Merge branch 'main' into dns_refresh

Signed-off-by: Alyssa Wilk <alyssar@chromium.org>

changelogs/current.yaml

@@ -25,6 +25,11 @@ behavior_changes:
 
 minor_behavior_changes:
 # *Changes that may cause incompatibilities for some users, but should not for most*
+- area: grpc
+  change: |
+    Changes in ``AsyncStreamImpl`` now propagate tracing context headers in bidirectional streams when using
+    :ref:`Envoy gRPC client <envoy_v3_api_field_config.core.v3.GrpcService.envoy_grpc>`. Previously, tracing context headers
+    were not being set when calling external services such as ``ext_proc``.
 - area: tracers
   change: |
     Set status code for OpenTelemetry tracers (previously unset).
@@ -177,6 +182,9 @@ new_features:
   change: |
     added support for :ref:`%UPSTREAM_HOST_NAME% <config_access_log_format_upstream_host_name>` for the upstream host
     identifier.
+- area: access_loggers
+  change: |
+    Added ``TRACE_ID`` :ref:`access log formatter <config_access_log_format>`.
 - area: healthcheck
   change: |
     Added support to healthcheck with ProxyProtocol in TCP Healthcheck by setting

docs/root/configuration/observability/access_log/usage.rst

@@ -1216,3 +1216,9 @@ UDP
 %ENVIRONMENT(X):Z%
   Environment value of environment variable X. If no valid environment variable X, '-' symbol will be used.
   Z is an optional parameter denoting string truncation up to Z characters long.
+
+%TRACE_ID%
+  HTTP
+    The trace ID of the request. If the request does not have a trace ID, this will be an empty string.
+  TCP/UDP
+    Not implemented ("-").

envoy/grpc/async_client.h

