Durable Execution Telemetry Exporter¶

The Durable Execution Telemetry Exporter is a plugin that builds a telemetry record for each durable execution and sends it to one or more destinations. It builds on the plugin interface. It runs inside your function, in the same process as your handler, and subscribes to the SDK's lifecycle hooks. It makes no calls of its own until a record is due, and an exporter failure never fails the execution.

Install the exporter¶

npm install @aws/durable-execution-sdk-js-telemetry-exporter

npm install @aws-sdk/client-s3 @aws-sdk/client-dynamodb @aws-sdk/util-dynamodb \
  @aws-sdk/client-rds-data @aws-sdk/client-cloudwatch-logs @aws-sdk/client-firehose \
  @aws-sdk/client-eventbridge @aws-sdk/client-redshift-data @aws-sdk/client-sqs \
  @aws-sdk/credential-provider-node "@smithy/signature-v4@^4" @aws-crypto/sha256-js

pip install aws-durable-execution-sdk-python-telemetry-exporter

pip install "aws-durable-execution-sdk-python-telemetry-exporter[s3]"

<dependency>
    <groupId>software.amazon.lambda.durable</groupId>
    <artifactId>aws-durable-execution-sdk-java-plugin-telemetry-exporter</artifactId>
    <version>${durable.sdk.version}</version>
</dependency>

go get github.com/aws/aws-durable-execution-sdk-go/telemetry

Register the exporter¶

Create the plugin and register it in your handler configuration. With no configuration, the exporter uses LambdaLogExporter, which writes each record as one INFO log line to your function's own CloudWatch log group. Your function already has permission to write its own logs, so this needs no new infrastructure and no new IAM permission. In TypeScript, Python, and Java, this exporter writes the operationsByName map, not the operations array. With an application log level of WARN or higher, Lambda drops these INFO lines, so no records arrive.

import { DurableContext, withDurableExecution } from "@aws/durable-execution-sdk-js";
import { durableExecutionTelemetry } from "@aws/durable-execution-sdk-js-telemetry-exporter";

export const handler = withDurableExecution(
  async (event: { id: string }, context: DurableContext) => {
    const data = await context.step("fetch-data", async () => event.id);
    await context.wait("cooldown", { seconds: 5 });
    return await context.step("process", async () => `processed ${data}`);
  },
  { plugins: [durableExecutionTelemetry({})] },
);

from aws_durable_execution_sdk_python import DurableContext, durable_execution
from aws_durable_execution_sdk_python.config import Duration
from aws_durable_execution_sdk_python_telemetry_exporter import (
    DurableExecutionTelemetryConfig,
    durable_execution_telemetry,
)


@durable_execution(
    plugins=[durable_execution_telemetry(DurableExecutionTelemetryConfig())]
)
def handler(event: dict, context: DurableContext) -> str:
    data = context.step(lambda ctx: event["id"], name="fetch-data")
    context.wait(Duration.from_seconds(5), name="cooldown")
    return context.step(lambda ctx: f"processed {data}", name="process")

import java.time.Duration;
import java.util.Map;
import software.amazon.lambda.durable.DurableConfig;
import software.amazon.lambda.durable.DurableContext;
import software.amazon.lambda.durable.DurableHandler;
import software.amazon.lambda.durable.telemetry.DurableExecutionTelemetry;
import software.amazon.lambda.durable.telemetry.DurableExecutionTelemetryConfig;

public class TelemetryExample extends DurableHandler<Map<String, String>, String> {
    @Override
    protected DurableConfig createConfiguration() {
        return DurableConfig.builder()
                .withPlugins(DurableExecutionTelemetry.durableExecutionTelemetry(
                        DurableExecutionTelemetryConfig.builder().build()))
                .build();
    }

    @Override
    public String handleRequest(Map<String, String> event, DurableContext context) {
        String data = context.step("fetch-data", String.class, ctx -> event.get("id"));
        context.wait("cooldown", Duration.ofSeconds(5));
        return context.step("process", String.class, ctx -> "processed " + data);
    }
}

