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
 133         if ok := n.SubscribeStatus("o-ran-sc-ric-alarm-v1", "/o-ran-sc-ric-alarm-v1:ric/alarms"); !ok {
 134                 return ok
 135         }
 136
 137         return true
 138 }
 139
 140 func (n *Nbi) SubscribeStatus(module, xpath string) bool {
 141         mod := C.CString(module)
 142         path := C.CString(xpath)
 143         defer C.free(unsafe.Pointer(mod))
 144         defer C.free(unsafe.Pointer(path))
 145
 146         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)
 147         if C.SR_ERR_OK != rc {
 148                 log.Error("NBI: sr_oper_get_items_subscribe failed: %s", C.GoString(C.sr_strerror(rc)))
 149                 return false
 150         }
 151         return true
 152 }
 153
 154 //export nbiModuleChangeCB
 155 func nbiModuleChangeCB(session *C.sr_session_ctx_t, module *C.char, xpath *C.char, event C.sr_event_t, reqId C.int) C.int {
 156         changedModule := C.GoString(module)
 157         changedXpath := C.GoString(xpath)
 158
 159         log.Info("NBI: change event='%d' module=%s xpath=%s reqId=%d", event, changedModule, changedXpath, reqId)
 160         if C.SR_EV_CHANGE != event {
 161                 log.Info("NBI: Changes finalized!")
 162                 return C.SR_ERR_OK
 163         }
 164
 165         if changedModule == "o-ran-sc-ric-xapp-desc-v1" {
 166                 configJson := C.yang_data_sr2json(session, module, event, &nbiClient.oper)
 167                 err := nbiClient.ManageXapps(changedModule, C.GoString(configJson), int(nbiClient.oper))
 168                 if err != nil {
 169                         return C.SR_ERR_OPERATION_FAILED
 170                 }
 171         }
 172
 173         if changedModule == "o-ran-sc-ric-ueec-config-v1" {
 174                 configJson := C.get_data_json(session, module)
 175                 err := nbiClient.ManageConfigmaps(changedModule, C.GoString(configJson), int(C.SR_OP_MODIFIED))
 176                 if err != nil {
 177                         return C.SR_ERR_OPERATION_FAILED
 178                 }
 179         }
 180
 181         return C.SR_ERR_OK
 182 }
 183
 184 func (n *Nbi) ManageXapps(module, configJson string, oper int) error {
 185         log.Info("ManageXapps: module=%s configJson=%s", module, configJson)
 186
 187         if configJson == "" {
 188                 return nil
 189         }
 190
 191         root := fmt.Sprintf("%s:ric", module)
 192         jsonList, err := n.ParseJsonArray(configJson, root, "xapps", "xapp")
 193         if err != nil {
 194                 return err
 195         }
 196
 197         for _, m := range jsonList {
 198                 xappName := string(m.GetStringBytes("name"))
 199                 namespace := string(m.GetStringBytes("namespace"))
 200                 relName := string(m.GetStringBytes("release-name"))
 201                 version := string(m.GetStringBytes("version"))
 202
 203                 desc := sbiClient.BuildXappDescriptor(xappName, namespace, relName, version)
 204                 switch oper {
 205                 case C.SR_OP_CREATED:
 206                         return sbiClient.DeployXapp(desc)
 207                 case C.SR_OP_DELETED:
 208                         return sbiClient.UndeployXapp(desc)
 209                 default:
 210                         return errors.New(fmt.Sprintf("Operation '%d' not supported!", oper))
 211                 }
 212         }
 213         return nil
 214 }
 215
 216 func (n *Nbi) ManageConfigmaps(module, configJson string, oper int) error {
 217         log.Info("ManageConfig: module=%s configJson=%s", module, configJson)
 218
 219         if configJson == "" {
 220                 return nil
 221         }
 222
 223         if oper != C.SR_OP_MODIFIED {
 224                 return errors.New(fmt.Sprintf("Operation '%d' not supported!", oper))
 225         }
 226
 227         value, err := n.ParseJson(configJson)
 228         if err != nil {
 229                 log.Info("ParseJson failed with error: %v", oper)
 230                 return err
 231         }
 232
 233         root := fmt.Sprintf("%s:ric", module)
 234         appName := string(value.GetStringBytes(root, "config", "name"))
 235         namespace := string(value.GetStringBytes(root, "config", "namespace"))
 236         control := value.Get(root, "config", "control").String()
 237
 238         var f interface{}
 239         err = json.Unmarshal([]byte(strings.ReplaceAll(control, "\\", "")), &f)
 240         if err != nil {
 241                 log.Info("json.Unmarshal failed: %v", err)
 242                 return err
 243         }
 244
 245         xappConfig := sbiClient.BuildXappConfig(appName, namespace, f)
 246         return sbiClient.ModifyXappConfig(xappConfig)
 247 }
 248
 249 func (n *Nbi) ParseJson(dsContent string) (*fastjson.Value, error) {
 250         var p fastjson.Parser
 251         v, err := p.Parse(dsContent)
 252         if err != nil {
 253                 log.Info("fastjson.Parser failed: %v", err)
 254         }
 255         return v, err
 256 }
 257
 258 func (n *Nbi) ParseJsonArray(dsContent, model, top, elem string) ([]*fastjson.Value, error) {
 259         v, err := n.ParseJson(dsContent)
 260         if err != nil {
 261                 return nil, err
 262         }
 263         return v.GetArray(model, top, elem), nil
 264 }
 265
 266 //export nbiGnbStateCB
 267 func nbiGnbStateCB(session *C.sr_session_ctx_t, module *C.char, xpath *C.char, rpath *C.char, reqid C.uint32_t, parent **C.char) C.int {
 268         mod := C.GoString(module)
 269         log.Info("nbiGnbStateCB: module='%s' xpath='%s' rpath='%s' [id=%d]", mod, C.GoString(xpath), C.GoString(rpath), reqid)
 270
 271         if mod == "o-ran-sc-ric-xapp-desc-v1" {
 272                 podList, _ := sbiClient.GetAllPodStatus("ricxapp")
 273                 for _, pod := range podList {
 274                         path := fmt.Sprintf("/o-ran-sc-ric-xapp-desc-v1:ric/health/status[name='%s']", pod.Name)
 275                         nbiClient.CreateNewElement(session, parent, path, "name", path)
 276                         nbiClient.CreateNewElement(session, parent, path, "health", pod.Health)
 277                         nbiClient.CreateNewElement(session, parent, path, "status", pod.Status)
 278                 }
 279                 return C.SR_ERR_OK
 280         }
 281
 282         if mod == "o-ran-sc-ric-alarm-v1" {
 283                 alerts, _ := sbiClient.GetAlerts()
 284                 for _, alert := range alerts.Payload {
 285                         id := alert.Annotations["alarm_id"]
 286                         path := fmt.Sprintf("/o-ran-sc-ric-alarm-v1:ric/alarms/alarm[alarm-id='%s']", id)
 287                         nbiClient.CreateNewElement(session, parent, path, "alarm-id", id)
 288                         nbiClient.CreateNewElement(session, parent, path, "fault-text", alert.Alert.Labels["alertname"])
 289                         nbiClient.CreateNewElement(session, parent, path, "severity", alert.Alert.Labels["severity"])
 290                         nbiClient.CreateNewElement(session, parent, path, "status", alert.Alert.Labels["status"])
 291                         nbiClient.CreateNewElement(session, parent, path, "additional-info", alert.Annotations["additional_info"])
 292                 }
 293                 return C.SR_ERR_OK
 294         }
 295
 296         gnbs, err := xapp.Rnib.GetListGnbIds()
 297         if err != nil || len(gnbs) == 0 {
 298                 log.Info("Rnib.GetListGnbIds() returned elementCount=%d err:%v", len(gnbs), err)
 299                 return C.SR_ERR_OK
 300         }
 301
 302         for _, gnb := range gnbs {
 303                 ranName := gnb.GetInventoryName()
 304                 info, err := xapp.Rnib.GetNodeb(ranName)
 305                 if err != nil {
 306                         log.Error("GetNodeb() failed for ranName=%s: %v", ranName, err)
 307                         continue
 308                 }
 309
 310                 prot := nbiClient.E2APProt2Str(int(info.E2ApplicationProtocol))
 311                 connStat := nbiClient.ConnStatus2Str(int(info.ConnectionStatus))
 312                 ntype := nbiClient.NodeType2Str(int(info.NodeType))
 313
 314                 log.Info("gNB info: %s -> %s %s %s -> %s %s", ranName, prot, connStat, ntype, gnb.GetGlobalNbId().GetPlmnId(), gnb.GetGlobalNbId().GetNbId())
 315
 316                 path := fmt.Sprintf("/o-ran-sc-ric-gnb-status-v1:ric/nodes/node[ran-name='%s']", ranName)
 317                 nbiClient.CreateNewElement(session, parent, path, "ran-name", ranName)
 318                 nbiClient.CreateNewElement(session, parent, path, "ip", info.Ip)
 319                 nbiClient.CreateNewElement(session, parent, path, "port", fmt.Sprintf("%d", info.Port))
 320                 nbiClient.CreateNewElement(session, parent, path, "plmn-id", gnb.GetGlobalNbId().GetPlmnId())
 321                 nbiClient.CreateNewElement(session, parent, path, "nb-id", gnb.GetGlobalNbId().GetNbId())
 322                 nbiClient.CreateNewElement(session, parent, path, "e2ap-protocol", prot)
 323                 nbiClient.CreateNewElement(session, parent, path, "connection-status", connStat)
 324                 nbiClient.CreateNewElement(session, parent, path, "node", ntype)
 325         }
 326         return C.SR_ERR_OK
 327 }
 328
 329 func (n *Nbi) CreateNewElement(session *C.sr_session_ctx_t, parent **C.char, key, name, value string) {
 330         basePath := fmt.Sprintf("%s/%s", key, name)
 331         log.Info("%s -> %s", basePath, value)
 332
 333         cPath := C.CString(basePath)
 334         defer C.free(unsafe.Pointer(cPath))
 335         cValue := C.CString(value)
 336         defer C.free(unsafe.Pointer(cValue))
 337
 338         C.create_new_path(session, parent, cPath, cValue)
 339 }
 340
 341 func (n *Nbi) ConnStatus2Str(connStatus int) string {
 342         switch connStatus {
 343         case 0:
 344                 return "not-specified"
 345         case 1:
 346                 return "connected"
 347         case 2:
 348                 return "disconnected"
 349         case 3:
 350                 return "setup-failed"
 351         case 4:
 352                 return "connecting"
 353         case 5:
 354                 return "shutting-down"
 355         case 6:
 356                 return "shutdown"
 357         }
 358         return "not-specified"
 359 }
 360
 361 func (n *Nbi) E2APProt2Str(prot int) string {
 362         switch prot {
 363         case 0:
 364                 return "not-specified"
 365         case 1:
 366                 return "x2-setup-request"
 367         case 2:
 368                 return "endc-x2-setup-request"
 369         }
 370         return "not-specified"
 371 }
 372
 373 func (n *Nbi) NodeType2Str(ntype int) string {
 374         switch ntype {
 375         case 0:
 376                 return "not-specified"
 377         case 1:
 378                 return "enb"
 379         case 2:
 380                 return "gnb"
 381         }
 382         return "not-specified"
 383 }