Added tests

Author: ryanrolds

projects/gloo/pkg/plugins/tunneling/plugin.go

@@ -1,8 +1,6 @@
 package tunneling
 
 import (
-	"fmt"
-
 	envoy_config_cluster_v3 "github.com/envoyproxy/go-control-plane/envoy/config/cluster/v3"
 	envoy_config_core_v3 "github.com/envoyproxy/go-control-plane/envoy/config/core/v3"
 	envoy_config_endpoint_v3 "github.com/envoyproxy/go-control-plane/envoy/config/endpoint/v3"
@@ -12,7 +10,6 @@ import (
 	"github.com/envoyproxy/go-control-plane/pkg/wellknown"
 	"github.com/golang/protobuf/ptypes/duration"
 	"github.com/golang/protobuf/ptypes/wrappers"
-	"github.com/solo-io/gloo/projects/gloo/constants"
 	v1 "github.com/solo-io/gloo/projects/gloo/pkg/api/v1"
 	"github.com/solo-io/gloo/projects/gloo/pkg/plugins"
 	"github.com/solo-io/gloo/projects/gloo/pkg/translator"
@@ -27,8 +24,10 @@ var (
 )
 
 const (
-	ExtensionName                       = "tunneling"
-	TunnelingAutogeneratedClusterPrefix = constants.SoloGeneratedClusterPrefix + "self_cluster_"
+	ExtensionName                 = "tunneling"
+	OriginalClusterSuffix         = "_original"
+	forwardingListenerPrefix      = "solo_io_generated_self_listener_"
+	forwardingListenerStatsPrefix = "soloioTcpStats"
 )
 
 type plugin struct{}
@@ -44,7 +43,15 @@ func (p *plugin) Name() string {
 func (p *plugin) Init(_ plugins.InitParams) {
 }
 
-// GeneratedResources generates the tunneling resources
+// GeneratedResources scans Upstreams for a tunneling configuration and sets up
+// clusters and listeners to forward traffic to an HTTP CONNECT supporting proxy.
+//
+// HTTP CONNECT tunneling is provided by an Envoy Listener filter. To send traffic to the Listener,
+// we must generate a new forwarding Cluster. The generated Cluster sends traffic over
+// a pipe to the Listener which forwards the traffic to the original Upstream.
+//
+// The SSL configuration for the original cluster is copied to the generated cluster and the
+// supplied proxy SSL configuration is set on the original cluster.
 func (p *plugin) GeneratedResources(params plugins.Params,
 	inClusters []*envoy_config_cluster_v3.Cluster,
 	inEndpoints []*envoy_config_endpoint_v3.ClusterLoadAssignment,
@@ -62,38 +69,15 @@ func (p *plugin) GeneratedResources(params plugins.Params,
 	var newClusters []*envoy_config_cluster_v3.Cluster
 	var newListeners []*envoy_config_listener_v3.Listener
 
-	upstreams := params.Snapshot.Upstreams
-
-	// keep track of clusters we've seen in case of multiple routes to same cluster
+	// track the clusters we have transformed so we don't do it twice
 	processedClusters := sets.Set[string]{}
 
-	// find all the route config that points to upstreams with tunneling
-	// for _, rtConfig := range inRouteConfigurations {
-	// 	for _, vh := range rtConfig.GetVirtualHosts() {
-	// 		for _, rt := range vh.GetRoutes() {
-	// 			rtAction := rt.GetRoute()
-	// 			// we do not handle the weighted cluster or cluster header cases
-	// 			if cluster := rtAction.GetCluster(); cluster != "" {
-	// 				fmt.Printf("candidate cluster for tunneling: %v\n", cluster)
-	// 				var err error
-	// 				newClusters, newListeners, err = setupTunnel(cluster, upstreams, params,
-	// 					inClusters, rtAction, newClusters, newListeners, processedClusters)
-	// 				if err != nil {
-	// 					// return what we have so far, so that any modified input resources can still route
-	// 					fmt.Printf("error setting up tunneling for cluster %v: %v\n", cluster, err)
-	// 					return newClusters, nil, nil, newListeners, nil
-	// 				}
-	// 			}
-	// 		}
-	// 	}
-	// }
-
 	// find all upstreams with tunneling enabled
-	for _, us := range upstreams {
+	for _, us := range params.Snapshot.Upstreams {
 		clusterName := translator.UpstreamToClusterName(us.GetMetadata().Ref())
 
-		// check if this cluster has already been updated, if so move on
 		if processedClusters.Has(clusterName) {
+			logger.Warnf("cluster %v already processed", clusterName)
 			continue
 		}
 
@@ -106,16 +90,15 @@ func (p *plugin) GeneratedResources(params plugins.Params,
 		// find the cluster to update
 		cluster := findClusters(inClusters, clusterName)
 		if cluster == nil {
-			fmt.Printf("cluster %v not found\n", clusterName)
+			logger.Warnf("cluster %v not found when setting up HTTP tunnel", clusterName)
 			continue
 		}
 
 		// change the original cluster name to avoid conflicts with the new cluster
-		newOriginalClusterName := clusterName + "_original"
-		cluster.Name = newOriginalClusterName
+		newOriginalClusterName := clusterName + OriginalClusterSuffix
 
 		// use an in-memory pipe to ourselves (only works on linux)
-		selfPipe := "@/" + clusterName
+		forwardingPipe := "@/" + clusterName
 
 		var originalTransportSocket *envoy_config_core_v3.TransportSocket
 		if cluster.GetTransportSocket() != nil {
@@ -149,195 +132,58 @@ func (p *plugin) GeneratedResources(params plugins.Params,
 			}
 		}
 
-		// generate new cluster and replace the old cluster
-		newCluster := generateSelfCluster(clusterName, selfPipe, originalTransportSocket)
-		newClusters = append(newClusters, newCluster)
+		// generate new cluster with original cluster's name and transport socket that points to
+		// the new listener's pipe
+		newCluster := generateForwardingCluster(clusterName, forwardingPipe, originalTransportSocket)
 
 		tunnelingHeaders := envoyHeadersFromHttpConnectHeaders(us)
 
-		// create the listener with the tunneling configuration
+		// create the listener with the tunneling configuration and point it the original clusters
 		listener, err := generateForwardingTcpListener(clusterName, newOriginalClusterName,
-			selfPipe, httpProxyHostname, tunnelingHeaders)
+			forwardingPipe, httpProxyHostname, tunnelingHeaders)
 		if err != nil {
 			return newClusters, nil, nil, newListeners, err
 		}
 
+		// update the original route's cluster name
+		cluster.Name = newOriginalClusterName
+
+		newClusters = append(newClusters, newCluster)
 		newListeners = append(newListeners, listener)
 
-		// mark this cluster as processed
 		processedClusters.Insert(clusterName)
-
-		// replace the cluster with a new cluster that routes to the generated listener
-
-		// var err error
-		// newClusters, newListeners, err = setupTunnel(cluster, upstreams, params,
-		// 	inClusters, nil, newClusters, newListeners, processedClusters)
-		// if err != nil {
-		// 	// return what we have so far, so that any modified input resources can still route
-		// 	fmt.Printf("error setting up tunneling for cluster %v: %v\n", cluster, err)
-		// 	return newClusters, nil, nil, newListeners, nil
-		// }
 	}
 
 	return newClusters, nil, nil, newListeners, nil
 }
 
-// setupTunnel prepare the tunneling configuration for the given cluster
-func setupTunnel(
-	clusterName string,
-	upstreams v1.UpstreamList,
-	params plugins.Params,
-	inClusters []*envoy_config_cluster_v3.Cluster,
-	rtAction *envoy_config_route_v3.RouteAction,
-	generatedClusters []*envoy_config_cluster_v3.Cluster,
-	generatedListeners []*envoy_config_listener_v3.Listener,
-	processedClusters sets.Set[string],
-) (
-	[]*envoy_config_cluster_v3.Cluster,
-	[]*envoy_config_listener_v3.Listener,
-	error,
-) {
-	ref, err := translator.ClusterToUpstreamRef(clusterName)
-	if err != nil {
-		// return what we have so far, so that any modified input resources can still route
-		// successfully to their generated targets
-		fmt.Printf("error getting upstream ref for cluster %v: %v\n", clusterName, err)
-		return generatedClusters, generatedListeners, nil
-	}
-
-	us, err := upstreams.Find(ref.GetNamespace(), ref.GetName())
-	if err != nil {
-		// return what we have so far, so that any modified input resources can still route
-		// successfully to their generated targets
-		fmt.Printf("error finding upstream %v: %v\n", ref, err)
-		return generatedClusters, generatedListeners, nil
-	}
-
-	// the existence of this value is our indicator that this is a tunneling upstream
-	tunnelingHostname := us.GetHttpProxyHostname().GetValue()
-	if tunnelingHostname == "" {
-		fmt.Print("tunneling hostname is empty\n")
-		return generatedClusters, generatedListeners, nil
-	}
-
-	var tunnelingHeaders []*envoy_config_core_v3.HeaderValueOption
-	for _, header := range us.GetHttpConnectHeaders() {
-		tunnelingHeaders = append(tunnelingHeaders, &envoy_config_core_v3.HeaderValueOption{
-			Header: &envoy_config_core_v3.HeaderValue{
-				Key:   header.GetKey(),
-				Value: header.GetValue(),
-			},
-			Append: &wrappers.BoolValue{Value: false},
-		})
-	}
-
-	selfClusterName := TunnelingAutogeneratedClusterPrefix + clusterName
-	selfPipe := "@/" + clusterName // use an in-memory pipe to ourselves (only works on linux)
-
-	// update the old cluster to route to ourselves first
-	if rtAction != nil {
-		rtAction.ClusterSpecifier = &envoy_config_route_v3.RouteAction_Cluster{Cluster: selfClusterName}
-	}
-
-	// we only want to generate a new encapsulating cluster and pipe to ourselves if we have not done so already
-	if processedClusters.Has(clusterName) {
-		fmt.Printf("cluster %v already processed\n", clusterName)
-		return generatedClusters, generatedListeners, nil
-	}
-
-	// find the cluster and updates it's transport socket to use the generated cluster if needed
-	var originalTransportSocket *envoy_config_core_v3.TransportSocket
-	for _, inCluster := range inClusters {
-		if inCluster.GetName() == clusterName {
-			if inCluster.GetTransportSocket() != nil {
-				tmp := *inCluster.GetTransportSocket()
-				originalTransportSocket = &tmp
-			}
-
-			// we copy the transport socket to the generated cluster.
-			// the generated cluster will use upstream TLS context to leverage TLS origination;
-			// when we encapsulate in HTTP Connect the tcp data being proxied will
-			// be encrypted (thus we don't need the original transport socket metadata here)
-			inCluster.TransportSocket = nil
-			inCluster.TransportSocketMatches = nil
-
-			if us.GetHttpConnectSslConfig() == nil {
-				break
-			}
-
-			// user told us to configure ssl for the http connect proxy
-			cfg, err := utils.NewSslConfigTranslator().ResolveUpstreamSslConfig(params.Snapshot.Secrets,
-				us.GetHttpConnectSslConfig())
-			if err != nil {
-				break
-			}
-
-			typedConfig, err := utils.MessageToAny(cfg)
-			if err != nil {
-				return generatedClusters, generatedListeners, err
-			}
-
-			inCluster.TransportSocket = &envoy_config_core_v3.TransportSocket{
-				Name:       wellknown.TransportSocketTls,
-				ConfigType: &envoy_config_core_v3.TransportSocket_TypedConfig{TypedConfig: typedConfig},
-			}
-
-			break
-		}
-	}
-
-	generatedClusters = append(generatedClusters, generateSelfCluster(selfClusterName,
-		selfPipe, originalTransportSocket))
-
-	forwardingTcpListener, err := generateForwardingTcpListener(clusterName, clusterName, selfPipe,
-		tunnelingHostname, tunnelingHeaders)
-	if err != nil {
-		return generatedClusters, generatedListeners, err
-	}
-
-	generatedListeners = append(generatedListeners, forwardingTcpListener)
-
-	processedClusters.Insert(clusterName)
-
-	return generatedClusters, generatedListeners, nil
-}
-
-// the initial route is updated to route to this generated cluster, which routes envoy back to itself (to the
-// generated TCP listener, which forwards to the original destination)
-//
-// the purpose of doing this is to allow both the HTTP Connection Manager filter and TCP filter to run.
-// the HTTP Connection Manager runs to allow route-level matching on HTTP parameters (such as request path),
-// but then we forward the bytes as raw TCP to the HTTP Connect proxy (which can only be done on a TCP listener)
-func generateSelfCluster(
-	selfClusterName,
-	selfPipe string,
+// generateForwardingCluster generates a cluster will replace the original cluster and send
+// the traffic to the generated listener
+func generateForwardingCluster(
+	clusterName,
+	pipePath string,
 	originalTransportSocket *envoy_config_core_v3.TransportSocket,
 ) *envoy_config_cluster_v3.Cluster {
 	return &envoy_config_cluster_v3.Cluster{
 		ClusterDiscoveryType: &envoy_config_cluster_v3.Cluster_Type{
 			Type: envoy_config_cluster_v3.Cluster_STATIC,
 		},
 		ConnectTimeout:  &duration.Duration{Seconds: 5},
-		Name:            selfClusterName,
+		Name:            clusterName,
 		TransportSocket: originalTransportSocket,
-		LoadAssignment:  selfPipeLoadAssignment(selfClusterName, selfPipe),
-	}
-}
-
-// selfPipeLoadAssignment returns a load assignment for the pipe
-func selfPipeLoadAssignment(clusterName, pipeName string) *envoy_config_endpoint_v3.ClusterLoadAssignment {
-	return &envoy_config_endpoint_v3.ClusterLoadAssignment{
-		ClusterName: clusterName,
-		Endpoints: []*envoy_config_endpoint_v3.LocalityLbEndpoints{
-			{
-				LbEndpoints: []*envoy_config_endpoint_v3.LbEndpoint{
-					{
-						HostIdentifier: &envoy_config_endpoint_v3.LbEndpoint_Endpoint{
-							Endpoint: &envoy_config_endpoint_v3.Endpoint{
-								Address: &envoy_config_core_v3.Address{
-									Address: &envoy_config_core_v3.Address_Pipe{
-										Pipe: &envoy_config_core_v3.Pipe{
-											Path: pipeName,
+		LoadAssignment: &envoy_config_endpoint_v3.ClusterLoadAssignment{
+			ClusterName: clusterName,
+			Endpoints: []*envoy_config_endpoint_v3.LocalityLbEndpoints{
+				{
+					LbEndpoints: []*envoy_config_endpoint_v3.LbEndpoint{
+						{
+							HostIdentifier: &envoy_config_endpoint_v3.LbEndpoint_Endpoint{
+								Endpoint: &envoy_config_endpoint_v3.Endpoint{
+									Address: &envoy_config_core_v3.Address{
+										Address: &envoy_config_core_v3.Address_Pipe{
+											Pipe: &envoy_config_core_v3.Pipe{
+												Path: pipePath,
+											},
 										},
 									},
 								},
@@ -350,16 +196,17 @@ func selfPipeLoadAssignment(clusterName, pipeName string) *envoy_config_endpoint
 	}
 }
 
-// the generated cluster routes to this generated listener, which forwards TCP traffic to an HTTP Connect proxy
+// generateForwardingTcpListener generates a listener that will forwards traffic to the
+// HTTP Connect proxy
 func generateForwardingTcpListener(
 	cluster,
 	originalCluster,
-	selfPipePath,
+	pipePath,
 	tunnelingHostname string,
 	tunnelingHeadersToAdd []*envoy_config_core_v3.HeaderValueOption,
 ) (*envoy_config_listener_v3.Listener, error) {
 	cfg := &envoytcp.TcpProxy{
-		StatPrefix: "soloioTcpStats" + cluster,
+		StatPrefix: forwardingListenerStatsPrefix + cluster,
 		TunnelingConfig: &envoytcp.TcpProxy_TunnelingConfig{Hostname: tunnelingHostname,
 			HeadersToAdd: tunnelingHeadersToAdd},
 		ClusterSpecifier: &envoytcp.TcpProxy_Cluster{Cluster: originalCluster}, // route to original target
@@ -370,11 +217,11 @@ func generateForwardingTcpListener(
 	}
 
 	return &envoy_config_listener_v3.Listener{
-		Name: "solo_io_generated_self_listener_" + cluster,
+		Name: forwardingListenerPrefix + cluster,
 		Address: &envoy_config_core_v3.Address{
 			Address: &envoy_config_core_v3.Address_Pipe{
 				Pipe: &envoy_config_core_v3.Pipe{
-					Path: selfPipePath,
+					Path: pipePath,
 				},
 			},
 		},
@@ -393,6 +240,7 @@ func generateForwardingTcpListener(
 	}, nil
 }
 
+// envoyHeadersFromHttpConnectHeaders converts the http connect headers to envoy headers
 func envoyHeadersFromHttpConnectHeaders(us *v1.Upstream) []*envoy_config_core_v3.HeaderValueOption {
 	var tunnelingHeaders []*envoy_config_core_v3.HeaderValueOption
 	for _, header := range us.GetHttpConnectHeaders() {