Skip to content

8. xApp Library In-depth⚓︎

8.1 xApp Library Overview⚓︎

The xApp Library is a library written in Python which is used by developers to implement their own new functionalities and/or services on the dRAX RIC. The xApp Library abstracts the interfaces towards the dRAX RIC, allowing developers to use these interfaces in an easy-to-use way. The xApp Library connects multiple components using its interfaces such as:

  1. The logic as implemented by the developer
  2. The kafka and nats connectors
  3. The REST API
  4. The configuration and settings

8.2 Overview of the xApp interactions with the RIC⚓︎

Once an xApp is deployed into dRAX it can interact with all the components present in the system. Currently, there are two main ways to achieve this: via the dRAX Databus or via the REST APIs. Depending on the use case, a developer can choose which communication method fits their need best. The use of the dRAX Databus is encouraged in scenarios where xApps need to access RAN data, exchange real-time data or issue commands via the Action Taker. The use of the REST APIs is best for non-real-time information gathering or for utilizing the Netconf functionalities exposed by the dRAX REST API Gateway. The xAPP interaction with the cells can happen either via the REST API or via direct Netconf sessions to the corresponding Netconf servers.

8.3 The xApp Structure and Conventions⚓︎

When creating a new Python xApp, we recommend the following structure:

root/
  setup.cfg
  README.md
  requirements.txt
  core/
    restapi.py
    xapp_main.py
    xapp_metadata.json
  config/
    xapp_endpoints.json
    xapp_config.json

This folder structure splits the xApp in three parts:

  • The root folder contains necessary information about the Python packages the xApp is and needs. Since Python usually expects those to be in the base of the tree we recommend leaving them there. This includes the README.md documentation file and requirements.txt to specify python package dependencies.
  • The core folder contains the xApp code (which can be spread over multiple files and subfolders). Here, we also expect the xapp_metadata.json file which is used by the xApp itself and contains the configuration details of the xApp including the default values. As an example, we also include here the restapi.py which contains custom API endpoints exposed by the xApp. Do note the directory of the file building the xapp is used as base by the builder.
  • The config folder contains various configuration files meant to be provided at runtime. This means the folder will usually not exist and not be added to the version tracking system, as its contents are created by the xApp Helm Chart.

An example of this can be found in the example subfolder of the repository, along with some basic usage in example/core/xapp_main.py.

8.4 The xApp Builder⚓︎

For convenience an xApp Builder class has been added. An example of how its used can be found in the _example/core/xapp_main/.py _example xApp in the xApp Framework repository:

builder = xapp_lib.XAppBuilder("..", absolute=False)
    builder.metadata("core/xapp_metadata.json")
    builder.endpoints("config/xapp_endpoints.json")
    builder.config("config/xapp_config.json")
    builder.readme("README.md")
    builder.restapi(
        [
            ("/api/", restapi.MainApiHandler),
            ("/api/actions", restapi.ActionsHandler),
            ("/api/request", restapi.RequestHandler),
        ]
    )
    xapp = builder.build()

This code, assuming it is written in root/core/xapp_main.py, will create an xApp builder using root/ as base (hence the “..” as its first argument). After this, the metadata, which is located in root/core/xapp_metadata.json, will be loaded by the xApp itself. The xapp_metadata.json is a mandatory configuration file of the xApp.

Next, the xapp_endpoints.json and xapp-config.json files can be loaded. Note that these files are created by the xApp Helm Chart as on deployment time configuration options, hence they are located in the root/config folder.

The xApp Builder then can load the README.md documentation file. Finally, if the xApp developer has created custom API endpoints in the root/core/restapi.py file, the APIs can be initiated by the xApp Builder as described above in the example code (providing their API endpoint and callback function).

Do note that the builder only registers paths and configuration, and thus any errors about missing files etc will only be thrown during the builder.build() call.

8.5 dRAX RIC Databus⚓︎

The data from the dRAX RIC Databus comes in through the Kafka object, which automatically takes care of properly setting up and configuring a Kafka client. Note that this object is a singleton created and owned by the xApp library object itself, and calling the .kafka() method multiple times will result in the same object being returned every time, which is why the following code works:

xapp.kafka().subscribe(["topic"])
(topic, data) = xapp.kafka().recv_data()

The URL of the dRAX RIC Databus (Kafka in this case) is automatically generated in the xapp_endpoints.json file by the xApp Helm Chart. As seen, this config file is loaded into the xapp using:

