nfn agent for configuring nodes

nfn-agent runs on all nodes in cluster
and on starting subscribes for notifications
from gRPC server and waits for message to
configure provider network by adding vlan
interface and ovs-bridge on the node

Change-Id: Ibec7d0fda28db7b444900e13c705d15cf012aa48
Signed-off-by: Ritu Sood <ritu.sood@intel.com>

2 files changed, 217 insertions, 1 deletions

diff --git a/Makefile b/Makefile
index f8d7460..cc83aca 100644
--- a/Makefile
+++ b/Makefile
@@ -12,7 +12,7 @@ export GOPATH ...
 export GO111MODULE=on
 
 .PHONY: all 
-all: clean nfn-operator  ovn4nfvk8s-cni
+all: clean nfn-operator  ovn4nfvk8s-cni nfn-agent
 
 nfn-operator:
 	@go build -o build/bin/nfn-operator ./cmd/nfn-operator
@@ -20,10 +20,14 @@ nfn-operator:
 ovn4nfvk8s-cni:
 	@go build -o build/bin/ovn4nfvk8s-cni ./cmd/ovn4nfvk8s-cni
 
+nfn-agent:
+	@go build -o build/bin/nfn-agent ./cmd/nfn-agent
+
 test:
 	@go test -v ./...
 
 clean:
 	@rm -f build/bin/ovn4nfvk8s*
 	@rm -f build/bin/nfn-operator*
+	@rm -f build/bin/nfn-agent*
 
