Python Program Failing to Receive Complete Data from OpenTelemetry Collector

[knut-bw]

Problem Description:

I am using FastAPI to receive trace data from OpenTelemetry Collector (otelcol), but I am experiencing data loss. Specifically, each API call should generate one trace record. However, when I use a shell script to rapidly call the API (about ten times per second), I expect to receive ten trace records. Instead, my FastAPI only receives 3-4 trace records.

What I’ve Tried:

Using Redis for caching
Asynchronous processing
Multithreading
None of these attempts resolved the data loss issue.

Environment Information:

Deployed on Kubernetes
Below are my YAML configuration files for the OpenTelemetry Collector and Instrumentation, as well as the FastAPI code (please include the actual code or configurations).

apiVersion: opentelemetry.io/v1alpha1
kind: OpenTelemetryCollector
metadata:
  name: otelcol
spec:
  mode: daemonset
  config: |
    receivers:
      otlp:
        protocols:
          grpc:
            endpoint: 0.0.0.0:4317
          http:
            endpoint: 0.0.0.0:4318
            cors:
              allowed_origins:
                - "http://*"
                - "https://*"
    processors:
      memory_limiter:
        check_interval: 1s
        limit_percentage: 75
        spike_limit_percentage: 15
      batch:
        send_batch_size: 10000
        timeout: 10s
    exporters:
      # NOTE: Prior to v0.86.0 use `logging` instead of `debug`.
      debug:
      otlp/jaeger/A:
        endpoint: "jaeger-collector.default.svc.cluster.local:14250"
        tls:
          insecure: true
      otlp/jaeger/B:
        endpoint: "jaeger-collector.default.svc.cluster.local:4317"
        tls:
          insecure: true
      otlp/python/C:
        endpoint: "10.20.1.230:8003"
        tls:
          insecure: true
      otlp/python/D:
        endpoint: "10.20.1.230:50051"
        tls:
          insecure: true
    service:
      pipelines:
        traces:
          receivers: [otlp]
          processors: [batch]
          exporters: [otlp/jaeger/A, otlp/jaeger/B, otlp/python/C, otlp/python/D, debug]
        metrics:
          receivers: [otlp]
          processors: [batch]
          exporters: [debug]
        logs:
          receivers: [otlp]
          processors: [batch]
          exporters: [debug]

apiVersion: opentelemetry.io/v1alpha1
kind: Instrumentation
metadata:
  name: instrumentation
spec:
  exporter:
    endpoint: http://otelcol-collector.default.svc.cluster.local:4317
    endpoint: http://otelcol-collector.default.svc.cluster.local:4318
  propagators:
    - tracecontext
    - baggage
    - b3
  java:
    image: ghcr.io/open-telemetry/opentelemetry-operator/autoinstrumentation-java:latest
  python:
    image: ghcr.io/open-telemetry/opentelemetry-operator/autoinstrumentation-python:latest
    env:
      - name: OTEL_LOGS_EXPORTER
        value: otlp_proto_http
      - name: OTEL_PYTHON_LOGGING_AUTO_INSTRUMENTATION_ENABLED
        value: 'true'
      - name: OTEL_TRACES_EXPORTER
        value: otlp_proto_http
      - name: OTEL_EXPORTER_OTLP_TRACES_ENDPOINT
        value: http://10.20.1.230:8004/v1/traces
  sampler:
    type: parentbased_traceidratio
    argument: "1"

import asyncio
from fastapi import FastAPI, Request
from opentelemetry.proto.collector.trace.v1 import trace_service_pb2
from google.protobuf.json_format import MessageToDict
import aioredis
import time
from contextlib import asynccontextmanager
import os

REDIS_URL = "redis://127.0.0.1:6379"
BATCH_SIZE = 100
BATCH_TIMEOUT = 1.0  # 1 second

request_queue = asyncio.Queue()

async def get_redis_connection():
    return await aioredis.from_url(REDIS_URL)

@asynccontextmanager
async def lifespan(app: FastAPI):
    process_task = asyncio.create_task(process_batch())
    yield
    process_task.cancel()
    try:
        await process_task
    except asyncio.CancelledError:
        pass

app = FastAPI(lifespan=lifespan)

async def update_stats(redis, worker_id):
    pipe = redis.pipeline()
    pipe.incr(f"worker:{worker_id}:count")
    pipe.incr("total_requests")
    results = await pipe.execute()
    return results[0], results[1]  # worker_count, total_count

@app.post("/v1/traces")
async def receive_traces(request: Request):
    content = await request.body()
    worker_id = os.getpid()
    
    async with await get_redis_connection() as redis:
        worker_count, total_count = await update_stats(redis, worker_id)

    print({
        "status": "received", 
        "worker_id": worker_id,
        "worker_count": worker_count,
        "total_count": total_count
    })
    print("total_count", total_count)
    
    await request_queue.put(content)
    return {"status": "received"}

async def process_batch():
    try:
        async with await get_redis_connection() as redis_conn:
            while True:
                batch = []
                start_time = time.time()

                while len(batch) < BATCH_SIZE and time.time() - start_time < BATCH_TIMEOUT:
                    try:
                        content = await asyncio.wait_for(request_queue.get(), timeout=0.1)
                        batch.append(content)
                    except asyncio.TimeoutError:
                        continue

                if batch:
                    trace_data_list = []
                    for content in batch:
                        trace_data = trace_service_pb2.ExportTraceServiceRequest()
                        trace_data.ParseFromString(content)
                        # print("trace_data", trace_data)
                        print("trace_data")
                        trace_data_list.append(MessageToDict(trace_data))

                    await redis_conn.set("latest_trace", str(trace_data_list[-1]))
    except Exception as e:
        print("ERROR", e)

@app.get("/stats")
async def get_stats():
    async with await get_redis_connection() as redis:
        worker_keys = await redis.keys("worker:*:count")
        
        pipe = redis.pipeline()
        for key in worker_keys:
            pipe.get(key)
        pipe.get("total_requests")
        results = await pipe.execute()
        
        stats = {key.split(":")[1]: int(count) for key, count in zip(worker_keys, results[:-1])}
        total_count = int(results[-1]) if results[-1] else 0
    
    return {
        "worker_stats": stats,
        "total_requests": total_count
    }

if __name__ == "__main__":
    import uvicorn
    uvicorn.run("fast_scaler:app", host="0.0.0.0", port=8004, workers=16)

Does anyone know why the data loss might be occurring? How can I ensure that FastAPI receives all the trace data?