telecom-platform/apps/api-server/internal/services/kubernetes.go

package services

import (
	"context"
	"fmt"
	"os"
	"path/filepath"
	"strings"
	"time"

	corev1 "k8s.io/api/core/v1"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/labels"
	"k8s.io/client-go/kubernetes"
	"k8s.io/client-go/rest"
	"k8s.io/client-go/tools/clientcmd"
	"k8s.io/client-go/util/homedir"

	"github.com/nutcas3/telecom-platform/apps/api-server/internal/circuitbreaker"
)

// KubernetesService provides real Kubernetes cluster operations using client-go
type KubernetesService struct {
	clientset      *kubernetes.Clientset
	namespace      string
	circuitBreaker *circuitbreaker.CircuitBreaker
}

// NewKubernetesService creates a Kubernetes client using in-cluster config or kubeconfig
func NewKubernetesService() (*KubernetesService, error) {
	var config *rest.Config
	var err error

	// Try in-cluster config first
	if config, err = rest.InClusterConfig(); err != nil {
		// Fall back to kubeconfig
		kubeconfig := filepath.Join(homedir.HomeDir(), ".kube", "config")
		if config, err = clientcmd.BuildConfigFromFlags("", kubeconfig); err != nil {
			return nil, fmt.Errorf("failed to create Kubernetes config: %w", err)
		}
	}

	clientset, err := kubernetes.NewForConfig(config)
	if err != nil {
		return nil, fmt.Errorf("failed to create Kubernetes clientset: %w", err)
	}

	// Determine namespace
	namespace := os.Getenv("KUBERNETES_NAMESPACE")
	if namespace == "" {
		if ns, err := getInClusterNamespace(); err == nil {
			namespace = ns
		} else {
			namespace = "default"
		}
	}

	// Initialize circuit breaker for Kubernetes API calls
	cbConfig := circuitbreaker.Config{
		FailureThreshold: getEnvUint("K8S_CB_FAILURE_THRESHOLD", 5),
		SuccessThreshold: getEnvUint("K8S_CB_SUCCESS_THRESHOLD", 2),
		Timeout:          time.Duration(getEnvUint("K8S_CB_TIMEOUT_SECONDS", 60)) * time.Second,
	}
	circuitBreaker := circuitbreaker.NewCircuitBreaker(cbConfig)

	return &KubernetesService{
		clientset:      clientset,
		namespace:      namespace,
		circuitBreaker: circuitBreaker,
	}, nil
}

func getEnvUint(key string, defaultVal uint) uint {
	val := os.Getenv(key)
	if val == "" {
		return defaultVal
	}
	var result uint
	_, err := fmt.Sscanf(val, "%d", &result)
	if err != nil {
		return defaultVal
	}
	return result
}

// getInClusterNamespace reads the namespace from the in-cluster service account
func getInClusterNamespace() (string, error) {
	data, err := os.ReadFile("/var/run/secrets/kubernetes.io/serviceaccount/namespace")
	if err != nil {
		return "", err
	}
	return string(data), nil
}

// Namespace returns the configured namespace
func (k *KubernetesService) Namespace() string {
	return k.namespace
}

// Deployment wraps Kubernetes Deployment fields we expose via API
type Deployment struct {
	Name                string            `json:"name"`
	Namespace           string            `json:"namespace"`
	CreationTimestamp   time.Time         `json:"creationTimestamp"`
	Labels              map[string]string `json:"labels"`
	Replicas            int32             `json:"replicas"`
	ReadyReplicas       int32             `json:"readyReplicas"`
	AvailableReplicas   int32             `json:"availableReplicas"`
	UpdatedReplicas     int32             `json:"updatedReplicas"`
	UnavailableReplicas int32             `json:"unavailableReplicas"`
}

// ListDeployments returns all deployments in the configured namespace
func (k *KubernetesService) ListDeployments(ctx context.Context) ([]Deployment, error) {
	var result []Deployment
	err := k.circuitBreaker.Execute(func() error {
		deployments, err := k.clientset.AppsV1().Deployments(k.namespace).List(ctx, metav1.ListOptions{})
		if err != nil {
			return fmt.Errorf("failed to list deployments: %w", err)
		}

		for _, d := range deployments.Items {
			result = append(result, Deployment{
				Name:                d.Name,
				Namespace:           d.Namespace,
				CreationTimestamp:   d.CreationTimestamp.Time,
				Labels:              d.Labels,
				Replicas:            *d.Spec.Replicas,
				ReadyReplicas:       d.Status.ReadyReplicas,
				AvailableReplicas:   d.Status.AvailableReplicas,
				UpdatedReplicas:     d.Status.UpdatedReplicas,
				UnavailableReplicas: d.Status.UnavailableReplicas,
			})
		}
		return nil
	})

	if err != nil {
		return nil, err
	}
	return result, nil
}

