775 use tiered cache in place of caffeine cache

VERSION

@@ -1 +1 @@
-1.17.2
+1.17.3-B0

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.MoshiSerializable
 
 /**
  * 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 MoshiSerializable {
 
     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/RegistryLookupCache.groovy

@@ -0,0 +1,83 @@
+/*
+ *  Wave, containers provisioning service
+ *  Copyright (c) 2023-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
+
+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.encoder.MoshiEncodeStrategy
+import io.seqera.wave.encoder.MoshiSerializable
+import io.seqera.wave.store.cache.AbstractTieredCache
+import io.seqera.wave.store.cache.L2TieredCache
+import jakarta.inject.Singleton
+/**
+ * Implement a tiered cache for {@link RegistryLookupService}
+ *
+ * @author Munish Chouhan <munish.chouhan@seqera.io>
+ */
+@Slf4j
+@Singleton
+@CompileStatic
+class RegistryLookupCache extends AbstractTieredCache<String, RegistryAuth> {
+
+    @Value('${wave.registry-lookup.cache.duration:1h}')
+    private Duration duration
+
+    @Value('${wave.registry-lookup.cache.max-size:10000}')
+    private int maxSize
+
+    RegistryLookupCache(@Nullable L2TieredCache l2) {
+        super(l2, encoder())
+    }
+
+    @Override
+    int getMaxSize() {
+        return maxSize
+    }
+
+    @Override
+    protected getName() {
+        return 'registry-lookup-cache'
+    }
+
+    Duration getDuration() {
+        return duration
+    }
+
+    @Override
+    protected String getPrefix() {
+        return 'registry-lookup-cache/v1'
+    }
+
+    static MoshiEncodeStrategy encoder() {
+        new MoshiEncodeStrategy<AbstractTieredCache.Entry>(factory()) {}
+    }
+
+    static JsonAdapter.Factory factory() {
+        PolymorphicJsonAdapterFactory.of(MoshiSerializable.class, "@type")
+                .withSubtype(AbstractTieredCache.Entry.class, AbstractTieredCache.Entry.name)
+                .withSubtype(RegistryAuth.class, RegistryAuth.name)
+    }
+
+}

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

@@ -22,19 +22,14 @@ import java.net.http.HttpRequest
 import java.net.http.HttpResponse
 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.scheduling.TaskExecutors
 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
@@ -56,43 +51,12 @@ class RegistryLookupServiceImpl implements RegistryLookupService {
     @Inject
     private HttpClientConfig httpConfig
 
-    @Inject
-    private RegistryAuthStore store
-
     @Inject
     @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)
