KubernetesDockerRunner implements DockerRunner { @Override public String start(RunState runState, RunSpec runSpec) throws IOException { var workflowInstance = runState.workflowInstance(); var existingPod = client.getPod(runSpec.executionId()); if (existingPod.isPresent()) { LOG.info("Pod already exists, not creating: {}: {}", workflowInstance, existingPod); return id; } final KubernetesSecretSpec secretSpec = ensureSecrets(workflowInstance, runSpec); try { var pod = createPod(workflowInstance, runSpec, secretSpec, styxEnvironment); LOG.info("Creating pod: {}: {}", workflowInstance, pod); var createdPod = client.createPod(pod); stats.recordSubmission(runSpec.executionId()); LOG.info("Created pod: {}: {}", workflowInstance, createdPod); } catch (KubernetesClientException kce) { if (kce.getCode() == 409 && kce.getStatus().getReason().equals("AlreadyExists")) { LOG.info("Pod already existed when creating: {}: {}", workflowInstance, runSpec.executionId()); } else { throw new IOException("Failed to create Kubernetes pod", kce); } } return id; } @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 shouldCheckIfPodExistsBeforeCreating() throws IOException { when(k8sClient.getPod(any())).thenReturn(Optional.of(createdPod)); kdr.start(RUN_STATE, RUN_SPEC); verify(k8sClient).getPod(RUN_SPEC.executionId()); verifyNoMoreInteractions(k8sClient); verifyZeroInteractions(stateManager); }
@Test(expected = InvalidExecutionException.class) public void shouldThrowIfSecretNotExist() throws IOException { kdr.start(RUN_STATE, RUN_SPEC_WITH_SECRET); }
@Test(expected = InvalidExecutionException.class) public void shouldThrowIfMountToReservedPath() throws IOException { kdr.start(RUN_STATE, RUN_SPEC_WITH_SECRET_AND_SA); }
@Test public void shouldNotRunIfUsingNonWhitelistedSecret() throws IOException { exception.expectMessage("Referenced secret 'secret-1' is not whitelisted"); kdr.start(RUN_STATE, RUN_SPEC_WITH_NON_WHITELISTED_SECRET); }
@Test public void shouldNotRunIfServiceAccountSecretEnsureFails() throws IOException { final InvalidExecutionException error = new InvalidExecutionException("SA not found"); when(serviceAccountSecretManager.ensureServiceAccountKeySecret( WORKFLOW_INSTANCE.workflowId().toString(), SERVICE_ACCOUNT)).thenThrow(error); exception.expect(is(error)); kdr.start(RUN_STATE, RUN_SPEC_WITH_SA); }
@Test public void shouldGenerateStartedAndRecordSubmitToRunningTimeWhenContainerIsReady() throws Exception { when(time.nanoTime()).thenReturn(TimeUnit.SECONDS.toNanos(17)); kdr.start(RUN_STATE, RunSpec.simple(POD_NAME, "busybox")); verify(stats).recordSubmission(POD_NAME); when(time.nanoTime()).thenReturn(TimeUnit.SECONDS.toNanos(18)); when(time.nanoTime()).thenReturn(TimeUnit.SECONDS.toNanos(19)); setRunning(createdPod, false); receiveAndProcessEvent(Watcher.Action.MODIFIED, createdPod); verify(stateManager, never()).receive(Event.started(WORKFLOW_INSTANCE), -1); when(time.nanoTime()).thenReturn(TimeUnit.SECONDS.toNanos(4711)); setRunning(createdPod, true); receiveAndProcessEvent(Watcher.Action.MODIFIED, createdPod); verify(stateManager).receive(Event.started(WORKFLOW_INSTANCE), -1); verify(stats).recordRunning(POD_NAME); }
@Test public void shouldDiscardChangesForOldExecutions() throws Exception { kdr.start(RUN_STATE, RUN_SPEC); createdPod.getMetadata().setName(POD_NAME + "-other"); setTerminated(createdPod, "Succeeded", 20, null); receiveAndProcessEvent(Watcher.Action.MODIFIED, createdPod); verify(stateManager, never()).receive(any(), anyLong()); }