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

Commit 288c6000 authored by Yao Chen's avatar Yao Chen
Browse files

Only create ProtoOutputStream when onGetData() is called. 

The exception is EventMetricProducer. Each EventMetricProducer will still have a ProtoOutputStream
Because LogEvent comes as a fixed 4K, it's more memory efficient to have an 8k ProtoOutputStream for
storing the events.

Also removed finish() api in MetricProducer, which was intended to use with Dropbox.

Test: statsd_test & dogfood app
Bug: 70393808
Change-Id: I2efe4ecc76a88060a9aa5eb49d1fa6ea60bc5da8
parent 32c21619

cmds/statsd/src/StatsLogProcessor.cpp

+5 −11
Original line number Original line Diff line number Diff line
@@ -111,9 +111,7 @@ void StatsLogProcessor::OnConfigUpdated(const ConfigKey& key, const StatsdConfig 
    unique_ptr<MetricsManager> newMetricsManager = std::make_unique<MetricsManager>(key, config);

    unique_ptr<MetricsManager> newMetricsManager = std::make_unique<MetricsManager>(key, config);





    auto it = mMetricsManagers.find(key);

    auto it = mMetricsManagers.find(key);

    if (it != mMetricsManagers.end()) {

    if (it == mMetricsManagers.end() && mMetricsManagers.size() > StatsdStats::kMaxConfigCount) {

        it->second->finish();

    } else if (mMetricsManagers.size() > StatsdStats::kMaxConfigCount) {

        ALOGE("Can't accept more configs!");

        ALOGE("Can't accept more configs!");

        return;

        return;

    }

    }
@@ -167,11 +165,7 @@ void StatsLogProcessor::onDumpReport(const ConfigKey& key, vector<uint8_t>* outD 




    // First, fill in ConfigMetricsReport using current data on memory, which

    // First, fill in ConfigMetricsReport using current data on memory, which

    // starts from filling in StatsLogReport's.

    // starts from filling in StatsLogReport's.

    for (auto& m : it->second->onDumpReport()) {

    it->second->onDumpReport(&proto);

        // Add each vector of StatsLogReport into a repeated field.

        proto.write(FIELD_TYPE_MESSAGE | FIELD_COUNT_REPEATED | FIELD_ID_METRICS,

                    reinterpret_cast<char*>(m.get()->data()), m.get()->size());

    }





    // Fill in UidMap.

    // Fill in UidMap.

    auto uidMap = mUidMap->getOutput(key);

    auto uidMap = mUidMap->getOutput(key);
@@ -205,7 +199,6 @@ void StatsLogProcessor::onDumpReport(const ConfigKey& key, vector<uint8_t>* outD 
void StatsLogProcessor::OnConfigRemoved(const ConfigKey& key) {

void StatsLogProcessor::OnConfigRemoved(const ConfigKey& key) {

    auto it = mMetricsManagers.find(key);

    auto it = mMetricsManagers.find(key);

    if (it != mMetricsManagers.end()) {

    if (it != mMetricsManagers.end()) {

        it->second->finish();

        mMetricsManagers.erase(it);

        mMetricsManagers.erase(it);

        mUidMap->OnConfigRemoved(key);

        mUidMap->OnConfigRemoved(key);

    }

    }
@@ -231,8 +224,9 @@ void StatsLogProcessor::flushIfNecessary(uint64_t timestampNs, const ConfigKey& 
    mLastByteSizeTimes[key] = timestampNs;

    mLastByteSizeTimes[key] = timestampNs;

    if (totalBytes >

    if (totalBytes >

        StatsdStats::kMaxMetricsBytesPerConfig) {  // Too late. We need to start clearing data.

        StatsdStats::kMaxMetricsBytesPerConfig) {  // Too late. We need to start clearing data.

        // We ignore the return value so we force each metric producer to clear its contents.

        // TODO(b/70571383): By 12/15/2017 add API to drop data directly

        metricsManager.onDumpReport();

        ProtoOutputStream proto;

        metricsManager.onDumpReport(&proto);

        StatsdStats::getInstance().noteDataDropped(key);

        StatsdStats::getInstance().noteDataDropped(key);

        VLOG("StatsD had to toss out metrics for %s", key.ToString().c_str());

        VLOG("StatsD had to toss out metrics for %s", key.ToString().c_str());

    } else if (totalBytes > .9 * StatsdStats::kMaxMetricsBytesPerConfig) {

    } else if (totalBytes > .9 * StatsdStats::kMaxMetricsBytesPerConfig) {

cmds/statsd/src/metrics/CountMetricProducer.cpp

+28 −45
Original line number Original line Diff line number Diff line
@@ -83,8 +83,6 @@ CountMetricProducer::CountMetricProducer(const ConfigKey& key, const CountMetric 
        mConditionSliced = true;

        mConditionSliced = true;

    }

    }





    startNewProtoOutputStreamLocked(mStartTimeNs);



    VLOG("metric %s created. bucket size %lld start_time: %lld", metric.name().c_str(),

    VLOG("metric %s created. bucket size %lld start_time: %lld", metric.name().c_str(),

         (long long)mBucketSizeNs, (long long)mStartTimeNs);

         (long long)mBucketSizeNs, (long long)mStartTimeNs);

}

}
@@ -93,27 +91,18 @@ CountMetricProducer::~CountMetricProducer() { 
    VLOG("~CountMetricProducer() called");

    VLOG("~CountMetricProducer() called");

}

}





void CountMetricProducer::startNewProtoOutputStreamLocked(long long startTime) {

    mProto = std::make_unique<ProtoOutputStream>();

    mProto->write(FIELD_TYPE_STRING | FIELD_ID_NAME, mMetric.name());

    mProto->write(FIELD_TYPE_INT64 | FIELD_ID_START_REPORT_NANOS, startTime);

    mProtoToken = mProto->start(FIELD_TYPE_MESSAGE | FIELD_ID_COUNT_METRICS);

}



void CountMetricProducer::finish() {

}



void CountMetricProducer::onSlicedConditionMayChangeLocked(const uint64_t eventTime) {

void CountMetricProducer::onSlicedConditionMayChangeLocked(const uint64_t eventTime) {

    VLOG("Metric %s onSlicedConditionMayChange", mMetric.name().c_str());

    VLOG("Metric %s onSlicedConditionMayChange", mMetric.name().c_str());

}

}





std::unique_ptr<std::vector<uint8_t>> CountMetricProducer::onDumpReportLocked() {

void CountMetricProducer::onDumpReportLocked(const uint64_t dumpTimeNs,

    long long endTime = time(nullptr) * NS_PER_SEC;

                                             ProtoOutputStream* protoOutput) {

    flushIfNeededLocked(dumpTimeNs);



    protoOutput->write(FIELD_TYPE_STRING | FIELD_ID_NAME, mMetric.name());

    protoOutput->write(FIELD_TYPE_INT64 | FIELD_ID_START_REPORT_NANOS, (long long)mStartTimeNs);

    long long protoToken = protoOutput->start(FIELD_TYPE_MESSAGE | FIELD_ID_COUNT_METRICS);





    // Dump current bucket if it's stale.

    // If current bucket is still on-going, don't force dump current bucket.

    // In finish(), We can force dump current bucket.

    flushIfNeededLocked(endTime);

    VLOG("metric %s dump report now...", mMetric.name().c_str());

    VLOG("metric %s dump report now...", mMetric.name().c_str());





    for (const auto& counter : mPastBuckets) {

    for (const auto& counter : mPastBuckets) {
@@ -125,52 +114,46 @@ std::unique_ptr<std::vector<uint8_t>> CountMetricProducer::onDumpReportLocked() 
            continue;

            continue;

        }

        }

        long long wrapperToken =

        long long wrapperToken =

                mProto->start(FIELD_TYPE_MESSAGE | FIELD_COUNT_REPEATED | FIELD_ID_DATA);

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





        // First fill dimension (KeyValuePairs).

        // First fill dimension (KeyValuePairs).

        for (const auto& kv : it->second) {

        for (const auto& kv : it->second) {

            long long dimensionToken =

            long long dimensionToken = protoOutput->start(

                    mProto->start(FIELD_TYPE_MESSAGE | FIELD_COUNT_REPEATED | FIELD_ID_DIMENSION);

                    FIELD_TYPE_MESSAGE | FIELD_COUNT_REPEATED | FIELD_ID_DIMENSION);

            mProto->write(FIELD_TYPE_INT32 | FIELD_ID_KEY, kv.key());

            protoOutput->write(FIELD_TYPE_INT32 | FIELD_ID_KEY, kv.key());

            if (kv.has_value_str()) {

            if (kv.has_value_str()) {

                mProto->write(FIELD_TYPE_STRING | FIELD_ID_VALUE_STR, kv.value_str());

                protoOutput->write(FIELD_TYPE_STRING | FIELD_ID_VALUE_STR, kv.value_str());

            } else if (kv.has_value_int()) {

            } else if (kv.has_value_int()) {

                mProto->write(FIELD_TYPE_INT64 | FIELD_ID_VALUE_INT, kv.value_int());

                protoOutput->write(FIELD_TYPE_INT64 | FIELD_ID_VALUE_INT, kv.value_int());

            } else if (kv.has_value_bool()) {

            } else if (kv.has_value_bool()) {

                mProto->write(FIELD_TYPE_BOOL | FIELD_ID_VALUE_BOOL, kv.value_bool());

                protoOutput->write(FIELD_TYPE_BOOL | FIELD_ID_VALUE_BOOL, kv.value_bool());

            } else if (kv.has_value_float()) {

            } else if (kv.has_value_float()) {

                mProto->write(FIELD_TYPE_FLOAT | FIELD_ID_VALUE_FLOAT, kv.value_float());

                protoOutput->write(FIELD_TYPE_FLOAT | FIELD_ID_VALUE_FLOAT, kv.value_float());

            }

            }

            mProto->end(dimensionToken);

            protoOutput->end(dimensionToken);

        }

        }





        // Then fill bucket_info (CountBucketInfo).

        // Then fill bucket_info (CountBucketInfo).

        for (const auto& bucket : counter.second) {

        for (const auto& bucket : counter.second) {

            long long bucketInfoToken =

            long long bucketInfoToken = protoOutput->start(

                    mProto->start(FIELD_TYPE_MESSAGE | FIELD_COUNT_REPEATED | FIELD_ID_BUCKET_INFO);

                    FIELD_TYPE_MESSAGE | FIELD_COUNT_REPEATED | FIELD_ID_BUCKET_INFO);

            mProto->write(FIELD_TYPE_INT64 | FIELD_ID_START_BUCKET_NANOS,

            protoOutput->write(FIELD_TYPE_INT64 | FIELD_ID_START_BUCKET_NANOS,

                               (long long)bucket.mBucketStartNs);

                               (long long)bucket.mBucketStartNs);

            mProto->write(FIELD_TYPE_INT64 | FIELD_ID_END_BUCKET_NANOS,

            protoOutput->write(FIELD_TYPE_INT64 | FIELD_ID_END_BUCKET_NANOS,

                               (long long)bucket.mBucketEndNs);

                               (long long)bucket.mBucketEndNs);

            mProto->write(FIELD_TYPE_INT64 | FIELD_ID_COUNT, (long long)bucket.mCount);

            protoOutput->write(FIELD_TYPE_INT64 | FIELD_ID_COUNT, (long long)bucket.mCount);

            mProto->end(bucketInfoToken);

            protoOutput->end(bucketInfoToken);

            VLOG("\t bucket [%lld - %lld] count: %lld", (long long)bucket.mBucketStartNs,

            VLOG("\t bucket [%lld - %lld] count: %lld", (long long)bucket.mBucketStartNs,

                 (long long)bucket.mBucketEndNs, (long long)bucket.mCount);

                 (long long)bucket.mBucketEndNs, (long long)bucket.mCount);

        }

        }

        mProto->end(wrapperToken);

        protoOutput->end(wrapperToken);

    }

    }





    mProto->end(mProtoToken);

    protoOutput->end(protoToken);

    mProto->write(FIELD_TYPE_INT64 | FIELD_ID_END_REPORT_NANOS,

    protoOutput->write(FIELD_TYPE_INT64 | FIELD_ID_END_REPORT_NANOS, (long long)dumpTimeNs);

                  (long long)mCurrentBucketStartTimeNs);





    VLOG("metric %s dump report now...", mMetric.name().c_str());

    std::unique_ptr<std::vector<uint8_t>> buffer = serializeProtoLocked();



    startNewProtoOutputStreamLocked(endTime);

    mPastBuckets.clear();

    mPastBuckets.clear();



    mStartTimeNs = mCurrentBucketStartTimeNs;

    return buffer;





    // TODO: Clear mDimensionKeyMap once the report is dumped.

    // TODO: Clear mDimensionKeyMap once the report is dumped.

}

}

cmds/statsd/src/metrics/CountMetricProducer.h

+2 −7
Original line number Original line Diff line number Diff line
@@ -48,8 +48,6 @@ public: 




    virtual ~CountMetricProducer();

    virtual ~CountMetricProducer();





    void finish() override;



    // TODO: Implement this later.

    // TODO: Implement this later.

    virtual void notifyAppUpgrade(const string& apk, const int uid, const int64_t version)

    virtual void notifyAppUpgrade(const string& apk, const int uid, const int64_t version)

            override{};

            override{};
@@ -63,8 +61,8 @@ protected: 
            const LogEvent& event, bool scheduledPull) override;

            const LogEvent& event, bool scheduledPull) override;