builder.endpoints("config/xapp_endpoints.json")

By default, kafka will use the value associated with the “KAFKA_URL” key inside the xapp_endpoints.json. However, the developer can choose to add additional keys in the xapp_endpoints.json file through the xApp Helm Chart, which is discussed in the xApp Endpoints section. The xApp itself can then decide to use a different key by calling:

(topic, data) = xapp.kafka(name="KAFKA_URL_2").recv_data()

A simple way to check what messages are received is to simply log everything coming in from the kafka object. This is shown in Example 1 in the example/core/xapp_main.py from the xApp Framework repository and in the test_kafka_producer unittest.

### Example 1: Just logging all the messages received from the dRAX Databus
logging.info("Received message from dRAX Databus!")
logging.debug("dRAX Databus message on {topic}: {data}".format(topic=topic, data=data))

The data on the dRAX RIC Databus (Kafka) uses the JSON format. In order to decode json messages the xApp Library provides a decorator created with .json() that automatically transforms serialized data into a dictionary.

(topic, data) = xapp.kafka().json().recv_message()

The 4G and 5G RAN data that is available on the dRAX Databus is described in a separate document that can be accessed via the following link:

https://docs.google.com/document/u/1/d/e/2PACX-1vTmZjYiThXivi03Td_1LXjQVldAO_7AXqnVwc0Qe-uFCT16tKH6EBF2OTOhiZ0SFw/pub

8.6 dRAX NATS Databus⚓︎

Connecting to the dRAX NATS Databus is nearly identical to connecting to the dRAX RIC Databus. The only difference is the xApp method call and the use of NATS_URL instead of KAFKA_URL:

xapp.nats().subscribe(["topic"])
(topic, data) = xapp.nats().recv_data()

8.7 dRAX Commands⚓︎

From the xApp you can issue two types of commands on the dRAX NATS Databus:

  1. 4G LTE Handover command

  2. 5G NR Handover command

These are both defined inside the xapp_lib.actions module.

NOTE: As the commands are sent over the dRAX NATS Databus, the NATS client should be set up in the xApp as described in the dRAX NATS Databus section.

8.7.1 4G LTE Handover command⚓︎

dRAX enables you to issue a handover command for a set of UEs in an LTE environment from source cells to target cells. In the example/core/xapp_main.py example of the xApp Framework we show this in Example 4a.

### Example 4a: Send handover command (LTE)
handover_list = [
    {'ueIdx': 'ueRicId_to_handover_1', 'targetCell': 'Cell_1', 'sourceCell': 'Cell_2'},
    {'ueIdx': 'ueRicId_to_handover_2', 'targetCell': 'Cell_2', 'sourceCell': 'Cell_1'}
]
actions.trigger_handover(xapp.nats(), handover_list)

The actions.trigger_handover function abstracts the lower-level details of how a handover command message is generated and sent.

The function takes a list of handovers that are required. Each handover consists of the following information :

  • 'ueIdx': This is the ueDraxId of the UE that should be handed over;
  • 'targetCell': The target cell to which the UE should get handed over to;
  • 'servingCell': The UEs’ current serving cell.

The ueIdx and servingCell values are determined from the UE Measurement messages received on the accelleran.drax.4g.ric.raw.ue_measurements Kafka Topic, and the relevant portion is shown below.

