Fix hardcoded self provisioned kafka broker URL (#841)

Signed-off-by: Pavol Loffay <[email protected]>
jaegertracing · Jan 8, 2020 · 5999ea6 · 5999ea6
1 parent 6c2a806
commit 5999ea6
Show file tree

Hide file tree

Showing 3 changed files with 11 additions and 4 deletions.
diff --git a/Makefile b/Makefile
@@ -202,8 +202,12 @@ kafka:
 ifeq ($(OLM),true)
 	@echo Skipping kafka-operator deployment, assuming it has been installed via OperatorHub
 else
+	@kubectl create clusterrolebinding strimzi-cluster-operator-namespaced --clusterrole=strimzi-cluster-operator-namespaced --serviceaccount ${KAFKA_NAMESPACE}:strimzi-cluster-operator 2>&1 | grep -v "already exists" || true
+	@kubectl create clusterrolebinding strimzi-cluster-operator-entity-operator-delegation --clusterrole=strimzi-entity-operator --serviceaccount ${KAFKA_NAMESPACE}:strimzi-cluster-operator 2>&1 | grep -v "already exists" || true
+	@kubectl create clusterrolebinding strimzi-cluster-operator-topic-operator-delegation --clusterrole=strimzi-topic-operator --serviceaccount ${KAFKA_NAMESPACE}:strimzi-cluster-operator 2>&1 | grep -v "already exists" || true
 	@curl --location $(KAFKA_YAML) --output deploy/test/kafka-operator.yaml
 	@sed 's/namespace: .*/namespace: $(KAFKA_NAMESPACE)/' deploy/test/kafka-operator.yaml | kubectl -n $(KAFKA_NAMESPACE) apply -f -  2>&1 | grep -v "already exists" || true
+	@kubectl set env deployment strimzi-cluster-operator -n ${KAFKA_NAMESPACE} STRIMZI_NAMESPACE="*"
 endif
 	@curl --location $(KAFKA_EXAMPLE) --output deploy/test/kafka-example.yaml
 	@kubectl -n $(KAFKA_NAMESPACE) apply -f deploy/test/kafka-example.yaml  2>&1 | grep -v "already exists" || true
@@ -215,6 +219,9 @@ ifeq ($(OLM),true)
 	@echo Skiping kafka-operator undeploy
 else
 	@kubectl delete --namespace $(KAFKA_NAMESPACE) -f deploy/test/kafka-operator.yaml 2>&1 || true
+	@kubectl delete clusterrolebinding strimzi-cluster-operator-namespaced
+	@kubectl delete clusterrolebinding strimzi-cluster-operator-entity-operator-delegation
+	@kubectl delete clusterrolebinding strimzi-cluster-operator-topic-operator-delegation
 endif
 	@kubectl delete namespace $(KAFKA_NAMESPACE) 2>&1 || true
 

diff --git a/pkg/strategy/streaming.go b/pkg/strategy/streaming.go
@@ -172,7 +172,7 @@ func autoProvisionKafka(ctx context.Context, jaeger *v1.Jaeger, manifest S) S {
 	}
 	commonSpec.VolumeMounts = append(commonSpec.VolumeMounts, kuVolumeMount, kuCAVolumeMount)
 
-	brokers := fmt.Sprintf("%s-kafka-bootstrap.kafka.svc.cluster.local:9093", k.Name)
+	brokers := fmt.Sprintf("%s-kafka-bootstrap.%s.svc.cluster.local:9093", k.Name, k.Namespace)
 
 	collectorOpts := jaeger.Spec.Collector.Options.GenericMap()
 	ingesterOpts := jaeger.Spec.Ingester.Options.GenericMap()

diff --git a/pkg/strategy/streaming_test.go b/pkg/strategy/streaming_test.go
@@ -239,7 +239,7 @@ func TestAgentSidecarIsInjectedIntoQueryForStreaming(t *testing.T) {
 
 func TestAutoProvisionedKafkaInjectsIntoInstance(t *testing.T) {
 	name := "my-instance"
-	jaeger := v1.NewJaeger(types.NamespacedName{Name: name})
+	jaeger := v1.NewJaeger(types.NamespacedName{Name: name, Namespace: "project"})
 	jaeger.Spec.Collector.Options = v1.NewOptions(map[string]interface{}{})
 	jaeger.Spec.Ingester.Options = v1.NewOptions(map[string]interface{}{})
 	manifest := S{typ: v1.DeploymentStrategyStreaming}
@@ -250,14 +250,14 @@ func TestAutoProvisionedKafkaInjectsIntoInstance(t *testing.T) {
 	// verify
 	assert.Equal(t, v1.AnnotationProvisionedKafkaValue, jaeger.Annotations[v1.AnnotationProvisionedKafkaKey])
 
-	assert.Contains(t, jaeger.Spec.Collector.Options.Map(), "kafka.producer.brokers")
+	assert.Equal(t, "my-instance-kafka-bootstrap.project.svc.cluster.local:9093", jaeger.Spec.Collector.Options.Map()["kafka.producer.brokers"])
 	assert.Contains(t, jaeger.Spec.Collector.Options.Map(), "kafka.producer.authentication")
 	assert.Contains(t, jaeger.Spec.Collector.Options.Map(), "kafka.producer.tls.key")
 	assert.Contains(t, jaeger.Spec.Collector.Options.Map(), "kafka.producer.tls.cert")
 	assert.Contains(t, jaeger.Spec.Collector.Options.Map(), "kafka.producer.tls.ca")
 	assert.NotContains(t, jaeger.Spec.Collector.Options.Map(), "kafka.consumer.brokers")
 
-	assert.Contains(t, jaeger.Spec.Ingester.Options.Map(), "kafka.consumer.brokers")
+	assert.Equal(t, "my-instance-kafka-bootstrap.project.svc.cluster.local:9093", jaeger.Spec.Ingester.Options.Map()["kafka.consumer.brokers"])
 	assert.Contains(t, jaeger.Spec.Ingester.Options.Map(), "kafka.consumer.authentication")
 	assert.Contains(t, jaeger.Spec.Ingester.Options.Map(), "kafka.consumer.tls.key")
 	assert.Contains(t, jaeger.Spec.Ingester.Options.Map(), "kafka.consumer.tls.cert")