mirror of
https://github.com/makeplane/plane.git
synced 2026-09-01 19:48:42 +02:00
fix(api/observability): address PR review — leak, fork-safety, bounded shutdown, gate unification
Review fixes for #9419: - Pin headers on the license telemetry exporter so it can no longer inherit OTEL_EXPORTER_OTLP_HEADERS and ship an operator's APM credential to telemetry.plane.so. An empty dict is falsy and does not suppress the exporters' env fallback, so the value has to be non-empty. - manage.py: apply --settings before the first plane.* import, which otherwise materializes LazySettings from the default module and silently ignored the flag (breaking the local runserver entrypoint). - Defer OTLP exporter creation to worker_process_init on the Celery prefork pool; the gRPC channel is created eagerly and is not fork-safe. - Bound interpreter-exit flushing: shutdown_on_exit=False on both providers plus an atexit hook on the existing bounded flush_otel(). With an unreachable collector a manage.py command now exits in ~3.5s instead of ~63s. - Stamp a per-process service.instance.id so gunicorn workers and prefork children stop exporting colliding cumulative http.server.* streams. - Resolve instrumentors lazily and isolate each import, and take opentelemetry-instrumentation-httpx[instruments] so a missing httpx cannot crash every entrypoint with OTEL_ENABLED=0. - Add a shared is_otel_active() (enabled AND any endpoint var, signal-specific included) used by setup, both settings modules and celery, so the log schema never changes in a process that exports nothing. - Emit the boot banner through an explicit stderr handler and name the observability loggers in LOGGING so disable_existing_loggers stops silencing export errors. - Strip whitespace in _protocol(); treat blank env values as unset so compose's ${VAR:-} interpolation no longer defeats the sampler defaults. - Deployments: give migrator the OTel env, spell out the sampler defaults, and ship OTEL_ENABLED commented out on AIO, where plane.env is exported after container env and clobbered it. - Docs: correct the emitted-metrics list, the log-schema gate, and the filelog pipeline (Docker envelope needs a second json_parser; trace context needs trace_parser, not move). - Tests: fix the conftest env leak (monkeypatch.delenv, delete-only teardown) and cover the new gating, defaults, deferred providers and logging config. Claude-Session: https://claude.ai/code/session_01BpGkdpVLNQ3Ziqcd6zK2qV
This commit is contained in:
Sriram Veeraghanta
@@ -6,11 +6,33 @@
|
||||
import os
|
||||
import sys
|
||||
|
||||
|
||||
def _apply_settings_argument(argv: list[str]) -> None:
|
||||
"""Honor `--settings=...` before anything imports the `plane` package.
|
||||
|
||||
Importing `plane.observability.setup` below runs `plane/__init__.py`, which
|
||||
imports `plane.celery`, which touches `django.conf.settings` at module
|
||||
scope. That materializes Django's LazySettings from DJANGO_SETTINGS_MODULE
|
||||
long before `execute_from_command_line` gets a chance to apply `--settings`,
|
||||
so the flag would be silently ignored (e.g. bin/docker-entrypoint-api-local.sh
|
||||
runs `runserver --settings=plane.settings.local`). Django itself does the
|
||||
same assignment, just later, so applying it here is equivalent.
|
||||
"""
|
||||
for index, arg in enumerate(argv):
|
||||
if arg.startswith("--settings="):
|
||||
os.environ["DJANGO_SETTINGS_MODULE"] = arg.split("=", 1)[1]
|
||||
return
|
||||
if arg == "--settings" and index + 1 < len(argv):
|
||||
os.environ["DJANGO_SETTINGS_MODULE"] = argv[index + 1]
|
||||
return
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
os.environ.setdefault("DJANGO_SETTINGS_MODULE", "plane.settings.production")
|
||||
_apply_settings_argument(sys.argv[1:])
|
||||
|
||||
# Bootstrap OpenTelemetry before Django imports so runserver and
|
||||
# management commands are instrumented too. No-op unless OTEL_ENABLED=1.
|
||||
# management commands are instrumented too. No-op unless OTel is active.
|
||||
from plane.observability.setup import configure_otel
|
||||
|
||||
configure_otel()
|
||||
|
||||
@@ -4,6 +4,7 @@
|
||||
|
||||
# Python imports
|
||||
import os
|
||||
import sys
|
||||
import logging
|
||||
from datetime import timedelta
|
||||
|
||||
@@ -13,6 +14,7 @@ from pythonjsonlogger.json import JsonFormatter
|
||||
from celery.signals import (
|
||||
after_setup_logger,
|
||||
after_setup_task_logger,
|
||||
worker_process_init,
|
||||
worker_process_shutdown,
|
||||
)
|
||||
from celery.schedules import crontab, schedule
|
||||
@@ -24,16 +26,55 @@ from plane.settings.redis import redis_instance
|
||||
os.environ.setdefault("DJANGO_SETTINGS_MODULE", "plane.settings.production")
|
||||
|
||||
# Bootstrap OpenTelemetry before Celery wires up so CeleryInstrumentor can
|
||||
# patch task execution. No-op unless OTEL_ENABLED=1.
|
||||
from plane.observability.setup import configure_otel, flush_otel # noqa: E402
|
||||
from plane.observability.logging import TraceContextFilter, is_otel_enabled # noqa: E402
|
||||
# patch task execution. No-op unless OTel is active.
|
||||
from plane.observability.setup import ( # noqa: E402
|
||||
configure_otel,
|
||||
flush_otel,
|
||||
init_process_providers,
|
||||
)
|
||||
from plane.observability.logging import TraceContextFilter, is_otel_active # noqa: E402
|
||||
|
||||
configure_otel()
|
||||
|
||||
# Whether to trace-correlate worker logs. Uses the same shared is_otel_enabled()
|
||||
def _is_prefork_worker() -> bool:
|
||||
"""True when this process is `celery ... worker` on the (default) prefork pool.
|
||||
|
||||
The prefork pool forks its task children *after* this module is imported.
|
||||
The OTLP gRPC exporter opens its channel eagerly in __init__ and registers
|
||||
no os.register_at_fork handler, so a channel created here in the MainProcess
|
||||
and inherited by a forked child is not safe to export on — child exports can
|
||||
hang or fail nondeterministically. For that pool we defer exporter creation
|
||||
to worker_process_init, which fires inside each child. `celery beat` and the
|
||||
non-forking pools keep the import-time bootstrap.
|
||||
"""
|
||||
argv = sys.argv
|
||||
if "worker" not in argv:
|
||||
return False
|
||||
for index, arg in enumerate(argv):
|
||||
if arg.startswith("--pool="):
|
||||
return arg.split("=", 1)[1] == "prefork"
|
||||
if arg in ("-P", "--pool"):
|
||||
return index + 1 < len(argv) and argv[index + 1] == "prefork"
|
||||
return True # prefork is Celery's default pool
|
||||
|
||||
|
||||
_DEFER_OTEL_PROVIDERS = _is_prefork_worker()
|
||||
|
||||
configure_otel(defer_providers=_DEFER_OTEL_PROVIDERS)
|
||||
|
||||
# Whether to trace-correlate worker logs. Uses the same shared is_otel_active()
|
||||
# gate as the bootstrap (setup.configure_otel) and the Django LOGGING gate, so
|
||||
# every documented OTEL_ENABLED token behaves identically across processes.
|
||||
_OTEL_LOG_ENABLED = is_otel_enabled()
|
||||
# the log schema never changes in a process that exports no telemetry.
|
||||
_OTEL_LOG_ENABLED = is_otel_active()
|
||||
|
||||
|
||||
@worker_process_init.connect
|
||||
def init_otel_in_worker_child(*args, **kwargs):
|
||||
"""Create this child's OTLP exporters after the prefork fork.
|
||||
|
||||
No-op unless configure_otel() deferred them (see _is_prefork_worker).
|
||||
"""
|
||||
init_process_providers()
|
||||
|
||||
|
||||
# Base JSON log fmt (unchanged off-path); the OTel variant appends the
|
||||
# trace-context fields that TraceContextFilter populates.
|
||||
|
||||
@@ -38,6 +38,18 @@ FLUSH_TIMEOUT_MILLIS = 30000
|
||||
EXPORT_INTERVAL_MILLIS = 20000
|
||||
|
||||
|
||||
# Headers pinned on the instance-metrics exporter so it never inherits the
|
||||
# operator's OTLP env. Both OTLP exporters fall back to
|
||||
# OTEL_EXPORTER_OTLP_HEADERS / OTEL_EXPORTER_OTLP_METRICS_HEADERS when the
|
||||
# caller passes no headers, and an empty dict is falsy so it does not suppress
|
||||
# that fallback — the value has to be non-empty. Without this, a self-hoster who
|
||||
# points their own APM at Plane with e.g.
|
||||
# `OTEL_EXPORTER_OTLP_HEADERS=Authorization=Bearer <token>` would have that
|
||||
# credential attached to the instance metrics this task sends to Plane's
|
||||
# telemetry endpoint.
|
||||
_TELEMETRY_EXPORTER_HEADERS = {"x-plane-telemetry": "instance-metrics"}
|
||||
|
||||
|
||||
def _create_otlp_metric_exporter():
|
||||
"""
|
||||
Create OTLP metric exporter based on OTLP_METRICS_PROTOCOL (http or grpc).
|
||||
@@ -53,14 +65,21 @@ def _create_otlp_metric_exporter():
|
||||
|
||||
grpc_endpoint = get_otlp_grpc_endpoint()
|
||||
insecure = os.environ.get("OTEL_EXPORTER_OTLP_METRICS_INSECURE", "").lower() == "true"
|
||||
return GrpcOTLPMetricExporter(endpoint=grpc_endpoint, insecure=insecure)
|
||||
return GrpcOTLPMetricExporter(
|
||||
endpoint=grpc_endpoint,
|
||||
insecure=insecure,
|
||||
headers=_TELEMETRY_EXPORTER_HEADERS,
|
||||
)
|
||||
|
||||
# HTTP fallback
|
||||
from opentelemetry.exporter.otlp.proto.http.metric_exporter import (
|
||||
OTLPMetricExporter as HttpOTLPMetricExporter,
|
||||
)
|
||||
|
||||
return HttpOTLPMetricExporter(endpoint=get_otlp_http_metrics_url())
|
||||
return HttpOTLPMetricExporter(
|
||||
endpoint=get_otlp_http_metrics_url(),
|
||||
headers=_TELEMETRY_EXPORTER_HEADERS,
|
||||
)
|
||||
|
||||
|
||||
def _collect_and_push_metrics() -> None:
|
||||
|
||||
@@ -12,6 +12,10 @@ plane/settings/local.py and plane/settings/production.py) and attached to
|
||||
every handler, so it works regardless of logger propagation settings.
|
||||
configure_otel() does not install it at runtime — Django's dictConfig
|
||||
would wipe a runtime-installed filter when settings are applied.
|
||||
|
||||
This module deliberately depends only on the OpenTelemetry *API* (no SDK, no
|
||||
instrumentation packages) so the Django settings modules can import it cheaply
|
||||
and so the env-gate predicates below have a single home.
|
||||
"""
|
||||
|
||||
import logging
|
||||
@@ -25,6 +29,14 @@ from opentelemetry import trace
|
||||
# behaves identically everywhere.
|
||||
_TRUTHY_VALUES = ("1", "true", "yes", "on")
|
||||
|
||||
# Any of these makes the pinned OTLP exporters able to reach a collector; the
|
||||
# spec lets an operator set only the signal-specific ones.
|
||||
_ENDPOINT_VARS = (
|
||||
"OTEL_EXPORTER_OTLP_ENDPOINT",
|
||||
"OTEL_EXPORTER_OTLP_TRACES_ENDPOINT",
|
||||
"OTEL_EXPORTER_OTLP_METRICS_ENDPOINT",
|
||||
)
|
||||
|
||||
# Trace-context fields appended to the JSON log formatter when OTel is enabled;
|
||||
# populated by TraceContextFilter.
|
||||
_TRACE_LOG_FIELDS = "%(service_name)s %(trace_id)s %(span_id)s %(trace_flags)s"
|
||||
@@ -35,6 +47,23 @@ def is_otel_enabled() -> bool:
|
||||
return os.environ.get("OTEL_ENABLED", "0").strip().lower() in _TRUTHY_VALUES
|
||||
|
||||
|
||||
def has_otel_endpoint() -> bool:
|
||||
"""Return True when at least one OTLP endpoint var is set."""
|
||||
return any(os.environ.get(name, "").strip() for name in _ENDPOINT_VARS)
|
||||
|
||||
|
||||
def is_otel_active() -> bool:
|
||||
"""Return True when OTel will actually be bootstrapped in this process.
|
||||
|
||||
The single predicate shared by setup.configure_otel(), both Django settings
|
||||
modules and plane/celery.py. Gating the log schema on OTEL_ENABLED alone
|
||||
would diverge from the bootstrap gate: `OTEL_ENABLED=1` with no endpoint
|
||||
installs no TracerProvider, so every log line would gain permanently empty
|
||||
trace_id / span_id fields for zero telemetry.
|
||||
"""
|
||||
return is_otel_enabled() and has_otel_endpoint()
|
||||
|
||||
|
||||
class TraceContextFilter(logging.Filter):
|
||||
"""Inject trace_id, span_id, trace_flags, service_name into LogRecord."""
|
||||
|
||||
@@ -53,18 +82,24 @@ class TraceContextFilter(logging.Filter):
|
||||
|
||||
|
||||
def extend_logging_config(logging_config: dict) -> None:
|
||||
"""Trace-correlate a Django LOGGING dict in place. No-op unless OTel is enabled.
|
||||
"""Trace-correlate a Django LOGGING dict in place. No-op unless OTel is active.
|
||||
|
||||
When enabled, extends the JSON formatter's fmt with the trace-context fields
|
||||
When active, extends the JSON formatter's fmt with the trace-context fields
|
||||
and attaches TraceContextFilter to every handler. Attaching at the handler
|
||||
level (rather than the root logger) is required because most plane.* loggers
|
||||
set propagate=False; runtime mutation also wouldn't survive Django's
|
||||
dictConfig. The off path leaves the log schema byte-for-byte unchanged.
|
||||
|
||||
Also names the observability loggers explicitly. Both settings modules apply
|
||||
LOGGING with disable_existing_loggers=True, and by then the bootstrap has
|
||||
already created `plane.observability.*` and `opentelemetry.*` loggers — any
|
||||
logger not named here would be disabled, silencing export errors,
|
||||
instrumentor failures and flush warnings for the rest of the process.
|
||||
|
||||
Shared by settings/local.py and settings/production.py so the two stay in
|
||||
lockstep.
|
||||
"""
|
||||
if not is_otel_enabled():
|
||||
if not is_otel_active():
|
||||
return
|
||||
logging_config["formatters"]["json"]["fmt"] = (
|
||||
"%(levelname)s %(asctime)s %(module)s %(name)s %(message)s " + _TRACE_LOG_FIELDS
|
||||
@@ -73,4 +108,18 @@ def extend_logging_config(logging_config: dict) -> None:
|
||||
"()": "plane.observability.logging.TraceContextFilter",
|
||||
}
|
||||
for handler in logging_config["handlers"].values():
|
||||
handler["filters"] = ["trace_context"]
|
||||
existing = list(handler.get("filters", []))
|
||||
if "trace_context" not in existing:
|
||||
existing.append("trace_context")
|
||||
handler["filters"] = existing
|
||||
|
||||
logging_config["loggers"]["plane.observability"] = {
|
||||
"level": "INFO",
|
||||
"handlers": ["console"],
|
||||
"propagate": False,
|
||||
}
|
||||
logging_config["loggers"]["opentelemetry"] = {
|
||||
"level": "WARNING",
|
||||
"handlers": ["console"],
|
||||
"propagate": False,
|
||||
}
|
||||
|
||||
@@ -4,36 +4,41 @@
|
||||
|
||||
"""OpenTelemetry bootstrap for the Plane API.
|
||||
|
||||
Single entry point: configure_otel(). Idempotent. No-op unless OTEL_ENABLED
|
||||
is truthy and OTEL_EXPORTER_OTLP_ENDPOINT is set.
|
||||
Single entry point: configure_otel(). Idempotent. No-op unless OTel is active
|
||||
(OTEL_ENABLED truthy *and* an OTLP endpoint configured — see
|
||||
plane.observability.logging.is_otel_active).
|
||||
|
||||
SDK and instrumentor imports are deliberately deferred into the functions that
|
||||
need them. This module is imported by manage.py, wsgi.py, asgi.py and celery.py
|
||||
in *every* process, including the OTEL-disabled default; a module-scope
|
||||
``import opentelemetry.instrumentation.httpx`` would make a missing optional
|
||||
dependency (httpx is only declared under that package's ``instruments`` extra)
|
||||
crash every entrypoint even with OTEL_ENABLED=0.
|
||||
"""
|
||||
|
||||
import atexit
|
||||
import logging
|
||||
import os
|
||||
import sys
|
||||
import uuid
|
||||
from importlib import import_module
|
||||
|
||||
from opentelemetry import metrics, trace
|
||||
from opentelemetry.instrumentation.celery import CeleryInstrumentor
|
||||
from opentelemetry.instrumentation.django import DjangoInstrumentor
|
||||
from opentelemetry.instrumentation.httpx import HTTPXClientInstrumentor
|
||||
from opentelemetry.instrumentation.psycopg import PsycopgInstrumentor
|
||||
from opentelemetry.instrumentation.redis import RedisInstrumentor
|
||||
from opentelemetry.instrumentation.requests import RequestsInstrumentor
|
||||
from opentelemetry.sdk.metrics import MeterProvider
|
||||
from opentelemetry.sdk.metrics.export import PeriodicExportingMetricReader
|
||||
from opentelemetry.sdk.resources import Resource
|
||||
from opentelemetry.sdk.trace import TracerProvider
|
||||
from opentelemetry.sdk.trace.export import BatchSpanProcessor
|
||||
|
||||
from plane.observability.logging import is_otel_enabled
|
||||
from plane.observability.logging import has_otel_endpoint, is_otel_enabled
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
_CONFIGURED = False
|
||||
|
||||
# Providers are created per *process*: gunicorn forks workers and the Celery
|
||||
# prefork pool forks task children, and each needs its own exporter channel.
|
||||
_PROVIDERS_READY = False
|
||||
|
||||
# Kept at module scope so flush_otel() can force-flush them on worker shutdown
|
||||
# (prefork children exit via os._exit and skip atexit).
|
||||
_TRACER_PROVIDER: "TracerProvider | None" = None
|
||||
_METER_PROVIDER: "MeterProvider | None" = None
|
||||
_TRACER_PROVIDER = None
|
||||
_METER_PROVIDER = None
|
||||
|
||||
_NOISY_OTEL_LOGGERS = (
|
||||
"opentelemetry",
|
||||
@@ -46,24 +51,49 @@ _NOISY_OTEL_LOGGERS = (
|
||||
|
||||
_HTTP_PROTOCOLS = ("http/protobuf", "http")
|
||||
|
||||
|
||||
def _has_endpoint() -> bool:
|
||||
return bool(os.environ.get("OTEL_EXPORTER_OTLP_ENDPOINT", "").strip())
|
||||
# (module path, class name, instrument() kwargs). Resolved lazily so an import
|
||||
# failure is isolated to the one instrumentor that failed.
|
||||
_INSTRUMENTORS = (
|
||||
("opentelemetry.instrumentation.django", "DjangoInstrumentor", {}),
|
||||
("opentelemetry.instrumentation.celery", "CeleryInstrumentor", {}),
|
||||
("opentelemetry.instrumentation.psycopg", "PsycopgInstrumentor", {"enable_commenter": False}),
|
||||
("opentelemetry.instrumentation.redis", "RedisInstrumentor", {}),
|
||||
("opentelemetry.instrumentation.requests", "RequestsInstrumentor", {}),
|
||||
("opentelemetry.instrumentation.httpx", "HTTPXClientInstrumentor", {}),
|
||||
)
|
||||
|
||||
|
||||
def _protocol() -> str:
|
||||
return os.environ.get("OTEL_EXPORTER_OTLP_PROTOCOL", "grpc").lower()
|
||||
"""Return the normalized OTEL_EXPORTER_OTLP_PROTOCOL value.
|
||||
|
||||
Stripped as well as lowercased: env-file values are taken literally, so a
|
||||
stray trailing space would otherwise fail the _HTTP_PROTOCOLS membership
|
||||
test and silently pick the gRPC exporter for an HTTP endpoint.
|
||||
"""
|
||||
return os.environ.get("OTEL_EXPORTER_OTLP_PROTOCOL", "grpc").strip().lower()
|
||||
|
||||
|
||||
def _setenv_default(key: str, value: str) -> None:
|
||||
"""os.environ.setdefault that also replaces a blank value.
|
||||
|
||||
Compose interpolation like ``${OTEL_TRACES_SAMPLER:-}`` injects an *empty
|
||||
string* rather than leaving the var unset, which makes plain setdefault a
|
||||
no-op and hands the SDK a value it rejects — silently falling back to 100%
|
||||
sampling, or raising on ``float("")`` for the sampler arg.
|
||||
"""
|
||||
if not os.environ.get(key, "").strip():
|
||||
os.environ[key] = value
|
||||
|
||||
|
||||
def _apply_defaults() -> None:
|
||||
"""Set defaults for standard OTEL env vars. Operator overrides win."""
|
||||
os.environ.setdefault("OTEL_SERVICE_NAME", "plane-api")
|
||||
os.environ.setdefault("OTEL_TRACES_SAMPLER", "parentbased_traceidratio")
|
||||
os.environ.setdefault("OTEL_TRACES_SAMPLER_ARG", "0.1")
|
||||
os.environ.setdefault("OTEL_EXPORTER_OTLP_PROTOCOL", "grpc")
|
||||
_setenv_default("OTEL_SERVICE_NAME", "plane-api")
|
||||
_setenv_default("OTEL_TRACES_SAMPLER", "parentbased_traceidratio")
|
||||
_setenv_default("OTEL_TRACES_SAMPLER_ARG", "0.1")
|
||||
_setenv_default("OTEL_EXPORTER_OTLP_PROTOCOL", "grpc")
|
||||
|
||||
|
||||
def _build_resource() -> Resource:
|
||||
def _build_resource():
|
||||
"""Build the OTEL Resource.
|
||||
|
||||
Resource.create() reads OTEL_SERVICE_NAME and OTEL_RESOURCE_ATTRIBUTES from
|
||||
@@ -71,8 +101,16 @@ def _build_resource() -> Resource:
|
||||
from a dedicated OTEL_ENVIRONMENT (or SENTRY_ENVIRONMENT) var. Note the
|
||||
current semconv key is `deployment.environment.name` (not the legacy
|
||||
`deployment.environment`).
|
||||
|
||||
service.instance.id is stamped per process: gunicorn runs N workers without
|
||||
--preload and the Celery prefork pool forks children, so without it every
|
||||
process would export its own cumulative http.server.* streams under one
|
||||
byte-identical resource identity. That breaks OTLP's single-writer rule and
|
||||
makes the resulting counters and histograms unusable in most backends.
|
||||
"""
|
||||
attributes: dict[str, str] = {}
|
||||
from opentelemetry.sdk.resources import Resource
|
||||
|
||||
attributes: dict[str, str] = {"service.instance.id": str(uuid.uuid4())}
|
||||
environment = os.environ.get("OTEL_ENVIRONMENT") or os.environ.get("SENTRY_ENVIRONMENT") or ""
|
||||
if environment:
|
||||
attributes["deployment.environment.name"] = environment
|
||||
@@ -109,22 +147,58 @@ def _create_metric_exporter():
|
||||
return GrpcMetricExporter()
|
||||
|
||||
|
||||
def _setup_tracing(resource: Resource) -> None:
|
||||
def _setup_tracing(resource) -> None:
|
||||
global _TRACER_PROVIDER
|
||||
provider = TracerProvider(resource=resource)
|
||||
from opentelemetry.sdk.trace import TracerProvider
|
||||
from opentelemetry.sdk.trace.export import BatchSpanProcessor
|
||||
|
||||
# shutdown_on_exit=False: the SDK's atexit hook joins the batch worker
|
||||
# thread *unbounded*, and the OTLP exporter retries an unreachable collector
|
||||
# with backoff summing to ~1 minute per batch. That would stall every
|
||||
# short-lived manage.py command at boot and every recycled gunicorn worker.
|
||||
# _init_providers() registers the bounded flush_otel() instead.
|
||||
provider = TracerProvider(resource=resource, shutdown_on_exit=False)
|
||||
provider.add_span_processor(BatchSpanProcessor(_create_span_exporter()))
|
||||
trace.set_tracer_provider(provider)
|
||||
_TRACER_PROVIDER = provider
|
||||
|
||||
|
||||
def _setup_metrics(resource: Resource) -> None:
|
||||
def _setup_metrics(resource) -> None:
|
||||
global _METER_PROVIDER
|
||||
from opentelemetry.sdk.metrics import MeterProvider
|
||||
from opentelemetry.sdk.metrics.export import PeriodicExportingMetricReader
|
||||
|
||||
reader = PeriodicExportingMetricReader(_create_metric_exporter())
|
||||
provider = MeterProvider(resource=resource, metric_readers=[reader])
|
||||
# shutdown_on_exit=False for the same reason as the tracer provider above.
|
||||
provider = MeterProvider(resource=resource, metric_readers=[reader], shutdown_on_exit=False)
|
||||
metrics.set_meter_provider(provider)
|
||||
_METER_PROVIDER = provider
|
||||
|
||||
|
||||
def _init_providers() -> None:
|
||||
"""Create the exporters + providers for the *current* process. Idempotent."""
|
||||
global _PROVIDERS_READY
|
||||
if _PROVIDERS_READY:
|
||||
return
|
||||
resource = _build_resource()
|
||||
_setup_tracing(resource)
|
||||
_setup_metrics(resource)
|
||||
atexit.register(flush_otel)
|
||||
_PROVIDERS_READY = True
|
||||
|
||||
|
||||
def init_process_providers() -> None:
|
||||
"""Create this process's providers after a fork.
|
||||
|
||||
Called from Celery's worker_process_init when configure_otel() was asked to
|
||||
defer provider creation (see configure_otel's defer_providers). No-op when
|
||||
OTel is not configured or the providers already exist.
|
||||
"""
|
||||
if not _CONFIGURED:
|
||||
return
|
||||
_init_providers()
|
||||
|
||||
|
||||
def _instrument_libraries() -> None:
|
||||
"""Patch Django + Celery + downstream client libraries.
|
||||
|
||||
@@ -135,23 +209,16 @@ def _instrument_libraries() -> None:
|
||||
The others add child spans so latency can be decomposed (SQL query,
|
||||
Redis op, outbound HTTP).
|
||||
|
||||
Each instrumentor is isolated: a single failure (version mismatch, missing
|
||||
optional dependency) is logged and skipped so it neither blocks the other
|
||||
instrumentors nor takes down application startup.
|
||||
Each instrumentor is isolated — *including its import* — so a single failure
|
||||
(version mismatch, missing optional dependency) is logged and skipped
|
||||
without blocking the other instrumentors or application startup.
|
||||
"""
|
||||
instrumentors = (
|
||||
(DjangoInstrumentor, {}),
|
||||
(CeleryInstrumentor, {}),
|
||||
(PsycopgInstrumentor, {"enable_commenter": False}),
|
||||
(RedisInstrumentor, {}),
|
||||
(RequestsInstrumentor, {}),
|
||||
(HTTPXClientInstrumentor, {}),
|
||||
)
|
||||
for instrumentor_cls, kwargs in instrumentors:
|
||||
for module_path, class_name, kwargs in _INSTRUMENTORS:
|
||||
try:
|
||||
instrumentor_cls = getattr(import_module(module_path), class_name)
|
||||
instrumentor_cls().instrument(**kwargs)
|
||||
except Exception as exc:
|
||||
logger.warning("Failed to instrument %s: %s", instrumentor_cls.__name__, exc)
|
||||
logger.warning("Failed to instrument %s: %s", class_name, exc)
|
||||
|
||||
|
||||
def _quiet_otel_loggers() -> None:
|
||||
@@ -164,14 +231,41 @@ def _quiet_otel_loggers() -> None:
|
||||
logging.getLogger(name).setLevel(logging.WARNING)
|
||||
|
||||
|
||||
def configure_otel() -> None:
|
||||
def _log_boot_banner(message: str, *args) -> None:
|
||||
"""Emit a one-off INFO line before Django has configured logging.
|
||||
|
||||
configure_otel() runs before dictConfig in every entrypoint, so the root
|
||||
logger has no handlers and logging.lastResort only emits WARNING and above —
|
||||
a plain logger.info() here would be dropped. Attach a temporary stderr
|
||||
handler for this single record so the boot confirmation the README tells
|
||||
operators to look for actually appears.
|
||||
"""
|
||||
handler = logging.StreamHandler(sys.stderr)
|
||||
handler.setFormatter(logging.Formatter("%(levelname)s %(asctime)s %(name)s %(message)s"))
|
||||
previous_level = logger.level
|
||||
logger.addHandler(handler)
|
||||
logger.setLevel(logging.INFO)
|
||||
try:
|
||||
logger.info(message, *args)
|
||||
finally:
|
||||
logger.removeHandler(handler)
|
||||
logger.setLevel(previous_level)
|
||||
handler.close()
|
||||
|
||||
|
||||
def configure_otel(*, defer_providers: bool = False) -> None:
|
||||
"""Configure OpenTelemetry tracing + metrics.
|
||||
|
||||
No-op unless OTEL_ENABLED is truthy and OTEL_EXPORTER_OTLP_ENDPOINT is set.
|
||||
No-op unless OTEL_ENABLED is truthy and an OTLP endpoint is configured.
|
||||
Idempotent — safe to call from wsgi.py, asgi.py, manage.py, and celery.py.
|
||||
|
||||
Pass defer_providers=True when this process will fork workers that do the
|
||||
actual exporting (the Celery prefork pool): instrumentation and env defaults
|
||||
are applied here, but the exporters — whose gRPC channel is not fork-safe —
|
||||
are left to init_process_providers() in each child.
|
||||
|
||||
Log correlation is wired separately via Django's LOGGING dict (see
|
||||
plane/settings/local.py and production.py) and, for Celery, via the
|
||||
plane.observability.logging.extend_logging_config) and, for Celery, via the
|
||||
after_setup_logger handlers in plane/celery.py.
|
||||
"""
|
||||
global _CONFIGURED
|
||||
@@ -179,36 +273,46 @@ def configure_otel() -> None:
|
||||
return
|
||||
if not is_otel_enabled():
|
||||
return
|
||||
if not _has_endpoint():
|
||||
logger.warning("OTEL_ENABLED=1 but OTEL_EXPORTER_OTLP_ENDPOINT is not set; OpenTelemetry bootstrap skipped")
|
||||
if not has_otel_endpoint():
|
||||
logger.warning(
|
||||
"OTEL_ENABLED=1 but OTEL_EXPORTER_OTLP_ENDPOINT is not set "
|
||||
"(nor OTEL_EXPORTER_OTLP_TRACES_ENDPOINT / OTEL_EXPORTER_OTLP_METRICS_ENDPOINT); "
|
||||
"OpenTelemetry bootstrap skipped"
|
||||
)
|
||||
return
|
||||
|
||||
_apply_defaults()
|
||||
resource = _build_resource()
|
||||
_setup_tracing(resource)
|
||||
_setup_metrics(resource)
|
||||
if not defer_providers:
|
||||
_init_providers()
|
||||
_instrument_libraries()
|
||||
_quiet_otel_loggers()
|
||||
|
||||
_CONFIGURED = True
|
||||
logger.info(
|
||||
"OpenTelemetry configured: service=%s, endpoint=%s, protocol=%s, sampler=%s(%s)",
|
||||
_log_boot_banner(
|
||||
"OpenTelemetry configured: service=%s, endpoint=%s, protocol=%s, sampler=%s(%s), providers=%s",
|
||||
os.environ.get("OTEL_SERVICE_NAME"),
|
||||
os.environ.get("OTEL_EXPORTER_OTLP_ENDPOINT"),
|
||||
os.environ.get("OTEL_EXPORTER_OTLP_PROTOCOL"),
|
||||
os.environ.get("OTEL_TRACES_SAMPLER"),
|
||||
os.environ.get("OTEL_TRACES_SAMPLER_ARG"),
|
||||
"deferred to worker children" if defer_providers else "ready",
|
||||
)
|
||||
|
||||
|
||||
def flush_otel(timeout_millis: int = 3000) -> None:
|
||||
"""Best-effort bounded flush of buffered spans + metrics.
|
||||
|
||||
Celery prefork children exit via os._exit and never run atexit hooks, so the
|
||||
worker_process_shutdown signal calls this to keep the tail of each child's
|
||||
spans/metrics from being dropped on --max-tasks-per-child recycling and warm
|
||||
shutdown. Never raises and is bounded by the timeout, so it cannot stall
|
||||
child replacement.
|
||||
Used in two places, both of which must not stall:
|
||||
|
||||
- Celery prefork children exit via os._exit and never run atexit hooks, so
|
||||
the worker_process_shutdown signal calls this to keep the tail of each
|
||||
child's spans/metrics from being dropped on --max-tasks-per-child
|
||||
recycling and warm shutdown.
|
||||
- Registered as this process's atexit hook in place of the SDK's own
|
||||
unbounded provider shutdown (see _setup_tracing).
|
||||
|
||||
Never raises and is bounded by the timeout, so it cannot stall interpreter
|
||||
exit or child replacement even when the collector is unreachable.
|
||||
"""
|
||||
if _TRACER_PROVIDER is not None:
|
||||
try:
|
||||
|
||||
@@ -9,10 +9,15 @@
|
||||
|
||||
- `_CONFIGURED` is reset before every test (otherwise the second test
|
||||
would always be a no-op because the first one flipped the flag).
|
||||
- Every `OTEL_*` env var is stripped before the test and restored after,
|
||||
so what `_apply_defaults()` writes in one test doesn't leak into the
|
||||
next (it uses `os.environ.setdefault`, which bypasses monkeypatch's
|
||||
bookkeeping).
|
||||
- Every `OTEL_*` env var is stripped before the test, so what
|
||||
`_apply_defaults()` writes in one test doesn't leak into the next (it
|
||||
writes to `os.environ` directly, bypassing monkeypatch's bookkeeping).
|
||||
|
||||
The stripping goes through `monkeypatch.delenv` rather than a manual
|
||||
save/restore: this fixture depends on `monkeypatch`, so `monkeypatch.undo()`
|
||||
runs *after* any post-yield teardown here. A manual restore would be undone
|
||||
again by that later `undo()`, permanently stripping any `OTEL_*` var the
|
||||
developer had exported in their shell.
|
||||
"""
|
||||
|
||||
import os
|
||||
@@ -25,15 +30,18 @@ def isolate_otel_state(monkeypatch):
|
||||
from plane.observability import setup as otel_setup
|
||||
|
||||
monkeypatch.setattr(otel_setup, "_CONFIGURED", False, raising=False)
|
||||
monkeypatch.setattr(otel_setup, "_PROVIDERS_READY", False, raising=False)
|
||||
monkeypatch.setattr(otel_setup, "_TRACER_PROVIDER", None, raising=False)
|
||||
monkeypatch.setattr(otel_setup, "_METER_PROVIDER", None, raising=False)
|
||||
|
||||
saved = {k: v for k, v in os.environ.items() if k.startswith("OTEL_")}
|
||||
for key in list(saved.keys()):
|
||||
del os.environ[key]
|
||||
for key in [k for k in os.environ if k.startswith("OTEL_")]:
|
||||
monkeypatch.delenv(key, raising=False)
|
||||
|
||||
yield
|
||||
|
||||
# Drop whatever `_apply_defaults()` wrote directly into os.environ during
|
||||
# the test — monkeypatch knows nothing about those keys and would leave
|
||||
# them behind. This is a delete-only teardown: the developer's original
|
||||
# values are restored by monkeypatch.undo(), which runs after this.
|
||||
for key in [k for k in os.environ if k.startswith("OTEL_")]:
|
||||
del os.environ[key]
|
||||
os.environ.update(saved)
|
||||
|
||||
@@ -4,12 +4,17 @@
|
||||
|
||||
"""Tests for plane.observability.logging.TraceContextFilter."""
|
||||
|
||||
import copy
|
||||
import logging
|
||||
|
||||
import pytest
|
||||
from opentelemetry.sdk.trace import TracerProvider
|
||||
|
||||
from plane.observability.logging import TraceContextFilter
|
||||
from plane.observability.logging import (
|
||||
TraceContextFilter,
|
||||
extend_logging_config,
|
||||
is_otel_active,
|
||||
)
|
||||
|
||||
|
||||
def _make_record() -> logging.LogRecord:
|
||||
@@ -78,3 +83,110 @@ def test_trace_flags_is_int_for_log_record_compat():
|
||||
record = _make_record()
|
||||
TraceContextFilter().filter(record)
|
||||
assert isinstance(record.trace_flags, int)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# extend_logging_config
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def _sample_logging_config() -> dict:
|
||||
"""A trimmed stand-in for the LOGGING dict in plane/settings/*.py."""
|
||||
return {
|
||||
"version": 1,
|
||||
"disable_existing_loggers": True,
|
||||
"formatters": {
|
||||
"json": {
|
||||
"()": "pythonjsonlogger.json.JsonFormatter",
|
||||
"fmt": "%(levelname)s %(asctime)s %(module)s %(name)s %(message)s",
|
||||
}
|
||||
},
|
||||
"handlers": {"console": {"level": "DEBUG", "class": "logging.StreamHandler", "formatter": "json"}},
|
||||
"loggers": {"plane.api": {"level": "INFO", "handlers": ["console"], "propagate": False}},
|
||||
}
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_extend_logging_config_is_a_noop_when_disabled():
|
||||
config = _sample_logging_config()
|
||||
before = copy.deepcopy(config)
|
||||
extend_logging_config(config)
|
||||
assert config == before
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_extend_logging_config_is_a_noop_when_enabled_without_endpoint(monkeypatch):
|
||||
# OTEL_ENABLED=1 with no endpoint installs no TracerProvider, so the log
|
||||
# schema must not change — otherwise every line gains empty trace ids.
|
||||
monkeypatch.setenv("OTEL_ENABLED", "1")
|
||||
config = _sample_logging_config()
|
||||
before = copy.deepcopy(config)
|
||||
extend_logging_config(config)
|
||||
assert config == before
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_extend_logging_config_adds_trace_fields_when_active(monkeypatch):
|
||||
monkeypatch.setenv("OTEL_ENABLED", "1")
|
||||
monkeypatch.setenv("OTEL_EXPORTER_OTLP_ENDPOINT", "http://localhost:4317")
|
||||
config = _sample_logging_config()
|
||||
|
||||
extend_logging_config(config)
|
||||
|
||||
assert "%(trace_id)s" in config["formatters"]["json"]["fmt"]
|
||||
assert "trace_context" in config["filters"]
|
||||
assert config["handlers"]["console"]["filters"] == ["trace_context"]
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_extend_logging_config_activates_on_signal_specific_endpoint(monkeypatch):
|
||||
monkeypatch.setenv("OTEL_ENABLED", "1")
|
||||
monkeypatch.setenv("OTEL_EXPORTER_OTLP_TRACES_ENDPOINT", "http://localhost:4318/v1/traces")
|
||||
config = _sample_logging_config()
|
||||
|
||||
extend_logging_config(config)
|
||||
|
||||
assert "trace_context" in config["filters"]
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_extend_logging_config_preserves_existing_handler_filters(monkeypatch):
|
||||
monkeypatch.setenv("OTEL_ENABLED", "1")
|
||||
monkeypatch.setenv("OTEL_EXPORTER_OTLP_ENDPOINT", "http://localhost:4317")
|
||||
config = _sample_logging_config()
|
||||
config["handlers"]["console"]["filters"] = ["require_debug_true"]
|
||||
|
||||
extend_logging_config(config)
|
||||
|
||||
assert config["handlers"]["console"]["filters"] == ["require_debug_true", "trace_context"]
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_extend_logging_config_keeps_observability_loggers_alive(monkeypatch):
|
||||
# disable_existing_loggers=True would otherwise silence the bootstrap and
|
||||
# exporter loggers created before dictConfig runs.
|
||||
monkeypatch.setenv("OTEL_ENABLED", "1")
|
||||
monkeypatch.setenv("OTEL_EXPORTER_OTLP_ENDPOINT", "http://localhost:4317")
|
||||
config = _sample_logging_config()
|
||||
|
||||
extend_logging_config(config)
|
||||
|
||||
assert "plane.observability" in config["loggers"]
|
||||
assert "opentelemetry" in config["loggers"]
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
@pytest.mark.parametrize(
|
||||
"enabled,endpoint,expected",
|
||||
[
|
||||
("1", "http://localhost:4317", True),
|
||||
("1", "", False),
|
||||
("0", "http://localhost:4317", False),
|
||||
("on", "http://localhost:4317", True),
|
||||
],
|
||||
)
|
||||
def test_is_otel_active(enabled, endpoint, expected, monkeypatch):
|
||||
monkeypatch.setenv("OTEL_ENABLED", enabled)
|
||||
if endpoint:
|
||||
monkeypatch.setenv("OTEL_EXPORTER_OTLP_ENDPOINT", endpoint)
|
||||
assert is_otel_active() is expected
|
||||
|
||||
@@ -7,9 +7,11 @@
|
||||
Strategy: mock at the boundary. The OTLP exporter factories and the
|
||||
opentelemetry global setters (`set_tracer_provider`, `set_meter_provider`)
|
||||
are patched so tests don't open real gRPC connections or mutate global
|
||||
OTEL state. Instrumentor classes are patched at the same point — the
|
||||
real DjangoInstrumentor is a module-level singleton that would survive
|
||||
the test process and pollute other suites.
|
||||
OTEL state. Instrumentor classes are patched in their own packages — setup.py
|
||||
resolves them lazily via import_module, so there is nothing to patch on
|
||||
`plane.observability.setup` itself — because the real DjangoInstrumentor is a
|
||||
module-level singleton that would survive the test process and pollute other
|
||||
suites.
|
||||
"""
|
||||
|
||||
import logging
|
||||
@@ -20,24 +22,30 @@ from unittest.mock import patch
|
||||
import pytest
|
||||
|
||||
from plane.observability.setup import (
|
||||
_build_resource,
|
||||
_create_metric_exporter,
|
||||
_create_span_exporter,
|
||||
_protocol,
|
||||
configure_otel,
|
||||
init_process_providers,
|
||||
)
|
||||
|
||||
|
||||
_INSTRUMENTOR_TARGETS = (
|
||||
"opentelemetry.instrumentation.django.DjangoInstrumentor",
|
||||
"opentelemetry.instrumentation.celery.CeleryInstrumentor",
|
||||
"opentelemetry.instrumentation.psycopg.PsycopgInstrumentor",
|
||||
"opentelemetry.instrumentation.redis.RedisInstrumentor",
|
||||
"opentelemetry.instrumentation.requests.RequestsInstrumentor",
|
||||
"opentelemetry.instrumentation.httpx.HTTPXClientInstrumentor",
|
||||
)
|
||||
|
||||
_MOCK_TARGETS = (
|
||||
"plane.observability.setup._create_span_exporter",
|
||||
"plane.observability.setup._create_metric_exporter",
|
||||
"opentelemetry.trace.set_tracer_provider",
|
||||
"opentelemetry.metrics.set_meter_provider",
|
||||
"plane.observability.setup.DjangoInstrumentor",
|
||||
"plane.observability.setup.CeleryInstrumentor",
|
||||
"plane.observability.setup.PsycopgInstrumentor",
|
||||
"plane.observability.setup.RedisInstrumentor",
|
||||
"plane.observability.setup.RequestsInstrumentor",
|
||||
"plane.observability.setup.HTTPXClientInstrumentor",
|
||||
)
|
||||
) + _INSTRUMENTOR_TARGETS
|
||||
|
||||
|
||||
def _full_enabled_env(monkeypatch):
|
||||
@@ -124,14 +132,7 @@ def test_enabled_with_endpoint_sets_providers_and_instruments(monkeypatch):
|
||||
mocks["opentelemetry.trace.set_tracer_provider"].assert_called_once()
|
||||
mocks["opentelemetry.metrics.set_meter_provider"].assert_called_once()
|
||||
|
||||
for instrumentor_target in (
|
||||
"plane.observability.setup.DjangoInstrumentor",
|
||||
"plane.observability.setup.CeleryInstrumentor",
|
||||
"plane.observability.setup.PsycopgInstrumentor",
|
||||
"plane.observability.setup.RedisInstrumentor",
|
||||
"plane.observability.setup.RequestsInstrumentor",
|
||||
"plane.observability.setup.HTTPXClientInstrumentor",
|
||||
):
|
||||
for instrumentor_target in _INSTRUMENTOR_TARGETS:
|
||||
mocks[instrumentor_target].return_value.instrument.assert_called_once()
|
||||
|
||||
|
||||
@@ -140,7 +141,7 @@ def test_psycopg_instrumentor_called_with_enable_commenter_false(monkeypatch):
|
||||
_full_enabled_env(monkeypatch)
|
||||
mocks = _patched_configure_otel()
|
||||
mocks[
|
||||
"plane.observability.setup.PsycopgInstrumentor"
|
||||
"opentelemetry.instrumentation.psycopg.PsycopgInstrumentor"
|
||||
].return_value.instrument.assert_called_once_with(enable_commenter=False)
|
||||
|
||||
|
||||
@@ -209,7 +210,7 @@ def test_idempotent_when_called_twice(monkeypatch):
|
||||
assert mocks["plane.observability.setup._create_span_exporter"].call_count == 1
|
||||
assert mocks["opentelemetry.trace.set_tracer_provider"].call_count == 1
|
||||
assert (
|
||||
mocks["plane.observability.setup.DjangoInstrumentor"]
|
||||
mocks["opentelemetry.instrumentation.django.DjangoInstrumentor"]
|
||||
.return_value.instrument.call_count
|
||||
== 1
|
||||
)
|
||||
@@ -281,3 +282,95 @@ def test_create_metric_exporter_uses_http_when_protocol_is_http(protocol, monkey
|
||||
_create_metric_exporter()
|
||||
Http.assert_called_once()
|
||||
Grpc.assert_not_called()
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Empty-string env values (compose injects these for unset host vars)
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
@pytest.mark.parametrize(
|
||||
"key,expected",
|
||||
[
|
||||
("OTEL_TRACES_SAMPLER", "parentbased_traceidratio"),
|
||||
("OTEL_TRACES_SAMPLER_ARG", "0.1"),
|
||||
("OTEL_SERVICE_NAME", "plane-api"),
|
||||
("OTEL_EXPORTER_OTLP_PROTOCOL", "grpc"),
|
||||
],
|
||||
)
|
||||
def test_blank_env_value_is_replaced_by_default(key, expected, monkeypatch):
|
||||
# `${OTEL_TRACES_SAMPLER:-}` in docker-compose sets the var to "", which
|
||||
# plain os.environ.setdefault would leave in place.
|
||||
_full_enabled_env(monkeypatch)
|
||||
monkeypatch.setenv(key, "")
|
||||
|
||||
_patched_configure_otel()
|
||||
|
||||
assert os.environ[key] == expected
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_protocol_strips_surrounding_whitespace(monkeypatch):
|
||||
monkeypatch.setenv("OTEL_EXPORTER_OTLP_PROTOCOL", " http/protobuf ")
|
||||
assert _protocol() == "http/protobuf"
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Resource identity
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_resource_carries_a_unique_service_instance_id():
|
||||
# Every gunicorn worker / prefork child must export under its own identity
|
||||
# or the cumulative http.server.* streams collide in the backend.
|
||||
first = _build_resource().attributes["service.instance.id"]
|
||||
second = _build_resource().attributes["service.instance.id"]
|
||||
assert first and second and first != second
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Deferred providers (Celery prefork pool)
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_defer_providers_instruments_but_creates_no_exporters(monkeypatch):
|
||||
_full_enabled_env(monkeypatch)
|
||||
|
||||
stack = ExitStack()
|
||||
mocks = {target: stack.enter_context(patch(target)) for target in _MOCK_TARGETS}
|
||||
try:
|
||||
configure_otel(defer_providers=True)
|
||||
finally:
|
||||
stack.close()
|
||||
|
||||
mocks["plane.observability.setup._create_span_exporter"].assert_not_called()
|
||||
mocks["plane.observability.setup._create_metric_exporter"].assert_not_called()
|
||||
mocks["opentelemetry.instrumentation.celery.CeleryInstrumentor"].return_value.instrument.assert_called_once()
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_init_process_providers_creates_deferred_exporters(monkeypatch):
|
||||
_full_enabled_env(monkeypatch)
|
||||
|
||||
stack = ExitStack()
|
||||
mocks = {target: stack.enter_context(patch(target)) for target in _MOCK_TARGETS}
|
||||
try:
|
||||
configure_otel(defer_providers=True)
|
||||
init_process_providers()
|
||||
# Idempotent: a second call must not build a second exporter.
|
||||
init_process_providers()
|
||||
finally:
|
||||
stack.close()
|
||||
|
||||
mocks["plane.observability.setup._create_span_exporter"].assert_called_once()
|
||||
mocks["plane.observability.setup._create_metric_exporter"].assert_called_once()
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_init_process_providers_is_a_noop_when_otel_is_disabled():
|
||||
with patch("plane.observability.setup._create_span_exporter") as create_exporter:
|
||||
init_process_providers()
|
||||
create_exporter.assert_not_called()
|
||||
|
||||
@@ -80,7 +80,11 @@ opentelemetry-instrumentation-celery==0.49b1
|
||||
opentelemetry-instrumentation-psycopg==0.49b1
|
||||
opentelemetry-instrumentation-redis==0.49b1
|
||||
opentelemetry-instrumentation-requests==0.49b1
|
||||
opentelemetry-instrumentation-httpx==0.49b1
|
||||
# [instruments] extra: this package imports httpx at module scope but declares it
|
||||
# only under that extra. Requesting it keeps the instrumentor from depending on
|
||||
# httpx arriving transitively (today only via openai), without adding a top-level
|
||||
# httpx pin that would clash with the one in requirements/test.txt.
|
||||
opentelemetry-instrumentation-httpx[instruments]==0.49b1
|
||||
# OpenAPI Specification
|
||||
drf-spectacular==0.28.0
|
||||
# html sanitizer
|
||||
|
||||
@@ -75,7 +75,11 @@ WEBHOOK_ALLOWED_HOSTS=
|
||||
|
||||
# OpenTelemetry APM (self-hoster observability). Off by default.
|
||||
# See docs/otel-api-observability/README.md.
|
||||
OTEL_ENABLED=0
|
||||
# Left commented out on purpose: start.sh exports this file *after* the
|
||||
# container env is applied, so an uncommented OTEL_ENABLED=0 would clobber a
|
||||
# `docker run -e OTEL_ENABLED=1 ...` and make OTel impossible to turn on.
|
||||
# The code already defaults to disabled when the var is unset.
|
||||
# OTEL_ENABLED=0
|
||||
# OTEL_EXPORTER_OTLP_ENDPOINT=http://otel-collector:4317
|
||||
# OTEL_EXPORTER_OTLP_PROTOCOL=grpc
|
||||
# OTEL_EXPORTER_OTLP_HEADERS=
|
||||
|
||||
@@ -61,8 +61,11 @@ x-app-env: &app-env
|
||||
WEBHOOK_ALLOWED_IPS: ${WEBHOOK_ALLOWED_IPS:-}
|
||||
WEBHOOK_ALLOWED_HOSTS: ${WEBHOOK_ALLOWED_HOSTS:-}
|
||||
|
||||
# OpenTelemetry APM for the Django API (api, worker, beat-worker).
|
||||
# OpenTelemetry APM for the Django API (api, worker, beat-worker, migrator).
|
||||
# Read at container RUNTIME; off by default. See docs/otel-api-observability/README.md.
|
||||
# Note the sampler defaults are spelled out here rather than left as `${VAR:-}`:
|
||||
# an unset host var would otherwise be injected as an empty string, which the
|
||||
# OTEL SDK rejects (silently falling back to 100% sampling).
|
||||
x-otel-env: &otel-env
|
||||
OTEL_ENABLED: ${OTEL_ENABLED:-0}
|
||||
OTEL_EXPORTER_OTLP_ENDPOINT: ${OTEL_EXPORTER_OTLP_ENDPOINT:-}
|
||||
@@ -70,8 +73,8 @@ x-otel-env: &otel-env
|
||||
OTEL_EXPORTER_OTLP_HEADERS: ${OTEL_EXPORTER_OTLP_HEADERS:-}
|
||||
OTEL_SERVICE_NAME: ${OTEL_SERVICE_NAME:-plane-api}
|
||||
OTEL_ENVIRONMENT: ${OTEL_ENVIRONMENT:-}
|
||||
OTEL_TRACES_SAMPLER: ${OTEL_TRACES_SAMPLER:-}
|
||||
OTEL_TRACES_SAMPLER_ARG: ${OTEL_TRACES_SAMPLER_ARG:-}
|
||||
OTEL_TRACES_SAMPLER: ${OTEL_TRACES_SAMPLER:-parentbased_traceidratio}
|
||||
OTEL_TRACES_SAMPLER_ARG: ${OTEL_TRACES_SAMPLER_ARG:-0.1}
|
||||
OTEL_RESOURCE_ATTRIBUTES: ${OTEL_RESOURCE_ATTRIBUTES:-}
|
||||
|
||||
services:
|
||||
@@ -178,7 +181,7 @@ services:
|
||||
volumes:
|
||||
- logs_migrator:/code/plane/logs
|
||||
environment:
|
||||
<<: [*app-env, *db-env, *redis-env, *minio-env, *aws-s3-env, *proxy-env]
|
||||
<<: [*app-env, *db-env, *redis-env, *minio-env, *aws-s3-env, *proxy-env, *otel-env]
|
||||
depends_on:
|
||||
- plane-db
|
||||
- plane-redis
|
||||
|
||||
@@ -17,24 +17,24 @@ Plane's API server can emit OpenTelemetry traces, HTTP metrics, and trace-correl
|
||||
- A server span per HTTP request, with `http.route`, `http.method`, `http.status_code`, `http.target`, and duration.
|
||||
- A span per Celery task with `celery.action` / `celery.task_name` / `celery.state`. Traceparent is propagated through the queue, so a request that enqueues a task is linked to that task's execution span in the same trace.
|
||||
- Child spans for every Postgres query, Redis op, and outbound `requests` / `httpx` call inside that request or task.
|
||||
- HTTP metrics: `http.server.duration` histogram, `http.server.active_requests`, `http.server.request.size`, `http.server.response.size`.
|
||||
- JSON logs on stdout with `trace_id` / `span_id` / `service_name` fields **added only when `OTEL_ENABLED=1`**. Off path leaves the existing log schema untouched. Point the collector's `filelog` receiver at your container log directory to link logs ↔ traces.
|
||||
- HTTP metrics: `http.server.duration` histogram and `http.server.active_requests`. (These are the only two `DjangoInstrumentor` emits — it does not produce request/response size histograms.)
|
||||
- JSON logs on stdout with `trace_id` / `span_id` / `service_name` fields **added only when OTel is active** — that is, `OTEL_ENABLED` is truthy _and_ an OTLP endpoint is set. Any other combination leaves the existing log schema untouched, so you never get the extra fields without the traces to match them. Point the collector's `filelog` receiver at your container log directory to link logs ↔ traces.
|
||||
|
||||
## Environment variables
|
||||
|
||||
| Var | Default | Purpose |
|
||||
| ----------------------------- | -------------------------- | ----------------------------------------------------------------------- |
|
||||
| `OTEL_ENABLED` | `0` | Plane gate. Must be `1` (or `true`/`yes`/`on`). |
|
||||
| `OTEL_SERVICE_NAME` | `plane-api` | Service identifier in your APM backend |
|
||||
| `OTEL_EXPORTER_OTLP_ENDPOINT` | _(required)_ | Your collector's OTLP receiver |
|
||||
| `OTEL_EXPORTER_OTLP_PROTOCOL` | `grpc` | `grpc` or `http/protobuf` |
|
||||
| `OTEL_EXPORTER_OTLP_HEADERS` | _(unset)_ | For SaaS backends needing auth headers |
|
||||
| `OTEL_ENVIRONMENT` | _(unset)_ | Sets `deployment.environment.name` (falls back to `SENTRY_ENVIRONMENT`) |
|
||||
| `OTEL_TRACES_SAMPLER` | `parentbased_traceidratio` | Standard OTEL sampler |
|
||||
| `OTEL_TRACES_SAMPLER_ARG` | `0.1` | 10 % head sampling. Set to `1.0` to capture every request. |
|
||||
| `OTEL_RESOURCE_ATTRIBUTES` | _(unset)_ | Extra resource attrs: `service.version=...` |
|
||||
| Var | Default | Purpose |
|
||||
| ----------------------------- | -------------------------- | ------------------------------------------------------------------------------------------------------------------------------------------------- |
|
||||
| `OTEL_ENABLED` | `0` | Plane gate. Must be `1` (or `true`/`yes`/`on`). |
|
||||
| `OTEL_SERVICE_NAME` | `plane-api` | Service identifier in your APM backend |
|
||||
| `OTEL_EXPORTER_OTLP_ENDPOINT` | _(required)_ | Your collector's OTLP receiver. The signal-specific `OTEL_EXPORTER_OTLP_TRACES_ENDPOINT` / `OTEL_EXPORTER_OTLP_METRICS_ENDPOINT` are honored too. |
|
||||
| `OTEL_EXPORTER_OTLP_PROTOCOL` | `grpc` | `grpc` or `http/protobuf` |
|
||||
| `OTEL_EXPORTER_OTLP_HEADERS` | _(unset)_ | For SaaS backends needing auth headers |
|
||||
| `OTEL_ENVIRONMENT` | _(unset)_ | Sets `deployment.environment.name` (falls back to `SENTRY_ENVIRONMENT`) |
|
||||
| `OTEL_TRACES_SAMPLER` | `parentbased_traceidratio` | Standard OTEL sampler |
|
||||
| `OTEL_TRACES_SAMPLER_ARG` | `0.1` | 10 % head sampling. Set to `1.0` to capture every request. |
|
||||
| `OTEL_RESOURCE_ATTRIBUTES` | _(unset)_ | Extra resource attrs: `service.version=...` |
|
||||
|
||||
If `OTEL_ENABLED=1` but `OTEL_EXPORTER_OTLP_ENDPOINT` is unset, the API logs a single WARNING at boot and continues without instrumentation — no silent local-host default.
|
||||
If `OTEL_ENABLED=1` but no endpoint var is set, the API logs a single WARNING at boot and continues without instrumentation — no silent local-host default, and the log schema is left unchanged.
|
||||
|
||||
## What's not instrumented yet
|
||||
|
||||
@@ -43,5 +43,5 @@ If `OTEL_ENABLED=1` but `OTEL_EXPORTER_OTLP_ENDPOINT` is unset, the API logs a s
|
||||
## Troubleshooting
|
||||
|
||||
- **No spans showing up.** Confirm `OTEL_ENABLED=1` is in the API container's env, not just the host shell. Check API logs for the `OpenTelemetry configured` INFO line at boot.
|
||||
- **`connection refused` floods stop appearing.** Good — they were dropped batches. The `opentelemetry.*` loggers are pinned to WARNING but the underlying gRPC retries still happen. Fix the collector reachability or unset `OTEL_ENABLED`.
|
||||
- **`connection refused` floods.** These are dropped batches. The `opentelemetry.*` loggers are pinned to WARNING (and kept alive across Django's `dictConfig`, which would otherwise disable them) so the errors still reach your logs, but the underlying gRPC retries keep happening. Fix the collector reachability or unset `OTEL_ENABLED`.
|
||||
- **`trace_id` is empty in logs.** Either you're outside a request/task or the sampler dropped the trace. Drop `OTEL_TRACES_SAMPLER_ARG` to `1.0` while debugging.
|
||||
|
||||
@@ -14,13 +14,22 @@ receivers:
|
||||
include:
|
||||
- /var/lib/docker/containers/*/*-json.log
|
||||
operators:
|
||||
# 1. Unwrap Docker's json-file envelope:
|
||||
# {"log": "<the app's JSON line>\n", "stream": "stdout", "time": "..."}
|
||||
# This leaves the app's own JSON as a string in attributes.log.
|
||||
- type: json_parser
|
||||
- type: move
|
||||
from: attributes.trace_id
|
||||
to: trace_id
|
||||
- type: move
|
||||
from: attributes.span_id
|
||||
to: span_id
|
||||
# 2. Parse the app's JSON out of that string, so trace_id / span_id /
|
||||
# service_name become real attributes.
|
||||
- type: json_parser
|
||||
parse_from: attributes.log
|
||||
# 3. Promote them to the log record's top-level trace context. A `move`
|
||||
# to a non-prefixed field would target body.trace_id and never set the
|
||||
# real trace context — trace_parser is the operator that does.
|
||||
- type: trace_parser
|
||||
trace_id:
|
||||
parse_from: attributes.trace_id
|
||||
span_id:
|
||||
parse_from: attributes.span_id
|
||||
|
||||
processors:
|
||||
batch:
|
||||
|
||||
Reference in New Issue
Block a user