Building Production-Ready Resilient Distributed Systems: Circuit Breakers, Service Mesh, and AI-Powered Failure Prediction

<!-- pom.xml -->
<properties>
    <resilience4j.version>2.1.0</resilience4j.version>
    <spring-boot.version>3.2.0</spring-boot.version>
</properties>

<dependencies>
    <!-- Spring Boot -->
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-web</artifactId>
    </dependency>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-actuator</artifactId>
    </dependency>

    <!-- Resilience4j Circuit Breaker -->
    <dependency>
        <groupId>io.github.resilience4j</groupId>
        <artifactId>resilience4j-spring-boot3</artifactId>
        <version>${resilience4j.version}</version>
    </dependency>
    <dependency>
        <groupId>io.github.resilience4j</groupId>
        <artifactId>resilience4j-circuitbreaker</artifactId>
        <version>${resilience4j.version}</version>
    </dependency>
    <dependency>
        <groupId>io.github.resilience4j</groupId>
        <artifactId>resilience4j-retry</artifactId>
        <version>${resilience4j.version}</version>
    </dependency>
    <dependency>
        <groupId>io.github.resilience4j</groupId>
        <artifactId>resilience4j-bulkhead</artifactId>
        <version>${resilience4j.version}</version>
    </dependency>
    <dependency>
        <groupId>io.github.resilience4j</groupId>
        <artifactId>resilience4j-timelimiter</artifactId>
        <version>${resilience4j.version}</version>
    </dependency>

    <!-- Micrometer for metrics -->
    <dependency>
        <groupId>io.micrometer</groupId>
        <artifactId>micrometer-registry-prometheus</artifactId>
    </dependency>

    <!-- OpenTelemetry for distributed tracing -->
    <dependency>
        <groupId>io.opentelemetry</groupId>
        <artifactId>opentelemetry-api</artifactId>
        <version>1.25.0</version>
    </dependency>
    <dependency>
        <groupId>io.opentelemetry</groupId>
        <artifactId>opentelemetry-sdk</artifactId>
        <version>1.25.0</version>
    </dependency>
</dependencies>

# application.yml
resilience4j:
  circuitbreaker:
    configs:
      default:
        registerHealthIndicator: true
        slidingWindowSize: 10
        minimumNumberOfCalls: 5
        permittedNumberOfCallsInHalfOpenState: 3
        automaticTransitionFromOpenToHalfOpenEnabled: true
        waitDurationInOpenState: 5s
        failureRateThreshold: 50
        eventConsumerBufferSize: 10
        recordExceptions:
          - org.springframework.web.client.HttpServerErrorException
          - java.util.concurrent.TimeoutException
          - java.io.IOException
        ignoreExceptions:
          - com.example.exceptions.BusinessException

    instances:
      paymentService:
        baseConfig: default
        waitDurationInOpenState: 10s
        failureRateThreshold: 60
      inventoryService:
        baseConfig: default
        failureRateThreshold: 40

  retry:
    configs:
      default:
        maxAttempts: 3
        waitDuration: 1000ms
        retryExceptions:
          - org.springframework.web.client.ResourceAccessException
          - java.net.SocketTimeoutException
        ignoreExceptions:
          - com.example.exceptions.BusinessException

    instances:
      paymentService:
        baseConfig: default
        maxAttempts: 5
        waitDuration: 2000ms
        enableExponentialBackoff: true
        exponentialBackoffMultiplier: 2

  bulkhead:
    configs:
      default:
        maxConcurrentCalls: 10
        maxWaitDuration: 1000ms

    instances:
      databaseOperations:
        maxConcurrentCalls: 25
        maxWaitDuration: 500ms

  timelimiter:
    configs:
      default:
        timeoutDuration: 5s
        cancelRunningFuture: true

    instances:
      paymentService:
        timeoutDuration: 10s

management:
  health:
    circuitbreakers:
      enabled: true
  endpoints:
    web:
      exposure:
        include: health,info,metrics,prometheus
  metrics:
    export:
      prometheus:
        enabled: true
    tags:
      application: ${spring.application.name}

package com.example.resilience.client;

import com.example.model.Payment;
import com.example.model.PaymentResponse;
import io.github.resilience4j.bulkhead.annotation.Bulkhead;
import io.github.resilience4j.circuitbreaker.annotation.CircuitBreaker;
import io.github.resilience4j.retry.annotation.Retry;
import io.github.resilience4j.timelimiter.annotation.TimeLimiter;
import io.micrometer.core.instrument.Counter;
import io.micrometer.core.instrument.MeterRegistry;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;
import org.springframework.web.client.RestTemplate;

import java.util.concurrent.CompletableFuture;

/**
 * Resilient payment service client with circuit breaker, retry, bulkhead, and timeout.
 * Demonstrates production-ready resilience patterns.
 */
@Service
public class PaymentServiceClient {

    private static final Logger logger = LoggerFactory.getLogger(PaymentServiceClient.class);

    private final RestTemplate restTemplate;
    private final String paymentServiceUrl;
    private final Counter successCounter;
    private final Counter failureCounter;
    private final Counter fallbackCounter;

    public PaymentServiceClient(
        RestTemplate restTemplate,
        @Value("${payment.service.url}") String paymentServiceUrl,
        MeterRegistry meterRegistry
    ) {
        this.restTemplate = restTemplate;
        this.paymentServiceUrl = paymentServiceUrl;

        // Initialize metrics
        this.successCounter = Counter.builder("payment.service.calls.success")
            .description("Successful payment service calls")
            .register(meterRegistry);
        this.failureCounter = Counter.builder("payment.service.calls.failure")
            .description("Failed payment service calls")
            .register(meterRegistry);
        this.fallbackCounter = Counter.builder("payment.service.calls.fallback")
            .description("Payment service calls that used fallback")
            .register(meterRegistry);
    }

