1 /*
2 * Copyright (C) 2017 The Android Open Source Project
3 *
4 * Licensed under the Apache License, Version 2.0 (the "License");
5 * you may not use this file except in compliance with the License.
6 * You may obtain a copy of the License at
7 *
8 * http://www.apache.org/licenses/LICENSE-2.0
9 *
10 * Unless required by applicable law or agreed to in writing, software
11 * distributed under the License is distributed on an "AS IS" BASIS,
12 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13 * See the License for the specific language governing permissions and
14 * limitations under the License.
15 */
16
17 #define DEBUG false // STOPSHIP if true
18 #include "Log.h"
19
20 #include "../guardrail/StatsdStats.h"
21 #include "GaugeMetricProducer.h"
22 #include "../stats_log_util.h"
23
24 #include <cutils/log.h>
25
26 using android::util::FIELD_COUNT_REPEATED;
27 using android::util::FIELD_TYPE_BOOL;
28 using android::util::FIELD_TYPE_FLOAT;
29 using android::util::FIELD_TYPE_INT32;
30 using android::util::FIELD_TYPE_INT64;
31 using android::util::FIELD_TYPE_MESSAGE;
32 using android::util::FIELD_TYPE_STRING;
33 using android::util::ProtoOutputStream;
34 using std::map;
35 using std::string;
36 using std::unordered_map;
37 using std::vector;
38 using std::make_shared;
39 using std::shared_ptr;
40
41 namespace android {
42 namespace os {
43 namespace statsd {
44
45 // for StatsLogReport
46 const int FIELD_ID_ID = 1;
47 const int FIELD_ID_GAUGE_METRICS = 8;
48 const int FIELD_ID_TIME_BASE = 9;
49 const int FIELD_ID_BUCKET_SIZE = 10;
50 const int FIELD_ID_DIMENSION_PATH_IN_WHAT = 11;
51 const int FIELD_ID_DIMENSION_PATH_IN_CONDITION = 12;
52 const int FIELD_ID_IS_ACTIVE = 14;
53 // for GaugeMetricDataWrapper
54 const int FIELD_ID_DATA = 1;
55 const int FIELD_ID_SKIPPED = 2;
56 const int FIELD_ID_SKIPPED_START_MILLIS = 3;
57 const int FIELD_ID_SKIPPED_END_MILLIS = 4;
58 // for GaugeMetricData
59 const int FIELD_ID_DIMENSION_IN_WHAT = 1;
60 const int FIELD_ID_DIMENSION_IN_CONDITION = 2;
61 const int FIELD_ID_BUCKET_INFO = 3;
62 const int FIELD_ID_DIMENSION_LEAF_IN_WHAT = 4;
63 const int FIELD_ID_DIMENSION_LEAF_IN_CONDITION = 5;
64 // for GaugeBucketInfo
65 const int FIELD_ID_ATOM = 3;
66 const int FIELD_ID_ELAPSED_ATOM_TIMESTAMP = 4;
67 const int FIELD_ID_BUCKET_NUM = 6;
68 const int FIELD_ID_START_BUCKET_ELAPSED_MILLIS = 7;
69 const int FIELD_ID_END_BUCKET_ELAPSED_MILLIS = 8;
70
GaugeMetricProducer(const ConfigKey & key,const GaugeMetric & metric,const int conditionIndex,const sp<ConditionWizard> & wizard,const int whatMatcherIndex,const sp<EventMatcherWizard> & matcherWizard,const int pullTagId,const int triggerAtomId,const int atomId,const int64_t timeBaseNs,const int64_t startTimeNs,const sp<StatsPullerManager> & pullerManager)71 GaugeMetricProducer::GaugeMetricProducer(
72 const ConfigKey& key, const GaugeMetric& metric, const int conditionIndex,
73 const sp<ConditionWizard>& wizard, const int whatMatcherIndex,
74 const sp<EventMatcherWizard>& matcherWizard, const int pullTagId, const int triggerAtomId,
75 const int atomId, const int64_t timeBaseNs, const int64_t startTimeNs,
76 const sp<StatsPullerManager>& pullerManager)
77 : MetricProducer(metric.id(), key, timeBaseNs, conditionIndex, wizard),
78 mWhatMatcherIndex(whatMatcherIndex),
79 mEventMatcherWizard(matcherWizard),
80 mPullerManager(pullerManager),
81 mPullTagId(pullTagId),
82 mTriggerAtomId(triggerAtomId),
83 mAtomId(atomId),
84 mIsPulled(pullTagId != -1),
85 mMinBucketSizeNs(metric.min_bucket_size_nanos()),
86 mMaxPullDelayNs(metric.max_pull_delay_sec() > 0 ? metric.max_pull_delay_sec() * NS_PER_SEC
87 : StatsdStats::kPullMaxDelayNs),
88 mDimensionSoftLimit(StatsdStats::kAtomDimensionKeySizeLimitMap.find(pullTagId) !=
89 StatsdStats::kAtomDimensionKeySizeLimitMap.end()
90 ? StatsdStats::kAtomDimensionKeySizeLimitMap.at(pullTagId).first
91 : StatsdStats::kDimensionKeySizeSoftLimit),
92 mDimensionHardLimit(StatsdStats::kAtomDimensionKeySizeLimitMap.find(pullTagId) !=
93 StatsdStats::kAtomDimensionKeySizeLimitMap.end()
94 ? StatsdStats::kAtomDimensionKeySizeLimitMap.at(pullTagId).second
95 : StatsdStats::kDimensionKeySizeHardLimit),
96 mGaugeAtomsPerDimensionLimit(metric.max_num_gauge_atoms_per_bucket()),
97 mSplitBucketForAppUpgrade(metric.split_bucket_for_app_upgrade()) {
98 mCurrentSlicedBucket = std::make_shared<DimToGaugeAtomsMap>();
99 mCurrentSlicedBucketForAnomaly = std::make_shared<DimToValMap>();
100 int64_t bucketSizeMills = 0;
101 if (metric.has_bucket()) {
102 bucketSizeMills = TimeUnitToBucketSizeInMillisGuardrailed(key.GetUid(), metric.bucket());
103 } else {
104 bucketSizeMills = TimeUnitToBucketSizeInMillis(ONE_HOUR);
105 }
106 mBucketSizeNs = bucketSizeMills * 1000000;
107
108 mSamplingType = metric.sampling_type();
109 if (!metric.gauge_fields_filter().include_all()) {
110 translateFieldMatcher(metric.gauge_fields_filter().fields(), &mFieldMatchers);
111 }
112
113 if (metric.has_dimensions_in_what()) {
114 translateFieldMatcher(metric.dimensions_in_what(), &mDimensionsInWhat);
115 mContainANYPositionInDimensionsInWhat = HasPositionANY(metric.dimensions_in_what());
116 }
117
118 if (metric.has_dimensions_in_condition()) {
119 translateFieldMatcher(metric.dimensions_in_condition(), &mDimensionsInCondition);
120 }
121
122 if (metric.links().size() > 0) {
123 for (const auto& link : metric.links()) {
124 Metric2Condition mc;
125 mc.conditionId = link.condition();
126 translateFieldMatcher(link.fields_in_what(), &mc.metricFields);
127 translateFieldMatcher(link.fields_in_condition(), &mc.conditionFields);
128 mMetric2ConditionLinks.push_back(mc);
129 }
130 }
131 mConditionSliced = (metric.links().size() > 0) || (mDimensionsInCondition.size() > 0);
132 mSliceByPositionALL = HasPositionALL(metric.dimensions_in_what()) ||
133 HasPositionALL(metric.dimensions_in_condition());
134
135 flushIfNeededLocked(startTimeNs);
136 // Kicks off the puller immediately.
137 if (mIsPulled && mSamplingType == GaugeMetric::RANDOM_ONE_SAMPLE) {
138 mPullerManager->RegisterReceiver(mPullTagId, this, getCurrentBucketEndTimeNs(),
139 mBucketSizeNs);
140 }
141
142 // Adjust start for partial bucket
143 mCurrentBucketStartTimeNs = startTimeNs;
144
145 VLOG("Gauge metric %lld created. bucket size %lld start_time: %lld sliced %d",
146 (long long)metric.id(), (long long)mBucketSizeNs, (long long)mTimeBaseNs,
147 mConditionSliced);
148 }
149
~GaugeMetricProducer()150 GaugeMetricProducer::~GaugeMetricProducer() {
151 VLOG("~GaugeMetricProducer() called");
152 if (mIsPulled && mSamplingType == GaugeMetric::RANDOM_ONE_SAMPLE) {
153 mPullerManager->UnRegisterReceiver(mPullTagId, this);
154 }
155 }
156
dumpStatesLocked(FILE * out,bool verbose) const157 void GaugeMetricProducer::dumpStatesLocked(FILE* out, bool verbose) const {
158 if (mCurrentSlicedBucket == nullptr ||
159 mCurrentSlicedBucket->size() == 0) {
160 return;
161 }
162
163 fprintf(out, "GaugeMetric %lld dimension size %lu\n", (long long)mMetricId,
164 (unsigned long)mCurrentSlicedBucket->size());
165 if (verbose) {
166 for (const auto& it : *mCurrentSlicedBucket) {
167 fprintf(out, "\t(what)%s\t(condition)%s %d atoms\n",
168 it.first.getDimensionKeyInWhat().toString().c_str(),
169 it.first.getDimensionKeyInCondition().toString().c_str(),
170 (int)it.second.size());
171 }
172 }
173 }
174
clearPastBucketsLocked(const int64_t dumpTimeNs)175 void GaugeMetricProducer::clearPastBucketsLocked(const int64_t dumpTimeNs) {
176 flushIfNeededLocked(dumpTimeNs);
177 mPastBuckets.clear();
178 mSkippedBuckets.clear();
179 }
180
onDumpReportLocked(const int64_t dumpTimeNs,const bool include_current_partial_bucket,const bool erase_data,const DumpLatency dumpLatency,std::set<string> * str_set,ProtoOutputStream * protoOutput)181 void GaugeMetricProducer::onDumpReportLocked(const int64_t dumpTimeNs,
182 const bool include_current_partial_bucket,
183 const bool erase_data,
184 const DumpLatency dumpLatency,
185 std::set<string> *str_set,
186 ProtoOutputStream* protoOutput) {
187 VLOG("Gauge metric %lld report now...", (long long)mMetricId);
188 if (include_current_partial_bucket) {
189 flushLocked(dumpTimeNs);
190 } else {
191 flushIfNeededLocked(dumpTimeNs);
192 }
193
194 protoOutput->write(FIELD_TYPE_INT64 | FIELD_ID_ID, (long long)mMetricId);
195 protoOutput->write(FIELD_TYPE_BOOL | FIELD_ID_IS_ACTIVE, isActiveLocked());
196
197 if (mPastBuckets.empty()) {
198 return;
199 }
200
201 protoOutput->write(FIELD_TYPE_INT64 | FIELD_ID_TIME_BASE, (long long)mTimeBaseNs);
202 protoOutput->write(FIELD_TYPE_INT64 | FIELD_ID_BUCKET_SIZE, (long long)mBucketSizeNs);
203
204 // Fills the dimension path if not slicing by ALL.
205 if (!mSliceByPositionALL) {
206 if (!mDimensionsInWhat.empty()) {
207 uint64_t dimenPathToken = protoOutput->start(
208 FIELD_TYPE_MESSAGE | FIELD_ID_DIMENSION_PATH_IN_WHAT);
209 writeDimensionPathToProto(mDimensionsInWhat, protoOutput);
210 protoOutput->end(dimenPathToken);
211 }
212 if (!mDimensionsInCondition.empty()) {
213 uint64_t dimenPathToken = protoOutput->start(
214 FIELD_TYPE_MESSAGE | FIELD_ID_DIMENSION_PATH_IN_CONDITION);
215 writeDimensionPathToProto(mDimensionsInCondition, protoOutput);
216 protoOutput->end(dimenPathToken);
217 }
218 }
219
220 uint64_t protoToken = protoOutput->start(FIELD_TYPE_MESSAGE | FIELD_ID_GAUGE_METRICS);
221
222 for (const auto& pair : mSkippedBuckets) {
223 uint64_t wrapperToken =
224 protoOutput->start(FIELD_TYPE_MESSAGE | FIELD_COUNT_REPEATED | FIELD_ID_SKIPPED);
225 protoOutput->write(FIELD_TYPE_INT64 | FIELD_ID_SKIPPED_START_MILLIS,
226 (long long)(NanoToMillis(pair.first)));
227 protoOutput->write(FIELD_TYPE_INT64 | FIELD_ID_SKIPPED_END_MILLIS,
228 (long long)(NanoToMillis(pair.second)));
229 protoOutput->end(wrapperToken);
230 }
231
232 for (const auto& pair : mPastBuckets) {
233 const MetricDimensionKey& dimensionKey = pair.first;
234
235 VLOG("Gauge dimension key %s", dimensionKey.toString().c_str());
236 uint64_t wrapperToken =
237 protoOutput->start(FIELD_TYPE_MESSAGE | FIELD_COUNT_REPEATED | FIELD_ID_DATA);
238
239 // First fill dimension.
240 if (mSliceByPositionALL) {
241 uint64_t dimensionToken = protoOutput->start(
242 FIELD_TYPE_MESSAGE | FIELD_ID_DIMENSION_IN_WHAT);
243 writeDimensionToProto(dimensionKey.getDimensionKeyInWhat(), str_set, protoOutput);
244 protoOutput->end(dimensionToken);
245
246 if (dimensionKey.hasDimensionKeyInCondition()) {
247 uint64_t dimensionInConditionToken = protoOutput->start(
248 FIELD_TYPE_MESSAGE | FIELD_ID_DIMENSION_IN_CONDITION);
249 writeDimensionToProto(dimensionKey.getDimensionKeyInCondition(),
250 str_set, protoOutput);
251 protoOutput->end(dimensionInConditionToken);
252 }
253 } else {
254 writeDimensionLeafNodesToProto(dimensionKey.getDimensionKeyInWhat(),
255 FIELD_ID_DIMENSION_LEAF_IN_WHAT, str_set, protoOutput);
256 if (dimensionKey.hasDimensionKeyInCondition()) {
257 writeDimensionLeafNodesToProto(dimensionKey.getDimensionKeyInCondition(),
258 FIELD_ID_DIMENSION_LEAF_IN_CONDITION,
259 str_set, protoOutput);
260 }
261 }
262
263 // Then fill bucket_info (GaugeBucketInfo).
264 for (const auto& bucket : pair.second) {
265 uint64_t bucketInfoToken = protoOutput->start(
266 FIELD_TYPE_MESSAGE | FIELD_COUNT_REPEATED | FIELD_ID_BUCKET_INFO);
267
268 if (bucket.mBucketEndNs - bucket.mBucketStartNs != mBucketSizeNs) {
269 protoOutput->write(FIELD_TYPE_INT64 | FIELD_ID_START_BUCKET_ELAPSED_MILLIS,
270 (long long)NanoToMillis(bucket.mBucketStartNs));
271 protoOutput->write(FIELD_TYPE_INT64 | FIELD_ID_END_BUCKET_ELAPSED_MILLIS,
272 (long long)NanoToMillis(bucket.mBucketEndNs));
273 } else {
274 protoOutput->write(FIELD_TYPE_INT64 | FIELD_ID_BUCKET_NUM,
275 (long long)(getBucketNumFromEndTimeNs(bucket.mBucketEndNs)));
276 }
277
278 if (!bucket.mGaugeAtoms.empty()) {
279 for (const auto& atom : bucket.mGaugeAtoms) {
280 uint64_t atomsToken =
281 protoOutput->start(FIELD_TYPE_MESSAGE | FIELD_COUNT_REPEATED |
282 FIELD_ID_ATOM);
283 writeFieldValueTreeToStream(mAtomId, *(atom.mFields), protoOutput);
284 protoOutput->end(atomsToken);
285 }
286 for (const auto& atom : bucket.mGaugeAtoms) {
287 const int64_t elapsedTimestampNs =
288 truncateTimestampIfNecessary(mAtomId, atom.mElapsedTimestamps);
289 protoOutput->write(
290 FIELD_TYPE_INT64 | FIELD_COUNT_REPEATED | FIELD_ID_ELAPSED_ATOM_TIMESTAMP,
291 (long long)elapsedTimestampNs);
292 }
293 }
294 protoOutput->end(bucketInfoToken);
295 VLOG("Gauge \t bucket [%lld - %lld] includes %d atoms.",
296 (long long)bucket.mBucketStartNs, (long long)bucket.mBucketEndNs,
297 (int)bucket.mGaugeAtoms.size());
298 }
299 protoOutput->end(wrapperToken);
300 }
301 protoOutput->end(protoToken);
302
303
304 if (erase_data) {
305 mPastBuckets.clear();
306 mSkippedBuckets.clear();
307 }
308 }
309
prepareFirstBucketLocked()310 void GaugeMetricProducer::prepareFirstBucketLocked() {
311 if (mIsActive && mIsPulled && mSamplingType == GaugeMetric::RANDOM_ONE_SAMPLE) {
312 pullAndMatchEventsLocked(mCurrentBucketStartTimeNs);
313 }
314 }
315
pullAndMatchEventsLocked(const int64_t timestampNs)316 void GaugeMetricProducer::pullAndMatchEventsLocked(const int64_t timestampNs) {
317 bool triggerPuller = false;
318 switch(mSamplingType) {
319 // When the metric wants to do random sampling and there is already one gauge atom for the
320 // current bucket, do not do it again.
321 case GaugeMetric::RANDOM_ONE_SAMPLE: {
322 triggerPuller = mCondition == ConditionState::kTrue && mCurrentSlicedBucket->empty();
323 break;
324 }
325 case GaugeMetric::CONDITION_CHANGE_TO_TRUE: {
326 triggerPuller = mCondition == ConditionState::kTrue;
327 break;
328 }
329 case GaugeMetric::FIRST_N_SAMPLES: {
330 triggerPuller = mCondition == ConditionState::kTrue;
331 break;
332 }
333 default:
334 break;
335 }
336 if (!triggerPuller) {
337 return;
338 }
339 vector<std::shared_ptr<LogEvent>> allData;
340 if (!mPullerManager->Pull(mPullTagId, &allData)) {
341 ALOGE("Gauge Stats puller failed for tag: %d at %lld", mPullTagId, (long long)timestampNs);
342 return;
343 }
344 const int64_t pullDelayNs = getElapsedRealtimeNs() - timestampNs;
345 if (pullDelayNs > mMaxPullDelayNs) {
346 ALOGE("Pull finish too late for atom %d", mPullTagId);
347 StatsdStats::getInstance().notePullExceedMaxDelay(mPullTagId);
348 StatsdStats::getInstance().notePullDelay(mPullTagId, pullDelayNs);
349 return;
350 }
351 StatsdStats::getInstance().notePullDelay(mPullTagId, pullDelayNs);
352 for (const auto& data : allData) {
353 LogEvent localCopy = data->makeCopy();
354 localCopy.setElapsedTimestampNs(timestampNs);
355 if (mEventMatcherWizard->matchLogEvent(localCopy, mWhatMatcherIndex) ==
356 MatchingState::kMatched) {
357 onMatchedLogEventLocked(mWhatMatcherIndex, localCopy);
358 }
359 }
360 }
361
onActiveStateChangedLocked(const int64_t & eventTimeNs)362 void GaugeMetricProducer::onActiveStateChangedLocked(const int64_t& eventTimeNs) {
363 MetricProducer::onActiveStateChangedLocked(eventTimeNs);
364 if (ConditionState::kTrue != mCondition || !mIsPulled) {
365 return;
366 }
367 if (mTriggerAtomId == -1 || (mIsActive && mSamplingType == GaugeMetric::RANDOM_ONE_SAMPLE)) {
368 pullAndMatchEventsLocked(eventTimeNs);
369 }
370
371 }
372
onConditionChangedLocked(const bool conditionMet,const int64_t eventTimeNs)373 void GaugeMetricProducer::onConditionChangedLocked(const bool conditionMet,
374 const int64_t eventTimeNs) {
375 VLOG("GaugeMetric %lld onConditionChanged", (long long)mMetricId);
376
377 mCondition = conditionMet ? ConditionState::kTrue : ConditionState::kFalse;
378 if (!mIsActive) {
379 return;
380 }
381
382 flushIfNeededLocked(eventTimeNs);
383 if (mIsPulled && mTriggerAtomId == -1) {
384 pullAndMatchEventsLocked(eventTimeNs);
385 } // else: Push mode. No need to proactively pull the gauge data.
386 }
387
onSlicedConditionMayChangeLocked(bool overallCondition,const int64_t eventTimeNs)388 void GaugeMetricProducer::onSlicedConditionMayChangeLocked(bool overallCondition,
389 const int64_t eventTimeNs) {
390 VLOG("GaugeMetric %lld onSlicedConditionMayChange overall condition %d", (long long)mMetricId,
391 overallCondition);
392 mCondition = overallCondition ? ConditionState::kTrue : ConditionState::kFalse;
393 if (!mIsActive) {
394 return;
395 }
396
397 flushIfNeededLocked(eventTimeNs);
398 // If the condition is sliced, mCondition is true if any of the dimensions is true. And we will
399 // pull for every dimension.
400 if (mIsPulled && mTriggerAtomId == -1) {
401 pullAndMatchEventsLocked(eventTimeNs);
402 } // else: Push mode. No need to proactively pull the gauge data.
403 }
404
getGaugeFields(const LogEvent & event)405 std::shared_ptr<vector<FieldValue>> GaugeMetricProducer::getGaugeFields(const LogEvent& event) {
406 std::shared_ptr<vector<FieldValue>> gaugeFields;
407 if (mFieldMatchers.size() > 0) {
408 gaugeFields = std::make_shared<vector<FieldValue>>();
409 filterGaugeValues(mFieldMatchers, event.getValues(), gaugeFields.get());
410 } else {
411 gaugeFields = std::make_shared<vector<FieldValue>>(event.getValues());
412 }
413 // Trim all dimension fields from output. Dimensions will appear in output report and will
414 // benefit from dictionary encoding. For large pulled atoms, this can give the benefit of
415 // optional repeated field.
416 for (const auto& field : mDimensionsInWhat) {
417 for (auto it = gaugeFields->begin(); it != gaugeFields->end();) {
418 if (it->mField.matches(field)) {
419 it = gaugeFields->erase(it);
420 } else {
421 it++;
422 }
423 }
424 }
425 return gaugeFields;
426 }
427
onDataPulled(const std::vector<std::shared_ptr<LogEvent>> & allData,bool pullSuccess,int64_t originalPullTimeNs)428 void GaugeMetricProducer::onDataPulled(const std::vector<std::shared_ptr<LogEvent>>& allData,
429 bool pullSuccess, int64_t originalPullTimeNs) {
430 std::lock_guard<std::mutex> lock(mMutex);
431 if (!pullSuccess || allData.size() == 0) {
432 return;
433 }
434 for (const auto& data : allData) {
435 if (mEventMatcherWizard->matchLogEvent(
436 *data, mWhatMatcherIndex) == MatchingState::kMatched) {
437 onMatchedLogEventLocked(mWhatMatcherIndex, *data);
438 }
439 }
440 }
441
hitGuardRailLocked(const MetricDimensionKey & newKey)442 bool GaugeMetricProducer::hitGuardRailLocked(const MetricDimensionKey& newKey) {
443 if (mCurrentSlicedBucket->find(newKey) != mCurrentSlicedBucket->end()) {
444 return false;
445 }
446 // 1. Report the tuple count if the tuple count > soft limit
447 if (mCurrentSlicedBucket->size() > mDimensionSoftLimit - 1) {
448 size_t newTupleCount = mCurrentSlicedBucket->size() + 1;
449 StatsdStats::getInstance().noteMetricDimensionSize(mConfigKey, mMetricId, newTupleCount);
450 // 2. Don't add more tuples, we are above the allowed threshold. Drop the data.
451 if (newTupleCount > mDimensionHardLimit) {
452 ALOGE("GaugeMetric %lld dropping data for dimension key %s",
453 (long long)mMetricId, newKey.toString().c_str());
454 return true;
455 }
456 }
457
458 return false;
459 }
460
onMatchedLogEventInternalLocked(const size_t matcherIndex,const MetricDimensionKey & eventKey,const ConditionKey & conditionKey,bool condition,const LogEvent & event)461 void GaugeMetricProducer::onMatchedLogEventInternalLocked(
462 const size_t matcherIndex, const MetricDimensionKey& eventKey,
463 const ConditionKey& conditionKey, bool condition,
464 const LogEvent& event) {
465 if (condition == false) {
466 return;
467 }
468 int64_t eventTimeNs = event.GetElapsedTimestampNs();
469 if (eventTimeNs < mCurrentBucketStartTimeNs) {
470 VLOG("Gauge Skip event due to late arrival: %lld vs %lld", (long long)eventTimeNs,
471 (long long)mCurrentBucketStartTimeNs);
472 return;
473 }
474 flushIfNeededLocked(eventTimeNs);
475
476 if (mTriggerAtomId == event.GetTagId()) {
477 pullAndMatchEventsLocked(eventTimeNs);
478 return;
479 }
480
481 // When gauge metric wants to randomly sample the output atom, we just simply use the first
482 // gauge in the given bucket.
483 if (mCurrentSlicedBucket->find(eventKey) != mCurrentSlicedBucket->end() &&
484 mSamplingType == GaugeMetric::RANDOM_ONE_SAMPLE) {
485 return;
486 }
487 if (hitGuardRailLocked(eventKey)) {
488 return;
489 }
490 if ((*mCurrentSlicedBucket)[eventKey].size() >= mGaugeAtomsPerDimensionLimit) {
491 return;
492 }
493 GaugeAtom gaugeAtom(getGaugeFields(event), eventTimeNs);
494 (*mCurrentSlicedBucket)[eventKey].push_back(gaugeAtom);
495 // Anomaly detection on gauge metric only works when there is one numeric
496 // field specified.
497 if (mAnomalyTrackers.size() > 0) {
498 if (gaugeAtom.mFields->size() == 1) {
499 const Value& value = gaugeAtom.mFields->begin()->mValue;
500 long gaugeVal = 0;
501 if (value.getType() == INT) {
502 gaugeVal = (long)value.int_value;
503 } else if (value.getType() == LONG) {
504 gaugeVal = value.long_value;
505 }
506 for (auto& tracker : mAnomalyTrackers) {
507 tracker->detectAndDeclareAnomaly(eventTimeNs, mCurrentBucketNum, mMetricId,
508 eventKey, gaugeVal);
509 }
510 }
511 }
512 }
513
updateCurrentSlicedBucketForAnomaly()514 void GaugeMetricProducer::updateCurrentSlicedBucketForAnomaly() {
515 for (const auto& slice : *mCurrentSlicedBucket) {
516 if (slice.second.empty()) {
517 continue;
518 }
519 const Value& value = slice.second.front().mFields->front().mValue;
520 long gaugeVal = 0;
521 if (value.getType() == INT) {
522 gaugeVal = (long)value.int_value;
523 } else if (value.getType() == LONG) {
524 gaugeVal = value.long_value;
525 }
526 (*mCurrentSlicedBucketForAnomaly)[slice.first] = gaugeVal;
527 }
528 }
529
dropDataLocked(const int64_t dropTimeNs)530 void GaugeMetricProducer::dropDataLocked(const int64_t dropTimeNs) {
531 flushIfNeededLocked(dropTimeNs);
532 StatsdStats::getInstance().noteBucketDropped(mMetricId);
533 mPastBuckets.clear();
534 }
535
536 // When a new matched event comes in, we check if event falls into the current
537 // bucket. If not, flush the old counter to past buckets and initialize the new
538 // bucket.
539 // if data is pushed, onMatchedLogEvent will only be called through onConditionChanged() inside
540 // the GaugeMetricProducer while holding the lock.
flushIfNeededLocked(const int64_t & eventTimeNs)541 void GaugeMetricProducer::flushIfNeededLocked(const int64_t& eventTimeNs) {
542 int64_t currentBucketEndTimeNs = getCurrentBucketEndTimeNs();
543
544 if (eventTimeNs < currentBucketEndTimeNs) {
545 VLOG("Gauge eventTime is %lld, less than next bucket start time %lld",
546 (long long)eventTimeNs, (long long)(mCurrentBucketStartTimeNs + mBucketSizeNs));
547 return;
548 }
549
550 // Adjusts the bucket start and end times.
551 int64_t numBucketsForward = 1 + (eventTimeNs - currentBucketEndTimeNs) / mBucketSizeNs;
552 int64_t nextBucketNs = currentBucketEndTimeNs + (numBucketsForward - 1) * mBucketSizeNs;
553 flushCurrentBucketLocked(eventTimeNs, nextBucketNs);
554
555 mCurrentBucketNum += numBucketsForward;
556 VLOG("Gauge metric %lld: new bucket start time: %lld", (long long)mMetricId,
557 (long long)mCurrentBucketStartTimeNs);
558 }
559
flushCurrentBucketLocked(const int64_t & eventTimeNs,const int64_t & nextBucketStartTimeNs)560 void GaugeMetricProducer::flushCurrentBucketLocked(const int64_t& eventTimeNs,
561 const int64_t& nextBucketStartTimeNs) {
562 int64_t fullBucketEndTimeNs = getCurrentBucketEndTimeNs();
563
564 GaugeBucket info;
565 info.mBucketStartNs = mCurrentBucketStartTimeNs;
566 if (eventTimeNs < fullBucketEndTimeNs) {
567 info.mBucketEndNs = eventTimeNs;
568 } else {
569 info.mBucketEndNs = fullBucketEndTimeNs;
570 }
571
572 if (info.mBucketEndNs - mCurrentBucketStartTimeNs >= mMinBucketSizeNs) {
573 for (const auto& slice : *mCurrentSlicedBucket) {
574 info.mGaugeAtoms = slice.second;
575 auto& bucketList = mPastBuckets[slice.first];
576 bucketList.push_back(info);
577 VLOG("Gauge gauge metric %lld, dump key value: %s", (long long)mMetricId,
578 slice.first.toString().c_str());
579 }
580 } else {
581 mSkippedBuckets.emplace_back(info.mBucketStartNs, info.mBucketEndNs);
582 }
583
584 // If we have anomaly trackers, we need to update the partial bucket values.
585 if (mAnomalyTrackers.size() > 0) {
586 updateCurrentSlicedBucketForAnomaly();
587
588 if (eventTimeNs > fullBucketEndTimeNs) {
589 // This is known to be a full bucket, so send this data to the anomaly tracker.
590 for (auto& tracker : mAnomalyTrackers) {
591 tracker->addPastBucket(mCurrentSlicedBucketForAnomaly, mCurrentBucketNum);
592 }
593 mCurrentSlicedBucketForAnomaly = std::make_shared<DimToValMap>();
594 }
595 }
596
597 StatsdStats::getInstance().noteBucketCount(mMetricId);
598 mCurrentSlicedBucket = std::make_shared<DimToGaugeAtomsMap>();
599 mCurrentBucketStartTimeNs = nextBucketStartTimeNs;
600 }
601
byteSizeLocked() const602 size_t GaugeMetricProducer::byteSizeLocked() const {
603 size_t totalSize = 0;
604 for (const auto& pair : mPastBuckets) {
605 for (const auto& bucket : pair.second) {
606 totalSize += bucket.mGaugeAtoms.size() * sizeof(GaugeAtom);
607 for (const auto& atom : bucket.mGaugeAtoms) {
608 if (atom.mFields != nullptr) {
609 totalSize += atom.mFields->size() * sizeof(FieldValue);
610 }
611 }
612 }
613 }
614 return totalSize;
615 }
616
617 } // namespace statsd
618 } // namespace os
619 } // namespace android
620