package main

import (
    "time"

    "github.com/aws/aws-durable-execution-sdk-go/durable"
    "github.com/aws/aws-durable-execution-sdk-go/telemetry"
)

func handler(ctx durable.Context, event map[string]string) (string, error) {
    data, err := durable.Step(ctx, "fetch-data", func(sc durable.StepContext) (string, error) {
        return event["id"], nil
    })
    if err != nil {
        return "", err
    }

    if err := durable.Wait(ctx, "cooldown", 5*time.Second); err != nil {
        return "", err
    }

    return durable.Step(ctx, "process", func(sc durable.StepContext) (string, error) {
        return "processed " + data, nil
    })
}

func main() {
    plugin := telemetry.New(telemetry.Config{})
    durable.Start(handler, durable.WithPlugins(plugin.Plugin()))
}

You can query the records with CloudWatch Logs Insights right away. Durable functions always use the JSON log format. In TypeScript, Python, and Java, each record arrives as a JSON string in the message field of an INFO log line. Match the line, then parse the fields you need out of the string:

filter level = "INFO" and message like /DurableExecutionTelemetry/
| parse message '"executionName":"*"' as executionName
| parse message '"status":"*"' as status
| parse message '"durationMs":*,' as duration
| filter status = "FAILED"
| fields tonumber(duration) as durationMs
| display executionName, status, durationMs
| sort durationMs desc

Each parse takes the first match in the record. The record writes the execution's executionName, status, and durationMs before its operations, so the query reads the execution's values and not those of an operation.

Go is the exception. Lambda leaves Go's application log lines as they are, so each Go record arrives as a bare JSON object, and Logs Insights discovers its fields directly:

fields executionName, status, durationMs
| filter recordType = "DurableExecutionTelemetry" and status = "FAILED"
| sort durationMs desc

Method signature¶

function durableExecutionTelemetry(
  config: DurableExecutionTelemetryConfig,
): DurableInstrumentationPluginFactory

def durable_execution_telemetry(
    config: DurableExecutionTelemetryConfig,
) -> DurableExecutionTelemetryPluginFactory

public static DurableExecutionPluginFactory durableExecutionTelemetry(
        DurableExecutionTelemetryConfig config)

func New(cfg Config) *TelemetryPlugin
func (p *TelemetryPlugin) Plugin() durable.Plugin

Choose exporters¶

Log groups do not suit questions across thousands of executions, such as the slowest step in the last week. The exporter ships with 13 destinations, so you can send records where those questions are cheap to answer:

The plugin sends each record to every configured exporter. If one exporter fails, the others still receive the record. TypeScript, Java, and Go send to the exporters concurrently. Python calls them one at a time on a single worker thread. A common setup uses one store for analytics, one for lookup, and one for alerting:

import { DurableContext, withDurableExecution } from "@aws/durable-execution-sdk-js";
import {
  DynamoDBExporter,
  EventBridgeExporter,
  S3Exporter,
  durableExecutionTelemetry,
} from "@aws/durable-execution-sdk-js-telemetry-exporter";

const telemetry = durableExecutionTelemetry({
  emitMode: "on-complete",
  samplingRate: 1.0,
  exporters: [
    new S3Exporter({ bucket: "durable-execution-telemetry-archive" }),
    new DynamoDBExporter({ tableName: "durable-execution-telemetry-live" }),
    new EventBridgeExporter({}),
  ],
});

export const handler = withDurableExecution(
  async (event: { id: string }, context: DurableContext) => {
    const data = await context.step("fetch-data", async () => event.id);
    await context.wait("cooldown", { seconds: 5 });
    return await context.step("process", async () => `processed ${data}`);
  },
  { plugins: [telemetry] },
);

