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.

Python

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())

Copy

Go

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
}

Copy

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:

Google Cloud Storage

Amazon S3

Local File System

Registering a Storage Location

To make a storage location available within Tilebox workflows, it must be registered first. This involves specifying the location and setting up a notification system that forwards events to Tilebox, enabling task triggering. The setup varies depending on the storage location type.

For example, a GCP storage bucket is integrated by setting up a PubSub Notification with a push subscription. A local file system requires installing a filesystem watcher. To set up a storage location registered with Tilebox, please get in touch.

Listing Available Storage Locations

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

Python

from tilebox.workflows import Client

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

storage_locations = automations_client.storage_locations()
print(storage_locations)

Copy

Go

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()),
    )
}

Copy

Output

[
    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,
    ),
]

Copy

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.

Python

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")

Copy

The read method instantiates a client for the specific storage location. This requires that the storage location is accessible by a runner and may require credentials for cloud storage or physical/network access to a locally mounted file system.

To set up authentication and enable access to a GCS storage bucket, check out the Google Client docs for authentication.

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.

Python

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"),  
    ],
)

Copy

The syntax for specifying glob patterns follows Standard Wildcards. Additionally, you can use ** as a super-asterisk, a matching operator not sensitive to slash separators.

Here are some examples of valid glob patterns:

Pattern	Matches
*.ext	Any file ending in .ext in the root directory
**/*.ext	Any file ending in .ext in any subdirectory, but not in the root directory
**.ext	Any file ending in .ext in any subdirectory, including the root directory
folder/*	Any file directly in a folder subdirectory
folder/**	Any file directly or recursively part of a folder subdirectory
[a-z].txt	Matches a.txt, b.txt, etc.

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.

Python

from tilebox.workflows import Client, Runner

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

Copy

Go

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))
}

Copy

Triggering an Event

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

Creating an object

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

Copy

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

Output

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'

Copy

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.

Python

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")

Copy

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.

Python

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"),
)

Copy