Storage Event Triggers

Creating a Storage Event Task

Storage Event Tasks are automations triggered when objects are created or modified in a storage location. In Python, use tilebox.workflows.automations.StorageEventTask as your task base class instead of the regular tilebox.workflows.Task. In Go, register an executable task with the same task identifier as the automation prototype.

from tilebox.workflows import ExecutionContext
from tilebox.workflows.automations import StorageEventTask, StorageEventType

class LogObjectCreation(StorageEventTask):
    head_bytes: int

    def execute(self, context: ExecutionContext) -> None:
        # self.trigger is an attribute of the StorageEventTask class,
        # which contains information about the storage event that triggered this task
        if self.trigger.type == StorageEventType.CREATED:
            path = self.trigger.location
            context.logger.info("A new object was created", path=path)

            # trigger.storage is a storage client for interacting with the storage
            # location that triggered the event
            # using it, we can read the object to get its content as a bytes object
            data = self.trigger.storage.read(path)
            context.logger.info("Read object data", path=path, size_bytes=len(data))
            context.logger.info("Object preview", path=path, preview=data[:self.head_bytes].hex())

type LogObjectCreation struct {
    HeadBytes int
}

func (t *LogObjectCreation) Execute(ctx context.Context) error {
    slog.InfoContext(ctx, "storage event task triggered", slog.Int("head_bytes", t.HeadBytes))
    return nil
}

Storage Locations

Storage Event tasks are triggered when objects are created or modified in a storage location. This location can be a cloud storage bucket or a local file system. Tilebox supports the following storage locations:

Registering a Storage Location

For cloud storage, connect a storage location and delivery subscription in the Console, then configure your provider to forward object-created notifications. The guide covers S3/SNS, GCS/Pub/Sub, and Azure Blob/Event Grid, including resources and permissions needed before saving in Tilebox.

Tilebox registration does not provision cloud resources or grant object-read access. Provider retries and overlapping subscriptions can create duplicate jobs; tasks should be safe to retry. A local file system requires a filesystem watcher rather than a cloud delivery subscription; get in touch for filesystem setup.

Listing Available Storage Locations

To list all available storage locations, use the automation client.

from tilebox.workflows import Client

client = Client()
automations_client = client.automations()

storage_locations = automations_client.storage_locations()
print(storage_locations)

ctx := context.Background()
client := workflows.NewClient()

storageLocations, err := client.Automations.ListStorageLocations(ctx)
if err != nil {
    slog.ErrorContext(ctx, "failed to list storage locations", slog.Any("error", err))
    return
}

for _, location := range storageLocations {
    slog.InfoContext(ctx,
        "storage location",
        slog.String("id", location.ID.String()),
        slog.String("location", location.Location),
        slog.String("type", location.Type.String()),
    )
}

[
    StorageLocation(
        location="gcp-project:gcs-bucket-fab3fa2",
        type=StorageType.GCS,
    ),
    StorageLocation(
        location="s3-bucket-263af15",
        type=StorageType.S3,
    ),
    StorageLocation(
        location='/path/to/a/local/folder',
        type=StorageType.FS,
    ),
]

Reading Files from a Storage Location

Once a storage location is registered, you can read files from it using the read method on the storage client.

gcs_bucket = storage_locations[0]
s3_bucket = storage_locations[1]
local_folder = storage_locations[2]

gcs_object = gcs_bucket.read("my-object.txt")
s3_object = s3_bucket.read("my-object.txt")
local_object = local_folder.read("my-object.txt")

Registering a Storage Event Trigger

After implementing a Storage Event task, register it to trigger each time a storage event occurs. The Python SDK provides a registration helper, and you can also register storage-event automations from the Tilebox Console. This registration submits a new job consisting of a single task instance derived from the registered Storage Event task prototype.

from tilebox.workflows import Client

client = Client()
automations = client.automations()
storage_event_automation = automations.create_storage_event_automation(
    "log-object-creations",  # name of the storage event automation
    LogObjectCreation(head_bytes=20),  # the task (and its input parameters) to run repeatedly
    triggers=[
        # you can specify a glob pattern:
        # run every time a .txt file is created anywhere in the gcs bucket
        (gcs_bucket, "**.txt"),  
    ],
)

Here are some examples of valid glob patterns:

Start a Storage Event runner

With the Storage Event automation registered, a job is submitted whenever a storage event occurs. But unless a runner is available to execute the Storage Event task the submitted jobs remain in a task queue. Once an eligible runner becomes available, all jobs in the queue are executed.

from tilebox.workflows import Client, Runner

client = Client()
runner = Runner(tasks=[LogObjectCreation])
runner.connect_to(client).run_forever()

ctx := context.Background()
client := workflows.NewClient()

runner, err := client.NewTaskRunner(ctx)
if err != nil {
    slog.ErrorContext(ctx, "failed to create task runner", slog.Any("error", err))
    return
}

if err := runner.RegisterTasks(&LogObjectCreation{}); err != nil {
    slog.ErrorContext(ctx, "failed to register tasks", slog.Any("error", err))
    return
}

if err := runner.RunForever(ctx); err != nil {
    slog.ErrorContext(ctx, "task runner stopped", slog.Any("error", err))
}

Triggering an Event

Creating an object in the bucket where the task is registered results in a job being submitted:

echo "Hello World" > my-object.txt
gcloud storage cp my-object.txt gs://gcs-bucket-fab3fa2

Inspecting the runner output reveals that the job was submitted and the task executed:

2024-09-25 16:51:45,621 INFO A new object was created: my-object.txt
2024-09-25 16:51:45,857 INFO The object's file size is 12 bytes
2024-09-25 16:51:45,858 INFO The object's first 20 bytes are: b'Hello World\n'

Inspecting in the Console

The Tilebox Console provides an easy way to inspect all registered storage event automations.

Deleting Storage Event automations

To delete a registered storage event automation from Python, use automations.delete. You can also delete storage-event automations from the Tilebox Console. After deletion, no new jobs will be submitted by the storage event trigger. Past jobs already triggered will still remain queued.

from tilebox.workflows import Client

client = Client()
automations = client.automations()

# delete the automation as returned by create_storage_event_automation
automations.delete(storage_event_automation)

# or manually by id:
automations.delete("0190bafc-b3b8-88c4-008b-a5db044380d0")

Submitting Storage Event jobs manually

In Python, you can submit Storage Event tasks as regular tasks for testing purposes or as part of a larger workflow. To do so, instantiate the task with a specific storage location and object name using the once method.

job_client = client.jobs()

task = LogObjectCreation(head_bytes=20)

# submitting it directly won't work; raises ValueError:
# job_client.submit("manual-storage-event-job", task)

# instead, we specify a trigger condition, and submit a job manually
job_client.submit(
    "manual-storage-event-job",
    # simulate an event that occurred in the gcs bucket for the object "my-object.txt"
    task.once(gcs_bucket, "my-object.txt"),
)