Add time-based caching for K8sConfig.getClient() (nextflow-io#6742)

Co-authored-by: Jorge Ejarque <jorgee@users.noreply.github.com>

Author: pditommaso

docs/reference/config.md

@@ -1057,6 +1057,12 @@ The following settings are available:
   - `clientKey`
   - `clientKeyFile`
 
+`k8s.clientRefreshInterval`
+: :::{versionadded} 26.01.0-edge
+  :::
+: The interval after which the Kubernetes client configuration is refreshed (default: `50m`).
+: This setting is useful when the Kubernetes authentication token has a limited lifespan and needs to be periodically refreshed. The client configuration will be automatically reloaded after the specified interval, allowing Nextflow to obtain fresh credentials from the Kubernetes configuration.
+
 `k8s.computeResourceType`
 : :::{versionadded} 22.05.0-edge
   :::

plugins/nf-k8s/src/main/nextflow/k8s/K8sConfig.groovy

@@ -18,10 +18,12 @@ package nextflow.k8s
 
 import nextflow.k8s.client.K8sRetryConfig
 
+import java.util.concurrent.TimeUnit
 import javax.annotation.Nullable
 
+import com.google.common.cache.Cache
+import com.google.common.cache.CacheBuilder
 import groovy.transform.CompileStatic
-import groovy.transform.Memoized
 import groovy.transform.PackageScope
 import groovy.util.logging.Slf4j
 import nextflow.BuildInfo
@@ -55,6 +57,8 @@ class K8sConfig implements ConfigScope {
 
     static final private Map<String,?> DEFAULT_FUSE_PLUGIN = Map.of('nextflow.io/fuse', 1)
 
+    private Cache<String, ClientConfig> clientCache
+
     @ConfigOption
     @Description("""
         Automatically mount host paths into the task pods (default: `false`). Only intended for development purposes when using a single node.
@@ -81,6 +85,12 @@ class K8sConfig implements ConfigScope {
     """)
     final Map client
 
+    @ConfigOption
+    @Description("""
+        The interval after which the Kubernetes client configuration is refreshed (default: `50m`).
+    """)
+    final Duration clientRefreshInterval
+
     @ConfigOption
     @Description("""
         The Kubernetes [configuration context](https://kubernetes.io/docs/tasks/access-application-cluster/configure-access-multiple-clusters/) to use.
@@ -215,6 +225,10 @@ class K8sConfig implements ConfigScope {
         autoMountHostPaths = opts.autoMountHostPaths as boolean
         cleanup = opts.cleanup as Boolean
         client = opts.client as Map
+        clientRefreshInterval = opts.clientRefreshInterval as Duration ?: Duration.of('50m')
+        clientCache = CacheBuilder.newBuilder()
+            .expireAfterWrite(clientRefreshInterval.toMillis(), TimeUnit.MILLISECONDS)
+            .build()
         computeResourceType = opts.computeResourceType
         context = opts.context
         cpuLimits = opts.cpuLimits as boolean
@@ -351,9 +365,11 @@ class K8sConfig implements ConfigScope {
         return result ? result.claimName : null
     }
 
-    @Memoized
     ClientConfig getClient() {
+        return clientCache.get('client', this::getClient0)
+    }
 
+    private ClientConfig getClient0() {
         final result = client != null
                 ? clientFromNextflow(client, namespace, serviceAccount)
                 : clientDiscovery(context, namespace, serviceAccount)

plugins/nf-k8s/src/test/nextflow/k8s/K8sConfigTest.groovy

@@ -466,4 +466,51 @@ class K8sConfigTest extends Specification {
         then:
         cfg.fetchNodeName() == false
     }
+
+    def 'should set clientRefreshInterval' () {
+        when:
+        def cfg = new K8sConfig()
+        then:
+        cfg.clientRefreshInterval == Duration.of('50m')
+
+        when:
+        cfg = new K8sConfig(clientRefreshInterval: '30m')
+        then:
+        cfg.clientRefreshInterval == Duration.of('30m')
+
+        when:
+        cfg = new K8sConfig(clientRefreshInterval: '1h')
+        then:
+        cfg.clientRefreshInterval == Duration.of('1h')
+    }
+
+    def 'should cache client config and refresh after expiration' () {
+        given:
+        def CONFIG = [
+            namespace: 'test-ns',
+            serviceAccount: 'test-sa',
+            client: [server: 'http://k8s-server'],
+            clientRefreshInterval: '100ms'
+        ]
+        K8sConfig config = Spy(K8sConfig, constructorArgs: [CONFIG])
+
+        when: 'first call to getClient'
+        def client1 = config.getClient()
+        then: 'client is created via clientFromNextflow'
+        1 * config.clientFromNextflow(_, _, _) >> new ClientConfig(server: 'http://k8s-server', namespace: 'test-ns')
+        client1.server == 'http://k8s-server'
+
+        when: 'second call within cache interval'
+        def client2 = config.getClient()
+        then: 'returns cached client without calling clientFromNextflow again'
+        0 * config.clientFromNextflow(_, _, _)
+        client2.is(client1)
+
+        when: 'call after cache expiration'
+        sleep(150) // wait for cache to expire
+        def client3 = config.getClient()
+        then: 'client is recreated'
+        1 * config.clientFromNextflow(_, _, _) >> new ClientConfig(server: 'http://k8s-server', namespace: 'test-ns')
+        !client3.is(client1)
+    }
 }