principalNetworkDomain.go

package nucleus

import (
	"code.fbi.h-da.de/cocsn/gosdn/forks/goarista/gnmi"
	. "code.fbi.h-da.de/cocsn/gosdn/nucleus/pnd" //nolint
	"context"
	"github.com/openconfig/ygot/ygot"
	"github.com/openconfig/ygot/ytypes"
	log "github.com/sirupsen/logrus"

	"encoding/json"
	"github.com/google/uuid"
)

// PrincipalNetworkDomain provides an
// interface for PND implementations
type PrincipalNetworkDomain interface {
	Destroy() error
	AddSbi(interface{}) error
	RemoveSbi(uuid.UUID) error
	AddDevice(interface{}) error
	GetDevice(uuid uuid.UUID) (ygot.GoStruct, error)
	RemoveDevice(uuid.UUID) error
	Devices() []uuid.UUID
	ChangeOND(uuid uuid.UUID, operation interface{}, path string, value ...string) error
	Request(uuid.UUID, string) error
	RequestAll(string) error
	GetName() string
	GetDescription() string
	MarshalDevice(uuid.UUID) (string, error)
	ContainsDevice(uuid.UUID) bool
	GetSBIs() interface{}
	ID() uuid.UUID
	Pending() []uuid.UUID
	Committed() []uuid.UUID
	Commit(uuid.UUID) error
	Confirm(uuid.UUID) error
}

// NewPND creates a Principle Network Domain
func NewPND(name, description string, id uuid.UUID, sbi SouthboundInterface) (PrincipalNetworkDomain, error) {
	pnd := &pndImplementation{
		name:             name,
		description:      description,
		sbic:             sbiStore{store{}},
		devices:          deviceStore{store{}},
		pendingChanges:   changeStore{store{}},
		committedChanges: changeStore{store{}},
		confirmedChanges: changeStore{store{}},
		id:               id,
		errChans:         make(map[uuid.UUID]chan error),
	}
	if err := pnd.sbic.add(sbi); err != nil {
		return nil, err
	}
	return pnd, nil
}

type pndImplementation struct {
	name             string
	description      string
	sbic             sbiStore
	devices          deviceStore
	pendingChanges   changeStore
	committedChanges changeStore
	confirmedChanges changeStore
	id               uuid.UUID
	errChans         map[uuid.UUID]chan error
}

func (pnd *pndImplementation) Pending() []uuid.UUID {
	return pnd.pendingChanges.UUIDs()
}

func (pnd *pndImplementation) Committed() []uuid.UUID {
	return pnd.committedChanges.UUIDs()
}

func (pnd *pndImplementation) Commit(u uuid.UUID) error {
	change, err := pnd.pendingChanges.get(u)
	if err != nil {
		return err
	}
	if err := change.Commit(); err != nil {
		return err
	}
	go func() {
		for {
			select {
			case err := <-pnd.errChans[u]:
				if err != nil {
					handleRollbackError(change.ID(), err)
				}
			case <-change.Done:
			}
		}
	}()
	if err := pnd.committedChanges.add(change); err != nil {
		return err
	}
	return pnd.pendingChanges.delete(u)
}

func (pnd *pndImplementation) Confirm(u uuid.UUID) error {
	change, err := pnd.committedChanges.get(u)
	if err != nil {
		return err
	}
	if err := change.Confirm(); err != nil {
		return err
	}
	if err := pnd.confirmedChanges.add(change); err != nil {
		return err
	}
	return pnd.committedChanges.delete(u)
}

func (pnd *pndImplementation) ID() uuid.UUID {
	return pnd.id
}

func (pnd *pndImplementation) Devices() []uuid.UUID {
	return pnd.devices.UUIDs()
}

// GetName returns the name of the PND
func (pnd *pndImplementation) GetName() string {
	return pnd.name
}

// 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 {
	return pnd.description
}

