---
title: "Collecting Custom Metrics"
slug: "collecting-custom-metrics"
tags: ["5.5", "DataEngine"]
updated: 2026-10-27T23:00:54Z
published: 2026-10-27T23:00:54Z
canonical: "docs.vastdata.com/collecting-custom-metrics"
---

> ## Documentation Index
> Fetch the complete documentation index at: https://docs.vastdata.com/llms.txt
> Use this file to discover all available pages before exploring further.

# Collecting Custom Metrics

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 | `ctx.counter` | Cumulative values that only increase. |
| gauge | `ctx.gauge` | Asynchronous snapshot of a live metric via a background function. |
| histogram | `ctx.histogram` | Values measured over time grouped into ranges. |
| updowncounter | `ctx.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:

```plaintext
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:

```plaintext
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)
```