Wallaroo provides the ability to publish a pipelines, which automatically packages pipelines with their models and models steps and publish the containerized pipelines with the inference engine to an Open Container Initiative (OCI) Registry as a pipeline publish.

Edge locations are added to pipeline publishes associated with the pipeline, and provides:

• Authentication credentials to allow edge deployments to connect back to the Wallaroo Ops instance.
• Inference results from the edge deployments are stored in the local configured storage with configurable storage size in a “first in first out” basis; the oldest inference results are deleted first. These records are stored locally even when connection to the Wallaroo Ops instance is lost.
• Edge locations store their inference logs locally through a persistent volume.
  • These logs are retrieved through the edge deployments /logs endpoint.
  • When a connection is not available from a Wallaroo Ops instance, or an “air gap” deployment, the logs stored in the local edge deployments are still available through the /logs endpoint, and are transmitted bck to the Wallaroo Ops instance when connectivity is restored.
  • Logs are stored in a First In-First Out method. When the storage capacity is reached, the oldest logs are deleted first.
• When the edge location is connected with the Wallaroo instance, inference result logs are transmitted back to the Wallaroo Ops instance and added to the pipeline the edge location is associated with, based on the DateTime stamp of when the inference occurred, not when the logs were transmitted back to the Wallaroo ops instance.
  • This allows operations such as model drift detection with assays and other observability operations to sync inference times across multiple locations.
  • Pipeline logs store the inference location in the field metadata.partition labeled with the edge location’s name. For example, inference requests to the the edge location named us-west1 would have its inference request logs stored in the pipeline logs metadata.partition as us-west1.
• Pipeline publishes replaced from new pipeline versions or entirely new pipelines allocate the edge locations associated with the pipeline to the new pipeline. Previous inference results and logs remain with the previous pipeline, and new inference records are added to the pipeline the edge locations are now associated with.

Edge Observability via the Wallaroo SDK

When connectivity is available, edge inference logs are synced with the pipeline they are referenced from. For more details on adding an edge location to a pipeline publish, see How to Publish and Deploy AI Workloads for For Edge and Multicloud Model Deployments.

Retrieve Edge Location Inference Logs from a Pipeline via the Wallaroo SDK

Inference logs from an edge location can be available for analysis in the Wallaroo Model Ops Center pipeline are retrieved by one of the two methods:

• wallaroo.pipeline.Pipeline.logs: Returns the pipeline inference logs.
• wallaroo.pipeline.Pipeline.export_logs: Creates a subdirectory and JSON files in pandas Record Format of the pipeline inference logs.

The field metadata.partition displays which location the inference was performed through. This is returned with the dataset parameter and requesting the metadata dataset for each pipeline log request. For example:

pipeline.log(dataset=["time", "in", "out", "metadata"]) # returns the time, in, out and metadata datasets
pipeline.export_logs(dataset=["time", "in", "out", "metadata"]) # returns the time, in, out and metadata datasets

Timeout Issues

If latency between the model’s edge deployment and the Wallaroo Ops Instance is higher than 0.5, the connection may time out and the inference results are not transmitted back to the Wallaroo Ops Center. To check the latency between the deployed model and the Wallaroo Ops Center, use the following command from the edge device:

curl -o /dev/null -s -w "%{time_total}\n" {Wallaroo Domain}:8443

For example, if the Wallaroo Domain of the Wallaroo Ops Center is example.wallaroo.ai, the command is:

curl -o /dev/null -s -w "%{time_total}\n" example.wallaroo.ai:8443
0.089316

For docker run deployments, increase the timeout with the option -e MTLS_CONNECTION_TIMEOUT=3.0. Increase the MTLS_CONNECTION_TIMEOUT further as needed for extreme low latency scenarios.

For example:

docker run -p 8080:8080 \
-v ./data:/persist \ 
-e DEBUG=true \
-e MTLS_CONNECTION_TIMEOUT=3.0 \ 
-e OCI_REGISTRY=$OCI_REGISTRY \
-e EDGE_BUNDLE=abc123 \
-e CONFIG_CPUS=1 \
-e OCI_USERNAME=$OCI_USERNAME \
-e OCI_PASSWORD=$OCI_PASSWORD \
-e PIPELINE_URL=example.registry.io/wallaroolabs/doc-samples/pipelines/edge-observability-pipeline:3b49fba8-94d8-4fca-aecc-c75257fd16c6 \
example.registry.io/wallaroolabs/doc-samples/engines/proxy/wallaroo/ghcr.io/wallaroolabs/standalone-mini:v2023.4.0-main-4079

Airgapped Edge Observability from Edge Deployments

In scenarios where connectivity is limited or unavailable for centralized Edge observability, Edge model deployments with Wallaroo can be configured to store inference logs locally in a “first in, first deleted” format: the oldest inference request logs are the first deleted when storage capacity runs out. Inference logs stored at the edge will be available in the model operations center, once connectivity is restored, following an automated retry mechanism.

Edge Observability Storage Configuration

Edge location inference log storage capacity is set with the LOCAL_INFERENCE_STORAGE and PLATEAU_PAGE_SIZE.

• LOCAL_INFERENCE_STORAGE (Optional): Sets amount of storage to allocate for the edge deployments inference log storage capacity. This is in the format {size as number}{unit value}. If used, must be used with PLATEAU_PAGE_SIZE. The values are similar to the Kubernetes memory resource units format. This is configurable based on the amount of storage needed during low/no connectivity periods before connectivity is restored for logs to be available in the Wallaroo model Ops center. The accepted unit values are:
  • Ki (for KiloBytes)
  • Mi (for MegaBytes)
  • Gi (for GigaBytes)
  • Ti (for TeraBytes)
• PLATEAU_PAGE_SIZE (Optional): How many inference log rows to upload from the edge deployment at a time. Must be used with LOCAL_INFERENCE_STORAGE.

The following example shows a standard deployment with the edge observability configured for 512Mi of log storage.

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 \
    -e PLATEAU_PAGE_SIZE=100 \
    -e PIPELINE_URL=registry.example.com/pipelines/edge-low-connection-demo:cc4a10a2-4866-4ed6-a1d8-8b731e7bf8f8\
    -e EDGE_BUNDLE=abc123 \
    -e CONFIG_CPUS=4 registry.example.com/engines/proxy/wallaroo/ghcr.io/wallaroolabs/fitzroy-mini:v2024.1.0-main-4963

For full details of model deployment in edge locations, see How to Publish and Deploy AI Workloads for For Edge/Multicloud Model Deployments.

Log Retrieval from Edge Locations

Inference logs are retrieved from edge location deployments through the /logs endpoint.

• Endpoint: /logs
• Type: POST
• Headers:
  • Content-Type: application/json: Submissions to the /logs endpoint in JSON text format.
  • Accept: application/json; format=pandas-records: The /logs endpoint returns JSON in pandas Record format.
• Parameters:
  • {}: An empty set.

Inference logs are returned as JSON in pandas Record Format with the following fields:

Examples

Edge Observability Low Connectivity Example

The following example demonstrates the following:

• From an edge location model deployment:
  • Perform inference requests and retrieving logs from both the edge location and the Wallaroo Ops center.
  • Simulate a disconnection between the edge location and the Wallaroo Ops center and showing how inference requests logs are available in the edge location.
  • Simulate a reconnection between the edge location and the Wallaroo Ops center and showing how inference request logs are added to the Wallaroo Ops instance.

For this example, we have deployed a house price model to an edge location. We perform an inference request and submit JSON in pandas Record Format.

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 Ops 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 Ops 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 Ops 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

Tutorials

• Edge Observability with No/Low Connection Tutorial