Wire in Logging and Metrics config maps to PullSubscription RA

cmd/pubsub/receive_adapter/main.go

@@ -18,6 +18,7 @@ package main
 
 import (
 	"flag"
+	"fmt"
 	"log"
 	"time"
 
@@ -28,6 +29,7 @@ import (
 	"github.com/cloudevents/sdk-go/pkg/cloudevents/datacodec/xml"
 	transporthttp "github.com/cloudevents/sdk-go/pkg/cloudevents/transport/http"
 	"github.com/google/knative-gcp/pkg/pubsub/adapter"
+	"github.com/google/knative-gcp/pkg/reconciler/pullsubscription/resources"
 	"github.com/kelseyhightower/envconfig"
 	"go.opencensus.io/stats/view"
 	"go.uber.org/zap"
@@ -43,18 +45,33 @@ const (
 func main() {
 	flag.Parse()
 
-	sl, _ := logging.NewLogger("", "INFO") // TODO: use logging config map.
-	logger := sl.Desugar()
-	defer logger.Sync()
-
-	ctx := logging.WithLogger(signals.NewContext(), logger.Sugar())
-
 	startable := adapter.Adapter{}
 	if err := envconfig.Process("", &startable); err != nil {
-		logger.Fatal("Failed to process env var", zap.Error(err))
+		panic(fmt.Sprintf("Failed to process env var: %s", err))
+	}
+
+	// Convert base64 encoded json logging.Config to logging.Config.
+	loggingConfig, err := resources.Base64ToLoggingConfig(startable.LoggingConfigBase64)
+	if err != nil {
+		fmt.Printf("[ERROR] filed to process logging config: %s", err.Error())
+		// Use default logging config.
+		if loggingConfig, err = logging.NewConfigFromMap(map[string]string{}); err != nil {
+			// If this fails, there is no recovering.
+			panic(err)
+		}
 	}
 
-	mainMetrics(sl)
+	logger, _ := logging.NewLoggerFromConfig(loggingConfig, component)
+	defer flush(logger)
+	ctx := logging.WithLogger(signals.NewContext(), logger)
+
+	// Convert base64 encoded json metrics.ExporterOptions to metrics.ExporterOptions.
+	metricsConfig, err := resources.Base64ToMetricsOptions(startable.MetricsConfigBase64)
+	if err != nil {
+		logger.Errorf("failed to process metrics options: %s", err.Error())
+	}
+
+	mainMetrics(logger, metricsConfig)
 
 	if startable.Project == "" {
 		project, err := metadata.ProjectID()
@@ -72,17 +89,13 @@ func main() {
 	}
 }
 
-func mainMetrics(logger *zap.SugaredLogger) {
-	cfg := map[string]string{
-		"metrics.backend-destination": "prometheus", // TODO: hard code for now while we test.
+func mainMetrics(logger *zap.SugaredLogger, opts *metrics.ExporterOptions) {
+	if opts == nil {
+		logger.Info("metrics disabled")
+		return
 	}
 
-	if err := metrics.UpdateExporter(
-		metrics.ExporterOptions{
-			Domain:    metrics.Domain(),
-			Component: component,
-			ConfigMap: cfg,
-		}, logger); err != nil {
+	if err := metrics.UpdateExporter(*opts, logger); err != nil {
 		log.Fatalf("Failed to create the metrics exporter: %v", err)
 	}
 
@@ -93,9 +106,15 @@ func mainMetrics(logger *zap.SugaredLogger) {
 		json.LatencyView,
 		xml.LatencyView,
 		datacodec.LatencyView,
+		adapter.LatencyView,
 	); err != nil {
 		log.Fatalf("Failed to register views: %v", err)
 	}
 
 	view.SetReportingPeriod(2 * time.Second)
 }
+
+func flush(logger *zap.SugaredLogger) {
+	_ = logger.Sync()
+	metrics.FlushExporter()
+}

pkg/pubsub/adapter/adapter.go

@@ -19,10 +19,8 @@ package adapter
 import (
 	"context"
 	"fmt"
-
-	"github.com/cloudevents/sdk-go/pkg/cloudevents/observability"
-
 	cloudevents "github.com/cloudevents/sdk-go"
+	"github.com/cloudevents/sdk-go/pkg/cloudevents/observability"
 	"github.com/cloudevents/sdk-go/pkg/cloudevents/transport"
 	"github.com/cloudevents/sdk-go/pkg/cloudevents/transport/http"
 	cepubsub "github.com/cloudevents/sdk-go/pkg/cloudevents/transport/pubsub"
@@ -31,7 +29,7 @@ import (
 
 	"github.com/google/knative-gcp/pkg/kncloudevents"
 	"github.com/google/knative-gcp/pkg/pubsub/adapter/converters"
-	"github.com/google/knative-gcp/pkg/reconciler/decorator/resources"
+	decoratorresources "github.com/google/knative-gcp/pkg/reconciler/decorator/resources"
 )
 
 // Adapter implements the Pub/Sub adapter to deliver Pub/Sub messages from a
@@ -55,10 +53,10 @@ type Adapter struct {
 	// subscription to use.
 	Subscription string `envconfig:"PUBSUB_SUBSCRIPTION_ID" required:"true"`
 
-	// ExtensionsBased64 is a based64 encoded json string of a map of
+	// ExtensionsBase64 is a based64 encoded json string of a map of
 	// CloudEvents extensions (key-value pairs) override onto the outbound
 	// event.
-	ExtensionsBased64 string `envconfig:"K_CE_EXTENSIONS" required:"true"`
+	ExtensionsBase64 string `envconfig:"K_CE_EXTENSIONS" required:"true"`
 
 	// extensions is the converted ExtensionsBased64 value.
 	extensions map[string]string
@@ -67,8 +65,15 @@ type Adapter struct {
 	// One of [binary, structured, push]. Default: binary
 	SendMode converters.ModeType `envconfig:"SEND_MODE" default:"binary" required:"true"`
 
-	// MetricsDomain holds the metrics domain to use for surfacing metrics.
-	MetricsDomain string `envconfig:"METRICS_DOMAIN" required:"true"`
+	// MetricsConfigBase64 is a base64 encoded json string of metrics.ExporterOptions.
+	// This is used to configure the metrics exporter options, the config is
+	// stored in a config map inside the controllers namespace and copied here.
+	MetricsConfigBase64 string `envconfig:"K_METRICS_CONFIG" required:"true"`
+
+	// LoggingConfigBase64 is a base64 encoded json string of logging.Config.
+	// This is used to configure the logging config, the config is stored in
+	// a config map inside the controllers namespace and copied here.
+	LoggingConfigBase64 string `envconfig:"K_LOGGING_CONFIG" required:"true"`
 
 	// inbound is the cloudevents client to use to receive events.
 	inbound cloudevents.Client
@@ -90,7 +95,7 @@ func (a *Adapter) Start(ctx context.Context) error {
 
 	// Convert base64 encoded json map to extensions map.
 	// This implementation comes from the Decorator object.
-	a.extensions = resources.MakeDecoratorExtensionsMap(a.ExtensionsBased64)
+	a.extensions = decoratorresources.MakeDecoratorExtensionsMap(a.ExtensionsBase64)
 
 	// Receive Events on Pub/Sub.
 	if a.inbound == nil {

pkg/reconciler/pullsubscription/controller.go

@@ -18,6 +18,7 @@ package pullsubscription
 
 import (
 	"context"
+	"knative.dev/pkg/metrics"
 
 	"github.com/kelseyhightower/envconfig"
 	"go.uber.org/zap"
@@ -97,5 +98,8 @@ func NewController(
 
 	c.tracker = tracker.New(impl.EnqueueKey, controller.GetTrackerLease(ctx))
 
+	cmw.Watch(logging.ConfigMapName(), c.UpdateFromLoggingConfigMap)
+	cmw.Watch(metrics.ConfigMapName(), c.UpdateFromMetricsConfigMap)
+
 	return impl
 }

pkg/reconciler/pullsubscription/controller_test.go

@@ -17,18 +17,26 @@ limitations under the License.
 package pullsubscription
 
 import (
+	"knative.dev/pkg/logging"
+	"knative.dev/pkg/metrics"
 	"os"
 	"testing"
 
+	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+
 	"knative.dev/pkg/configmap"
 	logtesting "knative.dev/pkg/logging/testing"
 	. "knative.dev/pkg/reconciler/testing"
+	"knative.dev/pkg/system"
 
-	// Fake injection informers
+	_ "knative.dev/pkg/metrics/testing"
 
-	_ "github.com/google/knative-gcp/pkg/client/injection/informers/pubsub/v1alpha1/pullsubscription/fake"
+	// Fake injection informers
 	_ "knative.dev/pkg/injection/informers/kubeinformers/appsv1/deployment/fake"
 	_ "knative.dev/pkg/injection/informers/kubeinformers/batchv1/job/fake"
+
+	_ "github.com/google/knative-gcp/pkg/client/injection/informers/pubsub/v1alpha1/pullsubscription/fake"
 )
 
 func TestNew(t *testing.T) {
@@ -38,7 +46,22 @@ func TestNew(t *testing.T) {
 	_ = os.Setenv("PUBSUB_RA_IMAGE", "PUBSUB_RA_IMAGE")
 	_ = os.Setenv("PUBSUB_SUB_IMAGE", "PUBSUB_SUB_IMAGE")
 
-	c := NewController(ctx, configmap.NewFixedWatcher())
+	c := NewController(ctx, configmap.NewStaticWatcher(
+		&corev1.ConfigMap{
+			ObjectMeta: metav1.ObjectMeta{
+				Name:      logging.ConfigMapName(),
+				Namespace: system.Namespace(),
+			},
+			Data: map[string]string{},
+		},
+		&corev1.ConfigMap{
+			ObjectMeta: metav1.ObjectMeta{
+				Name:      metrics.ConfigMapName(),
+				Namespace: system.Namespace(),
+			},
+			Data: map[string]string{},
+		},
+	))
 
 	if c == nil {
 		t.Fatal("Expected NewController to return a non-nil value")

pkg/reconciler/pullsubscription/pullsubscription.go

@@ -19,6 +19,7 @@ package pullsubscription
 import (
 	"context"
 	"encoding/json"
+	"knative.dev/pkg/metrics"
 	"time"
 
 	appsv1 "k8s.io/api/apps/v1"
@@ -54,6 +55,8 @@ const (
 	// ReconcilerName is the name of the reconciler
 	ReconcilerName = "PullSubscriptions"
 
+	component = "pullsubscriptions"
+
 	finalizerName = controllerAgentName
 )
 
@@ -70,6 +73,9 @@ type Reconciler struct {
 
 	receiveAdapterImage string
 
+	loggingConfig *logging.Config
+	metricsConfig *metrics.ExporterOptions
+
 	//	eventTypeReconciler eventtype.Reconciler // TODO: event types.
 
 }
@@ -408,6 +414,8 @@ func (r *Reconciler) createOrUpdateReceiveAdapter(ctx context.Context, src *v1al
 		SubscriptionID: src.Status.SubscriptionID,
 		SinkURI:        src.Status.SinkURI,
 		TransformerURI: src.Status.TransformerURI,
+		LoggingConfig:  resources.LoggingConfigToBase64(r.loggingConfig),
+		MetricsConfig:  resources.MetricsOptionsToBase64(r.metricsConfig),
 	})
 
 	if existing == nil {
@@ -447,6 +455,35 @@ func (r *Reconciler) getReceiveAdapter(ctx context.Context, src *v1alpha1.PullSu
 	return nil, apierrors.NewNotFound(schema.GroupResource{}, "")
 }
 
+func (r *Reconciler) UpdateFromLoggingConfigMap(cfg *corev1.ConfigMap) {
+	if cfg != nil {
+		delete(cfg.Data, "_example")
+	}
+
+	logcfg, err := logging.NewConfigFromConfigMap(cfg)
+	if err != nil {
+		r.Logger.Warn("failed to create logging config from configmap", zap.String("cfg.Name", cfg.Name))
+		return
+	}
+	r.loggingConfig = logcfg
+	r.Logger.Infow("Update from logging ConfigMap", zap.Any("ConfigMap", cfg))
+	// TODO: requeue all pullsubscriptions
+}
+
+func (r *Reconciler) UpdateFromMetricsConfigMap(cfg *corev1.ConfigMap) {
+	if cfg != nil {
+		delete(cfg.Data, "_example")
+	}
+
+	r.metricsConfig = &metrics.ExporterOptions{
+		Domain:    metrics.Domain(),
+		Component: component,
+		ConfigMap: cfg.Data,
+	}
+	r.Logger.Infow("Update from metrics ConfigMap", zap.Any("ConfigMap", cfg))
+	// TODO: requeue all pullsubscriptions
+}
+
 // TODO: Registry
 //func (r *Reconciler) reconcileEventTypes(ctx context.Context, src *v1alpha1.PullSubscription) error {
 //	args := r.newEventTypeReconcilerArgs(src)