refactor: agg dispatcher (#701)

traas-stack · Oct 19, 2023 · 9fae622 · 9fae622
1 parent 47a49e8
commit 9fae622
Show file tree

Hide file tree

Showing 6 changed files with 25 additions and 129 deletions.
diff --git a/...oot/src/main/java/io/holoinsight/server/agg/v1/executor/executor/MetricStorageOutput.java b/...oot/src/main/java/io/holoinsight/server/agg/v1/executor/executor/MetricStorageOutput.java
@@ -39,7 +39,6 @@ public String type() {
 
   @Override
   public void write(Batch batch) {
-    log.info("xxx");
     WriteMetricsParam param = new WriteMetricsParam();
     param.setTenant(batch.key.getTenant());
     List<WriteMetricsParam.Point> points = new ArrayList<>(batch.groups.size());

diff --git a/server/agg/agg-boot/src/main/java/temp.groovy b/server/agg/agg-boot/src/main/java/temp.groovy
diff --git a/...g/agg-dispatcher/src/main/java/io/holoinsight/server/agg/v1/dispatcher/AggDispatcher.java b/...g/agg-dispatcher/src/main/java/io/holoinsight/server/agg/v1/dispatcher/AggDispatcher.java
@@ -162,7 +162,8 @@ public void dispatch(AuthInfo authInfo, WriteMetricsRequestV4 request) {
         ProducerRecord<AggTaskKey, AggProtos.AggTaskValue> record =
             new ProducerRecord<>(aggProperties.getTopic(), aggKey, aggTaskValue.build());
         kafkaProducer.send(record);
-        StatUtils.KAFKA_SEND.add(StringsKey.of("v4"), Stats.V_1);
+        StatUtils.KAFKA_SEND.add(StringsKey.of("v4"),
+            new long[] {1, aggTaskValue.getInDataNodesCount()});
       }
     }
   }
@@ -204,7 +205,8 @@ public void dispatch(AuthInfo authInfo, WriteMetricsRequestV1 request) {
         ProducerRecord<AggTaskKey, AggProtos.AggTaskValue> record =
             new ProducerRecord<>(aggProperties.getTopic(), aggTaskKey, taskValue);
         kafkaProducer.send(record);
-        StatUtils.KAFKA_SEND.add(StringsKey.of("v1"), Stats.V_1);
+        StatUtils.KAFKA_SEND.add(StringsKey.of("v1"),
+            new long[] {1, taskValue.getInDataNodesCount()});
       }
     }
   }
@@ -291,9 +293,14 @@ public void dispatchDetailData(AuthInfo authInfo, GatewayHook.Table table) {
       ProducerRecord<AggTaskKey, AggProtos.AggTaskValue> record =
           new ProducerRecord<>(aggProperties.getTopic(), aggTaskKey, aggTaskValue);
       kafkaProducer.send(record);
+      StatUtils.KAFKA_SEND.add(StringsKey.of("detail"), new long[] {1, table.getRows().size()});
     }
   }
 
+  public boolean supportsDetail(String name) {
+    return CollectionUtils.isNotEmpty(this.storage.getByMetric(name));
+  }
+
   @NotNull
   private static String buildPartition(StringBuilder sb, AuthInfo authInfo, AggTask aggTask) {
     if (sb == null) {

diff --git a/.../agg-dispatcher/src/main/java/io/holoinsight/server/agg/v1/dispatcher/AggGatewayHook.java b/.../agg-dispatcher/src/main/java/io/holoinsight/server/agg/v1/dispatcher/AggGatewayHook.java
@@ -34,4 +34,9 @@ public void writeMetricsV4(AuthInfo authInfo, WriteMetricsRequestV4 request) {
   public void writeDetail(AuthInfo authInfo, Table table) {
     aggDispatcher.dispatchDetailData(authInfo, table);
   }
+
+  @Override
+  public boolean supportsDetail(String name) {
+    return aggDispatcher.supportsDetail(name);
+  }
 }
diff --git a/...gg/agg-executor/src/main/java/io/holoinsight/server/agg/v1/executor/executor/XSelect.java b/...gg/agg-executor/src/main/java/io/holoinsight/server/agg/v1/executor/executor/XSelect.java
@@ -59,7 +59,8 @@ public Map<String, List<XSelectItem>> getSelectItem(String metric) {
         XSelectItem item = items.get(index);
         SelectItem.Elect e = item.getInner().getElect();
 
-        if (item.getInner().getAgg().getTypeInt() == AggFunc.TYPE_HLL) {
+        int aggType = item.getInner().getAgg().getTypeInt();
+        if (aggType == AggFunc.TYPE_HLL || aggType == AggFunc.TYPE_COUNT) {
           electToItemMap.computeIfAbsent(e.getMetric(), i -> new HashMap<>())
               .computeIfAbsent(NO_FIELD, i -> new LinkedList<>()) //
               .add(item);

diff --git a/...teway/gateway-core/src/main/java/io/holoinsight/server/gateway/core/grpc/GatewayHook.java b/...teway/gateway-core/src/main/java/io/holoinsight/server/gateway/core/grpc/GatewayHook.java
@@ -4,7 +4,6 @@
 package io.holoinsight.server.gateway.core.grpc;
 
 import java.util.List;
-import java.util.Map;
 
 import io.holoinsight.server.common.auth.AuthInfo;
 import io.holoinsight.server.gateway.grpc.WriteMetricsRequestV1;
@@ -35,6 +34,8 @@ public interface GatewayHook {
 
   void writeDetail(AuthInfo authInfo, Table table);
 
+  boolean supportsDetail(String name);
+
   @lombok.Data
   class Table {
     private String name;
@@ -47,6 +48,13 @@ class Table {
   class Header {
     private List<String> tagKeys;
     private List<String> fieldKeys;
+
+    public Header() {}
+
+    public Header(List<String> tagKeys, List<String> fieldKeys) {
+      this.tagKeys = tagKeys;
+      this.fieldKeys = fieldKeys;
+    }
   }
 
   @lombok.Data