package tunnel import ( "context" "fmt" "log/slog" "net/http" "os" "os/exec" "reflect" "sync/atomic" "syscall" "time" "k8s.io/client-go/kubernetes" "k8s.io/client-go/rest" "quorate.dev/services/short/config" ) func readRuntime(file string) (config.Runtime, error) { r, err := config.Read[config.Runtime](file) if err == nil { err = r.Bundle.Validate() } if err == nil { err = r.Path.Validate() } return r, err } func Run(ctx context.Context, file, namespace, name string) error { kc, err := rest.InClusterConfig() if err != nil { return err } kube, err := kubernetes.NewForConfig(kc) if err != nil { return err } if err = label(ctx, kube, namespace, name, nil); err != nil { return err } var ready atomic.Bool var leading atomic.Bool server := &http.Server{Addr: ":9090", ReadHeaderTimeout: 5 * time.Second, Handler: http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { if r.URL.Path != "/ready" { http.NotFound(w, r) return } _, err := readRuntime(file) if err != nil || leading.Load() && !ready.Load() { w.WriteHeader(http.StatusServiceUnavailable) return } fmt.Fprintln(w, "ready") })} go func() { if err := server.ListenAndServe(); err != nil && err != http.ErrServerClosed { slog.Error("health server", "error", err) } }() defer server.Close() err = ownership(ctx, kube, namespace, name, func(leader context.Context) error { leading.Store(true) return serve(leader, file, &ready, func(request context.Context) error { return claim(request, kube, namespace, name) }) }) ready.Store(false) request, cancel := context.WithTimeout(context.Background(), 5*time.Second) defer cancel() if err := label(request, kube, namespace, name, nil); err != nil { slog.Error("remove active tunnel label", "error", err) } return err } func serve(ctx context.Context, file string, ready *atomic.Bool, publish func(context.Context) error) error { var stop context.CancelFunc var exited chan error var active *config.Runtime var loaded bool var published bool shutdown := func() { ready.Store(false) if stop != nil { stop() <-exited stop = nil } } defer shutdown() ticker := time.NewTicker(5 * time.Second) defer ticker.Stop() step := func() error { r, err := readRuntime(file) if err != nil { return err } restart := active == nil || !reflect.DeepEqual(active.Bundle, r.Bundle) || active.Path.Address != r.Path.Address || active.Path.Network != r.Path.Network if restart { shutdown() if err = network(r.Path); err != nil { return err } child, cancel := context.WithCancel(ctx) cmd := exec.CommandContext(child, "/usr/local/libexec/ipsec/charon") cmd.Cancel = func() error { return cmd.Process.Signal(syscall.SIGTERM) } cmd.WaitDelay = 10 * time.Second cmd.Stdout = os.Stdout cmd.Stderr = os.Stderr if err = cmd.Start(); err != nil { cancel() return err } stop = cancel exited = make(chan error, 1) go func() { exited <- cmd.Wait() }() active = &r loaded = false } else if !reflect.DeepEqual(active.Path, r.Path) { if err = firewall(r.Path); err != nil { return err } active = &r } if !loaded { for index, peer := range r.Peers { if err = load(r.WAN, peer, index); err != nil { return err } } loaded = true } if !published { if err = publish(ctx); err != nil { return err } published = true } up, err := installed() if err != nil { return err } if index := returnPeer(up); index >= 0 { if err = returnRoute(r.Path.Network, index); err != nil { return err } } ready.Store(returnPeer(up) >= 0) for index, live := range up { if !live { if err := initiate(index); err != nil { slog.Error("initiate headend", "index", index, "error", err) } } } return nil } for { if ctx.Err() != nil { return nil } if err := step(); err != nil { ready.Store(false) slog.Error("reconcile tunnel", "error", err) } select { case <-ctx.Done(): return nil case err := <-exited: stop = nil return fmt.Errorf("charon exited: %w", err) case <-ticker.C: } } }