from aws_durable_execution_sdk_python import DurableContext, durable_execution
from aws_durable_execution_sdk_python.config import Duration
from aws_durable_execution_sdk_python_telemetry_exporter import (
    DurableExecutionTelemetryConfig,
    DynamoDBExporter,
    EventBridgeExporter,
    S3Exporter,
    durable_execution_telemetry,
)

telemetry = durable_execution_telemetry(
    DurableExecutionTelemetryConfig(
        emit_mode="on-complete",
        sampling_rate=1.0,
        exporters=[
            S3Exporter(bucket="durable-execution-telemetry-archive"),
            DynamoDBExporter(table_name="durable-execution-telemetry-live"),
            EventBridgeExporter(),
        ],
    )
)


@durable_execution(plugins=[telemetry])
def handler(event: dict, context: DurableContext) -> str:
    data = context.step(lambda ctx: event["id"], name="fetch-data")
    context.wait(Duration.from_seconds(5), name="cooldown")
    return context.step(lambda ctx: f"processed {data}", name="process")

import java.time.Duration;
import java.util.Map;
import software.amazon.lambda.durable.DurableConfig;
import software.amazon.lambda.durable.DurableContext;
import software.amazon.lambda.durable.DurableHandler;
import software.amazon.lambda.durable.telemetry.DurableExecutionTelemetry;
import software.amazon.lambda.durable.telemetry.DurableExecutionTelemetryConfig;
import software.amazon.lambda.durable.telemetry.DurableExecutionTelemetryConfig.EmitMode;
import software.amazon.lambda.durable.telemetry.exporters.DynamoDBExporter;
import software.amazon.lambda.durable.telemetry.exporters.EventBridgeExporter;
import software.amazon.lambda.durable.telemetry.exporters.S3Exporter;

public class MultipleExportersExample extends DurableHandler<Map<String, String>, String> {
    @Override
    protected DurableConfig createConfiguration() {
        return DurableConfig.builder()
                .withPlugins(
                        DurableExecutionTelemetry.durableExecutionTelemetry(DurableExecutionTelemetryConfig.builder()
                                .emitMode(EmitMode.ON_COMPLETE)
                                .samplingRate(1.0)
                                .addExporter(S3Exporter.builder()
                                        .bucket("durable-execution-telemetry-archive")
                                        .build())
                                .addExporter(DynamoDBExporter.builder()
                                        .tableName("durable-execution-telemetry-live")
                                        .build())
                                .addExporter(EventBridgeExporter.builder().build())
                                .build()))
                .build();
    }

    @Override
    public String handleRequest(Map<String, String> event, DurableContext context) {
        String data = context.step("fetch-data", String.class, ctx -> event.get("id"));
        context.wait("cooldown", Duration.ofSeconds(5));
        return context.step("process", String.class, ctx -> "processed " + data);
    }
}

package main

import (
    "context"
    "log"
    "time"

    "github.com/aws/aws-durable-execution-sdk-go/durable"
    "github.com/aws/aws-durable-execution-sdk-go/telemetry"
    "github.com/aws/aws-sdk-go-v2/config"
    "github.com/aws/aws-sdk-go-v2/service/dynamodb"
    "github.com/aws/aws-sdk-go-v2/service/eventbridge"
    "github.com/aws/aws-sdk-go-v2/service/s3"
)

func handler(ctx durable.Context, event map[string]string) (string, error) {
    data, err := durable.Step(ctx, "fetch-data", func(sc durable.StepContext) (string, error) {
        return event["id"], nil
    })
    if err != nil {
        return "", err
    }

    if err := durable.Wait(ctx, "cooldown", 5*time.Second); err != nil {
        return "", err
    }

    return durable.Step(ctx, "process", func(sc durable.StepContext) (string, error) {
        return "processed " + data, nil
    })
}

