This tutorial and the assets can be downloaded as part of the Wallaroo Tutorials repository.

Airgapped Edge Observability with No/Low Connection Tutorial

Wallaroo pipeline publishes are packaged versions of the pipeline, models, and the inference engine published to an Open Container Initiative (OCI) Registry. Once published, they are used to deploy models to edge devices and used for inference requests.

Edge locations added to a pipeline published allow deployments on edge devices to connect with the Wallaroo Model Ops Center instance and transmit their inference results as part of the pipeline logs.

Edge Location Observability

Inference request logs are stored locally in the edge device. This storage space is configurable during model deployment on the edge devices via the LOCAL_INFERENCE_STORAGE flag. Inference logs are stored on the edge devices until the assigned storage capacity is reached. At that point, logs are pruned on a “FIFO” - First In First Out method, where the first inference logs stored are the one pruned first.

In conditions where the edge device is unable to connect with the Wallaroo Model Ops Center instance for an extended period of time, logs are stored until connection is restored to the Wallaroo Model Ops Center instance. When the connection is restored, logs are transmitted back to the Wallaroo Model Ops Center instance and added to the pipeline where the edge location is assigned to.

Edge inference logs have the inference DateTime is set to the DateTime the inference occurred on the edge device, not when the logs are uploaded to the Wallaroo Model Ops Center. Assays and other features that use inference logs are updated when they perform analyses with any new inference records uploaded by edge devices. This insures that inference log information is synced across different edge devices without concern for when the data was uploaded, just when the inference occurred.

Goals

This tutorial will perform the following:

1. Upload and deploy a sample real estate house price estimation to a Wallaroo Model Ops Center instance.
2. Publish the pipeline and deploy to an edge device.
3. Perform sample inferences on the edge devices.
   1. Show inference logs from the local edge device.
   2. Show inference logs submitted to the Wallaroo Model Ops Center instance from the edge device.
4. Simulate a loss of communication on the Wallaroo Model Ops Center instance from the edge device. Perform additional sample inferences on the edge device.
   1. Show new inference logs on the edge device.
   2. Show new inference logs are not yet transmitted to the Wallaroo Model Ops Center instance.
5. Simulate a restoration of communication on the Wallaroo Model Ops Center instance and show the inference logs from the edge device are restored to the Wallaroo Model Ops Center instance.

Resources

This tutorial provides the following:

• Models:
  • ./models/rf_model.onnx: The champion model that predicts house prices.
• Data:
  • ./data/normal-inputs.df.json: A set of inference requests across a span of different house types and values.
  • ./data/smallinputs.df.json: A set of inference requests that tend to generate house price predictions with low values.
  • ./data/biginputs.df.json: A set of inference requests that tend to generate house price predictions with high values.

Prerequisites

• A deployed Wallaroo instance with Edge Registry Services and Airgapped Edge Observability enabled.
• The following Python libraries installed:
  • wallaroo: The Wallaroo SDK. Included with the Wallaroo JupyterHub service by default.
  • pandas: Pandas, mainly used for Pandas DataFrame
• A X64 Docker deployment to deploy the model on an edge location.

Edge Deployment Preparation Steps

The following is used to set up the environment that is used for the pipeline publish replacement and in-line model updates on an edge tutorial.

In these steps, we will:

• Deploying a sample ML model used to determine house prices based on a set of input parameters.
• Publish the model deployment configuration to an OCI registry.
• Use the publish and set edge locations.
• Deploy the model in an edge location and perform sample inferences.

Import Libraries

The first step will be to import our libraries, and set variables used through this tutorial.

import wallaroo
from wallaroo.object import EntityNotFoundError
from wallaroo.framework import Framework

from IPython.display import display

# used to display DataFrame information without truncating
from IPython.display import display
import pandas as pd
pd.set_option('display.max_colwidth', None)

import json
import time

workspace_name = 'edge-low-connection-demonstration'

# ignoring warnings for demonstration
import warnings
warnings.filterwarnings('ignore')

# used to display DataFrame information without truncating
from IPython.display import display
import pandas as pd
pd.set_option('display.max_colwidth', None)

Connect to the Wallaroo Instance

The first step is to connect to Wallaroo through the Wallaroo client. The Python library is included in the Wallaroo install and available through the Jupyter Hub interface provided with your Wallaroo environment.

