# aiops-anomaly-detection.py
import numpy as np
import pandas as pd
from sklearn.ensemble import IsolationForest
from sklearn.preprocessing import StandardScaler
import matplotlib.pyplot as plt
from datetime import datetime, timedelta
import json

class AIOpsAnomalyDetector:
    def __init__(self):
        self.model = IsolationForest(contamination=0.1, random_state=42)
        self.scaler = StandardScaler()
        self.is_trained = False
        
    def prepare_data(self, metrics_data):
        """
        Prepare metrics data for anomaly detection
        Expected format: List of dictionaries with timestamp and metrics
        """
        df = pd.DataFrame(metrics_data)
        
        # Extract numerical features for anomaly detection
        feature_columns = ['cpu_usage', 'memory_usage', 'disk_usage', 
                          'network_in', 'network_out', 'request_rate', 'error_rate']
        
        # Handle missing values
        df[feature_columns] = df[feature_columns].fillna(df[feature_columns].mean())
        
        return df[feature_columns]
    
    def train(self, historical_metrics):
        """
        Train the anomaly detection model on historical data
        """
        features = self.prepare_data(historical_metrics)
        scaled_features = self.scaler.fit_transform(features)
        
        # Fit the isolation forest model
        self.model.fit(scaled_features)
        self.is_trained = True
        
        print(f"Model trained on {len(historical_metrics)} data points")
        return True
    
    def detect_anomalies(self, current_metrics):
        """
        Detect anomalies in current metrics
        """
        if not self.is_trained:
            raise ValueError("Model must be trained before detecting anomalies")
        
        features = self.prepare_data(current_metrics)
        scaled_features = self.scaler.transform(features)
        
        # Predict anomalies (-1 for anomaly, 1 for normal)
        predictions = self.model.predict(scaled_features)
        anomaly_scores = self.model.decision_function(scaled_features)
        
        # Prepare results
        results = []
        for i, (pred, score) in enumerate(zip(predictions, anomaly_scores)):
            is_anomaly = pred == -1
            results.append({
                'timestamp': current_metrics[i]['timestamp'],
                'is_anomaly': is_anomaly,
                'anomaly_score': float(score),
                'metrics': current_metrics[i],
                'severity': self._calculate_severity(score, is_anomaly)
            })
        
        return results
    
    def _calculate_severity(self, score, is_anomaly):
        """
        Calculate severity based on anomaly score
        """
        if not is_anomaly:
            return 'normal'
        
        abs_score = abs(score)
        if abs_score > 0.5:
            return 'critical'
        elif abs_score > 0.3:
            return 'high'
        elif abs_score > 0.1:
            return 'medium'
        else:
            return 'low'

# Example usage
def simulate_metrics_data(start_time, num_points=100):
    """
    Simulate realistic metrics data for demonstration
    """
    timestamps = [start_time + timedelta(minutes=i) for i in range(num_points)]
    
    data = []
    for i, ts in enumerate(timestamps):
        # Normal operating conditions
        base_cpu = np.random.normal(40, 10)
        base_memory = np.random.normal(50, 15)
        base_disk = np.random.normal(60, 10)
        
        # Add some anomalous periods
        if 30 <= i <= 35 or 70 <= i <= 75:
            base_cpu += np.random.normal(30, 5)  # High CPU usage
            base_memory += np.random.normal(20, 8)  # High memory usage
        
        data.append({
            'timestamp': ts.isoformat(),
            'cpu_usage': max(0, min(100, base_cpu)),
            'memory_usage': max(0, min(100, base_memory)),
            'disk_usage': max(0, min(100, base_disk)),
            'network_in': np.random.exponential(100),
            'network_out': np.random.exponential(80),
            'request_rate': np.random.poisson(100),
            'error_rate': np.random.uniform(0, 5)
        })
    
    return data

# Example implementation
detector = AIOpsAnomalyDetector()

# Simulate historical data for training
historical_data = simulate_metrics_data(datetime.now() - timedelta(days=7), 1000)
detector.train(historical_data)

# Simulate current data for anomaly detection
current_data = simulate_metrics_data(datetime.now(), 100)
anomalies = detector.detect_anomalies(current_data)

# Print results
anomaly_count = sum(1 for a in anomalies if a['is_anomaly'])
print(f"Detected {anomaly_count} anomalies out of {len(anomalies)} samples")

# aiops-incident-response.py
import asyncio
import openai
from typing import Dict, List, Optional
import json
from dataclasses import dataclass

@dataclass
class Incident:
    id: str
    title: str
    description: str
    severity: str
    timestamp: str
    affected_services: List[str]
    metrics_data: Dict
    logs: List[str]

