networkElement.go

package server

import (
	"context"
	"fmt"
	"strings"
	"time"

	mnepb "code.fbi.h-da.de/danet/gosdn/api/go/gosdn/networkelement"
	plugin_registry "code.fbi.h-da.de/danet/gosdn/api/go/gosdn/plugin-registry"
	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"
	"code.fbi.h-da.de/danet/gosdn/controller/interfaces/networkdomain"
	"code.fbi.h-da.de/danet/gosdn/controller/interfaces/networkelement"
	"code.fbi.h-da.de/danet/gosdn/controller/interfaces/plugin"
	"code.fbi.h-da.de/danet/gosdn/controller/metrics"
	"code.fbi.h-da.de/danet/gosdn/controller/nucleus"
	"code.fbi.h-da.de/danet/gosdn/controller/nucleus/types"
	util "code.fbi.h-da.de/danet/gosdn/controller/nucleus/util/gnmi"
	"code.fbi.h-da.de/danet/gosdn/controller/store"
	aGNMI "code.fbi.h-da.de/danet/gosdn/forks/goarista/gnmi"
	"github.com/bufbuild/protovalidate-go"
	"github.com/google/uuid"
	"github.com/hashicorp/go-multierror"
	"github.com/openconfig/gnmi/proto/gnmi"

	ygotutil "github.com/openconfig/ygot/util"
	"github.com/openconfig/ygot/ygot"
	"github.com/prometheus/client_golang/prometheus"
	log "github.com/sirupsen/logrus"
	"google.golang.org/grpc/codes"
	"google.golang.org/grpc/status"
	"google.golang.org/protobuf/proto"
)

// NetworkElementServer represents a NetworkElementServer.
type NetworkElementServer struct {
	mnepb.UnimplementedNetworkElementServiceServer
	mneService             networkelement.Service
	pndService             networkdomain.Service
	pluginService          plugin.Service
	changeStore            store.ChangeStore
	protoValidator         *protovalidate.Validator
	networkElementWatchter *nucleus.NetworkElementWatcher
}

// NewNetworkElementServer returns a new NetWorkElementServer.
func NewNetworkElementServer(
	mneService networkelement.Service,
	pndService networkdomain.Service,
	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,
		networkElementWatchter: networkElementWatchter,
	}
}

// Update updates a network element.
func (n *NetworkElementServer) Update(ctx context.Context, request *mnepb.UpdateNetworkElementRequest) (*mnepb.UpdateNetworkElementResponse, error) {
	if err := n.protoValidator.Validate(request); err != nil {
		return nil, status.Errorf(codes.Aborted, "%v", err)
	}

	mneID, err := uuid.Parse(request.NetworkElement.Id)
	if err != nil {
		return nil, err
	}

	err = n.mneService.UpdateModel(mneID, request.NetworkElement.Model)
	if err != nil {
		return nil, err
	}

	networkElement, err := n.mneService.Get(store.Query{ID: mneID})
	if err != nil {
		return nil, err
	}

	err = networkelement.EnsureIntendedConfigurationIsAppliedOnNetworkElement(networkElement)
	if err != nil {
		return nil, err
	}

	return &mnepb.UpdateNetworkElementResponse{
		Timestamp: time.Now().UnixNano(),
	}, nil
}

// Get gets a specific mne.
func (n *NetworkElementServer) Get(ctx context.Context, request *mnepb.GetRequest) (*mnepb.GetResponse, error) {
	if err := n.protoValidator.Validate(request); err != nil {
		return nil, status.Errorf(codes.Aborted, "%v", err)
	}

	labels := prometheus.Labels{"service": "mne", "rpc": "get"}
	start := metrics.StartHook(labels, grpcRequestsTotal)
	defer metrics.FinishHook(labels, start, grpcRequestDurationSecondsTotal, grpcRequestDurationSeconds)

	pnd, err := n.getPnd(request.Pid)
	if err != nil {
		log.Error(err)
		return nil, status.Errorf(codes.Aborted, "%v", err)
	}

	networkElement, err := n.getMne(request.Mneid)
	if err != nil {
		log.Error(err)
		return nil, status.Errorf(codes.Aborted, "%v", err)
	}

	mne := &mnepb.ManagedNetworkElement{
		Id:               networkElement.ID().String(),
		Name:             networkElement.Name(),
		TransportAddress: networkElement.TransportAddress(),
	}

	err = fillMneBySpecificPath(mne, networkElement, "/", false)
	if err != nil {
		log.Error(err)
		return nil, status.Errorf(codes.Aborted, "%v", err)
	}

	err = fillMneWithPluginInformation(mne, networkElement)
	if err != nil {
		log.Error(err)
		return nil, status.Errorf(codes.Aborted, "%v", err)
	}

	mne.Model, err = networkElement.GetModelAsString()
	if err != nil {
		log.Error(err)
		return nil, status.Errorf(codes.Aborted, "%v", err)
	}

	return &mnepb.GetResponse{
		Timestamp: time.Now().UnixNano(),
		Pnd: &ppb.PrincipalNetworkDomain{
			Id:          pnd.ID().String(),
			Name:        pnd.GetName(),
			Description: pnd.GetDescription(),
		},
		Mne: mne,
	}, nil
}

// GetFlattened gets a specific mne.
func (n *NetworkElementServer) GetFlattened(ctx context.Context, request *mnepb.GetFlattenedRequest) (*mnepb.GetFlattenedResponse, error) {
	if err := n.protoValidator.Validate(request); err != nil {
		return nil, status.Errorf(codes.Aborted, "%v", err)
	}

	labels := prometheus.Labels{"service": "pnd", "rpc": "get"}
	start := metrics.StartHook(labels, grpcRequestsTotal)
	defer metrics.FinishHook(labels, start, grpcRequestDurationSecondsTotal, grpcRequestDurationSeconds)

	networkElement, err := n.getMne(request.Mneid)
	if err != nil {
		log.Error(err)
		return nil, status.Errorf(codes.Aborted, "%v", err)
	}

	return &mnepb.GetFlattenedResponse{
		Timestamp: time.Now().UnixNano(),
		Mne: &mnepb.FlattenedManagedNetworkElement{
			Id:       networkElement.ID().String(),
			Name:     networkElement.Name(),
			Pid:      networkElement.PndID().String(),
			Pluginid: networkElement.GetPlugin().ID().String(),
		},
	}, nil
}

func (n *NetworkElementServer) getMne(identifier string) (networkelement.NetworkElement, error) {
	id, err := uuid.Parse(identifier)
	if err != nil {
		id = uuid.Nil
	}

	mne, err := n.mneService.Get(store.Query{
		ID: id,
	})
	if mne == nil {
		return nil, fmt.Errorf("no network element found")
	}
	if err != nil {
		return nil, err
	}

	return mne, nil
}

func (n *NetworkElementServer) getPnd(identifier string) (networkdomain.NetworkDomain, error) {
	pid, err := uuid.Parse(identifier)
	if err != nil {
		return nil, err
	}

	pnd, err := n.pndService.Get(store.Query{ID: pid})
	if err != nil {
		log.Error(err)
		return nil, status.Errorf(codes.Aborted, "%v", err)
	}

	return pnd, nil
}

// GetAll returns a list of existing mnes.
func (n *NetworkElementServer) GetAll(ctx context.Context, request *mnepb.GetAllRequest) (*mnepb.GetAllResponse, error) {
	if err := n.protoValidator.Validate(request); err != nil {
		return nil, status.Errorf(codes.Aborted, "%v", err)
	}

	labels := prometheus.Labels{"service": "mne", "rpc": "get"}
	start := metrics.StartHook(labels, grpcRequestsTotal)
	defer metrics.FinishHook(labels, start, grpcRequestDurationSecondsTotal, grpcRequestDurationSeconds)

	pnd, err := n.getPnd(request.Pid)
	if err != nil {
		log.Error(err)
		return nil, status.Errorf(codes.Aborted, "%v", err)
	}

	storedMNE, err := n.mneService.GetAll()
	if err != nil {
		return nil, err
	}

	mnes := make([]*mnepb.ManagedNetworkElement, len(storedMNE))
	for i, networkElement := range storedMNE {
		mne := &mnepb.ManagedNetworkElement{
			Id:               networkElement.ID().String(),
			Name:             networkElement.Name(),
			TransportAddress: networkElement.TransportAddress(),
		}

		err := fillMneBySpecificPath(mne, networkElement, "/", false)
		if err != nil {
			log.Error(err)
			return nil, status.Errorf(codes.Aborted, "%v", err)
		}

		err = fillMneWithPluginInformation(mne, networkElement)
		if err != nil {
			log.Error(err)
			return nil, status.Errorf(codes.Aborted, "%v", err)
		}
		mnes[i] = mne
	}

	return &mnepb.GetAllResponse{
		Timestamp: time.Now().UnixNano(),
		Pnd: &ppb.PrincipalNetworkDomain{
			Id:          pnd.ID().String(),
			Name:        pnd.GetName(),
			Description: pnd.GetDescription(),
		},
		Mne: mnes,
	}, nil
}

