Add opencensus reciever (#5029)

Signed-off-by: NGao Hongtao <hanahmily@gmail.com>

docs/en/setup/backend/backend-receivers.md

@@ -13,6 +13,7 @@ We have following receivers, and `default` implementors are provided in our Apac
 1. **receiver-profile**. gRPC services accept profile task status and snapshot reporter. 
 1. **receiver_zipkin**. See [details](#zipkin-receiver).
 1. **receiver_jaeger**. See [details](#jaeger-receiver).
+1. **receiver-oc**. See [details](#oc-receiver).
 
 The sample settings of these receivers should be already in default `application.yml`, and also list here
 ```yaml
@@ -109,4 +110,21 @@ receiver_jaeger:
     gRPCPort: ${SW_RECEIVER_JAEGER_PORT:14250}
 ``` 
 
-NOTICE, Jaeger receiver is only provided in `apache-skywalking-apm-x.y.z.tar.gz` tar.
\ No newline at end of file
+NOTICE, Jaeger receiver is only provided in `apache-skywalking-apm-x.y.z.tar.gz` tar.
+
+## Opencensus receiver
+
+Opencensus receiver supports to ingest agent metrics by meter-system. OAP can load the configuration at bootstrap. 
+If the new configuration is not well-formed, OAP fails to start up. The files are located at `$CLASSPATH/oc-rules`.
+
+The file is written in YAML format, defined by the scheme described in [prometheus-fetcher](./backend-fetcher.md).
+Notice, `receiver-oc` only support `metricsRules` node of scheme due to the push mode it opts to.
+
+To active the `default` implementation:
+```yaml
+receiver-oc:
+  selector: ${SW_OC_RECEIVER:-}
+  default:
+    gRPCHost: ${SW_OC_RECEIVER_GRPC_HOST:0.0.0.0}
+    gRPCPort: ${SW_OC_RECEIVER_GRPC_PORT:55678}
+```

oap-server/server-bootstrap/pom.xml

@@ -111,6 +111,11 @@
             <artifactId>skywalking-profile-receiver-plugin</artifactId>
             <version>${project.version}</version>
         </dependency>
+        <dependency>
+            <groupId>org.apache.skywalking</groupId>
+            <artifactId>opencensus-receiver-plugin</artifactId>
+            <version>${project.version}</version>
+        </dependency>
         <!-- receiver module -->
 
         <!-- fetcher module -->

oap-server/server-bootstrap/src/main/resources/application.yml

@@ -196,6 +196,12 @@ prometheus-fetcher:
   default:
     active: ${SW_PROMETHEUS_FETCHER_ACTIVE:false}
 
+receiver-oc:
+  selector: ${SW_OC_RECEIVER:-}
+  default:
+    gRPCHost: ${SW_OC_RECEIVER_GRPC_HOST:0.0.0.0}
+    gRPCPort: ${SW_OC_RECEIVER_GRPC_PORT:55678}
+
 receiver_zipkin:
   selector: ${SW_RECEIVER_ZIPKIN:-}
   default:

oap-server/server-bootstrap/src/main/resources/log4j2.xml

@@ -32,7 +32,7 @@
         <logger name="io.netty" level="INFO"/>
         <logger name="org.apache.http" level="INFO"/>
         <logger name="org.apache.skywalking.oap.server.core.alarm.AlarmStandardPersistence" level="DEBUG"/>
-        <logger name="org.apache.skywalking.oap.server.core" level="INFO"/>
+        <logger name="org.apache.skywalking.oap.server.core" level="DEBUG"/>
         <logger name="org.apache.skywalking.oap.server.core.analysis.worker" level="DEBUG" />
         <logger name="org.apache.skywalking.oap.server.core.remote.client" level="DEBUG"/>
         <logger name="org.apache.skywalking.oap.server.library.buffer" level="INFO"/>

oap-server/server-bootstrap/src/main/resources/oc-rules/oap.yaml

+# 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.
+
+# This will parse a textual representation of a duration. The formats
+# accepted are based on the ISO-8601 duration format {@code PnDTnHnMn.nS}
+# with days considered to be exactly 24 hours.
+# <p>
+# Examples:
+# <pre>
+#    "PT20.345S" -- parses as "20.345 seconds"
+#    "PT15M"     -- parses as "15 minutes" (where a minute is 60 seconds)
+#    "PT10H"     -- parses as "10 hours" (where an hour is 3600 seconds)
+#    "P2D"       -- parses as "2 days" (where a day is 24 hours or 86400 seconds)
+#    "P2DT3H4M"  -- parses as "2 days, 3 hours and 4 minutes"
+#    "P-6H3M"    -- parses as "-6 hours and +3 minutes"
+#    "-P6H3M"    -- parses as "-6 hours and -3 minutes"
+#    "-P-6H+3M"  -- parses as "+6 hours and -3 minutes"
+# </pre>
+metricsRules:
+  - name: instance_cpu_percentage
+    scope: SERVICE_INSTANCE
+    operation: avg
+    sources:
+      process_cpu_seconds_total:
+        counterFunction: RATE
+        range: PT1M
+        scale: 2
+        relabel:
+          service:
+            - service
+          instance:
+            - host_name
+  - name: instance_jvm_memory_bytes_used
+    scope: SERVICE_INSTANCE
+    operation: avg
+    sources:
+      jvm_memory_bytes_used:
+        relabel:
+          service:
+            - service
+          instance:
+            - instance
+  - name: instance_jvm_young_gc_count
+    scope: SERVICE_INSTANCE
+    operation: avg
+    sources:
+      jvm_gc_collection_seconds_count:
+        counterFunction: INCREASE
+        range: PT1M
+        labelFilter:
+          - key: gc
+            options:
+              - "PS Scavenge"
+              - "Copy"
+              - "ParNew"
+              - "G1 Young Generation"
+        relabel:
+          service:
+            - service
+          instance:
+            - instance
+  - name: instance_jvm_young_gc_time
+    scope: SERVICE_INSTANCE
+    operation: avg
+    sources:
+      jvm_gc_collection_seconds:
+        labelFilter:
+          - key: gc
+            options:
+              - "PS Scavenge"
+              - "Copy"
+              - "ParNew"
+              - "G1 Young Generation"
+        relabel:
+          service:
+            - service
+          instance:
+            - instance
+  - name: instance_jvm_old_gc_count
+    scope: SERVICE_INSTANCE
+    operation: avg
+    sources:
+      jvm_gc_collection_seconds_count:
+        counterFunction: INCREASE
+        range: PT1M
+        labelFilter:
+          - key: gc
+            options:
+              - "PS MarkSweep"
+              - "MarkSweepCompact"
+              - "ConcurrentMarkSweep"
+              - "G1 Old Generation"
+        relabel:
+          service:
+            - service
+          instance:
+            - instance
+  - name: instance_jvm_old_gc_time
+    scope: SERVICE_INSTANCE
+    operation: avg
+    sources:
+      jvm_gc_collection_seconds:
+        labelFilter:
+          - key: gc
+            options:
+              - "PS MarkSweep"
+              - "MarkSweepCompact"
+              - "ConcurrentMarkSweep"
+              - "G1 Old Generation"
+        relabel:
+          service:
+            - service
+          instance:
+            - instance
+  - name: instance_trace_count
+    scope: SERVICE_INSTANCE
+    operation: avg
+    sources:
+      trace_in_latency_count:
+        counterFunction: INCREASE
+        range: PT1M
+        relabel:
+          service:
+            - service
+          instance:
+            - instance
+  - name: instance_trace_latency_percentile
+    scope: SERVICE_INSTANCE
+    operation: avgHistogramPercentile
+    percentiles: [50, 70, 90, 99]
+    sources:
+      trace_in_latency:
+        counterFunction: INCREASE
+        range: PT1M
+        relabel:
+          service:
+            - service
+          instance:
+            - instance
+  - name: instance_trace_analysis_error_count
+    scope: SERVICE_INSTANCE
+    operation: avg
+    sources:
+      trace_analysis_error_count:
+        counterFunction: INCREASE
+        range: PT1M
+        relabel:
+          service:
+            - service
+          instance:
+            - instanc
+  - name: instance_mesh_count
+    scope: SERVICE_INSTANCE
+    operation: avg
+    sources:
+      mesh_analysis_latency_count:
+        counterFunction: INCREASE
+        range: PT1M
+        relabel:
+          service:
+            - service
+          instance:
+            - instance
+  - name: instance_mesh_latency_percentile
+    scope: SERVICE_INSTANCE
+    operation: avgHistogramPercentile
+    percentiles: [50, 70, 90, 99]
+    sources:
+      mesh_analysis_latency:
+        counterFunction: INCREASE
+        range: PT10M
+        relabel:
+          service:
+            - service
+          instance:
+            - instance
+  - name: instance_mesh_analysis_error_count
+    scope: SERVICE_INSTANCE
+    operation: avg
+    sources:
+      mesh_analysis_error_count:
+        counterFunction: INCREASE
+        range: PT1M
+        relabel:
+          service:
+            - service
+          instance:
+            - instance
+  - name: instance_metrics_first_aggregation
+    scope: SERVICE_INSTANCE
+    operation: avg
+    sources:
+      metrics_aggregation:
+        counterFunction: INCREASE
+        range: PT1M
+        labelFilter:
+          - key: dimensionality
+            options: ["min"]
+          - key: level
+            options: ["1"]
+        relabel:
+          service:
+            - service
+          instance:
+            - instance
+  - name: instance_metrics_second_aggregation
+    scope: SERVICE_INSTANCE
+    operation: avg
+    sources:
+      metrics_aggregation:
+        counterFunction: INCREASE
+        range: PT1M
+        labelFilter:
+          - key: dimensionality
+            options: ["min"]
+          - key: level
+            options: ["2"]
+        relabel:
+          service:
+            - service
+          instance:
+            - instance
+  - name: instance_persistence_execute_percentile
+    scope: SERVICE_INSTANCE
+    operation: avgHistogramPercentile
+    percentiles: [50, 70, 90, 99]
+    sources:
+      persistence_timer_bulk_execute_latency:
+        counterFunction: INCREASE
+        range: PT5M
+        relabel:
+          service:
+            - service
+          instance:
+            - instance
+  - name: instance_persistence_prepare_percentile
+    scope: SERVICE_INSTANCE
+    operation: avgHistogramPercentile
+    percentiles: [50, 70, 90, 99]
+    sources:
+      persistence_timer_bulk_prepare_latency:
+        counterFunction: INCREASE
+        range: PT5M
+        relabel:
+          service:
+            - service
+          instance:
+            - instance
+  - name: instance_persistence_error_count
+    scope: SERVICE_INSTANCE
+    operation: avg
+    sources:
+      persistence_timer_bulk_error_count:
+        counterFunction: INCREASE
+        range: PT1M
+        relabel:
+          service:
+            - service
+          instance:
+            - instance
+  - name: instance_persistence_execute_count
+    scope: SERVICE_INSTANCE
+    operation: avg
+    sources:
+      persistence_timer_bulk_execute_latency_count:
+        counterFunction: INCREASE
+        range: PT1M
+        relabel:
+          service:
+            - service
+          instance:
+            - instance
+  - name: instance_persistence_prepare_count
+    scope: SERVICE_INSTANCE
+    operation: avg
+    sources:
+      persistence_timer_bulk_prepare_latency_count:
+        counterFunction: INCREASE
+        range: PT1M
+        relabel:
+          service:
+            - service
+          instance:
+            - instance

oap-server/server-core/pom.xml

@@ -78,6 +78,10 @@
             <groupId>org.javassist</groupId>
             <artifactId>javassist</artifactId>
         </dependency>
+        <dependency>
+            <groupId>io.vavr</groupId>
+            <artifactId>vavr</artifactId>
+        </dependency>
 
         <dependency>
             <groupId>org.apache.skywalking</groupId>

oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/metric/promethues/PrometheusMetricConverter.java

+/*
+ * 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.metric.promethues;
+
+import com.google.common.base.Strings;
+import com.google.common.collect.ImmutableMap;
+import io.vavr.Function1;
+import io.vavr.Tuple;
+import io.vavr.Tuple3;
+import io.vavr.control.Try;
+import java.math.BigDecimal;
+import java.util.Comparator;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.StringJoiner;
+import java.util.concurrent.atomic.AtomicReference;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.lang3.Validate;
+import org.apache.skywalking.oap.server.core.analysis.NodeType;
+import org.apache.skywalking.oap.server.core.analysis.TimeBucket;
+import org.apache.skywalking.oap.server.core.analysis.manual.endpoint.EndpointTraffic;
+import org.apache.skywalking.oap.server.core.analysis.manual.instance.InstanceTraffic;
+import org.apache.skywalking.oap.server.core.analysis.manual.service.ServiceTraffic;
+import org.apache.skywalking.oap.server.core.analysis.meter.MeterEntity;
+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.worker.MetricsStreamProcessor;
+import org.apache.skywalking.oap.server.core.metric.promethues.counter.Window;
+import org.apache.skywalking.oap.server.core.metric.promethues.operation.MetricSource;
+import org.apache.skywalking.oap.server.core.metric.promethues.operation.Operation;
+import org.apache.skywalking.oap.server.core.metric.promethues.rule.MetricsRule;
+import org.apache.skywalking.oap.server.library.util.prometheus.metrics.Counter;
+import org.apache.skywalking.oap.server.library.util.prometheus.metrics.Histogram;
+import org.apache.skywalking.oap.server.library.util.prometheus.metrics.Metric;
+import org.apache.skywalking.oap.server.library.util.prometheus.metrics.Summary;
+
+import static java.util.Objects.requireNonNull;
+import static java.util.stream.Collectors.groupingBy;
+import static java.util.stream.Collectors.mapping;
+import static java.util.stream.Collectors.toList;
+
+@Slf4j
+public class PrometheusMetricConverter {
+
+    private final static BigDecimal SECOND_TO_MILLISECOND  = BigDecimal.TEN.pow(3);
+
+    private final static String AVG_HISTOGRAM = "avgHistogram";
+
+    private final static String AVG_PERCENTILE = "avgHistogramPercentile";
+
+    private final static String AVG = "avg";
+
+    private final Window window = new Window();
+
+    private final List<MetricsRule> rules;
+
+    private final MeterSystem service;
+
+    public PrometheusMetricConverter(List<MetricsRule> rules, MeterSystem service) {
+        this.rules = rules;
+        this.service = service;
+        final AtomicReference<String> lastRuleName = new AtomicReference<>();
+        rules.stream().sorted(Comparator.comparing(MetricsRule::getName)).forEach(rule -> {
+            if (rule.getName().equals(lastRuleName.get())) {
+                lastRuleName.set(rule.getName());
+                return;
+            }
+            service.create(formatMetricName(rule.getName()), rule.getOperation(), rule.getScope());
+            lastRuleName.set(rule.getName());
+        });
+    }
+
+    public void toMeter(Stream<Metric> metricStream) {
+        metricStream
+            .flatMap(metric -> {
+                if (metric instanceof Histogram) {
+                    Histogram h = (Histogram) metric;
+                    return Stream.of(metric,
+                        new Counter(h.getName() + "_count", h.getLabels(), h.getSampleCount(), h.getTimestamp()),
+                        new Counter(h.getName() + "_sum", h.getLabels(), h.getSampleSum(), h.getTimestamp()));
+                }
+                if (metric instanceof Summary) {
+                    Summary s = (Summary) metric;
+                    return Stream.of(metric,
+                        new Counter(s.getName() + "_count", s.getLabels(), s.getSampleCount(), s.getTimestamp()),
+                        new Counter(s.getName() + "_sum", s.getLabels(), s.getSampleSum(), s.getTimestamp()));
+                }
+                return Stream.of(metric);
+            })
+            .flatMap(metric ->
+                rules.stream()
+                    .flatMap(rule -> rule.getSources().entrySet().stream().map(source -> Tuple.of(rule, source.getKey(), source.getValue())))
+                    .filter(rule -> rule._2.equals(metric.getName()))
+                    .filter(rule -> metric.getLabels().keySet().containsAll(rule._3.getRelabel().labelKeys()))
+                    .filter(rule -> {
+                        if (Objects.isNull(rule._3.getLabelFilter())) {
+                            return true;
+                        }
+                        return rule._3.getLabelFilter().stream()
+                            .allMatch(matchRule -> matchRule.getOptions().contains(metric.getLabels().get(matchRule.getKey())));
+                    })
+                    .map(rule -> Tuple.of(rule._1, rule._2, rule._3, metric))
+            )
+            .peek(tuple -> log.debug("Mapped rules to metrics: {}", tuple))
+            .map(Function1.liftTry(tuple -> {
+                String serviceName = composeEntity(tuple._3.getRelabel().getService().stream(), tuple._4.getLabels());
+                Operation o = new Operation(tuple._1.getOperation(), tuple._1.getName(), tuple._1.getScope(), tuple._1.getPercentiles());
+                MetricSource.MetricSourceBuilder sb = MetricSource.builder();
+                sb.promMetricName(tuple._2)
+                    .timestamp(tuple._4.getTimestamp())
+                    .scale(tuple._3.getScale())
+                    .counterFunction(tuple._3.getCounterFunction())
+                    .range(tuple._3.getRange());
+                switch (tuple._1.getScope()) {
+                    case SERVICE:
+                        return Tuple.of(o, sb.entity(MeterEntity.newService(serviceName)).build(), tuple._4);
+                    case SERVICE_INSTANCE:
+                        String instanceName = composeEntity(tuple._3.getRelabel().getInstance().stream(), tuple._4.getLabels());
+                        return Tuple.of(o, sb.entity(MeterEntity.newServiceInstance(serviceName, instanceName)).build(), tuple._4);
+                    case ENDPOINT:
+                        String endpointName = composeEntity(tuple._3.getRelabel().getEndpoint().stream(), tuple._4.getLabels());
+                        return Tuple.of(o, sb.entity(MeterEntity.newEndpoint(serviceName, endpointName)).build(), tuple._4);
+                    default:
+                        throw new IllegalArgumentException("Unsupported scope" + tuple._1.getScope());
+                }
+            }))
+            .flatMap(tryIt -> PrometheusMetricConverter.log(tryIt, "Generated entity from labels"))
+            .collect(groupingBy(Tuple3::_1, groupingBy(Tuple3::_2, mapping(Tuple3::_3, toList()))))
+            .forEach((operation, sources) -> {
+                log.debug("Building metrics {} -> {}", operation, sources);
+                Try.run(() -> {
+                    switch (operation.getName()) {
+                        case AVG:
+                            sources.forEach((source, metrics) -> {
+                                AcceptableValue<Long> 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.setTimeBucket(TimeBucket.getMinuteTimeBucket(source.getTimestamp()));
+                                log.debug("Input metric {}", value.getTimeBucket());
+                                service.doStreamingCalculation(value);
+
+                                generateTraffic(source.getEntity());
+                            });
+                            break;
+                        case AVG_HISTOGRAM:
+                        case AVG_PERCENTILE:
+                            Validate.isTrue(sources.size() == 1, "Can't get source for histogram");
+                            Map.Entry<MetricSource, List<Metric>> smm = sources.entrySet().iterator().next();
+                            Histogram h = (Histogram) sum(smm.getValue());
+
+                            long[] vv = new long[h.getBuckets().size()];
+                            int[] bb = new int[h.getBuckets().size()];
+                            long v = 0L;
+                            int i = 0;
+                            for (Map.Entry<Double, Long> entry : h.getBuckets().entrySet()) {
+                                long increase = entry.getValue() - v;
+                                vv[i] = window.get(operation.getMetricName(), ImmutableMap.of("le", entry.getKey().toString()))
+                                    .apply(smm.getKey(), (double) increase).longValue();
+                                v = entry.getValue();
+
+                                if (i + 1 < h.getBuckets().size()) {
+                                    bb[i + 1] = BigDecimal.valueOf(entry.getKey()).multiply(SECOND_TO_MILLISECOND).intValue();
+                                }
+
+                                i++;
+                            }
+
+                            if (operation.getName().equals(AVG_HISTOGRAM)) {
+                                AcceptableValue<BucketedValues> heatmapMetrics = service.buildMetrics(
+                                    formatMetricName(operation.getMetricName()), BucketedValues.class);
+                                heatmapMetrics.setTimeBucket(TimeBucket.getMinuteTimeBucket(smm.getKey().getTimestamp()));
+                                heatmapMetrics.accept(smm.getKey().getEntity(), new BucketedValues(bb, vv));
+                                service.doStreamingCalculation(heatmapMetrics);
+                            } else {
+                                AcceptableValue<AvgHistogramPercentileFunction.AvgPercentileArgument> percentileMetrics =
+                                    service.buildMetrics(formatMetricName(operation.getMetricName()), AvgHistogramPercentileFunction.AvgPercentileArgument.class);
+                                percentileMetrics.setTimeBucket(TimeBucket.getMinuteTimeBucket(smm.getKey().getTimestamp()));
+                                percentileMetrics.accept(smm.getKey().getEntity(),
+                                    new AvgHistogramPercentileFunction.AvgPercentileArgument(new BucketedValues(bb, vv), operation.getPercentiles().stream().mapToInt(Integer::intValue).toArray()));
+                                service.doStreamingCalculation(percentileMetrics);
+                            }
+
+                            generateTraffic(smm.getKey().getEntity());
+                            break;
+                        default:
+                            throw new IllegalArgumentException(String.format("Unsupported downSampling %s", operation.getName()));
+                    }
+                }).onFailure(e -> log.debug("Building metric failed", e));
+            });
+    }
+
+    private String formatMetricName(String meterRuleName) {
+        StringJoiner metricName = new StringJoiner("_");
+        metricName.add("meter").add(meterRuleName);
+        return metricName.toString();
+    }
+
+    private String composeEntity(Stream<String> stream, Map<String, String> labels) {
+        return stream.map(key -> requireNonNull(labels.get(key), String.format("Getting %s from %s failed", key, labels)))
+            .collect(Collectors.joining("."));
+    }
+
+    private Metric sum(List<Metric> metrics) {
+        return metrics.stream().reduce(Metric::sum).orElseThrow(IllegalArgumentException::new);
+    }
+
+    private void generateTraffic(MeterEntity entity) {
+        ServiceTraffic s = new ServiceTraffic();
+        s.setName(requireNonNull(entity.getServiceName()));
+        s.setNodeType(NodeType.Normal);
+        s.setTimeBucket(TimeBucket.getMinuteTimeBucket(System.currentTimeMillis()));
+        MetricsStreamProcessor.getInstance().in(s);
+        if (!Strings.isNullOrEmpty(entity.getInstanceName())) {
+            InstanceTraffic instanceTraffic = new InstanceTraffic();
+            instanceTraffic.setName(entity.getInstanceName());
+            instanceTraffic.setServiceId(entity.serviceId());
+            instanceTraffic.setTimeBucket(TimeBucket.getMinuteTimeBucket(System.currentTimeMillis()));
+            instanceTraffic.setLastPingTimestamp(System.currentTimeMillis());
+            MetricsStreamProcessor.getInstance().in(instanceTraffic);
+        }
+        if (!Strings.isNullOrEmpty(entity.getEndpointName())) {
+            EndpointTraffic endpointTraffic = new EndpointTraffic();
+            endpointTraffic.setName(entity.getEndpointName());
+            endpointTraffic.setServiceId(entity.serviceId());
+            endpointTraffic.setTimeBucket(TimeBucket.getMinuteTimeBucket(System.currentTimeMillis()));
+            MetricsStreamProcessor.getInstance().in(endpointTraffic);
+        }
+    }
+
+    public static <T> Stream<T> log(Try<T> t, String debugMessage) {
+        return t
+            .onSuccess(i -> log.debug(debugMessage + " :{}", i))
+            .onFailure(e -> log.debug(debugMessage + " failed", e))
+            .toJavaStream();
+    }
+}

oap-server/server-fetcher-plugin/prometheus-fetcher-plugin/src/main/java/org/apache/skywalking/oap/server/fetcher/prometheus/provider/counter/ID.java → oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/metric/promethues/counter/ID.java

@@ -16,7 +16,7 @@
  *
  */
 
-package org.apache.skywalking.oap.server.fetcher.prometheus.provider.counter;
+package org.apache.skywalking.oap.server.core.metric.promethues.counter;
 
 import com.google.common.collect.ImmutableMap;
 import lombok.EqualsAndHashCode;

oap-server/server-fetcher-plugin/prometheus-fetcher-plugin/src/main/java/org/apache/skywalking/oap/server/fetcher/prometheus/provider/counter/Window.java → oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/metric/promethues/counter/Window.java

@@ -16,7 +16,7 @@
  *
  */
 
-package org.apache.skywalking.oap.server.fetcher.prometheus.provider.counter;
+package org.apache.skywalking.oap.server.core.metric.promethues.counter;
 
 import com.google.common.collect.ImmutableMap;
 import com.google.common.collect.Maps;
@@ -31,7 +31,7 @@ import java.util.Queue;
 import lombok.EqualsAndHashCode;
 import lombok.RequiredArgsConstructor;
 import lombok.ToString;
-import org.apache.skywalking.oap.server.fetcher.prometheus.provider.operation.MetricSource;
+import org.apache.skywalking.oap.server.core.metric.promethues.operation.MetricSource;
 
 /**
  * Window stores a series of counter samples in order to calculate the increase
@@ -57,28 +57,27 @@ public class Window {
         if (source.getCounterFunction() == null) {
             return sum;
         }
-        long now = System.currentTimeMillis();
+        long now = source.getTimestamp();
         switch (source.getCounterFunction()) {
             case INCREASE:
-                Tuple2<Long, Double> i = increase(sum, id, Duration.parse(source.getRange()).toMillis());
+                Tuple2<Long, Double> i = increase(sum, id, Duration.parse(source.getRange()).toMillis(), now);
                 return sum - i._2;
             case RATE:
-                i = increase(sum, id, Duration.parse(source.getRange()).toMillis());
+                i = increase(sum, id, Duration.parse(source.getRange()).toMillis(), now);
                 return (sum - i._2) / ((now - i._1) / 1000);
             case IRATE:
-                i = increase(sum, id, 0);
+                i = increase(sum, id, 0, now);
                 return (sum - i._2) / ((now - i._1) / 1000);
             default:
                 return sum;
         }
     }
 
-    private Tuple2<Long, Double> increase(Double value, ID id, long windowSize) {
+    private Tuple2<Long, Double> increase(Double value, ID id, long windowSize, long now) {
         if (!windows.containsKey(id)) {
             windows.put(id, new LinkedList<>());
         }
         Queue<Tuple2<Long, Double>> window = windows.get(id);
-        long now = System.currentTimeMillis();
         window.offer(Tuple.of(now, value));
         Tuple2<Long, Double> ps = window.element();
         if ((now - ps._1) >= windowSize) {

oap-server/server-fetcher-plugin/prometheus-fetcher-plugin/src/main/java/org/apache/skywalking/oap/server/fetcher/prometheus/provider/operation/MetricSource.java → oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/metric/promethues/operation/MetricSource.java

@@ -16,14 +16,14 @@
  *
  */
 
-package org.apache.skywalking.oap.server.fetcher.prometheus.provider.operation;
+package org.apache.skywalking.oap.server.core.metric.promethues.operation;
 
 import lombok.Builder;
 import lombok.EqualsAndHashCode;
 import lombok.Getter;
 import lombok.ToString;
 import org.apache.skywalking.oap.server.core.analysis.meter.MeterEntity;
-import org.apache.skywalking.oap.server.fetcher.prometheus.provider.rule.CounterFunction;
+import org.apache.skywalking.oap.server.core.metric.promethues.rule.CounterFunction;
 
 @EqualsAndHashCode
 @ToString
@@ -32,6 +32,8 @@ import org.apache.skywalking.oap.server.fetcher.prometheus.provider.rule.Counter
 public class MetricSource {
     private final String promMetricName;
 
+    private final long timestamp;
+
     private final MeterEntity entity;
 
     private final CounterFunction counterFunction;

oap-server/server-fetcher-plugin/prometheus-fetcher-plugin/src/main/java/org/apache/skywalking/oap/server/fetcher/prometheus/provider/operation/Operation.java → oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/metric/promethues/operation/Operation.java

@@ -16,7 +16,7 @@
  *
  */
 
-package org.apache.skywalking.oap.server.fetcher.prometheus.provider.operation;
+package org.apache.skywalking.oap.server.core.metric.promethues.operation;
 
 import java.util.List;
 import lombok.EqualsAndHashCode;

oap-server/server-fetcher-plugin/prometheus-fetcher-plugin/src/main/java/org/apache/skywalking/oap/server/fetcher/prometheus/provider/rule/CounterFunction.java → oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/metric/promethues/rule/CounterFunction.java

@@ -16,7 +16,7 @@
  *
  */
 
-package org.apache.skywalking.oap.server.fetcher.prometheus.provider.rule;
+package org.apache.skywalking.oap.server.core.metric.promethues.rule;
 
 public enum CounterFunction {
     INCREASE, RATE, IRATE

oap-server/server-fetcher-plugin/prometheus-fetcher-plugin/src/main/java/org/apache/skywalking/oap/server/fetcher/prometheus/provider/rule/LabelMatchRule.java → oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/metric/promethues/rule/LabelMatchRule.java

@@ -16,7 +16,7 @@
  *
  */
 
-package org.apache.skywalking.oap.server.fetcher.prometheus.provider.rule;
+package org.apache.skywalking.oap.server.core.metric.promethues.rule;
 
 import java.util.List;
 import lombok.Data;

oap-server/server-fetcher-plugin/prometheus-fetcher-plugin/src/main/java/org/apache/skywalking/oap/server/fetcher/prometheus/provider/rule/MetricsRule.java → oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/metric/promethues/rule/MetricsRule.java

@@ -16,7 +16,7 @@
  *
  */
 
-package org.apache.skywalking.oap.server.fetcher.prometheus.provider.rule;
+package org.apache.skywalking.oap.server.core.metric.promethues.rule;
 
 import java.util.List;
 import java.util.Map;

oap-server/server-fetcher-plugin/prometheus-fetcher-plugin/src/main/java/org/apache/skywalking/oap/server/fetcher/prometheus/provider/rule/PrometheusMetric.java → oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/metric/promethues/rule/PrometheusMetric.java

@@ -16,7 +16,7 @@
  *
  */
 
-package org.apache.skywalking.oap.server.fetcher.prometheus.provider.rule;
+package org.apache.skywalking.oap.server.core.metric.promethues.rule;
 
 import java.util.List;
 import lombok.Data;

oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/metric/promethues/rule/Relabel.java

+/*
+ * 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.metric.promethues.rule;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Optional;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+@Data
+@NoArgsConstructor
+public class Relabel {
+    private List<String> service;
+    private List<String> instance;
+    private List<String> endpoint;
+
+    public List<String> labelKeys() {
+        List<String> result = new ArrayList<>();
+        result.addAll(Optional.ofNullable(service).orElse(Collections.emptyList()));
+        result.addAll(Optional.ofNullable(instance).orElse(Collections.emptyList()));
+        result.addAll(Optional.ofNullable(endpoint).orElse(Collections.emptyList()));
+        return result;
+    }
+}

oap-server/server-fetcher-plugin/prometheus-fetcher-plugin/src/main/java/org/apache/skywalking/oap/server/fetcher/prometheus/provider/rule/Rule.java → oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/metric/promethues/rule/Rule.java

@@ -16,7 +16,7 @@
  *
  */
 
-package org.apache.skywalking.oap.server.fetcher.prometheus.provider.rule;
+package org.apache.skywalking.oap.server.core.metric.promethues.rule;
 
 import java.util.List;
 import lombok.Data;

oap-server/server-fetcher-plugin/prometheus-fetcher-plugin/src/main/java/org/apache/skywalking/oap/server/fetcher/prometheus/provider/rule/Rules.java → oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/metric/promethues/rule/Rules.java

@@ -16,7 +16,7 @@
  *
  */
 
-package org.apache.skywalking.oap.server.fetcher.prometheus.provider.rule;
+package org.apache.skywalking.oap.server.core.metric.promethues.rule;
 
 import java.io.File;
 import java.io.FileNotFoundException;

oap-server/server-fetcher-plugin/prometheus-fetcher-plugin/src/main/java/org/apache/skywalking/oap/server/fetcher/prometheus/provider/rule/StaticConfig.java → oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/metric/promethues/rule/StaticConfig.java

@@ -16,7 +16,7 @@
  *
  */
 
-package org.apache.skywalking.oap.server.fetcher.prometheus.provider.rule;
+package org.apache.skywalking.oap.server.core.metric.promethues.rule;
 
 import java.util.List;
 import java.util.Map;

oap-server/server-fetcher-plugin/prometheus-fetcher-plugin/src/main/java/org/apache/skywalking/oap/server/fetcher/prometheus/provider/PrometheusFetcherProvider.java

@@ -18,52 +18,27 @@
 
 package org.apache.skywalking.oap.server.fetcher.prometheus.provider;
 
-import com.google.common.collect.ImmutableMap;
 import com.google.common.collect.Maps;
 import io.vavr.CheckedFunction1;
-import io.vavr.Function1;
-import io.vavr.Tuple;
-import io.vavr.Tuple3;
-import io.vavr.control.Try;
-import java.math.BigDecimal;
 import java.time.Duration;
 import java.util.Collection;
-import java.util.Comparator;
 import java.util.LinkedList;
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
-import java.util.StringJoiner;
 import java.util.concurrent.Executors;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.TimeUnit;
-import java.util.concurrent.atomic.AtomicReference;
-import java.util.stream.Collectors;
-import java.util.stream.Stream;
 import okhttp3.OkHttpClient;
 import okhttp3.Request;
 import okhttp3.Response;
-import org.apache.commons.lang3.Validate;
 import org.apache.skywalking.oap.server.core.CoreModule;
-import org.apache.skywalking.oap.server.core.analysis.NodeType;
-import org.apache.skywalking.oap.server.core.analysis.TimeBucket;
-import org.apache.skywalking.oap.server.core.analysis.manual.endpoint.EndpointTraffic;
-import org.apache.skywalking.oap.server.core.analysis.manual.instance.InstanceTraffic;
-import org.apache.skywalking.oap.server.core.analysis.manual.service.ServiceTraffic;
-import org.apache.skywalking.oap.server.core.analysis.meter.MeterEntity;
 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.worker.MetricsStreamProcessor;
+import org.apache.skywalking.oap.server.core.metric.promethues.PrometheusMetricConverter;
+import org.apache.skywalking.oap.server.core.metric.promethues.rule.Rule;
+import org.apache.skywalking.oap.server.core.metric.promethues.rule.Rules;
+import org.apache.skywalking.oap.server.core.metric.promethues.rule.StaticConfig;
 import org.apache.skywalking.oap.server.fetcher.prometheus.module.PrometheusFetcherModule;
-import org.apache.skywalking.oap.server.fetcher.prometheus.provider.counter.Window;
-import org.apache.skywalking.oap.server.fetcher.prometheus.provider.operation.MetricSource;
-import org.apache.skywalking.oap.server.fetcher.prometheus.provider.operation.Operation;
-import org.apache.skywalking.oap.server.fetcher.prometheus.provider.rule.MetricsRule;
-import org.apache.skywalking.oap.server.fetcher.prometheus.provider.rule.Rule;
-import org.apache.skywalking.oap.server.fetcher.prometheus.provider.rule.Rules;
-import org.apache.skywalking.oap.server.fetcher.prometheus.provider.rule.StaticConfig;
 import org.apache.skywalking.oap.server.library.module.ModuleConfig;
 import org.apache.skywalking.oap.server.library.module.ModuleDefine;
 import org.apache.skywalking.oap.server.library.module.ModuleProvider;
@@ -71,32 +46,17 @@ import org.apache.skywalking.oap.server.library.module.ModuleStartException;
 import org.apache.skywalking.oap.server.library.module.ServiceNotProvidedException;
 import org.apache.skywalking.oap.server.library.util.prometheus.Parser;
 import org.apache.skywalking.oap.server.library.util.prometheus.Parsers;
-import org.apache.skywalking.oap.server.library.util.prometheus.metrics.Counter;
-import org.apache.skywalking.oap.server.library.util.prometheus.metrics.Histogram;
 import org.apache.skywalking.oap.server.library.util.prometheus.metrics.Metric;
 import org.apache.skywalking.oap.server.library.util.prometheus.metrics.MetricFamily;
-import org.apache.skywalking.oap.server.library.util.prometheus.metrics.MetricType;
-import org.apache.skywalking.oap.server.library.util.prometheus.metrics.Summary;
-import org.elasticsearch.common.Strings;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import static java.util.Objects.requireNonNull;
-import static java.util.stream.Collectors.groupingBy;
-import static java.util.stream.Collectors.mapping;
 import static java.util.stream.Collectors.toList;
 
 public class PrometheusFetcherProvider extends ModuleProvider {
     private static final Logger LOG = LoggerFactory.getLogger(PrometheusFetcherProvider.class);
 
-    private final static BigDecimal SECOND_TO_MILLISECOND  = BigDecimal.TEN.pow(3);
-
-    private final static String AVG_HISTOGRAM = "avgHistogram";
-
-    private final static String AVG_PERCENTILE = "avgHistogramPercentile";
-
-    private final static String AVG = "avg";
-
     private final PrometheusFetcherConfig config;
 
     private final OkHttpClient client = new OkHttpClient();
@@ -143,28 +103,18 @@ public class PrometheusFetcherProvider extends ModuleProvider {
             return;
         }
         final MeterSystem service = getManager().find(CoreModule.NAME).provider().getService(MeterSystem.class);
-
         rules.forEach(r -> {
-            final AtomicReference<String> lastRuleName = new AtomicReference<>();
-            r.getMetricsRules().stream().sorted(Comparator.comparing(MetricsRule::getName)).forEach(rule -> {
-                if (rule.getName().equals(lastRuleName.get())) {
-                    lastRuleName.set(rule.getName());
-                    return;
-                }
-                service.create(formatMetricName(rule.getName()), rule.getOperation(), rule.getScope());
-                lastRuleName.set(rule.getName());
-            });
             ses.scheduleAtFixedRate(new Runnable() {
 
-                private final Window window = new Window();
+                private final PrometheusMetricConverter converter = new PrometheusMetricConverter(r.getMetricsRules(), service);
 
                 @Override public void run() {
                     if (Objects.isNull(r.getStaticConfig())) {
                         return;
                     }
-                    long now = System.currentTimeMillis();
                     StaticConfig sc = r.getStaticConfig();
-                    sc.getTargets().stream()
+                    long now = System.currentTimeMillis();
+                    converter.toMeter(sc.getTargets().stream()
                         .map(CheckedFunction1.liftTry(url -> {
                             Request request = new Request.Builder()
                                 .url(String.format("http://%s%s", url, r.getMetricsPath().startsWith("/") ? r.getMetricsPath() : "/" + r.getMetricsPath()))
@@ -174,7 +124,7 @@ public class PrometheusFetcherProvider extends ModuleProvider {
                                 Parser p = Parsers.text(requireNonNull(response.body()).byteStream());
                                 MetricFamily mf;
 
-                                while ((mf = p.parse()) != null) {
+                                while ((mf = p.parse(now)) != null) {
                                     result.addAll(mf.getMetrics().stream()
                                         .peek(metric -> {
                                             Map<String, String> extraLabels = Maps.newHashMap(sc.getLabels());
@@ -187,123 +137,13 @@ public class PrometheusFetcherProvider extends ModuleProvider {
                                             });
                                         })
                                         .collect(toList()));
-                                    if (mf.getType() == MetricType.HISTOGRAM) {
-                                        Histogram h = (Histogram) mf.getMetrics().get(0);
-                                        result.add(new Counter(h.getName() + "_count", h.getLabels(), h.getSampleCount()));
-                                        result.add(new Counter(h.getName() + "_sum", h.getLabels(), h.getSampleSum()));
-                                    }
-                                    if (mf.getType() == MetricType.SUMMARY) {
-                                        Summary s = (Summary) mf.getMetrics().get(0);
-                                        result.add(new Counter(s.getName() + "_count", s.getLabels(), s.getSampleCount()));
-                                        result.add(new Counter(s.getName() + "_sum", s.getLabels(), s.getSampleSum()));
-                                    }
                                 }
                             }
                             return result;
                         }))
-                        .flatMap(tryIt -> PrometheusFetcherProvider.log(tryIt, "Load metric"))
-                        .flatMap(Collection::stream)
-                        .flatMap(metric ->
-                            r.getMetricsRules().stream()
-                                .flatMap(rule -> rule.getSources().entrySet().stream().map(source -> Tuple.of(rule, source.getKey(), source.getValue())))
-                                .filter(rule -> rule._2.equals(metric.getName()))
-                                .filter(rule -> {
-                                    if (Objects.isNull(rule._3.getLabelFilter())) {
-                                        return true;
-                                    }
-                                    return rule._3.getLabelFilter().stream()
-                                        .allMatch(matchRule -> matchRule.getOptions().contains(metric.getLabels().get(matchRule.getKey())));
-                                })
-                                .map(rule -> Tuple.of(rule._1, rule._2, rule._3, metric))
-                        )
-                        .peek(tuple -> LOG.debug("Mapped rules to metrics: {}", tuple))
-                        .map(Function1.liftTry(tuple -> {
-                            String serviceName = composeEntity(tuple._3.getRelabel().getService().stream(), tuple._4.getLabels());
-                            Operation o = new Operation(tuple._1.getOperation(), tuple._1.getName(), tuple._1.getScope(), tuple._1.getPercentiles());
-                            MetricSource.MetricSourceBuilder sb = MetricSource.builder();
-                            sb.promMetricName(tuple._2)
-                                .scale(tuple._3.getScale())
-                                .counterFunction(tuple._3.getCounterFunction())
-                                .range(tuple._3.getRange());
-                            switch (tuple._1.getScope()) {
-                                case SERVICE:
-                                    return Tuple.of(o, sb.entity(MeterEntity.newService(serviceName)).build(), tuple._4);
-                                case SERVICE_INSTANCE:
-                                    String instanceName = composeEntity(tuple._3.getRelabel().getInstance().stream(), tuple._4.getLabels());
-                                    return Tuple.of(o, sb.entity(MeterEntity.newServiceInstance(serviceName, instanceName)).build(), tuple._4);
-                                case ENDPOINT:
-                                    String endpointName = composeEntity(tuple._3.getRelabel().getEndpoint().stream(), tuple._4.getLabels());
-                                    return Tuple.of(o, sb.entity(MeterEntity.newEndpoint(serviceName, endpointName)).build(), tuple._4);
-                                default:
-                                    throw new IllegalArgumentException("Unsupported scope" + tuple._1.getScope());
-                            }
-                        }))
-                        .flatMap(tryIt -> PrometheusFetcherProvider.log(tryIt, "Generated entity from labels"))
-                        .collect(groupingBy(Tuple3::_1, groupingBy(Tuple3::_2, mapping(Tuple3::_3, toList()))))
-                        .forEach((operation, sources) -> {
-                            LOG.debug("Building metrics {} -> {}", operation, sources);
-                            Try.run(() -> {
-                                switch (operation.getName()) {
-                                    case AVG:
-                                        sources.forEach((source, metrics) -> {
-                                            AcceptableValue<Long> 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.setTimeBucket(TimeBucket.getMinuteTimeBucket(now));
-                                            LOG.debug("Input metric {}", value.getTimeBucket());
-                                            service.doStreamingCalculation(value);
-
-                                            generateTraffic(source.getEntity());
-                                        });
-                                        break;
-                                    case AVG_HISTOGRAM:
-                                    case AVG_PERCENTILE:
-                                        Validate.isTrue(sources.size() == 1, "Can't get source for histogram");
-                                        Map.Entry<MetricSource, List<Metric>> smm = sources.entrySet().iterator().next();
-                                        Histogram h = (Histogram) sum(smm.getValue());
-
-                                        long[] vv = new long[h.getBuckets().size()];
-                                        int[] bb = new int[h.getBuckets().size()];
-                                        long v = 0L;
-                                        int i = 0;
-                                        for (Map.Entry<Double, Long> entry : h.getBuckets().entrySet()) {
-                                            long increase = entry.getValue() - v;
-                                            vv[i] = window.get(operation.getMetricName(), ImmutableMap.of("le", entry.getKey().toString()))
-                                                .apply(smm.getKey(), (double) increase).longValue();
-                                            v = entry.getValue();
-
-                                            if (i + 1 < h.getBuckets().size()) {
-                                                bb[i + 1] = BigDecimal.valueOf(entry.getKey()).multiply(SECOND_TO_MILLISECOND).intValue();
-                                            }
-
-                                            i++;
-                                        }
-
-                                        if (operation.getName().equals(AVG_HISTOGRAM)) {
-                                            AcceptableValue<BucketedValues> heatmapMetrics = service.buildMetrics(
-                                                formatMetricName(operation.getMetricName()), BucketedValues.class);
-                                            heatmapMetrics.setTimeBucket(TimeBucket.getMinuteTimeBucket(now));
-                                            heatmapMetrics.accept(smm.getKey().getEntity(), new BucketedValues(bb, vv));
-                                            service.doStreamingCalculation(heatmapMetrics);
-                                        } else {
-                                            AcceptableValue<AvgHistogramPercentileFunction.AvgPercentileArgument> percentileMetrics =
-                                                service.buildMetrics(formatMetricName(operation.getMetricName()), AvgHistogramPercentileFunction.AvgPercentileArgument.class);
-                                            percentileMetrics.setTimeBucket(TimeBucket.getMinuteTimeBucket(now));
-                                            percentileMetrics.accept(smm.getKey().getEntity(),
-                                                new AvgHistogramPercentileFunction.AvgPercentileArgument(new BucketedValues(bb, vv), operation.getPercentiles().stream().mapToInt(Integer::intValue).toArray()));
-                                            service.doStreamingCalculation(percentileMetrics);
-                                        }
-
-                                        generateTraffic(smm.getKey().getEntity());
-                                        break;
-                                    default:
-                                        throw new IllegalArgumentException(String.format("Unsupported downSampling %s", operation.getName()));
-                                }
-                            }).onFailure(e -> LOG.debug("Building metric failed", e));
-                    });
-                }
+                        .flatMap(tryIt -> PrometheusMetricConverter.log(tryIt, "Load metric"))
+                        .flatMap(Collection::stream));
+                    }
             }, 0L, Duration.parse(r.getFetcherInterval()).getSeconds(), TimeUnit.SECONDS);
         });
     }
@@ -313,48 +153,4 @@ public class PrometheusFetcherProvider extends ModuleProvider {
         return new String[] {CoreModule.NAME};
     }
 
-    private String formatMetricName(String meterRuleName) {
-        StringJoiner metricName = new StringJoiner("_");
-        metricName.add("meter").add(meterRuleName);
-        return metricName.toString();
-    }
-
-    private String composeEntity(Stream<String> stream, Map<String, String> labels) {
-        return stream.map(key -> requireNonNull(labels.get(key), String.format("Getting %s from %s failed", key, labels)))
-            .collect(Collectors.joining("."));
-    }
-
-    private Metric sum(List<Metric> metrics) {
-        return metrics.stream().reduce(Metric::sum).orElseThrow(IllegalArgumentException::new);
-    }
-
-    private void generateTraffic(MeterEntity entity) {
-            ServiceTraffic s = new ServiceTraffic();
-            s.setName(requireNonNull(entity.getServiceName()));
-            s.setNodeType(NodeType.Normal);
-            s.setTimeBucket(TimeBucket.getMinuteTimeBucket(System.currentTimeMillis()));
-            MetricsStreamProcessor.getInstance().in(s);
-            if (!Strings.isNullOrEmpty(entity.getInstanceName())) {
-                InstanceTraffic instanceTraffic = new InstanceTraffic();
-                instanceTraffic.setName(entity.getInstanceName());
-                instanceTraffic.setServiceId(entity.serviceId());
-                instanceTraffic.setTimeBucket(TimeBucket.getMinuteTimeBucket(System.currentTimeMillis()));
-                instanceTraffic.setLastPingTimestamp(System.currentTimeMillis());
-                MetricsStreamProcessor.getInstance().in(instanceTraffic);
-            }
-            if (!Strings.isNullOrEmpty(entity.getEndpointName())) {
-                EndpointTraffic endpointTraffic = new EndpointTraffic();
-                endpointTraffic.setName(entity.getEndpointName());
-                endpointTraffic.setServiceId(entity.serviceId());
-                endpointTraffic.setTimeBucket(TimeBucket.getMinuteTimeBucket(System.currentTimeMillis()));
-                MetricsStreamProcessor.getInstance().in(endpointTraffic);
-            }
-    }
-
-    private static <T> Stream<T> log(Try<T> t, String debugMessage) {
-        return t
-            .onSuccess(i -> LOG.debug(debugMessage + " :{}", i))
-            .onFailure(e -> LOG.debug(debugMessage + " failed", e))
-            .toJavaStream();
-    }
 }

oap-server/server-fetcher-plugin/prometheus-fetcher-plugin/src/test/java/org/apache/skywalking/oap/server/fetcher/prometheus/provider/rule/RulesTest.java

@@ -19,6 +19,8 @@
 package org.apache.skywalking.oap.server.fetcher.prometheus.provider.rule;
 
 import java.util.List;
+import org.apache.skywalking.oap.server.core.metric.promethues.rule.Rule;
+import org.apache.skywalking.oap.server.core.metric.promethues.rule.Rules;
 import org.apache.skywalking.oap.server.library.module.ModuleStartException;
 import org.junit.Test;
 

oap-server/server-library/library-util/src/main/java/org/apache/skywalking/oap/server/library/util/prometheus/Parser.java

@@ -22,5 +22,5 @@ import java.io.IOException;
 import org.apache.skywalking.oap.server.library.util.prometheus.metrics.MetricFamily;
 
 public interface Parser {
-    MetricFamily parse() throws IOException;
+    MetricFamily parse(long now) throws IOException;
 }

oap-server/server-library/library-util/src/main/java/org/apache/skywalking/oap/server/library/util/prometheus/metrics/Counter.java

@@ -32,8 +32,8 @@ public class Counter extends Metric {
     private double value;
 
     @lombok.Builder
-    public Counter(String name, @Singular Map<String, String> labels, double value) {
-        super(name, labels);
+    public Counter(String name, @Singular Map<String, String> labels, double value, long timestamp) {
+        super(name, labels, timestamp);
         this.value = value;
     }
 

oap-server/server-library/library-util/src/main/java/org/apache/skywalking/oap/server/library/util/prometheus/metrics/Gauge.java

@@ -32,8 +32,8 @@ public class Gauge extends Metric {
     private double value;
 
     @lombok.Builder
-    public Gauge(String name, @Singular Map<String, String> labels, double value) {
-        super(name, labels);
+    public Gauge(String name, @Singular Map<String, String> labels, double value, long timestamp) {
+        super(name, labels, timestamp);
         this.value = value;
     }
 

oap-server/server-library/library-util/src/main/java/org/apache/skywalking/oap/server/library/util/prometheus/metrics/Histogram.java

@@ -38,8 +38,8 @@ public class Histogram extends Metric {
 
     @lombok.Builder
     public Histogram(String name, @Singular Map<String, String> labels, long sampleCount, double sampleSum,
-        @Singular Map<Double, Long> buckets) {
-        super(name, labels);
+        @Singular Map<Double, Long> buckets, long timestamp) {
+        super(name, labels, timestamp);
         getLabels().remove("le");
         this.sampleCount = sampleCount;
         this.sampleSum = sampleSum;

oap-server/server-library/library-util/src/main/java/org/apache/skywalking/oap/server/library/util/prometheus/metrics/Metric.java

@@ -31,10 +31,12 @@ public abstract class Metric {
 
     private final String name;
     private final Map<String, String> labels;
+    private final long timestamp;
 
-    protected Metric(String name, Map<String, String> labels) {
+    protected Metric(String name, Map<String, String> labels, long timestamp) {
         this.name = name;
         this.labels = Maps.newHashMap(labels);
+        this.timestamp = timestamp;
     }
 
     public abstract Metric sum(Metric m);

oap-server/server-library/library-util/src/main/java/org/apache/skywalking/oap/server/library/util/prometheus/metrics/Summary.java

@@ -35,8 +35,8 @@ public class Summary extends Metric {
 
     @lombok.Builder
     public Summary(String name, @Singular Map<String, String> labels, long sampleCount, double sampleSum,
-        @Singular Map<Double, Double> quantiles) {
-        super(name, labels);
+        @Singular Map<Double, Double> quantiles, long timestamp) {
+        super(name, labels, timestamp);
         getLabels().remove("quantile");
         this.sampleCount = sampleCount;
         this.sampleSum = sampleSum;

oap-server/server-library/library-util/src/main/java/org/apache/skywalking/oap/server/library/util/prometheus/parser/Context.java

@@ -22,6 +22,7 @@ import com.google.common.collect.Maps;
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Map;
+import lombok.RequiredArgsConstructor;
 import org.apache.commons.lang3.tuple.Pair;
 import org.apache.skywalking.oap.server.library.util.prometheus.metrics.Counter;
 import org.apache.skywalking.oap.server.library.util.prometheus.metrics.Gauge;
@@ -37,6 +38,7 @@ import static java.util.stream.Collectors.groupingBy;
 import static java.util.stream.Collectors.mapping;
 import static java.util.stream.Collectors.toList;
 
+@RequiredArgsConstructor
 public class Context {
     private static final Logger LOG = LoggerFactory.getLogger(Context.class);
     public MetricFamily metricFamily;
@@ -47,6 +49,8 @@ public class Context {
     public List<String> allowedNames = new ArrayList<>();
     public List<TextSample> samples = new ArrayList<>();
 
+    private final long now;
+
     void addAllowedNames(String type) {
         this.type = MetricType.valueOf(type.toUpperCase());
         allowedNames.clear();
@@ -96,6 +100,7 @@ public class Context {
                         .name(name)
                         .value(convertStringToDouble(textSample.getValue()))
                         .labels(textSample.getLabels())
+                        .timestamp(now)
                         .build()));
                 break;
             case COUNTER:
@@ -104,11 +109,12 @@ public class Context {
                         .name(name)
                         .value(convertStringToDouble(textSample.getValue()))
                         .labels(textSample.getLabels())
+                        .timestamp(now)
                         .build()));
                 break;
             case HISTOGRAM:
                 Histogram.HistogramBuilder hBuilder = Histogram.builder();
-                hBuilder.name(name);
+                hBuilder.name(name).timestamp(now);
                 samples.forEach(textSample -> {
                     if (textSample.getName().endsWith("_count")) {
                         hBuilder.sampleCount((long) convertStringToDouble(textSample.getValue()));
@@ -124,7 +130,6 @@ public class Context {
                 metricFamilyBuilder.addMetric(hBuilder.build());
                 break;
             case SUMMARY:
-
                 samples.stream()
                     .map(sample -> {
                         Map<String, String> labels = Maps.newHashMap(sample.getLabels());
@@ -134,7 +139,7 @@ public class Context {
                     .collect(groupingBy(Pair::getLeft, mapping(Pair::getRight, toList())))
                     .forEach((labels, samples) -> {
                         Summary.SummaryBuilder sBuilder = Summary.builder();
-                        sBuilder.name(name);
+                        sBuilder.name(name).timestamp(now);
                         sBuilder.labels(labels);
                         samples.forEach(textSample -> {
                             if (textSample.getName().endsWith("_count")) {

oap-server/server-library/library-util/src/main/java/org/apache/skywalking/oap/server/library/util/prometheus/parser/TextParser.java

@@ -42,7 +42,7 @@ public class TextParser implements Parser {
     }
 
     @Override
-    public MetricFamily parse() throws IOException {
+    public MetricFamily parse(long now) throws IOException {
         String line;
         if (lastLineReadFromStream != null) {
             line = lastLineReadFromStream;
@@ -54,7 +54,7 @@ public class TextParser implements Parser {
             return null;
         }
 
-        Context ctx = new Context();
+        Context ctx = new Context(now);
         while (line != null) {
             line = line.trim();
 

oap-server/server-library/library-util/src/test/java/org/apache/skywalking/oap/server/library/util/prometheus/parser/TextParserTest.java

@@ -39,8 +39,11 @@ public class TextParserTest {
 
     Queue<MetricFamily> expectedMfs = new LinkedList<>();
 
+    long now;
+
     @Before
     public void setup() {
+        now = System.currentTimeMillis();
         expectedMfs.offer(new MetricFamily.Builder()
             .setName("http_requests_total")
                               .setType(MetricType.COUNTER)
@@ -50,12 +53,14 @@ public class TextParserTest {
                                                 .label("method", "post")
                                                 .label("code", "200")
                                                 .value(1027D)
+                                                .timestamp(now)
                                                 .build())
                               .addMetric(Counter.builder()
                                                 .name("http_requests_total")
                                                 .label("method", "post")
                                                 .label("code", "400")
                                                 .value(3D)
+                                                .timestamp(now)
                                                 .build())
                               .build());
         expectedMfs.offer(new MetricFamily.Builder()
@@ -72,6 +77,7 @@ public class TextParserTest {
                                                   .bucket(0.5D, 129389L)
                                                   .bucket(1.0D, 133988L)
                                                   .bucket(Double.POSITIVE_INFINITY, 144320L)
+                                                  .timestamp(now)
                                                   .build())
                               .build());
         expectedMfs.offer(new MetricFamily.Builder()
@@ -87,6 +93,7 @@ public class TextParserTest {
                                              .quantile(0.5D, 4773D)
                                              .quantile(0.9D, 9001D)
                                              .quantile(0.99D, 76656D)
+                                             .timestamp(now)
                                              .build())
                               .build());
     }
@@ -97,7 +104,7 @@ public class TextParserTest {
             TextParser parser = new TextParser(is);
             MetricFamily mf;
             int mfNum = 0;
-            while ((mf = parser.parse()) != null) {
+            while ((mf = parser.parse(now)) != null) {
                 mfNum++;
                 MetricFamily expected = expectedMfs.poll();
                 assertNotNull(expected);

oap-server/server-receiver-plugin/opencensus-receiver-plugin/pom.xml

+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+  ~ 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.
+  ~
+  -->
+
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <parent>
+        <artifactId>server-receiver-plugin</artifactId>
+        <groupId>org.apache.skywalking</groupId>
+        <version>8.1.0-SNAPSHOT</version>
+    </parent>
+    <modelVersion>4.0.0</modelVersion>
+
+    <artifactId>opencensus-receiver-plugin</artifactId>
+    <packaging>jar</packaging>
+
+    <dependencies>
+        <dependency>
+            <groupId>org.apache.skywalking</groupId>
+            <artifactId>receiver-proto</artifactId>
+            <version>${project.version}</version>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.skywalking</groupId>
+            <artifactId>skywalking-sharing-server-plugin</artifactId>
+            <version>${project.version}</version>
+        </dependency>
+    </dependencies>
+
+</project>
\ No newline at end of file

oap-server/server-receiver-plugin/opencensus-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/opencensus/OCMetricHandler.java

+/*
+ * 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.receiver.opencensus;
+
+import com.google.protobuf.Timestamp;
+import io.grpc.stub.StreamObserver;
+import io.opencensus.proto.agent.metrics.v1.ExportMetricsServiceRequest;
+import io.opencensus.proto.agent.metrics.v1.ExportMetricsServiceResponse;
+import io.opencensus.proto.agent.metrics.v1.MetricsServiceGrpc;
+import io.opencensus.proto.metrics.v1.DistributionValue;
+import io.opencensus.proto.metrics.v1.LabelKey;
+import io.opencensus.proto.metrics.v1.LabelValue;
+import io.opencensus.proto.metrics.v1.SummaryValue;
+import io.vavr.Function1;
+import io.vavr.Tuple;
+import java.time.Instant;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.skywalking.oap.server.core.metric.promethues.PrometheusMetricConverter;
+import org.apache.skywalking.oap.server.library.util.prometheus.metrics.Counter;
+import org.apache.skywalking.oap.server.library.util.prometheus.metrics.Gauge;
+import org.apache.skywalking.oap.server.library.util.prometheus.metrics.Histogram;
+import org.apache.skywalking.oap.server.library.util.prometheus.metrics.Summary;
+
+@RequiredArgsConstructor
+@Slf4j
+public class OCMetricHandler extends MetricsServiceGrpc.MetricsServiceImplBase {
+
+    private final PrometheusMetricConverter prometheusMetric;
+
+    @Override public StreamObserver<ExportMetricsServiceRequest> export(
+        StreamObserver<ExportMetricsServiceResponse> responseObserver) {
+        return new StreamObserver<ExportMetricsServiceRequest>() {
+            @Override public void onNext(ExportMetricsServiceRequest request) {
+                prometheusMetric.toMeter(request.getMetricsList().stream()
+                    .flatMap(metric -> metric.getTimeseriesList().stream().map(timeSeries ->
+                        Tuple.of(metric.getMetricDescriptor(),
+                            buildLabels(metric.getMetricDescriptor().getLabelKeysList(), timeSeries.getLabelValuesList()),
+                            timeSeries)))
+                    .flatMap(t -> t._3.getPointsList().stream().map(point -> Tuple.of(t._1, t._2, point)))
+                    .map(Function1.liftTry(t -> {
+                        switch (t._1.getType()) {
+                            case GAUGE_INT64:
+                                return new Gauge(t._1.getName(), t._2, t._3.getInt64Value(), tsToMilli(t._3.getTimestamp()));
+                            case GAUGE_DOUBLE:
+                                return new Gauge(t._1.getName(), t._2, t._3.getDoubleValue(), tsToMilli(t._3.getTimestamp()));
+                            case CUMULATIVE_INT64:
+                                return new Counter(t._1.getName(), t._2, t._3.getInt64Value(), tsToMilli(t._3.getTimestamp()));
+                            case CUMULATIVE_DOUBLE:
+                                return new Counter(t._1.getName(), t._2, t._3.getDoubleValue(), tsToMilli(t._3.getTimestamp()));
+                            case CUMULATIVE_DISTRIBUTION:
+                                return new Histogram(t._1.getName(), t._2, t._3.getDistributionValue().getCount(),
+                                    t._3.getDistributionValue().getSum(),
+                                    buildBuckets(t._3.getDistributionValue()), tsToMilli(t._3.getTimestamp()));
+                            case SUMMARY:
+                                return new Summary(t._1.getName(), t._2, t._3.getSummaryValue().getCount().getValue(),
+                                    t._3.getSummaryValue().getSum().getValue(),
+                                    buildQuantiles(t._3.getSummaryValue().getSnapshot()), tsToMilli(t._3.getTimestamp()));
+                            default:
+                                throw new UnsupportedOperationException("Unsupported OC type:" + t._1.getType());
+                        }
+                    }))
+                    .flatMap(tryIt -> PrometheusMetricConverter.log(tryIt, "Convert OC metric to prometheus metric")));
+            }
+
+            @Override public void onError(Throwable throwable) {
+
+            }
+
+            @Override public void onCompleted() {
+                responseObserver.onCompleted();
+            }
+        };
+    }
+
+    private static Map<String, String> buildLabels(List<LabelKey> keys, List<LabelValue> values) {
+        Map<String, String> result = new HashMap<>();
+        for (int i = 0; i < keys.size(); i++) {
+            result.put(keys.get(i).getKey(), values.get(i).getValue());
+        }
+        return result;
+    }
+
+    private static Map<Double, Long> buildBuckets(DistributionValue distributionValue) {
+        Map<Double, Long> result = new HashMap<>();
+        List<Double> bounds = distributionValue.getBucketOptions().getExplicit().getBoundsList();
+        for (int i = 0; i < bounds.size(); i++) {
+            result.put(bounds.get(i), distributionValue.getBuckets(i).getCount());
+        }
+        result.put(Double.POSITIVE_INFINITY, distributionValue.getBuckets(bounds.size()).getCount());
+        return result;
+    }
+
+    private static Map<Double, Double> buildQuantiles(SummaryValue.Snapshot snapshot) {
+        Map<Double, Double> result = new HashMap<>();
+        snapshot.getPercentileValuesList().forEach(p -> result.put(p.getPercentile(), p.getValue()));
+        return result;
+    }
+
+    private static long tsToMilli(Timestamp timestamp) {
+        return timestamp.equals(Timestamp.getDefaultInstance()) ? System.currentTimeMillis() :
+            Instant.ofEpochSecond(timestamp.getSeconds(), timestamp.getNanos()).toEpochMilli();
+    }
+
+}

oap-server/server-receiver-plugin/opencensus-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/opencensus/OCMetricReceiverConfig.java

+/*
+ * 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.receiver.opencensus;
+
+import lombok.Getter;
+import lombok.Setter;
+import org.apache.skywalking.oap.server.library.module.ModuleConfig;
+
+@Getter
+@Setter
+public class OCMetricReceiverConfig extends ModuleConfig {
+    private String gRPCHost = "0.0.0.0";
+    private int gRPCPort = -1;
+    private int maxConcurrentCallsPerConnection;
+    private int maxMessageSize;
+    private int gRPCThreadPoolSize;
+    private int gRPCThreadPoolQueueSize;
+    private String rulePath = "oc-rules";
+}

oap-server/server-fetcher-plugin/prometheus-fetcher-plugin/src/main/java/org/apache/skywalking/oap/server/fetcher/prometheus/provider/rule/Relabel.java → oap-server/server-receiver-plugin/opencensus-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/opencensus/OCMetricReceiverModule.java

@@ -16,16 +16,19 @@
  *
  */
 
-package org.apache.skywalking.oap.server.fetcher.prometheus.provider.rule;
+package org.apache.skywalking.oap.server.receiver.opencensus;
 
-import java.util.List;
-import lombok.Data;
-import lombok.NoArgsConstructor;
+import org.apache.skywalking.oap.server.library.module.ModuleDefine;
 
-@Data
-@NoArgsConstructor
-public class Relabel {
-    private List<String> service;
-    private List<String> instance;
-    private List<String> endpoint;
+public class OCMetricReceiverModule extends ModuleDefine {
+    public static final String NAME = "receiver-oc";
+
+    public OCMetricReceiverModule() {
+        super(NAME);
+    }
+
+    @Override
+    public Class[] services() {
+        return new Class[0];
+    }
 }

oap-server/server-receiver-plugin/opencensus-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/opencensus/OCMetricReceiverProvider.java

+/*
+ * 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.receiver.opencensus;
+
+import java.util.List;
+import java.util.Objects;
+import java.util.stream.Collectors;
+import org.apache.skywalking.oap.server.core.CoreModule;
+import org.apache.skywalking.oap.server.core.analysis.meter.MeterSystem;
+import org.apache.skywalking.oap.server.core.metric.promethues.PrometheusMetricConverter;
+import org.apache.skywalking.oap.server.core.metric.promethues.rule.MetricsRule;
+import org.apache.skywalking.oap.server.core.metric.promethues.rule.Rules;
+import org.apache.skywalking.oap.server.library.module.ModuleConfig;
+import org.apache.skywalking.oap.server.library.module.ModuleDefine;
+import org.apache.skywalking.oap.server.library.module.ModuleProvider;
+import org.apache.skywalking.oap.server.library.module.ModuleStartException;
+import org.apache.skywalking.oap.server.library.module.ServiceNotProvidedException;
+import org.apache.skywalking.oap.server.library.server.ServerException;
+import org.apache.skywalking.oap.server.library.server.grpc.GRPCServer;
+import org.apache.skywalking.oap.server.receiver.sharing.server.SharingServerModule;
+
+public class OCMetricReceiverProvider extends ModuleProvider {
+    public static final String NAME = "default";
+    private OCMetricReceiverConfig config;
+    private GRPCServer grpcServer = null;
+    private List<MetricsRule> rules;
+
+    @Override
+    public String name() {
+        return NAME;
+    }
+
+    @Override
+    public Class<? extends ModuleDefine> module() {
+        return OCMetricReceiverModule.class;
+    }
+
+    @Override
+    public ModuleConfig createConfigBeanIfAbsent() {
+        config = new OCMetricReceiverConfig();
+        return config;
+    }
+
+    @Override
+    public void prepare() throws ServiceNotProvidedException, ModuleStartException {
+        if (config.getGRPCPort() <= 0) {
+           return;
+        }
+        rules = Rules.loadRules(config.getRulePath()).stream()
+            .flatMap(rule -> rule.getMetricsRules().stream())
+            .collect(Collectors.toList());
+        grpcServer = new GRPCServer(config.getGRPCHost(), config.getGRPCPort());
+        if (config.getMaxMessageSize() > 0) {
+            grpcServer.setMaxMessageSize(config.getMaxMessageSize());
+        }
+        if (config.getMaxConcurrentCallsPerConnection() > 0) {
+            grpcServer.setMaxConcurrentCallsPerConnection(config.getMaxConcurrentCallsPerConnection());
+        }
+        if (config.getGRPCThreadPoolQueueSize() > 0) {
+            grpcServer.setThreadPoolQueueSize(config.getGRPCThreadPoolQueueSize());
+        }
+        if (config.getGRPCThreadPoolSize() > 0) {
+            grpcServer.setThreadPoolSize(config.getGRPCThreadPoolSize());
+        }
+        grpcServer.initialize();
+    }
+
+    @Override
+    public void start() throws ServiceNotProvidedException, ModuleStartException {
+        if (Objects.nonNull(grpcServer)) {
+            final MeterSystem service = getManager().find(CoreModule.NAME).provider().getService(MeterSystem.class);
+            grpcServer.addHandler(new OCMetricHandler(new PrometheusMetricConverter(rules, service)));
+        }
+    }
+
+    @Override
+    public void notifyAfterCompleted() throws ServiceNotProvidedException, ModuleStartException {
+        try {
+            if (Objects.nonNull(grpcServer)) {
+                grpcServer.start();
+            }
+        } catch (ServerException e) {
+            throw new ModuleStartException(e.getMessage(), e);
+        }
+    }
+
+    @Override
+    public String[] requiredModules() {
+        return new String[] {SharingServerModule.NAME};
+    }
+}

oap-server/server-receiver-plugin/opencensus-receiver-plugin/src/main/resources/META-INF/services/org.apache.skywalking.oap.server.library.module.ModuleDefine

+#
+# 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.
+#
+#
+
+
+org.apache.skywalking.oap.server.receiver.opencensus.OCMetricReceiverModule
\ No newline at end of file

oap-server/server-receiver-plugin/opencensus-receiver-plugin/src/main/resources/META-INF/services/org.apache.skywalking.oap.server.library.module.ModuleProvider

+#
+# 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.
+#
+#
+
+org.apache.skywalking.oap.server.receiver.opencensus.OCMetricReceiverProvider
\ No newline at end of file

oap-server/server-receiver-plugin/pom.xml

@@ -40,6 +40,7 @@
         <module>jaeger-receiver-plugin</module>
         <module>receiver-proto</module>
         <module>skywalking-profile-receiver-plugin</module>
+        <module>opencensus-receiver-plugin</module>
     </modules>
 
     <dependencies>

oap-server/server-receiver-plugin/receiver-proto/src/main/proto/opencensus/proto/agent/common/v1/common.proto

+// Copyright 2018, OpenCensus Authors
+//
+// Licensed 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.
+
+syntax = "proto3";
+
+// NOTE: This proto is experimental and is subject to change at this point.
+// Please do not use it at the moment.
+
+package opencensus.proto.agent.common.v1;
+
+import "google/protobuf/timestamp.proto";
+
+option java_multiple_files = true;
+option java_package = "io.opencensus.proto.agent.common.v1";
+option java_outer_classname = "CommonProto";
+
+option go_package = "github.com/census-instrumentation/opencensus-proto/gen-go/agent/common/v1";
+
+// Identifier metadata of the Node that produces the span or tracing data.
+// Note, this is not the metadata about the Node or service that is described by associated spans.
+// In the future we plan to extend the identifier proto definition to support
+// additional information (e.g cloud id, etc.)
+message Node {
+  // Identifier that uniquely identifies a process within a VM/container.
+  ProcessIdentifier identifier = 1;
+
+  // Information on the OpenCensus Library that initiates the stream.
+  LibraryInfo library_info = 2;
+
+  // Additional information on service.
+  ServiceInfo service_info = 3;
+
+  // Additional attributes.
+  map<string, string> attributes = 4;
+
+  // TODO(songya): Add more identifiers in the future as needed, like cloud
+  // identifiers.
+}
+
+// Identifier that uniquely identifies a process within a VM/container.
+message ProcessIdentifier {
+
+  // The host name. Usually refers to the machine/container name.
+  // For example: os.Hostname() in Go, socket.gethostname() in Python.
+  string host_name = 1;
+
+  // Process id.
+  uint32 pid = 2;
+
+  // Start time of this ProcessIdentifier. Represented in epoch time.
+  google.protobuf.Timestamp start_timestamp = 3;
+}
+
+// Information on OpenCensus Library.
+message LibraryInfo {
+
+  enum Language {
+    LANGUAGE_UNSPECIFIED = 0;
+    CPP = 1;
+    C_SHARP = 2;
+    ERLANG = 3;
+    GO_LANG = 4;
+    JAVA = 5;
+    NODE_JS = 6;
+    PHP = 7;
+    PYTHON = 8;
+    RUBY = 9;
+    WEB_JS = 10;
+  }
+
+  // Language of OpenCensus Library.
+  Language language = 1;
+
+  // Version of Agent exporter of Library.
+  string exporter_version = 2;
+
+  // Version of OpenCensus Library.
+  string core_library_version = 3;
+}
+
+// Additional service information.
+message ServiceInfo {
+
+  // Name of the service.
+  string name = 1;
+
+  // TODO(songya): add more fields as needed.
+}

oap-server/server-receiver-plugin/receiver-proto/src/main/proto/opencensus/proto/agent/metrics/v1/metrics_service.proto

+// Copyright 2018, OpenCensus Authors
+//
+// Licensed 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.
+
+syntax = "proto3";
+
+package opencensus.proto.agent.metrics.v1;
+
+import "opencensus/proto/agent/common/v1/common.proto";
+import "opencensus/proto/metrics/v1/metrics.proto";
+import "opencensus/proto/resource/v1/resource.proto";
+
+option java_multiple_files = true;
+option java_package = "io.opencensus.proto.agent.metrics.v1";
+option java_outer_classname = "MetricsServiceProto";
+
+option go_package = "github.com/census-instrumentation/opencensus-proto/gen-go/agent/metrics/v1";
+
+// Service that can be used to push metrics between one Application
+// instrumented with OpenCensus and an agent, or between an agent and a
+// central collector.
+service MetricsService {
+  // For performance reasons, it is recommended to keep this RPC
+  // alive for the entire life of the application.
+  rpc Export(stream ExportMetricsServiceRequest) returns (stream ExportMetricsServiceResponse) {}
+}
+
+message ExportMetricsServiceRequest {
+  // This is required only in the first message on the stream or if the
+  // previous sent ExportMetricsServiceRequest message has a different Node (e.g.
+  // when the same RPC is used to send Metrics from multiple Applications).
+  opencensus.proto.agent.common.v1.Node node = 1;
+
+  // A list of metrics that belong to the last received Node.
+  repeated opencensus.proto.metrics.v1.Metric metrics = 2;
+
+  // The resource for the metrics in this message that do not have an explicit
+  // resource set.
+  // If unset, the most recently set resource in the RPC stream applies. It is
+  // valid to never be set within a stream, e.g. when no resource info is known
+  // at all or when all sent metrics have an explicit resource set.
+  opencensus.proto.resource.v1.Resource resource = 3;
+}
+
+message ExportMetricsServiceResponse {
+}

oap-server/server-receiver-plugin/receiver-proto/src/main/proto/opencensus/proto/agent/trace/v1/trace_service.proto

+// Copyright 2018, OpenCensus Authors
+//
+// Licensed 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.
+
+syntax = "proto3";
+
+// NOTE: This proto is experimental and is subject to change at this point.
+// Please do not use it at the moment.
+
+package opencensus.proto.agent.trace.v1;
+
+import "opencensus/proto/agent/common/v1/common.proto";
+import "opencensus/proto/resource/v1/resource.proto";
+import "opencensus/proto/trace/v1/trace.proto";
+import "opencensus/proto/trace/v1/trace_config.proto";
+
+option java_multiple_files = true;
+option java_package = "io.opencensus.proto.agent.trace.v1";
+option java_outer_classname = "TraceServiceProto";
+
+option go_package = "github.com/census-instrumentation/opencensus-proto/gen-go/agent/trace/v1";
+
+// Service that can be used to push spans and configs between one Application
+// instrumented with OpenCensus and an agent, or between an agent and a
+// central collector or config service (in this case spans and configs are
+// sent/received to/from multiple Applications).
+service TraceService {
+  // After initialization, this RPC must be kept alive for the entire life of
+  // the application. The agent pushes configs down to applications via a
+  // stream.
+  rpc Config(stream CurrentLibraryConfig) returns (stream UpdatedLibraryConfig) {}
+
+  // For performance reasons, it is recommended to keep this RPC
+  // alive for the entire life of the application.
+  rpc Export(stream ExportTraceServiceRequest) returns (stream ExportTraceServiceResponse) {}
+}
+
+message CurrentLibraryConfig {
+  // This is required only in the first message on the stream or if the
+  // previous sent CurrentLibraryConfig message has a different Node (e.g.
+  // when the same RPC is used to configure multiple Applications).
+  opencensus.proto.agent.common.v1.Node node = 1;
+
+  // Current configuration.
+  opencensus.proto.trace.v1.TraceConfig config = 2;
+}
+
+message UpdatedLibraryConfig {
+  // This field is ignored when the RPC is used to configure only one Application.
+  // This is required only in the first message on the stream or if the
+  // previous sent UpdatedLibraryConfig message has a different Node.
+  opencensus.proto.agent.common.v1.Node node = 1;
+
+  // Requested updated configuration.
+  opencensus.proto.trace.v1.TraceConfig config = 2;
+}
+
+message ExportTraceServiceRequest {
+  // This is required only in the first message on the stream or if the
+  // previous sent ExportTraceServiceRequest message has a different Node (e.g.
+  // when the same RPC is used to send Spans from multiple Applications).
+  opencensus.proto.agent.common.v1.Node node = 1;
+
+  // A list of Spans that belong to the last received Node.
+  repeated opencensus.proto.trace.v1.Span spans = 2;
+
+  // The resource for the spans in this message that do not have an explicit
+  // resource set.
+  // If unset, the most recently set resource in the RPC stream applies. It is
+  // valid to never be set within a stream, e.g. when no resource info is known.
+  opencensus.proto.resource.v1.Resource resource = 3;
+}
+
+message ExportTraceServiceResponse {
+}

oap-server/server-receiver-plugin/receiver-proto/src/main/proto/opencensus/proto/metrics/v1/metrics.proto

+// Copyright 2018, OpenCensus Authors
+//
+// Licensed 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.
+
+// This package describes the Metrics data model. It is currently experimental
+// but may eventually become the wire format for metrics. Please see
+// https://github.com/census-instrumentation/opencensus-specs/blob/master/stats/Metrics.md
+// for more details.
+
+syntax = "proto3";
+
+package opencensus.proto.metrics.v1;
+
+import "google/protobuf/timestamp.proto";
+import "google/protobuf/wrappers.proto";
+import "opencensus/proto/resource/v1/resource.proto";
+
+option go_package = "github.com/census-instrumentation/opencensus-proto/gen-go/metrics/v1";
+
+option java_multiple_files = true;
+option java_package = "io.opencensus.proto.metrics.v1";
+option java_outer_classname = "MetricsProto";
+
+// Defines a Metric which has one or more timeseries.
+message Metric {
+  // The descriptor of the Metric.
+  // TODO(issue #152): consider only sending the name of descriptor for
+  // optimization.
+  MetricDescriptor metric_descriptor = 1;
+
+  // One or more timeseries for a single metric, where each timeseries has
+  // one or more points.
+  repeated TimeSeries timeseries = 2;
+
+  // The resource for the metric. If unset, it may be set to a default value
+  // provided for a sequence of messages in an RPC stream.
+  opencensus.proto.resource.v1.Resource resource = 3;
+}
+
+// Defines a metric type and its schema.
+message MetricDescriptor {
+  // The metric type, including its DNS name prefix. It must be unique.
+  string name = 1;
+
+  // A detailed description of the metric, which can be used in documentation.
+  string description = 2;
+
+  // The unit in which the metric value is reported. Follows the format
+  // described by http://unitsofmeasure.org/ucum.html.
+  string unit = 3;
+
+  // The kind of metric. It describes how the data is reported.
+  //
+  // A gauge is an instantaneous measurement of a value.
+  //
+  // A cumulative measurement is a value accumulated over a time interval. In
+  // a time series, cumulative measurements should have the same start time,
+  // increasing values and increasing end times, until an event resets the
+  // cumulative value to zero and sets a new start time for the following
+  // points.
+  enum Type {
+    // Do not use this default value.
+    UNSPECIFIED = 0;
+
+    // Integer gauge. The value can go both up and down.
+    GAUGE_INT64 = 1;
+
+    // Floating point gauge. The value can go both up and down.
+    GAUGE_DOUBLE = 2;
+
+    // Distribution gauge measurement. The count and sum can go both up and
+    // down. Recorded values are always >= 0.
+    // Used in scenarios like a snapshot of time the current items in a queue
+    // have spent there.
+    GAUGE_DISTRIBUTION = 3;
+
+    // Integer cumulative measurement. The value cannot decrease, if resets
+    // then the start_time should also be reset.
+    CUMULATIVE_INT64 = 4;
+
+    // Floating point cumulative measurement. The value cannot decrease, if
+    // resets then the start_time should also be reset. Recorded values are
+    // always >= 0.
+    CUMULATIVE_DOUBLE = 5;
+
+    // Distribution cumulative measurement. The count and sum cannot decrease,
+    // if resets then the start_time should also be reset.
+    CUMULATIVE_DISTRIBUTION = 6;
+
+    // Some frameworks implemented Histograms as a summary of observations
+    // (usually things like request durations and response sizes). While it
+    // also provides a total count of observations and a sum of all observed
+    // values, it calculates configurable percentiles over a sliding time
+    // window. This is not recommended, since it cannot be aggregated.
+    SUMMARY = 7;
+  }
+  Type type = 4;
+
+  // The label keys associated with the metric descriptor.
+  repeated LabelKey label_keys = 5;
+}
+
+// Defines a label key associated with a metric descriptor.
+message LabelKey {
+  // The key for the label.
+  string key = 1;
+
+  // A human-readable description of what this label key represents.
+  string description = 2;
+}
+
+// A collection of data points that describes the time-varying values
+// of a metric.
+message TimeSeries {
+  // Must be present for cumulative metrics. The time when the cumulative value
+  // was reset to zero. Exclusive. The cumulative value is over the time interval
+  // (start_timestamp, timestamp]. If not specified, the backend can use the
+  // previous recorded value.
+  google.protobuf.Timestamp start_timestamp = 1;
+
+  // The set of label values that uniquely identify this timeseries. Applies to
+  // all points. The order of label values must match that of label keys in the
+  // metric descriptor.
+  repeated LabelValue label_values = 2;
+
+  // The data points of this timeseries. Point.value type MUST match the
+  // MetricDescriptor.type.
+  repeated Point points = 3;
+}
+
+message LabelValue {
+  // The value for the label.
+  string value = 1;
+  // If false the value field is ignored and considered not set.
+  // This is used to differentiate a missing label from an empty string.
+  bool has_value = 2;
+}
+
+// A timestamped measurement.
+message Point {
+  // The moment when this point was recorded. Inclusive.
+  // If not specified, the timestamp will be decided by the backend.
+  google.protobuf.Timestamp timestamp = 1;
+
+  // The actual point value.
+  oneof value {
+    // A 64-bit integer.
+    int64 int64_value = 2;
+
+    // A 64-bit double-precision floating-point number.
+    double double_value = 3;
+
+    // A distribution value.
+    DistributionValue distribution_value = 4;
+
+    // A summary value. This is not recommended, since it cannot be aggregated.
+    SummaryValue summary_value = 5;
+  }
+}
+
+// Distribution contains summary statistics for a population of values. It
+// optionally contains a histogram representing the distribution of those
+// values across a set of buckets.
+message DistributionValue {
+  // The number of values in the population. Must be non-negative. This value
+  // must equal the sum of the values in bucket_counts if a histogram is
+  // provided.
+  int64 count = 1;
+
+  // The sum of the values in the population. If count is zero then this field
+  // must be zero.
+  double sum = 2;
+
+  // The sum of squared deviations from the mean of the values in the
+  // population. For values x_i this is:
+  //
+  //     Sum[i=1..n]((x_i - mean)^2)
+  //
+  // Knuth, "The Art of Computer Programming", Vol. 2, page 323, 3rd edition
+  // describes Welford's method for accumulating this sum in one pass.
+  //
+  // If count is zero then this field must be zero.
+  double sum_of_squared_deviation = 3;
+
+  // A Distribution may optionally contain a histogram of the values in the
+  // population. The bucket boundaries for that histogram are described by
+  // BucketOptions.
+  //
+  // If bucket_options has no type, then there is no histogram associated with
+  // the Distribution.
+  message BucketOptions {
+    oneof type {
+      // Bucket with explicit bounds.
+      Explicit explicit = 1;
+    }
+
+    // Specifies a set of buckets with arbitrary upper-bounds.
+    // This defines size(bounds) + 1 (= N) buckets. The boundaries for bucket
+    // index i are:
+    //
+    // [0, bucket_bounds[i]) for i == 0
+    // [bucket_bounds[i-1], bucket_bounds[i]) for 0 < i < N-1
+    // [bucket_bounds[i], +infinity) for i == N-1
+    message Explicit {
+      // The values must be strictly increasing and > 0.
+      repeated double bounds = 1;
+    }
+
+    // TODO: If OpenMetrics decides to support (a, b] intervals we should add
+    // support for these by defining a boolean value here which decides what
+    // type of intervals to use.
+  }
+
+  // Don't change bucket boundaries within a TimeSeries if your backend doesn't
+  // support this.
+  // TODO(issue #152): consider not required to send bucket options for
+  // optimization.
+  BucketOptions bucket_options = 4;
+
+  message Bucket {
+    // The number of values in each bucket of the histogram, as described in
+    // bucket_bounds.
+    int64 count = 1;
+
+    // If the distribution does not have a histogram, then omit this field.
+    Exemplar exemplar = 2;
+  }
+
+  // If the distribution does not have a histogram, then omit this field.
+  // If there is a histogram, then the sum of the values in the Bucket counts
+  // must equal the value in the count field of the distribution.
+  repeated Bucket buckets = 5;
+
+  // Exemplars are example points that may be used to annotate aggregated
+  // Distribution values. They are metadata that gives information about a
+  // particular value added to a Distribution bucket.
+  message Exemplar {
+    // Value of the exemplar point. It determines which bucket the exemplar
+    // belongs to.
+    double value = 1;
+
+    // The observation (sampling) time of the above value.
+    google.protobuf.Timestamp timestamp = 2;
+
+    // Contextual information about the example value.
+    map<string, string> attachments = 3;
+  }
+}
+
+// The start_timestamp only applies to the count and sum in the SummaryValue.
+message SummaryValue {
+  // The total number of recorded values since start_time. Optional since
+  // some systems don't expose this.
+  google.protobuf.Int64Value count = 1;
+
+  // The total sum of recorded values since start_time. Optional since some
+  // systems don't expose this. If count is zero then this field must be zero.
+  // This field must be unset if the sum is not available.
+  google.protobuf.DoubleValue sum = 2;
+
+  // The values in this message can be reset at arbitrary unknown times, with
+  // the requirement that all of them are reset at the same time.
+  message Snapshot {
+    // The number of values in the snapshot. Optional since some systems don't
+    // expose this.
+    google.protobuf.Int64Value count = 1;
+
+    // The sum of values in the snapshot. Optional since some systems don't
+    // expose this. If count is zero then this field must be zero or not set
+    // (if not supported).
+    google.protobuf.DoubleValue sum = 2;
+
+    // Represents the value at a given percentile of a distribution.
+    message ValueAtPercentile {
+      // The percentile of a distribution. Must be in the interval
+      // (0.0, 100.0].
+      double percentile = 1;
+
+      // The value at the given percentile of a distribution.
+      double value = 2;
+    }
+
+    // A list of values at different percentiles of the distribution calculated
+    // from the current snapshot. The percentiles must be strictly increasing.
+    repeated ValueAtPercentile percentile_values = 3;
+  }
+
+  // Values calculated over an arbitrary time window.
+  Snapshot snapshot = 3;
+}
+

oap-server/server-receiver-plugin/receiver-proto/src/main/proto/opencensus/proto/resource/v1/resource.proto

+// Copyright 2018, OpenCensus Authors
+//
+// Licensed 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.
+
+syntax = "proto3";
+
+package opencensus.proto.resource.v1;
+
+option go_package = "github.com/census-instrumentation/opencensus-proto/gen-go/resource/v1";
+
+option java_multiple_files = true;
+option java_package = "io.opencensus.proto.resource.v1";
+option java_outer_classname = "ResourceProto";
+
+// Resource information.
+message Resource {
+
+  // Type identifier for the resource.
+  string type = 1;
+
+  // Set of labels that describe the resource.
+  map<string,string> labels = 2;
+}

oap-server/server-receiver-plugin/receiver-proto/src/main/proto/opencensus/proto/stats/v1/stats.proto

+// Copyright 2016-18, OpenCensus Authors
+//
+// Licensed 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.
+
+syntax = "proto3";
+
+package opencensus.proto.stats.v1;
+
+import "google/protobuf/timestamp.proto";
+
+option go_package = "github.com/census-instrumentation/opencensus-proto/gen-go/stats/v1";
+
+option java_multiple_files = true;
+option java_package = "io.opencensus.proto.stats.v1";
+option java_outer_classname = "StatsProto";
+
+// TODO(bdrutu): Consider if this should be moved to a "tags" directory to match the API structure.
+message Tag {
+  string key = 1;
+  string value = 2;
+}
+
+// Measure .
+message Measure {
+  // A string by which the measure will be referred to, e.g. "rpc_server_latency". Names MUST be
+  // unique within the library.
+  string name = 1;
+
+  // Describes the measure, e.g. "RPC latency in seconds".
+  string description = 2;
+
+  // Describes the unit used for the Measure. Follows the format described by
+  // http://unitsofmeasure.org/ucum.html.
+  string unit = 3;
+
+  enum Type {
+    // Unknown type.
+    TYPE_UNSPECIFIED = 0;
+    // Indicates an int64 Measure.
+    INT64 = 1;
+    // Indicates a double Measure.
+    DOUBLE = 2;
+  }
+
+  // The type used for this Measure.
+  Type type = 4;
+}
+
+message View {
+  // A string by which the View will be referred to, e.g. "rpc_latency". Names MUST be unique
+  // within the library.
+  string name = 1;
+
+  // Describes the view, e.g. "RPC latency distribution"
+  string description = 2;
+
+  // The Measure to which this view is applied.
+  Measure measure = 3;
+
+  // An array of tag keys. These values associated with tags of this name form the basis by which
+  // individual stats will be aggregated (one aggregation per unique tag value). If none are
+  // provided, then all data is recorded in a single aggregation.
+  repeated string columns = 4;
+
+  // The description of the aggregation used for this view which describes how data collected are
+  // aggregated.
+  oneof aggregation {
+    // Counts the number of measurements recorded.
+    CountAggregation count_aggregation = 5;
+    // Indicates that data collected and aggregated with this Aggregation will be summed up.
+    SumAggregation sum_aggregation = 6;
+    // Indicates that data collected and aggregated with this Aggregation will represent the last
+    // recorded value. This is useful to support Gauges.
+    LastValueAggregation last_value_aggregation = 7;
+    // Indicates that the desired Aggregation is a histogram distribution. A distribution
+    // Aggregation may contain a histogram of the values in the population. User should define the
+    // bucket boundaries for that histogram (see DistributionAggregation).
+    DistributionAggregation distribution_aggregation = 8;
+  }
+}
+
+message CountAggregation {}
+
+message SumAggregation {}
+
+message LastValueAggregation {}
+
+message DistributionAggregation {
+  // A Distribution may optionally contain a histogram of the values in the
+  // population. The bucket boundaries for that histogram are described by
+  // `bucket_bounds`. This defines `size(bucket_bounds) + 1` (= N)
+  // buckets. The boundaries for bucket index i are:
+  //
+  // (-infinity, bucket_bounds[i]) for i == 0
+  // [bucket_bounds[i-1], bucket_bounds[i]) for 0 < i < N-2
+  // [bucket_bounds[i-1], +infinity) for i == N-1
+  //
+  // i.e. an underflow bucket (number 0), zero or more finite buckets (1
+  // through N - 2, and an overflow bucket (N - 1), with inclusive lower
+  // bounds and exclusive upper bounds.
+  //
+  // If `bucket_bounds` has no elements (zero size), then there is no
+  // histogram associated with the Distribution. If `bucket_bounds` has only
+  // one element, there are no finite buckets, and that single element is the
+  // common boundary of the overflow and underflow buckets. The values must
+  // be monotonically increasing.
+  repeated double bucket_bounds = 1;
+}
+
+// Describes a data point to be collected for a Measure.
+message Measurement {
+  repeated Tag tags = 1;
+
+  // The name of the measure to which the value is applied.
+  string measure_name = 2;
+
+  // The recorded value, MUST have the appropriate type to match the Measure.
+  oneof value {
+    double double_value = 3;
+    int64 int_value = 4;
+  }
+
+  // The time when this measurement was recorded. If the implementation uses a async buffer to
+  // record measurements this may be the time when the measurement was read from the buffer.
+  google.protobuf.Timestamp time = 5;
+}

oap-server/server-receiver-plugin/receiver-proto/src/main/proto/opencensus/proto/trace/v1/trace.proto

+// Copyright 2017, OpenCensus Authors
+//
+// Licensed 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.
+
+syntax = "proto3";
+
+package opencensus.proto.trace.v1;
+
+import "opencensus/proto/resource/v1/resource.proto";
+import "google/protobuf/timestamp.proto";
+import "google/protobuf/wrappers.proto";
+
+option java_multiple_files = true;
+option java_package = "io.opencensus.proto.trace.v1";
+option java_outer_classname = "TraceProto";
+
+option go_package = "github.com/census-instrumentation/opencensus-proto/gen-go/trace/v1";
+
+// A span represents a single operation within a trace. Spans can be
+// nested to form a trace tree. Spans may also be linked to other spans
+// from the same or different trace. And form graphs. Often, a trace
+// contains a root span that describes the end-to-end latency, and one
+// or more subspans for its sub-operations. A trace can also contain
+// multiple root spans, or none at all. Spans do not need to be
+// contiguous - there may be gaps or overlaps between spans in a trace.
+//
+// The next id is 17.
+// TODO(bdrutu): Add an example.
+message Span {
+  // A unique identifier for a trace. All spans from the same trace share
+  // the same `trace_id`. The ID is a 16-byte array. An ID with all zeroes
+  // is considered invalid.
+  //
+  // This field is semantically required. Receiver should generate new
+  // random trace_id if empty or invalid trace_id was received.
+  //
+  // This field is required.
+  bytes trace_id = 1;
+
+  // A unique identifier for a span within a trace, assigned when the span
+  // is created. The ID is an 8-byte array. An ID with all zeroes is considered
+  // invalid.
+  //
+  // This field is semantically required. Receiver should generate new
+  // random span_id if empty or invalid span_id was received.
+  //
+  // This field is required.
+  bytes span_id = 2;
+
+  // This field conveys information about request position in multiple distributed tracing graphs.
+  // It is a list of Tracestate.Entry with a maximum of 32 members in the list.
+  //
+  // See the https://github.com/w3c/distributed-tracing for more details about this field.
+  message Tracestate {
+    message Entry {
+      // The key must begin with a lowercase letter, and can only contain
+      // lowercase letters 'a'-'z', digits '0'-'9', underscores '_', dashes
+      // '-', asterisks '*', and forward slashes '/'.
+      string key = 1;
+
+      // The value is opaque string up to 256 characters printable ASCII
+      // RFC0020 characters (i.e., the range 0x20 to 0x7E) except ',' and '='.
+      // Note that this also excludes tabs, newlines, carriage returns, etc.
+      string value = 2;
+    }
+
+    // A list of entries that represent the Tracestate.
+    repeated Entry entries = 1;
+  }
+
+  // The Tracestate on the span.
+  Tracestate tracestate = 15;
+
+  // The `span_id` of this span's parent span. If this is a root span, then this
+  // field must be empty. The ID is an 8-byte array.
+  bytes parent_span_id = 3;
+
+  // A description of the span's operation.
+  //
+  // For example, the name can be a qualified method name or a file name
+  // and a line number where the operation is called. A best practice is to use
+  // the same display name at the same call point in an application.
+  // This makes it easier to correlate spans in different traces.
+  //
+  // This field is semantically required to be set to non-empty string.
+  // When null or empty string received - receiver may use string "name"
+  // as a replacement. There might be smarted algorithms implemented by
+  // receiver to fix the empty span name.
+  //
+  // This field is required.
+  TruncatableString name = 4;
+
+  // Type of span. Can be used to specify additional relationships between spans
+  // in addition to a parent/child relationship.
+  enum SpanKind {
+    // Unspecified.
+    SPAN_KIND_UNSPECIFIED = 0;
+
+    // Indicates that the span covers server-side handling of an RPC or other
+    // remote network request.
+    SERVER = 1;
+
+    // Indicates that the span covers the client-side wrapper around an RPC or
+    // other remote request.
+    CLIENT = 2;
+  }
+
+  // Distinguishes between spans generated in a particular context. For example,
+  // two spans with the same name may be distinguished using `CLIENT` (caller)
+  // and `SERVER` (callee) to identify queueing latency associated with the span.
+  SpanKind kind = 14;
+
+  // The start time of the span. On the client side, this is the time kept by
+  // the local machine where the span execution starts. On the server side, this
+  // is the time when the server's application handler starts running.
+  //
+  // This field is semantically required. When not set on receive -
+  // receiver should set it to the value of end_time field if it was
+  // set. Or to the current time if neither was set. It is important to
+  // keep end_time > start_time for consistency.
+  //
+  // This field is required.
+  google.protobuf.Timestamp start_time = 5;
+
+  // The end time of the span. On the client side, this is the time kept by
+  // the local machine where the span execution ends. On the server side, this
+  // is the time when the server application handler stops running.
+  //
+  // This field is semantically required. When not set on receive -
+  // receiver should set it to start_time value. It is important to
+  // keep end_time > start_time for consistency.
+  //
+  // This field is required.
+  google.protobuf.Timestamp end_time = 6;
+
+  // A set of attributes, each with a key and a value.
+  message Attributes {
+    // The set of attributes. The value can be a string, an integer, a double
+    // or the Boolean values `true` or `false`. Note, global attributes like 
+    // server name can be set as tags using resource API. Examples of attributes:
+    //
+    //     "/http/user_agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_14_2) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/71.0.3578.98 Safari/537.36"
+    //     "/http/server_latency": 300
+    //     "abc.com/myattribute": true
+    //     "abc.com/score": 10.239
+    map<string, AttributeValue> attribute_map = 1;
+
+    // The number of attributes that were discarded. Attributes can be discarded
+    // because their keys are too long or because there are too many attributes.
+    // If this value is 0, then no attributes were dropped.
+    int32 dropped_attributes_count = 2;
+  }
+
+  // A set of attributes on the span.
+  Attributes attributes = 7;
+
+  // A stack trace captured at the start of the span.
+  StackTrace stack_trace = 8;
+
+  // A time-stamped annotation or message event in the Span.
+  message TimeEvent {
+    // The time the event occurred.
+    google.protobuf.Timestamp time = 1;
+
+    // A text annotation with a set of attributes.
+    message Annotation {
+      // A user-supplied message describing the event.
+      TruncatableString description = 1;
+
+      // A set of attributes on the annotation.
+      Attributes attributes = 2;
+    }
+
+    // An event describing a message sent/received between Spans.
+    message MessageEvent {
+      // Indicates whether the message was sent or received.
+      enum Type {
+        // Unknown event type.
+        TYPE_UNSPECIFIED = 0;
+        // Indicates a sent message.
+        SENT = 1;
+        // Indicates a received message.
+        RECEIVED = 2;
+      }
+
+      // The type of MessageEvent. Indicates whether the message was sent or
+      // received.
+      Type type = 1;
+
+      // An identifier for the MessageEvent's message that can be used to match
+      // SENT and RECEIVED MessageEvents. For example, this field could
+      // represent a sequence ID for a streaming RPC. It is recommended to be
+      // unique within a Span.
+      uint64 id = 2;
+
+      // The number of uncompressed bytes sent or received.
+      uint64 uncompressed_size = 3;
+
+      // The number of compressed bytes sent or received. If zero, assumed to
+      // be the same size as uncompressed.
+      uint64 compressed_size = 4;
+    }
+
+    // A `TimeEvent` can contain either an `Annotation` object or a
+    // `MessageEvent` object, but not both.
+    oneof value {
+      // A text annotation with a set of attributes.
+      Annotation annotation = 2;
+
+      // An event describing a message sent/received between Spans.
+      MessageEvent message_event = 3;
+    }
+  }
+
+  // A collection of `TimeEvent`s. A `TimeEvent` is a time-stamped annotation
+  // on the span, consisting of either user-supplied key-value pairs, or
+  // details of a message sent/received between Spans.
+  message TimeEvents {
+    // A collection of `TimeEvent`s.
+    repeated TimeEvent time_event = 1;
+
+    // The number of dropped annotations in all the included time events.
+    // If the value is 0, then no annotations were dropped.
+    int32 dropped_annotations_count = 2;
+
+    // The number of dropped message events in all the included time events.
+    // If the value is 0, then no message events were dropped.
+    int32 dropped_message_events_count = 3;
+  }
+
+  // The included time events.
+  TimeEvents time_events = 9;
+
+  // A pointer from the current span to another span in the same trace or in a
+  // different trace. For example, this can be used in batching operations,
+  // where a single batch handler processes multiple requests from different
+  // traces or when the handler receives a request from a different project.
+  message Link {
+    // A unique identifier of a trace that this linked span is part of. The ID is a 
+    // 16-byte array.
+    bytes trace_id = 1;
+
+    // A unique identifier for the linked span. The ID is an 8-byte array.
+    bytes span_id = 2;
+
+    // The relationship of the current span relative to the linked span: child,
+    // parent, or unspecified.
+    enum Type {
+      // The relationship of the two spans is unknown, or known but other
+      // than parent-child.
+      TYPE_UNSPECIFIED = 0;
+      // The linked span is a child of the current span.
+      CHILD_LINKED_SPAN = 1;
+      // The linked span is a parent of the current span.
+      PARENT_LINKED_SPAN = 2;
+    }
+
+    // The relationship of the current span relative to the linked span.
+    Type type = 3;
+
+    // A set of attributes on the link.
+    Attributes attributes = 4;
+
+    // The Tracestate associated with the link.
+    Tracestate tracestate = 5;
+  }
+
+  // A collection of links, which are references from this span to a span
+  // in the same or different trace.
+  message Links {
+    // A collection of links.
+    repeated Link link = 1;
+
+    // The number of dropped links after the maximum size was enforced. If
+    // this value is 0, then no links were dropped.
+    int32 dropped_links_count = 2;
+  }
+
+  // The included links.
+  Links links = 10;
+
+  // An optional final status for this span. Semantically when Status
+  // wasn't set it is means span ended without errors and assume
+  // Status.Ok (code = 0).
+  Status status = 11;
+
+  // An optional resource that is associated with this span. If not set, this span 
+  // should be part of a batch that does include the resource information, unless resource 
+  // information is unknown.
+  opencensus.proto.resource.v1.Resource resource = 16;
+
+  // A highly recommended but not required flag that identifies when a
+  // trace crosses a process boundary. True when the parent_span belongs
+  // to the same process as the current span. This flag is most commonly
+  // used to indicate the need to adjust time as clocks in different
+  // processes may not be synchronized.
+  google.protobuf.BoolValue same_process_as_parent_span = 12;
+
+  // An optional number of child spans that were generated while this span
+  // was active. If set, allows an implementation to detect missing child spans.
+  google.protobuf.UInt32Value child_span_count = 13;
+}
+
+// The `Status` type defines a logical error model that is suitable for different
+// programming environments, including REST APIs and RPC APIs. This proto's fields
+// are a subset of those of
+// [google.rpc.Status](https://github.com/googleapis/googleapis/blob/master/google/rpc/status.proto),
+// which is used by [gRPC](https://github.com/grpc).
+message Status {
+  // The status code. This is optional field. It is safe to assume 0 (OK)
+  // when not set.
+  int32 code = 1;
+
+  // A developer-facing error message, which should be in English.
+  string message = 2;
+}
+
+// The value of an Attribute.
+message AttributeValue {
+  // The type of the value.
+  oneof value {
+    // A string up to 256 bytes long.
+    TruncatableString string_value = 1;
+    // A 64-bit signed integer.
+    int64 int_value = 2;
+    // A Boolean value represented by `true` or `false`.
+    bool bool_value = 3;
+    // A double value.
+    double double_value = 4;
+  }
+}
+
+// The call stack which originated this span.
+message StackTrace {
+  // A single stack frame in a stack trace.
+  message StackFrame {
+    // The fully-qualified name that uniquely identifies the function or
+    // method that is active in this frame.
+    TruncatableString function_name = 1;
+    // An un-mangled function name, if `function_name` is
+    // [mangled](http://www.avabodh.com/cxxin/namemangling.html). The name can
+    // be fully qualified.
+    TruncatableString original_function_name = 2;
+    // The name of the source file where the function call appears.
+    TruncatableString file_name = 3;
+    // The line number in `file_name` where the function call appears.
+    int64 line_number = 4;
+    // The column number where the function call appears, if available.
+    // This is important in JavaScript because of its anonymous functions.
+    int64 column_number = 5;
+    // The binary module from where the code was loaded.
+    Module load_module = 6;
+    // The version of the deployed source code.
+    TruncatableString source_version = 7;
+  }
+
+  // A collection of stack frames, which can be truncated.
+  message StackFrames {
+    // Stack frames in this call stack.
+    repeated StackFrame frame = 1;
+    // The number of stack frames that were dropped because there
+    // were too many stack frames.
+    // If this value is 0, then no stack frames were dropped.
+    int32 dropped_frames_count = 2;
+  }
+
+  // Stack frames in this stack trace.
+  StackFrames stack_frames = 1;
+
+  // The hash ID is used to conserve network bandwidth for duplicate
+  // stack traces within a single trace.
+  //
+  // Often multiple spans will have identical stack traces.
+  // The first occurrence of a stack trace should contain both
+  // `stack_frames` and a value in `stack_trace_hash_id`.
+  //
+  // Subsequent spans within the same request can refer
+  // to that stack trace by setting only `stack_trace_hash_id`.
+  //
+  // TODO: describe how to deal with the case where stack_trace_hash_id is
+  // zero because it was not set.
+  uint64 stack_trace_hash_id = 2;
+}
+
+// A description of a binary module.
+message Module {
+  // TODO: document the meaning of this field.
+  // For example: main binary, kernel modules, and dynamic libraries
+  // such as libc.so, sharedlib.so.
+  TruncatableString module = 1;
+
+  // A unique identifier for the module, usually a hash of its
+  // contents.
+  TruncatableString build_id = 2;
+}
+
+// A string that might be shortened to a specified length.
+message TruncatableString {
+  // The shortened string. For example, if the original string was 500 bytes long and
+  // the limit of the string was 128 bytes, then this value contains the first 128
+  // bytes of the 500-byte string. Note that truncation always happens on a
+  // character boundary, to ensure that a truncated string is still valid UTF-8.
+  // Because it may contain multi-byte characters, the size of the truncated string
+  // may be less than the truncation limit.
+  string value = 1;
+
+  // The number of bytes removed from the original string. If this
+  // value is 0, then the string was not shortened.
+  int32 truncated_byte_count = 2;
+}

oap-server/server-receiver-plugin/receiver-proto/src/main/proto/opencensus/proto/trace/v1/trace_config.proto

+// Copyright 2018, OpenCensus Authors
+//
+// Licensed 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.
+
+syntax = "proto3";
+
+package opencensus.proto.trace.v1;
+
+option java_multiple_files = true;
+option java_package = "io.opencensus.proto.trace.v1";
+option java_outer_classname = "TraceConfigProto";
+
+option go_package = "github.com/census-instrumentation/opencensus-proto/gen-go/trace/v1";
+
+// Global configuration of the trace service. All fields must be specified, or
+// the default (zero) values will be used for each type.
+message TraceConfig {
+
+  // The global default sampler used to make decisions on span sampling.
+  oneof sampler {
+    ProbabilitySampler probability_sampler = 1;
+
+    ConstantSampler constant_sampler = 2;
+
+    RateLimitingSampler rate_limiting_sampler = 3;
+  }
+
+  // The global default max number of attributes per span.
+  int64 max_number_of_attributes = 4;
+
+  // The global default max number of annotation events per span.
+  int64 max_number_of_annotations = 5;
+
+  // The global default max number of message events per span.
+  int64 max_number_of_message_events = 6;
+
+  // The global default max number of link entries per span.
+  int64 max_number_of_links = 7;
+}
+
+// Sampler that tries to uniformly sample traces with a given probability.
+// The probability of sampling a trace is equal to that of the specified probability.
+message ProbabilitySampler {
+
+  // The desired probability of sampling. Must be within [0.0, 1.0].
+  double samplingProbability = 1;
+}
+
+// Sampler that always makes a constant decision on span sampling.
+message ConstantSampler {
+
+  // How spans should be sampled:
+  // - Always off
+  // - Always on
+  // - Always follow the parent Span's decision (off if no parent).
+  enum ConstantDecision {
+    ALWAYS_OFF = 0;
+    ALWAYS_ON = 1;
+    ALWAYS_PARENT = 2;
+  }
+  ConstantDecision decision = 1;
+}
+
+// Sampler that tries to sample with a rate per time window.
+message RateLimitingSampler {
+
+  // Rate per second.
+  int64 qps = 1;
+}

pom.xml

@@ -474,6 +474,7 @@
                         <exclude>**/src/main/proto/prometheus/client_model/metrics.proto</exclude>
                         <exclude>**/src/main/proto/protoc-gen-swagger/**</exclude>
                         <exclude>**/src/main/proto/validate/validate.proto</exclude>
+                        <exclude>**/src/main/proto/opencensus/**</exclude>
 
                         <!-- generated file from oal grammar and rt template -->
                         <exclude>**/src/main/antlr4/org/apache/skywalking/oal/rt/grammar/OALLexer.tokens</exclude>