From 0f861867c609205d23cdfb655e2a068841f87266 Mon Sep 17 00:00:00 2001 From: Yao Chen Date: Wed, 27 Mar 2019 11:51:15 -0700 Subject: [PATCH] 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 | 3 +- cmds/statsd/src/StatsService.cpp | 81 +++++++++----- cmds/statsd/src/StatsService.h | 12 ++- cmds/statsd/src/guardrail/StatsdStats.cpp | 37 +++++++ cmds/statsd/src/guardrail/StatsdStats.h | 17 +++ cmds/statsd/src/logd/LogEvent.cpp | 4 +- cmds/statsd/src/logd/LogEventQueue.cpp | 62 +++++++++++ cmds/statsd/src/logd/LogEventQueue.h | 59 +++++++++++ cmds/statsd/src/logd/LogListener.cpp | 31 ------ cmds/statsd/src/logd/LogListener.h | 40 ------- cmds/statsd/src/main.cpp | 17 +-- .../statsd/src/socket/StatsSocketListener.cpp | 11 +- cmds/statsd/src/socket/StatsSocketListener.h | 6 +- cmds/statsd/src/stats_log.proto | 8 ++ cmds/statsd/tests/StatsService_test.cpp | 8 +- .../tests/e2e/PartialBucket_e2e_test.cpp | 16 +-- .../tests/log_event/LogEventQueue_test.cpp | 100 ++++++++++++++++++ 17 files changed, 380 insertions(+), 132 deletions(-) create mode 100644 cmds/statsd/src/logd/LogEventQueue.cpp create mode 100644 cmds/statsd/src/logd/LogEventQueue.h delete mode 100644 cmds/statsd/src/logd/LogListener.cpp delete mode 100644 cmds/statsd/src/logd/LogListener.h create mode 100644 cmds/statsd/tests/log_event/LogEventQueue_test.cpp diff --git a/cmds/statsd/Android.bp b/cmds/statsd/Android.bp index 8cd409e5e8c1e..017cb6d9221ec 100644 --- a/cmds/statsd/Android.bp +++ b/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", diff --git a/cmds/statsd/src/StatsService.cpp b/cmds/statsd/src/StatsService.cpp index 52ecdc8425af4..bfe2da93db1d7 100644 --- a/cmds/statsd/src/StatsService.cpp +++ b/cmds/statsd/src/StatsService.cpp @@ -132,34 +132,36 @@ binder::Status checkDumpAndUsageStats(const String16& packageName) { } \ } -StatsService::StatsService(const sp& handlerLooper) - : mAnomalyAlarmMonitor(new AlarmMonitor(MIN_DIFF_TO_UPDATE_REGISTERED_ALARM_SECS, - [](const sp& sc, int64_t timeMillis) { - if (sc != nullptr) { - sc->setAnomalyAlarm(timeMillis); - StatsdStats::getInstance().noteRegisteredAnomalyAlarmChanged(); - } - }, - [](const sp& sc) { - if (sc != nullptr) { - sc->cancelAnomalyAlarm(); - StatsdStats::getInstance().noteRegisteredAnomalyAlarmChanged(); - } - })), - mPeriodicAlarmMonitor(new AlarmMonitor(MIN_DIFF_TO_UPDATE_REGISTERED_ALARM_SECS, - [](const sp& sc, int64_t timeMillis) { - if (sc != nullptr) { - sc->setAlarmForSubscriberTriggering(timeMillis); - StatsdStats::getInstance().noteRegisteredPeriodicAlarmChanged(); - } - }, - [](const sp& sc) { - if (sc != nullptr) { - sc->cancelAlarmForSubscriberTriggering(); - StatsdStats::getInstance().noteRegisteredPeriodicAlarmChanged(); - } - - })) { +StatsService::StatsService(const sp& handlerLooper, shared_ptr queue) + : mAnomalyAlarmMonitor(new AlarmMonitor( + MIN_DIFF_TO_UPDATE_REGISTERED_ALARM_SECS, + [](const sp& sc, int64_t timeMillis) { + if (sc != nullptr) { + sc->setAnomalyAlarm(timeMillis); + StatsdStats::getInstance().noteRegisteredAnomalyAlarmChanged(); + } + }, + [](const sp& sc) { + if (sc != nullptr) { + sc->cancelAnomalyAlarm(); + StatsdStats::getInstance().noteRegisteredAnomalyAlarmChanged(); + } + })), + mPeriodicAlarmMonitor(new AlarmMonitor( + MIN_DIFF_TO_UPDATE_REGISTERED_ALARM_SECS, + [](const sp& sc, int64_t timeMillis) { + if (sc != nullptr) { + sc->setAlarmForSubscriberTriggering(timeMillis); + StatsdStats::getInstance().noteRegisteredPeriodicAlarmChanged(); + } + }, + [](const sp& 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& 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) { diff --git a/cmds/statsd/src/StatsService.h b/cmds/statsd/src/StatsService.h index d24565a630544..cbe5b4838275d 100644 --- a/cmds/statsd/src/StatsService.h +++ b/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& handlerLooper); + StatsService(const sp& handlerLooper, std::shared_ptr 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 mShellSubscriber; + std::shared_ptr mEventQueue; + FRIEND_TEST(StatsServiceTest, TestAddConfig_simple); FRIEND_TEST(StatsServiceTest, TestAddConfig_empty); FRIEND_TEST(StatsServiceTest, TestAddConfig_invalid); diff --git a/cmds/statsd/src/guardrail/StatsdStats.cpp b/cmds/statsd/src/guardrail/StatsdStats.cpp index 29100aad81aa1..74a4c87af7c34 100644 --- a/cmds/statsd/src/guardrail/StatsdStats.cpp +++ b/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 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 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* 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); diff --git a/cmds/statsd/src/guardrail/StatsdStats.h b/cmds/statsd/src/guardrail/StatsdStats.h index 434920ebe8c72..88ecccc32407a 100644 --- a/cmds/statsd/src/guardrail/StatsdStats.h +++ b/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 mLogLossStats; diff --git a/cmds/statsd/src/logd/LogEvent.cpp b/cmds/statsd/src/logd/LogEvent.cpp index d9f5415463e33..d5e66fdf97904 100644 --- a/cmds/statsd/src/logd/LogEvent.cpp +++ b/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; diff --git a/cmds/statsd/src/logd/LogEventQueue.cpp b/cmds/statsd/src/logd/LogEventQueue.cpp new file mode 100644 index 0000000000000..146464bbe7748 --- /dev/null +++ b/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 LogEventQueue::waitPop() { + std::unique_lock lock(mMutex); + + if (mQueue.empty()) { + mCondition.wait(lock, [this] { return !this->mQueue.empty(); }); + } + + unique_ptr item = std::move(mQueue.front()); + mQueue.pop(); + + return item; +} + +bool LogEventQueue::push(unique_ptr item, int64_t* oldestTimestampNs) { + bool success; + { + std::unique_lock 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 diff --git a/cmds/statsd/src/logd/LogEventQueue.h b/cmds/statsd/src/logd/LogEventQueue.h new file mode 100644 index 0000000000000..b4fd63f119e60 --- /dev/null +++ b/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 +#include +#include +#include +#include + +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 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 event, int64_t* oldestTimestampNs); + +private: + const size_t mQueueLimit; + std::condition_variable mCondition; + std::mutex mMutex; + std::queue> mQueue; +}; + +} // namespace statsd +} // namespace os +} // namespace android diff --git a/cmds/statsd/src/logd/LogListener.cpp b/cmds/statsd/src/logd/LogListener.cpp deleted file mode 100644 index ddb26f9fe5654..0000000000000 --- a/cmds/statsd/src/logd/LogListener.cpp +++ /dev/null @@ -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 diff --git a/cmds/statsd/src/logd/LogListener.h b/cmds/statsd/src/logd/LogListener.h deleted file mode 100644 index d8b06e9fab924..0000000000000 --- a/cmds/statsd/src/logd/LogListener.h +++ /dev/null @@ -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 - -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 diff --git a/cmds/statsd/src/main.cpp b/cmds/statsd/src/main.cpp index eddc86eca798a..68082c2dc4d2f 100644 --- a/cmds/statsd/src/main.cpp +++ b/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 eventQueue = + std::make_shared(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 socketListener = new StatsSocketListener(gStatsService); + sp 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. diff --git a/cmds/statsd/src/socket/StatsSocketListener.cpp b/cmds/statsd/src/socket/StatsSocketListener.cpp index aed926d74b265..92200f99c3ccb 100755 --- a/cmds/statsd/src/socket/StatsSocketListener.cpp +++ b/cmds/statsd/src/socket/StatsSocketListener.cpp @@ -41,8 +41,8 @@ namespace statsd { static const int kLogMsgHeaderSize = 28; -StatsSocketListener::StatsSocketListener(const sp& listener) - : SocketListener(getLogSocket(), false /*start listen*/), mListener(listener) { +StatsSocketListener::StatsSocketListener(std::shared_ptr 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(msg), &oldestTimestamp)) { + StatsdStats::getInstance().noteEventQueueOverflow(oldestTimestamp); + } return true; } diff --git a/cmds/statsd/src/socket/StatsSocketListener.h b/cmds/statsd/src/socket/StatsSocketListener.h index b8185d2a32d22..2167a56445b9c 100644 --- a/cmds/statsd/src/socket/StatsSocketListener.h +++ b/cmds/statsd/src/socket/StatsSocketListener.h @@ -17,7 +17,7 @@ #include #include -#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& listener); + explicit StatsSocketListener(std::shared_ptr queue); virtual ~StatsSocketListener(); @@ -47,7 +47,7 @@ private: /** * Who is going to get the events when they're read. */ - sp mListener; + std::shared_ptr mQueue; }; } // namespace statsd } // namespace os diff --git a/cmds/statsd/src/stats_log.proto b/cmds/statsd/src/stats_log.proto index 967c352562f74..1dfc433cf0e2d 100644 --- a/cmds/statsd/src/stats_log.proto +++ b/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 { diff --git a/cmds/statsd/tests/StatsService_test.cpp b/cmds/statsd/tests/StatsService_test.cpp index 560fb9f021742..7c00531d560b1 100644 --- a/cmds/statsd/tests/StatsService_test.cpp +++ b/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" diff --git a/cmds/statsd/tests/e2e/PartialBucket_e2e_test.cpp b/cmds/statsd/tests/e2e/PartialBucket_e2e_test.cpp index 3dff7f5d1320e..309d251e4a890 100644 --- a/cmds/statsd/tests/e2e/PartialBucket_e2e_test.cpp +++ b/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 */)); diff --git a/cmds/statsd/tests/log_event/LogEventQueue_test.cpp b/cmds/statsd/tests/log_event/LogEventQueue_test.cpp new file mode 100644 index 0000000000000..f27d12957f111 --- /dev/null +++ b/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 +#include +#include + +#include + +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(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(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