-            return result
-        }
-    }
-
-    // FIXME https://github.com/seqeralabs/wave/issues/747
-    private AsyncLoadingCache<URI, RegistryAuth> cache
-
-    @PostConstruct
-    void init() {
-        cache = Caffeine
-                .newBuilder()
-                .maximumSize(10_000)
-                .expireAfterAccess(1, TimeUnit.HOURS)
-                .executor(ioExecutor)
-                .buildAsync(loader)
-    }
+    @Inject
+    private RegistryLookupCache cache
 
     protected RegistryAuth lookup0(URI endpoint) {
         final httpClient = HttpClientFactory.followRedirectsHttpClient()
@@ -131,8 +95,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(), (k) -> lookup0(endpoint), cache.duration)
             return new RegistryInfo(registry, endpoint, auth)
         }
         catch (CompletionException e) {

src/main/groovy/io/seqera/wave/encoder/MoshiExchange.groovy → src/main/groovy/io/seqera/wave/encoder/MoshiSerializable.groovy

@@ -19,9 +19,9 @@
 package io.seqera.wave.encoder
 
 /**
- * Marker interface for Moshi encoded exchange objects
+ * Marker interface for Moshi serializable objects
  * 
  * @author Paolo Di Tommaso <paolo.ditommaso@gmail.com>
  */
-interface MoshiExchange {
+interface MoshiSerializable {
 }

src/main/groovy/io/seqera/wave/proxy/DelegateResponse.groovy

@@ -19,14 +19,14 @@
 package io.seqera.wave.proxy
 
 import groovy.transform.EqualsAndHashCode
-import io.seqera.wave.encoder.MoshiExchange
+import io.seqera.wave.encoder.MoshiSerializable
 /**
  * Model a response object to be forwarded to the client
  * 
  * @author Paolo Di Tommaso <paolo.ditommaso@gmail.com>
  */
 @EqualsAndHashCode
-class DelegateResponse implements MoshiExchange {
+class DelegateResponse implements MoshiSerializable {
     int statusCode
     Map<String,List<String>> headers
     byte[] body

src/main/groovy/io/seqera/wave/proxy/ProxyCache.groovy

@@ -26,7 +26,7 @@ import groovy.util.logging.Slf4j
 import io.micronaut.core.annotation.Nullable
 import io.seqera.wave.configuration.ProxyCacheConfig
 import io.seqera.wave.encoder.MoshiEncodeStrategy
-import io.seqera.wave.encoder.MoshiExchange
+import io.seqera.wave.encoder.MoshiSerializable
 import io.seqera.wave.store.cache.AbstractTieredCache
 import io.seqera.wave.store.cache.L2TieredCache
 import jakarta.inject.Singleton
@@ -38,7 +38,7 @@ import jakarta.inject.Singleton
 @Slf4j
 @Singleton
 @CompileStatic
-class ProxyCache extends AbstractTieredCache<DelegateResponse> {
+class ProxyCache extends AbstractTieredCache<String, DelegateResponse> {
 
     private ProxyCacheConfig config
 
@@ -50,7 +50,7 @@ class ProxyCache extends AbstractTieredCache<DelegateResponse> {
 
     static MoshiEncodeStrategy encoder() {
         // json adapter factory
-        final factory = PolymorphicJsonAdapterFactory.of(MoshiExchange.class, "@type")
+        final factory = PolymorphicJsonAdapterFactory.of(MoshiSerializable.class, "@type")
                 .withSubtype(Entry.class, Entry.name)
                 .withSubtype(DelegateResponse.class, DelegateResponse.simpleName)
         // the encoding strategy

src/main/groovy/io/seqera/wave/service/aws/AwsEcrService.groovy

@@ -18,19 +18,20 @@
 
 package io.seqera.wave.service.aws
 
+import java.time.Duration
 import java.util.concurrent.ExecutorService
-import java.util.concurrent.TimeUnit
 import java.util.regex.Pattern
 
-import com.github.benmanes.caffeine.cache.AsyncLoadingCache
-import com.github.benmanes.caffeine.cache.CacheLoader
-import com.github.benmanes.caffeine.cache.Caffeine
 import groovy.transform.Canonical
 import groovy.transform.CompileStatic
 import groovy.util.logging.Slf4j
+import io.micronaut.context.annotation.Value
 import io.micronaut.scheduling.TaskExecutors
+import io.seqera.wave.service.aws.cache.AwsEcrAuthToken
+import io.seqera.wave.service.aws.cache.AwsEcrCache
+import io.seqera.wave.store.cache.TieredKey
+import io.seqera.wave.util.RegHelper
 import io.seqera.wave.util.StringUtils
-import jakarta.annotation.PostConstruct
 import jakarta.inject.Inject
 import jakarta.inject.Named
 import jakarta.inject.Singleton
@@ -56,11 +57,16 @@ class AwsEcrService {
     static final private Pattern AWS_ECR_PUBLIC = ~/public\.ecr\.aws/
 
     @Canonical
-    private static class AwsCreds {
+    private static class AwsCreds implements TieredKey {
         String accessKey
         String secretKey
         String region
         boolean ecrPublic
+
+        @Override
+        String stableHash() {
+            RegHelper.sipHash(accessKey, secretKey, region, ecrPublic)
+        }
     }
 
     @Canonical
@@ -69,31 +75,22 @@ class AwsEcrService {
         String region
     }
 
-    private CacheLoader<AwsCreds, String> loader = new CacheLoader<AwsCreds, String>() {
-        @Override
-        String load(AwsCreds creds) throws Exception {
-            return creds.ecrPublic
-                    ? getLoginToken1(creds.accessKey, creds.secretKey, creds.region)
-                    : getLoginToken0(creds.accessKey, creds.secretKey, creds.region)
-        }
+    AwsEcrAuthToken load(AwsCreds creds) throws Exception {
+        def token = creds.ecrPublic
+                ? getLoginToken1(creds.accessKey, creds.secretKey, creds.region)
+                : getLoginToken0(creds.accessKey, creds.secretKey, creds.region)
+        return new AwsEcrAuthToken(token)
     }
 
     @Inject
     @Named(TaskExecutors.BLOCKING)
     private ExecutorService ioExecutor
 
-    // FIXME https://github.com/seqeralabs/wave/issues/747
-    private AsyncLoadingCache<AwsCreds, String> cache
-
-    @PostConstruct
-    private void init() {
-        cache = Caffeine
-                .newBuilder()
-                .maximumSize(10_000)
-                .expireAfterWrite(3, TimeUnit.HOURS)
-                .executor(ioExecutor)
-                .buildAsync(loader)
-    }
+    @Inject
+    private AwsEcrCache cache
+
+    @Value('${wave.aws.ecr.cache.duration:24h}')
+    private Duration cacheDuration
 
     private EcrClient ecrClient(String accessKey, String secretKey, String region) {
         EcrClient.builder()
@@ -139,10 +136,8 @@ class AwsEcrService {
         assert region, "Missing AWS region argument"
 
         try {
-            // get the token from the cache, if missing the it's automatically
-            // fetch using the AWS ECR client
-            // FIXME https://github.com/seqeralabs/wave/issues/747
-            return cache.synchronous().get(new AwsCreds(accessKey,secretKey,region,isPublic))
+            final key = new AwsCreds(accessKey,secretKey,region,isPublic)
+            return cache.getOrCompute(key, (k) -> load(key), cacheDuration).value
         }
         catch (Exception e) {
             final type = isPublic ? "ECR public" : "ECR"