// GetDeployment returns a specific deployment by name
func (k *KubernetesService) GetDeployment(ctx context.Context, name string) (*Deployment, error) {
	var result *Deployment
	err := k.circuitBreaker.Execute(func() error {
		deployment, err := k.clientset.AppsV1().Deployments(k.namespace).Get(ctx, name, metav1.GetOptions{})
		if err != nil {
			return fmt.Errorf("failed to get deployment %s: %w", name, err)
		}

		result = &Deployment{
			Name:                deployment.Name,
			Namespace:           deployment.Namespace,
			CreationTimestamp:   deployment.CreationTimestamp.Time,
			Labels:              deployment.Labels,
			Replicas:            *deployment.Spec.Replicas,
			ReadyReplicas:       deployment.Status.ReadyReplicas,
			AvailableReplicas:   deployment.Status.AvailableReplicas,
			UpdatedReplicas:     deployment.Status.UpdatedReplicas,
			UnavailableReplicas: deployment.Status.UnavailableReplicas,
		}
		return nil
	})

	if err != nil {
		return nil, err
	}
	return result, nil
}

// ScaleDeployment updates the replica count for a deployment
func (k *KubernetesService) ScaleDeployment(ctx context.Context, name string, replicas int32) error {
	if replicas < 0 || replicas > 100 {
		return fmt.Errorf("replica count %d is out of valid range [0, 100]", replicas)
	}

	return k.circuitBreaker.Execute(func() error {
		// Get current deployment
		deployment, err := k.clientset.AppsV1().Deployments(k.namespace).Get(ctx, name, metav1.GetOptions{})
		if err != nil {
			return fmt.Errorf("failed to get deployment %s: %w", name, err)
		}

		// Update replicas
		deployment.Spec.Replicas = &replicas
		_, err = k.clientset.AppsV1().Deployments(k.namespace).Update(ctx, deployment, metav1.UpdateOptions{})
		if err != nil {
			return fmt.Errorf("failed to scale deployment %s: %w", name, err)
		}
		return nil
	})
}

// RestartDeployment restarts a deployment by rolling its pods
func (k *KubernetesService) RestartDeployment(ctx context.Context, name string) error {
	return k.circuitBreaker.Execute(func() error {
		deployment, err := k.clientset.AppsV1().Deployments(k.namespace).Get(ctx, name, metav1.GetOptions{})
		if err != nil {
			return fmt.Errorf("failed to get deployment %s for restart: %w", name, err)
		}

		// Add restart annotation to trigger rolling restart
		if deployment.Spec.Template.Annotations == nil {
			deployment.Spec.Template.Annotations = make(map[string]string)
		}
		deployment.Spec.Template.Annotations["kubectl.kubernetes.io/restartedAt"] = time.Now().UTC().Format(time.RFC3339)

		_, err = k.clientset.AppsV1().Deployments(k.namespace).Update(ctx, deployment, metav1.UpdateOptions{})
		if err != nil {
			return fmt.Errorf("failed to restart deployment %s: %w", name, err)
		}

		return nil
	})
}