    /**
     * Process payment with full resilience stack:
     * - Circuit Breaker: Fail fast if service is down
     * - Retry: Retry transient failures with exponential backoff
     * - Bulkhead: Limit concurrent calls to prevent resource exhaustion
     * - TimeLimiter: Enforce timeout on async operations
     *
     * @param payment Payment details
     * @return CompletableFuture with payment response
     */
    @CircuitBreaker(name = "paymentService", fallbackMethod = "processPaymentFallback")
    @Retry(name = "paymentService")
    @Bulkhead(name = "paymentService")
    @TimeLimiter(name = "paymentService")
    public CompletableFuture<PaymentResponse> processPayment(Payment payment) {
        return CompletableFuture.supplyAsync(() -> {
            try {
                logger.info("Processing payment: amount={}, currency={}",
                    payment.getAmount(), payment.getCurrency());

                // Make HTTP call to payment service
                PaymentResponse response = restTemplate.postForObject(
                    paymentServiceUrl + "/process",
                    payment,
                    PaymentResponse.class
                );

                if (response != null && response.isSuccess()) {
                    successCounter.increment();
                    logger.info("Payment processed successfully: transactionId={}",
                        response.getTransactionId());
                    return response;
                } else {
                    failureCounter.increment();
                    throw new PaymentProcessingException("Payment failed: " +
                        (response != null ? response.getErrorMessage() : "Unknown error"));
                }

            } catch (Exception e) {
                failureCounter.increment();
                logger.error("Error processing payment", e);
                throw new PaymentProcessingException("Payment processing failed", e);
            }
        });
    }

    /**
     * Fallback method when payment service is unavailable.
     * Returns a degraded response instead of failing completely.
     *
     * @param payment Payment details
     * @param throwable Exception that triggered fallback
     * @return CompletableFuture with fallback payment response
     */
    private CompletableFuture<PaymentResponse> processPaymentFallback(
        Payment payment,
        Throwable throwable
    ) {
        fallbackCounter.increment();
        logger.warn("Payment service fallback triggered for amount={}: {}",
            payment.getAmount(), throwable.getMessage());

        // Return degraded response
        PaymentResponse fallbackResponse = new PaymentResponse();
        fallbackResponse.setSuccess(false);
        fallbackResponse.setTransactionId("FALLBACK-" + System.currentTimeMillis());
        fallbackResponse.setStatus("PENDING");
        fallbackResponse.setErrorMessage("Payment service temporarily unavailable. " +
            "Your payment will be processed when service is restored.");

        // Queue payment for later processing
        queuePaymentForRetry(payment);

        return CompletableFuture.completedFuture(fallbackResponse);
    }

    /**
     * Queue failed payments for retry when service recovers.
     * In production, this would write to a message queue (Kafka, RabbitMQ, SQS).
     */
    private void queuePaymentForRetry(Payment payment) {
        logger.info("Queueing payment for retry: amount={}", payment.getAmount());
        // Implementation: Send to message queue
        // Example with Spring AMQP:
        // rabbitTemplate.convertAndSend("payment.retry.queue", payment);
    }

    /**
     * Check payment status with resilience patterns.
     * Uses different configuration (fewer retries, shorter timeout).
     */
    @CircuitBreaker(name = "paymentService")
    @Retry(name = "paymentService", fallbackMethod = "getPaymentStatusFallback")
    public PaymentResponse getPaymentStatus(String transactionId) {
        logger.info("Checking payment status: transactionId={}", transactionId);

        try {
            return restTemplate.getForObject(
                paymentServiceUrl + "/status/" + transactionId,
                PaymentResponse.class
            );
        } catch (Exception e) {
            logger.error("Error checking payment status", e);
            throw new PaymentProcessingException("Failed to check payment status", e);
        }
    }

    private PaymentResponse getPaymentStatusFallback(String transactionId, Throwable throwable) {
        logger.warn("Payment status check fallback: transactionId={}", transactionId);

        PaymentResponse response = new PaymentResponse();
        response.setTransactionId(transactionId);
        response.setStatus("UNKNOWN");
        response.setErrorMessage("Unable to retrieve payment status. Please try again later.");
        return response;
    }
}

/**
 * Custom exception for payment processing errors.
 */
class PaymentProcessingException extends RuntimeException {
    public PaymentProcessingException(String message) {
        super(message);
    }

    public PaymentProcessingException(String message, Throwable cause) {
        super(message, cause);
    }
}

package com.example.resilience.health;

import org.springframework.boot.actuate.health.Health;
import org.springframework.boot.actuate.health.HealthIndicator;
import org.springframework.stereotype.Component;
import org.springframework.web.client.RestTemplate;

/**
 * Custom health check for payment service dependency.
 * Kubernetes uses this to determine pod readiness.
 */
@Component("paymentServiceHealth")
public class PaymentServiceHealthIndicator implements HealthIndicator {

    private final RestTemplate restTemplate;
    private final String paymentServiceHealthUrl;

    public PaymentServiceHealthIndicator(
        RestTemplate restTemplate,
        @Value("${payment.service.url}") String paymentServiceUrl
    ) {
        this.restTemplate = restTemplate;
        this.paymentServiceHealthUrl = paymentServiceUrl + "/actuator/health";
    }

