Donate to e Foundation | Murena handsets with /e/OS | Own a part of Murena! Learn more

Commit 34ea1103 authored by Yangster-mac's avatar Yangster-mac
Browse files

Extend gauge metric to support memory metric. 

Test: statd unit test passed.

Test: statsd unit test passed
Change-Id: I2e3f26563678ae77d44afe168454b6d1ea449f3a
parent 752d7ca1

cmds/statsd/src/metrics/GaugeMetricProducer.cpp

+53 −21
Original line number Diff line number Diff line
@@ -58,6 +58,7 @@ const int FIELD_ID_BUCKET_INFO = 3; 
const int FIELD_ID_START_BUCKET_NANOS = 1;

const int FIELD_ID_END_BUCKET_NANOS = 2;

const int FIELD_ID_ATOM = 3;

const int FIELD_ID_TIMESTAMP = 4;



GaugeMetricProducer::GaugeMetricProducer(const ConfigKey& key, const GaugeMetric& metric,

                                         const int conditionIndex,
@@ -67,7 +68,7 @@ GaugeMetricProducer::GaugeMetricProducer(const ConfigKey& key, const GaugeMetric 
    : MetricProducer(metric.id(), key, startTimeNs, conditionIndex, wizard),

      mStatsPullerManager(statsPullerManager),

      mPullTagId(pullTagId) {

    mCurrentSlicedBucket = std::make_shared<DimToGaugeFieldsMap>();

    mCurrentSlicedBucket = std::make_shared<DimToGaugeAtomsMap>();

    mCurrentSlicedBucketForAnomaly = std::make_shared<DimToValMap>();

    int64_t bucketSizeMills = 0;

    if (metric.has_bucket()) {
@@ -77,6 +78,7 @@ GaugeMetricProducer::GaugeMetricProducer(const ConfigKey& key, const GaugeMetric 
    }

    mBucketSizeNs = bucketSizeMills * 1000000;



    mSamplingType = metric.sampling_type();

    mFieldFilter = metric.gauge_fields_filter();



    // TODO: use UidMap if uid->pkg_name is required
@@ -89,7 +91,7 @@ GaugeMetricProducer::GaugeMetricProducer(const ConfigKey& key, const GaugeMetric 
    }



    // Kicks off the puller immediately.

    if (mPullTagId != -1) {

    if (mPullTagId != -1 && mSamplingType == GaugeMetric::RANDOM_ONE_SAMPLE) {

        mStatsPullerManager->RegisterReceiver(mPullTagId, this, bucketSizeMills);

    }
@@ -154,12 +156,23 @@ void GaugeMetricProducer::onDumpReportLocked(const uint64_t dumpTimeNs, 
                               (long long)bucket.mBucketStartNs);

            protoOutput->write(FIELD_TYPE_INT64 | FIELD_ID_END_BUCKET_NANOS,

                               (long long)bucket.mBucketEndNs);

            long long atomToken = protoOutput->start(FIELD_TYPE_MESSAGE | FIELD_ID_ATOM);

            writeFieldValueTreeToStream(*bucket.mGaugeFields, protoOutput);

            protoOutput->end(atomToken);



            if (!bucket.mGaugeAtoms.empty()) {

                long long atomsToken =

                    protoOutput->start(FIELD_TYPE_MESSAGE | FIELD_COUNT_REPEATED | FIELD_ID_ATOM);

                for (const auto& atom : bucket.mGaugeAtoms) {

                    writeFieldValueTreeToStream(*atom.mFields, protoOutput);

                }

                protoOutput->end(atomsToken);



                for (const auto& atom : bucket.mGaugeAtoms) {

                    protoOutput->write(FIELD_TYPE_INT64 | FIELD_COUNT_REPEATED | FIELD_ID_TIMESTAMP,

                                       (long long)atom.mTimestamps);

                }

            }

            protoOutput->end(bucketInfoToken);

            VLOG("\t bucket [%lld - %lld] includes %d gauge fields.", (long long)bucket.mBucketStartNs,

                 (long long)bucket.mBucketEndNs, (int)bucket.mGaugeFields->size());

            VLOG("\t bucket [%lld - %lld] includes %d atoms.", (long long)bucket.mBucketStartNs,

                 (long long)bucket.mBucketEndNs, (int)bucket.mGaugeAtoms.size());

        }

        protoOutput->end(wrapperToken);

    }
@@ -181,14 +194,26 @@ void GaugeMetricProducer::onConditionChangedLocked(const bool conditionMet, 
    if (mPullTagId == -1) {

        return;

    }

    // No need to pull again. Either scheduled pull or condition on true happened

    if (!mCondition) {

        return;



    bool triggerPuller = false;

    switch(mSamplingType) {

        // When the metric wants to do random sampling and there is already one gauge atom for the

        // current bucket, do not do it again.

        case GaugeMetric::RANDOM_ONE_SAMPLE: {

            triggerPuller = mCondition && mCurrentSlicedBucket->empty();

            break;

        }

        case GaugeMetric::ALL_CONDITION_CHANGES: {

            triggerPuller = true;

            break;

        }

    // Already have gauge metric for the current bucket, do not do it again.

    if (mCurrentSlicedBucket->size() > 0) {

        default:

            break;

    }

    if (!triggerPuller) {

        return;

    }



    vector<std::shared_ptr<LogEvent>> allData;

    if (!mStatsPullerManager->Pull(mPullTagId, &allData)) {

        ALOGE("Stats puller failed for tag: %d", mPullTagId);
@@ -257,20 +282,24 @@ void GaugeMetricProducer::onMatchedLogEventInternalLocked( 
    }

    flushIfNeededLocked(eventTimeNs);



    // For gauge metric, we just simply use the first gauge in the given bucket.

    if (mCurrentSlicedBucket->find(eventKey) != mCurrentSlicedBucket->end()) {

    // When gauge metric wants to randomly sample the output atom, we just simply use the first

    // gauge in the given bucket.

    if (mCurrentSlicedBucket->find(eventKey) != mCurrentSlicedBucket->end() &&

        mSamplingType == GaugeMetric::RANDOM_ONE_SAMPLE) {

        return;

    }

    std::shared_ptr<FieldValueMap> gaugeFields = getGaugeFields(event);

    if (hitGuardRailLocked(eventKey)) {

        return;

    }

    (*mCurrentSlicedBucket)[eventKey] = gaugeFields;

    GaugeAtom gaugeAtom;

    gaugeAtom.mFields = getGaugeFields(event);

    gaugeAtom.mTimestamps = eventTimeNs;

    (*mCurrentSlicedBucket)[eventKey].push_back(gaugeAtom);

    // Anomaly detection on gauge metric only works when there is one numeric

    // field specified.

    if (mAnomalyTrackers.size() > 0) {

        if (gaugeFields->size() == 1) {

            const DimensionsValue& dimensionsValue = gaugeFields->begin()->second;

        if (gaugeAtom.mFields->size() == 1) {

            const DimensionsValue& dimensionsValue = gaugeAtom.mFields->begin()->second;

            long gaugeVal = 0;

            if (dimensionsValue.has_value_int()) {

                gaugeVal = (long)dimensionsValue.value_int();
@@ -289,7 +318,10 @@ void GaugeMetricProducer::updateCurrentSlicedBucketForAnomaly() { 
    mCurrentSlicedBucketForAnomaly->clear();

    status_t err = NO_ERROR;

    for (const auto& slice : *mCurrentSlicedBucket) {

        const DimensionsValue& dimensionsValue = slice.second->begin()->second;

        if (slice.second.empty() || slice.second.front().mFields->empty()) {

            continue;

        }

        const DimensionsValue& dimensionsValue = slice.second.front().mFields->begin()->second;

        long gaugeVal = 0;

        if (dimensionsValue.has_value_int()) {

            gaugeVal = (long)dimensionsValue.value_int();
@@ -318,7 +350,7 @@ void GaugeMetricProducer::flushIfNeededLocked(const uint64_t& eventTimeNs) { 
    info.mBucketNum = mCurrentBucketNum;



    for (const auto& slice : *mCurrentSlicedBucket) {

        info.mGaugeFields = slice.second;

        info.mGaugeAtoms = slice.second;

        auto& bucketList = mPastBuckets[slice.first];

        bucketList.push_back(info);

        VLOG("gauge metric %lld, dump key value: %s",
@@ -334,7 +366,7 @@ void GaugeMetricProducer::flushIfNeededLocked(const uint64_t& eventTimeNs) { 
    }



    mCurrentSlicedBucketForAnomaly = std::make_shared<DimToValMap>();

    mCurrentSlicedBucket = std::make_shared<DimToGaugeFieldsMap>();

    mCurrentSlicedBucket = std::make_shared<DimToGaugeAtomsMap>();



    // Adjusts the bucket start time

    int64_t numBucketsForward = (eventTimeNs - mCurrentBucketStartTimeNs) / mBucketSizeNs;

cmds/statsd/src/metrics/GaugeMetricProducer.h

+12 −5
Original line number Diff line number Diff line
@@ -32,15 +32,20 @@ namespace android { 
namespace os {

namespace statsd {



struct GaugeAtom {

    std::shared_ptr<FieldValueMap> mFields;

    int64_t mTimestamps;

};



struct GaugeBucket {

    int64_t mBucketStartNs;

    int64_t mBucketEndNs;

    std::shared_ptr<FieldValueMap> mGaugeFields;

    std::vector<GaugeAtom> mGaugeAtoms;

    uint64_t mBucketNum;

};



typedef std::unordered_map<HashableDimensionKey, std::shared_ptr<FieldValueMap>>

    DimToGaugeFieldsMap;

typedef std::unordered_map<HashableDimensionKey, std::vector<GaugeAtom>>

    DimToGaugeAtomsMap;



// This gauge metric producer first register the puller to automatically pull the gauge at the

// beginning of each bucket. If the condition is met, insert it to the bucket info. Otherwise
@@ -48,7 +53,7 @@ typedef std::unordered_map<HashableDimensionKey, std::shared_ptr<FieldValueMap>> 
// producer always reports the guage at the earliest time of the bucket when the condition is met.

class GaugeMetricProducer : public virtual MetricProducer, public virtual PullDataReceiver {

public:

    GaugeMetricProducer(const ConfigKey& key, const GaugeMetric& countMetric,

    GaugeMetricProducer(const ConfigKey& key, const GaugeMetric& gaugeMetric,

                        const int conditionIndex, const sp<ConditionWizard>& wizard,

                        const int pullTagId, const int64_t startTimeNs);
@@ -97,7 +102,7 @@ private: 
    std::unordered_map<HashableDimensionKey, std::vector<GaugeBucket>> mPastBuckets;



    // The current bucket.

    std::shared_ptr<DimToGaugeFieldsMap> mCurrentSlicedBucket;

    std::shared_ptr<DimToGaugeAtomsMap> mCurrentSlicedBucket;



    // The current bucket for anomaly detection.

    std::shared_ptr<DimToValMap> mCurrentSlicedBucketForAnomaly;
@@ -108,6 +113,8 @@ private: 
    // Whitelist of fields to report. Empty means all are reported.

    FieldFilter mFieldFilter;



    GaugeMetric::SamplingType mSamplingType;



    // apply a whitelist on the original input

    std::shared_ptr<FieldValueMap> getGaugeFields(const LogEvent& event);

cmds/statsd/src/stats_log.proto

+3 −1
Original line number Diff line number Diff line
@@ -100,7 +100,9 @@ message GaugeBucketInfo { 


  optional int64 end_bucket_nanos = 2;



  optional Atom atom = 3;

  repeated Atom atom = 3;



  repeated int64 timestamp_nanos = 4;

}



message GaugeMetricData {

cmds/statsd/src/statsd_config.proto

+6 −0
Original line number Diff line number Diff line
@@ -222,6 +222,12 @@ message GaugeMetric { 
  optional TimeUnit bucket = 6;



  repeated MetricConditionLink links = 7;



  enum SamplingType {

    RANDOM_ONE_SAMPLE = 1;

    ALL_CONDITION_CHANGES = 2;

  }

  optional SamplingType sampling_type = 9 [default = RANDOM_ONE_SAMPLE] ;

}



message ValueMetric {

cmds/statsd/tests/e2e/GaugeMetric_e2e_test.cpp

+16 −12
Original line number Diff line number Diff line
@@ -153,23 +153,26 @@ TEST(GaugeMetricE2eTest, TestMultipleFieldsForPushedEvent) { 
    EXPECT_EQ(data.dimensions_in_what().value_tuple().dimensions_value(0).field(), 1 /* uid field */);

    EXPECT_EQ(data.dimensions_in_what().value_tuple().dimensions_value(0).value_int(), appUid1);

    EXPECT_EQ(data.bucket_info_size(), 3);

    EXPECT_EQ(data.bucket_info(0).atom_size(), 1);

    EXPECT_EQ(data.bucket_info(0).start_bucket_nanos(), bucketStartTimeNs);

    EXPECT_EQ(data.bucket_info(0).end_bucket_nanos(), bucketStartTimeNs + bucketSizeNs);

    EXPECT_EQ(data.bucket_info(0).atom().app_start_changed().type(), AppStartChanged::HOT);

    EXPECT_EQ(data.bucket_info(0).atom().app_start_changed().activity_name(), "activity_name2");

    EXPECT_EQ(data.bucket_info(0).atom().app_start_changed().activity_start_msec(), 102L);

    EXPECT_EQ(data.bucket_info(0).atom(0).app_start_changed().type(), AppStartChanged::HOT);

    EXPECT_EQ(data.bucket_info(0).atom(0).app_start_changed().activity_name(), "activity_name2");

    EXPECT_EQ(data.bucket_info(0).atom(0).app_start_changed().activity_start_msec(), 102L);



    EXPECT_EQ(data.bucket_info(1).atom_size(), 1);

    EXPECT_EQ(data.bucket_info(1).start_bucket_nanos(), bucketStartTimeNs + bucketSizeNs);

    EXPECT_EQ(data.bucket_info(1).end_bucket_nanos(), bucketStartTimeNs + 2 * bucketSizeNs);

    EXPECT_EQ(data.bucket_info(1).atom().app_start_changed().type(), AppStartChanged::WARM);

    EXPECT_EQ(data.bucket_info(1).atom().app_start_changed().activity_name(), "activity_name4");

    EXPECT_EQ(data.bucket_info(1).atom().app_start_changed().activity_start_msec(), 104L);

    EXPECT_EQ(data.bucket_info(1).atom(0).app_start_changed().type(), AppStartChanged::WARM);

    EXPECT_EQ(data.bucket_info(1).atom(0).app_start_changed().activity_name(), "activity_name4");

    EXPECT_EQ(data.bucket_info(1).atom(0).app_start_changed().activity_start_msec(), 104L);



    EXPECT_EQ(data.bucket_info(2).atom_size(), 1);

    EXPECT_EQ(data.bucket_info(2).start_bucket_nanos(), bucketStartTimeNs + 2 * bucketSizeNs);

    EXPECT_EQ(data.bucket_info(2).end_bucket_nanos(), bucketStartTimeNs + 3 * bucketSizeNs);

    EXPECT_EQ(data.bucket_info(2).atom().app_start_changed().type(), AppStartChanged::COLD);

    EXPECT_EQ(data.bucket_info(2).atom().app_start_changed().activity_name(), "activity_name5");

    EXPECT_EQ(data.bucket_info(2).atom().app_start_changed().activity_start_msec(), 105L);

    EXPECT_EQ(data.bucket_info(2).atom(0).app_start_changed().type(), AppStartChanged::COLD);

    EXPECT_EQ(data.bucket_info(2).atom(0).app_start_changed().activity_name(), "activity_name5");

    EXPECT_EQ(data.bucket_info(2).atom(0).app_start_changed().activity_start_msec(), 105L);



    data = gaugeMetrics.data(1);
@@ -178,11 +181,12 @@ TEST(GaugeMetricE2eTest, TestMultipleFieldsForPushedEvent) { 
    EXPECT_EQ(data.dimensions_in_what().value_tuple().dimensions_value(0).field(), 1 /* uid field */);

    EXPECT_EQ(data.dimensions_in_what().value_tuple().dimensions_value(0).value_int(), appUid2);

    EXPECT_EQ(data.bucket_info_size(), 1);

    EXPECT_EQ(data.bucket_info(0).atom_size(), 1);

    EXPECT_EQ(data.bucket_info(0).start_bucket_nanos(), bucketStartTimeNs + 2 * bucketSizeNs);

    EXPECT_EQ(data.bucket_info(0).end_bucket_nanos(), bucketStartTimeNs + 3 * bucketSizeNs);

    EXPECT_EQ(data.bucket_info(0).atom().app_start_changed().type(), AppStartChanged::COLD);

    EXPECT_EQ(data.bucket_info(0).atom().app_start_changed().activity_name(), "activity_name7");

    EXPECT_EQ(data.bucket_info(0).atom().app_start_changed().activity_start_msec(), 201L);

    EXPECT_EQ(data.bucket_info(0).atom(0).app_start_changed().type(), AppStartChanged::COLD);

    EXPECT_EQ(data.bucket_info(0).atom(0).app_start_changed().activity_name(), "activity_name7");

    EXPECT_EQ(data.bucket_info(0).atom(0).app_start_changed().activity_start_msec(), 201L);

}



#else