sample-otel-metrics.py

# Inspired from post from Ashok Nagraj
import random
import time
from flask import Flask, request, jsonify

from opentelemetry import metrics
from opentelemetry.sdk.metrics import MeterProvider
from opentelemetry.sdk.metrics.export import PeriodicExportingMetricReader
from opentelemetry.exporter.otlp.proto.grpc.metric_exporter import OTLPMetricExporter
from opentelemetry.sdk.resources import SERVICE_NAME, SERVICE_NAMESPACE, SERVICE_VERSION, Resource

# Service name is required for most backends
resource = Resource(attributes={
    SERVICE_NAME: "dice-roller",
    SERVICE_NAMESPACE: "devx",
    SERVICE_VERSION: "1.0.0"
})

COLLECTOR_ENDPOINT = "http://localhost:4317"
INTERVAL_SEC = 10

# Boiler plate initialization
metric_reader = PeriodicExportingMetricReader(OTLPMetricExporter(endpoint=COLLECTOR_ENDPOINT), INTERVAL_SEC)
provider = MeterProvider(metric_readers=[metric_reader], resource=resource)

# Sets the global default meter provider
metrics.set_meter_provider(provider)

# Creates a meter from the global meter provider
meter = metrics.get_meter("dice-roller", "1.0.0")

# Add instruments
calls = meter.create_counter(name='api_calls')
duration = meter.create_up_down_counter(name='api_duration')
errors = meter.create_counter(name='api_errors')
size = meter.create_histogram(name='response_size')

app = Flask(__name__)

# decorator to add metrics to a function
def add_metrics(func):
    def wrapper(*args, **kwargs):
        attributes = {'path': func.__name__}
        calls.add(1, attributes)

        start_time = time.time()
        result = func(*args, **kwargs)
        end_time = time.time()

        duration.add(end_time - start_time, attributes)
        size.add(len(result), attributes)

        return result
    return wrapper


def do_roll():
    time.sleep(random.randint(1, 3))
    return random.randint(1, 8)


@add_metrics
@app.route('/roll')
def roll():
    roll = do_roll()
    if roll > 6:
        errors.add(1, {'path': '/roll'})
    return jsonify({'roll': roll})


@add_metrics
@app.route('/ping')
def ping():
    message = 'pong'
    return jsonify({'message': message})


if __name__ == '__main__':
    app.run(host='0.0.0.0', port=15000, debug=True)