import json
import asyncio
from datetime import datetime, timedelta
from typing import Dict, List, Optional
from kafka import KafkaProducer
from kafka.errors import KafkaError
import boto3
import pandas as pd
from dataclasses import dataclass, asdict

@dataclass
class CostEvent:
    event_id: str
    timestamp: datetime
    event_type: str
    service: str
    region: str
    cost_amount: float
    resource_id: str
    business_unit: str
    project_id: str
    environment: str
    metadata: Dict

@dataclass
class BusinessEvent:
    event_id: str
    timestamp: datetime
    event_type: str
    user_id: str
    feature: str
    action: str
    revenue_impact: float
    business_unit: str
    metadata: Dict

class FinOpsEventProducer:
    def __init__(self, bootstrap_servers: List[str]):
        self.producer = KafkaProducer(
            bootstrap_servers=bootstrap_servers,
            value_serializer=lambda v: json.dumps(v, default=str).encode('utf-8'),
            key_serializer=lambda v: v.encode('utf-8') if v else None,
            acks='all',
            retries=3
        )
        
        self.ce_client = boto3.client('ce')
        self.snowflake_conn = None  # Would be initialized with Snowflake connection
        
    async def produce_cost_events(self) -> None:
        """Continuously produce cost events from AWS Cost Explorer"""
        while True:
            try:
                # Get cost data from AWS Cost Explorer
                cost_data = self._get_current_cost_data()
                
                # Transform to cost events
                cost_events = self._transform_to_cost_events(cost_data)
                
                # Produce to Kafka
                for event in cost_events:
                    self._produce_event(
                        topic='finops.cost.events',
                        key=event.resource_id,
                        value=asdict(event)
                    )
                
                # Wait for next interval
                await asyncio.sleep(300)  # 5 minutes
                
            except Exception as e:
                print(f"Error producing cost events: {e}")
                await asyncio.sleep(60)  # Wait 1 minute before retry
    
    def _get_current_cost_data(self) -> List[Dict]:
        """Get current cost data from AWS Cost Explorer"""
        try:
            response = self.ce_client.get_cost_and_usage(
                TimePeriod={
                    'Start': (datetime.now() - timedelta(hours=1)).strftime('%Y-%m-%d'),
                    'End': datetime.now().strftime('%Y-%m-%d')
                },
                Granularity='HOURLY',
                Metrics=['UnblendedCost'],
                GroupBy=[
                    {'Type': 'DIMENSION', 'Key': 'SERVICE'},
                    {'Type': 'DIMENSION', 'Key': 'REGION'},
                    {'Type': 'TAG', 'Key': 'BusinessUnit'},
                    {'Type': 'TAG', 'Key': 'ProjectId'},
                    {'Type': 'TAG', 'Key': 'Environment'}
                ]
            )
            
            return response['ResultsByTime']
        except Exception as e:
            print(f"Error getting cost data: {e}")
            return []
    
    def _transform_to_cost_events(self, cost_data: List[Dict]) -> List[CostEvent]:
        """Transform AWS cost data to standardized cost events"""
        events = []
        
        for time_period in cost_data:
            for group in time_period.get('Groups', []):
                cost_amount = float(group['Metrics']['UnblendedCost']['Amount'])
                
                if cost_amount > 0:  # Only include actual costs
                    event = CostEvent(
                        event_id=f"cost_{datetime.now().strftime('%Y%m%d%H%M%S')}_{len(events)}",
                        timestamp=datetime.strptime(time_period['TimePeriod']['Start'], '%Y-%m-%d'),
                        event_type='cloud_cost_incurred',
                        service=group['Keys'][0],
                        region=group['Keys'][1],
                        cost_amount=cost_amount,
                        resource_id=f"{group['Keys'][0]}_{group['Keys'][1]}",
                        business_unit=group['Keys'][2] if len(group['Keys']) > 2 else 'unknown',
                        project_id=group['Keys'][3] if len(group['Keys']) > 3 else 'unknown',
                        environment=group['Keys'][4] if len(group['Keys']) > 4 else 'unknown',
                        metadata={
                            'time_period': time_period['TimePeriod'],
                            'granularity': 'HOURLY'
                        }
                    )
                    events.append(event)
        
        return events
    
    def produce_business_event(self, business_event: BusinessEvent) -> bool:
        """Produce a business event to Kafka"""
        try:
            self._produce_event(
                topic='finops.business.events',
                key=business_event.user_id,
                value=asdict(business_event)
            )
            return True
        except Exception as e:
            print(f"Error producing business event: {e}")
            return False
    
    def _produce_event(self, topic: str, key: str, value: Dict) -> None:
        """Produce a single event to Kafka"""
        future = self.producer.send(
            topic=topic,
            key=key,
            value=value
        )
        
        try:
            future.get(timeout=10)
        except KafkaError as e:
            print(f"Failed to send event to Kafka: {e}")
    
    async def produce_resource_events(self) -> None:
        """Produce resource utilization events"""
        while True:
            try:
                # Get resource metrics from CloudWatch
                resource_metrics = self._get_resource_metrics()
                
                for metric in resource_metrics:
                    event = CostEvent(
                        event_id=f"resource_{datetime.now().strftime('%Y%m%d%H%M%S')}",
                        timestamp=datetime.now(),
                        event_type='resource_utilization',
                        service=metric['service'],
                        region=metric['region'],
                        cost_amount=0,  # Will be calculated
                        resource_id=metric['resource_id'],
                        business_unit=metric.get('business_unit', 'unknown'),
                        project_id=metric.get('project_id', 'unknown'),
                        environment=metric.get('environment', 'unknown'),
                        metadata={
                            'utilization': metric['utilization'],
                            'resource_type': metric['resource_type'],
                            'cost_estimate': self._estimate_cost(metric)
                        }
                    )
                    
                    self._produce_event(
                        topic='finops.resource.events',
                        key=metric['resource_id'],
                        value=asdict(event)
                    )
                
                await asyncio.sleep(60)  # 1 minute intervals
                
            except Exception as e:
                print(f"Error producing resource events: {e}")
                await asyncio.sleep(30)
    
    def _get_resource_metrics(self) -> List[Dict]:
        """Get resource utilization metrics (simplified)"""
        # In production, this would query CloudWatch or similar
        return [
            {
                'service': 'ec2',
                'region': 'us-west-2',
                'resource_id': 'i-1234567890abcdef0',
                'resource_type': 'instance',
                'utilization': 0.65,
                'business_unit': 'ecommerce',
                'project_id': 'web-frontend',
                'environment': 'production'
            }
        ]
    
    def _estimate_cost(self, metric: Dict) -> float:
        """Estimate cost based on resource utilization"""
        # Simplified cost estimation
        base_costs = {
            'ec2': 0.10,  # per hour
            'rds': 0.15,
            's3': 0.023,  # per GB
        }
        
        base_cost = base_costs.get(metric['service'], 0.05)
        return base_cost * metric['utilization']

# Example usage
async def main():
    producer = FinOpsEventProducer(['kafka-broker1:9092', 'kafka-broker2:9092'])
    
    # Start producing events
    tasks = [
        asyncio.create_task(producer.produce_cost_events()),
        asyncio.create_task(producer.produce_resource_events())
    ]
    
    # Example business event
    business_event = BusinessEvent(
        event_id="biz_20250115093000",
        timestamp=datetime.now(),
        event_type="feature_usage",
        user_id="user_12345",
        feature="premium_checkout",
        action="completed_purchase",
        revenue_impact=199.99,
        business_unit="ecommerce",
        metadata={"order_id": "ORD-67890", "items_count": 3}
    )
    
    producer.produce_business_event(business_event)
    
    await asyncio.gather(*tasks)

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

  

// FinOpsStreamProcessor.java
package com.lktechacademy.finops;

import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.streams.KafkaStreams;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.StreamsConfig;
import org.apache.kafka.streams.kstream.*;
import org.apache.kafka.streams.state.KeyValueStore;
import org.apache.kafka.streams.state.StoreBuilder;
import org.apache.kafka.streams.state.Stores;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;

import java.time.Duration;
import java.util.Properties;
import java.util.concurrent.CountDownLatch;

public class FinOpsStreamProcessor {
    private final ObjectMapper objectMapper = new ObjectMapper();
    
