Merge remote-tracking branch 'upstream/master' into privacy-manifests

.github/workflows/pr-auto-fix.yaml

@@ -48,8 +48,9 @@ jobs:
         with:
           script: |
             // If you'd like not to run this code on your commits, add your github user id here:
-            NO_AUTOFIX_USERS = []
+            NO_AUTOFIX_USERS = ["copybara-service[bot]"]
             const { owner, repo } = context.repo
+            console.log("Actor: " + context.actor);
             if (NO_AUTOFIX_USERS.includes(context.actor)) {
               console.log('Cancelling');
               const run_id = "${{ github.run_id }}";

bazel/grpc_deps.bzl

@@ -512,11 +512,11 @@ def grpc_deps():
     if "io_opentelemetry_cpp" not in native.existing_rules():
         http_archive(
             name = "io_opentelemetry_cpp",
-            sha256 = "f30cd88bf898a5726d245eba882b8e81012021eb00df34109f4dfb203f005cea",
-            strip_prefix = "opentelemetry-cpp-1.11.0",
+            sha256 = "149f076cc7a79bbd3a3c34fb3ab61d3a3e8dcfe2b9596f79153e17123c32f897",
+            strip_prefix = "opentelemetry-cpp-064fef0d871c57ffac6739d3311659a5770a9db4",
             urls = [
-                "https://storage.googleapis.com/grpc-bazel-mirror/github.com/open-telemetry/opentelemetry-cpp/archive/refs/tags/v1.11.0.tar.gz",
-                "https://github.com/open-telemetry/opentelemetry-cpp/archive/refs/tags/v1.11.0.tar.gz",
+                "https://storage.googleapis.com/grpc-bazel-mirror/github.com/open-telemetry/opentelemetry-cpp/archive/064fef0d871c57ffac6739d3311659a5770a9db4.tar.gz",
+                "https://github.com/open-telemetry/opentelemetry-cpp/archive/064fef0d871c57ffac6739d3311659a5770a9db4.tar.gz",
             ],
         )
 

include/grpc/event_engine/event_engine.h

@@ -255,6 +255,45 @@ class EventEngine : public std::enable_shared_from_this<EventEngine> {
     /// values are expected to remain valid for the life of the Endpoint.
     virtual const ResolvedAddress& GetPeerAddress() const = 0;
     virtual const ResolvedAddress& GetLocalAddress() const = 0;
+
+    /// A method which allows users to query whether an Endpoint implementation
+    /// supports a specified extension. The name of the extension is provided
+    /// as an input.
+    ///
+    /// An extension could be any type with a unique string id. Each extension
+    /// may support additional capabilities and if the Endpoint implementation
+    /// supports the queried extension, it should return a valid pointer to the
+    /// extension type.
+    ///
+    /// E.g., use case of an EventEngine::Endpoint supporting a custom
+    /// extension.
+    ///
+    /// class CustomEndpointExtension {
+    ///  public:
+    ///    static constexpr std::string name = "my.namespace.extension_name";
+    ///    void Process() { ... }
+    /// }
+    ///
+    ///
+    /// class CustomEndpoint :
+    ///        public EventEngine::Endpoint, CustomEndpointExtension {
+    ///   public:
+    ///     void* QueryExtension(absl::string_view id) override {
+    ///       if (id == CustomEndpointExtension::name) {
+    ///         return static_cast<CustomEndpointExtension*>(this);
+    ///       }
+    ///       return nullptr;
+    ///     }
+    ///     ...
+    /// }
+    ///
+    /// auto ext_ =
+    /// static_cast<CustomEndpointExtension*>(
+    ///   endpoint->QueryExtension(CustomrEndpointExtension::name));
+    /// if (ext_ != nullptr) { ext_->Process(); }
+    ///
+    ///
+    virtual void* QueryExtension(absl::string_view /*id*/) { return nullptr; }
   };
 
   /// Called when a new connection is established.

include/grpc/impl/slice_type.h

@@ -74,7 +74,7 @@ struct grpc_slice {
   } data;
 };
 
-#define GRPC_SLICE_BUFFER_INLINE_ELEMENTS 8
+#define GRPC_SLICE_BUFFER_INLINE_ELEMENTS 7
 
 /** Represents an expandable array of slices, to be interpreted as a
    single item. */

src/core/BUILD

@@ -1540,6 +1540,18 @@ grpc_cc_library(
     ],
 )
 
