Monitoring and Observability

/**
 * Watch Kubernetes resources for changes
 */
class Watch {
  constructor(config: KubeConfig);
  
  /**
   * Watch for resource changes
   * @param path - API path to watch (e.g., '/api/v1/namespaces/default/pods')
   * @param queryParams - Query parameters for filtering
   * @param callback - Called for each change event
   * @param done - Called when watch ends or errors
   * @returns Promise that resolves to AbortController for canceling the watch
   */
  watch(
    path: string,
    queryParams: Record<string, string | number | boolean | undefined>,
    callback: (phase: string, apiObj: any, watchObj?: any) => void,
    done: (err: any) => void
  ): Promise<AbortController>;
  
  /** Server-side close indicator */
  static readonly SERVER_SIDE_CLOSE: object;
}

import { KubeConfig, Watch } from '@kubernetes/client-node';

const kc = new KubeConfig();
kc.loadFromDefault();

const watch = new Watch(kc);

// Watch pods in a namespace
const watchPods = async () => {
  try {
    const controller = await watch.watch(
      '/api/v1/namespaces/default/pods',
      {
        labelSelector: 'app=web'
      },
      (phase, pod) => {
        console.log(`Pod ${phase}:`, pod.metadata?.name);
        switch (phase) {
          case 'ADDED':
            console.log('New pod created');
            break;
          case 'MODIFIED':
            console.log('Pod updated, status:', pod.status?.phase);
            break;
          case 'DELETED':
            console.log('Pod deleted');
            break;
          case 'ERROR':
            console.error('Watch error:', pod);
            break;
        }
      },
      (err) => {
        if (err) {
          console.error('Watch ended with error:', err);
        } else {
          console.log('Watch ended normally');
        }
      }
    );
    
    // Cancel watch after 30 seconds
    setTimeout(() => {
      controller.abort();
    }, 30000);
    
  } catch (error) {
    console.error('Failed to start watch:', error);
  }
};

// Watch deployments with resource version
const watchDeployments = async () => {
  let resourceVersion: string | undefined;
  
  const startWatch = async () => {
    const controller = await watch.watch(
      '/apis/apps/v1/namespaces/default/deployments',
      {
        resourceVersion,
        watch: true
      },
      (phase, deployment, watchObj) => {
        console.log(`Deployment ${phase}:`, deployment.metadata?.name);
        
        // Update resource version for reconnection
        if (watchObj?.object?.metadata?.resourceVersion) {
          resourceVersion = watchObj.object.metadata.resourceVersion;
        }
      },
      (err) => {
        if (err && err !== Watch.SERVER_SIDE_CLOSE) {
          console.error('Watch error, restarting:', err);
          // Restart watch on error
          setTimeout(startWatch, 1000);
        }
      }
    );
  };
  
  startWatch();
};

// Watch all resources in a namespace
const watchNamespace = async () => {
  const resources = [
    '/api/v1/namespaces/default/pods',
    '/api/v1/namespaces/default/services',
    '/apis/apps/v1/namespaces/default/deployments'
  ];
  
  const controllers = await Promise.all(
    resources.map(path => 
      watch.watch(
        path,
        {},
        (phase, obj) => {
          console.log(`${obj.kind} ${phase}:`, obj.metadata?.name);
        },
        (err) => {
          if (err) console.error(`Watch error for ${path}:`, err);
        }
      )
    )
  );
  
  // Cancel all watches
  const cancelAll = () => {
    controllers.forEach(controller => controller.abort());
  };
  
  process.on('SIGINT', cancelAll);
};

/**
 * Stream logs from pod containers
 */
class Log {
  constructor(config: KubeConfig);
  
  /**
   * Stream container logs
   * @param namespace - Pod namespace
   * @param podName - Pod name
   * @param containerName - Container name
   * @param stream - Writable stream for log output
   * @param options - Log streaming options
   * @returns Promise that resolves to AbortController for canceling the stream
   */
  log(
    namespace: string,
    podName: string,
    containerName: string,
    stream: Writable,
    options?: LogOptions
  ): Promise<AbortController>;
}

