diff --git a/common/logging.py b/common/logging.py index 21c7834de..cb57d2e92 100644 --- a/common/logging.py +++ b/common/logging.py @@ -9,6 +9,7 @@ import watchtower from boto3 import client from botocore.exceptions import ClientError +from opentelemetry import trace as _otel_trace from .config import Config @@ -80,6 +81,7 @@ def setup_cw_logging(main_logger): stream_name=CFG.hostname, ) handler.setFormatter(CloudWatchLogFormatterCustom()) + handler.addFilter(ContextualFilter()) except ClientError: logger.exception("Unable to enable CloudWatch logging: ") else: # pragma: no cover @@ -87,6 +89,24 @@ def setup_cw_logging(main_logger): logger.info("CloudWatch logging ENABLED!") +class ContextualFilter(logging.Filter): + """Logging filter that injects OpenTelemetry trace context into log records. + + Adds hex-encoded trace_id (32 chars) and span_id (16 chars) to every log + record. When no active span exists, emits None without crashing. + """ + + def filter(self, log_record): + try: + span_context = _otel_trace.get_current_span().get_span_context() + log_record.trace_id = format(span_context.trace_id, "032x") if span_context.is_valid else None + log_record.span_id = format(span_context.span_id, "016x") if span_context.is_valid else None + except Exception: + log_record.trace_id = None + log_record.span_id = None + return True + + def init_logging(num_servers=1): """Setup root logger handler.""" logger = logging.getLogger() @@ -103,6 +123,7 @@ def init_logging(num_servers=1): handler = logging.StreamHandler() formatter = OneLineExceptionFormatter(log_fmt) handler.setFormatter(formatter) + handler.addFilter(ContextualFilter()) logger.addHandler(handler) setup_cw_logging(logger) diff --git a/common/mqueue.py b/common/mqueue.py index c41f6784a..50b59ec28 100644 --- a/common/mqueue.py +++ b/common/mqueue.py @@ -8,13 +8,17 @@ from aiokafka import AIOKafkaProducer from aiokafka.errors import KafkaError from aiokafka.partitioner import DefaultPartitioner +from opentelemetry.trace import SpanKind from common.config import Config from common.logging import get_logger +from common.telemetry import get_tracer +from common.telemetry import inject_context_to_msg_headers LOGGER = get_logger(__name__) CFG = Config() +TRACER = get_tracer(__name__) class Partitioners: @@ -142,8 +146,18 @@ async def send_one(self, msg, key=None, headers=None): await self.start() try: data = bytes(json.dumps(msg).encode("utf-8")) - res = await self.client.send_and_wait(self.topic, value=data, key=self._serialize_key(key), headers=headers) - LOGGER.debug(res) + with TRACER.start_as_current_span( + f"{self.topic} send", + kind=SpanKind.PRODUCER, + attributes={ + "messaging.system": "kafka", + "messaging.operation.name": "send", + "messaging.destination.name": self.topic, + }, + ): + headers = inject_context_to_msg_headers(headers) + res = await self.client.send_and_wait(self.topic, value=data, key=self._serialize_key(key), headers=headers) + LOGGER.debug(res) except KafkaError: self.connected = False @@ -153,7 +167,17 @@ async def send_many(self, msg_list, key=None, headers=None): try: for msg in msg_list: data = bytes(json.dumps(msg).encode("utf-8")) - res = await self.client.send_and_wait(self.topic, value=data, key=self._serialize_key(key), headers=headers) + with TRACER.start_as_current_span( + f"{self.topic} send", + kind=SpanKind.PRODUCER, + attributes={ + "messaging.system": "kafka", + "messaging.operation.name": "send", + "messaging.destination.name": self.topic, + }, + ): + msg_headers = inject_context_to_msg_headers(headers) + res = await self.client.send_and_wait(self.topic, value=data, key=self._serialize_key(key), headers=msg_headers) LOGGER.debug(res) except KafkaError: self.connected = False @@ -162,8 +186,18 @@ async def send_raw(self, msg: bytes, key=None, headers=None): """Logic around sending raw message""" await self.start() try: - res = await self.client.send_and_wait(self.topic, value=msg, key=self._serialize_key(key), headers=headers) - LOGGER.debug(res) + with TRACER.start_as_current_span( + f"{self.topic} send", + kind=SpanKind.PRODUCER, + attributes={ + "messaging.system": "kafka", + "messaging.operation.name": "send", + "messaging.destination.name": self.topic, + }, + ): + headers = inject_context_to_msg_headers(headers) + res = await self.client.send_and_wait(self.topic, value=msg, key=self._serialize_key(key), headers=headers) + LOGGER.debug(res) except KafkaError: self.connected = False diff --git a/common/telemetry.py b/common/telemetry.py new file mode 100644 index 000000000..eb967eb37 --- /dev/null +++ b/common/telemetry.py @@ -0,0 +1,474 @@ +"""OpenTelemetry initialization for Vulnerability Engine services. + +Provides centralized tracing setup for all Vulnerability Engine entry points +(Manager API, Listener, Evaluator, Grouper, VMaaS Sync, etc.). Every knob is +configurable via environment variables so stage and prod can run independent +configurations without code changes. + +Usage: + from common.telemetry import init_otel, instrument_flask_app, instrument_sqlalchemy, instrument_outbound_http + + # In service main(): + init_otel(service_name="vulnerability-engine") + + # After Flask app creation: + instrument_flask_app(flask_app) + + # After db engine creation: + instrument_sqlalchemy(engine) + + # For outbound HTTP (e.g., VMaaS calls): + instrument_outbound_http() +""" + +import os +from contextlib import contextmanager +from contextvars import ContextVar +from urllib.parse import urlparse + +from opentelemetry import context as otel_context_api + +from common.logging import get_logger + +logger = get_logger(__name__) + +# Coroutine-safe context storage for propagating org_id / request_id in async MQ handlers. +# ContextVar is safe under asyncio, unlike threading.local(). +_ctx_org_id: ContextVar[str | None] = ContextVar("ctx_org_id", default=None) +_ctx_request_id: ContextVar[str | None] = ContextVar("ctx_request_id", default=None) + + +class _AsyncSafeContext: + """Attribute-style wrapper over ContextVars for backward compatibility.""" + + _vars = { + "org_id": _ctx_org_id, + "request_id": _ctx_request_id, + } + + def __getattr__(self, name: str): + try: + return self._vars[name].get() + except KeyError: + raise AttributeError(name) from None + + def __setattr__(self, name: str, value): + if name in self._vars: + self._vars[name].set(value) + else: + super().__setattr__(name, value) + + +threadctx = _AsyncSafeContext() + +# --------------------------------------------------------------------------- +# Configuration — all tunables are environment-driven +# --------------------------------------------------------------------------- +OTEL_ENABLED = os.getenv("OTEL_ENABLED", "false").lower() == "true" +OTEL_SQL_ENABLED = os.getenv("OTEL_SQL_ENABLED", "true").lower() == "true" +OTEL_SQL_COMMENTER_ENABLED = os.getenv("OTEL_SQL_COMMENTER_ENABLED", "false").lower() == "true" +OTEL_HTTP_INBOUND_ENABLED = os.getenv("OTEL_HTTP_INBOUND_ENABLED", "true").lower() == "true" +OTEL_HTTP_OUTBOUND_ENABLED = os.getenv("OTEL_HTTP_OUTBOUND_ENABLED", "true").lower() == "true" +OTEL_MQ_ENABLED = os.getenv("OTEL_MQ_ENABLED", "true").lower() == "true" + + +def _parse_sampling_rate(raw: str, env_var: str = "OTEL_SAMPLING_RATE", default: float = 1.0) -> float: + """Parse a sampling-rate value, clamping to [0.0, 1.0] and falling back on bad values.""" + try: + rate = float(raw) + except (ValueError, TypeError): + logger.warning("Invalid %s=%r; falling back to %s", env_var, raw, default) + rate = default + return min(max(rate, 0.0), 1.0) + + +OTEL_SAMPLING_RATE = _parse_sampling_rate(os.getenv("OTEL_SAMPLING_RATE", "1.0")) + +OTEL_BSP_MAX_QUEUE_SIZE = int(os.getenv("OTEL_BSP_MAX_QUEUE_SIZE", "8192")) +OTEL_BSP_MAX_EXPORT_BATCH_SIZE = int(os.getenv("OTEL_BSP_MAX_EXPORT_BATCH_SIZE", "256")) +OTEL_BSP_SCHEDULE_DELAY = int(os.getenv("OTEL_BSP_SCHEDULE_DELAY", "2000")) +OTEL_BSP_EXPORT_TIMEOUT = int(os.getenv("OTEL_BSP_EXPORT_TIMEOUT", "10000")) + +OTEL_EXPORTER_OTLP_COMPRESSION = os.getenv("OTEL_EXPORTER_OTLP_COMPRESSION", "gzip").lower() +OTEL_SPAN_ATTRIBUTE_COUNT_LIMIT = int(os.getenv("OTEL_SPAN_ATTRIBUTE_COUNT_LIMIT", "64")) +OTEL_SPAN_ATTRIBUTE_VALUE_LENGTH_LIMIT = int(os.getenv("OTEL_SPAN_ATTRIBUTE_VALUE_LENGTH_LIMIT", "1024")) + +_RH_PROPAGATED_ATTRS = ("rh.org_id", "rh.request_id") + +_otel_initialized_pid = None + + +def _build_rh_attribute_span_processor(rh_service: str = "vulnerability-engine"): + """Build a SpanProcessor that sets platform rh.* attributes on every span. + + Defined as a factory so the SDK SpanProcessor base is imported only when OTel + is actually initialized. + """ + from opentelemetry import trace + from opentelemetry.sdk.trace import ReadableSpan + from opentelemetry.sdk.trace import SpanProcessor + + class RHAttributeSpanProcessor(SpanProcessor): + """Set rh.service on every span; copy rh.org_id / rh.request_id to children. + + Prefers org/request attrs already set on the parent span (HTTP request hook). + Falls back to threadctx for MQ, where those values are populated mid-message + processing. + """ + + def on_start(self, span, parent_context=None): + if not span or not span.is_recording(): + return + + span.set_attribute("rh.service", rh_service) + + parent = trace.get_current_span(parent_context) if parent_context is not None else trace.get_current_span() + parent_attrs = {} + if isinstance(parent, ReadableSpan) and parent.attributes: + parent_attrs = parent.attributes + + for attr in _RH_PROPAGATED_ATTRS: + if attr in (span.attributes or {}): + continue + value = parent_attrs.get(attr) + if value is None: + thread_key = "org_id" if attr == "rh.org_id" else "request_id" + value = getattr(threadctx, thread_key, None) + if value is not None: + span.set_attribute(attr, value) + + return RHAttributeSpanProcessor() + + +def get_tracer(name: str): + """Get an OpenTelemetry tracer for creating custom spans. + + When init_otel() has not been called (i.e. OTEL is disabled), the SDK's + default TracerProvider is a no-op that creates zero-overhead no-op spans. + """ + from opentelemetry import trace + + return trace.get_tracer(name) + + +@contextmanager +def use_otel_context(ctx): + """Temporarily activate an OTel context (or no-op if None).""" + token = otel_context_api.attach(ctx) if ctx is not None else None + try: + yield + finally: + if token is not None: + otel_context_api.detach(token) + + +def extract_context_from_msg_headers(headers): + """Extract OTel trace context from Kafka message headers. + + Args: + headers: List of (key, value) tuples from a Kafka message. + + Returns: + An OTel Context object with the extracted trace context, + or None if no traceparent header is found. + """ + if not OTEL_ENABLED or not headers: + return None + + from opentelemetry.propagate import extract + + carrier = {} + for key, value in headers: + if isinstance(value, bytes): + carrier[key] = value.decode("utf-8") + else: + carrier[key] = value + + if "traceparent" not in carrier: + return None + + return extract(carrier) + + +def inject_context_to_msg_headers(headers=None): + """Inject OTel trace context into Kafka message headers. + + Args: + headers: Optional existing list of (key, value) tuples. + + Returns: + List of (key, value) tuples with trace context injected. + """ + if not OTEL_ENABLED: + return headers or [] + + from opentelemetry.propagate import inject + + carrier = {} + inject(carrier) + + result = list(headers) if headers else [] + for key, value in carrier.items(): + result.append((key, value.encode("utf-8") if isinstance(value, str) else value)) + return result + + +def init_otel( + service_name: str = "vulnerability-engine", + service_version: str | None = None, + *, + rh_service: str = "vulnerability-engine", + sampling_rate: float | None = None, +): + """Initialize OpenTelemetry tracing. Safe to call multiple times. + + Uses the current PID to detect fork boundaries: if gunicorn master process + initializes and then post_fork calls again in a worker, the PID will differ + and the worker will re-initialize with a fresh TracerProvider and BatchSpanProcessor. + + Args: + service_name: OTel service.name resource attribute. + service_version: OTel service.version resource attribute. Defaults to IMAGE_TAG env var. + rh_service: Platform rh.service span attribute (logical service name). + sampling_rate: Optional override for OTEL_SAMPLING_RATE. + """ + global _otel_initialized_pid + + if _otel_initialized_pid == os.getpid(): + return + + if not OTEL_ENABLED: + logger.info("OpenTelemetry is disabled (OTEL_ENABLED != 'true')") + _otel_initialized_pid = os.getpid() + return + + from opentelemetry import trace + from opentelemetry.exporter.otlp.proto.http import Compression + from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter + from opentelemetry.sdk.resources import SERVICE_NAME + from opentelemetry.sdk.resources import SERVICE_VERSION + from opentelemetry.sdk.resources import Resource + from opentelemetry.sdk.trace import SpanLimits + from opentelemetry.sdk.trace import TracerProvider + from opentelemetry.sdk.trace.export import BatchSpanProcessor + from opentelemetry.sdk.trace.sampling import ParentBased + from opentelemetry.sdk.trace.sampling import TraceIdRatioBased + + if service_version is None: + service_version = os.getenv("IMAGE_TAG", "unknown") + + resource = Resource.create( + attributes={ + SERVICE_NAME: service_name, + SERVICE_VERSION: service_version, + "deployment.environment": os.getenv("NAMESPACE", "development"), + } + ) + + effective_rate = sampling_rate or OTEL_SAMPLING_RATE + sampler = ParentBased(TraceIdRatioBased(effective_rate)) + span_limits = SpanLimits( + max_attributes=OTEL_SPAN_ATTRIBUTE_COUNT_LIMIT, + max_attribute_length=OTEL_SPAN_ATTRIBUTE_VALUE_LENGTH_LIMIT, + ) + provider = TracerProvider(resource=resource, sampler=sampler, span_limits=span_limits) + + provider.add_span_processor(_build_rh_attribute_span_processor(rh_service=rh_service)) + + _compression_map = {"gzip": Compression.Gzip, "deflate": Compression.Deflate} + compression = _compression_map.get(OTEL_EXPORTER_OTLP_COMPRESSION, Compression.NoCompression) + exporter = OTLPSpanExporter(compression=compression) + provider.add_span_processor( + BatchSpanProcessor( + exporter, + max_queue_size=OTEL_BSP_MAX_QUEUE_SIZE, + max_export_batch_size=OTEL_BSP_MAX_EXPORT_BATCH_SIZE, + schedule_delay_millis=OTEL_BSP_SCHEDULE_DELAY, + export_timeout_millis=OTEL_BSP_EXPORT_TIMEOUT, + ) + ) + + trace.set_tracer_provider(provider) + _otel_initialized_pid = os.getpid() + logger.info("OpenTelemetry initialized for service=%s version=%s", service_name, service_version) + logger.info( + "OpenTelemetry config: sampling=%.2f sql=%s commenter=%s http_inbound=%s http_outbound=%s " + "mq=%s bsp_queue=%d bsp_batch=%d bsp_delay=%dms bsp_timeout=%dms " + "compression=%s attr_limit=%d attr_len_limit=%d", + effective_rate, + OTEL_SQL_ENABLED, + OTEL_SQL_COMMENTER_ENABLED, + OTEL_HTTP_INBOUND_ENABLED, + OTEL_HTTP_OUTBOUND_ENABLED, + OTEL_MQ_ENABLED, + OTEL_BSP_MAX_QUEUE_SIZE, + OTEL_BSP_MAX_EXPORT_BATCH_SIZE, + OTEL_BSP_SCHEDULE_DELAY, + OTEL_BSP_EXPORT_TIMEOUT, + OTEL_EXPORTER_OTLP_COMPRESSION, + OTEL_SPAN_ATTRIBUTE_COUNT_LIMIT, + OTEL_SPAN_ATTRIBUTE_VALUE_LENGTH_LIMIT, + ) + + +def instrument_flask_app(flask_app): + """Instrument a Flask app with OpenTelemetry request tracing. + + Each HTTP request becomes a span with method, path, status code, + plus RH-specific attributes (org_id, request_id). + """ + if not OTEL_ENABLED or not OTEL_HTTP_INBOUND_ENABLED: + return + + from opentelemetry.instrumentation.flask import FlaskInstrumentor + + FlaskInstrumentor().instrument_app( + flask_app, + excluded_urls=r"/healthz(?:\?.*)?$,/metrics(?:\?.*)?$", + request_hook=_request_hook, + response_hook=_response_hook, + ) + logger.info("Flask instrumented with OpenTelemetry") + + +def instrument_sqlalchemy(engine): + """Instrument a SQLAlchemy engine with OpenTelemetry query tracing. + + Controlled by OTEL_SQL_ENABLED (master toggle) and OTEL_SQL_COMMENTER_ENABLED + (adds traceparent to SQL comments for pg_stat_activity visibility). + """ + if not OTEL_ENABLED or not OTEL_SQL_ENABLED: + return + + from opentelemetry.instrumentation.sqlalchemy import SQLAlchemyInstrumentor + + SQLAlchemyInstrumentor().instrument( + engine=engine, + enable_commenter=OTEL_SQL_COMMENTER_ENABLED, + commenter_options={ + "db_framework": True, + "db_driver": True, + }, + ) + logger.info("SQLAlchemy engine instrumented with OpenTelemetry (commenter=%s)", OTEL_SQL_COMMENTER_ENABLED) + + +def instrument_psycopg2(): + """Instrument psycopg2 with OpenTelemetry query tracing. + + Used by components that use psycopg2 directly (VMaaS sync, taskomatic). + """ + if not OTEL_ENABLED or not OTEL_SQL_ENABLED: + return + + try: + from opentelemetry.instrumentation.psycopg2 import Psycopg2Instrumentor + + Psycopg2Instrumentor().instrument( + enable_commenter=OTEL_SQL_COMMENTER_ENABLED, + ) + logger.info("psycopg2 instrumented with OpenTelemetry (commenter=%s)", OTEL_SQL_COMMENTER_ENABLED) + except ImportError: + logger.warning("psycopg2 instrumentation unavailable; continuing without DB spans") + + +def instrument_psycopg(): + """Instrument psycopg (v3) with OpenTelemetry query tracing. + + Used by async components (listener, evaluator, grouper) that use psycopg3 + with AsyncConnectionPool. + """ + if not OTEL_ENABLED or not OTEL_SQL_ENABLED: + return + + try: + from opentelemetry.instrumentation.psycopg import PsycopgInstrumentor + + PsycopgInstrumentor().instrument( + enable_commenter=OTEL_SQL_COMMENTER_ENABLED, + ) + logger.info("psycopg (v3) instrumented with OpenTelemetry (commenter=%s)", OTEL_SQL_COMMENTER_ENABLED) + except ImportError: + logger.warning("psycopg (v3) instrumentation unavailable; continuing without async DB spans") + + +def instrument_outbound_http(): + """Instrument outbound HTTP calls (e.g., VMaaS) with OpenTelemetry. + + Controlled by OTEL_HTTP_OUTBOUND_ENABLED. Automatically creates spans for all + requests made via the `requests` library, including trace context + propagation to downstream services. + """ + if not OTEL_ENABLED or not OTEL_HTTP_OUTBOUND_ENABLED: + return + + from opentelemetry.instrumentation.requests import RequestsInstrumentor + + RequestsInstrumentor().instrument(request_hook=_outbound_request_hook) + logger.info("Outbound HTTP (requests library) instrumented with OpenTelemetry") + + +def instrument_aiohttp(): + """Instrument aiohttp client sessions with OpenTelemetry. + + Used by async components (evaluator, listener) that make HTTP calls via aiohttp. + """ + if not OTEL_ENABLED or not OTEL_HTTP_OUTBOUND_ENABLED: + return + + try: + from opentelemetry.instrumentation.aiohttp_client import ( + AioHttpClientInstrumentor, + ) + + AioHttpClientInstrumentor().instrument() + logger.info("aiohttp client instrumented with OpenTelemetry") + except ImportError: + logger.warning("aiohttp client instrumentation unavailable; continuing without outbound async HTTP spans") + + +def _outbound_request_hook(span, request, *_args, **_kwargs): + """Use METHOD + path as the client span name (default is METHOD only).""" + if not span or not span.is_recording(): + return + parsed = urlparse(request.path_url) + path = parsed.path or "/" + span.update_name(f"{request.method} {path}") + + +def _request_hook(span, environ): # noqa: ARG001 + """Add Red Hat platform attributes to every request span. + + Adds org_id and request_id so traces can be filtered in Grafana/Tempo. + """ + if not span or not span.is_recording(): + return + + from flask import request + + request_id = request.headers.get("x-rh-insights-request-id", "") + if request_id: + span.set_attribute("rh.request_id", request_id) + + try: + from common.identity import get_identity + + encoded_id = request.headers.get("x-rh-identity", "") + if encoded_id: + identity = get_identity(encoded_id) + if identity and "org_id" in identity.get("identity", {}): + span.set_attribute("rh.org_id", identity["identity"]["org_id"]) + except Exception: + pass + + +def _response_hook(span, status, response_headers): # noqa: ARG001 + """Add response-level attributes to request spans.""" + import contextlib + + if span and span.is_recording(): + if isinstance(status, str): + with contextlib.suppress(ValueError, IndexError): + span.set_attribute("http.status_code", int(status.split()[0])) + elif isinstance(status, int): + span.set_attribute("http.status_code", status) diff --git a/conf/common.env b/conf/common.env index 3438149d7..6bfd5c6b8 100644 --- a/conf/common.env +++ b/conf/common.env @@ -44,3 +44,19 @@ UNLEASH_BOOTSTRAP_FILE=develfeatureflags.json DB_POOL_TIMEOUT=300 DB_STATEMENT_TIMEOUT=1800000 DB_WORK_MEM=256000 +IMAGE_TAG=latest +NAMESPACE=development +OTEL_ENABLED=FALSE +OTEL_SAMPLING_RATE=1.0 +OTEL_SQL_ENABLED=TRUE +OTEL_SQL_COMMENTER_ENABLED=FALSE +OTEL_HTTP_INBOUND_ENABLED=TRUE +OTEL_HTTP_OUTBOUND_ENABLED=TRUE +OTEL_MQ_ENABLED=TRUE +OTEL_BSP_MAX_QUEUE_SIZE=8192 +OTEL_BSP_MAX_EXPORT_BATCH_SIZE=256 +OTEL_BSP_SCHEDULE_DELAY=2000 +OTEL_BSP_EXPORT_TIMEOUT=10000 +OTEL_EXPORTER_OTLP_COMPRESSION=gzip +OTEL_SPAN_ATTRIBUTE_COUNT_LIMIT=64 +OTEL_SPAN_ATTRIBUTE_VALUE_LENGTH_LIMIT=1024 diff --git a/deploy/clowdapp.yaml b/deploy/clowdapp.yaml index 28f18d638..4b31f390b 100644 --- a/deploy/clowdapp.yaml +++ b/deploy/clowdapp.yaml @@ -117,6 +117,38 @@ objects: value: ${KESSEL_AUTH_OIDC_ISSUER} - name: KESSEL_INSECURE value: ${KESSEL_INSECURE} + - name: OTEL_ENABLED + value: ${OTEL_ENABLED} + - name: OTEL_EXPORTER_OTLP_ENDPOINT + value: ${OTEL_EXPORTER_OTLP_ENDPOINT} + - name: OTEL_SAMPLING_RATE + value: ${OTEL_SAMPLING_RATE} + - name: OTEL_SQL_ENABLED + value: ${OTEL_SQL_ENABLED} + - name: OTEL_HTTP_INBOUND_ENABLED + value: ${OTEL_HTTP_INBOUND_ENABLED} + - name: OTEL_HTTP_OUTBOUND_ENABLED + value: ${OTEL_HTTP_OUTBOUND_ENABLED} + - name: OTEL_MQ_ENABLED + value: ${OTEL_MQ_ENABLED} + - name: OTEL_SQL_COMMENTER_ENABLED + value: ${OTEL_SQL_COMMENTER_ENABLED} + - name: OTEL_BSP_MAX_QUEUE_SIZE + value: ${OTEL_BSP_MAX_QUEUE_SIZE} + - name: OTEL_BSP_MAX_EXPORT_BATCH_SIZE + value: ${OTEL_BSP_MAX_EXPORT_BATCH_SIZE} + - name: OTEL_BSP_SCHEDULE_DELAY + value: ${OTEL_BSP_SCHEDULE_DELAY} + - name: OTEL_BSP_EXPORT_TIMEOUT + value: ${OTEL_BSP_EXPORT_TIMEOUT} + - name: OTEL_EXPORTER_OTLP_COMPRESSION + value: ${OTEL_EXPORTER_OTLP_COMPRESSION} + - name: OTEL_SPAN_ATTRIBUTE_COUNT_LIMIT + value: ${OTEL_SPAN_ATTRIBUTE_COUNT_LIMIT} + - name: OTEL_SPAN_ATTRIBUTE_VALUE_LENGTH_LIMIT + value: ${OTEL_SPAN_ATTRIBUTE_VALUE_LENGTH_LIMIT} + - name: IMAGE_TAG + value: ${IMAGE_TAG} resources: limits: cpu: ${CPU_LIMIT_MANAGER} @@ -237,6 +269,36 @@ objects: value: ${DB_STATEMENT_TIMEOUT} - name: DB_WORK_MEM value: ${DB_WORK_MEM} + - name: OTEL_ENABLED + value: ${OTEL_ENABLED} + - name: OTEL_EXPORTER_OTLP_ENDPOINT + value: ${OTEL_EXPORTER_OTLP_ENDPOINT} + - name: OTEL_SAMPLING_RATE + value: ${OTEL_SAMPLING_RATE} + - name: OTEL_SQL_ENABLED + value: ${OTEL_SQL_ENABLED} + - name: OTEL_HTTP_OUTBOUND_ENABLED + value: ${OTEL_HTTP_OUTBOUND_ENABLED} + - name: OTEL_MQ_ENABLED + value: ${OTEL_MQ_ENABLED} + - name: OTEL_SQL_COMMENTER_ENABLED + value: ${OTEL_SQL_COMMENTER_ENABLED} + - name: OTEL_BSP_MAX_QUEUE_SIZE + value: ${OTEL_BSP_MAX_QUEUE_SIZE} + - name: OTEL_BSP_MAX_EXPORT_BATCH_SIZE + value: ${OTEL_BSP_MAX_EXPORT_BATCH_SIZE} + - name: OTEL_BSP_SCHEDULE_DELAY + value: ${OTEL_BSP_SCHEDULE_DELAY} + - name: OTEL_BSP_EXPORT_TIMEOUT + value: ${OTEL_BSP_EXPORT_TIMEOUT} + - name: OTEL_EXPORTER_OTLP_COMPRESSION + value: ${OTEL_EXPORTER_OTLP_COMPRESSION} + - name: OTEL_SPAN_ATTRIBUTE_COUNT_LIMIT + value: ${OTEL_SPAN_ATTRIBUTE_COUNT_LIMIT} + - name: OTEL_SPAN_ATTRIBUTE_VALUE_LENGTH_LIMIT + value: ${OTEL_SPAN_ATTRIBUTE_VALUE_LENGTH_LIMIT} + - name: IMAGE_TAG + value: ${IMAGE_TAG} resources: limits: cpu: ${CPU_LIMIT_TASKOMATIC} @@ -277,6 +339,36 @@ objects: value: ${UNLEASH_BOOTSTRAP_FILE} - name: DB_STATEMENT_TIMEOUT value: ${DB_STATEMENT_TIMEOUT} + - name: OTEL_ENABLED + value: ${OTEL_ENABLED} + - name: OTEL_EXPORTER_OTLP_ENDPOINT + value: ${OTEL_EXPORTER_OTLP_ENDPOINT} + - name: OTEL_SAMPLING_RATE + value: ${OTEL_SAMPLING_RATE} + - name: OTEL_SQL_ENABLED + value: ${OTEL_SQL_ENABLED} + - name: OTEL_HTTP_OUTBOUND_ENABLED + value: ${OTEL_HTTP_OUTBOUND_ENABLED} + - name: OTEL_MQ_ENABLED + value: ${OTEL_MQ_ENABLED} + - name: OTEL_SQL_COMMENTER_ENABLED + value: ${OTEL_SQL_COMMENTER_ENABLED} + - name: OTEL_BSP_MAX_QUEUE_SIZE + value: ${OTEL_BSP_MAX_QUEUE_SIZE} + - name: OTEL_BSP_MAX_EXPORT_BATCH_SIZE + value: ${OTEL_BSP_MAX_EXPORT_BATCH_SIZE} + - name: OTEL_BSP_SCHEDULE_DELAY + value: ${OTEL_BSP_SCHEDULE_DELAY} + - name: OTEL_BSP_EXPORT_TIMEOUT + value: ${OTEL_BSP_EXPORT_TIMEOUT} + - name: OTEL_EXPORTER_OTLP_COMPRESSION + value: ${OTEL_EXPORTER_OTLP_COMPRESSION} + - name: OTEL_SPAN_ATTRIBUTE_COUNT_LIMIT + value: ${OTEL_SPAN_ATTRIBUTE_COUNT_LIMIT} + - name: OTEL_SPAN_ATTRIBUTE_VALUE_LENGTH_LIMIT + value: ${OTEL_SPAN_ATTRIBUTE_VALUE_LENGTH_LIMIT} + - name: IMAGE_TAG + value: ${IMAGE_TAG} resources: limits: cpu: ${CPU_LIMIT_NOTIFICATOR} @@ -321,6 +413,36 @@ objects: value: ${DISABLE_OPTIMISATION} - name: UNLEASH_BOOTSTRAP_FILE value: ${UNLEASH_BOOTSTRAP_FILE} + - name: OTEL_ENABLED + value: ${OTEL_ENABLED} + - name: OTEL_EXPORTER_OTLP_ENDPOINT + value: ${OTEL_EXPORTER_OTLP_ENDPOINT} + - name: OTEL_SAMPLING_RATE + value: ${OTEL_SAMPLING_RATE} + - name: OTEL_SQL_ENABLED + value: ${OTEL_SQL_ENABLED} + - name: OTEL_HTTP_OUTBOUND_ENABLED + value: ${OTEL_HTTP_OUTBOUND_ENABLED} + - name: OTEL_MQ_ENABLED + value: ${OTEL_MQ_ENABLED} + - name: OTEL_SQL_COMMENTER_ENABLED + value: ${OTEL_SQL_COMMENTER_ENABLED} + - name: OTEL_BSP_MAX_QUEUE_SIZE + value: ${OTEL_BSP_MAX_QUEUE_SIZE} + - name: OTEL_BSP_MAX_EXPORT_BATCH_SIZE + value: ${OTEL_BSP_MAX_EXPORT_BATCH_SIZE} + - name: OTEL_BSP_SCHEDULE_DELAY + value: ${OTEL_BSP_SCHEDULE_DELAY} + - name: OTEL_BSP_EXPORT_TIMEOUT + value: ${OTEL_BSP_EXPORT_TIMEOUT} + - name: OTEL_EXPORTER_OTLP_COMPRESSION + value: ${OTEL_EXPORTER_OTLP_COMPRESSION} + - name: OTEL_SPAN_ATTRIBUTE_COUNT_LIMIT + value: ${OTEL_SPAN_ATTRIBUTE_COUNT_LIMIT} + - name: OTEL_SPAN_ATTRIBUTE_VALUE_LENGTH_LIMIT + value: ${OTEL_SPAN_ATTRIBUTE_VALUE_LENGTH_LIMIT} + - name: IMAGE_TAG + value: ${IMAGE_TAG} resources: limits: cpu: ${CPU_LIMIT_GROUPER} @@ -367,6 +489,36 @@ objects: value: ${UNLEASH_BOOTSTRAP_FILE} - name: DB_STATEMENT_TIMEOUT value: ${DB_STATEMENT_TIMEOUT} + - name: OTEL_ENABLED + value: ${OTEL_ENABLED} + - name: OTEL_EXPORTER_OTLP_ENDPOINT + value: ${OTEL_EXPORTER_OTLP_ENDPOINT} + - name: OTEL_SAMPLING_RATE + value: ${OTEL_SAMPLING_RATE} + - name: OTEL_SQL_ENABLED + value: ${OTEL_SQL_ENABLED} + - name: OTEL_HTTP_OUTBOUND_ENABLED + value: ${OTEL_HTTP_OUTBOUND_ENABLED} + - name: OTEL_MQ_ENABLED + value: ${OTEL_MQ_ENABLED} + - name: OTEL_SQL_COMMENTER_ENABLED + value: ${OTEL_SQL_COMMENTER_ENABLED} + - name: OTEL_BSP_MAX_QUEUE_SIZE + value: ${OTEL_BSP_MAX_QUEUE_SIZE} + - name: OTEL_BSP_MAX_EXPORT_BATCH_SIZE + value: ${OTEL_BSP_MAX_EXPORT_BATCH_SIZE} + - name: OTEL_BSP_SCHEDULE_DELAY + value: ${OTEL_BSP_SCHEDULE_DELAY} + - name: OTEL_BSP_EXPORT_TIMEOUT + value: ${OTEL_BSP_EXPORT_TIMEOUT} + - name: OTEL_EXPORTER_OTLP_COMPRESSION + value: ${OTEL_EXPORTER_OTLP_COMPRESSION} + - name: OTEL_SPAN_ATTRIBUTE_COUNT_LIMIT + value: ${OTEL_SPAN_ATTRIBUTE_COUNT_LIMIT} + - name: OTEL_SPAN_ATTRIBUTE_VALUE_LENGTH_LIMIT + value: ${OTEL_SPAN_ATTRIBUTE_VALUE_LENGTH_LIMIT} + - name: IMAGE_TAG + value: ${IMAGE_TAG} resources: limits: cpu: ${CPU_LIMIT_LISTENER} @@ -417,6 +569,36 @@ objects: value: ${DB_STATEMENT_TIMEOUT} - name: INSIGHTS_LOAD_RULE_CACHE_TTL_SEC value: ${INSIGHTS_LOAD_RULE_CACHE_TTL_SEC} + - name: OTEL_ENABLED + value: ${OTEL_ENABLED} + - name: OTEL_EXPORTER_OTLP_ENDPOINT + value: ${OTEL_EXPORTER_OTLP_ENDPOINT} + - name: OTEL_SAMPLING_RATE + value: ${OTEL_SAMPLING_RATE} + - name: OTEL_SQL_ENABLED + value: ${OTEL_SQL_ENABLED} + - name: OTEL_HTTP_OUTBOUND_ENABLED + value: ${OTEL_HTTP_OUTBOUND_ENABLED} + - name: OTEL_MQ_ENABLED + value: ${OTEL_MQ_ENABLED} + - name: OTEL_SQL_COMMENTER_ENABLED + value: ${OTEL_SQL_COMMENTER_ENABLED} + - name: OTEL_BSP_MAX_QUEUE_SIZE + value: ${OTEL_BSP_MAX_QUEUE_SIZE} + - name: OTEL_BSP_MAX_EXPORT_BATCH_SIZE + value: ${OTEL_BSP_MAX_EXPORT_BATCH_SIZE} + - name: OTEL_BSP_SCHEDULE_DELAY + value: ${OTEL_BSP_SCHEDULE_DELAY} + - name: OTEL_BSP_EXPORT_TIMEOUT + value: ${OTEL_BSP_EXPORT_TIMEOUT} + - name: OTEL_EXPORTER_OTLP_COMPRESSION + value: ${OTEL_EXPORTER_OTLP_COMPRESSION} + - name: OTEL_SPAN_ATTRIBUTE_COUNT_LIMIT + value: ${OTEL_SPAN_ATTRIBUTE_COUNT_LIMIT} + - name: OTEL_SPAN_ATTRIBUTE_VALUE_LENGTH_LIMIT + value: ${OTEL_SPAN_ATTRIBUTE_VALUE_LENGTH_LIMIT} + - name: IMAGE_TAG + value: ${IMAGE_TAG} resources: limits: cpu: ${CPU_LIMIT_EVALUATOR_RECALC} @@ -467,6 +649,36 @@ objects: value: ${DB_STATEMENT_TIMEOUT} - name: INSIGHTS_LOAD_RULE_CACHE_TTL_SEC value: ${INSIGHTS_LOAD_RULE_CACHE_TTL_SEC} + - name: OTEL_ENABLED + value: ${OTEL_ENABLED} + - name: OTEL_EXPORTER_OTLP_ENDPOINT + value: ${OTEL_EXPORTER_OTLP_ENDPOINT} + - name: OTEL_SAMPLING_RATE + value: ${OTEL_SAMPLING_RATE} + - name: OTEL_SQL_ENABLED + value: ${OTEL_SQL_ENABLED} + - name: OTEL_HTTP_OUTBOUND_ENABLED + value: ${OTEL_HTTP_OUTBOUND_ENABLED} + - name: OTEL_MQ_ENABLED + value: ${OTEL_MQ_ENABLED} + - name: OTEL_SQL_COMMENTER_ENABLED + value: ${OTEL_SQL_COMMENTER_ENABLED} + - name: OTEL_BSP_MAX_QUEUE_SIZE + value: ${OTEL_BSP_MAX_QUEUE_SIZE} + - name: OTEL_BSP_MAX_EXPORT_BATCH_SIZE + value: ${OTEL_BSP_MAX_EXPORT_BATCH_SIZE} + - name: OTEL_BSP_SCHEDULE_DELAY + value: ${OTEL_BSP_SCHEDULE_DELAY} + - name: OTEL_BSP_EXPORT_TIMEOUT + value: ${OTEL_BSP_EXPORT_TIMEOUT} + - name: OTEL_EXPORTER_OTLP_COMPRESSION + value: ${OTEL_EXPORTER_OTLP_COMPRESSION} + - name: OTEL_SPAN_ATTRIBUTE_COUNT_LIMIT + value: ${OTEL_SPAN_ATTRIBUTE_COUNT_LIMIT} + - name: OTEL_SPAN_ATTRIBUTE_VALUE_LENGTH_LIMIT + value: ${OTEL_SPAN_ATTRIBUTE_VALUE_LENGTH_LIMIT} + - name: IMAGE_TAG + value: ${IMAGE_TAG} resources: limits: cpu: ${CPU_LIMIT_EVALUATOR_UPLOAD} @@ -513,6 +725,36 @@ objects: value: ${USE_VMAAS_GO} - name: UNLEASH_BOOTSTRAP_FILE value: ${UNLEASH_BOOTSTRAP_FILE} + - name: OTEL_ENABLED + value: ${OTEL_ENABLED} + - name: OTEL_EXPORTER_OTLP_ENDPOINT + value: ${OTEL_EXPORTER_OTLP_ENDPOINT} + - name: OTEL_SAMPLING_RATE + value: ${OTEL_SAMPLING_RATE} + - name: OTEL_SQL_ENABLED + value: ${OTEL_SQL_ENABLED} + - name: OTEL_HTTP_OUTBOUND_ENABLED + value: ${OTEL_HTTP_OUTBOUND_ENABLED} + - name: OTEL_MQ_ENABLED + value: ${OTEL_MQ_ENABLED} + - name: OTEL_SQL_COMMENTER_ENABLED + value: ${OTEL_SQL_COMMENTER_ENABLED} + - name: OTEL_BSP_MAX_QUEUE_SIZE + value: ${OTEL_BSP_MAX_QUEUE_SIZE} + - name: OTEL_BSP_MAX_EXPORT_BATCH_SIZE + value: ${OTEL_BSP_MAX_EXPORT_BATCH_SIZE} + - name: OTEL_BSP_SCHEDULE_DELAY + value: ${OTEL_BSP_SCHEDULE_DELAY} + - name: OTEL_BSP_EXPORT_TIMEOUT + value: ${OTEL_BSP_EXPORT_TIMEOUT} + - name: OTEL_EXPORTER_OTLP_COMPRESSION + value: ${OTEL_EXPORTER_OTLP_COMPRESSION} + - name: OTEL_SPAN_ATTRIBUTE_COUNT_LIMIT + value: ${OTEL_SPAN_ATTRIBUTE_COUNT_LIMIT} + - name: OTEL_SPAN_ATTRIBUTE_VALUE_LENGTH_LIMIT + value: ${OTEL_SPAN_ATTRIBUTE_VALUE_LENGTH_LIMIT} + - name: IMAGE_TAG + value: ${IMAGE_TAG} resources: limits: cpu: ${CPU_LIMIT_VMAAS_SYNC} @@ -1050,3 +1292,48 @@ parameters: - name: KESSEL_AUTH_OIDC_ISSUER - name: KESSEL_INSECURE value: "FALSE" +- name: OTEL_ENABLED + description: Enable or disable OpenTelemetry tracing + value: "false" +- name: OTEL_EXPORTER_OTLP_ENDPOINT + description: OTLP HTTP exporter endpoint + value: "http://otel-collector:4318" +- name: OTEL_SAMPLING_RATE + description: Trace sampling rate (0.0 to 1.0) + value: "1.0" +- name: OTEL_SQL_ENABLED + description: Enable or disable SQL query tracing + value: "true" +- name: OTEL_SQL_COMMENTER_ENABLED + description: Add traceparent to SQL comments for pg_stat_activity visibility + value: "false" +- name: OTEL_HTTP_INBOUND_ENABLED + description: Enable or disable inbound HTTP request tracing + value: "true" +- name: OTEL_HTTP_OUTBOUND_ENABLED + description: Enable or disable outbound HTTP call tracing + value: "true" +- name: OTEL_MQ_ENABLED + description: Enable or disable Kafka message context propagation + value: "true" +- name: OTEL_BSP_MAX_QUEUE_SIZE + description: BatchSpanProcessor max queue size + value: "8192" +- name: OTEL_BSP_MAX_EXPORT_BATCH_SIZE + description: BatchSpanProcessor max export batch size + value: "256" +- name: OTEL_BSP_SCHEDULE_DELAY + description: BatchSpanProcessor schedule delay in milliseconds + value: "2000" +- name: OTEL_BSP_EXPORT_TIMEOUT + description: BatchSpanProcessor export timeout in milliseconds + value: "10000" +- name: OTEL_EXPORTER_OTLP_COMPRESSION + description: OTLP exporter compression (gzip, deflate, or none) + value: "gzip" +- name: OTEL_SPAN_ATTRIBUTE_COUNT_LIMIT + description: Maximum number of attributes per span + value: "64" +- name: OTEL_SPAN_ATTRIBUTE_VALUE_LENGTH_LIMIT + description: Maximum length of span attribute values + value: "1024" diff --git a/evaluator/evaluator.py b/evaluator/evaluator.py index 5a18bba57..eceaeb9c0 100644 --- a/evaluator/evaluator.py +++ b/evaluator/evaluator.py @@ -8,6 +8,7 @@ from aiokafka import ConsumerRecord from dateutil import parser +from opentelemetry import trace from psycopg_pool.pool_async import AsyncConnectionPool from common.constants import EvaluatorMessageType @@ -19,6 +20,15 @@ from common.mqueue import MQWriter from common.status_app import create_status_app from common.status_app import create_status_runner +from common.telemetry import extract_context_from_msg_headers +from common.telemetry import get_tracer +from common.telemetry import init_otel +from common.telemetry import instrument_aiohttp +from common.telemetry import instrument_outbound_http +from common.telemetry import instrument_psycopg +from common.telemetry import instrument_psycopg2 +from common.telemetry import threadctx +from common.telemetry import use_otel_context from common.utils import a_ensure_minimal_schema_version from common.utils import create_task_and_log @@ -28,6 +38,7 @@ from .processor import EvaluatorProcessor LOGGER = get_logger(__name__) +TRACER = get_tracer(__name__) class Evaluator: @@ -77,7 +88,27 @@ async def _consume_message(self, msg: ConsumerRecord): LOGGER.error("received unknown message type: %s", msg_type) return - await self.processor.evaluate_system(inventory_id, org_id, request_id, timestamp, recalc_event_id=recalc_event_id) + parent_ctx = extract_context_from_msg_headers(msg.headers) + with use_otel_context(parent_ctx): + with TRACER.start_as_current_span( + f"process {msg.topic}", + kind=trace.SpanKind.CONSUMER, + attributes={ + "messaging.system": "kafka", + "messaging.operation.name": "process", + "messaging.destination.name": msg.topic, + "vulnerability.inventory_id": inventory_id, + "vulnerability.msg_type": msg_type.value, + }, + ) as span: + if org_id: + span.set_attribute("rh.org_id", org_id) + threadctx.org_id = org_id + if request_id: + span.set_attribute("rh.request_id", request_id) + threadctx.request_id = request_id + + await self.processor.evaluate_system(inventory_id, org_id, request_id, timestamp, recalc_event_id=recalc_event_id) async def consume_message(self, msg: ConsumerRecord): """Consume message for evaluation, wrapper for semaphore""" @@ -121,6 +152,11 @@ async def stop(self): def main(): """Main""" init_logging() + init_otel(service_name="vulnerability-engine-evaluator") + instrument_psycopg2() + instrument_psycopg() + instrument_aiohttp() + instrument_outbound_http() initialize_unleash() loop = asyncio.new_event_loop() diff --git a/evaluator/processor.py b/evaluator/processor.py index 527e644b4..8a96a251a 100644 --- a/evaluator/processor.py +++ b/evaluator/processor.py @@ -12,6 +12,7 @@ from typing import Tuple import pytz +from opentelemetry.trace import StatusCode from prometheus_async.aio import time from psycopg import AsyncConnection from psycopg import sql @@ -21,6 +22,7 @@ from common.logging import get_logger from common.mqueue import MQWriter from common.peewee_model import VulnerabilityState +from common.telemetry import get_tracer from common.utils import executemany_fetchall from common.utils import send_inventory_views from common.utils import send_msg_to_payload_tracker @@ -47,6 +49,7 @@ from .logic import EvaluatorLogic LOGGER = get_logger(__name__) +TRACER = get_tracer(__name__) class EvaluatorProcessor: @@ -409,18 +412,28 @@ async def evaluate_system( """Evaluate single system""" EVAL_COUNT.inc() msg = {"platform_metadata": {"request_id": request_id}, "host": {"org_id": org_id, "id": inventory_id}} - try: - with EVAL_TIME.time(): + with ( + TRACER.start_as_current_span( + "evaluator _evaluate_system", + attributes={"vulnerability.inventory_id": inventory_id}, + ) as span, + EVAL_TIME.time(), + ): + try: LOGGER.info("evaluating system: %s, org_id: %s", inventory_id, org_id) await self._evaluate_system(inventory_id, org_id, request_id, request_timestamp, recalc_event_id=recalc_event_id) - except EvaluatorException as ex: - LOGGER.error(str(ex)) - send_msg_to_payload_tracker(self.payload_tracker, msg, "error", status_msg="evaluation failed", loop=self.loop) - return - except VmaasErrorException as ex: - LOGGER.error(str(ex)) - VMAAS_ERRORS_SKIP.inc() - send_msg_to_payload_tracker(self.payload_tracker, msg, "error", status_msg="evaluation failed", loop=self.loop) - return + except EvaluatorException as ex: + LOGGER.error(str(ex)) + span.set_status(StatusCode.ERROR, str(ex)) + span.record_exception(ex) + send_msg_to_payload_tracker(self.payload_tracker, msg, "error", status_msg="evaluation failed", loop=self.loop) + return + except VmaasErrorException as ex: + LOGGER.error(str(ex)) + span.set_status(StatusCode.ERROR, "vmaas_error") + span.record_exception(ex) + VMAAS_ERRORS_SKIP.inc() + send_msg_to_payload_tracker(self.payload_tracker, msg, "error", status_msg="evaluation failed", loop=self.loop) + return send_msg_to_payload_tracker(self.payload_tracker, msg, "success", status_msg="evaluation succeeded", loop=self.loop) diff --git a/grouper/common.py b/grouper/common.py index de31bf162..8a23b3373 100644 --- a/grouper/common.py +++ b/grouper/common.py @@ -6,6 +6,7 @@ from dataclasses import dataclass from enum import Enum +from opentelemetry import context as otel_context_api from prometheus_client import Counter from prometheus_client import Gauge @@ -61,6 +62,7 @@ class QueueItem: request_id: str + otel_context: otel_context_api.Context | None = None second_upload_event: asyncio.Event = None def __post_init__(self): diff --git a/grouper/grouper.py b/grouper/grouper.py index 0aab425dc..6dfb5b07e 100644 --- a/grouper/grouper.py +++ b/grouper/grouper.py @@ -14,6 +14,14 @@ from common.mqueue import MQReader from common.status_app import create_status_app from common.status_app import create_status_runner +from common.telemetry import extract_context_from_msg_headers +from common.telemetry import get_tracer +from common.telemetry import init_otel +from common.telemetry import instrument_aiohttp +from common.telemetry import instrument_outbound_http +from common.telemetry import instrument_psycopg +from common.telemetry import instrument_psycopg2 +from common.telemetry import use_otel_context from common.utils import create_task_and_log from .common import CFG @@ -21,6 +29,7 @@ from .queue import GrouperQueue LOGGER = get_logger(__name__) +TRACER = get_tracer(__name__) class Grouper: @@ -74,14 +83,37 @@ async def consume_message(self, msg: ConsumerRecord, unlock: asyncio.BoundedSema unlock.release() return - if msg_type is GrouperMessageType.INVENTORY_UPLOAD: - self.queue.push_inventory_msg( - org_id, inventory_id, reporter, changed, (msg_dict.get("platform_metadata") or {}).get("request_id", "") - ) - elif msg_type is GrouperMessageType.ADVISOR_UPLOAD: - self.queue.push_advisor_msg( - org_id, inventory_id, reporter, changed, (msg_dict.get("platform_metadata") or {}).get("request_id", "") - ) + parent_ctx = extract_context_from_msg_headers(msg.headers) + + with use_otel_context(parent_ctx): + with TRACER.start_as_current_span( + f"process {msg.topic}", + attributes={ + "messaging.system": "kafka", + "messaging.operation.name": "process", + "messaging.destination.name": msg.topic, + "vulnerability.inventory_id": inventory_id, + "vulnerability.msg_type": msg_type.value, + }, + ): + if msg_type is GrouperMessageType.INVENTORY_UPLOAD: + self.queue.push_inventory_msg( + org_id, + inventory_id, + reporter, + changed, + (msg_dict.get("platform_metadata") or {}).get("request_id", ""), + otel_context=parent_ctx, + ) + elif msg_type is GrouperMessageType.ADVISOR_UPLOAD: + self.queue.push_advisor_msg( + org_id, + inventory_id, + reporter, + changed, + (msg_dict.get("platform_metadata") or {}).get("request_id", ""), + otel_context=parent_ctx, + ) async def _start_grouping_inventory(self) -> None: """Start of the grouping inventory uploads""" @@ -111,6 +143,11 @@ async def run(self) -> None: def main() -> None: """Start service""" init_logging() + init_otel(service_name="vulnerability-engine-grouper") + instrument_psycopg2() + instrument_psycopg() + instrument_aiohttp() + instrument_outbound_http() loop = asyncio.new_event_loop() diff --git a/grouper/queue.py b/grouper/queue.py index 16ed60e22..f8b52abc4 100644 --- a/grouper/queue.py +++ b/grouper/queue.py @@ -8,9 +8,12 @@ from datetime import timezone from typing import Dict +from opentelemetry import context as otel_context_api + from common.constants import EvaluatorMessageType from common.logging import get_logger from common.mqueue import MQWriter +from common.telemetry import use_otel_context from common.utils import create_task_and_log from common.utils import send_msg_to_payload_tracker @@ -53,14 +56,22 @@ async def stop(self): await self.payload_tracker.stop() await self.evaluator.stop() - def push_inventory_msg(self, org_id: str, inventory_id: str, reporter: str, inventory_changed: bool, request_id: str) -> None: + def push_inventory_msg( + self, + org_id: str, + inventory_id: str, + reporter: str, + inventory_changed: bool, + request_id: str, + otel_context: otel_context_api.Context | None = None, + ) -> None: """Push inventory upload message to queue""" is_updated = False item = self._queue.get(inventory_id) if not item: LOGGER.info("pushing listener upload to queue for system: %s, org_id: %s", inventory_id, org_id) - item = QueueItem(True, inventory_changed, False, False, request_id) + item = QueueItem(True, inventory_changed, False, False, request_id, otel_context=otel_context) self._queue[inventory_id] = item else: is_updated = True @@ -73,6 +84,8 @@ def push_inventory_msg(self, org_id: str, inventory_id: str, reporter: str, inve item.inventory_upload = True item.inventory_changed = inventory_changed item.request_id = request_id + if otel_context is not None: + item.otel_context = otel_context if item.inventory_upload and item.advisor_upload: LOGGER.info("obtained both uploads for system: %s, account: %s, releasing lock", inventory_id, org_id) @@ -81,14 +94,22 @@ def push_inventory_msg(self, org_id: str, inventory_id: str, reporter: str, inve QUEUE_SIZE.inc() create_task_and_log(self._start_item_processing(org_id, inventory_id, reporter), LOGGER, self.loop) - def push_advisor_msg(self, org_id: str, inventory_id: str, reporter: str, advisor_changed: bool, request_id: str) -> None: + def push_advisor_msg( + self, + org_id: str, + inventory_id: str, + reporter: str, + advisor_changed: bool, + request_id: str, + otel_context: otel_context_api.Context | None = None, + ) -> None: """Push advisor message to queue""" is_updated = False item = self._queue.get(inventory_id) if not item: LOGGER.info("pushing advisor upload to queue for system: %s, org_id: %s", inventory_id, org_id) - item = QueueItem(False, False, True, advisor_changed, request_id) + item = QueueItem(False, False, True, advisor_changed, request_id, otel_context=otel_context) self._queue[inventory_id] = item else: is_updated = True @@ -101,6 +122,8 @@ def push_advisor_msg(self, org_id: str, inventory_id: str, reporter: str, adviso item.advisor_upload = True item.advisor_changed = advisor_changed item.request_id = request_id + if otel_context is not None and item.otel_context is None: + item.otel_context = otel_context if item.inventory_upload and item.advisor_upload: LOGGER.info("obtained both messages for system: %s, account: %s, releasing lock", inventory_id, org_id) @@ -171,4 +194,5 @@ async def _send_for_evaluation(self, item: QueueItem, org_id: str, inventory_id: send_msg_to_payload_tracker( self.payload_tracker, msg, "processing", status_msg="changed system, sending to evaluator", loop=self.loop ) - self.evaluator.send(msg) + with use_otel_context(item.otel_context): + self.evaluator.send(msg) diff --git a/listener/listener.py b/listener/listener.py index dab63bdb6..1cb00b548 100644 --- a/listener/listener.py +++ b/listener/listener.py @@ -7,6 +7,7 @@ import signal from aiokafka.structs import ConsumerRecord +from opentelemetry import trace from psycopg_pool import AsyncConnectionPool from common.database_handler import setup_async_db_pool @@ -16,6 +17,15 @@ from common.mqueue import MQWriter from common.status_app import create_status_app from common.status_app import create_status_runner +from common.telemetry import extract_context_from_msg_headers +from common.telemetry import get_tracer +from common.telemetry import init_otel +from common.telemetry import instrument_aiohttp +from common.telemetry import instrument_outbound_http +from common.telemetry import instrument_psycopg +from common.telemetry import instrument_psycopg2 +from common.telemetry import threadctx +from common.telemetry import use_otel_context from common.utils import a_ensure_minimal_schema_version from common.utils import create_task_and_log from common.utils import send_msg_to_payload_tracker @@ -33,6 +43,7 @@ from .inventory_processor import InventoryMsgProcessor LOGGER = get_logger(__name__) +TRACER = get_tracer(__name__) class InvalidInventoryMsg(Exception): @@ -208,11 +219,41 @@ def _consume_message(self, msg: ConsumerRecord): LOGGER.exception("Unable to parse message: %s", exc) return + parent_ctx = extract_context_from_msg_headers(msg.headers) + if msg_dict.get("host") or msg_dict.get("type") == "delete": - create_task_and_log(self.consume_inventory_msg(msg_dict), LOGGER, self.loop) + with use_otel_context(parent_ctx): + with TRACER.start_as_current_span( + f"process {msg.topic}", + kind=trace.SpanKind.CONSUMER, + attributes={ + "messaging.system": "kafka", + "messaging.operation.name": "process", + "messaging.destination.name": msg.topic, + }, + ) as span: + org_id = msg_dict.get("host", {}).get("org_id") or msg_dict.get("org_id") + if org_id: + span.set_attribute("rh.org_id", org_id) + threadctx.org_id = org_id + request_id = (msg_dict.get("platform_metadata") or {}).get("request_id") + if request_id: + span.set_attribute("rh.request_id", request_id) + threadctx.request_id = request_id + create_task_and_log(self.consume_inventory_msg(msg_dict), LOGGER, self.loop) PROCESS_MESSAGES.inc() elif msg_dict.get("input"): - create_task_and_log(self.consume_advisor_msg(msg_dict), LOGGER, self.loop) + with use_otel_context(parent_ctx): + with TRACER.start_as_current_span( + f"process {msg.topic}", + kind=trace.SpanKind.CONSUMER, + attributes={ + "messaging.system": "kafka", + "messaging.operation.name": "process", + "messaging.destination.name": msg.topic, + }, + ): + create_task_and_log(self.consume_advisor_msg(msg_dict), LOGGER, self.loop) PROCESS_MESSAGES.inc() else: LOGGER.exception("Unknown message obtained: %s", msg) @@ -250,6 +291,11 @@ async def run(self): def main(): """Main""" init_logging() + init_otel(service_name="vulnerability-engine-listener") + instrument_psycopg2() + instrument_psycopg() + instrument_aiohttp() + instrument_outbound_http() loop = asyncio.new_event_loop() loop.run_until_complete(a_ensure_minimal_schema_version()) diff --git a/manager/gunicorn_conf.py b/manager/gunicorn_conf.py index c3da28abc..db1bab753 100644 --- a/manager/gunicorn_conf.py +++ b/manager/gunicorn_conf.py @@ -8,6 +8,7 @@ from prometheus_client import multiprocess from common.logging import setup_cw_logging +from common.telemetry import init_otel def get_account(headers: list) -> str: @@ -99,6 +100,10 @@ def child_exit(server, worker): multiprocess.mark_process_dead(worker.pid) +def post_fork(server, worker): + init_otel(service_name="vulnerability-engine-manager") + + accesslog="-" access_log_format="%(t)s %(x)s %(h)s %(s)s %(m)s %(U)s (%(q)s) %(L)ss" logger_class="manager.gunicorn_conf.CustomLogger" diff --git a/manager/main.py b/manager/main.py index 63b73e617..ea8e9718d 100755 --- a/manager/main.py +++ b/manager/main.py @@ -24,6 +24,10 @@ from common.logging import init_logging from common.peewee_database import DB from common.peewee_database import DB_READ_REPLICA +from common.telemetry import init_otel +from common.telemetry import instrument_flask_app +from common.telemetry import instrument_outbound_http +from common.telemetry import instrument_psycopg2 from .base import MissingEntitlementException from .base import forbidden_missing_entitlement @@ -122,10 +126,15 @@ def _on_first_request(): app.add_middleware(ErrorHandlerMiddleware, position=connexion.middleware.MiddlewarePosition.BEFORE_EXCEPTION) app.add_middleware(RequestTimeoutMiddleware, position=connexion.middleware.MiddlewarePosition.BEFORE_EXCEPTION) + instrument_flask_app(app.app) + return ASGIMiddleware(app) init_logging(num_servers=int(CFG.gunicorn_workers)) +init_otel(service_name="vulnerability-engine-manager") +instrument_psycopg2() +instrument_outbound_http() initialize_unleash() # gunicorn expects an object called 'application' hence the pylint disable application = create_app({CFG.default_route: 'manager.spec.yaml', # pylint: disable=invalid-name diff --git a/notificator/notificator.py b/notificator/notificator.py index 6efe0a1b3..ac38a91be 100644 --- a/notificator/notificator.py +++ b/notificator/notificator.py @@ -307,4 +307,12 @@ def main(): if __name__ == "__main__": init_logging() + + from common.telemetry import init_otel + from common.telemetry import instrument_outbound_http + from common.telemetry import instrument_psycopg2 + + init_otel(service_name="vulnerability-engine-notificator") + instrument_psycopg2() + instrument_outbound_http() main() diff --git a/poetry.lock b/poetry.lock index bb0ff1b6a..a92622cae 100644 --- a/poetry.lock +++ b/poetry.lock @@ -1450,6 +1450,24 @@ rsa = ["rsa (>=4.0.0,<5)"] testing = ["aiohttp (>=3.8.0,<4.0.0) ; python_version < \"3.14\"", "aiohttp (>=3.9.0,<4.0.0) ; python_version >= \"3.14\"", "aioresponses", "flask", "freezegun", "grpcio (>=1.59.0,<2.0.0) ; python_version < \"3.14\"", "grpcio (>=1.75.1,<2.0.0) ; python_version >= \"3.14\"", "packaging (>=20.0)", "pyjwt (>=2.0)", "pytest", "pytest-asyncio", "pytest-cov", "pytest-localserver", "pyu2f (>=0.1.5)", "requests (>=2.30.0,<3.0.0)", "responses", "urllib3 (>=1.26.15,<3.0.0)"] urllib3 = ["packaging (>=20.0)", "urllib3 (>=1.26.15,<3.0.0)"] +[[package]] +name = "googleapis-common-protos" +version = "1.75.1" +description = "Common protobufs used in Google APIs" +optional = false +python-versions = ">=3.10" +groups = ["main"] +files = [ + {file = "googleapis_common_protos-1.75.1-py3-none-any.whl", hash = "sha256:28a1934bcd33b9c9da66ac301a0a4227e3367f095a17d0375cb98f0a09d93b79"}, + {file = "googleapis_common_protos-1.75.1.tar.gz", hash = "sha256:d3042c6c5a2d4e67113104d6b6818b59b6bd92a197f2a91508e801fe815cf071"}, +] + +[package.dependencies] +protobuf = ">=6.33.5,<8.0.0" + +[package.extras] +grpc = ["grpcio (>=1.59.0,<2.0.0)"] + [[package]] name = "grpcio" version = "1.83.0" @@ -2341,6 +2359,305 @@ rsa = ["cryptography (>=3.0.0)"] signals = ["blinker (>=1.4.0)"] signedtoken = ["cryptography (>=3.0.0)", "pyjwt (>=2.0.0,<3)"] +[[package]] +name = "opentelemetry-api" +version = "1.44.0" +description = "OpenTelemetry Python API" +optional = false +python-versions = ">=3.10" +groups = ["main"] +files = [ + {file = "opentelemetry_api-1.44.0-py3-none-any.whl", hash = "sha256:94b98c893a91b88657eaac1e3ba89618cdb85be6918196705354f34728b2cdef"}, + {file = "opentelemetry_api-1.44.0.tar.gz", hash = "sha256:67647e5e9566edcf421166fdf022b3537f818635daa852b289e34604dc6fb33a"}, +] + +[package.dependencies] +typing-extensions = ">=4.5.0" + +[[package]] +name = "opentelemetry-exporter-otlp-proto-common" +version = "1.44.0" +description = "OpenTelemetry Protobuf encoding" +optional = false +python-versions = ">=3.10" +groups = ["main"] +files = [ + {file = "opentelemetry_exporter_otlp_proto_common-1.44.0-py3-none-any.whl", hash = "sha256:9a9fe61bba73d802904bc989f1d6b4a7b1ee40f06c40e98d6f85af65aaebb694"}, + {file = "opentelemetry_exporter_otlp_proto_common-1.44.0.tar.gz", hash = "sha256:dc87a5a5bc58f149a56d1547e4691588fa12994cdc3bc039a694ccb3375862ac"}, +] + +[package.dependencies] +opentelemetry-proto = "1.44.0" + +[[package]] +name = "opentelemetry-exporter-otlp-proto-http" +version = "1.44.0" +description = "OpenTelemetry Collector Protobuf over HTTP Exporter" +optional = false +python-versions = ">=3.10" +groups = ["main"] +files = [ + {file = "opentelemetry_exporter_otlp_proto_http-1.44.0-py3-none-any.whl", hash = "sha256:838592fce774c1c8bb7b9a0a7facbfa82e17be5a8a4e94cef10cb84ae026bae3"}, + {file = "opentelemetry_exporter_otlp_proto_http-1.44.0.tar.gz", hash = "sha256:c633d7270ad6b57cd4cfbe8b0007a9e2e7c0cb50bd6c50fe2a7b245f721a09d8"}, +] + +[package.dependencies] +googleapis-common-protos = ">=1.52,<2.0" +opentelemetry-api = ">=1.15,<2.0" +opentelemetry-exporter-otlp-proto-common = "1.44.0" +opentelemetry-proto = "1.44.0" +opentelemetry-sdk = ">=1.44.0,<1.45.0" +requests = ">=2.7,<3.0" +typing-extensions = ">=4.5.0" + +[package.extras] +gcp-auth = ["opentelemetry-exporter-credential-provider-gcp (>=0.59b0)"] + +[[package]] +name = "opentelemetry-instrumentation" +version = "0.65b0" +description = "Instrumentation Tools & Auto Instrumentation for OpenTelemetry Python" +optional = false +python-versions = ">=3.10" +groups = ["main"] +files = [ + {file = "opentelemetry_instrumentation-0.65b0-py3-none-any.whl", hash = "sha256:ea967a72b9939b5fcfdad572753b4306c59dcb99e3f382d95dae04286805e137"}, + {file = "opentelemetry_instrumentation-0.65b0.tar.gz", hash = "sha256:071d9d9eced9bd6460444ec3b0c77229870ed05a881c22c84fdede58e4eed09b"}, +] + +[package.dependencies] +opentelemetry-api = ">=1.4,<2.0" +opentelemetry-semantic-conventions = "0.65b0" +packaging = ">=18.0" +wrapt = ">=1.0.0,<3.0.0" + +[[package]] +name = "opentelemetry-instrumentation-aiohttp-client" +version = "0.65b0" +description = "OpenTelemetry aiohttp client instrumentation" +optional = false +python-versions = ">=3.10" +groups = ["main"] +files = [ + {file = "opentelemetry_instrumentation_aiohttp_client-0.65b0-py3-none-any.whl", hash = "sha256:3a060efa53fa44d02ba7372a7ed2b42cdfa6be6df81b089845067ad840e25729"}, + {file = "opentelemetry_instrumentation_aiohttp_client-0.65b0.tar.gz", hash = "sha256:85906a2806ee5641756b5c33274e9aa75c3cc2441e3b830aa5804cf0e1fa9dd1"}, +] + +[package.dependencies] +opentelemetry-api = ">=1.12,<2.0" +opentelemetry-instrumentation = "0.65b0" +opentelemetry-semantic-conventions = "0.65b0" +opentelemetry-util-http = "0.65b0" +wrapt = ">=1.0.0,<3.0.0" + +[package.extras] +instruments = ["aiohttp (>=3.0,<4.0)"] + +[[package]] +name = "opentelemetry-instrumentation-dbapi" +version = "0.65b0" +description = "OpenTelemetry Database API instrumentation" +optional = false +python-versions = ">=3.10" +groups = ["main"] +files = [ + {file = "opentelemetry_instrumentation_dbapi-0.65b0-py3-none-any.whl", hash = "sha256:50b662578a6903b028e09b73f604de687752f5f904aa0ca032969157b29d60f2"}, + {file = "opentelemetry_instrumentation_dbapi-0.65b0.tar.gz", hash = "sha256:da048bb683347ddad2f47344bacfe1e111bf7bfb2e5a39796b1083679ad4f0f3"}, +] + +[package.dependencies] +opentelemetry-api = ">=1.12,<2.0" +opentelemetry-instrumentation = "0.65b0" +opentelemetry-semantic-conventions = "0.65b0" +wrapt = ">=1.0.0,<3.0.0" + +[[package]] +name = "opentelemetry-instrumentation-flask" +version = "0.65b0" +description = "Flask instrumentation for OpenTelemetry" +optional = false +python-versions = ">=3.10" +groups = ["main"] +files = [ + {file = "opentelemetry_instrumentation_flask-0.65b0-py3-none-any.whl", hash = "sha256:d5337dac3b2af7f658fbc11c879667c9978910e38744b9706508f0b9908f7841"}, + {file = "opentelemetry_instrumentation_flask-0.65b0.tar.gz", hash = "sha256:887de3a97c09953da09ae713fbb777172900f33b2924d85dad314a033156ef66"}, +] + +[package.dependencies] +opentelemetry-api = ">=1.12,<2.0" +opentelemetry-instrumentation = "0.65b0" +opentelemetry-instrumentation-wsgi = "0.65b0" +opentelemetry-semantic-conventions = "0.65b0" +opentelemetry-util-http = "0.65b0" +packaging = ">=21.0" + +[package.extras] +instruments = ["flask (>=1.0)"] + +[[package]] +name = "opentelemetry-instrumentation-psycopg" +version = "0.65b0" +description = "OpenTelemetry psycopg instrumentation" +optional = false +python-versions = ">=3.10" +groups = ["main"] +files = [ + {file = "opentelemetry_instrumentation_psycopg-0.65b0-py3-none-any.whl", hash = "sha256:a363c594a0dfcb1f6c8db5b1a1549a2e7dfde91f4d17e5fe0b48e777119705f9"}, + {file = "opentelemetry_instrumentation_psycopg-0.65b0.tar.gz", hash = "sha256:62921ceaad2a0d0813a4185ce724aba40e583ead52db03bc3506de5ef53cab84"}, +] + +[package.dependencies] +opentelemetry-api = ">=1.12,<2.0" +opentelemetry-instrumentation = "0.65b0" +opentelemetry-instrumentation-dbapi = "0.65b0" + +[package.extras] +instruments = ["psycopg (>=3.1.0)"] + +[[package]] +name = "opentelemetry-instrumentation-psycopg2" +version = "0.65b0" +description = "OpenTelemetry psycopg2 instrumentation" +optional = false +python-versions = ">=3.10" +groups = ["main"] +files = [ + {file = "opentelemetry_instrumentation_psycopg2-0.65b0-py3-none-any.whl", hash = "sha256:91880c7dbcd2b9cc62694894abed7f69fdad4bae472d1d3664a82650294f9836"}, + {file = "opentelemetry_instrumentation_psycopg2-0.65b0.tar.gz", hash = "sha256:4eba60bef5f25d163a098109c2960773f3753e0b7e39824b9f398ba49ffab783"}, +] + +[package.dependencies] +opentelemetry-api = ">=1.12,<2.0" +opentelemetry-instrumentation = "0.65b0" +opentelemetry-instrumentation-dbapi = "0.65b0" + +[package.extras] +instruments-any = ["psycopg2 (>=2.7.3.1)", "psycopg2-binary (>=2.7.3.1)"] + +[[package]] +name = "opentelemetry-instrumentation-requests" +version = "0.65b0" +description = "OpenTelemetry requests instrumentation" +optional = false +python-versions = ">=3.10" +groups = ["main"] +files = [ + {file = "opentelemetry_instrumentation_requests-0.65b0-py3-none-any.whl", hash = "sha256:91688ec0d4d1fed75ea8d026ef2c66274ed9868c22b6be211ef85d832d16f957"}, + {file = "opentelemetry_instrumentation_requests-0.65b0.tar.gz", hash = "sha256:1d601548f89236d5ab373c7208a2e1e162a8d6462b5b972f9ad8fb0ed82d7438"}, +] + +[package.dependencies] +opentelemetry-api = ">=1.12,<2.0" +opentelemetry-instrumentation = "0.65b0" +opentelemetry-semantic-conventions = "0.65b0" +opentelemetry-util-http = "0.65b0" + +[package.extras] +instruments = ["requests (>=2.0,<3.0)"] + +[[package]] +name = "opentelemetry-instrumentation-sqlalchemy" +version = "0.65b0" +description = "OpenTelemetry SQLAlchemy instrumentation" +optional = false +python-versions = ">=3.10" +groups = ["main"] +files = [ + {file = "opentelemetry_instrumentation_sqlalchemy-0.65b0-py3-none-any.whl", hash = "sha256:4d8a2e5afc7b505a48d05cb1fb6db5f8b31681814c1d7767bd6d4b5bdd3a3047"}, + {file = "opentelemetry_instrumentation_sqlalchemy-0.65b0.tar.gz", hash = "sha256:8ec2e79f1e00808c5dc639ab5b2cfcdf9dcdef55efd6442c6019707fe2894028"}, +] + +[package.dependencies] +opentelemetry-api = ">=1.12,<2.0" +opentelemetry-instrumentation = "0.65b0" +opentelemetry-semantic-conventions = "0.65b0" +packaging = ">=21.0" +wrapt = ">=1.11.2" + +[package.extras] +instruments = ["sqlalchemy (>=1.0.0,<2.1.0)"] + +[[package]] +name = "opentelemetry-instrumentation-wsgi" +version = "0.65b0" +description = "WSGI Middleware for OpenTelemetry" +optional = false +python-versions = ">=3.10" +groups = ["main"] +files = [ + {file = "opentelemetry_instrumentation_wsgi-0.65b0-py3-none-any.whl", hash = "sha256:af23e6686c7cd2abcd7d14ac03fb7e3b438273eb2d54a8f8dc401dc71bc52a9f"}, + {file = "opentelemetry_instrumentation_wsgi-0.65b0.tar.gz", hash = "sha256:d4a62ae98667ddfe04fe538c3c54abad538feb8c9c7a407ba19f016e1ce4a89a"}, +] + +[package.dependencies] +opentelemetry-api = ">=1.12,<2.0" +opentelemetry-instrumentation = "0.65b0" +opentelemetry-semantic-conventions = "0.65b0" +opentelemetry-util-http = "0.65b0" + +[[package]] +name = "opentelemetry-proto" +version = "1.44.0" +description = "OpenTelemetry Python Proto" +optional = false +python-versions = ">=3.10" +groups = ["main"] +files = [ + {file = "opentelemetry_proto-1.44.0-py3-none-any.whl", hash = "sha256:898b155a0e1557afd867478fb6158e8122a46329ca0bb8dc53cc55e98f017f56"}, + {file = "opentelemetry_proto-1.44.0.tar.gz", hash = "sha256:c547a79c2f8c0c515d31509154682e5921c7cfd5ca67b70e1f9266e2c3e103f3"}, +] + +[package.dependencies] +protobuf = ">=5.0,<8.0" + +[[package]] +name = "opentelemetry-sdk" +version = "1.44.0" +description = "OpenTelemetry Python SDK" +optional = false +python-versions = ">=3.10" +groups = ["main"] +files = [ + {file = "opentelemetry_sdk-1.44.0-py3-none-any.whl", hash = "sha256:df081c4c6bcfdb1211e3e86140376792643128a25f8d72d1d27675936e7e96ad"}, + {file = "opentelemetry_sdk-1.44.0.tar.gz", hash = "sha256:cebe7f65dc12f26ead75c6064de12fd2a9052e5060c0272d402cfa203aae123b"}, +] + +[package.dependencies] +opentelemetry-api = "1.44.0" +opentelemetry-semantic-conventions = "0.65b0" +typing-extensions = ">=4.5.0" + +[package.extras] +file-configuration = ["opentelemetry-configuration (==0.65b0)"] + +[[package]] +name = "opentelemetry-semantic-conventions" +version = "0.65b0" +description = "OpenTelemetry Semantic Conventions" +optional = false +python-versions = ">=3.10" +groups = ["main"] +files = [ + {file = "opentelemetry_semantic_conventions-0.65b0-py3-none-any.whl", hash = "sha256:1cacde7b0ad306f84c5ef08c3dbe1bbaf20165bba6f8bff43b670e555a086bcb"}, + {file = "opentelemetry_semantic_conventions-0.65b0.tar.gz", hash = "sha256:f9b2b81e9d5b64f11bc952075e7e9c7fb0aab075c7fd1c46d597f1b919852d60"}, +] + +[package.dependencies] +opentelemetry-api = "1.44.0" +typing-extensions = ">=4.5.0" + +[[package]] +name = "opentelemetry-util-http" +version = "0.65b0" +description = "Web util for OpenTelemetry" +optional = false +python-versions = ">=3.10" +groups = ["main"] +files = [ + {file = "opentelemetry_util_http-0.65b0-py3-none-any.whl", hash = "sha256:7553b606f963097cb190536dc30556cce85090692e471a422fff30ca29b04348"}, + {file = "opentelemetry_util_http-0.65b0.tar.gz", hash = "sha256:84f82d826978bba416ab453460ff6a7391cdc3534c93a786595e4068680016b7"}, +] + [[package]] name = "packaging" version = "26.2" @@ -3789,4 +4106,4 @@ type = ["pytest-mypy (>=1.0.1) ; platform_python_implementation != \"PyPy\""] [metadata] lock-version = "2.1" python-versions = ">=3.12,<3.14" -content-hash = "7f468ce039730bf37bbf209eed441ab733e409dc0f5a85816f3a2e702f341508" +content-hash = "31d13d686be3946ce7cf476d620e2b70ec61101fd98b231a279f2e76660de5e0" diff --git a/pyproject.toml b/pyproject.toml index af10bc3ef..025ece9cd 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -54,6 +54,16 @@ kessel-sdk = {extras = ["auth"], version = "^2.1.0"} cython = "<3.2.0" # maturin@1.13.2 requires rustc 1.89 (not in RHEL 9 yet) maturin = "=1.13.1" +# OpenTelemetry +opentelemetry-api = "~1.44.0" +opentelemetry-sdk = "~1.44.0" +opentelemetry-exporter-otlp-proto-http = "~1.44.0" +opentelemetry-instrumentation-flask = "0.65b0" +opentelemetry-instrumentation-requests = "0.65b0" +opentelemetry-instrumentation-sqlalchemy = "0.65b0" +opentelemetry-instrumentation-psycopg2 = "0.65b0" +opentelemetry-instrumentation-psycopg = "0.65b0" +opentelemetry-instrumentation-aiohttp-client = "0.65b0" [tool.poetry.group.dev.dependencies] pre-commit = "^4.0.0" diff --git a/requirements-build.txt b/requirements-build.txt index 083dbeff6..6592c3bb1 100644 --- a/requirements-build.txt +++ b/requirements-build.txt @@ -313,14 +313,18 @@ flit-core==3.12.0 \ # click # coherent-licensed # flask - # idna # itsdangerous # jinja2 - # packaging - # pathspec # typing-extensions # werkzeug # wheel +flit-core==4.0.2 \ + --hash=sha256:8d80717e28d982b41594ed5b14d4e89fbc5bad55473fde27994a3101db18a94e \ + --hash=sha256:b6929defd93884b584d7c87829e0e7b5c26ed6be17b0b873979019314aa841c8 + # via + # idna + # packaging + # pathspec hatch-fancy-pypi-readme==25.1.0 \ --hash=sha256:9c58ed3dff90d51f43414ce37009ad1d5b0f08ffc9fc216998a06380f01c0045 \ --hash=sha256:ce0134c40d63d874ac48f48ccc678b8f3b62b8e50e9318520d2bffc752eedaf3 @@ -355,6 +359,22 @@ hatchling==1.31.0 \ # httpx # jsonschema # jsonschema-specifications + # opentelemetry-api + # opentelemetry-exporter-otlp-proto-common + # opentelemetry-exporter-otlp-proto-http + # opentelemetry-instrumentation + # opentelemetry-instrumentation-aiohttp-client + # opentelemetry-instrumentation-dbapi + # opentelemetry-instrumentation-flask + # opentelemetry-instrumentation-psycopg + # opentelemetry-instrumentation-psycopg2 + # opentelemetry-instrumentation-requests + # opentelemetry-instrumentation-sqlalchemy + # opentelemetry-instrumentation-wsgi + # opentelemetry-proto + # opentelemetry-sdk + # opentelemetry-semantic-conventions + # opentelemetry-util-http # platformdirs # prometheus-async # python-multipart @@ -493,6 +513,7 @@ setuptools==83.0.0 \ # cryptography # frozenlist # gitpython + # googleapis-common-protos # grpcio # gunicorn # importlib-metadata diff --git a/requirements.txt b/requirements.txt index 27c57eadd..fa30ee097 100644 --- a/requirements.txt +++ b/requirements.txt @@ -824,6 +824,9 @@ gitpython==3.1.57 ; python_version >= "3.12" and python_version < "3.14" \ google-auth==2.56.2 ; python_version >= "3.12" and python_version < "3.14" \ --hash=sha256:c8270ea95b2697b74e3d8438ae9c5b898e38b623b915c7b5c5635921e7de68a6 \ --hash=sha256:e28f103ca8091fb7012b99c44243d7366c29863713b8e34a220c3322b7a07051 +googleapis-common-protos==1.75.1 ; python_version >= "3.12" and python_version < "3.14" \ + --hash=sha256:28a1934bcd33b9c9da66ac301a0a4227e3367f095a17d0375cb98f0a09d93b79 \ + --hash=sha256:d3042c6c5a2d4e67113104d6b6818b59b6bd92a197f2a91508e801fe815cf071 grpcio==1.83.0 ; python_version >= "3.12" and python_version < "3.14" \ --hash=sha256:009667eaf3dcd5224c713589cdc98e7ca4ed0ff0b61132c6b276e930eb83a2df \ --hash=sha256:10b3fa0475eb572c9a81a6fe37fa16a9c500c0c91cfc148cac15692b7e3c2867 \ @@ -1278,6 +1281,54 @@ multidict==6.7.1 ; python_version >= "3.12" and python_version < "3.14" \ oauthlib==3.3.1 ; python_version >= "3.12" and python_version < "3.14" \ --hash=sha256:0f0f8aa759826a193cf66c12ea1af1637f87b9b4622d46e866952bb022e538c9 \ --hash=sha256:88119c938d2b8fb88561af5f6ee0eec8cc8d552b7bb1f712743136eb7523b7a1 +opentelemetry-api==1.44.0 ; python_version >= "3.12" and python_version < "3.14" \ + --hash=sha256:67647e5e9566edcf421166fdf022b3537f818635daa852b289e34604dc6fb33a \ + --hash=sha256:94b98c893a91b88657eaac1e3ba89618cdb85be6918196705354f34728b2cdef +opentelemetry-exporter-otlp-proto-common==1.44.0 ; python_version >= "3.12" and python_version < "3.14" \ + --hash=sha256:9a9fe61bba73d802904bc989f1d6b4a7b1ee40f06c40e98d6f85af65aaebb694 \ + --hash=sha256:dc87a5a5bc58f149a56d1547e4691588fa12994cdc3bc039a694ccb3375862ac +opentelemetry-exporter-otlp-proto-http==1.44.0 ; python_version >= "3.12" and python_version < "3.14" \ + --hash=sha256:838592fce774c1c8bb7b9a0a7facbfa82e17be5a8a4e94cef10cb84ae026bae3 \ + --hash=sha256:c633d7270ad6b57cd4cfbe8b0007a9e2e7c0cb50bd6c50fe2a7b245f721a09d8 +opentelemetry-instrumentation-aiohttp-client==0.65b0 ; python_version >= "3.12" and python_version < "3.14" \ + --hash=sha256:3a060efa53fa44d02ba7372a7ed2b42cdfa6be6df81b089845067ad840e25729 \ + --hash=sha256:85906a2806ee5641756b5c33274e9aa75c3cc2441e3b830aa5804cf0e1fa9dd1 +opentelemetry-instrumentation-dbapi==0.65b0 ; python_version >= "3.12" and python_version < "3.14" \ + --hash=sha256:50b662578a6903b028e09b73f604de687752f5f904aa0ca032969157b29d60f2 \ + --hash=sha256:da048bb683347ddad2f47344bacfe1e111bf7bfb2e5a39796b1083679ad4f0f3 +opentelemetry-instrumentation-flask==0.65b0 ; python_version >= "3.12" and python_version < "3.14" \ + --hash=sha256:887de3a97c09953da09ae713fbb777172900f33b2924d85dad314a033156ef66 \ + --hash=sha256:d5337dac3b2af7f658fbc11c879667c9978910e38744b9706508f0b9908f7841 +opentelemetry-instrumentation-psycopg2==0.65b0 ; python_version >= "3.12" and python_version < "3.14" \ + --hash=sha256:4eba60bef5f25d163a098109c2960773f3753e0b7e39824b9f398ba49ffab783 \ + --hash=sha256:91880c7dbcd2b9cc62694894abed7f69fdad4bae472d1d3664a82650294f9836 +opentelemetry-instrumentation-psycopg==0.65b0 ; python_version >= "3.12" and python_version < "3.14" \ + --hash=sha256:62921ceaad2a0d0813a4185ce724aba40e583ead52db03bc3506de5ef53cab84 \ + --hash=sha256:a363c594a0dfcb1f6c8db5b1a1549a2e7dfde91f4d17e5fe0b48e777119705f9 +opentelemetry-instrumentation-requests==0.65b0 ; python_version >= "3.12" and python_version < "3.14" \ + --hash=sha256:1d601548f89236d5ab373c7208a2e1e162a8d6462b5b972f9ad8fb0ed82d7438 \ + --hash=sha256:91688ec0d4d1fed75ea8d026ef2c66274ed9868c22b6be211ef85d832d16f957 +opentelemetry-instrumentation-sqlalchemy==0.65b0 ; python_version >= "3.12" and python_version < "3.14" \ + --hash=sha256:4d8a2e5afc7b505a48d05cb1fb6db5f8b31681814c1d7767bd6d4b5bdd3a3047 \ + --hash=sha256:8ec2e79f1e00808c5dc639ab5b2cfcdf9dcdef55efd6442c6019707fe2894028 +opentelemetry-instrumentation-wsgi==0.65b0 ; python_version >= "3.12" and python_version < "3.14" \ + --hash=sha256:af23e6686c7cd2abcd7d14ac03fb7e3b438273eb2d54a8f8dc401dc71bc52a9f \ + --hash=sha256:d4a62ae98667ddfe04fe538c3c54abad538feb8c9c7a407ba19f016e1ce4a89a +opentelemetry-instrumentation==0.65b0 ; python_version >= "3.12" and python_version < "3.14" \ + --hash=sha256:071d9d9eced9bd6460444ec3b0c77229870ed05a881c22c84fdede58e4eed09b \ + --hash=sha256:ea967a72b9939b5fcfdad572753b4306c59dcb99e3f382d95dae04286805e137 +opentelemetry-proto==1.44.0 ; python_version >= "3.12" and python_version < "3.14" \ + --hash=sha256:898b155a0e1557afd867478fb6158e8122a46329ca0bb8dc53cc55e98f017f56 \ + --hash=sha256:c547a79c2f8c0c515d31509154682e5921c7cfd5ca67b70e1f9266e2c3e103f3 +opentelemetry-sdk==1.44.0 ; python_version >= "3.12" and python_version < "3.14" \ + --hash=sha256:cebe7f65dc12f26ead75c6064de12fd2a9052e5060c0272d402cfa203aae123b \ + --hash=sha256:df081c4c6bcfdb1211e3e86140376792643128a25f8d72d1d27675936e7e96ad +opentelemetry-semantic-conventions==0.65b0 ; python_version >= "3.12" and python_version < "3.14" \ + --hash=sha256:1cacde7b0ad306f84c5ef08c3dbe1bbaf20165bba6f8bff43b670e555a086bcb \ + --hash=sha256:f9b2b81e9d5b64f11bc952075e7e9c7fb0aab075c7fd1c46d597f1b919852d60 +opentelemetry-util-http==0.65b0 ; python_version >= "3.12" and python_version < "3.14" \ + --hash=sha256:7553b606f963097cb190536dc30556cce85090692e471a422fff30ca29b04348 \ + --hash=sha256:84f82d826978bba416ab453460ff6a7391cdc3534c93a786595e4068680016b7 packaging==26.2 ; python_version >= "3.12" and python_version < "3.14" \ --hash=sha256:5fc45236b9446107ff2415ce77c807cee2862cb6fac22b8a73826d0693b0980e \ --hash=sha256:ff452ff5a3e828ce110190feff1178bb1f2ea2281fa2075aadb987c2fb221661 diff --git a/taskomatic/taskomatic.py b/taskomatic/taskomatic.py index cf10d7970..7ce9af6d1 100755 --- a/taskomatic/taskomatic.py +++ b/taskomatic/taskomatic.py @@ -73,6 +73,14 @@ def import_job(job_name): def main(): """Creates scheduler, fills it up with tasks and runs it""" init_logging() + + from common.telemetry import init_otel + from common.telemetry import instrument_outbound_http + from common.telemetry import instrument_psycopg2 + + init_otel(service_name="vulnerability-engine-taskomatic") + instrument_psycopg2() + instrument_outbound_http() initialize_unleash() LOGGER.info("Opening port [%s] for prometheus", PROMETHEUS_PORT) diff --git a/tests/common_tests/test_telemetry.py b/tests/common_tests/test_telemetry.py new file mode 100644 index 000000000..ee863db82 --- /dev/null +++ b/tests/common_tests/test_telemetry.py @@ -0,0 +1,432 @@ +# -*- coding: utf-8 -*- +# pylint: disable=no-self-use +""" +Tests for common.telemetry module. +""" + +import logging +import os +from unittest.mock import MagicMock +from unittest.mock import patch + + +class TestTelemetryConfig: + """Test telemetry configuration parsing.""" + + def test_otel_disabled_by_default(self): + """OTEL_ENABLED defaults to false.""" + with patch.dict(os.environ, {}, clear=False): + os.environ.pop("OTEL_ENABLED", None) + import importlib + + import common.telemetry as telemetry_mod + + importlib.reload(telemetry_mod) + assert telemetry_mod.OTEL_ENABLED is False + + def test_otel_enabled_when_set(self): + """OTEL_ENABLED=true enables tracing.""" + with patch.dict(os.environ, {"OTEL_ENABLED": "true"}): + import importlib + + import common.telemetry as telemetry_mod + + importlib.reload(telemetry_mod) + assert telemetry_mod.OTEL_ENABLED is True + + def test_sampling_rate_default(self): + """Default sampling rate is 1.0.""" + with patch.dict(os.environ, {}, clear=False): + os.environ.pop("OTEL_SAMPLING_RATE", None) + import importlib + + import common.telemetry as telemetry_mod + + importlib.reload(telemetry_mod) + assert telemetry_mod.OTEL_SAMPLING_RATE == 1.0 + + def test_sampling_rate_clamped(self): + """Sampling rate is clamped to [0.0, 1.0].""" + with patch.dict(os.environ, {"OTEL_SAMPLING_RATE": "2.5"}): + import importlib + + import common.telemetry as telemetry_mod + + importlib.reload(telemetry_mod) + assert telemetry_mod.OTEL_SAMPLING_RATE == 1.0 + + def test_sampling_rate_invalid_falls_back(self): + """Invalid sampling rate falls back to default.""" + with patch.dict(os.environ, {"OTEL_SAMPLING_RATE": "not_a_number"}): + import importlib + + import common.telemetry as telemetry_mod + + importlib.reload(telemetry_mod) + assert telemetry_mod.OTEL_SAMPLING_RATE == 1.0 + + +class TestInitOtel: + """Test init_otel initialization.""" + + def test_init_otel_disabled_noop(self): + """init_otel is a no-op when OTEL_ENABLED is false.""" + with patch.dict(os.environ, {"OTEL_ENABLED": "false"}): + import importlib + + import common.telemetry as telemetry_mod + + importlib.reload(telemetry_mod) + telemetry_mod._otel_initialized_pid = None + telemetry_mod.init_otel() + assert telemetry_mod._otel_initialized_pid == os.getpid() + + def test_init_otel_enabled(self): + """init_otel sets up TracerProvider when enabled.""" + with patch.dict(os.environ, {"OTEL_ENABLED": "true", "IMAGE_TAG": "v1.2.3"}): + import importlib + + import common.telemetry as telemetry_mod + + importlib.reload(telemetry_mod) + telemetry_mod._otel_initialized_pid = None + + with patch("opentelemetry.trace.set_tracer_provider") as mock_set_provider: + telemetry_mod.init_otel(service_name="test-service") + mock_set_provider.assert_called_once() + + assert telemetry_mod._otel_initialized_pid == os.getpid() + + def test_init_otel_idempotent(self): + """init_otel is safe to call multiple times (same PID).""" + with patch.dict(os.environ, {"OTEL_ENABLED": "true"}): + import importlib + + import common.telemetry as telemetry_mod + + importlib.reload(telemetry_mod) + telemetry_mod._otel_initialized_pid = None + + with patch("opentelemetry.trace.set_tracer_provider") as mock_set_provider: + telemetry_mod.init_otel(service_name="test-service") + telemetry_mod.init_otel(service_name="test-service") + assert mock_set_provider.call_count == 1 + + +class TestInstrumentToggle: + """Test that instrumentation functions respect toggle flags.""" + + def test_instrument_flask_disabled(self): + """instrument_flask_app is noop when OTEL_ENABLED=false.""" + import importlib + + import common.telemetry as telemetry_mod + + with patch.dict(os.environ, {"OTEL_ENABLED": "false"}): + importlib.reload(telemetry_mod) + mock_app = MagicMock() + telemetry_mod.instrument_flask_app(mock_app) + + def test_instrument_outbound_http_disabled(self): + """instrument_outbound_http is noop when OTEL_ENABLED=false.""" + import importlib + + import common.telemetry as telemetry_mod + + with patch.dict(os.environ, {"OTEL_ENABLED": "false"}): + importlib.reload(telemetry_mod) + telemetry_mod.instrument_outbound_http() + + def test_instrument_psycopg2_disabled(self): + """instrument_psycopg2 is noop when OTEL_ENABLED=false.""" + import importlib + + import common.telemetry as telemetry_mod + + with patch.dict(os.environ, {"OTEL_ENABLED": "false"}): + importlib.reload(telemetry_mod) + telemetry_mod.instrument_psycopg2() + + def test_outbound_http_toggle_disabled_when_otel_enabled(self): + """instrument_outbound_http skips when OTEL_HTTP_OUTBOUND_ENABLED=false.""" + import importlib + + import common.telemetry as telemetry_mod + + with patch.dict(os.environ, {"OTEL_ENABLED": "true", "OTEL_HTTP_OUTBOUND_ENABLED": "false"}): + importlib.reload(telemetry_mod) + with patch("opentelemetry.instrumentation.requests.RequestsInstrumentor.instrument") as mock_instrument: + telemetry_mod.instrument_outbound_http() + mock_instrument.assert_not_called() + + def test_sql_toggle_disabled_when_otel_enabled(self): + """instrument_psycopg2 skips when OTEL_SQL_ENABLED=false.""" + import importlib + + import common.telemetry as telemetry_mod + + with patch.dict(os.environ, {"OTEL_ENABLED": "true", "OTEL_SQL_ENABLED": "false"}): + importlib.reload(telemetry_mod) + with patch("opentelemetry.instrumentation.psycopg2.Psycopg2Instrumentor.instrument") as mock_instrument: + telemetry_mod.instrument_psycopg2() + mock_instrument.assert_not_called() + + def test_inbound_http_toggle_disabled_when_otel_enabled(self): + """instrument_flask_app skips when OTEL_HTTP_INBOUND_ENABLED=false.""" + import importlib + + import common.telemetry as telemetry_mod + + with patch.dict(os.environ, {"OTEL_ENABLED": "true", "OTEL_HTTP_INBOUND_ENABLED": "false"}): + importlib.reload(telemetry_mod) + mock_app = MagicMock() + with patch("opentelemetry.instrumentation.flask.FlaskInstrumentor.instrument_app") as mock_instrument: + telemetry_mod.instrument_flask_app(mock_app) + mock_instrument.assert_not_called() + + +class TestExtractContext: + """Test trace context extraction from Kafka headers.""" + + def test_extract_context_disabled(self): + """extract_context_from_msg_headers returns None when disabled.""" + import importlib + + import common.telemetry as telemetry_mod + + with patch.dict(os.environ, {"OTEL_ENABLED": "false"}): + importlib.reload(telemetry_mod) + headers = [("traceparent", b"00-abc123-def456-01")] + result = telemetry_mod.extract_context_from_msg_headers(headers) + assert result is None + + def test_extract_context_no_headers(self): + """extract_context_from_msg_headers returns None for empty headers.""" + import importlib + + import common.telemetry as telemetry_mod + + with patch.dict(os.environ, {"OTEL_ENABLED": "true"}): + importlib.reload(telemetry_mod) + result = telemetry_mod.extract_context_from_msg_headers(None) + assert result is None + + def test_extract_context_no_traceparent(self): + """extract_context_from_msg_headers returns None when no traceparent header.""" + import importlib + + import common.telemetry as telemetry_mod + + with patch.dict(os.environ, {"OTEL_ENABLED": "true"}): + importlib.reload(telemetry_mod) + headers = [("some-other-header", b"value")] + result = telemetry_mod.extract_context_from_msg_headers(headers) + assert result is None + + def test_extract_context_with_traceparent(self): + """extract_context_from_msg_headers returns context when traceparent exists.""" + import importlib + + import common.telemetry as telemetry_mod + + with patch.dict(os.environ, {"OTEL_ENABLED": "true"}): + importlib.reload(telemetry_mod) + traceparent = "00-0af7651916cd43dd8448eb211c80319c-b7ad6b7169203331-01" + headers = [("traceparent", traceparent.encode("utf-8"))] + result = telemetry_mod.extract_context_from_msg_headers(headers) + assert result is not None + + +class TestInjectContext: + """Test trace context injection into Kafka headers.""" + + def test_inject_context_disabled(self): + """inject_context_to_msg_headers returns empty list when disabled.""" + import importlib + + import common.telemetry as telemetry_mod + + with patch.dict(os.environ, {"OTEL_ENABLED": "false"}): + importlib.reload(telemetry_mod) + result = telemetry_mod.inject_context_to_msg_headers() + assert result == [] + + def test_inject_context_preserves_existing_headers(self): + """inject_context_to_msg_headers preserves existing headers.""" + import importlib + + import common.telemetry as telemetry_mod + + with patch.dict(os.environ, {"OTEL_ENABLED": "false"}): + importlib.reload(telemetry_mod) + existing = [("key", b"value")] + result = telemetry_mod.inject_context_to_msg_headers(existing) + assert result == existing + + def test_inject_context_enabled_adds_traceparent(self): + """inject_context_to_msg_headers injects traceparent when OTEL enabled and span active.""" + import importlib + + from opentelemetry.sdk.trace import TracerProvider + + import common.telemetry as telemetry_mod + + with patch.dict(os.environ, {"OTEL_ENABLED": "true"}): + importlib.reload(telemetry_mod) + + provider = TracerProvider() + tracer = provider.get_tracer("test") + with tracer.start_as_current_span("test-span"): + result = telemetry_mod.inject_context_to_msg_headers() + header_keys = [k for k, v in result] + assert "traceparent" in header_keys + traceparent_value = next(v for k, v in result if k == "traceparent") + assert isinstance(traceparent_value, bytes) + + def test_inject_context_enabled_preserves_existing_headers(self): + """inject_context_to_msg_headers preserves existing headers when adding trace context.""" + import importlib + + from opentelemetry.sdk.trace import TracerProvider + + import common.telemetry as telemetry_mod + + with patch.dict(os.environ, {"OTEL_ENABLED": "true"}): + importlib.reload(telemetry_mod) + + provider = TracerProvider() + tracer = provider.get_tracer("test") + with tracer.start_as_current_span("test-span"): + existing = [("custom-header", b"custom-value")] + result = telemetry_mod.inject_context_to_msg_headers(existing) + assert ("custom-header", b"custom-value") in result + header_keys = [k for k, v in result] + assert "traceparent" in header_keys + + def test_inject_extract_roundtrip(self): + """Injected context can be extracted back (round-trip propagation).""" + import importlib + + from opentelemetry.sdk.trace import TracerProvider + + import common.telemetry as telemetry_mod + + with patch.dict(os.environ, {"OTEL_ENABLED": "true"}): + importlib.reload(telemetry_mod) + + provider = TracerProvider() + tracer = provider.get_tracer("test") + with tracer.start_as_current_span("test-span"): + headers = telemetry_mod.inject_context_to_msg_headers() + ctx = telemetry_mod.extract_context_from_msg_headers(headers) + assert ctx is not None + + +class TestRHAttributeSpanProcessor: + """Test the RHAttributeSpanProcessor.""" + + def test_processor_sets_rh_service(self): + """Processor sets rh.service on every span.""" + import importlib + + import common.telemetry as telemetry_mod + + with patch.dict(os.environ, {"OTEL_ENABLED": "true"}): + importlib.reload(telemetry_mod) + processor = telemetry_mod._build_rh_attribute_span_processor(rh_service="test-service") + + mock_span = MagicMock() + mock_span.is_recording.return_value = True + mock_span.attributes = {} + + processor.on_start(mock_span, parent_context=None) + + mock_span.set_attribute.assert_any_call("rh.service", "test-service") + + def test_processor_noop_for_non_recording_span(self): + """Processor does nothing for non-recording spans.""" + import importlib + + import common.telemetry as telemetry_mod + + with patch.dict(os.environ, {"OTEL_ENABLED": "true"}): + importlib.reload(telemetry_mod) + processor = telemetry_mod._build_rh_attribute_span_processor() + + mock_span = MagicMock() + mock_span.is_recording.return_value = False + + processor.on_start(mock_span, parent_context=None) + + mock_span.set_attribute.assert_not_called() + + +class TestContextualFilter: + """Test the logging ContextualFilter for OTel trace correlation.""" + + def test_filter_no_active_span(self): + """Filter adds None trace_id/span_id when no active span.""" + from common.logging import ContextualFilter + + f = ContextualFilter() + record = logging.LogRecord("test", logging.INFO, "", 0, "msg", (), None) + result = f.filter(record) + assert result is True + assert record.trace_id is None + assert record.span_id is None + + def test_filter_with_active_span(self): + """Filter adds hex-encoded trace_id/span_id from active span.""" + from unittest.mock import patch as _patch + + from common.logging import ContextualFilter + + mock_span_context = MagicMock() + mock_span_context.is_valid = True + mock_span_context.trace_id = 0x0AF7651916CD43DD8448EB211C80319C + mock_span_context.span_id = 0xB7AD6B7169203331 + + mock_span = MagicMock() + mock_span.get_span_context.return_value = mock_span_context + + f = ContextualFilter() + record = logging.LogRecord("test", logging.INFO, "", 0, "msg", (), None) + + with _patch("opentelemetry.trace.get_current_span", return_value=mock_span): + result = f.filter(record) + + assert result is True + assert record.trace_id == "0af7651916cd43dd8448eb211c80319c" + assert record.span_id == "b7ad6b7169203331" + + +class TestGetTracer: + """Test get_tracer helper.""" + + def test_get_tracer_returns_tracer(self): + """get_tracer returns a Tracer object.""" + from common.telemetry import get_tracer + + tracer = get_tracer("test-module") + assert tracer is not None + + +class TestUseOtelContext: + """Test use_otel_context context manager.""" + + def test_use_otel_context_none(self): + """use_otel_context with None is a noop.""" + from common.telemetry import use_otel_context + + with use_otel_context(None): + pass + + def test_use_otel_context_valid(self): + """use_otel_context attaches and detaches context.""" + from opentelemetry import context as otel_context + + from common.telemetry import use_otel_context + + ctx = otel_context.get_current() + with use_otel_context(ctx): + pass diff --git a/vmaas_sync/vmaas_sync.py b/vmaas_sync/vmaas_sync.py index 026631f98..6556fa79f 100644 --- a/vmaas_sync/vmaas_sync.py +++ b/vmaas_sync/vmaas_sync.py @@ -494,6 +494,15 @@ def sync_os_md(): def main(): """Main VMaaS sync entrypoint.""" init_logging() + + from common.telemetry import init_otel + from common.telemetry import instrument_outbound_http + from common.telemetry import instrument_psycopg2 + + init_otel(service_name="vulnerability-engine-vmaas-sync") + instrument_psycopg2() + instrument_outbound_http() + ensure_minimal_schema_version() LOGGER.info("Starting VMaaS sync.") with DatabasePool(1):