func main() {
    cfg, err := config.LoadDefaultConfig(context.Background())
    if err != nil {
        log.Fatal(err)
    }

    plugin := telemetry.New(telemetry.Config{
        EmitMode:     telemetry.EmitOnComplete,
        SamplingRate: 1.0,
        Exporters: []telemetry.Exporter{
            &telemetry.S3Exporter{
                API:    s3.NewFromConfig(cfg),
                Bucket: "durable-execution-telemetry-archive",
            },
            &telemetry.DynamoDBExporter{
                API:       dynamodb.NewFromConfig(cfg),
                TableName: "durable-execution-telemetry-live",
            },
            &telemetry.EventBridgeExporter{
                API: eventbridge.NewFromConfig(cfg),
            },
        },
    })

    durable.Start(handler, durable.WithPlugins(plugin.Plugin()))
}

Each exporter has defaults that match the MCP server defaults. S3Exporter writes under the prefix durable-execution-telemetry/ in Hive-style date partitions (year=YYYY/month=MM/day=DD/). The TypeScript exporter's CDK stack creates a Glue table with partition projection over that prefix, so Athena can query it without a crawler. S3Exporter writes one object per execution, and later records overwrite it. DynamoDBExporter writes the execution ARN to the pk attribute and the emit time to the sk attribute. So a table keyed on pk and sk keeps every record, and a table keyed on pk alone, like the CDK stack's, keeps only the latest one. On a pk-only table, you can stop writing sk with sortKey: "" in TypeScript, sort_key=None in Python, .sortKey("") in Java, or SortKeyField: telemetry.NoSortKey in Go. On a table keyed on pk and sk, keep writing sk, because every write fails without it. AuroraExporter and RedshiftExporter write to the durable_execution_telemetry table, and OpenSearchExporter writes to the durable-execution-telemetry index. EventBridgeExporter sends events to the default bus with the source software.amazon.lambda.durable.telemetry and the record status as the detail type.

For example, this Athena query finds the steps that fail most often in June:

SELECT op.name, COUNT(*) AS failures, AVG(op.durationms) AS avg_ms
FROM durable_execution_telemetry
CROSS JOIN UNNEST(operations) AS t(op)
WHERE op.type = 'STEP' AND op.status = 'FAILED' AND year = '2026' AND month = '06'
GROUP BY op.name
ORDER BY failures DESC;

Emit mode¶

The emit mode decides when the plugin sends a record. The default, on-complete, sends one record when the execution reaches SUCCEEDED or FAILED. on-change also sends a RUNNING record as operations start and finish, so you can watch an execution progress. on-failure sends one record only when the execution fails, which has the lowest overhead.

Each record is a complete snapshot, so the plugin coalesces updates. If records arrive faster than an exporter can send them, the plugin sends only the latest one.

Sampling¶

The sampling rate reduces how many executions produce records. It is a number from 0 to 1, and the default is 1, which records every execution. The plugin derives the decision from a hash of the execution ARN. The ARN stays the same across replays, so the decision never changes during an execution. A sampled-in execution emits all of its records, and a sampled-out execution emits none. You never receive a partly recorded execution.

Operation detail¶

The operation detail decides which operations a record includes. The default, top-level, includes only operations without a parent. It produces the same set of operations no matter when the record is emitted. full-tree also includes the children of child contexts, such as map items and parallel branches.

By default, the durable execution backend removes a finished child context's children from the state it passes to later invocations. So after a suspend and resume, a full-tree record contains only the children of contexts that were still running in the invocation that emitted it.

export const handler = withDurableExecution(workflow, {
  plugins: [durableExecutionTelemetry({ operationDetail: "full-tree" })],
  pluginsConfig: { childOperationsDepth: 1 },
});

Content¶

By default, a record includes the execution input and output as they are, plus the errors of each operation. That is convenient during development, but production input and output often contain data that should not leave your function. Use the content configuration to drop fields or replace them with a transformed copy. If a transform throws, the plugin omits that field rather than send the raw value.

This example records 10% of executions and keeps only the order ID from the input and the status from the output:

import { DurableContext, withDurableExecution } from "@aws/durable-execution-sdk-js";
import { durableExecutionTelemetry } from "@aws/durable-execution-sdk-js-telemetry-exporter";

