Canary Releases

Canary Release Testing: Metrics-Driven Promotion and Automated Rollback

HelpMeTest

HelpMeTest

— 6 min read
Canary Release Testing: Metrics-Driven Promotion and Automated Rollback

Canary deployments are only as good as the analysis behind them. Shipping 5% of traffic to a new version and watching a dashboard manually is not a canary strategy — it's a slow, manual deployment. Real canary testing uses automated analysis: metrics-based promotion gates, statistical significance checks, and automatic rollback when things go wrong.

This guide covers how to build automated canary analysis that actually catches regressions.

What Good Canary Analysis Looks Like

The canary process should be automatic:

  1. Deploy new version to 5% of traffic
  2. Collect metrics for 15–30 minutes: error rate, latency, custom business metrics
  3. Compare canary metrics to baseline using statistical tests
  4. If metrics look good: promote to 25%, 50%, 100%
  5. If metrics degrade: automatic rollback, alert on-call

The key word is automatic. Human judgment in the middle of a canary analysis introduces delay and inconsistency.

Setting Up Canary Analysis with Flagger

Flagger automates canary deployments on Kubernetes with built-in metric analysis:

# canary.yaml
apiVersion: flagger.app/v1beta1
kind: Canary
metadata:
  name: myapp
  namespace: production
spec:
  targetRef:
    apiVersion: apps/v1
    kind: Deployment
    name: myapp
  
  progressDeadlineSeconds: 3600  # 1 hour max for full rollout
  
  service:
    port: 80
    targetPort: 8080
  
  analysis:
    interval: 2m          # Analyze every 2 minutes
    threshold: 5           # Fail after 5 consecutive metric failures
    maxWeight: 50          # Max canary traffic: 50%
    stepWeight: 10         # Increase by 10% per step
    
    # Prometheus metrics to watch
    metrics:
      - name: error-rate
        thresholdRange:
          max: 1           # Max 1% error rate
        interval: 2m
      
      - name: latency-p99
        templateRef:
          name: latency-p99
          namespace: flagger-system
        thresholdRange:
          max: 500         # Max 500ms p99
        interval: 2m
    
    # Webhooks for custom checks
    webhooks:
      - name: load-test
        type: rollout
        url: http://flagger-loadtester.test/
        metadata:
          cmd: "hey -z 2m -q 10 -c 2 http://myapp-canary.production/"
      
      - name: smoke-test
        type: pre-rollout
        url: http://flagger-loadtester.test/
        metadata:
          cmd: "curl -sf http://myapp-canary.production/health"

---
# Custom metric template for p99 latency
apiVersion: flagger.app/v1beta1
kind: MetricTemplate
metadata:
  name: latency-p99
  namespace: flagger-system
spec:
  provider:
    type: prometheus
    address: http://prometheus.monitoring:9090
  query: |
    histogram_quantile(0.99,
      sum(
        rate(http_request_duration_seconds_bucket{
          namespace="{{ namespace }}",
          ingress="{{ ingress }}"
        }[2m])
      ) by (le)
    ) * 1000

Argo Rollouts with Analysis Templates

Argo Rollouts provides another approach with explicit AnalysisTemplate objects:

# analysis-template.yaml
apiVersion: argoproj.io/v1alpha1
kind: AnalysisTemplate
metadata:
  name: http-benchmark
spec:
  args:
    - name: service-name
  
  metrics:
    # Error rate check via Prometheus
    - name: error-rate
      interval: 2m
      successCondition: result[0] <= 0.01
      failureLimit: 3
      provider:
        prometheus:
          address: http://prometheus.monitoring:9090
          query: |
            sum(
              rate(http_requests_total{
                service="{{args.service-name}}",
                status=~"5.."
              }[2m])
            ) /
            sum(
              rate(http_requests_total{
                service="{{args.service-name}}"
              }[2m])
            )
    
    # Latency check
    - name: p99-latency-ms
      interval: 2m
      successCondition: result[0] <= 500
      failureLimit: 3
      provider:
        prometheus:
          address: http://prometheus.monitoring:9090
          query: |
            histogram_quantile(0.99,
              rate(http_request_duration_seconds_bucket{
                service="{{args.service-name}}"
              }[2m])
            ) * 1000
    
    # Business metric: conversion rate should not drop
    - name: conversion-rate
      interval: 5m
      successCondition: result[0] >= 0.03  # Min 3% conversion
      failureLimit: 2
      provider:
        prometheus:
          address: http://prometheus.monitoring:9090
          query: |
            sum(rate(checkout_completed_total{service="{{args.service-name}}"}[5m])) /
            sum(rate(checkout_started_total{service="{{args.service-name}}"}[5m]))

---
# rollout.yaml
apiVersion: argoproj.io/v1alpha1
kind: Rollout
metadata:
  name: myapp