// GetAllFlattened returns a list of existing mnes.
func (n *NetworkElementServer) GetAllFlattened(ctx context.Context, request *mnepb.GetAllFlattenedRequest) (*mnepb.GetAllFlattenedResponse, error) {
	if err := n.protoValidator.Validate(request); err != nil {
		return nil, status.Errorf(codes.Aborted, "%v", err)
	}

	labels := prometheus.Labels{"service": "mne", "rpc": "get"}
	start := metrics.StartHook(labels, grpcRequestsTotal)
	defer metrics.FinishHook(labels, start, grpcRequestDurationSecondsTotal, grpcRequestDurationSeconds)

	pnd, err := n.getPnd(request.Pid)
	if err != nil {
		log.Error(err)
		return nil, status.Errorf(codes.Aborted, "%v", err)
	}

	mnes, err := n.mneService.GetAllAsLoaded()
	if err != nil {
		return nil, err
	}

	flattenedMnes := make([]*mnepb.FlattenedManagedNetworkElement, len(mnes))
	for i, mne := range mnes {
		mne := &mnepb.FlattenedManagedNetworkElement{
			Id:       mne.ID,
			Name:     mne.Name,
			Pluginid: mne.Plugin,
			Pid:      mne.PndID,
		}
		flattenedMnes[i] = mne
	}

	return &mnepb.GetAllFlattenedResponse{
		Timestamp: time.Now().UnixNano(),
		Pnd: &ppb.PrincipalNetworkDomain{
			Id:          pnd.ID().String(),
			Name:        pnd.GetName(),
			Description: pnd.GetDescription(),
		},
		Mne: flattenedMnes,
	}, nil
}

func fillMneWithPluginInformation(mne *mnepb.ManagedNetworkElement, nme networkelement.NetworkElement) error {
	plugin := nme.GetPlugin()
	pluginManifest := plugin.Manifest()

	mne.Plugin = &plugin_registry.Plugin{
		Id: plugin.ID().String(),
		Manifest: &plugin_registry.Manifest{
			Name:     pluginManifest.Name,
			Firmware: pluginManifest.Firmware,
			Author:   pluginManifest.Author,
			Version:  pluginManifest.Version,
		},
	}

	return nil
}

func fillMneBySpecificPath(mne *mnepb.ManagedNetworkElement, nme networkelement.NetworkElement, path string, requestForIntendedState bool) error {
	gnmiPath, err := ygot.StringToStructuredPath(path)
	if err != nil {
		log.Error(err)
		return status.Errorf(codes.Aborted, "%v", err)
	}

	plugin := nme.GetPlugin()

	notifications, err := plugin.GetNode(gnmiPath, requestForIntendedState)
	if err != nil {
		return status.Errorf(codes.Aborted, "%v", err)
	}

	mne.MneNotification = notifications

	return nil
}

// GetPath gets an actual state of the path on a mne.
func (n *NetworkElementServer) GetPath(ctx context.Context, request *mnepb.GetPathRequest) (*mnepb.GetPathResponse, error) {
	if err := n.protoValidator.Validate(request); err != nil {
		return nil, status.Errorf(codes.Aborted, "%v", err)
	}

	labels := prometheus.Labels{"service": "mne", "rpc": "get"}
	start := metrics.StartHook(labels, grpcRequestsTotal)
	defer metrics.FinishHook(labels, start, grpcRequestDurationSecondsTotal, grpcRequestDurationSeconds)

	pnd, err := n.getPnd(request.Pid)
	if err != nil {
		log.Error(err)
		return nil, status.Errorf(codes.Aborted, "%v", err)
	}

	networkElement, err := n.getMne(request.Mneid)
	if err != nil {
		log.Error(err)
		return nil, status.Errorf(codes.Aborted, "%v", err)
	}

	// In case we get the path from grpc-gateway we have to replace
	path := strings.ReplaceAll(request.Path, "||", "/")

	resp, err := n.getPath(ctx, networkElement, path)
	if err != nil {
		log.Error(err)
		return nil, status.Errorf(codes.Aborted, "%v", err)
	}

	return &mnepb.GetPathResponse{
		Timestamp: time.Now().UnixNano(),
		Pnd: &ppb.PrincipalNetworkDomain{
			Id:          pnd.ID().String(),
			Name:        pnd.GetName(),
			Description: pnd.GetDescription(),
		},
		MneNotification: resp.Notification,
	}, nil
}