interface LogOptions {
  /** Follow log stream (tail -f behavior) */
  follow?: boolean;
  /** Limit number of bytes returned */
  limitBytes?: number;
  /** Pretty-print the output */
  pretty?: boolean;
  /** Return logs from previous container instance */
  previous?: boolean;
  /** Relative time in seconds before now */
  sinceSeconds?: number;
  /** Absolute time (RFC3339) to start streaming from */
  sinceTime?: string;
  /** Number of lines from end of log to show */
  tailLines?: number;
  /** Include timestamps in log output */
  timestamps?: boolean;
}

/**
 * Utility function to add log options to URL search parameters
 */
function AddOptionsToSearchParams(options: LogOptions, searchParams: URLSearchParams): void;

import { KubeConfig, Log } from '@kubernetes/client-node';
import * as fs from 'fs';

const kc = new KubeConfig();
kc.loadFromDefault();

const logClient = new Log(kc);

// Stream logs to console
const streamLogs = async () => {
  try {
    const controller = await logClient.log(
      'default',
      'my-pod',
      'my-container',
      process.stdout,
      {
        follow: true,
        timestamps: true,
        tailLines: 100
      }
    );
    
    console.log('Streaming logs... Press Ctrl+C to stop');
    
    process.on('SIGINT', () => {
      controller.abort();
      process.exit(0);
    });
  } catch (error) {
    console.error('Log streaming error:', error);
  }
};

// Save logs to file
const saveLogs = async () => {
  const logFile = fs.createWriteStream('./pod-logs.txt');
  
  try {
    await logClient.log(
      'default',
      'my-pod',
      'my-container',
      logFile,
      {
        sinceSeconds: 3600, // Last hour
        timestamps: true
      }
    );
    
    console.log('Logs saved to pod-logs.txt');
  } catch (error) {
    console.error('Failed to save logs:', error);
  } finally {
    logFile.end();
  }
};

// Get previous container logs
const getPreviousLogs = async () => {
  try {
    await logClient.log(
      'default',
      'crashed-pod',
      'app-container',
      process.stdout,
      {
        previous: true,
        tailLines: 50
      }
    );
  } catch (error) {
    console.error('Failed to get previous logs:', error);
  }
};

// Multi-container pod logs
const multiContainerLogs = async () => {
  const containers = ['web', 'sidecar', 'init'];
  
  for (const container of containers) {
    console.log(`\n=== Logs for container: ${container} ===`);
    
    try {
      await logClient.log(
        'default',
        'multi-container-pod',
        container,
        process.stdout,
        {
          tailLines: 20,
          timestamps: true
        }
      );
    } catch (error) {
      console.error(`Error getting logs for ${container}:`, error);
    }
  }
};

/**
 * Create an informer for watching and caching resources
 * @param kubeconfig - Kubernetes configuration
 * @param path - API path to watch
 * @param listPromiseFn - Function to list resources
 * @param labelSelector - Optional label selector for filtering
 * @returns Informer instance
 */
function makeInformer<T>(
  kubeconfig: KubeConfig,
  path: string,
  listPromiseFn: ListPromise<T>,
  labelSelector?: string
): Informer<T>;

interface Informer<T> {
  /** Register event handler */
  on(verb: string, fn: ObjectCallback<T>): void;
  /** Unregister event handler */
  off(verb: string, fn: ObjectCallback<T>): void;
  /** Start the informer */
  start(): Promise<void>;
  /** Stop the informer */
  stop(): Promise<void>;
}

type ObjectCallback<T> = (obj: T) => void;
type ErrorCallback = (err?: any) => void;
type ListCallback<T> = (list: T[], ResourceVersion: string) => void;
type ListPromise<T> = () => Promise<KubernetesListObject<T>>;

// Event type constants
const ADD: string;
const UPDATE: string;
const CHANGE: string;
const DELETE: string;
const CONNECT: string;
const ERROR: string;

import { 
  KubeConfig, 
  CoreV1Api, 
  makeInformer,
  ADD, 
  UPDATE, 
  DELETE,
  ERROR 
} from '@kubernetes/client-node';

const kc = new KubeConfig();
kc.loadFromDefault();

const k8sApi = kc.makeApiClient(CoreV1Api);

