ExecutionDescriptionHandler implements OutputHandler { @Override public void transitionInto(RunState state, EventRouter eventRouter) { final WorkflowInstance workflowInstance = state.workflowInstance(); switch (state.state()) { case PREPARE: try { final Event submitEvent = Event.submit( state.workflowInstance(), getExecDescription(workflowInstance, state.data()), createExecutionId()); try { eventRouter.receive(submitEvent, state.counter()); } catch (IsClosedException isClosedException) { LOG.warn("Could not send 'submit' event", isClosedException); } } catch (ResourceNotFoundException e) { LOG.info("Halting {}: {}", workflowInstance, e.getMessage()); eventRouter.receiveIgnoreClosed(Event.halt(workflowInstance), state.counter()); } catch (MissingRequiredPropertyException e) { LOG.warn("Failed to prepare execution description for " + state.workflowInstance(), e); eventRouter.receiveIgnoreClosed(Event.halt(workflowInstance), state.counter()); } catch (IOException e) { try { LOG.error("Failed to retrieve execution description for " + state.workflowInstance(), e); eventRouter.receive(Event.runError(state.workflowInstance(), e.getMessage()), state.counter()); } catch (IsClosedException isClosedException) { LOG.warn("Failed to send 'runError' event", isClosedException); } } break; default: } } ExecutionDescriptionHandler(
      Storage storage,
      WorkflowValidator validator); @Override void transitionInto(RunState state, EventRouter eventRouter);  }
