Infrastructure Services - Comprehensive Overview

# OpenAPI specification example
openapi: 3.0.0
info:
  title: Configuration Service API
  version: 1.0.0
paths:
  /api/v1/config/{service}:
    get:
      summary: Get service configuration
      parameters:
        - name: service
          in: path
          required: true
          schema:
            type: string
      responses:
        '200':
          description: Configuration retrieved successfully
          content:
            application/json:
              schema:
                $ref: '#/components/schemas/Configuration'

// Configuration service definition
syntax = "proto3";

package sindhan.infrastructure.config.v1;

service ConfigurationService {
  rpc GetConfiguration(GetConfigurationRequest) returns (GetConfigurationResponse);
  rpc UpdateConfiguration(UpdateConfigurationRequest) returns (UpdateConfigurationResponse);
  rpc WatchConfiguration(WatchConfigurationRequest) returns (stream ConfigurationEvent);
}

message GetConfigurationRequest {
  string service_name = 1;
  string environment = 2;
  repeated string keys = 3;
}

message GetConfigurationResponse {
  map<string, string> configuration = 1;
  string version = 2;
}

{
  "eventType": "configuration.updated",
  "eventVersion": "1.0",
  "source": "configuration-service",
  "timestamp": "2024-01-15T10:30:00Z",
  "data": {
    "serviceName": "agent-service",
    "configurationKey": "feature.ai-model",
    "oldValue": "gpt-3.5-turbo",
    "newValue": "gpt-4",
    "environment": "production"
  },
  "correlationId": "req-12345-67890",
  "traceId": "trace-abcdef-123456"
}

# Kafka topic configuration
apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaTopic
metadata:
  name: infrastructure-events
  labels:
    strimzi.io/cluster: sindhan-kafka
spec:
  partitions: 12
  replicas: 3
  config:
    retention.ms: 604800000  # 7 days
    cleanup.policy: delete
    compression.type: lz4

// Event sourcing pattern implementation
use std::collections::HashMap;
use uuid::Uuid;
use chrono::{DateTime, Utc};
use serde::{Serialize, Deserialize};
use anyhow::Result;
 
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ConfigurationEvent {
    pub event_id: Uuid,
    pub event_type: String,
    pub event_data: serde_json::Value,
    pub timestamp: DateTime<Utc>,
    pub version: u64,
    pub aggregate_id: String,
}
 
pub struct ConfigurationEventStore {
    events: Vec<ConfigurationEvent>,
    snapshots: HashMap<String, serde_json::Value>,
}
 
impl ConfigurationEventStore {
    pub fn new() -> Self {
        Self {
            events: Vec::new(),
            snapshots: HashMap::new(),
        }
    }
    
    pub async fn append_event(&mut self, event: ConfigurationEvent) -> Result<()> {
        // Append event to the event store
        self.events.push(event.clone());
        
        // Async propagation to read models
        self.propagate_event_async(event).await?;
        Ok(())
    }
    
    pub fn get_aggregate_state(&self, aggregate_id: &str) -> Result<serde_json::Value> {
        // Reconstruct aggregate state from events
        let events: Vec<&ConfigurationEvent> = self.events
            .iter()
            .filter(|e| e.aggregate_id == aggregate_id)
            .collect();
        
        self.replay_events(events)
    }
    
    async fn propagate_event_async(&self, event: ConfigurationEvent) -> Result<()> {
        // Implementation for async event propagation
        todo!("Implement async event propagation")
    }
    
    fn replay_events(&self, events: Vec<&ConfigurationEvent>) -> Result<serde_json::Value> {
        // Implementation for event replay
        todo!("Implement event replay")
    }
}

// Distributed transaction pattern
use std::sync::Arc;
use tokio::sync::RwLock;
use anyhow::Result;
 
#[derive(Debug, Clone)]
pub struct TransactionParticipant {
    pub service: String,
    pub operation: String,
    pub status: TransactionStatus,
}
 
