X-Git-Url: https://gerrit.o-ran-sc.org/r/gitweb?a=blobdiff_plain;f=pkg%2Fnbi%2Fcontrol.go;h=98d7821dd0412062fa309ae78d4734fa5a483eb3;hb=930db799fe886da7b2c80f89ed3f67c0433c4ed7;hp=04c71a4f4aa32a2a95de863048415329fcf1eaf4;hpb=df23f25ac56eb0076e6e8801bb6080468ccf44b6;p=ric-plt%2Frtmgr.git diff --git a/pkg/nbi/control.go b/pkg/nbi/control.go index 04c71a4..98d7821 100644 --- a/pkg/nbi/control.go +++ b/pkg/nbi/control.go @@ -25,60 +25,104 @@ package nbi import "C" import ( - "errors" - "gerrit.o-ran-sc.org/r/ric-plt/xapp-frame/pkg/xapp" - "strconv" + "errors" + //"fmt" + "gerrit.o-ran-sc.org/r/ric-plt/xapp-frame/pkg/xapp" + "net/http" + "os" + "routing-manager/pkg/models" "routing-manager/pkg/rpe" "routing-manager/pkg/rtmgr" - "routing-manager/pkg/sdl" "routing-manager/pkg/sbi" + "routing-manager/pkg/sdl" + "strconv" + "strings" "sync" + "time" ) +var m sync.Mutex +var EndpointLock sync.Mutex -func NewControl() Control { +var nbiEngine Engine +var sbiEngine sbi.Engine +var sdlEngine sdl.Engine +var rpeEngine rpe.Engine - return Control{make(chan *xapp.RMRParams)} -} +const INTERVAL time.Duration = 60 +func NewControl() Control { + return Control{make(chan *xapp.RMRParams)} +} type Control struct { - rcChan chan *xapp.RMRParams + rcChan chan *xapp.RMRParams } +func (c *Control) Run() { + var err error + go c.controlLoop() + nbiEngine, sbiEngine, sdlEngine, rpeEngine, err = initRtmgr() + if err != nil { + xapp.Logger.Error(err.Error()) + os.Exit(1) + } + + xapp.Resource.InjectRoute("/ric/v1/symptomdata", c.SymptomDataHandler, "GET") -func (c *Control) Run(sbiEngine sbi.Engine, sdlEngine sdl.Engine, rpeEngine rpe.Engine, m *sync.Mutex) { - go c.controlLoop(sbiEngine, sdlEngine, rpeEngine, m) - xapp.Run(c) + xapp.Run(c) +} + +func (c *Control) SymptomDataHandler(w http.ResponseWriter, r *http.Request) { + resp, _ := DumpDebugData() + xapp.Resource.SendSymptomDataJson(w, r, resp, "platform/rttable.json") } func (c *Control) Consume(rp *xapp.RMRParams) (err error) { - c.rcChan <- rp - return + c.rcChan <- rp + return +} + +func initRtmgr() (nbiEngine Engine, sbiEngine sbi.Engine, sdlEngine sdl.Engine, rpeEngine rpe.Engine, err error) { + if nbiEngine, err = GetNbi(xapp.Config.GetString("nbi")); err == nil && nbiEngine != nil { + if sbiEngine, err = sbi.GetSbi(xapp.Config.GetString("sbi")); err == nil && sbiEngine != nil { + if sdlEngine, err = sdl.GetSdl(xapp.Config.GetString("sdl")); err == nil && sdlEngine != nil { + if rpeEngine, err = rpe.GetRpe(xapp.Config.GetString("rpe")); err == nil && rpeEngine != nil { + return nbiEngine, sbiEngine, sdlEngine, rpeEngine, nil + } + } + } + } + return nil, nil, nil, nil, err } -func (c *Control) controlLoop(sbiEngine sbi.Engine, sdlEngine sdl.Engine, rpeEngine rpe.Engine, m *sync.Mutex) { +func (c *Control) controlLoop() { for { msg := <-c.rcChan - switch msg.Mtype { - case xapp.RICMessageTypes["RMRRM_REQ_TABLE"]: - if (rtmgr.Rtmgr_ready == false) { - xapp.Logger.Info("Update Route Table Request(RMR to RM), message discarded as routing manager is not ready") - } else { - xapp.Logger.Info("Update Route Table Request(RMR to RM)") - go c.handleUpdateToRoutingManagerRequest(msg, sbiEngine, sdlEngine, rpeEngine, m) - } - case xapp.RICMessageTypes["RMRRM_TABLE_STATE"]: - xapp.Logger.Info("state of table to route mgr %v", msg) + c.recievermr(msg) + } +} - default: - err := errors.New("Message Type " + strconv.Itoa(msg.Mtype) + " is discarded") - xapp.Logger.Error("Unknown message type: %v", err) +func (c *Control) recievermr(msg *xapp.RMRParams) { + xapp_msg := sbi.RMRParams{msg} + switch msg.Mtype { + case xapp.RICMessageTypes["RMRRM_REQ_TABLE"]: + if rtmgr.Rtmgr_ready == false { + xapp.Logger.Info("Update route Table Request(RMR -> RM), message discarded as routing manager is not ready") + } else { + xapp.Logger.Info("Update Route Table Request(RMR -> RM)") + go c.handleUpdateToRoutingManagerRequest(msg) } + case xapp.RICMessageTypes["RMRRM_TABLE_STATE"]: + xapp.Logger.Info("State of route table to route mgr %s,payload %s", xapp_msg.String(), msg.Payload) + default: + err := errors.New("Message Type " + strconv.Itoa(msg.Mtype) + " is discarded") + xapp.Logger.Error("Unknown message type: %v", err) } + xapp.Rmr.Free(msg.Mbuf) } -func (c *Control) handleUpdateToRoutingManagerRequest(params *xapp.RMRParams, sbiEngine sbi.Engine, sdlEngine sdl.Engine, rpeEngine rpe.Engine, m *sync.Mutex) { +func (c *Control) handleUpdateToRoutingManagerRequest(params *xapp.RMRParams) { msg := sbi.RMRParams{params} @@ -88,21 +132,132 @@ func (c *Control) handleUpdateToRoutingManagerRequest(params *xapp.RMRParams, sb m.Lock() data, err := sdlEngine.ReadAll(xapp.Config.GetString("rtfile")) m.Unlock() - if err != nil || data == nil { - xapp.Logger.Error("Cannot get data from sdl interface due to: " + err.Error()) + if data == nil { + if err != nil { + xapp.Logger.Error("Cannot get data from sdl interface due to: " + err.Error()) + return + } else { + xapp.Logger.Debug("Cannot get data from sdl interface") + return + } + } + + /* hack with WA only for mcxapp in near future */ + if strings.Contains(msg.String(), "ricxapp") { + ep := sbiEngine.CheckEndpoint(string(params.Payload)) + if ep == nil { + xapp.Logger.Error("Update Routing Table Request, can't handle due to end point %s is not available in complete ep list: ", string(params.Payload)) + return + } + } + + epstr, whid := sbiEngine.CreateEndpoint(msg.String()) + if epstr == nil || whid < 0 { + xapp.Logger.Error("Wormhole Id creation failed %d for %s", whid, msg.String()) return } - ep := sbiEngine.CreateEndpoint(string(params.Payload)) - if ep == nil { - xapp.Logger.Error("Update Routing Table Request can't handle due to end point %s is not avail in complete ep list: ",string(params.Payload)) + /*This is to ensure the latest routes are sent. + Assumption is that in this time interval the routes are built for this endpoint */ + time.Sleep(100 * time.Millisecond) + policies := rpeEngine.GeneratePolicies(rtmgr.Eps, data) + err = sbiEngine.DistributeToEp(policies, *epstr, whid) + if err != nil { + xapp.Logger.Error("Not able to publish the routing table due to: " + err.Error()) return } +} + +func getConfigData() (*rtmgr.RicComponents, error) { + var data *rtmgr.RicComponents + m.Lock() + data, err := sdlEngine.ReadAll(xapp.Config.GetString("rtfile")) + + m.Unlock() + if data == nil { + if err != nil { + return nil, errors.New("Cannot get data from sdl interface due to: " + err.Error()) + } else { + xapp.Logger.Debug("Cannot get data from sdl interface due to data is null") + return nil, errors.New("Cannot get data from sdl interface") + } + } + + return data, nil +} + +func updateEp() (err error) { + data, err := getConfigData() + if err != nil { + return errors.New("Routing table cannot be published due to: " + err.Error()) + } + EndpointLock.Lock() + sbiEngine.UpdateEndpoints(data) + EndpointLock.Unlock() + + return nil +} + +func sendPartialRoutesToAll(xappSubData *models.XappSubscriptionData, updatetype rtmgr.RMRUpdateType) (err error) { + policies := rpeEngine.GeneratePartialPolicies(rtmgr.Eps, xappSubData, updatetype) + err = sbiEngine.DistributeAll(policies) + if err != nil { + return errors.New("Routing table cannot be published due to: " + err.Error()) + } + + return nil +} + +func sendRoutesToAll() (err error) { + + data, err := getConfigData() + if err != nil { + return errors.New("Routing table cannot be published due to: " + err.Error()) + } policies := rpeEngine.GeneratePolicies(rtmgr.Eps, data) - err = sbiEngine.DistributeToEp(policies, ep) + err = sbiEngine.DistributeAll(policies) + if err != nil { + return errors.New("Routing table cannot be published due to: " + err.Error()) + } + + return nil +} + +func Serve() { + + nbiErr := nbiEngine.Initialize(xapp.Config.GetString("xmurl"), xapp.Config.GetString("nbiurl"), xapp.Config.GetString("rtfile"), xapp.Config.GetString("cfgfile"), xapp.Config.GetString("e2murl"), sdlEngine, rpeEngine, &m) + if nbiErr != nil { + xapp.Logger.Error("Initialization of nbi failed due to: " + nbiErr.Error()) + return + } + + err := sbiEngine.Initialize(xapp.Config.GetString("sbiurl")) if err != nil { - xapp.Logger.Error("Routing table cannot be published due to: " + err.Error()) + xapp.Logger.Info("Failed to open push socket due to: " + err.Error()) return } + defer nbiEngine.Terminate() + defer sbiEngine.Terminate() + + /* used for rtmgr restart case to connect to Endpoints */ + go updateEp() + time.Sleep(5 * time.Second) + sendRoutesToAll() + /* Sometimes first message fails, retry after 5 sec */ + time.Sleep(5 * time.Second) + sendRoutesToAll() + + for { + xapp.Logger.Debug("Periodic Routes value = %s", xapp.Config.GetString("periodicroutes")) + if xapp.Config.GetString("periodicroutes") == "enable" { + go updateEp() + time.Sleep(5 * time.Second) + sendRoutesToAll() + } + + rtmgr.Rtmgr_ready = true + time.Sleep(INTERVAL * time.Second) + xapp.Logger.Debug("Periodic loop timed out. Setting triggerSBI flag to distribute updated routes.") + } }