org.apache.commons.io.output.CloseShieldOutputStream#io.fabric8.kubernetes.api.model.Pod源码实例Demo

private KubernetesAgentInstances unregisteredAfterTimeout(PluginSettings settings, Agents knownAgents) throws Exception {
    Period period = settings.getAutoRegisterPeriod();
    KubernetesAgentInstances unregisteredInstances = new KubernetesAgentInstances();
    KubernetesClient client = factory.client(settings);

    for (String instanceName : instances.keySet()) {
        if (knownAgents.containsAgentWithId(instanceName)) {
            continue;
        }

        Pod pod = getPod(client, instanceName);
        if (pod == null) {
            LOG.debug(String.format("[server-ping] Pod with name %s is already deleted.", instanceName));
            continue;
        }

        Date createdAt = getSimpleDateFormat().parse(pod.getMetadata().getCreationTimestamp());
        DateTime dateTimeCreated = new DateTime(createdAt);

        if (clock.now().isAfter(dateTimeCreated.plus(period))) {
            unregisteredInstances.register(kubernetesInstanceFactory.fromKubernetesPod(pod));
        }
    }

    return unregisteredInstances;
}
 

private static Pod createK8sPod(String uid, String name) {
    ObjectMeta meta = new ObjectMeta();
    meta.setUid(uid);
    meta.setName(name);

    PodStatus status = new PodStatus();
    status.setPhase("Running");

    Pod pod = new Pod();
    pod.setApiVersion("v1");
    pod.setKind("pod");
    pod.setMetadata(meta);
    pod.setStatus(status);

    return pod;
}
 

private List<PodData> getAllInjectablePods(
    ObjectMeta toCreateMeta,
    List<Container> toCreateContainers,
    Map<String, Map<String, Pod>> injectables) {
  return toCreateContainers
      .stream()
      .map(c -> Names.machineName(toCreateMeta, c))
      .map(injectables::get)
      // we're only interested in pods for which we require injection
      .filter(Objects::nonNull)
      // now reduce to a map keyed by injected pod name so that if 2 pods require injection
      // of the same thing, we don't inject twice
      .flatMap(m -> m.entrySet().stream())
      // collect to map, ignoring duplicate entries
      .collect(toMap(Entry::getKey, Entry::getValue, (v1, v2) -> v1))
      // ok, we only have 1 of each injectable pods keyed by their names, so let's just get them
      // all and return as list
      .values()
      .stream()
      .map(PodData::new)
      .collect(Collectors.toList());
}
 

protected KubernetesEnvironment(
    InternalRecipe internalRecipe,
    Map<String, InternalMachineConfig> machines,
    List<Warning> warnings,
    Map<String, Pod> pods,
    Map<String, Deployment> deployments,
    Map<String, Service> services,
    Map<String, Ingress> ingresses,
    Map<String, PersistentVolumeClaim> persistentVolumeClaims,
    Map<String, Secret> secrets,
    Map<String, ConfigMap> configMaps) {
  super(internalRecipe, machines, warnings);
  setType(TYPE);
  this.pods = pods;
  this.deployments = deployments;
  this.services = services;
  this.ingresses = ingresses;
  this.persistentVolumeClaims = persistentVolumeClaims;
  this.secrets = secrets;
  this.configMaps = configMaps;
  this.podData = new HashMap<>();
  this.injectablePods = new HashMap<>();
  pods.forEach((name, pod) -> podData.put(name, new PodData(pod)));
  deployments.forEach((name, deployment) -> podData.put(name, new PodData(deployment)));
}
 

@Test
public void testLocal() throws Exception {
    Pod pod = objectMapper.readValue(new File(localJson), Pod.class);
    Namespace namespace = createNamespace();
    KubernetesLookup lookup = new KubernetesLookup(pod, namespace, masterUrl);
    try {
        assertEquals("Incorrect container name", "sampleapp", lookup.lookup("containerName"));
        assertEquals("Incorrect container id",
                "docker://818b0098946c67e6ac56cb7c0934b7c2a9f50feb7244b422b2a7f566f7e5d0df",
                lookup.lookup("containerId"));
        assertEquals("Incorrect host name", "docker-desktop", lookup.lookup("host"));
        assertEquals("Incorrect pod name", "sampleapp-584f99476d-mnrp4", lookup.lookup("podName"));
    } finally {
        lookup.clearInfo();;
    }
}
 

