mirror of https://github.com/grpc/grpc.git
GCP Observability : Framework for detecting the environment (#32294)
This code is not plumbed through yet, but it provides the core infrastructure needed to detect the proper GCP environment resources needed to set up the labels/attributes/resources for stats, tracing and logging. Details on how the various environment resources are setup has been derived by looking at java's cloud logging library and OpenTelemetry's future plans. (Could be better explained in an offline review since some links are internal). Requesting @veblush for a full review and @markdroth for a structural review. <!-- If you know who should review your pull request, please assign it to that person, otherwise the pull request would get assigned randomly. If your pull request is for a specific language, please add the appropriate lang label. -->pull/32426/head
Yash Tibrewal
2 years ago
committed by
GitHub
parent
3013c7e9b3
commit
04e3a8e73d
23 changed files with 952 additions and 187 deletions
@ -0,0 +1,132 @@ |
||||
//
|
||||
//
|
||||
// Copyright 2023 gRPC authors.
|
||||
//
|
||||
// Licensed under the Apache License, Version 2.0 (the "License");
|
||||
// you may not use this file except in compliance with the License.
|
||||
// You may obtain a copy of the License at
|
||||
//
|
||||
// http://www.apache.org/licenses/LICENSE-2.0
|
||||
//
|
||||
// Unless required by applicable law or agreed to in writing, software
|
||||
// distributed under the License is distributed on an "AS IS" BASIS,
|
||||
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
// See the License for the specific language governing permissions and
|
||||
// limitations under the License.
|
||||
//
|
||||
//
|
||||
|
||||
#include <grpc/support/port_platform.h> |
||||
|
||||
#include "src/core/ext/gcp/metadata_query.h" |
||||
|
||||
#include <string.h> |
||||
|
||||
#include <initializer_list> |
||||
#include <memory> |
||||
#include <utility> |
||||
|
||||
#include "absl/status/status.h" |
||||
#include "absl/status/statusor.h" |
||||
#include "absl/strings/str_format.h" |
||||
#include "absl/strings/string_view.h" |
||||
|
||||
#include <grpc/grpc.h> |
||||
#include <grpc/grpc_security.h> |
||||
#include <grpc/support/log.h> |
||||
|
||||
#include "src/core/lib/debug/trace.h" |
||||
#include "src/core/lib/gprpp/ref_counted_ptr.h" |
||||
#include "src/core/lib/gprpp/status_helper.h" |
||||
#include "src/core/lib/gprpp/time.h" |
||||
#include "src/core/lib/security/credentials/credentials.h" |
||||
#include "src/core/lib/uri/uri_parser.h" |
||||
|
||||
namespace grpc_core { |
||||
|
||||
TraceFlag grpc_metadata_query_trace(false, "metadata_query"); |
||||
|
||||
constexpr const char MetadataQuery::kZoneAttribute[] = |
||||
"/computeMetadata/v1/instance/zone"; |
||||
constexpr const char MetadataQuery::kClusterNameAttribute[] = |
||||
"/computeMetadata/v1/instance/attributes/cluster-name"; |
||||
constexpr const char MetadataQuery::kRegionAttribute[] = |
||||
"/computeMetadata/v1/instance/region"; |
||||
constexpr const char MetadataQuery::kInstanceIdAttribute[] = |
||||
"/computeMetadata/v1/instance/id"; |
||||
constexpr const char MetadataQuery::kIPv6Attribute[] = |
||||
"/computeMetadata/v1/instance/network-interfaces/0/ipv6s"; |
||||
|
||||
MetadataQuery::MetadataQuery( |
||||
std::string attribute, grpc_polling_entity* pollent, |
||||
absl::AnyInvocable<void(std::string /* attribute */, |
||||
absl::StatusOr<std::string> /* result */)> |
||||
callback, |
||||
Duration timeout) |
||||
: InternallyRefCounted<MetadataQuery>(nullptr, 2), |
||||
attribute_(std::move(attribute)), |
||||
callback_(std::move(callback)) { |
||||
GRPC_CLOSURE_INIT(&on_done_, OnDone, this, nullptr); |
||||
auto uri = URI::Create("http", "metadata.google.internal.", attribute_, |
||||
{} /* query params */, "" /* fragment */); |
||||
GPR_ASSERT(uri.ok()); // params are hardcoded
|
||||
grpc_http_request request; |
||||
memset(&request, 0, sizeof(grpc_http_request)); |
||||
grpc_http_header header = {const_cast<char*>("Metadata-Flavor"), |
||||
const_cast<char*>("Google")}; |
||||
request.hdr_count = 1; |
||||
request.hdrs = &header; |
||||
http_request_ = HttpRequest::Get( |
||||
std::move(*uri), nullptr /* channel args */, pollent, &request, |
||||
Timestamp::Now() + timeout, &on_done_, &response_, |
||||
RefCountedPtr<grpc_channel_credentials>( |
||||
grpc_insecure_credentials_create())); |
||||
http_request_->Start(); |
||||
} |
||||
|
||||
MetadataQuery::~MetadataQuery() { grpc_http_response_destroy(&response_); } |
||||
|
||||
void MetadataQuery::Orphan() { |
||||
http_request_.reset(); |
||||
Unref(); |
||||
} |
||||
|
||||
void MetadataQuery::OnDone(void* arg, grpc_error_handle error) { |
||||
auto* self = static_cast<MetadataQuery*>(arg); |
||||
if (GRPC_TRACE_FLAG_ENABLED(grpc_metadata_query_trace)) { |
||||
gpr_log(GPR_INFO, "MetadataServer Query for %s: HTTP status: %d, error: %s", |
||||
self->attribute_.c_str(), self->response_.status, |
||||
StatusToString(error).c_str()); |
||||
} |
||||
absl::StatusOr<std::string> result; |
||||
if (!error.ok()) { |
||||
result = absl::UnavailableError(absl::StrFormat( |
||||
"MetadataServer Query failed for %s: %s", self->attribute_.c_str(), |
||||
StatusToString(error).c_str())); |
||||
} else if (self->response_.status != 200) { |
||||
result = absl::UnavailableError(absl::StrFormat( |
||||
"MetadataServer Query received non-200 status for %s: %s", |
||||
self->attribute_.c_str(), StatusToString(error).c_str())); |
||||
} else if (self->attribute_ == kZoneAttribute) { |
||||
absl::string_view body(self->response_.body, self->response_.body_length); |
||||
size_t pos = body.find_last_of('/'); |
||||
if (pos == body.npos) { |
||||
result = absl::UnavailableError( |
||||
absl::StrFormat("MetadataServer Could not parse zone: %s", |
||||
std::string(body).c_str())); |
||||
if (GRPC_TRACE_FLAG_ENABLED(grpc_metadata_query_trace)) { |
||||
gpr_log(GPR_INFO, "%s", result.status().ToString().c_str()); |
||||
} |
||||
} else { |
||||
result = std::string(body.substr(pos + 1)); |
||||
} |
||||
} else { |
||||
result = std::string(self->response_.body, self->response_.body_length); |
||||
} |
||||
auto callback = std::move(self->callback_); |
||||
auto attribute = std::move(self->attribute_); |
||||
self->Unref(); |
||||
return callback(std::move(attribute), std::move(result)); |
||||
} |
||||
|
||||
} // namespace grpc_core
|
||||
@ -0,0 +1,72 @@ |
||||
//
|
||||
// Copyright 2023 gRPC authors.
|
||||
//
|
||||
// Licensed under the Apache License, Version 2.0 (the "License");
|
||||
// you may not use this file except in compliance with the License.
|
||||
// You may obtain a copy of the License at
|
||||
//
|
||||
// http://www.apache.org/licenses/LICENSE-2.0
|
||||
//
|
||||
// Unless required by applicable law or agreed to in writing, software
|
||||
// distributed under the License is distributed on an "AS IS" BASIS,
|
||||
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
// See the License for the specific language governing permissions and
|
||||
// limitations under the License.
|
||||
//
|
||||
|
||||
#ifndef GRPC_SRC_CORE_EXT_GCP_METADATA_QUERY_H |
||||
#define GRPC_SRC_CORE_EXT_GCP_METADATA_QUERY_H |
||||
|
||||
#include <grpc/support/port_platform.h> |
||||
|
||||
#include <string> |
||||
|
||||
#include "absl/functional/any_invocable.h" |
||||
#include "absl/status/statusor.h" |
||||
|
||||
#include "src/core/lib/gprpp/orphanable.h" |
||||
#include "src/core/lib/gprpp/time.h" |
||||
#include "src/core/lib/http/httpcli.h" |
||||
#include "src/core/lib/http/parser.h" |
||||
#include "src/core/lib/iomgr/closure.h" |
||||
#include "src/core/lib/iomgr/error.h" |
||||
#include "src/core/lib/iomgr/polling_entity.h" |
||||
|
||||
namespace grpc_core { |
||||
|
||||
// Fetches the value of an attribute from the MetadataServer on a GCP
|
||||
// environment.
|
||||
class MetadataQuery : public InternallyRefCounted<MetadataQuery> { |
||||
public: |
||||
static const char kZoneAttribute[]; |
||||
static const char kClusterNameAttribute[]; |
||||
static const char kRegionAttribute[]; |
||||
static const char kInstanceIdAttribute[]; |
||||
static const char kIPv6Attribute[]; |
||||
|
||||
MetadataQuery( |
||||
std::string attribute, grpc_polling_entity* pollent, |
||||
absl::AnyInvocable<void(std::string /* attribute */, |
||||
absl::StatusOr<std::string> /* result */)> |
||||
callback, |
||||
Duration timeout); |
||||
|
||||
~MetadataQuery() override; |
||||
|
||||
void Orphan() override; |
||||
|
||||
private: |
||||
static void OnDone(void* arg, grpc_error_handle error); |
||||
|
||||
grpc_closure on_done_; |
||||
std::string attribute_; |
||||
absl::AnyInvocable<void(std::string /* attribute */, |
||||
absl::StatusOr<std::string> /* result */)> |
||||
callback_; |
||||
OrphanablePtr<HttpRequest> http_request_; |
||||
grpc_http_response response_; |
||||
}; |
||||
|
||||
} // namespace grpc_core
|
||||
|
||||
#endif // GRPC_SRC_CORE_EXT_GCP_METADATA_QUERY_H
|
||||
@ -0,0 +1,368 @@ |
||||
//
|
||||
//
|
||||
// Copyright 2023 gRPC authors.
|
||||
//
|
||||
// Licensed under the Apache License, Version 2.0 (the "License");
|
||||
// you may not use this file except in compliance with the License.
|
||||
// You may obtain a copy of the License at
|
||||
//
|
||||
// http://www.apache.org/licenses/LICENSE-2.0
|
||||
//
|
||||
// Unless required by applicable law or agreed to in writing, software
|
||||
// distributed under the License is distributed on an "AS IS" BASIS,
|
||||
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
// See the License for the specific language governing permissions and
|
||||
// limitations under the License.
|
||||
//
|
||||
//
|
||||
|
||||
#include <grpc/support/port_platform.h> |
||||
|
||||
#include "src/cpp/ext/gcp/environment_autodetect.h" |
||||
|
||||
#include <algorithm> |
||||
#include <memory> |
||||
#include <utility> |
||||
|
||||
#include "absl/container/flat_hash_map.h" |
||||
#include "absl/meta/type_traits.h" |
||||
#include "absl/status/status.h" |
||||
#include "absl/status/statusor.h" |
||||
#include "absl/types/optional.h" |
||||
|
||||
#include <grpc/support/alloc.h> |
||||
#include <grpc/support/log.h> |
||||
#include <grpc/support/sync.h> |
||||
#include <grpcpp/impl/grpc_library.h> |
||||
|
||||
#include "src/core/ext/gcp/metadata_query.h" |
||||
#include "src/core/lib/debug/trace.h" |
||||
#include "src/core/lib/event_engine/default_event_engine.h" |
||||
#include "src/core/lib/gprpp/crash.h" |
||||
#include "src/core/lib/gprpp/env.h" |
||||
#include "src/core/lib/gprpp/load_file.h" |
||||
#include "src/core/lib/gprpp/orphanable.h" |
||||
#include "src/core/lib/gprpp/status_helper.h" |
||||
#include "src/core/lib/gprpp/time.h" |
||||
#include "src/core/lib/iomgr/closure.h" |
||||
#include "src/core/lib/iomgr/error.h" |
||||
#include "src/core/lib/iomgr/exec_ctx.h" |
||||
#include "src/core/lib/iomgr/iomgr_fwd.h" |
||||
#include "src/core/lib/iomgr/polling_entity.h" |
||||
#include "src/core/lib/iomgr/pollset.h" |
||||
#include "src/core/lib/slice/slice.h" |
||||
|
||||
namespace grpc { |
||||
namespace internal { |
||||
|
||||
namespace { |
||||
|
||||
grpc_core::TraceFlag grpc_environment_autodetect_trace( |
||||
false, "environment_autodetect"); |
||||
|
||||
// This is not a definite method to get the namespace name for GKE, but it is
|
||||
// the best we have.
|
||||
std::string GetNamespaceName() { |
||||
// Read the root file.
|
||||
const char* filename = |
||||
"/var/run/secrets/kubernetes.io/serviceaccount/namespace"; |
||||
auto namespace_name = grpc_core::LoadFile(filename, false); |
||||
if (!namespace_name.ok()) { |
||||
if (GRPC_TRACE_FLAG_ENABLED(grpc_environment_autodetect_trace)) { |
||||
gpr_log(GPR_DEBUG, "Reading file %s failed: %s", filename, |
||||
grpc_core::StatusToString(namespace_name.status()).c_str()); |
||||
} |
||||
// Fallback on an environment variable
|
||||
return grpc_core::GetEnv("NAMESPACE_NAME").value_or(""); |
||||
} |
||||
return std::string(reinterpret_cast<const char*>((*namespace_name).begin()), |
||||
(*namespace_name).length()); |
||||
} |
||||
|
||||
// Get pod name for GKE
|
||||
std::string GetPodName() { |
||||
auto pod_name = grpc_core::GetEnv("POD_NAME"); |
||||
if (pod_name.has_value()) { |
||||
return pod_name.value(); |
||||
} |
||||
return grpc_core::GetEnv("HOSTNAME").value_or(""); |
||||
} |
||||
|
||||
// Get container name for GKE
|
||||
std::string GetContainerName() { |
||||
return grpc_core::GetEnv("HOSTNAME").value_or(""); |
||||
} |
||||
|
||||
// Get function name for Cloud Functions
|
||||
std::string GetFunctionName() { |
||||
auto k_service = grpc_core::GetEnv("K_SERVICE"); |
||||
if (k_service.has_value()) { |
||||
return k_service.value(); |
||||
} |
||||
return grpc_core::GetEnv("FUNCTION_NAME").value_or(""); |
||||
} |
||||
|
||||
// Get revision name for Cloud run
|
||||
std::string GetRevisionName() { |
||||
return grpc_core::GetEnv("K_REVISION").value_or(""); |
||||
} |
||||
|
||||
// Get service name for Cloud run
|
||||
std::string GetServiceName() { |
||||
return grpc_core::GetEnv("K_SERVICE").value_or(""); |
||||
} |
||||
|
||||
// Get configuration name for Cloud run
|
||||
std::string GetConfiguratioName() { |
||||
return grpc_core::GetEnv("K_CONFIGURATION").value_or(""); |
||||
} |
||||
|
||||
// Get module ID for App Engine
|
||||
std::string GetModuleId() { |
||||
return grpc_core::GetEnv("GAE_SERVICE").value_or(""); |
||||
} |
||||
|
||||
// Get version ID for App Engine
|
||||
std::string GetVersionId() { |
||||
return grpc_core::GetEnv("GAE_VERSION").value_or(""); |
||||
} |
||||
|
||||
// Fire and forget class
|
||||
class EnvironmentAutoDetectHelper |
||||
: public grpc_core::InternallyRefCounted<EnvironmentAutoDetectHelper>, |
||||
private internal::GrpcLibrary { |
||||
public: |
||||
EnvironmentAutoDetectHelper( |
||||
std::string project_id, |
||||
absl::AnyInvocable<void(EnvironmentAutoDetect::ResourceType)> on_done, |
||||
std::shared_ptr<grpc_event_engine::experimental::EventEngine> |
||||
event_engine) |
||||
: InternallyRefCounted(/*trace=*/nullptr, /*initial_refcount=*/2), |
||||
project_id_(std::move(project_id)), |
||||
on_done_(std::move(on_done)), |
||||
event_engine_(std::move(event_engine)) { |
||||
grpc_core::ExecCtx exec_ctx; |
||||
// TODO(yashykt): The pollset stuff should go away once the HTTP library is
|
||||
// ported over to use EventEngine.
|
||||
pollset_ = static_cast<grpc_pollset*>(gpr_zalloc(grpc_pollset_size())); |
||||
grpc_pollset_init(pollset_, &mu_poll_); |
||||
pollent_ = grpc_polling_entity_create_from_pollset(pollset_); |
||||
// TODO(yashykt): Note that using EventEngine::Run is not fork-safe. If we
|
||||
// want to make this fork-safe, we might need some re-work here.
|
||||
event_engine_->Run([this] { PollLoop(); }); |
||||
AutoDetect(); |
||||
} |
||||
|
||||
~EnvironmentAutoDetectHelper() override { |
||||
grpc_core::ExecCtx exec_ctx; |
||||
grpc_pollset_shutdown( |
||||
pollset_, GRPC_CLOSURE_CREATE( |
||||
[](void* arg, absl::Status /* status */) { |
||||
grpc_pollset_destroy(static_cast<grpc_pollset*>(arg)); |
||||
gpr_free(arg); |
||||
}, |
||||
pollset_, nullptr)); |
||||
} |
||||
|
||||
void Orphan() override { |
||||
grpc_core::Crash("Illegal Orphan() call on EnvironmentAutoDetectHelper."); |
||||
} |
||||
|
||||
private: |
||||
struct Attribute { |
||||
std::string resource_attribute; |
||||
std::string metadata_server_atttribute; |
||||
}; |
||||
|
||||
void PollLoop() { |
||||
grpc_core::ExecCtx exec_ctx; |
||||
bool done = false; |
||||
gpr_mu_lock(mu_poll_); |
||||
grpc_pollset_worker* worker = nullptr; |
||||
if (!GRPC_LOG_IF_ERROR( |
||||
"pollset_work", |
||||
grpc_pollset_work(grpc_polling_entity_pollset(&pollent_), &worker, |
||||
grpc_core::Timestamp::InfPast()))) { |
||||
notify_poller_ = true; |
||||
} |
||||
done = notify_poller_; |
||||
gpr_mu_unlock(mu_poll_); |
||||
if (!done) { |
||||
event_engine_->RunAfter(grpc_core::Duration::Milliseconds(100), |
||||
[this] { PollLoop(); }); |
||||
} else { |
||||
Unref(); |
||||
} |
||||
} |
||||
|
||||
void AutoDetect() { |
||||
grpc_core::MutexLock lock(&mu_); |
||||
// GKE
|
||||
resource_.labels.emplace("project_id", project_id_); |
||||
if (grpc_core::GetEnv("KUBERNETES_SERVICE_HOST").has_value()) { |
||||
resource_.resource_type = "k8s_container"; |
||||
resource_.labels.emplace("namespace_name", GetNamespaceName()); |
||||
resource_.labels.emplace("pod_name", GetPodName()); |
||||
resource_.labels.emplace("container_name", GetContainerName()); |
||||
attributes_to_fetch_.emplace(grpc_core::MetadataQuery::kZoneAttribute, |
||||
"location"); |
||||
attributes_to_fetch_.emplace( |
||||
grpc_core::MetadataQuery::kClusterNameAttribute, "cluster_name"); |
||||
} |
||||
// Cloud Functions
|
||||
else if (grpc_core::GetEnv("FUNCTION_NAME").has_value() || |
||||
grpc_core::GetEnv("FUNCTION_TARGET").has_value()) { |
||||
resource_.resource_type = "cloud_function"; |
||||
resource_.labels.emplace("function_name", GetFunctionName()); |
||||
attributes_to_fetch_.emplace(grpc_core::MetadataQuery::kRegionAttribute, |
||||
"region"); |
||||
} |
||||
// Cloud Run
|
||||
else if (grpc_core::GetEnv("K_CONFIGURATION").has_value()) { |
||||
resource_.resource_type = "cloud_run_revision"; |
||||
resource_.labels.emplace("revision_name", GetRevisionName()); |
||||
resource_.labels.emplace("service_name", GetServiceName()); |
||||
resource_.labels.emplace("configuration_name", GetConfiguratioName()); |
||||
attributes_to_fetch_.emplace(grpc_core::MetadataQuery::kRegionAttribute, |
||||
"location"); |
||||
} |
||||
// App Engine
|
||||
else if (grpc_core::GetEnv("GAE_SERVICE").has_value()) { |
||||
resource_.resource_type = "gae_app"; |
||||
resource_.labels.emplace("module_id", GetModuleId()); |
||||
resource_.labels.emplace("version_id", GetVersionId()); |
||||
attributes_to_fetch_.emplace(grpc_core::MetadataQuery::kZoneAttribute, |
||||
"zone"); |
||||
} |
||||
// Assume GCE
|
||||
else { |
||||
assuming_gce_ = true; |
||||
resource_.resource_type = "gce_instance"; |
||||
attributes_to_fetch_.emplace( |
||||
grpc_core::MetadataQuery::kInstanceIdAttribute, "instance_id"); |
||||
attributes_to_fetch_.emplace(grpc_core::MetadataQuery::kZoneAttribute, |
||||
"zone"); |
||||
} |
||||
FetchMetadataServerAttributesAsynchronouslyLocked(); |
||||
} |
||||
|
||||
void FetchMetadataServerAttributesAsynchronouslyLocked() |
||||
ABSL_EXCLUSIVE_LOCKS_REQUIRED(mu_) { |
||||
GPR_ASSERT(!attributes_to_fetch_.empty()); |
||||
for (auto& element : attributes_to_fetch_) { |
||||
queries_.push_back(grpc_core::MakeOrphanable<grpc_core::MetadataQuery>( |
||||
element.first, &pollent_, |
||||
[this](std::string attribute, absl::StatusOr<std::string> result) { |
||||
if (GRPC_TRACE_FLAG_ENABLED(grpc_environment_autodetect_trace)) { |
||||
gpr_log( |
||||
GPR_INFO, |
||||
"Environment AutoDetect: Attribute: \"%s\" Result: \"%s\"", |
||||
attribute.c_str(), |
||||
result.ok() |
||||
? result.value().c_str() |
||||
: grpc_core::StatusToString(result.status()).c_str()); |
||||
} |
||||
absl::optional<EnvironmentAutoDetect::ResourceType> resource; |
||||
{ |
||||
grpc_core::MutexLock lock(&mu_); |
||||
auto it = attributes_to_fetch_.find(attribute); |
||||
if (it != attributes_to_fetch_.end()) { |
||||
if (result.ok()) { |
||||
resource_.labels.emplace(std::move(it->second), |
||||
std::move(result).value()); |
||||
} |
||||
// If fetching from the MetadataServer failed and we were
|
||||
// assuming a GCE environment, fallback to "global".
|
||||
else if (assuming_gce_) { |
||||
assuming_gce_ = false; |
||||
resource_.resource_type = "global"; |
||||
} |
||||
attributes_to_fetch_.erase(it); |
||||
} else { |
||||
// This should not happen
|
||||
gpr_log(GPR_ERROR, |
||||
"An unexpected attribute was seen from the " |
||||
"MetadataServer: %s", |
||||
attribute.c_str()); |
||||
} |
||||
if (attributes_to_fetch_.empty()) { |
||||
resource = std::move(resource_); |
||||
} |
||||
} |
||||
if (resource.has_value()) { |
||||
gpr_mu_lock(mu_poll_); |
||||
notify_poller_ = true; |
||||
gpr_mu_unlock(mu_poll_); |
||||
auto on_done = std::move(on_done_); |
||||
Unref(); |
||||
on_done(std::move(resource).value()); |
||||
} |
||||
}, |
||||
grpc_core::Duration::Seconds(1))); |
||||
} |
||||
} |
||||
|
||||
const std::string project_id_; |
||||
grpc_pollset* pollset_ = nullptr; |
||||
grpc_polling_entity pollent_; |
||||
gpr_mu* mu_poll_ = nullptr; |
||||
absl::AnyInvocable<void(EnvironmentAutoDetect::ResourceType)> on_done_; |
||||
std::shared_ptr<grpc_event_engine::experimental::EventEngine> event_engine_; |
||||
grpc_core::Mutex mu_; |
||||
bool notify_poller_ = false; |
||||
absl::flat_hash_map<std::string /* metadata_server_attribute */, |
||||
std::string /* resource_attribute */> |
||||
attributes_to_fetch_ ABSL_GUARDED_BY(mu_); |
||||
std::vector<grpc_core::OrphanablePtr<grpc_core::MetadataQuery>> queries_ |
||||
ABSL_GUARDED_BY(mu_); |
||||
EnvironmentAutoDetect::ResourceType resource_ ABSL_GUARDED_BY(mu_); |
||||
// This would be true if we are assuming the resource to be GCE. In this case,
|
||||
// there is a chance that it will fail and we should instead just use
|
||||
// "global".
|
||||
bool assuming_gce_ ABSL_GUARDED_BY(mu_) = false; |
||||
}; |
||||
|
||||
EnvironmentAutoDetect* g_autodetect = nullptr; |
||||
|
||||
} // namespace
|
||||
|
||||
void EnvironmentAutoDetect::Create(std::string project_id) { |
||||
GPR_ASSERT(g_autodetect == nullptr && !project_id.empty()); |
||||
g_autodetect = new EnvironmentAutoDetect(project_id); |
||||
} |
||||
|
||||
EnvironmentAutoDetect& EnvironmentAutoDetect::Get() { return *g_autodetect; } |
||||
|
||||
EnvironmentAutoDetect::EnvironmentAutoDetect(std::string project_id) |
||||
: project_id_(std::move(project_id)), |
||||
event_engine_(grpc_event_engine::experimental::GetDefaultEventEngine()) { |
||||
GPR_ASSERT(!project_id_.empty()); |
||||
new EnvironmentAutoDetectHelper( |
||||
project_id_, |
||||
[this](EnvironmentAutoDetect::ResourceType resource) { |
||||
std::vector<absl::AnyInvocable<void()>> callbacks; |
||||
{ |
||||
grpc_core::MutexLock lock(&mu_); |
||||
resource_ = std::make_unique<EnvironmentAutoDetect::ResourceType>( |
||||
std::move(resource)); |
||||
callbacks = std::move(callbacks_); |
||||
} |
||||
for (auto& callback : callbacks) { |
||||
callback(); |
||||
} |
||||
}, |
||||
event_engine_); |
||||
} |
||||
|
||||
void EnvironmentAutoDetect::NotifyOnDone(absl::AnyInvocable<void()> callback) { |
||||
grpc_core::MutexLock lock(&mu_); |
||||
// Environment has already been detected
|
||||
if (resource_ != nullptr) { |
||||
// Execute on the event engine to avoid deadlocks.
|
||||
return event_engine_->Run(std::move(callback)); |
||||
} |
||||
callbacks_.push_back(std::move(callback)); |
||||
} |
||||
|
||||
} // namespace internal
|
||||
} // namespace grpc
|
||||
@ -0,0 +1,83 @@ |
||||
//
|
||||
// Copyright 2023 gRPC authors.
|
||||
//
|
||||
// Licensed under the Apache License, Version 2.0 (the "License");
|
||||
// you may not use this file except in compliance with the License.
|
||||
// You may obtain a copy of the License at
|
||||
//
|
||||
// http://www.apache.org/licenses/LICENSE-2.0
|
||||
//
|
||||
// Unless required by applicable law or agreed to in writing, software
|
||||
// distributed under the License is distributed on an "AS IS" BASIS,
|
||||
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
// See the License for the specific language governing permissions and
|
||||
// limitations under the License.
|
||||
//
|
||||
|
||||
#ifndef GRPC_SRC_CPP_EXT_GCP_ENVIRONMENT_AUTODETECT_H |
||||
#define GRPC_SRC_CPP_EXT_GCP_ENVIRONMENT_AUTODETECT_H |
||||
|
||||
#include <grpc/support/port_platform.h> |
||||
|
||||
#include <map> |
||||
#include <memory> |
||||
#include <string> |
||||
#include <vector> |
||||
|
||||
#include "absl/base/thread_annotations.h" |
||||
#include "absl/functional/any_invocable.h" |
||||
#include "absl/status/status.h" |
||||
|
||||
#include <grpc/event_engine/event_engine.h> |
||||
|
||||
#include "src/core/lib/gprpp/sync.h" |
||||
|
||||
namespace grpc { |
||||
|
||||
namespace experimental { |
||||
// Forward declaration for GcpObservabilityInit
|
||||
absl::Status GcpObservabilityInit(); |
||||
} // namespace experimental
|
||||
|
||||
namespace internal { |
||||
|
||||
class EnvironmentAutoDetect { |
||||
public: |
||||
struct ResourceType { |
||||
// For example, "gce_instance", "gke_container", etc.
|
||||
std::string resource_type; |
||||
// Values for all the labels listed in the associated resource type.
|
||||
std::map<std::string, std::string> labels; |
||||
}; |
||||
|
||||
EnvironmentAutoDetect& Get(); |
||||
|
||||
// Exposed for testing purposes only
|
||||
explicit EnvironmentAutoDetect(std::string project_id); |
||||
|
||||
// \a callback will be invoked once the environment is done being detected.
|
||||
void NotifyOnDone(absl::AnyInvocable<void()> callback); |
||||
|
||||
const ResourceType* resource() { |
||||
grpc_core::MutexLock lock(&mu_); |
||||
return resource_.get(); |
||||
} |
||||
|
||||
private: |
||||
friend absl::Status grpc::experimental::GcpObservabilityInit(); |
||||
|
||||
// GcpObservabilityInit() is responsible for setting up the singleton with the
|
||||
// project_id.
|
||||
void Create(std::string project_id); |
||||
|
||||
const std::string project_id_; |
||||
std::shared_ptr<grpc_event_engine::experimental::EventEngine> event_engine_; |
||||
grpc_core::Mutex mu_; |
||||
std::unique_ptr<ResourceType> resource_ ABSL_GUARDED_BY(mu_); |
||||
std::vector<absl::AnyInvocable<void()>> callbacks_ ABSL_GUARDED_BY(mu_); |
||||
}; |
||||
|
||||
} // namespace internal
|
||||
} // namespace grpc
|
||||
|
||||
#endif // GRPC_SRC_CPP_EXT_GCP_ENVIRONMENT_AUTODETECT_H
|
||||
@ -0,0 +1,146 @@ |
||||
//
|
||||
//
|
||||
// Copyright 2023 gRPC authors.
|
||||
//
|
||||
// Licensed under the Apache License, Version 2.0 (the "License");
|
||||
// you may not use this file except in compliance with the License.
|
||||
// You may obtain a copy of the License at
|
||||
//
|
||||
// http://www.apache.org/licenses/LICENSE-2.0
|
||||
//
|
||||
// Unless required by applicable law or agreed to in writing, software
|
||||
// distributed under the License is distributed on an "AS IS" BASIS,
|
||||
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
// See the License for the specific language governing permissions and
|
||||
// limitations under the License.
|
||||
//
|
||||
//
|
||||
|
||||
#include "src/cpp/ext/gcp/environment_autodetect.h" |
||||
|
||||
#include <string> |
||||
#include <thread> // NOLINT |
||||
#include <vector> |
||||
|
||||
#include "absl/strings/string_view.h" |
||||
#include "absl/synchronization/notification.h" |
||||
#include "gmock/gmock.h" |
||||
#include "gtest/gtest.h" |
||||
|
||||
#include <grpc/grpc.h> |
||||
|
||||
#include "src/core/lib/gprpp/env.h" |
||||
#include "src/core/lib/gprpp/notification.h" |
||||
#include "test/core/util/test_config.h" |
||||
|
||||
namespace grpc { |
||||
namespace testing { |
||||
|
||||
namespace { |
||||
|
||||
class EnvironmentAutoDetectTest : public ::testing::Test { |
||||
protected: |
||||
void GetNotifiedOnEnvironmentDetection( |
||||
grpc::internal::EnvironmentAutoDetect* env, |
||||
grpc_core::Notification* notify) { |
||||
env->NotifyOnDone([notify]() { notify->Notify(); }); |
||||
} |
||||
}; |
||||
|
||||
// TODO(yashykt): We could create a mock MetadataServer to test this more end to
|
||||
// end, but given that that should be covered by our integration testing so
|
||||
// deferring to that.
|
||||
|
||||
TEST_F(EnvironmentAutoDetectTest, Basic) { |
||||
grpc::internal::EnvironmentAutoDetect env("project"); |
||||
|
||||
grpc_core::Notification notify; |
||||
GetNotifiedOnEnvironmentDetection(&env, ¬ify); |
||||
notify.WaitForNotification(); |
||||
|
||||
// Unless we test in a specific GCP resource, we should get "global" here.
|
||||
// EXPECT_EQ(env.resource()->resource_type, "global");
|
||||
EXPECT_EQ((env.resource()->labels).at("project_id"), "project"); |
||||
} |
||||
|
||||
TEST_F(EnvironmentAutoDetectTest, GkeEnvironment) { |
||||
grpc_core::SetEnv("KUBERNETES_SERVICE_HOST", "k8s_service_host"); |
||||
grpc::internal::EnvironmentAutoDetect env("project"); |
||||
|
||||
grpc_core::Notification notify; |
||||
GetNotifiedOnEnvironmentDetection(&env, ¬ify); |
||||
notify.WaitForNotification(); |
||||
|
||||
EXPECT_EQ(env.resource()->resource_type, "k8s_container"); |
||||
EXPECT_EQ((env.resource()->labels).at("project_id"), "project"); |
||||
grpc_core::UnsetEnv("KUBERNETES_SERVICE_HOST"); |
||||
} |
||||
|
||||
TEST_F(EnvironmentAutoDetectTest, CloudFunctions) { |
||||
grpc_core::SetEnv("FUNCTION_NAME", "function_name"); |
||||
grpc::internal::EnvironmentAutoDetect env("project"); |
||||
|
||||
grpc_core::Notification notify; |
||||
GetNotifiedOnEnvironmentDetection(&env, ¬ify); |
||||
notify.WaitForNotification(); |
||||
|
||||
EXPECT_EQ(env.resource()->resource_type, "cloud_function"); |
||||
EXPECT_EQ((env.resource()->labels).at("project_id"), "project"); |
||||
grpc_core::UnsetEnv("FUNCTION_NAME"); |
||||
} |
||||
|
||||
TEST_F(EnvironmentAutoDetectTest, CloudRun) { |
||||
grpc_core::SetEnv("K_CONFIGURATION", "config"); |
||||
grpc::internal::EnvironmentAutoDetect env("project"); |
||||
|
||||
grpc_core::Notification notify; |
||||
GetNotifiedOnEnvironmentDetection(&env, ¬ify); |
||||
notify.WaitForNotification(); |
||||
|
||||
EXPECT_EQ(env.resource()->resource_type, "cloud_run_revision"); |
||||
EXPECT_EQ((env.resource()->labels).at("project_id"), "project"); |
||||
grpc_core::UnsetEnv("K_CONFIGURATION"); |
||||
} |
||||
|
||||
TEST_F(EnvironmentAutoDetectTest, AppEngine) { |
||||
grpc_core::SetEnv("K_CONFIGURATION", "config"); |
||||
grpc::internal::EnvironmentAutoDetect env("project"); |
||||
|
||||
grpc_core::Notification notify; |
||||
GetNotifiedOnEnvironmentDetection(&env, ¬ify); |
||||
notify.WaitForNotification(); |
||||
|
||||
EXPECT_EQ(env.resource()->resource_type, "cloud_run_revision"); |
||||
EXPECT_EQ((env.resource()->labels).at("project_id"), "project"); |
||||
grpc_core::UnsetEnv("K_CONFIGURATION"); |
||||
} |
||||
|
||||
TEST_F(EnvironmentAutoDetectTest, MultipleNotifyWaiters) { |
||||
grpc::internal::EnvironmentAutoDetect env("project"); |
||||
|
||||
grpc_core::Notification notify[10]; |
||||
for (int i = 0; i < 10; ++i) { |
||||
GetNotifiedOnEnvironmentDetection(&env, ¬ify[i]); |
||||
} |
||||
for (int i = 0; i < 10; ++i) { |
||||
notify[i].WaitForNotification(); |
||||
} |
||||
|
||||
// Unless we test in a specific GCP resource, we should get "global" here.
|
||||
// EXPECT_EQ(env.resource()->resource_type, "global");
|
||||
EXPECT_EQ((env.resource()->labels).at("project_id"), "project"); |
||||
} |
||||
|
||||
} // namespace
|
||||
|
||||
} // namespace testing
|
||||
} // namespace grpc
|
||||
|
||||
int main(int argc, char** argv) { |
||||
grpc::testing::TestEnvironment env(&argc, argv); |
||||
::testing::InitGoogleTest(&argc, argv); |
||||
grpc_init(); |
||||
int ret_val = RUN_ALL_TESTS(); |
||||
grpc_shutdown(); |
||||
return ret_val; |
||||
} |
||||
Loading…
Reference in new issue