Add an new tap sink

[coolg92003]

Title: Enhancement to add a new tap sink

Description:

current tap sinks:

1. File_per_tap, the performance is very low
2. Streaming/buffered_Admin, it will block main thread routine work. Also the performance is low.
3. Custom_sink,issue link until now, I don’t know how to use it based current investigation :(

Based above reason, I want to add a nice performance “UDP” sink.
As why this sink is needed, as you know, all upstream traffic is passed by TLS. Especially, TLS1.3, it can’t get original data even with tcpdump tool.

[optional Relevant Links:]

Any extra documentation required to understand the issue.

[zhmzhen]

wait for contribution! ^_^

Thanks,
zhmzhen

[coolg92003]

Note

Below is only private code for unix socket sink( not UT code). paste here for reference

1) configuration changes

--- a/api/envoy/config/tap/v3/common.proto
+++ b/api/envoy/config/tap/v3/common.proto
@@ -2,6 +2,7 @@ syntax = "proto3";

 package envoy.config.tap.v3; 
+import "envoy/config/core/v3/address.proto";
 import "envoy/config/common/matcher/v3/matcher.proto";
 import "envoy/config/core/v3/base.proto";
 import "envoy/config/core/v3/extension.proto";
@@ -263,6 +264,11 @@ message OutputSink {
 
     // Tap output filter will be defined by an extension type
     core.v3.TypedExtensionConfig custom_sink = 6;
+
+    // US output sink
+    UnixSocketSink us_sink = 8;
   }
 }
 
+//  The Unix socket sink for fast sinking
+message UnixSocketSink {
+  string su_path = 1 [(validate.rules).string = {min_len: 1}];
+}

2) Real code

2.1)

--- a/source/extensions/common/tap/tap_config_base.h
+++ b/source/extensions/common/tap/tap_config_base.h

+// UNIX socket sink definition
+class USTapSink : public Sink {
+public:
+  USTapSink(const envoy::config::tap::v3::UnixSocketSink& config) : config_(config) {
+    initUSData();
+  }
+  ~USTapSink() {
+    if (socket_fd_ > 0) {
+      close(socket_fd_);
+      socket_fd_ = -1;
+    }
+  }
+
+  // Sink
+  PerTapSinkHandlePtr
+  createPerTapSinkHandle(uint64_t trace_id,
+                         envoy::config::tap::v3::OutputSink::OutputSinkTypeCase) override {
+    return std::make_unique<USTapSinkHandle>(*this, trace_id);
+  }
+  void initUSData(void);
+
+private:
+  struct USTapSinkHandle : public PerTapSinkHandle {
+    USTapSinkHandle(USTapSink& parent, uint64_t trace_id)
+        : parent_(parent), trace_id_(trace_id) {}
+
+    // PerTapSinkHandle
+    void submitTrace(TraceWrapperPtr&& trace,
+                     envoy::config::tap::v3::OutputSink::Format format) override;
+
+    USTapSink& parent_;
+    const uint64_t trace_id_;
+  };
+
+  const envoy::config::tap::v3::UnixSocketSink config_;
+  int socket_fd_ = -1;
+  struct sockaddr_un server_addr_;
+};ore--
+// The end of Unix Socket sink
+

2.2) implementation

-- a/source/extensions/common/tap/tap_config_base.cc
+++ b/source/extensions/common/tap/tap_config_base.cc
@@ -8,6 +8,7 @@

 #include "source/common/common/assert.h"
 #include "source/common/common/fmt.h"
 #include "source/common/config/utility.h"
+#include "source/common/network/utility.h"
 #include "source/common/protobuf/utility.h"
 #include "source/extensions/common/matcher/matcher.h"
 
