* [custom_metric](custom_metric) - Custom metric to record the workflow type in the activity schedule to start latency.

This sample deminstrates two things: (1) how to make a custom metric, and (2) how to use an interceptor.

The custom metric in this sample is an activity schedule-to-start-latency with a workflow type tag.

Please see the top-level [README](../README.md) for prerequisites such as Python, uv, starting the local temporal development server, etc.

1. Run the worker with `uv run custom_metric/worker.py`

2. Request execution of the workflow with `uv run custom_metric/starter.py`

3. Go to `http://127.0.0.1:9090/metrics` in your browser

You'll get something like the following:

import time

from temporalio import activity

def print_and_sleep():

print("In the activity.")

time.sleep(1)

import asyncio

import uuid

from temporalio.client import Client

from custom_metric.workflow import StartTwoActivitiesWorkflow

async def main():

client = await Client.connect(

"localhost:7233",

)

await client.start_workflow(

StartTwoActivitiesWorkflow.run,

id="execute-activity-workflow-" + str(uuid.uuid4()),

task_queue="custom-metric-task-queue",

)

if __name__ == "__main__":

asyncio.run(main())

import asyncio

from concurrent.futures import ThreadPoolExecutor

from temporalio import activity

from temporalio.client import Client

from temporalio.runtime import PrometheusConfig, Runtime, TelemetryConfig

from temporalio.worker import (

ActivityInboundInterceptor,

ExecuteActivityInput,

Interceptor,

Worker,

)

from custom_metric.activity import print_and_sleep

from custom_metric.workflow import StartTwoActivitiesWorkflow

class SimpleWorkerInterceptor(Interceptor):

def intercept_activity(

self, next: ActivityInboundInterceptor

) -> ActivityInboundInterceptor:

return CustomScheduleToStartInterceptor(next)

class CustomScheduleToStartInterceptor(ActivityInboundInterceptor):

async def execute_activity(self, input: ExecuteActivityInput):

schedule_to_start = (

activity.info().started_time

- activity.info().current_attempt_scheduled_time

)

# Could do the original schedule time instead of current attempt

# schedule_to_start_second_option = activity.info().started_time - activity.info().scheduled_time

meter = activity.metric_meter()

histogram = meter.create_histogram_timedelta(

"custom_activity_schedule_to_start_latency",

description="Time between activity scheduling and start",

unit="duration",

)

histogram.record(

schedule_to_start, {"workflow_type": activity.info().workflow_type}

)

return await self.next.execute_activity(input)

async def main():

runtime = Runtime(

telemetry=TelemetryConfig(metrics=PrometheusConfig(bind_address="0.0.0.0:9090"))

)

client = await Client.connect(

"localhost:7233",

runtime=runtime,

)

worker = Worker(

client,

task_queue="custom-metric-task-queue",

interceptors=[SimpleWorkerInterceptor()],

workflows=[StartTwoActivitiesWorkflow],

activities=[print_and_sleep],

# only one activity executor with two concurrently scheduled activities

# to force a nontrivial schedule to start times

activity_executor=ThreadPoolExecutor(1),

max_concurrent_activities=1,

)

await worker.run()

if __name__ == "__main__":

asyncio.run(main())

import asyncio

from datetime import timedelta

from temporalio import workflow

with workflow.unsafe.imports_passed_through():

from custom_metric.activity import print_and_sleep

class StartTwoActivitiesWorkflow:

@workflow.run

async def run(self):

# Request two concurrent activities with only one task slot so

# we can see nontrivial schedule to start times.

activity1 = workflow.execute_activity(

print_and_sleep,

start_to_close_timeout=timedelta(seconds=5),

)

activity2 = workflow.execute_activity(

print_and_sleep,

start_to_close_timeout=timedelta(seconds=5),

)

await asyncio.gather(activity1, activity2)

return None

"custom_metric",