package me.dmexe.telemetry.kafka;

import static me.dmexe.telemetry.kafka.GaugesFactory.gauge;

import io.prometheus.client.Collector;
import java.util.ArrayList;
import java.util.LinkedList;
import java.util.List;
import java.util.Objects;

public class KafkaConsumerMetricsCollector extends Collector {
  private final KafkaMetricsProducer producer;

  public KafkaConsumerMetricsCollector(KafkaMetricsProducer producer) {
    Objects.requireNonNull(producer, "metrics producer cannot be null");
    this.producer = producer;
  }

  @Override
  public List<MetricFamilySamples> collect() {
    final List<MetricFamilySamples> samples = new LinkedList<>();

    producer.metrics().forEach((name, metric) -> {
      switch (name.group()) {
        case "consumer-metrics": {
          final String clientId = name.tags().get("client-id");
          if (clientId != null) {
            final List<String> labelValues = new ArrayList<>(1);
            labelValues.add(clientId);
            samples.add(gauge(name, "client_id").addMetric(labelValues, metric.value()));
          }
          break;
        }
        case "consumer-coordinator-metrics": {
          final String clientId = name.tags().get("client-id");
          if (clientId != null) {
            final List<String> labelValues = new ArrayList<>(1);
            labelValues.add(clientId);
            samples.add(gauge(name, "client_id").addMetric(labelValues, metric.value()));
          }
          break;
        }
        case "consumer-fetch-manager-metrics": {
          final String clientId = name.tags().get("client-id");
          if (clientId != null) {
            final List<String> labelValues = new ArrayList<>(1);
            labelValues.add(clientId);
            samples.add(gauge(name, "client_id").addMetric(labelValues, metric.value()));
          }
          break;
        }
        default:
          break;
      }
    });

    return samples;
  }
}

@@ -986,32 +986,49 @@
     samples.add(
+%0A                
 gauge(name, %22cli
@@ -1027,32 +1027,53 @@
 me, %22client_id%22)
+%0A                    
 .addMetric(label
@@ -1402,32 +1402,49 @@
     samples.add(
+%0A                
 gauge(name, %22cli
@@ -1443,32 +1443,53 @@
 me, %22client_id%22)
+%0A                    
 .addMetric(label
@@ -1688,36 +1688,470 @@
 clientId != null
-) %7B%0A
+ && !name.name().contains(%22.%22)) %7B%0A            final String topic = name.tags().get(%22topic%22);%0A            if (topic != null) %7B%0A              final List%3CString%3E labelValues = new ArrayList%3C%3E(2);%0A              labelValues.add(clientId);%0A              labelValues.add(topic);%0A              samples%0A                  .add(gauge(name, %22client_id%22, %22topic%22)%0A                      .addMetric(labelValues, metric.value()));%0A            %7D else %7B%0A  
             fina
@@ -2191,32 +2191,34 @@
 ArrayList%3C%3E(1);%0A
+  
             labe
@@ -2244,32 +2244,34 @@
 d);%0A            
+  
 samples.add(gaug
@@ -2262,24 +2262,43 @@
 samples.add(
+%0A                  
 gauge(name, 
@@ -2309,16 +2309,39 @@
 ent_id%22)
+%0A                      
 .addMetr
@@ -2366,32 +2366,46 @@
 tric.value()));%0A
+            %7D%0A
           %7D%0A    

fix consumer tags