StatefulSetOperator extends AbstractScalableResourceOperator<KubernetesClient, StatefulSet, StatefulSetList, DoneableStatefulSet, RollableScalableResource<StatefulSet, DoneableStatefulSet>> { Future<Void> maybeRestartPod(StatefulSet sts, String podName, Function<Pod, List<String>> podNeedsRestart) { long pollingIntervalMs = 1_000; long timeoutMs = operationTimeoutMs; String namespace = sts.getMetadata().getNamespace(); String name = sts.getMetadata().getName(); return podOperations.getAsync(sts.getMetadata().getNamespace(), podName).compose(pod -> { Future<Void> fut; List<String> reasons = podNeedsRestart.apply(pod); if (reasons != null && !reasons.isEmpty()) { log.debug("Rolling update of {}/{}: pod {} due to {}", namespace, name, podName, reasons); fut = restartPod(sts, pod); } else { log.debug("Rolling update of {}/{}: pod {} no need to roll", namespace, name, podName); fut = Future.succeededFuture(); } return fut.compose(ignored -> { log.debug("Rolling update of {}/{}: wait for pod {} readiness", namespace, name, podName); return podOperations.readiness(namespace, podName, pollingIntervalMs, timeoutMs); }); }); } StatefulSetOperator(Vertx vertx, KubernetesClient client, long operationTimeoutMs);  StatefulSetOperator(Vertx vertx, KubernetesClient client, long operationTimeoutMs,
                               PodOperator podOperator, PvcOperator pvcOperator); Future<Void> maybeRollingUpdate(StatefulSet sts, Function<Pod, List<String>> podNeedsRestart); abstract Future<Void> maybeRollingUpdate(StatefulSet sts, Function<Pod, List<String>> podNeedsRestart, Secret clusterCaSecret, Secret coKeySecret); Future<Void> deletePvc(StatefulSet sts, String pvcName); String getPodName(StatefulSet desired, int podId); @SuppressWarnings("deprecation") static int getStsGeneration(StatefulSet resource); @SuppressWarnings("deprecation") static int getPodGeneration(Pod resource); Future<Void> deleteAsync(String namespace, String name, boolean cascading);  }