@Test
public void testResourceReplaceFromLoad() throws Exception {
  server.expect().get().withPath("/api/v1/namespaces/test/pods/nginx").andReturn(200, new PodBuilder().withNewMetadata().withResourceVersion("12345").and().build()).always();

  server.expect().put().withPath("/api/v1/namespaces/test/pods/nginx").andReturn(200, new PodBuilder()
    .withNewMetadata().withResourceVersion("12345").and().build()).once();

  KubernetesClient client = server.getClient();
  List<HasMetadata> result = client.load(getClass().getResourceAsStream("/test-pod-create-from-load.yml")).createOrReplace();
  assertNotNull(result);
  assertEquals(1, result.size());
  Pod pod = (Pod) result.get(0);
  assertEquals("12345", pod.getMetadata().getResourceVersion());

  RecordedRequest request = server.getLastRequest();
  assertEquals("/api/v1/namespaces/test/pods/nginx", request.getPath());
  Pod requestPod = new ObjectMapper().readerFor(Pod.class).readValue(request.getBody().inputStream());
  assertEquals("nginx", requestPod.getMetadata().getName());
}
 

@Test
public void shouldDeleteUnwantedStyxPods() {
  final Pod createdPod1 = createPod(WORKFLOW_INSTANCE, RUN_SPEC, SECRET_SPEC);
  final Pod createdPod2 = createPod(WORKFLOW_INSTANCE_2, RUN_SPEC_2, SECRET_SPEC);

  podList.setItems(Arrays.asList(createdPod1, createdPod2));
  when(k8sClient.getPod(RUN_SPEC.executionId())).thenReturn(Optional.of(createdPod1));
  when(k8sClient.getPod(RUN_SPEC_2.executionId())).thenReturn(Optional.of(createdPod2));

  createdPod1.setStatus(podStatus1);
  when(podStatus1.getContainerStatuses()).thenReturn(List.of(containerStatus1));
  when(containerStatus1.getName()).thenReturn(RUN_SPEC.executionId());

  createdPod2.setStatus(podStatus2);
  when(podStatus2.getContainerStatuses()).thenReturn(List.of(containerStatus2));
  when(containerStatus2.getName()).thenReturn(RUN_SPEC_2.executionId());

  kdr.tryCleanupPods();

  verify(k8sClient).deletePod(createdPod1.getMetadata().getName());
  verify(k8sClient).deletePod(createdPod2.getMetadata().getName());
}
 

private List<PodResource<Pod, DoneablePod>> doGetLog(boolean isPretty) {
  List<PodResource<Pod, DoneablePod>> pods = new ArrayList<>();
  StatefulSet statefulSet = fromServer().get();
  String rcUid = statefulSet.getMetadata().getUid();

  PodOperationsImpl podOperations = new PodOperationsImpl(new PodOperationContext(context.getClient(),
    context.getConfig(), context.getPlural(), context.getNamespace(), context.getName(), null,
    "v1", context.getCascading(), context.getItem(), context.getLabels(), context.getLabelsNot(),
    context.getLabelsIn(), context.getLabelsNotIn(), context.getFields(), context.getFieldsNot(), context.getResourceVersion(),
    context.getReloadingFromServer(), context.getGracePeriodSeconds(), context.getPropagationPolicy(),
    context.getWatchRetryInitialBackoffMillis(), context.getWatchRetryBackoffMultiplier(), null, null, null, null, null,
    null, null, null, null, false, false, false, null, null,
    null, isPretty, null, null, null, null, null));
  PodList jobPodList = podOperations.withLabels(statefulSet.getSpec().getTemplate().getMetadata().getLabels()).list();

  for (Pod pod : jobPodList.getItems()) {
    OwnerReference ownerReference = KubernetesResourceUtil.getControllerUid(pod);
    if (ownerReference != null && ownerReference.getUid().equals(rcUid)) {
      pods.add(podOperations.withName(pod.getMetadata().getName()));
    }
  }
  return pods;
}
 

