controller.go

package controller

import (
	"context"
	"net"
	"net/http"
	"os"
	"os/signal"
	"sync"
	"syscall"

	"github.com/google/uuid"
	log "github.com/sirupsen/logrus"
	"github.com/spf13/viper"
	"google.golang.org/grpc"
	"google.golang.org/grpc/credentials/insecure"

	pb "code.fbi.h-da.de/danet/gosdn/api/go/gosdn/core"
	cpb "code.fbi.h-da.de/danet/gosdn/api/go/gosdn/csbi"
	ppb "code.fbi.h-da.de/danet/gosdn/api/go/gosdn/pnd"
	spb "code.fbi.h-da.de/danet/gosdn/api/go/gosdn/southbound"
	"code.fbi.h-da.de/danet/gosdn/controller/config"
	"code.fbi.h-da.de/danet/gosdn/controller/interfaces/southbound"
	nbi "code.fbi.h-da.de/danet/gosdn/controller/northbound/server"
	"code.fbi.h-da.de/danet/gosdn/controller/store"

	"code.fbi.h-da.de/danet/gosdn/controller/nucleus"
)

var coreLock sync.RWMutex
var coreOnce sync.Once

// Core is the representation of the controller's core
type Core struct {
	pndc       *store.PndStore
	httpServer *http.Server
	grpcServer *grpc.Server
	nbi        *nbi.NorthboundInterface
	stopChan   chan os.Signal

	csbiClient cpb.CsbiServiceClient
}

var c *Core

func init() {
	c = &Core{
		pndc:     store.NewPndStore(),
		stopChan: make(chan os.Signal, 1),
	}

	// Setting up signal capturing
	signal.Notify(c.stopChan, os.Interrupt, syscall.SIGTERM)
}

// initialize does start-up housekeeping like reading controller config files
func initialize() error {
	if err := startGrpc(); err != nil {
		return err
	}

	coreLock.Lock()
	startHttpServer()
	coreLock.Unlock()

	err := config.InitializeConfig()
	if err != nil {
		return err
	}

	err = restorePrincipalNetworkDomains()
	if err != nil {
		return err
	}

	sbi, err := createSouthboundInterfaces()
	if err != nil {
		return err
	}

	err = createPrincipalNetworkDomain(sbi)
	if err != nil {
		return err
	}

	return nil
}

func startGrpc() error {
	sock := viper.GetString("socket")
	lis, err := net.Listen("tcp", sock)
	if err != nil {
		return err
	}
	log.Infof("listening to %v", lis.Addr())
	c.grpcServer = grpc.NewServer()
	c.nbi = nbi.NewNBI(c.pndc)
	pb.RegisterCoreServiceServer(c.grpcServer, c.nbi.Core)
	ppb.RegisterPndServiceServer(c.grpcServer, c.nbi.Pnd)
	cpb.RegisterCsbiServiceServer(c.grpcServer, c.nbi.Csbi)
	spb.RegisterSbiServiceServer(c.grpcServer, c.nbi.Sbi)
	go func() {
		if err := c.grpcServer.Serve(lis); err != nil {
			log.Fatal(err)
		}
	}()

	orchestrator := viper.GetString("csbi-orchestrator")
	conn, err := grpc.Dial(orchestrator, grpc.WithTransportCredentials(insecure.NewCredentials()))
	if err != nil {
		log.Fatal(err)
	}
	c.csbiClient = cpb.NewCsbiServiceClient(conn)
	return nil
}

// createSouthboundInterfaces initializes the controller with its supported SBIs
func createSouthboundInterfaces() (southbound.SouthboundInterface, error) {
	sbi, err := nucleus.NewSBI(spb.Type(config.BaseSouthBoundType), config.BaseSouthBoundUUID)
	if err != nil {
		return nil, err
	}

	return sbi, nil
}

// createPrincipalNetworkDomain initializes the controller with an initial PND
func createPrincipalNetworkDomain(s southbound.SouthboundInterface) error {
	if !c.pndc.Exists(config.BasePndUUID) {
		pnd, err := nucleus.NewPND("base", "gosdn base pnd", config.BasePndUUID, s, c.csbiClient, callback)
		if err != nil {
			return err
		}
		err = c.pndc.Add(pnd)
		if err != nil {
			return err
		}
		return nil
	}

	return nil
}

// restorePrincipalNetworkDomains restores previously stored PNDs
func restorePrincipalNetworkDomains() error {
	pndsFromStore, err := c.pndc.Load()
	if err != nil {
		return err
	}

	sbi, err := createSouthboundInterfaces()
	if err != nil {
		return err
	}

	for _, pndFromStore := range pndsFromStore {
		log.Debugf("Restoring PND: %s\n", pndFromStore.Name)
		newPnd, err := nucleus.NewPND(
			pndFromStore.Name,
			pndFromStore.Description,
			pndFromStore.ID,
			sbi,
			c.csbiClient,
			callback,
		)
		if err != nil {
			return err
		}

		err = c.pndc.Add(newPnd)
		if err != nil {
			return err
		}
	}

	return nil
}

// Run calls initialize to start the controller
func Run(ctx context.Context) error {
	var initError error
	coreOnce.Do(func() {
		initError = initialize()
	})
	if initError != nil {
		log.WithFields(log.Fields{}).Error(initError)
		return initError
	}

	log.WithFields(log.Fields{}).Info("initialisation finished")

	select {
	case <-c.stopChan:
		return shutdown()
	case <-ctx.Done():
		return shutdown()
	}
}

func shutdown() error {
	log.Info("shutting down controller")
	coreLock.Lock()
	defer coreLock.Unlock()
	c.grpcServer.GracefulStop()
	return stopHttpServer()
}

func callback(id uuid.UUID, ch chan store.DeviceDetails) {
	if ch != nil {
		c.pndc.AddPendingChannel(id, ch)
		log.Infof("pending channel %v added", id)
	} else {
		c.pndc.RemovePendingChannel(id)
		log.Infof("pending channel %v removed", id)
	}
}