pkg/rpe/rpe.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    This source code is part of the near-RT RIC (RAN Intelligent Controller)
  20    platform project (RICP).
  21
  22 ==================================================================================
  23 */
  24 /*
  25   Mnemonic:     rpe.go
  26   Abstract:     Contains RPE (Route Policy Engine) module definitions and generic RPE components
  27   Date:         16 March 2019
  28 */
  29
  30 package rpe
  31
  32 import (
  33         "errors"
  34         "routing-manager/pkg/models"
  35         "routing-manager/pkg/rtmgr"
  36         "routing-manager/pkg/sbi"
  37         "runtime"
  38         "strconv"
  39
  40         "gerrit.o-ran-sc.org/r/ric-plt/xapp-frame/pkg/xapp"
  41 )
  42
  43 var (
  44         SupportedRpes = []*EngineConfig{
  45                 {
  46                         Name:        "rmrpush",
  47                         Version:     "pubsush",
  48                         Protocol:    "rmruta",
  49                         Instance:    NewRmrPush(),
  50                         IsAvailable: true,
  51                 },
  52         }
  53 )
  54
  55 func GetRpe(rpeName string) (Engine, error) {
  56         for _, rpe := range SupportedRpes {
  57                 if rpe.Name == rpeName && rpe.IsAvailable {
  58                         return rpe.Instance, nil
  59                 }
  60         }
  61         return nil, errors.New("SBI:" + rpeName + " is not supported or still not a available")
  62 }
  63
  64 type Rpe struct {
  65 }
  66
  67 func getEndpointByName(eps *rtmgr.Endpoints, name string) *rtmgr.Endpoint {
  68         for _, ep := range *eps {
  69                 if ep.Name == name {
  70                         xapp.Logger.Debug("name: %s", ep.Name)
  71                         xapp.Logger.Debug("ep: %v", ep)
  72                         return ep
  73                 }
  74         }
  75         return nil
  76 }
  77
  78 func getEndpointListByName(eps *rtmgr.Endpoints, name string) []rtmgr.Endpoint {
  79         var eplist []rtmgr.Endpoint
  80
  81         for _, ep := range *eps {
  82                 if ep.Name == name {
  83                         xapp.Logger.Debug("name: %s", ep.Name)
  84                         xapp.Logger.Debug("ep: %v", ep)
  85                         eplist = append(eplist, *ep)
  86                 }
  87         }
  88         return eplist
  89 }
  90
  91 func getEndpointByUuid(uuid string) *rtmgr.Endpoint {
  92         endPoints := rtmgr.Eps
  93         for _, ep := range endPoints {
  94                 if ep.Uuid == uuid {
  95                         xapp.Logger.Debug("name: %s", ep.Uuid)
  96                         xapp.Logger.Debug("ep: %v", ep)
  97                         return ep
  98                 }
  99         }
 100         return nil
 101 }
 102
 103 func (r *Rpe) addRoute(messageType string, tx *rtmgr.Endpoint, rx *rtmgr.Endpoint, routeTable *rtmgr.RouteTable, subId int32, routeType string) {
 104         txList := rtmgr.EndpointList{}
 105         rxList := []rtmgr.EndpointList{}
 106
 107         if tx == nil && rx == nil {
 108                 pc, _, _, ok := runtime.Caller(1)
 109                 details := runtime.FuncForPC(pc)
 110                 if ok && details != nil {
 111                         xapp.Logger.Error("Addition of route skipped due to either TX or RX endpoint not present. Caller function is %s", details.Name())
 112                 }
 113         } else {
 114                 if tx != nil {
 115                         txList = rtmgr.EndpointList{*tx}
 116                 }
 117                 if rx != nil {
 118                         rxList = []rtmgr.EndpointList{[]rtmgr.Endpoint{*rx}}
 119                 }
 120                 //messageId := strconv.Itoa(xapp.RICMessageTypes[messageType])
 121                 messageId := rtmgr.Mtype[messageType]
 122                 route := rtmgr.RouteTableEntry{
 123                         MessageType: messageId,
 124                         TxList:      txList,
 125                         RxGroups:    rxList,
 126                         SubID:       subId,
 127                         RouteType:   routeType}
 128                 *routeTable = append(*routeTable, route)
 129                 //              xapp.Logger.Debug("Route added: MessageType: %v, Tx: %v, Rx: %v, SubId: %v", messageId, tx.Uuid, rx.Uuid, subId)
 130                 //              xapp.Logger.Trace("Route added: MessageType: %v, Tx: %v, Rx: %v, SubId: %v", messageId, tx, rx, subId)
 131         }
 132 }
 133
 134 func (r *Rpe) addRoute_rx_list(messageType string, tx *rtmgr.Endpoint, rx []rtmgr.Endpoint, routeTable *rtmgr.RouteTable, subId int32, routeType string) {
 135         txList := rtmgr.EndpointList{}
 136         rxList := []rtmgr.EndpointList{}
 137
 138         if tx != nil {
 139                 txList = rtmgr.EndpointList{*tx}
 140         }
 141
 142         if rx != nil {
 143                 for _, item := range rx {
 144                         ep := []rtmgr.Endpoint{item}
 145                         rxList = append(rxList, ep)
 146                 }
 147         }
 148
 149         //messageId := strconv.Itoa(xapp.RICMessageTypes[messageType])
 150         messageId := rtmgr.Mtype[messageType]
 151         route := rtmgr.RouteTableEntry{
 152                 MessageType: messageId,
 153                 TxList:      txList,
 154                 RxGroups:    rxList,
 155                 SubID:       subId,
 156                 RouteType:   routeType}
 157         *routeTable = append(*routeTable, route)
 158         //      xapp.Logger.Debug("Route added: MessageType: %v, Tx: %v, Rx: %v, SubId: %v", messageId, tx.Uuid, rx.Uuid, SubId)
 159         //      xapp.Logger.Trace("Route added: MessageType: %v, Tx: %v, Rx: %v, SubId: %v", messageId, tx, rx, SubId)
 160 }
 161
 162 func (r *Rpe) generateXappRoutes(xAppEp *rtmgr.Endpoint, subManEp *rtmgr.Endpoint, routeTable *rtmgr.RouteTable) {
 163         xapp.Logger.Debug("rpe.generateXappRoutes invoked")
 164         xapp.Logger.Debug("Endpoint: %v, xAppType: %v", xAppEp.Name, xAppEp.XAppType)
 165         if xAppEp.XAppType != sbi.PlatformType && (len(xAppEp.TxMessages) > 0 || len(xAppEp.RxMessages) > 0) {
 166                 /// TODO ---
 167                 //xApp -> Subscription Manager
 168                 r.addRoute("RIC_SUB_REQ", xAppEp, subManEp, routeTable, -1, "")
 169                 r.addRoute("RIC_SUB_DEL_REQ", xAppEp, subManEp, routeTable, -1, "")
 170                 //xApp -> E2 Termination
 171                 //              r.addRoute("RIC_CONTROL_REQ", xAppEp, e2TermEp, routeTable, -1, "")
 172                 r.addRoute("RIC_CONTROL_REQ", xAppEp, nil, routeTable, -1, "%meid")
 173                 //E2 Termination -> xApp
 174                 ///             r.addRoute("RIC_CONTROL_ACK", e2TermEp, xAppEp, routeTable, -1, "")
 175                 ///             r.addRoute("RIC_CONTROL_FAILURE", e2TermEp, xAppEp, routeTable, -1, "")
 176                 r.addRoute("RIC_CONTROL_ACK", nil, xAppEp, routeTable, -1, "")
 177                 r.addRoute("RIC_CONTROL_FAILURE", nil, xAppEp, routeTable, -1, "")
 178         }
 179         //xApp->A1Mediator
 180         if xAppEp.XAppType != sbi.PlatformType && len(xAppEp.Policies) > 0 {
 181                 xapp.Logger.Debug("rpe.generateXappRoutes found policies section")
 182                 for _, policy := range xAppEp.Policies {
 183                         r.addRoute("A1_POLICY_REQ", nil, xAppEp, routeTable, policy, "")
 184                 }
 185         }
 186
 187 }
 188
 189 func (r *Rpe) generateXappToXappRoutes(RecvxAppEp *rtmgr.Endpoint, endPointList rtmgr.Endpoints, routeTable *rtmgr.RouteTable) {
 190         xapp.Logger.Debug("rpe.generateXappToXappRoutes invoked")
 191
 192         for _, rxmsg := range RecvxAppEp.RxMessages {
 193
 194                 var src_present bool
 195                 identicalMsg := false
 196                 var RxGrp []rtmgr.Endpoint
 197                 xapp.Logger.Debug("RecvxAppEp.RxMessages Endpoint: %v, xAppType: %v and rxmsg: %v ", RecvxAppEp.Name, RecvxAppEp.XAppType, rxmsg)
 198                 if rxmsg != "RIC_SUB_RESP" && rxmsg != "RIC_SUB_FAILURE" && rxmsg != "RIC_SUB_DEL_RESP" && rxmsg != "RIC_SUB_DEL_FAILURE" && rxmsg != "RIC_INDICATION" && rxmsg != "RIC_CONTROL_ACK" && rxmsg != "RIC_CONTROL_FAILURE" && rxmsg != "A1_POLICY_REQ" {
 199                         for _, SrcxAppEp := range endPointList {
 200                                 if SrcxAppEp.XAppType != sbi.PlatformType && (len(SrcxAppEp.TxMessages) > 0) && SrcxAppEp.Name != RecvxAppEp.Name {
 201                                         for _, txmsg := range SrcxAppEp.TxMessages {
 202                                                 if rxmsg == txmsg {
 203                                                         r.addRoute(rxmsg, SrcxAppEp, RecvxAppEp, routeTable, -1, "")
 204                                                         src_present = true
 205                                                         break
 206                                                 }
 207                                         }
 208                                 }
 209                         }
 210                         for _, SrcxAppEp := range endPointList {
 211
 212                                 if SrcxAppEp.XAppType != sbi.PlatformType && (len(SrcxAppEp.RxMessages) > 0) && SrcxAppEp.Name != RecvxAppEp.Name {
 213                                         for _, newrxmsg := range SrcxAppEp.RxMessages {
 214                                                 if newrxmsg == rxmsg {
 215                                                         RxGrp = append(RxGrp, *SrcxAppEp)
 216                                                         identicalMsg = true
 217                                                 }
 218                                         }
 219                                 }
 220                         }
 221                         if src_present == false && identicalMsg == false {
 222                                 xapp.Logger.Debug("Message type %v,for SrcxAppEp.Name %v", rxmsg, RecvxAppEp)
 223                                 r.addRoute(rxmsg, nil, RecvxAppEp, routeTable, -1, "")
 224                         }
 225                         if identicalMsg == true {
 226                                 xapp.Logger.Debug("Appending Message type %v,for SrcxAppEp.Name %v", rxmsg, RecvxAppEp)
 227                                 RxGrp = append(RxGrp, *RecvxAppEp)
 228                                 r.addRoute_rx_list(rxmsg, nil, RxGrp, routeTable, -1, "")
 229                                 //return
 230                         }
 231                 }
 232         }
 233 }
 234
 235 func (r *Rpe) generateSubscriptionRoutes(selectedxAppEp *rtmgr.Endpoint, subManEp *rtmgr.Endpoint, routeTable *rtmgr.RouteTable) {
 236         xapp.Logger.Debug("rpe.addSubscriptionRoutes invoked")
 237         subscriptionList := &rtmgr.Subs
 238         for _, subscription := range *subscriptionList {
 239                 xapp.Logger.Debug("Subscription: %v", subscription)
 240                 xAppUuid := subscription.Fqdn + ":" + strconv.Itoa(int(subscription.Port))
 241                 xapp.Logger.Debug("xApp UUID: %v", xAppUuid)
 242                 xAppEp := getEndpointByUuid(xAppUuid)
 243                 if xAppEp != nil {
 244                         if xAppEp.Uuid == selectedxAppEp.Uuid {
 245                                 xapp.Logger.Debug("xApp UUID is matched for selected xApp.UUID: %v and xApp.Name: %v", selectedxAppEp.Uuid, selectedxAppEp.Name)
 246                                 /// TODO
 247                                 //Subscription Manager -> xApp
 248                                 r.addRoute("RIC_SUB_RESP", subManEp, xAppEp, routeTable, subscription.SubID, "")
 249                                 r.addRoute("RIC_SUB_FAILURE", subManEp, xAppEp, routeTable, subscription.SubID, "")
 250                                 r.addRoute("RIC_SUB_DEL_RESP", subManEp, xAppEp, routeTable, subscription.SubID, "")
 251                                 r.addRoute("RIC_SUB_DEL_FAILURE", subManEp, xAppEp, routeTable, subscription.SubID, "")
 252                                 //E2 Termination -> xApp
 253                                 r.addRoute("RIC_INDICATION", nil, xAppEp, routeTable, subscription.SubID, "")
 254                                 r.addRoute("RIC_CONTROL_ACK", nil, xAppEp, routeTable, subscription.SubID, "")
 255                                 r.addRoute("RIC_CONTROL_FAILURE", nil, xAppEp, routeTable, subscription.SubID, "")
 256                         }
 257                 } else {
 258                         xapp.Logger.Error("generateSubscriptionRoutes xAppEp is nil, xApp UUID: %v", xAppUuid)
 259                 }
 260         }
 261 }
 262
 263 func (r *Rpe) generatePartialSubscriptionTable(xappSubData *models.XappSubscriptionData, subManEp *rtmgr.Endpoint, routeTable *rtmgr.RouteTable) {
 264         xapp.Logger.Debug("rpe.addSingleSubscriptionRoutes invoked")
 265         xAppUuid := *xappSubData.Address + ":" + strconv.Itoa(int(*xappSubData.Port))
 266         xapp.Logger.Debug("xApp UUID: %v", xAppUuid)
 267         xAppEp := getEndpointByUuid(xAppUuid)
 268         if xAppEp != nil {
 269                 //Subscription Manager -> xApp
 270                 r.addRoute("RIC_SUB_RESP", subManEp, xAppEp, routeTable, *xappSubData.SubscriptionID, "")
 271                 r.addRoute("RIC_SUB_FAILURE", subManEp, xAppEp, routeTable, *xappSubData.SubscriptionID, "")
 272                 r.addRoute("RIC_SUB_DEL_RESP", subManEp, xAppEp, routeTable, *xappSubData.SubscriptionID, "")
 273                 r.addRoute("RIC_SUB_DEL_FAILURE", subManEp, xAppEp, routeTable, *xappSubData.SubscriptionID, "")
 274                 //E2 Termination -> xApp
 275                 r.addRoute("RIC_INDICATION", nil, xAppEp, routeTable, *xappSubData.SubscriptionID, "")
 276                 r.addRoute("RIC_CONTROL_ACK", nil, xAppEp, routeTable, *xappSubData.SubscriptionID, "")
 277                 r.addRoute("RIC_CONTROL_FAILURE", nil, xAppEp, routeTable, *xappSubData.SubscriptionID, "")
 278         } else {
 279                 xapp.Logger.Error("generateSubscriptionRoutes xAppEp is nil, xApp UUID: %v", xAppUuid)
 280         }
 281 }
 282
 283 func (r *Rpe) generatePlatformRoutes(e2TermEp []rtmgr.Endpoint, subManEp *rtmgr.Endpoint, e2ManEp *rtmgr.Endpoint, a1mediatorEp *rtmgr.Endpoint, routeTable *rtmgr.RouteTable) {
 284         xapp.Logger.Debug("rpe.generatePlatformRoutes invoked")
 285         //Platform Routes --- Subscription Routes
 286         //Subscription Manager -> E2 Termination
 287         if rtmgr.PrsCfg == nil {
 288                 xapp.Logger.Info("No static routes configuration")
 289                 return
 290         }
 291         for _, routes := range *rtmgr.PrsCfg {
 292                 var sendEp *rtmgr.Endpoint
 293                 var Ep *rtmgr.Endpoint
 294                 switch routes.SenderEndPoint {
 295                 case "SUBMAN":
 296                         sendEp = subManEp
 297                 case "E2MAN":
 298                         sendEp = e2ManEp
 299                 //case "RSM":,
 300                 //      sendEp = rsmEp
 301                 case "A1MEDIATOR":
 302                         sendEp = a1mediatorEp
 303                 }
 304                 switch routes.EndPoint {
 305                 case "SUBMAN":
 306                         Ep = subManEp
 307                 case "E2MAN":
 308                         Ep = e2ManEp
 309                 //case "UEMAN":
 310                 //      Ep = ueManEp
 311                 //case "RSM":
 312                 //      Ep = rsmEp
 313                 case "A1MEDIATOR":
 314                         Ep = a1mediatorEp
 315                 }
 316
 317                 r.addRoute(routes.MessageType, sendEp, Ep, routeTable, routes.SubscriptionId, routes.Meid)
 318         }
 319
 320         if len(e2TermEp) > 0 {
 321                 r.addRoute_rx_list("RIC_SCTP_CLEAR_ALL", e2ManEp, e2TermEp, routeTable, -1, "")
 322                 r.addRoute_rx_list("E2_TERM_KEEP_ALIVE_REQ", e2ManEp, e2TermEp, routeTable, -1, "")
 323                 r.addRoute_rx_list("RIC_E2_SETUP_RESP", e2ManEp, e2TermEp, routeTable, -1, "")
 324                 r.addRoute_rx_list("RIC_E2_SETUP_FAILURE", e2ManEp, e2TermEp, routeTable, -1, "")
 325                 r.addRoute_rx_list("RIC_SERVICE_UPDATE_ACK", e2ManEp, e2TermEp, routeTable, -1, "")
 326                 r.addRoute_rx_list("RIC_SERVICE_UPDATE_FAILURE", e2ManEp, e2TermEp, routeTable, -1, "")
 327                 r.addRoute_rx_list("RIC_E2NODE_CONFIG_UPDATE_ACK", e2ManEp, e2TermEp, routeTable, -1, "")
 328                 r.addRoute_rx_list("RIC_E2NODE_CONFIG_UPDATE_FAILURE", e2ManEp, e2TermEp, routeTable, -1, "")
 329                 r.addRoute_rx_list("RIC_E2_RESET_RESP", e2ManEp, e2TermEp, routeTable, -1, "")
 330         }
 331 }
 332
 333 func (r *Rpe) generatePartialRouteTable(endPointList rtmgr.Endpoints, xappSubData *models.XappSubscriptionData, updatetype rtmgr.RMRUpdateType) *rtmgr.RouteTable {
 334         xapp.Logger.Debug("rpe.generatePartialRouteTable invoked")
 335         xapp.Logger.Debug("Endpoint List:  %v", endPointList)
 336         routeTable := &rtmgr.RouteTable{}
 337         subManEp := getEndpointByName(&endPointList, "SUBMAN")
 338         if subManEp == nil {
 339                 xapp.Logger.Error("Platform component not found: %v", "Subscription Manager")
 340                 xapp.Logger.Debug("Endpoints: %v", endPointList)
 341         }
 342         /*e2TermListEp := getEndpointListByName(&endPointList, "E2TERMINST")
 343         if len(e2TermListEp) == 0 {
 344                 xapp.Logger.Error("Platform component not found: %v", "E2 Termination List")
 345                 xapp.Logger.Debug("Endpoints: %v", endPointList)
 346         }
 347         e2ManEp := getEndpointByName(&endPointList, "E2MAN")
 348         if e2ManEp == nil {
 349                 xapp.Logger.Error("Platform component not found: %v", "E2 Manager")
 350                 xapp.Logger.Debug("Endpoints: %v", endPointList)
 351         }*/
 352
 353         if xappSubData != nil && updatetype == rtmgr.SubsType {
 354                 xapp.Logger.Info("Updating partial subscription routes")
 355                 r.generatePartialSubscriptionTable(xappSubData, subManEp, routeTable)
 356         }
 357         /*if updatetype == rtmgr.XappType {
 358                 xapp.Logger.Info("Updating partial xapp routes")
 359                 for _, endPoint := range endPointList {
 360                         xapp.Logger.Debug("Endpoint: %v, xAppType: %v", endPoint.Name, endPoint.XAppType)
 361                         if endPoint.XAppType != sbi.PlatformType && (len(endPoint.TxMessages) > 0 || len(endPoint.RxMessages) > 0) {
 362                                 r.generateXappRoutes(endPoint, subManEp, routeTable)
 363                                 r.generateXappToXappRoutes(endPoint, endPointList, routeTable)
 364                         }
 365                 }
 366         }
 367         if updatetype == rtmgr.E2Type {
 368                 xapp.Logger.Info("Updating partial E2 routes")
 369                 if len(e2TermListEp) > 0 {
 370                         r.addRoute_rx_list("RIC_SCTP_CLEAR_ALL", e2ManEp, e2TermListEp, routeTable, -1, "")
 371                         r.addRoute_rx_list("E2_TERM_KEEP_ALIVE_REQ", e2ManEp, e2TermListEp, routeTable, -1, "")
 372                         r.addRoute_rx_list("RIC_E2_SETUP_RESP", e2ManEp, e2TermListEp, routeTable, -1, "")
 373                         r.addRoute_rx_list("RIC_E2_SETUP_FAILURE", e2ManEp, e2TermListEp, routeTable, -1, "")
 374                 }
 375         }*/
 376
 377         return routeTable
 378
 379 }
 380
 381 func (r *Rpe) generateRouteTable(endPointList rtmgr.Endpoints) *rtmgr.RouteTable {
 382         xapp.Logger.Debug("rpe.generateRouteTable invoked")
 383         xapp.Logger.Debug("Endpoint List:  %v", endPointList)
 384         routeTable := &rtmgr.RouteTable{}
 385         /*e2TermEp := getEndpointByName(&endPointList, "E2TERM")
 386         if e2TermEp == nil {
 387                 xapp.Logger.Error("Platform component not found: %v", "E2 Termination")
 388                 xapp.Logger.Debug("Endpoints: %v", endPointList)
 389         }*/
 390         subManEp := getEndpointByName(&endPointList, "SUBMAN")
 391         if subManEp == nil {
 392                 xapp.Logger.Error("Platform component not found: %v", "Subscription Manager")
 393                 xapp.Logger.Debug("Endpoints: %v", endPointList)
 394         }
 395         e2ManEp := getEndpointByName(&endPointList, "E2MAN")
 396         if e2ManEp == nil {
 397                 xapp.Logger.Error("Platform component not found: %v", "E2 Manager")
 398                 xapp.Logger.Debug("Endpoints: %v", endPointList)
 399         }
 400         /*rsmEp := getEndpointByName(&endPointList, "RSM")
 401         if rsmEp == nil {
 402                 xapp.Logger.Error("Platform component not found: %v", "Resource Status Manager")
 403                 xapp.Logger.Debug("Endpoints: %v", endPointList)
 404         }*/
 405         A1MediatorEp := getEndpointByName(&endPointList, "A1MEDIATOR")
 406         if A1MediatorEp == nil {
 407                 xapp.Logger.Error("Platform component not found: %v", "A1Mediator")
 408                 xapp.Logger.Debug("Endpoints: %v", endPointList)
 409         }
 410
 411         e2TermListEp := getEndpointListByName(&endPointList, "E2TERMINST")
 412         if len(e2TermListEp) == 0 {
 413                 xapp.Logger.Error("Platform component not found: %v", "E2 Termination List")
 414                 xapp.Logger.Debug("Endpoints: %v", endPointList)
 415         }
 416         r.generatePlatformRoutes(e2TermListEp, subManEp, e2ManEp, A1MediatorEp, routeTable)
 417
 418         for _, endPoint := range endPointList {
 419                 xapp.Logger.Debug("Endpoint: %v, xAppType: %v", endPoint.Name, endPoint.XAppType)
 420                 if endPoint.XAppType != sbi.PlatformType && (len(endPoint.TxMessages) > 0 || len(endPoint.RxMessages) > 0) {
 421                         r.generateXappRoutes(endPoint, subManEp, routeTable)
 422                         r.generateSubscriptionRoutes(endPoint, subManEp, routeTable)
 423                         r.generateXappToXappRoutes(endPoint, endPointList, routeTable)
 424
 425                 }
 426         }
 427         return routeTable
 428 }