    @Override
    public Health health() {
        try {
            // Check if payment service is reachable
            restTemplate.getForObject(paymentServiceHealthUrl, String.class);

            return Health.up()
                .withDetail("paymentService", "Available")
                .build();

        } catch (Exception e) {
            return Health.down()
                .withDetail("paymentService", "Unavailable")
                .withDetail("error", e.getMessage())
                .build();
        }
    }
}

# deployment.yaml
apiVersion: apps/v1
kind: Deployment
metadata:
  name: order-service
spec:
  replicas: 3
  selector:
    matchLabels:
      app: order-service
  template:
    metadata:
      labels:
        app: order-service
    spec:
      containers:
      - name: order-service
        image: order-service:1.0.0
        ports:
        - containerPort: 8080

        # Liveness probe: Restart container if unhealthy
        livenessProbe:
          httpGet:
            path: /actuator/health/liveness
            port: 8080
          initialDelaySeconds: 30
          periodSeconds: 10
          timeoutSeconds: 5
          failureThreshold: 3

        # Readiness probe: Remove from service if not ready
        readinessProbe:
          httpGet:
            path: /actuator/health/readiness
            port: 8080
          initialDelaySeconds: 20
          periodSeconds: 5
          timeoutSeconds: 3
          failureThreshold: 2

        # Startup probe: Give slow-starting apps time to initialize
        startupProbe:
          httpGet:
            path: /actuator/health/liveness
            port: 8080
          initialDelaySeconds: 0
          periodSeconds: 10
          timeoutSeconds: 5
          failureThreshold: 30

        resources:
          requests:
            memory: "512Mi"
            cpu: "500m"
          limits:
            memory: "1Gi"
            cpu: "1000m"

        env:
        - name: PAYMENT_SERVICE_URL
          value: "http://payment-service:8080"
        - name: SPRING_PROFILES_ACTIVE
          value: "production"

# payment-service-virtualservice.yaml
apiVersion: networking.istio.io/v1beta1
kind: VirtualService
metadata:
  name: payment-service
spec:
  hosts:
  - payment-service
  http:
  - match:
    - headers:
        x-api-version:
          exact: "v2"
    route:
    - destination:
        host: payment-service
        subset: v2
      weight: 100
    retries:
      attempts: 3
      perTryTimeout: 2s
      retryOn: 5xx,reset,connect-failure,refused-stream
    timeout: 10s

  - route:
    - destination:
        host: payment-service
        subset: v1
      weight: 90
    - destination:
        host: payment-service
        subset: v2
      weight: 10  # Canary deployment: 10% traffic to v2

    retries:
      attempts: 2
      perTryTimeout: 3s
    timeout: 15s

    # Circuit breaker configuration
    fault:
      delay:
        percentage:
          value: 0.1  # 0.1% of requests get 5s delay (chaos testing)
        fixedDelay: 5s

# payment-service-destinationrule.yaml
apiVersion: networking.istio.io/v1beta1
kind: DestinationRule
metadata:
  name: payment-service
spec:
  host: payment-service
  trafficPolicy:
    connectionPool:
      tcp:
        maxConnections: 100
      http:
        http1MaxPendingRequests: 50
        http2MaxRequests: 100
        maxRequestsPerConnection: 2

    outlierDetection:
      consecutiveErrors: 5
      interval: 30s
      baseEjectionTime: 30s
      maxEjectionPercent: 50
      minHealthPercent: 50

  subsets:
  - name: v1
    labels:
      version: v1
  - name: v2
    labels:
      version: v2
    trafficPolicy:
      connectionPool:
        http:
          http1MaxPendingRequests: 100  # v2 can handle more load

package com.example.resilience.tracing;

import io.opentelemetry.api.OpenTelemetry;
import io.opentelemetry.api.trace.Tracer;
import io.opentelemetry.api.trace.Span;
import io.opentelemetry.api.trace.StatusCode;
import io.opentelemetry.context.Scope;
import io.opentelemetry.sdk.OpenTelemetrySdk;
import io.opentelemetry.sdk.trace.SdkTracerProvider;
import io.opentelemetry.sdk.trace.export.BatchSpanProcessor;
import io.opentelemetry.exporter.otlp.trace.OtlpGrpcSpanExporter;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

/**
 * OpenTelemetry configuration for distributed tracing.
 * Sends traces to Jaeger, Zipkin, or any OTLP-compatible backend.
 */
@Configuration
public class TracingConfiguration {

    @Bean
    public OpenTelemetry openTelemetry() {
        // Configure OTLP exporter (sends to Jaeger/Tempo/etc.)
        OtlpGrpcSpanExporter spanExporter = OtlpGrpcSpanExporter.builder()
            .setEndpoint("http://jaeger-collector:4317")
            .build();

        // Configure tracer provider
        SdkTracerProvider tracerProvider = SdkTracerProvider.builder()
            .addSpanProcessor(BatchSpanProcessor.builder(spanExporter).build())
            .build();

        return OpenTelemetrySdk.builder()
            .setTracerProvider(tracerProvider)
            .buildAndRegisterGlobal();
    }

    @Bean
    public Tracer tracer(OpenTelemetry openTelemetry) {
        return openTelemetry.getTracer("order-service", "1.0.0");
    }
}

package com.example.resilience.service;

import io.opentelemetry.api.trace.Span;
import io.opentelemetry.api.trace.Tracer;
import io.opentelemetry.api.trace.StatusCode;
import io.opentelemetry.context.Scope;
import org.springframework.stereotype.Service;

@Service
public class OrderService {

