Fixed integration and subscription routes related issues for R3
[ric-plt/rtmgr.git] / cmd / rtmgr.go
index e4f76d8..568b7ad 100644 (file)
    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"
        "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)
 }