This is accomplished using the wallaroo.Client() command, which provides a URL to grant the SDK permission to your specific Wallaroo environment. When displayed, enter the URL into a browser and confirm permissions. Store the connection into a variable that can be referenced later.

If logging into the Wallaroo instance through the internal JupyterHub service, use wl = wallaroo.Client(). For more information on Wallaroo Client settings, see the Client Connection guide.

# Login through local Wallaroo instance

wl = wallaroo.Client()

Create Workspace

We will create a workspace to manage our pipeline and models. The following variables will set the name of our sample workspace then set it as the current workspace.

Workspace names are unique across the Wallaroo instance. If a user requests a workspace they have not been granted access to, an error message is returned.

workspace = wl.get_workspace(name=workspace_name, create_if_not_exist=True)

wl.set_current_workspace(workspace)

{'name': 'edge-low-connection-demonstration', 'id': 12, 'archived': False, 'created_by': '76b893ff-5c30-4f01-bd9e-9579a20fc4ea', 'created_at': '2024-04-23T17:01:17.607894+00:00', 'models': [], 'pipelines': []}

Upload The Champion Model

For our example, we will upload the champion model that has been trained to derive house prices from a variety of inputs. The model file is rf_model.onnx, and is uploaded with the name rf-house-price-estimator.

housing_model_control = (wl.upload_model("rf-house-price-estimator", 
                                        './models/rf_model.onnx', 
                                        framework=Framework.ONNX)
                                        .configure(tensor_fields=["tensor"])
                        )

Build the Pipeline

This pipeline is made to be an example of an existing situation where a model is deployed and being used for inferences in a production environment. We’ll call it edge-inline-replacement-demonstration, set housing_model_control as a pipeline step.

pipeline = wl.build_pipeline('edge-low-connection-demonstration')

# clear the steps if used before
pipeline.clear()

_ = pipeline.add_model_step(housing_model_control)

Edge Deployment

We can now deploy the original pipeline to an edge device. This will require the following steps:

• Publish the pipeline: Publishes the pipeline to the OCI registry.
• Add Edge: Add the edge location to the pipeline publish.
• Deploy Edge: Deploy the edge device with the edge location settings.

Publish Pipeline

Publishing the pipeline uses the pipeline wallaroo.pipeline.publish() command. This requires that the Wallaroo Model Ops Center instance have Edge Registry Services enabled.

The following publishes the pipeline to the OCI registry and displays the container details. For more information, see Wallaroo SDK Essentials Guide: Pipeline Edge Publication.

#minimum deployment config
deploy_config = wallaroo.DeploymentConfigBuilder() \
                    .replica_count(1) \
                    .cpus(0.5) \
                    .memory("1Gi") \
                    .build()

pub = pipeline.publish(deployment_config = deploy_config)

Add Edge Location

The edge location is added with the wallaroo.pipeline_publish.add_edge(name) method. This returns the OCI registration information, and the EDGE_BUNDLE information. The EDGE_BUNDLE data is a base64 encoded set of parameters for the pipeline that the edge device is associated with, the workspace, and other data.

For full details, see Wallaroo SDK Essentials Guide: Pipeline Edge Publication: Airgapped Edge Observability.

For this example, we will add a location:

• houseprice-edge-inline-demo

edge_name_01 = "houseprice-low-connection-demonstration"
edge_publish_01 = pub.add_edge(edge_name_01)
display(edge_publish_01)

docker run -v $PERSISTENT_VOLUME_DIR:/persist \
    -p $EDGE_PORT:8080 \
    -e OCI_USERNAME=$OCI_USERNAME \
    -e OCI_PASSWORD=$OCI_PASSWORD \
    -e EDGE_BUNDLE=ZXhwb3J0IEJVTkRMRV9WRVJTSU9OPTEKZXhwb3J0IENPTkZJR19DUFVTPTEKZXhwb3J0IEVER0VfTkFNRT1ob3VzZXByaWNlLWxvdy1jb25uZWN0aW9uLWRlbW9uc3RyYXRpb24KZXhwb3J0IE9QU0NFTlRFUl9IT1NUPWRvYy10ZXN0LmVkZ2Uud2FsbGFyb29jb21tdW5pdHkubmluamEKZXhwb3J0IFBJUEVMSU5FX1VSTD1naGNyLmlvL3dhbGxhcm9vbGFicy9kb2Mtc2FtcGxlcy9waXBlbGluZXMvZWRnZS1sb3ctY29ubmVjdGlvbi1kZW1vbnN0cmF0aW9uOjIyNjUwOGM2LTcyMDUtNDg0Ny1hZmUzLTNkMGY2MmM3MzA5NgpleHBvcnQgSk9JTl9UT0tFTj0zOTQ5NzIzYS05MjA2LTQ5ZGEtYWU2Ni00YjZlNjNiYTkyYTMKZXhwb3J0IE9DSV9SRUdJU1RSWT1naGNyLmlv\
    -e PIPELINE_URL=ghcr.io/wallaroolabs/doc-samples/pipelines/edge-low-connection-demonstration:226508c6-7205-4847-afe3-3d0f62c73096 \
    -e CONFIG_CPUS=1 ghcr.io/wallaroolabs/doc-samples/engines/proxy/wallaroo/ghcr.io/wallaroolabs/fitzroy-mini:v2024.1.0-main-4963