@Test
public void testCluster() throws Exception {
    Pod pod = objectMapper.readValue(new File(clusterJson), Pod.class);
    Namespace namespace = createNamespace();
    KubernetesLookup lookup = new KubernetesLookup(pod, namespace, masterUrl);
    try {
        assertEquals("Incorrect container name", "platform-forms-service", lookup.lookup("containerName"));
        assertEquals("Incorrect container id",
                "docker://2b7c2a93dfb48334aa549e29fdd38039ddd256eec43ba64c145fa4b75a1542f0",
                lookup.lookup("containerId"));
        assertEquals("Incorrect host name", "k8s-tmpcrm-worker-s03-04", lookup.lookup("host"));
        assertEquals("Incorrect pod name", "platform-forms-service-primary-5ddfc4f9b8-kfpzv", lookup.lookup("podName"));
    } finally {
        lookup.clearInfo();
    }
}
 

@Exported
public List<Container> getContainers() throws KubernetesAuthException, IOException {
    if(!Jenkins.get().hasPermission(Jenkins.ADMINISTER)) {
        LOGGER.log(Level.FINE, " Computer {0} getContainers, lack of admin permission, returning empty list", this);
        return Collections.emptyList();
    }

    KubernetesSlave slave = getNode();
    if(slave == null) {
        return Collections.emptyList();
    }

    KubernetesCloud cloud = slave.getKubernetesCloud();
    KubernetesClient client = cloud.connect();

    String namespace = StringUtils.defaultIfBlank(slave.getNamespace(), client.getNamespace());
    Pod pod = client.pods().inNamespace(namespace).withName(getName()).get();

    return pod.getSpec().getContainers();
}
 

public static KubernetesPod buildTaskManagerKubernetesPod(KubernetesTaskManagerParameters kubernetesTaskManagerParameters) {
	FlinkPod flinkPod = new FlinkPod.Builder().build();

	final KubernetesStepDecorator[] stepDecorators = new KubernetesStepDecorator[] {
		new InitTaskManagerDecorator(kubernetesTaskManagerParameters),
		new JavaCmdTaskManagerDecorator(kubernetesTaskManagerParameters),
		new HadoopConfMountDecorator(kubernetesTaskManagerParameters),
		new FlinkConfMountDecorator(kubernetesTaskManagerParameters)};

	for (KubernetesStepDecorator stepDecorator: stepDecorators) {
		flinkPod = stepDecorator.decorateFlinkPod(flinkPod);
	}

	final Pod resolvedPod = new PodBuilder(flinkPod.getPod())
		.editOrNewSpec()
			.addToContainers(flinkPod.getMainContainer())
			.endSpec()
		.build();

	return new KubernetesPod(resolvedPod);
}
 

@Test
public void listFromServer() {
  // Wait for resources to get ready
  ReadyEntity<Pod> podReady = new ReadyEntity<>(Pod.class, client, pod1.getMetadata().getName(), currentNamespace);
  await().atMost(30, TimeUnit.SECONDS).until(podReady);

  List<HasMetadata> resources = client.resourceList(pod1).inNamespace(currentNamespace).fromServer().get();

  assertNotNull(resources);
  assertEquals(1, resources.size());
  assertNotNull(resources.get(0));

  HasMetadata fromServerPod = resources.get(0);

  assertEquals(pod1.getKind(), fromServerPod.getKind());
  assertEquals(currentNamespace, fromServerPod.getMetadata().getNamespace());
  assertEquals(pod1.getMetadata().getName(), fromServerPod.getMetadata().getName());
}
 

protected static String getPodCondition(Pod pod) {
    PodStatus podStatus = pod.getStatus();
    if (podStatus == null) {
        return "";
    }
    List<PodCondition> conditions = podStatus.getConditions();
    if (conditions == null || conditions.isEmpty()) {
        return "";
    }


    for (PodCondition condition : conditions) {
        String type = condition.getType();
        if (StringUtils.isNotBlank(type)) {
            if ("ready".equalsIgnoreCase(type)) {
                String statusText = condition.getStatus();
                if (StringUtils.isNotBlank(statusText)) {
                    if (Boolean.parseBoolean(statusText)) {
                        return type;
                    }
                }
            }
        }
    }
    return "";
}
 

@Test(groups = {"Live"})
public void testTomcatPodExtras() throws Exception {
    String yaml = Joiner.on("\n").join(
            locationYaml,
            "services:",
            "  - type: " + KubernetesPod.class.getName(),
            "    brooklyn.config:",
            "      docker.container.imageName: tomcat",
            "      docker.container.inboundPorts: [ \"8080\" ]",
            "      metadata:",
            "        extra: test");

    KubernetesPod entity = runTomcat(yaml, KubernetesPod.class);

    String namespace = entity.sensors().get(KubernetesPod.KUBERNETES_NAMESPACE);
    String podName = entity.sensors().get(KubernetesPod.KUBERNETES_POD);
    KubernetesClient client = getClient(entity);
    Pod pod = client.pods().inNamespace(namespace).withName(podName).get();
    Map<String, String> labels = pod.getMetadata().getLabels();
    assertTrue(labels.containsKey("extra"));
    assertEquals(labels.get("extra"), "test");
}
 

@Test
public void shouldNotApplyProxySettingsToJWTProxyContainer() throws Exception {

  Map<String, Pod> pods = new HashMap<>();
  pods.put(JWT_PROXY_POD_NAME, buildPod(JWT_PROXY_POD_NAME, buildContainers(2)));

  KubernetesEnvironment k8sEnv = KubernetesEnvironment.builder().setPods(pods).build();
  provisioner.provision(k8sEnv, runtimeId);

  assertTrue(
      k8sEnv
          .getPodsData()
          .values()
          .stream()
          .filter(pod -> pod.getMetadata().getName().equals(JWT_PROXY_POD_NAME))
          .flatMap(pod -> pod.getSpec().getContainers().stream())
          .noneMatch(
              container ->
                  container.getEnv().contains(new EnvVar(HTTP_PROXY, HTTP_PROXY_VALUE, null))
                      || container
                          .getEnv()
                          .contains(new EnvVar(HTTPS_PROXY, HTTPS_PROXY_VALUE, null))
                      || container
                          .getEnv()
                          .contains(new EnvVar(NO_PROXY, NO_PROXY_VALUE, null))));
}
 

@Test
void testReplaceWithDeleteDeltaIn() throws InterruptedException {
  Pod oldPod = new PodBuilder().withNewMetadata().withNamespace("default").withName("foo1").endMetadata().build();
  Pod newPod = new PodBuilder().withNewMetadata().withNamespace("default").withName("foo2").endMetadata().build();

  Cache<Pod> mockCache = mock(Cache.class);
  doReturn(oldPod).when(mockCache).getByKey(Cache.deletionHandlingMetaNamespaceKeyFunc(oldPod));
  DeltaFIFO<Pod> deltaFIFO =
    new DeltaFIFO<>(Cache::deletionHandlingMetaNamespaceKeyFunc, mockCache);

  deltaFIFO.delete(oldPod);
  deltaFIFO.replace(Collections.singletonList(newPod), "0");

  deltaFIFO.pop(
    (deltas) -> {
      assertEquals(DeltaFIFO.DeltaType.DELETION, deltas.getFirst().getKey());
      assertEquals(oldPod, deltas.getFirst().getValue());
    });

  deltaFIFO.pop(
    (deltas) -> {
      assertEquals(DeltaFIFO.DeltaType.SYNCHRONIZATION, deltas.getFirst().getKey());
      assertEquals(newPod, deltas.getFirst().getValue());
    });
}
 

@Test
public void shouldStartMachines() throws Exception {
  final Container container1 = mockContainer(CONTAINER_NAME_1, EXPOSED_PORT_1);
  final Container container2 = mockContainer(CONTAINER_NAME_2, EXPOSED_PORT_2, INTERNAL_PORT);
  final ImmutableMap<String, Pod> allPods =
      ImmutableMap.of(POD_NAME, mockPod(ImmutableList.of(container1, container2)));
  when(osEnv.getPodsCopy()).thenReturn(allPods);
  when(unrecoverablePodEventListenerFactory.isConfigured()).thenReturn(true);

  internalRuntime.startMachines();

  verify(deployments).deploy(any(Pod.class));
  verify(routes).create(any());
  verify(services).create(any());
  verify(secrets).create(any());
  verify(configMaps).create(any());

  verify(project.deployments(), times(2)).watchEvents(any());
  verify(eventService, times(2)).publish(any());
  verifyEventsOrder(newEvent(M1_NAME, STARTING), newEvent(M2_NAME, STARTING));
}
 

