(message)
Google PubSub callback.

    This function is called on all incoming messages on the configured topic.

    Args:
        message: (dict) PubSub message
Google PubSub callback.
Google PubSub callback .
def callback(message):
    """Google PubSub callback.

    This function is called on all incoming messages on the configured topic.

    Args:
        message: (dict) PubSub message
    """
    message.ack()
    gcs_full_path = message.attributes.get('objectId')

    # Exit early if the file type is wrong.
    if not gcs_full_path.endswith('.plaso.metadata.json'):
        return

    gcs_base_path = os.path.dirname(gcs_full_path)
    gcs_metadata_filename = os.path.basename(gcs_full_path)
    gcs_base_filename = gcs_metadata_filename.replace('.metadata.json', '')
    gcs_plaso_filename = gcs_base_filename

    # Download files from GCS
    local_metadata_file = download_from_gcs(
        gcs_base_path, gcs_metadata_filename)
    local_plaso_file = download_from_gcs(gcs_base_path, gcs_plaso_filename)

    with open(local_metadata_file, 'r') as metadata_file:
        metadata = json.load(metadata_file)
        username = metadata.get('requester')
        sketch_id_from_metadata = metadata.get('sketch_id')

    if not username:
        logger.error('Missing username')
        return

    timeline_name = os.path.splitext(gcs_plaso_filename)[0]
    index_name = uuid.uuid4().hex
    sketch_id, timeline_id = setup_sketch(
        timeline_name, index_name, 'admin', sketch_id_from_metadata)

    # Start indexing
    with app.app_context():
        pipeline = tasks.build_index_pipeline(
            file_path=local_plaso_file, timeline_name=gcs_base_filename,
            index_name=index_name, file_extension='plaso', sketch_id=sketch_id,
            timeline_id=timeline_id)
        pipeline.apply_async()
        logger.info('File sent for indexing: {}'. format(gcs_base_filename))
def callback ( message ) : message . ack ( ) gcs_full_path = message . attributes . get ( 'objectId' ) # Exit early if the file type is wrong. if not gcs_full_path . endswith ( '.plaso.metadata.json' ) : return gcs_base_path = os . path . dirname ( gcs_full_path ) gcs_metadata_filename = os . path . basename ( gcs_full_path ) gcs_base_filename = gcs_metadata_filename . replace ( '.metadata.json' , '' ) gcs_plaso_filename = gcs_base_filename # Download files from GCS local_metadata_file = download_from_gcs ( gcs_base_path , gcs_metadata_filename ) local_plaso_file = download_from_gcs ( gcs_base_path , gcs_plaso_filename ) with open ( local_metadata_file , 'r' ) as metadata_file : metadata = json . load ( metadata_file ) username = metadata . get ( 'requester' ) sketch_id_from_metadata = metadata . get ( 'sketch_id' ) if not username : logger . error ( 'Missing username' ) return timeline_name = os . path . splitext ( gcs_plaso_filename ) [ 0 ] index_name = uuid . uuid4 ( ) . hex sketch_id , timeline_id = setup_sketch ( timeline_name , index_name , 'admin' , sketch_id_from_metadata ) # Start indexing with app . app_context ( ) : pipeline = tasks . build_index_pipeline ( file_path = local_plaso_file , timeline_name = gcs_base_filename , index_name = index_name , file_extension = 'plaso' , sketch_id = sketch_id , timeline_id = timeline_id ) pipeline . apply_async ( ) logger . info ( 'File sent for indexing: {}' . format ( gcs_base_filename ) )