+grpc_cc_library(
+    name = "event_engine_query_extensions",
+    hdrs = [
+        "lib/event_engine/query_extensions.h",
+    ],
+    external_deps = ["absl/strings"],
+    deps = [
+        "//:event_engine_base_hdrs",
+        "//:gpr_platform",
+    ],
+)
+
 grpc_cc_library(
     name = "event_engine_work_queue",
     hdrs = [

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

@@ -199,21 +199,18 @@ class CdsLb : public LoadBalancingPolicy {
   // The root of the tree is config_->cluster().
   std::map<std::string, WatcherState> watchers_;
 
-  // TODO(roth, yashkt): These are here because we need to handle
-  // pollset_set linkage as clusters are added or removed from the
-  // XdsCertificateProvider.  However, in the aggregate cluster case,
-  // there may be multiple clusters in the same cert provider, and we're
-  // only tracking the cert providers for the most recent underlying
-  // cluster here.  I think this is a bug that could cause us to starve
-  // the underlying cert providers of polling.  However, it is not
-  // actually causing any problem in practice today, because (a) we have
-  // no cert provider impl that relies on gRPC's polling and (b)
-  // probably no one is actually configuring an aggregate cluster with
-  // different cert providers in different underlying clusters.
-  // Hopefully, this problem won't be an issue in practice until after
-  // the EventEngine migration is done, at which point the need for
-  // handling pollset_set linkage will go away, and these fields can
-  // simply be removed.
+  // TODO(roth, yashkt): These are here because XdsCertificateProvider
+  // does not store the actual underlying cert providers, it stores only
+  // their distributors, so we need to hold a ref to the cert providers
+  // here.  However, in the aggregate cluster case, there may be multiple
+  // clusters in the same cert provider, and we're only tracking the cert
+  // providers for the most recent underlying cluster here.  This is
+  // clearly a bug, and I think it will cause us to stop getting updates
+  // for all but one of the cert providers in the aggregate cluster
+  // case.  Need to figure out the right way to fix this -- I don't
+  // think we want to store another map here, so ideally, we should just
+  // have XdsCertificateProvider actually hold the refs to the cert
+  // providers instead of just the distributors.
   RefCountedPtr<grpc_tls_certificate_provider> root_certificate_provider_;
   RefCountedPtr<grpc_tls_certificate_provider> identity_certificate_provider_;
 
@@ -610,20 +607,7 @@ absl::Status CdsLb::UpdateXdsCertificateProvider(
                        root_provider_instance_name, "\" not recognized."));
     }
   }
-  if (root_certificate_provider_ != new_root_provider) {
-    if (root_certificate_provider_ != nullptr &&
-        root_certificate_provider_->interested_parties() != nullptr) {
-      grpc_pollset_set_del_pollset_set(
-          interested_parties(),
-          root_certificate_provider_->interested_parties());
-    }
-    if (new_root_provider != nullptr &&
-        new_root_provider->interested_parties() != nullptr) {
-      grpc_pollset_set_add_pollset_set(interested_parties(),
-                                       new_root_provider->interested_parties());
-    }
-    root_certificate_provider_ = std::move(new_root_provider);
-  }
+  root_certificate_provider_ = std::move(new_root_provider);
   xds_certificate_provider_->UpdateRootCertNameAndDistributor(
       cluster_name, root_provider_cert_name,
       root_certificate_provider_ == nullptr
@@ -647,20 +631,7 @@ absl::Status CdsLb::UpdateXdsCertificateProvider(
                        identity_provider_instance_name, "\" not recognized."));
     }
   }
