X-Git-Url: https://gerrit.o-ran-sc.org/r/gitweb?a=blobdiff_plain;f=pkg%2Fcontrol%2Fcontrol.go;h=27b2157e3ff8c14fd430a6a7f55ce3521e594f9a;hb=d00f25ffa52842524e52e161429ec4795574155a;hp=5ab86c002966e6248fe4cd7f2d03e16fa3d06c58;hpb=27bba84fdbebf215a9ebabe3b7e2a31cb579e669;p=ric-plt%2Fsubmgr.git diff --git a/pkg/control/control.go b/pkg/control/control.go index 5ab86c0..27b2157 100755 --- a/pkg/control/control.go +++ b/pkg/control/control.go @@ -71,19 +71,22 @@ var waitRouteCleanup_ms time.Duration var e2tMaxSubReqTryCount uint64 // Initial try + retry var e2tMaxSubDelReqTryCount uint64 // Initial try + retry var readSubsFromDb string -var restDuplicateCtrl duplicateCtrl +var dbRetryForever string +var dbTryCount int type Control struct { *xapp.RMRClient - e2ap *E2ap - registry *Registry - tracker *Tracker - e2SubsDb Sdlnterface - restSubsDb Sdlnterface - CntRecvMsg uint64 - ResetTestFlag bool - Counters map[string]xapp.Counter - LoggerLevel uint32 + e2ap *E2ap + registry *Registry + tracker *Tracker + restDuplicateCtrl *DuplicateCtrl + e2SubsDb Sdlnterface + restSubsDb Sdlnterface + CntRecvMsg uint64 + ResetTestFlag bool + Counters map[string]xapp.Counter + LoggerLevel uint32 + UTTesting bool } type RMRMeid struct { @@ -114,45 +117,35 @@ func NewControl() *Control { tracker := new(Tracker) tracker.Init() + restDuplicateCtrl := new(DuplicateCtrl) + restDuplicateCtrl.Init() + c := &Control{e2ap: new(E2ap), - registry: registry, - tracker: tracker, - e2SubsDb: CreateSdl(), - restSubsDb: CreateRESTSdl(), - Counters: xapp.Metric.RegisterCounterGroup(GetMetricsOpts(), "SUBMGR"), - LoggerLevel: 3, + registry: registry, + tracker: tracker, + restDuplicateCtrl: restDuplicateCtrl, + e2SubsDb: CreateSdl(), + restSubsDb: CreateRESTSdl(), + Counters: xapp.Metric.RegisterCounterGroup(GetMetricsOpts(), "SUBMGR"), + LoggerLevel: 4, } c.ReadConfigParameters("") // Register REST handler for testing support 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") - go xapp.Subscription.Listen(c.SubscriptionHandler, c.QueryHandler, c.SubscriptionDeleteHandlerCB) - if readSubsFromDb == "false" { return c } - restDuplicateCtrl.Init() - // Read subscriptions from db - xapp.Logger.Info("Reading subscriptions from db") - subIds, register, err := c.ReadAllSubscriptionsFromSdl() - if err != nil { - xapp.Logger.Error("%v", err) - } else { - c.registry.subIds = subIds - c.registry.register = register - c.HandleUncompletedSubscriptions(register) - } + c.ReadE2Subscriptions() + c.ReadRESTSubscriptions() + + go xapp.Subscription.Listen(c.SubscriptionHandler, c.QueryHandler, c.SubscriptionDeleteHandlerCB) - restSubscriptions, err := c.ReadAllRESTSubscriptionsFromSdl() - if err != nil { - xapp.Logger.Error("%v", err) - } else { - c.registry.restSubscriptions = restSubscriptions - } return c } @@ -161,6 +154,60 @@ func (c *Control) SymptomDataHandler(w http.ResponseWriter, r *http.Request) { xapp.Resource.SendSymptomDataJson(w, r, subscriptions, "platform/subscriptions.json") } +//------------------------------------------------------------------- +// +//------------------------------------------------------------------- +func (c *Control) GetAllRestSubscriptions(w http.ResponseWriter, r *http.Request) { + xapp.Logger.Info("GetAllRestSubscriptions() called") + response := c.registry.GetAllRestSubscriptions() + w.Write(response) +} + +//------------------------------------------------------------------- +// +//------------------------------------------------------------------- +func (c *Control) ReadE2Subscriptions() error { + var err error + var subIds []uint32 + var register map[uint32]*Subscription + for i := 0; dbRetryForever == "true" || i < dbTryCount; i++ { + xapp.Logger.Info("Reading E2 subscriptions from db") + subIds, register, err = c.ReadAllSubscriptionsFromSdl() + if err != nil { + xapp.Logger.Error("%v", err) + <-time.After(1 * time.Second) + } else { + c.registry.subIds = subIds + c.registry.register = register + c.HandleUncompletedSubscriptions(register) + return nil + } + } + xapp.Logger.Info("Continuing without retring") + return err +} + +//------------------------------------------------------------------- +// +//------------------------------------------------------------------- +func (c *Control) ReadRESTSubscriptions() error { + var err error + var restSubscriptions map[string]*RESTSubscription + for i := 0; dbRetryForever == "true" || i < dbTryCount; i++ { + xapp.Logger.Info("Reading REST subscriptions from db") + restSubscriptions, err = c.ReadAllRESTSubscriptionsFromSdl() + if err != nil { + xapp.Logger.Error("%v", err) + <-time.After(1 * time.Second) + } else { + c.registry.restSubscriptions = restSubscriptions + return nil + } + } + xapp.Logger.Info("Continuing without retring") + return err +} + //------------------------------------------------------------------- // //------------------------------------------------------------------- @@ -183,14 +230,6 @@ func (c *Control) ReadConfigParameters(f string) { } xapp.Logger.Info("e2tRecvMsgTimeout %v", e2tRecvMsgTimeout) - // Internal cfg parameter, used to define a wait time for RMR route clean-up. None default - // value 100ms used currently only in unittests. - waitRouteCleanup_ms = viper.GetDuration("controls.waitRouteCleanup_ms") * 1000000 - if waitRouteCleanup_ms == 0 { - waitRouteCleanup_ms = 5000 * 1000000 - } - xapp.Logger.Info("waitRouteCleanup %v", waitRouteCleanup_ms) - e2tMaxSubReqTryCount = viper.GetUint64("controls.e2tMaxSubReqTryCount") if e2tMaxSubReqTryCount == 0 { e2tMaxSubReqTryCount = 1 @@ -208,10 +247,32 @@ func (c *Control) ReadConfigParameters(f string) { readSubsFromDb = "true" } xapp.Logger.Info("readSubsFromDb %v", readSubsFromDb) + + dbTryCount = viper.GetInt("controls.dbTryCount") + if dbTryCount == 0 { + dbTryCount = 200 + } + xapp.Logger.Info("dbTryCount %v", dbTryCount) + + dbRetryForever = viper.GetString("controls.dbRetryForever") + if dbRetryForever == "" { + dbRetryForever = "true" + } + xapp.Logger.Info("dbRetryForever %v", dbRetryForever) + c.LoggerLevel = viper.GetUint32("logger.level") if c.LoggerLevel == 0 { c.LoggerLevel = 3 } + xapp.Logger.Info("LoggerLevel %v", c.LoggerLevel) + + // Internal cfg parameter, used to define a wait time for RMR route clean-up. None default + // value 100ms used currently only in unittests. + waitRouteCleanup_ms = viper.GetDuration("controls.waitRouteCleanup_ms") * 1000000 + if waitRouteCleanup_ms == 0 { + waitRouteCleanup_ms = 5000 * 1000000 + } + xapp.Logger.Info("waitRouteCleanup %v", waitRouteCleanup_ms) } //------------------------------------------------------------------- @@ -250,7 +311,7 @@ func (c *Control) GetOrCreateRestSubscription(p *models.SubscriptionParams, md5s var restSubscription *RESTSubscription var err error - prevRestSubsId, exists := restDuplicateCtrl.GetLastKnownRestSubsIdBasedOnMd5sum(md5sum) + prevRestSubsId, exists := c.restDuplicateCtrl.GetLastKnownRestSubsIdBasedOnMd5sum(md5sum) if p.SubscriptionID == "" { if exists { restSubscription, err = c.registry.GetRESTSubscription(prevRestSubsId, false) @@ -263,7 +324,7 @@ func (c *Control) GetOrCreateRestSubscription(p *models.SubscriptionParams, md5s } } else { xapp.Logger.Info("None existing restSubId %s referred by MD5sum %s for a request without subscription ID - deleting cached entry", prevRestSubsId, md5sum) - restDuplicateCtrl.DeleteLastKnownRestSubsIdBasedOnMd5sum(md5sum) + c.restDuplicateCtrl.DeleteLastKnownRestSubsIdBasedOnMd5sum(md5sum) } } @@ -342,19 +403,21 @@ func (c *Control) SubscriptionHandler(params interface{}) (*models.SubscriptionR err = c.e2ap.FillSubscriptionReqMsgs(params, &subReqList, restSubscription) if err != nil { xapp.Logger.Error("%s", err.Error()) - restDuplicateCtrl.DeleteLastKnownRestSubsIdBasedOnMd5sum(md5sum) + c.restDuplicateCtrl.DeleteLastKnownRestSubsIdBasedOnMd5sum(md5sum) c.registry.DeleteRESTSubscription(&restSubId) c.UpdateCounter(cRestSubFailToXapp) return nil, err } - duplicate := restDuplicateCtrl.IsDuplicateToOngoingTransaction(restSubId, md5sum) + duplicate := c.restDuplicateCtrl.IsDuplicateToOngoingTransaction(restSubId, md5sum) if duplicate { xapp.Logger.Info("Retransmission blocker direct ACK for request of restSubsId %s restSubId MD5sum %s as retransmission", restSubId, md5sum) c.UpdateCounter(cRestSubRespToXapp) return &subResp, nil } + c.WriteRESTSubscriptionToDb(restSubId, restSubscription) + go c.processSubscriptionRequests(restSubscription, &subReqList, p.ClientEndpoint, p.Meid, &restSubId, xAppRmrEndpoint, md5sum) c.UpdateCounter(cRestSubRespToXapp) @@ -368,12 +431,13 @@ func (c *Control) SubscriptionHandler(params interface{}) (*models.SubscriptionR func (c *Control) processSubscriptionRequests(restSubscription *RESTSubscription, subReqList *e2ap.SubscriptionRequestList, clientEndpoint *models.SubscriptionParamsClientEndpoint, meid *string, restSubId *string, xAppRmrEndpoint string, md5sum string) { - xapp.Logger.Info("Subscription Request count=%v ", len(subReqList.E2APSubscriptionRequests)) + c.SubscriptionProcessingStartDelay() + xapp.Logger.Debug("Subscription Request count=%v ", len(subReqList.E2APSubscriptionRequests)) var xAppEventInstanceID int64 var e2EventInstanceID int64 - defer restDuplicateCtrl.SetMd5sumFromLastOkRequest(*restSubId, md5sum) + defer c.restDuplicateCtrl.SetMd5sumFromLastOkRequest(*restSubId, md5sum) for index := 0; index < len(subReqList.E2APSubscriptionRequests); index++ { subReqMsg := subReqList.E2APSubscriptionRequests[index] @@ -391,7 +455,8 @@ func (c *Control) processSubscriptionRequests(restSubscription *RESTSubscription subRespMsg, err := c.handleSubscriptionRequest(trans, &subReqMsg, meid, *restSubId) - xapp.Logger.Info("Handled SubscriptionRequest index=%v, %s", index, idstring(nil, trans)) + xapp.Logger.Debug("Handled SubscriptionRequest index=%v, %s", index, idstring(nil, trans)) + trans.Release() if err != nil { c.sendUnsuccesfullResponseNotification(restSubId, restSubscription, xAppEventInstanceID, err, clientEndpoint, trans) @@ -402,7 +467,19 @@ func (c *Control) processSubscriptionRequests(restSubscription *RESTSubscription index, *restSubId, clientEndpoint.Host, *clientEndpoint.HTTPPort, xAppEventInstanceID, e2EventInstanceID, idstring(nil, trans)) c.sendSuccesfullResponseNotification(restSubId, restSubscription, xAppEventInstanceID, e2EventInstanceID, clientEndpoint, trans) } - trans.Release() + } +} + +//------------------------------------------------------------------- +// +//------------------------------------------------------------------ +func (c *Control) SubscriptionProcessingStartDelay() { + if c.UTTesting == true { + // This is temporary fix for the UT problem that notification arrives before subscription response + // Correct fix would be to allow notification come before response and process it correctly + xapp.Logger.Debug("Setting 50 ms delay before starting processing Subscriptions") + <-time.After(time.Millisecond * 50) + xapp.Logger.Debug("Continuing after delay") } } @@ -556,7 +633,7 @@ func (c *Control) SubscriptionDeleteHandlerCB(restSubId string) error { restSubscription.DeleteXappIdToE2Id(xAppEventInstanceID) restSubscription.DeleteE2InstanceId(instanceId) } - restDuplicateCtrl.DeleteLastKnownRestSubsIdBasedOnMd5sum(restSubscription.lastReqMd5sum) + c.restDuplicateCtrl.DeleteLastKnownRestSubsIdBasedOnMd5sum(restSubscription.lastReqMd5sum) c.registry.DeleteRESTSubscription(&restSubId) c.RemoveRESTSubscriptionFromDb(restSubId) }()