REST duplicate detection
[ric-plt/submgr.git] / pkg / control / control.go
index 0c2f917..9234a34 100755 (executable)
@@ -67,6 +67,7 @@ var waitRouteCleanup_ms time.Duration
 var e2tMaxSubReqTryCount uint64    // Initial try + retry
 var e2tMaxSubDelReqTryCount uint64 // Initial try + retry
 var readSubsFromDb string
+var restDuplicateCtrl duplicateCtrl
 
 type Control struct {
        *xapp.RMRClient
@@ -127,6 +128,8 @@ func NewControl() *Control {
                return c
        }
 
+       restDuplicateCtrl.Init()
+
        // Read subscriptions from db
        xapp.Logger.Info("Reading subscriptions from db")
        subIds, register, err := c.ReadAllSubscriptionsFromSdl()
@@ -233,9 +236,7 @@ func (c *Control) SubscriptionHandler(params interface{}) (*models.SubscriptionR
        c.CntRecvMsg++
        c.UpdateCounter(cRestSubReqFromXapp)
 
-       restSubId := ksuid.New().String()
        subResp := models.SubscriptionResponse{}
-       subResp.SubscriptionID = &restSubId
        p := params.(*models.SubscriptionParams)
 
        if c.LoggerLevel > 2 {
@@ -254,14 +255,28 @@ func (c *Control) SubscriptionHandler(params interface{}) (*models.SubscriptionR
                c.UpdateCounter(cRestSubFailToXapp)
                return nil, err
        }
+       var restSubId string
+       var restSubscription *RESTSubscription
+       if p.SubscriptionID == "" {
+               restSubId = ksuid.New().String()
+               restSubscription, err = c.registry.CreateRESTSubscription(&restSubId, &xAppRmrEndpoint, p.Meid)
+               if err != nil {
+                       xapp.Logger.Error("%s", err.Error())
+                       c.UpdateCounter(cRestSubFailToXapp)
+                       return nil, err
+               }
 
-       restSubscription, err := c.registry.CreateRESTSubscription(&restSubId, &xAppRmrEndpoint, p.Meid)
-       if err != nil {
-               xapp.Logger.Error("%s", err.Error())
-               c.UpdateCounter(cRestSubFailToXapp)
-               return nil, err
+       } else {
+               restSubId = p.SubscriptionID
+               restSubscription, err = c.registry.GetRESTSubscription(restSubId, false)
+               if err != nil {
+                       xapp.Logger.Error("%s", err.Error())
+                       c.UpdateCounter(cRestSubFailToXapp)
+                       return nil, err
+               }
        }
 
+       subResp.SubscriptionID = &restSubId
        subReqList := e2ap.SubscriptionRequestList{}
        err = c.e2ap.FillSubscriptionReqMsgs(params, &subReqList, restSubscription)
        if err != nil {
@@ -271,82 +286,117 @@ func (c *Control) SubscriptionHandler(params interface{}) (*models.SubscriptionR
                return nil, err
        }
 
-       go c.processSubscriptionRequests(restSubscription, &subReqList, p.ClientEndpoint, p.Meid, &restSubId)
+       err, duplicate, md5sum := restDuplicateCtrl.IsDuplicateToOngoingTransaction(restSubId, params)
+
+       if err != nil {
+               // We were unable to detect whether this request was duplicate or not, proceed
+               xapp.Logger.Info("%s - proceeding with the request", err.Error())
+       } else {
+               if duplicate {
+                       if *p.SubscriptionDetails[0].ActionToBeSetupList[0].ActionType == "report" {
+                               xapp.Logger.Info("Retransmission blocker dropped for report typer of request")
+                               c.UpdateCounter(cRestSubRespToXapp)
+                               return &subResp, nil
+                       }
+               }
+               restSubscription.Md5sumOngoing = md5sum
+       }
+
+       go c.processSubscriptionRequests(restSubscription, &subReqList, p.ClientEndpoint, p.Meid, &restSubId, xAppRmrEndpoint)
 
        c.UpdateCounter(cRestSubRespToXapp)
        return &subResp, nil
 }
 
+func (c *Control) sendUnsuccesfullResponseNotification(restSubId *string, restSubscription *RESTSubscription, xAppEventInstanceID int64, err error,
+       clientEndpoint *models.SubscriptionParamsClientEndpoint, trans *TransactionXapp) {
+
+       // Send notification to xApp that prosessing of a Subscription Request has failed.
+       e2EventInstanceID := (int64)(0)
+       errorCause := err.Error()
+       resp := &models.SubscriptionResponse{
+               SubscriptionID: restSubId,
+               SubscriptionInstances: []*models.SubscriptionInstance{
+                       &models.SubscriptionInstance{E2EventInstanceID: &e2EventInstanceID,
+                               ErrorCause:          &errorCause,
+                               XappEventInstanceID: &xAppEventInstanceID},
+               },
+       }
+       // Mark REST subscription request processed.
+       restSubscription.SetProcessed()
+       if trans != nil {
+               xapp.Logger.Info("Sending unsuccessful REST notification (cause %s) to endpoint=%v:%v, XappEventInstanceID=%v, E2EventInstanceID=%v, %s",
+                       errorCause, clientEndpoint.Host, *clientEndpoint.HTTPPort, xAppEventInstanceID, e2EventInstanceID, idstring(nil, trans))
+       } else {
+               xapp.Logger.Info("Sending unsuccessful REST notification (cause %s) to endpoint=%v:%v, XappEventInstanceID=%v, E2EventInstanceID=%v",
+                       errorCause, clientEndpoint.Host, *clientEndpoint.HTTPPort, xAppEventInstanceID, e2EventInstanceID)
+       }
+       xapp.Subscription.Notify(resp, *clientEndpoint)
+       c.UpdateCounter(cRestSubFailNotifToXapp)
+}
+
+func (c *Control) sendSuccesfullResponseNotification(restSubId *string, restSubscription *RESTSubscription, xAppEventInstanceID int64, e2EventInstanceID int64,
+       clientEndpoint *models.SubscriptionParamsClientEndpoint, trans *TransactionXapp) {
+
+       // Store successfully processed InstanceId for deletion
+       restSubscription.AddE2InstanceId((uint32)(e2EventInstanceID))
+       restSubscription.AddXappIdToE2Id(xAppEventInstanceID, e2EventInstanceID)
+
+       // Send notification to xApp that a Subscription Request has been processed.
+       resp := &models.SubscriptionResponse{
+               SubscriptionID: restSubId,
+               SubscriptionInstances: []*models.SubscriptionInstance{
+                       &models.SubscriptionInstance{E2EventInstanceID: &e2EventInstanceID,
+                               ErrorCause:          nil,
+                               XappEventInstanceID: &xAppEventInstanceID},
+               },
+       }
+       // Mark REST subscription request processesd.
+       restSubscription.SetProcessed()
+       xapp.Logger.Info("Sending successful REST notification to endpoint=%v:%v, XappEventInstanceID=%v, E2EventInstanceID=%v, %s",
+               clientEndpoint.Host, *clientEndpoint.HTTPPort, xAppEventInstanceID, e2EventInstanceID, idstring(nil, trans))
+       xapp.Subscription.Notify(resp, *clientEndpoint)
+       c.UpdateCounter(cRestSubNotifToXapp)
+}
+
 //-------------------------------------------------------------------
 //
 //-------------------------------------------------------------------
 
 func (c *Control) processSubscriptionRequests(restSubscription *RESTSubscription, subReqList *e2ap.SubscriptionRequestList,
-       clientEndpoint *models.SubscriptionParamsClientEndpoint, meid *string, restSubId *string) {
+       clientEndpoint *models.SubscriptionParamsClientEndpoint, meid *string, restSubId *string, xAppRmrEndpoint string) {
 
        xapp.Logger.Info("Subscription Request count=%v ", len(subReqList.E2APSubscriptionRequests))
 
-       _, xAppRmrEndpoint, err := ConstructEndpointAddresses(*clientEndpoint)
-       if err != nil {
-               xapp.Logger.Error("%s", err.Error())
-               return
-       }
+       var xAppEventInstanceID int64
+       var e2EventInstanceID int64
+
+       defer restDuplicateCtrl.TransactionComplete(restSubscription.Md5sumOngoing)
 
-       var requestorID int64
-       var instanceId int64
        for index := 0; index < len(subReqList.E2APSubscriptionRequests); index++ {
                subReqMsg := subReqList.E2APSubscriptionRequests[index]
+               xAppEventInstanceID = (int64)(subReqMsg.RequestId.Id)
 
-               xid := *restSubId + "_" + strconv.FormatUint(uint64(subReqMsg.RequestId.InstanceId), 10)
-               trans := c.tracker.NewXappTransaction(xapp.NewRmrEndpoint(xAppRmrEndpoint), xid, subReqMsg.RequestId, &xapp.RMRMeid{RanName: *meid})
-               //trans := c.tracker.NewXappTransaction(xapp.NewRmrEndpoint(xAppRmrEndpoint), *restSubId, subReqMsg.RequestId, &xapp.RMRMeid{RanName: *meid})
+               trans := c.tracker.NewXappTransaction(xapp.NewRmrEndpoint(xAppRmrEndpoint), *restSubId, subReqMsg.RequestId, &xapp.RMRMeid{RanName: *meid})
                if trans == nil {
-                       c.registry.DeleteRESTSubscription(restSubId)
-                       xapp.Logger.Error("XAPP-SubReq transaction not created. RESTSubId=%s, EndPoint=%s, Meid=%s", *restSubId, xAppRmrEndpoint, *meid)
-                       return
+                       // Send notification to xApp that prosessing of a Subscription Request has failed.
+                       err := fmt.Errorf("Tracking failure")
+                       c.sendUnsuccesfullResponseNotification(restSubId, restSubscription, xAppEventInstanceID, err, clientEndpoint, trans)
+                       continue
                }
 
-               defer trans.Release()
                xapp.Logger.Info("Handle SubscriptionRequest index=%v, %s", index, idstring(nil, trans))
+
                subRespMsg, err := c.handleSubscriptionRequest(trans, &subReqMsg, meid, restSubId)
                if err != nil {
-                       // Send notification to xApp that prosessing of a Subscription Request has failed. Currently it is not possible
-                       // to indicate error. Such possibility should be added. As a workaround requestorID and instanceId are set to zero value
-                       requestorID = (int64)(0)
-                       instanceId = (int64)(0)
-                       resp := &models.SubscriptionResponse{
-                               SubscriptionID: restSubId,
-                               SubscriptionInstances: []*models.SubscriptionInstance{
-                                       &models.SubscriptionInstance{RequestorID: &requestorID, InstanceID: &instanceId},
-                               },
-                       }
-                       // Mark REST subscription request processed.
-                       restSubscription.SetProcessed()
-                       xapp.Logger.Info("Sending unsuccessful REST notification to endpoint=%v:%v, InstanceId=%v, %s", clientEndpoint.Host, clientEndpoint.HTTPPort, instanceId, idstring(nil, trans))
-                       xapp.Subscription.Notify(resp, *clientEndpoint)
-                       c.UpdateCounter(cRestSubFailNotifToXapp)
+                       c.sendUnsuccesfullResponseNotification(restSubId, restSubscription, xAppEventInstanceID, err, clientEndpoint, trans)
                } else {
-                       xapp.Logger.Info("SubscriptionRequest index=%v processed successfully. endpoint=%v, InstanceId=%v, %s", index, *clientEndpoint, instanceId, idstring(nil, trans))
-
-                       // Store successfully processed InstanceId for deletion
-                       restSubscription.AddInstanceId(subRespMsg.RequestId.InstanceId)
-
-                       // Send notification to xApp that a Subscription Request has been processed.
-                       requestorID = (int64)(subRespMsg.RequestId.Id)
-                       instanceId = (int64)(subRespMsg.RequestId.InstanceId)
-                       resp := &models.SubscriptionResponse{
-                               SubscriptionID: restSubId,
-                               SubscriptionInstances: []*models.SubscriptionInstance{
-                                       &models.SubscriptionInstance{RequestorID: &requestorID, InstanceID: &instanceId},
-                               },
-                       }
-                       // Mark REST subscription request processesd.
-                       restSubscription.SetProcessed()
-                       xapp.Logger.Info("Sending successful REST notification to endpoint=%v, InstanceId=%v, %s", *clientEndpoint, instanceId, idstring(nil, trans))
-                       xapp.Subscription.Notify(resp, *clientEndpoint)
-                       c.UpdateCounter(cRestSubNotifToXapp)
-
+                       e2EventInstanceID = (int64)(subRespMsg.RequestId.InstanceId)
+                       xapp.Logger.Info("SubscriptionRequest index=%v processed successfully. endpoint=%v:%v, XappEventInstanceID=%v, E2EventInstanceID=%v, %s",
+                               index, clientEndpoint.Host, *clientEndpoint.HTTPPort, xAppEventInstanceID, e2EventInstanceID, idstring(nil, trans))
+                       c.sendSuccesfullResponseNotification(restSubId, restSubscription, xAppEventInstanceID, e2EventInstanceID, clientEndpoint, trans)
                }
+               trans.Release()
        }
 }
 
@@ -358,15 +408,14 @@ func (c *Control) handleSubscriptionRequest(trans *TransactionXapp, subReqMsg *e
 
        err := c.tracker.Track(trans)
        if err != nil {
-               err = fmt.Errorf("XAPP-SubReq: %s", idstring(err, trans))
-               xapp.Logger.Error("%s", err.Error())
+               xapp.Logger.Error("XAPP-SubReq Tracking error: %s", idstring(err, trans))
+               err = fmt.Errorf("Tracking failure")
                return nil, err
        }
 
        subs, err := c.registry.AssignToSubscription(trans, subReqMsg, c.ResetTestFlag, c)
        if err != nil {
-               err = fmt.Errorf("XAPP-SubReq: %s", idstring(err, trans))
-               xapp.Logger.Error("%s", err.Error())
+               xapp.Logger.Error("XAPP-SubReq Assign error: %s", idstring(err, trans))
                return nil, err
        }
 
@@ -383,14 +432,17 @@ func (c *Control) handleSubscriptionRequest(trans *TransactionXapp, subReqMsg *e
                        trans.Release()
                        return themsg, nil
                case *e2ap.E2APSubscriptionFailure:
-                       err = fmt.Errorf("SubscriptionFailure received")
+                       err = fmt.Errorf("E2 SubscriptionFailure received")
                        return nil, err
                default:
+                       err = fmt.Errorf("unexpected E2 subscription response received")
                        break
                }
+       } else {
+               err = fmt.Errorf("E2 subscription response timeout")
        }
-       err = fmt.Errorf("XAPP-SubReq: failed %s", idstring(err, trans, subs))
-       xapp.Logger.Error("%s", err.Error())
+
+       xapp.Logger.Error("XAPP-SubReq E2 subscription failed %s", idstring(err, trans, subs))
        c.registry.RemoveFromSubscription(subs, trans, waitRouteCleanup_ms, c)
        return nil, err
 }
@@ -405,7 +457,7 @@ func (c *Control) SubscriptionDeleteHandlerCB(restSubId string) error {
 
        xapp.Logger.Info("SubscriptionDeleteRequest from XAPP")
 
-       restSubscription, err := c.registry.GetRESTSubscription(restSubId)
+       restSubscription, err := c.registry.GetRESTSubscription(restSubId, true)
        if err != nil {
                xapp.Logger.Error("%s", err.Error())
                if restSubscription == nil {
@@ -426,13 +478,15 @@ func (c *Control) SubscriptionDeleteHandlerCB(restSubId string) error {
        xAppRmrEndPoint := restSubscription.xAppRmrEndPoint
        go func() {
                for _, instanceId := range restSubscription.InstanceIds {
-                       err := c.SubscriptionDeleteHandler(&restSubId, &xAppRmrEndPoint, &restSubscription.Meid, instanceId)
+                       xAppEventInstanceID, err := c.SubscriptionDeleteHandler(&restSubId, &xAppRmrEndPoint, &restSubscription.Meid, instanceId)
+
                        if err != nil {
                                xapp.Logger.Error("%s", err.Error())
                                //return err
                        }
                        xapp.Logger.Info("Deleteting instanceId = %v", instanceId)
-                       restSubscription.DeleteInstanceId(instanceId)
+                       restSubscription.DeleteXappIdToE2Id(xAppEventInstanceID)
+                       restSubscription.DeleteE2InstanceId(instanceId)
                }
                c.registry.DeleteRESTSubscription(&restSubId)
        }()
@@ -445,29 +499,29 @@ func (c *Control) SubscriptionDeleteHandlerCB(restSubId string) error {
 //-------------------------------------------------------------------
 //
 //-------------------------------------------------------------------
-func (c *Control) SubscriptionDeleteHandler(restSubId *string, endPoint *string, meid *string, instanceId uint32) error {
+func (c *Control) SubscriptionDeleteHandler(restSubId *string, endPoint *string, meid *string, instanceId uint32) (int64, error) {
 
-       xid := *restSubId + "_" + strconv.FormatUint(uint64(instanceId), 10)
-       trans := c.tracker.NewXappTransaction(xapp.NewRmrEndpoint(*endPoint), xid, e2ap.RequestId{0, 0}, &xapp.RMRMeid{RanName: *meid})
-       //trans := c.tracker.NewXappTransaction(xapp.NewRmrEndpoint(*endPoint), *restSubId, e2ap.RequestId{0, 0}, &xapp.RMRMeid{RanName: *meid})
+       var xAppEventInstanceID int64
+       subs, err := c.registry.GetSubscriptionFirstMatch([]uint32{instanceId})
+       if err != nil {
+               xapp.Logger.Info("Subscription Delete Handler subscription for restSubId=%v, E2EventInstanceID=%v not found %s",
+                       restSubId, instanceId, idstring(err, nil))
+               return xAppEventInstanceID, nil
+       }
+
+       xAppEventInstanceID = int64(subs.ReqId.Id)
+       trans := c.tracker.NewXappTransaction(xapp.NewRmrEndpoint(*endPoint), *restSubId, e2ap.RequestId{subs.ReqId.Id, 0}, &xapp.RMRMeid{RanName: *meid})
        if trans == nil {
                err := fmt.Errorf("XAPP-SubDelReq transaction not created. restSubId %s, endPoint %s, meid %s, instanceId %v", *restSubId, *endPoint, *meid, instanceId)
                xapp.Logger.Error("%s", err.Error())
        }
        defer trans.Release()
 
-       err := c.tracker.Track(trans)
-       if err != nil {
-               err := fmt.Errorf("XAPP-SubDelReq %s:", idstring(err, trans))
-               xapp.Logger.Error("%s", err.Error())
-               return &time.ParseError{}
-       }
-
-       subs, err := c.registry.GetSubscriptionFirstMatch([]uint32{instanceId})
+       err = c.tracker.Track(trans)
        if err != nil {
                err := fmt.Errorf("XAPP-SubDelReq %s:", idstring(err, trans))
                xapp.Logger.Error("%s", err.Error())
-               return err
+               return xAppEventInstanceID, &time.ParseError{}
        }
        //
        // Wake subs delete
@@ -479,7 +533,7 @@ func (c *Control) SubscriptionDeleteHandler(restSubId *string, endPoint *string,
 
        c.registry.RemoveFromSubscription(subs, trans, waitRouteCleanup_ms, c)
 
-       return nil
+       return xAppEventInstanceID, nil
 }
 
 //-------------------------------------------------------------------
@@ -820,9 +874,11 @@ func (c *Control) sendE2TSubscriptionRequest(subs *Subscription, trans *Transact
        var err error
        var event interface{} = nil
        var timedOut bool = false
+       const ricRequestorId = 123
 
        subReqMsg := subs.SubReqMsg
        subReqMsg.RequestId = subs.GetReqId().RequestId
+       subReqMsg.RequestId.Id = ricRequestorId
        trans.Mtype, trans.Payload, err = c.e2ap.PackSubscriptionRequest(subReqMsg)
        if err != nil {
                xapp.Logger.Error("SUBS-SubReq: %s", idstring(err, trans, subs, parentTrans))
@@ -864,9 +920,11 @@ func (c *Control) sendE2TSubscriptionDeleteRequest(subs *Subscription, trans *Tr
        var err error
        var event interface{}
        var timedOut bool
+       const ricRequestorId = 123
 
        subDelReqMsg := &e2ap.E2APSubscriptionDeleteRequest{}
        subDelReqMsg.RequestId = subs.GetReqId().RequestId
+       subDelReqMsg.RequestId.Id = ricRequestorId
        subDelReqMsg.FunctionId = subs.SubReqMsg.FunctionId
        trans.Mtype, trans.Payload, err = c.e2ap.PackSubscriptionDeleteRequest(subDelReqMsg)
        if err != nil {
@@ -1080,11 +1138,13 @@ func (c *Control) RemoveSubscriptionFromDb(subs *Subscription) {
 
 func (c *Control) SendSubscriptionDeleteReq(subs *Subscription) {
 
+       const ricRequestorId = 123
        xapp.Logger.Debug("Sending subscription delete due to restart. subId = %v", subs.ReqId.InstanceId)
 
        // Send delete for every endpoint in the subscription
        subDelReqMsg := &e2ap.E2APSubscriptionDeleteRequest{}
        subDelReqMsg.RequestId = subs.GetReqId().RequestId
+       subDelReqMsg.RequestId.Id = ricRequestorId
        subDelReqMsg.FunctionId = subs.SubReqMsg.FunctionId
        mType, payload, err := c.e2ap.PackSubscriptionDeleteRequest(subDelReqMsg)
        if err != nil {
@@ -1135,19 +1195,14 @@ func (c *Control) PrintRESTSubscriptionRequest(p *models.SubscriptionParams) {
                } else {
                        fmt.Println("  RANFunctionID = nil")
                }
-               fmt.Printf("  SubscriptionDetail.RequestorID = %v\n", *subscriptionDetail.RequestorID)
-               fmt.Printf("  SubscriptionDetail.InstanceID = %v\n", *subscriptionDetail.InstanceID)
-               fmt.Printf("  SubscriptionDetail.EventTriggers.OctetString = %X\n", subscriptionDetail.EventTriggers.OctetString)
+               fmt.Printf("  SubscriptionDetail.XappEventInstanceID = %v\n", *subscriptionDetail.XappEventInstanceID)
+               fmt.Printf("  SubscriptionDetail.EventTriggers = %v\n", subscriptionDetail.EventTriggers)
 
                for _, actionToBeSetup := range subscriptionDetail.ActionToBeSetupList {
                        fmt.Printf("  SubscriptionDetail.ActionToBeSetup.ActionID = %v\n", *actionToBeSetup.ActionID)
                        fmt.Printf("  SubscriptionDetail.ActionToBeSetup.ActionType = %s\n", *actionToBeSetup.ActionType)
-                       if actionToBeSetup.ActionDefinition != nil {
-                               fmt.Printf("  SubscriptionDetail.ActionToBeSetup.ActionDefinition.OctetString = %X\n", actionToBeSetup.ActionDefinition.OctetString)
-                       } else {
-                               fmt.Println("  SubscriptionDetail.ActionToBeSetup.ActionDefinition = nil")
+                       fmt.Printf("  SubscriptionDetail.ActionToBeSetup.ActionDefinition = %v\n", actionToBeSetup.ActionDefinition)
 
-                       }
                        if actionToBeSetup.SubsequentAction != nil {
                                fmt.Printf("  SubscriptionDetail.ActionToBeSetup.SubsequentAction.SubsequentActionType = %s\n", *actionToBeSetup.SubsequentAction.SubsequentActionType)
                                fmt.Printf("  SubscriptionDetail.ActionToBeSetup..SubsequentAction.TimeToWait = %s\n", *actionToBeSetup.SubsequentAction.TimeToWait)