Added wobberator-agent

Dockerfile

@@ -12,12 +12,17 @@ RUN go mod download
 COPY . .
 
 RUN CGO_ENABLED=0 GOOS=linux GOARCH=amd64 go build cmd/wobberator.go
+RUN CGO_ENABLED=0 GOOS=linux GOARCH=amd64 go build cmd/wobberator-agent.go
 
 FROM alpine:3.10 as certs
 RUN apk add ca-certificates && update-ca-certificates
 
 # final stage
 FROM scratch
+WORKDIR /app
+
 COPY --from=builder /app/wobberator /app/
+COPY --from=builder /app/wobberator-agent /app/
 COPY --from=certs /etc/ssl/certs/ca-certificates.crt /etc/ssl/certs/
+
 ENTRYPOINT ["/app/wobberator"]

cmd/wobberator-agent.go

@@ -0,0 +1,40 @@
+package main
+
+import (
+	"context"
+	"github.com/wobcom/wobberator/internal/app"
+	config2 "github.com/wobcom/wobberator/internal/pkg/config"
+	"github.com/wobcom/wobberator/internal/pkg/kubernetes"
+	"log/slog"
+	"os"
+	"sync"
+)
+
+func main() {
+
+	slog.SetLogLoggerLevel(slog.LevelDebug)
+
+	clientset, err := kubernetes.GetClient()
+	if err != nil {
+		slog.Error("Kubernetes client could not be loaded", "err", err)
+		os.Exit(1)
+	}
+
+	config, err := config2.GetConfig()
+	if err != nil {
+		slog.Error("Config could not be loaded", "err", err)
+		os.Exit(1)
+	}
+	ctx := context.Background()
+
+	var wg sync.WaitGroup
+
+	wg.Add(1)
+	go func() {
+		defer wg.Done()
+		slog.Info("Calling RunHostRouteAssignment")
+		app.RunHostRouteAssignment(ctx, clientset, &config.HostRouteAssignmentConfig)
+	}()
+
+	wg.Wait()
+}

go.mod

