Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

refactor: agg dispatcher #701

Merged
merged 2 commits into from
Oct 19, 2023
Merged

refactor: agg dispatcher #701

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
62887b7
refactor: agg dispatcher
Oct 19, 2023
210895f
fix: format
Oct 19, 2023
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 0 additions & 1 deletion ...oot/src/main/java/io/holoinsight/server/agg/v1/executor/executor/MetricStorageOutput.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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());
Expand Down
124 changes: 0 additions & 124 deletions server/agg/agg-boot/src/main/java/temp.groovy
View file
Open in desktop

This file was deleted.

11 changes: 9 additions & 2 deletions ...g/agg-dispatcher/src/main/java/io/holoinsight/server/agg/v1/dispatcher/AggDispatcher.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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()});
}
}
}
Expand Down Expand Up @@ -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()});
}
}
}
Expand Down Expand Up @@ -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) {
Expand Down
5 changes: 5 additions & 0 deletions .../agg-dispatcher/src/main/java/io/holoinsight/server/agg/v1/dispatcher/AggGatewayHook.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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);
}
}
3 changes: 2 additions & 1 deletion ...gg/agg-executor/src/main/java/io/holoinsight/server/agg/v1/executor/executor/XSelect.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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);
Expand Down
10 changes: 9 additions & 1 deletion ...teway/gateway-core/src/main/java/io/holoinsight/server/gateway/core/grpc/GatewayHook.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -35,6 +34,8 @@ public interface GatewayHook {

void writeDetail(AuthInfo authInfo, Table table);

boolean supportsDetail(String name);

@lombok.Data
class Table {
private String name;
Expand All @@ -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
Expand Down