KubernetesDockerRunner implements DockerRunner { @Override public void poll(RunState runState) { var executionId = runState.data().executionId().orElseThrow(IllegalArgumentException::new); var podOpt = client.getPod(executionId); if (podOpt.isEmpty()) { stateManager.receiveIgnoreClosed( Event.runError(runState.workflowInstance(), "No pod associated with this instance"), runState.counter()); return; } var pod = podOpt.orElseThrow(); logEvent(Watcher.Action.MODIFIED, pod, pod.getMetadata().getResourceVersion(), true); emitPodEvents(pod, runState); } @VisibleForTesting  KubernetesDockerRunner(String id, Fabric8KubernetesClient client, StateManager stateManager, Stats stats,
                         KubernetesGCPServiceAccountSecretManager serviceAccountSecretManager,
                         Debug debug, String styxEnvironment,
                         Set<String> secretWhitelist,
                         int cleanupPodsIntervalSeconds,
                         int podDeletionDelaySeconds,
                         Time time, ScheduledExecutorService scheduledExecutor);   KubernetesDockerRunner(String id, Fabric8KubernetesClient client, StateManager stateManager, Stats stats,
                         KubernetesGCPServiceAccountSecretManager serviceAccountSecretManager,
                         Debug debug, String styxEnvironment, Set<String> secretWhitelist); @Override String start(RunState runState, RunSpec runSpec); @Override void poll(RunState runState); @Override void cleanup(); @Override void close();  }
@Test public void shouldPollPodStatusAndEmitEvents() throws Exception { final Pod terminatedPod = new PodBuilder(createdPod) .withStatus(terminated("Succeeded", 20, null)) .build(); when(k8sClient.getPod(POD_NAME)).thenReturn(Optional.of(terminatedPod)); var stateData = StateData.newBuilder().executionId(POD_NAME).build(); var runState = RunState.create(WORKFLOW_INSTANCE, State.SUBMITTED, stateData); kdr.poll(runState); verify(stateManager, atLeastOnce()).receive( Event.started(WORKFLOW_INSTANCE), -1); verify(stateManager, atLeastOnce()).receive( Event.terminate(WORKFLOW_INSTANCE, Optional.of(20)), 0); }
@Test public void shouldTolerateTransitionConflictWhenEmittingEvents() throws Exception { final Pod terminatedPod = new PodBuilder(createdPod) .withStatus(terminated("Succeeded", 20, null)) .build(); when(k8sClient.getPod(POD_NAME)).thenReturn(Optional.of(terminatedPod)); doThrow(new StateTransitionConflictException("foo!")) .when(stateManager).receive(any(), anyLong()); verifyZeroInteractions(stateManager); var stateData = StateData.newBuilder().executionId(POD_NAME).build(); var runState = RunState.create(WORKFLOW_INSTANCE, State.SUBMITTED, stateData); kdr.poll(runState); verify(stateManager).receive(any(), anyLong()); }
@Test public void shouldTolerateCounterCapacityExceptionWhenEmittingEvents() throws Exception { var terminatedPod = new PodBuilder(createdPod) .withStatus(terminated("Succeeded", 20, null)) .build(); when(k8sClient.getPod(POD_NAME)).thenReturn(Optional.of(terminatedPod)); doThrow(new CounterCapacityException("foo!")) .when(stateManager).receive(any(), anyLong()); verifyZeroInteractions(stateManager); var stateData = StateData.newBuilder().executionId(POD_NAME).build(); var runState = RunState.create(WORKFLOW_INSTANCE, State.SUBMITTED, stateData); kdr.poll(runState); verify(stateManager).receive(any(), anyLong()); }