Event-Driven Packaging

EDP is in private preview. Ask your Quilt account manager for details.

Overview

Data tend to be created in logical batches by machines, people, and pipelines. Detecting these logical events from Amazon S3 events alone is complex and requires extensive logic.

Quilt's Event-Driven Packaging (EDP) service intelligently groups one or more Amazon S3 object events into a single batch-level event. You can easily (and if desired, automatically) trigger logical events like data package creation that depend on batches rather than on individual files.

Any AWS service or action that generates S3 object events may trigger the EDP service.

Requirements

A pre-existing VPC that either includes a NAT Gateway or the following VPC endpoints:
- Amazon S3 (gateway endpoint or interface endpoint).
- EventBridge (interface endpoint).
Enable EventBridge S3 Events for all S3 buckets to be monitored by EDP.

Deployment

EDP deploys Lambda and RDS resources to monitor S3 and generate EventBridge events under user-configurable conditions.

Networking

Lambda and RDS resources are placed in the VPC and Subnets that you provide.
Subnets are normally private and must be able to reach Amazon services such as EventBridge via port 443 (e.g. by means of a NAT gateway, or VPC endpoint).
SecurityGroup should allow outbound access to AWS services on port 443. Does not need inbound access.

Parameters

EDP is deployed by a standalone CloudFormation template with the following parameters:

Parameter

Description

VPC

For EDP resources and Subnets.

Subnets

For EDP Lambda, RDS (see above for configuration).

SecurityGroup

For EDP Lambdas (see above for configuration).

BucketName

Name of the Amazon S3 bucket to monitor.

BucketIgnorePrefixes

