// product-subgraph.ts - A federated Apollo subgraph
import { gql } from 'graphql-tag';
import { buildSubgraphSchema } from '@apollo/subgraph';
import { ApolloServer } from '@apollo/server';
import { startStandaloneServer } from '@apollo/server/standalone';

// 1. Define the GraphQL schema with @key directive for federation
const typeDefs = gql`
  extend schema
    @link(url: "https://specs.apollo.dev/federation/v2.3", 
          import: ["@key", "@shareable", "@external"])

  type Product @key(fields: "id") {
    id: ID!
    name: String!
    description: String
    price: Price!
    inventory: InventoryData
    reviews: [Review!]! @requires(fields: "id")
  }

  type Price {
    amount: Float!
    currency: String!
    discount: DiscountInfo
  }

  type DiscountInfo {
    percentage: Int
    validUntil: String
  }

  type InventoryData {
    stock: Int!
    warehouse: String
    lastRestocked: String
  }

  extend type Review @key(fields: "id") {
    id: ID! @external
    product: Product @requires(fields: "id")
  }

  type Query {
    product(id: ID!): Product
    productsByCategory(category: String!, limit: Int = 10): [Product!]!
    searchProducts(query: String!, filters: ProductFilters): ProductSearchResult!
  }

  input ProductFilters {
    minPrice: Float
    maxPrice: Float
    inStock: Boolean
    categories: [String!]
  }

  type ProductSearchResult {
    products: [Product!]!
    total: Int!
    pageInfo: PageInfo!
  }

  type PageInfo {
    hasNextPage: Boolean!
    endCursor: String
  }
`;

// 2. Implement resolvers with data loaders for N+1 prevention
const resolvers = {
  Product: {
    // Reference resolver for federated entities
    __resolveReference: async (reference, { dataSources }) => {
      return dataSources.productAPI.getProductById(reference.id);
    },
    
    // Resolver for reviews with batch loading
    reviews: async (product, _, { dataSources }) => {
      return dataSources.reviewAPI.getReviewsByProductId(product.id);
    },
    
    // Field-level resolver for computed fields
    inventory: async (product, _, { dataSources, cache }) => {
      const cacheKey = `inventory:${product.id}`;
      const cached = await cache.get(cacheKey);
      
      if (cached) return JSON.parse(cached);
      
      const inventory = await dataSources.inventoryAPI.getInventory(product.id);
      await cache.set(cacheKey, JSON.stringify(inventory), { ttl: 300 }); // 5 min cache
      return inventory;
    }
  },
  
  Query: {
    product: async (_, { id }, { dataSources, requestId }) => {
      console.log(`[${requestId}] Fetching product ${id}`);
      return dataSources.productAPI.getProductById(id);
    },
    
    productsByCategory: async (_, { category, limit }, { dataSources }) => {
      // Implement cursor-based pagination for scalability
      return dataSources.productAPI.getProductsByCategory(category, limit);
    },
    
    searchProducts: async (_, { query, filters }, { dataSources }) => {
      // Implement search with Elasticsearch/OpenSearch integration
      return dataSources.searchAPI.searchProducts(query, filters);
    }
  }
};

// 3. Data source implementation with Redis caching
class ProductAPI {
  private redis;
  private db;
  
  constructor(redisClient, dbConnection) {
    this.redis = redisClient;
    this.db = dbConnection;
  }
  
  async getProductById(id: string) {
    const cacheKey = `product:${id}`;
    
    // Check Redis cache first
    const cached = await this.redis.get(cacheKey);
    if (cached) {
      return JSON.parse(cached);
    }
    
    // Cache miss - query database
    const product = await this.db.query(
      `SELECT p.*, 
              json_build_object('amount', p.price_amount, 
                               'currency', p.price_currency) as price
       FROM products p 
       WHERE p.id = $1 AND p.status = 'active'`,
      [id]
    );
    
    if (product.rows.length === 0) return null;
    
    // Cache with adaptive TTL based on product popularity
    const ttl = await this.calculateAdaptiveTTL(id);
    await this.redis.setex(cacheKey, ttl, JSON.stringify(product.rows[0]));
    
    return product.rows[0];
  }
  