// GetSBIs returns the registered SBIs
func (pnd *pndImplementation) GetSBIs() interface{} {
	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(sbi interface{}) error {
	s, ok := sbi.(SouthboundInterface)
	if !ok {
		return &ErrInvalidTypeAssertion{
			v: sbi,
			t: "Device",
		}
	}
	return pnd.addSbi(s)
}

// RemoveSbi removes a SBI from the PND
// TODO: this should to recursivly through
// devices and remove the devices using
// this SBI
func (pnd *pndImplementation) RemoveSbi(id uuid.UUID) error {
	return pnd.removeSbi(id)
}

//AddDevice adds a new device to the PND
func (pnd *pndImplementation) AddDevice(device interface{}) error {
	d, ok := device.(*Device)
	if !ok {
		return &ErrInvalidTypeAssertion{
			v: device,
			t: "Device",
		}
	}
	return pnd.addDevice(d)
}

func (pnd *pndImplementation) GetDevice(uuid uuid.UUID) (ygot.GoStruct, error) {
	d, err := pnd.devices.get(uuid)
	if err != nil {
		return nil, err
	}
	return ygot.DeepCopy(d.GoStruct)
}

// RemoveDevice removes a device from the PND
func (pnd *pndImplementation) RemoveDevice(uuid uuid.UUID) error {
	return pnd.removeDevice(uuid)
}

// Actual implementation, bind to struct if
// neccessary
func destroy() error {
	return nil
}

func (pnd *pndImplementation) addSbi(sbi 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) error {
	return pnd.devices.add(device)
}

func (pnd *pndImplementation) getDevice(id uuid.UUID) (*Device, error) {
	return pnd.devices.get(id)
}

func (pnd *pndImplementation) removeDevice(id uuid.UUID) error {
	return pnd.devices.delete(id)
}

func (pnd *pndImplementation) MarshalDevice(uuid uuid.UUID) (string, error) {
	d, err := pnd.getDevice(uuid)
	if err != nil {
		return "", err
	}
	jsonTree, err := json.MarshalIndent(d.GoStruct, "", "\t")
	if err != nil {
		return "", err
	}
	log.WithFields(log.Fields{
		"pnd":    pnd.id,
		"device": uuid,
	}).Info("marshalled device")
	return string(jsonTree), nil
}

// Request sends a get request to a specific device
func (pnd *pndImplementation) Request(uuid uuid.UUID, path string) error {
	d, err := pnd.getDevice(uuid)
	if err != nil {
		return err
	}
	ctx := context.Background()
	res, err := d.Transport.Get(ctx, path)
	if err != nil {
		return err
	}
	err = d.Transport.ProcessResponse(res, d.GoStruct, d.SBI.Schema())
	if err != nil {
		return err
	}
	return nil
}

// RequestAll sends a request for all registered devices
func (pnd *pndImplementation) RequestAll(path string) error {
	for _, k := range pnd.devices.UUIDs() {
		if err := pnd.Request(k, path); err != nil {
			return err
		}
	}
	log.WithFields(log.Fields{
		"pnd":  pnd.id,
		"path": path,
	}).Info("sent request to all devices")
	return nil
}

// ChangeOND creates a change from the provided Operation, path and value. The Change is pending and
func (pnd *pndImplementation) ChangeOND(uuid uuid.UUID, operation interface{}, path string, value ...string) error {
	d, err := pnd.getDevice(uuid)
	if err != nil {
		return err
	}
	cpy, err := ygot.DeepCopy(d.GoStruct)
	ygot.BuildEmptyTree(cpy)

	p, err := ygot.StringToStructuredPath(path)
	if err != nil {
		return err
	}

	if operation != TransportDelete && len(value) != 1 {
		return &ErrInvalidParameters{
			f: pnd.ChangeOND,
			r: value,
		}
	}

	switch operation {
	case TransportUpdate, TransportReplace:
		typedValue := gnmi.TypedValue(value[0])
		if err := ytypes.SetNode(d.SBI.Schema().RootSchema(), cpy, p, typedValue); err != nil {
			return err
		}
	case TransportDelete:
		if err := ytypes.DeleteNode(d.SBI.Schema().RootSchema(), cpy, p); err != nil {
			return err
		}
	default:
		return &ErrOperationNotSupported{o: operation}
	}

	callback := func(state ygot.GoStruct, change ygot.GoStruct) error {
		ctx := context.Background()
		return d.Transport.Set(ctx, state, change)
	}

	errChan := make(chan error)
	change := NewChange(uuid, d.GoStruct, cpy, callback, errChan)
	pnd.errChans[change.ID()] = errChan

	return pnd.pendingChanges.add(change)
}

func handleRollbackError(id uuid.UUID, err error) {
	log.Error(err)
	// TODO: Notion of invalid state needed.
}