Skip to content

Instantly share code, notes, and snippets.

@prathamesh-sonpatki
Last active February 9, 2024 10:12
Show Gist options
  • Save prathamesh-sonpatki/a6ae38a88b3fede02f2f8ed71b27d181 to your computer and use it in GitHub Desktop.
Save prathamesh-sonpatki/a6ae38a88b3fede02f2f8ed71b27d181 to your computer and use it in GitHub Desktop.
# 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)
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment