Implemented a second service for the collector

Signed-off-by: Juraci Paixão Kröhling <juraci@kroehling.de>

deploy/examples/business-application-injected-sidecar.yaml

@@ -16,5 +16,3 @@ spec:
       containers:
       - name: myapp
         image: jaegertracing/vertx-create-span:operator-e2e-tests
-
-

pkg/deployment/agent.go

@@ -44,7 +44,7 @@ func (a *Agent) Get() *appsv1.DaemonSet {
 
 		// we only add the grpc host if we are adding the reporter type and there's no explicit value yet
 		if len(util.FindItem("--reporter.grpc.host-port=", args)) == 0 {
-			args = append(args, fmt.Sprintf("--reporter.grpc.host-port=dns:///%s.%s:14250", service.GetNameForCollectorService(a.jaeger), a.jaeger.Namespace))
+			args = append(args, fmt.Sprintf("--reporter.grpc.host-port=dns:///%s.%s:14250", service.GetNameForHeadlessCollectorService(a.jaeger), a.jaeger.Namespace))
 		}
 	}
 

pkg/deployment/all-in-one.go

@@ -175,11 +175,10 @@ func (a *AllInOne) Get() *appsv1.Deployment {
 // Services returns a list of services to be deployed along with the all-in-one deployment
 func (a *AllInOne) Services() []*corev1.Service {
 	labels := a.labels()
-	return []*corev1.Service{
-		service.NewCollectorService(a.jaeger, labels),
+	return append(service.NewCollectorServices(a.jaeger, labels),
 		service.NewQueryService(a.jaeger, labels),
 		service.NewAgentService(a.jaeger, labels),
-	}
+	)
 }
 
 func (a *AllInOne) labels() map[string]string {

pkg/deployment/all-in-one_test.go

@@ -69,7 +69,7 @@ func TestAllInOneHasOwner(t *testing.T) {
 func TestAllInOneNumberOfServices(t *testing.T) {
 	name := "TestNumberOfServices"
 	services := NewAllInOne(v1.NewJaeger(name)).Services()
-	assert.Len(t, services, 3) // collector, query, agent
+	assert.Len(t, services, 4) // collector (headless and cluster IP), query, agent
 
 	for _, svc := range services {
 		owners := svc.ObjectMeta.OwnerReferences

pkg/deployment/collector.go

@@ -169,9 +169,7 @@ func (c *Collector) Get() *appsv1.Deployment {
 
 // Services returns a list of services to be deployed along with the all-in-one deployment
 func (c *Collector) Services() []*corev1.Service {
-	return []*corev1.Service{
-		service.NewCollectorService(c.jaeger, c.labels()),
-	}
+	return service.NewCollectorServices(c.jaeger, c.labels())
 }
 
 func (c *Collector) labels() map[string]string {

pkg/deployment/collector_test.go

@@ -85,7 +85,7 @@ func TestName(t *testing.T) {
 func TestCollectorServices(t *testing.T) {
 	collector := NewCollector(v1.NewJaeger("TestName"))
 	svcs := collector.Services()
-	assert.Len(t, svcs, 1)
+	assert.Len(t, svcs, 2) // headless and cluster IP
 }
 
 func TestDefaultCollectorImage(t *testing.T) {

pkg/inject/sidecar.go

@@ -95,7 +95,7 @@ func container(jaeger *v1.Jaeger) corev1.Container {
 
 		// we only add the grpc host if we are adding the reporter type and there's no explicit value yet
 		if len(util.FindItem("--reporter.grpc.host-port=", args)) == 0 {
-			args = append(args, fmt.Sprintf("--reporter.grpc.host-port=dns:///%s.%s:14250", service.GetNameForCollectorService(jaeger), jaeger.Namespace))
+			args = append(args, fmt.Sprintf("--reporter.grpc.host-port=dns:///%s.%s:14250", service.GetNameForHeadlessCollectorService(jaeger), jaeger.Namespace))
 		}
 	}
 

pkg/inventory/service.go

@@ -21,6 +21,11 @@ func ForServices(existing []v1.Service, desired []v1.Service) Service {
 		if t, ok := mdelete[k]; ok {
 			tp := t.DeepCopy()
 
+			// we keep the ClusterIP that got assigned by the cluster, if it's empty in the "desired" and not empty on the "current"
+			if v.Spec.ClusterIP == "" && len(tp.Spec.ClusterIP) > 0 {
+				v.Spec.ClusterIP = tp.Spec.ClusterIP
+			}
+
 			// we can't blindly DeepCopyInto, so, we select what we bring from the new to the old object
 			tp.Spec = v.Spec
 			tp.ObjectMeta.OwnerReferences = v.ObjectMeta.OwnerReferences

pkg/service/collector.go

@@ -9,10 +9,27 @@ import (
 	"github.com/jaegertracing/jaeger-operator/pkg/apis/jaegertracing/v1"
 )
 
-// NewCollectorService returns a new Kubernetes service for Jaeger Collector backed by the pods matching the selector
-func NewCollectorService(jaeger *v1.Jaeger, selector map[string]string) *corev1.Service {
-	trueVar := true
+// NewCollectorServices returns a new Kubernetes service for Jaeger Collector backed by the pods matching the selector
+func NewCollectorServices(jaeger *v1.Jaeger, selector map[string]string) []*corev1.Service {
+	return []*corev1.Service{
+		headlessCollectorService(jaeger, selector),
+		clusteripCollectorService(jaeger, selector),
+	}
+}
 
+func headlessCollectorService(jaeger *v1.Jaeger, selector map[string]string) *corev1.Service {
+	svc := collectorService(jaeger, selector)
+	svc.Name = GetNameForHeadlessCollectorService(jaeger)
+	svc.Spec.ClusterIP = "None"
+	return svc
+}
+
+func clusteripCollectorService(jaeger *v1.Jaeger, selector map[string]string) *corev1.Service {
+	return collectorService(jaeger, selector)
+}
+
+func collectorService(jaeger *v1.Jaeger, selector map[string]string) *corev1.Service {
+	trueVar := true
 	return &corev1.Service{
 		TypeMeta: metav1.TypeMeta{
 			Kind:       "Service",
@@ -41,7 +58,7 @@ func NewCollectorService(jaeger *v1.Jaeger, selector map[string]string) *corev1.
 		},
 		Spec: corev1.ServiceSpec{
 			Selector:  selector,
-			ClusterIP: "None",
+			ClusterIP: "",
 			Ports: []corev1.ServicePort{
 				{
 					Name: "zipkin",
@@ -62,9 +79,15 @@ func NewCollectorService(jaeger *v1.Jaeger, selector map[string]string) *corev1.
 			},
 		},
 	}
+
 }
 
 // GetNameForCollectorService returns the service name for the collector in this Jaeger instance
 func GetNameForCollectorService(jaeger *v1.Jaeger) string {
 	return fmt.Sprintf("%s-collector", jaeger.Name)
 }
+
+// GetNameForHeadlessCollectorService returns the headless service name for the collector in this Jaeger instance
+func GetNameForHeadlessCollectorService(jaeger *v1.Jaeger) string {
+	return fmt.Sprintf("%s-collector-headless", jaeger.Name)
+}

pkg/service/collector_test.go

@@ -14,8 +14,10 @@ func TestCollectorServiceNameAndPorts(t *testing.T) {
 	selector := map[string]string{"app": "myapp", "jaeger": name, "jaeger-component": "collector"}
 
 	jaeger := v1.NewJaeger(name)
-	svc := NewCollectorService(jaeger, selector)
-	assert.Equal(t, svc.ObjectMeta.Name, fmt.Sprintf("%s-collector", name))
+	svcs := NewCollectorServices(jaeger, selector)
+
+	assert.Equal(t, svcs[0].Name, fmt.Sprintf("%s-collector-headless", name))
+	assert.Equal(t, svcs[1].Name, fmt.Sprintf("%s-collector", name))
 
 	ports := map[int32]bool{
 		9411:  false,
@@ -24,6 +26,7 @@ func TestCollectorServiceNameAndPorts(t *testing.T) {
 		14268: false,
 	}
 
+	svc := svcs[0]
 	for _, port := range svc.Spec.Ports {
 		ports[port.Port] = true
 	}
@@ -32,4 +35,21 @@ func TestCollectorServiceNameAndPorts(t *testing.T) {
 		assert.Equal(t, v, true, "Expected port %v to be specified, but wasn't", k)
 	}
 
+	// we ensure the ports are the same for both services
+	assert.Equal(t, svcs[0].Spec.Ports, svcs[1].Spec.Ports)
+}
+
+func TestCollectorServiceWithClusterIPEmptyAndNone(t *testing.T) {
+	name := "TestCollectorServiceWithClusterIP"
+	selector := map[string]string{"app": "myapp", "jaeger": name, "jaeger-component": "collector"}
+
+	jaeger := v1.NewJaeger(name)
+	svcs := NewCollectorServices(jaeger, selector)
+
+	// we want two services, one headless (load balanced by the client, possibly via DNS)
+	// and one with a cluster IP (load balanced by kube-proxy)
+	assert.Len(t, svcs, 2)
+	assert.NotEqual(t, svcs[0].Name, svcs[1].Name) // they can't have the same name
+	assert.Equal(t, "None", svcs[0].Spec.ClusterIP)
+	assert.Len(t, svcs[1].Spec.ClusterIP, 0)
 }

pkg/service/query.go

@@ -48,7 +48,7 @@ func NewQueryService(jaeger *v1.Jaeger, selector map[string]string) *corev1.Serv
 		},
 		Spec: corev1.ServiceSpec{
 			Selector:  selector,
-			ClusterIP: "None",
+			ClusterIP: "",
 			Ports: []corev1.ServicePort{
 				{
 					Name:       "query",

pkg/service/query_test.go

@@ -21,6 +21,7 @@ func TestQueryServiceNameAndPorts(t *testing.T) {
 	assert.Len(t, svc.Spec.Ports, 1)
 	assert.Equal(t, int32(16686), svc.Spec.Ports[0].Port)
 	assert.Equal(t, intstr.FromInt(16686), svc.Spec.Ports[0].TargetPort)
+	assert.Len(t, svc.Spec.ClusterIP, 0) // make sure we get a cluster IP
 }
 
 func TestQueryServiceNameAndPortsWithOAuthProxy(t *testing.T) {