agent/pkg/nbi/nbi.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 */
  19
  20 package nbi
  21
  22 import (
  23         "encoding/json"
  24         "errors"
  25         "fmt"
  26         "github.com/spf13/viper"
  27         "github.com/valyala/fastjson"
  28         "strings"
  29         "time"
  30         "unsafe"
  31
  32         "gerrit.o-ran-sc.org/r/ric-plt/xapp-frame/pkg/xapp"
  33         "gerrit.oran-osc.org/r/ric-plt/o1mediator/pkg/sbi"
  34 )
  35
  36 /*
  37 #cgo LDFLAGS: -lsysrepo -lyang
  38
  39 #include <stdio.h>
  40 #include <limits.h>
  41 #include <sysrepo.h>
  42 #include <sysrepo/values.h>
  43 #include "helper.h"
  44 */
  45 import "C"
  46
  47 var sbiClient sbi.SBIClientInterface
  48 var nbiClient *Nbi
  49 var log = xapp.Logger
  50
  51 func NewNbi(s sbi.SBIClientInterface) *Nbi {
  52         sbiClient = s
  53
  54         nbiClient = &Nbi{
  55                 schemas:     viper.GetStringSlice("nbi.schemas"),
  56                 cleanupChan: make(chan bool),
  57         }
  58         return nbiClient
  59 }
  60
  61 func (n *Nbi) Start() bool {
  62         if ok := n.Setup(n.schemas); !ok {
  63                 log.Error("NBI: SYSREPO initialization failed, bailing out!")
  64                 return false
  65         }
  66         log.Info("NBI: SYSREPO initialization done ... processing O1 requests!")
  67
  68         return true
  69 }
  70
  71 func (n *Nbi) Stop() {
  72         C.sr_unsubscribe(n.subscription)
  73         C.sr_session_stop(n.session)
  74         C.sr_disconnect(n.connection)
  75
  76         log.Info("NBI: SYSREPO cleanup done gracefully!")
  77 }
  78
  79 func (n *Nbi) Setup(schemas []string) bool {
  80         rc := C.sr_connect(0, &n.connection)
  81         if C.SR_ERR_OK != rc {
  82                 log.Error("NBI: sr_connect failed: %s", C.GoString(C.sr_strerror(rc)))
  83                 return false
  84         }
  85
  86         rc = C.sr_session_start(n.connection, C.SR_DS_RUNNING, &n.session)
  87         if C.SR_ERR_OK != rc {
  88                 log.Error("NBI: sr_session_start failed: %s", C.GoString(C.sr_strerror(rc)))
  89                 return false
  90         }
  91
  92         for {
  93                 if ok := n.DoSubscription(schemas); ok == true {
  94                         break
  95                 }
  96                 time.Sleep(time.Duration(5 * time.Second))
  97         }
  98         return true
  99 }
 100
 101 func (n *Nbi) DoSubscription(schemas []string) bool {
 102         log.Info("Subscribing YANG modules ... %v", schemas)
 103
 104         for _, module := range schemas {
 105                 modName := C.CString(module)
 106                 defer C.free(unsafe.Pointer(modName))
 107
 108                 if done := n.SubscribeModule(modName); !done {
 109                         return false
 110                 }
 111         }
 112         return n.SubscribeStatusData()
 113 }
 114
 115 func (n *Nbi) SubscribeModule(module *C.char) bool {
 116         rc := C.sr_module_change_subscribe(n.session, module, nil, C.sr_module_change_cb(C.module_change_cb), nil, 0, 0, &n.subscription)
 117         if C.SR_ERR_OK != rc {
 118                 log.Info("NBI: sr_module_change_subscribe failed: %s", C.GoString(C.sr_strerror(rc)))
 119                 return false
 120         }
 121         return true
 122 }
 123
 124 func (n *Nbi) SubscribeStatusData() bool {
 125         if ok := n.SubscribeStatus("o-ran-sc-ric-gnb-status-v1", "/o-ran-sc-ric-gnb-status-v1:ric/nodes"); !ok {
 126                 return ok
 127         }
 128
 129         if ok := n.SubscribeStatus("o-ran-sc-ric-xapp-desc-v1", "/o-ran-sc-ric-xapp-desc-v1:ric/health"); !ok {
 130                 return ok
 131         }
 132         return true
 133 }
 134
 135 func (n *Nbi) SubscribeStatus(module, xpath string) bool {
 136         mod := C.CString(module)
 137         path := C.CString(xpath)
 138         defer C.free(unsafe.Pointer(mod))
 139         defer C.free(unsafe.Pointer(path))
 140
 141         rc := C.sr_oper_get_items_subscribe(n.session, mod, path, C.sr_oper_get_items_cb(C.gnb_status_cb), nil, 0, &n.subscription)
 142         if C.SR_ERR_OK != rc {
 143                 log.Error("NBI: sr_oper_get_items_subscribe failed: %s", C.GoString(C.sr_strerror(rc)))
 144                 return false
 145         }
 146         return true
 147 }
 148
 149 //export nbiModuleChangeCB
 150 func nbiModuleChangeCB(session *C.sr_session_ctx_t, module *C.char, xpath *C.char, event C.sr_event_t, reqId C.int) C.int {
 151         changedModule := C.GoString(module)
 152         changedXpath := C.GoString(xpath)
 153
 154         log.Info("NBI: Module change callback - event='%d' module=%s xpath=%s reqId=%d", event, changedModule, changedXpath, reqId)
 155
 156         if C.SR_EV_CHANGE == event {
 157                 configJson := C.yang_data_sr2json(session, module, event, &nbiClient.oper)
 158                 err := nbiClient.ManageXapps(changedModule, C.GoString(configJson), int(nbiClient.oper))
 159                 if err != nil {
 160                         return C.SR_ERR_OPERATION_FAILED
 161                 }
 162         }
 163
 164         if C.SR_EV_DONE == event {
 165                 configJson := C.get_data_json(session, module)
 166                 err := nbiClient.ManageConfigmaps(changedModule, C.GoString(configJson), int(nbiClient.oper))
 167                 if err != nil {
 168                         return C.SR_ERR_OPERATION_FAILED
 169                 }
 170         }
 171
 172         return C.SR_ERR_OK
 173 }
 174
 175 func (n *Nbi) ManageXapps(module, configJson string, oper int) error {
 176         log.Info("ManageXapps: module=%s configJson=%s", module, configJson)
 177
 178         if configJson == "" || module != "o-ran-sc-ric-xapp-desc-v1" {
 179                 return nil
 180         }
 181
 182         root := fmt.Sprintf("%s:ric", module)
 183         jsonList, err := n.ParseJsonArray(configJson, root, "xapps", "xapp")
 184         if err != nil {
 185                 return err
 186         }
 187
 188         for _, m := range jsonList {
 189                 xappName := string(m.GetStringBytes("name"))
 190                 namespace := string(m.GetStringBytes("namespace"))
 191                 relName := string(m.GetStringBytes("release-name"))
 192                 version := string(m.GetStringBytes("version"))
 193
 194                 desc := sbiClient.BuildXappDescriptor(xappName, namespace, relName, version)
 195                 switch oper {
 196                 case C.SR_OP_CREATED:
 197                         return sbiClient.DeployXapp(desc)
 198                 case C.SR_OP_DELETED:
 199                         return sbiClient.UndeployXapp(desc)
 200                 default:
 201                         return errors.New(fmt.Sprintf("Operation '%d' not supported!", oper))
 202                 }
 203         }
 204         return nil
 205 }
 206
 207 func (n *Nbi) ManageConfigmaps(module, configJson string, oper int) error {
 208         log.Info("ManageConfig: module=%s configJson=%s", module, configJson)
 209
 210         if configJson == "" || module != "o-ran-sc-ric-ueec-config-v1" {
 211                 return nil
 212         }
 213
 214         if oper != C.SR_OP_MODIFIED {
 215                 return errors.New(fmt.Sprintf("Operation '%d' not supported!", oper))
 216         }
 217
 218         value, err := n.ParseJson(configJson)
 219         if err != nil {
 220                 return err
 221         }
 222
 223         root := fmt.Sprintf("%s:ric", module)
 224         appName := string(value.GetStringBytes(root, "config", "name"))
 225         namespace := string(value.GetStringBytes(root, "config", "namespace"))
 226         control := value.Get(root, "config", "control").String()
 227
 228         var f interface{}
 229         err = json.Unmarshal([]byte(strings.ReplaceAll(control, "\\", "")), &f)
 230         if err != nil {
 231                 log.Info("json.Unmarshal failed: %v", err)
 232                 return err
 233         }
 234
 235         xappConfig := sbiClient.BuildXappConfig(appName, namespace, f)
 236         return sbiClient.ModifyXappConfig(xappConfig)
 237 }
 238
 239 func (n *Nbi) ParseJson(dsContent string) (*fastjson.Value, error) {
 240         var p fastjson.Parser
 241         v, err := p.Parse(dsContent)
 242         if err != nil {
 243                 log.Info("fastjson.Parser failed: %v", err)
 244         }
 245         return v, err
 246 }
 247
 248 func (n *Nbi) ParseJsonArray(dsContent, model, top, elem string) ([]*fastjson.Value, error) {
 249         v, err := n.ParseJson(dsContent)
 250         if err != nil {
 251                 return nil, err
 252         }
 253         return v.GetArray(model, top, elem), nil
 254 }
 255
 256 //export nbiGnbStateCB
 257 func nbiGnbStateCB(session *C.sr_session_ctx_t, module *C.char, xpath *C.char, req_xpath *C.char, reqid C.uint32_t, parent **C.char) C.int {
 258         log.Info("NBI: Module state data for module='%s' path='%s' rpath='%s' requested [id=%d]", C.GoString(module), C.GoString(xpath), C.GoString(req_xpath), reqid)
 259
 260         if C.GoString(module) == "o-ran-sc-ric-xapp-desc-v1" {
 261                 log.Info("xApp health query not implemtented yet!")
 262                 return C.SR_ERR_OK
 263         }
 264
 265         gnbs, err := xapp.Rnib.GetListGnbIds()
 266         if err != nil || len(gnbs) == 0 {
 267                 log.Info("Rnib.GetListGnbIds() returned elementCount=%d err:%v", len(gnbs), err)
 268                 return C.SR_ERR_OK
 269         }
 270
 271         for _, gnb := range gnbs {
 272                 ranName := gnb.GetInventoryName()
 273                 info, err := xapp.Rnib.GetNodeb(ranName)
 274                 if err != nil {
 275                         log.Error("GetNodeb() failed for ranName=%s: %v", ranName, err)
 276                         continue
 277                 }
 278
 279                 prot := nbiClient.E2APProt2Str(int(info.E2ApplicationProtocol))
 280                 connStat := nbiClient.ConnStatus2Str(int(info.ConnectionStatus))
 281                 ntype := nbiClient.NodeType2Str(int(info.NodeType))
 282
 283                 log.Info("gNB info: %s -> %s %s %s -> %s %s", ranName, prot, connStat, ntype, gnb.GetGlobalNbId().GetPlmnId(), gnb.GetGlobalNbId().GetNbId())
 284
 285                 nbiClient.CreateNewElement(session, parent, ranName, "ran-name", ranName)
 286                 nbiClient.CreateNewElement(session, parent, ranName, "ip", info.Ip)
 287                 nbiClient.CreateNewElement(session, parent, ranName, "port", fmt.Sprintf("%d", info.Port))
 288                 nbiClient.CreateNewElement(session, parent, ranName, "plmn-id", gnb.GetGlobalNbId().GetPlmnId())
 289                 nbiClient.CreateNewElement(session, parent, ranName, "nb-id", gnb.GetGlobalNbId().GetNbId())
 290                 nbiClient.CreateNewElement(session, parent, ranName, "e2ap-protocol", prot)
 291                 nbiClient.CreateNewElement(session, parent, ranName, "connection-status", connStat)
 292                 nbiClient.CreateNewElement(session, parent, ranName, "node", ntype)
 293         }
 294         return C.SR_ERR_OK
 295 }
 296
 297 func (n *Nbi) CreateNewElement(session *C.sr_session_ctx_t, parent **C.char, key, name, value string) {
 298         basePath := fmt.Sprintf("/o-ran-sc-ric-gnb-status-v1:ric/nodes/node[ran-name='%s']/%s", key, name)
 299         log.Info("%s -> %s", basePath, value)
 300
 301         cPath := C.CString(basePath)
 302         defer C.free(unsafe.Pointer(cPath))
 303         cValue := C.CString(value)
 304         defer C.free(unsafe.Pointer(cValue))
 305
 306         C.create_new_path(session, parent, cPath, cValue)
 307 }
 308
 309 func (n *Nbi) ConnStatus2Str(connStatus int) string {
 310         switch connStatus {
 311         case 0:
 312                 return "not-specified"
 313         case 1:
 314                 return "connected"
 315         case 2:
 316                 return "disconnected"
 317         case 3:
 318                 return "setup-failed"
 319         case 4:
 320                 return "connecting"
 321         case 5:
 322                 return "shutting-down"
 323         case 6:
 324                 return "shutdown"
 325         }
 326         return "not-specified"
 327 }
 328
 329 func (n *Nbi) E2APProt2Str(prot int) string {
 330         switch prot {
 331         case 0:
 332                 return "not-specified"
 333         case 1:
 334                 return "x2-setup-request"
 335         case 2:
 336                 return "endc-x2-setup-request"
 337         }
 338         return "not-specified"
 339 }
 340
 341 func (n *Nbi) NodeType2Str(ntype int) string {
 342         switch ntype {
 343         case 0:
 344                 return "not-specified"
 345         case 1:
 346                 return "enb"
 347         case 2:
 348                 return "gnb"
 349         }
 350         return "not-specified"
 351 }