DataEngine supports collection of metrics from data processing functions using OpenTelemetry through the DataEngine SDK. The SDK wraps an OpenTelemetry metrics handler via four methods, supporting four metrics instruments:
Metric type | Method | What it measures |
|---|---|---|
Counter |
| Cumulative values that only increase. |
gauge |
| Asynchronous snapshot of a live metric via a background function. |
histogram |
| Values measured over time grouped into ranges. |
updowncounter |
| Values that can increase or decrease. |
In the DataEngine Dashboard, when a pipeline is actively deployed, you can visualize custom metrics that are emitted by functions in the pipeline.
To implement custom metrics in your function code:
Initialize each custom metric in the init() function. Give the metric a name, a description and a unit of measurement:
def init(ctx: Context): global request_counter, connection_counter, duration_histogram # 1. COUNTER request_counter = ctx.counter( "requests.total", description="Total number of requests processed", unit="", # Dimensionless count (default) ) # 2. UPDOWNCOUNTER connection_counter = ctx.updowncounter( "connections.active", description="Number of currently active connections", unit="", ) # 3. HISTOGRAM: Measure request duration distribution # Custom bucket boundaries tuned for 10-100 ms requests. Histograms without # the boundaries argument keep the OpenTelemetry SDK default boundaries. duration_histogram = ctx.histogram( "request.duration", description="Request processing duration", unit="ms", boundaries=[1, 2, 5, 10, 20, 50, 100, 200, 500], ) # 4. GAUGE ctx.gauge( "memory.usage", callback=get_memory_usage, description="Current memory usage", unit="MB", )In the handler() function, define how to measure the metrics:
def handler(ctx: Context, event: VastEvent): """ Minimal handler demonstrating each metric type. - Counter: Counts each request - UpDownCounter: Tracks active connections - Histogram: Records processing duration - Gauge: Reports memory via callback (automatic) """ global active_connections start_time = time.time() # 1. Start timer # UpDownCounter: Mark that a connection has started active_connections += 1 if connection_counter: connection_counter.add(1) try: data = event.get_data() message = data.get("message", "Hello World") # Counter: Count a new request if request_counter: request_counter.add(1, {"status": "processing"}) # Simulate work processing_time = random.uniform(0.01, 0.1) time.sleep(processing_time) processed_message = message.upper() # 4. Histogram: Record how long the request took (in ms) duration_ms = (time.time() - start_time) * 1000 if duration_histogram: duration_histogram.record(duration_ms) ctx.logger.info(f"Processed in {duration_ms:.2f}ms") return { "original": message, "processed": processed_message, "duration_ms": duration_ms, } finally: # UpDownCounter: Mark that the connection ended (-1) active_connections -= 1 if connection_counter: connection_counter.add(-1)