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