-  if (identity_certificate_provider_ != new_identity_provider) {
-    if (identity_certificate_provider_ != nullptr &&
-        identity_certificate_provider_->interested_parties() != nullptr) {
-      grpc_pollset_set_del_pollset_set(
-          interested_parties(),
-          identity_certificate_provider_->interested_parties());
-    }
-    if (new_identity_provider != nullptr &&
-        new_identity_provider->interested_parties() != nullptr) {
-      grpc_pollset_set_add_pollset_set(
-          interested_parties(), new_identity_provider->interested_parties());
-    }
-    identity_certificate_provider_ = std::move(new_identity_provider);
-  }
+  identity_certificate_provider_ = std::move(new_identity_provider);
   xds_certificate_provider_->UpdateIdentityCertNameAndDistributor(
       cluster_name, identity_provider_cert_name,
       identity_certificate_provider_ == nullptr

src/core/ext/filters/http/client/http_client_filter.cc

@@ -51,6 +51,9 @@
 
 namespace grpc_core {
 
+const NoInterceptor HttpClientFilter::Call::OnServerToClientMessage;
+const NoInterceptor HttpClientFilter::Call::OnClientToServerMessage;
+
 const grpc_channel_filter HttpClientFilter::kFilter =
     MakePromiseBasedFilter<HttpClientFilter, FilterEndpoint::kClient,
                            kFilterExaminesServerInitialMetadata>("http-client");
@@ -105,40 +108,27 @@ Slice UserAgentFromArgs(const ChannelArgs& args,
 }
 }  // namespace
 
-ArenaPromise<ServerMetadataHandle> HttpClientFilter::MakeCallPromise(
-    CallArgs call_args, NextPromiseFactory next_promise_factory) {
-  auto& md = call_args.client_initial_metadata;
-  if (test_only_use_put_requests_) {
-    md->Set(HttpMethodMetadata(), HttpMethodMetadata::kPut);
+void HttpClientFilter::Call::OnClientInitialMetadata(ClientMetadata& md,
+                                                     HttpClientFilter* filter) {
+  if (filter->test_only_use_put_requests_) {
+    md.Set(HttpMethodMetadata(), HttpMethodMetadata::kPut);
   } else {
-    md->Set(HttpMethodMetadata(), HttpMethodMetadata::kPost);
+    md.Set(HttpMethodMetadata(), HttpMethodMetadata::kPost);
   }
-  md->Set(HttpSchemeMetadata(), scheme_);
-  md->Set(TeMetadata(), TeMetadata::kTrailers);
-  md->Set(ContentTypeMetadata(), ContentTypeMetadata::kApplicationGrpc);
-  md->Set(UserAgentMetadata(), user_agent_.Ref());
-
-  auto* initial_metadata_err =
-      GetContext<Arena>()->New<Latch<ServerMetadataHandle>>();
-
-  call_args.server_initial_metadata->InterceptAndMap(
-      [initial_metadata_err](
-          ServerMetadataHandle md) -> absl::optional<ServerMetadataHandle> {
-        auto r = CheckServerMetadata(md.get());
-        if (!r.ok()) {
-          initial_metadata_err->Set(ServerMetadataFromStatus(r));
-          return absl::nullopt;
-        }
-        return std::move(md);
-      });
-
-  return Race(initial_metadata_err->Wait(),
-              Map(next_promise_factory(std::move(call_args)),
-                  [](ServerMetadataHandle md) -> ServerMetadataHandle {
-                    auto r = CheckServerMetadata(md.get());
-                    if (!r.ok()) return ServerMetadataFromStatus(r);
-                    return md;
-                  }));
+  md.Set(HttpSchemeMetadata(), filter->scheme_);
+  md.Set(TeMetadata(), TeMetadata::kTrailers);
+  md.Set(ContentTypeMetadata(), ContentTypeMetadata::kApplicationGrpc);
+  md.Set(UserAgentMetadata(), filter->user_agent_.Ref());
+}
+
+absl::Status HttpClientFilter::Call::OnServerInitialMetadata(
+    ServerMetadata& md) {
+  return CheckServerMetadata(&md);
+}
+
+absl::Status HttpClientFilter::Call::OnServerTrailingMetadata(
+    ServerMetadata& md) {
+  return CheckServerMetadata(&md);
 }
 
 HttpClientFilter::HttpClientFilter(HttpSchemeMetadata::ValueType scheme,

src/core/ext/filters/http/client/http_client_filter.h

@@ -25,23 +25,27 @@
 #include "src/core/lib/channel/channel_args.h"
 #include "src/core/lib/channel/channel_fwd.h"
 #include "src/core/lib/channel/promise_based_filter.h"
-#include "src/core/lib/promise/arena_promise.h"
 #include "src/core/lib/slice/slice.h"
 #include "src/core/lib/transport/metadata_batch.h"
 #include "src/core/lib/transport/transport.h"
 
 namespace grpc_core {
 
-class HttpClientFilter : public ChannelFilter {
+class HttpClientFilter : public ImplementChannelFilter<HttpClientFilter> {
  public:
   static const grpc_channel_filter kFilter;
 
   static absl::StatusOr<HttpClientFilter> Create(
       const ChannelArgs& args, ChannelFilter::Args filter_args);
 
-  // Construct a promise for one call.
-  ArenaPromise<ServerMetadataHandle> MakeCallPromise(
-      CallArgs call_args, NextPromiseFactory next_promise_factory) override;
+  class Call {
+   public:
+    void OnClientInitialMetadata(ClientMetadata& md, HttpClientFilter* filter);
+    absl::Status OnServerInitialMetadata(ServerMetadata& md);
+    absl::Status OnServerTrailingMetadata(ServerMetadata& md);
+    static const NoInterceptor OnClientToServerMessage;
+    static const NoInterceptor OnServerToClientMessage;
+  };
 
  private:
   HttpClientFilter(HttpSchemeMetadata::ValueType scheme, Slice user_agent,