func (n *NetworkElementServer) getPath(ctx context.Context, mne networkelement.NetworkElement, path string) (*gnmi.GetResponse, error) {
	res, err := mne.Transport().Get(ctx, path)
	if err != nil {
		return nil, err
	}

	protoMessage, ok := res.(proto.Message)
	if !ok {
		return nil, &customerrs.InvalidTypeAssertionError{
			Value: res,
			Type:  (*proto.Message)(nil),
		}
	}

	getResponse, ok := protoMessage.(*gnmi.GetResponse)
	if !ok {
		return nil, &customerrs.InvalidTypeAssertionError{
			Value: res,
			Type:  (*gnmi.GetResponse)(nil),
		}
	}

	return getResponse, nil
}

// GetIntendedPath gets a path as the intended state stored in the storage.
func (n *NetworkElementServer) GetIntendedPath(ctx context.Context, request *mnepb.GetIntendedPathRequest) (*mnepb.GetIntendedPathResponse, error) {
	if err := n.protoValidator.Validate(request); err != nil {
		return nil, status.Errorf(codes.Aborted, "%v", err)
	}

	pnd, err := n.getPnd(request.Pid)
	if err != nil {
		log.Error(err)
		return nil, status.Errorf(codes.Aborted, "%v", err)
	}

	// In case we get the path from grpc-gateway we have to replace
	intendedPath := strings.ReplaceAll(request.IntendedPath, "||", "/")

	networkElement, err := n.getMne(request.Mneid)
	if err != nil {
		log.Error(err)
		return nil, status.Errorf(codes.Aborted, "%v", err)
	}

	mne := &mnepb.ManagedNetworkElement{}

	err = fillMneBySpecificPath(mne, networkElement, intendedPath, true)
	if err != nil {
		return nil, err
	}

	return &mnepb.GetIntendedPathResponse{
		Timestamp: time.Now().UnixNano(),
		Pnd: &ppb.PrincipalNetworkDomain{
			Id:          pnd.ID().String(),
			Name:        pnd.GetName(),
			Description: pnd.GetDescription(),
		},
		MneNotification: mne.MneNotification,
	}, nil
}

// GetChange gets a specific change of a mne.
func (n *NetworkElementServer) GetChange(ctx context.Context, request *mnepb.GetChangeRequest) (*mnepb.GetChangeResponse, error) {
	if err := n.protoValidator.Validate(request); err != nil {
		return nil, status.Errorf(codes.Aborted, "%v", err)
	}

	labels := prometheus.Labels{"service": "mne", "rpc": "get"}
	start := metrics.StartHook(labels, grpcRequestsTotal)
	defer metrics.FinishHook(labels, start, grpcRequestDurationSecondsTotal, grpcRequestDurationSeconds)

	pnd, err := n.getPnd(request.Pid)
	if err != nil {
		log.Error(err)
		return nil, status.Errorf(codes.Aborted, "%v", err)
	}

	changes, err := n.fillChanges(false, request.Cuid...)
	if err != nil {
		log.Error(err)
		return nil, status.Errorf(codes.Aborted, "%v", err)
	}

	return &mnepb.GetChangeResponse{
		Timestamp: time.Now().UnixNano(),
		Pnd: &ppb.PrincipalNetworkDomain{
			Id:          pnd.ID().String(),
			Name:        pnd.GetName(),
			Description: pnd.GetDescription(),
		},
		Change: changes,
	}, nil
}

// GetChangeList gets all existing changes.
func (n *NetworkElementServer) GetChangeList(ctx context.Context, request *mnepb.GetChangeListRequest) (*mnepb.GetChangeListResponse, error) {
	if err := n.protoValidator.Validate(request); err != nil {
		return nil, status.Errorf(codes.Aborted, "%v", err)
	}

	labels := prometheus.Labels{"service": "mne", "rpc": "get"}
	start := metrics.StartHook(labels, grpcRequestsTotal)
	defer metrics.FinishHook(labels, start, grpcRequestDurationSecondsTotal, grpcRequestDurationSeconds)

	pnd, err := n.getPnd(request.Pid)
	if err != nil {
		log.Error(err)
		return nil, status.Errorf(codes.Aborted, "%v", err)
	}

	changes, err := n.fillChanges(true, "")
	if err != nil {
		log.Error(err)
		return nil, status.Errorf(codes.Aborted, "%v", err)
	}

	return &mnepb.GetChangeListResponse{
		Timestamp: time.Now().UnixNano(),
		Pnd: &ppb.PrincipalNetworkDomain{
			Id:          pnd.ID().String(),
			Name:        pnd.GetName(),
			Description: pnd.GetDescription(),
		},
		Change: changes,
	}, nil
}

func (n *NetworkElementServer) fillChanges(all bool, cuid ...string) ([]*mnepb.Change, error) {
	var changeList []uuid.UUID

	switch all {
	case true:
		changeList = n.changeStore.Pending()
		changeList = append(changeList, n.changeStore.Committed()...)
	default:
		var err error
		if len(cuid) == 0 {
			return nil, &customerrs.InvalidParametersError{
				Func:  n.fillChanges,
				Param: "length of 'mneID' cannot be '0' when 'all' is set to 'false'",
			}
		}
		changeList, err = stringArrayToUUIDs(cuid)
		if err != nil {
			log.Error(err)
			return nil, status.Errorf(codes.Aborted, "%v", err)
		}
	}

	changes := make([]*mnepb.Change, len(changeList))
	for i, ch := range changeList {
		c, err := n.changeStore.GetChange(ch)
		if err != nil {
			log.Error(err)
			return nil, status.Errorf(codes.Aborted, "%v", err)
		}

		// Diff could be added here.
		changes[i] = &mnepb.Change{
			Id:    ch.String(),
			Age:   c.Age().Microseconds(),
			State: c.State(),
		}
	}
	return changes, nil
}

func stringArrayToUUIDs(sid []string) ([]uuid.UUID, error) {
	UUIDs := make([]uuid.UUID, len(sid))
	for i, id := range sid {
		parsed, err := uuid.Parse(id)
		if err != nil {
			log.Error(err)
			return nil, status.Errorf(codes.Aborted, "%v", err)
		}
		UUIDs[i] = parsed
	}
	return UUIDs, nil
}

// AddList adds the list of mnes to the storage.
func (n *NetworkElementServer) AddList(ctx context.Context, request *mnepb.AddListRequest) (*mnepb.AddListResponse, error) {
	if err := n.protoValidator.Validate(request); err != nil {
		return nil, status.Errorf(codes.Aborted, "%v", err)
	}

	labels := prometheus.Labels{"service": "mne", "rpc": "set"}
	start := metrics.StartHook(labels, grpcRequestsTotal)
	defer metrics.FinishHook(labels, start, grpcRequestDurationSecondsTotal, grpcRequestDurationSeconds)

	pndID, err := uuid.Parse(request.Pid)
	if err != nil {
		log.Error(err)
		return nil, status.Errorf(codes.Aborted, "%v", err)
	}

	networkElementIDs := make([]uuid.UUID, 0, len(request.Mne))
	for _, r := range request.Mne {
		pluginId, err := uuid.Parse(r.GetPluginId())
		if err != nil {
			log.Error(err)
			return nil, status.Errorf(codes.Aborted, "%v", err)
		}

		gNMISubscriptionPaths := n.get2dSlicePathsFrom1dPaths(r.GnmiSubscribePaths)

		mneID, err := uuid.Parse(r.MneId)
		if err != nil {
			log.Errorf("Error parsing UUID: %v. For: %v, ", r.MneId, r.MneName)
		}

		mneID, err = n.addMne(ctx, r.MneName, r.TransportOption, nil, pluginId, pndID, gNMISubscriptionPaths, mneID)
		if err != nil {
			log.Error(err)
			return nil, status.Errorf(codes.Aborted, "%v", err)
		}
		networkElementIDs = append(networkElementIDs, mneID)
	}

	r := make([]*mnepb.SetResponse, len(networkElementIDs))
	for i, mneID := range networkElementIDs {
		r[i] = &mnepb.SetResponse{Id: mneID.String()}
	}

	return &mnepb.AddListResponse{
		Timestamp: time.Now().UnixNano(),
		Responses: r,
	}, nil
}

func (n *NetworkElementServer) addMne(ctx context.Context,
	name string,
	opt *tpb.TransportOption,
	requestPluginFunc func(uuid.UUID) (plugin.Plugin, error),
	pluginID uuid.UUID, pndID uuid.UUID,
	gnmiSubscriptionPaths [][]string,
	optionalNetworkElementID ...uuid.UUID) (mneId uuid.UUID, err error) {
	var plugin plugin.Plugin
	var mne networkelement.NetworkElement

	defer func() {
		if err != nil && plugin != nil {
			// clean up plugin
			if pErr := plugin.Remove(); pErr != nil {
				err = multierror.Append(err, pErr)
			}
		}
	}()

	networkElementID := uuid.Nil
	if len(optionalNetworkElementID) > 0 {
		networkElementID = optionalNetworkElementID[0]
	}

	if requestPluginFunc == nil {
		requestPluginFunc = n.pluginService.RequestPlugin
	}

	plugin, err = requestPluginFunc(pluginID)
	if err != nil {
		return uuid.Nil, err
	}

	mne, err = nucleus.NewNetworkElement(name, networkElementID, opt, pndID, plugin, gnmiSubscriptionPaths, conflict.Metadata{ResourceVersion: 0})
	if err != nil {
		return uuid.Nil, err
	}

	if mne.IsTransportValid() {
		err = n.initialNetworkElementRootPathRequest(ctx, mne, plugin)
		if err != nil {
			return uuid.Nil, err
		}

		err = n.mneService.Add(mne)
		if err != nil {
			return uuid.Nil, err
		}

		err = n.pluginService.Add(plugin)
		if err != nil {
			if err := n.mneService.Delete(mne); err != nil {
				return uuid.Nil, err
			}
			return uuid.Nil, err
		}

		if mne.GetGnmiSubscriptionPaths() != nil || config.GetGnmiSubscriptionPaths() != nil {
			n.networkElementWatchter.SubscribeToNetworkElement(mne, nil)
		}
	} else {
		err = fmt.Errorf("invalid transport data provided")
		return uuid.Nil, err
	}

	return mne.ID(), nil
}

func (n *NetworkElementServer) initialNetworkElementRootPathRequest(ctx context.Context, mne networkelement.NetworkElement, plugin plugin.Plugin) error {
	resp, err := n.getPath(ctx, mne, "/")
	if err != nil {
		return err
	}

	err = mne.ProcessResponse(resp)
	if err != nil {
		return err
	}
	return nil
}

// SetChangeList sets a list of changes.
func (n *NetworkElementServer) SetChangeList(ctx context.Context, request *mnepb.SetChangeListRequest) (*mnepb.SetChangeListResponse, error) {
	labels := prometheus.Labels{"service": "mne", "rpc": "set"}
	start := metrics.StartHook(labels, grpcRequestsTotal)
	defer metrics.FinishHook(labels, start, grpcRequestDurationSecondsTotal, grpcRequestDurationSeconds)

	if err := n.protoValidator.Validate(request); err != nil {
		return nil, status.Errorf(codes.Aborted, "%v", err)
	}

	responses := make([]*mnepb.SetResponse, len(request.Change))

	for i, r := range request.Change {
		cuid, err := uuid.Parse(r.Cuid)
		if err != nil {
			log.Error(err)
			return nil, status.Errorf(codes.Aborted, "%v", err)
		}
		switch r.Op {
		case mnepb.Operation_OPERATION_COMMIT:
			if err := n.Commit(cuid); err != nil {
				log.Error(err)
				return nil, status.Errorf(codes.Aborted, "%v", err)
			}
		case mnepb.Operation_OPERATION_CONFIRM:
			if err := n.Confirm(cuid); err != nil {
				log.Error(err)
				return nil, status.Errorf(codes.Aborted, "%v", err)
			}
		default:
			return nil, &customerrs.InvalidParametersError{
				Param: r.Op,
			}
		}

		responses[i] = &mnepb.SetResponse{
			Id: cuid.String(),
		}
	}
	return &mnepb.SetChangeListResponse{
		Timestamp: time.Now().UnixNano(),
		Responses: responses,
	}, nil
}

// Commit calls commit on the pending change with ID.
func (n *NetworkElementServer) Commit(u uuid.UUID) error {
	ch, err := n.changeStore.GetChange(u)
	if err != nil {
		return err
	}

	if err := ch.Commit(); err != nil {
		return err
	}

	// Set all the changes within the network elements model
	networkElement, err := n.mneService.Get(store.Query{ID: ch.AssociatedDeviceID()})
	if err != nil {
		return err
	}
	diff, err := networkElement.GetPlugin().Diff(ch.PreviousState(), ch.IntendedState())
	if err != nil {
		return err
	}
	for _, update := range diff.GetUpdate() {
		if err := networkElement.GetPlugin().SetNode(update.GetPath(), update.GetVal()); err != nil {
			return err
		}
	}
	for _, deletePath := range diff.GetDelete() {
		if err := networkElement.GetPlugin().DeleteNode(deletePath); err != nil {
			return err
		}
	}

	// update the network element
	return n.mneService.Update(networkElement)
}

// Confirm calls confirm on pending the pending change with ID.
func (n *NetworkElementServer) Confirm(u uuid.UUID) error {
	ch, err := n.changeStore.GetChange(u)
	if err != nil {
		return err
	}
	return ch.Confirm()
}

// SetPathList sets a list of paths.
func (n *NetworkElementServer) SetPathList(ctx context.Context, request *mnepb.SetPathListRequest) (*mnepb.SetPathListResponse, error) {
	labels := prometheus.Labels{"service": "mne", "rpc": "set"}
	start := metrics.StartHook(labels, grpcRequestsTotal)
	defer metrics.FinishHook(labels, start, grpcRequestDurationSecondsTotal, grpcRequestDurationSeconds)

	if err := n.protoValidator.Validate(request); err != nil {
		return nil, status.Errorf(codes.Aborted, "%v", err)
	}

	responses := make([]*mnepb.SetResponse, len(request.ChangeRequest))

	for i, r := range request.ChangeRequest {
		var cid uuid.UUID
		var err error

		mneID, err := uuid.Parse(r.Mneid)
		if err != nil {
			log.Error(err)
			return nil, status.Errorf(codes.Aborted, "%v", err)
		}

		// NOTE: Could be useful to split method into two, one for
		// update/replace and on for delete.
		cid, err = n.ChangeMNE(mneID, r.ApiOp, r.Path, r.Value)
		if err != nil {
			log.Error(err)
			return nil, status.Errorf(codes.Aborted, "%v", err)
		}

		responses[i] = &mnepb.SetResponse{
			Id: cid.String(),
		}
	}
	return &mnepb.SetPathListResponse{
		Timestamp: time.Now().UnixNano(),
		Responses: responses,
	}, nil
}

// ChangeMNE creates a change from the provided Operation, path and value.
// The Change is Pending and times out after the specified timeout period.
func (n *NetworkElementServer) ChangeMNE(duid uuid.UUID, operation mnepb.ApiOperation, path *gnmi.Path, value *gnmi.TypedValue) (uuid.UUID, error) {
	mne, err := n.mneService.Get(store.Query{
		ID: duid,
	})
	if err != nil {
		return uuid.Nil, err
	}

	plugin := mne.GetPlugin()

	validatedChangeModel, err := plugin.ValidateChange(operation, path, value)
	if err != nil {
		return uuid.Nil, err
	}

	filteredMarshaledModel, err := plugin.PruneConfigFalse(validatedChangeModel)
	if err != nil {
		return uuid.Nil, err
	}

	currentModel, err := mne.GetModelAsFilteredCopy()
	if err != nil {
		return uuid.Nil, err
	}

	diff, err := plugin.Diff(currentModel, filteredMarshaledModel)
	if err != nil {
		return uuid.Nil, err
	}

	if util.IsGNMINotificationEmpty(diff) {
		return uuid.Nil, customerrs.NoNewChangesError{Original: string(currentModel), Modified: string(filteredMarshaledModel)}
	}

	callback := func(original, modified []byte) error {
		ctx := context.WithValue(context.Background(), types.CtxKeyOperation, operation) // nolint
		payload := change.Payload{Original: original, Modified: modified, Diff: diff}
		return mne.Transport().Set(ctx, payload)
	}

	ch := nucleus.NewChange(duid, currentModel, filteredMarshaledModel, diff, callback)

	if err := n.changeStore.Add(ch); err != nil {
		return uuid.Nil, err
	}

	return ch.ID(), nil
}

// Delete deletes a mne.
func (n *NetworkElementServer) Delete(ctx context.Context, request *mnepb.DeleteRequest) (*mnepb.DeleteResponse, error) {
	if err := n.protoValidator.Validate(request); err != nil {
		return nil, status.Errorf(codes.Aborted, "%v", err)
	}

	mneID, err := uuid.Parse(request.Mneid)
	if err != nil {
		log.Error(err)
		return nil, status.Errorf(codes.Aborted, "%v", err)
	}

	if err := n.deleteMne(mneID); err != nil {
		log.Error(err)
		return nil, status.Errorf(codes.Aborted, "%v", err)
	}
	return &mnepb.DeleteResponse{
		Timestamp: time.Now().UnixNano(),
	}, nil
}

func (n *NetworkElementServer) deleteMne(id uuid.UUID) error {
	mne, err := n.mneService.Get(store.Query{
		ID:   id,
		Name: id.String(),
	})

	if err != nil {
		return err
	}

	// Note: cSBI not supported currently, so this is commented fow now.
	// Might be needed or removed in the future.
	//
	// switch mne.(type) {
	// case *CsbiNetworkElement:
	// 	return pnd.handleCsbiDeletion(mne)
	// default:
	// 	return pnd.networkElementService.Delete(mne)
	// }

	return n.mneService.Delete(mne)
}

// SubscribePath subscribes to specifc paths of an mne.
func (n *NetworkElementServer) SubscribePath(request *mnepb.SubscribePathRequest, stream mnepb.NetworkElementService_SubscribePathServer) error {
	networkElement, err := n.getMne(request.Mneid)
	if err != nil {
		log.Error(err)
		return status.Errorf(codes.Aborted, "%v", err)
	}

	if err := n.subscribePath(networkElement, request.Sublist); err != nil {
		return err
	}

	return nil
}

func (n *NetworkElementServer) subscribePath(networkElement networkelement.NetworkElement, subList *mnepb.SubscriptionList) error {
	mode, err := n.mapModeToAristaFork(subList.GetMode())
	if err != nil {
		return err
	}

	for _, sub := range subList.Subscription {
		streamMode, err := n.mapStreamModeToAristaFork(sub.GetStreamMode())
		if err != nil {
			return err
		}

		opts := &aGNMI.SubscribeOptions{
			Mode:           mode,
			StreamMode:     streamMode,
			Paths:          [][]string{n.splitStringPath(sub.GetPath())},
			SampleInterval: sub.SampleInterval,
		}

		ctx := context.Background()
		ctx = context.WithValue(ctx, types.CtxKeyOpts, opts)

		if err = networkElement.Transport().Subscribe(ctx); err != nil {
			return err
		}
	}

	return nil
}

func (n *NetworkElementServer) splitStringPath(s string) []string {
	return strings.Split(s, "/")
}

func (n *NetworkElementServer) mapStreamModeToAristaFork(mode mnepb.StreamMode) (string, error) {
	switch mode {
	case mnepb.StreamMode_STREAM_MODE_TARGET_DEFINED:
		return "target_defined", nil
	case mnepb.StreamMode_STREAM_MODE_ON_CHANGE:
		return "on_change", nil
	case mnepb.StreamMode_STREAM_MODE_SAMPLE:
		return "sample", nil
	default:
		return "", fmt.Errorf("StreamMode of type: %T is not supported", mode)
	}
}

func (n *NetworkElementServer) mapModeToAristaFork(mode mnepb.SubscriptionMode) (string, error) {
	switch mode {
	case mnepb.SubscriptionMode_SUBSCRIPTION_MODE_STREAM:
		return "stream", nil
	case mnepb.SubscriptionMode_SUBSCRIPTION_MODE_ONCE:
		return "once", nil
	case mnepb.SubscriptionMode_SUBSCRIPTION_MODE_POLL: