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         "fmt"
  39 )
  40
  41 var m sync.Mutex
  42
  43 var nbiEngine Engine
  44 var sbiEngine sbi.Engine
  45 var sdlEngine sdl.Engine
  46 var rpeEngine rpe.Engine
  47
  48 const INTERVAL time.Duration = 60
  49
  50 func NewControl() Control {
  51         return Control{make(chan *xapp.RMRParams)}
  52 }
  53
  54 type Control struct {
  55         rcChan chan *xapp.RMRParams
  56 }
  57
  58
  59 func (c *Control) Run() {
  60         var err error
  61         go c.controlLoop()
  62         nbiEngine, sbiEngine, sdlEngine, rpeEngine, err = initRtmgr()
  63         if err != nil {
  64                 xapp.Logger.Error(err.Error())
  65                 os.Exit(1)
  66         }
  67         xapp.Run(c)
  68 }
  69
  70 func (c *Control) Consume(rp *xapp.RMRParams) (err error) {
  71         c.rcChan <- rp
  72         return
  73 }
  74
  75 func initRtmgr() (nbiEngine Engine, sbiEngine sbi.Engine, sdlEngine sdl.Engine, rpeEngine rpe.Engine, err error) {
  76         if nbiEngine, err = GetNbi(xapp.Config.GetString("nbi")); err == nil && nbiEngine != nil {
  77                 if sbiEngine, err = sbi.GetSbi(xapp.Config.GetString("sbi")); err == nil && sbiEngine != nil {
  78                         if sdlEngine, err = sdl.GetSdl(xapp.Config.GetString("sdl")); err == nil && sdlEngine != nil {
  79                                 if rpeEngine, err = rpe.GetRpe(xapp.Config.GetString("rpe")); err == nil && rpeEngine != nil {
  80                                         return nbiEngine, sbiEngine, sdlEngine, rpeEngine, nil
  81                                 }
  82                         }
  83                 }
  84         }
  85         return nil, nil, nil, nil, err
  86 }
  87
  88 func (c *Control) controlLoop() {
  89         for {
  90                 msg := <-c.rcChan
  91                 c.recievermr(msg)
  92                 /*
  93                 xapp_msg := sbi.RMRParams{msg}
  94                 switch msg.Mtype {
  95                 case xapp.RICMessageTypes["RMRRM_REQ_TABLE"]:
  96                         if rtmgr.Rtmgr_ready == false {
  97                                 xapp.Logger.Info("Update Route Table Request(RMR to RM), message discarded as routing manager is not ready")
  98                         } else {
  99                                 xapp.Logger.Info("Update Route Table Request(RMR to RM)")
 100                                 go c.handleUpdateToRoutingManagerRequest(msg)
 101                         }
 102                 case xapp.RICMessageTypes["RMRRM_TABLE_STATE"]:
 103                         xapp.Logger.Info("state of table to route mgr %s,payload %s", xapp_msg.String(), msg.Payload)
 104
 105                 default:
 106                         err := errors.New("Message Type " + strconv.Itoa(msg.Mtype) + " is discarded")
 107                         xapp.Logger.Error("Unknown message type: %v", err)
 108                 }
 109                 xapp.Rmr.Free(msg.Mbuf)*/
 110         }
 111 }
 112
 113 func (c *Control) recievermr(msg *xapp.RMRParams) {
 114         xapp_msg := sbi.RMRParams{msg}
 115         switch msg.Mtype {
 116         case xapp.RICMessageTypes["RMRRM_REQ_TABLE"]:
 117         if rtmgr.Rtmgr_ready == false {
 118                 xapp.Logger.Info("Update Route Table Request(RMR to RM), message discarded as routing manager is not ready")
 119         } else {
 120                 xapp.Logger.Info("Update Route Table Request(RMR to RM)")
 121                 go c.handleUpdateToRoutingManagerRequest(msg)
 122         }
 123         case xapp.RICMessageTypes["RMRRM_TABLE_STATE"]:
 124                 xapp.Logger.Info("state of table to route mgr %s,payload %s", xapp_msg.String(), msg.Payload)
 125         default:
 126                 err := errors.New("Message Type " + strconv.Itoa(msg.Mtype) + " is discarded")
 127                 xapp.Logger.Error("Unknown message type: %v", err)
 128         }
 129         xapp.Rmr.Free(msg.Mbuf)
 130 }
 131
 132 func (c *Control) handleUpdateToRoutingManagerRequest(params *xapp.RMRParams) {
 133
 134         msg := sbi.RMRParams{params}
 135
 136         xapp.Logger.Info("Update Route Table Request, msg.String() : %s", msg.String())
 137         xapp.Logger.Info("Update Route Table Request, params.Payload : %s", string(params.Payload))
 138
 139         m.Lock()
 140         data, err := sdlEngine.ReadAll(xapp.Config.GetString("rtfile"))
 141         m.Unlock()
 142         if data == nil {
 143                 if err != nil {
 144                         xapp.Logger.Error("Cannot get data from sdl interface due to: " + err.Error())
 145                         return
 146                 } else {
 147                         xapp.Logger.Debug("Cannot get data from sdl interface")
 148                         return
 149                 }
 150         }
 151
 152         ep := sbiEngine.CheckEndpoint(string(params.Payload))
 153         if ep == nil {
 154                 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))
 155                 return
 156         }
 157         epstr,whid := sbiEngine.CreateEndpoint(msg.String())
 158         if epstr == nil || whid < 0 {
 159                 xapp.Logger.Error("Wormhole Id creation failed %d for %s",whid,msg.String() )
 160                 return
 161         }
 162
 163         /*This is to ensure the latest routes are sent.
 164         Assumption is that in this time interval the routes are built for this endpoint */
 165         time.Sleep(100 * time.Millisecond)
 166         policies := rpeEngine.GeneratePolicies(rtmgr.Eps, data)
 167         err = sbiEngine.DistributeToEp(policies, *epstr, whid)
 168         if err != nil {
 169                 xapp.Logger.Error("Routing table cannot be published due to: " + err.Error())
 170                 return
 171         }
 172 }
 173
 174 func sendRoutesToAll() (err error) {
 175
 176         m.Lock()
 177         data, err := sdlEngine.ReadAll(xapp.Config.GetString("rtfile"))
 178         fmt.Printf("data = %v,%v,%v",data,sdlEngine,sbiEngine)
 179         m.Unlock()
 180         if data == nil {
 181                 if err != nil {
 182                         return errors.New("Cannot get data from sdl interface due to: " + err.Error())
 183                 } else {
 184                         xapp.Logger.Debug("Cannot get data from sdl interface, data is null")
 185                         return errors.New("Cannot get data from sdl interface")
 186                 }
 187         }
 188
 189         if sbiEngine == nil {
 190                 fmt.Printf("SBI is nil")
 191         }
 192         sbiEngine.UpdateEndpoints(data)
 193         policies := rpeEngine.GeneratePolicies(rtmgr.Eps, data)
 194         err = sbiEngine.DistributeAll(policies)
 195         if err != nil {
 196                 return errors.New("Routing table cannot be published due to: " + err.Error())
 197         }
 198
 199         return nil
 200 }
 201
 202 func Serve() {
 203
 204         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)
 205         if nbiErr != nil {
 206                 xapp.Logger.Error("Failed to initialize nbi due to: " + nbiErr.Error())
 207                 return
 208         }
 209
 210         err := sbiEngine.Initialize(xapp.Config.GetString("sbiurl"))
 211         if err != nil {
 212                 xapp.Logger.Info("Failed to open push socket due to: " + err.Error())
 213                 return
 214         }
 215         defer nbiEngine.Terminate()
 216         defer sbiEngine.Terminate()
 217
 218         for {
 219                 sendRoutesToAll()
 220
 221                 rtmgr.Rtmgr_ready = true
 222                 time.Sleep(INTERVAL * time.Second)
 223                 xapp.Logger.Debug("Periodic loop timed out. Setting triggerSBI flag to distribute updated routes.")
 224         }
 225 }