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.server.receiver.telegraf.provider.handler;
  
  import com.google.common.collect.ImmutableMap;
  import com.linecorp.armeria.server.annotation.Post;
  import com.linecorp.armeria.server.annotation.RequestConverter;
  import lombok.extern.slf4j.Slf4j;
  import org.apache.skywalking.apm.network.common.v3.Commands;
  import org.apache.skywalking.oap.meter.analyzer.MetricConvert;
  import org.apache.skywalking.oap.meter.analyzer.dsl.Sample;
  import org.apache.skywalking.oap.meter.analyzer.dsl.SampleFamily;
  import org.apache.skywalking.oap.meter.analyzer.dsl.SampleFamilyBuilder;
  import org.apache.skywalking.oap.meter.analyzer.prometheus.rule.Rule;
  import org.apache.skywalking.oap.server.core.analysis.meter.MeterSystem;
  import org.apache.skywalking.oap.server.library.module.ModuleManager;
  import org.apache.skywalking.oap.server.receiver.telegraf.provider.handler.pojo.TelegrafData;
  import org.apache.skywalking.oap.server.receiver.telegraf.provider.handler.pojo.TelegrafDatum;
  import org.apache.skywalking.oap.server.telemetry.TelemetryModule;
  import org.apache.skywalking.oap.server.telemetry.api.CounterMetrics;
  import org.apache.skywalking.oap.server.telemetry.api.HistogramMetrics;
  import org.apache.skywalking.oap.server.telemetry.api.MetricsCreator;
  import org.apache.skywalking.oap.server.telemetry.api.MetricsTag;
  
  import java.util.ArrayList;
  import java.util.List;
  import java.util.Map;
  import java.util.stream.Collectors;
  
  @Slf4j
  public class TelegrafServiceHandler {
  
      private final HistogramMetrics histogram;
      private final CounterMetrics errorCounter;
      private List<MetricConvert> metricConvert;
  
      public TelegrafServiceHandler(ModuleManager moduleManager, MeterSystem meterSystem, List<Rule> rules) {
  
          this.metricConvert = rules.stream().map(r -> new MetricConvert(r, meterSystem)).collect(Collectors.toList());
  
          final MetricsCreator metricsCreator = moduleManager.find(TelemetryModule.NAME)
                                                             .provider()
                                                             .getService(MetricsCreator.class);
  
          histogram = metricsCreator.createHistogramMetric(
                  "telegraf_in_latency", "The process latency of telegraf data",
                  new MetricsTag.Keys("protocol"), new MetricsTag.Values("http")
          );
  
          errorCounter = metricsCreator.createCounter(
                  "telegraf_error_count", "The error number of telegraf analysis",
                  new MetricsTag.Keys("protocol"), new MetricsTag.Values("http")
          );
      }
  
      /**
      * Convert the TelegrafData object to meter {@link Sample}
      **/
      public List<Sample> convertTelegraf(TelegrafDatum telegrafData) {
  
          List<Sample> sampleList = new ArrayList<>();
  
          Map<String, Object> fields = telegrafData.getFields();
          String name = telegrafData.getName();
          Map<String, String> tags = telegrafData.getTags();
          ImmutableMap<String, String> immutableTags = ImmutableMap.copyOf(tags);
          long timestamp = telegrafData.getTimestamp();
  
          fields.forEach((key, value) -> {
              if (value instanceof Number) {
                  Sample.SampleBuilder builder = Sample.builder();
                  Sample sample = builder.name(name + "_" + key)
                          .timestamp(timestamp * 1000L)
                          .value(((Number) value).doubleValue())
                          .labels(immutableTags).build();
  
                  sampleList.add(sample);
              }
          });
          return sampleList;
  
      }
  
      public List<ImmutableMap<String, SampleFamily>> convertSampleFamily(TelegrafData telegrafData) {
          List<Sample> allSamples = new ArrayList<>();
  
          List<TelegrafDatum> metrics = telegrafData.getMetrics();
          for (TelegrafDatum m : metrics) {
              List<Sample> samples = convertTelegraf(m);
              allSamples.addAll(samples);
          }
  
          List<ImmutableMap<String, SampleFamily>> res = new ArrayList<>();
  
          // Grouping all samples by timestamp name
          Map<Long, List<Sample>> sampleFamilyByTime = allSamples.stream()
                  .collect(Collectors.groupingBy(Sample::getTimestamp));
  
          // Grouping all samples with the same timestamp by name
          for (List<Sample> s : sampleFamilyByTime.values()) {
              ImmutableMap.Builder<String, SampleFamily> builder = ImmutableMap.builder();
              Map<String, List<Sample>> sampleFamilyByName = s.stream()
                      .collect(Collectors.groupingBy(Sample::getName));
              sampleFamilyByName.forEach((k, v) ->
                      builder.put(k, SampleFamilyBuilder.newBuilder(v.toArray(new Sample[0])).build()));
              res.add(builder.build());
          }
  
          return res;
      }
  
      @Post("/telegraf")
      @RequestConverter(TelegrafData.class)
      public Commands collectData(TelegrafData telegrafData) {
          try (HistogramMetrics.Timer ignored = histogram.createTimer()) {
              List<ImmutableMap<String, SampleFamily>> sampleFamily = convertSampleFamily(telegrafData);
              sampleFamily.forEach(s -> metricConvert.forEach(m -> m.toMeter(s)));
          } catch (Exception e) {
              errorCounter.inc();
              log.error(e.getMessage(), e);
          }
          return Commands.newBuilder().build();
      }
  }
  