Feature: Golang Http Tcp Bridge

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

@@ -0,0 +1,12 @@
+# 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",
+        "@com_github_cncf_xds//xds/annotations/v3:pkg",
+    ],
+)

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

@@ -0,0 +1,47 @@
+syntax = "proto3";
+
+package envoy.extensions.upstreams.http.tcp.golang.v3alpha;
+
+import "google/protobuf/any.proto";
+
+import "validate/validate.proto";
+
+import "udpa/annotations/status.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 ``network::RegisterNetworkFilterConfigFactory``
+  //
+  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;
+}

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,54 @@ 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, GoInt p1) {
+  ASSERT(envoy_go_tcp_upstream_destroy_plugin_config_ != nullptr);
+  return envoy_go_tcp_upstream_destroy_plugin_config_(p0, p1);
+}
+
+void TcpUpstreamDsoImpl::envoyGoEncodeHeader(processState* p0, GoUint64 p1, GoUint64 p2,
+                                                      GoUint64 p3) {
+  ASSERT(envoy_go_encode_header_ != nullptr);
+  return envoy_go_encode_header_(p0, p1, p2, p3);
+}
+
+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 buf_ptr, GoUint64 buf_len) {
+  ASSERT(envoy_go_on_upstream_data_ != nullptr);
+  return envoy_go_on_upstream_data_(state, end_stream, buf_ptr, buf_len);
+}
+
+void TcpUpstreamDsoImpl::envoyGoOnTcpUpstreamDestroy(httpRequest* p0, int p1) {
+  ASSERT(envoy_go_on_tcp_upstream_destroy_ != nullptr);
+  envoy_go_on_tcp_upstream_destroy_(p0, GoUint64(p1));
+}
+
 } // 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, GoInt p1) PURE;
+
+  virtual void envoyGoEncodeHeader(processState* p0, GoUint64 p1, GoUint64 p2, GoUint64 p3) 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 buf_ptr, GoUint64 buf_len) PURE;
+
+  virtual void envoyGoOnTcpUpstreamDestroy(httpRequest* p0, int p1) PURE;
+
+};
+
+class TcpUpstreamDsoImpl : public TcpUpstreamDso {
+public:
+  TcpUpstreamDsoImpl(const std::string dso_name);
+  ~TcpUpstreamDsoImpl() override = default;
+
+  GoUint64 envoyGoOnTcpUpstreamConfig(httpConfig* p0) override;
+
+  void envoyGoTcpUpstreamDestroyPluginConfig(GoUint64 p0, GoInt p1) override;
+
+  void envoyGoEncodeHeader(processState* p0, GoUint64 p1, GoUint64 p2, GoUint64 p3) override;
+
+  GoUint64 envoyGoEncodeData(processState* state, GoUint64 end_stream, GoUint64 buf_ptr, GoUint64 buf_len) override;
+
+  GoUint64 envoyGoOnUpstreamData(processState* state, GoUint64 end_stream, GoUint64 buf_ptr, GoUint64 buf_len) override;
+
+  void envoyGoOnTcpUpstreamDestroy(httpRequest* p0, int p1) override;
+
+private:
+  GoUint64 (*envoy_go_on_tcp_upstream_config_)(httpConfig* p0) = {nullptr};
+
+  void (*envoy_go_tcp_upstream_destroy_plugin_config_)(GoUint64 p0, GoInt p1) = {nullptr};
+
+  void (*envoy_go_encode_header_)(processState* p0, GoUint64 p1, GoUint64 p2, GoUint64 p3) = {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 buf_ptr, GoUint64 buf_len) = {nullptr};
+
+  void (*envoy_go_on_tcp_upstream_destroy_)(httpRequest* p0, GoUint64 p1) = {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, GoInt need_delay);
+
+// go:linkname envoyGoEncodeHeader
+// github.com/envoyproxy/envoy/contrib/golang/filters/upstreams/http/tcp/source/go/pkg/upstreams/http/tcp.envoyGoEncodeHeader
+extern void envoyGoEncodeHeader(processState* r, GoUint64 end_stream, GoUint64 header_num,  GoUint64 header_bytes);
+
+// 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 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, GoUint64 reason);
+
 #ifdef __cplusplus
 }
 #endif

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

@@ -145,6 +145,18 @@ 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 envoyGoTcpUpstreamInfo(void* u, int info_type, void* ret);
+CAPIStatus envoyGoTcpUpstreamConnEnableHalfClose(void* u, int enable_half_close);
+CAPIStatus envoyGoTcpUpstreamGetHeader(void* s, void* key_data, int key_len, uint64_t* value_data, int* value_len);
+CAPIStatus envoyGoTcpUpstreamCopyHeaders(void* s, void* strs, void* buf);
+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);
+void envoyGoTcpUpstreamFinalize(void* r, int reason);
+void envoyGoConfigTcpUpstreamFinalize(void* c);
+
 #ifdef __cplusplus
 } // extern "C"
 #endif

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

@@ -98,6 +98,28 @@ type NetworkCAPI interface {
 	UpstreamInfo(f unsafe.Pointer, infoType int) string
 }
 
