The Netra SDK exposes a meter API that lets you:
- Create custom metrics – Counters, histograms, gauges, and more
- Export to Netra backend – Using OpenTelemetry OTLP/HTTP JSON
This page shows how to use meter utility in Netra to define and record your own metrics.
Getting Started
The metrics pipeline is configured automatically when you enable metrics in Netra.init.
from netra import Netra
Netra.init(
app_name="sample-app",
enable_metrics=True,
)
# Access a Meter (OpenTelemetry metrics entry point)
meter = Netra.get_meter("ai-service")
If enable_metrics is False or no OTLP endpoint is configured, Netra installs a no-op MeterProvider. Your code can still call get_meter and record metrics safely, they will simply be discarded.
get_meter
Return an OpenTelemetry Meter instance backed by Netra’s global MeterProvider.
from netra import Netra
Netra.init(
app_name="sample-app",
enable_metrics=True,
)
# Recommended: use your service or module name
meter = Netra.get_meter("payment_service")
Parameters
| Parameter | Type | Description |
|---|
name | str | Instrumentation scope name – typically your service or module name (e.g. "order_service"). Defaults to "netra". |
version | str? | Optional instrumentation scope version string (e.g. "1.0.0"). |
Creating Instruments
The returned Meter is a standard OpenTelemetry Meter and supports the usual instrument types.
Counter
Monotonically increasing value (e.g. request counts).
request_counter = meter.create_counter(
name="http.requests",
description="Number of HTTP requests processed",
unit="1"
)
request_counter.add(
amount=1,
attributes={"route": "/api/health", "status": "ok"}
)
Up Down Counter
Value that can increase or decrease (e.g. active connections).
active_connections = meter.create_up_down_counter(
name="connections.active",
description="Number of active client connections",
unit="1"
)
active_connections.add( # connection opened
amount=1,
attributes={"region": "us-east-1"}
)
active_connections.add( # connection closed
amount=-1,
attributes={"region": "us-east-1"}
)
Histogram
Distribution of measurements (e.g. latency, payload size).
latency = meter.create_histogram(
name="db.query.latency_ms",
unit="ms",
description="Database query latency",
)
latency.record(
amount=15.3,
attributes={"operation": "read", "table": "users"}
)
latency.record(
amount=30.7,
attributes={"operation": "write", "table": "orders"}
)
Observable Instruments (Advanced)
You can also create observable instruments (pull-based metrics), such as periodic gauges or counters, by registering callbacks.
Example for CPU and memory utilization gauges:
import psutil
from opentelemetry.metrics import Observation
def _cpu_utilization_callback(options):
value = psutil.cpu_percent()
yield Observation(
value=value,
attributes={"resource": "cpu"}
)
def _memory_usage_callback(options):
mem = psutil.virtual_memory()
yield Observation(
value=mem.used,
attributes={"unit": "bytes"}
)
cpu_gauge = meter.create_observable_gauge(
name="system.cpu.utilization",
description="CPU utilization percentage",
callbacks=[_cpu_utilization_callback],
)
memory_gauge = meter.create_observable_gauge(
name="system.memory.used_bytes",
description="Used memory in bytes",
callbacks=[_memory_usage_callback],
)
These callbacks are invoked periodically by the metrics reader configured by Netra.
Metrics Export Pipeline
Netra configures the metrics pipeline viaMetricsSetup(), which:
- Uses OTLP/HTTP JSON
Wraps the upstream OTLPMetricExporter to serialize metrics to JSON and send them
- Uses delta temporality for all instruments
Every export contains increments since the last export (delta), matching common observability platforms.
- Respects configuration
Uses app_name, environment, and optional resource_attributes from Netra.init as resource attributes on all metrics.
- Exports periodically
Uses metrics_export_interval_ms to control the export interval.
Complete Example
from netra import Netra
import time
import random
# Initialize Netra with metrics enabled
Netra.init(
app_name="custom-metrics-demo",
environment="staging",
enable_metrics=True,
metrics_export_interval_ms=2000
)
# Get a Meter for your service
meter = Netra.get_meter("checkout_service")
# Define instruments
request_counter = meter.create_counter(
name="checkout.requests",
description="Number of checkout requests",
unit="1"
)
error_counter = meter.create_counter(
name="checkout.errors",
description="Number of failed checkouts",
unit="1"
)
latency_histogram = meter.create_histogram(
name="checkout.latency_ms",
description="Checkout latency in milliseconds",
unit="ms",
)
# Application loop (simplified)
for _ in range(100):
start = time.time()
# Simulate request handling
success = random.random() > 0.1
latency_ms = random.uniform(50, 300)
# Record metrics
request_counter.add(
amount=1,
attributes={"status": "success" if success else "error"}
)
if not success:
error_counter.add(
amount=1,
attributes={"reason": "payment_failed"}
)
latency_histogram.record(
amount=latency_ms,
attributes={"status": "success" if success else "error"},
)
time.sleep(0.1)
Netra.shutdown()
Next Steps
Last modified on March 27, 2026