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         "gerrit.o-ran-sc.org/r/ric-plt/xapp-frame/pkg/xapp"
  35         "routing-manager/pkg/rtmgr"
  36         "routing-manager/pkg/sbi"
  37         "runtime"
  38         "strconv"
  39 )
  40
  41 var (
  42         SupportedRpes = []*EngineConfig{
  43                 {
  44                         Name:        "rmrpush",
  45                         Version:     "pubsush",
  46                         Protocol:    "rmruta",
  47                         Instance:    NewRmrPush(),
  48                         IsAvailable: true,
  49                 },
  50         }
  51 )
  52
  53 func GetRpe(rpeName string) (Engine, error) {
  54         for _, rpe := range SupportedRpes {
  55                 if rpe.Name == rpeName && rpe.IsAvailable {
  56                         return rpe.Instance, nil
  57                 }
  58         }
  59         return nil, errors.New("SBI:" + rpeName + " is not supported or still not a available")
  60 }
  61
  62 type Rpe struct {
  63 }
  64
  65 func getEndpointByName(eps *rtmgr.Endpoints, name string) *rtmgr.Endpoint {
  66         for _, ep := range *eps {
  67                 if ep.Name == name {
  68                         xapp.Logger.Debug("name: %s", ep.Name)
  69                         xapp.Logger.Debug("ep: %v", ep)
  70                         return ep
  71                 }
  72         }
  73         return nil
  74 }
  75
  76 func getEndpointListByName(eps *rtmgr.Endpoints, name string) []rtmgr.Endpoint {
  77         var eplist []rtmgr.Endpoint
  78
  79         for _, ep := range *eps {
  80                 if ep.Name == name {
  81                         xapp.Logger.Debug("name: %s", ep.Name)
  82                         xapp.Logger.Debug("ep: %v", ep)
  83                         eplist = append(eplist, *ep)
  84                 }
  85         }
  86         return eplist
  87 }
  88
  89 func getEndpointByUuid(uuid string) *rtmgr.Endpoint {
  90         endPoints := rtmgr.Eps
  91         for _, ep := range endPoints {
  92                 if ep.Uuid == uuid {
  93                         xapp.Logger.Debug("name: %s", ep.Uuid)
  94                         xapp.Logger.Debug("ep: %v", ep)
  95                         return ep
  96                 }
  97         }
  98         return nil
  99 }
 100
 101 func (r *Rpe) addRoute(messageType string, tx *rtmgr.Endpoint, rx *rtmgr.Endpoint, routeTable *rtmgr.RouteTable, subId int32, routeType string) {
 102         txList := rtmgr.EndpointList{}
 103         rxList := []rtmgr.EndpointList{}
 104
 105         if tx == nil && rx == nil {
 106                 pc, _, _, ok := runtime.Caller(1)
 107                 details := runtime.FuncForPC(pc)
 108                 if ok && details != nil {
 109                         xapp.Logger.Error("Route addition skipped: Either TX or RX endpoint not present. Caller function is %s", details.Name())
 110                 }
 111         } else {
 112                 if tx != nil {
 113                         txList = rtmgr.EndpointList{*tx}
 114                 }
 115                 if rx != nil {
 116                         rxList = []rtmgr.EndpointList{[]rtmgr.Endpoint{*rx}}
 117                 }
 118                 //messageId := strconv.Itoa(xapp.RICMessageTypes[messageType])
 119                 messageId := rtmgr.Mtype[messageType]
 120                 route := rtmgr.RouteTableEntry{
 121                         MessageType: messageId,
 122                         TxList:      txList,
 123                         RxGroups:    rxList,
 124                         SubID:       subId,
 125                         RouteType:   routeType}
 126                 *routeTable = append(*routeTable, route)
 127                 //              xapp.Logger.Debug("Route added: MessageTyp: %v, Tx: %v, Rx: %v, SubId: %v", messageId, tx.Uuid, rx.Uuid, subId)
 128                 //              xapp.Logger.Trace("Route added: MessageTyp: %v, Tx: %v, Rx: %v, SubId: %v", messageId, tx, rx, subId)
 129         }
 130 }
 131
 132 func (r *Rpe) addRoute_rx_list(messageType string, tx *rtmgr.Endpoint, rx []rtmgr.Endpoint, routeTable *rtmgr.RouteTable, subId int32, routeType string) {
 133         txList := rtmgr.EndpointList{}
 134         rxList := []rtmgr.EndpointList{}
 135
 136         if tx != nil {
 137                 txList = rtmgr.EndpointList{*tx}
 138         }
 139
 140         if rx != nil {
 141                 for _, item := range rx {
 142                         ep := []rtmgr.Endpoint{item}
 143                         rxList = append(rxList, ep)
 144                 }
 145         }
 146
 147         //messageId := strconv.Itoa(xapp.RICMessageTypes[messageType])
 148         messageId := rtmgr.Mtype[messageType]
 149         route := rtmgr.RouteTableEntry{
 150                 MessageType: messageId,
 151                 TxList:      txList,
 152                 RxGroups:    rxList,
 153                 SubID:       subId,
 154                 RouteType:   routeType}
 155         *routeTable = append(*routeTable, route)
 156         //      xapp.Logger.Debug("Route added: MessageTyp: %v, Tx: %v, Rx: %v, SubId: %v", messageId, tx.Uuid, rx.Uuid, subId)
 157         //      xapp.Logger.Trace("Route added: MessageTyp: %v, Tx: %v, Rx: %v, SubId: %v", messageId, tx, rx, subId)
 158 }
 159
 160 func (r *Rpe) generateXappRoutes(xAppEp *rtmgr.Endpoint, subManEp *rtmgr.Endpoint, routeTable *rtmgr.RouteTable) {
 161         xapp.Logger.Debug("rpe.generateXappRoutes invoked")
 162         xapp.Logger.Debug("Endpoint: %v, xAppType: %v", xAppEp.Name, xAppEp.XAppType)
 163         if xAppEp.XAppType != sbi.PlatformType && (len(xAppEp.TxMessages) > 0 || len(xAppEp.RxMessages) > 0) {
 164                 /// TODO ---
 165                 //xApp -> Subscription Manager
 166                 r.addRoute("RIC_SUB_REQ", xAppEp, subManEp, routeTable, -1, "")
 167                 r.addRoute("RIC_SUB_DEL_REQ", xAppEp, subManEp, routeTable, -1, "")
 168                 //xApp -> E2 Termination
 169                 //              r.addRoute("RIC_CONTROL_REQ", xAppEp, e2TermEp, routeTable, -1, "")
 170                 r.addRoute("RIC_CONTROL_REQ", xAppEp, nil, routeTable, -1, "%meid")
 171                 //E2 Termination -> xApp
 172                 ///             r.addRoute("RIC_CONTROL_ACK", e2TermEp, xAppEp, routeTable, -1, "")
 173                 ///             r.addRoute("RIC_CONTROL_FAILURE", e2TermEp, xAppEp, routeTable, -1, "")
 174                 r.addRoute("RIC_CONTROL_ACK", nil, xAppEp, routeTable, -1, "")
 175                 r.addRoute("RIC_CONTROL_FAILURE", nil, xAppEp, routeTable, -1, "")
 176         }
 177         //xApp->A1Mediator
 178         if xAppEp.XAppType != sbi.PlatformType && len(xAppEp.Policies) > 0 {
 179                 xapp.Logger.Debug("rpe.generateXappRoutes found policies section")
 180                 for _, policy := range xAppEp.Policies {
 181                         r.addRoute("A1_POLICY_REQ", nil, xAppEp, routeTable, policy, "")
 182                 }
 183         }
 184
 185 }
 186
 187
 188 func (r *Rpe) generateXappToXappRoutes(RecvxAppEp *rtmgr.Endpoint, endPointList rtmgr.Endpoints, routeTable *rtmgr.RouteTable) {
 189         xapp.Logger.Debug("rpe.generateXappToXappRoutes invoked")
 190
 191         for _, rxmsg := range RecvxAppEp.RxMessages {
 192
 193                 var src_present bool
 194                 xapp.Logger.Debug("RecvxAppEp.RxMessages Endpoint: %v, xAppType: %v and rxmsg: %v ", RecvxAppEp.Name, RecvxAppEp.XAppType, rxmsg)
 195                 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") {
 196                         for _, SrcxAppEp := range endPointList {
 197                                 if SrcxAppEp.XAppType != sbi.PlatformType && (len(SrcxAppEp.TxMessages) > 0) && SrcxAppEp.Name != RecvxAppEp.Name {
 198                                         for _, txmsg := range SrcxAppEp.TxMessages {
 199                                                         if (rxmsg == txmsg) {
 200                                                                 r.addRoute(rxmsg, SrcxAppEp, RecvxAppEp, routeTable, -1, "")
 201                                                                 src_present = true
 202                                                                 break
 203                                                         }
 204                                         }
 205                                 }
 206                         }
 207                         if src_present == false {
 208                                 r.addRoute(rxmsg, nil, RecvxAppEp, routeTable, -1, "")
 209                         }
 210                 }
 211
 212         }
 213 }
 214
 215 func (r *Rpe) generateSubscriptionRoutes(selectedxAppEp *rtmgr.Endpoint, subManEp *rtmgr.Endpoint, routeTable *rtmgr.RouteTable) {
 216         xapp.Logger.Debug("rpe.addSubscriptionRoutes invoked")
 217         subscriptionList := &rtmgr.Subs
 218         for _, subscription := range *subscriptionList {
 219                 xapp.Logger.Debug("Subscription: %v", subscription)
 220                 xAppUuid := subscription.Fqdn + ":" + strconv.Itoa(int(subscription.Port))
 221                 xapp.Logger.Debug("xApp UUID: %v", xAppUuid)
 222                 xAppEp := getEndpointByUuid(xAppUuid)
 223                 if xAppEp != nil {
 224                         if xAppEp.Uuid == selectedxAppEp.Uuid {
 225                                 xapp.Logger.Debug("xApp UUID is matched for selected xApp.UUID: %v and xApp.Name: %v", selectedxAppEp.Uuid, selectedxAppEp.Name)
 226                                 /// TODO
 227                                 //Subscription Manager -> xApp
 228                                 r.addRoute("RIC_SUB_RESP", subManEp, xAppEp, routeTable, subscription.SubID, "")
 229                                 r.addRoute("RIC_SUB_FAILURE", subManEp, xAppEp, routeTable, subscription.SubID, "")
 230                                 r.addRoute("RIC_SUB_DEL_RESP", subManEp, xAppEp, routeTable, subscription.SubID, "")
 231                                 r.addRoute("RIC_SUB_DEL_FAILURE", subManEp, xAppEp, routeTable, subscription.SubID, "")
 232                                 //E2 Termination -> xApp
 233                                 r.addRoute("RIC_INDICATION", nil, xAppEp, routeTable, subscription.SubID, "")
 234                                 r.addRoute("RIC_CONTROL_ACK", nil, xAppEp, routeTable, subscription.SubID, "")
 235                                 r.addRoute("RIC_CONTROL_FAILURE", nil, xAppEp, routeTable, subscription.SubID, "")
 236                         }
 237                 } else {
 238                                 xapp.Logger.Error("generateSubscriptionRoutes xAppEp is nil, xApp UUID: %v", xAppUuid)
 239                 }
 240         }
 241 }
 242
 243 func (r *Rpe) generatePlatformRoutes(e2TermEp []rtmgr.Endpoint, subManEp *rtmgr.Endpoint, e2ManEp *rtmgr.Endpoint, rsmEp *rtmgr.Endpoint, a1mediatorEp *rtmgr.Endpoint, routeTable *rtmgr.RouteTable) {
 244         xapp.Logger.Debug("rpe.generatePlatformRoutes invoked")
 245         //Platform Routes --- Subscription Routes
 246         //Subscription Manager -> E2 Termination
 247         for _, routes := range *rtmgr.PrsCfg {
 248                 var sendEp *rtmgr.Endpoint
 249                 var Ep *rtmgr.Endpoint
 250                 switch routes.SenderEndPoint {
 251                 case "SUBMAN":
 252                         sendEp = subManEp
 253                 case "E2MAN":
 254                         sendEp = e2ManEp
 255                 case "RSM":
 256                         sendEp = rsmEp
 257                 case "A1MEDIATOR":
 258                         sendEp = a1mediatorEp
 259                 }
 260                 switch routes.EndPoint {
 261                 case "SUBMAN":
 262                         Ep = subManEp
 263                 case "E2MAN":
 264                         Ep = e2ManEp
 265                 //case "UEMAN":
 266                 //      Ep = ueManEp
 267                 case "RSM":
 268                         Ep = rsmEp
 269                 case "A1MEDIATOR":
 270                         Ep = a1mediatorEp
 271                 }
 272
 273                 r.addRoute(routes.MessageType, sendEp, Ep, routeTable, routes.SubscriptionId, routes.Meid)
 274         }
 275
 276         if len(e2TermEp) > 0 {
 277                 r.addRoute_rx_list("RIC_SCTP_CLEAR_ALL", e2ManEp, e2TermEp, routeTable, -1, "")
 278                 r.addRoute_rx_list("E2_TERM_KEEP_ALIVE_REQ", e2ManEp, e2TermEp, routeTable, -1, "")
 279                 r.addRoute_rx_list("RIC_E2_SETUP_RESP", e2ManEp, e2TermEp, routeTable, -1, "")
 280                 r.addRoute_rx_list("RIC_E2_SETUP_FAILURE", e2ManEp, e2TermEp, routeTable, -1, "")
 281         }
 282 }
 283
 284 func (r *Rpe) generateRouteTable(endPointList rtmgr.Endpoints) *rtmgr.RouteTable {
 285         xapp.Logger.Debug("rpe.generateRouteTable invoked")
 286         xapp.Logger.Debug("Endpoint List:  %v", endPointList)
 287         routeTable := &rtmgr.RouteTable{}
 288         /*e2TermEp := getEndpointByName(&endPointList, "E2TERM")
 289         if e2TermEp == nil {
 290                 xapp.Logger.Error("Platform component not found: %v", "E2 Termination")
 291                 xapp.Logger.Debug("Endpoints: %v", endPointList)
 292         }*/
 293         subManEp := getEndpointByName(&endPointList, "SUBMAN")
 294         if subManEp == nil {
 295                 xapp.Logger.Error("Platform component not found: %v", "Subscription Manager")
 296                 xapp.Logger.Debug("Endpoints: %v", endPointList)
 297         }
 298         e2ManEp := getEndpointByName(&endPointList, "E2MAN")
 299         if e2ManEp == nil {
 300                 xapp.Logger.Error("Platform component not found: %v", "E2 Manager")
 301                 xapp.Logger.Debug("Endpoints: %v", endPointList)
 302         }
 303         rsmEp := getEndpointByName(&endPointList, "RSM")
 304         if rsmEp == nil {
 305                 xapp.Logger.Error("Platform component not found: %v", "Resource Status Manager")
 306                 xapp.Logger.Debug("Endpoints: %v", endPointList)
 307         }
 308         A1MediatorEp := getEndpointByName(&endPointList, "A1MEDIATOR")
 309         if A1MediatorEp == nil {
 310                 xapp.Logger.Error("Platform component not found: %v", "A1Mediator")
 311                 xapp.Logger.Debug("Endpoints: %v", endPointList)
 312         }
 313
 314         e2TermListEp := getEndpointListByName(&endPointList, "E2TERMINST")
 315         if len(e2TermListEp) == 0 {
 316                 xapp.Logger.Error("Platform component not found: %v", "E2 Termination List")
 317                 xapp.Logger.Debug("Endpoints: %v", endPointList)
 318         }
 319         r.generatePlatformRoutes(e2TermListEp, subManEp, e2ManEp, rsmEp, A1MediatorEp, routeTable)
 320
 321         for _, endPoint := range endPointList {
 322                 xapp.Logger.Debug("Endpoint: %v, xAppType: %v", endPoint.Name, endPoint.XAppType)
 323                 if endPoint.XAppType != sbi.PlatformType && (len(endPoint.TxMessages) > 0 || len(endPoint.RxMessages) > 0) {
 324                         r.generateXappRoutes(endPoint, subManEp, routeTable)
 325                         r.generateSubscriptionRoutes(endPoint, subManEp, routeTable)
 326                         r.generateXappToXappRoutes(endPoint, endPointList, routeTable)
 327
 328                 }
 329         }
 330         return routeTable
 331 }