Unit tests for ValueMetricProducer

StatsPullerManager is refactored so that we can mock it.
It may need more refactor pass to make is safer for longer runs.

Test: unit test
Change-Id: Ief0c99710e4d06e1454678f8b749c9599467d114

12 files changed, 534 insertions, 106 deletions

diff --git a/cmds/statsd/Android.mk b/cmds/statsd/Android.mk
index a1f5bb1a30f0..0f6d868708bd 100644
--- a/cmds/statsd/Android.mk
+++ b/cmds/statsd/Android.mk
@@ -33,7 +33,7 @@ statsd_common_src := \
     src/external/ResourcePowerManagerPuller.cpp \
     src/external/CpuTimePerUidPuller.cpp \
     src/external/CpuTimePerUidFreqPuller.cpp \
-    src/external/StatsPullerManager.cpp \
+    src/external/StatsPullerManagerImpl.cpp \
     src/logd/LogEvent.cpp \
     src/logd/LogListener.cpp \
     src/logd/LogReader.cpp \
@@ -164,7 +164,8 @@ LOCAL_SRC_FILES := \
     tests/metrics/OringDurationTracker_test.cpp \
     tests/metrics/MaxDurationTracker_test.cpp \
     tests/metrics/CountMetricProducer_test.cpp \
-    tests/metrics/EventMetricProducer_test.cpp
+    tests/metrics/EventMetricProducer_test.cpp \
+    tests/metrics/ValueMetricProducer_test.cpp
 
 LOCAL_STATIC_LIBRARIES := \
     libgmock
diff --git a/cmds/statsd/src/StatsService.h b/cmds/statsd/src/StatsService.h
index 1d7e5a618cc9..fa92f65d264d 100644
--- a/cmds/statsd/src/StatsService.h
+++ b/cmds/statsd/src/StatsService.h
@@ -157,7 +157,7 @@ private:
     /**
      * Fetches external metrics.
      */
