feat: ipam sync routine

Author: fra98

cmd/ipam/main.go

@@ -23,7 +23,6 @@ import (
 
 	"github.com/spf13/cobra"
 	"google.golang.org/grpc"
-	"google.golang.org/grpc/health"
 	"google.golang.org/grpc/health/grpc_health_v1"
 	corev1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/runtime"
@@ -71,6 +70,8 @@ func main() {
 	restcfg.InitFlags(cmd.Flags())
 
 	cmd.Flags().IntVar(&options.Port, "port", consts.IpamPort, "The port on which to listen for incoming gRPC requests.")
+	cmd.Flags().DurationVar(&options.SyncFrequency, "interval", consts.SyncFrequency,
+		"The interval at which the IPAM will synchronize the IPAM storage.")
 	cmd.Flags().BoolVar(&options.EnableLeaderElection, "leader-election", false, "Enable leader election for IPAM. "+
 		"Enabling this will ensure there is only one active IPAM.")
 	cmd.Flags().StringVar(&options.LeaderElectionNamespace, "leader-election-namespace", consts.DefaultLiqoNamespace,
@@ -102,14 +103,12 @@ func run(cmd *cobra.Command, _ []string) error {
 
 	// Get the rest config.
 	cfg := restcfg.SetRateLimiter(ctrl.GetConfigOrDie())
-	options.Config = cfg
 	cl, err := client.New(cfg, client.Options{
 		Scheme: scheme,
 	})
 	if err != nil {
 		return err
 	}
-	options.Client = cl
 
 	if options.EnableLeaderElection {
 		if leader, err := leaderelection.Blocking(ctx, cfg, record.NewBroadcaster(), &leaderelection.Opts{
@@ -129,10 +128,7 @@ func run(cmd *cobra.Command, _ []string) error {
 		}
 	}
 
-	hs := health.NewServer()
-	options.HealthServer = hs
-
-	liqoIPAM, err := ipam.New(ctx, &options)
+	liqoIPAM, err := ipam.New(ctx, cl, cfg, &options)
 	if err != nil {
 		return err
 	}
@@ -145,7 +141,7 @@ func run(cmd *cobra.Command, _ []string) error {
 	server := grpc.NewServer()
 
 	// Register health service
-	grpc_health_v1.RegisterHealthServer(server, hs)
+	grpc_health_v1.RegisterHealthServer(server, liqoIPAM.HealthServer)
 
 	// Register IPAM service
 	ipam.RegisterIPAMServer(server, liqoIPAM)

pkg/consts/ipam.go

@@ -14,12 +14,16 @@
 
 package consts
 
+import "time"
+
 // NetworkType indicates the type of Network.
 type NetworkType string
 
 const (
 	// IpamPort is the port used by the IPAM gRPC server.
-	IpamPort = 50051
+	IpamPort = 6000
+	// SyncFrequency is the frequency at which the IPAM should periodically sync its status.
+	SyncFrequency = 2 * time.Minute
 
 	// NetworkNotRemappedLabelKey is the label key used to mark a Network that does not need CIDR remapping.
 	NetworkNotRemappedLabelKey = "ipam.liqo.io/network-not-remapped"

pkg/ipam/initialize.go

@@ -18,19 +18,11 @@ import (
 	"context"
 
 	klog "k8s.io/klog/v2"
-
-	ipamv1alpha1 "github.com/liqotech/liqo/apis/ipam/v1alpha1"
-	"github.com/liqotech/liqo/pkg/consts"
 )
 
 // +kubebuilder:rbac:groups=ipam.liqo.io,resources=ips,verbs=get;list;watch
 // +kubebuilder:rbac:groups=ipam.liqo.io,resources=networks,verbs=get;list;watch
 
-type ipCidr struct {
-	ip   string
-	cidr string
-}
-
 func (lipam *LiqoIPAM) initialize(ctx context.Context) error {
 	if err := lipam.initializeNetworks(ctx); err != nil {
 		return err
@@ -77,71 +69,3 @@ func (lipam *LiqoIPAM) initializeIPs(ctx context.Context) error {
 
 	return nil
 }
-
-func (lipam *LiqoIPAM) getReservedNetworks(ctx context.Context) ([]string, error) {
-	var nets []string
-	var networks ipamv1alpha1.NetworkList
-	if err := lipam.Options.Client.List(ctx, &networks); err != nil {
-		return nil, err
-	}
-
-	for i := range networks.Items {
-		net := &networks.Items[i]
-
-		var cidr string
-		switch {
-		case net.Labels != nil && net.Labels[consts.NetworkNotRemappedLabelKey] == consts.NetworkNotRemappedLabelValue:
-			cidr = net.Spec.CIDR.String()
-		default:
-			cidr = net.Status.CIDR.String()
-		}
-		if cidr == "" {
-			klog.Warningf("Network %s has no CIDR", net.Name)
-			continue
-		}
-
-		nets = append(nets, cidr)
-	}
-
-	return nets, nil
-}
-
-func (lipam *LiqoIPAM) getReservedIPs(ctx context.Context) ([]ipCidr, error) {
-	var ips []ipCidr
-	var ipList ipamv1alpha1.IPList
-	if err := lipam.Options.Client.List(ctx, &ipList); err != nil {
-		return nil, err
-	}
-
-	for i := range ipList.Items {
-		ip := &ipList.Items[i]
-
-		address := ip.Status.IP.String()
-		if address == "" {
-			klog.Warningf("IP %s has no address", ip.Name)
-			continue
-		}
-
-		cidr := ip.Status.CIDR.String()
-		if cidr == "" {
-			klog.Warningf("IP %s has no CIDR", ip.Name)
-			continue
-		}
-
-		ips = append(ips, ipCidr{ip: address, cidr: cidr})
-	}
-
-	return ips, nil
-}
-
-func (lipam *LiqoIPAM) reserveNetwork(cidr string) error {
-	// TODO: Reserve the network.
-	klog.Infof("Reserved network %s", cidr)
-	return nil
-}
-
-func (lipam *LiqoIPAM) reserveIP(ip, cidr string) error {
-	// TODO: Reserve the IP.
-	klog.Infof("Reserved IP %s in network %s", ip, cidr)
-	return nil
-}

pkg/ipam/ipam.go

@@ -16,6 +16,7 @@ package ipam
 
 import (
 	"context"
+	"sync"
 	"time"
 
 	"google.golang.org/grpc/health"
@@ -27,15 +28,21 @@ import (
 // LiqoIPAM is the struct implementing the IPAM interface.
 type LiqoIPAM struct {
 	UnimplementedIPAMServer
+	HealthServer *health.Server
+
+	client.Client
+	*rest.Config
 
-	Options *Options
+	options       *Options
+	cacheNetworks map[string]networkInfo
+	cacheIPs      map[string]ipInfo
+	mutex         sync.Mutex
 }
 
 // Options contains the options to configure the IPAM.
 type Options struct {
-	Port   int
-	Config *rest.Config
-	Client client.Client
+	Port          int
+	SyncFrequency time.Duration
 
 	EnableLeaderElection    bool
 	LeaderElectionNamespace string
@@ -44,23 +51,34 @@ type Options struct {
 	RenewDeadline           time.Duration
 	RetryPeriod             time.Duration
 	PodName                 string
-
-	HealthServer *health.Server
 }
 
 // New creates a new instance of the LiqoIPAM.
-func New(ctx context.Context, opts *Options) (*LiqoIPAM, error) {
-	opts.HealthServer.SetServingStatus(IPAM_ServiceDesc.ServiceName, grpc_health_v1.HealthCheckResponse_NOT_SERVING)
+func New(ctx context.Context, cl client.Client, cfg *rest.Config, opts *Options) (*LiqoIPAM, error) {
+	hs := health.NewServer()
+	hs.SetServingStatus(IPAM_ServiceDesc.ServiceName, grpc_health_v1.HealthCheckResponse_NOT_SERVING)
 
 	lipam := &LiqoIPAM{
-		Options: opts,
+		HealthServer: hs,
+
+		Config: cfg,
+		Client: cl,
+
+		options:       opts,
+		cacheNetworks: make(map[string]networkInfo),
+		cacheIPs:      make(map[string]ipInfo),
 	}
 
+	// Initialize the IPAM instance
 	if err := lipam.initialize(ctx); err != nil {
 		return nil, err
 	}
 
-	opts.HealthServer.SetServingStatus(IPAM_ServiceDesc.ServiceName, grpc_health_v1.HealthCheckResponse_SERVING)
+	// Launch sync routine
+	go lipam.sync(ctx, opts.SyncFrequency)
+
+	hs.SetServingStatus(IPAM_ServiceDesc.ServiceName, grpc_health_v1.HealthCheckResponse_SERVING)
+
 	return lipam, nil
 }
 

pkg/ipam/ips.go

@@ -0,0 +1,91 @@
+// Copyright 2019-2024 The Liqo 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 ipam
+
+import (
+	"context"
+	"time"
+
+	klog "k8s.io/klog/v2"
+
+	ipamv1alpha1 "github.com/liqotech/liqo/apis/ipam/v1alpha1"
+)
+
+type ipInfo struct {
+	ipCidr
+	creationTimestamp time.Time
+}
+
+func (i *ipInfo) String() string {
+	return i.ipCidr.String()
+}
+
+type ipCidr struct {
+	ip   string
+	cidr string
+}
+
+func (i *ipCidr) String() string {
+	return i.ip + "-" + i.cidr
+}
+
+func (lipam *LiqoIPAM) reserveIP(ip, cidr string) error {
+	lipam.mutex.Lock()
+	defer lipam.mutex.Unlock()
+
+	if lipam.cacheIPs == nil {
+		lipam.cacheIPs = make(map[string]ipInfo)
+	}
+
+	// TODO: add correct logic.
+	ipI := ipInfo{
+		ipCidr:            ipCidr{ip: ip, cidr: cidr},
+		creationTimestamp: time.Now(),
+	}
+
+	// Save IP in cache.
+	lipam.cacheIPs[ipI.String()] = ipI
+	klog.Infof("Reserved IP %s in network %s", ip, cidr)
+
+	return nil
+}
+
+func (lipam *LiqoIPAM) getReservedIPs(ctx context.Context) ([]ipCidr, error) {
+	var ips []ipCidr
+	var ipList ipamv1alpha1.IPList
+	if err := lipam.Client.List(ctx, &ipList); err != nil {
+		return nil, err
+	}
+
+	for i := range ipList.Items {
+		ip := &ipList.Items[i]
+
+		address := ip.Status.IP.String()
+		if address == "" {
+			klog.Warningf("IP %s has no address", ip.Name)
+			continue
+		}
+
+		cidr := ip.Status.CIDR.String()
+		if cidr == "" {
+			klog.Warningf("IP %s has no CIDR", ip.Name)
+			continue
+		}
+
+		ips = append(ips, ipCidr{ip: address, cidr: cidr})
+	}
+
+	return ips, nil
+}