Cross-Environment Configuration Drift Detection & Drift-Proof Deployment Best Practices¶

# scripts/detect_infrastructure_drift.py
import subprocess
import yaml
from typing import Dict, List

class InfrastructureDriftDetector:
    def __init__(self, expected_config: Dict):
        self.expected = expected_config

    def detect_kernel_drift(self, node: str) -> bool:
        """Detect kernel version drift"""
        result = subprocess.run(
            ["ssh", node, "uname -r"],
            capture_output=True,
            text=True
        )
        actual_kernel = result.stdout.strip()
        expected_kernel = self.expected["nodes"][node]["kernel"]

        return actual_kernel != expected_kernel

    def detect_nvidia_driver_drift(self, node: str) -> bool:
        """Detect NVIDIA driver version drift"""
        result = subprocess.run(
            ["ssh", node, "nvidia-smi --query-gpu=driver_version --format=csv,noheader"],
            capture_output=True,
            text=True
        )
        actual_driver = result.stdout.strip()
        expected_driver = self.expected["nodes"][node]["nvidia_driver"]

        return actual_driver != expected_driver

    def detect_ulimit_drift(self, node: str) -> Dict[str, bool]:
        """Detect ulimit drift"""
        result = subprocess.run(
            ["ssh", node, "ulimit -n"],
            capture_output=True,
            text=True
        )
        actual_nofile = int(result.stdout.strip())
        expected_nofile = self.expected["nodes"][node]["ulimits"]["nofile"]

        return {
            "nofile": actual_nofile != expected_nofile,
            "actual": actual_nofile,
            "expected": expected_nofile
        }

    def detect_sysctl_drift(self, node: str) -> Dict[str, bool]:
        """Detect sysctl drift"""
        drifts = {}

        for key, expected_value in self.expected["nodes"][node]["sysctls"].items():
            result = subprocess.run(
                ["ssh", node, f"sysctl -n {key}"],
                capture_output=True,
                text=True
            )
            actual_value = int(result.stdout.strip())
            drifts[key] = actual_value != expected_value

        return drifts

# scripts/detect_sbom_drift.py
import json
from syft import syft
from deepdiff import DeepDiff

class SBOMDriftDetector:
    def __init__(self, expected_sbom_path: str):
        with open(expected_sbom_path) as f:
            self.expected_sbom = json.load(f)

    def detect_drift(self, image: str) -> Dict:
        """Detect SBOM drift for container image"""
        # Generate SBOM for current image
        actual_sbom = syft.scan(image, output="json")
        actual_data = json.loads(actual_sbom)

        # Compare packages
        expected_packages = {
            pkg["name"]: pkg["version"]
            for pkg in self.expected_sbom.get("artifacts", [])
        }

        actual_packages = {
            pkg["name"]: pkg["version"]
            for pkg in actual_data.get("artifacts", [])
        }

        # Find drift
        drift = DeepDiff(expected_packages, actual_packages)

        return {
            "has_drift": bool(drift),
            "added": drift.get("dictionary_item_added", []),
            "removed": drift.get("dictionary_item_removed", []),
            "changed": drift.get("values_changed", {})
        }

# scripts/detect_cuda_drift.py
import subprocess
import re

def detect_cuda_drift(expected_cuda: str, container: str) -> Dict:
    """Detect CUDA version drift in container"""
    # Check container CUDA version
    result = subprocess.run(
        ["docker", "exec", container, "nvcc", "--version"],
        capture_output=True,
        text=True
    )

    match = re.search(r"release (\d+\.\d+)", result.stdout)
    if match:
        actual_cuda = match.group(1)

        return {
            "has_drift": actual_cuda != expected_cuda,
            "expected": expected_cuda,
            "actual": actual_cuda
        }

    return {"has_drift": True, "error": "CUDA not found"}

# requirements-expected.txt
fastapi==0.104.1
pydantic==2.5.0
uvicorn==0.24.0

