Skip to main content

Functions + Object Storage integration

This integration uses object storage as the intake point and a Function as the processor. A trigger lists the bucket on an interval and POSTs each new object to the function, so you can validate, transform, or index a file without hand-wiring a worker loop.

Two measured facts decide the shape of everything below, so they come first:

  • The object's content is the request body. Your handler is handed the bytes, not a JSON description of them. It needs no S3 client and never talks to the bucket.
  • The object's key is not delivered. Nothing in the request says which file this was — see The object's key is not delivered before you write a filename check.

Use it when​

  • files arrive in a bucket,
  • each object can be processed independently, and is smaller than 8 MiB,
  • up to a minute of latency is fine — the trigger polls, it does not listen.

The pieces​

  • Bucket — the source of new files, in any store that speaks S3. The platform does not host it, so the trigger always needs an endpoint.
  • Trigger — lists the bucket every poll_seconds (default 60), POSTs what it finds, and retires each object according to after_read.
  • Function — receives the object's bytes and does the work.

Step 1: Deploy the processor​

def handle(event: dict) -> dict:
    text = event.get("data")               # the object's own bytes, decoded as UTF-8
    if text is None:                       # not UTF-8 text: event["data_base64"] holds it
        print("skipped: object is not text")
        return {"statusCode": 200}
    attrs = event.get("_cloudevent", {})   # id, source, specversion, subject, time, type
    title = next((line[2:].strip() for line in text.splitlines()
                  if line.startswith("# ")), "untitled")
    print(f"ingesting {len(text)} characters as {title!r} (ce.id={attrs.get('id')})")
    # ... write the result somewhere durable ...
    return {"statusCode": 200}
platformctl functions deploy ./doc-ingest --name doc-ingest

Three things about the delivery that change how you write the handler:

  • The shape depends on the object. A JSON object arrives parsed as the event itself. Any other text — markdown, CSV, a log line — arrives as event["data"]. A JSON scalar or array arrives as event["data"]. Bytes that are not valid UTF-8 arrive as event["data_base64"], base64-encoded. None of those is an error.
  • The return value is discarded. Every delivery carries a ce-id header, so the shim treats it as a CloudEvent: it answers 204 and throws the result away. The work has to happen inside handle(), not in what it returns. A handler exception becomes a terminal 400 and the delivery is not retried.
  • This is the Python runtime specifically. The Node.js, Go and Ruby shims answer 400 invalid JSON body: ... for a body that is not JSON, and none of them exposes _cloudevent. A markdown or CSV drop therefore never reaches those handlers. Write bucket-triggered functions in Python, or make sure the objects you drop are JSON.

The object's key is not delivered​

This is the trap that costs an afternoon. The CloudEvent attributes under event["_cloudevent"] were measured carrying id, source, specversion, subject, time and type — and subject is the literal string aws-s3-source, the name of the source, not the name of your object. It has no prefix and no extension, so a suffix check on it:

if not attrs.get("subject", "").endswith(".md"):   # rejects every file in the bucket
    return {"statusCode": 200}

rejects everything, and a dedupe keyed on it collapses every delivery into one. Nothing errors: the poller keeps retiring objects, the function keeps returning 200, and the work silently never happens.

Print event["_cloudevent"] on your first delivery and read what your store actually sends. When nothing there looks like a key — an object key contains a / or a . — name the object yourself:

  • Derive the name from the content — a markdown # H1, a CSV header row, an id field, as the handler above does. Fall back to a hash of the content so two unnamed objects stay distinct.
  • Or write the identity into the object when you produce it, so the delivery carries it in the body.

Step 2: Connect the trigger​

An object-store trigger needs four fields: bucket, endpoint, credentials_secret_name and after_read. None of them has a default.

platformctl serverless triggers create doc-drop \
  --target doc-ingest \
  --type objectstore \
  --bucket uploads \
  --endpoint https://object.us-southcentral1-a.crusoecloudcompute.com \
  --region us-southcentral1-a \
  --credentials-secret crusoe-object-store \
  --prefix incoming/ \
  --after-read move \
  --move-to-prefix processed/

Four things to get right here:

  • credentials_secret_name is not your Crusoe Cloud API key. Object storage answers that key with 403. On a project connected to Crusoe Cloud the platform has already minted a real S3 key and stored it as the Secret crusoe-object-store (keys aws.accessKey and aws.secretKey) — use that name. If it is missing, open Project Settings (the page that connects a project to Crusoe Cloud) and press Check again — the platform mints the key only for a project that has none.

  • region must match the bucket's. It goes straight to the poller's S3 client, so a mismatch fails at connect time, inside the poller, with nothing on the trigger to say why. The default is us-east-1, which is almost certainly wrong for a Crusoe bucket.

  • after_read is required and has no default, because a trigger that never retires an object fires on it forever. Leaving it out is refused:

    source.objectstore.after_read must be "move", "delete" or "none" - there is no default, because a bucket trigger that never retires an object fires on it forever

    move needs a destination (--move-to-bucket, --move-to-prefix, or both), or it would move the object onto itself and re-deliver it on every poll.

  • retry does not apply to this source. A trigger's retry.max_attempts and retry.backoff reach schedule and Pub/Sub sources only; the bucket poller reads neither, and there is no dead-letter destination anywhere in this path.

Step 3: Test the pipeline​

Upload a small test file under the watched prefix and watch the function's logs.

Wait a full minute before deciding nothing happened. This is a poller, not a notification listener: poll_seconds defaults to 60, so a drop can sit for up to a minute before your function runs. One poll consumes at most max_messages_per_poll objects (default 10), so a bucket that gains 10,000 files does not become 10,000 simultaneous invocations either.

The acceptance signal is not "the upload succeeded"; it is "the function logged the content and produced the downstream result", and — with after_read: move or delete — "the object is no longer sitting in the watched prefix".

Common mistakes​

  • Reading event["bucket"] or event["key"]. Neither field exists. The body is the object itself; see Step 1.
  • Trusting subject as a filename. It is the source's own name, aws-s3-source, on every delivery.
  • Sending objects larger than 8 MiB. The body is the object, so the shim's request cap is a cap on object size: a bigger object is refused with 413 body exceeds 8388608 bytes before handle() runs. Keep the watched prefix to small files, and drop a manifest that names the large object instead.
  • Making the function non-idempotent. With after_read: none the same object is delivered again on every poll, forever, by design — the handler has to be safe to run twice on the same file. The trigger says so too: its No repeat delivery readiness check stays waiting and names the bucket.
  • Expecting a retry budget or a dead-letter path. There is neither on this source. If the work can fail transiently, catch it in the handler and write the object somewhere you can retry from.

Next steps​