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 && !name.name().contains(".")) {
            final String topic = name.tags().get("topic");
            if (topic != null) {
              final List<String> labelValues = new ArrayList<>(2);
              labelValues.add(clientId);
              labelValues.add(topic);
              samples
                  .add(gauge(name, "client_id", "topic")
                      .addMetric(labelValues, metric.value()));
            } else {
              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;
  }
}

@@ -703,16 +703,118 @@
 c) -%3E %7B%0A
+      // skip, per topic metrics%0A      if (name.tags().size() == 1 && !name.name().contains(%22.%22)) %7B%0A  
       sw
@@ -831,24 +831,26 @@
 .group()) %7B%0A
+  
         case
@@ -874,32 +874,34 @@
 s%22: %7B%0A          
+ 
+ 
 final String cli
@@ -940,32 +940,34 @@
 id%22);%0A          
+  
 if (clientId != 
@@ -966,32 +966,34 @@
 ntId != null) %7B%0A
+  
             fina
@@ -1045,32 +1045,34 @@
 1);%0A            
+  
 labelValues.add(
@@ -1086,32 +1086,34 @@
 d);%0A            
+  
 samples.add(%0A   
@@ -1108,33 +1108,16 @@
 les.add(
-%0A                
 gauge(na
@@ -1132,37 +1132,16 @@
 ent_id%22)
-%0A                    
 .addMetr
@@ -1166,32 +1166,34 @@
 tric.value()));%0A
+  
           %7D%0A    
@@ -1190,32 +1190,34 @@
     %7D%0A          
+  
 break;%0A        %7D
@@ -1207,34 +1207,38 @@
  break;%0A        
-%7D%0A
+  %7D%0A  
         case %22co
@@ -1270,32 +1270,34 @@
 s%22: %7B%0A          
+  
 final String cli
@@ -1336,32 +1336,34 @@
 id%22);%0A          
+  
 if (clientId != 
@@ -1374,32 +1374,34 @@
 ) %7B%0A            
+ 
+ 
 final List%3CStrin
@@ -1441,32 +1441,34 @@
 1);%0A            
+  
 labelValues.add(
@@ -1482,32 +1482,34 @@
 d);%0A            
+  
 samples.add(%0A   
@@ -1504,33 +1504,16 @@
 les.add(
-%0A                
 gauge(na
@@ -1528,37 +1528,16 @@
 ent_id%22)
-%0A                    
 .addMetr
@@ -1572,34 +1572,38 @@
 ()));%0A          
-%7D%0A
+  %7D%0A  
           break;
@@ -1595,32 +1595,34 @@
          break;%0A
+  
         %7D%0A      
@@ -1615,32 +1615,34 @@
       %7D%0A        
+  
 case %22consumer-f
@@ -1658,32 +1658,34 @@
 ger-metrics%22: %7B%0A
+  
           final 
@@ -1734,32 +1734,34 @@
 id%22);%0A          
+ 
+ 
 if (clientId != 
@@ -1763,130 +1763,8 @@
 d !=
- null && !name.name().contains(%22.%22)) %7B%0A            final String topic = name.tags().get(%22topic%22);%0A            if (topic !=
  nul
@@ -1835,9 +1835,9 @@
 t%3C%3E(
-2
+1
 );%0A 
@@ -1894,133 +1894,44 @@
     
-labelValues.add(topic);%0A              samples%0A                  .add(gauge(name, %22client_id%22, %22topic%22)%0A                      
+samples.add(gauge(name, %22client_id%22)
 .add
@@ -1985,15 +1985,8 @@
    %7D
- else %7B
 %0A   
@@ -1998,324 +1998,74 @@
     
-  final List%3CString%3E labelValues = new ArrayList%3C%3E(1);%0A              labelValues.add(clientId);%0A              samples.add(%0A                  gauge(name, %22client_id%22)%0A                      .addMetric(labelValues, metric.value()));%0A            %7D%0A          %7D%0A          break;%0A        %7D%0A        default:%0A          break;
+break;%0A          %7D%0A          default:%0A            break;%0A        %7D
 %0A   

fix consumer metrics