mirror of
https://github.com/andwati/telecom-platform.git
synced 2026-10-21 04:04:28 +00:00
feat: Wrap all Kubernetes API operations with circuit breaker protection and improve error handling
- Add strings import to kubernetes.go - Wrap ListDeployments with circuitBreaker.Execute and move result building inside closure - Wrap GetDeployment with circuitBreaker.Execute and move result assignment inside closure - Wrap ScaleDeployment with circuitBreaker.Execute and simplify error messages - Wrap RestartDeployment with circuitBreaker.Execute - Wrap PodLogs with circuitBreaker.Execute and add streaming
This commit is contained in:
nutcas3
@@ -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
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user