-    StatsPullerManager& mStatsPullerManager = StatsPullerManager::GetInstance();
+    StatsPullerManager mStatsPullerManager;
 
     /**
      * Tracks the configurations that have been passed to statsd.
diff --git a/cmds/statsd/src/external/StatsPullerManager.h b/cmds/statsd/src/external/StatsPullerManager.h
index 67580d68c6f1..2e803c9bd2d6 100644
--- a/cmds/statsd/src/external/StatsPullerManager.h
+++ b/cmds/statsd/src/external/StatsPullerManager.h
@@ -16,71 +16,39 @@
 
 #pragma once
 
-#include <android/os/IStatsCompanionService.h>
-#include <binder/IServiceManager.h>
-#include <utils/RefBase.h>
-#include <utils/String16.h>
-#include <utils/String8.h>
-#include <utils/threads.h>
-#include <string>
-#include <unordered_map>
-#include <vector>
-#include "PullDataReceiver.h"
-#include "StatsPuller.h"
-#include "logd/LogEvent.h"
+#include "StatsPullerManagerImpl.h"
 
 namespace android {
 namespace os {
 namespace statsd {
 
-class StatsPullerManager : public virtual RefBase {
-public:
-    static StatsPullerManager& GetInstance();
+class StatsPullerManager{
+ public:
+  virtual ~StatsPullerManager() {}
 
-    void RegisterReceiver(int tagId, sp<PullDataReceiver> receiver, long intervalMs);
+  virtual void RegisterReceiver(int tagId, wp<PullDataReceiver> receiver, long intervalMs) {
+    mPullerManager.RegisterReceiver(tagId, receiver, intervalMs);
+  };
 
-    void UnRegisterReceiver(int tagId, sp<PullDataReceiver> receiver);
+  virtual void UnRegisterReceiver(int tagId, wp<PullDataReceiver> receiver) {
+    mPullerManager.UnRegisterReceiver(tagId, receiver);
+  };
 
-    // Verify if we know how to pull for this matcher
-    bool PullerForMatcherExists(int tagId);
+  // Verify if we know how to pull for this matcher
+  bool PullerForMatcherExists(int tagId) {
+    return mPullerManager.PullerForMatcherExists(tagId);
+  }
 
-    void OnAlarmFired();
+  void OnAlarmFired() {
+    mPullerManager.OnAlarmFired();
+  }
 
-    bool Pull(const int pullCode, vector<std::shared_ptr<LogEvent>>* data);
+  virtual bool Pull(const int tagId, vector<std::shared_ptr<LogEvent>>* data) {
+    return mPullerManager.Pull(tagId, data);
+  }
 
-private:
-    StatsPullerManager();
-
-    // use this to update alarm
-    sp<IStatsCompanionService> mStatsCompanionService = nullptr;
-
-    sp<IStatsCompanionService> get_stats_companion_service();
-
-    // mapping from simple matcher tagId to puller
-    std::map<int, std::shared_ptr<StatsPuller>> mPullers;
-
-      typedef struct {
-        // pull_interval_sec : last_pull_time_sec
-        std::pair<uint64_t, uint64_t> timeInfo;
-        sp<PullDataReceiver> receiver;
-      } ReceiverInfo;
-
-    // mapping from simple matcher tagId to receivers
-    std::map<int, std::vector<ReceiverInfo>> mReceivers;
-
-    Mutex mReceiversLock;
-
-    long mCurrentPullingInterval;
-
-    // for pulled metrics, it is important for the buckets to be aligned to multiple of smallest
-    // bucket size. All pulled metrics start pulling based on this time, so that they can be
-    // correctly attributed to the correct buckets. Pulled data attach a timestamp which is the
-    // request time.
-    const long mPullStartTimeMs;
-
-    long get_pull_start_time_ms();
-
-    LogEvent parse_pulled_data(String16 data);
+ private:
+  StatsPullerManagerImpl& mPullerManager = StatsPullerManagerImpl::GetInstance();
 };
 
 }  // namespace statsd
diff --git a/cmds/statsd/src/external/StatsPullerManager.cpp b/cmds/statsd/src/external/StatsPullerManagerImpl.cpp
index 5a05b453aa86..07d0b3e2aac5 100644
--- a/cmds/statsd/src/external/StatsPullerManager.cpp
+++ b/cmds/statsd/src/external/StatsPullerManagerImpl.cpp
@@ -25,7 +25,7 @@
 #include "CpuTimePerUidPuller.h"
 #include "ResourcePowerManagerPuller.h"
 #include "StatsCompanionServicePuller.h"
-#include "StatsPullerManager.h"
+#include "StatsPullerManagerImpl.h"
 #include "StatsService.h"
 #include "logd/LogEvent.h"
 #include "statslog.h"
@@ -37,12 +37,13 @@ using std::map;
 using std::shared_ptr;
 using std::string;
 using std::vector;
+using std::list;
 
 namespace android {
 namespace os {
 namespace statsd {
 
-StatsPullerManager::StatsPullerManager()
+StatsPullerManagerImpl::StatsPullerManagerImpl()
     : mCurrentPullingInterval(LONG_MAX), mPullStartTimeMs(get_pull_start_time_ms()) {
     shared_ptr<StatsPuller> statsCompanionServicePuller = make_shared<StatsCompanionServicePuller>();
     shared_ptr<StatsPuller> resourcePowerManagerPuller = make_shared<ResourcePowerManagerPuller>();
@@ -71,7 +72,7 @@ StatsPullerManager::StatsPullerManager()
     mStatsCompanionService = StatsService::getStatsCompanionService();
 }
 
-bool StatsPullerManager::Pull(int tagId, vector<shared_ptr<LogEvent>>* data) {
+bool StatsPullerManagerImpl::Pull(int tagId, vector<shared_ptr<LogEvent>>* data) {
     if (DEBUG) ALOGD("Initiating pulling %d", tagId);
 
     if (mPullers.find(tagId) != mPullers.end()) {
@@ -82,26 +83,26 @@ bool StatsPullerManager::Pull(int tagId, vector<shared_ptr<LogEvent>>* data) {
     }
 }
 
-StatsPullerManager& StatsPullerManager::GetInstance() {
-    static StatsPullerManager instance;
+StatsPullerManagerImpl& StatsPullerManagerImpl::GetInstance() {
+    static StatsPullerManagerImpl instance;
     return instance;
 }
 
-bool StatsPullerManager::PullerForMatcherExists(int tagId) {
+bool StatsPullerManagerImpl::PullerForMatcherExists(int tagId) {
     return mPullers.find(tagId) != mPullers.end();
 }
 
-long StatsPullerManager::get_pull_start_time_ms() {
+long StatsPullerManagerImpl::get_pull_start_time_ms() {
     // TODO: limit and align pull intervals to 10min boundaries if this turns out to be a problem
     return time(nullptr) * 1000;
 }
 
-void StatsPullerManager::RegisterReceiver(int tagId, sp<PullDataReceiver> receiver,
-                                          long intervalMs) {
+void StatsPullerManagerImpl::RegisterReceiver(int tagId, wp<PullDataReceiver> receiver,
+                                              long intervalMs) {
     AutoMutex _l(mReceiversLock);
-    vector<ReceiverInfo>& receivers = mReceivers[tagId];
+    auto& receivers = mReceivers[tagId];
     for (auto it = receivers.begin(); it != receivers.end(); it++) {
-        if (it->receiver.get() == receiver.get()) {
+        if (it->receiver == receiver) {
             VLOG("Receiver already registered of %d", (int)receivers.size());
             return;
         }
@@ -124,7 +125,7 @@ void StatsPullerManager::RegisterReceiver(int tagId, sp<PullDataReceiver> receiv
     VLOG("Puller for tagId %d registered of %d", tagId, (int)receivers.size());
 }
 
-void StatsPullerManager::UnRegisterReceiver(int tagId, sp<PullDataReceiver> receiver) {
+void StatsPullerManagerImpl::UnRegisterReceiver(int tagId, wp<PullDataReceiver> receiver) {
     AutoMutex _l(mReceiversLock);
     if (mReceivers.find(tagId) == mReceivers.end()) {
         VLOG("Unknown pull code or no receivers: %d", tagId);
@@ -132,7 +133,7 @@ void StatsPullerManager::UnRegisterReceiver(int tagId, sp<PullDataReceiver> rece
     }
     auto& receivers = mReceivers.find(tagId)->second;
     for (auto it = receivers.begin(); it != receivers.end(); it++) {
-        if (receiver.get() == it->receiver.get()) {
+        if (receiver == it->receiver) {
             receivers.erase(it);
             VLOG("Puller for tagId %d unregistered of %d", tagId, (int)receivers.size());
             return;
@@ -140,7 +141,7 @@ void StatsPullerManager::UnRegisterReceiver(int tagId, sp<PullDataReceiver> rece
     }
 }
 
-void StatsPullerManager::OnAlarmFired() {
+void StatsPullerManagerImpl::OnAlarmFired() {
     AutoMutex _l(mReceiversLock);
 
     uint64_t currentTimeMs = time(nullptr) * 1000;
@@ -165,8 +166,13 @@ void StatsPullerManager::OnAlarmFired() {
         vector<shared_ptr<LogEvent>> data;
         if (Pull(pullInfo.first, &data)) {
             for (const auto& receiverInfo : pullInfo.second) {
-                receiverInfo->receiver->onDataPulled(data);
-                receiverInfo->timeInfo.second = currentTimeMs;
+                sp<PullDataReceiver> receiverPtr = receiverInfo->receiver.promote();
+                if (receiverPtr != nullptr) {
+                    receiverPtr->onDataPulled(data);
+                    receiverInfo->timeInfo.second = currentTimeMs;
+                } else {
+                    VLOG("receiver already gone.");
+                }
             }
         }
     }
diff --git a/cmds/statsd/src/external/StatsPullerManagerImpl.h b/cmds/statsd/src/external/StatsPullerManagerImpl.h
new file mode 100644
index 000000000000..0b9f21e3217c
--- /dev/null
+++ b/cmds/statsd/src/external/StatsPullerManagerImpl.h
@@ -0,0 +1,87 @@
+/*
+ * 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 <android/os/IStatsCompanionService.h>
+#include <binder/IServiceManager.h>
+#include <utils/RefBase.h>
+#include <utils/threads.h>
+#include <string>
+#include <unordered_map>
+#include <vector>
+#include <list>
+#include "PullDataReceiver.h"
+#include "StatsPuller.h"
+#include "logd/LogEvent.h"
+
+namespace android {
+namespace os {
+namespace statsd {
+
+class StatsPullerManagerImpl : public virtual RefBase {
+public:
+    static StatsPullerManagerImpl& GetInstance();
+
+    void RegisterReceiver(int tagId, wp<PullDataReceiver> receiver, long intervalMs);
+
+    void UnRegisterReceiver(int tagId, wp<PullDataReceiver> receiver);
+
+    // Verify if we know how to pull for this matcher
+    bool PullerForMatcherExists(int tagId);
+
+    void OnAlarmFired();
+
+    bool Pull(const int tagId, vector<std::shared_ptr<LogEvent>>* data);
+
+private:
+    StatsPullerManagerImpl();
+
+    // use this to update alarm
+    sp<IStatsCompanionService> mStatsCompanionService = nullptr;
+
+    sp<IStatsCompanionService> get_stats_companion_service();
+
+    // mapping from simple matcher tagId to puller
+    std::map<int, std::shared_ptr<StatsPuller>> mPullers;
+
+    typedef struct {
+        // pull_interval_sec : last_pull_time_sec
+        std::pair<uint64_t, uint64_t> timeInfo;
+        wp<PullDataReceiver> receiver;
+    } ReceiverInfo;
+
+    // mapping from simple matcher tagId to receivers
+    std::map<int, std::list<ReceiverInfo>> mReceivers;
+
+    Mutex mReceiversLock;
+
+    long mCurrentPullingInterval;
+
+    // for pulled metrics, it is important for the buckets to be aligned to multiple of smallest
+    // bucket size. All pulled metrics start pulling based on this time, so that they can be
+    // correctly attributed to the correct buckets. Pulled data attach a timestamp which is the
+    // request time.
+    const long mPullStartTimeMs;
+
+    long get_pull_start_time_ms();
+
+    LogEvent parse_pulled_data(String16 data);
+};
+
+}  // namespace statsd
+}  // namespace os
+}  // namespace android
diff --git a/cmds/statsd/src/metrics/GaugeMetricProducer.h b/cmds/statsd/src/metrics/GaugeMetricProducer.h
index d80672d25957..f9e4deb9f04e 100644
--- a/cmds/statsd/src/metrics/GaugeMetricProducer.h
+++ b/cmds/statsd/src/metrics/GaugeMetricProducer.h
@@ -81,7 +81,7 @@ private:
     static const uint64_t kDefaultGaugemBucketSizeNs = 1000 * 1000 * 1000;
     const GaugeMetric mMetric;
 
-    StatsPullerManager& mStatsPullerManager = StatsPullerManager::GetInstance();
+    StatsPullerManager mStatsPullerManager;
     // tagId for pulled data. -1 if this is not pulled
     const int mPullTagId;
 
diff --git a/cmds/statsd/src/metrics/ValueMetricProducer.cpp b/cmds/statsd/src/metrics/ValueMetricProducer.cpp
index 5bd10fa3dad8..5cffec1b1bcf 100644
--- a/cmds/statsd/src/metrics/ValueMetricProducer.cpp
+++ b/cmds/statsd/src/metrics/ValueMetricProducer.cpp
@@ -31,6 +31,7 @@ using android::util::FIELD_TYPE_INT64;
 using android::util::FIELD_TYPE_MESSAGE;
 using android::util::ProtoOutputStream;
 using std::list;
+using std::make_pair;
 using std::make_shared;
 using std::map;
 using std::shared_ptr;
@@ -62,13 +63,23 @@ const int FIELD_ID_START_BUCKET_NANOS = 1;
 const int FIELD_ID_END_BUCKET_NANOS = 2;
 const int FIELD_ID_VALUE = 3;
 
+static const uint64_t kDefaultBucketSizeMillis = 60 * 60 * 1000L;
+
 // 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, const int pullTagId,
-                                         const uint64_t startTimeNs)
-    : MetricProducer(startTimeNs, conditionIndex, wizard), mMetric(metric), mPullTagId(pullTagId) {
+                                         const uint64_t startTimeNs,
+                                         shared_ptr<StatsPullerManager> statsPullerManager)
+    : MetricProducer(startTimeNs, conditionIndex, wizard),
+      mMetric(metric),
+      mStatsPullerManager(statsPullerManager),
+      mPullTagId(pullTagId) {
     // TODO: valuemetric for pushed events may need unlimited bucket length
-    mBucketSizeNs = mMetric.bucket().bucket_size_millis() * 1000 * 1000;
+    if (metric.has_bucket() && metric.bucket().has_bucket_size_millis()) {
+        mBucketSizeNs = mMetric.bucket().bucket_size_millis() * 1000 * 1000;
+    } else {
+        mBucketSizeNs = kDefaultBucketSizeMillis * 1000 * 1000;
+    }
 
     mDimension.insert(mDimension.begin(), metric.dimension().begin(), metric.dimension().end());
 
@@ -79,8 +90,9 @@ ValueMetricProducer::ValueMetricProducer(const ValueMetric& metric, const int co
     }
 
     if (!metric.has_condition() && mPullTagId != -1) {
-        mStatsPullerManager.RegisterReceiver(mPullTagId, this,
-                                             metric.bucket().bucket_size_millis());
+        VLOG("Setting up periodic pulling for %d", mPullTagId);
+        mStatsPullerManager->RegisterReceiver(mPullTagId, this,
+                                              metric.bucket().bucket_size_millis());
     }
 
     startNewProtoOutputStream(mStartTimeNs);
@@ -89,8 +101,19 @@ ValueMetricProducer::ValueMetricProducer(const ValueMetric& metric, const int co
          (long long)mBucketSizeNs, (long long)mStartTimeNs);
 }
 
+// for testing
+ValueMetricProducer::ValueMetricProducer(const ValueMetric& metric, const int conditionIndex,
+                                         const sp<ConditionWizard>& wizard, const int pullTagId,
+                                         const uint64_t startTimeNs)
+    : ValueMetricProducer(metric, conditionIndex, wizard, pullTagId, startTimeNs,
+                          make_shared<StatsPullerManager>()) {
+}
+
 ValueMetricProducer::~ValueMetricProducer() {
     VLOG("~ValueMetricProducer() called");
+    if (mPullTagId != -1) {
+        mStatsPullerManager->UnRegisterReceiver(mPullTagId, this);
+    }
 }
 
 void ValueMetricProducer::startNewProtoOutputStream(long long startTime) {
@@ -177,14 +200,14 @@ void ValueMetricProducer::onConditionChanged(const bool condition, const uint64_
 
     if (mPullTagId != -1) {
         if (mCondition == true) {
-            mStatsPullerManager.RegisterReceiver(mPullTagId, this,
-                                                 mMetric.bucket().bucket_size_millis());
-        } else if (mCondition == ConditionState::kFalse) {
-            mStatsPullerManager.UnRegisterReceiver(mPullTagId, this);
+            mStatsPullerManager->RegisterReceiver(mPullTagId, this,
+                                                  mMetric.bucket().bucket_size_millis());
+        } else if (mCondition == false) {
+            mStatsPullerManager->UnRegisterReceiver(mPullTagId, this);
         }
 
         vector<shared_ptr<LogEvent>> allData;
-        if (mStatsPullerManager.Pull(mPullTagId, &allData)) {
+        if (mStatsPullerManager->Pull(mPullTagId, &allData)) {
             if (allData.size() == 0) {
                 return;
             }
@@ -199,11 +222,15 @@ void ValueMetricProducer::onConditionChanged(const bool condition, const uint64_
 
 void ValueMetricProducer::onDataPulled(const std::vector<std::shared_ptr<LogEvent>>& allData) {
     AutoMutex _l(mLock);
-    if (mCondition == ConditionState::kTrue || !mMetric.has_condition()) {
+    if (mCondition == true || !mMetric.has_condition()) {
         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) {
+            flush_if_needed(eventTime);
+        }
         for (const auto& data : allData) {
             onMatchedLogEvent(0, *data, true);
         }
@@ -226,24 +253,36 @@ void ValueMetricProducer::onMatchedLogEventInternal(
 
     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) {
+    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;
+            }
+        } else {
+            if (mCondition == true) {
+                interval.raw.push_back(make_pair(value, 0));
+            } else {
+                if (interval.raw.size() != 0) {
+                    interval.raw.back().second = value;
+                } else {
+                    interval.tainted = true;
+                    VLOG("Data on condition true missing!");
+                }
             }
         }
-    }
-    if (mPullTagId == -1) {
+    } else {
         flush_if_needed(eventTimeNs);
+        interval.raw.push_back(make_pair(value, 0));
     }
 }
 
@@ -253,7 +292,7 @@ long ValueMetricProducer::get_value(const LogEvent& event) {
     if (err == NO_ERROR) {
         return val;
     } else {
-        VLOG("Can't find value in message.");
+        VLOG("Can't find value in message. %s", event.ToString().c_str());
         return 0;
     }
 }
@@ -271,13 +310,21 @@ void ValueMetricProducer::flush_if_needed(const uint64_t eventTimeNs) {
     info.mBucketStartNs = mCurrentBucketStartTimeNs;
     info.mBucketEndNs = mCurrentBucketStartTimeNs + mBucketSizeNs;
 
+    int tainted = 0;
     for (const auto& slice : mCurrentSlicedBucket) {
         long value = 0;
-        for (const auto& pair : slice.second.raw) {
-            value += pair.second - pair.first;
+        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", slice.first.c_str(), value);
+        VLOG(" %s, %ld, %d", slice.first.c_str(), value, tainted);
         // it will auto create new vector of ValuebucketInfo if the key is not found.
         auto& bucketList = mPastBuckets[slice.first];
         bucketList.push_back(info);
diff --git a/cmds/statsd/src/metrics/ValueMetricProducer.h b/cmds/statsd/src/metrics/ValueMetricProducer.h
index ef9868b613b2..c6c87f5b15f6 100644
--- a/cmds/statsd/src/metrics/ValueMetricProducer.h
+++ b/cmds/statsd/src/metrics/ValueMetricProducer.h
@@ -16,6 +16,7 @@
 
 #pragma once
 
+#include <gtest/gtest_prod.h>
 #include <utils/threads.h>
 #include <list>
 #include "../condition/ConditionTracker.h"
@@ -71,7 +72,13 @@ protected:
 private:
     const ValueMetric mMetric;
 
-    StatsPullerManager& mStatsPullerManager = StatsPullerManager::GetInstance();
+    std::shared_ptr<StatsPullerManager> mStatsPullerManager;
+
+    // for testing
+    ValueMetricProducer(const ValueMetric& valueMetric, const int conditionIndex,
+                        const sp<ConditionWizard>& wizard, const int pullTagId,
+                        const uint64_t startTimeNs,
+                        std::shared_ptr<StatsPullerManager> statsPullerManager);
 
     Mutex mLock;
 
@@ -81,6 +88,7 @@ private:
     // internal state of a bucket.
     typedef struct {
         std::vector<std::pair<long, long>> raw;
+        bool tainted;
     } Interval;
 
     std::unordered_map<HashableDimensionKey, Interval> mCurrentSlicedBucket;
@@ -97,6 +105,10 @@ private:
     void flush_if_needed(const uint64_t eventTimeNs);
 
     size_t mByteSize;
+
+    FRIEND_TEST(ValueMetricProducerTest, TestNonDimensionalEvents);
+    FRIEND_TEST(ValueMetricProducerTest, TestEventsWithNonSlicedCondition);
+    FRIEND_TEST(ValueMetricProducerTest, TestPushedEventsWithoutCondition);
 };
 
 }  // namespace statsd
diff --git a/cmds/statsd/src/metrics/metrics_manager_util.cpp b/cmds/statsd/src/metrics/metrics_manager_util.cpp
index ca9cdfb47f0a..226e4d13afb2 100644
--- a/cmds/statsd/src/metrics/metrics_manager_util.cpp
+++ b/cmds/statsd/src/metrics/metrics_manager_util.cpp
@@ -195,7 +195,7 @@ bool initMetrics(const StatsdConfig& config, const unordered_map<string, int>& l
     const int allMetricsCount = config.count_metric_size() + config.duration_metric_size() +
                                 config.event_metric_size() + config.value_metric_size();
     allMetricProducers.reserve(allMetricsCount);
-    StatsPullerManager& statsPullerManager = StatsPullerManager::GetInstance();
+    StatsPullerManager statsPullerManager;
     uint64_t startTimeNs = time(nullptr) * NS_PER_SEC;
 
     // Build MetricProducers for each metric defined in config.
diff --git a/cmds/statsd/src/metrics/metrics_manager_util.h b/cmds/statsd/src/metrics/metrics_manager_util.h
index e089d0653238..edf3af0aff8e 100644
--- a/cmds/statsd/src/metrics/metrics_manager_util.h
+++ b/cmds/statsd/src/metrics/metrics_manager_util.h
@@ -21,7 +21,7 @@
 #include <vector>
 
 #include "../condition/ConditionTracker.h"
-#include "../external/StatsPullerManager.h"
+#include "../external/StatsPullerManagerImpl.h"
 #include "../matchers/LogMatchingTracker.h"
 
 namespace android {
diff --git a/cmds/statsd/tests/metrics/ValueMetricProducer_test.cpp b/cmds/statsd/tests/metrics/ValueMetricProducer_test.cpp
new file mode 100644
index 000000000000..2a26388c7f73
--- /dev/null
+++ b/cmds/statsd/tests/metrics/ValueMetricProducer_test.cpp
@@ -0,0 +1,299 @@
+// 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.
+
+#include "metrics_test_helper.h"
+#include "src/metrics/ValueMetricProducer.h"
+
+#include <gmock/gmock.h>
+#include <gtest/gtest.h>
+#include <stdio.h>
+#include <vector>
+
+using namespace testing;
+using android::sp;
+using std::set;
+using std::unordered_map;
+using std::vector;
+using std::shared_ptr;
+using std::make_shared;
+
+#ifdef __ANDROID__
+
+namespace android {
+namespace os {
+namespace statsd {
+
+/*
+ * Tests pulled atoms with no conditions
+ */
+TEST(ValueMetricProducerTest, TestNonDimensionalEvents) {
+    int64_t bucketStartTimeNs = 10000000000;
+    int64_t bucketSizeNs = 60 * 1000 * 1000 * 1000LL;
+
+    int64_t bucket2StartTimeNs = bucketStartTimeNs + bucketSizeNs;
+    int64_t bucket3StartTimeNs = bucketStartTimeNs + 2*bucketSizeNs;
+
+    ValueMetric metric;
+    metric.set_metric_id(1);
+    metric.mutable_bucket()->set_bucket_size_millis(bucketSizeNs / 1000000);
+    metric.set_value_field(2);
+
+    int tagId = 1;
+
+    sp<MockConditionWizard> wizard = new NaggyMock<MockConditionWizard>();
+    // TODO: pending refactor of StatsPullerManager
+    // For now we still need this so that it doesn't do real pulling.
+    shared_ptr<MockStatsPullerManager> pullerManager = make_shared<StrictMock<MockStatsPullerManager>>();
+    EXPECT_CALL(*pullerManager, RegisterReceiver(tagId, _, _)).WillOnce(Return());
+    EXPECT_CALL(*pullerManager, UnRegisterReceiver(tagId, _)).WillOnce(Return());
+
+    ValueMetricProducer valueProducer(metric, -1 /*-1 meaning no condition*/, wizard,tagId,
+                                      bucketStartTimeNs, pullerManager);
+
+    vector<shared_ptr<LogEvent>> allData;
+    allData.clear();
+    shared_ptr<LogEvent> event = make_shared<LogEvent>(tagId, bucketStartTimeNs + 1);
+    auto list = event->GetAndroidLogEventList();
+    *list << 1;
+    *list << 11;
+    event->init();
+    allData.push_back(event);
+
+    valueProducer.onDataPulled(allData);
+    // has one slice
+    EXPECT_EQ(1UL, valueProducer.mCurrentSlicedBucket.size());
+    ValueMetricProducer::Interval curInterval = valueProducer.mCurrentSlicedBucket.begin()->second;
+    // has one raw pair
+    EXPECT_EQ(1UL, curInterval.raw.size());
+    // value is 11, 11
+    EXPECT_EQ(11, curInterval.raw.front().first);
+    EXPECT_EQ(11, curInterval.raw.front().second);
+    ValueMetricProducer::Interval nextInterval = valueProducer.mNextSlicedBucket.begin()->second;
+    // has one raw pair
+    EXPECT_EQ(1UL, nextInterval.raw.size());
+    // value is 11, 0
+    EXPECT_EQ(11, nextInterval.raw.front().first);
+    EXPECT_EQ(0, nextInterval.raw.front().second);
+    EXPECT_EQ(0UL, valueProducer.mPastBuckets.size());
+
+    allData.clear();
+    event = make_shared<LogEvent>(tagId, bucket2StartTimeNs + 1);
+    list = event->GetAndroidLogEventList();
+    *list << 1;
+    *list << 22;
+    event->init();
+    allData.push_back(event);
+    valueProducer.onDataPulled(allData);
+    // has one slice
+    EXPECT_EQ(1UL, valueProducer.mCurrentSlicedBucket.size());
+    curInterval = valueProducer.mCurrentSlicedBucket.begin()->second;
+    // has one raw pair
+    EXPECT_EQ(1UL, curInterval.raw.size());
+    // value is 22, 0
+    EXPECT_EQ(22, curInterval.raw.front().first);
+    EXPECT_EQ(0, curInterval.raw.front().second);
+    EXPECT_EQ(0UL, valueProducer.mNextSlicedBucket.size());
+    EXPECT_EQ(1UL, valueProducer.mPastBuckets.size());
+    EXPECT_EQ(1UL, valueProducer.mPastBuckets.begin()->second.size());
+    EXPECT_EQ(11, valueProducer.mPastBuckets.begin()->second.back().mValue);
+
+    allData.clear();
+    event = make_shared<LogEvent>(tagId, bucket3StartTimeNs + 1);
+    list = event->GetAndroidLogEventList();
+    *list << 1;
+    *list << 33;
+    event->init();
+    allData.push_back(event);
+    valueProducer.onDataPulled(allData);
+    EXPECT_EQ(1UL, valueProducer.mCurrentSlicedBucket.size());
+    curInterval = valueProducer.mCurrentSlicedBucket.begin()->second;
+    EXPECT_EQ(1UL, curInterval.raw.size());
+    // value is 33, 0
+    EXPECT_EQ(33, curInterval.raw.front().first);
+    EXPECT_EQ(0, curInterval.raw.front().second);
+    EXPECT_EQ(0UL, valueProducer.mNextSlicedBucket.size());
+    EXPECT_EQ(1UL, valueProducer.mPastBuckets.size());
+    EXPECT_EQ(2UL, valueProducer.mPastBuckets.begin()->second.size());
+    EXPECT_EQ(11, valueProducer.mPastBuckets.begin()->second.back().mValue);
+}
+
+/*
+ * Test pulled event with non sliced condition.
+ */
+TEST(ValueMetricProducerTest, TestEventsWithNonSlicedCondition) {
+    int64_t bucketStartTimeNs = 10000000000;
+    int64_t bucketSizeNs = 60 * 1000 * 1000 * 1000LL;
+
+    int64_t bucket2StartTimeNs = bucketStartTimeNs + bucketSizeNs;
+    int64_t bucket3StartTimeNs = bucketStartTimeNs + 2 * bucketSizeNs;
+
+    ValueMetric metric;
+    metric.set_metric_id(1);
+    metric.mutable_bucket()->set_bucket_size_millis(bucketSizeNs / 1000000);
+    metric.set_value_field(2);
+    metric.set_condition("SCREEN_ON");
+
+    int tagId = 1;
+
+    sp<MockConditionWizard> wizard = new NaggyMock<MockConditionWizard>();
+    shared_ptr<MockStatsPullerManager> pullerManager = make_shared<StrictMock<MockStatsPullerManager>>();
+    EXPECT_CALL(*pullerManager, RegisterReceiver(tagId, _, _)).WillOnce(Return());
+    EXPECT_CALL(*pullerManager, UnRegisterReceiver(tagId, _)).WillRepeatedly(Return());
+
+    EXPECT_CALL(*pullerManager, Pull(tagId, _)).WillOnce(Invoke([] (int tagId, vector<std::shared_ptr<LogEvent>>* data) {
+        int64_t bucketStartTimeNs = 10000000000;
+        int64_t bucketSizeNs = 60 * 1000 * 1000 * 1000LL;
+
+        int64_t bucket2StartTimeNs = bucketStartTimeNs + bucketSizeNs;
+        int64_t bucket3StartTimeNs = bucketStartTimeNs + 2 * bucketSizeNs;
+        data->clear();
+        shared_ptr<LogEvent> event = make_shared<LogEvent>(tagId, bucketStartTimeNs + 10);
+        auto list = event->GetAndroidLogEventList();
+        *list << 1;
+        *list << 100;
+        event->init();
+        data->push_back(event);
+        return true;
+    }))
+    .WillOnce(Invoke([] (int tagId, vector<std::shared_ptr<LogEvent>>* data) {
+        int64_t bucketStartTimeNs = 10000000000;
+        int64_t bucketSizeNs = 60 * 1000 * 1000 * 1000LL;
+
+        int64_t bucket2StartTimeNs = bucketStartTimeNs + bucketSizeNs;
+        int64_t bucket3StartTimeNs = bucketStartTimeNs + 2 * bucketSizeNs;
+        data->clear();
+        shared_ptr<LogEvent> event = make_shared<LogEvent>(tagId, bucket2StartTimeNs + 10);
+        auto list = event->GetAndroidLogEventList();
+        *list << 1;
+        *list << 120;
+        event->init();
+        data->push_back(event);
+        return true;
+    }));
+
+    ValueMetricProducer valueProducer(metric, 1, wizard,tagId,
+                                      bucketStartTimeNs, pullerManager);
+
+    valueProducer.onConditionChanged(true, bucketStartTimeNs + 10);
+
+    // has one slice
+    EXPECT_EQ(1UL, valueProducer.mCurrentSlicedBucket.size());
+    ValueMetricProducer::Interval curInterval = valueProducer.mCurrentSlicedBucket.begin()->second;
+    // has one raw pair
+    EXPECT_EQ(1UL, curInterval.raw.size());
+    // value is 100, 0
+    EXPECT_EQ(100, curInterval.raw.front().first);
+    EXPECT_EQ(0, curInterval.raw.front().second);
+    EXPECT_EQ(0UL, valueProducer.mNextSlicedBucket.size());
+    EXPECT_EQ(0UL, valueProducer.mPastBuckets.size());
+
+    vector<shared_ptr<LogEvent>> allData;
+    allData.clear();
+    shared_ptr<LogEvent> event = make_shared<LogEvent>(tagId, bucket2StartTimeNs + 1);
+    auto list = event->GetAndroidLogEventList();
+    *list << 1;
+    *list << 110;
+    event->init();
+    allData.push_back(event);
+    valueProducer.onDataPulled(allData);
+
+    // has one slice
+    EXPECT_EQ(1UL, valueProducer.mCurrentSlicedBucket.size());
+    curInterval = valueProducer.mCurrentSlicedBucket.begin()->second;
+    // has one raw pair
+    EXPECT_EQ(1UL, curInterval.raw.size());
+    // value is 110, 0
+    EXPECT_EQ(110, curInterval.raw.front().first);
+    EXPECT_EQ(0, curInterval.raw.front().second);
+    EXPECT_EQ(1UL, valueProducer.mPastBuckets.size());
+    EXPECT_EQ(1UL, valueProducer.mPastBuckets.begin()->second.size());
+    EXPECT_EQ(10, valueProducer.mPastBuckets.begin()->second.back().mValue);
+
+    valueProducer.onConditionChanged(false, bucket2StartTimeNs + 1);
+
+    // has one slice
+    EXPECT_EQ(1UL, valueProducer.mCurrentSlicedBucket.size());
+    curInterval = valueProducer.mCurrentSlicedBucket.begin()->second;
+    // has one raw pair
+    EXPECT_EQ(1UL, curInterval.raw.size());
+    // value is 110, 120
+    EXPECT_EQ(110, curInterval.raw.front().first);
+    EXPECT_EQ(120, curInterval.raw.front().second);
+}
+
+TEST(ValueMetricProducerTest, TestPushedEventsWithoutCondition) {
+    int64_t bucketStartTimeNs = 10000000000;
+    int64_t bucketSizeNs = 60 * 1000 * 1000 * 1000LL;
+
+    int64_t bucket2StartTimeNs = bucketStartTimeNs + bucketSizeNs;
+    int64_t bucket3StartTimeNs = bucketStartTimeNs + 2 * bucketSizeNs;
+
+    ValueMetric metric;
+    metric.set_metric_id(1);
+    metric.mutable_bucket()->set_bucket_size_millis(bucketSizeNs / 1000000);
+    metric.set_value_field(2);
+
+    int tagId = 1;
+
+    sp<MockConditionWizard> wizard = new NaggyMock<MockConditionWizard>();
+    shared_ptr<MockStatsPullerManager> pullerManager = make_shared<StrictMock<MockStatsPullerManager>>();
+
+    ValueMetricProducer valueProducer(metric, -1, wizard,-1,
+                                      bucketStartTimeNs, pullerManager);
+
+    shared_ptr<LogEvent> event1 = make_shared<LogEvent>(tagId, bucketStartTimeNs + 10);
+    auto list = event1->GetAndroidLogEventList();
+    *list << 1;
+    *list << 10;
+    event1->init();
+    shared_ptr<LogEvent> event2 = make_shared<LogEvent>(tagId, bucketStartTimeNs + 10);
+    auto list2 = event2->GetAndroidLogEventList();
+    *list2 << 1;
+    *list2 << 20;
+    event2->init();
+    valueProducer.onMatchedLogEvent(1 /*log matcher index*/, *event1, false);
+    // has one slice
+    EXPECT_EQ(1UL, valueProducer.mCurrentSlicedBucket.size());
+    ValueMetricProducer::Interval curInterval = valueProducer.mCurrentSlicedBucket.begin()->second;
+    // has one raw pair
+    EXPECT_EQ(1UL, curInterval.raw.size());
+    // value is 10, 0
+    EXPECT_EQ(10, curInterval.raw.front().first);
+    EXPECT_EQ(0, curInterval.raw.front().second);
+    EXPECT_EQ(0UL, valueProducer.mNextSlicedBucket.size());
+
+    valueProducer.onMatchedLogEvent(1 /*log matcher index*/, *event2, false);
+
+    // has one slice
+    EXPECT_EQ(1UL, valueProducer.mCurrentSlicedBucket.size());
+    curInterval = valueProducer.mCurrentSlicedBucket.begin()->second;
+    // has one raw pair
+    EXPECT_EQ(2UL, curInterval.raw.size());
+    // value is 10, 20
+    EXPECT_EQ(10, curInterval.raw.front().first);
+    EXPECT_EQ(20, curInterval.raw.back().first);
+    EXPECT_EQ(0UL, valueProducer.mNextSlicedBucket.size());
+
+    valueProducer.flush_if_needed(bucket3StartTimeNs);
+    EXPECT_EQ(1UL, valueProducer.mPastBuckets.size());
+    EXPECT_EQ(1UL, valueProducer.mPastBuckets.begin()->second.size());
+    EXPECT_EQ(30, valueProducer.mPastBuckets.begin()->second.back().mValue);
+}
+
+}  // namespace statsd
+}  // namespace os
+}  // namespace android
+#else
+GTEST_LOG_(INFO) << "This test does nothing.\n";
+#endif
diff --git a/cmds/statsd/tests/metrics/metrics_test_helper.h b/cmds/statsd/tests/metrics/metrics_test_helper.h
index 5fd7d621637d..fa221aae1731 100644
--- a/cmds/statsd/tests/metrics/metrics_test_helper.h
+++ b/cmds/statsd/tests/metrics/metrics_test_helper.h
@@ -14,6 +14,7 @@
 #pragma once
 
 #include "src/condition/ConditionWizard.h"
+#include "src/external/StatsPullerManager.h"
 
 #include <gmock/gmock.h>
 #include <gtest/gtest.h>
@@ -30,6 +31,13 @@ public:
                            const std::map<std::string, HashableDimensionKey>& conditionParameters));
 };
 
+class MockStatsPullerManager : public StatsPullerManager {
+public:
+    MOCK_METHOD3(RegisterReceiver, void(int tagId, wp<PullDataReceiver> receiver, long intervalMs));
+    MOCK_METHOD2(UnRegisterReceiver, void(int tagId, wp<PullDataReceiver> receiver));
+    MOCK_METHOD2(Pull, bool(const int pullCode, vector<std::shared_ptr<LogEvent>>* data));
+};
+
 }  // namespace statsd
 }  // namespace os
 }  // namespace android