helm 入门与应用：Kubernetes 包管理器

package config

import (
	"sync"
	"time"
)

type RouteRule struct {
	Path      string   `json:"path"`
	Methods   []string `json:"methods"`
	Backend   string   `json:"backend"`
	RateLimit struct {
		QPS   int `json:"qps"`
		Burst int `json:"burst"`
	} `json:"rateLimit"`
}

type GatewaySpec struct {
	GlobalRateLimit struct {
		QPS   int `json:"qps"`
		Burst int `json:"burst"`
	} `json:"globalRateLimit"`
	Routes []RouteRule `json:"routes"`
}

type GatewayStatus struct {
	ActiveRoutes   int       `json:"activeRoutes"`
	LastReloadTime time.Time `json:"lastReloadTime"`
}

// Store 使用 RWMutex 保护的配置存储，支持并发安全的读写
type Store struct {
	mu     sync.RWMutex
	spec   *GatewaySpec
	status *GatewayStatus
}

func NewStore() *Store {
	return &Store{
		spec:   &GatewaySpec{},
		status: &GatewayStatus{},
	}
}

func (s *Store) UpdateSpec(spec *GatewaySpec) {
	s.mu.Lock()
	defer s.mu.Unlock()
	s.spec = spec
	s.status = &GatewayStatus{
		ActiveRoutes:   len(spec.Routes),
		LastReloadTime: time.Now(),
	}
}

func (s *Store) GetSpec() *GatewaySpec {
	s.mu.RLock()
	defer s.mu.RUnlock()
	return s.spec
}

func (s *Store) GetStatus() *GatewayStatus {
	s.mu.RLock()
	defer s.mu.RUnlock()
	return s.status
}

package watcher

import (
	"context"
	"encoding/json"
	"fmt"
	"log"
	"time"

	"gin-gateway-clientgo/config"

	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
	"k8s.io/apimachinery/pkg/runtime/schema"
	"k8s.io/client-go/dynamic"
	"k8s.io/client-go/dynamic/dynamicinformer"
	"k8s.io/client-go/rest"
	"k8s.io/client-go/tools/cache"
	"k8s.io/client-go/tools/clientcmd"
)

var gvr = schema.GroupVersionResource{
	Group:    "gateway.example.com",
	Version:  "v1alpha1",
	Resource: "gatewayconfigs",
}

type Watcher struct {
	client    dynamic.Interface
	store     *config.Store
	namespace string
}

func NewWatcher(store *config.Store, namespace string) (*Watcher, error) {
	cfg, err := rest.InClusterConfig()
	if err != nil {
		// 本地开发时 fallback 到 kubeconfig
		cfg, err = clientcmd.BuildConfigFromFlags("", clientcmd.RecommendedHomeFile)
		if err != nil {
			return nil, fmt.Errorf("failed to build k8s config: %w", err)
		}
	}

	client, err := dynamic.NewForConfig(cfg)
	if err != nil {
		return nil, fmt.Errorf("failed to create dynamic client: %w", err)
	}

	return &Watcher{
		client:    client,
		store:     store,
		namespace: namespace,
	}, nil
}

func (w *Watcher) Start(ctx context.Context) error {
	factory := dynamicinformer.NewFilteredDynamicSharedInformerFactory(
		w.client,
		30*time.Second, // resync 周期
		w.namespace,
		func(opts *metav1.ListOptions) {},
	)

	informer := factory.ForResource(gvr).Informer()

	informer.AddEventHandler(cache.ResourceEventHandlerFuncs{
		AddFunc: func(obj interface{}) {
			w.handleEvent("ADD", obj)
		},
		UpdateFunc: func(oldObj, newObj interface{}) {
			w.handleEvent("UPDATE", newObj)
		},
		DeleteFunc: func(obj interface{}) {
			log.Println("[Watcher] GatewayConfig deleted, clearing config")
			w.store.UpdateSpec(&config.GatewaySpec{})
		},
	})

	log.Printf("[Watcher] Starting informer for %s in namespace %s", gvr.Resource, w.namespace)
	informer.Run(ctx.Done())
	return nil
}

func (w *Watcher) handleEvent(eventType string, obj interface{}) {
	unstr, ok := obj.(*unstructured.Unstructured)
	if !ok {
		log.Printf("[Watcher] unexpected object type: %T", obj)
		return
	}

	specRaw, found, err := unstructured.NestedMap(unstr.Object, "spec")
	if err != nil || !found {
		log.Printf("[Watcher] failed to extract spec: %v", err)
		return
	}

	specJSON, err := json.Marshal(specRaw)
	if err != nil {
		log.Printf("[Watcher] failed to marshal spec: %v", err)
		return
	}

	var spec config.GatewaySpec
	if err := json.Unmarshal(specJSON, &spec); err != nil {
		log.Printf("[Watcher] failed to unmarshal spec: %v", err)
		return
	}

	w.store.UpdateSpec(&spec)
	log.Printf("[Watcher] %s event: loaded %d routes from %s",
		eventType, len(spec.Routes), unstr.GetName())
}

