3 * Copyright 2018 gRPC authors.
5 * Licensed under the Apache License, Version 2.0 (the "License");
6 * you may not use this file except in compliance with the License.
7 * You may obtain a copy of the License at
9 * http://www.apache.org/licenses/LICENSE-2.0
11 * Unless required by applicable law or agreed to in writing, software
12 * distributed under the License is distributed on an "AS IS" BASIS,
13 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14 * See the License for the specific language governing permissions and
15 * limitations under the License.
19 #include <grpc/support/port_platform.h>
21 #include "src/cpp/ext/filters/census/server_filter.h"
23 #include "absl/strings/str_cat.h"
24 #include "absl/strings/string_view.h"
25 #include "absl/time/clock.h"
26 #include "absl/time/time.h"
27 #include "opencensus/stats/stats.h"
28 #include "src/core/lib/surface/call.h"
29 #include "src/cpp/ext/filters/census/grpc_plugin.h"
30 #include "src/cpp/ext/filters/census/measures.h"
34 constexpr uint32_t CensusServerCallData::kMaxServerStatsLen;
38 // server metadata elements
39 struct ServerMetadataElements {
41 grpc_slice tracing_slice;
42 grpc_slice census_proto;
45 void FilterInitialMetadata(grpc_metadata_batch* b,
46 ServerMetadataElements* sml) {
47 if (b->idx.named.path != nullptr) {
48 sml->path = grpc_slice_ref_internal(GRPC_MDVALUE(b->idx.named.path->md));
50 if (b->idx.named.grpc_trace_bin != nullptr) {
52 grpc_slice_ref_internal(GRPC_MDVALUE(b->idx.named.grpc_trace_bin->md));
53 grpc_metadata_batch_remove(b, GRPC_BATCH_GRPC_TRACE_BIN);
55 if (b->idx.named.grpc_tags_bin != nullptr) {
57 grpc_slice_ref_internal(GRPC_MDVALUE(b->idx.named.grpc_tags_bin->md));
58 grpc_metadata_batch_remove(b, GRPC_BATCH_GRPC_TAGS_BIN);
64 void CensusServerCallData::OnDoneRecvMessageCb(void* user_data,
66 grpc_call_element* elem = reinterpret_cast<grpc_call_element*>(user_data);
67 CensusServerCallData* calld =
68 reinterpret_cast<CensusServerCallData*>(elem->call_data);
69 CensusChannelData* channeld =
70 reinterpret_cast<CensusChannelData*>(elem->channel_data);
71 GPR_ASSERT(calld != nullptr);
72 GPR_ASSERT(channeld != nullptr);
73 // Stream messages are no longer valid after receiving trailing metadata.
74 if ((*calld->recv_message_) != nullptr) {
75 ++calld->recv_message_count_;
77 grpc_core::Closure::Run(DEBUG_LOCATION, calld->initial_on_done_recv_message_,
78 GRPC_ERROR_REF(error));
81 void CensusServerCallData::OnDoneRecvInitialMetadataCb(void* user_data,
83 grpc_call_element* elem = reinterpret_cast<grpc_call_element*>(user_data);
84 CensusServerCallData* calld =
85 reinterpret_cast<CensusServerCallData*>(elem->call_data);
86 GPR_ASSERT(calld != nullptr);
87 if (error == GRPC_ERROR_NONE) {
88 grpc_metadata_batch* initial_metadata = calld->recv_initial_metadata_;
89 GPR_ASSERT(initial_metadata != nullptr);
90 ServerMetadataElements sml;
91 sml.path = grpc_empty_slice();
92 sml.tracing_slice = grpc_empty_slice();
93 sml.census_proto = grpc_empty_slice();
94 FilterInitialMetadata(initial_metadata, &sml);
95 calld->path_ = grpc_slice_ref_internal(sml.path);
96 calld->method_ = GetMethod(&calld->path_);
97 calld->qualified_method_ = absl::StrCat("Recv.", calld->method_);
98 const char* tracing_str =
99 GRPC_SLICE_IS_EMPTY(sml.tracing_slice)
101 : reinterpret_cast<const char*>(
102 GRPC_SLICE_START_PTR(sml.tracing_slice));
103 size_t tracing_str_len = GRPC_SLICE_IS_EMPTY(sml.tracing_slice)
105 : GRPC_SLICE_LENGTH(sml.tracing_slice);
106 GenerateServerContext(absl::string_view(tracing_str, tracing_str_len),
107 calld->qualified_method_, &calld->context_);
108 grpc_slice_unref_internal(sml.tracing_slice);
109 grpc_slice_unref_internal(sml.census_proto);
110 grpc_slice_unref_internal(sml.path);
111 grpc_census_call_set_context(
112 calld->gc_, reinterpret_cast<census_context*>(&calld->context_));
114 grpc_core::Closure::Run(DEBUG_LOCATION,
115 calld->initial_on_done_recv_initial_metadata_,
116 GRPC_ERROR_REF(error));
119 void CensusServerCallData::StartTransportStreamOpBatch(
120 grpc_call_element* elem, TransportStreamOpBatch* op) {
121 if (op->recv_initial_metadata() != nullptr) {
122 // substitute our callback for the op callback
123 recv_initial_metadata_ = op->recv_initial_metadata()->batch();
124 initial_on_done_recv_initial_metadata_ = op->recv_initial_metadata_ready();
125 op->set_recv_initial_metadata_ready(&on_done_recv_initial_metadata_);
127 if (op->send_message() != nullptr) {
128 ++sent_message_count_;
130 if (op->recv_message() != nullptr) {
131 recv_message_ = op->op()->payload->recv_message.recv_message;
132 initial_on_done_recv_message_ =
133 op->op()->payload->recv_message.recv_message_ready;
134 op->op()->payload->recv_message.recv_message_ready = &on_done_recv_message_;
136 // We need to record the time when the trailing metadata was sent to mark the
137 // completeness of the request.
138 if (op->send_trailing_metadata() != nullptr) {
139 elapsed_time_ = absl::Now() - start_time_;
140 size_t len = ServerStatsSerialize(absl::ToInt64Nanoseconds(elapsed_time_),
141 stats_buf_, kMaxServerStatsLen);
144 "census grpc_filter",
145 grpc_metadata_batch_add_tail(
146 op->send_trailing_metadata()->batch(), &census_bin_,
147 grpc_mdelem_from_slices(
148 GRPC_MDSTR_GRPC_SERVER_STATS_BIN,
149 grpc_core::UnmanagedMemorySlice(stats_buf_, len)),
150 GRPC_BATCH_GRPC_SERVER_STATS_BIN));
154 grpc_call_next_op(elem, op->op());
157 grpc_error* CensusServerCallData::Init(grpc_call_element* elem,
158 const grpc_call_element_args* args) {
159 start_time_ = absl::Now();
161 grpc_call_from_top_element(grpc_call_stack_element(args->call_stack, 0));
162 GRPC_CLOSURE_INIT(&on_done_recv_initial_metadata_,
163 OnDoneRecvInitialMetadataCb, elem,
164 grpc_schedule_on_exec_ctx);
165 GRPC_CLOSURE_INIT(&on_done_recv_message_, OnDoneRecvMessageCb, elem,
166 grpc_schedule_on_exec_ctx);
167 auth_context_ = grpc_call_auth_context(gc_);
168 return GRPC_ERROR_NONE;
171 void CensusServerCallData::Destroy(grpc_call_element* /*elem*/,
172 const grpc_call_final_info* final_info,
173 grpc_closure* /*then_call_closure*/) {
174 const uint64_t request_size = GetOutgoingDataSize(final_info);
175 const uint64_t response_size = GetIncomingDataSize(final_info);
176 double elapsed_time_ms = absl::ToDoubleMilliseconds(elapsed_time_);
177 grpc_auth_context_release(auth_context_);
178 ::opencensus::stats::Record(
179 {{RpcServerSentBytesPerRpc(), static_cast<double>(response_size)},
180 {RpcServerReceivedBytesPerRpc(), static_cast<double>(request_size)},
181 {RpcServerServerLatency(), elapsed_time_ms},
182 {RpcServerSentMessagesPerRpc(), sent_message_count_},
183 {RpcServerReceivedMessagesPerRpc(), recv_message_count_}},
184 {{ServerMethodTagKey(), method_},
185 {ServerStatusTagKey(), StatusCodeToString(final_info->final_status)}});
186 grpc_slice_unref_internal(path_);