submanagement.go

package server

import (
	"context"
	"time"

	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.
type SubManagementServer struct {
	subpb.UnimplementedSubscriptionManagementServiceServer
	networkElementWatchter *nucleus.NetworkElementWatcher
}

// NewSubManagementServer returns a new SubManagementServer.
func NewSubManagementServer(networkElementWatchter *nucleus.NetworkElementWatcher) *SubManagementServer {
	return &SubManagementServer{
		networkElementWatchter: networkElementWatchter,
	}
}

// ResetAllSubscriptions stops all running subscriptions and restarts them the same way they ran before.
func (s *SubManagementServer) ResetAllSubscriptions(ctx context.Context, request *subpb.ResetAllSubscriptionsRequest) (*subpb.ResetAllSubscriptionsResponse, error) {
	subscriptions := s.networkElementWatchter.GetAllSubscriptionInformations()

	s.networkElementWatchter.StopAndRemoveAllNetworkElementSubscriptions()

	s.networkElementWatchter.SubscribeToNetworkElements(subscriptions)

	return &subpb.ResetAllSubscriptionsResponse{
		Timestamp: time.Now().UnixNano(),
	}, nil
}

// GetAll returns information about all running subscriptions.
func (s *SubManagementServer) GetAll(ctx context.Context, request *subpb.GetAllRequest) (*subpb.GetAllResponse, error) {
	subInfos := s.networkElementWatchter.GetAllSubscriptionInformations()

	subInfosToReturn := make([]*subpb.Subscription, 0)

	for _, info := range subInfos {
		tmpPaths := make([]*subpb.Path, 0)
		for _, path := range info.Opts.Paths {
			tmpPaths = append(tmpPaths, &subpb.Path{
				Elem: path,
			})
		}

		subInfosToReturn = append(subInfosToReturn, &subpb.Subscription{
			Subid:   info.SubID,
			Pid:     info.PndID,
			Mneid:   info.MneID,
			MneName: info.MneName,
			Paths:   tmpPaths,
			SubscribeOptions: &subpb.SubscribeOptions{
				GnmiMode:       info.Opts.Mode,
				GnmiStreamMode: info.Opts.StreamMode,
				SampleInterval: info.Opts.SampleInterval,
			},
		})
	}

	return &subpb.GetAllResponse{
		Timestamp:     time.Now().UnixNano(),
		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
}