reretry ValueMetric implementation and pulling mechanism
Note:
This is for value metric. The default operations is sum the diffs.
The test uses kernel wake lock, which also needs dimension by kernel
wake lock name.
The test is a bit cumbersome as it needs StatsCompanionService to do
the alarm, which is not exact alarm.
The internal state of a slice of bucket would look something like this:
4:ipc0000005e_727_android.hardwar
0 0
4:SensorService_wakelock
40 64
4:ipc0000005c_727_android.hardwar
...
Test: manual test on device.
Change-Id: I2ed0ac7d3c5fcba8b7611d46f38a38ffd8bdc92a
diff --git a/cmds/statsd/src/metrics/CountMetricProducer.cpp b/cmds/statsd/src/metrics/CountMetricProducer.cpp
index 10816f6..9f8558d 100644
--- a/cmds/statsd/src/metrics/CountMetricProducer.cpp
+++ b/cmds/statsd/src/metrics/CountMetricProducer.cpp
@@ -135,7 +135,7 @@
void CountMetricProducer::onMatchedLogEventInternal(
const size_t matcherIndex, const HashableDimensionKey& eventKey,
const map<string, HashableDimensionKey>& conditionKey, bool condition,
- const LogEvent& event) {
+ const LogEvent& event, bool scheduledPull) {
uint64_t eventTimeNs = event.GetTimestampNs();
flushCounterIfNeeded(eventTimeNs);
diff --git a/cmds/statsd/src/metrics/CountMetricProducer.h b/cmds/statsd/src/metrics/CountMetricProducer.h
index 5d1889a..80e80d9 100644
--- a/cmds/statsd/src/metrics/CountMetricProducer.h
+++ b/cmds/statsd/src/metrics/CountMetricProducer.h
@@ -59,7 +59,8 @@
protected:
void onMatchedLogEventInternal(const size_t matcherIndex, const HashableDimensionKey& eventKey,
const std::map<std::string, HashableDimensionKey>& conditionKey,
- bool condition, const LogEvent& event) override;
+ bool condition, const LogEvent& event,
+ bool scheduledPull) override;
private:
const CountMetric mMetric;
diff --git a/cmds/statsd/src/metrics/DurationMetricProducer.cpp b/cmds/statsd/src/metrics/DurationMetricProducer.cpp
index dfed275..340f503 100644
--- a/cmds/statsd/src/metrics/DurationMetricProducer.cpp
+++ b/cmds/statsd/src/metrics/DurationMetricProducer.cpp
@@ -168,7 +168,7 @@
void DurationMetricProducer::onMatchedLogEventInternal(
const size_t matcherIndex, const HashableDimensionKey& eventKey,
const map<string, HashableDimensionKey>& conditionKeys, bool condition,
- const LogEvent& event) {
+ const LogEvent& event, bool scheduledPull) {
flushIfNeeded(event.GetTimestampNs());
if (matcherIndex == mStopAllIndex) {
diff --git a/cmds/statsd/src/metrics/DurationMetricProducer.h b/cmds/statsd/src/metrics/DurationMetricProducer.h
index 5b302b4..febf25d 100644
--- a/cmds/statsd/src/metrics/DurationMetricProducer.h
+++ b/cmds/statsd/src/metrics/DurationMetricProducer.h
@@ -62,7 +62,8 @@
protected:
void onMatchedLogEventInternal(const size_t matcherIndex, const HashableDimensionKey& eventKey,
const std::map<std::string, HashableDimensionKey>& conditionKeys,
- bool condition, const LogEvent& event) override;
+ bool condition, const LogEvent& event,
+ bool scheduledPull) override;
private:
const DurationMetric mMetric;
diff --git a/cmds/statsd/src/metrics/EventMetricProducer.cpp b/cmds/statsd/src/metrics/EventMetricProducer.cpp
index dd23d66..d714179 100644
--- a/cmds/statsd/src/metrics/EventMetricProducer.cpp
+++ b/cmds/statsd/src/metrics/EventMetricProducer.cpp
@@ -112,7 +112,7 @@
void EventMetricProducer::onMatchedLogEventInternal(
const size_t matcherIndex, const HashableDimensionKey& eventKey,
const std::map<std::string, HashableDimensionKey>& conditionKey, bool condition,
- const LogEvent& event) {
+ const LogEvent& event, bool scheduledPull) {
if (!condition) {
return;
diff --git a/cmds/statsd/src/metrics/EventMetricProducer.h b/cmds/statsd/src/metrics/EventMetricProducer.h
index 72df0a7..7dd0e38 100644
--- a/cmds/statsd/src/metrics/EventMetricProducer.h
+++ b/cmds/statsd/src/metrics/EventMetricProducer.h
@@ -41,7 +41,7 @@
void onMatchedLogEventInternal(const size_t matcherIndex, const HashableDimensionKey& eventKey,
const std::map<std::string, HashableDimensionKey>& conditionKey,
- bool condition, const LogEvent& event) override;
+ bool condition, const LogEvent& event, bool scheduledPull) override;
void onConditionChanged(const bool conditionMet, const uint64_t eventTime) override;
diff --git a/cmds/statsd/src/metrics/MetricProducer.cpp b/cmds/statsd/src/metrics/MetricProducer.cpp
index 3c8ce6e..535f4a2 100644
--- a/cmds/statsd/src/metrics/MetricProducer.cpp
+++ b/cmds/statsd/src/metrics/MetricProducer.cpp
@@ -21,7 +21,8 @@
using std::map;
-void MetricProducer::onMatchedLogEvent(const size_t matcherIndex, const LogEvent& event) {
+void MetricProducer::onMatchedLogEvent(const size_t matcherIndex, const LogEvent& event,
+ bool scheduledPull) {
uint64_t eventTimeNs = event.GetTimestampNs();
// this is old event, maybe statsd restarted?
if (eventTimeNs < mStartTimeNs) {
@@ -59,7 +60,8 @@
condition = mCondition;
}
- onMatchedLogEventInternal(matcherIndex, eventKey, conditionKeys, condition, event);
+ onMatchedLogEventInternal(matcherIndex, eventKey, conditionKeys, condition, event,
+ scheduledPull);
}
} // namespace statsd
diff --git a/cmds/statsd/src/metrics/MetricProducer.h b/cmds/statsd/src/metrics/MetricProducer.h
index 2083695..3b117ec 100644
--- a/cmds/statsd/src/metrics/MetricProducer.h
+++ b/cmds/statsd/src/metrics/MetricProducer.h
@@ -48,7 +48,7 @@
virtual ~MetricProducer(){};
// Consume the parsed stats log entry that already matched the "what" of the metric.
- void onMatchedLogEvent(const size_t matcherIndex, const LogEvent& event);
+ void onMatchedLogEvent(const size_t matcherIndex, const LogEvent& event, bool scheduledPull);
virtual void onConditionChanged(const bool condition, const uint64_t eventTime) = 0;
@@ -107,7 +107,7 @@
virtual void onMatchedLogEventInternal(
const size_t matcherIndex, const HashableDimensionKey& eventKey,
const std::map<std::string, HashableDimensionKey>& conditionKey, bool condition,
- const LogEvent& event) = 0;
+ const LogEvent& event, bool scheduledPull) = 0;
};
} // namespace statsd
diff --git a/cmds/statsd/src/metrics/MetricsManager.cpp b/cmds/statsd/src/metrics/MetricsManager.cpp
index d1df8aa..521fcc3 100644
--- a/cmds/statsd/src/metrics/MetricsManager.cpp
+++ b/cmds/statsd/src/metrics/MetricsManager.cpp
@@ -145,7 +145,8 @@
if (pair != mTrackerToMetricMap.end()) {
auto& metricList = pair->second;
for (const int metricIndex : metricList) {
- mAllMetricProducers[metricIndex]->onMatchedLogEvent(i, event);
+ // pushed metrics are never scheduled pulls
+ mAllMetricProducers[metricIndex]->onMatchedLogEvent(i, event, false);
}
}
}
diff --git a/cmds/statsd/src/metrics/ValueMetricProducer.cpp b/cmds/statsd/src/metrics/ValueMetricProducer.cpp
new file mode 100644
index 0000000..cb6166d
--- /dev/null
+++ b/cmds/statsd/src/metrics/ValueMetricProducer.cpp
@@ -0,0 +1,246 @@
+/*
+ * 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
\ No newline at end of file
diff --git a/cmds/statsd/src/metrics/ValueMetricProducer.h b/cmds/statsd/src/metrics/ValueMetricProducer.h
new file mode 100644
index 0000000..4f17913
--- /dev/null
+++ b/cmds/statsd/src/metrics/ValueMetricProducer.h
@@ -0,0 +1,92 @@
+/*
+ * 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.
+ */
+
+#pragma once
+
+#include <utils/threads.h>
+#include <list>
+#include "../condition/ConditionTracker.h"
+#include "../external/PullDataReceiver.h"
+#include "../external/StatsPullerManager.h"
+#include "CountAnomalyTracker.h"
+#include "MetricProducer.h"
+#include "frameworks/base/cmds/statsd/src/stats_log.pb.h"
+#include "frameworks/base/cmds/statsd/src/statsd_config.pb.h"
+
+namespace android {
+namespace os {
+namespace statsd {
+
+class ValueMetricProducer : public virtual MetricProducer, public virtual PullDataReceiver {
+public:
+ ValueMetricProducer(const ValueMetric& valueMetric, const int conditionIndex,
+ const sp<ConditionWizard>& wizard);
+
+ virtual ~ValueMetricProducer();
+
+ void onConditionChanged(const bool condition, const uint64_t eventTime) override;
+
+ void finish() override;
+
+ StatsLogReport onDumpReport() override;
+
+ void onSlicedConditionMayChange(const uint64_t eventTime);
+
+ void onDataPulled(const std::vector<std::shared_ptr<LogEvent>>& data) override;
+ // TODO: Implement this later.
+ size_t byteSize() override{return 0;};
+
+ // TODO: Implement this later.
+ virtual void notifyAppUpgrade(const string& apk, const int uid, const int version) override{};
+ // TODO: Implement this later.
+ virtual void notifyAppRemoved(const string& apk, const int uid) override{};
+
+protected:
+ void onMatchedLogEventInternal(const size_t matcherIndex, const HashableDimensionKey& eventKey,
+ const std::map<std::string, HashableDimensionKey>& conditionKey,
+ bool condition, const LogEvent& event,
+ bool scheduledPull) override;
+
+private:
+ const ValueMetric mMetric;
+
+ StatsPullerManager& mStatsPullerManager = StatsPullerManager::GetInstance();
+
+ Mutex mLock;
+
+ const int mPullCode;
+
+ // internal state of a bucket.
+ typedef struct {
+ std::vector<std::pair<long, long>> raw;
+ } Interval;
+
+ std::unordered_map<HashableDimensionKey, Interval> mCurrentSlicedBucket;
+ // If condition is true and pulling on schedule, the previous bucket value needs to be carried
+ // over to the next bucket.
+ std::unordered_map<HashableDimensionKey, Interval> mNextSlicedBucket;
+
+ // Save the past buckets and we can clear when the StatsLogReport is dumped.
+ std::unordered_map<HashableDimensionKey, std::vector<ValueBucketInfo>> mPastBuckets;
+
+ long get_value(const LogEvent& event);
+
+ void flush_if_needed(const uint64_t eventTimeNs);
+};
+
+} // namespace statsd
+} // namespace os
+} // namespace android
diff --git a/cmds/statsd/src/metrics/metrics_manager_util.cpp b/cmds/statsd/src/metrics/metrics_manager_util.cpp
index 3b3ffb7..3d4036e 100644
--- a/cmds/statsd/src/metrics/metrics_manager_util.cpp
+++ b/cmds/statsd/src/metrics/metrics_manager_util.cpp
@@ -16,11 +16,13 @@
#include "../condition/CombinationConditionTracker.h"
#include "../condition/SimpleConditionTracker.h"
+#include "../external/StatsPullerManager.h"
#include "../matchers/CombinationLogMatchingTracker.h"
#include "../matchers/SimpleLogMatchingTracker.h"
#include "CountMetricProducer.h"
#include "DurationMetricProducer.h"
#include "EventMetricProducer.h"
+#include "ValueMetricProducer.h"
#include "stats_util.h"
using std::set;
@@ -192,6 +194,7 @@
const int allMetricsCount =
config.count_metric_size() + config.duration_metric_size() + config.event_metric_size();
allMetricProducers.reserve(allMetricsCount);
+ StatsPullerManager& statsPullerManager = StatsPullerManager::GetInstance();
// Build MetricProducers for each metric defined in config.
// (1) build CountMetricProducer
@@ -307,6 +310,34 @@
allMetricProducers.push_back(eventMetric);
}
+ // value metrics
+ for (int i = 0; i < config.value_metric_size(); i++) {
+ const ValueMetric& metric = config.value_metric(i);
+ if (!metric.has_what()) {
+ ALOGW("cannot find what in ValueMetric %lld", metric.metric_id());
+ return false;
+ }
+
+ int pullCode = statsPullerManager.GetPullCode(metric.what());
+ if (pullCode == -1) {
+ ALOGW("cannot find %s in pulled metrics", metric.what().c_str());
+ return false;
+ }
+
+ sp<MetricProducer> valueProducer;
+ auto condition_it = conditionTrackerMap.find(metric.condition());
+ if (condition_it == conditionTrackerMap.end()) {
+ ALOGW("cannot find the Condition %s in the config", metric.condition().c_str());
+ return false;
+ }
+ int metricIndex = allMetricProducers.size();
+ valueProducer = new ValueMetricProducer(metric, condition_it->second, wizard);
+ // will create new vector if not exist before.
+ auto& metricList = conditionToMetricMap[condition_it->second];
+ metricList.push_back(metricIndex);
+
+ allMetricProducers.push_back(valueProducer);
+ }
return true;
}
diff --git a/cmds/statsd/src/metrics/metrics_manager_util.h b/cmds/statsd/src/metrics/metrics_manager_util.h
index c91b7fc..e089d065 100644
--- a/cmds/statsd/src/metrics/metrics_manager_util.h
+++ b/cmds/statsd/src/metrics/metrics_manager_util.h
@@ -21,6 +21,7 @@
#include <vector>
#include "../condition/ConditionTracker.h"
+#include "../external/StatsPullerManager.h"
#include "../matchers/LogMatchingTracker.h"
namespace android {