Issue #170 Add level parameter to get_log_entries.

Setting minimum level should help to limit the amount of memory used in
Elasticsearch while retrieving logs.
The cause of the circuit_breaker_exception is that amount of memory
used is getting too close to the maximum limit allowed.

openeo_driver/_version.py

@@ -1 +1 @@
-__version__ = "0.39.1a1"
+__version__ = "0.40.0a1"

openeo_driver/backend.py

@@ -439,7 +439,13 @@ def get_results(self, job_id: str, user_id: str) -> Dict[str, dict]:
         # TODO: eliminate this method in favor of `get_result_assets`
         raise NotImplementedError
 
-    def get_log_entries(self, job_id: str, user_id: str, offset: Optional[str] = None) -> Iterable[dict]:
+    def get_log_entries(
+        self,
+        job_id: str,
+        user_id: str,
+        offset: Optional[str] = None,
+        level: Optional[str] = None,
+    ) -> Iterable[dict]:
         """
         https://openeo.org/documentation/1.0/developers/api/reference.html#operation/debug-job
         """

openeo_driver/dummy/dummy_backend.py

@@ -13,6 +13,7 @@
 from shapely.geometry.base import BaseGeometry, BaseMultipartGeometry
 from shapely.geometry.collection import GeometryCollection
 
+from openeo.api.logs import normalize_log_level
 from openeo.internal.process_graph_visitor import ProcessGraphVisitor
 from openeo.metadata import CollectionMetadata, Band
 from openeo_driver.ProcessGraphDeserializer import ConcreteProcessing
@@ -731,11 +732,21 @@ 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) -> Iterable[dict]:
+    def get_log_entries(
+        self,
+        job_id: str,
+        user_id: str,
+        offset: Optional[str] = None,
+        level: Optional[str] = None,
+    ) -> 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"))
+                if actual_level < requested_level:
+                    continue
                 yield log
             elif isinstance(log, Exception):
                 raise log

openeo_driver/views.py

@@ -1255,12 +1255,15 @@ def download_job_result_signed(job_id, user_base64, secure_key, filename):
         return _download_job_result(job_id=job_id, filename=filename, user_id=user_id)
 
     @api_endpoint
-    @blueprint.route('/jobs/<job_id>/logs', methods=['GET'])
+    @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')
+        offset = request.args.get("offset")
+        level = request.args.get("level")
         # TODO: implement paging support: `limit`, next/prev/first/last `links`, ...
-        logs = backend_implementation.batch_jobs.get_log_entries(job_id=job_id, user_id=user.user_id, offset=offset)
+        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":["""