Add Kafka client detector (#1932)

Author: jefchien

cmd/workload-discovery/discovery.go

@@ -7,6 +7,7 @@ import (
 	"context"
 	"encoding/json"
 	"errors"
+	"flag"
 	"log/slog"
 	"math"
 	"os"
@@ -218,9 +219,13 @@ func convertToPid32(pid int) (int32, error) {
 }
 
 func main() {
+	var debug bool
+	flag.BoolVar(&debug, "debug", false, "enable debug logging")
+	flag.Parse()
+
 	cfg := Config{
 		Concurrency: runtime.NumCPU(),
-		LogLevel:    slog.LevelDebug,
+		LogLevel:    slog.LevelInfo,
 		Timeout:     500 * time.Millisecond,
 		FilterConfig: filter.Config{
 			Process: filter.ProcessConfig{
@@ -229,6 +234,9 @@ func main() {
 			},
 		},
 	}
+	if debug {
+		cfg.LogLevel = slog.LevelDebug
+	}
 
 	logger := buildLogger(cfg.LogLevel)
 

extension/agenthealth/metadata/metadata.go

@@ -17,6 +17,8 @@ const (
 	ValueJVM                  = "jvm"
 	ValueTomcat               = "tomcat"
 	ValueKafkaBroker          = "kafka_broker"
+	ValueKafkaConsumer        = "kafka_consumer"
+	ValueKafkaProducer        = "kafka_producer"
 	ValueNVIDIA               = "nvidia_gpu"
 
 	shortKeyObservabilitySolutions = "obs"
@@ -29,6 +31,8 @@ var (
 		Build(KeyObservabilitySolutions, ValueJVM),
 		Build(KeyObservabilitySolutions, ValueTomcat),
 		Build(KeyObservabilitySolutions, ValueKafkaBroker),
+		Build(KeyObservabilitySolutions, ValueKafkaConsumer),
+		Build(KeyObservabilitySolutions, ValueKafkaProducer),
 		Build(KeyObservabilitySolutions, ValueNVIDIA),
 	)
 	shortKeyMapping = map[string]string{

internal/detector/common/constants.go

@@ -0,0 +1,9 @@
+// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+// SPDX-License-Identifier: MIT
+
+package common
+
+const (
+	ExtJAR = ".jar"
+	ExtWAR = ".war"
+)

internal/detector/detector.go

@@ -6,6 +6,8 @@ package detector
 import (
 	"context"
 	"errors"
+
+	"github.com/shirou/gopsutil/v4/process"
 )
 
 var (
@@ -27,6 +29,8 @@ type DeviceDetector interface {
 	Detect() (*Metadata, error)
 }
 
+type OpenFilesStat = process.OpenFilesStat
+
 // Process defines an interface for interacting with system processes.
 type Process interface {
 	// PID returns the process ID.
@@ -42,4 +46,6 @@ type Process interface {
 	EnvironWithContext(ctx context.Context) ([]string, error)
 	// CreateTimeWithContext returns the creation time of the process in milliseconds.
 	CreateTimeWithContext(ctx context.Context) (int64, error)
+	// OpenFilesWithContext returns the open file descriptors of the process.
+	OpenFilesWithContext(context.Context) ([]OpenFilesStat, error)
 }

internal/detector/detectortest/mock.go

@@ -52,6 +52,14 @@ func (m *MockProcess) CreateTimeWithContext(ctx context.Context) (int64, error)
 	return int64(args.Int(0)), args.Error(1)
 }
 
+func (m *MockProcess) OpenFilesWithContext(ctx context.Context) ([]detector.OpenFilesStat, error) {
+	args := m.Called(ctx)
+	if args.Get(0) == nil {
+		return nil, args.Error(1)
+	}
+	return args.Get(0).([]detector.OpenFilesStat), args.Error(1)
+}
+
 type MockProcessDetector struct {
 	mock.Mock
 }

internal/detector/java/extract/name.go

@@ -13,14 +13,12 @@ import (
 	"strings"
 
 	"github.com/aws/amazon-cloudwatch-agent/internal/detector"
+	"github.com/aws/amazon-cloudwatch-agent/internal/detector/common"
 	"github.com/aws/amazon-cloudwatch-agent/internal/detector/util"
 	"github.com/aws/amazon-cloudwatch-agent/internal/util/collections"
 )
 
 const (
-	extJAR = ".jar"
-	extWAR = ".war"
-
 	// 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.
@@ -186,7 +184,7 @@ func newArchiveManifestNameExtractor(logger *slog.Logger) argNameExtractor {
 // Extract opens the archive and reads the manifest file. Tries to extract the name from values of specific keys in the
 // manifest. Prioritizes keys in the order defined in the extractor.
 func (e *archiveManifestNameExtractor) Extract(ctx context.Context, process detector.Process, arg string) (string, error) {
-	if !strings.HasSuffix(arg, extJAR) && !strings.HasSuffix(arg, extWAR) {
+	if !strings.HasSuffix(arg, common.ExtJAR) && !strings.HasSuffix(arg, common.ExtWAR) {
 		return "", detector.ErrIncompatibleExtractor
 	}
 	fallback := strings.TrimSuffix(filepath.Base(arg), filepath.Ext(arg))
@@ -232,11 +230,12 @@ func (e *archiveManifestNameExtractor) readManifest(jarPath string) (string, err
 
 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 {
+	_ = util.ScanProperties(r, metaManifestSeparator, func(key, value string) bool {
 		if !fieldLookup.Contains(key) || value == "" {
 			return true
 		}
 		manifest[key] = value
+		// stop scanning if the highest priority field is found.
 		return key != fieldPriority[0]
 	})
 	for _, field := range fieldPriority {

internal/detector/java/java.go

@@ -11,6 +11,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/kafkabroker"
+	"github.com/aws/amazon-cloudwatch-agent/internal/detector/kafkaclient"
 	"github.com/aws/amazon-cloudwatch-agent/internal/detector/tomcat"
 	"github.com/aws/amazon-cloudwatch-agent/internal/detector/util"
 )
@@ -36,6 +37,7 @@ func NewDetector(logger *slog.Logger, nameFilter detector.NameFilter) detector.P
 		subDetectors: []detector.ProcessDetector{
 			tomcat.NewDetector(logger),
 			kafkabroker.NewDetector(logger),
+			kafkaclient.NewDetector(logger),
 		},
 		nameExtractor: extract.NewNameExtractor(logger, nameFilter),
 		portExtractor: extract.NewPortExtractor(),

internal/detector/kafkabroker/extract/attributes.go

@@ -6,6 +6,7 @@ package extract
 import (
 	"context"
 	"errors"
+	"fmt"
 	"log/slog"
 	"os"
 	"path/filepath"
@@ -127,6 +128,7 @@ func (e *attributesExtractor) extractAttributesFromMetaProperties(ctx context.Co
 			case propertyClusterID:
 				attributes[key] = value
 			}
+			// stop scanning the properties once both attributes have been set
 			return len(attributes) < 2
 		})
 		if err != nil {
@@ -170,7 +172,7 @@ func (e *attributesExtractor) parseArgs(args []string) (*brokerInfo, error) {
 	}
 
 	if info == nil {
-		return nil, detector.ErrIncompatibleExtractor
+		return nil, fmt.Errorf("%w: Class (%s) not found in command-line", detector.ErrIncompatibleExtractor, brokerClassName)
 	}
 	return info, nil
 }
@@ -215,7 +217,7 @@ func (e *attributesExtractor) parsePropertiesFile(path string, fn func(key, valu
 	}
 	defer file.Close()
 
-	err = util.ReadProperties(file, propertiesSeparator, fn)
+	err = util.ScanProperties(file, propertiesSeparator, fn)
 	if err != nil && !errors.Is(err, util.ErrLineLimitExceeded) {
 		return err
 	}

internal/detector/kafkabroker/kafkabroker.go

@@ -35,12 +35,10 @@ func NewDetector(logger *slog.Logger) detector.ProcessDetector {
 func (d *kafkaBrokerDetector) Detect(ctx context.Context, process detector.Process) (*detector.Metadata, error) {
 	attributes, err := d.attributesExtractor.Extract(ctx, process)
 	if err != nil {
-		d.logger.Debug("Kafka Broker not detected for process", "pid", process.PID(), "error", err)
+		d.logger.Debug("Kafka broker not detected for process", "pid", process.PID(), "error", err)
 		return nil, detector.ErrIncompatibleDetector
 	}
 
-	d.logger.Debug("Detected Kafka Broker", "pid", process.PID())
-
 	md := &detector.Metadata{
 		Categories: []detector.Category{detector.CategoryKafkaBroker},
 		Name:       brokerMetadataName,

internal/detector/kafkaclient/README.md

@@ -0,0 +1,23 @@
+# Kafka Client Process Detector
+
+Detects and classifies Kafka Client processes running on the system.
+
+> [!NOTE]
+> This detector only supports detection of processes that use the Kafka clients library.
+
+> [!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 client detector identifies Java applications that use the Kafka client library (`kafka-clients`) by searching in:
+1. Command line classpath (`-cp`/`-classpath`) arguments
+2. Open file descriptors for `kafka-clients.jar`
+
+### Sample Metadata Result
+```json
+{
+  "categories": ["KAFKA/CLIENT"]
+}
+```