package gateway

import (
	"net/http"
	"net/http/httputil"
	"net/url"
	"strings"
	"sync"

	"gin-gateway-clientgo/config"

	"github.com/gin-gonic/gin"
	"golang.org/x/time/rate"
)

type Handler struct {
	store    *config.Store
	limiters sync.Map // path -> *rate.Limiter
}

func NewHandler(store *config.Store) *Handler {
	return &Handler{store: store}
}

func (h *Handler) RegisterRoutes(r *gin.Engine) {
	r.GET("/healthz", func(c *gin.Context) {
		c.JSON(http.StatusOK, gin.H{"status": "ok"})
	})

	r.GET("/config", func(c *gin.Context) {
		c.JSON(http.StatusOK, gin.H{
			"spec":   h.store.GetSpec(),
			"status": h.store.GetStatus(),
		})
	})

	// 通配路由：所有 /api/* 请求走网关代理逻辑
	r.NoRoute(h.proxyHandler)
}

func (h *Handler) proxyHandler(c *gin.Context) {
	spec := h.store.GetSpec()

	// 匹配路由规则
	for _, route := range spec.Routes {
		if !strings.HasPrefix(c.Request.URL.Path, route.Path) {
			continue
		}

		// 检查 HTTP 方法
		if len(route.Methods) > 0 && !containsMethod(route.Methods, c.Request.Method) {
			continue
		}

		// 限流检查
		if route.RateLimit.QPS > 0 {
			limiter := h.getOrCreateLimiter(route.Path, route.RateLimit.QPS, route.RateLimit.Burst)
			if !limiter.Allow() {
				c.JSON(http.StatusTooManyRequests, gin.H{
					"error": "rate limit exceeded",
					"path":  route.Path,
				})
				return
			}
		}

		// 代理转发
		target, err := url.Parse(route.Backend)
		if err != nil {
			c.JSON(http.StatusBadGateway, gin.H{"error": "invalid backend"})
			return
		}

		proxy := httputil.NewSingleHostReverseProxy(target)
		proxy.ServeHTTP(c.Writer, c.Request)
		return
	}

	c.JSON(http.StatusNotFound, gin.H{"error": "no matching route"})
}

func (h *Handler) getOrCreateLimiter(path string, qps, burst int) *rate.Limiter {
	if v, ok := h.limiters.Load(path); ok {
		return v.(*rate.Limiter)
	}
	limiter := rate.NewLimiter(rate.Limit(qps), burst)
	h.limiters.Store(path, limiter)
	return limiter
}

func containsMethod(methods []string, method string) bool {
	for _, m := range methods {
		if strings.EqualFold(m, method) {
			return true
		}
	}
	return false
}

package main

import (
	"context"
	"log"
	"os"
	"os/signal"
	"syscall"

	"gin-gateway-clientgo/config"
	"gin-gateway-clientgo/gateway"
	"gin-gateway-clientgo/watcher"

	"github.com/gin-gonic/gin"
)

func main() {
	namespace := os.Getenv("WATCH_NAMESPACE")
	if namespace == "" {
		namespace = "gateway"
	}

	store := config.NewStore()

	w, err := watcher.NewWatcher(store, namespace)
	if err != nil {
		log.Fatalf("Failed to create watcher: %v", err)
	}

	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()

	// 启动 informer（后台 goroutine）
	go func() {
		if err := w.Start(ctx); err != nil {
			log.Fatalf("Watcher failed: %v", err)
		}
	}()

	r := gin.Default()
	handler := gateway.NewHandler(store)
	handler.RegisterRoutes(r)

	// 优雅退出
	go func() {
		sigCh := make(chan os.Signal, 1)
		signal.Notify(sigCh, syscall.SIGINT, syscall.SIGTERM)
		<-sigCh
		log.Println("Shutting down...")
		cancel()
	}()

	port := os.Getenv("PORT")
	if port == "" {
		port = "8080"
	}
	log.Printf("Starting gin-gateway on :%s", port)
	if err := r.Run(":" + port); err != nil {
		log.Fatalf("Server failed: %v", err)
	}
}

package v1alpha1

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

type RouteRule struct {
	Path    string   `json:"path"`
	Methods []string `json:"methods,omitempty"`
	Backend string   `json:"backend"`

	RateLimit *RateLimitConfig `json:"rateLimit,omitempty"`
}

type RateLimitConfig struct {
	QPS   int `json:"qps"`
	Burst int `json:"burst,omitempty"`
}

