kazusa
/
skywalking-java
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Provide labeled meter when receive meter (#5659)

This commit is contained in:
mrproliu 2020-10-17 22:36:12 +08:00 committed by GitHub
parent 45e02ec106
commit 7b2a0b4dac
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
7 changed files with 109 additions and 34 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
docs/en/setup/backend/backend-meter.md
View File

@ -49,6 +49,9 @@ meters:
operation: <string>
# Meter value parse groovy script.
value: <string>
# Aggregate metrics group by dedicated labels
groupBy:
- <labelName>
# <Optional> Appoint percentiles if using avgHistogramPercentile operation.
percentile:
- <rank>
@ -56,7 +59,7 @@ meters:
#### Meter transform operation
The available operations are `avg`, `avgHistogram` and `avgHistogramPercentile`. The `avg` and `avgXXX` mean to average
The available operations are `avg`, `avgLabeled`, `avgHistogram` and `avgHistogramPercentile`. The `avg` and `avgXXX` mean to average
the raw received metrics.
When you specify `avgHistogram` and `avgHistogramPercentile`, the source should be the type of `histogram`.

5
oap-server/analyzer/agent-analyzer/src/main/java/org/apache/skywalking/oap/server/analyzer/provider/meter/config/MeterDataConfig.java
View File

@ -42,4 +42,9 @@ public class MeterDataConfig {
*/
private List<Integer> percentile;
/**
* Aggregate meter group by dedicated labels
*/
private List<String> groupBy;
}

19
oap-server/analyzer/agent-analyzer/src/main/java/org/apache/skywalking/oap/server/analyzer/provider/meter/process/EvalMultipleData.java
View File

@ -22,8 +22,12 @@ import io.vavr.Function2;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.stream.Collectors;
import static java.util.stream.Collectors.groupingBy;
/**
* Combined meter data, has multiple meter data. Support batch process the express on meter
*/
@ -122,6 +126,21 @@ public class EvalMultipleData extends EvalData<EvalMultipleData> {
return dataList.stream().reduce(EvalData::combine).orElseThrow(IllegalArgumentException::new);
}
/**
* Combine all of the meter and group by labeled names
* @return Same labeled names and values
*/
Map<String, EvalData> combineAndGroupBy(List<String> labelNames) {
return dataList.stream()
// group by label
.collect(groupingBy(m -> labelNames.stream().map(l -> m.getLabels().getOrDefault(l, "")).map(Objects::toString).collect(Collectors.joining("-"))))
.entrySet().stream().collect(
// combine labeled values
Collectors.toMap(
e -> e.getKey(),
e -> e.getValue().stream().reduce(EvalData::combine).orElse(null)));
}
/**
* Append data to ready eval list
*/

74
oap-server/analyzer/agent-analyzer/src/main/java/org/apache/skywalking/oap/server/analyzer/provider/meter/process/MeterBuilder.java
View File

@ -28,8 +28,12 @@ import org.apache.skywalking.oap.server.core.analysis.meter.MeterSystem;
import org.apache.skywalking.oap.server.core.analysis.meter.function.AcceptableValue;
import org.apache.skywalking.oap.server.core.analysis.meter.function.AvgHistogramPercentileFunction;
import org.apache.skywalking.oap.server.core.analysis.meter.function.BucketedValues;
import org.apache.skywalking.oap.server.core.analysis.metrics.DataTable;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.StringJoiner;
import java.util.concurrent.atomic.AtomicBoolean;
@ -42,6 +46,9 @@ public class MeterBuilder {
private final MeterConfig config;
private final MeterSystem meterSystem;
private final static String DEFAULT_GROUP = "default";
private final static List<String> DEFAULT_GROUP_LIST = Collections.singletonList(DEFAULT_GROUP);
/**
* Current meter has init finished.
*/
@ -96,36 +103,49 @@ public class MeterBuilder {
log.warn("avg function not support histogram value, please check meter:{}", combinedSingleAvgData.getName());
}
break;
case "avgLabeled":
final DataTable dt = new DataTable();
values.combineAndGroupBy(Optional.ofNullable(config.getMeter().getGroupBy()).orElse(DEFAULT_GROUP_LIST)).entrySet().stream()
.forEach(e -> dt.put(e.getKey(), (long) ((EvalSingleData) e.getValue()).getValue()));
AcceptableValue<DataTable> value = meterSystem.buildMetrics(metricsName, DataTable.class);
value.accept(entity, dt);
value.setTimeBucket(TimeBucket.getMinuteTimeBucket(processor.timestamp()));
meterSystem.doStreamingCalculation(value);
break;
case "avgHistogram":
case "avgHistogramPercentile":
final EvalData combinedHistogramData = values.combineAsSingleData();
if (combinedHistogramData instanceof EvalHistogramData) {
final EvalHistogramData histogram = (EvalHistogramData) combinedHistogramData;
long[] buckets = new long[histogram.getBuckets().size()];
long[] bucketValues = new long[histogram.getBuckets().size()];
int i = 0;
for (Map.Entry<Double, Long> entry : histogram.getBuckets().entrySet()) {
buckets[i] = entry.getKey().intValue();
bucketValues[i] = entry.getValue();
i++;
}
values.combineAndGroupBy(Optional.ofNullable(config.getMeter().getGroupBy()).orElse(DEFAULT_GROUP_LIST)).entrySet().stream()
.forEach(e -> {
final String group = e.getKey();
final EvalData combinedHistogramData = e.getValue();
if (combinedHistogramData instanceof EvalHistogramData) {
final EvalHistogramData histogram = (EvalHistogramData) combinedHistogramData;
long[] buckets = new long[histogram.getBuckets().size()];
long[] bucketValues = new long[histogram.getBuckets().size()];
int i = 0;
for (Map.Entry<Double, Long> entry : histogram.getBuckets().entrySet()) {
buckets[i] = entry.getKey().intValue();
bucketValues[i] = entry.getValue();
i++;
}
if (config.getMeter().getOperation().equals("avgHistogram")) {
AcceptableValue<BucketedValues> avgHistogramValue = meterSystem.buildMetrics(metricsName, BucketedValues.class);
avgHistogramValue.accept(entity, new BucketedValues(buckets, bucketValues));
avgHistogramValue.setTimeBucket(TimeBucket.getMinuteTimeBucket(processor.timestamp()));
meterSystem.doStreamingCalculation(avgHistogramValue);
} else {
final AcceptableValue<AvgHistogramPercentileFunction.AvgPercentileArgument> percentileValue =
meterSystem.buildMetrics(metricsName, AvgHistogramPercentileFunction.AvgPercentileArgument.class);
percentileValue.accept(entity, new AvgHistogramPercentileFunction.AvgPercentileArgument(new BucketedValues(buckets, bucketValues),
config.getMeter().getPercentile().stream().mapToInt(Integer::intValue).toArray()));
percentileValue.setTimeBucket(TimeBucket.getMinuteTimeBucket(processor.timestamp()));
meterSystem.doStreamingCalculation(percentileValue);
}
} else {
log.warn(config.getMeter().getOperation() + " function not support single value, please check meter:{}", combinedHistogramData.getName());
}
final BucketedValues bucketedValues = new BucketedValues(buckets, bucketValues);
bucketedValues.setGroup(group);
if (config.getMeter().getOperation().equals("avgHistogram")) {
AcceptableValue<BucketedValues> avgHistogramValue = meterSystem.buildMetrics(metricsName, BucketedValues.class);
avgHistogramValue.accept(entity, bucketedValues);
avgHistogramValue.setTimeBucket(TimeBucket.getMinuteTimeBucket(processor.timestamp()));
meterSystem.doStreamingCalculation(avgHistogramValue);
} else {
final AcceptableValue<AvgHistogramPercentileFunction.AvgPercentileArgument> percentileValue =
meterSystem.buildMetrics(metricsName, AvgHistogramPercentileFunction.AvgPercentileArgument.class);
percentileValue.accept(entity, new AvgHistogramPercentileFunction.AvgPercentileArgument(bucketedValues,
config.getMeter().getPercentile().stream().mapToInt(Integer::intValue).toArray()));
percentileValue.setTimeBucket(TimeBucket.getMinuteTimeBucket(processor.timestamp()));
meterSystem.doStreamingCalculation(percentileValue);
}
}
});
break;
default:
log.warn("Cannot support function:{}", config.getMeter().getOperation());

30
oap-server/analyzer/agent-analyzer/src/test/java/org/apache/skywalking/oap/server/analyzer/provider/meter/process/EvalMultipleDataTest.java
View File

@ -19,13 +19,15 @@
package org.apache.skywalking.oap.server.analyzer.provider.meter.process;
import io.vavr.Function2;
import java.util.Arrays;
import java.util.List;
import org.junit.Assert;
import org.junit.Before;
import org.junit.Test;
import org.powermock.reflect.Whitebox;
import java.util.Arrays;
import java.util.List;
import java.util.Map;
public class EvalMultipleDataTest extends EvalDataBaseTest {
private EvalMultipleData singleMultiple;
@ -104,6 +106,30 @@ public class EvalMultipleDataTest extends EvalDataBaseTest {
Assert.assertEquals(35, combinedHistogramData.getBuckets().get(10d).longValue());
}
@Test
public void testCombineAndGroupBy() {
final Map<String, EvalData> singleValueCombineAndGroupBy = singleMultiple.combineAndGroupBy(Arrays.asList("k1"));
Assert.assertEquals(3, singleValueCombineAndGroupBy.size());
Assert.assertEquals(10d, ((EvalSingleData) singleValueCombineAndGroupBy.get("v1")).getValue(), 0.0);
Assert.assertEquals(20d, ((EvalSingleData) singleValueCombineAndGroupBy.get("v2")).getValue(), 0.0);
Assert.assertEquals(30d, ((EvalSingleData) singleValueCombineAndGroupBy.get("v3")).getValue(), 0.0);
final Map<String, EvalData> histogramCombineAndGroupBy = histogramMultiple.combineAndGroupBy(Arrays.asList("k1"));
Assert.assertEquals(3, histogramCombineAndGroupBy.size());
Assert.assertEquals(10, ((EvalHistogramData) histogramCombineAndGroupBy.get("v1")).getBuckets().get(1d).longValue());
Assert.assertEquals(20, ((EvalHistogramData) histogramCombineAndGroupBy.get("v1")).getBuckets().get(5d).longValue());
Assert.assertEquals(3, ((EvalHistogramData) histogramCombineAndGroupBy.get("v1")).getBuckets().get(10d).longValue());
Assert.assertEquals(5, ((EvalHistogramData) histogramCombineAndGroupBy.get("v2")).getBuckets().get(1d).longValue());
Assert.assertEquals(10, ((EvalHistogramData) histogramCombineAndGroupBy.get("v2")).getBuckets().get(5d).longValue());
Assert.assertEquals(7, ((EvalHistogramData) histogramCombineAndGroupBy.get("v2")).getBuckets().get(10d).longValue());
Assert.assertEquals(15, ((EvalHistogramData) histogramCombineAndGroupBy.get("v3")).getBuckets().get(1d).longValue());
Assert.assertEquals(20, ((EvalHistogramData) histogramCombineAndGroupBy.get("v3")).getBuckets().get(5d).longValue());
Assert.assertEquals(25, ((EvalHistogramData) histogramCombineAndGroupBy.get("v3")).getBuckets().get(10d).longValue());
}
/**
* Verify the multiple data operation
*/

5
oap-server/analyzer/agent-analyzer/src/test/java/org/apache/skywalking/oap/server/analyzer/provider/meter/process/MeterBuilderTest.java
View File

@ -18,8 +18,6 @@
package org.apache.skywalking.oap.server.analyzer.provider.meter.process;
import java.util.ArrayList;
import java.util.List;
import org.apache.skywalking.oap.server.core.analysis.IDManager;
import org.apache.skywalking.oap.server.core.analysis.TimeBucket;
import org.apache.skywalking.oap.server.core.analysis.meter.function.AcceptableValue;
@ -35,6 +33,9 @@ import org.powermock.core.classloader.annotations.PowerMockIgnore;
import org.powermock.modules.junit4.PowerMockRunner;
import org.powermock.reflect.Whitebox;
import java.util.ArrayList;
import java.util.List;
import static org.mockito.Matchers.any;
import static org.mockito.Mockito.doAnswer;
import static org.mockito.Mockito.doCallRealMethod;

5
oap-server/analyzer/agent-analyzer/src/test/java/org/apache/skywalking/oap/server/analyzer/provider/trace/TraceSampleRateWatcherTest.java
View File

@ -18,8 +18,6 @@
package org.apache.skywalking.oap.server.analyzer.provider.trace;
import java.util.Optional;
import java.util.Set;
import org.apache.skywalking.oap.server.analyzer.provider.AnalyzerModuleProvider;
import org.apache.skywalking.oap.server.configuration.api.ConfigChangeWatcher;
import org.apache.skywalking.oap.server.configuration.api.ConfigTable;
@ -30,6 +28,9 @@ import org.junit.Test;
import org.junit.runner.RunWith;
import org.mockito.runners.MockitoJUnitRunner;
import java.util.Optional;
import java.util.Set;
import static org.hamcrest.CoreMatchers.is;
import static org.hamcrest.MatcherAssert.assertThat;