Merge pull request #311 from andrewjjenkins/mirror

Add traffic mirroring for Istio service mesh
2026-04-15 06:57:34 +00:00 · 2019-10-05 10:34:25 +03:00
parent 0c60cf39f8 a21e53fa31
commit 9a9baadf0e
22 changed files with 430 additions and 74 deletions
@@ -42,11 +42,9 @@ type Mocks struct {
 	router        router.Interface
 }

-func SetupMocks(abtest bool) Mocks {
-	// init canary
-	c := newTestCanary()
-	if abtest {
-		c = newTestCanaryAB()
+func SetupMocks(c *v1alpha3.Canary) Mocks {
+	if c == nil {
+		c = newTestCanary()
 	}
 	flaggerClient := fakeFlagger.NewSimpleClientset(c)

@@ -269,6 +267,12 @@ func newTestCanary() *v1alpha3.Canary {
 	return cd
 }

+func newTestCanaryMirror() *v1alpha3.Canary {
+	cd := newTestCanary()
+	cd.Spec.CanaryAnalysis.Mirror = true
+	return cd
+}
+
 func newTestCanaryAB() *v1alpha3.Canary {
 	cd := &v1alpha3.Canary{
 		TypeMeta: metav1.TypeMeta{APIVersion: v1alpha3.SchemeGroupVersion.String()},
@@ -155,7 +155,7 @@ func (c *Controller) advanceCanary(name string, namespace string, skipLivenessCh

 	// check if virtual service exists
 	// and if it contains weighted destination routes to the primary and canary services
-	primaryWeight, canaryWeight, err := meshRouter.GetRoutes(cd)
+	primaryWeight, canaryWeight, mirrored, err := meshRouter.GetRoutes(cd)
 	if err != nil {
 		c.recordEventWarningf(cd, "%v", err)
 		return
@@ -176,7 +176,7 @@ func (c *Controller) advanceCanary(name string, namespace string, skipLivenessCh
 		// route all traffic back to primary
 		primaryWeight = 100
 		canaryWeight = 0
-		if err := meshRouter.SetRoutes(cd, primaryWeight, canaryWeight); err != nil {
+		if err := meshRouter.SetRoutes(cd, primaryWeight, canaryWeight, false); err != nil {
 			c.recordEventWarningf(cd, "%v", err)
 			return
 		}
@@ -218,7 +218,7 @@ func (c *Controller) advanceCanary(name string, namespace string, skipLivenessCh
 	if cd.Status.Phase == flaggerv1.CanaryPhasePromoting {
 		if provider != "kubernetes" {
 			c.recordEventInfof(cd, "Routing all traffic to primary")
-			if err := meshRouter.SetRoutes(cd, 100, 0); err != nil {
+			if err := meshRouter.SetRoutes(cd, 100, 0, false); err != nil {
 				c.recordEventWarningf(cd, "%v", err)
 				return
 			}
@@ -275,7 +275,7 @@ func (c *Controller) advanceCanary(name string, namespace string, skipLivenessCh
 		// route all traffic back to primary
 		primaryWeight = 100
 		canaryWeight = 0
-		if err := meshRouter.SetRoutes(cd, primaryWeight, canaryWeight); err != nil {
+		if err := meshRouter.SetRoutes(cd, primaryWeight, canaryWeight, false); err != nil {
 			c.recordEventWarningf(cd, "%v", err)
 			return
 		}
@@ -302,8 +302,9 @@ func (c *Controller) advanceCanary(name string, namespace string, skipLivenessCh
 	}

 	// check if the canary success rate is above the threshold
-	// skip check if no traffic is routed to canary
-	if canaryWeight == 0 && cd.Status.Iterations == 0 {
+	// skip check if no traffic is routed or mirrored to canary
+	if canaryWeight == 0 && cd.Status.Iterations == 0 &&
+		(cd.Spec.CanaryAnalysis.Mirror == false || mirrored == false) {
 		c.recordEventInfof(cd, "Starting canary analysis for %s.%s", cd.Spec.TargetRef.Name, cd.Namespace)

 		// run pre-rollout web hooks
@@ -328,7 +329,7 @@ func (c *Controller) advanceCanary(name string, namespace string, skipLivenessCh
 	if len(cd.Spec.CanaryAnalysis.Match) > 0 && cd.Spec.CanaryAnalysis.Iterations > 0 {
 		// route traffic to canary and increment iterations
 		if cd.Spec.CanaryAnalysis.Iterations > cd.Status.Iterations {
-			if err := meshRouter.SetRoutes(cd, 0, 100); err != nil {
+			if err := meshRouter.SetRoutes(cd, 0, 100, false); err != nil {
 				c.recordEventWarningf(cd, "%v", err)
 				return
 			}
@@ -372,6 +373,15 @@ func (c *Controller) advanceCanary(name string, namespace string, skipLivenessCh
 	if cd.Spec.CanaryAnalysis.Iterations > 0 {
 		// increment iterations
 		if cd.Spec.CanaryAnalysis.Iterations > cd.Status.Iterations {
+			// If in "mirror" mode, mirror requests during the entire B/G canary test
+			if provider != "kubernetes" &&
+				cd.Spec.CanaryAnalysis.Mirror == true && mirrored == false {
+				if err := meshRouter.SetRoutes(cd, 100, 0, true); err != nil {
+					c.recordEventWarningf(cd, "%v", err)
+				}
+				c.logger.With("canary", fmt.Sprintf("%s.%s", name, namespace)).
+					Infof("Enabling mirroring for Blue/Green")
+			}
 			if err := c.deployer.SetStatusIterations(cd, cd.Status.Iterations+1); err != nil {
 				c.recordEventWarningf(cd, "%v", err)
 				return
@@ -390,7 +400,7 @@ func (c *Controller) advanceCanary(name string, namespace string, skipLivenessCh
 		if cd.Spec.CanaryAnalysis.Iterations == cd.Status.Iterations {
 			if provider != "kubernetes" {
 				c.recordEventInfof(cd, "Routing all traffic to canary")
-				if err := meshRouter.SetRoutes(cd, 0, 100); err != nil {
+				if err := meshRouter.SetRoutes(cd, 0, 100, false); err != nil {
 					c.recordEventWarningf(cd, "%v", err)
 					return
 				}
@@ -429,16 +439,34 @@ func (c *Controller) advanceCanary(name string, namespace string, skipLivenessCh
 	if cd.Spec.CanaryAnalysis.StepWeight > 0 {
 		// increase traffic weight
 		if canaryWeight < maxWeight {
-			primaryWeight -= cd.Spec.CanaryAnalysis.StepWeight
-			if primaryWeight < 0 {
-				primaryWeight = 0
-			}
-			canaryWeight += cd.Spec.CanaryAnalysis.StepWeight
-			if canaryWeight > 100 {
-				canaryWeight = 100
+			// If in "mirror" mode, do one step of mirroring before shifting traffic to canary.
+			// When mirroring, all requests go to primary and canary, but only responses from
+			// primary go back to the user.
+			if cd.Spec.CanaryAnalysis.Mirror && canaryWeight == 0 {
+				if mirrored == false {
+					mirrored = true
+					primaryWeight = 100
+					canaryWeight = 0
+				} else {
+					mirrored = false
+					primaryWeight = 100 - cd.Spec.CanaryAnalysis.StepWeight
+					canaryWeight = cd.Spec.CanaryAnalysis.StepWeight
+				}
+				c.logger.With("canary", fmt.Sprintf("%s.%s", name, namespace)).
+					Infof("Running mirror step %d/%d/%t", primaryWeight, canaryWeight, mirrored)
+			} else {
+
+				primaryWeight -= cd.Spec.CanaryAnalysis.StepWeight
+				if primaryWeight < 0 {
+					primaryWeight = 0
+				}
+				canaryWeight += cd.Spec.CanaryAnalysis.StepWeight
+				if canaryWeight > 100 {
+					canaryWeight = 100
+				}
 			}

-			if err := meshRouter.SetRoutes(cd, primaryWeight, canaryWeight); err != nil {
+			if err := meshRouter.SetRoutes(cd, primaryWeight, canaryWeight, mirrored); err != nil {
 				c.recordEventWarningf(cd, "%v", err)
 				return
 			}
@@ -489,7 +517,7 @@ func (c *Controller) shouldSkipAnalysis(cd *flaggerv1.Canary, meshRouter router.
 	// route all traffic to primary
 	primaryWeight = 100
 	canaryWeight = 0
-	if err := meshRouter.SetRoutes(cd, primaryWeight, canaryWeight); err != nil {
+	if err := meshRouter.SetRoutes(cd, primaryWeight, canaryWeight, false); err != nil {
 		c.recordEventWarningf(cd, "%v", err)
 		return false
 	}
@@ -8,7 +8,7 @@ import (
 )

 func TestScheduler_Init(t *testing.T) {
-	mocks := SetupMocks(false)
+	mocks := SetupMocks(nil)
 	mocks.ctrl.advanceCanary("podinfo", "default", true)

 	_, err := mocks.kubeClient.AppsV1().Deployments("default").Get("podinfo-primary", metav1.GetOptions{})
@@ -18,7 +18,7 @@ func TestScheduler_Init(t *testing.T) {
 }

 func TestScheduler_NewRevision(t *testing.T) {
-	mocks := SetupMocks(false)
+	mocks := SetupMocks(nil)
 	mocks.ctrl.advanceCanary("podinfo", "default", true)

 	// update
@@ -42,7 +42,7 @@ func TestScheduler_NewRevision(t *testing.T) {
 }

 func TestScheduler_Rollback(t *testing.T) {
-	mocks := SetupMocks(false)
+	mocks := SetupMocks(nil)
 	// init
 	mocks.ctrl.advanceCanary("podinfo", "default", true)

@@ -66,7 +66,7 @@ func TestScheduler_Rollback(t *testing.T) {
 }

 func TestScheduler_SkipAnalysis(t *testing.T) {
-	mocks := SetupMocks(false)
+	mocks := SetupMocks(nil)
 	// init
 	mocks.ctrl.advanceCanary("podinfo", "default", true)

@@ -107,7 +107,7 @@ func TestScheduler_SkipAnalysis(t *testing.T) {
 }

 func TestScheduler_NewRevisionReset(t *testing.T) {
-	mocks := SetupMocks(false)
+	mocks := SetupMocks(nil)
 	// init
 	mocks.ctrl.advanceCanary("podinfo", "default", true)

@@ -123,7 +123,7 @@ func TestScheduler_NewRevisionReset(t *testing.T) {
 	// advance
 	mocks.ctrl.advanceCanary("podinfo", "default", true)

-	primaryWeight, canaryWeight, err := mocks.router.GetRoutes(mocks.canary)
+	primaryWeight, canaryWeight, mirrored, err := mocks.router.GetRoutes(mocks.canary)
 	if err != nil {
 		t.Fatal(err.Error())
 	}
@@ -136,6 +136,10 @@ func TestScheduler_NewRevisionReset(t *testing.T) {
 		t.Errorf("Got canary route %v wanted %v", canaryWeight, 10)
 	}

+	if mirrored != false {
+		t.Errorf("Got mirrored %v wanted %v", mirrored, false)
+	}
+
 	// second update
 	dep2.Spec.Template.Spec.ServiceAccountName = "test"
 	_, err = mocks.kubeClient.AppsV1().Deployments("default").Update(dep2)
@@ -146,7 +150,7 @@ func TestScheduler_NewRevisionReset(t *testing.T) {
 	// detect changes
 	mocks.ctrl.advanceCanary("podinfo", "default", true)

-	primaryWeight, canaryWeight, err = mocks.router.GetRoutes(mocks.canary)
+	primaryWeight, canaryWeight, mirrored, err = mocks.router.GetRoutes(mocks.canary)
 	if err != nil {
 		t.Fatal(err.Error())
 	}
@@ -158,10 +162,14 @@ func TestScheduler_NewRevisionReset(t *testing.T) {
 	if canaryWeight != 0 {
 		t.Errorf("Got canary route %v wanted %v", canaryWeight, 0)
 	}
+
+	if mirrored != false {
+		t.Errorf("Got mirrored %v wanted %v", mirrored, false)
+	}
 }

 func TestScheduler_Promotion(t *testing.T) {
-	mocks := SetupMocks(false)
+	mocks := SetupMocks(nil)

 	// init
 	mocks.ctrl.advanceCanary("podinfo", "default", true)
@@ -201,14 +209,14 @@ func TestScheduler_Promotion(t *testing.T) {
 	// detect configs changes
 	mocks.ctrl.advanceCanary("podinfo", "default", true)

-	primaryWeight, canaryWeight, err := mocks.router.GetRoutes(mocks.canary)
+	primaryWeight, canaryWeight, mirrored, err := mocks.router.GetRoutes(mocks.canary)
 	if err != nil {
 		t.Fatal(err.Error())
 	}

 	primaryWeight = 60
 	canaryWeight = 40
-	err = mocks.router.SetRoutes(mocks.canary, primaryWeight, canaryWeight)
+	err = mocks.router.SetRoutes(mocks.canary, primaryWeight, canaryWeight, mirrored)
 	if err != nil {
 		t.Fatal(err.Error())
 	}
@@ -242,7 +250,7 @@ func TestScheduler_Promotion(t *testing.T) {
 	// finalise
 	mocks.ctrl.advanceCanary("podinfo", "default", true)

-	primaryWeight, canaryWeight, err = mocks.router.GetRoutes(mocks.canary)
+	primaryWeight, canaryWeight, mirrored, err = mocks.router.GetRoutes(mocks.canary)
 	if err != nil {
 		t.Fatal(err.Error())
 	}
@@ -255,6 +263,10 @@ func TestScheduler_Promotion(t *testing.T) {
 		t.Errorf("Got canary route %v wanted %v", canaryWeight, 0)
 	}

+	if mirrored != false {
+		t.Errorf("Got mirrored %v wanted %v", mirrored, false)
+	}
+
 	primaryDep, err := mocks.kubeClient.AppsV1().Deployments("default").Get("podinfo-primary", metav1.GetOptions{})
 	if err != nil {
 		t.Fatal(err.Error())
@@ -307,8 +319,66 @@ func TestScheduler_Promotion(t *testing.T) {
 	}
 }

+func TestScheduler_Mirroring(t *testing.T) {
+	mocks := SetupMocks(newTestCanaryMirror())
+	// init
+	mocks.ctrl.advanceCanary("podinfo", "default", true)
+
+	// update
+	dep2 := newTestDeploymentV2()
+	_, err := mocks.kubeClient.AppsV1().Deployments("default").Update(dep2)
+	if err != nil {
+		t.Fatal(err.Error())
+	}
+
+	// detect pod spec changes
+	mocks.ctrl.advanceCanary("podinfo", "default", true)
+
+	// advance
+	mocks.ctrl.advanceCanary("podinfo", "default", true)
+
+	// check if traffic is mirrored to canary
+	primaryWeight, canaryWeight, mirrored, err := mocks.router.GetRoutes(mocks.canary)
+	if err != nil {
+		t.Fatal(err.Error())
+	}
+
+	if primaryWeight != 100 {
+		t.Errorf("Got primary route %v wanted %v", primaryWeight, 100)
+	}
+
+	if canaryWeight != 0 {
+		t.Errorf("Got canary route %v wanted %v", canaryWeight, 0)
+	}
+
+	if mirrored != true {
+		t.Errorf("Got mirrored %v wanted %v", mirrored, true)
+	}
+
+	// advance
+	mocks.ctrl.advanceCanary("podinfo", "default", true)
+
+	// check if traffic is mirrored to canary
+	primaryWeight, canaryWeight, mirrored, err = mocks.router.GetRoutes(mocks.canary)
+	if err != nil {
+		t.Fatal(err.Error())
+	}
+
+	if primaryWeight != 90 {
+		t.Errorf("Got primary route %v wanted %v", primaryWeight, 90)
+	}
+
+	if canaryWeight != 10 {
+		t.Errorf("Got canary route %v wanted %v", canaryWeight, 10)
+	}
+
+	if mirrored != false {
+		t.Errorf("Got mirrored %v wanted %v", mirrored, false)
+	}
+}
+
 func TestScheduler_ABTesting(t *testing.T) {
-	mocks := SetupMocks(true)
+	mocks := SetupMocks(newTestCanaryAB())
 	// init
 	mocks.ctrl.advanceCanary("podinfo", "default", true)

@@ -326,7 +396,7 @@ func TestScheduler_ABTesting(t *testing.T) {
 	mocks.ctrl.advanceCanary("podinfo", "default", true)

 	// check if traffic is routed to canary
-	primaryWeight, canaryWeight, err := mocks.router.GetRoutes(mocks.canary)
+	primaryWeight, canaryWeight, mirrored, err := mocks.router.GetRoutes(mocks.canary)
 	if err != nil {
 		t.Fatal(err.Error())
 	}
@@ -339,6 +409,10 @@ func TestScheduler_ABTesting(t *testing.T) {
 		t.Errorf("Got canary route %v wanted %v", canaryWeight, 100)
 	}

+	if mirrored != false {
+		t.Errorf("Got mirrored %v wanted %v", mirrored, false)
+	}
+
 	cd, err := mocks.flaggerClient.FlaggerV1alpha3().Canaries("default").Get("podinfo", metav1.GetOptions{})
 	if err != nil {
 		t.Fatal(err.Error())
@@ -392,7 +466,7 @@ func TestScheduler_ABTesting(t *testing.T) {
 }

 func TestScheduler_PortDiscovery(t *testing.T) {
-	mocks := SetupMocks(false)
+	mocks := SetupMocks(nil)

 	// enable port discovery
 	cd, err := mocks.flaggerClient.FlaggerV1alpha3().Canaries("default").Get("podinfo", metav1.GetOptions{})