Revert "Revert "Revert "get results metadata from disk/object storage (#1260)"""

This reverts commit efd754a.

Author: bossie

CHANGELOG.md

@@ -15,8 +15,6 @@ without compromising stable operations.
 
 ## In progress: 0.67.0
 
-- Avoid workaround with EJR to obtain job results metadata in the context of a failover ([#1255](https://github.com/Open-EO/openeo-geopyspark-driver/issues/1255))
-- Avoid 413 "Payload Too Large" response from EJR upon job results metadata update ([#1200](https://github.com/Open-EO/openeo-geopyspark-driver/issues/1200))
 - Configurable usage of `async_task` ([eu-cdse/openeo-cdse-infra#387](https://github.com/eu-cdse/openeo-cdse-infra/issues/387))
 - Add job option "omit-derived-from-links" to omit "derived_from" links in batch job results metadata ([ESA-WEED-project/eo_processing#175](https://github.com/ESA-WEED-project/eo_processing/issues/175))
 - Better freeIPA configurability for proxy user lookup ([#1261](https://github.com/Open-EO/openeo-geopyspark-driver/issues/1261), eu-cdse/openeo-cdse-infra#626)

openeogeotrellis/_version.py

@@ -1 +1 @@
-__version__ = "0.67.0a6"
+__version__ = "0.67.0a7"

openeogeotrellis/backend.py

@@ -32,7 +32,6 @@
 import pkg_resources
 import pystac
 import requests
-import reretry
 import shapely.geometry.base
 from deprecated import deprecated
 from geopyspark import LayerType, Pyramid, TiledRasterLayer
@@ -97,7 +96,6 @@
     k8s_get_batch_job_cfg_secret_name,
     truncate_user_id_k8s,
 )
-from openeogeotrellis.integrations.s3proxy.asset_urls import PresignedS3AssetUrls
 from openeogeotrellis.integrations.stac import ResilientStacIO
 from openeogeotrellis.integrations.traefik import Traefik
 from openeogeotrellis.integrations.yarn_jobrunner import YARNBatchJobRunner
@@ -2079,12 +2077,6 @@ def as_boolean_arg(job_option_key: str, default_value: str) -> str:
                     )
                     log.info(f"mapped job_id {job_id} to application ID {spark_app_id}")
                     dbl_registry.set_application_id(job_id=job_id, user_id=user_id, application_id=spark_app_id)
-                    dbl_registry.set_results_metadata_uri(
-                        job_id=job_id,
-                        user_id=user_id,
-                        results_metadata_uri=f"s3://{bucket}/{str(job_work_dir).strip('/')}/{JOB_METADATA_FILENAME}",
-                    )
-
                     status_response = {}
                     retry = 0
                     while "status" not in status_response and retry < 10:
@@ -2115,24 +2107,9 @@ def as_boolean_arg(job_option_key: str, default_value: str) -> str:
             runner = YARNBatchJobRunner(principal=self._principal, key_tab=self._key_tab)
             runner.set_default_sentinel_hub_credentials(self._default_sentinel_hub_client_id,self._default_sentinel_hub_client_secret)
             vault_token = None if sentinel_hub_client_alias == 'default' else get_vault_token(sentinel_hub_client_alias)
-            job_work_dir = self.get_job_work_dir(job_id=job_id)
-            application_id = runner.run_job(
-                job_info,
-                job_id,
-                job_work_dir=job_work_dir,
-                log=log,
-                user_id=user_id,
-                api_version=api_version,
-                proxy_user=proxy_user or job_info.get("proxy_user", None),
-                vault_token=vault_token,
-            )
+            application_id = runner.run_job(job_info, job_id, job_work_dir = self.get_job_work_dir(job_id=job_id), log=log, user_id=user_id, api_version=api_version,proxy_user=proxy_user or job_info.get('proxy_user',None), vault_token=vault_token)
             with self._double_job_registry as dbl_registry:
                 dbl_registry.set_application_id(job_id=job_id, user_id=user_id, application_id=application_id)
-                dbl_registry.set_results_metadata_uri(
-                    job_id=job_id,
-                    user_id=user_id,
-                    results_metadata_uri=f"file://{job_work_dir}/{JOB_METADATA_FILENAME}",
-                )
                 dbl_registry.set_status(job_id=job_id, user_id=user_id, status=JOB_STATUS.QUEUED)
 
 
@@ -2582,29 +2559,24 @@ def get_result_assets(self, job_id: str, user_id: str) -> Dict[str, dict]:
 
         :return: A mapping between a filename and a dict containing information about that file.
         """
-        with self._double_job_registry as registry:
-            job_dict = registry.get_job(job_id=job_id, user_id=user_id)
-
-        if job_dict["status"] != JOB_STATUS.FINISHED:
+        job_info = self.get_job_info(job_id=job_id, user_id=user_id)
+        if job_info.status != JOB_STATUS.FINISHED:
             raise JobNotFinishedException
 
         job_dir = self.get_job_output_dir(job_id=job_id)
 
-        results_metadata = self._load_results_metadata_from_uri(job_dict.get("results_metadata_uri"), job_id)  # TODO: expose a getter?
-        if not results_metadata:
-            try:
-                logger.debug(f"Loading results metadata from job registry", extra={"job_id": job_id})
-                with self._double_job_registry as registry:
-                    job_dict = registry.elastic_job_registry.get_job(job_id, user_id=user_id)
-                    if "results_metadata" in job_dict:
-                        results_metadata = job_dict["results_metadata"]
-            except Exception as e:
-                logger.warning(
-                    "Could not retrieve result metadata from job registry %s", e, exc_info=True, extra={"job_id": job_id}
-                )
-        if not results_metadata:
+        results_metadata = None
+        try:
+            with self._double_job_registry as registry:
+                job_dict = registry.elastic_job_registry.get_job(job_id, user_id=user_id)
+                if "results_metadata" in job_dict:
+                    results_metadata = job_dict["results_metadata"]
+        except Exception as e:
+            logger.warning(
+                "Could not retrieve result metadata from job tracker %s", e, exc_info=True, extra={"job_id": job_id}
+            )
+        if results_metadata is None or len(results_metadata) == 0:
             results_metadata = self.load_results_metadata(job_id, user_id)
-
         out_assets = results_metadata.get("assets", {})
         out_metadata = out_assets.get("out", {})
         bands = [Band(*properties) for properties in out_metadata.get("bands", [])]
@@ -2682,21 +2654,12 @@ def load_results_metadata(self, job_id: str, user_id: str) -> dict:
         """
         Reads the metadata json file from the job directory and returns it.
         """
-        with self._double_job_registry as registry:
-            job_dict = registry.get_job(job_id=job_id, user_id=user_id)
-
-        results_metadata = self._load_results_metadata_from_uri(job_dict.get("results_metadata_uri"), job_id)  # TODO: expose a getter?
-        if results_metadata is not None:
-            return results_metadata
 
         metadata_file = self.get_results_metadata_path(job_id=job_id)
 
         if ConfigParams().use_object_storage:
             try:
-                logger.debug(
-                    f"Loading results metadata from object storage at {metadata_file}", extra={"job_id": job_id}
-                )
-                contents = get_s3_file_contents(path=str(metadata_file))
+                contents = get_s3_file_contents(str(metadata_file))
                 return json.loads(contents)
             except Exception:
                 logger.warning(
@@ -2705,7 +2668,6 @@ def load_results_metadata(self, job_id: str, user_id: str) -> dict:
                     extra={'job_id': job_id})
 
         try:
-            logger.debug(f"Loading results metadata from file at {metadata_file}", extra={"job_id": job_id})
             with open(metadata_file) as f:
                 return json.load(f)
         except FileNotFoundError:
@@ -2714,55 +2676,6 @@ def load_results_metadata(self, job_id: str, user_id: str) -> dict:
 
         return {}
 
-    @staticmethod
-    @reretry.retry(exceptions=FileNotFoundError, tries=5, delay=1, backoff=2, logger=logger)
-    def _load_results_metadata_from_file(metadata_file: Path):
-        with open(metadata_file) as f:
-            return json.load(f)
-
-    @staticmethod
-    def _load_results_metadata_from_uri(results_metadata_uri: Optional[str], job_id: str) -> Optional[dict]:
-        # TODO: reduce code duplication with load_results_metadata
-        import botocore.exceptions
-
-        if results_metadata_uri is None:
-            return None
-
-        logger.debug(f"Loading results metadata from URI {results_metadata_uri}", extra={"job_id": job_id})
-
-        uri_parts = urlparse(results_metadata_uri)
-
-        if uri_parts.scheme == "file":
-            file_path = Path(uri_parts.path)
-            try:
-                return GpsBatchJobs._load_results_metadata_from_file(file_path)
-            except FileNotFoundError:
-                logger.debug(
-                    f"File with results metadata {file_path} does not exist; this is expected and not "
-                    f"an error if the batch job did not have the chance to write it yet.",
-                    exc_info=True,
-                    extra={"job_id": job_id},
-                )
-                return None
-
-        if uri_parts.scheme == "s3":
-            bucket, key = PresignedS3AssetUrls.get_bucket_key_from_uri(results_metadata_uri)
-            try:
-                return json.loads(get_s3_file_contents(key, bucket))
-            except botocore.exceptions.ClientError as e:
-                if e.response["Error"]["Code"] != "NoSuchKey":
-                    raise
-
-                logger.debug(
-                    f"Object with results metadata {key} does not exist in bucket {bucket}; this is "
-                    f"expected and not an error if the batch job did not have the chance to write it yet.",
-                    exc_info=True,
-                    extra={"job_id": job_id},
-                )
-                return None
-
-        raise ValueError(f"Unsupported results metadata URI: {results_metadata_uri}")
-
     def _get_providers(self, job_id: str, user_id: str) -> List[dict]:
         results_metadata = self.load_results_metadata(job_id, user_id)
         return results_metadata.get("providers", [])

openeogeotrellis/deploy/batch_job.py

@@ -44,7 +44,9 @@
 from shapely.geometry import mapping
 
 from openeogeotrellis._version import __version__
-from openeogeotrellis.backend import GeoPySparkBackendImplementation
+from openeogeotrellis.backend import (
+    GeoPySparkBackendImplementation,
+)
 from openeogeotrellis.collect_unique_process_ids_visitor import (
     CollectUniqueProcessIdsVisitor,
 )

openeogeotrellis/job_registry.py

@@ -1,12 +1,11 @@
 from __future__ import annotations
 import contextlib
-from copy import deepcopy
+import datetime as dt
 import json
 import logging
 import random
 from datetime import datetime, timedelta
 from decimal import Decimal
-from pathlib import Path
 from typing import Any, List, Dict, Callable, Union, Optional, Iterator, Tuple
 
 import kazoo
@@ -155,9 +154,6 @@ def set_dependencies(self, job_id: str, user_id: str, dependencies: List[Dict[st
     def remove_dependencies(self, job_id: str, user_id: str):
         self.patch(job_id, user_id, dependencies=None, dependency_status=None)
 
-    def set_results_metadata_uri(self, job_id: str, user_id: str, results_metadata_uri: str) -> None:
-        self.patch(job_id, user_id, results_metadata_uri=results_metadata_uri)
-
     def patch(
         self, job_id: str, user_id: str, auto_mark_done: bool = True, **kwargs
     ) -> None:
@@ -627,15 +623,15 @@ def create_job(
             "api_version": api_version,
             "job_options": job_options,
         }
-        return deepcopy(self.db[job_id])
+        return self.db[job_id]
 
     def get_job(self, job_id: str, *, user_id: Optional[str] = None) -> JobDict:
         job = self.db.get(job_id)
 
         if not job or (user_id is not None and job['user_id'] != user_id):
             raise JobNotFoundException(job_id=job_id)
 
-        return deepcopy(job)
+        return job
 
     def delete_job(self, job_id: str, *, user_id: Optional[str] = None) -> None:
         self.get_job(job_id=job_id, user_id=user_id)  # will raise on job not found
@@ -644,7 +640,7 @@ def delete_job(self, job_id: str, *, user_id: Optional[str] = None) -> None:
     def _update(self, job_id: str, **kwargs) -> JobDict:
         assert job_id in self.db
         self.db[job_id].update(**kwargs)
-        return deepcopy(self.db[job_id])
+        return self.db[job_id]
 
     def set_status(
         self,
@@ -695,10 +691,7 @@ def set_results_metadata(
         usage: dict,
         results_metadata: Optional[Dict[str, Any]] = None,
     ) -> None:
-        if results_metadata:
-            self._update(job_id=job_id, costs=costs, usage=usage, results_metadata=results_metadata)
-        else:
-            self._update(job_id=job_id, costs=costs, usage=usage)
+        self._update(job_id=job_id, costs=costs, usage=usage, results_metadata=results_metadata)
 
     def set_results_metadata_uri(
         self, job_id: str, *, user_id: Optional[str] = None, results_metadata_uri: str
@@ -714,7 +707,7 @@ def list_user_jobs(
         request_parameters: Optional[dict] = None,
         # TODO #959 settle on returning just `JobListing` and eliminate other options/code paths.
     ) -> Union[JobListing, List[JobDict]]:
-        jobs = [deepcopy(job) for job in self.db.values() if job["user_id"] == user_id]
+        jobs = [job for job in self.db.values() if job["user_id"] == user_id]
         if limit:
             pagination_param = "page"
             page_number = int((request_parameters or {}).get(pagination_param, 0))
@@ -741,7 +734,7 @@ def list_active_jobs(
         active = [JOB_STATUS.CREATED, JOB_STATUS.QUEUED, JOB_STATUS.RUNNING]
         # TODO: implement support for max_age, max_updated_ago, fields
         return [
-            deepcopy(job)
+            job
             for job in self.db.values()
             if job["status"] in active and (not require_application_id or job.get("application_id") is not None)
         ]
@@ -874,64 +867,10 @@ def get_job_metadata(self, job_id: str, user_id: str) -> BatchJobMetadata:
                 with contextlib.suppress(JobNotFoundException):
                     ejr_job_info = self.elastic_job_registry.get_job(job_id=job_id, user_id=user_id)
 
-                    # TODO: replace with getter once introduced?
-                    results_metadata = self._load_results_metadata_from_uri(
-                        ejr_job_info.get("results_metadata_uri"), job_id
-                    )
-                    if results_metadata is not None:
-                        ejr_job_info["results_metadata"] = results_metadata
-
         self._check_zk_ejr_job_info(job_id=job_id, zk_job_info=zk_job_info, ejr_job_info=ejr_job_info)
         job_metadata = zk_job_info_to_metadata(zk_job_info) if zk_job_info else ejr_job_info_to_metadata(ejr_job_info)
         return job_metadata
 
-    @staticmethod
-    def _load_results_metadata_from_uri(results_metadata_uri: Optional[str], job_id: str) -> Optional[dict]:
-        # TODO: reduce code duplication with openeogeotrellis.backend.GpsBatchJobs._load_results_metadata_from_uri
-        from openeogeotrellis.integrations.s3proxy.asset_urls import PresignedS3AssetUrls
-        from openeogeotrellis.utils import get_s3_file_contents
-        import botocore.exceptions
-        from urllib.parse import urlparse
-
-        if results_metadata_uri is None:
-            return None
-
-        _log.debug(f"Loading results metadata from URI {results_metadata_uri}", extra={"job_id": job_id})
-
-        uri_parts = urlparse(results_metadata_uri)
-
-        if uri_parts.scheme == "file":
-            file_path = Path(uri_parts.path)
-            try:
-                with open(file_path) as f:
-                    return json.load(f)
-            except FileNotFoundError:
-                _log.debug(
-                    f"File with results metadata {file_path} does not exist; this is expected and not "
-                    f"an error if the batch job did not have the chance to write it yet.",
-                    exc_info=True,
-                    extra={"job_id": job_id},
-                )
-                return None
-
-        if uri_parts.scheme == "s3":
-            bucket, key = PresignedS3AssetUrls.get_bucket_key_from_uri(results_metadata_uri)
-            try:
-                return json.loads(get_s3_file_contents(key, bucket))
-            except botocore.exceptions.ClientError as e:
-                if e.response["Error"]["Code"] != "NoSuchKey":
-                    raise
-
-                _log.debug(
-                    f"Object with results metadata {key} does not exist in bucket {bucket}; this is "
-                    f"expected and not an error if the batch job did not have the chance to write it yet.",
-                    exc_info=True,
-                    extra={"job_id": job_id},
-                )
-                return None
-
-        raise ValueError(f"Unsupported results metadata URI: {results_metadata_uri}")
-
     def _check_zk_ejr_job_info(self, job_id: str, zk_job_info: Union[dict, None], ejr_job_info: Union[dict, None]):
         # TODO #236/#498 For now: compare job metadata between Zk and EJR
         fields = ["job_id", "status", "created"]
@@ -1020,17 +959,6 @@ def set_application_id(self, job_id: str, *, user_id: Optional[str] = None, appl
         if self.elastic_job_registry:
             self.elastic_job_registry.set_application_id(job_id=job_id, user_id=user_id, application_id=application_id)
 
-    def set_results_metadata_uri(self, job_id: str, *, user_id: Optional[str] = None, results_metadata_uri: str):
-        if self.zk_job_registry:
-            assert user_id, "user_id is required in ZkJobRegistry"
-            self.zk_job_registry.set_results_metadata_uri(
-                job_id=job_id, user_id=user_id, results_metadata_uri=results_metadata_uri
-            )
-        if self.elastic_job_registry:
-            self.elastic_job_registry.set_results_metadata_uri(
-                job_id=job_id, user_id=user_id, results_metadata_uri=results_metadata_uri
-            )
-
     def mark_ongoing(self, job_id: str, user_id: str) -> None:
         # TODO #863/#1123 can this method be eliminated (e.g. integrate it directly in ZkJobRegistry.set_status)?
         if self.zk_job_registry:
@@ -1099,9 +1027,8 @@ def set_results_metadata(
     ) -> None:
         if self.zk_job_registry:
             assert user_id, "user_id is required in ZkJobRegistry"
-            self.zk_job_registry.patch(
-                job_id=job_id, user_id=user_id, **dict(results_metadata or {}, costs=costs, usage=usage)
-            )
+            self.zk_job_registry.patch(job_id=job_id, user_id=user_id,
+                                       **dict(results_metadata, costs=costs, usage=usage))
 
         if self.elastic_job_registry:
             self.elastic_job_registry.set_results_metadata(