mirror of
https://github.com/hanzo-ml/spark-operator.git
synced 2026-10-30 20:39:07 +00:00
Venkateswarlu KomirisettiandGitHub
5006eb7a3d
* feat: add REST-based spark submitter with mTLS support - Add RestSparkSubmitter with mTLS, retry with exponential backoff, and HTTP+TLS startup verification - Add typed SubmitErrorCode enum with structured error responses - Implement operator-side idempotent retry (DRIVER_POD_ALREADY_EXISTS + matching submission_id = success) - Switch retry logic from HTTP-status-based to error-code-based - Add RestSubmitter feature gate to toggle between default and REST submitter - Move feature gate check to start.go with newSparkSubmitter() helper - Document contract, request/response payloads, and error codes Signed-off-by: Venkateswarlu Komirisetti <[email protected]> * feat: integrate server-side idempotency (duplicate_submission) The submitter now handles duplicate submissions server-side, returning 200 OK with duplicate_submission=true instead of an error. Update the operator client to treat both 201 and 200 as success, and update DRIVER_POD_ALREADY_EXISTS to 409 (genuine conflict only). Signed-off-by: Venkateswarlu Komirisetti <[email protected]> * fix: check resp.Body.Close error to satisfy errcheck linter Signed-off-by: Venkateswarlu Komirisetti <[email protected]> * refactor: address PR review feedback for REST submitter - Pass context from SetupSignalHandler to newSparkSubmitter for proper signal propagation and return error instead of calling os.Exit - Move submitter-max-retries validation to PreRunE for early failure - Extract buildDriverPodTemplate/buildExecutorPodTemplate as pure functions to ensure pod templates include owner refs for both spark-submit and REST submission paths - Remove cert reload mention from comment (not yet implemented) Signed-off-by: Venkateswarlu Komirisetti <[email protected]> * refactor: simplify TLS flags and add submitter URL validation Replace --submitter-tls-cert-dir with --submitter-tls-enabled flag and direct path flags (--submitter-tls-cert-file, --submitter-tls-key-file, --submitter-tls-ca-file). Group all REST submitter validations under RestSubmitter feature gate check in PreRunE, including requiring --submitter-service-url when the gate is enabled. Signed-off-by: Venkateswarlu Komirisetti <[email protected]> * style: fix variable alignment to pass go-fmt Signed-off-by: Venkateswarlu Komirisetti <[email protected]> --------- Signed-off-by: Venkateswarlu Komirisetti <[email protected]>
1383 lines
47 KiB
Go
1383 lines
47 KiB
Go
/*
|
|
Copyright 2017 Google LLC
|
|
|
|
Licensed 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
|
|
|
|
https://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 sparkapplication
|
|
|
|
import (
|
|
"fmt"
|
|
"os"
|
|
"slices"
|
|
"strings"
|
|
"testing"
|
|
|
|
"github.com/google/uuid"
|
|
"github.com/stretchr/testify/assert"
|
|
corev1 "k8s.io/api/core/v1"
|
|
"k8s.io/apimachinery/pkg/api/resource"
|
|
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
|
|
"k8s.io/apimachinery/pkg/types"
|
|
"k8s.io/utils/ptr"
|
|
"sigs.k8s.io/yaml"
|
|
|
|
"github.com/kubeflow/spark-operator/v2/api/v1beta2"
|
|
"github.com/kubeflow/spark-operator/v2/pkg/common"
|
|
"github.com/kubeflow/spark-operator/v2/pkg/util"
|
|
)
|
|
|
|
func TestExecutorConfOption(t *testing.T) {
|
|
tests := []struct {
|
|
name string
|
|
app *v1beta2.SparkApplication
|
|
expected []string
|
|
}{
|
|
{
|
|
name: "minimal executor configuration",
|
|
app: &v1beta2.SparkApplication{
|
|
ObjectMeta: metav1.ObjectMeta{
|
|
Name: "spark-minimal",
|
|
},
|
|
Status: v1beta2.SparkApplicationStatus{
|
|
SubmissionID: "minimal-123",
|
|
},
|
|
Spec: v1beta2.SparkApplicationSpec{
|
|
Executor: v1beta2.ExecutorSpec{},
|
|
},
|
|
},
|
|
expected: []string{
|
|
"--conf", fmt.Sprintf("%s=%s", fmt.Sprintf(common.SparkKubernetesExecutorLabelTemplate, common.LabelSparkAppName), "spark-minimal"),
|
|
"--conf", fmt.Sprintf("%s=%s", fmt.Sprintf(common.SparkKubernetesExecutorLabelTemplate, common.LabelLaunchedBySparkOperator), "true"),
|
|
"--conf", fmt.Sprintf("%s=%s", fmt.Sprintf(common.SparkKubernetesExecutorLabelTemplate, common.LabelMutatedBySparkOperator), "true"),
|
|
"--conf", fmt.Sprintf("%s=%s", fmt.Sprintf(common.SparkKubernetesExecutorLabelTemplate, common.LabelSubmissionID), "minimal-123"),
|
|
},
|
|
},
|
|
{
|
|
name: "custom executor configs",
|
|
app: &v1beta2.SparkApplication{
|
|
ObjectMeta: metav1.ObjectMeta{
|
|
Name: "spark-executor",
|
|
},
|
|
Status: v1beta2.SparkApplicationStatus{
|
|
SubmissionID: "exec-123",
|
|
},
|
|
Spec: v1beta2.SparkApplicationSpec{
|
|
Executor: v1beta2.ExecutorSpec{
|
|
CoreRequest: ptr.To("2"),
|
|
SparkPodSpec: v1beta2.SparkPodSpec{
|
|
Image: ptr.To("custom-executor:v1"),
|
|
Cores: ptr.To[int32](4),
|
|
CoreLimit: ptr.To("4"),
|
|
Memory: ptr.To("4g"),
|
|
MemoryOverhead: ptr.To("1g"),
|
|
},
|
|
Instances: ptr.To[int32](3),
|
|
DeleteOnTermination: ptr.To(true),
|
|
},
|
|
},
|
|
},
|
|
expected: []string{
|
|
"--conf", fmt.Sprintf("%s=%s", fmt.Sprintf(common.SparkKubernetesExecutorLabelTemplate, common.LabelSparkAppName), "spark-executor"),
|
|
"--conf", fmt.Sprintf("%s=%s", fmt.Sprintf(common.SparkKubernetesExecutorLabelTemplate, common.LabelLaunchedBySparkOperator), "true"),
|
|
"--conf", fmt.Sprintf("%s=%s", fmt.Sprintf(common.SparkKubernetesExecutorLabelTemplate, common.LabelMutatedBySparkOperator), "true"),
|
|
"--conf", fmt.Sprintf("%s=%s", fmt.Sprintf(common.SparkKubernetesExecutorLabelTemplate, common.LabelSubmissionID), "exec-123"),
|
|
"--conf", fmt.Sprintf("%s=%s", common.SparkKubernetesExecutorContainerImage, "custom-executor:v1"),
|
|
"--conf", fmt.Sprintf("%s=%d", common.SparkExecutorCores, 4),
|
|
"--conf", fmt.Sprintf("%s=%s", common.SparkKubernetesExecutorLimitCores, "4"),
|
|
"--conf", fmt.Sprintf("%s=%s", common.SparkKubernetesExecutorRequestCores, "2"),
|
|
"--conf", fmt.Sprintf("%s=%s", common.SparkExecutorMemory, "4g"),
|
|
"--conf", fmt.Sprintf("%s=%s", common.SparkExecutorMemoryOverhead, "1g"),
|
|
"--conf", fmt.Sprintf("%s=%d", common.SparkExecutorInstances, 3),
|
|
"--conf", fmt.Sprintf("%s=%t", common.SparkKubernetesExecutorDeleteOnTermination, true),
|
|
},
|
|
},
|
|
{
|
|
name: "kueue labels on SparkApplication should not be propagated to executor conf",
|
|
app: &v1beta2.SparkApplication{
|
|
ObjectMeta: metav1.ObjectMeta{
|
|
Name: "spark-kueue",
|
|
Labels: map[string]string{
|
|
"environment": "production",
|
|
"kueue.x-k8s.io/queue-name": "high-priority",
|
|
},
|
|
},
|
|
Status: v1beta2.SparkApplicationStatus{
|
|
SubmissionID: "minimal-123",
|
|
},
|
|
Spec: v1beta2.SparkApplicationSpec{
|
|
Executor: v1beta2.ExecutorSpec{},
|
|
},
|
|
},
|
|
expected: []string{
|
|
"--conf", fmt.Sprintf("%s=%s", fmt.Sprintf(common.SparkKubernetesExecutorLabelTemplate, "environment"), "production"),
|
|
"--conf", fmt.Sprintf("%s=%s", fmt.Sprintf(common.SparkKubernetesExecutorLabelTemplate, common.LabelSparkAppName), "spark-kueue"),
|
|
"--conf", fmt.Sprintf("%s=%s", fmt.Sprintf(common.SparkKubernetesExecutorLabelTemplate, common.LabelLaunchedBySparkOperator), "true"),
|
|
"--conf", fmt.Sprintf("%s=%s", fmt.Sprintf(common.SparkKubernetesExecutorLabelTemplate, common.LabelMutatedBySparkOperator), "true"),
|
|
"--conf", fmt.Sprintf("%s=%s", fmt.Sprintf(common.SparkKubernetesExecutorLabelTemplate, common.LabelSubmissionID), "minimal-123"),
|
|
},
|
|
},
|
|
}
|
|
|
|
for _, testCase := range tests {
|
|
t.Run(testCase.name, func(t *testing.T) {
|
|
options, err := executorConfOption(testCase.app)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
|
|
slices.Sort(options)
|
|
slices.Sort(testCase.expected)
|
|
|
|
assert.Equal(t, testCase.expected, options, "Executor configuration options do not match expected values")
|
|
})
|
|
}
|
|
}
|
|
|
|
func TestDriverVolumeMountsOption(t *testing.T) {
|
|
tests := []struct {
|
|
name string
|
|
app *v1beta2.SparkApplication
|
|
expected []string
|
|
}{
|
|
{
|
|
name: "basic local volume",
|
|
app: &v1beta2.SparkApplication{
|
|
ObjectMeta: metav1.ObjectMeta{
|
|
Name: "spark-volumes",
|
|
},
|
|
Spec: v1beta2.SparkApplicationSpec{
|
|
Volumes: []corev1.Volume{
|
|
{
|
|
Name: fmt.Sprintf("%stest", common.SparkLocalDirVolumePrefix),
|
|
VolumeSource: corev1.VolumeSource{
|
|
HostPath: &corev1.HostPathVolumeSource{
|
|
Path: "/tmp/test-path",
|
|
},
|
|
},
|
|
},
|
|
},
|
|
Driver: v1beta2.DriverSpec{
|
|
SparkPodSpec: v1beta2.SparkPodSpec{
|
|
VolumeMounts: []corev1.VolumeMount{
|
|
{
|
|
Name: fmt.Sprintf("%stest", common.SparkLocalDirVolumePrefix),
|
|
MountPath: "/tmp/mount-path",
|
|
ReadOnly: true,
|
|
},
|
|
},
|
|
},
|
|
},
|
|
},
|
|
},
|
|
expected: []string{
|
|
"--conf", fmt.Sprintf("%s=%s",
|
|
fmt.Sprintf(common.SparkKubernetesDriverVolumesMountPathTemplate, common.VolumeTypeHostPath, fmt.Sprintf("%stest", common.SparkLocalDirVolumePrefix)),
|
|
"/tmp/mount-path"),
|
|
"--conf", fmt.Sprintf("%s=%s",
|
|
fmt.Sprintf(common.SparkKubernetesDriverVolumesOptionsTemplate, common.VolumeTypeHostPath, fmt.Sprintf("%stest", common.SparkLocalDirVolumePrefix), "path"),
|
|
"/tmp/test-path"),
|
|
"--conf", fmt.Sprintf("%s=%s",
|
|
fmt.Sprintf(common.SparkKubernetesDriverVolumesMountReadOnlyTemplate, common.VolumeTypeHostPath, fmt.Sprintf("%stest", common.SparkLocalDirVolumePrefix)),
|
|
"true"),
|
|
},
|
|
},
|
|
{
|
|
name: "emptydir volume with size limit",
|
|
app: &v1beta2.SparkApplication{
|
|
ObjectMeta: metav1.ObjectMeta{
|
|
Name: "spark-emptydir",
|
|
},
|
|
Spec: v1beta2.SparkApplicationSpec{
|
|
Volumes: []corev1.Volume{
|
|
{
|
|
Name: fmt.Sprintf("%semptydir", common.SparkLocalDirVolumePrefix),
|
|
VolumeSource: corev1.VolumeSource{
|
|
EmptyDir: &corev1.EmptyDirVolumeSource{
|
|
SizeLimit: resource.NewQuantity(10*1024*1024*1024, resource.BinarySI), // 10Gi
|
|
},
|
|
},
|
|
},
|
|
},
|
|
Driver: v1beta2.DriverSpec{
|
|
SparkPodSpec: v1beta2.SparkPodSpec{
|
|
VolumeMounts: []corev1.VolumeMount{
|
|
{
|
|
Name: fmt.Sprintf("%semptydir", common.SparkLocalDirVolumePrefix),
|
|
MountPath: "/tmp/scratch",
|
|
ReadOnly: false,
|
|
},
|
|
},
|
|
},
|
|
},
|
|
},
|
|
},
|
|
expected: []string{
|
|
"--conf", fmt.Sprintf("%s=%s",
|
|
fmt.Sprintf(common.SparkKubernetesDriverVolumesMountPathTemplate, common.VolumeTypeEmptyDir, fmt.Sprintf("%semptydir", common.SparkLocalDirVolumePrefix)),
|
|
"/tmp/scratch"),
|
|
"--conf", fmt.Sprintf("%s=%s",
|
|
fmt.Sprintf(common.SparkKubernetesDriverVolumesOptionsTemplate, common.VolumeTypeEmptyDir, fmt.Sprintf("%semptydir", common.SparkLocalDirVolumePrefix), "sizeLimit"),
|
|
"10Gi"),
|
|
},
|
|
},
|
|
{
|
|
name: "emptydir volume with memory medium",
|
|
app: &v1beta2.SparkApplication{
|
|
ObjectMeta: metav1.ObjectMeta{
|
|
Name: "spark-emptydir-memory",
|
|
},
|
|
Spec: v1beta2.SparkApplicationSpec{
|
|
Volumes: []corev1.Volume{
|
|
{
|
|
Name: fmt.Sprintf("%semptydir", common.SparkLocalDirVolumePrefix),
|
|
VolumeSource: corev1.VolumeSource{
|
|
EmptyDir: &corev1.EmptyDirVolumeSource{
|
|
Medium: corev1.StorageMediumMemory,
|
|
},
|
|
},
|
|
},
|
|
},
|
|
Driver: v1beta2.DriverSpec{
|
|
SparkPodSpec: v1beta2.SparkPodSpec{
|
|
VolumeMounts: []corev1.VolumeMount{
|
|
{
|
|
Name: fmt.Sprintf("%semptydir", common.SparkLocalDirVolumePrefix),
|
|
MountPath: "/tmp/scratch",
|
|
ReadOnly: false,
|
|
},
|
|
},
|
|
},
|
|
},
|
|
},
|
|
},
|
|
expected: []string{
|
|
"--conf", fmt.Sprintf("%s=%s",
|
|
fmt.Sprintf(common.SparkKubernetesDriverVolumesMountPathTemplate, common.VolumeTypeEmptyDir, fmt.Sprintf("%semptydir", common.SparkLocalDirVolumePrefix)),
|
|
"/tmp/scratch"),
|
|
"--conf", fmt.Sprintf("%s=%s",
|
|
fmt.Sprintf(common.SparkKubernetesDriverVolumesOptionsTemplate, common.VolumeTypeEmptyDir, fmt.Sprintf("%semptydir", common.SparkLocalDirVolumePrefix), "medium"),
|
|
string(corev1.StorageMediumMemory)),
|
|
},
|
|
},
|
|
}
|
|
|
|
for _, tt := range tests {
|
|
t.Run(tt.name, func(t *testing.T) {
|
|
options, err := driverVolumeMountsOption(tt.app)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
|
|
slices.Sort(options)
|
|
slices.Sort(tt.expected)
|
|
|
|
assert.Equal(t, tt.expected, options, "Driver volume mount options do not match expected values")
|
|
})
|
|
}
|
|
}
|
|
|
|
func TestDriverConfOption(t *testing.T) {
|
|
tests := []struct {
|
|
name string
|
|
app *v1beta2.SparkApplication
|
|
expected []string
|
|
}{
|
|
{
|
|
name: "basic driver configuration",
|
|
app: &v1beta2.SparkApplication{
|
|
ObjectMeta: metav1.ObjectMeta{
|
|
Name: "spark-test",
|
|
Labels: map[string]string{
|
|
"environment": "test",
|
|
"team": "spark",
|
|
},
|
|
},
|
|
Status: v1beta2.SparkApplicationStatus{
|
|
SubmissionID: "submission-id-123",
|
|
},
|
|
Spec: v1beta2.SparkApplicationSpec{
|
|
Driver: v1beta2.DriverSpec{
|
|
SparkPodSpec: v1beta2.SparkPodSpec{
|
|
Cores: ptr.To[int32](2),
|
|
CoreLimit: ptr.To("2"),
|
|
Memory: ptr.To("2g"),
|
|
Image: ptr.To("spark-driver:latest"),
|
|
ConfigMaps: []v1beta2.NamePath{{Name: "driver-config", Path: "/etc/config"}},
|
|
},
|
|
},
|
|
},
|
|
},
|
|
expected: []string{
|
|
"--conf", fmt.Sprintf("%s=%s", fmt.Sprintf(common.SparkKubernetesDriverLabelTemplate, common.LabelSparkAppName), "spark-test"),
|
|
"--conf", fmt.Sprintf("%s=%s", fmt.Sprintf(common.SparkKubernetesDriverLabelTemplate, common.LabelLaunchedBySparkOperator), "true"),
|
|
"--conf", fmt.Sprintf("%s=%s", fmt.Sprintf(common.SparkKubernetesDriverLabelTemplate, common.LabelMutatedBySparkOperator), "true"),
|
|
"--conf", fmt.Sprintf("%s=%s", fmt.Sprintf(common.SparkKubernetesDriverLabelTemplate, common.LabelSubmissionID), "submission-id-123"),
|
|
"--conf", fmt.Sprintf("%s=%s", common.SparkKubernetesDriverContainerImage, "spark-driver:latest"),
|
|
"--conf", fmt.Sprintf("%s=%d", common.SparkDriverCores, 2),
|
|
"--conf", fmt.Sprintf("%s=%s", common.SparkKubernetesDriverLimitCores, "2"),
|
|
"--conf", fmt.Sprintf("%s=%s", common.SparkDriverMemory, "2g"),
|
|
"--conf", fmt.Sprintf("%s=%s", fmt.Sprintf(common.SparkKubernetesDriverLabelTemplate, "environment"), "test"),
|
|
"--conf", fmt.Sprintf("%s=%s", fmt.Sprintf(common.SparkKubernetesDriverLabelTemplate, "team"), "spark"),
|
|
},
|
|
},
|
|
{
|
|
name: "minimal driver configuration",
|
|
app: &v1beta2.SparkApplication{
|
|
ObjectMeta: metav1.ObjectMeta{
|
|
Name: "spark-minimal",
|
|
},
|
|
Status: v1beta2.SparkApplicationStatus{
|
|
SubmissionID: "minimal-123",
|
|
},
|
|
Spec: v1beta2.SparkApplicationSpec{
|
|
Driver: v1beta2.DriverSpec{},
|
|
},
|
|
},
|
|
expected: []string{
|
|
"--conf", fmt.Sprintf("%s=%s", fmt.Sprintf(common.SparkKubernetesDriverLabelTemplate, common.LabelSparkAppName), "spark-minimal"),
|
|
"--conf", fmt.Sprintf("%s=%s", fmt.Sprintf(common.SparkKubernetesDriverLabelTemplate, common.LabelLaunchedBySparkOperator), "true"),
|
|
"--conf", fmt.Sprintf("%s=%s", fmt.Sprintf(common.SparkKubernetesDriverLabelTemplate, common.LabelMutatedBySparkOperator), "true"),
|
|
"--conf", fmt.Sprintf("%s=%s", fmt.Sprintf(common.SparkKubernetesDriverLabelTemplate, common.LabelSubmissionID), "minimal-123"),
|
|
},
|
|
},
|
|
{
|
|
name: "kueue labels on SparkApplication should not be propagated to driver conf",
|
|
app: &v1beta2.SparkApplication{
|
|
ObjectMeta: metav1.ObjectMeta{
|
|
Name: "spark-minimal",
|
|
Labels: map[string]string{
|
|
"environment": "production",
|
|
"kueue.x-k8s.io/queue-name": "high-priority",
|
|
},
|
|
},
|
|
Status: v1beta2.SparkApplicationStatus{
|
|
SubmissionID: "minimal-123",
|
|
},
|
|
Spec: v1beta2.SparkApplicationSpec{
|
|
Driver: v1beta2.DriverSpec{},
|
|
},
|
|
},
|
|
expected: []string{
|
|
"--conf", fmt.Sprintf("%s=%s", fmt.Sprintf(common.SparkKubernetesDriverLabelTemplate, "environment"), "production"),
|
|
"--conf", fmt.Sprintf("%s=%s", fmt.Sprintf(common.SparkKubernetesDriverLabelTemplate, common.LabelSparkAppName), "spark-minimal"),
|
|
"--conf", fmt.Sprintf("%s=%s", fmt.Sprintf(common.SparkKubernetesDriverLabelTemplate, common.LabelLaunchedBySparkOperator), "true"),
|
|
"--conf", fmt.Sprintf("%s=%s", fmt.Sprintf(common.SparkKubernetesDriverLabelTemplate, common.LabelMutatedBySparkOperator), "true"),
|
|
"--conf", fmt.Sprintf("%s=%s", fmt.Sprintf(common.SparkKubernetesDriverLabelTemplate, common.LabelSubmissionID), "minimal-123"),
|
|
},
|
|
},
|
|
{
|
|
name: "driver with labels and annotations",
|
|
app: &v1beta2.SparkApplication{
|
|
ObjectMeta: metav1.ObjectMeta{
|
|
Name: "spark-labels",
|
|
},
|
|
Status: v1beta2.SparkApplicationStatus{
|
|
SubmissionID: "labels-123",
|
|
},
|
|
Spec: v1beta2.SparkApplicationSpec{
|
|
Driver: v1beta2.DriverSpec{
|
|
SparkPodSpec: v1beta2.SparkPodSpec{
|
|
Labels: map[string]string{
|
|
"custom-label": "label-value",
|
|
"priority": "high",
|
|
},
|
|
Annotations: map[string]string{
|
|
"custom-annotation": "annotation-value",
|
|
"sidecar.istio.io/inject": "false",
|
|
},
|
|
},
|
|
ServiceLabels: map[string]string{
|
|
"service-label": "service-value",
|
|
"environment": "production",
|
|
},
|
|
ServiceAnnotations: map[string]string{
|
|
"service-annotation": "svc-annotation-value",
|
|
"prometheus.io/scrape": "true",
|
|
},
|
|
},
|
|
},
|
|
},
|
|
expected: []string{
|
|
"--conf", fmt.Sprintf("%s=%s", fmt.Sprintf(common.SparkKubernetesDriverLabelTemplate, common.LabelSparkAppName), "spark-labels"),
|
|
"--conf", fmt.Sprintf("%s=%s", fmt.Sprintf(common.SparkKubernetesDriverLabelTemplate, common.LabelLaunchedBySparkOperator), "true"),
|
|
"--conf", fmt.Sprintf("%s=%s", fmt.Sprintf(common.SparkKubernetesDriverLabelTemplate, common.LabelMutatedBySparkOperator), "true"),
|
|
"--conf", fmt.Sprintf("%s=%s", fmt.Sprintf(common.SparkKubernetesDriverLabelTemplate, common.LabelSubmissionID), "labels-123"),
|
|
"--conf", fmt.Sprintf("%s=%s", fmt.Sprintf(common.SparkKubernetesDriverLabelTemplate, "custom-label"), "label-value"),
|
|
"--conf", fmt.Sprintf("%s=%s", fmt.Sprintf(common.SparkKubernetesDriverLabelTemplate, "priority"), "high"),
|
|
"--conf", fmt.Sprintf("%s=%s", fmt.Sprintf(common.SparkKubernetesDriverAnnotationTemplate, "custom-annotation"), "annotation-value"),
|
|
"--conf", fmt.Sprintf("%s=%s", fmt.Sprintf(common.SparkKubernetesDriverAnnotationTemplate, "sidecar.istio.io/inject"), "false"),
|
|
"--conf", fmt.Sprintf("%s=%s", fmt.Sprintf(common.SparkKubernetesDriverServiceLabelTemplate, "service-label"), "service-value"),
|
|
"--conf", fmt.Sprintf("%s=%s", fmt.Sprintf(common.SparkKubernetesDriverServiceLabelTemplate, "environment"), "production"),
|
|
"--conf", fmt.Sprintf("%s=%s", fmt.Sprintf(common.SparkKubernetesDriverServiceAnnotationTemplate, "service-annotation"), "svc-annotation-value"),
|
|
"--conf", fmt.Sprintf("%s=%s", fmt.Sprintf(common.SparkKubernetesDriverServiceAnnotationTemplate, "prometheus.io/scrape"), "true"),
|
|
},
|
|
},
|
|
}
|
|
|
|
for _, testCase := range tests {
|
|
t.Run(testCase.name, func(t *testing.T) {
|
|
options, err := driverConfOption(testCase.app)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
|
|
slices.Sort(options)
|
|
slices.Sort(testCase.expected)
|
|
|
|
assert.Equal(t, testCase.expected, options, "Driver configuration options do not match expected values")
|
|
})
|
|
}
|
|
}
|
|
|
|
func TestDriverPodTemplateContents(t *testing.T) {
|
|
appName := "test"
|
|
uid := types.UID(uuid.New().String())
|
|
appOwnerReference := util.GetOwnerReference(&v1beta2.SparkApplication{
|
|
ObjectMeta: metav1.ObjectMeta{
|
|
Name: appName,
|
|
UID: uid,
|
|
},
|
|
})
|
|
for name, tc := range map[string]struct {
|
|
app *v1beta2.SparkApplication
|
|
expectedTemplate corev1.PodTemplateSpec
|
|
}{
|
|
"no pod template": {
|
|
app: &v1beta2.SparkApplication{},
|
|
expectedTemplate: corev1.PodTemplateSpec{
|
|
ObjectMeta: metav1.ObjectMeta{
|
|
OwnerReferences: []metav1.OwnerReference{appOwnerReference},
|
|
},
|
|
Spec: corev1.PodSpec{
|
|
Containers: []corev1.Container{{
|
|
Name: common.SparkDriverContainerName,
|
|
}},
|
|
},
|
|
},
|
|
},
|
|
"with pod template": {
|
|
app: &v1beta2.SparkApplication{
|
|
Spec: v1beta2.SparkApplicationSpec{
|
|
Driver: v1beta2.DriverSpec{
|
|
SparkPodSpec: v1beta2.SparkPodSpec{
|
|
Template: &corev1.PodTemplateSpec{
|
|
ObjectMeta: metav1.ObjectMeta{
|
|
OwnerReferences: []metav1.OwnerReference{{Name: "owner-in-template"}},
|
|
},
|
|
Spec: corev1.PodSpec{
|
|
Containers: []corev1.Container{
|
|
{Name: common.SparkDriverContainerName, Image: "image"},
|
|
},
|
|
},
|
|
},
|
|
},
|
|
},
|
|
},
|
|
},
|
|
expectedTemplate: corev1.PodTemplateSpec{
|
|
ObjectMeta: metav1.ObjectMeta{
|
|
OwnerReferences: []metav1.OwnerReference{
|
|
{Name: "owner-in-template"},
|
|
appOwnerReference,
|
|
},
|
|
},
|
|
Spec: corev1.PodSpec{
|
|
Containers: []corev1.Container{
|
|
{Name: common.SparkDriverContainerName, Image: "image"},
|
|
},
|
|
},
|
|
},
|
|
},
|
|
"pod template already has owner references": {
|
|
app: &v1beta2.SparkApplication{
|
|
Spec: v1beta2.SparkApplicationSpec{
|
|
Driver: v1beta2.DriverSpec{
|
|
SparkPodSpec: v1beta2.SparkPodSpec{
|
|
Template: &corev1.PodTemplateSpec{
|
|
ObjectMeta: metav1.ObjectMeta{
|
|
OwnerReferences: []metav1.OwnerReference{{Name: "owner-in-template"}, appOwnerReference},
|
|
},
|
|
},
|
|
},
|
|
},
|
|
},
|
|
},
|
|
expectedTemplate: corev1.PodTemplateSpec{
|
|
ObjectMeta: metav1.ObjectMeta{
|
|
OwnerReferences: []metav1.OwnerReference{
|
|
{Name: "owner-in-template"},
|
|
appOwnerReference,
|
|
},
|
|
},
|
|
},
|
|
},
|
|
} {
|
|
t.Run(name, func(t *testing.T) {
|
|
submissionID := "TestDriverPodTemplateOption-" + strings.ReplaceAll(name, " ", "_") + "-" + uuid.New().String()
|
|
tc.app.Name = appName
|
|
tc.app.UID = uid
|
|
tc.app.Status.SubmissionID = submissionID
|
|
|
|
_, err := driverPodTemplateOption(tc.app)
|
|
assert.NoError(t, err)
|
|
defer func() { _ = os.RemoveAll(fmt.Sprintf("/tmp/spark/%s", submissionID)) }()
|
|
|
|
expectedBytes, err := yaml.Marshal(tc.expectedTemplate)
|
|
assert.NoError(t, err)
|
|
|
|
actualBytes, err := os.ReadFile(fmt.Sprintf("/tmp/spark/%s/driver-pod-template.yaml", submissionID))
|
|
assert.NoError(t, err)
|
|
|
|
assert.Equal(t, string(expectedBytes), string(actualBytes))
|
|
})
|
|
}
|
|
}
|
|
|
|
func TestExecutorPodTemplateContents(t *testing.T) {
|
|
appName := "test"
|
|
uid := types.UID(uuid.New().String())
|
|
appNonControllerOwnerReference := util.GetOwnerReference(&v1beta2.SparkApplication{
|
|
ObjectMeta: metav1.ObjectMeta{
|
|
Name: appName,
|
|
UID: uid,
|
|
},
|
|
})
|
|
appNonControllerOwnerReference.Controller = nil
|
|
appNonControllerOwnerReference.BlockOwnerDeletion = nil
|
|
|
|
for name, tc := range map[string]struct {
|
|
app *v1beta2.SparkApplication
|
|
expectedTemplate corev1.PodTemplateSpec
|
|
}{
|
|
"no pod template": {
|
|
app: &v1beta2.SparkApplication{},
|
|
expectedTemplate: corev1.PodTemplateSpec{
|
|
ObjectMeta: metav1.ObjectMeta{
|
|
OwnerReferences: []metav1.OwnerReference{appNonControllerOwnerReference},
|
|
},
|
|
Spec: corev1.PodSpec{
|
|
Containers: []corev1.Container{{
|
|
Name: common.Spark3DefaultExecutorContainerName,
|
|
}},
|
|
},
|
|
},
|
|
},
|
|
"with pod template": {
|
|
app: &v1beta2.SparkApplication{
|
|
Spec: v1beta2.SparkApplicationSpec{
|
|
Executor: v1beta2.ExecutorSpec{
|
|
SparkPodSpec: v1beta2.SparkPodSpec{
|
|
Template: &corev1.PodTemplateSpec{
|
|
ObjectMeta: metav1.ObjectMeta{
|
|
OwnerReferences: []metav1.OwnerReference{{Name: "owner-in-template"}},
|
|
},
|
|
Spec: corev1.PodSpec{
|
|
Containers: []corev1.Container{
|
|
{Name: common.Spark3DefaultExecutorContainerName, Image: "image"},
|
|
},
|
|
},
|
|
},
|
|
},
|
|
},
|
|
},
|
|
},
|
|
expectedTemplate: corev1.PodTemplateSpec{
|
|
ObjectMeta: metav1.ObjectMeta{
|
|
OwnerReferences: []metav1.OwnerReference{
|
|
{Name: "owner-in-template"},
|
|
appNonControllerOwnerReference,
|
|
},
|
|
},
|
|
Spec: corev1.PodSpec{
|
|
Containers: []corev1.Container{
|
|
{Name: common.Spark3DefaultExecutorContainerName, Image: "image"},
|
|
},
|
|
},
|
|
},
|
|
},
|
|
"pod template already has owner references": {
|
|
app: &v1beta2.SparkApplication{
|
|
Spec: v1beta2.SparkApplicationSpec{
|
|
Executor: v1beta2.ExecutorSpec{
|
|
SparkPodSpec: v1beta2.SparkPodSpec{
|
|
Template: &corev1.PodTemplateSpec{
|
|
ObjectMeta: metav1.ObjectMeta{
|
|
OwnerReferences: []metav1.OwnerReference{{Name: "owner-in-template"}, appNonControllerOwnerReference},
|
|
},
|
|
},
|
|
},
|
|
},
|
|
},
|
|
},
|
|
expectedTemplate: corev1.PodTemplateSpec{
|
|
ObjectMeta: metav1.ObjectMeta{
|
|
OwnerReferences: []metav1.OwnerReference{
|
|
{Name: "owner-in-template"},
|
|
appNonControllerOwnerReference,
|
|
},
|
|
},
|
|
},
|
|
},
|
|
} {
|
|
t.Run(name, func(t *testing.T) {
|
|
submissionID := "TestExecutorPodTemplateOption-" + strings.ReplaceAll(name, " ", "_") + "-" + uuid.New().String()
|
|
tc.app.Name = appName
|
|
tc.app.UID = uid
|
|
tc.app.Status.SubmissionID = submissionID
|
|
|
|
_, err := executorPodTemplateOption(tc.app)
|
|
assert.NoError(t, err)
|
|
defer func() { _ = os.RemoveAll(fmt.Sprintf("/tmp/spark/%s", submissionID)) }()
|
|
|
|
expectedBytes, err := yaml.Marshal(tc.expectedTemplate)
|
|
assert.NoError(t, err)
|
|
|
|
actualBytes, err := os.ReadFile(fmt.Sprintf("/tmp/spark/%s/executor-pod-template.yaml", submissionID))
|
|
assert.NoError(t, err)
|
|
|
|
assert.Equal(t, string(expectedBytes), string(actualBytes))
|
|
})
|
|
}
|
|
}
|
|
|
|
func TestBuildSparkSubmitArgsSkipPodTemplates(t *testing.T) {
|
|
t.Setenv("KUBERNETES_SERVICE_HOST", "127.0.0.1")
|
|
t.Setenv("KUBERNETES_SERVICE_PORT", "6443")
|
|
|
|
app := &v1beta2.SparkApplication{
|
|
ObjectMeta: metav1.ObjectMeta{
|
|
Name: "test-app",
|
|
Namespace: "default",
|
|
UID: "uid-123",
|
|
},
|
|
Spec: v1beta2.SparkApplicationSpec{
|
|
Type: v1beta2.SparkApplicationTypeScala,
|
|
Mode: v1beta2.DeployModeCluster,
|
|
MainApplicationFile: ptr.To("local:///app.jar"),
|
|
MainClass: ptr.To("org.example.Main"),
|
|
SparkVersion: "4.0.1",
|
|
},
|
|
Status: v1beta2.SparkApplicationStatus{
|
|
SubmissionID: "sub-123",
|
|
},
|
|
}
|
|
|
|
t.Run("includes pod template args when skipPodTemplates is false", func(t *testing.T) {
|
|
args, err := buildSparkSubmitArgs(app, false)
|
|
assert.NoError(t, err)
|
|
|
|
hasDriverTemplate := slices.ContainsFunc(args, func(s string) bool {
|
|
return strings.Contains(s, common.SparkKubernetesDriverPodTemplateFile)
|
|
})
|
|
hasExecutorTemplate := slices.ContainsFunc(args, func(s string) bool {
|
|
return strings.Contains(s, common.SparkKubernetesExecutorPodTemplateFile)
|
|
})
|
|
assert.True(t, hasDriverTemplate, "should include driver pod template conf")
|
|
assert.True(t, hasExecutorTemplate, "should include executor pod template conf")
|
|
})
|
|
|
|
t.Run("skips pod template args when skipPodTemplates is true", func(t *testing.T) {
|
|
args, err := buildSparkSubmitArgs(app, true)
|
|
assert.NoError(t, err)
|
|
|
|
hasDriverTemplate := slices.ContainsFunc(args, func(s string) bool {
|
|
return strings.Contains(s, common.SparkKubernetesDriverPodTemplateFile)
|
|
})
|
|
hasExecutorTemplate := slices.ContainsFunc(args, func(s string) bool {
|
|
return strings.Contains(s, common.SparkKubernetesExecutorPodTemplateFile)
|
|
})
|
|
assert.False(t, hasDriverTemplate, "should not include driver pod template conf")
|
|
assert.False(t, hasExecutorTemplate, "should not include executor pod template conf")
|
|
})
|
|
}
|
|
|
|
// import (
|
|
// "fmt"
|
|
// "os"
|
|
// "reflect"
|
|
// "sort"
|
|
// "strconv"
|
|
// "testing"
|
|
|
|
// "github.com/google/uuid"
|
|
// "github.com/stretchr/testify/assert"
|
|
// corev1 "k8s.io/api/core/v1"
|
|
// "k8s.io/apimachinery/pkg/api/resource"
|
|
// metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
|
|
|
|
// "github.com/kubeflow/spark-operator/v2/api/v1beta2"
|
|
// "github.com/kubeflow/spark-operator/v2/pkg/common"
|
|
// "github.com/kubeflow/spark-operator/v2/pkg/util"
|
|
// )
|
|
|
|
// const (
|
|
// VolumeMountPathTemplate = "spark.kubernetes.%s.volumes.%s.%s.mount.path=%s"
|
|
// VolumeMountOptionPathTemplate = "spark.kubernetes.%s.volumes.%s.%s.options.%s=%s"
|
|
// SparkDriverLabelAnnotationTemplate = "spark.kubernetes.driver.label.sparkoperator.k8s.io/%s=%s"
|
|
// SparkDriverLabelTemplate = "spark.kubernetes.driver.label.%s=%s"
|
|
// SparkDriverServiceLabelTemplate = "spark.kubernetes.driver.service.label.%s=%s"
|
|
// SparkExecutorLabelAnnotationTemplate = "spark.kubernetes.executor.label.sparkoperator.k8s.io/%s=%s"
|
|
// SparkExecutorLabelTemplate = "spark.kubernetes.executor.label.%s=%s"
|
|
// )
|
|
|
|
// func TestAddLocalDir_HostPath(t *testing.T) {
|
|
// volumes := []corev1.Volume{
|
|
// {
|
|
// Name: "spark-local-dir-1",
|
|
// VolumeSource: corev1.VolumeSource{
|
|
// HostPath: &corev1.HostPathVolumeSource{
|
|
// Path: "/tmp/mnt",
|
|
// },
|
|
// },
|
|
// },
|
|
// }
|
|
|
|
// volumeMounts := []corev1.VolumeMount{
|
|
// {
|
|
// Name: "spark-local-dir-1",
|
|
// MountPath: "/tmp/mnt-1",
|
|
// },
|
|
// }
|
|
|
|
// app := &v1beta2.SparkApplication{
|
|
// ObjectMeta: metav1.ObjectMeta{
|
|
// Name: "spark-test",
|
|
// UID: "spark-test-1",
|
|
// },
|
|
// Spec: v1beta2.SparkApplicationSpec{
|
|
// Volumes: volumes,
|
|
// Driver: v1beta2.DriverSpec{
|
|
// SparkPodSpec: v1beta2.SparkPodSpec{
|
|
// VolumeMounts: volumeMounts,
|
|
// },
|
|
// },
|
|
// },
|
|
// }
|
|
|
|
// localDirOptions, err := addLocalDirConfOptions(app)
|
|
// if err != nil {
|
|
// t.Fatal(err)
|
|
// }
|
|
|
|
// assert.Equal(t, 0, len(app.Spec.Volumes))
|
|
// assert.Equal(t, 0, len(app.Spec.Driver.VolumeMounts))
|
|
// assert.Equal(t, 2, len(localDirOptions))
|
|
// assert.Equal(t, fmt.Sprintf(VolumeMountPathTemplate, "driver", "hostPath", volumes[0].Name, volumeMounts[0].MountPath), localDirOptions[0])
|
|
// assert.Equal(t, fmt.Sprintf(VolumeMountOptionPathTemplate, "driver", "hostPath", volumes[0].Name, "path", volumes[0].HostPath.Path), localDirOptions[1])
|
|
// }
|
|
|
|
// func TestAddLocalDir_PVC(t *testing.T) {
|
|
// volumes := []corev1.Volume{
|
|
// {
|
|
// Name: "spark-local-dir-1",
|
|
// VolumeSource: corev1.VolumeSource{
|
|
// PersistentVolumeClaim: &corev1.PersistentVolumeClaimVolumeSource{
|
|
// ClaimName: "/tmp/mnt-1",
|
|
// },
|
|
// },
|
|
// },
|
|
// }
|
|
|
|
// volumeMounts := []corev1.VolumeMount{
|
|
// {
|
|
// Name: "spark-local-dir-1",
|
|
// MountPath: "/tmp/mnt-1",
|
|
// },
|
|
// }
|
|
|
|
// app := &v1beta2.SparkApplication{
|
|
// ObjectMeta: metav1.ObjectMeta{
|
|
// Name: "spark-test",
|
|
// UID: "spark-test-1",
|
|
// },
|
|
// Spec: v1beta2.SparkApplicationSpec{
|
|
// Volumes: volumes,
|
|
// Driver: v1beta2.DriverSpec{
|
|
// SparkPodSpec: v1beta2.SparkPodSpec{
|
|
// VolumeMounts: volumeMounts,
|
|
// },
|
|
// },
|
|
// },
|
|
// }
|
|
|
|
// localDirOptions, err := addLocalDirConfOptions(app)
|
|
// if err != nil {
|
|
// t.Fatal(err)
|
|
// }
|
|
|
|
// assert.Equal(t, 0, len(app.Spec.Volumes))
|
|
// assert.Equal(t, 0, len(app.Spec.Driver.VolumeMounts))
|
|
// assert.Equal(t, 2, len(localDirOptions))
|
|
// assert.Equal(t, fmt.Sprintf(VolumeMountPathTemplate, "driver", "persistentVolumeClaim", volumes[0].Name, volumeMounts[0].MountPath), localDirOptions[0])
|
|
// assert.Equal(t, fmt.Sprintf(VolumeMountOptionPathTemplate, "driver", "persistentVolumeClaim", volumes[0].Name, "claimName", volumes[0].PersistentVolumeClaim.ClaimName), localDirOptions[1])
|
|
// }
|
|
|
|
// func TestAddLocalDir_MixedVolumes(t *testing.T) {
|
|
// volumes := []corev1.Volume{
|
|
// {
|
|
// Name: "spark-local-dir-1",
|
|
// VolumeSource: corev1.VolumeSource{
|
|
// HostPath: &corev1.HostPathVolumeSource{
|
|
// Path: "/tmp/mnt-1",
|
|
// },
|
|
// },
|
|
// },
|
|
// {
|
|
// Name: "log-dir",
|
|
// VolumeSource: corev1.VolumeSource{
|
|
// HostPath: &corev1.HostPathVolumeSource{
|
|
// Path: "/var/log/spark",
|
|
// },
|
|
// },
|
|
// },
|
|
// }
|
|
|
|
// volumeMounts := []corev1.VolumeMount{
|
|
// {
|
|
// Name: "spark-local-dir-1",
|
|
// MountPath: "/tmp/mnt-1",
|
|
// },
|
|
// {
|
|
// Name: "log-dir",
|
|
// MountPath: "/var/log/spark",
|
|
// },
|
|
// }
|
|
|
|
// app := &v1beta2.SparkApplication{
|
|
// ObjectMeta: metav1.ObjectMeta{
|
|
// Name: "spark-test",
|
|
// UID: "spark-test-1",
|
|
// },
|
|
// Spec: v1beta2.SparkApplicationSpec{
|
|
// Volumes: volumes,
|
|
// Driver: v1beta2.DriverSpec{
|
|
// SparkPodSpec: v1beta2.SparkPodSpec{
|
|
// VolumeMounts: volumeMounts,
|
|
// },
|
|
// },
|
|
// },
|
|
// }
|
|
|
|
// localDirOptions, err := addLocalDirConfOptions(app)
|
|
// if err != nil {
|
|
// t.Fatal(err)
|
|
// }
|
|
|
|
// assert.Equal(t, 1, len(app.Spec.Volumes))
|
|
// assert.Equal(t, 1, len(app.Spec.Driver.VolumeMounts))
|
|
// assert.Equal(t, 2, len(localDirOptions))
|
|
// assert.Equal(t, fmt.Sprintf(VolumeMountPathTemplate, "driver", "hostPath", volumes[0].Name, volumeMounts[0].MountPath), localDirOptions[0])
|
|
// assert.Equal(t, fmt.Sprintf(VolumeMountOptionPathTemplate, "driver", "hostPath", volumes[0].Name, "path", volumes[0].HostPath.Path), localDirOptions[1])
|
|
// }
|
|
|
|
// func TestAddLocalDir_MultipleScratchVolumes(t *testing.T) {
|
|
// volumes := []corev1.Volume{
|
|
// {
|
|
// Name: "spark-local-dir-1",
|
|
// VolumeSource: corev1.VolumeSource{
|
|
// HostPath: &corev1.HostPathVolumeSource{
|
|
// Path: "/tmp/mnt-1",
|
|
// },
|
|
// },
|
|
// },
|
|
// {
|
|
// Name: "spark-local-dir-2",
|
|
// VolumeSource: corev1.VolumeSource{
|
|
// HostPath: &corev1.HostPathVolumeSource{
|
|
// Path: "/tmp/mnt-2",
|
|
// },
|
|
// },
|
|
// },
|
|
// }
|
|
|
|
// volumeMounts := []corev1.VolumeMount{
|
|
// {
|
|
// Name: "spark-local-dir-1",
|
|
// MountPath: "/tmp/mnt-1",
|
|
// },
|
|
// {
|
|
// Name: "spark-local-dir-2",
|
|
// MountPath: "/tmp/mnt-2",
|
|
// },
|
|
// }
|
|
|
|
// app := &v1beta2.SparkApplication{
|
|
// ObjectMeta: metav1.ObjectMeta{
|
|
// Name: "spark-test",
|
|
// UID: "spark-test-1",
|
|
// },
|
|
// Spec: v1beta2.SparkApplicationSpec{
|
|
// Volumes: volumes,
|
|
// Driver: v1beta2.DriverSpec{
|
|
// SparkPodSpec: v1beta2.SparkPodSpec{
|
|
// VolumeMounts: volumeMounts,
|
|
// },
|
|
// },
|
|
// },
|
|
// }
|
|
|
|
// localDirOptions, err := addLocalDirConfOptions(app)
|
|
// if err != nil {
|
|
// t.Fatal(err)
|
|
// }
|
|
|
|
// assert.Equal(t, 0, len(app.Spec.Volumes))
|
|
// assert.Equal(t, 0, len(app.Spec.Driver.VolumeMounts))
|
|
// assert.Equal(t, 4, len(localDirOptions))
|
|
// assert.Equal(t, fmt.Sprintf(VolumeMountPathTemplate, "driver", "hostPath", volumes[0].Name, volumeMounts[0].MountPath), localDirOptions[0])
|
|
// assert.Equal(t, fmt.Sprintf(VolumeMountOptionPathTemplate, "driver", "hostPath", volumes[0].Name, "path", volumes[0].HostPath.Path), localDirOptions[1])
|
|
// assert.Equal(t, fmt.Sprintf(VolumeMountPathTemplate, "driver", "hostPath", volumes[1].Name, volumeMounts[1].MountPath), localDirOptions[2])
|
|
// assert.Equal(t, fmt.Sprintf(VolumeMountOptionPathTemplate, "driver", "hostPath", volumes[1].Name, "path", volumes[1].HostPath.Path), localDirOptions[3])
|
|
// }
|
|
|
|
// func TestAddLocalDir_Executor(t *testing.T) {
|
|
// volumes := []corev1.Volume{
|
|
// {
|
|
// Name: "spark-local-dir-1",
|
|
// VolumeSource: corev1.VolumeSource{
|
|
// HostPath: &corev1.HostPathVolumeSource{
|
|
// Path: "/tmp/mnt",
|
|
// },
|
|
// },
|
|
// },
|
|
// }
|
|
|
|
// volumeMounts := []corev1.VolumeMount{
|
|
// {
|
|
// Name: "spark-local-dir-1",
|
|
// MountPath: "/tmp/mnt-1",
|
|
// },
|
|
// }
|
|
|
|
// app := &v1beta2.SparkApplication{
|
|
// ObjectMeta: metav1.ObjectMeta{
|
|
// Name: "spark-test",
|
|
// UID: "spark-test-1",
|
|
// },
|
|
// Spec: v1beta2.SparkApplicationSpec{
|
|
// Volumes: volumes,
|
|
// Executor: v1beta2.ExecutorSpec{
|
|
// SparkPodSpec: v1beta2.SparkPodSpec{
|
|
// VolumeMounts: volumeMounts,
|
|
// },
|
|
// },
|
|
// },
|
|
// }
|
|
|
|
// localDirOptions, err := addLocalDirConfOptions(app)
|
|
// if err != nil {
|
|
// t.Fatal(err)
|
|
// }
|
|
|
|
// assert.Equal(t, 0, len(app.Spec.Volumes))
|
|
// assert.Equal(t, 0, len(app.Spec.Executor.VolumeMounts))
|
|
// assert.Equal(t, 2, len(localDirOptions))
|
|
// assert.Equal(t, fmt.Sprintf(VolumeMountPathTemplate, "executor", "hostPath", volumes[0].Name, volumeMounts[0].MountPath), localDirOptions[0])
|
|
// assert.Equal(t, fmt.Sprintf(VolumeMountOptionPathTemplate, "executor", "hostPath", volumes[0].Name, "path", volumes[0].HostPath.Path), localDirOptions[1])
|
|
// }
|
|
|
|
// func TestAddLocalDir_Driver_Executor(t *testing.T) {
|
|
// volumes := []corev1.Volume{
|
|
// {
|
|
// Name: "spark-local-dir-1",
|
|
// VolumeSource: corev1.VolumeSource{
|
|
// HostPath: &corev1.HostPathVolumeSource{
|
|
// Path: "/tmp/mnt",
|
|
// },
|
|
// },
|
|
// },
|
|
// {
|
|
// Name: "test-volume",
|
|
// VolumeSource: corev1.VolumeSource{
|
|
// HostPath: &corev1.HostPathVolumeSource{
|
|
// Path: "/tmp/test",
|
|
// },
|
|
// },
|
|
// },
|
|
// }
|
|
|
|
// volumeMounts := []corev1.VolumeMount{
|
|
// {
|
|
// Name: "spark-local-dir-1",
|
|
// MountPath: "/tmp/mnt-1",
|
|
// },
|
|
// {
|
|
// Name: "test-volume",
|
|
// MountPath: "/tmp/test",
|
|
// },
|
|
// }
|
|
|
|
// app := &v1beta2.SparkApplication{
|
|
// ObjectMeta: metav1.ObjectMeta{
|
|
// Name: "spark-test",
|
|
// UID: "spark-test-1",
|
|
// },
|
|
// Spec: v1beta2.SparkApplicationSpec{
|
|
// Volumes: volumes,
|
|
// Driver: v1beta2.DriverSpec{
|
|
// SparkPodSpec: v1beta2.SparkPodSpec{
|
|
// VolumeMounts: volumeMounts,
|
|
// },
|
|
// },
|
|
// Executor: v1beta2.ExecutorSpec{
|
|
// SparkPodSpec: v1beta2.SparkPodSpec{
|
|
// VolumeMounts: volumeMounts,
|
|
// },
|
|
// },
|
|
// },
|
|
// }
|
|
|
|
// localDirOptions, err := addLocalDirConfOptions(app)
|
|
// if err != nil {
|
|
// t.Fatal(err)
|
|
// }
|
|
|
|
// assert.Equal(t, 1, len(app.Spec.Volumes))
|
|
// assert.Equal(t, 1, len(app.Spec.Driver.VolumeMounts))
|
|
// assert.Equal(t, 1, len(app.Spec.Executor.VolumeMounts))
|
|
// assert.Equal(t, 4, len(localDirOptions))
|
|
// assert.Equal(t, fmt.Sprintf(VolumeMountPathTemplate, "driver", "hostPath", volumes[0].Name, volumeMounts[0].MountPath), localDirOptions[0])
|
|
// assert.Equal(t, fmt.Sprintf(VolumeMountOptionPathTemplate, "driver", "hostPath", volumes[0].Name, "path", volumes[0].HostPath.Path), localDirOptions[1])
|
|
// assert.Equal(t, fmt.Sprintf(VolumeMountPathTemplate, "executor", "hostPath", volumes[0].Name, volumeMounts[0].MountPath), localDirOptions[2])
|
|
// assert.Equal(t, fmt.Sprintf(VolumeMountOptionPathTemplate, "executor", "hostPath", volumes[0].Name, "path", volumes[0].HostPath.Path), localDirOptions[3])
|
|
// }
|
|
|
|
// func TestAddEmptyDir_Driver_Executor_WithSizeLimit(t *testing.T) {
|
|
// sizeLimit := resource.MustParse("5Gi")
|
|
// volumes := []corev1.Volume{
|
|
// {
|
|
// Name: "spark-local-dir-1",
|
|
// VolumeSource: corev1.VolumeSource{
|
|
// EmptyDir: &corev1.EmptyDirVolumeSource{
|
|
// SizeLimit: &sizeLimit,
|
|
// },
|
|
// },
|
|
// },
|
|
// }
|
|
|
|
// volumeMounts := []corev1.VolumeMount{
|
|
// {
|
|
// Name: "spark-local-dir-1",
|
|
// MountPath: "/tmp/mnt-1",
|
|
// },
|
|
// }
|
|
|
|
// app := &v1beta2.SparkApplication{
|
|
// ObjectMeta: metav1.ObjectMeta{
|
|
// Name: "spark-test",
|
|
// UID: "spark-test-1",
|
|
// },
|
|
// Spec: v1beta2.SparkApplicationSpec{
|
|
// Volumes: volumes,
|
|
// Driver: v1beta2.DriverSpec{
|
|
// SparkPodSpec: v1beta2.SparkPodSpec{
|
|
// VolumeMounts: volumeMounts,
|
|
// },
|
|
// },
|
|
// Executor: v1beta2.ExecutorSpec{
|
|
// SparkPodSpec: v1beta2.SparkPodSpec{
|
|
// VolumeMounts: volumeMounts,
|
|
// },
|
|
// },
|
|
// },
|
|
// }
|
|
|
|
// localDirOptions, err := addLocalDirConfOptions(app)
|
|
// if err != nil {
|
|
// t.Fatal(err)
|
|
// }
|
|
|
|
// assert.Equal(t, 0, len(app.Spec.Volumes))
|
|
// assert.Equal(t, 0, len(app.Spec.Driver.VolumeMounts))
|
|
// assert.Equal(t, 0, len(app.Spec.Executor.VolumeMounts))
|
|
// assert.Equal(t, 4, len(localDirOptions))
|
|
// assert.Equal(t, fmt.Sprintf(VolumeMountPathTemplate, "driver", "emptyDir", volumes[0].Name, volumeMounts[0].MountPath), localDirOptions[0])
|
|
// assert.Equal(t, fmt.Sprintf(VolumeMountOptionPathTemplate, "driver", "emptyDir", volumes[0].Name, "sizeLimit", volumes[0].EmptyDir.SizeLimit.String()), localDirOptions[1])
|
|
// assert.Equal(t, fmt.Sprintf(VolumeMountPathTemplate, "executor", "emptyDir", volumes[0].Name, volumeMounts[0].MountPath), localDirOptions[2])
|
|
// assert.Equal(t, fmt.Sprintf(VolumeMountOptionPathTemplate, "executor", "emptyDir", volumes[0].Name, "sizeLimit", volumes[0].EmptyDir.SizeLimit.String()), localDirOptions[3])
|
|
// }
|
|
|
|
// func TestPopulateLabels_Driver_Executor(t *testing.T) {
|
|
// const (
|
|
// AppLabelKey = "app-label-key"
|
|
// AppLabelValue = "app-label-value"
|
|
// DriverLabelKey = "driver-label-key"
|
|
// DriverLabelValue = "driver-label-key"
|
|
// DriverServiceLabelKey = "driver-svc-label-key"
|
|
// DriverServiceLabelValue = "driver-svc-label-value"
|
|
// ExecutorLabelKey = "executor-label-key"
|
|
// ExecutorLabelValue = "executor-label-key"
|
|
// )
|
|
|
|
// app := &v1beta2.SparkApplication{
|
|
// ObjectMeta: metav1.ObjectMeta{
|
|
// Name: "spark-test",
|
|
// UID: "spark-test-1",
|
|
// Labels: map[string]string{AppLabelKey: AppLabelValue},
|
|
// },
|
|
// Spec: v1beta2.SparkApplicationSpec{
|
|
// Driver: v1beta2.DriverSpec{
|
|
// ServiceLabels: map[string]string{DriverServiceLabelKey: DriverServiceLabelValue},
|
|
// SparkPodSpec: v1beta2.SparkPodSpec{
|
|
// Labels: map[string]string{DriverLabelKey: DriverLabelValue},
|
|
// },
|
|
// },
|
|
// Executor: v1beta2.ExecutorSpec{
|
|
// SparkPodSpec: v1beta2.SparkPodSpec{
|
|
// Labels: map[string]string{ExecutorLabelKey: ExecutorLabelValue},
|
|
// },
|
|
// },
|
|
// },
|
|
// }
|
|
|
|
// submissionID := uuid.New().String()
|
|
// driverOptions, err := addDriverConfOptions(app, submissionID)
|
|
// if err != nil {
|
|
// t.Fatal(err)
|
|
// }
|
|
// assert.Equal(t, 6, len(driverOptions))
|
|
// sort.Strings(driverOptions)
|
|
// expectedDriverLabels := []string{
|
|
// fmt.Sprintf(SparkDriverLabelAnnotationTemplate, "launched-by-spark-operator", strconv.FormatBool(true)),
|
|
// fmt.Sprintf(SparkDriverLabelAnnotationTemplate, "app-name", "spark-test"),
|
|
// fmt.Sprintf(SparkDriverLabelAnnotationTemplate, "submission-id", submissionID),
|
|
// fmt.Sprintf(SparkDriverLabelTemplate, AppLabelKey, AppLabelValue),
|
|
// fmt.Sprintf(SparkDriverLabelTemplate, DriverLabelKey, DriverLabelValue),
|
|
// fmt.Sprintf(SparkDriverServiceLabelTemplate, DriverServiceLabelKey, DriverServiceLabelValue),
|
|
// }
|
|
// sort.Strings(expectedDriverLabels)
|
|
|
|
// if !reflect.DeepEqual(expectedDriverLabels, driverOptions) {
|
|
// t.Errorf("Executor labels: wanted %+q got %+q", expectedDriverLabels, driverOptions)
|
|
// }
|
|
|
|
// executorOptions, err := addExecutorConfOptions(app, submissionID)
|
|
// sort.Strings(executorOptions)
|
|
// if err != nil {
|
|
// t.Fatal(err)
|
|
// }
|
|
// assert.Equal(t, 5, len(executorOptions))
|
|
// expectedExecutorLabels := []string{
|
|
// fmt.Sprintf(SparkExecutorLabelAnnotationTemplate, "app-name", "spark-test"),
|
|
// fmt.Sprintf(SparkExecutorLabelAnnotationTemplate, "launched-by-spark-operator", strconv.FormatBool(true)),
|
|
// fmt.Sprintf(SparkExecutorLabelAnnotationTemplate, "submission-id", submissionID),
|
|
// fmt.Sprintf(SparkExecutorLabelTemplate, AppLabelKey, AppLabelValue),
|
|
// fmt.Sprintf(SparkExecutorLabelTemplate, ExecutorLabelKey, ExecutorLabelValue),
|
|
// }
|
|
// sort.Strings(expectedExecutorLabels)
|
|
|
|
// if !reflect.DeepEqual(expectedExecutorLabels, executorOptions) {
|
|
// t.Errorf("Executor labels: wanted %+q got %+q", expectedExecutorLabels, executorOptions)
|
|
// }
|
|
// }
|
|
|
|
// func TestPopulateLabelsOverride_Driver_Executor(t *testing.T) {
|
|
// const (
|
|
// AppLabelKey = "app-label-key"
|
|
// AppLabelValue = "app-label-value"
|
|
// DriverLabelKey = "driver-label-key"
|
|
// DriverLabelValue = "driver-label-key"
|
|
// DriverAppLabelOverride = "driver-app-label-override"
|
|
// ExecutorLabelKey = "executor-label-key"
|
|
// ExecutorLabelValue = "executor-label-key"
|
|
// ExecutorAppLabelOverride = "executor-app-label-override"
|
|
// )
|
|
|
|
// app := &v1beta2.SparkApplication{
|
|
// ObjectMeta: metav1.ObjectMeta{
|
|
// Name: "spark-test",
|
|
// UID: "spark-test-1",
|
|
// Labels: map[string]string{AppLabelKey: AppLabelValue},
|
|
// },
|
|
// Spec: v1beta2.SparkApplicationSpec{
|
|
// Driver: v1beta2.DriverSpec{
|
|
// SparkPodSpec: v1beta2.SparkPodSpec{
|
|
// Labels: map[string]string{DriverLabelKey: DriverLabelValue, AppLabelKey: DriverAppLabelOverride},
|
|
// },
|
|
// },
|
|
// Executor: v1beta2.ExecutorSpec{
|
|
// SparkPodSpec: v1beta2.SparkPodSpec{
|
|
// Labels: map[string]string{ExecutorLabelKey: ExecutorLabelValue, AppLabelKey: ExecutorAppLabelOverride},
|
|
// },
|
|
// },
|
|
// },
|
|
// }
|
|
|
|
// submissionID := uuid.New().String()
|
|
// driverOptions, err := addDriverConfOptions(app, submissionID)
|
|
// if err != nil {
|
|
// t.Fatal(err)
|
|
// }
|
|
// sort.Strings(driverOptions)
|
|
// assert.Equal(t, 5, len(driverOptions))
|
|
// expectedDriverLabels := []string{
|
|
// fmt.Sprintf(SparkDriverLabelTemplate, AppLabelKey, DriverAppLabelOverride),
|
|
// fmt.Sprintf(SparkDriverLabelTemplate, DriverLabelKey, DriverLabelValue),
|
|
// fmt.Sprintf(SparkDriverLabelAnnotationTemplate, "app-name", "spark-test"),
|
|
// fmt.Sprintf(SparkDriverLabelAnnotationTemplate, "launched-by-spark-operator", strconv.FormatBool(true)),
|
|
// fmt.Sprintf(SparkDriverLabelAnnotationTemplate, "submission-id", submissionID),
|
|
// }
|
|
// sort.Strings(expectedDriverLabels)
|
|
|
|
// if !reflect.DeepEqual(expectedDriverLabels, driverOptions) {
|
|
// t.Errorf("Executor labels: wanted %+q got %+q", expectedDriverLabels, driverOptions)
|
|
// }
|
|
|
|
// executorOptions, err := addExecutorConfOptions(app, submissionID)
|
|
// if err != nil {
|
|
// t.Fatal(err)
|
|
// }
|
|
// sort.Strings(executorOptions)
|
|
// assert.Equal(t, 5, len(executorOptions))
|
|
// expectedExecutorLabels := []string{
|
|
// fmt.Sprintf(SparkExecutorLabelTemplate, AppLabelKey, ExecutorAppLabelOverride),
|
|
// fmt.Sprintf(SparkExecutorLabelTemplate, ExecutorLabelKey, ExecutorLabelValue),
|
|
// fmt.Sprintf(SparkExecutorLabelAnnotationTemplate, "launched-by-spark-operator", strconv.FormatBool(true)),
|
|
// fmt.Sprintf(SparkExecutorLabelAnnotationTemplate, "app-name", "spark-test"),
|
|
// fmt.Sprintf(SparkExecutorLabelAnnotationTemplate, "submission-id", submissionID),
|
|
// }
|
|
// sort.Strings(expectedExecutorLabels)
|
|
|
|
// if !reflect.DeepEqual(expectedExecutorLabels, executorOptions) {
|
|
// t.Errorf("Executor labels: wanted %+q got %+q", expectedExecutorLabels, executorOptions)
|
|
// }
|
|
// }
|
|
|
|
// func TestDynamicAllocationOptions(t *testing.T) {
|
|
// app := &v1beta2.SparkApplication{
|
|
// ObjectMeta: metav1.ObjectMeta{
|
|
// Name: "spark-test",
|
|
// UID: "spark-test-1",
|
|
// },
|
|
// Spec: v1beta2.SparkApplicationSpec{},
|
|
// }
|
|
// options := addDynamicAllocationConfOptions(app)
|
|
// assert.Equal(t, 0, len(options))
|
|
|
|
// app = &v1beta2.SparkApplication{
|
|
// ObjectMeta: metav1.ObjectMeta{
|
|
// Name: "spark-test",
|
|
// UID: "spark-test-1",
|
|
// },
|
|
// Spec: v1beta2.SparkApplicationSpec{
|
|
// DynamicAllocation: &v1beta2.DynamicAllocation{
|
|
// Enabled: true,
|
|
// InitialExecutors: ptr.To[int32](2),
|
|
// MinExecutors: ptr.To[int32](0),
|
|
// MaxExecutors: ptr.To[int32](10),
|
|
// ShuffleTrackingTimeout: ptr.To[int64](6000000),
|
|
// },
|
|
// },
|
|
// }
|
|
|
|
// options = addDynamicAllocationConfOptions(app)
|
|
// assert.Equal(t, 6, len(options))
|
|
// assert.Equal(t, fmt.Sprintf("%s=true", common.SparkDynamicAllocationEnabled), options[0])
|
|
// assert.Equal(t, fmt.Sprintf("%s=true", common.SparkDynamicAllocationShuffleTrackingEnabled), options[1])
|
|
// assert.Equal(t, fmt.Sprintf("%s=2", common.SparkDynamicAllocationInitialExecutors), options[2])
|
|
// assert.Equal(t, fmt.Sprintf("%s=0", common.SparkDynamicAllocationMinExecutors), options[3])
|
|
// assert.Equal(t, fmt.Sprintf("%s=10", common.SparkDynamicAllocationMaxExecutors), options[4])
|
|
// assert.Equal(t, fmt.Sprintf("%s=6000000", common.SparkDynamicAllocationShuffleTrackingTimeout), options[5])
|
|
// }
|
|
|
|
// func TestProxyUserArg(t *testing.T) {
|
|
// const (
|
|
// host = "localhost"
|
|
// port = "6443"
|
|
// )
|
|
|
|
// if err := os.Setenv(common.EnvKubernetesServiceHost, host); err != nil {
|
|
// t.Fatal(err)
|
|
// }
|
|
// if err := os.Setenv(common.EnvKubernetesServicePort, port); err != nil {
|
|
// t.Fatal(err)
|
|
// }
|
|
|
|
// app := &v1beta2.SparkApplication{
|
|
// ObjectMeta: metav1.ObjectMeta{
|
|
// Name: "spark-test",
|
|
// UID: "spark-test-1",
|
|
// },
|
|
// Spec: v1beta2.SparkApplicationSpec{
|
|
// Mode: v1beta2.ClusterMode,
|
|
// ProxyUser: ptr.To("foo"),
|
|
// },
|
|
// }
|
|
|
|
// submissionID := uuid.New().String()
|
|
// driverPodName := app.GetDriverPodName()
|
|
// args, err := buildSubmissionCommandArgs(app, driverPodName, submissionID)
|
|
// if err != nil {
|
|
// t.Fatal(err)
|
|
// }
|
|
|
|
// assert.Equal(t, "--master", args[0])
|
|
// assert.Equal(t, fmt.Sprintf("k8s://https://%s:%s", host, port), args[1])
|
|
// assert.Equal(t, "--deploy-mode", args[2])
|
|
// assert.Equal(t, string(v1beta2.ClusterMode), args[3])
|
|
// assert.Equal(t, "--proxy-user", args[4])
|
|
// assert.Equal(t, "foo", args[5])
|
|
// }
|
|
|
|
// func Test_getMasterURL(t *testing.T) {
|
|
// setEnv := func(host string, port string) {
|
|
// if err := os.Setenv(common.EnvKubernetesServiceHost, host); err != nil {
|
|
// t.Fatal(err)
|
|
// }
|
|
// if err := os.Setenv(common.EnvKubernetesServicePort, port); err != nil {
|
|
// t.Fatal(err)
|
|
// }
|
|
// }
|
|
|
|
// tests := []struct {
|
|
// name string
|
|
// host string
|
|
// port string
|
|
// want string
|
|
// wantErr assert.ErrorAssertionFunc
|
|
// }{
|
|
// {
|
|
// name: "should return a valid master url when IPv4 address is used",
|
|
// host: "localhost",
|
|
// port: "6443",
|
|
// want: "k8s://https://localhost:6443",
|
|
// wantErr: assert.NoError,
|
|
// },
|
|
// {
|
|
// name: "should return a valid master url when IPv6 address is used",
|
|
// host: "::1",
|
|
// port: "6443",
|
|
// want: "k8s://https://[::1]:6443",
|
|
// wantErr: assert.NoError,
|
|
// },
|
|
// {
|
|
// name: "should throw an error when the host is empty",
|
|
// host: "",
|
|
// port: "6443",
|
|
// want: "",
|
|
// wantErr: assert.Error,
|
|
// },
|
|
// }
|
|
// for _, tt := range tests {
|
|
// t.Run(tt.name, func(t *testing.T) {
|
|
// setEnv(tt.host, tt.port)
|
|
// got, err := getMasterURL()
|
|
// if !tt.wantErr(t, err, "getMasterURL()") {
|
|
// return
|
|
// }
|
|
// assert.Equalf(t, tt.want, got, "getMasterURL()")
|
|
// })
|
|
// }
|
|
// }
|