Kafka cluster changes

CHANGELOG.md

@@ -7,6 +7,13 @@ and this project adheres to [Semantic Versioning](http://semver.org/).
 
 ## Unreleased
 
+### 🚀 Enhancements
+- Added cluster-level metrics from kafka.controller, kafka.coordinator.group, and kafka.server domains
+- Added a new event type `KafkaClusterSample` with metrics reflecting the health of the entire Kafka cluster
+- Added a new command line option `collect_cluster_metrics` to enable collection of cluster-wide metrics
+- Enhanced cluster metrics collection to intelligently identify and use the controller broker
+- Added fallback mechanism to ensure cluster metrics are collected even if controller is unavailable
+
 ## v3.13.2 - 2025-07-24
 
 ### 🐞 Bug fixes

README.md

@@ -2,7 +2,7 @@
 
 # New Relic integration for Kafka
 
-The New Relic integration for Kafka captures critical performance metrics and inventory reported by Kafka clusters. Data on Brokers, Topics, Java Producers, and Java Consumers is collected.
+The New Relic integration for Kafka captures critical performance metrics and inventory reported by Kafka clusters. Data on Brokers, Topics, Java Producers, Java Consumers, and cluster-wide metrics is collected.
 
 Inventory data is obtained mainly from Zookeeper nodes and metrics are collected through JMX.
 
@@ -119,7 +119,41 @@ If you believe you have found a security vulnerability in this project or any of
 
 If you would like to contribute to this project, review [these guidelines](./CONTRIBUTING.md).
 
-To all contributors, we thank you!  Without your contribution, this project would not be what it is today.
+## Event Types
+
+The integration provides multiple event types that each capture a specific aspect of Kafka monitoring:
+
+| Event Type | Description |
+| --- | --- |
+| KafkaBrokerSample | Metrics specific to individual broker performance such as network I/O, request handling, and more. |
+| KafkaTopicSample | Metrics related to individual topics, including partition count and replication status. |
+| KafkaProducerSample | Performance metrics for producers connected to the Kafka cluster. |
+| KafkaConsumerSample | Performance metrics for consumers subscribed to the Kafka cluster. |
+| KafkaOffsetSample | Metrics related to consumer group offsets, providing lag information. |
+| KafkaClusterSample | Cluster-wide metrics that provide a holistic view of the entire cluster, including controller status, broker counts, and partition health. |
+
+Each event type provides a specific perspective on your Kafka ecosystem, allowing you to monitor different components with appropriate granularity.
+
+## Cluster Metrics
+
+The integration can collect cluster-wide metrics from the Kafka controller. These metrics provide insight into the overall health and performance of your Kafka cluster. To enable cluster metrics collection, use the following configuration:
+
+```yaml
+collect_cluster_metrics: true
+```
+
+When this setting is enabled, the integration will:
+1. Automatically identify which broker is currently the controller
+2. Collect cluster-wide metrics from the controller broker
+3. Fall back to another available broker if the controller cannot be identified or accessed
+
+The cluster metrics are collected from the following MBeans:
+- `kafka.controller:type=KafkaController,name=*` - Core cluster metrics such as active broker count, offline partitions, and more
+- `kafka.controller:type=ControllerStats,name=UncleanLeaderElectionsPerSec` - Rate of unclean leader elections
+- `kafka.coordinator.group:type=GroupMetadataManager,name=*` - Consumer group metrics
+- `kafka.server:type=KafkaServer,name=ClusterId` - Cluster identification
+
+These metrics are reported in a `KafkaClusterSample` event type, allowing you to monitor your entire Kafka cluster's health from a single entity.
 
 ## License
 

kafka-config.yml.k8s_sample

@@ -84,3 +84,8 @@ newrelic-infrastructure:
             TOPIC_MODE: list
             TOPIC_LIST: '["test"]'
             METRICS: "true"
+
+            # Enable collection of cluster-wide metrics from the Kafka controller
+            # These metrics provide a holistic view of the cluster state, including information about brokers, 
+            # controller, partitions, and replication status.
+            COLLECT_CLUSTER_METRICS: false

kafka-config.yml.sample

@@ -33,6 +33,11 @@ integrations:
     # or Kubernetes, is used on top of the integration.
     LOCAL_ONLY_COLLECTION: false
 
+    # Enable collection of cluster-wide metrics from the Kafka controller
+    # These metrics provide a holistic view of the cluster state, including information about brokers, 
+    # controller, partitions, and replication status.
+    COLLECT_CLUSTER_METRICS: false
+
     # Below are the fields used to fine tune/toggle topic metric collection. Can be one of:
     # - "all", where all topics found in all brokers will be collected.
     # - "none", which will disable topic collection alltogether.

kafka-win-config.yml.sample

@@ -137,6 +137,12 @@ integrations:
     # for any other broker, nor will it collect cluster-level metrics like topic metrics. This is useful for things
     # like deployment to kubernetes, where a single integration instance is desired per broker.
     LOCAL_ONLY_COLLECTION: false
+
+    # Enable collection of cluster-wide metrics from the Kafka controller
+    # These metrics provide a holistic view of the cluster state, including information about brokers, 
+    # controller, partitions, and replication status.
+    COLLECT_CLUSTER_METRICS: false
+
     TOPIC_MODE: "all"
     COLLECT_TOPIC_SIZE: false
   interval: 15s

src/args/args.go

@@ -76,6 +76,7 @@ type ArgumentList struct {
 	TopicBucket                string `default:"1/1" help:"Allows the partitioning of topic collection across multiple instances. The second number is the number of instances topics are partitioned across. The first number is the bucket number of the current instance, which should be between 1 and the second number."`
 	CollectTopicSize           bool   `default:"false" help:"Enablement of on disk Topic size metric collection. This metric can be very resource intensive to collect especially against many topics."`
 	CollectTopicOffset         bool   `default:"false" help:"Enablement of Topic offsets collection. This metric can be very resource intensive to collect especially against many topics."`
+	CollectClusterMetrics      bool   `default:"false" help:"Collect cluster-wide metrics from the Kafka controller."`
 
 	// Consumer offset arguments
 	ConsumerOffset              bool   `default:"false" help:"Populate consumer offset data"`
@@ -88,4 +89,6 @@ type ArgumentList struct {
 	ShowVersion bool `default:"false" help:"Print build information and exit"`
 
 	TopicSource string `default:"broker" help:"Collect topics list from either the Broker or Zookeeper"`
+
+	EnableBrokerTopicMetricsV2 bool `default:"false" help:"Enable the new BrokerTopicMetrics metrics. This is a new set of metrics that are essentials for some capabilities to work. "`
 }

src/args/parsed_args.go

@@ -90,6 +90,7 @@ type ParsedArguments struct {
 	TopicBucket                TopicBucket
 	CollectTopicSize           bool
 	CollectTopicOffset         bool
+	EnableBrokerTopicMetricsV2 bool
 
 	// Consumer offset arguments
 	ConsumerOffset              bool
@@ -252,6 +253,8 @@ func ParseArgs(a ArgumentList) (*ParsedArguments, error) {
 		return nil, fmt.Errorf("failed to parse kafka version: %s", err)
 	}
 
+	log.Info("Processing new BrokerTopic metrics flag is : %v", a.EnableBrokerTopicMetricsV2)
+
 	parsedArgs := &ParsedArguments{
 		DefaultArgumentList:              a.DefaultArgumentList,
 		AutodiscoverStrategy:             a.AutodiscoverStrategy,
@@ -283,6 +286,7 @@ func ParseArgs(a ArgumentList) (*ParsedArguments, error) {
 		TrustStore:                       a.TrustStore,
 		TrustStorePassword:               a.TrustStorePassword,
 		LocalOnlyCollection:              a.LocalOnlyCollection,
+		CollectClusterMetrics:            a.CollectClusterMetrics,
 		ForceTopicSampleCollection:       a.ForceTopicSampleCollection,
 		CollectTopicSize:                 a.CollectTopicSize,
 		CollectTopicOffset:               a.CollectTopicOffset,
@@ -300,6 +304,7 @@ func ParseArgs(a ArgumentList) (*ParsedArguments, error) {
 		SaslGssapiKerberosConfigPath:     a.SaslGssapiKerberosConfigPath,
 		SaslGssapiDisableFASTNegotiation: a.SaslGssapiDisableFASTNegotiation,
 		TopicSource:                      a.TopicSource,
+		EnableBrokerTopicMetricsV2: 	  a.EnableBrokerTopicMetricsV2,
 	}
 
 	return parsedArgs, nil

src/broker/broker_collection.go

@@ -182,7 +182,7 @@ func collectBrokerTopicMetrics(b *connection.Broker, collectedTopics []string, i
 		// Insert into map
 		topicSampleLookup[topicName] = sample
 
-		metrics.CollectMetricDefinitions(sample, metrics.BrokerTopicMetricDefs, metrics.ApplyTopicName(topicName), conn)
+		metrics.CollectMetricDefinitions(sample, metrics.GetFinalMetricSets(metrics.BrokerTopicMetricDefs, metrics.BrokerTopicV2MetricDefs), metrics.ApplyTopicName(topicName), conn)
 	}
 
 	return topicSampleLookup

src/cluster/collector.go

@@ -0,0 +1,101 @@
+// Package cluster handles collection of Kafka cluster-level metrics
+package cluster
+
+import (
+	"fmt"
+	"strings"
+
+	"github.com/newrelic/infra-integrations-sdk/v3/data/attribute"
+	"github.com/newrelic/infra-integrations-sdk/v3/integration"
+	"github.com/newrelic/nri-kafka/src/args"
+	"github.com/newrelic/nri-kafka/src/connection"
+	"github.com/newrelic/nri-kafka/src/metrics"
+)
+
+const (
+	// ClusterName is the entity name for the Kafka cluster
+	ClusterName = "ka-cluster"
+
+	// ClusterEventType is the event type for cluster metrics
+	ClusterEventType = "KafkaClusterSample"
+)
+
+// Collector collects metrics at the cluster level
+// Ideally, this collector should be initialized with a connection to the controller broker,
+// but it will work with any broker that has JMX enabled.
+type Collector struct {
+	jmxClient connection.JMXConnection
+	hostPort  string // Format: host:port to identify the broker used for metrics collection
+}
+
+// NewCollector creates a new collector for cluster metrics
+func NewCollector(jmxClient connection.JMXConnection, hostPort string) *Collector {
+	return &Collector{
+		jmxClient: jmxClient,
+		hostPort:  hostPort,
+	}
+}
+
+// CollectMetrics collects metrics from the Kafka controller
+func (c *Collector) CollectMetrics(integration *integration.Integration) error {
+	// Create entity for the cluster
+	clusterEntity, err := c.Entity(integration)
+	if err != nil {
+		return fmt.Errorf("failed to create cluster entity: %v", err)
+	}
+
+	// Collect metrics only if metrics collection is enabled
+	if args.GlobalArgs.HasMetrics() {
+		// Collect cluster metrics
+		populateClusterMetrics(clusterEntity, c.hostPort, c.jmxClient)
+	}
+
+	return nil
+}
+
+// Entity gets the entity object for the cluster
+func (c *Collector) Entity(i *integration.Integration) (*integration.Entity, error) {
+	host := c.hostPort
+	if host == "" {
+		host = "unknown:0"
+	}
+
+	// Get hostname and port from hostPort
+	hostParts := strings.Split(host, ":")
+	hostname := hostParts[0]
+	port := "0"
+	if len(hostParts) > 1 {
+		port = hostParts[1]
+	}
+
+	// For broker entities, the entityName is just the host:port
+	// and the namespace is "ka-broker". Let's follow the same pattern.
+	entityName := fmt.Sprintf("%s:%s", hostname, port)
+
+	// Get cluster name from args if available
+	clusterName := ""
+	if args.GlobalArgs != nil {
+		clusterName = args.GlobalArgs.ClusterName
+	}
+
+	// Follow the broker entity pattern: only use clusterName as an ID attribute
+	clusterIDAttr := integration.NewIDAttribute("clusterName", clusterName)
+
+	// Don't include host and port attributes in the entity key
+	// as they are already part of the entityName
+	return i.Entity(entityName, ClusterName, clusterIDAttr)
+}
+
+// populateClusterMetrics collects all cluster metrics and adds them to the entity
+func populateClusterMetrics(entity *integration.Entity, hostPort string, conn connection.JMXConnection) {
+	// Create a sample metric set for the cluster
+	sample := entity.NewMetricSet(ClusterEventType,
+		attribute.Attribute{Key: "displayName", Value: hostPort},
+		attribute.Attribute{Key: "entityName", Value: "cluster:" + hostPort},
+		attribute.Attribute{Key: "clusterName", Value: args.GlobalArgs.ClusterName},
+		attribute.Attribute{Key: "event_type", Value: ClusterEventType},
+	)
+
+	// Collect all cluster metrics
+	metrics.CollectMetricDefinitions(sample, metrics.ClusterMetricDefs, nil, conn)
+}

src/cluster/collector_test.go

@@ -0,0 +1,55 @@
+package cluster
+
+import (
+	"testing"
+
+	"github.com/newrelic/infra-integrations-sdk/v3/data/attribute"
+	"github.com/newrelic/infra-integrations-sdk/v3/integration"
+	"github.com/newrelic/nri-kafka/src/args"
+	"github.com/newrelic/nri-kafka/src/connection/mocks"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func TestCollector_CollectMetrics(t *testing.T) {
+	// Set up mock arguments
+	args.GlobalArgs = &args.ParsedArguments{
+		ClusterName: "test-cluster",
+	}
+
+	// Create integration
+	i, err := integration.New("test", "1.0.0")
+	require.NoError(t, err)
+
+	// Create a collector with test parameters
+	hostPort := "localhost:9999"
+
+	// Create a mock JMX client
+	mockJMX := mocks.NewEmptyMockJMXProvider()
+
+	// Create collector with mock JMX client
+	collector := NewCollector(mockJMX, hostPort)
+
+	// Create the entity using the collector's Entity method
+	entity, err := collector.Entity(i)
+	require.NoError(t, err)
+
+	// Verify entity was created with correct metadata
+	assert.Equal(t, ClusterName, entity.Metadata.Namespace)
+	assert.Equal(t, hostPort, entity.Metadata.Name)
+	assert.Equal(t, 1, len(i.Entities))
+
+	// Create a metric set to simulate metrics collection
+	ms := entity.NewMetricSet(ClusterEventType,
+		attribute.Attribute{Key: "displayName", Value: hostPort},
+		attribute.Attribute{Key: "entityName", Value: "cluster:" + hostPort},
+		attribute.Attribute{Key: "clusterName", Value: args.GlobalArgs.ClusterName},
+		attribute.Attribute{Key: "event_type", Value: ClusterEventType},
+	)
+
+	// Verify the metric set was created
+	assert.NotNil(t, ms)
+	assert.Equal(t, 1, len(entity.Metrics))
+
+	t.Log("Test completed successfully")
+}

src/connection/controller.go

@@ -0,0 +1,47 @@
+// Package connection implements connection code
+package connection
+
+import (
+	"fmt"
+
+	"github.com/newrelic/infra-integrations-sdk/v3/log"
+)
+
+// FindControllerBroker identifies and returns the controller broker from a list of brokers
+// If the controller cannot be found, it returns nil
+func FindControllerBroker(brokers []*Broker) *Broker {
+	if len(brokers) == 0 {
+		return nil
+	}
+
+	// Create a client from the broker list to make API calls
+	client, err := NewSaramaClientFromBrokerList(brokers)
+	if err != nil {
+		log.Error("Failed to create client to find controller: %s", err)
+		return nil
+	}
+	defer client.Close()
+
+	// Get the controller broker from the client
+	controllerBroker, err := client.Controller()
+	if err != nil {
+		log.Error("Failed to get controller broker: %s", err)
+		return nil
+	}
+
+	// Get the broker ID as a string
+	controllerIDStr := fmt.Sprintf("%d", controllerBroker.ID())
+	log.Debug("Found controller broker with ID: %s", controllerIDStr)
+
+	// Find the broker in our list that matches the controller ID
+	for _, broker := range brokers {
+		// The broker ID in our struct is a string, compare with the string version of controller ID
+		if broker.ID == controllerIDStr {
+			log.Debug("Found controller broker: %s (ID: %s)", broker.Host, broker.ID)
+			return broker
+		}
+	}
+
+	log.Debug("Controller broker with ID %s not found in provided broker list", controllerIDStr)
+	return nil
+}

src/connection/controller_test.go

@@ -0,0 +1,23 @@
+package connection
+
+import (
+	"testing"
+
+	"github.com/newrelic/infra-integrations-sdk/v3/log"
+	"github.com/stretchr/testify/assert"
+)
+
+// Since we can't easily override NewSaramaClientFromBrokerList for testing,
+// we'll focus on testing that an empty broker list returns nil
+func TestFindControllerBrokerWithEmptyList(t *testing.T) {
+	// Setup logging
+	log.SetupLogging(false)
+
+	// Test case: Empty broker list
+	emptyBrokers := []*Broker{}
+	controllerBroker := FindControllerBroker(emptyBrokers)
+	assert.Nil(t, controllerBroker, "Should return nil for empty broker list")
+}
+
+// In a real environment, we'd use dependency injection for testability
+// For now, we'll just test the empty case which doesn't require mocking