    public Properties getStreamsConfig() {
        Properties props = new Properties();
        props.put(StreamsConfig.APPLICATION_ID_CONFIG, "finops-cost-processor");
        props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "kafka-broker1:9092,kafka-broker2:9092");
        props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());
        props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass());
        props.put(StreamsConfig.PROCESSING_GUARANTEE_CONFIG, StreamsConfig.EXACTLY_ONCE_V2);
        return props;
    }
    
    public void buildCostProcessingPipeline() {
        StreamsBuilder builder = new StreamsBuilder();
        
        // Create state store for cost thresholds
        StoreBuilder> thresholdStore = 
            Stores.keyValueStoreBuilder(
                Stores.persistentKeyValueStore("cost-thresholds"),
                Serdes.String(),
                Serdes.Double()
            );
        builder.addStateStore(thresholdStore);
        
        // Source streams
        KStream costEvents = builder.stream("finops.cost.events");
        KStream businessEvents = builder.stream("finops.business.events");
        KStream resourceEvents = builder.stream("finops.resource.events");
        
        // 1. Real-time cost anomaly detection
        costEvents
            .filter((key, value) -> {
                try {
                    JsonNode event = objectMapper.readTree(value);
                    double cost = event.get("cost_amount").asDouble();
                    return cost > 100.0; // Filter significant costs
                } catch (Exception e) {
                    return false;
                }
            })
            .process(() -> new CostAnomalyProcessor(), "cost-thresholds")
            .to("finops.cost.anomalies", Produced.with(Serdes.String(), Serdes.String()));
        
        // 2. Cost aggregation by business unit (5-minute windows)
        costEvents
            .groupBy((key, value) -> {
                try {
                    JsonNode event = objectMapper.readTree(value);
                    return event.get("business_unit").asText();
                } catch (Exception e) {
                    return "unknown";
                }
            })
            .windowedBy(TimeWindows.ofSizeWithNoGrace(Duration.ofMinutes(5)))
            .aggregate(
                () -> 0.0,
                (key, value, aggregate) -> {
                    try {
                        JsonNode event = objectMapper.readTree(value);
                        return aggregate + event.get("cost_amount").asDouble();
                    } catch (Exception e) {
                        return aggregate;
                    }
                },
                Materialized.with(Serdes.String(), Serdes.Double())
            )
            .toStream()
            .mapValues((readOnlyKey, value) -> {
                // Create aggregation event
                return String.format(
                    "{\"business_unit\": \"%s\", \"total_cost\": %.2f, \"window_start\": \"%s\", \"window_end\": \"%s\"}",
                    readOnlyKey.key(), value, readOnlyKey.window().start(), readOnlyKey.window().end()
                );
            })
            .to("finops.cost.aggregations");
        
        // 3. Join cost events with business events for ROI calculation
        KStream significantCosts = costEvents
            .filter((key, value) -> {
                try {
                    JsonNode event = objectMapper.readTree(value);
                    return event.get("cost_amount").asDouble() > 50.0;
                } catch (Exception e) {
                    return false;
                }
            });
        
        KStream revenueEvents = businessEvents
            .filter((key, value) -> {
                try {
                    JsonNode event = objectMapper.readTree(value);
                    return event.get("revenue_impact").asDouble() > 0;
                } catch (Exception e) {
                    return false;
                }
            });
        
        significantCosts
            .join(
                revenueEvents,
                (costEvent, revenueEvent) -> {
                    try {
                        JsonNode cost = objectMapper.readTree(costEvent);
                        JsonNode revenue = objectMapper.readTree(revenueEvent);
                        
                        double costAmount = cost.get("cost_amount").asDouble();
                        double revenueAmount = revenue.get("revenue_impact").asDouble();
                        double roi = (revenueAmount - costAmount) / costAmount * 100;
                        
                        return String.format(
                            "{\"business_unit\": \"%s\", \"cost\": %.2f, \"revenue\": %.2f, \"roi\": %.2f, \"timestamp\": \"%s\"}",
                            cost.get("business_unit").asText(),
                            costAmount,
                            revenueAmount,
                            roi,
                            java.time.Instant.now().toString()
                        );
                    } catch (Exception e) {
                        return "{\"error\": \"processing_failed\"}";
                    }
                },
                JoinWindows.ofTimeDifferenceWithNoGrace(Duration.ofMinutes(30)),
                StreamJoined.with(Serdes.String(), Serdes.String(), Serdes.String())
            )
            .to("finops.roi.calculations");
        
        // 4. Resource optimization recommendations
        resourceEvents
            .process(() -> new ResourceOptimizationProcessor())
            .to("finops.optimization.recommendations");
        
        // Start the streams application
        KafkaStreams streams = new KafkaStreams(builder.build(), getStreamsConfig());
        
        final CountDownLatch latch = new CountDownLatch(1);
        
        // Attach shutdown handler to catch control-c
        Runtime.getRuntime().addShutdownHook(new Thread("finops-streams-shutdown-hook") {
            @Override
            public void run() {
                streams.close();
                latch.countDown();
            }
        });
        
        try {
            streams.start();
            latch.await();
        } catch (Throwable e) {
            System.exit(1);
        }
        System.exit(0);
    }
    
    // Custom processor for cost anomaly detection
    static class CostAnomalyProcessor implements Processor {
        private ProcessorContext context;
        private KeyValueStore thresholdStore;
        
        @Override
        public void init(ProcessorContext context) {
            this.context = context;
            this.thresholdStore = context.getStateStore("cost-thresholds");
        }
        
        @Override
        public void process(Record record) {
            try {
                ObjectMapper mapper = new ObjectMapper();
                JsonNode event = mapper.readTree(record.value());
                
                String service = event.get("service").asText();
                double currentCost = event.get("cost_amount").asDouble();
                Double historicalAverage = thresholdStore.get(service);
                
                // Check if cost exceeds 2x historical average
                if (historicalAverage != null && currentCost > historicalAverage * 2) {
                    String anomalyEvent = String.format(
                        "{\"anomaly_type\": \"cost_spike\", \"service\": \"%s\", \"current_cost\": %.2f, \"historical_average\": %.2f, \"timestamp\": \"%s\"}",
                        service, currentCost, historicalAverage, record.timestamp().toString()
                    );
                    
                    context.forward(new Record<>(
                        service, anomalyEvent, record.timestamp()
                    ));
                }
                
                // Update historical average (simple moving average)
                double newAverage = historicalAverage == null ? 
                    currentCost : (historicalAverage * 0.9 + currentCost * 0.1);
                thresholdStore.put(service, newAverage);
                
            } catch (Exception e) {
                System.err.println("Error processing cost event: " + e.getMessage());
            }
        }
        
        @Override
        public void close() {
            // Cleanup resources
        }
    }
    
    public static void main(String[] args) {
        FinOpsStreamProcessor processor = new FinOpsStreamProcessor();
        processor.buildCostProcessingPipeline();
    }
}

  

-- Snowflake FinOps Data Model

-- Create staging table for Kafka events
CREATE OR REPLACE TABLE finops_staging.cost_events_raw (
    record_content VARIANT,
    record_metadata VARIANT,
    loaded_at TIMESTAMP_NTZ DEFAULT CURRENT_TIMESTAMP()
);

CREATE OR REPLACE TABLE finops_staging.business_events_raw (
    record_content VARIANT,
    record_metadata VARIANT,
    loaded_at TIMESTAMP_NTZ DEFAULT CURRENT_TIMESTAMP()
);

-- Create curated tables
CREATE OR REPLACE TABLE finops_curated.cost_events (
    event_id STRING PRIMARY KEY,
    timestamp TIMESTAMP_NTZ,
    event_type STRING,
    service STRING,
    region STRING,
    cost_amount NUMBER(15,4),
    resource_id STRING,
    business_unit STRING,
    project_id STRING,
    environment STRING,
    metadata VARIANT,
    loaded_at TIMESTAMP_NTZ
);

CREATE OR REPLACE TABLE finops_curated.business_events (
    event_id STRING PRIMARY KEY,
    timestamp TIMESTAMP_NTZ,
    event_type STRING,
    user_id STRING,
    feature STRING,
    action STRING,
    revenue_impact NUMBER(15,4),
    business_unit STRING,
    metadata VARIANT,
    loaded_at TIMESTAMP_NTZ
);

-- Create cost aggregation tables
CREATE OR REPLACE TABLE finops_aggregated.daily_cost_summary (
    date DATE,
    business_unit STRING,
    service STRING,
    environment STRING,
    total_cost NUMBER(15,4),
    cost_trend STRING,
    week_over_week_change NUMBER(10,4),
    budget_utilization NUMBER(5,2),
    PRIMARY KEY (date, business_unit, service, environment)
);

CREATE OR REPLACE TABLE finops_aggregated.cost_anomalies (
    anomaly_id STRING PRIMARY KEY,
    detected_at TIMESTAMP_NTZ,
    anomaly_type STRING,
    service STRING,
    cost_amount NUMBER(15,4),
    expected_amount NUMBER(15,4),
    deviation_percent NUMBER(10,4),
    business_impact STRING,
    resolved BOOLEAN DEFAULT FALSE
);

-- Create views for common queries
CREATE OR REPLACE VIEW finops_reporting.cost_by_business_unit AS
SELECT 
    DATE_TRUNC('DAY', timestamp) as cost_date,
    business_unit,
    SUM(cost_amount) as daily_cost,
    LAG(SUM(cost_amount), 7) OVER (PARTITION BY business_unit ORDER BY cost_date) as cost_7_days_ago,
    (SUM(cost_amount) - LAG(SUM(cost_amount), 7) OVER (PARTITION BY business_unit ORDER BY cost_date)) / 
    LAG(SUM(cost_amount), 7) OVER (PARTITION BY business_unit ORDER BY cost_date) * 100 as week_over_week_change
FROM finops_curated.cost_events
WHERE timestamp >= DATEADD('DAY', -30, CURRENT_DATE())
GROUP BY cost_date, business_unit
ORDER BY cost_date DESC, business_unit;

CREATE OR REPLACE VIEW finops_reporting.roi_analysis AS
SELECT 
    ce.business_unit,
    DATE_TRUNC('DAY', ce.timestamp) as analysis_date,
    SUM(ce.cost_amount) as total_cost,
    SUM(be.revenue_impact) as total_revenue,
    CASE 
        WHEN SUM(ce.cost_amount) = 0 THEN NULL
        ELSE (SUM(be.revenue_impact) - SUM(ce.cost_amount)) / SUM(ce.cost_amount) * 100 
    END as roi_percentage,
    COUNT(DISTINCT ce.event_id) as cost_events,
    COUNT(DISTINCT be.event_id) as revenue_events
FROM finops_curated.cost_events ce
LEFT JOIN finops_curated.business_events be 
    ON ce.business_unit = be.business_unit
    AND DATE_TRUNC('HOUR', ce.timestamp) = DATE_TRUNC('HOUR', be.timestamp)
    AND be.revenue_impact > 0
WHERE ce.timestamp >= DATEADD('DAY', -7, CURRENT_DATE())
GROUP BY ce.business_unit, analysis_date
ORDER BY analysis_date DESC, roi_percentage DESC;

-- Stored procedure for cost forecasting
CREATE OR REPLACE PROCEDURE finops_analysis.forecast_costs(
    business_unit STRING, 
    forecast_days NUMBER
)
RETURNS TABLE (
    forecast_date DATE,
    predicted_cost NUMBER(15,4),
    confidence_interval_lower NUMBER(15,4),
    confidence_interval_upper NUMBER(15,4)
)
LANGUAGE SQL
AS
$$
DECLARE
    training_data RESULTSET;
BEGIN
    -- Use historical data for forecasting
    training_data := (
        SELECT 
            DATE_TRUNC('DAY', timestamp) as cost_date,
            SUM(cost_amount) as daily_cost
        FROM finops_curated.cost_events
        WHERE business_unit = :business_unit
            AND timestamp >= DATEADD('DAY', -90, CURRENT_DATE())
        GROUP BY cost_date
        ORDER BY cost_date
    );
    
    -- Simple linear regression forecast (in production, use more sophisticated models)
    RETURN (
        WITH historical AS (
            SELECT 
                cost_date,
                daily_cost,
                ROW_NUMBER() OVER (ORDER BY cost_date) as day_number
            FROM TABLE(:training_data)
        ),
        regression AS (
            SELECT 
                AVG(daily_cost) as avg_cost,
                AVG(day_number) as avg_day,
                SUM((day_number - avg_day) * (daily_cost - avg_cost)) / 
                SUM((day_number - avg_day) * (day_number - avg_day)) as slope
            FROM historical
            CROSS JOIN (SELECT AVG(daily_cost) as avg_cost, AVG(day_number) as avg_day FROM historical) stats
        ),
        forecast_dates AS (
            SELECT 
                DATEADD('DAY', ROW_NUMBER() OVER (ORDER BY SEQ4()), CURRENT_DATE()) as forecast_date,
                ROW_NUMBER() OVER (ORDER BY SEQ4()) as forecast_day
            FROM TABLE(GENERATOR(ROWCOUNT => :forecast_days))
        )
        SELECT 
            fd.forecast_date,
            r.avg_cost + r.slope * (MAX(h.day_number) + fd.forecast_day - r.avg_day) as predicted_cost,
            (r.avg_cost + r.slope * (MAX(h.day_number) + fd.forecast_day - r.avg_day)) * 0.9 as confidence_interval_lower,
            (r.avg_cost + r.slope * (MAX(h.day_number) + fd.forecast_day - r.avg_day)) * 1.1 as confidence_interval_upper
        FROM forecast_dates fd
        CROSS JOIN regression r
        CROSS JOIN historical h
        GROUP BY fd.forecast_date, fd.forecast_day, r.avg_cost, r.avg_day, r.slope
        ORDER BY fd.forecast_date
    );
END;
$$;

-- Automated anomaly detection task
CREATE OR REPLACE TASK finops_tasks.detect_cost_anomalies
    WAREHOUSE = 'finops_wh'
    SCHEDULE = '5 MINUTE'
AS
BEGIN
    INSERT INTO finops_aggregated.cost_anomalies (
        anomaly_id, detected_at, anomaly_type, service, cost_amount, 
        expected_amount, deviation_percent, business_impact
    )
    WITH current_period AS (
        SELECT 
            service,
            SUM(cost_amount) as current_cost
        FROM finops_curated.cost_events
        WHERE timestamp >= DATEADD('HOUR', -1, CURRENT_TIMESTAMP())
        GROUP BY service
    ),
    historical_avg AS (
        SELECT 
            service,
            AVG(cost_amount) as avg_cost,
            STDDEV(cost_amount) as std_cost
        FROM finops_curated.cost_events
        WHERE timestamp >= DATEADD('DAY', -7, CURRENT_TIMESTAMP())
            AND HOUR(timestamp) = HOUR(CURRENT_TIMESTAMP())
        GROUP BY service
    )
    SELECT 
        UUID_STRING() as anomaly_id,
        CURRENT_TIMESTAMP() as detected_at,
        'cost_spike' as anomaly_type,
        cp.service,
        cp.current_cost,
        ha.avg_cost as expected_amount,
        ((cp.current_cost - ha.avg_cost) / ha.avg_cost) * 100 as deviation_percent,
        CASE 
            WHEN ((cp.current_cost - ha.avg_cost) / ha.avg_cost) * 100 > 100 THEN 'CRITICAL'
            WHEN ((cp.current_cost - ha.avg_cost) / ha.avg_cost) * 100 > 50 THEN 'HIGH'
            ELSE 'MEDIUM'
        END as business_impact
    FROM current_period cp
    JOIN historical_avg ha ON cp.service = ha.service
    WHERE cp.current_cost > ha.avg_cost + (ha.std_cost * 2)
        AND cp.current_cost > 10; -- Minimum cost threshold
END;

-- Enable the task
ALTER TASK finops_tasks.detect_cost_anomalies RESUME;

  

import numpy as np
import pandas as pd
from sklearn.ensemble import IsolationForest
from sklearn.preprocessing import StandardScaler
from tensorflow.keras.models import Sequential
from tensorflow.keras.layers import LSTM, Dense, Dropout
from prometheus_api_client import PrometheusConnect
import warnings
warnings.filterwarnings('ignore')

class AIOpsAnomalyDetector:
    def __init__(self, prometheus_url: str, threshold: float = 0.85):
        self.prometheus = PrometheusConnect(url=prometheus_url)
        self.scaler = StandardScaler()
        self.isolation_forest = IsolationForest(
            contamination=0.1, 
            random_state=42,
            n_estimators=100
        )
        self.threshold = threshold
        self.metrics_history = {}
        
    def collect_metrics(self, query: str, hours: int = 24) -> pd.DataFrame:
        """Collect metrics from Prometheus for analysis"""
        try:
            # Query Prometheus for historical data
            metric_data = self.prometheus.custom_query_range(
                query=query,
                start_time=pd.Timestamp.now() - pd.Timedelta(hours=hours),
                end_time=pd.Timestamp.now(),
                step="1m"
            )
            
            # Convert to DataFrame
            if metric_data:
                df = pd.DataFrame(metric_data[0]['values'], 
                                columns=['timestamp', 'value'])
                df['timestamp'] = pd.to_datetime(df['timestamp'], unit='s')
                df['value'] = pd.to_numeric(df['value'])
                df.set_index('timestamp', inplace=True)
                return df
            return pd.DataFrame()
            
        except Exception as e:
            print(f"Error collecting metrics: {e}")
            return pd.DataFrame()
    
    def build_lstm_forecaster(self, sequence_length: int = 60) -> Sequential:
        """Build LSTM model for time series forecasting"""
        model = Sequential([
            LSTM(50, return_sequences=True, 
                 input_shape=(sequence_length, 1)),
            Dropout(0.2),
            LSTM(50, return_sequences=False),
            Dropout(0.2),
            Dense(25),
            Dense(1)
        ])
        
        model.compile(optimizer='adam', loss='mse')
        return model
    
    def detect_statistical_anomalies(self, metric_data: pd.DataFrame) -> pd.DataFrame:
        """Detect anomalies using statistical methods"""
        df = metric_data.copy()
        
        # Calculate rolling statistics
        df['rolling_mean'] = df['value'].rolling(window=30).mean()
        df['rolling_std'] = df['value'].rolling(window=30).std()
        
        # Define anomaly thresholds (3 sigma)
        df['upper_bound'] = df['rolling_mean'] + 3 * df['rolling_std']
        df['lower_bound'] = df['rolling_mean'] - 3 * df['rolling_std']
        
        # Identify anomalies
        df['is_anomaly_statistical'] = (
            (df['value'] > df['upper_bound']) | 
            (df['value'] < df['lower_bound'])
        )
        
        return df
    
    def detect_ml_anomalies(self, metric_data: pd.DataFrame) -> pd.DataFrame:
        """Detect anomalies using machine learning"""
        df = metric_data.copy()
        
        # Prepare features for ML
        features = self._engineer_features(df)
        
        # Scale features
        scaled_features = self.scaler.fit_transform(features)
        
        # Train Isolation Forest
        anomalies = self.isolation_forest.fit_predict(scaled_features)
        
        df['is_anomaly_ml'] = anomalies == -1
        df['anomaly_score'] = self.isolation_forest.decision_function(scaled_features)
        
        return df
    
    def _engineer_features(self, df: pd.DataFrame) -> np.ndarray:
        """Engineer features for anomaly detection"""
        features = []
        
        # Raw value
        features.append(df['value'].values.reshape(-1, 1))
        
        # Rolling statistics
        features.append(df['value'].rolling(window=5).mean().fillna(0).values.reshape(-1, 1))
        features.append(df['value'].rolling(window=15).std().fillna(0).values.reshape(-1, 1))
        
        # Rate of change
        features.append(df['value'].diff().fillna(0).values.reshape(-1, 1))
        
        # Hour of day and day of week (for seasonality)
        features.append(df.index.hour.values.reshape(-1, 1))
        features.append(df.index.dayofweek.values.reshape(-1, 1))
        
        return np.hstack(features)
    
    def predict_future_anomalies(self, metric_data: pd.DataFrame, 
                               forecast_hours: int = 1) -> dict:
        """Predict potential future anomalies using LSTM"""
        try:
            # Prepare data for LSTM
            sequence_data = self._prepare_sequences(metric_data['value'].values)
            
            if len(sequence_data) == 0:
                return {"error": "Insufficient data for forecasting"}
            
            # Build and train LSTM model
            model = self.build_lstm_forecaster()
            
            X, y = sequence_data[:, :-1], sequence_data[:, -1]
            X = X.reshape((X.shape[0], X.shape[1], 1))
            
            # Train model (in production, this would be pre-trained)
            model.fit(X, y, epochs=10, batch_size=32, verbose=0)
            
            # Generate forecast
            last_sequence = sequence_data[-1, :-1].reshape(1, -1, 1)
            predictions = []
            
            for _ in range(forecast_hours * 60):  # 1-minute intervals
                pred = model.predict(last_sequence, verbose=0)[0][0]
                predictions.append(pred)
                
                # Update sequence for next prediction
                last_sequence = np.roll(last_sequence, -1)
                last_sequence[0, -1, 0] = pred
            
            # Analyze predictions for anomalies
            forecast_df = pd.DataFrame({
                'timestamp': pd.date_range(
                    start=metric_data.index[-1] + pd.Timedelta(minutes=1),
                    periods=len(predictions),
                    freq='1min'
                ),
                'predicted_value': predictions
            })
            
            # Detect anomalies in forecast
            forecast_anomalies = self.detect_statistical_anomalies(
                forecast_df.set_index('timestamp')
            )
            
            return {
                'forecast': forecast_df,
                'anomaly_periods': forecast_anomalies[
                    forecast_anomalies['is_anomaly_statistical']
                ].index.tolist(),
                'confidence': 0.85
            }
            
        except Exception as e:
            return {"error": str(e)}
    
    def run_comprehensive_analysis(self, metric_queries: dict) -> dict:
        """Run comprehensive anomaly analysis across multiple metrics"""
        results = {}
        
        for metric_name, query in metric_queries.items():
            print(f"Analyzing {metric_name}...")
            
            # Collect data
            metric_data = self.collect_metrics(query)
            
            if metric_data.empty:
                continue
            
            # Run multiple detection methods
            statistical_result = self.detect_statistical_anomalies(metric_data)
            ml_result = self.detect_ml_anomalies(metric_data)
            
            # Combine results
            combined_anomalies = (
                statistical_result['is_anomaly_statistical'] | 
                ml_result['is_anomaly_ml']
            )
            
            # Calculate confidence scores
            confidence_scores = self._calculate_confidence(
                statistical_result, ml_result
            )
            
            results[metric_name] = {
                'data': metric_data,
                'anomalies': combined_anomalies,
                'confidence_scores': confidence_scores,
                'anomaly_count': combined_anomalies.sum(),
                'forecast': self.predict_future_anomalies(metric_data)
            }
        
        return results
    
    def _calculate_confidence(self, stat_result: pd.DataFrame, 
                            ml_result: pd.DataFrame) -> pd.Series:
        """Calculate confidence scores for anomaly detections"""
        # Simple weighted average of different detection methods
        stat_confidence = stat_result['is_anomaly_statistical'].astype(float) * 0.6
        ml_confidence = (ml_result['anomaly_score'] < -0.1).astype(float) * 0.4
        
        return stat_confidence + ml_confidence

# Example usage
def main():
    # Initialize detector
    detector = AIOpsAnomalyDetector("http://prometheus:9090")
    
    # Define metrics to monitor
    metric_queries = {
        'cpu_usage': 'rate(container_cpu_usage_seconds_total[5m])',
        'memory_usage': 'container_memory_usage_bytes',
        'http_requests': 'rate(http_requests_total[5m])',
        'response_time': 'histogram_quantile(0.95, rate(http_request_duration_seconds_bucket[5m]))'
    }
    
    # Run analysis
    results = detector.run_comprehensive_analysis(metric_queries)
    
    # Generate report
    for metric, result in results.items():
        print(f"\n{metric.upper()} Analysis:")
        print(f"Anomalies detected: {result['anomaly_count']}")
        print(f"Latest anomaly: {result['anomalies'].iloc[-1] if len(result['anomalies']) > 0 else 'None'}")
        
        if 'forecast' in result and 'anomaly_periods' in result['forecast']:
            print(f"Future anomalies predicted: {len(result['forecast']['anomaly_periods'])}")

if __name__ == "__main__":
    main()

  

import networkx as nx
import pandas as pd
import numpy as np
from sklearn.ensemble import RandomForestRegressor
from sklearn.metrics import mean_squared_error
from causalnex.structure import DAGRegressor
from typing import Dict, List, Tuple
import matplotlib.pyplot as plt