  private async calculateAdaptiveTTL(productId: string): Promise {
    // More popular products get shorter TTL for freshness
    const views = await this.redis.get(`views:${productId}`);
    const baseTTL = 300; // 5 minutes
    
    if (!views) return baseTTL;
    
    const viewCount = parseInt(views);
    if (viewCount > 1000) return 60; // 1 minute for popular items
    if (viewCount > 100) return 120; // 2 minutes
    return baseTTL;
  }
}

// 4. Build and start the server
const schema = buildSubgraphSchema({ typeDefs, resolvers });
const server = new ApolloServer({
  schema,
  plugins: [
    // Apollo Studio reporting
    ApolloServerPluginLandingPageLocalDefault({ embed: true }),
    // Query complexity analysis
    {
      async requestDidStart() {
        return {
          async didResolveOperation(context) {
            const complexity = calculateQueryComplexity(
              context.request.query,
              context.request.variables
            );
            if (complexity > 1000) {
              throw new GraphQLError('Query too complex');
            }
          }
        };
      }
    }
  ]
});

// Start server
const { url } = await startStandaloneServer(server, {
  listen: { port: 4001 },
  context: async ({ req }) => ({
    dataSources: {
      productAPI: new ProductAPI(redisClient, db),
      reviewAPI: new ReviewAPI(),
      inventoryAPI: new InventoryAPI(),
      searchAPI: new SearchAPI()
    },
    cache: redisClient,
    requestId: req.headers['x-request-id']
  })
});

console.log(`🚀 Product subgraph ready at ${url}`);

  

// advanced-caching.ts - Smart GraphQL caching with invalidation
import { parse, print, visit } from 'graphql';
import Redis from 'ioredis';
import { createHash } from 'crypto';

class GraphQLSmartCache {
  private redis: Redis;
  private cacheHits = 0;
  private cacheMisses = 0;
  
  constructor(redisUrl: string) {
    this.redis = new Redis(redisUrl);
  }
  
  // Generate cache key from query and variables
  private generateCacheKey(
    query: string, 
    variables: Record,
    userId?: string
  ): string {
    const ast = parse(query);
    
    // Normalize query (remove whitespace, sort fields)
    const normalizedQuery = this.normalizeQuery(ast);
    
    // Create hash of query + variables + user context
    const hashInput = JSON.stringify({
      query: normalizedQuery,
      variables: this.normalizeVariables(variables),
      user: userId || 'anonymous'
    });
    
    return `gql:${createHash('sha256').update(hashInput).digest('hex')}`;
  }
  
  // Cache GraphQL response with field-level invalidation tags
  async cacheResponse(
    query: string,
    variables: Record,
    response: any,
    options: {
      ttl: number;
      invalidationTags: string[];
      userId?: string;
    }
  ): Promise {
    const cacheKey = this.generateCacheKey(query, variables, options.userId);
    const cacheValue = JSON.stringify({
      data: response,
      timestamp: Date.now(),
      tags: options.invalidationTags
    });
    
    // Store main response
    await this.redis.setex(cacheKey, options.ttl, cacheValue);
    
    // Store reverse index for tag-based invalidation
    for (const tag of options.invalidationTags) {
      await this.redis.sadd(`tag:${tag}`, cacheKey);
    }
    
    // Store query pattern for pattern-based invalidation
    const queryPattern = this.extractQueryPattern(query);
    await this.redis.sadd(`pattern:${queryPattern}`, cacheKey);
  }
  
  // Retrieve cached response
  async getCachedResponse(
    query: string,
    variables: Record,
    userId?: string
  ): Promise {
    const cacheKey = this.generateCacheKey(query, variables, userId);
    const cached = await this.redis.get(cacheKey);
    
    if (cached) {
      this.cacheHits++;
      const parsed = JSON.parse(cached);
      
      // Check if cache is stale based on tags
      const isStale = await this.isCacheStale(parsed.tags);
      if (isStale) {
        await this.redis.del(cacheKey);
        this.cacheMisses++;
        return null;
      }
      
      return parsed.data;
    }
    
    this.cacheMisses++;
    return null;
  }
  
