Adding data plane metrics to ApiServerSource

cmd/apiserver_receive_adapter/main.go

@@ -18,25 +18,31 @@ package main
 
 import (
 	"flag"
-	"log"
+	"fmt"
 	"strings"
 
-	// Uncomment the following line to load the gcp plugin (only required to authenticate against GKE clusters).
+	// 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"
-
 	"github.com/kelseyhightower/envconfig"
 	"go.uber.org/zap"
-	"go.uber.org/zap/zapcore"
 	"k8s.io/apimachinery/pkg/api/meta"
 	"k8s.io/apimachinery/pkg/runtime/schema"
 	"k8s.io/client-go/dynamic"
 	"k8s.io/client-go/tools/clientcmd"
 	"knative.dev/eventing/pkg/adapter/apiserver"
 	"knative.dev/eventing/pkg/kncloudevents"
 	"knative.dev/eventing/pkg/tracing"
+	"knative.dev/eventing/pkg/utils"
+	"knative.dev/pkg/logging"
+	"knative.dev/pkg/metrics"
 	"knative.dev/pkg/signals"
 )
 
+const (
+	component = "apiserversource"
+)
+
 var (
 	masterURL  = flag.String("master", "", "The address of the Kubernetes API server. Overrides any value in kubeconfig. Only required if out-of-cluster.")
 	kubeconfig = flag.String("kubeconfig", "", "Path to a kubeconfig. Only required if out-of-cluster.")
@@ -60,52 +66,84 @@ type envConfig struct {
 	Kind          StringList `required:"true"`
 	Controller    []bool     `required:"true"`
 	LabelSelector StringList `envconfig:"SELECTOR" required:"true"`
+	Name          string     `envconfig:"NAME" 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"`
 }
 
 // TODO: the controller should take the list of GVR
 
 func main() {
 	flag.Parse()
 
-	logCfg := zap.NewProductionConfig()
-	logCfg.EncoderConfig.EncodeTime = zapcore.ISO8601TimeEncoder
-	dlogger, err := logCfg.Build()
+	var env envConfig
+	err := envconfig.Process("", &env)
 	if err != nil {
-		log.Fatalf("Error building logger: %v", err)
+		panic(fmt.Sprintf("Error processing env var: %s", err))
+	}
+	// TODO move this util to pkg
+	// Convert base64 encoded json logging.Config to logging.Config.
+	loggingConfig, err := utils.Base64ToLoggingConfig(env.LoggingConfigBase64)
+	if err != nil {
+		fmt.Printf("[ERROR] failed 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)
+		}
+	}
+	loggerSugared, _ := logging.NewLoggerFromConfig(loggingConfig, component)
+	logger := loggerSugared.Desugar()
+	defer flush(loggerSugared)
+
+	// Convert base64 encoded json metrics.ExporterOptions to
+	// metrics.ExporterOptions.
+	metricsConfig, err := utils.Base64ToMetricsOptions(
+		env.MetricsConfigBase64)
+	if err != nil {
+		logger.Error("failed to process metrics options ", zap.Error(err))
 	}
-	logger := dlogger.Sugar()
 
-	var env envConfig
-	err = envconfig.Process("", &env)
+	if err := metrics.UpdateExporter(*metricsConfig, loggerSugared); err != nil {
+		logger.Error("failed to create the metrics exporter ", zap.Error(err))
+	}
+
+	reporter, err := apiserver.NewStatsReporter()
 	if err != nil {
-		logger.Fatalw("Error processing environment", zap.Error(err))
+		logger.Error("error building statsreporter", zap.Error(err))
 	}
 
 	// set up signals so we handle the first shutdown signal gracefully
 	stopCh := signals.SetupSignalHandler()
 
 	cfg, err := clientcmd.BuildConfigFromFlags(*masterURL, *kubeconfig)
 	if err != nil {
-		logger.Fatalw("Error building kubeconfig", zap.Error(err))
+		logger.Fatal("error building kubeconfig", zap.Error(err))
 	}
 
-	logger = logger.With(zap.String("controller/apiserver", "adapter"))
 	logger.Info("Starting the controller")
-
 	client, err := dynamic.NewForConfig(cfg)
 	if err != nil {
-		logger.Fatalw("Error building dynamic client", zap.Error(err))
+		logger.Fatal("error building dynamic client", zap.Error(err))
 	}
 
-	if err = tracing.SetupStaticPublishing(logger, "apiserversource", tracing.OnePercentSampling); err != nil {
-		// If tracing doesn't work, we will log an error, but allow the importer to continue to
-		// start.
+	if err = tracing.SetupStaticPublishing(loggerSugared, "apiserversource", tracing.OnePercentSampling); err != nil {
+		// If tracing doesn't work, we will log an error, but allow the importer
+		// to continue to start.
 		logger.Error("Error setting up trace publishing", zap.Error(err))
 	}
 
 	eventsClient, err := kncloudevents.NewDefaultClient(env.SinkURI)
 	if err != nil {
-		logger.Fatalw("Error building cloud event client", zap.Error(err))
+		logger.Fatal("error building cloud event client", zap.Error(err))
 	}
 
 	gvrcs := []apiserver.GVRC(nil)
@@ -117,7 +155,7 @@ func main() {
 
 		gv, err := schema.ParseGroupVersion(apiVersion)
 		if err != nil {
-			logger.Fatalw("Error parsing APIVersion", zap.Error(err))
+			logger.Fatal("error parsing APIVersion", zap.Error(err))
 		}
 		// TODO: pass down the resource and the kind so we do not have to guess.
 		gvr, _ := meta.UnsafeGuessKindToResource(schema.GroupVersionKind{Kind: kind, Group: gv.Group, Version: gv.Version})
@@ -134,9 +172,14 @@ func main() {
 		GVRCs:     gvrcs,
 	}
 
-	a := apiserver.NewAdaptor(cfg.Host, client, eventsClient, logger, opt)
-	logger.Info("starting kubernetes api adapter")
+	a := apiserver.NewAdaptor(cfg.Host, client, eventsClient, loggerSugared, opt, reporter, env.Name)
+	logger.Info("starting kubernetes api adapter.", zap.Any("adapter", env))
 	if err := a.Start(stopCh); err != nil {
 		logger.Warn("start returned an error,", zap.Error(err))
 	}
 }
+
+func flush(logger *zap.SugaredLogger) {
+	_ = logger.Sync()
+	metrics.FlushExporter()
+}

config/monitoring/metrics/prometheus/100-prometheus-scrape-kn-eventing.yaml

@@ -109,3 +109,21 @@
     - source_labels: [__meta_kubernetes_service_name]
       target_label: service
 
+# apiserver-source
+- job_name: apiserver-source
+  scrape_interval: 3s
+  scrape_timeout: 3s
+  kubernetes_sd_configs:
+    - role: pod
+  relabel_configs:
+    # Scrape only the the targets matching the following metadata
+    - source_labels: [ __meta_kubernetes_pod_label_eventing_knative_dev_source, __meta_kubernetes_pod_container_port_name]
+      action: keep
+      regex: apiserver-source-controller;metrics
+    # Rename metadata labels to be reader friendly
+    - source_labels: [__meta_kubernetes_namespace]
+      target_label: namespace
+    - source_labels: [__meta_kubernetes_pod_name]
+      target_label: pod
+    - source_labels: [__meta_kubernetes_service_name]
+      target_label: service

pkg/adapter/apiserver/adapter.go

@@ -67,9 +67,13 @@ type adapter struct {
 
 	mode     string
 	delegate eventDelegate
+	reporter StatsReporter
+	name     string
 }
 
-func NewAdaptor(source string, k8sClient dynamic.Interface, ceClient cloudevents.Client, logger *zap.SugaredLogger, opt Options) Adapter {
+func NewAdaptor(source string, k8sClient dynamic.Interface,
+	ceClient cloudevents.Client, logger *zap.SugaredLogger,
+	opt Options, reporter StatsReporter, name string) Adapter {
 	mode := opt.Mode
 	switch mode {
 	case ResourceMode, RefMode:
@@ -88,6 +92,8 @@ func NewAdaptor(source string, k8sClient dynamic.Interface, ceClient cloudevents
 		gvrcs:     opt.GVRCs,
 		namespace: opt.Namespace,
 		mode:      mode,
+		reporter:  reporter,
+		name:      name,
 	}
 	return a
 }
@@ -102,21 +108,26 @@ func (a *adapter) Start(stopCh <-chan struct{}) error {
 	stop := make(chan struct{})
 
 	resyncPeriod := time.Duration(10 * time.Hour)
-
 	var d eventDelegate
 	switch a.mode {
 	case ResourceMode:
 		d = &resource{
-			ce:     a.ce,
-			source: a.source,
-			logger: a.logger,
+			ce:        a.ce,
+			source:    a.source,
+			logger:    a.logger,
+			reporter:  a.reporter,
+			namespace: a.namespace,
+			name:      a.name,
 		}
 
 	case RefMode:
 		d = &ref{
-			ce:     a.ce,
-			source: a.source,
-			logger: a.logger,
+			ce:        a.ce,
+			source:    a.source,
+			logger:    a.logger,
+			reporter:  a.reporter,
+			namespace: a.namespace,
+			name:      a.name,
 		}
 
 	default:

pkg/adapter/apiserver/adapter_test.go

@@ -32,6 +32,12 @@ import (
 	rectesting "knative.dev/eventing/pkg/reconciler/testing"
 )
 
+type mockReporter struct{}
+
+func (r *mockReporter) ReportEventCount(args *ReportArgs, responseCode int) error {
+	return nil
+}
+
 func TestNewAdaptor(t *testing.T) {
 	ce := kncetesting.NewTestClient()
 	logger := zap.NewExample().Sugar()
@@ -132,8 +138,8 @@ func TestNewAdaptor(t *testing.T) {
 	}
 	for n, tc := range testCases {
 		t.Run(n, func(t *testing.T) {
-
-			a := NewAdaptor(tc.source, k8s, ce, logger, tc.opt)
+			r := &mockReporter{}
+			a := NewAdaptor(tc.source, k8s, ce, logger, tc.opt, r, "test-importer")
 
 			got, ok := a.(*adapter)
 			if !ok {
@@ -171,8 +177,8 @@ func TestAdapter_StartRef(t *testing.T) {
 			},
 		}},
 	}
-
-	a := NewAdaptor(source, k8s, ce, logger, opt)
+	r := &mockReporter{}
+	a := NewAdaptor(source, k8s, ce, logger, opt, r, "test-importer")
 
 	err := errors.New("test never ran")
 	stopCh := make(chan struct{})
@@ -206,8 +212,8 @@ func TestAdapter_StartResource(t *testing.T) {
 			},
 		}},
 	}
-
-	a := NewAdaptor(source, k8s, ce, logger, opt)
+	r := &mockReporter{}
+	a := NewAdaptor(source, k8s, ce, logger, opt, r, "test-importer")
 
 	err := errors.New("test never ran")
 	stopCh := make(chan struct{})
@@ -292,22 +298,24 @@ func makeResourceAndTestingClient() (*resource, *kncetesting.TestCloudEventsClie
 	ce := kncetesting.NewTestClient()
 	source := "unit-test"
 	logger := zap.NewExample().Sugar()
-
+	r := &mockReporter{}
 	return &resource{
-		ce:     ce,
-		source: source,
-		logger: logger,
+		ce:       ce,
+		source:   source,
+		logger:   logger,
+		reporter: r,
 	}, ce
 }
 
 func makeRefAndTestingClient() (*ref, *kncetesting.TestCloudEventsClient) {
 	ce := kncetesting.NewTestClient()
 	source := "unit-test"
 	logger := zap.NewExample().Sugar()
-
+	r := &mockReporter{}
 	return &ref{
-		ce:     ce,
-		source: source,
-		logger: logger,
+		ce:       ce,
+		source:   source,
+		logger:   logger,
+		reporter: r,
 	}, ce
 }