def read_logs()

in cdsresponder/rabbitmq/K8MessageProcessor.py [0:0]

• 11 lines of code
• 3 McCabe index (conditional complexity)

    def read_logs(self, job_name:str, job_namespace:str)->int:
        if self.pod_log_basepath is None:
            logger.warning("If you want pod logs to be saved, then you must set POD_LOGS_BASEPATH to a valid writable filepath")
            return 0

        pod_list:V1PodList = self.k8core.list_namespaced_pod(job_namespace, label_selector="job-name={0}".format(job_name))

        # ensure path exists
        destpath = os.path.join(self.pod_log_basepath, job_name)
        pathlib.Path(destpath).mkdir(parents=True, exist_ok=True)

        for pod in pod_list.items:
            filename = os.path.join(self.pod_log_basepath, job_name, pod.metadata.name + ".log")
            k8s.k8utils.dump_pod_logs(pod.metadata.name, pod.metadata.namespace, filename)

        return len(pod_list.items)