[NOSQUASH] Resync with Kubernetes

common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/kubernetes/KubernetesExternalShuffleClient.java

@@ -17,64 +17,13 @@
 
 package org.apache.spark.network.shuffle.kubernetes;
 
-import org.apache.spark.network.client.RpcResponseCallback;
-import org.apache.spark.network.client.TransportClient;
-import org.apache.spark.network.sasl.SecretKeyHolder;
-import org.apache.spark.network.shuffle.ExternalShuffleClient;
-import org.apache.spark.network.shuffle.protocol.RegisterDriver;
-import org.apache.spark.network.util.TransportConf;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
+import java.io.Closeable;
 import java.io.IOException;
-import java.nio.ByteBuffer;
-
-/**
- * A client for talking to the external shuffle service in Kubernetes cluster mode.
- *
- * This is used by the each Spark executor to register with a corresponding external
- * shuffle service on the cluster. The purpose is for cleaning up shuffle files
- * reliably if the application exits unexpectedly.
- */
-public class KubernetesExternalShuffleClient extends ExternalShuffleClient {
-  private static final Logger logger = LoggerFactory
-          .getLogger(KubernetesExternalShuffleClient.class);
-
-  /**
-   * Creates an Kubernetes external shuffle client that wraps the {@link ExternalShuffleClient}.
-   * Please refer to docs on {@link ExternalShuffleClient} for more information.
-   */
-  public KubernetesExternalShuffleClient(
-      TransportConf conf,
-      SecretKeyHolder secretKeyHolder,
-      boolean authEnabled,
-      long registrationTimeoutMs) {
-    super(conf, secretKeyHolder, authEnabled, registrationTimeoutMs);
-  }
-
-  public void registerDriverWithShuffleService(String host, int port)
-      throws IOException, InterruptedException {
-    checkInit();
-    ByteBuffer registerDriver = new RegisterDriver(appId, 0).toByteBuffer();
-    TransportClient client = clientFactory.createClient(host, port);
-    client.sendRpc(registerDriver, new RegisterDriverCallback());
-  }
 
-  private class RegisterDriverCallback implements RpcResponseCallback {
-    @Override
-    public void onSuccess(ByteBuffer response) {
-      logger.info("Successfully registered app " + appId + " with external shuffle service.");
-    }
+public interface KubernetesExternalShuffleClient extends Closeable {
 
-    @Override
-    public void onFailure(Throwable e) {
-      logger.warn("Unable to register app " + appId + " with external shuffle service. " +
-          "Please manually remove shuffle data after driver exit. Error: " + e);
-    }
-  }
+  void init(String appId);
 
-  @Override
-  public void close() {
-    super.close();
-  }
+  void registerDriverWithShuffleService(String host, int port)
+      throws IOException, InterruptedException;
 }

common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/kubernetes/KubernetesExternalShuffleClientImpl.java

@@ -0,0 +1,78 @@
+/*
+ * 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.network.shuffle.kubernetes;
+
+import org.apache.spark.network.client.RpcResponseCallback;
+import org.apache.spark.network.client.TransportClient;
+import org.apache.spark.network.sasl.SecretKeyHolder;
+import org.apache.spark.network.shuffle.ExternalShuffleClient;
+import org.apache.spark.network.shuffle.protocol.RegisterDriver;
+import org.apache.spark.network.util.TransportConf;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.nio.ByteBuffer;
+
+/**
+ * A client for talking to the external shuffle service in Kubernetes cluster mode.
+ *
+ * This is used by the each Spark executor to register with a corresponding external
+ * shuffle service on the cluster. The purpose is for cleaning up shuffle files
+ * reliably if the application exits unexpectedly.
+ */
+public class KubernetesExternalShuffleClientImpl
+    extends ExternalShuffleClient implements KubernetesExternalShuffleClient {
+
+  private static final Logger logger = LoggerFactory
+          .getLogger(KubernetesExternalShuffleClientImpl.class);
+
+  /**
+   * Creates a Kubernetes external shuffle client that wraps the {@link ExternalShuffleClient}.
+   * Please refer to docs on {@link ExternalShuffleClient} for more information.
+   */
+  public KubernetesExternalShuffleClientImpl(
+      TransportConf conf,
+      SecretKeyHolder secretKeyHolder,
+      boolean saslEnabled,
+      long registrationTimeoutMs) {
+    super(conf, secretKeyHolder, saslEnabled, registrationTimeoutMs);
+  }
+
+  @Override
+  public void registerDriverWithShuffleService(String host, int port)
+      throws IOException, InterruptedException {
+    checkInit();
+    ByteBuffer registerDriver = new RegisterDriver(appId, 0).toByteBuffer();
+    TransportClient client = clientFactory.createClient(host, port);
+    client.sendRpc(registerDriver, new RegisterDriverCallback());
+  }
+
+  private class RegisterDriverCallback implements RpcResponseCallback {
+    @Override
+    public void onSuccess(ByteBuffer response) {
+      logger.info("Successfully registered app " + appId + " with external shuffle service.");
+    }
+
+    @Override
+    public void onFailure(Throwable e) {
+      logger.warn("Unable to register app " + appId + " with external shuffle service. " +
+          "Please manually remove shuffle data after driver exit. Error: " + e);
+    }
+  }
+}

conf/k8s-shuffle-service-rbac.yaml

@@ -0,0 +1,80 @@
+#
+# 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.
+#
+
+apiVersion: v1
+kind: ServiceAccount
+metadata:
+  name: spark-shuffle-service-service-account
+  namespace: default
+  labels:
+    app: spark-shuffle-service
+    spark-version: 2.2.0
+---
+apiVersion: extensions/v1beta1
+kind: PodSecurityPolicy
+metadata:
+  name: spark-shuffle-service-pod-security-policy
+  labels:
+    app: spark-shuffle-service
+    spark-version: 2.2.0
+spec:
+  privileged: false
+  fsGroup:
+    rule: RunAsAny
+  runAsUser:
+    rule: RunAsAny
+  volumes:
+  - "hostPath"
+  - "secret"
+---
+apiVersion: rbac.authorization.k8s.io/v1beta1
+kind: ClusterRole
+metadata:
+  name: spark-shuffle-service-role
+  labels:
+    app: spark-shuffle-service
+    spark-version: 2.2.0
+rules: 
+- apiGroups:
+  - "extensions"
+  resources:
+  - "podsecuritypolicies"
+  resourceNames:
+  - "spark-shuffle-service-pod-security-policy"
+  verbs:
+  - "use"
+- apiGroups:
+  - "" # "" indicates the core API group
+  resources:
+  - "pods"
+  verbs:
+  - "get"
+  - "list"
+  - "watch"
+---
+apiVersion: rbac.authorization.k8s.io/v1beta1
+kind: ClusterRoleBinding
+metadata:
+    name: spark-shuffle-service-role-binding
+subjects:
+- kind: ServiceAccount
+  name: spark-shuffle-service-service-account
+  namespace: default
+roleRef:
+   apiGroup: rbac.authorization.k8s.io
+   kind: ClusterRole
+   name: spark-shuffle-service-role

conf/k8s-spark-rbac.yaml

@@ -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.
+#
+
+apiVersion: v1
+kind: ServiceAccount
+metadata:
+  name: spark
+  namespace: default
+---
+apiVersion: rbac.authorization.k8s.io/v1beta1
+kind: Role
+metadata:
+  namespace: default
+  name: spark-role
+rules:
+- apiGroups:
+  - "" # "" indicates the core API group
+  resources:
+  - "pods"
+  verbs:
+  - "*"
+- apiGroups:
+  - "" # "" indicates the core API group
+  resources:
+  - "services"
+  verbs:
+  - "*"
+---
+apiVersion: rbac.authorization.k8s.io/v1beta1
+kind: RoleBinding
+metadata:
+  name: spark-role-binding
+  namespace: default
+subjects:
+- kind: ServiceAccount
+  name: spark
+  namespace: default
+roleRef:
+  kind: Role
+  name: spark-role
+  apiGroup: rbac.authorization.k8s.io

conf/kubernetes-resource-staging-server.yaml

@@ -15,7 +15,7 @@
 # limitations under the License.
 #
 ---
-apiVersion: extensions/v1beta1
+apiVersion: apps/v1beta1
 kind: Deployment
 metadata:
   name: spark-resource-staging-server
@@ -32,14 +32,14 @@ spec:
             name: spark-resource-staging-server-config
       containers:
         - name: spark-resource-staging-server
-          image: kubespark/spark-resource-staging-server:v2.2.0-kubernetes-0.3.0
+          image: kubespark/spark-resource-staging-server:v2.2.0-kubernetes-0.4.0
           resources:
             requests:
               cpu: 100m
               memory: 256Mi
             limits:
               cpu: 100m
-              memory: 256Mi
+              memory: 1Gi
           volumeMounts:
             - name: resource-staging-server-properties
               mountPath: '/etc/spark-resource-staging-server'

conf/kubernetes-shuffle-service.yaml

@@ -32,16 +32,16 @@ spec:
       volumes:
         - name: temp-volume
           hostPath:
-            path: '/var/tmp' # change this path according to your cluster configuration.
+            path: '/tmp/spark-local' # change this path according to your cluster configuration.
       containers:
         - name: shuffle
           # This is an official image that is built
           # from the dockerfiles/shuffle directory
           # in the spark distribution.
-          image: kubespark/spark-shuffle:v2.2.0-kubernetes-0.3.0
+          image: kubespark/spark-shuffle:v2.2.0-kubernetes-0.4.0
           imagePullPolicy: IfNotPresent
           volumeMounts:
-            - mountPath: '/tmp'
+            - mountPath: '/tmp/spark-local'
               name: temp-volume
               # more volumes can be mounted here.
               # The spark job must be configured to use these

core/src/main/scala/org/apache/spark/deploy/SparkSubmit.scala

@@ -715,7 +715,7 @@ object SparkSubmit extends CommandLineUtils with Logging {
     }
 
     if (isKubernetesCluster) {
-      childMainClass = "org.apache.spark.deploy.kubernetes.submit.Client"
+      childMainClass = "org.apache.spark.deploy.k8s.submit.Client"
       if (args.isPython) {
         childArgs ++= Array("--primary-py-file", args.primaryResource)
         childArgs ++= Array("--main-class", "org.apache.spark.deploy.PythonRunner")

core/src/main/scala/org/apache/spark/internal/config/package.scala

@@ -262,7 +262,7 @@ package object config {
   private[spark] val DRIVER_HOST_ADDRESS = ConfigBuilder("spark.driver.host")
     .doc("Address of driver endpoints.")
     .stringConf
-    .createWithDefault(Utils.localHostName())
+    .createWithDefault(Utils.localCanonicalHostName())
 
   private[spark] val DRIVER_BIND_ADDRESS = ConfigBuilder("spark.driver.bindAddress")
     .doc("Address where to bind network listen sockets on the driver.")

core/src/main/scala/org/apache/spark/util/Utils.scala

@@ -941,6 +941,13 @@ private[spark] object Utils extends Logging {
     customHostname = Some(hostname)
   }
 
+  /**
+   * Get the local machine's FQDN.
+   */
+  def localCanonicalHostName(): String = {
+    customHostname.getOrElse(localIpAddress.getCanonicalHostName)
+  }
+
   /**
    * Get the local machine's hostname.
    */