devops-expert

name	devops-expert
version	1.0.0
description	Expert-level DevOps practices, culture, automation, and continuous delivery
category	devops
tags	["devops","ci-cd","automation","infrastructure","culture"]
allowed-tools	["Read","Write","Edit","Bash(*)"]

DevOps Expert

Expert guidance for DevOps practices, culture, CI/CD pipelines, infrastructure automation, and operational excellence.

Core Concepts

DevOps Culture

• Collaboration and communication
• Shared responsibility
• Continuous improvement
• Breaking down silos
• Blameless culture
• Measuring everything

Automation

• Infrastructure as Code (IaC)
• Configuration management
• Deployment automation
• Testing automation
• Monitoring automation
• Self-service platforms

CI/CD

• Continuous Integration
• Continuous Delivery
• Continuous Deployment
• Pipeline as Code
• Artifact management
• Release strategies

CI/CD Pipeline

# GitHub Actions Example
name: CI/CD Pipeline

on:
  push:
    branches: [main, develop]
  pull_request:
    branches: [main]

env:
  REGISTRY: ghcr.io
  IMAGE_NAME: ${{ github.repository }}

jobs:
  test:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v3

      - name: Setup Node.js
        uses: actions/setup-node@v3
        with:
          node-version: '18'
          cache: 'npm'

      - name: Install dependencies
        run: npm ci

      - name: Run linting
        run: npm run lint

      - name: Run tests
          

          
          

         
         

  
     
     
    
       
       

    
        

            
         
        
             
             
             

         
         
         
        
               

            
         
        
           
           
             
             

  
     
       
     
     

    
          
         
          
              

          
           

  
     
       
     
     

    
          
         
          
              

Infrastructure as Code

# Pulumi Infrastructure as Code
import pulumi
import pulumi_aws as aws

# VPC
vpc = aws.ec2.Vpc("main-vpc",
    cidr_block="10.0.0.0/16",
    enable_dns_hostnames=True,
    enable_dns_support=True,
    tags={"Name": "main-vpc"})

# Subnets
public_subnet = aws.ec2.Subnet("public-subnet",
    vpc_id=vpc.id,
    cidr_block="10.0.1.0/24",
    availability_zone="us-east-1a",
    map_public_ip_on_launch=True,
    tags={"Name": "public-subnet"})

private_subnet = aws.ec2.Subnet("private-subnet",
    vpc_id=vpc.id,
    cidr_block="10.0.2.0/24",
    availability_zone="us-east-1b",
    tags={"Name": "private-subnet"})

# Internet Gateway
igw = aws.ec2.InternetGateway("igw",
    vpc_id=vpc.id,
    tags={"Name": "main-igw"})

# Route Table
route_table = aws.ec2.RouteTable("public-rt",
    vpc_id=vpc.id,
    routes=[
        aws.ec2.RouteTableRouteArgs(
            cidr_block="0.0.0.0/0",
            gateway_id=igw.id,
        )
    ],
    tags={"Name": "public-rt"})

# Security Group
security_group = aws.ec2.SecurityGroup("web-sg",
    vpc_id=vpc.id,
    description="Allow HTTP and HTTPS traffic",
    ingress=[
        aws.ec2.SecurityGroupIngressArgs(
            protocol="tcp",
            from_port=,
            to_port=,
            cidr_blocks=[],
        ),
        aws.ec2.SecurityGroupIngressArgs(
            protocol=,
            from_port=,
            to_port=,
            cidr_blocks=[],
        ),
    ],
    egress=[
        aws.ec2.SecurityGroupEgressArgs(
            protocol=,
            from_port=,
            to_port=,
            cidr_blocks=[],
        )
    ])


cluster = aws.eks.Cluster(,
    role_arn=cluster_role.arn,
    vpc_config=aws.eks.ClusterVpcConfigArgs(
        subnet_ids=[public_subnet., private_subnet.],
        security_group_ids=[security_group.],
    ))


pulumi.export(, vpc.)
pulumi.export(, cluster.name)
pulumi.export(, cluster.endpoint)

Deployment Strategies

from typing import List, Dict
import time

