pkg/control/registry.go

/*
==================================================================================
  Copyright (c) 2019 AT&T Intellectual Property.
  Copyright (c) 2019 Nokia

   Licensed under the Apache License, Version 2.0 (the "License");
   you may not use this file except in compliance with the License.
   You may obtain a copy of the License at

       http://www.apache.org/licenses/LICENSE-2.0

   Unless required by applicable law or agreed to in writing, software
   distributed under the License is distributed on an "AS IS" BASIS,
   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
   See the License for the specific language governing permissions and
   limitations under the License.
==================================================================================
*/

package control

import (
	"encoding/json"
	"fmt"
	"strings"
	"sync"
	"time"

	"gerrit.o-ran-sc.org/r/ric-plt/e2ap/pkg/e2ap"
	"gerrit.o-ran-sc.org/r/ric-plt/xapp-frame/pkg/models"
	"gerrit.o-ran-sc.org/r/ric-plt/xapp-frame/pkg/xapp"
)

//-----------------------------------------------------------------------------
//
//-----------------------------------------------------------------------------

type RESTSubscription struct {
	xAppRmrEndPoint  string
	Meid             string
	InstanceIds      []uint32
	xAppIdToE2Id     map[int64]int64
	SubReqOngoing    bool
	SubDelReqOngoing bool
	lastReqMd5sum    string
}

func (r *RESTSubscription) AddE2InstanceId(instanceId uint32) {

	for _, v := range r.InstanceIds {
		if v == instanceId {
			return
		}

	}

	r.InstanceIds = append(r.InstanceIds, instanceId)
}

func (r *RESTSubscription) AddMd5Sum(md5sum string) {
	if md5sum != "" {
		r.lastReqMd5sum = md5sum
	} else {
		xapp.Logger.Error("EMPTY md5sum attempted to be add to subscrition")
	}
}

func (r *RESTSubscription) DeleteE2InstanceId(instanceId uint32) {
	r.InstanceIds = r.InstanceIds[1:]
}

func (r *RESTSubscription) AddXappIdToE2Id(xAppEventInstanceID int64, e2EventInstanceID int64) {
	r.xAppIdToE2Id[xAppEventInstanceID] = e2EventInstanceID
}

func (r *RESTSubscription) GetE2IdFromXappIdToE2Id(xAppEventInstanceID int64) int64 {
	return r.xAppIdToE2Id[xAppEventInstanceID]
}

func (r *RESTSubscription) DeleteXappIdToE2Id(xAppEventInstanceID int64) {
	delete(r.xAppIdToE2Id, xAppEventInstanceID)
}

func (r *RESTSubscription) SetProcessed(err error) {
	r.SubReqOngoing = false
	if err != nil {
		r.lastReqMd5sum = ""
	}
}

type Registry struct {
	mutex             sync.Mutex
	register          map[uint32]*Subscription
	subIds            []uint32
	rtmgrClient       *RtmgrClient
	restSubscriptions map[string]*RESTSubscription
}

func (r *Registry) Initialize() {
	r.register = make(map[uint32]*Subscription)
	r.restSubscriptions = make(map[string]*RESTSubscription)

	var i uint32
	for i = 1; i < 65535; i++ {
		r.subIds = append(r.subIds, i)
	}
}

func (r *Registry) GetAllRestSubscriptions() []byte {
	r.mutex.Lock()
	defer r.mutex.Unlock()
	restSubscriptionsJson, err := json.Marshal(r.restSubscriptions)
	if err != nil {
		xapp.Logger.Error("GetAllRestSubscriptions(): %v", err)
	}
	return restSubscriptionsJson
}

func (r *Registry) CreateRESTSubscription(restSubId *string, xAppRmrEndPoint *string, maid *string) *RESTSubscription {
	r.mutex.Lock()
	defer r.mutex.Unlock()
	newRestSubscription := RESTSubscription{}
	newRestSubscription.xAppRmrEndPoint = *xAppRmrEndPoint
	newRestSubscription.Meid = *maid
	newRestSubscription.SubReqOngoing = true
	newRestSubscription.SubDelReqOngoing = false
	r.restSubscriptions[*restSubId] = &newRestSubscription
	newRestSubscription.xAppIdToE2Id = make(map[int64]int64)
	xapp.Logger.Debug("Registry: Created REST subscription successfully. restSubId=%v, subscriptionCount=%v, e2apSubscriptionCount=%v", *restSubId, len(r.restSubscriptions), len(r.register))
	return &newRestSubscription
}

