Merge "allow statsd pull based on event trigger"

7 files changed, 154 insertions, 29 deletions

diff --git a/cmds/statsd/src/config/ConfigManager.h b/cmds/statsd/src/config/ConfigManager.h
index 611c34250a38..122e669057b0 100644
--- a/cmds/statsd/src/config/ConfigManager.h
+++ b/cmds/statsd/src/config/ConfigManager.h
@@ -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.
  */
diff --git a/cmds/statsd/src/metrics/GaugeMetricProducer.cpp b/cmds/statsd/src/metrics/GaugeMetricProducer.cpp
index fbe0b2193556..284c4511a16f 100644
--- a/cmds/statsd/src/metrics/GaugeMetricProducer.cpp
+++ b/cmds/statsd/src/metrics/GaugeMetricProducer.cpp
@@ -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() &&
diff --git a/cmds/statsd/src/metrics/GaugeMetricProducer.h b/cmds/statsd/src/metrics/GaugeMetricProducer.h
index cc65440c9572..15be1d7c470b 100644
--- a/cmds/statsd/src/metrics/GaugeMetricProducer.h
+++ b/cmds/statsd/src/metrics/GaugeMetricProducer.h
@@ -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
diff --git a/cmds/statsd/src/metrics/ValueMetricProducer.cpp b/cmds/statsd/src/metrics/ValueMetricProducer.cpp
index 16447e850cf5..192a54b7e0a3 100644
--- a/cmds/statsd/src/metrics/ValueMetricProducer.cpp
+++ b/cmds/statsd/src/metrics/ValueMetricProducer.cpp
@@ -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"
diff --git a/cmds/statsd/src/metrics/metrics_manager_util.cpp b/cmds/statsd/src/metrics/metrics_manager_util.cpp
index ff48d0239059..75d6df95852d 100644
--- a/cmds/statsd/src/metrics/metrics_manager_util.cpp
+++ b/cmds/statsd/src/metrics/metrics_manager_util.cpp
@@ -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) {
diff --git a/cmds/statsd/src/statsd_config.proto b/cmds/statsd/src/statsd_config.proto
index 26dfda3985d1..d19e247ae6c7 100644
--- a/cmds/statsd/src/statsd_config.proto
+++ b/cmds/statsd/src/statsd_config.proto
@@ -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;
diff --git a/cmds/statsd/tests/metrics/GaugeMetricProducer_test.cpp b/cmds/statsd/tests/metrics/GaugeMetricProducer_test.cpp
index 2fda858db7d7..9471faa89547 100644
--- a/cmds/statsd/tests/metrics/GaugeMetricProducer_test.cpp
+++ b/cmds/statsd/tests/metrics/GaugeMetricProducer_test.cpp
@@ -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