CNS Pool Monitor - Init and basic resize logic

cns/api.go

@@ -3,7 +3,11 @@
 
 package cns
 
-import "encoding/json"
+import (
+	"encoding/json"
+
+	"github.com/Azure/azure-container-networking/cns/common"
+)
 
 // Container Network Service remote API Contract
 const (
@@ -25,6 +29,27 @@ const (
 	V2Prefix                      = "/v0.2"
 )
 
+// HTTPService describes the min API interface that every service should have.
+type HTTPService interface {
+	common.ServiceAPI
+	SendNCSnapShotPeriodically(int, chan bool)
+	SetNodeOrchestrator(*SetOrchestratorTypeRequest)
+	SyncNodeStatus(string, string, string, json.RawMessage) (int, string)
+	GetAvailableIPConfigs() []IPConfigurationStatus
+	GetPodIPConfigState() map[string]IPConfigurationStatus
+	MarkIPsAsPending(numberToMark int) (map[string]SecondaryIPConfig, error)
+}
+
+// This is used for KubernetesCRD orchastrator Type where NC has multiple ips.
+// This struct captures the state for SecondaryIPs associated to a given NC
+type IPConfigurationStatus struct {
+	NCID                string
+	ID                  string //uuid
+	IPAddress           string
+	State               string
+	OrchestratorContext json.RawMessage
+}
+
 // SetEnvironmentRequest describes the Request to set the environment in CNS.
 type SetEnvironmentRequest struct {
 	Location    string
@@ -136,6 +161,17 @@ type NodeConfiguration struct {
 	NodeSubnet Subnet
 }
 
+type IPAMPoolMonitor interface {
+	Start() error
+	UpdatePoolLimitsTransacted(ScalarUnits)
+}
+
+type ScalarUnits struct {
+	BatchSize               int64
+	RequestThresholdPercent int64
+	ReleaseThresholdPercent int64
+}
+
 // Response describes generic response from CNS.
 type Response struct {
 	ReturnCode int

cns/cnsclient/apiclient.go

@@ -4,6 +4,6 @@ import "github.com/Azure/azure-container-networking/cns"
 
 // APIClient interface to update cns state
 type APIClient interface {
-	ReconcileNCState(*cns.CreateNetworkContainerRequest, map[string]cns.KubernetesPodInfo) error
-	CreateOrUpdateNC(cns.CreateNetworkContainerRequest) error
+	ReconcileNCState(nc *cns.CreateNetworkContainerRequest, pods map[string]cns.KubernetesPodInfo, scalarUnits cns.ScalarUnits) error
+	CreateOrUpdateNC(nc cns.CreateNetworkContainerRequest, scalarUnits cns.ScalarUnits) error
 }

cns/cnsclient/cnsclient_test.go

@@ -15,6 +15,7 @@ import (
 	"github.com/Azure/azure-container-networking/cns"
 	"github.com/Azure/azure-container-networking/cns/common"
 	"github.com/Azure/azure-container-networking/cns/fakes"
+	"github.com/Azure/azure-container-networking/cns/ipampoolmonitor"
 	"github.com/Azure/azure-container-networking/cns/logger"
 	"github.com/Azure/azure-container-networking/cns/restserver"
 	"github.com/Azure/azure-container-networking/log"
@@ -38,6 +39,7 @@ var (
 
 func addTestStateToRestServer(t *testing.T, secondaryIps []string) {
 	var ipConfig cns.IPConfiguration
+	var scalarUnits cns.ScalarUnits
 	ipConfig.DNSServers = dnsservers
 	ipConfig.GatewayIPAddress = gatewayIp
 	var ipSubnet cns.IPSubnet
@@ -61,7 +63,7 @@ func addTestStateToRestServer(t *testing.T, secondaryIps []string) {
 		SecondaryIPConfigs:   secondaryIPConfigs,
 	}
 
-	returnCode := svc.CreateOrUpdateNetworkContainerInternal(req)
+	returnCode := svc.CreateOrUpdateNetworkContainerInternal(req, scalarUnits)
 	if returnCode != 0 {
 		t.Fatalf("Failed to createNetworkContainerRequest, req: %+v, err: %d", req, returnCode)
 	}
@@ -121,6 +123,8 @@ func TestMain(m *testing.M) {
 	httpRestService, err := restserver.NewHTTPRestService(&config, fakes.NewFakeImdsClient())
 	svc = httpRestService.(*restserver.HTTPRestService)
 	svc.Name = "cns-test-server"
+	svc.PoolMonitor = ipampoolmonitor.NewCNSIPAMPoolMonitor(nil, nil)
+
 	if err != nil {
 		logger.Errorf("Failed to create CNS object, err:%v.\n", err)
 		return

cns/cnsclient/httpapi/client.go

@@ -13,8 +13,8 @@ type Client struct {
 }
 
 // CreateOrUpdateNC updates cns state
-func (client *Client) CreateOrUpdateNC(ncRequest cns.CreateNetworkContainerRequest) error {
-	returnCode := client.RestService.CreateOrUpdateNetworkContainerInternal(ncRequest)
+func (client *Client) CreateOrUpdateNC(ncRequest cns.CreateNetworkContainerRequest, scalarUnits cns.ScalarUnits) error {
+	returnCode := client.RestService.CreateOrUpdateNetworkContainerInternal(ncRequest, scalarUnits)
 
 	if returnCode != 0 {
 		return fmt.Errorf("Failed to Create NC request: %+v, errorCode: %d", ncRequest, returnCode)
@@ -24,8 +24,8 @@ func (client *Client) CreateOrUpdateNC(ncRequest cns.CreateNetworkContainerReque
 }
 
 // ReconcileNCState initializes cns state
-func (client *Client) ReconcileNCState(ncRequest *cns.CreateNetworkContainerRequest, podInfoByIP map[string]cns.KubernetesPodInfo) error {
-	returnCode := client.RestService.ReconcileNCState(ncRequest, podInfoByIP)
+func (client *Client) ReconcileNCState(ncRequest *cns.CreateNetworkContainerRequest, podInfoByIP map[string]cns.KubernetesPodInfo, scalarUnits cns.ScalarUnits) error {
+	returnCode := client.RestService.ReconcileNCState(ncRequest, podInfoByIP, scalarUnits)
 
 	if returnCode != 0 {
 		return fmt.Errorf("Failed to Reconcile ncState: ncRequest %+v, podInfoMap: %+v, errorCode: %d", *ncRequest, podInfoByIP, returnCode)

cns/fakes/cnsfake.go

@@ -0,0 +1,56 @@
+package fakes
+
+import (
+	"encoding/json"
+
+	"github.com/Azure/azure-container-networking/cns"
+	"github.com/Azure/azure-container-networking/cns/common"
+)
+
+type HTTPServiceFake struct {
+	PoolMonitor cns.IPAMPoolMonitor
+}
+
+func NewHTTPServiceFake() *HTTPServiceFake {
+	return &HTTPServiceFake{}
+}
+
+func (fake *HTTPServiceFake) SendNCSnapShotPeriodically(int, chan bool) {
+
+}
+
+func (fake *HTTPServiceFake) SetNodeOrchestrator(*cns.SetOrchestratorTypeRequest) {
+
+}
+
+func (fake *HTTPServiceFake) SyncNodeStatus(string, string, string, json.RawMessage) (int, string) {
+	return 0, ""
+}
+
+func (fake *HTTPServiceFake) GetAvailableIPConfigs() []cns.IPConfigurationStatus {
+	return []cns.IPConfigurationStatus{}
+}
+
+func (fake *HTTPServiceFake) GetPodIPConfigState() map[string]cns.IPConfigurationStatus {
+	return make(map[string]cns.IPConfigurationStatus)
+}
+
+func (fake *HTTPServiceFake) MarkIPsAsPending(numberToMark int) (map[string]cns.SecondaryIPConfig, error) {
+	return make(map[string]cns.SecondaryIPConfig), nil
+}
+
+func (fake *HTTPServiceFake) GetOption(string) interface{} {
+	return nil
+}
+
+func (fake *HTTPServiceFake) SetOption(string, interface{}) {
+
+}
+
+func (fake *HTTPServiceFake) Start(*common.ServiceConfig) error {
+	return nil
+}
+
+func (fake *HTTPServiceFake) Stop() {
+
+}

cns/fakes/requestcontrollerfake.go

@@ -0,0 +1,22 @@
+package fakes
+
+import (
+	"context"
+
+	nnc "github.com/Azure/azure-container-networking/nodenetworkconfig/api/v1alpha"
+)
+
+type RequestControllerFake struct {
+}
+
+func NewRequestControllerFake() *RequestControllerFake {
+	return &RequestControllerFake{}
+}
+
+func (rc RequestControllerFake) StartRequestController(exitChan <-chan struct{}) error {
+	return nil
+}
+
+func (rc RequestControllerFake) UpdateCRDSpec(cntxt context.Context, crdSpec nnc.NodeNetworkConfigSpec) error {
+	return nil
+}

cns/ipampoolmonitor/ipampoolmonitor.go

@@ -0,0 +1,131 @@
+package ipampoolmonitor
+
+import (
+	"context"
+	"sync"
+
+	"github.com/Azure/azure-container-networking/cns"
+	"github.com/Azure/azure-container-networking/cns/logger"
+	"github.com/Azure/azure-container-networking/cns/requestcontroller"
+	nnc "github.com/Azure/azure-container-networking/nodenetworkconfig/api/v1alpha"
+)
+
+var (
+	increasePoolSize = 1
+	decreasePoolSize = -1
+	doNothing        = 0
+)
+
+type CNSIPAMPoolMonitor struct {
+	initialized bool
+
+	cns            cns.HTTPService
+	rc             requestcontroller.RequestController
+	scalarUnits    cns.ScalarUnits
+	MinimumFreeIps int
+	MaximumFreeIps int
+
+	sync.RWMutex
+}
+
+func NewCNSIPAMPoolMonitor(cnsService cns.HTTPService, requestController requestcontroller.RequestController) *CNSIPAMPoolMonitor {
+	return &CNSIPAMPoolMonitor{
+		initialized: false,
+		cns:         cnsService,
+		rc:          requestController,
+	}
+}
+
+// TODO: add looping and cancellation to this, and add to CNS MAIN
+func (pm *CNSIPAMPoolMonitor) Start() error {
+
+	if pm.initialized {
+		availableIPConfigs := pm.cns.GetAvailableIPConfigs()
+		rebatchAction := pm.checkForResize(len(availableIPConfigs))
+		switch rebatchAction {
+		case increasePoolSize:
+			return pm.increasePoolSize()
+		case decreasePoolSize:
+			return pm.decreasePoolSize()
+		}
+	}
+
+	return nil
+}
+
+// UpdatePoolLimitsTransacted called by request controller on reconcile to set the batch size limits
+func (pm *CNSIPAMPoolMonitor) UpdatePoolLimitsTransacted(scalarUnits cns.ScalarUnits) {
+	pm.Lock()
+	defer pm.Unlock()
+	pm.scalarUnits = scalarUnits
+
+	// TODO rounding?
+	pm.MinimumFreeIps = int(pm.scalarUnits.BatchSize * (pm.scalarUnits.RequestThresholdPercent / 100))
+	pm.MaximumFreeIps = int(pm.scalarUnits.BatchSize * (pm.scalarUnits.ReleaseThresholdPercent / 100))
+
+	pm.initialized = true
+}
+
+func (pm *CNSIPAMPoolMonitor) checkForResize(freeIPConfigCount int) int {
+	switch {
+	// pod count is increasing
+	case freeIPConfigCount < pm.MinimumFreeIps:
+		logger.Printf("Number of free IP's (%d) < minimum free IPs (%d), request batch increase\n", freeIPConfigCount, pm.MinimumFreeIps)
+		return increasePoolSize
+
+	// pod count is decreasing
+	case freeIPConfigCount > pm.MaximumFreeIps:
+		logger.Printf("Number of free IP's (%d) > maximum free IPs (%d), request batch decrease\n", freeIPConfigCount, pm.MaximumFreeIps)
+		return decreasePoolSize
+	}
+	return doNothing
+}
+
+func (pm *CNSIPAMPoolMonitor) increasePoolSize() error {
+	increaseIPCount := len(pm.cns.GetPodIPConfigState()) + int(pm.scalarUnits.BatchSize)
+
+	// pass nil map to CNStoCRDSpec because we don't want to modify the to be deleted ipconfigs
+	spec, err := CNSToCRDSpec(nil, increaseIPCount)
+	if err != nil {
+		return err
+	}
+
+	return pm.rc.UpdateCRDSpec(context.Background(), spec)
+}
+
+func (pm *CNSIPAMPoolMonitor) decreasePoolSize() error {
+
+	// TODO: Better handling here, negatives
+	// TODO: Maintain desired state to check against if pool size adjustment is already happening
+	decreaseIPCount := len(pm.cns.GetPodIPConfigState()) - int(pm.scalarUnits.BatchSize)
+
+	// mark n number of IP's as pending
+	pendingIPAddresses, err := pm.cns.MarkIPsAsPending(decreaseIPCount)
+	if err != nil {
+		return err
+	}
+
+	// convert the pending IP addresses to a spec
+	spec, err := CNSToCRDSpec(pendingIPAddresses, decreaseIPCount)
+	if err != nil {
+		return err
+	}
+
+	return pm.rc.UpdateCRDSpec(context.Background(), spec)
+}
+
+// CNSToCRDSpec translates CNS's map of Ips to be released and requested ip count into a CRD Spec
+func CNSToCRDSpec(toBeDeletedSecondaryIPConfigs map[string]cns.SecondaryIPConfig, ipCount int) (nnc.NodeNetworkConfigSpec, error) {
+	var (
+		spec nnc.NodeNetworkConfigSpec
+		uuid string
+	)
+
+	spec.RequestedIPCount = int64(ipCount)
+
+	for uuid = range toBeDeletedSecondaryIPConfigs {
+		spec.IPsNotInUse = append(spec.IPsNotInUse, uuid)
+	}
+
+	return spec, nil
+}

cns/ipampoolmonitor/ipampoolmonitor_test.go

@@ -0,0 +1,19 @@
+package ipampoolmonitor
+
+import (
+	"testing"
+
+	"github.com/Azure/azure-container-networking/cns"
+	"github.com/Azure/azure-container-networking/cns/fakes"
+)
+
+func TestInterfaces(t *testing.T) {
+	fakecns := fakes.NewHTTPServiceFake()
+	fakerc := fakes.NewRequestControllerFake()
+
+	fakecns.PoolMonitor = NewCNSIPAMPoolMonitor(fakecns, fakerc)
+
+	scalarUnits := cns.ScalarUnits{}
+
+	fakecns.PoolMonitor.UpdatePoolLimitsTransacted(scalarUnits)
+}

cns/requestcontroller/kubecontroller/crdreconciler.go

@@ -55,7 +55,13 @@ func (r *CrdReconciler) Reconcile(request reconcile.Request) (reconcile.Result,
 		return reconcile.Result{}, err
 	}
 
-	if err = r.CNSClient.CreateOrUpdateNC(ncRequest); err != nil {
+	scalarUnits := cns.ScalarUnits{
+		BatchSize:               nodeNetConfig.Status.Scaler.BatchSize,
+		RequestThresholdPercent: nodeNetConfig.Status.Scaler.RequestThresholdPercent,
+		ReleaseThresholdPercent: nodeNetConfig.Status.Scaler.ReleaseThresholdPercent,
+	}
+
+	if err = r.CNSClient.CreateOrUpdateNC(ncRequest, scalarUnits); err != nil {
 		logger.Errorf("[cns-rc] Error creating or updating NC in reconcile: %v", err)
 		// requeue
 		return reconcile.Result{}, err