pkg/restful/restful.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 restful
  21
  22 import (
  23         "encoding/json"
  24         "errors"
  25         "fmt"
  26         //"io/ioutil"
  27         "log"
  28         "net/http"
  29         "os"
  30         "strconv"
  31         "strings"
  32         "time"
  33
  34         "gerrit.o-ran-sc.org/r/ric-plt/appmgr/pkg/models"
  35         "gerrit.o-ran-sc.org/r/ric-plt/appmgr/pkg/restapi"
  36         "gerrit.o-ran-sc.org/r/ric-plt/appmgr/pkg/restapi/operations"
  37         "gerrit.o-ran-sc.org/r/ric-plt/appmgr/pkg/restapi/operations/health"
  38         "gerrit.o-ran-sc.org/r/ric-plt/appmgr/pkg/restapi/operations/xapp"
  39         "github.com/go-openapi/loads"
  40         "github.com/go-openapi/runtime/middleware"
  41         "github.com/valyala/fastjson"
  42
  43         "gerrit.o-ran-sc.org/r/ric-plt/appmgr/pkg/appmgr"
  44         "gerrit.o-ran-sc.org/r/ric-plt/appmgr/pkg/resthooks"
  45 )
  46
  47 type XappData struct {
  48         httpendpoint   string
  49         rmrendpoint    string
  50         rmrserviceep   string
  51         status         string
  52         xappname       string
  53         xappinstname   string
  54         xappversion    string
  55         xappconfigpath string
  56         xappInstance   *models.XappInstance
  57 }
  58
  59 var xappmap = map[string]map[string]*XappData{}
  60
  61 func NewRestful() *Restful {
  62         r := &Restful{
  63                 rh:    resthooks.NewResthook(true),
  64                 ready: false,
  65         }
  66         r.api = r.SetupHandler()
  67         return r
  68 }
  69
  70 func (r *Restful) Run() {
  71         server := restapi.NewServer(r.api)
  72         defer server.Shutdown()
  73         server.Port = 8080
  74         server.Host = "0.0.0.0"
  75
  76         appmgr.Logger.Info("Xapp manager started ... serving on %s:%d\n", server.Host, server.Port)
  77
  78         go r.RetrieveApps()
  79         if err := server.Serve(); err != nil {
  80                 log.Fatal(err.Error())
  81         }
  82
  83 }
  84
  85 func (r *Restful) RetrieveApps() {
  86         time.Sleep(5 * time.Second)
  87         var xlist models.RegisterRequest
  88         applist := r.rh.GetAppsInSDL()
  89         if applist != nil {
  90                 appmgr.Logger.Info("List obtained from GetAppsInSDL is %s", *applist)
  91                 newstring := strings.Split(*applist, " ")
  92                 for i, _ := range newstring {
  93                         appmgr.Logger.Debug("Checking for xapp %s", newstring[i])
  94                         if newstring[i] != "" {
  95                                 err := json.Unmarshal([]byte(newstring[i]), &xlist)
  96                                 if err != nil {
  97                                         appmgr.Logger.Error("Error while unmarshalling")
  98                                         continue
  99                                 }
 100                         } else {
 101                                 continue //SDL may have empty item,so need to skip
 102                         }
 103
 104                         xmodel, _ := r.PrepareConfig(xlist, false)
 105                         if xmodel == nil {
 106                                 appmgr.Logger.Error("Xapp not found, deleting it from DB")
 107                                 r.rh.UpdateAppData(xlist, true)
 108                         }
 109                 }
 110         }
 111
 112 }
 113
 114 func (r *Restful) SetupHandler() *operations.AppManagerAPI {
 115         swaggerSpec, err := loads.Embedded(restapi.SwaggerJSON, restapi.FlatSwaggerJSON)
 116         if err != nil {
 117                 appmgr.Logger.Error(err.Error())
 118                 os.Exit(1)
 119         }
 120         api := operations.NewAppManagerAPI(swaggerSpec)
 121
 122         // URL: /ric/v1/health
 123         api.HealthGetHealthAliveHandler = health.GetHealthAliveHandlerFunc(
 124                 func(params health.GetHealthAliveParams) middleware.Responder {
 125                         return health.NewGetHealthAliveOK()
 126                 })
 127
 128         api.HealthGetHealthReadyHandler = health.GetHealthReadyHandlerFunc(
 129                 func(params health.GetHealthReadyParams) middleware.Responder {
 130                         return health.NewGetHealthReadyOK()
 131                 })
 132
 133         // URL: /ric/v1/subscriptions
 134         api.GetSubscriptionsHandler = operations.GetSubscriptionsHandlerFunc(
 135                 func(params operations.GetSubscriptionsParams) middleware.Responder {
 136                         return operations.NewGetSubscriptionsOK().WithPayload(r.rh.GetAllSubscriptions())
 137                 })
 138
 139         api.GetSubscriptionByIDHandler = operations.GetSubscriptionByIDHandlerFunc(
 140                 func(params operations.GetSubscriptionByIDParams) middleware.Responder {
 141                         if result, found := r.rh.GetSubscriptionById(params.SubscriptionID); found {
 142                                 return operations.NewGetSubscriptionByIDOK().WithPayload(&result)
 143                         }
 144                         return operations.NewGetSubscriptionByIDNotFound()
 145                 })
 146
 147         api.AddSubscriptionHandler = operations.AddSubscriptionHandlerFunc(
 148                 func(params operations.AddSubscriptionParams) middleware.Responder {
 149                         return operations.NewAddSubscriptionCreated().WithPayload(r.rh.AddSubscription(*params.SubscriptionRequest))
 150                 })
 151
 152         api.ModifySubscriptionHandler = operations.ModifySubscriptionHandlerFunc(
 153                 func(params operations.ModifySubscriptionParams) middleware.Responder {
 154                         if _, ok := r.rh.ModifySubscription(params.SubscriptionID, *params.SubscriptionRequest); ok {
 155                                 return operations.NewModifySubscriptionOK()
 156                         }
 157                         return operations.NewModifySubscriptionBadRequest()
 158                 })
 159
 160         api.DeleteSubscriptionHandler = operations.DeleteSubscriptionHandlerFunc(
 161                 func(params operations.DeleteSubscriptionParams) middleware.Responder {
 162                         if _, ok := r.rh.DeleteSubscription(params.SubscriptionID); ok {
 163                                 return operations.NewDeleteSubscriptionNoContent()
 164                         }
 165                         return operations.NewDeleteSubscriptionBadRequest()
 166                 })
 167
 168         // URL: /ric/v1/xapp
 169         api.XappGetAllXappsHandler = xapp.GetAllXappsHandlerFunc(
 170                 func(params xapp.GetAllXappsParams) middleware.Responder {
 171                         if result, err := r.GetApps(); err == nil {
 172                                 return xapp.NewGetAllXappsOK().WithPayload(result)
 173                         }
 174                         return xapp.NewGetAllXappsInternalServerError()
 175                 })
 176
 177         // URL: /ric/v1/config
 178         api.XappGetAllXappConfigHandler = xapp.GetAllXappConfigHandlerFunc(
 179                 func(params xapp.GetAllXappConfigParams) middleware.Responder {
 180                         return xapp.NewGetAllXappConfigOK().WithPayload(r.getAppConfig())
 181                 })
 182
 183         api.RegisterXappHandler = operations.RegisterXappHandlerFunc(
 184                 func(params operations.RegisterXappParams) middleware.Responder {
 185                         appmgr.Logger.Info("appname is %s", (*params.RegisterRequest.AppName))
 186                         appmgr.Logger.Info("endpoint is %s", (*params.RegisterRequest.HTTPEndpoint))
 187                         appmgr.Logger.Info("rmrendpoint is %s", (*params.RegisterRequest.RmrEndpoint))
 188                         if result, err := r.RegisterXapp(*params.RegisterRequest); err == nil {
 189                                 go r.rh.PublishSubscription(*result, models.EventTypeDeployed)
 190                                 return operations.NewRegisterXappCreated()
 191                         }
 192                         return operations.NewRegisterXappBadRequest()
 193                 })
 194
 195         api.DeregisterXappHandler = operations.DeregisterXappHandlerFunc(
 196                 func(params operations.DeregisterXappParams) middleware.Responder {
 197                         appmgr.Logger.Info("appname is %s", (*params.DeregisterRequest.AppName))
 198                         if result, err := r.DeregisterXapp(*params.DeregisterRequest); err == nil {
 199                                 go r.rh.PublishSubscription(*result, models.EventTypeUndeployed)
 200                                 return operations.NewDeregisterXappNoContent()
 201                         }
 202                         return operations.NewDeregisterXappBadRequest()
 203                 })
 204
 205         return api
 206 }
 207
 208 func httpGetXAppsconfig(url string) *string {
 209         appmgr.Logger.Info("Invoked httprestful.httpGetXApps: " + url)
 210         resp, err := http.Get(url)
 211         if err != nil {
 212                 return nil
 213         }
 214         defer resp.Body.Close()
 215
 216         if resp.StatusCode == http.StatusOK {
 217                 var data XappConfigList
 218                 appmgr.Logger.Info("http client raw response: %v", resp)
 219                 if err := json.NewDecoder(resp.Body).Decode(&data); err != nil {
 220                         appmgr.Logger.Error("Json decode failed: " + err.Error())
 221                         return nil
 222                 }
 223                 //data[0] assuming only for one app
 224                 str := fmt.Sprintf("%v", data[0].Config)
 225                 appmgr.Logger.Info("HTTP BODY: %v", str)
 226
 227                 resp.Body.Close()
 228                 return &str
 229         } else {
 230                 appmgr.Logger.Info("httprestful got an unexpected http status code: %v", resp.StatusCode)
 231                 return nil
 232         }
 233 }
 234
 235 func parseConfig(config *string) *appmgr.RtmData {
 236         var p fastjson.Parser
 237         var msgs appmgr.RtmData
 238
 239         v, err := p.Parse(*config)
 240         if err != nil {
 241                 appmgr.Logger.Info("fastjson.Parser for failed: %v", err)
 242                 return nil
 243         }
 244
 245         if v.Exists("rmr") {
 246                 for _, m := range v.GetArray("rmr", "txMessages") {
 247                         msgs.TxMessages = append(msgs.TxMessages, strings.Trim(m.String(), `"`))
 248                 }
 249
 250                 for _, m := range v.GetArray("rmr", "rxMessages") {
 251                         msgs.RxMessages = append(msgs.RxMessages, strings.Trim(m.String(), `"`))
 252                 }
 253
 254                 for _, m := range v.GetArray("rmr", "policies") {
 255                         if val, err := strconv.Atoi(strings.Trim(m.String(), `"`)); err == nil {
 256                                 msgs.Policies = append(msgs.Policies, int64(val))
 257                         }
 258                 }
 259         } else {
 260                 for _, p := range v.GetArray("messaging", "ports") {
 261                         appmgr.Logger.Info("txMessages=%v, rxMessages=%v", p.GetArray("txMessages"), p.GetArray("rxMessages"))
 262                         for _, m := range p.GetArray("txMessages") {
 263                                 msgs.TxMessages = append(msgs.TxMessages, strings.Trim(m.String(), `"`))
 264                         }
 265
 266                         for _, m := range p.GetArray("rxMessages") {
 267                                 msgs.RxMessages = append(msgs.RxMessages, strings.Trim(m.String(), `"`))
 268                         }
 269
 270                         for _, m := range p.GetArray("policies") {
 271                                 if val, err := strconv.Atoi(strings.Trim(m.String(), `"`)); err == nil {
 272                                         msgs.Policies = append(msgs.Policies, int64(val))
 273                                 }
 274                         }
 275                 }
 276         }
 277         return &msgs
 278 }
 279
 280 func (r *Restful) RegisterXapp(params models.RegisterRequest) (xapp *models.Xapp, err error) {
 281         return r.PrepareConfig(params, true)
 282 }
 283
 284 func (r *Restful) DeregisterXapp(params models.DeregisterRequest) (xapp *models.Xapp, err error) {
 285         var registeredlist models.RegisterRequest
 286         registeredlist.AppName = params.AppName
 287         registeredlist.AppInstanceName = params.AppInstanceName
 288         if _, found := xappmap[*params.AppName]; found {
 289                 var x models.Xapp
 290                 x.Instances = append(x.Instances, xappmap[*params.AppName][*params.AppInstanceName].xappInstance)
 291                 registeredlist.HTTPEndpoint = &xappmap[*params.AppName][*params.AppInstanceName].httpendpoint
 292                 delete(xappmap[*params.AppName], *params.AppInstanceName)
 293                 if len(xappmap[*params.AppName]) == 0 {
 294                         delete(xappmap, *params.AppName)
 295                 }
 296                 r.rh.UpdateAppData(registeredlist, true)
 297                 return &x, nil
 298         } else {
 299                 appmgr.Logger.Error("XApp Instance %v Not Found", *params.AppName)
 300                 return nil, errors.New("XApp Instance Not Found")
 301         }
 302 }
 303
 304 func (r *Restful) PrepareConfig(params models.RegisterRequest, updateflag bool) (xapp *models.Xapp, err error) {
 305         maxRetries := 5
 306         //tmpString := strings.Split(*params.HTTPEndpoint, "//")
 307         appmgr.Logger.Info("http endpoint is %s", *params.HTTPEndpoint)
 308         for i := 1; i <= maxRetries; i++ {
 309                 xappconfig := httpGetXAppsconfig(fmt.Sprintf("http://%s%s", *params.HTTPEndpoint, params.ConfigPath))
 310
 311                 if xappconfig != nil {
 312                         data := parseConfig(xappconfig)
 313                         if data != nil {
 314                                 appmgr.Logger.Info("iRetry Count = %v", i)
 315                                 var xapp models.Xapp
 316
 317                                 xapp.Name = params.AppName
 318                                 xapp.Version = params.AppVersion
 319                                 //xapp.Status = params.Status
 320
 321                                 r.rh.UpdateAppData(params, updateflag)
 322                                 return r.FillInstanceData(params, &xapp, *data)
 323                                 break
 324                         } else {
 325                                 appmgr.Logger.Error("Couldn't get data due to" + err.Error())
 326                         }
 327                         time.Sleep(2 * time.Second)
 328                 }
 329         }
 330         return nil, errors.New("Unable to get configmap after 5 retries")
 331 }
 332
 333 func (r *Restful) FillInstanceData(params models.RegisterRequest, xapp *models.Xapp, rtData appmgr.RtmData) (xapps *models.Xapp, err error) {
 334
 335         //tmpString := strings.Split(*params.RmrEndpoint, "//")
 336         endPointStr := strings.Split(*params.RmrEndpoint, ":")
 337         var x models.XappInstance
 338         x.Name = params.AppInstanceName
 339         //x.Status = strings.ToLower(params.Status)
 340         x.Status = "deployed"
 341         //x.IP = endPointStr[0]
 342         x.IP = fmt.Sprintf("service-ricxapp-%s-rmr.ricxapp", *params.AppInstanceName)
 343         x.Port, _ = strconv.ParseInt(endPointStr[1], 10, 64)
 344         x.TxMessages = rtData.TxMessages
 345         x.RxMessages = rtData.RxMessages
 346         x.Policies = rtData.Policies
 347         xapp.Instances = append(xapp.Instances, &x)
 348         rmrsrvname := fmt.Sprintf("service-ricxapp-%s-rmr.ricxapp:%s", *params.AppInstanceName, x.Port)
 349
 350         a := &XappData{httpendpoint: *params.HTTPEndpoint,
 351                 rmrendpoint:    *params.RmrEndpoint,
 352                 rmrserviceep:   rmrsrvname,
 353                 status:         "deployed",
 354                 xappname:       *params.AppName,
 355                 xappversion:    params.AppVersion,
 356                 xappinstname:   *params.AppInstanceName,
 357                 xappconfigpath: params.ConfigPath,
 358                 xappInstance:   &x}
 359
 360         if _, ok := xappmap[*params.AppName]; ok {
 361                 xappmap[*params.AppName][*params.AppInstanceName] = a
 362                 appmgr.Logger.Info("appname already present, %v", xappmap[*params.AppName])
 363         } else {
 364                 xappmap[*params.AppName] = make(map[string]*XappData)
 365                 xappmap[*params.AppName][*params.AppInstanceName] = a
 366                 appmgr.Logger.Info("Creating app instance, %v", xappmap[*params.AppName])
 367         }
 368
 369         return xapp, nil
 370
 371 }
 372
 373 func (r *Restful) GetApps() (xapps models.AllDeployedXapps, err error) {
 374         xapps = models.AllDeployedXapps{}
 375         for _, v := range xappmap {
 376                 var x models.Xapp
 377                 for i, j := range v {
 378                         x.Status = j.status
 379                         x.Name = &j.xappname
 380                         x.Version = j.xappversion
 381                         appmgr.Logger.Info("Xapps details currently in map Appname = %v,rmrendpoint = %v,Status = %v", i, j.rmrendpoint, j.status)
 382                         x.Instances = append(x.Instances, j.xappInstance)
 383                 }
 384                 xapps = append(xapps, &x)
 385         }
 386
 387         return xapps, nil
 388
 389 }
 390
 391 func (r *Restful) getAppConfig() (configList models.AllXappConfig) {
 392         for _, v := range xappmap {
 393                 namespace := "ricxapp" //Namespace hardcode, to be removed later
 394                 for _, j := range v {
 395                         var activeConfig interface{}
 396                         xappconfig := httpGetXAppsconfig(fmt.Sprintf("http://%s%s", j.httpendpoint, j.xappconfigpath))
 397
 398                         if xappconfig == nil {
 399                                 appmgr.Logger.Info("config not found for %s", &j.xappname)
 400                                 continue
 401                         }
 402                         json.Unmarshal([]byte(*xappconfig), &activeConfig)
 403
 404                         c := models.XAppConfig{
 405                                 Metadata: &models.ConfigMetadata{XappName: &j.xappname, Namespace: &namespace},
 406                                 Config:   activeConfig,
 407                         }
 408                         configList = append(configList, &c)
 409
 410                 }
 411
 412         }
 413         return
 414 }