From 217bb624f6b39f9ac9dd289485200fe5fbbbdde5 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=90=B4=E6=99=9F=20Wu=20Sheng?= Date: Wed, 29 Apr 2020 08:43:26 +0800 Subject: [PATCH] Support percentile function in the meter system. (#4725) * Support percentile function in the meter system. * Update document. * Fix formats. --- docs/en/concepts-and-designs/meter.md | 7 +- .../oal/rt/parser/AnalysisResult.java | 2 +- .../oal/rt/parser/PersistenceColumns.java | 12 +- .../oal/rt/parser/PersistenceField.java | 4 +- .../code-templates/metrics/deserialize.ftl | 2 +- .../core/analysis/meter/MeterSystem.java | 19 +- .../meter/function/HistogramFunction.java | 15 - .../meter/function/PercentileFunction.java | 297 ++++++++++++++++++ .../server/core/analysis/metrics/IntList.java | 87 +++++ .../function/PercentileFunctionTest.java | 208 ++++++++++++ .../provider/PrometheusFetcherProvider.java | 37 +++ .../base/ColumnTypeEsMapping.java | 4 +- .../plugin/jdbc/h2/dao/H2TableInstaller.java | 4 +- .../jdbc/mysql/MySQLTableInstaller.java | 4 +- 14 files changed, 663 insertions(+), 39 deletions(-) create mode 100644 oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/meter/function/PercentileFunction.java create mode 100644 oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/metrics/IntList.java create mode 100644 oap-server/server-core/src/test/java/org/apache/skywalking/oap/server/core/analysis/meter/function/PercentileFunctionTest.java diff --git a/docs/en/concepts-and-designs/meter.md b/docs/en/concepts-and-designs/meter.md index 6ae796d806..d07160d678 100644 --- a/docs/en/concepts-and-designs/meter.md +++ b/docs/en/concepts-and-designs/meter.md @@ -17,5 +17,8 @@ The values of scope entity name, such as service name, are required when metrics NOTICE, the metrics must be declared in the bootstrap stage, no runtime changed. Meter System supports following binding functions -- **avg**, calculate the avg value for every entity in the same metrics name. -- **histogram**. aggregate the counts in the configurable buckets. +- **avg**. Calculate the avg value for every entity in the same metrics name. +- **histogram**. Aggregate the counts in the configurable buckets, buckets is configurable but must be assigned in the declaration stage. +- **percentile**. Read [percentile in WIKI](https://en.wikipedia.org/wiki/Percentile). Unlike in the OAL, we provide +50/75/90/95/99 in default, in the meter system function, percentile function accepts several ranks, which should be in +the (0, 100) range. diff --git a/oap-server/oal-rt/src/main/java/org/apache/skywalking/oal/rt/parser/AnalysisResult.java b/oap-server/oal-rt/src/main/java/org/apache/skywalking/oal/rt/parser/AnalysisResult.java index c0cbe700b9..31a975b96e 100644 --- a/oap-server/oal-rt/src/main/java/org/apache/skywalking/oal/rt/parser/AnalysisResult.java +++ b/oap-server/oal-rt/src/main/java/org/apache/skywalking/oal/rt/parser/AnalysisResult.java @@ -139,7 +139,7 @@ public class AnalysisResult { } else if (columnType.equals(long.class)) { serializeFields.addLongField(column.getFieldName()); } else if (StorageDataComplexObject.class.isAssignableFrom(columnType)) { - serializeFields.addObjectField(column.getFieldName()); + serializeFields.addObjectField(column.getFieldName(), columnType.getName()); } else { throw new IllegalStateException( "Unexpected field type [" + columnType.getSimpleName() + "] of persistence column [" + column diff --git a/oap-server/oal-rt/src/main/java/org/apache/skywalking/oal/rt/parser/PersistenceColumns.java b/oap-server/oal-rt/src/main/java/org/apache/skywalking/oal/rt/parser/PersistenceColumns.java index 0fab3aa94f..195e3f8474 100644 --- a/oap-server/oal-rt/src/main/java/org/apache/skywalking/oal/rt/parser/PersistenceColumns.java +++ b/oap-server/oal-rt/src/main/java/org/apache/skywalking/oal/rt/parser/PersistenceColumns.java @@ -29,23 +29,23 @@ public class PersistenceColumns { private List objectFields = new LinkedList<>(); public void addStringField(String fieldName) { - stringFields.add(new PersistenceField(fieldName)); + stringFields.add(new PersistenceField(fieldName, "String")); } public void addLongField(String fieldName) { - longFields.add(new PersistenceField(fieldName)); + longFields.add(new PersistenceField(fieldName, "long")); } public void addDoubleField(String fieldName) { - doubleFields.add(new PersistenceField(fieldName)); + doubleFields.add(new PersistenceField(fieldName, "double")); } public void addIntField(String fieldName) { - intFields.add(new PersistenceField(fieldName)); + intFields.add(new PersistenceField(fieldName, "int")); } - public void addObjectField(String fieldName) { - objectFields.add(new PersistenceField(fieldName)); + public void addObjectField(String fieldName, String fieldType) { + objectFields.add(new PersistenceField(fieldName, fieldType)); } public List getStringFields() { diff --git a/oap-server/oal-rt/src/main/java/org/apache/skywalking/oal/rt/parser/PersistenceField.java b/oap-server/oal-rt/src/main/java/org/apache/skywalking/oal/rt/parser/PersistenceField.java index 015cebb3ac..622e7e8519 100644 --- a/oap-server/oal-rt/src/main/java/org/apache/skywalking/oal/rt/parser/PersistenceField.java +++ b/oap-server/oal-rt/src/main/java/org/apache/skywalking/oal/rt/parser/PersistenceField.java @@ -29,10 +29,12 @@ public class PersistenceField { private String fieldName; private String setter; private String getter; + private String fieldType; - public PersistenceField(String fieldName) { + public PersistenceField(String fieldName, String fieldType) { this.fieldName = fieldName; this.setter = ClassMethodUtil.toSetMethod(fieldName); this.getter = ClassMethodUtil.toGetMethod(fieldName); + this.fieldType = fieldType; } } diff --git a/oap-server/oal-rt/src/main/resources/code-templates/metrics/deserialize.ftl b/oap-server/oal-rt/src/main/resources/code-templates/metrics/deserialize.ftl index 24841514c8..455634f520 100644 --- a/oap-server/oal-rt/src/main/resources/code-templates/metrics/deserialize.ftl +++ b/oap-server/oal-rt/src/main/resources/code-templates/metrics/deserialize.ftl @@ -16,7 +16,7 @@ public void deserialize(org.apache.skywalking.oap.server.core.remote.grpc.proto. <#list serializeFields.objectFields as field> - ${field.setter}(new org.apache.skywalking.oap.server.core.analysis.metrics.DataTable(remoteData.getDataObjectStrings(${field?index}))); + ${field.setter}(new ${field.fieldType}(remoteData.getDataObjectStrings(${field?index}))); } \ No newline at end of file diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/meter/MeterSystem.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/meter/MeterSystem.java index 2d5173be22..6ba9a479aa 100644 --- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/meter/MeterSystem.java +++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/meter/MeterSystem.java @@ -159,13 +159,18 @@ public class MeterSystem implements Service { boolean foundDataType = false; String acceptance = null; for (final Type genericInterface : meterFunction.getGenericInterfaces()) { - ParameterizedType parameterizedType = (ParameterizedType) genericInterface; - if (parameterizedType.getRawType().getTypeName().equals(AcceptableValue.class.getName())) { - Type[] arguments = parameterizedType.getActualTypeArguments(); - if (arguments[0].equals(dataType)) { - foundDataType = true; - } else { - acceptance = arguments[0].getTypeName(); + if (genericInterface instanceof ParameterizedType) { + ParameterizedType parameterizedType = (ParameterizedType) genericInterface; + if (parameterizedType.getRawType().getTypeName().equals(AcceptableValue.class.getName())) { + Type[] arguments = parameterizedType.getActualTypeArguments(); + if (arguments[0].equals(dataType)) { + foundDataType = true; + } else { + acceptance = arguments[0].getTypeName(); + } + } + if (foundDataType) { + break; } } } diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/meter/function/HistogramFunction.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/meter/function/HistogramFunction.java index 16f000aed5..f1150a64ad 100644 --- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/meter/function/HistogramFunction.java +++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/meter/function/HistogramFunction.java @@ -27,7 +27,6 @@ import lombok.ToString; import lombok.extern.slf4j.Slf4j; 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; @@ -58,13 +57,6 @@ public abstract class HistogramFunction extends Metrics implements AcceptableVal @Setter @Column(columnName = DATASET, dataType = Column.ValueDataType.HISTOGRAM, storageOnly = true, defaultValue = 0) private DataTable dataset = new DataTable(30); - /** - * Service ID is required for sort query. - */ - @Setter - @Getter - @Column(columnName = InstanceTraffic.SERVICE_ID) - private String serviceId; @Override public void accept(final MeterEntity entity, final BucketedValues value) { @@ -76,7 +68,6 @@ public abstract class HistogramFunction extends Metrics implements AcceptableVal } this.entityId = entity.id(); - this.serviceId = entity.serviceId(); final long[] values = value.getValues(); for (int i = 0; i < values.length; i++) { @@ -110,7 +101,6 @@ public abstract class HistogramFunction extends Metrics implements AcceptableVal HistogramFunction metrics = (HistogramFunction) createNew(); metrics.setEntityId(getEntityId()); metrics.setTimeBucket(toTimeBucketInHour()); - metrics.setServiceId(getServiceId()); metrics.setDataset(getDataset()); return metrics; } @@ -120,7 +110,6 @@ public abstract class HistogramFunction extends Metrics implements AcceptableVal HistogramFunction metrics = (HistogramFunction) createNew(); metrics.setEntityId(getEntityId()); metrics.setTimeBucket(toTimeBucketInDay()); - metrics.setServiceId(getServiceId()); metrics.setDataset(getDataset()); return metrics; } @@ -135,7 +124,6 @@ public abstract class HistogramFunction extends Metrics implements AcceptableVal this.setTimeBucket(remoteData.getDataLongs(0)); this.setEntityId(remoteData.getDataStrings(0)); - this.setServiceId(remoteData.getDataStrings(1)); this.setDataset(new DataTable(remoteData.getDataObjectStrings(0))); } @@ -146,7 +134,6 @@ public abstract class HistogramFunction extends Metrics implements AcceptableVal remoteBuilder.addDataLongs(getTimeBucket()); remoteBuilder.addDataStrings(entityId); - remoteBuilder.addDataStrings(serviceId); remoteBuilder.addDataObjectStrings(dataset.toStorageData()); @@ -175,7 +162,6 @@ public abstract class HistogramFunction extends Metrics implements AcceptableVal }; metrics.setDataset(new DataTable((String) dbMap.get(DATASET))); 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; } @@ -185,7 +171,6 @@ public abstract class HistogramFunction extends Metrics implements AcceptableVal Map map = new HashMap<>(); map.put(DATASET, storageData.getDataset()); map.put(TIME_BUCKET, storageData.getTimeBucket()); - map.put(InstanceTraffic.SERVICE_ID, storageData.getServiceId()); map.put(ENTITY_ID, storageData.getEntityId()); return map; } diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/meter/function/PercentileFunction.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/meter/function/PercentileFunction.java new file mode 100644 index 0000000000..01d767bfcb --- /dev/null +++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/meter/function/PercentileFunction.java @@ -0,0 +1,297 @@ +/* + * 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.Comparator; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.stream.IntStream; +import lombok.EqualsAndHashCode; +import lombok.Getter; +import lombok.RequiredArgsConstructor; +import lombok.Setter; +import lombok.extern.slf4j.Slf4j; +import org.apache.skywalking.oap.server.core.Const; +import org.apache.skywalking.oap.server.core.UnexpectedException; +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.IntList; +import org.apache.skywalking.oap.server.core.analysis.metrics.Metrics; +import org.apache.skywalking.oap.server.core.analysis.metrics.MultiIntValuesHolder; +import org.apache.skywalking.oap.server.core.analysis.metrics.PercentileMetrics; +import org.apache.skywalking.oap.server.core.query.type.Bucket; +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; + +/** + * PercentileFunction is the implementation of {@link PercentileMetrics} in the meter system. The major difference is + * the PercentileFunction accepts the {@link PercentileArgument} as input rather than every single request. + */ +@MeterFunction(functionName = "percentile") +@Slf4j +@EqualsAndHashCode(of = { + "entityId", + "timeBucket" +}) +public abstract class PercentileFunction extends Metrics implements AcceptableValue, MultiIntValuesHolder { + public static final String DATASET = "dataset"; + public static final String RANKS = "ranks"; + public static final String VALUE = "value"; + + @Setter + @Getter + @Column(columnName = ENTITY_ID) + private String entityId; + @Getter + @Setter + @Column(columnName = VALUE, dataType = Column.ValueDataType.LABELED_VALUE, storageOnly = true) + private DataTable percentileValues = new DataTable(10); + @Getter + @Setter + @Column(columnName = DATASET, storageOnly = true) + private DataTable dataset = new DataTable(30); + /** + * Rank + */ + @Getter + @Setter + @Column(columnName = RANKS, storageOnly = true) + private IntList ranks = new IntList(10); + + private boolean isCalculated = false; + + @Override + public void accept(final MeterEntity entity, final PercentileArgument value) { + if (dataset.size() > 0) { + if (!value.getBucketedValues().isCompatible(dataset)) { + throw new IllegalArgumentException( + "Incompatible BucketedValues [" + value + "] for current PercentileFunction[" + dataset + "]"); + } + } + + for (final int rank : value.getRanks()) { + if (rank <= 0) { + throw new IllegalArgumentException("Illegal rank value " + rank + ", must be positive"); + } + } + + if (ranks.size() > 0) { + if (ranks.size() != value.getRanks().length) { + throw new IllegalArgumentException( + "Incompatible ranks size = [" + value.getRanks().length + "] for current PercentileFunction[" + ranks + .size() + "]"); + } else { + for (final int rank : value.getRanks()) { + if (!ranks.include(rank)) { + throw new IllegalArgumentException( + "Rank " + rank + " doesn't exist in the previous ranks " + ranks); + } + } + } + } else { + for (final int rank : value.getRanks()) { + ranks.add(rank); + } + } + + this.entityId = entity.id(); + + final long[] values = value.getBucketedValues().getValues(); + for (int i = 0; i < values.length; i++) { + final long bucket = value.getBucketedValues().getBuckets()[i]; + String bucketName = bucket == Integer.MIN_VALUE ? Bucket.INFINITE_NEGATIVE : String.valueOf(bucket); + final long bucketValue = values[i]; + dataset.valueAccumulation(bucketName, bucketValue); + } + + this.isCalculated = false; + } + + @Override + public void combine(final Metrics metrics) { + PercentileFunction percentile = (PercentileFunction) metrics; + + if (!dataset.keysEqual(percentile.getDataset())) { + log.warn("Incompatible input [{}}] for current HistogramFunction[{}], entity {}", + percentile, this, entityId + ); + return; + } + if (ranks.size() > 0) { + IntList ranksOfThat = percentile.getRanks(); + if (this.ranks.size() != ranks.size()) { + log.warn("Incompatible ranks size = [{}}] for current PercentileFunction[{}]", + ranks.size(), this.ranks.size() + ); + return; + } else { + if (!this.ranks.equals(percentile.getRanks())) { + log.warn("Rank {} doesn't exist in the previous ranks {}", percentile.getRanks(), ranks); + return; + } + } + } + + this.dataset.append(percentile.dataset); + + this.isCalculated = false; + } + + @Override + public void calculate() { + if (!isCalculated) { + long total = dataset.sumOfValues(); + + int[] roofs = new int[ranks.size()]; + for (int i = 0; i < ranks.size(); i++) { + roofs[i] = Math.round(total * ranks.get(i) * 1.0f / 100); + } + + int count = 0; + final List sortedKeys = dataset.sortedKeys(Comparator.comparingInt(Integer::parseInt)); + + int loopIndex = 0; + + for (String key : sortedKeys) { + final Long value = dataset.get(key); + + count += value; + for (int rankIdx = loopIndex; rankIdx < roofs.length; rankIdx++) { + int roof = roofs[rankIdx]; + + if (count >= roof) { + percentileValues.put(String.valueOf(ranks.get(rankIdx)), Long.parseLong(key)); + loopIndex++; + } else { + break; + } + } + } + } + } + + @Override + public Metrics toHour() { + PercentileFunction metrics = (PercentileFunction) createNew(); + metrics.setEntityId(getEntityId()); + metrics.setTimeBucket(toTimeBucketInHour()); + metrics.setDataset(getDataset()); + metrics.setRanks(getRanks()); + metrics.setPercentileValues(getPercentileValues()); + return metrics; + } + + @Override + public Metrics toDay() { + PercentileFunction metrics = (PercentileFunction) createNew(); + metrics.setEntityId(getEntityId()); + metrics.setTimeBucket(toTimeBucketInDay()); + metrics.setDataset(getDataset()); + metrics.setRanks(getRanks()); + metrics.setPercentileValues(getPercentileValues()); + return metrics; + } + + @Override + public int[] getValues() { + return percentileValues.sortedValues(Comparator.comparingInt(Integer::parseInt)) + .stream() + .flatMapToInt(l -> IntStream.of(l.intValue())) + .toArray(); + } + + @Override + public int remoteHashCode() { + return entityId.hashCode(); + } + + @Override + public void deserialize(final RemoteData remoteData) { + this.setTimeBucket(remoteData.getDataLongs(0)); + + this.setEntityId(remoteData.getDataStrings(0)); + + this.setDataset(new DataTable(remoteData.getDataObjectStrings(0))); + this.setRanks(new IntList(remoteData.getDataObjectStrings(1))); + this.setPercentileValues(new DataTable(remoteData.getDataObjectStrings(2))); + } + + @Override + public RemoteData.Builder serialize() { + RemoteData.Builder remoteBuilder = RemoteData.newBuilder(); + remoteBuilder.addDataLongs(getTimeBucket()); + + remoteBuilder.addDataStrings(entityId); + + remoteBuilder.addDataObjectStrings(dataset.toStorageData()); + remoteBuilder.addDataObjectStrings(ranks.toStorageData()); + remoteBuilder.addDataObjectStrings(percentileValues.toStorageData()); + + return remoteBuilder; + } + + @Override + public String id() { + return getTimeBucket() + Const.ID_CONNECTOR + entityId; + } + + @Override + public Class builder() { + return PercentileFunctionBuilder.class; + } + + @RequiredArgsConstructor + @Getter + public static class PercentileArgument { + private final BucketedValues bucketedValues; + private final int[] ranks; + } + + public static class PercentileFunctionBuilder implements StorageBuilder { + + @Override + public PercentileFunction map2Data(final Map dbMap) { + PercentileFunction metrics = new PercentileFunction() { + @Override + public AcceptableValue createNew() { + throw new UnexpectedException("createNew should not be called"); + } + }; + metrics.setDataset(new DataTable((String) dbMap.get(DATASET))); + metrics.setRanks(new IntList((String) dbMap.get(RANKS))); + metrics.setPercentileValues(new DataTable((String) dbMap.get(VALUE))); + metrics.setTimeBucket(((Number) dbMap.get(TIME_BUCKET)).longValue()); + metrics.setEntityId((String) dbMap.get(ENTITY_ID)); + return metrics; + } + + @Override + public Map data2Map(final PercentileFunction storageData) { + Map map = new HashMap<>(); + map.put(DATASET, storageData.getDataset()); + map.put(RANKS, storageData.getRanks()); + map.put(VALUE, storageData.getPercentileValues()); + map.put(TIME_BUCKET, storageData.getTimeBucket()); + map.put(ENTITY_ID, storageData.getEntityId()); + return map; + } + } +} diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/metrics/IntList.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/metrics/IntList.java new file mode 100644 index 0000000000..b5bd9efe90 --- /dev/null +++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/metrics/IntList.java @@ -0,0 +1,87 @@ +/* + * 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.metrics; + +import java.util.ArrayList; +import java.util.List; +import lombok.EqualsAndHashCode; +import lombok.ToString; +import org.apache.skywalking.oap.server.core.Const; +import org.apache.skywalking.oap.server.core.storage.type.StorageDataComplexObject; + +/** + * IntList is a serializable array list carrying int values. + */ +@ToString +@EqualsAndHashCode +public class IntList implements StorageDataComplexObject { + private List data; + + public IntList(int initialSize) { + this.data = new ArrayList(initialSize); + } + + public IntList(String valueString) { + toObject(valueString); + } + + public int size() { + return data.size(); + } + + public boolean include(int value) { + return data.contains(value); + } + + @Override + public String toStorageData() { + StringBuilder builder = new StringBuilder(); + + this.data.forEach(element -> { + if (builder.length() != 0) { + // For the first element. + builder.append(Const.ARRAY_SPLIT); + } + builder.append(element); + }); + return builder.toString(); + } + + @Override + public void toObject(final String data) { + String[] elements = data.split(Const.ARRAY_PARSER_SPLIT); + this.data = new ArrayList<>(elements.length); + for (String element : elements) { + this.data.add(Integer.parseInt(element)); + } + } + + @Override + public void copyFrom(final IntList source) { + this.data.addAll(source.data); + } + + public void add(final int rank) { + this.data.add(rank); + } + + public int get(final int idx) { + return this.data.get(idx); + } +} diff --git a/oap-server/server-core/src/test/java/org/apache/skywalking/oap/server/core/analysis/meter/function/PercentileFunctionTest.java b/oap-server/server-core/src/test/java/org/apache/skywalking/oap/server/core/analysis/meter/function/PercentileFunctionTest.java new file mode 100644 index 0000000000..6409d7bdbf --- /dev/null +++ b/oap-server/server-core/src/test/java/org/apache/skywalking/oap/server/core/analysis/meter/function/PercentileFunctionTest.java @@ -0,0 +1,208 @@ +/* + * 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.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.analysis.metrics.IntList; +import org.apache.skywalking.oap.server.core.storage.StorageBuilder; +import org.junit.Assert; +import org.junit.Test; + +public class PercentileFunctionTest { + private static final int[] BUCKETS = new int[] { + 0, + 50, + 100, + 250 + }; + + private static final int[] BUCKETS_2ND = new int[] { + 0, + 51, + 100, + 250 + }; + + private static final int[] RANKS = new int[] { + 50, + 90 + }; + + @Test + public void testFunction() { + PercentileFunctionInst inst = new PercentileFunctionInst(); + inst.accept( + MeterEntity.newService("service-test"), + new PercentileFunction.PercentileArgument( + new BucketedValues( + BUCKETS, + new long[] { + 10, + 20, + 30, + 40 + } + ), + RANKS + ) + ); + + inst.accept( + MeterEntity.newService("service-test"), + new PercentileFunction.PercentileArgument( + new BucketedValues( + BUCKETS, + new long[] { + 10, + 20, + 30, + 40 + } + ), + RANKS + ) + ); + + inst.calculate(); + final int[] values = inst.getValues(); + /** + * Expected percentile dataset + * 
+         *     0  , 20
+         *     50 , 40
+         *     100, 60 <- P50
+         *     250, 80 <- P90
+         *
+ */ + Assert.assertArrayEquals(new int[] { + 100, + 250 + }, values); + } + + @Test(expected = IllegalArgumentException.class) + public void testIncompatible() { + PercentileFunctionInst inst = new PercentileFunctionInst(); + inst.accept( + MeterEntity.newService("service-test"), + new PercentileFunction.PercentileArgument( + new BucketedValues( + BUCKETS, + new long[] { + 10, + 20, + 30, + 40 + } + ), + RANKS + ) + ); + + inst.accept( + MeterEntity.newService("service-test"), + new PercentileFunction.PercentileArgument( + new BucketedValues( + BUCKETS_2ND, + new long[] { + 10, + 20, + 30, + 40 + } + ), + RANKS + ) + ); + } + + @Test + public void testSerialization() { + PercentileFunctionInst inst = new PercentileFunctionInst(); + inst.accept( + MeterEntity.newService("service-test"), + new PercentileFunction.PercentileArgument( + new BucketedValues( + BUCKETS, + new long[] { + 10, + 20, + 30, + 40 + } + ), + RANKS + ) + ); + + PercentileFunctionInst inst2 = new PercentileFunctionInst(); + inst2.deserialize(inst.serialize().build()); + + Assert.assertEquals(inst, inst2); + // HistogramFunction equal doesn't include dataset. + Assert.assertEquals(inst.getDataset(), inst2.getDataset()); + Assert.assertEquals(inst.getRanks(), inst2.getRanks()); + Assert.assertEquals(0, inst2.getPercentileValues().size()); + } + + @Test + public void testBuilder() throws IllegalAccessException, InstantiationException { + PercentileFunctionInst inst = new PercentileFunctionInst(); + inst.accept( + MeterEntity.newService("service-test"), + new PercentileFunction.PercentileArgument( + new BucketedValues( + BUCKETS, + new long[] { + 10, + 20, + 30, + 40 + } + ), + RANKS + ) + ); + inst.calculate(); + + final StorageBuilder storageBuilder = inst.builder().newInstance(); + + // Simulate the storage layer do, convert the datatable to string. + final Map map = storageBuilder.data2Map(inst); + map.put(PercentileFunction.DATASET, ((DataTable) map.get(PercentileFunction.DATASET)).toStorageData()); + map.put(PercentileFunction.VALUE, ((DataTable) map.get(PercentileFunction.VALUE)).toStorageData()); + map.put(PercentileFunction.RANKS, ((IntList) map.get(PercentileFunction.RANKS)).toStorageData()); + + final PercentileFunction inst2 = (PercentileFunction) storageBuilder.map2Data(map); + Assert.assertEquals(inst, inst2); + // HistogramFunction equal doesn't include dataset. + Assert.assertEquals(inst.getDataset(), inst2.getDataset()); + Assert.assertEquals(inst.getPercentileValues(), inst2.getPercentileValues()); + Assert.assertEquals(inst.getRanks(), inst2.getRanks()); + } + + private static class PercentileFunctionInst extends PercentileFunction { + @Override + public AcceptableValue createNew() { + return new PercentileFunctionInst(); + } + } +} 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 9b5efcd8d7..0e2a00f040 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 @@ -27,6 +27,7 @@ import org.apache.skywalking.oap.server.core.analysis.meter.MeterSystem; import org.apache.skywalking.oap.server.core.analysis.meter.ScopeType; import org.apache.skywalking.oap.server.core.analysis.meter.function.AcceptableValue; import org.apache.skywalking.oap.server.core.analysis.meter.function.BucketedValues; +import org.apache.skywalking.oap.server.core.analysis.meter.function.PercentileFunction; import org.apache.skywalking.oap.server.fetcher.prometheus.module.PrometheusFetcherModule; import org.apache.skywalking.oap.server.library.module.ModuleConfig; import org.apache.skywalking.oap.server.library.module.ModuleDefine; @@ -64,6 +65,10 @@ public class PrometheusFetcherProvider extends ModuleProvider { final MeterSystem meterSystem = MeterSystem.meterSystem(getManager()); meterSystem.create("test_long_metrics", "avg", ScopeType.SERVICE, Long.class); meterSystem.create("test_histogram_metrics", "histogram", ScopeType.SERVICE, BucketedValues.class); + meterSystem.create( + "test_percentile_metrics", "percentile", ScopeType.SERVICE, + PercentileFunction.PercentileArgument.class + ); } } @@ -109,6 +114,38 @@ public class PrometheusFetcherProvider extends ModuleProvider { } )); service.doStreamingCalculation(histogramMetrics); + + // Percentile Example + final AcceptableValue testPercentileMetrics = service.buildMetrics( + "test_percentile_metrics", PercentileFunction.PercentileArgument.class); + testPercentileMetrics.setTimeBucket(TimeBucket.getMinuteTimeBucket(System.currentTimeMillis())); + testPercentileMetrics.accept( + MeterEntity.newService("service-test"), + new PercentileFunction.PercentileArgument( + new BucketedValues( + // Buckets + new int[] { + 0, + 51, + 100, + 250 + }, + // Values + new long[] { + 10, + 20, + 30, + 40 + } + ), + // Ranks + new int[] { + 50, + 90 + } + ) + ); + service.doStreamingCalculation(testPercentileMetrics); } }, 2, 2, TimeUnit.SECONDS); } diff --git a/oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/base/ColumnTypeEsMapping.java b/oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/base/ColumnTypeEsMapping.java index 7eefe653ce..7cec27ec28 100644 --- a/oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/base/ColumnTypeEsMapping.java +++ b/oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/base/ColumnTypeEsMapping.java @@ -20,8 +20,8 @@ package org.apache.skywalking.oap.server.storage.plugin.elasticsearch.base; import com.google.gson.JsonObject; import org.apache.skywalking.oap.server.core.analysis.NodeType; -import org.apache.skywalking.oap.server.core.analysis.metrics.DataTable; import org.apache.skywalking.oap.server.core.storage.model.DataTypeMapping; +import org.apache.skywalking.oap.server.core.storage.type.StorageDataComplexObject; public class ColumnTypeEsMapping implements DataTypeMapping { @@ -35,7 +35,7 @@ public class ColumnTypeEsMapping implements DataTypeMapping { return "double"; } else if (String.class.equals(type)) { return "keyword"; - } else if (DataTable.class.equals(type)) { + } else if (StorageDataComplexObject.class.isAssignableFrom(type)) { return "text"; } else if (byte[].class.equals(type)) { return "binary"; diff --git a/oap-server/server-storage-plugin/storage-jdbc-hikaricp-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/jdbc/h2/dao/H2TableInstaller.java b/oap-server/server-storage-plugin/storage-jdbc-hikaricp-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/jdbc/h2/dao/H2TableInstaller.java index 4d10bf4684..53e5266ba5 100644 --- a/oap-server/server-storage-plugin/storage-jdbc-hikaricp-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/jdbc/h2/dao/H2TableInstaller.java +++ b/oap-server/server-storage-plugin/storage-jdbc-hikaricp-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/jdbc/h2/dao/H2TableInstaller.java @@ -23,12 +23,12 @@ import java.sql.Connection; import java.sql.SQLException; import lombok.extern.slf4j.Slf4j; import org.apache.skywalking.oap.server.core.analysis.NodeType; -import org.apache.skywalking.oap.server.core.analysis.metrics.DataTable; import org.apache.skywalking.oap.server.core.storage.StorageException; import org.apache.skywalking.oap.server.core.storage.model.ColumnName; import org.apache.skywalking.oap.server.core.storage.model.Model; import org.apache.skywalking.oap.server.core.storage.model.ModelColumn; import org.apache.skywalking.oap.server.core.storage.model.ModelInstaller; +import org.apache.skywalking.oap.server.core.storage.type.StorageDataComplexObject; import org.apache.skywalking.oap.server.library.client.Client; import org.apache.skywalking.oap.server.library.client.jdbc.JDBCClientException; import org.apache.skywalking.oap.server.library.client.jdbc.hikaricp.JDBCHikariCPClient; @@ -98,7 +98,7 @@ public class H2TableInstaller extends ModelInstaller { return "DOUBLE"; } else if (String.class.equals(type)) { return "VARCHAR(" + column.getLength() + ")"; - } else if (DataTable.class.equals(type)) { + } else if (StorageDataComplexObject.class.isAssignableFrom(type)) { return "VARCHAR(20000)"; } else if (byte[].class.equals(type)) { return "MEDIUMTEXT"; diff --git a/oap-server/server-storage-plugin/storage-jdbc-hikaricp-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/jdbc/mysql/MySQLTableInstaller.java b/oap-server/server-storage-plugin/storage-jdbc-hikaricp-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/jdbc/mysql/MySQLTableInstaller.java index 839fefb77d..a31a8d75e1 100644 --- a/oap-server/server-storage-plugin/storage-jdbc-hikaricp-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/jdbc/mysql/MySQLTableInstaller.java +++ b/oap-server/server-storage-plugin/storage-jdbc-hikaricp-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/jdbc/mysql/MySQLTableInstaller.java @@ -22,11 +22,11 @@ import java.sql.Connection; import java.sql.ResultSet; import java.sql.SQLException; import lombok.extern.slf4j.Slf4j; -import org.apache.skywalking.oap.server.core.analysis.metrics.DataTable; import org.apache.skywalking.oap.server.core.storage.StorageException; import org.apache.skywalking.oap.server.core.storage.model.ExtraQueryIndex; import org.apache.skywalking.oap.server.core.storage.model.Model; import org.apache.skywalking.oap.server.core.storage.model.ModelColumn; +import org.apache.skywalking.oap.server.core.storage.type.StorageDataComplexObject; import org.apache.skywalking.oap.server.library.client.Client; import org.apache.skywalking.oap.server.library.client.jdbc.JDBCClientException; import org.apache.skywalking.oap.server.library.client.jdbc.hikaricp.JDBCHikariCPClient; @@ -105,7 +105,7 @@ public class MySQLTableInstaller extends H2TableInstaller { @Override protected String getColumnType(final ModelColumn column) { - if (DataTable.class.equals(column.getType())) { + if (StorageDataComplexObject.class.isAssignableFrom(column.getType())) { return "MEDIUMTEXT"; } return super.getColumnType(column); -- GitLab