Pipeline logs are retrieved through the Wallaroo MLOps API with the following request.

Get Pipeline Logs

Pipeline logs are retrieved through the Wallaroo MLOps API with the following request.

• REQUEST URL
  • v1/api/pipelines/get_logs
• Headers
  • Accept:
    • application/json; format=pandas-records: For the logs returned as pandas DataFrame
    • application/vnd.apache.arrow.file: for the logs returned as Apache Arrow
• PARAMETERS
  • pipeline_name (String Required): The name of the pipeline.
  • workspace_id (Integer Required): The numerical identifier of the workspace.
  • cursor (String Optional): Cursor returned with a previous page of results from a pipeline log request, used to retrieve the next page of information.
  • order (String Optional Default: Desc): The order for log inserts returned. Valid values are:
    • Asc: In chronological order of inserts.
    • Desc: In reverse chronological order of inserts.
  • page_size (Integer Optional Default: 1000.): Max records per page.
  • start_time (String Optional): The start time of the period to retrieve logs for in RFC 3339 format for DateTime. Must be combined with end_time.
  • end_time (String Optional): The end time of the period to retrieve logs for in RFC 3339 format for DateTime. Must be combined with start_time.
• RETURNS
  • The logs are returned by default as 'application/json; format=pandas-records' format. To request the logs as Apache Arrow tables, set the submission header Accept to application/vnd.apache.arrow.file.
  • Headers:
    • x-iteration-cursor: Used to retrieve the next page of results. This is not included if x-iteration-status is All.
    • x-iteration-status: Informs whether there are more records available outside of this log request parameters.
      • All: This page includes all logs available from this request. If x-iteration-status is All, then x-iteration-cursor is not provided.
      • SchemaChange: A change in the log schema caused by actions such as pipeline version, etc.
      • RecordLimited: The number of records exceeded from the page size, more records can be requested as the next page. There may be more records available to retrieve OR the record limit was reached for this request even if no more records are available in next cursor request.
      • ByteLimited: The number of records exceeded the pipeline log limit which is around 100K.

Get Pipeline Logs Example

For our example, we will retrieve the pipeline logs. FIrst by specifying the date and time, then we will request the logs and continue to show them as long as the cursor has another log to display. Because of the size of the input and outputs, most logs may be constrained by the x-iteration-status as ByteLimited.

# retrieve the authorization token
headers = wl.auth.auth_header()

url = f"{wl.api_endpoint}/v1/api/pipelines/get_logs"

# Standard log retrieval

data = {
    'pipeline_name': main_pipeline_name,
    'workspace_id': workspace_id,
    'start_time': dataframe_start.isoformat(),
    'end_time': dataframe_end.isoformat()
}

response = requests.post(url, headers=headers, json=data)
standard_logs = pd.DataFrame.from_records(response.json())

display(len(standard_logs))
display(standard_logs.loc[:, ["time", "out"]])

1

# retrieve the authorization token
headers = wl.auth.auth_header()

url = f"{wl.api_endpoint}/v1/api/pipelines/get_logs"

# datetime set back one day to get more values

data = {
    'pipeline_name': main_pipeline_name,
    'workspace_id': workspace_id
}

response = requests.post(url, headers=headers, json=data)
standard_logs = pd.DataFrame.from_records(response.json())

display(standard_logs.loc[:, ["time", "out"]])
cursor = response.headers['x-iteration-cursor']

# if there's another record, get the next one

while 'x-iteration-cursor' in response.headers:
    # retrieve the authorization token
    headers = wl.auth.auth_header()

    url = f"{wl.api_endpoint}/v1/api/pipelines/get_logs"

    # datetime set back one day to get more values

    data = {
        'pipeline_name': main_pipeline_name,
        'workspace_id': workspace_id,
        'cursor': response.headers['x-iteration-cursor']
    }

    response = requests.post(url, headers=headers, json=data)
    # if there's no response, the logs are done
    if response.json() != []:
        standard_logs = pd.DataFrame.from_records(response.json())
        display(standard_logs.head(5).loc[:, ["time", "out"]])

Pipeline Log Storage

Pipeline logs have a set allocation of storage space and data requirements.

Pipeline Log Storage Warnings

To prevent storage and performance issues, inference result data may be dropped from pipeline logs by the following standards:

• Columns are progressively removed from the row starting with the largest input data size and working to the smallest, then the same for outputs.

For example, Computer Vision ML Models typically have large inputs and output values - a single pandas DataFrame inference request may be over 13 MB in size, and the inference results nearly as large. To prevent pipeline log storage issues, the input may be dropped from the pipeline logs, and if additional space is needed, the inference outputs would follow. The time column is preserved.

If a pipeline has dropped columns for space purposes, this will be displayed when a log request is made with the following warning, with {columns} replaced with the dropped columns.

The inference log is above the allowable limit and the following columns may have been suppressed for various rows in the logs: {columns}. To review the dropped columns for an individual inference’s suppressed data, include dataset=["metadata"] in the log request.

Review Dropped Columns

To review what columns are dropped from pipeline logs for storage reasons, include the dataset metadata in the request to view the column metadata.dropped. This metadata field displays a List of any columns dropped from the pipeline logs.

For example:

# retrieve the authorization token
headers = wl.auth.auth_header()

url = f"{APIURL}/v1/api/pipelines/get_logs"

# Standard log retrieval

data = {
    'pipeline_name': main_pipeline_name,
    'workspace_id': workspace_id
}

response = requests.post(url, headers=headers, json=data)
standard_logs = pd.DataFrame.from_records(response.json())

display(len(standard_logs))
display(standard_logs.head(5).loc[:, ["time", "metadata"]])
cursor = response.headers['x-iteration-cursor']

Suppressed Data Elements

Data elements that do not fit the supported data types below, such as None or Null values, are not supported in pipeline logs. When present, undefined data will be written in the place of the null value, typically zeroes. Any null list values will present an empty list.