Feature: Golang Http Tcp Bridge

api/BUILD

@@ -97,6 +97,7 @@ proto_library(
         "//contrib/envoy/extensions/private_key_providers/qat/v3alpha:pkg",
         "//contrib/envoy/extensions/regex_engines/hyperscan/v3alpha:pkg",
         "//contrib/envoy/extensions/router/cluster_specifier/golang/v3alpha:pkg",
+        "//contrib/envoy/extensions/upstreams/http/tcp/golang/v3alpha:pkg",
         "//contrib/envoy/extensions/vcl/v3alpha:pkg",
         "//envoy/admin/v3:pkg",
         "//envoy/config/accesslog/v3:pkg",

api/contrib/envoy/extensions/upstreams/http/tcp/golang/v3alpha/BUILD

@@ -0,0 +1,9 @@
+# DO NOT EDIT. This file is generated by tools/proto_format/proto_sync.py.
+
+load("@envoy_api//bazel:api_build_system.bzl", "api_proto_package")
+
+licenses(["notice"])  # Apache 2
+
+api_proto_package(
+    deps = ["@com_github_cncf_xds//udpa/annotations:pkg"],
+)

api/contrib/envoy/extensions/upstreams/http/tcp/golang/v3alpha/golang.proto

@@ -0,0 +1,46 @@
+syntax = "proto3";
+
+package envoy.extensions.upstreams.http.tcp.golang.v3alpha;
+
+import "google/protobuf/any.proto";
+
+import "udpa/annotations/status.proto";
+import "validate/validate.proto";
+
+option java_package = "io.envoyproxy.envoy.extensions.upstreams.http.tcp.golang.v3alpha";
+option java_outer_classname = "GolangProto";
+option java_multiple_files = true;
+option go_package = "github.com/envoyproxy/go-control-plane/contrib/envoy/extensions/upstreams/http/tcp/golang/v3alpha";
+option (udpa.annotations.file_status).package_version_status = ACTIVE;
+
+// [#protodoc-title: Tcp Connection Pool]
+
+// A connection pool which forwards downstream HTTP as TCP to upstream,
+// [#extension: envoy.upstreams.http.tcp]
+message Config {
+  // Globally unique ID for a dynamic library file.
+  string library_id = 1 [(validate.rules).string = {min_len: 1}];
+
+  // Path to a dynamic library implementing the
+  // :repo:`DownstreamFilter API <contrib/golang/common/go/api.DownstreamFilter>`
+  // interface.
+  // [#comment:TODO(wangfakang): Support for downloading libraries from remote repositories.]
+  string library_path = 2 [(validate.rules).string = {min_len: 1}];
+
+  // Globally unique name of the Go plugin.
+  //
+  // This name **must** be consistent with the name registered in ``tcp::RegisterTcpUpstreamFactoryAndConfigParser``
+  //
+  string plugin_name = 3 [(validate.rules).string = {min_len: 1}];
+
+  // Configuration for the Go plugin.
+  //
+  // .. note::
+  //     This configuration is only parsed in the go plugin, and is therefore not validated
+  //     by Envoy.
+  //
+  //     See the :repo:`DownstreamFilter API <contrib/golang/common/go/api/filter.go>`
+  //     for more information about how the plugin's configuration data can be accessed.
+  //
+  google.protobuf.Any plugin_config = 4;
+}

api/versioning/BUILD

@@ -35,6 +35,7 @@ proto_library(
         "//contrib/envoy/extensions/private_key_providers/qat/v3alpha:pkg",
         "//contrib/envoy/extensions/regex_engines/hyperscan/v3alpha:pkg",
         "//contrib/envoy/extensions/router/cluster_specifier/golang/v3alpha:pkg",
+        "//contrib/envoy/extensions/upstreams/http/tcp/golang/v3alpha:pkg",
         "//contrib/envoy/extensions/vcl/v3alpha:pkg",
         "//envoy/admin/v3:pkg",
         "//envoy/config/accesslog/v3:pkg",

contrib/contrib_build_config.bzl

@@ -28,6 +28,7 @@ CONTRIB_EXTENSIONS = {
     "envoy.filters.network.postgres_proxy":                     "//contrib/postgres_proxy/filters/network/source:config",
     "envoy.filters.network.rocketmq_proxy":                     "//contrib/rocketmq_proxy/filters/network/source:config",
     "envoy.filters.network.golang":                             "//contrib/golang/filters/network/source:config",
+    "envoy.upstreams.http.tcp.golang":                          "//contrib/golang/upstreams/http/tcp/source:config",
 
     #
     # Sip proxy

contrib/golang/common/dso/dso.cc

@@ -242,5 +242,53 @@ void NetworkFilterDsoImpl::envoyGoFilterOnSemaDec(void* w) {
   envoy_go_filter_on_sema_dec_(w);
 }
 
+TcpUpstreamDsoImpl::TcpUpstreamDsoImpl(const std::string dso_name)
+    : TcpUpstreamDso(dso_name) {
+  loaded_ &= dlsymInternal<decltype(envoy_go_on_tcp_upstream_config_)>(
+      envoy_go_on_tcp_upstream_config_, handler_, dso_name,
+      "envoyGoOnTcpUpstreamConfig");
+  loaded_ &= dlsymInternal<decltype(envoy_go_tcp_upstream_destroy_plugin_config_)>(
+    envoy_go_tcp_upstream_destroy_plugin_config_, handler_, dso_name,
+    "envoyGoTcpUpstreamDestroyPluginConfig");
+  loaded_ &= dlsymInternal<decltype(envoy_go_encode_header_)>(
+    envoy_go_encode_header_, handler_, dso_name, "envoyGoEncodeHeader");
+  loaded_ &= dlsymInternal<decltype(envoy_go_on_encode_data_)>(
+    envoy_go_on_encode_data_, handler_, dso_name, "envoyGoEncodeData");
+  loaded_ &= dlsymInternal<decltype(envoy_go_on_upstream_data_)>(
+      envoy_go_on_upstream_data_, handler_, dso_name, "envoyGoOnUpstreamData");
+  loaded_ &= dlsymInternal<decltype(envoy_go_on_tcp_upstream_destroy_)>(
+    envoy_go_on_tcp_upstream_destroy_, handler_, dso_name, "envoyGoOnTcpUpstreamDestroy");
+}
+
+GoUint64 TcpUpstreamDsoImpl::envoyGoOnTcpUpstreamConfig(httpConfig* p0) {
+  ASSERT(envoy_go_on_tcp_upstream_config_ != nullptr);
+  return envoy_go_on_tcp_upstream_config_(p0);
+}
+
+void TcpUpstreamDsoImpl::envoyGoTcpUpstreamDestroyPluginConfig(GoUint64 p0) {
+  ASSERT(envoy_go_tcp_upstream_destroy_plugin_config_ != nullptr);
+  return envoy_go_tcp_upstream_destroy_plugin_config_(p0);
+}
+
+GoUint64 TcpUpstreamDsoImpl::envoyGoEncodeHeader(processState* state, GoUint64 end_stream, GoUint64 header_num,  GoUint64 header_bytes, GoUint64 buf_ptr, GoUint64 buf_len) {
+  ASSERT(envoy_go_encode_header_ != nullptr);
+  return envoy_go_encode_header_(state, end_stream, header_num, header_bytes, buf_ptr, buf_len);
+}
+
+GoUint64 TcpUpstreamDsoImpl::envoyGoEncodeData(processState* state, GoUint64 end_stream, GoUint64 buf_ptr, GoUint64 buf_len) {
+  ASSERT(envoy_go_on_upstream_data_ != nullptr);
+  return envoy_go_on_encode_data_(state, end_stream, buf_ptr, buf_len);
+}
+
+GoUint64 TcpUpstreamDsoImpl::envoyGoOnUpstreamData(processState* state, GoUint64 end_stream, GoUint64 header_num, GoUint64 header_bytes, GoUint64 buf_ptr, GoUint64 buf_len) {
+  ASSERT(envoy_go_on_upstream_data_ != nullptr);
+  return envoy_go_on_upstream_data_(state, end_stream, header_num, header_bytes, buf_ptr, buf_len);
+}
+
+void TcpUpstreamDsoImpl::envoyGoOnTcpUpstreamDestroy(httpRequest* p0) {
+  ASSERT(envoy_go_on_tcp_upstream_destroy_ != nullptr);
+  envoy_go_on_tcp_upstream_destroy_(p0);
+}
+
 } // namespace Dso
 } // namespace Envoy

contrib/golang/common/dso/dso.h

@@ -201,6 +201,60 @@ class NetworkFilterDsoImpl : public NetworkFilterDso {
 
 using NetworkFilterDsoPtr = std::shared_ptr<NetworkFilterDso>;
 
+
+class TcpUpstreamDso : public Dso {
+public:
+  TcpUpstreamDso() = default;
+  TcpUpstreamDso(const std::string dso_name) : Dso(dso_name){};
+  ~TcpUpstreamDso() override = default;
+
+  virtual GoUint64 envoyGoOnTcpUpstreamConfig(httpConfig* p0) PURE;
+
+  virtual void envoyGoTcpUpstreamDestroyPluginConfig(GoUint64 p0) PURE;
+
+  virtual GoUint64 envoyGoEncodeHeader(processState* state, GoUint64 end_stream, GoUint64 header_num,  GoUint64 header_bytes, GoUint64 buf_ptr, GoUint64 buf_len) PURE;
+
+  virtual GoUint64 envoyGoEncodeData(processState* state, GoUint64 end_stream, GoUint64 buf_ptr, GoUint64 buf_len) PURE;
+
+  virtual GoUint64 envoyGoOnUpstreamData(processState* state, GoUint64 end_stream, GoUint64 header_num, GoUint64 header_bytes, GoUint64 buf_ptr, GoUint64 buf_len) PURE;
+
+  virtual void envoyGoOnTcpUpstreamDestroy(httpRequest* p0) PURE;
+
+};
+
+class TcpUpstreamDsoImpl : public TcpUpstreamDso {
+public:
+  TcpUpstreamDsoImpl(const std::string dso_name);
+  ~TcpUpstreamDsoImpl() override = default;
+
+  GoUint64 envoyGoOnTcpUpstreamConfig(httpConfig* p0) override;
+
+  void envoyGoTcpUpstreamDestroyPluginConfig(GoUint64 p0) override;
+
+  GoUint64 envoyGoEncodeHeader(processState* state, GoUint64 end_stream, GoUint64 header_num,  GoUint64 header_bytes, GoUint64 buf_ptr, GoUint64 buf_len) override;
+
+  GoUint64 envoyGoEncodeData(processState* state, GoUint64 end_stream, GoUint64 buf_ptr, GoUint64 buf_len) override;
+
+  GoUint64 envoyGoOnUpstreamData(processState* state, GoUint64 end_stream, GoUint64 header_num, GoUint64 header_bytes, GoUint64 buf_ptr, GoUint64 buf_len) override;
+
+  void envoyGoOnTcpUpstreamDestroy(httpRequest* p0) override;
+
+private:
+  GoUint64 (*envoy_go_on_tcp_upstream_config_)(httpConfig* p0) = {nullptr};
+
+  void (*envoy_go_tcp_upstream_destroy_plugin_config_)(GoUint64 p0) = {nullptr};
+
+  GoUint64 (*envoy_go_encode_header_)(processState* p0, GoUint64 p1, GoUint64 p2, GoUint64 p3, GoUint64 buf_ptr, GoUint64 buf_len) = {nullptr};
+
+  GoUint64 (*envoy_go_on_encode_data_)(processState* state, GoUint64 end_stream, GoUint64 buf_ptr, GoUint64 buf_len) = {nullptr};
+
+  GoUint64 (*envoy_go_on_upstream_data_)(processState* state, GoUint64 end_stream, GoUint64 header_size, GoUint64 header_byte_size,  GoUint64 buf_ptr, GoUint64 buf_len) = {nullptr};
+
+  void (*envoy_go_on_tcp_upstream_destroy_)(httpRequest* p0) = {nullptr};
+};
+
+using TcpUpstreamDsoPtr = std::shared_ptr<TcpUpstreamDso>;
+
 /*
  * We do not unload a dynamic library once it is loaded. This is because
  * Go shared library could not be unload by dlclose yet, see:

contrib/golang/common/dso/libgolang.h

@@ -193,6 +193,31 @@ extern GoUint64 envoyGoFilterOnUpstreamData(void* f, GoUint64 data_size, GoUint6
 // github.com/envoyproxy/envoy/contrib/golang/filters/network/source/go/pkg/network.envoyGoFilterOnUpstreamEvent
 extern void envoyGoFilterOnUpstreamEvent(void* f, GoInt event);
 
+// go:linkname envoyGoOnTcpUpstreamConfig
+// github.com/envoyproxy/envoy/contrib/golang/filters/upstreams/http/tcp/source/go/pkg/upstreams/http/tcp.envoyGoOnTcpUpstreamConfig
+extern GoUint64 envoyGoOnTcpUpstreamConfig(GoUint64 library_id_ptr, GoUint64 library_id_len,
+                                                   GoUint64 config_ptr, GoUint64 config_len);
+
+// go:linkname envoyGoTcpUpstreamDestroyHttpPluginConfig
+// github.com/envoyproxy/envoy/contrib/golang/filters/http/source/go/pkg/http.envoyGoTcpUpstreamDestroyPluginConfig
+extern void envoyGoTcpUpstreamDestroyPluginConfig(GoUint64 id);
+
+// go:linkname envoyGoEncodeHeader
+// github.com/envoyproxy/envoy/contrib/golang/filters/upstreams/http/tcp/source/go/pkg/upstreams/http/tcp.envoyGoEncodeHeader
+extern GoUint64 envoyGoEncodeHeader(processState* r, GoUint64 end_stream, GoUint64 header_num,  GoUint64 header_bytes, GoUint64 buf_ptr, GoUint64 buf_len);
+
+// go:linkname envoyGoEncodeData
+// github.com/envoyproxy/envoy/contrib/golang/filters/upstreams/http/tcp/source/go/pkg/upstreams/http/tcp.envoyGoEncodeData
+extern GoUint64 envoyGoEncodeData(processState* state, GoUint64 end_stream, GoUint64 buf_ptr, GoUint64 buf_len);
+
+// go:linkname envoyGoOnUpstreamData
+// github.com/envoyproxy/envoy/contrib/golang/filters/upstreams/http/tcp/source/go/pkg/upstreams/http/tcp.envoyGoOnUpstreamData
+extern GoUint64 envoyGoOnUpstreamData(processState* state, GoUint64 end_stream, GoUint64 header_num, GoUint64 header_bytes, GoUint64 buf_ptr, GoUint64 buf_len);
+
+// go:linkname envoyGoOnTcpUpstreamDestroy
+// github.com/envoyproxy/envoy/contrib/golang/filters/upstreams/http/tcp/source/go/pkg/upstreams/http/tcp.envoyGoOnTcpUpstreamDestroy
+extern void envoyGoOnTcpUpstreamDestroy(httpRequest* r);
+
 #ifdef __cplusplus
 }
 #endif

contrib/golang/common/go/api/api.h

@@ -147,6 +147,16 @@ CAPIStatus envoyGoFilterSetFilterState(void* wrapper, void* key, void* value, in
                                        int life_span, int stream_sharing);
 CAPIStatus envoyGoFilterGetFilterState(void* wrapper, void* key, void* value);
 
+// tcp upstream
+CAPIStatus envoyGoTcpUpstreamCopyHeaders(void* s, void* strs, void* buf);
+CAPIStatus envoyGoTcpUpstreamSetRespHeader(void* s, void* key_data, int key_len, void* value_data, int value_len, headerAction action);
+CAPIStatus envoyGoTcpUpstreamRemoveRespHeader(void* s, void* key_data, int key_len);
+CAPIStatus envoyGoTcpUpstreamGetBuffer(void* s, uint64_t buffer, void* value);
+CAPIStatus envoyGoTcpUpstreamDrainBuffer(void* s, uint64_t buffer, uint64_t length);
+CAPIStatus envoyGoTcpUpstreamSetBufferHelper(void* s, uint64_t buffer, void* data, int length, bufferAction action);
+CAPIStatus envoyGoTcpUpstreamGetStringValue(void* r, int id, uint64_t* value_data, int* value_len);
+CAPIStatus envoyGoTcpUpstreamSetSelfHalfCloseForUpstreamConn(void* r, int enabled);
+
 #ifdef __cplusplus
 } // extern "C"
 #endif

contrib/golang/common/go/api/capi.go

@@ -99,6 +99,22 @@ type NetworkCAPI interface {
 	UpstreamInfo(f unsafe.Pointer, infoType int) string
 }
 
+type TcpUpstreamCAPI interface {
+	// Header related
+	CopyHeaders(s unsafe.Pointer, num uint64, bytes uint64) map[string][]string
+	SetRespHeader(s unsafe.Pointer, key string, value string, add bool)
+	RemoveRespHeader(s unsafe.Pointer, key string)
+	// Buffer related
+	GetBuffer(s unsafe.Pointer, bufferPtr uint64, length uint64) []byte
+	DrainBuffer(s unsafe.Pointer, bufferPtr uint64, length uint64)
+	SetBufferHelper(s unsafe.Pointer, bufferPtr uint64, value string, action BufferAction)
+	SetBytesBufferHelper(s unsafe.Pointer, bufferPtr uint64, value []byte, action BufferAction)
+	// Get the specified value by key
+	GetStringValue(r unsafe.Pointer, id int) (string, bool)
+
+	SetSelfHalfCloseForUpstreamConn(r unsafe.Pointer, enabled int)
+}
+
 type CommonCAPI interface {
 	Log(level LogType, message string)
 	LogLevel() LogType

contrib/golang/common/go/api/filter.go

@@ -34,6 +34,12 @@ type (
 	EmptyDownstreamFilter struct{}
 	// EmptyUpstreamFilter provides the no-op implementation of the UpstreamFilter interface
 	EmptyUpstreamFilter struct{}
+	// EmptyTcpUpstreamFilter provides the no-op implementation of the EmptyTcpUpstreamFilter interface
+	EmptyTcpUpstreamFilter struct{}
+
+	PassThroughTcpUpstream struct {
+		EmptyTcpUpstreamFilter
+	}
 )
 
 // request
@@ -336,3 +342,52 @@ type GaugeMetric interface {
 // TODO
 type HistogramMetric interface {
 }
+
+type TcpUpstreamCallbackHandler interface {
+	// GetRouteName returns the name of the route which got matched
+	GetRouteName() string
+	// GetVirtualClusterName returns the name of the virtual cluster which got matched
+	GetVirtualClusterName() string
+	// SetSelfHalfCloseForUpstreamConn default is false
+	SetSelfHalfCloseForUpstreamConn(enabled bool)
+}
+
+type TcpUpstreamFilter interface {
+
+	// Invoked when header is delivered from the downstream.
+	// Notice: headerMap and dataToUpstream cannot be invoked after the func return.
+	EncodeHeaders(headerMap RequestHeaderMap, dataToUpstream BufferInstance, endOfStream bool) TcpUpstreamStatus
+
+	// Streaming, Invoked when data is delivered from the downstream.
+	// Notice: buffer cannot be invoked after the func return.
+	EncodeData(buffer BufferInstance, endOfStream bool) TcpUpstreamStatus
+
+	// Streaming, Called when data is read on from tcp upstream.
+	// Notice-1: when return TcpUpstreamContinue, resp headers will be send to http all at once; from then on, you MUST NOT invoke responseHeaderForSet at any time(or you will get panic).
+	// Notice-2: responseHeaderForSet and buffer cannot be invoked after the func return.
+	OnUpstreamData(responseHeaderForSet ResponseHeaderMap, buffer BufferInstance, endOfStream bool) TcpUpstreamStatus
+
+	// destroy filter
+	OnDestroy()
+}
+
+func (*EmptyTcpUpstreamFilter) EncodeHeaders(headerMap RequestHeaderMap, dataToUpstream BufferInstance, endOfStream bool) TcpUpstreamStatus {
+	return TcpUpstreamContinue
+}
+
+func (*EmptyTcpUpstreamFilter) EncodeData(buffer BufferInstance, endOfStream bool) TcpUpstreamStatus {
+	return TcpUpstreamContinue
+}
+
+func (*EmptyTcpUpstreamFilter) OnUpstreamData(responseHeaderForSet ResponseHeaderMap, buffer BufferInstance, endOfStream bool) TcpUpstreamStatus {
+	return TcpUpstreamContinue
+}
+
+func (*EmptyTcpUpstreamFilter) OnDestroy() {
+}
+
+type TcpUpstreamFactory func(config interface{}, callbacks TcpUpstreamCallbackHandler) TcpUpstreamFilter
+
+type TcpUpstreamConfigParser interface {
+	Parse(any *anypb.Any) (interface{}, error)
+}