Added endpoint for admins to check quotas of all namespaces (#216)

* Started add an endpoint for admins to check quotas of all namespaces * Fixed unit tests * Improved code coverage * Improved code coverage
michelin · Nov 7, 2022 · e01b14a · e01b14a
1 parent 6962d6f
commit e01b14a
Show file tree

Hide file tree

Showing 17 changed files with 644 additions and 140 deletions.
diff --git a/api/src/main/java/com/michelin/ns4kafka/controllers/NamespaceController.java b/api/src/main/java/com/michelin/ns4kafka/controllers/NamespaceController.java
@@ -24,7 +24,6 @@
 @Tag(name = "Namespaces")
 @Controller("/api/namespaces")
 public class NamespaceController extends NonNamespacedResourceController {
-
     @Inject
     NamespaceService namespaceService;
 

diff --git a/api/src/main/java/com/michelin/ns4kafka/controllers/ResourceQuotaController.java b/api/src/main/java/com/michelin/ns4kafka/controllers/ResourceQuotaController.java
@@ -4,6 +4,7 @@
 import com.michelin.ns4kafka.models.Namespace;
 import com.michelin.ns4kafka.models.quota.ResourceQuota;
 import com.michelin.ns4kafka.models.quota.ResourceQuotaResponse;
+import com.michelin.ns4kafka.security.ResourceBasedSecurityRule;
 import com.michelin.ns4kafka.services.ResourceQuotaService;
 import com.michelin.ns4kafka.utils.enums.ApplyStatus;
 import com.michelin.ns4kafka.utils.exceptions.ResourceValidationException;
@@ -14,6 +15,7 @@
 import io.micronaut.scheduling.annotation.ExecuteOn;
 import io.swagger.v3.oas.annotations.tags.Tag;
 
+import javax.annotation.security.RolesAllowed;
 import javax.inject.Inject;
 import javax.validation.Valid;
 import java.time.Instant;
@@ -25,12 +27,19 @@
 @Controller(value = "/api/namespaces/{namespace}/resource-quotas")
 @ExecuteOn(TaskExecutors.IO)
 public class ResourceQuotaController extends NamespacedResourceController {
-    /**
-     * The resource quota service
-     */
     @Inject
     ResourceQuotaService resourceQuotaService;
 
+    /**
+     * Get the sum of all quotas of all namespaces with all the current consumed resources
+     * @return A quota response
+     */
+    @Get("/all")
+    @RolesAllowed(ResourceBasedSecurityRule.IS_ADMIN)
+    public ResourceQuotaResponse listAllNamespaces() {
+        return resourceQuotaService.getCurrentResourcesQuotasAllNamespaces();
+    }
+
     /**
      * Get all the quotas by namespace
      * @param namespace The namespace
@@ -39,7 +48,8 @@ public class ResourceQuotaController extends NamespacedResourceController {
     @Get
     public List<ResourceQuotaResponse> list(String namespace) {
         Namespace ns = getNamespace(namespace);
-        return List.of(resourceQuotaService.toResponse(ns, resourceQuotaService.findByNamespace(namespace)));
+        return List.of(resourceQuotaService.getCurrentResourcesQuotasByNamespace(ns,
+                resourceQuotaService.findByNamespace(namespace)));
     }
 
     /**
@@ -55,7 +65,7 @@ public Optional<ResourceQuotaResponse> get(String namespace, String quota) {
         if (resourceQuota.isEmpty()) {
             return Optional.empty();
         }
-        return Optional.of(resourceQuotaService.toResponse(ns, resourceQuota));
+        return Optional.of(resourceQuotaService.getCurrentResourcesQuotasByNamespace(ns, resourceQuota));
     }
 
     /**

diff --git a/api/src/main/java/com/michelin/ns4kafka/repositories/ConnectorRepository.java b/api/src/main/java/com/michelin/ns4kafka/repositories/ConnectorRepository.java
@@ -5,7 +5,29 @@
 import java.util.List;
 
 public interface ConnectorRepository {
+    /**
+     * Find all connectors
+     * @return The list of connectors
+     */
+    List<Connector> findAll();
+
+    /**
+     * Find all connectors by cluster
+     * @param cluster The cluster
+     * @return The list of connectors
+     */
     List<Connector> findAllForCluster(String cluster);
+
+    /**
+     * Create a given connector
+     * @param connector The connector to create
+     * @return The created connector
+     */
     Connector create(Connector connector);
+
+    /**
+     * Delete a given connector
+     * @param connector The connector to delete
+     */
     void delete(Connector connector);
 }
diff --git a/api/src/main/java/com/michelin/ns4kafka/repositories/ResourceQuotaRepository.java b/api/src/main/java/com/michelin/ns4kafka/repositories/ResourceQuotaRepository.java
@@ -2,9 +2,16 @@
 
 import com.michelin.ns4kafka.models.quota.ResourceQuota;
 
+import java.util.List;
 import java.util.Optional;
 
 public interface ResourceQuotaRepository {
+    /**
+     * Find all quotas of all namespaces
+     * @return The resource quotas
+     */
+    List<ResourceQuota> findAll();
+
     /**
      * Get resource quota by namespace
      * @param namespace The namespace used to research

diff --git a/api/src/main/java/com/michelin/ns4kafka/repositories/TopicRepository.java b/api/src/main/java/com/michelin/ns4kafka/repositories/TopicRepository.java
@@ -5,18 +5,29 @@
 import java.util.List;
 
 public interface TopicRepository {
+    /**
+     * Find all topics
+     * @return The list of topics
+     */
+    List<Topic> findAll();
 
-    /***
-     *
-     * @param cluster the cluster id
-     * @return the list of all topics for this cluster as a KV Map with :<br>
-     * key : String : Topic Name<br>
-     * value : Topic : Topic data<br>
+    /**
+     * Find all topics by cluster
+     * @param cluster The cluster
+     * @return The list of topics
      */
     List<Topic> findAllForCluster(String cluster);
 
+    /**
+     * Create a given topic
+     * @param topic The topic to create
+     * @return The created topic
+     */
     Topic create(Topic topic);
 
+    /**
+     * Delete a given topic
+     * @param topic The topic to delete
+     */
     void delete(Topic topic);
-
 }
diff --git a/api/src/main/java/com/michelin/ns4kafka/repositories/kafka/KafkaConnectorRepository.java b/api/src/main/java/com/michelin/ns4kafka/repositories/kafka/KafkaConnectorRepository.java
@@ -8,6 +8,7 @@
 import org.apache.kafka.clients.producer.Producer;
 
 import javax.inject.Singleton;
+import java.util.ArrayList;
 import java.util.List;
 import java.util.stream.Collectors;
 
@@ -34,16 +35,39 @@ void receive(ConsumerRecord<String, Connector> record) {
         super.receive(record);
     }
 
+    /**
+     * Create a given connector
+     * @param connector The connector to create
+     * @return The created connector
+     */
     @Override
     public Connector create(Connector connector) {
         return this.produce(getMessageKey(connector),connector);
     }
 
+    /**
+     * Delete a given connector
+     * @param connector The connector to delete
+     */
     @Override
     public void delete(Connector connector) {
         this.produce(getMessageKey(connector),null);
     }
 
+    /**
+     * Find all connectors
+     * @return The list of connectors
+     */
+    @Override
+    public List<Connector> findAll() {
+        return new ArrayList<>(getKafkaStore().values());
+    }
+
+    /**
+     * Find all connectors by cluster
+     * @param cluster The cluster
+     * @return The list of connectors
+     */
     @Override
     public List<Connector> findAllForCluster(String cluster) {
         return getKafkaStore().values().stream()

diff --git a/api/src/main/java/com/michelin/ns4kafka/repositories/kafka/KafkaResourceQuotaRepository.java b/api/src/main/java/com/michelin/ns4kafka/repositories/kafka/KafkaResourceQuotaRepository.java
@@ -8,6 +8,8 @@
 import org.apache.kafka.clients.producer.Producer;
 
 import javax.inject.Singleton;
+import java.util.ArrayList;
+import java.util.List;
 import java.util.Optional;
 
 @Singleton
@@ -32,6 +34,15 @@ String getMessageKey(ResourceQuota message) {
         return message.getMetadata().getNamespace();
     }
 
+    /**
+     * Find all quotas of all namespaces
+     * @return The resource quotas
+     */
+    @Override
+    public List<ResourceQuota> findAll() {
+        return new ArrayList<>(getKafkaStore().values());
+    }
+
     /**
      * Get resource quota by namespace
      * @param namespace The namespace used to research

diff --git a/api/src/main/java/com/michelin/ns4kafka/repositories/kafka/KafkaTopicRepository.java b/api/src/main/java/com/michelin/ns4kafka/repositories/kafka/KafkaTopicRepository.java
@@ -11,6 +11,7 @@
 import org.apache.kafka.clients.producer.Producer;
 
 import javax.inject.Singleton;
+import java.util.ArrayList;
 import java.util.List;
 import java.util.stream.Collectors;
 
@@ -32,11 +33,20 @@ String getMessageKey(Topic topic) {
         return topic.getMetadata().getCluster()+"/"+topic.getMetadata().getName();
     }
 
+    /**
+     * Create a given topic
+     * @param topic The topic to create
+     * @return The created topic
+     */
     @Override
     public Topic create(Topic topic) {
         return this.produce(getMessageKey(topic), topic);
     }
 
+    /**
+     * Delete a given topic
+     * @param topic The topic to delete
+     */
     @Override
     public void delete(Topic topic) {
         this.produce(getMessageKey(topic),null);
@@ -47,20 +57,25 @@ void receive(ConsumerRecord<String, Topic> record) {
         super.receive(record);
     }
 
-    //@Override
-    //public List<Topic> findAllForNamespace(Namespace namespace) {
-    //        return getKafkaStore().values()
-    //                .stream()
-    //                .filter(topic -> topic.getMetadata().getCluster().equals(namespace.getCluster()))
-    //                .collect(Collectors.toList());
-    //}
+    /**
+     * Find all topics
+     * @return The list of topics
+     */
+    @Override
+    public List<Topic> findAll() {
+        return new ArrayList<>(getKafkaStore().values());
+    }
 
+    /**
+     * Find all topics by cluster
+     * @param cluster The cluster
+     * @return The list of topics
+     */
     @Override
     public List<Topic> findAllForCluster(String cluster) {
         return getKafkaStore().values()
                 .stream()
                 .filter(topic -> topic.getMetadata().getCluster().equals(cluster))
                 .collect(Collectors.toList());
     }
-
 }
diff --git a/api/src/main/java/com/michelin/ns4kafka/services/ConnectorService.java b/api/src/main/java/com/michelin/ns4kafka/services/ConnectorService.java
@@ -42,6 +42,15 @@ public class ConnectorService {
     @Inject
     ConnectClusterService connectClusterService;
 
+
+    /**
+     * Find all connectors
+     * @return The list of connectors
+     */
+    public List<Connector> findAll() {
+        return connectorRepository.findAll();
+    }
+
     /**
      * Find all connectors by given namespace
      * @param namespace The namespace

diff --git a/api/src/main/java/com/michelin/ns4kafka/services/NamespaceService.java b/api/src/main/java/com/michelin/ns4kafka/services/NamespaceService.java
@@ -14,43 +14,49 @@
 
 @Singleton
 public class NamespaceService {
-
     @Inject
     NamespaceRepository namespaceRepository;
+
     @Inject
     List<KafkaAsyncExecutorConfig> kafkaAsyncExecutorConfigList;
+
     @Inject
     TopicService topicService;
+
     @Inject
     RoleBindingService roleBindingService;
+
     @Inject
     AccessControlEntryService accessControlEntryService;
+
     @Inject
     ConnectorService connectorService;
 
     /**
-     * Namespace validation in case of new namespace
-     *
-     * @param namespace
-     * @return
+     * Validate new namespace creation
+     * @param namespace The namespace to create
+     * @return A list of validation errors
      */
     public List<String> validateCreation(Namespace namespace) {
-
         List<String> validationErrors = new ArrayList<>();
 
-        if (kafkaAsyncExecutorConfigList.stream()
-                .noneMatch(config -> config.getName().equals(namespace.getMetadata().getCluster()))) {
-            validationErrors.add("Invalid value " + namespace.getMetadata().getCluster()
-                    + " for cluster: Cluster doesn't exist");
+        if (kafkaAsyncExecutorConfigList.stream().noneMatch(config -> config.getName().equals(namespace.getMetadata().getCluster()))) {
+            validationErrors.add("Invalid value " + namespace.getMetadata().getCluster() + " for cluster: Cluster doesn't exist");
         }
+
         if (namespaceRepository.findAllForCluster(namespace.getMetadata().getCluster()).stream()
                 .anyMatch(namespace1 -> namespace1.getSpec().getKafkaUser().equals(namespace.getSpec().getKafkaUser()))) {
-            validationErrors.add("Invalid value " + namespace.getSpec().getKafkaUser()
-                    + " for user: KafkaUser already exists");
+            validationErrors.add("Invalid value " + namespace.getSpec().getKafkaUser() + " for user: KafkaUser already exists");
         }
+
         return validationErrors;
     }
 
+    /**
+     * Validate the Connect clusters of the namespace
+     * @param namespace The namespace
+     * @return A list of validation errors
+     */
     public List<String> validate(Namespace namespace) {
         return namespace.getSpec().getConnectClusters()
                 .stream()
@@ -59,6 +65,12 @@ public List<String> validate(Namespace namespace) {
                 .collect(Collectors.toList());
     }
 
+    /**
+     * Check if a given Connect cluster exists on a given Kafka cluster
+     * @param kafkaCluster The Kafka cluster
+     * @param connectCluster The Connect cluster
+     * @return true it does, false otherwise
+     */
     private boolean connectClusterExists(String kafkaCluster, String connectCluster) {
         return kafkaAsyncExecutorConfigList.stream()
                 .anyMatch(kafkaAsyncExecutorConfig -> kafkaAsyncExecutorConfig.getName().equals(kafkaCluster) &&