pkg/nbi/control.go

   1 /*
   2 ==================================================================================
   3   Copyright (c) 2019 AT&T Intellectual Property.
   4   Copyright (c) 2019 Nokia
   5
   6    Licensed under the Apache License, Version 2.0 (the "License");
   7    you may not use this file except in compliance with the License.
   8    You may obtain a copy of the License at
   9
  10        http://www.apache.org/licenses/LICENSE-2.0
  11
  12    Unless required by applicable law or agreed to in writing, software
  13    distributed under the License is distributed on an "AS IS" BASIS,
  14    WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  15    See the License for the specific language governing permissions and
  16    limitations under the License.
  17
  18    This source code is part of the near-RT RIC (RAN Intelligent Controller)
  19    platform project (RICP).
  20
  21 ==================================================================================
  22 */
  23 package nbi
  24
  25 import "C"
  26
  27 import (
  28         "errors"
  29         "gerrit.o-ran-sc.org/r/ric-plt/xapp-frame/pkg/xapp"
  30         "routing-manager/pkg/rpe"
  31         "routing-manager/pkg/rtmgr"
  32         "routing-manager/pkg/sbi"
  33         "routing-manager/pkg/sdl"
  34         "strconv"
  35         "sync"
  36         "time"
  37         "os"
  38 )
  39
  40 var m sync.Mutex
  41
  42 var nbiEngine Engine
  43 var sbiEngine sbi.Engine
  44 var sdlEngine sdl.Engine
  45 var rpeEngine rpe.Engine
  46
  47 const INTERVAL time.Duration = 60
  48
  49 func NewControl() Control {
  50         return Control{make(chan *xapp.RMRParams)}
  51 }
  52
  53 type Control struct {
  54         rcChan chan *xapp.RMRParams
  55 }
  56
  57
  58 func (c *Control) Run() {
  59         var err error
  60         go c.controlLoop()
  61         nbiEngine, sbiEngine, sdlEngine, rpeEngine, err = initRtmgr()
  62         if err != nil {
  63                 xapp.Logger.Error(err.Error())
  64                 os.Exit(1)
  65         }
  66         xapp.Run(c)
  67 }
  68
  69 func (c *Control) Consume(rp *xapp.RMRParams) (err error) {
  70         c.rcChan <- rp
  71         return
  72 }
  73
  74 func initRtmgr() (nbiEngine Engine, sbiEngine sbi.Engine, sdlEngine sdl.Engine, rpeEngine rpe.Engine, err error) {
  75         if nbiEngine, err = GetNbi(xapp.Config.GetString("nbi")); err == nil && nbiEngine != nil {
  76                 if sbiEngine, err = sbi.GetSbi(xapp.Config.GetString("sbi")); err == nil && sbiEngine != nil {
  77                         if sdlEngine, err = sdl.GetSdl(xapp.Config.GetString("sdl")); err == nil && sdlEngine != nil {
  78                                 if rpeEngine, err = rpe.GetRpe(xapp.Config.GetString("rpe")); err == nil && rpeEngine != nil {
  79                                         return nbiEngine, sbiEngine, sdlEngine, rpeEngine, nil
  80                                 }
  81                         }
  82                 }
  83         }
  84         return nil, nil, nil, nil, err
  85 }
  86
  87 func (c *Control) controlLoop() {
  88         for {
  89                 msg := <-c.rcChan
  90                 xapp_msg := sbi.RMRParams{msg}
  91                 switch msg.Mtype {
  92                 case xapp.RICMessageTypes["RMRRM_REQ_TABLE"]:
  93                         if rtmgr.Rtmgr_ready == false {
  94                                 xapp.Logger.Info("Update Route Table Request(RMR to RM), message discarded as routing manager is not ready")
  95                         } else {
  96                                 xapp.Logger.Info("Update Route Table Request(RMR to RM)")
  97                                 go c.handleUpdateToRoutingManagerRequest(msg)
  98                         }
  99                 case xapp.RICMessageTypes["RMRRM_TABLE_STATE"]:
 100                         xapp.Logger.Info("state of table to route mgr %s,payload %s", xapp_msg.String(), msg.Payload)
 101
 102                 default:
 103                         err := errors.New("Message Type " + strconv.Itoa(msg.Mtype) + " is discarded")
 104                         xapp.Logger.Error("Unknown message type: %v", err)
 105                 }
 106                 xapp.Rmr.Free(msg.Mbuf)
 107         }
 108 }
 109
 110 func (c *Control) handleUpdateToRoutingManagerRequest(params *xapp.RMRParams) {
 111
 112         msg := sbi.RMRParams{params}
 113
 114         xapp.Logger.Info("Update Route Table Request, msg.String() : %s", msg.String())
 115         xapp.Logger.Info("Update Route Table Request, params.Payload : %s", string(params.Payload))
 116
 117         m.Lock()
 118         data, err := sdlEngine.ReadAll(xapp.Config.GetString("rtfile"))
 119         m.Unlock()
 120         if err != nil || data == nil {
 121                 xapp.Logger.Error("Cannot get data from sdl interface due to: " + err.Error())
 122                 return
 123         }
 124
 125         ep := sbiEngine.CreateEndpoint(string(params.Payload))
 126         if ep == nil {
 127                 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))
 128                 return
 129         }
 130
 131         policies := rpeEngine.GeneratePolicies(rtmgr.Eps, data)
 132         err = sbiEngine.DistributeToEp(policies, ep)
 133         if err != nil {
 134                 xapp.Logger.Error("Routing table cannot be published due to: " + err.Error())
 135                 return
 136         }
 137 }
 138
 139 func sendRoutesToAll() (err error) {
 140
 141         m.Lock()
 142         data, err := sdlEngine.ReadAll(xapp.Config.GetString("rtfile"))
 143         m.Unlock()
 144         if err != nil || data == nil {
 145                 return errors.New("Cannot get data from sdl interface due to: " + err.Error())
 146         }
 147         sbiEngine.UpdateEndpoints(data)
 148         policies := rpeEngine.GeneratePolicies(rtmgr.Eps, data)
 149         err = sbiEngine.DistributeAll(policies)
 150         if err != nil {
 151                 return errors.New("Routing table cannot be published due to: " + err.Error())
 152         }
 153
 154         return nil
 155 }
 156
 157 func Serve() {
 158
 159         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)
 160         if nbiErr != nil {
 161                 xapp.Logger.Error("Failed to initialize nbi due to: " + nbiErr.Error())
 162                 return
 163         }
 164
 165         err := sbiEngine.Initialize(xapp.Config.GetString("sbiurl"))
 166         if err != nil {
 167                 xapp.Logger.Info("Failed to open push socket due to: " + err.Error())
 168                 return
 169         }
 170         defer nbiEngine.Terminate()
 171         defer sbiEngine.Terminate()
 172
 173         for {
 174                 sendRoutesToAll()
 175
 176                 rtmgr.Rtmgr_ready = true
 177                 time.Sleep(INTERVAL * time.Second)
 178                 xapp.Logger.Debug("Periodic loop timed out. Setting triggerSBI flag to distribute updated routes.")
 179         }
 180 }