    private final Tracer tracer;
    private final PaymentServiceClient paymentClient;
    private final InventoryServiceClient inventoryClient;

    public OrderService(
        Tracer tracer,
        PaymentServiceClient paymentClient,
        InventoryServiceClient inventoryClient
    ) {
        this.tracer = tracer;
        this.paymentClient = paymentClient;
        this.inventoryClient = inventoryClient;
    }

    /**
     * Create order with distributed tracing.
     * Each downstream call creates child spans for end-to-end visibility.
     */
    public OrderResponse createOrder(OrderRequest request) {
        // Create root span for this operation
        Span span = tracer.spanBuilder("createOrder")
            .setAttribute("order.id", request.getOrderId())
            .setAttribute("order.amount", request.getTotalAmount())
            .startSpan();

        try (Scope scope = span.makeCurrent()) {
            // Step 1: Check inventory (child span created automatically)
            boolean inventoryAvailable = checkInventory(request);
            span.addEvent("Inventory checked", Map.of("available", inventoryAvailable));

            if (!inventoryAvailable) {
                span.setStatus(StatusCode.ERROR, "Insufficient inventory");
                throw new InsufficientInventoryException();
            }

            // Step 2: Process payment (child span)
            PaymentResponse paymentResponse = processPayment(request);
            span.addEvent("Payment processed", Map.of(
                "transactionId", paymentResponse.getTransactionId()
            ));

            // Step 3: Reserve inventory (child span)
            reserveInventory(request);
            span.addEvent("Inventory reserved");

            // Success
            span.setStatus(StatusCode.OK);
            return new OrderResponse(request.getOrderId(), "SUCCESS");

        } catch (Exception e) {
            // Record exception in trace
            span.recordException(e);
            span.setStatus(StatusCode.ERROR, e.getMessage());
            throw new OrderProcessingException("Order creation failed", e);

        } finally {
            span.end();
        }
    }

    private boolean checkInventory(OrderRequest request) {
        Span span = tracer.spanBuilder("checkInventory").startSpan();
        try (Scope scope = span.makeCurrent()) {
            return inventoryClient.checkAvailability(request.getItems());
        } finally {
            span.end();
        }
    }

    private PaymentResponse processPayment(OrderRequest request) {
        Span span = tracer.spanBuilder("processPayment").startSpan();
        try (Scope scope = span.makeCurrent()) {
            Payment payment = new Payment(request.getTotalAmount(), request.getCurrency());
            return paymentClient.processPayment(payment).join();
        } finally {
            span.end();
        }
    }

    private void reserveInventory(OrderRequest request) {
        Span span = tracer.spanBuilder("reserveInventory").startSpan();
        try (Scope scope = span.makeCurrent()) {
            inventoryClient.reserve(request.getOrderId(), request.getItems());
        } finally {
            span.end();
        }
    }
}

# chaos-network-latency.yaml
apiVersion: chaos-mesh.org/v1alpha1
kind: NetworkChaos
metadata:
  name: payment-service-latency
spec:
  action: delay
  mode: one
  selector:
    namespaces:
      - production
    labelSelectors:
      app: payment-service
  delay:
    latency: "500ms"
    correlation: "50"
    jitter: "100ms"
  duration: "5m"
  scheduler:
    cron: "@every 1h"  # Run every hour

# chaos-pod-kill.yaml
apiVersion: chaos-mesh.org/v1alpha1
kind: PodChaos
metadata:
  name: order-service-pod-kill
spec:
  action: pod-kill
  mode: fixed
  value: "1"  # Kill 1 pod
  selector:
    namespaces:
      - production
    labelSelectors:
      app: order-service
  duration: "30s"
  scheduler:
    cron: "0 */6 * * *"  # Every 6 hours

# chaos-http-abort.yaml
apiVersion: chaos-mesh.org/v1alpha1
kind: HTTPChaos
metadata:
  name: payment-service-http-abort
spec:
  mode: one
  selector:
    namespaces:
      - production
    labelSelectors:
      app: payment-service
  target: Request
  port: 8080
  method: POST
  path: /process
  abort: true
  duration: "2m"

# litmus-workflow.yaml
apiVersion: argoproj.io/v1alpha1
kind: Workflow
metadata:
  name: resilience-test-workflow
spec:
  entrypoint: resilience-tests
  templates:
  - name: resilience-tests
    steps:
    - - name: pod-delete-test
        template: pod-delete
    - - name: network-latency-test
        template: network-latency
    - - name: cpu-hog-test
        template: cpu-hog

  - name: pod-delete
    container:
      image: litmuschaos/litmus-checker:latest
      args:
      - -file=/experiments/pod-delete.yaml
      - -saveName=pod-delete-results

  - name: network-latency
    container:
      image: litmuschaos/litmus-checker:latest
      args:
      - -file=/experiments/network-latency.yaml
      - -saveName=network-latency-results

  - name: cpu-hog
    container:
      image: litmuschaos/litmus-checker:latest
      args:
      - -file=/experiments/cpu-hog.yaml
      - -saveName=cpu-hog-results

"""
AI-powered failure prediction using historical metrics.
Predicts service failures 10-30 minutes before they occur.
"""

import pandas as pd
import numpy as np
from sklearn.ensemble import RandomForestClassifier
from sklearn.model_selection import train_test_split, cross_val_score
from sklearn.metrics import classification_report, confusion_matrix
import tensorflow as tf
from tensorflow import keras
from prometheus_api_client import PrometheusConnect
import joblib
from datetime import datetime, timedelta
import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

