fix(RHINENG-31023): keep consumer spans active

[rodrigonull]

Jira

RHINENG-31023

What

Move OpenTelemetry Kafka consumer spans in the vulnerability listener and grouper into the awaited processing coroutines so spans remain active until message processing and outgoing operations complete.

Why

In both the listener and grouper, the consumer spans (process <topic>) were previously opened around fire-and-forget task scheduling and enqueue calls, causing spans to end prematurely in ~50–300µs. Descendant operations (such as psycopg3 DB transactions and aiokafka sends) ran in background tasks under an already-ended parent span, violating the OpenTelemetry span lifecycle model. This distorted Tempo's service-graph metrics (server_seconds reported ≤5ms for ~99.98% of requests instead of their true 10–20ms duration).

How

• Listener (listener/listener.py): Removed span creation from _consume_message at scheduling time. Started process <topic> CONSUMER spans inside consume_inventory_msg and consume_advisor_msg wrapping the awaited _consume_*_msg coroutines while preserving rh.org_id, rh.request_id, and threadctx context variables.
• Grouper (grouper/grouper.py, grouper/queue.py, grouper/common.py): Removed the span wrapping only enqueue in consume_message. Wrapped _start_item_processing in process <topic> using the OTel context stored on QueueItem, covering pair waiting/timeout, semaphore release, and evaluator sends.
• Kafka Sends (listener/*_processor.py, grouper/queue.py, common/utils.py): Returned send futures from send_msg_to_payload_tracker and awaited outgoing sends so producer child spans complete before the parent consumer span closes.

Testing

• Unit Tests: Added tests/listener_tests/test_listener_tracing.py and tests/grouper_tests/test_grouper_tracing.py asserting that consumer spans remain active during processing, do not close at scheduling time, and outlive all child spans (parent.end_time >= child.end_time).
• End-to-End Verification: Ran the full microservices stack locally with Tempo and Grafana. Uploaded an archive via Ingress and verified the trace tree end-to-end (Ingress -> Puptoo -> HBI -> vulnerability-engine-listener -> vulnerability-engine-grouper -> vulnerability-engine-evaluator):
  • Listener inventory span duration: 12.87ms (previously ~0.05ms) correctly parenting all DB operations and Kafka sends.
  • Listener advisor span duration: 7.64ms.
  • Grouper span duration: 579.59ms when paired, outliving all send children.
• Linters: pre-commit passed (black, flake8, isort, whitespace).

Secure Coding Practices Checklist GitHub Link

• https://github.com/RedHatInsights/secure-coding-checklist

Secure Coding Checklist

• Input Validation
• Output Encoding
• Authentication and Password Management
• Session Management
• Access Control
• Cryptographic Practices
• Error Handling and Logging
• Data Protection
• Communication Security
• System Configuration
• Database Security
• File Management
• Memory Management
• General Coding Practices

Summary by Sourcery

Keep consumer tracing spans open until message processing and downstream sends complete.

Bug Fixes:

• Keep Kafka consumer spans active through complete listener and grouper message processing so descendant database and messaging operations retain correct parent spans.

Enhancements:

• Await outgoing Kafka sends before closing consumer spans and preserve tracing context across deferred grouper processing.

Tests:

• Add listener and grouper tracing tests verifying active consumer spans, context propagation, child relationships, and span lifetimes.

[sourcery-ai]

Reviewer's Guide

The PR fixes prematurely terminated Kafka consumer traces by creating listener and grouper consumer spans around the actual awaited processing work, carrying extracted context through queued items, and awaiting outbound Kafka futures so producer and database-related child operations remain correctly nested until completion. New tracing tests validate span lifetimes, hierarchy, attributes, and context propagation.

Sequence diagram for listener consumer span lifetime

sequenceDiagram
    participant Kafka
    participant Listener
    participant Processor
    participant Grouper
    participant PayloadTracker

    Kafka->>Listener: _consume_message
    Listener->>Listener: consume_inventory_msg
    activate Listener
    Listener->>Processor: _consume_inventory_msg
    activate Processor
    Processor->>Grouper: _send_for_evaluation
    Processor->>PayloadTracker: _send_to_payload_tracker
    Processor-->>Processor: await outbound futures
    deactivate Processor
    Listener-->>Listener: release semaphore
    deactivate Listener

Loading

Sequence diagram for grouper queued processing span lifetime

sequenceDiagram
    participant Kafka
    participant Grouper
    participant Queue
    participant Evaluator
    participant PayloadTracker

    Kafka->>Grouper: consume_message
    Grouper->>Queue: push_inventory_msg or push_advisor_msg
    Note over Queue: QueueItem stores otel_context and topic
    Queue->>Queue: _start_item_processing
    activate Queue
    Queue-->>Queue: await pair event or timeout
    Queue->>PayloadTracker: send_msg_to_payload_tracker
    Queue->>Evaluator: send
    Queue-->>Queue: await tracker and evaluator futures
    deactivate Queue

Loading

File-Level Changes

Change	Details	Files
Move Kafka consumer spans from message scheduling/enqueue boundaries into the awaited processing lifecycles.	Create listener consumer spans inside inventory and advisor processing wrappers, preserving extracted OTel and request context. Store the originating topic and OTel context on grouper queue items, then create the consumer span around pairing, timeout, release, and evaluation processing. Populate span attributes and thread context for organization and request identifiers.	listener/listener.py grouper/grouper.py grouper/queue.py grouper/common.py
Await outbound Kafka operations before closing consumer spans.	Return producer futures from payload-tracker helpers and processor send methods. Await tracker, grouper, evaluator, and related send futures during listener and grouper processing.	common/utils.py listener/advisor_processor.py listener/inventory_processor.py grouper/queue.py
Add tracing regression coverage for parent-span lifetime and context propagation.	Use in-memory OpenTelemetry exporters to verify consumer spans remain active during processing. Assert child-span nesting, parent end times, messaging attributes, and thread context values for listener and grouper flows.	tests/listener_tests/test_listener_tracing.py tests/grouper_tests/test_grouper_tracing.py tests/grouper_tests/__init__.py

---

Tips and commands

Interacting with Sourcery

• Trigger a new review: Comment @sourcery-ai review on the pull request.
• Continue discussions: Reply directly to Sourcery's review comments.
• Generate a GitHub issue from a review comment: Ask Sourcery to create an
  issue from a review comment by replying to it. You can also reply to a
  review comment with @sourcery-ai issue to create an issue from it.
• Generate a pull request title: Write @sourcery-ai anywhere in the pull
  request title to generate a title at any time. You can also comment
  @sourcery-ai title on the pull request to (re-)generate the title at any time.
• Generate a pull request summary: Write @sourcery-ai summary anywhere in
  the pull request body to generate a PR summary at any time exactly where you
  want it. You can also comment @sourcery-ai summary on the pull request to
  (re-)generate the summary at any time.
• Generate reviewer's guide: Comment @sourcery-ai guide on the pull
  request to (re-)generate the reviewer's guide at any time.
• Resolve all Sourcery comments: Comment @sourcery-ai resolve on the
  pull request to resolve all Sourcery comments. Useful if you've already
  addressed all the comments and don't want to see them anymore.
• Dismiss all Sourcery reviews: Comment @sourcery-ai dismiss on the pull
  request to dismiss all existing Sourcery reviews. Especially useful if you
  want to start fresh with a new review - don't forget to comment
  @sourcery-ai review to trigger a new review!

Customizing Your Experience

Access your dashboard to:

• Enable or disable review features such as the Sourcery-generated pull request
  summary, the reviewer's guide, and others.
• Change the review language.
• Add, remove or edit custom review instructions.
• Adjust other review settings.

Getting Help

• Contact our support team for questions or feedback.
• Visit our documentation for detailed guides and information.
• Keep in touch with the Sourcery team by following us on X/Twitter, LinkedIn or GitHub.

[sourcery-ai]

Hey - I've found 1 issue

Prompt for AI Agents

Please address the comments from this code review:

## Individual Comments

### Comment 1
<location path="tests/grouper_tests/test_grouper_tracing.py" line_range="85" />
<code_context>
+        topic="platform.vulnerability.inventory-upload",
+    )
+
+    # Let the background task run to completion
+    await asyncio.sleep(0.05)
+
</code_context>
<issue_to_address>
**issue (testing):** The test assumes the background grouper task completes within a fixed 50ms sleep and then immediately indexes `active_checks`; under CI scheduling or load, the task has not run yet and the test raises `KeyError` or observes incomplete state.

**Triggers:** When the test process is delayed for more than 50ms.

**Suggested fix:** Synchronize on an explicit completion event or await the task returned by `create_task_and_log` instead of sleeping for a fixed duration.
</issue_to_address>

---

Sourcery is free for open source - if you like our reviews please consider sharing them ✨

• X
• Mastodon
• LinkedIn
• Facebook