private:

private:

    // TODO: Pass a timestamp as a parameter in onDumpReport.

    void onDumpReportLocked(const uint64_t dumpTimeNs,

    std::unique_ptr<std::vector<uint8_t>> onDumpReportLocked() override;

                            android::util::ProtoOutputStream* protoOutput) override;





    // Internal interface to handle condition change.

    // Internal interface to handle condition change.

    void onConditionChangedLocked(const bool conditionMet, const uint64_t eventTime) override;

    void onConditionChangedLocked(const bool conditionMet, const uint64_t eventTime) override;
@@ -78,9 +76,6 @@ private: 
    // Util function to flush the old packet.

    // Util function to flush the old packet.

    void flushIfNeededLocked(const uint64_t& newEventTime);

    void flushIfNeededLocked(const uint64_t& newEventTime);





    // Util function to init/reset the proto output stream.

    void startNewProtoOutputStreamLocked(long long timestamp);



    const CountMetric mMetric;

    const CountMetric mMetric;





    // TODO: Add a lock to mPastBuckets.

    // TODO: Add a lock to mPastBuckets.

cmds/statsd/src/metrics/DurationMetricProducer.cpp

+28 −44
Original line number Original line Diff line number Diff line
@@ -93,8 +93,6 @@ DurationMetricProducer::DurationMetricProducer(const ConfigKey& key, const Durat 
        mConditionSliced = true;

        mConditionSliced = true;

    }

    }





    startNewProtoOutputStreamLocked(mStartTimeNs);



    VLOG("metric %s created. bucket size %lld start_time: %lld", metric.name().c_str(),

    VLOG("metric %s created. bucket size %lld start_time: %lld", metric.name().c_str(),

         (long long)mBucketSizeNs, (long long)mStartTimeNs);

         (long long)mBucketSizeNs, (long long)mStartTimeNs);

}

}
@@ -114,13 +112,6 @@ sp<AnomalyTracker> DurationMetricProducer::createAnomalyTracker(const Alert &ale 
    return new AnomalyTracker(alert, mConfigKey);

    return new AnomalyTracker(alert, mConfigKey);

}

}





