Merge "allow statsd pull based on event trigger"
This commit is contained in:
committed by
Android (Google) Code Review
commit
a5f28237e4
@@ -31,9 +31,6 @@ namespace android {
|
||||
namespace os {
|
||||
namespace statsd {
|
||||
|
||||
// Util function to build a hard coded config with test metrics.
|
||||
StatsdConfig build_fake_config();
|
||||
|
||||
/**
|
||||
* Keeps track of which configurations have been set from various sources.
|
||||
*/
|
||||
|
||||
@@ -71,11 +71,14 @@ const int FIELD_ID_END_BUCKET_ELAPSED_MILLIS = 8;
|
||||
GaugeMetricProducer::GaugeMetricProducer(const ConfigKey& key, const GaugeMetric& metric,
|
||||
const int conditionIndex,
|
||||
const sp<ConditionWizard>& wizard, const int pullTagId,
|
||||
const int triggerAtomId, const int atomId,
|
||||
const int64_t timeBaseNs, const int64_t startTimeNs,
|
||||
const sp<StatsPullerManager>& pullerManager)
|
||||
: MetricProducer(metric.id(), key, timeBaseNs, conditionIndex, wizard),
|
||||
mPullerManager(pullerManager),
|
||||
mPullTagId(pullTagId),
|
||||
mTriggerAtomId(triggerAtomId),
|
||||
mAtomId(atomId),
|
||||
mIsPulled(pullTagId != -1),
|
||||
mMinBucketSizeNs(metric.min_bucket_size_nanos()),
|
||||
mDimensionSoftLimit(StatsdStats::kAtomDimensionKeySizeLimitMap.find(pullTagId) !=
|
||||
@@ -272,12 +275,12 @@ void GaugeMetricProducer::onDumpReportLocked(const int64_t dumpTimeNs,
|
||||
uint64_t atomsToken =
|
||||
protoOutput->start(FIELD_TYPE_MESSAGE | FIELD_COUNT_REPEATED |
|
||||
FIELD_ID_ATOM);
|
||||
writeFieldValueTreeToStream(mTagId, *(atom.mFields), protoOutput);
|
||||
writeFieldValueTreeToStream(mAtomId, *(atom.mFields), protoOutput);
|
||||
protoOutput->end(atomsToken);
|
||||
}
|
||||
const bool truncateTimestamp =
|
||||
android::util::AtomsInfo::kNotTruncatingTimestampAtomWhiteList.find(
|
||||
mTagId) ==
|
||||
mAtomId) ==
|
||||
android::util::AtomsInfo::kNotTruncatingTimestampAtomWhiteList.end();
|
||||
for (const auto& atom : bucket.mGaugeAtoms) {
|
||||
const int64_t elapsedTimestampNs = truncateTimestamp ?
|
||||
@@ -410,7 +413,6 @@ void GaugeMetricProducer::onMatchedLogEventInternalLocked(
|
||||
return;
|
||||
}
|
||||
int64_t eventTimeNs = event.GetElapsedTimestampNs();
|
||||
mTagId = event.GetTagId();
|
||||
if (eventTimeNs < mCurrentBucketStartTimeNs) {
|
||||
VLOG("Gauge Skip event due to late arrival: %lld vs %lld", (long long)eventTimeNs,
|
||||
(long long)mCurrentBucketStartTimeNs);
|
||||
@@ -418,6 +420,11 @@ void GaugeMetricProducer::onMatchedLogEventInternalLocked(
|
||||
}
|
||||
flushIfNeededLocked(eventTimeNs);
|
||||
|
||||
if (mTriggerAtomId == event.GetTagId()) {
|
||||
pullLocked(eventTimeNs);
|
||||
return;
|
||||
}
|
||||
|
||||
// When gauge metric wants to randomly sample the output atom, we just simply use the first
|
||||
// gauge in the given bucket.
|
||||
if (mCurrentSlicedBucket->find(eventKey) != mCurrentSlicedBucket->end() &&
|
||||
|
||||
@@ -58,7 +58,8 @@ class GaugeMetricProducer : public virtual MetricProducer, public virtual PullDa
|
||||
public:
|
||||
GaugeMetricProducer(const ConfigKey& key, const GaugeMetric& gaugeMetric,
|
||||
const int conditionIndex, const sp<ConditionWizard>& wizard,
|
||||
const int pullTagId, const int64_t timeBaseNs, const int64_t startTimeNs,
|
||||
const int pullTagId, const int triggerAtomId, const int atomId,
|
||||
const int64_t timeBaseNs, const int64_t startTimeNs,
|
||||
const sp<StatsPullerManager>& pullerManager);
|
||||
|
||||
virtual ~GaugeMetricProducer();
|
||||
@@ -115,12 +116,16 @@ private:
|
||||
|
||||
void pullLocked(const int64_t timestampNs);
|
||||
|
||||
int mTagId;
|
||||
|
||||
sp<StatsPullerManager> mPullerManager;
|
||||
// tagId for pulled data. -1 if this is not pulled
|
||||
const int mPullTagId;
|
||||
|
||||
// tagId for atoms that trigger the pulling, if any
|
||||
const int mTriggerAtomId;
|
||||
|
||||
// tagId for output atom
|
||||
const int mAtomId;
|
||||
|
||||
// if this is pulled metric
|
||||
const bool mIsPulled;
|
||||
|
||||
@@ -169,6 +174,7 @@ private:
|
||||
FRIEND_TEST(GaugeMetricProducerTest, TestPulledWithUpgrade);
|
||||
FRIEND_TEST(GaugeMetricProducerTest, TestPulledEventsAnomalyDetection);
|
||||
FRIEND_TEST(GaugeMetricProducerTest, TestFirstBucket);
|
||||
FRIEND_TEST(GaugeMetricProducerTest, TestPullOnTrigger);
|
||||
};
|
||||
|
||||
} // namespace statsd
|
||||
|
||||
@@ -14,7 +14,7 @@
|
||||
* limitations under the License.
|
||||
*/
|
||||
|
||||
#define DEBUG true // STOPSHIP if true
|
||||
#define DEBUG false // STOPSHIP if true
|
||||
#include "Log.h"
|
||||
|
||||
#include "ValueMetricProducer.h"
|
||||
|
||||
@@ -82,6 +82,28 @@ bool handleMetricWithLogTrackers(const int64_t what, const int metricIndex,
|
||||
return true;
|
||||
}
|
||||
|
||||
bool handlePullMetricTriggerWithLogTrackers(
|
||||
const int64_t trigger, const int metricIndex,
|
||||
const vector<sp<LogMatchingTracker>>& allAtomMatchers,
|
||||
const unordered_map<int64_t, int>& logTrackerMap,
|
||||
unordered_map<int, std::vector<int>>& trackerToMetricMap, int& logTrackerIndex) {
|
||||
auto logTrackerIt = logTrackerMap.find(trigger);
|
||||
if (logTrackerIt == logTrackerMap.end()) {
|
||||
ALOGW("cannot find the AtomMatcher \"%lld\" in config", (long long)trigger);
|
||||
return false;
|
||||
}
|
||||
if (allAtomMatchers[logTrackerIt->second]->getAtomIds().size() > 1) {
|
||||
ALOGE("AtomMatcher \"%lld\" has more than one tag ids."
|
||||
"Trigger can only be one atom type.",
|
||||
(long long)trigger);
|
||||
return false;
|
||||
}
|
||||
logTrackerIndex = logTrackerIt->second;
|
||||
auto& metric_list = trackerToMetricMap[logTrackerIndex];
|
||||
metric_list.push_back(metricIndex);
|
||||
return true;
|
||||
}
|
||||
|
||||
bool handleMetricWithConditions(
|
||||
const int64_t condition, const int metricIndex,
|
||||
const unordered_map<int64_t, int>& conditionTrackerMap,
|
||||
@@ -502,13 +524,29 @@ bool initMetrics(const ConfigKey& key, const StatsdConfig& config, const int64_t
|
||||
}
|
||||
|
||||
sp<LogMatchingTracker> atomMatcher = allAtomMatchers.at(trackerIndex);
|
||||
// If it is pulled atom, it should be simple matcher with one tagId.
|
||||
// For GaugeMetric atom, it should be simple matcher with one tagId.
|
||||
if (atomMatcher->getAtomIds().size() != 1) {
|
||||
return false;
|
||||
}
|
||||
int atomTagId = *(atomMatcher->getAtomIds().begin());
|
||||
int pullTagId = statsPullerManager.PullerForMatcherExists(atomTagId) ? atomTagId : -1;
|
||||
|
||||
int triggerTrackerIndex;
|
||||
int triggerAtomId = -1;
|
||||
if (pullTagId != -1 && metric.has_trigger_event()) {
|
||||
// event_trigger should be used with ALL_CONDITION_CHANGES
|
||||
if (metric.sampling_type() != GaugeMetric::ALL_CONDITION_CHANGES) {
|
||||
return false;
|
||||
}
|
||||
if (!handlePullMetricTriggerWithLogTrackers(metric.trigger_event(), metricIndex,
|
||||
allAtomMatchers, logTrackerMap,
|
||||
trackerToMetricMap, triggerTrackerIndex)) {
|
||||
return false;
|
||||
}
|
||||
sp<LogMatchingTracker> triggerAtomMatcher = allAtomMatchers.at(triggerTrackerIndex);
|
||||
triggerAtomId = *(triggerAtomMatcher->getAtomIds().begin());
|
||||
}
|
||||
|
||||
int conditionIndex = -1;
|
||||
if (metric.has_condition()) {
|
||||
bool good = handleMetricWithConditions(
|
||||
@@ -524,9 +562,9 @@ bool initMetrics(const ConfigKey& key, const StatsdConfig& config, const int64_t
|
||||
}
|
||||
}
|
||||
|
||||
sp<MetricProducer> gaugeProducer =
|
||||
new GaugeMetricProducer(key, metric, conditionIndex, wizard, pullTagId,
|
||||
timeBaseTimeNs, currentTimeNs, pullerManager);
|
||||
sp<MetricProducer> gaugeProducer = new GaugeMetricProducer(
|
||||
key, metric, conditionIndex, wizard, pullTagId, triggerAtomId, atomTagId,
|
||||
timeBaseTimeNs, currentTimeNs, pullerManager);
|
||||
allMetricProducers.push_back(gaugeProducer);
|
||||
}
|
||||
for (int i = 0; i < config.no_report_metric_size(); ++i) {
|
||||
|
||||
@@ -217,6 +217,8 @@ message GaugeMetric {
|
||||
|
||||
optional int64 what = 2;
|
||||
|
||||
optional int64 trigger_event = 12;
|
||||
|
||||
optional FieldFilter gauge_fields_filter = 3;
|
||||
|
||||
optional int64 condition = 4;
|
||||
|
||||
@@ -66,7 +66,8 @@ TEST(GaugeMetricProducerTest, TestFirstBucket) {
|
||||
// statsd started long ago.
|
||||
// The metric starts in the middle of the bucket
|
||||
GaugeMetricProducer gaugeProducer(kConfigKey, metric, -1 /*-1 meaning no condition*/, wizard,
|
||||
-1, 5, 600 * NS_PER_SEC + NS_PER_SEC/2, pullerManager);
|
||||
-1, -1, tagId, 5, 600 * NS_PER_SEC + NS_PER_SEC / 2,
|
||||
pullerManager);
|
||||
|
||||
EXPECT_EQ(600500000000, gaugeProducer.mCurrentBucketStartTimeNs);
|
||||
EXPECT_EQ(10, gaugeProducer.mCurrentBucketNum);
|
||||
@@ -100,8 +101,8 @@ TEST(GaugeMetricProducerTest, TestPulledEventsNoCondition) {
|
||||
}));
|
||||
|
||||
GaugeMetricProducer gaugeProducer(kConfigKey, metric, -1 /*-1 meaning no condition*/, wizard,
|
||||
tagId, bucketStartTimeNs, bucketStartTimeNs, pullerManager);
|
||||
gaugeProducer.setBucketSize(60 * NS_PER_SEC);
|
||||
tagId, -1, tagId, bucketStartTimeNs, bucketStartTimeNs,
|
||||
pullerManager);
|
||||
|
||||
vector<shared_ptr<LogEvent>> allData;
|
||||
allData.clear();
|
||||
@@ -176,10 +177,9 @@ TEST(GaugeMetricProducerTest, TestPushedEventsWithUpgrade) {
|
||||
sp<MockConditionWizard> wizard = new NaggyMock<MockConditionWizard>();
|
||||
sp<MockStatsPullerManager> pullerManager = new StrictMock<MockStatsPullerManager>();
|
||||
GaugeMetricProducer gaugeProducer(kConfigKey, metric, -1 /*-1 meaning no condition*/, wizard,
|
||||
-1 /* -1 means no pulling */, bucketStartTimeNs,
|
||||
-1 /* -1 means no pulling */, -1, tagId, bucketStartTimeNs,
|
||||
bucketStartTimeNs, pullerManager);
|
||||
|
||||
gaugeProducer.setBucketSize(60 * NS_PER_SEC);
|
||||
sp<AnomalyTracker> anomalyTracker = gaugeProducer.addAnomalyTracker(alert, alarmMonitor);
|
||||
EXPECT_TRUE(anomalyTracker != nullptr);
|
||||
|
||||
@@ -261,8 +261,8 @@ TEST(GaugeMetricProducerTest, TestPulledWithUpgrade) {
|
||||
}));
|
||||
|
||||
GaugeMetricProducer gaugeProducer(kConfigKey, metric, -1 /*-1 meaning no condition*/, wizard,
|
||||
tagId, bucketStartTimeNs, bucketStartTimeNs, pullerManager);
|
||||
gaugeProducer.setBucketSize(60 * NS_PER_SEC);
|
||||
tagId, -1, tagId, bucketStartTimeNs, bucketStartTimeNs,
|
||||
pullerManager);
|
||||
|
||||
vector<shared_ptr<LogEvent>> allData;
|
||||
shared_ptr<LogEvent> event = make_shared<LogEvent>(tagId, bucketStartTimeNs + 1);
|
||||
@@ -328,9 +328,8 @@ TEST(GaugeMetricProducerTest, TestPulledEventsWithCondition) {
|
||||
return true;
|
||||
}));
|
||||
|
||||
GaugeMetricProducer gaugeProducer(kConfigKey, metric, 1, wizard, tagId, bucketStartTimeNs,
|
||||
bucketStartTimeNs, pullerManager);
|
||||
gaugeProducer.setBucketSize(60 * NS_PER_SEC);
|
||||
GaugeMetricProducer gaugeProducer(kConfigKey, metric, 1, wizard, tagId, -1, tagId,
|
||||
bucketStartTimeNs, bucketStartTimeNs, pullerManager);
|
||||
|
||||
gaugeProducer.onConditionChanged(true, bucketStartTimeNs + 8);
|
||||
EXPECT_EQ(1UL, gaugeProducer.mCurrentSlicedBucket->size());
|
||||
@@ -419,9 +418,8 @@ TEST(GaugeMetricProducerTest, TestPulledEventsWithSlicedCondition) {
|
||||
return true;
|
||||
}));
|
||||
|
||||
GaugeMetricProducer gaugeProducer(kConfigKey, metric, 1, wizard, tagId, bucketStartTimeNs,
|
||||
bucketStartTimeNs, pullerManager);
|
||||
gaugeProducer.setBucketSize(60 * NS_PER_SEC);
|
||||
GaugeMetricProducer gaugeProducer(kConfigKey, metric, 1, wizard, tagId, -1, tagId,
|
||||
bucketStartTimeNs, bucketStartTimeNs, pullerManager);
|
||||
|
||||
gaugeProducer.onSlicedConditionMayChange(true, bucketStartTimeNs + 8);
|
||||
|
||||
@@ -464,8 +462,8 @@ TEST(GaugeMetricProducerTest, TestPulledEventsAnomalyDetection) {
|
||||
gaugeFieldMatcher->set_field(tagId);
|
||||
gaugeFieldMatcher->add_child()->set_field(2);
|
||||
GaugeMetricProducer gaugeProducer(kConfigKey, metric, -1 /*-1 meaning no condition*/, wizard,
|
||||
tagId, bucketStartTimeNs, bucketStartTimeNs, pullerManager);
|
||||
gaugeProducer.setBucketSize(60 * NS_PER_SEC);
|
||||
tagId, -1, tagId, bucketStartTimeNs, bucketStartTimeNs,
|
||||
pullerManager);
|
||||
|
||||
Alert alert;
|
||||
alert.set_id(101);
|
||||
@@ -530,6 +528,83 @@ TEST(GaugeMetricProducerTest, TestPulledEventsAnomalyDetection) {
|
||||
EXPECT_TRUE(gaugeProducer.mCurrentSlicedBucket->begin()->second.front().mFields->empty());
|
||||
}
|
||||
|
||||
TEST(GaugeMetricProducerTest, TestPullOnTrigger) {
|
||||
GaugeMetric metric;
|
||||
metric.set_id(metricId);
|
||||
metric.set_bucket(ONE_MINUTE);
|
||||
metric.set_sampling_type(GaugeMetric::ALL_CONDITION_CHANGES);
|
||||
metric.mutable_gauge_fields_filter()->set_include_all(false);
|
||||
auto gaugeFieldMatcher = metric.mutable_gauge_fields_filter()->mutable_fields();
|
||||
gaugeFieldMatcher->set_field(tagId);
|
||||
gaugeFieldMatcher->add_child()->set_field(1);
|
||||
|
||||
sp<MockConditionWizard> wizard = new NaggyMock<MockConditionWizard>();
|
||||
|
||||
sp<MockStatsPullerManager> pullerManager = new StrictMock<MockStatsPullerManager>();
|
||||
EXPECT_CALL(*pullerManager, Pull(tagId, _, _))
|
||||
.WillOnce(Invoke([](int tagId, int64_t timeNs,
|
||||
vector<std::shared_ptr<LogEvent>>* data) {
|
||||
data->clear();
|
||||
shared_ptr<LogEvent> event = make_shared<LogEvent>(tagId, bucketStartTimeNs + 3);
|
||||
event->write(3);
|
||||
event->init();
|
||||
data->push_back(event);
|
||||
return true;
|
||||
}))
|
||||
.WillOnce(Invoke([](int tagId, int64_t timeNs,
|
||||
vector<std::shared_ptr<LogEvent>>* data) {
|
||||
data->clear();
|
||||
shared_ptr<LogEvent> event = make_shared<LogEvent>(tagId, bucketStartTimeNs + 10);
|
||||
event->write(4);
|
||||
event->init();
|
||||
data->push_back(event);
|
||||
return true;
|
||||
}))
|
||||
.WillOnce(Invoke([](int tagId, int64_t timeNs,
|
||||
vector<std::shared_ptr<LogEvent>>* data) {
|
||||
data->clear();
|
||||
shared_ptr<LogEvent> event = make_shared<LogEvent>(tagId, bucketStartTimeNs + 20);
|
||||
event->write(5);
|
||||
event->init();
|
||||
data->push_back(event);
|
||||
return true;
|
||||
}));
|
||||
|
||||
int triggerId = 5;
|
||||
GaugeMetricProducer gaugeProducer(kConfigKey, metric, -1 /*-1 meaning no condition*/, wizard,
|
||||
tagId, triggerId, tagId, bucketStartTimeNs, bucketStartTimeNs,
|
||||
pullerManager);
|
||||
|
||||
vector<shared_ptr<LogEvent>> allData;
|
||||
allData.clear();
|
||||
|
||||
EXPECT_EQ(1UL, gaugeProducer.mCurrentSlicedBucket->size());
|
||||
LogEvent trigger(triggerId, bucketStartTimeNs + 10);
|
||||
trigger.init();
|
||||
gaugeProducer.onMatchedLogEvent(1 /*log matcher index*/, trigger);
|
||||
EXPECT_EQ(2UL, gaugeProducer.mCurrentSlicedBucket->begin()->second.size());
|
||||
trigger.setElapsedTimestampNs(bucketStartTimeNs + 20);
|
||||
gaugeProducer.onMatchedLogEvent(1 /*log matcher index*/, trigger);
|
||||
EXPECT_EQ(3UL, gaugeProducer.mCurrentSlicedBucket->begin()->second.size());
|
||||
|
||||
allData.clear();
|
||||
shared_ptr<LogEvent> event = make_shared<LogEvent>(tagId, bucket2StartTimeNs + 1);
|
||||
event->write(10);
|
||||
event->init();
|
||||
allData.push_back(event);
|
||||
|
||||
gaugeProducer.onDataPulled(allData);
|
||||
EXPECT_EQ(1UL, gaugeProducer.mCurrentSlicedBucket->size());
|
||||
auto it = gaugeProducer.mCurrentSlicedBucket->begin()->second.front().mFields->begin();
|
||||
EXPECT_EQ(INT, it->mValue.getType());
|
||||
EXPECT_EQ(10, it->mValue.int_value);
|
||||
EXPECT_EQ(1UL, gaugeProducer.mPastBuckets.size());
|
||||
EXPECT_EQ(3UL, gaugeProducer.mPastBuckets.begin()->second.back().mGaugeAtoms.size());
|
||||
EXPECT_EQ(3, gaugeProducer.mPastBuckets.begin()->second.back().mGaugeAtoms[0].mFields->begin()->mValue.int_value);
|
||||
EXPECT_EQ(4, gaugeProducer.mPastBuckets.begin()->second.back().mGaugeAtoms[1].mFields->begin()->mValue.int_value);
|
||||
EXPECT_EQ(5, gaugeProducer.mPastBuckets.begin()->second.back().mGaugeAtoms[2].mFields->begin()->mValue.int_value);
|
||||
}
|
||||
|
||||
} // namespace statsd
|
||||
} // namespace os
|
||||
} // namespace android
|
||||
|
||||
Reference in New Issue
Block a user