Text string of comma separated bucket path segments to ignore, for example raw/*, scratch/*. Default value is an empty string (i.e. nothing ignored).

BucketPrefixDepth

The number of /-separated common path segments at the beginning of an S3 object key. Default value is 2.

BucketThresholdDuration

Trigger a notification when this number of seconds has elapsed since the last object event in the S3 bucket occurred. Default value is 300 seconds.

BucketThresholdEventCount

Trigger a notification when this number of files have been created (since the prior trigger). Default value is 20.

DBUser

Username for EDP RDS instance.

DBPassword

Password for EDP RDS instance.

EventBusName

Name of custom EventBridge event bus that receives events.

How EDP works

EDP monitors S3 object events for s3://bucket-name

After a fixed number of object events (BucketThresholdEventCount) or a maximum duration within a common prefix (BucketThresholdDuration), EDP creates a package-objects-ready event that signals there is sufficient information to make Quilt data packages from a batch of files:

S3 bucket name
Common prefix
Number of files
Timestamp of event

The event payload is JSON:

{
    "version":"0",
    "id":"XXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXX",
    "detail-type":"package-objects-ready",
    "source":"com.quiltdata.edp",
    "account":"XXXXXXXXXXXX",
    "time":"2022-12-08T20:01:34Z",
    "region":"us-east-1",
    "resources":[
        "arn:aws:s3:::bucket-name"
    ],
    "detail":{
        "version":"0.1",
        "bucket":"bucket-name",
        "prefix":"prefix-path-1/prefix-path-2/"
    }
}

EDP publishes the event to an AWS EventBridge bus. From there the event can be forwarded to any services that can be targeted from AWS EventBridge for additional manual or automatic processing.

EDP, upon completion and if configured to do so, may warm its contents to a File Gateway where it has read permissions to ensure that new EDP-created Quilt packages are available to Gateway clients like Windows Workspaces.

Users can optionally subscribe directly to the EDP SNS topic. This is useful for both debugging and viewing how events are structured.

Example: Lambda function to automatically create data packages

An instrument automatically uploads a folder containing files from a single experiment into s3://instrument-bucket/instrument-name/experiment-id/.
EDP listens for events in s3://instrument-bucket/instrument-name/experiment-id/*. After the specified duration or event count, a package-objects-ready event is generated and sent to EventBridge.
A custom SNS topic is created for monitoring data package creation that Lab and Computational scientists subscribe to (SNS_TOPIC_ARN).

A custom lambda function triggered by the package-objects-ready event processes the experiment files and generates a data package. Additional processing includes (but is not limited to):

Enhance the package with documentation, charts, and metadata, such as the following:
- README.md: Noting that the package was created by EDP, a custom lambda function, and validated with a Quilt workflow.
- quilt_summarize.json
- .quiltignore
Package metadata creation and validation: Send an SNS notification on metadata validation failure.

import datetime
import functools
import os
import pathlib
import tempfile
import boto3
import quilt3 as quilt3
from aws_lambda_powertools import Logger

logger = Logger()
s3 = boto3.client("s3")
sns = boto3.client("sns")

# Configuration environment variables defined for Lambda function
WORKFLOW_NAME = os.environ.get("WORKFLOW_NAME") or ...
QUARANTINE_BUCKET_NAME = os.environ["QUARANTINE_BUCKET_NAME"]
SNS_TOPIC_ARN = os.environ["SNS_TOPIC_ARN"]
QUILT_URL = os.environ["QUILT_URL"]

# README.md default Markdown
QUILT_README_STR = f"""#Quilt package auto-generated by EDP\n\n
Created on {datetime.date.today()} by an
automated Lambda agent for the {WORKFLOW_NAME} workflow."""

# File system files for Quilt to ignore
QUILT_IGNORE_STR = """.DS_*
Icon
._*
.TemporaryItems
.Trashes
.VolumeIcon.icns
"""

# Define helpful additional data package files
beautify_files = {
    "README.md": QUILT_README_STR,
    ".quiltignore": QUILT_IGNORE_STR,
}

@logger.inject_lambda_context
def lambda_handler(event, context):

    # EDP event data
    bucket = event["detail"]["bucket"]
    prefix = event["detail"]["prefix"]

    # Add every file in the prefix folder to the new data package
    pkg = quilt3.Package().set_dir(".", f"s3://{bucket}/{prefix}")

    # Decorate the data package with example required metadata (as defined by WORKFLOW_NAME)
    meta = {
        "Author": "EDP",
        "ComputerName": "Genome Lab - 1234",
        "Date": datetime.date.today().strftime("%Y-%m-%d"),
        "ProjectID": "YYD",
        "StudyID": "ABC-23-023394"
    }

    with tempfile.TemporaryDirectory() as tmpdir:
        tmpdir_path = pathlib.Path(tmpdir)
        for name, body in beautify_files.items():
            if name in pkg:
                logger.debug(f"File {name} already exists. Ignoring.")
                continue
            logger.debug(f"File {name} does not exist at {prefix}. Creating.")
            file_path = tmpdir_path / name
            file_path.write_text(body)
            pkg.set(name, file_path)

        # Add metadata to package
        pkg.set_meta(meta)
        # Remove leading & trailing characters
        pkg_name = prefix.strip("/")

        # Define callable Quilt push()
        push = functools.partial(
            pkg.push,
            pkg_name,
            registry=f"s3://{bucket}",
            force=True,
            message="Created by EDP",
            workflow=WORKFLOW_NAME
        )

        # Validate against the Quilt workflow schema
        try:
            push(dedupe=True)
        except quilt3.workflows.WorkflowValidationError as e:
            logger.warning("Workflow check failed")

            # Write out error to README.md file in quarantine bucket
            file_path = tmpdir_path / "README.md"
            file_path.write_text(str(e))
            pkg.set("README.md", file_path)

            # Push package to quarantine bucket
            push(registry=f"s3://{QUARANTINE_BUCKET_NAME}", workflow=...)

            # Error SNS notification content
            subject = f"Failed to create package"
            message = (
                f"Validation failed for workflow {WORKFLOW_NAME} while pushing "
                f"package with name {pkg_name} to {bucket}. It was pushed to "
                f"{QUARANTINE_BUCKET_NAME} instead.\n"
                f"{QUILT_URL}/b/{QUARANTINE_BUCKET_NAME}/packages/{pkg_name}\n\n"
                f"Error message is:\n{e}\n"
            )
            # Publish notification to SNS topic
            sns.publish(
                TopicArn=SNS_TOPIC_ARN,
                Message=message,
                Subject=subject,
            )

If a metadata validation error occurs, an SNS event is sent to SNS_TOPIC_ARN noting that the package was created in the quarantine bucket. The SNS notification is routed to subscribers.
Computational scientist opens the new data package for additional analysis, modeling, and versioning.

Debugging

EDP includes a CloudWatch dashboard which exposes some metrics useful for debugging:

EDP event bus topic: Displays the number of events emitted by EDP. If EDP is working correctly there should be one or more events received (depending on the time range selected).
Per-bucket metrics:
- S3 EventBridge rule: The number of events published to EventBridge from the specified Amazon S3 bucket. If there is no data, there are several possibilities:
  - Invocations: If this value is zero, the S3 bucket isn't correctly configured (Send notifications to Amazon EventBridge for all events in this bucket is not turned On).
  - TriggeredRules: If this value is zero, there was a problem with the automated EventBridge rule creation process during deployment. In general, you want the number of invocations to approximately equal the number of triggered rules.
  - Failed Invocations: This value should be zero. If greater than zero, there is an EDP configuration issue.
- Store in DB lambda: If EDP is configured correctly, there should be zero errors and a 100% success rate.
- Emit event lambda: If EDP is configured correctly, there should be zero errors and a 100% success rate.

Limitations

Each EDP stack monitors one S3 bucket.

PreviousBenchling Packager NextNextflow Plugin

Last updated 7 months ago