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         //"fmt"
  30         "gerrit.o-ran-sc.org/r/ric-plt/xapp-frame/pkg/xapp"
  31         "net/http"
  32         "os"
  33         "routing-manager/pkg/models"
  34         "routing-manager/pkg/rpe"
  35         "routing-manager/pkg/rtmgr"
  36         "routing-manager/pkg/sbi"
  37         "routing-manager/pkg/sdl"
  38         "strconv"
  39         "strings"
  40         "sync"
  41         "time"
  42 )
  43
  44 var m sync.Mutex
  45 var EndpointLock sync.Mutex
  46
  47 var nbiEngine Engine
  48 var sbiEngine sbi.Engine
  49 var sdlEngine sdl.Engine
  50 var rpeEngine rpe.Engine
  51
  52 const INTERVAL time.Duration = 60
  53
  54 func NewControl() Control {
  55         return Control{make(chan *xapp.RMRParams)}
  56 }
  57
  58 type Control struct {
  59         rcChan chan *xapp.RMRParams
  60 }
  61
  62 func (c *Control) Run() {
  63         var err error
  64         go c.controlLoop()
  65         nbiEngine, sbiEngine, sdlEngine, rpeEngine, err = initRtmgr()
  66         if err != nil {
  67                 xapp.Logger.Error(err.Error())
  68                 os.Exit(1)
  69         }
  70
  71         xapp.Resource.InjectRoute("/ric/v1/symptomdata", c.SymptomDataHandler, "GET")
  72
  73         xapp.Run(c)
  74 }
  75
  76 func (c *Control) SymptomDataHandler(w http.ResponseWriter, r *http.Request) {
  77         resp, _ := dumpDebugData()
  78         xapp.Resource.SendSymptomDataJson(w, r, resp, "platform/rttable.json")
  79 }
  80
  81 func (c *Control) Consume(rp *xapp.RMRParams) (err error) {
  82         c.rcChan <- rp
  83         return
  84 }
  85
  86 func initRtmgr() (nbiEngine Engine, sbiEngine sbi.Engine, sdlEngine sdl.Engine, rpeEngine rpe.Engine, err error) {
  87         if nbiEngine, err = GetNbi(xapp.Config.GetString("nbi")); err == nil && nbiEngine != nil {
  88                 if sbiEngine, err = sbi.GetSbi(xapp.Config.GetString("sbi")); err == nil && sbiEngine != nil {
  89                         if sdlEngine, err = sdl.GetSdl(xapp.Config.GetString("sdl")); err == nil && sdlEngine != nil {
  90                                 if rpeEngine, err = rpe.GetRpe(xapp.Config.GetString("rpe")); err == nil && rpeEngine != nil {
  91                                         return nbiEngine, sbiEngine, sdlEngine, rpeEngine, nil
  92                                 }
  93                         }
  94                 }
  95         }
  96         return nil, nil, nil, nil, err
  97 }
  98
  99 func (c *Control) controlLoop() {
 100         for {
 101                 msg := <-c.rcChan
 102                 c.recievermr(msg)
 103         }
 104 }
 105
 106 func (c *Control) recievermr(msg *xapp.RMRParams) {
 107         xapp_msg := sbi.RMRParams{msg}
 108         switch msg.Mtype {
 109         case xapp.RICMessageTypes["RMRRM_REQ_TABLE"]:
 110                 if rtmgr.Rtmgr_ready == false {
 111                         xapp.Logger.Info("Update Route Table Request(RMR to RM), message discarded as routing manager is not ready")
 112                 } else {
 113                         xapp.Logger.Info("Update Route Table Request(RMR to RM)")
 114                         go c.handleUpdateToRoutingManagerRequest(msg)
 115                 }
 116         case xapp.RICMessageTypes["RMRRM_TABLE_STATE"]:
 117                 xapp.Logger.Info("state of table to route mgr %s,payload %s", xapp_msg.String(), msg.Payload)
 118         default:
 119                 err := errors.New("Message Type " + strconv.Itoa(msg.Mtype) + " is discarded")
 120                 xapp.Logger.Error("Unknown message type: %v", err)
 121         }
 122         xapp.Rmr.Free(msg.Mbuf)
 123 }
 124
 125 func (c *Control) handleUpdateToRoutingManagerRequest(params *xapp.RMRParams) {
 126
 127         msg := sbi.RMRParams{params}
 128
 129         xapp.Logger.Info("Update Route Table Request, msg.String() : %s", msg.String())
 130         xapp.Logger.Info("Update Route Table Request, params.Payload : %s", string(params.Payload))
 131
 132         m.Lock()
 133         data, err := sdlEngine.ReadAll(xapp.Config.GetString("rtfile"))
 134         m.Unlock()
 135         if data == nil {
 136                 if err != nil {
 137                         xapp.Logger.Error("Cannot get data from sdl interface due to: " + err.Error())
 138                         return
 139                 } else {
 140                         xapp.Logger.Debug("Cannot get data from sdl interface")
 141                         return
 142                 }
 143         }
 144
 145         /* hack with WA only for mcxapp in near future */
 146         if strings.Contains(msg.String(), "ricxapp") {
 147                 ep := sbiEngine.CheckEndpoint(string(params.Payload))
 148                 if ep == nil {
 149                         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))
 150                         return
 151                 }
 152         }
 153
 154         epstr, whid := sbiEngine.CreateEndpoint(msg.String())
 155         if epstr == nil || whid < 0 {
 156                 xapp.Logger.Error("Wormhole Id creation failed %d for %s", whid, msg.String())
 157                 return
 158         }
 159
 160         /*This is to ensure the latest routes are sent.
 161         Assumption is that in this time interval the routes are built for this endpoint */
 162         time.Sleep(100 * time.Millisecond)
 163         policies := rpeEngine.GeneratePolicies(rtmgr.Eps, data)
 164         err = sbiEngine.DistributeToEp(policies, *epstr, whid)
 165         if err != nil {
 166                 xapp.Logger.Error("Routing table cannot be published due to: " + err.Error())
 167                 return
 168         }
 169 }
 170
 171 func getConfigData() (*rtmgr.RicComponents, error) {
 172         var data *rtmgr.RicComponents
 173         m.Lock()
 174         data, err := sdlEngine.ReadAll(xapp.Config.GetString("rtfile"))
 175
 176         m.Unlock()
 177         if data == nil {
 178                 if err != nil {
 179                         return nil, errors.New("Cannot get data from sdl interface due to: " + err.Error())
 180                 } else {
 181                         xapp.Logger.Debug("Cannot get data from sdl interface, data is null")
 182                         return nil, errors.New("Cannot get data from sdl interface")
 183                 }
 184         }
 185
 186         return data, nil
 187 }
 188
 189 func updateEp() (err error) {
 190         data, err := getConfigData()
 191         if err != nil {
 192                 return errors.New("Routing table cannot be published due to: " + err.Error())
 193         }
 194         EndpointLock.Lock()
 195         sbiEngine.UpdateEndpoints(data)
 196         EndpointLock.Unlock()
 197
 198         return nil
 199 }
 200
 201 func sendPartialRoutesToAll(xappSubData *models.XappSubscriptionData, updatetype rtmgr.RMRUpdateType) (err error) {
 202         policies := rpeEngine.GeneratePartialPolicies(rtmgr.Eps, xappSubData, updatetype)
 203         err = sbiEngine.DistributeAll(policies)
 204         if err != nil {
 205                 return errors.New("Routing table cannot be published due to: " + err.Error())
 206         }
 207
 208         return nil
 209 }
 210
 211 func sendRoutesToAll() (err error) {
 212
 213         data, err := getConfigData()
 214         if err != nil {
 215                 return errors.New("Routing table cannot be published due to: " + err.Error())
 216         }
 217
 218         policies := rpeEngine.GeneratePolicies(rtmgr.Eps, data)
 219         err = sbiEngine.DistributeAll(policies)
 220         if err != nil {
 221                 return errors.New("Routing table cannot be published due to: " + err.Error())
 222         }
 223
 224         return nil
 225 }
 226
 227 func Serve() {
 228
 229         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)
 230         if nbiErr != nil {
 231                 xapp.Logger.Error("Failed to initialize nbi due to: " + nbiErr.Error())
 232                 return
 233         }
 234
 235         err := sbiEngine.Initialize(xapp.Config.GetString("sbiurl"))
 236         if err != nil {
 237                 xapp.Logger.Info("Failed to open push socket due to: " + err.Error())
 238                 return
 239         }
 240         defer nbiEngine.Terminate()
 241         defer sbiEngine.Terminate()
 242
 243         /* used for rtmgr restart case to connect to Endpoints */
 244         go updateEp()
 245         time.Sleep(5 * time.Second)
 246         sendRoutesToAll()
 247         /* Sometimes first message  fails, retry after 5 sec */
 248         time.Sleep(5 * time.Second)
 249         sendRoutesToAll()
 250
 251         for {
 252                 xapp.Logger.Debug("Periodic Routes value = %s", xapp.Config.GetString("periodicroutes"))
 253                 if xapp.Config.GetString("periodicroutes") == "enable" {
 254                         go updateEp()
 255                         time.Sleep(5 * time.Second)
 256                         sendRoutesToAll()
 257                 }
 258
 259                 rtmgr.Rtmgr_ready = true
 260                 time.Sleep(INTERVAL * time.Second)
 261                 xapp.Logger.Debug("Periodic loop timed out. Setting triggerSBI flag to distribute updated routes.")
 262         }
 263 }