Request controller for CNS

Makefile

@@ -378,6 +378,7 @@ test-all:
         ./cnm/network/ \
         ./cni/ipam/ \
         ./cns/ipamclient/ \
+        ./cns/requestcontroller/kubecontroller/ \
         ./cnms/service/ \
         ./npm/iptm/ \
         ./npm/ipsm/

cns/cnsclient/apiclient.go

@@ -0,0 +1,8 @@
+package cnsclient
+
+import "github.com/Azure/azure-container-networking/cns"
+
+// APIClient interface to update cns state
+type APIClient interface {
+	UpdateCNSState(createNetworkContainerRequest *cns.CreateNetworkContainerRequest) error
+}

cns/cnsclient/httpapi/client.go

@@ -0,0 +1,17 @@
+package httpapi
+
+import (
+	"github.com/Azure/azure-container-networking/cns"
+	"github.com/Azure/azure-container-networking/cns/restserver"
+)
+
+// Client implements APIClient interface. Used to update CNS state
+type Client struct {
+	RestService *restserver.HTTPRestService
+}
+
+// UpdateCNSState updates cns state
+func (client *Client) UpdateCNSState(createNetworkContainerRequest *cns.CreateNetworkContainerRequest) error {
+	//Mat will pick up from here
+	return nil
+}

cns/requestcontroller/example/main.go

@@ -0,0 +1,83 @@
+package main
+
+import (
+	"time"
+
+	"github.com/Azure/azure-container-networking/cns/logger"
+	"github.com/Azure/azure-container-networking/cns/requestcontroller"
+	"github.com/Azure/azure-container-networking/cns/requestcontroller/kubecontroller"
+	"github.com/Azure/azure-container-networking/cns/restserver"
+	nnc "github.com/Azure/azure-container-networking/nodenetworkconfig/api/v1alpha"
+	"golang.org/x/net/context"
+)
+
+func goRequestController(rc requestcontroller.RequestController) {
+	//Exit channel for requestController, this channel is notified when requestController receives
+	//SIGINT or SIGTERM, requestControllerExitChan is sent 'true' and you can clean up anything then
+	requestControllerExitChan := make(chan bool, 1)
+
+	//Start the RequestController which starts the reconcile loop, blocks
+	go func() {
+		if err := rc.StartRequestController(requestControllerExitChan); err != nil {
+			logger.Errorf("Error starting requestController: %v", err)
+			return
+		}
+	}()
+
+	// After calling StartRequestController, there needs to be some pause before updating CRD spec
+	time.Sleep(5 * time.Second)
+
+	// We provide a context when making operations on CRD in case we need to cancel operation
+	cntxt := context.Background()
+
+	// Create some dummy uuids
+	uuids := make([]string, 5)
+	uuids[0] = "uuid0"
+	uuids[1] = "uuid1"
+	uuids[2] = "uuid2"
+	uuids[3] = "uuid3"
+	uuids[4] = "uuid4"
+
+	// newCount = oldCount - ips releasing
+	// In this example, say we had 20 allocated to the node, we want to release 5, new count would be 15
+	oldCount := 20
+	newRequestedIPCount := int64(oldCount - len(uuids))
+
+	//Create CRD spec
+	spec := &nnc.NodeNetworkConfigSpec{
+		RequestedIPCount: newRequestedIPCount,
+		IPsNotInUse:      uuids,
+	}
+
+	//Update CRD spec
+	rc.UpdateCRDSpec(cntxt, spec)
+
+	<-requestControllerExitChan
+	logger.Printf("Request controller received sigint or sigterm, time to cleanup")
+	// Clean clean...
+}
+
+//Example of using the requestcontroller package
+func main() {
+	var requestController requestcontroller.RequestController
+
+	//Assuming logger is already setup and stuff
+	logger.InitLogger("Azure CNS", 3, 3, "")
+
+	restService := &restserver.HTTPRestService{}
+
+	//Provide kubeconfig, this method was abstracted out for testing
+	kubeconfig, err := kubecontroller.GetKubeConfig()
+	if err != nil {
+		logger.Errorf("Error getting kubeconfig: %v", err)
+	}
+
+	requestController, err = kubecontroller.NewCrdRequestController(restService, kubeconfig)
+	if err != nil {
+		logger.Errorf("Error making new RequestController: %v", err)
+		return
+	}
+
+	//Rely on the interface
+	goRequestController(requestController)
+}

cns/requestcontroller/kubecontroller/apiclientinterface.go

