Merge "multi-value aggregation in ValueMetric" (69df0530) · Commits · e / os / android_frameworks_base

cmds/statsd/src/metrics/ValueMetricProducer.cpp

+165 −134

Original line number	Diff line number	Diff line
		@@ -64,8 +64,10 @@ const int FIELD_ID_BUCKET_INFO = 3;
		const int FIELD_ID_DIMENSION_LEAF_IN_WHAT = 4;
		const int FIELD_ID_DIMENSION_LEAF_IN_CONDITION = 5;
		// for ValueBucketInfo
		const int FIELD_ID_VALUE_LONG = 7;
		const int FIELD_ID_VALUE_DOUBLE = 8;
		const int FIELD_ID_VALUE_INDEX = 1;
		const int FIELD_ID_VALUE_LONG = 2;
		const int FIELD_ID_VALUE_DOUBLE = 3;
		const int FIELD_ID_VALUES = 9;
		const int FIELD_ID_BUCKET_NUM = 4;
		const int FIELD_ID_START_BUCKET_ELAPSED_MILLIS = 5;
		const int FIELD_ID_END_BUCKET_ELAPSED_MILLIS = 6;
		@@ -78,7 +80,6 @@ ValueMetricProducer::ValueMetricProducer(const ConfigKey& key, const ValueMetric
		const sp<StatsPullerManager>& pullerManager)
		: MetricProducer(metric.id(), key, timeBaseNs, conditionIndex, wizard),
		mPullerManager(pullerManager),
		mValueField(metric.value_field()),
		mPullTagId(pullTagId),
		mIsPulled(pullTagId != -1),
		mMinBucketSizeNs(metric.min_bucket_size_nanos()),
		@@ -103,6 +104,9 @@ ValueMetricProducer::ValueMetricProducer(const ConfigKey& key, const ValueMetric
		}

		mBucketSizeNs = bucketSizeMills * 1000000;

		translateFieldMatcher(metric.value_field(), &mFieldMatchers);

		if (metric.has_dimensions_in_what()) {
		translateFieldMatcher(metric.dimensions_in_what(), &mDimensionsInWhat);
		mContainANYPositionInDimensionsInWhat = HasPositionANY(metric.dimensions_in_what());
		@@ -122,9 +126,6 @@ ValueMetricProducer::ValueMetricProducer(const ConfigKey& key, const ValueMetric
		}
		}

		if (mValueField.child_size() > 0) {
		mField = mValueField.child(0).field();
		}
		mConditionSliced = (metric.links().size() > 0) \|\| (mDimensionsInCondition.size() > 0);
		mSliceByPositionALL = HasPositionALL(metric.dimensions_in_what()) \|\|
		HasPositionALL(metric.dimensions_in_condition());
		@@ -259,18 +260,27 @@ void ValueMetricProducer::onDumpReportLocked(const int64_t dumpTimeNs,
		protoOutput->write(FIELD_TYPE_INT64 \| FIELD_ID_BUCKET_NUM,
		(long long)(getBucketNumFromEndTimeNs(bucket.mBucketEndNs)));
		}
		if (bucket.value.getType() == LONG) {
		for (int i = 0; i < (int)bucket.valueIndex.size(); i ++) {
		int index = bucket.valueIndex[i];
		const Value& value = bucket.values[i];
		uint64_t valueToken = protoOutput->start(
		FIELD_TYPE_MESSAGE \| FIELD_COUNT_REPEATED \| FIELD_ID_VALUES);
		protoOutput->write(FIELD_TYPE_INT32 \| FIELD_ID_VALUE_INDEX,
		index);
		if (value.getType() == LONG) {
		protoOutput->write(FIELD_TYPE_INT64 \| FIELD_ID_VALUE_LONG,
		(long long)bucket.value.long_value);
		VLOG("\t bucket [%lld - %lld] count: %lld", (long long)bucket.mBucketStartNs,
		(long long)bucket.mBucketEndNs, (long long)bucket.value.long_value);
		} else if (bucket.value.getType() == DOUBLE) {
		(long long)value.long_value);
		VLOG("\t bucket [%lld - %lld] value %d: %lld", (long long)bucket.mBucketStartNs,
		(long long)bucket.mBucketEndNs, index, (long long)value.long_value);
		} else if (value.getType() == DOUBLE) {
		protoOutput->write(FIELD_TYPE_DOUBLE \| FIELD_ID_VALUE_DOUBLE,
		bucket.value.double_value);
		VLOG("\t bucket [%lld - %lld] count: %.2f", (long long)bucket.mBucketStartNs,
		(long long)bucket.mBucketEndNs, bucket.value.double_value);
		value.double_value);
		VLOG("\t bucket [%lld - %lld] value %d: %.2f", (long long)bucket.mBucketStartNs,
		(long long)bucket.mBucketEndNs, index, value.double_value);
		} else {
		VLOG("Wrong value type for ValueMetric output: %d", bucket.value.getType());
		VLOG("Wrong value type for ValueMetric output: %d", value.getType());
		}
		protoOutput->end(valueToken);
		}
		protoOutput->end(bucketInfoToken);
		}
		@@ -303,7 +313,9 @@ void ValueMetricProducer::onConditionChangedLocked(const bool condition,
		// when condition change from true to false, clear diff base
		if (mUseDiff && mCondition && !condition) {
		for (auto& slice : mCurrentSlicedBucket) {
		slice.second.hasBase = false;
		for (auto& interval : slice.second) {
		interval.hasBase = false;
		}
		}
		}

		@@ -363,10 +375,12 @@ void ValueMetricProducer::dumpStatesLocked(FILE* out, bool verbose) const {
		(unsigned long)mCurrentSlicedBucket.size());
		if (verbose) {
		for (const auto& it : mCurrentSlicedBucket) {
		for (const auto& interval : it.second) {
		fprintf(out, "\t(what)%s\t(condition)%s (value)%s\n",
		it.first.getDimensionKeyInWhat().toString().c_str(),
		it.first.getDimensionKeyInCondition().toString().c_str(),
		it.second.value.toString().c_str());
		interval.value.toString().c_str());
		}
		}
		}
		}
		@@ -391,25 +405,29 @@ bool ValueMetricProducer::hitGuardRailLocked(const MetricDimensionKey& newKey) {
		return false;
		}

		const Value getDoubleOrLong(const Value& value) {
		Value v;
		switch (value.type) {
		bool getDoubleOrLong(const LogEvent& event, const Matcher& matcher, Value& ret) {
		for (const FieldValue& value : event.getValues()) {
		if (value.mField.matches(matcher)) {
		switch (value.mValue.type) {
		case INT:
		v.setLong(value.int_value);
		ret.setLong(value.mValue.int_value);
		break;
		case LONG:
		v.setLong(value.long_value);
		ret.setLong(value.mValue.long_value);
		break;
		case FLOAT:
		v.setDouble(value.float_value);
		ret.setDouble(value.mValue.float_value);
		break;
		case DOUBLE:
		v.setDouble(value.double_value);
		ret.setDouble(value.mValue.double_value);
		break;
		default:
		break;
		}
		return v;
		return true;
		}
		}
		return false;
		}

		void ValueMetricProducer::onMatchedLogEventInternalLocked(const size_t matcherIndex,
		@@ -436,14 +454,21 @@ void ValueMetricProducer::onMatchedLogEventInternalLocked(const size_t matcherIn
		if (hitGuardRailLocked(eventKey)) {
		return;
		}
		Interval& interval = mCurrentSlicedBucket[eventKey];
		vector<Interval>& multiIntervals = mCurrentSlicedBucket[eventKey];
		if (multiIntervals.size() < mFieldMatchers.size()) {
		VLOG("Resizing number of intervals to %d", (int)mFieldMatchers.size());
		multiIntervals.resize(mFieldMatchers.size());
		}

		if (mField > event.size()) {
		VLOG("Failed to extract value field %d from atom %s. %d", mField, event.ToString().c_str(),
		(int)event.size());
		for (int i = 0; i < (int)mFieldMatchers.size(); i++) {
		const Matcher& matcher = mFieldMatchers[i];
		Interval& interval = multiIntervals[i];
		interval.valueIndex = i;
		Value value;
		if (!getDoubleOrLong(event, matcher, value)) {
		VLOG("Failed to get value %d from event %s", i, event.ToString().c_str());
		return;
		}
		Value value = getDoubleOrLong(event.getValues()[mField - 1].mValue);

		if (mUseDiff) {
		// no base. just update base and return.
		@@ -509,9 +534,10 @@ void ValueMetricProducer::onMatchedLogEventInternalLocked(const size_t matcherIn
		interval.hasValue = true;
		}
		interval.sampleSize += 1;
		}

		// TODO: propgate proper values down stream when anomaly support doubles
		long wholeBucketVal = interval.value.long_value;
		long wholeBucketVal = multiIntervals[0].value.long_value;
		auto prev = mCurrentFullBucket.find(eventKey);
		if (prev != mCurrentFullBucket.end()) {
		wholeBucketVal += prev->second;
		@@ -540,7 +566,9 @@ void ValueMetricProducer::flushIfNeededLocked(const int64_t& eventTimeNs) {
		VLOG("Skipping forward %lld buckets", (long long)numBucketsForward);
		// take base again in future good bucket.
		for (auto& slice : mCurrentSlicedBucket) {
		slice.second.hasBase = false;
		for (auto& interval : slice.second) {
		interval.hasBase = false;
		}
		}
		}
		VLOG("metric %lld: new bucket start time: %lld", (long long)mMetricId,
		@@ -552,37 +580,38 @@ void ValueMetricProducer::flushCurrentBucketLocked(const int64_t& eventTimeNs) {
		(int)mCurrentSlicedBucket.size());
		int64_t fullBucketEndTimeNs = getCurrentBucketEndTimeNs();

		ValueBucket info;
		info.mBucketStartNs = mCurrentBucketStartTimeNs;
		if (eventTimeNs < fullBucketEndTimeNs) {
		info.mBucketEndNs = eventTimeNs;
		} else {
		info.mBucketEndNs = fullBucketEndTimeNs;
		}
		int64_t bucketEndTime = eventTimeNs < fullBucketEndTimeNs ? eventTimeNs : fullBucketEndTimeNs;

		if (info.mBucketEndNs - mCurrentBucketStartTimeNs >= mMinBucketSizeNs) {
		if (bucketEndTime - mCurrentBucketStartTimeNs >= mMinBucketSizeNs) {
		// The current bucket is large enough to keep.
		for (const auto& slice : mCurrentSlicedBucket) {
		if (slice.second.hasValue) {
		ValueBucket bucket;
		bucket.mBucketStartNs = mCurrentBucketStartTimeNs;
		bucket.mBucketEndNs = bucketEndTime;
		for (const auto& interval : slice.second) {
		if (interval.hasValue) {
		// skip the output if the diff is zero
		if (mSkipZeroDiffOutput && mUseDiff && slice.second.value.isZero()) {
		if (mSkipZeroDiffOutput && mUseDiff && interval.value.isZero()) {
		continue;
		}
		bucket.valueIndex.push_back(interval.valueIndex);
		if (mAggregationType != ValueMetric::AVG) {
		info.value = slice.second.value;
		bucket.values.push_back(interval.value);
		} else {
		double sum = slice.second.value.type == LONG
		? (double)slice.second.value.long_value
		: slice.second.value.double_value;
		info.value.setDouble(sum / slice.second.sampleSize);
		double sum = interval.value.type == LONG ? (double)interval.value.long_value
		: interval.value.double_value;
		bucket.values.push_back(Value((double)sum / interval.sampleSize));
		}
		}
		}
		// it will auto create new vector of ValuebucketInfo if the key is not found.
		if (bucket.valueIndex.size() > 0) {
		auto& bucketList = mPastBuckets[slice.first];
		bucketList.push_back(info);
		bucketList.push_back(bucket);
		}
		}
		} else {
		mSkippedBuckets.emplace_back(info.mBucketStartNs, info.mBucketEndNs);
		mSkippedBuckets.emplace_back(mCurrentBucketStartTimeNs, bucketEndTime);
		}

		if (eventTimeNs > fullBucketEndTimeNs) { // If full bucket, send to anomaly tracker.
		@@ -590,7 +619,7 @@ void ValueMetricProducer::flushCurrentBucketLocked(const int64_t& eventTimeNs) {
		if (mCurrentFullBucket.size() > 0) {
		for (const auto& slice : mCurrentSlicedBucket) {
		// TODO: fix this when anomaly can accept double values
		mCurrentFullBucket[slice.first] += slice.second.value.long_value;
		mCurrentFullBucket[slice.first] += slice.second[0].value.long_value;
		}
		for (const auto& slice : mCurrentFullBucket) {
		for (auto& tracker : mAnomalyTrackers) {
		@@ -606,7 +635,7 @@ void ValueMetricProducer::flushCurrentBucketLocked(const int64_t& eventTimeNs) {
		for (auto& tracker : mAnomalyTrackers) {
		if (tracker != nullptr) {
		// TODO: fix this when anomaly can accept double values
		tracker->addPastBucket(slice.first, slice.second.value.long_value,
		tracker->addPastBucket(slice.first, slice.second[0].value.long_value,
		mCurrentBucketNum);
		}
		}
		@@ -616,14 +645,16 @@ void ValueMetricProducer::flushCurrentBucketLocked(const int64_t& eventTimeNs) {
		// Accumulate partial bucket.
		for (const auto& slice : mCurrentSlicedBucket) {
		// TODO: fix this when anomaly can accept double values
		mCurrentFullBucket[slice.first] += slice.second.value.long_value;
		mCurrentFullBucket[slice.first] += slice.second[0].value.long_value;
		}
		}

		// Reset counters
		for (auto& slice : mCurrentSlicedBucket) {
		slice.second.hasValue = false;
		slice.second.sampleSize = 0;
		for (auto& interval : slice.second) {
		interval.hasValue = false;
		interval.sampleSize = 0;
		}
		}
		}

cmds/statsd/src/metrics/ValueMetricProducer.h

+8 −6

Original line number	Diff line number	Diff line
		@@ -34,7 +34,8 @@ namespace statsd {
		struct ValueBucket {
		int64_t mBucketStartNs;
		int64_t mBucketEndNs;
		Value value;
		std::vector<int> valueIndex;
		std::vector<Value> values;
		};

		class ValueMetricProducer : public virtual MetricProducer, public virtual PullDataReceiver {
		@@ -97,7 +98,8 @@ private:

		sp<StatsPullerManager> mPullerManager;

		const FieldMatcher mValueField;
		// Value fields for matching.
		std::vector<Matcher> mFieldMatchers;

		// tagId for pulled data. -1 if this is not pulled
		const int mPullTagId;
		@@ -105,10 +107,10 @@ private:
		// if this is pulled metric
		const bool mIsPulled;

		int mField;

		// internal state of a bucket.
		// internal state of an ongoing aggregation bucket.
		typedef struct {
		// Index in multi value aggregation.
		int valueIndex;
		// Holds current base value of the dimension. Take diff and update if necessary.
		Value base;
		// Whether there is a base to diff to.
		@@ -122,7 +124,7 @@ private:
		bool hasValue;
		} Interval;

		std::unordered_map<MetricDimensionKey, Interval> mCurrentSlicedBucket;
		std::unordered_map<MetricDimensionKey, std::vector<Interval>> mCurrentSlicedBucket;

		std::unordered_map<MetricDimensionKey, int64_t> mCurrentFullBucket;

cmds/statsd/src/stats_log.proto

+13 −3

Original line number	Diff line number	Diff line
		@@ -108,12 +108,22 @@ message ValueBucketInfo {

		optional int64 value = 3 [deprecated = true];

		oneof values {
		int64 value_long = 7;
		oneof single_value {
		int64 value_long = 7 [deprecated = true];

		double value_double = 8;
		double value_double = 8 [deprecated = true];
		}

		message Value {
		optional int32 index = 1;
		oneof value {
		int64 value_long = 2;
		double value_double = 3;
		}
		}

		repeated Value values = 9;

		optional int64 bucket_num = 4;

		optional int64 start_bucket_elapsed_millis = 5;

cmds/statsd/tests/e2e/ValueMetric_pull_e2e_test.cpp

+8 −8

Original line number	Diff line number	Diff line
		@@ -142,23 +142,23 @@ TEST(ValueMetricE2eTest, TestPulledEvents) {

		EXPECT_EQ(baseTimeNs + 2 * bucketSizeNs, data.bucket_info(0).start_bucket_elapsed_nanos());
		EXPECT_EQ(baseTimeNs + 3 * bucketSizeNs, data.bucket_info(0).end_bucket_elapsed_nanos());
		EXPECT_TRUE(data.bucket_info(0).has_value_long());
		EXPECT_EQ(1, data.bucket_info(0).values_size());

		EXPECT_EQ(baseTimeNs + 3 * bucketSizeNs, data.bucket_info(1).start_bucket_elapsed_nanos());
		EXPECT_EQ(baseTimeNs + 4 * bucketSizeNs, data.bucket_info(1).end_bucket_elapsed_nanos());
		EXPECT_TRUE(data.bucket_info(1).has_value_long());
		EXPECT_EQ(1, data.bucket_info(1).values_size());

		EXPECT_EQ(baseTimeNs + 4 * bucketSizeNs, data.bucket_info(2).start_bucket_elapsed_nanos());
		EXPECT_EQ(baseTimeNs + 5 * bucketSizeNs, data.bucket_info(2).end_bucket_elapsed_nanos());
		EXPECT_TRUE(data.bucket_info(2).has_value_long());
		EXPECT_EQ(1, data.bucket_info(2).values_size());

		EXPECT_EQ(baseTimeNs + 6 * bucketSizeNs, data.bucket_info(3).start_bucket_elapsed_nanos());
		EXPECT_EQ(baseTimeNs + 7 * bucketSizeNs, data.bucket_info(3).end_bucket_elapsed_nanos());
		EXPECT_TRUE(data.bucket_info(3).has_value_long());
		EXPECT_EQ(1, data.bucket_info(3).values_size());

		EXPECT_EQ(baseTimeNs + 7 * bucketSizeNs, data.bucket_info(4).start_bucket_elapsed_nanos());
		EXPECT_EQ(baseTimeNs + 8 * bucketSizeNs, data.bucket_info(4).end_bucket_elapsed_nanos());
		EXPECT_TRUE(data.bucket_info(4).has_value_long());
		EXPECT_EQ(1, data.bucket_info(4).values_size());
		}

		TEST(ValueMetricE2eTest, TestPulledEvents_LateAlarm) {
		@@ -249,15 +249,15 @@ TEST(ValueMetricE2eTest, TestPulledEvents_LateAlarm) {

		EXPECT_EQ(baseTimeNs + 2 * bucketSizeNs, data.bucket_info(0).start_bucket_elapsed_nanos());
		EXPECT_EQ(baseTimeNs + 3 * bucketSizeNs, data.bucket_info(0).end_bucket_elapsed_nanos());
		EXPECT_TRUE(data.bucket_info(0).has_value_long());
		EXPECT_EQ(1, data.bucket_info(0).values_size());

		EXPECT_EQ(baseTimeNs + 8 * bucketSizeNs, data.bucket_info(1).start_bucket_elapsed_nanos());
		EXPECT_EQ(baseTimeNs + 9 * bucketSizeNs, data.bucket_info(1).end_bucket_elapsed_nanos());
		EXPECT_TRUE(data.bucket_info(1).has_value_long());
		EXPECT_EQ(1, data.bucket_info(1).values_size());

		EXPECT_EQ(baseTimeNs + 9 * bucketSizeNs, data.bucket_info(2).start_bucket_elapsed_nanos());
		EXPECT_EQ(baseTimeNs + 10 * bucketSizeNs, data.bucket_info(2).end_bucket_elapsed_nanos());
		EXPECT_TRUE(data.bucket_info(2).has_value_long());
		EXPECT_EQ(1, data.bucket_info(2).values_size());
		}

		#else

cmds/statsd/tests/metrics/ValueMetricProducer_test.cpp

+54 −54

File changed.

Preview size limit exceeded, changes collapsed.