const telemetry = durableExecutionTelemetry({
  samplingRate: 0.1,
  content: {
    input: (input) => ({ orderId: (input as { orderId?: string }).orderId }),
    output: (output) => ({ status: (output as { status?: string }).status }),
  },
});

export const handler = withDurableExecution(
  async (event: { orderId: string }, context: DurableContext) => {
    const order = await context.step("fetch-order", async () => event.orderId);
    await context.wait("cooldown", { seconds: 5 });
    return await context.step("process", async () => ({
      orderId: order,
      status: "processed",
    }));
  },
  { plugins: [telemetry] },
);

from aws_durable_execution_sdk_python import DurableContext, durable_execution
from aws_durable_execution_sdk_python.config import Duration
from aws_durable_execution_sdk_python_telemetry_exporter import (
    ContentConfig,
    DurableExecutionTelemetryConfig,
    durable_execution_telemetry,
)

telemetry = durable_execution_telemetry(
    DurableExecutionTelemetryConfig(
        sampling_rate=0.1,
        content=ContentConfig(
            input=lambda data: {"orderId": data.get("orderId")},
            output=lambda result: {"status": result.get("status")},
        ),
    )
)


@durable_execution(plugins=[telemetry])
def handler(event: dict, context: DurableContext) -> dict:
    order = context.step(lambda ctx: event["orderId"], name="fetch-order")
    context.wait(Duration.from_seconds(5), name="cooldown")
    return context.step(
        lambda ctx: {"orderId": order, "status": "processed"}, name="process"
    )

import java.time.Duration;
import java.util.Map;
import software.amazon.lambda.durable.DurableConfig;
import software.amazon.lambda.durable.DurableContext;
import software.amazon.lambda.durable.DurableHandler;
import software.amazon.lambda.durable.telemetry.ContentConfig;
import software.amazon.lambda.durable.telemetry.DurableExecutionTelemetry;
import software.amazon.lambda.durable.telemetry.DurableExecutionTelemetryConfig;

public class RedactContentExample extends DurableHandler<Map<String, String>, Map<String, String>> {
    @Override
    protected DurableConfig createConfiguration() {
        return DurableConfig.builder()
                .withPlugins(DurableExecutionTelemetry.durableExecutionTelemetry(
                        DurableExecutionTelemetryConfig.builder()
                                .samplingRate(0.1)
                                .content(ContentConfig.builder()
                                        .inputTransform(input -> Map.of("orderId", ((Map<?, ?>) input).get("orderId")))
                                        .outputTransform(output -> Map.of("status", ((Map<?, ?>) output).get("status")))
                                        .build())
                                .build()))
                .build();
    }

    @Override
    public Map<String, String> handleRequest(Map<String, String> event, DurableContext context) {
        String order = context.step("fetch-order", String.class, ctx -> event.get("orderId"));
        context.wait("cooldown", Duration.ofSeconds(5));
        return context.step("process", Map.class, ctx -> Map.of("orderId", order, "status", "processed"));
    }
}

package main

import (
    "encoding/json"
    "time"

    "github.com/aws/aws-durable-execution-sdk-go/durable"
    "github.com/aws/aws-durable-execution-sdk-go/telemetry"
)

type Result struct {
    OrderID string `json:"orderId"`
    Status  string `json:"status"`
}

func handler(ctx durable.Context, event map[string]string) (Result, error) {
    order, err := durable.Step(ctx, "fetch-order", func(sc durable.StepContext) (string, error) {
        return event["orderId"], nil
    })
    if err != nil {
        return Result{}, err
    }

    if err := durable.Wait(ctx, "cooldown", 5*time.Second); err != nil {
        return Result{}, err
    }

    return durable.Step(ctx, "process", func(sc durable.StepContext) (Result, error) {
        return Result{OrderID: order, Status: "processed"}, nil
    })
}