class RootCauseAnalyzer:
    def __init__(self):
        self.service_graph = nx.DiGraph()
        self.causal_model = None
        self.feature_importance = {}
        
    def build_service_dependency_graph(self, service_data: Dict) -> nx.DiGraph:
        """Build service dependency graph from monitoring data"""
        G = nx.DiGraph()
        
        # Add nodes (services)
        for service, metrics in service_data.items():
            G.add_node(service, 
                      metrics=metrics,
                      health_score=self._calculate_health_score(metrics))
        
        # Add edges based on call patterns and dependencies
        for service in service_data.keys():
            dependencies = self._infer_dependencies(service, service_data)
            for dep in dependencies:
                G.add_edge(dep, service, 
                          weight=self._calculate_dependency_strength(service, dep))
        
        return G
    
    def perform_causal_analysis(self, incident_data: pd.DataFrame, 
                              target_metric: str) -> Dict:
        """Perform causal analysis to identify root causes"""
        # Prepare data for causal inference
        causal_data = self._prepare_causal_data(incident_data)
        
        # Use DAG regressor for causal structure learning
        self.causal_model = DAGRegressor(
            alpha=0.1,
            beta=1.0,
            fit_intercept=True,
            hidden_layer_units=None
        )
        
        # Learn causal structure
        self.causal_model.fit(causal_data)
        
        # Identify potential causes for the target metric
        root_causes = self._identify_root_causes(
            causal_data, target_metric, self.causal_model
        )
        
        return {
            'root_causes': root_causes,
            'causal_graph': self.causal_model,
            'confidence_scores': self._calculate_causal_confidence(root_causes)
        }
    
    def analyze_incident_impact(self, service_graph: nx.DiGraph,
                              affected_service: str) -> Dict:
        """Analyze potential impact of an incident across the service graph"""
        # Calculate propagation paths
        propagation_paths = list(nx.all_simple_paths(
            service_graph, 
            affected_service,
            [node for node in service_graph.nodes() if node != affected_service]
        ))
        
        # Estimate impact severity
        impact_analysis = {}
        for path in propagation_paths:
            if len(path) > 1:  # Valid propagation path
                impact_score = self._calculate_impact_score(path, service_graph)
                impact_analysis[tuple(path)] = impact_score
        
        return {
            'affected_service': affected_service,
            'propagation_paths': impact_analysis,
            'blast_radius': len(impact_analysis),
            'critical_services_at_risk': self._identify_critical_services(impact_analysis)
        }
    
    def _prepare_causal_data(self, incident_data: pd.DataFrame) -> pd.DataFrame:
        """Prepare time series data for causal analysis"""
        # Feature engineering for causal inference
        features = []
        
        for column in incident_data.columns:
            # Original values
            features.append(incident_data[column])
            
            # Lagged features
            for lag in [1, 5, 15]:  # 1, 5, 15 minute lags
                features.append(incident_data[column].shift(lag).fillna(method='bfill'))
            
            # Rolling statistics
            features.append(incident_data[column].rolling(window=10).mean().fillna(method='bfill'))
            features.append(incident_data[column].rolling(window=10).std().fillna(method='bfill'))
            
            # Rate of change
            features.append(incident_data[column].diff().fillna(0))
        
        causal_df = pd.concat(features, axis=1)
        causal_df.columns = [f'feature_{i}' for i in range(len(causal_df.columns))]
        
        return causal_df.fillna(0)
    
    def _identify_root_causes(self, causal_data: pd.DataFrame,
                            target_metric: str, causal_model) -> List[Tuple]:
        """Identify potential root causes using causal inference"""
        root_causes = []
        
        # Get feature importance from causal model
        if hasattr(causal_model, 'feature_importances_'):
            importances = causal_model.feature_importances_
            
            # Map back to original metrics
            for idx, importance in enumerate(importances):
                if importance > 0.1:  # Threshold for significance
                    original_metric = self._map_feature_to_metric(idx, causal_data.columns)
                    root_causes.append((original_metric, importance))
        
        # Sort by importance
        root_causes.sort(key=lambda x: x[1], reverse=True)
        
        return root_causes
    
    def _calculate_impact_score(self, path: List[str], 
                              graph: nx.DiGraph) -> float:
        """Calculate impact score for a propagation path"""
        score = 0.0
        
        for i in range(len(path) - 1):
            source, target = path[i], path[i+1]
            
            # Consider edge weight and node criticality
            edge_weight = graph[source][target].get('weight', 1.0)
            target_criticality = graph.nodes[target].get('criticality', 1.0)
            
            score += edge_weight * target_criticality
        
        return score
    
    def _infer_dependencies(self, service: str, service_data: Dict) -> List[str]:
        """Infer service dependencies from monitoring data"""
        dependencies = []
        
        # Simple heuristic based on correlation in metrics
        for other_service, other_metrics in service_data.items():
            if other_service != service:
                # Calculate correlation between service metrics
                correlation = self._calculate_service_correlation(
                    service_data[service], 
                    other_metrics
                )
                
                if correlation > 0.7:  # High correlation threshold
                    dependencies.append(other_service)
        
        return dependencies
    
    def _calculate_service_correlation(self, metrics1: Dict, 
                                    metrics2: Dict) -> float:
        """Calculate correlation between two services' metrics"""
        # Convert metrics to comparable format
        m1_values = list(metrics1.values()) if isinstance(metrics1, dict) else [metrics1]
        m2_values = list(metrics2.values()) if isinstance(metrics2, dict) else [metrics2]
        
        # Ensure same length
        min_len = min(len(m1_values), len(m2_values))
        m1_values = m1_values[:min_len]
        m2_values = m2_values[:min_len]
        
        if min_len > 1:
            return np.corrcoef(m1_values, m2_values)[0, 1]
        return 0.0
    
    def _calculate_health_score(self, metrics: Dict) -> float:
        """Calculate overall health score for a service"""
        if not metrics:
            return 1.0
        
        # Simple weighted average of normalized metrics
        weights = {
            'cpu_usage': 0.3,
            'memory_usage': 0.3,
            'error_rate': 0.2,
            'latency': 0.2
        }
        
        score = 0.0
        total_weight = 0.0
        
        for metric, weight in weights.items():
            if metric in metrics:
                # Normalize metric value (lower is better for most metrics)
                normalized_value = 1.0 - min(metrics[metric] / 100.0, 1.0)
                score += normalized_value * weight
                total_weight += weight
        
        return score / total_weight if total_weight > 0 else 1.0

# Example usage
def analyze_production_incident():
    analyzer = RootCauseAnalyzer()
    
    # Simulate incident data
    incident_data = pd.DataFrame({
        'api_gateway_cpu': [45, 48, 85, 92, 88, 46, 44],
        'user_service_memory': [65, 68, 72, 95, 91, 67, 66],
        'database_connections': [120, 125, 580, 620, 590, 130, 125],
        'payment_service_errors': [2, 3, 45, 52, 48, 4, 2],
        'response_time_p95': [120, 125, 480, 520, 490, 130, 125]
    })
    
    # Build service dependency graph
    service_data = {
        'api_gateway': {'cpu': 85, 'memory': 45, 'errors': 2},
        'user_service': {'cpu': 72, 'memory': 95, 'errors': 45},
        'database': {'connections': 580, 'latency': 220},
        'payment_service': {'cpu': 65, 'errors': 52, 'latency': 480}
    }
    
    dependency_graph = analyzer.build_service_dependency_graph(service_data)
    
    # Perform root cause analysis
    rca_results = analyzer.perform_causal_analysis(incident_data, 'response_time_p95')
    
    # Analyze incident impact
    impact_analysis = analyzer.analyze_incident_impact(dependency_graph, 'user_service')
    
    print("=== ROOT CAUSE ANALYSIS RESULTS ===")
    print(f"Primary Root Cause: {rca_results['root_causes'][0] if rca_results['root_causes'] else 'Unknown'}")
    print(f"Blast Radius: {impact_analysis['blast_radius']} services affected")
    print(f"Critical Services at Risk: {impact_analysis['critical_services_at_risk']}")

if __name__ == "__main__":
    analyze_production_incident()

  

# ai-ops/incident-response-config.yaml
apiVersion: aiops.lktechacademy.com/v1
kind: IncidentResponsePolicy
metadata:
  name: production-auto-remediation
  namespace: ai-ops
spec:
  enabled: true
  severityThreshold: high
  autoRemediation:
    enabled: true
    maxConcurrentActions: 3
    coolDownPeriod: 300s

  detectionRules:
    - name: "high-cpu-anomaly"
      condition: "cpu_usage > 90 AND anomaly_score > 0.8"
      severity: "high"
      metrics:
        - "container_cpu_usage_seconds_total"
        - "node_cpu_usage"
      window: "5m"
      
    - name: "memory-leak-pattern"
      condition: "memory_usage_trend > 0.1 AND duration > 900"
      severity: "medium"
      metrics:
        - "container_memory_usage_bytes"
        - "container_memory_working_set_bytes"
      window: "15m"
      
    - name: "latency-spike-correlation"
      condition: "response_time_p95 > 1000 AND error_rate > 0.1"
      severity: "critical"
      metrics:
        - "http_request_duration_seconds"
        - "http_requests_total"
      window: "2m"

  remediationActions:
    - name: "restart-pod-high-cpu"
      trigger: "high-cpu-anomaly"
      action: "kubernetes_rollout_restart"
      parameters:
        namespace: "{{ .Namespace }}"
        deployment: "{{ .Deployment }}"
      conditions:
        - "restart_count < 3"
        - "uptime > 300"
        
    - name: "scale-out-latency-spike"
      trigger: "latency-spike-correlation"
      action: "kubernetes_scale"
      parameters:
        namespace: "{{ .Namespace }}"
        deployment: "{{ .Deployment }}"
        replicas: "{{ .CurrentReplicas | add 2 }}"
      conditions:
        - "current_cpu < 70"
        - "available_nodes > 1"
        
    - name: "failover-database-connections"
      trigger: "database_connection_exhaustion"
      action: "database_failover"
      parameters:
        cluster: "{{ .DatabaseCluster }}"
        failoverType: "reader"
      conditions:
        - "replica_lag < 30"
        - "failover_count_today < 2"

  escalationPolicies:
    - name: "immediate-sre-page"
      conditions:
        - "severity == 'critical'"
        - "business_impact == 'high'"
        - "auto_remediation_failed == true"
      actions:
        - "pagerduty_trigger_incident"
        - "slack_notify_channel"
        - "create_jira_ticket"
        
    - name: "engineering-notification"
      conditions:
        - "severity == 'high'"
        - "team_working_hours == true"
      actions:
        - "slack_notify_team"
        - "email_digest"

  learningConfiguration:
    feedbackLoop: true
    modelRetraining:
      schedule: "0 2 * * *"  # Daily at 2 AM
      metrics:
        - "false_positive_rate"
        - "mean_time_to_detect"
        - "mean_time_to_resolve"
    continuousImprovement:
      enabled: true
      optimizationGoal: "reduce_mttr"
---
# ai-ops/response-orchestrator.py
import asyncio
import json
import logging
from typing import Dict, List
from kubernetes import client, config
import redis
import aiohttp

class IncidentResponseOrchestrator:
    def __init__(self, kubeconfig_path: str = None):
        # Load Kubernetes configuration
        try:
            config.load_incluster_config()  # In-cluster
        except:
            config.load_kube_config(kubeconfig_path)  # Local development
        
        self.k8s_apps = client.AppsV1Api()
        self.k8s_core = client.CoreV1Api()
        self.redis_client = redis.Redis(host='redis', port=6379, db=0)
        self.session = aiohttp.ClientSession()
        
        self.logger = logging.getLogger(__name__)
        
    async def handle_incident(self, incident_data: Dict) -> Dict:
        """Orchestrate incident response based on AI analysis"""
        self.logger.info(f"Processing incident: {incident_data['incident_id']}")
        
        try:
            # Validate incident
            if not self._validate_incident(incident_data):
                return {"status": "skipped", "reason": "invalid_incident"}
            
            # Check if similar incident recently handled
            if await self._is_duplicate_incident(incident_data):
                return {"status": "skipped", "reason": "duplicate"}
            
            # Determine appropriate response
            response_plan = await self._create_response_plan(incident_data)
            
            # Execute remediation actions
            results = await self._execute_remediation(response_plan)
            
            # Log results for learning
            await self._log_incident_response(incident_data, results)
            
            return {
                "status": "completed",
                "incident_id": incident_data['incident_id'],
                "actions_taken": results,
                "response_time_seconds": response_plan.get('response_time', 0)
            }
            
        except Exception as e:
            self.logger.error(f"Error handling incident: {e}")
            return {"status": "failed", "error": str(e)}
    
    async def _create_response_plan(self, incident_data: Dict) -> Dict:
        """Create optimized response plan based on incident analysis"""
        response_plan = {
            'incident_id': incident_data['incident_id'],
            'severity': incident_data['severity'],
            'detected_at': incident_data['timestamp'],
            'actions': [],
            'escalation_required': False
        }
        
        # AI-powered decision making
        recommended_actions = await self._ai_recommend_actions(incident_data)
        
        # Filter actions based on current system state
        feasible_actions = await self._filter_feasible_actions(recommended_actions)
        
        # Prioritize actions
        prioritized_actions = self._prioritize_actions(feasible_actions, incident_data)
        
        response_plan['actions'] = prioritized_actions
        response_plan['escalation_required'] = self._requires_escalation(incident_data)
        
        return response_plan
    
    async def _ai_recommend_actions(self, incident_data: Dict) -> List[Dict]:
        """Use AI to recommend remediation actions"""
        # This would integrate with your ML model
        # For now, using rule-based recommendations
        
        recommendations = []
        
        if incident_data.get('root_cause') == 'high_cpu':
            recommendations.append({
                'type': 'restart_pod',
                'confidence': 0.85,
                'parameters': {
                    'namespace': incident_data.get('namespace'),
                    'deployment': incident_data.get('deployment')
                }
            })
            
        elif incident_data.get('root_cause') == 'memory_leak':
            recommendations.append({
                'type': 'scale_up',
                'confidence': 0.75,
                'parameters': {
                    'namespace': incident_data.get('namespace'),
                    'deployment': incident_data.get('deployment'),
                    'replicas': '+2'
                }
            })
            
        elif incident_data.get('root_cause') == 'database_contention':
            recommendations.append({
                'type': 'database_failover',
                'confidence': 0.90,
                'parameters': {
                    'cluster': incident_data.get('database_cluster')
                }
            })
        
        return recommendations
    
    async def _execute_remediation(self, response_plan: Dict) -> List[Dict]:
        """Execute remediation actions safely"""
        results = []
        
        for action in response_plan['actions']:
            try:
                if action['type'] == 'restart_pod':
                    result = await self._restart_deployment(
                        action['parameters']['namespace'],
                        action['parameters']['deployment']
                    )
                    results.append({
                        'action': 'restart_pod',
                        'status': 'success' if result else 'failed',
                        'details': result
                    })
                    
                elif action['type'] == 'scale_up':
                    result = await self._scale_deployment(
                        action['parameters']['namespace'],
                        action['parameters']['deployment'],
                        action['parameters']['replicas']
                    )
                    results.append({
                        'action': 'scale_up',
                        'status': 'success' if result else 'failed',
                        'details': result
                    })
                    
            except Exception as e:
                results.append({
                    'action': action['type'],
                    'status': 'error',
                    'error': str(e)
                })
        
        return results
    
    async def _restart_deployment(self, namespace: str, deployment: str) -> bool:
        """Restart a Kubernetes deployment"""
        try:
            # This would actually call Kubernetes API
            self.logger.info(f"Restarting deployment {deployment} in {namespace}")
            
            # Simulate API call
            await asyncio.sleep(2)
            
            return True
        except Exception as e:
            self.logger.error(f"Failed to restart deployment: {e}")
            return False
    
    async def _scale_deployment(self, namespace: str, deployment: str, replicas: str) -> bool:
        """Scale a Kubernetes deployment"""
        try:
            self.logger.info(f"Scaling deployment {deployment} in {namespace} to {replicas}")
            
            # Simulate API call
            await asyncio.sleep(1)
            
            return True
        except Exception as e:
            self.logger.error(f"Failed to scale deployment: {e}")
            return False

# Example usage
async def main():
    orchestrator = IncidentResponseOrchestrator()
    
    # Simulate incident
    incident = {
        'incident_id': 'inc-20250115-001',
        'timestamp': '2025-01-15T10:30:00Z',
        'severity': 'high',
        'root_cause': 'high_cpu',
        'namespace': 'production',
        'deployment': 'user-service',
        'metrics': {
            'cpu_usage': 95,
            'memory_usage': 65,
            'anomaly_score': 0.92
        }
    }
    
    result = await orchestrator.handle_incident(incident)
    print(f"Incident response result: {result}")

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

  

# infrastructure/carbon-aware-eks.tf

# Carbon intensity data source
data "http" "carbon_intensity" {
  url = "https://api.electricitymap.org/v3/carbon-intensity/latest?zone=US-CAL"
  
  request_headers = {
    Accept = "application/json"
    Auth-Token = var.carbon_api_key
  }
}

# Carbon-aware EKS cluster configuration
resource "aws_eks_cluster" "green_cluster" {
  name     = "carbon-aware-${var.environment}"
  version  = "1.28"
  role_arn = aws_iam_role.eks_cluster.arn

  vpc_config {
    subnet_ids = var.carbon_optimized_subnets
  }

  # Enable carbon-aware scaling
  scaling_config {
    desired_size = local.carbon_optimal_size
    max_size     = 10
    min_size     = 1
  }

  # Carbon optimization tags
  tags = {
    Environment     = var.environment
    CarbonOptimized = "true"
    CostCenter      = "sustainability"
    AutoShutdown    = "enabled"
  }
}

