content: /*
   * Copyright 1999-2018 Alibaba Group Holding Ltd.
   *
   * 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.
   */
  package com.alibaba.csp.sentinel.node;
  
  import java.util.List;
  import java.util.Map;
  import java.util.concurrent.ConcurrentHashMap;
  import java.util.concurrent.atomic.LongAdder;
  
  import com.alibaba.csp.sentinel.node.metric.MetricNode;
  import com.alibaba.csp.sentinel.slots.statistic.metric.ArrayMetric;
  import com.alibaba.csp.sentinel.slots.statistic.metric.Metric;
  import com.alibaba.csp.sentinel.util.TimeUtil;
  import com.alibaba.csp.sentinel.util.function.Predicate;
  
  /**
   * <p>The statistic node keep three kinds of real-time statistics metrics:</p>
   * <ol>
   * <li>metrics in second level ({@code rollingCounterInSecond})</li>
   * <li>metrics in minute level ({@code rollingCounterInMinute})</li>
   * <li>thread count</li>
   * </ol>
   *
   * <p>
   * Sentinel use sliding window to record and count the resource statistics in real-time.
   * The sliding window infrastructure behind the {@link ArrayMetric} is {@code LeapArray}.
   * </p>
   *
   * <p>
   * case 1: When the first request comes in, Sentinel will create a new window bucket of
   * a specified time-span to store running statics, such as total response time(rt),
   * incoming request(QPS), block request(bq), etc. And the time-span is defined by sample count.
   * </p>
   * <pre>
   * 	0      100ms
   *  +-------+--→ Sliding Windows
   * 	    ^
   * 	    |
   * 	  request
   * </pre>
   * <p>
   * Sentinel use the statics of the valid buckets to decide whether this request can be passed.
   * For example, if a rule defines that only 100 requests can be passed,
   * it will sum all qps in valid buckets, and compare it to the threshold defined in rule.
   * </p>
   *
   * <p>case 2: continuous requests</p>
   * <pre>
   *  0    100ms    200ms    300ms
   *  +-------+-------+-------+-----→ Sliding Windows
   *                      ^
   *                      |
   *                   request
   * </pre>
   *
   * <p>case 3: requests keeps coming, and previous buckets become invalid</p>
   * <pre>
   *  0    100ms    200ms	  800ms	   900ms  1000ms    1300ms
   *  +-------+-------+ ...... +-------+-------+ ...... +-------+-----→ Sliding Windows
   *                                                      ^
   *                                                      |
   *                                                    request
   * </pre>
   *
   * <p>The sliding window should become:</p>
   * <pre>
   * 300ms     800ms  900ms  1000ms  1300ms
   *  + ...... +-------+ ...... +-------+-----→ Sliding Windows
   *                                                      ^
   *                                                      |
   *                                                    request
   * </pre>
   *
   * @author qinan.qn
   * @author jialiang.linjl
   */
  public class StatisticNode implements Node {
  
      /**
       * Holds statistics of the recent {@code INTERVAL} milliseconds. The {@code INTERVAL} is divided into time spans
       * by given {@code sampleCount}.
       */
      private transient volatile Metric rollingCounterInSecond = new ArrayMetric(SampleCountProperty.SAMPLE_COUNT,
          IntervalProperty.INTERVAL);
  
      /**
       * Holds statistics of the recent 60 seconds. The windowLengthInMs is deliberately set to 1000 milliseconds,
       * meaning each bucket per second, in this way we can get accurate statistics of each second.
       */
      private transient Metric rollingCounterInMinute = new ArrayMetric(60, 60 * 1000, false);
  
      /**
       * The counter for thread count.
       */
      private LongAdder curThreadNum = new LongAdder();
  
      /**
       * The last timestamp when metrics were fetched.
       */
      private long lastFetchTime = -1;
  
      @Override
      public Map<Long, MetricNode> metrics() {
          // The fetch operation is thread-safe under a single-thread scheduler pool.
          long currentTime = TimeUtil.currentTimeMillis();
          currentTime = currentTime - currentTime % 1000;
          Map<Long, MetricNode> metrics = new ConcurrentHashMap<>();
          List<MetricNode> nodesOfEverySecond = rollingCounterInMinute.details();
          long newLastFetchTime = lastFetchTime;
          // Iterate metrics of all resources, filter valid metrics (not-empty and up-to-date).
          for (MetricNode node : nodesOfEverySecond) {
              if (isNodeInTime(node, currentTime) && isValidMetricNode(node)) {
                  metrics.put(node.getTimestamp(), node);
                  newLastFetchTime = Math.max(newLastFetchTime, node.getTimestamp());
              }
          }
          lastFetchTime = newLastFetchTime;
  
          return metrics;
      }
  
      @Override
      public List<MetricNode> rawMetricsInMin(Predicate<Long> timePredicate) {
          return rollingCounterInMinute.detailsOnCondition(timePredicate);
      }
  
      private boolean isNodeInTime(MetricNode node, long currentTime) {
          return node.getTimestamp() > lastFetchTime && node.getTimestamp() < currentTime;
      }
  
      private boolean isValidMetricNode(MetricNode node) {
          return node.getPassQps() > 0 || node.getBlockQps() > 0 || node.getSuccessQps() > 0
              || node.getExceptionQps() > 0 || node.getRt() > 0 || node.getOccupiedPassQps() > 0;
      }
  
      @Override
      public void reset() {
          rollingCounterInSecond = new ArrayMetric(SampleCountProperty.SAMPLE_COUNT, IntervalProperty.INTERVAL);
      }
  
      @Override
      public long totalRequest() {
          return rollingCounterInMinute.pass() + rollingCounterInMinute.block();
      }
  
      @Override
      public long blockRequest() {
          return rollingCounterInMinute.block();
      }
  
      @Override
      public double blockQps() {
          return rollingCounterInSecond.block() / rollingCounterInSecond.getWindowIntervalInSec();
      }
  
      @Override
      public double previousBlockQps() {
          return this.rollingCounterInMinute.previousWindowBlock();
      }
  
      @Override
      public double previousPassQps() {
          return this.rollingCounterInMinute.previousWindowPass();
      }
  
      @Override
      public double totalQps() {
          return passQps() + blockQps();
      }
  
      @Override
      public long totalSuccess() {
          return rollingCounterInMinute.success();
      }
  
      @Override
      public double exceptionQps() {
          return rollingCounterInSecond.exception() / rollingCounterInSecond.getWindowIntervalInSec();
      }
  
      @Override
      public long totalException() {
          return rollingCounterInMinute.exception();
      }
  
      @Override
      public double passQps() {
          return rollingCounterInSecond.pass() / rollingCounterInSecond.getWindowIntervalInSec();
      }
  
      @Override
      public long totalPass() {
          return rollingCounterInMinute.pass();
      }
  
      @Override
      public double successQps() {
          return rollingCounterInSecond.success() / rollingCounterInSecond.getWindowIntervalInSec();
      }
  
      @Override
      public double maxSuccessQps() {
          return (double) rollingCounterInSecond.maxSuccess() * rollingCounterInSecond.getSampleCount()
                  / rollingCounterInSecond.getWindowIntervalInSec();
      }
  
      @Override
      public double occupiedPassQps() {
          return rollingCounterInSecond.occupiedPass() / rollingCounterInSecond.getWindowIntervalInSec();
      }
  
      @Override
      public double avgRt() {
          long successCount = rollingCounterInSecond.success();
          if (successCount == 0) {
              return 0;
          }
  
          return rollingCounterInSecond.rt() * 1.0 / successCount;
      }
  
      @Override
      public double minRt() {
          return rollingCounterInSecond.minRt();
      }
  
      @Override
      public int curThreadNum() {
          return (int)curThreadNum.sum();
      }
  
      @Override
      public void addPassRequest(int count) {
          rollingCounterInSecond.addPass(count);
          rollingCounterInMinute.addPass(count);
      }
  
      @Override
      public void addRtAndSuccess(long rt, int successCount) {
          rollingCounterInSecond.addSuccess(successCount);
          rollingCounterInSecond.addRT(rt);
  
          rollingCounterInMinute.addSuccess(successCount);
          rollingCounterInMinute.addRT(rt);
      }
  
      @Override
      public void increaseBlockQps(int count) {
          rollingCounterInSecond.addBlock(count);
          rollingCounterInMinute.addBlock(count);
      }
  
      @Override
      public void increaseExceptionQps(int count) {
          rollingCounterInSecond.addException(count);
          rollingCounterInMinute.addException(count);
      }
  
      @Override
      public void increaseThreadNum() {
          curThreadNum.increment();
      }
  
      @Override
      public void decreaseThreadNum() {
          curThreadNum.decrement();
      }
  
      @Override
      public void debug() {
          rollingCounterInSecond.debug();
      }
  
      @Override
      public long tryOccupyNext(long currentTime, int acquireCount, double threshold) {
          double maxCount = threshold * IntervalProperty.INTERVAL / 1000;
          long currentBorrow = rollingCounterInSecond.waiting();
          if (currentBorrow >= maxCount) {
              return OccupyTimeoutProperty.getOccupyTimeout();
          }
  
          int windowLength = IntervalProperty.INTERVAL / SampleCountProperty.SAMPLE_COUNT;
          long earliestTime = currentTime - currentTime % windowLength + windowLength - IntervalProperty.INTERVAL;
  
          int idx = 0;
          /*
           * Note: here {@code currentPass} may be less than it really is NOW, because time difference
           * since call rollingCounterInSecond.pass(). So in high concurrency, the following code may
           * lead more tokens be borrowed.
           */
          long currentPass = rollingCounterInSecond.pass();
          while (earliestTime < currentTime) {
              long waitInMs = idx * windowLength + windowLength - currentTime % windowLength;
              if (waitInMs >= OccupyTimeoutProperty.getOccupyTimeout()) {
                  break;
              }
              long windowPass = rollingCounterInSecond.getWindowPass(earliestTime);
              if (currentPass + currentBorrow + acquireCount - windowPass <= maxCount) {
                  return waitInMs;
              }
              earliestTime += windowLength;
              currentPass -= windowPass;
              idx++;
          }
  
          return OccupyTimeoutProperty.getOccupyTimeout();
      }
  
      @Override
      public long waiting() {
          return rollingCounterInSecond.waiting();
      }
  
      @Override
      public void addWaitingRequest(long futureTime, int acquireCount) {
          rollingCounterInSecond.addWaiting(futureTime, acquireCount);
      }
  
      @Override
      public void addOccupiedPass(int acquireCount) {
          rollingCounterInMinute.addOccupiedPass(acquireCount);
          rollingCounterInMinute.addPass(acquireCount);
      }
  }
  