type GatewayConfigSpec struct {
	GlobalRateLimit *RateLimitConfig `json:"globalRateLimit,omitempty"`
	Routes          []RouteRule      `json:"routes"`
}

type GatewayConfigStatus struct {
	ActiveRoutes   int    `json:"activeRoutes,omitempty"`
	LastReloadTime string `json:"lastReloadTime,omitempty"`
}

// +kubebuilder:object:root=true
// +kubebuilder:subresource:status
// +kubebuilder:printcolumn:name="Routes",type=integer,JSONPath=`.status.activeRoutes`
// +kubebuilder:printcolumn:name="LastReload",type=string,JSONPath=`.status.lastReloadTime`
// +kubebuilder:printcolumn:name="Age",type=date,JSONPath=`.metadata.creationTimestamp`

type GatewayConfig struct {
	metav1.TypeMeta   `json:",inline"`
	metav1.ObjectMeta `json:"metadata,omitempty"`

	Spec   GatewayConfigSpec   `json:"spec,omitempty"`
	Status GatewayConfigStatus `json:"status,omitempty"`
}

// +kubebuilder:object:root=true

type GatewayConfigList struct {
	metav1.TypeMeta `json:",inline"`
	metav1.ListMeta `json:"metadata,omitempty"`
	Items           []GatewayConfig `json:"items"`
}

func init() {
	SchemeBuilder.Register(&GatewayConfig{}, &GatewayConfigList{})
}

// Package v1alpha1 contains API Schema definitions for the gateway v1alpha1 API group.
// +kubebuilder:object:generate=true
// +groupName=gateway.example.com
package v1alpha1

import (
	"k8s.io/apimachinery/pkg/runtime/schema"
	"sigs.k8s.io/controller-runtime/pkg/scheme"
)

var (
	GroupVersion = schema.GroupVersion{
		Group:   "gateway.example.com",
		Version: "v1alpha1",
	}

	SchemeBuilder = &scheme.Builder{GroupVersion: GroupVersion}

	AddToScheme = SchemeBuilder.AddToScheme
)

package controller

import (
	"context"
	"fmt"
	"time"

	"k8s.io/apimachinery/pkg/runtime"
	ctrl "sigs.k8s.io/controller-runtime"
	"sigs.k8s.io/controller-runtime/pkg/builder"
	"sigs.k8s.io/controller-runtime/pkg/client"
	"sigs.k8s.io/controller-runtime/pkg/log"
	"sigs.k8s.io/controller-runtime/pkg/predicate"

	gatewayv1alpha1 "gin-gateway-operator/api/v1alpha1"
	"gin-gateway-operator/internal/gwconfig"
)

type GatewayConfigReconciler struct {
	client.Client
	Scheme *runtime.Scheme
	Store  *gwconfig.Store
}

// +kubebuilder:rbac:groups=gateway.example.com,resources=gatewayconfigs,verbs=get;list;watch
// +kubebuilder:rbac:groups=gateway.example.com,resources=gatewayconfigs/status,verbs=get;update;patch

