Fix K8s token refresh by caching K8sClient at executor level

Move the Guava cache from K8sConfig to K8sExecutor so that the
K8sClient object itself is cached (not just the ClientConfig).
This avoids re-creating K8sClient (including SSL setup) on every
invocation while still refreshing the service account token when
the configured interval expires.

Fixes #6918

Signed-off-by: Paolo Di Tommaso <paolo.ditommaso@gmail.com>

Author: pditommaso

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

@@ -18,11 +18,8 @@ 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.PackageScope
 import groovy.util.logging.Slf4j
@@ -57,8 +54,6 @@ 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.
@@ -226,9 +221,6 @@ class K8sConfig implements ConfigScope {
         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
@@ -366,10 +358,6 @@ class K8sConfig implements ConfigScope {
     }
 
     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/main/nextflow/k8s/K8sExecutor.groovy

@@ -16,6 +16,10 @@
 
 package nextflow.k8s
 
+import java.util.concurrent.TimeUnit
+
+import com.google.common.cache.Cache
+import com.google.common.cache.CacheBuilder
 import groovy.transform.CompileStatic
 import groovy.transform.Memoized
 import groovy.util.logging.Slf4j
@@ -41,12 +45,17 @@ import org.pf4j.ExtensionPoint
 class K8sExecutor extends Executor implements ExtensionPoint {
 
     /**
-     * The Kubernetes HTTP client
+     * Cache for the Kubernetes HTTP client. The client is refreshed periodically
+     * so that the service account token is re-read when it expires.
      */
-    private K8sClient client
+    private Cache<String, K8sClient> clientCache
 
+    /**
+     * @return The Kubernetes HTTP client. Delegates to a Guava cache that refreshes
+     * the client (including the service account token) when the configured interval expires.
+     */
     protected K8sClient getClient() {
-        client
+        clientCache.get('client', () -> new K8sClient(k8sConfig.getClient()))
     }
 
     /**
@@ -64,9 +73,12 @@ class K8sExecutor extends Executor implements ExtensionPoint {
     protected void register() {
         super.register()
         final k8sConfig = getK8sConfig()
-        final clientConfig = k8sConfig.getClient()
-        this.client = new K8sClient(clientConfig)
-        log.debug "[K8s] config=$k8sConfig; API client config=$clientConfig"
+        final refreshInterval = k8sConfig.clientRefreshInterval
+        this.clientCache = CacheBuilder.newBuilder()
+            .expireAfterWrite(refreshInterval.toMillis(), TimeUnit.MILLISECONDS)
+            .build()
+        final client = getClient()
+        log.debug "[K8s] config=$k8sConfig; API client config=$client.config"
     }
 
     /**

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

@@ -93,7 +93,7 @@ class K8sTaskHandler extends TaskHandler implements FusionAwareTask {
     K8sTaskHandler( TaskRun task, K8sExecutor executor ) {
         super(task)
         this.executor = executor
-        this.client = executor.client
+        this.client = executor.getClient()
         this.outputFile = task.workDir.resolve(TaskRun.CMD_OUTFILE)
         this.errorFile = task.workDir.resolve(TaskRun.CMD_ERRFILE)
         this.exitFile = task.workDir.resolve(TaskRun.CMD_EXIT)

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

@@ -484,33 +484,4 @@ class K8sConfigTest extends Specification {
         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)
-    }
 }

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

@@ -0,0 +1,66 @@
+/*
+ * Copyright 2013-2026, Seqera Labs
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package nextflow.k8s
+
+import java.util.concurrent.TimeUnit
+
+import com.google.common.cache.CacheBuilder
+import nextflow.k8s.client.ClientConfig
+import nextflow.k8s.client.K8sClient
+import spock.lang.Specification
+
+/**
+ * @author Paolo Di Tommaso <paolo.ditommaso@gmail.com>
+ */
+class K8sExecutorTest extends Specification {
+
+    def 'should cache k8s client and refresh after expiration' () {
+        given:
+        def CONFIG = new K8sConfig(
+            client: [server: 'http://k8s-server'],
+            namespace: 'test-ns',
+            serviceAccount: 'test-sa',
+            clientRefreshInterval: '100ms'
+        )
+        and:
+        def executor = Spy(K8sExecutor)
+        executor.getK8sConfig() >> CONFIG
+        // use a short-lived cache for the test
+        executor.@clientCache = CacheBuilder.newBuilder()
+            .expireAfterWrite(100, TimeUnit.MILLISECONDS)
+            .build()
+
+        when: 'first call to getClient'
+        def client1 = executor.getClient()
+        then: 'a new K8sClient is created'
+        client1 instanceof K8sClient
+        client1.config.server == 'http://k8s-server'
+
+        when: 'second call within cache interval'
+        def client2 = executor.getClient()
+        then: 'returns the same cached instance'
+        client2.is(client1)
+
+        when: 'call after cache expiration'
+        sleep(150)
+        def client3 = executor.getClient()
+        then: 'a new K8sClient instance is created'
+        client3 instanceof K8sClient
+        !client3.is(client1)
+    }
+
+}