Repository: knative-sandbox/eventing-rabbitmq (go)

Issue:
Broker Delivery Backoff policy is not being correctly parsed in the Dispatcher and the Receive adapter
**Describe the bug**
The Broker's and Trigger's dispatcher, and the Source's Receive Adapter are getting into a CrashloopBackoff because is not being able to parse the Spec.Delivery.BackoffDelay duration value:

`"level":"fatal","ts":1652108562.8887987,"logger":"fallback","caller":"dispatcher/main.go:62","msg":"Failed to process env var: envconfig.Process: assigning BACKOFF_DELAY to BackoffDelay: converting 'PT0.2S' to type time.Duration. details: time: invalid duration \"PT0.2S\""`

It seems to be initializing everything to the eventing core default values:
 ```
   K_TRACING_CONFIG:    {"backend":"none","debug":"false","sample-rate":"0.1"}
      RETRY:               10
      BACKOFF_POLICY:      exponential
      BACKOFF_DELAY:       PT0.2S
      PARALLELISM:         100
```

In the case of the Source it just happens when the Backoff Delay is explicitly set

 **Expected behavior**
The dispatcher parse correctly the Spec.Delivery.BackoffDelay Duration and works correctly

 **To Reproduce**
Create a Broker with a DLQ or a Trigger
Create a Source and set a spec.backoffDelay = PT2S

 **Knative release version**
1.4



Pull request description:
Fixed Receive Adapter and Dispatcher BackoffDelay period parsing


<!-- Thanks for sending a pull request! -->

# Changes

<!-- 
Describe your changes here- ideally you can get that description straight from
your descriptive commit message(s)! 

- :gift: Add new feature
- :bug: Fix bug
- :broom: Update or clean up current behavior
- :wastebasket: Remove feature or internal logic
-->

- :bug: Now the backoffDelay env variable is parsed correctly on the Broker's and Trigger's Dispatcher
- :bug: Now the backoffDelay env variable is parsed correctly on the Source's receive adapter
- 🧹 Added condition to Broker's e2e tests to try to catch this kind of bugs earlier

<!--
In addition, categorize the changes you're making using the "/kind" Prow command, example:

/kind <kind>

Supported kinds are: api-change, bug, cleanup, deprecation, removal, documentation, enhancement, performance

-->
/kind bug

<!--
*Automatically closes linked issue when PR is merged.
Usage: `Fixes #<issue number>`, or `Fixes (paste link of issue)`.
_If PR is about `failing-tests or flakes`, please post the related issues/tests in a comment and do not use `Fixes`_*
-->
Fixes #748 

<!-- Please include the 'why' behind your changes if no issue exists -->

**Release Note**

<!--
:page_facing_up: If this change has user-visible impact, write a release note in the block
below. Include the string "action required" if additional action is required of
users switching to the new release, for example in case of a breaking change.

Write as if you are speaking to users, not other Knative contributors. If this
change has no user-visible impact, no release-note is needed.
-->
```release-note
- :bug: Now the backoffDelay env variable is parsed correctly on the Broker's and Trigger's Dispatcher
- :bug: Now the backoffDelay env variable is parsed correctly on the Source's receive adapter
```


Patch:
diff --git a/Makefile b/Makefile
index c59455b2..359b248e 100644
--- a/Makefile
+++ b/Makefile
@@ -153,7 +153,7 @@ K_CMD = create --dry-run=client --output=yaml
 endif
 
 K9S_RELEASES := https://github.com/derailed/k9s/releases
-K9S_VERSION := 0.25.12
+K9S_VERSION := 0.25.18
 K9S_BIN_DIR := $(LOCAL_BIN)/k9s-$(K9S_VERSION)-$(platform)-x86_64
 K9S_URL := $(K9S_RELEASES)/download/v$(K9S_VERSION)/k9s_$(platform)_x86_64.tar.gz
 K9S := $(K9S_BIN_DIR)/k9s
