[SPARK-48382] Add reconciler to spark-operator module

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<CR extends BaseResource<?, ?, ?, ?, ?>> {
+  public abstract CR getResource();
+
+  public abstract KubernetesClient getClient();
+}

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<SparkApplication> {
+  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<Pod> getDriverPod() {
+    return josdkContext
+        .getSecondaryResourcesAsStream(Pod.class)
+        .filter(
+            p ->
+                p.getMetadata()
+                    .getLabels()
+                    .entrySet()
+                    .containsAll(driverLabels(sparkApplication).entrySet()))
+        .findAny();
+  }
+
+  public Set<Pod> 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<HasMetadata> getDriverPreResourcesSpec() {
+    return getSecondaryResourceSpec().getDriverPreResources();
+  }
+
+  public Pod getDriverPodSpec() {
+    return getSecondaryResourceSpec().getConfiguredPod();
+  }
+
+  public List<HasMetadata> getDriverResourcesSpec() {
+    return getSecondaryResourceSpec().getDriverResources();
+  }
+}

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 extends HasMetadata> 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;
+  }
+}

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 extends HasMetadata> 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;
+  }
+}