X-Git-Url: https://gerrit.o-ran-sc.org/r/gitweb?a=blobdiff_plain;ds=inline;f=cmd%2Frtmgr.go;h=568b7ad4f3287a10dcadc1bee2007106393e7d01;hb=b47d12de30e72b6e1017c93ca487279b09038e00;hp=e4f76d8630d31c08eefd236d902399d955a3845d;hpb=871fa393844ce1b61b8d5218d27687d9fc05803a;p=ric-plt%2Frtmgr.git diff --git a/cmd/rtmgr.go b/cmd/rtmgr.go index e4f76d8..568b7ad 100644 --- a/cmd/rtmgr.go +++ b/cmd/rtmgr.go @@ -14,6 +14,12 @@ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the specific language governing permissions and limitations under the License. + + + This source code is part of the near-RT RIC (RAN Intelligent Controller) + platform project (RICP). + + ================================================================================== */ /* @@ -26,56 +32,49 @@ */ package main +//TODO: change flag to pflag (won't need any argument parse) import ( "flag" - "nbi" "os" - "rpe" - "rtmgr" - "sbi" - "sdl" + "os/signal" + "routing-manager/pkg/nbi" + "routing-manager/pkg/rpe" + "routing-manager/pkg/rtmgr" + "routing-manager/pkg/sbi" + "routing-manager/pkg/sdl" + "syscall" "time" ) const SERVICENAME = "rtmgr" -const INTERVAL time.Duration = 2 +const INTERVAL time.Duration = 60 var ( - args *map[string]string + args map[string]*string ) func parseArgs() { - a := make(map[string]string) - xmgeturl := flag.String("nbi-httpget", "http://localhost:3000/xapps", "xApp Manager URL") - nngpubsock := flag.String("sbi-nngsub", "tcp://0.0.0.0:4560", "NNG Subsciption Socket URI") - file := flag.String("sdl-file", "/db/rt.json", "Local file store location") - rpename := flag.String("rpe", "rmr", "Policy Engine Module name") - loglevel := flag.String("loglevel", "INFO", "INFO | WARN | ERROR | DEBUG") + // TODO: arguments should be validated (filename; xm-url; sbi-if; rest-url; rest-port) + args = make(map[string]*string) + args["configfile"] = flag.String("configfile", "/etc/rtmgrcfg.json", "Routing manager's configuration file path") + args["nbi"] = flag.String("nbi", "httpRESTful", "Northbound interface module to be used. Valid values are: 'httpGetter | httpRESTful'") + args["sbi"] = flag.String("sbi", "nngpush", "Southbound interface module to be used. Valid values are: 'nngpush") + args["rpe"] = flag.String("rpe", "rmrpush", "Route Policy Engine to be used. Valid values are: 'rmrpush'") + args["sdl"] = flag.String("sdl", "file", "Data store engine to be used. Valid values are: 'file'") + args["xm-url"] = flag.String("xm-url", "http://localhost:3000/xapps", "HTTP URL where xApp Manager exposes the entire xApp List") + args["nbi-if"] = flag.String("nbi-if", "http://localhost:8888", "Base HTTP URL where routing manager will be listening on") + args["sbi-if"] = flag.String("sbi-if", "0.0.0.0", "IPv4 address of interface where Southbound socket to be opened") + args["filename"] = flag.String("filename", "/db/rt.json", "Absolute path of file where the route information to be stored") + args["loglevel"] = flag.String("loglevel", "INFO", "INFO | WARN | ERROR | DEBUG | TRACE") flag.Parse() - if (*xmgeturl) != "" { - a["xmurl"] = (*xmgeturl) - a["nbiname"] = "httpGetter" - } - if (*nngpubsock) != "" { - a["socketuri"] = (*nngpubsock) - a["sbiname"] = "nngpub" - } - if (*file) != "" { - a["file"] = (*file) - a["sdlname"] = "file" - } - a["rpename"] = (*rpename) - a["loglevel"] = (*loglevel) - args = &a } -func initRtmgr() (*nbi.NbiEngineConfig, *sbi.SbiEngineConfig, *sdl.SdlEngineConfig, *rpe.RpeEngineConfig, error) { - var err error - if nbi, err := nbi.GetNbi((*args)["nbiname"]); err == nil && nbi != nil { - if sbi, err := sbi.GetSbi((*args)["sbiname"]); err == nil && sbi != nil { - if sdl, err := sdl.GetSdl((*args)["sdlname"]); err == nil && sdl != nil { - if rpe, err := rpe.GetRpe((*args)["rpename"]); err == nil && rpe != nil { - return nbi, sbi, sdl, rpe, nil +func initRtmgr() (nbiEngine nbi.Engine, sbiEngine sbi.Engine, sdlEngine sdl.Engine, rpeEngine rpe.Engine, err error) { + if nbiEngine, err = nbi.GetNbi(*args["nbi"]); err == nil && nbiEngine != nil { + if sbiEngine, err = sbi.GetSbi(*args["sbi"]); err == nil && sbiEngine != nil { + if sdlEngine, err = sdl.GetSdl(*args["sdl"]); err == nil && sdlEngine != nil { + if rpeEngine, err = rpe.GetRpe(*args["rpe"]); err == nil && rpeEngine != nil { + return nbiEngine, sbiEngine, sdlEngine, rpeEngine, nil } } } @@ -83,43 +82,84 @@ func initRtmgr() (*nbi.NbiEngineConfig, *sbi.SbiEngineConfig, *sdl.SdlEngineConf return nil, nil, nil, nil, err } -func serve(nbi *nbi.NbiEngineConfig, sbi *sbi.SbiEngineConfig, sdl *sdl.SdlEngineConfig, rpe *rpe.RpeEngineConfig) { - err := sbi.OpenSocket((*args)["socketuri"]) +func serveSBI(triggerSBI <-chan bool, sbiEngine sbi.Engine, sdlEngine sdl.Engine, rpeEngine rpe.Engine) { + for { + if <-triggerSBI { + data, err := sdlEngine.ReadAll(*args["filename"]) + if err != nil || data == nil { + rtmgr.Logger.Error("Cannot get data from sdl interface due to: " + err.Error()) + continue + } + sbiEngine.UpdateEndpoints(data) + policies := rpeEngine.GeneratePolicies(rtmgr.Eps) + err = sbiEngine.DistributeAll(policies) + if err != nil { + rtmgr.Logger.Error("Routing table cannot be published due to: " + err.Error()) + } + } + } +} + +func serve(nbiEngine nbi.Engine, sbiEngine sbi.Engine, sdlEngine sdl.Engine, rpeEngine rpe.Engine) { + + triggerSBI := make(chan bool) + + nbiErr := nbiEngine.Initialize(*args["xm-url"], *args["nbi-if"], *args["filename"], *args["configfile"], + sdlEngine, rpeEngine, triggerSBI) + if nbiErr != nil { + rtmgr.Logger.Error("Failed to initialize nbi due to: " + nbiErr.Error()) + return + } + + err := sbiEngine.Initialize(*args["sbi-if"]) if err != nil { - rtmgr.Logger.Info("fail to open pub socket due to: " + err.Error()) + rtmgr.Logger.Info("Failed to open push socket due to: " + err.Error()) return } - defer sbi.CloseSocket() + defer nbiEngine.Terminate() + defer sbiEngine.Terminate() + + // This SBI Go routine is trtiggered by periodic main loop and when data is recieved on REST interface. + go serveSBI(triggerSBI, sbiEngine, sdlEngine, rpeEngine) + for { - time.Sleep(INTERVAL * time.Second) - data, err := nbi.BatchFetch((*args)["xmurl"]) - if err != nil { - rtmgr.Logger.Error("cannot get data from " + nbi.Engine.Name + " interface dute to: " + err.Error()) - } else { - sdl.WriteAll((*args)["file"], data) - } - data, err = sdl.ReadAll((*args)["file"]) - if err != nil || data == nil { - rtmgr.Logger.Error("cannot get data from " + sdl.Engine.Name + " interface dute to: " + err.Error()) - continue - } - policies := rpe.GeneratePolicies(data) - err = sbi.DistributeAll(policies) - if err != nil { - rtmgr.Logger.Error("routing rable cannot be published due to: " + err.Error()) + if *args["nbi"] == "httpGetter" { + data, err := nbiEngine.(*nbi.HttpGetter).FetchAllXApps(*args["xm-url"]) + if err != nil { + rtmgr.Logger.Error("Cannot fetch xapp data due to: " + err.Error()) + } else if data != nil { + sdlEngine.WriteXApps(*args["filename"], data) + } } + + triggerSBI <- true + + time.Sleep(INTERVAL * time.Second) + rtmgr.Logger.Debug("Periodic loop timed out. Setting triggerSBI flag to distribute updated routes.") } } +func SetupCloseHandler() { + c := make(chan os.Signal, 2) + signal.Notify(c, os.Interrupt, syscall.SIGTERM) + go func() { + <-c + rtmgr.Logger.Info("\r- Ctrl+C pressed in Terminal") + os.Exit(0) + }() +} + func main() { parseArgs() - rtmgr.SetLogLevel((*args)["loglevel"]) - nbi, sbi, sdl, rpe, err := initRtmgr() + rtmgr.SetLogLevel(*args["loglevel"]) + nbiEngine, sbiEngine, sdlEngine, rpeEngine, err := initRtmgr() if err != nil { rtmgr.Logger.Error(err.Error()) os.Exit(1) } + SetupCloseHandler() rtmgr.Logger.Info("Start " + SERVICENAME + " service") - serve(nbi, sbi, sdl, rpe) + rtmgr.Eps = make(rtmgr.Endpoints) + serve(nbiEngine, sbiEngine, sdlEngine, rpeEngine) os.Exit(0) }