Add Kafka broker attributes extraction (#1927)

Author: jefchien

go.mod

@@ -174,6 +174,7 @@ require (
 	github.com/prometheus/prometheus v0.300.1
 	github.com/shirou/gopsutil v3.21.11+incompatible
 	github.com/shirou/gopsutil/v3 v3.24.5
+	github.com/shirou/gopsutil/v4 v4.25.3
 	github.com/stretchr/testify v1.10.0
 	github.com/xeipuuv/gojsonschema v1.2.0
 	go.opentelemetry.io/collector/client v1.30.0
@@ -497,7 +498,6 @@ require (
 	github.com/rs/cors v1.11.1 // indirect
 	github.com/safchain/ethtool v0.0.0-20210803160452-9aa261dae9b1 // indirect
 	github.com/scaleway/scaleway-sdk-go v1.0.0-beta.32 // indirect
-	github.com/shirou/gopsutil/v4 v4.25.3 // indirect
 	github.com/shurcooL/httpfs v0.0.0-20230704072500-f1e31cf0ba5c // indirect
 	github.com/sirupsen/logrus v1.9.3 // indirect
 	github.com/sleepinggenius2/gosmi v0.4.4 // indirect

internal/detector/java/extract/name.go

@@ -5,7 +5,6 @@ package extract
 
 import (
 	"archive/zip"
-	"bufio"
 	"context"
 	"fmt"
 	"io"
@@ -14,6 +13,7 @@ import (
 	"strings"
 
 	"github.com/aws/amazon-cloudwatch-agent/internal/detector"
+	"github.com/aws/amazon-cloudwatch-agent/internal/detector/util"
 	"github.com/aws/amazon-cloudwatch-agent/internal/util/collections"
 )
 
@@ -23,6 +23,8 @@ const (
 
 	// metaManifestFile is the path to the Manifest in the Java archive.
 	metaManifestFile = "META-INF/MANIFEST.MF"
+	// metaManifestSeparator is the separator of the key/value pairs in the file.
+	metaManifestSeparator = ':'
 	// metaManifestApplicationName is a non-standard, but explicit field that should be used if present.
 	metaManifestApplicationName = "Application-Name"
 	// metaManifestImplementationTitle is a standard field that is conventionally used to name the application.
@@ -188,7 +190,7 @@ func (e *archiveManifestNameExtractor) Extract(ctx context.Context, process dete
 		return "", detector.ErrIncompatibleExtractor
 	}
 	fallback := strings.TrimSuffix(filepath.Base(arg), filepath.Ext(arg))
-	path, err := absPath(ctx, process, arg)
+	path, err := util.AbsPath(ctx, process, arg)
 	e.logger.Debug("Trying to extract name from Java Archive", "pid", process.PID(), "path", path)
 	if err != nil {
 		return fallback, nil
@@ -225,40 +227,22 @@ func (e *archiveManifestNameExtractor) readManifest(jarPath string) (string, err
 		return "", err
 	}
 	defer rc.Close()
-	return e.parseManifest(rc), nil
+	return parseManifest(rc, e.fieldPriority, e.fieldLookup), nil
 }
 
-func (e *archiveManifestNameExtractor) parseManifest(r io.Reader) string {
-	manifest := make(map[string]string, len(e.fieldPriority))
-	scanner := bufio.NewScanner(r)
-	for scanner.Scan() {
-		line := scanner.Text()
-		if parts := strings.SplitN(line, ":", 2); len(parts) == 2 {
-			field := strings.TrimSpace(parts[0])
-			if e.fieldLookup.Contains(field) {
-				manifest[field] = strings.TrimSpace(parts[1])
-				// exit early if the highest priority field is found
-				if field == e.fieldPriority[0] && manifest[field] != "" {
-					return manifest[field]
-				}
-			}
+func parseManifest(r io.Reader, fieldPriority []string, fieldLookup collections.Set[string]) string {
+	manifest := make(map[string]string, len(fieldPriority))
+	_ = util.ReadProperties(r, metaManifestSeparator, func(key, value string) bool {
+		if !fieldLookup.Contains(key) || value == "" {
+			return true
 		}
-	}
-	for _, field := range e.fieldPriority {
+		manifest[key] = value
+		return key != fieldPriority[0]
+	})
+	for _, field := range fieldPriority {
 		if name := manifest[field]; name != "" {
 			return name
 		}
 	}
 	return ""
 }
-
-func absPath(ctx context.Context, process detector.Process, path string) (string, error) {
-	if filepath.IsAbs(path) {
-		return path, nil
-	}
-	cwd, err := process.CwdWithContext(ctx)
-	if err != nil {
-		return "", err
-	}
-	return filepath.Join(cwd, path), nil
-}

internal/detector/java/java.go

@@ -10,7 +10,7 @@ import (
 
 	"github.com/aws/amazon-cloudwatch-agent/internal/detector"
 	"github.com/aws/amazon-cloudwatch-agent/internal/detector/java/extract"
-	"github.com/aws/amazon-cloudwatch-agent/internal/detector/kafka"
+	"github.com/aws/amazon-cloudwatch-agent/internal/detector/kafkabroker"
 	"github.com/aws/amazon-cloudwatch-agent/internal/detector/tomcat"
 	"github.com/aws/amazon-cloudwatch-agent/internal/detector/util"
 )
@@ -35,7 +35,7 @@ func NewDetector(logger *slog.Logger, nameFilter detector.NameFilter) detector.P
 		logger: logger,
 		subDetectors: []detector.ProcessDetector{
 			tomcat.NewDetector(logger),
-			kafka.NewDetector(logger),
+			kafkabroker.NewDetector(logger),
 		},
 		nameExtractor: extract.NewNameExtractor(logger, nameFilter),
 		portExtractor: extract.NewPortExtractor(),

internal/detector/kafka/README.md

@@ -0,0 +1,36 @@
+# Kafka Broker Process Detector
+
+Detects and classifies Kafka Broker processes running on the system.
+
+> [!WARNING]
+> This is meant to be used as a sub-detector of the [Java Process Detector](../java) and will result in incomplete metadata
+> results if used on its own.
+
+## Overview
+
+The Kafka detector identifies Kafka broker instances by searching for the Kafka main class (`kafka.Kafka`) in command-line arguments.
+It extracts the broker configuration and identifying information from multiple sources.
+
+### Extracted Attributes
+
+The detector extracts the following attributes when available:
+- `broker.id` - Unique identifier for the Kafka broker (normalized from `node.id` in KRaft mode)
+- `cluster.id` - Unique identifier for the Kafka cluster
+
+### Configuration Sources (in priority order)
+
+1. Meta properties file in Kafka log directories
+2. Command line `--override` arguments
+3. Server properties file
+
+### Sample Metadata Result
+```json
+{
+  "categories": ["KAFKA/BROKER"],
+  "name": "Kafka Broker",
+  "attributes": {
+    "broker.id": "0",
+    "cluster.id": "WQSzAfd_RvO0TocjqhQoaA"
+  }
+}
+```