Coverage for backend/CoreRoot/helpers.py: 92%

1# Helper function for getting a header value from an ASGI scope 

2import logging 

3import os 

4 

5from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter 

6from opentelemetry.instrumentation.django import DjangoInstrumentor 

7from opentelemetry.instrumentation.logging import LoggingInstrumentor 

8from opentelemetry.instrumentation.requests import RequestsInstrumentor 

9from opentelemetry.sdk.resources import SERVICE_NAME, Resource 

10from opentelemetry.sdk.trace import TracerProvider 

11from opentelemetry.sdk.trace.export import BatchSpanProcessor, ConsoleSpanExporter 

12from opentelemetry import trace 

13 

14OTLP_ENDPOINT = os.getenv("OTLP_ENDPOINT", "http://localhost:4318/v1/traces") 

15 

16def get_asgi_header_value(headers, header_name): 

17 header_value: bytes | None = None 

18 

19 try: 

20 header_value = next(header_value for header_key, header_value in headers if header_key.decode() == header_name) 

21 except StopIteration: 

22 pass 

23 

24 return header_value.decode() if header_value else None 

25 

26def initialise_tracing(): 

27 RequestsInstrumentor().instrument() 

28 LoggingInstrumentor().instrument(set_logging_format=True) 

29 

30 resource = Resource.create(attributes={ 

31 SERVICE_NAME: os.getenv("OTLP_SERVICE_NAME", "django-api") 

32 }) 

33 

34 tracerProvider = TracerProvider(resource=resource) 

35 processor = BatchSpanProcessor(OTLPSpanExporter(endpoint=OTLP_ENDPOINT)) 

36 tracerProvider.add_span_processor(processor) 

37 trace.set_tracer_provider(tracerProvider)