class AIOpsIncidentResponder:
    def __init__(self, openai_api_key: str):
        openai.api_key = openai_api_key
        self.knowledge_base = self._load_knowledge_base()
        
    def _load_knowledge_base(self) -> Dict:
        """
        Load incident response knowledge base
        In practice, this would come from a database or external system
        """
        return {
            "high_cpu_usage": {
                "root_causes": [
                    "Inefficient algorithms",
                    "Memory leaks causing GC pressure",
                    "Unexpected traffic spikes",
                    "Third-party service dependencies"
                ],
                "diagnosis_steps": [
                    "Check application logs for errors",
                    "Analyze heap dumps",
                    "Review recent deployments",
                    "Monitor database queries"
                ],
                "remediation_steps": [
                    "Scale up resources temporarily",
                    "Identify and optimize hot code paths",
                    "Implement circuit breakers",
                    "Roll back recent changes if necessary"
                ]
            },
            "high_error_rate": {
                "root_causes": [
                    "Application bugs",
                    "Database connection issues",
                    "Third-party API failures",
                    "Configuration errors"
                ],
                "diagnosis_steps": [
                    "Check error logs and stack traces",
                    "Review application health checks",
                    "Monitor downstream dependencies",
                    "Validate configuration settings"
                ],
                "remediation_steps": [
                    "Implement retry logic with exponential backoff",
                    "Scale database connections",
                    "Fix application bugs",
                    "Roll back configuration changes"
                ]
            }
        }
    
    async def analyze_incident(self, incident: Incident) -> Dict:
        """
        Analyze incident using AI to suggest root causes and remediation
        """
        prompt = f"""
        Analyze this incident and provide recommendations:

        Incident: {incident.title}
        Description: {incident.description}
        Severity: {incident.severity}
        Affected Services: {', '.join(incident.affected_services)}
        
        Metrics: {json.dumps(incident.metrics_data, indent=2)}
        Recent Logs: {incident.logs[:10]}  # First 10 log entries
        
        Please provide:
        1. Most likely root causes
        2. Recommended diagnosis steps
        3. Immediate remediation actions
        4. Priority level for response
        5. Estimated time to resolution
        """
        
        try:
            response = await openai.ChatCompletion.acreate(
                model="gpt-4",
                messages=[
                    {"role": "system", "content": "You are an experienced DevOps engineer helping diagnose incidents."},
                    {"role": "user", "content": prompt}
                ],
                temperature=0.3,
                max_tokens=1000
            )
            
            analysis = response.choices[0].message.content
            
            # Parse AI response and structure it
            structured_analysis = self._parse_ai_response(analysis, incident)
            
            return structured_analysis
            
        except Exception as e:
            print(f"AI analysis failed: {str(e)}")
            return self._fallback_analysis(incident)
    
    def _parse_ai_response(self, ai_response: str, incident: Incident) -> Dict:
        """
        Parse AI response and structure it into a usable format
        """
        # This would be more sophisticated in production
        # For now, we'll create a basic structure
        return {
            "incident_id": incident.id,
            "ai_analysis": ai_response,
            "confidence_score": 0.8,  # Placeholder
            "recommended_actions": [
                "Review recent deployments",
                "Check application logs",
                "Monitor resource utilization"
            ],
            "estimated_resolution_time": "2-4 hours",
            "priority": "high" if incident.severity.lower() in ['critical', 'high'] else "medium"
        }
    
    def _fallback_analysis(self, incident: Incident) -> Dict:
        """
        Fallback analysis when AI is unavailable
        """
        # Use knowledge base for analysis
        if "cpu" in incident.title.lower() or "cpu" in incident.description.lower():
            knowledge = self.knowledge_base.get("high_cpu_usage", {})
        elif "error" in incident.title.lower() or "error" in incident.description.lower():
            knowledge = self.knowledge_base.get("high_error_rate", {})
        else:
            knowledge = {}
        
        return {
            "incident_id": incident.id,
            "ai_analysis": "AI unavailable, using knowledge base",
            "confidence_score": 0.6,
            "recommended_actions": knowledge.get("remediation_steps", ["Manual investigation required"]),
            "estimated_resolution_time": "4-8 hours",
            "priority": incident.severity.lower()
        }

# Example usage
async def main():
    responder = AIOpsIncidentResponder(openai_api_key="your-api-key-here")
    
    # Create a sample incident
    sample_incident = Incident(
        id="INC-12345",
        title="High CPU Usage on Web Service",
        description="CPU usage has exceeded 90% for the past 10 minutes",
        severity="high",
        timestamp="2024-01-08T10:30:00Z",
        affected_services=["web-service", "api-gateway"],
        metrics_data={
            "cpu_avg": 92.5,
            "memory_avg": 65.2,
            "request_rate": 1200,
            "error_rate": 3.2
        },
        logs=[
            "2024-01-08T10:25:00Z ERROR: Database connection timeout",
            "2024-01-08T10:26:00Z WARN: High GC pressure detected",
            "2024-01-08T10:27:00Z ERROR: Thread pool exhausted"
        ]
    )
    
    analysis = await responder.analyze_incident(sample_incident)
    print(json.dumps(analysis, indent=2))

if __name__ == "__main__":
    asyncio.run(main())

# ml-test-generation.py
import openai
from typing import List, Dict, Any
import json
from dataclasses import dataclass

@dataclass
class CodeAnalysis:
    functions: List[Dict]
    classes: List[Dict]
    dependencies: List[str]
    complexity_metrics: Dict

class MLTestGenerator:
    def __init__(self, openai_api_key: str):
        openai.api_key = openai_api_key
    
    async def analyze_code(self, source_code: str, language: str = "python") -> CodeAnalysis:
        """
        Analyze source code to understand structure and generate tests
        """
        prompt = f"""
        Analyze the following {language} code and provide:
        
        1. List of functions with parameters and return types
        2. List of classes with methods and properties
        3. External dependencies used
        4. Complexity metrics (cyclomatic complexity, etc.)
        
        Code:
        {source_code}
        
        Return the analysis in JSON format.
        """
        
        response = await openai.ChatCompletion.acreate(
            model="gpt-4",
            messages=[
                {"role": "system", "content": "You are an expert code analyzer."},
                {"role": "user", "content": prompt}
            ],
            temperature=0.1,
            max_tokens=1000
        )
        
        analysis_data = json.loads(response.choices[0].message.content)
        
        return CodeAnalysis(
            functions=analysis_data.get("functions", []),
            classes=analysis_data.get("classes", []),
            dependencies=analysis_data.get("dependencies", []),
            complexity_metrics=analysis_data.get("complexity_metrics", {})
        )
    
    async def generate_tests(self, source_code: str, analysis: CodeAnalysis) -> str:
        """
        Generate comprehensive tests based on code analysis
        """
        prompt = f"""
        Generate comprehensive unit tests for the following code:
        
        Source Code:
        {source_code}
        
        Code Analysis:
        {json.dumps({
            'functions': analysis.functions,
            'classes': analysis.classes,
            'dependencies': analysis.dependencies,
            'complexity_metrics': analysis.complexity_metrics
        }, indent=2)}
        
        Generate tests that cover:
        1. Happy path scenarios
        2. Edge cases and boundary conditions
        3. Error handling and exception cases
        4. Performance considerations if applicable
        5. Security considerations if applicable
        
        Use appropriate testing frameworks (pytest for Python, Jest for JavaScript, etc.)
        Include proper mocking for external dependencies.
        """
        
        response = await openai.ChatCompletion.acreate(
            model="gpt-4",
            messages=[
                {"role": "system", "content": "You are an expert test engineer generating comprehensive unit tests."},
                {"role": "user", "content": prompt}
            ],
            temperature=0.3,
            max_tokens=2000
        )
        
        return response.choices[0].message.content