#[derive(Debug, Clone, PartialEq)]
pub enum TransactionStatus {
    Prepared,
    Committed,
    Aborted,
}
 
pub struct DistributedTransaction {
    transaction_manager: Arc<dyn TransactionManager>,
    participants: Arc<RwLock<Vec<TransactionParticipant>>>,
}
 
#[async_trait::async_trait]
pub trait TransactionManager: Send + Sync {
    async fn prepare(&self, participant: &TransactionParticipant) -> Result<bool>;
    async fn commit(&self, participant: &TransactionParticipant) -> Result<()>;
    async fn abort(&self, participant: &TransactionParticipant) -> Result<()>;
}
 
impl DistributedTransaction {
    pub fn new(transaction_manager: Arc<dyn TransactionManager>) -> Self {
        Self {
            transaction_manager,
            participants: Arc::new(RwLock::new(Vec::new())),
        }
    }
    
    pub async fn add_participant(&self, service: String, operation: String) {
        let mut participants = self.participants.write().await;
        participants.push(TransactionParticipant {
            service,
            operation,
            status: TransactionStatus::Prepared,
        });
    }
    
    pub async fn commit(&self) -> Result<bool> {
        let participants = self.participants.read().await;
        
        // Phase 1: Prepare
        for participant in participants.iter() {
            if !self.transaction_manager.prepare(participant).await? {
                self.abort().await?;
                return Ok(false);
            }
        }
        
        // Phase 2: Commit
        for participant in participants.iter() {
            self.transaction_manager.commit(participant).await?;
        }
        
        Ok(true)
    }
    
    pub async fn abort(&self) -> Result<()> {
        let participants = self.participants.read().await;
        
        for participant in participants.iter() {
            self.transaction_manager.abort(participant).await?;
        }
        
        Ok(())
    }
}

# Istio security policy
apiVersion: security.istio.io/v1beta1
kind: PeerAuthentication
metadata:
  name: default
  namespace: sindhan-infrastructure
spec:
  mtls:
    mode: STRICT
 
---
apiVersion: security.istio.io/v1beta1
kind: AuthorizationPolicy
metadata:
  name: infrastructure-authz
  namespace: sindhan-infrastructure
spec:
  selector:
    matchLabels:
      app: infrastructure-service
  rules:
  - from:
    - source:
        principals: ["cluster.local/ns/sindhan-platform/sa/platform-service"]
    to:
    - operation:
        methods: ["GET", "POST"]

// JWT validation middleware
use jsonwebtoken::{decode, DecodingKey, Validation, Algorithm};
use serde::{Deserialize, Serialize};
use anyhow::{Result, Context};
 
#[derive(Debug, Serialize, Deserialize)]
struct Claims {
    sub: String,
    exp: usize,
    iss: String,
    aud: String,
}
 
pub struct JWTAuthenticationMiddleware {
    jwt_secret: String,
    issuer: String,
}
 
impl JWTAuthenticationMiddleware {
    pub fn new(jwt_secret: String, issuer: String) -> Self {
        Self { jwt_secret, issuer }
    }
    
    pub fn authenticate(&self, token: &str) -> Result<Claims> {
        let validation = Validation::new(Algorithm::HS256);
        
        let token_data = decode::<Claims>(
            token,
            &DecodingKey::from_secret(self.jwt_secret.as_ref()),
            &validation,
        )
        .context("Failed to decode JWT token")?;
        
        if token_data.claims.iss != self.issuer {
            return Err(anyhow::anyhow!("Invalid token issuer"));
        }
        
        Ok(token_data.claims)
    }
    
    fn extract_token(&self, authorization_header: &str) -> Option<&str> {
        if authorization_header.starts_with("Bearer ") {
            Some(&authorization_header[7..])
        } else {
            None
        }
    }
}

# Prometheus scraping configuration
apiVersion: v1
kind: ConfigMap
metadata:
  name: prometheus-config