class FailurePredictor:
    """
    Predicts service failures using historical metrics from Prometheus.

    Features used:
    - CPU utilization
    - Memory usage
    - Request latency (p50, p95, p99)
    - Error rate
    - Request rate
    - Circuit breaker state
    - Database connection pool utilization
    """

    def __init__(self, prometheus_url: str):
        self.prom = PrometheusConnect(url=prometheus_url, disable_ssl=True)
        self.model = None
        self.feature_columns = [
            'cpu_usage', 'memory_usage', 'request_latency_p50',
            'request_latency_p95', 'request_latency_p99', 'error_rate',
            'request_rate', 'circuit_breaker_open', 'db_pool_utilization'
        ]

    def fetch_metrics(
        self,
        service_name: str,
        start_time: datetime,
        end_time: datetime,
        step: str = "1m"
    ) -> pd.DataFrame:
        """
        Fetch metrics from Prometheus for model training/inference.

        Args:
            service_name: Service to fetch metrics for
            start_time: Start of time range
            end_time: End of time range
            step: Query resolution (1m, 5m, etc.)

        Returns:
            DataFrame with metrics and labels
        """
        queries = {
            'cpu_usage': f'rate(container_cpu_usage_seconds_total{{pod=~"{service_name}.*"}}[5m])',
            'memory_usage': f'container_memory_usage_bytes{{pod=~"{service_name}.*"}}',
            'request_latency_p50': f'histogram_quantile(0.50, rate(http_request_duration_seconds_bucket{{service="{service_name}"}}[5m]))',
            'request_latency_p95': f'histogram_quantile(0.95, rate(http_request_duration_seconds_bucket{{service="{service_name}"}}[5m]))',
            'request_latency_p99': f'histogram_quantile(0.99, rate(http_request_duration_seconds_bucket{{service="{service_name}"}}[5m]))',
            'error_rate': f'rate(http_requests_total{{service="{service_name}", status=~"5.."}}[5m])',
            'request_rate': f'rate(http_requests_total{{service="{service_name}"}}[5m])',
            'circuit_breaker_open': f'resilience4j_circuitbreaker_state{{name=~"{service_name}.*", state="open"}}',
            'db_pool_utilization': f'hikaricp_connections_active{{pool=~"{service_name}.*"}} / hikaricp_connections_max{{pool=~"{service_name}.*"}}'
        }

        # Fetch all metrics
        metrics_data = {}
        for metric_name, query in queries.items():
            try:
                result = self.prom.custom_query_range(
                    query=query,
                    start_time=start_time,
                    end_time=end_time,
                    step=step
                )

                if result:
                    # Convert to time series
                    values = [(float(v[1])) for v in result[0]['values']]
                    timestamps = [datetime.fromtimestamp(v[0]) for v in result[0]['values']]
                    metrics_data[metric_name] = pd.Series(values, index=timestamps)

            except Exception as e:
                logger.warning(f"Failed to fetch {metric_name}: {e}")
                metrics_data[metric_name] = pd.Series()

        # Combine into DataFrame
        df = pd.DataFrame(metrics_data)

        # Fill missing values with interpolation
        df = df.interpolate(method='linear', limit_direction='both')

        return df

    def label_failures(
        self,
        df: pd.DataFrame,
        failure_threshold: dict = None
    ) -> pd.DataFrame:
        """
        Label time periods as "failure" or "healthy" based on thresholds.

        Args:
            df: DataFrame with metrics
            failure_threshold: Dict of metric -> threshold for failure

        Returns:
            DataFrame with 'failure' column (1=failure imminent, 0=healthy)
        """
        if failure_threshold is None:
            failure_threshold = {
                'error_rate': 0.05,  # >5% error rate
                'request_latency_p99': 5.0,  # >5s p99 latency
                'cpu_usage': 0.9,  # >90% CPU
                'memory_usage': 0.85 * 1024**3,  # >85% of 1GB
                'circuit_breaker_open': 0.5  # Circuit breaker open
            }

        # Mark periods as failures if any threshold exceeded
        df['failure'] = 0
        for metric, threshold in failure_threshold.items():
            if metric in df.columns:
                df.loc[df[metric] > threshold, 'failure'] = 1

        # Shift labels back by 10 minutes (predict 10 min before failure)
        df['failure'] = df['failure'].shift(-10)

        # Remove rows with NaN labels
        df = df.dropna(subset=['failure'])

        return df

    def train_model(
        self,
        service_name: str,
        start_time: datetime,
        end_time: datetime
    ):
        """
        Train failure prediction model on historical data.

        Args:
            service_name: Service to train model for
            start_time: Training data start
            end_time: Training data end
        """
        logger.info(f"Training failure prediction model for {service_name}")

        # Fetch training data
        df = self.fetch_metrics(service_name, start_time, end_time)
        df = self.label_failures(df)

        logger.info(f"Training data: {len(df)} samples, {df['failure'].sum()} failures")

        # Prepare features and labels
        X = df[self.feature_columns].fillna(0)
        y = df['failure']

        # Train/test split
        X_train, X_test, y_train, y_test = train_test_split(
            X, y, test_size=0.2, random_state=42, stratify=y
        )

        # Train Random Forest classifier
        self.model = RandomForestClassifier(
            n_estimators=100,
            max_depth=10,
            min_samples_split=10,
            class_weight='balanced',  # Handle class imbalance
            random_state=42
        )

        self.model.fit(X_train, y_train)

        # Evaluate
        y_pred = self.model.predict(X_test)

        logger.info("Model Performance:")
        logger.info(f"\n{classification_report(y_test, y_pred)}")
        logger.info(f"\nConfusion Matrix:\n{confusion_matrix(y_test, y_pred)}")

        # Cross-validation
        cv_scores = cross_val_score(self.model, X, y, cv=5, scoring='f1')
        logger.info(f"Cross-validation F1 scores: {cv_scores}")
        logger.info(f"Mean F1: {cv_scores.mean():.3f} (+/- {cv_scores.std():.3f})")

        # Feature importance
        feature_importance = pd.DataFrame({
            'feature': self.feature_columns,
            'importance': self.model.feature_importances_
        }).sort_values('importance', ascending=False)

        logger.info(f"\nFeature Importance:\n{feature_importance}")

    def predict_failure(self, service_name: str) -> dict:
        """
        Predict if failure is imminent for a service.

        Args:
            service_name: Service to check

        Returns:
            Dict with prediction and probability
        """
        if self.model is None:
            raise ValueError("Model not trained. Call train_model() first.")

        # Fetch current metrics (last 5 minutes)
        end_time = datetime.now()
        start_time = end_time - timedelta(minutes=5)

        df = self.fetch_metrics(service_name, start_time, end_time)

        if df.empty:
            logger.warning(f"No metrics available for {service_name}")
            return {'failure_predicted': False, 'probability': 0.0}

        # Use latest metrics
        X = df[self.feature_columns].iloc[-1:].fillna(0)

        # Predict
        prediction = self.model.predict(X)[0]
        probability = self.model.predict_proba(X)[0][1]  # Probability of failure

        result = {
            'service': service_name,
            'failure_predicted': bool(prediction),
            'failure_probability': float(probability),
            'timestamp': datetime.now().isoformat(),
            'metrics': X.iloc[0].to_dict()
        }

        if prediction:
            logger.warning(
                f"FAILURE PREDICTED for {service_name}! "
                f"Probability: {probability:.2%}"
            )

        return result

    def save_model(self, filepath: str):
        """Save trained model to disk."""
        joblib.dump(self.model, filepath)
        logger.info(f"Model saved to {filepath}")

    def load_model(self, filepath: str):
        """Load trained model from disk."""
        self.model = joblib.load(filepath)
        logger.info(f"Model loaded from {filepath}")

