[receiver/receivercreator] Ffeat/add kafkatopicsobserver to receivercreator (open-telemetry#39086)

#### Description
Add kafkatopicsobserver to the receivercreator configuration

<!-- Issue number (e.g. open-telemetry#1234) or full URL to issue, if applicable. -->
#### Link to tracking issue
New component 
open-telemetry#37665 

#### Testing
Manual and unit test

#### Documentation
Added new sections in README file

---------

Co-authored-by: Antoine Toulme <[email protected]>

Author: 2 people

.chloggen/feat_add-kafkatopicsobserver-to-receivercreator.yaml

@@ -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: enhancement
+
+# The name of the component, or a single word describing the area of concern, (e.g. filelogreceiver)
+component: receivercreator
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: Add kafkatopicsobserver to the receivercreator configuration
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [37665]
+
+# (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]

receiver/receivercreator/README.md

@@ -153,6 +153,10 @@ None
 |--------------------|-------------------|
 | k8s.namespace.name | \`namespace\`     |
 
+`type == "kafka.topics"`
+
+None
+
 See `redis/2` in [examples](#examples).
 
 
@@ -289,6 +293,12 @@ targeting it will have different variables available.
 | labels                | A key-value map of user-specified node metadata                      | Map with String key and value |
 | kubelet_endpoint_port | The node Status object's DaemonEndpoints.KubeletEndpoint.Port value  | Integer                       |
 
+### Kafka Topics
+| Variable              | Description                                                          | Data Type                     |
+|-----------------------|----------------------------------------------------------------------|-------------------------------|
+| type                  | `"kafka.topics"`                                                     | String                        |
+| id                    | ID of source endpoint                                                | String                        |
+
 ## Examples
 
 ```yaml
@@ -299,6 +309,11 @@ extensions:
     observe_services: true
     observe_ingresses: true
   host_observer:
+  kafkatopics_observer:
+    brokers: ["1.2.3.4:9093"]
+    protocol_version: 3.9.0
+    topic_regex: "^foo_topic[0-9]$"
+    topics_sync_interval: 5s
 
 receivers:
   receiver_creator/1:
@@ -417,6 +432,20 @@ receivers:
             - type: add
               field: attributes.log.template
               value: lazybox
+  receiver_creator/kafka:
+    watch_observers: [kafkatopics_observer]
+    receivers:
+      kafka:
+        rule: type == "kafka.topics"
+        config:
+          protocol_version: 3.9.0
+          topic: '`endpoint`'
+          encoding: text
+          brokers: ["1.2.3.4:9093"]
+          initial_offset: earliest
+          header_extraction:
+            extract_headers: true
+            headers: ["index", "source", "sourcetype", "host"]
 
 processors:
   exampleprocessor:
@@ -431,10 +460,10 @@ service:
       processors: [exampleprocessor]
       exporters: [exampleexporter]
     logs:
-      receivers: [receiver_creator/logs]
+      receivers: [receiver_creator/logs, receiver_creator/kafka]
       processors: [exampleprocessor]
       exporters: [exampleexporter]
-  extensions: [k8s_observer, host_observer]
+  extensions: [k8s_observer, host_observer, kafkatopics_observer]
 ```
 
 The full list of settings exposed for this receiver are documented in [config.go](./config.go)

receiver/receivercreator/config.go

@@ -106,7 +106,7 @@ func (cfg *Config) Unmarshal(componentParser *confmap.Conf) error {
 
 	for endpointType := range cfg.ResourceAttributes {
 		switch endpointType {
-		case observer.ContainerType, observer.K8sServiceType, observer.K8sIngressType, observer.HostPortType, observer.K8sNodeType, observer.PodType, observer.PortType, observer.PodContainerType:
+		case observer.ContainerType, observer.K8sServiceType, observer.K8sIngressType, observer.HostPortType, observer.K8sNodeType, observer.PodType, observer.PortType, observer.PodContainerType, observer.KafkaTopicType:
 		default:
 			return fmt.Errorf("resource attributes for unsupported endpoint type %q", endpointType)
 		}

receiver/receivercreator/config_test.go

@@ -120,6 +120,7 @@ func TestLoadConfig(t *testing.T) {
 					observer.K8sServiceType:   {"k8s.service.key": "k8s.service.value"},
 					observer.K8sIngressType:   {"k8s.ingress.key": "k8s.ingress.value"},
 					observer.K8sNodeType:      {"k8s.node.key": "k8s.node.value"},
+					observer.KafkaTopicType:   {},
 				},
 			},
 		},

receiver/receivercreator/factory.go

@@ -66,6 +66,7 @@ func createDefaultConfig() component.Config {
 				conventions.AttributeK8SNodeName: "`name`",
 				conventions.AttributeK8SNodeUID:  "`uid`",
 			},
+			observer.KafkaTopicType: map[string]string{},
 		},
 		receiverTemplates: map[string]receiverTemplate{},
 	}

receiver/receivercreator/fixtures_test.go

@@ -150,6 +150,12 @@ var k8sNodeEndpoint = observer.Endpoint{
 	},
 }
 
+var kafkaTopicsEndpoint = observer.Endpoint{
+	ID:      "topic1",
+	Target:  "topic1",
+	Details: &observer.KafkaTopic{},
+}
+
 var unsupportedEndpoint = observer.Endpoint{
 	ID:      "endpoint-1",
 	Target:  "localhost:1234",

receiver/receivercreator/rules.go

@@ -22,7 +22,7 @@ type rule struct {
 
 // ruleRe is used to verify the rule starts type check.
 var ruleRe = regexp.MustCompile(
-	fmt.Sprintf(`^type\s*==\s*(%q|%q|%q|%q|%q|%q|%q|%q)`, observer.PodType, observer.K8sServiceType, observer.K8sIngressType, observer.PortType, observer.PodContainerType, observer.HostPortType, observer.ContainerType, observer.K8sNodeType),
+	fmt.Sprintf(`^type\s*==\s*(%q|%q|%q|%q|%q|%q|%q|%q|%q)`, observer.PodType, observer.K8sServiceType, observer.K8sIngressType, observer.PortType, observer.PodContainerType, observer.HostPortType, observer.ContainerType, observer.K8sNodeType, observer.KafkaTopicType),
 )
 
 // newRule creates a new rule instance.

receiver/receivercreator/rules_test.go

@@ -34,6 +34,7 @@ func Test_ruleEval(t *testing.T) {
 		{"basic k8s.node", args{`type == "k8s.node" && kubelet_endpoint_port == 10250`, k8sNodeEndpoint}, true, false},
 		{"relocated type builtin", args{`type == "k8s.node" && typeOf("some string") == "string"`, k8sNodeEndpoint}, true, false},
 		{"pod container", args{`type == "pod.container" and container_image matches "redis"`, podContainerEndpointWithHints}, true, false},
+		{"kafka topics", args{`type == "kafka.topics"`, kafkaTopicsEndpoint}, true, false},
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {