Resolve "Implement missing functionality of subscription management"

See merge request !1080

cli/cmd/subManagement.go

@@ -43,6 +43,8 @@ var subCmd = &cobra.Command{
 	Long:    `The subscription command contains all sub-commands for subscription management. It has no functionality in itself.`,
 }
 
+var subUUID string
+
 func init() {
 	rootCmd.AddCommand(subCmd)
 }

cli/cmd/subManagementAdd.go

+/*
+Copyright © 2021 da/net Research Group <danet@h-da.de>
+All rights reserved.
+
+Redistribution and use in source and binary forms, with or without
+modification, are permitted provided that the following conditions are met:
+
+1. Redistributions of source code must retain the above copyright notice,
+   this list of conditions and the following disclaimer.
+
+2. Redistributions in binary form must reproduce the above copyright notice,
+   this list of conditions and the following disclaimer in the documentation
+   and/or other materials provided with the distribution.
+
+3. Neither the name of the copyright holder nor the names of its contributors
+   may be used to endorse or promote products derived from this software
+   without specific prior written permission.
+
+THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
+AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
+IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
+ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE
+LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
+CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
+SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
+INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
+CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
+ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
+POSSIBILITY OF SUCH DAMAGE.
+*/
+
+package cmd
+
+import (
+	"code.fbi.h-da.de/danet/gosdn/controller/api"
+	"github.com/pterm/pterm"
+	"github.com/spf13/cobra"
+	"github.com/spf13/viper"
+)
+
+// subAddCmd represents the add one subscription command.
+var subAddCmd = &cobra.Command{
+	Use:   "add",
+	Short: "Add creates a new subscription for the network element matching the provided ID using the provided subscribe options.",
+	Long:  "Add creates a new subscription for the network element matching the provided ID using the provided subscribe options.",
+	Run: func(cmd *cobra.Command, args []string) {
+		spinner, _ := pterm.DefaultSpinner.Start("Adding subscription")
+
+		// Note: Currently, doesn't support providing options/paths.
+		_, err := api.Add(createContextWithAuthorization(), viper.GetString("controllerAPIEndpoint"), mneUUID, nil)
+		if err != nil {
+			spinner.Fail(err)
+			return
+		}
+		spinner.Success()
+	},
+
+	PostRun: func(cmd *cobra.Command, args []string) {
+		// Necessary for prompt mode. The flag variables have to be resetted,
+		// since in prompt mode the program keeps running.
+		mneUUID = ""
+	},
+}
+
+func init() {
+	subCmd.AddCommand(subAddCmd)
+
+	subAddCmd.Flags().StringVar(&mneUUID, "uuid", "", "network element UUID")
+}

cli/cmd/subManagementDelete.go

+/*
+Copyright © 2021 da/net Research Group <danet@h-da.de>
+All rights reserved.
+
+Redistribution and use in source and binary forms, with or without
+modification, are permitted provided that the following conditions are met:
+
+1. Redistributions of source code must retain the above copyright notice,
+   this list of conditions and the following disclaimer.
+
+2. Redistributions in binary form must reproduce the above copyright notice,
+   this list of conditions and the following disclaimer in the documentation
+   and/or other materials provided with the distribution.
+
+3. Neither the name of the copyright holder nor the names of its contributors
+   may be used to endorse or promote products derived from this software
+   without specific prior written permission.
+
+THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
+AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
+IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
+ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE
+LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
+CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
+SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
+INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
+CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
+ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
+POSSIBILITY OF SUCH DAMAGE.
+*/
+
+package cmd
+
+import (
+	"code.fbi.h-da.de/danet/gosdn/controller/api"
+	"github.com/pterm/pterm"
+	"github.com/spf13/cobra"
+	"github.com/spf13/viper"
+)
+
+// subDeleteCmd represents the delete one subscription command.
+var subDeleteCmd = &cobra.Command{
+	Use:   "delete",
+	Short: "Stops and removes one running gNMI subscription.",
+	Long:  "Stops and removes one running gNMI subscription.",
+	Run: func(cmd *cobra.Command, args []string) {
+		spinner, _ := pterm.DefaultSpinner.Start("Removing subscription")
+
+		_, err := api.Delete(createContextWithAuthorization(), viper.GetString("controllerAPIEndpoint"), subUUID)
+		if err != nil {
+			pterm.Error.Println(err)
+			return
+		}
+
+		spinner.Success()
+	},
+	PostRun: func(cmd *cobra.Command, args []string) {
+		// Necessary for prompt mode. The flag variables have to be resetted,
+		// since in prompt mode the program keeps running.
+		subUUID = ""
+	},
+}
+
+func init() {
+	subCmd.AddCommand(subDeleteCmd)
+
+	subDeleteCmd.Flags().StringVar(&subUUID, "uuid", "", "UUID of the subcription")
+}

cli/cmd/subManagementGet.go

+/*
+Copyright © 2021 da/net Research Group <danet@h-da.de>
+All rights reserved.
+
+Redistribution and use in source and binary forms, with or without
+modification, are permitted provided that the following conditions are met:
+
+1. Redistributions of source code must retain the above copyright notice,
+   this list of conditions and the following disclaimer.
+
+2. Redistributions in binary form must reproduce the above copyright notice,
+   this list of conditions and the following disclaimer in the documentation
+   and/or other materials provided with the distribution.
+
+3. Neither the name of the copyright holder nor the names of its contributors
+   may be used to endorse or promote products derived from this software
+   without specific prior written permission.
+
+THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
+AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
+IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
+ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE
+LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
+CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
+SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
+INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
+CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
+ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
+POSSIBILITY OF SUCH DAMAGE.
+*/
+
+package cmd
+
+import (
+	"fmt"
+
+	"code.fbi.h-da.de/danet/gosdn/controller/api"
+	"github.com/pterm/pterm"
+	"github.com/spf13/cobra"
+	"github.com/spf13/viper"
+)
+
+// subGetCmd represents the get one subscription command.
+var subGetCmd = &cobra.Command{
+	Use:   "get",
+	Short: "Fetches details about one specific currently running gNMI subscriptions.",
+	Long:  "Fetches details about one specific currently running gNMI subscriptions.",
+	Run: func(cmd *cobra.Command, args []string) {
+		spinner, _ := pterm.DefaultSpinner.Start("Fetching subscription")
+
+		resp, err := api.Get(createContextWithAuthorization(), viper.GetString("controllerAPIEndpoint"), subUUID)
+		if err != nil {
+			pterm.Error.Println(err)
+			return
+		}
+		spinner.Success()
+
+		data1 := pterm.TableData{[]string{"SubUUID", "MneID", "MneName", "PndID"}}
+		data1 = append(data1, []string{resp.Subscriptions.Subid, resp.Subscriptions.Mneid, resp.Subscriptions.MneName, resp.Subscriptions.Pid})
+		err = pterm.DefaultTable.WithHasHeader().WithData(data1).Render()
+		if err != nil {
+			return
+		}
+
+		pterm.Print(pterm.FgLightCyan.Sprintf("Subscribe Options\n"))
+		data2 := pterm.TableData{[]string{"GnmiMode", "GnmiStreamMode", "SampleInterval (ns)"}}
+		data2 = append(data2, []string{resp.Subscriptions.SubscribeOptions.GnmiMode, resp.Subscriptions.SubscribeOptions.GnmiStreamMode, fmt.Sprint(resp.Subscriptions.SubscribeOptions.SampleInterval)})
+		err = pterm.DefaultTable.WithHasHeader().WithData(data2).Render()
+		if err != nil {
+			return
+		}
+
+		data3 := pterm.TableData{[]string{"Paths"}}
+		for _, path := range resp.Subscriptions.Paths {
+			data3 = append(data3, path.Elem)
+		}
+
+		err = pterm.DefaultTable.WithHasHeader().WithData(data3).Render()
+		if err != nil {
+			return
+		}
+	},
+	PostRun: func(cmd *cobra.Command, args []string) {
+		// Necessary for prompt mode. The flag variables have to be resetted,
+		// since in prompt mode the program keeps running.
+		subUUID = ""
+	},
+}
+
+func init() {
+	subCmd.AddCommand(subGetCmd)
+
+	subGetCmd.Flags().StringVar(&subUUID, "uuid", "", "UUID of the subcription")
+}

controller/api/subManagement.go

@@ -35,3 +35,49 @@ func GetAll(ctx context.Context, addr string) (*subpb.GetAllResponse, error) {
 
 	return subClient.GetAll(ctx, req)
 }
+
+// Get fetches detailed info for one specific available subscription.
+func Get(ctx context.Context, addr string, subID string) (*subpb.GetResponse, error) {
+	subClient, err := nbi.SubManagementClient(addr, dialOptions...)
+	if err != nil {
+		return nil, err
+	}
+
+	req := &subpb.GetRequest{
+		Timestamp: time.Now().UnixNano(),
+		Subid:     subID,
+	}
+
+	return subClient.Get(ctx, req)
+}
+
+// Delete stops and removes one specific running subscription is a no-op if object connected to ID does not exist.
+func Delete(ctx context.Context, addr string, subID string) (*subpb.DeleteResponse, error) {
+	subClient, err := nbi.SubManagementClient(addr, dialOptions...)
+	if err != nil {
+		return nil, err
+	}
+
+	req := &subpb.DeleteRequest{
+		Timestamp: time.Now().UnixNano(),
+		Subid:     subID,
+	}
+
+	return subClient.Delete(ctx, req)
+}
+
+// Add creates a new subscription for the network element matching the provided ID using the provided subscribe options.
+func Add(ctx context.Context, addr string, mneID string, subscribeOptions *subpb.Subscription) (*subpb.AddResponse, error) {
+	subClient, err := nbi.SubManagementClient(addr, dialOptions...)
+	if err != nil {
+		return nil, err
+	}
+
+	req := &subpb.AddRequest{
+		Timestamp:    time.Now().UnixNano(),
+		Mneid:        mneID,
+		Subscription: subscribeOptions,
+	}
+
+	return subClient.Add(ctx, req)
+}

controller/controller.go

@@ -361,6 +361,9 @@ func ensureAdminRoleExists() error {
 			"/gosdn.topology.TopologyService/DeleteLink",
 			"/gosdn.subscriptionmanagement.SubscriptionManagementService/ResetAllSubscriptions",
 			"/gosdn.subscriptionmanagement.SubscriptionManagementService/GetAll",
+			"/gosdn.subscriptionmanagement.SubscriptionManagementService/Get",
+			"/gosdn.subscriptionmanagement.SubscriptionManagementService/Delete",
+			"/gosdn.subscriptionmanagement.SubscriptionManagementService/Add",
 		}))
 		if err != nil {
 			return err

controller/northbound/server/submanagement.go

@@ -6,6 +6,8 @@ import (
 
 	subpb "code.fbi.h-da.de/danet/gosdn/api/go/gosdn/subscriptionmanagement"
 	"code.fbi.h-da.de/danet/gosdn/controller/nucleus"
+	"code.fbi.h-da.de/danet/gosdn/forks/goarista/gnmi"
+	"github.com/google/uuid"
 )
 
 // SubManagementServer represents a SubsriptionManagementServer.
@@ -67,3 +69,103 @@ func (s *SubManagementServer) GetAll(ctx context.Context, request *subpb.GetAllR
 		Subscriptions: subInfosToReturn,
 	}, nil
 }
+
+// Get fetches detailed info for one specific available subscription.
+func (s *SubManagementServer) Get(ctx context.Context, request *subpb.GetRequest) (*subpb.GetResponse, error) {
+	subUUID, err := uuid.Parse(request.GetSubid())
+	if err != nil {
+		return &subpb.GetResponse{
+			Timestamp: time.Now().UnixNano(),
+		}, err
+	}
+
+	info, err := s.networkElementWatchter.GetSubscriptionInformations(subUUID)
+	if err != nil {
+		return &subpb.GetResponse{
+			Timestamp: time.Now().UnixNano(),
+		}, err
+	}
+
+	var pathsToReturn []*subpb.Path
+	for _, path := range info.Opts.Paths {
+		var elems []string
+		for _, elem := range path {
+			elems = append(elems, elem)
+		}
+
+		pathsToReturn = append(pathsToReturn, &subpb.Path{Elem: elems})
+	}
+
+	return &subpb.GetResponse{
+		Timestamp: time.Now().UnixNano(),
+		Subscriptions: &subpb.Subscription{
+			Subid:   info.SubID,
+			Pid:     info.PndID,
+			Mneid:   info.MneID,
+			MneName: info.MneName,
+			Paths:   pathsToReturn,
+			SubscribeOptions: &subpb.SubscribeOptions{
+				GnmiMode:       info.Opts.Mode,
+				GnmiStreamMode: info.Opts.StreamMode,
+				SampleInterval: info.Opts.SampleInterval,
+			},
+		},
+	}, nil
+}
+
+// Delete stops and removes one specific running subscription is a no-op if object connected to ID does not exist.
+func (s *SubManagementServer) Delete(ctx context.Context, request *subpb.DeleteRequest) (*subpb.DeleteResponse, error) {
+	subUUID, err := uuid.Parse(request.GetSubid())
+	if err != nil {
+		return &subpb.DeleteResponse{
+			Timestamp: time.Now().UnixNano(),
+		}, err
+	}
+
+	s.networkElementWatchter.StopAndRemoveNetworkElementSubscription(subUUID)
+
+	return &subpb.DeleteResponse{
+		Timestamp: time.Now().UnixNano(),
+	}, nil
+}
+
+// Add creates a new subscription for the network element matching the provided ID using the provided subscribe options.
+func (s *SubManagementServer) Add(ctx context.Context, request *subpb.AddRequest) (*subpb.AddResponse, error) {
+	mneID, err := uuid.Parse(request.GetMneid())
+	if err != nil {
+		return &subpb.AddResponse{
+			Timestamp: time.Now().UnixNano(),
+		}, err
+	}
+
+	var gNMISubScribeOptions *gnmi.SubscribeOptions
+	if request.Subscription != nil {
+		var paths [][]string
+		for _, path := range request.Subscription.Paths {
+			var elems []string
+			for _, elem := range path.Elem {
+				elems = append(elems, elem)
+			}
+			paths = append(paths, elems)
+		}
+		gNMISubScribeOptions = &gnmi.SubscribeOptions{
+			Mode:           request.Subscription.SubscribeOptions.GnmiMode,
+			StreamMode:     request.Subscription.SubscribeOptions.GnmiStreamMode,
+			SampleInterval: request.Subscription.SubscribeOptions.SampleInterval,
+			Paths:          paths,
+		}
+	}
+
+	err = s.networkElementWatchter.SubscribeToNetworkElementWithID(mneID,
+		gNMISubScribeOptions,
+	)
+	if err != nil {
+		return &subpb.AddResponse{
+			Timestamp: time.Now().UnixNano(),
+		}, err
+	}
+
+	return &subpb.AddResponse{
+		Timestamp: time.Now().UnixNano(),
+	}, nil
+}

controller/nucleus/networkElementWatcher.go

@@ -14,6 +14,7 @@ import (
 	"code.fbi.h-da.de/danet/gosdn/controller/interfaces/networkelement"
 	"code.fbi.h-da.de/danet/gosdn/controller/interfaces/transport"
 	"code.fbi.h-da.de/danet/gosdn/controller/nucleus/types"
+	"code.fbi.h-da.de/danet/gosdn/controller/store"
 	"code.fbi.h-da.de/danet/gosdn/forks/goarista/gnmi"
 	"github.com/google/uuid"
 	gpb "github.com/openconfig/gnmi/proto/gnmi"
@@ -104,6 +105,27 @@ func (n *NetworkElementWatcher) SubscribeToNetworkElement(mne networkelement.Net
 	n.subscribeToNetworkElement(mne, opts)
 }
 
+// SubscribeToNetworkElementWithID subscribes to the network element matching the provided ID according to provided SubscribeOptions.
+// 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) SubscribeToNetworkElementWithID(mneID uuid.UUID, opts *gnmi.SubscribeOptions) error {
+	if opts == nil {
+		opts = &gnmi.SubscribeOptions{
+			Mode:           gNMISubscribeMode,
+			StreamMode:     gNMIStreamMode,
+			SampleInterval: subscribeSampleInterval,
+		}
+	}
+
+	mne, err := n.mneService.Get(store.Query{ID: mneID})
+	if err != nil {
+		log.Error(err)
+		return err
+	}
+
+	n.subscribeToNetworkElement(mne, opts)
+	return nil
+}
+
 func (n *NetworkElementWatcher) subscribeToNetworkElement(mne networkelement.NetworkElement, opts *gnmi.SubscribeOptions) {
 	subID := uuid.New()