From 5f769ca2b462964a618a5f90ae9c16fd5fb04ae8 Mon Sep 17 00:00:00 2001 From: Gao Hongtao Date: Thu, 30 Jul 2020 07:32:29 +0800 Subject: [PATCH] Add AvgLabeledFunction to ingest multiple labels (#5187) --- .../meter/function/AvgLabeledFunction.java | 214 ++++++++++++++++++ .../core/analysis/metrics/DataTable.java | 5 + .../promethues/PrometheusMetricConverter.java | 38 +++- .../promethues/operation/MetricSource.java | 3 + .../promethues/rule/PrometheusMetric.java | 1 + .../core/metric/promethues/rule/Rules.java | 1 + .../function/AvgLabeledFunctionTest.java | 112 +++++++++ .../provider/PrometheusFetcherProvider.java | 3 + 8 files changed, 373 insertions(+), 4 deletions(-) create mode 100644 oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/meter/function/AvgLabeledFunction.java create mode 100644 oap-server/server-core/src/test/java/org/apache/skywalking/oap/server/core/analysis/meter/function/AvgLabeledFunctionTest.java diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/meter/function/AvgLabeledFunction.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/meter/function/AvgLabeledFunction.java new file mode 100644 index 000000000..84739e3da --- /dev/null +++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/meter/function/AvgLabeledFunction.java @@ -0,0 +1,214 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + * + */ + +package org.apache.skywalking.oap.server.core.analysis.meter.function; + +import java.util.HashMap; +import java.util.Map; +import java.util.Objects; +import java.util.Set; +import lombok.Getter; +import lombok.Setter; +import lombok.ToString; +import org.apache.skywalking.oap.server.core.Const; +import org.apache.skywalking.oap.server.core.UnexpectedException; +import org.apache.skywalking.oap.server.core.analysis.manual.instance.InstanceTraffic; +import org.apache.skywalking.oap.server.core.analysis.meter.MeterEntity; +import org.apache.skywalking.oap.server.core.analysis.metrics.DataTable; +import org.apache.skywalking.oap.server.core.analysis.metrics.Metrics; +import org.apache.skywalking.oap.server.core.remote.grpc.proto.RemoteData; +import org.apache.skywalking.oap.server.core.storage.StorageBuilder; +import org.apache.skywalking.oap.server.core.storage.annotation.Column; + +@MeterFunction(functionName = "avgLabeled") +@ToString +public abstract class AvgLabeledFunction extends Metrics implements AcceptableValue { + protected static final String SUMMATION = "summation"; + protected static final String COUNT = "count"; + protected static final String VALUE = "value"; + + @Setter + @Getter + @Column(columnName = ENTITY_ID, length = 512) + private String entityId; + + /** + * Service ID is required for sort query. + */ + @Setter + @Getter + @Column(columnName = InstanceTraffic.SERVICE_ID) + private String serviceId; + + @Getter + @Setter + @Column(columnName = SUMMATION, storageOnly = true) + protected DataTable summation = new DataTable(30); + @Getter + @Setter + @Column(columnName = COUNT, storageOnly = true) + protected DataTable count = new DataTable(30); + @Getter + @Setter + @Column(columnName = VALUE, dataType = Column.ValueDataType.LABELED_VALUE, storageOnly = true) + private DataTable value = new DataTable(30); + + @Override + public final void combine(Metrics metrics) { + AvgLabeledFunction longAvgMetrics = (AvgLabeledFunction) metrics; + summation.append(longAvgMetrics.summation); + count.append(longAvgMetrics.count); + } + + @Override + public final void calculate() { + Set keys = count.keys(); + for (String key : keys) { + Long s = summation.get(key); + if (Objects.isNull(s)) { + continue; + } + Long c = count.get(key); + if (Objects.isNull(c)) { + continue; + } + long result = s / c; + if (result == 0 && s > 0) { + result = 1; + } + value.put(key, result); + } + } + + @Override + public Metrics toHour() { + AvgLabeledFunction metrics = (AvgLabeledFunction) createNew(); + metrics.setEntityId(getEntityId()); + metrics.setTimeBucket(toTimeBucketInHour()); + metrics.setServiceId(getServiceId()); + metrics.setSummation(getSummation()); + metrics.setCount(getCount()); + return metrics; + } + + @Override + public Metrics toDay() { + AvgLabeledFunction metrics = (AvgLabeledFunction) createNew(); + metrics.setEntityId(getEntityId()); + metrics.setTimeBucket(toTimeBucketInDay()); + metrics.setServiceId(getServiceId()); + metrics.setSummation(getSummation()); + metrics.setCount(getCount()); + return metrics; + } + + @Override + public int remoteHashCode() { + return entityId.hashCode(); + } + + @Override + public void deserialize(final RemoteData remoteData) { + this.setCount(new DataTable(remoteData.getDataObjectStrings(0))); + this.setSummation(new DataTable(remoteData.getDataObjectStrings(1))); + setTimeBucket(remoteData.getDataLongs(0)); + + this.entityId = remoteData.getDataStrings(0); + this.serviceId = remoteData.getDataStrings(1); + } + + @Override + public RemoteData.Builder serialize() { + RemoteData.Builder remoteBuilder = RemoteData.newBuilder(); + remoteBuilder.addDataObjectStrings(count.toStorageData()); + remoteBuilder.addDataObjectStrings(summation.toStorageData()); + remoteBuilder.addDataLongs(getTimeBucket()); + + remoteBuilder.addDataStrings(entityId); + remoteBuilder.addDataStrings(serviceId); + + return remoteBuilder; + } + + @Override + public String id() { + return getTimeBucket() + Const.ID_CONNECTOR + entityId; + } + + @Override + public void accept(final MeterEntity entity, final DataTable value) { + this.entityId = entity.id(); + this.serviceId = entity.serviceId(); + this.summation.append(value); + DataTable c = new DataTable(); + value.keys().forEach(key -> c.put(key, 1L)); + this.count.append(c); + } + + @Override + public Class builder() { + return AvgLabeledStorageBuilder.class; + } + + public static class AvgLabeledStorageBuilder implements StorageBuilder { + @Override + public AvgLabeledFunction map2Data(final Map dbMap) { + AvgLabeledFunction metrics = new AvgLabeledFunction() { + @Override + public AcceptableValue createNew() { + throw new UnexpectedException("createNew should not be called"); + } + }; + metrics.setSummation(new DataTable((String) dbMap.get(SUMMATION))); + metrics.setValue(new DataTable((String) dbMap.get(VALUE))); + metrics.setCount(new DataTable((String) dbMap.get(COUNT))); + metrics.setTimeBucket(((Number) dbMap.get(TIME_BUCKET)).longValue()); + metrics.setServiceId((String) dbMap.get(InstanceTraffic.SERVICE_ID)); + metrics.setEntityId((String) dbMap.get(ENTITY_ID)); + return metrics; + } + + @Override + public Map data2Map(final AvgLabeledFunction storageData) { + Map map = new HashMap<>(); + map.put(SUMMATION, storageData.getSummation()); + map.put(VALUE, storageData.getValue()); + map.put(COUNT, storageData.getCount()); + map.put(TIME_BUCKET, storageData.getTimeBucket()); + map.put(InstanceTraffic.SERVICE_ID, storageData.getServiceId()); + map.put(ENTITY_ID, storageData.getEntityId()); + return map; + } + } + + @Override + public boolean equals(Object o) { + if (this == o) + return true; + if (!(o instanceof AvgLabeledFunction)) + return false; + AvgLabeledFunction function = (AvgLabeledFunction) o; + return Objects.equals(entityId, function.entityId) && + getTimeBucket() == function.getTimeBucket(); + } + + @Override + public int hashCode() { + return Objects.hash(entityId, getTimeBucket()); + } +} diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/metrics/DataTable.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/metrics/DataTable.java index a104ee6c7..2fca2da41 100644 --- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/metrics/DataTable.java +++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/metrics/DataTable.java @@ -21,6 +21,7 @@ import java.util.ArrayList; import java.util.Comparator; import java.util.HashMap; import java.util.List; +import java.util.Set; import java.util.stream.Collectors; import lombok.EqualsAndHashCode; import lombok.ToString; @@ -94,6 +95,10 @@ public class DataTable implements StorageDataComplexObject { return values; } + public Set keys() { + return data.keySet(); + } + public boolean hasData() { return !data.isEmpty(); } diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/metric/promethues/PrometheusMetricConverter.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/metric/promethues/PrometheusMetricConverter.java index 8e50f5090..c564c1a5a 100644 --- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/metric/promethues/PrometheusMetricConverter.java +++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/metric/promethues/PrometheusMetricConverter.java @@ -18,6 +18,7 @@ package org.apache.skywalking.oap.server.core.metric.promethues; +import com.google.common.base.Preconditions; import com.google.common.base.Strings; import com.google.common.collect.ImmutableMap; import io.vavr.Function1; @@ -25,6 +26,7 @@ import io.vavr.Tuple; import io.vavr.Tuple3; import io.vavr.control.Try; import java.math.BigDecimal; +import java.util.Collections; import java.util.Comparator; import java.util.List; import java.util.Map; @@ -45,6 +47,7 @@ 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 org.apache.skywalking.oap.server.core.analysis.worker.MetricsStreamProcessor; import org.apache.skywalking.oap.server.core.metric.promethues.counter.Window; import org.apache.skywalking.oap.server.core.metric.promethues.operation.MetricSource; @@ -71,6 +74,8 @@ public class PrometheusMetricConverter { private final static String AVG = "avg"; + private final static String AVG_LABELED = "avgLabeled"; + private final Window window = new Window(); private final List rules; @@ -132,6 +137,7 @@ public class PrometheusMetricConverter { .timestamp(tuple._4.getTimestamp()) .scale(tuple._3.getScale()) .counterFunction(tuple._3.getCounterFunction()) + .groupBy(tuple._3.getGroupBy()) .range(tuple._3.getRange()); switch (tuple._1.getScope()) { case SERVICE: @@ -155,10 +161,23 @@ public class PrometheusMetricConverter { case AVG: sources.forEach((source, metrics) -> { AcceptableValue value = service.buildMetrics(formatMetricName(operation.getMetricName()), Long.class); - Double sumDouble = sum(metrics).value(); - sumDouble = window.get(source.getPromMetricName()).apply(source, sumDouble); - value.accept(source.getEntity(), BigDecimal.valueOf(Double.isNaN(sumDouble) ? 0D : sumDouble) - .multiply(BigDecimal.TEN.pow(source.getScale())).longValue()); + value.accept(source.getEntity(), sum(metrics, source)); + value.setTimeBucket(TimeBucket.getMinuteTimeBucket(source.getTimestamp())); + log.debug("Input metric {}", value.getTimeBucket()); + service.doStreamingCalculation(value); + + generateTraffic(source.getEntity()); + }); + break; + case AVG_LABELED: + sources.forEach((source, metrics) -> { + Preconditions.checkArgument(Objects.nonNull(source.getGroupBy())); + DataTable dt = new DataTable(); + metrics.stream() + .collect(groupingBy(m -> source.getGroupBy().stream().map(m.getLabels()::get).collect(Collectors.joining("-")))) + .forEach((group, mm) -> dt.put(group, sum(mm, source, ImmutableMap.of("group", group)))); + AcceptableValue value = service.buildMetrics(formatMetricName(operation.getMetricName()), DataTable.class); + value.accept(source.getEntity(), dt); value.setTimeBucket(TimeBucket.getMinuteTimeBucket(source.getTimestamp())); log.debug("Input metric {}", value.getTimeBucket()); service.doStreamingCalculation(value); @@ -228,6 +247,17 @@ public class PrometheusMetricConverter { return metrics.stream().reduce(Metric::sum).orElseThrow(IllegalArgumentException::new); } + private long sum(List metrics, MetricSource source) { + return sum(metrics, source, Collections.emptyMap()); + } + + private long sum(List metrics, MetricSource source, Map labels) { + Double sumDouble = sum(metrics).value(); + sumDouble = window.get(source.getPromMetricName(), labels).apply(source, sumDouble); + return BigDecimal.valueOf(Double.isNaN(sumDouble) ? 0D : sumDouble) + .multiply(BigDecimal.TEN.pow(source.getScale())).longValue(); + } + private void generateTraffic(MeterEntity entity) { ServiceTraffic s = new ServiceTraffic(); s.setName(requireNonNull(entity.getServiceName())); diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/metric/promethues/operation/MetricSource.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/metric/promethues/operation/MetricSource.java index ed503a241..983077b78 100644 --- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/metric/promethues/operation/MetricSource.java +++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/metric/promethues/operation/MetricSource.java @@ -18,6 +18,7 @@ package org.apache.skywalking.oap.server.core.metric.promethues.operation; +import java.util.List; import lombok.Builder; import lombok.EqualsAndHashCode; import lombok.Getter; @@ -41,4 +42,6 @@ public class MetricSource { private final String range; private final int scale; + + private final List groupBy; } diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/metric/promethues/rule/PrometheusMetric.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/metric/promethues/rule/PrometheusMetric.java index f02baee5b..4c5cdd0c2 100644 --- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/metric/promethues/rule/PrometheusMetric.java +++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/metric/promethues/rule/PrometheusMetric.java @@ -28,6 +28,7 @@ public class PrometheusMetric { private CounterFunction counterFunction; private String range; private List labelFilter; + private List groupBy; private Relabel relabel; private int scale = 0; } diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/metric/promethues/rule/Rules.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/metric/promethues/rule/Rules.java index 6e39e722a..91537c49c 100644 --- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/metric/promethues/rule/Rules.java +++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/metric/promethues/rule/Rules.java @@ -44,6 +44,7 @@ public class Rules { throw new ModuleStartException("Load fetcher rules failed", e); } return Arrays.stream(rules) + .filter(File::isFile) .map(f -> { try (Reader r = new FileReader(f)) { Rule rule = new Yaml().loadAs(r, Rule.class); diff --git a/oap-server/server-core/src/test/java/org/apache/skywalking/oap/server/core/analysis/meter/function/AvgLabeledFunctionTest.java b/oap-server/server-core/src/test/java/org/apache/skywalking/oap/server/core/analysis/meter/function/AvgLabeledFunctionTest.java new file mode 100644 index 000000000..df1fbe095 --- /dev/null +++ b/oap-server/server-core/src/test/java/org/apache/skywalking/oap/server/core/analysis/meter/function/AvgLabeledFunctionTest.java @@ -0,0 +1,112 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + * + */ + +package org.apache.skywalking.oap.server.core.analysis.meter.function; + +import io.vavr.collection.Stream; +import java.util.Comparator; +import java.util.List; +import java.util.Map; +import org.apache.skywalking.oap.server.core.analysis.meter.MeterEntity; +import org.apache.skywalking.oap.server.core.analysis.metrics.DataTable; +import org.apache.skywalking.oap.server.core.storage.StorageBuilder; +import org.junit.Test; +import org.junit.runner.RunWith; +import org.mockito.Mockito; +import org.mockito.Spy; +import org.mockito.runners.MockitoJUnitRunner; + +import static org.apache.skywalking.oap.server.core.analysis.meter.function.AvgLabeledFunction.COUNT; +import static org.apache.skywalking.oap.server.core.analysis.meter.function.AvgLabeledFunction.SUMMATION; +import static org.apache.skywalking.oap.server.core.analysis.meter.function.AvgLabeledFunction.VALUE; +import static org.hamcrest.core.Is.is; +import static java.util.Arrays.asList; +import static org.junit.Assert.assertThat; + +@RunWith(MockitoJUnitRunner.class) +public class AvgLabeledFunctionTest { + @Spy + private AvgLabeledFunction function; + + @Test + public void testAccept() { + function.accept(MeterEntity.newService("request_count"), build(asList("200", "404"), asList(10L, 2L))); + assertResult(asList("200", "404"), asList(10L, 2L), asList(1L, 1L)); + function.accept(MeterEntity.newService("request_count"), build(asList("200", "500"), asList(2L, 3L))); + assertResult(asList("200", "404", "500"), asList(12L, 2L, 3L), asList(2L, 1L, 1L)); + } + + @Test + public void testCalculate() { + function.accept(MeterEntity.newService("request_count"), build(asList("200", "404"), asList(10L, 2L))); + function.accept(MeterEntity.newService("request_count"), build(asList("200", "500"), asList(2L, 3L))); + function.calculate(); + + assertThat(function.getValue().sortedKeys(Comparator.naturalOrder()), is(asList("200", "404", "500"))); + assertThat(function.getValue().sortedValues(Comparator.naturalOrder()), is(asList(6L, 2L, 3L))); + } + + @Test + public void testSerialize() { + function.accept(MeterEntity.newService("request_count"), build(asList("200", "404"), asList(10L, 2L))); + AvgLabeledFunction function2 = Mockito.spy(AvgLabeledFunction.class); + function2.deserialize(function.serialize().build()); + assertThat(function2.getEntityId(), is(function.getEntityId())); + assertThat(function2.getTimeBucket(), is(function.getTimeBucket())); + } + + @Test + public void testBuilder() throws IllegalAccessException, InstantiationException { + function.accept(MeterEntity.newService("request_count"), build(asList("200", "404"), asList(10L, 2L))); + function.calculate(); + StorageBuilder storageBuilder = function.builder().newInstance(); + + Map map = storageBuilder.data2Map(function); + map.put(SUMMATION, ((DataTable) map.get(SUMMATION)).toStorageData()); + map.put(COUNT, ((DataTable) map.get(COUNT)).toStorageData()); + map.put(VALUE, ((DataTable) map.get(VALUE)).toStorageData()); + + AvgLabeledFunction function2 = storageBuilder.map2Data(map); + assertThat(function2.getValue(), is(function.getValue())); + } + + private DataTable build(List keys, List values) { + DataTable result = new DataTable(); + Stream.ofAll(keys).forEachWithIndex((key, i) -> result.put(key, values.get(i))); + return result; + } + + private void assertResult(List expectedKeys, List expectedValues, List expectedCount) { + assertSummation(expectedKeys, expectedValues); + assertCount(expectedKeys, expectedCount); + } + + private void assertCount(List expectedKeys, List expectedCount) { + List keys = function.getCount().sortedKeys(Comparator.comparingInt(Integer::parseInt)); + assertThat(keys, is(expectedKeys)); + List values = function.getCount().sortedValues(Comparator.comparingLong(Long::parseLong)); + assertThat(values, is(expectedCount)); + } + + private void assertSummation(List expectedKeys, List expectedValues) { + List keys = function.getSummation().sortedKeys(Comparator.comparingInt(Integer::parseInt)); + assertThat(keys, is(expectedKeys)); + List values = function.getSummation().sortedValues(Comparator.comparingLong(Long::parseLong)); + assertThat(values, is(expectedValues)); + } +} \ No newline at end of file diff --git a/oap-server/server-fetcher-plugin/prometheus-fetcher-plugin/src/main/java/org/apache/skywalking/oap/server/fetcher/prometheus/provider/PrometheusFetcherProvider.java b/oap-server/server-fetcher-plugin/prometheus-fetcher-plugin/src/main/java/org/apache/skywalking/oap/server/fetcher/prometheus/provider/PrometheusFetcherProvider.java index 1c3178d86..ffd3c7635 100644 --- a/oap-server/server-fetcher-plugin/prometheus-fetcher-plugin/src/main/java/org/apache/skywalking/oap/server/fetcher/prometheus/provider/PrometheusFetcherProvider.java +++ b/oap-server/server-fetcher-plugin/prometheus-fetcher-plugin/src/main/java/org/apache/skywalking/oap/server/fetcher/prometheus/provider/PrometheusFetcherProvider.java @@ -127,6 +127,9 @@ public class PrometheusFetcherProvider extends ModuleProvider { while ((mf = p.parse(now)) != null) { result.addAll(mf.getMetrics().stream() .peek(metric -> { + if (Objects.isNull(sc.getLabels())) { + return; + } Map extraLabels = Maps.newHashMap(sc.getLabels()); extraLabels.put("instance", url); extraLabels.forEach((key, value) -> {