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).
+
+
==================================================================================
*/
/*
*/
package main
+//TODO: change flag to pflag (won't need any argument parse)
+
import (
- "flag"
- "nbi"
+ "gerrit.o-ran-sc.org/r/ric-plt/xapp-frame/pkg/xapp"
"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"
+ "sync"
)
const SERVICENAME = "rtmgr"
-const INTERVAL time.Duration = 2
-
-var (
- args *map[string]string
-)
+const INTERVAL time.Duration = 60
-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")
- 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"
+func initRtmgr() (nbiEngine nbi.Engine, sbiEngine sbi.Engine, sdlEngine sdl.Engine, rpeEngine rpe.Engine, err error) {
+ if nbiEngine, err = nbi.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
+ }
+ }
+ }
}
- a["rpename"] = (*rpename)
- a["loglevel"] = (*loglevel)
- args = &a
+ return nil, nil, nil, nil, err
}
-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 serveSBI(triggerSBI <-chan bool, sbiEngine sbi.Engine, sdlEngine sdl.Engine, rpeEngine rpe.Engine, m *sync.Mutex) {
+ for {
+ if <-triggerSBI {
+ 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())
+ continue
+ }
+ sbiEngine.UpdateEndpoints(data)
+ policies := rpeEngine.GeneratePolicies(rtmgr.Eps, data)
+ err = sbiEngine.DistributeAll(policies)
+ if err != nil {
+ xapp.Logger.Error("Routing table cannot be published due to: " + err.Error())
}
}
}
- 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 sendRoutesToAll(sbiEngine sbi.Engine, sdlEngine sdl.Engine, rpeEngine rpe.Engine) {
+
+ data, err := sdlEngine.ReadAll(xapp.Config.GetString("rtfile"))
+ if err != nil || data == nil {
+ xapp.Logger.Error("Cannot get data from sdl interface due to: " + err.Error())
+ return
+ }
+ sbiEngine.UpdateEndpoints(data)
+ policies := rpeEngine.GeneratePolicies(rtmgr.Eps, data)
+ err = sbiEngine.DistributeAll(policies)
if err != nil {
- rtmgr.Logger.Info("fail to open pub socket due to: " + err.Error())
+ xapp.Logger.Error("Routing table cannot be published due to: " + err.Error())
return
}
- defer sbi.CloseSocket()
+}
+
+
+func serve(nbiEngine nbi.Engine, sbiEngine sbi.Engine, sdlEngine sdl.Engine, rpeEngine rpe.Engine, m *sync.Mutex) {
+
+ triggerSBI := make(chan bool)
+
+ 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, triggerSBI, m)
+ if nbiErr != nil {
+ xapp.Logger.Error("Failed to initialize nbi due to: " + nbiErr.Error())
+ return
+ }
+
+ err := sbiEngine.Initialize(xapp.Config.GetString("sbiurl"))
+ if err != nil {
+ xapp.Logger.Info("Failed to open push socket due to: " + err.Error())
+ return
+ }
+ 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, m)
+
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 xapp.Config.GetString("nbi") == "httpGetter" {
+ data, err := nbiEngine.(*nbi.HttpGetter).FetchAllXApps(xapp.Config.GetString("xmurl"))
+ if err != nil {
+ xapp.Logger.Error("Cannot fetch xapp data due to: " + err.Error())
+ } else if data != nil {
+ sdlEngine.WriteXApps(xapp.Config.GetString("rtfile"), data)
+ }
}
+
+ sendRoutesToAll(sbiEngine, sdlEngine, rpeEngine)
+
+ rtmgr.Rtmgr_ready = true
+ time.Sleep(INTERVAL * time.Second)
+ xapp.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
+ xapp.Logger.Info("\r- Ctrl+C pressed in Terminal")
+ os.Exit(0)
+ }()
+}
+
func main() {
- parseArgs()
- rtmgr.SetLogLevel((*args)["loglevel"])
- nbi, sbi, sdl, rpe, err := initRtmgr()
+
+ nbiEngine, sbiEngine, sdlEngine, rpeEngine, err := initRtmgr()
if err != nil {
- rtmgr.Logger.Error(err.Error())
+ xapp.Logger.Error(err.Error())
os.Exit(1)
}
- rtmgr.Logger.Info("Start " + SERVICENAME + " service")
- serve(nbi, sbi, sdl, rpe)
+
+ SetupCloseHandler()
+
+ xapp.Logger.Info("Start " + SERVICENAME + " service")
+ rtmgr.Eps = make(rtmgr.Endpoints)
+ rtmgr.Rtmgr_ready = false
+
+ var m sync.Mutex
+
+// RMR thread is starting port: 4560
+ c := nbi.NewControl()
+ go c.Run(sbiEngine, sdlEngine, rpeEngine, &m)
+
+// Waiting for RMR to be ready
+ time.Sleep(time.Duration(2) * time.Second)
+ for xapp.Rmr.IsReady() == false {
+ time.Sleep(time.Duration(2) * time.Second)
+ }
+
+ dummy_whid := int(xapp.Rmr.Openwh("localhost:4560"))
+ xapp.Logger.Info("created dummy Wormhole ID for routingmanager and dummy_whid :%d", dummy_whid)
+
+ serve(nbiEngine, sbiEngine, sdlEngine, rpeEngine, &m)
os.Exit(0)
}
Code Review / ric-plt / rtmgr.git / blobdiff