1# Example GitLab CI pipeline with event-driven stages
2stages:
3  - validate
4  - build
5  - test
6  - security
7  - deploy
8  - verify
9  - promote
10 
11variables:
12  PIPELINE_ID: ${CI_PIPELINE_ID}
13  DEPLOYMENT_STRATEGY: "blue_green"
14 
15validate:
16  stage: validate
17  script:
18    - echo "Pipeline ${PIPELINE_ID} started"
19    - ./scripts/validate-environment.sh
20    - ./scripts/check-dependencies.sh
21...

1import time
2import logging
3from enum import Enum
4from typing import Callable, Any
5 
6class CircuitState(Enum):
7    CLOSED = "closed"
8    OPEN = "open"
9    HALF_OPEN = "half_open"
10 
11class CircuitBreaker:
12    def __init__(self, failure_threshold: int = 5, recovery_timeout: int = 60):
13        self.failure_threshold = failure_threshold
14        self.recovery_timeout = recovery_timeout
15        self.failure_count = 0
16        self.last_failure_time = None
17        self.state = CircuitState.CLOSED
18        self.logger = logging.getLogger(__name__)
19    
20    def call(self, func: Callable, *args, **kwargs) -> Any:
21...

1# Terraform-integrated pipeline stage
2deploy_infrastructure:
3  stage: deploy
4  image: hashicorp/terraform:1.5
5  before_script:
6    - cd infrastructure/
7    - terraform init -backend-config="bucket=${TF_STATE_BUCKET}"
8  script:
9    - terraform plan -var="app_version=${CI_COMMIT_SHA}" -out=tfplan
10    - terraform apply -auto-approve tfplan
11    - terraform output -json > ../terraform-outputs.json
12  artifacts:
13    paths:
14      - terraform-outputs.json
15    expire_in: 1 hour
16  environment:
17    name: production
18    url: https://api.production.company.com
19  rules:
20    - if: '$CI_COMMIT_BRANCH == "main"'
21...

1#!/bin/bash
2# Comprehensive error handling script
3 
4set -euo pipefail
5 
6# Global error handling configuration
7export PIPELINE_ID=${CI_PIPELINE_ID:-$(date +%s)}
8export SLACK_WEBHOOK_URL=${SLACK_WEBHOOK_URL}
9export ROLLBACK_ENABLED=${ROLLBACK_ENABLED:-true}
10 
11# Logging setup
12setup_logging() {
13    exec 1> >(tee -a "/tmp/pipeline-${PIPELINE_ID}.log")
14    exec 2> >(tee -a "/tmp/pipeline-${PIPELINE_ID}.error.log")
15    echo "Pipeline ${PIPELINE_ID} started at $(date)"
16}
17 
18# Error severity classification
19classify_error() {
20    local exit_code=$1
21...

1import time
2import random
3import logging
4from typing import Callable, Any, List
5from dataclasses import dataclass
6 
7@dataclass
8class RetryConfig:
9    max_attempts: int = 3
10    base_delay: float = 1.0
11    max_delay: float = 60.0
12    exponential_base: float = 2.0
13    jitter: bool = True
14    retryable_exceptions: List[type] = None
15 
16class IntelligentRetry:
17    def __init__(self, config: RetryConfig):
18        self.config = config
19        self.logger = logging.getLogger(__name__)
20    
21...

1# Contract testing stage
2contract_tests:
3  stage: test
4  image: pactfoundation/pact-cli:latest
5  services:
6    - name: postgres:13
7      alias: postgres
8  variables:
9    POSTGRES_DB: testdb
10    POSTGRES_USER: testuser
11    POSTGRES_PASSWORD: testpass
12    PACT_BROKER_BASE_URL: https://pact-broker.company.com
13  script:
14    # Provider contract testing
15    - echo "Running provider contract tests"
16    - pact-provider-verifier \
17        --provider-base-url=http://localhost:8080 \
18        --pact-broker-base-url=$PACT_BROKER_BASE_URL \
19        --broker-token=$PACT_BROKER_TOKEN \
20        --provider="user-service" \
21...

1#!/usr/bin/env python3
2"""
3Chaos engineering integration for CI/CD pipelines
4"""
5 
6import random
7import time
8import subprocess
9import logging
10from typing import Dict, Any
11 
12class ChaosExperiment:
13    def __init__(self, name: str, probability: float = 0.1):
14        self.name = name
15        self.probability = probability
16        self.logger = logging.getLogger(__name__)
17    
18    def should_run(self) -> bool:
19        """Determine if chaos experiment should run based on probability"""
20        return random.random() < self.probability
21...

1"""
2Comprehensive pipeline metrics collection
3"""
4 
5import time
6import json
7import requests
8import logging
9from dataclasses import dataclass, asdict
10from typing import Dict, Any, Optional
11from contextlib import contextmanager
12 
13@dataclass
14class PipelineMetrics:
15    pipeline_id: str
16    stage: str
17    start_time: float
18    end_time: Optional[float] = None
19    status: str = "running"
20    error_message: Optional[str] = None
21...