Issue #195 add log level to log responses

also fine-tune/unify some other log handling aspects

Author: soxofaan

CHANGELOG.md

@@ -21,6 +21,12 @@ and start a new "In Progress" section above it.
 
 ## In progress
 
+
+## 0.125.0
+
+- Add log level to batch job logs response ([#195](https://github.com/Open-EO/openeo-python-driver/issues/195))
+
+
 ## 0.124.0
 
 - Better argument validation in `resample_spatial`/`resample_cube_spatial` (related to [Open-EO/openeo-python-client#690](https://github.com/Open-EO/openeo-python-client/issues/690))

openeo_driver/_version.py

@@ -1 +1 @@
-__version__ = "0.124.0a1"
+__version__ = "0.125.0a1"

openeo_driver/backend.py

@@ -31,7 +31,7 @@
 from openeo_driver.datastructs import SarBackscatterArgs
 from openeo_driver.dry_run import SourceConstraint
 from openeo_driver.errors import CollectionNotFoundException, ServiceUnsupportedException, FeatureUnsupportedException
-from openeo_driver.constants import JOB_STATUS
+from openeo_driver.constants import JOB_STATUS, DEFAULT_LOG_LEVEL_RETRIEVAL
 from openeo_driver.processes import ProcessRegistry
 from openeo_driver.users import User
 from openeo_driver.users.oidc import OidcProvider
@@ -142,7 +142,15 @@ def remove_service(self, user_id: str, service_id: str) -> None:
         """https://openeo.org/documentation/1.0/developers/api/reference.html#operation/delete-service"""
         raise NotImplementedError()
 
-    def get_log_entries(self, service_id: str, user_id: str, offset: str) -> List[dict]:
+    def get_log_entries(
+        self,
+        service_id: str,
+        *,
+        user_id: str,
+        offset: Union[str, None] = None,
+        limit: Union[int, None] = None,
+        level: str = DEFAULT_LOG_LEVEL_RETRIEVAL,
+    ) -> List[dict]:
         """https://openeo.org/documentation/1.0/developers/api/reference.html#operation/debug-service"""
         # TODO require auth/user handle?
         return []
@@ -178,12 +186,12 @@ class LoadParameters(dict):
     """
     A buffer provided in the units of the target CRS. If target CRS is not provided, then it is assumed to be the native CRS
     of the collection.
-    
-    This buffer is applied to AOI when constructing the datacube, allowing operations that require neighbouring pixels 
+
+    This buffer is applied to AOI when constructing the datacube, allowing operations that require neighbouring pixels
     to be implemented correctly. Examples are apply_kernel and apply_neighborhood, but also certain resampling operations
     could be affected by this.
-    
-    The buffer has to be considered in the global extent! 
+
+    The buffer has to be considered in the global extent!
     """
     pixel_buffer = dict_item(default=None)
 
@@ -552,9 +560,11 @@ def get_results(self, job_id: str, user_id: str) -> Dict[str, dict]:
     def get_log_entries(
         self,
         job_id: str,
+        *,
         user_id: str,
-        offset: Optional[str] = None,
-        level: Optional[str] = None,
+        offset: Union[str, None] = None,
+        limit: Union[str, None] = None,
+        level: str = DEFAULT_LOG_LEVEL_RETRIEVAL,
     ) -> Iterable[dict]:
         """
         https://openeo.org/documentation/1.0/developers/api/reference.html#operation/debug-job

openeo_driver/constants.py

@@ -49,3 +49,9 @@ class JOB_STATUS:
     "lower-right",
     "upper-right",
 ]
+
+
+# Default value for `level` parameter in `POST /result`, `POST /jobs`, ... requests
+DEFAULT_LOG_LEVEL_PROCESSING = "info"
+# Default value for `level in `GET /jobs/{job_id}/logs`, `GET /services/{service_id}/logs` requests
+DEFAULT_LOG_LEVEL_RETRIEVAL = "debug"

openeo_driver/dummy/dummy_backend.py

@@ -43,7 +43,7 @@
     UserDefinedProcessMetadata,
 )
 from openeo_driver.config import OpenEoBackendConfig
-from openeo_driver.constants import JOB_STATUS, STAC_EXTENSION
+from openeo_driver.constants import JOB_STATUS, STAC_EXTENSION, DEFAULT_LOG_LEVEL_RETRIEVAL
 from openeo_driver.datacube import DriverDataCube, DriverMlModel, DriverVectorCube
 from openeo_driver.datastructs import StacAsset
 from openeo_driver.delayed_vector import DelayedVector
@@ -172,9 +172,17 @@ def list_services(self, user_id: str) -> List[ServiceMetadata]:
     def service_info(self, user_id: str, service_id: str) -> ServiceMetadata:
         return next(s for s in self._registry if s.id == service_id)
 
-    def get_log_entries(self, service_id: str, user_id: str, offset: str) -> List[dict]:
+    def get_log_entries(
+        self,
+        service_id: str,
+        *,
+        user_id: str,
+        offset: Union[str, None] = None,
+        limit: Union[int, None] = None,
+        level: str = DEFAULT_LOG_LEVEL_RETRIEVAL,
+    ) -> List[dict]:
         return [
-            {"id": 3, "level": "info", "message": "Loaded data."}
+            {"id": 3, "level": "info", "message": "Loaded data."},
         ]
 
 
@@ -921,16 +929,18 @@ def get_result_assets(self, job_id: str, user_id: str) -> Dict[str, dict]:
     def get_log_entries(
         self,
         job_id: str,
+        *,
         user_id: str,
-        offset: Optional[str] = None,
-        level: Optional[str] = None,
+        offset: Union[str, None] = None,
+        limit: Union[str, None] = None,
+        level: str = DEFAULT_LOG_LEVEL_RETRIEVAL,
     ) -> Iterable[dict]:
         self._get_job_info(job_id=job_id, user_id=user_id)
         default_logs = [{"id": "1", "level": "info", "message": "hello world"}]
         requested_level = normalize_log_level(level)
         for log in self._custom_job_logs.get(job_id, default_logs):
             if isinstance(log, dict):
-                actual_level = normalize_log_level(log.get("log_level"))
+                actual_level = normalize_log_level(log.get("level"))
                 if actual_level < requested_level:
                     continue
                 yield log

openeo_driver/testing.py

@@ -206,9 +206,11 @@ def ensure_auth_header(self):
         if not self.default_request_headers.get("Authorization"):
             self.set_auth_bearer_token()
 
-    def get(self, path: str, headers: dict = None) -> ApiResponse:
+    def get(self, path: str, headers: dict = None, params: Optional[dict] = None) -> ApiResponse:
         """Do versioned GET request, given non-versioned path"""
-        return ApiResponse(self.client.get(path=self.url(path), headers=self._request_headers(headers)))
+        return ApiResponse(
+            self.client.get(path=self.url(path), headers=self._request_headers(headers), query_string=params)
+        )
 
     def head(self, path: str, headers: dict = None) -> ApiResponse:
         """Do versioned GET request, given non-versioned path"""

openeo_driver/views.py

@@ -39,7 +39,7 @@
     JobListing,
 )
 from openeo_driver.config import get_backend_config, OpenEoBackendConfig
-from openeo_driver.constants import STAC_EXTENSION
+from openeo_driver.constants import STAC_EXTENSION, DEFAULT_LOG_LEVEL_RETRIEVAL, DEFAULT_LOG_LEVEL_PROCESSING
 from openeo_driver.datacube import DriverMlModel
 from openeo_driver.errors import (
     OpenEOApiException,
@@ -654,7 +654,7 @@ def result(user: User):
         process_graph = _extract_process_graph(post_data)
         budget = post_data.get("budget")
         plan = post_data.get("plan")
-        log_level = post_data.get("log_level", "info")
+        log_level = post_data.get("log_level", DEFAULT_LOG_LEVEL_PROCESSING)
         job_options = _extract_job_options(
             post_data, to_ignore=["process", "process_graph", "budget", "plan", "log_level"]
         )
@@ -880,7 +880,7 @@ def create_job(user: User):
             post_data, to_ignore=["process", "process_graph", "title", "description", "plan", "budget", "log_level"]
         )
         metadata = {k: post_data[k] for k in ["title", "description", "plan", "budget"] if k in post_data}
-        metadata["log_level"] = post_data.get("log_level", "info")
+        metadata["log_level"] = post_data.get("log_level", DEFAULT_LOG_LEVEL_PROCESSING)
         job_info = backend_implementation.batch_jobs.create_job(
             # TODO: remove `filter_supported_kwargs` (when all implementations have migrated to `user` iso `user_id`)
             **filter_supported_kwargs(
@@ -1513,16 +1513,26 @@ def download_job_result_signed(job_id, user_base64, secure_key, filename):
     @blueprint.route("/jobs/<job_id>/logs", methods=["GET"])
     @auth_handler.requires_bearer_auth
     def get_job_logs(job_id, user: User):
-        offset = request.args.get("offset")
-        level = request.args.get("level", "debug")
+        offset = request.args.get("offset", default=None)
+        limit = request.args.get("limit", default=None, type=int)
+        level = request.args.get("level", default=DEFAULT_LOG_LEVEL_RETRIEVAL)
         request_id = FlaskRequestCorrelationIdLogging.get_request_id()
         # TODO: implement paging support: `limit`, next/prev/first/last `links`, ...
-        logs = backend_implementation.batch_jobs.get_log_entries(
+
+        # TODO: remove this `function_has_argument` once all implementations are migrated
+        if function_has_argument(backend_implementation.batch_jobs.get_log_entries, argument="limit"):
+            logs = backend_implementation.batch_jobs.get_log_entries(
+                job_id=job_id, user_id=user.user_id, offset=offset, level=level, limit=limit
+            )
+        else:
+            logs = backend_implementation.batch_jobs.get_log_entries(
             job_id=job_id, user_id=user.user_id, offset=offset, level=level
         )
 
         def generate():
-            yield """{"logs":["""
+            yield "{"
+            yield f'"level": {json.dumps(level)},'
+            yield '"logs":['
 
             sep = ""
             try:
@@ -1550,7 +1560,9 @@ def generate():
                 }
                 yield sep + json.dumps(log)
 
-            yield """],"links":[]}"""
+            yield "],"
+            # TODO: add pagination links (next, prev, first, last)
+            yield '"links":[]}'
 
         return current_app.response_class(generate(), mimetype="application/json")
 
@@ -1665,12 +1677,13 @@ def service_delete(service_id, user: User):
     @blueprint.route('/services/<service_id>/logs', methods=['GET'])
     @auth_handler.requires_bearer_auth
     def service_logs(service_id, user: User):
-        level = request.args.get("level", "debug")
-        offset = request.args.get('offset', 0)
+        offset = request.args.get("offset", default=None)
+        limit = request.args.get("limit", default=None, type=int)
+        level = request.args.get("level", default=DEFAULT_LOG_LEVEL_RETRIEVAL)
         logs = backend_implementation.secondary_services.get_log_entries(
-            service_id=service_id, user_id=user.user_id, offset=offset
+            service_id=service_id, user_id=user.user_id, offset=offset, limit=limit, level=level
         )
-        return jsonify({"logs": logs, "links": []})
+        return jsonify({"logs": logs, "links": [], "level": level})
 
 
 def register_views_udp(

tests/test_views.py

@@ -3055,11 +3055,32 @@ def test_get_batch_job_logs(self, api):
         with self._fresh_job_registry():
             resp = api.get('/jobs/07024ee9-7847-4b8a-b260-6c879a2b3cdc/logs', headers=self.AUTH_HEADER)
         assert resp.assert_status_code(200).json == {
+            "level": "debug",
             "logs": [
-                {"id": "1", "level": "info", "message": "hello world"}
+                {"id": "1", "level": "info", "message": "hello world"},
             ],
-            "links": []
+            "links": [],
+        }
+
+    def test_get_batch_job_logs_level(self, api):
+        log_db = {
+            "07024ee9-7847-4b8a-b260-6c879a2b3cdc": [
+                {"id": "1", "level": "info", "message": "howdy world"},
+                {"id": "2", "level": "error", "message": "oh no"},
+            ]
         }
+        with self._fresh_job_registry():
+            with mock.patch.dict(dummy_backend.DummyBatchJobs._custom_job_logs, log_db):
+                resp = api.get(
+                    "/jobs/07024ee9-7847-4b8a-b260-6c879a2b3cdc/logs",
+                    headers=self.AUTH_HEADER,
+                    params={"level": "warning"},
+                )
+                assert resp.assert_status_code(200).json == {
+                    "level": "warning",
+                    "logs": [{"id": "2", "level": "error", "message": "oh no"}],
+                    "links": [],
+                }
 
     def test_get_batch_job_logs_failure(self, api):
         with self._fresh_job_registry():
@@ -3068,6 +3089,7 @@ def test_get_batch_job_logs_failure(self, api):
             }):
                 resp = api.get('/jobs/07024ee9-7847-4b8a-b260-6c879a2b3cdc/logs', headers=self.AUTH_HEADER)
                 assert resp.assert_status_code(200).json == {
+                    "level": "debug",
                     "logs": [
                         {
                             "id": "",
@@ -3170,6 +3192,7 @@ def test_delete_service_requires_authentication(self, api):
     def test_service_logs_100(self, api):
         logs = api.get('/services/wmts-foo/logs', headers=self.AUTH_HEADER).json
         assert logs == {
+            "level": "debug",
             "logs": [
                 {"id": 3, "level": "info", "message": "Loaded data."},
             ],