prometheus.exporter.kafka: fix for config option is ignored

CHANGELOG.md

@@ -22,6 +22,8 @@ Main (unreleased)
 - Fix an issue in `remote.s3` where the exported content of an object would be an empty string if `remote.s3` failed to fully retrieve
   the file in a single read call. (@grafana/agent-squad)
 
+- Utilize the `instance` Argument of `prometheus.exporter.kafka` when set. (@akhmatov-s)
+
 ### Other changes
 
 - Removed support for Windows 2012 in line with Microsoft end of life. (@mattdurham)

component/prometheus/exporter/kafka/kafka.go

@@ -90,6 +90,7 @@ func createExporter(opts component.Options, args component.Arguments, defaultIns
 
 func (a *Arguments) Convert() *kafka_exporter.Config {
 	return &kafka_exporter.Config{
+		Instance:                a.Instance,
 		KafkaURIs:               a.KafkaURIs,
 		UseSASL:                 a.UseSASL,
 		UseSASLHandshake:        a.UseSASLHandshake,

component/prometheus/exporter/kafka/kafka_test.go

@@ -83,6 +83,7 @@ func TestRiverConvert(t *testing.T) {
 	}
 	converted := orig.Convert()
 	expected := kafka_exporter.Config{
+		Instance:                "example",
 		KafkaURIs:               []string{"localhost:9092", "localhost:19092"},
 		KafkaVersion:            "2.0.0",
 		MetadataRefreshInterval: "1m",

pkg/integrations/kafka_exporter/kafka_exporter.go

@@ -28,6 +28,9 @@ var DefaultConfig = Config{
 
 // Config controls kafka_exporter
 type Config struct {
+	// The instance label for metrics.
+	Instance string `yaml:"instance,omitempty"`
+
 	// Address array (host:port) of Kafka server
 	KafkaURIs []string `yaml:"kafka_uris,omitempty"`
 
@@ -109,11 +112,14 @@ func (c *Config) Name() string {
 // there is not exactly one Kafka node, the user must manually provide
 // their own value for instance key in the common config.
 func (c *Config) InstanceKey(agentKey string) (string, error) {
-	if len(c.KafkaURIs) != 1 {
+	if len(c.KafkaURIs) == 1 {
+		return c.KafkaURIs[0], nil
+	}
+	if c.Instance == "" && len(c.KafkaURIs) > 1 {
 		return "", fmt.Errorf("an automatic value for `instance` cannot be determined from %d kafka servers, manually provide one for this integration", len(c.KafkaURIs))
 	}
 
-	return c.KafkaURIs[0], nil
+	return c.Instance, nil
 }
 
 // NewIntegration creates a new elasticsearch_exporter