@Test public void testRollingUpdateSuccess(VertxTestContext context) { StatefulSet resource = resource(); Resource mockResource = mock(resourceType()); when(mockResource.get()).thenReturn(resource); PodOperator podOperator = mock(PodOperator.class); when(podOperator.waitFor(anyString(), anyString(), anyLong(), anyLong(), any(BiPredicate.class))).thenReturn(Future.succeededFuture()); when(podOperator.readiness(anyString(), anyString(), anyLong(), anyLong())).thenReturn(Future.succeededFuture()); when(podOperator.reconcile(anyString(), anyString(), any())).thenReturn(Future.succeededFuture()); when(podOperator.getAsync(anyString(), anyString())).thenReturn(Future.succeededFuture(new PodBuilder().withNewMetadata().withName("my-pod-0").endMetadata().build())); when(podOperator.restart(anyString(), any(), anyLong())).thenReturn(Future.succeededFuture()); PvcOperator pvcOperator = mock(PvcOperator.class); when(pvcOperator.reconcile(anyString(), anyString(), any())).thenReturn(Future.succeededFuture()); SecretOperator secretOperator = mock(SecretOperator.class); when(secretOperator.reconcile(anyString(), anyString(), any())).thenReturn(Future.succeededFuture()); NonNamespaceOperation mockNameable = mock(NonNamespaceOperation.class); when(mockNameable.withName(matches(resource.getMetadata().getName()))).thenReturn(mockResource); MixedOperation mockCms = mock(MixedOperation.class); when(mockCms.inNamespace(matches(resource.getMetadata().getNamespace()))).thenReturn(mockNameable); KubernetesClient mockClient = mock(KubernetesClient.class); mocker(mockClient, mockCms); StatefulSetOperator op = new StatefulSetOperator(AbstractResourceOperatorTest.vertx, mockClient, 5_000L, podOperator, pvcOperator) { @Override public Future<Void> maybeRollingUpdate(StatefulSet sts, Function<Pod, List<String>> podNeedsRestart, Secret clusterCaSecret, Secret coKeySecret) { return Future.succeededFuture(); } @Override protected boolean shouldIncrementGeneration(StatefulSetDiff diff) { return true; } }; Checkpoint a = context.checkpoint(); op.maybeRestartPod(resource, "my-pod-0", pod -> singletonList("roll")) .onComplete(context.succeeding(v -> a.flag())); }
@Test public void testRollingUpdateDeletionTimeout(VertxTestContext context) { StatefulSet resource = resource(); Resource mockResource = mock(resourceType()); when(mockResource.get()).thenReturn(resource); PodOperator podOperator = mock(PodOperator.class); when(podOperator.readiness(anyString(), anyString(), anyLong(), anyLong())).thenReturn(Future.succeededFuture()); when(podOperator.reconcile(anyString(), anyString(), any())).thenReturn(Future.succeededFuture()); AtomicInteger call = new AtomicInteger(); when(podOperator.getAsync(anyString(), anyString())).thenAnswer(invocation -> { if (call.getAndIncrement() == 0) { return Future.succeededFuture(new PodBuilder().withNewMetadata().withName("my-pod-0").endMetadata().build()); } else { return null; } }); when(podOperator.restart(anyString(), any(), anyLong())).thenReturn(Future.failedFuture(new TimeoutException())); PvcOperator pvcOperator = mock(PvcOperator.class); when(pvcOperator.reconcile(anyString(), anyString(), any())).thenReturn(Future.succeededFuture()); SecretOperator secretOperator = mock(SecretOperator.class); when(secretOperator.reconcile(anyString(), anyString(), any())).thenReturn(Future.succeededFuture()); NonNamespaceOperation mockNameable = mock(NonNamespaceOperation.class); when(mockNameable.withName(matches(resource.getMetadata().getName()))).thenReturn(mockResource); MixedOperation mockCms = mock(MixedOperation.class); when(mockCms.inNamespace(matches(resource.getMetadata().getNamespace()))).thenReturn(mockNameable); KubernetesClient mockClient = mock(KubernetesClient.class); mocker(mockClient, mockCms); StatefulSetOperator op = new StatefulSetOperator(AbstractResourceOperatorTest.vertx, mockClient, 5_000L, podOperator, pvcOperator) { @Override public Future<Void> maybeRollingUpdate(StatefulSet sts, Function<Pod, List<String>> podNeedsRestart, Secret clusterCaSecret, Secret coKeySecret) { return Future.succeededFuture(); } @Override protected boolean shouldIncrementGeneration(StatefulSetDiff diff) { return true; } }; Checkpoint a = context.checkpoint(); op.maybeRestartPod(resource, "my-pod-0", pod -> singletonList("roll")) .onComplete(context.failing(e -> context.verify(() -> { assertThat(e, instanceOf(TimeoutException.class)); a.flag(); }))); }
@Test public void testRollingUpdateReadinessTimeout(VertxTestContext context) { StatefulSet resource = resource(); Resource mockResource = mock(resourceType()); when(mockResource.get()).thenReturn(resource); PodOperator podOperator = mock(PodOperator.class); when(podOperator.waitFor(anyString(), anyString(), anyLong(), anyLong(), any(BiPredicate.class))).thenReturn(Future.succeededFuture()); when(podOperator.readiness(anyString(), anyString(), anyLong(), anyLong())).thenReturn(Future.failedFuture(new TimeoutException())); when(podOperator.reconcile(anyString(), anyString(), any())).thenReturn(Future.succeededFuture()); when(podOperator.getAsync(anyString(), anyString())).thenReturn(Future.succeededFuture(new PodBuilder().withNewMetadata().withName("my-pod-0").endMetadata().build())); when(podOperator.restart(anyString(), any(), anyLong())).thenReturn(Future.succeededFuture()); PvcOperator pvcOperator = mock(PvcOperator.class); when(pvcOperator.reconcile(anyString(), anyString(), any())).thenReturn(Future.succeededFuture()); SecretOperator secretOperator = mock(SecretOperator.class); when(secretOperator.reconcile(anyString(), anyString(), any())).thenReturn(Future.succeededFuture()); NonNamespaceOperation mockNameable = mock(NonNamespaceOperation.class); when(mockNameable.withName(matches(resource.getMetadata().getName()))).thenReturn(mockResource); MixedOperation mockCms = mock(MixedOperation.class); when(mockCms.inNamespace(matches(resource.getMetadata().getNamespace()))).thenReturn(mockNameable); KubernetesClient mockClient = mock(KubernetesClient.class); mocker(mockClient, mockCms); StatefulSetOperator op = new StatefulSetOperator(AbstractResourceOperatorTest.vertx, mockClient, 5_000L, podOperator, pvcOperator) { @Override public Future<Void> maybeRollingUpdate(StatefulSet sts, Function<Pod, List<String>> podNeedsRestart, Secret clusterCaSecret, Secret coKeySecret) { return Future.succeededFuture(); } @Override protected boolean shouldIncrementGeneration(StatefulSetDiff diff) { return true; } }; Checkpoint a = context.checkpoint(); op.maybeRestartPod(resource, "my-pod-0", pod -> singletonList("roll")).onComplete(context.failing(e -> context.verify(() -> { assertThat(e, instanceOf(TimeoutException.class)); a.flag(); }))); }
@Test public void testRollingUpdateReconcileFailed(VertxTestContext context) { StatefulSet resource = resource(); Resource mockResource = mock(resourceType()); when(mockResource.get()).thenReturn(resource); PodOperator podOperator = mock(PodOperator.class); when(podOperator.waitFor(anyString(), anyString(), anyLong(), anyLong(), any(BiPredicate.class))).thenReturn(Future.succeededFuture()); when(podOperator.readiness(anyString(), anyString(), anyLong(), anyLong())).thenReturn(Future.succeededFuture()); when(podOperator.getAsync(anyString(), anyString())).thenReturn(Future.succeededFuture(new PodBuilder().withNewMetadata().withName("my-pod-0").endMetadata().build())); when(podOperator.restart(anyString(), any(), anyLong())).thenReturn(Future.failedFuture("reconcile failed")); PvcOperator pvcOperator = mock(PvcOperator.class); when(pvcOperator.reconcile(anyString(), anyString(), any())).thenReturn(Future.succeededFuture()); SecretOperator secretOperator = mock(SecretOperator.class); when(secretOperator.reconcile(anyString(), anyString(), any())).thenReturn(Future.succeededFuture()); NonNamespaceOperation mockNameable = mock(NonNamespaceOperation.class); when(mockNameable.withName(matches(resource.getMetadata().getName()))).thenReturn(mockResource); MixedOperation mockCms = mock(MixedOperation.class); when(mockCms.inNamespace(matches(resource.getMetadata().getNamespace()))).thenReturn(mockNameable); KubernetesClient mockClient = mock(KubernetesClient.class); mocker(mockClient, mockCms); StatefulSetOperator op = new StatefulSetOperator(AbstractResourceOperatorTest.vertx, mockClient, 5_000L, podOperator, pvcOperator) { @Override public Future<Void> maybeRollingUpdate(StatefulSet sts, Function<Pod, List<String>> podNeedsRestart, Secret clusterCaSecret, Secret coKeySecret) { return Future.succeededFuture(); } @Override protected boolean shouldIncrementGeneration(StatefulSetDiff diff) { return true; } }; Checkpoint a = context.checkpoint(); op.maybeRestartPod(resource, "my-pod-0", pod -> singletonList("roll")) .onComplete(context.failing(e -> context.verify(() -> { assertThat(e.getMessage(), is("reconcile failed")); a.flag(); }))); }