feat(vm-route-forge): impl route reconciliation (#242)

Signed-off-by: yaroslavborbat <[email protected]>
Signed-off-by: Yaroslav Borbat <[email protected]>
Co-authored-by: Ivan Mikheykin <[email protected]>

CONTRIBUTING.md

@@ -100,7 +100,7 @@ Supported scopes are the following:
   # Core mechanisms and low-level system functionalities.
   - core
     - api-service
-    - vmi-router
+    - vm-route-forge
     - kubevirt
     - kube-api-rewriter
     - cdi

Taskfile.yaml

@@ -5,9 +5,9 @@ version: "3"
 silent: true
 
 includes:
-  vmi-router:
-    taskfile: ./images/vmi-router
-    dir: ./images/vmi-router
+  vm-route-forge:
+    taskfile: ./images/vm-route-forge
+    dir: ./images/vm-route-forge
   virtualization-controller:
     taskfile: ./images/virtualization-artifact
     dir: ./images/virtualization-artifact

images/vmi-router/README.md → images/vm-route-forge/README.md

@@ -1,6 +1,4 @@
-## vmi-router
-
-> **NOTE:** Not an accurate name, should be 'cilium-route-updater'.
+## vm-route-forge
 
 This controller watches for VirtualMachines in virtualization.deckhouse.io group and updates routes in table 1490 to route traffic between VMs via Cilium agents.
 
@@ -23,7 +21,7 @@ Use --cidr flags to specify CIDRs to limit managed IPs. Controller will update r
 Example:
 
 ```
-vmi-router --cidr 10.2.0.0/24 --cidr 10.2.1.0/24 --cidr 10.2.2.0/24 
+vm-route-forge --cidr 10.2.0.0/24 --cidr 10.2.1.0/24 --cidr 10.2.2.0/24 
 ```
 
 Controller will update route for VM with IP 10.2.1.32, but will ignore VM with IP 10.2.4.5.
@@ -32,10 +30,9 @@ Controller will update route for VM with IP 10.2.1.32, but will ignore VM with I
 
 Use --dry-run flag to enable non destructive mode. The controller will not actually delete or replace rules and routes, only log these actions.
 
-#### Metrics and healthz addresses
+#### Healthz addresses
 
-Controller can't predict used ports when starting in host network mode. So, be default, metrics and healthz are started on random free ports. Use flags to specify these addresses:
+Controller can't predict used ports when starting in host network mode. So, be default, healthz are started on random free ports. Use flags to specify these addresses:
 
-`--metrics-bind-address` - set port for /metrics endpoint, e.g. `--metrics-bind-address=:9250`
 `--health-probe-bind-address` - set port for /healthz endpoint, e.g. `--health-probe-bind-address=:9321`
 

images/vm-route-forge/cmd/vm-route-forge/app/options/options.go

@@ -0,0 +1,80 @@
+/*
+Copyright 2024 Flant JSC
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+     http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package options
+
+import (
+	"os"
+	"strconv"
+
+	"github.com/spf13/pflag"
+	"sigs.k8s.io/controller-runtime/pkg/log/zap"
+
+	"vm-route-forge/internal/netutil"
+)
+
+type Options struct {
+	ZapOptions zap.Options
+
+	Verbosity int
+	Cidrs     netutil.CIDRSet
+	DryRun    bool
+	ProbeAddr string
+	NodeName  string
+	TableID   string
+}
+
+const (
+	flagCidr, flagCidrShort           = "cidr", "c"
+	flagDryRun, flagDryRunShort       = "dry-run", "d"
+	flagProbeAddr                     = "health-probe-bind-address"
+	flagVerbosity, flagVerbosityShort = "verbosity", "v"
+	flagNodeName, flagNodeNameShort   = "nodeName", "n"
+	flagTableId, flagTableIdShort     = "tableId", "t"
+	defaultVerbosity                  = 1
+
+	VerbosityEnv = "VERBOSITY"
+	NodeNameEnv  = "NODE_NAME"
+	TableIDEnv   = "TABLE_ID"
+)
+
+func NewOptions() Options {
+	return Options{
+		ZapOptions: zap.Options{
+			Development: true,
+		},
+	}
+}
+
+func (o *Options) Flags(fs *pflag.FlagSet) {
+	fs.StringSliceVarP((*[]string)(&o.Cidrs), flagCidr, flagCidrShort, []string{}, "CIDRs enabled to route (multiple flags allowed)")
+	fs.BoolVarP(&o.DryRun, flagDryRun, flagDryRunShort, false, "Don't perform any changes on the node.")
+	fs.StringVar(&o.ProbeAddr, flagProbeAddr, ":0", "The address the probe endpoint binds to.")
+	fs.StringVarP(&o.NodeName, flagNodeName, flagNodeNameShort, os.Getenv(NodeNameEnv), "The name of the node.")
+	fs.StringVarP(&o.TableID, flagTableId, flagTableIdShort, os.Getenv(TableIDEnv), "The id of the table.")
+	fs.IntVarP(&o.Verbosity, flagVerbosity, flagVerbosityShort, getDefaultVerbosity(), "Verbosity of output")
+}
+
+func getDefaultVerbosity() int {
+	if v, ok := os.LookupEnv(VerbosityEnv); ok {
+		verbosity, err := strconv.Atoi(v)
+		if err != nil {
+			return defaultVerbosity
+		}
+		return verbosity
+	}
+	return defaultVerbosity
+}

images/vm-route-forge/cmd/vm-route-forge/app/root.go

@@ -0,0 +1,203 @@
+/*
+Copyright 2024 Flant JSC
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+     http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package app
+
+import (
+	"context"
+	"fmt"
+	"net"
+	"strconv"
+
+	"github.com/spf13/cobra"
+	"go.uber.org/zap/zapcore"
+	"k8s.io/client-go/kubernetes"
+	ctrl "sigs.k8s.io/controller-runtime"
+	"sigs.k8s.io/controller-runtime/pkg/client/config"
+	logf "sigs.k8s.io/controller-runtime/pkg/log"
+	"sigs.k8s.io/controller-runtime/pkg/log/zap"
+	"sigs.k8s.io/controller-runtime/pkg/manager/signals"
+
+	"vm-route-forge/cmd/vm-route-forge/app/options"
+	"vm-route-forge/internal/cache"
+	"vm-route-forge/internal/controller/route"
+	"vm-route-forge/internal/informer"
+	"vm-route-forge/internal/netlinkmanager"
+	"vm-route-forge/internal/server"
+)
+
+const long = `
+                                       _              __
+__   ___ __ ___        _ __ ___  _   _| |_ ___       / _| ___  _ __ __ _  ___
+\ \ / / '_ ` + "`" + ` _ \ _____| '__/ _ \| | | | __/ _ \_____| |_ / _ \| '__/ _` + "`" + ` |/ _ \
+ \ V /| | | | | |_____| | | (_) | |_| | ||  __/_____|  _| (_) | | | (_| |  __/
+  \_/ |_| |_| |_|     |_|  \___/ \__,_|\__\___|     |_|  \___/|_|  \__, |\___|
+                                                                   |___/
+Managing virtual machine routes
+`
+
+const (
+	appName = "vm-route-forge"
+	// The count of workers that will be started for the route controller.
+	// We are currently supporting only one worker.
+	countWorkersRouteController = 1
+)
+
+var (
+	log = ctrl.Log.WithName(appName)
+)
+
+func NewVmRouteForgeCommand() *cobra.Command {
+	opts := options.NewOptions()
+	cmd := &cobra.Command{
+		Short: "Managing virtual machine routes",
+		Long:  long,
+		RunE: func(c *cobra.Command, args []string) error {
+			return run(opts)
+		},
+	}
+	opts.Flags(cmd.Flags())
+	return cmd
+}
+
+func setupLogger(verbosity int) {
+	debug := false
+	if verbosity > 1 {
+		debug = true
+	}
+
+	// The logger instantiated here can be changed to any logger
+	// implementing the logr.Logger interface. This logger will
+	// be propagated through the whole operator, generating
+	// uniform and structured logs.
+	logf.SetLogger(zap.New(zap.Level(zapcore.Level(-1*verbosity)), zap.UseDevMode(debug)))
+}
+
+func run(opts options.Options) error {
+	setupLogger(opts.Verbosity)
+	var parsedCIDRs []*net.IPNet
+	for _, cidr := range opts.Cidrs {
+		_, parsedCIDR, err := net.ParseCIDR(cidr)
+		if err != nil || parsedCIDR == nil {
+			log.Error(err, "failed to parse passed CIDRs")
+			return err
+		}
+		parsedCIDRs = append(parsedCIDRs, parsedCIDR)
+	}
+	log.Info(fmt.Sprintf("Got CIDRs to manage: %+v", opts.Cidrs))
+
+	if opts.DryRun {
+		log.Info("Dry run mode is enabled, will not change network rules and routes")
+	}
+
+	tableID := netlinkmanager.DefaultCiliumRouteTable
+	tableIDStr := opts.TableID
+	if tableIDStr != "" {
+		tableId, err := strconv.ParseInt(tableIDStr, 10, 32)
+		if err != nil {
+			log.Error(err, "failed to parse Cilium table id, should be integer")
+			return err
+		}
+		tableID = int(tableId)
+	}
+	log.Info(fmt.Sprintf("Use cilium route table id %d", tableID))
+
+	// Load configuration to connect to Kubernetes API Server.
+	kubeCfg, err := config.GetConfig()
+	if err != nil {
+		log.Error(err, "Failed to load Kubernetes config")
+		return err
+	}
+	kubeClient, err := kubernetes.NewForConfig(kubeCfg)
+	if err != nil {
+		log.Error(err, "Failed to create Kubernetes client")
+		return err
+	}
+
+	ctx := signals.SetupSignalHandler()
+
+	vmSharedInformerFactory, err := informer.VirtualizationInformerFactory(kubeCfg)
+	if err != nil {
+		log.Error(err, "Failed to create virtualization shared factory")
+		return err
+	}
+	go vmSharedInformerFactory.Virtualization().V1alpha2().VirtualMachines().Informer().Run(ctx.Done())
+
+	ciliumSharedInformerFactory, err := informer.CiliumInformerFactory(kubeCfg)
+	if err != nil {
+		log.Error(err, "Failed to create cilium shared factory")
+		return err
+	}
+	go ciliumSharedInformerFactory.Cilium().V2().CiliumNodes().Informer().Run(ctx.Done())
+
+	sharedCache := cache.NewCache()
+
+	netMgr := netlinkmanager.New(vmSharedInformerFactory.Virtualization().V1alpha2().VirtualMachines(),
+		ciliumSharedInformerFactory.Cilium().V2().CiliumNodes(),
+		sharedCache,
+		log,
+		tableID,
+		parsedCIDRs,
+		opts.DryRun,
+	)
+
+	err = startupSync(ctx, netMgr)
+	if err != nil {
+		log.Error(err, "Failed to run pre sync")
+		return err
+	}
+	routeCtrl, err := route.NewRouteController(
+		vmSharedInformerFactory.Virtualization().V1alpha2().VirtualMachines(),
+		ciliumSharedInformerFactory.Cilium().V2().CiliumNodes(),
+		netMgr,
+		sharedCache,
+		parsedCIDRs,
+		log,
+	)
+	if err != nil {
+		log.Error(err, "Failed to create route controller")
+		return err
+	}
+	go routeCtrl.Run(ctx, countWorkersRouteController)
+
+	srv, err := server.NewServer(
+		kubeClient,
+		server.Options{HealthProbeBindAddress: opts.ProbeAddr},
+		log,
+	)
+	if err != nil {
+		log.Error(err, "Failed to create server")
+		return err
+	}
+	return srv.Run(ctx)
+}
+
+func startupSync(ctx context.Context, mgr *netlinkmanager.Manager) error {
+	log.Info("Synchronize route rules at start")
+	err := mgr.SyncRules()
+	if err != nil {
+		log.Error(err, fmt.Sprintf("failed to synchronize routing rules ar start"))
+		return err
+	}
+
+	log.Info("Synchronize VM routes at start")
+	err = mgr.SyncRoutes(ctx)
+	if err != nil {
+		log.Error(err, fmt.Sprintf("failed to synchronize VM routes at start"))
+		return err
+	}
+	return nil
+}

images/vm-route-forge/cmd/vm-route-forge/main.go

@@ -0,0 +1,32 @@
+/*
+Copyright 2024 Flant JSC
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+     http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package main
+
+import (
+	"fmt"
+	"os"
+
+	"vm-route-forge/cmd/vm-route-forge/app"
+)
+
+func main() {
+	vmRouteForge := app.NewVmRouteForgeCommand()
+	if err := vmRouteForge.Execute(); err != nil {
+		fmt.Fprintf(os.Stderr, "vm-route-forge: %s\n", err)
+		os.Exit(1)
+	}
+}

images/vmi-router/go.mod → images/vm-route-forge/go.mod

@@ -1,4 +1,4 @@
-module vmi-router
+module vm-route-forge
 
 go 1.21.0