diff --git a/pkg/controller/deployer.go b/pkg/controller/deployer.go index fb7a2bd7..74f48c68 100644 --- a/pkg/controller/deployer.go +++ b/pkg/controller/deployer.go @@ -220,6 +220,32 @@ func (c *CanaryDeployer) SetStatusWeight(cd *flaggerv1.Canary, val int) error { return nil } +// SetStatusWeight updates the canary status weight value +func (c *CanaryDeployer) SetStatusIterations(cd *flaggerv1.Canary, val int) error { + cdCopy := cd.DeepCopy() + cdCopy.Status.Iterations = val + cdCopy.Status.LastTransitionTime = metav1.Now() + + cd, err := c.flaggerClient.FlaggerV1alpha3().Canaries(cd.Namespace).UpdateStatus(cdCopy) + if err != nil { + return fmt.Errorf("canary %s.%s status update error %v", cdCopy.Name, cdCopy.Namespace, err) + } + return nil +} + +// SetStatusWeight updates the canary status weight value +func (c *CanaryDeployer) IncrementStatusIterations(cd *flaggerv1.Canary) error { + cdCopy := cd.DeepCopy() + cdCopy.Status.Iterations = cdCopy.Status.Iterations + 1 + cdCopy.Status.LastTransitionTime = metav1.Now() + + cd, err := c.flaggerClient.FlaggerV1alpha3().Canaries(cd.Namespace).UpdateStatus(cdCopy) + if err != nil { + return fmt.Errorf("canary %s.%s status update error %v", cdCopy.Name, cdCopy.Namespace, err) + } + return nil +} + // SetStatusPhase updates the canary status phase func (c *CanaryDeployer) SetStatusPhase(cd *flaggerv1.Canary, phase flaggerv1.CanaryPhase) error { cdCopy := cd.DeepCopy() @@ -228,6 +254,7 @@ func (c *CanaryDeployer) SetStatusPhase(cd *flaggerv1.Canary, phase flaggerv1.Ca if phase != flaggerv1.CanaryProgressing { cdCopy.Status.CanaryWeight = 0 + cdCopy.Status.Iterations = 0 } cd, err := c.flaggerClient.FlaggerV1alpha3().Canaries(cd.Namespace).UpdateStatus(cdCopy) @@ -261,6 +288,7 @@ func (c *CanaryDeployer) SyncStatus(cd *flaggerv1.Canary, status flaggerv1.Canar cdCopy.Status.Phase = status.Phase cdCopy.Status.CanaryWeight = status.CanaryWeight cdCopy.Status.FailedChecks = status.FailedChecks + cdCopy.Status.Iterations = status.Iterations cdCopy.Status.LastAppliedSpec = base64.StdEncoding.EncodeToString(specJson) cdCopy.Status.LastTransitionTime = metav1.Now() cdCopy.Status.TrackedConfigs = configs diff --git a/pkg/controller/scheduler.go b/pkg/controller/scheduler.go index d60a7e95..f9047b75 100644 --- a/pkg/controller/scheduler.go +++ b/pkg/controller/scheduler.go @@ -76,10 +76,13 @@ func (c *Controller) advanceCanary(name string, namespace string, skipLivenessCh // check if the canary exists cd, err := c.flaggerClient.FlaggerV1alpha3().Canaries(namespace).Get(name, v1.GetOptions{}) if err != nil { - c.logger.With("canary", fmt.Sprintf("%s.%s", name, namespace)).Errorf("Canary %s.%s not found", name, namespace) + c.logger.With("canary", fmt.Sprintf("%s.%s", name, namespace)). + Errorf("Canary %s.%s not found", name, namespace) return } + primaryName := fmt.Sprintf("%s-primary", cd.Spec.TargetRef.Name) + // create primary deployment and hpa if needed if err := c.deployer.Sync(cd); err != nil { c.recordEventWarningf(cd, "%v", err) @@ -160,6 +163,7 @@ func (c *Controller) advanceCanary(name string, namespace string, skipLivenessCh Phase: flaggerv1.CanaryProgressing, CanaryWeight: 0, FailedChecks: 0, + Iterations: 0, } if err := c.deployer.SyncStatus(cd, status); err != nil { c.recordEventWarningf(cd, "%v", err) @@ -247,7 +251,72 @@ func (c *Controller) advanceCanary(name string, namespace string, skipLivenessCh } } - // increase canary traffic percentage + // canary fix routing: A/B testing + if len(cd.Spec.CanaryAnalysis.Match) > 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 { + c.recordEventWarningf(cd, "%v", err) + return + } + c.recorder.SetWeight(cd, 100, 0) + + if err := c.deployer.SetStatusIterations(cd, cd.Status.Iterations+1); err != nil { + c.recordEventWarningf(cd, "%v", err) + return + } + c.recordEventInfof(cd, "Advance %s.%s canary iteration %v/%v", + cd.Name, cd.Namespace, cd.Status.Iterations+1, cd.Spec.CanaryAnalysis.Iterations) + return + } + + // promote canary - max iterations reached + if cd.Spec.CanaryAnalysis.Iterations == cd.Status.Iterations { + c.recordEventInfof(cd, "Copying %s.%s template spec to %s.%s", + cd.Spec.TargetRef.Name, cd.Namespace, primaryName, cd.Namespace) + if err := c.deployer.Promote(cd); err != nil { + c.recordEventWarningf(cd, "%v", err) + return + } + // increment iterations + if err := c.deployer.SetStatusIterations(cd, cd.Status.Iterations+1); err != nil { + c.recordEventWarningf(cd, "%v", err) + return + } + return + } + + // shutdown canary + if cd.Spec.CanaryAnalysis.Iterations < cd.Status.Iterations { + // route all traffic to the primary + if err := meshRouter.SetRoutes(cd, 100, 0); err != nil { + c.recordEventWarningf(cd, "%v", err) + return + } + c.recorder.SetWeight(cd, 100, 0) + c.recordEventInfof(cd, "Promotion completed! Scaling down %s.%s", cd.Spec.TargetRef.Name, cd.Namespace) + + // canary scale to zero + if err := c.deployer.Scale(cd, 0); err != nil { + c.recordEventWarningf(cd, "%v", err) + return + } + + // update status phase + if err := c.deployer.SetStatusPhase(cd, flaggerv1.CanarySucceeded); err != nil { + c.recordEventWarningf(cd, "%v", err) + return + } + c.recorder.SetStatus(cd) + c.sendNotification(cd, "Canary analysis completed successfully, promotion finished.", + false, false) + return + } + + return + } + + // canary incremental traffic weight if canaryWeight < maxWeight { primaryWeight -= cd.Spec.CanaryAnalysis.StepWeight if primaryWeight < 0 { @@ -273,7 +342,6 @@ func (c *Controller) advanceCanary(name string, namespace string, skipLivenessCh c.recordEventInfof(cd, "Advance %s.%s canary weight %v", cd.Name, cd.Namespace, canaryWeight) // promote canary - primaryName := fmt.Sprintf("%s-primary", cd.Spec.TargetRef.Name) if canaryWeight == maxWeight { c.recordEventInfof(cd, "Copying %s.%s template spec to %s.%s", cd.Spec.TargetRef.Name, cd.Namespace, primaryName, cd.Namespace) @@ -403,6 +471,13 @@ func (c *Controller) hasCanaryRevisionChanged(cd *flaggerv1.Canary) bool { return false } +func (c *Controller) hasMaxIterations(cd *flaggerv1.Canary) bool { + if cd.Status.Iterations == cd.Status.Iterations { + return true + } + return false +} + func (c *Controller) analyseCanary(r *flaggerv1.Canary) bool { // run external checks for _, webhook := range r.Spec.CanaryAnalysis.Webhooks { diff --git a/pkg/router/istio.go b/pkg/router/istio.go index feef64d1..7f3ad674 100644 --- a/pkg/router/istio.go +++ b/pkg/router/istio.go @@ -55,7 +55,7 @@ func (ir *IstioRouter) Sync(canary *flaggerv1.Canary) error { } // create destinations with primary weight 100% and canary weight 0% - route := []istiov1alpha3.DestinationWeight{ + canaryRoute := []istiov1alpha3.DestinationWeight{ { Destination: istiov1alpha3.Destination{ Host: primaryName, @@ -87,11 +87,45 @@ func (ir *IstioRouter) Sync(canary *flaggerv1.Canary) error { Retries: canary.Spec.Service.Retries, CorsPolicy: canary.Spec.Service.CorsPolicy, AppendHeaders: addHeaders(canary), - Route: route, + Route: canaryRoute, }, }, } + if len(canary.Spec.CanaryAnalysis.Match) > 0 { + canaryMatch := append(canary.Spec.Service.Match, canary.Spec.CanaryAnalysis.Match...) + newSpec.Http = []istiov1alpha3.HTTPRoute{ + { + Match: canaryMatch, + Rewrite: canary.Spec.Service.Rewrite, + Timeout: canary.Spec.Service.Timeout, + Retries: canary.Spec.Service.Retries, + CorsPolicy: canary.Spec.Service.CorsPolicy, + AppendHeaders: addHeaders(canary), + Route: canaryRoute, + }, + { + Match: canary.Spec.Service.Match, + Rewrite: canary.Spec.Service.Rewrite, + Timeout: canary.Spec.Service.Timeout, + Retries: canary.Spec.Service.Retries, + CorsPolicy: canary.Spec.Service.CorsPolicy, + AppendHeaders: addHeaders(canary), + Route: []istiov1alpha3.DestinationWeight{ + { + Destination: istiov1alpha3.Destination{ + Host: primaryName, + Port: istiov1alpha3.PortSelector{ + Number: uint32(canary.Spec.Service.Port), + }, + }, + Weight: 100, + }, + }, + }, + } + } + virtualService, err := ir.istioClient.NetworkingV1alpha3().VirtualServices(canary.Namespace).Get(targetName, metav1.GetOptions{}) // insert if errors.IsNotFound(err) { @@ -160,17 +194,25 @@ func (ir *IstioRouter) GetRoutes(canary *flaggerv1.Canary) ( return } + var httpRoute istiov1alpha3.HTTPRoute for _, http := range vs.Spec.Http { - for _, route := range http.Route { - if route.Destination.Host == fmt.Sprintf("%s-primary", targetName) { - primaryWeight = route.Weight - } - if route.Destination.Host == fmt.Sprintf("%s-canary", targetName) { - canaryWeight = route.Weight + for _, r := range http.Route { + if r.Destination.Host == fmt.Sprintf("%s-canary", targetName) { + httpRoute = http + break } } } + for _, route := range httpRoute.Route { + if route.Destination.Host == fmt.Sprintf("%s-primary", targetName) { + primaryWeight = route.Weight + } + if route.Destination.Host == fmt.Sprintf("%s-canary", targetName) { + canaryWeight = route.Weight + } + } + if primaryWeight == 0 && canaryWeight == 0 { err = fmt.Errorf("VirtualService %s.%s does not contain routes for %s-primary and %s-canary", targetName, canary.Namespace, targetName, targetName) @@ -196,6 +238,8 @@ func (ir *IstioRouter) SetRoutes( } vsCopy := vs.DeepCopy() + + // weighted routing (progressive canary) vsCopy.Spec.Http = []istiov1alpha3.HTTPRoute{ { Match: canary.Spec.Service.Match, @@ -227,6 +271,61 @@ func (ir *IstioRouter) SetRoutes( }, } + // fix routing (A/B testing) + if len(canary.Spec.CanaryAnalysis.Match) > 0 { + // merge the common routes with the canary ones + canaryMatch := append(canary.Spec.Service.Match, canary.Spec.CanaryAnalysis.Match...) + vsCopy.Spec.Http = []istiov1alpha3.HTTPRoute{ + { + Match: canaryMatch, + Rewrite: canary.Spec.Service.Rewrite, + Timeout: canary.Spec.Service.Timeout, + Retries: canary.Spec.Service.Retries, + CorsPolicy: canary.Spec.Service.CorsPolicy, + AppendHeaders: addHeaders(canary), + Route: []istiov1alpha3.DestinationWeight{ + { + Destination: istiov1alpha3.Destination{ + Host: fmt.Sprintf("%s-primary", targetName), + Port: istiov1alpha3.PortSelector{ + Number: uint32(canary.Spec.Service.Port), + }, + }, + Weight: primaryWeight, + }, + { + Destination: istiov1alpha3.Destination{ + Host: fmt.Sprintf("%s-canary", targetName), + Port: istiov1alpha3.PortSelector{ + Number: uint32(canary.Spec.Service.Port), + }, + }, + Weight: canaryWeight, + }, + }, + }, + { + Match: canary.Spec.Service.Match, + Rewrite: canary.Spec.Service.Rewrite, + Timeout: canary.Spec.Service.Timeout, + Retries: canary.Spec.Service.Retries, + CorsPolicy: canary.Spec.Service.CorsPolicy, + AppendHeaders: addHeaders(canary), + Route: []istiov1alpha3.DestinationWeight{ + { + Destination: istiov1alpha3.Destination{ + Host: fmt.Sprintf("%s-primary", targetName), + Port: istiov1alpha3.PortSelector{ + Number: uint32(canary.Spec.Service.Port), + }, + }, + Weight: primaryWeight, + }, + }, + }, + } + } + vs, err = ir.istioClient.NetworkingV1alpha3().VirtualServices(canary.Namespace).Update(vsCopy) if err != nil { return fmt.Errorf("VirtualService %s.%s update failed: %v", targetName, canary.Namespace, err)