def poller_processor_handler(event, context):  # pylint: disable=W0613
    """
    Historical Security Group Poller Processor.

    This will receive events from the Poller Tasker, and will list all objects of a given technology for an
    account/region pair. This will generate `polling events` which simulate changes. These polling events contain
    configuration data such as the account/region defining where the collector should attempt to gather data from.
    """
    LOG.debug('[@] Running Poller...')

    collector_poller_queue_url = get_queue_url(os.environ.get('POLLER_QUEUE_NAME', 'HistoricalSecurityGroupPoller'))
    takser_queue_url = get_queue_url(os.environ.get('POLLER_TASKER_QUEUE_NAME', 'HistoricalSecurityGroupPollerTasker'))

    poller_task_schema = HistoricalPollerTaskEventModel()
    records = deserialize_records(event['Records'])

    for record in records:
        # Skip accounts that have role assumption errors:
        try:
            # Did we get a NextToken?
            if record.get('NextToken'):
                LOG.debug(f"[@] Received pagination token: {record['NextToken']}")
                groups = describe_security_groups(
                    account_number=record['account_id'],
                    assume_role=HISTORICAL_ROLE,
                    region=record['region'],
                    MaxResults=200,
                    NextToken=record['NextToken']
                )
            else:
                groups = describe_security_groups(
                    account_number=record['account_id'],
                    assume_role=HISTORICAL_ROLE,
                    region=record['region'],
                    MaxResults=200
                )

            # FIRST THINGS FIRST: Did we get a `NextToken`? If so, we need to enqueue that ASAP because
            # 'NextToken`s expire in 60 seconds!
            if groups.get('NextToken'):
                logging.debug(f"[-->] Pagination required {groups['NextToken']}. Tasking continuation.")
                produce_events(
                    [poller_task_schema.serialize_me(record['account_id'], record['region'],
                                                     next_token=groups['NextToken'])],
                    takser_queue_url
                )

            # Task the collector to perform all the DDB logic -- this will pass in the collected data to the
            # collector in very small batches.
            events = [SECURITY_GROUP_POLLING_SCHEMA.serialize(record['account_id'], g, record['region'])
                      for g in groups['SecurityGroups']]
            produce_events(events, collector_poller_queue_url, batch_size=3)

            LOG.debug(f"[@] Finished generating polling events. Account: {record['account_id']}/{record['region']} "
                      f"Events Created: {len(events)}")
        except ClientError as exc:
            LOG.error(f"[X] Unable to generate events for account/region. Account Id/Region: {record['account_id']}"
                      f"/{record['region']} Reason: {exc}")
def poller_processor_handler ( event , context ) : LOG . debug ( '[@] Running Poller...' ) collector_poller_queue_url = get_queue_url ( os . environ . get ( 'POLLER_QUEUE_NAME' , 'HistoricalSecurityGroupPoller' ) ) takser_queue_url = get_queue_url ( os . environ . get ( 'POLLER_TASKER_QUEUE_NAME' , 'HistoricalSecurityGroupPollerTasker' ) ) poller_task_schema = HistoricalPollerTaskEventModel ( ) records = deserialize_records ( event [ 'Records' ] ) for record in records : try : if record . get ( 'NextToken' ) : LOG . debug ( f"[@] Received pagination token: {record['NextToken']}" ) groups = describe_security_groups ( account_number = record [ 'account_id' ] , assume_role = HISTORICAL_ROLE , region = record [ 'region' ] , MaxResults = 200 , NextToken = record [ 'NextToken' ] ) else : groups = describe_security_groups ( account_number = record [ 'account_id' ] , assume_role = HISTORICAL_ROLE , region = record [ 'region' ] , MaxResults = 200 ) if groups . get ( 'NextToken' ) : logging . debug ( f"[-->] Pagination required {groups['NextToken']}. Tasking continuation." ) produce_events ( [ poller_task_schema . serialize_me ( record [ 'account_id' ] , record [ 'region' ] , next_token = groups [ 'NextToken' ] ) ] , takser_queue_url ) events = [ SECURITY_GROUP_POLLING_SCHEMA . serialize ( record [ 'account_id' ] , g , record [ 'region' ] ) for g in groups [ 'SecurityGroups' ] ] produce_events ( events , collector_poller_queue_url , batch_size = 3 ) LOG . debug ( f"[@] Finished generating polling events. Account: {record['account_id']}/{record['region']} " f"Events Created: {len(events)}" ) except ClientError as exc : LOG . error ( f"[X] Unable to generate events for account/region. Account Id/Region: {record['account_id']}" f"/{record['region']} Reason: {exc}" )
Historical Security Group Poller Processor.

    This will receive events from the Poller Tasker, and will list all objects of a given technology for an
    account/region pair. This will generate `polling events` which simulate changes. These polling events contain
    configuration data such as the account/region defining where the collector should attempt to gather data from.