Simple ec2 monitoring for opentelemetry

ec2_monitoring.py

import botocore.session
from botocore import monitoring
from opentelemetry.trace import get_tracer
from opentelemetry import trace
from opentelemetry.trace import SpanKind
from opentelemetry.semconv.trace import SpanAttributes
from opentelemetry.trace.status import Status, StatusCode
from opentelemetry.instrumentation.utils import (
    http_status_to_status_code,
)


class OpentelemetryPublisher(object):
    def publish(self, event):
        tracer = get_tracer(__name__)
        span = tracer.start_span(
            f"ec2.{event.operation}",
            kind=SpanKind.CLIENT,
        )

        with trace.use_span(span, end_on_exit=True, record_exception=True) as current_span:
            if current_span.is_recording():
                attributes = {
                    SpanAttributes.HTTP_METHOD: "POST",
                    SpanAttributes.RPC_METHOD: event.operation,
                    SpanAttributes.HTTP_URL: event.url,
                    SpanAttributes.HTTP_STATUS_CODE: event.http_status_code,
                    SpanAttributes.RPC_SYSTEM: "aws-api",
                }
                for key, value in attributes.items():
                    span.set_attribute(key, value)
                span.set_status(
                    Status(http_status_to_status_code(int(event.http_status_code)))
                )
                span.set_attribute(
                    SpanAttributes.HTTP_STATUS_CODE, event.http_status_code
                )


def _monitor_factory():
    return monitoring.Monitor(
        adapter=monitoring.MonitorEventAdapter(),
        publisher=OpentelemetryPublisher()
    )


def get_botocore_session():
    botocore_session = botocore.session.get_session()
    botocore_session._internal_components.lazy_register_component("monitor", _monitor_factory)
    return botocore_session