Istio Canary TCP service support

Signed-off-by: Alexey Kubrinsky <akubrinsky@zetaglobal.com>

Makefile

@@ -29,6 +29,11 @@ test-codegen:
 test: test-fmt test-codegen
 	go test ./...
 
+test-coverage: test-fmt test-codegen
+	go test -coverprofile cover.out ./...
+	go tool cover -html=cover.out
+	rm cover.out
+
 crd:
 	cat artifacts/flagger/crd.yaml > charts/flagger/crds/crd.yaml
 	cat artifacts/flagger/crd.yaml > kustomize/base/flagger/crd.yaml

docs/gitbook/tutorials/istio-progressive-delivery.md

@@ -480,3 +480,63 @@ With the above configuration, Flagger will run a canary release with the followi
 
 The above procedure can be extended with [custom metrics](../usage/metrics.md) checks, [webhooks](../usage/webhooks.md), [manual promotion](../usage/webhooks.md#manual-gating) approval and [Slack or MS Teams](../usage/alerting.md) notifications.
 
+
+## Canary Deployments for TCP Services
+
+Performing a Canary deployment on a TCP (non HTTP) service is nearly identical to an HTTP Canary. Besides updating your `Gateway` document to support the `TCP` routing, the only difference is you have to set the `appProtocol` field to `TCP` inside of the `service` section of your `Canary` document.
+
+#### Example:
+
+```yaml
+apiVersion: networking.istio.io/v1alpha3
+kind: Gateway
+metadata:
+  name: public-gateway
+  namespace: istio-system
+spec:
+  selector:
+    istio: ingressgateway
+  servers:
+    - port:
+        number: 7070
+        name: tcp-service
+        protocol: TCP # <== set the protocol to tcp here
+      hosts:
+        - "*"
+```
+
+```yaml
+apiVersion: flagger.app/v1beta1
+kind: Canary
+...
+...
+service:
+  port: 7070
+  appProtocol: TCP # <== set the appProtocol here
+  targetPort: 7070
+  portName: "tcp-service-port"
+...
+...
+```
+
+If the `appProtocol` equals `TCP` then Flagger will treat this as a Canary deployment for a `TCP` service. When it creates the `VirtualService` document it will add a `TCP` section to route requests between the `primary` and `canary` services. See Istio documentation for more information on this [spec](https://istio.io/latest/docs/reference/config/networking/virtual-service/#TCPRoute).
+
+The resulting `VirtualService` will include a `tcp` section similar to what is shown below:
+```yaml
+tcp:
+  - route:
+    - destination:
+        host: tcp-service-primary
+        port:
+          number: 7070
+      weight: 100
+    - destination:
+        host: tcp-service-canary
+        port:
+          number: 7070
+      weight: 0
+```
+
+Once the Canary analysis begins, Flagger will be able to adjust the weights inside of this `tcp` section to advance the Canary deployment until it either runs into an error (and is halted) or it successfully reaches the end of the analysis and is Promoted.
+
+It is also important to note that if you set `appProtocol` to anything other than `TCP`, for example if you set it to `HTTP`, it will perform the Canary and treat it as an `HTTP` service. The same remains true if you do not set `appProtocol` at all. It will __ONLY__ treat a Canary as a `TCP` service if `appProtocal` equals `TCP`.

pkg/apis/istio/v1alpha3/virtual_service.go

@@ -597,7 +597,7 @@ type TCPRoute struct {
 	// Currently, only one destination is allowed for TCP services. When TCP
 	// weighted routing support is introduced in Envoy, multiple destinations
 	// with weights can be specified.
-	Route HTTPRouteDestination `json:"route"`
+	Route []HTTPRouteDestination `json:"route"`
 }
 
 // L4 connection match attributes. Note that L4 connection matching support

pkg/apis/istio/v1alpha3/zz_generated.deepcopy.go

@@ -848,7 +848,13 @@ func (in *TCPRoute) DeepCopyInto(out *TCPRoute) {
 			(*in)[i].DeepCopyInto(&(*out)[i])
 		}
 	}
-	in.Route.DeepCopyInto(&out.Route)
+	if in.Route != nil {
+		in, out := &in.Route, &out.Route
+		*out = make([]HTTPRouteDestination, len(*in))
+		for i := range *in {
+			(*in)[i].DeepCopyInto(&(*out)[i])
+		}
+	}
 	return
 }
 