# requirements-actual.txt (drifted)
fastapi==0.104.1
pydantic==2.4.0        # Drift! Downgraded
uvicorn==0.25.0        # Drift! Upgraded

# scripts/detect_python_drift.py
import subprocess
import json
from packaging import version

class PythonDriftDetector:
    def detect_drift(self, expected_requirements: str, container: str) -> Dict:
        """Detect Python package drift"""
        # Get installed packages
        result = subprocess.run(
            ["docker", "exec", container, "pip", "freeze"],
            capture_output=True,
            text=True
        )

        actual_packages = {}
        for line in result.stdout.strip().split("\n"):
            if "==" in line:
                name, ver = line.split("==")
                actual_packages[name.lower()] = ver

        # Load expected requirements
        expected_packages = {}
        with open(expected_requirements) as f:
            for line in f:
                if "==" in line:
                    name, ver = line.strip().split("==")
                    expected_packages[name.lower()] = ver

        # Compare
        drift = {}
        for pkg, expected_ver in expected_packages.items():
            if pkg not in actual_packages:
                drift[pkg] = {"status": "missing", "expected": expected_ver}
            elif actual_packages[pkg] != expected_ver:
                drift[pkg] = {
                    "status": "version_mismatch",
                    "expected": expected_ver,
                    "actual": actual_packages[pkg]
                }

        return drift

# nicegui/detect_client_drift.py
class NiceGUIDriftDetector:
    def detect_asset_drift(self, expected_assets: List[str]) -> Dict:
        """Detect client-side asset drift"""
        # Check if assets match expected versions
        from nicegui import app

        actual_assets = []
        for route in app.routes:
            if route.path.endswith(('.js', '.css', '.html')):
                actual_assets.append(route.path)

        expected_set = set(expected_assets)
        actual_set = set(actual_assets)

        return {
            "missing": list(expected_set - actual_set),
            "extra": list(actual_set - expected_set),
            "has_drift": expected_set != actual_set
        }

# scripts/detect_postgres_drift.py
import psycopg2
from schema_diff import compare_schemas

class PostgresDriftDetector:
    def __init__(self, dev_conn, prod_conn):
        self.dev = dev_conn
        self.prod = prod_conn

    def detect_schema_drift(self) -> Dict:
        """Detect schema drift between dev and prod"""
        dev_schema = self.get_schema(self.dev)
        prod_schema = self.get_schema(self.prod)

        diff = compare_schemas(dev_schema, prod_schema)

        return {
            "has_drift": bool(diff),
            "tables_added": diff.get("tables_added", []),
            "tables_removed": diff.get("tables_removed", []),
            "columns_added": diff.get("columns_added", []),
            "columns_removed": diff.get("columns_removed", []),
            "indexes_added": diff.get("indexes_added", []),
            "indexes_removed": diff.get("indexes_removed", [])
        }

    def get_schema(self, conn) -> Dict:
        """Get complete database schema"""
        with conn.cursor() as cur:
            # Get tables
            cur.execute("""
                SELECT 
                    schemaname,
                    tablename,
                    columnname,
                    datatype,
                    is_nullable
                FROM pg_catalog.pg_table_def
                WHERE schemaname NOT IN ('pg_catalog', 'information_schema')
                ORDER BY schemaname, tablename, columnname
            """)

            schema = {}
            for row in cur.fetchall():
                schema_name, table_name, col_name, data_type, nullable = row
                key = f"{schema_name}.{table_name}"
                if key not in schema:
                    schema[key] = {}
                schema[key][col_name] = {
                    "type": data_type,
                    "nullable": nullable == "YES"
                }

            return schema

# scripts/detect_lakehouse_drift.py
import pyiceberg
from pyiceberg.catalog import load_catalog

