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         "time"
  24         "errors"
  25         "fmt"
  26         "strings"
  27         "unsafe"
  28         "encoding/json"
  29         "github.com/spf13/viper"
  30         "github.com/valyala/fastjson"
  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         for _, module := range schemas {
 104                 modName := C.CString(module)
 105                 defer C.free(unsafe.Pointer(modName))
 106
 107                 if done := n.SubscribeModule(modName); !done {
 108                         return false
 109                 }
 110         }
 111         return true
 112 }
 113
 114 func (n *Nbi) SubscribeModule(module *C.char) bool {
 115         rc := C.sr_module_change_subscribe(n.session, module, nil, C.sr_module_change_cb(C.module_change_cb), nil, 0, 0, &n.subscription)
 116         if C.SR_ERR_OK != rc {
 117                 log.Info("NBI: sr_module_change_subscribe failed: %s", C.GoString(C.sr_strerror(rc)))
 118                 return false
 119         }
 120         return true
 121 }
 122
 123 //export nbiModuleChangeCB
 124 func nbiModuleChangeCB(session *C.sr_session_ctx_t, module *C.char, xpath *C.char, event C.sr_event_t, reqId C.int) C.int {
 125         changedModule := C.GoString(module)
 126         changedXpath := C.GoString(xpath)
 127
 128         log.Info("NBI: Module change callback - event='%d' module=%s xpath=%s reqId=%d", event, changedModule, changedXpath, reqId)
 129
 130         if C.SR_EV_CHANGE == event {
 131                 configJson := C.yang_data_sr2json(session, module, event, &nbiClient.oper)
 132                 err := nbiClient.ManageXapps(changedModule, C.GoString(configJson), int(nbiClient.oper))
 133                 if err != nil {
 134                         return C.SR_ERR_OPERATION_FAILED
 135                 }
 136         }
 137
 138         if C.SR_EV_DONE == event {
 139                 configJson := C.get_data_json(session, module)
 140                 err := nbiClient.ManageConfigmaps(changedModule, C.GoString(configJson), int(nbiClient.oper))
 141                 if err != nil {
 142                         return C.SR_ERR_OPERATION_FAILED
 143                 }
 144         }
 145
 146         return C.SR_ERR_OK
 147 }
 148
 149 func (n *Nbi) ManageXapps(module, configJson string, oper int) error {
 150         log.Info("ManageXapps: module=%s configJson=%s", module, configJson)
 151
 152         if configJson == "" || module != "o-ran-sc-ric-xapp-desc-v1" {
 153                 return nil
 154         }
 155
 156         root := fmt.Sprintf("%s:ric", module)
 157         jsonList, err := n.ParseJsonArray(configJson, root, "xapps", "xapp")
 158         if err != nil {
 159                 return err
 160         }
 161
 162         for _, m := range jsonList {
 163                 xappName := string(m.GetStringBytes("name"))
 164                 namespace := string(m.GetStringBytes("namespace"))
 165                 relName := string(m.GetStringBytes("release-name"))
 166                 version := string(m.GetStringBytes("version"))
 167
 168                 desc := sbiClient.BuildXappDescriptor(xappName, namespace, relName, version)
 169                 switch oper {
 170                         case C.SR_OP_CREATED:
 171                                 return sbiClient.DeployXapp(desc)
 172                         case C.SR_OP_DELETED:
 173                                 return sbiClient.UndeployXapp(desc)
 174                         default:
 175                                 return errors.New(fmt.Sprintf("Operation '%d' not supported!", oper))
 176                 }
 177         }
 178         return nil
 179 }
 180
 181 func (n *Nbi) ManageConfigmaps(module, configJson string, oper int) error {
 182         log.Info("ManageConfig: module=%s configJson=%s", module, configJson)
 183
 184         if configJson == "" || module != "o-ran-sc-ric-ueec-config-v1" {
 185                 return nil
 186         }
 187
 188         if oper != C.SR_OP_MODIFIED {
 189                 return errors.New(fmt.Sprintf("Operation '%d' not supported!", oper))
 190         }
 191
 192         value, err := n.ParseJson(configJson)
 193         if err != nil {
 194                 return err
 195         }
 196
 197         root := fmt.Sprintf("%s:ric", module)
 198         appName := string(value.GetStringBytes(root, "config", "name"))
 199         namespace := string(value.GetStringBytes(root, "config", "namespace"))
 200         control := value.Get(root, "config", "control").String()
 201
 202         var f interface{}
 203         err = json.Unmarshal([]byte(strings.ReplaceAll(control, "\\", "")), &f)
 204         if err != nil {
 205                 log.Info("json.Unmarshal failed: %v", err)
 206                 return err
 207         }
 208
 209         xappConfig := sbiClient.BuildXappConfig(appName, namespace, f)
 210         return sbiClient.ModifyXappConfig(xappConfig)
 211 }
 212
 213 func (n *Nbi) ParseJson(dsContent string) (*fastjson.Value, error) {
 214         var p fastjson.Parser
 215         v, err := p.Parse(dsContent)
 216         if err != nil {
 217                 log.Info("fastjson.Parser failed: %v", err)
 218         }
 219         return v, err
 220 }
 221
 222 func (n *Nbi) ParseJsonArray(dsContent, model, top, elem string) ([]*fastjson.Value, error) {
 223         v, err := n.ParseJson(dsContent)
 224         if err != nil {
 225                 return nil, err
 226         }
 227         return v.GetArray(model, top, elem), nil
 228 }