void DurationMetricProducer::startNewProtoOutputStreamLocked(long long startTime) {

    mProto = std::make_unique<ProtoOutputStream>();

    mProto->write(FIELD_TYPE_STRING | FIELD_ID_NAME, mMetric.name());

    mProto->write(FIELD_TYPE_INT64 | FIELD_ID_START_REPORT_NANOS, startTime);

    mProtoToken = mProto->start(FIELD_TYPE_MESSAGE | FIELD_ID_DURATION_METRICS);

}



unique_ptr<DurationTracker> DurationMetricProducer::createDurationTracker(

unique_ptr<DurationTracker> DurationMetricProducer::createDurationTracker(

        const HashableDimensionKey& eventKey) const {

        const HashableDimensionKey& eventKey) const {

    switch (mMetric.aggregation_type()) {

    switch (mMetric.aggregation_type()) {
@@ -135,11 +126,6 @@ unique_ptr<DurationTracker> DurationMetricProducer::createDurationTracker( 
    }

    }

}

}





void DurationMetricProducer::finish() {

    // TODO: write the StatsLogReport to dropbox using

    // DropboxWriter.

}



void DurationMetricProducer::onSlicedConditionMayChangeLocked(const uint64_t eventTime) {

void DurationMetricProducer::onSlicedConditionMayChangeLocked(const uint64_t eventTime) {

    VLOG("Metric %s onSlicedConditionMayChange", mMetric.name().c_str());

    VLOG("Metric %s onSlicedConditionMayChange", mMetric.name().c_str());

    flushIfNeededLocked(eventTime);

    flushIfNeededLocked(eventTime);
@@ -161,13 +147,14 @@ void DurationMetricProducer::onConditionChangedLocked(const bool conditionMet, 
    }

    }

}

}





