1 // Copyright 2022 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_pwpb.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_metric_proto/metric_service.pwpb.h"
24 #include "pw_preprocessor/util.h"
25 #include "pw_protobuf/serialized_size.h"
26 #include "pw_rpc/raw/server_reader_writer.h"
27 #include "pw_span/span.h"
28 #include "pw_status/status.h"
29 #include "pw_status/try.h"
30
31 namespace pw::metric {
32
33 // TODO(amontanez): Make this follow the metric_service.options configuration.
34 constexpr size_t kMaxNumPackedEntries = 3;
35
36 namespace {
37
38 class PwpbMetricWriter : public virtual internal::MetricWriter {
39 public:
PwpbMetricWriter(span<std::byte> response,rpc::RawServerWriter & response_writer)40 PwpbMetricWriter(span<std::byte> response,
41 rpc::RawServerWriter& response_writer)
42 : response_(response),
43 response_writer_(response_writer),
44 encoder_(response) {}
45
46 // TODO(keir): Figure out a pw_rpc mechanism to fill a streaming packet based
47 // on transport MTU, rather than having this as a static knob. For example,
48 // some transports may be able to fit 30 metrics; others, only 5.
Write(const Metric & metric,const Vector<Token> & path)49 Status Write(const Metric& metric, const Vector<Token>& path) override {
50 { // Scope to control proto_encoder lifetime.
51
52 // Grab the next available Metric slot to write to in the response.
53 proto::pwpb::Metric::StreamEncoder proto_encoder =
54 encoder_.GetMetricsEncoder();
55 PW_TRY(proto_encoder.WriteTokenPath(path));
56 // Encode the metric value.
57 if (metric.is_float()) {
58 PW_TRY(proto_encoder.WriteAsFloat(metric.as_float()));
59 } else {
60 PW_TRY(proto_encoder.WriteAsInt(metric.as_int()));
61 }
62
63 metrics_count++;
64 }
65
66 if (metrics_count == kMaxNumPackedEntries) {
67 return Flush();
68 }
69 return OkStatus();
70 }
71
Flush()72 Status Flush() {
73 Status status;
74 if (metrics_count) {
75 status = response_writer_.Write(encoder_);
76 // Different way to clear MemoryEncoder. Copy constructor is disabled
77 // for memory encoder, and there is no "clear()" method.
78 encoder_.~MemoryEncoder();
79 new (&encoder_) proto::pwpb::MetricRequest::MemoryEncoder(response_);
80 metrics_count = 0;
81 }
82 return status;
83 }
84
85 private:
86 span<std::byte> response_;
87 // This RPC stream writer handle must be valid for the metric writer
88 // lifetime.
89 rpc::RawServerWriter& response_writer_;
90 proto::pwpb::MetricRequest::MemoryEncoder encoder_;
91 size_t metrics_count = 0;
92 };
93 } // namespace
94
Get(ConstByteSpan,rpc::RawServerWriter & raw_response)95 void MetricService::Get(ConstByteSpan /*request*/,
96 rpc::RawServerWriter& raw_response) {
97 // For now, ignore the request and just stream all the metrics back.
98
99 // The `string_path` field of Metric is not supported. The maximum size
100 // without values includes the maximum token path. Additionally, include the
101 // maximum size of the `as_int` field.
102 constexpr size_t kSizeOfOneMetric =
103 pw::metric::proto::pwpb::MetricResponse::
104 kMaxEncodedSizeBytesWithoutValues +
105 pw::metric::proto::pwpb::Metric::kMaxEncodedSizeBytesWithoutValues +
106 protobuf::SizeOfFieldUint32(
107 pw::metric::proto::pwpb::Metric::Fields::kAsInt);
108
109 // TODO(amontanez): Make this follow the metric_service.options configuration.
110 constexpr size_t kEncodeBufferSize = kMaxNumPackedEntries * kSizeOfOneMetric;
111
112 std::array<std::byte, kEncodeBufferSize> encode_buffer;
113
114 PwpbMetricWriter writer(encode_buffer, raw_response);
115 internal::MetricWalker walker(writer);
116
117 // This will stream all the metrics in the span of this Get() method call.
118 // This will have the effect of blocking the RPC thread until all the metrics
119 // are sent. That is likely to cause problems if there are many metrics, or
120 // if other RPCs are higher priority and should complete first.
121 //
122 // In the future, this should be replaced with an optional async solution
123 // that puts the application in control of when the response batches are sent.
124
125 // Propagate status through walker.
126 Status status;
127 status.Update(walker.Walk(metrics_));
128 status.Update(walker.Walk(groups_));
129 status.Update(writer.Flush());
130 raw_response.Finish(status).IgnoreError();
131 }
132 } // namespace pw::metric
133