/**
 * Verifies container environment variables passed as a map.
 * @param podNamePrefix Name of pod where container is located
 * @param containerName The container where verifying is expected
 * @param config Expected environment variables with values
 */
protected void checkSpecificVariablesInContainer(String podNamePrefix, String containerName, Map<String, String> config) {
    LOGGER.info("Getting pods by prefix in name {}", podNamePrefix);
    List<Pod> pods = kubeClient().listPodsByPrefixInName(podNamePrefix);

    if (pods.size() != 0) {
        LOGGER.info("Testing EnvVars configuration for container {}", containerName);

        Map<String, Object> actual = pods.stream()
            .flatMap(p -> p.getSpec().getContainers().stream()) // get containers
            .filter(c -> c.getName().equals(containerName))
            .flatMap(c -> c.getEnv().stream().filter(envVar -> config.containsKey(envVar.getName())))
            .collect(Collectors.toMap(EnvVar::getName, EnvVar::getValue, (item, duplicatedItem) -> item));
        assertThat(actual, is(config));
    } else {
        fail("Pod with prefix " + podNamePrefix + " in name, not found");
    }
}
 

public PodSetController(KubernetesClient kubernetesClient, MixedOperation<PodSet, PodSetList, DoneablePodSet, Resource<PodSet, DoneablePodSet>>  podSetClient, SharedIndexInformer<Pod> podInformer, SharedIndexInformer<PodSet> podSetInformer, String namespace) {
    this.kubernetesClient = kubernetesClient;
    this.podSetClient = podSetClient;
    this.podSetLister = new Lister<>(podSetInformer.getIndexer(), namespace);
    this.podSetInformer = podSetInformer;
    this.podLister = new Lister<>(podInformer.getIndexer(), namespace);
    this.podInformer = podInformer;
    this.workqueue = new ArrayBlockingQueue<>(1024);
}
 

private void handlePodObject(Pod pod) {
    logger.log(Level.INFO, "handlePodObject(" + pod.getMetadata().getName() + ")");
    OwnerReference ownerReference = getControllerOf(pod);
    if (!ownerReference.getKind().equalsIgnoreCase("PodSet")) {
        return;
    }
    PodSet podSet = podSetLister.get(ownerReference.getName());
    if (podSet != null) {
        enqueuePodSet(podSet);
    }
}
 

@Test
public void testGetHostReturnsCorrectHostForGivenPod() {
    Pod pod = new PodBuilder()
            .withNewMetadata()
                .withName(ZookeeperCluster.zookeeperPodName("my-cluster", 3))
                .withNamespace("myproject")
                .addToLabels(Labels.STRIMZI_CLUSTER_LABEL, "my-cluster")
            .endMetadata()
        .build();

    assertThat(new ZookeeperLeaderFinder(vertx, null, this::backoff).host(pod),
            is("my-cluster-zookeeper-3.my-cluster-zookeeper-nodes.myproject.svc.cluster.local"));
}
 

@Override
public Lander newInstance(LanderConfiguration config) {
  List<String> args = argsFactory.newInstance(config);
  Pod pod = podFactory.newInstance(config, args);
  String podName = podNameFactory.newName(config);
  return new KubernetesLander(config, pod, podName, client);
}
 