# Example usage with a sample function
sample_code = '''
def calculate_discount(price: float, discount_percent: float, customer_type: str) -> float:
    """
    Calculate discount amount based on price, discount percentage, and customer type.
    
    Args:
        price: Original price of the item
        discount_percent: Discount percentage (0-100)
        customer_type: Type of customer ('regular', 'premium', 'vip')
    
    Returns:
        Discount amount to be subtracted from original price
    
    Raises:
        ValueError: If price is negative, discount is invalid, or customer type is unknown
    """
    if price < 0:
        raise ValueError("Price cannot be negative")
    
    if not 0 <= discount_percent <= 100:
        raise ValueError("Discount percent must be between 0 and 100")
    
    if customer_type not in ['regular', 'premium', 'vip']:
        raise ValueError("Unknown customer type")
    
    base_discount = price * (discount_percent / 100)
    
    # Additional discounts based on customer type
    if customer_type == 'premium':
        base_discount *= 1.1  # 10% additional for premium
    elif customer_type == 'vip':
        base_discount *= 1.2  # 20% additional for VIP
    
    # Ensure discount doesn't exceed original price
    return min(base_discount, price)
'''

async def generate_discount_tests():
    generator = MLTestGenerator(openai_api_key="your-api-key-here")
    analysis = await generator.analyze_code(sample_code)
    tests = await generator.generate_tests(sample_code, analysis)
    
    print("Generated Tests:")
    print(tests)

# Run the example
# asyncio.run(generate_discount_tests())

// packages/backend/src/plugins/catalog.ts
import { CatalogBuilder } from '@backstage/plugin-catalog-backend';
import { Router } from 'express';
import { PluginEnvironment } from '../types';
import { GitlabDiscoveryProcessor } from '@backstage/plugin-catalog-backend-module-gitlab';

export default async function createPlugin(
  env: PluginEnvironment,
): Promise<Router> {
  const builder = await CatalogBuilder.create(env);
  
  // Add GitLab discovery processor for automatic entity registration
  builder.addProcessor(
    GitlabDiscoveryProcessor.fromConfig(env.config, { logger: env.logger })
  );
  
  // Add custom entity processors
  builder.addProcessor(new CustomEntityValidator());
  builder.addProcessor(new ComplianceChecker());
  
  const { processingEngine, router } = await builder.build();
  await processingEngine.start();
  
  return router;
}

// Custom processor for validation
class CustomEntityValidator {
  async validateEntityKind(entity: Entity): Promise<boolean> {
    // Custom validation logic
    if (entity.kind === 'Component') {
      const spec = entity.spec as ComponentEntityV1alpha1;
      return this.validateComponentSpec(spec);
    }
    return true;
  }
  
  private validateComponentSpec(spec: ComponentEntityV1alpha1): boolean {
    // Ensure required fields are present
    if (!spec.lifecycle || !spec.type) {
      return false;
    }
    
    // Validate owner exists in catalog
    if (!spec.owner) {
      return false;
    }
    
    return true;
  }
}

// Compliance checker processor
class ComplianceChecker {
  async postProcessEntity(entity: Entity, location: LocationSpec): Promise<Entity> {
    // Add compliance status to entities
    const complianceStatus = await this.checkCompliance(entity);
    
    return {
      ...entity,
      metadata: {
        ...entity.metadata,
        annotations: {
          ...entity.metadata.annotations,
          'internal-platform/compliance-status': complianceStatus.status,
          'internal-platform/compliance-last-checked': new Date().toISOString(),
        }
      }
    };
  }
  
  private async checkCompliance(entity: Entity): Promise<{status: string, issues: string[]}> {
    // Check various compliance requirements
    const issues: string[] = [];
    
    // Check if security scanning is enabled
    if (!entity.metadata.annotations?.['internal-platform/security-scanning']) {
      issues.push('Security scanning not configured');
    }
    
    // Check if monitoring is configured
    if (!entity.metadata.annotations?.['internal-platform/monitoring']) {
      issues.push('Monitoring not configured');
    }
    
    return {
      status: issues.length === 0 ? 'compliant' : 'non-compliant',
      issues
    };
  }
}

# app-config.yaml - Backstage configuration
app:
  title: 'Internal Developer Platform'
  baseUrl: 'https://platform.company.com'

organization:
  name: 'Company Name'

backend:
  baseUrl: 'https://platform.company.com'
  listen:
    port: 7007
    host: '0.0.0.0'
  csp:
    connect-src: ["'self'", 'http:', 'https:']
    img-src: ["'self'", 'data:', 'https:']
    media-src: ["'self'"]
    upgrade-insecure-requests: false

# TechDocs configuration
techdocs:
  builder: 'local'
  generator:
    runIn: 'docker'
  publisher:
    type: 'local'
    cache:
      store: 'memory'

# Catalog configuration
catalog:
  import:
    entityFilename: 'catalog-info.yaml'
    pullRequestBranchName: 'backstage-integration'
  rules:
    - allow: [Component, System, API, Resource, Location]
  locations:
    # GitLab groups to discover
    - type: 'gitlab-discovery'
      target: 'https://gitlab.company.com/api/v4/groups/10/projects?simple=true&membership=true'

# Kubernetes plugin configuration
kubernetes:
  serviceLocatorMethod:
    type: 'multiTenant'
  clusterLocatorMethods:
    - type: 'config'
      clusters:
        - url: https://kubernetes.company.com
          name: production
          authProvider: 'serviceAccount'
          serviceAccountToken: ${KUBERNETES_SERVICE_ACCOUNT_TOKEN}
          skipTLSVerify: false
          dashboardApp: 'kubernetes-dashboard'
          customResources:
            - group: 'argoproj.io'
              apiVersion: 'v1alpha1'
              plural: 'rollouts'

# infrastructure_as_a_service.py
from abc import ABC, abstractmethod
from typing import Dict, List, Optional
from dataclasses import dataclass
import boto3
import os

@dataclass
class ServiceSpec:
    name: str
    replicas: int
    cpu_request: str
    memory_request: str
    cpu_limit: str
    memory_limit: str
    environment: Dict[str, str]
    ports: List[int]
    health_check_path: str

@dataclass
class DatabaseSpec:
    name: str
    engine: str
    version: str
    instance_type: str
    storage_gb: int
    publicly_accessible: bool
    backup_retention_days: int

class InfrastructureService(ABC):
    @abstractmethod
    async def deploy_service(self, spec: ServiceSpec) -> str:
        """Deploy a service and return the URL"""
        pass
    
    @abstractmethod
    async def create_database(self, spec: DatabaseSpec) -> str:
        """Create a database and return the connection string"""
        pass
    
    @abstractmethod
    async def get_service_status(self, service_name: str) -> Dict:
        """Get the status of a deployed service"""
        pass