@@ -0,0 +1,18 @@
+package kubecontroller
+
+import (
+	"context"
+
+	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/types"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+)
+
+// ObjectKey identifies a Kubernetes Object.
+type ObjectKey = types.NamespacedName
+
+// KubeClient is an interface that talks to the API server
+type KubeClient interface {
+	Get(ctx context.Context, key client.ObjectKey, obj runtime.Object) error
+	Update(ctx context.Context, obj runtime.Object, opts ...client.UpdateOption) error
+}

cns/requestcontroller/kubecontroller/crdreconciler.go

@@ -0,0 +1,59 @@
+package kubecontroller
+
+import (
+	"context"
+
+	"github.com/Azure/azure-container-networking/cns/cnsclient"
+	"github.com/Azure/azure-container-networking/cns/logger"
+	nnc "github.com/Azure/azure-container-networking/nodenetworkconfig/api/v1alpha"
+	apierrors "k8s.io/apimachinery/pkg/api/errors"
+	ctrl "sigs.k8s.io/controller-runtime"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+	"sigs.k8s.io/controller-runtime/pkg/reconcile"
+)
+
+// CrdReconciler watches for CRD status changes
+type CrdReconciler struct {
+	KubeClient KubeClient
+	NodeName   string
+	CNSClient  cnsclient.APIClient
+}
+
+// Reconcile is called on CRD status changes
+func (r *CrdReconciler) Reconcile(request reconcile.Request) (reconcile.Result, error) {
+	var nodeNetConfig nnc.NodeNetworkConfig
+
+	//Get the CRD object
+	if err := r.KubeClient.Get(context.TODO(), request.NamespacedName, &nodeNetConfig); err != nil {
+		if apierrors.IsNotFound(err) {
+			logger.Printf("[cns-rc] CRD not found, ignoring %v", err)
+			return reconcile.Result{}, client.IgnoreNotFound(err)
+		} else {
+			logger.Errorf("[cns-rc] Error retrieving CRD from cache : %v", err)
+			return reconcile.Result{}, err
+		}
+	}
+
+	logger.Printf("[cns-rc] CRD object: %v", nodeNetConfig)
+
+	//TODO: Translate CRD status into NetworkContainer request
+	ncRequest, err := CRDStatusToNCRequest(&nodeNetConfig.Status)
+	if err != nil {
+		logger.Errorf("[cns-rc] Error translating crd status to nc request %v", err)
+		//requeue
+		return reconcile.Result{}, err
+	}
+
+	//TODO: process the nc request on CNS side
+	r.CNSClient.UpdateCNSState(ncRequest)
+
+	return reconcile.Result{}, nil
+}
+
+// SetupWithManager Sets up the reconciler with a new manager, filtering using NodeNetworkConfigFilter
+func (r *CrdReconciler) SetupWithManager(mgr ctrl.Manager) error {
+	return ctrl.NewControllerManagedBy(mgr).
+		For(&nnc.NodeNetworkConfig{}).
+		WithEventFilter(NodeNetworkConfigFilter{nodeName: r.NodeName}).
+		Complete(r)
+}

cns/requestcontroller/kubecontroller/crdrequestcontroller.go

@@ -0,0 +1,169 @@
+package kubecontroller
+
+import (
+	"context"
+	"errors"
+	"os"
+
+	"github.com/Azure/azure-container-networking/cns/cnsclient/httpapi"
+	"github.com/Azure/azure-container-networking/cns/logger"
+	"github.com/Azure/azure-container-networking/cns/restserver"
+	nnc "github.com/Azure/azure-container-networking/nodenetworkconfig/api/v1alpha"
+	"k8s.io/apimachinery/pkg/runtime"
+	clientgoscheme "k8s.io/client-go/kubernetes/scheme"
+	"k8s.io/client-go/rest"
+	ctrl "sigs.k8s.io/controller-runtime"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+	"sigs.k8s.io/controller-runtime/pkg/manager"
+)
+
+const nodeNameEnvVar = "NODENAME"
+const k8sNamespace = "kube-system"
+const prometheusAddress = "0" //0 means disabled
+
+// crdRequestController
+// - watches CRD status changes
+// - updates CRD spec
+type crdRequestController struct {
+	mgr        manager.Manager //Manager starts the reconcile loop which watches for crd status changes
+	KubeClient KubeClient      //KubeClient interacts with API server
+	nodeName   string          //name of node running this program
+	Reconciler *CrdReconciler
+}
+
+// GetKubeConfig precedence
+// * --kubeconfig flag pointing at a file at this cmd line
+// * KUBECONFIG environment variable pointing at a file
+// * In-cluster config if running in cluster
+// * $HOME/.kube/config if exists
+func GetKubeConfig() (*rest.Config, error) {
+	k8sconfig, err := ctrl.GetConfig()
+	if err != nil {
+		return nil, err
+	}
+	return k8sconfig, nil
+}
+
+//NewCrdRequestController given a reference to CNS's HTTPRestService state, returns a crdRequestController struct
+func NewCrdRequestController(restService *restserver.HTTPRestService, kubeconfig *rest.Config) (*crdRequestController, error) {
+
+	//Check that logger package has been intialized
+	if logger.Log == nil {
+		return nil, errors.New("Must initialize logger before calling")
+	}
+
+	// Check that NODENAME environment variable is set. NODENAME is name of node running this program
+	nodeName := os.Getenv(nodeNameEnvVar)
+	if nodeName == "" {
+		return nil, errors.New("Must declare " + nodeNameEnvVar + " environment variable.")
+	}
+
+	//Add client-go scheme to runtime sheme so manager can recognize it
+	var scheme = runtime.NewScheme()
+	if err := clientgoscheme.AddToScheme(scheme); err != nil {
+		return nil, errors.New("Error adding client-go scheme to runtime scheme")
+	}
+
+	//Add CRD scheme to runtime sheme so manager can recognize it
+	if err := nnc.AddToScheme(scheme); err != nil {
+		return nil, errors.New("Error adding NodeNetworkConfig scheme to runtime scheme")
+	}
+
+	// Create manager for CrdRequestController
+	// MetricsBindAddress is the tcp address that the controller should bind to
+	// for serving prometheus metrics, set to "0" to disable
+	mgr, err := ctrl.NewManager(kubeconfig, ctrl.Options{
+		Scheme:             scheme,
+		MetricsBindAddress: prometheusAddress,
+		Namespace:          k8sNamespace,
+	})
+	if err != nil {
+		logger.Errorf("[cns-rc] Error creating new request controller manager: %v", err)
+		return nil, err
+	}
+
+	//Create httpClient
+	httpClient := &httpapi.Client{
+		RestService: restService,
+	}
+
+	//Create reconciler
+	crdreconciler := &CrdReconciler{
+		KubeClient: mgr.GetClient(),
+		NodeName:   nodeName,
+		CNSClient:  httpClient,
+	}
+
+	// Setup manager with reconciler
+	if err := crdreconciler.SetupWithManager(mgr); err != nil {
+		logger.Errorf("[cns-rc] Error creating new CrdRequestController: %v", err)
+		return nil, err
+	}
+
+	// Create the requestController
+	crdRequestController := crdRequestController{
+		mgr:        mgr,
+		KubeClient: mgr.GetClient(),
+		nodeName:   nodeName,
+		Reconciler: crdreconciler,
+	}
+
+	return &crdRequestController, nil
+}
+
+// StartRequestController starts the Reconciler loop which watches for CRD status updates
+// Blocks until SIGINT or SIGTERM is received
+// Notifies exitChan when kill signal received
+func (crdRC *crdRequestController) StartRequestController(exitChan chan bool) error {
+	logger.Printf("Starting manager")
+	if err := crdRC.mgr.Start(SetupSignalHandler(exitChan)); err != nil {
+		logger.Errorf("[cns-rc] Error starting manager: %v", err)
+	}
+
+	return nil
+}
+
+// UpdateCRDSpec updates the CRD spec
+func (crdRC *crdRequestController) UpdateCRDSpec(cntxt context.Context, crdSpec *nnc.NodeNetworkConfigSpec) error {
+	nodeNetworkConfig, err := crdRC.getNodeNetConfig(cntxt, crdRC.nodeName, k8sNamespace)
+	if err != nil {
+		logger.Errorf("[cns-rc] Error getting CRD when updating spec %v", err)
+		return err
+	}
+
+	//Update the CRD spec
+	crdSpec.DeepCopyInto(&nodeNetworkConfig.Spec)
+
+	//Send update to API server
+	if err := crdRC.updateNodeNetConfig(cntxt, nodeNetworkConfig); err != nil {
+		logger.Errorf("[cns-rc] Error updating CRD spec %v", err)
+		return err
+	}
+
+	return nil
+}
+
+// getNodeNetConfig gets the nodeNetworkConfig CRD given the name and namespace of the CRD object
+func (crdRC *crdRequestController) getNodeNetConfig(cntxt context.Context, name, namespace string) (*nnc.NodeNetworkConfig, error) {
+	nodeNetworkConfig := &nnc.NodeNetworkConfig{}
+
+	err := crdRC.KubeClient.Get(cntxt, client.ObjectKey{
+		Namespace: namespace,
+		Name:      name,
+	}, nodeNetworkConfig)
+
+	if err != nil {
+		return nil, err
+	}
+
+	return nodeNetworkConfig, nil
+}
+
+// updateNodeNetConfig updates the nodeNetConfig object in the API server with the given nodeNetworkConfig object
+func (crdRC *crdRequestController) updateNodeNetConfig(cntxt context.Context, nodeNetworkConfig *nnc.NodeNetworkConfig) error {
+	if err := crdRC.KubeClient.Update(cntxt, nodeNetworkConfig); err != nil {
+		return err
+	}
+
+	return nil
+}