StatefulSetOperator extends AbstractScalableResourceOperator<KubernetesClient, StatefulSet, StatefulSetList, DoneableStatefulSet, RollableScalableResource<StatefulSet, DoneableStatefulSet>> { public Future<Void> deleteAsync(String namespace, String name, boolean cascading) { Promise<Void> result = Promise.promise(); vertx.createSharedWorkerExecutor("kubernetes-ops-tool").executeBlocking( future -> { try { Boolean deleted = operation().inNamespace(namespace).withName(name).withPropagationPolicy(cascading ? DeletionPropagation.FOREGROUND : DeletionPropagation.ORPHAN).withGracePeriod(-1L).delete(); if (deleted) { log.debug("{} {} in namespace {} has been deleted", resourceKind, name, namespace); future.complete(); } else { log.debug("{} {} in namespace {} has been not been deleted", resourceKind, name, namespace); future.fail(resourceKind + " " + name + " in namespace " + namespace + " has been not been deleted"); } } catch (Exception e) { log.debug("Caught exception while deleting {} {} in namespace {}", resourceKind, name, namespace, e); future.fail(e); } }, true, result ); return result.future(); } 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 testCascadingDeleteAsync(VertxTestContext context) { Deletable mockDeletable = mock(Deletable.class); when(mockDeletable.delete()).thenReturn(Boolean.TRUE); Resource mockERPD = mock(resourceType()); when(mockERPD.withPropagationPolicy(any(DeletionPropagation.class))).thenReturn(mockDeletable); when(mockERPD.withGracePeriod(anyLong())).thenReturn(mockDeletable); RollableScalableResource mockRSR = mock(RollableScalableResource.class); ArgumentCaptor<DeletionPropagation> cascadingCaptor = ArgumentCaptor.forClass(DeletionPropagation.class); when(mockRSR.withPropagationPolicy(cascadingCaptor.capture())).thenReturn(mockERPD); NonNamespaceOperation mockNameable = mock(NonNamespaceOperation.class); when(mockNameable.withName(matches(RESOURCE_NAME))).thenReturn(mockRSR); MixedOperation mockCms = mock(MixedOperation.class); when(mockCms.inNamespace(matches(NAMESPACE))).thenReturn(mockNameable); PodOperator podOperator = mock(PodOperator.class); PvcOperator pvcOperator = mock(PvcOperator.class); 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 async = context.checkpoint(); op.deleteAsync(NAMESPACE, RESOURCE_NAME, true) .onComplete(context.succeeding(v -> context.verify(() -> { assertThat(cascadingCaptor.getValue(), is(DeletionPropagation.FOREGROUND)); async.flag(); }))); }
@Test public void testNonCascadingDeleteAsync(VertxTestContext context) { Deletable mockDeletable = mock(Deletable.class); when(mockDeletable.delete()).thenReturn(Boolean.TRUE); Resource mockERPD = mock(resourceType()); when(mockERPD.withPropagationPolicy(any(DeletionPropagation.class))).thenReturn(mockDeletable); when(mockERPD.withGracePeriod(anyLong())).thenReturn(mockDeletable); RollableScalableResource mockRSR = mock(RollableScalableResource.class); ArgumentCaptor<DeletionPropagation> cascadingCaptor = ArgumentCaptor.forClass(DeletionPropagation.class); when(mockRSR.withPropagationPolicy(cascadingCaptor.capture())).thenReturn(mockERPD); NonNamespaceOperation mockNameable = mock(NonNamespaceOperation.class); when(mockNameable.withName(matches(RESOURCE_NAME))).thenReturn(mockRSR); MixedOperation mockCms = mock(MixedOperation.class); when(mockCms.inNamespace(matches(NAMESPACE))).thenReturn(mockNameable); PodOperator podOperator = mock(PodOperator.class); PvcOperator pvcOperator = mock(PvcOperator.class); 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.deleteAsync(NAMESPACE, RESOURCE_NAME, false) .onComplete(context.succeeding(v -> context.verify(() -> { assertThat(cascadingCaptor.getValue(), is(DeletionPropagation.ORPHAN)); a.flag(); }))); }
@Test public void testDeleteAsyncNotDeleted(VertxTestContext context) { EditReplacePatchDeletable mockERPD = mock(EditReplacePatchDeletable.class); when(mockERPD.delete()).thenReturn(Boolean.FALSE); RollableScalableResource mockRSR = mock(RollableScalableResource.class); when(mockRSR.withPropagationPolicy(any(DeletionPropagation.class))).thenReturn(mockERPD); NonNamespaceOperation mockNameable = mock(NonNamespaceOperation.class); when(mockNameable.withName(matches(RESOURCE_NAME))).thenReturn(mockRSR); MixedOperation mockCms = mock(MixedOperation.class); when(mockCms.inNamespace(matches(NAMESPACE))).thenReturn(mockNameable); PodOperator podOperator = mock(PodOperator.class); PvcOperator pvcOperator = mock(PvcOperator.class); 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.deleteAsync(NAMESPACE, RESOURCE_NAME, false) .onComplete(context.failing(e -> a.flag())); }
@Test public void testDeleteAsyncFailing(VertxTestContext context) { Deletable mockDeletable = mock(Deletable.class); when(mockDeletable.delete()).thenThrow(new MockitoException("Something failed")); Resource mockERPD = mock(resourceType()); when(mockERPD.withPropagationPolicy(any(DeletionPropagation.class))).thenReturn(mockDeletable); when(mockERPD.withGracePeriod(anyLong())).thenReturn(mockDeletable); RollableScalableResource mockRSR = mock(RollableScalableResource.class); when(mockRSR.withPropagationPolicy(any(DeletionPropagation.class))).thenReturn(mockERPD); NonNamespaceOperation mockNameable = mock(NonNamespaceOperation.class); when(mockNameable.withName(matches(RESOURCE_NAME))).thenReturn(mockRSR); MixedOperation mockCms = mock(MixedOperation.class); when(mockCms.inNamespace(matches(NAMESPACE))).thenReturn(mockNameable); PodOperator podOperator = mock(PodOperator.class); PvcOperator pvcOperator = mock(PvcOperator.class); 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 async = context.checkpoint(); op.deleteAsync(NAMESPACE, RESOURCE_NAME, false) .onComplete(context.failing(e -> context.verify(() -> { assertThat(e, instanceOf(MockitoException.class)); assertThat(e.getMessage(), is("Something failed")); async.flag(); }))); }