Using Streaming API Python Client
HPE Aruba Networking's Central streaming API follows publish–subscribe model. In this model, Central will publish continuous data streams to WebSocket client after the client subscribes to a topic. This approach is different from "polling", where frequent HTTP requests are made to REST API endpoints from the HTTP client to get the real-time information.
Some applications of streaming API include monitoring state of devices (UP/DOWN status), statistics of devices (AppRF statistics, location and presence analytics), Intrusion detection, configuration audit trail and more.
To know the list of available streaming topics, enabling streaming API and to understand the format of the data streams visit the Getting Started with Streaming API section.
How to Install?
- Clone the GitHub repository
git clone https://github.com/aruba/central-python-workflows.git
You can refer to this Git Installation document, if you don't have git installed on your machine.
- Navigate to streaming API client directory within the cloned repository
cd streaming-api-client
- Install the Python packages required for the Streaming API client
pip3 install -r requirements.txt
Link to the GitHub repository.
Package Structure
The dir
structure of streaming api client Python package is as follows
│ README.md
│ simple_app.py
│ wsclient_public.py
│ input.json
│ ...
│
└───lib
│ │ streamingExport.py
│ │ utilities.py
│ │ ...
│
└───proto
│ apprf.proto
│ apprf_pb2.py
│ ...
Two websocket client scripts are available for reference and use.
-
simple_app.py
script provides a simple Websocket client python example. This is useful for someone to understand how to connect with Central via Websocket and decode/deserialize protobuf messages. -
wsclient_public.py
script is an advanced Websocket client sample that can be used as is or further extended to export data to external data sources(usinglib\streamingExport.py
). This script utilizes multi-threading to receive streaming data and establishing Websocket connection. With multi-threading, multiple streaming topics can be subscribed for an Central customer and/or multiple Central customers (in case of MSP) can stream data from a single script execution.
Both the simple_app.py
and wsclient_public.py
utilizes the compiled protobuf files under proto
directory. Be sure to download and compile the latest proto files from Central and place it under the proto
directory.
More information about using the scripts are provided below.
Gathering Requirements
The Python streaming client requires the following
- WebSocket URL Endpoint
- WebSocket Authorization Key
- Subscription Topic
- Central User Email
- ProtoBuf files
1. WebSocket Base URL and Key can be obtained from the Account Home -> Webhooks -> Streaming
Page as show in the image below.
![streaming_api.png 2880](https://files.readme.io/074fdc6-streaming_api.png)
Streaming API Page
2. ProtoBuf files can be downloaded from the same page. Depending on the programming language used, these .proto
files need to be compiled to deserialize the data. Refer the following documentation on using Python ProtoBuf compiler to compile downloaded .proto
files. This will result in "*_pb2.py" for every proto file.
Note
Pre-compiled
.proto
files are available in the proto directory of the GitHub project. The sample scripts uses the files from this directory.For most recent
.proto
definitions, download the files from Central, compile the files and place them in the proto directory.
Script Execution
Simple Streaming Client Application
The provided script subscribes to monitoring topic in the Central Streaming API. The following variables hostname, header["Authorization"], header["UserName"] needs to be updated.
To run the script, enter the command python3 simple_app.py
In the code snippet below,
- MsgProto() and MonitoringInformation() are proto messages from compiled streaming.proto and monitoring.proto files respectively.
- on_message() block converts the data received in serialized protobuf format to python dictionary.
import websocket
def on_message(ws, message):
# Decode Message from serialized proto buffer format
from proto import streaming_pb2
stream_data = streaming_pb2.MsgProto()
stream_data.ParseFromString(message)
print("Timestamp in Epoch: %s" % str(stream_data.timestamp))
print("Customer_ID: %s" % str(stream_data.customer_id))
# Based on the topic import compiled proto file and decode 'data' field
from proto import monitoring_pb2
monitoring_data = monitoring_pb2.MonitoringInformation()
monitoring_data.ParseFromString(stream_data.data)
print(monitoring_data)
if __name__ == "__main__":
# URL for WebSocket Connection from Streaming API page
hostname = "internal-ui.central.arubanetworks.com"
url = "wss://{}/streaming/api".format(hostname)
# Construct Header for WebSocket Connection
header = {}
# Central User email
header["UserName"] = "[email protected]"
# WebSocket Key from Streaming API Page
header["Authorization"] = "XXXXXX"
# Subscription TOPIC for Streaming API
# (audit|apprf|location|monitoring|presence|security)
header["Topic"] = "monitoring"
# Create WebSocket connection
websocket.enableTrace(True)
ws = websocket.WebSocketApp(url=url,
header=header,
on_message = on_message,
on_error = on_error,
on_close = on_close)
ws.on_open = on_open
ws.run_forever()
Advanced Streaming Client Application
This GitHub repository contains usage of advanced Streaming API client.
Features of wsclient_public.py
python streaming client is as follows,
- Option to subscribe data streams for multiple customers
- Option to subscribe to multiple streaming topics for each customer
- Validate the Secure WebSocket Key. When connection is made for the first time,
- If the provided key is expired, the script obtains the working WebSocket key from Central via REST API.
- Updates the input JSON file with working key. Next time when the script is executed, you would have most recent key.
- Decode data streams received in google protobuf format and convert it to python dictionary.
- Provide structure for data transport/export to external apps/storage device. User need to implement the transport/storage logic based on their requirement.
Please Note: This sample script does not attempt to retry when the WebSocket connection is broken in-between data streaming. Validation of WebSocket Key only occurs during initial connection.
Command to execute the script
python3 wsclient_public.py --hostname internal-ui.central.arubanetworks.com --jsoninput input.json --decode_data
To view description of all available arguments, execute the following command
python3 wsclient_public.py --help
The required script arguments are,
--hostname
- The base url obtained from Central Account Home -> Webhooks -> Streaming -> Endpoint. For example from the Endpoint "wss://internal-ui.central.arubanetworks.com/streaming/api", copy only the highlighted section for hostname.
--jsoninput
- Input File where the Central customer information is provided in JSON format. The following json file is the sample input.json.
To view data on screen,
--decode_data
- to print the decoded data streams on screen during script execution.
{
"customers": {
"CustomerA_monitoring": {
"username": "[email protected]",
"wsskey": "xxxx",
"topic": "monitoring"
},
"CustomerA_security": {
"username": "[email protected]",
"wsskey": "xxxx",
"topic": "security"
},
"CustomerB_monitoring": {
"username": "[email protected]",
"wsskey": "YYYY",
"topic": "monitoring"
}
}
}
Handling "bytes" Message Type in Protobuf
The proto messages of type bytes
are base64 encoded in the decoded data. Example message fields of type bytes are MAC Address, IP Address and ESSID. You can know the type of a certain message field from the .proto
files.
To decode the bytes
data field, snippets of python code is provided below
import base64
# Base64 decoding of MAC Address
mac_address = base64.b64decode("<mac-address-value>")
# In some instances, the fields needs to processed further, after base64
# decoding. To get MAC Address in format FF:FF:FF:FF:FF:FF
print(':'.join('%02x' % ord(byte) for byte in mac_address))
# Similarly IP address received as b'\n\x01\ng',
# would translate to 10.1.10.103 after processing
print('.'.join('%d' % byte for byte in ip_address))
Updated about 1 month ago