// Create pod informer
const createPodInformer = () => {
  const informer = makeInformer(
    kc,
    '/api/v1/namespaces/default/pods',
    () => k8sApi.listNamespacedPod('default'),
    'app=web' // label selector
  );
  
  // Register event handlers
  informer.on(ADD, (pod) => {
    console.log('Pod added:', pod.metadata?.name);
    console.log('Status:', pod.status?.phase);
  });
  
  informer.on(UPDATE, (pod) => {
    console.log('Pod updated:', pod.metadata?.name);
    console.log('New status:', pod.status?.phase);
  });
  
  informer.on(DELETE, (pod) => {
    console.log('Pod deleted:', pod.metadata?.name);
  });
  
  informer.on(ERROR, (err) => {
    console.error('Informer error:', err);
  });
  
  return informer;
};

// Start and manage informer
const runInformer = async () => {
  const informer = createPodInformer();
  
  try {
    await informer.start();
    console.log('Informer started successfully');
    
    // Stop informer after 60 seconds
    setTimeout(async () => {
      await informer.stop();
      console.log('Informer stopped');
    }, 60000);
    
  } catch (error) {
    console.error('Failed to start informer:', error);
  }
};

// Deployment informer with custom handlers
const createDeploymentInformer = () => {
  const informer = makeInformer(
    kc,
    '/apis/apps/v1/namespaces/default/deployments',
    () => kc.makeApiClient(AppsV1Api).listNamespacedDeployment('default')
  );
  
  const handleDeploymentChange = (deployment) => {
    const name = deployment.metadata?.name;
    const replicas = deployment.status?.replicas || 0;
    const ready = deployment.status?.readyReplicas || 0;
    
    console.log(`Deployment ${name}: ${ready}/${replicas} ready`);
    
    if (ready < replicas) {
      console.log('Deployment not fully ready');
    }
  };
  
  informer.on(ADD, handleDeploymentChange);
  informer.on(UPDATE, handleDeploymentChange);
  
  informer.on(DELETE, (deployment) => {
    console.log('Deployment deleted:', deployment.metadata?.name);
  });
  
  return informer;
};

/**
 * Collect metrics from Kubernetes cluster
 */
class Metrics {
  constructor(config: KubeConfig);
  
  /**
   * Get node resource metrics
   * @returns Promise that resolves to node metrics list
   */
  getNodeMetrics(): Promise<NodeMetricsList>;
  
  /**
   * Get pod resource metrics
   * @param namespace - Optional namespace filter
   * @returns Promise that resolves to pod metrics list
   */
  getPodMetrics(namespace?: string): Promise<PodMetricsList>;
}

interface Usage {
  /** CPU usage (e.g., "250m" for 250 millicores) */
  cpu: string;
  /** Memory usage (e.g., "128Mi" for 128 mebibytes) */
  memory: string;
}

interface ContainerMetric {
  /** Container name */
  name: string;
  /** Resource usage */
  usage: Usage;
}

interface PodMetric {
  /** Pod metadata */
  metadata: {
    name: string;
    namespace: string;
    selfLink: string;
    creationTimestamp: string;
  };
  /** Timestamp of metrics collection */
  timestamp: string;
  /** Metrics collection window */
  window: string;
  /** Container metrics */
  containers: ContainerMetric[];
}

interface NodeMetric {
  /** Node metadata */
  metadata: {
    name: string;
    selfLink: string;
    creationTimestamp: string;
  };
  /** Timestamp of metrics collection */
  timestamp: string;
  /** Metrics collection window */
  window: string;
  /** Node resource usage */
  usage: Usage;
}

interface PodMetricsList {
  kind: string;
  apiVersion: string;
  metadata: object;
  items: PodMetric[];
}

interface NodeMetricsList {
  kind: string;
  apiVersion: string;
  metadata: object;
  items: NodeMetric[];
}

import { KubeConfig, Metrics } from '@kubernetes/client-node';

const kc = new KubeConfig();
kc.loadFromDefault();

const metrics = new Metrics(kc);

// Get node metrics
const getNodeMetrics = async () => {
  try {
    const nodeMetrics = await metrics.getNodeMetrics();
    
    console.log('Node Metrics:');
    nodeMetrics.items.forEach(node => {
      console.log(`Node: ${node.metadata.name}`);
      console.log(`CPU: ${node.usage.cpu}`);
      console.log(`Memory: ${node.usage.memory}`);
      console.log('---');
    });
  } catch (error) {
    console.error('Failed to get node metrics:', error);
  }
};

// Get pod metrics for specific namespace
const getPodMetrics = async () => {
  try {
    const podMetrics = await metrics.getPodMetrics('default');
    
    console.log('Pod Metrics:');
    podMetrics.items.forEach(pod => {
      console.log(`Pod: ${pod.metadata.name}`);
      console.log(`Namespace: ${pod.metadata.namespace}`);
      console.log('Containers:');
      
      pod.containers.forEach(container => {
        console.log(`  ${container.name}:`);
        console.log(`    CPU: ${container.usage.cpu}`);
        console.log(`    Memory: ${container.usage.memory}`);
      });
      console.log('---');
    });
  } catch (error) {
    console.error('Failed to get pod metrics:', error);
  }
};

// Monitor resource usage over time
const monitorResources = async () => {
  const interval = setInterval(async () => {
    try {
      const [nodeMetrics, podMetrics] = await Promise.all([
        metrics.getNodeMetrics(),
        metrics.getPodMetrics()
      ]);
      
      // Calculate total cluster resource usage
      const totalNodeCpu = nodeMetrics.items.reduce((sum, node) => {
        const cpu = parseInt(node.usage.cpu.replace('n', '')) / 1000000; // Convert nanocores to millicores
        return sum + cpu;
      }, 0);
      
      const totalPodCpu = podMetrics.items.reduce((sum, pod) => {
        const podCpu = pod.containers.reduce((containerSum, container) => {
          const cpu = parseInt(container.usage.cpu.replace('m', ''));
          return containerSum + cpu;
        }, 0);
        return sum + podCpu;
      }, 0);
      
      console.log(`Cluster CPU Usage: ${totalPodCpu}m / ${totalNodeCpu}m`);
      console.log(`CPU Utilization: ${((totalPodCpu / totalNodeCpu) * 100).toFixed(2)}%`);
      
    } catch (error) {
      console.error('Metrics collection error:', error);
    }
  }, 30000); // Every 30 seconds
  
  // Stop monitoring after 5 minutes
  setTimeout(() => {
    clearInterval(interval);
    console.log('Resource monitoring stopped');
  }, 300000);
};

/**
 * Resource usage information
 */
class ResourceUsage {
  /** Total resource capacity */
  Capacity: number | bigint;
  /** Total requested resources */
  RequestTotal: number | bigint;
  /** Total resource limits */
  LimitTotal: number | bigint;
}

/**
 * Current resource usage
 */
class CurrentResourceUsage {
  /** Current resource usage */
  CurrentUsage: number | bigint;
  /** Total requested resources */
  RequestTotal: number | bigint;
  /** Total resource limits */
  LimitTotal: number | bigint;
}

/**
 * Node resource status
 */
class NodeStatus {
  /** Node object */
  Node: V1Node;
  /** CPU usage information */
  CPU: ResourceUsage;
  /** Memory usage information */
  Memory: ResourceUsage;
}

/**
 * Container resource status
 */
class ContainerStatus {
  /** Container name */
  name: string;
  /** CPU usage */
  CPUUsage: CurrentResourceUsage;
  /** Memory usage */
  MemoryUsage: CurrentResourceUsage;
}

/**
 * Pod resource status
 */
class PodStatus {
  /** Pod object */
  Pod: V1Pod;
  /** Container statuses */
  Containers: ContainerStatus[];
}

import { KubeConfig, CoreV1Api, Metrics } from '@kubernetes/client-node';

const kc = new KubeConfig();
kc.loadFromDefault();

// Utility functions for resource analysis
const parseResourceQuantity = (quantity: string): number => {
  // Parse Kubernetes resource quantities (e.g., "100m", "1Gi")
  const units = {
    'n': 1e-9, 'u': 1e-6, 'm': 1e-3,
    'k': 1e3, 'M': 1e6, 'G': 1e9, 'T': 1e12,
    'Ki': 1024, 'Mi': 1024**2, 'Gi': 1024**3, 'Ti': 1024**4
  };
  
  const match = quantity.match(/^(\d+(?:\.\d+)?)(.*?)$/);
  if (!match) return 0;
  
  const [, value, unit] = match;
  const multiplier = units[unit] || 1;
  return parseFloat(value) * multiplier;
};

