Publish S3 bucket events to a topic

Applies to VAST Cluster 5.5.

VAST Cluster can publish an event to a VAST Event Broker topic each time an object in an S3 bucket is created, tagged or deleted. Your application then reacts to new data with an ordinary Kafka consumer, instead of listing the bucket. This guide sets up a bucket notification with the S3 API, consumes the events, and explains their format and delivery guarantees.

Run the code blocks in order in one Python session. Below each block is the output you should see.

How it works

1. You put a notification configuration on an S3 bucket: which events, which objects (by key prefix or suffix), and which topic.
2. When an S3 request on the bucket succeeds and matches a rule, the cluster queues an event record. Failed requests create no events (Event Publishing).
3. In the background, the cluster writes the record into the topic, as JSON.
4. Your consumers read it like any other Kafka record. The topic is also a table you can query.

The S3 request doesn't wait for the event: publishing is asynchronous, so expect a short delay between the request and the event.

A notification rule on the bucket publishes S3 events into a topic, asynchronously and at least once.

Before you start

You need:

• A Kafka view and a topic on it. The topic must exist before events are published to it: the notification configuration doesn't create it, and the cluster doesn't check that it exists when you save the configuration. This guide creates its topic with a Kafka client.
• An S3 bucket to watch, on a view with the S3 Bucket protocol and the S3 Native security flavor, in the same tenant as the Kafka view (Configuring S3 Event Notifications for a View).
• S3 credentials for a user that can change the bucket's notification configuration and write objects to it.
• The environment variables and clients from Query topics as tables, and boto3. Set SOURCE_BUCKET to the bucket to watch:

export SOURCE_BUCKET="<bucket to watch>"
pip install boto3

