如何从零开始构建Kubernetes Operator

package main

import (
	"fmt"
	"net/http"
	"os"
	"os/signal"
	"sync/atomic"
	"syscall"
)

var requestCount int32

func main() {
	http.HandleFunc("/", func(w http.ResponseWriter, r *http.Request) {
		count := atomic.AddInt32(&requestCount, 1)
		fmt.Fprintf(w, "Hello from the self-destructing app! %d\n", count)
		if count > 5 {
			fmt.Println("Request limit exceeded. Shutting down...")
			go func() {
				pid := os.Getpid()
				syscall.Kill(pid, syscall.SIGTERM)
			}()
		}
	})

	http.HandleFunc("/health", func(w http.ResponseWriter, r *http.Request) {
		w.WriteHeader(http.StatusOK)
		fmt.Fprintln(w, "OK")
	})

	go func() {
		fmt.Println("Server is running on :80")
		if err := http.ListenAndServe(":80", nil); err != nil {
			fmt.Printf("Server error: %v\n", err)
		}
	}()

	// Handle termination signals
	stop := make(chan os.Signal, 1)
	signal.Notify(stop, os.Interrupt, syscall.SIGTERM)
	<-stop
	fmt.Println("Server stopped")
}

package v1

import (
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
)

// PodDiagnoserSpec defines the desired state of PodDiagnoser
type PodDiagnoserSpec struct {
	// TargetLabels specifies which pods to monitor based on labels
	// +optional
	TargetLabels map[string]string `json:"targetLabels,omitempty"`
	// EnableAnnotations determines whether to add diagnostic annotations
	// +optional
	EnableAnnotations bool `json:"enableAnnotations,omitempty"`
	// RemediationAction specifies what action to take when crashes are detected
	// +optional
	RemediationAction string `json:"remediationAction,omitempty"`
}

// PodDiagnoserStatus defines the observed state of PodDiagnoser
type PodDiagnoserStatus struct {
	// LastProcessedPod is the last pod that was processed
	// +optional
	LastProcessedPod string `json:"lastProcessedPod,omitempty"`
	// DiagnosedPods contains the count of pods that have been diagnosed
	// +optional
	DiagnosedPods int `json:"diagnosedPods,omitempty"`
}

//+kubebuilder:object:root=true
//+kubebuilder:subresource:status

// PodDiagnoser is the Schema for the poddiagnosers API
type PodDiagnoser struct {
	metav1.TypeMeta   `json:",inline"`
	metav1.ObjectMeta `json:"metadata,omitempty"`
	Spec              PodDiagnoserSpec   `json:"spec,omitempty"`
	Status            PodDiagnoserStatus `json:"status,omitempty"`
}

//+kubebuilder:object:root=true

// PodDiagnoserList contains a list of PodDiagnoser
type PodDiagnoserList struct {
	metav1.TypeMeta `json:",inline"`
	metav1.ListMeta `json:"metadata,omitempty"`
	Items           []PodDiagnoser `json:"items"`
}

func init() {
	SchemeBuilder.Register(&PodDiagnoser{}, &PodDiagnoserList{})
}

package controller
 
import (
    "context"
    "fmt"
    "time"
 
    corev1 "k8s.io/api/core/v1"
    "k8s.io/apimachinery/pkg/api/errors"
    "k8s.io/apimachinery/pkg/runtime"
    "k8s.io/client-go/tools/record"
    "k8s.io/client-go/util/retry"
    ctrl "sigs.k8s.io/controller-runtime"
    "sigs.k8s.io/controller-runtime/pkg/client"
    "sigs.k8s.io/controller-runtime/pkg/log"
 
    diagnosticsv1 "local.io/pod-diagnoser/api/v1"
)
 
// PodDiagnoserReconciler reconciles a PodDiagnoser object
type PodDiagnoserReconciler struct {
    client.Client
    Scheme *runtime.Scheme
    Recorder record.EventRecorder // Add event recorder
}
 
//+kubebuilder:rbac:groups=diagnostics.local.io,resources=poddiagnosers,verbs=get;list;watch;create;update;patch;delete
//+kubebuilder:rbac:groups=diagnostics.local.io,resources=poddiagnosers/status,verbs=get;update;patch
//+kubebuilder:rbac:groups=diagnostics.local.io,resources=poddiagnosers/finalizers,verbs=update
//+kubebuilder:rbac:groups="",resources=pods,verbs=get;list;watch;update;patch
//+kubebuilder:rbac:groups="",resources=events,verbs=create;patch
 
