Flowrensic Python SDK

Pre-release. This package is an early preview and not ready for production. Its API can change in any version.

Flowrensic is a platform that runs beside Temporal. It watches what your workflows and workers do and tells you what went wrong, before and after a deploy.

What it does today

• Connects a Temporal Python worker to your Flowrensic server, with your env access key, so the worker shows up in Flowrensic. Call temporal_flowrensic.client.connect.
• Every call to Flowrensic carries your env, its access key and the ephemeral run in the headers Flowrensic-Current-Env, Flowrensic-Env-Access-Key and Flowrensic-Current-Ephemeral-Run. Each part takes them as current_env, env_access_key and current_ephemeral_run. An env of kind ephemeral takes the ephemeral run, such as the CI build number, and without one the run is default. The SDK reads no environment variable: you pass the values in code.
• Records each activity attempt: its result, error type, attempt number and time, with the timeouts and retry limit it ran under. Register temporal_flowrensic.activity.Interceptor in Worker(interceptors=[...]) and give it a sink from temporal_flowrensic.client.create_attempt_sink. Each attempt carries the namespace Temporal gives the activity, so there is no namespace setting. It never changes what an activity returns or raises.
• The sink keeps the attempts in the worker's memory and sends them to Flowrensic in batches, in the background. Each send has a time limit. When Flowrensic does not answer, the attempts wait and are sent again later. The buffer has a limit on records and on memory, and past it the oldest are dropped and counted.

import temporal_flowrensic.activity
import temporal_flowrensic.client

sink = temporal_flowrensic.client.create_attempt_sink(
    "http://flowrensic:8080", current_env="prod", env_access_key=env_access_key, current_ephemeral_run="prod-1",
    worker_name="orders-worker",
)
worker = Worker(
    client,
    task_queue="main",
    workflows=[...],
    activities=[...],
    interceptors=[temporal_flowrensic.activity.Interceptor(sink, current_env="prod", current_ephemeral_run="prod-1", version=build_id)],
)

• Worker telemetry: temporal_flowrensic.client.start_telemetry says hello, then sends regular beats so Flowrensic knows the worker is up, the engine's and the worker's own metrics, and the history of each run the worker saw. At shutdown it hands the ephemeral run over. When Flowrensic does not answer, the worker still starts and keeps trying in the background. Give telemetry.sink to the activity interceptor.

telemetry = await temporal_flowrensic.client.start_telemetry(
    server_base_url="http://flowrensic:8080",
    current_env="prod",
    env_access_key=env_access_key,
    current_ephemeral_run="prod-1",
    version=build_id,
    namespace="default",
    temporal_client=client,
)
interceptor = temporal_flowrensic.activity.Interceptor(telemetry.sink, current_env="prod", current_ephemeral_run="prod-1", version=build_id)
# ... run the worker ...
await telemetry.shutdown()

• Client calls: temporal_flowrensic.client.ClientInterceptor in Client.connect(plugins=[...]) sends every query and signal that fails, and whether it ended on its deadline, under the namespace of the client. temporal_flowrensic.workflow.Interceptor in Worker(interceptors=[...]) counts the queries your workflows answer. Neither changes what a call returns or raises, or what a workflow does.

• Replay before deploy: in CI, temporal_flowrensic.client.fetch_replay_set writes the histories Flowrensic picks for a workflow type from each env of replay_history_from_envs, such as ["prod", "staging"], asked as your current_env, which must be of kind ephemeral, since a persistent env is refused, you replay them with temporalio's Replayer, and temporal_flowrensic.client.publish_verdicts sends the result back, with the names of the workflow interceptors the replay ran with, as temporal_flowrensic.client.interceptor_name gives them. Give the same interceptors to start_telemetry, so Flowrensic can compare the two lists.

• Deploy reports from CI: temporal_flowrensic.client.report_deploy_stage sends each stage of a deploy. Every CI call has a time limit and stops with a message that says Flowrensic could not be reached or what it answered. A failed deploy report does not fail the job unless you ask it to.

Planned

• Each new part of the Node SDK, @flowrensic/temporal-flowrensic-node-sdk, comes to the Python SDK too.

Install

pip install temporal-flowrensic-python-sdk