helm install --atomic $HELM_INSTALL_NAME \
    oci://ghcr.io/wallaroolabs/doc-samples/charts/edge-low-connection-demonstration \
    --namespace $HELM_INSTALL_NAMESPACE \
    --version 0.0.1-226508c6-7205-4847-afe3-3d0f62c73096 \
    --set ociRegistry.username=$OCI_USERNAME \
    --set ociRegistry.password=$OCI_PASSWORD \
    --set edgeBundle=ZXhwb3J0IEJVTkRMRV9WRVJTSU9OPTEKZXhwb3J0IENPTkZJR19DUFVTPTEKZXhwb3J0IEVER0VfTkFNRT1ob3VzZXByaWNlLWxvdy1jb25uZWN0aW9uLWRlbW9uc3RyYXRpb24KZXhwb3J0IE9QU0NFTlRFUl9IT1NUPWRvYy10ZXN0LmVkZ2Uud2FsbGFyb29jb21tdW5pdHkubmluamEKZXhwb3J0IFBJUEVMSU5FX1VSTD1naGNyLmlvL3dhbGxhcm9vbGFicy9kb2Mtc2FtcGxlcy9waXBlbGluZXMvZWRnZS1sb3ctY29ubmVjdGlvbi1kZW1vbnN0cmF0aW9uOjIyNjUwOGM2LTcyMDUtNDg0Ny1hZmUzLTNkMGY2MmM3MzA5NgpleHBvcnQgSk9JTl9UT0tFTj0zOTQ5NzIzYS05MjA2LTQ5ZGEtYWU2Ni00YjZlNjNiYTkyYTMKZXhwb3J0IE9DSV9SRUdJU1RSWT1naGNyLmlv

DevOps Deployment

The following shows two methods of model deployments on edge devices. For more details on model edge deployments with Wallaroo, see Model Operations: Run Anywhere.

Docker Run Based Deployment

For the model deployment to the edge location via docker run, use the Docker Run Command displayed with the pipeline publish. The following variables must be set first.

• $EDGE_PORT: The external port to connect to the model deployment.
• $OCI_USERNAME: The username for the OCI registry service.
• $OCI_PASSWORD: The password or token for the OCI registry service.

Edge location inference log storage capacity is set with the LOCAL_INFERENCE_STORAGE. This sets amount of storage to allocate for the edge deployments inference log storage capacity. This is in the format {size as number}{unit value}. The accepted unit values are:

• Ki (for KiloBytes)
• Mi (for MegaBytes)
• Gi (for GigaBytes)
• Ti (for TeraBytes)

For example, to set the inference storage capacity to 512Mi, the following command would be used:

