func (s *blockAssignState) findOrClaimBlock(ctx context.Context, minFreeIps int) (*model.KVPair, bool, error) {
	logCtx := log.WithFields(log.Fields{"host": s.host})
	for len(s.remainingAffineBlocks) > 0 {
		cidr := s.remainingAffineBlocks[0]
		s.remainingAffineBlocks = s.remainingAffineBlocks[1:]
		if s.reservations.MatchesWholeCIDR(&cidr) {
			log.WithField("cidr", cidr).Debug("Ignoring block, it is entirely reserved.")
			continue
		}
		for i := 0; i < datastoreRetries; i++ {
			logCtx.Infof("Trying affinity for %s", cidr)
			aff, err := s.client.blockReaderWriter.queryAffinity(ctx, s.host, cidr, "")
			if err != nil {
				logCtx.WithError(err).Warnf("Error getting affinity")
				break
			}
			b, err := s.client.getBlockFromAffinity(ctx, aff, s.hostReservedAttr)
			if err != nil {
				if _, ok := err.(cerrors.ErrorResourceUpdateConflict); ok {
					logCtx.WithError(err).Debug("CAS error getting affine block - retry")
					continue
				}
				logCtx.WithError(err).Warn("Couldn't get block for affinity, try next one")
				break
			}
			block := allocationBlock{b.Value.(*model.AllocationBlock)}
			numFreeAddresses := block.NumFreeAddresses(s.reservations)
			if numFreeAddresses >= minFreeIps {
				logCtx.Debugf("Block '%s' has %d free ips which is more than %d ips required.", cidr.String(), numFreeAddresses, minFreeIps)
				return b, false, nil
			} else {
				logCtx.Debugf("Block '%s' has %d free ips which is less than %d ips required.", cidr.String(), numFreeAddresses, minFreeIps)
				break
			}
		}
	}
	logCtx.Infof("Ran out of existing affine blocks for host")
	if !s.allowNewClaim {
		return nil, false, ErrBlockLimit
	}
	config, err := s.client.GetIPAMConfig(ctx)
	if err != nil {
		return nil, false, err
	}
	logCtx.Debugf("Allocate new blocks? Config: %+v", config)
	if config.AutoAllocateBlocks {
		for i := 0; i < datastoreRetries; i++ {
			logCtx.Info("Tried all affine blocks. Looking for an affine block with space, or a new unclaimed block")
			subnet, err := s.client.blockReaderWriter.findUsableBlock(ctx, s.host, s.version, s.pools, s.reservations, *config)
			if err != nil {
				if _, ok := err.(noFreeBlocksError); ok {
					logCtx.Info("No free blocks available for allocation")
				} else {
					log.WithError(err).Error("Failed to find an unclaimed block")
				}
				return nil, false, err
			}
			logCtx := log.WithFields(log.Fields{"host": s.host, "subnet": subnet})
			logCtx.Info("Found unclaimed block")
			for j := 0; j < datastoreRetries; j++ {
				pa, err := s.client.blockReaderWriter.getPendingAffinity(ctx, s.host, *subnet)
				if err != nil {
					if _, ok := err.(cerrors.ErrorResourceUpdateConflict); ok {
						logCtx.WithError(err).Debug("CAS error claiming pending affinity, retry")
						continue
					}
					logCtx.WithError(err).Errorf("Error claiming pending affinity")
					return nil, false, err
				}
				b, err := s.client.getBlockFromAffinity(ctx, pa, s.hostReservedAttr)
				if err != nil {
					if _, ok := err.(cerrors.ErrorResourceUpdateConflict); ok {
						logCtx.WithError(err).Debug("CAS error getting block, retry")
						continue
					} else if _, ok := err.(errBlockClaimConflict); ok {
						logCtx.WithError(err).Debug("Block taken by someone else, find a new one")
						break
					} else if _, ok := err.(errStaleAffinity); ok {
						logCtx.WithError(err).Debug("Affinity is stale, find a new one")
						break
					}
					logCtx.WithError(err).Errorf("Error getting block for affinity")
					return nil, false, err
				}
				block := allocationBlock{b.Value.(*model.AllocationBlock)}
				numFree := block.NumFreeAddresses(s.reservations)
				if numFree >= minFreeIps {
					logCtx.Infof("Block '%s' has %d free ips which is more than %d ips required.",
						b.Key.(model.BlockKey).CIDR, numFree, minFreeIps)
					return b, true, nil
				} else {
					errString := fmt.Sprintf("Block '%s' has %d free ips which is less than %d ips required.",
						b.Key.(model.BlockKey).CIDR, numFree, minFreeIps)
					logCtx.Errorf(errString)
					return nil, false, errors.New(errString)
				}
			}
			s.datastoreRetryCount++
		}
		return nil, false, errors.New("Max retries hit - excessive concurrent IPAM requests")
	}
	return nil, false, errors.New("failed to find or claim a block")
}
Given a list of host-affine blocks, findOrClaimBlock returns one block with minimum free ips.
It tries to use one of the current host-affine blocks first and if not found, it will claim a new block
and assign affinity.
It returns a block, a boolean if block is newly claimed and any error encountered.
package ipam
import (
	"context"
	"errors"
	"fmt"
	"math/bits"
	"runtime"
	"strings"
	log "github.com/sirupsen/logrus"
	"golang.org/x/sync/semaphore"
	v3 "github.com/projectcalico/api/pkg/apis/projectcalico/v3"
	libapiv3 "github.com/projectcalico/libcalico-go/lib/apis/v3"
	bapi "github.com/projectcalico/libcalico-go/lib/backend/api"
	"github.com/projectcalico/libcalico-go/lib/backend/model"
	cerrors "github.com/projectcalico/libcalico-go/lib/errors"
	"github.com/projectcalico/libcalico-go/lib/names"
	"github.com/projectcalico/libcalico-go/lib/net"
	"github.com/projectcalico/libcalico-go/lib/options"
	"github.com/projectcalico/libcalico-go/lib/set"
)
const (
	datastoreRetries  = 100
	ipamKeyErrRetries = 3
	AttributePod           = model.IPAMBlockAttributePod
	AttributeNamespace     = model.IPAMBlockAttributeNamespace
	AttributeNode          = model.IPAMBlockAttributeNode
	AttributeTimestamp     = model.IPAMBlockAttributeTimestamp
	AttributeType          = model.IPAMBlockAttributeType
	AttributeTypeIPIP      = model.IPAMBlockAttributeTypeIPIP
	AttributeTypeVXLAN     = model.IPAMBlockAttributeTypeVXLAN
	AttributeTypeWireguard = model.IPAMBlockAttributeTypeWireguard
)
var (
	ErrBlockLimit      = errors.New("cannot allocate new block due to per host block limit")
	ErrNoQualifiedPool = errors.New("cannot find a qualified ippool")
	ErrStrictAffinity  = errors.New("global strict affinity should not be false for Windows node")
)
type IPReservationInterface interface {
	List(ctx context.Context, opts options.ListOptions) (*v3.IPReservationList, error)
}
func NewIPAMClient(client bapi.Client, pools PoolAccessorInterface, reservations IPReservationInterface) Interface {
	return &ipamClient{
		client:       client,
		pools:        pools,
		reservations: reservations,
		blockReaderWriter: blockReaderWriter{
			client: client,
			pools:  pools,
		},
	}
}
type ipamClient struct {
	client            bapi.Client
	pools             PoolAccessorInterface
	blockReaderWriter blockReaderWriter
	reservations      IPReservationInterface
}
func (c ipamClient) AutoAssign(ctx context.Context, args AutoAssignArgs) (*IPAMAssignments, *IPAMAssignments, error) {
	hostname, err := decideHostname(args.Hostname)
	if err != nil {
		return nil, nil, err
	}
	log.Infof("Auto-assign %d ipv4, %d ipv6 addrs for host '%s'", args.Num4, args.Num6, hostname)
	var v4ia, v6ia *IPAMAssignments
	if args.Num4 != 0 {
		log.Debugf("Assigning IPv4 addresses")
		for _, pool := range args.IPv4Pools {
			if pool.IP.To4() == nil {
				return nil, nil, fmt.Errorf("provided IPv4 IPPools list contains one or more IPv6 IPPools")
			}
		}
		v4ia, err = c.autoAssign(ctx, args.Num4, args.HandleID, args.Attrs, args.IPv4Pools, 4, hostname, args.MaxBlocksPerHost, args.HostReservedAttrIPv4s, args.IntendedUse)
		if err != nil {
			log.Errorf("Error assigning IPV4 addresses: %v", err)
			return v4ia, nil, err
		}
	}
	if args.Num6 != 0 {
		log.Debugf("Assigning IPv6 addresses")
		for _, pool := range args.IPv6Pools {
			if pool.IP.To4() != nil {
				return nil, nil, fmt.Errorf("provided IPv6 IPPools list contains one or more IPv4 IPPools")
			}
		}
		v6ia, err = c.autoAssign(ctx, args.Num6, args.HandleID, args.Attrs, args.IPv6Pools, 6, hostname, args.MaxBlocksPerHost, args.HostReservedAttrIPv6s, args.IntendedUse)
		if err != nil {
			log.Errorf("Error assigning IPV6 addresses: %v", err)
			return v4ia, v6ia, err
		}
	}
	return v4ia, v6ia, nil
}
func (c ipamClient) getBlockFromAffinity(ctx context.Context, aff *model.KVPair, rsvdAttr *HostReservedAttr) (*model.KVPair, error) {
	cidr := aff.Key.(model.BlockAffinityKey).CIDR
	host := aff.Key.(model.BlockAffinityKey).Host
	state := aff.Value.(*model.BlockAffinity).State
	logCtx := log.WithFields(log.Fields{"host": host, "cidr": cidr})
	logCtx.Info("Attempting to load block")
	b, err := c.blockReaderWriter.queryBlock(ctx, cidr, "")
	if err != nil {
		if _, ok := err.(cerrors.ErrorResourceDoesNotExist); ok {
			logCtx.Info("The referenced block doesn't exist, trying to create it")
			aff.Value.(*model.BlockAffinity).State = model.StatePending
			aff, err = c.blockReaderWriter.updateAffinity(ctx, aff)
			if err != nil {
				logCtx.WithError(err).Warn("Error updating block affinity")
				return nil, err
			}
			logCtx.Info("Wrote affinity as pending")
			cfg, err := c.GetIPAMConfig(ctx)
			if err != nil {
				logCtx.WithError(err).Errorf("Error getting IPAM Config")
				return nil, err
			}
			logCtx.Info("Attempting to claim the block")
			b, err := c.blockReaderWriter.claimAffineBlock(ctx, aff, *cfg, rsvdAttr)
			if err != nil {
				logCtx.WithError(err).Warn("Error claiming block")
				return nil, err
			}
			return b, nil
		}
		logCtx.WithError(err).Error("Error getting block")
		return nil, err
	}
	blockAffinity := b.Value.(*model.AllocationBlock).Affinity
	if blockAffinity == nil || *blockAffinity != fmt.Sprintf("host:%s", host) {
		logCtx.WithField("blockAffinity", blockAffinity).Warn("Block does not match the provided affinity, deleting stale affinity")
		err := c.blockReaderWriter.deleteAffinity(ctx, aff)
		if err != nil {
			logCtx.WithError(err).Warn("Error deleting stale affinity")
			return nil, err
		}
		return nil, errStaleAffinity(fmt.Sprintf("Affinity is stale: %+v", aff))
	}
	if state != model.StateConfirmed && state != "" {
		logCtx.Info("Affinity has not been confirmed - attempt to confirm it")
		aff.Value.(*model.BlockAffinity).State = model.StatePending
		aff, err = c.blockReaderWriter.updateAffinity(ctx, aff)
		if err != nil {
			logCtx.WithError(err).Warn("Error marking affinity as pending as part of confirmation process")
			return nil, err
		}
		logCtx.Info("Writing block to get a new revision")
		b, err = c.blockReaderWriter.updateBlock(ctx, b)
		if err != nil {
			logCtx.WithError(err).Debug("Error writing block")
			return nil, err
		}
		logCtx.Info("Attempting to confirm affinity")
		aff.Value.(*model.BlockAffinity).State = model.StateConfirmed
		aff, err = c.blockReaderWriter.updateAffinity(ctx, aff)
		if err != nil {
			logCtx.WithError(err).Debug("Error confirming affinity")
			return nil, err
		}
		logCtx.Info("Affinity confirmed successfully")
	}
	logCtx.Info("Affinity is confirmed and block has been loaded")
	return b, nil
}
func detectOS(ctx context.Context) string {
	if osOverride := ctx.Value("windowsHost"); osOverride != nil {
		return osOverride.(string)
	}
	return runtime.GOOS
}
func (c ipamClient) determinePools(ctx context.Context, requestedPoolNets []net.IPNet, version int, node libapiv3.Node, maxPrefixLen int) (matchingPools, enabledPools []v3.IPPool, err error) {
	enabledPools, err = c.pools.GetEnabledPools(version)
	if err != nil {
		log.WithError(err).Errorf("Error getting IP pools")
		return
	}
	log.Debugf("enabled pools: %v", enabledPools)
	log.Debugf("requested pools: %v", requestedPoolNets)
	pm := map[string]v3.IPPool{}
	var cidr *net.IPNet
	for _, p := range enabledPools {
		if p.Spec.BlockSize > maxPrefixLen {
			log.Warningf("skipping pool %v due to blockSize %d bigger than %d", p, p.Spec.BlockSize, maxPrefixLen)
			continue
		}
		_, cidr, err = net.ParseCIDR(p.Spec.CIDR)
		if err != nil {
			log.WithError(err).Errorf("Pool %s has invalid CIDR %s", p.Name, p.Spec.CIDR)
			return
		}
		pm[cidr.String()] = p
	}
	if len(pm) == 0 {
		err = ErrNoQualifiedPool
		return
	}
	requestedPools := []v3.IPPool{}
	for _, rp := range requestedPoolNets {
		cidr := rp.Network()
		if pool, ok := pm[cidr.String()]; !ok {
			err = fmt.Errorf("the given pool (%s) does not exist, or is not enabled", cidr.String())
			return
		} else {
			requestedPools = append(requestedPools, pool)
		}
	}
	if len(requestedPools) > 0 {
		log.Debugf("Using the requested IP pools")
		matchingPools = requestedPools
		return
	}
	for _, pool := range enabledPools {
		var matches bool
		matches, err = SelectsNode(pool, node)
		if err != nil {
			log.WithError(err).WithField("pool", pool).Error("failed to determine if node matches pool")
			return
		}
		if !matches {
			log.Debugf("IP pool does not match this node: %s", pool.Name)
			continue
		}
		log.Debugf("IP pool matches this node: %s", pool.Name)
		matchingPools = append(matchingPools, pool)
	}
	return
}
func (c ipamClient) prepareAffinityBlocksForHost(ctx context.Context, requestedPools []net.IPNet, version int, host string, rsvdAttr *HostReservedAttr, use v3.IPPoolAllowedUse) ([]v3.IPPool, []net.IPNet, error) {
	node, err := c.client.Get(ctx, model.ResourceKey{Kind: libapiv3.KindNode, Name: host}, "")
	if err != nil {
		log.WithError(err).WithField("node", host).Error("failed to get node for host")
		return nil, nil, err
	}
	v3n, ok := node.Value.(*libapiv3.Node)
	if !ok {
		return nil, nil, fmt.Errorf("Datastore returned malformed node object")
	}
	maxPrefixLen, err := getMaxPrefixLen(version, rsvdAttr)
	if err != nil {
		return nil, nil, err
	}
	poolsSelectingNode, allPools, err := c.determinePools(ctx, requestedPools, version, *v3n, maxPrefixLen)
	if err != nil {
		return nil, nil, err
	}
	if len(poolsSelectingNode) == 0 {
		return nil, nil, fmt.Errorf("no configured Calico pools for node %s", host)
	}
	poolsAllowedByUse := filterPoolsByUse(poolsSelectingNode, use)
	log.Debugf("Pools filtered by allowed use: %v", poolsAllowedByUse)
	if len(poolsAllowedByUse) == 0 {
		return nil, nil, fmt.Errorf("%w, no pools match the required use (%v)", ErrNoQualifiedPool, use)
	}
	logCtx := log.WithFields(log.Fields{"host": host})
	logCtx.Info("Looking up existing affinities for host")
	allAffBlocks, err := c.blockReaderWriter.getAffineBlocks(ctx, host, version)
	if err != nil {
		return nil, nil, err
	}
	allowedAffBlocks, nonAllowedAffBlocks, err := filterBlocksByPools(allAffBlocks, poolsAllowedByUse)
	if err != nil {
		return nil, nil, err
	}
	_, affBlocksToRelease, err := filterBlocksByPools(nonAllowedAffBlocks, poolsSelectingNode)
	if err != nil {
		return nil, nil, err
	}
	for _, block := range affBlocksToRelease {
		pool, err := c.blockReaderWriter.getPoolForIP(net.IP{IP: block.IP}, allPools)
		if err != nil {
			log.WithError(err).Warnf("Failed to get pool for IP")
			continue
		}
		if pool == nil {
			logCtx.WithFields(log.Fields{"block": block}).Warn("No pool found for block, skipping")
			continue
		}
		blockSelectsNode, err := SelectsNode(*pool, *v3n)
		if err != nil {
			logCtx.WithError(err).WithField("pool", pool).Error("Failed to determine if node matches pool, skipping")
			continue
		}
		if blockSelectsNode {
			logCtx.WithFields(log.Fields{"pool": pool, "block": block}).Debug("Block's pool still selects node, refusing to remove affinity")
			continue
		}
		for i := 0; i < datastoreRetries; i++ {
			if err = c.blockReaderWriter.releaseBlockAffinity(ctx, host, block, true); err != nil {
				if _, ok := err.(errBlockClaimConflict); ok {
				} else if _, ok := err.(errBlockNotEmpty); ok {
				} else if _, ok := err.(cerrors.ErrorResourceDoesNotExist); ok {
				} else {
					logCtx.WithError(err).WithField("block", block).Warn("Error occurred releasing block, trying again")
					continue
				}
			}
			logCtx.WithField("block", block).Info("Released affine block that no longer selects this host")
			break
		}
	}
	return poolsAllowedByUse, allowedAffBlocks, nil
}
func filterPoolsByUse(pools []v3.IPPool, use v3.IPPoolAllowedUse) []v3.IPPool {
	var filteredPools []v3.IPPool
	for _, p := range pools {
		for _, allowed := range p.Spec.AllowedUses {
			if allowed == use {
				filteredPools = append(filteredPools, p)
				break
			}
		}
	}
	return filteredPools
}
type blockAssignState struct {
	client                ipamClient
	version               int
	host                  string
	pools                 []v3.IPPool
	remainingAffineBlocks []net.IPNet
	hostReservedAttr      *HostReservedAttr
	allowNewClaim         bool
	reservations          addrFilter
	datastoreRetryCount int
}