生产环境Django+Gunicorn+Gevent+OpenTelemetry批量处理器报错求助

有没有人在生产环境中使用Django + Gunicorn + Gevent + OpenTelemetry组合？想了解成功部署运行的方案。目前无法使用BatchSpanProcessor或BatchLogRecordProcessor，遇到报错，相关问题有大量未解决的issue，但没找到可行方案。

启动Gunicorn后会出现如下AssertionError，但服务器仍能接收并处理请求：

Traceback (most recent call last):
  File "src/gevent/_abstract_linkable.py", line 287, in gevent._gevent_c_abstract_linkable.AbstractLinkable._notify_links
  File "src/gevent/_abstract_linkable.py", line 333, in gevent._gevent_c_abstract_linkable.AbstractLinkable._notify_links
AssertionError: (None, <callback at 0xffffb32b0640 args=([],)>)
2025-06-09T14:58:34Z <callback at 0xffffb32b0640 args=([],)> failed with AssertionError

2025-06-09T14:58:34Z <callback at 0xffffb32903c0 args=([],)> failed with AssertionError

当发起大量请求时，会出现影响服务的KeyError：

2025-06-09T14:40:18Z <Greenlet at 0xffff65bb5b20: <bound method Thread._bootstrap of <Thread(OtelBatchSpanRecordProcessor, stopped daemon 281472388520736)>>> failed with KeyError

Traceback (most recent call last):
  File "src/gevent/greenlet.py", line 900, in gevent._gevent_cgreenlet.Greenlet.run
  File "/usr/local/lib/python3.11/threading.py", line 1002, in _bootstrap
    self._bootstrap_inner()
  File "/usr/local/lib/python3.11/threading.py", line 1037, in _bootstrap_inner
    del _limbo[self]
        ~~~~~~^^^^^^
KeyError: <Thread(OtelBatchSpanRecordProcessor, stopped daemon 281472388522016)>

当前实现细节：

• 在wsgi.py中启动OpenTelemetry初始化代码
• Gunicorn的preload app设置为false
• 初始化代码在fork和Gevent猴子补丁完成后执行，已验证

以下是OpenTelemetry埋点的初始化函数：

def init_opentelemetry():
    """
    Why we're not adding this in post_fork method in gunicorn.conf.py?
    When we use gevent worker class with gunicorn, the monkey patching is done after the post_fork method is complete.
    But, if we import a lot of things before that, gevent monkey patching will not work properly and cause errors.
    Hence, had to move this away from post_fork method.
    """
    import logging
    from uuid import uuid4

    from opentelemetry import metrics, trace
    from opentelemetry.exporter.otlp.proto.grpc._log_exporter import (
        OTLPLogExporter,
    )
    from opentelemetry.exporter.otlp.proto.grpc.metric_exporter import (
        OTLPMetricExporter,
    )
    from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import (
        OTLPSpanExporter,
    )

    # support for logs is currently experimental
    from opentelemetry.sdk._logs import LoggerProvider, LoggingHandler
    from opentelemetry.sdk._logs.export import BatchLogRecordProcessor
    from opentelemetry.sdk.environment_variables import OTEL_EXPORTER_OTLP_ENDPOINT
    from opentelemetry.sdk.metrics import MeterProvider
    from opentelemetry.sdk.metrics.export import PeriodicExportingMetricReader
    from opentelemetry.sdk.resources import Resource
    from opentelemetry.sdk.resources import SERVICE_INSTANCE_ID
    from opentelemetry.sdk.trace import TracerProvider
    from opentelemetry.sdk.trace.export import BatchSpanProcessor
    collector_endpoint = os.environ.get(OTEL_EXPORTER_OTLP_ENDPOINT, "")

    resource = Resource.create(
        attributes={
            # each worker needs a unique service.instance.id to distinguish the created metrics in prometheus
            SERVICE_INSTANCE_ID: str(uuid4()),
            # "worker": worker.pid,
            "service.name": API_V2_APP_NAME,
        }
    )

    tracer_provider = TracerProvider(resource=resource)
    tracer_provider.add_span_processor(BatchSpanProcessor(OTLPSpanExporter(endpoint=collector_endpoint)))
    trace.set_tracer_provider(tracer_provider)

    metrics.set_meter_provider(
        MeterProvider(
            resource=resource,
            metric_readers=[PeriodicExportingMetricReader(OTLPMetricExporter(endpoint=collector_endpoint))],
        )
    )

    logger_provider = LoggerProvider(resource=resource)
    logger_provider.add_log_record_processor(BatchLogRecordProcessor(OTLPLogExporter(endpoint=collector_endpoint)))
    from opentelemetry import _logs
    _logs.set_logger_provider(logger_provider)
    logging.getLogger().addHandler(LoggingHandler(level=logging.NOTSET, logger_provider=logger_provider))
    from opentelemetry.instrumentation.django import DjangoInstrumentor
    from opentelemetry.instrumentation.psycopg2 import Psycopg2Instrumentor
    from opentelemetry.instrumentation.requests import RequestsInstrumentor
    from opentelemetry.instrumentation.logging import LoggingInstrumentor

    DjangoInstrumentor().instrument()
    Psycopg2Instrumentor().instrument()
    RequestsInstrumentor().instrument()
    LoggingInstrumentor().instrument(set_logging_format=True)

备注：禁用指标、切换为SimpleSpanProcessor和SimpleLogRecordProcessor可解决报错，但性能表现很差。

内容的提问来源于stack exchange，提问作者Sai Chander