+ type TcpUpstreamCAPI interface {
+	// UpstreamConnEnableHalfClose upstream conn EnableHalfClose
+	UpstreamConnEnableHalfClose(f unsafe.Pointer, enableHalfClose int)
+
+	HttpGetHeader(s unsafe.Pointer, key string) string
+	HttpCopyHeaders(s unsafe.Pointer, num uint64, bytes uint64) map[string][]string
+	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)
+	GetStringValue(r unsafe.Pointer, id int) (string, bool)
+
+	Finalize(r unsafe.Pointer, reason int)
+
+	// /* These APIs are related to config, use the pointer of config. */
+	ConfigFinalize(c unsafe.Pointer)
+
+	/* These APIs have nothing to do with request */
+	Log(level LogType, message string)
+	LogLevel() LogType
+ }
+
 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
@@ -255,6 +261,12 @@ func (*EmptyUpstreamFilter) OnData(buffer []byte, endOfStream bool) FilterStatus
 func (*EmptyUpstreamFilter) OnEvent(event ConnectionEvent) {
 }
 
+type TcpUpstreamCallbackHandler interface {
+	FilterCallbackHandler
+	// EnableHalfClose only for upstream connection
+	EnableHalfClose(enabled bool)
+}
+
 type ConnectionCallback interface {
 	// StreamInfo returns the stream info of the connection
 	StreamInfo() StreamInfo
@@ -330,3 +342,36 @@ type GaugeMetric interface {
 // TODO
 type HistogramMetric interface {
 }
+
+type TcpUpstreamFilter interface {
+	EncodeHeaders(headerMap RequestHeaderMap, endOfStream bool)
+	// Invoked when data is delivered from the upstream connection.
+	EncodeData(buffer BufferInstance, endOfStream bool) (EndStream bool)
+	// Called when data is read on from tcp upstream.
+	OnUpstreamData(buffer BufferInstance, endOfStream bool) UpstreamDataStatus
+	// destroy filter
+	OnDestroy(DestroyReason)
+}
+
+func (*EmptyTcpUpstreamFilter) EncodeHeaders(headerMap RequestHeaderMap, endOfStream bool) {
+}
+
+func (*EmptyTcpUpstreamFilter) EncodeData(buffer BufferInstance, endOfStream bool) bool {
+	return true
+}
+
+func (*EmptyTcpUpstreamFilter) OnUpstreamData(buffer BufferInstance, endOfStream bool) UpstreamDataStatus {
+	return UpstreamDataFinish
+}
+
+func (*EmptyTcpUpstreamFilter) OnDestroy(DestroyReason) {
+}
+
+type TcpUpstreamFactory func(config interface{}, callbacks TcpUpstreamCallbackHandler) TcpUpstreamFilter
+
+type TcpUpstreamConfigParser interface {
+	// Parse the proto message to any Go value, and return error to reject the config.
+	// This is called when Envoy receives the config from the control plane.
+	// Also, you can define Metrics through the callbacks, and the callbacks will be nil when parsing the route config.
+	Parse(any *anypb.Any, callbacks ConfigCallbackHandler) (interface{}, error)
+}

contrib/golang/common/go/api/type.go

@@ -450,3 +450,63 @@ var (
 )
 
 // *************** errors end **************//
+
+ // Status codes returned by tcp upstream that can deal with data.
+ type UpstreamDataStatus int
+
+ const (
+	 // Continue to deal with further data.
+	 UpstreamDataContinue UpstreamDataStatus = 0
+	 // Finish dealing with data.
+	 UpstreamDataFinish UpstreamDataStatus = 1
+	 // Failure when dealing with data.
+	 UpstreamDataFailure UpstreamDataStatus = 2
+ )
+
+ func (s UpstreamDataStatus) String() string {
+	 switch s {
+	 case UpstreamDataContinue:
+		 return "Continue"
+	 case UpstreamDataFinish:
+		 return "Finish"
+	 case UpstreamDataFailure:
+		 return "Failure"
+	 }
+	 return "unknown"
+ }
+
+ // Info types called by tcp upstream that can get the corresponding info from c++ side.
+ type TcpUpstreamInfoType int
+
+ const (
+	 TcpUpstreamInfoRouterName  TcpUpstreamInfoType = 0
+	 TcpUpstreamInfoClusterName TcpUpstreamInfoType = 1
+ )
+
+ func (t TcpUpstreamInfoType) String() string {
+	 switch t {
+	 case TcpUpstreamInfoRouterName:
+		 return "TcpUpstreamInfoRouterName"
+	 case TcpUpstreamInfoClusterName:
+		 return "TcpUpstreamInfoClusterName"
+	 }
+	 return "unknown"
+ }
+
+ type EndStreamType int
+
+ const (
+	 NotEndStream EndStreamType = 0
+	 EndStream    EndStreamType = 1
+ )
+
+ func (t EndStreamType) String() string {
+	 switch t {
+	 case NotEndStream:
+		 return "NotEndStream"
+	 case EndStream:
+		 return "EndStream"
+	 }
+	 return "unknown"
+ }
+