func keepOrderIDAndStatus(value string) string {
    var fields map[string]any
    if err := json.Unmarshal([]byte(value), &fields); err != nil {
        return "{}"
    }
    kept := map[string]any{}
    for _, key := range []string{"orderId", "status"} {
        if v, ok := fields[key]; ok {
            kept[key] = v
        }
    }
    out, _ := json.Marshal(kept)
    return string(out)
}

func main() {
    plugin := telemetry.New(telemetry.Config{
        SamplingRate: 0.1,
        Content: telemetry.ContentConfig{
            IncludeInput:           true,
            IncludeOutput:          true,
            IncludeOperationErrors: true,
            IncludeExecutionError:  true,
            Redactor:               keepOrderIDAndStatus,
        },
    })
    durable.Start(handler, durable.WithPlugins(plugin.Plugin()))
}

Record size and truncation¶

Every destination limits the size of a payload. For example, a CloudWatch Logs event can be 1 MB and a DynamoDB item 400 KB. Each exporter has a maximum record size. When a record is larger, the plugin trims a copy for that exporter only, so the same record can reach one exporter whole and another trimmed.

The plugin trims in this order until the record fits:

1. Every error cause
2. Operation results, oldest operation first
3. Whole operations, oldest first
4. The execution input, then the output

The plugin never removes the identity and timeline fields. A trimmed record has truncated: true. Each operation that lost its result has truncated: true, and the record counts what it removed in droppedCauses, droppedOperations, droppedInput, and droppedOutput. So you can always tell a trimmed record from a complete one. Content transforms run before truncation, so prefer them to bound input and output on purpose.

The default limits sit below each destination's hard limit, for example 256 KB for CloudWatch Logs and 400 KB for DynamoDB. The HTTP and file exporters have no default limit. Override a limit with the exporter's maximum record size setting.

Custom exporters¶

To send records somewhere else, implement the exporter interface. The plugin calls export with each record. It calls flush at the end of each sampled-in invocation, and never at the same time as export. In TypeScript's on-change mode, an operation change that arrives during or after that flush triggers another flush, so write flush to allow repeated calls. An exporter that sets a maximum record size receives a trimmed copy when a record is larger.

interface TelemetryExporter {
  export(record: DurableExecutionTelemetryRecord): Promise<void>;
  flush?(): Promise<void>;
  maxRecordSizeBytes?: number;
  render?(record: DurableExecutionTelemetryRecord): unknown;
}

class TelemetryExporter(Protocol):
    max_record_size_bytes: int | None

    def render(self, record: dict) -> Any: ...
    def export(self, record: dict) -> None: ...
    def flush(self) -> None: ...

public interface TelemetryExporter {
    void export(DurableExecutionTelemetryRecord record);
    default void flush() {}
    default Integer maxRecordSizeBytes() { return null; }
    default Object render(DurableExecutionTelemetryRecord record) { return record.toWireMap(); }
}

type Exporter interface {
    Export(ctx context.Context, record Record) error
}

Executions that end outside your function¶

The exporter runs inside your function, so it can emit a record only when your function runs. Two terminal states start in the Lambda service without an invocation, so the exporter never sees them. A manual stop ends the execution as STOPPED, and an exceeded execution timeout ends it as TIMED_OUT. To record those states too, add an EventBridge rule on the default bus for Lambda durable execution status changes. Point it at a small function that updates your destination:

{
  "source": ["aws.lambda"],
  "detail-type": ["Durable Execution Status Change"],
  "detail": { "status": ["STOPPED", "TIMED_OUT"] }
}

Naming operations¶

The plugin uses operation names as they are, as keys and identifiers, and never escapes them. A . in a name separates field paths in CloudWatch Logs Insights and OpenSearch, so a name with a dot is hard to query there. Names made of letters, digits, -, and _ work in every destination.

See also¶

• Durable Execution Telemetry
• Durable Execution Telemetry MCP server
• Plugins
• Child Context