Your platform team can also set up the notification for you, in the VAST Web UI (the view's Event Notifications tab) or with the VAST CLI. Admin task: send to your platform team. See Publishing S3 Bucket Events to VAST Event Broker. S3 event notifications aren't supported on Global Access destination buckets (5.5.0 release notes).

Set up

Create the topic, and an S3 client:

import json
import os
import time
from urllib.parse import unquote_plus

import boto3
from confluent_kafka import Consumer, KafkaException
from confluent_kafka.admin import AdminClient, NewTopic

EVENTS_BUCKET = os.environ["EVENTS_BUCKET"]
SOURCE_BUCKET = os.environ["SOURCE_BUCKET"]
TOPIC = "docs-s3-bucket-events-objects"
GROUP = "docs-s3-bucket-events-reader"
PREFIX = "docs-s3-bucket-events/"

def kafka_config():
    cfg = {"bootstrap.servers": os.environ["KAFKA_BOOTSTRAP"]}
    if os.environ.get("KAFKA_USERNAME"):
        cfg.update({
            "security.protocol": "SASL_PLAINTEXT",
            "sasl.mechanism": "PLAIN",
            "sasl.username": os.environ["KAFKA_USERNAME"],
            "sasl.password": os.environ["KAFKA_PASSWORD"],
        })
    return cfg

admin = AdminClient(kafka_config())
admin.create_topics([NewTopic(TOPIC, num_partitions=3, replication_factor=1)])[TOPIC].result()

s3 = boto3.client("s3", endpoint_url=os.environ["AWS_S3_ENDPOINT_URL"])
print("ready")

ready

Configure the notification

Use the standard S3 PutBucketNotificationConfiguration call, with a topic configuration. Write the topic as:

vast-event-broker:<bucket of the Kafka view>:<topic>

vast-event-broker is a fixed word that means the cluster's own broker. For a third-party Kafka broker that your platform team has added to the cluster, the format is <broker name>:<topic> instead.

s3.put_bucket_notification_configuration(
    Bucket=SOURCE_BUCKET,
    NotificationConfiguration={
        "TopicConfigurations": [{
            "Id": "docs-s3-bucket-events",
            "TopicArn": f"vast-event-broker:{EVENTS_BUCKET}:{TOPIC}",
            "Events": ["ObjectCreated:*", "ObjectRemoved:*"],
            "Filter": {"Key": {"FilterRules": [{"Name": "prefix", "Value": PREFIX}]}},
        }],
    },
)
print("notification configured")

notification configured

The rules:

• Event names don't have the s3: prefix that AWS uses:

  Events	Names
  Object creation	ObjectCreated:Put, ObjectCreated:Post, ObjectCreated:Copy, ObjectCreated:CompleteMultipartUpload, or ObjectCreated:* for all
  Object tagging	ObjectTagging:Put, ObjectTagging:Delete, or ObjectTagging:*
  Object deletion	ObjectRemoved:Delete, ObjectRemoved:DeleteMarkerCreated, or ObjectRemoved:*

  Other names, including the AWS form s3:ObjectCreated:*, are rejected with InvalidArgument. Don't list a wildcard and one of its own events in the same rule, such as ObjectCreated:* and ObjectCreated:Put.

• Filters take a key prefix, a suffix or both. Write the rule names in lowercase, prefix and suffix.

• Rules can't overlap. Two rules can't both match the same request, for example one for keys starting abc and one for keys starting abcdef (Configuring S3 Event Notifications for a View).

• The Kafka view's bucket must be in the same tenant as the source bucket. Otherwise the call fails with kafka bucket name not found.

• The call replaces the bucket's whole notification configuration, as in AWS S3. To change one rule, send all of them.

Make some events

Upload two objects and delete one:

s3.put_object(Bucket=SOURCE_BUCKET, Key=PREFIX + "a.json", Body=b'{"n": 1}')
s3.put_object(Bucket=SOURCE_BUCKET, Key=PREFIX + "b.json", Body=b'{"n": 2}')
s3.delete_object(Bucket=SOURCE_BUCKET, Key=PREFIX + "a.json")
print("done")

done

Consume the events

Each Kafka record's value is a JSON document with a Records list. Poll until the three events have arrived, or a timeout passes:

consumer = Consumer({**kafka_config(), "group.id": GROUP,
                     "auto.offset.reset": "earliest", "enable.auto.commit": False})
consumer.subscribe([TOPIC])

events = []
deadline = time.time() + 120
while len(events) < 3 and time.time() < deadline:
    msg = consumer.poll(1.0)
    if msg is None:
        continue
    if msg.error():
        raise KafkaException(msg.error())
    body = json.loads(msg.value())
    for record in body.get("Records", []):      # skip anything else, such as test events
        events.append((msg.key(), record))
    consumer.commit(message=msg, asynchronous=False)

for kafka_key, record in sorted(events, key=lambda e: e[1]["s3"]["object"]["key"]):
    name = record["eventName"].removeprefix("s3:")
    obj = record["s3"]["object"]
    print(name, record["s3"]["bucket"]["name"], unquote_plus(obj["key"]), obj.get("size"), kafka_key)

ObjectCreated:Put docs-s3-source docs-s3-bucket-events/a.json 8 b'docs-s3-source/docs-s3-bucket-events/a.json'
ObjectRemoved:Delete docs-s3-source docs-s3-bucket-events/a.json 8 b'docs-s3-source/docs-s3-bucket-events/a.json'
ObjectCreated:Put docs-s3-source docs-s3-bucket-events/b.json 8 b'docs-s3-source/docs-s3-bucket-events/b.json'

The sort is stable, so the events for each object stay in the order they arrived.

Event format

The record follows the AWS S3 event structure (S3 Event Record Format):

{"Records": [{
  "eventVersion": "2.2",
  "eventSource": "vast:s3",
  "awsRegion": "<region>",
  "eventTime": "<timestamp>",
  "eventName": "<event type>",
  "userIdentity": {"principalId": "<principal ID>"},
  "requestParameters": {"sourceIPAddress": "<source IP>"},
  "responseElements": {"x-amz-request-id": "<request ID>", "x-amz-id-2": "<request ID>"},
  "s3": {
    "s3SchemaVersion": "1.0",
    "configurationId": "<rule Id>",
    "bucket": {"name": "<bucket name>", "ownerIdentity": {"principalId": "<principal ID>"}, "arn": "<bucket ARN>"},
    "object": {"key": "<key>", "size": <size>, "eTag": "<eTag>", "versionId": "<version>", "sequencer": "<sequencer>"}
  }
}]}

What to rely on:

The Kafka record's key is <bucket>/<object key>, and all events for one object go to the same partition, so they arrive in order for that object. Events for different objects can arrive in any order.

The cluster sends a test event ("Event": "s3:TestEvent") when you save a notification definition. It has no Records list, so skip records without one, as the example does.

Delivery: at least once

The cluster retries publishing after a failure, so an event can be delivered more than once. Make your consumer idempotent:

• Deduplicate on the request ID (responseElements["x-amz-request-id"]) plus eventName and the object key, or
• Make the processing safe to repeat, for example by writing results keyed by bucket, object key and eTag.

Don't use the topic as an audit log of every request. In rare failures, such as the cluster being unable to read the notification configuration, the S3 request still succeeds, and the event is dropped with an alert to your platform team.

The consumer side works like any other topic: commit offsets after you've processed a record, as the example does, so a restart resumes where you left off. See Consuming.

Clean up

Remove the notification rules, delete the test object, and delete the consumer group and topic:

s3.put_bucket_notification_configuration(Bucket=SOURCE_BUCKET, NotificationConfiguration={})
s3.delete_object(Bucket=SOURCE_BUCKET, Key=PREFIX + "b.json")
consumer.close()
admin.delete_consumer_groups([GROUP])[GROUP].result()
admin.delete_topics([TOPIC])[TOPIC].result()
print("cleaned up")

cleaned up

Remove the notification rules before you delete their topic, so the cluster doesn't try to publish to a topic that no longer exists.

See also

• Query topics as tables: read bucket events with SQL, such as all objects created in the last hour.
• Consuming: consumer groups and offset commits.
• Topics: create topics and set their retention.
• Event Publishing in the KB, including third-party brokers.