Repository: knative-sandbox/eventing-kafka (go)

Issue:
Support Eventing Tracing & Observability Configuration
In order to improve our alignment with the core Knative Eventing implementations we should support the `config-tracing` and `config-observability` global ConfigMaps in the `knative-eventing` namespace.  This is similar to the work we did to align on the `config-logging` implementation.

Pull request:
Add InitializeTracing function and call it from controller and dispatcher main
Fixes #22

In order to have tracing (via the config-tracing configmap) work properly, a call to SetupDynamicPublishing() needs to be made.  This adds those calls to the main functions of the controller and dispatcher.

Patch:
diff --git a/cmd/channel/main.go b/cmd/channel/main.go
index 1dfca67d0..c5b3c5d4f 100644
--- a/cmd/channel/main.go
+++ b/cmd/channel/main.go
@@ -45,6 +45,9 @@ func main() {
 		logger.Fatal("Invalid / Missing Environment Variables - Terminating", zap.Error(err))
 	}
 
+	// Initialize Tracing (Watching config-tracing ConfigMap, Assumes Context Came From LoggingContext With Embedded K8S Client Key)
+	commonk8s.InitializeTracing(logger.Sugar(), ctx, environment.ServiceName)
+
 	// Start The Liveness And Readiness Servers
 	healthServer := channelhealth.NewChannelHealthServer(environment.HealthPort)
 	healthServer.Start(logger)
