1 files changed, 94 insertions, 51 deletions

diff --git a/cmds/statsd/src/StatsService.cpp b/cmds/statsd/src/StatsService.cpp
index f78ae38aabd8..3bcebd9e3f76 100644
--- a/cmds/statsd/src/StatsService.cpp
+++ b/cmds/statsd/src/StatsService.cpp
@@ -50,6 +50,7 @@ using android::base::StringPrintf;
 using android::util::FIELD_COUNT_REPEATED;
 using android::util::FIELD_TYPE_INT64;
 using android::util::FIELD_TYPE_MESSAGE;
+using android::util::ProtoReader;
 
 namespace android {
 namespace os {
@@ -64,8 +65,6 @@ constexpr const char* kOpUsage = "android:get_usage_stats";
 
 // for StatsDataDumpProto
 const int FIELD_ID_REPORTS_LIST = 1;
-// for TrainInfo experiment id serialization
-const int FIELD_ID_EXPERIMENT_ID = 1;
 
 static binder::Status ok() {
     return binder::Status::ok();
@@ -131,34 +130,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);
@@ -200,11 +201,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");
@@ -1008,6 +1031,7 @@ void StatsService::Terminate() {
     }
 }
 
+// Test only interface!!!
 void StatsService::OnLogEvent(LogEvent* event) {
     mProcessor->OnLogEvent(event);
     if (mShellSubscriber != nullptr) {
@@ -1180,7 +1204,7 @@ Status StatsService::unregisterPullerCallback(int32_t atomTag, const String16& p
 Status StatsService::sendBinaryPushStateChangedAtom(const android::String16& trainName,
                                                     int64_t trainVersionCode, int options,
                                                     int32_t state,
-                                                    const std::vector<int64_t>& experimentIds) {
+                                                    const std::vector<int64_t>& experimentIdsIn) {
     uid_t uid = IPCThreadState::self()->getCallingUid();
     // For testing
     if (uid == AID_ROOT || uid == AID_SYSTEM || uid == AID_SHELL) {
@@ -1200,7 +1224,7 @@ Status StatsService::sendBinaryPushStateChangedAtom(const android::String16& tra
 
     bool readTrainInfoSuccess = false;
     InstallTrainInfo trainInfo;
-    if (trainVersionCode == -1 || experimentIds.empty() || trainName.size() == 0) {
+    if (trainVersionCode == -1 || experimentIdsIn.empty() || trainName.size() == 0) {
         readTrainInfoSuccess = StorageManager::readTrainInfo(trainInfo);
     }
 
@@ -1208,27 +1232,19 @@ Status StatsService::sendBinaryPushStateChangedAtom(const android::String16& tra
         trainVersionCode = trainInfo.trainVersionCode;
     }
 
-    vector<uint8_t> experimentIdsProtoBuffer;
-    if (readTrainInfoSuccess && experimentIds.empty()) {
-        experimentIdsProtoBuffer = trainInfo.experimentIds;
+    // Find the right experiment IDs
+    std::vector<int64_t> experimentIds;
+    if (readTrainInfoSuccess && experimentIdsIn.empty()) {
+        experimentIds = trainInfo.experimentIds;
     } else {
-        ProtoOutputStream proto;
-        for (const auto& expId : experimentIds) {
-            proto.write(FIELD_TYPE_INT64 | FIELD_COUNT_REPEATED | FIELD_ID_EXPERIMENT_ID,
-                        (long long)expId);
-        }
-
-        experimentIdsProtoBuffer.resize(proto.size());
-        size_t pos = 0;
-        auto iter = proto.data();
-        while (iter.readBuffer() != NULL) {
-            size_t toRead = iter.currentToRead();
-            std::memcpy(&(experimentIdsProtoBuffer[pos]), iter.readBuffer(), toRead);
-            pos += toRead;
-            iter.rp()->move(toRead);
-        }
+        experimentIds = experimentIdsIn;
     }
 
+    // Flatten the experiment IDs to proto
+    vector<uint8_t> experimentIdsProtoBuffer;
+    writeExperimentIdsToProto(experimentIds, &experimentIdsProtoBuffer);
+
+    // Find the right train name
     std::string trainNameUtf8;
     if (readTrainInfoSuccess && trainName.size() == 0) {
         trainNameUtf8 = trainInfo.trainName;
@@ -1243,7 +1259,34 @@ Status StatsService::sendBinaryPushStateChangedAtom(const android::String16& tra
     LogEvent event(trainNameUtf8, trainVersionCode, requiresStaging, rollbackEnabled,
                    requiresLowLatencyMonitor, state, experimentIdsProtoBuffer, userId);
     mProcessor->OnLogEvent(&event);
-    StorageManager::writeTrainInfo(trainVersionCode, trainNameUtf8, state, experimentIdsProtoBuffer);
+    StorageManager::writeTrainInfo(trainVersionCode, trainNameUtf8, state, experimentIds);
+    return Status::ok();
+}
+
+Status StatsService::getRegisteredExperimentIds(std::vector<int64_t>* experimentIdsOut) {
+    uid_t uid = IPCThreadState::self()->getCallingUid();
+
+    // Caller must be granted these permissions
+    if (!checkCallingPermission(String16(kPermissionDump))) {
+        return exception(binder::Status::EX_SECURITY,
+                         StringPrintf("UID %d lacks permission %s", uid, kPermissionDump));
+    }
+    if (!checkCallingPermission(String16(kPermissionUsage))) {
+        return exception(binder::Status::EX_SECURITY,
+                         StringPrintf("UID %d lacks permission %s", uid, kPermissionUsage));
+    }
+    // TODO: add verifier permission
+
+    // Read the latest train info
+    InstallTrainInfo trainInfo;
+    if (!StorageManager::readTrainInfo(trainInfo)) {
+        // No train info means no experiment IDs, return an empty list
+        experimentIdsOut->clear();
+        return Status::ok();
+    }
+
+    // Copy the experiment IDs to the out vector
+    experimentIdsOut->assign(trainInfo.experimentIds.begin(), trainInfo.experimentIds.end());
     return Status::ok();
 }