Streaming Events

Polling the API for events can be time-consuming and is only as reliable as your polling script. As an alternative, you can use our streaming event API:
https://docs.packetfabric.com/api/v2/redoc/#tag/Streaming-Events-Beta

You can also stream events using the PacketFabric Terraform provider: https://registry.terraform.io/providers/PacketFabric/packetfabric/latest/docs/resources/packetfabric_streaming_events

This API uses server-sent events to push events to you without using client-side polling. Streaming events are also more easily consumed programmatically, with less effort and overhead than polling solutions.

For example, the following shows a stream that begins with a user disabling and then enabling a port, and then proceeds to display errors related to the port metrics (the port used in this example was a lab port named “boop”):

Animated gif of the stream python script

The example above used the following script:

import asyncio
import datetime
import requests
import sys

BASE_URL = "https://api.packetfabric.com"

LOGIN = f"{BASE_URL}/v2/auth/login"
LOGOUT = f"{BASE_URL}/v2/auth/logout"
EVENTS_BASE = f"{BASE_URL}/v2/events"


def login(username, password):
    print("Logging in...")
    resp = requests.post(LOGIN, json={"login": username, "password": password})
    print(f"Login response: {resp.json()}")
    return resp.json()["token"]


def logout(token):
    headers = {"Authorization": f"Bearer {token}"}
    requests.get(LOGOUT, headers=headers)
    print("Logged out")


def create_subscription(token):
    url = EVENTS_BASE
    headers = {"Authorization": f"Bearer {token}"}
    body = {
        "streams": [
           # {
           #     "type": "vc",
                # "vcs": ["PF-BC-MIA-LAS-1234567-PF&PF-AP-LAS1-987654"],
                # "events": ["metrics"]
           # },
            {
                "type": "customer",
                # "events": ["auth", "physical_interface"]
            },
             {
                 "type": "port",
            #     "events": ["etherstats"],
                 "ifds": ["PF-AP-LAB6-2201365"]
             },

        ]
    }
    resp = requests.post(url, json=body, headers=headers)
    try:
        return resp.json()["subscription_uuid"]
    except KeyError:
        print("Error creating subscription")
        print(resp.content)
        sys.exit()


async def consume_simple(token, subscription_uuid):
    headers = {"Authorization": f"Bearer {token}", "Accept": "text/event-stream"}
    url = f"{EVENTS_BASE}/{subscription_uuid}"
    r = requests.get(url, stream=True, headers=headers)
    for line in r.iter_lines():
        decoded_line = line.decode('utf-8')
        print(decoded_line)


if __name__ == "__main__":
    username = "caitlin.technicalwriter@packetfabric.com"
    password = "ILoveKn0wledgebases!!"
    token = login(username, password)
    subscription_uuid = create_subscription(token)
    print(f"Subscription UUID: {subscription_uuid}")
    _ = input("Continue?")
    try:
        print(f"Starting stream at {datetime.datetime.utcnow().isoformat()}")
        asyncio.run(consume_simple(token, subscription_uuid))
    finally:
        print(f"Ending stream at {datetime.datetime.utcnow().isoformat()}")
        logout(token)

Updated on 26 Apr 2023