// PodLogs fetches recent logs from pods matching the deployment's label selector
func (k *KubernetesService) PodLogs(ctx context.Context, name string, tailLines int) (string, error) {
	var result string
	err := k.circuitBreaker.Execute(func() error {
		// Get deployment to find label selector
		deployment, err := k.clientset.AppsV1().Deployments(k.namespace).Get(ctx, name, metav1.GetOptions{})
		if err != nil {
			return fmt.Errorf("failed to get deployment %s for logs: %w", name, err)
		}

		// Find pods using deployment's label selector
		labelSelector := labels.Set(deployment.Spec.Selector.MatchLabels).AsSelector()
		pods, err := k.clientset.CoreV1().Pods(k.namespace).List(ctx, metav1.ListOptions{
			LabelSelector: labelSelector.String(),
		})
		if err != nil {
			return fmt.Errorf("failed to list pods for deployment %s: %w", name, err)
		}

		if len(pods.Items) == 0 {
			return fmt.Errorf("no pods found for deployment %s", name)
		}

		// Get logs from the first running pod
		var pod *corev1.Pod
		for _, p := range pods.Items {
			if p.Status.Phase == corev1.PodRunning {
				pod = &p
				break
			}
		}

		if pod == nil {
			return fmt.Errorf("no running pods found for deployment %s", name)
		}

		req := k.clientset.CoreV1().Pods(k.namespace).GetLogs(pod.Name, &corev1.PodLogOptions{
			TailLines: func(i int64) *int64 { return &i }(int64(tailLines)),
		})

		logData, err := req.Stream(ctx)
		if err != nil {
			return fmt.Errorf("failed to stream logs from pod %s: %w", pod.Name, err)
		}
		defer logData.Close()

		buf := make([]byte, 1024)
		var logs strings.Builder
		for {
			n, err := logData.Read(buf)
			if err != nil {
				break
			}
			logs.WriteString(string(buf[:n]))
		}
		result = logs.String()
		return nil
	})

	if err != nil {
		return "", err
	}
	return result, nil
}

// GetPodStatus returns detailed pod status for a deployment
func (k *KubernetesService) GetPodStatus(ctx context.Context, name string) (map[string]any, error) {
	var result map[string]any
	err := k.circuitBreaker.Execute(func() error {
		deployment, err := k.clientset.AppsV1().Deployments(k.namespace).Get(ctx, name, metav1.GetOptions{})
		if err != nil {
			return fmt.Errorf("failed to get deployment %s: %w", name, err)
		}

		labelSelector := labels.Set(deployment.Spec.Selector.MatchLabels).AsSelector()
		pods, err := k.clientset.CoreV1().Pods(k.namespace).List(ctx, metav1.ListOptions{
			LabelSelector: labelSelector.String(),
		})
		if err != nil {
			return fmt.Errorf("failed to list pods for deployment %s: %w", name, err)
		}

		// Build response with deployment replica info and pod details
		podList := make([]map[string]any, 0, len(pods.Items))
		for _, pod := range pods.Items {
			conditions := make([]map[string]string, 0, len(pod.Status.Conditions))
			for _, c := range pod.Status.Conditions {
				conditions = append(conditions, map[string]string{
					"type":   string(c.Type),
					"status": string(c.Status),
					"reason": c.Reason,
				})
			}

			podList = append(podList, map[string]any{
				"name":       pod.Name,
				"phase":      string(pod.Status.Phase),
				"node_name":  pod.Spec.NodeName,
				"created":    pod.CreationTimestamp.Format(time.RFC3339),
				"conditions": conditions,
			})
		}

		result = map[string]any{
			"deployment": map[string]any{
				"replicas":             deployment.Spec.Replicas,
				"ready_replicas":       deployment.Status.ReadyReplicas,
				"available_replicas":   deployment.Status.AvailableReplicas,
				"updated_replicas":     deployment.Status.UpdatedReplicas,
				"unavailable_replicas": deployment.Status.UnavailableReplicas,
			},
			"pods": podList,
		}
		return nil
	})

	if err != nil {
		return nil, err
	}
	return result, nil
}

// GetEvents returns recent events for a deployment
func (k *KubernetesService) GetEvents(ctx context.Context, name string, limit int) ([]map[string]any, error) {
	var result []map[string]any
	err := k.circuitBreaker.Execute(func() error {
		deployment, err := k.clientset.AppsV1().Deployments(k.namespace).Get(ctx, name, metav1.GetOptions{})
		if err != nil {
			return fmt.Errorf("failed to get deployment %s: %w", name, err)
		}

		labelSelector := labels.Set(deployment.Spec.Selector.MatchLabels).AsSelector()
		events, err := k.clientset.CoreV1().Events(k.namespace).List(ctx, metav1.ListOptions{
			LabelSelector: labelSelector.String(),
			Limit:         int64(limit),
		})
		if err != nil {
			return fmt.Errorf("failed to get events for deployment %s: %w", name, err)
		}

		result = make([]map[string]any, 0)
		for _, event := range events.Items {
			result = append(result, map[string]any{
				"type":            event.Type,
				"reason":          event.Reason,
				"message":         event.Message,
				"first_timestamp": event.FirstTimestamp.Time,
				"last_timestamp":  event.LastTimestamp.Time,
				"count":           event.Count,
			})
		}
		return nil
	})

	if err != nil {
		return nil, err
	}
	return result, nil
}