X-Git-Url: https://gerrit.o-ran-sc.org/r/gitweb?a=blobdiff_plain;f=pkg%2Fcontrol%2Fcontrol.go;h=ef27530e14491858efea0884c01815d4d988f98b;hb=662f68d35c10fa67382235a20e1e08c50359989a;hp=1bd27ebef23a00cde26a312bd8143e0c0c793105;hpb=60bfcf92de9e13b60acf1fd3e18bfaeb6a2d389b;p=ric-plt%2Fsubmgr.git diff --git a/pkg/control/control.go b/pkg/control/control.go index 1bd27eb..ef27530 100755 --- a/pkg/control/control.go +++ b/pkg/control/control.go @@ -45,7 +45,6 @@ var maxSubDelReqTryCount uint64 = 2 // Initial try + retry type Control struct { e2ap *E2ap registry *Registry - rtmgrClient *RtmgrClient tracker *Tracker timerMap *TimerMap rmrSendMutex sync.Mutex @@ -85,8 +84,15 @@ func init() { func NewControl() *Control { + transport := httptransport.New(viper.GetString("rtmgr.HostAddr")+":"+viper.GetString("rtmgr.port"), viper.GetString("rtmgr.baseUrl"), []string{"http"}) + client := rtmgrclient.New(transport, strfmt.Default) + handle := rtmgrhandle.NewProvideXappSubscriptionHandleParamsWithTimeout(10 * time.Second) + deleteHandle := rtmgrhandle.NewDeleteXappSubscriptionHandleParamsWithTimeout(10 * time.Second) + rtmgrClient := RtmgrClient{client, handle, deleteHandle} + registry := new(Registry) registry.Initialize(seedSN) + registry.rtmgrClient = &rtmgrClient tracker := new(Tracker) tracker.Init() @@ -94,23 +100,11 @@ func NewControl() *Control { timerMap := new(TimerMap) timerMap.Init() - transport := httptransport.New(viper.GetString("rtmgr.HostAddr")+":"+viper.GetString("rtmgr.port"), viper.GetString("rtmgr.baseUrl"), []string{"http"}) - client := rtmgrclient.New(transport, strfmt.Default) - handle := rtmgrhandle.NewProvideXappSubscriptionHandleParamsWithTimeout(10 * time.Second) - deleteHandle := rtmgrhandle.NewDeleteXappSubscriptionHandleParamsWithTimeout(10 * time.Second) - rtmgrClient := RtmgrClient{client, handle, deleteHandle} - - rtmgrClientPtr := &rtmgrClient - - //TODO: to make this better. Now it is just a hack. - registry.rtmgrClient = rtmgrClientPtr - return &Control{e2ap: new(E2ap), - registry: registry, - rtmgrClient: rtmgrClientPtr, - tracker: tracker, - timerMap: timerMap, - msgCounter: 0, + registry: registry, + tracker: tracker, + timerMap: timerMap, + msgCounter: 0, } } @@ -140,29 +134,29 @@ func (c *Control) rmrSendRaw(desc string, params *RMRParams) (err error) { return } -func (c *Control) rmrSend(desc string, subs *Subscription, trans *Transaction, payload []byte, payloadLen int) (err error) { +func (c *Control) rmrSend(desc string, subs *Subscription, trans *Transaction) (err error) { params := &RMRParams{&xapp.RMRParams{}} params.Mtype = trans.GetMtype() params.SubId = int(subs.GetSubId()) params.Xid = "" params.Meid = subs.GetMeid() params.Src = "" - params.PayloadLen = payloadLen - params.Payload = payload + params.PayloadLen = len(trans.Payload.Buf) + params.Payload = trans.Payload.Buf params.Mbuf = nil return c.rmrSendRaw(desc, params) } -func (c *Control) rmrReplyToSender(desc string, subs *Subscription, trans *Transaction, mType int, payload []byte, payloadLen int) (err error) { +func (c *Control) rmrReplyToSender(desc string, subs *Subscription, trans *Transaction) (err error) { params := &RMRParams{&xapp.RMRParams{}} - params.Mtype = mType + params.Mtype = trans.GetMtype() params.SubId = int(subs.GetSubId()) params.Xid = trans.GetXid() params.Meid = trans.GetMeid() params.Src = "" - params.PayloadLen = payloadLen - params.Payload = payload + params.PayloadLen = len(trans.Payload.Buf) + params.Payload = trans.Payload.Buf params.Mbuf = nil return c.rmrSendRaw(desc, params) @@ -200,7 +194,6 @@ func (c *Control) handleSubscriptionRequest(params *RMRParams) { // // trans, err := c.tracker.TrackTransaction(NewRmrEndpoint(params.Src), - params.Mtype, params.Xid, params.Meid, false, @@ -234,7 +227,7 @@ func (c *Control) handleSubscriptionRequest(params *RMRParams) { err = subs.SetTransaction(trans) if err != nil { xapp.Logger.Error("SubReq: %s, Dropping this msg. %s", err.Error(), trans) - c.registry.DelSubscription(subs.Seq) + subs.Release() trans.Release() return } @@ -249,19 +242,15 @@ func (c *Control) handleSubscriptionRequest(params *RMRParams) { // // This is intermediate solution while improving message handling // - packedData, err := c.e2ap.PackSubscriptionRequest(trans.SubReqMsg) + trans.Mtype, trans.Payload, err = c.e2ap.PackSubscriptionRequest(trans.SubReqMsg) if err != nil { xapp.Logger.Error("SubReq: %s for trans %s", err.Error(), trans) - c.registry.DelSubscription(subs.Seq) + subs.Release() trans.Release() return } - //Optimize and store packed message to be sent (for retransmission). Again owned by subscription? - trans.Payload = packedData.Buf - trans.PayloadLen = len(packedData.Buf) - - c.rmrSend("SubReq to E2T", subs, trans, packedData.Buf, len(packedData.Buf)) + c.rmrSend("SubReq: SubReq to E2T", subs, trans) c.timerMap.StartTimer("RIC_SUB_REQ", int(subs.GetSubId()), subReqTime, FirstTry, c.handleSubscriptionRequestTimer) xapp.Logger.Debug("SubReq: Debugging trans table = %v", c.tracker.transactionXappTable) @@ -276,7 +265,7 @@ func (c *Control) handleSubscriptionResponse(params *RMRParams) { // SubRespMsg, err := c.e2ap.UnpackSubscriptionResponse(params.Payload) if err != nil { - xapp.Logger.Error("SubDelReq: %s Dropping this msg. %s", err.Error(), params.String()) + xapp.Logger.Error("SubResp: %s Dropping this msg. %s", err.Error(), params.String()) return } @@ -316,20 +305,16 @@ func (c *Control) handleSubscriptionResponse(params *RMRParams) { return } - packedData, err := c.e2ap.PackSubscriptionResponse(trans.SubRespMsg) + trans.Mtype, trans.Payload, err = c.e2ap.PackSubscriptionResponse(trans.SubRespMsg) if err != nil { xapp.Logger.Error("SubResp: %s for trans %s", err.Error(), trans) trans.Release() return } - //Optimize and store packed message to be sent. - trans.Payload = packedData.Buf - trans.PayloadLen = len(packedData.Buf) - subs.Confirmed() trans.Release() - c.rmrReplyToSender("SubResp to xapp", subs, trans, 12011, trans.Payload, trans.PayloadLen) + c.rmrReplyToSender("SubResp: SubResp to xapp", subs, trans) return } @@ -384,22 +369,17 @@ func (c *Control) handleSubscriptionFailure(params *RMRParams) { return } - packedData, err := c.e2ap.PackSubscriptionFailure(trans.SubFailMsg) - if err != nil { - //TODO error handling improvement - xapp.Logger.Error("SubFail: %s for trans %s (continue still)", err.Error(), trans) - } else { - //Optimize and store packed message to be sent. - trans.Payload = packedData.Buf - trans.PayloadLen = len(packedData.Buf) - c.rmrReplyToSender("SubFail to xapp", subs, trans, 12012, trans.Payload, trans.PayloadLen) + trans.Mtype, trans.Payload, err = c.e2ap.PackSubscriptionFailure(trans.SubFailMsg) + if err == nil { + c.rmrReplyToSender("SubFail: SubFail to xapp", subs, trans) time.Sleep(3 * time.Second) + } else { + //TODO error handling improvement + xapp.Logger.Error("SubFail: %s for trans %s (continuing cleaning)", err.Error(), trans) } trans.Release() - if !c.registry.DelSubscription(subs.GetSubId()) { - xapp.Logger.Error("SubFail: Failed to release sequency number. %s", subs) - } + subs.Release() return } @@ -426,23 +406,22 @@ func (c *Control) handleSubscriptionRequestTimer(strId string, nbrId int, tryCou } if tryCount < maxSubReqTryCount { - xapp.Logger.Info("SubReq timeout: Resending SubReq to E2T: Mtype: %v, SubId: %v, Xid %s, Meid %v", trans.GetMtype(), subs.GetSubId(), trans.GetXid(), trans.GetMeid()) + xapp.Logger.Info("SubReq timeout: subs: %s trans: %s", subs, trans) trans.RetryTransaction() - c.rmrSend("SubReq(SubReq timer) to E2T", subs, trans, trans.Payload, trans.PayloadLen) + c.rmrSend("SubReq timeout: SubReq to E2T", subs, trans) tryCount++ c.timerMap.StartTimer("RIC_SUB_REQ", int(subs.GetSubId()), subReqTime, tryCount, c.handleSubscriptionRequestTimer) return } - // Delete CREATE transaction + // Release CREATE transaction trans.Release() // Create DELETE transaction (internal and no messages toward xapp) deltrans, err := c.tracker.TrackTransaction(&trans.RmrEndpoint, - 12020, // RIC SUBSCRIPTION DELETE trans.GetXid(), trans.GetMeid(), false, @@ -451,7 +430,7 @@ func (c *Control) handleSubscriptionRequestTimer(strId string, nbrId int, tryCou if err != nil { xapp.Logger.Error("SubReq timeout: %s, Dropping this msg.", err.Error()) //TODO improve error handling. Important at least in merge - c.registry.DelSubscription(subs.GetSubId()) + subs.Release() return } @@ -459,16 +438,14 @@ func (c *Control) handleSubscriptionRequestTimer(strId string, nbrId int, tryCou deltrans.SubDelReqMsg.RequestId.Id = trans.SubReqMsg.RequestId.Id deltrans.SubDelReqMsg.RequestId.Seq = uint32(subs.GetSubId()) deltrans.SubDelReqMsg.FunctionId = trans.SubReqMsg.FunctionId - packedData, err := c.e2ap.PackSubscriptionDeleteRequest(deltrans.SubDelReqMsg) + deltrans.Mtype, deltrans.Payload, err = c.e2ap.PackSubscriptionDeleteRequest(deltrans.SubDelReqMsg) if err != nil { xapp.Logger.Error("SubReq timeout: Packing SubDelReq failed. Err: %v", err) //TODO improve error handling. Important at least in merge deltrans.Release() - c.registry.DelSubscription(subs.GetSubId()) + subs.Release() return } - deltrans.PayloadLen = len(packedData.Buf) - deltrans.Payload = packedData.Buf err = subs.SetTransaction(deltrans) if err != nil { @@ -478,8 +455,7 @@ func (c *Control) handleSubscriptionRequestTimer(strId string, nbrId int, tryCou return } - c.rmrSend("SubDelReq(SubReq timer) to E2T", subs, deltrans, deltrans.Payload, deltrans.PayloadLen) - + c.rmrSend("SubReq timer: SubDelReq to E2T", subs, deltrans) c.timerMap.StartTimer("RIC_SUB_DEL_REQ", int(subs.GetSubId()), subDelReqTime, FirstTry, c.handleSubscriptionDeleteRequestTimer) return } @@ -491,7 +467,6 @@ func (c *Control) handleSubscriptionDeleteRequest(params *RMRParams) { // // trans, err := c.tracker.TrackTransaction(NewRmrEndpoint(params.Src), - params.Mtype, params.Xid, params.Meid, false, @@ -542,20 +517,16 @@ func (c *Control) handleSubscriptionDeleteRequest(params *RMRParams) { // // This is intermediate solution while improving message handling // - packedData, err := c.e2ap.PackSubscriptionDeleteRequest(trans.SubDelReqMsg) + trans.Mtype, trans.Payload, err = c.e2ap.PackSubscriptionDeleteRequest(trans.SubDelReqMsg) if err != nil { xapp.Logger.Error("SubDelReq: %s for trans %s", err.Error(), trans) trans.Release() return } - //Optimize and store packed message to be sent (for retransmission). Again owned by subscription? - trans.Payload = packedData.Buf - trans.PayloadLen = len(packedData.Buf) - subs.UnConfirmed() - c.rmrSend("SubDelReq to E2T", subs, trans, trans.Payload, trans.PayloadLen) + c.rmrSend("SubDelReq: SubDelReq to E2T", subs, trans) c.timerMap.StartTimer("RIC_SUB_DEL_REQ", int(subs.GetSubId()), subDelReqTime, FirstTry, c.handleSubscriptionDeleteRequestTimer) return @@ -564,25 +535,43 @@ func (c *Control) handleSubscriptionDeleteRequest(params *RMRParams) { func (c *Control) handleSubscriptionDeleteResponse(params *RMRParams) (err error) { xapp.Logger.Info("SubDelResp from E2T:%s", params.String()) - payloadSeqNum, err := c.e2ap.GetSubscriptionDeleteResponseSequenceNumber(params.Payload) + // + // + // + SubDelRespMsg, err := c.e2ap.UnpackSubscriptionDeleteResponse(params.Payload) if err != nil { - xapp.Logger.Error("SubDelResp: Unable to get Sequence Number from Payload. Dropping this msg. Err: %v, SubId: %v, Xid: %s, Payload %X", err, params.SubId, params.Xid, params.Payload) + xapp.Logger.Error("SubDelResp: %s Dropping this msg. %s", err.Error(), params.String()) return } - xapp.Logger.Info("SubDelResp: Received payloadSeqNum: %v", payloadSeqNum) - subs := c.registry.GetSubscription(payloadSeqNum) + // + // + // + subs := c.registry.GetSubscription(uint16(SubDelRespMsg.RequestId.Seq)) + if subs == nil && params.SubId > 0 { + subs = c.registry.GetSubscription(uint16(params.SubId)) + } + if subs == nil { - xapp.Logger.Error("SubDelResp: Unknown payloadSeqNum. Dropping this msg. PayloadSeqNum: %v, SubId: %v", payloadSeqNum, params.SubId) + xapp.Logger.Error("SubDelResp: Not valid subscription found payloadSeqNum: %d, SubId: %d. Dropping this msg. %s", SubDelRespMsg.RequestId.Seq, params.SubId, params.String()) return } + xapp.Logger.Info("SubDelResp: subscription found payloadSeqNum: %d, SubId: %d", SubDelRespMsg.RequestId.Seq, subs.GetSubId()) + // + // + // trans := subs.GetTransaction() if trans == nil { - xapp.Logger.Error("SubDelResp: Unknown trans. Dropping this msg. PayloadSeqNum: %v, SubId: %v", subs.GetSubId(), params.SubId) + xapp.Logger.Error("SubDelResp: Unknown trans. Dropping this msg. SubId: %d", subs.GetSubId()) return } + trans.SubDelRespMsg = SubDelRespMsg + + // + // + // c.timerMap.StopTimer("RIC_SUB_DEL_REQ", int(subs.GetSubId())) responseReceived := trans.CheckResponseReceived() @@ -591,43 +580,49 @@ func (c *Control) handleSubscriptionDeleteResponse(params *RMRParams) (err error return } - trans.Release() - - if trans.ForwardRespToXapp == true { - c.rmrReplyToSender("SubDelResp to xapp", subs, trans, params.Mtype, params.Payload, params.PayloadLen) - time.Sleep(3 * time.Second) - } - - xapp.Logger.Info("SubDelResp: Deleting trans record. SubId: %v, Xid: %s", subs.GetSubId(), trans.GetXid()) - if !c.registry.DelSubscription(subs.GetSubId()) { - xapp.Logger.Error("SubDelResp: Failed to release sequency number. SubId: %v, Xid: %s", subs.GetSubId(), trans.GetXid()) - return - } + c.sendSubscriptionDeleteResponse("SubDelResp", trans, subs) return } func (c *Control) handleSubscriptionDeleteFailure(params *RMRParams) { xapp.Logger.Info("SubDelFail from E2T:%s", params.String()) - payloadSeqNum, err := c.e2ap.GetSubscriptionDeleteFailureSequenceNumber(params.Payload) + // + // + // + SubDelFailMsg, err := c.e2ap.UnpackSubscriptionDeleteFailure(params.Payload) if err != nil { - xapp.Logger.Error("SubDelFail: Unable to get Sequence Number from Payload. Dropping this msg. Err: %v, %s", err, params.String()) + xapp.Logger.Error("SubDelFail: %s Dropping this msg. %s", err.Error(), params.String()) return } - xapp.Logger.Info("SubDelFail: Received payloadSeqNum: %v", payloadSeqNum) - subs := c.registry.GetSubscription(payloadSeqNum) + // + // + // + subs := c.registry.GetSubscription(uint16(SubDelFailMsg.RequestId.Seq)) + if subs == nil && params.SubId > 0 { + subs = c.registry.GetSubscription(uint16(params.SubId)) + } + if subs == nil { - xapp.Logger.Error("SubDelFail: Unknown payloadSeqNum. Dropping this msg. PayloadSeqNum: %v, SubId: %v", payloadSeqNum, params.SubId) + xapp.Logger.Error("SubDelFail: Not valid subscription found payloadSeqNum: %d, SubId: %d. Dropping this msg. %s", SubDelFailMsg.RequestId.Seq, params.SubId, params.String()) return } + xapp.Logger.Info("SubDelFail: subscription found payloadSeqNum: %d, SubId: %d", SubDelFailMsg.RequestId.Seq, subs.GetSubId()) + // + // + // trans := subs.GetTransaction() if trans == nil { - xapp.Logger.Error("SubDelFail: Unknown trans. Dropping this msg. PayloadSeqNum: %v, SubId: %v", subs.GetSubId(), params.SubId) + xapp.Logger.Error("SubDelFail: Unknown trans. Dropping this msg. SubId: %d", subs.GetSubId()) return } + trans.SubDelFailMsg = SubDelFailMsg + // + // + // c.timerMap.StopTimer("RIC_SUB_DEL_REQ", int(subs.GetSubId())) responseReceived := trans.CheckResponseReceived() @@ -635,25 +630,8 @@ func (c *Control) handleSubscriptionDeleteFailure(params *RMRParams) { // Subscription Delete timer already received return } - if trans.ForwardRespToXapp == true { - var subDelRespPayload []byte - subDelRespPayload, err = c.e2ap.PackSubscriptionDeleteResponseFromSubDelReq(trans.Payload, subs.GetSubId()) - if err != nil { - xapp.Logger.Error("SubDelFail:Packing SubDelResp failed. Err: %v", err) - return - } - - // RIC SUBSCRIPTION DELETE RESPONSE - c.rmrReplyToSender("SubDelFail to xapp", subs, trans, 12021, subDelRespPayload, len(subDelRespPayload)) - time.Sleep(3 * time.Second) - } - xapp.Logger.Info("SubDelFail: Deleting trans record. SubId: %v, Xid: %s", subs.GetSubId(), trans.GetXid()) - trans.Release() - if !c.registry.DelSubscription(subs.GetSubId()) { - xapp.Logger.Error("SubDelFail: Failed to release sequency number. Err: %v, SubId: %v, Xid: %s", err, subs.GetSubId(), trans.GetXid()) - return - } + c.sendSubscriptionDeleteResponse("SubDelFail", trans, subs) return } @@ -679,37 +657,38 @@ func (c *Control) handleSubscriptionDeleteRequestTimer(strId string, nbrId int, } if tryCount < maxSubDelReqTryCount { - xapp.Logger.Info("SubDelReq timeout: Resending SubDelReq to E2T: Mtype: %v, SubId: %v, Xid %s, Meid %v", trans.GetMtype(), subs.GetSubId(), trans.GetXid(), trans.GetMeid()) // Set possible to handle new response for the subId - trans.RetryTransaction() - - c.rmrSend("SubDelReq(SubDelReq timer) to E2T", subs, trans, trans.Payload, trans.PayloadLen) - + c.rmrSend("SubDelReq timeout: SubDelReq to E2T", subs, trans) tryCount++ c.timerMap.StartTimer("RIC_SUB_DEL_REQ", int(subs.GetSubId()), subReqTime, tryCount, c.handleSubscriptionDeleteRequestTimer) return } - if trans.ForwardRespToXapp == true { - var subDelRespPayload []byte - subDelRespPayload, err := c.e2ap.PackSubscriptionDeleteResponseFromSubDelReq(trans.Payload, subs.GetSubId()) - if err != nil { - xapp.Logger.Error("SubDelReq timeout: Unable to pack payload. Dropping this this msg. Err: %v, SubId: %v, Xid: %s, Payload %x", err, subs.GetSubId(), trans.GetXid(), trans.Payload) - return - } - - // RIC SUBSCRIPTION DELETE RESPONSE - c.rmrReplyToSender("SubDelResp(SubDelReq timer) to xapp", subs, trans, 12021, subDelRespPayload, len(subDelRespPayload)) + c.sendSubscriptionDeleteResponse("SubDelReq(timer)", trans, subs) + return +} - time.Sleep(3 * time.Second) +func (c *Control) sendSubscriptionDeleteResponse(desc string, trans *Transaction, subs *Subscription) { + if trans.ForwardRespToXapp == true { + //Always generate SubDelResp + trans.SubDelRespMsg = &e2ap.E2APSubscriptionDeleteResponse{} + trans.SubDelRespMsg.RequestId.Id = trans.SubDelReqMsg.RequestId.Id + trans.SubDelRespMsg.RequestId.Seq = uint32(subs.GetSubId()) + trans.SubDelRespMsg.FunctionId = trans.SubDelReqMsg.FunctionId + + var err error + trans.Mtype, trans.Payload, err = c.e2ap.PackSubscriptionDeleteResponse(trans.SubDelRespMsg) + if err == nil { + c.rmrReplyToSender(desc+": SubDelResp to xapp", subs, trans) + time.Sleep(3 * time.Second) + } else { + //TODO error handling improvement + xapp.Logger.Error("%s: %s for trans %s (continuing cleaning)", desc, err.Error(), trans) + } } - xapp.Logger.Info("SubDelReq timeout: Deleting trans record. SubId: %v, Xid: %s", subs.GetSubId(), trans.GetXid()) trans.Release() - if !c.registry.DelSubscription(subs.GetSubId()) { - xapp.Logger.Error("SubDelReq timeout: Failed to release sequency number. SubId: %v, Xid: %s", subs.GetSubId(), trans.GetXid()) - } - return + subs.Release() }