Show inconsistent scheduler state errors in dagit

helloworld · helloworld · commit f200174253b7 · 2020-05-21T19:52:22.000-04:00
Summary: Show consistenty issues with cron tab and schedule state in the UI Test Plan: manual Reviewers: alangenfeld Reviewed By: alangenfeld Differential Revision: https://dagster.phacility.com/D3043
diff --git a/CHANGES.md b/CHANGES.md
@@ -8,13 +8,16 @@
   the previous pipeline failed or not.
 - Added a search filter for the root Assets page
 - Adds tooltip explanations for disabled run actions
+- The last output of the cron job command created by the scheduler is now stored in a file. A new `dagster schedule logs {schedule_name}` command will show the log file for a given schedule. This helps uncover errors like missing environment variables and import errors. 
+- The dagit schedule page will now show inconsistency errors between schedule state and the cron tab that were previously only displayed by the `dagster schedule debug` command. As before, these errors can be resolve using `dagster schedule up`
 
 **Bugfix**
 
 - Fixes an issue with config schema validation on Arrays
 - Fixes an issue with initializing K8sRunLauncher when configured via `dagster.yaml`
 - Fixes a race condition in Airflow injection logic that happens when multiple Operators try to
   create PipelineRun entries simultaneously.
+- Fixed an issue with schedules that had invalid config not logging the appropriate error. 
 
 ## 0.7.13
 
diff --git a/js_modules/dagit/src/schedules/ScheduleRoot.tsx b/js_modules/dagit/src/schedules/ScheduleRoot.tsx
@@ -137,9 +137,12 @@ const RenderEventSpecificData: React.FunctionComponent<{
           minimal={true}
           href={`/runs/${data.run?.pipeline.name}/${data.run?.runId}`}
         >
-          <Tag fill={true} minimal={true} intent={Intent.SUCCESS}>
-            Run {data.run?.runId}
-          </Tag>
+          <div style={{ display: "flex" }}>
+            {data.run?.status && <RunStatus status={data.run?.status} />}
+            <Tag fill={true} minimal={true} style={{ marginLeft: 10 }}>
+              Run {data.run?.runId}
+            </Tag>
+          </div>
         </AnchorButton>
       );
   }
@@ -152,7 +155,7 @@ const TickTag: React.FunctionComponent<{ status: ScheduleTickStatus }> = ({
     case ScheduleTickStatus.STARTED:
       return (
         <Tag minimal={true} intent={Intent.PRIMARY}>
-          Success
+          Started
         </Tag>
       );
     case ScheduleTickStatus.SUCCESS:
@@ -339,6 +342,7 @@ export const SCHEDULE_ROOT_QUERY = gql`
                 pipeline {
                   name
                 }
+                status
                 runId
               }
             }
diff --git a/js_modules/dagit/src/schedules/ScheduleRow.tsx b/js_modules/dagit/src/schedules/ScheduleRow.tsx
@@ -13,7 +13,9 @@ import {
   Popover,
   Tooltip,
   Tag,
-  Intent
+  Intent,
+  PopoverInteractionKind,
+  Position
 } from "@blueprintjs/core";
 import { HighlightedCodeBlock } from "../HighlightedCodeBlock";
 import { RowColumn, RowContainer } from "../ListComponents";
@@ -51,12 +53,60 @@ const getNaturalLanguageCronString = (cronSchedule: string) => {
   }
 };
 
+const errorDisplay = (status: ScheduleStatus, runningJobCount: number) => {
+  if (status === ScheduleStatus.STOPPED && runningJobCount == 0) {
+    return null;
+  } else if (status === ScheduleStatus.RUNNING && runningJobCount == 1) {
+    return null;
+  }
+
+  const errors = [];
+  if (status === ScheduleStatus.RUNNING && runningJobCount === 0) {
+    errors.push(
+      "Schedule is set to be running, but the scheduler is not running the schedule"
+    );
+  } else if (status === ScheduleStatus.STOPPED && runningJobCount > 0) {
+    errors.push(
+      "Schedule is set to be stopped, but the scheduler is still running the schedule"
+    );
+  }
+
+  if (runningJobCount > 0) {
+    errors.push("Duplicate cron job for schedule found.");
+  }
+
+  return (
+    <Popover
+      interactionKind={PopoverInteractionKind.CLICK}
+      popoverClassName="bp3-popover-content-sizing"
+      position={Position.RIGHT}
+      fill={true}
+    >
+      <Tag fill={true} interactive={true} intent={Intent.DANGER}>
+        Error
+      </Tag>
+      <div>
+        <h3>There are errors with this schedule.</h3>
+
+        <p>To resolve, run `dagster schedule up`.</p>
+        <p>Errors:</p>
+        <ul>
+          {errors.map((error, index) => (
+            <li key={index}>{error}</li>
+          ))}
+        </ul>
+      </div>
+    </Popover>
+  );
+};
+
 export const ScheduleRow: React.FunctionComponent<{
   schedule: ScheduleFragment;
 }> = ({ schedule }) => {
   const {
     status,
     scheduleDefinition,
+    runningJobCount,
     logsPath,
     stats,
     ticks,
@@ -143,6 +193,8 @@ export const ScheduleRow: React.FunctionComponent<{
             }
           }}
         />
+
+        {errorDisplay(status, runningJobCount)}
       </RowColumn>
       <RowColumn style={{ flex: 1.4 }}>{displayName}</RowColumn>
       <RowColumn>
@@ -316,6 +368,7 @@ export const ScheduleRow: React.FunctionComponent<{
 export const ScheduleRowFragment = gql`
   fragment ScheduleFragment on RunningSchedule {
     __typename
+    runningJobCount
     scheduleDefinition {
       name
       cronSchedule
@@ -375,6 +428,7 @@ const START_SCHEDULE_MUTATION = gql`
       ... on RunningScheduleResult {
         schedule {
           __typename
+          runningJobCount
           scheduleDefinition {
             __typename
             name
@@ -397,6 +451,7 @@ const STOP_SCHEDULE_MUTATION = gql`
       ... on RunningScheduleResult {
         schedule {
           __typename
+          runningJobCount
           scheduleDefinition {
             __typename
             name
diff --git a/js_modules/dagit/src/schedules/types/ScheduleFragment.ts b/js_modules/dagit/src/schedules/types/ScheduleFragment.ts
@@ -47,6 +47,7 @@ export interface ScheduleFragment_stats {
 
 export interface ScheduleFragment {
   __typename: "RunningSchedule";
+  runningJobCount: number;
   scheduleDefinition: ScheduleFragment_scheduleDefinition;
   logsPath: string;
   ticks: ScheduleFragment_ticks[];
diff --git a/js_modules/dagit/src/schedules/types/ScheduleRootQuery.ts b/js_modules/dagit/src/schedules/types/ScheduleRootQuery.ts
@@ -59,6 +59,7 @@ export interface ScheduleRootQuery_scheduleOrError_RunningSchedule_ticksList_tic
 export interface ScheduleRootQuery_scheduleOrError_RunningSchedule_ticksList_tickSpecificData_ScheduleTickSuccessData_run {
   __typename: "PipelineRun";
   pipeline: ScheduleRootQuery_scheduleOrError_RunningSchedule_ticksList_tickSpecificData_ScheduleTickSuccessData_run_pipeline;
+  status: PipelineRunStatus;
   runId: string;
 }
 
@@ -111,6 +112,7 @@ export interface ScheduleRootQuery_scheduleOrError_RunningSchedule_attemptList {
 
 export interface ScheduleRootQuery_scheduleOrError_RunningSchedule {
   __typename: "RunningSchedule";
+  runningJobCount: number;
   scheduleDefinition: ScheduleRootQuery_scheduleOrError_RunningSchedule_scheduleDefinition;
   logsPath: string;
   ticks: ScheduleRootQuery_scheduleOrError_RunningSchedule_ticks[];
diff --git a/js_modules/dagit/src/schedules/types/SchedulesRootQuery.ts b/js_modules/dagit/src/schedules/types/SchedulesRootQuery.ts
@@ -52,6 +52,7 @@ export interface SchedulesRootQuery_scheduler_Scheduler_runningSchedules_stats {
 
 export interface SchedulesRootQuery_scheduler_Scheduler_runningSchedules {
   __typename: "RunningSchedule";
+  runningJobCount: number;
   scheduleDefinition: SchedulesRootQuery_scheduler_Scheduler_runningSchedules_scheduleDefinition;
   logsPath: string;
   ticks: SchedulesRootQuery_scheduler_Scheduler_runningSchedules_ticks[];
diff --git a/js_modules/dagit/src/schedules/types/StartSchedule.ts b/js_modules/dagit/src/schedules/types/StartSchedule.ts
@@ -16,6 +16,7 @@ export interface StartSchedule_startSchedule_RunningScheduleResult_schedule_sche
 
 export interface StartSchedule_startSchedule_RunningScheduleResult_schedule {
   __typename: "RunningSchedule";
+  runningJobCount: number;
   scheduleDefinition: StartSchedule_startSchedule_RunningScheduleResult_schedule_scheduleDefinition;
   status: ScheduleStatus;
 }
diff --git a/js_modules/dagit/src/schedules/types/StopSchedule.ts b/js_modules/dagit/src/schedules/types/StopSchedule.ts
@@ -16,6 +16,7 @@ export interface StopSchedule_stopRunningSchedule_RunningScheduleResult_schedule
 
 export interface StopSchedule_stopRunningSchedule_RunningScheduleResult_schedule {
   __typename: "RunningSchedule";
+  runningJobCount: number;
   scheduleDefinition: StopSchedule_stopRunningSchedule_RunningScheduleResult_schedule_scheduleDefinition;
   status: ScheduleStatus;
 }
diff --git a/js_modules/dagit/src/schema.graphql b/js_modules/dagit/src/schema.graphql
@@ -1003,6 +1003,7 @@ type RunningSchedule {
   attempts(limit: Int): [ScheduleAttempt!]!
   attemptsCount: Int!
   logsPath: String!
+  runningJobCount: Int!
 }
 
 type RunningScheduleResult {
diff --git a/python_modules/dagster-graphql/dagster_graphql/schema/schedules.py b/python_modules/dagster-graphql/dagster_graphql/schema/schedules.py
@@ -208,6 +208,7 @@ class Meta(object):
     attempts = dauphin.Field(dauphin.non_null_list('ScheduleAttempt'), limit=dauphin.Int())
     attempts_count = dauphin.NonNull(dauphin.Int)
     logs_path = dauphin.NonNull(dauphin.String)
+    running_job_count = dauphin.NonNull(dauphin.Int)
 
     def __init__(self, graphene_info, schedule):
         self._schedule = check.inst_param(schedule, 'schedule', Schedule)
@@ -222,6 +223,13 @@ def __init__(self, graphene_info, schedule):
             repository_path=schedule.repository_path,
         )
 
+    def resolve_running_job_count(self, graphene_info):
+        repository = graphene_info.context.get_repository()
+        running_job_count = graphene_info.context.instance.running_job_count(
+            repository.name, self._schedule.name
+        )
+        return running_job_count
+
     # TODO: Delete in 0.8.0 release
     # https://github.com/dagster-io/dagster/issues/228
     def resolve_attempts(self, graphene_info, **kwargs):
diff --git a/python_modules/dagster/dagster/cli/schedule.py b/python_modules/dagster/dagster/cli/schedule.py
@@ -59,10 +59,13 @@ def repository_target_argument(f):
 
 
 def print_changes(repository, instance, print_fn=print, preview=False):
+    debug_info = instance.scheduler_debug_info()
+    errors = debug_info.errors
     changeset = get_schedule_change_set(
         instance.all_schedules(repository), repository.schedule_defs
     )
-    if len(changeset) == 0:
+
+    if len(errors) == 0 and len(changeset) == 0:
         if preview:
             print_fn(click.style('No planned changes to schedules.', fg='magenta', bold=True))
             print_fn(
@@ -73,7 +76,20 @@ def print_changes(repository, instance, print_fn=print, preview=False):
             print_fn('{num} schedules unchanged'.format(num=len(repository.schedule_defs)))
         return
 
-    print_fn(click.style('Planned Changes:' if preview else 'Changes:', fg='magenta', bold=True))
+    if len(errors):
+        print_fn(
+            click.style(
+                'Planned Error Fixes:' if preview else 'Errors Resolved:', fg='magenta', bold=True
+            )
+        )
+        print_fn("\n".join(debug_info.errors))
+
+    if len(changeset):
+        print_fn(
+            click.style(
+                'Planned Schedule Changes:' if preview else 'Changes:', fg='magenta', bold=True
+            )
+        )
 
     for change in changeset:
         change_type, schedule_name, changes = change
@@ -415,7 +431,7 @@ def execute_debug_command(print_fn):
 
     errors = debug_info.errors
     if len(errors):
-        title = "Errors"
+        title = "Errors (Run `dagster schedule up` to resolve)"
         output += "\n{title}\n{sep}\n{info}\n\n".format(
             title=title, sep="=" * len(title), info="\n".join(debug_info.errors),
         )
diff --git a/python_modules/dagster/dagster/core/instance/__init__.py b/python_modules/dagster/dagster/core/instance/__init__.py
@@ -811,6 +811,9 @@ def stop_schedule(self, repository, schedule_name):
     def end_schedule(self, repository, schedule_name):
         return self._scheduler.end_schedule(self, repository, schedule_name)
 
+    def running_job_count(self, repository_name, schedule_name):
+        return self._scheduler.running_job_count(repository_name, schedule_name)
+
     def scheduler_debug_info(self):
         from dagster.core.scheduler import SchedulerDebugInfo, ScheduleStatus
 
@@ -819,32 +822,25 @@ def scheduler_debug_info(self):
         schedule_info = self.all_schedules_info()
         schedules = []
         for repository_name, schedule in schedule_info:
-            if (
-                schedule.status == ScheduleStatus.RUNNING
-                and not self._scheduler.is_scheduler_job_running(repository_name, schedule.name)
+            if schedule.status == ScheduleStatus.RUNNING and not self.running_job_count(
+                repository_name, schedule.name
             ):
                 errors.append(
                     "Schedule {schedule_name} is set to be running, but the scheduler is not "
-                    "running the schedule. Run `dagster schedule up` to resolve".format(
-                        schedule_name=schedule.name
-                    )
+                    "running the schedule.".format(schedule_name=schedule.name)
                 )
-            elif (
-                schedule.status == ScheduleStatus.STOPPED
-                and self._scheduler.is_scheduler_job_running(repository_name, schedule.name)
+            elif schedule.status == ScheduleStatus.STOPPED and self.running_job_count(
+                repository_name, schedule.name
             ):
                 errors.append(
                     "Schedule {schedule_name} is set to be stopped, but the scheduler is still running "
-                    "the schedule. Run `dagster schedule up` to resolve".format(
-                        schedule_name=schedule.name
-                    )
+                    "the schedule.".format(schedule_name=schedule.name)
                 )
 
-            if self._scheduler.is_scheduler_job_running(repository_name, schedule.name) > 1:
+            if self.running_job_count(repository_name, schedule.name) > 1:
                 errors.append(
                     "Duplicate jobs found: More than one job for schedule {schedule_name} are "
-                    "running on the scheduler.  "
-                    "Run `dagster schedule up` to resolve".format(schedule_name=schedule.name)
+                    "running on the scheduler.".format(schedule_name=schedule.name)
                 )
 
             schedule_info = {
diff --git a/python_modules/dagster/dagster/core/scheduler/scheduler.py b/python_modules/dagster/dagster/core/scheduler/scheduler.py
@@ -174,7 +174,7 @@ def end_schedule(self, instance, repository, schedule_name):
         '''
 
     @abc.abstractmethod
-    def is_scheduler_job_running(self, repository_name, schedule_name):
+    def running_job_count(self, repository_name, schedule_name):
         '''Resume a pipeline schedule.
 
         Args:
diff --git a/python_modules/dagster/dagster/utils/test/__init__.py b/python_modules/dagster/dagster/utils/test/__init__.py
@@ -426,9 +426,9 @@ def stop_schedule(self, instance, repository, schedule_name):
         instance.update_schedule(repository, stopped_schedule)
         return stopped_schedule
 
-    def is_scheduler_job_running(self, repository_name, schedule_name):
+    def running_job_count(self, repository_name, schedule_name):
         # Not currently tested in dagster core
-        return None
+        return 0
 
     def end_schedule(self, instance, repository, schedule_name):
         schedule = instance.get_schedule_by_name(repository, schedule_name)
diff --git a/python_modules/dagster/dagster_tests/cli_tests/test_cli_commands.py b/python_modules/dagster/dagster_tests/cli_tests/test_cli_commands.py
@@ -807,7 +807,7 @@ def test_schedules_wipe_correct_delete_message(_patch_scheduler_instance):
 
     # Verify schedules were wiped
     assert result.exit_code == 0
-    assert result.output == 'Planned Changes:\n  + foo_schedule (add)\n'
+    assert result.output == 'Planned Schedule Changes:\n  + foo_schedule (add)\n'
 
 
 def test_schedules_wipe_incorrect_delete_message(_patch_scheduler_instance):
@@ -892,7 +892,7 @@ def test_schedules_logs(_patch_scheduler_instance):
     )
 
     assert result.exit_code == 0
-    assert result.output.endswith('/scheduler.log\n')
+    assert result.output.endswith('scheduler.log\n')
 
 
 @pytest.mark.skipif(
diff --git a/python_modules/libraries/dagster-cron/dagster_cron/cron_scheduler.py b/python_modules/libraries/dagster-cron/dagster_cron/cron_scheduler.py
@@ -64,7 +64,7 @@ def start_schedule(self, instance, repository, schedule_name):
         self._start_cron_job(instance, repository, started_schedule)
 
         # Check that the schedule made it to the cron tab
-        if not self.is_scheduler_job_running(repository.name, schedule.name):
+        if not self.running_job_count(repository.name, schedule.name):
             raise DagsterInvariantViolationError(
                 "Attempted to write cron job for schedule {schedule_name}, but failed".format(
                     schedule_name=schedule.name
@@ -86,7 +86,7 @@ def stop_schedule(self, instance, repository, schedule_name):
         stopped_schedule = schedule.with_status(ScheduleStatus.STOPPED)
         self._end_cron_job(instance, repository, stopped_schedule)
 
-        if self.is_scheduler_job_running(repository.name, schedule.name):
+        if self.running_job_count(repository.name, schedule.name):
             raise DagsterInvariantViolationError(
                 "Attempted to remove cron job for schedule {schedule_name}, but failed. The cron "
                 "job for the schedule is still running".format(schedule_name=schedule.name)
@@ -176,8 +176,9 @@ def _end_cron_job(self, instance, repository, schedule):
         if os.path.isfile(script_file):
             os.remove(script_file)
 
-    def is_scheduler_job_running(self, repository_name, schedule_name):
-        matching_jobs = self._cron_tab.find_comment(
+    def running_job_count(self, repository_name, schedule_name):
+        cron_tab = CronTab(user=True)
+        matching_jobs = cron_tab.find_comment(
             self._cron_tag_for_schedule(repository_name, schedule_name)
         )
 

Original file line number	Diff line number	Diff line change
`@@ -16,6 +16,7 @@ export interface StartSchedule_startSchedule_RunningScheduleResult_schedule_sche`
`16`	`16`
`17`	`17`	`export interface StartSchedule_startSchedule_RunningScheduleResult_schedule {`
`18`	`18`	`__typename: "RunningSchedule";`
	`19`	`+ runningJobCount: number;`
`19`	`20`	`scheduleDefinition: StartSchedule_startSchedule_RunningScheduleResult_schedule_scheduleDefinition;`
`20`	`21`	`status: ScheduleStatus;`
`21`	`22`	`}`
Original file line number	Diff line number	Diff line change
`@@ -16,6 +16,7 @@ export interface StopSchedule_stopRunningSchedule_RunningScheduleResult_schedule`
`16`	`16`
`17`	`17`	`export interface StopSchedule_stopRunningSchedule_RunningScheduleResult_schedule {`
`18`	`18`	`__typename: "RunningSchedule";`
	`19`	`+ runningJobCount: number;`
`19`	`20`	`scheduleDefinition: StopSchedule_stopRunningSchedule_RunningScheduleResult_schedule_scheduleDefinition;`
`20`	`21`	`status: ScheduleStatus;`
`21`	`22`	`}`
Original file line number	Diff line number	Diff line change
`@@ -1003,6 +1003,7 @@ type RunningSchedule {`
`1003`	`1003`	`attempts(limit: Int): [ScheduleAttempt!]!`
`1004`	`1004`	`attemptsCount: Int!`
`1005`	`1005`	`logsPath: String!`
	`1006`	`+ runningJobCount: Int!`
`1006`	`1007`	`}`
`1007`	`1008`
`1008`	`1009`	`type RunningScheduleResult {`