Feat/MET-5383 Add rate control to validation workflow

pom.xml

@@ -62,8 +62,28 @@
     <version.netty>4.1.91.Final</version.netty>
     <version.reactor>3.4.24</version.reactor>
     <version.rabbittest>2.4.5</version.rabbittest>
+    <version.bucket4j>8.3.0</version.bucket4j>
+    <version.postgresql.bucket4j>8.1.1</version.postgresql.bucket4j>
   </properties>
   <dependencies>
+    <!-- Rate limiting dependencies-->
+    <dependency>
+      <groupId>com.bucket4j</groupId>
+      <artifactId>bucket4j-core</artifactId>
+      <version>${version.bucket4j}</version>
+    </dependency>
+    <dependency>
+      <groupId>com.bucket4j</groupId>
+      <artifactId>bucket4j-postgresql</artifactId>
+      <version>${version.postgresql.bucket4j}</version>
+      <exclusions>
+        <exclusion>
+          <groupId>com.zaxxer</groupId>
+          <artifactId>HikariCP-java6</artifactId>
+        </exclusion>
+      </exclusions>
+    </dependency>
+
     <!-- Spring boot -->
     <dependency>
       <groupId>org.springframework.boot</groupId>

src/main/java/eu/europeana/metis/sandbox/config/RateLimitConfig.java

@@ -0,0 +1,37 @@
+package eu.europeana.metis.sandbox.config;
+
+import eu.europeana.metis.sandbox.controller.ratelimit.RateLimitInterceptor;
+import io.github.bucket4j.TimeMeter;
+import io.github.bucket4j.distributed.jdbc.BucketTableSettings;
+import io.github.bucket4j.distributed.jdbc.SQLProxyConfiguration;
+import io.github.bucket4j.distributed.proxy.ClientSideConfig;
+import io.github.bucket4j.postgresql.PostgreSQLadvisoryLockBasedProxyManager;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+
+import javax.sql.DataSource;
+
+/**
+ * Configuration file for rate limiting feature
+ */
+@Configuration
+public class RateLimitConfig {
+
+    @Value("${sandbox.rate-limit.bandwidth.capacity}")
+    private String capacity;
+
+    @Value("${sandbox.rate-limit.bandwidth.time}")
+    private String time;
+
+    @Bean
+    RateLimitInterceptor rateLimitInterceptor(DataSource dataSource){
+        SQLProxyConfiguration<Long> sqlProxyConfiguration = SQLProxyConfiguration.builder()
+                .withClientSideConfig(ClientSideConfig.getDefault().withClientClock(TimeMeter.SYSTEM_MILLISECONDS))
+                .withTableSettings(BucketTableSettings.customSettings("rate_limit.buckets", "id", "bucket_state"))
+                .build(dataSource);
+        PostgreSQLadvisoryLockBasedProxyManager proxyManager = new PostgreSQLadvisoryLockBasedProxyManager(sqlProxyConfiguration);
+        return new RateLimitInterceptor(Integer.parseInt(capacity), Long.parseLong(time), proxyManager);
+
+    }
+}

src/main/java/eu/europeana/metis/sandbox/config/web/MvcConfig.java

@@ -1,10 +1,13 @@
 package eu.europeana.metis.sandbox.config.web;
 
+import eu.europeana.metis.sandbox.controller.ratelimit.RateLimitInterceptor;
 import org.apache.commons.lang3.ArrayUtils;
+import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.annotation.Value;
 import org.springframework.context.annotation.Configuration;
 import org.springframework.format.FormatterRegistry;
 import org.springframework.web.servlet.config.annotation.CorsRegistry;
+import org.springframework.web.servlet.config.annotation.InterceptorRegistry;
 import org.springframework.web.servlet.config.annotation.ViewControllerRegistry;
 import org.springframework.web.servlet.config.annotation.WebMvcConfigurer;
 
@@ -18,6 +21,9 @@ class MvcConfig implements WebMvcConfigurer {
   @Value("${sandbox.cors.mapping}")
   private String[] corsMapping;
 
+  @Autowired
+  private RateLimitInterceptor rateLimitInterceptor;
+
   @Override
   public void addViewControllers(ViewControllerRegistry registry) {
     registry.addRedirectViewController("/", "swagger-ui.html");
@@ -37,4 +43,9 @@ public void addCorsMappings(CorsRegistry registry) {
           .allowedMethods("GET", "HEAD", "POST", "PUT", "DELETE", "OPTIONS");
     }
   }
+
+  @Override
+  public void addInterceptors(InterceptorRegistry registry){
+    registry.addInterceptor(rateLimitInterceptor).addPathPatterns("/record/validation/**");
+  }
 }

src/main/java/eu/europeana/metis/sandbox/controller/ratelimit/RateLimitInterceptor.java

@@ -0,0 +1,69 @@
+package eu.europeana.metis.sandbox.controller.ratelimit;
+
+
+import io.github.bucket4j.*;
+import io.github.bucket4j.distributed.remote.CommandResult;
+import io.github.bucket4j.distributed.remote.Request;
+import io.github.bucket4j.distributed.remote.commands.TryConsumeAndReturnRemainingTokensCommand;
+import io.github.bucket4j.postgresql.PostgreSQLadvisoryLockBasedProxyManager;
+import org.jetbrains.annotations.NotNull;
+import org.springframework.http.HttpStatus;
+import org.springframework.web.servlet.HandlerInterceptor;
+
+import javax.servlet.http.HttpServletRequest;
+import javax.servlet.http.HttpServletResponse;
+import java.time.Duration;
+
+/**
+ * Implementation of Rate Limit interceptor to intercept the requests
+ */
+public class RateLimitInterceptor implements HandlerInterceptor {
+
+    private final Integer capacity;
+    private final PostgreSQLadvisoryLockBasedProxyManager postgreSQLManager;
+    private final BucketConfiguration bucketConfiguration;
+
+    /**
+     * Constructor
+     * @param capacity The max number of tokens per user
+     * @param time The time it takes to refresh the tokens per uset
+     * @param postgreSQLManager The database manager of the tokens per user
+     */
+    public RateLimitInterceptor(Integer capacity, Long time, PostgreSQLadvisoryLockBasedProxyManager postgreSQLManager){
+        this.capacity = capacity;
+        this.postgreSQLManager = postgreSQLManager;
+        bucketConfiguration = BucketConfiguration.builder()
+                .addLimit(Bandwidth.classic(capacity, Refill.intervally(capacity, Duration.ofSeconds(time))))
+                .build();
+    }
+
+    @Override
+    public boolean preHandle(HttpServletRequest request, @NotNull HttpServletResponse response, @NotNull Object handler) throws Exception {
+        Long key = (long) request.getRemoteAddr().hashCode();
+        ConsumptionProbe probe = resolveBucket(key);
+        response.addHeader("X-Rate-Limit-Limit", String.valueOf(capacity));
+        if (probe.isConsumed()) {
+            response.addHeader("X-Rate-Limit-Remaining", String.valueOf(probe.getRemainingTokens()));
+            return true;
+        } else {
+            long waitForRefill = probe.getNanosToWaitForRefill() / 1_000_000_000;
+            response.addHeader("X-Rate-Limit-Reset", String.valueOf(waitForRefill));
+            response.sendError(HttpStatus.TOO_MANY_REQUESTS.value(),
+                    "You have exhausted your API Request Quota");
+            return false;
+
+        }
+    }
+
+    private ConsumptionProbe resolveBucket(Long apiKey) {
+        Request<ConsumptionProbe> request = new Request<>(new TryConsumeAndReturnRemainingTokensCommand(1), null, null);
+        CommandResult<ConsumptionProbe> commandResult = postgreSQLManager.execute(apiKey, request);
+        if(commandResult.isBucketNotFound()){
+            Bucket bucket = postgreSQLManager.builder().build(apiKey, bucketConfiguration);
+            return bucket.tryConsumeAndReturnRemaining(1);
+        } else {
+            return commandResult.getData();
+        }
+    }
+
+}

src/main/resources/database/schema_validation.sql

