Repository: knative-sandbox/eventing-kafka
Pull request #642 (2021-05-21 14:06:56)

[main] Upgrade to latest dependencies
Cron -knative-prow-robot

/cc knative-sandbox/channel-wg-leads
/assign knative-sandbox/channel-wg-leads

Produced by: knative-sandbox/knobots/actions/update-deps

Details:
```
bumping knative.dev/eventing 40fa622...8fbacd0:
  > 8fbacd0 mt broker dispatch error log (# 5426)
  > cca60b3 [QE] Allow to change SUT for upgrade tests (# 5321)
```

knative-prow-robot: [APPROVALNOTIFIER] This PR is **NOT APPROVED**

This pull-request has been approved by:
To complete the [pull request process](https://git.k8s.io/community/contributors/guide/owners.md#the-code-review-process), please assign **davyodom** after the PR has been reviewed.
You can assign the PR to them by writing `/assign @davyodom` in a comment when ready.

The full list of commands accepted by this bot can be found [here](https://go.k8s.io/bot-commands?repo=knative-sandbox%2Feventing-kafka).

<details open>
Needs approval from an approver in each of these files:

- **[OWNERS](https://github.com/knative-sandbox/eventing-kafka/blob/main/OWNERS)**

Approvers can indicate their approval by writing `/approve` in a comment
Approvers can cancel approval by writing `/approve cancel` in a comment
</details>
<!-- META={"approvers":["davyodom"]} -->
knative-prow-robot: @knative-automation: The following test **failed**, say `/retest` to rerun all failed tests:

Test name | Commit | Details | Rerun command
--- | --- | --- | ---
pull-knative-sandbox-eventing-kafka-go-coverage | 9d8b133a22cc4b81b713ef0c9f53fa6aa476ff74 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/642/pull-knative-sandbox-eventing-kafka-go-coverage/1395742921587691520) | `/test pull-knative-sandbox-eventing-kafka-go-coverage`

[Full PR test history](https://gubernator.knative.dev/pr/knative-sandbox_eventing-kafka/642). [Your PR dashboard](https://gubernator.knative.dev/pr/knative-automation).

<details>

Instructions for interacting with me using PR comments are available [here](https://git.k8s.io/community/contributors/guide/pull-requests.md).  If you have questions or suggestions related to my behavior, please file an issue against the [kubernetes/test-infra](https://github.com/kubernetes/test-infra/issues/new?title=Prow%20issue:) repository. I understand the commands that are listed [here](https://go.k8s.io/bot-commands).
</details>
<!-- test report -->
codecov[bot]: # [Codecov](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/642?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) Report
> Merging [#642](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/642?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) (9d8b133) into [main](https://codecov.io/gh/knative-sandbox/eventing-kafka/commit/fbb9090ab2a9407cdcbc562d54429c20518930db?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) (fbb9090) will **not change** coverage.
> The diff coverage is `n/a`.

[![Impacted file tree graph](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/642/graphs/tree.svg?width=650&height=150&src=pr&token=W765CT93QC&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox)](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/642?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox)

```diff
@@           Coverage Diff           @@
##             main     #642   +/-   ##
=======================================
  Coverage   72.83%   72.83%           
=======================================
  Files         128      128           
  Lines        5555     5555           
=======================================
  Hits         4046     4046           
  Misses       1285     1285           
  Partials      224      224           
```



------

[Continue to review full report at Codecov](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/642?src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox).
> **Legend** - [Click here to learn more](https://docs.codecov.io/docs/codecov-delta?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox)
> `Δ = absolute <relative> (impact)`, `ø = not affected`, `? = missing data`
> Powered by [Codecov](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/642?src=pr&el=footer&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox). Last update [fbb9090...9d8b133](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/642?src=pr&el=lastupdated&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox).

knative-prow-robot: @knative-automation: The following tests **failed**, say `/retest` to rerun all failed tests:

Test name | Commit | Details | Rerun command
--- | --- | --- | ---
pull-knative-sandbox-eventing-kafka-go-coverage | 9d8b133a22cc4b81b713ef0c9f53fa6aa476ff74 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/642/pull-knative-sandbox-eventing-kafka-go-coverage/1395742921587691520) | `/test pull-knative-sandbox-eventing-kafka-go-coverage`
pull-knative-sandbox-eventing-kafka-integration-test-mt-source | 9d8b133a22cc4b81b713ef0c9f53fa6aa476ff74 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/642/pull-knative-sandbox-eventing-kafka-integration-test-mt-source/1395742921516388352) | `/test pull-knative-sandbox-eventing-kafka-integration-test-mt-source`

[Full PR test history](https://gubernator.knative.dev/pr/knative-sandbox_eventing-kafka/642). [Your PR dashboard](https://gubernator.knative.dev/pr/knative-automation).

<details>

Instructions for interacting with me using PR comments are available [here](https://git.k8s.io/community/contributors/guide/pull-requests.md).  If you have questions or suggestions related to my behavior, please file an issue against the [kubernetes/test-infra](https://github.com/kubernetes/test-infra/issues/new?title=Prow%20issue:) repository. I understand the commands that are listed [here](https://go.k8s.io/bot-commands).
</details>
<!-- test report -->
knative-prow-robot: @knative-automation: The following tests **failed**, say `/retest` to rerun all failed tests:

Test name | Commit | Details | Rerun command
--- | --- | --- | ---
pull-knative-sandbox-eventing-kafka-go-coverage | 9d8b133a22cc4b81b713ef0c9f53fa6aa476ff74 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/642/pull-knative-sandbox-eventing-kafka-go-coverage/1395742921587691520) | `/test pull-knative-sandbox-eventing-kafka-go-coverage`
pull-knative-sandbox-eventing-kafka-integration-test-mt-source | 9d8b133a22cc4b81b713ef0c9f53fa6aa476ff74 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/642/pull-knative-sandbox-eventing-kafka-integration-test-mt-source/1395742921516388352) | `/test pull-knative-sandbox-eventing-kafka-integration-test-mt-source`
pull-knative-sandbox-eventing-kafka-upgrade-tests | 9d8b133a22cc4b81b713ef0c9f53fa6aa476ff74 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/642/pull-knative-sandbox-eventing-kafka-upgrade-tests/1395742921533165568) | `/test pull-knative-sandbox-eventing-kafka-upgrade-tests`

[Full PR test history](https://gubernator.knative.dev/pr/knative-sandbox_eventing-kafka/642). [Your PR dashboard](https://gubernator.knative.dev/pr/knative-automation).

<details>

Instructions for interacting with me using PR comments are available [here](https://git.k8s.io/community/contributors/guide/pull-requests.md).  If you have questions or suggestions related to my behavior, please file an issue against the [kubernetes/test-infra](https://github.com/kubernetes/test-infra/issues/new?title=Prow%20issue:) repository. I understand the commands that are listed [here](https://go.k8s.io/bot-commands).
</details>
<!-- test report -->
lionelvillard: /test pull-knative-sandbox-eventing-kafka-integration-test-mt-source
/test pull-knative-sandbox-eventing-kafka-upgrade-tests

Patch:
diff --git a/go.mod b/go.mod
index aa98a9380..c592b979f 100644
--- a/go.mod
+++ b/go.mod
@@ -34,7 +34,7 @@ require (
 	k8s.io/client-go v0.19.7
 	k8s.io/utils v0.0.0-20200729134348-d5654de09c73
 	knative.dev/control-protocol v0.0.0-20210520180517-3214e528bf78
-	knative.dev/eventing v0.23.1-0.20210520220117-40fa622c08a4
+	knative.dev/eventing v0.23.1-0.20210521082618-8fbacd01b514
 	knative.dev/hack v0.0.0-20210428122153-93ad9129c268
 	knative.dev/networking v0.0.0-20210520062317-b2552eca09f7
 	knative.dev/pkg v0.0.0-20210520062216-e749d6a2ad0e
diff --git a/go.sum b/go.sum
index f202712d3..09cc42534 100644
--- a/go.sum
+++ b/go.sum
@@ -1311,8 +1311,8 @@ k8s.io/utils v0.0.0-20200729134348-d5654de09c73 h1:uJmqzgNWG7XyClnU/mLPBWwfKKF1K
 k8s.io/utils v0.0.0-20200729134348-d5654de09c73/go.mod h1:jPW/WVKK9YHAvNhRxK0md/EJ228hCsBRufyofKtW8HA=
 knative.dev/control-protocol v0.0.0-20210520180517-3214e528bf78 h1:OzXUGWQFLxO+O00rIUmB5Ir7ZiHg5AVlBxgnVdYzves=
 knative.dev/control-protocol v0.0.0-20210520180517-3214e528bf78/go.mod h1:z0X6o/iPLE0cwK+YVDZSdPH/VtrJqbqSlZXBWiWEv+I=
-knative.dev/eventing v0.23.1-0.20210520220117-40fa622c08a4 h1:6gKN/qztFuGfKLEXOkDJ+F5cvU994aMJIqeXWNy4ci0=
-knative.dev/eventing v0.23.1-0.20210520220117-40fa622c08a4/go.mod h1:4WewoHOfKYH1nGMrFrZzNz4ZXnCy8WThWley770Zyjc=
+knative.dev/eventing v0.23.1-0.20210521082618-8fbacd01b514 h1:vXMnDVSuFSaRm7Hbz20lTHxbyLjHlp3dkixrLfCxXLg=
+knative.dev/eventing v0.23.1-0.20210521082618-8fbacd01b514/go.mod h1:4WewoHOfKYH1nGMrFrZzNz4ZXnCy8WThWley770Zyjc=
 knative.dev/hack v0.0.0-20210428122153-93ad9129c268 h1:lBIj9Epd9UQ55NEaHzAdY/UZbuaegCdGPKVC2+Z68Q0=
 knative.dev/hack v0.0.0-20210428122153-93ad9129c268/go.mod h1:PHt8x8yX5Z9pPquBEfIj0X66f8iWkWfR0S/sarACJrI=
 knative.dev/hack/schema v0.0.0-20210428122153-93ad9129c268/go.mod h1:ffjwmdcrH5vN3mPhO8RrF2KfNnbHeCE2C60A+2cv3U0=
diff --git a/pkg/common/scheduler/statefulset/scheduler.go b/pkg/common/scheduler/statefulset/scheduler.go
index 4ff729986..6e1126b3f 100644
--- a/pkg/common/scheduler/statefulset/scheduler.go
+++ b/pkg/common/scheduler/statefulset/scheduler.go
@@ -21,6 +21,7 @@ import (
 	"errors"
 	"math"
 	"sort"
+	"strconv"
 	"sync"
 	"time"
 
@@ -122,6 +123,19 @@ func NewStatefulSetScheduler(ctx context.Context,
 }
 
 func (s *StatefulSetScheduler) Schedule(vpod scheduler.VPod) ([]duckv1alpha1.Placement, error) {
+	placements, err := s.scheduleVPod(vpod)
+	if placements == nil {
+		return placements, err
+	}
+	sort.SliceStable(placements, func(i int, j int) bool {
+		podIOrdinal, _ := strconv.Atoi(placements[i].PodName[len(placements[i].PodName)-1:])
+		podJOrdinal, _ := strconv.Atoi(placements[j].PodName[len(placements[j].PodName)-1:])
+		return podIOrdinal < podJOrdinal
+	})
+	return placements, err
+}
+
+func (s *StatefulSetScheduler) scheduleVPod(vpod scheduler.VPod) ([]duckv1alpha1.Placement, error) {
 	s.lock.Lock()
 	defer s.lock.Unlock()
 
@@ -206,7 +220,7 @@ func (s *StatefulSetScheduler) Schedule(vpod scheduler.VPod) ([]duckv1alpha1.Pla
 
 func (s *StatefulSetScheduler) removeReplicas(diff int32, placements []duckv1alpha1.Placement) []duckv1alpha1.Placement {
 	newPlacements := make([]duckv1alpha1.Placement, 0, len(placements))
-	for i := 0; i < len(placements); i++ {
+	for i := len(placements) - 1; i > -1; i-- {
 		if diff >= placements[i].VReplicas {
 			// remove the entire placement
 			diff -= placements[i].VReplicas
diff --git a/vendor/modules.txt b/vendor/modules.txt
index 4cb090f18..28947e1a7 100644
--- a/vendor/modules.txt
+++ b/vendor/modules.txt
@@ -1105,7 +1105,7 @@ knative.dev/control-protocol/pkg/message
 knative.dev/control-protocol/pkg/network
 knative.dev/control-protocol/pkg/reconciler
 knative.dev/control-protocol/pkg/service
-# knative.dev/eventing v0.23.1-0.20210520220117-40fa622c08a4
+# knative.dev/eventing v0.23.1-0.20210521082618-8fbacd01b514
 ## explicit
 knative.dev/eventing/pkg/adapter/v2
 knative.dev/eventing/pkg/adapter/v2/test
@@ -1211,6 +1211,7 @@ knative.dev/eventing/test/test_images/wathola-forwarder
 knative.dev/eventing/test/test_images/wathola-receiver
 knative.dev/eventing/test/test_images/wathola-sender
 knative.dev/eventing/test/upgrade/prober
+knative.dev/eventing/test/upgrade/prober/sut
 knative.dev/eventing/test/upgrade/prober/wathola/client
 knative.dev/eventing/test/upgrade/prober/wathola/config
 knative.dev/eventing/test/upgrade/prober/wathola/event


Test patch:
diff --git a/pkg/common/scheduler/statefulset/scheduler_test.go b/pkg/common/scheduler/statefulset/scheduler_test.go
index 0d68a1afd..3df6250db 100644
--- a/pkg/common/scheduler/statefulset/scheduler_test.go
+++ b/pkg/common/scheduler/statefulset/scheduler_test.go
@@ -135,8 +135,8 @@ func TestStatefulsetScheduler(t *testing.T) {
 				{PodName: "statefulset-name-1", VReplicas: 10},
 			},
 			expected: []duckv1alpha1.Placement{
-				{PodName: "statefulset-name-0", VReplicas: 5},
-				{PodName: "statefulset-name-1", VReplicas: 10},
+				{PodName: "statefulset-name-0", VReplicas: 10},
+				{PodName: "statefulset-name-1", VReplicas: 5},
 			},
 		},
 		{
diff --git a/vendor/knative.dev/eventing/test/lib/resources/eventing.go b/vendor/knative.dev/eventing/test/lib/resources/eventing.go
index cf7667330..d4720d2b8 100644
--- a/vendor/knative.dev/eventing/test/lib/resources/eventing.go
+++ b/vendor/knative.dev/eventing/test/lib/resources/eventing.go
@@ -242,22 +242,31 @@ func WithDependencyAnnotationTrigger(dependencyAnnotation string) TriggerOption
 
 // WithSubscriberServiceRefForTrigger returns an option that adds a Subscriber Knative Service Ref for the given v1 Trigger.
 func WithSubscriberServiceRefForTrigger(name string) TriggerOption {
-	return func(t *eventingv1.Trigger) {
-		if name != "" {
-			t.Spec.Subscriber = duckv1.Destination{
-				Ref: KnativeRefForService(name, t.Namespace),
-			}
+	return WithSubscriberDestination(func(t *eventingv1.Trigger) duckv1.Destination {
+		return duckv1.Destination{
+			Ref: KnativeRefForService(name, t.Namespace),
 		}
-	}
+	})
 }
 
 // WithSubscriberURIForTrigger returns an option that adds a Subscriber URI for the given v1 Trigger.
 func WithSubscriberURIForTrigger(uri string) TriggerOption {
-	apisURI, _ := apis.ParseURL(uri)
-	return func(t *eventingv1.Trigger) {
-		t.Spec.Subscriber = duckv1.Destination{
+	return WithSubscriberDestination(func(t *eventingv1.Trigger) duckv1.Destination {
+		apisURI, _ := apis.ParseURL(uri)
+		return duckv1.Destination{
 			URI: apisURI,
 		}
+	})
+}
+
+// WithSubscriberDestination returns an option that adds a Subscriber for given
+// duckv1.Destination.
+func WithSubscriberDestination(destFactory func(t *eventingv1.Trigger) duckv1.Destination) TriggerOption {
+	return func(t *eventingv1.Trigger) {
+		dest := destFactory(t)
+		if dest.Ref != nil || dest.URI != nil {
+			t.Spec.Subscriber = dest
+		}
 	}
 }
 
diff --git a/vendor/knative.dev/eventing/test/upgrade/prober/config.toml b/vendor/knative.dev/eventing/test/upgrade/prober/config.toml
index ca5c1c7c9..20afa87d6 100644
--- a/vendor/knative.dev/eventing/test/upgrade/prober/config.toml
+++ b/vendor/knative.dev/eventing/test/upgrade/prober/config.toml
@@ -1,6 +1,6 @@
 # logLevel = 'DEBUG'
 [sender]
-address = '{{- .BrokerURL -}}'
+address = '{{- .Endpoint -}}'
 interval = {{ .Config.Interval.Nanoseconds }}
 [forwarder]
-target = 'http://wathola-receiver.{{- .Config.Namespace -}}.svc.cluster.local'
+target = 'http://wathola-receiver.{{- .Namespace -}}.svc.cluster.local'
diff --git a/vendor/knative.dev/eventing/test/upgrade/prober/configuration.go b/vendor/knative.dev/eventing/test/upgrade/prober/configuration.go
index a195f7d4f..ba1057afd 100644
--- a/vendor/knative.dev/eventing/test/upgrade/prober/configuration.go
+++ b/vendor/knative.dev/eventing/test/upgrade/prober/configuration.go
@@ -20,68 +20,77 @@ import (
 	"context"
 	"fmt"
 	"io/ioutil"
+	"os"
 	"path"
 	"runtime"
+	"strings"
 	"text/template"
 	"time"
 
 	"github.com/kelseyhightower/envconfig"
 	"github.com/wavesoftware/go-ensure"
-	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
-	eventingv1 "knative.dev/eventing/pkg/apis/eventing/v1"
-	testlib "knative.dev/eventing/test/lib"
-	"knative.dev/eventing/test/lib/duck"
 	"knative.dev/eventing/test/lib/resources"
-	"knative.dev/pkg/apis"
+	"knative.dev/eventing/test/upgrade/prober/sut"
+	duckv1 "knative.dev/pkg/apis/duck/v1"
 	pkgTest "knative.dev/pkg/test"
+	pkgupgrade "knative.dev/pkg/test/upgrade"
 )
 
 const (
-	defaultConfigName          = "wathola-config"
-	defaultConfigHomedirPath   = ".config/wathola"
-	defaultHomedir             = "/home/nonroot"
-	defaultConfigFilename      = "config.toml"
-	defaultWatholaEventsPrefix = "com.github.cardil.wathola"
-	defaultBrokerName          = "default"
-	defaultHealthEndpoint      = "/healthz"
-	defaultFinishedSleep       = 5 * time.Second
+	defaultConfigName        = "wathola-config"
+	defaultConfigHomedirPath = ".config/wathola"
+	defaultHomedir           = "/home/nonroot"
+	defaultConfigFilename    = "config.toml"
+	defaultHealthEndpoint    = "/healthz"
+	defaultFinishedSleep     = 5 * time.Second
 
 	Silence DuplicateAction = "silence"
 	Warn    DuplicateAction = "warn"
 	Error   DuplicateAction = "error"
+
+	prefix = "eventing_upgrade_tests"
+
+	// TODO(ksuszyns): Remove this val in next release
+	// Deprecated: use 'eventing_upgrade_tests' prefix instead
+	deprecatedPrefix = "e2e_upgrade_tests"
 )
 
 // DuplicateAction is the action to take in case of duplicated events
 type DuplicateAction string
 
-var eventTypes = []string{"step", "finished"}
-
 // Config represents a configuration for prober.
 type Config struct {
 	Wathola
-	Namespace     string
 	Interval      time.Duration
 	FinishedSleep time.Duration
 	Serving       ServingConfig
 	FailOnErrors  bool
 	OnDuplicate   DuplicateAction
-	BrokerOpts    []resources.BrokerOption
+	Ctx           context.Context
+	// BrokerOpts holds opts for broker.
+	// TODO(ksuszyns): Remove this opt in next release
+	// Deprecated: use Wathola.SystemUnderTest instead.
+	BrokerOpts []resources.BrokerOption
+	// Namespace holds namespace in which test is about to be executed.
+	// TODO(ksuszyns): Remove this opt in next release
+	// Deprecated: namespace is about to be taken from testlib.Client created by
+	// NewRunner
+	Namespace string
 }
 
 // Wathola represents options related strictly to wathola testing tool.
 type Wathola struct {
-	ConfigMap
+	ConfigToml
 	ImageResolver
-	EventsTypePrefix string
-	HealthEndpoint   string
-	BrokerName       string
+	SystemUnderTest sut.SystemUnderTest
+	HealthEndpoint  string
 }
 
 // ImageResolver will resolve the container image for given component.
 type ImageResolver func(component string) string
 
-// ConfigMap represents options of wathola config toml file.
-type ConfigMap struct {
+// ConfigToml represents options of wathola config toml file.
+type ConfigToml struct {
 	// ConfigTemplate is a template file that will be compiled to the configmap
 	ConfigTemplate   string
 	ConfigMapName    string
@@ -95,123 +104,130 @@ type ServingConfig struct {
 	ScaleToZero bool
 }
 
+// NewConfigOrFail will create a prober.Config or fail trying.
+func NewConfigOrFail(c pkgupgrade.Context) *Config {
+	errSink := func(err error) {
+		c.T.Fatal(err)
+	}
+	warnf := c.Log.Warnf
+	return newConfig(errSink, warnf)
+}
+
 // NewConfig creates a new configuration object with default values filled in.
 // Values can be influenced by kelseyhightower/envconfig with
-// `e2e_upgrade_tests` prefix.
-func NewConfig(namespace string) *Config {
+// `eventing_upgrade_tests` prefix.
+// TODO(ksuszyns): Remove this func in next release
+// Deprecated: use NewContinualVerification or NewConfigOrFail
+func NewConfig(namespace ...string) *Config {
+	errSink := func(err error) {
+		ensure.NoError(err)
+	}
+	warnf := func(template string, args ...interface{}) {
+		_, err := fmt.Fprintf(os.Stderr, template, args)
+		ensure.NoError(err)
+	}
+	config := newConfig(errSink, warnf)
+	if len(namespace) > 0 {
+		config.Namespace = strings.Join(namespace, ",")
+	}
+	return config
+}
+
+func newConfig(
+	errSink func(err error), warnf func(template string, args ...interface{}),
+) *Config {
 	config := &Config{
-		Namespace:     "",
 		Interval:      Interval,
 		FinishedSleep: defaultFinishedSleep,
 		FailOnErrors:  true,
 		OnDuplicate:   Warn,
-		BrokerOpts:    make([]resources.BrokerOption, 0),
+		Ctx:           context.Background(),
 		Serving: ServingConfig{
 			Use:         false,
 			ScaleToZero: true,
 		},
 		Wathola: Wathola{
 			ImageResolver: pkgTest.ImagePath,
-			ConfigMap: ConfigMap{
+			ConfigToml: ConfigToml{
 				ConfigTemplate:   defaultConfigFilename,
 				ConfigMapName:    defaultConfigName,
 				ConfigMountPoint: fmt.Sprintf("%s/%s", defaultHomedir, defaultConfigHomedirPath),
 				ConfigFilename:   defaultConfigFilename,
 			},
-			EventsTypePrefix: defaultWatholaEventsPrefix,
-			HealthEndpoint:   defaultHealthEndpoint,
-			BrokerName:       defaultBrokerName,
+			HealthEndpoint:  defaultHealthEndpoint,
+			SystemUnderTest: sut.NewDefault(),
 		},
 	}
 
-	// FIXME: remove while fixing https://github.com/knative/eventing/issues/2665
-	config.FailOnErrors = false
+	if err := envconfig.Process(prefix, config); err != nil {
+		errSink(err)
+	}
 
-	err := envconfig.Process("e2e_upgrade_tests", config)
-	ensure.NoError(err)
-	config.Namespace = namespace
+	// TODO(ksuszyns): Remove this block in next release
+	for _, enventry := range os.Environ() {
+		if strings.HasPrefix(strings.ToLower(enventry), deprecatedPrefix) {
+			warnf(
+				"DEPRECATED: using deprecated '%s' prefix. Use '%s' instead.",
+				deprecatedPrefix, prefix)
+			if err := envconfig.Process(deprecatedPrefix, config); err != nil {
+				errSink(err)
+			}
+			break
+		}
+	}
 	return config
 }
 
 func (p *prober) deployConfiguration() {
-	p.deployBroker()
-	p.deployConfigMap()
-	p.deployTriggers()
-}
-
-func (p *prober) deployBroker() {
-	p.client.CreateBrokerOrFail(p.config.BrokerName, p.config.BrokerOpts...)
-}
-
-func (p *prober) fetchBrokerURL() (*apis.URL, error) {
-	namespace := p.config.Namespace
-	p.log.Debugf("Fetching %s broker URL for ns %s",
-		p.config.BrokerName, namespace)
-	meta := resources.NewMetaResource(
-		p.config.BrokerName, p.config.Namespace, testlib.BrokerTypeMeta,
-	)
-	err := duck.WaitForResourceReady(p.client.Dynamic, meta)
-	if err != nil {
-		return nil, err
+	sc := sut.Context{
+		Ctx:    p.config.Ctx,
+		Log:    p.log,
+		Client: p.client,
 	}
-	broker, err := p.client.Eventing.EventingV1().Brokers(namespace).Get(
-		context.Background(), p.config.BrokerName, metav1.GetOptions{},
-	)
-	if err != nil {
-		return nil, err
+	ref := resources.KnativeRefForService(receiverName, p.client.Namespace)
+	if p.config.Serving.Use {
+		ref = resources.KnativeRefForKservice(forwarderName, p.client.Namespace)
 	}
-	url := broker.Status.Address.URL
-	p.log.Debugf("%s broker URL for ns %s is %v",
-		p.config.BrokerName, namespace, url)
-	return url, nil
+	dest := duckv1.Destination{Ref: ref}
+	s := p.config.SystemUnderTest
+	endpoint := s.Deploy(sc, dest)
+	p.client.Cleanup(func() {
+		if tr, ok := s.(sut.HasTeardown); ok {
+			tr.Teardown(sc)
+		}
+	})
+	p.deployConfigToml(endpoint)
 }
 
-func (p *prober) deployConfigMap() {
+func (p *prober) deployConfigToml(endpoint interface{}) {
 	name := p.config.ConfigMapName
-	p.log.Infof("Deploying config map: \"%s/%s\"", p.config.Namespace, name)
-	brokerURL, err := p.fetchBrokerURL()
-	ensure.NoError(err)
-	configData := p.compileTemplate(p.config.ConfigTemplate, brokerURL)
-	p.client.CreateConfigMapOrFail(name, p.config.Namespace, map[string]string{
+	p.log.Infof("Deploying config map: \"%s/%s\"", p.client.Namespace, name)
+	configData := p.compileTemplate(p.config.ConfigTemplate, endpoint)
+	p.client.CreateConfigMapOrFail(name, p.client.Namespace, map[string]string{
 		p.config.ConfigFilename: configData,
 	})
 }
 
-func (p *prober) deployTriggers() {
-	for _, eventType := range eventTypes {
-		name := fmt.Sprintf("wathola-trigger-%v", eventType)
-		fullType := fmt.Sprintf("%v.%v", p.config.EventsTypePrefix, eventType)
-		subscriberOption := resources.WithSubscriberServiceRefForTrigger(receiverName)
-		if p.config.Serving.Use {
-			subscriberOption = resources.WithSubscriberKServiceRefForTrigger(forwarderName)
-		}
-		p.client.CreateTriggerOrFail(name,
-			resources.WithBroker(p.config.BrokerName),
-			resources.WithAttributesTriggerFilter(
-				eventingv1.TriggerAnyFilter,
-				fullType,
-				map[string]interface{}{},
-			),
-			subscriberOption,
-		)
-	}
-}
-
-func (p *prober) compileTemplate(templateName string, brokerURL fmt.Stringer) string {
+func (p *prober) compileTemplate(templateName string, endpoint interface{}) string {
 	_, filename, _, _ := runtime.Caller(0)
 	templateFilepath := path.Join(path.Dir(filename), templateName)
 	templateBytes, err := ioutil.ReadFile(templateFilepath)
-	ensure.NoError(err)
+	p.ensureNoError(err)
 	tmpl, err := template.New(templateName).Parse(string(templateBytes))
-	ensure.NoError(err)
+	p.ensureNoError(err)
 	var buff bytes.Buffer
 	data := struct {
 		*Config
+		Namespace string
+		// Deprecated: use Endpoint
 		BrokerURL string
+		Endpoint  interface{}
 	}{
 		p.config,
-		brokerURL.String(),
+		p.client.Namespace,
+		fmt.Sprintf("%v", endpoint),
+		endpoint,
 	}
-	ensure.NoError(tmpl.Execute(&buff, data))
+	p.ensureNoError(tmpl.Execute(&buff, data))
 	return buff.String()
 }
diff --git a/vendor/knative.dev/eventing/test/upgrade/prober/continual.go b/vendor/knative.dev/eventing/test/upgrade/prober/continual.go
new file mode 100644
index 000000000..237fdb582
--- /dev/null
+++ b/vendor/knative.dev/eventing/test/upgrade/prober/continual.go
@@ -0,0 +1,59 @@
+/*
+Copyright 2021 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 prober
+
+import (
+	testlib "knative.dev/eventing/test/lib"
+	pkgupgrade "knative.dev/pkg/test/upgrade"
+)
+
+// Configurator will customize default config.
+type Configurator func(config *Config) error
+
+// ContinualVerificationOptions holds options for NewContinualVerification func.
+type ContinualVerificationOptions struct {
+	Configurators []Configurator
+	ClientOptions []testlib.SetupClientOption
+}
+
+// NewContinualVerification will create a new continual verification operation
+// that will verify that SUT is propagating events well through the test
+// process. It's a general pattern that can be used to validate upgrade or
+// performance testing.
+func NewContinualVerification(
+	name string, opts ContinualVerificationOptions,
+) pkgupgrade.BackgroundOperation {
+	var config *Config
+	var runner Runner
+	setup := func(c pkgupgrade.Context) {
+		config = NewConfigOrFail(c)
+		if len(opts.Configurators) > 0 {
+			for _, fn := range opts.Configurators {
+				err := fn(config)
+				if err != nil {
+					c.T.Fatal(err)
+				}
+			}
+		}
+		runner = NewRunner(config, opts.ClientOptions...)
+		runner.Setup(c)
+	}
+	verify := func(c pkgupgrade.Context) {
+		runner.Verify(c)
+	}
+	return pkgupgrade.NewBackgroundVerification(name, setup, verify)
+}
diff --git a/vendor/knative.dev/eventing/test/upgrade/prober/errors.go b/vendor/knative.dev/eventing/test/upgrade/prober/errors.go
new file mode 100644
index 000000000..e70da17bd
--- /dev/null
+++ b/vendor/knative.dev/eventing/test/upgrade/prober/errors.go
@@ -0,0 +1,23 @@
+/*
+Copyright 2021 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 prober
+
+func (p *prober) ensureNoError(err error) {
+	if err != nil {
+		p.client.T.Fatal(err)
+	}
+}
diff --git a/vendor/knative.dev/eventing/test/upgrade/prober/forwarder.go b/vendor/knative.dev/eventing/test/upgrade/prober/forwarder.go
index 3406254ca..14965fafb 100644
--- a/vendor/knative.dev/eventing/test/upgrade/prober/forwarder.go
+++ b/vendor/knative.dev/eventing/test/upgrade/prober/forwarder.go
@@ -16,10 +16,8 @@
 package prober
 
 import (
-	"context"
 	"fmt"
 
-	"github.com/wavesoftware/go-ensure"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
 	testlib "knative.dev/eventing/test/lib"
@@ -31,12 +29,13 @@ var (
 	forwarderName = "wathola-forwarder"
 )
 
-func (p *prober) deployForwarder(ctx context.Context) {
+func (p *prober) deployForwarder() {
 	p.log.Infof("Deploy forwarder knative service: %v", forwarderName)
 	serving := p.client.Dynamic.Resource(resources.KServicesGVR).Namespace(p.client.Namespace)
 	service := p.forwarderKService(forwarderName, p.client.Namespace)
-	_, err := serving.Create(context.Background(), service, metav1.CreateOptions{})
-	ensure.NoError(err)
+	if _, err := serving.Create(p.config.Ctx, service, metav1.CreateOptions{}); err != nil {
+		p.client.T.Fatal(err)
+	}
 
 	sc := p.servingClient()
 	testlib.WaitFor(fmt.Sprintf("forwarder ksvc be ready: %v", forwarderName), func() error {
@@ -45,7 +44,7 @@ func (p *prober) deployForwarder(ctx context.Context) {
 
 	if p.config.Serving.ScaleToZero {
 		testlib.WaitFor(fmt.Sprintf("forwarder scales to zero: %v", forwarderName), func() error {
-			return duck.WaitForKServiceScales(ctx, sc, forwarderName, p.client.Namespace, func(scale int) bool {
+			return duck.WaitForKServiceScales(p.config.Ctx, sc, forwarderName, p.client.Namespace, func(scale int) bool {
 				return scale == 0
 			})
 		})
@@ -55,8 +54,8 @@ func (p *prober) deployForwarder(ctx context.Context) {
 func (p *prober) removeForwarder() {
 	p.log.Infof("Remove forwarder knative service: %v", forwarderName)
 	serving := p.client.Dynamic.Resource(resources.KServicesGVR).Namespace(p.client.Namespace)
-	err := serving.Delete(context.Background(), forwarderName, metav1.DeleteOptions{})
-	ensure.NoError(err)
+	err := serving.Delete(p.config.Ctx, forwarderName, metav1.DeleteOptions{})
+	p.ensureNoError(err)
 }
 
 func (p *prober) forwarderKService(name, namespace string) *unstructured.Unstructured {
diff --git a/vendor/knative.dev/eventing/test/upgrade/prober/prober.go b/vendor/knative.dev/eventing/test/upgrade/prober/prober.go
index bc6623bf5..201dc6862 100644
--- a/vendor/knative.dev/eventing/test/upgrade/prober/prober.go
+++ b/vendor/knative.dev/eventing/test/upgrade/prober/prober.go
@@ -17,63 +17,144 @@ package prober
 
 import (
 	"context"
+	"reflect"
 	"testing"
 	"time"
 
-	"github.com/wavesoftware/go-ensure"
 	"go.uber.org/zap"
 	testlib "knative.dev/eventing/test/lib"
 	"knative.dev/eventing/test/lib/resources"
+	"knative.dev/eventing/test/upgrade/prober/sut"
+	pkgupgrade "knative.dev/pkg/test/upgrade"
 )
 
 var (
-	// FIXME: Interval is set to 200 msec, as lower values will result in errors: knative/eventing#2357
-	// Interval = 10 * time.Millisecond
-	Interval = 200 * time.Millisecond
+	// Interval is used to send events in specific rate.
+	Interval = 10 * time.Millisecond
 )
 
-// Prober is the interface for a prober, which checks the result of the probes when stopped.
+// Prober is the interface for a prober, which checks the result of the probes
+// when stopped.
+// TODO(ksuszyns): Remove this interface in next release
+// Deprecated: use Runner instead, create it with NewRunner func.
 type Prober interface {
 	// Verify will verify prober state after finished has been send
-	Verify(ctx context.Context) ([]error, int, error)
+	Verify() ([]error, int)
 
 	// Finish send finished event
-	Finish(ctx context.Context)
+	Finish()
 
 	// ReportErrors will reports found errors in proper way
-	ReportErrors(t *testing.T, errors []error)
+	ReportErrors(errors []error)
+}
+
+// Runner will run continual verification with provided configuration.
+type Runner interface {
+	// Setup will start a continual prober in background.
+	Setup(ctx pkgupgrade.Context)
 
-	// deploy a prober to a cluster
-	deploy(ctx context.Context)
-	// remove a prober from cluster
-	remove()
+	// Verify will verify that all sent events propagated at least once.
+	Verify(ctx pkgupgrade.Context)
 }
 
-// RunEventProber starts a single Prober of the given domain.
-func RunEventProber(ctx context.Context, log *zap.SugaredLogger, client *testlib.Client, config *Config) Prober {
-	pm := newProber(log, client, config)
-	pm.deploy(ctx)
-	return pm
+// NewRunner will create a runner compatible with NewContinualVerification
+// func.
+func NewRunner(config *Config, options ...testlib.SetupClientOption) Runner {
+	return &probeRunner{
+		prober:  &prober{config: config},
+		options: options,
+	}
 }
 
-// AssertEventProber will send finish event and then verify if all events propagated well
-func AssertEventProber(ctx context.Context, t *testing.T, prober Prober) {
-	prober.Finish(ctx)
-	defer prober.remove()
+type probeRunner struct {
+	*prober
+	options []testlib.SetupClientOption
+}
 
-	waitAfterFinished(prober)
+func (p *probeRunner) Setup(ctx pkgupgrade.Context) {
+	p.validate(ctx)
+	p.log = ctx.Log
+	p.client = testlib.Setup(ctx.T, false, p.options...)
+	p.deploy()
+}
 
-	eventErrs, eventCount, err := prober.Verify(ctx)
-	if err != nil {
-		t.Fatal("fetch error:", err)
+func (p *probeRunner) Verify(ctx pkgupgrade.Context) {
+	if p.client == nil {
+		ctx.T.Fatal("prober isn't initiated (client is nil)")
+		return
 	}
-	if len(eventErrs) == 0 {
-		t.Logf("All %d events propagated well", eventCount)
+	// use T from new test
+	p.client.T = ctx.T
+	t := ctx.T
+	defer testlib.TearDown(p.client)
+	defer p.remove()
+	p.Finish()
+	waitAfterFinished(p.prober)
+
+	errors, events := p.prober.Verify()
+	if len(errors) == 0 {
+		t.Logf("All %d events propagated well", events)
 	} else {
 		t.Logf("There were %d events propagated, but %d errors occurred. "+
-			"Listing them below.", eventCount, len(eventErrs))
+			"Listing them below.", events, len(errors))
+	}
+
+	p.ReportErrors(errors)
+}
+
+func (p *probeRunner) validate(ctx pkgupgrade.Context) {
+	if p.config.Namespace != "" {
+		ctx.Log.Warnf(
+			"DEPRECATED: namespace set in Config: %s. Ignoring it.",
+			p.client.Namespace)
+	}
+	if len(p.config.BrokerOpts) > 0 {
+		ctx.Log.Warn(
+			"DEPRECATED: BrokerOpts set in Config. Use custom SystemUnderTest")
+		if reflect.ValueOf(p.config.Wathola.SystemUnderTest) == reflect.ValueOf(sut.NewDefault) {
+			bt := sut.NewDefault().(*sut.BrokerAndTriggers)
+			bt.Opts = append(bt.Opts, p.config.BrokerOpts...)
+			p.config.Wathola.SystemUnderTest = bt
+		} else {
+			ctx.T.Fatal("Can't use given BrokerOpts, as custom SUT is used as " +
+				"well. Drop using BrokerOpts in favor of custom SUT.")
+		}
 	}
-	prober.ReportErrors(t, eventErrs)
+}
+
+// RunEventProber starts a single Prober of the given domain.
+// TODO(ksuszyns): Remove this func in next release
+// Deprecated: use NewRunner func instead.
+func RunEventProber(ctx context.Context, log *zap.SugaredLogger, client *testlib.Client, config *Config) Prober {
+	log.Warn("prober.RunEventProber is deprecated. Use NewRunner instead.")
+	config.Ctx = ctx
+	p := &prober{
+		log:    log,
+		client: client,
+		config: config,
+	}
+	p.deploy()
+	return p
+}
+
+// AssertEventProber will send finish event and then verify if all events
+// propagated well.
+// TODO(ksuszyns): Remove this func in next release
+// Deprecated: use NewRunner func instead.
+func AssertEventProber(ctx context.Context, t *testing.T, probe Prober) {
+	t.Log("WARN: prober.AssertEventProber is deprecated. " +
+		"Use NewRunner instead.")
+	p := probe.(*prober)
+	p.client.T = t
+	p.config.Ctx = ctx
+	pr := &probeRunner{
+		prober:  p,
+		options: nil,
+	}
+	pr.Verify(pkgupgrade.Context{
+		T:   t,
+		Log: p.log,
+	})
 }
 
 type prober struct {
@@ -89,7 +170,8 @@ func (p *prober) servingClient() resources.ServingClient {
 	}
 }
 
-func (p *prober) ReportErrors(t *testing.T, errors []error) {
+func (p *prober) ReportErrors(errors []error) {
+	t := p.client.T
 	for _, err := range errors {
 		if p.config.FailOnErrors {
 			t.Error(err)
@@ -105,17 +187,17 @@ func (p *prober) ReportErrors(t *testing.T, errors []error) {
 	}
 }
 
-func (p *prober) deploy(ctx context.Context) {
+func (p *prober) deploy() {
 	p.log.Infof("Using namespace for probe testing: %v", p.client.Namespace)
 	p.deployConfiguration()
-	p.deployReceiver(ctx)
+	p.deployReceiver()
 	if p.config.Serving.Use {
-		p.deployForwarder(ctx)
+		p.deployForwarder()
 	}
-	p.client.WaitForAllTestResourcesReadyOrFail(ctx)
+	p.client.WaitForAllTestResourcesReadyOrFail(p.config.Ctx)
 
-	p.deploySender(ctx)
-	ensure.NoError(testlib.AwaitForAll(p.log))
+	p.deploySender()
+	p.ensureNoError(testlib.AwaitForAll(p.log))
 	// allow sender to send at least some events, 2 sec wait
 	time.Sleep(2 * time.Second)
 	p.log.Infof("Prober is now sending events with interval of %v in "+
@@ -126,20 +208,10 @@ func (p *prober) remove() {
 	if p.config.Serving.Use {
 		p.removeForwarder()
 	}
-	ensure.NoError(p.client.Tracker.Clean(true))
-}
-
-func newProber(log *zap.SugaredLogger, client *testlib.Client, config *Config) Prober {
-	return &prober{
-		log:    log,
-		client: client,
-		config: config,
-	}
+	p.ensureNoError(p.client.Tracker.Clean(true))
 }
 
-func waitAfterFinished(p Prober) {
-	s := p.(*prober)
-	cfg := s.config
-	s.log.Infof("Waiting %v after sender finished...", cfg.FinishedSleep)
-	time.Sleep(cfg.FinishedSleep)
+func waitAfterFinished(p *prober) {
+	p.log.Infof("Waiting %v after sender finished...", p.config.FinishedSleep)
+	time.Sleep(p.config.FinishedSleep)
 }
diff --git a/vendor/knative.dev/eventing/test/upgrade/prober/receiver.go b/vendor/knative.dev/eventing/test/upgrade/prober/receiver.go
index f10145f8d..d55379881 100644
--- a/vendor/knative.dev/eventing/test/upgrade/prober/receiver.go
+++ b/vendor/knative.dev/eventing/test/upgrade/prober/receiver.go
@@ -16,10 +16,8 @@
 package prober
 
 import (
-	"context"
 	"fmt"
 
-	"github.com/wavesoftware/go-ensure"
 	appsv1 "k8s.io/api/apps/v1"
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -30,36 +28,32 @@ import (
 )
 
 var (
-	receiverName           = "wathola-receiver"
-	receiverNodePort int32 = -1
+	receiverName = "wathola-receiver"
 )
 
-func (p *prober) deployReceiver(ctx context.Context) {
-	p.deployReceiverDeployment(ctx)
-	p.deployReceiverService(ctx)
+func (p *prober) deployReceiver() {
+	p.deployReceiverDeployment()
+	p.deployReceiverService()
 }
 
-func (p *prober) deployReceiverDeployment(ctx context.Context) {
+func (p *prober) deployReceiverDeployment() {
 	p.log.Info("Deploy of receiver deployment: ", receiverName)
 	deployment := p.createReceiverDeployment()
-	_, err := p.client.Kube.AppsV1().
-		Deployments(deployment.Namespace).
-		Create(ctx, deployment, metav1.CreateOptions{})
-	ensure.NoError(err)
+	p.client.CreateDeploymentOrFail(deployment)
 
 	testlib.WaitFor(fmt.Sprint("receiver deployment be ready: ", receiverName), func() error {
 		return pkgTest.WaitForDeploymentScale(
-			ctx, p.client.Kube, receiverName, p.client.Namespace, 1,
+			p.config.Ctx, p.client.Kube, receiverName, p.client.Namespace, 1,
 		)
 	})
 }
 
-func (p *prober) deployReceiverService(ctx context.Context) {
+func (p *prober) deployReceiverService() {
 	p.log.Infof("Deploy of receiver service: %v", receiverName)
 	service := &corev1.Service{
 		ObjectMeta: metav1.ObjectMeta{
 			Name:      receiverName,
-			Namespace: p.config.Namespace,
+			Namespace: p.client.Namespace,
 		},
 		Spec: corev1.ServiceSpec{
 			Ports: []corev1.ServicePort{
@@ -76,22 +70,10 @@ func (p *prober) deployReceiverService(ctx context.Context) {
 			Selector: map[string]string{
 				"app": receiverName,
 			},
-			Type: corev1.ServiceTypeNodePort,
+			Type: corev1.ServiceTypeClusterIP,
 		},
 	}
-	created, err := p.client.Kube.CoreV1().Services(p.config.Namespace).
-		Create(ctx, service, metav1.CreateOptions{})
-	ensure.NoError(err)
-	for _, portSpec := range created.Spec.Ports {
-		if portSpec.Port == 80 {
-			receiverNodePort = portSpec.NodePort
-		}
-	}
-	if receiverNodePort == -1 {
-		panic(fmt.Errorf("couldn't find a node port for service: %v", receiverName))
-	} else {
-		p.log.Debugf("Node port for service: %v is %v", receiverName, receiverNodePort)
-	}
+	p.client.CreateServiceOrFail(service)
 }
 
 func (p *prober) createReceiverDeployment() *appsv1.Deployment {
@@ -99,7 +81,7 @@ func (p *prober) createReceiverDeployment() *appsv1.Deployment {
 	return &appsv1.Deployment{
 		ObjectMeta: metav1.ObjectMeta{
 			Name:      receiverName,
-			Namespace: p.config.Namespace,
+			Namespace: p.client.Namespace,
 		},
 		Spec: appsv1.DeploymentSpec{
 			Replicas: &replicas,
diff --git a/vendor/knative.dev/eventing/test/upgrade/prober/sender.go b/vendor/knative.dev/eventing/test/upgrade/prober/sender.go
index bde438fab..5f3bdd4ec 100644
--- a/vendor/knative.dev/eventing/test/upgrade/prober/sender.go
+++ b/vendor/knative.dev/eventing/test/upgrade/prober/sender.go
@@ -16,10 +16,8 @@
 package prober
 
 import (
-	"context"
 	"fmt"
 
-	"github.com/wavesoftware/go-ensure"
 	appsv1 "k8s.io/api/apps/v1"
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -29,13 +27,13 @@ import (
 
 var senderName = "wathola-sender"
 
-func (p *prober) deploySender(ctx context.Context) {
+func (p *prober) deploySender() {
 	p.log.Info("Deploy sender deployment: ", senderName)
 	var replicas int32 = 1
 	deployment := &appsv1.Deployment{
 		ObjectMeta: metav1.ObjectMeta{
 			Name:      senderName,
-			Namespace: p.config.Namespace,
+			Namespace: p.client.Namespace,
 		},
 		Spec: appsv1.DeploymentSpec{
 			Replicas: &replicas,
@@ -77,21 +75,21 @@ func (p *prober) deploySender(ctx context.Context) {
 
 	_, err := p.client.Kube.AppsV1().
 		Deployments(p.client.Namespace).
-		Create(ctx, deployment, metav1.CreateOptions{})
-	ensure.NoError(err)
+		Create(p.config.Ctx, deployment, metav1.CreateOptions{})
+	p.ensureNoError(err)
 
 	testlib.WaitFor(fmt.Sprint("sender deployment be ready: ", senderName), func() error {
 		return pkgTest.WaitForDeploymentScale(
-			ctx, p.client.Kube, senderName, p.client.Namespace, int(replicas),
+			p.config.Ctx, p.client.Kube, senderName, p.client.Namespace, int(replicas),
 		)
 	})
 }
 
-func (p *prober) removeSender(ctx context.Context) {
+func (p *prober) removeSender() {
 	p.log.Info("Remove of sender deployment: ", senderName)
 
 	err := p.client.Kube.AppsV1().
 		Deployments(p.client.Namespace).
-		Delete(ctx, senderName, metav1.DeleteOptions{})
-	ensure.NoError(err)
+		Delete(p.config.Ctx, senderName, metav1.DeleteOptions{})
+	p.ensureNoError(err)
 }
diff --git a/vendor/knative.dev/eventing/test/upgrade/prober/sut/broker.go b/vendor/knative.dev/eventing/test/upgrade/prober/sut/broker.go
new file mode 100644
index 000000000..a468a3b4a
--- /dev/null
+++ b/vendor/knative.dev/eventing/test/upgrade/prober/sut/broker.go
@@ -0,0 +1,141 @@
+/*
+Copyright 2021 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 sut
+
+import (
+	"fmt"
+
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	eventingduckv1 "knative.dev/eventing/pkg/apis/duck/v1"
+	eventingv1 "knative.dev/eventing/pkg/apis/eventing/v1"
+	testlib "knative.dev/eventing/test/lib"
+	"knative.dev/eventing/test/lib/duck"
+	"knative.dev/eventing/test/lib/resources"
+	"knative.dev/pkg/apis"
+	duckv1 "knative.dev/pkg/apis/duck/v1"
+)
+
+var (
+	retryCount    = int32(12)
+	backoffPolicy = eventingduckv1.BackoffPolicyExponential
+	backoffDelay  = "PT1S"
+)
+
+// BrokerAndTriggers will deploy a default broker and 2 triggers to route two types of
+// events to receiver.
+type BrokerAndTriggers struct {
+	Broker
+	Triggers
+}
+
+// Broker will hold settings for broker itself
+type Broker struct {
+	Name string
+	Opts []resources.BrokerOption
+}
+
+// Triggers will hold settings for triggers
+type Triggers struct {
+	Types []string
+}
+
+// NewBrokerAndTriggers will create default configuration for BrokerAndTriggers
+// based SUT.
+func NewBrokerAndTriggers() SystemUnderTest {
+	return &BrokerAndTriggers{
+		Broker: Broker{
+			Name: "sut",
+			Opts: []resources.BrokerOption{
+				resources.WithDeliveryForBroker(
+					&eventingduckv1.DeliverySpec{
+						Retry:         &retryCount,
+						BackoffPolicy: &backoffPolicy,
+						BackoffDelay:  &backoffDelay,
+					}),
+			},
+		},
+		Triggers: Triggers{
+			Types: eventTypes,
+		},
+	}
+}
+
+func (b *BrokerAndTriggers) Deploy(ctx Context, dest duckv1.Destination) interface{} {
+	b.deployBroker(ctx)
+	url := b.fetchURL(ctx)
+	b.deployTriggers(ctx, dest)
+	return url
+}
+
+func (b *BrokerAndTriggers) deployBroker(ctx Context) {
+	ctx.Client.CreateBrokerOrFail(b.Name, b.Broker.Opts...)
+}
+
+func (b *BrokerAndTriggers) fetchURL(ctx Context) *apis.URL {
+	namespace := ctx.Client.Namespace
+	ctx.Log.Debugf("Fetching \"%s\" broker URL for ns %s",
+		b.Name, namespace)
+	meta := resources.NewMetaResource(
+		b.Name, namespace, testlib.BrokerTypeMeta,
+	)
+	err := duck.WaitForResourceReady(ctx.Client.Dynamic, meta)
+	if err != nil {
+		ctx.T.Fatal(err)
+	}
+	broker, err := ctx.Client.Eventing.EventingV1().Brokers(namespace).Get(
+		ctx.Ctx, b.Name, metav1.GetOptions{},
+	)
+	if err != nil {
+		ctx.T.Fatal(err)
+	}
+	url := broker.Status.Address.URL
+	ctx.Log.Debugf("\"%s\" broker URL for ns %s is %v",
+		b.Name, namespace, url)
+	return url
+}
+
+func (b *BrokerAndTriggers) deployTriggers(ctx Context, dest duckv1.Destination) {
+	triggers := make([]*eventingv1.Trigger, 0, len(b.Triggers.Types))
+	for _, eventType := range b.Triggers.Types {
+		name := fmt.Sprintf("%s-%s", b.Name, eventType)
+		subscriberOption := resources.WithSubscriberDestination(func(t *eventingv1.Trigger) duckv1.Destination {
+			return dest
+		})
+		ctx.Log.Debugf("Creating trigger \"%s\" for type %s to route to %#v",
+			name, eventType, dest)
+		trgr := ctx.Client.CreateTriggerOrFail(
+			name,
+			resources.WithBroker(b.Name),
+			resources.WithAttributesTriggerFilter(
+				eventingv1.TriggerAnyFilter,
+				eventType,
+				map[string]interface{}{},
+			),
+			subscriberOption,
+		)
+		triggers = append(triggers, trgr)
+	}
+	for _, trgr := range triggers {
+		meta := resources.NewMetaResource(
+			trgr.Name, trgr.Namespace, testlib.TriggerTypeMeta,
+		)
+		err := duck.WaitForResourceReady(ctx.Client.Dynamic, meta)
+		if err != nil {
+			ctx.T.Fatal(err)
+		}
+	}
+}
diff --git a/vendor/knative.dev/eventing/test/upgrade/prober/sut/default.go b/vendor/knative.dev/eventing/test/upgrade/prober/sut/default.go
new file mode 100644
index 000000000..1f11edbdd
--- /dev/null
+++ b/vendor/knative.dev/eventing/test/upgrade/prober/sut/default.go
@@ -0,0 +1,22 @@
+/*
+Copyright 2021 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 sut
+
+// NewDefault will return a default SUT for this repo.
+func NewDefault() SystemUnderTest {
+	return NewBrokerAndTriggers()
+}
diff --git a/vendor/knative.dev/eventing/test/upgrade/prober/sut/types.go b/vendor/knative.dev/eventing/test/upgrade/prober/sut/types.go
new file mode 100644
index 000000000..8f1341019
--- /dev/null
+++ b/vendor/knative.dev/eventing/test/upgrade/prober/sut/types.go
@@ -0,0 +1,54 @@
+/*
+Copyright 2021 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 sut
+
+import (
+	"context"
+
+	"go.uber.org/zap"
+	testlib "knative.dev/eventing/test/lib"
+	watholaevent "knative.dev/eventing/test/upgrade/prober/wathola/event"
+	duckv1 "knative.dev/pkg/apis/duck/v1"
+)
+
+var eventTypes = []string{
+	watholaevent.Step{}.Type(),
+	watholaevent.Finished{}.Type(),
+}
+
+// SystemUnderTest (SUT) represents a system that we'd like to test with
+// continual prober.
+type SystemUnderTest interface {
+	// Deploy is responsible for deploying SUT and returning a URL to feed
+	// events into.
+	Deploy(ctx Context, destination duckv1.Destination) interface{}
+}
+
+// HasTeardown indicates that SystemUnderTest supports custom teardown that
+// exceeds regular teardown via usage of testlib.Tracker.
+type HasTeardown interface {
+	// Teardown will remove all deployed SUT resources.
+	Teardown(ctx Context)
+}
+
+// Context represents a context of system under test that we'd
+// like to deploy and teardown.
+type Context struct {
+	Ctx context.Context
+	Log *zap.SugaredLogger
+	*testlib.Client
+}
diff --git a/vendor/knative.dev/eventing/test/upgrade/prober/verify.go b/vendor/knative.dev/eventing/test/upgrade/prober/verify.go
index 1687753a0..90ce927a6 100644
--- a/vendor/knative.dev/eventing/test/upgrade/prober/verify.go
+++ b/vendor/knative.dev/eventing/test/upgrade/prober/verify.go
@@ -22,7 +22,6 @@ import (
 	"fmt"
 	"time"
 
-	"github.com/wavesoftware/go-ensure"
 	"go.uber.org/zap"
 	batchv1 "k8s.io/api/batch/v1"
 	corev1 "k8s.io/api/core/v1"
@@ -42,8 +41,8 @@ const (
 )
 
 // Verify will verify prober state after finished has been sent.
-func (p *prober) Verify(ctx context.Context) (eventErrs []error, eventsSent int, fetchErr error) {
-	report := p.fetchReport(ctx)
+func (p *prober) Verify() (eventErrs []error, eventsSent int) {
+	report := p.fetchReport()
 	availRate := 0.0
 	if report.TotalRequests != 0 {
 		availRate = float64(report.EventsSent*100) / float64(report.TotalRequests)
@@ -53,7 +52,7 @@ func (p *prober) Verify(ctx context.Context) (eventErrs []error, eventsSent int,
 	p.log.Infof("Availability: %.3f%%, Requests sent: %d.",
 		availRate, report.TotalRequests)
 	if report.State == "active" {
-		return nil, report.EventsSent, errors.New("report fetched too early, receiver is in active state")
+		p.client.T.Fatal("report fetched too early, receiver is in active state")
 	}
 	for _, t := range report.Thrown.Missing {
 		eventErrs = append(eventErrs, errors.New(t))
@@ -71,16 +70,16 @@ func (p *prober) Verify(ctx context.Context) (eventErrs []error, eventsSent int,
 			eventErrs = append(eventErrs, errors.New(t))
 		}
 	}
-	return eventErrs, report.EventsSent, nil
+	return eventErrs, report.EventsSent
 }
 
 // Finish terminates sender which sends finished event.
-func (p *prober) Finish(ctx context.Context) {
-	p.removeSender(ctx)
+func (p *prober) Finish() {
+	p.removeSender()
 }
 
-func (p *prober) fetchReport(ctx context.Context) *receiver.Report {
-	exec := p.fetchExecution(ctx)
+func (p *prober) fetchReport() *receiver.Report {
+	exec := p.fetchExecution()
 	replayLogs(p.log, exec)
 	return exec.Report
 }
@@ -100,14 +99,14 @@ func replayLogs(log *zap.SugaredLogger, exec *fetcher.Execution) {
 	}
 }
 
-func (p *prober) fetchExecution(ctx context.Context) *fetcher.Execution {
-	ns := p.config.Namespace
-	job := p.deployFetcher(ctx)
-	defer p.deleteFetcher(ctx)
-	pod, err := p.findSucceededPod(ctx, job)
-	ensure.NoError(err)
-	bytes, err := pkgTest.PodLogs(ctx, p.client.Kube, pod.Name, fetcherName, ns)
-	ensure.NoError(err)
+func (p *prober) fetchExecution() *fetcher.Execution {
+	ns := p.client.Namespace
+	job := p.deployFetcher()
+	defer p.deleteFetcher()
+	pod, err := p.findSucceededPod(job)
+	p.ensureNoError(err)
+	bytes, err := pkgTest.PodLogs(p.config.Ctx, p.client.Kube, pod.Name, fetcherName, ns)
+	p.ensureNoError(err)
 	ex := &fetcher.Execution{
 		Logs: []fetcher.LogEntry{},
 		Report: &receiver.Report{
@@ -123,13 +122,13 @@ func (p *prober) fetchExecution(ctx context.Context) *fetcher.Execution {
 		},
 	}
 	err = json.Unmarshal(bytes, ex)
-	ensure.NoError(err)
+	p.ensureNoError(err)
 	return ex
 }
 
-func (p *prober) deployFetcher(ctx context.Context) *batchv1.Job {
+func (p *prober) deployFetcher() *batchv1.Job {
 	p.log.Info("Deploying fetcher job: ", fetcherName)
-	jobs := p.client.Kube.BatchV1().Jobs(p.config.Namespace)
+	jobs := p.client.Kube.BatchV1().Jobs(p.client.Namespace)
 	var replicas int32 = 1
 	fetcherJob := &batchv1.Job{
 		ObjectMeta: metav1.ObjectMeta{
@@ -170,28 +169,28 @@ func (p *prober) deployFetcher(ctx context.Context) *batchv1.Job {
 			},
 		},
 	}
-	created, err := jobs.Create(ctx, fetcherJob, metav1.CreateOptions{})
-	ensure.NoError(err)
+	created, err := jobs.Create(p.config.Ctx, fetcherJob, metav1.CreateOptions{})
+	p.ensureNoError(err)
 	p.log.Info("Waiting for fetcher job to succeed: ", fetcherName)
-	err = waitForJobToComplete(ctx, p.client.Kube, fetcherName, p.config.Namespace)
-	ensure.NoError(err)
+	err = waitForJobToComplete(p.config.Ctx, p.client.Kube, fetcherName, p.client.Namespace)
+	p.ensureNoError(err)
 
 	return created
 }
 
-func (p *prober) deleteFetcher(ctx context.Context) {
-	ns := p.config.Namespace
+func (p *prober) deleteFetcher() {
+	ns := p.client.Namespace
 	jobs := p.client.Kube.BatchV1().Jobs(ns)
-	err := jobs.Delete(ctx, fetcherName, metav1.DeleteOptions{})
-	ensure.NoError(err)
+	err := jobs.Delete(p.config.Ctx, fetcherName, metav1.DeleteOptions{})
+	p.ensureNoError(err)
 }
 
-func (p *prober) findSucceededPod(ctx context.Context, job *batchv1.Job) (*corev1.Pod, error) {
+func (p *prober) findSucceededPod(job *batchv1.Job) (*corev1.Pod, error) {
 	pods := p.client.Kube.CoreV1().Pods(job.Namespace)
-	podList, err := pods.List(ctx, metav1.ListOptions{
+	podList, err := pods.List(p.config.Ctx, metav1.ListOptions{
 		LabelSelector: fmt.Sprint("job-name=", job.Name),
 	})
-	ensure.NoError(err)
+	p.ensureNoError(err)
 	for _, pod := range podList.Items {
 		if pod.Status.Phase == corev1.PodSucceeded {
 			return &pod, nil
diff --git a/vendor/knative.dev/eventing/test/upgrade/prober/wathola/config/defaults.go b/vendor/knative.dev/eventing/test/upgrade/prober/wathola/config/defaults.go
index e25f0a92d..77f300a61 100644
--- a/vendor/knative.dev/eventing/test/upgrade/prober/wathola/config/defaults.go
+++ b/vendor/knative.dev/eventing/test/upgrade/prober/wathola/config/defaults.go
@@ -62,7 +62,7 @@ func defaultValues() *Config {
 				Duration: time.Second,
 			},
 			Errors: ReceiverErrorConfig{
-				UnavailablePeriodToReport: 5 * time.Second,
+				UnavailablePeriodToReport: time.Minute,
 			},
 		},
 		Forwarder: ForwarderConfig{
diff --git a/vendor/knative.dev/eventing/test/upgrade/prober/wathola/event/types.go b/vendor/knative.dev/eventing/test/upgrade/prober/wathola/event/types.go
index f6559dab5..2a7fe264b 100644
--- a/vendor/knative.dev/eventing/test/upgrade/prober/wathola/event/types.go
+++ b/vendor/knative.dev/eventing/test/upgrade/prober/wathola/event/types.go
@@ -23,9 +23,9 @@ import (
 
 const (
 	// StepType is a string type representation of step event
-	StepType = "com.github.cardil.wathola.step"
+	StepType = "dev.knative.eventing.wathola.step"
 	// FinishedType os a string type representation of finished event
-	FinishedType = "com.github.cardil.wathola.finished"
+	FinishedType = "dev.knative.eventing.wathola.finished"
 )
 
 // Step is a event call at each step of verification