func (r *Registry) DeleteRESTSubscription(restSubId *string) {
	r.mutex.Lock()
	defer r.mutex.Unlock()
	delete(r.restSubscriptions, *restSubId)
	xapp.Logger.Debug("Registry: Deleted REST subscription successfully. restSubId=%v, subscriptionCount=%v", *restSubId, len(r.restSubscriptions))
}

func (r *Registry) GetRESTSubscription(restSubId string, IsDelReqOngoing bool) (*RESTSubscription, error) {
	r.mutex.Lock()
	defer r.mutex.Unlock()
	if restSubscription, ok := r.restSubscriptions[restSubId]; ok {
		// Subscription deletion is not allowed if prosessing subscription request in not ready
		if restSubscription.SubDelReqOngoing == false && restSubscription.SubReqOngoing == false {
			if IsDelReqOngoing == true {
				restSubscription.SubDelReqOngoing = true
			}
			r.restSubscriptions[restSubId] = restSubscription
			return restSubscription, nil
		} else {
			return restSubscription, fmt.Errorf("Registry: REST request is still ongoing for the endpoint=%v, restSubId=%v, SubDelReqOngoing=%v, SubReqOngoing=%v", restSubscription, restSubId, restSubscription.SubDelReqOngoing, restSubscription.SubReqOngoing)
		}
	}
	return nil, fmt.Errorf("Registry: No valid subscription found with restSubId=%v", restSubId)
}

func (r *Registry) QueryHandler() (models.SubscriptionList, error) {
	r.mutex.Lock()
	defer r.mutex.Unlock()

	resp := models.SubscriptionList{}
	for _, subs := range r.register {
		subs.mutex.Lock()
		resp = append(resp, &models.SubscriptionData{SubscriptionID: int64(subs.ReqId.InstanceId), Meid: subs.Meid.RanName, ClientEndpoint: subs.EpList.StringList()})
		subs.mutex.Unlock()
	}
	return resp, nil
}

func (r *Registry) allocateSubs(trans *TransactionXapp, subReqMsg *e2ap.E2APSubscriptionRequest, resetTestFlag bool, rmrRoutecreated bool) (*Subscription, error) {
	if len(r.subIds) > 0 {
		subId := r.subIds[0]
		r.subIds = r.subIds[1:]
		if _, ok := r.register[subId]; ok == true {
			r.subIds = append(r.subIds, subId)
			return nil, fmt.Errorf("Registry: Failed to reserve subscription exists")
		}
		subs := &Subscription{
			registry:         r,
			Meid:             trans.Meid,
			RMRRouteCreated:  rmrRoutecreated,
			SubReqMsg:        subReqMsg,
			OngoingReqCount:  0,
			OngoingDelCount:  0,
			valid:            true,
			PolicyUpdate:     false,
			RetryFromXapp:    false,
			SubRespRcvd:      false,
			DeleteFromDb:     false,
			NoRespToXapp:     false,
			DoNotWaitSubResp: false,
		}
		subs.ReqId.Id = subReqMsg.RequestId.Id
		subs.ReqId.InstanceId = subId
		r.SetResetTestFlag(resetTestFlag, subs)

		if subs.EpList.AddEndpoint(trans.GetEndpoint()) == false {
			r.subIds = append(r.subIds, subs.ReqId.InstanceId)
			return nil, fmt.Errorf("Registry: Endpoint existing already in subscription")
		}
		return subs, nil
	}
	return nil, fmt.Errorf("Registry: Failed to reserve subscription no free ids")
}

func (r *Registry) findExistingSubs(trans *TransactionXapp, subReqMsg *e2ap.E2APSubscriptionRequest) (*Subscription, bool) {

	for _, subs := range r.register {
		if subs.IsMergeable(trans, subReqMsg) {

			//
			// check if there has been race conditions
			//
			subs.mutex.Lock()
			//subs has been set to invalid
			if subs.valid == false {
				subs.mutex.Unlock()
				continue
			}
			// If size is zero, entry is to be deleted
			if subs.EpList.Size() == 0 {
				subs.mutex.Unlock()
				continue
			}
			// Try to add to endpointlist. Adding fails if endpoint is already in the list
			if subs.EpList.AddEndpoint(trans.GetEndpoint()) == false {
				subs.mutex.Unlock()
				xapp.Logger.Debug("Registry: Subs with requesting endpoint found. %s for %s", subs.String(), trans.String())
				return subs, true
			}
			subs.mutex.Unlock()

			xapp.Logger.Debug("Registry: Mergeable subs found. %s for %s", subs.String(), trans.String())
			return subs, false
		}
	}
	return nil, false
}

