[1.14] ipam: Support for static IP allocation in AWS (#569)

Author: antonipp

pkg/alibabacloud/eni/node.go

@@ -306,6 +306,11 @@ func (n *Node) AllocateIPs(ctx context.Context, a *ipam.AllocationAction) error
 	return err
 }
 
+func (n *Node) AllocateStaticIP(ctx context.Context, staticIPTags ipamTypes.Tags) (string, error) {
+	// TODO
+	return "", nil
+}
+
 // PrepareIPRelease prepares the release of ENI IPs.
 func (n *Node) PrepareIPRelease(excessIPs int, scopedLog *logrus.Entry) *ipam.ReleaseAction {
 	r := &ipam.ReleaseAction{}

pkg/aws/ec2/ec2.go

@@ -14,7 +14,6 @@ import (
 	"github.com/aws/aws-sdk-go-v2/feature/ec2/imds"
 	"github.com/aws/aws-sdk-go-v2/service/ec2"
 	ec2_types "github.com/aws/aws-sdk-go-v2/service/ec2/types"
-	log "github.com/sirupsen/logrus"
 
 	"github.com/cilium/cilium/pkg/api/helpers"
 	"github.com/cilium/cilium/pkg/aws/endpoints"
@@ -24,6 +23,8 @@ import (
 	ipPkg "github.com/cilium/cilium/pkg/ip"
 	"github.com/cilium/cilium/pkg/ipam/option"
 	ipamTypes "github.com/cilium/cilium/pkg/ipam/types"
+	"github.com/cilium/cilium/pkg/logging"
+	"github.com/cilium/cilium/pkg/logging/logfields"
 	"github.com/cilium/cilium/pkg/spanstat"
 )
 
@@ -39,6 +40,8 @@ const (
 	InvalidParameterValueStr = "InvalidParameterValue"
 )
 
+var log = logging.DefaultLogger.WithField(logfields.LogSubsys, "ec2")
+
 // Client represents an EC2 API client
 type Client struct {
 	ec2Client           *ec2.Client
@@ -388,6 +391,10 @@ func parseENI(iface *ec2_types.NetworkInterface, vpcs ipamTypes.VirtualNetworkMa
 		eni.Prefixes = append(eni.Prefixes, aws.ToString(prefix.Ipv4Prefix))
 	}
 
+	if iface.Association != nil && aws.ToString(iface.Association.PublicIp) != "" {
+		eni.PublicIP = aws.ToString(iface.Association.PublicIp)
+	}
+
 	for _, g := range iface.Groups {
 		if g.GroupId != nil {
 			eni.SecurityGroups = append(eni.SecurityGroups, aws.ToString(g.GroupId))
@@ -691,6 +698,56 @@ func (c *Client) UnassignENIPrefixes(ctx context.Context, eniID string, prefixes
 	return err
 }
 
+func (c *Client) AssociateEIP(ctx context.Context, instanceID string, eipTags ipamTypes.Tags) (string, error) {
+	if len(eipTags) == 0 {
+		return "", fmt.Errorf("no EIP tags were provided")
+	}
+
+	filters := make([]ec2_types.Filter, 0, len(eipTags))
+	for k, v := range eipTags {
+		filters = append(filters, ec2_types.Filter{
+			Name:   aws.String(fmt.Sprintf("tag:%s", k)),
+			Values: []string{v},
+		})
+	}
+
+	describeAddressesInput := &ec2.DescribeAddressesInput{
+		Filters: filters,
+	}
+	c.limiter.Limit(ctx, "DescribeAddresses")
+	sinceStart := spanstat.Start()
+	addresses, err := c.ec2Client.DescribeAddresses(ctx, describeAddressesInput)
+	c.metricsAPI.ObserveAPICall("DescribeAddresses", deriveStatus(err), sinceStart.Seconds())
+	if err != nil {
+		return "", err
+	}
+	log.Infof("Found %d EIPs corresponding to tags %v", len(addresses.Addresses), eipTags)
+
+	for _, address := range addresses.Addresses {
+		// Only pick unassociated EIPs
+		if address.AssociationId == nil {
+			associateAddressInput := &ec2.AssociateAddressInput{
+				AllocationId:       address.AllocationId,
+				AllowReassociation: aws.Bool(false),
+				InstanceId:         aws.String(instanceID),
+			}
+			c.limiter.Limit(ctx, "AssociateAddress")
+			sinceStart = spanstat.Start()
+			association, err := c.ec2Client.AssociateAddress(ctx, associateAddressInput)
+			c.metricsAPI.ObserveAPICall("AssociateAddress", deriveStatus(err), sinceStart.Seconds())
+			if err != nil {
+				// In case there is an issue with associating the EIP, this will be retried in the next
+				// iteration of the reconciliation loop
+				return "", err
+			}
+			log.Infof("Associated EIP %s with instance %s (association ID: %s)", *address.PublicIp, instanceID, *association.AssociationId)
+			return *address.PublicIp, nil
+		}
+	}
+
+	return "", fmt.Errorf("no unassociated EIPs found for tags %v", eipTags)
+}
+
 func createAWSTagSlice(tags map[string]string) []ec2_types.Tag {
 	awsTags := make([]ec2_types.Tag, 0, len(tags))
 	for k, v := range tags {

pkg/aws/ec2/mock/mock.go

@@ -41,6 +41,7 @@ const (
 	AssignPrivateIpAddresses
 	UnassignPrivateIpAddresses
 	TagENI
+	AssociateEIP
 	MaxOperation
 )
 
@@ -531,6 +532,34 @@ func (e *API) UnassignENIPrefixes(ctx context.Context, eniID string, prefixes []
 	return fmt.Errorf("Unable to find ENI with ID %s", eniID)
 }
 
+func (e *API) AssociateEIP(ctx context.Context, instanceID string, eipTags ipamTypes.Tags) (string, error) {
+	e.rateLimit()
+	e.delaySim.Delay(AssociateEIP)
+
+	e.mutex.Lock()
+	defer e.mutex.Unlock()
+
+	if err, ok := e.errors[AssociateEIP]; ok {
+		return "", err
+	}
+
+	ipAddr := "192.0.2.254"
+
+	// Assign the EIP to the ENI 0 of the instance
+	for iid, enis := range e.enis {
+		if iid == instanceID {
+			for _, eni := range enis {
+				if eni.Number == 0 {
+					eni.PublicIP = ipAddr
+					return ipAddr, nil
+				}
+			}
+		}
+	}
+
+	return "", fmt.Errorf("unable to find ENI 0 for instance %s", instanceID)
+}
+
 func (e *API) GetInstances(ctx context.Context, vpcs ipamTypes.VirtualNetworkMap, subnets ipamTypes.SubnetMap) (*ipamTypes.InstanceMap, error) {
 	instances := ipamTypes.NewInstanceMap()
 

pkg/aws/eni/instances.go

@@ -34,6 +34,7 @@ type EC2API interface {
 	UnassignPrivateIpAddresses(ctx context.Context, eniID string, addresses []string) error
 	AssignENIPrefixes(ctx context.Context, eniID string, prefixes int32) error
 	UnassignENIPrefixes(ctx context.Context, eniID string, prefixes []string) error
+	AssociateEIP(ctx context.Context, instanceID string, eipTags ipamTypes.Tags) (string, error)
 }
 
 // InstancesManager maintains the list of instances. It must be kept up to date

pkg/aws/eni/node.go

@@ -295,6 +295,10 @@ func (n *Node) AllocateIPs(ctx context.Context, a *ipam.AllocationAction) error
 	return n.manager.api.AssignPrivateIpAddresses(ctx, a.InterfaceID, int32(a.AvailableForAllocation))
 }
 
+func (n *Node) AllocateStaticIP(ctx context.Context, staticIPTags ipamTypes.Tags) (string, error) {
+	return n.manager.api.AssociateEIP(ctx, n.node.InstanceID(), staticIPTags)
+}
+
 func (n *Node) getSecurityGroupIDs(ctx context.Context, eniSpec eniTypes.ENISpec) ([]string, error) {
 	// 1. check explicit security groups associations via checking Spec.ENI.SecurityGroups
 	// 2. check if Spec.ENI.SecurityGroupTags is passed and if so filter by those
@@ -600,6 +604,11 @@ func (n *Node) ResyncInterfacesAndIPs(ctx context.Context, scopedLog *logrus.Ent
 			for _, ip := range e.Addresses {
 				available[ip] = ipamTypes.AllocationIP{Resource: e.ID}
 			}
+
+			if e.Number == 0 && e.PublicIP != "" {
+				stats.AssignedStaticIP = e.PublicIP
+			}
+
 			return nil
 		})
 	enis := len(n.enis)

pkg/aws/eni/node_manager_test.go

@@ -844,6 +844,84 @@ func (e *ENISuite) TestInstanceBeenDeleted(c *check.C) {
 	c.Assert(node.Stats().ExcessIPs, check.Equals, 0)
 }
 
+// TestNodeManagerStaticIP tests allocation with a static IP
+//
+// - m5.large (3x ENIs, 2x10-2 IPs)
+// - MinAllocate 0
+// - MaxAllocate 0
+// - PreAllocate 8
+// - FirstInterfaceIndex 0
+func (e *ENISuite) TestNodeManagerStaticIP(c *check.C) {
+	const instanceID = "i-testNodeManagerStaticIP-0"
+
+	ec2api := ec2mock.NewAPI([]*ipamTypes.Subnet{testSubnet}, []*ipamTypes.VirtualNetwork{testVpc}, testSecurityGroups)
+	instances := NewInstancesManager(ec2api)
+	c.Assert(instances, check.Not(check.IsNil))
+	eniID1, _, err := ec2api.CreateNetworkInterface(context.TODO(), 0, "s-1", "desc", []string{"sg1", "sg2"}, false)
+	c.Assert(err, check.IsNil)
+	_, err = ec2api.AttachNetworkInterface(context.TODO(), 0, instanceID, eniID1)
+	c.Assert(err, check.IsNil)
+	instances.Resync(context.TODO())
+	mngr, err := ipam.NewNodeManager(instances, k8sapi, metricsapi, 10, false, false)
+	c.Assert(err, check.IsNil)
+	c.Assert(mngr, check.Not(check.IsNil))
+
+	staticIPTags := map[string]string{"some-eip-tag": "some-value"}
+	cn := newCiliumNode("node1", withTestDefaults(), withInstanceID(instanceID), withInstanceType("m5.large"), withIPAMPreAllocate(8), withIPAMStaticIPTags(staticIPTags))
+	mngr.Upsert(cn)
+	c.Assert(testutils.WaitUntil(func() bool { return reachedAddressesNeeded(mngr, "node1", 0) }, 5*time.Second), check.IsNil)
+
+	node := mngr.Get("node1")
+	c.Assert(node, check.Not(check.IsNil))
+	c.Assert(node.Stats().AvailableIPs, check.Equals, 8)
+	c.Assert(node.Stats().UsedIPs, check.Equals, 0)
+
+	// Use 1 IP
+	mngr.Upsert(updateCiliumNode(cn, 8, 1))
+	c.Assert(testutils.WaitUntil(func() bool { return reachedAddressesNeeded(mngr, "node1", 0) }, 5*time.Second), check.IsNil)
+
+	node = mngr.Get("node1")
+	c.Assert(node, check.Not(check.IsNil))
+	// Verify that the static IP has been successfully assigned
+	c.Assert(node.Stats().AssignedStaticIP, check.Equals, "192.0.2.254")
+}
+
+// TestNodeManagerStaticIPAlreadyAssociated verifies that when an ENI already has a public IP assigned to it, it is properly detected
+//
+// - m5.large (3x ENIs, 2x10-2 IPs)
+// - MinAllocate 0
+// - MaxAllocate 0
+// - PreAllocate 8
+// - FirstInterfaceIndex 0
+func (e *ENISuite) TestNodeManagerStaticIPAlreadyAssociated(c *check.C) {
+	const instanceID = "i-testNodeManagerStaticIPAlreadyAssociated-0"
+
+	ec2api := ec2mock.NewAPI([]*ipamTypes.Subnet{testSubnet}, []*ipamTypes.VirtualNetwork{testVpc}, testSecurityGroups)
+	instances := NewInstancesManager(ec2api)
+	c.Assert(instances, check.Not(check.IsNil))
+	eniID1, _, err := ec2api.CreateNetworkInterface(context.TODO(), 0, "s-1", "desc", []string{"sg1", "sg2"}, false)
+	c.Assert(err, check.IsNil)
+	_, err = ec2api.AttachNetworkInterface(context.TODO(), 0, instanceID, eniID1)
+	c.Assert(err, check.IsNil)
+	staticIP, err := ec2api.AssociateEIP(context.TODO(), instanceID, make(ipamTypes.Tags))
+	c.Assert(err, check.IsNil)
+	instances.Resync(context.TODO())
+	mngr, err := ipam.NewNodeManager(instances, k8sapi, metricsapi, 10, false, false)
+	c.Assert(err, check.IsNil)
+	c.Assert(mngr, check.Not(check.IsNil))
+
+	cn := newCiliumNode("node1", withTestDefaults(), withInstanceID(instanceID), withInstanceType("m5.large"), withIPAMPreAllocate(8))
+	mngr.Upsert(cn)
+	c.Assert(testutils.WaitUntil(func() bool { return reachedAddressesNeeded(mngr, "node1", 0) }, 5*time.Second), check.IsNil)
+
+	node := mngr.Get("node1")
+	c.Assert(node, check.Not(check.IsNil))
+	c.Assert(node.Stats().AvailableIPs, check.Equals, 8)
+	c.Assert(node.Stats().UsedIPs, check.Equals, 0)
+	// Verify that the static IP which has already been assigned to the ENI has been successfully detected
+	c.Assert(node.Stats().AssignedStaticIP, check.Equals, staticIP)
+}
+
 func benchmarkAllocWorker(c *check.C, workers int64, delay time.Duration, rateLimit float64, burst int) {
 	testSubnet1 := &ipamTypes.Subnet{ID: "s-1", AvailabilityZone: "us-west-1", VirtualNetworkID: "vpc-1", AvailableAddresses: 1000000}
 	testSubnet2 := &ipamTypes.Subnet{ID: "s-2", AvailabilityZone: "us-west-1", VirtualNetworkID: "vpc-1", AvailableAddresses: 1000000}
@@ -1004,6 +1082,12 @@ func withIPAMMaxAboveWatermark(aboveWM int) func(*v2.CiliumNode) {
 	}
 }
 
+func withIPAMStaticIPTags(tags map[string]string) func(*v2.CiliumNode) {
+	return func(cn *v2.CiliumNode) {
+		cn.Spec.IPAM.StaticIPTags = tags
+	}
+}
+
 func withExcludeInterfaceTags(tags map[string]string) func(*v2.CiliumNode) {
 	return func(cn *v2.CiliumNode) {
 		cn.Spec.ENI.ExcludeInterfaceTags = tags

pkg/aws/eni/types/types.go

@@ -206,6 +206,9 @@ type ENI struct {
 	//
 	// +optional
 	Tags map[string]string `json:"tags,omitempty"`
+
+	// +optional
+	PublicIP string `json:"public-ip,omitempty"`
 }
 
 func (e *ENI) DeepCopyInterface() types.Interface {

pkg/azure/ipam/node.go

@@ -130,6 +130,11 @@ func (n *Node) AllocateIPs(ctx context.Context, a *ipam.AllocationAction) error
 	}
 }
 
+func (n *Node) AllocateStaticIP(ctx context.Context, staticIPTags ipamTypes.Tags) (string, error) {
+	// TODO
+	return "", nil
+}
+
 // CreateInterface is called to create a new interface. This operation is
 // currently not supported on Azure.
 func (n *Node) CreateInterface(ctx context.Context, allocation *ipam.AllocationAction, scopedLog *logrus.Entry) (int, string, error) {

pkg/ipam/node.go

@@ -167,6 +167,8 @@ type Statistics struct {
 	// EmptyInterfaceSlots is the number of empty interface slots available
 	// for interfaces to be attached
 	EmptyInterfaceSlots int
+
+	AssignedStaticIP string
 }
 
 // IsRunning returns true if the node is considered to be running
@@ -264,6 +266,13 @@ func (n *Node) getMaxAllocate() int {
 	return instanceMax
 }
 
+func (n *Node) getStaticIPTags() ipamTypes.Tags {
+	if n.resource.Spec.IPAM.StaticIPTags != nil {
+		return n.resource.Spec.IPAM.StaticIPTags
+	}
+	return ipamTypes.Tags{}
+}
+
 // GetNeededAddresses returns the number of needed addresses that need to be
 // allocated or released. A positive number is returned to indicate allocation.
 // A negative number is returned to indicate release of addresses.
@@ -431,6 +440,10 @@ func (n *Node) recalculate() {
 	}
 
 	n.available = a
+	if stats.AssignedStaticIP != "" {
+		n.stats.AssignedStaticIP = stats.AssignedStaticIP
+	}
+	n.stats.AssignedStaticIP = stats.AssignedStaticIP
 	n.stats.UsedIPs = len(n.resource.Status.IPAM.Used)
 
 	// Get used IP count with prefixes included
@@ -887,6 +900,16 @@ func (n *Node) maintainIPPool(ctx context.Context) (instanceMutated bool, err er
 		n.removeStaleReleaseIPs()
 	}
 
+	if len(n.getStaticIPTags()) >= 1 {
+		if n.stats.AssignedStaticIP == "" {
+			ip, err := n.ops.AllocateStaticIP(ctx, n.getStaticIPTags())
+			if err != nil {
+				return false, err
+			}
+			n.stats.AssignedStaticIP = ip
+		}
+	}
+
 	a, err := n.determineMaintenanceAction()
 	if err != nil {
 		n.abortNoLongerExcessIPs(nil)
@@ -981,6 +1004,14 @@ func (n *Node) PopulateIPReleaseStatus(node *v2.CiliumNode) {
 	node.Status.IPAM.ReleaseIPs = releaseStatus
 }
 
+func (n *Node) PopulateStaticIPStatus(node *v2.CiliumNode) {
+	n.mutex.Lock()
+	defer n.mutex.Unlock()
+	if n.stats.AssignedStaticIP != "" {
+		node.Status.IPAM.AssignedStaticIP = n.stats.AssignedStaticIP
+	}
+}
+
 // syncToAPIServer synchronizes the contents of the CiliumNode resource
 // [(*Node).resource)] with the K8s apiserver. This operation occurs on an
 // interval to refresh the CiliumNode resource.
@@ -1029,6 +1060,7 @@ func (n *Node) syncToAPIServer() (err error) {
 
 		n.ops.PopulateStatusFields(node)
 		n.PopulateIPReleaseStatus(node)
+		n.PopulateStaticIPStatus(node)
 
 		err = n.update(origNode, node, retry, true)
 		if err == nil {

pkg/ipam/node_manager.go

@@ -75,6 +75,8 @@ type NodeOperations interface {
 	// to perform the actual allocation.
 	AllocateIPs(ctx context.Context, allocation *AllocationAction) error
 
+	AllocateStaticIP(ctx context.Context, staticIPTags ipamTypes.Tags) (string, error)
+
 	// PrepareIPRelease is called to calculate whether any IP excess needs
 	// to be resolved. It behaves identical to PrepareIPAllocation but
 	// indicates a need to release IPs.

pkg/ipam/node_manager_test.go

@@ -123,6 +123,10 @@ func (n *nodeOperationsMock) AllocateIPs(ctx context.Context, allocation *Alloca
 	return nil
 }
 
+func (n *nodeOperationsMock) AllocateStaticIP(ctx context.Context, staticIPTags ipamTypes.Tags) (string, error) {
+	return "", nil
+}
+
 func (n *nodeOperationsMock) PrepareIPRelease(excessIPs int, scopedLog *logrus.Entry) *ReleaseAction {
 	n.mutex.RLock()
 	excessIPs = math.IntMin(excessIPs, len(n.allocatedIPs))