OPEN-5105: Changes to support monitoring data ingestion without InferencePipelineData objects

Author: Parthib

openlayer/__init__.py

@@ -31,6 +31,7 @@
 from typing import Dict, Optional, Tuple
 
 import pandas as pd
+import urllib.parse
 import yaml
 
 from . import api, constants, exceptions, utils
@@ -882,6 +883,7 @@ def create_inference_pipeline(
  inference_pipeline_config = {
  "name": name or "Production",
  "description": description or "Monitoring production data.",
+ "storageType": api.STORAGE.value,
  }
  inference_pipeline_validator = (
  inference_pipeline_validators.InferencePipelineValidator(
@@ -920,7 +922,7 @@ def create_inference_pipeline(
  {"task_type": task_type.value, **reference_dataset_config}
  )
 
- with tempfile.TemporaryDirectory() as tmp_dir:
+ with tempfile.TemporaryDirectory() as tmp_dir: 
  # Copy relevant files to tmp dir if reference dataset is provided
  if reference_dataset_config_file_path is not None:
  utils.write_yaml(
@@ -1134,21 +1136,31 @@ def publish_batch_data(
  tar.add(tmp_dir, arcname=os.path.basename("batch_data"))
 
  payload = {
+ "performGroundTruthMerge": False,
+ }
+
+ presigned_url_query_params_dict = {
  "earliestTimestamp": int(earliest_timestamp),
  "latestTimestamp": int(latest_timestamp),
- "inferenceIdColumnName": batch_data.get("inferenceIdColumnName"),
- "timestampColumnName": batch_data.get("timestampColumnName"),
- "performGroundTruthMerge": False,
+ "storageInterface": api.STORAGE.value,
+ "dataType": "data",
  }
 
+ presigned_url_query_params = urllib.parse.urlencode(
+ presigned_url_query_params_dict
+ )
+
  self.api.upload(
  endpoint=f"inference-pipelines/{inference_pipeline_id}/data",
  file_path=tar_file_path,
  object_name="tarfile",
  body=payload,
  storage_uri_key="storageUri",
  method="POST",
+ presigned_url_endpoint=f"inference-pipelines/{inference_pipeline_id}/presigned-url",
+ presigned_url_query_params=presigned_url_query_params
  )
+
  print("Batch of data published!")
 
  def _add_default_column(
@@ -1196,18 +1208,30 @@ def publish_ground_truths(
  with tempfile.TemporaryDirectory() as tmp_dir:
  # Copy save files to tmp dir
  df.to_csv(f"{tmp_dir}/dataset.csv", index=False)
+
  payload = {
  "performGroundTruthMerge": True,
  "groundTruthColumnName": ground_truth_column_name,
  "inferenceIdColumnName": inference_id_column_name,
  }
 
+ presigned_url_query_params_dict = {
+ "storageInterface": api.STORAGE.value,
+ "dataType": "groundTruths",
+ }
+
+ presigned_url_query_params = urllib.parse.urlencode(
+ presigned_url_query_params_dict
+ )
+
  self.api.upload(
  endpoint=f"inference-pipelines/{inference_pipeline_id}/data",
  file_path=f"{tmp_dir}/dataset.csv",
  object_name="dataset.csv",
  body=payload,
  storage_uri_key="storageUri",
  method="POST",
+ presigned_url_endpoint=f"inference-pipelines/{inference_pipeline_id}/presigned-url",
+ presigned_url_query_params=presigned_url_query_params
  )
  print("Ground truths published!")

openlayer/api.py

@@ -28,6 +28,7 @@
 from requests_toolbelt import MultipartEncoder, MultipartEncoderMonitor
 from tqdm import tqdm
 from tqdm.utils import CallbackIOWrapper
+import urllib.parse
 
 from . import constants
 from .exceptions import ExceptionMap, OpenlayerException
@@ -189,6 +190,8 @@ def upload(
  body=None,
  method: str = "POST",
  storage_uri_key: str = "storageUri",
+ presigned_url_endpoint: str = "storage/presigned-url",
+ presigned_url_query_params: str = "",
  ):
  """Generic method to upload data to the default storage medium and create the
  appropriate resource in the backend.
@@ -201,13 +204,16 @@ def upload(
  upload = self.upload_blob_azure
  else:
  upload = self.transfer_blob
+
  return upload(
  endpoint=endpoint,
  file_path=file_path,
  object_name=object_name,
  body=body,
  method=method,
  storage_uri_key=storage_uri_key,
+ presigned_url_endpoint=presigned_url_endpoint,
+ presigned_url_query_params=presigned_url_query_params,
  )
 
  def upload_blob_s3(
@@ -218,12 +224,18 @@ def upload_blob_s3(
  body=None,
  method: str = "POST",
  storage_uri_key: str = "storageUri",
+ presigned_url_endpoint: str = "storage/presigned-url",
+ presigned_url_query_params: str = "",
  ):
  """Generic method to upload data to S3 storage and create the appropriate
  resource in the backend.
  """
+
  presigned_json = self.post_request(
- f"storage/presigned-url?objectName={object_name}"
+ (
+ f"{presigned_url_endpoint}?objectName={object_name}"
+ f"&{presigned_url_query_params}"
+ )
  )
 
  with tqdm(
@@ -236,7 +248,7 @@ def upload_blob_s3(
  with open(file_path, "rb") as f:
  # Avoid logging here as it will break the progress bar
  fields = presigned_json["fields"]
- fields["file"] = (presigned_json["id"], f, "application/x-tar")
+ fields["file"] = (object_name, f, "application/x-tar")
  e = MultipartEncoder(fields=fields)
  m = MultipartEncoderMonitor(
  e, lambda monitor: t.update(min(t.total, monitor.bytes_read) - t.n)
@@ -267,12 +279,17 @@ def upload_blob_gcs(
  body=None,
  method: str = "POST",
  storage_uri_key: str = "storageUri",
+ presigned_url_endpoint: str = "storage/presigned-url",
+ presigned_url_query_params: str = "",
  ):
  """Generic method to upload data to Google Cloud Storage and create the
  appropriate resource in the backend.
  """
  presigned_json = self.post_request(
- f"storage/presigned-url?objectName={object_name}"
+ (
+ f"{presigned_url_endpoint}?objectName={object_name}"
+ f"&{presigned_url_query_params}"
+ )
  )
  with open(file_path, "rb") as f:
  with tqdm(
@@ -306,12 +323,17 @@ def upload_blob_azure(
  body=None,
  method: str = "POST",
  storage_uri_key: str = "storageUri",
+ presigned_url_endpoint: str = "storage/presigned-url",
+ presigned_url_query_params: str = "",
  ):
  """Generic method to upload data to Azure Blob Storage and create the
  appropriate resource in the backend.
  """
  presigned_json = self.post_request(
- f"storage/presigned-url?objectName={object_name}"
+ (
+ f"{presigned_url_endpoint}?objectName={object_name}"
+ f"&{presigned_url_query_params}"
+ )
  )
  with open(file_path, "rb") as f:
  with tqdm(
@@ -348,12 +370,17 @@ def transfer_blob(
  body=None,
  method: str = "POST",
  storage_uri_key: str = "storageUri",
+ presigned_url_endpoint: str = "storage/presigned-url",
+ presigned_url_query_params: str = "",
  ):
  """Generic method to transfer data to the openlayer folder and create the
  appropriate resource in the backend when using a local deployment.
  """
  presigned_json = self.post_request(
- f"storage/presigned-url?objectName={object_name}"
+ (
+ f"{presigned_url_endpoint}?objectName={object_name}"
+ f"&{presigned_url_query_params}"
+ )
  )
  blob_path = presigned_json["storageUri"].replace("local://", "")
  dir_path = os.path.dirname(blob_path)