fix(kuma-cp) resync IPAM in case of Kuma CP restart (bp #1213)

pkg/dns/ip.go

@@ -2,13 +2,19 @@ package dns
 
 import (
 	"net"
+	"strings"
 
 	"github.com/Nordix/simple-ipam/pkg/ipam"
 )
 
 type IPAM interface {
 	AllocateIP() (string, error)
 	FreeIP(ip string) error
+	ReserveIP(ip string) error
+}
+
+func IsAddressAlreadyAllocated(err error) bool {
+	return err != nil && strings.HasPrefix(err.Error(), "Address already allocated")
 }
 
 type SimpleIPAM struct {
@@ -38,11 +44,15 @@ func (i *SimpleIPAM) FreeIP(ip string) error {
 
 	// ensure the IP is reserved before deleting it
 	err := i.Reserve(parsedIP)
-	if err != nil && err.Error() != "Address already allocated" {
+	if err != nil && !IsAddressAlreadyAllocated(err) {
 		return err
 	}
 
 	i.Free(parsedIP)
 
 	return nil
 }
+
+func (i *SimpleIPAM) ReserveIP(ip string) error {
+	return i.Reserve(net.ParseIP(ip))
+}

pkg/dns/vips/persistence.go

@@ -48,16 +48,17 @@ func NewPersistence(resourceManager manager.ReadOnlyResourceManager, configManag
 	}
 }
 
-func (m *Persistence) Get() (List, error) {
+func (m *Persistence) Get() (global List, meshed map[string]List, errs error) {
 	resourceList := &config_model.ConfigResourceList{}
 	if err := m.configManager.List(context.Background(), resourceList); err != nil {
-		return nil, err
+		return nil, nil, err
 	}
 
-	var errs error
-	vips := List{}
+	global = List{}
+	meshed = map[string]List{}
 	for _, resource := range resourceList.Items {
-		if _, ok := MeshFromConfigKey(resource.Meta.GetName()); !ok {
+		mesh, ok := MeshFromConfigKey(resource.Meta.GetName())
+		if !ok {
 			continue
 		}
 		if resource.Spec.Config == "" {
@@ -68,9 +69,10 @@ func (m *Persistence) Get() (List, error) {
 			errs = multierr.Append(errs, err)
 			continue
 		}
-		vips.Append(v)
+		global.Append(v)
+		meshed[mesh] = v
 	}
-	return vips, nil
+	return
 }
 
 func (m *Persistence) GetByMesh(mesh string) (List, error) {

pkg/dns/vips/persistence_test.go

@@ -109,7 +109,7 @@ var _ = Describe("Meshed Persistence", func() {
 		})
 
 		It("should merge vips from several configs", func() {
-			actual, err := meshedPersistence.Get()
+			actual, _, err := meshedPersistence.Get()
 			Expect(err).ToNot(HaveOccurred())
 
 			expected := vips.List{

pkg/dns/vips_allocator.go

@@ -23,25 +23,21 @@ var vipsAllocatorLog = core.Log.WithName("dns-vips-allocator")
 
 type VIPsAllocator struct {
 	rm          manager.ReadOnlyResourceManager
-	ipam        IPAM
 	persistence *vips.Persistence
 	resolver    resolver.DNSResolver
 	newTicker   func() *time.Ticker
+	cidr        string
 }
 
 // NewVIPsAllocator creates new object of VIPsAllocator. You can either
 // call method CreateOrUpdateVIPConfig manually or start VIPsAllocator as a component.
 // In the latter scenario it will call CreateOrUpdateVIPConfig every 'tickInterval'
 // for all meshes in the store.
 func NewVIPsAllocator(rm manager.ReadOnlyResourceManager, configManager config_manager.ConfigManager, cidr string, resolver resolver.DNSResolver) (*VIPsAllocator, error) {
-	ipam, err := NewSimpleIPAM(cidr)
-	if err != nil {
-		return nil, err
-	}
 	return &VIPsAllocator{
 		rm:          rm,
 		persistence: vips.NewPersistence(rm, configManager),
-		ipam:        ipam,
+		cidr:        cidr,
 		resolver:    resolver,
 		newTicker: func() *time.Ticker {
 			return time.NewTicker(tickInterval)
@@ -61,8 +57,8 @@ func (d *VIPsAllocator) Start(stop <-chan struct{}) error {
 	for {
 		select {
 		case <-ticker.C:
-			if err := d.createOrUpdateVIPConfigs(); err != nil {
-				vipsAllocatorLog.Error(err, "unable to create or update VIP configs")
+			if err := d.CreateOrUpdateVIPConfigs(); err != nil {
+				vipsAllocatorLog.Error(err, "errors during updating VIP configs")
 			}
 		case <-stop:
 			vipsAllocatorLog.Info("stopping")
@@ -71,49 +67,80 @@ func (d *VIPsAllocator) Start(stop <-chan struct{}) error {
 	}
 }
 
-func (d *VIPsAllocator) createOrUpdateVIPConfigs() error {
-	meshes := core_mesh.MeshResourceList{}
-	err := d.rm.List(context.Background(), &meshes)
-	if err != nil {
+func (d *VIPsAllocator) CreateOrUpdateVIPConfigs() error {
+	meshRes := core_mesh.MeshResourceList{}
+	if err := d.rm.List(context.Background(), &meshRes); err != nil {
 		return err
 	}
 
-	for _, mesh := range meshes.Items {
-		if err := d.CreateOrUpdateVIPConfig(mesh.GetMeta().GetName()); err != nil {
-			vipsAllocatorLog.Error(err, "unable to create or update VIP config", "mesh", mesh.GetMeta().GetName())
-		}
+	meshes := []string{}
+	for _, mesh := range meshRes.Items {
+		meshes = append(meshes, mesh.GetMeta().GetName())
 	}
-	return nil
+
+	return d.createOrUpdateVIPConfigs(meshes...)
 }
 
 func (d *VIPsAllocator) CreateOrUpdateVIPConfig(mesh string) error {
-	serviceSet, err := BuildServiceSet(d.rm, mesh)
+	return d.createOrUpdateVIPConfigs(mesh)
+}
+
+func (d *VIPsAllocator) createOrUpdateVIPConfigs(meshes ...string) (errs error) {
+	global, byMesh, err := d.persistence.Get()
 	if err != nil {
 		return err
 	}
 
-	global, err := d.persistence.Get()
+	ipam, err := d.newIPAM(global)
 	if err != nil {
 		return err
 	}
 
-	meshed, err := d.persistence.GetByMesh(mesh)
-	if err != nil {
-		return err
+	forEachMesh := func(mesh string, meshed vips.List) error {
+		serviceSet, err := BuildServiceSet(d.rm, mesh)
+		if err != nil {
+			return err
+		}
+
+		updated, updError := UpdateMeshedVIPs(global, meshed, ipam, serviceSet)
+		if !updated {
+			return updError
+		}
+
+		if err := d.persistence.Set(mesh, meshed); err != nil {
+			return multierr.Append(updError, err)
+		}
+
+		d.resolver.SetVIPs(meshed)
+		return updError
 	}
 
-	updated, updError := UpdateMeshedVIPs(global, meshed, d.ipam, serviceSet)
-	if !updated {
-		return err
+	for _, mesh := range meshes {
+		meshed, ok := byMesh[mesh]
+		if !ok {
+			meshed = vips.List{}
+		}
+		if err := forEachMesh(mesh, meshed); err != nil {
+			errs = multierr.Append(errs, errors.Wrapf(err, "errors during updating VIP config for mesh %s", mesh))
+		}
 	}
 
-	if err := d.persistence.Set(mesh, meshed); err != nil {
-		return multierr.Append(updError, err)
+	return errs
+}
+
+func (d *VIPsAllocator) newIPAM(initialVIPs vips.List) (IPAM, error) {
+	ipam, err := NewSimpleIPAM(d.cidr)
+	if err != nil {
+		return nil, err
 	}
 
-	d.resolver.SetVIPs(meshed)
+	for _, vip := range initialVIPs {
+		if err := ipam.ReserveIP(vip); err != nil && !IsAddressAlreadyAllocated(err) {
+			return nil, err
+		}
+	}
 
-	return updError
+	return ipam, nil
 }
 
 type ServiceSet map[string]bool
@@ -178,12 +205,14 @@ func UpdateMeshedVIPs(global, meshed vips.List, ipam IPAM, serviceSet ServiceSet
 		}
 		meshed[service] = ip
 		updated = true
+		vipsAllocatorLog.Info("adding", "service", service, "ip", ip)
 	}
 	for service, ip := range meshed {
 		if _, found := serviceSet[service]; !found {
 			updated = true
 			_ = ipam.FreeIP(ip)
 			delete(meshed, service)
+			vipsAllocatorLog.Info("deleting", "service", service, "ip", ip)
 		}
 	}
 	return