diff --git a/spark-operator/src/main/java/org/apache/spark/k8s/operator/context/BaseContext.java b/spark-operator/src/main/java/org/apache/spark/k8s/operator/context/BaseContext.java new file mode 100644 index 00000000..98bddb5a --- /dev/null +++ b/spark-operator/src/main/java/org/apache/spark/k8s/operator/context/BaseContext.java @@ -0,0 +1,30 @@ +/* + * 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.spark.k8s.operator.context; + +import io.fabric8.kubernetes.client.KubernetesClient; + +import org.apache.spark.k8s.operator.BaseResource; + +public abstract class BaseContext> { + public abstract CR getResource(); + + public abstract KubernetesClient getClient(); +} diff --git a/spark-operator/src/main/java/org/apache/spark/k8s/operator/context/SparkAppContext.java b/spark-operator/src/main/java/org/apache/spark/k8s/operator/context/SparkAppContext.java new file mode 100644 index 00000000..7dba5a2c --- /dev/null +++ b/spark-operator/src/main/java/org/apache/spark/k8s/operator/context/SparkAppContext.java @@ -0,0 +1,112 @@ +/* + * 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.spark.k8s.operator.context; + +import static org.apache.spark.k8s.operator.utils.Utils.driverLabels; +import static org.apache.spark.k8s.operator.utils.Utils.executorLabels; + +import java.util.List; +import java.util.Optional; +import java.util.Set; +import java.util.stream.Collectors; + +import io.fabric8.kubernetes.api.model.HasMetadata; +import io.fabric8.kubernetes.api.model.Pod; +import io.fabric8.kubernetes.client.KubernetesClient; +import io.javaoperatorsdk.operator.api.reconciler.Context; +import lombok.RequiredArgsConstructor; +import lombok.extern.slf4j.Slf4j; + +import org.apache.spark.k8s.operator.SparkAppResourceSpec; +import org.apache.spark.k8s.operator.SparkAppSubmissionWorker; +import org.apache.spark.k8s.operator.SparkApplication; +import org.apache.spark.k8s.operator.reconciler.SparkAppResourceSpecFactory; + +/** + * Context for {@link org.apache.spark.k8s.operator.SparkApplication} resource, including secondary + * resource(s) and desired secondary resource spec + */ +@RequiredArgsConstructor +@Slf4j +public class SparkAppContext extends BaseContext { + private final SparkApplication sparkApplication; + private final Context josdkContext; + private final SparkAppSubmissionWorker submissionWorker; + + /** secondaryResourceSpec is initialized in a lazy fashion - built upon the first attempt */ + private SparkAppResourceSpec secondaryResourceSpec; + + public Optional getDriverPod() { + return josdkContext + .getSecondaryResourcesAsStream(Pod.class) + .filter( + p -> + p.getMetadata() + .getLabels() + .entrySet() + .containsAll(driverLabels(sparkApplication).entrySet())) + .findAny(); + } + + public Set getExecutorsForApplication() { + return josdkContext + .getSecondaryResourcesAsStream(Pod.class) + .filter( + p -> + p.getMetadata() + .getLabels() + .entrySet() + .containsAll(executorLabels(sparkApplication).entrySet())) + .collect(Collectors.toSet()); + } + + private SparkAppResourceSpec getSecondaryResourceSpec() { + synchronized (this) { + if (secondaryResourceSpec == null) { + secondaryResourceSpec = + SparkAppResourceSpecFactory.buildResourceSpec( + sparkApplication, josdkContext.getClient(), submissionWorker); + } + return secondaryResourceSpec; + } + } + + @Override + public SparkApplication getResource() { + return sparkApplication; + } + + @Override + public KubernetesClient getClient() { + return josdkContext.getClient(); + } + + public List getDriverPreResourcesSpec() { + return getSecondaryResourceSpec().getDriverPreResources(); + } + + public Pod getDriverPodSpec() { + return getSecondaryResourceSpec().getConfiguredPod(); + } + + public List getDriverResourcesSpec() { + return getSecondaryResourceSpec().getDriverResources(); + } +} diff --git a/spark-operator/src/main/java/org/apache/spark/k8s/operator/decorators/DriverDecorator.java b/spark-operator/src/main/java/org/apache/spark/k8s/operator/decorators/DriverDecorator.java new file mode 100644 index 00000000..7cab1f56 --- /dev/null +++ b/spark-operator/src/main/java/org/apache/spark/k8s/operator/decorators/DriverDecorator.java @@ -0,0 +1,53 @@ +/* + * 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.spark.k8s.operator.decorators; + +import static org.apache.spark.k8s.operator.utils.ModelUtils.buildOwnerReferenceTo; +import static org.apache.spark.k8s.operator.utils.Utils.sparkAppResourceLabels; + +import io.fabric8.kubernetes.api.model.HasMetadata; +import io.fabric8.kubernetes.api.model.ObjectMeta; +import io.fabric8.kubernetes.api.model.ObjectMetaBuilder; +import lombok.RequiredArgsConstructor; + +import org.apache.spark.k8s.operator.SparkApplication; + +/** + * Decorates driver (pod) to make sure its metadata matches event source Also adds owner reference + * to the owner SparkApplication for garbage collection + */ +@RequiredArgsConstructor +public class DriverDecorator implements ResourceDecorator { + + private final SparkApplication app; + + /** Add labels and owner references to the app for all secondary resources */ + @Override + public T decorate(T resource) { + ObjectMeta metaData = + new ObjectMetaBuilder(resource.getMetadata()) + .addToOwnerReferences(buildOwnerReferenceTo(app)) + .addToLabels(sparkAppResourceLabels(app)) + .withNamespace(app.getMetadata().getNamespace()) + .build(); + resource.setMetadata(metaData); + return resource; + } +} diff --git a/spark-operator/src/main/java/org/apache/spark/k8s/operator/decorators/DriverResourceDecorator.java b/spark-operator/src/main/java/org/apache/spark/k8s/operator/decorators/DriverResourceDecorator.java new file mode 100644 index 00000000..95a61554 --- /dev/null +++ b/spark-operator/src/main/java/org/apache/spark/k8s/operator/decorators/DriverResourceDecorator.java @@ -0,0 +1,73 @@ +/* + * 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.spark.k8s.operator.decorators; + +import static org.apache.spark.k8s.operator.utils.ModelUtils.buildOwnerReferenceTo; + +import io.fabric8.kubernetes.api.model.HasMetadata; +import io.fabric8.kubernetes.api.model.ObjectMeta; +import io.fabric8.kubernetes.api.model.ObjectMetaBuilder; +import io.fabric8.kubernetes.api.model.OwnerReference; +import io.fabric8.kubernetes.api.model.Pod; +import lombok.RequiredArgsConstructor; +import lombok.extern.slf4j.Slf4j; + +/** + * Decorates Driver resources (except the pod). This makes sure all resources have owner reference + * to the driver pod, so they can be garbage collected upon termination. Secondary resources would + * be garbage-collected if ALL owners are deleted. Therefore, operator makes only driver pod has + * owned by the SparkApplication while all other secondary resources are owned by the driver. In + * this way, after driver pod is deleted at the end of each attempt, all other resources would be + * garbage collected automatically. If given secondary resource already has owner reference to + * additional resources, it's reference to driver pod would be added to the list. Note - this is + * uncommon as additional owner reference might impact the garbage collection at the end of each + * attempt + */ +@RequiredArgsConstructor +@Slf4j +public class DriverResourceDecorator implements ResourceDecorator { + private final Pod driverPod; + + @Override + public T decorate(T resource) { + boolean ownerReferenceExists = false; + if (resource.getMetadata().getOwnerReferences() != null + && !resource.getMetadata().getOwnerReferences().isEmpty()) { + for (OwnerReference o : resource.getMetadata().getOwnerReferences()) { + if (driverPod.getKind().equals(o.getKind()) + && driverPod.getMetadata().getName().equals(o.getName()) + && driverPod.getMetadata().getUid().equals(o.getUid())) { + ownerReferenceExists = true; + break; + } + } + } + if (!ownerReferenceExists) { + log.debug("Adding OwnerReference to driver for secondary resource"); + ObjectMeta metaData = + new ObjectMetaBuilder(resource.getMetadata()) + .addToOwnerReferences(buildOwnerReferenceTo(driverPod)) + .addToLabels(driverPod.getMetadata().getLabels()) + .build(); + resource.setMetadata(metaData); + } + return resource; + } +} diff --git a/spark-operator/src/main/java/org/apache/spark/k8s/operator/metrics/healthcheck/SentinelManager.java b/spark-operator/src/main/java/org/apache/spark/k8s/operator/metrics/healthcheck/SentinelManager.java new file mode 100644 index 00000000..108af26d --- /dev/null +++ b/spark-operator/src/main/java/org/apache/spark/k8s/operator/metrics/healthcheck/SentinelManager.java @@ -0,0 +1,216 @@ +/* + * 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.spark.k8s.operator.metrics.healthcheck; + +import java.util.HashSet; +import java.util.Map; +import java.util.Set; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.Executors; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.TimeUnit; + +import com.google.common.annotations.VisibleForTesting; +import io.fabric8.kubernetes.api.model.HasMetadata; +import io.fabric8.kubernetes.client.KubernetesClient; +import io.javaoperatorsdk.operator.processing.event.ResourceID; +import lombok.Getter; +import lombok.RequiredArgsConstructor; +import lombok.extern.slf4j.Slf4j; +import org.apache.commons.lang3.builder.ToStringBuilder; + +import org.apache.spark.k8s.operator.BaseResource; +import org.apache.spark.k8s.operator.Constants; +import org.apache.spark.k8s.operator.config.SparkOperatorConf; +import org.apache.spark.k8s.operator.utils.ReconcilerUtils; +import org.apache.spark.k8s.operator.utils.Utils; + +/** + * Sentinel manager monitors dedicated sentinel resources to make sure the operator is healthy + * + * @param custom resource type + */ +@RequiredArgsConstructor +@Slf4j +public class SentinelManager> { + + private final ConcurrentHashMap sentinelResources = + new ConcurrentHashMap<>(); + + private final ScheduledExecutorService executorService = + Executors.newScheduledThreadPool( + SparkOperatorConf.SENTINEL_EXECUTOR_SERVICE_POOL_SIZE.getValue()); + + public static boolean isSentinelResource(HasMetadata resource) { + Map labels = resource.getMetadata().getLabels(); + if (labels == null) { + return false; + } + String namespace = resource.getMetadata().getNamespace(); + return shouldSentinelWatchGivenNamespace(namespace) + && Boolean.TRUE + .toString() + .equalsIgnoreCase( + labels.getOrDefault(Constants.LABEL_SENTINEL_RESOURCE, Boolean.FALSE.toString())); + } + + private static boolean shouldSentinelWatchGivenNamespace(String namespace) { + if ((!Utils.getWatchedNamespaces().isEmpty()) + && !Utils.getWatchedNamespaces().contains(namespace)) { + if (log.isErrorEnabled()) { + log.error("Skip watching sentinel resource in namespace {}", namespace); + } + return false; + } + return true; + } + + public boolean allSentinelsAreHealthy() { + Set unWatchedKey = new HashSet<>(); + boolean result = + sentinelResources.entrySet().stream() + .filter( + x -> { + if (x.getKey().getNamespace().isPresent() + && shouldSentinelWatchGivenNamespace(x.getKey().getNamespace().get())) { + return true; + } + unWatchedKey.add(x.getKey()); + return false; + }) + .map(Map.Entry::getValue) + .allMatch(SentinelResourceState::isHealthy); + sentinelResources.keySet().removeAll(unWatchedKey); + return result; + } + + public void checkHealth(ResourceID resourceID, KubernetesClient client) { + SentinelResourceState sentinelResourceState = sentinelResources.get(resourceID); + if (sentinelResourceState == null) { + if (log.isErrorEnabled()) { + log.error("Sentinel resources {} not found. Stopping sentinel health checks", resourceID); + } + return; + } + + if (sentinelResourceState.reconciledSinceUpdate()) { + log.info("Sentinel reports healthy state globally"); + sentinelResourceState.isHealthy = true; + } else { + if (log.isErrorEnabled()) { + log.error( + "Sentinel deployment {} latest spec not was reconciled. Expected " + + "generation larger than {}, received {}", + resourceID, + sentinelResourceState.previousGeneration, + sentinelResourceState.resource.getMetadata().getGeneration()); + } + sentinelResourceState.isHealthy = false; + } + + updateSpecAndScheduleHealthCheck(resourceID, sentinelResourceState, client); + } + + public boolean handleSentinelResourceReconciliation(CR resource, KubernetesClient client) { + if (!isSentinelResource(resource)) { + return false; + } + + ResourceID resourceId = ResourceID.fromResource(resource); + sentinelResources.compute( + resourceId, + (id, previousState) -> { + boolean firstReconcile = false; + if (previousState == null) { + firstReconcile = true; + previousState = new SentinelResourceState(); + } + previousState.onReconcile(resource); + if (firstReconcile) { + updateSpecAndScheduleHealthCheck(resourceId, previousState, client); + } + return previousState; + }); + return true; + } + + private void updateSpecAndScheduleHealthCheck( + ResourceID resourceID, SentinelResourceState sentinelResourceState, KubernetesClient client) { + Map sparkConf = sentinelResourceState.resource.getSpec().getSparkConf(); + sparkConf.compute( + Constants.SENTINEL_RESOURCE_DUMMY_FIELD, + (key, value) -> { + if (value == null) { + return "1"; + } else { + return String.valueOf(Long.parseLong(value) + 1); + } + }); + sentinelResourceState.previousGeneration = + sentinelResourceState.resource.getMetadata().getGeneration(); + try { + if (log.isDebugEnabled()) { + log.debug("Update the sentinel kubernetes resource spec {}", sentinelResourceState); + } + client.resource(ReconcilerUtils.clone(sentinelResourceState.resource)).replace(); + } catch (Throwable t) { + if (log.isWarnEnabled()) { + log.warn( + "Could not replace the sentinel deployment spark conf {}", + Constants.SENTINEL_RESOURCE_DUMMY_FIELD, + t); + } + } + int delay = SparkOperatorConf.SENTINEL_RESOURCE_RECONCILIATION_DELAY.getValue(); + if (log.isInfoEnabled()) { + log.info("Scheduling sentinel check for {} in {} seconds", resourceID, delay); + } + executorService.schedule(() -> checkHealth(resourceID, client), delay, TimeUnit.SECONDS); + } + + public class SentinelResourceState { + CR resource; + long previousGeneration; + + @Getter boolean isHealthy = true; + + void onReconcile(CR cr) { + resource = cr; + } + + boolean reconciledSinceUpdate() { + return resource.getMetadata().getGeneration() > previousGeneration; + } + + @Override + public String toString() { + return new ToStringBuilder(this) + .append("resource", resource) + .append("previousGeneration", previousGeneration) + .append("isHealthy", isHealthy) + .toString(); + } + } + + @VisibleForTesting + public ConcurrentHashMap getSentinelResources() { + return sentinelResources; + } +} diff --git a/spark-operator/src/main/java/org/apache/spark/k8s/operator/probe/HealthProbe.java b/spark-operator/src/main/java/org/apache/spark/k8s/operator/probe/HealthProbe.java new file mode 100644 index 00000000..6e03e0ea --- /dev/null +++ b/spark-operator/src/main/java/org/apache/spark/k8s/operator/probe/HealthProbe.java @@ -0,0 +1,113 @@ +/* + * 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.spark.k8s.operator.probe; + +import static org.apache.spark.k8s.operator.utils.ProbeUtil.areOperatorsStarted; +import static org.apache.spark.k8s.operator.utils.ProbeUtil.sendMessage; + +import java.io.IOException; +import java.util.ArrayList; +import java.util.List; +import java.util.Map; +import java.util.Optional; + +import com.sun.net.httpserver.HttpExchange; +import com.sun.net.httpserver.HttpHandler; +import io.javaoperatorsdk.operator.Operator; +import io.javaoperatorsdk.operator.RuntimeInfo; +import io.javaoperatorsdk.operator.health.InformerHealthIndicator; +import io.javaoperatorsdk.operator.health.InformerWrappingEventSourceHealthIndicator; +import io.javaoperatorsdk.operator.health.Status; +import lombok.Getter; +import lombok.RequiredArgsConstructor; +import lombok.extern.slf4j.Slf4j; + +import org.apache.spark.k8s.operator.metrics.healthcheck.SentinelManager; + +@Getter +@Slf4j +@RequiredArgsConstructor +public class HealthProbe implements HttpHandler { + private final List operators; + private final List> sentinelManagers; + + public boolean isHealthy() { + Optional operatorsAreReady = areOperatorsStarted(operators); + if (operatorsAreReady.isEmpty() || !operatorsAreReady.get()) { + return false; + } + + Optional runtimeInfosAreHealthy = + operators.stream() + .map(operator -> checkInformersHealth(operator.getRuntimeInfo())) + .reduce((a, b) -> a && b); + + if (runtimeInfosAreHealthy.isEmpty() || !runtimeInfosAreHealthy.get()) { + return false; + } + + for (SentinelManager sentinelManager : sentinelManagers) { + if (!sentinelManager.allSentinelsAreHealthy()) { + log.error("One sentinel manager {} reported an unhealthy condition.", sentinelManager); + return false; + } + } + + return true; + } + + @Override + public void handle(HttpExchange exchange) throws IOException { + if (isHealthy()) { + sendMessage(exchange, 200, "healthy"); + } else { + sendMessage(exchange, 500, "unhealthy"); + } + } + + private boolean checkInformersHealth(RuntimeInfo operatorRuntimeInfo) { + log.debug("Checking informer health"); + List informersHealthList = new ArrayList<>(); + for (Map.Entry> + controllerEntry : + operatorRuntimeInfo.unhealthyInformerWrappingEventSourceHealthIndicator().entrySet()) { + for (Map.Entry eventSourceEntry : + controllerEntry.getValue().entrySet()) { + Map informers = + eventSourceEntry.getValue().informerHealthIndicators(); + for (Map.Entry informerEntry : informers.entrySet()) { + if (informerEntry.getValue().getStatus() == Status.HEALTHY) { + informersHealthList.add(true); + } else { + if (log.isErrorEnabled()) { + log.error( + "Controller: {}, Event Source: {}, Informer: {} is not in a healthy state", + controllerEntry.getKey(), + eventSourceEntry.getKey(), + informerEntry.getKey()); + } + informersHealthList.add(false); + } + } + } + } + return informersHealthList.stream().reduce((a, b) -> a && b).orElse(true); + } +} diff --git a/spark-operator/src/main/java/org/apache/spark/k8s/operator/probe/ProbeService.java b/spark-operator/src/main/java/org/apache/spark/k8s/operator/probe/ProbeService.java new file mode 100644 index 00000000..48159f5b --- /dev/null +++ b/spark-operator/src/main/java/org/apache/spark/k8s/operator/probe/ProbeService.java @@ -0,0 +1,64 @@ +/* + * 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.spark.k8s.operator.probe; + +import static org.apache.spark.k8s.operator.config.SparkOperatorConf.OPERATOR_PROBE_PORT; + +import java.io.IOException; +import java.net.InetSocketAddress; +import java.util.List; +import java.util.concurrent.Executor; + +import com.sun.net.httpserver.HttpServer; +import io.javaoperatorsdk.operator.Operator; +import lombok.Getter; +import lombok.extern.slf4j.Slf4j; + +import org.apache.spark.k8s.operator.metrics.healthcheck.SentinelManager; + +@Slf4j +public class ProbeService { + public static final String HEALTHZ = "/healthz"; + public static final String READYZ = "/readyz"; + @Getter private final HttpServer server; + + public ProbeService( + List operators, List> sentinelManagers, Executor executor) { + HealthProbe healthProbe = new HealthProbe(operators, sentinelManagers); + try { + this.server = HttpServer.create(new InetSocketAddress(OPERATOR_PROBE_PORT.getValue()), 0); + } catch (IOException e) { + throw new RuntimeException("Failed to create Probe Service Server", e); + } + server.createContext(READYZ, new ReadinessProbe(operators)); + server.createContext(HEALTHZ, healthProbe); + server.setExecutor(executor); + } + + public void start() { + log.info("Probe service started"); + server.start(); + } + + public void stop() { + log.info("Probe service stopped"); + server.stop(0); + } +} diff --git a/spark-operator/src/main/java/org/apache/spark/k8s/operator/probe/ReadinessProbe.java b/spark-operator/src/main/java/org/apache/spark/k8s/operator/probe/ReadinessProbe.java new file mode 100644 index 00000000..be08c363 --- /dev/null +++ b/spark-operator/src/main/java/org/apache/spark/k8s/operator/probe/ReadinessProbe.java @@ -0,0 +1,59 @@ +/* + * 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.spark.k8s.operator.probe; + +import static org.apache.spark.k8s.operator.utils.ProbeUtil.areOperatorsStarted; +import static org.apache.spark.k8s.operator.utils.ProbeUtil.sendMessage; + +import java.io.IOException; +import java.util.List; +import java.util.Optional; + +import com.sun.net.httpserver.HttpExchange; +import com.sun.net.httpserver.HttpHandler; +import io.javaoperatorsdk.operator.Operator; +import lombok.extern.slf4j.Slf4j; + +@Slf4j +public class ReadinessProbe implements HttpHandler { + private final List operators; + + public ReadinessProbe(List operators) { + this.operators = operators; + } + + @Override + public void handle(HttpExchange httpExchange) throws IOException { + Optional operatorsAreReady = areOperatorsStarted(operators); + if (operatorsAreReady.isEmpty() || !operatorsAreReady.get()) { + sendMessage(httpExchange, 400, "spark operators are not ready yet"); + } + + if (!passRbacCheck()) { + sendMessage(httpExchange, 403, "required rbac test failed, operators are not ready"); + } + + sendMessage(httpExchange, 200, "started"); + } + + public boolean passRbacCheck() { + return true; + } +} diff --git a/spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/ReconcileProgress.java b/spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/ReconcileProgress.java new file mode 100644 index 00000000..d59bf4b5 --- /dev/null +++ b/spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/ReconcileProgress.java @@ -0,0 +1,69 @@ +/* + * 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.spark.k8s.operator.reconciler; + +import static org.apache.spark.k8s.operator.config.SparkOperatorConf.RECONCILER_INTERVAL_SECONDS; + +import java.time.Duration; + +import lombok.Data; + +/** + * Represents the progress of a reconcile request. + * + *

completed is set to true if there's no more actions expected in the same reconciliation. + * + *

requeue : describes whether the mentioned resource need to be reconciled again - and if so, + * the frequency + */ +@Data +public class ReconcileProgress { + private boolean completed; + boolean requeue; + private Duration requeueAfterDuration; + + private ReconcileProgress(boolean completed, boolean requeue, Duration requeueAfterDuration) { + this.completed = completed; + this.requeue = requeue; + this.requeueAfterDuration = requeueAfterDuration; + } + + public static ReconcileProgress proceed() { + return new ReconcileProgress( + false, true, Duration.ofSeconds(RECONCILER_INTERVAL_SECONDS.getValue())); + } + + public static ReconcileProgress completeAndDefaultRequeue() { + return new ReconcileProgress( + true, true, Duration.ofSeconds(RECONCILER_INTERVAL_SECONDS.getValue())); + } + + public static ReconcileProgress completeAndRequeueAfter(Duration requeueAfterDuration) { + return new ReconcileProgress(true, true, requeueAfterDuration); + } + + public static ReconcileProgress completeAndImmediateRequeue() { + return new ReconcileProgress(true, true, Duration.ZERO); + } + + public static ReconcileProgress completeAndNoRequeue() { + return new ReconcileProgress(true, false, Duration.ZERO); + } +} diff --git a/spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/SparkAppReconciler.java b/spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/SparkAppReconciler.java new file mode 100644 index 00000000..0be8a6aa --- /dev/null +++ b/spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/SparkAppReconciler.java @@ -0,0 +1,234 @@ +/* + * 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.spark.k8s.operator.reconciler; + +import static org.apache.spark.k8s.operator.reconciler.ReconcileProgress.completeAndDefaultRequeue; +import static org.apache.spark.k8s.operator.utils.Utils.commonResourceLabelsStr; + +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; +import java.util.Map; + +import io.fabric8.kubernetes.api.model.Pod; +import io.javaoperatorsdk.operator.api.config.informer.InformerConfiguration; +import io.javaoperatorsdk.operator.api.reconciler.Cleaner; +import io.javaoperatorsdk.operator.api.reconciler.Context; +import io.javaoperatorsdk.operator.api.reconciler.ControllerConfiguration; +import io.javaoperatorsdk.operator.api.reconciler.DeleteControl; +import io.javaoperatorsdk.operator.api.reconciler.ErrorStatusHandler; +import io.javaoperatorsdk.operator.api.reconciler.ErrorStatusUpdateControl; +import io.javaoperatorsdk.operator.api.reconciler.EventSourceContext; +import io.javaoperatorsdk.operator.api.reconciler.EventSourceInitializer; +import io.javaoperatorsdk.operator.api.reconciler.Reconciler; +import io.javaoperatorsdk.operator.api.reconciler.UpdateControl; +import io.javaoperatorsdk.operator.processing.event.source.EventSource; +import io.javaoperatorsdk.operator.processing.event.source.informer.InformerEventSource; +import io.javaoperatorsdk.operator.processing.event.source.informer.Mappers; +import lombok.RequiredArgsConstructor; +import lombok.extern.slf4j.Slf4j; + +import org.apache.spark.k8s.operator.Constants; +import org.apache.spark.k8s.operator.SparkAppSubmissionWorker; +import org.apache.spark.k8s.operator.SparkApplication; +import org.apache.spark.k8s.operator.context.SparkAppContext; +import org.apache.spark.k8s.operator.metrics.healthcheck.SentinelManager; +import org.apache.spark.k8s.operator.reconciler.observers.AppDriverReadyObserver; +import org.apache.spark.k8s.operator.reconciler.observers.AppDriverRunningObserver; +import org.apache.spark.k8s.operator.reconciler.observers.AppDriverStartObserver; +import org.apache.spark.k8s.operator.reconciler.observers.AppDriverTimeoutObserver; +import org.apache.spark.k8s.operator.reconciler.reconcilesteps.AppCleanUpStep; +import org.apache.spark.k8s.operator.reconciler.reconcilesteps.AppInitStep; +import org.apache.spark.k8s.operator.reconciler.reconcilesteps.AppReconcileStep; +import org.apache.spark.k8s.operator.reconciler.reconcilesteps.AppResourceObserveStep; +import org.apache.spark.k8s.operator.reconciler.reconcilesteps.AppRunningStep; +import org.apache.spark.k8s.operator.reconciler.reconcilesteps.AppTerminatedStep; +import org.apache.spark.k8s.operator.reconciler.reconcilesteps.AppValidateStep; +import org.apache.spark.k8s.operator.reconciler.reconcilesteps.UnknownStateStep; +import org.apache.spark.k8s.operator.utils.LoggingUtils; +import org.apache.spark.k8s.operator.utils.ReconcilerUtils; +import org.apache.spark.k8s.operator.utils.SparkAppStatusRecorder; +import org.apache.spark.k8s.operator.utils.SparkAppStatusUtils; + +/** + * Reconciler for Spark Application. Performs sanity check on the app, identify the reconcile steps + * based on App status and execute the steps. + */ +@ControllerConfiguration +@Slf4j +@RequiredArgsConstructor +public class SparkAppReconciler + implements Reconciler, + ErrorStatusHandler, + EventSourceInitializer, + Cleaner { + private final SparkAppSubmissionWorker submissionWorker; + private final SparkAppStatusRecorder sparkAppStatusRecorder; + private final SentinelManager sentinelManager; + + @Override + public UpdateControl reconcile( + SparkApplication sparkApplication, Context context) throws Exception { + LoggingUtils.TrackedMDC trackedMDC = new LoggingUtils.TrackedMDC(); + try { + trackedMDC.set(sparkApplication); + if (sentinelManager.handleSentinelResourceReconciliation( + sparkApplication, context.getClient())) { + return UpdateControl.noUpdate(); + } + log.debug("Start reconciliation."); + sparkAppStatusRecorder.updateStatusFromCache(sparkApplication); + SparkAppContext ctx = new SparkAppContext(sparkApplication, context, submissionWorker); + List reconcileSteps = getReconcileSteps(sparkApplication); + for (AppReconcileStep step : reconcileSteps) { + ReconcileProgress progress = step.reconcile(ctx, sparkAppStatusRecorder); + if (progress.isCompleted()) { + return ReconcilerUtils.toUpdateControl(sparkApplication, progress); + } + } + return ReconcilerUtils.toUpdateControl(sparkApplication, completeAndDefaultRequeue()); + + } finally { + log.debug("Reconciliation completed."); + trackedMDC.reset(); + } + } + + @Override + public ErrorStatusUpdateControl updateErrorStatus( + SparkApplication sparkApplication, Context context, Exception e) { + LoggingUtils.TrackedMDC trackedMDC = new LoggingUtils.TrackedMDC(); + try { + trackedMDC.set(sparkApplication); + context + .getRetryInfo() + .ifPresent( + retryInfo -> { + if (log.isErrorEnabled()) { + log.error( + "Failed attempt: {}, last attempt: {}", + retryInfo.getAttemptCount(), + retryInfo.isLastAttempt()); + } + }); + return ErrorStatusUpdateControl.noStatusUpdate(); + } finally { + trackedMDC.reset(); + } + } + + @Override + public Map prepareEventSources( + EventSourceContext context) { + EventSource podEventSource = + new InformerEventSource<>( + InformerConfiguration.from(Pod.class, context) + .withSecondaryToPrimaryMapper( + Mappers.fromLabel(Constants.LABEL_SPARK_APPLICATION_NAME)) + .withLabelSelector(commonResourceLabelsStr()) + .build(), + context); + return EventSourceInitializer.nameEventSources(podEventSource); + } + + protected List getReconcileSteps(final SparkApplication app) { + List steps = new ArrayList<>(); + steps.add(new AppValidateStep()); + steps.add(new AppTerminatedStep()); + switch (app.getStatus().getCurrentState().getCurrentStateSummary()) { + case Submitted: + case ScheduledToRestart: + steps.add(new AppInitStep()); + break; + case DriverRequested: + case DriverStarted: + steps.add( + new AppResourceObserveStep( + List.of(new AppDriverStartObserver(), new AppDriverReadyObserver()))); + steps.add( + new AppResourceObserveStep(Collections.singletonList(new AppDriverRunningObserver()))); + steps.add( + new AppResourceObserveStep(Collections.singletonList(new AppDriverTimeoutObserver()))); + break; + case DriverReady: + case InitializedBelowThresholdExecutors: + case RunningHealthy: + case RunningWithBelowThresholdExecutors: + steps.add(new AppRunningStep()); + steps.add( + new AppResourceObserveStep(Collections.singletonList(new AppDriverRunningObserver()))); + steps.add( + new AppResourceObserveStep(Collections.singletonList(new AppDriverTimeoutObserver()))); + break; + case DriverReadyTimedOut: + case DriverStartTimedOut: + case ExecutorsStartTimedOut: + case Succeeded: + case DriverEvicted: + case Failed: + case SchedulingFailure: + steps.add(new AppCleanUpStep()); + break; + default: + steps.add(new UnknownStateStep()); + break; + } + return steps; + } + + /** + * Best-effort graceful termination upon delete. + * + * @param sparkApplication the resource that is marked for deletion + * @param context the context with which the operation is executed + * @return DeleteControl, with requeue if needed + */ + @Override + public DeleteControl cleanup( + SparkApplication sparkApplication, Context context) { + LoggingUtils.TrackedMDC trackedMDC = new LoggingUtils.TrackedMDC(); + DeleteControl deleteControl = DeleteControl.defaultDelete(); + try { + trackedMDC.set(sparkApplication); + log.info("Cleaning up resources for SparkApp."); + SparkAppContext ctx = new SparkAppContext(sparkApplication, context, submissionWorker); + List cleanupSteps = new ArrayList<>(); + cleanupSteps.add(new AppValidateStep()); + cleanupSteps.add(new AppTerminatedStep()); + cleanupSteps.add(new AppCleanUpStep(SparkAppStatusUtils::appCancelled)); + for (AppReconcileStep step : cleanupSteps) { + ReconcileProgress progress = step.reconcile(ctx, sparkAppStatusRecorder); + if (progress.isCompleted()) { + if (progress.isRequeue()) { + return DeleteControl.noFinalizerRemoval() + .rescheduleAfter(progress.getRequeueAfterDuration()); + } else { + break; + } + } + } + } finally { + log.info("Cleanup completed"); + trackedMDC.reset(); + } + sparkAppStatusRecorder.removeCachedStatus(sparkApplication); + return deleteControl; + } +} diff --git a/spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/SparkAppResourceSpecFactory.java b/spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/SparkAppResourceSpecFactory.java new file mode 100644 index 00000000..59c78838 --- /dev/null +++ b/spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/SparkAppResourceSpecFactory.java @@ -0,0 +1,175 @@ +/* + * 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.spark.k8s.operator.reconciler; + +import static org.apache.spark.k8s.operator.Constants.DRIVER_SPARK_TEMPLATE_FILE_PROP_KEY; +import static org.apache.spark.k8s.operator.Constants.EXECUTOR_SPARK_TEMPLATE_FILE_PROP_KEY; +import static org.apache.spark.k8s.operator.utils.ModelUtils.overrideDriverTemplateEnabled; +import static org.apache.spark.k8s.operator.utils.ModelUtils.overrideExecutorTemplateEnabled; +import static org.apache.spark.k8s.operator.utils.Utils.sparkAppResourceLabels; + +import java.io.File; +import java.io.FileOutputStream; +import java.io.IOException; +import java.io.OutputStreamWriter; +import java.util.Collections; +import java.util.HashMap; +import java.util.Map; +import java.util.Optional; + +import io.fabric8.kubernetes.api.model.PodTemplateSpec; +import io.fabric8.kubernetes.client.KubernetesClient; +import lombok.extern.slf4j.Slf4j; + +import org.apache.spark.k8s.operator.SparkAppResourceSpec; +import org.apache.spark.k8s.operator.SparkAppSubmissionWorker; +import org.apache.spark.k8s.operator.SparkApplication; +import org.apache.spark.k8s.operator.decorators.DriverDecorator; +import org.apache.spark.k8s.operator.utils.ModelUtils; + +@Slf4j +public class SparkAppResourceSpecFactory { + public static SparkAppResourceSpec buildResourceSpec( + final SparkApplication app, + final KubernetesClient client, + final SparkAppSubmissionWorker worker) { + Map confOverrides = overrideDependencyConf(app); + SparkAppResourceSpec resourceSpec = worker.getResourceSpec(app, client, confOverrides); + cleanUpTempResourcesForApp(app, confOverrides); + DriverDecorator decorator = new DriverDecorator(app); + decorator.decorate(resourceSpec.getConfiguredPod()); + return resourceSpec; + } + + private static Map overrideDependencyConf(final SparkApplication app) { + Map confOverrides = new HashMap<>(); + sparkAppResourceLabels(app) + .forEach( + (k, v) -> { + confOverrides.put("spark.kubernetes.driver.label." + k, v); + confOverrides.put("spark.kubernetes.driver.service.label." + k, v); + confOverrides.put("spark.kubernetes.executor.label." + k, v); + }); + confOverrides.put("spark.kubernetes.namespace", app.getMetadata().getNamespace()); + if (app.getSpec().getSparkConf().containsKey("spark.app.name")) { + confOverrides.put("spark.app.name", app.getMetadata().getName()); + } + // FIXME: avoid this file flushing + confOverrides.putAll(getOrCreateLocalFileForDriverSpec(app, confOverrides)); + confOverrides.putAll(getOrCreateLocalFileForExecutorSpec(app, confOverrides)); + return confOverrides; + } + + private static void cleanUpTempResourcesForApp( + final SparkApplication app, Map confOverrides) { + if (overrideDriverTemplateEnabled(app.getSpec())) { + deleteLocalFileFromPathKey(confOverrides, DRIVER_SPARK_TEMPLATE_FILE_PROP_KEY); + } + if (overrideExecutorTemplateEnabled(app.getSpec())) { + deleteLocalFileFromPathKey(confOverrides, EXECUTOR_SPARK_TEMPLATE_FILE_PROP_KEY); + } + } + + private static Optional getLocalFileFromPathKey( + Map confOverrides, String pathKey) { + if (confOverrides.containsKey(pathKey)) { + String filePath = confOverrides.get(pathKey); + if (filePath.startsWith("local") || filePath.startsWith("file") || filePath.startsWith("/")) { + return Optional.of(new File(filePath)); + } + } + return Optional.empty(); + } + + private static void deleteLocalFileFromPathKey( + Map confOverrides, String pathKey) { + Optional localFile = Optional.empty(); + boolean deleted = false; + try { + localFile = getLocalFileFromPathKey(confOverrides, pathKey); + if (localFile.isPresent() && localFile.get().exists() && localFile.get().isFile()) { + deleted = localFile.get().delete(); + } else { + log.warn("Local temp file not found at {}", pathKey); + } + } catch (Throwable t) { + log.error("Failed to delete temp file. Attempting delete upon exit.", t); + } finally { + if (!deleted && localFile.isPresent() && localFile.get().exists()) { + localFile.get().deleteOnExit(); + } + } + } + + private static Map getOrCreateLocalFileForDriverSpec( + final SparkApplication app, final Map confOverrides) { + if (overrideDriverTemplateEnabled(app.getSpec())) { + Optional localFile = + getLocalFileFromPathKey(confOverrides, DRIVER_SPARK_TEMPLATE_FILE_PROP_KEY); + if (localFile.isEmpty() || !localFile.get().exists() || !localFile.get().isFile()) { + String filePath = + createLocalFileForPodTemplateSpec( + app.getSpec().getDriverSpec().getPodTemplateSpec(), + app.getMetadata().getUid() + "-driver-"); + return Collections.singletonMap(DRIVER_SPARK_TEMPLATE_FILE_PROP_KEY, filePath); + } + } + return Collections.emptyMap(); + } + + private static Map getOrCreateLocalFileForExecutorSpec( + final SparkApplication app, final Map confOverrides) { + if (overrideExecutorTemplateEnabled(app.getSpec())) { + Optional localFile = + getLocalFileFromPathKey(confOverrides, EXECUTOR_SPARK_TEMPLATE_FILE_PROP_KEY); + if (localFile.isEmpty() || !localFile.get().exists() || !localFile.get().isFile()) { + String filePath = + createLocalFileForPodTemplateSpec( + app.getSpec().getExecutorSpec().getPodTemplateSpec(), + app.getMetadata().getUid() + "-executor-"); + return Collections.singletonMap(EXECUTOR_SPARK_TEMPLATE_FILE_PROP_KEY, filePath); + } + } + return Collections.emptyMap(); + } + + /** + * Flush driver pod template spec to a local file + * + * @return temp file path + */ + private static String createLocalFileForPodTemplateSpec( + final PodTemplateSpec podTemplateSpec, final String tempFilePrefix) { + try { + File tmpFile = File.createTempFile(tempFilePrefix, ".json"); + FileOutputStream fileStream = new FileOutputStream(tmpFile); + OutputStreamWriter writer = new OutputStreamWriter(fileStream, "UTF-8"); + writer.write(ModelUtils.asJsonString(ModelUtils.getPodFromTemplateSpec(podTemplateSpec))); + writer.close(); + String path = tmpFile.getAbsolutePath(); + if (log.isDebugEnabled()) { + log.debug("Temp file wrote to {}", tmpFile.getAbsolutePath()); + } + return path; + } catch (IOException e) { + throw new RuntimeException(e); + } + } +} diff --git a/spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/observers/AppDriverReadyObserver.java b/spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/observers/AppDriverReadyObserver.java new file mode 100644 index 00000000..41c45b53 --- /dev/null +++ b/spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/observers/AppDriverReadyObserver.java @@ -0,0 +1,48 @@ +/* + * 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.spark.k8s.operator.reconciler.observers; + +import java.util.Optional; + +import io.fabric8.kubernetes.api.model.Pod; + +import org.apache.spark.k8s.operator.Constants; +import org.apache.spark.k8s.operator.spec.ApplicationSpec; +import org.apache.spark.k8s.operator.status.ApplicationState; +import org.apache.spark.k8s.operator.status.ApplicationStateSummary; +import org.apache.spark.k8s.operator.status.ApplicationStatus; +import org.apache.spark.k8s.operator.utils.PodUtils; + +/** Observes whether driver is ready */ +public class AppDriverReadyObserver extends BaseAppDriverObserver { + @Override + public Optional observe( + Pod driver, ApplicationSpec spec, ApplicationStatus currentStatus) { + if (ApplicationStateSummary.DriverReady.ordinal() + <= currentStatus.getCurrentState().getCurrentStateSummary().ordinal()) { + return Optional.empty(); + } + if (PodUtils.isPodReady(driver)) { + return Optional.of( + new ApplicationState(ApplicationStateSummary.DriverReady, Constants.DRIVER_READY)); + } + return observeDriverTermination(driver, true, spec); + } +} diff --git a/spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/observers/AppDriverRunningObserver.java b/spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/observers/AppDriverRunningObserver.java new file mode 100644 index 00000000..c49fc333 --- /dev/null +++ b/spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/observers/AppDriverRunningObserver.java @@ -0,0 +1,39 @@ +/* + * 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.spark.k8s.operator.reconciler.observers; + +import java.util.Optional; + +import io.fabric8.kubernetes.api.model.Pod; + +import org.apache.spark.k8s.operator.spec.ApplicationSpec; +import org.apache.spark.k8s.operator.status.ApplicationState; +import org.apache.spark.k8s.operator.status.ApplicationStatus; + +/** + * Observes whether driver reaches running state (in other words, whether its at least scheduled) + */ +public class AppDriverRunningObserver extends BaseAppDriverObserver { + @Override + public Optional observe( + Pod driver, ApplicationSpec spec, ApplicationStatus currentStatus) { + return observeDriverTermination(driver, true, spec); + } +} diff --git a/spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/observers/AppDriverStartObserver.java b/spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/observers/AppDriverStartObserver.java new file mode 100644 index 00000000..0d3d90f9 --- /dev/null +++ b/spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/observers/AppDriverStartObserver.java @@ -0,0 +1,47 @@ +/* + * 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.spark.k8s.operator.reconciler.observers; + +import java.util.Optional; + +import io.fabric8.kubernetes.api.model.Pod; + +import org.apache.spark.k8s.operator.Constants; +import org.apache.spark.k8s.operator.spec.ApplicationSpec; +import org.apache.spark.k8s.operator.status.ApplicationState; +import org.apache.spark.k8s.operator.status.ApplicationStateSummary; +import org.apache.spark.k8s.operator.status.ApplicationStatus; +import org.apache.spark.k8s.operator.utils.PodUtils; + +public class AppDriverStartObserver extends BaseAppDriverObserver { + @Override + public Optional observe( + Pod driver, ApplicationSpec spec, ApplicationStatus currentStatus) { + if (ApplicationStateSummary.DriverStarted.ordinal() + <= currentStatus.getCurrentState().getCurrentStateSummary().ordinal()) { + return Optional.empty(); + } + if (PodUtils.isDriverPodStarted(driver, spec)) { + return Optional.of( + new ApplicationState(ApplicationStateSummary.DriverStarted, Constants.DRIVER_RUNNING)); + } + return observeDriverTermination(driver, false, spec); + } +} diff --git a/spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/observers/AppDriverTimeoutObserver.java b/spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/observers/AppDriverTimeoutObserver.java new file mode 100644 index 00000000..a32968ac --- /dev/null +++ b/spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/observers/AppDriverTimeoutObserver.java @@ -0,0 +1,93 @@ +/* + * 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.spark.k8s.operator.reconciler.observers; + +import java.time.Instant; +import java.util.Optional; +import java.util.function.Supplier; + +import io.fabric8.kubernetes.api.model.Pod; + +import org.apache.spark.k8s.operator.spec.ApplicationSpec; +import org.apache.spark.k8s.operator.spec.ApplicationTimeoutConfig; +import org.apache.spark.k8s.operator.status.ApplicationState; +import org.apache.spark.k8s.operator.status.ApplicationStatus; +import org.apache.spark.k8s.operator.utils.SparkAppStatusUtils; + +/** Observes driver status and time-out as configured in app spec */ +public class AppDriverTimeoutObserver extends BaseAppDriverObserver { + + /** + * Operator may proactively terminate application if it has stay in certain state for a while. + * This helps to avoid resource deadlock when app cannot proceed. Such states include: + * + *

    + *
  • DRIVER_REQUESTED -> goes to DRIVER_LAUNCH_TIMED_OUT if driver pod cannot be scheduled or + * cannot start running + *
  • DRIVER_REQUESTED -> goes to DRIVER_LAUNCH_TIMED_OUT if driver pod cannot be scheduled or + * cannot start running + *
  • DRIVER_STARTED -> goes to SPARK_SESSION_INITIALIZATION_TIMED_OUT if Spark session cannot + * be initialized + *
  • DRIVER_READY / EXECUTOR_REQUESTED / EXECUTOR_SCHEDULED / + * INITIALIZED_BELOW_THRESHOLD_EXECUTORS -> go to EXECUTORS_LAUNCH_TIMED_OUT if app cannot + * acquire at least minimal executors in given time + *
+ * + *

Operator will NOT proactively stop the app if it has acquired enough executors and later + * lose them. User may build additional layers to alert and act on such scenario. Timeout check + * would be performed at the end of reconcile - and it would be performed only if there's no other + * updates to be performed in the same reconcile action + */ + @Override + public Optional observe( + Pod driver, ApplicationSpec spec, ApplicationStatus currentStatus) { + Instant lastTransitionTime = + Instant.parse(currentStatus.getCurrentState().getLastTransitionTime()); + long timeoutThreshold; + Supplier supplier; + ApplicationTimeoutConfig timeoutConfig = + spec.getApplicationTolerations().getApplicationTimeoutConfig(); + switch (currentStatus.getCurrentState().getCurrentStateSummary()) { + case DriverRequested: + timeoutThreshold = timeoutConfig.getDriverStartTimeoutMillis(); + supplier = SparkAppStatusUtils::driverLaunchTimedOut; + break; + case DriverStarted: + timeoutThreshold = timeoutConfig.getDriverReadyTimeoutMillis(); + supplier = SparkAppStatusUtils::driverReadyTimedOut; + break; + case DriverReady: + case InitializedBelowThresholdExecutors: + timeoutThreshold = timeoutConfig.getExecutorStartTimeoutMillis(); + supplier = SparkAppStatusUtils::executorLaunchTimedOut; + break; + default: + // No timeout check needed for other states + return Optional.empty(); + } + if (timeoutThreshold > 0L + && lastTransitionTime.plusMillis(timeoutThreshold).isBefore(Instant.now())) { + ApplicationState state = supplier.get(); + state.setLastObservedDriverStatus(driver.getStatus()); + return Optional.of(state); + } + return Optional.empty(); + } +} diff --git a/spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/observers/BaseAppDriverObserver.java b/spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/observers/BaseAppDriverObserver.java new file mode 100644 index 00000000..6d984423 --- /dev/null +++ b/spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/observers/BaseAppDriverObserver.java @@ -0,0 +1,152 @@ +/* + * 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.spark.k8s.operator.reconciler.observers; + +import java.util.List; +import java.util.Optional; + +import io.fabric8.kubernetes.api.model.ContainerStatus; +import io.fabric8.kubernetes.api.model.Pod; +import lombok.extern.slf4j.Slf4j; + +import org.apache.spark.k8s.operator.Constants; +import org.apache.spark.k8s.operator.spec.ApplicationSpec; +import org.apache.spark.k8s.operator.status.ApplicationAttemptSummary; +import org.apache.spark.k8s.operator.status.ApplicationState; +import org.apache.spark.k8s.operator.status.ApplicationStateSummary; +import org.apache.spark.k8s.operator.status.ApplicationStatus; +import org.apache.spark.k8s.operator.utils.ModelUtils; +import org.apache.spark.k8s.operator.utils.PodPhase; +import org.apache.spark.k8s.operator.utils.PodUtils; + +/** Observes driver pod status and update Application status as needed */ +@Slf4j +public abstract class BaseAppDriverObserver + extends BaseSecondaryResourceObserver< + ApplicationStateSummary, + ApplicationAttemptSummary, + ApplicationState, + ApplicationSpec, + ApplicationStatus, + Pod> { + + /** + * Check whether the driver pod (and thus the application) has actually terminated This would be + * determined by status of containers - and only by the containers with name matches given filter + * e.g. you can use "s -> 'true'" to evaluate all containers Driver is considered as 'failed', if + * any init container failed, or any of the matched container(s) has + * + *

    + *
  • 1. failed (isTerminated, non-zero) + *
  • 2. restarted + *
  • 3. (corner case) exited 0 without SparkContext / SparkSession initialization + *
+ * + *

Driver is considered as 'succeeded', if + * + *

    + *
  • 1. The pod is succeeded phase, or + *
  • 2. The container(s) has exited 0 after SparkContext / SparkSession initialization + *
+ * + * @param driverPod the driverPod + * @param driverReady whether SparkContext / SparkSession has ever been initialized for this pod + * @return the ApplicationState to be updated if pod is terminated. Returning empty if pod is + * still running + */ + protected Optional observeDriverTermination( + final Pod driverPod, final boolean driverReady, final ApplicationSpec spec) { + if (driverPod.getStatus() == null + || driverPod.getStatus().getContainerStatuses() == null + || driverPod.getStatus().getContainerStatuses().isEmpty()) { + log.warn("Cannot determine driver pod status, the pod may in pending state."); + return Optional.empty(); + } + + if (PodPhase.FAILED.equals(PodPhase.getPhase(driverPod))) { + ApplicationState applicationState = + new ApplicationState(ApplicationStateSummary.Failed, Constants.DRIVER_FAILED_MESSAGE); + if ("Evicted".equalsIgnoreCase(driverPod.getStatus().getReason())) { + applicationState = + new ApplicationState( + ApplicationStateSummary.DriverEvicted, Constants.DRIVER_FAILED_MESSAGE); + } + applicationState.setLastObservedDriverStatus(driverPod.getStatus()); + return Optional.of(applicationState); + } + + if (PodPhase.SUCCEEDED.equals(PodPhase.getPhase(driverPod))) { + ApplicationState state; + if (driverReady) { + state = + new ApplicationState( + ApplicationStateSummary.Succeeded, Constants.DRIVER_COMPLETED_MESSAGE); + } else { + state = + new ApplicationState( + ApplicationStateSummary.Failed, + Constants.DRIVER_TERMINATED_BEFORE_INITIALIZATION_MESSAGE); + state.setLastObservedDriverStatus(driverPod.getStatus()); + } + return Optional.of(state); + } + + List initContainerStatusList = + driverPod.getStatus().getInitContainerStatuses(); + if (initContainerStatusList != null + && initContainerStatusList.parallelStream().anyMatch(PodUtils::isContainerFailed)) { + ApplicationState applicationState = + new ApplicationState( + ApplicationStateSummary.Failed, Constants.DRIVER_FAILED_INIT_CONTAINERS_MESSAGE); + applicationState.setLastObservedDriverStatus(driverPod.getStatus()); + return Optional.of(applicationState); + } + List containerStatusList = driverPod.getStatus().getContainerStatuses(); + List terminatedCriticalContainers = + ModelUtils.findDriverMainContainerStatus(spec, containerStatusList).stream() + .filter(PodUtils::isContainerTerminated) + .toList(); + + if (!terminatedCriticalContainers.isEmpty()) { + ApplicationState applicationState; + if (terminatedCriticalContainers.parallelStream().anyMatch(PodUtils::isContainerFailed)) { + applicationState = + new ApplicationState(ApplicationStateSummary.Failed, Constants.DRIVER_FAILED_MESSAGE); + } else { + applicationState = + new ApplicationState( + ApplicationStateSummary.Succeeded, Constants.DRIVER_SUCCEEDED_MESSAGE); + } + applicationState.setLastObservedDriverStatus(driverPod.getStatus()); + return Optional.of(applicationState); + } + boolean driverRestarted = + ModelUtils.findDriverMainContainerStatus(spec, containerStatusList).stream() + .anyMatch(PodUtils::isContainerRestarted); + + if (driverRestarted) { + ApplicationState state = + new ApplicationState(ApplicationStateSummary.Failed, Constants.DRIVER_RESTARTED_MESSAGE); + state.setLastObservedDriverStatus(driverPod.getStatus()); + return Optional.of(state); + } + return Optional.empty(); + } +} diff --git a/spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/observers/BaseSecondaryResourceObserver.java b/spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/observers/BaseSecondaryResourceObserver.java new file mode 100644 index 00000000..a2c5bbd6 --- /dev/null +++ b/spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/observers/BaseSecondaryResourceObserver.java @@ -0,0 +1,44 @@ +/* + * 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.spark.k8s.operator.reconciler.observers; + +import java.util.Optional; + +import io.fabric8.kubernetes.api.model.HasMetadata; + +import org.apache.spark.k8s.operator.spec.BaseSpec; +import org.apache.spark.k8s.operator.status.BaseAttemptSummary; +import org.apache.spark.k8s.operator.status.BaseState; +import org.apache.spark.k8s.operator.status.BaseStatus; + +/** + * Observe given secondary resource, return state to be updated if applicable. These observers do + * not act on secondary resource. They only observe secondary resource status and update owner + * SparkApplication status if needed + */ +public abstract class BaseSecondaryResourceObserver< + S, + AS extends BaseAttemptSummary, + STATE extends BaseState, + SPEC extends BaseSpec, + STATUS extends BaseStatus, + SR extends HasMetadata> { + public abstract Optional observe(SR secondaryResource, SPEC spec, STATUS currentStatus); +} diff --git a/spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/AppCleanUpStep.java b/spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/AppCleanUpStep.java new file mode 100644 index 00000000..f75967a1 --- /dev/null +++ b/spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/AppCleanUpStep.java @@ -0,0 +1,164 @@ +/* + * 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.spark.k8s.operator.reconciler.reconcilesteps; + +import java.time.Duration; +import java.time.Instant; +import java.util.ArrayList; +import java.util.List; +import java.util.Optional; +import java.util.function.Supplier; + +import io.fabric8.kubernetes.api.model.HasMetadata; +import io.fabric8.kubernetes.api.model.Pod; +import lombok.AllArgsConstructor; +import lombok.NoArgsConstructor; +import lombok.extern.slf4j.Slf4j; +import org.apache.commons.lang3.StringUtils; + +import org.apache.spark.k8s.operator.SparkApplication; +import org.apache.spark.k8s.operator.config.SparkOperatorConf; +import org.apache.spark.k8s.operator.context.SparkAppContext; +import org.apache.spark.k8s.operator.reconciler.ReconcileProgress; +import org.apache.spark.k8s.operator.spec.ApplicationTolerations; +import org.apache.spark.k8s.operator.spec.ResourceRetainPolicy; +import org.apache.spark.k8s.operator.spec.RestartPolicy; +import org.apache.spark.k8s.operator.status.ApplicationState; +import org.apache.spark.k8s.operator.status.ApplicationStateSummary; +import org.apache.spark.k8s.operator.status.ApplicationStatus; +import org.apache.spark.k8s.operator.utils.ReconcilerUtils; +import org.apache.spark.k8s.operator.utils.SparkAppStatusRecorder; + +/** + * Cleanup all secondary resources when application is deleted, or at the end of each attempt. + * Update Application status to indicate whether another attempt would be made + */ +@AllArgsConstructor +@NoArgsConstructor +@Slf4j +public class AppCleanUpStep extends AppReconcileStep { + private Supplier cleanUpSuccessStateSupplier; + + @Override + public ReconcileProgress reconcile( + SparkAppContext context, SparkAppStatusRecorder statusRecorder) { + ApplicationStatus currentStatus = context.getResource().getStatus(); + ApplicationTolerations tolerations = + context.getResource().getSpec().getApplicationTolerations(); + ResourceRetainPolicy resourceRetainPolicy = tolerations.getResourceRetainPolicy(); + String stateMessage = null; + + if (retainReleaseResource(resourceRetainPolicy, currentStatus.getCurrentState())) { + if (tolerations.getRestartConfig() != null + && !RestartPolicy.Never.equals(tolerations.getRestartConfig().getRestartPolicy())) { + stateMessage = + "Application is configured to restart, resources created in current " + + "attempt would be force released."; + log.warn(stateMessage); + } else { + ApplicationState terminationState = + new ApplicationState( + ApplicationStateSummary.TerminatedWithoutReleaseResources, + "Application is terminated without releasing resources as configured."); + long requeueAfterMillis = + tolerations.getApplicationTimeoutConfig().getTerminationRequeuePeriodMillis(); + return appendStateAndRequeueAfter( + context, statusRecorder, terminationState, Duration.ofMillis(requeueAfterMillis)); + } + } + List resourcesToRemove = new ArrayList<>(); + if (ApplicationStateSummary.SchedulingFailure.equals( + currentStatus.getCurrentState().getCurrentStateSummary())) { + // if app failed at scheduling, re-compute all spec and delete as they may not be fully + // owned by driver + try { + resourcesToRemove.addAll(context.getDriverPreResourcesSpec()); + resourcesToRemove.add(context.getDriverPodSpec()); + resourcesToRemove.addAll(context.getDriverResourcesSpec()); + } catch (Exception e) { + if (log.isErrorEnabled()) { + log.error("Failed to build resources for application.", e); + } + ApplicationState updatedState = + new ApplicationState( + ApplicationStateSummary.ResourceReleased, + "Cannot build Spark spec for given application, " + + "consider all resources as released."); + long requeueAfterMillis = + tolerations.getApplicationTimeoutConfig().getTerminationRequeuePeriodMillis(); + return appendStateAndRequeueAfter( + context, statusRecorder, updatedState, Duration.ofMillis(requeueAfterMillis)); + } + } else { + Optional driver = context.getDriverPod(); + driver.ifPresent(resourcesToRemove::add); + } + boolean forceDelete = enableForceDelete(context.getResource()); + for (HasMetadata resource : resourcesToRemove) { + ReconcilerUtils.deleteResourceIfExists(context.getClient(), resource, forceDelete); + } + ApplicationStatus updatedStatus; + if (cleanUpSuccessStateSupplier != null) { + ApplicationState state = cleanUpSuccessStateSupplier.get(); + if (StringUtils.isNotEmpty(stateMessage)) { + state.setMessage(stateMessage); + } + long requeueAfterMillis = + tolerations.getApplicationTimeoutConfig().getTerminationRequeuePeriodMillis(); + return appendStateAndRequeueAfter( + context, statusRecorder, state, Duration.ofMillis(requeueAfterMillis)); + } else { + updatedStatus = + currentStatus.terminateOrRestart( + tolerations.getRestartConfig(), + tolerations.getResourceRetainPolicy(), + stateMessage, + SparkOperatorConf.TRIM_ATTEMPT_STATE_TRANSITION_HISTORY.getValue()); + long requeueAfterMillis = + tolerations.getApplicationTimeoutConfig().getTerminationRequeuePeriodMillis(); + if (ApplicationStateSummary.ScheduledToRestart.equals( + updatedStatus.getCurrentState().getCurrentStateSummary())) { + requeueAfterMillis = tolerations.getRestartConfig().getRestartBackoffMillis(); + } + return updateStatusAndRequeueAfter( + context, statusRecorder, updatedStatus, Duration.ofMillis(requeueAfterMillis)); + } + } + + protected boolean retainReleaseResource( + ResourceRetainPolicy resourceRetainPolicy, ApplicationState currentState) { + return switch (resourceRetainPolicy) { + case Never -> false; + case OnFailure -> currentState.getCurrentStateSummary().isFailure(); + default -> true; + }; + } + + protected boolean enableForceDelete(SparkApplication app) { + long timeoutThreshold = + app.getSpec() + .getApplicationTolerations() + .getApplicationTimeoutConfig() + .getForceTerminationGracePeriodMillis(); + Instant lastTransitionTime = + Instant.parse(app.getStatus().getCurrentState().getLastTransitionTime()); + return lastTransitionTime.plusMillis(timeoutThreshold).isBefore(Instant.now()); + } +} diff --git a/spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/AppInitStep.java b/spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/AppInitStep.java new file mode 100644 index 00000000..e102bce1 --- /dev/null +++ b/spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/AppInitStep.java @@ -0,0 +1,136 @@ +/* + * 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.spark.k8s.operator.reconciler.reconcilesteps; + +import static org.apache.spark.k8s.operator.reconciler.ReconcileProgress.completeAndDefaultRequeue; +import static org.apache.spark.k8s.operator.reconciler.ReconcileProgress.completeAndImmediateRequeue; +import static org.apache.spark.k8s.operator.reconciler.ReconcileProgress.proceed; +import static org.apache.spark.k8s.operator.utils.SparkExceptionUtils.buildGeneralErrorMessage; + +import java.time.Duration; +import java.time.Instant; +import java.util.ArrayList; +import java.util.List; +import java.util.Optional; + +import io.fabric8.kubernetes.api.model.HasMetadata; +import io.fabric8.kubernetes.api.model.Pod; +import lombok.extern.slf4j.Slf4j; + +import org.apache.spark.k8s.operator.Constants; +import org.apache.spark.k8s.operator.SparkApplication; +import org.apache.spark.k8s.operator.context.SparkAppContext; +import org.apache.spark.k8s.operator.decorators.DriverResourceDecorator; +import org.apache.spark.k8s.operator.reconciler.ReconcileProgress; +import org.apache.spark.k8s.operator.status.ApplicationState; +import org.apache.spark.k8s.operator.status.ApplicationStateSummary; +import org.apache.spark.k8s.operator.status.ApplicationStatus; +import org.apache.spark.k8s.operator.utils.ReconcilerUtils; +import org.apache.spark.k8s.operator.utils.SparkAppStatusRecorder; + +/** Request all driver and driver resources when starting an attempt */ +@Slf4j +public class AppInitStep extends AppReconcileStep { + @Override + public ReconcileProgress reconcile( + SparkAppContext context, SparkAppStatusRecorder statusRecorder) { + ApplicationState currentState = context.getResource().getStatus().getCurrentState(); + if (!currentState.getCurrentStateSummary().isInitializing()) { + return proceed(); + } + SparkApplication app = context.getResource(); + if (app.getStatus().getPreviousAttemptSummary() != null) { + Instant lastTransitionTime = Instant.parse(currentState.getLastTransitionTime()); + Instant restartTime = + lastTransitionTime.plusMillis( + app.getSpec() + .getApplicationTolerations() + .getRestartConfig() + .getRestartBackoffMillis()); + Instant now = Instant.now(); + if (restartTime.isAfter(now)) { + return ReconcileProgress.completeAndRequeueAfter(Duration.between(now, restartTime)); + } + } + try { + List createdPreResources = new ArrayList<>(); + for (HasMetadata resource : context.getDriverPreResourcesSpec()) { + Optional createdResource = + ReconcilerUtils.getOrCreateSecondaryResource(context.getClient(), resource); + if (createdResource.isPresent()) { + createdPreResources.add(createdResource.get()); + } else { + return appendStateAndImmediateRequeue( + context, statusRecorder, creationFailureState(resource)); + } + } + Optional driverPod = + ReconcilerUtils.getOrCreateSecondaryResource( + context.getClient(), context.getDriverPodSpec()); + if (driverPod.isPresent()) { + DriverResourceDecorator decorator = new DriverResourceDecorator(driverPod.get()); + createdPreResources.forEach(decorator::decorate); + context.getClient().resourceList(createdPreResources).forceConflicts().serverSideApply(); + List driverResources = context.getDriverResourcesSpec(); + driverResources.forEach(decorator::decorate); + for (HasMetadata resource : driverResources) { + Optional createdResource = + ReconcilerUtils.getOrCreateSecondaryResource(context.getClient(), resource); + if (createdResource.isEmpty()) { + return appendStateAndImmediateRequeue( + context, statusRecorder, creationFailureState(resource)); + } + } + } + ApplicationStatus updatedStatus = + context + .getResource() + .getStatus() + .appendNewState( + new ApplicationState( + ApplicationStateSummary.DriverRequested, Constants.DRIVER_REQUESTED_MESSAGE)); + statusRecorder.persistStatus(context, updatedStatus); + return completeAndDefaultRequeue(); + } catch (Exception e) { + if (log.isErrorEnabled()) { + log.error("Failed to request driver resource.", e); + } + String errorMessage = + Constants.SCHEDULE_FAILURE_MESSAGE + " StackTrace: " + buildGeneralErrorMessage(e); + statusRecorder.persistStatus( + context, + context + .getResource() + .getStatus() + .appendNewState( + new ApplicationState(ApplicationStateSummary.SchedulingFailure, errorMessage))); + return completeAndImmediateRequeue(); + } + } + + private ApplicationState creationFailureState(HasMetadata failedResource) { + return new ApplicationState( + ApplicationStateSummary.SchedulingFailure, + "Failed to request resource for driver with kind: " + + failedResource.getKind() + + ", name: " + + failedResource.getMetadata().getName()); + } +} diff --git a/spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/AppReconcileStep.java b/spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/AppReconcileStep.java new file mode 100644 index 00000000..14f6b995 --- /dev/null +++ b/spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/AppReconcileStep.java @@ -0,0 +1,98 @@ +/* + * 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.spark.k8s.operator.reconciler.reconcilesteps; + +import static org.apache.spark.k8s.operator.reconciler.ReconcileProgress.completeAndImmediateRequeue; +import static org.apache.spark.k8s.operator.reconciler.ReconcileProgress.proceed; +import static org.apache.spark.k8s.operator.utils.SparkAppStatusUtils.driverUnexpectedRemoved; + +import java.time.Duration; +import java.util.List; +import java.util.Optional; +import java.util.stream.Collectors; + +import io.fabric8.kubernetes.api.model.Pod; + +import org.apache.spark.k8s.operator.SparkApplication; +import org.apache.spark.k8s.operator.context.SparkAppContext; +import org.apache.spark.k8s.operator.reconciler.ReconcileProgress; +import org.apache.spark.k8s.operator.reconciler.observers.BaseAppDriverObserver; +import org.apache.spark.k8s.operator.status.ApplicationState; +import org.apache.spark.k8s.operator.status.ApplicationStatus; +import org.apache.spark.k8s.operator.utils.SparkAppStatusRecorder; + +/** Basic reconcile step for application */ +public abstract class AppReconcileStep { + public abstract ReconcileProgress reconcile( + SparkAppContext context, SparkAppStatusRecorder statusRecorder); + + protected ReconcileProgress observeDriver( + final SparkAppContext context, + final SparkAppStatusRecorder statusRecorder, + final List observers) { + Optional driverPodOptional = context.getDriverPod(); + SparkApplication app = context.getResource(); + ApplicationStatus currentStatus = app.getStatus(); + if (driverPodOptional.isPresent()) { + List stateUpdates = + observers.stream() + .map(o -> o.observe(driverPodOptional.get(), app.getSpec(), app.getStatus())) + .filter(Optional::isPresent) + .map(Optional::get) + .collect(Collectors.toList()); + if (stateUpdates.isEmpty()) { + return proceed(); + } else { + for (ApplicationState state : stateUpdates) { + currentStatus = currentStatus.appendNewState(state); + } + statusRecorder.persistStatus(context, currentStatus); + return completeAndImmediateRequeue(); + } + } else { + ApplicationStatus updatedStatus = currentStatus.appendNewState(driverUnexpectedRemoved()); + statusRecorder.persistStatus(context, updatedStatus); + return completeAndImmediateRequeue(); + } + } + + protected ReconcileProgress updateStatusAndRequeueAfter( + SparkAppContext context, + SparkAppStatusRecorder statusRecorder, + ApplicationStatus updatedStatus, + Duration requeueAfter) { + statusRecorder.persistStatus(context, updatedStatus); + return ReconcileProgress.completeAndRequeueAfter(requeueAfter); + } + + protected ReconcileProgress appendStateAndRequeueAfter( + SparkAppContext context, + SparkAppStatusRecorder statusRecorder, + ApplicationState newState, + Duration requeueAfter) { + statusRecorder.appendNewStateAndPersist(context, newState); + return ReconcileProgress.completeAndRequeueAfter(requeueAfter); + } + + protected ReconcileProgress appendStateAndImmediateRequeue( + SparkAppContext context, SparkAppStatusRecorder statusRecorder, ApplicationState newState) { + return appendStateAndRequeueAfter(context, statusRecorder, newState, Duration.ZERO); + } +} diff --git a/spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/AppResourceObserveStep.java b/spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/AppResourceObserveStep.java new file mode 100644 index 00000000..0db6dfdf --- /dev/null +++ b/spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/AppResourceObserveStep.java @@ -0,0 +1,42 @@ +/* + * 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.spark.k8s.operator.reconciler.reconcilesteps; + +import java.util.List; + +import lombok.RequiredArgsConstructor; + +import org.apache.spark.k8s.operator.context.SparkAppContext; +import org.apache.spark.k8s.operator.reconciler.ReconcileProgress; +import org.apache.spark.k8s.operator.reconciler.observers.BaseAppDriverObserver; +import org.apache.spark.k8s.operator.utils.SparkAppStatusRecorder; + +/** Observes secondary resource and update app status if needed */ +@RequiredArgsConstructor +public class AppResourceObserveStep extends AppReconcileStep { + + private final List observers; + + @Override + public ReconcileProgress reconcile( + final SparkAppContext context, final SparkAppStatusRecorder statusRecorder) { + return observeDriver(context, statusRecorder, observers); + } +} diff --git a/spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/AppRunningStep.java b/spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/AppRunningStep.java new file mode 100644 index 00000000..02db7819 --- /dev/null +++ b/spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/AppRunningStep.java @@ -0,0 +1,92 @@ +/* + * 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.spark.k8s.operator.reconciler.reconcilesteps; + +import static org.apache.spark.k8s.operator.reconciler.ReconcileProgress.completeAndDefaultRequeue; + +import java.util.Collections; +import java.util.Set; + +import io.fabric8.kubernetes.api.model.Pod; + +import org.apache.spark.k8s.operator.Constants; +import org.apache.spark.k8s.operator.context.SparkAppContext; +import org.apache.spark.k8s.operator.reconciler.ReconcileProgress; +import org.apache.spark.k8s.operator.reconciler.observers.AppDriverRunningObserver; +import org.apache.spark.k8s.operator.spec.InstanceConfig; +import org.apache.spark.k8s.operator.status.ApplicationState; +import org.apache.spark.k8s.operator.status.ApplicationStateSummary; +import org.apache.spark.k8s.operator.utils.PodUtils; +import org.apache.spark.k8s.operator.utils.SparkAppStatusRecorder; + +/** Observe whether app acquires enough executors as configured in spec */ +public class AppRunningStep extends AppReconcileStep { + @Override + public ReconcileProgress reconcile( + SparkAppContext context, SparkAppStatusRecorder statusRecorder) { + InstanceConfig instanceConfig = + context.getResource().getSpec().getApplicationTolerations().getInstanceConfig(); + ApplicationStateSummary prevStateSummary = + context.getResource().getStatus().getCurrentState().getCurrentStateSummary(); + ApplicationStateSummary proposedStateSummary; + String stateMessage = context.getResource().getStatus().getCurrentState().getMessage(); + if (instanceConfig == null + || instanceConfig.getInitExecutors() == 0L + || (!prevStateSummary.isStarting() && instanceConfig.getMinExecutors() == 0L)) { + proposedStateSummary = ApplicationStateSummary.RunningHealthy; + stateMessage = Constants.RUNNING_HEALTHY_MESSAGE; + } else { + Set executors = context.getExecutorsForApplication(); + long runningExecutors = executors.stream().filter(PodUtils::isPodReady).count(); + if (prevStateSummary.isStarting()) { + if (runningExecutors >= instanceConfig.getInitExecutors()) { + proposedStateSummary = ApplicationStateSummary.RunningHealthy; + stateMessage = Constants.RUNNING_HEALTHY_MESSAGE; + } else if (runningExecutors > 0L) { + proposedStateSummary = ApplicationStateSummary.InitializedBelowThresholdExecutors; + stateMessage = Constants.INITIALIZED_WITH_BELOW_THRESHOLD_EXECUTORS_MESSAGE; + } else { + // keep previous state for 0 executor + proposedStateSummary = prevStateSummary; + } + } else { + if (runningExecutors >= instanceConfig.getMinExecutors()) { + proposedStateSummary = ApplicationStateSummary.RunningHealthy; + stateMessage = Constants.RUNNING_HEALTHY_MESSAGE; + } else { + proposedStateSummary = ApplicationStateSummary.RunningWithBelowThresholdExecutors; + stateMessage = Constants.RUNNING_WITH_BELOW_THRESHOLD_EXECUTORS_MESSAGE; + } + } + } + if (!proposedStateSummary.equals(prevStateSummary)) { + statusRecorder.persistStatus( + context, + context + .getResource() + .getStatus() + .appendNewState(new ApplicationState(proposedStateSummary, stateMessage))); + return completeAndDefaultRequeue(); + } else { + return observeDriver( + context, statusRecorder, Collections.singletonList(new AppDriverRunningObserver())); + } + } +} diff --git a/spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/AppTerminatedStep.java b/spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/AppTerminatedStep.java new file mode 100644 index 00000000..16a4155d --- /dev/null +++ b/spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/AppTerminatedStep.java @@ -0,0 +1,44 @@ +/* + * 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.spark.k8s.operator.reconciler.reconcilesteps; + +import static org.apache.spark.k8s.operator.reconciler.ReconcileProgress.proceed; + +import org.apache.spark.k8s.operator.context.SparkAppContext; +import org.apache.spark.k8s.operator.reconciler.ReconcileProgress; +import org.apache.spark.k8s.operator.utils.SparkAppStatusRecorder; + +/** Observes whether app is already terminated. If so, end the reconcile. */ +public class AppTerminatedStep extends AppReconcileStep { + @Override + public ReconcileProgress reconcile( + SparkAppContext context, SparkAppStatusRecorder statusRecorder) { + if (context + .getResource() + .getStatus() + .getCurrentState() + .getCurrentStateSummary() + .isTerminated()) { + statusRecorder.removeCachedStatus(context.getResource()); + return ReconcileProgress.completeAndNoRequeue(); + } + return proceed(); + } +} diff --git a/spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/AppValidateStep.java b/spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/AppValidateStep.java new file mode 100644 index 00000000..b8f50123 --- /dev/null +++ b/spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/AppValidateStep.java @@ -0,0 +1,55 @@ +/* + * 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.spark.k8s.operator.reconciler.reconcilesteps; + +import static org.apache.spark.k8s.operator.reconciler.ReconcileProgress.completeAndImmediateRequeue; +import static org.apache.spark.k8s.operator.reconciler.ReconcileProgress.proceed; +import static org.apache.spark.k8s.operator.utils.SparkAppStatusUtils.isValidApplicationStatus; + +import lombok.extern.slf4j.Slf4j; + +import org.apache.spark.k8s.operator.context.SparkAppContext; +import org.apache.spark.k8s.operator.reconciler.ReconcileProgress; +import org.apache.spark.k8s.operator.spec.DeploymentMode; +import org.apache.spark.k8s.operator.status.ApplicationState; +import org.apache.spark.k8s.operator.status.ApplicationStateSummary; +import org.apache.spark.k8s.operator.status.ApplicationStatus; +import org.apache.spark.k8s.operator.utils.SparkAppStatusRecorder; + +/** Validates the submitted app. This can be re-factored into webhook in future. */ +@Slf4j +public class AppValidateStep extends AppReconcileStep { + @Override + public ReconcileProgress reconcile( + SparkAppContext context, SparkAppStatusRecorder statusRecorder) { + if (!isValidApplicationStatus(context.getResource())) { + log.warn("Spark application found with empty status. Resetting to initial state."); + statusRecorder.persistStatus(context, new ApplicationStatus()); + } + if (DeploymentMode.ClientMode.equals(context.getResource().getSpec())) { + ApplicationState failure = + new ApplicationState(ApplicationStateSummary.Failed, "Client mode is not supported yet."); + statusRecorder.persistStatus( + context, context.getResource().getStatus().appendNewState(failure)); + return completeAndImmediateRequeue(); + } + return proceed(); + } +} diff --git a/spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/UnknownStateStep.java b/spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/UnknownStateStep.java new file mode 100644 index 00000000..0f6159d7 --- /dev/null +++ b/spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/UnknownStateStep.java @@ -0,0 +1,45 @@ +/* + * 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.spark.k8s.operator.reconciler.reconcilesteps; + +import java.util.Optional; + +import io.fabric8.kubernetes.api.model.Pod; + +import org.apache.spark.k8s.operator.Constants; +import org.apache.spark.k8s.operator.context.SparkAppContext; +import org.apache.spark.k8s.operator.reconciler.ReconcileProgress; +import org.apache.spark.k8s.operator.status.ApplicationState; +import org.apache.spark.k8s.operator.status.ApplicationStateSummary; +import org.apache.spark.k8s.operator.utils.SparkAppStatusRecorder; + +/** Abnormal state handler */ +public class UnknownStateStep extends AppReconcileStep { + @Override + public ReconcileProgress reconcile( + SparkAppContext context, SparkAppStatusRecorder statusRecorder) { + ApplicationState state = + new ApplicationState(ApplicationStateSummary.Failed, Constants.UNKNOWN_STATE_MESSAGE); + Optional driver = context.getDriverPod(); + driver.ifPresent(pod -> state.setLastObservedDriverStatus(pod.getStatus())); + statusRecorder.persistStatus(context, context.getResource().getStatus().appendNewState(state)); + return ReconcileProgress.completeAndImmediateRequeue(); + } +} diff --git a/spark-operator/src/main/java/org/apache/spark/k8s/operator/utils/ReconcilerUtils.java b/spark-operator/src/main/java/org/apache/spark/k8s/operator/utils/ReconcilerUtils.java new file mode 100644 index 00000000..ca25833b --- /dev/null +++ b/spark-operator/src/main/java/org/apache/spark/k8s/operator/utils/ReconcilerUtils.java @@ -0,0 +1,169 @@ +/* + * 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.spark.k8s.operator.utils; + +import static org.apache.spark.k8s.operator.config.SparkOperatorConf.API_SECONDARY_RESOURCE_CREATE_MAX_ATTEMPTS; +import static org.apache.spark.k8s.operator.config.SparkOperatorConf.RECONCILER_FOREGROUND_REQUEST_TIMEOUT_SECONDS; +import static org.apache.spark.k8s.operator.utils.ModelUtils.buildOwnerReferenceTo; +import static org.apache.spark.k8s.operator.utils.SparkExceptionUtils.isConflictForExistingResource; + +import java.util.List; +import java.util.Optional; +import java.util.concurrent.TimeUnit; + +import com.fasterxml.jackson.core.JsonProcessingException; +import com.fasterxml.jackson.databind.ObjectMapper; +import io.fabric8.kubernetes.api.model.DeletionPropagation; +import io.fabric8.kubernetes.api.model.HasMetadata; +import io.fabric8.kubernetes.api.model.ObjectMeta; +import io.fabric8.kubernetes.api.model.ObjectMetaBuilder; +import io.fabric8.kubernetes.client.KubernetesClient; +import io.fabric8.kubernetes.client.KubernetesClientException; +import io.javaoperatorsdk.operator.api.reconciler.DeleteControl; +import io.javaoperatorsdk.operator.api.reconciler.UpdateControl; +import lombok.extern.slf4j.Slf4j; + +import org.apache.spark.k8s.operator.BaseResource; +import org.apache.spark.k8s.operator.reconciler.ReconcileProgress; + +@Slf4j +public class ReconcilerUtils { + public static > UpdateControl toUpdateControl( + O resource, ReconcileProgress reconcileProgress) { + // reconciler already handled resource and status update, skip update at lower level + UpdateControl updateControl = UpdateControl.noUpdate(); + if (reconcileProgress.isRequeue()) { + return updateControl.rescheduleAfter(reconcileProgress.getRequeueAfterDuration()); + } else { + return updateControl; + } + } + + public static > DeleteControl toDeleteControl( + O resource, ReconcileProgress reconcileProgress) { + if (reconcileProgress.isRequeue()) { + return DeleteControl.noFinalizerRemoval() + .rescheduleAfter(reconcileProgress.getRequeueAfterDuration()); + } else { + return DeleteControl.defaultDelete(); + } + } + + public static Optional getOrCreateSecondaryResource( + final KubernetesClient client, final T resource) { + Optional current = getResource(client, resource); + if (current.isEmpty()) { + // Adding retry logic to overcome known k8s issue: + // https://github.com/kubernetes/kubernetes/issues/67761 + long maxAttempts = API_SECONDARY_RESOURCE_CREATE_MAX_ATTEMPTS.getValue(); + long attemptCount = 1; + while (true) { + try { + current = Optional.ofNullable(client.resource(resource).create()); + break; + } catch (KubernetesClientException e) { + if (log.isErrorEnabled()) { + log.error( + "Failed to request resource with responseCode={} attemptCount={}/{}", + e.getCode(), + attemptCount, + maxAttempts); + } + // retry only on 409 Conflict + if (e.getCode() != 409) { + throw e; + } else { + if (isConflictForExistingResource(e)) { + current = getResource(client, resource); + if (current.isPresent()) { + return current; + } + } + if (++attemptCount > maxAttempts) { + log.error("Max Retries exceeded while trying to create resource"); + throw e; + } + } + } + } + } + return current; + } + + public static void addOwnerReferenceSecondaryResource( + final KubernetesClient client, final List resources, final HasMetadata owner) { + + resources.forEach( + r -> { + ObjectMeta metaData = + new ObjectMetaBuilder(r.getMetadata()) + .addToOwnerReferences(buildOwnerReferenceTo(owner)) + .build(); + r.setMetadata(metaData); + }); + client.resourceList(resources).forceConflicts().serverSideApply(); + } + + public static Optional getResource( + final KubernetesClient client, final T desired) { + T resource = null; + try { + resource = client.resource(desired).get(); + } catch (KubernetesClientException e) { + if (e.getCode() == 404) { + return Optional.empty(); + } + } + return Optional.ofNullable(resource); + } + + public static void deleteResourceIfExists( + final KubernetesClient client, final T resource, boolean forceDelete) { + try { + if (forceDelete) { + client.resource(resource).withGracePeriod(0L).delete(); + } else { + client + .resource(resource) + .withPropagationPolicy(DeletionPropagation.FOREGROUND) + .withTimeout(RECONCILER_FOREGROUND_REQUEST_TIMEOUT_SECONDS.getValue(), TimeUnit.SECONDS) + .delete(); + } + } catch (KubernetesClientException e) { + if (e.getCode() != 404) { + throw e; + } else { + log.info("Pod to delete does not exist, proceeding..."); + } + } + } + + public static T clone(T object) { + if (object == null) { + return null; + } + try { + ObjectMapper mapper = ModelUtils.objectMapper; + return (T) mapper.readValue(mapper.writeValueAsString(object), object.getClass()); + } catch (JsonProcessingException e) { + throw new IllegalStateException(e); + } + } +} diff --git a/spark-operator/src/main/java/org/apache/spark/k8s/operator/utils/SparkAppStatusRecorder.java b/spark-operator/src/main/java/org/apache/spark/k8s/operator/utils/SparkAppStatusRecorder.java new file mode 100644 index 00000000..4ca31909 --- /dev/null +++ b/spark-operator/src/main/java/org/apache/spark/k8s/operator/utils/SparkAppStatusRecorder.java @@ -0,0 +1,40 @@ +/* + * 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.spark.k8s.operator.utils; + +import java.util.List; + +import org.apache.spark.k8s.operator.SparkApplication; +import org.apache.spark.k8s.operator.context.SparkAppContext; +import org.apache.spark.k8s.operator.listeners.SparkAppStatusListener; +import org.apache.spark.k8s.operator.status.ApplicationState; +import org.apache.spark.k8s.operator.status.ApplicationStatus; + +public class SparkAppStatusRecorder + extends StatusRecorder { + public SparkAppStatusRecorder(List statusListeners) { + super(statusListeners, ApplicationStatus.class, SparkApplication.class); + } + + public void appendNewStateAndPersist(SparkAppContext context, ApplicationState newState) { + ApplicationStatus updatedStatus = context.getResource().getStatus().appendNewState(newState); + persistStatus(context, updatedStatus); + } +} diff --git a/spark-operator/src/main/java/org/apache/spark/k8s/operator/utils/StatusRecorder.java b/spark-operator/src/main/java/org/apache/spark/k8s/operator/utils/StatusRecorder.java new file mode 100644 index 00000000..997d7093 --- /dev/null +++ b/spark-operator/src/main/java/org/apache/spark/k8s/operator/utils/StatusRecorder.java @@ -0,0 +1,149 @@ +/* + * 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.spark.k8s.operator.utils; + +import static org.apache.spark.k8s.operator.config.SparkOperatorConf.API_RETRY_ATTEMPT_AFTER_SECONDS; +import static org.apache.spark.k8s.operator.config.SparkOperatorConf.API_STATUS_PATCH_MAX_ATTEMPTS; + +import java.util.List; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.TimeUnit; + +import com.fasterxml.jackson.databind.ObjectMapper; +import com.fasterxml.jackson.databind.node.ObjectNode; +import io.fabric8.kubernetes.client.KubernetesClient; +import io.fabric8.kubernetes.client.KubernetesClientException; +import io.javaoperatorsdk.operator.processing.event.ResourceID; +import lombok.SneakyThrows; +import lombok.extern.slf4j.Slf4j; + +import org.apache.spark.k8s.operator.BaseResource; +import org.apache.spark.k8s.operator.context.BaseContext; +import org.apache.spark.k8s.operator.listeners.BaseStatusListener; +import org.apache.spark.k8s.operator.status.BaseStatus; + +/** + * + * + * 
+ * Note - this is inspired by
+ * Flink Operator Status Recorder
+ *
+ * + * Enables additional (extendable) observers for Spark App status. Cache & version locking might be + * removed in future version as batch app does not expect spec change after submitted. + */ +@Slf4j +public class StatusRecorder< + STATUS extends BaseStatus, + CR extends BaseResource, + LISTENER extends BaseStatusListener> { + protected final List statusListeners; + protected final ObjectMapper objectMapper = ModelUtils.objectMapper; + protected final Class statusClass; + protected final Class resourceClass; + protected final ConcurrentHashMap statusCache; + + protected StatusRecorder( + List statusListeners, Class statusClass, Class resourceClass) { + this.statusListeners = statusListeners; + this.statusClass = statusClass; + this.resourceClass = resourceClass; + this.statusCache = new ConcurrentHashMap<>(); + } + + /** + * Update the status of the provided kubernetes resource on the k8s cluster. We use patch together + * with null resourceVersion to try to guarantee that the status update succeeds even if the + * underlying resource spec was update in the meantime. This is necessary for the correct operator + * behavior. + * + * @param resource Resource for which status update should be performed + */ + @SneakyThrows + private void patchAndStatusWithVersionLocked(CR resource, KubernetesClient client) { + ObjectNode newStatusNode = objectMapper.convertValue(resource.getStatus(), ObjectNode.class); + ResourceID resourceId = ResourceID.fromResource(resource); + ObjectNode previousStatusNode = statusCache.get(resourceId); + + if (newStatusNode.equals(previousStatusNode)) { + log.debug("No status change."); + return; + } + + STATUS prevStatus = objectMapper.convertValue(previousStatusNode, statusClass); + + Exception err = null; + long maxRetry = API_STATUS_PATCH_MAX_ATTEMPTS.getValue(); + for (long i = 0; i < maxRetry; i++) { + // We retry the status update 3 times to avoid some intermittent connectivity errors + try { + CR updated = client.resource(resource).lockResourceVersion().updateStatus(); + resource.getMetadata().setResourceVersion(updated.getMetadata().getResourceVersion()); + err = null; + } catch (KubernetesClientException e) { + log.error("Error while patching status, retrying {}/{}...", (i + 1), maxRetry, e); + Thread.sleep(TimeUnit.SECONDS.toMillis(API_RETRY_ATTEMPT_AFTER_SECONDS.getValue())); + err = e; + } + } + + if (err != null) { + throw err; + } + + statusCache.put(resourceId, newStatusNode); + statusListeners.forEach( + listener -> { + listener.listenStatus(resource, prevStatus, resource.getStatus()); + }); + } + + public void persistStatus(BaseContext context, STATUS newStatus) { + context.getResource().setStatus(newStatus); + patchAndStatusWithVersionLocked(context.getResource(), context.getClient()); + } + + /** + * Update the custom resource status based on the in-memory cached to ensure that any status + * updates that we made previously are always visible in the reconciliation loop. This is required + * due to our custom status patching logic. + * + *

If the cache doesn't have a status stored, we do no update. This happens when the operator + * reconciles a resource for the first time after a restart. + * + * @param resource Resource for which the status should be updated from the cache + */ + public void updateStatusFromCache(CR resource) { + ResourceID key = ResourceID.fromResource(resource); + Object cachedStatus = statusCache.get(key); + if (cachedStatus != null) { + resource.setStatus(objectMapper.convertValue(cachedStatus, statusClass)); + } else { + // Initialize cache with current status copy + statusCache.put(key, objectMapper.convertValue(resource.getStatus(), ObjectNode.class)); + } + } + + /** Remove cached status */ + public void removeCachedStatus(CR resource) { + statusCache.remove(ResourceID.fromResource(resource)); + } +} diff --git a/spark-operator/src/test/java/org/apache/spark/k8s/operator/metrics/healthcheck/SentinelManagerTest.java b/spark-operator/src/test/java/org/apache/spark/k8s/operator/metrics/healthcheck/SentinelManagerTest.java new file mode 100644 index 00000000..da1ffafd --- /dev/null +++ b/spark-operator/src/test/java/org/apache/spark/k8s/operator/metrics/healthcheck/SentinelManagerTest.java @@ -0,0 +1,216 @@ +/* + * 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.spark.k8s.operator.metrics.healthcheck; + +import static org.apache.spark.k8s.operator.utils.TestUtils.createMockApp; +import static org.apache.spark.k8s.operator.utils.TestUtils.notTimedOut; +import static org.mockito.Mockito.mockStatic; + +import java.time.Duration; +import java.util.Collections; +import java.util.HashMap; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.concurrent.TimeUnit; + +import edu.umd.cs.findbugs.annotations.SuppressFBWarnings; +import io.fabric8.kubernetes.api.model.KubernetesResourceList; +import io.fabric8.kubernetes.client.KubernetesClient; +import io.fabric8.kubernetes.client.dsl.NonNamespaceOperation; +import io.fabric8.kubernetes.client.dsl.Resource; +import io.fabric8.kubernetes.client.server.mock.EnableKubernetesMockClient; +import io.fabric8.kubernetes.client.server.mock.KubernetesMockServer; +import io.javaoperatorsdk.operator.processing.event.ResourceID; +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.BeforeAll; +import org.junit.jupiter.api.MethodOrderer; +import org.junit.jupiter.api.Order; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.TestMethodOrder; +import org.mockito.MockedStatic; + +import org.apache.spark.k8s.operator.Constants; +import org.apache.spark.k8s.operator.SparkApplication; +import org.apache.spark.k8s.operator.config.SparkOperatorConf; +import org.apache.spark.k8s.operator.config.SparkOperatorConfManager; +import org.apache.spark.k8s.operator.utils.ReconcilerUtils; +import org.apache.spark.k8s.operator.utils.Utils; + +@EnableKubernetesMockClient(crud = true) +@TestMethodOrder(MethodOrderer.OrderAnnotation.class) +@SuppressFBWarnings( + value = {"UWF_UNWRITTEN_FIELD", "NP_UNWRITTEN_FIELD", "UUF_UNUSED_FIELD"}, + justification = "Unwritten fields are covered by Kubernetes mock client") +class SentinelManagerTest { + private static final String DEFAULT = "default"; + private static final String SPARK_DEMO = "spark-demo"; + KubernetesClient kubernetesClient; + KubernetesMockServer server; + public static final int SENTINEL_RESOURCE_RECONCILIATION_DELAY_SECONDS = 10; + + @BeforeAll + static void beforeAll() { + Map overrideValue = + Collections.singletonMap( + SparkOperatorConf.SENTINEL_RESOURCE_RECONCILIATION_DELAY.getKey(), + Duration.ofSeconds(SENTINEL_RESOURCE_RECONCILIATION_DELAY_SECONDS).toString()); + SparkOperatorConfManager.INSTANCE.refresh(overrideValue); + } + + @Test + @Order(1) + void testIsSentinelResource() { + SparkApplication sparkApplication = new SparkApplication(); + Map lableMap = sparkApplication.getMetadata().getLabels(); + lableMap.put(Constants.LABEL_SENTINEL_RESOURCE, "true"); + Set namespaces = new HashSet<>(); + sparkApplication.getMetadata().setNamespace("spark-test"); + namespaces.add("spark-test"); + try (MockedStatic mockUtils = mockStatic(Utils.class)) { + mockUtils.when(Utils::getWatchedNamespaces).thenReturn(namespaces); + Assertions.assertTrue(SentinelManager.isSentinelResource(sparkApplication)); + } + } + + @Test + @Order(3) + void testHandleSentinelResourceReconciliation() throws InterruptedException { + // Reduce the SENTINEL_RESOURCE_RECONCILIATION_DELAY time to 0 + SparkOperatorConfManager.INSTANCE.refresh( + Collections.singletonMap( + SparkOperatorConf.SENTINEL_RESOURCE_RECONCILIATION_DELAY.getKey(), "10")); + + // Before Spark Reconciler Started + SparkApplication mockApp = createMockApp(DEFAULT); + kubernetesClient.resource(ReconcilerUtils.clone(mockApp)).create(); + KubernetesResourceList crList = + kubernetesClient.resources(SparkApplication.class).inNamespace(DEFAULT).list(); + SparkApplication sparkApplication = crList.getItems().get(0); + Long generation = sparkApplication.getMetadata().getGeneration(); + Assertions.assertEquals(generation, 1L); + + // Spark Reconciler Handle Sentinel Resources at the first time + SentinelManager sentinelManager = new SentinelManager(); + sentinelManager.handleSentinelResourceReconciliation(sparkApplication, kubernetesClient); + KubernetesResourceList crList2 = + kubernetesClient.resources(SparkApplication.class).inNamespace(DEFAULT).list(); + SparkApplication sparkApplication2 = crList2.getItems().get(0); + Map sparkConf2 = new HashMap<>(sparkApplication2.getSpec().getSparkConf()); + long generation2 = sparkApplication2.getMetadata().getGeneration(); + + Assertions.assertEquals(sparkConf2.get(Constants.SENTINEL_RESOURCE_DUMMY_FIELD), "1"); + Assertions.assertEquals(generation2, 2L); + SentinelManager.SentinelResourceState state2 = + (SentinelManager.SentinelResourceState) + sentinelManager.getSentinelResources().get(ResourceID.fromResource(mockApp)); + long previousGeneration2 = state2.previousGeneration; + Assertions.assertTrue(sentinelManager.allSentinelsAreHealthy()); + Assertions.assertEquals(previousGeneration2, 1L); + + Thread.sleep(Duration.ofSeconds(SENTINEL_RESOURCE_RECONCILIATION_DELAY_SECONDS * 2).toMillis()); + List crList3 = + kubernetesClient.resources(SparkApplication.class).inNamespace(DEFAULT).list().getItems(); + SparkApplication sparkApplication3 = crList3.get(0); + Map sparkConf3 = new HashMap<>(sparkApplication3.getSpec().getSparkConf()); + // Spark Sentinel Applications' s k8s generation should change + Assertions.assertNotEquals(sparkApplication3.getMetadata().getGeneration(), generation2); + // Spark conf SPARK_CONF_SENTINEL_DUMMY_FIELD values should increase + Assertions.assertNotEquals( + sparkConf2.get(Constants.SENTINEL_RESOURCE_DUMMY_FIELD), + sparkConf3.get(Constants.SENTINEL_RESOURCE_DUMMY_FIELD)); + SentinelManager.SentinelResourceState state3 = + (SentinelManager.SentinelResourceState) + sentinelManager.getSentinelResources().get(ResourceID.fromResource(mockApp)); + Assertions.assertEquals(state3.previousGeneration, previousGeneration2); + // Given the 2 * SENTINEL_RESOURCE_RECONCILIATION_DELAY_SECONDS, the reconcile method is + // not called to handleSentinelResourceReconciliation to update + Assertions.assertFalse(sentinelManager.allSentinelsAreHealthy()); + + sentinelManager.handleSentinelResourceReconciliation(sparkApplication2, kubernetesClient); + sentinelManager.handleSentinelResourceReconciliation(sparkApplication2, kubernetesClient); + boolean isHealthy; + long currentTimeInMills = System.currentTimeMillis(); + do { + isHealthy = sentinelManager.allSentinelsAreHealthy(); + } while (!isHealthy + && notTimedOut( + currentTimeInMills, + TimeUnit.MILLISECONDS.convert( + Duration.ofSeconds(SENTINEL_RESOURCE_RECONCILIATION_DELAY_SECONDS)))); + Assertions.assertTrue(isHealthy); + kubernetesClient.resources(SparkApplication.class).inNamespace(DEFAULT).delete(); + } + + @Test + @Order(2) + void sentinelManagerShouldReportHealthyWhenWatchedNamespaceIsReduced() + throws InterruptedException { + Set namespaces = new HashSet<>(); + namespaces.add(DEFAULT); + namespaces.add(SPARK_DEMO); + + try (MockedStatic mockUtils = mockStatic(Utils.class)) { + mockUtils.when(Utils::getWatchedNamespaces).thenReturn(namespaces); + SentinelManager sentinelManager = new SentinelManager(); + NonNamespaceOperation< + SparkApplication, + KubernetesResourceList, + Resource> + cr1 = kubernetesClient.resources(SparkApplication.class).inNamespace(DEFAULT); + NonNamespaceOperation< + SparkApplication, + KubernetesResourceList, + Resource> + cr2 = kubernetesClient.resources(SparkApplication.class).inNamespace(SPARK_DEMO); + + SparkApplication mockApp1 = createMockApp(DEFAULT); + SparkApplication mockApp2 = createMockApp(SPARK_DEMO); + cr1.create(mockApp1); + cr2.create(mockApp2); + + KubernetesResourceList crList1 = + kubernetesClient.resources(SparkApplication.class).inNamespace(DEFAULT).list(); + KubernetesResourceList crList2 = + kubernetesClient.resources(SparkApplication.class).inNamespace(SPARK_DEMO).list(); + SparkApplication sparkApplication1 = crList1.getItems().get(0); + SparkApplication sparkApplication2 = crList2.getItems().get(0); + sentinelManager.handleSentinelResourceReconciliation(sparkApplication1, kubernetesClient); + sentinelManager.handleSentinelResourceReconciliation(sparkApplication2, kubernetesClient); + Assertions.assertEquals( + sentinelManager.getSentinelResources().size(), + 2, + "Sentinel Manager should watch on resources in two namespaces"); + Assertions.assertTrue( + sentinelManager.allSentinelsAreHealthy(), "Sentinel Manager should report healthy"); + namespaces.remove(SPARK_DEMO); + Thread.sleep(Duration.ofSeconds(SENTINEL_RESOURCE_RECONCILIATION_DELAY_SECONDS).toMillis()); + Assertions.assertTrue( + sentinelManager.allSentinelsAreHealthy(), + "Sentinel Manager should report healthy after one namespace is " + + "removed from the watch"); + Assertions.assertEquals( + sentinelManager.getSentinelResources().size(), + 1, + "Sentinel Manager should only watch on one namespace"); + } + } +} diff --git a/spark-operator/src/test/java/org/apache/spark/k8s/operator/probe/HealthProbeTest.java b/spark-operator/src/test/java/org/apache/spark/k8s/operator/probe/HealthProbeTest.java new file mode 100644 index 00000000..81341def --- /dev/null +++ b/spark-operator/src/test/java/org/apache/spark/k8s/operator/probe/HealthProbeTest.java @@ -0,0 +1,205 @@ +/* + * 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.spark.k8s.operator.probe; + +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; + +import java.util.Arrays; +import java.util.Collections; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.concurrent.atomic.AtomicBoolean; + +import edu.umd.cs.findbugs.annotations.SuppressFBWarnings; +import io.fabric8.kubernetes.client.KubernetesClient; +import io.fabric8.kubernetes.client.server.mock.EnableKubernetesMockClient; +import io.javaoperatorsdk.operator.Operator; +import io.javaoperatorsdk.operator.RuntimeInfo; +import io.javaoperatorsdk.operator.api.config.ResourceConfiguration; +import io.javaoperatorsdk.operator.health.InformerHealthIndicator; +import io.javaoperatorsdk.operator.health.InformerWrappingEventSourceHealthIndicator; +import io.javaoperatorsdk.operator.health.Status; +import org.junit.jupiter.api.BeforeAll; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; + +import org.apache.spark.k8s.operator.metrics.healthcheck.SentinelManager; + +@EnableKubernetesMockClient(crud = true) +@SuppressFBWarnings( + value = {"UWF_UNWRITTEN_FIELD", "NP_UNWRITTEN_FIELD", "UUF_UNUSED_FIELD"}, + justification = "Unwritten fields are covered by Kubernetes mock client") +class HealthProbeTest { + private static Operator operator; + private static Operator sparkConfMonitor; + private static List operators; + private KubernetesClient kubernetesClient; + private AtomicBoolean isRunning; + private AtomicBoolean isRunning2; + private Map> + unhealthyEventSources = new HashMap<>(); + private Map> + unhealthyEventSources2 = new HashMap<>(); + + @BeforeAll + public static void beforeAll() { + operator = mock(Operator.class); + sparkConfMonitor = mock(Operator.class); + operators = Arrays.asList(operator, sparkConfMonitor); + } + + @BeforeEach + public void beforeEach() { + isRunning = new AtomicBoolean(false); + isRunning2 = new AtomicBoolean(false); + RuntimeInfo runtimeInfo = + new RuntimeInfo( + new Operator(overrider -> overrider.withKubernetesClient(kubernetesClient))) { + @Override + public boolean isStarted() { + return isRunning.get(); + } + + @Override + public Map> + unhealthyInformerWrappingEventSourceHealthIndicator() { + return unhealthyEventSources; + } + }; + + RuntimeInfo runtimeInfo2 = + new RuntimeInfo( + new Operator(overrider -> overrider.withKubernetesClient(kubernetesClient))) { + @Override + public boolean isStarted() { + return isRunning2.get(); + } + + @Override + public Map> + unhealthyInformerWrappingEventSourceHealthIndicator() { + return unhealthyEventSources2; + } + }; + + when(operator.getRuntimeInfo()).thenReturn(runtimeInfo); + when(sparkConfMonitor.getRuntimeInfo()).thenReturn(runtimeInfo2); + } + + @Test + void testHealthProbeWithInformerHealthWithMultiOperators() { + HealthProbe healthyProbe = new HealthProbe(operators, Collections.emptyList()); + isRunning.set(true); + assertFalse( + healthyProbe.isHealthy(), + "Healthy Probe should fail when the spark conf monitor operator is not running"); + isRunning2.set(true); + assertTrue( + healthyProbe.isHealthy(), "Healthy Probe should pass when both operators are running"); + + unhealthyEventSources2.put( + "c1", Map.of("e1", informerHealthIndicator(Map.of("i1", Status.UNHEALTHY)))); + assertFalse( + healthyProbe.isHealthy(), + "Healthy Probe should fail when monitor's informer health is not healthy"); + unhealthyEventSources2.clear(); + assertTrue(healthyProbe.isHealthy(), "Healthy Probe should pass"); + } + + @Test + void testHealthProbeWithInformerHealthWithSingleOperator() { + HealthProbe healthyProbe = + new HealthProbe(Collections.singletonList(operator), Collections.emptyList()); + assertFalse(healthyProbe.isHealthy(), "Health Probe should fail when operator is not running"); + isRunning.set(true); + unhealthyEventSources.put( + "c1", Map.of("e1", informerHealthIndicator(Map.of("i1", Status.UNHEALTHY)))); + assertFalse( + healthyProbe.isHealthy(), "Healthy Probe should fail when informer health is not healthy"); + unhealthyEventSources.clear(); + assertTrue(healthyProbe.isHealthy(), "Healthy Probe should pass"); + } + + @Test + void testHealthProbeWithSentinelHealthWithMultiOperators() { + SentinelManager sentinelManager = mock(SentinelManager.class); + HealthProbe healthyProbe = + new HealthProbe(operators, Collections.singletonList(sentinelManager)); + isRunning.set(true); + isRunning2.set(true); + when(sentinelManager.allSentinelsAreHealthy()).thenReturn(false); + assertFalse( + healthyProbe.isHealthy(), "Healthy Probe should fail when sentinels report failures"); + + when(sentinelManager.allSentinelsAreHealthy()).thenReturn(true); + assertTrue(healthyProbe.isHealthy(), "Healthy Probe should pass"); + } + + private static InformerWrappingEventSourceHealthIndicator informerHealthIndicator( + Map informerStatuses) { + Map informers = new HashMap<>(); + informerStatuses.forEach( + (n, s) -> + informers.put( + n, + new InformerHealthIndicator() { + @Override + public boolean hasSynced() { + return false; + } + + @Override + public boolean isWatching() { + return false; + } + + @Override + public boolean isRunning() { + return false; + } + + @Override + public Status getStatus() { + return s; + } + + @Override + public String getTargetNamespace() { + return null; + } + })); + + return new InformerWrappingEventSourceHealthIndicator() { + @Override + public Map informerHealthIndicators() { + return informers; + } + + @Override + public ResourceConfiguration getInformerConfiguration() { + return null; + } + }; + } +} diff --git a/spark-operator/src/test/java/org/apache/spark/k8s/operator/probe/ProbeServiceTest.java b/spark-operator/src/test/java/org/apache/spark/k8s/operator/probe/ProbeServiceTest.java new file mode 100644 index 00000000..c7915b2d --- /dev/null +++ b/spark-operator/src/test/java/org/apache/spark/k8s/operator/probe/ProbeServiceTest.java @@ -0,0 +1,132 @@ +/* + * 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.spark.k8s.operator.probe; + +import static org.apache.spark.k8s.operator.config.SparkOperatorConf.OPERATOR_PROBE_PORT; +import static org.apache.spark.k8s.operator.probe.ProbeService.HEALTHZ; +import static org.apache.spark.k8s.operator.probe.ProbeService.READYZ; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; + +import java.net.HttpURLConnection; +import java.net.URL; +import java.util.Collections; +import java.util.HashMap; +import java.util.List; + +import io.fabric8.kubernetes.client.KubernetesClient; +import io.fabric8.kubernetes.client.server.mock.EnableKubernetesMockClient; +import io.javaoperatorsdk.operator.Operator; +import io.javaoperatorsdk.operator.RuntimeInfo; +import org.junit.jupiter.api.Test; + +import org.apache.spark.k8s.operator.metrics.healthcheck.SentinelManager; + +@SuppressWarnings("PMD.JUnitTestsShouldIncludeAssert") +@EnableKubernetesMockClient +class ProbeServiceTest { + @Test + void testHealthProbeEndpointWithStaticProperties() throws Exception { + Operator operator = mock(Operator.class); + RuntimeInfo runtimeInfo = mock(RuntimeInfo.class); + when(operator.getRuntimeInfo()).thenReturn(runtimeInfo); + when(runtimeInfo.isStarted()).thenReturn(true).thenReturn(true); + SentinelManager sentinelManager = mock(SentinelManager.class); + when(runtimeInfo.unhealthyInformerWrappingEventSourceHealthIndicator()) + .thenReturn(new HashMap<>()); + when(sentinelManager.allSentinelsAreHealthy()).thenReturn(true); + ProbeService probeService = + new ProbeService( + Collections.singletonList(operator), Collections.singletonList(sentinelManager), null); + probeService.start(); + hitHealthyEndpoint(); + probeService.stop(); + } + + @Test + void testHealthProbeEndpointWithDynamicProperties() throws Exception { + Operator operator = mock(Operator.class); + Operator operator1 = mock(Operator.class); + RuntimeInfo runtimeInfo = mock(RuntimeInfo.class); + RuntimeInfo runtimeInfo1 = mock(RuntimeInfo.class); + when(operator.getRuntimeInfo()).thenReturn(runtimeInfo); + when(operator1.getRuntimeInfo()).thenReturn(runtimeInfo1); + + when(runtimeInfo.isStarted()).thenReturn(true).thenReturn(true); + when(runtimeInfo1.isStarted()).thenReturn(true).thenReturn(true); + + SentinelManager sentinelManager = mock(SentinelManager.class); + when(runtimeInfo.unhealthyInformerWrappingEventSourceHealthIndicator()) + .thenReturn(new HashMap<>()); + when(runtimeInfo1.unhealthyInformerWrappingEventSourceHealthIndicator()) + .thenReturn(new HashMap<>()); + when(sentinelManager.allSentinelsAreHealthy()).thenReturn(true); + ProbeService probeService = + new ProbeService( + List.of(operator, operator1), Collections.singletonList(sentinelManager), null); + probeService.start(); + hitHealthyEndpoint(); + probeService.stop(); + } + + @Test + void testReadinessProbeEndpointWithDynamicProperties() throws Exception { + Operator operator = mock(Operator.class); + Operator operator1 = mock(Operator.class); + RuntimeInfo runtimeInfo = mock(RuntimeInfo.class); + RuntimeInfo runtimeInfo1 = mock(RuntimeInfo.class); + when(operator.getRuntimeInfo()).thenReturn(runtimeInfo); + when(operator1.getRuntimeInfo()).thenReturn(runtimeInfo1); + + when(runtimeInfo.isStarted()).thenReturn(true).thenReturn(true); + when(runtimeInfo1.isStarted()).thenReturn(true).thenReturn(true); + + SentinelManager sentinelManager = mock(SentinelManager.class); + KubernetesClient client = mock(KubernetesClient.class); + when(runtimeInfo.unhealthyInformerWrappingEventSourceHealthIndicator()) + .thenReturn(new HashMap<>()); + when(runtimeInfo1.unhealthyInformerWrappingEventSourceHealthIndicator()) + .thenReturn(new HashMap<>()); + when(operator1.getKubernetesClient()).thenReturn(client); + ProbeService probeService = + new ProbeService( + List.of(operator, operator1), Collections.singletonList(sentinelManager), null); + probeService.start(); + hitStartedUpEndpoint(); + probeService.stop(); + } + + private void hitHealthyEndpoint() throws Exception { + URL u = new URL("http://localhost:" + OPERATOR_PROBE_PORT.getValue() + HEALTHZ); + HttpURLConnection connection = (HttpURLConnection) u.openConnection(); + connection.setConnectTimeout(100000); + connection.connect(); + assertEquals(connection.getResponseCode(), 200, "Health Probe should return 200"); + } + + private void hitStartedUpEndpoint() throws Exception { + URL u = new URL("http://localhost:" + OPERATOR_PROBE_PORT.getValue() + READYZ); + HttpURLConnection connection = (HttpURLConnection) u.openConnection(); + connection.setConnectTimeout(100000); + connection.connect(); + assertEquals(connection.getResponseCode(), 200, "operators are not ready"); + } +} diff --git a/spark-operator/src/test/java/org/apache/spark/k8s/operator/probe/ReadinessProbeTest.java b/spark-operator/src/test/java/org/apache/spark/k8s/operator/probe/ReadinessProbeTest.java new file mode 100644 index 00000000..5a878d35 --- /dev/null +++ b/spark-operator/src/test/java/org/apache/spark/k8s/operator/probe/ReadinessProbeTest.java @@ -0,0 +1,70 @@ +/* + * 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.spark.k8s.operator.probe; + +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; + +import java.io.IOException; +import java.io.OutputStream; +import java.util.Arrays; + +import com.sun.net.httpserver.HttpExchange; +import io.fabric8.kubernetes.client.KubernetesClient; +import io.javaoperatorsdk.operator.Operator; +import io.javaoperatorsdk.operator.RuntimeInfo; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.mockito.MockedStatic; +import org.mockito.Mockito; + +import org.apache.spark.k8s.operator.utils.ProbeUtil; + +@SuppressWarnings("PMD.JUnitTestsShouldIncludeAssert") +class ReadinessProbeTest { + KubernetesClient client; + HttpExchange httpExchange; + + @BeforeEach + public void beforeEach() { + OutputStream outputStream = mock(OutputStream.class); + httpExchange = mock(HttpExchange.class); + client = mock(KubernetesClient.class); + when(httpExchange.getResponseBody()).thenReturn(outputStream); + } + + @Test + void testHandleSucceed() throws IOException { + Operator operator = mock(Operator.class); + Operator sparkConfMonitor = mock(Operator.class); + RuntimeInfo runtimeInfo = mock(RuntimeInfo.class); + RuntimeInfo sparkConfMonitorRuntimeInfo = mock(RuntimeInfo.class); + when(operator.getRuntimeInfo()).thenReturn(runtimeInfo); + when(runtimeInfo.isStarted()).thenReturn(true); + when(sparkConfMonitor.getRuntimeInfo()).thenReturn(sparkConfMonitorRuntimeInfo); + when(sparkConfMonitorRuntimeInfo.isStarted()).thenReturn(true); + when(sparkConfMonitor.getKubernetesClient()).thenReturn(client); + ReadinessProbe readinessProbe = new ReadinessProbe(Arrays.asList(operator)); + try (MockedStatic mockedStatic = Mockito.mockStatic(ProbeUtil.class)) { + readinessProbe.handle(httpExchange); + mockedStatic.verify(() -> ProbeUtil.sendMessage(httpExchange, 200, "started")); + } + } +} diff --git a/spark-operator/src/test/java/org/apache/spark/k8s/operator/reconciler/SparkAppReconcilerTest.java b/spark-operator/src/test/java/org/apache/spark/k8s/operator/reconciler/SparkAppReconcilerTest.java new file mode 100644 index 00000000..3b77e15a --- /dev/null +++ b/spark-operator/src/test/java/org/apache/spark/k8s/operator/reconciler/SparkAppReconcilerTest.java @@ -0,0 +1,116 @@ +/* + * 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.spark.k8s.operator.reconciler; + +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.doAnswer; +import static org.mockito.Mockito.doNothing; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.mockConstruction; +import static org.mockito.Mockito.when; + +import java.util.Collections; +import java.util.Optional; + +import io.fabric8.kubernetes.api.model.Pod; +import io.fabric8.kubernetes.client.KubernetesClient; +import io.javaoperatorsdk.operator.api.reconciler.Context; +import io.javaoperatorsdk.operator.api.reconciler.DeleteControl; +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.mockito.MockedConstruction; +import org.mockito.MockedStatic; +import org.mockito.Mockito; + +import org.apache.spark.k8s.operator.SparkAppSubmissionWorker; +import org.apache.spark.k8s.operator.SparkApplication; +import org.apache.spark.k8s.operator.context.SparkAppContext; +import org.apache.spark.k8s.operator.metrics.healthcheck.SentinelManager; +import org.apache.spark.k8s.operator.status.ApplicationState; +import org.apache.spark.k8s.operator.status.ApplicationStateSummary; +import org.apache.spark.k8s.operator.status.ApplicationStatus; +import org.apache.spark.k8s.operator.utils.ReconcilerUtils; +import org.apache.spark.k8s.operator.utils.SparkAppStatusRecorder; + +class SparkAppReconcilerTest { + private SparkAppStatusRecorder mockRecorder = mock(SparkAppStatusRecorder.class); + private SentinelManager mockSentinelManager = mock(SentinelManager.class); + private KubernetesClient mockClient = mock(KubernetesClient.class); + private Context mockContext = mock(Context.class); + private Pod mockDriver = mock(Pod.class); + private SparkAppSubmissionWorker mockWorker = mock(SparkAppSubmissionWorker.class); + SparkApplication app = new SparkApplication(); + SparkAppReconciler reconciler = + new SparkAppReconciler(mockWorker, mockRecorder, mockSentinelManager); + + @BeforeEach + public void beforeEach() { + when(mockContext.getClient()).thenReturn(mockClient); + doNothing().when(mockRecorder).removeCachedStatus(any(SparkApplication.class)); + doAnswer( + invocation -> { + app.setStatus(invocation.getArgument(1)); + return null; + }) + .when(mockRecorder) + .persistStatus(any(SparkAppContext.class), any(ApplicationStatus.class)); + doAnswer( + invocation -> { + ApplicationStatus updatedStatus = + app.getStatus().appendNewState(invocation.getArgument(1)); + app.setStatus(updatedStatus); + return null; + }) + .when(mockRecorder) + .appendNewStateAndPersist(any(SparkAppContext.class), any(ApplicationState.class)); + } + + @Test + void testCleanupRunningApp() { + try (MockedConstruction mockAppContext = + mockConstruction( + SparkAppContext.class, + (mock, context) -> { + when(mock.getResource()).thenReturn(app); + when(mock.getClient()).thenReturn(mockClient); + when(mock.getDriverPod()).thenReturn(Optional.of(mockDriver)); + when(mock.getDriverPodSpec()).thenReturn(mockDriver); + when(mock.getDriverPreResourcesSpec()).thenReturn(Collections.emptyList()); + when(mock.getDriverResourcesSpec()).thenReturn(Collections.emptyList()); + }); + MockedStatic utils = Mockito.mockStatic(ReconcilerUtils.class)) { + // delete running app + app.setStatus( + app.getStatus() + .appendNewState(new ApplicationState(ApplicationStateSummary.RunningHealthy, ""))); + DeleteControl deleteControl = reconciler.cleanup(app, mockContext); + Assertions.assertFalse(deleteControl.isRemoveFinalizer()); + utils.verify(() -> ReconcilerUtils.deleteResourceIfExists(mockClient, mockDriver, false)); + Assertions.assertEquals( + ApplicationStateSummary.ResourceReleased, + app.getStatus().getCurrentState().getCurrentStateSummary()); + + // proceed delete for terminated app + deleteControl = reconciler.cleanup(app, mockContext); + Assertions.assertTrue(deleteControl.isRemoveFinalizer()); + } + } +} diff --git a/spark-operator/src/test/java/org/apache/spark/k8s/operator/reconciler/SparkAppResourceSpecFactoryTest.java b/spark-operator/src/test/java/org/apache/spark/k8s/operator/reconciler/SparkAppResourceSpecFactoryTest.java new file mode 100644 index 00000000..2bbf8fa7 --- /dev/null +++ b/spark-operator/src/test/java/org/apache/spark/k8s/operator/reconciler/SparkAppResourceSpecFactoryTest.java @@ -0,0 +1,72 @@ +/* + * 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.spark.k8s.operator.reconciler; + +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.eq; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +import java.util.Map; + +import io.fabric8.kubernetes.api.model.ObjectMeta; +import io.fabric8.kubernetes.api.model.ObjectMetaBuilder; +import io.fabric8.kubernetes.api.model.Pod; +import io.fabric8.kubernetes.client.KubernetesClient; +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.Test; +import org.mockito.ArgumentCaptor; + +import org.apache.spark.k8s.operator.SparkAppResourceSpec; +import org.apache.spark.k8s.operator.SparkAppSubmissionWorker; +import org.apache.spark.k8s.operator.SparkApplication; + +class SparkAppResourceSpecFactoryTest { + + @Test + void testBuildResourceSpecCoversBasicOverride() { + SparkApplication app = new SparkApplication(); + app.setMetadata( + new ObjectMetaBuilder().withNamespace("foo").withName("bar-app").withUid("uid").build()); + KubernetesClient mockClient = mock(KubernetesClient.class); + Pod mockDriver = mock(Pod.class); + when(mockDriver.getMetadata()).thenReturn(new ObjectMeta()); + SparkAppResourceSpec mockSpec = mock(SparkAppResourceSpec.class); + when(mockSpec.getConfiguredPod()).thenReturn(mockDriver); + ArgumentCaptor> captor = ArgumentCaptor.forClass(Map.class); + SparkAppSubmissionWorker mockWorker = mock(SparkAppSubmissionWorker.class); + when(mockWorker.getResourceSpec(any(), any(), captor.capture())).thenReturn(mockSpec); + SparkAppResourceSpec spec = + SparkAppResourceSpecFactory.buildResourceSpec(app, mockClient, mockWorker); + verify(mockWorker).getResourceSpec(eq(app), eq(mockClient), any()); + Map props = captor.getValue(); + Assertions.assertTrue(props.containsKey("spark.kubernetes.namespace")); + Assertions.assertEquals("foo", props.get("spark.kubernetes.namespace")); + ArgumentCaptor metaArgumentCaptor = ArgumentCaptor.forClass(ObjectMeta.class); + verify(mockDriver).setMetadata(metaArgumentCaptor.capture()); + Assertions.assertEquals(mockSpec, spec); + ObjectMeta metaOverride = metaArgumentCaptor.getValue(); + Assertions.assertEquals(1, metaOverride.getOwnerReferences().size()); + Assertions.assertEquals("bar-app", metaOverride.getOwnerReferences().get(0).getName()); + Assertions.assertEquals("uid", metaOverride.getOwnerReferences().get(0).getUid()); + Assertions.assertEquals(app.getKind(), metaOverride.getOwnerReferences().get(0).getKind()); + } +} diff --git a/spark-operator/src/test/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/AppCleanUpStepTest.java b/spark-operator/src/test/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/AppCleanUpStepTest.java new file mode 100644 index 00000000..076295ea --- /dev/null +++ b/spark-operator/src/test/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/AppCleanUpStepTest.java @@ -0,0 +1,45 @@ +/* + * 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.spark.k8s.operator.reconciler.reconcilesteps; + +import static org.junit.jupiter.api.Assertions.*; + +import java.time.Instant; + +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.Test; + +import org.apache.spark.k8s.operator.SparkApplication; + +class AppCleanUpStepTest { + @Test + void enableForceDelete() { + AppCleanUpStep appCleanUpStep = new AppCleanUpStep(); + SparkApplication app = new SparkApplication(); + app.getStatus() + .getCurrentState() + .setLastTransitionTime(Instant.now().minusSeconds(5).toString()); + app.getSpec() + .getApplicationTolerations() + .getApplicationTimeoutConfig() + .setForceTerminationGracePeriodMillis(3000L); + Assertions.assertTrue(appCleanUpStep.enableForceDelete(app)); + } +}