From 1d7b0cd6b54e3b3ec8efdffd47b40cdd54c4e8d7 Mon Sep 17 00:00:00 2001 From: David Chen Date: Wed, 15 Nov 2017 14:20:04 -0800 Subject: [PATCH] Support StatsD sending broadcasts. StatsD will send a broadcast when we're 90% of the way to our allocated memory limit for the configuration. If the memory usage goes over the limit, we just lose all the data for this config. Also modifies the adb shell commands to facilitate debugging of the broadcasts. Test: Manually tested on marlin-eng with custom gmscore code. Change-Id: I517a15bd4c959aa221802f84a51f13141a725102 --- cmds/statsd/src/StatsLogProcessor.cpp | 88 ++++++++------- cmds/statsd/src/StatsLogProcessor.h | 24 +++-- cmds/statsd/src/StatsService.cpp | 102 +++++++++++++++--- cmds/statsd/src/StatsService.h | 7 +- cmds/statsd/src/config/ConfigManager.cpp | 22 ++++ cmds/statsd/src/config/ConfigManager.h | 10 ++ .../src/metrics/EventMetricProducer.cpp | 5 +- cmds/statsd/src/metrics/MetricProducer.h | 3 + cmds/statsd/src/metrics/MetricsManager.h | 2 + 9 files changed, 193 insertions(+), 70 deletions(-) diff --git a/cmds/statsd/src/StatsLogProcessor.cpp b/cmds/statsd/src/StatsLogProcessor.cpp index b764ce56767b2..abd2a35d76dfd 100644 --- a/cmds/statsd/src/StatsLogProcessor.cpp +++ b/cmds/statsd/src/StatsLogProcessor.cpp @@ -50,8 +50,8 @@ const int FIELD_ID_UID = 1; const int FIELD_ID_NAME = 2; StatsLogProcessor::StatsLogProcessor(const sp& uidMap, - const std::function&)>& pushLog) - : mUidMap(uidMap), mPushLog(pushLog) { + const std::function& sendBroadcast) + : mUidMap(uidMap), mSendBroadcast(sendBroadcast) { } StatsLogProcessor::~StatsLogProcessor() { @@ -102,12 +102,27 @@ void StatsLogProcessor::OnConfigUpdated(const ConfigKey& key, const StatsdConfig } } -vector StatsLogProcessor::onDumpReport(const ConfigKey& key) { +size_t StatsLogProcessor::GetMetricsSize(const ConfigKey& key) { auto it = mMetricsManagers.find(key); if (it == mMetricsManagers.end()) { ALOGW("Config source %s does not exist", key.ToString().c_str()); - return vector(); + return 0; } + return it->second->byteSize(); +} + +void StatsLogProcessor::onDumpReport(const ConfigKey& key, vector* outData) { + auto it = mMetricsManagers.find(key); + if (it == mMetricsManagers.end()) { + ALOGW("Config source %s does not exist", key.ToString().c_str()); + return; + } + + // This allows another broadcast to be sent within the rate-limit period if we get close to + // filling the buffer again soon. + mBroadcastTimesMutex.lock(); + mLastBroadcastTimes.erase(key); + mBroadcastTimesMutex.unlock(); ProtoOutputStream proto; @@ -131,17 +146,18 @@ vector StatsLogProcessor::onDumpReport(const ConfigKey& key) { uidMap.SerializeToArray(&uidMapBuffer[0], uidMapSize); proto.write(FIELD_TYPE_MESSAGE | FIELD_ID_UID_MAP, uidMapBuffer, uidMapSize); - vector buffer(proto.size()); - size_t pos = 0; - auto iter = proto.data(); - while (iter.readBuffer() != NULL) { - size_t toRead = iter.currentToRead(); - std::memcpy(&buffer[pos], iter.readBuffer(), toRead); - pos += toRead; - iter.rp()->move(toRead); + if (outData != nullptr) { + outData->clear(); + outData->resize(proto.size()); + size_t pos = 0; + auto iter = proto.data(); + while (iter.readBuffer() != NULL) { + size_t toRead = iter.currentToRead(); + std::memcpy(&((*outData)[pos]), iter.readBuffer(), toRead); + pos += toRead; + iter.rp()->move(toRead); + } } - - return buffer; } void StatsLogProcessor::OnConfigRemoved(const ConfigKey& key) { @@ -151,42 +167,34 @@ void StatsLogProcessor::OnConfigRemoved(const ConfigKey& key) { mMetricsManagers.erase(it); mUidMap->OnConfigRemoved(key); } - auto flushTime = mLastFlushTimes.find(key); - if (flushTime != mLastFlushTimes.end()) { - mLastFlushTimes.erase(flushTime); - } + + std::lock_guard lock(mBroadcastTimesMutex); + mLastBroadcastTimes.erase(key); } void StatsLogProcessor::flushIfNecessary(uint64_t timestampNs, const ConfigKey& key, const unique_ptr& metricsManager) { - auto lastFlushNs = mLastFlushTimes.find(key); - if (lastFlushNs != mLastFlushTimes.end()) { - if (timestampNs - lastFlushNs->second < kMinFlushPeriod) { - return; - } - } + std::lock_guard lock(mBroadcastTimesMutex); size_t totalBytes = metricsManager->byteSize(); - if (totalBytes > kMaxSerializedBytes) { - flush(); - mLastFlushTimes[key] = std::move(timestampNs); + if (totalBytes > .9 * kMaxSerializedBytes) { // Send broadcast so that receivers can pull data. + auto lastFlushNs = mLastBroadcastTimes.find(key); + if (lastFlushNs != mLastBroadcastTimes.end()) { + if (timestampNs - lastFlushNs->second < kMinBroadcastPeriod) { + return; + } + } + mLastBroadcastTimes[key] = timestampNs; + ALOGD("StatsD requesting broadcast for %s", key.ToString().c_str()); + mSendBroadcast(key); + } else if (totalBytes > kMaxSerializedBytes) { // Too late. We need to start clearing data. + // We ignore the return value so we force each metric producer to clear its contents. + metricsManager->onDumpReport(); + ALOGD("StatsD had to toss out metrics for %s", key.ToString().c_str()); } } -void StatsLogProcessor::flush() { - // TODO: Take ConfigKey as an argument and flush metrics related to the - // ConfigKey. Also, create a wrapper that holds a repeated field of - // StatsLogReport's. - /* - StatsLogReport logReport; - const int numBytes = logReport.ByteSize(); - vector logReportBuffer(numBytes); - logReport.SerializeToArray(&logReportBuffer[0], numBytes); - mPushLog(logReportBuffer); - */ -} - } // namespace statsd } // namespace os } // namespace android diff --git a/cmds/statsd/src/StatsLogProcessor.h b/cmds/statsd/src/StatsLogProcessor.h index f38d71550eda7..2091774b108ee 100644 --- a/cmds/statsd/src/StatsLogProcessor.h +++ b/cmds/statsd/src/StatsLogProcessor.h @@ -33,7 +33,7 @@ namespace statsd { class StatsLogProcessor : public ConfigListener { public: StatsLogProcessor(const sp& uidMap, - const std::function&)>& pushLog); + const std::function& sendBroadcast); virtual ~StatsLogProcessor(); virtual void OnLogEvent(const LogEvent& event); @@ -41,15 +41,16 @@ public: void OnConfigUpdated(const ConfigKey& key, const StatsdConfig& config); void OnConfigRemoved(const ConfigKey& key); - vector onDumpReport(const ConfigKey& key); + size_t GetMetricsSize(const ConfigKey& key); - /* Request a flush through a binder call. */ - void flush(); + void onDumpReport(const ConfigKey& key, vector* outData); private: + mutable mutex mBroadcastTimesMutex; + std::unordered_map> mMetricsManagers; - std::unordered_map mLastFlushTimes; + std::unordered_map mLastBroadcastTimes; sp mUidMap; // Reference to the UidMap to lookup app name and version for each uid. @@ -60,17 +61,18 @@ private: */ static const size_t kMaxSerializedBytes = 16 * 1024; - /* Check if the buffer size exceeds the max buffer size when the new entry is added, and flush - the logs to callback clients if true. */ + /* Check if we should send a broadcast if approaching memory limits and if we're over, we + * actually delete the data. */ void flushIfNecessary(uint64_t timestampNs, const ConfigKey& key, const unique_ptr& metricsManager); - std::function&)> mPushLog; + // Function used to send a broadcast so that receiver for the config key can call getData + // to retrieve the stored data. + std::function mSendBroadcast; - /* Minimum period between two flushes in nanoseconds. Currently set to 10 - * minutes. */ - static const unsigned long long kMinFlushPeriod = 600 * NS_PER_SEC; + /* Minimum period between two broadcasts in nanoseconds. Currently set to 60 seconds. */ + static const unsigned long long kMinBroadcastPeriod = 60 * NS_PER_SEC; }; } // namespace statsd diff --git a/cmds/statsd/src/StatsService.cpp b/cmds/statsd/src/StatsService.cpp index 8bc776f7bfc2b..779f60f3cab59 100644 --- a/cmds/statsd/src/StatsService.cpp +++ b/cmds/statsd/src/StatsService.cpp @@ -73,8 +73,18 @@ StatsService::StatsService(const sp& handlerLooper) { mUidMap = new UidMap(); mConfigManager = new ConfigManager(); - mProcessor = new StatsLogProcessor(mUidMap, [](const vector& log) { - // TODO: Update how we send data out of StatsD. + mProcessor = new StatsLogProcessor(mUidMap, [this](const ConfigKey& key) { + auto sc = getStatsCompanionService(); + auto receiver = mConfigManager->GetConfigReceiver(key); + if (sc == nullptr) { + ALOGD("Could not find StatsCompanionService"); + } else if (receiver.first.size() == 0) { + ALOGD("Statscompanion could not find a broadcast receiver for %s", + key.ToString().c_str()); + } else { + sc->sendBroadcast(String16(receiver.first.c_str()), + String16(receiver.second.c_str())); + } }); mConfigManager->AddListener(mProcessor); @@ -206,7 +216,11 @@ status_t StatsService::command(FILE* in, FILE* out, FILE* err, Vector& } if (!args[0].compare(String8("send-broadcast"))) { - return cmd_trigger_broadcast(args); + return cmd_trigger_broadcast(out, args); + } + + if (!args[0].compare(String8("print-stats"))) { + return cmd_print_stats(out); } if (!args[0].compare(String8("clear-config"))) { @@ -259,16 +273,56 @@ void StatsService::print_cmd_help(FILE* out) { fprintf(out, " NAME The name of the configuration\n"); fprintf(out, "\n"); fprintf(out, "\n"); - fprintf(out, "usage: adb shell cmd stats send-broadcast PACKAGE CLASS\n"); - fprintf(out, " Send a broadcast that triggers one subscriber to fetch metrics.\n"); - fprintf(out, " PACKAGE The name of the package to receive the broadcast.\n"); - fprintf(out, " CLASS The name of the class to receive the broadcast.\n"); + fprintf(out, "usage: adb shell cmd stats send-broadcast [UID] NAME\n"); + fprintf(out, " Send a broadcast that triggers the subscriber to fetch metrics.\n"); + fprintf(out, " UID The uid of the configuration. It is only possible to pass\n"); + fprintf(out, " the UID parameter on eng builds. If UID is omitted the\n"); + fprintf(out, " calling uid is used.\n"); + fprintf(out, " NAME The name of the configuration\n"); + fprintf(out, "\n"); + fprintf(out, "\n"); + fprintf(out, "usage: adb shell cmd stats print-stats\n"); + fprintf(out, " Prints some basic stats.\n"); } -status_t StatsService::cmd_trigger_broadcast(Vector& args) { +status_t StatsService::cmd_trigger_broadcast(FILE* out, Vector& args) { + string name; + bool good = false; + int uid; + const int argCount = args.size(); + if (argCount == 2) { + // Automatically pick the UID + uid = IPCThreadState::self()->getCallingUid(); + // TODO: What if this isn't a binder call? Should we fail? + name.assign(args[1].c_str(), args[1].size()); + good = true; + } else if (argCount == 3) { + // If it's a userdebug or eng build, then the shell user can + // impersonate other uids. + if (mEngBuild) { + const char* s = args[1].c_str(); + if (*s != '\0') { + char* end = NULL; + uid = strtol(s, &end, 0); + if (*end == '\0') { + name.assign(args[2].c_str(), args[2].size()); + good = true; + } + } + } else { + fprintf(out, + "The metrics can only be dumped for other UIDs on eng or userdebug " + "builds.\n"); + } + } + if (!good) { + print_cmd_help(out); + return UNKNOWN_ERROR; + } + auto receiver = mConfigManager->GetConfigReceiver(ConfigKey(uid, name)); auto sc = getStatsCompanionService(); - sc->sendBroadcast(String16(args[1]), String16(args[2])); - ALOGD("StatsService::trigger broadcast succeeded"); + sc->sendBroadcast(String16(receiver.first.c_str()), String16(receiver.second.c_str())); + ALOGD("StatsService::trigger broadcast succeeded to %s, %s", args[1].c_str(), args[2].c_str()); return NO_ERROR; } @@ -373,7 +427,8 @@ status_t StatsService::cmd_dump_report(FILE* out, FILE* err, const VectoronDumpReport(ConfigKey(uid, name)); + vector data; + mProcessor->onDumpReport(ConfigKey(uid, name), &data); // TODO: print the returned StatsLogReport to file instead of printing to logcat. fprintf(out, "Dump report for Config [%d,%s]\n", uid, name.c_str()); fprintf(out, "See the StatsLogReport in logcat...\n"); @@ -389,6 +444,15 @@ status_t StatsService::cmd_dump_report(FILE* out, FILE* err, const Vector configs = mConfigManager->GetAllConfigKeys(); + for (const ConfigKey& key : configs) { + fprintf(out, "Config %s uses %zu bytes\n", key.ToString().c_str(), + mProcessor->GetMetricsSize(key)); + } + return NO_ERROR; +} + status_t StatsService::cmd_print_stats_log(FILE* out, const Vector& args) { long msec = 0; @@ -573,10 +637,14 @@ void StatsService::OnLogEvent(const LogEvent& event) { mProcessor->OnLogEvent(event); } -Status StatsService::getData(const String16& key, vector* output) { +Status StatsService::getData(const String16& key, vector * output) { IPCThreadState* ipc = IPCThreadState::self(); + ALOGD("StatsService::getData with Pid %i, Uid %i", ipc->getCallingPid(), + ipc->getCallingUid()); if (checkCallingPermission(String16(kPermissionDump))) { - // TODO: Implement this. + string keyStr = string(String8(key).string()); + ConfigKey configKey(ipc->getCallingUid(), keyStr); + mProcessor->onDumpReport(configKey, output); return Status::ok(); } else { return Status::fromExceptionCode(binder::Status::EX_SECURITY); @@ -588,10 +656,9 @@ Status StatsService::addConfiguration(const String16& key, const String16& package, const String16& cls, bool* success) { IPCThreadState* ipc = IPCThreadState::self(); - int32_t* uid = reinterpret_cast(ipc->getCallingUid()); if (checkCallingPermission(String16(kPermissionDump))) { string keyString = string(String8(key).string()); - ConfigKey configKey(*uid, keyString); + ConfigKey configKey(ipc->getCallingUid(), keyString); StatsdConfig cfg; cfg.ParseFromArray(&config[0], config.size()); mConfigManager->UpdateConfig(configKey, cfg); @@ -607,7 +674,8 @@ Status StatsService::addConfiguration(const String16& key, Status StatsService::removeConfiguration(const String16& key, bool* success) { IPCThreadState* ipc = IPCThreadState::self(); if (checkCallingPermission(String16(kPermissionDump))) { - // TODO: Implement this. + string keyStr = string(String8(key).string()); + mConfigManager->RemoveConfig(ConfigKey(ipc->getCallingUid(), keyStr)); return Status::ok(); } else { *success = false; @@ -615,7 +683,7 @@ Status StatsService::removeConfiguration(const String16& key, bool* success) { } } -void StatsService::binderDied(const wp& who) { +void StatsService::binderDied(const wp & who) { for (size_t i = 0; i < mCallbacks.size(); i++) { if (IInterface::asBinder(mCallbacks[i]) == who) { mCallbacks.removeAt(i); diff --git a/cmds/statsd/src/StatsService.h b/cmds/statsd/src/StatsService.h index 0163f946448cb..888f97b6e954e 100644 --- a/cmds/statsd/src/StatsService.h +++ b/cmds/statsd/src/StatsService.h @@ -124,13 +124,18 @@ private: /** * Trigger a broadcast. */ - status_t cmd_trigger_broadcast(Vector& args); + status_t cmd_trigger_broadcast(FILE* out, Vector& args); /** * Handle the config sub-command. */ status_t cmd_config(FILE* in, FILE* out, FILE* err, Vector& args); + /** + * Prints some basic stats to std out. + */ + status_t cmd_print_stats(FILE* out); + /** * Print the event log. */ diff --git a/cmds/statsd/src/config/ConfigManager.cpp b/cmds/statsd/src/config/ConfigManager.cpp index 3319f6d47d65d..9680e4ac81091 100644 --- a/cmds/statsd/src/config/ConfigManager.cpp +++ b/cmds/statsd/src/config/ConfigManager.cpp @@ -134,12 +134,34 @@ void ConfigManager::RemoveConfigs(int uid) { } } +vector ConfigManager::GetAllConfigKeys() { + vector ret; + for (auto it = mConfigs.cbegin(); it != mConfigs.cend(); ++it) { + ret.push_back(it->first); + } + return ret; +} + +const pair ConfigManager::GetConfigReceiver(const ConfigKey& key) { + auto it = mConfigReceivers.find(key); + if (it == mConfigReceivers.end()) { + return pair(); + } else { + return it->second; + } +} + void ConfigManager::Dump(FILE* out) { fprintf(out, "CONFIGURATIONS (%d)\n", (int)mConfigs.size()); fprintf(out, " uid name\n"); for (unordered_map::const_iterator it = mConfigs.begin(); it != mConfigs.end(); it++) { fprintf(out, " %6d %s\n", it->first.GetUid(), it->first.GetName().c_str()); + auto receiverIt = mConfigReceivers.find(it->first); + if (receiverIt != mConfigReceivers.end()) { + fprintf(out, " -> received by %s, %s\n", receiverIt->second.first.c_str(), + receiverIt->second.second.c_str()); + } // TODO: Print the contents of the config too. } } diff --git a/cmds/statsd/src/config/ConfigManager.h b/cmds/statsd/src/config/ConfigManager.h index 5b612cc5fb795..01d7fb969230e 100644 --- a/cmds/statsd/src/config/ConfigManager.h +++ b/cmds/statsd/src/config/ConfigManager.h @@ -67,6 +67,16 @@ public: */ void SetConfigReceiver(const ConfigKey& key, const string& pkg, const string& cls); + /** + * Returns the package name and class name representing the broadcast receiver for this config. + */ + const pair GetConfigReceiver(const ConfigKey& key); + + /** + * Returns all config keys registered. + */ + vector GetAllConfigKeys(); + /** * Erase any broadcast receiver associated with this config key. */ diff --git a/cmds/statsd/src/metrics/EventMetricProducer.cpp b/cmds/statsd/src/metrics/EventMetricProducer.cpp index 74ba40ba20aed..bd288a16fbe31 100644 --- a/cmds/statsd/src/metrics/EventMetricProducer.cpp +++ b/cmds/statsd/src/metrics/EventMetricProducer.cpp @@ -121,7 +121,10 @@ void EventMetricProducer::onMatchedLogEventInternal( event.ToProto(*mProto); mProto->end(eventToken); mProto->end(wrapperToken); - // TODO: Find a proper way to derive the size of incoming LogEvent. + + // TODO: Increment mByteSize with a real value. Until this feature is working, we assume 50 + // bytes. + mByteSize += 50; } size_t EventMetricProducer::byteSize() { diff --git a/cmds/statsd/src/metrics/MetricProducer.h b/cmds/statsd/src/metrics/MetricProducer.h index c0930e381ce3a..c7982a837c140 100644 --- a/cmds/statsd/src/metrics/MetricProducer.h +++ b/cmds/statsd/src/metrics/MetricProducer.h @@ -61,12 +61,15 @@ public: // TODO: Pass a timestamp as a parameter in onDumpReport and update all its // implementations. + // onDumpReport returns the proto-serialized output and clears the previously stored contents. virtual std::unique_ptr> onDumpReport() = 0; virtual bool isConditionSliced() const { return mConditionSliced; }; + // Returns the memory in bytes currently used to store this metric's data. Does not change + // state. virtual size_t byteSize() = 0; protected: diff --git a/cmds/statsd/src/metrics/MetricsManager.h b/cmds/statsd/src/metrics/MetricsManager.h index 39c79f99dacaf..fb167798a44fb 100644 --- a/cmds/statsd/src/metrics/MetricsManager.h +++ b/cmds/statsd/src/metrics/MetricsManager.h @@ -46,6 +46,8 @@ public: // Config source owner can call onDumpReport() to get all the metrics collected. std::vector>> onDumpReport(); + // Computes the total byte size of all metrics managed by a single config source. + // Does not change the state. size_t byteSize(); private: