1 //
2 // Copyright 2022 gRPC authors.
3 //
4 // Licensed under the Apache License, Version 2.0 (the "License");
5 // you may not use this file except in compliance with the License.
6 // You may obtain a copy of the License at
7 //
8 // http://www.apache.org/licenses/LICENSE-2.0
9 //
10 // Unless required by applicable law or agreed to in writing, software
11 // distributed under the License is distributed on an "AS IS" BASIS,
12 // WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13 // See the License for the specific language governing permissions and
14 // limitations under the License.
15 //
16
17 #include <grpc/support/port_platform.h>
18
19 #include <stdint.h>
20
21 #include <algorithm>
22 #include <map>
23 #include <memory>
24 #include <string>
25 #include <utility>
26 #include <vector>
27
28 #include "absl/status/status.h"
29 #include "absl/status/statusor.h"
30 #include "absl/strings/str_cat.h"
31 #include "absl/time/clock.h"
32 #include "absl/time/time.h"
33 #include "absl/types/optional.h"
34 #include "google/api/monitored_resource.pb.h"
35 #include "google/devtools/cloudtrace/v2/tracing.grpc.pb.h"
36 #include "google/monitoring/v3/metric_service.grpc.pb.h"
37 #include "opencensus/exporters/stats/stackdriver/stackdriver_exporter.h"
38 #include "opencensus/exporters/trace/stackdriver/stackdriver_exporter.h"
39 #include "opencensus/stats/stats.h"
40 #include "opencensus/trace/sampler.h"
41 #include "opencensus/trace/trace_config.h"
42
43 #include <grpc/grpc.h>
44 #include <grpcpp/ext/gcp_observability.h>
45 #include <grpcpp/opencensus.h>
46 #include <grpcpp/security/credentials.h>
47 #include <grpcpp/support/channel_arguments.h>
48
49 #include "src/core/ext/filters/logging/logging_filter.h"
50 #include "src/core/lib/gprpp/crash.h"
51 #include "src/core/lib/gprpp/notification.h"
52 #include "src/cpp/client/client_stats_interceptor.h"
53 #include "src/cpp/ext/filters/census/client_filter.h"
54 #include "src/cpp/ext/filters/census/grpc_plugin.h"
55 #include "src/cpp/ext/filters/census/open_census_call_tracer.h"
56 #include "src/cpp/ext/gcp/environment_autodetect.h"
57 #include "src/cpp/ext/gcp/observability_config.h"
58 #include "src/cpp/ext/gcp/observability_logging_sink.h"
59
60 namespace grpc {
61
62 namespace internal {
63 namespace {
64
65 grpc::internal::ObservabilityLoggingSink* g_logging_sink = nullptr;
66
67 bool g_gcp_observability_initialized = false;
68
69 // TODO(yashykt): These constants are currently derived from the example at
70 // https://cloud.google.com/traffic-director/docs/observability-proxyless#c++.
71 // We might want these to be configurable.
72 constexpr uint32_t kMaxAttributes = 128;
73 constexpr uint32_t kMaxAnnotations = 128;
74 constexpr uint32_t kMaxMessageEvents = 128;
75 constexpr uint32_t kMaxLinks = 128;
76
77 constexpr char kGoogleStackdriverTraceAddress[] = "cloudtrace.googleapis.com";
78 constexpr char kGoogleStackdriverStatsAddress[] = "monitoring.googleapis.com";
79
RegisterOpenCensusViewsForGcpObservability()80 void RegisterOpenCensusViewsForGcpObservability() {
81 // Register client default views for GCP observability
82 experimental::ClientStartedRpcs().RegisterForExport();
83 experimental::ClientCompletedRpcs().RegisterForExport();
84 experimental::ClientRoundtripLatency().RegisterForExport();
85 internal::ClientApiLatency().RegisterForExport();
86 experimental::ClientSentCompressedMessageBytesPerRpc().RegisterForExport();
87 experimental::ClientReceivedCompressedMessageBytesPerRpc()
88 .RegisterForExport();
89 // Register server default views for GCP observability
90 experimental::ServerStartedRpcs().RegisterForExport();
91 experimental::ServerCompletedRpcs().RegisterForExport();
92 experimental::ServerSentCompressedMessageBytesPerRpc().RegisterForExport();
93 experimental::ServerReceivedCompressedMessageBytesPerRpc()
94 .RegisterForExport();
95 experimental::ServerServerLatency().RegisterForExport();
96 }
97
98 } // namespace
99
GcpObservabilityInit()100 absl::Status GcpObservabilityInit() {
101 auto config = grpc::internal::GcpObservabilityConfig::ReadFromEnv();
102 if (!config.ok()) {
103 return config.status();
104 }
105 if (!config->cloud_trace.has_value() &&
106 !config->cloud_monitoring.has_value() &&
107 !config->cloud_logging.has_value()) {
108 return absl::OkStatus();
109 }
110 if (g_gcp_observability_initialized) {
111 grpc_core::Crash("GCP Observability for gRPC was already initialized.");
112 }
113 g_gcp_observability_initialized = true;
114 grpc::internal::EnvironmentAutoDetect::Create(config->project_id);
115 if (!config->cloud_trace.has_value()) {
116 // Disable OpenCensus tracing
117 grpc::internal::EnableOpenCensusTracing(false);
118 }
119 if (!config->cloud_monitoring.has_value()) {
120 // Disable OpenCensus stats
121 grpc::internal::EnableOpenCensusStats(false);
122 } else {
123 // Register the OpenCensus client stats interceptor factory if stats are
124 // enabled. Note that this is currently separate from the OpenCensus Plugin
125 // to avoid changing the behavior of the currently available OpenCensus
126 // plugin.
127 grpc::internal::RegisterGlobalClientStatsInterceptorFactory(
128 new grpc::internal::OpenCensusClientInterceptorFactory);
129 }
130 if (config->cloud_logging.has_value()) {
131 g_logging_sink = new grpc::internal::ObservabilityLoggingSink(
132 config->cloud_logging.value(), config->project_id, config->labels);
133 grpc_core::RegisterLoggingFilter(g_logging_sink);
134 }
135 // If tracing or monitoring is enabled, we need to register the OpenCensus
136 // plugin as well.
137 if (config->cloud_trace.has_value() || config->cloud_monitoring.has_value()) {
138 grpc::RegisterOpenCensusPlugin();
139 }
140 // If tracing or monitoring is enabled, we need to detect the environment for
141 // OpenCensus, set the labels and attributes and prepare the StackDriver
142 // exporter.
143 // Note that this should be the last step of GcpObservabilityInit() since we
144 // can't register any more filters after grpc_init.
145 if (config->cloud_trace.has_value() || config->cloud_monitoring.has_value()) {
146 grpc_init();
147 grpc_core::Notification notification;
148 grpc::internal::EnvironmentAutoDetect::Get().NotifyOnDone(
149 [&]() { notification.Notify(); });
150 notification.WaitForNotification();
151 auto* resource = grpc::internal::EnvironmentAutoDetect::Get().resource();
152 if (config->cloud_trace.has_value()) {
153 // Set up attributes for constant tracing
154 std::vector<internal::OpenCensusRegistry::Attribute> attributes;
155 attributes.reserve(resource->labels.size() + config->labels.size());
156 // First insert in environment labels
157 for (const auto& resource_label : resource->labels) {
158 attributes.push_back(internal::OpenCensusRegistry::Attribute{
159 absl::StrCat(resource->resource_type, ".", resource_label.first),
160 resource_label.second});
161 }
162 // Then insert in labels from the GCP Observability config.
163 for (const auto& constant_label : config->labels) {
164 attributes.push_back(internal::OpenCensusRegistry::Attribute{
165 constant_label.first, constant_label.second});
166 }
167 grpc::internal::OpenCensusRegistry::Get().RegisterConstantAttributes(
168 std::move(attributes));
169 }
170 if (config->cloud_monitoring.has_value()) {
171 grpc::internal::OpenCensusRegistry::Get().RegisterConstantLabels(
172 config->labels);
173 RegisterOpenCensusViewsForGcpObservability();
174 }
175 // Note that we are setting up the exporters after registering the
176 // attributes and labels to avoid a case where the exporters start an RPC
177 // before we are ready.
178 if (config->cloud_trace.has_value()) {
179 // Set up the StackDriver Exporter for tracing.
180 opencensus::trace::TraceConfig::SetCurrentTraceParams(
181 {kMaxAttributes, kMaxAnnotations, kMaxMessageEvents, kMaxLinks,
182 opencensus::trace::ProbabilitySampler(
183 config->cloud_trace->sampling_rate)});
184 opencensus::exporters::trace::StackdriverOptions trace_opts;
185 trace_opts.project_id = config->project_id;
186 ChannelArguments args;
187 args.SetInt(GRPC_ARG_ENABLE_OBSERVABILITY, 0);
188 trace_opts.trace_service_stub =
189 ::google::devtools::cloudtrace::v2::TraceService::NewStub(
190 CreateCustomChannel(kGoogleStackdriverTraceAddress,
191 GoogleDefaultCredentials(), args));
192 opencensus::exporters::trace::StackdriverExporter::Register(
193 std::move(trace_opts));
194 }
195 if (config->cloud_monitoring.has_value()) {
196 // Set up the StackDriver Exporter for monitoring.
197 opencensus::exporters::stats::StackdriverOptions stats_opts;
198 stats_opts.project_id = config->project_id;
199 stats_opts.monitored_resource.set_type(resource->resource_type);
200 stats_opts.monitored_resource.mutable_labels()->insert(
201 resource->labels.begin(), resource->labels.end());
202 ChannelArguments args;
203 args.SetInt(GRPC_ARG_ENABLE_OBSERVABILITY, 0);
204 stats_opts.metric_service_stub =
205 google::monitoring::v3::MetricService::NewStub(
206 CreateCustomChannel(kGoogleStackdriverStatsAddress,
207 GoogleDefaultCredentials(), args));
208 opencensus::exporters::stats::StackdriverExporter::Register(
209 std::move(stats_opts));
210 }
211 grpc_shutdown();
212 }
213 return absl::OkStatus();
214 }
215
GcpObservabilityClose()216 void GcpObservabilityClose() {
217 if (g_logging_sink != nullptr) {
218 g_logging_sink->FlushAndClose();
219 }
220 // Currently, GcpObservabilityClose() only supports flushing logs. Stats and
221 // tracing get automatically flushed at a regular interval, so sleep for an
222 // interval to make sure that those are flushed too.
223 absl::SleepFor(absl::Seconds(25));
224 }
225
226 } // namespace internal
227
228 namespace experimental {
229
GcpObservabilityInit()230 absl::Status GcpObservabilityInit() {
231 return grpc::internal::GcpObservabilityInit();
232 }
233
GcpObservabilityClose()234 void GcpObservabilityClose() { return grpc::internal::GcpObservabilityClose(); }
235
236 } // namespace experimental
237
238 //
239 // GcpObservability
240 //
241
Init()242 absl::StatusOr<GcpObservability> GcpObservability::Init() {
243 absl::Status status = grpc::internal::GcpObservabilityInit();
244 if (!status.ok()) {
245 return status;
246 }
247 GcpObservability obj;
248 obj.impl_ = std::make_unique<GcpObservabilityImpl>();
249 return obj;
250 }
251
GcpObservability(GcpObservability && other)252 GcpObservability::GcpObservability(GcpObservability&& other) noexcept
253 : impl_(std::move(other.impl_)) {}
254
operator =(GcpObservability && other)255 GcpObservability& GcpObservability::operator=(
256 GcpObservability&& other) noexcept {
257 if (this != &other) {
258 impl_ = std::move(other.impl_);
259 }
260 return *this;
261 }
262
263 //
264 // GcpObservability::GcpObservabilityImpl
265 //
266
~GcpObservabilityImpl()267 GcpObservability::GcpObservabilityImpl::~GcpObservabilityImpl() {
268 grpc::internal::GcpObservabilityClose();
269 }
270
271 } // namespace grpc
272