// Reconcile is part of the main kubernetes reconciliation loop
func (r *PodDiagnoserReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {
    logger := log.FromContext(ctx)
    
    // Get the pod that triggered reconciliation
    pod := &corev1.Pod{}
    if err := r.Get(ctx, req.NamespacedName, pod); err != nil {
        if errors.IsNotFound(err) {
            // Pod was deleted, nothing to do
            return ctrl.Result{}, nil
        }
        logger.Error(err, "Unable to fetch Pod")
        return ctrl.Result{}, err
    }
 
    // Get all PodDiagnosers
    diagnoserList := &diagnosticsv1.PodDiagnoserList{}
    if err := r.List(ctx, diagnoserList); err != nil {
        logger.Error(err, "Unable to list PodDiagnosers")
        return ctrl.Result{}, err
    }
 
    // Check if this pod should be diagnosed based on PodDiagnoser rules
    var matchingDiagnoser *diagnosticsv1.PodDiagnoser
    for i := range diagnoserList.Items {
        diagnoser := &diagnoserList.Items[i]
        if shouldDiagnosePod(diagnoser, pod) {
            matchingDiagnoser = diagnoser
            break
        }
    }
 
    if matchingDiagnoser == nil {
        // No matching diagnoser for this pod
        return ctrl.Result{}, nil
    }
 
    // Check for container restarts and diagnose
    updated := false
    for _, containerStatus := range pod.Status.ContainerStatuses {
        if containerStatus.RestartCount > 0 && containerStatus.LastTerminationState.Terminated != nil {
            // Container has restarted at least once
            if pod.Annotations == nil {
                pod.Annotations = make(map[string]string)
            }
 
            // Add diagnostic annotations
            restartReason := containerStatus.LastTerminationState.Terminated.Reason
            exitCode := containerStatus.LastTerminationState.Terminated.ExitCode
            restartTime := containerStatus.LastTerminationState.Terminated.FinishedAt.Time
            
            pod.Annotations["diagnostics.local.io/restart-reason"] = 
                fmt.Sprintf("%s (Exit Code: %d)", restartReason, exitCode)
            pod.Annotations["diagnostics.local.io/restart-time"] = 
                restartTime.Format(time.RFC3339)
            
            updated = true
            
            logger.Info("Diagnosed pod restart",
                "pod", pod.Name,
                "container", containerStatus.Name,
                "reason", restartReason,
                "exitCode", exitCode)
        }
    }
 
    if updated {
        // Update the pod with new annotations using retry for robustness
        if err := retry.RetryOnConflict(retry.DefaultRetry, func() error {
            // Get the latest version to avoid conflicts
            latest := &corev1.Pod{}
            if err := r.Get(ctx, req.NamespacedName, latest); err != nil {
                return err
            }
            
            // Update annotations
            if latest.Annotations == nil {
                latest.Annotations = make(map[string]string)
            }
            latest.Annotations["diagnostics.local.io/restart-reason"] = 
                pod.Annotations["diagnostics.local.io/restart-reason"]
            latest.Annotations["diagnostics.local.io/restart-time"] = 
                pod.Annotations["diagnostics.local.io/restart-time"]
                
            return r.Update(ctx, latest)
        }); err != nil {
            logger.Error(err, "Failed to update Pod with diagnostic annotations")
            return ctrl.Result{}, err
        }
        
        // Update the diagnoser status with retry for robustness
        if err := retry.RetryOnConflict(retry.DefaultRetry, func() error {
            // Get the latest version
            latest := &diagnosticsv1.PodDiagnoser{}
            if err := r.Get(ctx, client.ObjectKey{
                Namespace: matchingDiagnoser.Namespace,
                Name:      matchingDiagnoser.Name,
            }, latest); err != nil {
                return err
            }
            
            latest.Status.LastProcessedPod = pod.Name
            latest.Status.DiagnosedPods++
            return r.Status().Update(ctx, latest)
        }); err != nil {
            logger.Error(err, "Unable to update PodDiagnoser status")
            return ctrl.Result{}, err
        }
        
        // Apply remediation if configured
        if matchingDiagnoser.Spec.RemediationAction != "" {
            if err := r.applyRemediation(ctx, matchingDiagnoser, pod); err != nil {
                logger.Error(err, "Failed to apply remediation")
                return ctrl.Result{}, err
            }
        }
    }
 
    return ctrl.Result{}, nil
}
 
// shouldDiagnosePod determines if this pod should be diagnosed by the given diagnoser
func shouldDiagnosePod(diagnoser *diagnosticsv1.PodDiagnoser, pod *corev1.Pod) bool {
    if !diagnoser.Spec.EnableAnnotations {
        return false
    }
    
    // If no target labels specified, match all pods
    if len(diagnoser.Spec.TargetLabels) == 0 {
        return true
    }
    
    // Check if pod matches all target labels
    for key, value := range diagnoser.Spec.TargetLabels {
        if pod.Labels[key] != value {
            return false
        }
    }
    
    return true
}
 
// applyRemediation applies remediation based on the diagnoser configuration
func (r *PodDiagnoserReconciler) applyRemediation(ctx context.Context, 
    diagnoser *diagnosticsv1.PodDiagnoser, pod *corev1.Pod) error {
    
    logger := log.FromContext(ctx)
    
    switch diagnoser.Spec.RemediationAction {
    case "RestartPod":
        logger.Info("Applying remediation: restarting pod", "pod", pod.Name)
        // Just delete the pod, the deployment controller will create a new one
        return r.Delete(ctx, pod)
    
    case "LogEvent":
        // Use the Kubernetes event recorder to log an event
        r.Recorder.Event(pod, 
            corev1.EventTypeWarning, 
            "PodRestarted", 
            fmt.Sprintf("Pod %s restarted due to %s (Exit Code: %d)",
                pod.Name,
                pod.Annotations["diagnostics.local.io/restart-reason"],
                pod.Status.ContainerStatuses[0].LastTerminationState.Terminated.ExitCode))
        return nil
        
    default:
        logger.Info("No remediation action defined or unknown action", 
            "action", diagnoser.Spec.RemediationAction)
        return nil
    }
}
 
// SetupWithManager sets up the controller with the Manager.
func (r *PodDiagnoserReconciler) SetupWithManager(mgr ctrl.Manager) error {
    // Initialize the event recorder
    r.Recorder = mgr.GetEventRecorderFor("pod-diagnoser")
    
    return ctrl.NewControllerManagedBy(mgr).
        For(&corev1.Pod{}).
        Complete(r)
}