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         if rtmgr.PrsCfg == nil {
 248                 xapp.Logger.Info("No static routes configuration")
 249                 return
 250         }
 251         for _, routes := range *rtmgr.PrsCfg {
 252                 var sendEp *rtmgr.Endpoint
 253                 var Ep *rtmgr.Endpoint
 254                 switch routes.SenderEndPoint {
 255                 case "SUBMAN":
 256                         sendEp = subManEp
 257                 case "E2MAN":
 258                         sendEp = e2ManEp
 259                 case "RSM":
 260                         sendEp = rsmEp
 261                 case "A1MEDIATOR":
 262                         sendEp = a1mediatorEp
 263                 }
 264                 switch routes.EndPoint {
 265                 case "SUBMAN":
 266                         Ep = subManEp
 267                 case "E2MAN":
 268                         Ep = e2ManEp
 269                 //case "UEMAN":
 270                 //      Ep = ueManEp
 271                 case "RSM":
 272                         Ep = rsmEp
 273                 case "A1MEDIATOR":
 274                         Ep = a1mediatorEp
 275                 }
 276
 277                 r.addRoute(routes.MessageType, sendEp, Ep, routeTable, routes.SubscriptionId, routes.Meid)
 278         }
 279
 280         if len(e2TermEp) > 0 {
 281                 r.addRoute_rx_list("RIC_SCTP_CLEAR_ALL", e2ManEp, e2TermEp, routeTable, -1, "")
 282                 r.addRoute_rx_list("E2_TERM_KEEP_ALIVE_REQ", e2ManEp, e2TermEp, routeTable, -1, "")
 283                 r.addRoute_rx_list("RIC_E2_SETUP_RESP", e2ManEp, e2TermEp, routeTable, -1, "")
 284                 r.addRoute_rx_list("RIC_E2_SETUP_FAILURE", e2ManEp, e2TermEp, routeTable, -1, "")
 285         }
 286 }
 287
 288 func (r *Rpe) generateRouteTable(endPointList rtmgr.Endpoints) *rtmgr.RouteTable {
 289         xapp.Logger.Debug("rpe.generateRouteTable invoked")
 290         xapp.Logger.Debug("Endpoint List:  %v", endPointList)
 291         routeTable := &rtmgr.RouteTable{}
 292         /*e2TermEp := getEndpointByName(&endPointList, "E2TERM")
 293         if e2TermEp == nil {
 294                 xapp.Logger.Error("Platform component not found: %v", "E2 Termination")
 295                 xapp.Logger.Debug("Endpoints: %v", endPointList)
 296         }*/
 297         subManEp := getEndpointByName(&endPointList, "SUBMAN")
 298         if subManEp == nil {
 299                 xapp.Logger.Error("Platform component not found: %v", "Subscription Manager")
 300                 xapp.Logger.Debug("Endpoints: %v", endPointList)
 301         }
 302         e2ManEp := getEndpointByName(&endPointList, "E2MAN")
 303         if e2ManEp == nil {
 304                 xapp.Logger.Error("Platform component not found: %v", "E2 Manager")
 305                 xapp.Logger.Debug("Endpoints: %v", endPointList)
 306         }
 307         rsmEp := getEndpointByName(&endPointList, "RSM")
 308         if rsmEp == nil {
 309                 xapp.Logger.Error("Platform component not found: %v", "Resource Status Manager")
 310                 xapp.Logger.Debug("Endpoints: %v", endPointList)
 311         }
 312         A1MediatorEp := getEndpointByName(&endPointList, "A1MEDIATOR")
 313         if A1MediatorEp == nil {
 314                 xapp.Logger.Error("Platform component not found: %v", "A1Mediator")
 315                 xapp.Logger.Debug("Endpoints: %v", endPointList)
 316         }
 317
 318         e2TermListEp := getEndpointListByName(&endPointList, "E2TERMINST")
 319         if len(e2TermListEp) == 0 {
 320                 xapp.Logger.Error("Platform component not found: %v", "E2 Termination List")
 321                 xapp.Logger.Debug("Endpoints: %v", endPointList)
 322         }
 323         r.generatePlatformRoutes(e2TermListEp, subManEp, e2ManEp, rsmEp, A1MediatorEp, routeTable)
 324
 325         for _, endPoint := range endPointList {
 326                 xapp.Logger.Debug("Endpoint: %v, xAppType: %v", endPoint.Name, endPoint.XAppType)
 327                 if endPoint.XAppType != sbi.PlatformType && (len(endPoint.TxMessages) > 0 || len(endPoint.RxMessages) > 0) {
 328                         r.generateXappRoutes(endPoint, subManEp, routeTable)
 329                         r.generateSubscriptionRoutes(endPoint, subManEp, routeTable)
 330                         r.generateXappToXappRoutes(endPoint, endPointList, routeTable)
 331
 332                 }
 333         }
 334         return routeTable
 335 }