Configuration for tracing in sources

pkg/adapter/config.go

@@ -40,6 +40,12 @@ type EnvConfig struct {
 	// This is used to configure the logging config, the config is stored in
 	// a config map inside the controllers namespace and copied here.
 	LoggingConfigJson string `envconfig:"K_LOGGING_CONFIG" required:"true"`
+
+	// TracingConfigJson is a json string of tracing.Config.
+	// This is used to configure the tracing config, the config is stored in
+	// a config map inside the controllers namespace and copied here.
+	// Default is no-op.
+	TracingConfigJson string `envconfig:"K_TRACING_CONFIG"`
 }
 
 // EnvConfigAccessor defines accessors for the minimal
@@ -56,6 +62,9 @@ type EnvConfigAccessor interface {
 
 	// Get the json string of logging.Config.
 	GetLoggingConfigJson() string
+
+	// Get the json string of tracubg.Config.
+	GetTracingConfigJson() string
 }
 
 func (e *EnvConfig) GetMetricsConfigJson() string {
@@ -66,6 +75,10 @@ func (e *EnvConfig) GetLoggingConfigJson() string {
 	return e.LoggingConfigJson
 }
 
+func (e *EnvConfig) GetTracingConfigJson() string {
+	return e.TracingConfigJson
+}
+
 func (e *EnvConfig) GetSinkURI() string {
 	if e.Sink != "" {
 		return e.Sink

pkg/adapter/config_test.go

@@ -33,6 +33,7 @@ func TestEnvConfig(t *testing.T) {
 	os.Setenv("NAMESPACE", "ns")
 	os.Setenv("K_METRICS_CONFIG", "metrics")
 	os.Setenv("K_LOGGING_CONFIG", "logging")
+	os.Setenv("K_TRACING_CONFIG", "tracing")
 	os.Setenv("MODE", "mymode")
 
 	var env myEnvConfig

pkg/adapter/main.go

@@ -36,6 +36,7 @@ import (
 	"knative.dev/pkg/profiling"
 	"knative.dev/pkg/signals"
 	"knative.dev/pkg/source"
+	tracingconfig "knative.dev/pkg/tracing/config"
 
 	"knative.dev/eventing/pkg/kncloudevents"
 	"knative.dev/eventing/pkg/tracing"
@@ -115,7 +116,12 @@ func MainWithContext(ctx context.Context, component string, ector EnvConfigConst
 		logger.Error("error building statsreporter", zap.Error(err))
 	}
 
-	if err = tracing.SetupStaticPublishing(logger, "", tracing.OnePercentSampling); err != nil {
+	// Retrieve tracing config
+	config, err := tracingconfig.JsonToTracingConfig(env.GetTracingConfigJson())
+	if err != nil {
+		logger.Warn("Tracing configuration is invalid, using the no-op default", zap.Error(err))
+	}
+	if err := tracing.SetupStaticPublishing(logger, component, config); err != nil {
 		// If tracing doesn't work, we will log an error, but allow the adapter
 		// to continue to start.
 		logger.Error("Error setting up trace publishing", zap.Error(err))

pkg/adapter/main_message_adapter.go

@@ -26,6 +26,7 @@ import (
 
 	"knative.dev/pkg/profiling"
 	"knative.dev/pkg/signals"
+	tracingconfig "knative.dev/pkg/tracing/config"
 
 	"github.com/kelseyhightower/envconfig"
 	"go.opencensus.io/stats/view"
@@ -112,7 +113,12 @@ func MainMessageAdapterWithContext(ctx context.Context, component string, ector
 		logger.Error("error building statsreporter", zap.Error(err))
 	}
 
-	if err = tracing.SetupStaticPublishing(logger, "", tracing.OnePercentSampling); err != nil {
+	// Retrieve tracing config
+	config, err := tracingconfig.JsonToTracingConfig(env.GetTracingConfigJson())
+	if err != nil {
+		logger.Warn("Tracing configuration is invalid, using the no-op default", zap.Error(err))
+	}
+	if err := tracing.SetupStaticPublishing(logger, component, config); err != nil {
 		// If tracing doesn't work, we will log an error, but allow the adapter
 		// to continue to start.
 		logger.Error("Error setting up trace publishing", zap.Error(err))

pkg/adapter/v2/config.go

@@ -19,10 +19,13 @@ import (
 	"encoding/json"
 
 	"go.uber.org/zap"
+	tracingconfig "knative.dev/pkg/tracing/config"
 
 	duckv1 "knative.dev/pkg/apis/duck/v1"
 	"knative.dev/pkg/logging"
 	"knative.dev/pkg/metrics"
+
+	"knative.dev/eventing/pkg/tracing"
 )
 
 type EnvConfigConstructor func() EnvConfigAccessor
@@ -58,6 +61,12 @@ type EnvConfig struct {
 	// This is used to configure the logging config, the config is stored in
 	// a config map inside the controllers namespace and copied here.
 	LoggingConfigJson string `envconfig:"K_LOGGING_CONFIG" required:"true"`
+
+	// TracingConfigJson is a json string of tracing.Config.
+	// This is used to configure the tracing config, the config is stored in
+	// a config map inside the controllers namespace and copied here.
+	// Default is no-op.
+	TracingConfigJson string `envconfig:"K_TRACING_CONFIG"`
 }
 
 // EnvConfigAccessor defines accessors for the minimal
@@ -81,6 +90,8 @@ type EnvConfigAccessor interface {
 	// Get the parsed logger.
 	GetLogger() *zap.SugaredLogger
 
+	SetupTracing(*zap.SugaredLogger) error
+
 	GetCloudEventOverrides() (*duckv1.CloudEventOverrides, error)
 }
 
@@ -126,6 +137,14 @@ func (e *EnvConfig) GetName() string {
 	return e.Name
 }
 
+func (e *EnvConfig) SetupTracing(logger *zap.SugaredLogger) error {
+	config, err := tracingconfig.JsonToTracingConfig(e.TracingConfigJson)
+	if err != nil {
+		logger.Warn("Tracing configuration is invalid, using the no-op default", zap.Error(err))
+	}
+	return tracing.SetupStaticPublishing(logger, e.Component, config)
+}
+
 func (e *EnvConfig) GetCloudEventOverrides() (*duckv1.CloudEventOverrides, error) {
 	var ceOverrides duckv1.CloudEventOverrides
 	if len(e.CEOverrides) > 0 {

pkg/adapter/v2/config_test.go

@@ -33,6 +33,7 @@ func TestEnvConfig(t *testing.T) {
 	os.Setenv("NAMESPACE", "ns")
 	os.Setenv("K_METRICS_CONFIG", "metrics")
 	os.Setenv("K_LOGGING_CONFIG", "logging")
+	os.Setenv("K_TRACING_CONFIG", "tracing")
 	os.Setenv("MODE", "mymode") // note: custom to this test impl
 
 	var env myEnvConfig

pkg/adapter/v2/main.go

@@ -24,7 +24,6 @@ import (
 	"time"
 
 	cloudevents "github.com/cloudevents/sdk-go/v2"
-
 	"github.com/kelseyhightower/envconfig"
 	"go.opencensus.io/stats/view"
 	"go.uber.org/zap"
@@ -33,8 +32,6 @@ import (
 	"knative.dev/pkg/profiling"
 	"knative.dev/pkg/signals"
 	"knative.dev/pkg/source"
-
-	"knative.dev/eventing/pkg/tracing"
 )
 
 type Adapter interface {
@@ -95,7 +92,7 @@ func MainWithContext(ctx context.Context, component string, ector EnvConfigConst
 		logger.Error("error building statsreporter", zap.Error(err))
 	}
 
-	if err = tracing.SetupStaticPublishing(logger, "", tracing.OnePercentSampling); err != nil {
+	if err := env.SetupTracing(logger); err != nil {
 		// If tracing doesn't work, we will log an error, but allow the adapter
 		// to continue to start.
 		logger.Error("Error setting up trace publishing", zap.Error(err))

pkg/reconciler/apiserversource/apiserversource.go

@@ -32,18 +32,20 @@ import (
 	"k8s.io/apimachinery/pkg/runtime/schema"
 	"k8s.io/client-go/kubernetes"
 
-	"knative.dev/eventing/pkg/apis/sources/v1alpha2"
-	apiserversourcereconciler "knative.dev/eventing/pkg/client/injection/reconciler/sources/v1alpha2/apiserversource"
-	listers "knative.dev/eventing/pkg/client/listers/sources/v1alpha2"
-	"knative.dev/eventing/pkg/logging"
-	"knative.dev/eventing/pkg/reconciler/apiserversource/resources"
-	"knative.dev/eventing/pkg/utils"
 	duckv1 "knative.dev/pkg/apis/duck/v1"
 	"knative.dev/pkg/controller"
 	pkgLogging "knative.dev/pkg/logging"
 	"knative.dev/pkg/metrics"
 	pkgreconciler "knative.dev/pkg/reconciler"
 	"knative.dev/pkg/resolver"
+	tracingconfig "knative.dev/pkg/tracing/config"
+
+	"knative.dev/eventing/pkg/apis/sources/v1alpha2"
+	apiserversourcereconciler "knative.dev/eventing/pkg/client/injection/reconciler/sources/v1alpha2/apiserversource"
+	listers "knative.dev/eventing/pkg/client/listers/sources/v1alpha2"
+	"knative.dev/eventing/pkg/logging"
+	"knative.dev/eventing/pkg/reconciler/apiserversource/resources"
+	"knative.dev/eventing/pkg/utils"
 )
 
 const (
@@ -80,6 +82,7 @@ type Reconciler struct {
 	loggingContext context.Context
 	loggingConfig  *pkgLogging.Config
 	metricsConfig  *metrics.ExporterOptions
+	tracingCfg     *tracingconfig.Config
 }
 
 var _ apiserversourcereconciler.Interface = (*Reconciler)(nil)
@@ -148,13 +151,19 @@ func (r *Reconciler) createReceiveAdapter(ctx context.Context, src *v1alpha2.Api
 		logging.FromContext(ctx).Error("error while converting metrics config to json", zap.Any("receiveAdapter", err))
 	}
 
+	tracingCfg, err := tracingconfig.TracingConfigToJson(r.tracingCfg)
+	if err != nil {
+		logging.FromContext(ctx).Error("error while converting tracing config to json", zap.Any("receiveAdapter", err))
+	}
+
 	adapterArgs := resources.ReceiveAdapterArgs{
 		Image:         r.receiveAdapterImage,
 		Source:        src,
 		Labels:        resources.Labels(src.Name),
 		SinkURI:       sinkURI,
 		LoggingConfig: loggingConfig,
 		MetricsConfig: metricsConfig,
+		TracingConfig: tracingCfg,
 	}
 	expected, err := resources.MakeReceiveAdapter(&adapterArgs)
 	if err != nil {
@@ -240,6 +249,21 @@ func (r *Reconciler) UpdateFromMetricsConfigMap(cfg *corev1.ConfigMap) {
 	logging.FromContext(r.loggingContext).Debug("Update from metrics ConfigMap", zap.Any("ConfigMap", cfg))
 }
 
+// TODO determine how to push the updated metrics config to existing data plane Pods.
+func (r *Reconciler) UpdateFromTracingConfigMap(cfg *corev1.ConfigMap) {
+	if cfg != nil {
+		delete(cfg.Data, "_example")
+	}
+
+	tracingCfg, err := tracingconfig.NewTracingConfigFromMap(cfg.Data)
+	if err != nil {
+		logging.FromContext(r.loggingContext).Warn("failed to create tracing config from configmap", zap.String("cfg.Name", cfg.Name))
+		return
+	}
+
+	r.tracingCfg = tracingCfg
+}
+
 func (r *Reconciler) runAccessCheck(src *v1alpha2.ApiServerSource) error {
 	if src.Spec.Resources == nil || len(src.Spec.Resources) == 0 {
 		src.Status.MarkSufficientPermissions()

pkg/reconciler/apiserversource/controller.go

@@ -26,13 +26,15 @@ import (
 	"knative.dev/pkg/logging"
 	"knative.dev/pkg/metrics"
 	"knative.dev/pkg/resolver"
+	tracingconfig "knative.dev/pkg/tracing/config"
 
 	"knative.dev/eventing/pkg/apis/sources/v1alpha1"
 
-	apiserversourceinformer "knative.dev/eventing/pkg/client/injection/informers/sources/v1alpha2/apiserversource"
-	apiserversourcereconciler "knative.dev/eventing/pkg/client/injection/reconciler/sources/v1alpha2/apiserversource"
 	kubeclient "knative.dev/pkg/client/injection/kube/client"
 	deploymentinformer "knative.dev/pkg/client/injection/kube/informers/apps/v1/deployment"
+
+	apiserversourceinformer "knative.dev/eventing/pkg/client/injection/informers/sources/v1alpha2/apiserversource"
+	apiserversourcereconciler "knative.dev/eventing/pkg/client/injection/reconciler/sources/v1alpha2/apiserversource"
 )
 
 // envConfig will be used to extract the required environment variables using
@@ -79,6 +81,7 @@ func NewController(
 
 	cmw.Watch(logging.ConfigMapName(), r.UpdateFromLoggingConfigMap)
 	cmw.Watch(metrics.ConfigMapName(), r.UpdateFromMetricsConfigMap)
+	cmw.Watch(tracingconfig.ConfigName, r.UpdateFromTracingConfigMap)
 
 	return impl
 }

pkg/reconciler/apiserversource/controller_test.go

@@ -26,10 +26,11 @@ import (
 	"knative.dev/pkg/configmap"
 	. "knative.dev/pkg/reconciler/testing"
 
-	// Fake injection informers
-	_ "knative.dev/eventing/pkg/client/injection/informers/sources/v1alpha2/apiserversource/fake"
 	"knative.dev/pkg/client/injection/ducks/duck/v1/addressable"
 	_ "knative.dev/pkg/client/injection/kube/informers/apps/v1/deployment/fake"
+
+	// Fake injection informers
+	_ "knative.dev/eventing/pkg/client/injection/informers/sources/v1alpha2/apiserversource/fake"
 )
 
 func TestNew(t *testing.T) {
@@ -56,6 +57,14 @@ func TestNew(t *testing.T) {
 			"loglevel.controller": "info",
 			"loglevel.webhook":    "info",
 		},
+	}, &corev1.ConfigMap{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      "config-tracing",
+			Namespace: "knative-eventing",
+		},
+		Data: map[string]string{
+			"_example": "test-config",
+		},
 	}))
 
 	if c == nil {

pkg/reconciler/apiserversource/resources/receive_adapter.go

@@ -26,10 +26,11 @@ import (
 	v1 "k8s.io/api/apps/v1"
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
-	"knative.dev/eventing/pkg/adapter/apiserver"
-	"knative.dev/eventing/pkg/apis/sources/v1alpha2"
 	"knative.dev/pkg/kmeta"
 	"knative.dev/pkg/system"
+
+	"knative.dev/eventing/pkg/adapter/apiserver"
+	"knative.dev/eventing/pkg/apis/sources/v1alpha2"
 )
 
 // ReceiveAdapterArgs are the arguments needed to create a ApiServer Receive Adapter.
@@ -41,6 +42,7 @@ type ReceiveAdapterArgs struct {
 	SinkURI       string
 	MetricsConfig string
 	LoggingConfig string
+	TracingConfig string
 }
 
 // MakeReceiveAdapter generates (but does not insert into K8s) the Receive Adapter Deployment for
@@ -151,5 +153,9 @@ func makeEnv(args *ReceiveAdapterArgs) ([]corev1.EnvVar, error) {
 	}, {
 		Name:  "K_LOGGING_CONFIG",
 		Value: args.LoggingConfig,
-	}}, nil
+	}, {
+		Name:  "K_TRACING_CONFIG",
+		Value: args.TracingConfig,
+	},
+	}, nil
 }

pkg/reconciler/apiserversource/resources/receive_adapter_test.go

@@ -24,9 +24,10 @@ import (
 	v1 "k8s.io/api/apps/v1"
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
-	"knative.dev/eventing/pkg/apis/sources/v1alpha2"
 	"knative.dev/pkg/kmeta"
 
+	"knative.dev/eventing/pkg/apis/sources/v1alpha2"
+
 	_ "knative.dev/pkg/metrics/testing"
 	_ "knative.dev/pkg/system/testing"
 )
@@ -152,6 +153,10 @@ func TestMakeReceiveAdapter(t *testing.T) {
 									Name:  "K_LOGGING_CONFIG",
 									Value: "",
 								},
+								{
+									Name:  "K_TRACING_CONFIG",
+									Value: "",
+								},
 							},
 						},
 					},