kfaraz commented on code in PR #18599: URL: https://github.com/apache/druid/pull/18599#discussion_r2596893365
########## extensions-core/kubernetes-overlord-extensions/src/main/java/org/apache/druid/k8s/overlord/common/CachingKubernetesPeonClient.java: ########## @@ -0,0 +1,284 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you 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 org.apache.druid.k8s.overlord.common; + +import com.google.common.base.Optional; +import io.fabric8.kubernetes.api.model.Pod; +import io.fabric8.kubernetes.api.model.batch.v1.Job; +import org.apache.druid.error.DruidException; +import org.apache.druid.java.util.emitter.EmittingLogger; +import org.apache.druid.java.util.emitter.service.ServiceEmitter; + +import javax.annotation.Nullable; +import java.util.List; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; + +/** + * A KubernetesPeonClient implementation that uses shared informers to read Job and Pod state from a local cache. + * <p> + * This implementation greatly reduces load on the Kubernetes API server by centralizing watches and allowing + * tasks to query cached resource state instead of making per-task API calls. Mutable operations (job creation, + * deletion) still contact the API server directly. + * </p> + */ +public class CachingKubernetesPeonClient extends AbstractKubernetesPeonClient +{ + protected static final EmittingLogger log = new EmittingLogger(CachingKubernetesPeonClient.class); + + public CachingKubernetesPeonClient( + KubernetesClientApi clientApi, + String namespace, + String overlordNamespace, + boolean debugJobs, + ServiceEmitter emitter + ) + { + super(clientApi, namespace, overlordNamespace, debugJobs, emitter); + } + + public CachingKubernetesPeonClient( + KubernetesClientApi clientApi, + String namespace, + boolean debugJobs, + ServiceEmitter emitter + ) + { + super(clientApi, namespace, "", debugJobs, emitter); + } + + @Override + public JobResponse waitForPeonJobCompletion(K8sTaskId taskId, long howLong, TimeUnit unit) + { + long timeoutMs = unit.toMillis(howLong); + long startTime = System.currentTimeMillis(); + + // Give the informer 2 resync periods to see the job. if it isn't seen by then, we assume the job was canceled. + // This is to prevent us from waiting for entire max job runtime on a job that was canceled before it even started. + long jobMustBeSeenBy = startTime + (clientApi.getInformerResyncPeriodMillis() * 2); + boolean jobSeenInCache = false; + + // Set up to watch for job changes + CompletableFuture<Job> jobFuture = clientApi.getEventNotifier().waitForJobChange(taskId.getK8sJobName()); + + // We will loop until the full timeout is reached if the job is seen in cache. If the job does not show up in the cache we will exit earlier. + // In this loop we first check the cache to see if our job is there and complete. This avoids missing notifications that happened before we set up the watch. + // If the job is not complete we wait for a notification of a job change or a timeout. + // If it is a timeout, we loop back to check the cache again. + // If it is a job change notification, we check the job state and exit if complete, or loop again if still running. + do { + try { + Optional<Job> maybeJob = getPeonJob(taskId.getK8sJobName()); + if (maybeJob.isPresent()) { + jobSeenInCache = true; + Job job = maybeJob.get(); + JobResponse currentResponse = determineJobResponse(job); + if (currentResponse.getPhase() != PeonPhase.RUNNING) { + return currentResponse; + } else { + log.debug("K8s job[%s] found in cache and is still running", taskId.getK8sJobName()); + } + } else if (jobSeenInCache) { + // Job was in cache before, but now it's gone - it was deleted and will never complete. + log.warn("K8s Job[%s] was not found. It can happen if the task was canceled", taskId.getK8sJobName()); + return new JobResponse(null, PeonPhase.FAILED); + } + + // We wake up every informer resync period to avoid event notifier misses. + Job job = jobFuture.get(clientApi.getInformerResyncPeriodMillis(), TimeUnit.MILLISECONDS); + + // Immediately set up to watch for the next change in case we need to wait again + jobFuture = clientApi.getEventNotifier().waitForJobChange(taskId.getK8sJobName()); + log.debug("Received job[%s] change notification", taskId.getK8sJobName()); + jobSeenInCache = true; + + if (job == null) { + log.warn("K8s job[%s] was not found. It can happen if the task was canceled", taskId.getK8sJobName()); + return new JobResponse(null, PeonPhase.FAILED); + } + + JobResponse currentResponse = determineJobResponse(job); + if (currentResponse.getPhase() != PeonPhase.RUNNING) { + return currentResponse; + } else { + log.debug("K8s job[%s] is still running", taskId.getK8sJobName()); + } + } + catch (TimeoutException e) { + // A timeout here is not a problem, it forces us to loop around and check the cache again. Review Comment: Yeah, if we can be sure that only one thread is going to perform this code at a time (by putting this behind a taskId gate), we can get rid of this complexity. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