std::unique_ptr<std::vector<uint8_t>> DurationMetricProducer::onDumpReportLocked() {

void DurationMetricProducer::onDumpReportLocked(const uint64_t dumpTimeNs,

    long long endTime = time(nullptr) * NS_PER_SEC;

                                                ProtoOutputStream* protoOutput) {

    flushIfNeededLocked(dumpTimeNs);



    protoOutput->write(FIELD_TYPE_STRING | FIELD_ID_NAME, mMetric.name());

    protoOutput->write(FIELD_TYPE_INT64 | FIELD_ID_START_REPORT_NANOS, (long long)mStartTimeNs);

    long long protoToken = protoOutput->start(FIELD_TYPE_MESSAGE | FIELD_ID_DURATION_METRICS);





    // Dump current bucket if it's stale.

    // If current bucket is still on-going, don't force dump current bucket.

    // In finish(), We can force dump current bucket.

    flushIfNeededLocked(endTime);

    VLOG("metric %s dump report now...", mMetric.name().c_str());

    VLOG("metric %s dump report now...", mMetric.name().c_str());





    for (const auto& pair : mPastBuckets) {

    for (const auto& pair : mPastBuckets) {
@@ -180,49 +167,46 @@ std::unique_ptr<std::vector<uint8_t>> DurationMetricProducer::onDumpReportLocked 
        }

        }





        long long wrapperToken =

        long long wrapperToken =

                mProto->start(FIELD_TYPE_MESSAGE | FIELD_COUNT_REPEATED | FIELD_ID_DATA);

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





        // First fill dimension (KeyValuePairs).

        // First fill dimension (KeyValuePairs).

        for (const auto& kv : it->second) {

        for (const auto& kv : it->second) {

            long long dimensionToken =

            long long dimensionToken = protoOutput->start(

                    mProto->start(FIELD_TYPE_MESSAGE | FIELD_COUNT_REPEATED | FIELD_ID_DIMENSION);

                    FIELD_TYPE_MESSAGE | FIELD_COUNT_REPEATED | FIELD_ID_DIMENSION);

            mProto->write(FIELD_TYPE_INT32 | FIELD_ID_KEY, kv.key());

            protoOutput->write(FIELD_TYPE_INT32 | FIELD_ID_KEY, kv.key());

            if (kv.has_value_str()) {

            if (kv.has_value_str()) {

                mProto->write(FIELD_TYPE_STRING | FIELD_ID_VALUE_STR, kv.value_str());

                protoOutput->write(FIELD_TYPE_STRING | FIELD_ID_VALUE_STR, kv.value_str());

            } else if (kv.has_value_int()) {

            } else if (kv.has_value_int()) {

                mProto->write(FIELD_TYPE_INT64 | FIELD_ID_VALUE_INT, kv.value_int());

                protoOutput->write(FIELD_TYPE_INT64 | FIELD_ID_VALUE_INT, kv.value_int());

            } else if (kv.has_value_bool()) {

            } else if (kv.has_value_bool()) {

                mProto->write(FIELD_TYPE_BOOL | FIELD_ID_VALUE_BOOL, kv.value_bool());

                protoOutput->write(FIELD_TYPE_BOOL | FIELD_ID_VALUE_BOOL, kv.value_bool());

            } else if (kv.has_value_float()) {

            } else if (kv.has_value_float()) {

                mProto->write(FIELD_TYPE_FLOAT | FIELD_ID_VALUE_FLOAT, kv.value_float());

                protoOutput->write(FIELD_TYPE_FLOAT | FIELD_ID_VALUE_FLOAT, kv.value_float());

            }

            }

            mProto->end(dimensionToken);

            protoOutput->end(dimensionToken);

        }

        }





        // Then fill bucket_info (DurationBucketInfo).

        // Then fill bucket_info (DurationBucketInfo).

        for (const auto& bucket : pair.second) {

        for (const auto& bucket : pair.second) {

            long long bucketInfoToken =

            long long bucketInfoToken = protoOutput->start(

                    mProto->start(FIELD_TYPE_MESSAGE | FIELD_COUNT_REPEATED | FIELD_ID_BUCKET_INFO);

                    FIELD_TYPE_MESSAGE | FIELD_COUNT_REPEATED | FIELD_ID_BUCKET_INFO);

            mProto->write(FIELD_TYPE_INT64 | FIELD_ID_START_BUCKET_NANOS,

            protoOutput->write(FIELD_TYPE_INT64 | FIELD_ID_START_BUCKET_NANOS,

                               (long long)bucket.mBucketStartNs);

                               (long long)bucket.mBucketStartNs);

            mProto->write(FIELD_TYPE_INT64 | FIELD_ID_END_BUCKET_NANOS,

            protoOutput->write(FIELD_TYPE_INT64 | FIELD_ID_END_BUCKET_NANOS,

                               (long long)bucket.mBucketEndNs);

                               (long long)bucket.mBucketEndNs);

            mProto->write(FIELD_TYPE_INT64 | FIELD_ID_DURATION, (long long)bucket.mDuration);

            protoOutput->write(FIELD_TYPE_INT64 | FIELD_ID_DURATION, (long long)bucket.mDuration);

            mProto->end(bucketInfoToken);

            protoOutput->end(bucketInfoToken);

            VLOG("\t bucket [%lld - %lld] duration: %lld", (long long)bucket.mBucketStartNs,

            VLOG("\t bucket [%lld - %lld] duration: %lld", (long long)bucket.mBucketStartNs,

                 (long long)bucket.mBucketEndNs, (long long)bucket.mDuration);

                 (long long)bucket.mBucketEndNs, (long long)bucket.mDuration);

        }

        }





        mProto->end(wrapperToken);

        protoOutput->end(wrapperToken);

    }

    }





    mProto->end(mProtoToken);

    protoOutput->end(protoToken);

    mProto->write(FIELD_TYPE_INT64 | FIELD_ID_END_REPORT_NANOS,

    protoOutput->write(FIELD_TYPE_INT64 | FIELD_ID_END_REPORT_NANOS, (long long)dumpTimeNs);

                  (long long)mCurrentBucketStartTimeNs);

    std::unique_ptr<std::vector<uint8_t>> buffer = serializeProtoLocked();

    startNewProtoOutputStreamLocked(endTime);

    mPastBuckets.clear();

    mPastBuckets.clear();

    return buffer;

    mStartTimeNs = mCurrentBucketStartTimeNs;

}

}