  // Invalidate cache by tags (e.g., when product data updates)
  async invalidateByTags(tags: string[]): Promise {
    for (const tag of tags) {
      const cacheKeys = await this.redis.smembers(`tag:${tag}`);
      
      if (cacheKeys.length > 0) {
        // Delete all cached entries with this tag
        await this.redis.del(...cacheKeys);
        await this.redis.del(`tag:${tag}`);
        
        console.log(`Invalidated ${cacheKeys.length} entries for tag: ${tag}`);
      }
    }
  }
  
  // Partial cache invalidation based on query patterns
  async invalidateByPattern(pattern: string): Promise {
    const cacheKeys = await this.redis.smembers(`pattern:${pattern}`);
    
    if (cacheKeys.length > 0) {
      // Invalidate matching queries
      await this.redis.del(...cacheKeys);
      await this.redis.del(`pattern:${pattern}`);
    }
  }
  
  // Extract invalidation tags from GraphQL query
  extractInvalidationTags(query: string): string[] {
    const ast = parse(query);
    const tags: string[] = [];
    
    visit(ast, {
      Field(node) {
        // Map fields to entity types for tagging
        const fieldToTagMap: Record = {
          'product': ['product'],
          'products': ['product:list'],
          'user': ['user'],
          'order': ['order', 'user:${userId}:orders']
        };
        
        if (fieldToTagMap[node.name.value]) {
          tags.push(...fieldToTagMap[node.name.value]);
        }
      }
    });
    
    return [...new Set(tags)]; // Remove duplicates
  }
  
  // Adaptive TTL based on query characteristics
  calculateAdaptiveTTL(query: string, userId?: string): number {
    const ast = parse(query);
    let maxTTL = 300; // Default 5 minutes
    
    // Adjust TTL based on query type
    visit(ast, {
      Field(node) {
        const fieldTTLs: Record = {
          'product': 60,           // Products update frequently
          'inventory': 30,         // Inventory changes often
          'userProfile': 86400,    // User profiles change rarely
          'catalog': 3600,         // Catalog changes daily
          'reviews': 1800          // Reviews update every 30 min
        };
        
        if (fieldTTLs[node.name.value]) {
          maxTTL = Math.min(maxTTL, fieldTTLs[node.name.value]);
        }
      }
    });
    
    // Authenticated users get fresher data
    if (userId) {
      maxTTL = Math.min(maxTTL, 120);
    }
    
    return maxTTL;
  }
  
  // Get cache statistics
  getStats() {
    const total = this.cacheHits + this.cacheMisses;
    const hitRate = total > 0 ? (this.cacheHits / total) * 100 : 0;
    
    return {
      hits: this.cacheHits,
      misses: this.cacheMisses,
      hitRate: `${hitRate.toFixed(2)}%`,
      total
    };
  }
}

// Usage example in a GraphQL resolver
const smartCache = new GraphQLSmartCache(process.env.REDIS_URL);

const productResolvers = {
  Query: {
    product: async (_, { id }, context) => {
      const query = context.queryString; // Original GraphQL query
      const userId = context.user?.id;
      
      // Try cache first
      const cached = await smartCache.getCachedResponse(query, { id }, userId);
      if (cached) {
        context.metrics.cacheHit();
        return cached;
      }
      
      // Cache miss - fetch from database
      const product = await db.products.findUnique({ where: { id } });
      
      // Cache the response
      const invalidationTags = smartCache.extractInvalidationTags(query);
      const ttl = smartCache.calculateAdaptiveTTL(query, userId);
      
      await smartCache.cacheResponse(
        query,
        { id },
        product,
        {
          ttl,
          invalidationTags,
          userId
        }
      );
      
      context.metrics.cacheMiss();
      return product;
    }
  },
  
  Mutation: {
    updateProduct: async (_, { id, input }, context) => {
      // Update product in database
      const updated = await db.products.update({
        where: { id },
        data: input
      });
      
      // Invalidate all caches related to this product
      await smartCache.invalidateByTags(['product', `product:${id}`]);
      
      return updated;
    }
  }
};

  

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())