wip

kafka-ui-api/src/main/java/com/provectus/kafka/ui/controller/ClientQuotasController.java

@@ -4,10 +4,10 @@ import static java.util.stream.Collectors.toMap;
 
 import com.provectus.kafka.ui.api.ClientQuotasApi;
 import com.provectus.kafka.ui.model.ClientQuotasDTO;
-import com.provectus.kafka.ui.service.audit.AuditService;
+import com.provectus.kafka.ui.model.rbac.AccessContext;
+import com.provectus.kafka.ui.model.rbac.permission.ClientQuotaAction;
 import com.provectus.kafka.ui.service.quota.ClientQuotaRecord;
-import com.provectus.kafka.ui.service.quota.QuotaService;
+import com.provectus.kafka.ui.service.quota.ClientQuotaService;
-import com.provectus.kafka.ui.service.rbac.AccessControlService;
 import java.math.BigDecimal;
 import java.util.LinkedHashMap;
 import java.util.Map;
@@ -23,37 +23,56 @@ import reactor.core.publisher.Mono;
 @RequiredArgsConstructor
 public class ClientQuotasController extends AbstractController implements ClientQuotasApi {
 
-  private final QuotaService quotaService;
+  private final ClientQuotaService clientQuotaService;
-  private final AccessControlService accessControlService;
-  private final AuditService auditService;
 
   @Override
   public Mono<ResponseEntity<Flux<ClientQuotasDTO>>> listQuotas(String clusterName,
                                                                 ServerWebExchange exchange) {
-    return Mono.just(quotaService.list(getCluster(clusterName)).map(this::map))
+    var context = AccessContext.builder()
-        .map(ResponseEntity::ok);
+        .cluster(clusterName)
+        .operationName("listClientQuotas")
+        .clientQuotaActions(ClientQuotaAction.VIEW)
+        .build();
+
+    Mono<ResponseEntity<Flux<ClientQuotasDTO>>> operation =
+        Mono.just(clientQuotaService.list(getCluster(clusterName)).map(this::mapToDto))
+            .map(ResponseEntity::ok);
+
+    return validateAccess(context)
+        .then(operation)
+        .doOnEach(sig -> audit(context, sig));
   }
 
   @Override
   public Mono<ResponseEntity<Void>> upsertClientQuotas(String clusterName,
-                                                       Mono<ClientQuotasDTO> clientQuotasDto,
+                                                       Mono<ClientQuotasDTO> quotasDto,
                                                        ServerWebExchange exchange) {
-    return clientQuotasDto.flatMap(
+    var context = AccessContext.builder()
-        quotas ->
+        .cluster(clusterName)
-            quotaService.upsert(
+        .operationName("upsertClientQuotas")
+        .clientQuotaActions(ClientQuotaAction.EDIT)
+        .build();
+
+    Mono<ResponseEntity<Void>> operation = quotasDto.flatMap(
+        newQuotas ->
+            clientQuotaService.upsert(
                 getCluster(clusterName),
-                quotas.getUser(),
+                newQuotas.getUser(),
-                quotas.getClientId(),
+                newQuotas.getClientId(),
-                quotas.getIp(),
+                newQuotas.getIp(),
-                Optional.ofNullable(quotas.getQuotas()).orElse(Map.of())
+                Optional.ofNullable(newQuotas.getQuotas()).orElse(Map.of())
                     .entrySet()
                     .stream()
                     .collect(toMap(Map.Entry::getKey, e -> e.getValue().doubleValue()))
             )
     ).map(statusCode -> ResponseEntity.status(statusCode).build());
+
+    return validateAccess(context)
+        .then(operation)
+        .doOnEach(sig -> audit(context, sig));
   }
 
-  private ClientQuotasDTO map(ClientQuotaRecord quotaRecord) {
+  private ClientQuotasDTO mapToDto(ClientQuotaRecord quotaRecord) {
     return new ClientQuotasDTO()
         .user(quotaRecord.user())
         .clientId(quotaRecord.clientId())

kafka-ui-api/src/main/java/com/provectus/kafka/ui/model/rbac/AccessContext.java

@@ -3,6 +3,7 @@ package com.provectus.kafka.ui.model.rbac;
 import com.provectus.kafka.ui.model.rbac.permission.AclAction;
 import com.provectus.kafka.ui.model.rbac.permission.ApplicationConfigAction;
 import com.provectus.kafka.ui.model.rbac.permission.AuditAction;
+import com.provectus.kafka.ui.model.rbac.permission.ClientQuotaAction;
 import com.provectus.kafka.ui.model.rbac.permission.ClusterConfigAction;
 import com.provectus.kafka.ui.model.rbac.permission.ConnectAction;
 import com.provectus.kafka.ui.model.rbac.permission.ConsumerGroupAction;
@@ -44,6 +45,8 @@ public class AccessContext {
 
   Collection<AuditAction> auditAction;
 
+  Collection<ClientQuotaAction> clientQuotaActions;
+
   String operationName;
   Object operationParams;
 
@@ -67,6 +70,7 @@ public class AccessContext {
     private Collection<KsqlAction> ksqlActions = Collections.emptySet();
     private Collection<AclAction> aclActions = Collections.emptySet();
     private Collection<AuditAction> auditActions = Collections.emptySet();
+    private Collection<ClientQuotaAction> clientQuotaActions = Collections.emptySet();
 
     private String operationName;
     private Object operationParams;
@@ -158,6 +162,12 @@ public class AccessContext {
       return this;
     }
 
+    public AccessContextBuilder clientQuotaActions(ClientQuotaAction... actions) {
+      Assert.isTrue(actions.length > 0, "actions not present");
+      this.clientQuotaActions = List.of(actions);
+      return this;
+    }
+
     public AccessContextBuilder operationName(String operationName) {
       this.operationName = operationName;
       return this;
@@ -182,7 +192,7 @@ public class AccessContext {
           connect, connectActions,
           connector,
           schema, schemaActions,
-          ksqlActions, aclActions, auditActions,
+          ksqlActions, aclActions, auditActions, clientQuotaActions,
           operationName, operationParams);
     }
   }

kafka-ui-api/src/main/java/com/provectus/kafka/ui/model/rbac/Permission.java

@@ -3,12 +3,14 @@ package com.provectus.kafka.ui.model.rbac;
 import static com.provectus.kafka.ui.model.rbac.Resource.ACL;
 import static com.provectus.kafka.ui.model.rbac.Resource.APPLICATIONCONFIG;
 import static com.provectus.kafka.ui.model.rbac.Resource.AUDIT;
+import static com.provectus.kafka.ui.model.rbac.Resource.CLIENT_QUOTAS;
 import static com.provectus.kafka.ui.model.rbac.Resource.CLUSTERCONFIG;
 import static com.provectus.kafka.ui.model.rbac.Resource.KSQL;
 
 import com.provectus.kafka.ui.model.rbac.permission.AclAction;
 import com.provectus.kafka.ui.model.rbac.permission.ApplicationConfigAction;
 import com.provectus.kafka.ui.model.rbac.permission.AuditAction;
+import com.provectus.kafka.ui.model.rbac.permission.ClientQuotaAction;
 import com.provectus.kafka.ui.model.rbac.permission.ClusterConfigAction;
 import com.provectus.kafka.ui.model.rbac.permission.ConnectAction;
 import com.provectus.kafka.ui.model.rbac.permission.ConsumerGroupAction;
@@ -32,7 +34,7 @@ import org.springframework.util.Assert;
 public class Permission {
 
   private static final List<Resource> RBAC_ACTION_EXEMPT_LIST =
-      List.of(KSQL, CLUSTERCONFIG, APPLICATIONCONFIG, ACL, AUDIT);
+      List.of(KSQL, CLUSTERCONFIG, APPLICATIONCONFIG, ACL, AUDIT, CLIENT_QUOTAS);
 
   Resource resource;
   List<String> actions;
@@ -88,6 +90,7 @@ public class Permission {
       case KSQL -> Arrays.stream(KsqlAction.values()).map(Enum::toString).toList();
       case ACL -> Arrays.stream(AclAction.values()).map(Enum::toString).toList();
       case AUDIT -> Arrays.stream(AuditAction.values()).map(Enum::toString).toList();
+      case CLIENT_QUOTAS -> Arrays.stream(ClientQuotaAction.values()).map(Enum::toString).toList();
     };
   }
 

kafka-ui-api/src/main/java/com/provectus/kafka/ui/model/rbac/Resource.java

@@ -13,7 +13,8 @@ public enum Resource {
   CONNECT,
   KSQL,
   ACL,
-  AUDIT;
+  AUDIT,
+  CLIENT_QUOTAS;
 
   @Nullable
   public static Resource fromString(String name) {

kafka-ui-api/src/main/java/com/provectus/kafka/ui/model/rbac/permission/ClientQuotaAction.java

@@ -0,0 +1,18 @@
+package com.provectus.kafka.ui.model.rbac.permission;
+
+import java.util.Set;
+
+public enum ClientQuotaAction implements PermissibleAction {
+
+  VIEW,
+  EDIT
+
+  ;
+
+  public static final Set<ClientQuotaAction> ALTER_ACTIONS = Set.of(EDIT);
+
+  public boolean isAlter() {
+    return ALTER_ACTIONS.contains(this);
+  }
+
+}

kafka-ui-api/src/main/java/com/provectus/kafka/ui/model/rbac/permission/PermissibleAction.java

@@ -4,5 +4,5 @@ public sealed interface PermissibleAction permits
     AclAction, ApplicationConfigAction,
     ConsumerGroupAction, SchemaAction,
     ConnectAction, ClusterConfigAction,
-    KsqlAction, TopicAction, AuditAction {
+    KsqlAction, TopicAction, AuditAction, ClientQuotaAction {
 }

kafka-ui-api/src/main/java/com/provectus/kafka/ui/service/quota/ClientQuotaRecord.java

@@ -1,6 +1,7 @@
 package com.provectus.kafka.ui.service.quota;
 
 import jakarta.annotation.Nullable;
+import java.util.Comparator;
 import java.util.Map;
 import org.apache.kafka.common.quota.ClientQuotaEntity;
 
@@ -9,12 +10,17 @@ public record ClientQuotaRecord(@Nullable String user,
                                 @Nullable String ip,
                                 Map<String, Double> quotas) {
 
-  static ClientQuotaRecord create(ClientQuotaEntity entity, Map<String, Double> qoutas) {
+  static final Comparator<ClientQuotaRecord> COMPARATOR =
+      Comparator.<ClientQuotaRecord, String>comparing(r -> r.user)
+          .thenComparing(r -> r.clientId)
+          .thenComparing(r -> r.ip);
+
+  static ClientQuotaRecord create(ClientQuotaEntity entity, Map<String, Double> quotas) {
     return new ClientQuotaRecord(
         entity.entries().get(ClientQuotaEntity.USER),
         entity.entries().get(ClientQuotaEntity.CLIENT_ID),
         entity.entries().get(ClientQuotaEntity.IP),
-        qoutas
+        quotas
     );
   }
 }

kafka-ui-api/src/main/java/com/provectus/kafka/ui/service/quota/ClientQuotaService.java

@@ -24,14 +24,13 @@ import org.apache.kafka.common.quota.ClientQuotaEntity;
 import org.apache.kafka.common.quota.ClientQuotaFilter;
 import org.apache.kafka.common.quota.ClientQuotaFilterComponent;
 import org.springframework.http.HttpStatus;
-import org.springframework.http.HttpStatusCode;
 import org.springframework.stereotype.Service;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 
 @Service
 @RequiredArgsConstructor
-public class QuotaService {
+public class ClientQuotaService {
 
   private final AdminClientService adminClientService;
 
@@ -39,36 +38,29 @@ public class QuotaService {
     return adminClientService.get(cluster)
         .flatMap(ac -> ac.getClientQuotas(ClientQuotaFilter.all()))
         .flatMapIterable(map ->
-            map.entrySet().stream().map(e -> ClientQuotaRecord.create(e.getKey(), e.getValue())).toList());
+            map.entrySet().stream().map(e -> ClientQuotaRecord.create(e.getKey(), e.getValue())).toList())
+        .sort(ClientQuotaRecord.COMPARATOR);
   }
 
-  //returns 201 is new entity was created, 200 if existing was updated, 204 if it was deleted
+  //returns 201 if new entity was created, 200 if existing was updated, 204 if existing was deleted
   public Mono<HttpStatus> upsert(KafkaCluster cluster,
-                                     @Nullable String user,
+                                 @Nullable String user,
-                                     @Nullable String clientId,
+                                 @Nullable String clientId,
-                                     @Nullable String ip,
+                                 @Nullable String ip,
-                                     Map<String, Double> newQuotas) {
+                                 Map<String, Double> newQuotas) {
     ClientQuotaEntity quotaEntity = quotaEntity(user, clientId, ip);
     return adminClientService.get(cluster)
         .flatMap(ac ->
             findQuotas(ac, quotaEntity)
                 .flatMap(currentQuotas -> {
-                  Set<String> quotasToClear = Sets.difference(currentQuotas.keySet(), newQuotas.keySet());
-                  List<ClientQuotaAlteration.Op> ops = Stream.concat(
-                      quotasToClear.stream()
-                          //setting null value to clear current state
-                          .map(name -> new ClientQuotaAlteration.Op(name, null)),
-                      newQuotas.entrySet().stream()
-                          .map(e -> new ClientQuotaAlteration.Op(e.getKey(), e.getValue()))
-                  ).toList();
-
                   HttpStatus result = HttpStatus.OK; //updated
                   if (newQuotas.isEmpty()) {
                     result = HttpStatus.NO_CONTENT; //deleted
                   } else if (currentQuotas.isEmpty()) {
                     result = HttpStatus.CREATED;
                   }
-                  return ac.alterClientQuota(new ClientQuotaAlteration(quotaEntity, ops))
+                  var alteration = createAlteration(quotaEntity, currentQuotas, newQuotas);
+                  return ac.alterClientQuota(alteration)
                       .thenReturn(result);
                 })
         );
@@ -85,12 +77,26 @@ public class QuotaService {
     return new ClientQuotaEntity(id);
   }
 
-  private Mono<Map<String, Double>> findQuotas(ReactiveAdminClient ac, ClientQuotaEntity quotaEntity) {
+  private ClientQuotaAlteration createAlteration(ClientQuotaEntity quotaEntity,
-    return ac.getClientQuotas(searchFilter(quotaEntity))
+                                                 Map<String, Double> currentQuotas,
-        .map(foundRecords -> Optional.ofNullable(foundRecords.get(quotaEntity)).orElse(Map.of()));
+                                                 Map<String, Double> newQuotas) {
+    Set<String> quotasToClear = Sets.difference(currentQuotas.keySet(), newQuotas.keySet());
+    List<ClientQuotaAlteration.Op> ops = Stream.concat(
+        quotasToClear.stream()
+            .map(name -> new ClientQuotaAlteration.Op(name, null)), //setting null value to clear current state
+        newQuotas.entrySet().stream()
+            .map(e -> new ClientQuotaAlteration.Op(e.getKey(), e.getValue()))
+    ).toList();
+    return new ClientQuotaAlteration(quotaEntity, ops);
   }
 
-  private ClientQuotaFilter searchFilter(ClientQuotaEntity quotaEntity) {
+  // returns empty map if no quotas found for an entity
+  private Mono<Map<String, Double>> findQuotas(ReactiveAdminClient ac, ClientQuotaEntity quotaEntity) {
+    return ac.getClientQuotas(crateSearchFilter(quotaEntity))
+        .map(found -> Optional.ofNullable(found.get(quotaEntity)).orElse(Map.of()));
+  }
+
+  private ClientQuotaFilter crateSearchFilter(ClientQuotaEntity quotaEntity) {
     List<ClientQuotaFilterComponent> filters = new ArrayList<>();
     quotaEntity.entries().forEach((type, name) -> filters.add(ClientQuotaFilterComponent.ofEntity(type, name)));
     return ClientQuotaFilter.contains(filters);

kafka-ui-api/src/main/java/com/provectus/kafka/ui/service/rbac/AccessControlService.java

@@ -123,7 +123,8 @@ public class AccessControlService {
                   && isSchemaAccessible(context, user)
                   && isKsqlAccessible(context, user)
                   && isAclAccessible(context, user)
-                  && isAuditAccessible(context, user);
+                  && isAuditAccessible(context, user)
+                  && isClientQuotaAccessible(context, user);
 
           if (!accessGranted) {
             throw new AccessDeniedException(ACCESS_DENIED);
@@ -417,6 +418,23 @@ public class AccessControlService {
     return isAccessible(Resource.AUDIT, null, user, context, requiredActions);
   }
 
+  private boolean isClientQuotaAccessible(AccessContext context, AuthenticatedUser user) {
+    if (!rbacEnabled) {
+      return true;
+    }
+
+    if (context.getClientQuotaActions().isEmpty()) {
+      return true;
+    }
+
+    Set<String> requiredActions = context.getClientQuotaActions()
+        .stream()
+        .map(a -> a.toString().toUpperCase())
+        .collect(Collectors.toSet());
+
+    return isAccessible(Resource.CLIENT_QUOTAS, null, user, context, requiredActions);
+  }
+
   public Set<ProviderAuthorityExtractor> getOauthExtractors() {
     return oauthExtractors;
   }

kafka-ui-api/src/test/java/com/provectus/kafka/ui/service/quota/ClientQuotaServiceTest.java

@@ -0,0 +1,11 @@
+package com.provectus.kafka.ui.service.quota;
+
+import com.provectus.kafka.ui.AbstractIntegrationTest;
+import org.springframework.beans.factory.annotation.Autowired;
+
+class ClientQuotaServiceTest extends AbstractIntegrationTest {
+
+  @Autowired
+  ClientQuotaService quotaService;
+
+}

kafka-ui-contract/src/main/resources/swagger/kafka-ui-api.yaml

@@ -1936,6 +1936,9 @@ paths:
         - ClientQuotas
       summary: upsertClientQuotas
       operationId: upsertClientQuotas
+      description: |
+        - updates/creates client quota record if `quotas` field is non-empty
+        - deletes client quota record if `quotas` field is null or empty
       parameters:
         - name: clusterName
           in: path
@@ -1951,7 +1954,7 @@ paths:
         200:
           description: Existing quota updated
         201:
-          description: Quota created
+          description: New quota created
         204:
           description: Existing quota deleted