PullSubscription and Storage importer metrics

Gopkg.toml

@@ -21,7 +21,8 @@ required = [
 
 [[override]]
   name = "knative.dev/pkg"
-  branch = "master"
+  source = "github.com/nachocano/pkg"
+  branch = "source_stats"
 
 # This is a preemptive override.
 [[override]]
@@ -62,8 +63,7 @@ required = [
 
 [[override]]
   name = "github.com/cloudevents/sdk-go"
-  # HEAD as of 2019-06-04
-  revision = "6fe382645effbc5d4a7055f4280e9c3d7f58a2b6"
+  version = "0.9.0"
 
 [prune]
   go-tests = true

cmd/pubsub/receive_adapter/main.go

@@ -17,25 +17,16 @@ limitations under the License.
 package main
 
 import (
+	"cloud.google.com/go/compute/metadata"
 	"flag"
 	"fmt"
-	"log"
-	"time"
-
-	"cloud.google.com/go/compute/metadata"
-	"github.com/cloudevents/sdk-go/pkg/cloudevents/client"
-	"github.com/cloudevents/sdk-go/pkg/cloudevents/datacodec"
-	"github.com/cloudevents/sdk-go/pkg/cloudevents/datacodec/json"
-	"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"
 	"knative.dev/pkg/logging"
 	"knative.dev/pkg/metrics"
 	"knative.dev/pkg/signals"
+	"log"
 )
 
 const (
@@ -50,8 +41,8 @@ func main() {
 		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)
+	// Convert json logging.Config to logging.Config.
+	loggingConfig, err := logging.JsonToLoggingConfig(startable.LoggingConfigJson)
 	if err != nil {
 		fmt.Printf("[ERROR] filed to process logging config: %s", err.Error())
 		// Use default logging config.
@@ -65,8 +56,8 @@ func main() {
 	defer flush(logger)
 	ctx := logging.WithLogger(signals.NewContext(), logger)
 
-	// Convert base64 encoded json metrics.ExporterOptions to metrics.ExporterOptions.
-	metricsConfig, err := resources.Base64ToMetricsOptions(startable.MetricsConfigBase64)
+	// Convert json metrics.ExporterOptions to metrics.ExporterOptions.
+	metricsConfig, err := metrics.JsonToMetricsOptions(startable.MetricsConfigJson)
 	if err != nil {
 		logger.Errorf("failed to process metrics options: %s", err.Error())
 	}
@@ -99,19 +90,22 @@ func mainMetrics(logger *zap.SugaredLogger, opts *metrics.ExporterOptions) {
 		log.Fatalf("Failed to create the metrics exporter: %v", err)
 	}
 
-	// Register the views
-	if err := view.Register(
-		client.LatencyView,
-		transporthttp.LatencyView,
-		json.LatencyView,
-		xml.LatencyView,
-		datacodec.LatencyView,
-		adapter.LatencyView,
-	); err != nil {
-		log.Fatalf("Failed to register views: %v", err)
-	}
-
-	view.SetReportingPeriod(2 * time.Second)
+	// TODO metrics are API surface, so make sure we need to expose this before doing so.
+	//  They seem to be private ones and more profiling related ones.
+	//  Commenting them for now, as we will use pkg stats_reporter.
+	// Register the views.
+	//if err := view.Register(
+	//	client.LatencyView,
+	//	transporthttp.LatencyView,
+	//	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) {

cmd/webhook/main.go

@@ -20,11 +20,10 @@ import (
 	"context"
 	"flag"
 	"fmt"
-	"log"
-	"net/http"
-
 	"golang.org/x/sync/errgroup"
 	"knative.dev/pkg/profiling"
+	"log"
+	"net/http"
 
 	"go.uber.org/zap"
 	"k8s.io/apimachinery/pkg/runtime/schema"
@@ -103,21 +102,33 @@ func SharedMain(handlers map[schema.GroupVersionKind]webhook.GenericCRD) {
 		logger.Fatalw("Failed to start the ConfigMap watcher", zap.Error(err))
 	}
 
+	stats, err := webhook.NewStatsReporter()
+	if err != nil {
+		logger.Fatalw("failed to initialize the stats reporter", zap.Error(err))
+	}
+
 	options := webhook.ControllerOptions{
-		ServiceName:    "webhook",
-		DeploymentName: "webhook",
-		Namespace:      system.Namespace(),
-		Port:           8443,
-		SecretName:     "webhook-certs",
-		WebhookName:    fmt.Sprintf("webhook.%s.events.cloud.run", system.Namespace()),
+		ServiceName:                     "webhook",
+		DeploymentName:                  "webhook",
+		Namespace:                       system.Namespace(),
+		Port:                            8443,
+		SecretName:                      "webhook-certs",
+		WebhookName:                     fmt.Sprintf("webhook.%s.events.cloud.run", system.Namespace()),
+		StatsReporter:                   stats,
+		ResourceAdmissionControllerPath: "/",
 	}
 
 	// Decorate contexts with the current state of the config.
 	ctxFunc := func(ctx context.Context) context.Context {
 		return ctx
 	}
 
-	controller, err := webhook.NewAdmissionController(kubeClient, options, handlers, logger, ctxFunc, true)
+	resourceAdmissionController := webhook.NewResourceAdmissionController(handlers, options, true)
+	admissionControllers := map[string]webhook.AdmissionController{
+		options.ResourceAdmissionControllerPath: resourceAdmissionController,
+	}
+
+	controller, err := webhook.New(kubeClient, options, admissionControllers, logger, ctxFunc)
 
 	if err != nil {
 		logger.Fatalw("Failed to create admission controller", zap.Error(err))

config/500-controller.yaml

@@ -60,7 +60,7 @@ spec:
         - name: CONFIG_OBSERVABILITY_NAME
           value: config-observability
         - name: METRICS_DOMAIN
-          value: cloud.run/events
+          value: knative.dev/eventing
         volumeMounts:
         - name: config-logging
           mountPath: /etc/config-logging

config/500-webhook.yaml

@@ -59,6 +59,8 @@ spec:
                   fieldPath: metadata.namespace
             - name: CONFIG_LOGGING_NAME
               value: config-logging
+            - name: METRICS_DOMAIN
+              value: knative.dev/eventing
       volumes:
         - name: config-logging
           configMap:

config/config-observability.yaml

@@ -37,80 +37,19 @@ data:
     # this example block and unindented to be in the data block
     # to actually change the configuration.
 
-    # logging.enable-var-log-collection defaults to false.
-    # A fluentd sidecar will be set up to collect var log if
-    # this flag is true.
-    logging.enable-var-log-collection: false
-
-    # logging.fluentd-sidecar-image provides the fluentd sidecar image
-    # to inject as a sidecar to collect logs from /var/log.
-    # Must be presented if logging.enable-var-log-collection is true.
-    logging.fluentd-sidecar-image: k8s.gcr.io/fluentd-elasticsearch:v2.0.4
-
-    # logging.fluentd-sidecar-output-config provides the configuration
-    # for the fluentd sidecar, which will be placed into a configmap and
-    # mounted into the fluentd sidecar image.
-    logging.fluentd-sidecar-output-config: |
-      # Parse json log before sending to Elastic Search
-      <filter **>
-        @type parser
-        key_name log
-        <parse>
-          @type multi_format
-          <pattern>
-            format json
-            time_key fluentd-time # fluentd-time is reserved for structured logs
-            time_format %Y-%m-%dT%H:%M:%S.%NZ
-          </pattern>
-          <pattern>
-            format none
-            message_key log
-          </pattern>
-        </parse>
-      </filter>
-      # Send to Elastic Search
-      <match **>
-        @id elasticsearch
-        @type elasticsearch
-        @log_level info
-        include_tag_key true
-        # Elasticsearch service is in monitoring namespace.
-        host elasticsearch-logging.knative-monitoring
-        port 9200
-        logstash_format true
-        <buffer>
-          @type file
-          path /var/log/fluentd-buffers/kubernetes.system.buffer
-          flush_mode interval
-          retry_type exponential_backoff
-          flush_thread_count 2
-          flush_interval 5s
-          retry_forever
-          retry_max_interval 30
-          chunk_limit_size 2M
-          queue_limit_length 8
-          overflow_action block
-        </buffer>
-      </match>
-
     # metrics.backend-destination field specifies the system metrics destination.
     # It supports either prometheus (the default) or stackdriver.
     # Note: Using stackdriver will incur additional charges
     metrics.backend-destination: prometheus
 
-    # metrics.request-metrics-backend-destination specifies the request metrics
-    # destination. If non-empty, it enables queue proxy to send request metrics.
-    # Currently supported values: prometheus, stackdriver.
-    metrics.request-metrics-backend-destination: prometheus
-
     # metrics.stackdriver-project-id field specifies the stackdriver project ID. This
     # field is optional. When running on GCE, application default credentials will be
     # used if this field is not provided.
     metrics.stackdriver-project-id: "<your stackdriver project id>"
 
     # metrics.allow-stackdriver-custom-metrics indicates whether it is allowed to send metrics to
     # Stackdriver using "global" resource type and custom metric type if the
-    # metrics are not supported by "knative_revision" resource type. Setting this
+    # metrics are not supported by "knative_importer" resource type. Setting this
     # flag to "true" could cause extra Stackdriver charge.
     # If metrics.backend-destination is not Stackdriver, this is ignored.
     metrics.allow-stackdriver-custom-metrics: "false"

pkg/apis/pubsub/v1alpha1/pull_subscription_types.go

@@ -148,10 +148,6 @@ func (ps PullSubscriptionSpec) GetRetentionDuration() time.Duration {
 type ModeType string
 
 const (
-	// PubSubEventType is the GcpPubSub CloudEvent type, in case PullSubscription
-	// doesn't send a CloudEvent itself.
-	PubSubEventType = "google.pubsub.topic.publish"
-
 	// ModeCloudEventsBinary will use CloudEvents binary HTTP mode with
 	// flattened Pub/Sub payload.
 	ModeCloudEventsBinary ModeType = "CloudEventsBinary"

pkg/decorator/decorator.go

@@ -70,7 +70,7 @@ func (a *Decorator) receive(ctx context.Context, event cloudevents.Event, resp *
 		event.SetExtension(k, v)
 	}
 
-	if r, err := a.outbound.Send(ctx, event); err != nil {
+	if _, r, err := a.outbound.Send(ctx, event); err != nil {
 		return err
 	} else if r != nil {
 		resp.RespondWith(200, r)