data:
  prometheus.yml: |
    global:
      scrape_interval: 15s
    scrape_configs:
    - job_name: 'infrastructure-services'
      kubernetes_sd_configs:
      - role: endpoints
        namespaces:
          names:
          - sindhan-infrastructure
      relabel_configs:
      - source_labels: [__meta_kubernetes_service_annotation_prometheus_io_scrape]
        action: keep
        regex: true

// OpenTelemetry tracing setup
use opentelemetry::trace::{TraceError, Tracer};
use opentelemetry::{global, sdk::trace as sdktrace};
use opentelemetry_jaeger::new_agent_pipeline;
use std::time::Duration;
 
pub fn init_tracer() -> Result<sdktrace::Tracer, TraceError> {
    new_agent_pipeline()
        .with_service_name("configuration-service")
        .with_agent_endpoint("http://jaeger-agent:14268/api/traces")
        .with_max_packet_size(65_000)
        .install_batch(opentelemetry::runtime::Tokio)
}
 
// Instrument service calls
pub async fn call_downstream_service(
    service_name: &str,
    operation: &str,
) -> Result<String> {
    let tracer = global::tracer("configuration-service");
    
    let span = tracer
        .span_builder(format!("{}.{}", service_name, operation))
        .with_attributes(vec![
            opentelemetry::KeyValue::new("service.name", service_name.to_string()),
            opentelemetry::KeyValue::new("operation.name", operation.to_string()),
        ])
        .start(&tracer);
    
    let _guard = span.clone();
    
    // Make the actual service call
    let result = make_service_call(service_name, operation).await;
    
    match &result {
        Ok(_) => span.set_attribute(opentelemetry::KeyValue::new("response.status", "success")),
        Err(e) => {
            span.set_attribute(opentelemetry::KeyValue::new("response.status", "error"));
            span.set_attribute(opentelemetry::KeyValue::new("error.message", e.to_string()));
        }
    }
    
    result
}
 
async fn make_service_call(service_name: &str, operation: &str) -> Result<String> {
    // Implementation for actual service call
    todo!("Implement service call")
}

{
  "timestamp": "2024-01-15T10:30:00.123Z",
  "level": "INFO",
  "service": "configuration-service",
  "version": "1.2.3",
  "environment": "production",
  "correlation_id": "req-12345-67890",
  "trace_id": "trace-abcdef-123456",
  "span_id": "span-123456-abcdef",
  "message": "Configuration updated successfully",
  "context": {
    "service_name": "agent-service",
    "configuration_key": "feature.ai-model",
    "old_value": "gpt-3.5-turbo",
    "new_value": "gpt-4"
  },
  "duration_ms": 45,
  "user_id": "user-789",
  "request_id": "req-abc-def-ghi"
}

# Service deployment example
apiVersion: apps/v1
kind: Deployment
metadata:
  name: configuration-service
  namespace: sindhan-infrastructure
spec:
  replicas: 3
  selector:
    matchLabels:
      app: configuration-service
  template:
    metadata:
      labels:
        app: configuration-service
        version: v1.2.3
    spec:
      serviceAccountName: configuration-service
      containers:
      - name: configuration-service
        image: sindhan/configuration-service:v1.2.3
        ports:
        - containerPort: 8080
          name: http
        - containerPort: 9090
          name: metrics
        env:
        - name: DATABASE_URL
          valueFrom:
            secretKeyRef:
              name: configuration-db-secret
              key: url
        resources:
          requests:
            memory: "256Mi"
            cpu: "100m"
          limits:
            memory: "512Mi"
            cpu: "500m"
        livenessProbe:
          httpGet:
            path: /health
            port: 8080
          initialDelaySeconds: 30
          periodSeconds: 10
        readinessProbe:
          httpGet:
            path: /ready
            port: 8080
          initialDelaySeconds: 5
          periodSeconds: 5

# ArgoCD application
apiVersion: argoproj.io/v1alpha1
kind: Application
metadata:
  name: infrastructure-services
  namespace: argocd
