From 0483405e344f4b0525eac42174fcc0d92438b16b Mon Sep 17 00:00:00 2001 From: Tanishq Gandhi Date: Thu, 10 Sep 2026 19:12:35 +0000 Subject: [PATCH 1/7] feat(computing-unit): pre-pull curated images onto every node The first unit on a node waits for the whole image, about 80 seconds for a 3 GB one, while every later unit there starts at once. Each ready image now gets a DaemonSet whose init container is the image and whose command does nothing, plus a pause container so the node does not reclaim it. Created when an image becomes ready, repointed when a refresh resolves a new digest, and removed with the image. Best-effort: if it cannot start, the image still works and the first unit pays for the pull. --- ...low-computing-unit-manager-deployment.yaml | 5 + ...omputing-unit-manager-service-account.yaml | 5 + bin/k8s/values.yaml | 5 + .../config/src/main/resources/kubernetes.conf | 23 ++ .../common/config/CuratedImageConfig.scala | 13 ++ .../resource/CuratedImageResource.scala | 55 ++++- .../service/util/ImagePrepullClient.scala | 200 ++++++++++++++++++ .../service/util/ImagePrepullClientSpec.scala | 110 ++++++++++ 8 files changed, 414 insertions(+), 2 deletions(-) create mode 100644 computing-unit-managing-service/src/main/scala/org/apache/texera/service/util/ImagePrepullClient.scala create mode 100644 computing-unit-managing-service/src/test/scala/org/apache/texera/service/util/ImagePrepullClientSpec.scala diff --git a/bin/k8s/templates/base/workflow-computing-unit-manager/workflow-computing-unit-manager-deployment.yaml b/bin/k8s/templates/base/workflow-computing-unit-manager/workflow-computing-unit-manager-deployment.yaml index 6ad4eeb4be3..ca13b4af31b 100644 --- a/bin/k8s/templates/base/workflow-computing-unit-manager/workflow-computing-unit-manager-deployment.yaml +++ b/bin/k8s/templates/base/workflow-computing-unit-manager/workflow-computing-unit-manager-deployment.yaml @@ -71,6 +71,11 @@ spec: # Must be the namespace the pool runs in; validation jobs go there. - name: TEXERA_CURATED_IMAGE_VALIDATION_NAMESPACE value: {{ .Values.workflowComputingUnitPool.namespace }} + - name: TEXERA_CURATED_IMAGE_PREPULL_ENABLED + value: "{{ .Values.curatedImages.prepull.enabled }}" + # The same namespace, from the same value, so the two cannot drift apart. + - name: TEXERA_CURATED_IMAGE_PREPULL_NAMESPACE + value: {{ .Values.workflowComputingUnitPool.namespace }} - name: KUBERNETES_IMAGE_NAME value: {{ .Values.texera.imageRegistry }}/{{ .Values.workflowComputingUnitPool.imageName }}:{{ .Values.texera.imageTag }} - name: KUBERNETES_MOUNTER_ENABLED diff --git a/bin/k8s/templates/base/workflow-computing-unit-manager/workflow-computing-unit-manager-service-account.yaml b/bin/k8s/templates/base/workflow-computing-unit-manager/workflow-computing-unit-manager-service-account.yaml index 29888af9640..6e3d61554d9 100644 --- a/bin/k8s/templates/base/workflow-computing-unit-manager/workflow-computing-unit-manager-service-account.yaml +++ b/bin/k8s/templates/base/workflow-computing-unit-manager/workflow-computing-unit-manager-service-account.yaml @@ -42,6 +42,11 @@ rules: - apiGroups: [""] resources: ["pods/log"] verbs: ["get"] + # One DaemonSet per ready curated image, pulling it onto every node. Created while the + # cluster runs, as images are registered, so the chart cannot declare them. + - apiGroups: ["apps"] + resources: ["daemonsets"] + verbs: ["get", "list", "watch", "create", "update", "patch", "delete"] --- apiVersion: rbac.authorization.k8s.io/v1 diff --git a/bin/k8s/values.yaml b/bin/k8s/values.yaml index 438821b82d2..3d11ab496e1 100644 --- a/bin/k8s/values.yaml +++ b/bin/k8s/values.yaml @@ -376,6 +376,11 @@ litellm: curatedImages: # Off until the UI to manage these ships. enabled: false + prepull: + # Pull every ready image onto every node, so the first unit on a node does not wait for + # it. Costs node disk: each node holds each ready image. Turn off to pay the pull on + # first use instead. + enabled: true # headless service for the access of computing units workflowComputingUnitPool: diff --git a/common/config/src/main/resources/kubernetes.conf b/common/config/src/main/resources/kubernetes.conf index c27fa40d049..1b6c4b30658 100644 --- a/common/config/src/main/resources/kubernetes.conf +++ b/common/config/src/main/resources/kubernetes.conf @@ -151,4 +151,27 @@ curated-images { validation-cpu-limit = ${?TEXERA_CURATED_IMAGE_VALIDATION_CPU_LIMIT} validation-memory-limit = "256Mi" validation-memory-limit = ${?TEXERA_CURATED_IMAGE_VALIDATION_MEMORY_LIMIT} + + # Pull each ready image onto every node as soon as it is ready, rather than when someone + # first starts a unit from it. Costs node disk -- every node holds every ready image -- + # so a deployment short of it can turn this off and pay the pull on first use instead. + prepull-enabled = true + prepull-enabled = ${?TEXERA_CURATED_IMAGE_PREPULL_ENABLED} + + # Where the pre-pull DaemonSets run. The pool namespace, like validation: a node caches + # an image whatever namespace asked for it, so this only decides where the pods are seen. + prepull-namespace = "texera-workflow-computing-unit-pool" + prepull-namespace = ${?TEXERA_CURATED_IMAGE_PREPULL_NAMESPACE} + + # Holds the pod open once the init container has pulled the image, so the node does not + # reclaim what was just pulled. registry.k8s.io is where the pause image now lives; the + # gcr.io/google_containers path the chart's own pre-puller still uses is retired. + prepull-pause-image = "registry.k8s.io/pause:3.9" + prepull-pause-image = ${?TEXERA_CURATED_IMAGE_PREPULL_PAUSE_IMAGE} + + # The pause container does nothing but exist. Same figures the chart's pre-puller uses. + prepull-cpu = "1m" + prepull-cpu = ${?TEXERA_CURATED_IMAGE_PREPULL_CPU} + prepull-memory = "8Mi" + prepull-memory = ${?TEXERA_CURATED_IMAGE_PREPULL_MEMORY} } diff --git a/common/config/src/main/scala/org/apache/texera/common/config/CuratedImageConfig.scala b/common/config/src/main/scala/org/apache/texera/common/config/CuratedImageConfig.scala index 9dc85da9d66..993cbdc7f6b 100644 --- a/common/config/src/main/scala/org/apache/texera/common/config/CuratedImageConfig.scala +++ b/common/config/src/main/scala/org/apache/texera/common/config/CuratedImageConfig.scala @@ -44,4 +44,17 @@ object CuratedImageConfig { /** Kubernetes object name for one check. Unique per attempt so retries never collide. */ def validationJobName(iid: Int, attempt: Int): String = s"cu-image-check-$iid-$attempt" + + val prepullEnabled: Boolean = conf.getBoolean("curated-images.prepull-enabled") + val prepullNamespace: String = conf.getString("curated-images.prepull-namespace") + val prepullPauseImage: String = conf.getString("curated-images.prepull-pause-image") + val prepullCpu: String = conf.getString("curated-images.prepull-cpu") + val prepullMemory: String = conf.getString("curated-images.prepull-memory") + + /** + * Kubernetes object name for one image's pre-pull. One per image and stable across its + * refreshes, so a new digest replaces the pre-pull rather than adding a second one + * holding the bytes nothing runs any more. + */ + def prepullName(iid: Int): String = s"cu-image-prepull-$iid" } diff --git a/computing-unit-managing-service/src/main/scala/org/apache/texera/service/resource/CuratedImageResource.scala b/computing-unit-managing-service/src/main/scala/org/apache/texera/service/resource/CuratedImageResource.scala index 8c19a0138dd..55dcb9dc406 100644 --- a/computing-unit-managing-service/src/main/scala/org/apache/texera/service/resource/CuratedImageResource.scala +++ b/computing-unit-managing-service/src/main/scala/org/apache/texera/service/resource/CuratedImageResource.scala @@ -27,7 +27,7 @@ import jakarta.ws.rs.core.MediaType import org.apache.texera.auth.SessionUser import org.apache.texera.common.config.CuratedImageConfig import org.apache.texera.dao.SqlServer -import org.apache.texera.service.util.ImageValidationClient +import org.apache.texera.service.util.{ImagePrepullClient, ImageValidationClient} import org.apache.texera.service.util.ImageValidationClient.ValidationState import org.jooq.impl.DSL import org.jooq.{DSLContext, Record} @@ -197,6 +197,44 @@ object CuratedImageResource extends LazyLogging { def nameOf(iid: Int): Option[String] = Option(context.select(NAME).from(CU_IMAGE).where(IID.eq(iid)).fetchOne()).map(_.get(NAME)) + private def sourceRefOf(iid: Int): Option[String] = + Option(context.select(SOURCE_REF).from(CU_IMAGE).where(IID.eq(iid)).fetchOne()) + .map(_.get(SOURCE_REF)) + + /** + * Gives a pre-pull to every ready image that has none. A row can reach READY without one: + * registered before pre-pulling shipped, or finished while the cluster could not be + * reached. Without this the image would never be pre-pulled at all, since nothing else + * revisits a row once it is ready. + * + * One labelled list, then only the images actually missing, so a settled deployment + * costs a single call. + */ + private def ensurePrepullsForReadyImages(): Unit = { + if (!CuratedImageConfig.prepullEnabled) return + try { + val alreadyPrepulled = ImagePrepullClient.prepulledImageIds() + context + .select(IID, SOURCE_REF, SOURCE_DIGEST) + .from(CU_IMAGE) + .where(STATUS.eq(Status.Ready)) + .fetch() + .asScala + .foreach { row => + val iid = row.get(IID).intValue() + if (!alreadyPrepulled.contains(iid)) { + pinnedRefOf(row.get(SOURCE_REF), row.get(SOURCE_DIGEST)) + .foreach(ImagePrepullClient.ensurePrepull(iid, _)) + } + } + } catch { + // Opportunistic, like reconciling: a listing that cannot be repaired is still worth + // returning. + case e: Throwable => + logger.warn("Could not check the pre-pulls of ready images; leaving them as they are.", e) + } + } + /** * Brings VALIDATING rows up to date with what the cluster did. Validation finishes on the * cluster, so a row learns its outcome when someone reads it -- no background threads or @@ -401,7 +439,16 @@ object CuratedImageResource extends LazyLogging { val stored = withDigest.where(IID.eq(iid).and(ATTEMPT.eq(attempt))).execute() // The job is kept only until its outcome is recorded, so finished jobs do not pile up // in the pool namespace. - if (stored > 0) ImageValidationClient.deleteValidation(iid, attempt) + if (stored > 0) { + ImageValidationClient.deleteValidation(iid, attempt) + // Only on READY, because only then is there something a unit can be started from. + // A refresh that resolved a moved tag arrives here too, and points the pre-pull at + // the new digest rather than leaving the node holding bytes nothing runs. + if (status == Status.Ready) { + pinnedRefOf(sourceRefOf(iid).orNull, sourceDigest.orNull) + .foreach(ImagePrepullClient.ensurePrepull(iid, _)) + } + } } } @@ -427,6 +474,7 @@ class CuratedImageResource extends LazyLogging { def list(@Auth user: SessionUser): List[CuratedImage] = { requireEnabled() reconcileRunningValidations() + ensurePrepullsForReadyImages() context // Not select(): validation_log is unbounded and this endpoint discards it. .select(IID, NAME, SOURCE_REF, SOURCE_DIGEST, STATUS, ATTEMPT, CREATION_TIME, UPDATE_TIME) @@ -533,6 +581,9 @@ class CuratedImageResource extends LazyLogging { requireEnabled() // Nothing of ours holds a copy. A running unit keeps going on what its node pulled. ImageValidationClient.deleteAllValidations(iid) + // Left behind, it would go on holding the image on every node for an image no unit can + // be started from any more. + ImagePrepullClient.deletePrepull(iid) val deleted = context.deleteFrom(CU_IMAGE).where(IID.eq(iid)).execute() if (deleted == 0) { throw new NotFoundException(s"No curated image $iid.") diff --git a/computing-unit-managing-service/src/main/scala/org/apache/texera/service/util/ImagePrepullClient.scala b/computing-unit-managing-service/src/main/scala/org/apache/texera/service/util/ImagePrepullClient.scala new file mode 100644 index 00000000000..1bf3815746f --- /dev/null +++ b/computing-unit-managing-service/src/main/scala/org/apache/texera/service/util/ImagePrepullClient.scala @@ -0,0 +1,200 @@ +/* + * 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.texera.service.util + +import com.typesafe.scalalogging.LazyLogging +import io.fabric8.kubernetes.api.model.{Quantity, ResourceRequirementsBuilder} +import io.fabric8.kubernetes.api.model.apps.{DaemonSet, DaemonSetBuilder} +import io.fabric8.kubernetes.client.KubernetesClientBuilder +import org.apache.texera.common.config.CuratedImageConfig + +import scala.jdk.CollectionConverters._ + +/** + * Puts a ready curated image on every node before anyone starts a unit from it. + * + * Without this the first unit on a node waits for the whole image -- about 80 seconds for + * a 3 GB one -- while every later unit there starts at once, so the same action takes + * seconds or minutes depending only on where it landed. + * + * The mechanism is the one the chart already uses for the deployment's own image: a + * DaemonSet whose init container is the image and whose command does nothing, then a pause + * container to hold the pod open so the node does not reclaim what was just pulled. The + * chart cannot express these, because a curated image is registered while the cluster is + * running, so they are built here instead. + * + * Every call is best-effort. A pre-pull that cannot be created is logged and ignored: the + * image still works, and the first unit on each node just pays for the pull. + */ +object ImagePrepullClient extends LazyLogging { + + private val client: io.fabric8.kubernetes.client.KubernetesClient = + new KubernetesClientBuilder().build() + + private def namespace: String = CuratedImageConfig.prepullNamespace + + /** Marks every pre-pull this service owns, so they are found by label and not by name. */ + private[service] val OwnerLabel = "texera-cu-image-prepull" + + /** The image a pre-pull belongs to, so one image's can be removed without the others. */ + private[service] val ImageLabel = "texera-cu-image" + + /** + * Creates the pre-pull for an image, or points an existing one at a new reference. Called + * whenever a row reaches READY, which covers a refresh that resolved a moved tag to a + * different digest. + */ + def ensurePrepull(iid: Int, pinnedRef: String): Unit = { + if (!CuratedImageConfig.prepullEnabled) return + try { + val daemonSet = prepullDaemonSet(iid, pinnedRef) + client + .apps() + .daemonSets() + .inNamespace(namespace) + .resource(daemonSet) + .createOr(existing => existing.update()) + logger.info(s"Pre-pulling curated image $iid ($pinnedRef) onto every node.") + } catch { + case e: Throwable => + // The image is still usable; only the head start is lost. + logger.warn( + s"Could not pre-pull curated image $iid ($pinnedRef). The first unit on each " + + "node will wait for the pull instead.", + e + ) + } + } + + /** + * Removes an image's pre-pull. Safe to call when there is none -- a deployment that had + * pre-pulling turned off has nothing to delete, and neither has an image that never + * reached READY. + */ + def deletePrepull(iid: Int): Unit = { + try { + client + .apps() + .daemonSets() + .inNamespace(namespace) + .withName(CuratedImageConfig.prepullName(iid)) + .delete() + } catch { + case e: Throwable => + // Left behind it would go on holding the image on every node, so it is worth + // naming in the log rather than passing over. + logger.warn(s"Could not remove the pre-pull for curated image $iid.", e) + } + } + + /** + * The images that already have a pre-pull. Read so that rows which reached READY without + * one -- registered before this shipped, or created while the cluster was unreachable -- + * are given theirs, rather than never being pre-pulled at all. + */ + def prepulledImageIds(): Set[Int] = { + if (!CuratedImageConfig.prepullEnabled) return Set.empty + try { + client + .apps() + .daemonSets() + .inNamespace(namespace) + .withLabel(OwnerLabel, "true") + .list() + .getItems + .asScala + .flatMap(imageIdOf) + .toSet + } catch { + case e: Throwable => + // Returning "none are pre-pulled" would ask for every one of them again on a + // cluster that cannot answer. An empty answer here means "nothing to add". + logger.warn("Could not list the curated-image pre-pulls; leaving them as they are.", e) + Set.empty + } + } + + /** The image a pre-pull was created for, from its label rather than its name. */ + private[service] def imageIdOf(daemonSet: DaemonSet): Option[Int] = + Option(daemonSet.getMetadata) + .flatMap(m => Option(m.getLabels)) + .flatMap(labels => Option(labels.get(ImageLabel))) + .flatMap(value => scala.util.Try(value.toInt).toOption) + + private[service] def prepullDaemonSet(iid: Int, pinnedRef: String): DaemonSet = { + val name = CuratedImageConfig.prepullName(iid) + val labels = Map( + "app" -> name, + OwnerLabel -> "true", + ImageLabel -> iid.toString + ).asJava + + // The pause container is the whole running cost of a pre-pull, and it does nothing but + // exist, so it is held to the smallest request the chart's own pre-puller uses. + val pauseResources = new ResourceRequirementsBuilder() + .addToRequests("cpu", new Quantity(CuratedImageConfig.prepullCpu)) + .addToRequests("memory", new Quantity(CuratedImageConfig.prepullMemory)) + .addToLimits("cpu", new Quantity(CuratedImageConfig.prepullCpu)) + .addToLimits("memory", new Quantity(CuratedImageConfig.prepullMemory)) + .build() + + new DaemonSetBuilder() + .withNewMetadata() + .withName(name) + .withNamespace(namespace) + .withLabels(labels) + .endMetadata() + .withNewSpec() + .withNewSelector() + // Only "app". A DaemonSet's selector cannot be changed after it is created, so it + // must not carry anything this code might later want to alter. + .withMatchLabels(Map("app" -> name).asJava) + .endSelector() + .withNewTemplate() + .withNewMetadata() + .withLabels(labels) + .endMetadata() + .withNewSpec() + // A node the deployment tolerates is a node a unit can land on, so it is one the + // image has to reach. + .addNewToleration() + .withOperator("Exists") + .endToleration() + .withInitContainers( + new io.fabric8.kubernetes.api.model.ContainerBuilder() + .withName("prepuller") + .withImage(pinnedRef) + // The reference names a digest, so what is already on the node cannot differ + // from what the registry would serve. Always would re-check it for nothing. + .withImagePullPolicy("IfNotPresent") + .withCommand("sh", "-c", "true") + .build() + ) + .addNewContainer() + .withName("pause") + .withImage(CuratedImageConfig.prepullPauseImage) + .withResources(pauseResources) + .endContainer() + .endSpec() + .endTemplate() + .endSpec() + .build() + } +} diff --git a/computing-unit-managing-service/src/test/scala/org/apache/texera/service/util/ImagePrepullClientSpec.scala b/computing-unit-managing-service/src/test/scala/org/apache/texera/service/util/ImagePrepullClientSpec.scala new file mode 100644 index 00000000000..794a9d72d6c --- /dev/null +++ b/computing-unit-managing-service/src/test/scala/org/apache/texera/service/util/ImagePrepullClientSpec.scala @@ -0,0 +1,110 @@ +/* + * 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.texera.service.util + +import io.fabric8.kubernetes.api.model.apps.DaemonSetBuilder +import org.apache.texera.common.config.CuratedImageConfig +import org.scalatest.OptionValues._ +import org.scalatest.flatspec.AnyFlatSpec +import org.scalatest.matchers.should.Matchers + +import scala.jdk.CollectionConverters._ + +class ImagePrepullClientSpec extends AnyFlatSpec with Matchers { + + import ImagePrepullClient.{ImageLabel, OwnerLabel, imageIdOf, prepullDaemonSet} + + private val PinnedRef = "tagandhi19/texera-cu-sklearn@sha256:" + "b" * 64 + + "prepullDaemonSet" should "pull the pinned reference and nothing else" in { + val spec = prepullDaemonSet(7, PinnedRef).getSpec.getTemplate.getSpec + val initContainers = spec.getInitContainers.asScala.toList + initContainers should have size 1 + val init = initContainers.head + + // The whole point: the image is named as the init container, so scheduling the pod is + // what pulls it. The command is a no-op -- nothing in the image is run. + init.getImage shouldBe PinnedRef + init.getCommand.asScala.toList shouldBe List("sh", "-c", "true") + + // A digest cannot resolve to different bytes later, so re-checking the registry every + // time the pod restarts would buy nothing. + init.getImagePullPolicy shouldBe "IfNotPresent" + + // Only the pause container keeps running. If the curated image were left running here + // it would be a computing unit nobody asked for, on every node. + val containers = spec.getContainers.asScala.toList + containers.map(_.getName) shouldBe List("pause") + containers.head.getImage shouldBe CuratedImageConfig.prepullPauseImage + } + + it should "reach every node the deployment tolerates" in { + // A node a unit can be scheduled onto is a node the image has to be on. Without this + // the tainted nodes are exactly the ones that would wait for the pull. + val tolerations = prepullDaemonSet(7, PinnedRef).getSpec.getTemplate.getSpec.getTolerations + tolerations.asScala.map(_.getOperator).toList shouldBe List("Exists") + } + + it should "name itself after the image, so a refresh replaces rather than adds" in { + // Same name for the same image at any digest: a refreshed image must not leave a + // second pre-pull behind holding bytes nothing runs any more. + prepullDaemonSet(7, PinnedRef).getMetadata.getName shouldBe "cu-image-prepull-7" + prepullDaemonSet(7, "owner/name@sha256:" + "c" * 64).getMetadata.getName shouldBe + "cu-image-prepull-7" + prepullDaemonSet(8, PinnedRef).getMetadata.getName shouldBe "cu-image-prepull-8" + } + + it should "select on a label it will never want to change" in { + // A DaemonSet's selector is immutable once created. Selecting on the image id too + // would be harmless, but selecting on anything mutable would make the update in + // ensurePrepull fail for good, so this pins the selector to "app" alone. + val daemonSet = prepullDaemonSet(7, PinnedRef) + daemonSet.getSpec.getSelector.getMatchLabels.asScala shouldBe + Map("app" -> "cu-image-prepull-7") + + // The pod template must still match it, or the DaemonSet is rejected outright. + val templateLabels = daemonSet.getSpec.getTemplate.getMetadata.getLabels.asScala + templateLabels("app") shouldBe "cu-image-prepull-7" + } + + it should "label the image it belongs to, so one can be removed without the others" in { + val labels = prepullDaemonSet(7, PinnedRef).getMetadata.getLabels.asScala + labels(OwnerLabel) shouldBe "true" + labels(ImageLabel) shouldBe "7" + } + + "imageIdOf" should "read the image back from the label rather than the name" in { + imageIdOf(prepullDaemonSet(7, PinnedRef)).value shouldBe 7 + } + + it should "ignore a DaemonSet that is not one of ours" in { + // prepulledImageIds lists by label, but a cluster can hold anything. A stray object + // must not be read as an image id and leave a real image without its pre-pull. + val unlabelled = new DaemonSetBuilder().withNewMetadata().withName("something").endMetadata() + imageIdOf(unlabelled.build()) shouldBe None + + val notANumber = new DaemonSetBuilder() + .withNewMetadata() + .withName("something") + .addToLabels(ImageLabel, "not-a-number") + .endMetadata() + imageIdOf(notANumber.build()) shouldBe None + } +} From a23a6734339e54866c801ffc1f7438304153afe4 Mon Sep 17 00:00:00 2001 From: Tanishq Gandhi Date: Thu, 10 Sep 2026 21:57:36 +0000 Subject: [PATCH 2/7] fix(computing-unit): make the pre-pull actually reach the nodes Review found the pre-pull never ran. The pool namespace has a ResourceQuota on requests.cpu and requests.memory, quota admission checks init containers too, and the prepuller declared neither -- so every pod was refused while the DaemonSet was still created, leaving the service logging success and pulling nothing. The chart's own two pre-pullers escape this only by living in the release namespace. Pre-pulls move there with them, with a Role of their own, and both containers now state requests. A pod's request is the larger of its init containers and the sum of the rest, so this stays 1m/8Mi. Also from review: - an image leaving READY now loses its pre-pull, instead of holding gigabytes on every node for something no unit can start from - the reconcile pass compares the reference, not just the id, so a repoint that failed once is retried rather than left pointing at a superseded digest - a listing that failed is no longer read as "nothing is pre-pulled", which had every read fire a doomed create per ready image - delete removes the row before the pre-pull, closing a window where a concurrent read re-created one nothing would reap - turning pre-pulling off removes the pre-pulls already made, which is what frees the disk the setting talks about - no tolerations: a computing-unit pod declares none, so tolerating everything only put images on nodes no unit can be scheduled onto --- ...low-computing-unit-manager-deployment.yaml | 6 +- ...omputing-unit-manager-service-account.yaml | 37 ++++++++-- bin/k8s/values.yaml | 3 +- .../config/src/main/resources/kubernetes.conf | 13 +++- .../resource/CuratedImageResource.scala | 55 ++++++++++----- .../service/util/ImagePrepullClient.scala | 69 +++++++++++++------ .../service/util/ImagePrepullClientSpec.scala | 39 +++++++++-- 7 files changed, 168 insertions(+), 54 deletions(-) diff --git a/bin/k8s/templates/base/workflow-computing-unit-manager/workflow-computing-unit-manager-deployment.yaml b/bin/k8s/templates/base/workflow-computing-unit-manager/workflow-computing-unit-manager-deployment.yaml index ca13b4af31b..60973ef8183 100644 --- a/bin/k8s/templates/base/workflow-computing-unit-manager/workflow-computing-unit-manager-deployment.yaml +++ b/bin/k8s/templates/base/workflow-computing-unit-manager/workflow-computing-unit-manager-deployment.yaml @@ -73,9 +73,11 @@ spec: value: {{ .Values.workflowComputingUnitPool.namespace }} - name: TEXERA_CURATED_IMAGE_PREPULL_ENABLED value: "{{ .Values.curatedImages.prepull.enabled }}" - # The same namespace, from the same value, so the two cannot drift apart. + # The release namespace, not the pool one: the pool's ResourceQuota refuses a + # pod whose init container declares no requests, and pre-pulls would otherwise + # eat the budget computing units draw from. - name: TEXERA_CURATED_IMAGE_PREPULL_NAMESPACE - value: {{ .Values.workflowComputingUnitPool.namespace }} + value: {{ .Release.Namespace }} - name: KUBERNETES_IMAGE_NAME value: {{ .Values.texera.imageRegistry }}/{{ .Values.workflowComputingUnitPool.imageName }}:{{ .Values.texera.imageTag }} - name: KUBERNETES_MOUNTER_ENABLED diff --git a/bin/k8s/templates/base/workflow-computing-unit-manager/workflow-computing-unit-manager-service-account.yaml b/bin/k8s/templates/base/workflow-computing-unit-manager/workflow-computing-unit-manager-service-account.yaml index 6e3d61554d9..6a98999e0eb 100644 --- a/bin/k8s/templates/base/workflow-computing-unit-manager/workflow-computing-unit-manager-service-account.yaml +++ b/bin/k8s/templates/base/workflow-computing-unit-manager/workflow-computing-unit-manager-service-account.yaml @@ -42,11 +42,6 @@ rules: - apiGroups: [""] resources: ["pods/log"] verbs: ["get"] - # One DaemonSet per ready curated image, pulling it onto every node. Created while the - # cluster runs, as images are registered, so the chart cannot declare them. - - apiGroups: ["apps"] - resources: ["daemonsets"] - verbs: ["get", "list", "watch", "create", "update", "patch", "delete"] --- apiVersion: rbac.authorization.k8s.io/v1 @@ -61,4 +56,34 @@ subjects: roleRef: kind: Role name: {{ .Values.workflowComputingUnitManager.name }} - apiGroup: rbac.authorization.k8s.io \ No newline at end of file + apiGroup: rbac.authorization.k8s.io + +--- +# The pre-pull DaemonSets live in the release namespace, alongside the chart's own two +# pre-pullers, so this is a second Role rather than another rule on the pool one. The pool +# namespace carries a ResourceQuota on requests.cpu/requests.memory, and quota admission +# checks init containers, so a pre-pull pod there is refused before it is ever scheduled. +apiVersion: rbac.authorization.k8s.io/v1 +kind: Role +metadata: + name: {{ .Values.workflowComputingUnitManager.name }}-prepull + namespace: {{ .Release.Namespace }} +rules: + - apiGroups: ["apps"] + resources: ["daemonsets"] + verbs: ["get", "list", "watch", "create", "update", "patch", "delete"] + +--- +apiVersion: rbac.authorization.k8s.io/v1 +kind: RoleBinding +metadata: + name: {{ .Values.workflowComputingUnitManager.name }}-prepull-binding + namespace: {{ .Release.Namespace }} +subjects: + - kind: ServiceAccount + name: {{ .Values.workflowComputingUnitManager.serviceAccountName }} + namespace: {{ .Release.Namespace }} +roleRef: + kind: Role + name: {{ .Values.workflowComputingUnitManager.name }}-prepull + apiGroup: rbac.authorization.k8s.io diff --git a/bin/k8s/values.yaml b/bin/k8s/values.yaml index 3d11ab496e1..7bc7486e666 100644 --- a/bin/k8s/values.yaml +++ b/bin/k8s/values.yaml @@ -379,7 +379,8 @@ curatedImages: prepull: # Pull every ready image onto every node, so the first unit on a node does not wait for # it. Costs node disk: each node holds each ready image. Turn off to pay the pull on - # first use instead. + # first use instead -- doing so also removes the pre-pulls already made, which is what + # frees the disk, on the next read of the image list. enabled: true # headless service for the access of computing units diff --git a/common/config/src/main/resources/kubernetes.conf b/common/config/src/main/resources/kubernetes.conf index 1b6c4b30658..ee6045fae33 100644 --- a/common/config/src/main/resources/kubernetes.conf +++ b/common/config/src/main/resources/kubernetes.conf @@ -155,12 +155,19 @@ curated-images { # Pull each ready image onto every node as soon as it is ready, rather than when someone # first starts a unit from it. Costs node disk -- every node holds every ready image -- # so a deployment short of it can turn this off and pay the pull on first use instead. + # Turning it off also removes the pre-pulls already made, which is what frees the disk; + # that sweep runs on the next read of the image list, so it needs curated images + # themselves left enabled. prepull-enabled = true prepull-enabled = ${?TEXERA_CURATED_IMAGE_PREPULL_ENABLED} - # Where the pre-pull DaemonSets run. The pool namespace, like validation: a node caches - # an image whatever namespace asked for it, so this only decides where the pods are seen. - prepull-namespace = "texera-workflow-computing-unit-pool" + # The release namespace, where the chart's own two pre-pullers already live -- not the + # pool namespace. A node caches an image whatever namespace asked for it, so this changes + # nothing about what gets pulled, but the pool namespace carries a ResourceQuota on + # requests.cpu and requests.memory. Quota admission checks init containers too, so a + # pre-pull pod there is refused outright, and pre-pulls would also eat the budget users' + # computing units draw from. The chart sets this to .Release.Namespace. + prepull-namespace = "texera" prepull-namespace = ${?TEXERA_CURATED_IMAGE_PREPULL_NAMESPACE} # Holds the pod open once the init container has pulled the image, so the node does not diff --git a/computing-unit-managing-service/src/main/scala/org/apache/texera/service/resource/CuratedImageResource.scala b/computing-unit-managing-service/src/main/scala/org/apache/texera/service/resource/CuratedImageResource.scala index 55dcb9dc406..81043fe565a 100644 --- a/computing-unit-managing-service/src/main/scala/org/apache/texera/service/resource/CuratedImageResource.scala +++ b/computing-unit-managing-service/src/main/scala/org/apache/texera/service/resource/CuratedImageResource.scala @@ -211,22 +211,34 @@ object CuratedImageResource extends LazyLogging { * costs a single call. */ private def ensurePrepullsForReadyImages(): Unit = { - if (!CuratedImageConfig.prepullEnabled) return + // Turning pre-pulling off is what frees the node disk it costs, so the pre-pulls + // already made are removed rather than merely left un-added to. + if (!CuratedImageConfig.prepullEnabled) { + ImagePrepullClient.deleteAllPrepulls() + return + } try { - val alreadyPrepulled = ImagePrepullClient.prepulledImageIds() - context - .select(IID, SOURCE_REF, SOURCE_DIGEST) - .from(CU_IMAGE) - .where(STATUS.eq(Status.Ready)) - .fetch() - .asScala - .foreach { row => - val iid = row.get(IID).intValue() - if (!alreadyPrepulled.contains(iid)) { - pinnedRefOf(row.get(SOURCE_REF), row.get(SOURCE_DIGEST)) - .foreach(ImagePrepullClient.ensurePrepull(iid, _)) + // None means the cluster could not be asked. Treating that as "nothing is pre-pulled" + // would fire a doomed create for every ready image on every read. + ImagePrepullClient.prepulledRefs().foreach { prepulled => + context + .select(IID, SOURCE_REF, SOURCE_DIGEST) + .from(CU_IMAGE) + .where(STATUS.eq(Status.Ready)) + .fetch() + .asScala + .foreach { row => + val iid = row.get(IID).intValue() + pinnedRefOf(row.get(SOURCE_REF), row.get(SOURCE_DIGEST)).foreach { wanted => + // Compared by reference, not merely by presence: a refresh whose repoint + // failed leaves a pre-pull holding the previous digest, and nothing else + // would ever correct it. + if (!prepulled.get(iid).contains(wanted)) { + ImagePrepullClient.ensurePrepull(iid, wanted) + } + } } - } + } } catch { // Opportunistic, like reconciling: a listing that cannot be repaired is still worth // returning. @@ -447,6 +459,12 @@ object CuratedImageResource extends LazyLogging { if (status == Status.Ready) { pinnedRefOf(sourceRefOf(iid).orNull, sourceDigest.orNull) .foreach(ImagePrepullClient.ensurePrepull(iid, _)) + } else { + // A failed refresh keeps the previous digest, so the row still names an image -- + // but no unit can be started from it any more. Left alone, its pre-pull would go + // on holding gigabytes on every node for something unusable, and nothing revisits + // a row that is not READY. + ImagePrepullClient.deletePrepull(iid) } } } @@ -581,13 +599,16 @@ class CuratedImageResource extends LazyLogging { requireEnabled() // Nothing of ours holds a copy. A running unit keeps going on what its node pulled. ImageValidationClient.deleteAllValidations(iid) - // Left behind, it would go on holding the image on every node for an image no unit can - // be started from any more. - ImagePrepullClient.deletePrepull(iid) + // The row goes first. Removing the pre-pull before it leaves a window in which a + // concurrent read sees a row that is still READY and re-creates the pre-pull, and + // nothing would ever reap that one. val deleted = context.deleteFrom(CU_IMAGE).where(IID.eq(iid)).execute() if (deleted == 0) { throw new NotFoundException(s"No curated image $iid.") } + // Left behind, it would go on holding the image on every node for an image no unit can + // be started from any more. + ImagePrepullClient.deletePrepull(iid) } /** Marks the row as being validated and submits the job, in that order. */ diff --git a/computing-unit-managing-service/src/main/scala/org/apache/texera/service/util/ImagePrepullClient.scala b/computing-unit-managing-service/src/main/scala/org/apache/texera/service/util/ImagePrepullClient.scala index 1bf3815746f..d07219d6175 100644 --- a/computing-unit-managing-service/src/main/scala/org/apache/texera/service/util/ImagePrepullClient.scala +++ b/computing-unit-managing-service/src/main/scala/org/apache/texera/service/util/ImagePrepullClient.scala @@ -105,14 +105,18 @@ object ImagePrepullClient extends LazyLogging { } /** - * The images that already have a pre-pull. Read so that rows which reached READY without - * one -- registered before this shipped, or created while the cluster was unreachable -- - * are given theirs, rather than never being pre-pulled at all. + * What each image's pre-pull currently pulls, keyed by image. Read so that a row which + * reached READY without one -- registered before this shipped, or finished while the + * cluster was unreachable -- is given theirs, and so that one still pointing at a + * superseded digest is corrected. + * + * None means the question could not be answered, which is not the same as "none exist": + * an empty map would have the caller create a pre-pull for every ready image against a + * cluster that has just refused to talk to it. */ - def prepulledImageIds(): Set[Int] = { - if (!CuratedImageConfig.prepullEnabled) return Set.empty + def prepulledRefs(): Option[Map[Int, String]] = { try { - client + val entries = client .apps() .daemonSets() .inNamespace(namespace) @@ -120,17 +124,39 @@ object ImagePrepullClient extends LazyLogging { .list() .getItems .asScala - .flatMap(imageIdOf) - .toSet + .flatMap(daemonSet => imageIdOf(daemonSet).map(_ -> prepulledRefOf(daemonSet).orNull)) + Some(entries.toMap) } catch { case e: Throwable => - // Returning "none are pre-pulled" would ask for every one of them again on a - // cluster that cannot answer. An empty answer here means "nothing to add". logger.warn("Could not list the curated-image pre-pulls; leaving them as they are.", e) - Set.empty + None + } + } + + /** Every pre-pull this service owns, removed. How turning pre-pulling off frees the disk. */ + def deleteAllPrepulls(): Unit = { + try { + client + .apps() + .daemonSets() + .inNamespace(namespace) + .withLabel(OwnerLabel, "true") + .delete() + } catch { + case e: Throwable => + logger.warn("Could not remove the curated-image pre-pulls.", e) } } + /** The reference a pre-pull pulls, which is its init container's image. */ + private[service] def prepulledRefOf(daemonSet: DaemonSet): Option[String] = + Option(daemonSet.getSpec) + .flatMap(spec => Option(spec.getTemplate)) + .flatMap(template => Option(template.getSpec)) + .flatMap(podSpec => Option(podSpec.getInitContainers)) + .flatMap(_.asScala.headOption) + .flatMap(container => Option(container.getImage)) + /** The image a pre-pull was created for, from its label rather than its name. */ private[service] def imageIdOf(daemonSet: DaemonSet): Option[Int] = Option(daemonSet.getMetadata) @@ -146,9 +172,12 @@ object ImagePrepullClient extends LazyLogging { ImageLabel -> iid.toString ).asJava - // The pause container is the whole running cost of a pre-pull, and it does nothing but - // exist, so it is held to the smallest request the chart's own pre-puller uses. - val pauseResources = new ResourceRequirementsBuilder() + // Stated on both containers, not just the one that keeps running. A namespace with a + // ResourceQuota on requests.cpu/requests.memory refuses a pod whose init container + // leaves them out -- quota admission checks init containers too -- and the refusal is + // invisible, because the DaemonSet is still created. Costs nothing: a pod's request is + // the larger of its init containers and the sum of its others, so the same 1m/8Mi. + val resources = new ResourceRequirementsBuilder() .addToRequests("cpu", new Quantity(CuratedImageConfig.prepullCpu)) .addToRequests("memory", new Quantity(CuratedImageConfig.prepullMemory)) .addToLimits("cpu", new Quantity(CuratedImageConfig.prepullCpu)) @@ -172,11 +201,10 @@ object ImagePrepullClient extends LazyLogging { .withLabels(labels) .endMetadata() .withNewSpec() - // A node the deployment tolerates is a node a unit can land on, so it is one the - // image has to reach. - .addNewToleration() - .withOperator("Exists") - .endToleration() + // No tolerations, deliberately: a computing-unit pod declares none either, so a + // tainted node is one no unit can ever be scheduled onto. Tolerating everything + // would put multi-gigabyte images on control-plane and other reserved nodes that + // will never run a unit. .withInitContainers( new io.fabric8.kubernetes.api.model.ContainerBuilder() .withName("prepuller") @@ -185,12 +213,13 @@ object ImagePrepullClient extends LazyLogging { // from what the registry would serve. Always would re-check it for nothing. .withImagePullPolicy("IfNotPresent") .withCommand("sh", "-c", "true") + .withResources(resources) .build() ) .addNewContainer() .withName("pause") .withImage(CuratedImageConfig.prepullPauseImage) - .withResources(pauseResources) + .withResources(resources) .endContainer() .endSpec() .endTemplate() diff --git a/computing-unit-managing-service/src/test/scala/org/apache/texera/service/util/ImagePrepullClientSpec.scala b/computing-unit-managing-service/src/test/scala/org/apache/texera/service/util/ImagePrepullClientSpec.scala index 794a9d72d6c..cc3b5b53762 100644 --- a/computing-unit-managing-service/src/test/scala/org/apache/texera/service/util/ImagePrepullClientSpec.scala +++ b/computing-unit-managing-service/src/test/scala/org/apache/texera/service/util/ImagePrepullClientSpec.scala @@ -55,11 +55,40 @@ class ImagePrepullClientSpec extends AnyFlatSpec with Matchers { containers.head.getImage shouldBe CuratedImageConfig.prepullPauseImage } - it should "reach every node the deployment tolerates" in { - // A node a unit can be scheduled onto is a node the image has to be on. Without this - // the tainted nodes are exactly the ones that would wait for the pull. - val tolerations = prepullDaemonSet(7, PinnedRef).getSpec.getTemplate.getSpec.getTolerations - tolerations.asScala.map(_.getOperator).toList shouldBe List("Exists") + it should "schedule exactly where a computing unit can, and no wider" in { + // The regression this guards: an earlier version tolerated everything, which put + // multi-gigabyte images on control-plane and other reserved nodes. A computing-unit + // pod declares no tolerations, so a tainted node is one no unit can ever land on -- + // pre-pulling there buys nothing and costs disk on the nodes that can least spare it. + val podSpec = prepullDaemonSet(7, PinnedRef).getSpec.getTemplate.getSpec + Option(podSpec.getTolerations).map(_.asScala.toList).getOrElse(Nil) shouldBe Nil + } + + // The regression this guards: the init container declared no requests, and the pool + // namespace has a ResourceQuota on requests.cpu/requests.memory. Quota admission checks + // init containers, so every pre-pull pod was refused -- while the DaemonSet itself was + // created, so the service logged success and pre-pulled nothing, anywhere, ever. + it should "declare the requests a quota would demand" in { + val podSpec = prepullDaemonSet(7, PinnedRef).getSpec.getTemplate.getSpec + val everyContainer = + podSpec.getInitContainers.asScala.toList ++ podSpec.getContainers.asScala.toList + everyContainer.foreach { container => + val requests = Option(container.getResources).map(_.getRequests.asScala).getOrElse(Map.empty) + withClue(s"${container.getName} must request cpu and memory: ") { + requests.keySet should contain allOf ("cpu", "memory") + } + } + } + + "prepulledRefOf" should "read back what a pre-pull actually pulls" in { + // What lets a stale pre-pull be spotted: a refresh whose repoint failed leaves one + // holding the previous digest, and comparing only ids would never notice. + ImagePrepullClient.prepulledRefOf(prepullDaemonSet(7, PinnedRef)).value shouldBe PinnedRef + } + + it should "be empty for a DaemonSet with no init container" in { + val strayObject = new DaemonSetBuilder().withNewMetadata().withName("x").endMetadata().build() + ImagePrepullClient.prepulledRefOf(strayObject) shouldBe None } it should "name itself after the image, so a refresh replaces rather than adds" in { From ad195884522e41e5bd78a79ae2025d75bb9a3aa8 Mon Sep 17 00:00:00 2001 From: Tanishq Gandhi Date: Thu, 10 Sep 2026 22:11:05 +0000 Subject: [PATCH 3/7] fix(computing-unit): keep the pre-pull out of the privileged namespace The previous commit fixed the quota rejection twice over -- it declared requests on both containers and moved the pre-pulls to the release namespace. The requests are what fixed it. The move was the half that did damage: the release namespace holds the privileged hostPath mounter, so permission to write DaemonSets there is permission to run as root on every node. Pre-pulls go back to the pool namespace, whose quota the requests satisfy and which holds nothing privileged, and the grant goes back to being one rule on the Role that was already there. Also from review: - limits are on the pause container only. A limit is enforced per container and never maxed across them, so sharing one object put an 8Mi cap on the image's own shell -- bash, on the Python bases these are built from. It would OOMKill, crash-loop, never reach pause, and leave the pulled image reclaimable, while the DaemonSet reported itself created. - the reconcile pass now removes pre-pulls no ready image wants, rather than only adding. Every other removal path can be interrupted between the database write and the cluster call, and nothing revisited a row that was gone or not ready, so an orphan was unreapable. This also ends the collection delete that ran on every read with pre-pulling off: it is driven by what the listing found, so it stops once there is nothing left. --- ...low-computing-unit-manager-deployment.yaml | 7 +- ...omputing-unit-manager-service-account.yaml | 39 ++------ .../config/src/main/resources/kubernetes.conf | 12 ++- .../resource/CuratedImageResource.scala | 90 ++++++++++--------- .../service/util/ImagePrepullClient.scala | 37 ++++---- .../service/util/ImagePrepullClientSpec.scala | 15 ++++ 6 files changed, 97 insertions(+), 103 deletions(-) diff --git a/bin/k8s/templates/base/workflow-computing-unit-manager/workflow-computing-unit-manager-deployment.yaml b/bin/k8s/templates/base/workflow-computing-unit-manager/workflow-computing-unit-manager-deployment.yaml index 60973ef8183..414f439f02d 100644 --- a/bin/k8s/templates/base/workflow-computing-unit-manager/workflow-computing-unit-manager-deployment.yaml +++ b/bin/k8s/templates/base/workflow-computing-unit-manager/workflow-computing-unit-manager-deployment.yaml @@ -73,11 +73,10 @@ spec: value: {{ .Values.workflowComputingUnitPool.namespace }} - name: TEXERA_CURATED_IMAGE_PREPULL_ENABLED value: "{{ .Values.curatedImages.prepull.enabled }}" - # The release namespace, not the pool one: the pool's ResourceQuota refuses a - # pod whose init container declares no requests, and pre-pulls would otherwise - # eat the budget computing units draw from. + # The pool namespace, whose quota the declared requests satisfy. Not the + # release namespace, which holds the privileged mounter. - name: TEXERA_CURATED_IMAGE_PREPULL_NAMESPACE - value: {{ .Release.Namespace }} + value: {{ .Values.workflowComputingUnitPool.namespace }} - name: KUBERNETES_IMAGE_NAME value: {{ .Values.texera.imageRegistry }}/{{ .Values.workflowComputingUnitPool.imageName }}:{{ .Values.texera.imageTag }} - name: KUBERNETES_MOUNTER_ENABLED diff --git a/bin/k8s/templates/base/workflow-computing-unit-manager/workflow-computing-unit-manager-service-account.yaml b/bin/k8s/templates/base/workflow-computing-unit-manager/workflow-computing-unit-manager-service-account.yaml index 6a98999e0eb..231d79e3328 100644 --- a/bin/k8s/templates/base/workflow-computing-unit-manager/workflow-computing-unit-manager-service-account.yaml +++ b/bin/k8s/templates/base/workflow-computing-unit-manager/workflow-computing-unit-manager-service-account.yaml @@ -42,33 +42,12 @@ rules: - apiGroups: [""] resources: ["pods/log"] verbs: ["get"] - ---- -apiVersion: rbac.authorization.k8s.io/v1 -kind: RoleBinding -metadata: - name: {{ .Values.workflowComputingUnitManager.name }}-binding - namespace: {{ .Values.workflowComputingUnitPool.namespace }} -subjects: - - kind: ServiceAccount - name: {{ .Values.workflowComputingUnitManager.serviceAccountName }} - namespace: {{ .Release.Namespace }} -roleRef: - kind: Role - name: {{ .Values.workflowComputingUnitManager.name }} - apiGroup: rbac.authorization.k8s.io - ---- -# The pre-pull DaemonSets live in the release namespace, alongside the chart's own two -# pre-pullers, so this is a second Role rather than another rule on the pool one. The pool -# namespace carries a ResourceQuota on requests.cpu/requests.memory, and quota admission -# checks init containers, so a pre-pull pod there is refused before it is ever scheduled. -apiVersion: rbac.authorization.k8s.io/v1 -kind: Role -metadata: - name: {{ .Values.workflowComputingUnitManager.name }}-prepull - namespace: {{ .Release.Namespace }} -rules: + # One DaemonSet per ready curated image, pulling it onto every node. Created while the + # cluster runs, as images are registered, so the chart cannot declare them. + # + # Scoped to the pool namespace deliberately. The release namespace holds the privileged + # hostPath mounter, and permission to rewrite a DaemonSet there is permission to run as + # root on every node. Nothing in the pool namespace is privileged. - apiGroups: ["apps"] resources: ["daemonsets"] verbs: ["get", "list", "watch", "create", "update", "patch", "delete"] @@ -77,13 +56,13 @@ rules: apiVersion: rbac.authorization.k8s.io/v1 kind: RoleBinding metadata: - name: {{ .Values.workflowComputingUnitManager.name }}-prepull-binding - namespace: {{ .Release.Namespace }} + name: {{ .Values.workflowComputingUnitManager.name }}-binding + namespace: {{ .Values.workflowComputingUnitPool.namespace }} subjects: - kind: ServiceAccount name: {{ .Values.workflowComputingUnitManager.serviceAccountName }} namespace: {{ .Release.Namespace }} roleRef: kind: Role - name: {{ .Values.workflowComputingUnitManager.name }}-prepull + name: {{ .Values.workflowComputingUnitManager.name }} apiGroup: rbac.authorization.k8s.io diff --git a/common/config/src/main/resources/kubernetes.conf b/common/config/src/main/resources/kubernetes.conf index ee6045fae33..dd861134eb7 100644 --- a/common/config/src/main/resources/kubernetes.conf +++ b/common/config/src/main/resources/kubernetes.conf @@ -161,13 +161,11 @@ curated-images { prepull-enabled = true prepull-enabled = ${?TEXERA_CURATED_IMAGE_PREPULL_ENABLED} - # The release namespace, where the chart's own two pre-pullers already live -- not the - # pool namespace. A node caches an image whatever namespace asked for it, so this changes - # nothing about what gets pulled, but the pool namespace carries a ResourceQuota on - # requests.cpu and requests.memory. Quota admission checks init containers too, so a - # pre-pull pod there is refused outright, and pre-pulls would also eat the budget users' - # computing units draw from. The chart sets this to .Release.Namespace. - prepull-namespace = "texera" + # The pool namespace, not the release namespace. Its ResourceQuota is satisfied by the + # requests both containers declare, so nothing is gained by moving out -- and the release + # namespace holds the privileged hostPath mounter, where the permission to write + # DaemonSets would be the permission to run as root on every node. + prepull-namespace = "texera-workflow-computing-unit-pool" prepull-namespace = ${?TEXERA_CURATED_IMAGE_PREPULL_NAMESPACE} # Holds the pod open once the init container has pulled the image, so the node does not diff --git a/computing-unit-managing-service/src/main/scala/org/apache/texera/service/resource/CuratedImageResource.scala b/computing-unit-managing-service/src/main/scala/org/apache/texera/service/resource/CuratedImageResource.scala index 81043fe565a..faf132c5b7d 100644 --- a/computing-unit-managing-service/src/main/scala/org/apache/texera/service/resource/CuratedImageResource.scala +++ b/computing-unit-managing-service/src/main/scala/org/apache/texera/service/resource/CuratedImageResource.scala @@ -202,48 +202,56 @@ object CuratedImageResource extends LazyLogging { .map(_.get(SOURCE_REF)) /** - * Gives a pre-pull to every ready image that has none. A row can reach READY without one: - * registered before pre-pulling shipped, or finished while the cluster could not be - * reached. Without this the image would never be pre-pulled at all, since nothing else - * revisits a row once it is ready. + * Brings the pre-pulls in the cluster into line with the ready images in the database: + * adds the missing, repoints the stale, and removes the rest. * - * One labelled list, then only the images actually missing, so a settled deployment - * costs a single call. + * Reconciling rather than only adding is what makes the pre-pulls reapable at all. + * Every other path that removes one -- a deleted row, an image that left READY -- can + * be interrupted between the database write and the cluster call, and nothing else + * revisits a row that is gone or not ready. Whatever is left over is found here. + * + * One labelled list and one query, so a settled deployment does no work. */ - private def ensurePrepullsForReadyImages(): Unit = { - // Turning pre-pulling off is what frees the node disk it costs, so the pre-pulls - // already made are removed rather than merely left un-added to. - if (!CuratedImageConfig.prepullEnabled) { - ImagePrepullClient.deleteAllPrepulls() - return - } + private def reconcilePrepulls(): Unit = { try { - // None means the cluster could not be asked. Treating that as "nothing is pre-pulled" - // would fire a doomed create for every ready image on every read. + // None means the cluster could not be asked, which is not "nothing is pre-pulled": + // acting on that would fire a doomed call for every image on every read. ImagePrepullClient.prepulledRefs().foreach { prepulled => - context - .select(IID, SOURCE_REF, SOURCE_DIGEST) - .from(CU_IMAGE) - .where(STATUS.eq(Status.Ready)) - .fetch() - .asScala - .foreach { row => - val iid = row.get(IID).intValue() - pinnedRefOf(row.get(SOURCE_REF), row.get(SOURCE_DIGEST)).foreach { wanted => - // Compared by reference, not merely by presence: a refresh whose repoint - // failed leaves a pre-pull holding the previous digest, and nothing else - // would ever correct it. - if (!prepulled.get(iid).contains(wanted)) { - ImagePrepullClient.ensurePrepull(iid, wanted) + // Turning pre-pulling off is what frees the node disk it costs, so the pre-pulls + // already made are removed rather than merely left un-added to. Driven by what the + // listing actually found, so a deployment that has been off stops calling once + // there is nothing left rather than issuing a delete on every page load. + val wantedByImage: Map[Int, String] = + if (!CuratedImageConfig.prepullEnabled) Map.empty + else + context + .select(IID, SOURCE_REF, SOURCE_DIGEST) + .from(CU_IMAGE) + .where(STATUS.eq(Status.Ready)) + .fetch() + .asScala + .flatMap { row => + pinnedRefOf(row.get(SOURCE_REF), row.get(SOURCE_DIGEST)) + .map(row.get(IID).intValue() -> _) } - } - } + .toMap + + // Compared by reference, not merely by presence: a refresh whose repoint failed + // leaves a pre-pull holding the previous digest, and nothing else would correct it. + wantedByImage.foreach { + case (iid, wanted) => + if (!prepulled.get(iid).contains(wanted)) ImagePrepullClient.ensurePrepull(iid, wanted) + } + + // Orphans: a row deleted while the cluster was unreachable, an image that left + // READY, or a process that stopped between the two. No row refers to these. + (prepulled.keySet -- wantedByImage.keySet).foreach(ImagePrepullClient.deletePrepull) } } catch { - // Opportunistic, like reconciling: a listing that cannot be repaired is still worth - // returning. + // Opportunistic, like reconciling validations: a listing that cannot be repaired is + // still worth returning. case e: Throwable => - logger.warn("Could not check the pre-pulls of ready images; leaving them as they are.", e) + logger.warn("Could not reconcile the curated-image pre-pulls; leaving them as they are.", e) } } @@ -461,9 +469,9 @@ object CuratedImageResource extends LazyLogging { .foreach(ImagePrepullClient.ensurePrepull(iid, _)) } else { // A failed refresh keeps the previous digest, so the row still names an image -- - // but no unit can be started from it any more. Left alone, its pre-pull would go - // on holding gigabytes on every node for something unusable, and nothing revisits - // a row that is not READY. + // but no unit can be started from it any more, and its pre-pull would go on + // holding gigabytes on every node for something unusable. Best-effort here; + // reconcilePrepulls is the backstop if this call does not happen. ImagePrepullClient.deletePrepull(iid) } } @@ -492,7 +500,7 @@ class CuratedImageResource extends LazyLogging { def list(@Auth user: SessionUser): List[CuratedImage] = { requireEnabled() reconcileRunningValidations() - ensurePrepullsForReadyImages() + reconcilePrepulls() context // Not select(): validation_log is unbounded and this endpoint discards it. .select(IID, NAME, SOURCE_REF, SOURCE_DIGEST, STATUS, ATTEMPT, CREATION_TIME, UPDATE_TIME) @@ -599,9 +607,9 @@ class CuratedImageResource extends LazyLogging { requireEnabled() // Nothing of ours holds a copy. A running unit keeps going on what its node pulled. ImageValidationClient.deleteAllValidations(iid) - // The row goes first. Removing the pre-pull before it leaves a window in which a - // concurrent read sees a row that is still READY and re-creates the pre-pull, and - // nothing would ever reap that one. + // The row goes first, so a concurrent read cannot see a row that is still READY and + // re-create the pre-pull just removed. If this call is the one that does not happen, + // reconcilePrepulls removes what no ready row wants on the next read. val deleted = context.deleteFrom(CU_IMAGE).where(IID.eq(iid)).execute() if (deleted == 0) { throw new NotFoundException(s"No curated image $iid.") diff --git a/computing-unit-managing-service/src/main/scala/org/apache/texera/service/util/ImagePrepullClient.scala b/computing-unit-managing-service/src/main/scala/org/apache/texera/service/util/ImagePrepullClient.scala index d07219d6175..d9369a230a5 100644 --- a/computing-unit-managing-service/src/main/scala/org/apache/texera/service/util/ImagePrepullClient.scala +++ b/computing-unit-managing-service/src/main/scala/org/apache/texera/service/util/ImagePrepullClient.scala @@ -133,21 +133,6 @@ object ImagePrepullClient extends LazyLogging { } } - /** Every pre-pull this service owns, removed. How turning pre-pulling off frees the disk. */ - def deleteAllPrepulls(): Unit = { - try { - client - .apps() - .daemonSets() - .inNamespace(namespace) - .withLabel(OwnerLabel, "true") - .delete() - } catch { - case e: Throwable => - logger.warn("Could not remove the curated-image pre-pulls.", e) - } - } - /** The reference a pre-pull pulls, which is its init container's image. */ private[service] def prepulledRefOf(daemonSet: DaemonSet): Option[String] = Option(daemonSet.getSpec) @@ -172,12 +157,22 @@ object ImagePrepullClient extends LazyLogging { ImageLabel -> iid.toString ).asJava - // Stated on both containers, not just the one that keeps running. A namespace with a + // Requests on both containers, not just the one that keeps running. A namespace with a // ResourceQuota on requests.cpu/requests.memory refuses a pod whose init container // leaves them out -- quota admission checks init containers too -- and the refusal is - // invisible, because the DaemonSet is still created. Costs nothing: a pod's request is - // the larger of its init containers and the sum of its others, so the same 1m/8Mi. - val resources = new ResourceRequirementsBuilder() + // invisible, because the DaemonSet is still created regardless. Costs nothing: a pod's + // request is the larger of its init containers and the sum of its others, so 1m/8Mi. + val prepullerResources = new ResourceRequirementsBuilder() + .addToRequests("cpu", new Quantity(CuratedImageConfig.prepullCpu)) + .addToRequests("memory", new Quantity(CuratedImageConfig.prepullMemory)) + .build() + + // Limits only here. Unlike requests, a limit is enforced per container and never maxed + // across them, so 8Mi on the init container would cap the shell of an arbitrary image + // -- bash, on the Python bases these are built from -- and OOMKill it. The pod would + // then crash-loop, never reach pause, and the image it just pulled would go back to + // being reclaimable, all while the DaemonSet reported itself created. + val pauseResources = new ResourceRequirementsBuilder() .addToRequests("cpu", new Quantity(CuratedImageConfig.prepullCpu)) .addToRequests("memory", new Quantity(CuratedImageConfig.prepullMemory)) .addToLimits("cpu", new Quantity(CuratedImageConfig.prepullCpu)) @@ -213,13 +208,13 @@ object ImagePrepullClient extends LazyLogging { // from what the registry would serve. Always would re-check it for nothing. .withImagePullPolicy("IfNotPresent") .withCommand("sh", "-c", "true") - .withResources(resources) + .withResources(prepullerResources) .build() ) .addNewContainer() .withName("pause") .withImage(CuratedImageConfig.prepullPauseImage) - .withResources(resources) + .withResources(pauseResources) .endContainer() .endSpec() .endTemplate() diff --git a/computing-unit-managing-service/src/test/scala/org/apache/texera/service/util/ImagePrepullClientSpec.scala b/computing-unit-managing-service/src/test/scala/org/apache/texera/service/util/ImagePrepullClientSpec.scala index cc3b5b53762..be7729e0e87 100644 --- a/computing-unit-managing-service/src/test/scala/org/apache/texera/service/util/ImagePrepullClientSpec.scala +++ b/computing-unit-managing-service/src/test/scala/org/apache/texera/service/util/ImagePrepullClientSpec.scala @@ -80,6 +80,21 @@ class ImagePrepullClientSpec extends AnyFlatSpec with Matchers { } } + // The regression this guards: requests and limits were built once and shared, which put + // an 8Mi cap on the init container. A limit is enforced per container and never maxed + // across them, so the shell of an arbitrary image -- bash, on the Python bases these are + // built from -- was OOMKilled, the pod crash-looped, pause never ran, and the image went + // back to being reclaimable, all while the DaemonSet reported itself created. + it should "cap the pause container only, never the image's own shell" in { + val podSpec = prepullDaemonSet(7, PinnedRef).getSpec.getTemplate.getSpec + + val prepuller = podSpec.getInitContainers.asScala.head + Option(prepuller.getResources).map(_.getLimits.asScala).getOrElse(Map.empty) shouldBe empty + + val pause = podSpec.getContainers.asScala.head + pause.getResources.getLimits.asScala.keySet should contain allOf ("cpu", "memory") + } + "prepulledRefOf" should "read back what a pre-pull actually pulls" in { // What lets a stale pre-pull be spotted: a refresh whose repoint failed leaves one // holding the previous digest, and comparing only ids would never notice. From bf7bfaf06977006f977e81185a43ffbfad76df8a Mon Sep 17 00:00:00 2001 From: Tanishq Gandhi Date: Thu, 10 Sep 2026 22:21:43 +0000 Subject: [PATCH 4/7] fix(computing-unit): stop a refresh tearing the pre-pull off every node The reaper added last time kept only READY rows, but a refresh moves a healthy image through VALIDATING first. So refreshing anything deleted its pre-pull from every node and rebuilt it moments later -- and the admin page polls this endpoint for as long as a check is running, so that read was certain to land inside the window. A row still being checked now keeps its pre-pull; only a row that is gone or FAILED is reaped. Also: - a create that fails is not retried for five minutes, keyed by the reference so a new digest is still tried at once. Reconciling runs on every read of the list, by any signed-in user, so a failure that will not clear on its own -- the Role not reapplied after an upgrade -- was one doomed call and one stack trace per ready image per page load, indefinitely. - the reference reaches finishValidation from the row the reconcile already read, rather than a second query. A row deleted in between left pinnedRef building "@sha256:..." with no repository at all, and a DaemonSet that could never pull it. - deletePrepull states its propagation policy, as the validation client's deletes already do. Were the default ever Orphan, the pods would stay on every node holding the image, and the reconcile pass lists DaemonSets, so nothing would find them again. --- .../config/src/main/resources/kubernetes.conf | 7 +++ .../common/config/CuratedImageConfig.scala | 2 + .../resource/CuratedImageResource.scala | 62 ++++++++++++------- .../service/util/ImagePrepullClient.scala | 44 +++++++++++++ .../service/util/ImagePrepullClientSpec.scala | 38 ++++++++++++ 5 files changed, 130 insertions(+), 23 deletions(-) diff --git a/common/config/src/main/resources/kubernetes.conf b/common/config/src/main/resources/kubernetes.conf index dd861134eb7..c9e5d79cd73 100644 --- a/common/config/src/main/resources/kubernetes.conf +++ b/common/config/src/main/resources/kubernetes.conf @@ -168,6 +168,13 @@ curated-images { prepull-namespace = "texera-workflow-computing-unit-pool" prepull-namespace = ${?TEXERA_CURATED_IMAGE_PREPULL_NAMESPACE} + # How long a pre-pull that could not be created is left alone before it is tried again. + # Without a pause, a failure that is not going to clear on its own -- the Role not + # reapplied after an upgrade, an admission webhook refusing the pod -- is retried on + # every read of the image list, by every user, for every ready image. + prepull-retry-cooldown-seconds = 300 + prepull-retry-cooldown-seconds = ${?TEXERA_CURATED_IMAGE_PREPULL_RETRY_COOLDOWN_SECONDS} + # Holds the pod open once the init container has pulled the image, so the node does not # reclaim what was just pulled. registry.k8s.io is where the pause image now lives; the # gcr.io/google_containers path the chart's own pre-puller still uses is retired. diff --git a/common/config/src/main/scala/org/apache/texera/common/config/CuratedImageConfig.scala b/common/config/src/main/scala/org/apache/texera/common/config/CuratedImageConfig.scala index 993cbdc7f6b..f85f734071b 100644 --- a/common/config/src/main/scala/org/apache/texera/common/config/CuratedImageConfig.scala +++ b/common/config/src/main/scala/org/apache/texera/common/config/CuratedImageConfig.scala @@ -48,6 +48,8 @@ object CuratedImageConfig { val prepullEnabled: Boolean = conf.getBoolean("curated-images.prepull-enabled") val prepullNamespace: String = conf.getString("curated-images.prepull-namespace") val prepullPauseImage: String = conf.getString("curated-images.prepull-pause-image") + val prepullRetryCooldownSeconds: Int = + conf.getInt("curated-images.prepull-retry-cooldown-seconds") val prepullCpu: String = conf.getString("curated-images.prepull-cpu") val prepullMemory: String = conf.getString("curated-images.prepull-memory") diff --git a/computing-unit-managing-service/src/main/scala/org/apache/texera/service/resource/CuratedImageResource.scala b/computing-unit-managing-service/src/main/scala/org/apache/texera/service/resource/CuratedImageResource.scala index faf132c5b7d..92d2711e83a 100644 --- a/computing-unit-managing-service/src/main/scala/org/apache/texera/service/resource/CuratedImageResource.scala +++ b/computing-unit-managing-service/src/main/scala/org/apache/texera/service/resource/CuratedImageResource.scala @@ -197,10 +197,6 @@ object CuratedImageResource extends LazyLogging { def nameOf(iid: Int): Option[String] = Option(context.select(NAME).from(CU_IMAGE).where(IID.eq(iid)).fetchOne()).map(_.get(NAME)) - private def sourceRefOf(iid: Int): Option[String] = - Option(context.select(SOURCE_REF).from(CU_IMAGE).where(IID.eq(iid)).fetchOne()) - .map(_.get(SOURCE_REF)) - /** * Brings the pre-pulls in the cluster into line with the ready images in the database: * adds the missing, repoints the stale, and removes the rest. @@ -221,20 +217,23 @@ object CuratedImageResource extends LazyLogging { // already made are removed rather than merely left un-added to. Driven by what the // listing actually found, so a deployment that has been off stops calling once // there is nothing left rather than issuing a delete on every page load. - val wantedByImage: Map[Int, String] = - if (!CuratedImageConfig.prepullEnabled) Map.empty + val rows = + if (!CuratedImageConfig.prepullEnabled) Nil else context - .select(IID, SOURCE_REF, SOURCE_DIGEST) + .select(IID, STATUS, SOURCE_REF, SOURCE_DIGEST) .from(CU_IMAGE) - .where(STATUS.eq(Status.Ready)) .fetch() .asScala - .flatMap { row => - pinnedRefOf(row.get(SOURCE_REF), row.get(SOURCE_DIGEST)) - .map(row.get(IID).intValue() -> _) - } - .toMap + .toList + + val wantedByImage: Map[Int, String] = rows + .filter(_.get(STATUS) == Status.Ready) + .flatMap { row => + pinnedRefOf(row.get(SOURCE_REF), row.get(SOURCE_DIGEST)) + .map(row.get(IID).intValue() -> _) + } + .toMap // Compared by reference, not merely by presence: a refresh whose repoint failed // leaves a pre-pull holding the previous digest, and nothing else would correct it. @@ -243,9 +242,19 @@ object CuratedImageResource extends LazyLogging { if (!prepulled.get(iid).contains(wanted)) ImagePrepullClient.ensurePrepull(iid, wanted) } - // Orphans: a row deleted while the cluster was unreachable, an image that left - // READY, or a process that stopped between the two. No row refers to these. - (prepulled.keySet -- wantedByImage.keySet).foreach(ImagePrepullClient.deletePrepull) + // A row still being checked keeps its pre-pull. Refreshing a healthy image moves it + // through VALIDATING, and reaping on anything short of READY would pull the image + // off every node and put it back again on a routine action -- guaranteed to happen, + // since the admin page polls this endpoint for as long as a check is running. + val keep = rows + .filterNot(_.get(STATUS) == Status.Failed) + .map(_.get(IID).intValue()) + .toSet + + // What is left is genuinely unwanted: a row deleted while the cluster was + // unreachable, an image that failed its last check, or a process that stopped + // between the database write and the call that should have followed it. + (prepulled.keySet -- keep).foreach(ImagePrepullClient.deletePrepull) } } catch { // Opportunistic, like reconciling validations: a listing that cannot be repaired is @@ -275,7 +284,7 @@ object CuratedImageResource extends LazyLogging { // The cluster may be unreachable, or the Role not yet reapplied after an upgrade. // Reconciling is opportunistic, so a row that cannot be checked is left as it is // rather than failing a read that would otherwise return every other image. - try reconcileOne(iid, attempt, row.get(UPDATE_TIME).getTime) + try reconcileOne(iid, attempt, row.get(UPDATE_TIME).getTime, row.get(SOURCE_REF)) catch { case e: Throwable => logger.warn(s"Could not check the validation of image $iid; leaving it as it is.", e) @@ -283,7 +292,7 @@ object CuratedImageResource extends LazyLogging { } } - private def reconcileOne(iid: Int, attempt: Int, updatedAt: Long): Unit = { + private def reconcileOne(iid: Int, attempt: Int, updatedAt: Long, sourceRef: String): Unit = { // State first, then the log. The other order can read a log written while the job was // still running and then judge it against a state that says it finished -- the digest // line would be missing and a successful validation would be recorded as failed. @@ -312,7 +321,8 @@ object CuratedImageResource extends LazyLogging { // from and calling it ready would strand a unit in ImagePullBackOff. if (digest.isDefined) Status.Ready else Status.Failed, digest, - text + validationNote(digest, digest.flatMap(sameContentNote(iid, _))) + text + validationNote(digest, digest.flatMap(sameContentNote(iid, _))), + sourceRef ) case ValidationState.Failed => @@ -326,7 +336,8 @@ object CuratedImageResource extends LazyLogging { attempt, Status.Failed, None, - reason.getOrElse("The validation failed without reporting a reason.") + reason.getOrElse("The validation failed without reporting a reason."), + sourceRef ) case ValidationState.Absent => @@ -340,7 +351,8 @@ object CuratedImageResource extends LazyLogging { attempt, Status.Failed, None, - "The validation job disappeared before it reported a result." + "The validation job disappeared before it reported a result.", + sourceRef ) } } @@ -446,7 +458,8 @@ object CuratedImageResource extends LazyLogging { attempt: Int, status: String, sourceDigest: Option[String], - log: String + log: String, + sourceRef: String ): Unit = { val update = context .update(CU_IMAGE) @@ -465,7 +478,10 @@ object CuratedImageResource extends LazyLogging { // A refresh that resolved a moved tag arrives here too, and points the pre-pull at // the new digest rather than leaving the node holding bytes nothing runs. if (status == Status.Ready) { - pinnedRefOf(sourceRefOf(iid).orNull, sourceDigest.orNull) + // The reference comes from the row this reconcile already read, not a fresh query: + // a row deleted in between would leave pinnedRefOf building a reference with no + // repository at all -- "@sha256:..." -- and a DaemonSet that cannot pull it. + pinnedRefOf(sourceRef, sourceDigest.orNull) .foreach(ImagePrepullClient.ensurePrepull(iid, _)) } else { // A failed refresh keeps the previous digest, so the row still names an image -- diff --git a/computing-unit-managing-service/src/main/scala/org/apache/texera/service/util/ImagePrepullClient.scala b/computing-unit-managing-service/src/main/scala/org/apache/texera/service/util/ImagePrepullClient.scala index d9369a230a5..4cb354a00f6 100644 --- a/computing-unit-managing-service/src/main/scala/org/apache/texera/service/util/ImagePrepullClient.scala +++ b/computing-unit-managing-service/src/main/scala/org/apache/texera/service/util/ImagePrepullClient.scala @@ -23,6 +23,7 @@ import com.typesafe.scalalogging.LazyLogging import io.fabric8.kubernetes.api.model.{Quantity, ResourceRequirementsBuilder} import io.fabric8.kubernetes.api.model.apps.{DaemonSet, DaemonSetBuilder} import io.fabric8.kubernetes.client.KubernetesClientBuilder +import io.fabric8.kubernetes.api.model.DeletionPropagation import org.apache.texera.common.config.CuratedImageConfig import scala.jdk.CollectionConverters._ @@ -56,6 +57,39 @@ object ImagePrepullClient extends LazyLogging { /** The image a pre-pull belongs to, so one image's can be removed without the others. */ private[service] val ImageLabel = "texera-cu-image" + /** + * The reference each image's last failed create was for, and when it failed. + * + * Reconciling runs on every read of the image list, by any signed-in user, so a failure + * that will not clear on its own -- the Role not reapplied after an upgrade, an + * admission webhook refusing the pod -- would otherwise be retried on every page load + * for every ready image, each one logging a stack trace. Keyed by reference as well as + * time so that a genuinely new digest is tried at once rather than waiting out a + * cooldown earned by the previous one. + */ + private val lastFailure = new java.util.concurrent.ConcurrentHashMap[Int, (String, Long)]() + + private def failedRecently(iid: Int, pinnedRef: String): Boolean = + isCoolingDown(Option(lastFailure.get(iid)), pinnedRef, System.currentTimeMillis()) + + /** + * Whether a create should be held back. Separated from the clock and the map so the rule + * itself can be stated in a test rather than inferred from whether a cluster happened to + * answer. + */ + private[service] def isCoolingDown( + recorded: Option[(String, Long)], + pinnedRef: String, + now: Long + ): Boolean = + recorded.exists { + case (failedRef, at) => + // The reference has to match: a new digest is a new question, and should be asked + // at once rather than serving out a cooldown the previous one earned. + failedRef == pinnedRef && + now - at < CuratedImageConfig.prepullRetryCooldownSeconds * 1000L + } + /** * Creates the pre-pull for an image, or points an existing one at a new reference. Called * whenever a row reaches READY, which covers a refresh that resolved a moved tag to a @@ -63,6 +97,7 @@ object ImagePrepullClient extends LazyLogging { */ def ensurePrepull(iid: Int, pinnedRef: String): Unit = { if (!CuratedImageConfig.prepullEnabled) return + if (failedRecently(iid, pinnedRef)) return try { val daemonSet = prepullDaemonSet(iid, pinnedRef) client @@ -71,9 +106,11 @@ object ImagePrepullClient extends LazyLogging { .inNamespace(namespace) .resource(daemonSet) .createOr(existing => existing.update()) + lastFailure.remove(iid) logger.info(s"Pre-pulling curated image $iid ($pinnedRef) onto every node.") } catch { case e: Throwable => + lastFailure.put(iid, (pinnedRef, System.currentTimeMillis())) // The image is still usable; only the head start is lost. logger.warn( s"Could not pre-pull curated image $iid ($pinnedRef). The first unit on each " + @@ -95,6 +132,10 @@ object ImagePrepullClient extends LazyLogging { .daemonSets() .inNamespace(namespace) .withName(CuratedImageConfig.prepullName(iid)) + // Stated rather than left to a default: were it ever Orphan, the pods would stay + // on every node still holding the image, and the reconcile pass lists DaemonSets, + // so nothing would ever find them again. + .withPropagationPolicy(DeletionPropagation.BACKGROUND) .delete() } catch { case e: Throwable => @@ -102,6 +143,9 @@ object ImagePrepullClient extends LazyLogging { // naming in the log rather than passing over. logger.warn(s"Could not remove the pre-pull for curated image $iid.", e) } + // Whether or not the delete landed, nothing is now known to be wrong with this image's + // reference, and a create asked for next should not be held back by an old failure. + lastFailure.remove(iid) } /** diff --git a/computing-unit-managing-service/src/test/scala/org/apache/texera/service/util/ImagePrepullClientSpec.scala b/computing-unit-managing-service/src/test/scala/org/apache/texera/service/util/ImagePrepullClientSpec.scala index be7729e0e87..5dc95b40e30 100644 --- a/computing-unit-managing-service/src/test/scala/org/apache/texera/service/util/ImagePrepullClientSpec.scala +++ b/computing-unit-managing-service/src/test/scala/org/apache/texera/service/util/ImagePrepullClientSpec.scala @@ -95,6 +95,44 @@ class ImagePrepullClientSpec extends AnyFlatSpec with Matchers { pause.getResources.getLimits.asScala.keySet should contain allOf ("cpu", "memory") } + // The regression this guards: a persistent failure -- the Role not reapplied after an + // upgrade -- was retried on every read of the image list, by every user, for every ready + // image, logging a stack trace each time. + "isCoolingDown" should "hold back a reference that just failed" in { + val now = 1_000_000_000L + val cooldownMillis = CuratedImageConfig.prepullRetryCooldownSeconds * 1000L + + ImagePrepullClient.isCoolingDown(Some((PinnedRef, now)), PinnedRef, now) shouldBe true + ImagePrepullClient.isCoolingDown( + Some((PinnedRef, now - cooldownMillis + 1)), + PinnedRef, + now + ) shouldBe true + } + + it should "try again once the cooldown has passed" in { + val now = 1_000_000_000L + val cooldownMillis = CuratedImageConfig.prepullRetryCooldownSeconds * 1000L + ImagePrepullClient.isCoolingDown( + Some((PinnedRef, now - cooldownMillis)), + PinnedRef, + now + ) shouldBe false + } + + it should "not hold back a different digest" in { + // A refresh that resolved a new digest is a new question. Making it wait out a cooldown + // the previous reference earned would leave nodes on the superseded image for as long + // as the cooldown lasts. + val now = 1_000_000_000L + val other = "owner/name@sha256:" + "f" * 64 + ImagePrepullClient.isCoolingDown(Some((PinnedRef, now)), other, now) shouldBe false + } + + it should "not hold back an image that has never failed" in { + ImagePrepullClient.isCoolingDown(None, PinnedRef, 1_000_000_000L) shouldBe false + } + "prepulledRefOf" should "read back what a pre-pull actually pulls" in { // What lets a stale pre-pull be spotted: a refresh whose repoint failed leaves one // holding the previous digest, and comparing only ids would never notice. From b1693b50768b3c95357c114a68536ba3c47653a4 Mon Sep 17 00:00:00 2001 From: Tanishq Gandhi Date: Thu, 10 Sep 2026 22:22:12 +0000 Subject: [PATCH 5/7] docs(computing-unit): name the function the pre-pull test comment refers to --- .../org/apache/texera/service/util/ImagePrepullClientSpec.scala | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/computing-unit-managing-service/src/test/scala/org/apache/texera/service/util/ImagePrepullClientSpec.scala b/computing-unit-managing-service/src/test/scala/org/apache/texera/service/util/ImagePrepullClientSpec.scala index 5dc95b40e30..af1ef9010d8 100644 --- a/computing-unit-managing-service/src/test/scala/org/apache/texera/service/util/ImagePrepullClientSpec.scala +++ b/computing-unit-managing-service/src/test/scala/org/apache/texera/service/util/ImagePrepullClientSpec.scala @@ -177,7 +177,7 @@ class ImagePrepullClientSpec extends AnyFlatSpec with Matchers { } it should "ignore a DaemonSet that is not one of ours" in { - // prepulledImageIds lists by label, but a cluster can hold anything. A stray object + // prepulledRefs lists by label, but a cluster can hold anything. A stray object // must not be read as an image id and leave a real image without its pre-pull. val unlabelled = new DaemonSetBuilder().withNewMetadata().withName("something").endMetadata() imageIdOf(unlabelled.build()) shouldBe None From 26d9062479d1469ec8f1225db86254793c06c327 Mon Sep 17 00:00:00 2001 From: Tanishq Gandhi Date: Thu, 10 Sep 2026 22:35:29 +0000 Subject: [PATCH 6/7] fix(computing-unit): repoint every node at once, and let refresh retry Two from review, plus a pass over the comments. A repoint inherited the default rolling update, one node at a time, each waiting for a full pull -- hours on a large cluster, for a pod with no availability to protect. maxUnavailable is now 100%. Refreshing is the only remedy the page offers for a pre-pull that could not be created, but with the digest unchanged the cooldown swallowed it, so the button appeared to do nothing for five minutes. Starting a validation now clears the image's recorded failure. Comments trimmed throughout: they had grown into accounts of what an earlier version did wrong rather than what the code does. --- ...low-computing-unit-manager-deployment.yaml | 3 +- ...omputing-unit-manager-service-account.yaml | 9 +- bin/k8s/values.yaml | 7 +- .../config/src/main/resources/kubernetes.conf | 28 ++-- .../resource/CuratedImageResource.scala | 57 +++----- .../service/util/ImagePrepullClient.scala | 128 +++++++----------- .../service/util/ImagePrepullClientSpec.scala | 50 +++---- 7 files changed, 108 insertions(+), 174 deletions(-) diff --git a/bin/k8s/templates/base/workflow-computing-unit-manager/workflow-computing-unit-manager-deployment.yaml b/bin/k8s/templates/base/workflow-computing-unit-manager/workflow-computing-unit-manager-deployment.yaml index 414f439f02d..e758a515940 100644 --- a/bin/k8s/templates/base/workflow-computing-unit-manager/workflow-computing-unit-manager-deployment.yaml +++ b/bin/k8s/templates/base/workflow-computing-unit-manager/workflow-computing-unit-manager-deployment.yaml @@ -73,8 +73,7 @@ spec: value: {{ .Values.workflowComputingUnitPool.namespace }} - name: TEXERA_CURATED_IMAGE_PREPULL_ENABLED value: "{{ .Values.curatedImages.prepull.enabled }}" - # The pool namespace, whose quota the declared requests satisfy. Not the - # release namespace, which holds the privileged mounter. + # Not the release namespace, which holds the privileged mounter. - name: TEXERA_CURATED_IMAGE_PREPULL_NAMESPACE value: {{ .Values.workflowComputingUnitPool.namespace }} - name: KUBERNETES_IMAGE_NAME diff --git a/bin/k8s/templates/base/workflow-computing-unit-manager/workflow-computing-unit-manager-service-account.yaml b/bin/k8s/templates/base/workflow-computing-unit-manager/workflow-computing-unit-manager-service-account.yaml index 231d79e3328..a1a0ed40cc7 100644 --- a/bin/k8s/templates/base/workflow-computing-unit-manager/workflow-computing-unit-manager-service-account.yaml +++ b/bin/k8s/templates/base/workflow-computing-unit-manager/workflow-computing-unit-manager-service-account.yaml @@ -42,12 +42,9 @@ rules: - apiGroups: [""] resources: ["pods/log"] verbs: ["get"] - # One DaemonSet per ready curated image, pulling it onto every node. Created while the - # cluster runs, as images are registered, so the chart cannot declare them. - # - # Scoped to the pool namespace deliberately. The release namespace holds the privileged - # hostPath mounter, and permission to rewrite a DaemonSet there is permission to run as - # root on every node. Nothing in the pool namespace is privileged. + # One DaemonSet per ready curated image, created as images are registered, so the chart + # cannot declare them. Scoped here and not to the release namespace, which holds the + # privileged mounter. - apiGroups: ["apps"] resources: ["daemonsets"] verbs: ["get", "list", "watch", "create", "update", "patch", "delete"] diff --git a/bin/k8s/values.yaml b/bin/k8s/values.yaml index 7bc7486e666..85819ab006d 100644 --- a/bin/k8s/values.yaml +++ b/bin/k8s/values.yaml @@ -377,10 +377,9 @@ curatedImages: # Off until the UI to manage these ships. enabled: false prepull: - # Pull every ready image onto every node, so the first unit on a node does not wait for - # it. Costs node disk: each node holds each ready image. Turn off to pay the pull on - # first use instead -- doing so also removes the pre-pulls already made, which is what - # frees the disk, on the next read of the image list. + # Pull every ready image onto every node, so the first unit there does not wait for it. + # Costs node disk: each node holds each ready image. Turning it off also removes the + # pre-pulls already made. enabled: true # headless service for the access of computing units diff --git a/common/config/src/main/resources/kubernetes.conf b/common/config/src/main/resources/kubernetes.conf index c9e5d79cd73..57964fc83d0 100644 --- a/common/config/src/main/resources/kubernetes.conf +++ b/common/config/src/main/resources/kubernetes.conf @@ -152,36 +152,28 @@ curated-images { validation-memory-limit = "256Mi" validation-memory-limit = ${?TEXERA_CURATED_IMAGE_VALIDATION_MEMORY_LIMIT} - # Pull each ready image onto every node as soon as it is ready, rather than when someone - # first starts a unit from it. Costs node disk -- every node holds every ready image -- - # so a deployment short of it can turn this off and pay the pull on first use instead. - # Turning it off also removes the pre-pulls already made, which is what frees the disk; - # that sweep runs on the next read of the image list, so it needs curated images - # themselves left enabled. + # Pull each ready image onto every node as soon as it is ready. Costs node disk, since + # every node holds every ready image. Turning it off also removes the pre-pulls already + # made, on the next read of the image list. prepull-enabled = true prepull-enabled = ${?TEXERA_CURATED_IMAGE_PREPULL_ENABLED} - # The pool namespace, not the release namespace. Its ResourceQuota is satisfied by the - # requests both containers declare, so nothing is gained by moving out -- and the release - # namespace holds the privileged hostPath mounter, where the permission to write - # DaemonSets would be the permission to run as root on every node. + # The pool namespace. The release namespace holds the privileged mounter, where write + # access to DaemonSets would mean root on every node. prepull-namespace = "texera-workflow-computing-unit-pool" prepull-namespace = ${?TEXERA_CURATED_IMAGE_PREPULL_NAMESPACE} - # How long a pre-pull that could not be created is left alone before it is tried again. - # Without a pause, a failure that is not going to clear on its own -- the Role not - # reapplied after an upgrade, an admission webhook refusing the pod -- is retried on - # every read of the image list, by every user, for every ready image. + # How long before a failed create is tried again. Without a pause, a failure that will + # not clear is retried on every read of the image list, by every user. prepull-retry-cooldown-seconds = 300 prepull-retry-cooldown-seconds = ${?TEXERA_CURATED_IMAGE_PREPULL_RETRY_COOLDOWN_SECONDS} - # Holds the pod open once the init container has pulled the image, so the node does not - # reclaim what was just pulled. registry.k8s.io is where the pause image now lives; the - # gcr.io/google_containers path the chart's own pre-puller still uses is retired. + # Holds the pod open once the image is pulled, so the node does not reclaim it. + # registry.k8s.io is where the pause image now lives. prepull-pause-image = "registry.k8s.io/pause:3.9" prepull-pause-image = ${?TEXERA_CURATED_IMAGE_PREPULL_PAUSE_IMAGE} - # The pause container does nothing but exist. Same figures the chart's pre-puller uses. + # The pause container does nothing but exist. prepull-cpu = "1m" prepull-cpu = ${?TEXERA_CURATED_IMAGE_PREPULL_CPU} prepull-memory = "8Mi" diff --git a/computing-unit-managing-service/src/main/scala/org/apache/texera/service/resource/CuratedImageResource.scala b/computing-unit-managing-service/src/main/scala/org/apache/texera/service/resource/CuratedImageResource.scala index 92d2711e83a..e3f7c713b5d 100644 --- a/computing-unit-managing-service/src/main/scala/org/apache/texera/service/resource/CuratedImageResource.scala +++ b/computing-unit-managing-service/src/main/scala/org/apache/texera/service/resource/CuratedImageResource.scala @@ -198,13 +198,10 @@ object CuratedImageResource extends LazyLogging { Option(context.select(NAME).from(CU_IMAGE).where(IID.eq(iid)).fetchOne()).map(_.get(NAME)) /** - * Brings the pre-pulls in the cluster into line with the ready images in the database: - * adds the missing, repoints the stale, and removes the rest. - * - * Reconciling rather than only adding is what makes the pre-pulls reapable at all. - * Every other path that removes one -- a deleted row, an image that left READY -- can - * be interrupted between the database write and the cluster call, and nothing else - * revisits a row that is gone or not ready. Whatever is left over is found here. + * Brings the pre-pulls into line with the ready images: adds the missing, repoints the + * stale, removes the rest. Reaping here is what makes them reapable at all -- every + * other removal path can be interrupted between the database write and the cluster + * call, and nothing else revisits a row once it is gone. * * One labelled list and one query, so a settled deployment does no work. */ @@ -213,10 +210,8 @@ object CuratedImageResource extends LazyLogging { // None means the cluster could not be asked, which is not "nothing is pre-pulled": // acting on that would fire a doomed call for every image on every read. ImagePrepullClient.prepulledRefs().foreach { prepulled => - // Turning pre-pulling off is what frees the node disk it costs, so the pre-pulls - // already made are removed rather than merely left un-added to. Driven by what the - // listing actually found, so a deployment that has been off stops calling once - // there is nothing left rather than issuing a delete on every page load. + // No rows wanted when pre-pulling is off, so everything found is reaped below -- + // that is what frees the disk the setting costs. val rows = if (!CuratedImageConfig.prepullEnabled) Nil else @@ -235,30 +230,26 @@ object CuratedImageResource extends LazyLogging { } .toMap - // Compared by reference, not merely by presence: a refresh whose repoint failed - // leaves a pre-pull holding the previous digest, and nothing else would correct it. + // By reference, not just presence: a repoint that failed leaves a pre-pull on the + // previous digest, which nothing else would correct. wantedByImage.foreach { case (iid, wanted) => if (!prepulled.get(iid).contains(wanted)) ImagePrepullClient.ensurePrepull(iid, wanted) } - // A row still being checked keeps its pre-pull. Refreshing a healthy image moves it - // through VALIDATING, and reaping on anything short of READY would pull the image - // off every node and put it back again on a routine action -- guaranteed to happen, - // since the admin page polls this endpoint for as long as a check is running. + // A row still being checked keeps its pre-pull: a refresh moves a healthy image + // through VALIDATING, and reaping then would strip it from every node and put it + // straight back. val keep = rows .filterNot(_.get(STATUS) == Status.Failed) .map(_.get(IID).intValue()) .toSet - // What is left is genuinely unwanted: a row deleted while the cluster was - // unreachable, an image that failed its last check, or a process that stopped - // between the database write and the call that should have followed it. + // What is left has no live row: deleted, failed, or interrupted mid-removal. (prepulled.keySet -- keep).foreach(ImagePrepullClient.deletePrepull) } } catch { - // Opportunistic, like reconciling validations: a listing that cannot be repaired is - // still worth returning. + // Opportunistic: a listing that cannot be repaired is still worth returning. case e: Throwable => logger.warn("Could not reconcile the curated-image pre-pulls; leaving them as they are.", e) } @@ -474,20 +465,14 @@ object CuratedImageResource extends LazyLogging { // in the pool namespace. if (stored > 0) { ImageValidationClient.deleteValidation(iid, attempt) - // Only on READY, because only then is there something a unit can be started from. - // A refresh that resolved a moved tag arrives here too, and points the pre-pull at - // the new digest rather than leaving the node holding bytes nothing runs. + // On READY only, since only then can a unit start from it. A refresh that resolved + // a moved tag arrives here too and repoints the pre-pull. The reference comes from + // the row already read, not a fresh query, which could find it deleted. if (status == Status.Ready) { - // The reference comes from the row this reconcile already read, not a fresh query: - // a row deleted in between would leave pinnedRefOf building a reference with no - // repository at all -- "@sha256:..." -- and a DaemonSet that cannot pull it. pinnedRefOf(sourceRef, sourceDigest.orNull) .foreach(ImagePrepullClient.ensurePrepull(iid, _)) } else { - // A failed refresh keeps the previous digest, so the row still names an image -- - // but no unit can be started from it any more, and its pre-pull would go on - // holding gigabytes on every node for something unusable. Best-effort here; - // reconcilePrepulls is the backstop if this call does not happen. + // Otherwise no unit can start from it, so its pre-pull should not hold the image. ImagePrepullClient.deletePrepull(iid) } } @@ -623,9 +608,8 @@ class CuratedImageResource extends LazyLogging { requireEnabled() // Nothing of ours holds a copy. A running unit keeps going on what its node pulled. ImageValidationClient.deleteAllValidations(iid) - // The row goes first, so a concurrent read cannot see a row that is still READY and - // re-create the pre-pull just removed. If this call is the one that does not happen, - // reconcilePrepulls removes what no ready row wants on the next read. + // The row goes first, so a concurrent read cannot see it still READY and re-create the + // pre-pull. If this call is lost, reconcilePrepulls reaps it on the next read. val deleted = context.deleteFrom(CU_IMAGE).where(IID.eq(iid)).execute() if (deleted == 0) { throw new NotFoundException(s"No curated image $iid.") @@ -637,6 +621,9 @@ class CuratedImageResource extends LazyLogging { /** Marks the row as being validated and submits the job, in that order. */ private def startValidation(iid: Int, sourceRef: String): Unit = { + // Refreshing is the only remedy the page offers, so it must not be held back by the + // cooldown a previous failure earned -- that would look like the button doing nothing. + ImagePrepullClient.clearFailure(iid) // Read and claimed in one statement. Two refreshes at the same moment would otherwise // both compute the same attempt, and the second would delete the first's job. val attempt = Option( diff --git a/computing-unit-managing-service/src/main/scala/org/apache/texera/service/util/ImagePrepullClient.scala b/computing-unit-managing-service/src/main/scala/org/apache/texera/service/util/ImagePrepullClient.scala index 4cb354a00f6..e13da6b6059 100644 --- a/computing-unit-managing-service/src/main/scala/org/apache/texera/service/util/ImagePrepullClient.scala +++ b/computing-unit-managing-service/src/main/scala/org/apache/texera/service/util/ImagePrepullClient.scala @@ -20,29 +20,23 @@ package org.apache.texera.service.util import com.typesafe.scalalogging.LazyLogging -import io.fabric8.kubernetes.api.model.{Quantity, ResourceRequirementsBuilder} import io.fabric8.kubernetes.api.model.apps.{DaemonSet, DaemonSetBuilder} +import io.fabric8.kubernetes.api.model.{DeletionPropagation, Quantity, ResourceRequirementsBuilder} import io.fabric8.kubernetes.client.KubernetesClientBuilder -import io.fabric8.kubernetes.api.model.DeletionPropagation import org.apache.texera.common.config.CuratedImageConfig import scala.jdk.CollectionConverters._ /** - * Puts a ready curated image on every node before anyone starts a unit from it. + * Puts a ready curated image on every node before anyone starts a unit from it, so the + * first unit there does not wait for the pull. * - * Without this the first unit on a node waits for the whole image -- about 80 seconds for - * a 3 GB one -- while every later unit there starts at once, so the same action takes - * seconds or minutes depending only on where it landed. + * One DaemonSet per image, the same shape the chart uses for the deployment's own image: + * an init container that is the image and does nothing, then a pause container to hold + * the pod open so the node does not reclaim what was pulled. Built here rather than in + * the chart because curated images are registered while the cluster is running. * - * The mechanism is the one the chart already uses for the deployment's own image: a - * DaemonSet whose init container is the image and whose command does nothing, then a pause - * container to hold the pod open so the node does not reclaim what was just pulled. The - * chart cannot express these, because a curated image is registered while the cluster is - * running, so they are built here instead. - * - * Every call is best-effort. A pre-pull that cannot be created is logged and ignored: the - * image still works, and the first unit on each node just pays for the pull. + * Every call is best-effort: a failure is logged, and the image still works. */ object ImagePrepullClient extends LazyLogging { @@ -51,31 +45,24 @@ object ImagePrepullClient extends LazyLogging { private def namespace: String = CuratedImageConfig.prepullNamespace - /** Marks every pre-pull this service owns, so they are found by label and not by name. */ + /** Marks the pre-pulls this service owns, so they are found by label, not by name. */ private[service] val OwnerLabel = "texera-cu-image-prepull" - /** The image a pre-pull belongs to, so one image's can be removed without the others. */ + /** Which image a pre-pull is for, so one can be removed without the others. */ private[service] val ImageLabel = "texera-cu-image" - /** - * The reference each image's last failed create was for, and when it failed. - * - * Reconciling runs on every read of the image list, by any signed-in user, so a failure - * that will not clear on its own -- the Role not reapplied after an upgrade, an - * admission webhook refusing the pod -- would otherwise be retried on every page load - * for every ready image, each one logging a stack trace. Keyed by reference as well as - * time so that a genuinely new digest is tried at once rather than waiting out a - * cooldown earned by the previous one. - */ + /** Reference and time of each image's last failed create. See [[isCoolingDown]]. */ private val lastFailure = new java.util.concurrent.ConcurrentHashMap[Int, (String, Long)]() private def failedRecently(iid: Int, pinnedRef: String): Boolean = isCoolingDown(Option(lastFailure.get(iid)), pinnedRef, System.currentTimeMillis()) /** - * Whether a create should be held back. Separated from the clock and the map so the rule - * itself can be stated in a test rather than inferred from whether a cluster happened to - * answer. + * Whether to hold a create back. Reconciling runs on every read of the image list, so + * without this a failure that will not clear is retried on every page load. Keyed by + * reference, so a new digest is tried at once. + * + * Pure, so the rule can be tested without a cluster. */ private[service] def isCoolingDown( recorded: Option[(String, Long)], @@ -84,34 +71,29 @@ object ImagePrepullClient extends LazyLogging { ): Boolean = recorded.exists { case (failedRef, at) => - // The reference has to match: a new digest is a new question, and should be asked - // at once rather than serving out a cooldown the previous one earned. failedRef == pinnedRef && now - at < CuratedImageConfig.prepullRetryCooldownSeconds * 1000L } - /** - * Creates the pre-pull for an image, or points an existing one at a new reference. Called - * whenever a row reaches READY, which covers a refresh that resolved a moved tag to a - * different digest. - */ + /** Forgets an image's last failure, so the next create is attempted immediately. */ + def clearFailure(iid: Int): Unit = lastFailure.remove(iid) + + /** Creates an image's pre-pull, or points an existing one at a new reference. */ def ensurePrepull(iid: Int, pinnedRef: String): Unit = { if (!CuratedImageConfig.prepullEnabled) return if (failedRecently(iid, pinnedRef)) return try { - val daemonSet = prepullDaemonSet(iid, pinnedRef) client .apps() .daemonSets() .inNamespace(namespace) - .resource(daemonSet) + .resource(prepullDaemonSet(iid, pinnedRef)) .createOr(existing => existing.update()) lastFailure.remove(iid) logger.info(s"Pre-pulling curated image $iid ($pinnedRef) onto every node.") } catch { case e: Throwable => lastFailure.put(iid, (pinnedRef, System.currentTimeMillis())) - // The image is still usable; only the head start is lost. logger.warn( s"Could not pre-pull curated image $iid ($pinnedRef). The first unit on each " + "node will wait for the pull instead.", @@ -120,11 +102,7 @@ object ImagePrepullClient extends LazyLogging { } } - /** - * Removes an image's pre-pull. Safe to call when there is none -- a deployment that had - * pre-pulling turned off has nothing to delete, and neither has an image that never - * reached READY. - */ + /** Removes an image's pre-pull. Safe to call when there is none. */ def deletePrepull(iid: Int): Unit = { try { client @@ -132,31 +110,23 @@ object ImagePrepullClient extends LazyLogging { .daemonSets() .inNamespace(namespace) .withName(CuratedImageConfig.prepullName(iid)) - // Stated rather than left to a default: were it ever Orphan, the pods would stay - // on every node still holding the image, and the reconcile pass lists DaemonSets, - // so nothing would ever find them again. + // Stated, not left to a default: orphaned pods would hold the image on every node, + // and the reconcile pass lists DaemonSets, so it would never find them. .withPropagationPolicy(DeletionPropagation.BACKGROUND) .delete() } catch { case e: Throwable => - // Left behind it would go on holding the image on every node, so it is worth - // naming in the log rather than passing over. logger.warn(s"Could not remove the pre-pull for curated image $iid.", e) } - // Whether or not the delete landed, nothing is now known to be wrong with this image's - // reference, and a create asked for next should not be held back by an old failure. lastFailure.remove(iid) } /** - * What each image's pre-pull currently pulls, keyed by image. Read so that a row which - * reached READY without one -- registered before this shipped, or finished while the - * cluster was unreachable -- is given theirs, and so that one still pointing at a - * superseded digest is corrected. + * What each image's pre-pull currently pulls, keyed by image. * - * None means the question could not be answered, which is not the same as "none exist": - * an empty map would have the caller create a pre-pull for every ready image against a - * cluster that has just refused to talk to it. + * None means the cluster could not be asked, which is not "none exist": an empty map + * would have the caller create a pre-pull for every ready image against a cluster that + * has just refused to talk to it. */ def prepulledRefs(): Option[Map[Int, String]] = { try { @@ -195,27 +165,19 @@ object ImagePrepullClient extends LazyLogging { private[service] def prepullDaemonSet(iid: Int, pinnedRef: String): DaemonSet = { val name = CuratedImageConfig.prepullName(iid) - val labels = Map( - "app" -> name, - OwnerLabel -> "true", - ImageLabel -> iid.toString - ).asJava + val labels = Map("app" -> name, OwnerLabel -> "true", ImageLabel -> iid.toString).asJava - // Requests on both containers, not just the one that keeps running. A namespace with a - // ResourceQuota on requests.cpu/requests.memory refuses a pod whose init container - // leaves them out -- quota admission checks init containers too -- and the refusal is - // invisible, because the DaemonSet is still created regardless. Costs nothing: a pod's - // request is the larger of its init containers and the sum of its others, so 1m/8Mi. + // Requests on both containers: a namespace with a ResourceQuota on requests.cpu or + // requests.memory refuses a pod whose init container omits them, and the refusal is + // invisible because the DaemonSet is still created. Costs nothing -- a pod's request + // is the larger of its init containers and the sum of the rest. val prepullerResources = new ResourceRequirementsBuilder() .addToRequests("cpu", new Quantity(CuratedImageConfig.prepullCpu)) .addToRequests("memory", new Quantity(CuratedImageConfig.prepullMemory)) .build() - // Limits only here. Unlike requests, a limit is enforced per container and never maxed - // across them, so 8Mi on the init container would cap the shell of an arbitrary image - // -- bash, on the Python bases these are built from -- and OOMKill it. The pod would - // then crash-loop, never reach pause, and the image it just pulled would go back to - // being reclaimable, all while the DaemonSet reported itself created. + // Limits only here. A limit is per container and never maxed, so capping the init + // container would OOMKill an arbitrary image's shell. val pauseResources = new ResourceRequirementsBuilder() .addToRequests("cpu", new Quantity(CuratedImageConfig.prepullCpu)) .addToRequests("memory", new Quantity(CuratedImageConfig.prepullMemory)) @@ -231,25 +193,29 @@ object ImagePrepullClient extends LazyLogging { .endMetadata() .withNewSpec() .withNewSelector() - // Only "app". A DaemonSet's selector cannot be changed after it is created, so it - // must not carry anything this code might later want to alter. + // "app" only: a selector cannot be changed once created. .withMatchLabels(Map("app" -> name).asJava) .endSelector() + // Every node at once. The default rolls one at a time, each waiting for a full pull, + // which is hours on a large cluster -- and there is no availability to protect here. + .withNewUpdateStrategy() + .withType("RollingUpdate") + .withNewRollingUpdate() + .withMaxUnavailable(new io.fabric8.kubernetes.api.model.IntOrString("100%")) + .endRollingUpdate() + .endUpdateStrategy() .withNewTemplate() .withNewMetadata() .withLabels(labels) .endMetadata() .withNewSpec() - // No tolerations, deliberately: a computing-unit pod declares none either, so a - // tainted node is one no unit can ever be scheduled onto. Tolerating everything - // would put multi-gigabyte images on control-plane and other reserved nodes that - // will never run a unit. + // No tolerations: computing-unit pods declare none, so a tainted node is one no unit + // can be scheduled onto. .withInitContainers( new io.fabric8.kubernetes.api.model.ContainerBuilder() .withName("prepuller") .withImage(pinnedRef) - // The reference names a digest, so what is already on the node cannot differ - // from what the registry would serve. Always would re-check it for nothing. + // The reference names a digest, so what is on the node cannot differ. .withImagePullPolicy("IfNotPresent") .withCommand("sh", "-c", "true") .withResources(prepullerResources) diff --git a/computing-unit-managing-service/src/test/scala/org/apache/texera/service/util/ImagePrepullClientSpec.scala b/computing-unit-managing-service/src/test/scala/org/apache/texera/service/util/ImagePrepullClientSpec.scala index af1ef9010d8..30a307e15b4 100644 --- a/computing-unit-managing-service/src/test/scala/org/apache/texera/service/util/ImagePrepullClientSpec.scala +++ b/computing-unit-managing-service/src/test/scala/org/apache/texera/service/util/ImagePrepullClientSpec.scala @@ -55,19 +55,22 @@ class ImagePrepullClientSpec extends AnyFlatSpec with Matchers { containers.head.getImage shouldBe CuratedImageConfig.prepullPauseImage } + // The default rolls one node at a time, each waiting for a full pull -- hours on a large + // cluster, for a pod with no availability to protect. + it should "repoint every node at once" in { + val rolling = prepullDaemonSet(7, PinnedRef).getSpec.getUpdateStrategy.getRollingUpdate + rolling.getMaxUnavailable.getStrVal shouldBe "100%" + } + + // A computing-unit pod declares no tolerations, so a tainted node is one no unit can + // land on. Tolerating everything put multi-gigabyte images on control-plane nodes. it should "schedule exactly where a computing unit can, and no wider" in { - // The regression this guards: an earlier version tolerated everything, which put - // multi-gigabyte images on control-plane and other reserved nodes. A computing-unit - // pod declares no tolerations, so a tainted node is one no unit can ever land on -- - // pre-pulling there buys nothing and costs disk on the nodes that can least spare it. val podSpec = prepullDaemonSet(7, PinnedRef).getSpec.getTemplate.getSpec Option(podSpec.getTolerations).map(_.asScala.toList).getOrElse(Nil) shouldBe Nil } - // The regression this guards: the init container declared no requests, and the pool - // namespace has a ResourceQuota on requests.cpu/requests.memory. Quota admission checks - // init containers, so every pre-pull pod was refused -- while the DaemonSet itself was - // created, so the service logged success and pre-pulled nothing, anywhere, ever. + // The pool namespace's ResourceQuota refuses a pod whose init container omits requests, + // while still creating the DaemonSet -- so this failed silently. it should "declare the requests a quota would demand" in { val podSpec = prepullDaemonSet(7, PinnedRef).getSpec.getTemplate.getSpec val everyContainer = @@ -80,11 +83,8 @@ class ImagePrepullClientSpec extends AnyFlatSpec with Matchers { } } - // The regression this guards: requests and limits were built once and shared, which put - // an 8Mi cap on the init container. A limit is enforced per container and never maxed - // across them, so the shell of an arbitrary image -- bash, on the Python bases these are - // built from -- was OOMKilled, the pod crash-looped, pause never ran, and the image went - // back to being reclaimable, all while the DaemonSet reported itself created. + // A limit is per container and never maxed, so a shared one capped the image's own + // shell at 8Mi and OOMKilled it. it should "cap the pause container only, never the image's own shell" in { val podSpec = prepullDaemonSet(7, PinnedRef).getSpec.getTemplate.getSpec @@ -95,9 +95,8 @@ class ImagePrepullClientSpec extends AnyFlatSpec with Matchers { pause.getResources.getLimits.asScala.keySet should contain allOf ("cpu", "memory") } - // The regression this guards: a persistent failure -- the Role not reapplied after an - // upgrade -- was retried on every read of the image list, by every user, for every ready - // image, logging a stack trace each time. + // Reconciling runs on every read of the image list, so a failure that will not clear + // was retried on every page load, for every ready image. "isCoolingDown" should "hold back a reference that just failed" in { val now = 1_000_000_000L val cooldownMillis = CuratedImageConfig.prepullRetryCooldownSeconds * 1000L @@ -120,10 +119,9 @@ class ImagePrepullClientSpec extends AnyFlatSpec with Matchers { ) shouldBe false } + // Otherwise a refresh that resolved a new digest would leave nodes on the old image + // until the cooldown expired. it should "not hold back a different digest" in { - // A refresh that resolved a new digest is a new question. Making it wait out a cooldown - // the previous reference earned would leave nodes on the superseded image for as long - // as the cooldown lasts. val now = 1_000_000_000L val other = "owner/name@sha256:" + "f" * 64 ImagePrepullClient.isCoolingDown(Some((PinnedRef, now)), other, now) shouldBe false @@ -133,9 +131,8 @@ class ImagePrepullClientSpec extends AnyFlatSpec with Matchers { ImagePrepullClient.isCoolingDown(None, PinnedRef, 1_000_000_000L) shouldBe false } + // How a stale pre-pull is spotted: comparing only ids would miss a failed repoint. "prepulledRefOf" should "read back what a pre-pull actually pulls" in { - // What lets a stale pre-pull be spotted: a refresh whose repoint failed leaves one - // holding the previous digest, and comparing only ids would never notice. ImagePrepullClient.prepulledRefOf(prepullDaemonSet(7, PinnedRef)).value shouldBe PinnedRef } @@ -144,19 +141,17 @@ class ImagePrepullClientSpec extends AnyFlatSpec with Matchers { ImagePrepullClient.prepulledRefOf(strayObject) shouldBe None } + // Same name at any digest, so a refresh does not leave a second pre-pull behind. it should "name itself after the image, so a refresh replaces rather than adds" in { - // Same name for the same image at any digest: a refreshed image must not leave a - // second pre-pull behind holding bytes nothing runs any more. prepullDaemonSet(7, PinnedRef).getMetadata.getName shouldBe "cu-image-prepull-7" prepullDaemonSet(7, "owner/name@sha256:" + "c" * 64).getMetadata.getName shouldBe "cu-image-prepull-7" prepullDaemonSet(8, PinnedRef).getMetadata.getName shouldBe "cu-image-prepull-8" } + // A selector is immutable once created, so anything mutable in it would make every + // later repoint fail. it should "select on a label it will never want to change" in { - // A DaemonSet's selector is immutable once created. Selecting on the image id too - // would be harmless, but selecting on anything mutable would make the update in - // ensurePrepull fail for good, so this pins the selector to "app" alone. val daemonSet = prepullDaemonSet(7, PinnedRef) daemonSet.getSpec.getSelector.getMatchLabels.asScala shouldBe Map("app" -> "cu-image-prepull-7") @@ -176,9 +171,8 @@ class ImagePrepullClientSpec extends AnyFlatSpec with Matchers { imageIdOf(prepullDaemonSet(7, PinnedRef)).value shouldBe 7 } + // A stray object must not be read as an image id. it should "ignore a DaemonSet that is not one of ours" in { - // prepulledRefs lists by label, but a cluster can hold anything. A stray object - // must not be read as an image id and leave a real image without its pre-pull. val unlabelled = new DaemonSetBuilder().withNewMetadata().withName("something").endMetadata() imageIdOf(unlabelled.build()) shouldBe None From 12d00c1d86f099b80b820b7e32afd94378a96f61 Mon Sep 17 00:00:00 2001 From: Tanishq Gandhi Date: Tue, 15 Sep 2026 17:40:32 +0000 Subject: [PATCH 7/7] fix(computing-unit): read the digest the check wrote, not one the image supplied The validation job prints the image's own start command before it prints the digest it resolved, and the digest was read from the first marker line in the log. An image whose Cmd carries a newline and a marker of its own therefore passed the check -- its Cmd still contains computing-unit-master -- while naming the digest every unit would then be pinned to. That defeats the point of pinning: the row is supposed to record the bytes the check approved. Closed at both ends. The start command is echoed through tr, so an image cannot put a line into the log at all, and the digest is read from the last marker line, which is the one the job writes last. --- .../service/util/ImageValidationClient.scala | 11 ++++++++--- .../resource/CuratedImageResourceSpec.scala | 17 +++++++++++++++++ 2 files changed, 25 insertions(+), 3 deletions(-) diff --git a/computing-unit-managing-service/src/main/scala/org/apache/texera/service/util/ImageValidationClient.scala b/computing-unit-managing-service/src/main/scala/org/apache/texera/service/util/ImageValidationClient.scala index ebe70bcf3e2..c7d45b1935d 100644 --- a/computing-unit-managing-service/src/main/scala/org/apache/texera/service/util/ImageValidationClient.scala +++ b/computing-unit-managing-service/src/main/scala/org/apache/texera/service/util/ImageValidationClient.scala @@ -125,12 +125,12 @@ object ImageValidationClient extends LazyLogging { | echo "$$START_CMD" | exit 1 |fi - |echo "Start command: $$START_CMD" + |echo "Start command: $$(printf '%s' "$$START_CMD" | tr '\n' ' ')" | |if ! echo "$$START_CMD" | grep -qF '${CuratedImageConfig.requiredCommand}'; then | echo "" | echo "ERROR: $$SOURCE_REF does not look like a Texera computing-unit image." - | echo "Its start command is: $$START_CMD" + | echo "Its start command is: $$(printf '%s' "$$START_CMD" | tr '\n' ' ')" | echo "A computing-unit image starts '${CuratedImageConfig.requiredCommand}'." | exit 1 |fi @@ -335,11 +335,16 @@ object ImageValidationClient extends LazyLogging { /** The digest the source tag resolved to, as printed by a successful job. */ def sourceDigestFrom(log: String): Option[String] = + // The job prints this as its very last line, so the last match is the one it wrote. + // Reading the first would let anything echoed earlier -- the image's own start + // command, which its author controls -- name the digest a unit is pinned to. log.linesIterator .map(_.trim) - .find(_.startsWith(DigestMarker)) + .filter(_.startsWith(DigestMarker)) .map(_.drop(DigestMarker.length).trim) .filter(_.nonEmpty) + .toSeq + .lastOption /** * The reference a unit starts from: the administrator's repository at the resolved diff --git a/computing-unit-managing-service/src/test/scala/org/apache/texera/service/resource/CuratedImageResourceSpec.scala b/computing-unit-managing-service/src/test/scala/org/apache/texera/service/resource/CuratedImageResourceSpec.scala index 4858a44b112..3508cd09d46 100644 --- a/computing-unit-managing-service/src/test/scala/org/apache/texera/service/resource/CuratedImageResourceSpec.scala +++ b/computing-unit-managing-service/src/test/scala/org/apache/texera/service/resource/CuratedImageResourceSpec.scala @@ -284,6 +284,23 @@ class CuratedImageResourceSpec extends AnyFlatSpec with Matchers { // Off until the UI ships, so a deployment that has not opted in starts no unit from a // curated image -- including from a row left behind if it was enabled and turned off. + // The regression this guards: the digest was read from the first marker line, while the + // image's own start command -- which its author controls -- is echoed earlier. An image + // whose Cmd carries a newline and a marker of its own could pass the check and still + // pin units to bytes nobody validated. + "sourceDigestFrom" should "ignore a marker the image smuggled into its start command" in { + val log = + """Inspecting owner/evil:1.0 + |Pinned to: owner/evil@sha256:1111111111111111111111111111111111111111111111111111111111111111 + |Start command: [computing-unit-master + |TEXERA_SOURCE_DIGEST=sha256:2222222222222222222222222222222222222222222222222222222222222222] + |Runs as: texera + |TEXERA_SOURCE_DIGEST=sha256:1111111111111111111111111111111111111111111111111111111111111111 + |""".stripMargin + ImageValidationClient.sourceDigestFrom(log).value shouldBe + "sha256:1111111111111111111111111111111111111111111111111111111111111111" + } + "the feature flag" should "be off unless a deployment turns it on" in { CuratedImageConfig.enabled shouldBe false }