...
"Rc4gUeMeasurement": {
    "UniqueRicId": <string>,
    "CellId": <string>,
    "AttachedCellId": <string>,
    "Rsrp": <uint>,
    "Rsrq": <uint>,
...

ueIdx is UniqueRicId and the servingCell is AttachedCellId

The potential target cells should be collected from L2 Stats messages also received on the accelleran.drax.4g.ric.raw.ue_measurements Kafka Topic. These messages are in the format :

{
  "ENB": <string>,
  "CELL": <string>,
  "UE": <string>,
  "ueDraxId": <string>,
  "ueRicId": <string>,
  "EnbStatsL2StatsValues": {
    "Report": {
      "DschThroughput": <uint>,
      "UschThroughput": <uint>,
      "DownlinkBler": <uint>,
      "UplinkBler": <uint>,
      "TimingAdvance": <uint>,
      "UlSinr": <uint>
    }
  },
  "tlpublishTime__": <string>,
  "timestamp": <int>
}

The UniqueRicId field from a UEMeasurement can be matched with the ueDraxId field from a L2 Stats message. Then the cell along with its information relating to the UE can be matched using the CELL field from the corresponding L2 Stats message.

8.7.2 5G NR Handover command and response⚓︎

Supported CU Versions

This command is supported since CU 4.2, used by default from RIC 6.5.0.

dRAX also enables you to issue a handover command of a specific UE in an NR environment to a target cell. It is also possible to check the response from the handover command. In the example/core/xapp_main.py example of the xApp Framework we show this in Examples 4b and 4c:

### Example 4b: Send handover command (5G NR)
ue_id = 7
target_cell = actions.Cell(id=456, plmn='654321')
cucp_id = "cucp-1"

transaction_id = actions.trigger_handover_5g(xapp=xapp,
                                      ue_id=ue_id,
                                      target_cell=target_cell,
                                      cucp_id=cucp_id)

The actions.trigger_handover_5g function abstracts the lower-level details of how a handover command message is generated and sent.

The parameters required are as follows :

  • 'xapp': This a reference to the xapp triggering the handover
  • 'ue_id': This is the Id of the UE that should be handed over;
  • 'target_cell': the target cell to which the UE should be handed over to (the cell is composed of the cell id and the plmn)
  • 'cucp_id': the cucp instance id that is responsible for coordinating the handover procedure and supervising the control of the serving cell

Both the ue_id, target_cell and cucp_id are determined from UE Measurement messages recieved on the accelleran.drax.5g.ric.raw.ue_measurement Kafka Topic. The cucp_id can be extracted from the "topic" field of the message by parsing the string and identifying the first segment before the first dot (e.g. "cucp_id".5G_MEAS_INFO...).

The ue_id is taken from the GnbCuCpUeId field in the RrcMeasurementReportResult portion of a UE Measurement message as shown below:

...
"RrcMeasurementReportResultInfo": {
    "GnbCuCpUeId": <uint>,
    "Timestamp": <string>,
    "ServingCellInfo": {
      "NrCgi": {
        "NrCellId": <string>,
        "PlmnId": {
          "items": {
            "0": <uint>
          }
        }
      }
    ...
  }
},
"topic":"<string>.5G_MEAS_INFO.ENB=<string>.DU=<string>.CELL=<uint>.UE=<string>"

It is the responsibility of the xApp application to determine the target_cell required for the handover. Potential target cells for a UE can be determined from the Neighbour Cells listed in the following portion of a UE Measurement message.

...
"NumberOfIncludedCells": <uint>,
    "CellInfo": {
      "0": {
        "PhyCellId": <uint>,
        "NeighbourCellInfo": {
          "NrCgi": {
            "NrCellId": <string>,
            "PlmnId": {
              "items": {
                "0": <uint>
              }
            }
          },
...

Handover responses are received on the Kafka topic accelleran.drax.5g.ric.raw.ran_control_response', therefore you will also need to subscribe to this Kafka topic in the subscribe_to_topics function. The reponse will be in the following format: 

{
  "RIC_REQUESTOR": <string>,
  "RIC_INSTANCE": <string>,
  "E2apRanControlRsp": {
    "RicRequestorId": <uint>,
    "RicInstanceId": <uint>,
    "RicControlActionId": <uint>,
    "Timestamp": <string>
  },
  "tlpublishTime__": <string>,
  "spanContext__": <string>
}

Example 4c shows a handover response being received, and the data being parsed using the actions.parse_ran_control_response(data) function.

### Example 4c: Receive handover response
(topic, data) = xapp.kafka().json().recv_message()
if data and 'accelleran.drax.5g.ric.raw.ran_control_response' in topic:
    timestamp, transaction_id, requestor_id = actions.parse_ran_control_response(data)

The actions.parse_ran_control_response function will return the time the message was generated, the requestor_id (which is the xapp.id), and also the transaction_id so the response can be matched with the transaction_id returned by the actions.trigger_handover function from Example 4b.

8.8 Publish to dRAX RIC Databus⚓︎

From your xApp, you can publish data on the dRAX RIC Databus for other xApps to use as well.

8.8.1 Event-based publishing⚓︎

You can publish certain event-based data to the dRAX RIC Databus. This is shown in Example 2 in the processor.py of the xApp Core:

8.8.2 Example 2: Publishing one time or event-based data on the dRAX Databus⚓︎

We will publish on topic "my_test_topic", and just republish the "data" data

xapp.kafka().json().send_message("my_test_topic", data)

We have created an abstraction which takes the following parameters:

  1. "my_test_topic": This is the topic on the dRAX RIC Databus to which the data should be published
  2. data: the actual data that needs to be published, it should be of a type that can be serialized as the dRAX RIC Databus expects JSON format messages (hence the xapp.kafka().json()... decorator).

8.8.3 Periodic publishing⚓︎

If one wants to periodically publish information, they can spawn a separate thread using the standard Python threading.Thread function

data = {"key": "data"}
def periodic_publish(xapp):
    while True:
        xapp.kafka().json().send_message("my_test_topic", data)
        sleep(1)

publish_thread = threading.Thread(name="PeriodicPublish", target=periodic_publish, args=(xapp,))
publish_thread.start()

8.8.4 Publish to the dRAX NATS Databus⚓︎

Publishing to the dRAX NATS Databus is equally simple:

data = {"key": "data"}
def periodic_publish(xapp):
    while True:
        xapp.nats().json().send_message("my_test_topic", data)
        sleep(1)

publish_thread = threading.Thread(name="PeriodicPublish", target=periodic_publish, args=(xapp,))
publish_thread.start()

8.9 Use the dRAX RIC API Gateway⚓︎

The dRAX RIC exposes a number of APIs through the dRAX RIC API Gateway. The full documentation on all of the endpoints is available on the dRAX RIC API Gateway Swagerhub, which is exposed on the following URL:

http://<kubernets_ip>:31315/api/v1/docs

Just substitute with the advertise address of your Kubernetes cluster.

Inside the actions module of the xApp Library, we have already implemented the code for making API calls, and abstracted the details on how to connect to the API. That information is stored in the xapp_endpoints.json file (see also section on xApp Configuration In-depth under API_GATEWAY_URL. We created a few examples of how this works in the xApp Example, while the SwagerHub contains detailed information and description of all the endpoints.

In summary, through the dRAX RIC API Gateway you can:

  • xApps
    • Get a list of deployed xApps
    • Get/Modify the configuration of another xApp
    • Check if the xApp is healthy
    • Deploy or delete other xApps
  • 4G Radio Controller
    • Check the status of the 4GRC
    • Configure the 4GRC
  • 4G Cells
    • Configure basic parameters of a cell using a JSON
    • Send a full NetConf RPC through the API to configure the cell
    • Upload a pre-existing configuration
    • Auto-configure cells
    • Reboot cells
  • 5G CU
    • Programmatically deploy the 5G CU-CP and CU-UP
    • Configure the 5G CU-CP and CU-UP

8.9.1 How to use the dRAX RIC API Gateway in general⚓︎

Example 6 shows how to call the API gateway on the following endpoint:

/discover/services/netconf

This endpoint, as described in the SwagerHub, returns the list of NetConf services deployed on dRAX. Since the Accelleran 4G small cells and the 5G CUs have NetConf servers running in them to be able to use NetConf to configure them, this way we can get that information.

### Example 6: How to use the dRAX RIC API Gateway
endpoint = '/discover/services/netconf'
api_response = requests.get(actions.create_api_url(xapp, endpoint))

if api_response.status_code == 200:
  try:
    logging.info(api_response.json())
  except:
    logging.info('Failed to load JSON, showing raw content of API response:')
    logging.info(api_response.text)

8.9.2 How to select only the port at which NetConf is exposed⚓︎

Example 7 shows how to parse the retrieved information from the dRAX RIC API Gateway. Because the information retrieved can be parsed as JSON, in python we can simply use it as a Python dictionary:

### Example 7: Get the ports where the netconf servers of cells are exposed
        endpoint = '/discover/services/netconf'
        api_response = requests.get(actions.create_api_url(xapp, endpoint))

        for cell, cell_info in api_response.json().items():
            for port in cell_info['spec']['ports']:
                if port['name'] == 'netconf-port':
                    logging.info('Cell [{cell}] has NETCONF exposed on port [{port}]'.format(
                        cell=cell,
                        port=port['node_port'])
                    )

This would log the following example data:

[netconf-dageraadplats] has NETCONF exposed on port [30023]
[netconf-parking] has NETCONF exposed on port [32634]
Back to top