# Carbon-aware node group
resource "aws_eks_node_group" "carbon_optimized" {
  cluster_name    = aws_eks_cluster.green_cluster.name
  node_group_name = "carbon-optimized-nodes"
  node_role_arn   = aws_iam_role.eks_node_group.arn
  subnet_ids      = var.carbon_optimized_subnets

  scaling_config {
    desired_size = local.calculate_optimal_capacity()
    max_size     = 15
    min_size     = 1
  }

  # Instance types optimized for energy efficiency
  instance_types = ["c6g.4xlarge", "m6g.4xlarge", "r6g.4xlarge"] # Graviton processors

  # Carbon-aware update strategy
  update_config {
    max_unavailable = 1
  }

  lifecycle {
    ignore_changes = [scaling_config[0].desired_size]
  }
}

# Carbon-aware auto-scaling policy
resource "aws_autoscaling_policy" "carbon_aware_scaling" {
  name                   = "carbon-aware-scaling"
  autoscaling_group_name = aws_eks_node_group.carbon_optimized.resources[0].autoscaling_groups[0].name
  policy_type            = "TargetTrackingScaling"

  target_tracking_configuration {
    predefined_metric_specification {
      predefined_metric_type = "ASGAverageCPUUtilization"
    }
    target_value = 65.0 # Optimized for energy efficiency
  }
}

# Locals for carbon calculations
locals {
  carbon_intensity = jsondecode(data.http.carbon_intensity.body).carbonIntensity
  
  # Calculate optimal cluster size based on carbon intensity
  calculate_optimal_capacity = () => {
    var.carbon_intensity < 200 ? 3 : (
      var.carbon_intensity < 400 ? 2 : 1
    )
  }
  
  carbon_optimal_size = local.calculate_optimal_capacity()
}

# Carbon monitoring and alerts
resource "aws_cloudwatch_dashboard" "carbon_dashboard" {
  dashboard_name = "Carbon-Monitoring-${var.environment}"

  dashboard_body = jsonencode({
    widgets = [
      {
        type   = "metric"
        x      = 0
        y      = 0
        width  = 12
        height = 6

        properties = {
          metrics = [
            ["AWS/EKS", "CPUUtilization", "ClusterName", aws_eks_cluster.green_cluster.name],
            [".", "MemoryUtilization", ".", "."],
            [".", "NetworkRxBytes", ".", "."],
            [".", "NetworkTxBytes", ".", "."]
          ]
          view    = "timeSeries"
          stacked = false
          region  = var.aws_region
          title   = "Cluster Performance vs Carbon Intensity"
          period  = 300
        }
      }
    ]
  })
}

# Output carbon efficiency metrics
output "carbon_efficiency_metrics" {
  description = "Carbon efficiency metrics for the deployment"
  value = {
    cluster_name          = aws_eks_cluster.green_cluster.name
    estimated_carbon_savings = local.calculate_carbon_savings()
    optimal_instance_type = "Graviton-based for 40% better performance per watt"
    carbon_aware_scaling  = "Enabled"
  }
}

  

# k8s/carbon-aware-scheduler.yaml
apiVersion: apps/v1
kind: Deployment
metadata:
  name: carbon-aware-scheduler
  namespace: kube-system
  labels:
    app: carbon-aware-scheduler
    sustainability: enabled
spec:
  replicas: 2
  selector:
    matchLabels:
      app: carbon-aware-scheduler
  template:
    metadata:
      labels:
        app: carbon-aware-scheduler
      annotations:
        carbon.optimization/enabled: "true"
    spec:
      serviceAccountName: carbon-scheduler
      containers:
      - name: scheduler
        image: k8s.gcr.io/carbon-aware-scheduler:v2.1.0
        args:
        - --carbon-api-endpoint=https://api.carbonintensity.org
        - --optimization-mode=balanced
        - --carbon-threshold=300
        - --region-preference=us-west-2,eu-west-1,us-east-1
        resources:
          requests:
            cpu: 100m
            memory: 256Mi
          limits:
            cpu: 500m
            memory: 1Gi
        env:
        - name: CARBON_API_KEY
          valueFrom:
            secretKeyRef:
              name: carbon-credentials
              key: api-key
        - name: SCHEDULING_STRATEGY
          value: "carbon-aware"
---
# Carbon-aware deployment with resource optimization
apiVersion: apps/v1
kind: Deployment
metadata:
  name: web-app-carbon-optimized
  labels:
    app: web-app
    sustainability-tier: "optimized"
spec:
  replicas: 3
  selector:
    matchLabels:
      app: web-app
  template:
    metadata:
      labels:
        app: web-app
      annotations:
        carbon.scheduling/preferred-time: "low-carbon-hours"
        carbon.scaling/strategy: "carbon-aware"
        autoscaling.alpha.kubernetes.io/conditions: '
          [{
            "type": "CarbonOptimized",
            "status": "True",
            "lastTransitionTime": "2025-01-15T10:00:00Z"
          }]'
    spec:
      affinity:
        nodeAffinity:
          preferredDuringSchedulingIgnoredDuringExecution:
          - weight: 100
            preference:
              matchExpressions:
              - key: kubernetes.io/arch
                operator: In
                values:
                - arm64
          - weight: 80
            preference:
              matchExpressions:
              - key: carbon.efficiency/score
                operator: Gt
                values:
                - "80"
          - weight: 60
            preference:
              matchExpressions:
              - key: topology.kubernetes.io/region
                operator: In
                values:
                - us-west-2
                - eu-west-1
      containers:
      - name: web-app
        image: my-registry/web-app:green-optimized
        ports:
        - containerPort: 8080
        resources:
          requests:
            cpu: 200m
            memory: 256Mi
          limits:
            cpu: 500m
            memory: 512Mi
        env:
        - name: CARBON_OPTIMIZATION
          value: "enabled"
        - name: ENERGY_EFFICIENT_MODE
          value: "true"
        livenessProbe:
          httpGet:
            path: /health
            port: 8080
          initialDelaySeconds: 30
          periodSeconds: 10
          timeoutSeconds: 5
          failureThreshold: 3
        readinessProbe:
          httpGet:
            path: /ready
            port: 8080
          initialDelaySeconds: 5
          periodSeconds: 5
        # Carbon-aware lifecycle hooks
        lifecycle:
          preStop:
            exec:
              command: ["/bin/sh", "-c", "echo 'Shutting down during high carbon hours'"]
---
# Carbon-aware HPA configuration
apiVersion: autoscaling/v2
kind: HorizontalPodAutoscaler
metadata:
  name: web-app-carbon-hpa
  annotations:
    carbon.scaling/strategy: "time-aware"
spec:
  scaleTargetRef:
    apiVersion: apps/v1
    kind: Deployment
    name: web-app-carbon-optimized
  minReplicas: 2
  maxReplicas: 10
  metrics:
  - type: Resource
    resource:
      name: cpu
      target:
        type: Utilization
        averageUtilization: 65
  - type: Resource
    resource:
      name: memory
      target:
        type: Utilization
        averageUtilization: 75
  behavior:
    scaleDown:
      stabilizationWindowSeconds: 300
      policies:
      - type: Percent
        value: 50
        periodSeconds: 60
      - type: Pods
        value: 2
        periodSeconds: 60
      selectPolicy: Min
    scaleUp:
      stabilizationWindowSeconds: 180
      policies:
      - type: Percent
        value: 25
        periodSeconds: 60
      - type: Pods
        value: 2
        periodSeconds: 60
      selectPolicy: Max
---
# Carbon metrics collector
apiVersion: v1
kind: ConfigMap
metadata:
  name: carbon-metrics-config
data:
  config.yaml: |
    carbon:
      enabled: true
      collection_interval: 5m
      metrics:
        - carbon_intensity
        - energy_consumption
        - cost_per_carbon_unit
      exporters:
        - prometheus
        - cloudwatch
      optimization_rules:
        - name: "scale_down_high_carbon"
          condition: "carbon_intensity > 400"
          action: "scale_replicas_by_percent"
          value: -50
        - name: "prefer_graviton"
          condition: "always"
          action: "node_selector"
          value: "kubernetes.io/arch=arm64"

  

#!/usr/bin/env python3
"""
Green Cloud Analytics: Carbon Footprint Monitoring and Optimization
"""

import asyncio
import aiohttp
import pandas as pd
from datetime import datetime, timedelta
from typing import Dict, List, Optional
from dataclasses import dataclass
import boto3
from prometheus_api_client import PrometheusConnect

@dataclass
class CarbonMetrics:
    timestamp: datetime
    carbon_intensity: float  # gCO2/kWh
    energy_consumption: float  # kWh
    estimated_emissions: float  # gCO2
    cost_usd: float
    region: str
    service: str