@@ -27,6 +27,8 @@ require (
 	github.com/modern-go/reflect2 v1.0.2 // indirect
 	github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 // indirect
 	github.com/spf13/pflag v1.0.5 // indirect
+	github.com/vishvananda/netlink v1.1.0 // indirect
+	github.com/vishvananda/netns v0.0.0-20191106174202-0a2b9b5464df // indirect
 	golang.org/x/net v0.23.0 // indirect
 	golang.org/x/oauth2 v0.10.0 // indirect
 	golang.org/x/sys v0.18.0 // indirect

go.sum

@@ -74,6 +74,10 @@ github.com/stretchr/testify v1.8.0/go.mod h1:yNjHg4UonilssWZ8iaSj1OCr/vHnekPRkoO
 github.com/stretchr/testify v1.8.1/go.mod h1:w2LPCIKwWwSfY2zedu0+kehJoqGctiVI29o6fzry7u4=
 github.com/stretchr/testify v1.8.4 h1:CcVxjf3Q8PM0mHUKJCdn+eZZtm5yQwehR5yeSVQQcUk=
 github.com/stretchr/testify v1.8.4/go.mod h1:sz/lmYIOXD/1dqDmKjjqLyZ2RngseejIcXlSw2iwfAo=
+github.com/vishvananda/netlink v1.1.0 h1:1iyaYNBLmP6L0220aDnYQpo1QEV4t4hJ+xEEhhJH8j0=
+github.com/vishvananda/netlink v1.1.0/go.mod h1:cTgwzPIzzgDAYoQrMm0EdrjRUBkTqKYppBueQtXaqoE=
+github.com/vishvananda/netns v0.0.0-20191106174202-0a2b9b5464df h1:OviZH7qLw/7ZovXvuNyL3XQl8UFofeikI1NW1Gypu7k=
+github.com/vishvananda/netns v0.0.0-20191106174202-0a2b9b5464df/go.mod h1:JP3t17pCcGlemwknint6hfoeCVQrEMVwxRLRjXpq+BU=
 github.com/yuin/goldmark v1.1.27/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74=
 github.com/yuin/goldmark v1.2.1/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74=
 golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w=
@@ -95,6 +99,7 @@ golang.org/x/sync v0.0.0-20190911185100-cd5d95a43a6e/go.mod h1:RxMgew5VJxzue5/jJ
 golang.org/x/sync v0.0.0-20201020160332-67f06af15bc9/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
 golang.org/x/sys v0.0.0-20190412213103-97732733099d/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
+golang.org/x/sys v0.0.0-20190606203320-7fc4e5ec1444/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20200930185726-fdedc70b468f/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.18.0 h1:DBdB3niSjOA/O0blCZBqDefyWNYveAYMNF1Wum0DYQ4=
 golang.org/x/sys v0.18.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=

internal/app/wobberator.go

@@ -2,7 +2,9 @@ package app
 
 import (
 	"context"
+	"errors"
 	"fmt"
+	"github.com/vishvananda/netlink"
 	"github.com/wobcom/wobberator/internal/pkg/config"
 	"k8s.io/client-go/kubernetes"
 	"log/slog"
@@ -39,6 +41,118 @@ func getAvailableAddreses(cidr string, alreadyUsedAddresses []netip.Addr) []neti
 	return availableIPs
 }
 
+func EnsureIPAddressesOnInterface(addresses []netip.Addr) error {
+
+	link, err := netlink.LinkByName("antikubermatik")
+	var linkNotFoundError netlink.LinkNotFoundError
+	isLNF := errors.As(err, &linkNotFoundError)
+	if isLNF {
+		dummyAttrs := netlink.LinkAttrs{Name: "antikubermatik"}
+		dummyInterface := netlink.Dummy{LinkAttrs: dummyAttrs}
+		err := netlink.LinkAdd(&dummyInterface)
+		if err != nil {
+			return err
+		}
+	} else if err != nil {
+		return err
+	}
+
+	existingAddresses, err := netlink.AddrList(link, netlink.FAMILY_V6)
+	if err != nil {
+		return err
+	}
+
+	for _, eAddress := range existingAddresses {
+		netAddress := netip.MustParsePrefix(eAddress.String())
+		if !slices.Contains(addresses, netAddress.Addr()) {
+			slog.Info("Removing address from interface", "address", eAddress.String())
+			err := netlink.AddrDel(link, &eAddress)
+			if err != nil {
+				return err
+			}
+		}
+	}
+
+	for _, ensureAddress := range addresses {
+		netlinkAddress, err := netlink.ParseAddr(fmt.Sprintf("%v/128", ensureAddress.String()))
+		if err != nil {
+			return err
+		}
+		alreadyExists := false
+		for _, eA := range existingAddresses {
+			if netlinkAddress.IP.Equal(eA.IP) {
+				alreadyExists = true
+			}
+		}
+
+		if alreadyExists {
+			continue
+		}
+
+		slog.Info("Adding address to interface", "address", netlinkAddress.String())
+		err = netlink.AddrAdd(link, netlinkAddress)
+		if err != nil {
+			return err
+		}
+	}
+
+	return nil
+}
+
+func RunHostRouteAssignment(ctx context.Context, clientset *kubernetes.Clientset, cfg *config.HostRouteAssignmentConfig) {
+	for {
+
+		allowedNets := make([]netip.Prefix, 0)
+		for _, network := range cfg.ServiceNetworks {
+			prefix := netip.MustParsePrefix(network)
+			allowedNets = append(allowedNets, prefix)
+		}
+
+		services, err := clientset.CoreV1().Services("").List(ctx, metav1.ListOptions{})
+		if err != nil {
+			slog.Error("Unable to fetch services", "err", err)
+		}
+
+		activeServiceIPs := make([]netip.Addr, 0)
+
+		slog.Info("Loaded services", "len", len(services.Items))
+
+		for _, service := range services.Items {
+
+			for _, ingress := range service.Status.LoadBalancer.Ingress {
+				ip, err := netip.ParseAddr(ingress.IP)
+				if err != nil {
+					slog.Warn("ingress.IP was invalid", "err", err, "ip", ingress.IP)
+				}
+
+				isAllowed := false
+				for _, net := range allowedNets {
+					if net.Contains(ip) {
+						isAllowed = true
+					}
+				}
+
+				if !isAllowed {
+					slog.Info("Skipping IP address", "ip", ip.String())
+					continue
+				}
+
+				activeServiceIPs = append(activeServiceIPs, ip)
+
+			}
+
+		}
+
+		err = EnsureIPAddressesOnInterface(activeServiceIPs)
+		if err != nil {
+			slog.Error("Error while ensuring ip addresses on interface", "err", err)
+		}
+
+		time.Sleep(10 * time.Second)
+	}
+
+}
+
 func RunRouterIDAssignment(ctx context.Context, clientset *kubernetes.Clientset, cfg *config.RouterIDAssignmentConfig) {
 	slog.Debug("Called RunRouterIDAssignment")
 

internal/pkg/config/config.go

@@ -10,8 +10,13 @@ type RouterIDAssignmentConfig struct {
 	} `yaml:"asns"`
 }
 
+type HostRouteAssignmentConfig struct {
+	ServiceNetworks []string `yaml:"serviceNetworks"`
+}
+
 type Config struct {
-	RouterIDAssignmentConfig RouterIDAssignmentConfig `yaml:"routerIdAssignment"`
+	RouterIDAssignmentConfig  RouterIDAssignmentConfig  `yaml:"routerIdAssignment"`
+	HostRouteAssignmentConfig HostRouteAssignmentConfig `yaml:"hostRouteAssignment"`
 }
 
 func GetConfig() (*Config, error) {