diff --git a/cmd/dispatcher/main.go b/cmd/dispatcher/main.go
index 7bd40119d..f76ec6b54 100644
--- a/cmd/dispatcher/main.go
+++ b/cmd/dispatcher/main.go
@@ -8,6 +8,7 @@ import (
 	"k8s.io/client-go/tools/clientcmd"
 	"knative.dev/eventing-contrib/kafka/channel/pkg/client/clientset/versioned"
 	"knative.dev/eventing-contrib/kafka/channel/pkg/client/informers/externalversions"
+	commonenv "knative.dev/eventing-kafka/pkg/common/env"
 	commonk8s "knative.dev/eventing-kafka/pkg/common/k8s"
 	"knative.dev/eventing-kafka/pkg/common/prometheus"
 	"knative.dev/eventing-kafka/pkg/dispatcher/controller"
@@ -51,20 +52,21 @@ func main() {
 	defer func() { _ = logger.Sync() }()
 
 	// Load Environment Variables
-	metricsPort := os.Getenv("METRICS_PORT")
-	healthPort := os.Getenv("HEALTH_PORT")
-	rawExpBackoff, expBackoffPresent := os.LookupEnv("EXPONENTIAL_BACKOFF")
+	metricsPort := os.Getenv(commonenv.MetricsPortEnvVarKey)
+	healthPort := os.Getenv(commonenv.HealthPortEnvVarKey)
+	rawExpBackoff, expBackoffPresent := os.LookupEnv(commonenv.ExponentialBackoffEnvVarKey)
 	exponentialBackoff, _ := strconv.ParseBool(rawExpBackoff)
-	maxRetryTime, _ := strconv.ParseInt(os.Getenv("MAX_RETRY_TIME"), 10, 64)
-	initialRetryInterval, _ := strconv.ParseInt(os.Getenv("INITIAL_RETRY_INTERVAL"), 10, 64)
-	kafkaBrokers := os.Getenv("KAFKA_BROKERS")
-	kafkaTopic := os.Getenv("KAFKA_TOPIC")
-	channelKey := os.Getenv("CHANNEL_KEY")
-	kafkaUsername := os.Getenv("KAFKA_USERNAME")
-	kafkaPassword := os.Getenv("KAFKA_PASSWORD")
+	maxRetryTime, _ := strconv.ParseInt(os.Getenv(commonenv.MaxRetryTimeEnvVarKey), 10, 64)
+	initialRetryInterval, _ := strconv.ParseInt(os.Getenv(commonenv.InitialRetryIntervalEnvVarKey), 10, 64)
+	kafkaBrokers := os.Getenv(commonenv.KafkaBrokerEnvVarKey)
+	kafkaTopic := os.Getenv(commonenv.KafkaTopicEnvVarKey)
+	channelKey := os.Getenv(commonenv.ChannelKeyEnvVarKey)
+	serviceName := os.Getenv(commonenv.ServiceNameEnvVarKey)
+	kafkaUsername := os.Getenv(commonenv.KafkaUsernameEnvVarKey)
+	kafkaPassword := os.Getenv(commonenv.KafkaPasswordEnvVarKey)
 	kafkaPasswordLog := ""
-	kafkaOffsetCommitMessageCount, _ := strconv.ParseInt(os.Getenv("KAFKA_OFFSET_COMMIT_MESSAGE_COUNT"), 10, 64)
-	kafkaOffsetCommitDurationMillis, _ := strconv.ParseInt(os.Getenv("KAFKA_OFFSET_COMMIT_DURATION_MILLIS"), 10, 64)
+	kafkaOffsetCommitMessageCount, _ := strconv.ParseInt(os.Getenv(commonenv.KafkaOffsetCommitMessageCountEnvVarKey), 10, 64)
+	kafkaOffsetCommitDurationMillis, _ := strconv.ParseInt(os.Getenv(commonenv.KafkaOffsetCommitDurationMillisEnvVarKey), 10, 64)
 
 	if len(kafkaPassword) > 0 {
 		kafkaPasswordLog = "*************"
@@ -72,18 +74,19 @@ func main() {
 
 	// Log Environment Variables
 	logger.Info("Environment Variables",
-		zap.String("HEALTH_PORT", healthPort),
-		zap.String("METRICS_PORT", metricsPort),
-		zap.Bool("EXPONENTIAL_BACKOFF", exponentialBackoff),
-		zap.Int64("INITIAL_RETRY_INTERVAL", initialRetryInterval),
-		zap.Int64("MAX_RETRY_TIME", maxRetryTime),
-		zap.String("KAFKA_BROKERS", kafkaBrokers),
-		zap.String("KAFKA_TOPIC", kafkaTopic),
-		zap.Int64("KAFKA_OFFSET_COMMIT_MESSAGE_COUNT", kafkaOffsetCommitMessageCount),
-		zap.Int64("KAFKA_OFFSET_COMMIT_DURATION_MILLIS", kafkaOffsetCommitDurationMillis),
-		zap.String("KAFKA_USERNAME", kafkaUsername),
-		zap.String("KAFKA_PASSWORD", kafkaPasswordLog),
-		zap.String("CHANNEL_KEY", channelKey))
+		zap.String(commonenv.HealthPortEnvVarKey, healthPort),
+		zap.String(commonenv.MetricsPortEnvVarKey, metricsPort),
+		zap.Bool(commonenv.ExponentialBackoffEnvVarKey, exponentialBackoff),
+		zap.Int64(commonenv.InitialRetryIntervalEnvVarKey, initialRetryInterval),
+		zap.Int64(commonenv.MaxRetryTimeEnvVarKey, maxRetryTime),
+		zap.String(commonenv.KafkaBrokerEnvVarKey, kafkaBrokers),
+		zap.String(commonenv.KafkaTopicEnvVarKey, kafkaTopic),
+		zap.Int64(commonenv.KafkaOffsetCommitMessageCountEnvVarKey, kafkaOffsetCommitMessageCount),
+		zap.Int64(commonenv.KafkaOffsetCommitDurationMillisEnvVarKey, kafkaOffsetCommitDurationMillis),
+		zap.String(commonenv.KafkaUsernameEnvVarKey, kafkaUsername),
+		zap.String(commonenv.KafkaPasswordEnvVarKey, kafkaPasswordLog),
+		zap.String(commonenv.ChannelKeyEnvVarKey, channelKey),
+		zap.String(commonenv.ServiceNameEnvVarKey, serviceName))
 
 	// Validate Required Environment Variables
 	if len(metricsPort) == 0 ||
@@ -94,11 +97,15 @@ func main() {
 		len(kafkaBrokers) == 0 ||
 		len(kafkaTopic) == 0 ||
 		len(channelKey) == 0 ||
+		len(serviceName) == 0 ||
 		kafkaOffsetCommitMessageCount <= 0 ||
 		kafkaOffsetCommitDurationMillis <= 0 {
 		logger.Fatal("Invalid / Missing Environment Variables - Terminating")
 	}
 
+	// Initialize Tracing (Watching config-tracing ConfigMap, Assumes Context Came From LoggingContext With Embedded K8S Client Key)
+	commonk8s.InitializeTracing(logger.Sugar(), ctx, serviceName)
+
 	// Start The Liveness And Readiness Servers
 	healthServer := dispatcherhealth.NewDispatcherHealthServer(healthPort)
 	healthServer.Start(logger)
diff --git a/pkg/channel/env/environment.go b/pkg/channel/env/environment.go
index be7bb9422..1b668463f 100644
--- a/pkg/channel/env/environment.go
+++ b/pkg/channel/env/environment.go
@@ -3,20 +3,10 @@ package env
 import (
 	"errors"
 	"go.uber.org/zap"
+	commonenv "knative.dev/eventing-kafka/pkg/common/env"
 	"os"
 )
 
-// Environment Constants
-const (
-
-	// Environment Variable Keys
-	MetricsPortEnvVarKey   = "METRICS_PORT"
-	HealthPortEnvVarKey    = "HEALTH_PORT"
-	KafkaBrokersEnvVarKey  = "KAFKA_BROKERS"
-	KafkaUsernameEnvVarKey = "KAFKA_USERNAME"
-	KafkaPasswordEnvVarKey = "KAFKA_PASSWORD"
-)
-
 // The Environment Struct
 type Environment struct {
 	MetricsPort   string
@@ -24,6 +14,7 @@ type Environment struct {
 	KafkaBrokers  string
 	KafkaUsername string
 	KafkaPassword string
+	ServiceName   string
 }
 
 // Load & Return The Environment Variables
@@ -31,11 +22,12 @@ func GetEnvironment(logger *zap.Logger) (Environment, error) {
 
 	// Create The Environment With Current Values
 	environment := Environment{
-		MetricsPort:   os.Getenv(MetricsPortEnvVarKey),
-		HealthPort:    os.Getenv(HealthPortEnvVarKey),
-		KafkaBrokers:  os.Getenv(KafkaBrokersEnvVarKey),
-		KafkaUsername: os.Getenv(KafkaUsernameEnvVarKey),
-		KafkaPassword: os.Getenv(KafkaPasswordEnvVarKey),
+		MetricsPort:   os.Getenv(commonenv.MetricsPortEnvVarKey),
+		HealthPort:    os.Getenv(commonenv.HealthPortEnvVarKey),
+		KafkaBrokers:  os.Getenv(commonenv.KafkaBrokerEnvVarKey),
+		KafkaUsername: os.Getenv(commonenv.KafkaUsernameEnvVarKey),
+		KafkaPassword: os.Getenv(commonenv.KafkaPasswordEnvVarKey),
+		ServiceName:   os.Getenv(commonenv.ServiceNameEnvVarKey),
 	}
 
 	// Safely Log The ControllerConfig Loaded From Environment Variables
@@ -56,9 +48,10 @@ func GetEnvironment(logger *zap.Logger) (Environment, error) {
 // Validate The Specified Environment Variables
 func validateEnvironment(logger *zap.Logger, environment Environment) error {
 
-	valid := validateRequiredEnvironmentVariable(logger, MetricsPortEnvVarKey, environment.MetricsPort) &&
-		validateRequiredEnvironmentVariable(logger, HealthPortEnvVarKey, environment.HealthPort) &&
-		validateRequiredEnvironmentVariable(logger, KafkaBrokersEnvVarKey, environment.KafkaBrokers)
+	valid := validateRequiredEnvironmentVariable(logger, commonenv.MetricsPortEnvVarKey, environment.MetricsPort) &&
+		validateRequiredEnvironmentVariable(logger, commonenv.HealthPortEnvVarKey, environment.HealthPort) &&
+		validateRequiredEnvironmentVariable(logger, commonenv.KafkaBrokerEnvVarKey, environment.KafkaBrokers) &&
+		validateRequiredEnvironmentVariable(logger, commonenv.ServiceNameEnvVarKey, environment.ServiceName)
 
 	if valid {
 		return nil
diff --git a/pkg/common/env/constants.go b/pkg/common/env/constants.go
new file mode 100644
index 000000000..10bbb31d9
--- /dev/null
+++ b/pkg/common/env/constants.go
@@ -0,0 +1,28 @@
+package env
+
+// Package Constants
+const (
+	// Eventing-Kafka Configuration
+	ServiceAccountEnvVarKey = "SERVICE_ACCOUNT"
+	MetricsPortEnvVarKey    = "METRICS_PORT"
+	HealthPortEnvVarKey     = "HEALTH_PORT"
+
+	// Kafka Authorization
+	KafkaBrokerEnvVarKey   = "KAFKA_BROKERS"
+	KafkaUsernameEnvVarKey = "KAFKA_USERNAME"
+	KafkaPasswordEnvVarKey = "KAFKA_PASSWORD"
+
+	// Kafka Configuration
+	KafkaProviderEnvVarKey                   = "KAFKA_PROVIDER"
+	KafkaOffsetCommitMessageCountEnvVarKey   = "KAFKA_OFFSET_COMMIT_MESSAGE_COUNT"
+	KafkaOffsetCommitDurationMillisEnvVarKey = "KAFKA_OFFSET_COMMIT_DURATION_MILLIS"
+	KafkaTopicEnvVarKey                      = "KAFKA_TOPIC"
+
+	// Dispatcher Configuration
+	ChannelKeyEnvVarKey           = "CHANNEL_KEY"
+	ServiceNameEnvVarKey          = "SERVICE_NAME"
+	ExponentialBackoffEnvVarKey   = "EXPONENTIAL_BACKOFF"
+	InitialRetryIntervalEnvVarKey = "INITIAL_RETRY_INTERVAL"
+	MaxRetryTimeEnvVarKey         = "MAX_RETRY_TIME"
+)
+
diff --git a/pkg/common/k8s/tracing.go b/pkg/common/k8s/tracing.go
new file mode 100644
index 000000000..8a6847def
--- /dev/null
+++ b/pkg/common/k8s/tracing.go
@@ -0,0 +1,35 @@
+package k8s
+
+import (
+	"context"
+	"go.uber.org/zap"
+	"k8s.io/client-go/kubernetes"
+	"knative.dev/eventing/pkg/tracing"
+	injectionclient "knative.dev/pkg/client/injection/kube/client"
+	"knative.dev/pkg/configmap"
+	"knative.dev/pkg/system"
+	tracingconfig "knative.dev/pkg/tracing/config"
+)
+
+//
+// Initialize The Specified Context With A Tracer (ConfigMap Watcher)
+// Assumes ctx Has K8S Client Injected Via LoggingContext (Or Similar)
+//
+func InitializeTracing(logger *zap.SugaredLogger, ctx context.Context, service string) {
+	k8sClient, ok := ctx.Value(injectionclient.Key{}).(kubernetes.Interface)
+	if !ok {
+		logger.Fatalw("Error getting kubernetes client from context (invalid interface)")
+	}
+
+	// Create A Watcher On The Tracing ConfigMap & Dynamically Update Tracing Configuration
+	cmw := configmap.NewInformedWatcher(k8sClient, system.Namespace())
+
+	if err := tracing.SetupDynamicPublishing(logger, cmw, service, tracingconfig.ConfigName); err != nil {
+		logger.Fatalw("Error setting up dynamic trace publishing", zap.Error(err))
+	}
+
+	// Start The Tracing ConfigMap Watcher
+	if err := cmw.Start(ctx.Done()); err != nil {
+		logger.Fatalw("Failed To Start ConfigMap Watcher", zap.Error(err))
+	}
+}
diff --git a/pkg/controller/env/environment.go b/pkg/controller/env/environment.go
index 4eaef1b32..98cc6c932 100644
--- a/pkg/controller/env/environment.go
+++ b/pkg/controller/env/environment.go
@@ -4,6 +4,7 @@ import (
 	"fmt"
 	"go.uber.org/zap"
 	"k8s.io/apimachinery/pkg/api/resource"
+	"knative.dev/eventing-kafka/pkg/common/env"
 	"os"
 	"strconv"
 	"strings"
@@ -11,28 +12,6 @@ import (
 
 // Package Constants
 const (
-	// Eventing-Kafka Configuration
-	ServiceAccountEnvVarKey = "SERVICE_ACCOUNT"
-	MetricsPortEnvVarKey    = "METRICS_PORT"
-	HealthPortEnvVarKey     = "HEALTH_PORT"
-
-	// Kafka Authorization
-	KafkaBrokerEnvVarKey   = "KAFKA_BROKERS"
-	KafkaUsernameEnvVarKey = "KAFKA_USERNAME"
-	KafkaPasswordEnvVarKey = "KAFKA_PASSWORD"
-
-	// Kafka Configuration
-	KafkaProviderEnvVarKey                   = "KAFKA_PROVIDER"
-	KafkaOffsetCommitMessageCountEnvVarKey   = "KAFKA_OFFSET_COMMIT_MESSAGE_COUNT"
-	KafkaOffsetCommitDurationMillisEnvVarKey = "KAFKA_OFFSET_COMMIT_DURATION_MILLIS"
-	KafkaTopicEnvVarKey                      = "KAFKA_TOPIC"
-
-	// Dispatcher Configuration
-	ChannelKeyEnvVarKey           = "CHANNEL_KEY"
-	ExponentialBackoffEnvVarKey   = "EXPONENTIAL_BACKOFF"
-	InitialRetryIntervalEnvVarKey = "INITIAL_RETRY_INTERVAL"
-	MaxRetryTimeEnvVarKey         = "MAX_RETRY_TIME"
-
 	// Default Values To Use If Not Available In Env Variables
 	DefaultKafkaOffsetCommitMessageCount   = "100"
 	DefaultKafkaOffsetCommitDurationMillis = "5000"
@@ -124,25 +103,25 @@ func GetEnvironment(logger *zap.Logger) (*Environment, error) {
 	environment := &Environment{}
 
 	// Get The Required K8S ServiceAccount Config Value
-	environment.ServiceAccount, err = getRequiredConfigValue(logger, ServiceAccountEnvVarKey)
+	environment.ServiceAccount, err = getRequiredConfigValue(logger, env.ServiceAccountEnvVarKey)
 	if err != nil {
 		return nil, err
 	}
 
 	// Get The Required Metrics Port Config Value & Convert To Int
-	metricsPortString, err := getRequiredConfigValue(logger, MetricsPortEnvVarKey)
+	metricsPortString, err := getRequiredConfigValue(logger, env.MetricsPortEnvVarKey)
 	if err != nil {
 		return nil, err
 	} else {
 		environment.MetricsPort, err = strconv.Atoi(metricsPortString)
 		if err != nil {
 			logger.Error("Invalid MetricsPort (Non Integer)", zap.String("Value", metricsPortString), zap.Error(err))
-			return nil, fmt.Errorf("invalid (non-integer) value '%s' for environment variable '%s'", metricsPortString, MetricsPortEnvVarKey)
+			return nil, fmt.Errorf("invalid (non-integer) value '%s' for environment variable '%s'", metricsPortString, env.MetricsPortEnvVarKey)
 		}
 	}
 
 	// Get The Required Kafka Provider Config Value
-	kafkaProviderString, err := getRequiredConfigValue(logger, KafkaProviderEnvVarKey)
+	kafkaProviderString, err := getRequiredConfigValue(logger, env.KafkaProviderEnvVarKey)
 	if err != nil {
 		return nil, err
 	} else {
@@ -155,12 +134,12 @@ func GetEnvironment(logger *zap.Logger) (*Environment, error) {
 			environment.KafkaProvider = KafkaProviderValueAzure
 		default:
 			logger.Error("Invalid / Unknown KafkaProvider", zap.String("Value", kafkaProviderString), zap.Error(err))
-			return nil, fmt.Errorf("invalid (unknown) value '%s' for environment variable '%s'", kafkaProviderString, KafkaProviderEnvVarKey)
+			return nil, fmt.Errorf("invalid (unknown) value '%s' for environment variable '%s'", kafkaProviderString, env.KafkaProviderEnvVarKey)
 		}
 	}
 
 	// Get The Optional KafkaOffsetCommitMessageCount Config Value
-	kafkaOffsetCommitMessageCountString := getOptionalConfigValue(logger, KafkaOffsetCommitMessageCountEnvVarKey, DefaultKafkaOffsetCommitMessageCount)
+	kafkaOffsetCommitMessageCountString := getOptionalConfigValue(logger, env.KafkaOffsetCommitMessageCountEnvVarKey, DefaultKafkaOffsetCommitMessageCount)
 	environment.KafkaOffsetCommitMessageCount, err = strconv.ParseInt(kafkaOffsetCommitMessageCountString, 10, 64)
 	if err != nil {
 		logger.Error("Invalid KafkaOffsetCommitMessageCount (Non Integer)", zap.String("Value", kafkaOffsetCommitMessageCountString), zap.Error(err))
@@ -168,7 +147,7 @@ func GetEnvironment(logger *zap.Logger) (*Environment, error) {
 	}
 
 	// Get The Optional KafkaOffsetCommitDurationMillis Config Value
-	kafkaOffsetCommitDurationMillisString := getOptionalConfigValue(logger, KafkaOffsetCommitDurationMillisEnvVarKey, DefaultKafkaOffsetCommitDurationMillis)
+	kafkaOffsetCommitDurationMillisString := getOptionalConfigValue(logger, env.KafkaOffsetCommitDurationMillisEnvVarKey, DefaultKafkaOffsetCommitDurationMillis)
 	environment.KafkaOffsetCommitDurationMillis, err = strconv.ParseInt(kafkaOffsetCommitDurationMillisString, 10, 64)
 	if err != nil {
 		logger.Error("Invalid KafkaOffsetCommitDurationMillis (Non Integer)", zap.String("Value", kafkaOffsetCommitDurationMillisString), zap.Error(err))
diff --git a/pkg/controller/kafkachannel/dispatcher.go b/pkg/controller/kafkachannel/dispatcher.go
index c4c0cc836..72c2e23a0 100644
--- a/pkg/controller/kafkachannel/dispatcher.go
+++ b/pkg/controller/kafkachannel/dispatcher.go
@@ -10,9 +10,9 @@ import (
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/util/intstr"
 	kafkav1alpha1 "knative.dev/eventing-contrib/kafka/channel/pkg/apis/messaging/v1alpha1"
+	commonenv "knative.dev/eventing-kafka/pkg/common/env"
 	"knative.dev/eventing-kafka/pkg/common/health"
 	"knative.dev/eventing-kafka/pkg/controller/constants"
-	"knative.dev/eventing-kafka/pkg/controller/env"
 	"knative.dev/eventing-kafka/pkg/controller/event"
 	"knative.dev/eventing-kafka/pkg/controller/util"
 	"knative.dev/pkg/controller"
@@ -317,39 +317,43 @@ func (r *Reconciler) dispatcherDeploymentEnvVars(channel *kafkav1alpha1.KafkaCha
 			Value: logging.ConfigMapName(),
 		},
 		{
-			Name:  env.MetricsPortEnvVarKey,
+			Name:  commonenv.MetricsPortEnvVarKey,
 			Value: strconv.Itoa(r.environment.MetricsPort),
 		},
 		{
-			Name:  env.HealthPortEnvVarKey,
+			Name:  commonenv.HealthPortEnvVarKey,
 			Value: strconv.Itoa(constants.HealthPort),
 		},
 		{
-			Name:  env.ChannelKeyEnvVarKey,
+			Name:  commonenv.ChannelKeyEnvVarKey,
 			Value: util.ChannelKey(channel),
 		},
 		{
-			Name:  env.KafkaTopicEnvVarKey,
+			Name:  commonenv.ServiceNameEnvVarKey,
+			Value: util.DispatcherDnsSafeName(channel),
+		},
+		{
+			Name:  commonenv.KafkaTopicEnvVarKey,
 			Value: topicName,
 		},
 		{
-			Name:  env.KafkaOffsetCommitMessageCountEnvVarKey,
+			Name:  commonenv.KafkaOffsetCommitMessageCountEnvVarKey,
 			Value: strconv.FormatInt(r.environment.KafkaOffsetCommitMessageCount, 10),
 		},
 		{
-			Name:  env.KafkaOffsetCommitDurationMillisEnvVarKey,
+			Name:  commonenv.KafkaOffsetCommitDurationMillisEnvVarKey,
 			Value: strconv.FormatInt(r.environment.KafkaOffsetCommitDurationMillis, 10),
 		},
 		{
-			Name:  env.ExponentialBackoffEnvVarKey,
+			Name:  commonenv.ExponentialBackoffEnvVarKey,
 			Value: strconv.FormatBool(r.environment.DispatcherRetryExponentialBackoff),
 		},
 		{
-			Name:  env.InitialRetryIntervalEnvVarKey,
+			Name:  commonenv.InitialRetryIntervalEnvVarKey,
 			Value: strconv.FormatInt(r.environment.DispatcherRetryInitialIntervalMillis, 10),
 		},
 		{
-			Name:  env.MaxRetryTimeEnvVarKey,
+			Name:  commonenv.MaxRetryTimeEnvVarKey,
 			Value: strconv.FormatInt(r.environment.DispatcherRetryTimeMillisMax, 10),
 		},
 	}
@@ -367,7 +371,7 @@ func (r *Reconciler) dispatcherDeploymentEnvVars(channel *kafkav1alpha1.KafkaCha
 
 		// Append The Kafka Brokers As Env Var
 		envVars = append(envVars, corev1.EnvVar{
-			Name: env.KafkaBrokerEnvVarKey,
+			Name: commonenv.KafkaBrokerEnvVarKey,
 			ValueFrom: &corev1.EnvVarSource{
 				SecretKeyRef: &corev1.SecretKeySelector{
 					LocalObjectReference: corev1.LocalObjectReference{Name: kafkaSecret},
@@ -378,7 +382,7 @@ func (r *Reconciler) dispatcherDeploymentEnvVars(channel *kafkav1alpha1.KafkaCha
 
 		// Append The Kafka Username As Env Var
 		envVars = append(envVars, corev1.EnvVar{
-			Name: env.KafkaUsernameEnvVarKey,
+			Name: commonenv.KafkaUsernameEnvVarKey,
 			ValueFrom: &corev1.EnvVarSource{
 				SecretKeyRef: &corev1.SecretKeySelector{
 					LocalObjectReference: corev1.LocalObjectReference{Name: kafkaSecret},
@@ -389,7 +393,7 @@ func (r *Reconciler) dispatcherDeploymentEnvVars(channel *kafkav1alpha1.KafkaCha
 
 		// Append The Kafka Password As Env Var
 		envVars = append(envVars, corev1.EnvVar{
-			Name: env.KafkaPasswordEnvVarKey,
+			Name: commonenv.KafkaPasswordEnvVarKey,
 			ValueFrom: &corev1.EnvVarSource{
 				SecretKeyRef: &corev1.SecretKeySelector{
 					LocalObjectReference: corev1.LocalObjectReference{Name: kafkaSecret},
diff --git a/pkg/controller/kafkasecret/channel.go b/pkg/controller/kafkasecret/channel.go
index c8851f365..f337de504 100644
--- a/pkg/controller/kafkasecret/channel.go
+++ b/pkg/controller/kafkasecret/channel.go
@@ -9,9 +9,9 @@ import (
 	"k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/util/intstr"
+	commonenv "knative.dev/eventing-kafka/pkg/common/env"
 	"knative.dev/eventing-kafka/pkg/common/health"
 	"knative.dev/eventing-kafka/pkg/controller/constants"
-	"knative.dev/eventing-kafka/pkg/controller/env"
 	"knative.dev/eventing-kafka/pkg/controller/event"
 	"knative.dev/eventing-kafka/pkg/controller/util"
 	"knative.dev/pkg/controller"
@@ -334,18 +334,22 @@ func (r *Reconciler) channelDeploymentEnvVars(secret *corev1.Secret) ([]corev1.E
 			Value: logging.ConfigMapName(),
 		},
 		{
-			Name:  env.MetricsPortEnvVarKey,
+			Name:  commonenv.ServiceNameEnvVarKey,
+			Value: util.ChannelDnsSafeName(secret.Name),
+		},
+		{
+			Name:  commonenv.MetricsPortEnvVarKey,
 			Value: strconv.Itoa(r.environment.MetricsPort),
 		},
 		{
-			Name:  env.HealthPortEnvVarKey,
+			Name:  commonenv.HealthPortEnvVarKey,
 			Value: strconv.Itoa(constants.HealthPort),
 		},
 	}
 
 	// Append The Kafka Brokers As Env Var
 	envVars = append(envVars, corev1.EnvVar{
-		Name: env.KafkaBrokerEnvVarKey,
+		Name: commonenv.KafkaBrokerEnvVarKey,
 		ValueFrom: &corev1.EnvVarSource{
 			SecretKeyRef: &corev1.SecretKeySelector{
 				LocalObjectReference: corev1.LocalObjectReference{Name: secret.Name},
@@ -356,7 +360,7 @@ func (r *Reconciler) channelDeploymentEnvVars(secret *corev1.Secret) ([]corev1.E
 
 	// Append The Kafka Username As Env Var
 	envVars = append(envVars, corev1.EnvVar{
-		Name: env.KafkaUsernameEnvVarKey,
+		Name: commonenv.KafkaUsernameEnvVarKey,
 		ValueFrom: &corev1.EnvVarSource{
 			SecretKeyRef: &corev1.SecretKeySelector{
 				LocalObjectReference: corev1.LocalObjectReference{Name: secret.Name},
@@ -367,7 +371,7 @@ func (r *Reconciler) channelDeploymentEnvVars(secret *corev1.Secret) ([]corev1.E
 
 	// Append The Kafka Password As Env Var
 	envVars = append(envVars, corev1.EnvVar{
-		Name: env.KafkaPasswordEnvVarKey,
+		Name: commonenv.KafkaPasswordEnvVarKey,
 		ValueFrom: &corev1.EnvVarSource{
 			SecretKeyRef: &corev1.SecretKeySelector{
 				LocalObjectReference: corev1.LocalObjectReference{Name: secret.Name},


Test patch:
diff --git a/pkg/channel/env/environment_test.go b/pkg/channel/env/environment_test.go
index bbdf2daa3..78262e808 100644
--- a/pkg/channel/env/environment_test.go
+++ b/pkg/channel/env/environment_test.go
@@ -3,6 +3,7 @@ package env
 import (
 	"errors"
 	"github.com/stretchr/testify/assert"
+	"knative.dev/eventing-kafka/pkg/common/env"
 	logtesting "knative.dev/pkg/logging/testing"
 	"os"
 	"testing"
@@ -12,6 +13,7 @@ import (
 const (
 	metricsPort   = "TestMetricsPort"
 	healthPort    = "TestHealthPort"
+	serviceName   = "TestServiceName"
 	kafkaBrokers  = "TestKafkaBrokers"
 	kafkaUsername = "TestKafkaUsername"
 	kafkaPassword = "TestKafkaPassword"
@@ -22,6 +24,7 @@ type TestCase struct {
 	name          string
 	metricsPort   string
 	healthPort    string
+	serviceName   string
 	kafkaBrokers  string
 	kafkaUsername string
 	kafkaPassword string
@@ -59,11 +62,12 @@ func TestGetEnvironment(t *testing.T) {
 
 		// (Re)Setup The Environment Variables From TestCase
 		os.Clearenv()
-		assert.Nil(t, os.Setenv(MetricsPortEnvVarKey, testCase.metricsPort))
-		assert.Nil(t, os.Setenv(HealthPortEnvVarKey, testCase.healthPort))
-		assert.Nil(t, os.Setenv(KafkaBrokersEnvVarKey, testCase.kafkaBrokers))
-		assert.Nil(t, os.Setenv(KafkaUsernameEnvVarKey, testCase.kafkaUsername))
-		assert.Nil(t, os.Setenv(KafkaPasswordEnvVarKey, testCase.kafkaPassword))
+		assert.Nil(t, os.Setenv(env.MetricsPortEnvVarKey, testCase.metricsPort))
+		assert.Nil(t, os.Setenv(env.HealthPortEnvVarKey, testCase.healthPort))
+		assert.Nil(t, os.Setenv(env.ServiceNameEnvVarKey, testCase.serviceName))
+		assert.Nil(t, os.Setenv(env.KafkaBrokerEnvVarKey, testCase.kafkaBrokers))
+		assert.Nil(t, os.Setenv(env.KafkaUsernameEnvVarKey, testCase.kafkaUsername))
+		assert.Nil(t, os.Setenv(env.KafkaPasswordEnvVarKey, testCase.kafkaPassword))
 
 		// Perform The Test
 		environment, err := GetEnvironment(logger)
@@ -88,6 +92,7 @@ func getValidTestCase(name string) TestCase {
 		name:          name,
 		metricsPort:   metricsPort,
 		healthPort:    healthPort,
+		serviceName:   serviceName,
 		kafkaBrokers:  kafkaBrokers,
 		kafkaUsername: kafkaUsername,
 		kafkaPassword: kafkaPassword,
diff --git a/pkg/common/k8s/tracing_test.go b/pkg/common/k8s/tracing_test.go
new file mode 100644
index 000000000..f61d9f36a
--- /dev/null
+++ b/pkg/common/k8s/tracing_test.go
@@ -0,0 +1,55 @@
+package k8s
+
+import (
+	"context"
+	"github.com/stretchr/testify/assert"
+	corev1 "k8s.io/api/core/v1"
+	v1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/client-go/kubernetes/fake"
+	"knative.dev/eventing-kafka/pkg/common/kafka/constants"
+	injectionclient "knative.dev/pkg/client/injection/kube/client"
+	logtesting "knative.dev/pkg/logging/testing"
+	"knative.dev/pkg/system"
+	tracingconfig "knative.dev/pkg/tracing/config"
+	"os"
+	"testing"
+)
+
+// Test The InitializeTracing() Functionality
+func TestInitializeTracing(t *testing.T) {
+
+	// Test Data
+	ctx := context.TODO()
+	service := "TestService"
+
+	// Obtain a Test Logger (Required By Tracing Function)
+	logger := logtesting.TestLogger(t)
+
+	// Setup Environment
+	assert.Nil(t, os.Setenv(system.NamespaceEnvKey, constants.KnativeEventingNamespace))
+
+	// Create A Test Tracing ConfigMap For The SetupDynamicPublishing() Call To Watch
+	tracingConfigMap := &corev1.ConfigMap{
+		TypeMeta: v1.TypeMeta{
+			Kind:       "ConfigMap",
+			APIVersion: corev1.SchemeGroupVersion.String(),
+		},
+		ObjectMeta: v1.ObjectMeta{
+			Name:      tracingconfig.ConfigName,
+			Namespace: system.Namespace(),
+		},
+	}
+
+	// Create The Fake K8S Client And Add It To The ConfigMap
+	fakeK8sClient := fake.NewSimpleClientset(tracingConfigMap)
+
+	// Add The Fake K8S Client To The Context (Required By InitializeTracing)
+	ctx = context.WithValue(ctx, injectionclient.Key{}, fakeK8sClient)
+
+	// Perform The Test (Initialize The Tracing Watcher)
+	InitializeTracing(logger, ctx, service)
+
+	// If the InitializeTracing Succeeds, it will not fatally exit
+	// (Not the best test of failure conditions but it does run through the SetupDynamicPublishing() call at least
+	//  and verify that the happy-path doesn't error out)
+}
diff --git a/pkg/controller/env/environment_test.go b/pkg/controller/env/environment_test.go
index 18bc7c720..fc6c761e2 100644
--- a/pkg/controller/env/environment_test.go
+++ b/pkg/controller/env/environment_test.go
@@ -4,6 +4,7 @@ import (
 	"fmt"
 	"github.com/stretchr/testify/assert"
 	"go.uber.org/zap"
+	commonenv "knative.dev/eventing-kafka/pkg/common/env"
 	"log"
 	"os"
 	"strconv"
@@ -85,27 +86,27 @@ func TestGetEnvironment(t *testing.T) {
 
 	testCase = getValidTestCase("Missing Required Config - ServiceAccount")
 	testCase.serviceAccount = ""
-	testCase.expectedError = getMissingRequiredEnvironmentVariableError(ServiceAccountEnvVarKey)
+	testCase.expectedError = getMissingRequiredEnvironmentVariableError(commonenv.ServiceAccountEnvVarKey)
 	testCases = append(testCases, testCase)
 
 	testCase = getValidTestCase("Missing Required Config - MetricsPort")
 	testCase.metricsPort = ""
-	testCase.expectedError = getMissingRequiredEnvironmentVariableError(MetricsPortEnvVarKey)
+	testCase.expectedError = getMissingRequiredEnvironmentVariableError(commonenv.MetricsPortEnvVarKey)
 	testCases = append(testCases, testCase)
 
 	testCase = getValidTestCase("Invalid Config - MetricsPort")
 	testCase.metricsPort = "NAN"
-	testCase.expectedError = getInvalidIntegerEnvironmentVariableError(testCase.metricsPort, MetricsPortEnvVarKey)
+	testCase.expectedError = getInvalidIntegerEnvironmentVariableError(testCase.metricsPort, commonenv.MetricsPortEnvVarKey)
 	testCases = append(testCases, testCase)
 
 	testCase = getValidTestCase("Missing Required Config - KafkaProvider")
 	testCase.kafkaProvider = ""
-	testCase.expectedError = getMissingRequiredEnvironmentVariableError(KafkaProviderEnvVarKey)
+	testCase.expectedError = getMissingRequiredEnvironmentVariableError(commonenv.KafkaProviderEnvVarKey)
 	testCases = append(testCases, testCase)
 
 	testCase = getValidTestCase("Invalid Config - KafkaProvider")
 	testCase.kafkaProvider = "foo"
-	testCase.expectedError = fmt.Errorf("invalid (unknown) value 'foo' for environment variable '%s'", KafkaProviderEnvVarKey)
+	testCase.expectedError = fmt.Errorf("invalid (unknown) value 'foo' for environment variable '%s'", commonenv.KafkaProviderEnvVarKey)
 	testCases = append(testCases, testCase)
 
 	testCase = getValidTestCase("Missing Optional Config - KafkaOffsetCommitMessageCount")
@@ -287,14 +288,14 @@ func TestGetEnvironment(t *testing.T) {
 
 		// (Re)Setup The Environment Variables From TestCase
 		os.Clearenv()
-		assert.Nil(t, os.Setenv(ServiceAccountEnvVarKey, testCase.serviceAccount))
+		assert.Nil(t, os.Setenv(commonenv.ServiceAccountEnvVarKey, testCase.serviceAccount))
 		if len(testCase.metricsPort) > 0 {
-			assert.Nil(t, os.Setenv(MetricsPortEnvVarKey, testCase.metricsPort))
+			assert.Nil(t, os.Setenv(commonenv.MetricsPortEnvVarKey, testCase.metricsPort))
 		}
 
-		assert.Nil(t, os.Setenv(KafkaProviderEnvVarKey, testCase.kafkaProvider))
-		assert.Nil(t, os.Setenv(KafkaOffsetCommitMessageCountEnvVarKey, testCase.kafkaOffsetCommitMessageCount))
-		assert.Nil(t, os.Setenv(KafkaOffsetCommitDurationMillisEnvVarKey, testCase.kafkaOffsetCommitDurationMillis))
+		assert.Nil(t, os.Setenv(commonenv.KafkaProviderEnvVarKey, testCase.kafkaProvider))
+		assert.Nil(t, os.Setenv(commonenv.KafkaOffsetCommitMessageCountEnvVarKey, testCase.kafkaOffsetCommitMessageCount))
+		assert.Nil(t, os.Setenv(commonenv.KafkaOffsetCommitDurationMillisEnvVarKey, testCase.kafkaOffsetCommitDurationMillis))
 		if len(testCase.defaultNumPartitions) > 0 {
 			assert.Nil(t, os.Setenv(DefaultNumPartitionsEnvVarKey, testCase.defaultNumPartitions))
 		}
diff --git a/pkg/controller/test/data.go b/pkg/controller/test/data.go
index a003fdb15..b0f406103 100644
--- a/pkg/controller/test/data.go
+++ b/pkg/controller/test/data.go
@@ -10,6 +10,7 @@ import (
 	"k8s.io/apimachinery/pkg/util/intstr"
 	clientgotesting "k8s.io/client-go/testing"
 	kafkav1alpha1 "knative.dev/eventing-contrib/kafka/channel/pkg/apis/messaging/v1alpha1"
+	commonenv "knative.dev/eventing-kafka/pkg/common/env"
 	"knative.dev/eventing-kafka/pkg/common/health"
 	kafkautil "knative.dev/eventing-kafka/pkg/common/kafka/util"
 	"knative.dev/eventing-kafka/pkg/controller/constants"
@@ -55,6 +56,7 @@ const (
 	KafkaSecretName       = "kafkasecret-name"
 	KafkaSecretKey        = KafkaSecretNamespace + "/" + KafkaSecretName
 	ChannelDeploymentName = KafkaSecretName + "-channel"
+	ChannelServiceName    = ChannelDeploymentName
 	TopicName             = KafkaChannelNamespace + "." + KafkaChannelName
 
 	KafkaSecretDataValueBrokers  = "TestKafkaSecretDataBrokers"
@@ -475,15 +477,19 @@ func NewKafkaChannelChannelDeployment() *appsv1.Deployment {
 									Value: logging.ConfigMapName(),
 								},
 								{
-									Name:  env.MetricsPortEnvVarKey,
+									Name:  commonenv.ServiceNameEnvVarKey,
+									Value: ChannelServiceName,
+								},
+								{
+									Name:  commonenv.MetricsPortEnvVarKey,
 									Value: strconv.Itoa(MetricsPort),
 								},
 								{
-									Name:  env.HealthPortEnvVarKey,
+									Name:  commonenv.HealthPortEnvVarKey,
 									Value: strconv.Itoa(HealthPort),
 								},
 								{
-									Name: env.KafkaBrokerEnvVarKey,
+									Name: commonenv.KafkaBrokerEnvVarKey,
 									ValueFrom: &corev1.EnvVarSource{
 										SecretKeyRef: &corev1.SecretKeySelector{
 											LocalObjectReference: corev1.LocalObjectReference{Name: KafkaSecretName},
@@ -492,7 +498,7 @@ func NewKafkaChannelChannelDeployment() *appsv1.Deployment {
 									},
 								},
 								{
-									Name: env.KafkaUsernameEnvVarKey,
+									Name: commonenv.KafkaUsernameEnvVarKey,
 									ValueFrom: &corev1.EnvVarSource{
 										SecretKeyRef: &corev1.SecretKeySelector{
 											LocalObjectReference: corev1.LocalObjectReference{Name: KafkaSecretName},
@@ -501,7 +507,7 @@ func NewKafkaChannelChannelDeployment() *appsv1.Deployment {
 									},
 								},
 								{
-									Name: env.KafkaPasswordEnvVarKey,
+									Name: commonenv.KafkaPasswordEnvVarKey,
 									ValueFrom: &corev1.EnvVarSource{
 										SecretKeyRef: &corev1.SecretKeySelector{
 											LocalObjectReference: corev1.LocalObjectReference{Name: KafkaSecretName},
@@ -648,43 +654,47 @@ func NewKafkaChannelDispatcherDeployment() *appsv1.Deployment {
 									Value: logging.ConfigMapName(),
 								},
 								{
-									Name:  env.MetricsPortEnvVarKey,
+									Name:  commonenv.MetricsPortEnvVarKey,
 									Value: strconv.Itoa(MetricsPort),
 								},
 								{
-									Name:  env.HealthPortEnvVarKey,
+									Name:  commonenv.HealthPortEnvVarKey,
 									Value: strconv.Itoa(HealthPort),
 								},
 								{
-									Name:  env.ChannelKeyEnvVarKey,
+									Name:  commonenv.ChannelKeyEnvVarKey,
 									Value: fmt.Sprintf("%s/%s", KafkaChannelNamespace, KafkaChannelName),
 								},
 								{
-									Name:  env.KafkaTopicEnvVarKey,
+									Name:  commonenv.ServiceNameEnvVarKey,
+									Value: dispatcherName,
+								},
+								{
+									Name:  commonenv.KafkaTopicEnvVarKey,
 									Value: topicName,
 								},
 								{
-									Name:  env.KafkaOffsetCommitMessageCountEnvVarKey,
+									Name:  commonenv.KafkaOffsetCommitMessageCountEnvVarKey,
 									Value: strconv.Itoa(KafkaOffsetCommitMessageCount),
 								},
 								{
-									Name:  env.KafkaOffsetCommitDurationMillisEnvVarKey,
+									Name:  commonenv.KafkaOffsetCommitDurationMillisEnvVarKey,
 									Value: strconv.Itoa(KafkaOffsetCommitDurationMillis),
 								},
 								{
-									Name:  env.ExponentialBackoffEnvVarKey,
+									Name:  commonenv.ExponentialBackoffEnvVarKey,
 									Value: strconv.FormatBool(DefaultExponentialBackoff),
 								},
 								{
-									Name:  env.InitialRetryIntervalEnvVarKey,
+									Name:  commonenv.InitialRetryIntervalEnvVarKey,
 									Value: strconv.Itoa(DefaultEventRetryInitialIntervalMillis),
 								},
 								{
-									Name:  env.MaxRetryTimeEnvVarKey,
+									Name:  commonenv.MaxRetryTimeEnvVarKey,
 									Value: strconv.Itoa(DefaultEventRetryTimeMillisMax),
 								},
 								{
-									Name: env.KafkaBrokerEnvVarKey,
+									Name: commonenv.KafkaBrokerEnvVarKey,
 									ValueFrom: &corev1.EnvVarSource{
 										SecretKeyRef: &corev1.SecretKeySelector{
 											LocalObjectReference: corev1.LocalObjectReference{Name: KafkaSecretName},
@@ -693,7 +703,7 @@ func NewKafkaChannelDispatcherDeployment() *appsv1.Deployment {
 									},
 								},
 								{
-									Name: env.KafkaUsernameEnvVarKey,
+									Name: commonenv.KafkaUsernameEnvVarKey,
 									ValueFrom: &corev1.EnvVarSource{
 										SecretKeyRef: &corev1.SecretKeySelector{
 											LocalObjectReference: corev1.LocalObjectReference{Name: KafkaSecretName},
@@ -702,7 +712,7 @@ func NewKafkaChannelDispatcherDeployment() *appsv1.Deployment {
 									},
 								},
 								{
-									Name: env.KafkaPasswordEnvVarKey,
+									Name: commonenv.KafkaPasswordEnvVarKey,
 									ValueFrom: &corev1.EnvVarSource{
 										SecretKeyRef: &corev1.SecretKeySelector{
 											LocalObjectReference: corev1.LocalObjectReference{Name: KafkaSecretName},