add namespace wildcard deletion

michelin · Sep 17, 2024 · e94a24e · e94a24e
1 parent d529e59
commit e94a24e
Show file tree

Hide file tree

Showing 2 changed files with 134 additions and 8 deletions.
diff --git a/src/main/java/com/michelin/ns4kafka/controller/NamespaceController.java b/src/main/java/com/michelin/ns4kafka/controller/NamespaceController.java
@@ -24,9 +24,13 @@
 import jakarta.validation.Valid;
 import java.time.Instant;
 import java.util.ArrayList;
+import java.util.Collection;
 import java.util.Date;
+import java.util.HashMap;
 import java.util.List;
+import java.util.Map;
 import java.util.Optional;
+import java.util.stream.Collectors;
 
 /**
  * Controller to manage the namespaces.
@@ -141,17 +145,69 @@ public HttpResponse<Void> delete(String namespace, @QueryValue(defaultValue = "f
             return HttpResponse.noContent();
         }
 
-        var namespaceToDelete = optionalNamespace.get();
+        performDeletion(optionalNamespace.get());
+        return HttpResponse.noContent();
+    }
 
-        sendEventLog(
-            namespaceToDelete,
-            ApplyStatus.deleted,
-            namespaceToDelete.getSpec(),
-            null,
-            EMPTY_STRING
+    /**
+     * Delete namespaces.
+     *
+     * @param dryrun Is dry run mode or not ?
+     * @param name The name parameter
+     * @return An HTTP response
+     */
+    @Delete
+    public HttpResponse<Void> bulkDelete(@QueryValue(defaultValue = "*") String name,
+                                         @QueryValue(defaultValue = "false") boolean dryrun) {
+        List<Namespace> namespaces = namespaceService.findByWildcardName(name);
+        if (namespaces.isEmpty()) {
+            return HttpResponse.notFound();
+        }
+
+        Map<String, List<String>> validationErrors = new HashMap<>();
+
+        namespaces.forEach(namespace ->
+            validationErrors.put(
+                    namespace.getMetadata().getName(),
+                    namespaceService.findAllResourcesByNamespace(namespace)
+                            .stream().filter(r -> !r.isEmpty())
+                            .map(FormatErrorUtils::invalidNamespaceDeleteOperation)
+                            .toList())
         );
 
-        namespaceService.delete(optionalNamespace.get());
+        if (validationErrors.values().stream()
+                .anyMatch(list -> !list.isEmpty())) {
+            throw new ResourceValidationException(
+                    NAMESPACE,
+                    validationErrors.keySet().stream()
+                            .map(Object::toString)
+                            .collect(Collectors.joining("/")),
+                    validationErrors.values().stream()
+                            .flatMap(Collection::stream)
+                            .collect(Collectors.toList()));
+        }
+
+        if (dryrun) {
+            return HttpResponse.noContent();
+        }
+
+        namespaces.forEach(this::performDeletion);
         return HttpResponse.noContent();
     }
+
+    /**
+     * Perform the deletion of the namespace and send an event log.
+     *
+     * @param namespace The namespace to delete
+     */
+    private void performDeletion(Namespace namespace) {
+        sendEventLog(
+                namespace,
+                ApplyStatus.deleted,
+                namespace.getSpec(),
+                null,
+                EMPTY_STRING
+        );
+        namespaceService.delete(namespace);
+    }
 }
diff --git a/src/test/java/com/michelin/ns4kafka/controller/NamespaceControllerTest.java b/src/test/java/com/michelin/ns4kafka/controller/NamespaceControllerTest.java
@@ -371,4 +371,74 @@ void shouldNotDeleteNamespaceWhenResourcesAreStillLinkedWithIt() {
             () -> namespaceController.delete("namespace", false));
         verify(namespaceService, never()).delete(any());
     }
+
+    @Test
+    void shouldDeleteNamespaces() {
+        Namespace existing1 = Namespace.builder()
+                .metadata(Metadata.builder()
+                        .name("namespace1")
+                        .cluster("local")
+                        .build())
+                .spec(Namespace.NamespaceSpec.builder()
+                        .kafkaUser("user")
+                        .build())
+                .build();
+        Namespace existing2 = Namespace.builder()
+                .metadata(Metadata.builder()
+                        .name("namespace2")
+                        .cluster("local")
+                        .build())
+                .spec(Namespace.NamespaceSpec.builder()
+                        .kafkaUser("user")
+                        .build())
+                .build();
+
+        when(namespaceService.findByWildcardName("namespace*"))
+                .thenReturn(List.of(existing1, existing2));
+        when(namespaceService.findAllResourcesByNamespace(existing1))
+                .thenReturn(List.of());
+        when(namespaceService.findAllResourcesByNamespace(existing2))
+                .thenReturn(List.of());
+        when(securityService.username())
+                .thenReturn(Optional.of("test-user"));
+        when(securityService.hasRole(ResourceBasedSecurityRule.IS_ADMIN))
+                .thenReturn(false);
+
+        doNothing().when(applicationEventPublisher).publishEvent(any());
+        var result = namespaceController.bulkDelete("namespace*", false);
+        assertEquals(HttpResponse.noContent().getStatus(), result.getStatus());
+    }
+
+    @Test
+    void shouldNotDeleteNamespacesWhenResourcesAreStillLinkedWithIt() {
+        Namespace existing1 = Namespace.builder()
+                .metadata(Metadata.builder()
+                        .name("namespace1")
+                        .cluster("local")
+                        .build())
+                .spec(Namespace.NamespaceSpec.builder()
+                        .kafkaUser("user")
+                        .build())
+                .build();
+        Namespace existing2 = Namespace.builder()
+                .metadata(Metadata.builder()
+                        .name("namespace2")
+                        .cluster("local")
+                        .build())
+                .spec(Namespace.NamespaceSpec.builder()
+                        .kafkaUser("user")
+                        .build())
+                .build();
+
+        when(namespaceService.findByWildcardName("namespace*"))
+                .thenReturn(List.of(existing1, existing2));
+        when(namespaceService.findAllResourcesByNamespace(existing1))
+                .thenReturn(List.of("Topic/topic1"));
+        when(namespaceService.findAllResourcesByNamespace(existing2))
+                .thenReturn(List.of());
+
+        assertThrows(ResourceValidationException.class,
+                () -> namespaceController.bulkDelete("namespace*", false));
+        verify(namespaceService, never()).delete(any());
+    }
 }