Add joblog label to job

CONFIG.md

@@ -46,6 +46,36 @@ For Kubernetes, it is important to set resource limits.
 TODO: explain how to set limits, with default, project and spider specificity.
 
 ### [joblogs] section
+
+The joblogs section is used to configure the job logs feature. It is not enabled by default, but can be activated if you
+choose to use it. The job logs feature allows you to collect logs from the Kubernetes cluster and store them in a specified
+directory. The logs are collected from the pods running on the cluster and can be compressed using a specified method.
+The job logs feature is implemented only for Kubernetes and is not available for Docker.
+
+When the log file was uploaded to the specified storage, the pod is labeled with org.scrapy.log_file_uploaded=true, it also
+contains labels to specify an end location of the log files in the storage.
+Since the label field has limitation in the number of symbols, an end location is split into several labels, each of them
+contains a part of the end location. The labels are named as follows:
+  * org.scrapy.project
+  * org.scrapy.spider
+  * org.scrapy.job_id
+  * org.scrapy.extension
+
+How can you build the end location from these labels? Complete the following line
+  * logs/`org.scrapy.project`/`org.scrapy.spider`/`org.scrapy.job_id`.`org.scrapy.extension` 
+
+**Important**: if you used the `compression_method` parameter, the extension of the log file then have two extensions. The
+first extension is `.log` and the second extension is extracted to the label `org.scrapy.extension`. 
+
+For example, if you used
+`compression_method = gzip`, the log file will be named as 
+  * logs/`org.scrapy.project`/`org.scrapy.spider`/`org.scrapy.job_id`.log.`org.scrapy.extension`
+
+If you haven't specified a compression method, then none was applied and the final location of the log file is
+* logs/`org.scrapy.project`/`org.scrapy.spider`/`org.scrapy.job_id`.`org.scrapy.extension`
+
+The joblogs section contains the following parameters:
+
   * `logs_dir`     - a directory to store log files collected on k8s cluster (implemented only for Kubernetes). If you are using a Persistent Volume, keep in mind, that the provided path should be mounted in the deployment manifest. Read and write permissions should be granted to allow actions with log files in the provided directory, thus a [securityContext](https://kubernetes.io/docs/tasks/configure-pod-container/security-context/) was added to the deployment manifest.
   * `compression_method` - a method to compress log files. Available options are `gzip` `bzip2`, `lzma`, `brotli` and `none`. If no compression_method is provided, it defaults to `none`.
 

kubernetes.yaml

@@ -166,6 +166,7 @@ metadata:
 rules:
   - apiGroups: [""]
     resources: ["pods"]
+    # add "patch" if you use joblogs feature
     verbs: ["get", "list", "watch"]
   - apiGroups: [""]
     resources: ["pods/exec"]

scrapyd_k8s/joblogs/log_handler_k8s.py

@@ -84,6 +84,62 @@ def __init__(self, config):
             raise ValueError("Configuration error: 'logs_dir' is missing in joblogs configuration section.")
         self.object_storage_provider = LibcloudObjectStorage(self.config)
 
+    def _add_object_name_label(self, pod, object_name):
+        """
+        Adds extension as label to the pod.
+        Adds label that the log file was uploaded.
+
+        Parameters
+        ----------
+        pod : V1Pod
+            The Kubernetes pod object.
+        object_name : str
+            The object name in storage.
+
+        Returns
+        -------
+        bool
+            True if the label was added successfully, False otherwise.
+        """
+        try:
+            extension = os.path.splitext(object_name)[1][1:]
+            v1 = client.CoreV1Api()
+
+            # Create a body for the patch operation
+            patch_extension = {
+                "metadata": {
+                    "labels": {
+                        "org.scrapy.log_file_extension": extension
+                    }
+                }
+            }
+
+            patch_upload_label = {
+                "metadata": {
+                    "labels": {
+                        "org.scrapy.log_file_uploaded": "true"
+                    }
+                }
+            }
+
+            # Patch the pod with the new label
+            v1.patch_namespaced_pod(
+                name=pod.metadata.name,
+                namespace=self.namespace,
+                body=patch_extension
+            )
+            v1.patch_namespaced_pod(
+                name=pod.metadata.name,
+                namespace=self.namespace,
+                body=patch_upload_label
+            )
+
+            logger.info(f"Added log_file_extension labels '{extension}' and log_file_uploaded: 'true' to pod '{pod.metadata.name}'")
+            return True
+        except Exception as e:
+            logger.error(f"Failed to add labels to pod '{pod.metadata.name}': {e}")
+            return False
+
     def get_existing_log_filename(self, job_id):
         """
         Retrieves the existing temporary log file path for a job without creating a new one.
@@ -322,7 +378,12 @@ def handle_events(self, event):
                                 logger.info(
                                     f"Removed local log file '{log_filename}' since it already exists in storage.")
                         else:
-                            self.object_storage_provider.upload_file(project, spider, log_filename)
+                            object_name = self.object_storage_provider.upload_file(project, spider, log_filename)
+                            if object_name:
+                                try:
+                                    self._add_object_name_label(pod, object_name)
+                                except Exception as e:
+                                    logger.error(f"Exception while adding label for job '{job_id}': {e}")
                             os.remove(log_filename)
                             logger.info(f"Removed local log file '{log_filename}' after successful upload.")
                     else:

scrapyd_k8s/object_storage/libcloud_driver.py

@@ -133,7 +133,8 @@ def upload_file(self, project, spider, local_path):
 
         Returns
         -------
-        None
+        str or None
+            The object name in storage if upload is successful, None otherwise.
 
         Logs
         ----
@@ -172,6 +173,10 @@ def upload_file(self, project, spider, local_path):
                     f"Successfully uploaded compressed file '{object_name}' to container '{self._container_name}'.")
             else:
                 logger.info(f"Successfully uploaded file '{object_name}' to container '{self._container_name}'.")
+
+            # Return object_name on successful upload
+            return object_name
+
         except (ObjectError, ContainerDoesNotExistError, InvalidContainerNameError) as e:
             logger.exception(f"Error uploading the file '{object_name}': {e}")
         except Exception as e: