Add more detail to schedule logs command

Summary: Previously, the schedule log command only showed the top level log file, which captured errors at the cron and bash script level. This diff updates the schedule log command to also link to the individual execution logs, which are needed to debug user-code error issues.

Test Plan: Run `dagster schedule logs` in the toys directory for (1) a schedule with result logs and (2) a schedule without result logs

Reviewers: dgibson

Reviewed By: dgibson

Differential Revision: https://dagster.phacility.com/D5653

python_modules/dagster/dagster/cli/schedule.py

@@ -1,3 +1,4 @@
+import glob
 import os
 
 import click
@@ -10,6 +11,7 @@
 from dagster.core.host_representation import ExternalRepository
 from dagster.core.instance import DagsterInstance
 from dagster.core.scheduler.job import JobStatus
+from dagster.core.scheduler.scheduler import DagsterDaemonScheduler
 
 
 def create_schedule_cli_group():
@@ -371,12 +373,46 @@ def execute_logs_command(schedule_name, cli_args, print_fn, instance=None):
  with DagsterInstance.get() as instance:
  with get_external_repository_from_kwargs(cli_args) as external_repo:
  check_repo_and_scheduler(external_repo, instance)
+
+ if isinstance(instance.scheduler, DagsterDaemonScheduler):
+ return print_fn(
+ "This command is deprecated for the DagsterDaemonScheduler. "
+ "Logs for the DagsterDaemonScheduler written to the process output. "
+ "For help troubleshooting the Daemon Scheduler, see "
+ "https://docs.dagster.io/troubleshooting/schedules"
+ )
+
  logs_path = os.path.join(
  instance.logs_path_for_schedule(
  external_repo.get_external_schedule(schedule_name).get_external_origin_id()
  )
  )
- print_fn(logs_path)
+
+ logs_directory = os.path.dirname(logs_path)
+ result_files = glob.glob("{}/*.result".format(logs_directory))
+ most_recent_log = max(result_files, key=os.path.getctime) if result_files else None
+
+ output = ""
+
+ title = "Scheduler Logs:"
+ output += "{title}\n{sep}\n{info}\n".format(
+ title=title, sep="=" * len(title), info=logs_path,
+ )
+
+ title = (
+ "Schedule Execution Logs:"
+ "\nEvent logs from schedule execution. "
+ "Errors that caused schedule executions to not run or fail can be found here. "
+ )
+ most_recent_info = (
+ "\nMost recent execution log: {}".format(most_recent_log) if most_recent_log else ""
+ )
+ info = "All execution logs: {}{}".format(logs_directory, most_recent_info)
+ output += "\n{title}\n{sep}\n{info}\n".format(
+ title=title, sep="=" * len(title), info=info,
+ )
+
+ print_fn(output)
 
 
 @click.command(name="restart", help="Restart a running schedule")

python_modules/dagster/dagster_tests/cli_tests/command_tests/test_schedule_commands.py

@@ -211,10 +211,10 @@ def test_schedules_logs(gen_schedule_args):
  _instance.return_value = instance
  runner = CliRunner()
 
- result = runner.invoke(schedule_logs_command, cli_args + ["foo_schedule"],)
+ result = runner.invoke(schedule_logs_command, cli_args + ["foo_schedule"])
 
  assert result.exit_code == 0
- assert result.output.endswith("scheduler.log\n")
+ assert "scheduler.log" in result.output
 
 
 def test_check_repo_and_scheduler_no_external_schedules():