test: Add test socket interface that allows overriding IoHandle behavior

include/envoy/network/socket.h

@@ -246,7 +246,7 @@ class SocketInterface {
    * @return @ref Network::IoHandlePtr that wraps the underlying socket file descriptor
    */
   virtual IoHandlePtr socket(Socket::Type type, Address::Type addr_type, Address::IpVersion version,
-                             bool socket_v6only) const PURE;
+                             bool socket_v6only) PURE;
 
   /**
    * Low level api to create a socket in the underlying host stack. Does not create an
@@ -256,7 +256,7 @@ class SocketInterface {
    * @return @ref Network::IoHandlePtr that wraps the underlying socket file descriptor
    */
   virtual IoHandlePtr socket(Socket::Type socket_type,
-                             const Address::InstanceConstSharedPtr addr) const PURE;
+                             const Address::InstanceConstSharedPtr addr) PURE;
 
   /**
    * Wrap socket file descriptor in IoHandle

source/common/network/socket_interface.h

@@ -43,7 +43,7 @@ class SocketInterfaceBase : public SocketInterface,
  * @param name Name of the socket interface to be looked up
  * @return Pointer to @ref SocketInterface instance that registered using the name of nullptr
  */
-static inline const SocketInterface* socketInterface(std::string name) {
+static inline SocketInterface* socketInterface(std::string name) {
   auto factory =
       Registry::FactoryRegistry<Server::Configuration::BootstrapExtensionFactory>::getFactory(name);
   return dynamic_cast<SocketInterface*>(factory);

source/common/network/socket_interface_impl.cc

@@ -13,7 +13,7 @@ namespace Envoy {
 namespace Network {
 
 IoHandlePtr SocketInterfaceImpl::socket(Socket::Type socket_type, Address::Type addr_type,
-                                        Address::IpVersion version, bool socket_v6only) const {
+                                        Address::IpVersion version, bool socket_v6only) {
 #if defined(__APPLE__) || defined(WIN32)
   int flags = 0;
 #else
@@ -54,7 +54,7 @@ IoHandlePtr SocketInterfaceImpl::socket(Socket::Type socket_type, Address::Type
 }
 
 IoHandlePtr SocketInterfaceImpl::socket(Socket::Type socket_type,
-                                        const Address::InstanceConstSharedPtr addr) const {
+                                        const Address::InstanceConstSharedPtr addr) {
   Address::IpVersion ip_version = addr->ip() ? addr->ip()->version() : Address::IpVersion::v4;
   int v6only = 0;
   if (addr->type() == Address::Type::Ip && ip_version == Address::IpVersion::v6) {

source/common/network/socket_interface_impl.h

@@ -12,9 +12,8 @@ class SocketInterfaceImpl : public SocketInterfaceBase {
 public:
   // SocketInterface
   IoHandlePtr socket(Socket::Type socket_type, Address::Type addr_type, Address::IpVersion version,
-                     bool socket_v6only) const override;
-  IoHandlePtr socket(Socket::Type socket_type,
-                     const Address::InstanceConstSharedPtr addr) const override;
+                     bool socket_v6only) override;
+  IoHandlePtr socket(Socket::Type socket_type, const Address::InstanceConstSharedPtr addr) override;
   IoHandlePtr socket(os_fd_t fd) override;
   bool ipFamilySupported(int domain) override;
 

test/common/http/http2/http2_frame.cc

@@ -111,6 +111,14 @@ void Http2Frame::appendHeaderWithoutIndexing(StaticHeaderIndex index, absl::stri
   appendData(value);
 }
 
+void Http2Frame::appendHeaderWithoutIndexing(const Header& header) {
+  data_.push_back(0);
+  appendHpackInt(header.key_.size(), 0x7f);
+  appendData(header.key_);
+  appendHpackInt(header.value_.size(), 0x7f);
+  appendData(header.value_);
+}
+
 void Http2Frame::appendEmptyHeader() {
   data_.push_back(0x40);
   data_.push_back(0x00);
@@ -250,6 +258,17 @@ Http2Frame Http2Frame::makeRequest(uint32_t stream_index, absl::string_view host
   return frame;
 }
 
+Http2Frame Http2Frame::makeRequest(uint32_t stream_index, absl::string_view host,
+                                   absl::string_view path,
+                                   const std::vector<Header> extra_headers) {
+  auto frame = makeRequest(stream_index, host, path);
+  for (const auto& header : extra_headers) {
+    frame.appendHeaderWithoutIndexing(header);
+  }
+  frame.adjustPayloadSize();
+  return frame;
+}
+
 Http2Frame Http2Frame::makePostRequest(uint32_t stream_index, absl::string_view host,
                                        absl::string_view path) {
   Http2Frame frame;

test/common/http/http2/http2_frame.h

@@ -85,6 +85,12 @@ class Http2Frame {
 
   enum class ResponseStatus { Unknown, Ok, NotFound };
 
+  struct Header {
+    Header(absl::string_view key, absl::string_view value) : key_(key), value_(value) {}
+    std::string key_;
+    std::string value_;
+  };
+
   // Methods for creating HTTP2 frames
   static Http2Frame makePingFrame(absl::string_view data = {});
   static Http2Frame makeEmptySettingsFrame(SettingsFlags flags = SettingsFlags::None);
@@ -107,6 +113,8 @@ class Http2Frame {
                                                           absl::string_view path);
   static Http2Frame makeRequest(uint32_t stream_index, absl::string_view host,
                                 absl::string_view path);
+  static Http2Frame makeRequest(uint32_t stream_index, absl::string_view host,
+                                absl::string_view path, const std::vector<Header> extra_headers);
   static Http2Frame makePostRequest(uint32_t stream_index, absl::string_view host,
                                     absl::string_view path);
   /**
@@ -163,6 +171,7 @@ class Http2Frame {
   // Headers are directly encoded
   void appendStaticHeader(StaticHeaderIndex index);
   void appendHeaderWithoutIndexing(StaticHeaderIndex index, absl::string_view value);
+  void appendHeaderWithoutIndexing(const Header& header);
   void appendEmptyHeader();
 
   // This method updates payload length in the HTTP2 header based on the size of the data_

test/integration/BUILD

@@ -345,6 +345,7 @@ envoy_cc_test(
         "//test/integration/filters:request_metadata_filter_config_lib",
         "//test/integration/filters:response_metadata_filter_config_lib",
         "//test/integration/filters:stop_iteration_and_continue",
+        "//test/integration/filters:test_socket_interface_lib",
         "//test/mocks/http:http_mocks",
         "//test/mocks/upstream:upstream_mocks",
         "//test/test_common:utility_lib",

test/integration/autonomous_upstream.cc

@@ -18,6 +18,7 @@ void HeaderToInt(const char header_name[], int32_t& return_int,
 } // namespace
 
 const char AutonomousStream::RESPONSE_SIZE_BYTES[] = "response_size_bytes";
+const char AutonomousStream::RESPONSE_DATA_BLOCKS[] = "response_data_blocks";
 const char AutonomousStream::EXPECT_REQUEST_SIZE_BYTES[] = "expect_request_size_bytes";
 const char AutonomousStream::RESET_AFTER_REQUEST[] = "reset_after_request";
 
@@ -59,8 +60,13 @@ void AutonomousStream::sendResponse() {
   int32_t response_body_length = 10;
   HeaderToInt(RESPONSE_SIZE_BYTES, response_body_length, headers);
 
+  int32_t response_data_blocks = 1;
+  HeaderToInt(RESPONSE_DATA_BLOCKS, response_data_blocks, headers);
+
   encodeHeaders(upstream_.responseHeaders(), false);
-  encodeData(response_body_length, false);
+  for (int32_t i = 0; i < response_data_blocks; ++i) {
+    encodeData(response_body_length, false);
+  }
   encodeTrailers(upstream_.responseTrailers());
 }
 

test/integration/autonomous_upstream.h

@@ -13,6 +13,8 @@ class AutonomousStream : public FakeStream {
 public:
   // The number of response bytes to send. Payload is randomized.
   static const char RESPONSE_SIZE_BYTES[];
+  // The number of data blocks send.
+  static const char RESPONSE_DATA_BLOCKS[];
   // If set to an integer, the AutonomousStream will expect the response body to
   // be this large.
   static const char EXPECT_REQUEST_SIZE_BYTES[];

test/integration/filters/BUILD

@@ -339,3 +339,23 @@ envoy_cc_test_library(
         "//test/extensions/filters/http/common:empty_http_filter_config_lib",
     ],
 )
+
+envoy_cc_test_library(
+    name = "test_socket_interface_lib",
+    srcs = [
+        "test_socket_interface.cc",
+    ],
+    hdrs = [
+        "test_socket_interface.h",
+    ],
+    deps = [
+        "//include/envoy/network:socket_interface",
+        "//source/common/common:assert_lib",
+        "//source/common/common:utility_lib",
+        "//source/common/network:address_lib",
+        "//source/common/network:socket_lib",
+        "@com_google_absl//absl/synchronization",
+        "@com_google_absl//absl/types:optional",
+        "@envoy_api//envoy/extensions/network/socket_interface/v3:pkg_cc_proto",
+    ],
+)

test/integration/filters/test_socket_interface.cc

@@ -0,0 +1,54 @@
+#include "test/integration/filters/test_socket_interface.h"
+
+#include <algorithm>
+
+#include "envoy/common/exception.h"
+#include "envoy/extensions/network/socket_interface/v3/default_socket_interface.pb.h"
+#include "envoy/network/socket.h"
+
+#include "common/api/os_sys_calls_impl.h"
+#include "common/common/utility.h"
+#include "common/network/address_impl.h"
+
+namespace Envoy {
+namespace Network {
+
+Api::IoCallUint64Result TestIoSocketHandle::writev(const Buffer::RawSlice* slices,
+                                                   uint64_t num_slice) {
+  if (writev_override_) {
+    auto result = writev_override_(index_, slices, num_slice);
+    if (result.has_value()) {
+      return std::move(result).value();
+    }
+  }
+  return IoSocketHandleImpl::writev(slices, num_slice);
+}
+
+TestSocketInterface::TestSocketInterface()
+    : previous_socket_interface_(SocketInterfaceSingleton::getExisting()) {
+  SocketInterfaceSingleton::clear();
+  SocketInterfaceSingleton::initialize(this);
+}
+
+TestSocketInterface::~TestSocketInterface() {
+  SocketInterfaceSingleton::clear();
+  if (previous_socket_interface_) {
+    SocketInterfaceSingleton::initialize(previous_socket_interface_);
+  }
+}
+
+void TestSocketInterface::overrideWritev(TestIoSocketHandle::WritevOverrideProc writev) {
+  writev_overide_proc_ = writev;
+}
+
+IoHandlePtr TestSocketInterface::socket(os_fd_t fd) {
+  return std::make_unique<TestIoSocketHandle>(getAcceptedSocketIndex(), writev_overide_proc_, fd);
+}
+
+ProtobufTypes::MessagePtr TestSocketInterface::createEmptyConfigProto() {
+  return std::make_unique<
+      envoy::extensions::network::socket_interface::v3::DefaultSocketInterface>();
+}
+
+} // namespace Network
+} // namespace Envoy

test/integration/filters/test_socket_interface.h

@@ -0,0 +1,85 @@
+#pragma once
+
+#include <functional>
+
+#include "envoy/network/address.h"
+#include "envoy/network/socket.h"
+
+#include "common/network/io_socket_handle_impl.h"
+#include "common/network/socket_interface_impl.h"
+
+#include "absl/synchronization/mutex.h"
+#include "absl/types/optional.h"
+
+/**
+ * TestSocketInterface allows overriding the behavior of the IoHandle interface.
+ */
+namespace Envoy {
+namespace Network {
+
+class TestIoSocketHandle : public IoSocketHandleImpl {
+public:
+  using WritevOverrideType = absl::optional<Api::IoCallUint64Result>(uint32_t index,
+                                                                     const Buffer::RawSlice* slices,
+                                                                     uint64_t num_slice);
+  using WritevOverrideProc = std::function<WritevOverrideType>;
+
+  TestIoSocketHandle(uint32_t index, WritevOverrideProc writev_override_proc,
+                     os_fd_t fd = INVALID_SOCKET, bool socket_v6only = false)
+      : IoSocketHandleImpl(fd, socket_v6only), index_(index),
+        writev_override_(writev_override_proc) {}
+
+private:
+  Api::IoCallUint64Result writev(const Buffer::RawSlice* slices, uint64_t num_slice) override;
+
+  const uint32_t index_;
+  const WritevOverrideProc writev_override_;
+};
+
+/**
+ * TestSocketInterface allows overriding of the behavior of the IoHandle interface of
+ * accepted sockets.
+ * Most integration tests have deterministic order in which Envoy accepts connections.
+ * For example a test with one client connection will result in two accepted sockets. First
+ * is for the client<->Envoy connection and the second is for the Envoy<->upstream connection.
+ */
+
+class TestSocketInterface : public SocketInterfaceImpl {
+public:
+  TestSocketInterface();
+  ~TestSocketInterface() override;
+
+  /**
+   * Override the behavior of the IoSocketHandleImpl::writev() method.
+   * The supplied callback is invoked with the arguments of the writev method.
+   * Returning absl::nullopt from the callback continues normal execution of the
+   * IoSocketHandleImpl::writev() method. Returning a Api::IoCallUint64Result from callback skips
+   * the IoSocketHandleImpl::writev() with the returned result value.
+   * NOTE: this method must be called before any sockets are created to avoid race condition in
+   * setting the `writev_overide_proc_` member.
+   */
+  void overrideWritev(TestIoSocketHandle::WritevOverrideProc writev);
+
+private:
+  // SocketInterface
+  using SocketInterfaceImpl::socket;
+  IoHandlePtr socket(os_fd_t fd) override;
+
+  ProtobufTypes::MessagePtr createEmptyConfigProto() override;
+  std::string name() const override {
+    return "envoy.extensions.network.socket_interface.test_socket_interface";
+  };
+
+  uint32_t getAcceptedSocketIndex() {
+    absl::MutexLock lock(&mutex_);
+    return accepted_socket_index_++;
+  }
+
+  SocketInterface* const previous_socket_interface_;
+  TestIoSocketHandle::WritevOverrideProc writev_overide_proc_;
+  mutable absl::Mutex mutex_;
+  uint32_t accepted_socket_index_ ABSL_GUARDED_BY(mutex_) = 0;
+};
+
+} // namespace Network
+} // namespace Envoy