principalNetworkDomain.go

package nucleus

import (
	"context"
	"encoding/json"
	"reflect"
	"time"

	cpb "code.fbi.h-da.de/cocsn/api/go/gosdn/csbi"
	ppb "code.fbi.h-da.de/cocsn/api/go/gosdn/pnd"
	tpb "code.fbi.h-da.de/cocsn/api/go/gosdn/transport"
	"google.golang.org/protobuf/proto"

	"code.fbi.h-da.de/cocsn/gosdn/forks/goarista/gnmi"
	"code.fbi.h-da.de/cocsn/gosdn/interfaces/change"
	"code.fbi.h-da.de/cocsn/gosdn/interfaces/device"
	"code.fbi.h-da.de/cocsn/gosdn/interfaces/networkdomain"
	"code.fbi.h-da.de/cocsn/gosdn/interfaces/southbound"
	"code.fbi.h-da.de/cocsn/gosdn/interfaces/store"
	"code.fbi.h-da.de/cocsn/gosdn/nucleus/errors"

	"github.com/google/uuid"
	"github.com/openconfig/ygot/ygot"
	"github.com/openconfig/ygot/ytypes"
	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 DeviceDetails)) (networkdomain.NetworkDomain, error) {
	pnd := &pndImplementation{
		name:             name,
		description:      description,
		sbic:             SbiStore{genericStore{}},
		devices:          NewDeviceStore(),
		pendingChanges:   ChangeStore{genericStore{}},
		committedChanges: ChangeStore{genericStore{}},
		confirmedChanges: ChangeStore{genericStore{}},
		id:               id,
		errChans:         make(map[uuid.UUID]chan error),

		csbiClient: c,
		callback:   callback,
	}
	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

	csbiClient cpb.CsbiClient
	callback   func(uuid.UUID, chan DeviceDetails)
}

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

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

func (pnd *pndImplementation) GetChange(cuid uuid.UUID, i ...int) (change.Change, error) {
	var index int
	if len(i) == 1 {
		index = i[0]
	} else if len(i) > 1 {
		return nil, errors.ErrInvalidParameters{
			Func:  pnd.GetChange,
			Param: "length of 'i' cannot be greater than '1'",
		}
	}
	stores := []*ChangeStore{
		&pnd.pendingChanges,
		&pnd.committedChanges,
		&pnd.confirmedChanges,
	}
	ch, err := stores[index].GetChange(cuid)
	index++
	if err != nil {
		switch err.(type) {
		case *errors.ErrNotFound:
			c, err := pnd.GetChange(cuid, index)
			if err != nil {
				return nil, err
			}
			var ok bool
			ch, ok = c.(*Change)
			if !ok {
				return nil, &errors.ErrInvalidTypeAssertion{
					Value: c,
					Type:  reflect.TypeOf(&Change{}),
				}
			}

		default:
			return nil, err
		}
	}
	return ch, err
}

func (pnd *pndImplementation) Commit(u uuid.UUID) error {
	ch, err := pnd.pendingChanges.GetChange(u)
	if err != nil {
		return err
	}
	if err := ch.Commit(); err != nil {
		return err
	}
	go func() {
		for {
			select {
			case err := <-pnd.errChans[u]:
				if err != nil {
					handleRollbackError(ch.ID(), err)
				}
			case <-ch.done:
			}
		}
	}()
	if err := pnd.committedChanges.Add(ch); err != nil {
		return err
	}
	return pnd.pendingChanges.Delete(u)
}

func (pnd *pndImplementation) Confirm(u uuid.UUID) error {
	ch, err := pnd.committedChanges.GetChange(u)
	if err != nil {
		return err
	}
	if err := ch.Confirm(); err != nil {
		return err
	}
	if err := pnd.confirmedChanges.Add(ch); 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() store.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
// 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)
}