private List<Pod> generateDrainPods(
        final Set<String> podNames,
        final String clusterName,
        final RabbitMQCustomResourceSpec rabbit,
        final RabbitMQNetworkPartitionCustomResource networkPartition
) {
    final Container container = rabbitMQContainers.buildContainer(
            namespace,
            clusterName,
            rabbit.getRabbitMQImage(),
            rabbit.getComputeResources(),
            0);

    return podNames.stream().map(podName ->
            new PodBuilder()
                    .withNewMetadata()
                    .withName(podName)
                    .withNamespace(namespace)
                    .addToLabels(Labels.Kubernetes.INSTANCE, clusterName)
                    .addToLabels(Labels.Kubernetes.MANAGED_BY, Labels.Values.RABBITMQ_OPERATOR)
                    .addToLabels(Labels.Kubernetes.PART_OF, Labels.Values.RABBITMQ)
                    .addToLabels(Labels.Indeed.LOCKED_BY, "network-partition")
                    .addToLabels(Labels.Indeed.getIndeedLabels(networkPartition))
                    .addToOwnerReferences(new OwnerReference(networkPartition.getApiVersion(), false, true, networkPartition.getKind(), networkPartition.getName(), networkPartition.getMetadata().getUid()))
                    .endMetadata()
                    .withSpec(rabbitMQPods.buildPodSpec(clusterName, rabbit.getInitContainerImage(), container))
                    .editSpec()
                    .withHostname(podName)
                    .withSubdomain(RabbitMQServices.getDiscoveryServiceName(clusterName))
                    .addNewVolume().withName(RABBITMQ_STORAGE_NAME).withNewPersistentVolumeClaim().withClaimName(RABBITMQ_STORAGE_NAME + "-" + podName).endPersistentVolumeClaim().endVolume()
                    .endSpec()
                    .build()
    ).collect(Collectors.toList());
}
 

@Test
public void readFileEscapedParams() throws IOException {
  // Wait for resources to get ready
  ReadyEntity<Pod> podReady = new ReadyEntity<>(Pod.class, client, pod1.getMetadata().getName(), currentNamespace);
  await().atMost(30, TimeUnit.SECONDS).until(podReady);
  ExecWatch watch = client.pods().inNamespace(currentNamespace).withName(pod1.getMetadata().getName()).writingOutput(System.out).exec("sh", "-c", "echo 'H$ll* (W&RLD}' > /msg");
  try (InputStream is = client.pods().inNamespace(currentNamespace).withName(pod1.getMetadata().getName()).file("/msg").read())  {
    String result = new BufferedReader(new InputStreamReader(is)).lines().collect(Collectors.joining("\n"));
    assertEquals("H$ll* (W&RLD}", result);
  }
}
 

private void waitForPodsToBeDeleted(final List<Pod> pods) throws InterruptedException {
    for (final Pod pod : pods) {
        final String podName = ModelFieldLookups.getName(pod);
        log.info("Waiting for pod {} to be deleted", podName);
        // if we timeout here, this reconciliation will abort and we'll try again on the next reconciliation loop
        podController.waitForDeletion(podName, namespace, 1, TimeUnit.MINUTES);
    }
}
 

private void waitForPodsToDrain(
        final String clusterName,
        final List<Pod> pods
) throws InterruptedException {
    final RabbitMQConnectionInfo connectionInfo = new RabbitMQConnectionInfo(clusterName, namespace, RabbitMQServices.getDiscoveryServiceName(clusterName), ModelFieldLookups.getName(pods.get(0)));

    while(!queuesEmptyOperation.execute(connectionInfo)) {
        Thread.sleep(TimeUnit.SECONDS.toMillis(5));
    }
}
 

/**
 * Creates pod from resource
 *
 * @param namespace
 * @param resources
 * @throws Exception
 */
public void createPodFromResource(String namespace, Pod resources) throws Exception {
    if (getPod(namespace, resources.getMetadata().getName()) == null) {
        Pod podRes = client.pods().inNamespace(namespace).create(resources);
        if (verboseLog) {
            log.info("Pod {} created", podRes.getMetadata().getName());
        }
    } else {
        if (verboseLog) {
            log.info("Pod {} already exists", resources.getMetadata().getName());
        }
    }
}
 

@Test
public void testOnCloseEvent() throws InterruptedException {
  logger.info("testOnCloseEvent");
  final CountDownLatch eventLatch = new CountDownLatch(2);
  final CountDownLatch closeLatch = new CountDownLatch(1);
  KubernetesClient client = server.getClient().inNamespace("test");

  server.expect()
    .withPath("/api/v1/namespaces/test/pods?fieldSelector=metadata.name%3Dpod1&resourceVersion=1&watch=true")
    .andUpgradeToWebSocket().open().waitFor(EVENT_WAIT_PERIOD).andEmit(new WatchEvent(pod1, "MODIFIED")).waitFor(EVENT_WAIT_PERIOD)
    .andEmit(new WatchEvent(pod1, "MODIFIED")).done().once();

  Watch watch = client.pods().withName("pod1").withResourceVersion("1").watch(new Watcher<Pod>() {
    @Override
    public void eventReceived(Action action, Pod resource) {
      eventLatch.countDown();
    }

    @Override
    public void onClose(KubernetesClientException cause) {
      closeLatch.countDown();
    }
  });

  assertTrue(eventLatch.await(10, TimeUnit.SECONDS));
  watch.close();
  assertTrue(closeLatch.await(10, TimeUnit.SECONDS));
}
 