diff --git a/cmd/nfn-agent/nfn-agent.go b/cmd/nfn-agent/nfn-agent.go
new file mode 100644
index 0000000..766e5f3
--- /dev/null
+++ b/cmd/nfn-agent/nfn-agent.go
@@ -0,0 +1,212 @@
+package main
+
+import (
+	"context"
+	"fmt"
+	"google.golang.org/grpc"
+	"io"
+	kexec "k8s.io/utils/exec"
+	"os"
+	"os/signal"
+	pb "ovn4nfv-k8s-plugin/internal/pkg/nfnNotify/proto"
+	"ovn4nfv-k8s-plugin/internal/pkg/ovn"
+	logf "sigs.k8s.io/controller-runtime/pkg/runtime/log"
+	"strings"
+	"syscall"
+	"time"
+	//"google.golang.org/grpc/keepalive"
+	"google.golang.org/grpc/status"
+	"sigs.k8s.io/controller-runtime/pkg/log/zap"
+)
+
+var log = logf.Log.WithName("nfn-agent")
+var errorChannel chan string
+var inSync bool
+var pnCreateStore []*pb.Notification_ProviderNwCreate
+
+// subscribe Notifications
+func subscribeNotif(client pb.NfnNotifyClient) error {
+	log.Info("Subscribe Notification from server")
+	ctx := context.Background()
+	var n pb.SubscribeContext
+	n.NodeName = os.Getenv("NFN_NODE_NAME")
+	for {
+		stream, err := client.Subscribe(ctx, &n, grpc.WaitForReady(true))
+		if err != nil {
+			log.Error(err, "Subscribe", "client", client, "status", status.Code(err))
+			continue
+		}
+		log.Info("Subscribe Notification success")
+
+		for {
+			in, err := stream.Recv()
+			if err == io.EOF {
+				// read done.
+				shutDownAgent("Stream closed")
+				return err
+			}
+			if err != nil {
+				log.Error(err, "Stream closed from server")
+				shutDownAgent("Stream closed from server")
+				return err
+			}
+			log.Info("Got message", "msg", in)
+			handleNotif(in)
+		}
+	}
+}
+
+func handleNotif(msg *pb.Notification) {
+	var err error
+	switch msg.GetCniType() {
+	case "ovn4nfv":
+		switch payload := msg.Payload.(type) {
+		case *pb.Notification_ProviderNwCreate:
+			if !inSync {
+				// Store Msgs
+				pnCreateStore = append(pnCreateStore, payload)
+				return
+			}
+			vlanID := payload.ProviderNwCreate.GetVlan().GetVlanId()
+			ln := payload.ProviderNwCreate.GetVlan().GetLogicalIntf()
+			pn := payload.ProviderNwCreate.GetVlan().GetProviderIntf()
+			name := payload.ProviderNwCreate.GetProviderNwName()
+			if ln == "" {
+				ln = name + "." + vlanID
+			}
+			err = ovn.CreateVlan(vlanID, pn, ln)
+			if err != nil {
+				log.Error(err, "Unable to create VLAN", "vlan", ln)
+				return
+			}
+			ovn.CreatePnBridge("nw_"+name, "br-"+name, ln)
+		case *pb.Notification_ProviderNwRemove:
+			if !inSync {
+				// Unexpected Remove message
+				return
+			}
+			ln := payload.ProviderNwRemove.GetVlanLogicalIntf()
+			name := payload.ProviderNwRemove.GetProviderNwName()
+			ovn.DeleteVlan(ln)
+			ovn.DeletePnBridge("nw_"+name, "br-"+name)
+		case *pb.Notification_InSync:
+			// Read config from node
+			vlanList := ovn.GetVlan()
+			pnBridgeList := ovn.GetPnBridge("nfn")
+			diffVlan := make(map[string]bool)
+			diffPnBridge := make(map[string]bool)
+		VLAN:
+			for _, pn := range pnCreateStore {
+				id := pn.ProviderNwCreate.GetVlan().GetVlanId()
+				ln := pn.ProviderNwCreate.GetVlan().GetLogicalIntf()
+				pn := pn.ProviderNwCreate.GetVlan().GetProviderIntf()
+				if ln == "" {
+					ln = pn + "." + id
+				}
+				for _, vlan := range vlanList {
+					if vlan == ln {
+						// VLAN already present
+						diffVlan[vlan] = true
+						continue VLAN
+					}
+				}
+				// Vlan not found
+				err = ovn.CreateVlan(id, pn, ln)
+				if err != nil {
+					log.Error(err, "Unable to create VLAN", "vlan", ln)
+					return
+				}
+			}
+		PRNETWORK:
+			for _, pn := range pnCreateStore {
+				ln := pn.ProviderNwCreate.GetVlan().GetLogicalIntf()
+				name := pn.ProviderNwCreate.GetProviderNwName()
+				for _, br := range pnBridgeList {
+					pnName := strings.Replace(br, "br-", "", -1)
+					if name == pnName {
+						diffPnBridge[br] = true
+						continue PRNETWORK
+					}
+				}
+				// Provider Network not found
+				ovn.CreatePnBridge("nw_"+name, "br-"+name, ln)
+			}
+			// Delete VLAN not in the list
+			for _, vlan := range vlanList {
+				if diffVlan[vlan] == false {
+					ovn.DeleteVlan(vlan)
+				}
+			}
+			// Delete Provider Bridge not in the list
+			for _, br := range pnBridgeList {
+				if diffPnBridge[br] == false {
+					name := strings.Replace(br, "br-", "", -1)
+					ovn.DeletePnBridge("nw_"+name, "br-"+name)
+				}
+			}
+
+			pnCreateStore = nil
+			inSync = true
+
+		}
+	// Add other Types here
+	default:
+		log.Info("Not supported cni type", "cni type", msg.GetCniType())
+	}
+}
+
+func shutdownHandler(errorChannel <-chan string) {
+	// Register to receive term/int signal.
+	signalChan := make(chan os.Signal, 1)
+	signal.Notify(signalChan, syscall.SIGTERM)
+	signal.Notify(signalChan, syscall.SIGINT)
+	signal.Notify(signalChan, syscall.SIGHUP)
+
+	var reason string
+	select {
+	case sig := <-signalChan:
+		if sig == syscall.SIGHUP {
+			log.Info("Received a SIGHUP")
+		}
+		reason = fmt.Sprintf("Received OS signal %v", sig)
+	case reason = <-errorChannel:
+		log.Info("Error", "reason", reason)
+	}
+	log.Info("nfn-agent is shutting down", "reason", reason)
+}
+
+func shutDownAgent(reason string) {
+	// Send a failure message and give few seconds complete shutdown.
+	log.Info("shutDownAgent recieved")
+	errorChannel <- reason
+	time.Sleep(10 * time.Second)
+	// The graceful shutdown failed, terminate the process.
+	panic("Shutdown failed. Panicking.")
+}
+
+func main() {
+	logf.SetLogger(zap.Logger(true))
+	log.Info("nfn-agent Started")
+
+	serverAddr := os.Getenv("NFN_OPERATOR_SERVICE_HOST") + ":" + os.Getenv("NFN_OPERATOR_SERVICE_PORT")
+	// Setup ovn utilities
+	exec := kexec.New()
+	err := ovn.SetExec(exec)
+	if err != nil {
+		log.Error(err, "Unable to setup OVN Utils")
+		return
+	}
+	conn, err := grpc.Dial(serverAddr, grpc.WithInsecure())
+	if err != nil {
+		log.Error(err, "fail to dial")
+		return
+	}
+	defer conn.Close()
+	client := pb.NewNfnNotifyClient(conn)
+	errorChannel = make(chan string)
+
+	// Run client in background
+	go subscribeNotif(client)
+	shutdownHandler(errorChannel)
+
+}