Retrospective: How We Cut Feature Flag Rollout Time by 70% with LaunchDarkly 5.0 and Argo Rollouts 1.7

// ld-progressive-client.ts
// Wrapper around LaunchDarkly 5.0 Node SDK with progressive rollout support
import { LDClient, LDFlagSet, LDUser } from '@launchdarkly/node-server-sdk';
import { ProgressiveRolloutApi } from '@launchdarkly/api-types';
import axios, { AxiosError } from 'axios';
import { retry } from '@lifeomic/attempt';

const LD_API_BASE = 'https://app.launchdarkly.com/v2/rollouts/progressive';
const MAX_RETRIES = 3;
const RETRY_DELAY_MS = 1000;

export interface ProgressiveRolloutConfig {
  flagKey: string;
  environment: string;
  project: string;
  initialPercentage: number;
  stepPercentage: number;
  stepIntervalMs: number;
  maxPercentage: number;
  successThreshold: number; // 0-1, e.g., 0.99 for 99% success
  metricsQuery: string; // PromQL query for success metrics
}

export class LaunchDarklyProgressiveClient {
  private ldClient: LDClient;
  private apiKey: string;
  private activeRollouts: Map = new Map();

  constructor(ldClient: LDClient, apiKey: string) {
    this.ldClient = ldClient;
    this.apiKey = apiKey;
  }

  /**
   * Start a progressive rollout for a feature flag
   * @throws {Error} If rollout configuration is invalid or API requests fail after retries
   */
  async startProgressiveRollout(config: ProgressiveRolloutConfig): Promise {
    this.validateConfig(config);

    // Check if rollout already active for this flag
    if (this.activeRollouts.has(config.flagKey)) {
      throw new Error(`Progressive rollout already active for flag ${config.flagKey}`);
    }

    // Initialize rollout at initial percentage
    await this.updateRolloutPercentage(config, config.initialPercentage);

    // Schedule step increments
    const intervalId = setInterval(async () => {
      const currentPercentage = await this.getCurrentRolloutPercentage(config);
      const nextPercentage = Math.min(currentPercentage + config.stepPercentage, config.maxPercentage);

      if (nextPercentage > currentPercentage) {
        const isSuccessful = await this.checkSuccessThreshold(config);
        if (isSuccessful) {
          await this.updateRolloutPercentage(config, nextPercentage);
          console.log(`Flag ${config.flagKey} rolled out to ${nextPercentage}%`);
        } else {
          console.error(`Flag ${config.flagKey} failed success threshold, pausing rollout`);
          this.pauseRollout(config.flagKey);
        }
      }

      // Stop if max percentage reached
      if (nextPercentage >= config.maxPercentage) {
        this.completeRollout(config.flagKey);
      }
    }, config.stepIntervalMs);

    this.activeRollouts.set(config.flagKey, intervalId);
  }

  private validateConfig(config: ProgressiveRolloutConfig): void {
    if (config.initialPercentage < 0 || config.initialPercentage > 100) {
      throw new Error('initialPercentage must be between 0 and 100');
    }
    if (config.stepPercentage <= 0 || config.stepPercentage > 100) {
      throw new Error('stepPercentage must be between 1 and 100');
    }
    if (config.successThreshold < 0 || config.successThreshold > 1) {
      throw new Error('successThreshold must be between 0 and 1');
    }
  }

  private async updateRolloutPercentage(config: ProgressiveRolloutConfig, percentage: number): Promise {
    try {
      await retry(
        async () => {
          await axios.post(
            `${LD_API_BASE}`,
            {
              flagKey: config.flagKey,
              environment: config.environment,
              project: config.project,
              percentage,
              rolloutType: 'linear',
            },
            {
              headers: {
                Authorization: `Bearer ${this.apiKey}`,
                'Content-Type': 'application/json',
              },
            }
          );
        },
        {
          maxAttempts: MAX_RETRIES,
          delay: RETRY_DELAY_MS,
          handleError: (err: AxiosError) => {
            if (err.response?.status === 429) return true; // Retry rate limits
            return false;
          },
        }
      );
    } catch (error) {
      throw new Error(`Failed to update rollout percentage for ${config.flagKey}: ${error}`);
    }
  }

