diff --git a/src/client/id_detection.go b/src/client/id_detection.go index c700ae12..1cbf6d26 100644 --- a/src/client/id_detection.go +++ b/src/client/id_detection.go @@ -40,8 +40,8 @@ func idsFromMBeanNames(mBeanNames []string, idExtractor idFromMBeanNameFn) []str return ids } -// idFromMBeanWithClientIdField Gets the identifier given a type=app-info MBean name. Example: "name:type=app-info,client-id=my-id" -func idFromMBeanWithClientIdField(mBeanName string) string { +// idFromMBeanWithClientIDField Gets the identifier given a type=app-info MBean name. Example: "name:type=app-info,client-id=my-id" +func idFromMBeanWithClientIDField(mBeanName string) string { _, info, valid := strings.Cut(mBeanName, ":") if !valid { return "" diff --git a/src/client/id_detection_test.go b/src/client/id_detection_test.go index 631c2f33..6c19f745 100644 --- a/src/client/id_detection_test.go +++ b/src/client/id_detection_test.go @@ -39,7 +39,7 @@ func TestIdFromAppInfoMBean(t *testing.T) { for _, c := range cases { t.Run(c.MBeanName, func(t *testing.T) { - assert.Equal(t, c.Expected, idFromMBeanWithClientIdField(c.MBeanName)) + assert.Equal(t, c.Expected, idFromMBeanWithClientIDField(c.MBeanName)) }) } } diff --git a/src/client/producer_consumer_collection.go b/src/client/producer_consumer_collection.go index 966cf5cb..466ce704 100644 --- a/src/client/producer_consumer_collection.go +++ b/src/client/producer_consumer_collection.go @@ -70,7 +70,7 @@ func CollectConsumerMetrics(i *integration.Integration, jmxInfo *args.JMXHost, j } defer conn.Close() // Get client identifiers for all the consumers - clientIDs, err := getClientIDS(jmxInfo, consumerDetectionPattern, idFromMBeanWithClientIdField, conn) + clientIDs, err := getClientIDS(jmxInfo, consumerDetectionPattern, idFromMBeanWithClientIDField, conn) if err != nil { log.Error("Unable to detect consumer/producers for '%s:%s': %s", jmxInfo.Host, jmxInfo.Port, err) return @@ -115,7 +115,7 @@ func CollectProducerMetrics(i *integration.Integration, jmxInfo *args.JMXHost, j } defer conn.Close() // Get client identifiers for all the producers - clientIDs, err := getClientIDS(jmxInfo, producerDetectionPattern, idFromMBeanWithClientIdField, conn) + clientIDs, err := getClientIDS(jmxInfo, producerDetectionPattern, idFromMBeanWithClientIDField, conn) if err != nil { log.Error("Unable to detect producers for '%s:%s': %s", jmxInfo.Host, jmxInfo.Port, err) return diff --git a/src/client/producer_consumer_collection_test.go b/src/client/producer_consumer_collection_test.go index 4bd992f2..b9d69b93 100644 --- a/src/client/producer_consumer_collection_test.go +++ b/src/client/producer_consumer_collection_test.go @@ -184,72 +184,70 @@ func TestProducerWorker_JmxOpenFuncErr(t *testing.T) { wg.Wait() } -func TestCollectConsumerMetricsNameProvided(t *testing.T) { - i, err := integration.New(t.Name(), "1.0.0") - require.NoError(t, err) - connProvider := mocks.NewEmptyMockJMXProvider() - connProvider.Names = []string{ - "kafka.consumer:type=consumer-fetch-manager-metrics,client-id=consumer-1", - "kafka.consumer:type=consumer-fetch-manager-metrics,client-id=consumer-2", - } - clientID := "consumer-1" - jmxInfo := &args.JMXHost{Name: clientID} - testutils.SetupTestArgs() - CollectConsumerMetrics(i, jmxInfo, connProvider) - require.Len(t, i.Entities, 1, "only metrics from the provided consumer should be fetched") - assert.Equal(t, clientID, i.Entities[0].Metadata.Name) -} - -func TestCollectProducerMetricsCreateEntity(t *testing.T) { - i, err := integration.New(t.Name(), "1.0.0") - require.NoError(t, err) - connProvider := mocks.NewEmptyMockJMXProvider() - connProvider.Names = []string{ - "kafka.producer:type=producer-metrics,client-id=producer-1", - "kafka.producer:type=producer-metrics,client-id=producer-2", - } - clientID := "producer-1" - jmxInfo := &args.JMXHost{Name: clientID} - testutils.SetupTestArgs() - CollectProducerMetrics(i, jmxInfo, connProvider) - require.Len(t, i.Entities, 1, "only metrics from the provided producer should be fetched") - assert.Equal(t, clientID, i.Entities[0].Metadata.Name) -} - -func TestCollectConsumerMetricsNoNameProvided(t *testing.T) { - i, err := integration.New(t.Name(), "1.0.0") - require.NoError(t, err) - connProvider := mocks.NewEmptyMockJMXProvider() - connProvider.Names = []string{ - "kafka.consumer:type=consumer-fetch-manager-metrics,client-id=consumer-1", - "kafka.consumer:type=consumer-fetch-manager-metrics,client-id=consumer-2", - } - jmxInfo := &args.JMXHost{} - testutils.SetupTestArgs() - CollectConsumerMetrics(i, jmxInfo, connProvider) - require.Len(t, i.Entities, 2) - var entityNames []string - for _, entity := range i.Entities { - entityNames = append(entityNames, entity.Metadata.Name) - } - assert.ElementsMatch(t, []string{"consumer-1", "consumer-2"}, entityNames) -} - -func TestCollectProducerMetricsNoNameProvided(t *testing.T) { - i, err := integration.New(t.Name(), "1.0.0") - require.NoError(t, err) - connProvider := mocks.NewEmptyMockJMXProvider() - connProvider.Names = []string{ - "kafka.producer:type=producer-metrics,client-id=producer-1", - "kafka.producer:type=producer-metrics,client-id=producer-2", - } - jmxInfo := &args.JMXHost{} - testutils.SetupTestArgs() - CollectProducerMetrics(i, jmxInfo, connProvider) - require.Len(t, i.Entities, 2) - var entityNames []string - for _, entity := range i.Entities { - entityNames = append(entityNames, entity.Metadata.Name) +func TestProducerConsumerEntitiesCreation(t *testing.T) { + cases := []struct { + Name string + JMXNames []string + JMXInfo *args.JMXHost + CollectionFn CollectionFn + ExpectedEntityNames []string + }{ + { + Name: "Collect consumer metrics with name provided", + JMXNames: []string{ + "kafka.consumer:type=consumer-fetch-manager-metrics,client-id=consumer-1", + "kafka.consumer:type=consumer-fetch-manager-metrics,client-id=consumer-2", + }, + JMXInfo: &args.JMXHost{Name: "consumer-1"}, + CollectionFn: CollectConsumerMetrics, + ExpectedEntityNames: []string{"consumer-1"}, + }, + { + Name: "Collect consumer metrics with no name provided", + JMXNames: []string{ + "kafka.consumer:type=consumer-fetch-manager-metrics,client-id=consumer-1", + "kafka.consumer:type=consumer-fetch-manager-metrics,client-id=consumer-2", + }, + JMXInfo: &args.JMXHost{}, + CollectionFn: CollectConsumerMetrics, + ExpectedEntityNames: []string{"consumer-1", "consumer-2"}, + }, + { + Name: "Collect producer metrics with name provided", + JMXNames: []string{ + "kafka.producer:type=producer-metrics,client-id=producer-1", + "kafka.producer:type=producer-metrics,client-id=producer-2", + }, + JMXInfo: &args.JMXHost{Name: "producer-1"}, + CollectionFn: CollectProducerMetrics, + ExpectedEntityNames: []string{"producer-1"}, + }, + { + Name: "Collect producer metrics with no name provided", + JMXNames: []string{ + "kafka.producer:type=producer-metrics,client-id=producer-1", + "kafka.producer:type=producer-metrics,client-id=producer-2", + }, + JMXInfo: &args.JMXHost{}, + CollectionFn: CollectProducerMetrics, + ExpectedEntityNames: []string{"producer-1", "producer-2"}, + }, + } + for _, c := range cases { + t.Run(c.Name, func(t *testing.T) { + // setup integration + i, err := integration.New(c.Name, "1.0.0") + require.NoError(t, err) + testutils.SetupTestArgs() + connProvider := mocks.NewEmptyMockJMXProvider() + connProvider.Names = c.JMXNames + // run collection + c.CollectionFn(i, c.JMXInfo, connProvider) + var entityNames []string + for _, entity := range i.Entities { + entityNames = append(entityNames, entity.Metadata.Name) + } + assert.ElementsMatch(t, c.ExpectedEntityNames, entityNames) + }) } - assert.ElementsMatch(t, []string{"producer-1", "producer-2"}, entityNames) } diff --git a/tests/integration/kafka_test.go b/tests/integration/kafka_test.go index 31f48d6b..ebee1dfe 100644 --- a/tests/integration/kafka_test.go +++ b/tests/integration/kafka_test.go @@ -367,7 +367,7 @@ func TestKafkaIntegration_consumer_test(t *testing.T) { consumerConfig := func(command []string) []string { return append( command, - "--consumers", "[{\"name\": \"kafka_dummy_consumer\", \"host\": \"kafka_dummy_consumer\", \"port\": 1087},{\"name\": \"kafka_dummy_consumer2\", \"host\": \"kafka_dummy_consumer2\", \"port\": 1088}]", + "--consumers", "[{\"host\": \"kafka_dummy_consumer\", \"port\": 1087},{\"name\": \"kafka_dummy_consumer2\", \"host\": \"kafka_dummy_consumer2\", \"port\": 1088}]", ) }