feat: Add Kubernetes service with deployment management, scaling, restart, logs, and pod status monitoring

- Add KubernetesService struct with kubernetes.Clientset/namespace and NewKubernetesService constructor with in-cluster/kubeconfig fallback - Add getInClusterNamespace helper with /var/run/secrets/kubernetes.io/serviceaccount/namespace reading - Add Deployment struct with name/namespace/creationTimestamp/labels/replicas/readyReplicas/availableReplicas/updatedReplicas/unavailableReplicas fields - Add ListDeployments method with AppsV1().Deployments().
2026-10-18 11:02:54 +00:00 · 2026-04-20 20:05:23 +03:00
parent 501c558b9f
commit 4875fa7c88
1 changed files with 306 additions and 0 deletions
@@ -0,0 +1,306 @@
+package services
+
+import (
+	"context"
+	"fmt"
+	"os"
+	"path/filepath"
+	"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"
+)
+
+// KubernetesService provides real Kubernetes cluster operations using client-go
+type KubernetesService struct {
+	clientset *kubernetes.Clientset
+	namespace string
+}
+
+// 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"
+		}
+	}
+
+	return &KubernetesService{
+		clientset: clientset,
+		namespace: namespace,
+	}, nil
+}
+
+// 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) {
+	deployments, err := k.clientset.AppsV1().Deployments(k.namespace).List(ctx, metav1.ListOptions{})
+	if err != nil {
+		return nil, fmt.Errorf("failed to list deployments: %w", err)
+	}
+
+	var result []Deployment
+	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 result, nil
+}
+
+// GetDeployment returns a specific deployment by name
+func (k *KubernetesService) GetDeployment(ctx context.Context, name string) (*Deployment, error) {
+	deployment, err := k.clientset.AppsV1().Deployments(k.namespace).Get(ctx, name, metav1.GetOptions{})
+	if err != nil {
+		return nil, fmt.Errorf("failed to get deployment %s: %w", name, err)
+	}
+
+	return &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,
+	}, 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)
+	}
+
+	// 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 for scaling: %w", name, err)
+	}
+
+	// Update replica count
+	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 to %d replicas: %w", name, replicas, err)
+	}
+
+	return nil
+}
+
+// RestartDeployment triggers a rolling restart by updating the pod template annotation
+func (k *KubernetesService) RestartDeployment(ctx context.Context, name string) 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) {
+	// 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 {
+		// If no running pod, use the first one
+		pod = &pods.Items[0]
+	}
+
+	if tailLines <= 0 {
+		tailLines = 100
+	}
+
+	tailLines64 := int64(tailLines)
+	req := k.clientset.CoreV1().Pods(k.namespace).GetLogs(pod.Name, &corev1.PodLogOptions{
+		TailLines: &tailLines64,
+		Follow:    false,
+	})
+
+	logs, err := req.Stream(ctx)
+	if err != nil {
+		return "", fmt.Errorf("failed to get logs for pod %s: %w", pod.Name, err)
+	}
+	defer logs.Close()
+
+	data := make([]byte, 1024*1024) // 1MB limit
+	n, err := logs.Read(data)
+	if err != nil && err.Error() != "EOF" {
+		return "", fmt.Errorf("failed to read logs: %w", err)
+	}
+
+	return string(data[:n]), nil
+}
+
+// GetPodStatus returns detailed pod status for a deployment
+func (k *KubernetesService) GetPodStatus(ctx context.Context, name string) (map[string]any, error) {
+	deployment, err := k.clientset.AppsV1().Deployments(k.namespace).Get(ctx, name, metav1.GetOptions{})
+	if err != nil {
+		return nil, 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 nil, fmt.Errorf("failed to list pods for deployment %s: %w", name, err)
+	}
+
+	status := 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": make([]map[string]any, 0),
+	}
+
+	for _, pod := range pods.Items {
+		podStatus := map[string]any{
+			"name":       pod.Name,
+			"phase":      string(pod.Status.Phase),
+			"node_name":  pod.Spec.NodeName,
+			"created":    pod.CreationTimestamp.Time,
+			"conditions": make([]map[string]any, 0),
+		}
+
+		for _, condition := range pod.Status.Conditions {
+			podStatus["conditions"] = append(podStatus["conditions"].([]map[string]any), map[string]any{
+				"type":   string(condition.Type),
+				"status": string(condition.Status),
+				"reason": condition.Reason,
+			})
+		}
+
+		status["pods"] = append(status["pods"].([]map[string]any), podStatus)
+	}
+
+	return status, nil
+}
+
+// GetEvents returns recent events for a deployment
+func (k *KubernetesService) GetEvents(ctx context.Context, name string, limit int) ([]map[string]any, error) {
+	events, err := k.clientset.CoreV1().Events(k.namespace).List(ctx, metav1.ListOptions{
+		FieldSelector: fmt.Sprintf("involvedObject.name=%s,involvedObject.kind=Deployment", name),
+		Limit:         int64(limit),
+	})
+	if err != nil {
+		return nil, 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 result, nil
+}