quorate/services/short/tunnel/run.go

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:
		}
	}
}