//AddDevice adds a new device to the PND
func (pnd *pndImplementation) AddDevice(name string, opt *tpb.TransportOption, sid uuid.UUID) error {
	if opt.Csbi {
		return pnd.handleCsbiEnrolment(name, opt)
	}

	sbi, err := pnd.sbic.GetSBI(sid)
	if err != nil {
		return err
	}

	d, err := NewDevice(name, 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(FromString(identifier))
	if err != nil {
		return nil, err
	}

	copiedGoStruct, err := ygot.DeepCopy(d.Model())
	if err != nil {
		return nil, err
	}

	copiedDevice := &CommonDevice{name: d.Name(), UUID: d.ID(), GoStruct: copiedGoStruct}

	return copiedDevice, nil
}

// 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 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 {
	err := pnd.devices.Add(device, device.Name())
	if err != nil {
		return err
	}

	return nil
}

func (pnd *pndImplementation) removeDevice(id uuid.UUID) error {
	d, err := pnd.devices.GetDevice(id)
	if err != nil {
		return err
	}
	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(FromString(identifier))
	if err != nil {
		return "", err
	}

	jsonTree, err := json.MarshalIndent(foundDevice.Model(), "", "\t")
	if err != nil {
		return "", err
	}
	log.WithFields(log.Fields{
		"pnd":        pnd.id,
		"Identifier": identifier,
		"Name":       foundDevice.Name,
	}).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.devices.GetDevice(FromString(uuid.String()))
	if err != nil {
		return err
	}
	ctx := context.Background()
	res, err := d.Transport().Get(ctx, path)
	if err != nil {
		return err
	}
	err = d.ProcessResponse(res.(proto.Message))
	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 times out after the specified timeout period
func (pnd *pndImplementation) ChangeOND(uuid uuid.UUID, operation ppb.ApiOperation, path string, value ...string) error {
	d, err := pnd.devices.GetDevice(uuid)
	if err != nil {
		return err
	}
	cpy, err := ygot.DeepCopy(d.Model())
	ygot.BuildEmptyTree(cpy)
	if err != nil {
		return err
	}

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

	if operation != ppb.ApiOperation_DELETE && len(value) != 1 {
		return &errors.ErrInvalidParameters{
			Func:  pnd.ChangeOND,
			Param: value,
		}
	}

	switch operation {
	case ppb.ApiOperation_UPDATE, ppb.ApiOperation_REPLACE:
		typedValue := gnmi.TypedValue(value[0])
		if err := ytypes.SetNode(d.SBI().Schema().RootSchema(), cpy, p, typedValue); err != nil {
			return err
		}
	case ppb.ApiOperation_DELETE:
		if err := ytypes.DeleteNode(d.SBI().Schema().RootSchema(), cpy, p); err != nil {
			return err
		}
	default:
		return &errors.ErrOperationNotSupported{Op: operation}
	}

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

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

	return pnd.pendingChanges.Add(ch)
}

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 {
		if err := pnd.createCsbiDevice(name, d, opt); err != nil {
			log.Error(err)
		}
	}
	return nil
}

// DeviceDetails contains details of a device used by the cSBI mechanism
type DeviceDetails struct {
	ID              string
	Address         string
	TransportOption *tpb.TransportOption
}

func (pnd *pndImplementation) createCsbiDevice(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 DeviceDetails, 1)
	pnd.callback(id, ch)
	go func() {
		deviceDetails := <-ch
		log.Infof("syn from csbi %v", deviceDetails.ID)
		id, err := uuid.Parse(deviceDetails.ID)
		if err != nil {
			panic(err)
		}
		opt.Address = deviceDetails.Address

		d, err := NewDevice(name, opt, csbi)
		if err != nil {
			panic(err)
		}
		d.(*CsbiDevice).UUID = id
		ch <- DeviceDetails{TransportOption: opt}
		if err := pnd.devices.Add(d, d.Name()); err != nil {
			panic(err)
		}
		pnd.callback(id, nil)
		close(ch)
	}()
	return nil
}