spec:
  project: default
  source:
    repoURL: https://github.com/sindhan-ai/infrastructure-manifests
    targetRevision: HEAD
    path: infrastructure-services
  destination:
    server: https://kubernetes.default.svc
    namespace: sindhan-infrastructure
  syncPolicy:
    automated:
      prune: true
      selfHeal: true
    syncOptions:
    - CreateNamespace=true

apiVersion: autoscaling/v2
kind: HorizontalPodAutoscaler
metadata:
  name: configuration-service-hpa
spec:
  scaleTargetRef:
    apiVersion: apps/v1
    kind: Deployment
    name: configuration-service
  minReplicas: 3
  maxReplicas: 20
  metrics:
  - type: Resource
    resource:
      name: cpu
      target:
        type: Utilization
        averageUtilization: 70
  - type: Resource
    resource:
      name: memory
      target:
        type: Utilization
        averageUtilization: 80

// Multi-level caching implementation
use std::collections::HashMap;
use std::sync::Arc;
use tokio::sync::RwLock;
use redis::AsyncCommands;
use anyhow::Result;
 
pub struct ConfigurationCache {
    l1_cache: Arc<RwLock<HashMap<String, String>>>,
    l2_cache: redis::Client,
    ttl: u64,
}
 
impl ConfigurationCache {
    pub fn new(redis_url: &str, ttl: u64) -> Result<Self> {
        let l2_cache = redis::Client::open(redis_url)?;
        
        Ok(Self {
            l1_cache: Arc::new(RwLock::new(HashMap::new())),
            l2_cache,
            ttl,
        })
    }
    
    pub async fn get_configuration(
        &self,
        service_name: &str,
        key: &str,
    ) -> Result<Option<String>> {
        let cache_key = format!("{}:{}", service_name, key);
        
        // L1 Cache check
        {
            let l1 = self.l1_cache.read().await;
            if let Some(value) = l1.get(&cache_key) {
                return Ok(Some(value.clone()));
            }
        }
        
        // L2 Cache check
        let mut conn = self.l2_cache.get_async_connection().await?;
        if let Some(value) = conn.get::<_, Option<String>>(&cache_key).await? {
            // Update L1 cache
            {
                let mut l1 = self.l1_cache.write().await;
                l1.insert(cache_key, value.clone());
            }
            return Ok(Some(value));
        }
        
        // Database lookup
        if let Some(value) = self.database_lookup(service_name, key).await? {
            // Update both caches
            conn.setex(&cache_key, self.ttl, &value).await?;
            {
                let mut l1 = self.l1_cache.write().await;
                l1.insert(cache_key, value.clone());
            }
            return Ok(Some(value));
        }
        
        Ok(None)
    }
    
    async fn database_lookup(
        &self,
        service_name: &str,
        key: &str,
    ) -> Result<Option<String>> {
        // Implementation for database lookup
        todo!("Implement database lookup")
    }
}

# Velero backup configuration
apiVersion: velero.io/v1
kind: Backup
metadata:
  name: infrastructure-services-backup
spec:
  includedNamespaces:
  - sindhan-infrastructure
  storageLocation: aws-s3
  ttl: 720h0m0s  # 30 days
  includeClusterResources: true
  hooks:
    resources:
    - name: database-backup-hook
      includedNamespaces:
      - sindhan-infrastructure
      pre:
      - exec:
          container: postgres
          command:
          - /bin/bash
          - -c
          - pg_dump -h localhost -U postgres sindhan_config > /tmp/backup.sql

# Cross-region replication
apiVersion: networking.istio.io/v1beta1
kind: DestinationRule
metadata:
  name: configuration-service-dr
spec:
  host: configuration-service
  trafficPolicy:
    outlierDetection:
      consecutiveErrors: 3
      interval: 30s
      baseEjectionTime: 30s
  subsets:
  - name: us-east-1
    labels:
      region: us-east-1
  - name: us-west-2
    labels:
      region: us-west-2