package main import ( "context" "crypto/tls" "crypto/x509" "errors" "fmt" "log/slog" "net/http" "os" "os/signal" "strings" "syscall" "time" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/client-go/dynamic" coordinationv1 "k8s.io/client-go/kubernetes/typed/coordination/v1" "k8s.io/client-go/rest" "k8s.io/client-go/tools/leaderelection" "k8s.io/client-go/tools/leaderelection/resourcelock" ) func main() { ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM) defer stop() if err := run(ctx, os.Args[1:]); err != nil && !errors.Is(err, context.Canceled) { slog.Error("project delivery", "error", err) os.Exit(1) } } func run(ctx context.Context, arguments []string) error { if len(arguments) != 1 { return errors.New("expected one command: run") } switch arguments[0] { case "run": return runController(ctx) default: return fmt.Errorf("unknown command %q", arguments[0]) } } func runController(ctx context.Context) error { clientConfig, err := rest.InClusterConfig() if err != nil { return fmt.Errorf("load Kubernetes configuration: %w", err) } client, err := dynamic.NewForConfig(clientConfig) if err != nil { return fmt.Errorf("create Kubernetes client: %w", err) } forgejoHTTP, err := httpClient(os.Getenv("FORGEJO_CA_PATH"), 30*time.Second) if err != nil { return err } publicToken, err := secretFile(os.Getenv("FORGEJO_PUBLIC_TOKEN_PATH")) if err != nil { return err } privateToken, err := secretFile(os.Getenv("FORGEJO_PRIVATE_TOKEN_PATH")) if err != nil { return err } publicRefs, err := newForgejoClient( valueOrDefault("FORGEJO_PUBLIC_URL", "http://forge-git-http.forgejo-public.svc.cluster.local:3000"), forgejoHTTP, publicToken, ) if err != nil { return err } privateRefs, err := newForgejoClient( valueOrDefault("FORGEJO_PRIVATE_URL", "http://forgejo-http.forgejo.svc.cluster.local:3000"), forgejoHTTP, privateToken, ) if err != nil { return err } refs, err := newForgejoRouter( forgejoRoute{authority: valueOrDefault("FORGEJO_PUBLIC_AUTHORITY", "git.quorate.dev"), source: publicRefs}, forgejoRoute{authority: valueOrDefault("FORGEJO_PRIVATE_AUTHORITY", "forge.quorate.internal"), source: privateRefs}, ) if err != nil { return err } config, err := loadControllerConfig() if err != nil { return err } instance := controller{client: client, source: refs, now: time.Now, config: config} coordination, err := coordinationv1.NewForConfig(clientConfig) if err != nil { return err } identity, err := os.Hostname() if err != nil { return err } namespace := os.Getenv("POD_NAMESPACE") if namespace == "" { return errors.New("POD_NAMESPACE is required") } return instance.lead(ctx, &resourcelock.LeaseLock{ LeaseMeta: metav1.ObjectMeta{Name: "project-delivery", Namespace: namespace}, Client: coordination, LockConfig: resourcelock.ResourceLockConfig{Identity: identity}, }) } func secretFile(path string) (string, error) { if path == "" { return "", nil } value, err := os.ReadFile(path) if err != nil { return "", fmt.Errorf("read secret file: %w", err) } secret := strings.TrimSpace(string(value)) if secret == "" { return "", errors.New("secret file is empty") } return secret, nil } func loadControllerConfig() (controllerConfig, error) { pollInterval, err := durationFromEnvironment("POLL_INTERVAL", 20*time.Second) if err != nil { return controllerConfig{}, err } bucketInterval, err := durationFromEnvironment("OBSERVATION_BUCKET_INTERVAL", time.Minute) if err != nil { return controllerConfig{}, err } requestTimeout, err := durationFromEnvironment("REQUEST_TIMEOUT", 15*time.Second) if err != nil { return controllerConfig{}, err } return controllerConfig{ pollInterval: pollInterval, bucketInterval: bucketInterval, requestTimeout: requestTimeout, templateQuiet: bucketInterval, }, nil } func durationFromEnvironment(name string, fallback time.Duration) (time.Duration, error) { value := os.Getenv(name) if value == "" { return fallback, nil } duration, err := time.ParseDuration(value) if err != nil || duration <= 0 { return 0, fmt.Errorf("%s must be a positive duration", name) } return duration, nil } func valueOrDefault(name, fallback string) string { if value := os.Getenv(name); value != "" { return value } return fallback } func httpClient(caPath string, timeout time.Duration) (*http.Client, error) { transport := http.DefaultTransport.(*http.Transport).Clone() if caPath != "" { pem, err := os.ReadFile(caPath) if err != nil { return nil, fmt.Errorf("read HTTP CA: %w", err) } roots, err := x509.SystemCertPool() if err != nil { return nil, fmt.Errorf("load system roots: %w", err) } if !roots.AppendCertsFromPEM(pem) { return nil, errors.New("HTTP CA contains no certificate") } transport.TLSClientConfig = &tls.Config{RootCAs: roots} } return &http.Client{Transport: transport, Timeout: timeout}, nil } func (c *controller) lead(ctx context.Context, lock resourcelock.Interface) error { electionCtx, cancel := context.WithCancel(ctx) defer cancel() controllerErr := make(chan error, 1) elector, err := leaderelection.NewLeaderElector(leaderelection.LeaderElectionConfig{ Lock: lock, LeaseDuration: 30 * time.Second, RenewDeadline: 20 * time.Second, RetryPeriod: 5 * time.Second, Name: "project-delivery", Callbacks: leaderelection.LeaderCallbacks{ OnStartedLeading: func(leaderCtx context.Context) { controllerErr <- c.run(leaderCtx) cancel() }, OnStoppedLeading: func() { if ctx.Err() == nil { select { case controllerErr <- errors.New("leadership lost"): default: } } }, }, }) if err != nil { return err } elector.Run(electionCtx) select { case err := <-controllerErr: return err default: return ctx.Err() } }