# Example usage
if __name__ == "__main__":
    # Initialize predictor
    predictor = FailurePredictor(prometheus_url="http://prometheus:9090")

    # Train model on last 30 days of data
    end_time = datetime.now()
    start_time = end_time - timedelta(days=30)

    predictor.train_model(
        service_name="payment-service",
        start_time=start_time,
        end_time=end_time
    )

    # Save model
    predictor.save_model("payment_service_failure_model.pkl")

    # Make prediction
    prediction = predictor.predict_failure("payment-service")
    print(f"\nPrediction: {prediction}")

    # If failure predicted, take preventive action
    if prediction['failure_predicted']:
        print("\nTaking preventive action:")
        print("- Scaling up replicas")
        print("- Draining traffic to healthy instances")
        print("- Alerting SRE team")

"""
Automated remediation based on failure predictions.
Takes action to prevent predicted failures.
"""

from kubernetes import client, config
import logging

logger = logging.getLogger(__name__)

class AutomatedRemediation:
    """
    Automated actions to prevent predicted failures.
    """

    def __init__(self):
        config.load_kube_config()
        self.apps_v1 = client.AppsV1Api()
        self.core_v1 = client.CoreV1Api()

    def scale_service(self, service_name: str, namespace: str, replicas: int):
        """
        Scale service to specified number of replicas.

        Args:
            service_name: Deployment name
            namespace: Kubernetes namespace
            replicas: Target replica count
        """
        try:
            logger.info(f"Scaling {service_name} to {replicas} replicas")

            # Get current deployment
            deployment = self.apps_v1.read_namespaced_deployment(
                name=service_name,
                namespace=namespace
            )

            # Update replica count
            deployment.spec.replicas = replicas

            # Apply update
            self.apps_v1.patch_namespaced_deployment(
                name=service_name,
                namespace=namespace,
                body=deployment
            )

            logger.info(f"Successfully scaled {service_name} to {replicas}")

        except Exception as e:
            logger.error(f"Failed to scale {service_name}: {e}")

    def restart_pods(self, service_name: str, namespace: str):
        """
        Restart pods by deleting them (deployment controller recreates them).

        Args:
            service_name: Service name (label selector)
            namespace: Kubernetes namespace
        """
        try:
            logger.info(f"Restarting pods for {service_name}")

            # List pods with label selector
            pods = self.core_v1.list_namespaced_pod(
                namespace=namespace,
                label_selector=f"app={service_name}"
            )

            # Delete each pod
            for pod in pods.items:
                logger.info(f"Deleting pod {pod.metadata.name}")
                self.core_v1.delete_namespaced_pod(
                    name=pod.metadata.name,
                    namespace=namespace
                )

            logger.info(f"Restarted {len(pods.items)} pods")

        except Exception as e:
            logger.error(f"Failed to restart pods for {service_name}: {e}")

    def drain_traffic(self, service_name: str, namespace: str):
        """
        Drain traffic from unhealthy instances by updating service selector.

        Args:
            service_name: Service name
            namespace: Kubernetes namespace
        """
        try:
            logger.info(f"Draining traffic from {service_name}")

            # This is a simplified example
            # In production, use Istio VirtualService or similar

            # Update service to only route to healthy pods
            service = self.core_v1.read_namespaced_service(
                name=service_name,
                namespace=namespace
            )

            # Add label selector for healthy pods
            service.spec.selector['health'] = 'healthy'

            self.core_v1.patch_namespaced_service(
                name=service_name,
                namespace=namespace,
                body=service
            )

            logger.info(f"Traffic drained from {service_name}")

        except Exception as e:
            logger.error(f"Failed to drain traffic for {service_name}: {e}")

    def handle_predicted_failure(self, prediction: dict):
        """
        Take automated action based on failure prediction.

        Args:
            prediction: Prediction dict from FailurePredictor
        """
        service_name = prediction['service']
        probability = prediction['failure_probability']

        logger.info(
            f"Handling predicted failure for {service_name} "
            f"(probability: {probability:.2%})"
        )

        # Action thresholds
        if probability > 0.8:
            # High probability: Aggressive action
            logger.warning("HIGH PROBABILITY FAILURE - Taking aggressive action")
            self.scale_service(service_name, "production", replicas=10)
            self.restart_pods(service_name, "production")

        elif probability > 0.5:
            # Medium probability: Moderate action
            logger.warning("MEDIUM PROBABILITY FAILURE - Taking moderate action")
            self.scale_service(service_name, "production", replicas=6)

        else:
            # Low probability: Monitor only
            logger.info("LOW PROBABILITY FAILURE - Monitoring")

