Change: improve usage stats collection with a local cache

server/.gitignore

@@ -1,9 +1,11 @@
 /bin/
 /build/
+
 .gradle/
 .idea/
 .settings/
 .classpath
 .project
+
 DEPENDENCIES
 /src/dev/resources/application-ovsx.properties

server/src/main/java/org/eclipse/openvsx/ratelimit/UsageStatsService.java

@@ -12,6 +12,7 @@
  *****************************************************************************/
 package org.eclipse.openvsx.ratelimit;
 
+import com.github.benmanes.caffeine.cache.Cache;
 import jakarta.annotation.Nonnull;
 import org.eclipse.openvsx.entities.Customer;
 import org.eclipse.openvsx.entities.DailyUsageStats;
@@ -22,7 +23,7 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
-import org.springframework.scheduling.annotation.Async;
+import org.springframework.scheduling.annotation.Scheduled;
 import org.springframework.stereotype.Service;
 import redis.clients.jedis.JedisCluster;
 import redis.clients.jedis.params.ScanParams;
@@ -31,7 +32,6 @@
 import java.time.*;
 import java.time.temporal.ChronoUnit;
 import java.util.Arrays;
-import java.util.Comparator;
 import java.util.List;
 import java.util.Map;
 
@@ -48,20 +48,44 @@ public class UsageStatsService {
 
     private final RepositoryService repositories;
     private final CustomerService customerService;
+    private final Cache<Object, Object> usageCache;
     private final JedisCluster jedisCluster;
 
-    public UsageStatsService(RepositoryService repositories, CustomerService customerService, JedisCluster jedisCluster) {
+    public UsageStatsService(
+            RepositoryService repositories,
+            CustomerService customerService,
+            Cache<Object, Object> usageCache,
+            JedisCluster jedisCluster
+    ) {
         this.repositories = repositories;
         this.customerService = customerService;
+        this.usageCache = usageCache;
         this.jedisCluster = jedisCluster;
     }
 
-    @Async
     public void incrementUsage(Customer customer) {
         var key = customer.getId();
         var window = getCurrentUsageWindow();
-        var old = jedisCluster.hincrBy(USAGE_DATA_KEY, key + ":" + window, 1);
-        logger.debug("Usage count for {}: {}", customer.getName(), old + 1);
+
+        var count = usageCache.asMap().compute(key + ":" + window, (_, v) -> v == null ? 1 : ((Long) v) + 1);
+        logger.debug("Local usage count for {}: {}", customer.getName(), count);
+    }
+
+    @Scheduled(cron = "*/30 * * * * *")
+    public void syncDistributedUsageData() {
+        logger.debug("Updating distributed usage data from local cache");
+        var cacheMap = usageCache.asMap();
+        for (var key : cacheMap.keySet()) {
+            cacheMap.compute(key, (_, v) -> {
+                if (v != null) {
+                    var value = (Long) v;
+                    jedisCluster.hincrBy(USAGE_DATA_KEY, key.toString(), value.intValue());
+                    return 0L;
+                } else {
+                    return null;
+                }
+            });
+        }
     }
 
     public void persistUsageStats() {

server/src/main/java/org/eclipse/openvsx/ratelimit/cache/RateLimitCacheService.java

@@ -39,6 +39,7 @@ public class RateLimitCacheService extends JedisPubSub {
     public static final String CACHE_CUSTOMER = "ratelimit.customer";
     public static final String CACHE_TIER = "ratelimit.tier";
     public static final String CACHE_TOKEN = "ratelimit.token";
+    public static final String CACHE_USAGE = "ratelimit.usage";
 
     private static final String CONFIG_UPDATE_CHANNEL = "ratelimit.config";
 

server/src/main/java/org/eclipse/openvsx/ratelimit/config/RateLimitConfig.java

@@ -120,18 +120,31 @@ public Cache<Object, Object> tokenCache(
                 .build();
     }
 
+    @Bean
+    public Cache<Object, Object> usageCache(
+            @Value("${ovsx.caching.usage.ttl:PT1H}") Duration timeToLive
+    ) {
+        return Caffeine.newBuilder()
+                .expireAfterWrite(timeToLive)
+                .scheduler(Scheduler.systemScheduler())
+                .recordStats()
+                .build();
+    }
+
     @Bean
     @Qualifier(CACHE_MANAGER)
     public CacheManager rateLimitCacheManager(
             Cache<Object, Object> customerCache,
             Cache<Object, Object> tierCache,
-            Cache<Object, Object> tokenCache
+            Cache<Object, Object> tokenCache,
+            Cache<Object, Object> usageCache
     ) {
         logger.info("Configure rate limit cache manager");
         CaffeineCacheManager caffeineCacheManager = new CaffeineCacheManager();
         caffeineCacheManager.registerCustomCache(CACHE_CUSTOMER, customerCache);
         caffeineCacheManager.registerCustomCache(CACHE_TIER, tierCache);
         caffeineCacheManager.registerCustomCache(CACHE_TOKEN, tokenCache);
+        caffeineCacheManager.registerCustomCache(CACHE_USAGE, usageCache);
 
         return caffeineCacheManager;
     }

server/src/main/java/org/eclipse/openvsx/ratelimit/filter/RateLimitServletFilter.java

@@ -68,6 +68,12 @@ protected void doFilterInternal(
         var identity = identityService.resolveIdentity(request);
         logger.debug("Rate limit filter: {}: {}", request.getRequestURI(), identity.ipAddress());
 
+        if (identity.isCustomer()) {
+            var customer = identity.getCustomer();
+            logger.debug("Increasing usage stats for customer {}", customer.getName());
+            usageStatsService.incrementUsage(customer);
+        }
+
         var bucketPair = rateLimitService.getBucket(identity);
         var bucket = bucketPair.bucket();
         if (bucket == null) {
@@ -85,12 +91,6 @@ protected void doFilterInternal(
         } else {
             handleHttpResponseOnRateLimiting(response, probe);
         }
-
-        if (identity.isCustomer()) {
-            var customer = identity.getCustomer();
-            logger.debug("Increasing usage stats for customer {}", customer.getName());
-            usageStatsService.incrementUsage(customer);
-        }
     }
 
     private void handleHttpResponseOnRateLimiting(HttpServletResponse response, ConsumptionProbe probe) throws IOException {