blob: 41ae6af04489d3655a42a62dca20c161b021780b [file] [log] [blame]
// Copyright 2020 The Pigweed Authors
//
// Licensed under the Apache License, Version 2.0 (the "License"); you may not
// use this file except in compliance with the License. You may obtain a copy of
// the License at
//
// https://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
// WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
// License for the specific language governing permissions and limitations under
// the License.
#include "pw_metric/metric_service_nanopb.h"
#include <cstring>
#include "pw_assert/check.h"
#include "pw_containers/vector.h"
#include "pw_metric/metric.h"
#include "pw_preprocessor/util.h"
#include "pw_span/span.h"
namespace pw::metric {
namespace {
class MetricWriter {
public:
MetricWriter(MetricService::ServerWriter<pw_metric_proto_MetricResponse>&
response_writer)
: response_(pw_metric_proto_MetricResponse_init_zero),
response_writer_(response_writer) {}
// TODO(keir): Figure out a pw_rpc mechanism to fill a streaming packet based
// on transport MTU, rather than having this as a static knob. For example,
// some transports may be able to fit 30 metrics; others, only 5.
void Write(const Metric& metric, const Vector<Token>& path) {
// Nanopb doesn't offer an easy way to do bounds checking, so use span's
// type deduction magic to figure out the max size.
span<pw_metric_proto_Metric> metrics(response_.metrics);
PW_CHECK_INT_LT(response_.metrics_count, metrics.size());
// Grab the next available Metric slot to write to in the response.
pw_metric_proto_Metric& proto_metric =
response_.metrics[response_.metrics_count];
// Copy the path.
span<Token> proto_path(proto_metric.token_path);
PW_CHECK_INT_LE(path.size(), proto_path.size());
std::copy(path.begin(), path.end(), proto_path.begin());
proto_metric.token_path_count = path.size();
// Copy the metric value.
if (metric.is_float()) {
proto_metric.value.as_float = metric.as_float();
proto_metric.which_value = pw_metric_proto_Metric_as_float_tag;
} else {
proto_metric.value.as_int = metric.as_int();
proto_metric.which_value = pw_metric_proto_Metric_as_int_tag;
}
// Move write head to the next slot.
response_.metrics_count++;
// If the metric response object is full, send the response and reset.
// TODO(keir): Support runtime batch sizes < max proto size.
if (response_.metrics_count == metrics.size()) {
Flush();
}
}
void Flush() {
if (response_.metrics_count) {
response_writer_.Write(response_)
.IgnoreError(); // TODO(b/242598609): Handle Status properly
response_ = pw_metric_proto_MetricResponse_init_zero;
}
}
private:
pw_metric_proto_MetricResponse response_;
// This RPC stream writer handle must be valid for the metric writer lifetime.
MetricService::ServerWriter<pw_metric_proto_MetricResponse>& response_writer_;
};
// Walk a metric tree recursively; passing metrics with their path (names) to a
// metric writer which can consume them.
//
// TODO(keir): Generalize this to support a generic visitor.
class MetricWalker {
public:
MetricWalker(MetricWriter& writer) : writer_(writer) {}
void Walk(const IntrusiveList<Metric>& metrics) {
for (const auto& m : metrics) {
ScopedName scoped_name(m.name(), *this);
writer_.Write(m, path_);
}
}
void Walk(const IntrusiveList<Group>& groups) {
for (const auto& g : groups) {
Walk(g);
}
}
void Walk(const Group& group) {
ScopedName scoped_name(group.name(), *this);
Walk(group.children());
Walk(group.metrics());
}
private:
// Exists to safely push/pop parent groups from the explicit stack.
struct ScopedName {
ScopedName(Token name, MetricWalker& rhs) : walker(rhs) {
PW_CHECK_INT_LT(walker.path_.size(),
walker.path_.capacity(),
"Metrics are too deep; bump path_ capacity");
walker.path_.push_back(name);
}
~ScopedName() { walker.path_.pop_back(); }
MetricWalker& walker;
};
Vector<Token, 4 /* max depth */> path_;
MetricWriter& writer_;
};
} // namespace
void MetricService::Get(
const pw_metric_proto_MetricRequest& /* request */,
ServerWriter<pw_metric_proto_MetricResponse>& response) {
// For now, ignore the request and just stream all the metrics back.
MetricWriter writer(response);
MetricWalker walker(writer);
// This will stream all the metrics in the span of this Get() method call.
// This will have the effect of blocking the RPC thread until all the metrics
// are sent. That is likely to cause problems if there are many metrics, or
// if other RPCs are higher priority and should complete first.
//
// In the future, this should be replaced with an optional async solution
// that puts the application in control of when the response batches are sent.
walker.Walk(metrics_);
walker.Walk(groups_);
writer.Flush();
}
} // namespace pw::metric