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"
	"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, error) {
	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.Info("Registry: Created REST subscription successfully. restSubId=%v, subscriptionCount=%v, e2apSubscriptionCount=%v", *restSubId, len(r.restSubscriptions), len(r.register))
	return &newRestSubscription, nil
}

func (r *Registry) DeleteRESTSubscription(restSubId *string) {
	r.mutex.Lock()
	defer r.mutex.Unlock()
	delete(r.restSubscriptions, *restSubId)
	xapp.Logger.Info("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 restSubscription, nil
	}
	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) (*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,
			SubReqMsg:        subReqMsg,
			valid:            true,
			RetryFromXapp:    false,
			SubRespRcvd:      false,
			DeleteFromDb:     false,
			NoRespToXapp:     false,
			DoNotWaitSubResp: false,
		}
		subs.ReqId.Id = subReqMsg.RequestId.Id
		subs.ReqId.InstanceId = subId
		if resetTestFlag == true {
			subs.DoNotWaitSubResp = true
		}

		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) (*Subscription, error) {
	var err error
	var newAlloc bool
	r.mutex.Lock()
	defer r.mutex.Unlock()

	//
	// Check validity of subscription action types
	//
	actionType, err := r.CheckActionTypes(subReqMsg)
	if err != nil {
		xapp.Logger.Info("CREATE %s", err)
		err = fmt.Errorf("E2 content validation failed")
		return nil, 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.SetCachedResponse(nil, true)
			return subs, nil
		}
	}

	subs, endPointFound := r.findExistingSubs(trans, subReqMsg)
	if subs == nil {
		if subs, err = r.allocateSubs(trans, subReqMsg, resetTestFlag); err != nil {
			xapp.Logger.Error("%s", err.Error())
			err = fmt.Errorf("subscription not allocated")
			return nil, 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, nil
	}

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

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

	r.mutex.Unlock()
	//
	// Subscription route updates
	//
	if epamount == 1 {
		err = r.RouteCreate(subs, c)
	} else {
		err = r.RouteCreateUpdate(subs, c)
	}
	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, 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, nil
}

func (r *Registry) RouteCreate(subs *Subscription, c *Control) error {
	subRouteAction := SubRouteInfo{subs.EpList, uint16(subs.ReqId.InstanceId)}
	err := r.rtmgrClient.SubscriptionRequestCreate(subRouteAction)
	if err != nil {
		c.UpdateCounter(cRouteCreateFail)
		xapp.Logger.Error("%s", err.Error())
		err = fmt.Errorf("RTMGR route create failure")
	}
	return err
}

func (r *Registry) RouteCreateUpdate(subs *Subscription, c *Control) error {
	subRouteAction := SubRouteInfo{subs.EpList, uint16(subs.ReqId.InstanceId)}
	err := r.rtmgrClient.SubscriptionRequestUpdate(subRouteAction)
	if err != nil {
		c.UpdateCounter(cRouteCreateUpdateFail)
		xapp.Logger.Error("%s", err.Error())
		err = fmt.Errorf("RTMGR route update failure")
		return err
	}
	c.UpdateCounter(cMergedSubscriptions)
	return 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.Info("CLEAN %s", subs.String())

		if epamount == 0 {
			//
			// Subscription route delete
			//
			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
			//
			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)
}