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 getEndpointByUuid(uuid string) *rtmgr.Endpoint {
  77         endPoints := rtmgr.Eps
  78         for _, ep := range endPoints {
  79                 if ep.Uuid == uuid {
  80                         xapp.Logger.Debug("name: %s", ep.Uuid)
  81                         xapp.Logger.Debug("ep: %v", ep)
  82                         return ep
  83                 }
  84         }
  85         return nil
  86 }
  87
  88 func (r *Rpe) addRoute(messageType string, tx *rtmgr.Endpoint, rx *rtmgr.Endpoint, routeTable *rtmgr.RouteTable, subId int32) {
  89         var txList rtmgr.EndpointList
  90         if rx != nil {
  91                 rxList := []rtmgr.EndpointList{[]rtmgr.Endpoint{*rx}}
  92                 if tx != nil {
  93                         txList = rtmgr.EndpointList{*tx}
  94                 }
  95                 messageId := rtmgr.MessageTypes[messageType]
  96                 route := rtmgr.RouteTableEntry{
  97                         MessageType: messageId,
  98                         TxList:      txList,
  99                         RxGroups:    rxList,
 100                         SubID:       subId}
 101                 *routeTable = append(*routeTable, route)
 102                 xapp.Logger.Debug("Route added: MessageTyp: %v,  Rx: %v, SubId: %v", messageId, rx.Uuid, subId)
 103         } else {
 104                 pc, _, _, ok := runtime.Caller(1)
 105                 details := runtime.FuncForPC(pc)
 106                 if ok && details != nil {
 107                         xapp.Logger.Error("Route addition skipped: Either TX or RX endpoint not present. Caller function is %s", details.Name())
 108                 }
 109         }
 110 }
 111
 112 func (r *Rpe) generateXappRoutes(xAppEp *rtmgr.Endpoint, e2TermEp *rtmgr.Endpoint, subManEp *rtmgr.Endpoint, routeTable *rtmgr.RouteTable) {
 113         xapp.Logger.Debug("rpe.generateXappRoutes invoked")
 114         xapp.Logger.Debug("Endpoint: %v, xAppType: %v", xAppEp.Name, xAppEp.XAppType)
 115         if xAppEp.XAppType != sbi.PlatformType && (len(xAppEp.TxMessages) > 0 || len(xAppEp.RxMessages) > 0) {
 116                 //xApp -> Subscription Manager
 117                 r.addRoute("RIC_SUB_REQ", xAppEp, subManEp, routeTable, -1)
 118                 r.addRoute("RIC_SUB_DEL_REQ", xAppEp, subManEp, routeTable, -1)
 119                 //xApp -> E2 Termination
 120                 r.addRoute("RIC_CONTROL_REQ", xAppEp, e2TermEp, routeTable, -1)
 121                 //E2 Termination -> xApp
 122                 r.addRoute("RIC_CONTROL_ACK", e2TermEp, xAppEp, routeTable, -1)
 123                 r.addRoute("RIC_CONTROL_FAILURE", e2TermEp, xAppEp, routeTable, -1)
 124         }
 125         //xApp->A1Mediator
 126         if xAppEp.XAppType != sbi.PlatformType && len(xAppEp.Policies) > 0 {
 127                 xapp.Logger.Debug("rpe.generateXappRoutes found policies section")
 128                 for _, policy := range xAppEp.Policies {
 129                         r.addRoute("A1_POLICY_REQ", nil, xAppEp, routeTable, policy)
 130                 }
 131         }
 132
 133 }
 134
 135 func (r *Rpe) generateSubscriptionRoutes(selectedxAppEp *rtmgr.Endpoint, e2TermEp *rtmgr.Endpoint, subManEp *rtmgr.Endpoint, routeTable *rtmgr.RouteTable) {
 136         xapp.Logger.Debug("rpe.addSubscriptionRoutes invoked")
 137         subscriptionList := &rtmgr.Subs
 138         for _, subscription := range *subscriptionList {
 139                 xapp.Logger.Debug("Subscription: %v", subscription)
 140                 xAppUuid := subscription.Fqdn + ":" + strconv.Itoa(int(subscription.Port))
 141                 xapp.Logger.Debug("xApp UUID: %v", xAppUuid)
 142                 xAppEp := getEndpointByUuid(xAppUuid)
 143                 if xAppEp.Uuid == selectedxAppEp.Uuid {
 144                         xapp.Logger.Debug("xApp UUID is matched for selected xApp.UUID: %v and xApp.Name: %v", selectedxAppEp.Uuid, selectedxAppEp.Name)
 145                         //Subscription Manager -> xApp
 146                         r.addRoute("RIC_SUB_RESP", subManEp, xAppEp, routeTable, subscription.SubID)
 147                         r.addRoute("RIC_SUB_FAILURE", subManEp, xAppEp, routeTable, subscription.SubID)
 148                         r.addRoute("RIC_SUB_DEL_RESP", subManEp, xAppEp, routeTable, subscription.SubID)
 149                         r.addRoute("RIC_SUB_DEL_FAILURE", subManEp, xAppEp, routeTable, subscription.SubID)
 150                         //E2 Termination -> xApp
 151                         r.addRoute("RIC_INDICATION", e2TermEp, xAppEp, routeTable, subscription.SubID)
 152                         r.addRoute("RIC_CONTROL_ACK", e2TermEp, xAppEp, routeTable, subscription.SubID)
 153                         r.addRoute("RIC_CONTROL_FAILURE", e2TermEp, xAppEp, routeTable, subscription.SubID)
 154                 }
 155         }
 156 }
 157
 158 func (r *Rpe) generatePlatformRoutes(e2TermEp *rtmgr.Endpoint, subManEp *rtmgr.Endpoint, e2ManEp *rtmgr.Endpoint, ueManEp *rtmgr.Endpoint, rsmEp *rtmgr.Endpoint, a1mediatorEp *rtmgr.Endpoint, routeTable *rtmgr.RouteTable) {
 159         xapp.Logger.Debug("rpe.generatePlatformRoutes invoked")
 160         //Platform Routes --- Subscription Routes
 161         //Subscription Manager -> E2 Termination
 162         r.addRoute("RIC_SUB_REQ", subManEp, e2TermEp, routeTable, -1)
 163         r.addRoute("RIC_SUB_DEL_REQ", subManEp, e2TermEp, routeTable, -1)
 164         //E2 Termination -> Subscription Manager
 165         r.addRoute("RIC_SUB_RESP", e2TermEp, subManEp, routeTable, -1)
 166         r.addRoute("RIC_SUB_DEL_RESP", e2TermEp, subManEp, routeTable, -1)
 167         r.addRoute("RIC_SUB_FAILURE", e2TermEp, subManEp, routeTable, -1)
 168         r.addRoute("RIC_SUB_DEL_FAILURE", e2TermEp, subManEp, routeTable, -1)
 169         //TODO: UE Man Routes removed (since it is not existing)
 170         //UE Manager -> Subscription Manager
 171         //r.addRoute("RIC_SUB_REQ", ueManEp, subManEp, routeTable)
 172         //r.addRoute("RIC_SUB_DEL_REQ", ueManEp, subManEp, routeTable)
 173         ////UE Manager -> E2 Termination
 174         //r.addRoute("RIC_CONTROL_REQ", ueManEp, e2TermEp, routeTable)
 175
 176         //Platform Routes --- X2 Routes
 177         //E2 Manager -> E2 Termination
 178         r.addRoute("RIC_X2_SETUP_REQ", e2ManEp, e2TermEp, routeTable, -1)
 179         r.addRoute("RIC_X2_RESET_REQ", e2ManEp, e2TermEp, routeTable, -1)
 180         r.addRoute("RIC_X2_RESET_RESP", e2ManEp, e2TermEp, routeTable, -1)
 181         r.addRoute("RIC_ENDC_X2_SETUP_REQ", e2ManEp, e2TermEp, routeTable, -1)
 182         r.addRoute("RIC_SCTP_CLEAR_ALL", e2ManEp, e2TermEp, routeTable, -1)
 183         r.addRoute("RIC_ENB_CONF_UPDATE_ACK", e2ManEp, e2TermEp, routeTable, -1)
 184         r.addRoute("RIC_ENB_CONF_UPDATE_FAILURE", e2ManEp, e2TermEp, routeTable, -1)
 185         r.addRoute("RIC_ENDC_CONF_UPDATE_ACK", e2ManEp, e2TermEp, routeTable, -1)
 186         r.addRoute("RIC_ENDC_CONF_UPDATE_FAILURE", e2ManEp, e2TermEp, routeTable, -1)
 187         //E2 Termination -> E2 Manager
 188         r.addRoute("E2_TERM_INIT", e2TermEp, e2ManEp, routeTable, -1)
 189         r.addRoute("RIC_X2_SETUP_RESP", e2TermEp, e2ManEp, routeTable, -1)
 190         r.addRoute("RIC_X2_SETUP_FAILURE", e2TermEp, e2ManEp, routeTable, -1)
 191         r.addRoute("RIC_X2_RESET_REQ", e2TermEp, e2ManEp, routeTable, -1)
 192         r.addRoute("RIC_X2_RESET_RESP", e2TermEp, e2ManEp, routeTable, -1)
 193         r.addRoute("RIC_ENDC_X2_SETUP_RESP", e2TermEp, e2ManEp, routeTable, -1)
 194         r.addRoute("RIC_ENDC_X2_SETUP_FAILURE", e2TermEp, e2ManEp, routeTable, -1)
 195         r.addRoute("RIC_ENDC_CONF_UPDATE", e2TermEp, e2ManEp, routeTable, -1)
 196         r.addRoute("RIC_SCTP_CONNECTION_FAILURE", e2TermEp, e2ManEp, routeTable, -1)
 197         r.addRoute("RIC_ERROR_INDICATION", e2TermEp, e2ManEp, routeTable, -1)
 198         r.addRoute("RIC_ENB_CONF_UPDATE", e2TermEp, e2ManEp, routeTable, -1)
 199         r.addRoute("RIC_ENB_LOAD_INFORMATION", e2TermEp, e2ManEp, routeTable, -1)
 200         //E2 Manager -> Resource Status Manager
 201         r.addRoute("RAN_CONNECTED", e2ManEp, rsmEp, routeTable, -1)
 202         r.addRoute("RAN_RESTARTED", e2ManEp, rsmEp, routeTable, -1)
 203         r.addRoute("RAN_RECONFIGURED", e2ManEp, rsmEp, routeTable, -1)
 204         //Resource Status Manager -> E2 Termination
 205         r.addRoute("RIC_RES_STATUS_REQ", rsmEp, e2TermEp, routeTable, -1)
 206         //E2 Termination -> Resource Status Manager
 207         r.addRoute("RIC_RES_STATUS_RESP", e2TermEp, rsmEp, routeTable, -1)
 208         r.addRoute("RIC_RES_STATUS_FAILURE", e2TermEp, rsmEp, routeTable, -1)
 209         //ACxapp -> A1 Mediator
 210         r.addRoute("A1_POLICY_QUERY", nil, a1mediatorEp, routeTable, -1)
 211         r.addRoute("A1_POLICY_RESPONSE", nil, a1mediatorEp, routeTable, -1)
 212 }
 213
 214 func (r *Rpe) generateRouteTable(endPointList rtmgr.Endpoints) *rtmgr.RouteTable {
 215         xapp.Logger.Debug("rpe.generateRouteTable invoked")
 216         xapp.Logger.Debug("Endpoint List:  %v", endPointList)
 217         routeTable := &rtmgr.RouteTable{}
 218         e2TermEp := getEndpointByName(&endPointList, "E2TERM")
 219         if e2TermEp == nil {
 220                 xapp.Logger.Error("Platform component not found: %v", "E2 Termination")
 221                 xapp.Logger.Debug("Endpoints: %v", endPointList)
 222         }
 223         subManEp := getEndpointByName(&endPointList, "SUBMAN")
 224         if subManEp == nil {
 225                 xapp.Logger.Error("Platform component not found: %v", "Subscription Manager")
 226                 xapp.Logger.Debug("Endpoints: %v", endPointList)
 227         }
 228         e2ManEp := getEndpointByName(&endPointList, "E2MAN")
 229         if e2ManEp == nil {
 230                 xapp.Logger.Error("Platform component not found: %v", "E2 Manager")
 231                 xapp.Logger.Debug("Endpoints: %v", endPointList)
 232         }
 233         ueManEp := getEndpointByName(&endPointList, "UEMAN")
 234         if ueManEp == nil {
 235                 xapp.Logger.Error("Platform component not found: %v", "UE Manger")
 236                 xapp.Logger.Debug("Endpoints: %v", endPointList)
 237         }
 238         rsmEp := getEndpointByName(&endPointList, "RSM")
 239         if rsmEp == nil {
 240                 xapp.Logger.Error("Platform component not found: %v", "Resource Status Manager")
 241                 xapp.Logger.Debug("Endpoints: %v", endPointList)
 242         }
 243         A1MediatorEp := getEndpointByName(&endPointList, "A1MEDIATOR")
 244         if A1MediatorEp == nil {
 245                 xapp.Logger.Error("Platform component not found: %v", "A1Mediator")
 246                 xapp.Logger.Debug("Endpoints: %v", endPointList)
 247         }
 248
 249         r.generatePlatformRoutes(e2TermEp, subManEp, e2ManEp, ueManEp, rsmEp, A1MediatorEp, routeTable)
 250
 251         for _, endPoint := range endPointList {
 252                 xapp.Logger.Debug("Endpoint: %v, xAppType: %v", endPoint.Name, endPoint.XAppType)
 253                 if endPoint.XAppType != sbi.PlatformType && (len(endPoint.TxMessages) > 0 || len(endPoint.RxMessages) > 0) {
 254                         r.generateXappRoutes(endPoint, e2TermEp, subManEp, routeTable)
 255                         r.generateSubscriptionRoutes(endPoint, e2TermEp, subManEp, routeTable)
 256                 }
 257         }
 258         return routeTable
 259 }