Fix scheduler, nic nat release, metalnet instances

api/core/v1alpha1/networkinterface_types.go

@@ -128,6 +128,15 @@ func GetNetworkInterfaceNATClaimer(nic *NetworkInterface, ipFamily corev1.IPFami
 	return nil
 }
 
+func IsNetworkInterfaceNATClaimedBy(nic *NetworkInterface, claimer *NATGateway) bool {
+	for _, nat := range nic.Spec.NATs {
+		if nat.ClaimRef.UID == claimer.UID {
+			return true
+		}
+	}
+	return false
+}
+
 func GetNetworkInterfacePublicIPs(nic *NetworkInterface) []net.IP {
 	res := make([]net.IP, len(nic.Spec.PublicIPs))
 	for i, publicIP := range nic.Spec.PublicIPs {

cmd/controller-manager/main.go

@@ -150,6 +150,14 @@ func main() {
 		os.Exit(1)
 	}
 
+	if err = (&controllers.NetworkInterfaceNATReleaseReconciler{
+		Client:       mgr.GetClient(),
+		APIReader:    mgr.GetAPIReader(),
+		AbsenceCache: lru.New(500),
+	}).SetupWithManager(mgr); err != nil {
+		setupLog.Error(err, "unable to create controller", "controller", "NetworkInterfaceNATRelease")
+	}
+
 	schedulerCache := scheduler.NewCache(
 		mgr.GetLogger().WithName("scheduler").WithName("cache"),
 		scheduler.DefaultCacheStrategy,

internal/apis/core/networkinterface_types.go

@@ -128,6 +128,15 @@ func GetNetworkInterfaceNATClaimer(nic *NetworkInterface, ipFamily corev1.IPFami
 	return nil
 }
 
+func IsNetworkInterfaceNATClaimedBy(nic *NetworkInterface, claimer *NATGateway) bool {
+	for _, nat := range nic.Spec.NATs {
+		if nat.ClaimRef.UID == claimer.UID {
+			return true
+		}
+	}
+	return false
+}
+
 func GetNetworkInterfacePublicIPs(nic *NetworkInterface) []net.IP {
 	res := make([]net.IP, len(nic.Spec.PublicIPs))
 	for i, publicIP := range nic.Spec.PublicIPs {

internal/controllers/controllers_suite_test.go

@@ -173,6 +173,12 @@ var _ = BeforeSuite(func() {
 		Cache:         schedulerCache,
 	}).SetupWithManager(k8sManager)).To(Succeed())
 
+	Expect((&NetworkInterfaceNATReleaseReconciler{
+		Client:       k8sManager.GetClient(),
+		APIReader:    k8sManager.GetAPIReader(),
+		AbsenceCache: lru.New(100),
+	}).SetupWithManager(k8sManager)).To(Succeed())
+
 	mgrCtx, cancel := context.WithCancel(context.Background())
 	DeferCleanup(cancel)
 	go func() {

internal/controllers/daemonset_controller.go

@@ -185,13 +185,35 @@ func (r *DaemonSetReconciler) instancesShouldBeOnNode(
 ) (nodesNeedingDaemonInsts []string, instsToDelete []string) {
 	_, _ = log, hash
 	shouldRun := r.nodeShouldRunDaemonInstance(node, ds)
-	_, exists := nodeToDaemonInsts[node.Name]
+	insts, exists := nodeToDaemonInsts[node.Name]
 
 	switch {
 	case shouldRun && !exists:
 		// If a daemon instance is supposed to be running on a node but isn't, create one.
 		nodesNeedingDaemonInsts = append(nodesNeedingDaemonInsts, node.Name)
-		// TODO: Add cases handling deletion of instances that should not be on a node anymore.
+	case shouldRun:
+		var filtered []*v1alpha1.Instance
+		for _, inst := range insts {
+			if !inst.DeletionTimestamp.IsZero() {
+				continue
+			}
+			filtered = append(filtered, inst)
+		}
+		if len(filtered) == 0 {
+			nodesNeedingDaemonInsts = append(nodesNeedingDaemonInsts, node.Name)
+		} else if len(filtered) > 1 {
+			// Delete any unnecessary instance, keeping the oldest ones.
+			slices.SortFunc(filtered, func(a, b *v1alpha1.Instance) bool {
+				return a.CreationTimestamp.Compare(b.CreationTimestamp.Time) < 0
+			})
+			for _, inst := range filtered[1:] {
+				instsToDelete = append(instsToDelete, inst.Name)
+			}
+		}
+	case !shouldRun:
+		for _, inst := range insts {
+			instsToDelete = append(instsToDelete, inst.Name)
+		}
 	}
 
 	return nodesNeedingDaemonInsts, instsToDelete

internal/controllers/daemonset_controller_test.go

@@ -100,5 +100,18 @@ var _ = Describe("DaemonSetController", func() {
 		Eventually(ObjectList(&v1alpha1.InstanceList{},
 			client.InNamespace(ns.Name),
 		)).Should(HaveField("Items", HaveEach(HaveField("Spec.IPs", []net.IP{net.MustParseIP("192.168.178.1")}))))
+
+		By("deleting the instances")
+		Expect(k8sClient.DeleteAllOf(ctx, &v1alpha1.Instance{}, client.InNamespace(ns.Name))).To(Succeed())
+
+		By("waiting for new instances to be created again")
+		Eventually(ObjectList(&v1alpha1.InstanceList{},
+			client.InNamespace(ns.Name),
+		)).Should(HaveField("Items", SatisfyAll(
+			ContainElements(
+				HaveField("DeletionTimestamp", BeNil()),
+				HaveField("DeletionTimestamp", BeNil()),
+			)),
+		))
 	})
 })

internal/controllers/ipaddressgc_controller.go

@@ -116,6 +116,10 @@ func (r *IPAddressGCReconciler) ipAddressClaimerExists(ctx context.Context, addr
 		r.AbsenceCache.Add(claimRef.UID, nil)
 		return false, nil
 	}
+	if claimRef.UID != claimer.UID {
+		r.AbsenceCache.Add(claimRef.UID, nil)
+		return false, nil
+	}
 	return true, nil
 }
 

internal/controllers/networkidgc_controller.go

@@ -117,6 +117,10 @@ func (r *NetworkIDGCReconciler) networkIDClaimerExists(ctx context.Context, netw
 		r.AbsenceCache.Add(claimRef.UID, nil)
 		return false, nil
 	}
+	if claimRef.UID != claimer.UID {
+		r.AbsenceCache.Add(claimRef.UID, nil)
+		return false, nil
+	}
 	return true, nil
 }
 

internal/controllers/networkinterfacenatrelease_controller.go

@@ -0,0 +1,154 @@
+// Copyright 2023 OnMetal authors
+//
+// 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 controllers
+
+import (
+	"context"
+	"fmt"
+
+	"github.com/onmetal/onmetal-api-net/api/core/v1alpha1"
+	apierrors "k8s.io/apimachinery/pkg/api/errors"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/client-go/util/workqueue"
+	"k8s.io/utils/lru"
+	ctrl "sigs.k8s.io/controller-runtime"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+	"sigs.k8s.io/controller-runtime/pkg/event"
+	"sigs.k8s.io/controller-runtime/pkg/handler"
+)
+
+type NetworkInterfaceNATReleaseReconciler struct {
+	client.Client
+	APIReader client.Reader
+
+	AbsenceCache *lru.Cache
+}
+
+func (r *NetworkInterfaceNATReleaseReconciler) networkInterfaceNATExists(
+	ctx context.Context,
+	nic *v1alpha1.NetworkInterface,
+	nat *v1alpha1.NetworkInterfaceNAT,
+) (bool, error) {
+	claimRef := nat.ClaimRef
+	if _, ok := r.AbsenceCache.Get(claimRef.UID); ok {
+		return false, nil
+	}
+
+	natGateway := &metav1.PartialObjectMetadata{
+		TypeMeta: metav1.TypeMeta{
+			APIVersion: v1alpha1.SchemeGroupVersion.String(),
+			Kind:       "NATGateway",
+		},
+	}
+	natGatewayKey := client.ObjectKey{Namespace: nic.Namespace, Name: claimRef.Name}
+	if err := r.APIReader.Get(ctx, natGatewayKey, natGateway); err != nil {
+		if !apierrors.IsNotFound(err) {
+			return false, fmt.Errorf("error getting NAT gateway: %w", err)
+		}
+
+		r.AbsenceCache.Add(claimRef.UID, nil)
+		return false, nil
+	}
+	if claimRef.UID != natGateway.UID {
+		r.AbsenceCache.Add(claimRef.UID, nil)
+		return false, nil
+	}
+	return true, nil
+}
+
+//+kubebuilder:rbac:groups=core.apinet.api.onmetal.de,resources=networkinterfaces,verbs=get;list;watch;update;patch
+//+kubebuilder:rbac:groups=core.apinet.api.onmetal.de,resources=natgateways,verbs=get;list;watch
+
+func (r *NetworkInterfaceNATReleaseReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {
+	log := ctrl.LoggerFrom(ctx)
+	nic := &v1alpha1.NetworkInterface{}
+	if err := r.Get(ctx, req.NamespacedName, nic); err != nil {
+		return ctrl.Result{}, client.IgnoreNotFound(err)
+	}
+
+	if !nic.DeletionTimestamp.IsZero() {
+		log.V(1).Info("Network interface is already deleting")
+		return ctrl.Result{}, nil
+	}
+
+	var filtered []v1alpha1.NetworkInterfaceNAT
+	for _, nat := range nic.Spec.NATs {
+		ok, err := r.networkInterfaceNATExists(ctx, nic, &nat)
+		if err != nil {
+			return ctrl.Result{}, fmt.Errorf("error checking whether NAT %s exists: %w", nat.IPFamily, err)
+		}
+		if !ok {
+			continue
+		}
+
+		filtered = append(filtered, nat)
+	}
+	if len(filtered) == len(nic.Spec.NATs) {
+		log.V(1).Info("All NATs are present, nothing to do")
+		return ctrl.Result{}, nil
+	}
+
+	base := nic.DeepCopy()
+	nic.Spec.NATs = filtered
+	if err := r.Patch(ctx, nic, client.StrategicMergeFrom(base)); err != nil {
+		return ctrl.Result{}, fmt.Errorf("error patching network interface: %w", err)
+	}
+
+	log.V(1).Info("Filtered NATs", "Filtered", filtered, "Original", nic.Spec.NATs)
+	return ctrl.Result{}, nil
+}
+
+func (r *NetworkInterfaceNATReleaseReconciler) enqueueByNATGateway() handler.EventHandler {
+	mapAndEnqueue := func(ctx context.Context, natGateway *v1alpha1.NATGateway, queue workqueue.RateLimitingInterface) {
+		log := ctrl.LoggerFrom(ctx)
+
+		nicList := &v1alpha1.NetworkInterfaceList{}
+		if err := r.List(ctx, nicList,
+			client.InNamespace(natGateway.GetNamespace()),
+		); err != nil {
+			log.Error(err, "Error listing network interfaces")
+			return
+		}
+
+		for _, nic := range nicList.Items {
+			if v1alpha1.IsNetworkInterfaceNATClaimedBy(&nic, natGateway) {
+				queue.Add(ctrl.Request{NamespacedName: client.ObjectKeyFromObject(&nic)})
+			}
+		}
+	}
+
+	return &handler.Funcs{
+		DeleteFunc: func(ctx context.Context, event event.DeleteEvent, queue workqueue.RateLimitingInterface) {
+			natGateway := event.Object.(*v1alpha1.NATGateway)
+			mapAndEnqueue(ctx, natGateway, queue)
+		},
+		GenericFunc: func(ctx context.Context, event event.GenericEvent, queue workqueue.RateLimitingInterface) {
+			natGateway := event.Object.(*v1alpha1.NATGateway)
+			if !natGateway.GetDeletionTimestamp().IsZero() {
+				mapAndEnqueue(ctx, natGateway, queue)
+			}
+		},
+	}
+}
+
+func (r *NetworkInterfaceNATReleaseReconciler) SetupWithManager(mgr ctrl.Manager) error {
+	return ctrl.NewControllerManagedBy(mgr).
+		For(&v1alpha1.NetworkInterface{}).
+		Watches(
+			&v1alpha1.NATGateway{},
+			r.enqueueByNATGateway(),
+		).
+		Complete(r)
+}

internal/controllers/networkinterfacenatrelease_controller_test.go

@@ -0,0 +1,103 @@
+// Copyright 2022 OnMetal authors
+//
+// 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 controllers
+
+import (
+	"github.com/onmetal/onmetal-api-net/api/core/v1alpha1"
+	"github.com/onmetal/onmetal-api-net/apimachinery/api/net"
+	. "github.com/onmetal/onmetal-api/utils/testing"
+	. "github.com/onsi/ginkgo/v2"
+	. "github.com/onsi/gomega"
+	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	. "sigs.k8s.io/controller-runtime/pkg/envtest/komega"
+)
+
+var _ = Describe("NetworkInterfaceNATReleaseReconciler", func() {
+	ns := SetupNamespace(&k8sClient)
+	network := SetupNetwork(ns)
+
+	It("should not release NATs that exist", func(ctx SpecContext) {
+		By("creating a network interface")
+		nic := &v1alpha1.NetworkInterface{
+			ObjectMeta: metav1.ObjectMeta{
+				Namespace:    ns.Name,
+				GenerateName: "nic-",
+			},
+			Spec: v1alpha1.NetworkInterfaceSpec{
+				NodeRef:    corev1.LocalObjectReference{Name: "my-node"},
+				NetworkRef: corev1.LocalObjectReference{Name: network.Name},
+				IPs:        []net.IP{net.MustParseIP("10.0.0.1")},
+			},
+		}
+		Expect(k8sClient.Create(ctx, nic)).To(Succeed())
+
+		By("creating a NAT gateway")
+		natGateway := &v1alpha1.NATGateway{
+			ObjectMeta: metav1.ObjectMeta{
+				Namespace:    ns.Name,
+				GenerateName: "nat-gateway-",
+			},
+			Spec: v1alpha1.NATGatewaySpec{
+				IPFamily:                 corev1.IPv4Protocol,
+				NetworkRef:               corev1.LocalObjectReference{Name: network.Name},
+				IPs:                      []v1alpha1.NATGatewayIP{{Name: "ip-1"}},
+				PortsPerNetworkInterface: 64,
+			},
+		}
+		Expect(k8sClient.Create(ctx, natGateway)).To(Succeed())
+
+		By("waiting for the network interface to have a NAT")
+		nat := v1alpha1.NetworkInterfaceNAT{
+			IPFamily: corev1.IPv4Protocol,
+			ClaimRef: v1alpha1.NetworkInterfaceNATClaimRef{
+				Name: natGateway.Name,
+				UID:  natGateway.UID,
+			},
+		}
+		Eventually(Object(nic)).Should(HaveField("Spec.NATs", ConsistOf(nat)))
+
+		By("ensuring it stays that way")
+		Consistently(Object(nic)).Should(HaveField("Spec.NATs", ConsistOf(nat)))
+	})
+
+	It("should release a NAT address of a non-existent NAT gateway", func(ctx SpecContext) {
+		By("creating a network interface")
+		nic := &v1alpha1.NetworkInterface{
+			ObjectMeta: metav1.ObjectMeta{
+				Namespace:    ns.Name,
+				GenerateName: "nic-",
+			},
+			Spec: v1alpha1.NetworkInterfaceSpec{
+				NodeRef:    corev1.LocalObjectReference{Name: "my-node"},
+				NetworkRef: corev1.LocalObjectReference{Name: network.Name},
+				IPs:        []net.IP{net.MustParseIP("10.0.0.1")},
+				NATs: []v1alpha1.NetworkInterfaceNAT{
+					{
+						IPFamily: corev1.IPv4Protocol,
+						ClaimRef: v1alpha1.NetworkInterfaceNATClaimRef{
+							Name: "should-not-exist",
+							UID:  "should-not-exist",
+						},
+					},
+				},
+			},
+		}
+		Expect(k8sClient.Create(ctx, nic)).To(Succeed())
+
+		By("waiting for the NAT to be released")
+		Eventually(Object(nic)).Should(HaveField("Spec.NATs", BeEmpty()))
+	})
+})