deviceWatcher.go

package nucleus

import (
	"context"
	"fmt"

	"code.fbi.h-da.de/danet/gosdn/controller/interfaces/device"
	"code.fbi.h-da.de/danet/gosdn/controller/interfaces/networkdomain"
	"code.fbi.h-da.de/danet/gosdn/controller/nucleus/types"
	"code.fbi.h-da.de/danet/gosdn/forks/goarista/gnmi"
	gpb "github.com/openconfig/gnmi/proto/gnmi"
	"github.com/sirupsen/logrus"
	log "github.com/sirupsen/logrus"
)

const (
	subscribeSampleInterval uint64 = 1000000000 // 1 second in nanoseconds
	// TODO: These gNMI options are adjusted to arista gNMI fork. Change when switching to native gNMI
	gNMISubscribeMode string = "stream"
	gNMIStreamMode    string = "sample"
)

// DeviceWatcher is a component that subscribes to devices via gNMI from within the controller and handles
// responses by triggering the internal event process.
type DeviceWatcher struct {
	pndStore networkdomain.PndStore
}

// NewDeviceWatcher takes a pndStore to subscribe to device paths.
func NewDeviceWatcher(pndStore networkdomain.PndStore) *DeviceWatcher {
	return &DeviceWatcher{
		pndStore: pndStore,
	}
}

// SubToDevices subscribes to every available device in each network domain with fixed gNMI subscription options (streaming in sample mode each second).
// Paths should be provided in the following format [][]string{{"system", "config", "hostname"}}
func (d *DeviceWatcher) SubToDevices(paths [][]string) {
	opts := &gnmi.SubscribeOptions{
		Mode:           gNMISubscribeMode,
		StreamMode:     gNMIStreamMode,
		Paths:          paths,
		SampleInterval: subscribeSampleInterval,
	}

	pnds, err := d.pndStore.GetAll()
	if err != nil {
		log.Error(err)
	}

	for _, pnd := range pnds {
		d.subscribeToPndDevices(pnd, opts)
	}
}

func (d *DeviceWatcher) subscribeToPndDevices(pnd networkdomain.NetworkDomain, opts *gnmi.SubscribeOptions) {
	for _, device := range pnd.Devices() {
		go d.callSubscribe(device, opts)
	}
}

func (d *DeviceWatcher) callSubscribe(device device.Device, opts *gnmi.SubscribeOptions) {
	ctx := context.Background()
	ctx = context.WithValue(ctx, types.CtxKeyOpts, opts)

	if err := device.Transport().SubscribeInternal(ctx, handleSubscribeRepsonse); err != nil {
		logrus.Error(err)
	}
}

func handleSubscribeRepsonse(resp *gpb.SubscribeResponse) {
	fmt.Printf("YEP HANDLER CALLED Type:%T ExampleMessage:%s\n", resp, resp.String())
}