Merge branch 'main' into fix/add-retry-logic-pipeline-components

Author: fMurugi

gfmstudio/fine_tuning/api.py

@@ -110,20 +110,6 @@ def parse_hpo_form(tune_metadata: str = Form(...)) -> schemas.TuneSubmitHPO:
 inference_crud = crud.ItemCrud(model=Inference)
 inference_model_crud = crud.ItemCrud(model=InferenceModel)
 
-"""
-Create file folders if they don't exist already
-"""
-if settings.ENVIRONMENT.lower() in ["local", "crc"]:
-    tasks_dir = os.path.join(settings.TUNE_BASEDIR, "tune-tasks")
-    if os.path.isdir(tasks_dir) is False:
-        logger.info(f"Creating tasks directory: {tasks_dir}")
-        os.makedirs(tasks_dir)
-
-    trained_dir = os.path.join(settings.TUNE_BASEDIR, "pre-trained")
-    if os.path.isdir(trained_dir) is False:
-        logger.info(f"Creating pre-trained directory: {trained_dir}")
-        os.makedirs(trained_dir)
-
 
 ###############################################
 # ---- Get token endpoint
@@ -996,7 +982,7 @@ async def submit_hpo_tune_yaml(
         bucket_key = f"tune-tasks/{tune_id}/{tune_id}_config.yaml"
         tune_dir = os.path.join(settings.TUNE_BASEDIR, f"tune-tasks/{tune_id}")
         if os.path.isdir(tune_dir) is False:
-            os.mkdir(tune_dir)
+            os.makedirs(tune_dir, exist_ok=True)
 
         bucket_dir = os.path.join(settings.TUNE_BASEDIR, bucket_key)
         config_data = yaml.load(config_content, Loader=yaml.SafeLoader)

gfmstudio/fine_tuning/deployment/k8-dataset-onboarding-v2-pipeline.tpl.yaml

@@ -9,6 +9,7 @@ metadata:
 spec:
   template:
     spec:
+      serviceAccountName: api-gateway-sa
       imagePullSecrets:
         - name: IMAGE_PULL_SECRET
       containers:
@@ -17,7 +18,7 @@ spec:
         imagePullPolicy: Always
         volumeMounts:
           - name: mount-data
-            mountPath: /data
+            mountPath: /pipeline/data
         env:
         - name: LOGLEVEL
           valueFrom:

gfmstudio/fine_tuning/utils/dataset_handlers.py

@@ -269,7 +269,7 @@ def obtain_job_log(pod_name: str, dataset_id: str) -> str:
     destination: str
         the filepath at which the job log is stored
     """
-    logs_dir = "deployment/logs"
+    logs_dir = "/app/deployment/logs"
     os.makedirs(logs_dir, exist_ok=True)
     log_path = f"{logs_dir}/{dataset_id}.log"
     try:

gfmstudio/fine_tuning/utils/tune_handlers.py

@@ -655,23 +655,6 @@ async def save_tune_config(
     HTTPException
         500 if COS upload fails
     """
-    # Upload to COS if not local environment
-    if settings.ENVIRONMENT.lower() not in ["local", "crc"]:
-        s3 = object_storage.object_storage_client()
-        try:
-            await asyncify(s3.put_object)(
-                Bucket=settings.TUNES_FILES_BUCKET,
-                Body=rendered_template,
-                Key=bucket_key,
-            )
-            logger.debug(
-                f"Config uploaded to COS: {settings.TUNES_FILES_BUCKET}/{bucket_key}"
-            )
-        except Exception as exc:
-            logger.exception("Failed to upload config to COS")
-            raise HTTPException(
-                status_code=500, detail="Failed to upload configuration to storage"
-            ) from exc
 
     # Save to local storage
     tune_dir = os.path.join(settings.TUNE_BASEDIR, f"tune-tasks/{tune_id}")

gfmstudio/fine_tuning/utils/webhook_event_handlers.py

@@ -319,7 +319,15 @@ async def handle_dataset_factory_webhooks(
                 f"kubectl delete job onboarding-v2-pipeline-{dataset_id}"
             )
             k8s_delete_secret_command = f"kubectl delete secret dataset-onboarding-v2-pipeline-params-{dataset_id}"
-            remove_job_deployment_file_command = f"rm {BASE_DIR}/deployment/jobs/onboarding-v2-pipeline-{dataset_id}.yaml"
+
+            # Validate BASE_DIR before using it in rm command
+            if not BASE_DIR or str(BASE_DIR).strip() == "":
+                logger.info(f"BASE_DIR is empty or invalid: '{BASE_DIR}'. Try using /app as BASE_DIR")
+                deployment_file_path = f"/app/deployment/jobs/onboarding-v2-pipeline-{dataset_id}.yaml"
+                remove_job_deployment_file_command = f"rm -f {deployment_file_path}"
+            else:
+                deployment_file_path = f"{BASE_DIR}/deployment/jobs/onboarding-v2-pipeline-{dataset_id}.yaml"
+                remove_job_deployment_file_command = f"rm -f {deployment_file_path}"
 
             try:
                 delete_job_output = subprocess.check_output(

gfmstudio/inference/services.py

@@ -123,84 +123,39 @@ async def invoke_tune_upload_handler(
         logger.exception(f"Network error during download:{e}")
         raise
 
-    if settings.ENVIRONMENT.lower() in ["local", "crc"]:
-        tune_dir = os.path.join(settings.TUNE_BASEDIR, f"tune-tasks/{tune_id}")
-        if os.path.isdir(tune_dir) is False:
-            os.mkdir(tune_dir)
+    tune_dir = os.path.join(settings.TUNE_BASEDIR, f"tune-tasks/{tune_id}")
+    if os.path.isdir(tune_dir) is False:
+        os.makedirs(tune_dir, exist_ok=True)
 
-        tune_config_deploy_bucket_dir = os.path.join(
-            settings.TUNE_BASEDIR, tune_config_deploy_bucket_key
-        )
-        tune_config_bucket_dir = os.path.join(
-            settings.TUNE_BASEDIR, tune_config_bucket_key
-        )
-        tune_checkpoint_bucket_dir = os.path.join(
-            settings.TUNE_BASEDIR, tune_checkpoint_bucket_key
-        )
-
-        if tune_config_url[0:4] == "http":
-            with open(tune_config_deploy_bucket_dir, "w") as config_file:
-                config_file.write(tune_config_response.text)
-            with open(tune_config_bucket_dir, "w") as config_file:
-                config_file.write(tune_config_response.text)
-
-        elif tune_config_url[0:4] == "file":
-            if not os.path.exists(tune_config_deploy_bucket_dir):
-                shutil.copyfile(tune_config_url[7:], tune_config_deploy_bucket_dir)
-            if not os.path.exists(tune_config_bucket_dir):
-                shutil.copyfile(tune_config_url[7:], tune_config_bucket_dir)
-
-        if tune_checkpoint_url[0:4] == "http":
-            with open(tune_checkpoint_bucket_dir, "wb") as checkpoint_file:
-                checkpoint_file.write(tune_checkpoint_response.content)
-        elif tune_checkpoint_url[0:4] == "file":
-            if not os.path.exists(tune_checkpoint_bucket_dir):
-                shutil.copyfile(tune_checkpoint_url[7:], tune_checkpoint_bucket_dir)
+    tune_config_deploy_bucket_dir = os.path.join(
+        settings.TUNE_BASEDIR, tune_config_deploy_bucket_key
+    )
+    tune_config_bucket_dir = os.path.join(
+        settings.TUNE_BASEDIR, tune_config_bucket_key
+    )
+    tune_checkpoint_bucket_dir = os.path.join(
+        settings.TUNE_BASEDIR, tune_checkpoint_bucket_key
+    )
 
-    else:
-        pipelines_bucket_name = settings.TUNES_FILES_BUCKET
-        try:
-            logger.info(f"Connect to  cos bucket{pipelines_bucket_name}")
-            pipeline_s3_client = boto3.client(
-                "s3",
-                aws_access_key_id=settings.OBJECT_STORAGE_KEY_ID,
-                aws_secret_access_key=settings.OBJECT_STORAGE_SEC_KEY,
-                endpoint_url=settings.OBJECT_STORAGE_ENDPOINT,
-                config=Config(
-                    signature_version=settings.OBJECT_STORAGE_SIGNATURE_VERSION
-                ),
-                verify=(settings.ENVIRONMENT.lower() not in ["local", "crc"]),
-            )
-        except ValueError as exc:
-            logger.error(f"pipeline_s3_client Misconfiguration: {str(exc)}")
-
-        try:
-            logger.info("Uploading tune config file")
-            # Uploading both config_deploy.yaml and {tune_id}_config.yaml
-            pipeline_s3_client.upload_fileobj(
-                tune_config_response.raw,
-                Bucket=pipelines_bucket_name,
-                Key=tune_config_deploy_bucket_key,
-            )
-            pipeline_s3_client.upload_fileobj(
-                tune_config_response.raw,
-                Bucket=pipelines_bucket_name,
-                Key=tune_config_bucket_key,
-            )
-            logger.info("Uploading tune config Succeeded")
+    if tune_config_url[0:4] == "http":
+        with open(tune_config_deploy_bucket_dir, "w") as config_file:
+            config_file.write(tune_config_response.text)
+        with open(tune_config_bucket_dir, "w") as config_file:
+            config_file.write(tune_config_response.text)
+
+    elif tune_config_url[0:4] == "file":
+        if not os.path.exists(tune_config_deploy_bucket_dir):
+            shutil.copyfile(tune_config_url[7:], tune_config_deploy_bucket_dir)
+        if not os.path.exists(tune_config_bucket_dir):
+            shutil.copyfile(tune_config_url[7:], tune_config_bucket_dir)
+
+    if tune_checkpoint_url[0:4] == "http":
+        with open(tune_checkpoint_bucket_dir, "wb") as checkpoint_file:
+            checkpoint_file.write(tune_checkpoint_response.content)
+    elif tune_checkpoint_url[0:4] == "file":
+        if not os.path.exists(tune_checkpoint_bucket_dir):
+            shutil.copyfile(tune_checkpoint_url[7:], tune_checkpoint_bucket_dir)
 
-            logger.info("uploading tune checkpoint file")
-            pipeline_s3_client.upload_fileobj(
-                tune_checkpoint_response.raw,
-                Bucket=pipelines_bucket_name,
-                Key=tune_checkpoint_bucket_key,
-            )
-        except (requests.exceptions.ConnectionError, requests.exceptions.Timeout) as e:
-            logger.exception(f"Network error during download:{e}")
-            raise
-        except ClientError as e:
-            logger.exception(f"Failed to upload files to cos: {e}")
-            raise
     logger.info("Updating the tune with {tune_id} status to Finished")
     tunes_crud.update(
         db=db,

pipelines/Dockerfile

@@ -99,8 +99,8 @@ RUN mkdir /pipeline/data \
 && mkdir /.docker && chmod -R 777 /.docker \
 && mkdir /.local && chmod -R 777 /.local \
 && mkdir /.cache && chmod -R 777 /.cache \
-&& chown -R 10001:10001 /pipeline/ \
-&& chmod -R 777 /pipeline/
+&& chown -R 10001:10001 /pipeline/ /pipeline/data \
+&& chmod -R 777 /pipeline/ /pipeline/data
 
 COPY ./components/terrakit_data_fetch/sentinelhub_config.toml /.config/sentinelhub/config.toml
 RUN chmod -R 777 /.config/sentinelhub/

pipelines/components/curated_upload/claimed_curated_upload_v2.py

@@ -714,7 +714,7 @@ def main():
 
     onboarding_details = {}
 
-    working_path = "/data/" + payload["dataset_id"]
+    working_path = "/pipeline/data/" + payload["dataset_id"]
 
     dataset_bucket = os.getenv("DATA_BUCKET", "geoft-service-datasets")
 
@@ -895,9 +895,10 @@ def main():
         main()
     except Exception as e:
         logger.error(
-            "An exception occurred when onboarding the dataset", stack_info=True
+            f"An exception occurred when onboarding the dataset: {str(e)}",
+            exc_info=True,
+            stack_info=True
         )
-        logger.error("Exception - " + str(e))
         error["message"] = str(e)
         onboarding_details = {}
         populate_onboarding_details(

pipelines/components/push_to_geoserver/push_to_geoserver_helper_functions.py

@@ -56,7 +56,7 @@ def add_imagemosaic_to_geoserver(geo, workspace, task_folder, layer_name, retrie
     logger.debug(f"Save to geoserver create coverage store for file: {retrieved_file_paths}")
 
     layer_folder = f"{task_folder}/{layer_name}"
-    os.mkdir(layer_folder)
+    os.makedirs(layer_folder, exist_ok=True)
     for f in retrieved_file_paths:
         shutil.copy(f, f"{layer_folder}/")