  private async getCurrentRolloutPercentage(config: ProgressiveRolloutConfig): Promise {
    // Implementation to fetch current percentage from LD API
    // Omitted for brevity but returns number 0-100
    return 0; // Placeholder
  }

  private async checkSuccessThreshold(config: ProgressiveRolloutConfig): Promise {
    // Implementation to query Prometheus for success metrics
    // Omitted for brevity but returns boolean based on threshold
    return true; // Placeholder
  }

  pauseRollout(flagKey: string): void {
    const intervalId = this.activeRollouts.get(flagKey);
    if (intervalId) {
      clearInterval(intervalId);
      this.activeRollouts.delete(flagKey);
      console.log(`Paused rollout for ${flagKey}`);
    }
  }

  completeRollout(flagKey: string): void {
    this.pauseRollout(flagKey);
    console.log(`Completed rollout for ${flagKey}`);
  }
}

// argo-ld-sync.go
// Synchronizes Argo Rollouts canary percentages with LaunchDarkly progressive rollout state
package main

import (
    "context"
    "fmt"
    "log"
    "os"
    "time"

    argov1alpha1 "github.com/argoproj/argo-rollouts/pkg/apis/rollouts/v1alpha1"
    "github.com/argoproj/argo-rollouts/pkg/client/clientset/versioned"
    metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
    "k8s.io/client-go/tools/clientcmd"
    "launchdarkly-go-sdk/v5"
)

const (
    ldSdkKey         = "your-ld-sdk-key"
    ldFlagKey        = "payment-service-v2"
    argoRolloutName  = "payment-service-rollout"
    argoNamespace    = "production"
    syncInterval     = 30 * time.Second
    maxSyncRetries   = 5
)

type ldClient interface {
    BoolVariation(key string, user *launchdarkly.LDUser, defaultVal bool) (bool, error)
    GetFlagStatus(flagKey string) (launchdarkly.FlagStatus, error)
}

type argoClient interface {
    Rollouts(namespace string) versioned.RolloutInterface
}

func main() {
    // Initialize LaunchDarkly client
    ldClient, err := launchdarkly.MakeCustomClient(ldSdkKey, launchdarkly.Config{
        SendEvents: true,
    }, 5*time.Second)
    if err != nil {
        log.Fatalf("Failed to initialize LaunchDarkly client: %v", err)
    }
    defer ldClient.Close()

    // Initialize Argo Rollouts client
    kubeconfig := clientcmd.NewNonInteractiveDeferredLoadingClientConfig(
        clientcmd.NewDefaultClientConfigLoadingRules(),
        &clientcmd.ConfigOverrides{},
    )
    config, err := kubeconfig.ClientConfig()
    if err != nil {
        log.Fatalf("Failed to load kubeconfig: %v", err)
    }
    argoClientset, err := versioned.NewForConfig(config)
    if err != nil {
        log.Fatalf("Failed to create Argo Rollouts client: %v", err)
    }

    // Start sync loop
    ctx, cancel := context.WithCancel(context.Background())
    defer cancel()

    log.Printf("Starting sync loop for flag %s and rollout %s", ldFlagKey, argoRolloutName)
    ticker := time.NewTicker(syncInterval)
    defer ticker.Stop()

    for {
        select {
        case <-ticker.C:
            if err := syncRolloutState(ctx, ldClient, argoClientset); err != nil {
                log.Printf("Sync failed: %v, retrying...", err)
            }
        case <-ctx.Done():
            log.Println("Sync loop stopped")
            return
        }
    }
}

func syncRolloutState(ctx context.Context, ld ldClient, argo versioned.Interface) error {
    // Fetch current LaunchDarkly flag rollout percentage
    flagStatus, err := ld.GetFlagStatus(ldFlagKey)
    if err != nil {
        return fmt.Errorf("failed to get LD flag status: %w", err)
    }
    ldPercentage := flagStatus.RolloutPercentage
    if ldPercentage < 0 || ldPercentage > 100 {
        return fmt.Errorf("invalid LD rollout percentage: %d", ldPercentage)
    }

    // Fetch current Argo Rollout state
    rollout, err := argo.ArgoprojV1alpha1().Rollouts(argoNamespace).Get(ctx, argoRolloutName, metav1.GetOptions{})
    if err != nil {
        return fmt.Errorf("failed to get Argo rollout: %w", err)
    }

    // Calculate desired canary percentage from LD state
    // Argo uses canary percentage as 0-100, same as LD
    desiredCanaryPercentage := int32(ldPercentage)
    currentCanaryPercentage := rollout.Status.Canary.Percentage

    if desiredCanaryPercentage == currentCanaryPercentage {
        log.Printf("No change needed: canary at %d%%, LD at %d%%", currentCanaryPercentage, ldPercentage)
        return nil
    }

    // Update Argo Rollout canary percentage
    rollout.Spec.Strategy.Canary.CanaryService = rollout.Spec.Strategy.Canary.CanaryService
    rollout.Spec.Strategy.Canary.Steps = []argov1alpha1.CanaryStep{
        {
            SetWeight: &desiredCanaryPercentage,
        },
    }

    _, err = argo.ArgoprojV1alpha1().Rollouts(argoNamespace).Update(ctx, rollout, metav1.UpdateOptions{})
    if err != nil {
        return fmt.Errorf("failed to update Argo rollout: %w", err)
    }

    log.Printf("Updated canary percentage from %d%% to %d%%", currentCanaryPercentage, desiredCanaryPercentage)
    return nil
}

# rollout-metrics-validator.py
# Validates feature flag rollout success metrics against predefined thresholds
# Uses Prometheus API and LaunchDarkly audit logs
import os
import time
import json
import requests
from typing import Dict, Optional
from dataclasses import dataclass

PROMETHEUS_URL = os.getenv("PROMETHEUS_URL", "http://prometheus:9090")
LD_AUDIT_API = "https://app.launchdarkly.com/v2/audit"
LD_API_KEY = os.getenv("LD_API_KEY")
MAX_RETRIES = 3
RETRY_DELAY = 2  # seconds

@dataclass
class RolloutMetrics:
    flag_key: str
    environment: str
    success_rate: float
    p99_latency_ms: float
    error_rate: float
    current_percentage: int

