Scheduler { void tick() { try (Scope ignored = tracer.spanBuilder("Styx.Scheduler.tick") .setRecordEvents(true) .setSampler(Samplers.alwaysSample()) .startScopedSpan()) { tick0(); } } Scheduler(Time time, StateManager stateManager, Storage storage,
            WorkflowResourceDecorator resourceDecorator, Stats stats, RateLimiter dequeueRateLimiter,
            ShardedCounter shardedCounter, Executor executor);   Scheduler(Time time, StateManager stateManager, Storage storage,
            WorkflowResourceDecorator resourceDecorator, Stats stats, RateLimiter dequeueRateLimiter,
            ShardedCounter shardedCounter, Executor executor, Logger log);   }
@Test public void shouldNotScheduleExecutionWhenFailedToReadConfig() throws IOException { when(storage.config()).thenThrow(new IOException()); scheduler.tick(); verify(storage, never()).listActiveInstances(); }
@Test public void shouldBeRateLimiting() { when(rateLimiter.acquire()).thenReturn(1.0); setResourceLimit("r1", 2); initWorkflow(workflowUsingResources(WORKFLOW_ID1, "r1")); populateActiveStates(RunState.create(INSTANCE_1, State.QUEUED, time.get())); scheduler.tick(); verify(stats).recordResourceUsed("r1", 0L); verify(stateManager).receiveIgnoreClosed( eq(Event.dequeue(INSTANCE_1, ImmutableSet.of("r1"))), anyLong()); verify(rateLimiter).acquire(); }
@Test public void shouldExecuteRetryIfDelayHasPassed() { initWorkflow(workflowUsingResources(WORKFLOW_ID1)); StateData stateData = StateData.newBuilder().retryDelayMillis(15_000L).tries(10).build(); populateActiveStates(RunState.create(INSTANCE_1, State.QUEUED, stateData, time.get())); now = now.plus(15, ChronoUnit.SECONDS); scheduler.tick(); verify(stateManager).receiveIgnoreClosed(eq(Event.dequeue(INSTANCE_1, ImmutableSet.of())), anyLong()); }
@Test public void shouldReadRunStateCounterForEachTick() { initWorkflow(workflowUsingResources(WORKFLOW_ID1)); populateActiveStates(RunState.create( INSTANCE_1, State.QUEUED, StateData.zero(), time.get(), 15L)); scheduler.tick(); verify(stateManager).receiveIgnoreClosed(any(Event.class), eq(15L)); populateActiveStates(RunState.create( INSTANCE_1, State.QUEUED, StateData.zero(), time.get(), 23L)); scheduler.tick(); verify(stateManager).receiveIgnoreClosed(any(Event.class), eq(23L)); }
@Test public void shouldExecuteRetryIfDelayIsReset() { initWorkflow(workflowUsingResources(WORKFLOW_ID1)); StateData stateData = StateData.newBuilder().retryDelayMillis(15_000L).tries(10).build(); populateActiveStates(RunState.create(INSTANCE_1, State.QUEUED, stateData, time.get())); now = now.plus(10, ChronoUnit.SECONDS); scheduler.tick(); verify(stateManager, never()).receiveIgnoreClosed(any()); stateData = StateData.newBuilder().retryDelayMillis(0L).tries(10).build(); populateActiveStates(RunState.create(INSTANCE_1, State.QUEUED, stateData, time.get())); scheduler.tick(); verify(stateManager).receiveIgnoreClosed(eq(Event.dequeue(INSTANCE_1, ImmutableSet.of())), anyLong()); }
@Test public void shouldExecuteNewTriggers() { initWorkflow(workflowUsingResources(WORKFLOW_ID1)); StateData stateData = StateData.newBuilder().tries(0).build(); populateActiveStates(RunState.create(INSTANCE_1, State.QUEUED, stateData, time.get())); List<WorkflowInstance> workflowInstances = new ArrayList<>(); for (int i = 1; i <= 10; i++) { WorkflowId workflowId = WorkflowId.create("styx2", "example" + i); initWorkflow(workflowUsingResources(workflowId)); WorkflowInstance instance = WorkflowInstance.create(workflowId, "2016-12-02T01"); populateActiveStates(RunState.create(instance, State.QUEUED, stateData, time.get().minus(i, ChronoUnit.SECONDS))); workflowInstances.add(instance); } scheduler.tick(); Lists.reverse(workflowInstances) .forEach(x -> verify(stateManager) .receiveIgnoreClosed(eq(Event.dequeue(x, ImmutableSet.of())), anyLong())); verify(stateManager).receiveIgnoreClosed(eq( Event.dequeue(INSTANCE_1, ImmutableSet.of())), anyLong()); }
@Test public void shouldDequeueEvenWhenMissingWorkflows() { workflows.clear(); StateData stateData = StateData.newBuilder().tries(0).build(); populateActiveStates(RunState.create(INSTANCE_1, State.QUEUED, stateData, time.get())); List<WorkflowInstance> workflowInstances = new ArrayList<>(); for (int i = 1; i <= 10; i++) { WorkflowId workflowId = WorkflowId.create("styx2", "example" + i); WorkflowInstance instance = WorkflowInstance.create(workflowId, "2016-12-02T01"); populateActiveStates(RunState.create(instance, State.QUEUED, stateData, time.get().minus(i, ChronoUnit.SECONDS))); workflowInstances.add(instance); } scheduler.tick(); Lists.reverse(workflowInstances) .forEach(x -> verify(stateManager, timeout(30_000)) .receiveIgnoreClosed(eq(Event.dequeue(x, ImmutableSet.of())), anyLong())); verify(stateManager, timeout(30_000)).receiveIgnoreClosed(eq( Event.dequeue(INSTANCE_1, ImmutableSet.of())), anyLong()); }
@Test public void shouldFailWhenUnknownResourceReference() { initWorkflow(workflowUsingResources(WORKFLOW_ID1, "unknown")); populateActiveStates(RunState.create(INSTANCE_1, State.QUEUED, time.get())); scheduler.tick(); verify(stateManager).receiveIgnoreClosed(eq(Event.runError(INSTANCE_1, "Referenced resources not found: [unknown]")), anyLong()); }
@Test public void shouldFailWhenUnknownAndDepletedResources() { setResourceLimit("r1", 0); initWorkflow(workflowUsingResources(WORKFLOW_ID1, "r1", "r2", "r3")); populateActiveStates(RunState.create(INSTANCE_1, State.QUEUED, time.get())); scheduler.tick(); verify(stateManager).receiveIgnoreClosed( eq(Event.runError(INSTANCE_1, "Referenced resources not found: [r2, r3]")), anyLong()); }
@Test public void shouldIssueInfoOnceIfRepeated() throws Exception { setResourceLimit("r1", 0); initWorkflow(workflowUsingResources(WORKFLOW_ID1, "r1")); when(shardedCounter.counterHasSpareCapacity("r1")).thenReturn(false); InOrder inOrder = Mockito.inOrder(stateManager); StateData stateDataWithoutInfo = StateData.newBuilder() .build(); RunState rsWithoutInfo = RunState.create(INSTANCE_1, State.QUEUED, stateDataWithoutInfo, time.get(), 17); activeStates.put(INSTANCE_1, rsWithoutInfo); scheduler.tick(); inOrder.verify(stateManager).listActiveInstances(); inOrder.verify(stateManager).getActiveState(INSTANCE_1); inOrder.verify(stateManager).receiveIgnoreClosed( Event.info(INSTANCE_1, Message.info("Resource limit reached for: [r1]")), rsWithoutInfo.counter()); StateData stateDataWithInfo = StateData.newBuilder() .addMessage(Message.info("Resource limit reached for: [r1]")) .build(); activeStates.put(INSTANCE_1, RunState.create(INSTANCE_1, State.QUEUED, stateDataWithInfo, time.get())); scheduler.tick(); inOrder.verify(stateManager).listActiveInstances(); inOrder.verify(stateManager).getActiveState(INSTANCE_1); inOrder.verifyNoMoreInteractions(); }
@Test public void shouldRecordAggregateResourceUsageAndDemand() { setResourceLimit("r1", 2); setResourceLimit("r2", 3); initWorkflow(workflowUsingResources(WORKFLOW_ID1, "r1")); initWorkflow(workflowUsingResources(WORKFLOW_ID2, "r2")); populateActiveStates(runStateWithResources(instance(WORKFLOW_ID1, "i0"), State.NEW, "r1")); populateActiveStates(runStateWithResources(instance(WORKFLOW_ID1, "i1"), State.QUEUED, "r1")); populateActiveStates(runStateWithResources(instance(WORKFLOW_ID1, "i5"), State.QUEUED, "r1")); populateActiveStates(runStateWithResources(instance(WORKFLOW_ID2, "i6"), State.QUEUED, "r2")); populateActiveStates(runStateWithResources(instance(WORKFLOW_ID1, "i4"), State.TERMINATED, "r1")); populateActiveStates(runStateWithResources(instance(WORKFLOW_ID1, "i2"), State.SUBMITTING, "r1")); populateActiveStates(runStateWithResources(instance(WORKFLOW_ID1, "i3"), State.PREPARE, "r1")); scheduler.tick(); verify(stats).recordResourceDemanded("r1", 2L); verify(stats).recordResourceDemanded("r2", 1L); verify(stats).recordResourceUsed("r1", 2L); }
@Test public void shouldNotExceedResourceLimitsIfAlreadyAtLimit() throws Exception { setResourceLimit("r1", 3); initWorkflow(workflowUsingResources(WORKFLOW_ID1, "r1")); when(shardedCounter.counterHasSpareCapacity("r1")).thenReturn(false); final WorkflowInstance i0 = instance(WORKFLOW_ID1, "i0"); final WorkflowInstance i1 = instance(WORKFLOW_ID1, "i1"); final RunState rs0 = RunState.create(i0, State.NEW, time.get(), 17); final RunState rs1 = RunState.create(i1, State.QUEUED, time.get(), 4711); populateActiveStates(rs0); populateActiveStates(rs1); populateActiveStates(runStateWithResources(instance(WORKFLOW_ID1, "i2"), State.PREPARE, "r1")); populateActiveStates(runStateWithResources(instance(WORKFLOW_ID1, "i3"), State.PREPARE, "r1")); populateActiveStates(runStateWithResources(instance(WORKFLOW_ID1, "i4"), State.PREPARE, "r1")); scheduler.tick(); verify(shardedCounter, times(1)).counterHasSpareCapacity("r1"); assertThat(eventCaptor.getAllValues().stream() .anyMatch(e -> EventUtil.name(e).equals("dequeue")), is(false)); verify(stats).recordResourceDemanded("r1", 1L); verify(stats).recordResourceUsed("r1", 3L); scheduler.tick(); verify(stateManager, times(2)).listActiveInstances(); verify(shardedCounter, times(2)).counterHasSpareCapacity("r1"); assertThat(eventCaptor.getAllValues().stream() .anyMatch(e -> EventUtil.name(e).equals("dequeue")), is(false)); verify(stats, times(2)).recordResourceDemanded("r1", 1L); verify(stats, times(2)).recordResourceUsed("r1", 3L); }
@Test public void shouldCacheResourceUsageExceededLookup() throws Exception { setResourceLimit("r1", 2); initWorkflow(workflowUsingResources(WORKFLOW_ID1, "r1")); when(shardedCounter.counterHasSpareCapacity("r1")).thenReturn(false); final WorkflowInstance i0 = instance(WORKFLOW_ID1, "i0"); final WorkflowInstance i1 = instance(WORKFLOW_ID1, "i1"); final RunState rs0 = RunState.create(i0, State.QUEUED, time.get(), 17); final RunState rs1 = RunState.create(i1, State.QUEUED, time.get(), 4711); populateActiveStates(rs0); populateActiveStates(rs1); scheduler.tick(); verify(shardedCounter, times(1)).counterHasSpareCapacity("r1"); verifyNoMoreInteractions(shardedCounter); }
@Test public void shouldHandleResourceUsageExceededLookupFailure() throws Exception { setResourceLimit("r1", 2); initWorkflow(workflowUsingResources(WORKFLOW_ID1, "r1")); when(shardedCounter.counterHasSpareCapacity("r1")).thenThrow(new RuntimeException("error!")); final WorkflowInstance i0 = instance(WORKFLOW_ID1, "i0"); final RunState rs0 = RunState.create(i0, State.QUEUED, time.get(), 17); populateActiveStates(rs0); scheduler.tick(); verify(shardedCounter, times(1)).counterHasSpareCapacity("r1"); verify(stateManager).receiveIgnoreClosed(Event.dequeue(i0, ImmutableSet.of("r1")), 17); }
@Test public void shouldDecorateWorkflowInstanceResources() { Workflow workflow = workflowUsingResources(WORKFLOW_ID1, "foo", "bar"); when(resourceDecorator.decorateResources( any(RunState.class), any(WorkflowConfiguration.class), anySet())) .thenReturn(ImmutableSet.of("baz", "quux", "GLOBAL_STYX_CLUSTER")); when(config.globalConcurrency()).thenReturn(Optional.of(17L)); setResourceLimit("baz", 4); setResourceLimit("quux", 4); initWorkflow(workflow); populateActiveStates(RunState.create(INSTANCE_1, State.QUEUED, time.get())); scheduler.tick(); verify(resourceDecorator).decorateResources(any(RunState.class), eq(workflow.configuration()), eq(ImmutableSet.of("foo", "bar", "GLOBAL_STYX_CLUSTER"))); verify(stateManager).receiveIgnoreClosed(eq(Event.dequeue(INSTANCE_1, ImmutableSet.of("baz", "quux", "GLOBAL_STYX_CLUSTER"))), anyLong()); }
@Test public void shouldCountDecoratedResourcesOnNonQueuedStates() { Workflow workflow = workflowUsingResources(WORKFLOW_ID1, "foo", "bar"); when(resourceDecorator.decorateResources( any(RunState.class), any(WorkflowConfiguration.class), anySet())) .thenReturn(ImmutableSet.of("baz", "GLOBAL_STYX_CLUSTER")); when(config.globalConcurrency()).thenReturn(Optional.of(17L)); setResourceLimit("baz", 3); initWorkflow(workflow); WorkflowInstance i0 = instance(WORKFLOW_ID1, "i0"); WorkflowInstance i1 = instance(WORKFLOW_ID1, "i1"); WorkflowInstance i2 = instance(WORKFLOW_ID1, "i2"); WorkflowInstance i3 = instance(WORKFLOW_ID1, "i3"); WorkflowInstance i4 = instance(WORKFLOW_ID1, "i4"); populateActiveStates( RunState.create(i0, State.QUEUED, time.get()), RunState.create(i1, State.SUBMITTING, time.get()), RunState.create(i2, State.PREPARE, time.get()), RunState.create(i3, State.TERMINATED, time.get()), RunState.create(i4, State.QUEUED, time.get())); scheduler.tick(); verify(resourceDecorator).decorateResources(eq(RunState.create(i0, State.QUEUED, time.get())), eq(workflow.configuration()), eq(Set.of("foo", "bar", "GLOBAL_STYX_CLUSTER"))); verify(resourceDecorator).decorateResources(eq(RunState.create(i4, State.QUEUED, time.get())), eq(workflow.configuration()), eq(Set.of("foo", "bar", "GLOBAL_STYX_CLUSTER"))); verify(stateManager, times(2)).receiveIgnoreClosed(argThat( either(is(Event.dequeue(i0, ImmutableSet.of("baz", "GLOBAL_STYX_CLUSTER")))) .or(is(Event.dequeue(i4, ImmutableSet.of("baz", "GLOBAL_STYX_CLUSTER"))))), anyLong()); }
@Test public void shouldHandleDequeueFailures() { var workflow1 = workflowUsingResources(WORKFLOW_ID1); var workflow2 = workflowUsingResources(WORKFLOW_ID2); initWorkflow(workflow1); initWorkflow(workflow2); var counter1 = 17; var stateData1 = StateData.newBuilder().tries(0).build(); var runState1 = RunState.create(INSTANCE_1, State.QUEUED, stateData1, time.get(), counter1); var counter2 = 4711; var stateData2 = StateData.newBuilder().tries(0).build(); var runState2 = RunState.create(INSTANCE_2, State.QUEUED, stateData2, time.get(), counter2); populateActiveStates(runState1, runState2); doThrow(new RuntimeException("fail!")) .when(stateManager).receiveIgnoreClosed(Event.dequeue(INSTANCE_1, ImmutableSet.of()), counter1); scheduler.tick(); verify(stateManager).receiveIgnoreClosed(Event.dequeue(INSTANCE_1, ImmutableSet.of()), counter1); verify(stateManager).receiveIgnoreClosed(Event.dequeue(INSTANCE_2, ImmutableSet.of()), counter2); verify(stats).recordTickDuration(any(), anyLong()); }
@Test public void shouldHandleStateTransitionConflicts() { var workflow1 = workflowUsingResources(WORKFLOW_ID1); var workflow2 = workflowUsingResources(WORKFLOW_ID2); initWorkflow(workflow1); initWorkflow(workflow2); var counter1 = 17; var stateData1 = StateData.newBuilder().tries(0).build(); var runState1 = RunState.create(INSTANCE_1, State.QUEUED, stateData1, time.get(), counter1); var counter2 = 4711; var stateData2 = StateData.newBuilder().tries(0).build(); var runState2 = RunState.create(INSTANCE_2, State.QUEUED, stateData2, time.get(), counter2); populateActiveStates(runState1, runState2); var cause = new StateTransitionConflictException("conflict!"); doThrow(cause).when(stateManager).receiveIgnoreClosed(Event.dequeue(INSTANCE_1, ImmutableSet.of()), counter1); scheduler.tick(); verify(stateManager).receiveIgnoreClosed(Event.dequeue(INSTANCE_1, ImmutableSet.of()), counter1); verify(stateManager).receiveIgnoreClosed(Event.dequeue(INSTANCE_2, ImmutableSet.of()), counter2); verify(stats).recordTickDuration(any(), anyLong()); verify(log).debug("State transition conflict when scheduling instance: {}", INSTANCE_1, cause); }
@Test public void shouldHandleCounterCapacityExceptions() { var workflow1 = workflowUsingResources(WORKFLOW_ID1); var workflow2 = workflowUsingResources(WORKFLOW_ID2); initWorkflow(workflow1); initWorkflow(workflow2); var counter1 = 17; var stateData1 = StateData.newBuilder().tries(0).build(); var runState1 = RunState.create(INSTANCE_1, State.QUEUED, stateData1, time.get(), counter1); var counter2 = 4711; var stateData2 = StateData.newBuilder().tries(0).build(); var runState2 = RunState.create(INSTANCE_2, State.QUEUED, stateData2, time.get(), counter2); populateActiveStates(runState1, runState2); var cause = new CounterCapacityException("foo!"); doThrow(cause).when(stateManager).receiveIgnoreClosed(Event.dequeue(INSTANCE_1, ImmutableSet.of()), counter1); scheduler.tick(); verify(stateManager).receiveIgnoreClosed(Event.dequeue(INSTANCE_1, ImmutableSet.of()), counter1); verify(stateManager).receiveIgnoreClosed(Event.dequeue(INSTANCE_2, ImmutableSet.of()), counter2); verify(stats).recordTickDuration(any(), anyLong()); verify(log).debug("Counter capacity exhausted when scheduling instance: {}", INSTANCE_1, cause); }
@Test public void shouldSendBackToQueueOnDisabledGlobally() { when(config.globalEnabled()).thenReturn(false); initWorkflow(workflowUsingResources(WORKFLOW_ID1)); var stateData = StateData.newBuilder().tries(0).build(); populateActiveStates(RunState.create(INSTANCE_1, State.QUEUED, stateData, time.get())); scheduler.tick(); var event = eventCaptor.getValue(); assertThat(event.workflowInstance(), is(INSTANCE_1)); assertThat(EventUtil.name(event).equals("retryAfter"), is(true)); }