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)