func (r *GatewayConfigReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {
	logger := log.FromContext(ctx)

	var gwConfig gatewayv1alpha1.GatewayConfig
	if err := r.Get(ctx, req.NamespacedName, &gwConfig); err != nil {
		if client.IgnoreNotFound(err) == nil {
			// CR 被删除，清空配置
			logger.Info("GatewayConfig deleted, clearing config")
			r.Store.UpdateSpec(&gwconfig.GatewaySpec{})
			return ctrl.Result{}, nil
		}
		return ctrl.Result{}, fmt.Errorf("failed to get GatewayConfig: %w", err)
	}

	// 将 CR spec 转换为内部配置格式
	spec := convertSpec(&gwConfig.Spec)
	r.Store.UpdateSpec(spec)

	logger.Info("GatewayConfig reconciled",
		"name", req.Name,
		"routes", len(spec.Routes))

	// 更新 CR status
	gwConfig.Status.ActiveRoutes = len(spec.Routes)
	gwConfig.Status.LastReloadTime = time.Now().Format(time.RFC3339)
	if err := r.Status().Update(ctx, &gwConfig); err != nil {
		logger.Error(err, "failed to update status")
		return ctrl.Result{}, err
	}

	return ctrl.Result{}, nil
}

func convertSpec(crSpec *gatewayv1alpha1.GatewayConfigSpec) *gwconfig.GatewaySpec {
	spec := &gwconfig.GatewaySpec{}

	if crSpec.GlobalRateLimit != nil {
		spec.GlobalRateLimit.QPS = crSpec.GlobalRateLimit.QPS
		spec.GlobalRateLimit.Burst = crSpec.GlobalRateLimit.Burst
	}

	for _, r := range crSpec.Routes {
		route := gwconfig.RouteRule{
			Path:    r.Path,
			Methods: r.Methods,
			Backend: r.Backend,
		}
		if r.RateLimit != nil {
			route.RateLimit.QPS = r.RateLimit.QPS
			route.RateLimit.Burst = r.RateLimit.Burst
		}
		spec.Routes = append(spec.Routes, route)
	}

	return spec
}

func (r *GatewayConfigReconciler) SetupWithManager(mgr ctrl.Manager) error {
	// Reconcile 每次都会写入 lastReloadTime，若不过滤 status 变更会导致无限调和
	return ctrl.NewControllerManagedBy(mgr).
		For(&gatewayv1alpha1.GatewayConfig{},
			builder.WithPredicates(predicate.GenerationChangedPredicate{})).
		Complete(r)
}

package main

import (
	"context"
	"log"
	"net/http"
	"os"
	"os/signal"
	"syscall"
	"time"

	"github.com/gin-gonic/gin"
	"k8s.io/apimachinery/pkg/runtime"
	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
	clientgoscheme "k8s.io/client-go/kubernetes/scheme"
	ctrl "sigs.k8s.io/controller-runtime"
	"sigs.k8s.io/controller-runtime/pkg/healthz"
	ctrllog "sigs.k8s.io/controller-runtime/pkg/log"
	"sigs.k8s.io/controller-runtime/pkg/log/zap"

	gatewayv1alpha1 "gin-gateway-operator/api/v1alpha1"
	controller "gin-gateway-operator/internal/controller"
	"gin-gateway-operator/internal/gateway"
	"gin-gateway-operator/internal/gwconfig"
)

var sch = runtime.NewScheme()

func init() {
	utilruntime.Must(clientgoscheme.AddToScheme(sch))
	utilruntime.Must(gatewayv1alpha1.AddToScheme(sch))
}

func main() {
	ctrllog.SetLogger(zap.New(zap.UseDevMode(true)))

	namespace := os.Getenv("WATCH_NAMESPACE")

	store := gwconfig.NewStore()

	// 创建 controller-manager
	mgr, err := ctrl.NewManager(ctrl.GetConfigOrDie(), ctrl.Options{
		Scheme:                 sch,
		HealthProbeBindAddress: ":8081",
		LeaderElection:         false,
	})
	if err != nil {
		log.Fatalf("Failed to create manager: %v", err)
	}

	// 注册 Reconciler
	if err := (&controller.GatewayConfigReconciler{
		Client: mgr.GetClient(),
		Scheme: mgr.GetScheme(),
		Store:  store,
	}).SetupWithManager(mgr); err != nil {
		log.Fatalf("Failed to setup controller: %v", err)
	}

	if err := mgr.AddHealthzCheck("healthz", healthz.Ping); err != nil {
		log.Fatalf("Failed to set up health check: %v", err)
	}

	// 启动 Gin 服务（独立 goroutine）
	port := os.Getenv("PORT")
	if port == "" {
		port = "8080"
	}

	r := gin.Default()
	handler := gateway.NewHandler(store)
	handler.RegisterRoutes(r)

	srv := &http.Server{
		Addr:    ":" + port,
		Handler: r,
	}

	go func() {
		log.Printf("Starting gin-gateway on :%s", port)
		if err := srv.ListenAndServe(); err != nil && err != http.ErrServerClosed {
			log.Fatalf("Gin server failed: %v", err)
		}
	}()

	// 启动 controller-manager（阻塞）
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()

	go func() {
		sigCh := make(chan os.Signal, 1)
		signal.Notify(sigCh, syscall.SIGINT, syscall.SIGTERM)
		<-sigCh
		log.Println("Shutting down...")
		cancel()
		shutdownCtx, shutdownCancel := context.WithTimeout(context.Background(), 5*time.Second)
		defer shutdownCancel()
		_ = srv.Shutdown(shutdownCtx)
	}()

	_ = namespace // 如需限定 namespace，在 Manager Options 中设置 Cache.DefaultNamespaces
	log.Println("Starting controller manager")
	if err := mgr.Start(ctx); err != nil {
		log.Fatalf("Manager failed: %v", err)
	}
}

package v1alpha1

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

type RateLimitConfig struct {
	// +kubebuilder:validation:Minimum=1
	QPS int32 `json:"qps"`
	// +optional
	Burst int32 `json:"burst,omitempty"`
}

type RouteRule struct {
	// +kubebuilder:validation:Pattern=`^/`
	Path string `json:"path"`
	// +optional
	Methods []string `json:"methods,omitempty"`
	// +kubebuilder:validation:MinLength=1
	Backend string `json:"backend"`
	// +optional
	RateLimit *RateLimitConfig `json:"rateLimit,omitempty"`
}

// GatewayWorkload 描述由 Operator 创建的 gin-gateway Deployment/Service
type GatewayWorkload struct {
	// +kubebuilder:validation:MinLength=1
	Image string `json:"image"`
	// +kubebuilder:default=2
	// +kubebuilder:validation:Minimum=0
	// +optional
	Replicas *int32 `json:"replicas,omitempty"`
	// +kubebuilder:default=8080
	// +optional
	Port int32 `json:"port,omitempty"`
	// +kubebuilder:default=ClusterIP
	// +kubebuilder:validation:Enum=ClusterIP;NodePort;LoadBalancer
	// +optional
	ServiceType corev1.ServiceType `json:"serviceType,omitempty"`
	// +optional
	Resources corev1.ResourceRequirements `json:"resources,omitempty"`
}

type GatewayConfigSpec struct {
	Gateway GatewayWorkload `json:"gateway"`
	// +optional
	GlobalRateLimit *RateLimitConfig `json:"globalRateLimit,omitempty"`
	// +kubebuilder:validation:MinItems=1
	Routes []RouteRule `json:"routes"`
}

type GatewayConfigStatus struct {
	// +optional
	ObservedGeneration int64 `json:"observedGeneration,omitempty"`
	// +optional
	ActiveRoutes int32 `json:"activeRoutes,omitempty"`
	// +optional
	ReadyReplicas int32 `json:"readyReplicas,omitempty"`
	// +optional
	ConfigHash string `json:"configHash,omitempty"`
	// +optional
	LastRenderTime string `json:"lastRenderTime,omitempty"`
	// +listType=map
	// +listMapKey=type
	// +optional
	Conditions []metav1.Condition `json:"conditions,omitempty"`
}

// +kubebuilder:object:root=true
// +kubebuilder:subresource:status
// +kubebuilder:resource:shortName=gwc
// +kubebuilder:printcolumn:name="Routes",type=integer,JSONPath=`.status.activeRoutes`
// +kubebuilder:printcolumn:name="Ready",type=integer,JSONPath=`.status.readyReplicas`
// +kubebuilder:printcolumn:name="Hash",type=string,JSONPath=`.status.configHash`
// +kubebuilder:printcolumn:name="Age",type=date,JSONPath=`.metadata.creationTimestamp`

type GatewayConfig struct {
	metav1.TypeMeta   `json:",inline"`
	metav1.ObjectMeta `json:"metadata,omitempty"`

	Spec   GatewayConfigSpec   `json:"spec,omitempty"`
	Status GatewayConfigStatus `json:"status,omitempty"`
}

// +kubebuilder:object:root=true

type GatewayConfigList struct {
	metav1.TypeMeta `json:",inline"`
	metav1.ListMeta `json:"metadata,omitempty"`
	Items           []GatewayConfig `json:"items"`	//slice
}

func init() {
	SchemeBuilder.Register(&GatewayConfig{}, &GatewayConfigList{})
}

package controller

import (
	"crypto/sha256"
	"encoding/hex"
	"encoding/json"
	"fmt"
	"net/url"
	"sort"

	gatewayv1alpha1 "gin-gateway-split/api/v1alpha1"
	"gin-gateway-split/internal/gwconfig"
)

// renderConfig 返回 routes.json 内容及其短 hash
func renderConfig(spec *gatewayv1alpha1.GatewayConfigSpec) (string, string, error) {
	out := gwconfig.GatewaySpec{}
	if spec.GlobalRateLimit != nil {
		out.GlobalRateLimit.QPS = int(spec.GlobalRateLimit.QPS)
		out.GlobalRateLimit.Burst = int(spec.GlobalRateLimit.Burst)
	}

	seen := make(map[string]struct{}, len(spec.Routes))
	for _, rt := range spec.Routes {
		if _, dup := seen[rt.Path]; dup {
			return "", "", fmt.Errorf("duplicate route path %q", rt.Path)
		}
		seen[rt.Path] = struct{}{}

		u, err := url.Parse(rt.Backend)
		if err != nil || (u.Scheme != "http" && u.Scheme != "https") || u.Host == "" {
			return "", "", fmt.Errorf("route %q: invalid backend %q", rt.Path, rt.Backend)
		}

		rule := gwconfig.RouteRule{Path: rt.Path, Methods: rt.Methods, Backend: rt.Backend}
		if rt.RateLimit != nil {
			rule.RateLimit.QPS = int(rt.RateLimit.QPS)
			rule.RateLimit.Burst = int(rt.RateLimit.Burst)
		}
		out.Routes = append(out.Routes, rule)
	}

	// gateway 按顺序做前缀匹配，按路径长度降序排列以保证最长前缀优先
	sort.SliceStable(out.Routes, func(i, j int) bool {
		return len(out.Routes[i].Path) > len(out.Routes[j].Path)
	})

	data, err := json.MarshalIndent(out, "", "  ")
	if err != nil {
		return "", "", err
	}
	sum := sha256.Sum256(data)
	return string(data), hex.EncodeToString(sum[:8]), nil
}

package controller

import (
	"context"
	"fmt"
	"time"

	appsv1 "k8s.io/api/apps/v1"
	corev1 "k8s.io/api/core/v1"
	apierrors "k8s.io/apimachinery/pkg/api/errors"
	"k8s.io/apimachinery/pkg/api/meta"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/runtime"
	"k8s.io/apimachinery/pkg/util/intstr"
	ctrl "sigs.k8s.io/controller-runtime"
	"sigs.k8s.io/controller-runtime/pkg/builder"
	"sigs.k8s.io/controller-runtime/pkg/client"
	"sigs.k8s.io/controller-runtime/pkg/controller/controllerutil"
	logf "sigs.k8s.io/controller-runtime/pkg/log"
	"sigs.k8s.io/controller-runtime/pkg/predicate"

	gatewayv1alpha1 "gin-gateway-split/api/v1alpha1"
)

const (
	configFileName = "routes.json"
	configMountDir = "/etc/gin-gateway"
	conditionReady = "Ready"
)

type GatewayConfigReconciler struct {
	client.Client
	Scheme *runtime.Scheme
}

// +kubebuilder:rbac:groups=gateway.example.com,resources=gatewayconfigs,verbs=get;list;watch;update;patch
// +kubebuilder:rbac:groups=gateway.example.com,resources=gatewayconfigs/status,verbs=get;update;patch
// +kubebuilder:rbac:groups=gateway.example.com,resources=gatewayconfigs/finalizers,verbs=update
// +kubebuilder:rbac:groups=core,resources=configmaps;services,verbs=get;list;watch;create;update;patch;delete
// +kubebuilder:rbac:groups=apps,resources=deployments,verbs=get;list;watch;create;update;patch;delete

func (r *GatewayConfigReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {
	logger := logf.FromContext(ctx)

	var gwc gatewayv1alpha1.GatewayConfig
	if err := r.Get(ctx, req.NamespacedName, &gwc); err != nil {
		// 子资源都带有 OwnerReference，CR 删除后由 GC 级联清理
		return ctrl.Result{}, client.IgnoreNotFound(err)
	}

	data, hash, err := renderConfig(&gwc.Spec)
	if err != nil {
		logger.Info("invalid GatewayConfig spec", "error", err.Error())
		meta.SetStatusCondition(&gwc.Status.Conditions, metav1.Condition{
			Type:               conditionReady,
			Status:             metav1.ConditionFalse,
			Reason:             "InvalidSpec",
			Message:            err.Error(),
			ObservedGeneration: gwc.Generation,
		})
		gwc.Status.ObservedGeneration = gwc.Generation
		// 配置错误重试没有意义，等待用户修改 CR 后再次触发
		return ctrl.Result{}, r.updateStatus(ctx, &gwc)
	}

	if err := r.reconcileConfigMap(ctx, &gwc, data); err != nil {
		return ctrl.Result{}, fmt.Errorf("reconcile configmap: %w", err)
	}
	deploy, err := r.reconcileDeployment(ctx, &gwc)
	if err != nil {
		return ctrl.Result{}, fmt.Errorf("reconcile deployment: %w", err)
	}
	if err := r.reconcileService(ctx, &gwc); err != nil {
		return ctrl.Result{}, fmt.Errorf("reconcile service: %w", err)
	}

	desired := int32(1)
	if gwc.Spec.Gateway.Replicas != nil {
		desired = *gwc.Spec.Gateway.Replicas
	}
	gwc.Status.ObservedGeneration = gwc.Generation
	gwc.Status.ActiveRoutes = int32(len(gwc.Spec.Routes))
	gwc.Status.ReadyReplicas = deploy.Status.ReadyReplicas
	if gwc.Status.ConfigHash != hash {
		gwc.Status.ConfigHash = hash
		gwc.Status.LastRenderTime = time.Now().Format(time.RFC3339)
	}
	cond := metav1.Condition{
		Type:               conditionReady,
		Status:             metav1.ConditionTrue,
		Reason:             "Available",
		Message:            fmt.Sprintf("%d/%d replicas ready", deploy.Status.ReadyReplicas, desired),
		ObservedGeneration: gwc.Generation,
	}
	if deploy.Status.ReadyReplicas < desired {
		cond.Status = metav1.ConditionFalse
		cond.Reason = "Progressing"
	}
	meta.SetStatusCondition(&gwc.Status.Conditions, cond)

	return ctrl.Result{}, r.updateStatus(ctx, &gwc)
}

func (r *GatewayConfigReconciler) updateStatus(ctx context.Context, gwc *gatewayv1alpha1.GatewayConfig) error {
	err := r.Status().Update(ctx, gwc)
	if apierrors.IsConflict(err) {
		// 返回错误让 controller-runtime 按退避策略重新入队，下次会读取到最新版本
		return fmt.Errorf("status conflict, requeue: %w", err)
	}
	return err
}

func labelsFor(gwc *gatewayv1alpha1.GatewayConfig) map[string]string {
	return map[string]string{
		"app.kubernetes.io/name":       "gin-gateway",
		"app.kubernetes.io/instance":   gwc.Name,
		"app.kubernetes.io/managed-by": "gin-gateway-operator",
	}
}

func configMapName(gwc *gatewayv1alpha1.GatewayConfig) string { return gwc.Name + "-routes" }
func workloadName(gwc *gatewayv1alpha1.GatewayConfig) string  { return gwc.Name + "-gateway" }

func (r *GatewayConfigReconciler) reconcileConfigMap(ctx context.Context, gwc *gatewayv1alpha1.GatewayConfig, data string) error {
	cm := &corev1.ConfigMap{ObjectMeta: metav1.ObjectMeta{Name: configMapName(gwc), Namespace: gwc.Namespace}}
	_, err := controllerutil.CreateOrUpdate(ctx, r.Client, cm, func() error {
		cm.Labels = labelsFor(gwc)
		cm.Data = map[string]string{configFileName: data}
		return controllerutil.SetControllerReference(gwc, cm, r.Scheme)
	})
	return err
}

func (r *GatewayConfigReconciler) reconcileDeployment(ctx context.Context, gwc *gatewayv1alpha1.GatewayConfig) (*appsv1.Deployment, error) {
	gw := gwc.Spec.Gateway
	labels := labelsFor(gwc)
	automountToken := false

	deploy := &appsv1.Deployment{ObjectMeta: metav1.ObjectMeta{Name: workloadName(gwc), Namespace: gwc.Namespace}}
	_, err := controllerutil.CreateOrUpdate(ctx, r.Client, deploy, func() error {
		deploy.Labels = labels
		// selector 创建后不可变更
		if deploy.CreationTimestamp.IsZero() {
			deploy.Spec.Selector = &metav1.LabelSelector{MatchLabels: labels}
		}
		deploy.Spec.Replicas = gw.Replicas

		tpl := &deploy.Spec.Template
		tpl.Labels = labels
		tpl.Spec.AutomountServiceAccountToken = &automountToken
		tpl.Spec.Volumes = []corev1.Volume{{
			Name: "routes",
			VolumeSource: corev1.VolumeSource{
				ConfigMap: &corev1.ConfigMapVolumeSource{
					LocalObjectReference: corev1.LocalObjectReference{Name: configMapName(gwc)},
				},
			},
		}}
		probe := func(delay int32) *corev1.Probe {
			return &corev1.Probe{
				ProbeHandler: corev1.ProbeHandler{
					HTTPGet: &corev1.HTTPGetAction{Path: "/healthz", Port: intstr.FromString("http")},
				},
				InitialDelaySeconds: delay,
				PeriodSeconds:       10,
			}
		}
		tpl.Spec.Containers = []corev1.Container{{
			Name:  "gateway",
			Image: gw.Image,
			Env: []corev1.EnvVar{
				{Name: "PORT", Value: fmt.Sprint(gw.Port)},
				{Name: "CONFIG_FILE", Value: configMountDir + "/" + configFileName},
			},
			Ports: []corev1.ContainerPort{{Name: "http", ContainerPort: gw.Port, Protocol: corev1.ProtocolTCP}},
			// 不能使用 subPath，否则 ConfigMap 更新不会同步到容器内
			VolumeMounts:   []corev1.VolumeMount{{Name: "routes", MountPath: configMountDir, ReadOnly: true}},
			ReadinessProbe: probe(3),
			LivenessProbe:  probe(5),
			Resources:      gw.Resources,
		}}
		return controllerutil.SetControllerReference(gwc, deploy, r.Scheme)
	})
	return deploy, err
}

func (r *GatewayConfigReconciler) reconcileService(ctx context.Context, gwc *gatewayv1alpha1.GatewayConfig) error {
	svc := &corev1.Service{ObjectMeta: metav1.ObjectMeta{Name: workloadName(gwc), Namespace: gwc.Namespace}}
	_, err := controllerutil.CreateOrUpdate(ctx, r.Client, svc, func() error {
		svc.Labels = labelsFor(gwc)
		svc.Spec.Type = gwc.Spec.Gateway.ServiceType
		svc.Spec.Selector = labelsFor(gwc)
		svc.Spec.Ports = []corev1.ServicePort{{
			Name:       "http",
			Port:       80,
			TargetPort: intstr.FromString("http"),
			Protocol:   corev1.ProtocolTCP,
		}}
		return controllerutil.SetControllerReference(gwc, svc, r.Scheme)
	})
	return err
}

func (r *GatewayConfigReconciler) SetupWithManager(mgr ctrl.Manager) error {
	return ctrl.NewControllerManagedBy(mgr).
		// 只在 spec 变化时触发，避免自身写 status 引发的循环调和
		For(&gatewayv1alpha1.GatewayConfig{},
			builder.WithPredicates(predicate.GenerationChangedPredicate{})).
		// 子资源被修改/删除或 Deployment 就绪副本变化时，重新调和
		Owns(&corev1.ConfigMap{}).
		Owns(&appsv1.Deployment{}).
		Owns(&corev1.Service{}).
		Named("gatewayconfig").
		Complete(r)
}

func (h *Handler) ResetLimiters() {
	h.limiters.Range(func(key, _ any) bool {
		h.limiters.Delete(key)
		return true
	})
}

package reloader

import (
	"context"
	"crypto/sha256"
	"encoding/json"
	"fmt"
	"log"
	"os"
	"path/filepath"

	"github.com/fsnotify/fsnotify"

	"gin-gateway-split/internal/gwconfig"
)

type Reloader struct {
	path     string
	onReload func(*gwconfig.GatewaySpec)
	lastHash [sha256.Size]byte
}

func New(path string, onReload func(*gwconfig.GatewaySpec)) *Reloader {
	return &Reloader{path: path, onReload: onReload}
}

// Load 读取并解析配置文件，内容未变化时直接返回
func (r *Reloader) Load() error {
	data, err := os.ReadFile(r.path)
	if err != nil {
		return fmt.Errorf("read %s: %w", r.path, err)
	}
	sum := sha256.Sum256(data)
	if sum == r.lastHash {
		return nil
	}

	var spec gwconfig.GatewaySpec
	if err := json.Unmarshal(data, &spec); err != nil {
		return fmt.Errorf("parse %s: %w", r.path, err)
	}
	r.lastHash = sum
	r.onReload(&spec)
	log.Printf("[Reloader] loaded %d routes from %s", len(spec.Routes), r.path)
	return nil
}

// Watch 阻塞监听配置目录，直到 ctx 结束
func (r *Reloader) Watch(ctx context.Context) error {
	w, err := fsnotify.NewWatcher()
	if err != nil {
		return err
	}
	defer w.Close()

	// kubelet 通过原子替换 ..data 符号链接更新 ConfigMap 卷，只能 watch 目录
	if err := w.Add(filepath.Dir(r.path)); err != nil {
		return err
	}

	for {
		select {
		case <-ctx.Done():
			return nil
		case _, ok := <-w.Events:
			if !ok {
				return nil
			}
			if err := r.Load(); err != nil {
				log.Printf("[Reloader] reload failed, keep last config: %v", err)
			}
		case err, ok := <-w.Errors:
			if !ok {
				return nil
			}
			log.Printf("[Reloader] watch error: %v", err)
		}
	}
}

package main

import (
	"context"
	"errors"
	"log"
	"net/http"
	"os"
	"os/signal"
	"syscall"
	"time"

	"github.com/gin-gonic/gin"

	"gin-gateway-split/internal/gateway"
	"gin-gateway-split/internal/gwconfig"
	"gin-gateway-split/internal/reloader"
)

func getenv(key, def string) string {
	if v := os.Getenv(key); v != "" {
		return v
	}
	return def
}

func main() {
	configFile := getenv("CONFIG_FILE", "/etc/gin-gateway/routes.json")
	port := getenv("PORT", "8080")

	store := gwconfig.NewStore()
	handler := gateway.NewHandler(store)

	rl := reloader.New(configFile, func(spec *gwconfig.GatewaySpec) {
		store.UpdateSpec(spec)
		handler.ResetLimiters()
	})
	// ConfigMap 由 Operator 先于 Deployment 创建，首次加载失败说明配置异常，直接退出
	if err := rl.Load(); err != nil {
		log.Fatalf("initial config load failed: %v", err)
	}

	ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM)
	defer stop()

	go func() {
		if err := rl.Watch(ctx); err != nil {
			log.Printf("config watcher exited: %v", err)
		}
	}()

	r := gin.Default()
	handler.RegisterRoutes(r)
	srv := &http.Server{Addr: ":" + port, Handler: r}

	go func() {
		<-ctx.Done()
		log.Println("Shutting down...")
		shutdownCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
		defer cancel()
		_ = srv.Shutdown(shutdownCtx)
	}()

	log.Printf("Starting gin-gateway on :%s, config=%s", port, configFile)
	if err := srv.ListenAndServe(); err != nil && !errors.Is(err, http.ErrServerClosed) {
		log.Fatalf("Gin server failed: %v", err)
	}
}