Merge remote-tracking branch 'upstream/master' into dualstack_ring_hash

.github/workflows/psm-interop.yaml

@@ -5,7 +5,6 @@ on:
   push:
     branches:
       - master
-      - 'v1.*'
 
 permissions:
   contents: read

src/compiler/config.h

@@ -40,7 +40,8 @@ namespace protobuf {
 namespace compiler {
 typedef GRPC_CUSTOM_CODEGENERATOR CodeGenerator;
 typedef GRPC_CUSTOM_GENERATORCONTEXT GeneratorContext;
-static inline int PluginMain(int argc, char* argv[], CodeGenerator* generator) {
+static inline int PluginMain(int argc, char* argv[],
+                             const CodeGenerator* generator) {
   return GRPC_CUSTOM_PLUGINMAIN(argc, argv, generator);
 }
 static inline void ParseGeneratorParameter(

src/core/ext/filters/client_channel/lb_policy/xds/xds_override_host.cc

@@ -18,6 +18,7 @@
 
 #include "src/core/ext/filters/client_channel/lb_policy/xds/xds_override_host.h"
 
+#include <inttypes.h>
 #include <stddef.h>
 
 #include <algorithm>
@@ -430,7 +431,10 @@ void XdsOverrideHostLb::ResetBackoffLocked() {
 
 absl::Status XdsOverrideHostLb::UpdateLocked(UpdateArgs args) {
   if (GRPC_TRACE_FLAG_ENABLED(grpc_lb_xds_override_host_trace)) {
-    gpr_log(GPR_INFO, "[xds_override_host_lb %p] Received update", this);
+    gpr_log(GPR_INFO,
+            "[xds_override_host_lb %p] Received update with %" PRIuPTR
+            " addresses",
+            this, args.addresses.ok() ? args.addresses->size() : 0);
   }
   auto old_config = std::move(config_);
   // Update config.
@@ -498,27 +502,52 @@ OrphanablePtr<LoadBalancingPolicy> XdsOverrideHostLb::CreateChildPolicyLocked(
 absl::StatusOr<ServerAddressList> XdsOverrideHostLb::UpdateAddressMap(
     absl::StatusOr<ServerAddressList> addresses) {
   if (!addresses.ok()) {
+    if (GRPC_TRACE_FLAG_ENABLED(grpc_lb_xds_override_host_trace)) {
+      gpr_log(GPR_INFO, "[xds_override_host_lb %p] address error: %s", this,
+              addresses.status().ToString().c_str());
+    }
     return addresses;
   }
   ServerAddressList return_value;
   std::map<const std::string, XdsHealthStatus> addresses_for_map;
   for (const auto& address : *addresses) {
     XdsHealthStatus status = GetAddressHealthStatus(address);
     if (status.status() != XdsHealthStatus::kDraining) {
+      if (GRPC_TRACE_FLAG_ENABLED(grpc_lb_xds_override_host_trace)) {
+        gpr_log(GPR_INFO,
+                "[xds_override_host_lb %p] address %s: not draining, "
+                "passing to child",
+                this, address.ToString().c_str());
+      }
       return_value.push_back(address);
     } else if (!config_->override_host_status_set().Contains(status)) {
       // Skip draining hosts if not in the override status set.
+      if (GRPC_TRACE_FLAG_ENABLED(grpc_lb_xds_override_host_trace)) {
+        gpr_log(GPR_INFO,
+                "[xds_override_host_lb %p] address %s: draining but not in "
+                "override_host_status set -- ignoring",
+                this, address.ToString().c_str());
+      }
       continue;
     }
     auto key = grpc_sockaddr_to_uri(&address.address());
     if (key.ok()) {
+      if (GRPC_TRACE_FLAG_ENABLED(grpc_lb_xds_override_host_trace)) {
+        gpr_log(GPR_INFO,
+                "[xds_override_host_lb %p] address %s: adding map key %s", this,
+                address.ToString().c_str(), key->c_str());
+      }
       addresses_for_map.emplace(std::move(*key), status);
     }
   }
   {
     MutexLock lock(&subchannel_map_mu_);
     for (auto it = subchannel_map_.begin(); it != subchannel_map_.end();) {
       if (addresses_for_map.find(it->first) == addresses_for_map.end()) {
+        if (GRPC_TRACE_FLAG_ENABLED(grpc_lb_xds_override_host_trace)) {
+          gpr_log(GPR_INFO, "[xds_override_host_lb %p] removing map key %s",
+                  this, it->first.c_str());
+        }
         it = subchannel_map_.erase(it);
       } else {
         ++it;
@@ -527,10 +556,20 @@ absl::StatusOr<ServerAddressList> XdsOverrideHostLb::UpdateAddressMap(
     for (const auto& key_status : addresses_for_map) {
       auto it = subchannel_map_.find(key_status.first);
       if (it == subchannel_map_.end()) {
+        if (GRPC_TRACE_FLAG_ENABLED(grpc_lb_xds_override_host_trace)) {
+          gpr_log(GPR_INFO, "[xds_override_host_lb %p] adding map key %s", this,
+                  key_status.first.c_str());
+        }
         subchannel_map_.emplace(std::piecewise_construct,
                                 std::forward_as_tuple(key_status.first),
                                 std::forward_as_tuple(key_status.second));
       } else {
+        if (GRPC_TRACE_FLAG_ENABLED(grpc_lb_xds_override_host_trace)) {
+          gpr_log(GPR_INFO,
+                  "[xds_override_host_lb %p] setting EDS health status for "
+                  "%s to %s",
+                  this, key_status.first.c_str(), key_status.second.ToString());
+        }
         it->second.SetEdsHealthStatus(key_status.second);
       }
     }

src/core/ext/transport/chttp2/transport/frame.cc

@@ -40,6 +40,7 @@ constexpr uint8_t kFrameTypeSettings = 4;
 constexpr uint8_t kFrameTypePing = 6;
 constexpr uint8_t kFrameTypeGoaway = 7;
 constexpr uint8_t kFrameTypeWindowUpdate = 8;
+constexpr uint8_t kFrameTypePushPromise = 5;
 
 constexpr uint8_t kFlagEndStream = 1;
 constexpr uint8_t kFlagAck = 1;
@@ -175,6 +176,7 @@ class SerializeHeaderAndPayload {
   }
 
   void operator()(Http2SettingsFrame& frame) {
+    // Six bytes per setting (u16 id, u32 value)
     const size_t payload_size = 6 * frame.settings.size();
     auto hdr_and_payload =
         extra_bytes_.TakeFirst(kFrameHeaderSize + payload_size);
@@ -463,7 +465,7 @@ void Serialize(absl::Span<Http2Frame> frames, SliceBuffer& out) {
   for (auto& frame : frames) {
     // Bytes needed for framing
     buffer_needed += kFrameHeaderSize;
-    // Bytes needed for unserialized payload
+    // Bytes needed for frame payload
     buffer_needed += absl::visit(SerializeExtraBytesRequired(), frame);
   }
   SerializeHeaderAndPayload serialize(buffer_needed, out);
@@ -492,6 +494,10 @@ absl::StatusOr<Http2Frame> ParseFramePayload(const Http2FrameHeader& hdr,
       return ParseGoawayFrame(hdr, payload);
     case kFrameTypeWindowUpdate:
       return ParseWindowUpdateFrame(hdr, payload);
+    case kFrameTypePushPromise:
+      return absl::InternalError(
+          "push promise not supported (and SETTINGS_ENABLE_PUSH explicitly "
+          "disabled).");
     default:
       return Http2UnknownFrame{};
   }

src/core/lib/transport/metadata_batch.h

@@ -305,6 +305,13 @@ struct GrpcTagsBinMetadata : public SimpleSliceBasedMetadata {
   static absl::string_view key() { return "grpc-tags-bin"; }
 };
 
+// XEnvoyPeerMetadata
+struct XEnvoyPeerMetadata : public SimpleSliceBasedMetadata {
+  static constexpr bool kRepeatable = false;
+  using CompressionTraits = StableValueCompressor;
+  static absl::string_view key() { return "x-envoy-peer-metadata"; }
+};
+
 // :authority metadata trait.
 struct HttpAuthorityMetadata : public SimpleSliceBasedMetadata {
   static constexpr bool kRepeatable = false;
@@ -1474,6 +1481,7 @@ using grpc_metadata_batch_base = grpc_core::MetadataMap<
     grpc_core::GrpcServerStatsBinMetadata, grpc_core::GrpcTraceBinMetadata,
     grpc_core::GrpcTagsBinMetadata, grpc_core::GrpcLbClientStatsMetadata,
     grpc_core::LbCostBinMetadata, grpc_core::LbTokenMetadata,
+    grpc_core::XEnvoyPeerMetadata,
     // Non-encodable things
     grpc_core::GrpcStreamNetworkState, grpc_core::PeerString,
     grpc_core::GrpcStatusContext, grpc_core::GrpcStatusFromWire,

src/cpp/ext/gsm/BUILD

@@ -32,21 +32,32 @@ grpc_cc_library(
     name = "gsm_observability",
     srcs = [
         "gsm_observability.cc",
+        "metadata_exchange.cc",
     ],
     hdrs = [
         "gsm_observability.h",
+        "metadata_exchange.h",
     ],
     external_deps = [
         "absl/container:flat_hash_set",
+        "absl/meta:type_traits",
         "absl/status",
         "absl/status:statusor",
         "absl/strings",
+        "absl/types:optional",
+        "absl/types:variant",
         "otel/sdk/src/metrics",
+        "otel/sdk:headers",
+        "upb_lib",
     ],
     language = "c++",
     visibility = ["//:__subpackages__"],
     deps = [
         "//:gpr_platform",
+        "//:grpc_base",
+        "//:protobuf_struct_upb",
+        "//src/core:env",
+        "//src/core:slice",
         "//src/cpp/ext/otel:otel_plugin",
     ],
 )