@Test
public void shouldConfigureEnvironmentVariables() {
  final Pod pod = createPod(
      WORKFLOW_INSTANCE,
      DockerRunner.RunSpec.builder()
          .executionId(TEST_EXECUTION_ID)
          .args("hello", "world")
          .imageName("busybox")
          .trigger(Trigger.unknown("trigger-id"))
          .commitSha("abc123")
          .build(),
      EMPTY_SECRET_SPEC);

  final List<EnvVar> envVars = pod.getSpec().getContainers().get(0).getEnv();

  assertThat(envVars, hasItem(envVar(COMPONENT_ID, WORKFLOW_INSTANCE.workflowId().componentId())));
  assertThat(envVars, hasItem(envVar(WORKFLOW_ID, WORKFLOW_INSTANCE.workflowId().id())));
  assertThat(envVars, hasItem(envVar(PARAMETER, WORKFLOW_INSTANCE.parameter())));
  assertThat(envVars, hasItem(envVar(SERVICE_ACCOUNT, "")));
  assertThat(envVars, hasItem(envVar(DOCKER_IMAGE, "busybox")));
  assertThat(envVars, hasItem(envVar(TRIGGER_ID, "trigger-id")));
  assertThat(envVars, hasItem(envVar(TRIGGER_TYPE, "unknown")));
  assertThat(envVars, hasItem(envVar(COMMIT_SHA, "abc123")));
  assertThat(envVars, hasItem(envVar(EXECUTION_ID, TEST_EXECUTION_ID)));
  assertThat(envVars, hasItem(envVar(DOCKER_ARGS, "hello world")));
  assertThat(envVars, hasItem(envVar(ENVIRONMENT, STYX_ENVIRONMENT)));
  assertThat(envVars, hasItem(envVar(LOGGING, "structured")));
}
 

/**
 * Lets wait until there are enough Ready pods of the given RC
 */
private void waitUntilPodsAreReady(final T obj, final String namespace, final int requiredPodCount) {
  final CountDownLatch countDownLatch = new CountDownLatch(1);
  final AtomicInteger podCount = new AtomicInteger(0);

  final Runnable readyPodsPoller = () -> {
    PodList podList = listSelectedPods(obj);
    int count = 0;
    List<Pod> items = podList.getItems();
    for (Pod item : items) {
      for (PodCondition c : item.getStatus().getConditions()) {
        if (c.getType().equals("Ready") && c.getStatus().equals("True")) {
          count++;
        }
      }
    }
    podCount.set(count);
    if (count == requiredPodCount) {
      countDownLatch.countDown();
    }
  };

  ScheduledExecutorService executor = Executors.newSingleThreadScheduledExecutor();
  ScheduledFuture poller = executor.scheduleWithFixedDelay(readyPodsPoller, 0, 1, TimeUnit.SECONDS);
  ScheduledFuture logger = executor.scheduleWithFixedDelay(() -> LOG.debug("Only {}/{} pod(s) ready for {}: {} in namespace: {} seconds so waiting...",
      podCount.get(), requiredPodCount, obj.getKind(), obj.getMetadata().getName(), namespace), 0, loggingIntervalMillis, TimeUnit.MILLISECONDS);
  try {
    countDownLatch.await(rollingTimeoutMillis, TimeUnit.MILLISECONDS);
    executor.shutdown();
  } catch (InterruptedException e) {
    Thread.currentThread().interrupt();
    poller.cancel(true);
    logger.cancel(true);
    executor.shutdown();
    LOG.warn("Only {}/{} pod(s) ready for {}: {} in namespace: {}  after waiting for {} seconds so giving up",
        podCount.get(), requiredPodCount, obj.getKind(), obj.getMetadata().getName(), namespace, TimeUnit.MILLISECONDS.toSeconds(rollingTimeoutMillis));
  }
}