Repository: GoogleCloudPlatform/spark-on-k8s-operator
Pull request #1205 (2021-03-22 21:17:06)

Add support to expose custom ports
I'd like to expose other ports in my spark app, similar to what the spark-operator already has done to prometheus.

Its related to: 

https://github.com/GoogleCloudPlatform/spark-on-k8s-operator/issues/1202 
https://github.com/GoogleCloudPlatform/spark-on-k8s-operator/issues/1120



google-cla[bot]: 
Thanks for your pull request. It looks like this may be your first contribution to a Google open source project (if not, look below for help). Before we can look at your pull request, you'll need to sign a Contributor License Agreement (CLA).

:memo: **Please visit <https://cla.developers.google.com/> to sign.**

Once you've signed (or fixed any issues), please reply here with `@googlebot I signed it!` and we'll verify it.

----

#### What to do if you already signed the CLA

##### Individual signers

*   It's possible we don't have your GitHub username or you're using a different email address on your commit. Check [your existing CLA data](https://cla.developers.google.com/clas) and verify that your [email is set on your git commits](https://help.github.com/articles/setting-your-email-in-git/).

##### Corporate signers

*   Your company has a Point of Contact who decides which employees are authorized to participate. Ask your POC to be added to the group of authorized contributors. If you don't know who your Point of Contact is, direct the Google project maintainer to [go/cla#troubleshoot](http://go/cla#troubleshoot) ([Public version](https://opensource.google/docs/cla/#troubleshoot)).
*   The email used to register you as an authorized contributor must be the email used for the Git commit. Check [your existing CLA data](https://cla.developers.google.com/clas) and verify that your [email is set on your git commits](https://help.github.com/articles/setting-your-email-in-git/).
*   The email used to register you as an authorized contributor must also be [attached to your GitHub account](https://github.com/settings/emails).
		

ℹ️ **Googlers: [Go here](https://goto.google.com/prinfo/https%3A%2F%2Fgithub.com%2FGoogleCloudPlatform%2Fspark-on-k8s-operator%2Fpull%2F1205) for more info**.

<!-- need_sender_cla -->
google-cla[bot]: 
Thanks for your pull request. It looks like this may be your first contribution to a Google open source project (if not, look below for help). Before we can look at your pull request, you'll need to sign a Contributor License Agreement (CLA).

:memo: **Please visit <https://cla.developers.google.com/> to sign.**

Once you've signed (or fixed any issues), please reply here with `@googlebot I signed it!` and we'll verify it.

----

#### What to do if you already signed the CLA

##### Individual signers

*   It's possible we don't have your GitHub username or you're using a different email address on your commit. Check [your existing CLA data](https://cla.developers.google.com/clas) and verify that your [email is set on your git commits](https://help.github.com/articles/setting-your-email-in-git/).

##### Corporate signers

*   Your company has a Point of Contact who decides which employees are authorized to participate. Ask your POC to be added to the group of authorized contributors. If you don't know who your Point of Contact is, direct the Google project maintainer to [go/cla#troubleshoot](http://go/cla#troubleshoot) ([Public version](https://opensource.google/docs/cla/#troubleshoot)).
*   The email used to register you as an authorized contributor must be the email used for the Git commit. Check [your existing CLA data](https://cla.developers.google.com/clas) and verify that your [email is set on your git commits](https://help.github.com/articles/setting-your-email-in-git/).
*   The email used to register you as an authorized contributor must also be [attached to your GitHub account](https://github.com/settings/emails).
		

ℹ️ **Googlers: [Go here](https://goto.google.com/prinfo/https%3A%2F%2Fgithub.com%2FGoogleCloudPlatform%2Fspark-on-k8s-operator%2Fpull%2F1205) for more info**.

<!-- need_sender_cla -->
luizm: @googlebot I signed it!

Patch:
diff --git a/docs/api-docs.md b/docs/api-docs.md
index 7a26539..3997b47 100644
--- a/docs/api-docs.md
+++ b/docs/api-docs.md
@@ -719,8 +719,7 @@ string
 </td>
 <td>
 <em>(Optional)</em>
-<p>PriorityClassName stands for the name of k8s PriorityClass resource, it&rsquo;s being used for job scheduling and 
-preemption, whether with the Volcano batch scheduler or the Kubernetes default scheduler.</p>
+<p>PriorityClassName stands for the name of k8s PriorityClass resource, it&rsquo;s being used in Volcano batch scheduler.</p>
 </td>
 </tr>
 <tr>
@@ -1047,6 +1046,20 @@ map[string]string
 executors to connect to the driver.</p>
 </td>
 </tr>
+<tr>
+<td>
+<code>ports</code></br>
+<em>
+<a href="#sparkoperator.k8s.io/v1beta2.Port">
+[]Port
+</a>
+</em>
+</td>
+<td>
+<em>(Optional)</em>
+<p>Ports settings for the pods, following the Kubernetes specifications.</p>
+</td>
+</tr>
 </tbody>
 </table>
 <h3 id="sparkoperator.k8s.io/v1beta2.DriverState">DriverState
@@ -1217,6 +1230,20 @@ bool
 Maps to <code>spark.kubernetes.executor.deleteOnTermination</code> that is available since Spark 3.0.</p>
 </td>
 </tr>
+<tr>
+<td>
+<code>ports</code></br>
+<em>
+<a href="#sparkoperator.k8s.io/v1beta2.Port">
+[]Port
+</a>
+</em>
+</td>
+<td>
+<em>(Optional)</em>
+<p>Ports settings for the pods, following the Kubernetes specifications.</p>
+</td>
+</tr>
 </tbody>
 </table>
 <h3 id="sparkoperator.k8s.io/v1beta2.ExecutorState">ExecutorState
@@ -1427,6 +1454,56 @@ string
 </tr>
 </tbody>
 </table>
+<h3 id="sparkoperator.k8s.io/v1beta2.Port">Port
+</h3>
+<p>
+(<em>Appears on:</em>
+<a href="#sparkoperator.k8s.io/v1beta2.DriverSpec">DriverSpec</a>, 
+<a href="#sparkoperator.k8s.io/v1beta2.ExecutorSpec">ExecutorSpec</a>)
+</p>
+<p>
+<p>Port represents the port definition in the pods objects.</p>
+</p>
+<table>
+<thead>
+<tr>
+<th>Field</th>
+<th>Description</th>
+</tr>
+</thead>
+<tbody>
+<tr>
+<td>
+<code>name</code></br>
+<em>
+string
+</em>
+</td>
+<td>
+</td>
+</tr>
+<tr>
+<td>
+<code>protocol</code></br>
+<em>
+string
+</em>
+</td>
+<td>
+</td>
+</tr>
+<tr>
+<td>
+<code>containerPort</code></br>
+<em>
+int32
+</em>
+</td>
+<td>
+</td>
+</tr>
+</tbody>
+</table>
 <h3 id="sparkoperator.k8s.io/v1beta2.PrometheusSpec">PrometheusSpec
 </h3>
 <p>
@@ -1471,6 +1548,19 @@ If not specified, 8090 will be used as the default.</p>
 </tr>
 <tr>
 <td>
+<code>portName</code></br>
+<em>
+string
+</em>
+</td>
+<td>
+<em>(Optional)</em>
+<p>PortName is the port name of prometheus JMX exporter port.
+If not specified, jmx-exporter will be used as the default.</p>
+</td>
+</tr>
+<tr>
+<td>
 <code>configFile</code></br>
 <em>
 string
@@ -2828,6 +2918,20 @@ TargetPort should be the same as the one defined in spark.ui.port</p>
 </tr>
 <tr>
 <td>
+<code>serviceType</code></br>
+<em>
+<a href="https://kubernetes.io/docs/reference/generated/kubernetes-api/v1.13/#servicetype-v1-core">
+Kubernetes core/v1.ServiceType
+</a>
+</em>
+</td>
+<td>
+<em>(Optional)</em>
+<p>ServiceType allows configuring the type of the service. Defaults to ClusterIP.</p>
+</td>
+</tr>
+<tr>
+<td>
 <code>ingressAnnotations</code></br>
 <em>
 map[string]string
@@ -2857,5 +2961,5 @@ map[string]string
 <hr/>
 <p><em>
 Generated with <code>gen-crd-api-reference-docs</code>
-on git commit <code>cf34460</code>.
+on git commit <code>2f6a3f9</code>.
 </em></p>
diff --git a/pkg/apis/sparkoperator.k8s.io/v1beta2/types.go b/pkg/apis/sparkoperator.k8s.io/v1beta2/types.go
index 16474c6..ea34834 100644
--- a/pkg/apis/sparkoperator.k8s.io/v1beta2/types.go
+++ b/pkg/apis/sparkoperator.k8s.io/v1beta2/types.go
@@ -549,6 +549,9 @@ type DriverSpec struct {
 	// executors to connect to the driver.
 	// +optional
 	ServiceAnnotations map[string]string `json:"serviceAnnotations,omitempty"`
+	// Ports settings for the pods, following the Kubernetes specifications.
+	// +optional
+	Ports []Port `json:"ports,omitempty"`
 }
 
 // ExecutorSpec is specification of the executor.
@@ -570,6 +573,9 @@ type ExecutorSpec struct {
 	// Maps to `spark.kubernetes.executor.deleteOnTermination` that is available since Spark 3.0.
 	// +optional
 	DeleteOnTermination *bool `json:"deleteOnTermination,omitempty"`
+	// Ports settings for the pods, following the Kubernetes specifications.
+	// +optional
+	Ports []Port `json:"ports,omitempty"`
 }
 
 // NamePath is a pair of a name and a path to which the named objects should be mounted to.
@@ -618,6 +624,13 @@ type NameKey struct {
 	Key  string `json:"key"`
 }
 
+// Port represents the port definition in the pods objects.
+type Port struct {
+	Name          string `json:"name"`
+	Protocol      string `json:"protocol"`
+	ContainerPort int32  `json:"containerPort"`
+}
+
 // MonitoringSpec defines the monitoring specification.
 type MonitoringSpec struct {
 	// ExposeDriverMetrics specifies whether to expose metrics on the driver.
diff --git a/pkg/webhook/patch.go b/pkg/webhook/patch.go
index 986baf7..14c4fe7 100644
--- a/pkg/webhook/patch.go
+++ b/pkg/webhook/patch.go
@@ -63,6 +63,7 @@ func patchSparkPod(pod *corev1.Pod, app *v1beta2.SparkApplication) []patchOperat
 	patchOps = append(patchOps, addEnvVars(pod, app)...)
 	patchOps = append(patchOps, addEnvFrom(pod, app)...)
 	patchOps = append(patchOps, addHostAliases(pod, app)...)
+	patchOps = append(patchOps, addContainerPorts(pod, app)...)
 	patchOps = append(patchOps, addPriorityClassName(pod, app)...)
 
 	op := addSchedulerName(pod, app)
@@ -367,14 +368,14 @@ func getPrometheusConfigPatches(pod *corev1.Pod, app *v1beta2.SparkApplication)
 	name := config.GetPrometheusConfigMapName(app)
 	volumeName := name + "-vol"
 	mountPath := config.PrometheusConfigMapMountPath
-	port := config.DefaultPrometheusJavaAgentPort
+	promPort := config.DefaultPrometheusJavaAgentPort
 	if app.Spec.Monitoring.Prometheus.Port != nil {
-		port = *app.Spec.Monitoring.Prometheus.Port
+		promPort = *app.Spec.Monitoring.Prometheus.Port
 	}
-	protocol := config.DefaultPrometheusPortProtocol
-	portName := config.DefaultPrometheusPortName
+	promProtocol := config.DefaultPrometheusPortProtocol
+	promPortName := config.DefaultPrometheusPortName
 	if app.Spec.Monitoring.Prometheus.PortName != nil {
-		portName = *app.Spec.Monitoring.Prometheus.PortName
+		promPortName = *app.Spec.Monitoring.Prometheus.PortName
 	}
 
 	patchOps = append(patchOps, addConfigMapVolume(pod, name, volumeName))
@@ -384,13 +385,33 @@ func getPrometheusConfigPatches(pod *corev1.Pod, app *v1beta2.SparkApplication)
 		return nil
 	}
 	patchOps = append(patchOps, *vmPatchOp)
-	portPatchOp := addContainerPort(pod, port, protocol, portName)
-	if portPatchOp == nil {
-		glog.Warningf("could not expose port %d to scrape metrics outside the pod", port)
+	promPortPatchOp := addContainerPort(pod, promPort, promProtocol, promPortName)
+	if promPortPatchOp == nil {
+		glog.Warningf("could not expose port %d to scrape metrics outside the pod", promPort)
 		return nil
 	}
-	patchOps = append(patchOps, *portPatchOp)
+	patchOps = append(patchOps, *promPortPatchOp)
+	return patchOps
+}
+
+func addContainerPorts(pod *corev1.Pod, app *v1beta2.SparkApplication) []patchOperation {
+	var ports []v1beta2.Port
+
+	if util.IsDriverPod(pod) {
+		ports = app.Spec.Driver.Ports
+	} else if util.IsExecutorPod(pod) {
+		ports = app.Spec.Executor.Ports
+	}
 
+	var patchOps []patchOperation
+	for _, p := range ports {
+		portPatchOp := addContainerPort(pod, p.ContainerPort, p.Protocol, p.Name)
+		if portPatchOp == nil {
+			glog.Warningf("could not expose port named %s", p.Name)
+			continue
+		}
+		patchOps = append(patchOps, *portPatchOp)
+	}
 	return patchOps
 }
 
@@ -801,12 +822,11 @@ func addHostAliases(pod *corev1.Pod, app *v1beta2.SparkApplication) []patchOpera
 	}
 
 	var ops []patchOperation
-	for _, v := range hostAliases {
+	if len(hostAliases) > 0 {
 		if first {
-			ops = append(ops, patchOperation{Op: "add", Path: "/spec/hostAliases", Value: []corev1.HostAlias{v}})
-			first = false
+			ops = append(ops, patchOperation{Op: "add", Path: "/spec/hostAliases", Value: hostAliases})
 		} else {
-			ops = append(ops, patchOperation{Op: "add", Path: "/spec/hostAliases/-", Value: &v})
+			ops = append(ops, patchOperation{Op: "add", Path: "/spec/hostAliases/-", Value: hostAliases})
 		}
 	}
 	return ops
diff --git a/sparkctl/README.md b/sparkctl/README.md
index 4f94e69..e9fdc90 100644
--- a/sparkctl/README.md
+++ b/sparkctl/README.md
@@ -110,6 +110,12 @@ If you want to use custom S3 endpoint or region, add `--upload-to-endpoint` and
 $ sparkctl create <path to YAML file> --upload-to-endpoint https://<endpoint-url> --upload-to-region <endpoint-region> --upload-to s3://<bucket>
 ```
 
+If you want to force path style URLs for S3 objects add `--s3-force-path-style`: 
+
+```bash
+$ sparkctl create <path to YAML file> --s3-force-path-style
+```
+
 If you want to make uploaded dependencies publicly available, add `--public` to the `create` command, as the following example shows:
 
 ```bash
diff --git a/sparkctl/cmd/create.go b/sparkctl/cmd/create.go
index d79fe5d..318829a 100644
--- a/sparkctl/cmd/create.go
+++ b/sparkctl/cmd/create.go
@@ -40,12 +40,13 @@ import (
 )
 
 const bufferSize = 1024
-const rootPath = "spark-app-dependencies"
 
+var RootPath string
 var UploadToPath string
 var UploadToEndpoint string
 var UploadToRegion string
 var Public bool
+var S3ForcePathStyle bool
 var Override bool
 var From string
 
@@ -90,13 +91,17 @@ var createCmd = &cobra.Command{
 
 func init() {
 	createCmd.Flags().StringVarP(&UploadToPath, "upload-to", "u", "",
-		"a URL of the remote location where local application dependencies are to be uploaded to")
+		"the name of the bucket where local application dependencies are to be uploaded")
+	createCmd.Flags().StringVarP(&RootPath, "upload-prefix", "p", "",
+		"the prefix to use for the dependency uploads")
 	createCmd.Flags().StringVarP(&UploadToRegion, "upload-to-region", "r", "",
 		"the GCS or S3 storage region for the bucket")
 	createCmd.Flags().StringVarP(&UploadToEndpoint, "upload-to-endpoint", "e",
 		"https://storage.googleapis.com", "the GCS or S3 storage api endpoint url")
 	createCmd.Flags().BoolVarP(&Public, "public", "c", false,
 		"whether to make uploaded files publicly available")
+	createCmd.Flags().BoolVar(&S3ForcePathStyle, "s3-force-path-style", false,
+		"whether to force path style URLs for S3 objects")
 	createCmd.Flags().BoolVarP(&Override, "override", "o", false,
 		"whether to override remote files with the same names")
 	createCmd.Flags().StringVarP(&From, "from", "f", "",
@@ -379,7 +384,7 @@ func uploadLocalDependencies(app *v1beta2.SparkApplication, files []string) ([]s
 	case "gs":
 		uh, err = newGCSBlob(ctx, uploadBucket, UploadToEndpoint, UploadToRegion)
 	case "s3":
-		uh, err = newS3Blob(ctx, uploadBucket, UploadToEndpoint, UploadToRegion)
+		uh, err = newS3Blob(ctx, uploadBucket, UploadToEndpoint, UploadToRegion, S3ForcePathStyle)
 	default:
 		return nil, fmt.Errorf("unsupported upload location URL scheme: %s", uploadLocationUrl.Scheme)
 	}
@@ -390,7 +395,7 @@ func uploadLocalDependencies(app *v1beta2.SparkApplication, files []string) ([]s
 	}
 
 	var uploadedFilePaths []string
-	uploadPath := filepath.Join(rootPath, app.Namespace, app.Name)
+	uploadPath := filepath.Join(RootPath, app.Namespace, app.Name)
 	for _, localFilePath := range files {
 		uploadFilePath, err := uh.uploadToBucket(uploadPath, localFilePath)
 		if err != nil {
diff --git a/sparkctl/cmd/s3.go b/sparkctl/cmd/s3.go
index d6bf9e9..8b9544d 100644
--- a/sparkctl/cmd/s3.go
+++ b/sparkctl/cmd/s3.go
@@ -48,14 +48,16 @@ func newS3Blob(
 	ctx context.Context,
 	bucket string,
 	endpoint string,
-	region string) (*uploadHandler, error) {
+	region string,
+	forcePathStyle bool) (*uploadHandler, error) {
 	// AWS SDK does require specifying regions, thus set it to default S3 region
 	if region == "" {
 		region = "us-east1"
 	}
 	c := &aws.Config{
-		Region:   aws.String(region),
-		Endpoint: aws.String(endpoint),
+		Region:           aws.String(region),
+		Endpoint:         aws.String(endpoint),
+		S3ForcePathStyle: aws.Bool(forcePathStyle),
 	}
 	sess := session.Must(session.NewSession(c))
 	b, err := s3blob.OpenBucket(ctx, sess, bucket)


Test patch:
diff --git a/pkg/webhook/patch_test.go b/pkg/webhook/patch_test.go
index 5b3b707..8db4fbb 100644
--- a/pkg/webhook/patch_test.go
+++ b/pkg/webhook/patch_test.go
@@ -1833,6 +1833,14 @@ func TestPatchSparkPod_HostAliases(t *testing.T) {
 							IP:        "192.168.0.1",
 							Hostnames: []string{"test.com", "test2.com"},
 						},
+						{
+							IP:        "192.168.0.2",
+							Hostnames: []string{"test3.com"},
+						},
+						{
+							IP:        "192.168.0.3",
+							Hostnames: []string{"test4.com"},
+						},
 					},
 				},
 			},
@@ -1875,9 +1883,11 @@ func TestPatchSparkPod_HostAliases(t *testing.T) {
 	if err != nil {
 		t.Fatal(err)
 	}
-	assert.Equal(t, 2, len(modifiedDriverPod.Spec.HostAliases))
+	assert.Equal(t, 4, len(modifiedDriverPod.Spec.HostAliases))
 	assert.Equal(t, "127.0.0.1", modifiedDriverPod.Spec.HostAliases[0].IP)
 	assert.Equal(t, "192.168.0.1", modifiedDriverPod.Spec.HostAliases[1].IP)
+	assert.Equal(t, "192.168.0.2", modifiedDriverPod.Spec.HostAliases[2].IP)
+	assert.Equal(t, "192.168.0.3", modifiedDriverPod.Spec.HostAliases[3].IP)
 
 	executorPod := &corev1.Pod{
 		ObjectMeta: metav1.ObjectMeta{
@@ -1905,3 +1915,83 @@ func TestPatchSparkPod_HostAliases(t *testing.T) {
 	assert.Equal(t, "127.0.0.1", modifiedExecutorPod.Spec.HostAliases[0].IP)
 	assert.Equal(t, "192.168.0.1", modifiedExecutorPod.Spec.HostAliases[1].IP)
 }
+
+func TestPatchSparkPod_Ports(t *testing.T) {
+	app := &v1beta2.SparkApplication{
+		ObjectMeta: metav1.ObjectMeta{
+			Name: "spark-test",
+			UID:  "spark-test-1",
+		},
+		Spec: v1beta2.SparkApplicationSpec{
+			Driver: v1beta2.DriverSpec{
+				Ports: []v1beta2.Port{
+					{Name: "driverPort1", ContainerPort: 8080, Protocol: "TCP"},
+					{Name: "driverPort2", ContainerPort: 8081, Protocol: "TCP"},
+				},
+			},
+			Executor: v1beta2.ExecutorSpec{
+				Ports: []v1beta2.Port{
+					{Name: "executorPort1", ContainerPort: 8082, Protocol: "TCP"},
+					{Name: "executorPort2", ContainerPort: 8083, Protocol: "TCP"},
+				},
+			},
+		},
+	}
+
+	driverPod := &corev1.Pod{
+		ObjectMeta: metav1.ObjectMeta{
+			Name: "spark-driver",
+			Labels: map[string]string{
+				config.SparkRoleLabel:               config.SparkDriverRole,
+				config.LaunchedBySparkOperatorLabel: "true",
+			},
+		},
+		Spec: corev1.PodSpec{
+			Containers: []corev1.Container{
+				{
+					Name:  config.SparkDriverContainerName,
+					Image: "spark-driver:latest",
+				},
+			},
+		},
+	}
+
+	modifiedDriverPod, err := getModifiedPod(driverPod, app)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	assert.Equal(t, 2, len(modifiedDriverPod.Spec.Containers[0].Ports))
+	assert.Equal(t, "driverPort1", modifiedDriverPod.Spec.Containers[0].Ports[0].Name)
+	assert.Equal(t, "driverPort2", modifiedDriverPod.Spec.Containers[0].Ports[1].Name)
+	assert.Equal(t, int32(8080), modifiedDriverPod.Spec.Containers[0].Ports[0].ContainerPort)
+	assert.Equal(t, int32(8081), modifiedDriverPod.Spec.Containers[0].Ports[1].ContainerPort)
+
+	executorPod := &corev1.Pod{
+		ObjectMeta: metav1.ObjectMeta{
+			Name: "spark-executor",
+			Labels: map[string]string{
+				config.SparkRoleLabel:               config.SparkExecutorRole,
+				config.LaunchedBySparkOperatorLabel: "true",
+			},
+		},
+		Spec: corev1.PodSpec{
+			Containers: []corev1.Container{
+				{
+					Name:  config.SparkExecutorContainerName,
+					Image: "spark-executor:latest",
+				},
+			},
+		},
+	}
+
+	modifiedExecutorPod, err := getModifiedPod(executorPod, app)
+	if err != nil {
+		t.Fatal(err)
+	}
+	assert.Equal(t, 2, len(modifiedExecutorPod.Spec.Containers[0].Ports))
+	assert.Equal(t, "executorPort1", modifiedExecutorPod.Spec.Containers[0].Ports[0].Name)
+	assert.Equal(t, "executorPort2", modifiedExecutorPod.Spec.Containers[0].Ports[1].Name)
+	assert.Equal(t, int32(8082), modifiedExecutorPod.Spec.Containers[0].Ports[0].ContainerPort)
+	assert.Equal(t, int32(8083), modifiedExecutorPod.Spec.Containers[0].Ports[1].ContainerPort)
+}