fix(controller): keep scheduled run names within DNS label limit

internal/controller/scheduledsparkapplication/controller.go

@@ -273,9 +273,15 @@ func (r *Reconciler) createSparkApplication(
 	for key, value := range scheduledApp.Labels {
 		labels[key] = value
 	}
+	timestamp := formatTimestamp(t, r.options.TimestampPrecision)
+	runName := util.GetScheduledSparkApplicationRunName(scheduledApp, timestamp)
+	if originalName := fmt.Sprintf("%s-%s", scheduledApp.Name, timestamp); runName != originalName {
+		logger.Info("Shortened scheduled SparkApplication run name", "name", scheduledApp.Name, "namespace", scheduledApp.Namespace,
+			"originalName", originalName, "runName", runName)
+	}
 	app := &v1beta2.SparkApplication{
 		ObjectMeta: metav1.ObjectMeta{
-			Name:      fmt.Sprintf("%s-%s", scheduledApp.Name, formatTimestamp(t, r.options.TimestampPrecision)),
+			Name:      runName,
 			Namespace: scheduledApp.Namespace,
 			Labels:    labels,
 			OwnerReferences: []metav1.OwnerReference{{

internal/controller/scheduledsparkapplication/controller_test.go

@@ -18,6 +18,7 @@ package scheduledsparkapplication
 
 import (
 	"context"
+	"strings"
 	"time"
 
 	. "github.com/onsi/ginkgo/v2"
@@ -26,11 +27,13 @@ import (
 	"k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/types"
+	"k8s.io/apimachinery/pkg/util/validation"
 	"k8s.io/utils/clock"
 	"k8s.io/utils/ptr"
 	"sigs.k8s.io/controller-runtime/pkg/reconcile"
 
 	"github.com/kubeflow/spark-operator/v2/api/v1beta2"
+	"github.com/kubeflow/spark-operator/v2/internal/webhook"
 )
 
 var _ = Describe("ScheduledSparkApplication Controller", func() {
@@ -134,3 +137,37 @@ var _ = Describe("formatTimestamp", func() {
 		Expect(result).To(Equal("1234567890123456789"))
 	})
 })
+
+var _ = Describe("createSparkApplication", func() {
+	It("should create a valid run name for a long accepted schedule name", func(ctx SpecContext) {
+		scheduledApp := &v1beta2.ScheduledSparkApplication{
+			ObjectMeta: metav1.ObjectMeta{
+				Name:      strings.Repeat("a", 55),
+				Namespace: "default",
+				UID:       "test-scheduled-app-uid",
+			},
+			Spec: v1beta2.ScheduledSparkApplicationSpec{
+				Schedule: "@every 1h",
+				Template: v1beta2.SparkApplicationSpec{
+					Type:                v1beta2.SparkApplicationTypeScala,
+					MainApplicationFile: ptr.To("local:///dummy.jar"),
+				},
+			},
+		}
+
+		_, err := webhook.NewScheduledSparkApplicationValidator().ValidateCreate(ctx, scheduledApp)
+		Expect(err).NotTo(HaveOccurred())
+
+		reconciler := NewReconciler(k8sClient.Scheme(), k8sClient, nil, clock.RealClock{}, Options{TimestampPrecision: "minutes"})
+		app, err := reconciler.createSparkApplication(scheduledApp, time.Unix(1_800_000_000, 0))
+		Expect(err).NotTo(HaveOccurred())
+		DeferCleanup(func() {
+			Expect(k8sClient.Delete(context.Background(), app)).To(Succeed())
+		})
+
+		Expect(app.Name).To(HaveLen(63))
+		Expect(validation.IsDNS1035Label(app.Name)).To(BeEmpty())
+		_, err = webhook.NewSparkApplicationValidator(nil, false).ValidateCreate(ctx, app)
+		Expect(err).NotTo(HaveOccurred())
+	})
+})

pkg/util/sparkapplication.go

@@ -201,6 +201,18 @@ func generateName(name, suffix string) string {
 	return fmt.Sprintf("%s-%s-%s", name[:maxNameLength], hash[:8], suffix)
 }
 
+// GetScheduledSparkApplicationRunName preserves the timestamp and hashes only the schedule name when truncating.
+func GetScheduledSparkApplicationRunName(app *v1beta2.ScheduledSparkApplication, timestamp string) string {
+	preferredName := fmt.Sprintf("%s-%s", app.Name, timestamp)
+	if len(preferredName) <= 63 {
+		return preferredName
+	}
+
+	maxNameLength := 63 - len(timestamp) - 10 // 8 for the hash, 2 for the dashes.
+	hash := fmt.Sprintf("%x", md5.Sum([]byte(app.Name)))
+	return fmt.Sprintf("%s-%s-%s", app.Name[:maxNameLength], hash[:8], timestamp)
+}
+
 func GetDefaultUIServiceName(app *v1beta2.SparkApplication) string {
 	return generateName(app.Name, "ui-svc")
 }

pkg/util/sparkapplication_test.go

@@ -17,20 +17,76 @@ limitations under the License.
 package util_test
 
 import (
+	"strings"
 	"time"
 
 	. "github.com/onsi/ginkgo/v2"
 	. "github.com/onsi/gomega"
 
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/util/validation"
 	"k8s.io/utils/ptr"
 
 	"github.com/kubeflow/spark-operator/v2/api/v1beta2"
 	"github.com/kubeflow/spark-operator/v2/pkg/common"
 	"github.com/kubeflow/spark-operator/v2/pkg/util"
 )
 
+var _ = Describe("GetScheduledSparkApplicationRunName", func() {
+	It("Should preserve names that fit", func() {
+		app := &v1beta2.ScheduledSparkApplication{ObjectMeta: metav1.ObjectMeta{Name: "scheduled-app"}}
+		Expect(util.GetScheduledSparkApplicationRunName(app, "1234567890")).To(Equal("scheduled-app-1234567890"))
+	})
+
+	It("Should generate valid unique names when truncation is needed", func() {
+		prefix := strings.Repeat("a", 50)
+		firstApp := &v1beta2.ScheduledSparkApplication{ObjectMeta: metav1.ObjectMeta{Name: prefix + "first"}}
+		secondApp := &v1beta2.ScheduledSparkApplication{ObjectMeta: metav1.ObjectMeta{Name: prefix + "secon"}}
+		first := util.GetScheduledSparkApplicationRunName(firstApp, "1234567890123456789")
+		second := util.GetScheduledSparkApplicationRunName(secondApp, "1234567890123456789")
+
+		Expect(first).To(HaveLen(63))
+		Expect(validation.IsDNS1035Label(first)).To(BeEmpty())
+		Expect(first).To(HaveSuffix("-1234567890123456789"))
+		Expect(first).NotTo(Equal(second))
+	})
+
+	DescribeTable("Should preserve names at the length limit",
+		func(timestamp string) {
+			app := &v1beta2.ScheduledSparkApplication{ObjectMeta: metav1.ObjectMeta{Name: strings.Repeat("a", 62-len(timestamp))}}
+			Expect(util.GetScheduledSparkApplicationRunName(app, timestamp)).To(Equal(app.Name + "-" + timestamp))
+		},
+		Entry("minutes", "30000000"),
+		Entry("seconds", "1800000000"),
+		Entry("milliseconds", "1800000000000"),
+		Entry("microseconds", "1800000000000000"),
+		Entry("nanoseconds", "1800000000000000000"),
+	)
+
+	DescribeTable("Should keep a stable prefix and timestamp order across runs",
+		func(firstTimestamp, secondTimestamp string) {
+			app := &v1beta2.ScheduledSparkApplication{ObjectMeta: metav1.ObjectMeta{Name: strings.Repeat("a", 55)}}
+			first := util.GetScheduledSparkApplicationRunName(app, firstTimestamp)
+			second := util.GetScheduledSparkApplicationRunName(app, secondTimestamp)
+
+			Expect(first).To(HaveLen(63))
+			Expect(second).To(HaveLen(63))
+			Expect(validation.IsDNS1035Label(first)).To(BeEmpty())
+			Expect(validation.IsDNS1035Label(second)).To(BeEmpty())
+			Expect(first).To(HaveSuffix("-" + firstTimestamp))
+			Expect(second).To(HaveSuffix("-" + secondTimestamp))
+			Expect(strings.TrimSuffix(first, firstTimestamp)).To(Equal(strings.TrimSuffix(second, secondTimestamp)))
+			Expect(first < second).To(BeTrue())
+		},
+		Entry("minutes", "30000000", "30000001"),
+		Entry("seconds", "1800000000", "1800000001"),
+		Entry("milliseconds", "1800000000000", "1800000000001"),
+		Entry("microseconds", "1800000000000000", "1800000000000001"),
+		Entry("nanoseconds", "1800000000000000000", "1800000000000000001"),
+	)
+})
+
 var _ = Describe("GetDriverPodName", func() {
 	Context("SparkApplication without driver pod name field and driver pod name conf", func() {
 		app := &v1beta2.SparkApplication{