775 use tiered cache in place of caffeine cache

src/main/groovy/io/seqera/wave/auth/RegistryAuth.groovy

@@ -23,6 +23,7 @@ import java.util.regex.Pattern
 import groovy.transform.Canonical
 import groovy.transform.CompileStatic
 import groovy.transform.ToString
+import io.seqera.wave.encoder.MoshiExchange
 
 /**
  * Model container registry authentication meta-info
@@ -32,7 +33,7 @@ import groovy.transform.ToString
 @Canonical
 @CompileStatic
 @ToString(includePackage = false, includeNames = true)
-class RegistryAuth {
+class RegistryAuth implements MoshiExchange {
 
     private static final Pattern AUTH = ~/(?i)(?<type>.+) realm="(?<realm>[^"]+)",service="(?<service>[^"]+)"/
     // some registries doesn't send the service

src/main/groovy/io/seqera/wave/auth/RegistryAuthServiceImpl.groovy

@@ -28,17 +28,21 @@ import java.util.concurrent.TimeUnit
 import com.github.benmanes.caffeine.cache.AsyncLoadingCache
 import com.github.benmanes.caffeine.cache.CacheLoader
 import com.github.benmanes.caffeine.cache.Caffeine
+import com.google.common.hash.Hashing
 import groovy.json.JsonSlurper
 import groovy.transform.Canonical
 import groovy.transform.CompileStatic
 import groovy.transform.PackageScope
 import groovy.transform.ToString
 import groovy.util.logging.Slf4j
 import io.micronaut.scheduling.TaskExecutors
+import io.seqera.wave.auth.cache.RegistryAuthCache
+import io.seqera.wave.auth.model.RegistryAuthToken
 import io.seqera.wave.configuration.HttpClientConfig
 import io.seqera.wave.exception.RegistryForwardException
 import io.seqera.wave.exception.RegistryUnauthorizedAccessException
 import io.seqera.wave.http.HttpClientFactory
+import io.seqera.wave.store.cache.TieredCacheKey
 import io.seqera.wave.util.RegHelper
 import io.seqera.wave.util.Retryable
 import io.seqera.wave.util.StringUtils
@@ -74,21 +78,19 @@ class RegistryAuthServiceImpl implements RegistryAuthService {
 
     @Canonical
     @ToString(includePackage = false, includeNames = true)
-    static private class CacheKey {
+    static private class CacheKey implements TieredCacheKey{
         final String image
         final RegistryAuth auth
         final RegistryCredentials creds
 
-        String stableKey() {
+        @Override
+        String stableHash() {
             return RegHelper.sipHash(['content': toString()])
         }
     }
 
-    private CacheLoader<CacheKey, String> loader = new CacheLoader<CacheKey, String>() {
-        @Override
-        String load(CacheKey key) throws Exception {
-            return getToken(key)
-        }
+    RegistryAuthToken load(CacheKey key) throws Exception {
+        return new RegistryAuthToken(getToken(key))
     }
 
     protected String getToken(CacheKey key){
@@ -108,25 +110,15 @@ class RegistryAuthServiceImpl implements RegistryAuthService {
         return result
     }
 
-    // FIXME https://github.com/seqeralabs/wave/issues/747
-    private AsyncLoadingCache<CacheKey, String> cacheTokens
+    @Inject
+    private RegistryAuthCache cache
 
     @Inject
     private RegistryLookupService lookupService
 
     @Inject
     private RegistryCredentialsFactory credentialsFactory
 
-    @PostConstruct
-    private void init() {
-        cacheTokens = Caffeine
-                .newBuilder()
-                .maximumSize(10_000)
-                .expireAfterAccess(_1_HOUR.toMillis(), TimeUnit.MILLISECONDS)
-                .executor(ioExecutor)
-                .buildAsync(loader)
-    }
-
     /**
      * Implements container registry login
      *
@@ -284,8 +276,7 @@ class RegistryAuthServiceImpl implements RegistryAuthService {
     protected String getAuthToken(String image, RegistryAuth auth, RegistryCredentials creds) {
         final key = new CacheKey(image, auth, creds)
         try {
-            // FIXME https://github.com/seqeralabs/wave/issues/747
-            return cacheTokens.synchronous().get(key)
+            return cache.getOrCompute(key, (k)->load(key), _1_HOUR).value
         }
         catch (CompletionException e) {
             // this catches the exception thrown in the cache loader lookup
@@ -303,15 +294,14 @@ class RegistryAuthServiceImpl implements RegistryAuthService {
      */
     void invalidateAuthorization(String image, RegistryAuth auth, RegistryCredentials creds) {
         final key = new CacheKey(image, auth, creds)
-        // FIXME https://github.com/seqeralabs/wave/issues/747
-        cacheTokens.synchronous().invalidate(key)
+        cache.invalidate(key)
         tokenStore.remove(getStableKey(key))
     }
 
     /**
      * Invalidate all cached authorization tokens
      */
     private static String getStableKey(CacheKey key) {
-        return "key-" + key.stableKey()
+        return "key-" + key.stableHash()
     }
 }

