[FLINK-28087] Add validation for the meta.name of FlinkDeployment CR

Author: Grypse

flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/validation/DefaultValidator.java

@@ -48,10 +48,13 @@
 import java.util.Map;
 import java.util.Optional;
 import java.util.Set;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
 
 /** Default validator implementation for {@link FlinkDeployment}. */
 public class DefaultValidator implements FlinkResourceValidator {
-
+    private static final Pattern DEPLOYMENT_NAME_PATTERN =
+            Pattern.compile("[a-z]([-a-z\\d]{0,43}[a-z\\d])?");
     private static final String[] FORBIDDEN_CONF_KEYS =
             new String[] {
                 KubernetesConfigOptions.NAMESPACE.key(), KubernetesConfigOptions.CLUSTER_ID.key()
@@ -77,6 +80,7 @@ public Optional<String> validateDeployment(FlinkDeployment deployment) {
             effectiveConfig.putAll(spec.getFlinkConfiguration());
         }
         return firstPresent(
+                validateDeploymentName(deployment.getMetadata().getName()),
                 validateFlinkVersion(spec.getFlinkVersion()),
                 validateFlinkDeploymentConfig(effectiveConfig),
                 validateIngress(
@@ -100,6 +104,17 @@ private static Optional<String> firstPresent(Optional<String>... errOpts) {
         return Optional.empty();
     }
 
+    private Optional<String> validateDeploymentName(String name) {
+        Matcher matcher = DEPLOYMENT_NAME_PATTERN.matcher(name);
+        if (!matcher.matches()) {
+            return Optional.of(
+                    String.format(
+                            "The FlinkDeployment name: %s is invalid, must consist of lower case alphanumeric characters or '-', start with an alphabetic character, and end with an alphanumeric character (e.g. 'my-name',  or 'abc-123'), and the length must be no more than 45 characters.",
+                            name));
+        }
+        return Optional.empty();
+    }
+
     private Optional<String> validateFlinkVersion(FlinkVersion version) {
         if (version == null) {
             return Optional.of("Flink Version must be defined.");

flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/validation/DefaultValidatorTest.java

@@ -68,6 +68,12 @@ public class DefaultValidatorTest {
     public void testValidationWithoutDefaultConfig() {
         testSuccess(dep -> {});
 
+        // Test meta.name
+        testSuccess(dep -> dep.getMetadata().setName("session-cluster"));
+        testError(
+                dep -> dep.getMetadata().setName("session-cluster-1.13"),
+                "The FlinkDeployment name: session-cluster-1.13 is invalid, must consist of lower case alphanumeric characters or '-', start with an alphabetic character, and end with an alphanumeric character (e.g. 'my-name',  or 'abc-123'), and the length must be no more than 45 characters.");
+
         // Test job validation
         testError(dep -> dep.getSpec().getJob().setJarURI(null), "Jar URI must be defined");
         testError(

flink-kubernetes-webhook/src/test/java/org/apache/flink/kubernetes/operator/admission/AdmissionHandlerTest.java

@@ -37,6 +37,7 @@
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.fasterxml.jackson.databind.exc.MismatchedInputException;
 import io.fabric8.kubernetes.api.model.GroupVersionKind;
+import io.fabric8.kubernetes.api.model.ObjectMeta;
 import io.fabric8.kubernetes.api.model.admission.v1.AdmissionRequest;
 import io.fabric8.kubernetes.api.model.admission.v1.AdmissionReview;
 import org.junit.jupiter.api.Assertions;
@@ -99,6 +100,9 @@ public void testHandleValidateRequestWithoutContent() {
     public void testHandleValidateRequestWithAdmissionReview() throws IOException {
         final EmbeddedChannel embeddedChannel = new EmbeddedChannel(admissionHandler);
         final FlinkDeployment flinkDeployment = new FlinkDeployment();
+        ObjectMeta objectMeta = new ObjectMeta();
+        objectMeta.setName("basic-session-cluster");
+        flinkDeployment.setMetadata(objectMeta);
         flinkDeployment.setSpec(new FlinkDeploymentSpec());
         final AdmissionRequest admissionRequest = new AdmissionRequest();
         admissionRequest.setOperation(CREATE.name());