feat(cli): add --follow flag to logs command (#731)

AUTHORS.md

@@ -15,6 +15,7 @@ The list of contributors in alphabetical order:
 - [Giuseppe Steduto](https://orcid.org/0009-0002-1258-8553)
 - [Harri Hirvonsalo](https://orcid.org/0000-0002-5503-510X)
 - [Jan Okraska](https://orcid.org/0000-0002-1416-3244)
+- [Jelizaveta Lemeševa](https://orcid.org/0009-0003-6606-9270)
 - [Leticia Wanderley](https://orcid.org/0000-0003-4649-6630)
 - [Marco Donadoni](https://orcid.org/0000-0003-2922-5505)
 - [Marco Vidal](https://orcid.org/0000-0002-9363-4971)

reana_client/cli/utils.py

@@ -12,6 +12,7 @@
 import os
 import shlex
 import sys
+import time
 from typing import Callable, NoReturn, Optional, List, Tuple, Union
 
 import click
@@ -24,6 +25,8 @@
  RUN_STATUSES,
  JOB_STATUS_TO_MSG_COLOR,
  JSON,
+ CLI_LOGS_FOLLOW_MIN_INTERVAL,
+ CLI_LOGS_FOLLOW_DEFAULT_INTERVAL,
 )
 from reana_client.printer import display_message
 from reana_client.utils import workflow_uuid_or_name
@@ -409,3 +412,114 @@ def output_user_friendly_logs(workflow_logs, steps):
  f"Step {job_name_or_id} emitted no logs.",
  msg_type="info",
  )
+
+
+def retrieve_workflow_logs(
+ workflow,
+ access_token,
+ json_format,
+ filters,
+ steps,
+ chosen_filters,
+ available_filters,
+ page=None,
+ size=None,
+): # noqa: D301
+ """Retrieve workflow logs."""
+ from reana_client.api.client import get_workflow_logs
+
+ response = get_workflow_logs(
+ workflow,
+ access_token,
+ steps=None if not steps else list(set(steps)),
+ page=page,
+ size=size,
+ )
+ workflow_logs = json.loads(response["logs"])
+ if filters:
+ for key, value in chosen_filters.items():
+ unwanted_steps = [
+ k
+ for k, v in workflow_logs["job_logs"].items()
+ if v[available_filters[key]] != value
+ ]
+ for job_id in unwanted_steps:
+ del workflow_logs["job_logs"][job_id]
+
+ if json_format:
+ display_message(json.dumps(workflow_logs, indent=2))
+ sys.exit(0)
+ else:
+ from reana_client.cli.utils import output_user_friendly_logs
+
+ output_user_friendly_logs(workflow_logs, None if not steps else list(set(steps)))
+
+
+def follow_workflow_logs(
+ workflow,
+ access_token,
+ interval,
+ steps,
+): # noqa: D301
+ """Continuously poll for workflow or job logs."""
+ from reana_client.api.client import get_workflow_logs, get_workflow_status
+
+ if len(steps) > 1:
+ display_message(
+ "Only one step can be followed at a time, ignoring additional steps.",
+ "warning",
+ )
+ if interval < CLI_LOGS_FOLLOW_MIN_INTERVAL:
+ interval = CLI_LOGS_FOLLOW_DEFAULT_INTERVAL
+ display_message(
+ f"Interval should be an integer greater than or equal to {CLI_LOGS_FOLLOW_MIN_INTERVAL}, resetting to default ({CLI_LOGS_FOLLOW_DEFAULT_INTERVAL} s).",
+ "warning",
+ )
+ step = steps[0] if steps else None
+
+ previous_logs = ""
+
+ while True:
+ response = get_workflow_logs(
+ workflow,
+ access_token,
+ steps=None if not step else [step],
+ )
+ if response.get("live_logs_enabled", False) is False:
+ display_message(
+ "Live logs are not enabled, please rerun the command without the --follow flag.",
+ "error",
+ )
+ return
+
+ json_response = json.loads(response.get("logs"))
+
+ if step:
+ jobs = json_response["job_logs"]
+
+ if not jobs:
+ raise Exception(f"Step data not found: {step}")
+
+ job = next(iter(jobs.values())) # get values of the first job
+ logs = job["logs"]
+ status = job["status"]
+ else:
+ logs = json_response["workflow_logs"]
+ status = get_workflow_status(workflow, access_token).get("status")
+
+ previous_lines = previous_logs.splitlines()
+ new_lines = logs.splitlines()
+
+ diff = "\n".join([x for x in new_lines if x not in previous_lines])
+ if diff != "" and diff != "\n":
+ display_message(diff)
+
+ if status in ["finished", "failed", "stopped", "deleted"]:
+ subject = "Workflow" if not step else "Job"
+ display_message(
+ f"{subject} has completed, you might want to rerun the command without the --follow flag.",
+ "info",
+ )
+ return
+ previous_logs = logs
+ time.sleep(interval)

reana_client/cli/workflow.py

@@ -31,8 +31,15 @@
  key_value_to_dict,
  parse_filter_parameters,
  requires_environments,
+ retrieve_workflow_logs,
+ follow_workflow_logs,
+)
+from reana_client.config import (
+ ERROR_MESSAGES,
+ RUN_STATUSES,
+ TIMECHECK,
+ CLI_LOGS_FOLLOW_DEFAULT_INTERVAL,
 )
-from reana_client.config import ERROR_MESSAGES, RUN_STATUSES, TIMECHECK
 from reana_client.printer import display_message
 from reana_client.utils import (
  get_reana_yaml_file_path,
@@ -886,6 +893,20 @@ def add_verbose_data_from_response(response, verbose_headers, headers, data):
  multiple=True,
  help="Filter job logs to include only those steps that match certain filtering criteria. Use --filter name=value pairs. Available filters are compute_backend, docker_img, status and step.",
 )
+@click.option(
+ "--follow",
+ "follow",
+ is_flag=True,
+ default=False,
+ help="Follow the logs of a running workflow or job (similar to tail -f).",
+)
+@click.option(
+ "-i",
+ "--interval",
+ "interval",
+ default=CLI_LOGS_FOLLOW_DEFAULT_INTERVAL,
+ help=f"Sleep time in seconds between log polling if log following is enabled. [default={CLI_LOGS_FOLLOW_DEFAULT_INTERVAL}]",
+)
 @add_pagination_options
 @check_connection
 @click.pass_context
@@ -894,22 +915,32 @@ def workflow_logs(
  workflow,
  access_token,
  json_format,
- steps=None,
+ follow,
+ interval,
  filters=None,
  page=None,
  size=None,
 ): # noqa: D301
  """Get workflow logs.
 
- The ``logs`` command allows to retrieve logs of running workflow. Note that
- only finished steps of the workflow are returned, the logs of the currently
- processed step is not returned until it is finished.
+ The ``logs`` command allows to retrieve logs of a running workflow.
+ Either retrive logs and print the result or follow the logs of a running workflow/job.
 
  Examples:\n
- \t $ reana-client logs -w myanalysis.42
- \t $ reana-client logs -w myanalysis.42 -s 1st_step
+ \t $ reana-client logs -w myanalysis.42\n
+ \t $ reana-client logs -w myanalysis.42 --json\n
+ \t $ reana-client logs -w myanalysis.42 --filter status=running\n
+ \t $ reana-client logs -w myanalysis.42 --filter step=myfit --follow\n
  """
- from reana_client.api.client import get_workflow_logs
+ logging.debug("command: {}".format(ctx.command_path.replace(" ", ".")))
+ for p in ctx.params:
+ logging.debug("{param}: {value}".format(param=p, value=ctx.params[p]))
+
+ if json_format and follow:
+ display_message(
+ "Ignoring --json as it cannot be used together with --follow.",
+ msg_type="warning",
+ )
 
  available_filters = {
  "step": "job_name",
@@ -920,90 +951,73 @@ def workflow_logs(
  steps = []
  chosen_filters = dict()
 
- logging.debug("command: {}".format(ctx.command_path.replace(" ", ".")))
- for p in ctx.params:
- logging.debug("{param}: {value}".format(param=p, value=ctx.params[p]))
- if workflow:
- if filters:
- try:
- for f in filters:
- key, value = f.split("=")
- if key not in available_filters:
+ if filters:
+ try:
+ for f in filters:
+ key, value = f.split("=")
+ if key not in available_filters:
+ display_message(
+ "Filter '{}' is not valid.\n"
+ "Available filters are '{}'.".format(
+ key,
+ "' '".join(sorted(available_filters.keys())),
+ ),
+ msg_type="error",
+ )
+ sys.exit(1)
+ elif key == "step":
+ steps.append(value)
+ else:
+ # Case insensitive for compute backends
+ if (
+ key == "compute_backend"
+ and value.lower() in REANA_COMPUTE_BACKENDS
+ ):
+ value = REANA_COMPUTE_BACKENDS[value.lower()]
+ elif key == "status" and value not in RUN_STATUSES:
  display_message(
- "Filter '{}' is not valid.\n"
- "Available filters are '{}'.".format(
- key,
- "' '".join(sorted(available_filters.keys())),
- ),
+ "Input status value {} is not valid. ".format(value),
  msg_type="error",
- )
+ ),
  sys.exit(1)
- elif key == "step":
- steps.append(value)
- else:
- # Case insensitive for compute backends
- if (
- key == "compute_backend"
- and value.lower() in REANA_COMPUTE_BACKENDS
- ):
- value = REANA_COMPUTE_BACKENDS[value.lower()]
- elif key == "status" and value not in RUN_STATUSES:
- display_message(
- "Input status value {} is not valid. ".format(value),
- msg_type="error",
- ),
- sys.exit(1)
- chosen_filters[key] = value
- except Exception as e:
- logging.debug(traceback.format_exc())
- logging.debug(str(e))
- display_message(
- "Please provide complete --filter name=value pairs, "
- "for example --filter status=running.\n"
- "Available filters are '{}'.".format(
- "' '".join(sorted(available_filters.keys()))
- ),
- msg_type="error",
- )
- sys.exit(1)
- try:
- response = get_workflow_logs(
- workflow,
- access_token,
- steps=None if not steps else list(set(steps)),
- page=page,
- size=size,
- )
- workflow_logs = json.loads(response["logs"])
- if filters:
- for key, value in chosen_filters.items():
- unwanted_steps = [
- k
- for k, v in workflow_logs["job_logs"].items()
- if v[available_filters[key]] != value
- ]
- for job_id in unwanted_steps:
- del workflow_logs["job_logs"][job_id]
-
- if json_format:
- display_message(json.dumps(workflow_logs, indent=2))
- sys.exit(0)
- else:
- from reana_client.cli.utils import output_user_friendly_logs
-
- output_user_friendly_logs(
- workflow_logs, None if not steps else list(set(steps))
- )
+ chosen_filters[key] = value
  except Exception as e:
  logging.debug(traceback.format_exc())
  logging.debug(str(e))
  display_message(
- "Cannot retrieve the logs of a workflow {}: \n"
- "{}".format(workflow, str(e)),
+ "Please provide complete --filter name=value pairs, "
+ "for example --filter status=running.\n"
+ "Available filters are '{}'.".format(
+ "' '".join(sorted(available_filters.keys()))
+ ),
  msg_type="error",
  )
  sys.exit(1)
 
+ try:
+ if follow:
+ follow_workflow_logs(workflow, access_token, interval, steps)
+ else:
+ retrieve_workflow_logs(
+ workflow,
+ access_token,
+ json_format,
+ filters,
+ steps,
+ chosen_filters,
+ available_filters,
+ page,
+ size,
+ )
+ except Exception as e:
+ logging.debug(traceback.format_exc())
+ logging.debug(str(e))
+ display_message(
+ "Cannot retrieve logs for workflow {}: \n{}".format(workflow, str(e)),
+ msg_type="error",
+ )
+ sys.exit(1)
+
 
 @workflow_execution_group.command("validate")
 @click.option(

reana_client/config.py

@@ -77,3 +77,9 @@
 
 STD_OUTPUT_CHAR = "-"
 """Character used to refer to the standard output."""
+
+CLI_LOGS_FOLLOW_MIN_INTERVAL = 1
+"""Minimum interval between log requests in seconds."""
+
+CLI_LOGS_FOLLOW_DEFAULT_INTERVAL = 10
+"""Default interval between log requests in seconds."""