Skip to content
Snippets Groups Projects
principalNetworkDomain.go 14.1 KiB
Newer Older
  • Learn to ignore specific revisions
  • package nucleus
    
    
    Manuel Kieweg's avatar
    Manuel Kieweg committed
    import (
    
    Andre Sterba's avatar
    Andre Sterba committed
    	"context"
    
    Manuel Kieweg's avatar
    Manuel Kieweg committed
    	"encoding/json"
    
    Malte Bauch's avatar
    Malte Bauch committed
    	"io"
    	"os"
    	"plugin"
    
    	"time"
    
    Malte Bauch's avatar
    Malte Bauch committed
    	"code.fbi.h-da.de/danet/gosdn/metrics"
    
    Manuel Kieweg's avatar
    Manuel Kieweg committed
    	"code.fbi.h-da.de/danet/gosdn/nucleus/types"
    
    Manuel Kieweg's avatar
    Manuel Kieweg committed
    
    
    Manuel Kieweg's avatar
    Manuel Kieweg committed
    	cpb "code.fbi.h-da.de/danet/api/go/gosdn/csbi"
    	ppb "code.fbi.h-da.de/danet/api/go/gosdn/pnd"
    
    Malte Bauch's avatar
    Malte Bauch committed
    	spb "code.fbi.h-da.de/danet/api/go/gosdn/southbound"
    
    Manuel Kieweg's avatar
    Manuel Kieweg committed
    	tpb "code.fbi.h-da.de/danet/api/go/gosdn/transport"
    
    	"google.golang.org/protobuf/proto"
    
    Manuel Kieweg's avatar
    Manuel Kieweg committed
    	"code.fbi.h-da.de/danet/forks/goarista/gnmi"
    
    Manuel Kieweg's avatar
    Manuel Kieweg committed
    	"code.fbi.h-da.de/danet/gosdn/interfaces/change"
    	"code.fbi.h-da.de/danet/gosdn/interfaces/device"
    	"code.fbi.h-da.de/danet/gosdn/interfaces/networkdomain"
    	"code.fbi.h-da.de/danet/gosdn/interfaces/southbound"
    
    	si "code.fbi.h-da.de/danet/gosdn/interfaces/store"
    
    Manuel Kieweg's avatar
    Manuel Kieweg committed
    	"code.fbi.h-da.de/danet/gosdn/nucleus/errors"
    
    	"code.fbi.h-da.de/danet/gosdn/store"
    
    
    Manuel Kieweg's avatar
    Manuel Kieweg committed
    	"github.com/google/uuid"
    
    	"github.com/openconfig/ygot/ygot"
    	"github.com/openconfig/ygot/ytypes"
    
    Malte Bauch's avatar
    Malte Bauch committed
    	"github.com/prometheus/client_golang/prometheus"
    
    	log "github.com/sirupsen/logrus"
    
    // NewPND creates a Principle Network Domain
    
    func NewPND(name, description string, id uuid.UUID, sbi southbound.SouthboundInterface, c cpb.CsbiClient, callback func(uuid.UUID, chan store.DeviceDetails)) (networkdomain.NetworkDomain, error) {
    
    	pnd := &pndImplementation{
    
    		Name:        name,
    		Description: description,
    
    		sbic:        store.NewSbiStore(),
    
    		devices:     store.NewDeviceStore(id),
    
    		changes:     store.NewChangeStore(),
    
    
    		csbiClient: c,
    		callback:   callback,
    
    	if err := pnd.sbic.Add(sbi); err != nil {
    
    Manuel Kieweg's avatar
    Manuel Kieweg committed
    		return nil, err
    
    
    	if err := pnd.loadStoredDevices(); err != nil {
    		return nil, err
    	}
    
    
    type pndImplementation struct {
    
    	Name        string `json:"name,omitempty"`
    	Description string `json:"description,omitempty"`
    
    	sbic        *store.SbiStore
    	devices     *store.DeviceStore
    	changes     *store.ChangeStore
    
    	//nolint
    	Id uuid.UUID `json:"id,omitempty"`
    
    
    	csbiClient cpb.CsbiClient
    
    	callback   func(uuid.UUID, chan store.DeviceDetails)
    
    Manuel Kieweg's avatar
    Manuel Kieweg committed
    func (pnd *pndImplementation) PendingChanges() []uuid.UUID {
    
    Manuel Kieweg's avatar
    Manuel Kieweg committed
    	return pnd.changes.Pending()
    
    Manuel Kieweg's avatar
    Manuel Kieweg committed
    }
    
    
    Manuel Kieweg's avatar
    Manuel Kieweg committed
    func (pnd *pndImplementation) CommittedChanges() []uuid.UUID {
    
    Manuel Kieweg's avatar
    Manuel Kieweg committed
    	return pnd.changes.Committed()
    
    Manuel Kieweg's avatar
    Manuel Kieweg committed
    }
    
    
    Manuel Kieweg's avatar
    Manuel Kieweg committed
    func (pnd *pndImplementation) ConfirmedChanges() []uuid.UUID {
    	return pnd.changes.Confirmed()
    }
    
    func (pnd *pndImplementation) GetChange(cuid uuid.UUID) (change.Change, error) {
    	return pnd.changes.GetChange(cuid)
    
    func (pnd *pndImplementation) Commit(u uuid.UUID) error {
    
    Manuel Kieweg's avatar
    Manuel Kieweg committed
    	ch, err := pnd.changes.GetChange(u)
    
    	if err != nil {
    		return err
    	}
    
    Manuel Kieweg's avatar
    Manuel Kieweg committed
    	return ch.Commit()
    
    Manuel Kieweg's avatar
    Manuel Kieweg committed
    }
    
    
    func (pnd *pndImplementation) Confirm(u uuid.UUID) error {
    
    Manuel Kieweg's avatar
    Manuel Kieweg committed
    	ch, err := pnd.changes.GetChange(u)
    
    	if err != nil {
    		return err
    	}
    
    Manuel Kieweg's avatar
    Manuel Kieweg committed
    	return ch.Confirm()
    
    Manuel Kieweg's avatar
    Manuel Kieweg committed
    }
    
    
    func (pnd *pndImplementation) ID() uuid.UUID {
    
    func (pnd *pndImplementation) Devices() []uuid.UUID {
    	return pnd.devices.UUIDs()
    }
    
    
    // GetName returns the name of the PND
    
    Malte Bauch's avatar
    Malte Bauch committed
    func (pnd *pndImplementation) GetName() string {
    
    // ContainsDevice checks if the given device uuid is registered for this PND
    
    func (pnd *pndImplementation) ContainsDevice(id uuid.UUID) bool {
    
    	return pnd.devices.Exists(id)
    
    // GetDescription returns the current description of the PND
    
    func (pnd *pndImplementation) GetDescription() string {
    
    // GetSBIs returns the registered SBIs
    
    func (pnd *pndImplementation) GetSBIs() si.Store {
    	return pnd.sbic
    
    // Destroy destroys the PND
    
    func (pnd *pndImplementation) Destroy() error {
    	return destroy()
    }
    
    
    // AddSbi adds a SBI to the PND which will be supported
    
    func (pnd *pndImplementation) AddSbi(s southbound.SouthboundInterface) error {
    
    	return pnd.addSbi(s)
    
    // RemoveSbi removes a SBI from the PND
    
    Manuel Kieweg's avatar
    Manuel Kieweg committed
    // TODO: this should to recursively through
    
    // devices and remove the devices using
    // this SBI
    
    func (pnd *pndImplementation) RemoveSbi(id uuid.UUID) error {
    	return pnd.removeSbi(id)
    
    Manuel Kieweg's avatar
    Manuel Kieweg committed
    //AddDevice adds a new device to the PND
    func (pnd *pndImplementation) AddDevice(name string, opt *tpb.TransportOption, sid uuid.UUID) error {
    
    Malte Bauch's avatar
    Malte Bauch committed
    	labels := prometheus.Labels{"type": opt.Type.String()}
    	start := metrics.StartHook(labels, deviceCreationsTotal)
    	defer metrics.FinishHook(labels, start, deviceCreationDurationSecondsTotal, deviceCreationDurationSeconds)
    	var sbi southbound.SouthboundInterface
    	switch t := opt.Type; t {
    	case spb.Type_CONTAINERISED:
    
    		return pnd.handleCsbiEnrolment(name, opt)
    
    Malte Bauch's avatar
    Malte Bauch committed
    	case spb.Type_PLUGIN:
    		var err error
    		sbi, err = pnd.requestPlugin(name, opt)
    		if err != nil {
    			return err
    		}
    	default:
    		var err error
    		sbi, err = pnd.sbic.GetSBI(sid)
    		if err != nil {
    			return err
    		}
    
    	d, err := NewDevice(name, uuid.Nil, opt, sbi)
    
    Manuel Kieweg's avatar
    Manuel Kieweg committed
    	if err != nil {
    		return err
    
    	}
    	return pnd.addDevice(d)
    
    //AddDeviceFromStore adds a new device to the PND
    func (pnd *pndImplementation) AddDeviceFromStore(name string, deviceUUID uuid.UUID, opt *tpb.TransportOption, sid uuid.UUID) error {
    
    Malte Bauch's avatar
    Malte Bauch committed
    	if opt.Type == spb.Type_CONTAINERISED {
    
    		return pnd.handleCsbiEnrolment(name, opt)
    	}
    
    	sbi, err := pnd.sbic.GetSBI(sid)
    	if err != nil {
    		return err
    	}
    
    
    	d, err := NewDevice(name, deviceUUID, opt, sbi)
    
    	if err != nil {
    		return err
    	}
    	return pnd.addDevice(d)
    }
    
    
    func (pnd *pndImplementation) GetDevice(identifier string) (device.Device, error) {
    
    	d, err := pnd.devices.GetDevice(store.FromString(identifier))
    
    	if err != nil {
    		return nil, err
    	}
    
    Manuel Kieweg's avatar
    Manuel Kieweg committed
    	copiedGoStruct, err := ygot.DeepCopy(d.Model())
    
    Manuel Kieweg's avatar
    Manuel Kieweg committed
    	copiedDevice := &CommonDevice{name: d.Name(), UUID: d.ID(), GoStruct: copiedGoStruct}
    
    // RemoveDevice removes a device from the PND
    
    Manuel Kieweg's avatar
    Manuel Kieweg committed
    func (pnd *pndImplementation) RemoveDevice(uuid uuid.UUID) error {
    
    Manuel Kieweg's avatar
    Manuel Kieweg committed
    	return pnd.removeDevice(uuid)
    
    // Actual implementation, bind to struct if
    // neccessary
    func destroy() error {
    	return nil
    }
    
    
    func (pnd *pndImplementation) addSbi(sbi southbound.SouthboundInterface) error {
    
    	return pnd.sbic.Add(sbi)
    
    func (pnd *pndImplementation) removeSbi(id uuid.UUID) error {
    
    	return pnd.sbic.Delete(id)
    
    func (pnd *pndImplementation) addDevice(device device.Device) error {
    
    Manuel Kieweg's avatar
    Manuel Kieweg committed
    	err := pnd.devices.Add(device, device.Name())
    
    func (pnd *pndImplementation) removeDevice(id uuid.UUID) error {
    
    	d, err := pnd.devices.GetDevice(id)
    	if err != nil {
    		return err
    	}
    
    Malte Bauch's avatar
    Malte Bauch committed
    	labels := prometheus.Labels{"type": d.SBI().SbiIdentifier()}
    	start := metrics.StartHook(labels, deviceDeletionsTotal)
    	defer metrics.FinishHook(labels, start, deviceDeletionDurationSecondsTotal, deviceDeletionDurationSeconds)
    
    	switch d.(type) {
    	case *CsbiDevice:
    		return pnd.handleCsbiDeletion(id)
    	default:
    		return pnd.devices.Delete(id)
    	}
    
    func (pnd *pndImplementation) MarshalDevice(identifier string) (string, error) {
    
    	foundDevice, err := pnd.devices.GetDevice(store.FromString(identifier))
    
    	if err != nil {
    		return "", err
    	}
    
    Manuel Kieweg's avatar
    Manuel Kieweg committed
    	jsonTree, err := json.MarshalIndent(foundDevice.Model(), "", "\t")
    
    	if err != nil {
    		return "", err
    	}
    
    	log.WithFields(log.Fields{
    
    		"Identifier": identifier,
    		"Name":       foundDevice.Name,
    
    	}).Info("marshalled device")
    
    	return string(jsonTree), nil
    
    Manuel Kieweg's avatar
    Manuel Kieweg committed
    // Request sends a get request to a specific device
    
    func (pnd *pndImplementation) Request(uuid uuid.UUID, path string) (proto.Message, error) {
    
    	d, err := pnd.devices.GetDevice(store.FromString(uuid.String()))
    
    	ctx := context.Background()
    
    Manuel Kieweg's avatar
    Manuel Kieweg committed
    	res, err := d.Transport().Get(ctx, path)
    
    	if err != nil {
    
    	resp, ok := res.(proto.Message)
    	if !ok {
    
    		return nil, &errors.ErrInvalidTypeAssertion{
    			Value: res,
    			Type:  (*proto.Message)(nil),
    		}
    
    	}
    	err = d.ProcessResponse(resp)
    
    	if err != nil {
    
    	return resp, nil
    
    Manuel Kieweg's avatar
    Manuel Kieweg committed
    // RequestAll sends a request for all registered devices
    
    func (pnd *pndImplementation) RequestAll(path string) error {
    
    	for _, k := range pnd.devices.UUIDs() {
    
    		_, err := pnd.Request(k, path)
    		if err != nil {
    
    	log.WithFields(log.Fields{
    
    		"path": path,
    	}).Info("sent request to all devices")
    
    Manuel Kieweg's avatar
    Manuel Kieweg committed
    // ChangeOND creates a change from the provided Operation, path and value.
    // The Change is Pending and times out after the specified timeout period
    
    func (pnd *pndImplementation) ChangeOND(duid uuid.UUID, operation ppb.ApiOperation, path string, value ...string) (uuid.UUID, error) {
    	d, err := pnd.devices.GetDevice(duid)
    
    Manuel Kieweg's avatar
    Manuel Kieweg committed
    	if err != nil {
    
    		return uuid.Nil, err
    
    Manuel Kieweg's avatar
    Manuel Kieweg committed
    	cpy, err := ygot.DeepCopy(d.Model())
    
    Manuel Kieweg's avatar
    Manuel Kieweg committed
    	ygot.BuildEmptyTree(cpy)
    
    		return uuid.Nil, err
    
    
    	p, err := ygot.StringToStructuredPath(path)
    	if err != nil {
    
    		return uuid.Nil, err
    
    Manuel Kieweg's avatar
    Manuel Kieweg committed
    	if operation != ppb.ApiOperation_DELETE && len(value) != 1 {
    
    		return uuid.Nil, &errors.ErrInvalidParameters{
    
    			Func:  pnd.ChangeOND,
    			Param: value,
    
    	switch operation {
    
    Manuel Kieweg's avatar
    Manuel Kieweg committed
    	case ppb.ApiOperation_UPDATE, ppb.ApiOperation_REPLACE:
    
    Manuel Kieweg's avatar
    Manuel Kieweg committed
    		typedValue := gnmi.TypedValue(value[0])
    
    Manuel Kieweg's avatar
    Manuel Kieweg committed
    		if err := ytypes.SetNode(d.SBI().Schema().RootSchema(), cpy, p, typedValue); err != nil {
    
    			return uuid.Nil, err
    
    Manuel Kieweg's avatar
    Manuel Kieweg committed
    	case ppb.ApiOperation_DELETE:
    
    Manuel Kieweg's avatar
    Manuel Kieweg committed
    		if err := ytypes.DeleteNode(d.SBI().Schema().RootSchema(), cpy, p); err != nil {
    
    			return uuid.Nil, err
    
    		return uuid.Nil, &errors.ErrOperationNotSupported{Op: operation}
    
    Manuel Kieweg's avatar
    Manuel Kieweg committed
    	ygot.PruneEmptyBranches(cpy)
    
    	callback := func(original ygot.GoStruct, modified ygot.GoStruct) error {
    
    Manuel Kieweg's avatar
    Manuel Kieweg committed
    		ctx := context.WithValue(context.Background(), types.CtxKeyOperation, operation) // nolint
    
    		payload := change.Payload{Original: original, Modified: modified}
    		return d.Transport().Set(ctx, payload)
    
    	ch := NewChange(duid, d.Model(), cpy, callback)
    
    	if err := pnd.changes.Add(ch); err != nil {
    		return uuid.Nil, err
    	}
    	return ch.cuid, nil
    
    // nolint will be implemented in the near future
    
    func handleRollbackError(id uuid.UUID, err error) {
    	log.Error(err)
    	// TODO: Notion of invalid state needed.
    }
    
    
    func (pnd *pndImplementation) handleCsbiDeletion(id uuid.UUID) error {
    	log.Infof("csbi deletion triggered for %v", id)
    	ctx, cancel := context.WithTimeout(context.Background(), time.Minute)
    	defer cancel()
    	req := &cpb.DeleteRequest{
    		Timestamp: time.Now().UnixNano(),
    		Did:       []string{id.String()},
    	}
    	resp, err := pnd.csbiClient.Delete(ctx, req)
    	if err != nil {
    		return err
    	}
    	log.WithFields(log.Fields{
    		"uuid":   id,
    		"status": resp.Status,
    	}).Info("csbi deleted")
    	return nil
    }
    
    func (pnd *pndImplementation) handleCsbiEnrolment(name string, opt *tpb.TransportOption) error {
    	ctx, cancel := context.WithTimeout(context.Background(), time.Minute*5)
    	defer cancel()
    	req := &cpb.CreateRequest{
    		Timestamp:       time.Now().UnixNano(),
    		TransportOption: []*tpb.TransportOption{opt},
    	}
    	resp, err := pnd.csbiClient.Create(ctx, req)
    	if err != nil {
    		return err
    	}
    	for _, d := range resp.Deployments {
    
    Malte Bauch's avatar
    Malte Bauch committed
    		if err := pnd.createCsbiDevice(ctx, name, d, opt); err != nil {
    
    			log.Error(err)
    		}
    	}
    	return nil
    }
    
    
    Malte Bauch's avatar
    Malte Bauch committed
    func (pnd *pndImplementation) createCsbiDevice(ctx context.Context, name string, d *cpb.Deployment, opt *tpb.TransportOption) error {
    
    	defer func() {
    		if r := recover(); r != nil {
    			log.Errorf("recovered in sbi enrolment: %v", r)
    		}
    	}()
    	id, err := uuid.Parse(d.Id)
    	if err != nil {
    		return err
    	}
    
    	ch := make(chan store.DeviceDetails, 1)
    
    	pnd.callback(id, ch)
    
    Malte Bauch's avatar
    Malte Bauch committed
    	tickatus := time.NewTicker(time.Minute * 1)
    
    	go func() {
    
    Malte Bauch's avatar
    Malte Bauch committed
    		select {
    		case <-tickatus.C:
    			log.WithFields(log.Fields{
    				"id":  d.Id,
    				"err": ctx.Err(),
    			}).Error("csbi handshake timed out")
    		case deviceDetails := <-ch:
    			log.Infof("syn from csbi %v", deviceDetails.ID)
    			id, err := uuid.Parse(deviceDetails.ID)
    			if err != nil {
    				panic(err)
    			}
    			csbiTransportOptions := &tpb.TransportOption{
    				Address:         deviceDetails.Address,
    				Username:        opt.Username,
    				Password:        opt.Password,
    				Tls:             opt.Tls,
    				Type:            opt.Type,
    				TransportOption: opt.TransportOption,
    			}
    			log.WithField("transport option", csbiTransportOptions).Debug("gosdn gnmi transport options")
    
    			d, err := NewDevice(name, uuid.Nil, csbiTransportOptions, csbi)
    
    Malte Bauch's avatar
    Malte Bauch committed
    			if err != nil {
    				panic(err)
    			}
    			d.(*CsbiDevice).UUID = id
    			ch <- store.DeviceDetails{TransportOption: opt}
    			if err := pnd.devices.Add(d, d.Name()); err != nil {
    				panic(err)
    			}
    
    		}
    		pnd.callback(id, nil)
    		close(ch)
    	}()
    	return nil
    }
    
    Malte Bauch's avatar
    Malte Bauch committed
    func (pnd *pndImplementation) requestPlugin(name string, opt *tpb.TransportOption) (southbound.SouthboundInterface, error) {
    	ctx, cancel := context.WithTimeout(context.Background(), time.Minute*10)
    	defer cancel()
    	req := &cpb.CreateRequest{
    		Timestamp:       time.Now().UnixNano(),
    		TransportOption: []*tpb.TransportOption{opt},
    	}
    	client, err := pnd.csbiClient.CreatePlugin(ctx, req)
    	if err != nil {
    		return nil, err
    	}
    
    	id := uuid.New()
    	f, err := os.Create("plugin-" + id.String() + ".so")
    	if err != nil {
    		return nil, err
    	}
    	defer f.Close()
    	for {
    		payload, err := client.Recv()
    		if err != nil {
    			if err == io.EOF {
    				break
    			}
    			client.CloseSend()
    			return nil, err
    		}
    		n, err := f.Write(payload.Chunk)
    		if err != nil {
    			client.CloseSend()
    			return nil, err
    		}
    		log.WithField("n", n).Trace("wrote bytes")
    	}
    	if err := f.Sync(); err != nil {
    		return nil, err
    	}
    
    	return loadPlugin(id)
    }
    
    func loadPlugin(id uuid.UUID) (southbound.SouthboundInterface, error) {
    	p, err := plugin.Open("plugin-" + id.String() + ".so")
    	if err != nil {
    		return nil, err
    	}
    
    	symbol, err := p.Lookup("PluginSymbol")
    	if err != nil {
    		return nil, err
    	}
    
    	var sbi southbound.SouthboundInterface
    	sbi, ok := symbol.(southbound.SouthboundInterface)
    	if !ok {
    
    		return nil, &errors.ErrInvalidTypeAssertion{
    			Value: symbol,
    			Type:  (*southbound.SouthboundInterface)(nil),
    		}
    
    Malte Bauch's avatar
    Malte Bauch committed
    	}
    	log.WithFields(log.Fields{
    		"identifier": sbi.SbiIdentifier(),
    		"id":         sbi.ID(),
    		"type":       sbi.Type(),
    	}).Trace("plugin information")
    	return sbi, nil
    }
    
    
    func (pnd *pndImplementation) loadStoredDevices() error {
    	devices, err := pnd.devices.Load()
    	if err != nil {
    		return err
    	}
    
    	for _, device := range devices {
    		err := pnd.AddDeviceFromStore(
    			device.Name,
    			device.DeviceID,
    			&tpb.TransportOption{
    				Address:  device.TransportAddress,
    				Username: device.TransportUsername,
    				Password: device.TransportPassword,
    				TransportOption: &tpb.TransportOption_GnmiTransportOption{
    					GnmiTransportOption: &tpb.GnmiTransportOption{},
    				},
    
    Malte Bauch's avatar
    Malte Bauch committed
    				Type: spb.Type_OPENCONFIG,
    
    			}, device.SBI)
    		if err != nil {
    			return err
    		}
    	}
    	return nil
    }