added carbon ingester port config to cluster spec

.gitignore

@@ -3,6 +3,7 @@
 
 # Dev artifacts
 .vscode/
+.idea
 
 # Binaries for programs and plugins
 *.so

assets/default-config.tmpl

@@ -18,6 +18,11 @@ coordinator:
     extended: none
   tagOptions:
     idScheme: quoted
+{{- if .CarbonIngester }}
+  carbon:
+    ingester:
+      listenAddress: "0.0.0.0:7204"
+{{- end  }}
 
 db:
   logging:

assets/default-config.yaml

@@ -18,6 +18,9 @@ coordinator:
     extended: none
   tagOptions:
     idScheme: quoted
+  carbon:
+    ingester:
+      listenAddress: "0.0.0.0:7204"
 
 db:
   logging:

docs/api.md

@@ -46,6 +46,7 @@ ClusterSpec defines the desired state for a M3 cluster to be converge to.
 | namespaces | Namespaces specifies the namespaces this cluster will hold. | [][Namespace](#namespace) | false |
 | etcdEndpoints | EtcdEndpoints defines the etcd endpoints to use for service discovery. Must be set if no custom configmap is defined. If set, etcd endpoints will be templated in to the default configmap template. | []string | false |
 | keepEtcdDataOnDelete | KeepEtcdDataOnDelete determines whether the operator will remove cluster metadata (placement + namespaces) in etcd when the cluster is deleted. Unless true, etcd data will be cleared when the cluster is deleted. | bool | false |
+| enableCarbonIngester | EnableCarbonIngester enables the listener port for the carbon ingester | bool | false |
 | configMapName | ConfigMapName specifies the ConfigMap to use for this cluster. If unset a default configmap with template variables for etcd endpoints will be used. See \"Configuring M3DB\" in the docs for more. | *string | false |
 | podIdentityConfig | PodIdentityConfig sets the configuration for pod identity. If unset only pod name and UID will be used. | *PodIdentityConfig | false |
 | containerResources | Resources defines memory / cpu constraints for each container in the cluster. | [corev1.ResourceRequirements](https://kubernetes.io/docs/reference/generated/kubernetes-api/v1.10/#resourcerequirements-v1-core) | false |

pkg/apis/m3dboperator/v1alpha1/cluster.go

@@ -200,6 +200,10 @@ type ClusterSpec struct {
 	// +optional
 	KeepEtcdDataOnDelete bool `json:"keepEtcdDataOnDelete,omitempty"`
 
+	// EnableCarbonIngester enables the listener port for the carbon ingester
+	// +optional
+	EnableCarbonIngester bool `json:"enableCarbonIngester,omitempty"`
+
 	// ConfigMapName specifies the ConfigMap to use for this cluster. If unset a
 	// default configmap with template variables for etcd endpoints will be used.
 	// See "Configuring M3DB" in the docs for more.

pkg/k8sops/config_map.go

@@ -46,6 +46,7 @@ var (
 type configData struct {
 	Env       string
 	Endpoints []string
+	CarbonIngester bool
 }
 
 // GenerateDefaultConfigMap creates a ConfigMap for the clusters with the
@@ -77,10 +78,13 @@ func GenerateDefaultConfigMap(cluster *myspec.M3DBCluster) (*corev1.ConfigMap, e
 	config := &configData{
 		Env:       DefaultM3ClusterEnvironmentName(cluster),
 		Endpoints: cluster.Spec.EtcdEndpoints,
+		CarbonIngester: cluster.Spec.EnableCarbonIngester,
 	}
 
 	buf := &bytes.Buffer{}
-	tmpl.Execute(buf, &config)
+	if err:= tmpl.Execute(buf, &config); err != nil {
+		return nil, err
+	}
 
 	ownerRef := GenerateOwnerRef(cluster)
 

pkg/k8sops/config_map_test.go

@@ -75,6 +75,9 @@ func TestGenerateDefaultConfigMap(t *testing.T) {
 	assert.Contains(t, data, `env: "foo/m3db-cluster"`)
 	assert.Contains(t, data, `- "ep0"`)
 	assert.Contains(t, data, `- "ep1"`)
+	assert.Contains(t, data, `carbon:`)
+	assert.Contains(t, data, `ingester:`)
+	assert.Contains(t, data, `listenAddress: "0.0.0.0:7204"`)
 }
 
 func TestGenerateDefaultConfigMap_Err(t *testing.T) {

pkg/k8sops/fixtures/testM3DBCluster.yaml

@@ -8,6 +8,7 @@ spec:
   image: fake.fake/fake/m3dbnode:latest
   replicationFactor: 3
   numberOfShards: 8
+  enableCarbonIngester: true
   isolationGroups:
     - name: us-fake1-a
       numInstances: 1

pkg/k8sops/generators.go

@@ -82,6 +82,8 @@ var baseCoordinatorPorts = [...]m3dbPort{
 	{"coord-metrics", PortM3CoordinatorMetrics, v1.ProtocolTCP},
 }
 
+var carbonListenerPort = m3dbPort{"coord-carbon", PortM3CoordinatorCarbon, v1.ProtocolTCP}
+
 // GenerateCRD generates the crd object needed for the M3DBCluster
 func GenerateCRD(enableValidation bool) *apiextensionsv1beta1.CustomResourceDefinition {
 	crd := &apiextensionsv1beta1.CustomResourceDefinition{
@@ -149,7 +151,7 @@ func GenerateStatefulSet(
 	statefulSet := NewBaseStatefulSet(ssName, isolationGroupName, cluster, instanceAmount)
 	m3dbContainer := &statefulSet.Spec.Template.Spec.Containers[0]
 	m3dbContainer.Resources = clusterSpec.ContainerResources
-	m3dbContainer.Ports = generateContainerPorts()
+	m3dbContainer.Ports = generateContainerPorts(cluster)
 	statefulSet.Spec.Template.Spec.Affinity = affinity
 	statefulSet.Spec.Template.Spec.Tolerations = cluster.Spec.Tolerations
 
@@ -205,7 +207,7 @@ func GenerateM3DBService(cluster *myspec.M3DBCluster) (*v1.Service, error) {
 		},
 		Spec: v1.ServiceSpec{
 			Selector:  svcLabels,
-			Ports:     generateM3DBServicePorts(),
+			Ports:     generateM3DBServicePorts(cluster),
 			ClusterIP: v1.ClusterIPNone,
 			Type:      v1.ServiceTypeClusterIP,
 		},
@@ -232,10 +234,11 @@ func GenerateCoordinatorService(cluster *myspec.M3DBCluster) (*v1.Service, error
 		},
 		Spec: v1.ServiceSpec{
 			Selector: selectorLabels,
-			Ports:    generateCoordinatorServicePorts(),
+			Ports:    generateCoordinatorServicePorts(cluster),
 			Type:     v1.ServiceTypeClusterIP,
 		},
 	}, nil
+
 }
 
 func buildServicePorts(ports []m3dbPort) []v1.ServicePort {
@@ -251,18 +254,30 @@ func buildServicePorts(ports []m3dbPort) []v1.ServicePort {
 	return svcPorts
 }
 
-func generateM3DBServicePorts() []v1.ServicePort {
-	return buildServicePorts(baseM3DBPorts[:])
+func generateM3DBServicePorts(cluster *myspec.M3DBCluster) []v1.ServicePort {
+	ports := baseM3DBPorts[:]
+	if cluster.Spec.EnableCarbonIngester {
+		ports = append(ports, carbonListenerPort)
+	}
+	return buildServicePorts(ports)
 }
 
-func generateCoordinatorServicePorts() []v1.ServicePort {
-	return buildServicePorts(baseCoordinatorPorts[:])
+func generateCoordinatorServicePorts(cluster *myspec.M3DBCluster) []v1.ServicePort {
+	ports := baseCoordinatorPorts[:]
+	if cluster.Spec.EnableCarbonIngester {
+		ports = append(ports, carbonListenerPort)
+	}
+	return buildServicePorts(ports)
 }
 
 // generateContainerPorts will produce default container ports.
-func generateContainerPorts() []v1.ContainerPort {
+func generateContainerPorts(cluster *myspec.M3DBCluster) []v1.ContainerPort {
 	cntPorts := []v1.ContainerPort{}
-	for _, v := range baseM3DBPorts {
+	basePorts := baseM3DBPorts[:]
+	if cluster.Spec.EnableCarbonIngester {
+		basePorts = append(basePorts, carbonListenerPort)
+	}
+	for _, v := range basePorts {
 		newPortMapping := v1.ContainerPort{
 			Name:          v.name,
 			ContainerPort: int32(v.port),

pkg/k8sops/generators_test.go

@@ -190,7 +190,7 @@ func TestGenerateStatefulSet(t *testing.T) {
 									},
 								},
 							},
-							Ports: generateContainerPorts(),
+							Ports: generateContainerPorts(fixture),
 							VolumeMounts: []v1.VolumeMount{
 								{
 									Name:      _dataVolumeName,
@@ -380,7 +380,7 @@ func TestGenerateM3DBService(t *testing.T) {
 		},
 		Spec: v1.ServiceSpec{
 			Selector:  baseLabels,
-			Ports:     generateM3DBServicePorts(),
+			Ports:     generateM3DBServicePorts(cluster),
 			ClusterIP: v1.ClusterIPNone,
 			Type:      v1.ServiceTypeClusterIP,
 		},
@@ -419,7 +419,7 @@ func TestGenerateCoordinatorService(t *testing.T) {
 		},
 		Spec: v1.ServiceSpec{
 			Selector: selectLabels,
-			Ports:    generateCoordinatorServicePorts(),
+			Ports:    generateCoordinatorServicePorts(cluster),
 			Type:     v1.ServiceTypeClusterIP,
 		},
 	}

pkg/k8sops/ports.go

@@ -33,4 +33,5 @@ const (
 
 	PortM3Coordinator        = 7201
 	PortM3CoordinatorMetrics = 7203
+	PortM3CoordinatorCarbon  = 7204
 )