spec:
  replicas: 10
  strategy:
    canary:
      canaryService: myapp-canary
      stableService: myapp-stable
      
      steps:
        - setWeight: 10
        - pause: {duration: 5m}
        - analysis:
            templates:
              - templateName: http-benchmark
            args:
              - name: service-name
                value: myapp
        
        - setWeight: 30
        - pause: {duration: 5m}
        - analysis:
            templates:
              - templateName: http-benchmark
            args:
              - name: service-name
                value: myapp
        
        - setWeight: 100

Writing Custom Canary Analysis

For more control, write your own canary analysis:

# canary_analysis.py
import requests
import statistics
import scipy.stats as stats
from dataclasses import dataclass
from typing import Optional
from datetime import datetime, timedelta
import logging

logger = logging.getLogger(__name__)

@dataclass
class MetricSample:
    timestamp: datetime
    canary_value: float
    baseline_value: float

class CanaryAnalyzer:
    def __init__(self, prometheus_url: str):
        self.prometheus = prometheus_url
    
    def query_metric(self, query: str, duration_minutes: int = 30) -> list[float]:
        """Query Prometheus for metric values over the analysis window."""
        end = datetime.utcnow()
        start = end - timedelta(minutes=duration_minutes)
        
        response = requests.get(
            f"{self.prometheus}/api/v1/query_range",
            params={
                "query": query,
                "start": start.timestamp(),
                "end": end.timestamp(),
                "step": "60",  # 1-minute resolution
            }
        )
        
        data = response.json()
        if data["status"] != "success":
            raise ValueError(f"Prometheus query failed: {data}")
        
        results = data["data"]["result"]
        if not results:
            return []
        
        return [float(v[1]) for v in results[0]["values"]]
    
    def analyze(
        self,
        service: str,
        canary_version: str,
        baseline_version: str,
    ) -> dict:
        """
        Full canary analysis: compare canary vs baseline across key metrics.
        Returns promotion decision with rationale.
        """
        results = {}
        
        # Error rate analysis
        canary_errors = self.query_metric(
            f'rate(http_requests_total{{version="{canary_version}",status=~"5.."}}[5m])'
            f' / rate(http_requests_total{{version="{canary_version}"}}[5m])'
        )
        baseline_errors = self.query_metric(
            f'rate(http_requests_total{{version="{baseline_version}",status=~"5.."}}[5m])'
            f' / rate(http_requests_total{{version="{baseline_version}"}}[5m])'
        )
        
        results["error_rate"] = self._compare_metric(
            canary_errors, baseline_errors,
            max_absolute=0.01,   # Max 1% error rate
            max_relative_increase=0.50,  # Max 50% increase over baseline
            lower_is_better=True,
        )
        
        # Latency analysis
        canary_latency = self.query_metric(
            f'histogram_quantile(0.99, rate(http_request_duration_seconds_bucket{{version="{canary_version}"}}[5m])) * 1000'
        )
        baseline_latency = self.query_metric(
            f'histogram_quantile(0.99, rate(http_request_duration_seconds_bucket{{version="{baseline_version}"}}[5m])) * 1000'
        )
        
        results["p99_latency_ms"] = self._compare_metric(
            canary_latency, baseline_latency,
            max_absolute=500,    # Max 500ms p99
            max_relative_increase=0.20,  # Max 20% latency increase
            lower_is_better=True,
        )
        
        # Overall decision
        all_passed = all(r["passed"] for r in results.values())
        
        return {
            "decision": "PROMOTE" if all_passed else "ROLLBACK",
            "metrics": results,
            "summary": self._format_summary(results),
        }
    
    def _compare_metric(
        self,
        canary: list[float],
        baseline: list[float],
        max_absolute: float,
        max_relative_increase: float,
        lower_is_better: bool,
    ) -> dict:
        if not canary or not baseline:
            return {"passed": False, "reason": "Insufficient data"}
        
        canary_mean = statistics.mean(canary)
        baseline_mean = statistics.mean(baseline)
        
        # Absolute threshold check
        if lower_is_better and canary_mean > max_absolute:
            return {
                "passed": False,
                "reason": f"Absolute threshold exceeded: {canary_mean:.4f} > {max_absolute}",
                "canary": canary_mean,
                "baseline": baseline_mean,
            }
        
        # Relative regression check
        if baseline_mean > 0:
            relative_change = (canary_mean - baseline_mean) / baseline_mean
            if lower_is_better and relative_change > max_relative_increase:
                return {
                    "passed": False,
                    "reason": f"Relative regression: {relative_change:.0%} increase over baseline",
                    "canary": canary_mean,
                    "baseline": baseline_mean,
                }
        
        # Statistical significance (Mann-Whitney U test)
        if len(canary) >= 5 and len(baseline) >= 5:
            _, p_value = stats.mannwhitneyu(canary, baseline, alternative="greater" if lower_is_better else "less")
            if p_value < 0.05:
                return {
                    "passed": False,
                    "reason": f"Statistically significant regression (p={p_value:.3f})",
                    "canary": canary_mean,
                    "baseline": baseline_mean,
                }
        
        return {
            "passed": True,
            "canary": canary_mean,
            "baseline": baseline_mean,
        }

