Merge "Statsd: pull once per event time" into rvc-dev

This commit is contained in:
Tej Singh
2020-05-15 02:59:26 +00:00
committed by Android (Google) Code Review
15 changed files with 544 additions and 366 deletions

View File

@@ -826,7 +826,7 @@ status_t StatsService::cmd_print_pulled_metrics(int out, const Vector<String8>&
uids.push_back(AID_SYSTEM);
}
vector<shared_ptr<LogEvent>> stats;
if (mPullerManager->Pull(s, uids, &stats)) {
if (mPullerManager->Pull(s, uids, getElapsedRealtimeNs(), &stats)) {
for (const auto& it : stats) {
dprintf(out, "Pull from %d: %s\n", s, it->ToString().c_str());
}

View File

@@ -38,14 +38,16 @@ StatsPuller::StatsPuller(const int tagId, const int64_t coolDownNs, const int64_
mPullTimeoutNs(pullTimeoutNs),
mCoolDownNs(coolDownNs),
mAdditiveFields(additiveFields),
mLastPullTimeNs(0) {
mLastPullTimeNs(0),
mLastEventTimeNs(0) {
}
bool StatsPuller::Pull(std::vector<std::shared_ptr<LogEvent>>* data) {
bool StatsPuller::Pull(const int64_t eventTimeNs, std::vector<std::shared_ptr<LogEvent>>* data) {
lock_guard<std::mutex> lock(mLock);
int64_t elapsedTimeNs = getElapsedRealtimeNs();
StatsdStats::getInstance().notePull(mTagId);
const bool shouldUseCache = elapsedTimeNs - mLastPullTimeNs < mCoolDownNs;
const bool shouldUseCache =
(mLastEventTimeNs == eventTimeNs) || (elapsedTimeNs - mLastPullTimeNs < mCoolDownNs);
if (shouldUseCache) {
if (mHasGoodData) {
(*data) = mCachedData;
@@ -54,13 +56,13 @@ bool StatsPuller::Pull(std::vector<std::shared_ptr<LogEvent>>* data) {
}
return mHasGoodData;
}
if (mLastPullTimeNs > 0) {
StatsdStats::getInstance().updateMinPullIntervalSec(
mTagId, (elapsedTimeNs - mLastPullTimeNs) / NS_PER_SEC);
}
mCachedData.clear();
mLastPullTimeNs = elapsedTimeNs;
mLastEventTimeNs = eventTimeNs;
mHasGoodData = PullInternal(&mCachedData);
if (!mHasGoodData) {
return mHasGoodData;
@@ -70,7 +72,7 @@ bool StatsPuller::Pull(std::vector<std::shared_ptr<LogEvent>>* data) {
const bool pullTimeOut = pullDurationNs > mPullTimeoutNs;
if (pullTimeOut) {
// Something went wrong. Discard the data.
clearCacheLocked();
mCachedData.clear();
mHasGoodData = false;
StatsdStats::getInstance().notePullTimeout(mTagId);
ALOGW("Pull for atom %d exceeds timeout %lld nano seconds.", mTagId,
@@ -104,6 +106,7 @@ int StatsPuller::clearCacheLocked() {
int ret = mCachedData.size();
mCachedData.clear();
mLastPullTimeNs = 0;
mLastEventTimeNs = 0;
return ret;
}

View File

@@ -51,7 +51,7 @@ public:
// 2) pull takes longer than mPullTimeoutNs (intrinsic to puller)
// If a metric wants to make any change to the data, like timestamps, it
// should make a copy as this data may be shared with multiple metrics.
bool Pull(std::vector<std::shared_ptr<LogEvent>>* data);
bool Pull(const int64_t eventTimeNs, std::vector<std::shared_ptr<LogEvent>>* data);
// Clear cache immediately
int ForceClearCache();
@@ -94,6 +94,11 @@ private:
int64_t mLastPullTimeNs;
// All pulls happen due to an event (app upgrade, bucket boundary, condition change, etc).
// If multiple pulls need to be done at the same event time, we will always use the cache after
// the first pull.
int64_t mLastEventTimeNs;
// Cache of data from last pull. If next request comes before cool down finishes,
// cached data will be returned.
// Cached data is cleared when

View File

@@ -91,20 +91,21 @@ StatsPullerManager::StatsPullerManager()
mPullAtomCallbackDeathRecipient(AIBinder_DeathRecipient_new(pullAtomCallbackDied)) {
}
bool StatsPullerManager::Pull(int tagId, const ConfigKey& configKey,
bool StatsPullerManager::Pull(int tagId, const ConfigKey& configKey, const int64_t eventTimeNs,
vector<shared_ptr<LogEvent>>* data, bool useUids) {
std::lock_guard<std::mutex> _l(mLock);
return PullLocked(tagId, configKey, data, useUids);
return PullLocked(tagId, configKey, eventTimeNs, data, useUids);
}
bool StatsPullerManager::Pull(int tagId, const vector<int32_t>& uids,
bool StatsPullerManager::Pull(int tagId, const vector<int32_t>& uids, const int64_t eventTimeNs,
vector<std::shared_ptr<LogEvent>>* data, bool useUids) {
std::lock_guard<std::mutex> _l(mLock);
return PullLocked(tagId, uids, data, useUids);
return PullLocked(tagId, uids, eventTimeNs, data, useUids);
}
bool StatsPullerManager::PullLocked(int tagId, const ConfigKey& configKey,
vector<shared_ptr<LogEvent>>* data, bool useUids) {
const int64_t eventTimeNs, vector<shared_ptr<LogEvent>>* data,
bool useUids) {
vector<int32_t> uids;
if (useUids) {
auto uidProviderIt = mPullUidProviders.find(configKey);
@@ -123,18 +124,19 @@ bool StatsPullerManager::PullLocked(int tagId, const ConfigKey& configKey,
}
uids = pullUidProvider->getPullAtomUids(tagId);
}
return PullLocked(tagId, uids, data, useUids);
return PullLocked(tagId, uids, eventTimeNs, data, useUids);
}
bool StatsPullerManager::PullLocked(int tagId, const vector<int32_t>& uids,
vector<shared_ptr<LogEvent>>* data, bool useUids) {
const int64_t eventTimeNs, vector<shared_ptr<LogEvent>>* data,
bool useUids) {
VLOG("Initiating pulling %d", tagId);
if (useUids) {
for (int32_t uid : uids) {
PullerKey key = {.atomTag = tagId, .uid = uid};
auto pullerIt = kAllPullAtomInfo.find(key);
if (pullerIt != kAllPullAtomInfo.end()) {
bool ret = pullerIt->second->Pull(data);
bool ret = pullerIt->second->Pull(eventTimeNs, data);
VLOG("pulled %zu items", data->size());
if (!ret) {
StatsdStats::getInstance().notePullFailed(tagId);
@@ -149,7 +151,7 @@ bool StatsPullerManager::PullLocked(int tagId, const vector<int32_t>& uids,
PullerKey key = {.atomTag = tagId, .uid = -1};
auto pullerIt = kAllPullAtomInfo.find(key);
if (pullerIt != kAllPullAtomInfo.end()) {
bool ret = pullerIt->second->Pull(data);
bool ret = pullerIt->second->Pull(eventTimeNs, data);
VLOG("pulled %zu items", data->size());
if (!ret) {
StatsdStats::getInstance().notePullFailed(tagId);
@@ -290,7 +292,8 @@ void StatsPullerManager::OnAlarmFired(int64_t elapsedTimeNs) {
}
for (const auto& pullInfo : needToPull) {
vector<shared_ptr<LogEvent>> data;
bool pullSuccess = PullLocked(pullInfo.first->atomTag, pullInfo.first->configKey, &data);
bool pullSuccess = PullLocked(pullInfo.first->atomTag, pullInfo.first->configKey,
elapsedTimeNs, &data);
if (!pullSuccess) {
VLOG("pull failed at %lld, will try again later", (long long)elapsedTimeNs);
}

View File

@@ -101,11 +101,11 @@ public:
// registered for any of the uids for this atom.
// If the metric wants to make any change to the data, like timestamps, they
// should make a copy as this data may be shared with multiple metrics.
virtual bool Pull(int tagId, const ConfigKey& configKey,
virtual bool Pull(int tagId, const ConfigKey& configKey, const int64_t eventTimeNs,
vector<std::shared_ptr<LogEvent>>* data, bool useUids = true);
// Same as above, but directly specify the allowed uids to pull from.
virtual bool Pull(int tagId, const vector<int32_t>& uids,
virtual bool Pull(int tagId, const vector<int32_t>& uids, const int64_t eventTimeNs,
vector<std::shared_ptr<LogEvent>>* data, bool useUids = true);
// Clear pull data cache immediately.
@@ -152,11 +152,11 @@ private:
// mapping from Config Key to the PullUidProvider for that config
std::map<ConfigKey, wp<PullUidProvider>> mPullUidProviders;
bool PullLocked(int tagId, const ConfigKey& configKey, vector<std::shared_ptr<LogEvent>>* data,
bool useUids = true);
bool PullLocked(int tagId, const ConfigKey& configKey, const int64_t eventTimeNs,
vector<std::shared_ptr<LogEvent>>* data, bool useUids = true);
bool PullLocked(int tagId, const vector<int32_t>& uids, vector<std::shared_ptr<LogEvent>>* data,
bool useUids);
bool PullLocked(int tagId, const vector<int32_t>& uids, const int64_t eventTimeNs,
vector<std::shared_ptr<LogEvent>>* data, bool useUids);
// locks for data receiver and StatsCompanionService changes
std::mutex mLock;

View File

@@ -321,7 +321,7 @@ void GaugeMetricProducer::pullAndMatchEventsLocked(const int64_t timestampNs) {
return;
}
vector<std::shared_ptr<LogEvent>> allData;
if (!mPullerManager->Pull(mPullTagId, mConfigKey, &allData)) {
if (!mPullerManager->Pull(mPullTagId, mConfigKey, timestampNs, &allData)) {
ALOGE("Gauge Stats puller failed for tag: %d at %lld", mPullTagId, (long long)timestampNs);
return;
}

View File

@@ -508,7 +508,7 @@ void ValueMetricProducer::prepareFirstBucketLocked() {
void ValueMetricProducer::pullAndMatchEventsLocked(const int64_t timestampNs) {
vector<std::shared_ptr<LogEvent>> allData;
if (!mPullerManager->Pull(mPullTagId, mConfigKey, &allData)) {
if (!mPullerManager->Pull(mPullTagId, mConfigKey, timestampNs, &allData)) {
ALOGE("Stats puller failed for tag: %d at %lld", mPullTagId, (long long)timestampNs);
invalidateCurrentBucket(timestampNs, BucketDropReason::PULL_FAILED);
return;

View File

@@ -152,6 +152,7 @@ void ShellSubscriber::startPull(int myToken) {
}
int64_t nowMillis = getElapsedRealtimeMillis();
int64_t nowNanos = getElapsedRealtimeNs();
for (PullInfo& pullInfo : mSubscriptionInfo->mPulledInfo) {
if (pullInfo.mPrevPullElapsedRealtimeMs + pullInfo.mInterval >= nowMillis) {
continue;
@@ -161,7 +162,7 @@ void ShellSubscriber::startPull(int myToken) {
getUidsForPullAtom(&uids, pullInfo);
vector<std::shared_ptr<LogEvent>> data;
mPullerMgr->Pull(pullInfo.mPullerMatcher.atom_id(), uids, &data);
mPullerMgr->Pull(pullInfo.mPullerMatcher.atom_id(), uids, nowNanos, &data);
VLOG("Pulled %zu atoms with id %d", data.size(), pullInfo.mPullerMatcher.atom_id());
writePulledAtomsLocked(data, pullInfo.mPullerMatcher);

View File

@@ -155,7 +155,7 @@ TEST_F(StatsCallbackPullerTest, PullFail) {
TEST_F(StatsCallbackPullerTest, PullTimeout) {
shared_ptr<FakePullAtomCallback> cb = SharedRefBase::make<FakePullAtomCallback>();
pullSuccess = true;
pullDelayNs = 500000000; // 500ms.
pullDelayNs = MillisToNano(5); // 5ms.
pullTimeoutNs = 10000; // 10 microseconds.
int64_t value = 4321;
values.push_back(value);
@@ -184,7 +184,7 @@ TEST_F(StatsCallbackPullerTest, PullTimeout) {
TEST_F(StatsCallbackPullerTest, RegisterAndTimeout) {
shared_ptr<FakePullAtomCallback> cb = SharedRefBase::make<FakePullAtomCallback>();
pullSuccess = true;
pullDelayNs = 500000000; // 500 ms.
pullDelayNs = MillisToNano(5); // 5 ms.
pullTimeoutNs = 10000; // 10 microsseconds.
int64_t value = 4321;
int32_t uid = 123;
@@ -196,7 +196,7 @@ TEST_F(StatsCallbackPullerTest, RegisterAndTimeout) {
vector<shared_ptr<LogEvent>> dataHolder;
int64_t startTimeNs = getElapsedRealtimeNs();
// Returns false, since StatsPuller code will evaluate the timeout.
EXPECT_FALSE(pullerManager->Pull(pullTagId, {uid}, &dataHolder));
EXPECT_FALSE(pullerManager->Pull(pullTagId, {uid}, startTimeNs, &dataHolder));
int64_t endTimeNs = getElapsedRealtimeNs();
int64_t actualPullDurationNs = endTimeNs - startTimeNs;

View File

@@ -101,14 +101,14 @@ TEST(StatsPullerManagerTest, TestPullInvalidUid) {
sp<StatsPullerManager> pullerManager = createPullerManagerAndRegister();
vector<shared_ptr<LogEvent>> data;
EXPECT_FALSE(pullerManager->Pull(pullTagId1, {unregisteredUid}, &data, true));
EXPECT_FALSE(pullerManager->Pull(pullTagId1, {unregisteredUid}, /*timestamp =*/1, &data, true));
}
TEST(StatsPullerManagerTest, TestPullChoosesCorrectUid) {
sp<StatsPullerManager> pullerManager = createPullerManagerAndRegister();
vector<shared_ptr<LogEvent>> data;
EXPECT_TRUE(pullerManager->Pull(pullTagId1, {uid1}, &data, true));
EXPECT_TRUE(pullerManager->Pull(pullTagId1, {uid1}, /*timestamp =*/1, &data, true));
ASSERT_EQ(data.size(), 1);
EXPECT_EQ(data[0]->GetTagId(), pullTagId1);
ASSERT_EQ(data[0]->getValues().size(), 1);
@@ -121,7 +121,7 @@ TEST(StatsPullerManagerTest, TestPullInvalidConfigKey) {
pullerManager->RegisterPullUidProvider(configKey, uidProvider);
vector<shared_ptr<LogEvent>> data;
EXPECT_FALSE(pullerManager->Pull(pullTagId1, badConfigKey, &data, true));
EXPECT_FALSE(pullerManager->Pull(pullTagId1, badConfigKey, /*timestamp =*/1, &data, true));
}
TEST(StatsPullerManagerTest, TestPullConfigKeyGood) {
@@ -130,7 +130,7 @@ TEST(StatsPullerManagerTest, TestPullConfigKeyGood) {
pullerManager->RegisterPullUidProvider(configKey, uidProvider);
vector<shared_ptr<LogEvent>> data;
EXPECT_TRUE(pullerManager->Pull(pullTagId1, configKey, &data, true));
EXPECT_TRUE(pullerManager->Pull(pullTagId1, configKey, /*timestamp =*/1, &data, true));
EXPECT_EQ(data[0]->GetTagId(), pullTagId1);
ASSERT_EQ(data[0]->getValues().size(), 1);
EXPECT_EQ(data[0]->getValues()[0].mValue.int_value, uid2);
@@ -142,7 +142,7 @@ TEST(StatsPullerManagerTest, TestPullConfigKeyNoPullerWithUid) {
pullerManager->RegisterPullUidProvider(configKey, uidProvider);
vector<shared_ptr<LogEvent>> data;
EXPECT_FALSE(pullerManager->Pull(pullTagId2, configKey, &data, true));
EXPECT_FALSE(pullerManager->Pull(pullTagId2, configKey, /*timestamp =*/1, &data, true));
}
} // namespace statsd

View File

@@ -39,7 +39,6 @@ using std::this_thread::sleep_for;
using testing::Contains;
namespace {
// cooldown time 1sec.
int pullTagId = 10014;
bool pullSuccess;
@@ -48,7 +47,8 @@ long pullDelayNs;
class FakePuller : public StatsPuller {
public:
FakePuller() : StatsPuller(pullTagId, /*coolDown=*/NS_PER_SEC, /*timeout=*/NS_PER_SEC / 2){};
FakePuller()
: StatsPuller(pullTagId, /*coolDownNs=*/MillisToNano(10), /*timeoutNs=*/MillisToNano(5)){};
private:
bool PullInternal(vector<std::shared_ptr<LogEvent>>* data) override {
@@ -92,21 +92,21 @@ TEST_F(StatsPullerTest, PullSuccess) {
pullSuccess = true;
vector<std::shared_ptr<LogEvent>> dataHolder;
EXPECT_TRUE(puller.Pull(&dataHolder));
EXPECT_TRUE(puller.Pull(getElapsedRealtimeNs(), &dataHolder));
ASSERT_EQ(1, dataHolder.size());
EXPECT_EQ(pullTagId, dataHolder[0]->GetTagId());
EXPECT_EQ(1111L, dataHolder[0]->GetElapsedTimestampNs());
ASSERT_EQ(1, dataHolder[0]->size());
EXPECT_EQ(33, dataHolder[0]->getValues()[0].mValue.int_value);
sleep_for(std::chrono::seconds(1));
sleep_for(std::chrono::milliseconds(11));
pullData.clear();
pullData.push_back(createSimpleEvent(2222L, 44));
pullSuccess = true;
EXPECT_TRUE(puller.Pull(&dataHolder));
EXPECT_TRUE(puller.Pull(getElapsedRealtimeNs(), &dataHolder));
ASSERT_EQ(1, dataHolder.size());
EXPECT_EQ(pullTagId, dataHolder[0]->GetTagId());
EXPECT_EQ(2222L, dataHolder[0]->GetElapsedTimestampNs());
@@ -120,26 +120,27 @@ TEST_F(StatsPullerTest, PullFailAfterSuccess) {
pullSuccess = true;
vector<std::shared_ptr<LogEvent>> dataHolder;
EXPECT_TRUE(puller.Pull(&dataHolder));
EXPECT_TRUE(puller.Pull(getElapsedRealtimeNs(), &dataHolder));
ASSERT_EQ(1, dataHolder.size());
EXPECT_EQ(pullTagId, dataHolder[0]->GetTagId());
EXPECT_EQ(1111L, dataHolder[0]->GetElapsedTimestampNs());
ASSERT_EQ(1, dataHolder[0]->size());
EXPECT_EQ(33, dataHolder[0]->getValues()[0].mValue.int_value);
sleep_for(std::chrono::seconds(1));
sleep_for(std::chrono::milliseconds(11));
pullData.clear();
pullData.push_back(createSimpleEvent(2222L, 44));
pullSuccess = false;
dataHolder.clear();
EXPECT_FALSE(puller.Pull(&dataHolder));
EXPECT_FALSE(puller.Pull(getElapsedRealtimeNs(), &dataHolder));
ASSERT_EQ(0, dataHolder.size());
// Fails due to hitting the cool down.
pullSuccess = true;
dataHolder.clear();
EXPECT_FALSE(puller.Pull(&dataHolder));
EXPECT_FALSE(puller.Pull(getElapsedRealtimeNs(), &dataHolder));
ASSERT_EQ(0, dataHolder.size());
}
@@ -147,19 +148,20 @@ TEST_F(StatsPullerTest, PullFailAfterSuccess) {
TEST_F(StatsPullerTest, PullTakeTooLongAndPullFast) {
pullData.push_back(createSimpleEvent(1111L, 33));
pullSuccess = true;
// timeout is 0.5
pullDelayNs = (long)(0.8 * NS_PER_SEC);
// timeout is 5ms
pullDelayNs = MillisToNano(6);
vector<std::shared_ptr<LogEvent>> dataHolder;
EXPECT_FALSE(puller.Pull(&dataHolder));
EXPECT_FALSE(puller.Pull(getElapsedRealtimeNs(), &dataHolder));
ASSERT_EQ(0, dataHolder.size());
pullData.clear();
pullData.push_back(createSimpleEvent(2222L, 44));
pullDelayNs = 0;
pullSuccess = true;
dataHolder.clear();
EXPECT_FALSE(puller.Pull(&dataHolder));
EXPECT_FALSE(puller.Pull(getElapsedRealtimeNs(), &dataHolder));
ASSERT_EQ(0, dataHolder.size());
}
@@ -169,7 +171,7 @@ TEST_F(StatsPullerTest, PullFail) {
pullSuccess = false;
vector<std::shared_ptr<LogEvent>> dataHolder;
EXPECT_FALSE(puller.Pull(&dataHolder));
EXPECT_FALSE(puller.Pull(getElapsedRealtimeNs(), &dataHolder));
ASSERT_EQ(0, dataHolder.size());
}
@@ -177,10 +179,10 @@ TEST_F(StatsPullerTest, PullTakeTooLong) {
pullData.push_back(createSimpleEvent(1111L, 33));
pullSuccess = true;
pullDelayNs = NS_PER_SEC;
pullDelayNs = MillisToNano(6);
vector<std::shared_ptr<LogEvent>> dataHolder;
EXPECT_FALSE(puller.Pull(&dataHolder));
EXPECT_FALSE(puller.Pull(getElapsedRealtimeNs(), &dataHolder));
ASSERT_EQ(0, dataHolder.size());
}
@@ -190,7 +192,7 @@ TEST_F(StatsPullerTest, PullTooFast) {
pullSuccess = true;
vector<std::shared_ptr<LogEvent>> dataHolder;
EXPECT_TRUE(puller.Pull(&dataHolder));
EXPECT_TRUE(puller.Pull(getElapsedRealtimeNs(), &dataHolder));
ASSERT_EQ(1, dataHolder.size());
EXPECT_EQ(pullTagId, dataHolder[0]->GetTagId());
EXPECT_EQ(1111L, dataHolder[0]->GetElapsedTimestampNs());
@@ -203,7 +205,7 @@ TEST_F(StatsPullerTest, PullTooFast) {
pullSuccess = true;
dataHolder.clear();
EXPECT_TRUE(puller.Pull(&dataHolder));
EXPECT_TRUE(puller.Pull(getElapsedRealtimeNs(), &dataHolder));
ASSERT_EQ(1, dataHolder.size());
EXPECT_EQ(pullTagId, dataHolder[0]->GetTagId());
EXPECT_EQ(1111L, dataHolder[0]->GetElapsedTimestampNs());
@@ -217,7 +219,7 @@ TEST_F(StatsPullerTest, PullFailsAndTooFast) {
pullSuccess = false;
vector<std::shared_ptr<LogEvent>> dataHolder;
EXPECT_FALSE(puller.Pull(&dataHolder));
EXPECT_FALSE(puller.Pull(getElapsedRealtimeNs(), &dataHolder));
ASSERT_EQ(0, dataHolder.size());
pullData.clear();
@@ -225,7 +227,84 @@ TEST_F(StatsPullerTest, PullFailsAndTooFast) {
pullSuccess = true;
EXPECT_FALSE(puller.Pull(&dataHolder));
EXPECT_FALSE(puller.Pull(getElapsedRealtimeNs(), &dataHolder));
ASSERT_EQ(0, dataHolder.size());
}
TEST_F(StatsPullerTest, PullSameEventTime) {
pullData.push_back(createSimpleEvent(1111L, 33));
pullSuccess = true;
int64_t eventTimeNs = getElapsedRealtimeNs();
vector<std::shared_ptr<LogEvent>> dataHolder;
EXPECT_TRUE(puller.Pull(eventTimeNs, &dataHolder));
ASSERT_EQ(1, dataHolder.size());
EXPECT_EQ(pullTagId, dataHolder[0]->GetTagId());
EXPECT_EQ(1111L, dataHolder[0]->GetElapsedTimestampNs());
ASSERT_EQ(1, dataHolder[0]->size());
EXPECT_EQ(33, dataHolder[0]->getValues()[0].mValue.int_value);
pullData.clear();
pullData.push_back(createSimpleEvent(2222L, 44));
// Sleep to ensure the cool down expires.
sleep_for(std::chrono::milliseconds(11));
pullSuccess = true;
dataHolder.clear();
EXPECT_TRUE(puller.Pull(eventTimeNs, &dataHolder));
ASSERT_EQ(1, dataHolder.size());
EXPECT_EQ(pullTagId, dataHolder[0]->GetTagId());
EXPECT_EQ(1111L, dataHolder[0]->GetElapsedTimestampNs());
ASSERT_EQ(1, dataHolder[0]->size());
EXPECT_EQ(33, dataHolder[0]->getValues()[0].mValue.int_value);
}
// Test pull takes longer than timeout, 2nd pull happens at same event time
TEST_F(StatsPullerTest, PullTakeTooLongAndPullSameEventTime) {
pullData.push_back(createSimpleEvent(1111L, 33));
pullSuccess = true;
int64_t eventTimeNs = getElapsedRealtimeNs();
// timeout is 5ms
pullDelayNs = MillisToNano(6);
vector<std::shared_ptr<LogEvent>> dataHolder;
EXPECT_FALSE(puller.Pull(eventTimeNs, &dataHolder));
ASSERT_EQ(0, dataHolder.size());
// Sleep to ensure the cool down expires. 6ms is taken by the delay, so only 5 is needed here.
sleep_for(std::chrono::milliseconds(5));
pullData.clear();
pullData.push_back(createSimpleEvent(2222L, 44));
pullDelayNs = 0;
pullSuccess = true;
dataHolder.clear();
EXPECT_FALSE(puller.Pull(eventTimeNs, &dataHolder));
ASSERT_EQ(0, dataHolder.size());
}
TEST_F(StatsPullerTest, PullFailsAndPullSameEventTime) {
pullData.push_back(createSimpleEvent(1111L, 33));
pullSuccess = false;
int64_t eventTimeNs = getElapsedRealtimeNs();
vector<std::shared_ptr<LogEvent>> dataHolder;
EXPECT_FALSE(puller.Pull(eventTimeNs, &dataHolder));
ASSERT_EQ(0, dataHolder.size());
// Sleep to ensure the cool down expires.
sleep_for(std::chrono::milliseconds(11));
pullData.clear();
pullData.push_back(createSimpleEvent(2222L, 44));
pullSuccess = true;
EXPECT_FALSE(puller.Pull(eventTimeNs, &dataHolder));
ASSERT_EQ(0, dataHolder.size());
}

View File

@@ -138,11 +138,12 @@ TEST(GaugeMetricProducerTest, TestPulledEventsNoCondition) {
sp<MockStatsPullerManager> pullerManager = new StrictMock<MockStatsPullerManager>();
EXPECT_CALL(*pullerManager, RegisterReceiver(tagId, kConfigKey, _, _, _)).WillOnce(Return());
EXPECT_CALL(*pullerManager, UnRegisterReceiver(tagId, kConfigKey, _)).WillOnce(Return());
EXPECT_CALL(*pullerManager, Pull(tagId, kConfigKey, _, _))
.WillOnce(Invoke([](int tagId, const ConfigKey&,
EXPECT_CALL(*pullerManager, Pull(tagId, kConfigKey, bucketStartTimeNs, _, _))
.WillOnce(Invoke([](int tagId, const ConfigKey&, const int64_t eventTimeNs,
vector<std::shared_ptr<LogEvent>>* data, bool) {
EXPECT_EQ(eventTimeNs, bucketStartTimeNs);
data->clear();
data->push_back(makeLogEvent(tagId, bucketStartTimeNs + 10, 3, "some value", 11));
data->push_back(makeLogEvent(tagId, eventTimeNs + 10, 3, "some value", 11));
return true;
}));
@@ -311,15 +312,15 @@ TEST_P(GaugeMetricProducerTest_PartialBucket, TestPulled) {
sp<MockStatsPullerManager> pullerManager = new StrictMock<MockStatsPullerManager>();
EXPECT_CALL(*pullerManager, RegisterReceiver(tagId, kConfigKey, _, _, _)).WillOnce(Return());
EXPECT_CALL(*pullerManager, UnRegisterReceiver(tagId, kConfigKey, _)).WillOnce(Return());
EXPECT_CALL(*pullerManager, Pull(tagId, kConfigKey, _, _))
EXPECT_CALL(*pullerManager, Pull(tagId, kConfigKey, _, _, _))
.WillOnce(Return(false))
.WillOnce(Invoke(
[](int tagId, const ConfigKey&, vector<std::shared_ptr<LogEvent>>* data, bool) {
data->clear();
data->push_back(
CreateRepeatedValueLogEvent(tagId, partialBucketSplitTimeNs, 2));
return true;
}));
.WillOnce(Invoke([](int tagId, const ConfigKey&, const int64_t eventTimeNs,
vector<std::shared_ptr<LogEvent>>* data, bool) {
EXPECT_EQ(eventTimeNs, partialBucketSplitTimeNs);
data->clear();
data->push_back(CreateRepeatedValueLogEvent(tagId, eventTimeNs, 2));
return true;
}));
GaugeMetricProducer gaugeProducer(kConfigKey, metric, -1 /*-1 meaning no condition*/, wizard,
logEventMatcherIndex, eventMatcherWizard, tagId, -1, tagId,
@@ -389,7 +390,8 @@ TEST(GaugeMetricProducerTest, TestPulledWithAppUpgradeDisabled) {
sp<MockStatsPullerManager> pullerManager = new StrictMock<MockStatsPullerManager>();
EXPECT_CALL(*pullerManager, RegisterReceiver(tagId, kConfigKey, _, _, _)).WillOnce(Return());
EXPECT_CALL(*pullerManager, UnRegisterReceiver(tagId, kConfigKey, _)).WillOnce(Return());
EXPECT_CALL(*pullerManager, Pull(tagId, kConfigKey, _, _)).WillOnce(Return(false));
EXPECT_CALL(*pullerManager, Pull(tagId, kConfigKey, bucketStartTimeNs, _, _))
.WillOnce(Return(false));
GaugeMetricProducer gaugeProducer(kConfigKey, metric, -1 /*-1 meaning no condition*/, wizard,
logEventMatcherIndex, eventMatcherWizard, tagId, -1, tagId,
@@ -435,14 +437,16 @@ TEST(GaugeMetricProducerTest, TestPulledEventsWithCondition) {
new EventMatcherWizard({new SimpleLogMatchingTracker(
atomMatcherId, logEventMatcherIndex, atomMatcher, uidMap)});
int64_t conditionChangeNs = bucketStartTimeNs + 8;
sp<MockStatsPullerManager> pullerManager = new StrictMock<MockStatsPullerManager>();
EXPECT_CALL(*pullerManager, RegisterReceiver(tagId, kConfigKey, _, _, _)).WillOnce(Return());
EXPECT_CALL(*pullerManager, UnRegisterReceiver(tagId, kConfigKey, _)).WillOnce(Return());
EXPECT_CALL(*pullerManager, Pull(tagId, kConfigKey, _, _))
.WillOnce(Invoke([](int tagId, const ConfigKey&,
EXPECT_CALL(*pullerManager, Pull(tagId, kConfigKey, conditionChangeNs, _, _))
.WillOnce(Invoke([](int tagId, const ConfigKey&, const int64_t eventTimeNs,
vector<std::shared_ptr<LogEvent>>* data, bool) {
data->clear();
data->push_back(CreateRepeatedValueLogEvent(tagId, bucketStartTimeNs + 10, 100));
data->push_back(CreateRepeatedValueLogEvent(tagId, eventTimeNs + 10, 100));
return true;
}));
@@ -451,7 +455,7 @@ TEST(GaugeMetricProducerTest, TestPulledEventsWithCondition) {
bucketStartTimeNs, pullerManager);
gaugeProducer.prepareFirstBucket();
gaugeProducer.onConditionChanged(true, bucketStartTimeNs + 8);
gaugeProducer.onConditionChanged(true, conditionChangeNs);
ASSERT_EQ(1UL, gaugeProducer.mCurrentSlicedBucket->size());
EXPECT_EQ(100, gaugeProducer.mCurrentSlicedBucket->begin()
->second.front()
@@ -519,14 +523,16 @@ TEST(GaugeMetricProducerTest, TestPulledEventsWithSlicedCondition) {
return ConditionState::kTrue;
}));
int64_t sliceConditionChangeNs = bucketStartTimeNs + 8;
sp<MockStatsPullerManager> pullerManager = new StrictMock<MockStatsPullerManager>();
EXPECT_CALL(*pullerManager, RegisterReceiver(tagId, kConfigKey, _, _, _)).WillOnce(Return());
EXPECT_CALL(*pullerManager, UnRegisterReceiver(tagId, kConfigKey, _)).WillOnce(Return());
EXPECT_CALL(*pullerManager, Pull(tagId, kConfigKey, _, _))
.WillOnce(Invoke([](int tagId, const ConfigKey&,
EXPECT_CALL(*pullerManager, Pull(tagId, kConfigKey, sliceConditionChangeNs, _, _))
.WillOnce(Invoke([](int tagId, const ConfigKey&, const int64_t eventTimeNs,
vector<std::shared_ptr<LogEvent>>* data, bool) {
data->clear();
data->push_back(CreateTwoValueLogEvent(tagId, bucketStartTimeNs + 10, 1000, 100));
data->push_back(CreateTwoValueLogEvent(tagId, eventTimeNs + 10, 1000, 100));
return true;
}));
@@ -535,7 +541,7 @@ TEST(GaugeMetricProducerTest, TestPulledEventsWithSlicedCondition) {
bucketStartTimeNs, pullerManager);
gaugeProducer.prepareFirstBucket();
gaugeProducer.onSlicedConditionMayChange(true, bucketStartTimeNs + 8);
gaugeProducer.onSlicedConditionMayChange(true, sliceConditionChangeNs);
ASSERT_EQ(1UL, gaugeProducer.mCurrentSlicedBucket->size());
const auto& key = gaugeProducer.mCurrentSlicedBucket->begin()->first;
@@ -560,7 +566,8 @@ TEST(GaugeMetricProducerTest, TestPulledEventsAnomalyDetection) {
sp<MockStatsPullerManager> pullerManager = new StrictMock<MockStatsPullerManager>();
EXPECT_CALL(*pullerManager, RegisterReceiver(tagId, kConfigKey, _, _, _)).WillOnce(Return());
EXPECT_CALL(*pullerManager, UnRegisterReceiver(tagId, kConfigKey, _)).WillOnce(Return());
EXPECT_CALL(*pullerManager, Pull(tagId, kConfigKey, _, _)).WillOnce(Return(false));
EXPECT_CALL(*pullerManager, Pull(tagId, kConfigKey, bucketStartTimeNs, _, _))
.WillOnce(Return(false));
GaugeMetric metric;
metric.set_id(metricId);
@@ -658,17 +665,19 @@ TEST(GaugeMetricProducerTest, TestPullOnTrigger) {
atomMatcherId, logEventMatcherIndex, atomMatcher, uidMap)});
sp<MockStatsPullerManager> pullerManager = new StrictMock<MockStatsPullerManager>();
EXPECT_CALL(*pullerManager, Pull(tagId, kConfigKey, _, _))
.WillOnce(Invoke([](int tagId, const ConfigKey&,
EXPECT_CALL(*pullerManager, Pull(tagId, kConfigKey, _, _, _))
.WillOnce(Invoke([](int tagId, const ConfigKey&, const int64_t eventTimeNs,
vector<std::shared_ptr<LogEvent>>* data, bool) {
EXPECT_EQ(eventTimeNs, bucketStartTimeNs + 10);
data->clear();
data->push_back(CreateRepeatedValueLogEvent(tagId, bucketStartTimeNs + 10, 4));
data->push_back(CreateRepeatedValueLogEvent(tagId, eventTimeNs, 4));
return true;
}))
.WillOnce(Invoke([](int tagId, const ConfigKey&,
.WillOnce(Invoke([](int tagId, const ConfigKey&, const int64_t eventTimeNs,
vector<std::shared_ptr<LogEvent>>* data, bool) {
EXPECT_EQ(eventTimeNs, bucketStartTimeNs + 20);
data->clear();
data->push_back(CreateRepeatedValueLogEvent(tagId, bucketStartTimeNs + 20, 5));
data->push_back(CreateRepeatedValueLogEvent(tagId, eventTimeNs, 5));
return true;
}))
.WillOnce(Return(true));
@@ -727,23 +736,26 @@ TEST(GaugeMetricProducerTest, TestRemoveDimensionInOutput) {
atomMatcherId, logEventMatcherIndex, atomMatcher, uidMap)});
sp<MockStatsPullerManager> pullerManager = new StrictMock<MockStatsPullerManager>();
EXPECT_CALL(*pullerManager, Pull(tagId, kConfigKey, _, _))
.WillOnce(Invoke(
[](int tagId, const ConfigKey&, vector<std::shared_ptr<LogEvent>>* data, bool) {
data->clear();
data->push_back(CreateTwoValueLogEvent(tagId, bucketStartTimeNs + 3, 3, 4));
return true;
}))
.WillOnce(Invoke([](int tagId, const ConfigKey&,
EXPECT_CALL(*pullerManager, Pull(tagId, kConfigKey, _, _, _))
.WillOnce(Invoke([](int tagId, const ConfigKey&, const int64_t eventTimeNs,
vector<std::shared_ptr<LogEvent>>* data, bool) {
EXPECT_EQ(eventTimeNs, bucketStartTimeNs + 3);
data->clear();
data->push_back(CreateTwoValueLogEvent(tagId, bucketStartTimeNs + 10, 4, 5));
data->push_back(CreateTwoValueLogEvent(tagId, eventTimeNs, 3, 4));
return true;
}))
.WillOnce(Invoke([](int tagId, const ConfigKey&,
.WillOnce(Invoke([](int tagId, const ConfigKey&, const int64_t eventTimeNs,
vector<std::shared_ptr<LogEvent>>* data, bool) {
EXPECT_EQ(eventTimeNs, bucketStartTimeNs + 10);
data->clear();
data->push_back(CreateTwoValueLogEvent(tagId, bucketStartTimeNs + 20, 4, 6));
data->push_back(CreateTwoValueLogEvent(tagId, eventTimeNs, 4, 5));
return true;
}))
.WillOnce(Invoke([](int tagId, const ConfigKey&, const int64_t eventTimeNs,
vector<std::shared_ptr<LogEvent>>* data, bool) {
EXPECT_EQ(eventTimeNs, bucketStartTimeNs + 20);
data->clear();
data->push_back(CreateTwoValueLogEvent(tagId, eventTimeNs, 4, 6));
return true;
}))
.WillOnce(Return(true));
@@ -801,14 +813,14 @@ TEST(GaugeMetricProducerTest_BucketDrop, TestBucketDropWhenBucketTooSmall) {
atomMatcherId, logEventMatcherIndex, atomMatcher, uidMap)});
sp<MockStatsPullerManager> pullerManager = new StrictMock<MockStatsPullerManager>();
EXPECT_CALL(*pullerManager, Pull(tagId, kConfigKey, _, _))
EXPECT_CALL(*pullerManager, Pull(tagId, kConfigKey, bucketStartTimeNs + 3, _, _))
// Bucket start.
.WillOnce(Invoke(
[](int tagId, const ConfigKey&, vector<std::shared_ptr<LogEvent>>* data, bool) {
data->clear();
data->push_back(CreateRepeatedValueLogEvent(tagId, bucketStartTimeNs, 10));
return true;
}));
.WillOnce(Invoke([](int tagId, const ConfigKey&, const int64_t eventTimeNs,
vector<std::shared_ptr<LogEvent>>* data, bool) {
data->clear();
data->push_back(CreateRepeatedValueLogEvent(tagId, eventTimeNs, 10));
return true;
}));
int triggerId = 5;
GaugeMetricProducer gaugeProducer(kConfigKey, metric, -1 /*-1 meaning no condition*/, wizard,

File diff suppressed because it is too large Load Diff

View File

@@ -38,10 +38,11 @@ public:
int64_t nextPulltimeNs, int64_t intervalNs));
MOCK_METHOD3(UnRegisterReceiver,
void(int tagId, const ConfigKey& key, wp<PullDataReceiver> receiver));
MOCK_METHOD4(Pull, bool(const int pullCode, const ConfigKey& key,
vector<std::shared_ptr<LogEvent>>* data, bool useUids));
MOCK_METHOD4(Pull, bool(const int pullCode, const vector<int32_t>& uids,
MOCK_METHOD5(Pull, bool(const int pullCode, const ConfigKey& key, const int64_t eventTimeNs,
vector<std::shared_ptr<LogEvent>>* data, bool useUids));
MOCK_METHOD5(Pull,
bool(const int pullCode, const vector<int32_t>& uids, const int64_t eventTimeNs,
vector<std::shared_ptr<LogEvent>>* data, bool useUids));
MOCK_METHOD2(RegisterPullUidProvider,
void(const ConfigKey& configKey, wp<PullUidProvider> provider));
MOCK_METHOD2(UnregisterPullUidProvider,

View File

@@ -190,8 +190,8 @@ TEST(ShellSubscriberTest, testPulledSubscription) {
sp<MockStatsPullerManager> pullerManager = new StrictMock<MockStatsPullerManager>();
const vector<int32_t> uids = {AID_SYSTEM};
EXPECT_CALL(*pullerManager, Pull(10016, uids, _, _))
.WillRepeatedly(Invoke([](int tagId, const vector<int32_t>&,
EXPECT_CALL(*pullerManager, Pull(10016, uids, _, _, _))
.WillRepeatedly(Invoke([](int tagId, const vector<int32_t>&, const int64_t,
vector<std::shared_ptr<LogEvent>>* data, bool) {
data->clear();
data->push_back(makeCpuActiveTimeAtom(/*uid=*/kUid1, /*timeMillis=*/kCpuTime1));