mirror of https://github.com/grpc/grpc.git
Revert "[GSM Observability] "Revert Metadata Exchange Implementation"" (#34234)
Reverts grpc/grpc#34233pull/33646/head
Yash Tibrewal
1 year ago
committed by
GitHub
parent
4f80a4f9aa
commit
0dd8a056b8
20 changed files with 1223 additions and 182 deletions
@ -0,0 +1,344 @@ |
||||
//
|
||||
//
|
||||
// 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/gsm/metadata_exchange.h" |
||||
|
||||
#include <stddef.h> |
||||
|
||||
#include <array> |
||||
#include <cstdint> |
||||
#include <unordered_map> |
||||
|
||||
#include "absl/meta/type_traits.h" |
||||
#include "absl/strings/escaping.h" |
||||
#include "absl/strings/string_view.h" |
||||
#include "absl/types/optional.h" |
||||
#include "absl/types/variant.h" |
||||
#include "google/protobuf/struct.upb.h" |
||||
#include "opentelemetry/sdk/resource/semantic_conventions.h" |
||||
#include "upb/base/string_view.h" |
||||
#include "upb/mem/arena.h" |
||||
#include "upb/upb.hpp" |
||||
|
||||
#include "src/core/lib/gprpp/env.h" |
||||
|
||||
namespace grpc { |
||||
namespace internal { |
||||
|
||||
namespace { |
||||
|
||||
// The keys that will be used in the Metadata Exchange between local and remote.
|
||||
constexpr absl::string_view kMetadataExchangeTypeKey = "type"; |
||||
constexpr absl::string_view kMetadataExchangePodNameKey = "pod_name"; |
||||
constexpr absl::string_view kMetadataExchangeContainerNameKey = |
||||
"container_name"; |
||||
constexpr absl::string_view kMetadataExchangeNamespaceNameKey = |
||||
"namespace_name"; |
||||
constexpr absl::string_view kMetadataExchangeClusterNameKey = "cluster_name"; |
||||
constexpr absl::string_view kMetadataExchangeLocationKey = "location"; |
||||
constexpr absl::string_view kMetadataExchangeProjectIdKey = "project_id"; |
||||
constexpr absl::string_view kMetadataExchangeCanonicalServiceKey = |
||||
"canonical_service"; |
||||
// The keys that will be used for the peer attributes when recording metrics.
|
||||
constexpr absl::string_view kPeerTypeAttribute = "gsm.remote_workload_type"; |
||||
constexpr absl::string_view kPeerPodNameAttribute = |
||||
"gsm.remote_workload_pod_name"; |
||||
constexpr absl::string_view kPeerContainerNameAttribute = |
||||
"gsm.remote_workload_container_name"; |
||||
constexpr absl::string_view kPeerNamespaceNameAttribute = |
||||
"gsm.remote_workload_namespace_name"; |
||||
constexpr absl::string_view kPeerClusterNameAttribute = |
||||
"gsm.remote_workload_cluster_name"; |
||||
constexpr absl::string_view kPeerLocationAttribute = |
||||
"gsm.remote_workload_location"; |
||||
constexpr absl::string_view kPeerProjectIdAttribute = |
||||
"gsm.remote_workload_project_id"; |
||||
constexpr absl::string_view kPeerCanonicalServiceAttribute = |
||||
"gsm.remote_workload_canonical_service"; |
||||
// Type values used by Google Cloud Resource Detector
|
||||
constexpr absl::string_view kGkeType = "gcp_kubernetes_engine"; |
||||
|
||||
enum class GcpResourceType : std::uint8_t { kGke, kUnknown }; |
||||
|
||||
GcpResourceType StringToGcpResourceType(absl::string_view type) { |
||||
if (type == kGkeType) { |
||||
return GcpResourceType::kGke; |
||||
} |
||||
return GcpResourceType::kUnknown; |
||||
} |
||||
|
||||
upb_StringView AbslStrToUpbStr(absl::string_view str) { |
||||
return upb_StringView_FromDataAndSize(str.data(), str.size()); |
||||
} |
||||
|
||||
absl::string_view UpbStrToAbslStr(upb_StringView str) { |
||||
return absl::string_view(str.data, str.size); |
||||
} |
||||
|
||||
void AddStringKeyValueToStructProto(google_protobuf_Struct* struct_pb, |
||||
absl::string_view key, |
||||
absl::string_view value, upb_Arena* arena) { |
||||
google_protobuf_Value* value_pb = google_protobuf_Value_new(arena); |
||||
google_protobuf_Value_set_string_value(value_pb, AbslStrToUpbStr(value)); |
||||
google_protobuf_Struct_fields_set(struct_pb, AbslStrToUpbStr(key), value_pb, |
||||
arena); |
||||
} |
||||
|
||||
absl::string_view GetStringValueFromAttributeMap( |
||||
const opentelemetry::sdk::common::AttributeMap& map, |
||||
absl::string_view key) { |
||||
const auto& attributes = map.GetAttributes(); |
||||
const auto it = attributes.find(std::string(key)); |
||||
if (it == attributes.end()) { |
||||
return "unknown"; |
||||
} |
||||
const auto* string_value = absl::get_if<std::string>(&it->second); |
||||
if (string_value == nullptr) { |
||||
return "unknown"; |
||||
} |
||||
return *string_value; |
||||
} |
||||
|
||||
absl::string_view GetStringValueFromUpbStruct(google_protobuf_Struct* struct_pb, |
||||
absl::string_view key, |
||||
upb_Arena* arena) { |
||||
if (struct_pb == nullptr) { |
||||
return "unknown"; |
||||
} |
||||
google_protobuf_Value* value_pb = google_protobuf_Value_new(arena); |
||||
bool present = google_protobuf_Struct_fields_get( |
||||
struct_pb, AbslStrToUpbStr(key), &value_pb); |
||||
if (present) { |
||||
if (google_protobuf_Value_has_string_value(value_pb)) { |
||||
return UpbStrToAbslStr(google_protobuf_Value_string_value(value_pb)); |
||||
} |
||||
} |
||||
return "unknown"; |
||||
} |
||||
|
||||
class LocalLabelsIterable : public LabelsIterable { |
||||
public: |
||||
explicit LocalLabelsIterable( |
||||
const std::vector<std::pair<absl::string_view, std::string>>& labels) |
||||
: labels_(labels) {} |
||||
|
||||
absl::optional<std::pair<absl::string_view, absl::string_view>> Next() |
||||
override { |
||||
if (pos_ >= labels_.size()) { |
||||
return absl::nullopt; |
||||
} |
||||
return labels_[pos_++]; |
||||
} |
||||
|
||||
size_t Size() const override { return labels_.size(); } |
||||
|
||||
void ResetIteratorPosition() override { pos_ = 0; } |
||||
|
||||
private: |
||||
size_t pos_ = 0; |
||||
const std::vector<std::pair<absl::string_view, std::string>>& labels_; |
||||
}; |
||||
|
||||
class PeerLabelsIterable : public LabelsIterable { |
||||
public: |
||||
explicit PeerLabelsIterable(grpc_core::Slice remote_metadata) |
||||
: metadata_(std::move(remote_metadata)) {} |
||||
|
||||
absl::optional<std::pair<absl::string_view, absl::string_view>> Next() |
||||
override { |
||||
auto& struct_pb = GetDecodedMetadata(); |
||||
if (struct_pb.struct_pb == nullptr) { |
||||
return absl::nullopt; |
||||
} |
||||
if (++pos_ == 1) { |
||||
return std::make_pair(kPeerTypeAttribute, |
||||
GetStringValueFromUpbStruct( |
||||
struct_pb.struct_pb, kMetadataExchangeTypeKey, |
||||
struct_pb.arena.ptr())); |
||||
} |
||||
// Only handle GKE type for now.
|
||||
switch (type_) { |
||||
case GcpResourceType::kGke: |
||||
if (pos_ - 2 >= kGkeAttributeList.size()) { |
||||
return absl::nullopt; |
||||
} |
||||
return std::make_pair( |
||||
kGkeAttributeList[pos_ - 2].otel_attribute, |
||||
GetStringValueFromUpbStruct( |
||||
struct_pb.struct_pb, |
||||
kGkeAttributeList[pos_ - 2].metadata_attribute, |
||||
struct_pb.arena.ptr())); |
||||
case GcpResourceType::kUnknown: |
||||
return absl::nullopt; |
||||
} |
||||
} |
||||
|
||||
size_t Size() const override { |
||||
auto& struct_pb = GetDecodedMetadata(); |
||||
if (struct_pb.struct_pb == nullptr) { |
||||
return 0; |
||||
} |
||||
if (type_ != GcpResourceType::kGke) { |
||||
return 1; |
||||
} |
||||
return kGkeAttributeList.size(); |
||||
} |
||||
|
||||
void ResetIteratorPosition() override { pos_ = 0; } |
||||
|
||||
private: |
||||
struct GkeAttribute { |
||||
absl::string_view otel_attribute; |
||||
absl::string_view metadata_attribute; |
||||
}; |
||||
|
||||
struct StructPb { |
||||
upb::Arena arena; |
||||
google_protobuf_Struct* struct_pb = nullptr; |
||||
}; |
||||
|
||||
static constexpr std::array<GkeAttribute, 7> kGkeAttributeList = { |
||||
GkeAttribute{kPeerPodNameAttribute, kMetadataExchangePodNameKey}, |
||||
GkeAttribute{kPeerContainerNameAttribute, |
||||
kMetadataExchangeContainerNameKey}, |
||||
GkeAttribute{kPeerNamespaceNameAttribute, |
||||
kMetadataExchangeNamespaceNameKey}, |
||||
GkeAttribute{kPeerClusterNameAttribute, kMetadataExchangeClusterNameKey}, |
||||
GkeAttribute{kPeerLocationAttribute, kMetadataExchangeLocationKey}, |
||||
GkeAttribute{kPeerProjectIdAttribute, kMetadataExchangeProjectIdKey}, |
||||
GkeAttribute{kPeerCanonicalServiceAttribute, |
||||
kMetadataExchangeCanonicalServiceKey}, |
||||
}; |
||||
|
||||
StructPb& GetDecodedMetadata() const { |
||||
auto* slice = absl::get_if<grpc_core::Slice>(&metadata_); |
||||
if (slice == nullptr) { |
||||
return absl::get<StructPb>(metadata_); |
||||
} |
||||
std::string decoded_metadata; |
||||
bool metadata_decoded = |
||||
absl::Base64Unescape(slice->as_string_view(), &decoded_metadata); |
||||
metadata_ = StructPb{}; |
||||
auto& struct_pb = absl::get<StructPb>(metadata_); |
||||
if (metadata_decoded) { |
||||
struct_pb.struct_pb = google_protobuf_Struct_parse( |
||||
decoded_metadata.c_str(), decoded_metadata.size(), |
||||
struct_pb.arena.ptr()); |
||||
type_ = StringToGcpResourceType(GetStringValueFromUpbStruct( |
||||
struct_pb.struct_pb, kMetadataExchangeTypeKey, |
||||
struct_pb.arena.ptr())); |
||||
} |
||||
return struct_pb; |
||||
} |
||||
|
||||
// Holds either the metadata slice or the decoded proto struct.
|
||||
mutable absl::variant<grpc_core::Slice, StructPb> metadata_; |
||||
mutable GcpResourceType type_; |
||||
uint32_t pos_ = 0; |
||||
}; |
||||
|
||||
constexpr std::array<PeerLabelsIterable::GkeAttribute, 7> |
||||
PeerLabelsIterable::kGkeAttributeList; |
||||
|
||||
} // namespace
|
||||
|
||||
ServiceMeshLabelsInjector::ServiceMeshLabelsInjector( |
||||
const opentelemetry::sdk::common::AttributeMap& map) { |
||||
upb::Arena arena; |
||||
auto* metadata = google_protobuf_Struct_new(arena.ptr()); |
||||
// Assume kubernetes for now
|
||||
absl::string_view type_value = GetStringValueFromAttributeMap( |
||||
map, opentelemetry::sdk::resource::SemanticConventions::kCloudPlatform); |
||||
absl::string_view pod_name_value = GetStringValueFromAttributeMap( |
||||
map, opentelemetry::sdk::resource::SemanticConventions::kK8sPodName); |
||||
absl::string_view container_name_value = GetStringValueFromAttributeMap( |
||||
map, |
||||
opentelemetry::sdk::resource::SemanticConventions::kK8sContainerName); |
||||
absl::string_view namespace_value = GetStringValueFromAttributeMap( |
||||
map, |
||||
opentelemetry::sdk::resource::SemanticConventions::kK8sNamespaceName); |
||||
absl::string_view cluster_name_value = GetStringValueFromAttributeMap( |
||||
map, opentelemetry::sdk::resource::SemanticConventions::kK8sClusterName); |
||||
absl::string_view cluster_location_value = GetStringValueFromAttributeMap( |
||||
map, opentelemetry::sdk::resource::SemanticConventions:: |
||||
kCloudRegion); // if regional
|
||||
if (cluster_location_value == "unknown") { |
||||
cluster_location_value = GetStringValueFromAttributeMap( |
||||
map, opentelemetry::sdk::resource::SemanticConventions:: |
||||
kCloudAvailabilityZone); // if zonal
|
||||
} |
||||
absl::string_view project_id_value = GetStringValueFromAttributeMap( |
||||
map, opentelemetry::sdk::resource::SemanticConventions::kCloudAccountId); |
||||
std::string canonical_service_value = |
||||
grpc_core::GetEnv("GSM_CANONICAL_SERVICE_NAME").value_or("unknown"); |
||||
// Create metadata to be sent over wire.
|
||||
AddStringKeyValueToStructProto(metadata, kMetadataExchangeTypeKey, type_value, |
||||
arena.ptr()); |
||||
// Only handle GKE for now
|
||||
if (type_value == kGkeType) { |
||||
AddStringKeyValueToStructProto(metadata, kMetadataExchangePodNameKey, |
||||
pod_name_value, arena.ptr()); |
||||
AddStringKeyValueToStructProto(metadata, kMetadataExchangeContainerNameKey, |
||||
container_name_value, arena.ptr()); |
||||
AddStringKeyValueToStructProto(metadata, kMetadataExchangeNamespaceNameKey, |
||||
namespace_value, arena.ptr()); |
||||
AddStringKeyValueToStructProto(metadata, kMetadataExchangeClusterNameKey, |
||||
cluster_name_value, arena.ptr()); |
||||
AddStringKeyValueToStructProto(metadata, kMetadataExchangeLocationKey, |
||||
cluster_location_value, arena.ptr()); |
||||
AddStringKeyValueToStructProto(metadata, kMetadataExchangeProjectIdKey, |
||||
project_id_value, arena.ptr()); |
||||
AddStringKeyValueToStructProto(metadata, |
||||
kMetadataExchangeCanonicalServiceKey, |
||||
canonical_service_value, arena.ptr()); |
||||
} |
||||
|
||||
size_t output_length; |
||||
char* output = |
||||
google_protobuf_Struct_serialize(metadata, arena.ptr(), &output_length); |
||||
serialized_labels_to_send_ = grpc_core::Slice::FromCopiedString( |
||||
absl::Base64Escape(absl::string_view(output, output_length))); |
||||
// Fill up local labels map. The rest we get from the detected Resource and
|
||||
// from the peer.
|
||||
// TODO(yashykt): Add mesh_id
|
||||
} |
||||
|
||||
std::unique_ptr<LabelsIterable> ServiceMeshLabelsInjector::GetPeerLabels( |
||||
grpc_metadata_batch* incoming_initial_metadata) { |
||||
auto peer_metadata = |
||||
incoming_initial_metadata->Take(grpc_core::XEnvoyPeerMetadata()); |
||||
if (!peer_metadata.has_value()) { |
||||
return nullptr; |
||||
} |
||||
return std::make_unique<PeerLabelsIterable>(*std::move(peer_metadata)); |
||||
} |
||||
|
||||
std::unique_ptr<LabelsIterable> ServiceMeshLabelsInjector::GetLocalLabels() { |
||||
return std::make_unique<LocalLabelsIterable>(local_labels_); |
||||
} |
||||
|
||||
void ServiceMeshLabelsInjector::AddLabels( |
||||
grpc_metadata_batch* outgoing_initial_metadata) { |
||||
outgoing_initial_metadata->Set(grpc_core::XEnvoyPeerMetadata(), |
||||
serialized_labels_to_send_.Ref()); |
||||
} |
||||
|
||||
} // namespace internal
|
||||
} // namespace grpc
|
||||
@ -0,0 +1,64 @@ |
||||
//
|
||||
//
|
||||
// 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_GSM_METADATA_EXCHANGE_H |
||||
#define GRPC_SRC_CPP_EXT_GSM_METADATA_EXCHANGE_H |
||||
|
||||
#include <grpc/support/port_platform.h> |
||||
|
||||
#include <memory> |
||||
#include <string> |
||||
#include <utility> |
||||
#include <vector> |
||||
|
||||
#include "absl/strings/string_view.h" |
||||
#include "opentelemetry/sdk/common/attribute_utils.h" |
||||
|
||||
#include "src/core/lib/slice/slice.h" |
||||
#include "src/core/lib/transport/metadata_batch.h" |
||||
#include "src/cpp/ext/otel/otel_plugin.h" |
||||
|
||||
namespace grpc { |
||||
namespace internal { |
||||
|
||||
class ServiceMeshLabelsInjector : public LabelsInjector { |
||||
public: |
||||
explicit ServiceMeshLabelsInjector( |
||||
const opentelemetry::sdk::common::AttributeMap& map); |
||||
// Read the incoming initial metadata to get the set of labels to be added to
|
||||
// metrics.
|
||||
std::unique_ptr<LabelsIterable> GetPeerLabels( |
||||
grpc_metadata_batch* incoming_initial_metadata) override; |
||||
|
||||
// Get the local labels to be added to metrics. To be used when the peer
|
||||
// metadata is not available, for example, for started RPCs metric.
|
||||
std::unique_ptr<LabelsIterable> GetLocalLabels() override; |
||||
|
||||
// Modify the outgoing initial metadata with metadata information to be sent
|
||||
// to the peer.
|
||||
void AddLabels(grpc_metadata_batch* outgoing_initial_metadata) override; |
||||
|
||||
private: |
||||
std::vector<std::pair<absl::string_view, std::string>> local_labels_; |
||||
grpc_core::Slice serialized_labels_to_send_; |
||||
}; |
||||
|
||||
} // namespace internal
|
||||
} // namespace grpc
|
||||
|
||||
#endif // GRPC_SRC_CPP_EXT_GSM_METADATA_EXCHANGE_H
|
||||
@ -0,0 +1,109 @@ |
||||
//
|
||||
//
|
||||
// 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_OTEL_KEY_VALUE_ITERABLE_H |
||||
#define GRPC_SRC_CPP_EXT_OTEL_KEY_VALUE_ITERABLE_H |
||||
|
||||
#include <grpc/support/port_platform.h> |
||||
|
||||
#include <stddef.h> |
||||
|
||||
#include <utility> |
||||
|
||||
#include "absl/strings/string_view.h" |
||||
#include "absl/types/optional.h" |
||||
#include "absl/types/span.h" |
||||
#include "opentelemetry/common/attribute_value.h" |
||||
#include "opentelemetry/common/key_value_iterable.h" |
||||
#include "opentelemetry/nostd/function_ref.h" |
||||
#include "opentelemetry/nostd/string_view.h" |
||||
|
||||
#include "src/cpp/ext/otel/otel_plugin.h" |
||||
|
||||
namespace grpc { |
||||
namespace internal { |
||||
|
||||
inline opentelemetry::nostd::string_view AbslStrViewToOTelStrView( |
||||
absl::string_view str) { |
||||
return opentelemetry::nostd::string_view(str.data(), str.size()); |
||||
} |
||||
|
||||
// An iterable class based on opentelemetry::common::KeyValueIterable that
|
||||
// allows gRPC to iterate on its various sources of attributes and avoid an
|
||||
// allocation in cases wherever possible.
|
||||
class KeyValueIterable : public opentelemetry::common::KeyValueIterable { |
||||
public: |
||||
explicit KeyValueIterable( |
||||
LabelsIterable* local_labels_iterable, |
||||
LabelsIterable* peer_labels_iterable, |
||||
absl::Span<const std::pair<absl::string_view, absl::string_view>> |
||||
additional_labels) |
||||
: local_labels_iterable_(local_labels_iterable), |
||||
peer_labels_iterable_(peer_labels_iterable), |
||||
additional_labels_(additional_labels) {} |
||||
|
||||
bool ForEachKeyValue(opentelemetry::nostd::function_ref< |
||||
bool(opentelemetry::nostd::string_view, |
||||
opentelemetry::common::AttributeValue)> |
||||
callback) const noexcept override { |
||||
if (local_labels_iterable_ != nullptr) { |
||||
local_labels_iterable_->ResetIteratorPosition(); |
||||
while (const auto& pair = local_labels_iterable_->Next()) { |
||||
if (!callback(AbslStrViewToOTelStrView(pair->first), |
||||
AbslStrViewToOTelStrView(pair->second))) { |
||||
return false; |
||||
} |
||||
} |
||||
} |
||||
if (peer_labels_iterable_ != nullptr) { |
||||
peer_labels_iterable_->ResetIteratorPosition(); |
||||
while (const auto& pair = peer_labels_iterable_->Next()) { |
||||
if (!callback(AbslStrViewToOTelStrView(pair->first), |
||||
AbslStrViewToOTelStrView(pair->second))) { |
||||
return false; |
||||
} |
||||
} |
||||
} |
||||
for (const auto& pair : additional_labels_) { |
||||
if (!callback(AbslStrViewToOTelStrView(pair.first), |
||||
AbslStrViewToOTelStrView(pair.second))) { |
||||
return false; |
||||
} |
||||
} |
||||
return true; |
||||
} |
||||
|
||||
size_t size() const noexcept override { |
||||
return (local_labels_iterable_ != nullptr ? local_labels_iterable_->Size() |
||||
: 0) + |
||||
(peer_labels_iterable_ != nullptr ? peer_labels_iterable_->Size() |
||||
: 0) + |
||||
additional_labels_.size(); |
||||
} |
||||
|
||||
private: |
||||
LabelsIterable* local_labels_iterable_; |
||||
LabelsIterable* peer_labels_iterable_; |
||||
absl::Span<const std::pair<absl::string_view, absl::string_view>> |
||||
additional_labels_; |
||||
}; |
||||
|
||||
} // namespace internal
|
||||
} // namespace grpc
|
||||
|
||||
#endif // GRPC_SRC_CPP_EXT_OTEL_KEY_VALUE_ITERABLE_H
|
||||
@ -0,0 +1,232 @@ |
||||
//
|
||||
//
|
||||
// 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/gsm/metadata_exchange.h" |
||||
|
||||
#include "absl/functional/any_invocable.h" |
||||
#include "api/include/opentelemetry/metrics/provider.h" |
||||
#include "gmock/gmock.h" |
||||
#include "google/cloud/opentelemetry/resource_detector.h" |
||||
#include "gtest/gtest.h" |
||||
#include "opentelemetry/sdk/metrics/meter_provider.h" |
||||
#include "opentelemetry/sdk/metrics/metric_reader.h" |
||||
|
||||
#include <grpcpp/grpcpp.h> |
||||
|
||||
#include "src/core/lib/channel/call_tracer.h" |
||||
#include "src/core/lib/config/core_configuration.h" |
||||
#include "src/core/lib/gprpp/env.h" |
||||
#include "src/cpp/ext/gsm/gsm_observability.h" |
||||
#include "src/cpp/ext/otel/otel_plugin.h" |
||||
#include "test/core/util/test_config.h" |
||||
#include "test/cpp/end2end/test_service_impl.h" |
||||
#include "test/cpp/ext/otel/otel_test_library.h" |
||||
|
||||
namespace grpc { |
||||
namespace testing { |
||||
namespace { |
||||
|
||||
class TestScenario { |
||||
public: |
||||
enum class Type : std::uint8_t { kGke, kUnknown }; |
||||
|
||||
explicit TestScenario(Type type) : type_(type) {} |
||||
|
||||
opentelemetry::sdk::resource::Resource GetTestResource() const { |
||||
switch (type_) { |
||||
case Type::kGke: |
||||
return TestGkeResource(); |
||||
case Type::kUnknown: |
||||
return TestUnknownResource(); |
||||
} |
||||
} |
||||
|
||||
static std::string Name(const ::testing::TestParamInfo<TestScenario>& info) { |
||||
switch (info.param.type_) { |
||||
case Type::kGke: |
||||
return "gke"; |
||||
case Type::kUnknown: |
||||
return "unknown"; |
||||
} |
||||
} |
||||
|
||||
Type type() const { return type_; } |
||||
|
||||
private: |
||||
static opentelemetry::sdk::resource::Resource TestGkeResource() { |
||||
opentelemetry::sdk::common::AttributeMap attributes; |
||||
attributes.SetAttribute("cloud.platform", "gcp_kubernetes_engine"); |
||||
attributes.SetAttribute("k8s.pod.name", "pod"); |
||||
attributes.SetAttribute("k8s.container.name", "container"); |
||||
attributes.SetAttribute("k8s.namespace.name", "namespace"); |
||||
attributes.SetAttribute("k8s.cluster.name", "cluster"); |
||||
attributes.SetAttribute("cloud.region", "region"); |
||||
attributes.SetAttribute("cloud.account.id", "id"); |
||||
return opentelemetry::sdk::resource::Resource::Create(attributes); |
||||
} |
||||
|
||||
static opentelemetry::sdk::resource::Resource TestUnknownResource() { |
||||
opentelemetry::sdk::common::AttributeMap attributes; |
||||
attributes.SetAttribute("cloud.platform", "random"); |
||||
return opentelemetry::sdk::resource::Resource::Create(attributes); |
||||
} |
||||
|
||||
Type type_; |
||||
}; |
||||
|
||||
class MetadataExchangeTest |
||||
: public OTelPluginEnd2EndTest, |
||||
public ::testing::WithParamInterface<TestScenario> { |
||||
protected: |
||||
void Init(const absl::flat_hash_set<absl::string_view>& metric_names) { |
||||
OTelPluginEnd2EndTest::Init( |
||||
metric_names, /*resource=*/GetParam().GetTestResource(), |
||||
/*labels_injector=*/ |
||||
std::make_unique<grpc::internal::ServiceMeshLabelsInjector>( |
||||
GetParam().GetTestResource().GetAttributes())); |
||||
} |
||||
|
||||
void VerifyGkeServiceMeshAttributes( |
||||
const std::map<std::string, |
||||
opentelemetry::sdk::common::OwnedAttributeValue>& |
||||
attributes, |
||||
bool local_only = false) { |
||||
if (!local_only) { |
||||
switch (GetParam().type()) { |
||||
case TestScenario::Type::kGke: |
||||
EXPECT_EQ( |
||||
absl::get<std::string>(attributes.at("gsm.remote_workload_type")), |
||||
"gcp_kubernetes_engine"); |
||||
EXPECT_EQ(absl::get<std::string>( |
||||
attributes.at("gsm.remote_workload_pod_name")), |
||||
"pod"); |
||||
EXPECT_EQ(absl::get<std::string>( |
||||
attributes.at("gsm.remote_workload_container_name")), |
||||
"container"); |
||||
EXPECT_EQ(absl::get<std::string>( |
||||
attributes.at("gsm.remote_workload_namespace_name")), |
||||
"namespace"); |
||||
EXPECT_EQ(absl::get<std::string>( |
||||
attributes.at("gsm.remote_workload_cluster_name")), |
||||
"cluster"); |
||||
EXPECT_EQ(absl::get<std::string>( |
||||
attributes.at("gsm.remote_workload_location")), |
||||
"region"); |
||||
EXPECT_EQ(absl::get<std::string>( |
||||
attributes.at("gsm.remote_workload_project_id")), |
||||
"id"); |
||||
EXPECT_EQ(absl::get<std::string>( |
||||
attributes.at("gsm.remote_workload_canonical_service")), |
||||
"canonical_service"); |
||||
break; |
||||
case TestScenario::Type::kUnknown: |
||||
EXPECT_EQ( |
||||
absl::get<std::string>(attributes.at("gsm.remote_workload_type")), |
||||
"random"); |
||||
break; |
||||
} |
||||
} |
||||
} |
||||
}; |
||||
|
||||
TEST_P(MetadataExchangeTest, ClientAttemptStarted) { |
||||
Init(/*metric_names=*/{ |
||||
grpc::internal::OTelClientAttemptStartedInstrumentName()}); |
||||
SendRPC(); |
||||
const char* kMetricName = "grpc.client.attempt.started"; |
||||
auto data = ReadCurrentMetricsData( |
||||
[&](const absl::flat_hash_map< |
||||
std::string, |
||||
std::vector<opentelemetry::sdk::metrics::PointDataAttributes>>& |
||||
data) { return !data.contains(kMetricName); }); |
||||
ASSERT_EQ(data[kMetricName].size(), 1); |
||||
auto point_data = absl::get_if<opentelemetry::sdk::metrics::SumPointData>( |
||||
&data[kMetricName][0].point_data); |
||||
ASSERT_NE(point_data, nullptr); |
||||
auto client_started_value = absl::get_if<int64_t>(&point_data->value_); |
||||
ASSERT_NE(client_started_value, nullptr); |
||||
EXPECT_EQ(*client_started_value, 1); |
||||
const auto& attributes = data[kMetricName][0].attributes.GetAttributes(); |
||||
EXPECT_EQ(absl::get<std::string>(attributes.at("grpc.method")), kMethodName); |
||||
EXPECT_EQ(absl::get<std::string>(attributes.at("grpc.target")), |
||||
canonical_server_address_); |
||||
VerifyGkeServiceMeshAttributes(attributes, /*local_only=*/true); |
||||
} |
||||
|
||||
TEST_P(MetadataExchangeTest, ClientAttemptDuration) { |
||||
Init(/*metric_names=*/{ |
||||
grpc::internal::OTelClientAttemptDurationInstrumentName()}); |
||||
SendRPC(); |
||||
const char* kMetricName = "grpc.client.attempt.duration"; |
||||
auto data = ReadCurrentMetricsData( |
||||
[&](const absl::flat_hash_map< |
||||
std::string, |
||||
std::vector<opentelemetry::sdk::metrics::PointDataAttributes>>& |
||||
data) { return !data.contains(kMetricName); }); |
||||
ASSERT_EQ(data[kMetricName].size(), 1); |
||||
auto point_data = |
||||
absl::get_if<opentelemetry::sdk::metrics::HistogramPointData>( |
||||
&data[kMetricName][0].point_data); |
||||
ASSERT_NE(point_data, nullptr); |
||||
ASSERT_EQ(point_data->count_, 1); |
||||
const auto& attributes = data[kMetricName][0].attributes.GetAttributes(); |
||||
EXPECT_EQ(absl::get<std::string>(attributes.at("grpc.method")), kMethodName); |
||||
EXPECT_EQ(absl::get<std::string>(attributes.at("grpc.target")), |
||||
canonical_server_address_); |
||||
EXPECT_EQ(absl::get<std::string>(attributes.at("grpc.status")), "OK"); |
||||
VerifyGkeServiceMeshAttributes(attributes); |
||||
} |
||||
|
||||
TEST_P(MetadataExchangeTest, ServerCallDuration) { |
||||
Init(/*metric_names=*/{ |
||||
grpc::internal::OTelServerCallDurationInstrumentName()}); |
||||
SendRPC(); |
||||
const char* kMetricName = "grpc.server.call.duration"; |
||||
auto data = ReadCurrentMetricsData( |
||||
[&](const absl::flat_hash_map< |
||||
std::string, |
||||
std::vector<opentelemetry::sdk::metrics::PointDataAttributes>>& |
||||
data) { return !data.contains(kMetricName); }); |
||||
ASSERT_EQ(data[kMetricName].size(), 1); |
||||
auto point_data = |
||||
absl::get_if<opentelemetry::sdk::metrics::HistogramPointData>( |
||||
&data[kMetricName][0].point_data); |
||||
ASSERT_NE(point_data, nullptr); |
||||
ASSERT_EQ(point_data->count_, 1); |
||||
const auto& attributes = data[kMetricName][0].attributes.GetAttributes(); |
||||
EXPECT_EQ(absl::get<std::string>(attributes.at("grpc.method")), kMethodName); |
||||
EXPECT_EQ(absl::get<std::string>(attributes.at("grpc.status")), "OK"); |
||||
VerifyGkeServiceMeshAttributes(attributes); |
||||
} |
||||
|
||||
INSTANTIATE_TEST_SUITE_P( |
||||
MetadataExchange, MetadataExchangeTest, |
||||
::testing::Values(TestScenario(TestScenario::Type::kGke), |
||||
TestScenario(TestScenario::Type::kUnknown)), |
||||
&TestScenario::Name); |
||||
|
||||
} // namespace
|
||||
} // namespace testing
|
||||
} // namespace grpc
|
||||
|
||||
int main(int argc, char** argv) { |
||||
grpc::testing::TestEnvironment env(&argc, argv); |
||||
::testing::InitGoogleTest(&argc, argv); |
||||
grpc_core::SetEnv("GSM_CANONICAL_SERVICE_NAME", "canonical_service"); |
||||
return RUN_ALL_TESTS(); |
||||
} |
||||
@ -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 "test/cpp/ext/otel/otel_test_library.h" |
||||
|
||||
#include "absl/functional/any_invocable.h" |
||||
#include "api/include/opentelemetry/metrics/provider.h" |
||||
#include "gmock/gmock.h" |
||||
#include "gtest/gtest.h" |
||||
#include "opentelemetry/sdk/metrics/meter_provider.h" |
||||
#include "opentelemetry/sdk/metrics/metric_reader.h" |
||||
|
||||
#include <grpcpp/grpcpp.h> |
||||
|
||||
#include "src/core/lib/channel/call_tracer.h" |
||||
#include "src/core/lib/config/core_configuration.h" |
||||
#include "test/core/util/test_config.h" |
||||
#include "test/cpp/end2end/test_service_impl.h" |
||||
|
||||
namespace grpc { |
||||
namespace testing { |
||||
|
||||
void OTelPluginEnd2EndTest::Init( |
||||
const absl::flat_hash_set<absl::string_view>& metric_names, |
||||
opentelemetry::sdk::resource::Resource resource, |
||||
std::unique_ptr<grpc::internal::LabelsInjector> labels_injector, |
||||
bool test_no_meter_provider) { |
||||
// We are resetting the MeterProvider and OpenTelemetry plugin at the start
|
||||
// of each test to avoid test results from one test carrying over to another
|
||||
// test. (Some measurements can get arbitrarily delayed.)
|
||||
auto meter_provider = |
||||
std::make_shared<opentelemetry::sdk::metrics::MeterProvider>( |
||||
std::make_unique<opentelemetry::sdk::metrics::ViewRegistry>(), |
||||
std::move(resource)); |
||||
reader_.reset(new grpc::testing::MockMetricReader); |
||||
meter_provider->AddMetricReader(reader_); |
||||
grpc_core::CoreConfiguration::Reset(); |
||||
grpc::internal::OpenTelemetryPluginBuilder ot_builder; |
||||
ot_builder.EnableMetrics(metric_names); |
||||
if (!test_no_meter_provider) { |
||||
auto meter_provider = |
||||
std::make_shared<opentelemetry::sdk::metrics::MeterProvider>(); |
||||
reader_.reset(new grpc::testing::MockMetricReader); |
||||
meter_provider->AddMetricReader(reader_); |
||||
ot_builder.SetMeterProvider(std::move(meter_provider)); |
||||
} |
||||
ot_builder.SetLabelsInjector(std::move(labels_injector)); |
||||
ot_builder.BuildAndRegisterGlobal(); |
||||
grpc_init(); |
||||
grpc::ServerBuilder builder; |
||||
int port; |
||||
// Use IPv4 here because it's less flaky than IPv6 ("[::]:0") on Travis.
|
||||
builder.AddListeningPort("0.0.0.0:0", grpc::InsecureServerCredentials(), |
||||
&port); |
||||
builder.RegisterService(&service_); |
||||
server_ = builder.BuildAndStart(); |
||||
ASSERT_NE(nullptr, server_); |
||||
ASSERT_NE(0, port); |
||||
server_address_ = absl::StrCat("localhost:", port); |
||||
canonical_server_address_ = absl::StrCat("dns:///", server_address_); |
||||
|
||||
stub_ = EchoTestService::NewStub( |
||||
grpc::CreateChannel(server_address_, grpc::InsecureChannelCredentials())); |
||||
} |
||||
|
||||
void OTelPluginEnd2EndTest::TearDown() { |
||||
server_->Shutdown(); |
||||
grpc_shutdown_blocking(); |
||||
delete grpc_core::ServerCallTracerFactory::Get(grpc_core::ChannelArgs()); |
||||
grpc_core::ServerCallTracerFactory::RegisterGlobal(nullptr); |
||||
} |
||||
|
||||
void OTelPluginEnd2EndTest::ResetStub(std::shared_ptr<Channel> channel) { |
||||
stub_ = EchoTestService::NewStub(std::move(channel)); |
||||
} |
||||
|
||||
void OTelPluginEnd2EndTest::SendRPC() { |
||||
EchoRequest request; |
||||
request.set_message("foo"); |
||||
EchoResponse response; |
||||
grpc::ClientContext context; |
||||
grpc::Status status = stub_->Echo(&context, request, &response); |
||||
} |
||||
|
||||
absl::flat_hash_map< |
||||
std::string, std::vector<opentelemetry::sdk::metrics::PointDataAttributes>> |
||||
OTelPluginEnd2EndTest::ReadCurrentMetricsData( |
||||
absl::AnyInvocable< |
||||
bool(const absl::flat_hash_map< |
||||
std::string, |
||||
std::vector<opentelemetry::sdk::metrics::PointDataAttributes>>&)> |
||||
continue_predicate) { |
||||
absl::flat_hash_map< |
||||
std::string, |
||||
std::vector<opentelemetry::sdk::metrics::PointDataAttributes>> |
||||
data; |
||||
auto deadline = absl::Now() + absl::Seconds(5); |
||||
do { |
||||
reader_->Collect([&](opentelemetry::sdk::metrics::ResourceMetrics& rm) { |
||||
for (const opentelemetry::sdk::metrics::ScopeMetrics& smd : |
||||
rm.scope_metric_data_) { |
||||
for (const opentelemetry::sdk::metrics::MetricData& md : |
||||
smd.metric_data_) { |
||||
for (const opentelemetry::sdk::metrics::PointDataAttributes& dp : |
||||
md.point_data_attr_) { |
||||
data[md.instrument_descriptor.name_].push_back(dp); |
||||
} |
||||
} |
||||
} |
||||
return true; |
||||
}); |
||||
} while (continue_predicate(data) && deadline > absl::Now()); |
||||
return data; |
||||
} |
||||
|
||||
} // namespace testing
|
||||
} // namespace grpc
|
||||
@ -0,0 +1,94 @@ |
||||
//
|
||||
//
|
||||
// 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_TEST_CPP_EXT_OTEL_OTEL_TEST_LIBRARY_H |
||||
#define GRPC_TEST_CPP_EXT_OTEL_OTEL_TEST_LIBRARY_H |
||||
|
||||
#include "absl/functional/any_invocable.h" |
||||
#include "api/include/opentelemetry/metrics/provider.h" |
||||
#include "gmock/gmock.h" |
||||
#include "gtest/gtest.h" |
||||
#include "opentelemetry/sdk/metrics/meter_provider.h" |
||||
#include "opentelemetry/sdk/metrics/metric_reader.h" |
||||
|
||||
#include <grpcpp/grpcpp.h> |
||||
|
||||
#include "src/core/lib/channel/call_tracer.h" |
||||
#include "src/core/lib/config/core_configuration.h" |
||||
#include "src/cpp/ext/otel/otel_plugin.h" |
||||
#include "test/core/util/test_config.h" |
||||
#include "test/cpp/end2end/test_service_impl.h" |
||||
|
||||
namespace grpc { |
||||
namespace testing { |
||||
|
||||
class MockMetricReader : public opentelemetry::sdk::metrics::MetricReader { |
||||
public: |
||||
opentelemetry::sdk::metrics::AggregationTemporality GetAggregationTemporality( |
||||
opentelemetry::sdk::metrics::InstrumentType) const noexcept override { |
||||
return opentelemetry::sdk::metrics::AggregationTemporality::kDelta; |
||||
} |
||||
|
||||
bool OnForceFlush(std::chrono::microseconds) noexcept override { |
||||
return true; |
||||
} |
||||
|
||||
bool OnShutDown(std::chrono::microseconds) noexcept override { return true; } |
||||
|
||||
void OnInitialized() noexcept override {} |
||||
}; |
||||
|
||||
class OTelPluginEnd2EndTest : public ::testing::Test { |
||||
protected: |
||||
// Note that we can't use SetUp() here since we want to send in parameters.
|
||||
void Init( |
||||
const absl::flat_hash_set<absl::string_view>& metric_names, |
||||
opentelemetry::sdk::resource::Resource resource = |
||||
opentelemetry::sdk::resource::Resource::Create({}), |
||||
std::unique_ptr<grpc::internal::LabelsInjector> labels_injector = nullptr, |
||||
bool test_no_meter_provider = false); |
||||
|
||||
void TearDown() override; |
||||
|
||||
void ResetStub(std::shared_ptr<Channel> channel); |
||||
|
||||
void SendRPC(); |
||||
|
||||
absl::flat_hash_map< |
||||
std::string, |
||||
std::vector<opentelemetry::sdk::metrics::PointDataAttributes>> |
||||
ReadCurrentMetricsData( |
||||
absl::AnyInvocable< |
||||
bool(const absl::flat_hash_map< |
||||
std::string, |
||||
std::vector<opentelemetry::sdk::metrics::PointDataAttributes>>&)> |
||||
continue_predicate); |
||||
|
||||
const absl::string_view kMethodName = "grpc.testing.EchoTestService/Echo"; |
||||
std::shared_ptr<opentelemetry::sdk::metrics::MetricReader> reader_; |
||||
std::string server_address_; |
||||
std::string canonical_server_address_; |
||||
CallbackTestServiceImpl service_; |
||||
std::unique_ptr<grpc::Server> server_; |
||||
std::unique_ptr<EchoTestService::Stub> stub_; |
||||
}; |
||||
|
||||
} // namespace testing
|
||||
} // namespace grpc
|
||||
|
||||
#endif // GRPC_TEST_CPP_EXT_OTEL_OTEL_TEST_LIBRARY_H
|
||||
Loading…
Reference in new issue