X-Git-Url: https://gerrit.o-ran-sc.org/r/gitweb?a=blobdiff_plain;f=pkg%2Fcontrol%2Fcontrol.go;h=977949880448dbb7213ae8d7e86a3b226d64ea20;hb=987f8b1fdb2dd8413d6c8faef1eb8618226e0421;hp=0c2f917e4b256baad519a935b51f722602ab947b;hpb=de457a14f98c546fdb009e9915745ef398388ddd;p=ric-plt%2Fsubmgr.git diff --git a/pkg/control/control.go b/pkg/control/control.go index 0c2f917..9779498 100755 --- a/pkg/control/control.go +++ b/pkg/control/control.go @@ -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,119 @@ 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) + } + + c.UpdateCounter(cRestSubFailNotifToXapp) + xapp.Subscription.Notify(resp, *clientEndpoint) +} + +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)) + + c.UpdateCounter(cRestSubNotifToXapp) + xapp.Subscription.Notify(resp, *clientEndpoint) +} + //------------------------------------------------------------------- // //------------------------------------------------------------------- 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 +410,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 +434,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 +459,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 +480,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 +501,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) { + + 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 + } - 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}) + 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 +535,7 @@ func (c *Control) SubscriptionDeleteHandler(restSubId *string, endPoint *string, c.registry.RemoveFromSubscription(subs, trans, waitRouteCleanup_ms, c) - return nil + return xAppEventInstanceID, nil } //------------------------------------------------------------------- @@ -820,9 +876,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 +922,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 +1140,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 { @@ -1109,6 +1171,13 @@ func (c *Control) SendSubscriptionDeleteReq(subs *Subscription) { func (c *Control) PrintRESTSubscriptionRequest(p *models.SubscriptionParams) { fmt.Println("CRESTSubscriptionRequest") + + if p.SubscriptionID != "" { + fmt.Println(" SubscriptionID = ", p.SubscriptionID) + } else { + fmt.Println(" SubscriptionID = ''") + } + fmt.Printf(" ClientEndpoint.Host = %s\n", p.ClientEndpoint.Host) if p.ClientEndpoint.HTTPPort != nil { @@ -1135,19 +1204,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)