class AWSInfrastructureService(InfrastructureService):
    def __init__(self):
        self.ecs_client = boto3.client('ecs')
        self.ec2_client = boto3.client('ec2')
        self.rds_client = boto3.client('rds')
        self.cluster_name = os.getenv('ECS_CLUSTER_NAME', 'default-cluster')
    
    async def deploy_service(self, spec: ServiceSpec) -> str:
        """
        Deploy a service to ECS with Fargate
        """
        # Create task definition
        task_definition = {
            'family': spec.name,
            'networkMode': 'awsvpc',
            'requiresCompatibilities': ['FARGATE'],
            'cpu': '256',  # Default, can be adjusted
            'memory': '512',  # Default, can be adjusted
            'executionRoleArn': os.getenv('ECS_EXECUTION_ROLE_ARN'),
            'taskRoleArn': os.getenv('ECS_TASK_ROLE_ARN'),
            'containerDefinitions': [
                {
                    'name': spec.name,
                    'image': f'{spec.name}:latest',
                    'cpu': 256,
                    'memoryReservation': 512,
                    'portMappings': [
                        {
                            'containerPort': port,
                            'protocol': 'tcp'
                        } for port in spec.ports
                    ],
                    'environment': [
                        {
                            'name': key,
                            'value': value
                        } for key, value in spec.environment.items()
                    ],
                    'healthCheck': {
                        'command': ['CMD-SHELL', f'curl -f http://localhost:{spec.ports[0]}{spec.health_check_path} || exit 1'],
                        'interval': 30,
                        'timeout': 5,
                        'retries': 3,
                        'startPeriod': 60
                    }
                }
            ]
        }
        
        # Register task definition
        response = self.ecs_client.register_task_definition(**task_definition)
        task_def_arn = response['taskDefinition']['taskDefinitionArn']
        
        # Create/update service
        service_response = self.ecs_client.create_service(
            cluster=self.cluster_name,
            serviceName=spec.name,
            taskDefinition=task_def_arn,
            desiredCount=spec.replicas,
            launchType='FARGATE',
            networkConfiguration={
                'awsvpcConfiguration': {
                    'subnets': os.getenv('SUBNET_IDS', '').split(','),
                    'securityGroups': os.getenv('SECURITY_GROUP_IDS', '').split(','),
                    'assignPublicIp': 'ENABLED'
                }
            }
        )
        
        # Return service URL (would need ALB/Route53 setup in real implementation)
        return f"https://{spec.name}.company.internal"
    
    async def create_database(self, spec: DatabaseSpec) -> str:
        """
        Create a database instance
        """
        db_identifier = f"{spec.name}-{os.getenv('ENVIRONMENT', 'dev')}"
        
        response = self.rds_client.create_db_instance(
            DBInstanceIdentifier=db_identifier,
            DBInstanceClass=spec.instance_type,
            Engine=spec.engine,
            EngineVersion=spec.version,
            MasterUsername='admin',
            MasterUserPassword=os.getenv('DB_MASTER_PASSWORD'),
            AllocatedStorage=spec.storage_gb,
            BackupRetentionPeriod=spec.backup_retention_days,
            PubliclyAccessible=spec.publicly_accessible,
            VpcSecurityGroupIds=os.getenv('DB_SECURITY_GROUPS', '').split(',')
        )
        
        # Return connection string (in real implementation, would wait for availability)
        endpoint = response['DBInstance']['Endpoint']
        return f"{spec.engine}://{endpoint.Address}:{endpoint.Port}/{spec.name}"
    
    async def get_service_status(self, service_name: str) -> Dict:
        """
        Get service status
        """
        try:
            response = self.ecs_client.describe_services(
                cluster=self.cluster_name,
                services=[service_name]
            )
            
            service = response['services'][0] if response['services'] else {}
            
            return {
                'status': service.get('status', 'UNKNOWN'),
                'running_tasks_count': service.get('runningCount', 0),
                'pending_tasks_count': service.get('pendingCount', 0),
                'desired_count': service.get('desiredCount', 0),
                'events': [event['message'] for event in service.get('events', [])[:5]]
            }
        except Exception as e:
            return {
                'status': 'ERROR',
                'error': str(e)
            }

# Developer SDK
class PlatformSDK:
    def __init__(self, infra_service: InfrastructureService):
        self.infra = infra_service
    
    async def deploy_microservice(self, 
                                 name: str, 
                                 docker_image: str, 
                                 replicas: int = 1,
                                 environment: Dict[str, str] = None,
                                 database_required: bool = False) -> Dict:
        """
        Simple deployment method for developers
        """
        # Deploy the service
        service_spec = ServiceSpec(
            name=name,
            replicas=replicas,
            cpu_request='256m',
            memory_request='512Mi',
            cpu_limit='512m',
            memory_limit='1Gi',
            environment=environment or {},
            ports=[8080],
            health_check_path='/health'
        )
        
        service_url = await self.infra.deploy_service(service_spec)
        
        result = {
            'service_name': name,
            'url': service_url,
            'status': 'deployed'
        }
        
        # Optionally create database
        if database_required:
            db_spec = DatabaseSpec(
                name=f"{name}_db",
                engine='postgres',
                version='13',
                instance_type='db.t3.micro',
                storage_gb=20,
                publicly_accessible=False,
                backup_retention_days=7
            )
            
            connection_string = await self.infra.create_database(db_spec)
            result['database'] = {
                'connection_string': connection_string,
                'status': 'created'
            }
        
        return result

# Usage example
async def deploy_new_service():
    infra = AWSInfrastructureService()
    sdk = PlatformSDK(infra)
    
    deployment_result = await sdk.deploy_microservice(
        name='user-service',
        docker_image='company/user-service:latest',
        replicas=3,
        environment={
            'DATABASE_URL': 'postgresql://...',  # Would be auto-generated
            'API_KEY': '...'  # Would come from secrets manager
        },
        database_required=True
    )
    
    print(f"Service deployed: {deployment_result}")

# otel-collector-config.yaml
receivers:
  otlp:
    protocols:
      grpc:
        endpoint: 0.0.0.0:4317
      http:
        endpoint: 0.0.0.0:4318
  prometheus:
    config:
      scrape_configs:
      - job_name: 'kubernetes-pods'
        kubernetes_sd_configs:
        - role: pod
        relabel_configs:
        - source_labels: [__meta_kubernetes_pod_annotation_prometheus_io_scrape]
          action: keep
          regex: true
        - source_labels: [__meta_kubernetes_pod_annotation_prometheus_io_path]
          action: replace
          target_label: __metrics_path__
          regex: (.+)
        - source_labels: [__address__, __meta_kubernetes_pod_annotation_prometheus_io_port]
          action: replace
          regex: ([^:]+)(?::\d+)?;(\d+)
          replacement: $1:$2
          target_label: __address__

processors:
  batch:
    timeout: 1s
    send_batch_size: 1024
  memory_limiter:
    limit_mib: 1000
    spike_limit_mib: 200
  transform:
    log_statements:
      - context: log
        statements:
          - set(attributes["deployment.environment"], "production") where attributes["k8s.namespace.name"] == "production"
    span_statements:
      - context: span
        statements:
          - set(attributes["service.version"], resource.attributes["service.version"]) where attributes["service.version"] == nil

exporters:
  otlp:
    endpoint: tempo:4317
    tls:
      insecure: true
  prometheus:
    endpoint: "0.0.0.0:8889"
    resource_to_telemetry_conversion:
      enabled: true
  logging:
    loglevel: debug

service:
  pipelines:
    traces:
      receivers: [otlp]
      processors: [memory_limiter, batch, transform]
      exporters: [otlp, logging]
    metrics:
      receivers: [otlp, prometheus]
      processors: [memory_limiter, batch, transform]
      exporters: [prometheus, logging]
    logs:
      receivers: [otlp]
      processors: [memory_limiter, batch, transform]
      exporters: [logging]

# istio-telemetry.yaml
apiVersion: telemetry.istio.io/v1alpha1
kind: Telemetry
metadata:
  name: default
  namespace: istio-system
spec:
  # Enable tracing for all workloads in mesh
  tracing:
  - providers:
    - name: otel
    randomSamplingPercentage: 100.0
    customTags:
      baggage:
        kind: W3CBaggage
      tenant:
        literal:
          value: "default"
  
  # Enable metrics for all workloads
  metrics:
  - providers:
    - name: prometheus
    overrides:
    - match:
        metric: ALL_METRICS
      tagOverrides:
        # Remove high-cardinality tags that could cause performance issues
        connection_security_policy:
          operation: REMOVE
        destination_canonical_revision:
          operation: REMOVE
        source_canonical_revision:
          operation: REMOVE

---
# OpenTelemetry Collector Deployment
apiVersion: v1
kind: ConfigMap
metadata:
  name: otel-collector-conf
  namespace: istio-system
  labels:
    app: opentelemetry
    component: collector
    otel: config
data:
  relay: |
    receivers:
      otlp:
        protocols:
          grpc:
            endpoint: 0.0.0.0:4317
          http:
            endpoint: 0.0.0.0:4318
    processors:
      batch:
        timeout: 1s
        send_batch_size: 1024
      attributes:
        actions:
          - key: service.name
            action: upsert
            from_attribute: service.name
    exporters:
      logging:
        loglevel: debug
      otlp:
        endpoint: jaeger-collector:4317
        tls:
          insecure: true
    service:
      pipelines:
        traces:
          receivers: [otlp]
          processors: [attributes, batch]
          exporters: [logging, otlp]

# observability_data_pipeline.py
from pyspark.sql import SparkSession
from pyspark.sql.functions import col, from_json, unix_timestamp, date_format
from pyspark.sql.types import StructType, StructField, StringType, TimestampType, DoubleType
import json

class ObservabilityDataLake:
    def __init__(self):
        self.spark = SparkSession.builder \
            .appName("ObservabilityDataLake") \
            .config("spark.sql.streaming.checkpointLocation", "/tmp/checkpoint") \
            .getOrCreate()
    
    def create_metrics_schema(self):
        """Define schema for metrics data"""
        return StructType([
            StructField("timestamp", TimestampType(), True),
            StructField("metric_name", StringType(), True),
            StructField("value", DoubleType(), True),
            StructField("labels", StringType(), True),  # JSON string
            StructField("source", StringType(), True),
            StructField("environment", StringType(), True)
        ])
    
    def create_logs_schema(self):
        """Define schema for logs data"""
        return StructType([
            StructField("timestamp", TimestampType(), True),
            StructField("level", StringType(), True),
            StructField("message", StringType(), True),
            StructField("service", StringType(), True),
            StructField("trace_id", StringType(), True),
            StructField("span_id", StringType(), True),
            StructField("fields", StringType(), True)  # Additional structured fields
        ])
    
    def process_metrics_stream(self, input_path: str, output_path: str):
        """Process streaming metrics data"""
        # Read streaming metrics data
        metrics_df = (self.spark.readStream
                     .format("kafka")
                     .option("kafka.bootstrap.servers", "kafka:9092")
                     .option("subscribe", "metrics")
                     .load())
        
        # Parse JSON data
        parsed_df = (metrics_df
                    .select(col("value").cast("string").alias("json_value"))
                    .select(from_json(col("json_value"), self.create_metrics_schema()).alias("data"))
                    .select("data.*"))
        
        # Process and enrich data
        processed_df = (parsed_df
                       .withColumn("date_partition", date_format(col("timestamp"), "yyyy-MM-dd"))
                       .withColumn("hour_partition", date_format(col("timestamp"), "HH")))
        
        # Write to data lake
        query = (processed_df.writeStream
                .format("parquet")
                .option("path", f"{output_path}/metrics")
                .option("checkpointLocation", f"{output_path}/checkpoints/metrics")
                .partitionBy("date_partition", "hour_partition")
                .trigger(processingTime='1 minute')
                .start())
        
        return query
    
    def process_logs_stream(self, input_path: str, output_path: str):
        """Process streaming logs data"""
        # Read streaming logs data
        logs_df = (self.spark.readStream
                  .format("kafka")
                  .option("kafka.bootstrap.servers", "kafka:9092")
                  .option("subscribe", "logs")
                  .load())
        
        # Parse JSON data
        parsed_df = (logs_df
                    .select(col("value").cast("string").alias("json_value"))
                    .select(from_json(col("json_value"), self.create_logs_schema()).alias("data"))
                    .select("data.*"))
        
        # Process and enrich data
        processed_df = (parsed_df
                       .withColumn("date_partition", date_format(col("timestamp"), "yyyy-MM-dd"))
                       .withColumn("hour_partition", date_format(col("timestamp"), "HH")))
        
        # Write to data lake
        query = (processed_df.writeStream
                .format("parquet")
                .option("path", f"{output_path}/logs")
                .option("checkpointLocation", f"{output_path}/checkpoints/logs")
                .partitionBy("date_partition", "hour_partition")
                .trigger(processingTime='1 minute')
                .start())
        
        return query
    
    def run_analytics_queries(self):
        """Run analytical queries on the data lake"""
        # Example: Identify top error-prone services
        logs_df = self.spark.read.parquet("/data-lake/logs/*/*/")
        
        error_analysis = (logs_df
                         .filter(col("level").isin(["ERROR", "FATAL"]))
                         .groupBy("service")
                         .count()
                         .orderBy(col("count").desc()))
        
        print("Top Error-Prone Services:")
        error_analysis.show(10)
        
        # Example: Performance trends
        metrics_df = self.spark.read.parquet("/data-lake/metrics/*/*/")
        
        performance_trends = (metrics_df
                             .filter(col("metric_name").startswith("http_request_duration"))
                             .groupBy("service", "date_partition")
                             .agg({"value": "avg"})
                             .withColumnRenamed("avg(value)", "avg_duration"))
        
        print("Performance Trends:")
        performance_trends.show(20)

# Usage
if __name__ == "__main__":
    obs_lake = ObservabilityDataLake()
    
    # Start processing streams
    metrics_query = obs_lake.process_metrics_stream("kafka://kafka:9092", "/data-lake")
    logs_query = obs_lake.process_logs_stream("kafka://kafka:9092", "/data-lake")
    
    # Run analytics
    obs_lake.run_analytics_queries()
    
    # Wait for queries to finish
    metrics_query.awaitTermination()
    logs_query.awaitTermination()

# istio-security.yaml
apiVersion: security.istio.io/v1beta1
kind: PeerAuthentication
metadata:
  name: default
  namespace: istio-system
spec:
  mtls:
    mode: STRICT  # Enforce mTLS for all communication

---
apiVersion: security.istio.io/v1beta1
kind: AuthorizationPolicy
metadata:
  name: deny-all
  namespace: default
spec:
  action: DENY
  rules:
  - from:
    - source:
        notNames: ["istio-system"]  # Allow istio-system communication

---
apiVersion: security.istio.io/v1beta1
kind: AuthorizationPolicy
metadata:
  name: allow-product-page
  namespace: default
spec:
  selector:
    matchLabels:
      app: productpage
  rules:
  - from:
    - source:
        principals: ["cluster.local/ns/default/sa/bookinfo-productpage"]
    to:
    - operation:
        methods: ["GET"]
    when:
    - key: request.headers[authorization]
      values: ["Bearer*"]

---
apiVersion: networking.istio.io/v1alpha3
kind: Sidecar
metadata:
  name: restricted-pod
  namespace: default
spec:
  workloadSelector:
    labels:
      app: restricted-app
  outboundTrafficPolicy:
    mode: REGISTRY_ONLY  # Only allow traffic to known services

# compliance_checker.py
import boto3
import kubernetes
from kubernetes import client, config
from typing import List, Dict, Any
import json

class ComplianceChecker:
    def __init__(self):
        # Initialize AWS clients
        self.ec2_client = boto3.client('ec2')
        self.rds_client = boto3.client('rds')
        self.iam_client = boto3.client('iam')
        
        # Initialize Kubernetes client
        try:
            config.load_incluster_config()
        except:
            config.load_kube_config()
        self.k8s_client = client.ApiClient()
    
    def check_aws_compliance(self) -> Dict[str, Any]:
        """Check AWS resource compliance"""
        results = {
            'ec2_instances': self._check_ec2_compliance(),
            'rds_instances': self._check_rds_compliance(),
            'iam_policies': self._check_iam_compliance(),
            'overall_score': 0
        }
        
        # Calculate overall compliance score
        total_checks = sum(len(v) if isinstance(v, list) else 1 for v in results.values() if v != 0)
        compliant_checks = sum(1 for v in results.values() if isinstance(v, list) and all(item.get('compliant') for item in v))
        
        results['overall_score'] = (compliant_checks / total_checks * 100) if total_checks > 0 else 0
        
        return results
    
    def _check_ec2_compliance(self) -> List[Dict[str, Any]]:
        """Check EC2 instance compliance"""
        instances = self.ec2_client.describe_instances()
        results = []
        
        for reservation in instances['Reservations']:
            for instance in reservation['Instances']:
                instance_id = instance['InstanceId']
                
                # Check if instance has required tags
                has_compliant_tags = any(
                    tag['Key'] in ['Environment', 'Owner', 'CostCenter'] 
                    for tag in instance.get('Tags', [])
                )
                
                # Check if instance has public IP (security concern)
                has_public_ip = 'PublicIpAddress' in instance
                
                # Check if instance is in correct VPC
                vpc_id = instance['VpcId']
                expected_vpc_tag = any(
                    tag['Key'] == 'NetworkTier' and tag['Value'] in ['Production', 'Development']
                    for tag in instance.get('Tags', [])
                )
                
                results.append({
                    'instance_id': instance_id,
                    'compliant': has_compliant_tags and not has_public_ip and expected_vpc_tag,
                    'issues': [
                        'missing_required_tags' if not has_compliant_tags else None,
                        'has_public_ip' if has_public_ip else None,
                        'wrong_vpc_tier' if not expected_vpc_tag else None
                    ],
                    'region': instance['Placement']['AvailabilityZone'][:-1]
                })
        
        return results
    
    def _check_rds_compliance(self) -> List[Dict[str, Any]]:
        """Check RDS instance compliance"""
        instances = self.rds_client.describe_db_instances()
        results = []
        
        for instance in instances['DBInstances']:
            instance_id = instance['DBInstanceIdentifier']
            
            # Check if encrypted at rest
            encrypted = instance['StorageEncrypted']
            
            # Check if public access is disabled
            publicly_accessible = instance['PubliclyAccessible']
            
            # Check backup retention
            backup_retention = instance['BackupRetentionPeriod']
            has_adequate_backup = backup_retention >= 7
            
            # Check if enhanced monitoring is enabled
            has_enhanced_monitoring = instance.get('MonitoringInterval', 0) > 0
            
            results.append({
                'db_instance_id': instance_id,
                'compliant': encrypted and not publicly_accessible and has_adequate_backup,
                'issues': [
                    'not_encrypted' if not encrypted else None,
                    'publicly_accessible' if publicly_accessible else None,
                    'inadequate_backup' if not has_adequate_backup else None,
                    'no_enhanced_monitoring' if not has_enhanced_monitoring else None
                ],
                'engine': instance['Engine']
            })
        
        return results
    
    def check_kubernetes_compliance(self) -> Dict[str, Any]:
        """Check Kubernetes cluster compliance"""
        v1 = client.CoreV1Api()
        apps_v1 = client.AppsV1Api()
        
        results = {
            'pods': self._check_pod_security(),
            'rbac': self._check_rbac_compliance(),
            'network_policies': self._check_network_policies(),
            'overall_score': 0
        }
        
        return results
    
    def _check_pod_security(self) -> List[Dict[str, Any]]:
        """Check pod security compliance"""
        v1 = client.CoreV1Api()
        pods = v1.list_pod_for_all_namespaces()
        results = []
        
        for pod in pods.items:
            pod_name = pod.metadata.name
            namespace = pod.metadata.namespace
            
            compliant = True
            issues = []
            
            # Check if pod runs as non-root user
            if pod.spec.security_context and hasattr(pod.spec.security_context, 'run_as_non_root'):
                if not pod.spec.security_context.run_as_non_root:
                    compliant = False
                    issues.append('runs_as_root')
            
            # Check if privileged containers are used
            for container in pod.spec.containers:
                if container.security_context and getattr(container.security_context, 'privileged', False):
                    compliant = False
                    issues.append('privileged_container')
            
            # Check if read-only root filesystem is used
            for container in pod.spec.containers:
                if not (container.security_context and getattr(container.security_context, 'read_only_root_filesystem', False)):
                    compliant = False
                    issues.append('writable_root_filesystem')
            
            results.append({
                'pod_name': pod_name,
                'namespace': namespace,
                'compliant': compliant,
                'issues': issues
            })
        
        return results
    
    def generate_compliance_report(self) -> str:
        """Generate comprehensive compliance report"""
        aws_results = self.check_aws_compliance()
        k8s_results = self.check_kubernetes_compliance()
        
        report = {
            'timestamp': '2024-01-08T10:30:00Z',
            'aws_compliance': aws_results,
            'kubernetes_compliance': k8s_results,
            'recommendations': self._generate_recommendations(aws_results, k8s_results)
        }
        
        return json.dumps(report, indent=2)
    
    def _generate_recommendations(self, aws_results: Dict, k8s_results: Dict) -> List[str]:
        """Generate compliance recommendations"""
        recommendations = []
        
        # AWS recommendations
        for result_list in aws_results.values():
            if isinstance(result_list, list):
                for item in result_list:
                    if not item.get('compliant'):
                        issues = [issue for issue in item.get('issues', []) if issue]
                        for issue in issues:
                            recommendations.append(f"AWS - {item.get('instance_id', item.get('db_instance_id', 'Resource'))}: Fix {issue}")
        
        # Kubernetes recommendations
        for result_list in k8s_results.values():
            if isinstance(result_list, list):
                for item in result_list:
                    if not item.get('compliant'):
                        issues = [issue for issue in item.get('issues', []) if issue]
                        for issue in issues:
                            recommendations.append(f"Kubernetes - {item.get('pod_name', 'Resource')} in {item.get('namespace', 'unknown')}: Fix {issue}")
        
        return recommendations

# Usage example
checker = ComplianceChecker()
report = checker.generate_compliance_report()
print(report)

# argo-cd-application.yaml
apiVersion: argoproj.io/v1alpha1
kind: Application
metadata:
  name: my-app-prod
  namespace: argocd
spec:
  project: default
  source:
    repoURL: https://github.com/company/infrastructure.git
    targetRevision: HEAD
    path: environments/production/my-app
    helm:
      valueFiles:
        - values-prod.yaml
        - values-global.yaml
  destination:
    server: https://kubernetes.default.svc
    namespace: my-app-prod
  syncPolicy:
    automated:
      prune: true
      selfHeal: true
    syncOptions:
      - CreateNamespace=true
      - ApplyOutOfSyncOnly=true
    retry:
      limit: 5
      backoff:
        duration: 5s
        factor: 2
        maxDuration: 3m

---
# argo-rollouts for progressive delivery
apiVersion: argoproj.io/v1alpha1
kind: Rollout
metadata:
  name: my-app-rollout
spec:
  replicas: 5
  strategy:
    canary:
      steps:
      - setWeight: 10
      - pause: {duration: 2m}
      - setWeight: 20
      - pause: {duration: 2m}
      - setWeight: 40
      - pause: {duration: 2m}
      - setWeight: 60
      - pause: {duration: 2m}
      - setWeight: 80
      - pause: {duration: 2m}
      - setWeight: 100
      canaryService: my-app-canary
      stableService: my-app-stable
      trafficRouting:
        nginx:
          stableIngress: my-app-ingress
      analysis:
        templates:
        - templateName: success-rate
        args:
        - name: service-name
          value: my-app-canary
  revisionHistoryLimit: 2
  selector:
    matchLabels:
      app: my-app
  template:
    metadata:
      labels:
        app: my-app
    spec:
      containers:
      - name: my-app
        image: my-app:latest
        ports:
        - containerPort: 8080
        resources:
          requests:
            memory: "64Mi"
            cpu: "250m"
          limits:
            memory: "128Mi"
            cpu: "500m"

# infrastructure_from_code.py
from typing import Dict, List, Any, Optional
from pydantic import BaseModel, Field
import json
from enum import Enum

class ResourceType(str, Enum):
    COMPUTE = "compute"
    DATABASE = "database"
    NETWORK = "network"
    STORAGE = "storage"
    SECURITY = "security"

class ComputeSpec(BaseModel):
    type: str = Field(description="Instance type (e.g., t3.micro, Standard_D2s_v3)")
    count: int = Field(default=1, description="Number of instances")
    os: str = Field(description="Operating system")
    region: str = Field(default="us-east-1")

class DatabaseSpec(BaseModel):
    engine: str = Field(description="Database engine (mysql, postgresql, mongodb)")
    version: str = Field(default="latest")
    instance_type: str = Field(description="Database instance type")
    storage_gb: int = Field(description="Storage size in GB")
    multi_az: bool = Field(default=False)

class NetworkSpec(BaseModel):
    vpc_cidr: str = Field(default="10.0.0.0/16")
    subnets: List[str] = Field(default_factory=list)
    enable_nat: bool = Field(default=True)

class InfrastructureSpec(BaseModel):
    name: str
    environment: str = Field(description="Environment (dev, staging, prod)")
    region: str = Field(default="us-east-1")
    compute: Optional[List[ComputeSpec]] = Field(default_factory=list)
    databases: Optional[List[DatabaseSpec]] = Field(default_factory=list)
    network: Optional[NetworkSpec] = Field(default=None)
    tags: Dict[str, str] = Field(default_factory=dict)

class InfrastructureGenerator:
    def __init__(self):
        self.providers = {
            'aws': self._generate_aws_infrastructure,
            'azure': self._generate_azure_infrastructure,
            'gcp': self._generate_gcp_infrastructure
        }
    
    def generate_from_spec(self, spec: InfrastructureSpec, provider: str = 'aws') -> Dict[str, Any]:
        """Generate infrastructure code from specification"""
        if provider not in self.providers:
            raise ValueError(f"Unsupported provider: {provider}")
        
        return self.providers[provider](spec)
    
    def _generate_aws_infrastructure(self, spec: InfrastructureSpec) -> Dict[str, Any]:
        """Generate AWS infrastructure using Terraform"""
        terraform_config = {
            'terraform': {
                'required_providers': {
                    'aws': {
                        'source': 'hashicorp/aws',
                        'version': '~> 5.0'
                    }
                }
            },
            'provider': {
                'aws': {
                    'region': spec.region
                }
            },
            'resource': {},
            'output': {}
        }
        
        # Generate VPC if network spec exists
        if spec.network:
            self._add_vpc_resources(terraform_config, spec.network, spec.name)
        
        # Generate compute resources
        for i, compute in enumerate(spec.compute or []):
            self._add_compute_resources(terraform_config, compute, f"{spec.name}-compute-{i}")
        
        # Generate database resources
        for i, db in enumerate(spec.databases or []):
            self._add_database_resources(terraform_config, db, f"{spec.name}-db-{i}")
        
        return terraform_config
    
    def _add_vpc_resources(self, config: Dict, network_spec: NetworkSpec, name_prefix: str):
        """Add VPC resources to Terraform config"""
        config['resource']['aws_vpc'] = {
            f"{name_prefix}_vpc": {
                'cidr_block': network_spec.vpc_cidr,
                'enable_dns_hostnames': True,
                'enable_dns_support': True,
                'tags': {
                    'Name': f"{name_prefix}-vpc",
                    'Environment': name_prefix.split('-')[0]
                }
            }
        }
        
        # Add subnets
        for i, subnet_cidr in enumerate(network_spec.subnets):
            config['resource']['aws_subnet'] = {
                f"{name_prefix}_subnet_{i}": {
                    'vpc_id': f"${{aws_vpc.{name_prefix}_vpc.id}}",
                    'cidr_block': subnet_cidr,
                    'availability_zone': f"${{data.aws_availability_zones.available.names[{i}]}}",
                    'tags': {
                        'Name': f"{name_prefix}-subnet-{i}",
                        'Environment': name_prefix.split('-')[0]
                    }
                }
            }
    
    def _add_compute_resources(self, config: Dict, compute_spec: ComputeSpec, name_prefix: str):
        """Add compute resources to Terraform config"""
        config['resource']['aws_instance'] = {
            f"{name_prefix}_instance": {
                'ami': self._get_ami_for_os(compute_spec.os),
                'instance_type': compute_spec.type,
                'count': compute_spec.count,
                'tags': {
                    'Name': f"{name_prefix}-instance-${{count.index}}",
                    'Environment': name_prefix.split('-')[0]
                }
            }
        }
    
    def _add_database_resources(self, config: Dict, db_spec: DatabaseSpec, name_prefix: str):
        """Add database resources to Terraform config"""
        config['resource']['aws_db_instance'] = {
            f"{name_prefix}_db": {
                'identifier': f"{name_prefix}-db",
                'engine': db_spec.engine,
                'engine_version': db_spec.version,
                'instance_class': db_spec.instance_type,
                'allocated_storage': db_spec.storage_gb,
                'storage_encrypted': True,
                'backup_retention_period': 7,
                'multi_az': db_spec.multi_az,
                'tags': {
                    'Name': f"{name_prefix}-db",
                    'Environment': name_prefix.split('-')[0]
                }
            }
        }
    
    def _get_ami_for_os(self, os: str) -> str:
        """Get appropriate AMI for OS (simplified)"""
        ami_map = {
            'ubuntu': 'ami-0c02fb55956c7d316',  # Ubuntu 22.04 LTS
            'centos': 'ami-0c55b18b98952c964',  # CentOS Stream 9
            'amazon-linux': 'ami-0c02fb55956c7d316'
        }
        return ami_map.get(os.lower(), 'ami-0c02fb55956c7d316')

# Example usage
def generate_production_infrastructure():
    spec = InfrastructureSpec(
        name="my-app",
        environment="production",
        region="us-west-2",
        network=NetworkSpec(
            vpc_cidr="10.0.0.0/16",
            subnets=["10.0.1.0/24", "10.0.2.0/24"]
        ),
        compute=[
            ComputeSpec(
                type="t3.medium",
                count=3,
                os="ubuntu"
            )
        ],
        databases=[
            DatabaseSpec(
                engine="postgresql",
                instance_type="db.t3.micro",
                storage_gb=100,
                multi_az=True
            )
        ],
        tags={
            "Project": "MyApp",
            "Owner": "DevOps Team",
            "CostCenter": "Engineering"
        }
    )
    
    generator = InfrastructureGenerator()
    terraform_code = generator.generate_from_spec(spec, 'aws')
    
    # Write to file
    with open('generated_infrastructure.tf.json', 'w') as f:
        json.dump(terraform_code, f, indent=2)
    
    print("Infrastructure code generated successfully!")
    print(json.dumps(terraform_code, indent=2))

# Run the example
generate_production_infrastructure()