ApiServerSource updates

cmd/apiserver_receive_adapter/main.go

@@ -36,6 +36,7 @@ import (
 	"knative.dev/pkg/logging"
 	"knative.dev/pkg/metrics"
 	"knative.dev/pkg/signals"
+	"knative.dev/pkg/source"
 )
 
 const (
@@ -112,7 +113,7 @@ func main() {
 		logger.Error("failed to create the metrics exporter", zap.Error(err))
 	}
 
-	reporter, err := metrics.NewStatsReporter()
+	reporter, err := source.NewStatsReporter()
 	if err != nil {
 		logger.Error("error building statsreporter", zap.Error(err))
 	}

pkg/adapter/apiserver/adapter.go

@@ -20,17 +20,16 @@ import (
 	"fmt"
 	"time"
 
+	cloudevents "github.com/cloudevents/sdk-go"
+	"go.uber.org/zap"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/runtime/schema"
 	"k8s.io/apimachinery/pkg/watch"
 	"k8s.io/client-go/dynamic"
 	"k8s.io/client-go/tools/cache"
-
-	cloudevents "github.com/cloudevents/sdk-go"
-	"go.uber.org/zap"
-	"knative.dev/pkg/metrics"
+	"knative.dev/pkg/source"
 )
 
 type Adapter interface {
@@ -70,13 +69,13 @@ type adapter struct {
 
 	mode     string
 	delegate eventDelegate
-	reporter metrics.StatsReporter
+	reporter source.StatsReporter
 	name     string
 }
 
 func NewAdaptor(source string, k8sClient dynamic.Interface,
 	ceClient cloudevents.Client, logger *zap.SugaredLogger,
-	opt Options, reporter metrics.StatsReporter, name string) Adapter {
+	opt Options, reporter source.StatsReporter, name string) Adapter {
 	mode := opt.Mode
 	switch mode {
 	case ResourceMode, RefMode:

pkg/adapter/apiserver/adapter_test.go

@@ -30,14 +30,14 @@ import (
 	dynamicfake "k8s.io/client-go/dynamic/fake"
 	kncetesting "knative.dev/eventing/pkg/kncloudevents/testing"
 	rectesting "knative.dev/eventing/pkg/reconciler/testing"
-	"knative.dev/pkg/metrics"
+	"knative.dev/pkg/source"
 )
 
 type mockReporter struct {
 	eventCount int
 }
 
-func (r *mockReporter) ReportEventCount(args *metrics.ReportArgs, responseCode int) error {
+func (r *mockReporter) ReportEventCount(args *source.ReportArgs, responseCode int) error {
 	r.eventCount += 1
 	return nil
 }
@@ -324,7 +324,7 @@ func makeRefAndTestingClient() (*ref, *kncetesting.TestCloudEventsClient) {
 	}, ce
 }
 
-func validateMetric(t *testing.T, reporter metrics.StatsReporter, want int) {
+func validateMetric(t *testing.T, reporter source.StatsReporter, want int) {
 	if mockReporter, ok := reporter.(*mockReporter); !ok {
 		t.Errorf("reporter is not a mockReporter")
 	} else if mockReporter.eventCount != want {

pkg/adapter/apiserver/ref.go

@@ -18,17 +18,16 @@ package apiserver
 
 import (
 	"context"
-	"knative.dev/pkg/metrics"
 	"reflect"
 
+	cloudevents "github.com/cloudevents/sdk-go"
+	"go.uber.org/zap"
 	"k8s.io/apimachinery/pkg/api/meta"
 	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
 	"k8s.io/apimachinery/pkg/runtime/schema"
 	"k8s.io/client-go/tools/cache"
-
-	cloudevents "github.com/cloudevents/sdk-go"
-	"go.uber.org/zap"
 	"knative.dev/eventing/pkg/adapter/apiserver/events"
+	"knative.dev/pkg/source"
 )
 
 type ref struct {
@@ -38,7 +37,7 @@ type ref struct {
 	logger    *zap.SugaredLogger
 
 	controlledGVRs []schema.GroupVersionResource
-	reporter       metrics.StatsReporter
+	reporter       source.StatsReporter
 	namespace      string
 	name           string
 }
@@ -106,7 +105,7 @@ func (a *ref) addControllerWatch(gvr schema.GroupVersionResource) {
 }
 
 func (a *ref) sendEvent(ctx context.Context, event *cloudevents.Event) error {
-	reportArgs := &metrics.ReportArgs{
+	reportArgs := &source.ReportArgs{
 		Namespace:     a.namespace,
 		EventSource:   event.Source(),
 		EventType:     event.Type(),

pkg/adapter/apiserver/resource.go

@@ -24,15 +24,15 @@ import (
 	"k8s.io/apimachinery/pkg/runtime/schema"
 	"k8s.io/client-go/tools/cache"
 	"knative.dev/eventing/pkg/adapter/apiserver/events"
-	"knative.dev/pkg/metrics"
+	"knative.dev/pkg/source"
 )
 
 type resource struct {
 	ce        cloudevents.Client
 	source    string
 	eventType string
 	logger    *zap.SugaredLogger
-	reporter  metrics.StatsReporter
+	reporter  source.StatsReporter
 	namespace string
 	name      string
 }
@@ -72,7 +72,7 @@ func (a *resource) Delete(obj interface{}) error {
 }
 
 func (a *resource) sendEvent(ctx context.Context, event *cloudevents.Event) error {
-	reportArgs := &metrics.ReportArgs{
+	reportArgs := &source.ReportArgs{
 		Namespace:     a.namespace,
 		EventSource:   event.Source(),
 		EventType:     event.Type(),

vendor/knative.dev/pkg/metrics/metricskey/constants_eventing.go

@@ -27,8 +27,8 @@ const (
 	// ResourceTypeKnativeBroker is the Stackdriver resource type for Knative Brokers.
 	ResourceTypeKnativeBroker = "knative_broker"
 
-	// ResourceTypeKnativeImporter is the Stackdriver resource type for Knative Importers.
-	ResourceTypeKnativeImporter = "knative_importer"
+	// ResourceTypeKnativeSource is the Stackdriver resource type for Knative Sources.
+	ResourceTypeKnativeSource = "knative_source"
 
 	// LabelTriggerName is the label for the name of the Trigger.
 	LabelTriggerName = "trigger_name"
@@ -48,11 +48,11 @@ const (
 	// LabelFilterSource is the label for the Trigger filter attribute "source".
 	LabelFilterSource = "filter_source"
 
-	// LabelImporterName is the label for the name of the Importer.
-	LabelImporterName = "importer_name"
+	// LabelSourceName is the label for the name of the Source.
+	LabelSourceName = "source_name"
 
-	// LabelImporterResourceGroup is the name of the Importer CRD.
-	LabelImporterResourceGroup = "importer_resource_group"
+	// LabelSourceResourceGroup is the name of the Source CRD.
+	LabelSourceResourceGroup = "source_resource_group"
 )
 
 var (
@@ -89,19 +89,19 @@ var (
 		"knative.dev/eventing/broker/event_count",
 	)
 
-	// KnativeImporterLabels stores the set of resource labels for resource type knative_importer.
-	KnativeImporterLabels = sets.NewString(
+	// KnativeSourceLabels stores the set of resource labels for resource type knative_source.
+	KnativeSourceLabels = sets.NewString(
 		LabelProject,
 		LabelLocation,
 		LabelClusterName,
 		LabelNamespaceName,
-		LabelImporterName,
-		LabelImporterResourceGroup,
+		LabelSourceName,
+		LabelSourceResourceGroup,
 	)
 
-	// KnativeImporterMetrics stores a set of metric types which are supported
-	// by resource type knative_importer.
-	KnativeImporterMetrics = sets.NewString(
-		"knative.dev/eventing/importer/event_count",
+	// KnativeSourceMetrics stores a set of metric types which are supported
+	// by resource type knative_source.
+	KnativeSourceMetrics = sets.NewString(
+		"knative.dev/eventing/source/event_count",
 	)
 )

vendor/knative.dev/pkg/metrics/monitored_resources_eventing.go

@@ -44,13 +44,13 @@ type KnativeBroker struct {
 	BrokerName    string
 }
 
-type KnativeImporter struct {
-	Project               string
-	Location              string
-	ClusterName           string
-	NamespaceName         string
-	ImporterName          string
-	ImporterResourceGroup string
+type KnativeSource struct {
+	Project             string
+	Location            string
+	ClusterName         string
+	NamespaceName       string
+	SourceName          string
+	SourceResourceGroup string
 }
 
 func (kt *KnativeTrigger) MonitoredResource() (resType string, labels map[string]string) {
@@ -76,16 +76,16 @@ func (kb *KnativeBroker) MonitoredResource() (resType string, labels map[string]
 	return metricskey.ResourceTypeKnativeBroker, labels
 }
 
-func (ki *KnativeImporter) MonitoredResource() (resType string, labels map[string]string) {
+func (ki *KnativeSource) MonitoredResource() (resType string, labels map[string]string) {
 	labels = map[string]string{
-		metricskey.LabelProject:               ki.Project,
-		metricskey.LabelLocation:              ki.Location,
-		metricskey.LabelClusterName:           ki.ClusterName,
-		metricskey.LabelNamespaceName:         ki.NamespaceName,
-		metricskey.LabelImporterName:          ki.ImporterName,
-		metricskey.LabelImporterResourceGroup: ki.ImporterResourceGroup,
+		metricskey.LabelProject:             ki.Project,
+		metricskey.LabelLocation:            ki.Location,
+		metricskey.LabelClusterName:         ki.ClusterName,
+		metricskey.LabelNamespaceName:       ki.NamespaceName,
+		metricskey.LabelSourceName:          ki.SourceName,
+		metricskey.LabelSourceResourceGroup: ki.SourceResourceGroup,
 	}
-	return metricskey.ResourceTypeKnativeImporter, labels
+	return metricskey.ResourceTypeKnativeSource, labels
 }
 
 func GetKnativeBrokerMonitoredResource(
@@ -137,24 +137,24 @@ func GetKnativeTriggerMonitoredResource(
 	return newTags, kt
 }
 
-func GetKnativeImporterMonitoredResource(
+func GetKnativeSourceMonitoredResource(
 	v *view.View, tags []tag.Tag, gm *gcpMetadata) ([]tag.Tag, monitoredresource.Interface) {
 	tagsMap := getTagsMap(tags)
-	ki := &KnativeImporter{
+	ki := &KnativeSource{
 		// The first three resource labels are from metadata.
 		Project:     gm.project,
 		Location:    gm.location,
 		ClusterName: gm.cluster,
 		// The rest resource labels are from metrics labels.
-		NamespaceName:         valueOrUnknown(metricskey.LabelNamespaceName, tagsMap),
-		ImporterName:          valueOrUnknown(metricskey.LabelImporterName, tagsMap),
-		ImporterResourceGroup: valueOrUnknown(metricskey.LabelImporterResourceGroup, tagsMap),
+		NamespaceName:       valueOrUnknown(metricskey.LabelNamespaceName, tagsMap),
+		SourceName:          valueOrUnknown(metricskey.LabelSourceName, tagsMap),
+		SourceResourceGroup: valueOrUnknown(metricskey.LabelSourceResourceGroup, tagsMap),
 	}
 
 	var newTags []tag.Tag
 	for _, t := range tags {
 		// Keep the metrics labels that are not resource labels
-		if !metricskey.KnativeImporterLabels.Has(t.Key.Name()) {
+		if !metricskey.KnativeSourceLabels.Has(t.Key.Name()) {
 			newTags = append(newTags, t)
 		}
 	}

vendor/knative.dev/pkg/metrics/record.go

@@ -36,7 +36,7 @@ import (
 //   2) The backend is not Stackdriver.
 //   3) The backend is Stackdriver and it is allowed to use custom metrics.
 //   4) The backend is Stackdriver and the metric is one of the built-in metrics: "knative_revision", "knative_broker",
-//      "knative_trigger", "knative_importer".
+//      "knative_trigger", "knative_source".
 func Record(ctx context.Context, ms stats.Measurement) {
 	mc := getCurMetricsConfig()
 
@@ -57,7 +57,7 @@ func Record(ctx context.Context, ms stats.Measurement) {
 	isServingBuiltIn := metricskey.KnativeRevisionMetrics.Has(metricType)
 	isEventingBuiltIn := metricskey.KnativeTriggerMetrics.Has(metricType) ||
 		metricskey.KnativeBrokerMetrics.Has(metricType) ||
-		metricskey.KnativeImporterMetrics.Has(metricType)
+		metricskey.KnativeSourceMetrics.Has(metricType)
 
 	if isServingBuiltIn || isEventingBuiltIn {
 		stats.Record(ctx, ms)

vendor/knative.dev/pkg/metrics/stackdriver_exporter.go

@@ -87,10 +87,10 @@ func getMonitoredResourceFunc(metricTypePrefix string, gm *gcpMetadata) func(v *
 			return GetKnativeBrokerMonitoredResource(view, tags, gm)
 		} else if metricskey.KnativeTriggerMetrics.Has(metricType) {
 			return GetKnativeTriggerMonitoredResource(view, tags, gm)
-		} else if metricskey.KnativeImporterMetrics.Has(metricType) {
-			return GetKnativeImporterMonitoredResource(view, tags, gm)
+		} else if metricskey.KnativeSourceMetrics.Has(metricType) {
+			return GetKnativeSourceMonitoredResource(view, tags, gm)
 		}
-		// Unsupported metric by knative_revision, knative_broker, knative_trigger, and knative_importer, use "global" resource type.
+		// Unsupported metric by knative_revision, knative_broker, knative_trigger, and knative_source, use "global" resource type.
 		return getGlobalMonitoredResource(view, tags)
 	}
 }
@@ -105,7 +105,7 @@ func getMetricTypeFunc(metricTypePrefix, customMetricTypePrefix string) func(vie
 		inServing := metricskey.KnativeRevisionMetrics.Has(metricType)
 		inEventing := metricskey.KnativeBrokerMetrics.Has(metricType) ||
 			metricskey.KnativeTriggerMetrics.Has(metricType) ||
-			metricskey.KnativeImporterMetrics.Has(metricType)
+			metricskey.KnativeSourceMetrics.Has(metricType)
 		if inServing || inEventing {
 			return metricType
 		}