Testing the Canary Analysis Itself

Your canary analysis logic needs tests too:

# tests/test_canary_analysis.py
import pytest
from unittest.mock import patch, MagicMock
from canary_analysis import CanaryAnalyzer

@pytest.fixture
def analyzer():
    return CanaryAnalyzer(prometheus_url="http://prometheus:9090")

def test_promote_when_metrics_healthy(analyzer):
    """Should promote when both error rate and latency are within bounds."""
    with patch.object(analyzer, 'query_metric') as mock_query:
        # Return healthy metrics for both canary and baseline
        mock_query.side_effect = [
            [0.001] * 20,  # Canary error rate: 0.1%
            [0.001] * 20,  # Baseline error rate: 0.1%
            [150.0] * 20,  # Canary latency: 150ms
            [145.0] * 20,  # Baseline latency: 145ms
        ]
        
        result = analyzer.analyze("myapp", "v2", "v1")
        
        assert result["decision"] == "PROMOTE"

def test_rollback_on_high_error_rate(analyzer):
    """Should rollback when error rate exceeds 1%."""
    with patch.object(analyzer, 'query_metric') as mock_query:
        mock_query.side_effect = [
            [0.05] * 20,   # Canary error rate: 5% - too high!
            [0.001] * 20,  # Baseline error rate: 0.1%
            [150.0] * 20,  # Latency fine
            [145.0] * 20,
        ]
        
        result = analyzer.analyze("myapp", "v2", "v1")
        
        assert result["decision"] == "ROLLBACK"
        assert "error_rate" in result["metrics"]
        assert not result["metrics"]["error_rate"]["passed"]

def test_rollback_on_latency_regression(analyzer):
    """Should rollback when p99 latency increases more than 20%."""
    with patch.object(analyzer, 'query_metric') as mock_query:
        mock_query.side_effect = [
            [0.001] * 20,  # Error rate fine
            [0.001] * 20,
            [620.0] * 20,  # Canary latency: 620ms (25% increase)
            [500.0] * 20,  # Baseline latency: 500ms
        ]
        
        result = analyzer.analyze("myapp", "v2", "v1")
        
        assert result["decision"] == "ROLLBACK"

def test_handles_empty_metrics_gracefully(analyzer):
    """Should not crash when metrics are unavailable."""
    with patch.object(analyzer, 'query_metric', return_value=[]):
        result = analyzer.analyze("myapp", "v2", "v1")
        
        assert result["decision"] == "ROLLBACK"  # Fail safe on no data

Rollback Automation

# canary_controller.py
import time
from kubernetes import client, config

class CanaryController:
    def __init__(self, namespace: str, rollout_name: str):
        config.load_incluster_config()
        self.custom_api = client.CustomObjectsApi()
        self.namespace = namespace
        self.rollout_name = rollout_name
    
    def get_canary_weight(self) -> int:
        """Get current canary traffic percentage."""
        rollout = self.custom_api.get_namespaced_custom_object(
            group="argoproj.io",
            version="v1alpha1",
            namespace=self.namespace,
            plural="rollouts",
            name=self.rollout_name,
        )
        return rollout["status"].get("canaryReplicas", 0)
    
    def abort_rollout(self, reason: str):
        """Abort and roll back the canary."""
        patch = {"spec": {"abort": True}}
        self.custom_api.patch_namespaced_custom_object(
            group="argoproj.io",
            version="v1alpha1",
            namespace=self.namespace,
            plural="rollouts",
            name=self.rollout_name,
            body=patch,
        )
        logger.warning(f"Rollout aborted: {reason}")
    
    def run_analysis_loop(self, analyzer: CanaryAnalyzer, interval_seconds: int = 120):
        """Continuously analyze canary and rollback if needed."""
        while True:
            weight = self.get_canary_weight()
            
            if weight == 0:
                break  # Rollout complete or already rolled back
            
            result = analyzer.analyze(
                service=self.rollout_name,
                canary_version="canary",
                baseline_version="stable",
            )
            
            if result["decision"] == "ROLLBACK":
                self.abort_rollout(result["summary"])
                break
            
            time.sleep(interval_seconds)

Summary

Automated canary analysis is the difference between a real canary strategy and a manual deployment with extra steps:

  1. Define metrics before deploying — error rate, latency, business KPIs
  2. Use statistical tests — don't rollback on noise; require significance
  3. Set absolute and relative thresholds — catch regressions even when baseline is degraded
  4. Automate the rollback — human approval in the rollback path means regressions last longer
  5. Test your analysis logic — the canary analyzer is code; it needs tests

Use HelpMeTest to run smoke tests as part of your canary webhook — an automated test that exercises critical paths gives you fast signal before statistical analysis completes.

Read more

Start now free