class DeploymentStrategy:
    """Implement various deployment strategies"""

    def __init__(self, service_name: str):
        self.service_name = service_name

    def blue_green_deployment(self, blue_version: str, green_version: str):
        """Blue-Green deployment"""
        # Deploy green environment
        self.deploy_environment("green", green_version)

        # Run tests on green
        if self.run_tests("green"):
            # Switch traffic to green
            self.switch_traffic("green")

            # Keep blue for rollback
            print(f"Deployment successful. Blue ({blue_version}) kept for rollback.")
        else:
            # Rollback - keep blue active
            print("Tests failed on green. Keeping blue active.")

    def canary_deployment(self, current_version: str, new_version: str,
                         canary_percentage: int = 10):
        """Canary deployment"""
        
        .deploy_canary(new_version, canary_percentage)

        
        metrics = .monitor_canary_metrics(duration_minutes=)

         metrics[] <   metrics[] < :
            
             percentage  [, , , ]:
                .update_canary_traffic(percentage)
                time.sleep()  

                  .check_health():
                    .rollback(current_version)
                     

            ()
             
        :
            .rollback(current_version)
            ()
             

     ():
        
        instances = .get_instances()

         i  (, (instances), batch_size):
            batch = instances[i:i + batch_size]

            
             instance  batch:
                .update_instance(instance, version)
                .wait_for_healthy(instance)

            
              .check_health():
                ()
                 

        ()
         

     ():
        
         {
            : feature_name,
            : enabled,
            : rollout_percentage,
            : {
                : []  rollout_percentage <   []
            }
        }

Configuration Management

from typing import Dict, Any
import yaml
import json

class ConfigurationManager:
    """Manage application configuration"""

    def __init__(self, environment: str):
        self.environment = environment
        self.config = {}

    def load_config(self, config_file: str):
        """Load configuration from file"""
        with open(config_file, 'r') as f:
            if config_file.endswith('.yaml') or config_file.endswith('.yml'):
                self.config = yaml.safe_load(f)
            elif config_file.endswith('.json'):
                self.config = json.load(f)

    def get(self, key: str, default: Any = None) -> Any:
        """Get configuration value"""
        keys = key.split('.')
        value = self.config

        for k in keys:
            if isinstance(value, dict):
                value = value.get(k)
            :
                 default

             value  :
                 default

         value

     ():
        
        .config = ._deep_merge(.config, env_config)

     () -> :
        
        result = base.copy()

         key, value  override.items():
             key  result  (result[key], )  (value, ):
                result[key] = ._deep_merge(result[key], value)
            :
                result[key] = value

         result

     () -> []:
        
        missing = []

         key  required_keys:
             .get(key)  :
                missing.append(key)

         missing

Monitoring and Observability

import logging
from opencensus.ext.azure import metrics_exporter
from opencensus.stats import aggregation as aggregation_module
from opencensus.stats import measure as measure_module
from opencensus.stats import stats as stats_module
from opencensus.stats import view as view_module
from opencensus.tags import tag_map as tag_map_module

class ObservabilityStack:
    """Implement observability best practices"""

    def __init__(self):
        self.logger = self._setup_logging()
        self.stats = stats_module.stats
        self.view_manager = self.stats.view_manager

    def _setup_logging(self) -> logging.Logger:
        """Setup structured logging"""
        logger = logging.getLogger(__name__)
        handler = logging.StreamHandler()

        formatter = logging.Formatter(
            '{"time": "%(asctime)s", "level": "%(levelname)s", '
            '"service": "%(name)s", "message": "%(message)s"}'
        )
        handler.setFormatter(formatter)
        logger.addHandler(handler)
        logger.setLevel(logging.INFO)

        return logger

    def log_with_context(self, level: str, message: str, **context):
        
        log_func = (.logger, level)
        log_func(message, extra=context)

     ():
        
        
        

     ():
        
        
        

Best Practices

Culture & Process

• Foster collaboration between Dev and Ops
• Automate everything possible
• Measure and monitor continuously
• Practice blameless post-mortems
• Share knowledge and documentation
• Encourage experimentation
• Celebrate successes and learn from failures

CI/CD

• Keep builds fast (<10 minutes)
• Run tests in parallel
• Use pipeline as code
• Implement automated rollbacks
• Require code review before merge
• Use trunk-based development
• Deploy small, frequent changes

Infrastructure

• Use Infrastructure as Code
• Version everything (code, config, infrastructure)
• Implement disaster recovery
• Practice chaos engineering
• Use immutable infrastructure
• Automate security scanning
• Monitor cloud costs

Anti-Patterns

❌ Manual deployments ❌ Configuration drift ❌ No automated testing ❌ Long-lived feature branches ❌ Blame culture ❌ Siloed teams ❌ Ignoring technical debt

Resources

• The Phoenix Project: https://itrevolution.com/the-phoenix-project/
• DevOps Handbook: https://itrevolution.com/the-devops-handbook/
• State of DevOps Report: https://www.devops-research.com/research.html
• GitLab CI/CD: https://docs.gitlab.com/ee/ci/
• GitHub Actions: https://docs.github.com/en/actions