code: /*
   * 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.meter.analyzer;
  
  import com.google.common.base.Strings;
  import com.google.common.collect.ImmutableMap;
  import com.google.gson.JsonObject;
  import io.vavr.Tuple;
  import io.vavr.Tuple2;
  import java.util.List;
  import java.util.Map;
  import java.util.Objects;
  import java.util.function.Predicate;
  import java.util.stream.Collectors;
  import java.util.stream.Stream;
  import lombok.AccessLevel;
  import lombok.RequiredArgsConstructor;
  import lombok.ToString;
  import lombok.extern.slf4j.Slf4j;
  import org.apache.commons.lang3.StringUtils;
  import org.apache.commons.text.CaseUtils;
  import org.apache.skywalking.oap.meter.analyzer.dsl.DSL;
  import org.apache.skywalking.oap.meter.analyzer.dsl.DownsamplingType;
  import org.apache.skywalking.oap.meter.analyzer.dsl.Expression;
  import org.apache.skywalking.oap.meter.analyzer.dsl.ExpressionParsingContext;
  import org.apache.skywalking.oap.meter.analyzer.dsl.FilterExpression;
  import org.apache.skywalking.oap.meter.analyzer.dsl.Result;
  import org.apache.skywalking.oap.meter.analyzer.dsl.Sample;
  import org.apache.skywalking.oap.meter.analyzer.dsl.SampleFamily;
  import org.apache.skywalking.oap.server.core.analysis.Layer;
  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.relation.process.ProcessRelationClientSideMetrics;
  import org.apache.skywalking.oap.server.core.analysis.manual.relation.process.ProcessRelationServerSideMetrics;
  import org.apache.skywalking.oap.server.core.analysis.manual.relation.service.ServiceRelationClientSideMetrics;
  import org.apache.skywalking.oap.server.core.analysis.manual.relation.service.ServiceRelationServerSideMetrics;
  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.ScopeType;
  import org.apache.skywalking.oap.server.core.analysis.meter.function.AcceptableValue;
  import org.apache.skywalking.oap.server.core.analysis.meter.function.BucketedValues;
  import org.apache.skywalking.oap.server.core.analysis.meter.function.PercentileArgument;
  import org.apache.skywalking.oap.server.core.analysis.metrics.DataTable;
  import org.apache.skywalking.oap.server.core.analysis.worker.MetricsStreamProcessor;
  
  import static com.google.common.collect.ImmutableMap.toImmutableMap;
  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;
  
  /**
   * Analyzer analyses DSL expression with input samples, then to generate meter-system metrics.
   */
  @Slf4j
  @RequiredArgsConstructor(access = AccessLevel.PRIVATE)
  @ToString(of = {
      "metricName",
      "expression"
  })
  public class Analyzer {
  
      public static final Tuple2<String, SampleFamily> NIL = Tuple.of("", null);
  
      public static Analyzer build(final String metricName,
                                   final String filterExpression,
                                   final String expression,
                                   final MeterSystem meterSystem) {
          Expression e = DSL.parse(expression);
          FilterExpression filter = null;
          if (!Strings.isNullOrEmpty(filterExpression)) {
              filter = new FilterExpression(filterExpression);
          }
          ExpressionParsingContext ctx = e.parse();
          Analyzer analyzer = new Analyzer(metricName, filter, e, meterSystem, ctx);
          analyzer.init();
          return analyzer;
      }
  
      private List<String> samples;
  
      private final String metricName;
  
      private final FilterExpression filterExpression;
  
      private final Expression expression;
  
      private final MeterSystem meterSystem;
  
      private final ExpressionParsingContext ctx;
  
      private MetricType metricType;
  
      private int[] percentiles;
  
      /**
       * analyse intends to parse expression with input samples to meter-system metrics.
       *
       * @param sampleFamilies input samples.
       */
      public void analyse(final ImmutableMap<String, SampleFamily> sampleFamilies) {
          Map<String, SampleFamily> input = samples.stream()
                                                   .map(s -> Tuple.of(s, sampleFamilies.get(s)))
                                                   .filter(t -> t._2 != null)
                                                   .collect(toImmutableMap(t -> t._1, t -> t._2));
          if (input.size() < 1) {
              if (log.isDebugEnabled()) {
                  log.debug("{} is ignored due to the lack of {}", expression, samples);
              }
              return;
          }
          if (filterExpression != null) {
              input = filterExpression.filter(input);
          }
          Result r = expression.run(input);
          if (!r.isSuccess()) {
              return;
          }
          SampleFamily.RunningContext ctx = r.getData().context;
          Map<MeterEntity, Sample[]> meterSamples = ctx.getMeterSamples();
          meterSamples.forEach((meterEntity, ss) -> {
              generateTraffic(meterEntity);
              switch (metricType) {
                  case single:
                      AcceptableValue<Long> sv = meterSystem.buildMetrics(metricName, Long.class);
                      sv.accept(meterEntity, getValue(ss[0]));
                      send(sv, ss[0].getTimestamp());
                      break;
                  case labeled:
                      AcceptableValue<DataTable> lv = meterSystem.buildMetrics(metricName, DataTable.class);
                      DataTable dt = new DataTable();
                      for (Sample each : ss) {
                          dt.put(composeGroup(each.getLabels()), getValue(each));
                      }
                      lv.accept(meterEntity, dt);
                      send(lv, ss[0].getTimestamp());
                      break;
                  case histogram:
                  case histogramPercentile:
                      Stream.of(ss).map(s -> Tuple.of(composeGroup(s.getLabels(), k -> !Objects.equals("le", k)), s))
                            .collect(groupingBy(Tuple2::_1, mapping(Tuple2::_2, toList())))
                            .forEach((group, subSs) -> {
                                if (subSs.size() < 1) {
                                    return;
                                }
                                long[] bb = new long[subSs.size()];
                                long[] vv = new long[bb.length];
                                for (int i = 0; i < subSs.size(); i++) {
                                    Sample s = subSs.get(i);
                                    final double leVal = Double.parseDouble(s.getLabels().get("le"));
                                    if (leVal == Double.NEGATIVE_INFINITY) {
                                        bb[i] = Long.MIN_VALUE;
                                    } else {
                                        bb[i] = (long) leVal;
                                    }
                                    vv[i] = getValue(s);
                                }
                                BucketedValues bv = new BucketedValues(bb, vv);
                                bv.setGroup(group);
                                long time = subSs.get(0).getTimestamp();
                                if (metricType == MetricType.histogram) {
                                    AcceptableValue<BucketedValues> v = meterSystem.buildMetrics(
                                        metricName, BucketedValues.class);
                                    v.accept(meterEntity, bv);
                                    send(v, time);
                                    return;
                                }
                                AcceptableValue<PercentileArgument> v = meterSystem.buildMetrics(
                                    metricName, PercentileArgument.class);
                                v.accept(meterEntity, new PercentileArgument(bv, percentiles));
                                send(v, time);
                            });
                      break;
              }
          });
      }
  
      private long getValue(Sample sample) {
          if (sample.getValue() <= 0.0) {
              return 0L;
          }
          if (sample.getValue() < 1.0) {
              return 1L;
          }
          return Math.round(sample.getValue());
      }
  
      private String composeGroup(ImmutableMap<String, String> labels) {
          return composeGroup(labels, k -> true);
      }
  
      private String composeGroup(ImmutableMap<String, String> labels, Predicate<String> filter) {
          return labels.keySet().stream().filter(filter).sorted().map(labels::get)
                       .collect(Collectors.joining("-"));
      }
  
      @RequiredArgsConstructor
      private enum MetricType {
          // metrics is aggregated by histogram function.
          histogram("histogram"),
          // metrics is aggregated by histogram based percentile function.
          histogramPercentile("histogramPercentile"),
          // metrics is aggregated by labeled function.
          labeled("labeled"),
          // metrics is aggregated by single value function.
          single("");
  
          private final String literal;
      }
  
      private void init() {
          this.samples = ctx.getSamples();
          if (ctx.isHistogram()) {
              if (ctx.getPercentiles() != null && ctx.getPercentiles().length > 0) {
                  metricType = MetricType.histogramPercentile;
                  this.percentiles = ctx.getPercentiles();
              } else {
                  metricType = MetricType.histogram;
              }
          } else {
              if (ctx.getLabels().isEmpty()) {
                  metricType = MetricType.single;
              } else {
                  metricType = MetricType.labeled;
              }
          }
          createMetric(ctx.getScopeType(), metricType.literal, ctx.getDownsampling());
      }
  
      private void createMetric(final ScopeType scopeType,
                                final String dataType,
                                final DownsamplingType downsamplingType) {
          String downSamplingStr = CaseUtils.toCamelCase(downsamplingType.toString().toLowerCase(), false, '_');
          String functionName = String.format("%s%s", downSamplingStr, StringUtils.capitalize(dataType));
          meterSystem.create(metricName, functionName, scopeType);
      }
  
      private void send(final AcceptableValue<?> v, final long time) {
          v.setTimeBucket(TimeBucket.getMinuteTimeBucket(time));
          meterSystem.doStreamingCalculation(v);
      }
  
      private void generateTraffic(MeterEntity entity) {
          if (entity.getDetectPoint() != null) {
              switch (entity.getScopeType()) {
                  case SERVICE_RELATION:
                      serviceRelationTraffic(entity);
                      break;
                  case PROCESS_RELATION:
                      processRelationTraffic(entity);
                      break;
                  default:
              }
          } else {
              toService(requireNonNull(entity.getServiceName()), entity.getLayer());
          }
  
          if (!com.google.common.base.Strings.isNullOrEmpty(entity.getInstanceName())) {
              InstanceTraffic instanceTraffic = new InstanceTraffic();
              instanceTraffic.setName(entity.getInstanceName());
              instanceTraffic.setServiceId(entity.serviceId());
              instanceTraffic.setTimeBucket(TimeBucket.getMinuteTimeBucket(System.currentTimeMillis()));
              instanceTraffic.setLastPingTimestamp(TimeBucket.getMinuteTimeBucket(System.currentTimeMillis()));
              if (entity.getInstanceProperties() != null && !entity.getInstanceProperties().isEmpty()) {
                  final JsonObject properties = new JsonObject();
                  entity.getInstanceProperties().forEach((k, v) -> properties.addProperty(k, v));
                  instanceTraffic.setProperties(properties);
              }
              MetricsStreamProcessor.getInstance().in(instanceTraffic);
          }
          if (!com.google.common.base.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 void toService(String serviceName, Layer layer) {
          ServiceTraffic s = new ServiceTraffic();
          s.setName(requireNonNull(serviceName));
          s.setTimeBucket(TimeBucket.getMinuteTimeBucket(System.currentTimeMillis()));
          s.setLayer(layer);
          MetricsStreamProcessor.getInstance().in(s);
      }
  
      private void serviceRelationTraffic(MeterEntity entity) {
          switch (entity.getDetectPoint()) {
              case SERVER:
                  entity.setServiceName(entity.getDestServiceName());
                  toService(requireNonNull(entity.getDestServiceName()), entity.getLayer());
                  serviceRelationServerSide(entity);
                  break;
              case CLIENT:
                  entity.setServiceName(entity.getSourceServiceName());
                  toService(requireNonNull(entity.getSourceServiceName()), entity.getLayer());
                  serviceRelationClientSide(entity);
                  break;
              default:
          }
      }
  
      private void serviceRelationServerSide(MeterEntity entity) {
          ServiceRelationServerSideMetrics metrics = new ServiceRelationServerSideMetrics();
          metrics.setTimeBucket(TimeBucket.getMinuteTimeBucket(System.currentTimeMillis()));
          metrics.setSourceServiceId(entity.sourceServiceId());
          metrics.setDestServiceId(entity.destServiceId());
          metrics.getComponentIds().add(0);
          metrics.setEntityId(entity.id());
          MetricsStreamProcessor.getInstance().in(metrics);
      }
  
      private void serviceRelationClientSide(MeterEntity entity) {
          ServiceRelationClientSideMetrics metrics = new ServiceRelationClientSideMetrics();
          metrics.setTimeBucket(TimeBucket.getMinuteTimeBucket(System.currentTimeMillis()));
          metrics.setSourceServiceId(entity.sourceServiceId());
          metrics.setDestServiceId(entity.destServiceId());
          metrics.getComponentIds().add(0);
          metrics.setEntityId(entity.id());
          MetricsStreamProcessor.getInstance().in(metrics);
      }
  
      private void processRelationTraffic(MeterEntity entity) {
          switch (entity.getDetectPoint()) {
              case SERVER:
                  processRelationServerSide(entity);
                  break;
              case CLIENT:
                  processRelationClientSide(entity);
                  break;
              default:
          }
      }
  
      private void processRelationServerSide(MeterEntity entity) {
          ProcessRelationServerSideMetrics metrics = new ProcessRelationServerSideMetrics();
          metrics.setTimeBucket(TimeBucket.getMinuteTimeBucket(System.currentTimeMillis()));
          metrics.setServiceInstanceId(entity.serviceInstanceId());
          metrics.setSourceProcessId(entity.getSourceProcessId());
          metrics.setDestProcessId(entity.getDestProcessId());
          metrics.setEntityId(entity.id());
          metrics.setComponentId(entity.getComponentId());
          MetricsStreamProcessor.getInstance().in(metrics);
      }
  
      private void processRelationClientSide(MeterEntity entity) {
          ProcessRelationClientSideMetrics metrics = new ProcessRelationClientSideMetrics();
          metrics.setTimeBucket(TimeBucket.getMinuteTimeBucket(System.currentTimeMillis()));
          metrics.setServiceInstanceId(entity.serviceInstanceId());
          metrics.setSourceProcessId(entity.getSourceProcessId());
          metrics.setDestProcessId(entity.getDestProcessId());
          metrics.setEntityId(entity.id());
          metrics.setComponentId(entity.getComponentId());
          MetricsStreamProcessor.getInstance().in(metrics);
      }
  
  }
  