KubernetesConnector

from connectors import KubernetesConnector

connector = KubernetesConnector()
nodes, edges = connector.parse("k8s-deployments.yaml")

def parse(self, file_path: str) -> tuple[List[Node], List[Edge]]:
    """Parse k8s-deployments.yaml file."""
    self.logger.info(f"Parsing Kubernetes file: {file_path}")
    
    try:
        with open(file_path, 'r') as f:
            # Parse multiple YAML documents
            k8s_docs = list(yaml.safe_load_all(f))
    except Exception as e:
        self.logger.error(f"Failed to parse {file_path}: {e}")
        return [], []
    
    nodes = []
    edges = []
    
    deployments = {}
    services = {}
    
    # First pass: collect all resources
    for doc in k8s_docs:
        if not doc or 'kind' not in doc:
            continue
            
        kind = doc['kind']
        metadata = doc.get('metadata', {})
        name = metadata.get('name')
        
        if kind == 'Deployment':
            deployments[name] = doc
        elif kind == 'Service':
            services[name] = doc
    
    # Second pass: create nodes and edges
    for deployment_name, deployment in deployments.items():
        # Create deployment node
        deployment_node = self._create_deployment_node(deployment)
        nodes.append(deployment_node)
        
        # Extract dependencies from environment variables
        spec = deployment.get('spec', {})
        template = spec.get('template', {})
        pod_spec = template.get('spec', {})
        containers = pod_spec.get('containers', [])
        
        for container in containers:
            env_vars = self._extract_k8s_env_vars(container)
            
            # Create edges for service dependencies
            service_deps = self._extract_k8s_service_dependencies(env_vars)
            for dep_service in service_deps:
                edge = self._create_edge(
                    'calls',
                    deployment_node.id,
                    f"service:{dep_service}"
                )
                edges.append(edge)
    
    # Create service nodes
    for service_name, service in services.items():
        service_node = self._create_k8s_service_node(service)
        nodes.append(service_node)
        
        # Link service to deployment if they have the same name
        if service_name in deployments:
            edge = self._create_edge(
                'exposes',
                f"service:{service_name}",
                f"deployment:{service_name}"
            )
            edges.append(edge)
    
    self.logger.info(f"Parsed {len(nodes)} nodes and {len(edges)} edges from Kubernetes")
    return nodes, edges

def _create_deployment_node(self, deployment: Dict[str, Any]) -> Node:
    """Create a deployment node from Kubernetes deployment."""
    metadata = deployment.get('metadata', {})
    name = metadata.get('name')
    labels = metadata.get('labels', {})
    
    spec = deployment.get('spec', {})
    replicas = spec.get('replicas', 1)
    
    template = spec.get('template', {})
    pod_spec = template.get('spec', {})
    containers = pod_spec.get('containers', [])
    
    # Extract container information
    container_info = {}
    if containers:
        container = containers[0]  # Take first container
        container_info = {
            'image': container.get('image'),
            'resources': container.get('resources', {})
        }
    
    properties = {
        'namespace': metadata.get('namespace'),
        'team': labels.get('team'),
        'replicas': replicas,
        **container_info
    }
    
    # Filter out None values
    properties = {k: v for k, v in properties.items() if v is not None}
    
    return self._create_node('deployment', name, properties)

def _create_k8s_service_node(self, service: Dict[str, Any]) -> Node:
    """Create a service node from Kubernetes service."""
    metadata = service.get('metadata', {})
    name = metadata.get('name')
    
    spec = service.get('spec', {})
    ports = spec.get('ports', [])
    
    # Extract port information
    port_info = {}
    if ports:
        port = ports[0]  # Take first port
        port_info = {
            'port': port.get('port'),
            'target_port': port.get('targetPort')
        }
    
    properties = {
        'namespace': metadata.get('namespace'),
        **port_info
    }
    
    # Filter out None values
    properties = {k: v for k, v in properties.items() if v is not None}
    
    return self._create_node('service', name, properties)

def _extract_k8s_env_vars(self, container: Dict[str, Any]) -> Dict[str, str]:
    """Extract environment variables from Kubernetes container spec."""
    env_vars = {}
    env_list = container.get('env', [])
    
    for env_var in env_list:
        name = env_var.get('name')
        value = env_var.get('value')
        
        # Skip secret references for now
        if value and name:
            env_vars[name] = value
    
    return env_vars

def _extract_k8s_service_dependencies(self, env_vars: Dict[str, str]) -> List[str]:
    """Extract service dependencies from Kubernetes environment variables."""
    dependencies = []
    
    for key, value in env_vars.items():
        if key.endswith('_SERVICE_URL') or key.endswith('_URL'):
            # Extract service name from Kubernetes service URL
            # Format: http://service-name.namespace.svc.cluster.local:port
            if 'svc.cluster.local' in value:
                parts = value.split('://')
                if len(parts) > 1:
                    host_part = parts[1].split('.')[0]
                    dependencies.append(host_part)
            elif '://' in value:
                # Fallback for simpler URLs
                url_part = value.split('://')[1]
                if ':' in url_part:
                    service_name = url_part.split(':')[0]
                    dependencies.append(service_name)
    
    return dependencies

from connectors import KubernetesConnector

# Initialize connector
connector = KubernetesConnector()

# Parse Kubernetes YAML file
nodes, edges = connector.parse("./k8s/deployments.yaml")

# Process results
for node in nodes:
    if node.type == 'deployment':
        print(f"Deployment: {node.name}")
        print(f"  Replicas: {node.properties.get('replicas', 1)}")
        print(f"  Image: {node.properties.get('image')}")
        if node.properties.get('team'):
            print(f"  Team: {node.properties['team']}")
    elif node.type == 'service':
        print(f"Service: {node.name}:{node.properties.get('port')}")

for edge in edges:
    print(f"{edge.source} --{edge.type}--> {edge.target}")

apiVersion: apps/v1
kind: Deployment
metadata:
  name: user-service
  namespace: production
  labels:
    app: user-service
    team: identity
spec:
  replicas: 3
  selector:
    matchLabels:
      app: user-service
  template:
    metadata:
      labels:
        app: user-service
    spec:
      containers:
      - name: user-service
        image: user-service:v1.2.3
        resources:
          requests:
            memory: "256Mi"
            cpu: "500m"
          limits:
            memory: "512Mi"
            cpu: "1000m"
        env:
        - name: AUTH_SERVICE_URL
          value: http://auth-service.production.svc.cluster.local:8080
        - name: DATABASE_URL
          valueFrom:
            secretKeyRef:
              name: db-credentials
              key: url
---
apiVersion: v1
kind: Service
metadata:
  name: user-service
  namespace: production
spec:
  selector:
    app: user-service
  ports:
  - port: 8080
    targetPort: 8080
  type: ClusterIP