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.core.analysis.worker;
  
  import java.util.ArrayList;
  import java.util.Collections;
  import java.util.List;
  import java.util.Optional;
  import java.util.stream.Collectors;
  import lombok.extern.slf4j.Slf4j;
  import org.apache.skywalking.oap.server.core.CoreModule;
  import org.apache.skywalking.oap.server.core.UnexpectedException;
  import org.apache.skywalking.oap.server.core.analysis.DownSampling;
  import org.apache.skywalking.oap.server.core.analysis.TimeBucket;
  import org.apache.skywalking.oap.server.core.analysis.data.MergableBufferedData;
  import org.apache.skywalking.oap.server.core.analysis.data.ReadWriteSafeCache;
  import org.apache.skywalking.oap.server.core.analysis.metrics.Metrics;
  import org.apache.skywalking.oap.server.core.exporter.ExportEvent;
  import org.apache.skywalking.oap.server.core.status.BootingStatus;
  import org.apache.skywalking.oap.server.core.status.ClusterStatus;
  import org.apache.skywalking.oap.server.core.status.ServerStatusService;
  import org.apache.skywalking.oap.server.core.status.ServerStatusWatcher;
  import org.apache.skywalking.oap.server.core.storage.IMetricsDAO;
  import org.apache.skywalking.oap.server.core.storage.SessionCacheCallback;
  import org.apache.skywalking.oap.server.core.storage.model.Model;
  import org.apache.skywalking.oap.server.core.worker.AbstractWorker;
  import org.apache.skywalking.oap.server.library.client.request.PrepareRequest;
  import org.apache.skywalking.oap.server.library.datacarrier.DataCarrier;
  import org.apache.skywalking.oap.server.library.datacarrier.consumer.BulkConsumePool;
  import org.apache.skywalking.oap.server.library.datacarrier.consumer.ConsumerPoolFactory;
  import org.apache.skywalking.oap.server.library.datacarrier.consumer.IConsumer;
  import org.apache.skywalking.oap.server.library.module.ModuleDefineHolder;
  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.MetricsCreator;
  import org.apache.skywalking.oap.server.telemetry.api.MetricsTag;
  
  /**
   * MetricsPersistentWorker is an extension of {@link PersistenceWorker} and focuses on the Metrics data persistent.
   */
  @Slf4j
  public class MetricsPersistentWorker extends PersistenceWorker<Metrics> implements ServerStatusWatcher {
      private final Model model;
      private final MetricsSessionCache sessionCache;
      private final IMetricsDAO metricsDAO;
      private final Optional<AbstractWorker<Metrics>> nextAlarmWorker;
      private final Optional<AbstractWorker<ExportEvent>> nextExportWorker;
      private final DataCarrier<Metrics> dataCarrier;
      private final Optional<MetricsTransWorker> transWorker;
      private final boolean supportUpdate;
      /**
       * The counter of L2 aggregation.
       */
      private CounterMetrics aggregationCounter;
      /**
       * The counter of metrics reading from Database.
       */
      private CounterMetrics readMetricsCounter;
      /**
       * The counter of metrics cached in-memory.
       */
      private CounterMetrics cachedMetricsCounter;
      /**
       * The counter for the round of persistent.
       */
      private int persistentCounter;
      /**
       * The mod value to control persistent. The MetricsPersistentWorker is driven by the {@link
       * org.apache.skywalking.oap.server.core.storage.PersistenceTimer}. The down sampling level workers only execute in
       * every {@link #persistentMod} periods. And minute level workers execute every time.
       */
      private int persistentMod;
      /**
       * @since 8.7.0 TTL settings from {@link org.apache.skywalking.oap.server.core.CoreModuleConfig#getMetricsDataTTL()}
       */
      private int metricsDataTTL;
      /**
       * @since 9.4.0
       */
      private final ServerStatusService serverStatusService;
      /**
       * The time bucket is 0 or in minute dimensionality of the system in the latest stability status.
       *
       * @since 9.4.0
       */
      private volatile long timeOfLatestStabilitySts = 0;
  
      MetricsPersistentWorker(ModuleDefineHolder moduleDefineHolder, Model model, IMetricsDAO metricsDAO,
                              AbstractWorker<Metrics> nextAlarmWorker, AbstractWorker<ExportEvent> nextExportWorker,
                              MetricsTransWorker transWorker, boolean supportUpdate,
                              long storageSessionTimeout, int metricsDataTTL, MetricStreamKind kind) {
          super(moduleDefineHolder, new ReadWriteSafeCache<>(new MergableBufferedData(), new MergableBufferedData()));
          this.model = model;
          this.sessionCache = new MetricsSessionCache(storageSessionTimeout);
          this.metricsDAO = metricsDAO;
          this.nextAlarmWorker = Optional.ofNullable(nextAlarmWorker);
          this.nextExportWorker = Optional.ofNullable(nextExportWorker);
          this.transWorker = Optional.ofNullable(transWorker);
          this.supportUpdate = supportUpdate;
          this.persistentCounter = 0;
          this.persistentMod = 1;
          this.metricsDataTTL = metricsDataTTL;
  
          String name = "METRICS_L2_AGGREGATION";
          int size = BulkConsumePool.Creator.recommendMaxSize() / 8;
          if (size == 0) {
              size = 1;
          }
          BulkConsumePool.Creator creator = new BulkConsumePool.Creator(name, size, 20);
          try {
              ConsumerPoolFactory.INSTANCE.createIfAbsent(name, creator);
          } catch (Exception e) {
              throw new UnexpectedException(e.getMessage(), e);
          }
  
          int bufferSize = 2000;
          if (MetricStreamKind.MAL == kind) {
              // In MAL meter streaming, the load of data flow is much less as they are statistics already,
              // but in OAL sources, they are raw data.
              // Set the buffer(size of queue) as 1/2 to reduce unnecessary resource costs.
              bufferSize = 1000;
          }
          this.dataCarrier = new DataCarrier<>("MetricsPersistentWorker." + model.getName(), name, 1, bufferSize);
          this.dataCarrier.consume(ConsumerPoolFactory.INSTANCE.get(name), new PersistentConsumer());
  
          MetricsCreator metricsCreator = moduleDefineHolder.find(TelemetryModule.NAME)
                                                            .provider()
                                                            .getService(MetricsCreator.class);
          aggregationCounter = metricsCreator.createCounter(
              "metrics_aggregation", "The number of rows in aggregation",
              new MetricsTag.Keys("metricName", "level", "dimensionality"),
              new MetricsTag.Values(model.getName(), "2", model.getDownsampling().getName())
          );
          readMetricsCounter = metricsCreator.createCounter(
              "metrics_persistent_cache", "The counter of metrics status, new or cached.",
              new MetricsTag.Keys("status"), new MetricsTag.Values("new")
          );
          cachedMetricsCounter = metricsCreator.createCounter(
              "metrics_persistent_cache", "The counter of metrics status, new or cached.",
              new MetricsTag.Keys("status"), new MetricsTag.Values("cached")
          );
          serverStatusService = moduleDefineHolder.find(CoreModule.NAME).provider().getService(ServerStatusService.class);
          if (model.getDownsampling().equals(DownSampling.Minute)) {
              serverStatusService.registerWatcher(this);
          }
      }
  
      /**
       * Create the leaf and down-sampling MetricsPersistentWorker, no next step.
       */
      MetricsPersistentWorker(ModuleDefineHolder moduleDefineHolder,
                              Model model,
                              IMetricsDAO metricsDAO,
                              boolean supportUpdate,
                              long storageSessionTimeout,
                              int metricsDataTTL,
                              MetricStreamKind kind) {
          this(moduleDefineHolder, model, metricsDAO,
               null, null, null,
               supportUpdate, storageSessionTimeout, metricsDataTTL, kind
          );
          // For a down-sampling metrics, we prolong the session timeout for 4 times, nearly 5 minutes.
          sessionCache.setTimeoutThreshold(storageSessionTimeout * 4);
          // The down sampling level worker executes every 4 periods.
          this.persistentMod = 4;
      }
  
      /**
       * Accept all metrics data and push them into the queue for serial processing
       */
      @Override
      public void in(Metrics metrics) {
          aggregationCounter.inc();
          dataCarrier.produce(metrics);
      }
  
      @Override
      public List<PrepareRequest> buildBatchRequests() {
          if (persistentCounter++ % persistentMod != 0) {
              return Collections.emptyList();
          }
  
          final List<Metrics> lastCollection = getCache().read();
  
          long start = System.currentTimeMillis();
          if (lastCollection.size() == 0) {
              return Collections.emptyList();
          }
  
          /*
           * Hard coded the max size. This only affect the multiIDRead if the data doesn't hit the cache.
           */
          int maxBatchGetSize = 2000;
          final int batchSize = Math.min(maxBatchGetSize, lastCollection.size());
          List<Metrics> metricsList = new ArrayList<>();
          List<PrepareRequest> prepareRequests = new ArrayList<>(lastCollection.size());
          for (Metrics data : lastCollection) {
              transWorker.ifPresent(metricsTransWorker -> metricsTransWorker.in(data));
  
              metricsList.add(data);
  
              if (metricsList.size() == batchSize) {
                  prepareFlushDataToStorage(metricsList, prepareRequests);
              }
          }
  
          if (metricsList.size() > 0) {
              prepareFlushDataToStorage(metricsList, prepareRequests);
          }
  
          if (prepareRequests.size() > 0) {
              log.debug(
                  "prepare batch requests for model {}, took time: {}, size: {}", model.getName(),
                  System.currentTimeMillis() - start, prepareRequests.size()
              );
          }
          return prepareRequests;
      }
  
      /**
       * Build given prepareRequests to prepare database flush
       *
       * @param metricsList     the metrics in the last read from the in-memory aggregated cache.
       * @param prepareRequests the results for final execution.
       */
      private void prepareFlushDataToStorage(List<Metrics> metricsList,
                                             List<PrepareRequest> prepareRequests) {
          try {
              loadFromStorage(metricsList);
  
              long timestamp = System.currentTimeMillis();
              for (Metrics metrics : metricsList) {
                  Metrics cachedMetrics = sessionCache.get(metrics);
                  if (cachedMetrics != null) {
                      cachedMetrics.setLastUpdateTimestamp(timestamp);
                      /*
                       * If the metrics is not supportUpdate, defined through MetricsExtension#supportUpdate,
                       * then no merge and further process happens.
                       */
                      if (!supportUpdate) {
                          continue;
                      }
                      /*
                       * Merge metrics into cachedMetrics, change only happens inside cachedMetrics.
                       */
                      final boolean isAbandoned = !cachedMetrics.combine(metrics);
                      if (isAbandoned) {
                          continue;
                      }
                      cachedMetrics.calculate();
                      prepareRequests.add(
                          metricsDAO.prepareBatchUpdate(
                              model,
                              cachedMetrics,
                              new SessionCacheCallback(sessionCache, cachedMetrics)
                          ));
                      nextWorker(cachedMetrics);
                  } else {
                      metrics.calculate();
                      prepareRequests.add(
                          metricsDAO.prepareBatchInsert(
                              model,
                              metrics,
                              new SessionCacheCallback(sessionCache, metrics)
                          ));
                      nextWorker(metrics);
                      metrics.setLastUpdateTimestamp(timestamp);
                  }
  
                  /*
                   * The `metrics` should be not changed in all above process. Exporter is an async process.
                   */
                  nextExportWorker.ifPresent(exportEvenWorker -> exportEvenWorker.in(
                      new ExportEvent(metrics, ExportEvent.EventType.INCREMENT)));
              }
          } catch (Throwable t) {
              log.error(t.getMessage(), t);
          } finally {
              metricsList.clear();
          }
      }
  
      private void nextWorker(Metrics metrics) {
          nextAlarmWorker.ifPresent(nextAlarmWorker -> nextAlarmWorker.in(metrics));
          nextExportWorker.ifPresent(
              nextExportWorker -> nextExportWorker.in(new ExportEvent(metrics, ExportEvent.EventType.TOTAL)));
      }
  
      /**
       * Load data from the storage, only load data when the id doesn't exist.
       */
      private void loadFromStorage(List<Metrics> metrics) {
          final long currentTimeMillis = System.currentTimeMillis();
          try {
              List<Metrics> notInCacheMetrics =
                  metrics.stream()
                         .filter(m -> {
                             final Metrics cachedValue = requireInitialization(m);
                             // the metric is tagged `not in cache`.
                             if (cachedValue == null) {
                                 return true;
                             }
                             // The metric is in the cache, but still we have to check
                             // whether the cache is expired due to TTL.
                             // This is a cache-DB inconsistent case:
                             // Metrics keep coming due to traffic, but the entity in the
                             // database has been removed due to TTL.
                             if (!model.isTimeRelativeID()) {
                                 // Mostly all updatable metadata level metrics are required to do this check.
  
                                 if (metricsDAO.isExpiredCache(model, cachedValue, currentTimeMillis, metricsDataTTL)) {
                                     // The expired metrics should be removed from the context and tagged `not in cache` directly.
                                     sessionCache.remove(m);
                                     return true;
                                 }
                             }
  
                             return false;
                         })
                         .collect(Collectors.toList());
  
              readMetricsCounter.inc(notInCacheMetrics.size());
              cachedMetricsCounter.inc(metrics.size() - notInCacheMetrics.size());
              if (notInCacheMetrics.isEmpty()) {
                  return;
              }
              metricsDAO.multiGet(model, notInCacheMetrics).forEach(m -> {
                  m.setLastUpdateTimestamp(currentTimeMillis);
                  sessionCache.put(m);
              });
          } catch (final Exception e) {
              log.error("Failed to load metrics for merging", e);
          }
      }
  
      @Override
      public void endOfRound() {
          sessionCache.removeExpired();
      }
  
      /**
       * Check the metrics whether in the cache, and whether the worker should go further to load from database.
       *
       * @param metrics the metrics in the streaming process.
       * @return metrics in cache or null if try to read the metrics from the database.
       */
      private Metrics requireInitialization(Metrics metrics) {
          final Metrics cached = sessionCache.get(metrics);
  
          // All cached metrics, it at least had been written once.
          if (cached != null) {
              return cached;
          }
  
          // If the metrics do not have time bucket relative ID
          // it is treated as high dimensionality metrics.
          // Policy, always try load from the database when miss in the cache.
          if (!model.isTimeRelativeID()) {
              return null;
          }
  
          // When
          // (1) the time bucket of the server's latest stability status is provided
          //     1.1 the OAP has booted successfully
          //     1.2 the current dimensionality is in minute
          //     1.3 the OAP cluster is rebalanced due to scaling
          // (2) the metrics are from the time after the timeOfLatestStabilitySts
          // (3) the metrics don't exist in the cache
          // the kernel should NOT try to load it from the database.
          //
          // Notice, about condition (2),
          // for the specific minute of booted successfully, the metrics are expected to load from database when
          // it doesn't exist in the cache.
          if (timeOfLatestStabilitySts > 0 &&
              metrics.getTimeBucket() > timeOfLatestStabilitySts
              && cached == null) {
              // Return metrics as input to avoid reading from database.
              return metrics;
          }
  
          return null;
      }
  
      @Override
      public void onServerBooted(final BootingStatus bootingStatus) {
          timeOfLatestStabilitySts = TimeBucket.getMinuteTimeBucket(
              bootingStatus.getUptime());
      }
  
      @Override
      public void onClusterRebalanced(final ClusterStatus clusterStatus) {
          timeOfLatestStabilitySts = TimeBucket.getMinuteTimeBucket(
              clusterStatus.getRebalancedTime());
      }
  
      /**
       * Metrics queue processor, merge the received metrics if existing one with same ID(s) and time bucket.
       *
       * ID is declared through {@link Object#hashCode()} and {@link Object#equals(Object)} as usual.
       */
      private class PersistentConsumer implements IConsumer<Metrics> {
          @Override
          public void consume(List<Metrics> data) {
              MetricsPersistentWorker.this.onWork(data);
          }
  
          @Override
          public void onError(List<Metrics> data, Throwable t) {
              log.error(t.getMessage(), t);
          }
      }
  }
  