Add Headless AccessScope to jobmanager service spec (#280)

This allows you to have the operator create a headless service
(https://kubernetes.io/docs/concepts/services-networking/service/#headless-services)
for the jobmanager. This is useful when e.g. your pod network is directly routable
and you'd like to use SRV or A record round-robin to discover the underlying pods.

Author: bnu0

api/v1beta1/flinkcluster_types.go

@@ -58,6 +58,7 @@ const (
 	AccessScopeVPC      = "VPC"
 	AccessScopeExternal = "External"
 	AccessScopeNodePort = "NodePort"
+	AccessScopeHeadless = "Headless"
 )
 
 // JobRestartPolicy defines the restart policy when a job fails.

api/v1beta1/flinkcluster_validate.go

@@ -296,6 +296,7 @@ func (v *Validator) validateJobManager(jmSpec *JobManagerSpec) error {
 	case AccessScopeVPC:
 	case AccessScopeExternal:
 	case AccessScopeNodePort:
+	case AccessScopeHeadless:
 	default:
 		return fmt.Errorf("invalid JobManager access scope: %v", jmSpec.AccessScope)
 	}

controllers/flinkcluster_converter.go

@@ -280,6 +280,11 @@ func getDesiredJobManagerService(
 		jobManagerService.Spec.Type = corev1.ServiceTypeLoadBalancer
 	case v1beta1.AccessScopeNodePort:
 		jobManagerService.Spec.Type = corev1.ServiceTypeNodePort
+	case v1beta1.AccessScopeHeadless:
+		// Headless services do not allocate any sort of VIP or LoadBalancer, and merely
+		// collect a set of Pod IPs that are assumed to be independently routable:
+		jobManagerService.Spec.Type = corev1.ServiceTypeClusterIP
+		jobManagerService.Spec.ClusterIP = "None"
 	default:
 		panic(fmt.Sprintf(
 			"Unknown service access cope: %v", jobManagerSpec.AccessScope))

docs/crd.md

@@ -144,9 +144,10 @@ FlinkCluster
     * **batchSchedulerName** (optional): BatchSchedulerName specifies the batch scheduler name for JobManager, TaskManager.
       If empty, no batch scheduling is enabled.
     * **jobManager** (required): JobManager spec.
-      * **accessScope** (optional): Access scope of the JobManager service. `enum("Cluster", "VPC", "External",
-      "NodePort")`.`Cluster`: accessible from within the same cluster; `VPC`: accessible from within the same VPC;
-      `External`:accessible from the internet. `NodePort`: accessible through node port.
+      * **accessScope** (optional): Access scope of the JobManager service. `enum("Cluster", "VPC", "External", 
+      "NodePort", "Headless")`. `Cluster`: accessible from within the same cluster; `VPC`: accessible from within the same VPC; 
+      `External`: accessible from the internet. `NodePort`: accessible through node port; `Headless`: pod IPs assumed to 
+      be routable and advertised directly with `clusterIP: None`.  
       Currently `VPC` and `External` are only available for GKE.
       * **ports** (optional): Ports that JobManager listening on.
         * **rpc** (optional): RPC port, default: 6123.