1 // Copyright 2020 The Pigweed Authors
2 //
3 // Licensed under the Apache License, Version 2.0 (the "License"); you may not
4 // use this file except in compliance with the License. You may obtain a copy of
5 // the License at
6 //
7 // https://www.apache.org/licenses/LICENSE-2.0
8 //
9 // Unless required by applicable law or agreed to in writing, software
10 // distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
11 // WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
12 // License for the specific language governing permissions and limitations under
13 // the License.
14
15 #include "pw_metric/metric_service_nanopb.h"
16
17 #include <cstring>
18
19 #include "pw_assert/check.h"
20 #include "pw_containers/vector.h"
21 #include "pw_metric/metric.h"
22 #include "pw_metric_private/metric_walker.h"
23 #include "pw_preprocessor/util.h"
24 #include "pw_span/span.h"
25
26 namespace pw::metric {
27 namespace {
28
29 class NanopbMetricWriter : public virtual internal::MetricWriter {
30 public:
NanopbMetricWriter(MetricService::ServerWriter<pw_metric_proto_MetricResponse> & response_writer)31 NanopbMetricWriter(
32 MetricService::ServerWriter<pw_metric_proto_MetricResponse>&
33 response_writer)
34 : response_(pw_metric_proto_MetricResponse_init_zero),
35 response_writer_(response_writer) {}
36
37 // TODO(keir): Figure out a pw_rpc mechanism to fill a streaming packet based
38 // on transport MTU, rather than having this as a static knob. For example,
39 // some transports may be able to fit 30 metrics; others, only 5.
Write(const Metric & metric,const Vector<Token> & path)40 Status Write(const Metric& metric, const Vector<Token>& path) override {
41 // Nanopb doesn't offer an easy way to do bounds checking, so use span's
42 // type deduction magic to figure out the max size.
43 span<pw_metric_proto_Metric> metrics(response_.metrics);
44 PW_CHECK_INT_LT(response_.metrics_count, metrics.size());
45
46 // Grab the next available Metric slot to write to in the response.
47 pw_metric_proto_Metric& proto_metric =
48 response_.metrics[response_.metrics_count];
49
50 // Copy the path.
51 span<Token> proto_path(proto_metric.token_path);
52 PW_CHECK_INT_LE(path.size(), proto_path.size());
53 std::copy(path.begin(), path.end(), proto_path.begin());
54 proto_metric.token_path_count = path.size();
55
56 // Copy the metric value.
57 if (metric.is_float()) {
58 proto_metric.value.as_float = metric.as_float();
59 proto_metric.which_value = pw_metric_proto_Metric_as_float_tag;
60 } else {
61 proto_metric.value.as_int = metric.as_int();
62 proto_metric.which_value = pw_metric_proto_Metric_as_int_tag;
63 }
64
65 // Move write head to the next slot.
66 response_.metrics_count++;
67
68 // If the metric response object is full, send the response and reset.
69 // TODO(keir): Support runtime batch sizes < max proto size.
70 if (response_.metrics_count == metrics.size()) {
71 Flush();
72 }
73
74 return OkStatus();
75 }
76
Flush()77 void Flush() {
78 if (response_.metrics_count) {
79 response_writer_.Write(response_)
80 .IgnoreError(); // TODO: b/242598609 - Handle Status properly
81 response_ = pw_metric_proto_MetricResponse_init_zero;
82 }
83 }
84
85 private:
86 pw_metric_proto_MetricResponse response_;
87 // This RPC stream writer handle must be valid for the metric writer lifetime.
88 MetricService::ServerWriter<pw_metric_proto_MetricResponse>& response_writer_;
89 };
90
91 } // namespace
92
Get(const pw_metric_proto_MetricRequest &,ServerWriter<pw_metric_proto_MetricResponse> & response)93 void MetricService::Get(
94 const pw_metric_proto_MetricRequest& /* request */,
95 ServerWriter<pw_metric_proto_MetricResponse>& response) {
96 // For now, ignore the request and just stream all the metrics back.
97 NanopbMetricWriter writer(response);
98 internal::MetricWalker walker(writer);
99
100 // This will stream all the metrics in the span of this Get() method call.
101 // This will have the effect of blocking the RPC thread until all the metrics
102 // are sent. That is likely to cause problems if there are many metrics, or
103 // if other RPCs are higher priority and should complete first.
104 //
105 // In the future, this should be replaced with an optional async solution
106 // that puts the application in control of when the response batches are sent.
107 walker.Walk(metrics_).IgnoreError();
108 walker.Walk(groups_).IgnoreError();
109 writer.Flush();
110 }
111
112 } // namespace pw::metric
113