import (
"fmt"
"net/http"
- "os"
- "strconv"
- "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/xapp"
httptransport "github.com/go-openapi/runtime/client"
"github.com/go-openapi/strfmt"
- "github.com/gorilla/mux"
"github.com/segmentio/ksuid"
"github.com/spf13/viper"
)
var readSubsFromDb string
var dbRetryForever string
var dbTryCount int
+var e2IEOrderCheckValue uint8
type Control struct {
*xapp.RMRClient
e2SubsDb: CreateSdl(),
restSubsDb: CreateRESTSdl(),
Counters: xapp.Metric.RegisterCounterGroup(GetMetricsOpts(), "SUBMGR"),
- LoggerLevel: 4,
+ LoggerLevel: 1,
}
e2IfState.Init(c)
c.ReadConfigParameters("")
// Register REST handler for testing support
+ xapp.Resource.InjectRoute("/ric/v1/symptomdata", c.SymptomDataHandler, "GET")
xapp.Resource.InjectRoute("/ric/v1/test/{testId}", c.TestRestHandler, "POST")
xapp.Resource.InjectRoute("/ric/v1/restsubscriptions", c.GetAllRestSubscriptions, "GET")
- xapp.Resource.InjectRoute("/ric/v1/symptomdata", c.SymptomDataHandler, "GET")
- if readSubsFromDb == "false" {
- return c
- }
+ xapp.Resource.InjectRoute("/ric/v1/get_all_e2nodes", c.GetAllE2Nodes, "GET")
+ xapp.Resource.InjectRoute("/ric/v1/get_e2node_rest_subscriptions/{ranName}", c.GetAllE2NodeRestSubscriptions, "GET")
- // Read subscriptions from db
- c.ReadE2Subscriptions()
- c.ReadRESTSubscriptions()
+ xapp.Resource.InjectRoute("/ric/v1/get_all_xapps", c.GetAllXapps, "GET")
+ xapp.Resource.InjectRoute("/ric/v1/get_xapp_rest_restsubscriptions/{xappServiceName}", c.GetAllXappRestSubscriptions, "GET")
+ xapp.Resource.InjectRoute("/ric/v1/get_e2subscriptions/{restId}", c.GetE2Subscriptions, "GET")
- go xapp.Subscription.Listen(c.RESTSubscriptionHandler, c.RESTQueryHandler, c.RESTSubscriptionDeleteHandler)
+ xapp.Resource.InjectRoute("/ric/v1/delete_all_e2node_subscriptions/{ranName}", c.DeleteAllE2nodeSubscriptions, "DELETE")
+ xapp.Resource.InjectRoute("/ric/v1/delete_all_xapp_subscriptions/{xappServiceName}", c.DeleteAllXappSubscriptions, "DELETE")
+ if readSubsFromDb == "true" {
+ // Read subscriptions from db
+ err := c.ReadE2Subscriptions()
+ if err != nil {
+ xapp.Logger.Error("ReadE2Subscriptions() failed %s", err.Error())
+ }
+ err = c.ReadRESTSubscriptions()
+ if err != nil {
+ xapp.Logger.Error("ReadRESTSubscriptions() failed %s", err.Error())
+ }
+ }
+
+ go func() {
+ err := xapp.Subscription.Listen(c.RESTSubscriptionHandler, c.RESTQueryHandler, c.RESTSubscriptionDeleteHandler)
+ if err != nil {
+ xapp.Logger.Error("xapp.Subscription.Listen failure: %s", err.Error())
+ }
+ }()
return c
}
func (c *Control) SymptomDataHandler(w http.ResponseWriter, r *http.Request) {
- subscriptions, _ := c.registry.QueryHandler()
+ subscriptions, err := c.registry.QueryHandler()
+ if err != nil {
+ xapp.Logger.Error("QueryHandler() failed %s", err.Error())
+ }
+
xapp.Resource.SendSymptomDataJson(w, r, subscriptions, "platform/subscriptions.json")
}
//-------------------------------------------------------------------
//
//-------------------------------------------------------------------
-func (c *Control) GetAllRestSubscriptions(w http.ResponseWriter, r *http.Request) {
- xapp.Logger.Debug("GetAllRestSubscriptions() called")
- response := c.registry.GetAllRestSubscriptions()
- w.Write(response)
+func (c *Control) RESTQueryHandler() (models.SubscriptionList, error) {
+ xapp.Logger.Debug("RESTQueryHandler() called")
+
+ c.CntRecvMsg++
+
+ return c.registry.QueryHandler()
}
//-------------------------------------------------------------------
} else {
c.registry.subIds = subIds
c.registry.register = register
- c.HandleUncompletedSubscriptions(register)
+ go c.HandleUncompletedSubscriptions(register)
return nil
}
}
//
//-------------------------------------------------------------------
func (c *Control) ReadRESTSubscriptions() error {
+
+ xapp.Logger.Debug("ReadRESTSubscriptions()")
var err error
var restSubscriptions map[string]*RESTSubscription
for i := 0; dbRetryForever == "true" || i < dbTryCount; i++ {
xapp.Logger.Error("%v", err)
<-time.After(1 * time.Second)
} else {
+ // Fix REST subscriptions ongoing status after restart
+ for restSubId, restSubscription := range restSubscriptions {
+ restSubscription.SubReqOngoing = false
+ restSubscription.SubDelReqOngoing = false
+ err := c.WriteRESTSubscriptionToSdl(restSubId, restSubscription)
+ if err != nil {
+ xapp.Logger.Error("WriteRESTSubscriptionToSdl() failed:%s", err.Error())
+ }
+ }
c.registry.restSubscriptions = restSubscriptions
return nil
}
xapp.Logger.Debug("ReadConfigParameters")
c.LoggerLevel = int(xapp.Logger.GetLevel())
- xapp.Logger.Debug("LoggerLevel= %v", c.LoggerLevel)
+ xapp.Logger.Info("LoggerLevel = %v", c.LoggerLevel)
+ c.e2ap.SetASN1DebugPrintStatus(c.LoggerLevel)
// viper.GetDuration returns nanoseconds
e2tSubReqTimeout = viper.GetDuration("controls.e2tSubReqTimeout_ms") * 1000000
xapp.Logger.Debug("WARNING: Using hard coded default value for waitRouteCleanup_ms")
}
xapp.Logger.Debug("waitRouteCleanup= %v", waitRouteCleanup_ms)
+
+ viper.SetDefault("controls.checkE2IEOrder", 1)
+ e2IEOrderCheckValue = uint8(viper.GetUint("controls.checkE2IEOrder"))
+ c.e2ap.SetE2IEOrderCheck(e2IEOrderCheckValue)
+ xapp.Logger.Debug("e2IEOrderCheck= %v", e2IEOrderCheckValue)
}
//-------------------------------------------------------------------
if subs.PolicyUpdate == false {
subs.NoRespToXapp = true
xapp.Logger.Debug("SendSubscriptionDeleteReq. subId = %v", subId)
- c.SendSubscriptionDeleteReq(subs)
+ c.SendSubscriptionDeleteReq(subs, false)
}
}
}
//-------------------------------------------------------------------
//
//-------------------------------------------------------------------
-func (c *Control) GetOrCreateRestSubscription(p *models.SubscriptionParams, md5sum string, xAppRmrEndpoint string) (*RESTSubscription, string, error) {
+func (c *Control) GetOrCreateRestSubscription(p *models.SubscriptionParams, md5sum string, xAppRmrEndpoint string, xAppServiceName string) (*RESTSubscription, string, error) {
var restSubId string
var restSubscription *RESTSubscription
if restSubscription == nil {
restSubId = ksuid.New().String()
- restSubscription = c.registry.CreateRESTSubscription(&restSubId, &xAppRmrEndpoint, p.Meid)
+ restSubscription = c.registry.CreateRESTSubscription(&restSubId, &xAppServiceName, &xAppRmrEndpoint, p.Meid)
}
} else {
// Subscription contains REST subscription Id
c.PrintRESTSubscriptionRequest(p)
}
- if c.e2IfState.IsE2ConnectionUp(p.Meid) == false {
- xapp.Logger.Error("No E2 connection for ranName %v", *p.Meid)
+ if c.e2IfState.IsE2ConnectionUp(p.Meid) == false || c.e2IfState.IsE2ConnectionUnderReset(p.Meid) == true {
+ if c.e2IfState.IsE2ConnectionUp(p.Meid) == false {
+ xapp.Logger.Error("No E2 connection for ranName %v", *p.Meid)
+ } else if c.e2IfState.IsE2ConnectionUnderReset(p.Meid) == true {
+ xapp.Logger.Error("E2 Node for ranName %v UNDER RESET", *p.Meid)
+ }
c.UpdateCounter(cRestReqRejDueE2Down)
return nil, common.SubscribeServiceUnavailableCode
}
return nil, common.SubscribeBadRequestCode
}
+ e2SubscriptionDirectives, err := c.GetE2SubscriptionDirectives(p)
+ if err != nil {
+ xapp.Logger.Error("%s", err)
+ c.UpdateCounter(cRestSubFailToXapp)
+ return nil, common.SubscribeBadRequestCode
+ }
_, xAppRmrEndpoint, err := ConstructEndpointAddresses(*p.ClientEndpoint)
if err != nil {
xapp.Logger.Error("%s", err.Error())
xapp.Logger.Error("Failed to generate md5sum from incoming request - %s", err.Error())
}
- restSubscription, restSubId, err := c.GetOrCreateRestSubscription(p, md5sum, xAppRmrEndpoint)
+ restSubscription, restSubId, err := c.GetOrCreateRestSubscription(p, md5sum, xAppRmrEndpoint, p.ClientEndpoint.Host)
if err != nil {
xapp.Logger.Error("Subscription with id in REST request does not exist")
return nil, common.SubscribeNotFoundCode
if duplicate {
err := fmt.Errorf("Retransmission blocker direct ACK for request of restSubsId %s restSubId MD5sum %s as retransmission", restSubId, md5sum)
xapp.Logger.Debug("%s", err)
+ c.registry.DeleteRESTSubscription(&restSubId)
c.UpdateCounter(cRestSubRespToXapp)
return &subResp, common.SubscribeCreatedCode
}
c.WriteRESTSubscriptionToDb(restSubId, restSubscription)
- e2SubscriptionDirectives, err := c.GetE2SubscriptionDirectives(p)
- if err != nil {
- xapp.Logger.Error("%s", err)
- return nil, common.SubscribeBadRequestCode
- }
go c.processSubscriptionRequests(restSubscription, &subReqList, p.ClientEndpoint, p.Meid, &restSubId, xAppRmrEndpoint, md5sum, e2SubscriptionDirectives)
c.UpdateCounter(cRestSubRespToXapp)
clientEndpoint *models.SubscriptionParamsClientEndpoint, meid *string, restSubId *string, xAppRmrEndpoint string, md5sum string, e2SubscriptionDirectives *E2SubscriptionDirectives) {
c.SubscriptionProcessingStartDelay()
- xapp.Logger.Debug("Subscription Request count=%v ", len(subReqList.E2APSubscriptionRequests))
+ xapp.Logger.Debug("E2 SubscriptionRequest count = %v ", len(subReqList.E2APSubscriptionRequests))
var xAppEventInstanceID int64
var e2EventInstanceID int64
trans.Release()
if err != nil {
+ if err.Error() == "TEST: restart event received" {
+ // This is just for UT cases. Stop here subscription processing
+ return
+ }
c.sendUnsuccesfullResponseNotification(restSubId, restSubscription, xAppEventInstanceID, err, clientEndpoint, trans, errorInfo)
} else {
e2EventInstanceID = (int64)(subRespMsg.RequestId.InstanceId)
restSubscription.AddMd5Sum(md5sum)
xapp.Logger.Debug("SubscriptionRequest index=%v processed successfullyfor %s. endpoint=%v:%v, XappEventInstanceID=%v, E2EventInstanceID=%v, %s",
index, *restSubId, clientEndpoint.Host, *clientEndpoint.HTTPPort, xAppEventInstanceID, e2EventInstanceID, idstring(nil, trans))
- c.sendSuccesfullResponseNotification(restSubId, restSubscription, xAppEventInstanceID, e2EventInstanceID, clientEndpoint, trans)
+ c.sendSuccesfullResponseNotification(restSubId, restSubscription, xAppEventInstanceID, e2EventInstanceID, clientEndpoint, trans, errorInfo)
}
}
}
// Wake subs request
//
subs.OngoingReqCount++
- go c.handleSubscriptionCreate(subs, trans, e2SubscriptionDirectives)
+ go c.handleSubscriptionCreate(subs, trans, e2SubscriptionDirectives, 0)
event, _ := trans.WaitEvent(0) //blocked wait as timeout is handled in subs side
subs.OngoingReqCount--
case *e2ap.E2APSubscriptionResponse:
trans.Release()
if c.e2IfState.IsE2ConnectionUp(meid) == true {
+ errorInfo = c.e2ap.CheckActionNotAdmittedList(xapp.RIC_SUB_RESP, themsg.ActionNotAdmittedList, c)
return themsg, &errorInfo, nil
} else {
c.registry.RemoveFromSubscription(subs, trans, waitRouteCleanup_ms, c)
c.RemoveSubscriptionFromDb(subs)
err = fmt.Errorf("E2 interface down")
errorInfo.SetInfo(err.Error(), models.SubscriptionInstanceErrorSourceE2Node, "")
- return nil, &errorInfo, err
}
case *e2ap.E2APSubscriptionFailure:
- err = fmt.Errorf("E2 SubscriptionFailure received")
+ err = fmt.Errorf("RICSubscriptionFailure. E2NodeCause: (Cause:%v, Value %v)", themsg.Cause.Content, themsg.Cause.Value)
errorInfo.SetInfo(err.Error(), models.SubscriptionInstanceErrorSourceE2Node, "")
- return nil, &errorInfo, err
case *PackSubscriptionRequestErrortEvent:
- err = fmt.Errorf("E2 SubscriptionRequest pack failure")
- return nil, &themsg.ErrorInfo, err
+ err = fmt.Errorf("E2 RICSubscriptionRequest pack failure")
+ errorInfo = themsg.ErrorInfo
case *SDLWriteErrortEvent:
err = fmt.Errorf("SDL write failure")
- return nil, &themsg.ErrorInfo, err
+ errorInfo = themsg.ErrorInfo
+ case *SubmgrRestartTestEvent:
+ err = fmt.Errorf("TEST: restart event received")
+ xapp.Logger.Debug("%s", err)
+ return nil, &errorInfo, err
default:
err = fmt.Errorf("Unexpected E2 subscription response received")
errorInfo.SetInfo(err.Error(), models.SubscriptionInstanceErrorSourceE2Node, "")
break
}
} else {
- err = fmt.Errorf("E2 subscription response timeout")
+ // Timer expiry
+ err = fmt.Errorf("E2 RICSubscriptionResponse timeout")
errorInfo.SetInfo(err.Error(), "", models.SubscriptionInstanceTimeoutTypeE2Timeout)
if subs.PolicyUpdate == true {
return nil, &errorInfo, err
}
}
- xapp.Logger.Error("XAPP-SubReq E2 subscription failed %s", idstring(err, trans, subs))
- c.registry.RemoveFromSubscription(subs, trans, waitRouteCleanup_ms, c)
+ xapp.Logger.Error("XAPP-SubReq E2 subscription failed: %s", idstring(err, trans, subs))
+ // If policy type subscription fails we cannot remove it only internally. Once subscription has been created
+ // successfully, it must be deleted on both sides.
+ if subs.PolicyUpdate == false {
+ c.registry.RemoveFromSubscription(subs, trans, waitRouteCleanup_ms, c)
+ }
+
return nil, &errorInfo, err
}
restSubscription.SetProcessed(err)
c.UpdateRESTSubscriptionInDB(*restSubId, restSubscription, false)
if trans != nil {
- xapp.Logger.Debug("Sending unsuccessful REST notification (cause %s) to endpoint=%v:%v, XappEventInstanceID=%v, E2EventInstanceID=%v, %s",
- errorInfo.ErrorCause, clientEndpoint.Host, *clientEndpoint.HTTPPort, xAppEventInstanceID, e2EventInstanceID, idstring(nil, trans))
+ xapp.Logger.Debug("Sending unsuccessful REST notification: ErrorCause:%s, ErrorSource:%s, TimeoutType:%s, to Endpoint=%v:%v, XappEventInstanceID=%v, E2EventInstanceID=%v, %s",
+ errorInfo.ErrorCause, errorInfo.ErrorSource, errorInfo.TimeoutType, clientEndpoint.Host, *clientEndpoint.HTTPPort, xAppEventInstanceID, e2EventInstanceID, idstring(nil, trans))
} else {
- xapp.Logger.Debug("Sending unsuccessful REST notification (cause %s) to endpoint=%v:%v, XappEventInstanceID=%v, E2EventInstanceID=%v",
- errorInfo.ErrorCause, clientEndpoint.Host, *clientEndpoint.HTTPPort, xAppEventInstanceID, e2EventInstanceID)
+ xapp.Logger.Debug("Sending unsuccessful REST notification: ErrorCause:%s, ErrorSource:%s, TimeoutType:%s, to Endpoint=%v:%v, XappEventInstanceID=%v, E2EventInstanceID=%v",
+ errorInfo.ErrorCause, errorInfo.ErrorSource, errorInfo.TimeoutType, clientEndpoint.Host, *clientEndpoint.HTTPPort, xAppEventInstanceID, e2EventInstanceID)
}
c.UpdateCounter(cRestSubFailNotifToXapp)
- xapp.Subscription.Notify(resp, *clientEndpoint)
+ err = xapp.Subscription.Notify(resp, *clientEndpoint)
+ if err != nil {
+ xapp.Logger.Error("xapp.Subscription.Notify failed %s", err.Error())
+ }
+ // E2 is down. Delete completely processed request safely now
if c.e2IfState.IsE2ConnectionUp(&restSubscription.Meid) == false && restSubscription.SubReqOngoing == false {
c.registry.DeleteRESTSubscription(restSubId)
c.RemoveRESTSubscriptionFromDb(*restSubId)
//
//-------------------------------------------------------------------
func (c *Control) sendSuccesfullResponseNotification(restSubId *string, restSubscription *RESTSubscription, xAppEventInstanceID int64, e2EventInstanceID int64,
- clientEndpoint *models.SubscriptionParamsClientEndpoint, trans *TransactionXapp) {
+ clientEndpoint *models.SubscriptionParamsClientEndpoint, trans *TransactionXapp, errorInfo *ErrorInfo) {
// Store successfully processed InstanceId for deletion
restSubscription.AddE2InstanceId((uint32)(e2EventInstanceID))
SubscriptionID: restSubId,
SubscriptionInstances: []*models.SubscriptionInstance{
&models.SubscriptionInstance{E2EventInstanceID: &e2EventInstanceID,
- ErrorCause: "",
+ ErrorCause: errorInfo.ErrorCause,
+ ErrorSource: errorInfo.ErrorSource,
XappEventInstanceID: &xAppEventInstanceID},
},
}
// Mark REST subscription request processesd.
restSubscription.SetProcessed(nil)
c.UpdateRESTSubscriptionInDB(*restSubId, restSubscription, false)
- xapp.Logger.Debug("Sending successful REST notification to endpoint=%v:%v, XappEventInstanceID=%v, E2EventInstanceID=%v, %s",
- clientEndpoint.Host, *clientEndpoint.HTTPPort, xAppEventInstanceID, e2EventInstanceID, idstring(nil, trans))
-
+ xapp.Logger.Debug("Sending successful REST notification: ErrorCause:%s, ErrorSource:%s, TimeoutType:%s, to Endpoint=%v:%v, XappEventInstanceID=%v, E2EventInstanceID=%v, %s",
+ errorInfo.ErrorCause, errorInfo.ErrorSource, errorInfo.TimeoutType, clientEndpoint.Host, *clientEndpoint.HTTPPort, xAppEventInstanceID, e2EventInstanceID, idstring(nil, trans))
c.UpdateCounter(cRestSubNotifToXapp)
- xapp.Subscription.Notify(resp, *clientEndpoint)
+ err := xapp.Subscription.Notify(resp, *clientEndpoint)
+ if err != nil {
+ xapp.Logger.Error("xapp.Subscription.Notify failed %s", err.Error())
+ }
+ // E2 is down. Delete completely processed request safely now
if c.e2IfState.IsE2ConnectionUp(&restSubscription.Meid) == false && restSubscription.SubReqOngoing == false {
c.registry.DeleteRESTSubscription(restSubId)
c.RemoveRESTSubscriptionFromDb(*restSubId)
return common.UnsubscribeBadRequestCode
} else if restSubscription.SubDelReqOngoing == true {
// Previous request for same restSubId still ongoing
- c.UpdateCounter(cRestSubDelFailToXapp)
- return common.UnsubscribeBadRequestCode
+ c.UpdateCounter(cRestSubDelRespToXapp)
+ return common.UnsubscribeNoContentCode
}
}
}
go func() {
xapp.Logger.Debug("Deleteting handler: processing instances = %v", restSubscription.InstanceIds)
for _, instanceId := range restSubscription.InstanceIds {
- xAppEventInstanceID, err := c.SubscriptionDeleteHandler(&restSubId, &xAppRmrEndPoint, &restSubscription.Meid, instanceId)
+ xAppEventInstanceID, err := c.SubscriptionDeleteHandler(&restSubId, &xAppRmrEndPoint, &restSubscription.Meid, instanceId, 0)
if err != nil {
xapp.Logger.Error("%s", err.Error())
//-------------------------------------------------------------------
//
//-------------------------------------------------------------------
-func (c *Control) SubscriptionDeleteHandler(restSubId *string, endPoint *string, meid *string, instanceId uint32) (int64, error) {
+func (c *Control) SubscriptionDeleteHandler(restSubId *string, endPoint *string, meid *string, instanceId uint32, waitRouteCleanupTime time.Duration) (int64, error) {
var xAppEventInstanceID int64
subs, err := c.registry.GetSubscriptionFirstMatch([]uint32{instanceId})
// Wake subs delete
//
subs.OngoingDelCount++
- go c.handleSubscriptionDelete(subs, trans)
+ go c.handleSubscriptionDelete(subs, trans, waitRouteCleanupTime)
trans.WaitEvent(0) //blocked wait as timeout is handled in subs side
subs.OngoingDelCount--
return xAppEventInstanceID, nil
}
-//-------------------------------------------------------------------
-//
-//-------------------------------------------------------------------
-func (c *Control) RESTQueryHandler() (models.SubscriptionList, error) {
- xapp.Logger.Debug("RESTQueryHandler() called")
-
- c.CntRecvMsg++
-
- return c.registry.QueryHandler()
-}
-
-func (c *Control) TestRestHandler(w http.ResponseWriter, r *http.Request) {
- xapp.Logger.Debug("RESTTestRestHandler() called")
-
- pathParams := mux.Vars(r)
- s := pathParams["testId"]
-
- // This can be used to delete single subscription from db
- if contains := strings.Contains(s, "deletesubid="); contains == true {
- var splits = strings.Split(s, "=")
- if subId, err := strconv.ParseInt(splits[1], 10, 64); err == nil {
- xapp.Logger.Debug("RemoveSubscriptionFromSdl() called. subId = %v", subId)
- c.RemoveSubscriptionFromSdl(uint32(subId))
- return
- }
- }
-
- // This can be used to remove all subscriptions db from
- if s == "emptydb" {
- xapp.Logger.Debug("RemoveAllSubscriptionsFromSdl() called")
- c.RemoveAllSubscriptionsFromSdl()
- c.RemoveAllRESTSubscriptionsFromSdl()
- return
- }
-
- // This is meant to cause submgr's restart in testing
- if s == "restart" {
- xapp.Logger.Debug("os.Exit(1) called")
- os.Exit(1)
- }
-
- xapp.Logger.Debug("Unsupported rest command received %s", s)
-}
-
//-------------------------------------------------------------------
//
//-------------------------------------------------------------------
go c.handleE2TSubscriptionDeleteResponse(msg)
case xapp.RIC_SUB_DEL_FAILURE:
go c.handleE2TSubscriptionDeleteFailure(msg)
+ case xapp.RIC_SUB_DEL_REQUIRED:
+ go c.handleE2TSubscriptionDeleteRequired(msg)
default:
xapp.Logger.Debug("Unknown Message Type '%d', discarding", msg.Mtype)
}
return
}
- //TODO handle subscription toward e2term inside AssignToSubscription / hide handleSubscriptionCreate in it?
subs, _, err := c.registry.AssignToSubscription(trans, subReqMsg, c.ResetTestFlag, c, true)
if err != nil {
xapp.Logger.Error("XAPP-SubReq: %s", idstring(err, trans))
//------------------------------------------------------------------
func (c *Control) wakeSubscriptionRequest(subs *Subscription, trans *TransactionXapp) {
- e2SubscriptionDirectives, _ := c.GetE2SubscriptionDirectives(nil)
+ e2SubscriptionDirectives, err := c.GetE2SubscriptionDirectives(nil)
+ if err != nil {
+ xapp.Logger.Error("c.GetE2SubscriptionDirectives failure: %s", err.Error())
+ }
subs.OngoingReqCount++
- go c.handleSubscriptionCreate(subs, trans, e2SubscriptionDirectives)
+ go c.handleSubscriptionCreate(subs, trans, e2SubscriptionDirectives, waitRouteCleanup_ms)
event, _ := trans.WaitEvent(0) //blocked wait as timeout is handled in subs side
subs.OngoingReqCount--
- var err error
if event != nil {
switch themsg := event.(type) {
case *e2ap.E2APSubscriptionResponse:
if err == nil {
trans.Release()
c.UpdateCounter(cSubRespToXapp)
- c.rmrSendToXapp("", subs, trans)
+ err := c.rmrSendToXapp("", subs, trans)
+ if err != nil {
+ xapp.Logger.Error("rmrSendToXapp() failed:%s", err.Error())
+ }
return
}
case *e2ap.E2APSubscriptionFailure:
}
}
xapp.Logger.Debug("XAPP-SubReq: failed %s", idstring(err, trans, subs))
- //c.registry.RemoveFromSubscription(subs, trans, 5*time.Second)
}
//-------------------------------------------------------------------
// Wake subs delete
//
subs.OngoingDelCount++
- go c.handleSubscriptionDelete(subs, trans)
+ go c.handleSubscriptionDelete(subs, trans, waitRouteCleanup_ms)
trans.WaitEvent(0) //blocked wait as timeout is handled in subs side
subs.OngoingDelCount--
trans.Mtype, trans.Payload, err = c.e2ap.PackSubscriptionDeleteResponse(subDelRespMsg)
if err == nil {
c.UpdateCounter(cSubDelRespToXapp)
- c.rmrSendToXapp("", subs, trans)
+ err := c.rmrSendToXapp("", subs, trans)
+ if err != nil {
+ xapp.Logger.Error("rmrSendToXapp() failed:%s", err.Error())
+ }
}
-
- //TODO handle subscription toward e2term insiged RemoveFromSubscription / hide handleSubscriptionDelete in it?
- //c.registry.RemoveFromSubscription(subs, trans, 5*time.Second)
}
//-------------------------------------------------------------------
// SUBS CREATE Handling
//-------------------------------------------------------------------
-func (c *Control) handleSubscriptionCreate(subs *Subscription, parentTrans *TransactionXapp, e2SubscriptionDirectives *E2SubscriptionDirectives) {
+func (c *Control) handleSubscriptionCreate(subs *Subscription, parentTrans *TransactionXapp, e2SubscriptionDirectives *E2SubscriptionDirectives, waitRouteCleanupTime time.Duration) {
var event interface{} = nil
var removeSubscriptionFromDb bool = false
subRfMsg, valid = subs.SetCachedResponse(event, true)
subs.SubRespRcvd = true
case *e2ap.E2APSubscriptionFailure:
- removeSubscriptionFromDb = true
- subRfMsg, valid = subs.SetCachedResponse(event, false)
+ if subs.PolicyUpdate == false {
+ subRfMsg, valid = subs.SetCachedResponse(event, false)
+ } else {
+ // In policy update case where subscription has already been created successfully in Gnb
+ // we cannot delete subscription internally in submgr
+ subRfMsg, valid = subs.SetCachedResponse(event, true)
+ }
xapp.Logger.Debug("SUBS-SubReq: internal delete due failure event(%s) %s", typeofSubsMessage(event), idstring(nil, trans, subs, parentTrans))
case *SubmgrRestartTestEvent:
- // This simulates that no response has been received and after restart subscriptions are restored from db
+ // This is used to simulate that no response has been received and after restart, subscriptions are restored from db
xapp.Logger.Debug("Test restart flag is active. Dropping this transaction to test restart case")
+ subRfMsg, valid = subs.SetCachedResponse(event, false)
+ parentTrans.SendEvent(subRfMsg, 0)
+ return
case *PackSubscriptionRequestErrortEvent, *SDLWriteErrortEvent:
subRfMsg, valid = subs.SetCachedResponse(event, false)
default:
+ // Timer expiry
if subs.PolicyUpdate == false {
xapp.Logger.Debug("SUBS-SubReq: internal delete due default event(%s) %s", typeofSubsMessage(event), idstring(nil, trans, subs, parentTrans))
- removeSubscriptionFromDb = true
subRfMsg, valid = subs.SetCachedResponse(nil, false)
c.sendE2TSubscriptionDeleteRequest(subs, trans, parentTrans)
+ } else {
+ subRfMsg, valid = subs.SetCachedResponse(nil, true)
}
}
xapp.Logger.Debug("SUBS-SubReq: Handling (e2t response %s) %s", typeofSubsMessage(subRfMsg), idstring(nil, trans, subs, parentTrans))
} else {
xapp.Logger.Debug("SUBS-SubReq: Handling (cached response %s) %s", typeofSubsMessage(subRfMsg), idstring(nil, trans, subs, parentTrans))
}
+ xapp.Logger.Debug("subs.PolicyUpdate: %v", subs.PolicyUpdate)
+ xapp.Logger.Debug("subs: %v", subs)
+
+ if valid == false {
+ removeSubscriptionFromDb = true
+ }
err := c.UpdateSubscriptionInDB(subs, removeSubscriptionFromDb)
if err != nil {
- subRfMsg, valid = subs.SetCachedResponse(event, false)
+ valid = false
c.sendE2TSubscriptionDeleteRequest(subs, trans, parentTrans)
+
}
- //Now RemoveFromSubscription in here to avoid race conditions (mostly concerns delete)
+ // Now RemoveFromSubscription in here to avoid race conditions (mostly concerns delete)
if valid == false {
- c.registry.RemoveFromSubscription(subs, parentTrans, waitRouteCleanup_ms, c)
+ c.registry.RemoveFromSubscription(subs, parentTrans, waitRouteCleanupTime, c)
}
parentTrans.SendEvent(subRfMsg, 0)
// SUBS DELETE Handling
//-------------------------------------------------------------------
-func (c *Control) handleSubscriptionDelete(subs *Subscription, parentTrans *TransactionXapp) {
+func (c *Control) handleSubscriptionDelete(subs *Subscription, parentTrans *TransactionXapp, waitRouteCleanupTime time.Duration) {
trans := c.tracker.NewSubsTransaction(subs)
subs.WaitTransactionTurn(trans)
} else {
subs.mutex.Unlock()
}
- //Now RemoveFromSubscription in here to avoid race conditions (mostly concerns delete)
- // If parallel deletes ongoing both might pass earlier sendE2TSubscriptionDeleteRequest(...) if
- // RemoveFromSubscription locates in caller side (now in handleXAPPSubscriptionDeleteRequest(...))
- c.registry.RemoveFromSubscription(subs, parentTrans, waitRouteCleanup_ms, c)
- c.registry.UpdateSubscriptionToDb(subs, c)
+
+ // Now RemoveFromSubscription in here to avoid race conditions (mostly concerns delete)
+ c.registry.RemoveFromSubscription(subs, parentTrans, waitRouteCleanupTime, c)
parentTrans.SendEvent(nil, 0)
}
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))
+ xapp.Logger.Error("SUBS-SubReq ASN1 pack error: %s", idstring(err, trans, subs, parentTrans))
return &PackSubscriptionRequestErrortEvent{
ErrorInfo{
ErrorSource: models.SubscriptionInstanceErrorSourceASN1,
} else {
c.UpdateCounter(cSubReReqToE2)
}
- c.rmrSendToE2T(desc, subs, trans)
+ err := c.rmrSendToE2T(desc, subs, trans)
+ if err != nil {
+ xapp.Logger.Error("rmrSendToE2T() failed:%s", err.Error())
+ }
+
if subs.DoNotWaitSubResp == false {
event, timedOut = trans.WaitEvent(e2SubscriptionDirectives.E2TimeoutTimerValue)
if timedOut {
} else {
c.UpdateCounter(cSubDelReReqToE2)
}
- c.rmrSendToE2T(desc, subs, trans)
+ err := c.rmrSendToE2T(desc, subs, trans)
+ if err != nil {
+ xapp.Logger.Error("SUBS-SubDelReq: rmrSendToE2T failure: %s", idstring(err, trans, subs, parentTrans))
+ }
event, timedOut = trans.WaitEvent(e2tSubDelReqTime)
if timedOut {
c.UpdateCounter(cSubDelReqTimerExpiry)
xapp.Logger.Error("MSG-SubResp: %s", idstring(err, params, subs))
return
}
+ xapp.Logger.Debug("SUBS-SubResp: Sending event, trans= %v", trans)
sendOk, timedOut := trans.SendEvent(subRespMsg, e2tRecvMsgTimeout)
if sendOk == false {
err = fmt.Errorf("Passing event to transaction failed: sendOk(%t) timedOut(%t)", sendOk, timedOut)
//-------------------------------------------------------------------
// handle from E2T Subscription Delete Response
//-------------------------------------------------------------------
-func (c *Control) handleE2TSubscriptionDeleteResponse(params *xapp.RMRParams) (err error) {
+func (c *Control) handleE2TSubscriptionDeleteResponse(params *xapp.RMRParams) {
xapp.Logger.Debug("MSG from E2T: %s", params.String())
c.UpdateCounter(cSubDelRespFromE2)
subDelRespMsg, err := c.e2ap.UnpackSubscriptionDeleteResponse(params.Payload)
}
}
-func (c *Control) SendSubscriptionDeleteReq(subs *Subscription) {
+func (c *Control) SendSubscriptionDeleteReq(subs *Subscription, e2SubsDelRequired bool) {
+
+ if c.UTTesting == true {
+ // Reqistry mutex is not locked after real restart but it can be when restart is simulated in unit tests
+ c.registry.mutex = new(sync.Mutex)
+ }
const ricRequestorId = 123
xapp.Logger.Debug("Sending subscription delete due to restart. subId = %v", subs.ReqId.InstanceId)
params.Payload = payload.Buf
params.Mbuf = nil
subs.DeleteFromDb = true
- c.handleXAPPSubscriptionDeleteRequest(params)
+ if !e2SubsDelRequired {
+ c.handleXAPPSubscriptionDeleteRequest(params)
+ } else {
+ c.SendSubscriptionDeleteReqToE2T(subs, params)
+ }
}
}
}
}
}
}
+
+//-------------------------------------------------------------------
+// handle from E2T Subscription Delete Required
+//-------------------------------------------------------------------
+func (c *Control) handleE2TSubscriptionDeleteRequired(params *xapp.RMRParams) {
+ xapp.Logger.Info("MSG from E2T: %s", params.String())
+ c.UpdateCounter(cSubDelRequFromE2)
+ subsDelRequMsg, err := c.e2ap.UnpackSubscriptionDeleteRequired(params.Payload)
+ if err != nil {
+ xapp.Logger.Error("MSG-SubDelRequired: %s", idstring(err, params))
+ //c.sendE2TErrorIndication(nil)
+ return
+ }
+ var subscriptions = map[string][]e2ap.E2APSubscriptionDeleteRequired{}
+ var subDB = []*Subscription{}
+ for _, subsTobeRemove := range subsDelRequMsg.E2APSubscriptionDeleteRequiredRequests {
+ subs, err := c.registry.GetSubscriptionFirstMatch([]uint32{subsTobeRemove.RequestId.InstanceId})
+ if err != nil {
+ xapp.Logger.Error("MSG-SubDelFail: %s", idstring(err, params))
+ continue
+ }
+ // Check if Delete Subscription Already triggered
+ if subs.OngoingDelCount > 0 {
+ continue
+ }
+ subDB = append(subDB, subs)
+ for _, endpoint := range subs.EpList.Endpoints {
+ subscriptions[endpoint.Addr] = append(subscriptions[endpoint.Addr], subsTobeRemove)
+ }
+ // Sending Subscription Delete Request to E2T
+ // c.SendSubscriptionDeleteReq(subs, true)
+ }
+ for _, subsTobeRemove := range subDB {
+ // Sending Subscription Delete Request to E2T
+ c.SendSubscriptionDeleteReq(subsTobeRemove, true)
+ }
+}
+
+//-----------------------------------------------------------------
+// Initiate RIC Subscription Delete Request after receiving
+// RIC Subscription Delete Required from E2T
+//-----------------------------------------------------------------
+func (c *Control) SendSubscriptionDeleteReqToE2T(subs *Subscription, params *xapp.RMRParams) {
+ xapp.Logger.Debug("MSG TO E2T: %s", params.String())
+ c.UpdateCounter(cSubDelReqToE2)
+
+ if c.e2IfState.IsE2ConnectionUp(¶ms.Meid.RanName) == false {
+ xapp.Logger.Error("No E2 connection for ranName %v", params.Meid.RanName)
+ return
+ }
+
+ trans := c.tracker.NewXappTransaction(xapp.NewRmrEndpoint(params.Src), params.Xid, subs.ReqId.RequestId, params.Meid)
+ if trans == nil {
+ xapp.Logger.Error("XAPP-SubDelReq: %s", idstring(fmt.Errorf("transaction not created"), params))
+ return
+ }
+ defer trans.Release()
+
+ err := c.tracker.Track(trans)
+ if err != nil {
+ xapp.Logger.Error("XAPP-SubReq: %s", idstring(err, trans))
+ return
+ }
+
+ //
+ // Wake subs delete
+ //
+ subs.OngoingDelCount++
+ go c.handleSubscriptionDelete(subs, trans, waitRouteCleanup_ms)
+ trans.WaitEvent(0) //blocked wait as timeout is handled in subs side
+ subs.OngoingDelCount--
+
+ xapp.Logger.Debug("XAPP-SubDelReq: Handling event %s ", idstring(nil, trans, subs))
+
+ if subs.NoRespToXapp == true {
+ // Do no send delete responses to xapps due to submgr restart is deleting uncompleted subscriptions
+ xapp.Logger.Debug("XAPP-SubDelReq: subs.NoRespToXapp == true")
+ return
+ }
+}
Code Review / ric-plt / submgr.git / blobdiff