use running sum for ValueMetricProducer bucket
simplify ValueMetricProducer logic for pulled data
Test: unit test
Change-Id: Ic0a21a543166cc5c34c1fa505dba08d1fc2f510a
diff --git a/cmds/statsd/src/metrics/CountMetricProducer.cpp b/cmds/statsd/src/metrics/CountMetricProducer.cpp
index fc12013..ae297d9 100644
--- a/cmds/statsd/src/metrics/CountMetricProducer.cpp
+++ b/cmds/statsd/src/metrics/CountMetricProducer.cpp
@@ -188,7 +188,7 @@
void CountMetricProducer::onMatchedLogEventInternalLocked(
const size_t matcherIndex, const HashableDimensionKey& eventKey,
const map<string, HashableDimensionKey>& conditionKey, bool condition,
- const LogEvent& event, bool scheduledPull) {
+ const LogEvent& event) {
uint64_t eventTimeNs = event.GetTimestampNs();
flushIfNeededLocked(eventTimeNs);
diff --git a/cmds/statsd/src/metrics/CountMetricProducer.h b/cmds/statsd/src/metrics/CountMetricProducer.h
index 21bd9d6..8a17169 100644
--- a/cmds/statsd/src/metrics/CountMetricProducer.h
+++ b/cmds/statsd/src/metrics/CountMetricProducer.h
@@ -58,7 +58,7 @@
void onMatchedLogEventInternalLocked(
const size_t matcherIndex, const HashableDimensionKey& eventKey,
const std::map<std::string, HashableDimensionKey>& conditionKey, bool condition,
- const LogEvent& event, bool scheduledPull) override;
+ const LogEvent& event) override;
private:
void onDumpReportLocked(const uint64_t dumpTimeNs,
diff --git a/cmds/statsd/src/metrics/DurationMetricProducer.cpp b/cmds/statsd/src/metrics/DurationMetricProducer.cpp
index 9920f65..c268798 100644
--- a/cmds/statsd/src/metrics/DurationMetricProducer.cpp
+++ b/cmds/statsd/src/metrics/DurationMetricProducer.cpp
@@ -251,7 +251,7 @@
void DurationMetricProducer::onMatchedLogEventInternalLocked(
const size_t matcherIndex, const HashableDimensionKey& eventKey,
const map<string, HashableDimensionKey>& conditionKeys, bool condition,
- const LogEvent& event, bool scheduledPull) {
+ const LogEvent& event) {
flushIfNeededLocked(event.GetTimestampNs());
if (matcherIndex == mStopAllIndex) {
diff --git a/cmds/statsd/src/metrics/DurationMetricProducer.h b/cmds/statsd/src/metrics/DurationMetricProducer.h
index e509af4..14504c1 100644
--- a/cmds/statsd/src/metrics/DurationMetricProducer.h
+++ b/cmds/statsd/src/metrics/DurationMetricProducer.h
@@ -57,7 +57,7 @@
void onMatchedLogEventInternalLocked(
const size_t matcherIndex, const HashableDimensionKey& eventKey,
const std::map<std::string, HashableDimensionKey>& conditionKeys, bool condition,
- const LogEvent& event, bool scheduledPull) override;
+ const LogEvent& event) override;
private:
void onDumpReportLocked(const uint64_t dumpTimeNs,
diff --git a/cmds/statsd/src/metrics/EventMetricProducer.cpp b/cmds/statsd/src/metrics/EventMetricProducer.cpp
index 217aff0..bcecf16 100644
--- a/cmds/statsd/src/metrics/EventMetricProducer.cpp
+++ b/cmds/statsd/src/metrics/EventMetricProducer.cpp
@@ -122,7 +122,7 @@
void EventMetricProducer::onMatchedLogEventInternalLocked(
const size_t matcherIndex, const HashableDimensionKey& eventKey,
const std::map<std::string, HashableDimensionKey>& conditionKey, bool condition,
- const LogEvent& event, bool scheduledPull) {
+ const LogEvent& event) {
if (!condition) {
return;
}
diff --git a/cmds/statsd/src/metrics/EventMetricProducer.h b/cmds/statsd/src/metrics/EventMetricProducer.h
index 75ccf47..49ba9d8 100644
--- a/cmds/statsd/src/metrics/EventMetricProducer.h
+++ b/cmds/statsd/src/metrics/EventMetricProducer.h
@@ -53,7 +53,7 @@
void onMatchedLogEventInternalLocked(
const size_t matcherIndex, const HashableDimensionKey& eventKey,
const std::map<std::string, HashableDimensionKey>& conditionKey, bool condition,
- const LogEvent& event, bool scheduledPull) override;
+ const LogEvent& event) override;
void onDumpReportLocked(const uint64_t dumpTimeNs,
android::util::ProtoOutputStream* protoOutput) override;
diff --git a/cmds/statsd/src/metrics/GaugeMetricProducer.cpp b/cmds/statsd/src/metrics/GaugeMetricProducer.cpp
index 55d84e0..fffb2bf 100644
--- a/cmds/statsd/src/metrics/GaugeMetricProducer.cpp
+++ b/cmds/statsd/src/metrics/GaugeMetricProducer.cpp
@@ -187,7 +187,7 @@
return;
}
for (const auto& data : allData) {
- onMatchedLogEventLocked(0, *data, false /*scheduledPull*/);
+ onMatchedLogEventLocked(0, *data);
}
flushIfNeededLocked(eventTime);
}
@@ -211,7 +211,7 @@
std::lock_guard<std::mutex> lock(mMutex);
for (const auto& data : allData) {
- onMatchedLogEventLocked(0, *data, true /*scheduledPull*/);
+ onMatchedLogEventLocked(0, *data);
}
}
@@ -238,7 +238,7 @@
void GaugeMetricProducer::onMatchedLogEventInternalLocked(
const size_t matcherIndex, const HashableDimensionKey& eventKey,
const map<string, HashableDimensionKey>& conditionKey, bool condition,
- const LogEvent& event, bool scheduledPull) {
+ const LogEvent& event) {
if (condition == false) {
return;
}
diff --git a/cmds/statsd/src/metrics/GaugeMetricProducer.h b/cmds/statsd/src/metrics/GaugeMetricProducer.h
index e4bda02..ee4f40c 100644
--- a/cmds/statsd/src/metrics/GaugeMetricProducer.h
+++ b/cmds/statsd/src/metrics/GaugeMetricProducer.h
@@ -66,7 +66,7 @@
void onMatchedLogEventInternalLocked(
const size_t matcherIndex, const HashableDimensionKey& eventKey,
const std::map<std::string, HashableDimensionKey>& conditionKey, bool condition,
- const LogEvent& event, bool scheduledPull) override;
+ const LogEvent& event) override;
private:
void onDumpReportLocked(const uint64_t dumpTimeNs,
diff --git a/cmds/statsd/src/metrics/MetricProducer.cpp b/cmds/statsd/src/metrics/MetricProducer.cpp
index 5a0a7c7f..f38f3df 100644
--- a/cmds/statsd/src/metrics/MetricProducer.cpp
+++ b/cmds/statsd/src/metrics/MetricProducer.cpp
@@ -21,8 +21,7 @@
using std::map;
-void MetricProducer::onMatchedLogEventLocked(const size_t matcherIndex, const LogEvent& event,
- bool scheduledPull) {
+void MetricProducer::onMatchedLogEventLocked(const size_t matcherIndex, const LogEvent& event) {
uint64_t eventTimeNs = event.GetTimestampNs();
// this is old event, maybe statsd restarted?
if (eventTimeNs < mStartTimeNs) {
@@ -60,8 +59,7 @@
condition = mCondition;
}
- onMatchedLogEventInternalLocked(matcherIndex, eventKey, conditionKeys, condition, event,
- scheduledPull);
+ onMatchedLogEventInternalLocked(matcherIndex, eventKey, conditionKeys, condition, event);
}
} // namespace statsd
diff --git a/cmds/statsd/src/metrics/MetricProducer.h b/cmds/statsd/src/metrics/MetricProducer.h
index ef2ef29..269bd43 100644
--- a/cmds/statsd/src/metrics/MetricProducer.h
+++ b/cmds/statsd/src/metrics/MetricProducer.h
@@ -54,9 +54,9 @@
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, bool scheduledPull) {
+ void onMatchedLogEvent(const size_t matcherIndex, const LogEvent& event) {
std::lock_guard<std::mutex> lock(mMutex);
- onMatchedLogEventLocked(matcherIndex, event, scheduledPull);
+ onMatchedLogEventLocked(matcherIndex, event);
}
void onConditionChanged(const bool condition, const uint64_t eventTime) {
@@ -155,11 +155,10 @@
virtual void onMatchedLogEventInternalLocked(
const size_t matcherIndex, const HashableDimensionKey& eventKey,
const std::map<std::string, HashableDimensionKey>& conditionKey, bool condition,
- const LogEvent& event, bool scheduledPull) = 0;
+ const LogEvent& event) = 0;
// Consume the parsed stats log entry that already matched the "what" of the metric.
- void onMatchedLogEventLocked(const size_t matcherIndex, const LogEvent& event,
- bool scheduledPull);
+ void onMatchedLogEventLocked(const size_t matcherIndex, const LogEvent& event);
mutable std::mutex mMutex;
};
diff --git a/cmds/statsd/src/metrics/MetricsManager.cpp b/cmds/statsd/src/metrics/MetricsManager.cpp
index 0510fff..b0f0135 100644
--- a/cmds/statsd/src/metrics/MetricsManager.cpp
+++ b/cmds/statsd/src/metrics/MetricsManager.cpp
@@ -162,8 +162,7 @@
auto& metricList = pair->second;
for (const int metricIndex : metricList) {
// pushed metrics are never scheduled pulls
- mAllMetricProducers[metricIndex]->onMatchedLogEvent(i, event,
- false /* schedulePull */);
+ mAllMetricProducers[metricIndex]->onMatchedLogEvent(i, event);
}
}
}
diff --git a/cmds/statsd/src/metrics/ValueMetricProducer.cpp b/cmds/statsd/src/metrics/ValueMetricProducer.cpp
index c20c302..aabe5af 100644
--- a/cmds/statsd/src/metrics/ValueMetricProducer.cpp
+++ b/cmds/statsd/src/metrics/ValueMetricProducer.cpp
@@ -185,9 +185,13 @@
mCondition = condition;
if (eventTime < mCurrentBucketStartTimeNs) {
+ VLOG("Skip event due to late arrival: %lld vs %lld", (long long)eventTime,
+ (long long)mCurrentBucketStartTimeNs);
return;
}
+ flushIfNeededLocked(eventTime);
+
if (mPullTagId != -1) {
if (mCondition == true) {
mStatsPullerManager->RegisterReceiver(mPullTagId, this,
@@ -202,9 +206,8 @@
return;
}
for (const auto& data : allData) {
- onMatchedLogEventLocked(0, *data, false);
+ onMatchedLogEventLocked(0, *data);
}
- flushIfNeededLocked(eventTime);
}
return;
}
@@ -217,15 +220,22 @@
if (allData.size() == 0) {
return;
}
- uint64_t eventTime = allData.at(0)->GetTimestampNs();
- // alarm is not accurate and might drift.
- if (eventTime > mCurrentBucketStartTimeNs + mBucketSizeNs * 3 / 2) {
- flushIfNeededLocked(eventTime);
- }
+ // For scheduled pulled data, the effective event time is snap to the nearest
+ // bucket boundary to make bucket finalize.
+ uint64_t realEventTime = allData.at(0)->GetTimestampNs();
+ uint64_t eventTime = mStartTimeNs + ((realEventTime - mStartTimeNs)/mBucketSizeNs) * mBucketSizeNs;
+
+ mCondition = false;
for (const auto& data : allData) {
- onMatchedLogEventLocked(0, *data, true);
+ data->setTimestampNs(eventTime-1);
+ onMatchedLogEventLocked(0, *data);
}
- flushIfNeededLocked(eventTime);
+
+ mCondition = true;
+ for (const auto& data : allData) {
+ data->setTimestampNs(eventTime);
+ onMatchedLogEventLocked(0, *data);
+ }
}
}
@@ -253,7 +263,7 @@
void ValueMetricProducer::onMatchedLogEventInternalLocked(
const size_t matcherIndex, const HashableDimensionKey& eventKey,
const map<string, HashableDimensionKey>& conditionKey, bool condition,
- const LogEvent& event, bool scheduledPull) {
+ const LogEvent& event) {
uint64_t eventTimeNs = event.GetTimestampNs();
if (eventTimeNs < mCurrentBucketStartTimeNs) {
VLOG("Skip event due to late arrival: %lld vs %lld", (long long)eventTimeNs,
@@ -261,6 +271,8 @@
return;
}
+ flushIfNeededLocked(eventTimeNs);
+
if (hitGuardRailLocked(eventKey)) {
return;
}
@@ -268,36 +280,21 @@
long value = get_value(event);
- if (mPullTagId != -1) {
- if (scheduledPull) {
- // scheduled pull always sets beginning of current bucket and end
- // of next bucket
- if (interval.raw.size() > 0) {
- interval.raw.back().second = value;
- } else {
- interval.raw.push_back(make_pair(value, value));
- }
- Interval& nextInterval = mNextSlicedBucket[eventKey];
- if (nextInterval.raw.size() == 0) {
- nextInterval.raw.push_back(make_pair(value, 0));
- } else {
- nextInterval.raw.front().first = value;
- }
+ if (mPullTagId != -1) { // for pulled events
+ if (mCondition == true) {
+ interval.start = value;
+ interval.startUpdated = true;
} else {
- if (mCondition == true) {
- interval.raw.push_back(make_pair(value, 0));
+ if (interval.startUpdated) {
+ interval.sum += (value - interval.start);
+ interval.startUpdated = false;
} else {
- if (interval.raw.size() != 0) {
- interval.raw.back().second = value;
- } else {
- interval.tainted = true;
- VLOG("Data on condition true missing!");
- }
+ VLOG("No start for matching end %ld", value);
+ interval.tainted += 1;
}
}
- } else {
- flushIfNeededLocked(eventTimeNs);
- interval.raw.push_back(make_pair(value, 0));
+ } else { // for pushed events
+ interval.sum += value;
}
}
@@ -327,27 +324,16 @@
int tainted = 0;
for (const auto& slice : mCurrentSlicedBucket) {
- long value = 0;
- if (mPullTagId != -1) {
- for (const auto& pair : slice.second.raw) {
- value += (pair.second - pair.first);
- }
- } else {
- for (const auto& pair : slice.second.raw) {
- value += pair.first;
- }
- }
tainted += slice.second.tainted;
- info.mValue = value;
- VLOG(" %s, %ld, %d", slice.first.c_str(), value, tainted);
+ info.mValue = slice.second.sum;
// it will auto create new vector of ValuebucketInfo if the key is not found.
auto& bucketList = mPastBuckets[slice.first];
bucketList.push_back(info);
}
+ VLOG("%d tainted pairs in the bucket", tainted);
// Reset counters
- mCurrentSlicedBucket.swap(mNextSlicedBucket);
- mNextSlicedBucket.clear();
+ mCurrentSlicedBucket.clear();
int64_t numBucketsForward = (eventTimeNs - mCurrentBucketStartTimeNs) / mBucketSizeNs;
mCurrentBucketStartTimeNs = mCurrentBucketStartTimeNs + numBucketsForward * mBucketSizeNs;
diff --git a/cmds/statsd/src/metrics/ValueMetricProducer.h b/cmds/statsd/src/metrics/ValueMetricProducer.h
index 8d60ff6..4c49927 100644
--- a/cmds/statsd/src/metrics/ValueMetricProducer.h
+++ b/cmds/statsd/src/metrics/ValueMetricProducer.h
@@ -56,7 +56,7 @@
void onMatchedLogEventInternalLocked(
const size_t matcherIndex, const HashableDimensionKey& eventKey,
const std::map<std::string, HashableDimensionKey>& conditionKey, bool condition,
- const LogEvent& event, bool scheduledPull) override;
+ const LogEvent& event) override;
private:
void onDumpReportLocked(const uint64_t dumpTimeNs,
@@ -89,14 +89,19 @@
// internal state of a bucket.
typedef struct {
- std::vector<std::pair<long, long>> raw;
- bool tainted;
+ // Pulled data always come in pair of <start, end>. This holds the value
+ // for start. The diff (end - start) is added to sum.
+ long start;
+ // Whether the start data point is updated
+ bool startUpdated;
+ // If end data point comes before the start, record this pair as tainted
+ // and the value is not added to the running sum.
+ int tainted;
+ // Running sum of known pairs in this bucket
+ long sum;
} 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.
// TODO: Add a lock to mPastBuckets.