notify: add Kafka receiver - #5410
Conversation
Signed-off-by: cshorvath <cshorvath@ultinous.com>
📝 WalkthroughWalkthroughAdded an Apache Kafka notification receiver that publishes webhook v4 JSON messages. Added Kafka configuration, receiver wiring, documentation, metrics registration, and tests. Added integration lifecycle cleanup during receiver construction, reload, and shutdown. ChangesKafka receiver
Estimated code review effort: 4 (Complex) | ~45 minutes Sequence Diagram(s)sequenceDiagram
participant Alertmanager
participant KafkaNotifier as notify/kafka.Notifier
participant KafkaProducer
participant KafkaBroker
Alertmanager->>KafkaNotifier: Notify grouped alerts
KafkaNotifier->>KafkaNotifier: Render webhook v4 JSON
KafkaNotifier->>KafkaProducer: ProduceSync topic, group key, JSON value
KafkaProducer->>KafkaBroker: Publish Kafka record
Possibly related PRs
🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (4 passed)
✨ 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 |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 1
🧹 Nitpick comments (2)
notify/kafka/config.go (1)
45-53: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winWrap propagated Kafka errors with operation context.
Direct error returns make YAML parsing and producer construction failures harder to locate. Wrap each propagated error with
fmt.Errorf("kafka: <operation>: %w", err).
notify/kafka/config.go#L45-L53: wrap YAML unmarshal and client-option validation errors.notify/kafka/kafka.go#L46-L55: wrap configuration validation and producer-option construction errors.As per coding guidelines, “Wrap errors with
fmt.Errorf("...: %w", err)and check witherrors.Is/errors.Asin Go code.”🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@notify/kafka/config.go` around lines 45 - 53, Wrap the propagated errors in Config unmarshalling and validate using fmt.Errorf with “kafka: <operation>: %w” context, covering both sites: notify/kafka/config.go lines 45-53 for YAML unmarshal and client-option validation, and notify/kafka/kafka.go lines 46-55 for configuration validation and producer-option construction. Preserve error unwrapping so callers can continue using errors.Is and errors.As.Source: Coding guidelines
app/reloader.go (1)
127-130: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winAdd receiver context to the joined error.
A receiver build failure is returned without the receiver name. Wrap
errbefore joining it with cleanup errors.As per coding guidelines, “Wrap errors with
fmt.Errorf("...: %w", err)and check witherrors.Is/errors.Asin Go code.”Proposed fix
- return errors.Join(err, notify.CloseIntegrations(integrations)) + return errors.Join( + fmt.Errorf("build receiver %q: %w", rcv.Name, err), + notify.CloseIntegrations(integrations), + )🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@app/reloader.go` around lines 127 - 130, Update the receiver build error handling around BuildReceiverIntegrations to wrap err with receiver context using fmt.Errorf and %w before joining it with notify.CloseIntegrations(integrations), including the receiver name in the message while preserving errors.Is/errors.As unwrapping.Source: Coding guidelines
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@docs/configuration.md`:
- Around line 2039-2042: Update both occurrences of the producer acknowledgement
documentation in docs/configuration.md to remove the claim that acks: "all"
enables idempotent writes. Keep the description focused on waiting for all
in-sync replicas, while preserving the existing acknowledgement-level behavior
and syntax.
---
Nitpick comments:
In `@app/reloader.go`:
- Around line 127-130: Update the receiver build error handling around
BuildReceiverIntegrations to wrap err with receiver context using fmt.Errorf and
%w before joining it with notify.CloseIntegrations(integrations), including the
receiver name in the message while preserving errors.Is/errors.As unwrapping.
In `@notify/kafka/config.go`:
- Around line 45-53: Wrap the propagated errors in Config unmarshalling and
validate using fmt.Errorf with “kafka: <operation>: %w” context, covering both
sites: notify/kafka/config.go lines 45-53 for YAML unmarshal and client-option
validation, and notify/kafka/kafka.go lines 46-55 for configuration validation
and producer-option construction. Preserve error unwrapping so callers can
continue using errors.Is and errors.As.
🪄 Autofix
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Path: .coderabbit.yaml
Review profile: CHILL
Plan: Pro Plus
Run ID: 00bb60ba-8a61-44d4-9c0e-4ab3746914f7
📒 Files selected for processing (17)
CHANGELOG.mdapp/reloader.goapp/reloader_test.goconfig/config.goconfig/config_test.goconfig/receiver/receiver.goconfig/receiver/receiver_test.godocs/configuration.mddocs/integrations.mdkafka/kafka.gonotify/integration_close_test.gonotify/kafka/config.gonotify/kafka/config_test.gonotify/kafka/kafka.gonotify/kafka/kafka_test.gonotify/metrics.gonotify/notify.go
| # Producer acknowledgement level. With "none", broker-side delivery failures | ||
| # cannot be reported to Alertmanager. "leader" waits for the partition leader. | ||
| # "all" waits for all in-sync replicas and enables idempotent writes. | ||
| [ acks: <"none" | "leader" | "all"> | default = "leader" ] |
There was a problem hiding this comment.
🎯 Functional Correctness | 🟡 Minor | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo "Occurrences of acks configuration in docs/configuration.md:"
rg -n 'acks|idempotent|franz-go|kgo' docs/configuration.md || true
echo
echo "Relevant lines around docs/configuration.md:2028-2048:"
sed -n '2028,2050p' docs/configuration.md | nl -ba -v2028
echo
echo "Check git diff for docs/configuration.md and references to producer/acks/idempotent:"
git diff -- docs/configuration.md | sed -n '1,160p' || trueRepository: prometheus/alertmanager
Length of output: 2812
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo "Configuration file references for franz-go producer settings:"
rg -n 'franz-go|idempotent|acks|enable-idempotent|idempotent' . -g '!vendor/**' -g '!node_modules/**' || true
echo
echo "docs/configuration.md lines 2305-2328:"
sed -n '2305,2328p' docs/configuration.mdRepository: prometheus/alertmanager
Length of output: 15698
🌐 Web query:
twmb franz-go kgo NewClient idempotent producer default produce producer transaction producer doc DefaultProduceTopic idempotent
💡 Result:
The twmb/franz-go Kafka client is designed with an idempotent producer enabled by default [1][2][3]. This aligns with modern Kafka standards where idempotent production is considered a best practice for ensuring exactly-once semantics per partition and avoiding duplicates in the face of retries [1][4]. Key details regarding this behavior include: Idempotent Producer Default: The kgo client automatically opts into idempotent writes [1][3]. This functionality can be explicitly disabled using the DisableIdempotentWrite producer option if required (e.g., due to specific cluster permission constraints like missing IDEMPOTENT_WRITE privileges) [5][3]. When idempotency is disabled, the client defaults to a lower number of in-flight requests per broker compared to the default of 5 used when idempotency is enabled [5]. Transactional Producer: Transactional producing is also fully supported [1][6]. While the producer itself is idempotent by default, transactional operations are treated as an extension of the idempotent producer [1]. You initiate transactions manually via BeginTransaction and EndTransaction calls [1][7]. Note that configuring a transactional ID is incompatible with disabling idempotency; the client will return an error if you attempt to use both [5]. Regarding DefaultProduceTopic: There is no specific "DefaultProduceTopic" setting in the kgo client configuration that automatically assigns a topic to all produce requests. When using the Produce method, you must explicitly specify the topic (typically within the kmsg.ProduceRecord or similar record structure) for each message you intend to send [1][8]. Summary of Key Options: - DisableIdempotentWrite: Used to opt out of idempotent production [5]. - TransactionalID: Setting this enables transactional producer capabilities [5][7]. In practice, users should generally keep the default idempotent settings unless there is a specific architectural reason to disable them, as they provide strong delivery guarantees without performance penalties [1][5][4].
Citations:
- 1: https://github.com/twmb/franz-go/blob/master/docs/producing-and-consuming.md
- 2: https://github.com/twmb/franz-go
- 3: CLUSTER_AUTHORIZATION_FAILED when producing. twmb/franz-go#26
- 4: Is it possible to specify the concurrency level of produce requests with idempotent disabled? twmb/franz-go#129
- 5: https://github.com/twmb/franz-go/blob/master/pkg/kgo/config.go
- 6: https://pkg.go.dev/github.com/twmb/franz-go/pkg/kgo
- 7: https://github.com/twmb/franz-go/blob/master/docs/transactions.md
- 8: https://github.com/twmb/franz-go/blob/master/pkg/kgo/client.go
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo "kafka/kafka.go factory/body around kopts construction:"
sed -n '95,140p' kafka/kafka.go
echo
echo "Shared kafka config validation construction:"
sed -n '65,135p' notify/kafka/config.go
echo
echo "Docs around line 2038-2043 and 2314-2320:"
python3 - <<'PY'
from pathlib import Path
for path in ['docs/configuration.md']:
text = Path(path).read_text()
lines = text.splitlines()
for start,end in [(2038,2043),(2314,2320)]:
print(f"\n--- {path}:{start}-{end} ---")
for i,l in enumerate(lines[start-1:end], start=start):
print(f"{i}: {l}")
PYRepository: prometheus/alertmanager
Length of output: 2675
🌐 Web query:
kw "acks=all" "EnableIdempotentWrite" "EnableIdempotentWrites" "DisableIdempotentWrite" franz-go config.go
💡 Result:
In the franz-go (kgo) library, idempotent production is enabled by default [1][2]. The library does not use an "EnableIdempotentWrite" or "EnableIdempotentWrites" option because idempotency is the standard operating mode [3][4]. To opt out of this behavior, you use the DisableIdempotentWrite() producer option [3][5]. This is typically only necessary if you require specific configurations, such as producing with no acknowledgments (acks=none) or only leader acknowledgments (acks=leader), which are incompatible with the default idempotent settings [1][6]. Regarding Kafka acknowledgments and idempotency: - Idempotent production requires acks=all (or equivalent) to function correctly, as it relies on server-side deduplication [3]. - If you use DisableIdempotentWrite(), you can configure other acknowledgment levels like kgo.LeaderAck() or kgo.NoAck() [6]. - When idempotency is disabled, the default maximum number of in-flight produce requests per broker is 1 [3]. If idempotency is enabled, this limit is 5 (for Kafka v1+) or 1 (for Kafka v0.11) [3]. In summary, there is no "Enable" option; simply omit DisableIdempotentWrite() to maintain the default idempotent behavior [1][2].
Citations:
- 1: https://github.com/twmb/franz-go/blob/master/docs/producing-and-consuming.md
- 2: https://pkg.go.dev/github.com/twmb/franz-go/pkg/kgo
- 3: https://github.com/twmb/franz-go/blob/master/pkg/kgo/config.go
- 4: CLUSTER_AUTHORIZATION_FAILED when producing. twmb/franz-go#26
- 5: https://github.com/twmb/franz-go/blob/master/pkg/kgo/client.go
- 6: https://github.com/open-telemetry/opentelemetry-collector-contrib/blob/main/internal/kafka/franz_client.go
🌐 Web query:
twmb franz-go package config.go acks default producer Acks default ProduceClient defaults producer option
💡 Result:
In the twmb/franz-go library, the default configuration for producer acknowledgments (acks) is AllISRAcks [1]. While the Acks documentation in config.go contains a comment stating "The default is LeaderAck" [1], this appears to be a documentation discrepancy when compared to the library's actual implementation and stated delivery guarantees. The library supports and enforces strong delivery guarantees by default, consistent with KIP-679 [2][3][4]. Specifically, franz-go defaults to idempotent production [5][3], and the library code explicitly initializes the producer configuration with AllISRAcks() [1]. Key details regarding producer acks in franz-go: 1. Default Acks: The client defaults to AllISRAcks() (corresponding to Kafka acks=-1), which ensures all in-sync replicas acknowledge the record before the leader replies [1]. 2. Configuration Options: - RequiredAcks(Acks): This ProducerOpt is used to override the default acknowledgment setting [1]. - Acks Types: - NoAck: Corresponds to acks=0 (no acknowledgment) [1]. - LeaderAck: Corresponds to acks=1 (leader acknowledgment only) [1]. - AllISRAcks: Corresponds to acks=-1 (all in-sync replicas) [1]. 3. Idempotency: franz-go uses idempotent production by default [5]. If idempotency is disabled, certain constraints apply (e.g., specific requirements for inflight requests and acks) [1]. If you need to change the acknowledgment behavior, you can pass the kgo.RequiredAcks option when initializing the kgo.Client [1].
Citations:
- 1: https://github.com/twmb/franz-go/blob/master/pkg/kgo/config.go
- 2: https://pkg.go.dev/github.com/twmb/franz-go/pkg/kgo
- 3: https://pkg.go.dev/github.com/twmb/franz-go
- 4: https://github.com/twmb/franz-go/
- 5: https://github.com/twmb/franz-go/blob/master/docs/producing-and-consuming.md
Do not treat acks: all as enabling idempotent writes.
franz-go enables idempotent production by default. acks: all only affects the broker acknowledgment level, not the idempotency setting. Remove the idempotency claim from both docs/configuration.md occurrences.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@docs/configuration.md` around lines 2039 - 2042, Update both occurrences of
the producer acknowledgement documentation in docs/configuration.md to remove
the claim that acks: "all" enables idempotent writes. Keep the description
focused on waiting for all in-sync replicas, while preserving the existing
acknowledgement-level behavior and syntax.
2 participants
Pull Request Checklist
Please check all the applicable boxes.
benchstatto compare benchmarksWhich user-facing changes does this PR introduce?