pkg/router/istio.go

@@ -125,6 +125,23 @@ func (ir *IstioRouter) reconcileDestinationRule(canary *flaggerv1.Canary, name s
 	return nil
 }
 
+// return true if canary service has appProtocol == tcp
+func isTcp(canary *flaggerv1.Canary) bool {
+	return strings.ToLower(canary.Spec.Service.AppProtocol) == "tcp"
+}
+
+// map canary.spec.service.match into L4Match
+func canaryToL4Match(canary *flaggerv1.Canary) []istiov1alpha3.L4MatchAttributes {
+	var match []istiov1alpha3.L4MatchAttributes
+	for _, m := range canary.Spec.Service.Match {
+		match = append(match, istiov1alpha3.L4MatchAttributes{
+			Port: int(m.Port),
+		})
+	}
+
+	return match
+}
+
 func (ir *IstioRouter) reconcileVirtualService(canary *flaggerv1.Canary) error {
 	apexName, primaryName, canaryName := canary.GetServiceNames()
 
@@ -175,7 +192,21 @@ func (ir *IstioRouter) reconcileVirtualService(canary *flaggerv1.Canary) error {
 		gateways = []string{}
 	}
 
-	newSpec := istiov1alpha3.VirtualServiceSpec{
+	var newSpec istiov1alpha3.VirtualServiceSpec
+
+	if isTcp(canary) {
+		newSpec = istiov1alpha3.VirtualServiceSpec{
+			Hosts:    hosts,
+			Gateways: gateways,
+			Tcp: []istiov1alpha3.TCPRoute{
+				{
+					Match: canaryToL4Match(canary),
+					Route: canaryRoute,
+				},
+			},
+		}
+	} else {
+		newSpec = istiov1alpha3.VirtualServiceSpec{
 			Hosts:    hosts,
 			Gateways: gateways,
 			Http: []istiov1alpha3.HTTPRoute{
@@ -190,6 +221,7 @@ func (ir *IstioRouter) reconcileVirtualService(canary *flaggerv1.Canary) error {
 				},
 			},
 		}
+	}
 
 	newMetadata := canary.Spec.Service.Apex
 	if newMetadata == nil {
@@ -203,7 +235,7 @@ func (ir *IstioRouter) reconcileVirtualService(canary *flaggerv1.Canary) error {
 	}
 	newMetadata.Annotations = filterMetadata(newMetadata.Annotations)
 
-	if len(canary.GetAnalysis().Match) > 0 {
+	if !isTcp(canary) && len(canary.GetAnalysis().Match) > 0 {
 		canaryMatch := mergeMatchConditions(canary.GetAnalysis().Match, canary.Spec.Service.Match)
 		newSpec.Http = []istiov1alpha3.HTTPRoute{
 			{
@@ -349,6 +381,38 @@ func (ir *IstioRouter) GetRoutes(canary *flaggerv1.Canary) (
 		return
 	}
 
+	if isTcp(canary) {
+		ir.logger.Infof("Canary %s.%s uses TCP service", canary.Name, canary.Namespace)
+		var tcpRoute istiov1alpha3.TCPRoute
+		for _, tcp := range vs.Spec.Tcp {
+			for _, r := range tcp.Route {
+				if r.Destination.Host == canaryName {
+					tcpRoute = tcp
+					break
+				}
+			}
+		}
+		for _, route := range tcpRoute.Route {
+			if route.Destination.Host == primaryName {
+				primaryWeight = route.Weight
+			}
+			if route.Destination.Host == canaryName {
+				canaryWeight = route.Weight
+			}
+		}
+
+		mirrored = false
+
+		if primaryWeight == 0 && canaryWeight == 0 {
+			err = fmt.Errorf("VirtualService %s.%s does not contain routes for %s-primary and %s-canary",
+				apexName, canary.Namespace, apexName, apexName)
+		}
+
+		return
+	}
+
+	ir.logger.Infof("Canary %s.%s uses HTTP service", canary.Name, canary.Namespace)
+
 	var httpRoute istiov1alpha3.HTTPRoute
 	for _, http := range vs.Spec.Http {
 		for _, r := range http.Route {
@@ -412,6 +476,26 @@ func (ir *IstioRouter) SetRoutes(
 
 	vsCopy := vs.DeepCopy()
 
+	if isTcp(canary) {
+		// weighted routing (progressive canary)
+		weightedRoute := istiov1alpha3.TCPRoute{
+			Match: canaryToL4Match(canary),
+			Route: []istiov1alpha3.HTTPRouteDestination{
+				makeDestination(canary, primaryName, primaryWeight),
+				makeDestination(canary, canaryName, canaryWeight),
+			},
+		}
+		vsCopy.Spec.Tcp = []istiov1alpha3.TCPRoute{
+			weightedRoute,
+		}
+
+		vs, err = ir.istioClient.NetworkingV1alpha3().VirtualServices(canary.Namespace).Update(context.TODO(), vsCopy, metav1.UpdateOptions{})
+		if err != nil {
+			return fmt.Errorf("VirtualService %s.%s update failed: %w", apexName, canary.Namespace, err)
+		}
+		return nil
+	}
+
 	// weighted routing (progressive canary)
 	weightedRoute := istiov1alpha3.HTTPRoute{
 		Match:      canary.Spec.Service.Match,

pkg/router/istio_test.go

@@ -29,10 +29,51 @@ import (
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 
 	"github.com/fluxcd/flagger/pkg/apis/flagger/v1beta1"
+	flaggerv1 "github.com/fluxcd/flagger/pkg/apis/flagger/v1beta1"
 	istiov1alpha1 "github.com/fluxcd/flagger/pkg/apis/istio/common/v1alpha1"
 	istiov1alpha3 "github.com/fluxcd/flagger/pkg/apis/istio/v1alpha3"
 )
 
+func TestUnmarshalVirtualService(t *testing.T) {
+	body := `
+	{
+		"apiVersion": "networking.istio.io/v1beta1",
+		"kind": "VirtualService",
+		"spec": {
+			"gateways": [
+				"default/gateway"
+			],
+			"hosts": [
+				"my.example.com"
+			],
+			"tcp": [
+				{
+					"match": [
+						{
+							"port": 9898
+						}
+					],
+					"route": [
+						{
+							"destination": {
+								"host": "my.example.com",
+								"port": {
+									"number": 9898
+								}
+							}
+						}
+					]
+				}
+			]
+		}
+	}
+	`
+
+	var vs istiov1alpha3.VirtualService
+	err := json.Unmarshal([]byte(body), &vs)
+	require.NoError(t, err)
+}
+
 func TestIstioRouter_Sync(t *testing.T) {
 	mocks := newFixture(nil)
 	router := &IstioRouter{
@@ -727,3 +768,124 @@ func TestIstioRouter_Match(t *testing.T) {
 	assert.Len(t, vs.Spec.Http[1].Match, 1) // check for abtest-primary
 	require.Equal(t, vs.Spec.Http[1].Match[0].Uri.Prefix, "/podinfo")
 }
+
+// TCP Canary
+func newTestCanaryTCP() *flaggerv1.Canary {
+	cd := &flaggerv1.Canary{
+		TypeMeta: metav1.TypeMeta{APIVersion: flaggerv1.SchemeGroupVersion.String()},
+		ObjectMeta: metav1.ObjectMeta{
+			Namespace: "default",
+			Name:      "podinfo",
+		},
+		Spec: flaggerv1.CanarySpec{
+			TargetRef: flaggerv1.LocalObjectReference{
+				Name:       "podinfo",
+				APIVersion: "apps/v1",
+				Kind:       "Deployment",
+			},
+			Service: flaggerv1.CanaryService{
+				Port:          9898,
+				PortDiscovery: true,
+				AppProtocol:   "TCP",
+				Match: []istiov1alpha3.HTTPMatchRequest{
+					{
+						Port: 9898,
+					},
+				},
+				Gateways: []string{
+					"istio/public-gateway",
+					"mesh",
+				},
+			}, Analysis: &flaggerv1.CanaryAnalysis{
+				Threshold:  10,
+				StepWeight: 10,
+				MaxWeight:  50,
+				Metrics: []flaggerv1.CanaryMetric{
+					{
+						Name:      "request-success-rate",
+						Threshold: 99,
+						Interval:  "1m",
+					},
+					{
+						Name:      "request-duration",
+						Threshold: 500,
+						Interval:  "1m",
+					},
+				},
+			},
+		},
+	}
+	return cd
+}
+
+func TestIstioRouter_SetRoutesTCP(t *testing.T) {
+	mocks := newFixture(newTestCanaryTCP())
+	router := &IstioRouter{
+		logger:        mocks.logger,
+		flaggerClient: mocks.flaggerClient,
+		istioClient:   mocks.meshClient,
+		kubeClient:    mocks.kubeClient,
+	}
+
+	err := router.Reconcile(mocks.canary)
+	require.NoError(t, err)
+
+	pHost := fmt.Sprintf("%s-primary", mocks.canary.Spec.TargetRef.Name)
+	cHost := fmt.Sprintf("%s-canary", mocks.canary.Spec.TargetRef.Name)
+
+	t.Run("normal", func(t *testing.T) {
+		p, c := 60, 40
+		err := router.SetRoutes(mocks.canary, p, c, false)
+		require.NoError(t, err)
+
+		vs, err := mocks.meshClient.NetworkingV1alpha3().VirtualServices("default").Get(context.TODO(), "podinfo", metav1.GetOptions{})
+		require.NoError(t, err)
+
+		var pRoute, cRoute istiov1alpha3.HTTPRouteDestination
+		for _, tcp := range vs.Spec.Tcp {
+			for _, route := range tcp.Route {
+				if route.Destination.Host == pHost {
+					pRoute = route
+				}
+				if route.Destination.Host == cHost {
+					cRoute = route
+				}
+			}
+		}
+
+		assert.Equal(t, p, pRoute.Weight)
+		assert.Equal(t, c, cRoute.Weight)
+	})
+}
+
+func TestIstioRouter_GetRoutesTCP(t *testing.T) {
+	mocks := newFixture(newTestCanaryTCP())
+	router := &IstioRouter{
+		logger:        mocks.logger,
+		flaggerClient: mocks.flaggerClient,
+		istioClient:   mocks.meshClient,
+		kubeClient:    mocks.kubeClient,
+	}
+
+	err := router.Reconcile(mocks.canary)
+	require.NoError(t, err)
+
+	p, c, m, err := router.GetRoutes(mocks.canary)
+	require.NoError(t, err)
+	assert.Equal(t, 100, p)
+	assert.Equal(t, 0, c)
+	assert.False(t, m)
+
+	mocks.canary = newTestMirror()
+
+	err = router.Reconcile(mocks.canary)
+	require.NoError(t, err)
+
+	p, c, m, err = router.GetRoutes(mocks.canary)
+	require.NoError(t, err)
+	assert.Equal(t, 100, p)
+	assert.Equal(t, 0, c)
+
+	// A TCP Canary resource has mirroring disabled
+	assert.False(t, m)
+}