johnandersen777.bsky.social/socialweb-computer-kcp

Switch repository

Loading…

package nodeagent

import (
	"bufio"
	"bytes"
	"context"
	"encoding/json"
	"fmt"
	"log/slog"
	"net"
	"os"
	"os/exec"
	"path/filepath"
	"strings"
	"time"

	"github.com/johnandersen777/socialweb-computer-kcp/internal/kit/clock"
)

type cniRange struct {
	Subnet string `json:"subnet"`
}

type cniRoute struct {
	Dst string `json:"dst"`
}

type cniIPAM struct {
	Type string `json:"type"`

	Ranges [][]cniRange `json:"ranges"`

	Routes []cniRoute `json:"routes"`
}

type cniBridge struct {
	Type string `json:"type"`

	Bridge string `json:"bridge"`

	IsGateway bool `json:"isGateway"`

	IPMasq bool `json:"ipMasq"`

	HairpinMode bool `json:"hairpinMode"`

	MTU int `json:"mtu"`

	IPAM cniIPAM `json:"ipam"`
}

type cniPortmap struct {
	Type string `json:"type"`

	Capabilities map[string]bool `json:"capabilities"`
}

type cniConflist struct {
	CNIVersion string `json:"cniVersion"`

	Name string `json:"name"`

	Plugins []any `json:"plugins"`
}

func CNI(ctx context.Context, cfg Config, log *slog.Logger) error {
	if log == nil {
		log = slog.Default()
	}
	nodeName := os.Getenv("NODE_NAME")
	if nodeName == "" {
		env, err := readEnvFile(cfg.APIServerEnvPath)
		if err != nil {
			return fmt.Errorf("nodeagent: reading %s for NODE_NAME: %w", cfg.APIServerEnvPath, err)
		}
		nodeName = env["NODE_NAME"]
	}
	if !validName(nodeName) {
		return fmt.Errorf("nodeagent: NODE_NAME %q is not a node name", nodeName)
	}
	cidr, err := awaitPodCIDR(ctx, cfg, nodeName, log)
	if err != nil {
		return err
	}
	if err := validPodCIDR(cidr); err != nil {
		return err
	}
	if err := os.MkdirAll(cfg.CNIConfDir, 0o755); err != nil {
		return fmt.Errorf("nodeagent: creating %s: %w", cfg.CNIConfDir, err)
	}
	if err := os.MkdirAll(cfg.CNIBinDir, 0o755); err != nil {
		return fmt.Errorf("nodeagent: creating %s: %w", cfg.CNIBinDir, err)
	}
	for _, plugin := range []string{"bridge", "host-local", "portmap", "loopback"} {
		link := filepath.Join(cfg.CNIBinDir, plugin)
		if err := os.Remove(link); err != nil && !os.IsNotExist(err) {
			return fmt.Errorf("nodeagent: replacing the CNI plugin link %s: %w", link, err)
		}
		if err := os.Symlink(cfg.CNIPluginPath(), link); err != nil {
			return fmt.Errorf("nodeagent: linking %s to %s: %w", link, cfg.CNIPluginPath(), err)
		}
	}
	content, err := json.MarshalIndent(conflistFor(cidr), "", "  ")
	if err != nil {
		return fmt.Errorf("nodeagent: encoding the CNI conflist: %w", err)
	}
	content = append(content, '\n')
	conf := filepath.Join(cfg.CNIConfDir, CNIConflistName)
	if err := writeAtomic(conf, content, 0o644); err != nil {
		return err
	}
	log.Info("wrote the bridge CNI conflist", "path", conf, "podCIDR", cidr, "plugins", cfg.CNIPluginPath())
	return nil
}

func awaitPodCIDR(ctx context.Context, cfg Config, nodeName string, log *slog.Logger) (string, error) {
	limit := time.Duration(cfg.CNIWaitAttempts) * cfg.CNIWaitInterval()
	wctx, cancel := context.WithTimeout(ctx, limit)
	defer cancel()
	last := "the kubelet kubeconfig " + cfg.KubeletKubeconfig + " is not written yet"
	for {
		if fileNonEmpty(cfg.KubeletKubeconfig) {
			cidr, err := followPodCIDR(wctx, cfg, nodeName)
			if cidr != "" {
				return cidr, nil
			}
			if ctx.Err() != nil {
				return "", ctx.Err()
			}
			if err != nil {
				last = err.Error()
				log.Info("the node watch ended before a podCIDR; it restarts", "node", nodeName, "err", last)
			}
		}
		if err := clock.SleepErr(wctx, cfg.CNIWaitInterval()); err != nil {
			if ctx.Err() != nil {
				return "", ctx.Err()
			}
			return "", fmt.Errorf("nodeagent: node %s has no podCIDR yet after %s: %s", nodeName, limit, last)
		}
	}
}

func followPodCIDR(ctx context.Context, cfg Config, nodeName string) (string, error) {
	ctx, cancel := context.WithCancel(ctx)
	defer cancel()
	cmd := exec.CommandContext(ctx, cfg.K3sPath, "kubectl", "--kubeconfig", cfg.KubeletKubeconfig,
		"get", "node", nodeName, "--watch", "-o", `jsonpath={.spec.podCIDR}{"\n"}`)
	var stderr bytes.Buffer
	cmd.Stderr = &stderr
	out, err := cmd.StdoutPipe()
	if err != nil {
		return "", err
	}
	if err := cmd.Start(); err != nil {
		return "", fmt.Errorf("nodeagent: watching node %s: %w", nodeName, err)
	}
	lines := bufio.NewScanner(out)
	for lines.Scan() {
		if cidr := strings.TrimSpace(lines.Text()); cidr != "" {
			cancel()
			_ = cmd.Wait()
			return cidr, nil
		}
	}
	err = cmd.Wait()
	if msg := strings.TrimSpace(stderr.String()); msg != "" {
		return "", fmt.Errorf("nodeagent: watching node %s: %s", nodeName, msg)
	}
	if err != nil {
		return "", fmt.Errorf("nodeagent: watching node %s: %w", nodeName, err)
	}
	return "", fmt.Errorf("nodeagent: the watch of node %s closed", nodeName)
}

func validPodCIDR(cidr string) error {
	ip, network, err := net.ParseCIDR(cidr)
	if err != nil || ip.To4() == nil || network.IP.To4() == nil {
		return fmt.Errorf("nodeagent: podCIDR %s is not an IPv4 CIDR", cidr)
	}
	return nil
}

func conflistFor(cidr string) cniConflist {
	return cniConflist{
		CNIVersion: "1.0.0",
		Name:       "socialweb-pods",
		Plugins: []any{
			cniBridge{
				Type:        "bridge",
				Bridge:      PodBridge,
				IsGateway:   true,
				IPMasq:      false,
				HairpinMode: true,
				MTU:         PodMTU,
				IPAM: cniIPAM{
					Type:   "host-local",
					Ranges: [][]cniRange{{{Subnet: cidr}}},
					Routes: []cniRoute{{Dst: "0.0.0.0/0"}},
				},
			},
			cniPortmap{Type: "portmap", Capabilities: map[string]bool{"portMappings": true}},
		},
	}
}