class LakehouseDriftDetector:
    def detect_iceberg_drift(self, table_name: str, expected_manifest: str) -> Dict:
        """Detect Iceberg table metadata drift"""
        catalog = load_catalog("s3", **catalog_config)
        table = catalog.load_table(table_name)

        # Get current manifest
        current_manifest = table.metadata_location()

        # Compare with expected
        with open(expected_manifest) as f:
            expected_data = json.load(f)

        current_data = table.metadata()

        drift = {
            "schema_drift": current_data.schema() != expected_data["schema"],
            "partition_drift": current_data.partition_spec() != expected_data["partition_spec"],
            "properties_drift": current_data.properties != expected_data["properties"]
        }

        return drift

# scripts/detect_geoparquet_drift.py
import geopandas as gpd
import pyarrow.parquet as pq

class GeoParquetDriftDetector:
    def detect_schema_drift(self, expected_path: str, actual_path: str) -> Dict:
        """Detect GeoParquet schema drift"""
        # Read expected schema
        expected_schema = pq.read_schema(expected_path)

        # Read actual schema
        actual_schema = pq.read_schema(actual_path)

        # Compare
        expected_fields = {f.name: str(f.type) for f in expected_schema}
        actual_fields = {f.name: str(f.type) for f in actual_schema}

        drift = {
            "fields_added": set(actual_fields) - set(expected_fields),
            "fields_removed": set(expected_fields) - set(actual_fields),
            "type_changes": {
                field: {"expected": expected_fields[field], "actual": actual_fields[field]}
                for field in set(expected_fields) & set(actual_fields)
                if expected_fields[field] != actual_fields[field]
            }
        }

        return drift

# scripts/detect_mlflow_drift.py
import mlflow
from mlflow.tracking import MlflowClient

class MLflowDriftDetector:
    def detect_signature_drift(self, model_name: str, expected_version: int) -> Dict:
        """Detect MLflow model signature drift"""
        client = MlflowClient()

        # Get expected model
        expected_model = client.get_model_version(model_name, expected_version)
        expected_signature = expected_model.signature

        # Get latest model
        latest_model = client.get_latest_versions(model_name, stages=["Production"])[0]
        latest_signature = latest_model.signature

        # Compare signatures
        drift = {
            "has_drift": expected_signature != latest_signature,
            "expected_inputs": expected_signature.inputs.to_dict() if expected_signature else None,
            "actual_inputs": latest_signature.inputs.to_dict() if latest_signature else None,
            "expected_outputs": expected_signature.outputs.to_dict() if expected_signature else None,
            "actual_outputs": latest_signature.outputs.to_dict() if latest_signature else None
        }

        return drift

# scripts/detect_etl_contract_drift.py
from prefect import flow
import json

class ETLContractDriftDetector:
    def detect_contract_drift(self, flow_name: str, expected_contract: Dict) -> Dict:
        """Detect ETL contract drift"""
        # Get actual flow definition
        from prefect.client import get_client
        client = get_client()

        flow_def = client.read_flow_by_name(flow_name)

        # Compare contracts
        actual_contract = {
            "inputs": flow_def.inputs,
            "outputs": flow_def.outputs,
            "transformations": flow_def.transformations
        }

        drift = {
            "has_drift": actual_contract != expected_contract,
            "input_drift": actual_contract["inputs"] != expected_contract["inputs"],
            "output_drift": actual_contract["outputs"] != expected_contract["outputs"],
            "transformation_drift": actual_contract["transformations"] != expected_contract["transformations"]
        }

        return drift

# scripts/validate_helm_values.py
import yaml
import jsonschema

class HelmValuesValidator:
    def __init__(self, schema_path: str):
        with open(schema_path) as f:
            self.schema = yaml.safe_load(f)

    def validate(self, values_path: str) -> Dict:
        """Validate Helm values against schema"""
        with open(values_path) as f:
            values = yaml.safe_load(f)

        try:
            jsonschema.validate(instance=values, schema=self.schema)
            return {"valid": True}
        except jsonschema.ValidationError as e:
            return {
                "valid": False,
                "error": str(e),
                "path": list(e.path)
            }

# scripts/enforce_schema_contract.py
from alembic import config, script
from alembic.runtime import migration
import sqlalchemy

