1 //
2 //
3 // Copyright 2023 gRPC authors.
4 //
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
8 //
9 // http://www.apache.org/licenses/LICENSE-2.0
10 //
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.
16 //
17 //
18
19 #include "src/cpp/ext/gcp/environment_autodetect.h"
20
21 #include <grpc/support/alloc.h>
22 #include <grpc/support/port_platform.h>
23 #include <grpc/support/sync.h>
24 #include <grpcpp/impl/grpc_library.h>
25
26 #include <memory>
27 #include <utility>
28
29 #include "absl/container/flat_hash_map.h"
30 #include "absl/log/check.h"
31 #include "absl/log/log.h"
32 #include "absl/status/status.h"
33 #include "absl/status/statusor.h"
34 #include "absl/types/optional.h"
35 #include "src/core/lib/debug/trace.h"
36 #include "src/core/lib/event_engine/default_event_engine.h"
37 #include "src/core/lib/iomgr/closure.h"
38 #include "src/core/lib/iomgr/error.h"
39 #include "src/core/lib/iomgr/exec_ctx.h"
40 #include "src/core/lib/iomgr/iomgr_fwd.h"
41 #include "src/core/lib/iomgr/polling_entity.h"
42 #include "src/core/lib/iomgr/pollset.h"
43 #include "src/core/lib/slice/slice.h"
44 #include "src/core/util/crash.h"
45 #include "src/core/util/env.h"
46 #include "src/core/util/gcp_metadata_query.h"
47 #include "src/core/util/load_file.h"
48 #include "src/core/util/orphanable.h"
49 #include "src/core/util/status_helper.h"
50 #include "src/core/util/time.h"
51
52 namespace grpc {
53 namespace internal {
54
55 namespace {
56
57 // This is not a definite method to get the namespace name for GKE, but it is
58 // the best we have.
GetNamespaceName()59 std::string GetNamespaceName() {
60 // Read the root file.
61 const char* filename =
62 "/var/run/secrets/kubernetes.io/serviceaccount/namespace";
63 auto namespace_name = grpc_core::LoadFile(filename, false);
64 if (!namespace_name.ok()) {
65 GRPC_TRACE_VLOG(environment_autodetect, 2)
66 << "Reading file " << filename
67 << " failed: " << grpc_core::StatusToString(namespace_name.status());
68 // Fallback on an environment variable
69 return grpc_core::GetEnv("NAMESPACE_NAME").value_or("");
70 }
71 return std::string(reinterpret_cast<const char*>((*namespace_name).begin()),
72 (*namespace_name).length());
73 }
74
75 // Get pod name for GKE
GetPodName()76 std::string GetPodName() {
77 auto pod_name = grpc_core::GetEnv("POD_NAME");
78 if (pod_name.has_value()) {
79 return pod_name.value();
80 }
81 return grpc_core::GetEnv("HOSTNAME").value_or("");
82 }
83
84 // Get container name for GKE
GetContainerName()85 std::string GetContainerName() {
86 return grpc_core::GetEnv("HOSTNAME").value_or("");
87 }
88
89 // Get function name for Cloud Functions
GetFunctionName()90 std::string GetFunctionName() {
91 auto k_service = grpc_core::GetEnv("K_SERVICE");
92 if (k_service.has_value()) {
93 return k_service.value();
94 }
95 return grpc_core::GetEnv("FUNCTION_NAME").value_or("");
96 }
97
98 // Get revision name for Cloud run
GetRevisionName()99 std::string GetRevisionName() {
100 return grpc_core::GetEnv("K_REVISION").value_or("");
101 }
102
103 // Get service name for Cloud run
GetServiceName()104 std::string GetServiceName() {
105 return grpc_core::GetEnv("K_SERVICE").value_or("");
106 }
107
108 // Get configuration name for Cloud run
GetConfiguratioName()109 std::string GetConfiguratioName() {
110 return grpc_core::GetEnv("K_CONFIGURATION").value_or("");
111 }
112
113 // Get module ID for App Engine
GetModuleId()114 std::string GetModuleId() {
115 return grpc_core::GetEnv("GAE_SERVICE").value_or("");
116 }
117
118 // Get version ID for App Engine
GetVersionId()119 std::string GetVersionId() {
120 return grpc_core::GetEnv("GAE_VERSION").value_or("");
121 }
122
123 // Fire and forget class
124 class EnvironmentAutoDetectHelper
125 : public grpc_core::InternallyRefCounted<EnvironmentAutoDetectHelper>,
126 private internal::GrpcLibrary {
127 public:
EnvironmentAutoDetectHelper(std::string project_id,absl::AnyInvocable<void (EnvironmentAutoDetect::ResourceType)> on_done,std::shared_ptr<grpc_event_engine::experimental::EventEngine> event_engine)128 EnvironmentAutoDetectHelper(
129 std::string project_id,
130 absl::AnyInvocable<void(EnvironmentAutoDetect::ResourceType)> on_done,
131 std::shared_ptr<grpc_event_engine::experimental::EventEngine>
132 event_engine)
133 : InternallyRefCounted(/*trace=*/nullptr, /*initial_refcount=*/2),
134 project_id_(std::move(project_id)),
135 on_done_(std::move(on_done)),
136 event_engine_(std::move(event_engine)) {
137 grpc_core::ExecCtx exec_ctx;
138 // TODO(yashykt): The pollset stuff should go away once the HTTP library is
139 // ported over to use EventEngine.
140 pollset_ = static_cast<grpc_pollset*>(gpr_zalloc(grpc_pollset_size()));
141 grpc_pollset_init(pollset_, &mu_poll_);
142 pollent_ = grpc_polling_entity_create_from_pollset(pollset_);
143 // TODO(yashykt): Note that using EventEngine::Run is not fork-safe. If we
144 // want to make this fork-safe, we might need some re-work here.
145 event_engine_->Run([this] { PollLoop(); });
146 AutoDetect();
147 }
148
~EnvironmentAutoDetectHelper()149 ~EnvironmentAutoDetectHelper() override {
150 grpc_core::ExecCtx exec_ctx;
151 grpc_pollset_shutdown(
152 pollset_, GRPC_CLOSURE_CREATE(
153 [](void* arg, absl::Status /* status */) {
154 grpc_pollset_destroy(static_cast<grpc_pollset*>(arg));
155 gpr_free(arg);
156 },
157 pollset_, nullptr));
158 }
159
Orphan()160 void Orphan() override {
161 grpc_core::Crash("Illegal Orphan() call on EnvironmentAutoDetectHelper.");
162 }
163
164 private:
165 struct Attribute {
166 std::string resource_attribute;
167 std::string metadata_server_atttribute;
168 };
169
PollLoop()170 void PollLoop() {
171 grpc_core::ExecCtx exec_ctx;
172 bool done = false;
173 gpr_mu_lock(mu_poll_);
174 grpc_pollset_worker* worker = nullptr;
175 if (!GRPC_LOG_IF_ERROR(
176 "pollset_work",
177 grpc_pollset_work(grpc_polling_entity_pollset(&pollent_), &worker,
178 grpc_core::Timestamp::InfPast()))) {
179 notify_poller_ = true;
180 }
181 done = notify_poller_;
182 gpr_mu_unlock(mu_poll_);
183 if (!done) {
184 event_engine_->RunAfter(grpc_core::Duration::Milliseconds(100),
185 [this] { PollLoop(); });
186 } else {
187 Unref();
188 }
189 }
190
AutoDetect()191 void AutoDetect() {
192 grpc_core::MutexLock lock(&mu_);
193 // GKE
194 resource_.labels.emplace("project_id", project_id_);
195 if (grpc_core::GetEnv("KUBERNETES_SERVICE_HOST").has_value()) {
196 resource_.resource_type = "k8s_container";
197 resource_.labels.emplace("namespace_name", GetNamespaceName());
198 resource_.labels.emplace("pod_name", GetPodName());
199 resource_.labels.emplace("container_name", GetContainerName());
200 attributes_to_fetch_.emplace(grpc_core::GcpMetadataQuery::kZoneAttribute,
201 "location");
202 attributes_to_fetch_.emplace(
203 grpc_core::GcpMetadataQuery::kClusterNameAttribute, "cluster_name");
204 }
205 // Cloud Functions
206 else if (grpc_core::GetEnv("FUNCTION_NAME").has_value() ||
207 grpc_core::GetEnv("FUNCTION_TARGET").has_value()) {
208 resource_.resource_type = "cloud_function";
209 resource_.labels.emplace("function_name", GetFunctionName());
210 attributes_to_fetch_.emplace(
211 grpc_core::GcpMetadataQuery::kRegionAttribute, "region");
212 }
213 // Cloud Run
214 else if (grpc_core::GetEnv("K_CONFIGURATION").has_value()) {
215 resource_.resource_type = "cloud_run_revision";
216 resource_.labels.emplace("revision_name", GetRevisionName());
217 resource_.labels.emplace("service_name", GetServiceName());
218 resource_.labels.emplace("configuration_name", GetConfiguratioName());
219 attributes_to_fetch_.emplace(
220 grpc_core::GcpMetadataQuery::kRegionAttribute, "location");
221 }
222 // App Engine
223 else if (grpc_core::GetEnv("GAE_SERVICE").has_value()) {
224 resource_.resource_type = "gae_app";
225 resource_.labels.emplace("module_id", GetModuleId());
226 resource_.labels.emplace("version_id", GetVersionId());
227 attributes_to_fetch_.emplace(grpc_core::GcpMetadataQuery::kZoneAttribute,
228 "zone");
229 }
230 // Assume GCE
231 else {
232 assuming_gce_ = true;
233 resource_.resource_type = "gce_instance";
234 attributes_to_fetch_.emplace(
235 grpc_core::GcpMetadataQuery::kInstanceIdAttribute, "instance_id");
236 attributes_to_fetch_.emplace(grpc_core::GcpMetadataQuery::kZoneAttribute,
237 "zone");
238 }
239 FetchMetadataServerAttributesAsynchronouslyLocked();
240 }
241
FetchMetadataServerAttributesAsynchronouslyLocked()242 void FetchMetadataServerAttributesAsynchronouslyLocked()
243 ABSL_EXCLUSIVE_LOCKS_REQUIRED(mu_) {
244 CHECK(!attributes_to_fetch_.empty());
245 for (auto& element : attributes_to_fetch_) {
246 queries_.push_back(grpc_core::MakeOrphanable<grpc_core::GcpMetadataQuery>(
247 element.first, &pollent_,
248 [this](std::string attribute, absl::StatusOr<std::string> result) {
249 GRPC_TRACE_LOG(environment_autodetect, INFO)
250 << "Environment AutoDetect: Attribute: \"" << attribute
251 << "\" Result: \""
252 << (result.ok() ? result.value()
253 : grpc_core::StatusToString(result.status()))
254 << "\"";
255 absl::optional<EnvironmentAutoDetect::ResourceType> resource;
256 {
257 grpc_core::MutexLock lock(&mu_);
258 auto it = attributes_to_fetch_.find(attribute);
259 if (it != attributes_to_fetch_.end()) {
260 if (result.ok()) {
261 resource_.labels.emplace(std::move(it->second),
262 std::move(result).value());
263 }
264 // If fetching from the MetadataServer failed and we were
265 // assuming a GCE environment, fallback to "global".
266 else if (assuming_gce_) {
267 GRPC_TRACE_LOG(environment_autodetect, INFO)
268 << "Environment Autodetect: Falling back to "
269 << "global resource type";
270 assuming_gce_ = false;
271 resource_.resource_type = "global";
272 }
273 attributes_to_fetch_.erase(it);
274 } else {
275 // This should not happen
276 LOG(ERROR) << "An unexpected attribute was seen from the "
277 "MetadataServer: "
278 << attribute;
279 }
280 if (attributes_to_fetch_.empty()) {
281 resource = std::move(resource_);
282 }
283 }
284 if (resource.has_value()) {
285 gpr_mu_lock(mu_poll_);
286 notify_poller_ = true;
287 gpr_mu_unlock(mu_poll_);
288 auto on_done = std::move(on_done_);
289 Unref();
290 on_done(std::move(resource).value());
291 }
292 },
293 grpc_core::Duration::Seconds(10)));
294 }
295 }
296
297 const std::string project_id_;
298 grpc_pollset* pollset_ = nullptr;
299 grpc_polling_entity pollent_;
300 gpr_mu* mu_poll_ = nullptr;
301 absl::AnyInvocable<void(EnvironmentAutoDetect::ResourceType)> on_done_;
302 std::shared_ptr<grpc_event_engine::experimental::EventEngine> event_engine_;
303 grpc_core::Mutex mu_;
304 bool notify_poller_ = false;
305 absl::flat_hash_map<std::string /* metadata_server_attribute */,
306 std::string /* resource_attribute */>
307 attributes_to_fetch_ ABSL_GUARDED_BY(mu_);
308 std::vector<grpc_core::OrphanablePtr<grpc_core::GcpMetadataQuery>> queries_
309 ABSL_GUARDED_BY(mu_);
310 EnvironmentAutoDetect::ResourceType resource_ ABSL_GUARDED_BY(mu_);
311 // This would be true if we are assuming the resource to be GCE. In this case,
312 // there is a chance that it will fail and we should instead just use
313 // "global".
314 bool assuming_gce_ ABSL_GUARDED_BY(mu_) = false;
315 };
316
317 EnvironmentAutoDetect* g_autodetect = nullptr;
318
319 } // namespace
320
Create(std::string project_id)321 void EnvironmentAutoDetect::Create(std::string project_id) {
322 CHECK_EQ(g_autodetect, nullptr);
323 CHECK(!project_id.empty());
324
325 g_autodetect = new EnvironmentAutoDetect(project_id);
326 }
327
Get()328 EnvironmentAutoDetect& EnvironmentAutoDetect::Get() { return *g_autodetect; }
329
EnvironmentAutoDetect(std::string project_id)330 EnvironmentAutoDetect::EnvironmentAutoDetect(std::string project_id)
331 : project_id_(std::move(project_id)) {
332 CHECK(!project_id_.empty());
333 }
334
NotifyOnDone(absl::AnyInvocable<void ()> callback)335 void EnvironmentAutoDetect::NotifyOnDone(absl::AnyInvocable<void()> callback) {
336 std::shared_ptr<grpc_event_engine::experimental::EventEngine> event_engine;
337 {
338 grpc_core::MutexLock lock(&mu_);
339 // Environment has already been detected
340 if (resource_ != nullptr) {
341 // Execute on the event engine to avoid deadlocks.
342 return event_engine_->Run(std::move(callback));
343 }
344 callbacks_.push_back(std::move(callback));
345 // Use the event_engine_ pointer as a signal to judge whether we've started
346 // detecting the environment.
347 if (event_engine_ == nullptr) {
348 event_engine_ = grpc_event_engine::experimental::GetDefaultEventEngine();
349 event_engine = event_engine_;
350 }
351 }
352 if (event_engine) {
353 new EnvironmentAutoDetectHelper(
354 project_id_,
355 [this](EnvironmentAutoDetect::ResourceType resource) {
356 std::vector<absl::AnyInvocable<void()>> callbacks;
357 {
358 grpc_core::MutexLock lock(&mu_);
359 resource_ = std::make_unique<EnvironmentAutoDetect::ResourceType>(
360 std::move(resource));
361 callbacks = std::move(callbacks_);
362 }
363 for (auto& callback : callbacks) {
364 callback();
365 }
366 },
367 std::move(event_engine));
368 }
369 }
370
371 } // namespace internal
372 } // namespace grpc
373