[receiver/kafka] Add support for topic and exclude_topic as an array #44575
There are no files selected for viewing
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,27 @@ | ||
| # Use this changelog template to create an entry for release notes. | ||
|
|
||
| # One of 'breaking', 'deprecation', 'new_component', 'enhancement', 'bug_fix' | ||
| change_type: deprecation | ||
|
|
||
| # The name of the component, or a single word describing the area of concern, (e.g. receiver/filelog) | ||
| component: receiver/kafka | ||
|
|
||
| # A brief description of the change. Surround your text with quotes ("") if it needs to start with a backtick (`). | ||
| note: Support configuring a list of topics and exclude_topics; deprecate topic and exclude_topic | ||
|
|
||
| # Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists. | ||
| issues: [44477] | ||
|
|
||
| # (Optional) One or more lines of additional information to render under the primary note. | ||
| # These lines will be padded with 2 spaces and then inserted directly into the document. | ||
| # Use pipe (|) for multiline entries. | ||
| subtext: | ||
|
|
||
| # If your change doesn't affect end users or the exported elements of any package, | ||
| # you should instead start your pull request title with [chore] or use the "Skip Changelog" label. | ||
| # Optional: The change log or logs in which this entry should be included. | ||
| # e.g. '[user]' or '[user, api]' | ||
| # Include 'user' if the change is relevant to end users. | ||
| # Include 'api' if there is a change to a library API. | ||
| # Default: '[user]' | ||
| change_logs: [user] |
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -53,6 +53,79 @@ | |
| return err | ||
| } | ||
|
|
||
| // Check if deprecated fields have been explicitly set | ||
| // give them precedence | ||
| var zeroConfig Config | ||
| if err := conf.Unmarshal(&zeroConfig); err != nil { | ||
| return err | ||
| } | ||
|
|
||
| // handle deprecated topic and exclude_topic for log signal | ||
| if zeroConfig.Logs.Topic != "" { | ||
| if len(zeroConfig.Logs.Topics) == 0 { | ||
|
Check failure on line 65 in receiver/kafkareceiver/config.go
|
||
| c.Logs.Topics = []string{zeroConfig.Logs.Topic} | ||
| } else { | ||
| return fmt.Errorf("both logs.topic and logs.topics cannot be set") | ||
|
Check failure on line 68 in receiver/kafkareceiver/config.go
|
||
|
||
| } | ||
| } | ||
| if zeroConfig.Logs.ExcludeTopic != "" { | ||
| if len(zeroConfig.Logs.ExcludeTopics) == 0 { | ||
|
Check failure on line 72 in receiver/kafkareceiver/config.go
|
||
| c.Logs.ExcludeTopics = []string{zeroConfig.Logs.ExcludeTopic} | ||
| } else { | ||
| return fmt.Errorf("both logs.exclude_topic and logs.exclude_topics cannot be set") | ||
|
Check failure on line 75 in receiver/kafkareceiver/config.go
|
||
| } | ||
| } | ||
|
|
||
| // handle deprecated topic and exclude_topic for metric signal | ||
| if zeroConfig.Metrics.Topic != "" { | ||
| if len(zeroConfig.Logs.Topics) == 0 { | ||
|
Check failure on line 81 in receiver/kafkareceiver/config.go
|
||
| c.Metrics.Topics = []string{zeroConfig.Metrics.Topic} | ||
| } else { | ||
| return fmt.Errorf("both metrics.topic and metrics.topics cannot be set") | ||
|
Check failure on line 84 in receiver/kafkareceiver/config.go
|
||
| } | ||
| } | ||
|
|
||
| if zeroConfig.Metrics.ExcludeTopic != "" { | ||
| if len(zeroConfig.Logs.ExcludeTopics) == 0 { | ||
|
Check failure on line 89 in receiver/kafkareceiver/config.go
|
||
| c.Metrics.ExcludeTopics = []string{zeroConfig.Metrics.ExcludeTopic} | ||
| } else { | ||
| return fmt.Errorf("both metrics.exclude_topic and metrics.exclude_topics cannot be set") | ||
|
Check failure on line 92 in receiver/kafkareceiver/config.go
|
||
| } | ||
| } | ||
|
|
||
| // handle deprecated topic and exclude_topic for trace signal | ||
| if zeroConfig.Traces.Topic != "" { | ||
| if len(zeroConfig.Logs.Topics) == 0 { | ||
|
Check failure on line 98 in receiver/kafkareceiver/config.go
|
||
| c.Traces.Topics = []string{zeroConfig.Traces.Topic} | ||
| } else { | ||
| return fmt.Errorf("both traces.topic and traces.topics cannot be set") | ||
|
Check failure on line 101 in receiver/kafkareceiver/config.go
|
||
| } | ||
| } | ||
|
|
||
| if zeroConfig.Traces.ExcludeTopic != "" { | ||
| if len(zeroConfig.Logs.ExcludeTopics) == 0 { | ||
| c.Traces.ExcludeTopics = []string{zeroConfig.Traces.ExcludeTopic} | ||
| } else { | ||
| return fmt.Errorf("both traces.exclude_topic and traces.exclude_topics cannot be set") | ||
| } | ||
| } | ||
|
|
||
| // handle deprecated topic and exclude_topic for profile signal | ||
| if zeroConfig.Profiles.Topic != "" { | ||
| if len(zeroConfig.Logs.Topics) == 0 { | ||
| c.Profiles.Topics = []string{zeroConfig.Profiles.Topic} | ||
| } else { | ||
| return fmt.Errorf("both profiles.topic and profiles.topics cannot be set") | ||
| } | ||
| } | ||
| if zeroConfig.Profiles.ExcludeTopic != "" { | ||
| if len(zeroConfig.Logs.ExcludeTopics) == 0 { | ||
| c.Profiles.ExcludeTopics = []string{zeroConfig.Profiles.ExcludeTopic} | ||
| } else { | ||
| return fmt.Errorf("both profiles.exclude_topic and profiles.exclude_topics cannot be set") | ||
| } | ||
| } | ||
|
|
||
| // Set OnPermanentError default value to inherit from OnError for backward compatibility | ||
| // Only if OnPermanentError was not explicitly set in the config | ||
| rawConf := conf.Get("message_marking") | ||
|
|
@@ -73,61 +146,81 @@ | |
| // Validate checks the receiver configuration is valid. | ||
| func (c *Config) Validate() error { | ||
axw marked this conversation as resolved.
Show resolved
Hide resolved
|
||
| // Validate that exclude_topic is only used with regex topic patterns | ||
| if err := validateExcludeTopic("logs", c.Logs.Topics, c.Logs.ExcludeTopics); err != nil { | ||
| return err | ||
| } | ||
| if err := validateExcludeTopic("metrics", c.Metrics.Topics, c.Metrics.ExcludeTopics); err != nil { | ||
| return err | ||
| } | ||
| if err := validateExcludeTopic("traces", c.Traces.Topics, c.Traces.ExcludeTopics); err != nil { | ||
| return err | ||
| } | ||
| if err := validateExcludeTopic("profiles", c.Profiles.Topics, c.Profiles.ExcludeTopics); err != nil { | ||
| return err | ||
| } | ||
| return nil | ||
| } | ||
|
|
||
| // validateExcludeTopic checks that exclude_topic is only configured when topics uses regex pattern | ||
| func validateExcludeTopic(signalType string, topics, excludeTopics []string) error { | ||
| if len(excludeTopics) == 0 { | ||
| return nil // No exclude_topic configured, nothing to validate | ||
| } | ||
|
|
||
| // if none of the configured topic uses regex return error | ||
| var usesRegex bool | ||
| for _, topic := range topics { | ||
| if strings.HasPrefix(topic, "^") { | ||
| usesRegex = true | ||
khushijain21 marked this conversation as resolved.
Show resolved
Hide resolved
|
||
| break | ||
| } | ||
| } | ||
|
|
||
| if !usesRegex { | ||
| return fmt.Errorf( | ||
| "%s.exclude_topics is configured but none of the configured %s.topics use regex pattern (must start with '^')", | ||
| signalType, signalType, | ||
| ) | ||
| } | ||
|
|
||
| for _, excludeTopic := range excludeTopics { | ||
| // Validate that exclude_topic is a valid regex pattern | ||
| if _, err := regexp.Compile(excludeTopic); err != nil { | ||
| return fmt.Errorf( | ||
| "%s.exclude_topic contains invalid regex pattern: %w", | ||
| signalType, err, | ||
| ) | ||
| } | ||
| } | ||
|
|
||
| return nil | ||
| } | ||
|
|
||
| // TopicEncodingConfig holds signal-specific topic and encoding configuration. | ||
| type TopicEncodingConfig struct { | ||
| // Deprecated [v0.142.0]: Use Topics | ||
| Topic string `mapstructure:"topic"` | ||
|
|
||
| // Topics holds the name of the Kafka topics from which messages of the | ||
| // signal type should be consumed. | ||
| // | ||
| // The default depends on the signal type: | ||
| // - "otlp_spans" for traces | ||
| // - "otlp_metrics" for metrics | ||
| // - "otlp_logs" for logs | ||
| // - "otlp_profiles" for profiles | ||
| Topics []string `mapstructure:"topics"` | ||
|
|
||
| // Encoding holds the expected encoding of messages for the signal type | ||
| // | ||
| // Defaults to "otlp_proto". | ||
| Encoding string `mapstructure:"encoding"` | ||
|
|
||
| // Deprecated [v0.142.0]: Use ExcludeTopics | ||
| ExcludeTopic string `mapstructure:"exclude_topic"` | ||
|
|
||
| // Optional exclude topics option, used only in regex mode. | ||
| ExcludeTopics []string `mapstructure:"exclude_topics"` | ||
| } | ||
|
|
||
| type MessageMarking struct { | ||
|
|
||
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.