class SchemaContractEnforcer:
    def __init__(self, db_url: str, alembic_cfg_path: str):
        self.engine = sqlalchemy.create_engine(db_url)
        self.alembic_cfg = config.Config(alembic_cfg_path)

    def enforce_contract(self) -> bool:
        """Enforce schema contract - fail if drift detected"""
        # Get current database revision
        with self.engine.connect() as conn:
            context = migration.MigrationContext.configure(conn)
            current_rev = context.get_current_revision()

        # Get expected revision from Alembic
        script_dir = script.ScriptDirectory.from_config(self.alembic_cfg)
        head_rev = script_dir.get_current_head()

        if current_rev != head_rev:
            raise ValueError(
                f"Schema drift detected! "
                f"Expected: {head_rev}, Actual: {current_rev}"
            )

        return True

# scripts/detect_redis_drift.py
import redis
import hashlib
import json

class RedisDriftDetector:
    def __init__(self, redis_client: redis.Redis, expected_config_path: str):
        self.redis = redis_client
        with open(expected_config_path) as f:
            self.expected_config = json.load(f)

    def detect_drift(self) -> Dict:
        """Detect Redis configuration drift"""
        # Get actual config
        actual_config = self.redis.config_get()

        # Compare checksums
        expected_hash = hashlib.sha256(
            json.dumps(self.expected_config, sort_keys=True).encode()
        ).hexdigest()

        actual_hash = hashlib.sha256(
            json.dumps(actual_config, sort_keys=True).encode()
        ).hexdigest()

        # Find differences
        drift = {}
        for key in set(self.expected_config.keys()) | set(actual_config.keys()):
            expected_val = self.expected_config.get(key)
            actual_val = actual_config.get(key)

            if expected_val != actual_val:
                drift[key] = {
                    "expected": expected_val,
                    "actual": actual_val
                }

        return {
            "has_drift": expected_hash != actual_hash,
            "differences": drift
        }

# scripts/detect_yaml_drift.py
import yaml
import hashlib
from deepdiff import DeepDiff

class YAMLDriftDetector:
    def detect_drift(self, expected_path: str, actual_path: str) -> Dict:
        """Detect YAML manifest drift"""
        with open(expected_path) as f:
            expected = yaml.safe_load(f)

        with open(actual_path) as f:
            actual = yaml.safe_load(f)

        # Compare structure
        diff = DeepDiff(expected, actual, ignore_order=True)

        return {
            "has_drift": bool(diff),
            "added": diff.get("dictionary_item_added", []),
            "removed": diff.get("dictionary_item_removed", []),
            "changed": diff.get("values_changed", {}),
            "type_changes": diff.get("type_changes", {})
        }

# scripts/detect_prefect_drift.py
from prefect.blocks.core import Block
from prefect.client import get_client

class PrefectDriftDetector:
    def detect_block_drift(self, block_name: str, expected_config: Dict) -> Dict:
        """Detect Prefect block configuration drift"""
        client = get_client()

        # Get actual block
        block = Block.load(block_name)
        actual_config = block.dict()

        # Compare
        drift = {}
        for key, expected_val in expected_config.items():
            actual_val = actual_config.get(key)
            if actual_val != expected_val:
                drift[key] = {
                    "expected": expected_val,
                    "actual": actual_val
                }

        return {
            "has_drift": bool(drift),
            "differences": drift
        }

# monitoring/drift_metrics.py
from prometheus_client import Gauge, Counter

# Drift detection metrics
drift_detected = Gauge(
    'config_drift_detected',
    'Configuration drift detected',
    ['environment', 'component', 'drift_type']
)

drift_detection_errors = Counter(
    'drift_detection_errors_total',
    'Drift detection errors',
    ['environment', 'component']
)

drift_remediation_attempts = Counter(
    'drift_remediation_attempts_total',
    'Drift remediation attempts',
    ['environment', 'component', 'status']
)