Merge pull request #1 from lukasrynt/kprom/dynamic-client-id-label

kprom: Multiple clients

pkg/kgo/broker.go

@@ -477,6 +477,7 @@ func (cxn *brokerCxn) hookWriteE2E(key int16, bytesWritten int, writeWait, timeT
 				WriteWait:    writeWait,
 				TimeToWrite:  timeToWrite,
 				WriteErr:     writeErr,
+				ClientID:     cxn.cl.clientIDString(),
 			})
 		}
 	})
@@ -1232,6 +1233,7 @@ func (cxn *brokerCxn) readResponse(
 				ReadWait:     readWait,
 				TimeToRead:   timeToRead,
 				ReadErr:      readErr,
+				ClientID:     cxn.cl.clientIDString(),
 			})
 		}
 	})

pkg/kgo/hooks.go

@@ -134,6 +134,9 @@ type BrokerE2E struct {
 	WriteErr error
 	// ReadErr is any error encountered during reading.
 	ReadErr error
+
+	// ClientID is pointer to ID of the client that made the request
+	ClientID string
 }
 
 // DurationE2E returns the e2e time from the start of when a request is written
@@ -239,6 +242,9 @@ type ProduceBatchMetrics struct {
 	// 0 is no compression, 1 is gzip, 2 is snappy, 3 is lz4, and 4 is
 	// zstd.
 	CompressionType uint8
+
+	// ClientID is pointer to ID of the client that made the request
+	ClientID string
 }
 
 // HookProduceBatchWritten is called whenever a batch is known to be
@@ -288,6 +294,9 @@ type FetchBatchMetrics struct {
 	// 0 is no compression, 1 is gzip, 2 is snappy, 3 is lz4, and 4 is
 	// zstd.
 	CompressionType uint8
+
+	// ClientID is pointer to ID of the client that made the request
+	ClientID string
 }
 
 // HookFetchBatchRead is called whenever a batch if read within the client.
@@ -418,3 +427,11 @@ func implementsAnyHook(h Hook) bool {
 	}
 	return false
 }
+
+func (cl *Client) clientIDString() string {
+	resolved := "kgo"
+	if cl.cfg.id != nil {
+		resolved = *cl.cfg.id
+	}
+	return resolved
+}

pkg/kgo/sink.go

@@ -1738,6 +1738,7 @@ func (p produceMetrics) hook(cfg *cfg, br *broker) {
 		for _, h := range hooks {
 			for topic, partitions := range p {
 				for partition, metrics := range partitions {
+					metrics.ClientID = br.cl.clientIDString()
 					h.OnProduceBatchWritten(br.meta, topic, partition, metrics)
 				}
 			}

pkg/kgo/source.go

@@ -1362,6 +1362,7 @@ func (o *cursorOffsetNext) processRespPartition(br *broker, rp *kmsg.FetchRespon
 		in = in[length:]
 
 		var m FetchBatchMetrics
+		m.ClientID = br.cl.clientIDString()
 
 		switch t := r.(type) {
 		case *kmsg.MessageV0:

plugin/kprom/config.go

@@ -96,9 +96,12 @@ func HandlerOpts(opts promhttp.HandlerOpts) Opt {
 	return opt{func(c *cfg) { c.handlerOpts = opts }}
 }
 
-// WithClientLabel adds a "cliend_id" label to all metrics.
+// WithClientLabel adds a "client_id" label to all metrics.
 func WithClientLabel() Opt {
-	return opt{func(c *cfg) { c.withClientLabel = true }}
+	return opt{func(c *cfg) {
+		c.withClientLabel = true
+		c.fetchProduceOpts.labels = append(c.fetchProduceOpts.labels, "client_id")
+	}}
 }
 
 // Subsystem sets the subsystem for the kprom metrics, overriding the default
@@ -182,6 +185,7 @@ type Detail uint8
 const (
 	ByNode            Detail = iota // Include label "node_id" for fetch and produce metrics.
 	ByTopic                         // Include label "topic" for fetch and produce metrics.
+	ByClient                        // Include label "client_id" for fetch and produce metrics
 	Batches                         // Report number of fetched and produced batches.
 	Records                         // Report the number of fetched and produced records.
 	CompressedBytes                 // Report the number of fetched and produced compressed bytes.
@@ -211,6 +215,8 @@ func FetchAndProduceDetail(details ...Detail) Opt {
 					labelsDeduped[ByTopic] = "topic"
 				case ByNode:
 					labelsDeduped[ByNode] = "node_id"
+				case ByClient:
+					labelsDeduped[ByClient] = "client_id"
 				case Batches:
 					c.fetchProduceOpts.batches = true
 				case Records: