KubernetesDockerRunner implements DockerRunner { @VisibleForTesting static Pod createPod(WorkflowInstance workflowInstance, RunSpec runSpec, KubernetesSecretSpec secretSpec, String styxEnvironment) { final String imageWithTag = runSpec.imageName().contains(":") ? runSpec.imageName() : runSpec.imageName() + ":latest"; final String executionId = runSpec.executionId(); final PodBuilder podBuilder = new PodBuilder() .withNewMetadata() .withName(executionId) .addToAnnotations(STYX_WORKFLOW_INSTANCE_ANNOTATION, workflowInstance.toKey()) .addToAnnotations(DOCKER_TERMINATION_LOGGING_ANNOTATION, String.valueOf(runSpec.terminationLogging())) .endMetadata(); final PodSpecBuilder specBuilder = new PodSpecBuilder() .withRestartPolicy("Never"); final ResourceRequirementsBuilder resourceRequirements = new ResourceRequirementsBuilder(); runSpec.memRequest().ifPresent(s -> resourceRequirements.addToRequests("memory", new Quantity(s))); runSpec.memLimit().ifPresent(s -> resourceRequirements.addToLimits("memory", new Quantity(s))); final ContainerBuilder mainContainerBuilder = new ContainerBuilder() .withName(MAIN_CONTAINER_NAME) .withImage(imageWithTag) .withArgs(runSpec.args()) .withEnv(buildEnv(workflowInstance, runSpec, styxEnvironment)) .withResources(resourceRequirements.build()); secretSpec.serviceAccountSecret().ifPresent(serviceAccountSecret -> { final SecretVolumeSource saVolumeSource = new SecretVolumeSourceBuilder() .withSecretName(serviceAccountSecret) .build(); final Volume saVolume = new VolumeBuilder() .withName(STYX_WORKFLOW_SA_SECRET_NAME) .withSecret(saVolumeSource) .build(); specBuilder.addToVolumes(saVolume); final VolumeMount saMount = new VolumeMountBuilder() .withMountPath(STYX_WORKFLOW_SA_SECRET_MOUNT_PATH) .withName(saVolume.getName()) .withReadOnly(true) .build(); mainContainerBuilder.addToVolumeMounts(saMount); mainContainerBuilder.addToEnv(envVar(STYX_WORKFLOW_SA_ENV_VARIABLE, saMount.getMountPath() + STYX_WORKFLOW_SA_JSON_KEY)); }); secretSpec.customSecret().ifPresent(secret -> { final SecretVolumeSource secretVolumeSource = new SecretVolumeSourceBuilder() .withSecretName(secret.name()) .build(); final Volume secretVolume = new VolumeBuilder() .withName(secret.name()) .withSecret(secretVolumeSource) .build(); specBuilder.addToVolumes(secretVolume); final VolumeMount secretMount = new VolumeMountBuilder() .withMountPath(secret.mountPath()) .withName(secretVolume.getName()) .withReadOnly(true) .build(); mainContainerBuilder.addToVolumeMounts(secretMount); }); specBuilder.addToContainers(mainContainerBuilder.build()); specBuilder.addToContainers(keepaliveContainer()); podBuilder.withSpec(specBuilder.build()); return podBuilder.build(); } @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 shouldMountSecret() { final Pod pod = createPod(WORKFLOW_INSTANCE, RUN_SPEC_WITH_SECRET, SECRET_SPEC_WITH_CUSTOM_SECRET); assertThat(pod.getSpec().getVolumes().size(), is(1)); assertThat(pod.getSpec().getVolumes().get(0).getName(), is(RUN_SPEC_WITH_SECRET.secret().orElseThrow().name())); assertThat(pod.getSpec().getContainers().get(0).getVolumeMounts().get(0).getMountPath(), is(RUN_SPEC_WITH_SECRET.secret().orElseThrow().mountPath())); assertThat(pod.getSpec().getContainers().get(0).getVolumeMounts().get(0).getName(), is(RUN_SPEC_WITH_SECRET.secret().orElseThrow().name())); }
@Test public void shouldMountServiceAccount() { final Pod pod = createPod(WORKFLOW_INSTANCE, RUN_SPEC_WITH_SA, SECRET_SPEC_WITH_SA); assertThat(pod.getSpec().getVolumes().size(), is(1)); assertThat(pod.getSpec().getVolumes().get(0).getName(), is(KubernetesDockerRunner.STYX_WORKFLOW_SA_SECRET_NAME)); assertThat(pod.getSpec().getContainers().size(), is(2)); final Container mainContainer = pod.getSpec().getContainers().get(0); assertThat(mainContainer.getEnv().stream() .anyMatch(e -> e.getName().equals(KubernetesDockerRunner.STYX_WORKFLOW_SA_ENV_VARIABLE)), is(true)); }
@Test public void shouldConfigureResourceRequirements() { final String memRequest = "17Mi"; final String memLimit = "4711Mi"; final Pod pod = createPod(WORKFLOW_INSTANCE, RunSpec.builder() .executionId("eid1") .imageName("busybox") .memRequest(memRequest) .memLimit(memLimit) .build(), EMPTY_SECRET_SPEC); final ResourceRequirements resourceReqs = pod.getSpec().getContainers().get(0).getResources(); assertThat(resourceReqs.getRequests().get("memory"), is(new Quantity(memRequest))); assertThat(resourceReqs.getLimits().get("memory"), is(new Quantity(memLimit))); }