feat(RHINENG-28313): add OpenTelemetry distributed tracing - #2454
Open
rodrigonull wants to merge 2 commits into
Open
feat(RHINENG-28313): add OpenTelemetry distributed tracing#2454rodrigonull wants to merge 2 commits into
rodrigonull wants to merge 2 commits into
Conversation
Reviewer's GuideAdds centralized OpenTelemetry tracing to Vulnerability Engine services, wiring in Kafka context propagation, HTTP/DB instrumentation, log correlation, and ClowdApp OTEL env configuration, plus tests and dependencies. Sequence diagram for OpenTelemetry trace propagation across Kafka and internal queuesequenceDiagram
actor Ingress
participant Manager
participant Listener
participant Kafka
participant Grouper
participant Queue
participant Evaluator
participant OTLPExporter
Ingress->>Manager: HTTP request
Manager->>Manager: init_otel
Manager->>Listener: MQWriter.send_one
Listener->>Kafka: MQWriter.send_one
Kafka-->>Grouper: ConsumerRecord
Grouper->>Grouper: extract_context_from_msg_headers
Grouper->>Grouper: use_otel_context
Grouper->>Queue: push_inventory_msg
Queue->>Queue: QueueItem
Queue->>Evaluator: MQWriter.send_one
Kafka-->>Evaluator: ConsumerRecord
Evaluator->>Evaluator: extract_context_from_msg_headers
Evaluator->>Evaluator: use_otel_context
Evaluator->>Evaluator: EvaluatorProcessor.evaluate_system
Evaluator->>OTLPExporter: OTLPSpanExporter.export
File-Level Changes
Tips and commandsInteracting with Sourcery
Customizing Your ExperienceAccess your dashboard to:
Getting Help
|
![sourcery-ai[bot]](https://avatars.githubusercontent.com/in/48477?s=60&v=4){kind=small}
There was a problem hiding this comment.
Hey - I've found 8 issues, and left some high level feedback:
- The OTEL_* environment variable blocks are duplicated for every deployment in clowdapp.yaml; consider factoring these into a shared template or helper to reduce repetition and the risk of config drift between services.
- There is an inconsistency in Kafka span attributes between send_one and send_many (messaging.operation.name vs messaging.operation.type); aligning on a single key that matches the chosen semantic convention will make traces easier to query and analyze.
- The ContextualFilter is only attached to the stream handler in init_logging; if you want trace_id/span_id available in CloudWatch logs as well, add the filter to the watchtower/CloudWatch handlers created in setup_cw_logging.
Prompt for AI Agents
Please address the comments from this code review:
## Overall Comments
- The OTEL_* environment variable blocks are duplicated for every deployment in clowdapp.yaml; consider factoring these into a shared template or helper to reduce repetition and the risk of config drift between services.
- There is an inconsistency in Kafka span attributes between send_one and send_many (messaging.operation.name vs messaging.operation.type); aligning on a single key that matches the chosen semantic convention will make traces easier to query and analyze.
- The ContextualFilter is only attached to the stream handler in init_logging; if you want trace_id/span_id available in CloudWatch logs as well, add the filter to the watchtower/CloudWatch handlers created in setup_cw_logging.
## Individual Comments
### Comment 1
<location path="grouper/queue.py" line_range="57-66" />
<code_context>
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=None) -> None:
"""Push inventory upload message to queue"""
is_updated = False
</code_context>
<issue_to_address>
**suggestion:** The otel_context field is loosely typed and may benefit from a more specific type.
`QueueItem.otel_context` is declared as `object = None`, and the `push_*` APIs accept `otel_context=None` without a type hint. Since this is intended to hold an OpenTelemetry context, consider annotating it as `context.Context | None` (or the appropriate OTel type) and updating the `push_*` signatures accordingly to improve clarity and static analysis.
Suggested implementation:
```python
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
from opentelemetry import context
```
```python
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,
otel_context: context.Context | None = None,
) -> None:
```
To fully implement the suggestion across the codebase:
1. Update all other `push_*` methods in `grouper/queue.py` that accept `otel_context` to use the same `otel_context: context.Context | None = None` annotation.
2. In the definition of `QueueItem` (likely in this file or a related module), change the `otel_context` field’s type from `object = None` to `context.Context | None = None` (or an equivalent `typing`-based annotation if using `dataclasses` or `attrs`).
3. Ensure any other call sites passing `otel_context` are compatible with `context.Context` (e.g., if they currently pass a different type, align them with OpenTelemetry’s context object).
</issue_to_address>
### Comment 2
<location path="tests/common_tests/test_telemetry.py" line_range="189-198" />
<code_context>
+class TestInjectContext:
</code_context>
<issue_to_address>
**suggestion (testing):** Add positive-path tests for context injection when OTEL is enabled
Currently `TestInjectContext` only exercises the OTEL-disabled path and header preservation in that mode. There’s no coverage of actual header injection when `OTEL_ENABLED=true`.
Please add tests that:
- Run with `OTEL_ENABLED=true`, call `inject_context_to_msg_headers()` with no headers, and assert a `("traceparent", <bytes>)` header is injected.
- Call `inject_context_to_msg_headers()` with existing headers and assert the original headers are retained and trace headers are appended.
- Optionally, add a round-trip test using `extract_context_from_msg_headers()` (inject then extract and assert a non-`None` context) to validate propagation end-to-end.
</issue_to_address>
### Comment 3
<location path="tests/common_tests/test_telemetry.py" line_range="109-118" />
<code_context>
+class TestInstrumentToggle:
</code_context>
<issue_to_address>
**suggestion (testing):** Extend instrumentation toggle tests to cover feature-specific flags when OTEL is enabled
Beyond the global `OTEL_ENABLED` flag, we also rely on feature-specific toggles (`OTEL_HTTP_INBOUND_ENABLED`, `OTEL_HTTP_OUTBOUND_ENABLED`, `OTEL_SQL_ENABLED`, etc.) that currently aren’t tested.
Please add tests that cover cases like:
- `OTEL_ENABLED=true` with `OTEL_HTTP_OUTBOUND_ENABLED=false`: assert `instrument_outbound_http` does not call `RequestsInstrumentor().instrument` (via patch), while other OTel behavior remains active.
- `OTEL_ENABLED=true` with `OTEL_SQL_ENABLED=false`: assert `instrument_psycopg2` / `instrument_psycopg` return early and don’t call their instrumentors.
- `OTEL_ENABLED=true` with `OTEL_HTTP_INBOUND_ENABLED=false`: assert `instrument_flask_app` skips inbound HTTP instrumentation.
These tests can rely on mocking the instrumentors to keep them fast and focused on environment-driven toggle behavior.
Suggested implementation:
```python
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()
# When OTEL_ENABLED is false, inbound HTTP instrumentation should be skipped
with patch("opentelemetry.instrumentation.flask.FlaskInstrumentor.instrument_app") as mock_instrument_app:
telemetry_mod.instrument_flask_app(mock_app)
mock_instrument_app.assert_not_called()
def test_outbound_http_disabled_when_otel_enabled(self):
"""
OTEL_ENABLED=true with OTEL_HTTP_OUTBOUND_ENABLED=false:
instrument_outbound_http should not call RequestsInstrumentor().instrument.
"""
import importlib
import common.telemetry as telemetry_mod
env = {
"OTEL_ENABLED": "true",
"OTEL_HTTP_OUTBOUND_ENABLED": "false",
}
with patch.dict(os.environ, env, clear=True):
importlib.reload(telemetry_mod)
with patch(
"opentelemetry.instrumentation.requests.RequestsInstrumentor.instrument"
) as mock_requests_instrument:
telemetry_mod.instrument_outbound_http()
mock_requests_instrument.assert_not_called()
def test_sql_psycopg2_disabled_when_otel_enabled(self):
"""
OTEL_ENABLED=true with OTEL_SQL_ENABLED=false:
instrument_psycopg2 should return early and not call its instrumentor.
"""
import importlib
import common.telemetry as telemetry_mod
env = {
"OTEL_ENABLED": "true",
"OTEL_SQL_ENABLED": "false",
}
with patch.dict(os.environ, env, clear=True):
importlib.reload(telemetry_mod)
with patch(
"opentelemetry.instrumentation.psycopg2.Psycopg2Instrumentor.instrument"
) as mock_psycopg2_instrument:
telemetry_mod.instrument_psycopg2()
mock_psycopg2_instrument.assert_not_called()
def test_sql_psycopg_disabled_when_otel_enabled(self):
"""
OTEL_ENABLED=true with OTEL_SQL_ENABLED=false:
instrument_psycopg should return early and not call its instrumentor.
"""
import importlib
import common.telemetry as telemetry_mod
env = {
"OTEL_ENABLED": "true",
"OTEL_SQL_ENABLED": "false",
}
with patch.dict(os.environ, env, clear=True):
importlib.reload(telemetry_mod)
with patch(
"opentelemetry.instrumentation.psycopg.PsycopgInstrumentor.instrument"
) as mock_psycopg_instrument:
telemetry_mod.instrument_psycopg()
mock_psycopg_instrument.assert_not_called()
def test_inbound_http_disabled_when_otel_enabled(self):
"""
OTEL_ENABLED=true with OTEL_HTTP_INBOUND_ENABLED=false:
instrument_flask_app should skip inbound HTTP instrumentation.
"""
import importlib
import common.telemetry as telemetry_mod
env = {
"OTEL_ENABLED": "true",
"OTEL_HTTP_INBOUND_ENABLED": "false",
}
with patch.dict(os.environ, env, clear=True):
importlib.reload(telemetry_mod)
mock_app = MagicMock()
with patch(
"opentelemetry.instrumentation.flask.FlaskInstrumentor.instrument_app"
) as mock_instrument_app:
telemetry_mod.instrument_flask_app(mock_app)
mock_instrument_app.assert_not_called()
```
The exact patch targets for the instrumentors may need to be aligned with your telemetry implementation:
1. If `common.telemetry` wraps or re-exports the instrumentors, you may want to patch
`common.telemetry.RequestsInstrumentor.instrument`,
`common.telemetry.Psycopg2Instrumentor.instrument`,
`common.telemetry.PsycopgInstrumentor.instrument`,
and `common.telemetry.FlaskInstrumentor.instrument_app` instead of the `opentelemetry.*` paths used above.
2. Ensure that `instrument_outbound_http`, `instrument_psycopg2`, `instrument_psycopg`, and `instrument_flask_app`
exist in `common.telemetry` and that they read the corresponding environment variables
(`OTEL_HTTP_OUTBOUND_ENABLED`, `OTEL_SQL_ENABLED`, `OTEL_HTTP_INBOUND_ENABLED`) to short-circuit as tested.
3. If your inbound HTTP instrumentation uses a different method (e.g. `instrument_app` vs a custom wrapper),
adjust the patched attribute accordingly while keeping the assertions the same.
</issue_to_address>
### Comment 4
<location path="listener/listener.py" line_range="222" />
<code_context>
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(
</code_context>
<issue_to_address>
**issue (complexity):** Consider extracting helper functions for Kafka consumer span creation and context population to centralize telemetry logic and simplify `_consume_message` control flow.
The duplicated tracing blocks and inline `threadctx` wiring in `_consume_message` can be simplified by extracting helpers, keeping telemetry intact while reducing nesting and repetition.
You can:
1. Extract a helper to run a handler inside the Kafka consumer span.
2. Extract a helper to set `org_id` / `request_id` and update `threadctx`.
Example:
```python
def _run_in_kafka_consumer_span(self, msg: ConsumerRecord, msg_dict: dict, handler: callable, set_ctx_attrs: bool = False):
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,
},
) as span:
if set_ctx_attrs:
self._populate_span_and_threadctx(span, msg_dict)
create_task_and_log(handler(msg_dict), LOGGER, self.loop)
def _populate_span_and_threadctx(self, span, msg_dict: dict):
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
```
Then `_consume_message` becomes flatter and easier to read:
```python
def _consume_message(self, msg: ConsumerRecord):
try:
msg_dict = json.loads(msg.value.decode("utf-8"))
except json.decoder.JSONDecodeError as exc:
MESSAGE_PARSE_ERROR.inc()
LOGGER.exception("Unable to parse message: %s", exc)
return
if msg_dict.get("host") or msg_dict.get("type") == "delete":
self._run_in_kafka_consumer_span(msg, msg_dict, self.consume_inventory_msg, set_ctx_attrs=True)
PROCESS_MESSAGES.inc()
elif msg_dict.get("input"):
self._run_in_kafka_consumer_span(msg, msg_dict, self.consume_advisor_msg)
PROCESS_MESSAGES.inc()
else:
LOGGER.exception("Unknown message obtained: %s", msg)
SKIPPED_MESSAGES.inc()
```
This keeps all the new telemetry behavior, but centralizes span creation and context propagation, reduces duplication, and restores a clearer “what do we do with this message?” control flow.
</issue_to_address>
### Comment 5
<location path="grouper/grouper.py" line_range="86" />
<code_context>
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(
</code_context>
<issue_to_address>
**issue (complexity):** Consider extracting the OpenTelemetry setup and queue routing logic in consume_message into a small helper and a simple dispatcher to decouple telemetry from business logic and reduce nesting.
You can simplify `consume_message` and reduce coupling to telemetry by extracting a small helper to encapsulate the OpenTelemetry boilerplate and by isolating the queue call decision.
One possible refactor:
```python
# keep this as a private helper on Grouper (or a shared module if listener/evaluator need it)
def _with_kafka_consumer_span(
self,
msg: ConsumerRecord,
msg_dict: dict,
msg_type: GrouperMessageType,
inventory_id: str,
fn: Callable[[Context], None],
) -> None:
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,
},
):
fn(parent_ctx)
```
Then `consume_message` becomes a simple routing function:
```python
# inside consume_message, after msg_type is resolved
def _enqueue(parent_ctx):
request_id = (msg_dict.get("platform_metadata") or {}).get("request_id", "")
if msg_type is GrouperMessageType.INVENTORY_UPLOAD:
self.queue.push_inventory_msg(
org_id, inventory_id, reporter, changed, request_id,
otel_context=parent_ctx,
)
elif msg_type is GrouperMessageType.ADVISOR_UPLOAD:
self.queue.push_advisor_msg(
org_id, inventory_id, reporter, changed, request_id,
otel_context=parent_ctx,
)
self._with_kafka_consumer_span(msg, msg_dict, msg_type, inventory_id, _enqueue)
```
Benefits while preserving behavior:
- Core decision path (`if msg_type is ...`) is no longer nested inside two context managers.
- Telemetry setup (`extract_context_from_msg_headers`, `use_otel_context`, span attributes) lives in one focused helper.
- The helper can be reused in `listener` / `evaluator`, removing duplicated span creation logic and keeping “Kafka consumer span” semantics in one place.
If possible in your codebase, you can take this one step further by moving `otel_context` handling entirely into queue-layer helpers, so your routing call sites never see `otel_context`:
```python
# in GrouperQueue (example)
def push_inventory_msg_with_otel(self, parent_ctx, *args, **kwargs):
with use_otel_context(parent_ctx):
return self.push_inventory_msg(*args, **kwargs, otel_context=parent_ctx)
```
Then your `_enqueue` just calls `push_inventory_msg_with_otel` instead of threading `otel_context` explicitly, keeping telemetry concerns localized.
</issue_to_address>
### Comment 6
<location path="evaluator/evaluator.py" line_range="91" />
<code_context>
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(
</code_context>
<issue_to_address>
**issue (complexity):** Consider extracting the tracing and thread context setup into a reusable helper/context manager so `_consume_message` remains focused on its core message-processing logic.
You can encapsulate the tracing + threadctx wiring into a reusable helper/context manager to keep `_consume_message` focused on core behavior and reduce indentation.
For example, in `common.telemetry` (or a local helper module), add something like:
```python
# common/telemetry.py
from opentelemetry import trace
from contextlib import contextmanager
TRACER = get_tracer(__name__)
@contextmanager
def kafka_consumer_span(msg, inventory_id, msg_type, org_id=None, request_id=None):
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
yield
```
Then `_consume_message` can stay visually simple:
```python
async def _consume_message(self, msg: ConsumerRecord):
msg_dict = json.loads(msg.value)
inventory_id = msg_dict.get("inventory_id")
org_id = (msg_dict.get("platform_metadata") or {}).get("org_id")
request_id = (msg_dict.get("platform_metadata") or {}).get("request_id")
timestamp = parser.parse(ts) if (ts := msg_dict.get("timestamp")) else None
recalc_event_id = msg_dict.get("recalc_event_id")
try:
msg_type = EvaluatorMessageType(msg_dict.get("type"))
except ValueError:
LOGGER.error("received unknown message type: %s", msg_type)
return
with kafka_consumer_span(msg, inventory_id, msg_type, org_id, request_id):
await self.processor.evaluate_system(
inventory_id,
org_id,
request_id,
timestamp,
recalc_event_id=recalc_event_id,
)
```
This keeps functionality/tracing intact, removes nested blocks from the core logic, and gives you a shared pattern that other consumers can reuse without duplicating span setup and `threadctx` handling.
</issue_to_address>
### Comment 7
<location path="evaluator/processor.py" line_range="414" />
<code_context>
- res = await self.client.send_and_wait(self.topic, value=data, key=self._serialize_key(key), headers=headers)
- LOGGER.debug(res)
+ headers = inject_context_to_msg_headers(headers)
+ with TRACER.start_as_current_span(
+ f"{self.topic} send",
+ kind=SpanKind.PRODUCER,
</code_context>
<issue_to_address>
**issue (complexity):** Consider combining the context managers and using a helper for span error tagging to keep telemetry while simplifying control flow and reducing duplication.
You can keep all telemetry while reducing nesting and centralizing error tagging with small, local changes.
**1. Flatten nesting by combining context managers**
You don’t need an inner `with` for `EVAL_TIME`:
```python
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))
_set_span_error(span, 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))
_set_span_error(span, "vmaas_error")
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,
)
```
**2. Centralize span error tagging**
Extract the repeated attribute setting into a tiny helper to reduce duplication and keep conventions in one place:
```python
def _set_span_error(span, error_value: str) -> None:
span.set_attribute("error", True)
span.set_attribute("vulnerability.error", error_value)
```
You can keep this as a private function in the same module or as a `@staticmethod` on the class, and the rest of `evaluate_system` stays close to its original structure while retaining all telemetry behavior.
</issue_to_address>
### Comment 8
<location path="common/telemetry.py" line_range="190" />
<code_context>
+ return result
+
+
+def init_otel(
+ service_name: str = "vulnerability-engine",
+ service_version: str | None = None,
</code_context>
<issue_to_address>
**issue (complexity):** Consider decomposing the OpenTelemetry setup and instrumentation helpers so init_otel and span processors are smaller, single-purpose, and easier to scan and test.
You can keep all current behavior but reduce complexity a bit by decomposing some pieces.
### 1. Split `init_otel` into small helpers
Right now `init_otel` is parsing config, building the provider, span limits, exporter, and logging. You can move those into internal helpers so `init_otel` becomes easier to read and safer to modify.
Example (keep names private so API stays the same):
```python
def _build_resource(service_name: str, service_version: str | None) -> Resource:
from opentelemetry.sdk.resources import SERVICE_NAME, SERVICE_VERSION, Resource
if service_version is None:
service_version = os.getenv("IMAGE_TAG", "unknown")
return Resource.create(
attributes={
SERVICE_NAME: service_name,
SERVICE_VERSION: service_version,
"deployment.environment": os.getenv("NAMESPACE", "development"),
}
)
def _build_tracer_provider(resource: Resource, sampling_rate: float) -> TracerProvider:
from opentelemetry.sdk.trace import SpanLimits, TracerProvider
from opentelemetry.sdk.trace.sampling import ParentBased, TraceIdRatioBased
sampler = ParentBased(TraceIdRatioBased(sampling_rate))
span_limits = SpanLimits(
max_attributes=OTEL_SPAN_ATTRIBUTE_COUNT_LIMIT,
max_attribute_length=OTEL_SPAN_ATTRIBUTE_VALUE_LENGTH_LIMIT,
)
return TracerProvider(resource=resource, sampler=sampler, span_limits=span_limits)
def _configure_exporter(provider: TracerProvider) -> None:
from opentelemetry.exporter.otlp.proto.http import Compression
from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter
from opentelemetry.sdk.trace.export import BatchSpanProcessor
_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,
)
)
```
Then `init_otel` becomes:
```python
def init_otel(...):
...
resource = _build_resource(service_name, service_version)
effective_rate = sampling_rate or OTEL_SAMPLING_RATE
provider = _build_tracer_provider(resource, effective_rate)
provider.add_span_processor(_build_rh_attribute_span_processor(rh_service=rh_service))
_configure_exporter(provider)
trace.set_tracer_provider(provider)
...
```
This keeps behavior identical but isolates config parsing / object-wiring into testable pieces.
### 2. Separate `rh.service` injection from org/request propagation
`_build_rh_attribute_span_processor` currently does two unrelated things. You can keep the org/request propagation logic where it is but make `rh.service` a simple processor that is clearly single-purpose and easier to test.
Example:
```python
def _build_rh_service_span_processor(rh_service: str):
from opentelemetry.sdk.trace import SpanProcessor
class RHServiceSpanProcessor(SpanProcessor):
def on_start(self, span, parent_context=None): # noqa: ARG002
if span and span.is_recording():
span.set_attribute("rh.service", rh_service)
return RHServiceSpanProcessor()
```
Then scope the propagation processor to just that concern:
```python
def _build_rh_propagation_span_processor():
from opentelemetry import trace
from opentelemetry.sdk.trace import ReadableSpan, SpanProcessor
class RHPropagationSpanProcessor(SpanProcessor):
def on_start(self, span, parent_context=None):
if not span or not span.is_recording():
return
parent = (
trace.get_current_span(parent_context)
if parent_context is not None
else trace.get_current_span()
)
parent_attrs = parent.attributes if isinstance(parent, ReadableSpan) and parent.attributes else {}
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 RHPropagationSpanProcessor()
```
And in `init_otel`:
```python
provider.add_span_processor(_build_rh_service_span_processor(rh_service=rh_service))
provider.add_span_processor(_build_rh_propagation_span_processor())
```
This makes each processor’s responsibility obvious and reduces the branching inside `on_start` without changing semantics.
### 3. Centralize toggle checks for instrumentation
The repeated guards (`if not OTEL_ENABLED or not OTEL_HTTP_OUTBOUND_ENABLED`) can be factored into a tiny helper so each `instrument_*` function is shorter and more uniform.
Example:
```python
def _is_enabled(*flags: bool) -> bool:
return OTEL_ENABLED and all(flags)
```
Then:
```python
def instrument_flask_app(flask_app):
if not _is_enabled(OTEL_HTTP_INBOUND_ENABLED):
return
...
def instrument_sqlalchemy(engine):
if not _is_enabled(OTEL_SQL_ENABLED):
return
...
def instrument_outbound_http():
if not _is_enabled(OTEL_HTTP_OUTBOUND_ENABLED):
return
...
```
Optionally you can reuse this in `extract_context_from_msg_headers` / `inject_context_to_msg_headers` as well. This doesn’t remove any toggles but makes the control flow easier to scan and reduces the chance of inconsistent checks in future additions.
</issue_to_address>Help me be more useful! Please click 👍 or 👎 on each comment and I'll use the feedback to improve your reviews.
rodrigonull
force-pushed
the
RHINENG-28313
branch
from
August 10, 2026 11:37
4da2de7 to
cae2deb
Compare
Implement end-to-end distributed tracing across all vulnerability-engine components following patterns established in insights-host-inventory. - Add common/telemetry.py with centralized OTel initialization, sampling, instrumentation toggles, RHAttributeSpanProcessor, and log correlation - Instrument Kafka consumers (listener, evaluator, grouper) with trace context extraction/injection and semantic convention attributes - Instrument Flask/Connexion API (manager) with excluded health endpoints - Instrument outbound HTTP, psycopg2, and SQLAlchemy queries - Propagate traceparent through grouper queue to evaluator - Add Gunicorn post_fork hook for worker-safe OTel re-initialization - Inject trace context into all outgoing Kafka messages via MQWriter - Add OTEL_* environment variables to ClowdApp deployment manifest - Update poetry.lock and requirements files with OTel dependencies Assisted-by: Cursor:claude-4.6-opus
… contextvars Move inject_context_to_msg_headers() inside the TRACER.start_as_current_span() block so injected headers carry the producer span's trace context instead of the previous span's. Also add a missing producer span to send_raw. Replace threading.local() with contextvars.ContextVar for org_id/request_id propagation, which is safe under asyncio concurrency. Assisted-by: Cursor:claude-4.6-opus
rodrigonull
force-pushed
the
RHINENG-28313
branch
from
August 17, 2026 13:54
cae2deb to
2d7ddbb
Compare
MichaelMraka
approved these changes
Aug 18, 2026
Member
Author
|
@jdobes could you please take a look at this PR? Thanks! |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
2 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Jira
RHINENG-28313
What
Add OpenTelemetry distributed tracing to all Vulnerability Engine components, enabling end-to-end trace propagation from upstream services (Ingress, HBI) through the Listener, Grouper, Evaluator, Manager API, and background workers.
Why
Vulnerability Engine currently lacks observability into request latency and cross-service trace correlation. This makes it difficult to debug slow evaluations, identify bottlenecks in the host ingestion pipeline, and correlate issues across the platform. With OTel tracing, we gain full visibility into the request lifecycle from inventory event to CVE evaluation completion.
How
common/telemetry.pymodule: Centralized OTel initialization following the same patterns asinsights-host-inventory. All knobs are environment-driven (OTEL_ENABLED,OTEL_SAMPLING_RATE,OTEL_SQL_ENABLED, etc.) so stage/prod can tune independently.RHAttributeSpanProcessor: Custom SpanProcessor that automatically injectsrh.service,rh.org_id, andrh.request_idattributes into all spans via contextvars.MQWriterinjectstraceparentheaders on all outgoing messages. Consumers (Listener, Evaluator) extract parent context and create child spans with properSpanKind.CONSUMERand semantic convention attributes (messaging.system,messaging.operation.name,messaging.destination.name).QueueItemand reactivated when the grouped message is sent to the Evaluator, preserving the trace chain across the internal queue./healthzand/metricsexcluded. Outbound HTTP (VMaaS, etc.) and aiohttp-client instrumented automatically.ContextualFilterinjectstrace_idandspan_idinto all log records.post_forkhook re-initializes OTel in each worker process.OTEL_*environment variables added to every service deployment with template parameters for per-environment control.opentelemetry-api,opentelemetry-sdk,opentelemetry-exporter-otlp-proto-http, and instrumentation packages for Flask, requests, aiohttp-client, psycopg2, psycopg, and SQLAlchemy (all pinned to 1.44.0 / 0.65b0).Testing
tests/common_tests/test_telemetry.pycovering:RHAttributeSpanProcessorattribute injectionOTEL_ENABLED=false).Assisted-by: Cursor:claude-4.6-opus
Secure Coding Practices Checklist GitHub Link
Secure Coding Checklist
Summary by Sourcery
Introduce centralized OpenTelemetry-based distributed tracing across Vulnerability Engine services, including Kafka message flows, HTTP/API handling, database access, and background workers, controlled entirely via environment configuration.
New Features:
Enhancements:
Build:
Deployment:
Tests: