cmds/statsd/src/metrics/ValueMetricProducer.cpp

/*
 * Copyright (C) 2017 The Android Open Source Project
 *
 * 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.
 */

#define DEBUG true  // STOPSHIP if true
#include "Log.h"

#include "ValueMetricProducer.h"

#include <cutils/log.h>
#include <limits.h>
#include <stdlib.h>

using std::map;
using std::unordered_map;
using std::list;
using std::make_shared;
using std::shared_ptr;
using std::unique_ptr;

namespace android {
namespace os {
namespace statsd {

// ValueMetric has a minimum bucket size of 10min so that we don't pull too frequently
ValueMetricProducer::ValueMetricProducer(const ValueMetric& metric, const int conditionIndex,
                                         const sp<ConditionWizard>& wizard)
    : MetricProducer((time(nullptr) / 600 * 600 * NANO_SECONDS_IN_A_SECOND), conditionIndex,
                     wizard),
      mMetric(metric),
      mPullCode(mStatsPullerManager.GetPullCode(mMetric.what())) {
  // TODO: valuemetric for pushed events may need unlimited bucket length
  mBucketSizeNs = mMetric.bucket().bucket_size_millis() * 1000 * 1000;

  mDimension.insert(mDimension.begin(), metric.dimension().begin(), metric.dimension().end());

  if (metric.links().size() > 0) {
    mConditionLinks.insert(mConditionLinks.begin(), metric.links().begin(),
                           metric.links().end());
    mConditionSliced = true;
  }

  if (!metric.has_condition() && mPullCode != -1) {
    mStatsPullerManager.RegisterReceiver(mPullCode, this, metric.bucket().bucket_size_millis());
  }

  VLOG("value metric %lld created. bucket size %lld start_time: %lld", metric.metric_id(),
       (long long)mBucketSizeNs, (long long)mStartTimeNs);
}

ValueMetricProducer::~ValueMetricProducer() {
  VLOG("~ValueMetricProducer() called");
}

void ValueMetricProducer::finish() {
  // TODO: write the StatsLogReport to dropbox using
  // DropboxWriter.
}

static void addSlicedCounterToReport(StatsLogReport_ValueMetricDataWrapper& wrapper,
                                     const vector<KeyValuePair>& key,
                                     const vector<ValueBucketInfo>& buckets) {
  ValueMetricData* data = wrapper.add_data();
  for (const auto& kv : key) {
    data->add_dimension()->CopyFrom(kv);
  }
  for (const auto& bucket : buckets) {
    data->add_bucket_info()->CopyFrom(bucket);
    VLOG("\t bucket [%lld - %lld] value: %lld", bucket.start_bucket_nanos(),
         bucket.end_bucket_nanos(), bucket.value());
  }
}

void ValueMetricProducer::onSlicedConditionMayChange(const uint64_t eventTime) {
    VLOG("Metric %lld onSlicedConditionMayChange", mMetric.metric_id());
}

StatsLogReport ValueMetricProducer::onDumpReport() {
  VLOG("metric %lld dump report now...", mMetric.metric_id());

  StatsLogReport report;
  report.set_metric_id(mMetric.metric_id());
  report.set_start_report_nanos(mStartTimeNs);

  // Dump current bucket if it's stale.
  // If current bucket is still on-going, don't force dump current bucket.
  // In finish(), We can force dump current bucket.
  //    flush_if_needed(time(nullptr) * NANO_SECONDS_IN_A_SECOND);
  report.set_end_report_nanos(mCurrentBucketStartTimeNs);

  StatsLogReport_ValueMetricDataWrapper* wrapper = report.mutable_value_metrics();

  for (const auto& pair : mPastBuckets) {
    const HashableDimensionKey& hashableKey = pair.first;
    auto it = mDimensionKeyMap.find(hashableKey);
    if (it == mDimensionKeyMap.end()) {
      ALOGE("Dimension key %s not found?!?! skip...", hashableKey.c_str());
      continue;
    }

    VLOG("  dimension key %s", hashableKey.c_str());
    addSlicedCounterToReport(*wrapper, it->second, pair.second);
  }
  return report;
  // TODO: Clear mPastBuckets, mDimensionKeyMap once the report is dumped.
}

void ValueMetricProducer::onConditionChanged(const bool condition, const uint64_t eventTime) {
    mCondition = condition;

    if (mPullCode != -1) {
        vector<shared_ptr<LogEvent>> allData = mStatsPullerManager.Pull(mPullCode, eventTime);
        if (mCondition == true) {
            mStatsPullerManager.RegisterReceiver(mPullCode, this,
                                                 mMetric.bucket().bucket_size_millis());
        } else if (mCondition == ConditionState::kFalse) {
            mStatsPullerManager.UnRegisterReceiver(mPullCode, this);
        }
        if (allData.size() == 0) {
            return;
        }
        AutoMutex _l(mLock);
        if (allData.size() == 0) {
            return;
        }
        for (const auto& data : allData) {
            onMatchedLogEvent(0, *data, false);
        }
        flush_if_needed(eventTime);
    }
    return;
}

void ValueMetricProducer::onDataPulled(const std::vector<std::shared_ptr<LogEvent>>& allData) {
    if (mCondition == ConditionState::kTrue || !mMetric.has_condition()) {
        AutoMutex _l(mLock);
        if (allData.size() == 0) {
            return;
        }
        uint64_t eventTime = allData.at(0)->GetTimestampNs();
        for (const auto& data : allData) {
            onMatchedLogEvent(0, *data, true);
        }
        flush_if_needed(eventTime);
    }
}

void ValueMetricProducer::onMatchedLogEventInternal(
    const size_t matcherIndex, const HashableDimensionKey& eventKey,
    const map<string, HashableDimensionKey>& conditionKey, bool condition,
    const LogEvent& event, bool scheduledPull) {
  uint64_t eventTimeNs = event.GetTimestampNs();
  if (eventTimeNs < mCurrentBucketStartTimeNs) {
      VLOG("Skip event due to late arrival: %lld vs %lld", (long long)eventTimeNs,
           (long long)mCurrentBucketStartTimeNs);
      return;
  }

  Interval& interval = mCurrentSlicedBucket[eventKey];

  long value = get_value(event);

  if (scheduledPull) {
    if (interval.raw.size() > 0) {
      interval.raw.back().second = value;
    } else {
      interval.raw.push_back(std::make_pair(value, value));
    }
    mNextSlicedBucket[eventKey].raw[0].first = value;
  } else {
    if (mCondition == ConditionState::kTrue) {
      interval.raw.push_back(std::make_pair(value, 0));
    } else {
      if (interval.raw.size() != 0) {
        interval.raw.back().second = value;
      }
    }
  }
  if (mPullCode == -1) {
      flush_if_needed(eventTimeNs);
  }
}

long ValueMetricProducer::get_value(const LogEvent& event) {
  status_t err = NO_ERROR;
  long val = event.GetLong(mMetric.value_field(), &err);
  if (err == NO_ERROR) {
    return val;
  } else {
    VLOG("Can't find value in message.");
    return 0;
  }
}

void ValueMetricProducer::flush_if_needed(const uint64_t eventTimeNs) {
    if (mCurrentBucketStartTimeNs + mBucketSizeNs > eventTimeNs) {
        VLOG("eventTime is %lld, less than next bucket start time %lld", (long long)eventTimeNs,
             (long long)(mCurrentBucketStartTimeNs + mBucketSizeNs));
        return;
    }

    VLOG("finalizing bucket for %ld, dumping %d slices", (long)mCurrentBucketStartTimeNs,
         (int)mCurrentSlicedBucket.size());
    ValueBucketInfo info;
    info.set_start_bucket_nanos(mCurrentBucketStartTimeNs);
    info.set_end_bucket_nanos(mCurrentBucketStartTimeNs + mBucketSizeNs);

    for (const auto& slice : mCurrentSlicedBucket) {
      long value = 0;
      for (const auto& pair : slice.second.raw) {
        value += pair.second - pair.first;
      }
      info.set_value(value);
      VLOG(" %s, %ld", slice.first.c_str(), value);
      // it will auto create new vector of ValuebucketInfo if the key is not found.
      auto& bucketList = mPastBuckets[slice.first];
      bucketList.push_back(info);
    }

    // Reset counters
    mCurrentSlicedBucket.swap(mNextSlicedBucket);
    mNextSlicedBucket.clear();
    int64_t numBucketsForward = (eventTimeNs - mCurrentBucketStartTimeNs) / mBucketSizeNs;
    if (numBucketsForward >1) {
        VLOG("Skipping forward %lld buckets", (long long)numBucketsForward);
    }
    mCurrentBucketStartTimeNs = mCurrentBucketStartTimeNs + numBucketsForward * mBucketSizeNs;
    VLOG("metric %lld: new bucket start time: %lld", mMetric.metric_id(),
         (long long)mCurrentBucketStartTimeNs);
}

}  // namespace statsd
}  // namespace os
}  // namespace android