@@ -0,0 +1,11 @@
+BEGIN;
+CREATE SCHEMA rate_limit;
+
+CREATE TABLE IF NOT EXISTS rate_limit.buckets
+(
+    id            BIGINT    NOT NULL,
+    bucket_state  BYTEA     NOT NULL,
+    PRIMARY KEY (id)
+);
+
+COMMIT;

src/main/resources/database/schema_validation_drop.sql

@@ -0,0 +1,3 @@
+BEGIN;
+DROP SCHEMA IF EXISTS rate_limit CASCADE;
+COMMIT;

src/main/resources/sample.application.yml

@@ -89,6 +89,10 @@ sandbox:
     password:
     maximumPoolSize: 10           # this is the default we make it explicit in the config.
     leakDetectionThreshold: 5000  # five second should be enough
+  rate-limit:
+    bandwidth:
+      capacity:
+      time: # in seconds
   dataset:
     clean:
       frequency: 0 0 0 * * ? # At 00:00:00am every day

src/test/java/eu/europeana/metis/sandbox/controller/DatasetControllerTest.java

@@ -36,6 +36,7 @@
 import eu.europeana.metis.sandbox.common.exception.ServiceException;
 import eu.europeana.metis.sandbox.common.locale.Country;
 import eu.europeana.metis.sandbox.common.locale.Language;
+import eu.europeana.metis.sandbox.controller.ratelimit.RateLimitInterceptor;
 import eu.europeana.metis.sandbox.domain.DatasetMetadata;
 import eu.europeana.metis.sandbox.dto.DatasetInfoDto;
 import eu.europeana.metis.sandbox.dto.RecordTiersInfoDto;
@@ -88,6 +89,9 @@ class DatasetControllerTest {
   @Autowired
   private MockMvc mvc;
 
+  @MockBean
+  private RateLimitInterceptor rateLimitInterceptor;
+
   @MockBean
   private DatasetService datasetService;
 

src/test/java/eu/europeana/metis/sandbox/controller/PatternAnalysisControllerTest.java

@@ -13,6 +13,7 @@
 import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.status;
 
 import eu.europeana.metis.sandbox.common.Step;
+import eu.europeana.metis.sandbox.controller.ratelimit.RateLimitInterceptor;
 import eu.europeana.metis.sandbox.dto.report.ProgressInfoDto;
 import eu.europeana.metis.sandbox.entity.RecordLogEntity;
 import eu.europeana.metis.sandbox.entity.problempatterns.ExecutionPoint;
@@ -54,6 +55,9 @@ class PatternAnalysisControllerTest {
   @Autowired
   private MockMvc mvc;
 
+  @MockBean
+  private RateLimitInterceptor rateLimitInterceptor;
+
   @MockBean
   private PatternAnalysisService<Step, ExecutionPoint> mockPatternAnalysisService;
 

src/test/java/eu/europeana/metis/sandbox/controller/ValidationControllerTest.java

@@ -1,6 +1,7 @@
 package eu.europeana.metis.sandbox.controller;
 
 import eu.europeana.metis.sandbox.common.Step;
+import eu.europeana.metis.sandbox.controller.ratelimit.RateLimitInterceptor;
 import eu.europeana.metis.sandbox.service.validationworkflow.RecordValidationMessage;
 import eu.europeana.metis.sandbox.service.validationworkflow.ValidationResult;
 import eu.europeana.metis.sandbox.service.validationworkflow.ValidationWorkflowReport;
@@ -42,6 +43,8 @@ class ValidationControllerTest {
     ValidationWorkflowService validationWorkflowService;
     @Autowired
     private MockMvc mvc;
+    @MockBean
+    private RateLimitInterceptor rateLimitInterceptor;
 
     @NotNull
     private static ProblemPatternAnalysis getProblemPatternAnalysis() {

src/test/resources/application.yml

@@ -56,6 +56,10 @@ sandbox:
     #MET-3929 connection leak detection
     maximumPoolSize: 10           # this is the default we make it explicit in the config.
     leakDetectionThreshold: 5000  # five second should be enough
+  rate-limit:
+    bandwidth:
+      capacity: 20
+      time: 3600 # in seconds
   dataset:
     clean:
       frequency: 0 0 0 * * ? # At 00:00:00am every day