Make StatsLog drop less.

+ Create a thread-safe LogEventQueue to buffer log events.

+ The socket listner thread will read from socket and write to the buffer as quickly as possible
  to minimize the data loss in socket.

+ All pushed data is fetched from the the buffer and processed in a dedicated thread. After an
  event is fetched from the queue, we no longer block the socket listener thread.

+ Report event queue stats via statsdstats, including the min and max queue event history span in
  the queue (to understand how slow statsd can be and how fast the events can be)

Bug: 119031518
Test: unit tests added in statsd_test

Change-Id: I6b65ed9a678935b2e24302ba4b36e69c157adde4

cmds/statsd/Android.bp

@@ -81,7 +81,7 @@ cc_defaults {
         "src/external/StatsPullerManager.cpp",
         "src/external/puller_util.cpp",
         "src/logd/LogEvent.cpp",
-        "src/logd/LogListener.cpp",
+        "src/logd/LogEventQueue.cpp",
         "src/matchers/CombinationLogMatchingTracker.cpp",
         "src/matchers/EventMatcherWizard.cpp",
         "src/matchers/matcher_util.cpp",
@@ -226,6 +226,7 @@ cc_test {
         "tests/indexed_priority_queue_test.cpp",
         "tests/LogEntryMatcher_test.cpp",
         "tests/LogEvent_test.cpp",
+        "tests/log_event/LogEventQueue_test.cpp",
         "tests/MetricsManager_test.cpp",
         "tests/StatsLogProcessor_test.cpp",
         "tests/StatsService_test.cpp",

cmds/statsd/src/StatsService.cpp

@@ -132,34 +132,36 @@ binder::Status checkDumpAndUsageStats(const String16& packageName) {
     }                                                             \
 }
 
-StatsService::StatsService(const sp<Looper>& handlerLooper)
-    : mAnomalyAlarmMonitor(new AlarmMonitor(MIN_DIFF_TO_UPDATE_REGISTERED_ALARM_SECS,
-       [](const sp<IStatsCompanionService>& sc, int64_t timeMillis) {
-           if (sc != nullptr) {
-               sc->setAnomalyAlarm(timeMillis);
-               StatsdStats::getInstance().noteRegisteredAnomalyAlarmChanged();
-           }
-       },
-       [](const sp<IStatsCompanionService>& sc) {
-           if (sc != nullptr) {
-               sc->cancelAnomalyAlarm();
-               StatsdStats::getInstance().noteRegisteredAnomalyAlarmChanged();
-           }
-       })),
-   mPeriodicAlarmMonitor(new AlarmMonitor(MIN_DIFF_TO_UPDATE_REGISTERED_ALARM_SECS,
-      [](const sp<IStatsCompanionService>& sc, int64_t timeMillis) {
-           if (sc != nullptr) {
-               sc->setAlarmForSubscriberTriggering(timeMillis);
-               StatsdStats::getInstance().noteRegisteredPeriodicAlarmChanged();
-           }
-      },
-      [](const sp<IStatsCompanionService>& sc) {
-           if (sc != nullptr) {
-               sc->cancelAlarmForSubscriberTriggering();
-               StatsdStats::getInstance().noteRegisteredPeriodicAlarmChanged();
-           }
-
-      }))  {
+StatsService::StatsService(const sp<Looper>& handlerLooper, shared_ptr<LogEventQueue> queue)
+    : mAnomalyAlarmMonitor(new AlarmMonitor(
+              MIN_DIFF_TO_UPDATE_REGISTERED_ALARM_SECS,
+              [](const sp<IStatsCompanionService>& sc, int64_t timeMillis) {
+                  if (sc != nullptr) {
+                      sc->setAnomalyAlarm(timeMillis);
+                      StatsdStats::getInstance().noteRegisteredAnomalyAlarmChanged();
+                  }
+              },
+              [](const sp<IStatsCompanionService>& sc) {
+                  if (sc != nullptr) {
+                      sc->cancelAnomalyAlarm();
+                      StatsdStats::getInstance().noteRegisteredAnomalyAlarmChanged();
+                  }
+              })),
+      mPeriodicAlarmMonitor(new AlarmMonitor(
+              MIN_DIFF_TO_UPDATE_REGISTERED_ALARM_SECS,
+              [](const sp<IStatsCompanionService>& sc, int64_t timeMillis) {
+                  if (sc != nullptr) {
+                      sc->setAlarmForSubscriberTriggering(timeMillis);
+                      StatsdStats::getInstance().noteRegisteredPeriodicAlarmChanged();
+                  }
+              },
+              [](const sp<IStatsCompanionService>& sc) {
+                  if (sc != nullptr) {
+                      sc->cancelAlarmForSubscriberTriggering();
+                      StatsdStats::getInstance().noteRegisteredPeriodicAlarmChanged();
+                  }
+              })),
+      mEventQueue(queue) {
     mUidMap = UidMap::getInstance();
     mPullerManager = new StatsPullerManager();
     StatsPuller::SetUidMap(mUidMap);
@@ -201,11 +203,33 @@ StatsService::StatsService(const sp<Looper>& handlerLooper)
     mConfigManager->AddListener(mProcessor);
 
     init_system_properties();
+
+    if (mEventQueue != nullptr) {
+        std::thread pushedEventThread([this] { readLogs(); });
+        pushedEventThread.detach();
+    }
 }
 
 StatsService::~StatsService() {
 }
 
+/* Runs on a dedicated thread to process pushed events. */
+void StatsService::readLogs() {
+    // Read forever..... long live statsd
+    while (1) {
+        // Block until an event is available.
+        auto event = mEventQueue->waitPop();
+        // Pass it to StatsLogProcess to all configs/metrics
+        // At this point, the LogEventQueue is not blocked, so that the socketListener
+        // can read events from the socket and write to buffer to avoid data drop.
+        mProcessor->OnLogEvent(event.get());
+        // The ShellSubscriber is only used by shell for local debugging.
+        if (mShellSubscriber != nullptr) {
+            mShellSubscriber->onLogEvent(*event);
+        }
+    }
+}
+
 void StatsService::init_system_properties() {
     mEngBuild = false;
     const prop_info* buildType = __system_property_find("ro.build.type");
@@ -1009,6 +1033,7 @@ void StatsService::Terminate() {
     }
 }
 
+// Test only interface!!!
 void StatsService::OnLogEvent(LogEvent* event) {
     mProcessor->OnLogEvent(event);
     if (mShellSubscriber != nullptr) {

cmds/statsd/src/StatsService.h

@@ -22,7 +22,7 @@
 #include "anomaly/AlarmMonitor.h"
 #include "config/ConfigManager.h"
 #include "external/StatsPullerManager.h"
-#include "logd/LogListener.h"
+#include "logd/LogEventQueue.h"
 #include "packages/UidMap.h"
 #include "shell/ShellSubscriber.h"
 #include "statscompanion_util.h"
@@ -52,11 +52,10 @@ namespace statsd {
 using android::hardware::Return;
 
 class StatsService : public BnStatsManager,
-                     public LogListener,
                      public IStats,
                      public IBinder::DeathRecipient {
 public:
-    StatsService(const sp<Looper>& handlerLooper);
+    StatsService(const sp<Looper>& handlerLooper, std::shared_ptr<LogEventQueue> queue);
     virtual ~StatsService();
 
     /** The anomaly alarm registered with AlarmManager won't be updated by less than this. */
@@ -92,7 +91,7 @@ public:
     void Terminate();
 
     /**
-     * Called by LogReader when there's a log event to process.
+     * Test ONLY interface. In real world, StatsService reads from LogEventQueue.
      */
     virtual void OnLogEvent(LogEvent* event);
 
@@ -278,6 +277,9 @@ private:
      */
     void print_cmd_help(int out);
 
+    /* Runs on its dedicated thread to process pushed stats event from socket. */
+    void readLogs();
+
     /**
      * Trigger a broadcast.
      */
@@ -410,6 +412,8 @@ private:
 
     sp<ShellSubscriber> mShellSubscriber;
 
+    std::shared_ptr<LogEventQueue> mEventQueue;
+
     FRIEND_TEST(StatsServiceTest, TestAddConfig_simple);
     FRIEND_TEST(StatsServiceTest, TestAddConfig_empty);
     FRIEND_TEST(StatsServiceTest, TestAddConfig_invalid);

cmds/statsd/src/guardrail/StatsdStats.cpp

@@ -50,6 +50,7 @@ const int FIELD_ID_ANOMALY_ALARM_STATS = 9;
 const int FIELD_ID_PERIODIC_ALARM_STATS = 12;
 const int FIELD_ID_SYSTEM_SERVER_RESTART = 15;
 const int FIELD_ID_LOGGER_ERROR_STATS = 16;
+const int FIELD_ID_OVERFLOW = 18;
 
 const int FIELD_ID_ATOM_STATS_TAG = 1;
 const int FIELD_ID_ATOM_STATS_COUNT = 2;
@@ -64,6 +65,10 @@ const int FIELD_ID_LOG_LOSS_STATS_TAG = 4;
 const int FIELD_ID_LOG_LOSS_STATS_UID = 5;
 const int FIELD_ID_LOG_LOSS_STATS_PID = 6;
 
+const int FIELD_ID_OVERFLOW_COUNT = 1;
+const int FIELD_ID_OVERFLOW_MAX_HISTORY = 2;
+const int FIELD_ID_OVERFLOW_MIN_HISTORY = 3;
+
 const int FIELD_ID_CONFIG_STATS_UID = 1;
 const int FIELD_ID_CONFIG_STATS_ID = 2;
 const int FIELD_ID_CONFIG_STATS_CREATION = 3;
@@ -235,6 +240,22 @@ void StatsdStats::noteDataDropped(const ConfigKey& key, const size_t totalBytes)
     noteDataDropped(key, totalBytes, getWallClockSec());
 }
 
+void StatsdStats::noteEventQueueOverflow(int64_t oldestEventTimestampNs) {
+    lock_guard<std::mutex> lock(mLock);
+
+    mOverflowCount++;
+
+    int64_t history = getElapsedRealtimeNs() - oldestEventTimestampNs;
+
+    if (history > mMaxQueueHistoryNs) {
+        mMaxQueueHistoryNs = history;
+    }
+
+    if (history < mMinQueueHistoryNs) {
+        mMinQueueHistoryNs = history;
+    }
+}
+
 void StatsdStats::noteDataDropped(const ConfigKey& key, const size_t totalBytes, int32_t timeSec) {
     lock_guard<std::mutex> lock(mLock);
     auto it = mConfigStats.find(key);
@@ -534,6 +555,9 @@ void StatsdStats::resetInternalLocked() {
     mPeriodicAlarmRegisteredStats = 0;
     mSystemServerRestartSec.clear();
     mLogLossStats.clear();
+    mOverflowCount = 0;
+    mMinQueueHistoryNs = kInt64Max;
+    mMaxQueueHistoryNs = 0;
     for (auto& config : mConfigStats) {
         config.second->broadcast_sent_time_sec.clear();
         config.second->activation_time_sec.clear();
@@ -726,6 +750,9 @@ void StatsdStats::dumpStats(int out) const {
                 (long long)loss.mWallClockSec, loss.mCount, loss.mLastError, loss.mLastTag,
                 loss.mUid, loss.mPid);
     }
+
+    dprintf(out, "Event queue overflow: %d; MaxHistoryNs: %lld; MinHistoryNs: %lld\n",
+            mOverflowCount, (long long)mMaxQueueHistoryNs, (long long)mMinQueueHistoryNs);
 }
 
 void addConfigStatsToProto(const ConfigStats& configStats, ProtoOutputStream* proto) {
@@ -904,6 +931,16 @@ void StatsdStats::dumpStats(std::vector<uint8_t>* output, bool reset) {
         proto.end(token);
     }
 
+    if (mOverflowCount > 0) {
+        uint64_t token = proto.start(FIELD_TYPE_MESSAGE | FIELD_ID_OVERFLOW);
+        proto.write(FIELD_TYPE_INT32 | FIELD_ID_OVERFLOW_COUNT, (int32_t)mOverflowCount);
+        proto.write(FIELD_TYPE_INT64 | FIELD_ID_OVERFLOW_MAX_HISTORY,
+                    (long long)mMaxQueueHistoryNs);
+        proto.write(FIELD_TYPE_INT64 | FIELD_ID_OVERFLOW_MIN_HISTORY,
+                    (long long)mMinQueueHistoryNs);
+        proto.end(token);
+    }
+
     for (const auto& restart : mSystemServerRestartSec) {
         proto.write(FIELD_TYPE_INT32 | FIELD_ID_SYSTEM_SERVER_RESTART | FIELD_COUNT_REPEATED,
                     restart);

cmds/statsd/src/guardrail/StatsdStats.h

@@ -160,6 +160,8 @@ public:
     // Max platform atom tag number.
     static const int32_t kMaxPlatformAtomTag = 100000;
 
+    static const int64_t kInt64Max = 0x7fffffffffffffffLL;
+
     /**
      * Report a new config has been received and report the static stats about the config.
      *
@@ -419,6 +421,10 @@ public:
      */
     void noteBucketUnknownCondition(int64_t metricId);
 
+    /* Reports one event has been dropped due to queue overflow, and the oldest event timestamp in
+     * the queue */
+    void noteEventQueueOverflow(int64_t oldestEventTimestampNs);
+
     /**
      * Reset the historical stats. Including all stats in icebox, and the tracked stats about
      * metrics, matchers, and atoms. The active configs will be kept and StatsdStats will continue
@@ -522,6 +528,17 @@ private:
         int32_t mPid;
     };
 
+    // Max of {(now - oldestEventTimestamp) when overflow happens}.
+    // This number is helpful to understand how SLOW statsd can be.
+    int64_t mMaxQueueHistoryNs = 0;
+
+    // Min of {(now - oldestEventTimestamp) when overflow happens}.
+    // This number is helpful to understand how FAST the events floods to statsd.
+    int64_t mMinQueueHistoryNs = kInt64Max;
+
+    // Total number of events that are lost due to queue overflow.
+    int32_t mOverflowCount = 0;
+
     // Timestamps when we detect log loss, and the number of logs lost.
     std::list<LogLossStats> mLogLossStats;
 

cmds/statsd/src/logd/LogEvent.cpp

@@ -118,6 +118,7 @@ void LogEvent::createLogEvents(const StatsLogEventWrapper& statsLogEventWrapper,
 
 LogEvent::LogEvent(int32_t tagId, int64_t wallClockTimestampNs, int64_t elapsedTimestampNs) {
     mLogdTimestampNs = wallClockTimestampNs;
+    mElapsedTimestampNs = elapsedTimestampNs;
     mTagId = tagId;
     mLogUid = 0;
     mContext = create_android_logger(1937006964); // the event tag shared by all stats logs
@@ -246,7 +247,8 @@ LogEvent::LogEvent(int64_t wallClockTimestampNs, int64_t elapsedTimestampNs,
     mValues.push_back(FieldValue(Field(mTagId, getSimpleField(4)), Value(trainInfo.status)));
 }
 
-LogEvent::LogEvent(int32_t tagId, int64_t timestampNs) : LogEvent(tagId, timestampNs, 0) {}
+LogEvent::LogEvent(int32_t tagId, int64_t timestampNs) : LogEvent(tagId, timestampNs, timestampNs) {
+}
 
 LogEvent::LogEvent(int32_t tagId, int64_t timestampNs, int32_t uid) {
     mLogdTimestampNs = timestampNs;

cmds/statsd/src/logd/LogEventQueue.cpp

@@ -0,0 +1,62 @@
+/*
+ * Copyright (C) 2019 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.
+ */
+
+#define DEBUG false  // STOPSHIP if true
+#include "Log.h"
+
+#include "LogEventQueue.h"
+
+namespace android {
+namespace os {
+namespace statsd {
+
+using std::unique_lock;
+using std::unique_ptr;
+
+unique_ptr<LogEvent> LogEventQueue::waitPop() {
+    std::unique_lock<std::mutex> lock(mMutex);
+
+    if (mQueue.empty()) {
+        mCondition.wait(lock, [this] { return !this->mQueue.empty(); });
+    }
+
+    unique_ptr<LogEvent> item = std::move(mQueue.front());
+    mQueue.pop();
+
+    return item;
+}
+
+bool LogEventQueue::push(unique_ptr<LogEvent> item, int64_t* oldestTimestampNs) {
+    bool success;
+    {
+        std::unique_lock<std::mutex> lock(mMutex);
+        if (mQueue.size() < mQueueLimit) {
+            mQueue.push(std::move(item));
+            success = true;
+        } else {
+            // safe operation as queue must not be empty.
+            *oldestTimestampNs = mQueue.front()->GetElapsedTimestampNs();
+            success = false;
+        }
+    }
+
+    mCondition.notify_one();
+    return success;
+}
+
+}  // namespace statsd
+}  // namespace os
+}  // namespace android

cmds/statsd/src/logd/LogEventQueue.h

@@ -0,0 +1,59 @@
+/*
+ * Copyright (C) 2019 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 "LogEvent.h"
+
+#include <condition_variable>
+#include <memory>
+#include <mutex>
+#include <queue>
+#include <thread>
+
+namespace android {
+namespace os {
+namespace statsd {
+
+/**
+ * A zero copy thread safe queue buffer for producing and consuming LogEvent.
+ */
+class LogEventQueue {
+public:
+    explicit LogEventQueue(size_t maxSize) : mQueueLimit(maxSize){};
+
+    /**
+     * Blocking read one event from the queue.
+     */
+    std::unique_ptr<LogEvent> waitPop();
+
+    /**
+     * Puts a LogEvent ptr to the end of the queue.
+     * Returns false on failure when the queue is full, and output the oldest event timestamp
+     * in the queue.
+     */
+    bool push(std::unique_ptr<LogEvent> event, int64_t* oldestTimestampNs);
+
+private:
+    const size_t mQueueLimit;
+    std::condition_variable mCondition;
+    std::mutex mMutex;
+    std::queue<std::unique_ptr<LogEvent>> mQueue;
+};
+
+}  // namespace statsd
+}  // namespace os
+}  // namespace android

cmds/statsd/src/logd/LogListener.cpp

@@ -1,31 +0,0 @@
-/*
- * 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 "logd/LogListener.h"
-
-namespace android {
-namespace os {
-namespace statsd {
-
-LogListener::LogListener() {
-}
-
-LogListener::~LogListener() {
-}
-
-}  // namespace statsd
-}  // namespace os
-}  // namespace android

cmds/statsd/src/logd/LogListener.h

@@ -1,40 +0,0 @@
-/*
- * 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 "logd/LogEvent.h"
-
-#include <utils/RefBase.h>
-
-namespace android {
-namespace os {
-namespace statsd {
-
-/**
- * Callback for LogReader
- */
-class LogListener : public virtual android::RefBase {
-public:
-    LogListener();
-    virtual ~LogListener();
-
-    virtual void OnLogEvent(LogEvent* msg) = 0;
-};
-
-}  // namespace statsd
-}  // namespace os
-}  // namespace android

cmds/statsd/src/main.cpp

@@ -80,8 +80,11 @@ int main(int /*argc*/, char** /*argv*/) {
 
     ::android::hardware::configureRpcThreadpool(1 /*threads*/, false /*willJoin*/);
 
+    std::shared_ptr<LogEventQueue> eventQueue =
+            std::make_shared<LogEventQueue>(2000 /*buffer limit. Buffer is NOT pre-allocated*/);
+
     // Create the service
-    gStatsService = new StatsService(looper);
+    gStatsService = new StatsService(looper, eventQueue);
     if (defaultServiceManager()->addService(String16("stats"), gStatsService, false,
                 IServiceManager::DUMP_FLAG_PRIORITY_NORMAL | IServiceManager::DUMP_FLAG_PROTO)
             != 0) {
@@ -101,13 +104,13 @@ int main(int /*argc*/, char** /*argv*/) {
 
     gStatsService->Startup();
 
-    sp<StatsSocketListener> socketListener = new StatsSocketListener(gStatsService);
+    sp<StatsSocketListener> socketListener = new StatsSocketListener(eventQueue);
 
-        ALOGI("using statsd socket");
-        // Backlog and /proc/sys/net/unix/max_dgram_qlen set to large value
-        if (socketListener->startListener(600)) {
-            exit(1);
-        }
+    ALOGI("Statsd starts to listen to socket.");
+    // Backlog and /proc/sys/net/unix/max_dgram_qlen set to large value
+    if (socketListener->startListener(600)) {
+        exit(1);
+    }
 
     // Loop forever -- the reports run on this thread in a handler, and the
     // binder calls remain responsive in their pool of one thread.

cmds/statsd/src/socket/StatsSocketListener.cpp

@@ -41,8 +41,8 @@ namespace statsd {
 
 static const int kLogMsgHeaderSize = 28;
 
-StatsSocketListener::StatsSocketListener(const sp<LogListener>& listener)
-    : SocketListener(getLogSocket(), false /*start listen*/), mListener(listener) {
+StatsSocketListener::StatsSocketListener(std::shared_ptr<LogEventQueue> queue)
+    : SocketListener(getLogSocket(), false /*start listen*/), mQueue(queue) {
 }
 
 StatsSocketListener::~StatsSocketListener() {
@@ -134,10 +134,11 @@ bool StatsSocketListener::onDataAvailable(SocketClient* cli) {
     msg.entry.uid = cred->uid;
 
     memcpy(msg.buf + kLogMsgHeaderSize, ptr, n + 1);
-    LogEvent event(msg);
 
-    // Call the listener
-    mListener->OnLogEvent(&event);
+    int64_t oldestTimestamp;
+    if (!mQueue->push(std::make_unique<LogEvent>(msg), &oldestTimestamp)) {
+        StatsdStats::getInstance().noteEventQueueOverflow(oldestTimestamp);
+    }
 
     return true;
 }

cmds/statsd/src/socket/StatsSocketListener.h

@@ -17,7 +17,7 @@
 
 #include <sysutils/SocketListener.h>
 #include <utils/RefBase.h>
-#include "logd/LogListener.h"
+#include "logd/LogEventQueue.h"
 
 // DEFAULT_OVERFLOWUID is defined in linux/highuid.h, which is not part of
 // the uapi headers for userspace to use.  This value is filled in on the
@@ -35,7 +35,7 @@ namespace statsd {
 
 class StatsSocketListener : public SocketListener, public virtual android::RefBase {
 public:
-    explicit StatsSocketListener(const sp<LogListener>& listener);
+    explicit StatsSocketListener(std::shared_ptr<LogEventQueue> queue);
 
     virtual ~StatsSocketListener();
 
@@ -47,7 +47,7 @@ private:
     /**
      * Who is going to get the events when they're read.
      */
-    sp<LogListener> mListener;
+    std::shared_ptr<LogEventQueue> mQueue;
 };
 }  // namespace statsd
 }  // namespace os

cmds/statsd/src/stats_log.proto

@@ -460,6 +460,14 @@ message StatsdStatsReport {
         optional int32 pid = 6;
     }
     repeated LogLossStats detected_log_loss = 16;
+
+    message EventQueueOverflow {
+        optional int32 count = 1;
+        optional int64 max_queue_history_ns = 2;
+        optional int64 min_queue_history_ns = 3;
+    }
+
+    optional EventQueueOverflow queue_overflow = 18;
 }
 
 message AlertTriggerDetails {

cmds/statsd/tests/StatsService_test.cpp

@@ -33,7 +33,7 @@ using android::util::ProtoOutputStream;
 #ifdef __ANDROID__
 
 TEST(StatsServiceTest, TestAddConfig_simple) {
-    StatsService service(nullptr);
+    StatsService service(nullptr, nullptr);
     StatsdConfig config;
     config.set_id(12345);
     string serialized = config.SerializeAsString();
@@ -43,7 +43,7 @@ TEST(StatsServiceTest, TestAddConfig_simple) {
 }
 
 TEST(StatsServiceTest, TestAddConfig_empty) {
-    StatsService service(nullptr);
+    StatsService service(nullptr, nullptr);
     string serialized = "";
 
     EXPECT_TRUE(
@@ -51,7 +51,7 @@ TEST(StatsServiceTest, TestAddConfig_empty) {
 }
 
 TEST(StatsServiceTest, TestAddConfig_invalid) {
-    StatsService service(nullptr);
+    StatsService service(nullptr, nullptr);
     string serialized = "Invalid config!";
 
     EXPECT_FALSE(
@@ -69,7 +69,7 @@ TEST(StatsServiceTest, TestGetUidFromArgs) {
 
     int32_t uid;
 
-    StatsService service(nullptr);
+    StatsService service(nullptr, nullptr);
     service.mEngBuild = true;
 
     // "-1"

cmds/statsd/tests/e2e/PartialBucket_e2e_test.cpp

@@ -111,7 +111,7 @@ StatsdConfig MakeGaugeMetricConfig(int64_t minTime) {
 }
 
 TEST(PartialBucketE2eTest, TestCountMetricWithoutSplit) {
-    StatsService service(nullptr);
+    StatsService service(nullptr, nullptr);
     SendConfig(service, MakeConfig());
     int64_t start = getElapsedRealtimeNs();  // This is the start-time the metrics producers are
                                              // initialized with.
@@ -126,7 +126,7 @@ TEST(PartialBucketE2eTest, TestCountMetricWithoutSplit) {
 }
 
 TEST(PartialBucketE2eTest, TestCountMetricNoSplitOnNewApp) {
-    StatsService service(nullptr);
+    StatsService service(nullptr, nullptr);
     SendConfig(service, MakeConfig());
     int64_t start = getElapsedRealtimeNs();  // This is the start-time the metrics producers are
                                              // initialized with.
@@ -146,7 +146,7 @@ TEST(PartialBucketE2eTest, TestCountMetricNoSplitOnNewApp) {
 }
 
 TEST(PartialBucketE2eTest, TestCountMetricSplitOnUpgrade) {
-    StatsService service(nullptr);
+    StatsService service(nullptr, nullptr);
     SendConfig(service, MakeConfig());
     int64_t start = getElapsedRealtimeNs();  // This is the start-time the metrics producers are
                                              // initialized with.
@@ -171,7 +171,7 @@ TEST(PartialBucketE2eTest, TestCountMetricSplitOnUpgrade) {
 }
 
 TEST(PartialBucketE2eTest, TestCountMetricSplitOnRemoval) {
-    StatsService service(nullptr);
+    StatsService service(nullptr, nullptr);
     SendConfig(service, MakeConfig());
     int64_t start = getElapsedRealtimeNs();  // This is the start-time the metrics producers are
                                              // initialized with.
@@ -195,7 +195,7 @@ TEST(PartialBucketE2eTest, TestCountMetricSplitOnRemoval) {
 }
 
 TEST(PartialBucketE2eTest, TestValueMetricWithoutMinPartialBucket) {
-    StatsService service(nullptr);
+    StatsService service(nullptr, nullptr);
     // Partial buckets don't occur when app is first installed.
     service.mUidMap->updateApp(1, String16(kApp1.c_str()), 1, 1, String16("v1"), String16(""));
     SendConfig(service, MakeValueMetricConfig(0));
@@ -213,7 +213,7 @@ TEST(PartialBucketE2eTest, TestValueMetricWithoutMinPartialBucket) {
 }
 
 TEST(PartialBucketE2eTest, TestValueMetricWithMinPartialBucket) {
-    StatsService service(nullptr);
+    StatsService service(nullptr, nullptr);
     // Partial buckets don't occur when app is first installed.
     service.mUidMap->updateApp(1, String16(kApp1.c_str()), 1, 1, String16("v1"), String16(""));
     SendConfig(service, MakeValueMetricConfig(60 * NS_PER_SEC /* One minute */));
@@ -237,7 +237,7 @@ TEST(PartialBucketE2eTest, TestValueMetricWithMinPartialBucket) {
 }
 
 TEST(PartialBucketE2eTest, TestGaugeMetricWithoutMinPartialBucket) {
-    StatsService service(nullptr);
+    StatsService service(nullptr, nullptr);
     // Partial buckets don't occur when app is first installed.
     service.mUidMap->updateApp(1, String16(kApp1.c_str()), 1, 1, String16("v1"), String16(""));
     SendConfig(service, MakeGaugeMetricConfig(0));
@@ -255,7 +255,7 @@ TEST(PartialBucketE2eTest, TestGaugeMetricWithoutMinPartialBucket) {
 }
 
 TEST(PartialBucketE2eTest, TestGaugeMetricWithMinPartialBucket) {
-    StatsService service(nullptr);
+    StatsService service(nullptr, nullptr);
     // Partial buckets don't occur when app is first installed.
     service.mUidMap->updateApp(1, String16(kApp1.c_str()), 1, 1, String16("v1"), String16(""));
     SendConfig(service, MakeGaugeMetricConfig(60 * NS_PER_SEC /* One minute */));

cmds/statsd/tests/log_event/LogEventQueue_test.cpp

@@ -0,0 +1,100 @@
+// Copyright (C) 2019 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 "logd/LogEventQueue.h"
+
+#include <gmock/gmock.h>
+#include <gtest/gtest.h>
+#include <thread>
+
+#include <stdio.h>
+
+namespace android {
+namespace os {
+namespace statsd {
+
+using namespace android;
+using namespace testing;
+
+using std::unique_ptr;
+
+#ifdef __ANDROID__
+TEST(LogEventQueue_test, TestGoodConsumer) {
+    LogEventQueue queue(50);
+    int64_t timeBaseNs = 100;
+    std::thread writer([&queue, timeBaseNs] {
+        for (int i = 0; i < 100; i++) {
+            int64_t oldestEventNs;
+            bool success = queue.push(std::make_unique<LogEvent>(10, timeBaseNs + i * 1000),
+                                      &oldestEventNs);
+            EXPECT_TRUE(success);
+            std::this_thread::sleep_for(std::chrono::milliseconds(1));
+        }
+    });
+
+    std::thread reader([&queue, timeBaseNs] {
+        for (int i = 0; i < 100; i++) {
+            auto event = queue.waitPop();
+            EXPECT_TRUE(event != nullptr);
+            // All events are in right order.
+            EXPECT_EQ(timeBaseNs + i * 1000, event->GetElapsedTimestampNs());
+        }
+    });
+
+    reader.join();
+    writer.join();
+}
+
+TEST(LogEventQueue_test, TestSlowConsumer) {
+    LogEventQueue queue(50);
+    int64_t timeBaseNs = 100;
+    std::thread writer([&queue, timeBaseNs] {
+        int failure_count = 0;
+        int64_t oldestEventNs;
+        for (int i = 0; i < 100; i++) {
+            bool success = queue.push(std::make_unique<LogEvent>(10, timeBaseNs + i * 1000),
+                                      &oldestEventNs);
+            if (!success) failure_count++;
+            std::this_thread::sleep_for(std::chrono::milliseconds(1));
+        }
+
+        // There is some remote chance that reader thread not get chance to run before writer thread
+        // ends. That's why the following comparison is not "==".
+        // There will be at least 45 events lost due to overflow.
+        EXPECT_TRUE(failure_count >= 45);
+        // The oldest event must be at least the 6th event.
+        EXPECT_TRUE(oldestEventNs <= (100 + 5 * 1000));
+    });
+
+    std::thread reader([&queue, timeBaseNs] {
+        // The consumer quickly processed 5 events, then it got stuck (not reading anymore).
+        for (int i = 0; i < 5; i++) {
+            auto event = queue.waitPop();
+            EXPECT_TRUE(event != nullptr);
+            // All events are in right order.
+            EXPECT_EQ(timeBaseNs + i * 1000, event->GetElapsedTimestampNs());
+        }
+    });
+
+    reader.join();
+    writer.join();
+}
+
+#else
+GTEST_LOG_(INFO) << "This test does nothing.\n";
+#endif
+
+}  // namespace statsd
+}  // namespace os
+}  // namespace android