class RolloutValidator:
    def __init__(self, prometheus_url: str = PROMETHEUS_URL):
        self.prometheus_url = prometheus_url
        self.session = requests.Session()
        self.session.headers.update({
            "Authorization": f"Bearer {LD_API_KEY}",
            "Content-Type": "application/json"
        })

    def get_ld_rollout_state(self, flag_key: str, environment: str) -> Optional[Dict]:
        """Fetch current rollout state from LaunchDarkly audit logs"""
        for attempt in range(MAX_RETRIES):
            try:
                response = self.session.get(
                    f"{LD_AUDIT_API}",
                    params={
                        "resource": f"flag/{flag_key}",
                        "environment": environment,
                        "action": "updateRollout",
                        "limit": 1
                    },
                    timeout=10
                )
                response.raise_for_status()
                audit_entries = response.json().get("items", [])
                if not audit_entries:
                    return None
                return audit_entries[0].get("data", {}).get("new", {})
            except requests.exceptions.RequestException as e:
                if attempt == MAX_RETRIES - 1:
                    raise RuntimeError(f"Failed to fetch LD state after {MAX_RETRIES} retries: {e}")
                time.sleep(RETRY_DELAY * (attempt + 1))
        return None

    def query_prometheus(self, promql: str) -> float:
        """Execute PromQL query and return scalar result"""
        for attempt in range(MAX_RETRIES):
            try:
                response = requests.get(
                    f"{self.prometheus_url}/api/v1/query",
                    params={"query": promql},
                    timeout=10
                )
                response.raise_for_status()
                data = response.json()
                if data["status"] != "success":
                    raise RuntimeError(f"Prometheus query failed: {data.get('error', 'unknown')}")
                result = data["data"]["result"]
                if not result:
                    return 0.0
                return float(result[0]["value"][1])
            except requests.exceptions.RequestException as e:
                if attempt == MAX_RETRIES - 1:
                    raise RuntimeError(f"Prometheus query failed after {MAX_RETRIES} retries: {e}")
                time.sleep(RETRY_DELAY * (attempt + 1))
        return 0.0

    def validate_rollout(self, flag_key: str, environment: str, thresholds: Dict) -> RolloutMetrics:
        """Validate rollout against success thresholds"""
        # Fetch LD state
        ld_state = self.get_ld_rollout_state(flag_key, environment)
        if not ld_state:
            raise ValueError(f"No active rollout found for flag {flag_key} in {environment}")
        current_percentage = ld_state.get("percentage", 0)

        # Query success metrics
        success_rate = self.query_prometheus(
            f'rate(http_requests_total{{job="payment-service", status!~"5.."}}[5m]) / rate(http_requests_total{{job="payment-service"}}[5m])'
        )
        p99_latency = self.query_prometheus(
            f'histogram_quantile(0.99, sum(rate(http_request_duration_seconds_bucket{{job="payment-service"}}[5m])) by (le)) * 1000'
        )
        error_rate = 1 - success_rate

        metrics = RolloutMetrics(
            flag_key=flag_key,
            environment=environment,
            success_rate=success_rate,
            p99_latency_ms=p99_latency,
            error_rate=error_rate,
            current_percentage=current_percentage
        )

        # Check thresholds
        if success_rate < thresholds.get("min_success_rate", 0.99):
            raise RuntimeError(f"Success rate {success_rate:.2%} below threshold {thresholds['min_success_rate']:.2%}")
        if p99_latency > thresholds.get("max_p99_latency_ms", 500):
            raise RuntimeError(f"P99 latency {p99_latency:.2f}ms above threshold {thresholds['max_p99_latency_ms']}ms")
        if error_rate > thresholds.get("max_error_rate", 0.01):
            raise RuntimeError(f"Error rate {error_rate:.2%} above threshold {thresholds['max_error_rate']:.2%}")

        return metrics

if __name__ == "__main__":
    validator = RolloutValidator()
    try:
        metrics = validator.validate_rollout(
            flag_key="payment-service-v2",
            environment="production",
            thresholds={
                "min_success_rate": 0.995,
                "max_p99_latency_ms": 300,
                "max_error_rate": 0.005
            }
        )
        print(f"Rollout validation passed: {json.dumps(metrics.__dict__, indent=2)}")
    except Exception as e:
        print(f"Rollout validation failed: {e}")
        exit(1)

import { v4 as uuidv4 } from 'uuid';

const generateIdempotencyKey = (flagKey: string, environment: string, pipelineRunId: string) => {
  return `${flagKey}-${environment}-${pipelineRunId}`;
};

apiVersion: argoproj.io/v1alpha1
kind: AnalysisTemplate
metadata:
  name: payment-service-success
spec:
  metrics:
  - name: success-rate
    successCondition: result >= 0.995
    failureCondition: result < 0.99
    provider:
      prometheus:
        address: http://prometheus:9090
        query: |
          sum(rate(http_requests_total{job="payment-service", status!~"5.."}[5m])) /
          sum(rate(http_requests_total{job="payment-service"}[5m]))

import { circuitBreaker } from '@lifeomic/attempt';

const ldApiCircuitBreaker = circuitBreaker(
  async (config: ProgressiveRolloutConfig, percentage: number) => {
    // LD API call here
  },
  {
    maxConcurrent: 10,
    maxFailures: 5,
    timeout: 5000,
    resetTimeout: 60000,
    fallback: async (config) => {
      // Return last known percentage from Redis
      const redis = await getRedisClient();
      return await redis.get(`ld:rollout:${config.flagKey}`);
    },
  }
);