introduce connectTimeout

in4it · Feb 24, 2022 · dac1a45 · dac1a45
1 parent f5bbdf3
commit dac1a45
Show file tree

Hide file tree

Showing 6 changed files with 103 additions and 2 deletions.
diff --git a/pkg/api/rule.go b/pkg/api/rule.go
@@ -37,6 +37,7 @@ type RuleActionsProxy struct {
 	EnableWebsockets bool                    `json:"enableWebsockets" yaml:"enableWebsockets"`
 	PrefixRewrite    string                  `json:"prefixRewrite" yaml:"prefixRewrite"`
 	RegexRewrite     RuleActionsRegexRewrite `json:"regexRewrite" yaml:"regexRewrite"`
+	ConnectTimeout   int64                   `json:"connectTimeout" yaml:"connectTimeout"`
 }
 type RuleActionsRegexRewrite struct {
 	Regex        string `json:"regex" yaml:"regex"`

diff --git a/pkg/envoy/cluster.go b/pkg/envoy/cluster.go
@@ -77,6 +77,10 @@ func (c *Cluster) createCluster(params ClusterParams) *api.Cluster {
 
 	connectTimeout := 2 * time.Second
 
+	if params.ConnectTimeout > 0 {
+		connectTimeout = time.Duration(params.ConnectTimeout) * time.Second
+	}
+
 	// add healthchecks
 	healthChecks := []*core.HealthCheck{}
 	if params.HealthCheck.HTTPHealthCheck.Path != "" {

diff --git a/pkg/envoy/testdata/test-cluster-connection-timeout.yaml b/pkg/envoy/testdata/test-cluster-connection-timeout.yaml
@@ -0,0 +1,13 @@
+api: proxy.in4it.io/v1
+kind: rule
+metadata:
+  name: test-cluster
+spec:
+  conditions:
+    - hostname: test.example.com
+      path: /test
+  actions:
+    - proxy:
+        hostname: target-example-1.com
+        port: 443
+        connectTimeout: 5
diff --git a/pkg/envoy/types.go b/pkg/envoy/types.go
@@ -48,6 +48,7 @@ type ClusterParams struct {
 	Port           int64
 	HTTP2          bool
 	HealthCheck    HealthCheck
+	ConnectTimeout int64
 }
 type ListenerParams struct {
 	Name             string
@@ -111,6 +112,7 @@ type ActionProxy struct {
 	EnableWebsockets bool
 	PrefixRewrite    string
 	RegexRewrite     RegexRewrite
+	ConnectTimeout   int64
 }
 
 type RegexRewrite struct {

diff --git a/pkg/envoy/xds.go b/pkg/envoy/xds.go
@@ -564,6 +564,7 @@ func (x *XDS) getAction(ruleName string, actions []pkgApi.RuleActions) Action {
 			action.RuleName = ruleName
 			action.Proxy.TargetHostname = ruleAction.Proxy.Hostname
 			action.Proxy.Port = ruleAction.Proxy.Port
+			action.Proxy.ConnectTimeout = ruleAction.Proxy.ConnectTimeout
 			if ruleAction.Proxy.HealthCheck.HTTPHealthCheck.Path != "" {
 				action.Proxy.HealthCheck.HTTPHealthCheck.Path = ruleAction.Proxy.HealthCheck.HTTPHealthCheck.Path
 				action.Proxy.HealthCheck.Timeout = ruleAction.Proxy.HealthCheck.Timeout
@@ -633,6 +634,7 @@ func (x *XDS) getClusterParams(action Action) ClusterParams {
 		TargetHostname: action.Proxy.TargetHostname,
 		Port:           action.Proxy.Port,
 		HealthCheck:    action.Proxy.HealthCheck,
+		ConnectTimeout: action.Proxy.ConnectTimeout,
 	}
 }
 func (x *XDS) getAuthParams(jwtProviderName string, jwtProvider pkgApi.JwtProvider) Auth {

diff --git a/pkg/envoy/xds_test.go b/pkg/envoy/xds_test.go
@@ -1262,9 +1262,11 @@ func TestLuaFilterWithMTLS(t *testing.T) {
 			return
 		}
 	}
+	found := false
 	for listenerKey := range x.workQueue.cache.listeners {
 		ll := x.workQueue.cache.listeners[listenerKey].(*api.Listener)
 		if ll.GetName() != "l_http" {
+			found = true
 			manager, err := getListenerHTTPConnectionManager(ll)
 			if err != nil {
 				t.Errorf("getListenerHTTPConnectionManager error: %s", err)
@@ -1277,6 +1279,9 @@ func TestLuaFilterWithMTLS(t *testing.T) {
 		}
 
 	}
+	if found == false {
+		t.Errorf("listener not found")
+	}
 }
 func TestLuaFilterWithMTLS2(t *testing.T) {
 	logger.SetLogLevel(loggo.DEBUG)
@@ -1299,9 +1304,11 @@ func TestLuaFilterWithMTLS2(t *testing.T) {
 			return
 		}
 	}
+	found := false
 	for listenerKey := range x.workQueue.cache.listeners {
 		ll := x.workQueue.cache.listeners[listenerKey].(*api.Listener)
 		if ll.GetName() != "l_http" {
+			found = true
 			manager, err := getListenerHTTPConnectionManager(ll)
 			if err != nil {
 				t.Errorf("getListenerHTTPConnectionManager error: %s", err)
@@ -1312,9 +1319,52 @@ func TestLuaFilterWithMTLS2(t *testing.T) {
 				return
 			}
 		}
+	}
+	if found == false {
+		t.Errorf("listener not found")
+	}
+}
 
+func TestLuaFilterWithMultipleListeners(t *testing.T) {
+	logger.SetLogLevel(loggo.DEBUG)
+	s, err := initStorage()
+	if err != nil {
+		t.Errorf("Couldn't initialize storage: %s", err)
+		return
+	}
+	x := NewXDS(s, "", "")
+	ObjectFileNames := []string{"test-mtls.yaml", "test-cluster-1.yaml", "test-luafilter.yaml"}
+	for _, filename := range ObjectFileNames {
+		newItems, err := x.putObject(filename)
+		if err != nil {
+			t.Errorf("PutObject failed: %s", err)
+			return
+		}
+		_, err = x.workQueue.Submit(newItems)
+		if err != nil {
+			t.Errorf("WorkQueue error: %s", err)
+			return
+		}
+	}
+	found := 0
+	for listenerKey := range x.workQueue.cache.listeners {
+		ll := x.workQueue.cache.listeners[listenerKey].(*api.Listener)
+		found++
+		manager, err := getListenerHTTPConnectionManager(ll)
+		if err != nil {
+			t.Errorf("getListenerHTTPConnectionManager error: %s", err)
+			return
+		}
+		if getListenerHTTPFilterIndex("envoy.filters.http.lua", manager.HttpFilters) == -1 {
+			t.Errorf("envoy.filters.http.lua not found in httprouter filter - should be found")
+			return
+		}
+	}
+	if found != 2 {
+		t.Errorf("Not all listeners found")
 	}
 }
+
 func TestRuleWithNoConditions(t *testing.T) {
 	logger.SetLogLevel(loggo.DEBUG)
 	s, err := initStorage()
@@ -1337,8 +1387,37 @@ func TestRuleWithNoConditions(t *testing.T) {
 		}
 	}
 	allClusters := x.workQueue.cache.clusters
-	for _, v := range allClusters {
-		fmt.Printf("%+v", v.(*clusterAPI.Cluster))
+	if len(allClusters) != 1 {
+		t.Errorf("Expected to have a 1 cluster (got %d)", len(allClusters))
 	}
 
 }
+func TestRuleWithConnectionTimeout(t *testing.T) {
+	logger.SetLogLevel(loggo.DEBUG)
+	s, err := initStorage()
+	if err != nil {
+		t.Errorf("Couldn't initialize storage: %s", err)
+		return
+	}
+	x := NewXDS(s, "", "")
+	ObjectFileNames := []string{"test-cluster-connection-timeout.yaml"}
+	for _, filename := range ObjectFileNames {
+		newItems, err := x.putObject(filename)
+		if err != nil {
+			t.Errorf("PutObject failed: %s", err)
+			return
+		}
+		_, err = x.workQueue.Submit(newItems)
+		if err != nil {
+			t.Errorf("WorkQueue error: %s", err)
+			return
+		}
+	}
+	allClusters := x.workQueue.cache.clusters
+	for _, v := range allClusters {
+		cluster := v.(*clusterAPI.Cluster)
+		if cluster.ConnectTimeout.Seconds != 5 {
+			t.Errorf("Cluster Connect timeout is not 5 (got %d)", cluster.ConnectTimeout.Seconds)
+		}
+	}
+}