Merge branch 'master' into bump-mn-4.7.0

VERSION

@@ -1 +1 @@
-1.14.1
+1.15.1

build.gradle

@@ -49,7 +49,7 @@ dependencies {
     implementation 'io.micronaut:micronaut-websocket'
     implementation 'org.apache.groovy:groovy-json'
     implementation 'org.apache.groovy:groovy-nio'
-    implementation 'com.google.guava:guava:32.1.2-jre'
+    implementation 'com.google.guava:guava:33.3.1-jre'
     implementation 'dev.failsafe:failsafe:3.1.0'
     implementation 'io.micronaut.reactor:micronaut-reactor'
     implementation 'io.micronaut.reactor:micronaut-reactor-http-client'
@@ -156,8 +156,7 @@ jib {
 
 run{
     def envs = findProperty('micronautEnvs')
-    // note: "--enable-preview" is required to use virtual threads on Java 19 and 20
-    def args = ["-Dmicronaut.environments=$envs","--enable-preview"]
+    def args = ["-Dmicronaut.environments=$envs","-Djdk.tracePinnedThreads=short"]
     if( environment['JVM_OPTS'] ) args.add(environment['JVM_OPTS'])
     jvmArgs args
     systemProperties 'DOCKER_USER': project.findProperty('DOCKER_USER') ?: environment['DOCKER_USER'],

changelog.txt

@@ -1,4 +1,19 @@
 # Wave changelog
+1.15.1 - 20 Nov 2024
+- Check block existence with object operation (#750) [86ef526c]
+- Add /v1alpha2/validate-creds (#752) [e24ec62c]
+- Switched env vars around (#748) [080d5cce]
+
+1.15.0 - 18 Nov 2024
+- Migration to virtual threads - phase 1 (#746) [aaf0420c]
+- Use runAsync instead supplyAsync [ffd0dacd]
+- Remove deprecated ThreadPoolBuilder [7af3046f]
+- Replace Guava cache with Caffeine (#745) [cf813e0a]
+- Update project deps [f24b684d]
+- Bump  guava to version 33.3.1-jre [328e9ea3]
+- Bump Netty version 4.1.115.Final [9ba433ce]
+- Bump gradle 8.10.2 [52272fe1]
+
 1.14.1 - 14 Nov 2024
 - Fix creds validation endpoint (#740) [8c0f3a4c]
 

configuration.md

@@ -184,6 +184,16 @@ Rate limit configuration controls the limits of anonymous and authenticated user
 
 - **`redis.pool.enabled`**: whether to enable the Redis pool. It is set to `true` by default, enabling the use of a connection pool for efficient management of connections to the Redis server. *Optional*.
 
+- **`redis.pool.minIdle`**: Specifies the minimum number of idle connections to maintain in the Redis connection pool. The default value is `0`. This ensures that connections are readily available for use.  *Optional*.
+
+- **`redis.pool.maxIdle`**: Specifies the maximum number of idle connections to maintain in the Redis connection pool. The default value is `10`.  *Optional*.
+
+- **`redis.pool.maxTotal`**: Specifies the maximum number of connections that can be maintained in the Redis connection pool. The default value is `50`. This helps to manage resource usage efficiently while supporting high demand.  *Optional*.
+
+- **`redis.client.timeout`**: Defines the timeout duration (in milliseconds) for Redis client operations. The default value is `5000` (5 seconds).  *Optional*.
+
+- **`redis.password`**: Specifies the password used to authenticate with the Redis server. This is needed when redis authentication is enabled.  *Optional*.
+
 - **`surreal.default.ns`**: the namespace for the Surreal database. It can be set using `${SURREALDB_NS}` environment variable. *Mandatory*.
 
 - **`surreal.default.db`**: the name of the Surreal database. It can be set using`${SURREALDB_DB}` environment variable. This setting defines the target database within the Surreal database system that Wave should interact with. *Mandatory*.

docs/cli/index.mdx

@@ -27,8 +27,8 @@ The following CLI arguments are available for Seqera Platform integration:
 
 The following environment variables are available for Seqera Platform integration:
 
-- `TOWER_API_ENDPOINT`: A Seqera Platform auth token so that Wave can access your private registry credentials.
-- `TOWER_ACCESS_TOKEN`: For Enterprise customers, the URL endpoint for your instance, such as `https://api.cloud.seqera.io`.
+- `TOWER_ACCESS_TOKEN`: A Seqera Platform auth token so that Wave can access your private registry credentials.
+- `TOWER_API_ENDPOINT`: For Enterprise customers, the URL endpoint for your instance, such as `https://api.cloud.seqera.io`.
 - `TOWER_WORKSPACE_ID`: A Seqera Platform workspace ID, such as `1234567890`, where credentials may be stored.
 
 ## Usage limits

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

@@ -21,13 +21,12 @@ 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.CompletionException
 import java.util.concurrent.TimeUnit
 
-import com.google.common.cache.CacheBuilder
-import com.google.common.cache.CacheLoader
-import com.google.common.cache.LoadingCache
-import com.google.common.util.concurrent.UncheckedExecutionException
+import com.github.benmanes.caffeine.cache.AsyncLoadingCache
+import com.github.benmanes.caffeine.cache.CacheLoader
+import com.github.benmanes.caffeine.cache.Caffeine
 import groovy.json.JsonSlurper
 import groovy.transform.Canonical
 import groovy.transform.CompileStatic
@@ -101,11 +100,12 @@ class RegistryAuthServiceImpl implements RegistryAuthService {
         return result
     }
 
-    private LoadingCache<CacheKey, String> cacheTokens = CacheBuilder<CacheKey, String>
+    // FIXME https://github.com/seqeralabs/wave/issues/747
+    private AsyncLoadingCache<CacheKey, String> cacheTokens = Caffeine.newBuilder()
                     .newBuilder()
                     .maximumSize(10_000)
                     .expireAfterAccess(_1_HOUR.toMillis(), TimeUnit.MILLISECONDS)
-                    .build(loader)
+                    .buildAsync(loader)
 
     @Inject
     private RegistryLookupService lookupService
@@ -269,9 +269,10 @@ class RegistryAuthServiceImpl implements RegistryAuthService {
     protected String getAuthToken(String image, RegistryAuth auth, RegistryCredentials creds) {
         final key = new CacheKey(image, auth, creds)
         try {
-            return cacheTokens.get(key)
+            // FIXME https://github.com/seqeralabs/wave/issues/747
+            return cacheTokens.synchronous().get(key)
         }
-        catch (UncheckedExecutionException | ExecutionException e) {
+        catch (CompletionException e) {
             // this catches the exception thrown in the cache loader lookup
             // and throws the causing exception that should be `RegistryUnauthorizedAccessException`
             throw e.cause
@@ -287,7 +288,8 @@ class RegistryAuthServiceImpl implements RegistryAuthService {
      */
     void invalidateAuthorization(String image, RegistryAuth auth, RegistryCredentials creds) {
         final key = new CacheKey(image, auth, creds)
-        cacheTokens.invalidate(key)
+        // FIXME https://github.com/seqeralabs/wave/issues/747
+        cacheTokens.synchronous().invalidate(key)
         tokenStore.remove(getStableKey(key))
     }
 

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

@@ -20,13 +20,12 @@ package io.seqera.wave.auth
 
 import java.net.http.HttpRequest
 import java.net.http.HttpResponse
-import java.util.concurrent.ExecutionException
+import java.util.concurrent.CompletionException
 import java.util.concurrent.TimeUnit
 
-import com.google.common.cache.CacheBuilder
-import com.google.common.cache.CacheLoader
-import com.google.common.cache.LoadingCache
-import com.google.common.util.concurrent.UncheckedExecutionException
+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.seqera.wave.configuration.HttpClientConfig
@@ -74,11 +73,12 @@ class RegistryLookupServiceImpl implements RegistryLookupService {
         }
     }
 
-    private LoadingCache<URI, RegistryAuth> cache = CacheBuilder<URI, RegistryAuth>
+    // FIXME https://github.com/seqeralabs/wave/issues/747
+    private AsyncLoadingCache<URI, RegistryAuth> cache = Caffeine.newBuilder()
                 .newBuilder()
                 .maximumSize(10_000)
                 .expireAfterAccess(1, TimeUnit.HOURS)
-                .build(loader)
+                .buildAsync(loader)
 
     protected RegistryAuth lookup0(URI endpoint) {
         final httpClient = HttpClientFactory.followRedirectsHttpClient()
@@ -117,10 +117,11 @@ class RegistryLookupServiceImpl implements RegistryLookupService {
     RegistryInfo lookup(String registry) {
         try {
             final endpoint = registryEndpoint(registry)
-            final auth = cache.get(endpoint)
+            // FIXME https://github.com/seqeralabs/wave/issues/747
+            final auth = cache.synchronous().get(endpoint)
             return new RegistryInfo(registry, endpoint, auth)
         }
-        catch (UncheckedExecutionException | ExecutionException e) {
+        catch (CompletionException e) {
             // this catches the exception thrown in the cache loader lookup
             // and throws the causing exception that should be `RegistryUnauthorizedAccessException`
             throw e.cause

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

@@ -42,7 +42,7 @@ import jakarta.inject.Inject
 @Slf4j
 @CompileStatic
 @Controller("/")
-@ExecuteOn(TaskExecutors.IO)
+@ExecuteOn(TaskExecutors.BLOCKING)
 class BuildController {
 
     @Inject

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

@@ -103,7 +103,7 @@ import static java.util.concurrent.CompletableFuture.completedFuture
 @Slf4j
 @CompileStatic
 @Controller("/")
-@ExecuteOn(TaskExecutors.IO)
+@ExecuteOn(TaskExecutors.BLOCKING)
 class ContainerController {
 
     @Inject
@@ -181,13 +181,13 @@ class ContainerController {
 
     @Deprecated
     @Post('/container-token')
-    @ExecuteOn(TaskExecutors.IO)
+    @ExecuteOn(TaskExecutors.BLOCKING)
     CompletableFuture<HttpResponse<SubmitContainerTokenResponse>> getToken(HttpRequest httpRequest, @Body SubmitContainerTokenRequest req) {
         return getContainerImpl(httpRequest, req, false)
     }
 
     @Post('/v1alpha2/container')
-    @ExecuteOn(TaskExecutors.IO)
+    @ExecuteOn(TaskExecutors.BLOCKING)
     CompletableFuture<HttpResponse<SubmitContainerTokenResponse>> getTokenV2(HttpRequest httpRequest, @Body SubmitContainerTokenRequest req) {
         return getContainerImpl(httpRequest, req, true)
     }

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

@@ -50,7 +50,7 @@ import static io.seqera.wave.util.ContainerHelper.patchPlatformEndpoint
 @Slf4j
 @CompileStatic
 @Controller("/")
-@ExecuteOn(TaskExecutors.IO)
+@ExecuteOn(TaskExecutors.BLOCKING)
 class InspectController {
 
     @Inject

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

@@ -52,7 +52,7 @@ import static io.micronaut.http.HttpHeaders.WWW_AUTHENTICATE
 @Requires(property = 'wave.metrics.enabled', value = 'true')
 @Secured(SecurityRule.IS_AUTHENTICATED)
 @Controller
-@ExecuteOn(TaskExecutors.IO)
+@ExecuteOn(TaskExecutors.BLOCKING)
 class MetricsController {
 
     @Inject

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

@@ -36,7 +36,7 @@ import jakarta.inject.Inject
 @Slf4j
 @CompileStatic
 @Controller("/")
-@ExecuteOn(TaskExecutors.IO)
+@ExecuteOn(TaskExecutors.BLOCKING)
 class MirrorController {
 
     @Inject

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

@@ -67,7 +67,7 @@ import reactor.core.publisher.Mono
 @Slf4j
 @CompileStatic
 @Controller("/v2")
-@ExecuteOn(TaskExecutors.IO)
+@ExecuteOn(TaskExecutors.BLOCKING)
 class RegistryProxyController {
 
     @Inject

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

@@ -39,7 +39,7 @@ import jakarta.inject.Inject
 @CompileStatic
 @Requires(bean = ContainerScanService)
 @Controller("/")
-@ExecuteOn(TaskExecutors.IO)
+@ExecuteOn(TaskExecutors.BLOCKING)
 class ScanController {
 
     @Inject

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

@@ -39,7 +39,7 @@ import io.seqera.wave.util.BuildInfo
 @Slf4j
 @Controller("/")
 @CompileStatic
-@ExecuteOn(TaskExecutors.IO)
+@ExecuteOn(TaskExecutors.BLOCKING)
 class ServiceInfoController {
 
     @Value('${wave.landing.url}')

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

@@ -18,6 +18,7 @@
 
 package io.seqera.wave.controller
 
+import io.micronaut.http.annotation.Body
 import io.micronaut.http.annotation.Controller
 import io.micronaut.http.annotation.Post
 import io.micronaut.scheduling.TaskExecutors
@@ -26,15 +27,21 @@ import io.seqera.wave.auth.RegistryAuthService
 import jakarta.inject.Inject
 import jakarta.validation.Valid
 
-@ExecuteOn(TaskExecutors.IO)
-@Controller("/validate-creds")
+@Controller("/")
+@ExecuteOn(TaskExecutors.BLOCKING)
 class ValidateController {
 
     @Inject RegistryAuthService loginService
 
-    @Post
+    @Deprecated
+    @Post("/validate-creds")
     Boolean validateCreds(@Valid ValidateRegistryCredsRequest request){
         loginService.validateUser(request.registry, request.userName, request.password)
     }
 
+    @Post("/v1alpha2/validate-creds")
+    Boolean validateCredsV2(@Valid @Body ValidateRegistryCredsRequest request){
+        loginService.validateUser(request.registry, request.userName, request.password)
+    }
+
 }

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

@@ -59,7 +59,7 @@ import static io.seqera.wave.util.DataTimeUtils.formatTimestamp
 @Slf4j
 @CompileStatic
 @Controller("/view")
-@ExecuteOn(TaskExecutors.IO)
+@ExecuteOn(TaskExecutors.BLOCKING)
 class ViewController {
 
     @Inject

src/main/groovy/io/seqera/wave/core/ContainerAugmenter.groovy

@@ -280,7 +280,7 @@ class ContainerAugmenter {
         return result
     }
 
-    synchronized protected Map layerBlob(String image, ContainerLayer layer) {
+    protected Map layerBlob(String image, ContainerLayer layer) {
         log.debug "Adding layer: $layer to image: $client.registry.name/$image"
         // store the layer blob in the cache
         final String path = "$client.registry.name/v2/$image/blobs/$layer.gzipDigest"
@@ -295,7 +295,6 @@ class ContainerAugmenter {
 
 
     protected Tuple2<String,Integer> updateImageManifest(String imageName, String imageManifest, String newImageConfigDigest, newImageConfigSize, boolean oci) {
-
         // turn the json string into a json map
         // and append the new layer
         final manifest = (Map) new JsonSlurper().parseText(imageManifest)

src/main/groovy/io/seqera/wave/core/RegistryProxyService.groovy

@@ -18,6 +18,8 @@
 
 package io.seqera.wave.core
 
+import java.util.concurrent.CompletableFuture
+
 import groovy.transform.CompileStatic
 import groovy.transform.ToString
 import groovy.util.logging.Slf4j
@@ -193,7 +195,7 @@ class RegistryProxyService {
 
     String getImageDigest(String containerImage, PlatformId identity, boolean retryOnNotFound=false) {
         try {
-            return getImageDigest0(containerImage, identity, retryOnNotFound)
+            return getImageDigest0(containerImage, identity, retryOnNotFound).get()
         }
         catch(Exception e) {
             log.warn "Unable to retrieve digest for image '${containerImage}' -- cause: ${e.message}"
@@ -203,8 +205,15 @@ class RegistryProxyService {
 
     static private List<Integer> RETRY_ON_NOT_FOUND = HTTP_RETRYABLE_ERRORS + 404
 
+    // note: return a CompletableFuture to force micronaut to use caffeine AsyncCache
+    // that provides a workaround about the use of virtual threads with SyncCache
+    // see https://github.com/ben-manes/caffeine/issues/1468#issuecomment-1906733926
     @Cacheable(value = 'cache-registry-proxy', atomic = true, parameters = ['image'])
-    protected String getImageDigest0(String image, PlatformId identity, boolean retryOnNotFound) {
+    protected CompletableFuture<String> getImageDigest0(String image, PlatformId identity, boolean retryOnNotFound) {
+        CompletableFuture.completedFuture(getImageDigest1(image, identity, retryOnNotFound))
+    }
+
+    protected String getImageDigest1(String image, PlatformId identity, boolean retryOnNotFound) {
         final coords = ContainerCoordinates.parse(image)
         final route = RoutePath.v2manifestPath(coords, identity)
         final proxyClient = client(route)

src/main/groovy/io/seqera/wave/filter/PullMetricsRequestsFilter.groovy

@@ -83,10 +83,10 @@ class PullMetricsRequestsFilter implements HttpServerFilter {
         final contentType = response.headers.get(HttpHeaders.CONTENT_TYPE)
         if( contentType && contentType in MANIFEST_TYPES ) {
             final route = routeHelper.parse(request.path)
-            CompletableFuture.supplyAsync(() -> metricsService.incrementPullsCounter(route.identity), executor)
+            CompletableFuture.runAsync(() -> metricsService.incrementPullsCounter(route.identity), executor)
             final version = route.request?.containerConfig?.fusionVersion()
             if (version) {
-                CompletableFuture.supplyAsync(() -> metricsService.incrementFusionPullsCounter(route.identity), executor)
+                CompletableFuture.runAsync(() -> metricsService.incrementFusionPullsCounter(route.identity), executor)
             }
         }
     }