@@ -175,7 +175,6 @@ class RawAsyncClient {
 
   /**
    * Start a gRPC stream asynchronously.
-   * TODO(mattklein123): Determine if tracing should be added to streaming requests.
    * @param service_full_name full name of the service (i.e. service_method.service()->full_name()).
    * @param method_name name of the method (i.e. service_method.name()).
    * @param callbacks the callbacks to be notified of stream status.

envoy/http/async_client.h

@@ -321,10 +321,25 @@ class AsyncClient {
       return *this;
     }
 
+    StreamOptions& setParentSpan(Tracing::Span& parent_span) {
+      parent_span_ = &parent_span;
+      return *this;
+    }
+    StreamOptions& setChildSpanName(const std::string& child_span_name) {
+      child_span_name_ = child_span_name;
+      return *this;
+    }
+    StreamOptions& setSampled(absl::optional<bool> sampled) {
+      sampled_ = sampled;
+      return *this;
+    }
+
     // For gmock test
     bool operator==(const StreamOptions& src) const {
       return timeout == src.timeout && buffer_body_for_retry == src.buffer_body_for_retry &&
-             send_xff == src.send_xff && send_internal == src.send_internal;
+             send_xff == src.send_xff && send_internal == src.send_internal &&
+             parent_span_ == src.parent_span_ && child_span_name_ == src.child_span_name_ &&
+             sampled_ == src.sampled_;
     }
 
     // The timeout supplies the stream timeout, measured since when the frame with
@@ -365,91 +380,6 @@ class AsyncClient {
     bool is_shadow{false};
 
     bool is_shadow_suffixed_disabled{false};
-  };
-
-  /**
-   * A structure to hold the options for AsyncRequest object.
-   */
-  struct RequestOptions : public StreamOptions {
-    RequestOptions& setTimeout(const absl::optional<std::chrono::milliseconds>& v) {
-      StreamOptions::setTimeout(v);
-      return *this;
-    }
-    RequestOptions& setTimeout(const std::chrono::milliseconds& v) {
-      StreamOptions::setTimeout(v);
-      return *this;
-    }
-    RequestOptions& setBufferBodyForRetry(bool v) {
-      StreamOptions::setBufferBodyForRetry(v);
-      return *this;
-    }
-    RequestOptions& setSendXff(bool v) {
-      StreamOptions::setSendXff(v);
-      return *this;
-    }
-    RequestOptions& setSendInternal(bool v) {
-      StreamOptions::setSendInternal(v);
-      return *this;
-    }
-    RequestOptions& setHashPolicy(
-        const Protobuf::RepeatedPtrField<envoy::config::route::v3::RouteAction::HashPolicy>& v) {
-      StreamOptions::setHashPolicy(v);
-      return *this;
-    }
-    RequestOptions& setParentContext(const ParentContext& v) {
-      StreamOptions::setParentContext(v);
-      return *this;
-    }
-    RequestOptions& setMetadata(const envoy::config::core::v3::Metadata& m) {
-      StreamOptions::setMetadata(m);
-      return *this;
-    }
-    RequestOptions& setFilterState(Envoy::StreamInfo::FilterStateSharedPtr fs) {
-      StreamOptions::setFilterState(fs);
-      return *this;
-    }
-    RequestOptions& setRetryPolicy(const envoy::config::route::v3::RetryPolicy& p) {
-      StreamOptions::setRetryPolicy(p);
-      return *this;
-    }
-    RequestOptions& setRetryPolicy(const Router::RetryPolicy& p) {
-      StreamOptions::setRetryPolicy(p);
-      return *this;
-    }
-    RequestOptions& setIsShadow(bool s) {
-      StreamOptions::setIsShadow(s);
-      return *this;
-    }
-    RequestOptions& setIsShadowSuffixDisabled(bool d) {
-      StreamOptions::setIsShadowSuffixDisabled(d);
-      return *this;
-    }
-    RequestOptions& setParentSpan(Tracing::Span& parent_span) {
-      parent_span_ = &parent_span;
-      return *this;
-    }
-    RequestOptions& setChildSpanName(const std::string& child_span_name) {
-      child_span_name_ = child_span_name;
-      return *this;
-    }
-    RequestOptions& setSampled(absl::optional<bool> sampled) {
-      sampled_ = sampled;
-      return *this;
-    }
-    RequestOptions& setBufferAccount(const Buffer::BufferMemoryAccountSharedPtr& account) {
-      account_ = account;
-      return *this;
-    }
-    RequestOptions& setBufferLimit(uint32_t limit) {
-      buffer_limit_ = limit;
-      return *this;
-    }
-
-    // For gmock test
-    bool operator==(const RequestOptions& src) const {
-      return StreamOptions::operator==(src) && parent_span_ == src.parent_span_ &&
-             child_span_name_ == src.child_span_name_ && sampled_ == src.sampled_;
-    }
 
     // The parent span that child spans are created under to trace egress requests/responses.
     // If not set, requests will not be traced.
@@ -462,6 +392,11 @@ class AsyncClient {
     absl::optional<bool> sampled_{true};
   };
 
+  /**
+   * A structure to hold the options for AsyncRequest object.
+   */
+  using RequestOptions = StreamOptions;
+
   /**
    * Send an HTTP request asynchronously
    * @param request the request to send.

mobile/library/common/engine_types.h

@@ -64,6 +64,7 @@ struct EnvoyStreamCallbacks {
    *
    * The callback function pases the following parameters.
    * - buffer: the data received.
+   * - length: the length of data to read. It will always be <= `buffer.length()`
    * - end_stream: whether the data is the last data frame.
    * - stream_intel: contains internal stream metrics.
    */

mobile/library/common/internal_engine.cc

@@ -11,6 +11,9 @@
 #include "library/common/stats/utility.h"
 
 namespace Envoy {
+namespace {
+constexpr absl::Duration ENGINE_RUNNING_TIMEOUT = absl::Seconds(3);
+} // namespace
 
 static std::atomic<envoy_stream_t> current_stream_handle_{0};
 
@@ -173,6 +176,7 @@ envoy_status_t InternalEngine::main(std::shared_ptr<Envoy::OptionsImplBase> opti
                                                         server_->serverFactoryContext().scope(),
                                                         server_->api().randomGenerator());
           dispatcher_->drain(server_->dispatcher());
+          engine_running_.Notify();
           callbacks_->on_engine_running_();
         });
   } // mutex_
@@ -212,6 +216,10 @@ envoy_status_t InternalEngine::terminate() {
     return ENVOY_FAILURE;
   }
 
+  // Wait until the Engine is ready before calling terminate to avoid assertion failures.
+  // TODO(fredyw): Fix this without having to wait.
+  ASSERT(engine_running_.WaitForNotificationWithTimeout(ENGINE_RUNNING_TIMEOUT));
+
   // We need to be sure that MainCommon is finished being constructed so we can dispatch shutdown.
   {
     Thread::LockGuard lock(mutex_);

mobile/library/common/internal_engine.h

@@ -7,6 +7,7 @@
 #include "source/common/common/posix/thread_impl.h"
 #include "source/common/common/thread.h"
 
+#include "absl/synchronization/notification.h"
 #include "absl/types/optional.h"
 #include "extension_registry.h"
 #include "library/common/engine_common.h"
@@ -163,6 +164,7 @@ class InternalEngine : public Logger::Loggable<Logger::Id::main> {
   // instructions scheduled on the main_thread_ need to have a longer lifetime.
   Thread::PosixThreadPtr main_thread_{nullptr}; // Empty placeholder to be populated later.
   bool terminated_{false};
+  absl::Notification engine_running_;
 };
 
 } // namespace Envoy

