Broker Ingress Tracing

cmd/broker/ingress/main.go

@@ -24,27 +24,36 @@ import (
 	"log"
 	"net/http"
 	"net/url"
-	"os"
 	"reflect"
 	"sync"
 	"time"
 
+	// Uncomment the following line to load the gcp plugin (only required to authenticate against GKE clusters).
+	// _ "k8s.io/client-go/plugin/pkg/client/auth/gcp"
+
 	cloudevents "github.com/cloudevents/sdk-go"
+	cehttp "github.com/cloudevents/sdk-go/pkg/cloudevents/transport/http"
 	eventingv1alpha1 "github.com/knative/eventing/pkg/apis/eventing/v1alpha1"
 	"github.com/knative/eventing/pkg/broker"
 	"github.com/knative/eventing/pkg/provisioners"
+	"github.com/knative/eventing/pkg/tracing"
 	"github.com/knative/eventing/pkg/utils"
+	"github.com/knative/pkg/configmap"
 	"github.com/knative/pkg/signals"
+	pkgtracing "github.com/knative/pkg/tracing"
 	"go.opencensus.io/exporter/prometheus"
 	"go.opencensus.io/stats"
 	"go.opencensus.io/stats/view"
 	"go.opencensus.io/tag"
 	"go.uber.org/zap"
+	"k8s.io/client-go/kubernetes"
 	"sigs.k8s.io/controller-runtime/pkg/client/config"
 	"sigs.k8s.io/controller-runtime/pkg/manager"
 	crlog "sigs.k8s.io/controller-runtime/pkg/runtime/log"
-	// Uncomment the following line to load the gcp plugin (only required to authenticate against GKE clusters).
-	// _ "k8s.io/client-go/plugin/pkg/client/auth/gcp"
+)
+
+const (
+	NAMESPACE = "NAMESPACE"
 )
 
 var (
@@ -76,15 +85,26 @@ func main() {
 		logger.Fatal("Unable to add eventingv1alpha1 scheme", zap.Error(err))
 	}
 
-	brokerName := getRequiredEnv("BROKER")
+	brokerName := utils.GetRequiredEnvOrFatal("BROKER")
 
 	channelURI := &url.URL{
 		Scheme: "http",
-		Host:   getRequiredEnv("CHANNEL"),
+		Host:   utils.GetRequiredEnvOrFatal("CHANNEL"),
 		Path:   "/",
 	}
 
-	ceClient, err := cloudevents.NewDefaultClient()
+	kc := kubernetes.NewForConfigOrDie(mgr.GetConfig())
+	configMapWatcher := configmap.NewInformedWatcher(kc, utils.GetRequiredEnvOrFatal(NAMESPACE))
+
+	if err = tracing.SetupDynamicZipkinPublishing(logger.Sugar(), configMapWatcher, utils.GetRequiredEnvOrFatal("ZIPKIN_SERVICE_NAME")); err != nil {
+		logger.Fatal("Error setting up Zipkin publishing", zap.Error(err))
+	}
+
+	httpTransport, err := cloudevents.NewHTTPTransport(cloudevents.WithBinaryEncoding(), cehttp.WithMiddleware(pkgtracing.HTTPSpanMiddleware))
+	if err != nil {
+		logger.Fatal("Unable to create CE transport", zap.Error(err))
+	}
+	ceClient, err := cloudevents.NewClient(httpTransport, cloudevents.WithTimeNow(), cloudevents.WithUUIDs())
 	if err != nil {
 		logger.Fatal("Unable to create CE client", zap.Error(err))
 	}
@@ -127,6 +147,12 @@ func main() {
 
 	// Set up signals so we handle the first shutdown signal gracefully.
 	stopCh := signals.SetupSignalHandler()
+
+	// configMapWatcher does not block, so start it first.
+	if err = configMapWatcher.Start(stopCh); err != nil {
+		logger.Fatal("Failed to start ConfigMap watcher", zap.Error(err))
+	}
+
 	// Start blocks forever.
 	if err = mgr.Start(stopCh); err != nil {
 		logger.Error("manager.Start() returned an error", zap.Error(err))
@@ -146,14 +172,6 @@ func main() {
 	logger.Info("Done.")
 }
 
-func getRequiredEnv(envKey string) string {
-	val, defined := os.LookupEnv(envKey)
-	if !defined {
-		log.Fatalf("required environment variable not defined '%s'", envKey)
-	}
-	return val
-}
-
 type handler struct {
 	logger     *zap.Logger
 	ceClient   cloudevents.Client

config/200-broker-clusterrole.yaml

@@ -33,4 +33,12 @@ apiVersion: rbac.authorization.k8s.io/v1
 kind: ClusterRole
 metadata:
   name: eventing-broker-ingress
-rules: []
+rules:
+  - apiGroups:
+      - ""
+    resources:
+      - "configmaps"
+    verbs:
+      - "get"
+      - "list"
+      - "watch"

pkg/configmap/store.go

@@ -0,0 +1,43 @@
+package configmap
+
+import (
+	"github.com/knative/pkg/configmap"
+	v1 "k8s.io/api/core/v1"
+)
+
+// TODO Move this to knative/pkg.
+
+// DefaultConstructors are something.
+type DefaultConstructors map[*v1.ConfigMap]interface{}
+
+// DefaultUntypedStore is an UntypedStore with default values for ConfigMaps that do not exist.
+type DefaultUntypedStore struct {
+	store      *configmap.UntypedStore
+	defaultCMs []v1.ConfigMap
+}
+
+// NewDefaultUntypedStore creates a new DefaultUntypedStore.
+func NewDefaultUntypedStore(
+	name string,
+	logger configmap.Logger,
+	defaultConstructors DefaultConstructors,
+	onAfterStore ...func(name string, value interface{})) *DefaultUntypedStore {
+	constructors := configmap.Constructors{}
+	defaultCMs := make([]v1.ConfigMap, 0, len(defaultConstructors))
+	for cm, ctor := range defaultConstructors {
+		constructors[cm.Name] = ctor
+		defaultCMs = append(defaultCMs, *cm)
+	}
+	return &DefaultUntypedStore{
+		store:      configmap.NewUntypedStore(name, logger, constructors, onAfterStore...),
+		defaultCMs: defaultCMs,
+	}
+}
+
+// WatchConfigs uses the provided configmap.DefaultingWatcher to setup watches for the config maps
+// provided in defaultCMs.
+func (s *DefaultUntypedStore) WatchConfigs(w configmap.DefaultingWatcher) {
+	for _, cm := range s.defaultCMs {
+		w.WatchWithDefault(cm, s.store.OnConfigChanged)
+	}
+}

pkg/reconciler/broker/broker_test.go

@@ -930,6 +930,14 @@ func envVars(containerName string) []corev1.EnvVar {
 		}
 	case ingressContainerName:
 		return []corev1.EnvVar{
+			{
+				Name: "NAMESPACE",
+				ValueFrom: &corev1.EnvVarSource{
+					FieldRef: &corev1.ObjectFieldSelector{
+						FieldPath: "metadata.namespace",
+					},
+				},
+			},
 			{
 				Name:  "FILTER",
 				Value: "",
@@ -942,6 +950,10 @@ func envVars(containerName string) []corev1.EnvVar {
 				Name:  "BROKER",
 				Value: brokerName,
 			},
+			{
+				Name:  "ZIPKIN_SERVICE_NAME",
+				Value: fmt.Sprintf("%s-broker-ingress.%s", brokerName, testNS),
+			},
 		}
 	}
 	return []corev1.EnvVar{}

pkg/reconciler/broker/resources/ingress.go

@@ -63,6 +63,14 @@ func MakeIngress(args *IngressArgs) *appsv1.Deployment {
 							Image: args.Image,
 							Name:  "ingress",
 							Env: []corev1.EnvVar{
+								{
+									Name: "NAMESPACE",
+									ValueFrom: &corev1.EnvVarSource{
+										FieldRef: &corev1.ObjectFieldSelector{
+											FieldPath: "metadata.namespace",
+										},
+									},
+								},
 								{
 									Name:  "FILTER",
 									Value: "", // TODO Add one.
@@ -75,6 +83,10 @@ func MakeIngress(args *IngressArgs) *appsv1.Deployment {
 									Name:  "BROKER",
 									Value: args.Broker.Name,
 								},
+								{
+									Name:  "ZIPKIN_SERVICE_NAME",
+									Value: fmt.Sprintf("%s-broker-ingress.%s", args.Broker.Name, args.Broker.Namespace),
+								},
 							},
 							Ports: []corev1.ContainerPort{
 								{

pkg/tracing/setup.go

@@ -19,11 +19,14 @@ package tracing
 import (
 	"fmt"
 
+	eventingconfigmap "github.com/knative/eventing/pkg/configmap"
 	"github.com/knative/pkg/configmap"
 	"github.com/knative/pkg/tracing"
 	tracingconfig "github.com/knative/pkg/tracing/config"
-	zipkin "github.com/openzipkin/zipkin-go"
+	"github.com/openzipkin/zipkin-go"
 	"go.uber.org/zap"
+	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 )
 
 // TODO Move this to knative/pkg.
@@ -45,6 +48,18 @@ var (
 		SampleRate:     0,
 		ZipkinEndpoint: "http://zipkin.istio-system.svc.cluster.local:9411/api/v2/spans",
 	}
+
+	enableZeroSamplingCM = &corev1.ConfigMap{
+		ObjectMeta: metav1.ObjectMeta{
+			Name: tracingconfig.ConfigName,
+		},
+		Data: map[string]string{
+			"enable":          "True",
+			"debug":           "False",
+			"sample-rate":     "0",
+			"zipkin-endpoint": "http://zipkin.istio-system.svc.cluster.local:9411/api/v2/spans",
+		},
+	}
 )
 
 // setupZipkinPublishing sets up Zipkin trace publishing for the process. Note that other pieces
@@ -82,7 +97,7 @@ func SetupStaticZipkinPublishing(serviceName string, cfg *tracingconfig.Config)
 // just ensures that if generated, they are collected appropriately. This is normally done by using
 // tracing.HTTPSpanMiddleware as a middleware HTTP handler. The configuration will be dynamically
 // updated when the ConfigMap is updated.
-func SetupDynamicZipkinPublishing(logger *zap.SugaredLogger, configMapWatcher configmap.Watcher, serviceName string) error {
+func SetupDynamicZipkinPublishing(logger *zap.SugaredLogger, configMapWatcher configmap.DefaultingWatcher, serviceName string) error {
 	oct, err := setupZipkinPublishing(serviceName)
 	if err != nil {
 		return err
@@ -101,11 +116,11 @@ func SetupDynamicZipkinPublishing(logger *zap.SugaredLogger, configMapWatcher co
 	}
 
 	// Set up our config store.
-	configStore := configmap.NewUntypedStore(
+	configStore := eventingconfigmap.NewDefaultUntypedStore(
 		"tracing-config",
 		logger,
-		configmap.Constructors{
-			tracingconfig.ConfigName: tracingconfig.NewTracingConfigFromConfigMap,
+		eventingconfigmap.DefaultConstructors{
+			enableZeroSamplingCM: tracingconfig.NewTracingConfigFromConfigMap,
 		},
 		tracerUpdater)
 	configStore.WatchConfigs(configMapWatcher)

pkg/utils/env.go

@@ -0,0 +1,29 @@
+/*
+Copyright 2019 The Knative Authors
+ Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+     http://www.apache.org/licenses/LICENSE-2.0
+ Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package utils
+
+import (
+	"log"
+	"os"
+)
+
+// GetRequiredEnvOrFatal gets the environment variable's value. If the environment variable is not
+// defined, then log.Fatal will be called, exiting the program.
+func GetRequiredEnvOrFatal(envKey string) string {
+	val, defined := os.LookupEnv(envKey)
+	if !defined {
+		log.Fatalf("required environment variable not defined '%s'", envKey)
+	}
+	return val
+}