diff --git a/apps/api-server/internal/services/kubernetes.go b/apps/api-server/internal/services/kubernetes.go index f1a354f..3927afd 100644 --- a/apps/api-server/internal/services/kubernetes.go +++ b/apps/api-server/internal/services/kubernetes.go @@ -5,6 +5,7 @@ import ( "fmt" "os" "path/filepath" + "strings" "time" corev1 "k8s.io/api/core/v1" @@ -111,47 +112,62 @@ type Deployment struct { // 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, - }) - } + 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) { - 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) - } + 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) + } - 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 + 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 @@ -160,172 +176,202 @@ func (k *KubernetesService) ScaleDeployment(ctx context.Context, name string, re 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) - } + 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 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 + // 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 triggers a rolling restart by updating the pod template annotation +// RestartDeployment restarts a deployment by rolling its pods 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) - } + 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) + // 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) - } + _, 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 + 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 + 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) } - } - if pod == nil { - // If no running pod, use the first one - pod = &pods.Items[0] - } - if tailLines <= 0 { - tailLines = 100 - } + // 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) + } - tailLines64 := int64(tailLines) - req := k.clientset.CoreV1().Pods(k.namespace).GetLogs(pod.Name, &corev1.PodLogOptions{ - TailLines: &tailLines64, - Follow: false, + 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 }) - logs, err := req.Stream(ctx) if err != nil { - return "", fmt.Errorf("failed to get logs for pod %s: %w", pod.Name, err) + return "", 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 + return result, 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), + 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) } - 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, + 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, }) } - status["pods"] = append(status["pods"].([]map[string]any), podStatus) - } + 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 + }) - return status, 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) { - 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) - } + 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) + } - 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, + 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 }