mobile/test/cc/engine_test.cc

@@ -141,4 +141,56 @@ TEST(EngineTest, SetEventTracker) {
   EXPECT_TRUE(on_track.WaitForNotificationWithTimeout(absl::Seconds(3)));
 }
 
+TEST(EngineTest, DontWaitForOnEngineRunning) {
+  Platform::EngineBuilder engine_builder;
+  engine_builder.setLogLevel(Logger::Logger::debug).enforceTrustChainVerification(false);
+  EngineWithTestServer engine_with_test_server(engine_builder, TestServerType::HTTP2_WITH_TLS);
+
+  std::string actual_status_code;
+  bool actual_end_stream = false;
+  absl::Notification stream_complete;
+  EnvoyStreamCallbacks stream_callbacks;
+  stream_callbacks.on_headers_ = [&](const Http::ResponseHeaderMap& headers, bool end_stream,
+                                     envoy_stream_intel) {
+    actual_status_code = headers.getStatusValue();
+    actual_end_stream = end_stream;
+  };
+  stream_callbacks.on_data_ = [&](const Buffer::Instance&, uint64_t /* length */, bool end_stream,
+                                  envoy_stream_intel) { actual_end_stream = end_stream; };
+  stream_callbacks.on_complete_ = [&](envoy_stream_intel, envoy_final_stream_intel) {
+    stream_complete.Notify();
+  };
+  stream_callbacks.on_error_ = [&](EnvoyError, envoy_stream_intel, envoy_final_stream_intel) {
+    stream_complete.Notify();
+  };
+  stream_callbacks.on_cancel_ = [&](envoy_stream_intel, envoy_final_stream_intel) {
+    stream_complete.Notify();
+  };
+  auto stream_prototype = engine_with_test_server.engine()->streamClient()->newStreamPrototype();
+  Platform::StreamSharedPtr stream = stream_prototype->start(std::move(stream_callbacks));
+
+  auto headers = Http::Utility::createRequestHeaderMapPtr();
+  headers->addCopy(Http::LowerCaseString(":method"), "GET");
+  headers->addCopy(Http::LowerCaseString(":scheme"), "https");
+  headers->addCopy(Http::LowerCaseString(":authority"),
+                   engine_with_test_server.testServer().getAddress());
+  headers->addCopy(Http::LowerCaseString(":path"), "/");
+  stream->sendHeaders(std::move(headers), true);
+  stream_complete.WaitForNotification();
+
+  EXPECT_EQ(actual_status_code, "200");
+  EXPECT_TRUE(actual_end_stream);
+}
+
+TEST(EngineTest, TerminateWithoutWaitingForOnEngineRunning) {
+  absl::Notification engine_running;
+  auto engine_callbacks = std::make_unique<EngineCallbacks>();
+  engine_callbacks->on_engine_running_ = [&] { engine_running.Notify(); };
+
+  Platform::EngineBuilder engine_builder;
+  auto engine = engine_builder.setLogLevel(Logger::Logger::debug).build();
+
+  engine->terminate();
+}
+
 } // namespace Envoy

mobile/test/common/integration/xds_integration_test.cc

@@ -3,6 +3,8 @@
 #include "envoy/config/bootstrap/v3/bootstrap.pb.h"
 #include "envoy/config/cluster/v3/cluster.pb.h"
 
+#include "source/common/tls/server_context_impl.h"
+
 #include "test/common/grpc/grpc_client_integration.h"
 #include "test/common/integration/base_client_integration_test.h"
 #include "test/test_common/environment.h"
@@ -26,6 +28,8 @@ void XdsIntegrationTest::initialize() {
 
   // Register the extensions required for Envoy Mobile.
   ExtensionRegistry::registerFactories();
+  // For server TLS.
+  Extensions::TransportSockets::Tls::forceRegisterServerContextFactoryImpl();
 
   if (sotw_or_delta_ == Grpc::SotwOrDelta::UnifiedSotw ||
       sotw_or_delta_ == Grpc::SotwOrDelta::UnifiedDelta) {

source/common/common/BUILD

@@ -473,6 +473,18 @@ envoy_cc_library(
     ],
 )
 
+envoy_cc_library(
+    name = "compiled_string_map_lib",
+    hdrs = ["compiled_string_map.h"],
+    external_deps = [
+        "abseil_strings",
+    ],
+    deps = [
+        "//envoy/common:pure_lib",
+        "//source/common/common:assert_lib",
+    ],
+)
+
 envoy_cc_library(
     name = "packed_struct_lib",
     hdrs = ["packed_struct.h"],