void DurationMetricProducer::flushIfNeededLocked(const uint64_t& eventTime) {

void DurationMetricProducer::flushIfNeededLocked(const uint64_t& eventTime) {

cmds/statsd/src/metrics/DurationMetricProducer.h

+2 −7
Original line number Original line Diff line number Diff line
@@ -47,8 +47,6 @@ public: 




    virtual sp<AnomalyTracker> createAnomalyTracker(const Alert &alert) override;

    virtual sp<AnomalyTracker> createAnomalyTracker(const Alert &alert) override;





    void finish() override;



    // TODO: Implement this later.

    // TODO: Implement this later.

    virtual void notifyAppUpgrade(const string& apk, const int uid, const int64_t version)

    virtual void notifyAppUpgrade(const string& apk, const int uid, const int64_t version)

            override{};

            override{};
@@ -62,8 +60,8 @@ protected: 
            const LogEvent& event, bool scheduledPull) override;

            const LogEvent& event, bool scheduledPull) override;





private:

private:

    // TODO: Pass a timestamp as a parameter in onDumpReport.

    void onDumpReportLocked(const uint64_t dumpTimeNs,

    std::unique_ptr<std::vector<uint8_t>> onDumpReportLocked() override;

                            android::util::ProtoOutputStream* protoOutput) override;





    // Internal interface to handle condition change.

    // Internal interface to handle condition change.

    void onConditionChangedLocked(const bool conditionMet, const uint64_t eventTime) override;

    void onConditionChangedLocked(const bool conditionMet, const uint64_t eventTime) override;
@@ -77,9 +75,6 @@ private: 
    // Util function to flush the old packet.

    // Util function to flush the old packet.

    void flushIfNeededLocked(const uint64_t& eventTime);

    void flushIfNeededLocked(const uint64_t& eventTime);





    // Util function to init/reset the proto output stream.

    void startNewProtoOutputStreamLocked(long long timestamp);



    const DurationMetric mMetric;

    const DurationMetric mMetric;





    // Index of the SimpleAtomMatcher which defines the start.

    // Index of the SimpleAtomMatcher which defines the start.