Merge branch 'master' into 314-update-to-micronaut-41

VERSION

@@ -1 +1 @@
-1.10.1
+1.10.3

build.gradle

@@ -119,7 +119,7 @@ micronaut {
 //
 jib {
     from {
-        image = 'cr.seqera.io/public/nf-jdk:corretto-17.0.10-al2023-jemalloc'
+        image = 'cr.seqera.io/public/nf-jdk:corretto-21-al2023-jemalloc'
         platforms {
             platform { architecture = 'amd64'; os = 'linux' }
         }

changelog.txt

@@ -1,4 +1,10 @@
 # Wave changelog
+1.10.3 - 5 Aug 2024
+- Fix from invalidateAuthorization from redis cache store (#591) [8f293d7c]
+
+1.10.2 - 3 Aug 2024
+- Add second level cache in RegistryAuthService (#588) [4ce329a7]
+
 1.10.1 - 26 Jul 2024
 - Add second level cache for Registry auth record (#582) [702528df]
 - Improve error report when platform cannot be found (#579) [5a9120b8]

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

@@ -20,6 +20,7 @@ package io.seqera.wave.auth
 
 import java.net.http.HttpRequest
 import java.net.http.HttpResponse
+import java.time.Duration
 import java.util.concurrent.ExecutionException
 import java.util.concurrent.TimeUnit
 
@@ -30,10 +31,12 @@ import com.google.common.util.concurrent.UncheckedExecutionException
 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.seqera.wave.configuration.HttpClientConfig
 import io.seqera.wave.http.HttpClientFactory
+import io.seqera.wave.util.RegHelper
 import io.seqera.wave.util.Retryable
 import io.seqera.wave.util.StringUtils
 import jakarta.inject.Inject
@@ -52,28 +55,54 @@ import static io.seqera.wave.WaveDefault.HTTP_RETRYABLE_ERRORS
 @CompileStatic
 class RegistryAuthServiceImpl implements RegistryAuthService {
 
+    @PackageScope static final Duration _1_HOUR = Duration.ofHours(1)
+
     @Inject
     private HttpClientConfig httpConfig
 
+    @Inject
+    private RegistryTokenCacheStore tokenStore
+
     @Canonical
     @ToString(includePackage = false, includeNames = true)
     static private class CacheKey {
         final String image
         final RegistryAuth auth
         final RegistryCredentials creds
+
+        String stableKey() {
+            return RegHelper.sipHash(['content': toString()])
+        }
     }
 
     private CacheLoader<CacheKey, String> loader = new CacheLoader<CacheKey, String>() {
         @Override
         String load(CacheKey key) throws Exception {
-            return getToken0(key)
+            return getToken(key)
+        }
+    }
+
+    protected String getToken(CacheKey key){
+        // check if there's a record in the store cache (redis)
+        // since the key is shared across replicas, it should be stable (java hashCode is not good)
+        final stableKey = getStableKey(key)
+        def result = tokenStore.get(stableKey)
+        if( result ) {
+            log.debug "Registry auth token for cachekey: '$key' [$stableKey] => $result [from store]"
+            return result
         }
+        // look-up using the corresponding API endpoint
+        result = getToken0(key)
+        log.debug "Registry auth token for cachekey: '$key' [$stableKey] => $result"
+        // save it in the store cache (redis)
+        tokenStore.put(stableKey, result)
+        return result
     }
 
     private LoadingCache<CacheKey, String> cacheTokens = CacheBuilder<CacheKey, String>
                     .newBuilder()
                     .maximumSize(10_000)
-                    .expireAfterAccess(1, TimeUnit.HOURS)
+                    .expireAfterAccess(_1_HOUR.toMillis(), TimeUnit.MILLISECONDS)
                     .build(loader)
 
     @Inject
@@ -253,7 +282,13 @@ class RegistryAuthServiceImpl implements RegistryAuthService {
     void invalidateAuthorization(String image, RegistryAuth auth, RegistryCredentials creds) {
         final key = new CacheKey(image, auth, creds)
         cacheTokens.invalidate(key)
+        tokenStore.remove(getStableKey(key))
     }
 
-
+    /**
+     * Invalidate all cached authorization tokens
+     */
+    private static String getStableKey(CacheKey key) {
+        return "key-" + key.stableKey()
+    }
 }

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

@@ -0,0 +1,54 @@
+/*
+ *  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 groovy.transform.CompileStatic
+import groovy.util.logging.Slf4j
+import io.seqera.wave.encoder.MoshiEncodeStrategy
+import io.seqera.wave.service.cache.AbstractCacheStore
+import io.seqera.wave.service.cache.impl.CacheProvider
+import jakarta.inject.Singleton
+/**
+ * Implement a cache store for {@link io.seqera.wave.auth.RegistryAuthServiceImpl.CacheKey} object and token that
+ * can be distributed across wave replicas
+ *
+ * @author Munish Chouhan <[email protected]>
+ */
+@Slf4j
+@Singleton
+@CompileStatic
+class RegistryTokenCacheStore extends AbstractCacheStore<String> {
+
+    RegistryTokenCacheStore(CacheProvider<String, String> provider) {
+        super(provider, new MoshiEncodeStrategy<String>() {})
+        log.info "Creating Registry Auth token cache store"
+    }
+
+    @Override
+    protected String getPrefix() {
+        return 'registry-token/v1:'
+    }
+
+    @Override
+    protected Duration getDuration() {
+        return RegistryAuthServiceImpl._1_HOUR
+    }
+}

src/main/groovy/io/seqera/wave/controller/RegistryProxyController.groovy

@@ -53,7 +53,6 @@ import io.seqera.wave.service.builder.ContainerBuildService
 import io.seqera.wave.storage.DigestStore
 import io.seqera.wave.storage.DockerDigestStore
 import io.seqera.wave.storage.HttpDigestStore
-import io.seqera.wave.storage.LazyDigestStore
 import io.seqera.wave.storage.Storage
 import io.seqera.wave.util.Retryable
 import jakarta.inject.Inject
@@ -173,15 +172,16 @@ class RegistryProxyController {
             final entry = storage.getBlob(route.getTargetPath()).orElse(null)
             String location
             if( location=dockerRedirection(entry) ) {
-                log.debug "Blob found in the cache: $route.path ==> mapping to: ${location}"
+                log.debug "Blob found in the cache [docker]: ${route.path} ==> mapping to: ${location}"
                 final target = RoutePath.parse(location, route.identity)
                 return handleDelegate0(target, httpRequest)
             }
             else if ( location=httpRedirect(entry) ) {
-                log.debug "Blob found in the cache: $route.path  ==> mapping to: $location"
+                log.debug "Blob found in the cache [http]: ${route.path} ==> mapping to: ${location}"
                 return fromCacheRedirect(location)
             }
             else if( entry ) {
+                log.trace "Blob found in the cache [digest]: ${route.path} ==> entry: ${entry}"
                 return fromCacheDigest(entry)
             }
         }
@@ -197,8 +197,6 @@ class RegistryProxyController {
     private String httpRedirect(DigestStore entry) {
         if( entry instanceof HttpDigestStore )
             return (entry as HttpDigestStore).location
-        if( entry instanceof LazyDigestStore )
-            return (entry as LazyDigestStore).location
         return null
     }
 

src/main/groovy/io/seqera/wave/cron/ThreadMonitorCron.groovy

@@ -96,16 +96,16 @@ class ThreadMonitorCron {
     }
 
     private String dumpThreads() {
-        def buffer = new StringBuffer()
+        final result = new StringBuilder()
         Map<Thread, StackTraceElement[]> m = Thread.getAllStackTraces();
         for(Map.Entry<Thread,  StackTraceElement[]> e : m.entrySet()) {
-            buffer.append('\n').append(e.getKey().toString()).append('\n')
+            result.append('\n').append(e.getKey().toString()).append('\n')
             for (StackTraceElement s : e.getValue()) {
-                buffer.append("  " + s).append('\n')
+                result.append("  " + s).append('\n')
             }
         }
 
-        return buffer.toString()
+        return result.toString()
     }
 
     static protected String getDumpFile(String file) {

src/main/groovy/io/seqera/wave/encoder/MoshiEncodeStrategy.groovy

@@ -32,12 +32,7 @@ import io.seqera.wave.service.pairing.socket.msg.ProxyHttpResponse
 import io.seqera.wave.storage.DigestStore
 import io.seqera.wave.storage.DockerDigestStore
 import io.seqera.wave.storage.HttpDigestStore
-import io.seqera.wave.storage.LazyDigestStore
 import io.seqera.wave.storage.ZippedDigestStore
-import io.seqera.wave.storage.reader.ContentReader
-import io.seqera.wave.storage.reader.DataContentReader
-import io.seqera.wave.storage.reader.GzipContentReader
-import io.seqera.wave.storage.reader.HttpContentReader
 import io.seqera.wave.util.TypeHelper
 /**
  * Implements a JSON {@link EncodingStrategy} based on Mosh JSON serializer
@@ -71,14 +66,9 @@ abstract class MoshiEncodeStrategy<V> implements EncodingStrategy<V> {
                 .add(new PathAdapter())
                 .add(new UriAdapter())
                 .add(PolymorphicJsonAdapterFactory.of(DigestStore.class, "@type")
-                        .withSubtype(LazyDigestStore, LazyDigestStore.simpleName)
                         .withSubtype(ZippedDigestStore, ZippedDigestStore.simpleName)
                         .withSubtype(HttpDigestStore, HttpDigestStore.simpleName)
                         .withSubtype(DockerDigestStore, DockerDigestStore.simpleName) )
-                .add(PolymorphicJsonAdapterFactory.of(ContentReader.class, "@type")
-                        .withSubtype(DataContentReader.class, DataContentReader.simpleName)
-                        .withSubtype(GzipContentReader.class, GzipContentReader.simpleName)
-                        .withSubtype(HttpContentReader.class, HttpContentReader.simpleName))
                 .add(PolymorphicJsonAdapterFactory.of(PairingMessage.class, "@type")
                         .withSubtype(ProxyHttpRequest.class, ProxyHttpRequest.simpleName)
                         .withSubtype(ProxyHttpResponse.class, ProxyHttpResponse.simpleName)

src/main/groovy/io/seqera/wave/storage/ManifestCacheStore.groovy

@@ -27,7 +27,6 @@ import io.seqera.wave.api.ContainerLayer
 import io.seqera.wave.encoder.MoshiEncodeStrategy
 import io.seqera.wave.service.cache.AbstractCacheStore
 import io.seqera.wave.service.cache.impl.CacheProvider
-import io.seqera.wave.storage.reader.ContentReader
 import jakarta.inject.Singleton
 /**
  * Implements manifest cache for {@link DigestStore}
@@ -97,15 +96,6 @@ class ManifestCacheStore extends AbstractCacheStore<DigestStore> implements Stor
         return result
     }
 
-    @Override
-    @Deprecated
-    DigestStore saveBlob(String path, ContentReader content, String type, String digest, int size) {
-        log.trace "Save Blob ==> $path"
-        final result = new LazyDigestStore(content, type, digest, size)
-        this.put(path, result)
-        return result
-    }
-
     @Override
     DigestStore saveBlob(String path, ContainerLayer layer) {
         log.trace "Save Blob ==> $path; layer=${layer}"

src/main/groovy/io/seqera/wave/storage/Storage.java

@@ -22,7 +22,6 @@
 import java.util.Optional;
 
 import io.seqera.wave.api.ContainerLayer;
-import io.seqera.wave.storage.reader.ContentReader;
 
 /**
  * @author : jorge <[email protected]>
@@ -39,8 +38,5 @@ public interface Storage {
 
     DigestStore saveBlob(String path, byte[] content, String type, String digest);
 
-    @Deprecated
-    DigestStore saveBlob(String path, ContentReader content, String type, String digest, int size);
-
     DigestStore saveBlob(String path, ContainerLayer layer);
 }