@Test public void shouldTransitionIntoSubmittingIfMissingDockerArgs() throws Exception { var workflow = Workflow.create("id", workflowConfiguration()); var workflowState = WorkflowState.builder().enabled(true).build(); var workflowWithState = WorkflowWithState.create(workflow, workflowState); var workflowInstance = WorkflowInstance.create(workflow.id(), "2016-03-14"); var runState = RunState.create(workflowInstance, PREPARE, NOW, COUNTER); when(storage.workflowWithState(workflow.id())).thenReturn(Optional.of(workflowWithState)); toTest.transitionInto(runState, eventRouter); verify(eventRouter).receive(eventCaptor.capture(), eq(COUNTER)); var event = eventCaptor.getValue(); event.accept(eventVisitor); verify(eventVisitor) .submit(workflowInstanceCaptor.capture(), executionDescriptionCaptor.capture(), executionIdCaptor.capture()); assertThat(executionIdCaptor.getValue(), startsWith("styx-run-")); assertThat(executionDescriptionCaptor.getValue().dockerImage().orElseThrow(), is(DOCKER_IMAGE)); assertThat(executionDescriptionCaptor.getValue().dockerArgs(), hasSize(0)); assertThat(executionDescriptionCaptor.getValue().commitSha(), isPresentAndIs(COMMIT_SHA)); }
@Test public void shouldTransitionDockerWorkflowIntoSubmitting() throws Exception { var workflow = Workflow.create("id", workflowConfiguration("--date", "{}", "--bar")); var workflowState = WorkflowState.builder().enabled(true).build(); var workflowWithState = WorkflowWithState.create(workflow, workflowState); var workflowInstance = WorkflowInstance.create(workflow.id(), "2016-03-14"); var runState = RunState.create(workflowInstance, PREPARE, StateData.newBuilder().trigger(Trigger.natural()).build(), NOW, COUNTER); when(storage.workflowWithState(workflow.id())).thenReturn(Optional.of(workflowWithState)); toTest.transitionInto(runState, eventRouter); verify(eventRouter).receive(eventCaptor.capture(), eq(COUNTER)); var event = eventCaptor.getValue(); event.accept(eventVisitor); verify(eventVisitor) .submit(workflowInstanceCaptor.capture(), executionDescriptionCaptor.capture(), executionIdCaptor.capture()); assertThat(executionIdCaptor.getValue(), startsWith("styx-run-")); assertThat(executionDescriptionCaptor.getValue().dockerImage().orElseThrow(), is(DOCKER_IMAGE)); assertThat(executionDescriptionCaptor.getValue().commitSha(), isPresentAndIs(COMMIT_SHA)); assertThat( executionDescriptionCaptor.getValue().dockerArgs(), contains("--date", "2016-03-14", "--bar") ); assertThat(executionDescriptionCaptor.getValue().env(), is(Map.of("foo", "bar"))); assertThat(executionDescriptionCaptor.getValue().runningTimeout(), is(Optional.of(Duration.ZERO))); assertThat(executionDescriptionCaptor.getValue().retryCondition(), is(Optional.of("#tries<2"))); }
@Test public void shouldTransitionFlyteWorkflowIntoSubmitting() throws Exception { var workflow = Workflow.create("id", FLYTE_WORKFLOW_CONFIGURATION); var workflowState = WorkflowState.builder().enabled(true).build(); var workflowWithState = WorkflowWithState.create(workflow, workflowState); var workflowInstance = WorkflowInstance.create(workflow.id(), "2016-03-14"); var runState = RunState.create(workflowInstance, PREPARE, StateData.newBuilder().trigger(Trigger.natural()).build(), NOW, COUNTER); when(storage.workflowWithState(workflow.id())).thenReturn(Optional.of(workflowWithState)); toTest.transitionInto(runState, eventRouter); verify(eventRouter).receive(eventCaptor.capture(), eq(COUNTER)); var event = eventCaptor.getValue(); event.accept(eventVisitor); verify(eventVisitor) .submit(workflowInstanceCaptor.capture(), executionDescriptionCaptor.capture(), executionIdCaptor.capture()); assertThat(executionIdCaptor.getValue(), startsWith("styx-run-")); assertThat(executionDescriptionCaptor.getValue().flyteExecConf(), is(FLYTE_WORKFLOW_CONFIGURATION.flyteExecConf())); assertThat(executionDescriptionCaptor.getValue().commitSha(), isPresentAndIs(FLYTE_WORKFLOW_CONFIGURATION.commitSha().get())); assertThat(executionDescriptionCaptor.getValue().env(), is(FLYTE_WORKFLOW_CONFIGURATION.env())); assertThat(executionDescriptionCaptor.getValue().runningTimeout(), is(FLYTE_WORKFLOW_CONFIGURATION.runningTimeout())); assertThat(executionDescriptionCaptor.getValue().retryCondition(), is(FLYTE_WORKFLOW_CONFIGURATION.retryCondition())); }
@Test public void shouldTransitionIntoFailedIfStorageError() throws Exception { var workflow = Workflow.create("id", workflowConfiguration("--date", "{}", "--bar")); var workflowInstance = WorkflowInstance.create(workflow.id(), "2016-03-14"); var exception = new IOException("TEST"); when(storage.workflowWithState(workflow.id())).thenThrow(exception); RunState runState = RunState.create(workflowInstance, PREPARE, NOW, COUNTER); toTest.transitionInto(runState, eventRouter); verify(eventRouter).receive(Event.runError(workflowInstance, exception.getMessage()), COUNTER); }
@Test public void shouldHaltIfMissingWorkflow() throws Exception { var workflowInstance = WorkflowInstance.create(WorkflowId.create("c", "e"), "2016-03-14T15"); var runState = RunState.create(workflowInstance, PREPARE, NOW, COUNTER); when(storage.workflowWithState(any())).thenReturn(Optional.empty()); toTest.transitionInto(runState, eventRouter); verify(eventRouter).receiveIgnoreClosed(Event.halt(workflowInstance), COUNTER); }
@Test public void shouldHaltIfMissingDockerImageAndFlyteExecDescription() throws Exception { var workflowConfiguration = WorkflowConfigurationBuilder.from(MINIMAL_WORKFLOW_CONFIGURATION) .dockerImage(Optional.empty()) .flyteExecConf(Optional.empty()) .build(); var workflow = Workflow.create("id", workflowConfiguration); var workflowState = WorkflowState.builder().enabled(true).build(); var workflowWithState = WorkflowWithState.create(workflow, workflowState); var workflowInstance = WorkflowInstance.create(workflow.id(), "2016-03-14T15"); var runState = RunState.create(workflowInstance, PREPARE, NOW, COUNTER); when(storage.workflowWithState(workflow.id())).thenReturn(Optional.of(workflowWithState)); toTest.transitionInto(runState, eventRouter); verify(eventRouter).receiveIgnoreClosed(Event.halt(workflowInstance), COUNTER); }
@Test public void shouldHaltIfDisabledAndNaturallyTriggered() throws Exception { var workflow = Workflow.create("id", workflowConfiguration("--date", "{}", "--bar")); var workflowState = WorkflowState.builder().enabled(false).build(); var workflowWithState = WorkflowWithState.create(workflow, workflowState); var workflowInstance = WorkflowInstance.create(workflow.id(), "2016-03-14T15"); var runState = RunState.create(workflowInstance, PREPARE, StateData.newBuilder().trigger(Trigger.natural()).build(), NOW, COUNTER); when(storage.workflowWithState(workflow.id())).thenReturn(Optional.of(workflowWithState)); toTest.transitionInto(runState, eventRouter); verify(eventRouter).receiveIgnoreClosed(Event.halt(workflowInstance), COUNTER); }
@Test public void shouldHaltIfInvalidConfiguration() throws Exception { when(workflowValidator.validateWorkflow(any())).thenReturn(List.of("foo", "bar")); var workflow = Workflow.create("id", FULL_WORKFLOW_CONFIGURATION); var workflowState = WorkflowState.builder().enabled(true).build(); var workflowWithState = WorkflowWithState.create(workflow, workflowState); var workflowInstance = WorkflowInstance.create(workflow.id(), "2016-03-14T15"); var runState = RunState.create(workflowInstance, PREPARE, NOW, COUNTER); when(storage.workflowWithState(workflow.id())).thenReturn(Optional.of(workflowWithState)); toTest.transitionInto(runState, eventRouter); verify(eventRouter).receiveIgnoreClosed(Event.halt(workflowInstance), COUNTER); }