class GreenCloudAnalytics:
    def __init__(self, prometheus_url: str, aws_region: str = "us-west-2"):
        self.prometheus = PrometheusConnect(url=prometheus_url)
        self.cloudwatch = boto3.client('cloudwatch', region_name=aws_region)
        self.ce = boto3.client('ce', region_name=aws_region)
        self.carbon_data_cache = {}
        
    async def get_carbon_intensity(self, region: str) -> float:
        """Get real-time carbon intensity for cloud region"""
        cache_key = f"{region}_{datetime.now().strftime('%Y-%m-%d-%H')}"
        
        if cache_key in self.carbon_data_cache:
            return self.carbon_data_cache[cache_key]
        
        # Carbon intensity API (example using Electricity Maps)
        async with aiohttp.ClientSession() as session:
            async with session.get(
                f"https://api.electricitymap.org/v3/carbon-intensity/latest?zone={self._region_to_zone(region)}",
                headers={"auth-token": "YOUR_API_KEY"}
            ) as response:
                data = await response.json()
                carbon_intensity = data.get('carbonIntensity', 300)  # Default fallback
                self.carbon_data_cache[cache_key] = carbon_intensity
                return carbon_intensity
    
    def _region_to_zone(self, region: str) -> str:
        """Map AWS regions to carbon intensity zones"""
        zone_mapping = {
            'us-east-1': 'US-MIDA',
            'us-west-2': 'US-NW-PAC',
            'eu-west-1': 'IE',
            'eu-central-1': 'DE',
            'ap-southeast-1': 'SG'
        }
        return zone_mapping.get(region, 'US-CAL')
    
    async def calculate_service_emissions(self, service: str, region: str, 
                                        duration_hours: int = 1) -> CarbonMetrics:
        """Calculate carbon emissions for a specific cloud service"""
        # Get resource utilization metrics
        cpu_usage = self._get_cpu_usage(service, region, duration_hours)
        memory_usage = self._get_memory_usage(service, region, duration_hours)
        network_io = self._get_network_usage(service, region, duration_hours)
        
        # Calculate energy consumption (simplified model)
        energy_kwh = self._estimate_energy_consumption(cpu_usage, memory_usage, network_io)
        
        # Get carbon intensity
        carbon_intensity = await self.get_carbon_intensity(region)
        
        # Calculate emissions
        emissions_gco2 = energy_kwh * carbon_intensity
        
        # Get cost data
        cost = self._get_service_cost(service, region, duration_hours)
        
        return CarbonMetrics(
            timestamp=datetime.now(),
            carbon_intensity=carbon_intensity,
            energy_consumption=energy_kwh,
            estimated_emissions=emissions_gco2,
            cost_usd=cost,
            region=region,
            service=service
        )
    
    def _estimate_energy_consumption(self, cpu_usage: float, memory_usage: float, 
                                   network_io: float) -> float:
        """Estimate energy consumption based on resource usage"""
        # Simplified energy estimation model
        base_power_w = 50  # Base power for idle instance
        cpu_power_w = cpu_usage * 100  # CPU power scaling
        memory_power_w = memory_usage * 20  # Memory power scaling
        network_power_w = network_io * 5  # Network power scaling
        
        total_power_w = base_power_w + cpu_power_w + memory_power_w + network_power_w
        energy_kwh = (total_power_w * 1) / 1000  # Convert to kWh for 1 hour
        
        return energy_kwh
    
    def _get_cpu_usage(self, service: str, region: str, duration_hours: int) -> float:
        """Get average CPU usage for service"""
        query = f'avg(rate(container_cpu_usage_seconds_total{{service="{service}"}}[{duration_hours}h]))'
        result = self.prometheus.custom_query(query)
        return float(result[0]['value'][1]) if result else 0.5  # Default 50%
    
    def _get_memory_usage(self, service: str, region: str, duration_hours: int) -> float:
        """Get average memory usage for service"""
        query = f'avg(container_memory_usage_bytes{{service="{service}"}} / container_spec_memory_limit_bytes{{service="{service}"}})'
        result = self.prometheus.custom_query(query)
        return float(result[0]['value'][1]) if result else 0.6  # Default 60%
    
    def _get_network_usage(self, service: str, region: str, duration_hours: int) -> float:
        """Get network I/O usage"""
        query = f'avg(rate(container_network_receive_bytes_total{{service="{service}"}}[{duration_hours}h]))'
        result = self.prometheus.custom_query(query)
        return float(result[0]['value'][1]) / 1e6 if result else 10  # Default 10 MB/s
    
    def _get_service_cost(self, service: str, region: str, duration_hours: int) -> float:
        """Get cost for service usage"""
        # Simplified cost estimation
        instance_costs = {
            'c6g.4xlarge': 0.544,
            'm6g.4xlarge': 0.616,
            'r6g.4xlarge': 0.724
        }
        base_cost = instance_costs.get('c6g.4xlarge', 0.5)
        return base_cost * duration_hours
    
    def generate_optimization_recommendations(self, metrics: CarbonMetrics) -> List[Dict]:
        """Generate carbon optimization recommendations"""
        recommendations = []
        
        # High carbon intensity recommendation
        if metrics.carbon_intensity > 400:
            recommendations.append({
                'type': 'carbon_timing',
                'priority': 'high',
                'message': f'High carbon intensity ({metrics.carbon_intensity} gCO2/kWh). Consider shifting workload to low-carbon hours.',
                'estimated_savings': f'{metrics.estimated_emissions * 0.3:.2f} gCO2'
            })
        
        # Resource optimization
        if metrics.energy_consumption > 0.5:  # High energy usage
            recommendations.append({
                'type': 'resource_optimization',
                'priority': 'medium',
                'message': 'High energy consumption detected. Consider right-sizing instances.',
                'estimated_savings': f'{metrics.energy_consumption * 0.2:.2f} kWh'
            })
        
        # Architecture optimization
        if metrics.cost_usd > 1.0:  # High cost
            recommendations.append({
                'type': 'architecture',
                'priority': 'medium',
                'message': 'Consider migrating to Graviton instances for better performance per watt.',
                'estimated_savings': '40% better performance per watt'
            })
        
        return recommendations
    
    async def create_sustainability_report(self, services: List[str]) -> Dict:
        """Generate comprehensive sustainability report"""
        report = {
            'timestamp': datetime.now().isoformat(),
            'services_analyzed': [],
            'total_emissions_gco2': 0,
            'total_energy_kwh': 0,
            'total_cost_usd': 0,
            'recommendations': [],
            'carbon_efficiency_score': 0
        }
        
        for service in services:
            metrics = await self.calculate_service_emissions(service, 'us-west-2')
            report['services_analyzed'].append({
                'service': service,
                'emissions_gco2': metrics.estimated_emissions,
                'energy_kwh': metrics.energy_consumption,
                'cost_usd': metrics.cost_usd,
                'carbon_intensity': metrics.carbon_intensity
            })
            
            report['total_emissions_gco2'] += metrics.estimated_emissions
            report['total_energy_kwh'] += metrics.energy_consumption
            report['total_cost_usd'] += metrics.cost_usd
            
            # Add recommendations
            service_recommendations = self.generate_optimization_recommendations(metrics)
            report['recommendations'].extend(service_recommendations)
        
        # Calculate carbon efficiency score (0-100)
        report['carbon_efficiency_score'] = self._calculate_efficiency_score(report)
        
        return report
    
    def _calculate_efficiency_score(self, report: Dict) -> float:
        """Calculate overall carbon efficiency score"""
        total_work = sum(s['cost_usd'] for s in report['services_analyzed'])  # Using cost as proxy for work
        total_emissions = report['total_emissions_gco2']
        
        if total_emissions == 0:
            return 100
        
        efficiency = total_work / total_emissions
        max_efficiency = 1000  # Theoretical maximum
        score = min(100, (efficiency / max_efficiency) * 100)
        
        return score

# Example usage
async def main():
    analytics = GreenCloudAnalytics(
        prometheus_url="http://prometheus:9090",
        aws_region="us-west-2"
    )
    
    services = ["web-app", "api-service", "database-service"]
    report = await analytics.create_sustainability_report(services)
    
    print("=== Green Cloud Sustainability Report ===")
    print(f"Total Emissions: {report['total_emissions_gco2']:.2f} gCO2")
    print(f"Total Energy: {report['total_energy_kwh']:.2f} kWh")
    print(f"Carbon Efficiency Score: {report['carbon_efficiency_score']:.1f}/100")
    print(f"Recommendations: {len(report['recommendations'])}")
    
    for rec in report['recommendations']:
        print(f"- [{rec['priority'].upper()}] {rec['message']}")

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