# Integration example
if __name__ == "__main__":
    predictor = FailurePredictor(prometheus_url="http://prometheus:9090")
    predictor.load_model("payment_service_failure_model.pkl")

    remediation = AutomatedRemediation()

    # Continuous monitoring loop
    import time
    while True:
        try:
            # Predict failure
            prediction = predictor.predict_failure("payment-service")

            # Take action if needed
            if prediction['failure_predicted']:
                remediation.handle_predicted_failure(prediction)

            # Check every minute
            time.sleep(60)

        except KeyboardInterrupt:
            logger.info("Stopping monitoring")
            break
        except Exception as e:
            logger.error(f"Error in monitoring loop: {e}")
            time.sleep(60)

// Symptom: Circuit breaker opens and never closes despite service recovery

// Solution 1: Check minimumNumberOfCalls configuration
resilience4j.circuitbreaker.instances.paymentService.minimumNumberOfCalls=5

// If too high, circuit breaker may not collect enough samples in half-open state
// Recommendation: Set to 3-5 for most services

// Solution 2: Verify health check endpoint is responding
curl http://payment-service:8080/actuator/health

// Solution 3: Check if exceptions are being recorded
resilience4j.circuitbreaker.instances.paymentService.recordExceptions=\
  java.net.ConnectException,\
  java.net.SocketTimeoutException

// Solution 4: Manually transition circuit breaker for testing
@Autowired
private CircuitBreakerRegistry circuitBreakerRegistry;

public void forceCircuitBreakerClosed() {
    CircuitBreaker cb = circuitBreakerRegistry.circuitBreaker("paymentService");
    cb.transitionToClosedState();
}

// Symptom: All retry attempts fail, overwhelming downstream service

// Solution 1: Implement exponential backoff with jitter
resilience4j.retry.instances.paymentService.enableExponentialBackoff=true
resilience4j.retry.instances.paymentService.exponentialBackoffMultiplier=2
resilience4j.retry.instances.paymentService.enableRandomizedWait=true

// Solution 2: Check if retrying appropriate exceptions
// Don't retry business logic errors (4xx) - only transient failures (5xx, timeouts)
@Retry(name = "paymentService", fallbackMethod = "paymentFallback")
public PaymentResponse processPayment(Payment payment) {
    try {
        return paymentClient.call(payment);
    } catch (BusinessException e) {
        // Don't retry business exceptions
        throw e;
    } catch (Exception e) {
        // Retry transient failures
        throw e;
    }
}

// Solution 3: Add circuit breaker before retry
// This prevents retry storm when service is down
@CircuitBreaker(name = "paymentService")
@Retry(name = "paymentService")
public PaymentResponse processPayment(Payment payment) {
    // ...
}

# Symptom: Traces incomplete or missing in Jaeger/Zipkin

# Solution 1: Verify OTLP exporter endpoint is correct
curl http://jaeger-collector:4317
# Should return gRPC service info

# Solution 2: Check sampling configuration
# By default, OpenTelemetry samples 100% of traces in development
# Verify sampling rate is appropriate
export OTEL_TRACES_SAMPLER=always_on  # For testing
export OTEL_TRACES_SAMPLER=traceidratio  # For production
export OTEL_TRACES_SAMPLER_ARG=0.1  # Sample 10%

# Solution 3: Verify trace context propagation
# Check HTTP headers contain trace IDs
curl -v http://order-service:8080/orders
# Look for: traceparent: 00-<trace-id>-<span-id>-01

# Solution 4: Enable debug logging
logging.level.io.opentelemetry=DEBUG

# Solution 5: Check span processor is flushing
# Spans are batched - may take 5-30 seconds to appear
# For immediate export in tests:
OtlpGrpcSpanExporter spanExporter = OtlpGrpcSpanExporter.builder()
    .setEndpoint("http://jaeger-collector:4317")
    .build();

SimpleSpanProcessor spanProcessor = SimpleSpanProcessor.create(spanExporter);
// SimpleSpanProcessor exports immediately (no batching)

# Symptom: Chaos Mesh experiment took down entire service

# Solution 1: Start with limited scope
# Use 'mode: one' to affect only 1 pod
apiVersion: chaos-mesh.org/v1alpha1
kind: PodChaos
spec:
  mode: one  # Only affect 1 pod, not all
  value: "1"

# Solution 2: Set short durations initially
spec:
  duration: "30s"  # Start with 30 seconds, not minutes

