Enable pull sampling for pulled gauge metrics

Test: atest statsd_test
Bug: 326980425
Change-Id: Ieef950da0d2968d08b01b8343d3178ec6afa98dd

9 files changed, 523 insertions, 6 deletions

diff --git a/statsd/src/external/StatsPullerManager.cpp b/statsd/src/external/StatsPullerManager.cpp
index 4fb63a66..df36d832 100644
--- a/statsd/src/external/StatsPullerManager.cpp
+++ b/statsd/src/external/StatsPullerManager.cpp
@@ -231,8 +231,8 @@ void StatsPullerManager::OnAlarmFired(int64_t elapsedTimeNs) {
                 // receiver to the list that will pull on this alarm.
                 // If pullNecessary is false, check if next pull time needs to be updated.
                 sp<PullDataReceiver> receiverPtr = receiverInfo.receiver.promote();
-                const bool pullNecessary = receiverPtr != nullptr && receiverPtr->isPullNeeded();
-                if (receiverInfo.nextPullTimeNs <= elapsedTimeNs && pullNecessary) {
+                if (receiverInfo.nextPullTimeNs <= elapsedTimeNs && receiverPtr != nullptr &&
+                    receiverPtr->isPullNeeded()) {
                     receivers.push_back(&receiverInfo);
                 } else {
                     if (receiverInfo.nextPullTimeNs <= elapsedTimeNs) {
diff --git a/statsd/src/guardrail/stats_log_enums.proto b/statsd/src/guardrail/stats_log_enums.proto
index d47bc80b..7b4c79a4 100644
--- a/statsd/src/guardrail/stats_log_enums.proto
+++ b/statsd/src/guardrail/stats_log_enums.proto
@@ -148,6 +148,9 @@ enum InvalidConfigReasonEnum {
     INVALID_CONFIG_REASON_MATCHER_INVALID_VALUE_MATCHER_WITH_STRING_REPLACE = 90;
     INVALID_CONFIG_REASON_MATCHER_COMBINATION_WITH_STRING_REPLACE = 91;
     INVALID_CONFIG_REASON_MATCHER_STRING_REPLACE_WITH_NO_VALUE_MATCHER_WITH_POSITION_ANY = 92;
+    INVALID_CONFIG_REASON_METRIC_INCORRECT_PULL_PROBABILITY = 93;
+    INVALID_CONFIG_REASON_GAUGE_METRIC_PUSHED_WITH_PULL_PROBABILITY = 94;
+    INVALID_CONFIG_REASON_GAUGE_METRIC_RANDOM_ONE_SAMPLE_WITH_PULL_PROBABILITY = 95;
 };
 
 enum InvalidQueryReason {
diff --git a/statsd/src/metrics/GaugeMetricProducer.cpp b/statsd/src/metrics/GaugeMetricProducer.cpp
index 51872916..974c9ae6 100644
--- a/statsd/src/metrics/GaugeMetricProducer.cpp
+++ b/statsd/src/metrics/GaugeMetricProducer.cpp
@@ -102,7 +102,8 @@ GaugeMetricProducer::GaugeMetricProducer(
       mDimensionHardLimit(dimensionHardLimit),
       mGaugeAtomsPerDimensionLimit(metric.max_num_gauge_atoms_per_bucket()),
       mDimensionGuardrailHit(false),
-      mSamplingPercentage(metric.sampling_percentage()) {
+      mSamplingPercentage(metric.sampling_percentage()),
+      mPullProbability(metric.pull_probability()) {
     mCurrentSlicedBucket = std::make_shared<DimToGaugeAtomsMap>();
     mCurrentSlicedBucketForAnomaly = std::make_shared<DimToValMap>();
     int64_t bucketSizeMills = 0;
@@ -397,7 +398,7 @@ void GaugeMetricProducer::pullAndMatchEventsLocked(const int64_t timestampNs) {
         default:
             break;
     }
-    if (!triggerPuller) {
+    if (!triggerPuller || !shouldKeepRandomSample(mPullProbability)) {
         return;
     }
     vector<std::shared_ptr<LogEvent>> allData;
diff --git a/statsd/src/metrics/GaugeMetricProducer.h b/statsd/src/metrics/GaugeMetricProducer.h
index 7c737a41..6563ab50 100644
--- a/statsd/src/metrics/GaugeMetricProducer.h
+++ b/statsd/src/metrics/GaugeMetricProducer.h
@@ -81,7 +81,8 @@ public:
     // Determine if metric needs to pull
     bool isPullNeeded() const override {
         std::lock_guard<std::mutex> lock(mMutex);
-        return mIsActive && (mCondition == ConditionState::kTrue);
+        return mIsActive && (mCondition == ConditionState::kTrue) &&
+               shouldKeepRandomSample(mPullProbability);
     };
 
     // GaugeMetric needs to immediately trigger another pull when we create the partial bucket.
@@ -227,6 +228,8 @@ private:
 
     const int mSamplingPercentage;
 
+    const int mPullProbability;
+
     FRIEND_TEST(GaugeMetricProducerTest, TestPulledEventsWithCondition);
     FRIEND_TEST(GaugeMetricProducerTest, TestPulledEventsWithSlicedCondition);
     FRIEND_TEST(GaugeMetricProducerTest, TestPulledEventsNoCondition);
diff --git a/statsd/src/metrics/MetricsManager.h b/statsd/src/metrics/MetricsManager.h
index 86cacf3a..2072d61b 100644
--- a/statsd/src/metrics/MetricsManager.h
+++ b/statsd/src/metrics/MetricsManager.h
@@ -368,6 +368,8 @@ private:
     FRIEND_TEST(MetricConditionLinkE2eTest, TestMultiplePredicatesAndLinks);
     FRIEND_TEST(AttributionE2eTest, TestAttributionMatchAndSliceByFirstUid);
     FRIEND_TEST(AttributionE2eTest, TestAttributionMatchAndSliceByChain);
+
+    FRIEND_TEST(GaugeMetricE2ePushedTest, TestDimensionalSampling);
     FRIEND_TEST(GaugeMetricE2ePulledTest, TestFirstNSamplesPulledNoTrigger);
     FRIEND_TEST(GaugeMetricE2ePulledTest, TestFirstNSamplesPulledNoTriggerWithActivation);
     FRIEND_TEST(GaugeMetricE2ePushedTest, TestMultipleFieldsForPushedEvent);
@@ -435,7 +437,6 @@ private:
     FRIEND_TEST(ValueMetricE2eTest, TestInitWithSlicedState);
     FRIEND_TEST(ValueMetricE2eTest, TestInitWithSlicedState_WithDimensions);
     FRIEND_TEST(ValueMetricE2eTest, TestInitWithSlicedState_WithIncorrectDimensions);
-    FRIEND_TEST(GaugeMetricE2ePushedTest, TestDimensionalSampling);
 };
 
 }  // namespace statsd
diff --git a/statsd/src/metrics/parsing_utils/metrics_manager_util.cpp b/statsd/src/metrics/parsing_utils/metrics_manager_util.cpp
index 673ff45a..f0c5bb25 100644
--- a/statsd/src/metrics/parsing_utils/metrics_manager_util.cpp
+++ b/statsd/src/metrics/parsing_utils/metrics_manager_util.cpp
@@ -1299,6 +1299,26 @@ optional<sp<MetricProducer>> createGaugeMetricProducerAndUpdateMetadata(
         return nullopt;
     }
 
+    if (metric.pull_probability() < 1 || metric.pull_probability() > 100) {
+        invalidConfigReason = InvalidConfigReason(
+                INVALID_CONFIG_REASON_METRIC_INCORRECT_PULL_PROBABILITY, metric.id());
+        return nullopt;
+    }
+
+    if (metric.pull_probability() != 100) {
+        if (pullTagId == -1) {
+            invalidConfigReason = InvalidConfigReason(
+                    INVALID_CONFIG_REASON_GAUGE_METRIC_PUSHED_WITH_PULL_PROBABILITY, metric.id());
+            return nullopt;
+        }
+        if (metric.sampling_type() == GaugeMetric::RANDOM_ONE_SAMPLE) {
+            invalidConfigReason = InvalidConfigReason(
+                    INVALID_CONFIG_REASON_GAUGE_METRIC_RANDOM_ONE_SAMPLE_WITH_PULL_PROBABILITY,
+                    metric.id());
+            return nullopt;
+        }
+    }
+
     unordered_map<int, shared_ptr<Activation>> eventActivationMap;
     unordered_map<int, vector<shared_ptr<Activation>>> eventDeactivationMap;
     invalidConfigReason = handleMetricActivation(
diff --git a/statsd/src/statsd_config.proto b/statsd/src/statsd_config.proto
index c4cffcac..b49d3e89 100644
--- a/statsd/src/statsd_config.proto
+++ b/statsd/src/statsd_config.proto
@@ -358,6 +358,8 @@ message GaugeMetric {
 
   optional int32 sampling_percentage = 17 [default = 100];
 
+  optional int32 pull_probability = 18 [default = 100];
+
   reserved 100;
   reserved 101;
 }
diff --git a/statsd/tests/e2e/GaugeMetric_e2e_pull_test.cpp b/statsd/tests/e2e/GaugeMetric_e2e_pull_test.cpp
index 4349a3c7..e9d5b0ec 100644
--- a/statsd/tests/e2e/GaugeMetric_e2e_pull_test.cpp
+++ b/statsd/tests/e2e/GaugeMetric_e2e_pull_test.cpp
@@ -956,6 +956,420 @@ TEST(GaugeMetricE2ePulledTest, TestRandomSamplePulledEventsNoCondition) {
     EXPECT_GT(data.bucket_info(2).atom(0).subsystem_sleep_state().time_millis(), 0);
 }
 
+TEST(GaugeMetricE2ePulledTest, TestGaugeMetricPullProbabilityWithTriggerEvent) {
+    // Initiating StatsdStats at the start of this test, so it doesn't call rand() during the test.
+    StatsdStats::getInstance();
+    // Set srand seed to make rand deterministic for testing.
+    srand(0);
+
+    auto config = CreateStatsdConfig(GaugeMetric::FIRST_N_SAMPLES, /*useCondition=*/false);
+    auto gaugeMetric = config.mutable_gauge_metric(0);
+    gaugeMetric->set_pull_probability(50);
+    auto triggerEventMatcher = CreateScreenTurnedOnAtomMatcher();
+    gaugeMetric->set_trigger_event(triggerEventMatcher.id());
+    gaugeMetric->set_max_num_gauge_atoms_per_bucket(200);
+    gaugeMetric->set_bucket(ONE_HOUR);
+
+    int64_t configAddedTimeNs = 60 * NS_PER_SEC;
+    int64_t bucketSizeNs = TimeUnitToBucketSizeInMillis(config.gauge_metric(0).bucket()) * 1000000;
+
+    ConfigKey cfgKey;
+    auto processor =
+            CreateStatsLogProcessor(configAddedTimeNs, configAddedTimeNs, config, cfgKey,
+                                    SharedRefBase::make<FakeSubsystemSleepCallback>(), ATOM_TAG);
+
+    std::vector<std::unique_ptr<LogEvent>> events;
+    // First bucket events.
+    for (int i = 0; i < 30; i++) {
+        events.push_back(CreateScreenStateChangedEvent(configAddedTimeNs + (i * 10 * NS_PER_SEC),
+                                                       android::view::DISPLAY_STATE_ON));
+    }
+    // Second bucket events.
+    for (int i = 0; i < 30; i++) {
+        events.push_back(CreateScreenStateChangedEvent(
+                configAddedTimeNs + bucketSizeNs + (i * 10 * NS_PER_SEC),
+                android::view::DISPLAY_STATE_ON));
+    }
+
+    // Send log events to StatsLogProcessor.
+    for (auto& event : events) {
+        processor->OnLogEvent(event.get());
+    }
+
+    ConfigMetricsReportList reports;
+    vector<uint8_t> buffer;
+    processor->onDumpReport(cfgKey, configAddedTimeNs + 7 * bucketSizeNs + 10, false, true,
+                            ADB_DUMP, FAST, &buffer);
+
+    EXPECT_TRUE(buffer.size() > 0);
+    EXPECT_TRUE(reports.ParseFromArray(&buffer[0], buffer.size()));
+    backfillDimensionPath(&reports);
+    backfillStringInReport(&reports);
+    backfillStartEndTimestamp(&reports);
+    backfillAggregatedAtoms(&reports);
+    ASSERT_EQ(1, reports.reports_size());
+    ASSERT_EQ(1, reports.reports(0).metrics_size());
+    StatsLogReport::GaugeMetricDataWrapper gaugeMetrics;
+    sortMetricDataByDimensionsValue(reports.reports(0).metrics(0).gauge_metrics(), &gaugeMetrics);
+    ASSERT_EQ((int)gaugeMetrics.data_size(), 2);  // 2 sets of data for each pull.
+
+    // Data 1
+    auto data = gaugeMetrics.data(0);
+    EXPECT_EQ(ATOM_TAG, data.dimensions_in_what().field());
+    ASSERT_EQ(1, data.dimensions_in_what().value_tuple().dimensions_value_size());
+    EXPECT_EQ(1 /* subsystem name field */,
+              data.dimensions_in_what().value_tuple().dimensions_value(0).field());
+    EXPECT_EQ("subsystem_name_1",
+              data.dimensions_in_what().value_tuple().dimensions_value(0).value_str());
+    ASSERT_EQ(2, data.bucket_info_size());
+
+    // Data 1, Bucket 1
+    ASSERT_EQ(13, data.bucket_info(0).atom_size());
+    ValidateGaugeBucketTimes(
+            data.bucket_info(0), configAddedTimeNs, configAddedTimeNs + bucketSizeNs,
+            {(int64_t)60 * NS_PER_SEC, (int64_t)80 * NS_PER_SEC, (int64_t)90 * NS_PER_SEC,
+             (int64_t)130 * NS_PER_SEC, (int64_t)150 * NS_PER_SEC, (int64_t)170 * NS_PER_SEC,
+             (int64_t)190 * NS_PER_SEC, (int64_t)200 * NS_PER_SEC, (int64_t)240 * NS_PER_SEC,
+             (int64_t)250 * NS_PER_SEC, (int64_t)300 * NS_PER_SEC, (int64_t)330 * NS_PER_SEC,
+             (int64_t)340 * NS_PER_SEC});
+
+    // Data 1, Bucket 2
+    ASSERT_EQ(18, data.bucket_info(1).atom_size());
+    ValidateGaugeBucketTimes(
+            data.bucket_info(1), configAddedTimeNs + bucketSizeNs,
+            configAddedTimeNs + 2 * bucketSizeNs,
+            {(int64_t)3660 * NS_PER_SEC, (int64_t)3680 * NS_PER_SEC, (int64_t)3700 * NS_PER_SEC,
+             (int64_t)3710 * NS_PER_SEC, (int64_t)3720 * NS_PER_SEC, (int64_t)3740 * NS_PER_SEC,
+             (int64_t)3780 * NS_PER_SEC, (int64_t)3790 * NS_PER_SEC, (int64_t)3820 * NS_PER_SEC,
+             (int64_t)3850 * NS_PER_SEC, (int64_t)3860 * NS_PER_SEC, (int64_t)3870 * NS_PER_SEC,
+             (int64_t)3880 * NS_PER_SEC, (int64_t)3900 * NS_PER_SEC, (int64_t)3910 * NS_PER_SEC,
+             (int64_t)3920 * NS_PER_SEC, (int64_t)3930 * NS_PER_SEC, (int64_t)3940 * NS_PER_SEC});
+
+    // Data 2
+    data = gaugeMetrics.data(1);
+    EXPECT_EQ(ATOM_TAG, data.dimensions_in_what().field());
+    ASSERT_EQ(1, data.dimensions_in_what().value_tuple().dimensions_value_size());
+    EXPECT_EQ(1 /* subsystem name field */,
+              data.dimensions_in_what().value_tuple().dimensions_value(0).field());
+    EXPECT_EQ("subsystem_name_2",
+              data.dimensions_in_what().value_tuple().dimensions_value(0).value_str());
+    ASSERT_EQ(2, data.bucket_info_size());
+
+    // Data 2, Bucket 1
+    ASSERT_EQ(13, data.bucket_info(0).atom_size());
+    ValidateGaugeBucketTimes(
+            data.bucket_info(0), configAddedTimeNs, configAddedTimeNs + bucketSizeNs,
+            {(int64_t)60 * NS_PER_SEC, (int64_t)80 * NS_PER_SEC, (int64_t)90 * NS_PER_SEC,
+             (int64_t)130 * NS_PER_SEC, (int64_t)150 * NS_PER_SEC, (int64_t)170 * NS_PER_SEC,
+             (int64_t)190 * NS_PER_SEC, (int64_t)200 * NS_PER_SEC, (int64_t)240 * NS_PER_SEC,
+             (int64_t)250 * NS_PER_SEC, (int64_t)300 * NS_PER_SEC, (int64_t)330 * NS_PER_SEC,
+             (int64_t)340 * NS_PER_SEC});
+
+    // Data 2, Bucket 2
+    ASSERT_EQ(18, data.bucket_info(1).atom_size());
+    ValidateGaugeBucketTimes(
+            data.bucket_info(1), configAddedTimeNs + bucketSizeNs,
+            configAddedTimeNs + 2 * bucketSizeNs,
+            {(int64_t)3660 * NS_PER_SEC, (int64_t)3680 * NS_PER_SEC, (int64_t)3700 * NS_PER_SEC,
+             (int64_t)3710 * NS_PER_SEC, (int64_t)3720 * NS_PER_SEC, (int64_t)3740 * NS_PER_SEC,
+             (int64_t)3780 * NS_PER_SEC, (int64_t)3790 * NS_PER_SEC, (int64_t)3820 * NS_PER_SEC,
+             (int64_t)3850 * NS_PER_SEC, (int64_t)3860 * NS_PER_SEC, (int64_t)3870 * NS_PER_SEC,
+             (int64_t)3880 * NS_PER_SEC, (int64_t)3900 * NS_PER_SEC, (int64_t)3910 * NS_PER_SEC,
+             (int64_t)3920 * NS_PER_SEC, (int64_t)3930 * NS_PER_SEC, (int64_t)3940 * NS_PER_SEC});
+}
+
+TEST(GaugeMetricE2ePulledTest, TestGaugeMetricPullProbabilityWithBucketBoundaryAlarm) {
+    // Initiating StatsdStats at the start of this test, so it doesn't call rand() during the test.
+    StatsdStats::getInstance();
+    // Set srand seed to make rand deterministic for testing.
+    srand(0);
+
+    auto config = CreateStatsdConfig(GaugeMetric::FIRST_N_SAMPLES, /*useCondition=*/false);
+    auto gaugeMetric = config.mutable_gauge_metric(0);
+    gaugeMetric->set_pull_probability(50);
+    gaugeMetric->set_max_num_gauge_atoms_per_bucket(200);
+
+    int64_t baseTimeNs = 5 * 60 * NS_PER_SEC;
+    int64_t configAddedTimeNs = 10 * 60 * NS_PER_SEC;
+    int64_t bucketSizeNs = TimeUnitToBucketSizeInMillis(config.gauge_metric(0).bucket()) * 1000000;
+
+    ConfigKey cfgKey;
+    auto processor =
+            CreateStatsLogProcessor(configAddedTimeNs, configAddedTimeNs, config, cfgKey,
+                                    SharedRefBase::make<FakeSubsystemSleepCallback>(), ATOM_TAG);
+
+    // Pulling alarm arrives on time and resets the sequential pulling alarm.
+    for (int i = 1; i < 31; i++) {
+        processor->informPullAlarmFired(configAddedTimeNs + i * bucketSizeNs);
+    }
+
+    ConfigMetricsReportList reports;
+    vector<uint8_t> buffer;
+    processor->onDumpReport(cfgKey, configAddedTimeNs + 32 * bucketSizeNs + 10, false, true,
+                            ADB_DUMP, FAST, &buffer);
+    EXPECT_TRUE(buffer.size() > 0);
+    EXPECT_TRUE(reports.ParseFromArray(&buffer[0], buffer.size()));
+    backfillDimensionPath(&reports);
+    backfillStringInReport(&reports);
+    backfillStartEndTimestamp(&reports);
+    backfillAggregatedAtoms(&reports);
+    ASSERT_EQ(1, reports.reports_size());
+    ASSERT_EQ(1, reports.reports(0).metrics_size());
+    StatsLogReport::GaugeMetricDataWrapper gaugeMetrics;
+    sortMetricDataByDimensionsValue(reports.reports(0).metrics(0).gauge_metrics(), &gaugeMetrics);
+    ASSERT_EQ((int)gaugeMetrics.data_size(), 2);
+
+    // Data 1
+    auto data = gaugeMetrics.data(0);
+    EXPECT_EQ(ATOM_TAG, data.dimensions_in_what().field());
+    ASSERT_EQ(1, data.dimensions_in_what().value_tuple().dimensions_value_size());
+    EXPECT_EQ(1 /* subsystem name field */,
+              data.dimensions_in_what().value_tuple().dimensions_value(0).field());
+    EXPECT_EQ("subsystem_name_1",
+              data.dimensions_in_what().value_tuple().dimensions_value(0).value_str());
+    ASSERT_EQ(14, data.bucket_info_size());
+
+    EXPECT_EQ(1, data.bucket_info(0).atom_size());
+    ValidateGaugeBucketTimes(data.bucket_info(0), configAddedTimeNs,
+                             configAddedTimeNs + bucketSizeNs, {configAddedTimeNs});
+
+    EXPECT_EQ(1, data.bucket_info(1).atom_size());
+    ValidateGaugeBucketTimes(data.bucket_info(1), configAddedTimeNs + 2 * bucketSizeNs,
+                             configAddedTimeNs + 3 * bucketSizeNs,
+                             {configAddedTimeNs + 2 * bucketSizeNs});  // 1200000000000ns
+
+    EXPECT_EQ(1, data.bucket_info(2).atom_size());
+    ValidateGaugeBucketTimes(data.bucket_info(2), configAddedTimeNs + 3 * bucketSizeNs,
+                             configAddedTimeNs + 4 * bucketSizeNs,
+                             {(int64_t)configAddedTimeNs + 3 * bucketSizeNs});  // 1500000000000ns
+
+    EXPECT_EQ(1, data.bucket_info(3).atom_size());
+    ValidateGaugeBucketTimes(data.bucket_info(3), configAddedTimeNs + 7 * bucketSizeNs,
+                             configAddedTimeNs + 8 * bucketSizeNs,
+                             {configAddedTimeNs + 7 * bucketSizeNs});  // 2700000000000ns
+
+    EXPECT_EQ(1, data.bucket_info(4).atom_size());
+    ValidateGaugeBucketTimes(data.bucket_info(4), configAddedTimeNs + 9 * bucketSizeNs,
+                             configAddedTimeNs + 10 * bucketSizeNs,
+                             {configAddedTimeNs + 9 * bucketSizeNs});  // 3300000000000ns
+
+    EXPECT_EQ(1, data.bucket_info(5).atom_size());
+    ValidateGaugeBucketTimes(data.bucket_info(5), configAddedTimeNs + 11 * bucketSizeNs,
+                             configAddedTimeNs + 12 * bucketSizeNs,
+                             {configAddedTimeNs + 11 * bucketSizeNs});  // 3900000000000ns
+
+    EXPECT_EQ(1, data.bucket_info(6).atom_size());
+    ValidateGaugeBucketTimes(data.bucket_info(6), configAddedTimeNs + 13 * bucketSizeNs,
+                             configAddedTimeNs + 14 * bucketSizeNs,
+                             {configAddedTimeNs + 13 * bucketSizeNs});  // 4500000000000ns
+
+    EXPECT_EQ(1, data.bucket_info(7).atom_size());
+    ValidateGaugeBucketTimes(data.bucket_info(7), configAddedTimeNs + 14 * bucketSizeNs,
+                             configAddedTimeNs + 15 * bucketSizeNs,
+                             {configAddedTimeNs + 14 * bucketSizeNs});  // 4800000000000ns
+
+    EXPECT_EQ(1, data.bucket_info(8).atom_size());
+    ValidateGaugeBucketTimes(data.bucket_info(8), configAddedTimeNs + 18 * bucketSizeNs,
+                             configAddedTimeNs + 19 * bucketSizeNs,
+                             {configAddedTimeNs + 18 * bucketSizeNs});  // 6000000000000ns
+
+    EXPECT_EQ(1, data.bucket_info(9).atom_size());
+    ValidateGaugeBucketTimes(data.bucket_info(9), configAddedTimeNs + 19 * bucketSizeNs,
+                             configAddedTimeNs + 20 * bucketSizeNs,
+                             {configAddedTimeNs + 19 * bucketSizeNs});  // 6300000000000ns
+
+    EXPECT_EQ(1, data.bucket_info(10).atom_size());
+    ValidateGaugeBucketTimes(data.bucket_info(10), configAddedTimeNs + 24 * bucketSizeNs,
+                             configAddedTimeNs + 25 * bucketSizeNs,
+                             {configAddedTimeNs + 24 * bucketSizeNs});  // 7800000000000ns
+
+    EXPECT_EQ(1, data.bucket_info(11).atom_size());
+    ValidateGaugeBucketTimes(data.bucket_info(11), configAddedTimeNs + 27 * bucketSizeNs,
+                             configAddedTimeNs + 28 * bucketSizeNs,
+                             {configAddedTimeNs + 27 * bucketSizeNs});  // 8700000000000ns
+
+    EXPECT_EQ(1, data.bucket_info(12).atom_size());
+    ValidateGaugeBucketTimes(data.bucket_info(12), configAddedTimeNs + 28 * bucketSizeNs,
+                             configAddedTimeNs + 29 * bucketSizeNs,
+                             {configAddedTimeNs + 28 * bucketSizeNs});  // 9000000000000ns
+
+    EXPECT_EQ(1, data.bucket_info(13).atom_size());
+    ValidateGaugeBucketTimes(data.bucket_info(13), configAddedTimeNs + 30 * bucketSizeNs,
+                             configAddedTimeNs + 31 * bucketSizeNs,
+                             {configAddedTimeNs + 30 * bucketSizeNs});  // 9600000000000ns
+
+    // Data 2
+    data = gaugeMetrics.data(1);
+    EXPECT_EQ(ATOM_TAG, data.dimensions_in_what().field());
+    ASSERT_EQ(1, data.dimensions_in_what().value_tuple().dimensions_value_size());
+    EXPECT_EQ(1 /* subsystem name field */,
+              data.dimensions_in_what().value_tuple().dimensions_value(0).field());
+    EXPECT_EQ("subsystem_name_2",
+              data.dimensions_in_what().value_tuple().dimensions_value(0).value_str());
+    ASSERT_EQ(14, data.bucket_info_size());
+
+    EXPECT_EQ(1, data.bucket_info(0).atom_size());
+    ValidateGaugeBucketTimes(data.bucket_info(0), configAddedTimeNs,
+                             configAddedTimeNs + bucketSizeNs, {configAddedTimeNs});
+
+    EXPECT_EQ(1, data.bucket_info(1).atom_size());
+    ValidateGaugeBucketTimes(data.bucket_info(1), configAddedTimeNs + 2 * bucketSizeNs,
+                             configAddedTimeNs + 3 * bucketSizeNs,
+                             {configAddedTimeNs + 2 * bucketSizeNs});
+
+    EXPECT_EQ(1, data.bucket_info(2).atom_size());
+    ValidateGaugeBucketTimes(data.bucket_info(2), configAddedTimeNs + 3 * bucketSizeNs,
+                             configAddedTimeNs + 4 * bucketSizeNs,
+                             {(int64_t)configAddedTimeNs + 3 * bucketSizeNs});
+
+    EXPECT_EQ(1, data.bucket_info(3).atom_size());
+    ValidateGaugeBucketTimes(data.bucket_info(3), configAddedTimeNs + 7 * bucketSizeNs,
+                             configAddedTimeNs + 8 * bucketSizeNs,
+                             {configAddedTimeNs + 7 * bucketSizeNs});
+
+    EXPECT_EQ(1, data.bucket_info(4).atom_size());
+    ValidateGaugeBucketTimes(data.bucket_info(4), configAddedTimeNs + 9 * bucketSizeNs,
+                             configAddedTimeNs + 10 * bucketSizeNs,
+                             {configAddedTimeNs + 9 * bucketSizeNs});
+
+    EXPECT_EQ(1, data.bucket_info(5).atom_size());
+    ValidateGaugeBucketTimes(data.bucket_info(5), configAddedTimeNs + 11 * bucketSizeNs,
+                             configAddedTimeNs + 12 * bucketSizeNs,
+                             {configAddedTimeNs + 11 * bucketSizeNs});
+
+    EXPECT_EQ(1, data.bucket_info(6).atom_size());
+    ValidateGaugeBucketTimes(data.bucket_info(6), configAddedTimeNs + 13 * bucketSizeNs,
+                             configAddedTimeNs + 14 * bucketSizeNs,
+                             {configAddedTimeNs + 13 * bucketSizeNs});
+
+    EXPECT_EQ(1, data.bucket_info(7).atom_size());
+    ValidateGaugeBucketTimes(data.bucket_info(7), configAddedTimeNs + 14 * bucketSizeNs,
+                             configAddedTimeNs + 15 * bucketSizeNs,
+                             {configAddedTimeNs + 14 * bucketSizeNs});
+
+    EXPECT_EQ(1, data.bucket_info(8).atom_size());
+    ValidateGaugeBucketTimes(data.bucket_info(8), configAddedTimeNs + 18 * bucketSizeNs,
+                             configAddedTimeNs + 19 * bucketSizeNs,
+                             {configAddedTimeNs + 18 * bucketSizeNs});
+
+    EXPECT_EQ(1, data.bucket_info(9).atom_size());
+    ValidateGaugeBucketTimes(data.bucket_info(9), configAddedTimeNs + 19 * bucketSizeNs,
+                             configAddedTimeNs + 20 * bucketSizeNs,
+                             {configAddedTimeNs + 19 * bucketSizeNs});
+
+    EXPECT_EQ(1, data.bucket_info(10).atom_size());
+    ValidateGaugeBucketTimes(data.bucket_info(10), configAddedTimeNs + 24 * bucketSizeNs,
+                             configAddedTimeNs + 25 * bucketSizeNs,
+                             {configAddedTimeNs + 24 * bucketSizeNs});
+
+    EXPECT_EQ(1, data.bucket_info(11).atom_size());
+    ValidateGaugeBucketTimes(data.bucket_info(11), configAddedTimeNs + 27 * bucketSizeNs,
+                             configAddedTimeNs + 28 * bucketSizeNs,
+                             {configAddedTimeNs + 27 * bucketSizeNs});
+
+    EXPECT_EQ(1, data.bucket_info(12).atom_size());
+    ValidateGaugeBucketTimes(data.bucket_info(12), configAddedTimeNs + 28 * bucketSizeNs,
+                             configAddedTimeNs + 29 * bucketSizeNs,
+                             {configAddedTimeNs + 28 * bucketSizeNs});
+
+    EXPECT_EQ(1, data.bucket_info(13).atom_size());
+    ValidateGaugeBucketTimes(data.bucket_info(13), configAddedTimeNs + 30 * bucketSizeNs,
+                             configAddedTimeNs + 31 * bucketSizeNs,
+                             {configAddedTimeNs + 30 * bucketSizeNs});
+}
+
+TEST(GaugeMetricE2ePulledTest, TestGaugeMetricPullProbabilityWithCondition) {
+    // Initiating StatsdStats at the start of this test, so it doesn't call rand() during the test.
+    StatsdStats::getInstance();
+    // Set srand seed to make rand deterministic for testing.
+    srand(0);
+
+    auto config = CreateStatsdConfig(GaugeMetric::CONDITION_CHANGE_TO_TRUE, /*useCondition=*/true);
+    auto gaugeMetric = config.mutable_gauge_metric(0);
+    gaugeMetric->set_pull_probability(50);
+    gaugeMetric->set_max_num_gauge_atoms_per_bucket(200);
+    gaugeMetric->set_bucket(ONE_HOUR);
+
+    int64_t configAddedTimeNs = 60 * NS_PER_SEC;
+    int64_t bucketSizeNs = TimeUnitToBucketSizeInMillis(config.gauge_metric(0).bucket()) * 1000000;
+
+    ConfigKey cfgKey;
+    auto processor =
+            CreateStatsLogProcessor(configAddedTimeNs, configAddedTimeNs, config, cfgKey,
+                                    SharedRefBase::make<FakeSubsystemSleepCallback>(), ATOM_TAG);
+
+    std::vector<std::unique_ptr<LogEvent>> events;
+    // First bucket events.
+    for (int i = 0; i < 30; i++) {
+        events.push_back(CreateScreenStateChangedEvent(configAddedTimeNs + (i * 10 * NS_PER_SEC),
+                                                       android::view::DISPLAY_STATE_OFF));
+        events.push_back(CreateScreenStateChangedEvent(configAddedTimeNs + (i * 11 * NS_PER_SEC),
+                                                       android::view::DISPLAY_STATE_ON));
+    }
+
+    // Send log events to StatsLogProcessor.
+    for (auto& event : events) {
+        processor->OnLogEvent(event.get());
+    }
+
+    ConfigMetricsReportList reports;
+    vector<uint8_t> buffer;
+    processor->onDumpReport(cfgKey, configAddedTimeNs + 2 * bucketSizeNs, false, true, ADB_DUMP,
+                            FAST, &buffer);
+
+    EXPECT_TRUE(buffer.size() > 0);
+    EXPECT_TRUE(reports.ParseFromArray(&buffer[0], buffer.size()));
+    backfillDimensionPath(&reports);
+    backfillStringInReport(&reports);
+    backfillStartEndTimestamp(&reports);
+    backfillAggregatedAtoms(&reports);
+    ASSERT_EQ(1, reports.reports_size());
+    ASSERT_EQ(1, reports.reports(0).metrics_size());
+    StatsLogReport::GaugeMetricDataWrapper gaugeMetrics;
+    sortMetricDataByDimensionsValue(reports.reports(0).metrics(0).gauge_metrics(), &gaugeMetrics);
+    ASSERT_EQ((int)gaugeMetrics.data_size(), 2);  // 2 sets of data for each pull.
+
+    // Data 1
+    auto data = gaugeMetrics.data(0);
+    EXPECT_EQ(ATOM_TAG, data.dimensions_in_what().field());
+    ASSERT_EQ(1, data.dimensions_in_what().value_tuple().dimensions_value_size());
+    EXPECT_EQ(1 /* subsystem name field */,
+              data.dimensions_in_what().value_tuple().dimensions_value(0).field());
+    EXPECT_EQ("subsystem_name_1",
+              data.dimensions_in_what().value_tuple().dimensions_value(0).value_str());
+    ASSERT_EQ(1, data.bucket_info_size());
+
+    // Data 1, Bucket 1
+    ASSERT_EQ(13, data.bucket_info(0).atom_size());
+    ValidateGaugeBucketTimes(
+            data.bucket_info(0), configAddedTimeNs, configAddedTimeNs + bucketSizeNs,
+            {(int64_t)60 * NS_PER_SEC, (int64_t)80 * NS_PER_SEC, (int64_t)90 * NS_PER_SEC,
+             (int64_t)130 * NS_PER_SEC, (int64_t)150 * NS_PER_SEC, (int64_t)170 * NS_PER_SEC,
+             (int64_t)190 * NS_PER_SEC, (int64_t)200 * NS_PER_SEC, (int64_t)240 * NS_PER_SEC,
+             (int64_t)250 * NS_PER_SEC, (int64_t)300 * NS_PER_SEC, (int64_t)330 * NS_PER_SEC,
+             (int64_t)340 * NS_PER_SEC});
+
+    // Data 2
+    data = gaugeMetrics.data(1);
+    EXPECT_EQ(ATOM_TAG, data.dimensions_in_what().field());
+    ASSERT_EQ(1, data.dimensions_in_what().value_tuple().dimensions_value_size());
+    EXPECT_EQ(1 /* subsystem name field */,
+              data.dimensions_in_what().value_tuple().dimensions_value(0).field());
+    EXPECT_EQ("subsystem_name_2",
+              data.dimensions_in_what().value_tuple().dimensions_value(0).value_str());
+    ASSERT_EQ(1, data.bucket_info_size());
+
+    // Data 2, Bucket 1
+    ASSERT_EQ(13, data.bucket_info(0).atom_size());
+    ValidateGaugeBucketTimes(
+            data.bucket_info(0), configAddedTimeNs, configAddedTimeNs + bucketSizeNs,
+            {(int64_t)60 * NS_PER_SEC, (int64_t)80 * NS_PER_SEC, (int64_t)90 * NS_PER_SEC,
+             (int64_t)130 * NS_PER_SEC, (int64_t)150 * NS_PER_SEC, (int64_t)170 * NS_PER_SEC,
+             (int64_t)190 * NS_PER_SEC, (int64_t)200 * NS_PER_SEC, (int64_t)240 * NS_PER_SEC,
+             (int64_t)250 * NS_PER_SEC, (int64_t)300 * NS_PER_SEC, (int64_t)330 * NS_PER_SEC,
+             (int64_t)340 * NS_PER_SEC});
+}
+
 #else
 GTEST_LOG_(INFO) << "This test does nothing.\n";
 #endif
diff --git a/statsd/tests/metrics/parsing_utils/metrics_manager_util_test.cpp b/statsd/tests/metrics/parsing_utils/metrics_manager_util_test.cpp
index da678b4b..67aa8056 100644
--- a/statsd/tests/metrics/parsing_utils/metrics_manager_util_test.cpp
+++ b/statsd/tests/metrics/parsing_utils/metrics_manager_util_test.cpp
@@ -707,6 +707,79 @@ TEST_F(MetricsManagerUtilTest, TestPulledGaugeMetricWithSamplingPercentage) {
                                   StringToId("Gauge")));
 }
 
+TEST_F(MetricsManagerUtilTest, TestGaugeMetricInvalidPullProbability) {
+    StatsdConfig config;
+    GaugeMetric* metric = config.add_gauge_metric();
+    *metric = createGaugeMetric(/*name=*/"Gauge", /*what=*/StringToId("SubsystemSleep"),
+                                GaugeMetric::FIRST_N_SAMPLES,
+                                /*condition=*/nullopt, /*triggerEvent=*/nullopt);
+    metric->set_pull_probability(101);
+    *config.add_atom_matcher() =
+            CreateSimpleAtomMatcher("SubsystemSleep", util::SUBSYSTEM_SLEEP_STATE);
+
+    EXPECT_EQ(initConfig(config),
+              InvalidConfigReason(INVALID_CONFIG_REASON_METRIC_INCORRECT_PULL_PROBABILITY,
+                                  StringToId("Gauge")));
+}
+
+TEST_F(MetricsManagerUtilTest, TestGaugeMetricInvalidPullProbabilityZero) {
+    StatsdConfig config;
+    GaugeMetric* metric = config.add_gauge_metric();
+    *metric = createGaugeMetric(/*name=*/"Gauge", /*what=*/StringToId("SubsystemSleep"),
+                                GaugeMetric::FIRST_N_SAMPLES,
+                                /*condition=*/nullopt, /*triggerEvent=*/nullopt);
+    metric->set_pull_probability(0);
+    *config.add_atom_matcher() =
+            CreateSimpleAtomMatcher("SubsystemSleep", util::SUBSYSTEM_SLEEP_STATE);
+
+    EXPECT_EQ(initConfig(config),
+              InvalidConfigReason(INVALID_CONFIG_REASON_METRIC_INCORRECT_PULL_PROBABILITY,
+                                  StringToId("Gauge")));
+}
+
+TEST_F(MetricsManagerUtilTest, TestGaugeMetricValidPullProbability) {
+    StatsdConfig config;
+    GaugeMetric* metric = config.add_gauge_metric();
+    *metric = createGaugeMetric(/*name=*/"Gauge", /*what=*/StringToId("SubsystemSleep"),
+                                GaugeMetric::FIRST_N_SAMPLES,
+                                /*condition=*/nullopt, /*triggerEvent=*/nullopt);
+    metric->set_pull_probability(50);
+    *config.add_atom_matcher() =
+            CreateSimpleAtomMatcher("SubsystemSleep", util::SUBSYSTEM_SLEEP_STATE);
+
+    EXPECT_EQ(initConfig(config), nullopt);
+}
+
+TEST_F(MetricsManagerUtilTest, TestPushedGaugeMetricWithPullProbability) {
+    StatsdConfig config;
+    GaugeMetric* metric = config.add_gauge_metric();
+    *metric = createGaugeMetric(/*name=*/"Gauge", /*what=*/StringToId("ScreenTurnedOn"),
+                                GaugeMetric::FIRST_N_SAMPLES,
+                                /*condition=*/nullopt, /*triggerEvent=*/nullopt);
+    metric->set_pull_probability(50);
+    *config.add_atom_matcher() = CreateScreenTurnedOnAtomMatcher();
+
+    EXPECT_EQ(initConfig(config),
+              InvalidConfigReason(INVALID_CONFIG_REASON_GAUGE_METRIC_PUSHED_WITH_PULL_PROBABILITY,
+                                  StringToId("Gauge")));
+}
+
+TEST_F(MetricsManagerUtilTest, TestGaugeMetricRandomOneSampleWithPullProbability) {
+    StatsdConfig config;
+    GaugeMetric* metric = config.add_gauge_metric();
+    *metric = createGaugeMetric(/*name=*/"Gauge", /*what=*/StringToId("SubsystemSleep"),
+                                GaugeMetric::RANDOM_ONE_SAMPLE,
+                                /*condition=*/nullopt, /*triggerEvent=*/nullopt);
+    metric->set_pull_probability(50);
+    *config.add_atom_matcher() =
+            CreateSimpleAtomMatcher("SubsystemSleep", util::SUBSYSTEM_SLEEP_STATE);
+
+    EXPECT_EQ(initConfig(config),
+              InvalidConfigReason(
+                      INVALID_CONFIG_REASON_GAUGE_METRIC_RANDOM_ONE_SAMPLE_WITH_PULL_PROBABILITY,
+                      StringToId("Gauge")));
+}
+
 TEST_F(MetricsManagerUtilTest, TestNumericValueMetricMissingIdOrWhat) {
     StatsdConfig config;
     int64_t metricId = 1;