src/main/groovy/io/seqera/wave/auth/RegistryLookupServiceImpl.groovy

@@ -20,21 +20,19 @@ package io.seqera.wave.auth
 
 import java.net.http.HttpRequest
 import java.net.http.HttpResponse
+import java.time.Duration
 import java.util.concurrent.CompletionException
 import java.util.concurrent.ExecutorService
-import java.util.concurrent.TimeUnit
 
-import com.github.benmanes.caffeine.cache.AsyncLoadingCache
-import com.github.benmanes.caffeine.cache.CacheLoader
-import com.github.benmanes.caffeine.cache.Caffeine
 import groovy.transform.CompileStatic
 import groovy.util.logging.Slf4j
+import io.micronaut.context.annotation.Value
 import io.micronaut.scheduling.TaskExecutors
+import io.seqera.wave.auth.cache.RegistryLookupCache
 import io.seqera.wave.configuration.HttpClientConfig
 import io.seqera.wave.exception.RegistryForwardException
 import io.seqera.wave.http.HttpClientFactory
 import io.seqera.wave.util.Retryable
-import jakarta.annotation.PostConstruct
 import jakarta.inject.Inject
 import jakarta.inject.Named
 import jakarta.inject.Singleton
@@ -63,36 +61,26 @@ class RegistryLookupServiceImpl implements RegistryLookupService {
     @Named(TaskExecutors.BLOCKING)
     private ExecutorService ioExecutor
 
-    private CacheLoader<URI, RegistryAuth> loader = new CacheLoader<URI, RegistryAuth>() {
-        @Override
-        RegistryAuth load(URI endpoint) throws Exception {
-            // check if there's a record in the store cache (redis)
-            def result = store.get(endpoint.toString())
-            if( result ) {
-                log.debug "Authority lookup for endpoint: '$endpoint' => $result [from store]"
-               return result
-            }
-            // look-up using the corresponding API endpoint
-            result = lookup0(endpoint)
-            log.debug "Authority lookup for endpoint: '$endpoint' => $result"
-            // save it in the store cache (redis)
-            store.put(endpoint.toString(), result)
+    RegistryAuth load(URI endpoint) throws Exception {
+        // check if there's a record in the store cache (redis)
+        def result = store.get(endpoint.toString())
+        if( result ) {
+            log.debug "Authority lookup for endpoint: '$endpoint' => $result [from store]"
             return result
         }
+        // look-up using the corresponding API endpoint
+        result = lookup0(endpoint)
+        log.debug "Authority lookup for endpoint: '$endpoint' => $result"
+        // save it in the store cache (redis)
+        store.put(endpoint.toString(), result)
+        return result
     }
 
-    // FIXME https://github.com/seqeralabs/wave/issues/747
-    private AsyncLoadingCache<URI, RegistryAuth> cache
+    @Inject
+    private RegistryLookupCache cache
 
-    @PostConstruct
-    void init() {
-        cache = Caffeine
-                .newBuilder()
-                .maximumSize(10_000)
-                .expireAfterAccess(1, TimeUnit.HOURS)
-                .executor(ioExecutor)
-                .buildAsync(loader)
-    }
+    @Value('${wave.registry.cache.duration:24h}')
+    private Duration cacheDuration
 
     protected RegistryAuth lookup0(URI endpoint) {
         final httpClient = HttpClientFactory.followRedirectsHttpClient()
@@ -131,8 +119,7 @@ class RegistryLookupServiceImpl implements RegistryLookupService {
     RegistryInfo lookup(String registry) {
         try {
             final endpoint = registryEndpoint(registry)
-            // FIXME https://github.com/seqeralabs/wave/issues/747
-            final auth = cache.synchronous().get(endpoint)
+            final auth = cache.getOrCompute(endpoint.toString(), (key) -> load(endpoint), cacheDuration)
             return new RegistryInfo(registry, endpoint, auth)
         }
         catch (CompletionException e) {

src/main/groovy/io/seqera/wave/auth/cache/RegistryAuthCache.groovy

@@ -0,0 +1,81 @@
+/*
+ *  Wave, containers provisioning service
+ *  Copyright (c) 2024, Seqera Labs
+ *
+ *  This program is free software: you can redistribute it and/or modify
+ *  it under the terms of the GNU Affero General Public License as published by
+ *  the Free Software Foundation, either version 3 of the License, or
+ *  (at your option) any later version.
+ *
+ *  This program is distributed in the hope that it will be useful,
+ *  but WITHOUT ANY WARRANTY; without even the implied warranty of
+ *  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ *  GNU Affero General Public License for more details.
+ *
+ *  You should have received a copy of the GNU Affero General Public License
+ *  along with this program.  If not, see <https://www.gnu.org/licenses/>.
+ */
+
+package io.seqera.wave.auth.cache
+
+import java.time.Duration
+
+import com.squareup.moshi.JsonAdapter
+import com.squareup.moshi.adapters.PolymorphicJsonAdapterFactory
+import groovy.transform.CompileStatic
+import groovy.util.logging.Slf4j
+import io.micronaut.context.annotation.Value
+import io.micronaut.core.annotation.Nullable
+import io.seqera.wave.auth.model.RegistryAuthToken
+import io.seqera.wave.encoder.MoshiEncodeStrategy
+import io.seqera.wave.encoder.MoshiExchange
+import io.seqera.wave.store.cache.AbstractTieredCache
+import io.seqera.wave.store.cache.L2TieredCache
+import io.seqera.wave.store.cache.TieredCacheKey
+import jakarta.inject.Singleton
+/**
+ * Implement a tiered cache for Registry Auth
+ *
+ * @author Munish Chouhan <munish.chouhan@seqera.io>
+ */
+@Slf4j
+@Singleton
+@CompileStatic
+class RegistryAuthCache  extends AbstractTieredCache<TieredCacheKey, RegistryAuthToken> {
+
+    @Value('${wave.registry.auth.cache.duration:1h}')
+    private Duration duration
+
+    @Value('${wave.registry.auth.cache.max-size:10000}')
+    private int maxSize
+
+    RegistryAuthCache(@Nullable L2TieredCache l2) {
+        super(l2, encoder())
+    }
+
+    @Override
+    int getMaxSize() {
+        return maxSize
+    }
+
+    @Override
+    protected getName() {
+        return 'registry-auth-cache'
+    }
+
+    @Override
+    protected String getPrefix() {
+        return 'registry-auth-cache/v1'
+    }
+
+    static MoshiEncodeStrategy encoder() {
+        new MoshiEncodeStrategy<AbstractTieredCache.Entry>(factory()) {}
+    }
+
+    static JsonAdapter.Factory factory() {
+        PolymorphicJsonAdapterFactory.of(MoshiExchange.class, "@type")
+                .withSubtype(AbstractTieredCache.Entry.class, AbstractTieredCache.Entry.name)
+                .withSubtype(RegistryAuthToken.class, RegistryAuthToken.name)
+    }
+
+}

src/main/groovy/io/seqera/wave/auth/cache/RegistryLookupCache.groovy

@@ -0,0 +1,81 @@
+/*
+ *  Wave, containers provisioning service
+ *  Copyright (c) 2024, Seqera Labs
+ *
+ *  This program is free software: you can redistribute it and/or modify
+ *  it under the terms of the GNU Affero General Public License as published by
+ *  the Free Software Foundation, either version 3 of the License, or
+ *  (at your option) any later version.
+ *
+ *  This program is distributed in the hope that it will be useful,
+ *  but WITHOUT ANY WARRANTY; without even the implied warranty of
+ *  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ *  GNU Affero General Public License for more details.
+ *
+ *  You should have received a copy of the GNU Affero General Public License
+ *  along with this program.  If not, see <https://www.gnu.org/licenses/>.
+ */
+
+package io.seqera.wave.auth.cache
+
+import java.time.Duration
+
+import com.squareup.moshi.JsonAdapter
+import com.squareup.moshi.adapters.PolymorphicJsonAdapterFactory
+import groovy.transform.CompileStatic
+import groovy.util.logging.Slf4j
+import io.micronaut.context.annotation.Value
+import io.micronaut.core.annotation.Nullable
+import io.seqera.wave.auth.RegistryAuth
+import io.seqera.wave.encoder.MoshiEncodeStrategy
+import io.seqera.wave.encoder.MoshiExchange
+import io.seqera.wave.store.cache.AbstractTieredCache
+import io.seqera.wave.store.cache.L2TieredCache
+import io.seqera.wave.store.cache.TieredCacheKey
+import jakarta.inject.Singleton
+/**
+ * Implement a tiered cache for Registry lookup
+ *
+ * @author Munish Chouhan <munish.chouhan@seqera.io>
+ */
+@Slf4j
+@Singleton
+@CompileStatic
+class RegistryLookupCache extends AbstractTieredCache<TieredCacheKey, RegistryAuth> {
+
+    @Value('${wave.registry.cache.duration:1h}')
+    private Duration duration
+
+    @Value('${wave.registry.cache.max-size:10000}')
+    private int maxSize
+
+    RegistryLookupCache(@Nullable L2TieredCache l2) {
+        super(l2, encoder())
+    }
+
+    @Override
+    int getMaxSize() {
+        return maxSize
+    }
+
+    @Override
+    protected getName() {
+        return 'registry-cache'
+    }
+
+    @Override
+    protected String getPrefix() {
+        return 'registry-cache/v1'
+    }
+
+    static MoshiEncodeStrategy encoder() {
+        new MoshiEncodeStrategy<AbstractTieredCache.Entry>(factory()) {}
+    }
+
+    static JsonAdapter.Factory factory() {
+        PolymorphicJsonAdapterFactory.of(MoshiExchange.class, "@type")
+                .withSubtype(AbstractTieredCache.Entry.class, AbstractTieredCache.Entry.name)
+                .withSubtype(RegistryAuth.class, RegistryAuth.name)
+    }
+
+}