# Solution 3: Use percentage mode for gradual testing
spec:
  mode: fixed-percent
  value: "10"  # Affect 10% of pods

# Solution 4: Implement automatic rollback
# Chaos Mesh can auto-pause if metrics exceed thresholds
apiVersion: chaos-mesh.org/v1alpha1
kind: Workflow
spec:
  templates:
  - name: chaos-with-checks
    steps:
    - - name: run-chaos
        template: network-chaos
    - - name: check-slo
        template: slo-check

  - name: slo-check
    container:
      image: prom/prometheus:latest
      command:
      - sh
      - -c
      - |
        # Query error rate
        ERROR_RATE=$(promtool query instant 'rate(http_errors[5m])' | ...)
        if [ $ERROR_RATE -gt 0.05 ]; then
          echo "SLO violated! Stopping chaos experiment"
          exit 1
        fi

# Symptom: Failure prediction model has low F1 score or too many false positives

# Solution 1: Check class imbalance
df['failure'].value_counts()
# If failures <1% of data, model may always predict "no failure"

# Solution 2: Use class weights
model = RandomForestClassifier(
    class_weight={0: 1, 1: 100},  # Weight failures 100x more
    # ... other params
)

# Solution 3: Collect more failure examples
# Train on longer time period or inject synthetic failures

# Solution 4: Feature engineering
# Add rate-of-change features
df['latency_change_rate'] = df['request_latency_p99'].pct_change()
df['error_rate_change'] = df['error_rate'].diff()

# Add rolling window statistics
df['latency_rolling_mean'] = df['request_latency_p99'].rolling(window=10).mean()
df['latency_rolling_std'] = df['request_latency_p99'].rolling(window=10).std()

# Solution 5: Hyperparameter tuning
from sklearn.model_selection import GridSearchCV

param_grid = {
    'n_estimators': [50, 100, 200],
    'max_depth': [5, 10, 20],
    'min_samples_split': [2, 5, 10],
    'class_weight': ['balanced', {0: 1, 1: 50}, {0: 1, 1: 100}]
}

grid_search = GridSearchCV(
    RandomForestClassifier(),
    param_grid,
    cv=5,
    scoring='f1',
    n_jobs=-1
)
grid_search.fit(X_train, y_train)
best_model = grid_search.best_estimator_

// BAD: No resilience patterns
public OrderResponse createOrder(Order order) {
    Payment payment = paymentClient.process(order.getPayment());
    Inventory inventory = inventoryClient.reserve(order.getItems());
    return new OrderResponse(order.getId(), "SUCCESS");
}

// GOOD: Multiple resilience layers
@CircuitBreaker(name = "orderService", fallbackMethod = "createOrderFallback")
@Retry(name = "orderService")
@TimeLimiter(name = "orderService")
public CompletableFuture<OrderResponse> createOrder(Order order) {
    return CompletableFuture.supplyAsync(() -> {
        // Process with resilience patterns
        Payment payment = paymentService.processWithResilience(order);
        Inventory inventory = inventoryService.reserveWithResilience(order);
        return new OrderResponse(order.getId(), "SUCCESS");
    });
}

# Every service must expose:
# - Health checks (/health/liveness, /health/readiness)
# - Metrics endpoint (/metrics)
# - Distributed tracing (OpenTelemetry instrumentation)
# - Structured logging (JSON format)

# Example: Spring Boot Actuator configuration
management:
  endpoints:
    web:
      exposure:
        include: health,metrics,prometheus,info
  health:
    circuitbreakers.enabled: true
    ratelimiters.enabled: true
  metrics:
    export:
      prometheus.enabled: true
    tags:
      application: ${spring.application.name}
      environment: ${ENV:development}

# Run chaos experiments on schedule (not ad-hoc)
# Example: Daily chaos testing schedule

# Monday: Network latency
# Tuesday: Pod failures
# Wednesday: Memory pressure
# Thursday: Database slowdown
# Friday: Full disaster recovery drill

# Automate with GitOps (ArgoCD + Chaos Mesh)
apiVersion: chaos-mesh.org/v1alpha1
kind: Schedule
metadata:
  name: weekly-chaos-tests
spec:
  schedule: "0 10 * * 1-5"  # Weekdays at 10 AM
  type: NetworkChaos
  networkChaos:
    action: delay
    mode: all
    selector:
      namespaces: ["production"]
    delay:
      latency: "100ms"
    duration: "10m"

# Define Service Level Objectives (SLOs) based on business requirements
# Example: E-commerce checkout service

SLOs = {
    "availability": 0.999,  # 99.9% uptime (43.2 min/month downtime budget)
    "latency_p99": 1.0,     # 99th percentile <1 second
    "error_rate": 0.001     # <0.1% errors
}

# Calculate error budget
monthly_requests = 10_000_000
error_budget = monthly_requests * (1 - SLOs["availability"])
print(f"Monthly error budget: {error_budget} failed requests")

# Monitor error budget consumption
def check_error_budget(current_errors: int) -> dict:
    """Check if we're within error budget."""
    budget_remaining = error_budget - current_errors
    budget_pct = (budget_remaining / error_budget) * 100

    if budget_pct < 10:
        return {
            "status": "CRITICAL",
            "action": "STOP all risky deployments. Focus on reliability."
        }
    elif budget_pct < 25:
        return {
            "status": "WARNING",
            "action": "Slow down feature velocity. Prioritize fixes."
        }
    else:
        return {
            "status": "HEALTHY",
            "action": "Continue normal operations."
        }