// Create top-like analysis
const analyzeClusterResources = async () => {
  const coreApi = kc.makeApiClient(CoreV1Api);
  const metrics = new Metrics(kc);
  
  try {
    const [nodes, nodeMetrics, podMetrics] = await Promise.all([
      coreApi.listNode(),
      metrics.getNodeMetrics(),
      metrics.getPodMetrics()
    ]);
    
    console.log('CLUSTER RESOURCE ANALYSIS');
    console.log('=========================');
    
    // Analyze nodes
    console.log('\nNODES:');
    console.log('NAME\t\tCPU\t\tMEMORY');
    
    nodes.body.items.forEach(node => {
      const nodeName = node.metadata?.name || 'unknown';
      const nodeMetric = nodeMetrics.items.find(m => m.metadata.name === nodeName);
      
      if (nodeMetric) {
        const cpuUsage = parseResourceQuantity(nodeMetric.usage.cpu);
        const memoryUsage = parseResourceQuantity(nodeMetric.usage.memory);
        
        console.log(`${nodeName}\t${(cpuUsage * 1000).toFixed(0)}m\t\t${(memoryUsage / 1024**2).toFixed(0)}Mi`);
      }
    });
    
    // Analyze pods
    console.log('\nPODS:');
    console.log('NAMESPACE\tNAME\t\tCPU\t\tMEMORY');
    
    podMetrics.items.forEach(pod => {
      const namespace = pod.metadata.namespace;
      const name = pod.metadata.name;
      
      const totalCpu = pod.containers.reduce((sum, container) => {
        return sum + parseResourceQuantity(container.usage.cpu);
      }, 0);
      
      const totalMemory = pod.containers.reduce((sum, container) => {
        return sum + parseResourceQuantity(container.usage.memory);
      }, 0);
      
      console.log(`${namespace}\t\t${name}\t${(totalCpu * 1000).toFixed(0)}m\t\t${(totalMemory / 1024**2).toFixed(0)}Mi`);
    });
    
  } catch (error) {
    console.error('Resource analysis failed:', error);
  }
};

/**
 * Cache interface for Kubernetes objects
 */
interface ObjectCache<T> {
  /** Get object by name and optional namespace */
  get(name: string, namespace?: string): T | undefined;
  /** List all objects, optionally filtered by namespace */
  list(namespace?: string): ReadonlyArray<T>;
}

/**
 * Internal cache structure
 */
type CacheMap<T> = Map<string, Map<string, T>>;

/**
 * Combined list/watch with caching
 */
class ListWatch<T> implements ObjectCache<T>, Informer<T> {
  constructor(
    path: string,
    watch: Watch,
    listFn: ListPromise<T>,
    autoStart?: boolean,
    labelSelector?: string,
    fieldSelector?: string
  );
  
  // ObjectCache methods
  get(name: string, namespace?: string): T | undefined;
  list(namespace?: string): ReadonlyArray<T>;
  
  // Informer methods
  on(verb: string, fn: ObjectCallback<T>): void;
  off(verb: string, fn: ObjectCallback<T>): void;
  start(): Promise<void>;
  stop(): Promise<void>;
}

import { KubeConfig, CoreV1Api, Watch, ListWatch } from '@kubernetes/client-node';

const kc = new KubeConfig();
kc.loadFromDefault();

// Create cached pod watcher
const createPodCache = () => {
  const watch = new Watch(kc);
  const coreApi = kc.makeApiClient(CoreV1Api);
  
  const podCache = new ListWatch(
    '/api/v1/namespaces/default/pods',
    watch,
    () => coreApi.listNamespacedPod('default'),
    true, // auto-start
    'app=web' // label selector
  );
  
  // Use cache methods
  podCache.on('ADD', (pod) => {
    console.log('Pod cached:', pod.metadata?.name);
  });
  
  return podCache;
};

// Query cached data
const queryCachedPods = (cache: ObjectCache<V1Pod>) => {
  // Get specific pod
  const pod = cache.get('my-pod', 'default');
  if (pod) {
    console.log('Found pod:', pod.metadata?.name);
  }
  
  // List all cached pods
  const allPods = cache.list();
  console.log(`Total cached pods: ${allPods.length}`);
  
  // List pods in specific namespace
  const namespacedPods = cache.list('production');
  console.log(`Pods in production: ${namespacedPods.length}`);
};