Resolve "Subscribe to network element on creation in controller"

See merge request !564

controller/api/initialise_test.go

@@ -206,6 +206,8 @@ func bootstrapUnitTest() {
 		conflict.Metadata{ResourceVersion: 0})
 	_ = networkElementService.Add(mne)
 
+	networkElementWatcher := nucleus.NewNetworkElementWatcher(networkElementService, eventService)
+
 	pndService := &mocks.PndService{}
 	pndService.On("GetAll").Return([]networkdomain.NetworkDomain{}, nil)
 	pndService.On("Add", mock.Anything).Return(nil)
@@ -234,6 +236,7 @@ func bootstrapUnitTest() {
 		rpb.NewPluginRegistryServiceClient(&grpc.ClientConn{}),
 		csbi.NewCsbiServiceClient(&grpc.ClientConn{}),
 		func(u uuid.UUID, c chan networkelement.Details) {},
+		networkElementWatcher,
 	)
 
 	ppb.RegisterPndServiceServer(s, northbound.Pnd)

controller/controller.go

@@ -213,6 +213,7 @@ func startGrpc() error {
 		c.pluginRegistryClient,
 		c.csbiClient,
 		callback,
+		c.networkElementWatcher,
 	)
 
 	ppb.RegisterPndServiceServer(c.grpcServer, c.nbi.Pnd)

controller/northbound/server/nbi.go

@@ -8,6 +8,7 @@ import (
 	"code.fbi.h-da.de/danet/gosdn/controller/interfaces/networkelement"
 	"code.fbi.h-da.de/danet/gosdn/controller/interfaces/plugin"
 	rbacInterfaces "code.fbi.h-da.de/danet/gosdn/controller/interfaces/rbac"
+	"code.fbi.h-da.de/danet/gosdn/controller/nucleus"
 	"code.fbi.h-da.de/danet/gosdn/controller/rbac"
 	"code.fbi.h-da.de/danet/gosdn/controller/store"
 	"code.fbi.h-da.de/danet/gosdn/controller/topology"
@@ -58,6 +59,7 @@ func NewNBI(
 	pluginRegistryClient rpb.PluginRegistryServiceClient,
 	csbiClient cpb.CsbiServiceClient,
 	pndCallbackFn func(uuid.UUID, chan networkelement.Details),
+	networkElementWatchter *nucleus.NetworkElementWatcher,
 ) *NorthboundInterface {
 	protoValidator, err := protovalidate.New()
 	if err != nil {
@@ -73,7 +75,7 @@ func NewNBI(
 		Role:                    NewRoleServer(&jwt, roles, protoValidator),
 		Topology:                NewTopologyServer(topologyService, nodeService, portService, protoValidator),
 		App:                     NewAppServer(apps, protoValidator),
-		NetworkElement:          NewNetworkElementServer(mneService, pndService, pluginService, changeStore, protoValidator),
+		NetworkElement:          NewNetworkElementServer(mneService, pndService, pluginService, changeStore, protoValidator, networkElementWatchter),
 		Routes:                  NewRoutingTableServiceServer(routeService, nodeService, portService, protoValidator),
 		ConfigurationManagement: NewConfigurationManagementServer(pndService, mneService, topologyService, nodeService, portService, pluginService, protoValidator),
 	}

controller/northbound/server/networkElement.go

@@ -9,6 +9,7 @@ import (
 	mnepb "code.fbi.h-da.de/danet/gosdn/api/go/gosdn/networkelement"
 	ppb "code.fbi.h-da.de/danet/gosdn/api/go/gosdn/pnd"
 	tpb "code.fbi.h-da.de/danet/gosdn/api/go/gosdn/transport"
+	"code.fbi.h-da.de/danet/gosdn/controller/config"
 	"code.fbi.h-da.de/danet/gosdn/controller/conflict"
 	"code.fbi.h-da.de/danet/gosdn/controller/customerrs"
 	"code.fbi.h-da.de/danet/gosdn/controller/interfaces/change"
@@ -36,11 +37,12 @@ import (
 // NetworkElementServer represents a NetworkElementServer.
 type NetworkElementServer struct {
 	mnepb.UnimplementedNetworkElementServiceServer
-	mneService     networkelement.Service
-	pndService     networkdomain.Service
-	pluginService  plugin.Service
-	changeStore    store.ChangeStore
-	protoValidator *protovalidate.Validator
+	mneService             networkelement.Service
+	pndService             networkdomain.Service
+	pluginService          plugin.Service
+	changeStore            store.ChangeStore
+	protoValidator         *protovalidate.Validator
+	networkElementWatchter *nucleus.NetworkElementWatcher
 }
 
 // NewNetworkElementServer returns a new NetWorkElementServer.
@@ -50,13 +52,15 @@ func NewNetworkElementServer(
 	pluginService plugin.Service,
 	changeStore store.ChangeStore,
 	protoValidator *protovalidate.Validator,
+	networkElementWatchter *nucleus.NetworkElementWatcher,
 ) *NetworkElementServer {
 	return &NetworkElementServer{
-		mneService:     mneService,
-		pndService:     pndService,
-		pluginService:  pluginService,
-		changeStore:    changeStore,
-		protoValidator: protoValidator,
+		mneService:             mneService,
+		pndService:             pndService,
+		pluginService:          pluginService,
+		changeStore:            changeStore,
+		protoValidator:         protoValidator,
+		networkElementWatchter: networkElementWatchter,
 	}
 }
 
@@ -756,6 +760,8 @@ func (n *NetworkElementServer) addMne(ctx context.Context, name string, opt *tpb
 		if err != nil {
 			return uuid.Nil, err
 		}
+
+		n.networkElementWatchter.SubscribeToNetworkElement(mne, config.GetGnmiSubscriptionPaths(), nil)
 	} else {
 		return uuid.Nil, status.Errorf(codes.InvalidArgument, "invalid transport data provided")
 	}

controller/nucleus/networkElementWatcher.go

@@ -74,6 +74,21 @@ func (n *NetworkElementWatcher) SubscribeToNetworkElements(paths [][]string, opt
 	}
 }
 
+// SubscribeToNetworkElement subscribes to the provided network element according to provided SubscribeOptions.
+// Paths should be provided in the following format [][]string{{"system", "config", "hostname"}}
+// SubscribeOptions can be nil. Use nil for a fixed, pre-defined set of gNMI subscription options (streaming in sample mode each second).
+func (n *NetworkElementWatcher) SubscribeToNetworkElement(mne networkelement.NetworkElement, paths [][]string, opts *gnmi.SubscribeOptions) {
+	if opts == nil {
+		opts = &gnmi.SubscribeOptions{
+			Mode:           gNMISubscribeMode,
+			StreamMode:     gNMIStreamMode,
+			Paths:          paths,
+			SampleInterval: subscribeSampleInterval,
+		}
+	}
+	n.subscribeToNetworkElement(mne, opts)
+}
+
 func (n *NetworkElementWatcher) subscribeToNetworkElement(mne networkelement.NetworkElement, opts *gnmi.SubscribeOptions) {
 	subID := uuid.New()