From 5bbe96ad27ea19e9025fb83acb383d6a37f6a0fe Mon Sep 17 00:00:00 2001 From: IvanBorislavovDimitrov Date: Mon, 27 Jul 2026 13:20:12 +0300 Subject: [PATCH 01/10] Introduce per-user/per-space operation rate limits Add an application-level limiter that bounds the START of MTA operations along two independent keys (per CF user and per space) using two complementary mechanisms: - a token-bucket RATE limiter (bucket4j) whose state is persisted and synchronized in PostgreSQL via a SELECT FOR UPDATE proxy manager, so the limit holds across all service instances sharing the database; and - a concurrency cap on simultaneously non-final operations, counted over the shared operation table. The limiter is gated behind a feature flag (disabled by default) and all caps are configurable via environment variables. When triggered, the operation-start endpoint returns HTTP 429 with a Retry-After header and no process is started. Developed test-first: unit tests for config, key derivation, the limiter (mocked bucket + concurrency), and the 429 wiring; a Testcontainers integration test proves the cross-instance PostgreSQL synchronization. --- .../multiapps/controller/core/Messages.java | 7 + .../core/util/ApplicationConfiguration.java | 123 ++++++++++- .../util/ApplicationConfigurationTest.java | 125 +++++++++++ .../db-changelog-2.52.0-persistence.xml | 25 +++ .../persistence/db/changelog/db-changelog.xml | 4 +- multiapps-controller-web/pom.xml | 36 ++++ .../multiapps/controller/web/Messages.java | 1 + .../api/impl/OperationsApiServiceImpl.java | 20 +- .../controller/web/util/BucketStore.java | 13 ++ .../OperationRateLimitExceededException.java | 21 ++ .../web/util/OperationRateLimitKeys.java | 39 ++++ .../web/util/OperationRateLimiter.java | 102 +++++++++ .../web/util/PostgresBucketStore.java | 33 +++ .../impl/OperationsApiServiceImplTest.java | 46 +++- .../web/util/OperationRateLimitKeysTest.java | 72 +++++++ .../OperationRateLimiterIntegrationTest.java | 174 +++++++++++++++ .../web/util/OperationRateLimiterTest.java | 201 ++++++++++++++++++ pom.xml | 33 +++ 18 files changed, 1071 insertions(+), 4 deletions(-) create mode 100644 multiapps-controller-persistence/src/main/resources/org/cloudfoundry/multiapps/controller/persistence/db/changelog/db-changelog-2.52.0-persistence.xml create mode 100644 multiapps-controller-web/src/main/java/org/cloudfoundry/multiapps/controller/web/util/BucketStore.java create mode 100644 multiapps-controller-web/src/main/java/org/cloudfoundry/multiapps/controller/web/util/OperationRateLimitExceededException.java create mode 100644 multiapps-controller-web/src/main/java/org/cloudfoundry/multiapps/controller/web/util/OperationRateLimitKeys.java create mode 100644 multiapps-controller-web/src/main/java/org/cloudfoundry/multiapps/controller/web/util/OperationRateLimiter.java create mode 100644 multiapps-controller-web/src/main/java/org/cloudfoundry/multiapps/controller/web/util/PostgresBucketStore.java create mode 100644 multiapps-controller-web/src/test/java/org/cloudfoundry/multiapps/controller/web/util/OperationRateLimitKeysTest.java create mode 100644 multiapps-controller-web/src/test/java/org/cloudfoundry/multiapps/controller/web/util/OperationRateLimiterIntegrationTest.java create mode 100644 multiapps-controller-web/src/test/java/org/cloudfoundry/multiapps/controller/web/util/OperationRateLimiterTest.java diff --git a/multiapps-controller-core/src/main/java/org/cloudfoundry/multiapps/controller/core/Messages.java b/multiapps-controller-core/src/main/java/org/cloudfoundry/multiapps/controller/core/Messages.java index d0b9eec793..16e8fada28 100644 --- a/multiapps-controller-core/src/main/java/org/cloudfoundry/multiapps/controller/core/Messages.java +++ b/multiapps-controller-core/src/main/java/org/cloudfoundry/multiapps/controller/core/Messages.java @@ -198,6 +198,13 @@ public final class Messages { public static final String THREADS_FOR_FILE_STORAGE_UPLOAD_0 = "Threads for file storage upload: {0}"; public static final String DELETED_ORPHANED_MTA_DESCRIPTORS_COUNT = "Deleted orphaned mta descriptors count: {0}"; public static final String IS_HEALTH_CHECK_ENABLED = "Is health check enabled: {0}"; + public static final String OPERATION_RATE_LIMITING_ENABLED = "Operation rate limiting enabled: {0}"; + public static final String MAX_ACTIVE_OPERATIONS_PER_SPACE = "Max active operations per space: {0}"; + public static final String MAX_ACTIVE_OPERATIONS_PER_USER = "Max active operations per user: {0}"; + public static final String OPERATION_RATE_LIMIT_PER_SPACE_CAPACITY = "Operation rate limit per space capacity: {0}"; + public static final String OPERATION_RATE_LIMIT_PER_SPACE_REFILL_PER_HOUR = "Operation rate limit per space refill per hour: {0}"; + public static final String OPERATION_RATE_LIMIT_PER_USER_CAPACITY = "Operation rate limit per user capacity: {0}"; + public static final String OPERATION_RATE_LIMIT_PER_USER_REFILL_PER_HOUR = "Operation rate limit per user refill per hour: {0}"; // Debug messages public static final String DEPLOYMENT_DESCRIPTOR = "Deployment descriptor: {0}"; diff --git a/multiapps-controller-core/src/main/java/org/cloudfoundry/multiapps/controller/core/util/ApplicationConfiguration.java b/multiapps-controller-core/src/main/java/org/cloudfoundry/multiapps/controller/core/util/ApplicationConfiguration.java index 143f7972e5..d460e88a9a 100644 --- a/multiapps-controller-core/src/main/java/org/cloudfoundry/multiapps/controller/core/util/ApplicationConfiguration.java +++ b/multiapps-controller-core/src/main/java/org/cloudfoundry/multiapps/controller/core/util/ApplicationConfiguration.java @@ -100,6 +100,13 @@ public class ApplicationConfiguration { static final String CFG_THREADS_FOR_FILE_UPLOAD_TO_CONTROLLER = "THREADS_FOR_FILE_UPLOAD_TO_CONTROLLER"; static final String CFG_THREADS_FOR_FILE_STORAGE_UPLOAD = "THREADS_FOR_FILE_STORAGE_UPLOAD"; static final String CFG_IS_HEALTH_CHECK_ENABLED = "IS_HEALTH_CHECK_ENABLED"; + static final String CFG_OPERATION_RATE_LIMITING_ENABLED = "OPERATION_RATE_LIMITING_ENABLED"; + static final String CFG_MAX_ACTIVE_OPERATIONS_PER_SPACE = "MAX_ACTIVE_OPERATIONS_PER_SPACE"; + static final String CFG_MAX_ACTIVE_OPERATIONS_PER_USER = "MAX_ACTIVE_OPERATIONS_PER_USER"; + static final String CFG_OPERATION_RATE_LIMIT_PER_SPACE_CAPACITY = "OP_RATE_LIMIT_PER_SPACE_CAPACITY"; + static final String CFG_OPERATION_RATE_LIMIT_PER_SPACE_REFILL_PER_HOUR = "OP_RATE_LIMIT_PER_SPACE_REFILL_PER_HOUR"; + static final String CFG_OPERATION_RATE_LIMIT_PER_USER_CAPACITY = "OP_RATE_LIMIT_PER_USER_CAPACITY"; + static final String CFG_OPERATION_RATE_LIMIT_PER_USER_REFILL_PER_HOUR = "OP_RATE_LIMIT_PER_USER_REFILL_PER_HOUR"; private static final List VCAP_APPLICATION_URIS_KEYS = List.of("full_application_uris", "application_uris", "uris"); @@ -158,6 +165,13 @@ public class ApplicationConfiguration { public static final int DEFAULT_THREADS_FOR_FILE_UPLOAD_TO_CONTROLLER = 6; public static final int DEFAULT_THREADS_FOR_FILE_STORAGE_UPLOAD = 7; public static final boolean DEFAULT_IS_HEALTH_CHECK_ENABLED = false; + public static final boolean DEFAULT_OPERATION_RATE_LIMITING_ENABLED = false; + public static final int DEFAULT_MAX_ACTIVE_OPERATIONS_PER_SPACE = 500; + public static final int DEFAULT_MAX_ACTIVE_OPERATIONS_PER_USER = 200; + public static final int DEFAULT_OPERATION_RATE_LIMIT_PER_SPACE_CAPACITY = 300; + public static final int DEFAULT_OPERATION_RATE_LIMIT_PER_SPACE_REFILL_PER_HOUR = 800; + public static final int DEFAULT_OPERATION_RATE_LIMIT_PER_USER_CAPACITY = 150; + public static final int DEFAULT_OPERATION_RATE_LIMIT_PER_USER_REFILL_PER_HOUR = 300; protected final Environment environment; @@ -217,6 +231,13 @@ public class ApplicationConfiguration { private Integer threadsForFileStorageUpload; private Boolean isHealthCheckEnabled; private Set objectStoreRegions; + private Boolean operationRateLimitingEnabled; + private Integer maxActiveOperationsPerSpace; + private Integer maxActiveOperationsPerUser; + private Integer operationRateLimitPerSpaceCapacity; + private Integer operationRateLimitPerSpaceRefillPerHour; + private Integer operationRateLimitPerUserCapacity; + private Integer operationRateLimitPerUserRefillPerHour; public ApplicationConfiguration() { this(new Environment()); @@ -286,7 +307,10 @@ private Set getNotSensitiveConfigVariables() { CFG_FLOWABLE_JOB_EXECUTOR_QUEUE_CAPACITY, CFG_CONTROLLER_CLIENT_CONNECTION_POOL_SIZE, CFG_CONTROLLER_CLIENT_THREAD_POOL_SIZE, CFG_CONTROLLER_CLIENT_RESPONSE_TIMEOUT, CFG_DB_TRANSACTION_TIMEOUT_IN_SECONDS, - CFG_SNAKEYAML_MAX_ALIASES_FOR_COLLECTIONS, CFG_SERVICE_HANDLING_MAX_PARALLEL_THREADS); + CFG_SNAKEYAML_MAX_ALIASES_FOR_COLLECTIONS, CFG_SERVICE_HANDLING_MAX_PARALLEL_THREADS, + CFG_OPERATION_RATE_LIMITING_ENABLED, CFG_MAX_ACTIVE_OPERATIONS_PER_SPACE, CFG_MAX_ACTIVE_OPERATIONS_PER_USER, + CFG_OPERATION_RATE_LIMIT_PER_SPACE_CAPACITY, CFG_OPERATION_RATE_LIMIT_PER_SPACE_REFILL_PER_HOUR, + CFG_OPERATION_RATE_LIMIT_PER_USER_CAPACITY, CFG_OPERATION_RATE_LIMIT_PER_USER_REFILL_PER_HOUR); } public URL getControllerUrl() { @@ -668,6 +692,55 @@ public boolean isHealthCheckEnabled() { return isHealthCheckEnabled; } + public boolean isOperationRateLimitingEnabled() { + if (operationRateLimitingEnabled == null) { + operationRateLimitingEnabled = isOperationRateLimitingEnabledThroughEnvironment(); + } + return operationRateLimitingEnabled; + } + + public Integer getMaxActiveOperationsPerSpace() { + if (maxActiveOperationsPerSpace == null) { + maxActiveOperationsPerSpace = getMaxActiveOperationsPerSpaceFromEnvironment(); + } + return maxActiveOperationsPerSpace; + } + + public Integer getMaxActiveOperationsPerUser() { + if (maxActiveOperationsPerUser == null) { + maxActiveOperationsPerUser = getMaxActiveOperationsPerUserFromEnvironment(); + } + return maxActiveOperationsPerUser; + } + + public Integer getOperationRateLimitPerSpaceCapacity() { + if (operationRateLimitPerSpaceCapacity == null) { + operationRateLimitPerSpaceCapacity = getOperationRateLimitPerSpaceCapacityFromEnvironment(); + } + return operationRateLimitPerSpaceCapacity; + } + + public Integer getOperationRateLimitPerSpaceRefillPerHour() { + if (operationRateLimitPerSpaceRefillPerHour == null) { + operationRateLimitPerSpaceRefillPerHour = getOperationRateLimitPerSpaceRefillPerHourFromEnvironment(); + } + return operationRateLimitPerSpaceRefillPerHour; + } + + public Integer getOperationRateLimitPerUserCapacity() { + if (operationRateLimitPerUserCapacity == null) { + operationRateLimitPerUserCapacity = getOperationRateLimitPerUserCapacityFromEnvironment(); + } + return operationRateLimitPerUserCapacity; + } + + public Integer getOperationRateLimitPerUserRefillPerHour() { + if (operationRateLimitPerUserRefillPerHour == null) { + operationRateLimitPerUserRefillPerHour = getOperationRateLimitPerUserRefillPerHourFromEnvironment(); + } + return operationRateLimitPerUserRefillPerHour; + } + private URL getControllerUrlFromEnvironment() { String controllerUrlString = environment.getString("CF_API"); if (StringUtils.isEmpty(controllerUrlString)) { @@ -1098,6 +1171,54 @@ public boolean isHealthCheckEnabledFromEnvironment() { return value; } + private Boolean isOperationRateLimitingEnabledThroughEnvironment() { + Boolean value = environment.getBoolean(CFG_OPERATION_RATE_LIMITING_ENABLED, DEFAULT_OPERATION_RATE_LIMITING_ENABLED); + logEnvironmentVariable(CFG_OPERATION_RATE_LIMITING_ENABLED, Messages.OPERATION_RATE_LIMITING_ENABLED, value); + return value; + } + + private Integer getMaxActiveOperationsPerSpaceFromEnvironment() { + Integer value = environment.getPositiveInteger(CFG_MAX_ACTIVE_OPERATIONS_PER_SPACE, DEFAULT_MAX_ACTIVE_OPERATIONS_PER_SPACE); + logEnvironmentVariable(CFG_MAX_ACTIVE_OPERATIONS_PER_SPACE, Messages.MAX_ACTIVE_OPERATIONS_PER_SPACE, value); + return value; + } + + private Integer getMaxActiveOperationsPerUserFromEnvironment() { + Integer value = environment.getPositiveInteger(CFG_MAX_ACTIVE_OPERATIONS_PER_USER, DEFAULT_MAX_ACTIVE_OPERATIONS_PER_USER); + logEnvironmentVariable(CFG_MAX_ACTIVE_OPERATIONS_PER_USER, Messages.MAX_ACTIVE_OPERATIONS_PER_USER, value); + return value; + } + + private Integer getOperationRateLimitPerSpaceCapacityFromEnvironment() { + Integer value = environment.getPositiveInteger(CFG_OPERATION_RATE_LIMIT_PER_SPACE_CAPACITY, + DEFAULT_OPERATION_RATE_LIMIT_PER_SPACE_CAPACITY); + logEnvironmentVariable(CFG_OPERATION_RATE_LIMIT_PER_SPACE_CAPACITY, Messages.OPERATION_RATE_LIMIT_PER_SPACE_CAPACITY, value); + return value; + } + + private Integer getOperationRateLimitPerSpaceRefillPerHourFromEnvironment() { + Integer value = environment.getPositiveInteger(CFG_OPERATION_RATE_LIMIT_PER_SPACE_REFILL_PER_HOUR, + DEFAULT_OPERATION_RATE_LIMIT_PER_SPACE_REFILL_PER_HOUR); + logEnvironmentVariable(CFG_OPERATION_RATE_LIMIT_PER_SPACE_REFILL_PER_HOUR, + Messages.OPERATION_RATE_LIMIT_PER_SPACE_REFILL_PER_HOUR, value); + return value; + } + + private Integer getOperationRateLimitPerUserCapacityFromEnvironment() { + Integer value = environment.getPositiveInteger(CFG_OPERATION_RATE_LIMIT_PER_USER_CAPACITY, + DEFAULT_OPERATION_RATE_LIMIT_PER_USER_CAPACITY); + logEnvironmentVariable(CFG_OPERATION_RATE_LIMIT_PER_USER_CAPACITY, Messages.OPERATION_RATE_LIMIT_PER_USER_CAPACITY, value); + return value; + } + + private Integer getOperationRateLimitPerUserRefillPerHourFromEnvironment() { + Integer value = environment.getPositiveInteger(CFG_OPERATION_RATE_LIMIT_PER_USER_REFILL_PER_HOUR, + DEFAULT_OPERATION_RATE_LIMIT_PER_USER_REFILL_PER_HOUR); + logEnvironmentVariable(CFG_OPERATION_RATE_LIMIT_PER_USER_REFILL_PER_HOUR, + Messages.OPERATION_RATE_LIMIT_PER_USER_REFILL_PER_HOUR, value); + return value; + } + public Boolean isInternalEnvironment() { return environment.getBoolean(SAP_INTERNAL_DELIVERY, DEFAULT_SAP_INTERNAL_DELIVERY); } diff --git a/multiapps-controller-core/src/test/java/org/cloudfoundry/multiapps/controller/core/util/ApplicationConfigurationTest.java b/multiapps-controller-core/src/test/java/org/cloudfoundry/multiapps/controller/core/util/ApplicationConfigurationTest.java index e968387926..8404349eac 100644 --- a/multiapps-controller-core/src/test/java/org/cloudfoundry/multiapps/controller/core/util/ApplicationConfigurationTest.java +++ b/multiapps-controller-core/src/test/java/org/cloudfoundry/multiapps/controller/core/util/ApplicationConfigurationTest.java @@ -464,6 +464,131 @@ void testGetSpringSchedulerTaskExecutorThreads() { Assertions.assertEquals(executorThreads, configuration.getSpringSchedulerTaskExecutorThreads()); } + @Test + void testIsOperationRateLimitingEnabled() { + Mockito.when(environment.getBoolean(ApplicationConfiguration.CFG_OPERATION_RATE_LIMITING_ENABLED, + ApplicationConfiguration.DEFAULT_OPERATION_RATE_LIMITING_ENABLED)) + .thenReturn(ApplicationConfiguration.DEFAULT_OPERATION_RATE_LIMITING_ENABLED); + assertEquals(ApplicationConfiguration.DEFAULT_OPERATION_RATE_LIMITING_ENABLED, + configuration.isOperationRateLimitingEnabled()); + } + + @Test + void testIsOperationRateLimitingEnabledWithCustomValue() { + Mockito.when(environment.getBoolean(ApplicationConfiguration.CFG_OPERATION_RATE_LIMITING_ENABLED, + ApplicationConfiguration.DEFAULT_OPERATION_RATE_LIMITING_ENABLED)) + .thenReturn(true); + assertTrue(configuration.isOperationRateLimitingEnabled()); + } + + @Test + void testGetMaxActiveOperationsPerSpace() { + Mockito.when(environment.getPositiveInteger(ApplicationConfiguration.CFG_MAX_ACTIVE_OPERATIONS_PER_SPACE, + ApplicationConfiguration.DEFAULT_MAX_ACTIVE_OPERATIONS_PER_SPACE)) + .thenReturn(ApplicationConfiguration.DEFAULT_MAX_ACTIVE_OPERATIONS_PER_SPACE); + assertEquals(ApplicationConfiguration.DEFAULT_MAX_ACTIVE_OPERATIONS_PER_SPACE, + configuration.getMaxActiveOperationsPerSpace()); + } + + @Test + void testGetMaxActiveOperationsPerSpaceWithCustomValue() { + int customValue = 750; + Mockito.when(environment.getPositiveInteger(ApplicationConfiguration.CFG_MAX_ACTIVE_OPERATIONS_PER_SPACE, + ApplicationConfiguration.DEFAULT_MAX_ACTIVE_OPERATIONS_PER_SPACE)) + .thenReturn(customValue); + assertEquals(customValue, configuration.getMaxActiveOperationsPerSpace()); + } + + @Test + void testGetMaxActiveOperationsPerUser() { + Mockito.when(environment.getPositiveInteger(ApplicationConfiguration.CFG_MAX_ACTIVE_OPERATIONS_PER_USER, + ApplicationConfiguration.DEFAULT_MAX_ACTIVE_OPERATIONS_PER_USER)) + .thenReturn(ApplicationConfiguration.DEFAULT_MAX_ACTIVE_OPERATIONS_PER_USER); + assertEquals(ApplicationConfiguration.DEFAULT_MAX_ACTIVE_OPERATIONS_PER_USER, + configuration.getMaxActiveOperationsPerUser()); + } + + @Test + void testGetMaxActiveOperationsPerUserWithCustomValue() { + int customValue = 250; + Mockito.when(environment.getPositiveInteger(ApplicationConfiguration.CFG_MAX_ACTIVE_OPERATIONS_PER_USER, + ApplicationConfiguration.DEFAULT_MAX_ACTIVE_OPERATIONS_PER_USER)) + .thenReturn(customValue); + assertEquals(customValue, configuration.getMaxActiveOperationsPerUser()); + } + + @Test + void testGetOperationRateLimitPerSpaceCapacity() { + Mockito.when(environment.getPositiveInteger(ApplicationConfiguration.CFG_OPERATION_RATE_LIMIT_PER_SPACE_CAPACITY, + ApplicationConfiguration.DEFAULT_OPERATION_RATE_LIMIT_PER_SPACE_CAPACITY)) + .thenReturn(ApplicationConfiguration.DEFAULT_OPERATION_RATE_LIMIT_PER_SPACE_CAPACITY); + assertEquals(ApplicationConfiguration.DEFAULT_OPERATION_RATE_LIMIT_PER_SPACE_CAPACITY, + configuration.getOperationRateLimitPerSpaceCapacity()); + } + + @Test + void testGetOperationRateLimitPerSpaceCapacityWithCustomValue() { + int customValue = 400; + Mockito.when(environment.getPositiveInteger(ApplicationConfiguration.CFG_OPERATION_RATE_LIMIT_PER_SPACE_CAPACITY, + ApplicationConfiguration.DEFAULT_OPERATION_RATE_LIMIT_PER_SPACE_CAPACITY)) + .thenReturn(customValue); + assertEquals(customValue, configuration.getOperationRateLimitPerSpaceCapacity()); + } + + @Test + void testGetOperationRateLimitPerSpaceRefillPerHour() { + Mockito.when(environment.getPositiveInteger(ApplicationConfiguration.CFG_OPERATION_RATE_LIMIT_PER_SPACE_REFILL_PER_HOUR, + ApplicationConfiguration.DEFAULT_OPERATION_RATE_LIMIT_PER_SPACE_REFILL_PER_HOUR)) + .thenReturn(ApplicationConfiguration.DEFAULT_OPERATION_RATE_LIMIT_PER_SPACE_REFILL_PER_HOUR); + assertEquals(ApplicationConfiguration.DEFAULT_OPERATION_RATE_LIMIT_PER_SPACE_REFILL_PER_HOUR, + configuration.getOperationRateLimitPerSpaceRefillPerHour()); + } + + @Test + void testGetOperationRateLimitPerSpaceRefillPerHourWithCustomValue() { + int customValue = 1000; + Mockito.when(environment.getPositiveInteger(ApplicationConfiguration.CFG_OPERATION_RATE_LIMIT_PER_SPACE_REFILL_PER_HOUR, + ApplicationConfiguration.DEFAULT_OPERATION_RATE_LIMIT_PER_SPACE_REFILL_PER_HOUR)) + .thenReturn(customValue); + assertEquals(customValue, configuration.getOperationRateLimitPerSpaceRefillPerHour()); + } + + @Test + void testGetOperationRateLimitPerUserCapacity() { + Mockito.when(environment.getPositiveInteger(ApplicationConfiguration.CFG_OPERATION_RATE_LIMIT_PER_USER_CAPACITY, + ApplicationConfiguration.DEFAULT_OPERATION_RATE_LIMIT_PER_USER_CAPACITY)) + .thenReturn(ApplicationConfiguration.DEFAULT_OPERATION_RATE_LIMIT_PER_USER_CAPACITY); + assertEquals(ApplicationConfiguration.DEFAULT_OPERATION_RATE_LIMIT_PER_USER_CAPACITY, + configuration.getOperationRateLimitPerUserCapacity()); + } + + @Test + void testGetOperationRateLimitPerUserCapacityWithCustomValue() { + int customValue = 200; + Mockito.when(environment.getPositiveInteger(ApplicationConfiguration.CFG_OPERATION_RATE_LIMIT_PER_USER_CAPACITY, + ApplicationConfiguration.DEFAULT_OPERATION_RATE_LIMIT_PER_USER_CAPACITY)) + .thenReturn(customValue); + assertEquals(customValue, configuration.getOperationRateLimitPerUserCapacity()); + } + + @Test + void testGetOperationRateLimitPerUserRefillPerHour() { + Mockito.when(environment.getPositiveInteger(ApplicationConfiguration.CFG_OPERATION_RATE_LIMIT_PER_USER_REFILL_PER_HOUR, + ApplicationConfiguration.DEFAULT_OPERATION_RATE_LIMIT_PER_USER_REFILL_PER_HOUR)) + .thenReturn(ApplicationConfiguration.DEFAULT_OPERATION_RATE_LIMIT_PER_USER_REFILL_PER_HOUR); + assertEquals(ApplicationConfiguration.DEFAULT_OPERATION_RATE_LIMIT_PER_USER_REFILL_PER_HOUR, + configuration.getOperationRateLimitPerUserRefillPerHour()); + } + + @Test + void testGetOperationRateLimitPerUserRefillPerHourWithCustomValue() { + int customValue = 500; + Mockito.when(environment.getPositiveInteger(ApplicationConfiguration.CFG_OPERATION_RATE_LIMIT_PER_USER_REFILL_PER_HOUR, + ApplicationConfiguration.DEFAULT_OPERATION_RATE_LIMIT_PER_USER_REFILL_PER_HOUR)) + .thenReturn(customValue); + assertEquals(customValue, configuration.getOperationRateLimitPerUserRefillPerHour()); + } + @Test void testGetFilteredEnv() { Map filteredEnvironment = new HashMap<>(); diff --git a/multiapps-controller-persistence/src/main/resources/org/cloudfoundry/multiapps/controller/persistence/db/changelog/db-changelog-2.52.0-persistence.xml b/multiapps-controller-persistence/src/main/resources/org/cloudfoundry/multiapps/controller/persistence/db/changelog/db-changelog-2.52.0-persistence.xml new file mode 100644 index 0000000000..08f20ca848 --- /dev/null +++ b/multiapps-controller-persistence/src/main/resources/org/cloudfoundry/multiapps/controller/persistence/db/changelog/db-changelog-2.52.0-persistence.xml @@ -0,0 +1,25 @@ + + + + + + + + + + + + + + + + + + + + + + diff --git a/multiapps-controller-persistence/src/main/resources/org/cloudfoundry/multiapps/controller/persistence/db/changelog/db-changelog.xml b/multiapps-controller-persistence/src/main/resources/org/cloudfoundry/multiapps/controller/persistence/db/changelog/db-changelog.xml index 0592d75f42..de059351b2 100644 --- a/multiapps-controller-persistence/src/main/resources/org/cloudfoundry/multiapps/controller/persistence/db/changelog/db-changelog.xml +++ b/multiapps-controller-persistence/src/main/resources/org/cloudfoundry/multiapps/controller/persistence/db/changelog/db-changelog.xml @@ -47,5 +47,7 @@ - + + diff --git a/multiapps-controller-web/pom.xml b/multiapps-controller-web/pom.xml index 1c377426ed..e25e1e1303 100644 --- a/multiapps-controller-web/pom.xml +++ b/multiapps-controller-web/pom.xml @@ -82,6 +82,23 @@ + + org.apache.maven.plugins + maven-failsafe-plugin + + + + integration-test + verify + + + + + + **/*IntegrationTest + + + @@ -151,6 +168,10 @@ io.github.resilience4j resilience4j-ratelimiter + + com.bucket4j + bucket4j_jdk17-postgresql + org.apache.jclouds.common googlecloud @@ -215,5 +236,20 @@ software.amazon.awssdk url-connection-client + + org.testcontainers + postgresql + test + + + org.testcontainers + junit-jupiter + test + + + org.postgresql + postgresql + test + diff --git a/multiapps-controller-web/src/main/java/org/cloudfoundry/multiapps/controller/web/Messages.java b/multiapps-controller-web/src/main/java/org/cloudfoundry/multiapps/controller/web/Messages.java index 2be0acf27a..d38680e7f1 100644 --- a/multiapps-controller-web/src/main/java/org/cloudfoundry/multiapps/controller/web/Messages.java +++ b/multiapps-controller-web/src/main/java/org/cloudfoundry/multiapps/controller/web/Messages.java @@ -87,6 +87,7 @@ public final class Messages { public static final String ASYNC_UPLOAD_JOB_EXISTS = "Async upload job for URL {} exists: {}"; public static final String CREATING_ASYNC_UPLOAD_JOB = "Creating async upload job for URL {} with ID: {}"; public static final String ASYNC_UPLOAD_JOB_REJECTED = "Async upload job with space guid: {}, namespace: {}, URL: {} rejected."; + public static final String OPERATION_START_RATE_LIMITED = "Start of operation in space {} rejected due to rate limiting: {}"; public static final String STARTING_DOWNLOAD_OF_MTAR_WITH_JOB_ID = "Starting download of MTAR from remote endpoint: {}. Job id: {}"; public static final String UPLOADED_MTAR_FROM_REMOTE_ENDPOINT_AND_JOB_ID = "Uploaded MTAR from remote endpoint {}. Job id: {} in {} ms"; public static final String ASYNC_UPLOAD_JOB_FINISHED = "Async upload job {} finished"; diff --git a/multiapps-controller-web/src/main/java/org/cloudfoundry/multiapps/controller/web/api/impl/OperationsApiServiceImpl.java b/multiapps-controller-web/src/main/java/org/cloudfoundry/multiapps/controller/web/api/impl/OperationsApiServiceImpl.java index f0752061fc..eb92cd01da 100644 --- a/multiapps-controller-web/src/main/java/org/cloudfoundry/multiapps/controller/web/api/impl/OperationsApiServiceImpl.java +++ b/multiapps-controller-web/src/main/java/org/cloudfoundry/multiapps/controller/web/api/impl/OperationsApiServiceImpl.java @@ -56,10 +56,13 @@ import org.cloudfoundry.multiapps.controller.web.Constants; import org.cloudfoundry.multiapps.controller.web.Messages; import org.cloudfoundry.multiapps.controller.web.monitoring.ApiUsageLogger; +import org.cloudfoundry.multiapps.controller.web.util.OperationRateLimitExceededException; +import org.cloudfoundry.multiapps.controller.web.util.OperationRateLimiter; import org.cloudfoundry.multiapps.controller.web.util.SecurityContextUtil; import org.flowable.engine.runtime.ProcessInstance; import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import org.springframework.http.HttpHeaders; import org.springframework.http.HttpStatus; import org.springframework.http.ResponseEntity; import org.springframework.web.server.ResponseStatusException; @@ -84,6 +87,7 @@ public class OperationsApiServiceImpl implements OperationsApiService { private final OperationsApiServiceAuditLog operationsApiServiceAuditLog; private final ApiUsageLogger apiUsageLogger; private final HttpServletRequest httpServletRequest; + private final OperationRateLimiter operationRateLimiter; @Inject public OperationsApiServiceImpl(CloudControllerClientFactory clientFactory, TokenService tokenService, @@ -93,7 +97,8 @@ public OperationsApiServiceImpl(CloudControllerClientFactory clientFactory, Toke OperationsHelper operationsHelper, ProgressMessageService progressMessageService, ProcessActionRegistry processActionRegistry, OperationsApiServiceAuditLog operationsApiServiceAuditLog, - ApiUsageLogger apiUsageLogger, HttpServletRequest httpServletRequest) { + ApiUsageLogger apiUsageLogger, HttpServletRequest httpServletRequest, + OperationRateLimiter operationRateLimiter) { this.clientFactory = clientFactory; this.tokenService = tokenService; this.operationService = operationService; @@ -106,6 +111,7 @@ public OperationsApiServiceImpl(CloudControllerClientFactory clientFactory, Toke this.operationsApiServiceAuditLog = operationsApiServiceAuditLog; this.apiUsageLogger = apiUsageLogger; this.httpServletRequest = httpServletRequest; + this.operationRateLimiter = operationRateLimiter; } @Override @@ -174,6 +180,11 @@ public ResponseEntity startOperation(String spaceGuid, Operation oper operation.getNamespace(), httpServletRequest); operationsApiServiceAuditLog.logStartOperation(SecurityContextUtil.getUsername(), spaceGuid, operation); UserInfo authenticatedUser = getAuthenticatedUser(); + try { + operationRateLimiter.checkStartAllowed(authenticatedUser.getName(), spaceGuid); + } catch (OperationRateLimitExceededException e) { + return buildRateLimitExceededResponse(spaceGuid, e); + } String processDefinitionKey = operationsHelper.getProcessDefinitionKey(operation); Set predefinedParameters = operationMetadataMapper.getOperationMetadata(operation.getProcessType()) .getParameters(); @@ -188,6 +199,13 @@ public ResponseEntity startOperation(String spaceGuid, Operation oper .build(); } + private ResponseEntity buildRateLimitExceededResponse(String spaceGuid, OperationRateLimitExceededException e) { + LOGGER.debug(Messages.OPERATION_START_RATE_LIMITED, spaceGuid, e.getMessage()); + return ResponseEntity.status(HttpStatus.TOO_MANY_REQUESTS) + .header(HttpHeaders.RETRY_AFTER, String.valueOf(e.getRetryAfterSeconds())) + .build(); + } + protected void logStartOperation(String processInstanceId, UserInfo authenticatedUser) { LOGGER.info(MessageFormat.format(Messages.STARTED_OPERATION_0_BY_USER_1_AND_ORIGIN_OF_2, processInstanceId, authenticatedUser.getId(), authenticatedUser.getToken() diff --git a/multiapps-controller-web/src/main/java/org/cloudfoundry/multiapps/controller/web/util/BucketStore.java b/multiapps-controller-web/src/main/java/org/cloudfoundry/multiapps/controller/web/util/BucketStore.java new file mode 100644 index 0000000000..ad6046614d --- /dev/null +++ b/multiapps-controller-web/src/main/java/org/cloudfoundry/multiapps/controller/web/util/BucketStore.java @@ -0,0 +1,13 @@ +package org.cloudfoundry.multiapps.controller.web.util; + +import io.github.bucket4j.Bucket; +import io.github.bucket4j.BucketConfiguration; + +/** + * Resolves distributed token buckets by key. Abstracting this behind an interface keeps the concrete bucket4j proxy manager (and its backing + * data store) out of the rate limiter, so unit tests can supply mocked buckets without touching a real database. + */ +public interface BucketStore { + + Bucket getBucket(long key, BucketConfiguration configuration); +} diff --git a/multiapps-controller-web/src/main/java/org/cloudfoundry/multiapps/controller/web/util/OperationRateLimitExceededException.java b/multiapps-controller-web/src/main/java/org/cloudfoundry/multiapps/controller/web/util/OperationRateLimitExceededException.java new file mode 100644 index 0000000000..8084c40ccb --- /dev/null +++ b/multiapps-controller-web/src/main/java/org/cloudfoundry/multiapps/controller/web/util/OperationRateLimitExceededException.java @@ -0,0 +1,21 @@ +package org.cloudfoundry.multiapps.controller.web.util; + +/** + * Thrown when an operation cannot be started because a rate limit has been reached. Carries the number of seconds after which the caller may + * retry, which is mappable to an HTTP 429 {@code Retry-After} response header. + */ +public class OperationRateLimitExceededException extends RuntimeException { + + private static final long serialVersionUID = 1L; + + private final long retryAfterSeconds; + + public OperationRateLimitExceededException(String message, long retryAfterSeconds) { + super(message); + this.retryAfterSeconds = retryAfterSeconds; + } + + public long getRetryAfterSeconds() { + return retryAfterSeconds; + } +} diff --git a/multiapps-controller-web/src/main/java/org/cloudfoundry/multiapps/controller/web/util/OperationRateLimitKeys.java b/multiapps-controller-web/src/main/java/org/cloudfoundry/multiapps/controller/web/util/OperationRateLimitKeys.java new file mode 100644 index 0000000000..620c6cd68c --- /dev/null +++ b/multiapps-controller-web/src/main/java/org/cloudfoundry/multiapps/controller/web/util/OperationRateLimitKeys.java @@ -0,0 +1,39 @@ +package org.cloudfoundry.multiapps.controller.web.util; + +import static java.nio.charset.StandardCharsets.UTF_8; + +import com.google.common.hash.HashFunction; +import com.google.common.hash.Hashing; +import com.google.common.primitives.Longs; + +/** + * Derives stable {@code long} bucket keys for operation rate limiting. + *

+ * Keys are computed from the SHA-256 digest of a namespaced input string and are therefore deterministic across restarts and JVMs. The + * space and user namespaces are disjoint by construction, so a space key can never collide with a user key. Truncating the 256-bit digest + * to its first 64 bits keeps the collision probability negligible for the number of distinct spaces and users a single landscape handles. + */ +public final class OperationRateLimitKeys { + + private static final String SPACE_NAMESPACE_PREFIX = "space:"; + private static final String USER_NAMESPACE_PREFIX = "user:"; + private static final String SEGMENT_SEPARATOR = ":"; + private static final HashFunction HASH_FUNCTION = Hashing.sha256(); + + private OperationRateLimitKeys() { + } + + public static long spaceKey(String spaceGuid) { + return hashToLong(SPACE_NAMESPACE_PREFIX + spaceGuid); + } + + public static long userKey(String spaceGuid, String user) { + return hashToLong(USER_NAMESPACE_PREFIX + spaceGuid + SEGMENT_SEPARATOR + user); + } + + private static long hashToLong(String input) { + byte[] digest = HASH_FUNCTION.hashString(input, UTF_8) + .asBytes(); + return Longs.fromBytes(digest[0], digest[1], digest[2], digest[3], digest[4], digest[5], digest[6], digest[7]); + } +} diff --git a/multiapps-controller-web/src/main/java/org/cloudfoundry/multiapps/controller/web/util/OperationRateLimiter.java b/multiapps-controller-web/src/main/java/org/cloudfoundry/multiapps/controller/web/util/OperationRateLimiter.java new file mode 100644 index 0000000000..7576767ce1 --- /dev/null +++ b/multiapps-controller-web/src/main/java/org/cloudfoundry/multiapps/controller/web/util/OperationRateLimiter.java @@ -0,0 +1,102 @@ +package org.cloudfoundry.multiapps.controller.web.util; + +import java.time.Duration; +import java.util.concurrent.TimeUnit; + +import jakarta.inject.Named; + +import org.cloudfoundry.multiapps.controller.core.util.ApplicationConfiguration; +import org.cloudfoundry.multiapps.controller.persistence.services.OperationService; + +import io.github.bucket4j.Bandwidth; +import io.github.bucket4j.Bucket; +import io.github.bucket4j.BucketConfiguration; +import io.github.bucket4j.ConsumptionProbe; + +/** + * Guards the start of MTA operations against per-space and per-user rate limits. Each start attempt consumes a single token from both the + * space bucket and the user bucket; if either is exhausted an {@link OperationRateLimitExceededException} is raised. + */ +@Named +public class OperationRateLimiter { + + private static final Duration REFILL_PERIOD = Duration.ofHours(1); + private static final long TOKENS_PER_OPERATION = 1; + private static final long NO_RETRY_AFTER_SECONDS = 0; + + private final ApplicationConfiguration applicationConfiguration; + private final OperationService operationService; + private final BucketStore bucketStore; + + public OperationRateLimiter(ApplicationConfiguration applicationConfiguration, OperationService operationService, + BucketStore bucketStore) { + this.applicationConfiguration = applicationConfiguration; + this.operationService = operationService; + this.bucketStore = bucketStore; + } + + public void checkStartAllowed(String user, String spaceGuid) { + if (!applicationConfiguration.isOperationRateLimitingEnabled()) { + return; + } + checkActiveOperationCaps(user, spaceGuid); + checkTokenBuckets(user, spaceGuid); + } + + private void checkActiveOperationCaps(String user, String spaceGuid) { + int activeOperationsPerSpace = operationService.createQuery() + .spaceId(spaceGuid) + .inNonFinalState() + .list() + .size(); + if (activeOperationsPerSpace >= applicationConfiguration.getMaxActiveOperationsPerSpace()) { + throw new OperationRateLimitExceededException("Too many active operations in space", NO_RETRY_AFTER_SECONDS); + } + int activeOperationsPerUser = operationService.createQuery() + .user(user) + .spaceId(spaceGuid) + .inNonFinalState() + .list() + .size(); + if (activeOperationsPerUser >= applicationConfiguration.getMaxActiveOperationsPerUser()) { + throw new OperationRateLimitExceededException("Too many active operations for user", NO_RETRY_AFTER_SECONDS); + } + } + + private void checkTokenBuckets(String user, String spaceGuid) { + consumeSpaceToken(spaceGuid); + consumeUserToken(spaceGuid, user); + } + + private void consumeSpaceToken(String spaceGuid) { + BucketConfiguration configuration = buildBucketConfiguration(applicationConfiguration.getOperationRateLimitPerSpaceCapacity(), + applicationConfiguration.getOperationRateLimitPerSpaceRefillPerHour()); + Bucket bucket = bucketStore.getBucket(OperationRateLimitKeys.spaceKey(spaceGuid), configuration); + consumeToken(bucket); + } + + private void consumeUserToken(String spaceGuid, String user) { + BucketConfiguration configuration = buildBucketConfiguration(applicationConfiguration.getOperationRateLimitPerUserCapacity(), + applicationConfiguration.getOperationRateLimitPerUserRefillPerHour()); + Bucket bucket = bucketStore.getBucket(OperationRateLimitKeys.userKey(spaceGuid, user), configuration); + consumeToken(bucket); + } + + private BucketConfiguration buildBucketConfiguration(int capacity, int refillTokensPerHour) { + Bandwidth bandwidth = Bandwidth.builder() + .capacity(capacity) + .refillGreedy(refillTokensPerHour, REFILL_PERIOD) + .build(); + return BucketConfiguration.builder() + .addLimit(bandwidth) + .build(); + } + + private void consumeToken(Bucket bucket) { + ConsumptionProbe probe = bucket.tryConsumeAndReturnRemaining(TOKENS_PER_OPERATION); + if (!probe.isConsumed()) { + long retryAfterSeconds = TimeUnit.NANOSECONDS.toSeconds(probe.getNanosToWaitForRefill()); + throw new OperationRateLimitExceededException("Operation rate limit exceeded", retryAfterSeconds); + } + } +} diff --git a/multiapps-controller-web/src/main/java/org/cloudfoundry/multiapps/controller/web/util/PostgresBucketStore.java b/multiapps-controller-web/src/main/java/org/cloudfoundry/multiapps/controller/web/util/PostgresBucketStore.java new file mode 100644 index 0000000000..24f4a030a4 --- /dev/null +++ b/multiapps-controller-web/src/main/java/org/cloudfoundry/multiapps/controller/web/util/PostgresBucketStore.java @@ -0,0 +1,33 @@ +package org.cloudfoundry.multiapps.controller.web.util; + +import javax.sql.DataSource; + +import jakarta.inject.Named; + +import io.github.bucket4j.Bucket; +import io.github.bucket4j.BucketConfiguration; +import io.github.bucket4j.distributed.proxy.ProxyManager; +import io.github.bucket4j.postgresql.Bucket4jPostgreSQL; + +/** + * {@link BucketStore} backed by a PostgreSQL {@link ProxyManager} that uses SELECT ... FOR UPDATE row locking to coordinate token + * consumption across all controller instances sharing the database. + */ +@Named +public class PostgresBucketStore implements BucketStore { + + private static final String BUCKET_TABLE_NAME = "operation_rate_limit_bucket"; + + private final ProxyManager proxyManager; + + public PostgresBucketStore(DataSource dataSource) { + this.proxyManager = Bucket4jPostgreSQL.selectForUpdateBasedBuilder(dataSource) + .table(BUCKET_TABLE_NAME) + .build(); + } + + @Override + public Bucket getBucket(long key, BucketConfiguration configuration) { + return proxyManager.getProxy(key, () -> configuration); + } +} diff --git a/multiapps-controller-web/src/test/java/org/cloudfoundry/multiapps/controller/web/api/impl/OperationsApiServiceImplTest.java b/multiapps-controller-web/src/test/java/org/cloudfoundry/multiapps/controller/web/api/impl/OperationsApiServiceImplTest.java index ea44fb8882..2bc5ae9148 100644 --- a/multiapps-controller-web/src/test/java/org/cloudfoundry/multiapps/controller/web/api/impl/OperationsApiServiceImplTest.java +++ b/multiapps-controller-web/src/test/java/org/cloudfoundry/multiapps/controller/web/api/impl/OperationsApiServiceImplTest.java @@ -39,6 +39,8 @@ import org.cloudfoundry.multiapps.controller.process.util.OperationsHelper; import org.cloudfoundry.multiapps.controller.process.variables.Variables; import org.cloudfoundry.multiapps.controller.web.monitoring.ApiUsageLogger; +import org.cloudfoundry.multiapps.controller.web.util.OperationRateLimitExceededException; +import org.cloudfoundry.multiapps.controller.web.util.OperationRateLimiter; import org.flowable.engine.runtime.ProcessInstance; import org.junit.jupiter.api.Assertions; import org.junit.jupiter.api.BeforeEach; @@ -50,6 +52,8 @@ import org.mockito.Mockito; import org.mockito.MockitoAnnotations; import org.mockito.Spy; +import org.springframework.http.HttpHeaders; +import org.springframework.http.HttpStatus; import org.springframework.http.ResponseEntity; import org.springframework.security.core.context.SecurityContextHolder; import org.springframework.security.oauth2.client.authentication.OAuth2AuthenticationToken; @@ -91,6 +95,8 @@ class OperationsApiServiceImplTest { private ApiUsageLogger apiUsageLogger; @Mock private HttpServletRequest httpServletRequest; + @Mock + private OperationRateLimiter operationRateLimiter; private OperationsApiServiceImpl operationsApiService; @@ -120,7 +126,7 @@ public void initialize() throws Exception { operationsApiService = new OperationsApiServiceImpl(clientFactory, tokenService, operationService, operationMetadataMapper, logsService, flowableFacade, operationsHelper, progressMessageService, processActionRegistry, operationsApiServiceAuditLog, apiUsageLogger, - httpServletRequest); + httpServletRequest, operationRateLimiter); operations = new LinkedList<>(); operations.add(createOperation(FINISHED_PROCESS, Operation.State.FINISHED, Collections.emptyMap())); operations.add(createOperation(RUNNING_PROCESS, Operation.State.RUNNING, Collections.emptyMap())); @@ -213,6 +219,44 @@ void testStartOperation() { .startProcess(Mockito.any(), Mockito.anyMap()); } + @Test + void testStartOperationWhenRateLimitAllowsStartsProcess() { + Map parameters = Map.of(Variables.MTA_ID.getName(), "test"); + Operation operation = createOperation(null, null, parameters); + Mockito.when(operationsHelper.getProcessDefinitionKey(operation)) + .thenReturn("deploy"); + HttpServletRequest httpServletRequestMock = Mockito.mock(HttpServletRequest.class); + Mockito.when(httpServletRequestMock.getRequestURL()) + .thenReturn(new StringBuffer("test/api/path")); + + ResponseEntity response = operationsApiService.startOperation(SPACE_GUID, operation, httpServletRequestMock); + + assertEquals(HttpStatus.ACCEPTED, response.getStatusCode()); + Mockito.verify(operationRateLimiter) + .checkStartAllowed(EXAMPLE_USER, SPACE_GUID); + Mockito.verify(flowableFacade) + .startProcess(Mockito.any(), Mockito.anyMap()); + } + + @Test + void testStartOperationWhenRateLimitExceededReturnsTooManyRequests() { + long retryAfterSeconds = 42; + Map parameters = Map.of(Variables.MTA_ID.getName(), "test"); + Operation operation = createOperation(null, null, parameters); + Mockito.doThrow(new OperationRateLimitExceededException("Operation rate limit exceeded", retryAfterSeconds)) + .when(operationRateLimiter) + .checkStartAllowed(EXAMPLE_USER, SPACE_GUID); + HttpServletRequest httpServletRequestMock = Mockito.mock(HttpServletRequest.class); + + ResponseEntity response = operationsApiService.startOperation(SPACE_GUID, operation, httpServletRequestMock); + + assertEquals(HttpStatus.TOO_MANY_REQUESTS, response.getStatusCode()); + assertEquals(String.valueOf(retryAfterSeconds), response.getHeaders() + .getFirst(HttpHeaders.RETRY_AFTER)); + Mockito.verify(flowableFacade, Mockito.never()) + .startProcess(Mockito.any(), Mockito.anyMap()); + } + @Test void testStartOperationLogsUserGuidAndOriginButNotUsername() { String processInstanceId = "process-instance-id-1"; diff --git a/multiapps-controller-web/src/test/java/org/cloudfoundry/multiapps/controller/web/util/OperationRateLimitKeysTest.java b/multiapps-controller-web/src/test/java/org/cloudfoundry/multiapps/controller/web/util/OperationRateLimitKeysTest.java new file mode 100644 index 0000000000..66b08da17d --- /dev/null +++ b/multiapps-controller-web/src/test/java/org/cloudfoundry/multiapps/controller/web/util/OperationRateLimitKeysTest.java @@ -0,0 +1,72 @@ +package org.cloudfoundry.multiapps.controller.web.util; + +import java.util.HashSet; +import java.util.List; +import java.util.Set; + +import org.junit.jupiter.api.Test; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNotEquals; +import static org.junit.jupiter.api.Assertions.assertTrue; + +class OperationRateLimitKeysTest { + + private static final String SPACE_GUID = "3d4d3f9a-1a2b-4c5d-8e9f-0a1b2c3d4e5f"; + private static final String OTHER_SPACE_GUID = "9f8e7d6c-5b4a-3c2d-1e0f-abcdef123456"; + private static final String USER = "john.doe"; + private static final String OTHER_USER = "jane.roe"; + + @Test + void testSpaceKeyIsDeterministic() { + assertEquals(OperationRateLimitKeys.spaceKey(SPACE_GUID), OperationRateLimitKeys.spaceKey(SPACE_GUID)); + } + + @Test + void testUserKeyIsDeterministic() { + assertEquals(OperationRateLimitKeys.userKey(SPACE_GUID, USER), OperationRateLimitKeys.userKey(SPACE_GUID, USER)); + } + + @Test + void testSpaceKeyAndUserKeyAreDisjointForSameSpace() { + assertNotEquals(OperationRateLimitKeys.spaceKey(SPACE_GUID), OperationRateLimitKeys.userKey(SPACE_GUID, USER)); + } + + @Test + void testDifferentSpacesProduceDifferentKeys() { + assertNotEquals(OperationRateLimitKeys.spaceKey(SPACE_GUID), OperationRateLimitKeys.spaceKey(OTHER_SPACE_GUID)); + } + + @Test + void testDifferentUsersInSameSpaceProduceDifferentKeys() { + assertNotEquals(OperationRateLimitKeys.userKey(SPACE_GUID, USER), OperationRateLimitKeys.userKey(SPACE_GUID, OTHER_USER)); + } + + @Test + void testSameUserInDifferentSpacesProduceDifferentKeys() { + assertNotEquals(OperationRateLimitKeys.userKey(SPACE_GUID, USER), OperationRateLimitKeys.userKey(OTHER_SPACE_GUID, USER)); + } + + @Test + void testNoCollisionsAcrossRealisticSamples() { + List spaceGuids = List.of(SPACE_GUID, OTHER_SPACE_GUID, "11111111-2222-3333-4444-555555555555", + "aaaaaaaa-bbbb-cccc-dddd-eeeeeeeeeeee"); + List users = List.of(USER, OTHER_USER, "admin", "service-account-1"); + Set keys = new HashSet<>(); + for (String spaceGuid : spaceGuids) { + keys.add(OperationRateLimitKeys.spaceKey(spaceGuid)); + for (String user : users) { + keys.add(OperationRateLimitKeys.userKey(spaceGuid, user)); + } + } + int expectedDistinctKeys = spaceGuids.size() + spaceGuids.size() * users.size(); + assertEquals(expectedDistinctKeys, keys.size()); + } + + @Test + void testSpaceKeyIsStableAcrossRuns() { + long firstValue = OperationRateLimitKeys.spaceKey(SPACE_GUID); + long secondValue = OperationRateLimitKeys.spaceKey(SPACE_GUID); + assertTrue(firstValue == secondValue); + } +} diff --git a/multiapps-controller-web/src/test/java/org/cloudfoundry/multiapps/controller/web/util/OperationRateLimiterIntegrationTest.java b/multiapps-controller-web/src/test/java/org/cloudfoundry/multiapps/controller/web/util/OperationRateLimiterIntegrationTest.java new file mode 100644 index 0000000000..cda845fccc --- /dev/null +++ b/multiapps-controller-web/src/test/java/org/cloudfoundry/multiapps/controller/web/util/OperationRateLimiterIntegrationTest.java @@ -0,0 +1,174 @@ +package org.cloudfoundry.multiapps.controller.web.util; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import java.sql.Connection; +import java.sql.ResultSet; +import java.time.Duration; + +import javax.sql.DataSource; + +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.postgresql.ds.PGSimpleDataSource; +import org.testcontainers.containers.PostgreSQLContainer; +import org.testcontainers.junit.jupiter.Container; +import org.testcontainers.junit.jupiter.Testcontainers; + +import io.github.bucket4j.Bandwidth; +import io.github.bucket4j.BucketConfiguration; +import io.github.bucket4j.ConsumptionProbe; +import io.github.bucket4j.distributed.BucketProxy; +import io.github.bucket4j.distributed.proxy.ProxyManager; +import io.github.bucket4j.postgresql.Bucket4jPostgreSQL; +import liquibase.Liquibase; +import liquibase.database.Database; +import liquibase.database.DatabaseFactory; +import liquibase.database.jvm.JdbcConnection; +import liquibase.resource.ClassLoaderResourceAccessor; + +/** + * Proves the real PostgreSQL SELECT ... FOR UPDATE round-trip behind {@link PostgresBucketStore}: token consumption is coordinated through + * the shared database, so two controller instances pointed at the same {@code operation_rate_limit_bucket} table drain a single logical + * bucket. A real PostgreSQL instance is required, so this runs under failsafe (name ends in {@code IntegrationTest}) and needs Docker. + */ +@Testcontainers +class OperationRateLimiterIntegrationTest { + + private static final String BUCKET_TABLE_NAME = "operation_rate_limit_bucket"; + // Step 1's changeset lives in this changelog; running it here validates the exact DDL the application uses. + private static final String BUCKET_CHANGELOG_LOCATION = "org/cloudfoundry/multiapps/controller/persistence/db/changelog/db-changelog-2.52.0-persistence.xml"; + // The master db-changelog.xml resolves this property per-dbms; on PostgreSQL it maps to BYTEA. We supply it directly because we run only + // the 2.52.0 changelog, not the master that declares the property. + private static final String SMALL_BLOB_TYPE_PARAMETER = "small-blob.type"; + private static final String POSTGRES_SMALL_BLOB_TYPE = "BYTEA"; + + private static final long BUCKET_CAPACITY = 3; + private static final long TOKENS_PER_CONSUME = 1; + // A refill window far larger than the test runtime guarantees no tokens are replenished mid-test, keeping the capacity assertions + // deterministic without any sleeps. + private static final Duration REFILL_WINDOW = Duration.ofHours(1); + + private static final long OPERATION_KEY = 42L; + private static final long OTHER_OPERATION_KEY = 43L; + + @Container + private final PostgreSQLContainer postgres = new PostgreSQLContainer<>("postgres:16"); + + private DataSource dataSource; + + @BeforeEach + void setUp() throws Exception { + dataSource = createDataSource(); + applyBucketSchema(); + } + + private DataSource createDataSource() { + var pgDataSource = new PGSimpleDataSource(); + pgDataSource.setUrl(postgres.getJdbcUrl()); + pgDataSource.setUser(postgres.getUsername()); + pgDataSource.setPassword(postgres.getPassword()); + return pgDataSource; + } + + private void applyBucketSchema() throws Exception { + try (Connection connection = dataSource.getConnection()) { + Database database = DatabaseFactory.getInstance() + .findCorrectDatabaseImplementation(new JdbcConnection(connection)); + try (Liquibase liquibase = new Liquibase(BUCKET_CHANGELOG_LOCATION, new ClassLoaderResourceAccessor(), database)) { + liquibase.setChangeLogParameter(SMALL_BLOB_TYPE_PARAMETER, POSTGRES_SMALL_BLOB_TYPE); + liquibase.update(""); + } + } + } + + @Test + void testConsumingAcrossTwoInstancesDrainsSharedBucket() { + var firstInstance = buildProxyManager(); + var secondInstance = buildProxyManager(); + + // Drain the shared bucket by splitting the capacity across two independent proxy managers over the same database. + assertTrue(consumeOneToken(firstInstance, OPERATION_KEY), "first token (instance one) should be consumed"); + assertTrue(consumeOneToken(secondInstance, OPERATION_KEY), "second token (instance two) should be consumed"); + assertTrue(consumeOneToken(firstInstance, OPERATION_KEY), "third token (instance one) should be consumed"); + + // The bucket is now empty. A further consume from EITHER instance must be rejected, proving the SELECT ... FOR UPDATE state is + // shared through Postgres rather than held per-instance. + assertFalse(consumeOneToken(secondInstance, OPERATION_KEY), "instance two must see the shared bucket as drained"); + assertFalse(consumeOneToken(firstInstance, OPERATION_KEY), "instance one must see the shared bucket as drained"); + } + + @Test + void testDifferentKeysAreIndependentBuckets() { + var firstInstance = buildProxyManager(); + var secondInstance = buildProxyManager(); + + for (var token = 0; token < BUCKET_CAPACITY; token++) { + assertTrue(consumeOneToken(firstInstance, OPERATION_KEY), "draining the bucket for the first key should succeed"); + } + assertFalse(consumeOneToken(firstInstance, OPERATION_KEY), "the first key's bucket should be drained"); + + // A different key resolves to a different row, so its bucket is untouched even from the other instance. + assertTrue(consumeOneToken(secondInstance, OTHER_OPERATION_KEY), "a different key must have its own independent bucket"); + } + + @Test + void testBucketStateRowIsPersistedWithExpiresAt() throws Exception { + var proxyManager = buildProxyManager(); + + assertTrue(consumeOneToken(proxyManager, OPERATION_KEY), "the first consume should persist bucket state"); + + assertBucketRowPersisted(OPERATION_KEY); + } + + private ProxyManager buildProxyManager() { + return Bucket4jPostgreSQL.selectForUpdateBasedBuilder(dataSource) + .table(BUCKET_TABLE_NAME) + .build(); + } + + private boolean consumeOneToken(ProxyManager proxyManager, long key) { + BucketProxy bucket = proxyManager.getProxy(key, this::buildBucketConfiguration); + ConsumptionProbe probe = bucket.tryConsumeAndReturnRemaining(TOKENS_PER_CONSUME); + return probe.isConsumed(); + } + + private BucketConfiguration buildBucketConfiguration() { + Bandwidth bandwidth = Bandwidth.builder() + .capacity(BUCKET_CAPACITY) + .refillGreedy(BUCKET_CAPACITY, REFILL_WINDOW) + .build(); + return BucketConfiguration.builder() + .addLimit(bandwidth) + .build(); + } + + private void assertBucketRowPersisted(long key) throws Exception { + var selectRow = "SELECT state, expires_at FROM " + BUCKET_TABLE_NAME + " WHERE id = ?"; + try (Connection connection = dataSource.getConnection(); + var statement = connection.prepareStatement(selectRow)) { + statement.setLong(1, key); + try (ResultSet resultSet = statement.executeQuery()) { + assertTrue(resultSet.next(), "a bucket row must be persisted for the consumed key"); + assertNotNull(resultSet.getBytes("state"), "the serialized bucket state must be stored"); + var expiresAt = resultSet.getLong("expires_at"); + assertFalse(resultSet.wasNull(), "expires_at must be populated so bucket4j can expire stale rows"); + assertTrue(expiresAt > 0, "expires_at must be a positive epoch value"); + assertFalse(resultSet.next(), "a single key must map to exactly one bucket row"); + } + } + assertEquals(1, countBucketRows(), "only the consumed key's bucket row should exist"); + } + + private long countBucketRows() throws Exception { + try (Connection connection = dataSource.getConnection(); + var statement = connection.prepareStatement("SELECT COUNT(*) FROM " + BUCKET_TABLE_NAME); + ResultSet resultSet = statement.executeQuery()) { + resultSet.next(); + return resultSet.getLong(1); + } + } +} diff --git a/multiapps-controller-web/src/test/java/org/cloudfoundry/multiapps/controller/web/util/OperationRateLimiterTest.java b/multiapps-controller-web/src/test/java/org/cloudfoundry/multiapps/controller/web/util/OperationRateLimiterTest.java new file mode 100644 index 0000000000..5644e9ecf5 --- /dev/null +++ b/multiapps-controller-web/src/test/java/org/cloudfoundry/multiapps/controller/web/util/OperationRateLimiterTest.java @@ -0,0 +1,201 @@ +package org.cloudfoundry.multiapps.controller.web.util; + +import org.cloudfoundry.multiapps.controller.api.model.Operation; +import org.cloudfoundry.multiapps.controller.core.util.ApplicationConfiguration; +import org.cloudfoundry.multiapps.controller.persistence.query.OperationQuery; +import org.cloudfoundry.multiapps.controller.persistence.services.OperationService; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.mockito.InjectMocks; +import org.mockito.Mock; +import org.mockito.MockitoAnnotations; + +import io.github.bucket4j.Bucket; +import io.github.bucket4j.BucketConfiguration; +import io.github.bucket4j.ConsumptionProbe; + +import java.util.Collections; +import java.util.List; +import java.util.concurrent.TimeUnit; +import java.util.stream.Stream; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertDoesNotThrow; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.eq; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.verifyNoInteractions; +import static org.mockito.Mockito.when; + +class OperationRateLimiterTest { + + private static final String SPACE_GUID = "3d4d3f9a-1a2b-4c5d-8e9f-0a1b2c3d4e5f"; + private static final String USER = "john.doe"; + private static final int PER_SPACE_CAPACITY = 300; + private static final int PER_SPACE_REFILL_PER_HOUR = 800; + private static final int PER_USER_CAPACITY = 150; + private static final int PER_USER_REFILL_PER_HOUR = 300; + private static final int MAX_ACTIVE_OPERATIONS_PER_SPACE = 500; + private static final int MAX_ACTIVE_OPERATIONS_PER_USER = 200; + private static final long NANOS_TO_WAIT = TimeUnit.SECONDS.toNanos(42); + + @Mock + private ApplicationConfiguration applicationConfiguration; + @Mock + private OperationService operationService; + @Mock + private BucketStore bucketStore; + @Mock + private Bucket spaceBucket; + @Mock + private Bucket userBucket; + @InjectMocks + private OperationRateLimiter operationRateLimiter; + + @BeforeEach + void setUp() throws Exception { + MockitoAnnotations.openMocks(this) + .close(); + } + + private void enableRateLimiting() { + when(applicationConfiguration.isOperationRateLimitingEnabled()).thenReturn(true); + } + + private void stubRateLimitConfiguration() { + when(applicationConfiguration.getMaxActiveOperationsPerSpace()).thenReturn(MAX_ACTIVE_OPERATIONS_PER_SPACE); + when(applicationConfiguration.getMaxActiveOperationsPerUser()).thenReturn(MAX_ACTIVE_OPERATIONS_PER_USER); + when(applicationConfiguration.getOperationRateLimitPerSpaceCapacity()).thenReturn(PER_SPACE_CAPACITY); + when(applicationConfiguration.getOperationRateLimitPerSpaceRefillPerHour()).thenReturn(PER_SPACE_REFILL_PER_HOUR); + when(applicationConfiguration.getOperationRateLimitPerUserCapacity()).thenReturn(PER_USER_CAPACITY); + when(applicationConfiguration.getOperationRateLimitPerUserRefillPerHour()).thenReturn(PER_USER_REFILL_PER_HOUR); + } + + @Test + void testAllowedWhenFeatureFlagOffAndNothingIsTouched() { + when(applicationConfiguration.isOperationRateLimitingEnabled()).thenReturn(false); + + assertDoesNotThrow(() -> operationRateLimiter.checkStartAllowed(USER, SPACE_GUID)); + + verifyNoInteractions(operationService); + verifyNoInteractions(bucketStore); + } + + @Test + void testCheckStartAllowedWhenUnderAllLimits() { + enableRateLimiting(); + stubRateLimitConfiguration(); + stubActiveOperationCounts(0, 0); + stubBucketForKey(OperationRateLimitKeys.spaceKey(SPACE_GUID), spaceBucket); + stubBucketForKey(OperationRateLimitKeys.userKey(SPACE_GUID, USER), userBucket); + stubConsumption(spaceBucket, true); + stubConsumption(userBucket, true); + + assertDoesNotThrow(() -> operationRateLimiter.checkStartAllowed(USER, SPACE_GUID)); + } + + @Test + void testChecksSpaceAndUserBucketsIndependently() { + enableRateLimiting(); + stubRateLimitConfiguration(); + stubActiveOperationCounts(0, 0); + stubBucketForKey(OperationRateLimitKeys.spaceKey(SPACE_GUID), spaceBucket); + stubBucketForKey(OperationRateLimitKeys.userKey(SPACE_GUID, USER), userBucket); + stubConsumption(spaceBucket, true); + stubConsumption(userBucket, true); + + operationRateLimiter.checkStartAllowed(USER, SPACE_GUID); + + verify(bucketStore).getBucket(eq(OperationRateLimitKeys.spaceKey(SPACE_GUID)), any()); + verify(bucketStore).getBucket(eq(OperationRateLimitKeys.userKey(SPACE_GUID, USER)), any()); + } + + @Test + void testThrowExceptionWhenSpaceBucketExhausted() { + enableRateLimiting(); + stubRateLimitConfiguration(); + stubActiveOperationCounts(0, 0); + stubBucketForKey(OperationRateLimitKeys.spaceKey(SPACE_GUID), spaceBucket); + stubConsumption(spaceBucket, false); + + OperationRateLimitExceededException exception = assertThrows(OperationRateLimitExceededException.class, + () -> operationRateLimiter.checkStartAllowed(USER, SPACE_GUID)); + assertEquals(TimeUnit.NANOSECONDS.toSeconds(NANOS_TO_WAIT), exception.getRetryAfterSeconds()); + } + + @Test + void testThrowExceptionWhenUserBucketExhausted() { + enableRateLimiting(); + stubRateLimitConfiguration(); + stubActiveOperationCounts(0, 0); + stubBucketForKey(OperationRateLimitKeys.spaceKey(SPACE_GUID), spaceBucket); + stubBucketForKey(OperationRateLimitKeys.userKey(SPACE_GUID, USER), userBucket); + stubConsumption(spaceBucket, true); + stubConsumption(userBucket, false); + + OperationRateLimitExceededException exception = assertThrows(OperationRateLimitExceededException.class, + () -> operationRateLimiter.checkStartAllowed(USER, SPACE_GUID)); + assertEquals(TimeUnit.NANOSECONDS.toSeconds(NANOS_TO_WAIT), exception.getRetryAfterSeconds()); + } + + @Test + void testThrowExceptionWhenActiveOperationsPerSpaceReachCap() { + enableRateLimiting(); + when(applicationConfiguration.getMaxActiveOperationsPerSpace()).thenReturn(MAX_ACTIVE_OPERATIONS_PER_SPACE); + stubActiveOperationCounts(MAX_ACTIVE_OPERATIONS_PER_SPACE, 0); + + assertThrows(OperationRateLimitExceededException.class, () -> operationRateLimiter.checkStartAllowed(USER, SPACE_GUID)); + verifyNoInteractions(bucketStore); + } + + @Test + void testThrowExceptionWhenActiveOperationsPerUserReachCap() { + enableRateLimiting(); + when(applicationConfiguration.getMaxActiveOperationsPerSpace()).thenReturn(MAX_ACTIVE_OPERATIONS_PER_SPACE); + when(applicationConfiguration.getMaxActiveOperationsPerUser()).thenReturn(MAX_ACTIVE_OPERATIONS_PER_USER); + stubActiveOperationCounts(0, MAX_ACTIVE_OPERATIONS_PER_USER); + + assertThrows(OperationRateLimitExceededException.class, () -> operationRateLimiter.checkStartAllowed(USER, SPACE_GUID)); + verifyNoInteractions(bucketStore); + } + + private void stubActiveOperationCounts(int perSpace, int perUser) { + OperationQuery spaceQuery = mockQuery(); + when(spaceQuery.spaceId(SPACE_GUID)).thenReturn(spaceQuery); + when(spaceQuery.inNonFinalState()).thenReturn(spaceQuery); + when(spaceQuery.list()).thenReturn(activeOperations(perSpace)); + + OperationQuery userQuery = mockQuery(); + when(userQuery.user(USER)).thenReturn(userQuery); + when(userQuery.spaceId(SPACE_GUID)).thenReturn(userQuery); + when(userQuery.inNonFinalState()).thenReturn(userQuery); + when(userQuery.list()).thenReturn(activeOperations(perUser)); + + when(operationService.createQuery()).thenReturn(spaceQuery, userQuery); + } + + private OperationQuery mockQuery() { + return mock(OperationQuery.class); + } + + private List activeOperations(int count) { + if (count == 0) { + return Collections.emptyList(); + } + return Stream.generate(() -> mock(Operation.class)) + .limit(count) + .toList(); + } + + private void stubBucketForKey(long key, Bucket bucket) { + when(bucketStore.getBucket(eq(key), any(BucketConfiguration.class))).thenReturn(bucket); + } + + private void stubConsumption(Bucket bucket, boolean consumed) { + ConsumptionProbe probe = consumed ? ConsumptionProbe.consumed(1, NANOS_TO_WAIT) + : ConsumptionProbe.rejected(0, NANOS_TO_WAIT, NANOS_TO_WAIT); + when(bucket.tryConsumeAndReturnRemaining(1)).thenReturn(probe); + } +} diff --git a/pom.xml b/pom.xml index 87d54ee76e..16cbd38eea 100644 --- a/pom.xml +++ b/pom.xml @@ -65,6 +65,9 @@ 1.0.4 4.0.1 6.3.0 + 8.14.0 + 1.20.6 + 42.7.9 multiapps-controller-client @@ -176,6 +179,12 @@ random + + + org.apache.maven.plugins + maven-failsafe-plugin + 3.5.4 + org.apache.maven.plugins @@ -825,6 +834,30 @@ resilience4j-ratelimiter ${resilience4j.version} + + + com.bucket4j + bucket4j_jdk17-postgresql + ${bucket4j.version} + + + + org.testcontainers + postgresql + ${testcontainers.version} + + + + org.testcontainers + junit-jupiter + ${testcontainers.version} + + + + org.postgresql + postgresql + ${postgresql-jdbc.version} + From bfe61a9fda3603d73767b60322e93bb9d2dc2749 Mon Sep 17 00:00:00 2001 From: IvanBorislavovDimitrov Date: Mon, 27 Jul 2026 17:05:07 +0300 Subject: [PATCH 02/10] Delete expired operation rate-limit buckets Rows in operation_rate_limit_bucket were never removed: one row per space and per (space,user) key was inserted on first use and left forever, since the token-bucket store set no expiration and nothing swept the table. Set an expiration strategy so each row records when its bucket would be fully refilled (i.e. indistinguishable from a fresh bucket), and add a scheduled cleaner that deletes expired rows in batches. The cleaner runs on a single instance, only while rate limiting is enabled, and swallows/logs failures so it never disrupts the scheduler. --- .../multiapps/controller/web/Messages.java | 3 + .../controller/web/util/BucketStore.java | 2 + .../util/OperationRateLimitBucketCleaner.java | 64 ++++++++++++++ .../web/util/PostgresBucketStore.java | 13 ++- .../OperationRateLimitBucketCleanerTest.java | 87 +++++++++++++++++++ 5 files changed, 168 insertions(+), 1 deletion(-) create mode 100644 multiapps-controller-web/src/main/java/org/cloudfoundry/multiapps/controller/web/util/OperationRateLimitBucketCleaner.java create mode 100644 multiapps-controller-web/src/test/java/org/cloudfoundry/multiapps/controller/web/util/OperationRateLimitBucketCleanerTest.java diff --git a/multiapps-controller-web/src/main/java/org/cloudfoundry/multiapps/controller/web/Messages.java b/multiapps-controller-web/src/main/java/org/cloudfoundry/multiapps/controller/web/Messages.java index d38680e7f1..06ff82f30c 100644 --- a/multiapps-controller-web/src/main/java/org/cloudfoundry/multiapps/controller/web/Messages.java +++ b/multiapps-controller-web/src/main/java/org/cloudfoundry/multiapps/controller/web/Messages.java @@ -79,6 +79,9 @@ public final class Messages { public static final String JOB_WITH_ID_WAS_NOT_UPDATED_WITHIN_SECONDS_ON_GET_0_1 = "getUploadFromUrlJob: Job with ID: {0} was not updated within: {1} seconds, returning error"; public static final String CLEARING_OLD_ENTRY = "Clearing old entry with id: {0}"; public static final String STARTED_OPERATION_0_BY_USER_1_AND_ORIGIN_OF_2 = "Started operation \"{0}\" by user \"{1}\" and origin of \"{2}\"."; + public static final String STARTING_CLEAN_UP_OF_EXPIRED_OPERATION_RATE_LIMIT_BUCKETS = "Starting clean up of expired operation rate limit buckets..."; + public static final String DELETED_EXPIRED_OPERATION_RATE_LIMIT_BUCKETS_0 = "Deleted {0} expired operation rate limit buckets"; + public static final String COULD_NOT_CLEAN_UP_EXPIRED_OPERATION_RATE_LIMIT_BUCKETS = "Could not clean up expired operation rate limit buckets"; // DEBUG log messages public static final String RECEIVED_UPLOAD_REQUEST = "Received upload request on URI: {}"; diff --git a/multiapps-controller-web/src/main/java/org/cloudfoundry/multiapps/controller/web/util/BucketStore.java b/multiapps-controller-web/src/main/java/org/cloudfoundry/multiapps/controller/web/util/BucketStore.java index ad6046614d..ca6bd3a2ca 100644 --- a/multiapps-controller-web/src/main/java/org/cloudfoundry/multiapps/controller/web/util/BucketStore.java +++ b/multiapps-controller-web/src/main/java/org/cloudfoundry/multiapps/controller/web/util/BucketStore.java @@ -10,4 +10,6 @@ public interface BucketStore { Bucket getBucket(long key, BucketConfiguration configuration); + + int removeExpiredEntries(int batchSize); } diff --git a/multiapps-controller-web/src/main/java/org/cloudfoundry/multiapps/controller/web/util/OperationRateLimitBucketCleaner.java b/multiapps-controller-web/src/main/java/org/cloudfoundry/multiapps/controller/web/util/OperationRateLimitBucketCleaner.java new file mode 100644 index 0000000000..98ce80f980 --- /dev/null +++ b/multiapps-controller-web/src/main/java/org/cloudfoundry/multiapps/controller/web/util/OperationRateLimitBucketCleaner.java @@ -0,0 +1,64 @@ +package org.cloudfoundry.multiapps.controller.web.util; + +import java.text.MessageFormat; +import java.util.concurrent.TimeUnit; + +import jakarta.inject.Inject; +import jakarta.inject.Named; + +import org.cloudfoundry.multiapps.controller.core.util.ApplicationConfiguration; +import org.cloudfoundry.multiapps.controller.web.Messages; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.springframework.scheduling.annotation.Scheduled; + +/** + * Periodically deletes expired rows from the operation rate limit bucket table. bucket4j populates each row's expiry but never removes + * expired rows on its own, so without this sweeper the table grows without bound as new spaces and users start operations. + */ +@Named +public class OperationRateLimitBucketCleaner { + + private static final Logger LOGGER = LoggerFactory.getLogger(OperationRateLimitBucketCleaner.class); + private static final int SELECTED_INSTANCE_FOR_CLEAN_UP = 0; + private static final int DELETE_BATCH_SIZE = 100; + private static final int MAX_ITERATIONS = 1000; + + private final ApplicationConfiguration applicationConfiguration; + private final BucketStore bucketStore; + + @Inject + public OperationRateLimitBucketCleaner(ApplicationConfiguration applicationConfiguration, BucketStore bucketStore) { + this.applicationConfiguration = applicationConfiguration; + this.bucketStore = bucketStore; + } + + @Scheduled(fixedRate = 1, timeUnit = TimeUnit.HOURS) + public void cleanUpExpiredBuckets() { + if (!applicationConfiguration.isOperationRateLimitingEnabled()) { + return; + } + if (applicationConfiguration.getApplicationInstanceIndex() != SELECTED_INSTANCE_FOR_CLEAN_UP) { + return; + } + LOGGER.info(Messages.STARTING_CLEAN_UP_OF_EXPIRED_OPERATION_RATE_LIMIT_BUCKETS); + try { + int totalDeleted = deleteExpiredBucketsInBatches(); + LOGGER.info(MessageFormat.format(Messages.DELETED_EXPIRED_OPERATION_RATE_LIMIT_BUCKETS_0, totalDeleted)); + } catch (Exception e) { + LOGGER.error(Messages.COULD_NOT_CLEAN_UP_EXPIRED_OPERATION_RATE_LIMIT_BUCKETS, e); + } + } + + private int deleteExpiredBucketsInBatches() { + int totalDeleted = 0; + for (int iteration = 0; iteration < MAX_ITERATIONS; iteration++) { + int deleted = bucketStore.removeExpiredEntries(DELETE_BATCH_SIZE); + totalDeleted += deleted; + if (deleted < DELETE_BATCH_SIZE) { + break; + } + } + return totalDeleted; + } +} diff --git a/multiapps-controller-web/src/main/java/org/cloudfoundry/multiapps/controller/web/util/PostgresBucketStore.java b/multiapps-controller-web/src/main/java/org/cloudfoundry/multiapps/controller/web/util/PostgresBucketStore.java index 24f4a030a4..fcf9af4a89 100644 --- a/multiapps-controller-web/src/main/java/org/cloudfoundry/multiapps/controller/web/util/PostgresBucketStore.java +++ b/multiapps-controller-web/src/main/java/org/cloudfoundry/multiapps/controller/web/util/PostgresBucketStore.java @@ -1,13 +1,17 @@ package org.cloudfoundry.multiapps.controller.web.util; +import java.time.Duration; + import javax.sql.DataSource; import jakarta.inject.Named; import io.github.bucket4j.Bucket; import io.github.bucket4j.BucketConfiguration; +import io.github.bucket4j.distributed.ExpirationAfterWriteStrategy; import io.github.bucket4j.distributed.proxy.ProxyManager; import io.github.bucket4j.postgresql.Bucket4jPostgreSQL; +import io.github.bucket4j.postgresql.PostgreSQLSelectForUpdateBasedProxyManager; /** * {@link BucketStore} backed by a PostgreSQL {@link ProxyManager} that uses SELECT ... FOR UPDATE row locking to coordinate token @@ -17,12 +21,14 @@ public class PostgresBucketStore implements BucketStore { private static final String BUCKET_TABLE_NAME = "operation_rate_limit_bucket"; + private static final Duration BUCKET_TIME_TO_LIVE = Duration.ofHours(1); - private final ProxyManager proxyManager; + private final PostgreSQLSelectForUpdateBasedProxyManager proxyManager; public PostgresBucketStore(DataSource dataSource) { this.proxyManager = Bucket4jPostgreSQL.selectForUpdateBasedBuilder(dataSource) .table(BUCKET_TABLE_NAME) + .expirationAfterWrite(ExpirationAfterWriteStrategy.basedOnTimeForRefillingBucketUpToMax(BUCKET_TIME_TO_LIVE)) .build(); } @@ -30,4 +36,9 @@ public PostgresBucketStore(DataSource dataSource) { public Bucket getBucket(long key, BucketConfiguration configuration) { return proxyManager.getProxy(key, () -> configuration); } + + @Override + public int removeExpiredEntries(int batchSize) { + return proxyManager.removeExpired(batchSize); + } } diff --git a/multiapps-controller-web/src/test/java/org/cloudfoundry/multiapps/controller/web/util/OperationRateLimitBucketCleanerTest.java b/multiapps-controller-web/src/test/java/org/cloudfoundry/multiapps/controller/web/util/OperationRateLimitBucketCleanerTest.java new file mode 100644 index 0000000000..0ecfa233a9 --- /dev/null +++ b/multiapps-controller-web/src/test/java/org/cloudfoundry/multiapps/controller/web/util/OperationRateLimitBucketCleanerTest.java @@ -0,0 +1,87 @@ +package org.cloudfoundry.multiapps.controller.web.util; + +import static org.junit.jupiter.api.Assertions.assertDoesNotThrow; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.verifyNoInteractions; +import static org.mockito.Mockito.when; + +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.mockito.InjectMocks; +import org.mockito.Mock; +import org.mockito.MockitoAnnotations; + +import org.cloudfoundry.multiapps.controller.core.util.ApplicationConfiguration; + +class OperationRateLimitBucketCleanerTest { + + private static final int SELECTED_INSTANCE = 0; + private static final int OTHER_INSTANCE = 3; + private static final int BATCH_SIZE = 100; + + @Mock + private ApplicationConfiguration applicationConfiguration; + @Mock + private BucketStore bucketStore; + @InjectMocks + private OperationRateLimitBucketCleaner cleaner; + + @BeforeEach + void setUp() throws Exception { + MockitoAnnotations.openMocks(this) + .close(); + } + + @Test + void testDoesNothingWhenRateLimitingDisabled() { + when(applicationConfiguration.isOperationRateLimitingEnabled()).thenReturn(false); + + cleaner.cleanUpExpiredBuckets(); + + verifyNoInteractions(bucketStore); + } + + @Test + void testDoesNothingWhenNotSelectedInstance() { + when(applicationConfiguration.isOperationRateLimitingEnabled()).thenReturn(true); + when(applicationConfiguration.getApplicationInstanceIndex()).thenReturn(OTHER_INSTANCE); + + cleaner.cleanUpExpiredBuckets(); + + verifyNoInteractions(bucketStore); + } + + @Test + void testDeletesInASingleBatchWhenFewerThanBatchSizeExpired() { + enableCleaningOnSelectedInstance(); + when(bucketStore.removeExpiredEntries(BATCH_SIZE)).thenReturn(0); + + cleaner.cleanUpExpiredBuckets(); + + verify(bucketStore, times(1)).removeExpiredEntries(BATCH_SIZE); + } + + @Test + void testKeepsDeletingUntilBatchNotFull() { + enableCleaningOnSelectedInstance(); + when(bucketStore.removeExpiredEntries(BATCH_SIZE)).thenReturn(BATCH_SIZE, 30); + + cleaner.cleanUpExpiredBuckets(); + + verify(bucketStore, times(2)).removeExpiredEntries(BATCH_SIZE); + } + + @Test + void testSwallowsExceptionFromBucketStore() { + enableCleaningOnSelectedInstance(); + when(bucketStore.removeExpiredEntries(BATCH_SIZE)).thenThrow(new RuntimeException("boom")); + + assertDoesNotThrow(() -> cleaner.cleanUpExpiredBuckets()); + } + + private void enableCleaningOnSelectedInstance() { + when(applicationConfiguration.isOperationRateLimitingEnabled()).thenReturn(true); + when(applicationConfiguration.getApplicationInstanceIndex()).thenReturn(SELECTED_INSTANCE); + } +} From 4441fb4e5e8aba403a7d424e26a03f20319bf489 Mon Sep 17 00:00:00 2001 From: IvanBorislavovDimitrov Date: Tue, 28 Jul 2026 11:45:22 +0300 Subject: [PATCH 03/10] Extract operation rate-limit exception messages into constants Replace the three inline exception-message string literals in the rate limiter with named constants in the web Messages class, matching the existing exception-message convention. No behavior change. --- .../cloudfoundry/multiapps/controller/web/Messages.java | 3 +++ .../controller/web/util/OperationRateLimiter.java | 7 ++++--- 2 files changed, 7 insertions(+), 3 deletions(-) diff --git a/multiapps-controller-web/src/main/java/org/cloudfoundry/multiapps/controller/web/Messages.java b/multiapps-controller-web/src/main/java/org/cloudfoundry/multiapps/controller/web/Messages.java index 06ff82f30c..3f0e571042 100644 --- a/multiapps-controller-web/src/main/java/org/cloudfoundry/multiapps/controller/web/Messages.java +++ b/multiapps-controller-web/src/main/java/org/cloudfoundry/multiapps/controller/web/Messages.java @@ -24,6 +24,9 @@ public final class Messages { public static final String MISSING_PROPERTIES_FOR_CREATING_THE_SPECIFIC_PROVIDER = "Missing properties for creating the specific provider!"; public static final String DEPLOY_FROM_URL_WRONG_CREDENTIALS_FOR_JOB_WITH_ID = "Credentials to {0} are wrong. Make sure that they are correct. Job id: {1}"; public static final String JOB_NOT_UPDATED_FOR_0_SECONDS = "Job not updated for {0} seconds"; + public static final String TOO_MANY_ACTIVE_OPERATIONS_IN_SPACE = "Too many active operations in space"; + public static final String TOO_MANY_ACTIVE_OPERATIONS_FOR_USER = "Too many active operations for user"; + public static final String OPERATION_RATE_LIMIT_EXCEEDED = "Operation rate limit exceeded"; public static final String FAILED_TO_CREATE_BLOB_STORE_CONTEXT = "Failed to create BlobStoreContext"; diff --git a/multiapps-controller-web/src/main/java/org/cloudfoundry/multiapps/controller/web/util/OperationRateLimiter.java b/multiapps-controller-web/src/main/java/org/cloudfoundry/multiapps/controller/web/util/OperationRateLimiter.java index 7576767ce1..608e63e022 100644 --- a/multiapps-controller-web/src/main/java/org/cloudfoundry/multiapps/controller/web/util/OperationRateLimiter.java +++ b/multiapps-controller-web/src/main/java/org/cloudfoundry/multiapps/controller/web/util/OperationRateLimiter.java @@ -7,6 +7,7 @@ import org.cloudfoundry.multiapps.controller.core.util.ApplicationConfiguration; import org.cloudfoundry.multiapps.controller.persistence.services.OperationService; +import org.cloudfoundry.multiapps.controller.web.Messages; import io.github.bucket4j.Bandwidth; import io.github.bucket4j.Bucket; @@ -50,7 +51,7 @@ private void checkActiveOperationCaps(String user, String spaceGuid) { .list() .size(); if (activeOperationsPerSpace >= applicationConfiguration.getMaxActiveOperationsPerSpace()) { - throw new OperationRateLimitExceededException("Too many active operations in space", NO_RETRY_AFTER_SECONDS); + throw new OperationRateLimitExceededException(Messages.TOO_MANY_ACTIVE_OPERATIONS_IN_SPACE, NO_RETRY_AFTER_SECONDS); } int activeOperationsPerUser = operationService.createQuery() .user(user) @@ -59,7 +60,7 @@ private void checkActiveOperationCaps(String user, String spaceGuid) { .list() .size(); if (activeOperationsPerUser >= applicationConfiguration.getMaxActiveOperationsPerUser()) { - throw new OperationRateLimitExceededException("Too many active operations for user", NO_RETRY_AFTER_SECONDS); + throw new OperationRateLimitExceededException(Messages.TOO_MANY_ACTIVE_OPERATIONS_FOR_USER, NO_RETRY_AFTER_SECONDS); } } @@ -96,7 +97,7 @@ private void consumeToken(Bucket bucket) { ConsumptionProbe probe = bucket.tryConsumeAndReturnRemaining(TOKENS_PER_OPERATION); if (!probe.isConsumed()) { long retryAfterSeconds = TimeUnit.NANOSECONDS.toSeconds(probe.getNanosToWaitForRefill()); - throw new OperationRateLimitExceededException("Operation rate limit exceeded", retryAfterSeconds); + throw new OperationRateLimitExceededException(Messages.OPERATION_RATE_LIMIT_EXCEEDED, retryAfterSeconds); } } } From 4543b7e3c4f7a7d8c5396be5605c4e98b941d070 Mon Sep 17 00:00:00 2001 From: IvanBorislavovDimitrov Date: Tue, 28 Jul 2026 12:01:10 +0300 Subject: [PATCH 04/10] Raise expired-bucket cleanup batch size and iteration cap Increase the sweep batch to 1000 and the iteration cap to 10000 so a single cleanup run can clear far more expired rows, matching landscapes that accumulate many distinct rate-limit keys. --- .../controller/web/util/OperationRateLimitBucketCleaner.java | 4 ++-- .../web/util/OperationRateLimitBucketCleanerTest.java | 2 +- 2 files changed, 3 insertions(+), 3 deletions(-) diff --git a/multiapps-controller-web/src/main/java/org/cloudfoundry/multiapps/controller/web/util/OperationRateLimitBucketCleaner.java b/multiapps-controller-web/src/main/java/org/cloudfoundry/multiapps/controller/web/util/OperationRateLimitBucketCleaner.java index 98ce80f980..afbc81a4b2 100644 --- a/multiapps-controller-web/src/main/java/org/cloudfoundry/multiapps/controller/web/util/OperationRateLimitBucketCleaner.java +++ b/multiapps-controller-web/src/main/java/org/cloudfoundry/multiapps/controller/web/util/OperationRateLimitBucketCleaner.java @@ -21,8 +21,8 @@ public class OperationRateLimitBucketCleaner { private static final Logger LOGGER = LoggerFactory.getLogger(OperationRateLimitBucketCleaner.class); private static final int SELECTED_INSTANCE_FOR_CLEAN_UP = 0; - private static final int DELETE_BATCH_SIZE = 100; - private static final int MAX_ITERATIONS = 1000; + private static final int DELETE_BATCH_SIZE = 1000; + private static final int MAX_ITERATIONS = 10000; private final ApplicationConfiguration applicationConfiguration; private final BucketStore bucketStore; diff --git a/multiapps-controller-web/src/test/java/org/cloudfoundry/multiapps/controller/web/util/OperationRateLimitBucketCleanerTest.java b/multiapps-controller-web/src/test/java/org/cloudfoundry/multiapps/controller/web/util/OperationRateLimitBucketCleanerTest.java index 0ecfa233a9..689d4915ba 100644 --- a/multiapps-controller-web/src/test/java/org/cloudfoundry/multiapps/controller/web/util/OperationRateLimitBucketCleanerTest.java +++ b/multiapps-controller-web/src/test/java/org/cloudfoundry/multiapps/controller/web/util/OperationRateLimitBucketCleanerTest.java @@ -18,7 +18,7 @@ class OperationRateLimitBucketCleanerTest { private static final int SELECTED_INSTANCE = 0; private static final int OTHER_INSTANCE = 3; - private static final int BATCH_SIZE = 100; + private static final int BATCH_SIZE = 1000; @Mock private ApplicationConfiguration applicationConfiguration; From 68111bec38bc11a94584065a12bb8ba287fe0bb6 Mon Sep 17 00:00:00 2001 From: IvanBorislavovDimitrov Date: Tue, 28 Jul 2026 12:37:13 +0300 Subject: [PATCH 05/10] Move operation rate-limit bucket cleaner next to the other clean-up jobs Relocate OperationRateLimitBucketCleaner into the process module's jobs package, alongside the existing clean-up jobs, and move its BucketStore / PostgresBucketStore collaborators into the process util package. The bucket4j dependency, the Postgres integration test, and the failsafe plugin move to the process module accordingly; the web limiter now uses the bucket store transitively. The three cleaner log messages move to the process Messages class. No behavior change. --- multiapps-controller-process/pom.xml | 36 +++++++++++++++++++ .../src/main/java/module-info.java | 2 ++ .../controller/process/Messages.java | 3 ++ .../OperationRateLimitBucketCleaner.java | 5 +-- .../controller/process}/util/BucketStore.java | 2 +- .../process}/util/PostgresBucketStore.java | 2 +- .../OperationRateLimitBucketCleanerTest.java | 3 +- .../OperationRateLimiterIntegrationTest.java | 2 +- multiapps-controller-web/pom.xml | 36 ------------------- .../multiapps/controller/web/Messages.java | 3 -- .../web/util/OperationRateLimiter.java | 1 + .../web/util/OperationRateLimiterTest.java | 1 + 12 files changed, 51 insertions(+), 45 deletions(-) rename {multiapps-controller-web/src/main/java/org/cloudfoundry/multiapps/controller/web/util => multiapps-controller-process/src/main/java/org/cloudfoundry/multiapps/controller/process/jobs}/OperationRateLimitBucketCleaner.java (92%) rename {multiapps-controller-web/src/main/java/org/cloudfoundry/multiapps/controller/web => multiapps-controller-process/src/main/java/org/cloudfoundry/multiapps/controller/process}/util/BucketStore.java (89%) rename {multiapps-controller-web/src/main/java/org/cloudfoundry/multiapps/controller/web => multiapps-controller-process/src/main/java/org/cloudfoundry/multiapps/controller/process}/util/PostgresBucketStore.java (96%) rename {multiapps-controller-web/src/test/java/org/cloudfoundry/multiapps/controller/web/util => multiapps-controller-process/src/test/java/org/cloudfoundry/multiapps/controller/process/jobs}/OperationRateLimitBucketCleanerTest.java (95%) rename {multiapps-controller-web/src/test/java/org/cloudfoundry/multiapps/controller/web => multiapps-controller-process/src/test/java/org/cloudfoundry/multiapps/controller/process}/util/OperationRateLimiterIntegrationTest.java (99%) diff --git a/multiapps-controller-process/pom.xml b/multiapps-controller-process/pom.xml index f33f5c6386..a94e07f8bf 100644 --- a/multiapps-controller-process/pom.xml +++ b/multiapps-controller-process/pom.xml @@ -25,6 +25,23 @@ + + org.apache.maven.plugins + maven-failsafe-plugin + + + + integration-test + verify + + + + + + **/*IntegrationTest + + + @@ -110,5 +127,24 @@ org.cloudfoundry.multiapps multiapps-controller-shutdown-client + + com.bucket4j + bucket4j_jdk17-postgresql + + + org.testcontainers + postgresql + test + + + org.testcontainers + junit-jupiter + test + + + org.postgresql + postgresql + test + \ No newline at end of file diff --git a/multiapps-controller-process/src/main/java/module-info.java b/multiapps-controller-process/src/main/java/module-info.java index 6b87de4bd9..6a67ce6fab 100644 --- a/multiapps-controller-process/src/main/java/module-info.java +++ b/multiapps-controller-process/src/main/java/module-info.java @@ -67,5 +67,7 @@ requires static java.compiler; requires static org.immutables.value; requires org.cloudfoundry.multiapps.controller.shutdown.client; + requires io.github.bucket4j.core; + requires io.github.bucket4j.postgresql; } \ No newline at end of file diff --git a/multiapps-controller-process/src/main/java/org/cloudfoundry/multiapps/controller/process/Messages.java b/multiapps-controller-process/src/main/java/org/cloudfoundry/multiapps/controller/process/Messages.java index ac8279c2f8..86ef2d9b56 100755 --- a/multiapps-controller-process/src/main/java/org/cloudfoundry/multiapps/controller/process/Messages.java +++ b/multiapps-controller-process/src/main/java/org/cloudfoundry/multiapps/controller/process/Messages.java @@ -105,6 +105,7 @@ public class Messages { public static final String ERROR_DURING_INCREMENTAL_INSTANCE_UPDATE_OF_MODULE_0 = "Error during incremental instance update of module \"{0}\""; public static final String ERROR_DURING_POLL_OF_INCREMENTAL_INSTANCE_UPDATE_OF_MODULE_0 = "Error during poll of incremental instance update of module \"{0}\""; public static final String FAILED_TO_GET_CLOUD_LOGGING_SERVICE_KEY = "Failed to get Cloud Logging service key"; + public static final String COULD_NOT_CLEAN_UP_EXPIRED_OPERATION_RATE_LIMIT_BUCKETS = "Could not clean up expired operation rate limit buckets"; // Process step errors public static final String ERROR_VALIDATING_PARAMS = "Error validating parameters"; @@ -358,6 +359,8 @@ public class Messages { public static final String DELETING_BACKUP_DESCRIPTORS_WITH_MTA_ID_0_SPACE_1_NAMESPACE_2_AND_SKIP_VERSIONS_3 = "Deleting backup descriptors with mta id \"{0}\" in space \"{1}\" namespace \"{2}\" and skip the following mta versions \"{3}\""; public static final String EXISTING_APPS_TO_BACKUP = "Existing apps to backup: {0}"; public static final String TASK_0_ON_APPLICATION_1_IS_STILL_2 = "Task \"{0}\" on application \"{1}\" is still \"{2}\""; + public static final String STARTING_CLEAN_UP_OF_EXPIRED_OPERATION_RATE_LIMIT_BUCKETS = "Starting clean up of expired operation rate limit buckets..."; + public static final String DELETED_EXPIRED_OPERATION_RATE_LIMIT_BUCKETS_0 = "Deleted {0} expired operation rate limit buckets"; // Progress messages public static final String OPERATION_ID = "Operation ID: {0}"; diff --git a/multiapps-controller-web/src/main/java/org/cloudfoundry/multiapps/controller/web/util/OperationRateLimitBucketCleaner.java b/multiapps-controller-process/src/main/java/org/cloudfoundry/multiapps/controller/process/jobs/OperationRateLimitBucketCleaner.java similarity index 92% rename from multiapps-controller-web/src/main/java/org/cloudfoundry/multiapps/controller/web/util/OperationRateLimitBucketCleaner.java rename to multiapps-controller-process/src/main/java/org/cloudfoundry/multiapps/controller/process/jobs/OperationRateLimitBucketCleaner.java index afbc81a4b2..7ce526176d 100644 --- a/multiapps-controller-web/src/main/java/org/cloudfoundry/multiapps/controller/web/util/OperationRateLimitBucketCleaner.java +++ b/multiapps-controller-process/src/main/java/org/cloudfoundry/multiapps/controller/process/jobs/OperationRateLimitBucketCleaner.java @@ -1,4 +1,4 @@ -package org.cloudfoundry.multiapps.controller.web.util; +package org.cloudfoundry.multiapps.controller.process.jobs; import java.text.MessageFormat; import java.util.concurrent.TimeUnit; @@ -7,7 +7,8 @@ import jakarta.inject.Named; import org.cloudfoundry.multiapps.controller.core.util.ApplicationConfiguration; -import org.cloudfoundry.multiapps.controller.web.Messages; +import org.cloudfoundry.multiapps.controller.process.Messages; +import org.cloudfoundry.multiapps.controller.process.util.BucketStore; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.scheduling.annotation.Scheduled; diff --git a/multiapps-controller-web/src/main/java/org/cloudfoundry/multiapps/controller/web/util/BucketStore.java b/multiapps-controller-process/src/main/java/org/cloudfoundry/multiapps/controller/process/util/BucketStore.java similarity index 89% rename from multiapps-controller-web/src/main/java/org/cloudfoundry/multiapps/controller/web/util/BucketStore.java rename to multiapps-controller-process/src/main/java/org/cloudfoundry/multiapps/controller/process/util/BucketStore.java index ca6bd3a2ca..56d1961a39 100644 --- a/multiapps-controller-web/src/main/java/org/cloudfoundry/multiapps/controller/web/util/BucketStore.java +++ b/multiapps-controller-process/src/main/java/org/cloudfoundry/multiapps/controller/process/util/BucketStore.java @@ -1,4 +1,4 @@ -package org.cloudfoundry.multiapps.controller.web.util; +package org.cloudfoundry.multiapps.controller.process.util; import io.github.bucket4j.Bucket; import io.github.bucket4j.BucketConfiguration; diff --git a/multiapps-controller-web/src/main/java/org/cloudfoundry/multiapps/controller/web/util/PostgresBucketStore.java b/multiapps-controller-process/src/main/java/org/cloudfoundry/multiapps/controller/process/util/PostgresBucketStore.java similarity index 96% rename from multiapps-controller-web/src/main/java/org/cloudfoundry/multiapps/controller/web/util/PostgresBucketStore.java rename to multiapps-controller-process/src/main/java/org/cloudfoundry/multiapps/controller/process/util/PostgresBucketStore.java index fcf9af4a89..fe93e52f5d 100644 --- a/multiapps-controller-web/src/main/java/org/cloudfoundry/multiapps/controller/web/util/PostgresBucketStore.java +++ b/multiapps-controller-process/src/main/java/org/cloudfoundry/multiapps/controller/process/util/PostgresBucketStore.java @@ -1,4 +1,4 @@ -package org.cloudfoundry.multiapps.controller.web.util; +package org.cloudfoundry.multiapps.controller.process.util; import java.time.Duration; diff --git a/multiapps-controller-web/src/test/java/org/cloudfoundry/multiapps/controller/web/util/OperationRateLimitBucketCleanerTest.java b/multiapps-controller-process/src/test/java/org/cloudfoundry/multiapps/controller/process/jobs/OperationRateLimitBucketCleanerTest.java similarity index 95% rename from multiapps-controller-web/src/test/java/org/cloudfoundry/multiapps/controller/web/util/OperationRateLimitBucketCleanerTest.java rename to multiapps-controller-process/src/test/java/org/cloudfoundry/multiapps/controller/process/jobs/OperationRateLimitBucketCleanerTest.java index 689d4915ba..cad08ea837 100644 --- a/multiapps-controller-web/src/test/java/org/cloudfoundry/multiapps/controller/web/util/OperationRateLimitBucketCleanerTest.java +++ b/multiapps-controller-process/src/test/java/org/cloudfoundry/multiapps/controller/process/jobs/OperationRateLimitBucketCleanerTest.java @@ -1,4 +1,4 @@ -package org.cloudfoundry.multiapps.controller.web.util; +package org.cloudfoundry.multiapps.controller.process.jobs; import static org.junit.jupiter.api.Assertions.assertDoesNotThrow; import static org.mockito.Mockito.times; @@ -13,6 +13,7 @@ import org.mockito.MockitoAnnotations; import org.cloudfoundry.multiapps.controller.core.util.ApplicationConfiguration; +import org.cloudfoundry.multiapps.controller.process.util.BucketStore; class OperationRateLimitBucketCleanerTest { diff --git a/multiapps-controller-web/src/test/java/org/cloudfoundry/multiapps/controller/web/util/OperationRateLimiterIntegrationTest.java b/multiapps-controller-process/src/test/java/org/cloudfoundry/multiapps/controller/process/util/OperationRateLimiterIntegrationTest.java similarity index 99% rename from multiapps-controller-web/src/test/java/org/cloudfoundry/multiapps/controller/web/util/OperationRateLimiterIntegrationTest.java rename to multiapps-controller-process/src/test/java/org/cloudfoundry/multiapps/controller/process/util/OperationRateLimiterIntegrationTest.java index cda845fccc..dbf171ee03 100644 --- a/multiapps-controller-web/src/test/java/org/cloudfoundry/multiapps/controller/web/util/OperationRateLimiterIntegrationTest.java +++ b/multiapps-controller-process/src/test/java/org/cloudfoundry/multiapps/controller/process/util/OperationRateLimiterIntegrationTest.java @@ -1,4 +1,4 @@ -package org.cloudfoundry.multiapps.controller.web.util; +package org.cloudfoundry.multiapps.controller.process.util; import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertFalse; diff --git a/multiapps-controller-web/pom.xml b/multiapps-controller-web/pom.xml index e25e1e1303..1c377426ed 100644 --- a/multiapps-controller-web/pom.xml +++ b/multiapps-controller-web/pom.xml @@ -82,23 +82,6 @@ - - org.apache.maven.plugins - maven-failsafe-plugin - - - - integration-test - verify - - - - - - **/*IntegrationTest - - - @@ -168,10 +151,6 @@ io.github.resilience4j resilience4j-ratelimiter - - com.bucket4j - bucket4j_jdk17-postgresql - org.apache.jclouds.common googlecloud @@ -236,20 +215,5 @@ software.amazon.awssdk url-connection-client - - org.testcontainers - postgresql - test - - - org.testcontainers - junit-jupiter - test - - - org.postgresql - postgresql - test - diff --git a/multiapps-controller-web/src/main/java/org/cloudfoundry/multiapps/controller/web/Messages.java b/multiapps-controller-web/src/main/java/org/cloudfoundry/multiapps/controller/web/Messages.java index 3f0e571042..57175a2a82 100644 --- a/multiapps-controller-web/src/main/java/org/cloudfoundry/multiapps/controller/web/Messages.java +++ b/multiapps-controller-web/src/main/java/org/cloudfoundry/multiapps/controller/web/Messages.java @@ -82,9 +82,6 @@ public final class Messages { public static final String JOB_WITH_ID_WAS_NOT_UPDATED_WITHIN_SECONDS_ON_GET_0_1 = "getUploadFromUrlJob: Job with ID: {0} was not updated within: {1} seconds, returning error"; public static final String CLEARING_OLD_ENTRY = "Clearing old entry with id: {0}"; public static final String STARTED_OPERATION_0_BY_USER_1_AND_ORIGIN_OF_2 = "Started operation \"{0}\" by user \"{1}\" and origin of \"{2}\"."; - public static final String STARTING_CLEAN_UP_OF_EXPIRED_OPERATION_RATE_LIMIT_BUCKETS = "Starting clean up of expired operation rate limit buckets..."; - public static final String DELETED_EXPIRED_OPERATION_RATE_LIMIT_BUCKETS_0 = "Deleted {0} expired operation rate limit buckets"; - public static final String COULD_NOT_CLEAN_UP_EXPIRED_OPERATION_RATE_LIMIT_BUCKETS = "Could not clean up expired operation rate limit buckets"; // DEBUG log messages public static final String RECEIVED_UPLOAD_REQUEST = "Received upload request on URI: {}"; diff --git a/multiapps-controller-web/src/main/java/org/cloudfoundry/multiapps/controller/web/util/OperationRateLimiter.java b/multiapps-controller-web/src/main/java/org/cloudfoundry/multiapps/controller/web/util/OperationRateLimiter.java index 608e63e022..7745482e5f 100644 --- a/multiapps-controller-web/src/main/java/org/cloudfoundry/multiapps/controller/web/util/OperationRateLimiter.java +++ b/multiapps-controller-web/src/main/java/org/cloudfoundry/multiapps/controller/web/util/OperationRateLimiter.java @@ -7,6 +7,7 @@ import org.cloudfoundry.multiapps.controller.core.util.ApplicationConfiguration; import org.cloudfoundry.multiapps.controller.persistence.services.OperationService; +import org.cloudfoundry.multiapps.controller.process.util.BucketStore; import org.cloudfoundry.multiapps.controller.web.Messages; import io.github.bucket4j.Bandwidth; diff --git a/multiapps-controller-web/src/test/java/org/cloudfoundry/multiapps/controller/web/util/OperationRateLimiterTest.java b/multiapps-controller-web/src/test/java/org/cloudfoundry/multiapps/controller/web/util/OperationRateLimiterTest.java index 5644e9ecf5..5eb3599022 100644 --- a/multiapps-controller-web/src/test/java/org/cloudfoundry/multiapps/controller/web/util/OperationRateLimiterTest.java +++ b/multiapps-controller-web/src/test/java/org/cloudfoundry/multiapps/controller/web/util/OperationRateLimiterTest.java @@ -4,6 +4,7 @@ import org.cloudfoundry.multiapps.controller.core.util.ApplicationConfiguration; import org.cloudfoundry.multiapps.controller.persistence.query.OperationQuery; import org.cloudfoundry.multiapps.controller.persistence.services.OperationService; +import org.cloudfoundry.multiapps.controller.process.util.BucketStore; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; import org.mockito.InjectMocks; From aee7c1e4e2715f4e7f8159b7eeafeb056ee910d0 Mon Sep 17 00:00:00 2001 From: IvanBorislavovDimitrov Date: Tue, 28 Jul 2026 16:41:04 +0300 Subject: [PATCH 06/10] Annotate PostgresBucketStore constructor with @Inject Match the constructor-injection convention of the other @Named beans in the process util package. --- .../multiapps/controller/process/util/PostgresBucketStore.java | 2 ++ 1 file changed, 2 insertions(+) diff --git a/multiapps-controller-process/src/main/java/org/cloudfoundry/multiapps/controller/process/util/PostgresBucketStore.java b/multiapps-controller-process/src/main/java/org/cloudfoundry/multiapps/controller/process/util/PostgresBucketStore.java index fe93e52f5d..4ece62969a 100644 --- a/multiapps-controller-process/src/main/java/org/cloudfoundry/multiapps/controller/process/util/PostgresBucketStore.java +++ b/multiapps-controller-process/src/main/java/org/cloudfoundry/multiapps/controller/process/util/PostgresBucketStore.java @@ -4,6 +4,7 @@ import javax.sql.DataSource; +import jakarta.inject.Inject; import jakarta.inject.Named; import io.github.bucket4j.Bucket; @@ -25,6 +26,7 @@ public class PostgresBucketStore implements BucketStore { private final PostgreSQLSelectForUpdateBasedProxyManager proxyManager; + @Inject public PostgresBucketStore(DataSource dataSource) { this.proxyManager = Bucket4jPostgreSQL.selectForUpdateBasedBuilder(dataSource) .table(BUCKET_TABLE_NAME) From 24166c75afd1a47f70bb558368f2ca997403da9e Mon Sep 17 00:00:00 2001 From: Yavor16 Date: Wed, 9 Sep 2026 12:17:45 +0300 Subject: [PATCH 07/10] Log structured INFO event on every rate-limit rejection for Dynatrace OperationRateLimiter now emits an INFO log line on every rejection with user, spaceGuid, and reason in a fixed format that Dynatrace can parse as structured fields: Operation start rejected: user="" spaceGuid="" reason="" This covers all three rejection paths: per-space active-op cap, per-user active-op cap, and token-bucket exhaustion. Also restores bucket4j + testcontainers dependencies in the web module pom that were dropped during a prior rebase. --- multiapps-controller-web/pom.xml | 4 +++ .../multiapps/controller/web/Messages.java | 1 + .../web/util/OperationRateLimiter.java | 25 +++++++++++++------ 3 files changed, 22 insertions(+), 8 deletions(-) diff --git a/multiapps-controller-web/pom.xml b/multiapps-controller-web/pom.xml index 1c377426ed..5e180abdcc 100644 --- a/multiapps-controller-web/pom.xml +++ b/multiapps-controller-web/pom.xml @@ -151,6 +151,10 @@ io.github.resilience4j resilience4j-ratelimiter + + com.bucket4j + bucket4j_jdk17-postgresql + org.apache.jclouds.common googlecloud diff --git a/multiapps-controller-web/src/main/java/org/cloudfoundry/multiapps/controller/web/Messages.java b/multiapps-controller-web/src/main/java/org/cloudfoundry/multiapps/controller/web/Messages.java index 57175a2a82..8875564c41 100644 --- a/multiapps-controller-web/src/main/java/org/cloudfoundry/multiapps/controller/web/Messages.java +++ b/multiapps-controller-web/src/main/java/org/cloudfoundry/multiapps/controller/web/Messages.java @@ -91,6 +91,7 @@ public final class Messages { public static final String CREATING_ASYNC_UPLOAD_JOB = "Creating async upload job for URL {} with ID: {}"; public static final String ASYNC_UPLOAD_JOB_REJECTED = "Async upload job with space guid: {}, namespace: {}, URL: {} rejected."; public static final String OPERATION_START_RATE_LIMITED = "Start of operation in space {} rejected due to rate limiting: {}"; + public static final String OPERATION_START_RATE_LIMITED_STRUCTURED = "Operation start rejected: user=\"{0}\" spaceGuid=\"{1}\" reason=\"{2}\""; public static final String STARTING_DOWNLOAD_OF_MTAR_WITH_JOB_ID = "Starting download of MTAR from remote endpoint: {}. Job id: {}"; public static final String UPLOADED_MTAR_FROM_REMOTE_ENDPOINT_AND_JOB_ID = "Uploaded MTAR from remote endpoint {}. Job id: {} in {} ms"; public static final String ASYNC_UPLOAD_JOB_FINISHED = "Async upload job {} finished"; diff --git a/multiapps-controller-web/src/main/java/org/cloudfoundry/multiapps/controller/web/util/OperationRateLimiter.java b/multiapps-controller-web/src/main/java/org/cloudfoundry/multiapps/controller/web/util/OperationRateLimiter.java index 7745482e5f..544244c907 100644 --- a/multiapps-controller-web/src/main/java/org/cloudfoundry/multiapps/controller/web/util/OperationRateLimiter.java +++ b/multiapps-controller-web/src/main/java/org/cloudfoundry/multiapps/controller/web/util/OperationRateLimiter.java @@ -1,5 +1,6 @@ package org.cloudfoundry.multiapps.controller.web.util; +import java.text.MessageFormat; import java.time.Duration; import java.util.concurrent.TimeUnit; @@ -9,6 +10,8 @@ import org.cloudfoundry.multiapps.controller.persistence.services.OperationService; import org.cloudfoundry.multiapps.controller.process.util.BucketStore; import org.cloudfoundry.multiapps.controller.web.Messages; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; import io.github.bucket4j.Bandwidth; import io.github.bucket4j.Bucket; @@ -22,6 +25,7 @@ @Named public class OperationRateLimiter { + private static final Logger LOGGER = LoggerFactory.getLogger(OperationRateLimiter.class); private static final Duration REFILL_PERIOD = Duration.ofHours(1); private static final long TOKENS_PER_OPERATION = 1; private static final long NO_RETRY_AFTER_SECONDS = 0; @@ -45,6 +49,11 @@ public void checkStartAllowed(String user, String spaceGuid) { checkTokenBuckets(user, spaceGuid); } + private void rejectAndLog(String user, String spaceGuid, String reason, long retryAfterSeconds) { + LOGGER.info(MessageFormat.format(Messages.OPERATION_START_RATE_LIMITED_STRUCTURED, user, spaceGuid, reason)); + throw new OperationRateLimitExceededException(reason, retryAfterSeconds); + } + private void checkActiveOperationCaps(String user, String spaceGuid) { int activeOperationsPerSpace = operationService.createQuery() .spaceId(spaceGuid) @@ -52,7 +61,7 @@ private void checkActiveOperationCaps(String user, String spaceGuid) { .list() .size(); if (activeOperationsPerSpace >= applicationConfiguration.getMaxActiveOperationsPerSpace()) { - throw new OperationRateLimitExceededException(Messages.TOO_MANY_ACTIVE_OPERATIONS_IN_SPACE, NO_RETRY_AFTER_SECONDS); + rejectAndLog(user, spaceGuid, Messages.TOO_MANY_ACTIVE_OPERATIONS_IN_SPACE, NO_RETRY_AFTER_SECONDS); } int activeOperationsPerUser = operationService.createQuery() .user(user) @@ -61,27 +70,27 @@ private void checkActiveOperationCaps(String user, String spaceGuid) { .list() .size(); if (activeOperationsPerUser >= applicationConfiguration.getMaxActiveOperationsPerUser()) { - throw new OperationRateLimitExceededException(Messages.TOO_MANY_ACTIVE_OPERATIONS_FOR_USER, NO_RETRY_AFTER_SECONDS); + rejectAndLog(user, spaceGuid, Messages.TOO_MANY_ACTIVE_OPERATIONS_FOR_USER, NO_RETRY_AFTER_SECONDS); } } private void checkTokenBuckets(String user, String spaceGuid) { - consumeSpaceToken(spaceGuid); + consumeSpaceToken(spaceGuid, user); consumeUserToken(spaceGuid, user); } - private void consumeSpaceToken(String spaceGuid) { + private void consumeSpaceToken(String spaceGuid, String user) { BucketConfiguration configuration = buildBucketConfiguration(applicationConfiguration.getOperationRateLimitPerSpaceCapacity(), applicationConfiguration.getOperationRateLimitPerSpaceRefillPerHour()); Bucket bucket = bucketStore.getBucket(OperationRateLimitKeys.spaceKey(spaceGuid), configuration); - consumeToken(bucket); + consumeToken(bucket, user, spaceGuid); } private void consumeUserToken(String spaceGuid, String user) { BucketConfiguration configuration = buildBucketConfiguration(applicationConfiguration.getOperationRateLimitPerUserCapacity(), applicationConfiguration.getOperationRateLimitPerUserRefillPerHour()); Bucket bucket = bucketStore.getBucket(OperationRateLimitKeys.userKey(spaceGuid, user), configuration); - consumeToken(bucket); + consumeToken(bucket, user, spaceGuid); } private BucketConfiguration buildBucketConfiguration(int capacity, int refillTokensPerHour) { @@ -94,11 +103,11 @@ private BucketConfiguration buildBucketConfiguration(int capacity, int refillTok .build(); } - private void consumeToken(Bucket bucket) { + private void consumeToken(Bucket bucket, String user, String spaceGuid) { ConsumptionProbe probe = bucket.tryConsumeAndReturnRemaining(TOKENS_PER_OPERATION); if (!probe.isConsumed()) { long retryAfterSeconds = TimeUnit.NANOSECONDS.toSeconds(probe.getNanosToWaitForRefill()); - throw new OperationRateLimitExceededException(Messages.OPERATION_RATE_LIMIT_EXCEEDED, retryAfterSeconds); + rejectAndLog(user, spaceGuid, Messages.OPERATION_RATE_LIMIT_EXCEEDED, retryAfterSeconds); } } } From b3f70b3e0941eeaa994ef893dd2e704736dfc6ae Mon Sep 17 00:00:00 2001 From: Yavor16 Date: Wed, 9 Sep 2026 16:00:07 +0300 Subject: [PATCH 08/10] Expose operation rate-limit rejection counters as JMX MBean MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Adds OperationRateLimitMetricsMBean / OperationRateLimitMetrics with two attributes: - RateLimitRejectionCount — all-time total since last restart - RateLimitRejectionCountInWindow — rolling window counter (resets on each Dynatrace poll cycle, consistent with UploadDurationMetrics) OperationRateLimiter calls metrics.recordRejection() from the shared rejectAndLog helper so every rejection — active-op cap, space bucket, or user bucket — is counted. Registered in JmxConfiguration under the object name org.cloudfoundry.multiapps.controller.web.monitoring:type=Metrics,name=OperationRateLimitMetricsMBean. --- .../web/configuration/JmxConfiguration.java | 10 +++++-- .../monitoring/OperationRateLimitMetrics.java | 29 +++++++++++++++++++ .../OperationRateLimitMetricsMBean.java | 9 ++++++ .../web/util/OperationRateLimiter.java | 6 +++- .../web/util/OperationRateLimiterTest.java | 3 ++ 5 files changed, 54 insertions(+), 3 deletions(-) create mode 100644 multiapps-controller-web/src/main/java/org/cloudfoundry/multiapps/controller/web/monitoring/OperationRateLimitMetrics.java create mode 100644 multiapps-controller-web/src/main/java/org/cloudfoundry/multiapps/controller/web/monitoring/OperationRateLimitMetricsMBean.java diff --git a/multiapps-controller-web/src/main/java/org/cloudfoundry/multiapps/controller/web/configuration/JmxConfiguration.java b/multiapps-controller-web/src/main/java/org/cloudfoundry/multiapps/controller/web/configuration/JmxConfiguration.java index a8aca8efc2..8ca8cbf127 100644 --- a/multiapps-controller-web/src/main/java/org/cloudfoundry/multiapps/controller/web/configuration/JmxConfiguration.java +++ b/multiapps-controller-web/src/main/java/org/cloudfoundry/multiapps/controller/web/configuration/JmxConfiguration.java @@ -3,6 +3,7 @@ import java.util.Map; import org.cloudfoundry.multiapps.controller.web.monitoring.Metrics; +import org.cloudfoundry.multiapps.controller.web.monitoring.OperationRateLimitMetrics; import org.cloudfoundry.multiapps.controller.web.monitoring.UploadDurationMetrics; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; @@ -15,10 +16,15 @@ public class JmxConfiguration { private static final String UPLOAD_METRICS_BEAN = "org.cloudfoundry.multiapps.controller.web.monitoring:type=Metrics,name=UploadMetricsMBean"; + private static final String RATE_LIMIT_METRICS_BEAN = "org.cloudfoundry.multiapps.controller.web.monitoring:type=Metrics,name=OperationRateLimitMetricsMBean"; + @Bean - public MBeanExporter jmxExporter(Metrics metrics, UploadDurationMetrics uploadDurationMetrics) { + public MBeanExporter jmxExporter(Metrics metrics, UploadDurationMetrics uploadDurationMetrics, + OperationRateLimitMetrics operationRateLimitMetrics) { MBeanExporter mBeanExporter = new MBeanExporter(); - mBeanExporter.setBeans(Map.of(METRICS_BEAN, metrics, UPLOAD_METRICS_BEAN, uploadDurationMetrics)); + mBeanExporter.setBeans(Map.of(METRICS_BEAN, metrics, + UPLOAD_METRICS_BEAN, uploadDurationMetrics, + RATE_LIMIT_METRICS_BEAN, operationRateLimitMetrics)); return mBeanExporter; } } diff --git a/multiapps-controller-web/src/main/java/org/cloudfoundry/multiapps/controller/web/monitoring/OperationRateLimitMetrics.java b/multiapps-controller-web/src/main/java/org/cloudfoundry/multiapps/controller/web/monitoring/OperationRateLimitMetrics.java new file mode 100644 index 0000000000..aa49755d36 --- /dev/null +++ b/multiapps-controller-web/src/main/java/org/cloudfoundry/multiapps/controller/web/monitoring/OperationRateLimitMetrics.java @@ -0,0 +1,29 @@ +package org.cloudfoundry.multiapps.controller.web.monitoring; + +import java.util.concurrent.atomic.AtomicLong; +import java.util.concurrent.atomic.LongAdder; + +import jakarta.inject.Named; + +@Named +public class OperationRateLimitMetrics implements OperationRateLimitMetricsMBean { + + private final LongAdder totalRejections = new LongAdder(); + private final AtomicLong rejectionsInWindow = new AtomicLong(0); + + public void recordRejection() { + totalRejections.increment(); + rejectionsInWindow.incrementAndGet(); + } + + @Override + public long getRateLimitRejectionCount() { + return totalRejections.sum(); + } + + @Override + public long getRateLimitRejectionCountInWindow() { + return rejectionsInWindow.getAndSet(0); + } + +} diff --git a/multiapps-controller-web/src/main/java/org/cloudfoundry/multiapps/controller/web/monitoring/OperationRateLimitMetricsMBean.java b/multiapps-controller-web/src/main/java/org/cloudfoundry/multiapps/controller/web/monitoring/OperationRateLimitMetricsMBean.java new file mode 100644 index 0000000000..adfb93bb71 --- /dev/null +++ b/multiapps-controller-web/src/main/java/org/cloudfoundry/multiapps/controller/web/monitoring/OperationRateLimitMetricsMBean.java @@ -0,0 +1,9 @@ +package org.cloudfoundry.multiapps.controller.web.monitoring; + +public interface OperationRateLimitMetricsMBean { + + long getRateLimitRejectionCount(); + + long getRateLimitRejectionCountInWindow(); + +} diff --git a/multiapps-controller-web/src/main/java/org/cloudfoundry/multiapps/controller/web/util/OperationRateLimiter.java b/multiapps-controller-web/src/main/java/org/cloudfoundry/multiapps/controller/web/util/OperationRateLimiter.java index 544244c907..0b61b1c06e 100644 --- a/multiapps-controller-web/src/main/java/org/cloudfoundry/multiapps/controller/web/util/OperationRateLimiter.java +++ b/multiapps-controller-web/src/main/java/org/cloudfoundry/multiapps/controller/web/util/OperationRateLimiter.java @@ -10,6 +10,7 @@ import org.cloudfoundry.multiapps.controller.persistence.services.OperationService; import org.cloudfoundry.multiapps.controller.process.util.BucketStore; import org.cloudfoundry.multiapps.controller.web.Messages; +import org.cloudfoundry.multiapps.controller.web.monitoring.OperationRateLimitMetrics; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -33,12 +34,14 @@ public class OperationRateLimiter { private final ApplicationConfiguration applicationConfiguration; private final OperationService operationService; private final BucketStore bucketStore; + private final OperationRateLimitMetrics metrics; public OperationRateLimiter(ApplicationConfiguration applicationConfiguration, OperationService operationService, - BucketStore bucketStore) { + BucketStore bucketStore, OperationRateLimitMetrics metrics) { this.applicationConfiguration = applicationConfiguration; this.operationService = operationService; this.bucketStore = bucketStore; + this.metrics = metrics; } public void checkStartAllowed(String user, String spaceGuid) { @@ -51,6 +54,7 @@ public void checkStartAllowed(String user, String spaceGuid) { private void rejectAndLog(String user, String spaceGuid, String reason, long retryAfterSeconds) { LOGGER.info(MessageFormat.format(Messages.OPERATION_START_RATE_LIMITED_STRUCTURED, user, spaceGuid, reason)); + metrics.recordRejection(); throw new OperationRateLimitExceededException(reason, retryAfterSeconds); } diff --git a/multiapps-controller-web/src/test/java/org/cloudfoundry/multiapps/controller/web/util/OperationRateLimiterTest.java b/multiapps-controller-web/src/test/java/org/cloudfoundry/multiapps/controller/web/util/OperationRateLimiterTest.java index 5eb3599022..7011cc902f 100644 --- a/multiapps-controller-web/src/test/java/org/cloudfoundry/multiapps/controller/web/util/OperationRateLimiterTest.java +++ b/multiapps-controller-web/src/test/java/org/cloudfoundry/multiapps/controller/web/util/OperationRateLimiterTest.java @@ -5,6 +5,7 @@ import org.cloudfoundry.multiapps.controller.persistence.query.OperationQuery; import org.cloudfoundry.multiapps.controller.persistence.services.OperationService; import org.cloudfoundry.multiapps.controller.process.util.BucketStore; +import org.cloudfoundry.multiapps.controller.web.monitoring.OperationRateLimitMetrics; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; import org.mockito.InjectMocks; @@ -49,6 +50,8 @@ class OperationRateLimiterTest { @Mock private BucketStore bucketStore; @Mock + private OperationRateLimitMetrics metrics; + @Mock private Bucket spaceBucket; @Mock private Bucket userBucket; From 08b0c76c769ad6e57884cb3ceb04bb057640d1b5 Mon Sep 17 00:00:00 2001 From: Yavor16 Date: Fri, 11 Sep 2026 13:29:28 +0300 Subject: [PATCH 09/10] Finish implemenation --- .../web/configuration/JmxConfiguration.java | 16 +-- .../web/monitoring/ActiveOperationsCount.java | 20 ++++ .../ActiveOperationsCountMBean.java | 7 ++ .../ActiveOperationsJmxReporter.java | 97 +++++++++++++++++ .../monitoring/OperationRateLimitMetrics.java | 29 ----- .../OperationRateLimitMetricsMBean.java | 9 -- .../web/util/OperationRateLimiter.java | 6 +- .../ActiveOperationsJmxReporterTest.java | 101 ++++++++++++++++++ .../web/util/OperationRateLimiterTest.java | 3 - 9 files changed, 236 insertions(+), 52 deletions(-) create mode 100644 multiapps-controller-web/src/main/java/org/cloudfoundry/multiapps/controller/web/monitoring/ActiveOperationsCount.java create mode 100644 multiapps-controller-web/src/main/java/org/cloudfoundry/multiapps/controller/web/monitoring/ActiveOperationsCountMBean.java create mode 100644 multiapps-controller-web/src/main/java/org/cloudfoundry/multiapps/controller/web/monitoring/ActiveOperationsJmxReporter.java delete mode 100644 multiapps-controller-web/src/main/java/org/cloudfoundry/multiapps/controller/web/monitoring/OperationRateLimitMetrics.java delete mode 100644 multiapps-controller-web/src/main/java/org/cloudfoundry/multiapps/controller/web/monitoring/OperationRateLimitMetricsMBean.java create mode 100644 multiapps-controller-web/src/test/java/org/cloudfoundry/multiapps/controller/web/monitoring/ActiveOperationsJmxReporterTest.java diff --git a/multiapps-controller-web/src/main/java/org/cloudfoundry/multiapps/controller/web/configuration/JmxConfiguration.java b/multiapps-controller-web/src/main/java/org/cloudfoundry/multiapps/controller/web/configuration/JmxConfiguration.java index 8ca8cbf127..0ec32c40eb 100644 --- a/multiapps-controller-web/src/main/java/org/cloudfoundry/multiapps/controller/web/configuration/JmxConfiguration.java +++ b/multiapps-controller-web/src/main/java/org/cloudfoundry/multiapps/controller/web/configuration/JmxConfiguration.java @@ -1,9 +1,11 @@ package org.cloudfoundry.multiapps.controller.web.configuration; +import java.lang.management.ManagementFactory; import java.util.Map; +import javax.management.MBeanServer; + import org.cloudfoundry.multiapps.controller.web.monitoring.Metrics; -import org.cloudfoundry.multiapps.controller.web.monitoring.OperationRateLimitMetrics; import org.cloudfoundry.multiapps.controller.web.monitoring.UploadDurationMetrics; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; @@ -16,15 +18,17 @@ public class JmxConfiguration { private static final String UPLOAD_METRICS_BEAN = "org.cloudfoundry.multiapps.controller.web.monitoring:type=Metrics,name=UploadMetricsMBean"; - private static final String RATE_LIMIT_METRICS_BEAN = "org.cloudfoundry.multiapps.controller.web.monitoring:type=Metrics,name=OperationRateLimitMetricsMBean"; + @Bean + public MBeanServer mBeanServer() { + return ManagementFactory.getPlatformMBeanServer(); + } @Bean - public MBeanExporter jmxExporter(Metrics metrics, UploadDurationMetrics uploadDurationMetrics, - OperationRateLimitMetrics operationRateLimitMetrics) { + public MBeanExporter jmxExporter(Metrics metrics, UploadDurationMetrics uploadDurationMetrics) { MBeanExporter mBeanExporter = new MBeanExporter(); + mBeanExporter.setServer(mBeanServer()); mBeanExporter.setBeans(Map.of(METRICS_BEAN, metrics, - UPLOAD_METRICS_BEAN, uploadDurationMetrics, - RATE_LIMIT_METRICS_BEAN, operationRateLimitMetrics)); + UPLOAD_METRICS_BEAN, uploadDurationMetrics)); return mBeanExporter; } } diff --git a/multiapps-controller-web/src/main/java/org/cloudfoundry/multiapps/controller/web/monitoring/ActiveOperationsCount.java b/multiapps-controller-web/src/main/java/org/cloudfoundry/multiapps/controller/web/monitoring/ActiveOperationsCount.java new file mode 100644 index 0000000000..773668d8e3 --- /dev/null +++ b/multiapps-controller-web/src/main/java/org/cloudfoundry/multiapps/controller/web/monitoring/ActiveOperationsCount.java @@ -0,0 +1,20 @@ +package org.cloudfoundry.multiapps.controller.web.monitoring; + +public class ActiveOperationsCount implements ActiveOperationsCountMBean { + + private volatile long count; + + public ActiveOperationsCount(long count) { + this.count = count; + } + + public void setCount(long count) { + this.count = count; + } + + @Override + public long getCount() { + return count; + } + +} diff --git a/multiapps-controller-web/src/main/java/org/cloudfoundry/multiapps/controller/web/monitoring/ActiveOperationsCountMBean.java b/multiapps-controller-web/src/main/java/org/cloudfoundry/multiapps/controller/web/monitoring/ActiveOperationsCountMBean.java new file mode 100644 index 0000000000..b2901e0245 --- /dev/null +++ b/multiapps-controller-web/src/main/java/org/cloudfoundry/multiapps/controller/web/monitoring/ActiveOperationsCountMBean.java @@ -0,0 +1,7 @@ +package org.cloudfoundry.multiapps.controller.web.monitoring; + +public interface ActiveOperationsCountMBean { + + long getCount(); + +} diff --git a/multiapps-controller-web/src/main/java/org/cloudfoundry/multiapps/controller/web/monitoring/ActiveOperationsJmxReporter.java b/multiapps-controller-web/src/main/java/org/cloudfoundry/multiapps/controller/web/monitoring/ActiveOperationsJmxReporter.java new file mode 100644 index 0000000000..bf740cd55b --- /dev/null +++ b/multiapps-controller-web/src/main/java/org/cloudfoundry/multiapps/controller/web/monitoring/ActiveOperationsJmxReporter.java @@ -0,0 +1,97 @@ +package org.cloudfoundry.multiapps.controller.web.monitoring; + +import java.nio.charset.StandardCharsets; +import java.security.MessageDigest; +import java.security.NoSuchAlgorithmException; +import java.util.HashMap; +import java.util.HashSet; +import java.util.HexFormat; +import java.util.Map; +import java.util.Set; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.TimeUnit; + +import jakarta.inject.Inject; +import jakarta.inject.Named; + +import javax.management.MBeanServer; +import javax.management.ObjectName; + +import org.cloudfoundry.multiapps.controller.persistence.services.OperationService; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.springframework.scheduling.annotation.Scheduled; + +@Named +public class ActiveOperationsJmxReporter { + + static final String DOMAIN = "org.cloudfoundry.multiapps.controller.web.monitoring"; + static final String USER_OBJECT_NAME_PATTERN = DOMAIN + ":type=Metrics,name=ActiveOps,user=%s"; + static final String SPACE_OBJECT_NAME_PATTERN = DOMAIN + ":type=Metrics,name=ActiveOps,space=%s"; + + private static final Logger LOGGER = LoggerFactory.getLogger(ActiveOperationsJmxReporter.class); + + private final OperationService operationService; + private final MBeanServer mBeanServer; + private final Map userBeans = new ConcurrentHashMap<>(); + private final Map spaceBeans = new ConcurrentHashMap<>(); + + @Inject + public ActiveOperationsJmxReporter(OperationService operationService, MBeanServer mBeanServer) { + this.operationService = operationService; + this.mBeanServer = mBeanServer; + } + + @Scheduled(fixedRate = 1, timeUnit = TimeUnit.MINUTES) + public void refresh() { + try { + Map byUser = new HashMap<>(); + Map bySpace = new HashMap<>(); + operationService.createQuery() + .inNonFinalState() + .list() + .forEach(op -> { + byUser.merge(hashUser(op.getUser()), 1L, Long::sum); + bySpace.merge(op.getSpaceId(), 1L, Long::sum); + }); + syncMBeans(byUser, USER_OBJECT_NAME_PATTERN, userBeans); + syncMBeans(bySpace, SPACE_OBJECT_NAME_PATTERN, spaceBeans); + } catch (Exception e) { + LOGGER.warn("Failed to refresh active operations JMX metrics", e); + } + } + + private void syncMBeans(Map counts, String pattern, + Map registry) throws Exception { + Set toRemove = new HashSet<>(registry.keySet()); + + for (Map.Entry entry : counts.entrySet()) { + ObjectName name = new ObjectName(pattern.formatted(ObjectName.quote(entry.getKey()))); + toRemove.remove(name); + ActiveOperationsCount bean = registry.get(name); + if (bean == null) { + bean = new ActiveOperationsCount(entry.getValue()); + mBeanServer.registerMBean(bean, name); + registry.put(name, bean); + } else { + bean.setCount(entry.getValue()); + } + } + + for (ObjectName name : toRemove) { + mBeanServer.unregisterMBean(name); + registry.remove(name); + } + } + + static String hashUser(String user) { + try { + byte[] digest = MessageDigest.getInstance("SHA-256") + .digest(user.getBytes(StandardCharsets.UTF_8)); + return HexFormat.of().formatHex(digest); + } catch (NoSuchAlgorithmException e) { + throw new IllegalStateException(e); + } + } + +} diff --git a/multiapps-controller-web/src/main/java/org/cloudfoundry/multiapps/controller/web/monitoring/OperationRateLimitMetrics.java b/multiapps-controller-web/src/main/java/org/cloudfoundry/multiapps/controller/web/monitoring/OperationRateLimitMetrics.java deleted file mode 100644 index aa49755d36..0000000000 --- a/multiapps-controller-web/src/main/java/org/cloudfoundry/multiapps/controller/web/monitoring/OperationRateLimitMetrics.java +++ /dev/null @@ -1,29 +0,0 @@ -package org.cloudfoundry.multiapps.controller.web.monitoring; - -import java.util.concurrent.atomic.AtomicLong; -import java.util.concurrent.atomic.LongAdder; - -import jakarta.inject.Named; - -@Named -public class OperationRateLimitMetrics implements OperationRateLimitMetricsMBean { - - private final LongAdder totalRejections = new LongAdder(); - private final AtomicLong rejectionsInWindow = new AtomicLong(0); - - public void recordRejection() { - totalRejections.increment(); - rejectionsInWindow.incrementAndGet(); - } - - @Override - public long getRateLimitRejectionCount() { - return totalRejections.sum(); - } - - @Override - public long getRateLimitRejectionCountInWindow() { - return rejectionsInWindow.getAndSet(0); - } - -} diff --git a/multiapps-controller-web/src/main/java/org/cloudfoundry/multiapps/controller/web/monitoring/OperationRateLimitMetricsMBean.java b/multiapps-controller-web/src/main/java/org/cloudfoundry/multiapps/controller/web/monitoring/OperationRateLimitMetricsMBean.java deleted file mode 100644 index adfb93bb71..0000000000 --- a/multiapps-controller-web/src/main/java/org/cloudfoundry/multiapps/controller/web/monitoring/OperationRateLimitMetricsMBean.java +++ /dev/null @@ -1,9 +0,0 @@ -package org.cloudfoundry.multiapps.controller.web.monitoring; - -public interface OperationRateLimitMetricsMBean { - - long getRateLimitRejectionCount(); - - long getRateLimitRejectionCountInWindow(); - -} diff --git a/multiapps-controller-web/src/main/java/org/cloudfoundry/multiapps/controller/web/util/OperationRateLimiter.java b/multiapps-controller-web/src/main/java/org/cloudfoundry/multiapps/controller/web/util/OperationRateLimiter.java index 0b61b1c06e..544244c907 100644 --- a/multiapps-controller-web/src/main/java/org/cloudfoundry/multiapps/controller/web/util/OperationRateLimiter.java +++ b/multiapps-controller-web/src/main/java/org/cloudfoundry/multiapps/controller/web/util/OperationRateLimiter.java @@ -10,7 +10,6 @@ import org.cloudfoundry.multiapps.controller.persistence.services.OperationService; import org.cloudfoundry.multiapps.controller.process.util.BucketStore; import org.cloudfoundry.multiapps.controller.web.Messages; -import org.cloudfoundry.multiapps.controller.web.monitoring.OperationRateLimitMetrics; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -34,14 +33,12 @@ public class OperationRateLimiter { private final ApplicationConfiguration applicationConfiguration; private final OperationService operationService; private final BucketStore bucketStore; - private final OperationRateLimitMetrics metrics; public OperationRateLimiter(ApplicationConfiguration applicationConfiguration, OperationService operationService, - BucketStore bucketStore, OperationRateLimitMetrics metrics) { + BucketStore bucketStore) { this.applicationConfiguration = applicationConfiguration; this.operationService = operationService; this.bucketStore = bucketStore; - this.metrics = metrics; } public void checkStartAllowed(String user, String spaceGuid) { @@ -54,7 +51,6 @@ public void checkStartAllowed(String user, String spaceGuid) { private void rejectAndLog(String user, String spaceGuid, String reason, long retryAfterSeconds) { LOGGER.info(MessageFormat.format(Messages.OPERATION_START_RATE_LIMITED_STRUCTURED, user, spaceGuid, reason)); - metrics.recordRejection(); throw new OperationRateLimitExceededException(reason, retryAfterSeconds); } diff --git a/multiapps-controller-web/src/test/java/org/cloudfoundry/multiapps/controller/web/monitoring/ActiveOperationsJmxReporterTest.java b/multiapps-controller-web/src/test/java/org/cloudfoundry/multiapps/controller/web/monitoring/ActiveOperationsJmxReporterTest.java new file mode 100644 index 0000000000..59a83b6a59 --- /dev/null +++ b/multiapps-controller-web/src/test/java/org/cloudfoundry/multiapps/controller/web/monitoring/ActiveOperationsJmxReporterTest.java @@ -0,0 +1,101 @@ +package org.cloudfoundry.multiapps.controller.web.monitoring; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; + +import java.util.List; + +import javax.management.MBeanServer; +import javax.management.MBeanServerFactory; +import javax.management.ObjectName; + +import org.cloudfoundry.multiapps.controller.api.model.Operation; +import org.cloudfoundry.multiapps.controller.persistence.query.OperationQuery; +import org.cloudfoundry.multiapps.controller.persistence.services.OperationService; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.mockito.Mock; +import org.mockito.MockitoAnnotations; + +class ActiveOperationsJmxReporterTest { + + @Mock + private OperationService operationService; + + private MBeanServer mBeanServer; + private ActiveOperationsJmxReporter reporter; + private AutoCloseable mocks; + + @BeforeEach + void setUp() { + mocks = MockitoAnnotations.openMocks(this); + mBeanServer = MBeanServerFactory.createMBeanServer(); + reporter = new ActiveOperationsJmxReporter(operationService, mBeanServer); + } + + @AfterEach + void tearDown() throws Exception { + MBeanServerFactory.releaseMBeanServer(mBeanServer); + mocks.close(); + } + + @Test + void testRegistersMBeansPerUserAndSpace() throws Exception { + stubOperations(List.of(operation("alice@sap.com", "space-1"), + operation("alice@sap.com", "space-2"), + operation("bob@sap.com", "space-1"))); + + reporter.refresh(); + + assertEquals(2L, getCount(ActiveOperationsJmxReporter.USER_OBJECT_NAME_PATTERN, ActiveOperationsJmxReporter.hashUser("alice@sap.com"))); + assertEquals(1L, getCount(ActiveOperationsJmxReporter.USER_OBJECT_NAME_PATTERN, ActiveOperationsJmxReporter.hashUser("bob@sap.com"))); + assertEquals(2L, getCount(ActiveOperationsJmxReporter.SPACE_OBJECT_NAME_PATTERN, "space-1")); + assertEquals(1L, getCount(ActiveOperationsJmxReporter.SPACE_OBJECT_NAME_PATTERN, "space-2")); + } + + @Test + void testUnregistersMBeanWhenUserHasNoMoreActiveOperations() throws Exception { + stubOperations(List.of(operation("alice@sap.com", "space-1"))); + reporter.refresh(); + + stubOperations(List.of()); + reporter.refresh(); + + ObjectName name = new ObjectName(ActiveOperationsJmxReporter.USER_OBJECT_NAME_PATTERN.formatted(ObjectName.quote(ActiveOperationsJmxReporter.hashUser("alice@sap.com")))); + assertEquals(false, mBeanServer.isRegistered(name)); + } + + @Test + void testUpdatesCountWhenOperationsChange() throws Exception { + stubOperations(List.of(operation("alice@sap.com", "space-1"))); + reporter.refresh(); + assertEquals(1L, getCount(ActiveOperationsJmxReporter.USER_OBJECT_NAME_PATTERN, ActiveOperationsJmxReporter.hashUser("alice@sap.com"))); + + stubOperations(List.of(operation("alice@sap.com", "space-1"), + operation("alice@sap.com", "space-2"))); + reporter.refresh(); + assertEquals(2L, getCount(ActiveOperationsJmxReporter.USER_OBJECT_NAME_PATTERN, ActiveOperationsJmxReporter.hashUser("alice@sap.com"))); + } + + private void stubOperations(List operations) { + OperationQuery query = mock(OperationQuery.class); + when(operationService.createQuery()).thenReturn(query); + when(query.inNonFinalState()).thenReturn(query); + when(query.list()).thenReturn(operations); + } + + private long getCount(String pattern, String key) throws Exception { + ObjectName name = new ObjectName(pattern.formatted(ObjectName.quote(key))); + return (long) mBeanServer.getAttribute(name, "Count"); + } + + private Operation operation(String user, String spaceId) { + Operation op = mock(Operation.class); + when(op.getUser()).thenReturn(user); + when(op.getSpaceId()).thenReturn(spaceId); + return op; + } + +} diff --git a/multiapps-controller-web/src/test/java/org/cloudfoundry/multiapps/controller/web/util/OperationRateLimiterTest.java b/multiapps-controller-web/src/test/java/org/cloudfoundry/multiapps/controller/web/util/OperationRateLimiterTest.java index 7011cc902f..5eb3599022 100644 --- a/multiapps-controller-web/src/test/java/org/cloudfoundry/multiapps/controller/web/util/OperationRateLimiterTest.java +++ b/multiapps-controller-web/src/test/java/org/cloudfoundry/multiapps/controller/web/util/OperationRateLimiterTest.java @@ -5,7 +5,6 @@ import org.cloudfoundry.multiapps.controller.persistence.query.OperationQuery; import org.cloudfoundry.multiapps.controller.persistence.services.OperationService; import org.cloudfoundry.multiapps.controller.process.util.BucketStore; -import org.cloudfoundry.multiapps.controller.web.monitoring.OperationRateLimitMetrics; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; import org.mockito.InjectMocks; @@ -50,8 +49,6 @@ class OperationRateLimiterTest { @Mock private BucketStore bucketStore; @Mock - private OperationRateLimitMetrics metrics; - @Mock private Bucket spaceBucket; @Mock private Bucket userBucket; From 138ade3cc6b05a76bf39922391f5895b4d83dc4e Mon Sep 17 00:00:00 2001 From: Yavor16 Date: Fri, 11 Sep 2026 13:55:51 +0300 Subject: [PATCH 10/10] Remove unused dependnecies --- multiapps-controller-process/pom.xml | 15 -- .../OperationRateLimiterIntegrationTest.java | 174 ------------------ .../ActiveOperationsJmxReporter.java | 63 ++++--- pom.xml | 20 -- 4 files changed, 40 insertions(+), 232 deletions(-) delete mode 100644 multiapps-controller-process/src/test/java/org/cloudfoundry/multiapps/controller/process/util/OperationRateLimiterIntegrationTest.java diff --git a/multiapps-controller-process/pom.xml b/multiapps-controller-process/pom.xml index a94e07f8bf..6a77ba2163 100644 --- a/multiapps-controller-process/pom.xml +++ b/multiapps-controller-process/pom.xml @@ -131,20 +131,5 @@ com.bucket4j bucket4j_jdk17-postgresql - - org.testcontainers - postgresql - test - - - org.testcontainers - junit-jupiter - test - - - org.postgresql - postgresql - test - \ No newline at end of file diff --git a/multiapps-controller-process/src/test/java/org/cloudfoundry/multiapps/controller/process/util/OperationRateLimiterIntegrationTest.java b/multiapps-controller-process/src/test/java/org/cloudfoundry/multiapps/controller/process/util/OperationRateLimiterIntegrationTest.java deleted file mode 100644 index dbf171ee03..0000000000 --- a/multiapps-controller-process/src/test/java/org/cloudfoundry/multiapps/controller/process/util/OperationRateLimiterIntegrationTest.java +++ /dev/null @@ -1,174 +0,0 @@ -package org.cloudfoundry.multiapps.controller.process.util; - -import static org.junit.jupiter.api.Assertions.assertEquals; -import static org.junit.jupiter.api.Assertions.assertFalse; -import static org.junit.jupiter.api.Assertions.assertNotNull; -import static org.junit.jupiter.api.Assertions.assertTrue; - -import java.sql.Connection; -import java.sql.ResultSet; -import java.time.Duration; - -import javax.sql.DataSource; - -import org.junit.jupiter.api.BeforeEach; -import org.junit.jupiter.api.Test; -import org.postgresql.ds.PGSimpleDataSource; -import org.testcontainers.containers.PostgreSQLContainer; -import org.testcontainers.junit.jupiter.Container; -import org.testcontainers.junit.jupiter.Testcontainers; - -import io.github.bucket4j.Bandwidth; -import io.github.bucket4j.BucketConfiguration; -import io.github.bucket4j.ConsumptionProbe; -import io.github.bucket4j.distributed.BucketProxy; -import io.github.bucket4j.distributed.proxy.ProxyManager; -import io.github.bucket4j.postgresql.Bucket4jPostgreSQL; -import liquibase.Liquibase; -import liquibase.database.Database; -import liquibase.database.DatabaseFactory; -import liquibase.database.jvm.JdbcConnection; -import liquibase.resource.ClassLoaderResourceAccessor; - -/** - * Proves the real PostgreSQL SELECT ... FOR UPDATE round-trip behind {@link PostgresBucketStore}: token consumption is coordinated through - * the shared database, so two controller instances pointed at the same {@code operation_rate_limit_bucket} table drain a single logical - * bucket. A real PostgreSQL instance is required, so this runs under failsafe (name ends in {@code IntegrationTest}) and needs Docker. - */ -@Testcontainers -class OperationRateLimiterIntegrationTest { - - private static final String BUCKET_TABLE_NAME = "operation_rate_limit_bucket"; - // Step 1's changeset lives in this changelog; running it here validates the exact DDL the application uses. - private static final String BUCKET_CHANGELOG_LOCATION = "org/cloudfoundry/multiapps/controller/persistence/db/changelog/db-changelog-2.52.0-persistence.xml"; - // The master db-changelog.xml resolves this property per-dbms; on PostgreSQL it maps to BYTEA. We supply it directly because we run only - // the 2.52.0 changelog, not the master that declares the property. - private static final String SMALL_BLOB_TYPE_PARAMETER = "small-blob.type"; - private static final String POSTGRES_SMALL_BLOB_TYPE = "BYTEA"; - - private static final long BUCKET_CAPACITY = 3; - private static final long TOKENS_PER_CONSUME = 1; - // A refill window far larger than the test runtime guarantees no tokens are replenished mid-test, keeping the capacity assertions - // deterministic without any sleeps. - private static final Duration REFILL_WINDOW = Duration.ofHours(1); - - private static final long OPERATION_KEY = 42L; - private static final long OTHER_OPERATION_KEY = 43L; - - @Container - private final PostgreSQLContainer postgres = new PostgreSQLContainer<>("postgres:16"); - - private DataSource dataSource; - - @BeforeEach - void setUp() throws Exception { - dataSource = createDataSource(); - applyBucketSchema(); - } - - private DataSource createDataSource() { - var pgDataSource = new PGSimpleDataSource(); - pgDataSource.setUrl(postgres.getJdbcUrl()); - pgDataSource.setUser(postgres.getUsername()); - pgDataSource.setPassword(postgres.getPassword()); - return pgDataSource; - } - - private void applyBucketSchema() throws Exception { - try (Connection connection = dataSource.getConnection()) { - Database database = DatabaseFactory.getInstance() - .findCorrectDatabaseImplementation(new JdbcConnection(connection)); - try (Liquibase liquibase = new Liquibase(BUCKET_CHANGELOG_LOCATION, new ClassLoaderResourceAccessor(), database)) { - liquibase.setChangeLogParameter(SMALL_BLOB_TYPE_PARAMETER, POSTGRES_SMALL_BLOB_TYPE); - liquibase.update(""); - } - } - } - - @Test - void testConsumingAcrossTwoInstancesDrainsSharedBucket() { - var firstInstance = buildProxyManager(); - var secondInstance = buildProxyManager(); - - // Drain the shared bucket by splitting the capacity across two independent proxy managers over the same database. - assertTrue(consumeOneToken(firstInstance, OPERATION_KEY), "first token (instance one) should be consumed"); - assertTrue(consumeOneToken(secondInstance, OPERATION_KEY), "second token (instance two) should be consumed"); - assertTrue(consumeOneToken(firstInstance, OPERATION_KEY), "third token (instance one) should be consumed"); - - // The bucket is now empty. A further consume from EITHER instance must be rejected, proving the SELECT ... FOR UPDATE state is - // shared through Postgres rather than held per-instance. - assertFalse(consumeOneToken(secondInstance, OPERATION_KEY), "instance two must see the shared bucket as drained"); - assertFalse(consumeOneToken(firstInstance, OPERATION_KEY), "instance one must see the shared bucket as drained"); - } - - @Test - void testDifferentKeysAreIndependentBuckets() { - var firstInstance = buildProxyManager(); - var secondInstance = buildProxyManager(); - - for (var token = 0; token < BUCKET_CAPACITY; token++) { - assertTrue(consumeOneToken(firstInstance, OPERATION_KEY), "draining the bucket for the first key should succeed"); - } - assertFalse(consumeOneToken(firstInstance, OPERATION_KEY), "the first key's bucket should be drained"); - - // A different key resolves to a different row, so its bucket is untouched even from the other instance. - assertTrue(consumeOneToken(secondInstance, OTHER_OPERATION_KEY), "a different key must have its own independent bucket"); - } - - @Test - void testBucketStateRowIsPersistedWithExpiresAt() throws Exception { - var proxyManager = buildProxyManager(); - - assertTrue(consumeOneToken(proxyManager, OPERATION_KEY), "the first consume should persist bucket state"); - - assertBucketRowPersisted(OPERATION_KEY); - } - - private ProxyManager buildProxyManager() { - return Bucket4jPostgreSQL.selectForUpdateBasedBuilder(dataSource) - .table(BUCKET_TABLE_NAME) - .build(); - } - - private boolean consumeOneToken(ProxyManager proxyManager, long key) { - BucketProxy bucket = proxyManager.getProxy(key, this::buildBucketConfiguration); - ConsumptionProbe probe = bucket.tryConsumeAndReturnRemaining(TOKENS_PER_CONSUME); - return probe.isConsumed(); - } - - private BucketConfiguration buildBucketConfiguration() { - Bandwidth bandwidth = Bandwidth.builder() - .capacity(BUCKET_CAPACITY) - .refillGreedy(BUCKET_CAPACITY, REFILL_WINDOW) - .build(); - return BucketConfiguration.builder() - .addLimit(bandwidth) - .build(); - } - - private void assertBucketRowPersisted(long key) throws Exception { - var selectRow = "SELECT state, expires_at FROM " + BUCKET_TABLE_NAME + " WHERE id = ?"; - try (Connection connection = dataSource.getConnection(); - var statement = connection.prepareStatement(selectRow)) { - statement.setLong(1, key); - try (ResultSet resultSet = statement.executeQuery()) { - assertTrue(resultSet.next(), "a bucket row must be persisted for the consumed key"); - assertNotNull(resultSet.getBytes("state"), "the serialized bucket state must be stored"); - var expiresAt = resultSet.getLong("expires_at"); - assertFalse(resultSet.wasNull(), "expires_at must be populated so bucket4j can expire stale rows"); - assertTrue(expiresAt > 0, "expires_at must be a positive epoch value"); - assertFalse(resultSet.next(), "a single key must map to exactly one bucket row"); - } - } - assertEquals(1, countBucketRows(), "only the consumed key's bucket row should exist"); - } - - private long countBucketRows() throws Exception { - try (Connection connection = dataSource.getConnection(); - var statement = connection.prepareStatement("SELECT COUNT(*) FROM " + BUCKET_TABLE_NAME); - ResultSet resultSet = statement.executeQuery()) { - resultSet.next(); - return resultSet.getLong(1); - } - } -} diff --git a/multiapps-controller-web/src/main/java/org/cloudfoundry/multiapps/controller/web/monitoring/ActiveOperationsJmxReporter.java b/multiapps-controller-web/src/main/java/org/cloudfoundry/multiapps/controller/web/monitoring/ActiveOperationsJmxReporter.java index bf740cd55b..de04773329 100644 --- a/multiapps-controller-web/src/main/java/org/cloudfoundry/multiapps/controller/web/monitoring/ActiveOperationsJmxReporter.java +++ b/multiapps-controller-web/src/main/java/org/cloudfoundry/multiapps/controller/web/monitoring/ActiveOperationsJmxReporter.java @@ -45,40 +45,57 @@ public ActiveOperationsJmxReporter(OperationService operationService, MBeanServe @Scheduled(fixedRate = 1, timeUnit = TimeUnit.MINUTES) public void refresh() { try { - Map byUser = new HashMap<>(); - Map bySpace = new HashMap<>(); - operationService.createQuery() - .inNonFinalState() - .list() - .forEach(op -> { - byUser.merge(hashUser(op.getUser()), 1L, Long::sum); - bySpace.merge(op.getSpaceId(), 1L, Long::sum); - }); - syncMBeans(byUser, USER_OBJECT_NAME_PATTERN, userBeans); - syncMBeans(bySpace, SPACE_OBJECT_NAME_PATTERN, spaceBeans); + syncMBeans(countActiveOperationsByUser(), USER_OBJECT_NAME_PATTERN, userBeans); + syncMBeans(countActiveOperationsBySpace(), SPACE_OBJECT_NAME_PATTERN, spaceBeans); } catch (Exception e) { LOGGER.warn("Failed to refresh active operations JMX metrics", e); } } + private Map countActiveOperationsByUser() { + Map counts = new HashMap<>(); + operationService.createQuery() + .inNonFinalState() + .list() + .forEach(op -> counts.merge(hashUser(op.getUser()), 1L, Long::sum)); + return counts; + } + + private Map countActiveOperationsBySpace() { + Map counts = new HashMap<>(); + operationService.createQuery() + .inNonFinalState() + .list() + .forEach(op -> counts.merge(op.getSpaceId(), 1L, Long::sum)); + return counts; + } + private void syncMBeans(Map counts, String pattern, Map registry) throws Exception { - Set toRemove = new HashSet<>(registry.keySet()); - + Set stale = new HashSet<>(registry.keySet()); for (Map.Entry entry : counts.entrySet()) { ObjectName name = new ObjectName(pattern.formatted(ObjectName.quote(entry.getKey()))); - toRemove.remove(name); - ActiveOperationsCount bean = registry.get(name); - if (bean == null) { - bean = new ActiveOperationsCount(entry.getValue()); - mBeanServer.registerMBean(bean, name); - registry.put(name, bean); - } else { - bean.setCount(entry.getValue()); - } + stale.remove(name); + upsertMBean(name, entry.getValue(), registry); } + removeStaleMBeans(stale, registry); + } + + private void upsertMBean(ObjectName name, long count, + Map registry) throws Exception { + ActiveOperationsCount bean = registry.get(name); + if (bean == null) { + bean = new ActiveOperationsCount(count); + mBeanServer.registerMBean(bean, name); + registry.put(name, bean); + } else { + bean.setCount(count); + } + } - for (ObjectName name : toRemove) { + private void removeStaleMBeans(Set stale, + Map registry) throws Exception { + for (ObjectName name : stale) { mBeanServer.unregisterMBean(name); registry.remove(name); } diff --git a/pom.xml b/pom.xml index 16cbd38eea..f8ab346eac 100644 --- a/pom.xml +++ b/pom.xml @@ -66,8 +66,6 @@ 4.0.1 6.3.0 8.14.0 - 1.20.6 - 42.7.9 multiapps-controller-client @@ -840,24 +838,6 @@ bucket4j_jdk17-postgresql ${bucket4j.version} - - - org.testcontainers - postgresql - ${testcontainers.version} - - - - org.testcontainers - junit-jupiter - ${testcontainers.version} - - - - org.postgresql - postgresql - ${postgresql-jdbc.version} -