func (r *Registry) AssignToSubscription(trans *TransactionXapp, subReqMsg *e2ap.E2APSubscriptionRequest, resetTestFlag bool, c *Control, createRMRRoute bool) (*Subscription, ErrorInfo, error) {
	var err error
	var newAlloc bool
	errorInfo := ErrorInfo{}
	r.mutex.Lock()
	defer r.mutex.Unlock()

	//
	// Check validity of subscription action types
	//
	actionType, err := r.CheckActionTypes(subReqMsg)
	if err != nil {
		xapp.Logger.Debug("CREATE %s", err)
		err = fmt.Errorf("E2 content validation failed")
		return nil, errorInfo, err
	}

	//
	// Find possible existing Policy subscription
	//
	if actionType == e2ap.E2AP_ActionTypePolicy {
		if subs, ok := r.register[trans.GetSubId()]; ok {
			xapp.Logger.Debug("CREATE %s. Existing subscription for Policy found.", subs.String())
			// Update message data to subscription
			subs.SubReqMsg = subReqMsg
			subs.PolicyUpdate = true
			subs.SetCachedResponse(nil, true)
			r.SetResetTestFlag(resetTestFlag, subs)
			return subs, errorInfo, nil
		}
	}

	subs, endPointFound := r.findExistingSubs(trans, subReqMsg)
	if subs == nil {
		if subs, err = r.allocateSubs(trans, subReqMsg, resetTestFlag, createRMRRoute); err != nil {
			xapp.Logger.Error("%s", err.Error())
			err = fmt.Errorf("subscription not allocated")
			return nil, errorInfo, err
		}
		newAlloc = true
	} else if endPointFound == true {
		// Requesting endpoint is already present in existing subscription. This can happen if xApp is restarted.
		subs.RetryFromXapp = true
		xapp.Logger.Debug("CREATE subReqMsg.InstanceId=%v. Same subscription %s already exists.", subReqMsg.InstanceId, subs.String())
		c.UpdateCounter(cDuplicateE2SubReq)
		return subs, errorInfo, nil
	}

	//
	// Add to subscription
	//
	subs.mutex.Lock()
	defer subs.mutex.Unlock()

	epamount := subs.EpList.Size()
	xapp.Logger.Debug("AssignToSubscription subs.EpList.Size()=%v", subs.EpList.Size())

	r.mutex.Unlock()
	//
	// Subscription route updates
	//
	if createRMRRoute == true {
		if epamount == 1 {
			errorInfo, err = r.RouteCreate(subs, c)
		} else {
			errorInfo, err = r.RouteCreateUpdate(subs, c)
		}
	} else {
		xapp.Logger.Debug("RMR route not created: createRMRRoute=%v", createRMRRoute)
	}
	r.mutex.Lock()

	if err != nil {
		if newAlloc {
			r.subIds = append(r.subIds, subs.ReqId.InstanceId)
		}
		// Delete already added endpoint for the request
		subs.EpList.DelEndpoint(trans.GetEndpoint())
		return nil, errorInfo, err
	}

	if newAlloc {
		r.register[subs.ReqId.InstanceId] = subs
	}
	xapp.Logger.Debug("CREATE %s", subs.String())
	xapp.Logger.Debug("Registry: substable=%v", r.register)
	return subs, errorInfo, nil
}

func (r *Registry) RouteCreate(subs *Subscription, c *Control) (ErrorInfo, error) {
	errorInfo := ErrorInfo{}
	subRouteAction := SubRouteInfo{subs.EpList, uint16(subs.ReqId.InstanceId)}
	err := r.rtmgrClient.SubscriptionRequestCreate(subRouteAction)
	if err != nil {
		if strings.Contains(err.Error(), "status 400") {
			errorInfo.TimeoutType = models.SubscriptionInstanceTimeoutTypeRTMGRTimeout
		} else {
			errorInfo.ErrorSource = models.SubscriptionInstanceErrorSourceRTMGR
		}
		errorInfo.ErrorCause = err.Error()
		c.UpdateCounter(cRouteCreateFail)
		xapp.Logger.Error("%s", err.Error())
		err = fmt.Errorf("RTMGR route create failure")
	}
	return errorInfo, err
}

func (r *Registry) RouteCreateUpdate(subs *Subscription, c *Control) (ErrorInfo, error) {
	errorInfo := ErrorInfo{}
	subRouteAction := SubRouteInfo{subs.EpList, uint16(subs.ReqId.InstanceId)}
	err := r.rtmgrClient.SubscriptionRequestUpdate(subRouteAction)
	if err != nil {
		if strings.Contains(err.Error(), "status 400") {
			errorInfo.TimeoutType = models.SubscriptionInstanceTimeoutTypeRTMGRTimeout
		} else {
			errorInfo.ErrorSource = models.SubscriptionInstanceErrorSourceRTMGR
		}
		errorInfo.ErrorCause = err.Error()
		c.UpdateCounter(cRouteCreateUpdateFail)
		xapp.Logger.Error("%s", err.Error())
		err = fmt.Errorf("RTMGR route update failure")
		return errorInfo, err
	}
	c.UpdateCounter(cMergedSubscriptions)
	return errorInfo, err
}

func (r *Registry) CheckActionTypes(subReqMsg *e2ap.E2APSubscriptionRequest) (uint64, error) {
	var reportFound bool = false
	var policyFound bool = false
	var insertFound bool = false

	for _, acts := range subReqMsg.ActionSetups {
		if acts.ActionType == e2ap.E2AP_ActionTypeReport {
			reportFound = true
		}
		if acts.ActionType == e2ap.E2AP_ActionTypePolicy {
			policyFound = true
		}
		if acts.ActionType == e2ap.E2AP_ActionTypeInsert {
			insertFound = true
		}
	}
	if reportFound == true && policyFound == true || reportFound == true && insertFound == true || policyFound == true && insertFound == true {
		return e2ap.E2AP_ActionTypeInvalid, fmt.Errorf("Different action types (Report, Policy or Insert) in same RICactions-ToBeSetup-List")
	}
	if reportFound == true {
		return e2ap.E2AP_ActionTypeReport, nil
	}
	if policyFound == true {
		return e2ap.E2AP_ActionTypePolicy, nil
	}
	if insertFound == true {
		return e2ap.E2AP_ActionTypeInsert, nil
	}
	return e2ap.E2AP_ActionTypeInvalid, fmt.Errorf("Invalid action type in RICactions-ToBeSetup-List")
}

// TODO: Works with concurrent calls, but check if can be improved
func (r *Registry) RemoveFromSubscription(subs *Subscription, trans *TransactionXapp, waitRouteClean time.Duration, c *Control) error {

	r.mutex.Lock()
	defer r.mutex.Unlock()
	subs.mutex.Lock()
	defer subs.mutex.Unlock()

	delStatus := subs.EpList.DelEndpoint(trans.GetEndpoint())
	epamount := subs.EpList.Size()
	subId := subs.ReqId.InstanceId
	if delStatus == false {
		return nil
	}

	go func() {
		if waitRouteClean > 0 {
			xapp.Logger.Debug("Pending %v in order to wait route cleanup", waitRouteClean)
			time.Sleep(waitRouteClean)
		}

		subs.mutex.Lock()
		defer subs.mutex.Unlock()
		xapp.Logger.Debug("CLEAN %s", subs.String())

		if epamount == 0 {
			//
			// Subscription route delete
			//
			if subs.RMRRouteCreated == true {
				r.RouteDelete(subs, trans, c)
			}

			//
			// Subscription release
			//
			r.mutex.Lock()
			defer r.mutex.Unlock()

			if _, ok := r.register[subId]; ok {
				xapp.Logger.Debug("RELEASE %s", subs.String())
				delete(r.register, subId)
				xapp.Logger.Debug("Registry: substable=%v", r.register)
			}
			r.subIds = append(r.subIds, subId)
		} else if subs.EpList.Size() > 0 {
			//
			// Subscription route update
			//
			if subs.RMRRouteCreated == true {
				r.RouteDeleteUpdate(subs, c)
			}
		}
	}()

	return nil
}

func (r *Registry) RouteDelete(subs *Subscription, trans *TransactionXapp, c *Control) {
	tmpList := xapp.RmrEndpointList{}
	tmpList.AddEndpoint(trans.GetEndpoint())
	subRouteAction := SubRouteInfo{tmpList, uint16(subs.ReqId.InstanceId)}
	if err := r.rtmgrClient.SubscriptionRequestDelete(subRouteAction); err != nil {
		c.UpdateCounter(cRouteDeleteFail)
	}
}

func (r *Registry) RouteDeleteUpdate(subs *Subscription, c *Control) {
	subRouteAction := SubRouteInfo{subs.EpList, uint16(subs.ReqId.InstanceId)}
	if err := r.rtmgrClient.SubscriptionRequestUpdate(subRouteAction); err != nil {
		c.UpdateCounter(cRouteDeleteUpdateFail)
	}
}

func (r *Registry) UpdateSubscriptionToDb(subs *Subscription, c *Control) {
	r.mutex.Lock()
	defer r.mutex.Unlock()
	subs.mutex.Lock()
	defer subs.mutex.Unlock()

	epamount := subs.EpList.Size()
	if epamount == 0 {
		if _, ok := r.register[subs.ReqId.InstanceId]; ok {
			// Not merged subscription is being deleted
			c.RemoveSubscriptionFromDb(subs)

		}
	} else if subs.EpList.Size() > 0 {
		// Endpoint of merged subscription is being deleted
		c.WriteSubscriptionToDb(subs)
		c.UpdateCounter(cUnmergedSubscriptions)
	}
}

func (r *Registry) GetSubscription(subId uint32) *Subscription {
	r.mutex.Lock()
	defer r.mutex.Unlock()
	if _, ok := r.register[subId]; ok {
		return r.register[subId]
	}
	return nil
}

func (r *Registry) GetSubscriptionFirstMatch(subIds []uint32) (*Subscription, error) {
	r.mutex.Lock()
	defer r.mutex.Unlock()
	for _, subId := range subIds {
		if _, ok := r.register[subId]; ok {
			return r.register[subId], nil
		}
	}
	return nil, fmt.Errorf("No valid subscription found with subIds %v", subIds)
}

func (r *Registry) SetResetTestFlag(resetTestFlag bool, subs *Subscription) {
	if resetTestFlag == true {
		// This is used in submgr restart unit tests
		xapp.Logger.Debug("resetTestFlag == true")
		subs.DoNotWaitSubResp = true
	} else {
		xapp.Logger.Debug("resetTestFlag == false")
	}
}

func (r *Registry) DeleteAllE2Subscriptions(ranName string, c *Control) {

	xapp.Logger.Debug("Registry: DeleteAllE2Subscriptions()")
	for subId, subs := range r.register {
		if subs.Meid.RanName == ranName {
			if subs.OngoingReqCount != 0 || subs.OngoingDelCount != 0 {
				// Subscription creation or deletion processes need to be processed gracefully till the end.
				// Subscription is deleted at end of the process in both cases.
				xapp.Logger.Debug("Registry: E2 subscription under prosessing ongoing cannot delete it yet. subId=%v, OngoingReqCount=%v, OngoingDelCount=%v", subId, subs.OngoingReqCount, subs.OngoingDelCount)
				continue
			} else {
				// Delete route
				if subs.RMRRouteCreated == true {
					for _, ep := range subs.EpList.Endpoints {
						tmpList := xapp.RmrEndpointList{}
						tmpList.AddEndpoint(&ep)
						subRouteAction := SubRouteInfo{tmpList, uint16(subs.ReqId.InstanceId)}
						if err := r.rtmgrClient.SubscriptionRequestDelete(subRouteAction); err != nil {
							c.UpdateCounter(cRouteDeleteFail)
						}
					}
				}
				// Delete E2 subscription from registry and db
				xapp.Logger.Debug("Registry: Subscription delete. subId=%v", subId)
				delete(r.register, subId)
				r.subIds = append(r.subIds, subId)
				c.RemoveSubscriptionFromDb(subs)
			}
		}
	}

	// Delete REST subscription from registry and db
	for restSubId, restSubs := range r.restSubscriptions {
		if restSubs.Meid == ranName && restSubs.SubReqOngoing == true || restSubs.SubDelReqOngoing == true {
			// Subscription creation or deletion processes need to be processed gracefully till the end.
			// Subscription is deleted at end of the process in both cases.
			xapp.Logger.Debug("Registry: REST subscription under prosessing ongoing cannot delete it yet. RestSubId=%v, SubReqOngoing=%v, SubDelReqOngoing=%v", restSubId, restSubs.SubReqOngoing, restSubs.SubDelReqOngoing)
			continue
		} else {
			xapp.Logger.Debug("Registry: REST subscription delete. subId=%v", restSubId)
			delete(r.restSubscriptions, restSubId)
			c.RemoveRESTSubscriptionFromDb(restSubId)
		}
	}
}