diff --git a/cmd/dispatcher/main.go b/cmd/dispatcher/main.go
index 3daa6712..c4c2095c 100644
--- a/cmd/dispatcher/main.go
+++ b/cmd/dispatcher/main.go
@@ -28,6 +28,7 @@ import (
 	amqperr "github.com/rabbitmq/amqp091-go"
 	eventingduckv1 "knative.dev/eventing/pkg/apis/duck/v1"
 	"knative.dev/pkg/logging"
+	"knative.dev/pkg/metrics"
 	"knative.dev/pkg/signals"
 
 	"github.com/NeowayLabs/wabbit"
@@ -57,6 +58,9 @@ type envConfig struct {
 func main() {
 	ctx := signals.NewContext()
 
+	// Report stats on Go memory usage every 30 seconds.
+	metrics.MemStatsOrDie(ctx)
+
 	var env envConfig
 	if err := envconfig.Process("", &env); err != nil {
 		logging.FromContext(ctx).Fatal("Failed to process env var: ", err)
diff --git a/cmd/ingress/main.go b/cmd/ingress/main.go
index 8e73a820..529bdd16 100644
--- a/cmd/ingress/main.go
+++ b/cmd/ingress/main.go
@@ -21,19 +21,24 @@ import (
 	"errors"
 	"fmt"
 	"net/http"
+	"time"
 
 	cloudevents "github.com/cloudevents/sdk-go/v2"
 	"github.com/cloudevents/sdk-go/v2/binding"
 	cehttp "github.com/cloudevents/sdk-go/v2/protocol/http"
+	"github.com/google/uuid"
 	"github.com/kelseyhightower/envconfig"
 	amqp "github.com/rabbitmq/amqp091-go"
 	"go.opencensus.io/plugin/ochttp/propagation/tracecontext"
 	"go.opencensus.io/trace"
 	"go.uber.org/zap"
+	"knative.dev/eventing-rabbitmq/pkg/broker/ingress"
 	"knative.dev/eventing-rabbitmq/pkg/rabbit"
 	"knative.dev/eventing-rabbitmq/pkg/utils"
 	"knative.dev/eventing/pkg/kncloudevents"
+	"knative.dev/pkg/kmeta"
 	"knative.dev/pkg/logging"
+	"knative.dev/pkg/metrics"
 	"knative.dev/pkg/signals"
 )
 
@@ -41,6 +46,9 @@ const (
 	defaultMaxIdleConnections        = 1000
 	defaultMaxIdleConnectionsPerHost = 1000
 	component                        = "rabbitmq-ingress"
+
+	// noDuration signals that the dispatch step hasn't started
+	noDuration = -1
 )
 
 type envConfig struct {
@@ -52,11 +60,22 @@ type envConfig struct {
 
 	connection *amqp.Connection
 	channel    *amqp.Channel
+
+	ContainerName   string `envconfig:"CONTAINER_NAME" default:"ingress"`
+	PodName         string `envconfig:"POD_NAME" default:"rabbitmq-broker-ingress"`
+	BrokerName      string `envconfig:"BROKER_NAME"`
+	BrokerNamespace string `envconfig:"BROKER_NAMESPACE"`
+
+	reporter ingress.StatsReporter
 }
 
 func main() {
 	ctx := signals.NewContext()
 	var err error
+
+	// Report stats on Go memory usage every 30 seconds.
+	metrics.MemStatsOrDie(ctx)
+
 	var env envConfig
 	if err = envconfig.Process("", &env); err != nil {
 		logging.FromContext(ctx).Fatal("Failed to process env var: ", err)
@@ -97,6 +116,8 @@ func main() {
 	}
 	defer rmqHelper.CleanupRabbitMQ(env.connection, env.channel, retryChannel, logger)
 
+	env.reporter = ingress.NewStatsReporter(env.ContainerName, kmeta.ChildName(env.PodName, uuid.New().String()))
+
 	connectionArgs := kncloudevents.ConnectionArgs{
 		MaxIdleConns:        defaultMaxIdleConnections,
 		MaxIdleConnsPerHost: defaultMaxIdleConnectionsPerHost,
@@ -146,17 +167,28 @@ func (env *envConfig) ServeHTTP(writer http.ResponseWriter, request *http.Reques
 	span := trace.FromContext(ctx)
 	defer span.End()
 
-	statusCode, err := env.send(event, span)
+	reporterArgs := &ingress.ReportArgs{
+		Namespace:  env.BrokerNamespace,
+		BrokerName: env.BrokerName,
+		EventType:  event.Type(),
+	}
+
+	statusCode, dispatchTime, err := env.send(event, span)
 	if err != nil {
 		logger.Errorw("failed to send event", zap.Error(err))
 	}
+	if dispatchTime > noDuration {
+		_ = env.reporter.ReportEventDispatchTime(reporterArgs, statusCode, dispatchTime)
+	}
+	_ = env.reporter.ReportEventCount(reporterArgs, statusCode)
+
 	writer.WriteHeader(statusCode)
 }
 
-func (env *envConfig) send(event *cloudevents.Event, span *trace.Span) (int, error) {
+func (env *envConfig) send(event *cloudevents.Event, span *trace.Span) (int, time.Duration, error) {
 	bytes, err := json.Marshal(event)
 	if err != nil {
-		return http.StatusBadRequest, fmt.Errorf("failed to marshal event, %w", err)
+		return http.StatusBadRequest, noDuration, fmt.Errorf("failed to marshal event, %w", err)
 	}
 
 	tp, ts := (&tracecontext.HTTPFormat{}).SpanContextToHeaders(span.SpanContext())
@@ -171,7 +203,7 @@ func (env *envConfig) send(event *cloudevents.Event, span *trace.Span) (int, err
 	for key, val := range event.Extensions() {
 		headers[key] = val
 	}
-
+	start := time.Now()
 	dc, err := env.channel.PublishWithDeferredConfirm(
 		env.ExchangeName,
 		"",    // routing key
@@ -185,15 +217,15 @@ func (env *envConfig) send(event *cloudevents.Event, span *trace.Span) (int, err
 		})
 
 	if err != nil {
-		return http.StatusInternalServerError, fmt.Errorf("failed to publish message: %w", err)
+		return http.StatusInternalServerError, noDuration, fmt.Errorf("failed to publish message: %w", err)
 	}
 
 	ack := dc.Wait()
+	dispatchTime := time.Since(start)
 	if !ack {
-		return http.StatusInternalServerError, errors.New("failed to publish message: nacked")
+		return http.StatusInternalServerError, noDuration, errors.New("failed to publish message: nacked")
 	}
-
-	return http.StatusAccepted, nil
+	return http.StatusAccepted, dispatchTime, nil
 }
 
 func (env *envConfig) CreateRabbitMQConnections(
diff --git a/go.mod b/go.mod
index a058b759..654e1f60 100644
--- a/go.mod
+++ b/go.mod
@@ -33,6 +33,7 @@ require (
 )
 
 require (
+	github.com/google/uuid v1.3.0
 	github.com/rickb777/date v1.13.0
 	k8s.io/utils v0.0.0-20220210201930-3a6ce19ff2f9
 )
@@ -76,7 +77,6 @@ require (
 	github.com/google/go-querystring v1.0.0 // indirect
 	github.com/google/gofuzz v1.2.0 // indirect
 	github.com/google/mako v0.0.0-20190821191249-122f8dcef9e3 // indirect
-	github.com/google/uuid v1.3.0 // indirect
 	github.com/googleapis/gnostic v0.5.5 // indirect
 	github.com/grpc-ecosystem/grpc-gateway v1.16.0 // indirect
 	github.com/hashicorp/go-cleanhttp v0.5.2 // indirect
diff --git a/pkg/broker/ingress/stats_exporter.go b/pkg/broker/ingress/stats_exporter.go
new file mode 100644
index 00000000..15c5a3d2
--- /dev/null
+++ b/pkg/broker/ingress/stats_exporter.go
@@ -0,0 +1,159 @@
+/*
+Copyright 2022 The Knative Authors
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package ingress
+
+import (
+	"context"
+	"log"
+	"strconv"
+	"time"
+
+	"go.opencensus.io/resource"
+	"go.opencensus.io/stats"
+	"go.opencensus.io/stats/view"
+	"go.opencensus.io/tag"
+	broker "knative.dev/eventing-rabbitmq/pkg/broker"
+	eventingmetrics "knative.dev/eventing/pkg/metrics"
+	"knative.dev/pkg/metrics"
+	"knative.dev/pkg/metrics/metricskey"
+)
+
+var (
+	// eventCountM is a counter which records the number of events received
+	// by the Broker.
+	eventCountM = stats.Int64(
+		"event_count",
+		"Number of events received by a Broker",
+		stats.UnitDimensionless,
+	)
+
+	// dispatchTimeInMsecM records the time spent dispatching an event to
+	// a Channel, in milliseconds.
+	dispatchTimeInMsecM = stats.Float64(
+		"event_dispatch_latencies",
+		"The time spent dispatching an event to a Channel",
+		stats.UnitMilliseconds,
+	)
+
+	// Create the tag keys that will be used to add tags to our measurements.
+	// Tag keys must conform to the restrictions described in
+	// go.opencensus.io/tag/validate.go. Currently those restrictions are:
+	// - length between 1 and 255 inclusive
+	// - characters are printable US-ASCII
+	eventTypeKey         = tag.MustNewKey(eventingmetrics.LabelEventType)
+	responseCodeKey      = tag.MustNewKey(eventingmetrics.LabelResponseCode)
+	responseCodeClassKey = tag.MustNewKey(eventingmetrics.LabelResponseCodeClass)
+)
+
+type ReportArgs struct {
+	Namespace  string
+	BrokerName string
+	EventType  string
+}
+
+func init() {
+	register()
+}
+
+// StatsReporter defines the interface for sending ingress metrics.
+type StatsReporter interface {
+	ReportEventCount(args *ReportArgs, responseCode int) error
+	ReportEventDispatchTime(args *ReportArgs, responseCode int, d time.Duration) error
+}
+
+var _ StatsReporter = (*reporter)(nil)
+var emptyContext = context.Background()
+
+// Reporter holds cached metric objects to report ingress metrics.
+type reporter struct {
+	container  string
+	uniqueName string
+}
+
+// NewStatsReporter creates a reporter that collects and reports ingress metrics.
+func NewStatsReporter(container, uniqueName string) StatsReporter {
+	return &reporter{
+		container:  container,
+		uniqueName: uniqueName,
+	}
+}
+
+func register() {
+	tagKeys := []tag.Key{
+		eventTypeKey,
+		responseCodeKey,
+		responseCodeClassKey,
+		broker.ContainerTagKey,
+		broker.UniqueTagKey}
+
+	// Create view to see our measurements.
+	err := metrics.RegisterResourceView(
+		&view.View{
+			Description: eventCountM.Description(),
+			Measure:     eventCountM,
+			Aggregation: view.Count(),
+			TagKeys:     tagKeys,
+		},
+		&view.View{
+			Description: dispatchTimeInMsecM.Description(),
+			Measure:     dispatchTimeInMsecM,
+			Aggregation: view.Distribution(metrics.Buckets125(1, 10000)...), // 1, 2, 5, 10, 20, 50, 100, 500, 1000, 5000, 10000
+			TagKeys:     tagKeys,
+		},
+	)
+	if err != nil {
+		log.Printf("failed to register opencensus views, %s", err)
+	}
+}
+
+// ReportEventCount captures the event count.
+func (r *reporter) ReportEventCount(args *ReportArgs, responseCode int) error {
+	ctx, err := r.generateTag(args, responseCode)
+	if err != nil {
+		return err
+	}
+	metrics.Record(ctx, eventCountM.M(1))
+	return nil
+}
+
+// ReportEventDispatchTime captures dispatch times.
+func (r *reporter) ReportEventDispatchTime(args *ReportArgs, responseCode int, d time.Duration) error {
+	ctx, err := r.generateTag(args, responseCode)
+	if err != nil {
+		return err
+	}
+	// convert time.Duration in nanoseconds to milliseconds.
+	metrics.Record(ctx, dispatchTimeInMsecM.M(float64(d/time.Millisecond)))
+	return nil
+}
+
+func (r *reporter) generateTag(args *ReportArgs, responseCode int) (context.Context, error) {
+	ctx := metricskey.WithResource(emptyContext, resource.Resource{
+		Type: eventingmetrics.ResourceTypeKnativeBroker,
+		Labels: map[string]string{
+			eventingmetrics.LabelNamespaceName: args.Namespace,
+			eventingmetrics.LabelBrokerName:    args.BrokerName,
+		},
+	})
+	return tag.New(
+		ctx,
+		tag.Insert(broker.ContainerTagKey, r.container),
+		tag.Insert(broker.UniqueTagKey, r.uniqueName),
+		tag.Insert(eventTypeKey, args.EventType),
+		tag.Insert(responseCodeKey, strconv.Itoa(responseCode)),
+		tag.Insert(responseCodeClassKey, metrics.ResponseCodeClass(responseCode)))
+}
diff --git a/pkg/broker/metrics.go b/pkg/broker/metrics.go
new file mode 100644
index 00000000..a8909e0a
--- /dev/null
+++ b/pkg/broker/metrics.go
@@ -0,0 +1,41 @@
+/*
+Copyright 2022 The Knative Authors
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package broker
+
+import (
+	"go.opencensus.io/tag"
+	"knative.dev/eventing/pkg/metrics"
+)
+
+const (
+	// EventArrivalTime is used to access the metadata stored on a
+	// CloudEvent to measure the time difference between when an event is
+	// received on a broker and before it is dispatched to the trigger function.
+	// The format is an RFC3339 time in string format. For example: 2019-08-26T23:38:17.834384404Z.
+	EventArrivalTime = "knativearrivaltime"
+
+	// LabelUniqueName is the label for the unique name per stats_reporter instance.
+	LabelUniqueName = "unique_name"
+
+	// LabelContainerName is the label for the immutable name of the container.
+	LabelContainerName = metrics.LabelContainerName
+)
+
+var (
+	ContainerTagKey = tag.MustNewKey(LabelContainerName)
+	UniqueTagKey    = tag.MustNewKey(LabelUniqueName)
+)
diff --git a/pkg/reconciler/broker/resources/dispatcher.go b/pkg/reconciler/broker/resources/dispatcher.go
index 5dffa3f0..c8243526 100644
--- a/pkg/reconciler/broker/resources/dispatcher.go
+++ b/pkg/reconciler/broker/resources/dispatcher.go
@@ -23,6 +23,7 @@ import (
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 
+	"github.com/rickb777/date/period"
 	eventingduckv1 "knative.dev/eventing/pkg/apis/duck/v1"
 	"knative.dev/eventing/pkg/apis/eventing"
 	eventingv1 "knative.dev/eventing/pkg/apis/eventing/v1"
@@ -113,10 +114,11 @@ func MakeDispatcherDeployment(args *DispatcherArgs) *appsv1.Deployment {
 				})
 		}
 		if args.Delivery.BackoffDelay != nil {
+			p, _ := period.Parse(*args.Delivery.BackoffDelay)
 			envs = append(envs,
 				corev1.EnvVar{
 					Name:  "BACKOFF_DELAY",
-					Value: *args.Delivery.BackoffDelay,
+					Value: p.DurationApprox().String(),
 				})
 		}
 	}
diff --git a/pkg/reconciler/broker/resources/ingress.go b/pkg/reconciler/broker/resources/ingress.go
index 3827dd58..69af9c28 100644
--- a/pkg/reconciler/broker/resources/ingress.go
+++ b/pkg/reconciler/broker/resources/ingress.go
@@ -48,6 +48,8 @@ type IngressArgs struct {
 
 // MakeIngressDeployment creates the in-memory representation of the Broker's ingress Deployment.
 func MakeIngressDeployment(args *IngressArgs) *appsv1.Deployment {
+	DeploymentName := fmt.Sprintf("%s-broker-ingress", args.Broker.Name)
+
 	envs := []corev1.EnvVar{{
 		Name:  system.NamespaceEnvKey,
 		Value: system.Namespace(),
@@ -64,6 +66,18 @@ func MakeIngressDeployment(args *IngressArgs) *appsv1.Deployment {
 	}, {
 		Name:  "EXCHANGE_NAME",
 		Value: naming.BrokerExchangeName(args.Broker, false),
+	}, {
+		Name:  "CONTAINER_NAME",
+		Value: ingressContainerName,
+	}, {
+		Name:  "POD_NAME",
+		Value: DeploymentName,
+	}, {
+		Name:  "BROKER_NAME",
+		Value: args.Broker.Name,
+	}, {
+		Name:  "BROKER_NAMESPACE",
+		Value: args.Broker.Namespace,
 	}}
 	if args.Configs != nil {
 		envs = append(envs, args.Configs.ToEnvVars()...)
@@ -72,7 +86,7 @@ func MakeIngressDeployment(args *IngressArgs) *appsv1.Deployment {
 	return &appsv1.Deployment{
 		ObjectMeta: metav1.ObjectMeta{
 			Namespace: args.Broker.Namespace,
-			Name:      fmt.Sprintf("%s-broker-ingress", args.Broker.Name),
+			Name:      DeploymentName,
 			OwnerReferences: []metav1.OwnerReference{
 				*kmeta.NewControllerRef(args.Broker),
 			},
diff --git a/pkg/reconciler/source/resources/receive_adapter.go b/pkg/reconciler/source/resources/receive_adapter.go
index 592593a3..1ba1d7cf 100644
--- a/pkg/reconciler/source/resources/receive_adapter.go
+++ b/pkg/reconciler/source/resources/receive_adapter.go
@@ -20,6 +20,7 @@ import (
 	"fmt"
 	"strconv"
 
+	"github.com/rickb777/date/period"
 	v1 "k8s.io/api/apps/v1"
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -151,9 +152,10 @@ func MakeReceiveAdapter(args *ReceiveAdapterArgs) *v1.Deployment {
 	}
 
 	if args.Source.Spec.BackoffDelay != nil {
+		p, _ := period.Parse(*args.Source.Spec.BackoffDelay)
 		env = append(env, corev1.EnvVar{
 			Name:  "HTTP_SENDER_BACKOFF_DELAY",
-			Value: *args.Source.Spec.BackoffDelay,
+			Value: p.DurationApprox().String(),
 		})
 	}
 
diff --git a/pkg/reconciler/trigger/resources/dispatcher.go b/pkg/reconciler/trigger/resources/dispatcher.go
index e18fd92d..d63f6fad 100644
--- a/pkg/reconciler/trigger/resources/dispatcher.go
+++ b/pkg/reconciler/trigger/resources/dispatcher.go
@@ -118,10 +118,11 @@ func MakeDispatcherDeployment(args *DispatcherArgs) *appsv1.Deployment {
 				})
 		}
 		if args.Delivery.BackoffDelay != nil {
+			p, _ := period.Parse(*args.Delivery.BackoffDelay)
 			dispatcher.Env = append(dispatcher.Env,
 				corev1.EnvVar{
 					Name:  "BACKOFF_DELAY",
-					Value: *args.Delivery.BackoffDelay,
+					Value: p.DurationApprox().String(),
 				})
 		}
 		if args.Delivery.Timeout != nil {
diff --git a/vendor/knative.dev/eventing/pkg/broker/metrics.go b/vendor/knative.dev/eventing/pkg/broker/metrics.go
new file mode 100644
index 00000000..4daca105
--- /dev/null
+++ b/vendor/knative.dev/eventing/pkg/broker/metrics.go
@@ -0,0 +1,41 @@
+/*
+Copyright 2019 The Knative Authors
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package broker
+
+import (
+	"go.opencensus.io/tag"
+	"knative.dev/eventing/pkg/metrics"
+)
+
+const (
+	// EventArrivalTime is used to access the metadata stored on a
+	// CloudEvent to measure the time difference between when an event is
+	// received on a broker and before it is dispatched to the trigger function.
+	// The format is an RFC3339 time in string format. For example: 2019-08-26T23:38:17.834384404Z.
+	EventArrivalTime = "knativearrivaltime"
+
+	// LabelUniqueName is the label for the unique name per stats_reporter instance.
+	LabelUniqueName = "unique_name"
+
+	// LabelContainerName is the label for the immutable name of the container.
+	LabelContainerName = metrics.LabelContainerName
+)
+
+var (
+	ContainerTagKey = tag.MustNewKey(LabelContainerName)
+	UniqueTagKey    = tag.MustNewKey(LabelUniqueName)
+)
diff --git a/vendor/knative.dev/eventing/pkg/broker/ttl.go b/vendor/knative.dev/eventing/pkg/broker/ttl.go
new file mode 100644
index 00000000..60c930e8
--- /dev/null
+++ b/vendor/knative.dev/eventing/pkg/broker/ttl.go
@@ -0,0 +1,97 @@
+/*
+Copyright 2019 The Knative Authors
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package broker
+
+import (
+	"context"
+
+	cloudevents "github.com/cloudevents/sdk-go/v2"
+	"github.com/cloudevents/sdk-go/v2/client"
+	cetypes "github.com/cloudevents/sdk-go/v2/types"
+	"go.uber.org/zap"
+)
+
+const (
+	// TTLAttribute is the name of the CloudEvents extension attribute used to store the
+	// Broker's TTL (number of times a single event can reply through a Broker continuously). All
+	// interactions with the attribute should be done through the GetTTL and SetTTL functions.
+	TTLAttribute = "knativebrokerttl"
+)
+
+// GetTTL finds the TTL in the EventContext using a case insensitive comparison
+// for the key. The second return param, is the case preserved key that matched.
+// Depending on the encoding/transport, the extension case could be changed.
+func GetTTL(ctx cloudevents.EventContext) (int32, error) {
+	ttl, err := ctx.GetExtension(TTLAttribute)
+	if err != nil {
+		return 0, err
+	}
+	return cetypes.ToInteger(ttl)
+}
+
+// SetTTL sets the TTL into the EventContext. ttl should be a positive integer.
+func SetTTL(ctx cloudevents.EventContext, ttl int32) error {
+	return ctx.SetExtension(TTLAttribute, ttl)
+}
+
+// DeleteTTL removes the TTL CE extension attribute
+func DeleteTTL(ctx cloudevents.EventContext) error {
+	return ctx.SetExtension(TTLAttribute, nil)
+}
+
+// TTLDefaulter returns a cloudevents event defaulter that will manage the TTL
+// for events with the following rules:
+//   If TTL is not found, it will set it to the default passed in.
+//   If TTL is <= 0, it will remain 0.
+//   If TTL is > 1, it will be reduced by one.
+func TTLDefaulter(logger *zap.Logger, defaultTTL int32) client.EventDefaulter {
+	return func(ctx context.Context, event cloudevents.Event) cloudevents.Event {
+		// Get the current or default TTL from the event.
+		var ttl int32
+		if ttlraw, err := event.Context.GetExtension(TTLAttribute); err != nil {
+			logger.Debug("TTL not found in outbound event, defaulting.",
+				zap.String("event.id", event.ID()),
+				zap.Int32(TTLAttribute, defaultTTL),
+				zap.Error(err),
+			)
+			ttl = defaultTTL
+		} else if ttl, err = cetypes.ToInteger(ttlraw); err != nil {
+			logger.Warn("Failed to convert existing TTL into integer, defaulting.",
+				zap.String("event.id", event.ID()),
+				zap.Any(TTLAttribute, ttlraw),
+				zap.Error(err),
+			)
+			ttl = defaultTTL
+		} else {
+			// Decrement TTL.
+			ttl = ttl - 1
+			if ttl < 0 {
+				ttl = 0
+			}
+		}
+		// Overwrite the TTL into the event.
+		if err := SetTTL(event.Context, ttl); err != nil {
+			logger.Error("Failed to set TTL on outbound event.",
+				zap.String("event.id", event.ID()),
+				zap.Int32(TTLAttribute, ttl),
+				zap.Error(err),
+			)
+		}
+
+		return event
+	}
+}
diff --git a/vendor/modules.txt b/vendor/modules.txt
index b4979f27..1e3a1b6a 100644
--- a/vendor/modules.txt
+++ b/vendor/modules.txt
@@ -1076,6 +1076,7 @@ knative.dev/eventing/pkg/apis/sources
 knative.dev/eventing/pkg/apis/sources/config
 knative.dev/eventing/pkg/apis/sources/v1
 knative.dev/eventing/pkg/apis/sources/v1beta2
+knative.dev/eventing/pkg/broker
 knative.dev/eventing/pkg/client/clientset/versioned
 knative.dev/eventing/pkg/client/clientset/versioned/fake
 knative.dev/eventing/pkg/client/clientset/versioned/scheme
@@ -1218,6 +1219,7 @@ knative.dev/pkg/logging/logkey
 knative.dev/pkg/logging/testing
 knative.dev/pkg/metrics
 knative.dev/pkg/metrics/metricskey
+knative.dev/pkg/metrics/metricstest
 knative.dev/pkg/metrics/testing
 knative.dev/pkg/network
 knative.dev/pkg/network/handlers


Test patch:
diff --git a/pkg/broker/ingress/stats_exporter_test.go b/pkg/broker/ingress/stats_exporter_test.go
new file mode 100644
index 00000000..9e845cd9
--- /dev/null
+++ b/pkg/broker/ingress/stats_exporter_test.go
@@ -0,0 +1,95 @@
+/*
+Copyright 2022 The Knative Authors
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package ingress
+
+import (
+	"net/http"
+	"testing"
+	"time"
+
+	"go.opencensus.io/resource"
+	broker "knative.dev/eventing/pkg/broker"
+	"knative.dev/eventing/pkg/metrics"
+	"knative.dev/pkg/metrics/metricstest"
+	_ "knative.dev/pkg/metrics/testing"
+)
+
+func TestStatsReporter(t *testing.T) {
+	setup()
+
+	args := &ReportArgs{
+		Namespace:  "testns",
+		BrokerName: "testbroker",
+		EventType:  "testeventtype",
+	}
+
+	r := NewStatsReporter("testcontainer", "testpod")
+
+	wantTags := map[string]string{
+		metrics.LabelEventType:         "testeventtype",
+		metrics.LabelResponseCode:      "202",
+		metrics.LabelResponseCodeClass: "2xx",
+		broker.LabelUniqueName:         "testpod",
+		broker.LabelContainerName:      "testcontainer",
+	}
+
+	resource := resource.Resource{
+		Type: metrics.ResourceTypeKnativeBroker,
+		Labels: map[string]string{
+			metrics.LabelNamespaceName: "testns",
+			metrics.LabelBrokerName:    "testbroker",
+		},
+	}
+
+	// test ReportEventCount
+	expectSuccess(t, func() error {
+		return r.ReportEventCount(args, http.StatusAccepted)
+	})
+	expectSuccess(t, func() error {
+		return r.ReportEventCount(args, http.StatusAccepted)
+	})
+	metricstest.AssertMetric(t, metricstest.IntMetric("event_count", 2, wantTags).WithResource(&resource))
+
+	// test ReportDispatchTime
+	expectSuccess(t, func() error {
+		return r.ReportEventDispatchTime(args, http.StatusAccepted, 1100*time.Millisecond)
+	})
+	expectSuccess(t, func() error {
+		return r.ReportEventDispatchTime(args, http.StatusAccepted, 9100*time.Millisecond)
+	})
+	metricstest.AssertMetric(t, metricstest.DistributionCountOnlyMetric("event_dispatch_latencies", 2, wantTags))
+	metricstest.CheckDistributionData(t, "event_dispatch_latencies", wantTags, 2, 1100.0, 9100.0)
+}
+
+func expectSuccess(t *testing.T, f func() error) {
+	t.Helper()
+	if err := f(); err != nil {
+		t.Error("Reporter expected success but got error:", err)
+	}
+}
+
+func setup() {
+	resetMetrics()
+}
+
+func resetMetrics() {
+	// OpenCensus metrics carry global state that need to be reset between unit tests.
+	metricstest.Unregister(
+		"event_count",
+		"event_dispatch_latencies")
+	register()
+}
diff --git a/pkg/reconciler/broker/resources/dispatcher_test.go b/pkg/reconciler/broker/resources/dispatcher_test.go
index 069fca48..e199ddd0 100644
--- a/pkg/reconciler/broker/resources/dispatcher_test.go
+++ b/pkg/reconciler/broker/resources/dispatcher_test.go
@@ -63,7 +63,7 @@ func TestMakeDispatcherDeployment(t *testing.T) {
 		BrokerIngressURL:   bURL,
 		Delivery: &v1.DeliverySpec{
 			Retry:         ptr.Int32(10),
-			BackoffDelay:  ptr.String("20s"),
+			BackoffDelay:  ptr.String("PT20S"),
 			BackoffPolicy: &linear,
 		},
 	}
diff --git a/pkg/reconciler/broker/resources/ingress_test.go b/pkg/reconciler/broker/resources/ingress_test.go
index 09d1e586..31119838 100644
--- a/pkg/reconciler/broker/resources/ingress_test.go
+++ b/pkg/reconciler/broker/resources/ingress_test.go
@@ -98,6 +98,18 @@ func TestMakeIngressDeployment(t *testing.T) {
 						}, {
 							Name:  "EXCHANGE_NAME",
 							Value: brokerExchangeName,
+						}, {
+							Name:  "CONTAINER_NAME",
+							Value: "ingress",
+						}, {
+							Name:  "POD_NAME",
+							Value: "testbroker-broker-ingress",
+						}, {
+							Name:  "BROKER_NAME",
+							Value: "testbroker",
+						}, {
+							Name:  "BROKER_NAMESPACE",
+							Value: "testnamespace",
 						}},
 						Ports: []corev1.ContainerPort{{
 							ContainerPort: 8080,
diff --git a/pkg/reconciler/source/resources/receive_adapter_test.go b/pkg/reconciler/source/resources/receive_adapter_test.go
index 0a8b2604..2a3e7d79 100644
--- a/pkg/reconciler/source/resources/receive_adapter_test.go
+++ b/pkg/reconciler/source/resources/receive_adapter_test.go
@@ -30,7 +30,7 @@ import (
 func TestMakeReceiveAdapter(t *testing.T) {
 	var retry int32 = 5
 	parallelism := 10
-	backoffDelay := "50ms"
+	backoffDelay := "PT0.1S"
 
 	for _, tt := range []struct {
 		name          string
@@ -259,7 +259,7 @@ func TestMakeReceiveAdapter(t *testing.T) {
 										},
 										{
 											Name:  "HTTP_SENDER_BACKOFF_DELAY",
-											Value: "50ms",
+											Value: "100ms",
 										},
 									},
 								},
diff --git a/pkg/reconciler/trigger/resources/dispatcher_test.go b/pkg/reconciler/trigger/resources/dispatcher_test.go
index 1ef4dfb0..56846570 100644
--- a/pkg/reconciler/trigger/resources/dispatcher_test.go
+++ b/pkg/reconciler/trigger/resources/dispatcher_test.go
@@ -63,7 +63,7 @@ func TestMakeDispatcherDeployment(t *testing.T) {
 			args: dispatcherArgs(withDelivery(&eventingduckv1.DeliverySpec{
 				Retry:         Int32Ptr(10),
 				BackoffPolicy: &exponentialBackoff,
-				BackoffDelay:  ptr.String("20s"),
+				BackoffDelay:  ptr.String("PT20S"),
 				Timeout:       pointer.StringPtr("PT10S"),
 			})),
 			want: deployment(
diff --git a/vendor/knative.dev/pkg/metrics/metricstest/metricstest.go b/vendor/knative.dev/pkg/metrics/metricstest/metricstest.go
new file mode 100644
index 00000000..3d054a18
--- /dev/null
+++ b/vendor/knative.dev/pkg/metrics/metricstest/metricstest.go
@@ -0,0 +1,262 @@
+/*
+Copyright 2019 The Knative Authors
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package metricstest
+
+import (
+	"fmt"
+	"reflect"
+
+	"go.opencensus.io/metric/metricproducer"
+	"go.opencensus.io/stats/view"
+)
+
+type ti interface {
+	Helper()
+	Error(args ...interface{})
+}
+
+// CheckStatsReported checks that there is a view registered with the given name for each string in names,
+// and that each view has at least one record.
+func CheckStatsReported(t ti, names ...string) {
+	t.Helper()
+	for _, name := range names {
+		d, err := readRowsFromAllMeters(name)
+		if err != nil {
+			t.Error("For metric, Reporter.Report() error", "metric", name, "error", err)
+		}
+		if len(d) < 1 {
+			t.Error("For metric, no data reported when data was expected, view data is empty.", "metric", name)
+		}
+	}
+}
+
+// CheckStatsNotReported checks that there are no records for any views that a name matching a string in names.
+// Names that do not match registered views are considered not reported.
+func CheckStatsNotReported(t ti, names ...string) {
+	t.Helper()
+	for _, name := range names {
+		d, err := readRowsFromAllMeters(name)
+		// err == nil means a valid stat exists matching "name"
+		// len(d) > 0 means a component recorded metrics for that stat
+		if err == nil && len(d) > 0 {
+			t.Error("For metric, unexpected data reported when no data was expected.", "metric", name, "Reporter len(d)", len(d))
+		}
+	}
+}
+
+// CheckCountData checks the view with a name matching string name to verify that the CountData stats
+// reported are tagged with the tags in wantTags and that wantValue matches reported count.
+func CheckCountData(t ti, name string, wantTags map[string]string, wantValue int64) {
+	t.Helper()
+	row, err := checkExactlyOneRow(t, name)
+	if err != nil {
+		t.Error(err)
+		return
+	}
+	checkRowTags(t, row, name, wantTags)
+
+	if s, ok := row.Data.(*view.CountData); !ok {
+		t.Error("want CountData", "metric", name, "got", reflect.TypeOf(row.Data))
+	} else if s.Value != wantValue {
+		t.Error("Wrong value", "metric", name, "value", s.Value, "want", wantValue)
+	}
+}
+
+// CheckDistributionData checks the view with a name matching string name to verify that the DistributionData stats reported
+// are tagged with the tags in wantTags and that expectedCount number of records were reported.
+// It also checks that expectedMin and expectedMax match the minimum and maximum reported values, respectively.
+func CheckDistributionData(t ti, name string, wantTags map[string]string, expectedCount int64, expectedMin float64, expectedMax float64) {
+	t.Helper()
+	row, err := checkExactlyOneRow(t, name)
+	if err != nil {
+		t.Error(err)
+		return
+	}
+	checkRowTags(t, row, name, wantTags)
+
+	if s, ok := row.Data.(*view.DistributionData); !ok {
+		t.Error("want DistributionData", "metric", name, "got", reflect.TypeOf(row.Data))
+	} else {
+		if s.Count != expectedCount {
+			t.Error("reporter count wrong", "metric", name, "got", s.Count, "want", expectedCount)
+		}
+		if s.Min != expectedMin {
+			t.Error("reporter min wrong", "metric", name, "got", s.Min, "want", expectedMin)
+		}
+		if s.Max != expectedMax {
+			t.Error("reporter max wrong", "metric", name, "got", s.Max, "want", expectedMax)
+		}
+	}
+}
+
+// CheckDistributionCount checks the view with a name matching string name to verify that the DistributionData stats reported
+// are tagged with the tags in wantTags and that expectedCount number of records were reported.
+func CheckDistributionCount(t ti, name string, wantTags map[string]string, expectedCount int64) {
+	t.Helper()
+	row, err := checkExactlyOneRow(t, name)
+	if err != nil {
+		t.Error(err)
+		return
+	}
+	checkRowTags(t, row, name, wantTags)
+
+	if s, ok := row.Data.(*view.DistributionData); !ok {
+		t.Error("want DistributionData", "metric", name, "got", reflect.TypeOf(row.Data))
+	} else if s.Count != expectedCount {
+		t.Error("reporter count wrong", "metric", name, "got", s.Count, "want", expectedCount)
+	}
+
+}
+
+// GetLastValueData returns the last value for the given metric, verifying tags.
+func GetLastValueData(t ti, name string, tags map[string]string) float64 {
+	t.Helper()
+	return GetLastValueDataWithMeter(t, name, tags, nil)
+}
+
+// GetLastValueDataWithMeter returns the last value of the given metric using meter, verifying tags.
+func GetLastValueDataWithMeter(t ti, name string, tags map[string]string, meter view.Meter) float64 {
+	t.Helper()
+	if row := lastRow(t, name, meter); row != nil {
+		checkRowTags(t, row, name, tags)
+
+		s, ok := row.Data.(*view.LastValueData)
+		if !ok {
+			t.Error("want LastValueData", "metric", name, "got", reflect.TypeOf(row.Data))
+		}
+		return s.Value
+	}
+	return 0
+}
+
+// CheckLastValueData checks the view with a name matching string name to verify that the LastValueData stats
+// reported are tagged with the tags in wantTags and that wantValue matches reported last value.
+func CheckLastValueData(t ti, name string, wantTags map[string]string, wantValue float64) {
+	t.Helper()
+	CheckLastValueDataWithMeter(t, name, wantTags, wantValue, nil)
+}
+
+// CheckLastValueDataWithMeter checks the  view with a name matching the string name in the
+// specified Meter (resource-specific view) to verify that the LastValueData stats are tagged with
+// the tags in wantTags and that wantValue matches the last reported value.
+func CheckLastValueDataWithMeter(t ti, name string, wantTags map[string]string, wantValue float64, meter view.Meter) {
+	t.Helper()
+	if v := GetLastValueDataWithMeter(t, name, wantTags, meter); v != wantValue {
+		t.Error("Reporter.Report() wrong value", "metric", name, "got", v, "want", wantValue)
+	}
+}
+
+// CheckSumData checks the view with a name matching string name to verify that the SumData stats
+// reported are tagged with the tags in wantTags and that wantValue matches the reported sum.
+func CheckSumData(t ti, name string, wantTags map[string]string, wantValue float64) {
+	t.Helper()
+	row, err := checkExactlyOneRow(t, name)
+	if err != nil {
+		t.Error(err)
+		return
+	}
+	checkRowTags(t, row, name, wantTags)
+
+	if s, ok := row.Data.(*view.SumData); !ok {
+		t.Error("Wrong type", "metric", name, "got", reflect.TypeOf(row.Data), "want", "SumData")
+	} else if s.Value != wantValue {
+		t.Error("Wrong sumdata", "metric", name, "got", s.Value, "want", wantValue)
+	}
+}
+
+// Unregister unregisters the metrics that were registered.
+// This is useful for testing since golang execute test iterations within the same process and
+// opencensus views maintain global state. At the beginning of each test, tests should
+// unregister for all metrics and then re-register for the same metrics. This effectively clears
+// out any existing data and avoids a panic due to re-registering a metric.
+//
+// In normal process shutdown, metrics do not need to be unregistered.
+func Unregister(names ...string) {
+	for _, producer := range metricproducer.GlobalManager().GetAll() {
+		meter := producer.(view.Meter)
+		for _, n := range names {
+			if v := meter.Find(n); v != nil {
+				meter.Unregister(v)
+			}
+		}
+	}
+}
+
+func lastRow(t ti, name string, meter view.Meter) *view.Row {
+	t.Helper()
+	var d []*view.Row
+	var err error
+	if meter != nil {
+		d, err = meter.RetrieveData(name)
+	} else {
+		d, err = readRowsFromAllMeters(name)
+	}
+	if err != nil {
+		t.Error("Reporter.Report() error", "metric", name, "error", err)
+		return nil
+	}
+	if len(d) < 1 {
+		t.Error("Reporter.Report() wrong length", "metric", name, "got", len(d), "want at least", 1)
+		return nil
+	}
+
+	return d[len(d)-1]
+}
+
+func checkExactlyOneRow(t ti, name string) (*view.Row, error) {
+	rows, err := readRowsFromAllMeters(name)
+	if err != nil || len(rows) == 0 {
+		return nil, fmt.Errorf("could not find row for %q", name)
+	}
+	if len(rows) > 1 {
+		return nil, fmt.Errorf("expected 1 row for metric %q got %d", name, len(rows))
+	}
+	return rows[0], nil
+}
+
+func readRowsFromAllMeters(name string) ([]*view.Row, error) {
+	// view.Meter implements (and is exposed by) metricproducer.GetAll. Since
+	// this is a test, reach around and cast these to view.Meter.
+	var rows []*view.Row
+	for _, producer := range metricproducer.GlobalManager().GetAll() {
+		meter := producer.(view.Meter)
+		d, err := meter.RetrieveData(name)
+		if err != nil || len(d) == 0 {
+			continue
+		}
+		if rows != nil {
+			return nil, fmt.Errorf("got metrics for the same name from different meters: %+v, %+v", rows, d)
+		}
+		rows = d
+	}
+	return rows, nil
+}
+
+func checkRowTags(t ti, row *view.Row, name string, wantTags map[string]string) {
+	t.Helper()
+	if wantlen, gotlen := len(wantTags), len(row.Tags); gotlen != wantlen {
+		t.Error("Reporter got wrong number of tags", "metric", name, "got", gotlen, "want", wantlen)
+	}
+	for _, got := range row.Tags {
+		n := got.Key.Name()
+		if want, ok := wantTags[n]; !ok {
+			t.Error("Reporter got an extra tag", "metric", name, "gotName", n, "gotValue", got.Value)
+		} else if got.Value != want {
+			t.Error("Reporter expected a different tag value for key", "metric", name, "key", n, "got", got.Value, "want", want)
+		}
+	}
+}
diff --git a/vendor/knative.dev/pkg/metrics/metricstest/resource_metrics.go b/vendor/knative.dev/pkg/metrics/metricstest/resource_metrics.go
new file mode 100644
index 00000000..10e2e4c9
--- /dev/null
+++ b/vendor/knative.dev/pkg/metrics/metricstest/resource_metrics.go
@@ -0,0 +1,350 @@
+/*
+Copyright 2020 The Knative Authors
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+// Package metricstest simplifies some of the common boilerplate around testing
+// metrics exports. It should work with or without the code in metrics, but this
+// code particularly knows how to deal with metrics which are exported for
+// multiple Resources in the same process.
+package metricstest
+
+import (
+	"fmt"
+	"sort"
+	"strings"
+	"testing"
+
+	"github.com/google/go-cmp/cmp"
+	"go.opencensus.io/metric/metricdata"
+	"go.opencensus.io/metric/metricproducer"
+	"go.opencensus.io/resource"
+	"go.opencensus.io/stats/view"
+)
+
+// Value provides a simplified implementation of a metric Value suitable for
+// easy testing.
+type Value struct {
+	Tags map[string]string
+	// union interface, only one of these will be set
+	Int64        *int64
+	Float64      *float64
+	Distribution *metricdata.Distribution
+	// VerifyDistributionCountOnly makes Equal compare the Distribution with the
+	// field Count only, and ignore all other fields of Distribution.
+	// This is ignored when the value is not a Distribution.
+	VerifyDistributionCountOnly bool
+}
+
+// Metric provides a simplified (for testing) implementation of a metric report
+// for a given metric name in a given Resource.
+type Metric struct {
+	// Name is the exported name of the metric, probably from the View's name.
+	Name string
+	// Unit is the units of measure of the metric. This is only checked for
+	// equality if Unit is non-empty or VerifyMetadata is true on both Metrics.
+	Unit metricdata.Unit
+	// Type is the type of measurement represented by the metric. This is only
+	// checked for equality if VerifyMetadata is true on both Metrics.
+	Type metricdata.Type
+
+	// Resource is the reported Resource (if any) for this metric. This is only
+	// checked for equality if Resource is non-nil or VerifyResource is true on
+	// both Metrics.
+	Resource *resource.Resource
+
+	// Values contains the values recorded for different Key=Value Tag
+	// combinations. Value is checked for equality if present.
+	Values []Value
+
+	// Equality testing/validation settings on the Metric. These are used to
+	// allow simple construction and usage with github.com/google/go-cmp/cmp
+
+	// VerifyMetadata makes Equal compare Unit and Type if it is true on both
+	// Metrics.
+	VerifyMetadata bool
+	// VerifyResource makes Equal compare Resource if it is true on Metrics with
+	// nil Resource. Metrics with non-nil Resource are always compared.
+	VerifyResource bool
+}
+
+// NewMetric creates a Metric from a metricdata.Metric, which is designed for
+// compact wire representation.
+func NewMetric(metric *metricdata.Metric) Metric {
+	value := Metric{
+		Name:     metric.Descriptor.Name,
+		Unit:     metric.Descriptor.Unit,
+		Type:     metric.Descriptor.Type,
+		Resource: metric.Resource,
+
+		VerifyMetadata: true,
+		VerifyResource: true,
+
+		Values: make([]Value, 0, len(metric.TimeSeries)),
+	}
+
+	for _, ts := range metric.TimeSeries {
+		tags := make(map[string]string, len(metric.Descriptor.LabelKeys))
+		for i, k := range metric.Descriptor.LabelKeys {
+			if ts.LabelValues[i].Present {
+				tags[k.Key] = ts.LabelValues[i].Value
+			}
+		}
+		v := Value{Tags: tags}
+		ts.Points[0].ReadValue(&v)
+		value.Values = append(value.Values, v)
+	}
+
+	return value
+}
+
+// EnsureRecorded makes sure that all stats metrics are actually flushed and recorded.
+func EnsureRecorded() {
+	// stats.Record queues the actual record to a channel to be accounted for by
+	// a background goroutine (nonblocking). Call a method which does a
+	// round-trip to that goroutine to ensure that records have been flushed.
+	for _, producer := range metricproducer.GlobalManager().GetAll() {
+		if meter, ok := producer.(view.Meter); ok {
+			meter.Find("nonexistent")
+		}
+	}
+}
+
+// GetMetric returns all values for the named metric.
+func GetMetric(name string) []Metric {
+	producers := metricproducer.GlobalManager().GetAll()
+	retval := make([]Metric, 0, len(producers))
+	for _, p := range producers {
+		for _, m := range p.Read() {
+			if m.Descriptor.Name == name && len(m.TimeSeries) > 0 {
+				retval = append(retval, NewMetric(m))
+			}
+		}
+	}
+	return retval
+}
+
+// GetOneMetric is like GetMetric, but it panics if more than a single Metric is
+// found.
+func GetOneMetric(name string) Metric {
+	m := GetMetric(name)
+	if len(m) != 1 {
+		panic(fmt.Sprint("Got wrong number of metrics:", m))
+	}
+	return m[0]
+}
+
+// IntMetric creates an Int64 metric.
+func IntMetric(name string, value int64, tags map[string]string) Metric {
+	return Metric{
+		Name:   name,
+		Values: []Value{{Int64: &value, Tags: tags}},
+	}
+}
+
+// FloatMetric creates a Float64 metric
+func FloatMetric(name string, value float64, tags map[string]string) Metric {
+	return Metric{
+		Name:   name,
+		Values: []Value{{Float64: &value, Tags: tags}},
+	}
+}
+
+// DistributionCountOnlyMetric creates a distribution metric for test, and verifying only the count.
+func DistributionCountOnlyMetric(name string, count int64, tags map[string]string) Metric {
+	return Metric{
+		Name: name,
+		Values: []Value{{
+			Distribution:                &metricdata.Distribution{Count: count},
+			Tags:                        tags,
+			VerifyDistributionCountOnly: true}},
+	}
+}
+
+// WithResource sets the resource of the metric.
+func (m Metric) WithResource(r *resource.Resource) Metric {
+	m.Resource = r
+	return m
+}
+
+// AssertMetric verifies that the metrics have the specified values. Note that
+// this method will spuriously fail if there are multiple metrics with the same
+// name on different Meters. Calls EnsureRecorded internally before fetching the
+// batch of metrics.
+func AssertMetric(t *testing.T, values ...Metric) {
+	t.Helper()
+	EnsureRecorded()
+	for _, v := range values {
+		if diff := cmp.Diff(v, GetOneMetric(v.Name)); diff != "" {
+			t.Error("Wrong metric (-want +got):", diff)
+		}
+	}
+}
+
+// AssertMetricExists verifies that at least one metric values has been reported for
+// each of metric names.
+// Calls EnsureRecorded internally before fetching the batch of metrics.
+func AssertMetricExists(t *testing.T, names ...string) {
+	metrics := make([]Metric, 0, len(names))
+	for _, n := range names {
+		metrics = append(metrics, Metric{Name: n})
+	}
+	AssertMetric(t, metrics...)
+}
+
+// AssertNoMetric verifies that no metrics have been reported for any of the
+// metric names.
+// Calls EnsureRecorded internally before fetching the batch of metrics.
+func AssertNoMetric(t *testing.T, names ...string) {
+	t.Helper()
+	EnsureRecorded()
+	for _, name := range names {
+		if m := GetMetric(name); len(m) != 0 {
+			t.Error("Found unexpected data for:", m)
+		}
+	}
+}
+
+// VisitFloat64Value implements metricdata.ValueVisitor.
+func (v *Value) VisitFloat64Value(f float64) {
+	v.Float64 = &f
+	v.Int64 = nil
+	v.Distribution = nil
+}
+
+// VisitInt64Value implements metricdata.ValueVisitor.
+func (v *Value) VisitInt64Value(i int64) {
+	v.Int64 = &i
+	v.Float64 = nil
+	v.Distribution = nil
+}
+
+// VisitDistributionValue implements metricdata.ValueVisitor.
+func (v *Value) VisitDistributionValue(d *metricdata.Distribution) {
+	v.Distribution = d
+	v.Int64 = nil
+	v.Float64 = nil
+}
+
+// VisitSummaryValue implements metricdata.ValueVisitor.
+func (v *Value) VisitSummaryValue(*metricdata.Summary) {
+	panic("Attempted to fetch summary value, which we never use!")
+}
+
+// Equal provides a contract for use with github.com/google/go-cmp/cmp. Due to
+// the reflection in cmp, it only works if the type of the two arguments to cmp
+// are the same.
+func (m Metric) Equal(other Metric) bool {
+	if m.Name != other.Name {
+		return false
+	}
+	if (m.Unit != "" || m.VerifyMetadata) && (other.Unit != "" || other.VerifyMetadata) {
+		if m.Unit != other.Unit {
+			return false
+		}
+	}
+	if m.VerifyMetadata && other.VerifyMetadata {
+		if m.Type != other.Type {
+			return false
+		}
+	}
+
+	if (m.Resource != nil || m.VerifyResource) && (other.Resource != nil || other.VerifyResource) {
+		if !cmp.Equal(m.Resource, other.Resource) {
+			return false
+		}
+	}
+
+	if len(m.Values) > 0 && len(other.Values) > 0 {
+		if len(m.Values) != len(other.Values) {
+			return false
+		}
+		myValues := make(map[string]Value, len(m.Values))
+		for _, v := range m.Values {
+			myValues[tagsToString(v.Tags)] = v
+		}
+		for _, v := range other.Values {
+			myV, ok := myValues[tagsToString(v.Tags)]
+			if !ok || !myV.Equal(v) {
+				return false
+			}
+		}
+	}
+
+	return true
+}
+
+// Equal provides a contract for github.com/google/go-cmp/cmp. It compares two
+// values, including deep comparison of Distributions. (Exemplars are
+// intentional not included in the comparison, but other fields are considered).
+func (v Value) Equal(other Value) bool {
+	if len(v.Tags) != len(other.Tags) {
+		return false
+	}
+	for k, v := range v.Tags {
+		if v != other.Tags[k] {
+			return false
+		}
+	}
+	if v.Int64 != nil {
+		return other.Int64 != nil && *v.Int64 == *other.Int64
+	}
+	if v.Float64 != nil {
+		return other.Float64 != nil && *v.Float64 == *other.Float64
+	}
+
+	if v.Distribution != nil {
+		if other.Distribution == nil {
+			return false
+		}
+		if v.Distribution.Count != other.Distribution.Count {
+			return false
+		}
+		if v.VerifyDistributionCountOnly || other.VerifyDistributionCountOnly {
+			return true
+		}
+		if v.Distribution.Sum != other.Distribution.Sum {
+			return false
+		}
+		if v.Distribution.SumOfSquaredDeviation != other.Distribution.SumOfSquaredDeviation {
+			return false
+		}
+		if v.Distribution.BucketOptions != nil {
+			if other.Distribution.BucketOptions == nil {
+				return false
+			}
+			for i, bo := range v.Distribution.BucketOptions.Bounds {
+				if bo != other.Distribution.BucketOptions.Bounds[i] {
+					return false
+				}
+			}
+		}
+		for i, b := range v.Distribution.Buckets {
+			if b.Count != other.Distribution.Buckets[i].Count {
+				return false
+			}
+		}
+	}
+
+	return true
+}
+
+func tagsToString(tags map[string]string) string {
+	pairs := make([]string, 0, len(tags))
+	for k, v := range tags {
+		pairs = append(pairs, fmt.Sprintf("%s=%s", k, v))
+	}
+	sort.Strings(pairs)
+	return strings.Join(pairs, ",")
+}