@@ -98,6 +99,20 @@ TapConfigBaseImpl::TapConfigBaseImpl(const envoy::config::tap::v3::TapConfig& pr
     sink_ = std::make_unique<FilePerTapSink>(sinks[0].file_per_tap());
     sink_to_use_ = sink_.get();
     break;
+  case ProtoOutputSink::OutputSinkTypeCase::kUsSink:
+    {
+    sink_ = std::make_unique<USTapSink>(sinks[0].us_sink());
+    sink_to_use_ = sink_.get();
+    ENVOY_LOG_MISC(debug, "CFXUS, Done init data");
+    }
+    break;
   case ProtoOutputSink::OutputSinkTypeCase::kCustomSink: {

+void USTapSink::initUSData(void) {
+   ENVOY_LOG_MISC(debug, "CFXUS, USTapSink::initUSData, enter");
+
+   if ((socket_fd_ = socket(AF_UNIX, SOCK_STREAM, 0)) < 0) {
+     ENVOY_LOG_MISC(debug, "CFXUS, USTapSink::initUDPData failed to create US socket\n");
+     return;
+   }
+   server_addr_.sun_family = AF_UNIX;
+   strcpy(server_addr_.sun_path, config_.su_path().c_str());
+   int len = sizeof(server_addr_.sun_family) + strlen(server_addr_.sun_path);
+   if (connect(socket_fd_, reinterpret_cast<struct sockaddr *>(&server_addr_), len) < 0) {
+     ENVOY_LOG_MISC(debug, "CFXU, USTapSink::initUSData Connect to server, su_path={}", config_.su_path().c_str());
+     close(socket_fd_);
+     socket_fd_ = -1;
+     return;
+   }
+
+   ENVOY_LOG_MISC(debug, "CFXUS, exit USTapSink::initUSData, sokcet_fd={}, su_path={}",
+                   socket_fd_,  config_.su_path().c_str());
+}
+
+void USTapSink::USTapSinkHandle::submitTrace(
+    TraceWrapperPtr&& trace, envoy::config::tap::v3::OutputSink::Format format) {
+  ENVOY_LOG_MISC(debug, "CFXUS, UsTapSink::UsTapSinkHandle::submitTrace, enter [id={}] ", trace_id_);
+  switch (format) {
+    PANIC_ON_PROTO_ENUM_SENTINEL_VALUES;
+  case envoy::config::tap::v3::OutputSink::PROTO_BINARY:
+  case envoy::config::tap::v3::OutputSink::PROTO_BINARY_LENGTH_DELIMITED:
+  case envoy::config::tap::v3::OutputSink::PROTO_TEXT:
+    ENVOY_LOG_MISC(debug, "CFXUS, Notsupport PROTO_BINARY, PROTO_BINARY_LENGTH_DELIMITED,  PROTO_TEXT\n");
+    break;
+  case envoy::config::tap::v3::OutputSink::JSON_BODY_AS_BYTES:
+  case envoy::config::tap::v3::OutputSink::JSON_BODY_AS_STRING:
+    {
+      std::string json_string =  MessageUtil::getJsonStringFromMessageOrError(*trace, true, true);
+      ENVOY_LOG_MISC(debug, "CFXU, message is <\n{}\n", json_string);
+      if (parent_.socket_fd_ < 0) {
+        ENVOY_LOG_MISC(debug, "CFXU, parent_.socket_fd_ < 0 {}\n", parent_.socket_fd_);
+      } else {
+        int l_str_len = json_string.size();
+        int l_w_len = write(parent_.socket_fd_, json_string.c_str(), l_str_len);
+        int l_w_total_len = 0;
+        int l_con_len = 0;
+        while ( 1 )
+        {
+          if (l_w_len <= 0) { 
+           if (errno == EINTR || errno == EAGAIN || errno == EWOULDBLOCK) {
+              if (l_con_len++ >= 3) {
+                ENVOY_LOG_MISC(debug, "CFXUS, Failed to write US after 3 times\n");
+                break;
+              }
+             continue;
+            }
+            // other errors
+            ENVOY_LOG_MISC(debug, "CFXUS, other Failure to write US\n");
+            break;
+          } else {
+            l_w_total_len = l_w_total_len + l_w_len;
+            if (l_w_total_len == l_str_len) {
+               // write done
+               ENVOY_LOG_MISC(debug, "CFXUS, write US size {}\n", l_w_total_len);
+               break;
+            } else {
+               l_w_len = write(parent_.socket_fd_,
+                               json_string.c_str() + l_w_total_len, l_str_len - l_w_total_len);
+            }
+          }
+       }
+     }
+   }
+   break;
+  }
+  ENVOY_LOG_MISC(debug, "CFXUS, USTapSink::USTapSinkHandle::submitTrace: End");
+  return;
+}
 
 } // namespace Tap
 } // namespace Common

Envoy configuration:

See attachment for full configuration file

[coolg92003]

below is the full configuration file for unixsocket
unixSocketSink.yaml.txt

Important

When using above file, please remove .txt

[nezdolik]

cc @mattklein123 @xu1zhou

[coolg92003]

Please let me know if anyone want to try private code changes.
thanks
Cliff

[coolg92003]

Changing UDP sink code ......

[coolg92003]

PR is #37172
See this pic for test configuration

If the cpu is A before enable "udp sink", then after enable "udp sink", then cpu is B
the total increasing (B-A)/A = about 40%-50%
thanks
Cliff

[coolg92003]

PR is wait for review

[coolg92003]

change to add UDP extension sink for tap custom sink