docker run -v $PERSISTENT_VOLUME_DIR:/persist \
    -p $EDGE_PORT:8080 \
    -e OCI_USERNAME=$OCI_USERNAME \
    -e OCI_PASSWORD=$OCI_PASSWORD \
    -e LOCAL_INFERENCE_STORAGE=512Mi \ # sets the log storage to 512MB
    -e PIPELINE_URL=ghcr.io/wallaroolabs/doc-samples/pipelines/edge-low-connection-demo:cc4a10a2-4866-4ed6-a1d8-8b731e7bf8f8\
    -e EDGE_BUNDLE=ZXhwb3J0IEJVTkRMRV9WRVJTSU9OPTEKZXhwb3J0IENPTkZJR19DUFVTPTQKZXhwb3J0IEVER0VfTkFNRT1ob3VzZXByaWNlLWxvdy1jb25uZWN0aW9uLWRlbW9uc3RyYXRpb24tMDEKZXhwb3J0IE9QU0NFTlRFUl9IT1NUPWRvYy10ZXN0LmVkZ2Uud2FsbGFyb29jb21tdW5pdHkubmluamEKZXhwb3J0IFBJUEVMSU5FX1VSTD1naGNyLmlvL3dhbGxhcm9vbGFicy9kb2Mtc2FtcGxlcy9waXBlbGluZXMvZWRnZS1sb3ctY29ubmVjdGlvbi1kZW1vOmNjNGExMGEyLTQ4NjYtNGVkNi1hMWQ4LThiNzMxZTdiZjhmOApleHBvcnQgSk9JTl9UT0tFTj1iMmEzZGVlMS0zYjQzLTRhNmQtOGEzMS00MDk1ZjE1YmY5MmIKZXhwb3J0IE9DSV9SRUdJU1RSWT1naGNyLmlv \
    -e CONFIG_CPUS=4 ghcr.io/wallaroolabs/doc-samples/engines/proxy/wallaroo/ghcr.io/wallaroolabs/fitzroy-mini:v2024.1.0-main-4963

Edge Inference Example

We will perform sample inferences on the current model edge deployment to demonstrate how the inferences change as the models are updated through the in-line edge deployments.

With the model deployed on the edge location, we will perform sample inferences on the edge deployment. For this example, the hostname is HOSTNAME. Adjust the hostname to fit your edge location deployment.

The following endpoints are available.

For more details on model edge deployments with Wallaroo, see Model Operations: Run Anywhere.

First we’ll retrieve the model. We should see the same name we set when we uploaded it to the Wallaroo instance.

!curl HOSTNAME:8080/models

{"models":[{"sha":"e22a0831aafd9917f3cc87a15ed267797f80e2afa12ad7d8810ca58f173b8cc6","name":"rf-house-price-estimator","version":"d4287885-3a9a-4229-ae48-d2617d26e7ce","status":"Running"}]}

Now we’ll retrieve the pipeline details, and verify the name is the same as set earlier and it is running.

!curl HOSTNAME:8080/pipelines

{"pipelines":[{"id":"edge-low-connection-demonstration","version":"226508c6-7205-4847-afe3-3d0f62c73096","status":"Running"}]}

Now do the infer with the same DataFrame we used for our sample inference when the model was deployed in the Wallaroo instance.

import json

df = pd.DataFrame.from_records({"tensor": [[4.0, 
                                            2.5, 
                                            2900.0, 
                                            5505.0, 
                                            2.0, 
                                            0.0, 
                                            0.0, 
                                            3.0, 
                                            8.0, 
                                            2900.0, 
                                            0.0, 
                                            47.6063, 
                                            -122.02, 
                                            2970.0, 
                                            5251.0, 
                                            12.0, 
                                            0.0, 
                                            0.0]]})

data = df.to_dict(orient="records")

!curl -X POST HOSTNAME:8080/infer \
    -H "Content-Type: Content-Type: application/json; format=pandas-records" \
    --data '{json.dumps(data)}' > ./inferenceoutput.df.json

# get the dataframe from what we just did

df_result = pd.read_json('./inferenceoutput.df.json', orient="records")
df_result.loc[:, ['time', 'out', 'metadata']]

  % Total    % Received % Xferd  Average Speed   Time    Time     Time  Current
                                 Dload  Upload   Total   Spent    Left  Speed
  0     0    0     0    0     0      0      0 --:--:-- --:--:-- --:--:--     0

100   626  100   498  100   128    755    194 --:--:-- --:--:-- --:--:--   949

pipeline.logs(dataset=['time', 'out', 'metadata'])

Let’s do a request with more records.

import datetime
import time

!curl -X POST http://HOSTNAME:8080/infer \
    -H "Content-Type: Content-Type: application/json; format=pandas-records" \
    --data @./data/smallinputs.df.json > ./inferenceoutput2.df.json

# get the dataframe from what we just did

df_result = pd.read_json('./inferenceoutput2.df.json', orient="records")
df_result.tail(10).loc[:, ['time', 'out', 'metadata']]

  % Total    % Received % Xferd  Average Speed   Time    Time     Time  Current
                                 Dload  Upload   Total   Spent    Left  Speed
  0     0    0     0    0     0      0      0 --:--:-- --:--:-- --:--:--     0

100  624k  100  502k  100  121k   209k  51771  0:00:02  0:00:02 --:--:--  260k

We now show the logs from the edge location via the /logs endpoint.

# local logs - show the last 10 log records to show the updates

!curl -X POST http://HOSTNAME:8080/logs \
    -H "Content-Type: Content-Type: application/json; format=pandas-records" \
    --data '{ }' > ./edge-logs.df.json

df_logs = pd.read_json("./edge-logs.df.json", orient="records")
df_logs.tail(10)

  % Total    % Received % Xferd  Average Speed   Time    Time     Time  Current
                                 Dload  Upload   Total   Spent    Left  Speed
  0     0    0     0    0     0      0      0 --:--:-- --:--:-- --:--:--     0

100  502k  100  502k  100     3   264k      1  0:00:03  0:00:01  0:00:02  264k

Get the most recent logs to verify the edge records were transmitted. We’ll see in the metadata.partition section the name matching our edge location.

pipeline.logs(dataset=['time', 'out', 'metadata'])

Warning: There are more logs available. Please set a larger limit or request a file using export_logs.

100 rows × 7 columns

Simulate Connection Loss

For this example, we’ll simulate a connection loss, then perform another set of inferences, this time from the ./data/biginputs.df.json list. When we retrieve the pipeline logs again from the Wallaroo Model Ops Center instance, we should not see any additional records.

!curl -X POST HOSTNAME:8080/infer \
    -H "Content-Type: Content-Type: application/json; format=pandas-records" \
    --data @./data/biginputs.df.json > ./inferenceoutput2.df.json

# get the dataframe from what we just did

df_result = pd.read_json('./inferenceoutput2.df.json', orient="records")
df_result.tail(10).loc[:, ['time', 'out', 'metadata']]

  % Total    % Received % Xferd  Average Speed   Time    Time     Time  Current
                                 Dload  Upload   Total   Spent    Left  Speed
100  5786  100  4639  100  1147  13402   3313 --:--:-- --:--:-- --:--:-- 16722

These are the logs retrieved from the Wallaroo Model Ops Center center during the connection loss. We can see the new records from the edge deployment have not been added.

pipeline.logs(dataset=['time', 'out', 'metadata'])

Warning: There are more logs available. Please set a larger limit or request a file using export_logs.

100 rows × 7 columns

The logs retrieved from the local Edge location, which is not in communication with the Wallaroo Model Ops Center instance.

# local logs

!curl -X POST http://HOSTNAME:8080/logs \
    -H "Content-Type: Content-Type: application/json; format=pandas-records" \
    --data '{}' > ./edge-logs.df.json

df_logs = pd.read_json("./edge-logs.df.json", orient="records")
df_logs

  % Total    % Received % Xferd  Average Speed   Time    Time     Time  Current
                                 Dload  Upload   Total   Spent    Left  Speed
100  502k  100  502k  100     2   344k      1  0:00:02  0:00:01  0:00:01  344k

1000 rows × 5 columns

We’ll now reenable communication, then wait 60 seconds and check the pipeline inference logs to see if communication is restored.

import time
time.sleep(60)

import json

df = pd.DataFrame.from_records({"tensor": [[4.0, 
                                            2.5, 
                                            2900.0, 
                                            5505.0, 
                                            2.0, 
                                            0.0, 
                                            0.0, 
                                            3.0, 
                                            8.0, 
                                            2900.0, 
                                            0.0, 
                                            47.6063, 
                                            -122.02, 
                                            2970.0, 
                                            5251.0, 
                                            12.0, 
                                            0.0, 
                                            0.0]]})

data = df.to_dict(orient="records")

!curl -X POST HOSTNAME:8080/infer \
    -H "Content-Type: Content-Type: application/json; format=pandas-records" \
    --data '{json.dumps(data)}' > ./inferenceoutput.df.json

# get the dataframe from what we just did

df_result = pd.read_json('./inferenceoutput.df.json', orient="records")
df_result.loc[:, ['time', 'out', 'metadata']]

pipeline.logs(dataset=['time', 'out', 'metadata'])

  % Total    % Received % Xferd  Average Speed   Time    Time     Time  Current
                                 Dload  Upload   Total   Spent    Left  Speed
100   625  100   497  100   128   1242    319 --:--:-- --:--:-- --:--:--  1562

Warning: There are more logs available. Please set a larger limit or request a file using export_logs.

100 rows × 7 columns