Repository: knative-sandbox/eventing-kafka
Pull request #91 (2020-10-09 06:48:39)

dep updates and http to HTTP from eventing
Signed-off-by: Matthias Wessendorf <mwessend@redhat.com>

Fixes #

<!-- Please include the 'why' behind your changes if no issue exists -->

## Proposed Changes

-
-
-

<!--
If this change has user-visible impact, follow the instructions below.
Examples include:

- 🎁 Add new feature
- 🐛 Fix bug
- 🧽 Update or clean up current behavior
- 🗑️ Remove feature or internal logic

Otherwise delete the rest of this template.
-->

**Release Note**

<!--
🗒️ 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

```

**Docs**

<!--
📖 If this change has user-visible impact, link to an issue or PR in
https://github.com/knative/docs.
-->


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

This pull-request has been approved by: *<a href="https://github.com/knative-sandbox/eventing-kafka/pull/91#" title="Author self-approved">matzew</a>*

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

The pull request process is described [here](https://git.k8s.io/community/contributors/guide/owners.md#the-code-review-process)

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

- ~~[OWNERS](https://github.com/knative-sandbox/eventing-kafka/blob/master/OWNERS)~~ [matzew]

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":[]} -->
knative-prow-robot: @matzew: The following test **failed**, say `/retest` to rerun all failed tests:

Test name | Commit | Details | Rerun command
--- | --- | --- | ---
pull-knative-sandbox-eventing-kafka-unit-tests | ae8603621159688147b41c44aa19d3341ec15f7a | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/91/pull-knative-sandbox-eventing-kafka-unit-tests/1314457876244205568) | `/test pull-knative-sandbox-eventing-kafka-unit-tests`

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

<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: @matzew: The following tests **failed**, say `/retest` to rerun all failed tests:

Test name | Commit | Details | Rerun command
--- | --- | --- | ---
pull-knative-sandbox-eventing-kafka-unit-tests | ae8603621159688147b41c44aa19d3341ec15f7a | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/91/pull-knative-sandbox-eventing-kafka-unit-tests/1314457876244205568) | `/test pull-knative-sandbox-eventing-kafka-unit-tests`
pull-knative-sandbox-eventing-kafka-build-tests | ae8603621159688147b41c44aa19d3341ec15f7a | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/91/pull-knative-sandbox-eventing-kafka-build-tests/1314457876248399873) | `/test pull-knative-sandbox-eventing-kafka-build-tests`

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

<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: @matzew: The following tests **failed**, say `/retest` to rerun all failed tests:

Test name | Commit | Details | Rerun command
--- | --- | --- | ---
pull-knative-sandbox-eventing-kafka-unit-tests | ae8603621159688147b41c44aa19d3341ec15f7a | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/91/pull-knative-sandbox-eventing-kafka-unit-tests/1314457876244205568) | `/test pull-knative-sandbox-eventing-kafka-unit-tests`
pull-knative-sandbox-eventing-kafka-build-tests | ae8603621159688147b41c44aa19d3341ec15f7a | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/91/pull-knative-sandbox-eventing-kafka-build-tests/1314457876248399873) | `/test pull-knative-sandbox-eventing-kafka-build-tests`
pull-knative-sandbox-eventing-kafka-integration-tests | ae8603621159688147b41c44aa19d3341ec15f7a | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/91/pull-knative-sandbox-eventing-kafka-integration-tests/1314457876248399872) | `/test pull-knative-sandbox-eventing-kafka-integration-tests`

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

<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: [APPROVALNOTIFIER] This PR is **APPROVED**

This pull-request has been approved by: *<a href="https://github.com/knative-sandbox/eventing-kafka/pull/91#" title="Author self-approved">matzew</a>*, *<a href="https://github.com/knative-sandbox/eventing-kafka/pull/91#pullrequestreview-505469709" title="LGTM">pierDipi</a>*

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

The pull request process is described [here](https://git.k8s.io/community/contributors/guide/owners.md#the-code-review-process)

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

- ~~[OWNERS](https://github.com/knative-sandbox/eventing-kafka/blob/master/OWNERS)~~ [matzew]

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":[]} -->

Patch:
diff --git a/go.mod b/go.mod
index d355b77b0..4913aec91 100644
--- a/go.mod
+++ b/go.mod
@@ -27,9 +27,9 @@ require (
 	k8s.io/apimachinery v0.18.8
 	k8s.io/client-go v11.0.1-0.20190805182717-6502b5e7b1b5+incompatible
 	k8s.io/utils v0.0.0-20200603063816-c1c6865ac451
-	knative.dev/eventing v0.18.1-0.20201006061220-cca58af424d3
-	knative.dev/pkg v0.0.0-20201006044120-247841408e57
-	knative.dev/test-infra v0.0.0-20201006044420-26457f2cad45
+	knative.dev/eventing v0.18.1-0.20201009013021-d37dbc88e6fd
+	knative.dev/pkg v0.0.0-20201009023821-73da5142da72
+	knative.dev/test-infra v0.0.0-20201008221921-aeca2ff2371a
 	pack.ag/amqp v0.12.4 // indirect
 )
 
diff --git a/go.sum b/go.sum
index 6d6253a7a..3f4230855 100644
--- a/go.sum
+++ b/go.sum
@@ -266,6 +266,7 @@ github.com/blang/semver v3.5.0+incompatible h1:CGxCgetQ64DKk7rdZ++Vfnb1+ogGNnB17
 github.com/blang/semver v3.5.0+incompatible/go.mod h1:kRBLl5iJ+tD4TcOOxsy/0fnwebNt5EWlYSAyrTnjyyk=
 github.com/blang/semver v3.5.1+incompatible h1:cQNTCjp13qL8KC3Nbxr/y2Bqb63oX6wdnnjpJbkM4JQ=
 github.com/blang/semver v3.5.1+incompatible/go.mod h1:kRBLl5iJ+tD4TcOOxsy/0fnwebNt5EWlYSAyrTnjyyk=
+github.com/blang/semver/v4 v4.0.0/go.mod h1:IbckMUScFkM3pff0VJDNKRiT6TG/YpiHIM2yvyW5YoQ=
 github.com/bmizerany/assert v0.0.0-20160611221934-b7ed37b82869/go.mod h1:Ekp36dRnpXw/yCqJaO+ZrUyxD+3VXMFFr56k5XYrpB4=
 github.com/bmizerany/perks v0.0.0-20141205001514-d9a9656a3a4b h1:AP/Y7sqYicnjGDfD5VcY4CIfh1hRXBUavxrvELjTiOE=
 github.com/bmizerany/perks v0.0.0-20141205001514-d9a9656a3a4b/go.mod h1:ac9efd0D1fsDb3EJvhqgXRbFx7bs2wqZ10HQPeU8U/Q=
@@ -439,6 +440,10 @@ github.com/globalsign/mgo v0.0.0-20181015135952-eeefdecb41b8/go.mod h1:xkRDCp4j0
 github.com/go-bindata/go-bindata/v3 v3.1.3/go.mod h1:1/zrpXsLD8YDIbhZRqXzm1Ghc7NhEvIN9+Z6R5/xH4I=
 github.com/go-critic/go-critic v0.4.1/go.mod h1:7/14rZGnZbY6E38VEGk2kVhoq6itzc1E68facVDK23g=
 github.com/go-critic/go-critic v0.4.3/go.mod h1:j4O3D4RoIwRqlZw5jJpx0BNfXWWbpcJoKu5cYSe4YmQ=
+github.com/go-git/gcfg v1.5.0/go.mod h1:5m20vg6GwYabIxaOonVkTdrILxQMpEShl1xiMF4ua+E=
+github.com/go-git/go-billy/v5 v5.0.0/go.mod h1:pmpqyWchKfYfrkb/UVH4otLvyi/5gJlGI4Hb3ZqZ3W0=
+github.com/go-git/go-git-fixtures/v4 v4.0.1/go.mod h1:m+ICp2rF3jDhFgEZ/8yziagdT1C+ZpZcrJjappBCDSw=
+github.com/go-git/go-git/v5 v5.1.0/go.mod h1:ZKfuPUoY1ZqIG4QG9BDBh3G4gLM5zvPuSJAozQrZuyM=
 github.com/go-gl/glfw v0.0.0-20190409004039-e6da0acd62b1/go.mod h1:vR7hzQXu2zJy9AVAgeJqvqgH9Q5CA+iKCZ2gyEVpxRU=
 github.com/go-gl/glfw/v3.3/glfw v0.0.0-20191125211704-12ad95a8df72/go.mod h1:tQ2UAYgL5IevRw8kRxooKSPJfGvJ9fJQFa0TUsXzTg8=
 github.com/go-gl/glfw/v3.3/glfw v0.0.0-20200222043503-6f7a984d4dc4/go.mod h1:tQ2UAYgL5IevRw8kRxooKSPJfGvJ9fJQFa0TUsXzTg8=
@@ -1913,10 +1918,11 @@ gopkg.in/yaml.v2 v2.2.8/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
 gopkg.in/yaml.v2 v2.3.0 h1:clyUAQHOM3G0M3f5vQj7LuJrETvjVot3Z5el9nffUtU=
 gopkg.in/yaml.v2 v2.3.0/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
 gopkg.in/yaml.v3 v3.0.0-20190709130402-674ba3eaed22/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
-gopkg.in/yaml.v3 v3.0.0-20191026110619-0b21df46bc1d/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
 gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
 gopkg.in/yaml.v3 v3.0.0-20200601152816-913338de1bd2 h1:VEmvx0P+GVTgkNu2EdTN988YCZPcD3lo9AoczZpucwc=
 gopkg.in/yaml.v3 v3.0.0-20200601152816-913338de1bd2/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
+gopkg.in/yaml.v3 v3.0.0-20200615113413-eeeca48fe776 h1:tQIYjPdBoyREyB9XMu+nnTclpTYkz2zFM+lzLJFO4gQ=
+gopkg.in/yaml.v3 v3.0.0-20200615113413-eeeca48fe776/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
 gotest.tools v2.2.0+incompatible/go.mod h1:DsYFclhRJ6vuDpmuTbkuFWG+y2sxOXAzmJt81HFBacw=
 helm.sh/helm/v3 v3.1.1/go.mod h1:WYsFJuMASa/4XUqLyv54s0U/f3mlAaRErGmyy4z921g=
 honnef.co/go/tools v0.0.0-20180728063816-88497007e858/go.mod h1:rf3lG4BRIbNafJWhAfAdb/ePZxsR/4RtNHQocxwk9r4=
@@ -1997,8 +2003,8 @@ k8s.io/utils v0.0.0-20200603063816-c1c6865ac451 h1:v8ud2Up6QK1lNOKFgiIVrZdMg7Mpm
 k8s.io/utils v0.0.0-20200603063816-c1c6865ac451/go.mod h1:jPW/WVKK9YHAvNhRxK0md/EJ228hCsBRufyofKtW8HA=
 knative.dev/caching v0.0.0-20190719140829-2032732871ff/go.mod h1:dHXFU6CGlLlbzaWc32g80cR92iuBSpsslDNBWI8C7eg=
 knative.dev/caching v0.0.0-20200116200605-67bca2c83dfa/go.mod h1:dHXFU6CGlLlbzaWc32g80cR92iuBSpsslDNBWI8C7eg=
-knative.dev/eventing v0.18.1-0.20201006061220-cca58af424d3 h1:B3EgOUEOkCctR/wJv/dEUc03cqCTaVmSAWIdpId0r+U=
-knative.dev/eventing v0.18.1-0.20201006061220-cca58af424d3/go.mod h1:gPb7tc5qIHsO6DHNPcRK4CNbQLeYzA2n8YlIW9eSVc0=
+knative.dev/eventing v0.18.1-0.20201009013021-d37dbc88e6fd h1:6Tun+LeSvTmnY1ZBTsWFvvd3cfgZTfFYKvxrFjuaRm0=
+knative.dev/eventing v0.18.1-0.20201009013021-d37dbc88e6fd/go.mod h1:05WPTpHdblD/8+IgrN2f6jFLLiUIxVEfZCwDEGK44mk=
 knative.dev/eventing-contrib v0.6.1-0.20190723221543-5ce18048c08b/go.mod h1:SnXZgSGgMSMLNFTwTnpaOH7hXDzTFtw0J8OmHflNx3g=
 knative.dev/eventing-contrib v0.11.2/go.mod h1:SnXZgSGgMSMLNFTwTnpaOH7hXDzTFtw0J8OmHflNx3g=
 knative.dev/pkg v0.0.0-20191101194912-56c2594e4f11/go.mod h1:pgODObA1dTyhNoFxPZTTjNWfx6F0aKsKzn+vaT9XO/Q=
@@ -2009,18 +2015,18 @@ knative.dev/pkg v0.0.0-20200505191044-3da93ebb24c2/go.mod h1:Q6sL35DdGs8hIQZKdaC
 knative.dev/pkg v0.0.0-20200515002500-16d7b963416f/go.mod h1:tMOHGbxtRz8zYFGEGpV/bpoTEM1o89MwYFC4YJXl3GY=
 knative.dev/pkg v0.0.0-20200528142800-1c6815d7e4c9/go.mod h1:QgNZTxnwpB/oSpNcfnLVlw+WpEwwyKAvJlvR3hgeltA=
 knative.dev/pkg v0.0.0-20200711004937-22502028e31a/go.mod h1:AqAJV6rYi8IGikDjJ/9ZQd9qKdkXVlesVnVjwx62YB8=
-knative.dev/pkg v0.0.0-20201003175733-ea7374e81105/go.mod h1:o9w/M7OnPc0ou07PIUORHUBSZgszeZebAYZkxQ8GdFw=
-knative.dev/pkg v0.0.0-20201006044120-247841408e57 h1:5tWwqp0FC7nPELsgpmvs35/gbo7Lw8MIROW8jGuxX0A=
-knative.dev/pkg v0.0.0-20201006044120-247841408e57/go.mod h1:xpPYLHYXh306wQISWkRC/5t3+CD76nOcEHrCQ+5pxBM=
+knative.dev/pkg v0.0.0-20201008052621-c0496d5e1b2b/go.mod h1:h4jEOUhdPbt4nTdoEmyGk3bKcEr1tu5US+ou36dEeJc=
+knative.dev/pkg v0.0.0-20201009023821-73da5142da72 h1:7JOLxLzCs9GcoBdKpEyGsrDE1ZrNCVOCyXQg32uYo9c=
+knative.dev/pkg v0.0.0-20201009023821-73da5142da72/go.mod h1:AqiTNwkpHWuywH8MtgvqzYrph/qbzACviH5iyEfDAZ8=
 knative.dev/test-infra v0.0.0-20200407185800-1b88cb3b45a5/go.mod h1:xcdUkMJrLlBswIZqL5zCuBFOC22WIPMQoVX1L35i0vQ=
 knative.dev/test-infra v0.0.0-20200505052144-5ea2f705bb55/go.mod h1:WqF1Azka+FxPZ20keR2zCNtiQA1MP9ZB4BH4HuI+SIU=
 knative.dev/test-infra v0.0.0-20200513011557-d03429a76034/go.mod h1:aMif0KXL4g19YCYwsy4Ocjjz5xgPlseYV+B95Oo4JGE=
 knative.dev/test-infra v0.0.0-20200519015156-82551620b0a9/go.mod h1:A5b2OAXTOeHT3hHhVQm3dmtbuWvIDP7qzgtqxA3/2pE=
 knative.dev/test-infra v0.0.0-20200707183444-aed09e56ddc7/go.mod h1:RjYAhXnZqeHw9+B0zsbqSPlae0lCvjekO/nw5ZMpLCs=
-knative.dev/test-infra v0.0.0-20201001200229-a6988e3b3b38/go.mod h1:Pmg2c7Z7q7BGFUV/GOpU5BlrD3ePJft4MPqx8AYBplc=
-knative.dev/test-infra v0.0.0-20201002164834-8c07ff018549/go.mod h1:Pmg2c7Z7q7BGFUV/GOpU5BlrD3ePJft4MPqx8AYBplc=
-knative.dev/test-infra v0.0.0-20201006044420-26457f2cad45 h1:uuttx/O6SNZZ4lzvvDKc1HWy+72TxcKLBM3ponGwGdU=
-knative.dev/test-infra v0.0.0-20201006044420-26457f2cad45/go.mod h1:Pmg2c7Z7q7BGFUV/GOpU5BlrD3ePJft4MPqx8AYBplc=
+knative.dev/test-infra v0.0.0-20201005021934-df2084cec789/go.mod h1:Pmg2c7Z7q7BGFUV/GOpU5BlrD3ePJft4MPqx8AYBplc=
+knative.dev/test-infra v0.0.0-20201008043522-20ae6ee1a359/go.mod h1:Qj8sZPA2mTFusTOH13YVns7jY7xsnrLsZdPB0aTdejs=
+knative.dev/test-infra v0.0.0-20201008221921-aeca2ff2371a h1:Z+IRS8COiXGmnxgoeqUEvQii8QFc0mbSRjUh4c1jQdM=
+knative.dev/test-infra v0.0.0-20201008221921-aeca2ff2371a/go.mod h1:W4EfffNKnIkYuBHdOjw0UP3YtFbK4ImwDiEoIAbG1zk=
 mvdan.cc/interfacer v0.0.0-20180901003855-c20040233aed/go.mod h1:Xkxe497xwlCKkIaQYRfC7CSLworTXY9RMqwhhCm+8Nc=
 mvdan.cc/lint v0.0.0-20170908181259-adc824a0674b/go.mod h1:2odslEg/xrtNQqCYg2/jCoyKnw3vv5biOc3JnIcYfL4=
 mvdan.cc/unparam v0.0.0-20190720180237-d51796306d8f/go.mod h1:4G1h5nDURzA3bwVMZIVpwbkw+04kSxk3rAtzlimaUJw=
diff --git a/pkg/source/adapter/adapter.go b/pkg/source/adapter/adapter.go
index 492f70e20..c4230a40c 100644
--- a/pkg/source/adapter/adapter.go
+++ b/pkg/source/adapter/adapter.go
@@ -55,7 +55,7 @@ func NewEnvConfig() adapter.EnvConfigAccessor {
 
 type Adapter struct {
 	config            *adapterConfig
-	httpMessageSender *kncloudevents.HttpMessageSender
+	httpMessageSender *kncloudevents.HTTPMessageSender
 	reporter          pkgsource.StatsReporter
 	logger            *zap.SugaredLogger
 	keyTypeMapper     func([]byte) interface{}
@@ -64,7 +64,7 @@ type Adapter struct {
 var _ adapter.MessageAdapter = (*Adapter)(nil)
 var _ adapter.MessageAdapterConstructor = NewAdapter
 
-func NewAdapter(ctx context.Context, processed adapter.EnvConfigAccessor, httpMessageSender *kncloudevents.HttpMessageSender, reporter pkgsource.StatsReporter) adapter.MessageAdapter {
+func NewAdapter(ctx context.Context, processed adapter.EnvConfigAccessor, httpMessageSender *kncloudevents.HTTPMessageSender, reporter pkgsource.StatsReporter) adapter.MessageAdapter {
 	logger := logging.FromContext(ctx)
 	config := processed.(*adapterConfig)
 
diff --git a/vendor/gopkg.in/yaml.v3/.travis.yml b/vendor/gopkg.in/yaml.v3/.travis.yml
index 04d4dae09..a130fe883 100644
--- a/vendor/gopkg.in/yaml.v3/.travis.yml
+++ b/vendor/gopkg.in/yaml.v3/.travis.yml
@@ -11,6 +11,7 @@ go:
     - "1.11.x"
     - "1.12.x"
     - "1.13.x"
+    - "1.14.x"
     - "tip"
 
 go_import_path: gopkg.in/yaml.v3
diff --git a/vendor/gopkg.in/yaml.v3/decode.go b/vendor/gopkg.in/yaml.v3/decode.go
index d50c61aae..21c0dacfd 100644
--- a/vendor/gopkg.in/yaml.v3/decode.go
+++ b/vendor/gopkg.in/yaml.v3/decode.go
@@ -35,6 +35,7 @@ type parser struct {
 	doc      *Node
 	anchors  map[string]*Node
 	doneInit bool
+	textless bool
 }
 
 func newParser(b []byte) *parser {
@@ -173,17 +174,20 @@ func (p *parser) node(kind Kind, defaultTag, tag, value string) *Node {
 	} else if kind == ScalarNode {
 		tag, _ = resolve("", value)
 	}
-	return &Node{
-		Kind:        kind,
-		Tag:         tag,
-		Value:       value,
-		Style:       style,
-		Line:        p.event.start_mark.line + 1,
-		Column:      p.event.start_mark.column + 1,
-		HeadComment: string(p.event.head_comment),
-		LineComment: string(p.event.line_comment),
-		FootComment: string(p.event.foot_comment),
+	n := &Node{
+		Kind:  kind,
+		Tag:   tag,
+		Value: value,
+		Style: style,
+	}
+	if !p.textless {
+		n.Line = p.event.start_mark.line + 1
+		n.Column = p.event.start_mark.column + 1
+		n.HeadComment = string(p.event.head_comment)
+		n.LineComment = string(p.event.line_comment)
+		n.FootComment = string(p.event.foot_comment)
 	}
+	return n
 }
 
 func (p *parser) parseChild(parent *Node) *Node {
@@ -395,7 +399,7 @@ func (d *decoder) callObsoleteUnmarshaler(n *Node, u obsoleteUnmarshaler) (good
 //
 // If n holds a null value, prepare returns before doing anything.
 func (d *decoder) prepare(n *Node, out reflect.Value) (newout reflect.Value, unmarshaled, good bool) {
-	if n.ShortTag() == nullTag {
+	if n.ShortTag() == nullTag || n.Kind == 0 && n.IsZero() {
 		return out, false, false
 	}
 	again := true
@@ -501,8 +505,13 @@ func (d *decoder) unmarshal(n *Node, out reflect.Value) (good bool) {
 		good = d.mapping(n, out)
 	case SequenceNode:
 		good = d.sequence(n, out)
+	case 0:
+		if n.IsZero() {
+			return d.null(out)
+		}
+		fallthrough
 	default:
-		panic("internal error: unknown node kind: " + strconv.Itoa(int(n.Kind)))
+		failf("cannot decode node with unknown kind %d", n.Kind)
 	}
 	return good
 }
@@ -537,6 +546,17 @@ func resetMap(out reflect.Value) {
 	}
 }
 
+func (d *decoder) null(out reflect.Value) bool {
+	if out.CanAddr() {
+		switch out.Kind() {
+		case reflect.Interface, reflect.Ptr, reflect.Map, reflect.Slice:
+			out.Set(reflect.Zero(out.Type()))
+			return true
+		}
+	}
+	return false
+}
+
 func (d *decoder) scalar(n *Node, out reflect.Value) bool {
 	var tag string
 	var resolved interface{}
@@ -554,14 +574,7 @@ func (d *decoder) scalar(n *Node, out reflect.Value) bool {
 		}
 	}
 	if resolved == nil {
-		if out.CanAddr() {
-			switch out.Kind() {
-			case reflect.Interface, reflect.Ptr, reflect.Map, reflect.Slice:
-				out.Set(reflect.Zero(out.Type()))
-				return true
-			}
-		}
-		return false
+		return d.null(out)
 	}
 	if resolvedv := reflect.ValueOf(resolved); out.Type() == resolvedv.Type() {
 		// We've resolved to exactly the type we want, so use that.
diff --git a/vendor/gopkg.in/yaml.v3/emitterc.go b/vendor/gopkg.in/yaml.v3/emitterc.go
index c7bd5d2a3..c29217ef5 100644
--- a/vendor/gopkg.in/yaml.v3/emitterc.go
+++ b/vendor/gopkg.in/yaml.v3/emitterc.go
@@ -235,10 +235,13 @@ func yaml_emitter_increase_indent(emitter *yaml_emitter_t, flow, indentless bool
 			emitter.indent = 0
 		}
 	} else if !indentless {
-		emitter.indent += emitter.best_indent
-		// [Go] If inside a block sequence item, discount the space taken by the indicator.
-		if emitter.best_indent > 2 && emitter.states[len(emitter.states)-1] == yaml_EMIT_BLOCK_SEQUENCE_ITEM_STATE {
-			emitter.indent -= 2
+		// [Go] This was changed so that indentations are more regular.
+		if emitter.states[len(emitter.states)-1] == yaml_EMIT_BLOCK_SEQUENCE_ITEM_STATE {
+			// The first indent inside a sequence will just skip the "- " indicator.
+			emitter.indent += 2
+		} else {
+			// Everything else aligns to the chosen indentation.
+			emitter.indent = emitter.best_indent*((emitter.indent+emitter.best_indent)/emitter.best_indent)
 		}
 	}
 	return true
@@ -725,16 +728,9 @@ func yaml_emitter_emit_flow_mapping_value(emitter *yaml_emitter_t, event *yaml_e
 // Expect a block item node.
 func yaml_emitter_emit_block_sequence_item(emitter *yaml_emitter_t, event *yaml_event_t, first bool) bool {
 	if first {
-		// [Go] The original logic here would not indent the sequence when
-		//      inside a mapping. In Go we always indent it.
-		indentless := false
-		original := emitter.indent
-		if !yaml_emitter_increase_indent(emitter, false, indentless) {
+		if !yaml_emitter_increase_indent(emitter, false, false) {
 			return false
 		}
-		if emitter.indent > original+2 {
-			emitter.indent -= 2
-		}
 	}
 	if event.typ == yaml_SEQUENCE_END_EVENT {
 		emitter.indent = emitter.indents[len(emitter.indents)-1]
@@ -785,6 +781,13 @@ func yaml_emitter_emit_block_mapping_key(emitter *yaml_emitter_t, event *yaml_ev
 	if !yaml_emitter_write_indent(emitter) {
 		return false
 	}
+	if len(emitter.line_comment) > 0 {
+		// [Go] A line comment was provided for the key. That's unusual as the
+		//      scanner associates line comments with the value. Either way,
+		//      save the line comment and render it appropriately later.
+		emitter.key_line_comment = emitter.line_comment
+		emitter.line_comment = nil
+	}
 	if yaml_emitter_check_simple_key(emitter) {
 		emitter.states = append(emitter.states, yaml_EMIT_BLOCK_MAPPING_SIMPLE_VALUE_STATE)
 		return yaml_emitter_emit_node(emitter, event, false, false, true, true)
@@ -810,6 +813,29 @@ func yaml_emitter_emit_block_mapping_value(emitter *yaml_emitter_t, event *yaml_
 			return false
 		}
 	}
+	if len(emitter.key_line_comment) > 0 {
+		// [Go] A line comment was previously provided for the key. Handle it before
+		//      the value so the inline comments are placed correctly.
+		if yaml_emitter_silent_nil_event(emitter, event) && len(emitter.line_comment) == 0 {
+			// Nothing other than the line comment will be written on the line.
+			emitter.line_comment = emitter.key_line_comment
+			emitter.key_line_comment = nil
+		} else {
+			// An actual value is coming, so emit the comment line.
+			emitter.line_comment, emitter.key_line_comment = emitter.key_line_comment, emitter.line_comment
+			if !yaml_emitter_process_line_comment(emitter) {
+				return false
+			}
+			emitter.line_comment, emitter.key_line_comment = emitter.key_line_comment, emitter.line_comment
+			// Indent in unless it's a block that will reindent anyway.
+			if event.sequence_style() == yaml_FLOW_SEQUENCE_STYLE || (event.typ != yaml_MAPPING_START_EVENT && event.typ != yaml_SEQUENCE_START_EVENT) {
+				emitter.indent = emitter.best_indent*((emitter.indent+emitter.best_indent)/emitter.best_indent)
+				if !yaml_emitter_write_indent(emitter) {
+					return false
+				}
+			}
+		}
+	}
 	emitter.states = append(emitter.states, yaml_EMIT_BLOCK_MAPPING_KEY_STATE)
 	if !yaml_emitter_emit_node(emitter, event, false, false, true, false) {
 		return false
@@ -823,6 +849,10 @@ func yaml_emitter_emit_block_mapping_value(emitter *yaml_emitter_t, event *yaml_
 	return true
 }
 
+func yaml_emitter_silent_nil_event(emitter *yaml_emitter_t, event *yaml_event_t) bool {
+	return event.typ == yaml_SCALAR_EVENT && event.implicit && !emitter.canonical && len(emitter.scalar_data.value) == 0
+}
+
 // Expect a node.
 func yaml_emitter_emit_node(emitter *yaml_emitter_t, event *yaml_event_t,
 	root bool, sequence bool, mapping bool, simple_key bool) bool {
diff --git a/vendor/gopkg.in/yaml.v3/encode.go b/vendor/gopkg.in/yaml.v3/encode.go
index e998db941..45e8d1e1b 100644
--- a/vendor/gopkg.in/yaml.v3/encode.go
+++ b/vendor/gopkg.in/yaml.v3/encode.go
@@ -425,18 +425,23 @@ func (e *encoder) nodev(in reflect.Value) {
 }
 
 func (e *encoder) node(node *Node, tail string) {
+	// Zero nodes behave as nil.
+	if node.Kind == 0 && node.IsZero() {
+		e.nilv()
+		return
+	}
+
 	// If the tag was not explicitly requested, and dropping it won't change the
 	// implicit tag of the value, don't include it in the presentation.
 	var tag = node.Tag
 	var stag = shortTag(tag)
-	var rtag string
 	var forceQuoting bool
 	if tag != "" && node.Style&TaggedStyle == 0 {
 		if node.Kind == ScalarNode {
 			if stag == strTag && node.Style&(SingleQuotedStyle|DoubleQuotedStyle|LiteralStyle|FoldedStyle) != 0 {
 				tag = ""
 			} else {
-				rtag, _ = resolve("", node.Value)
+				rtag, _ := resolve("", node.Value)
 				if rtag == stag {
 					tag = ""
 				} else if stag == strTag {
@@ -445,6 +450,7 @@ func (e *encoder) node(node *Node, tail string) {
 				}
 			}
 		} else {
+			var rtag string
 			switch node.Kind {
 			case MappingNode:
 				rtag = mapTag
@@ -474,7 +480,7 @@ func (e *encoder) node(node *Node, tail string) {
 		if node.Style&FlowStyle != 0 {
 			style = yaml_FLOW_SEQUENCE_STYLE
 		}
-		e.must(yaml_sequence_start_event_initialize(&e.event, []byte(node.Anchor), []byte(tag), tag == "", style))
+		e.must(yaml_sequence_start_event_initialize(&e.event, []byte(node.Anchor), []byte(longTag(tag)), tag == "", style))
 		e.event.head_comment = []byte(node.HeadComment)
 		e.emit()
 		for _, node := range node.Content {
@@ -490,7 +496,7 @@ func (e *encoder) node(node *Node, tail string) {
 		if node.Style&FlowStyle != 0 {
 			style = yaml_FLOW_MAPPING_STYLE
 		}
-		yaml_mapping_start_event_initialize(&e.event, []byte(node.Anchor), []byte(tag), tag == "", style)
+		yaml_mapping_start_event_initialize(&e.event, []byte(node.Anchor), []byte(longTag(tag)), tag == "", style)
 		e.event.tail_comment = []byte(tail)
 		e.event.head_comment = []byte(node.HeadComment)
 		e.emit()
@@ -531,11 +537,11 @@ func (e *encoder) node(node *Node, tail string) {
 	case ScalarNode:
 		value := node.Value
 		if !utf8.ValidString(value) {
-			if tag == binaryTag {
+			if stag == binaryTag {
 				failf("explicitly tagged !!binary data must be base64-encoded")
 			}
-			if tag != "" {
-				failf("cannot marshal invalid UTF-8 data as %s", shortTag(tag))
+			if stag != "" {
+				failf("cannot marshal invalid UTF-8 data as %s", stag)
 			}
 			// It can't be encoded directly as YAML so use a binary tag
 			// and encode it as base64.
@@ -560,5 +566,7 @@ func (e *encoder) node(node *Node, tail string) {
 		}
 
 		e.emitScalar(value, node.Anchor, tag, style, []byte(node.HeadComment), []byte(node.LineComment), []byte(node.FootComment), []byte(tail))
+	default:
+		failf("cannot encode node with unknown kind %d", node.Kind)
 	}
 }
diff --git a/vendor/gopkg.in/yaml.v3/parserc.go b/vendor/gopkg.in/yaml.v3/parserc.go
index aea9050b8..ac66fccc0 100644
--- a/vendor/gopkg.in/yaml.v3/parserc.go
+++ b/vendor/gopkg.in/yaml.v3/parserc.go
@@ -648,6 +648,10 @@ func yaml_parser_parse_node(parser *yaml_parser_t, event *yaml_event_t, block, i
 			implicit:   implicit,
 			style:      yaml_style_t(yaml_BLOCK_MAPPING_STYLE),
 		}
+		if parser.stem_comment != nil {
+			event.head_comment = parser.stem_comment
+			parser.stem_comment = nil
+		}
 		return true
 	}
 	if len(anchor) > 0 || len(tag) > 0 {
@@ -694,25 +698,13 @@ func yaml_parser_parse_block_sequence_entry(parser *yaml_parser_t, event *yaml_e
 
 	if token.typ == yaml_BLOCK_ENTRY_TOKEN {
 		mark := token.end_mark
-		prior_head := len(parser.head_comment)
+		prior_head_len := len(parser.head_comment)
 		skip_token(parser)
+		yaml_parser_split_stem_comment(parser, prior_head_len)
 		token = peek_token(parser)
 		if token == nil {
 			return false
 		}
-		if prior_head > 0 && token.typ == yaml_BLOCK_SEQUENCE_START_TOKEN {
-			// [Go] It's a sequence under a sequence entry, so the former head comment
-			//      is for the list itself, not the first list item under it.
-			parser.stem_comment = parser.head_comment[:prior_head]
-			if len(parser.head_comment) == prior_head {
-				parser.head_comment = nil
-			} else {
-				// Copy suffix to prevent very strange bugs if someone ever appends
-				// further bytes to the prefix in the stem_comment slice above.
-				parser.head_comment = append([]byte(nil), parser.head_comment[prior_head+1:]...)
-			}
-
-		}
 		if token.typ != yaml_BLOCK_ENTRY_TOKEN && token.typ != yaml_BLOCK_END_TOKEN {
 			parser.states = append(parser.states, yaml_PARSE_BLOCK_SEQUENCE_ENTRY_STATE)
 			return yaml_parser_parse_node(parser, event, true, false)
@@ -754,7 +746,9 @@ func yaml_parser_parse_indentless_sequence_entry(parser *yaml_parser_t, event *y
 
 	if token.typ == yaml_BLOCK_ENTRY_TOKEN {
 		mark := token.end_mark
+		prior_head_len := len(parser.head_comment)
 		skip_token(parser)
+		yaml_parser_split_stem_comment(parser, prior_head_len)
 		token = peek_token(parser)
 		if token == nil {
 			return false
@@ -780,6 +774,32 @@ func yaml_parser_parse_indentless_sequence_entry(parser *yaml_parser_t, event *y
 	return true
 }
 
+// Split stem comment from head comment.
+//
+// When a sequence or map is found under a sequence entry, the former head comment
+// is assigned to the underlying sequence or map as a whole, not the individual
+// sequence or map entry as would be expected otherwise. To handle this case the
+// previous head comment is moved aside as the stem comment.
+func yaml_parser_split_stem_comment(parser *yaml_parser_t, stem_len int) {
+	if stem_len == 0 {
+		return
+	}
+
+	token := peek_token(parser)
+	if token.typ != yaml_BLOCK_SEQUENCE_START_TOKEN && token.typ != yaml_BLOCK_MAPPING_START_TOKEN {
+		return
+	}
+
+	parser.stem_comment = parser.head_comment[:stem_len]
+	if len(parser.head_comment) == stem_len {
+		parser.head_comment = nil
+	} else {
+		// Copy suffix to prevent very strange bugs if someone ever appends
+		// further bytes to the prefix in the stem_comment slice above.
+		parser.head_comment = append([]byte(nil), parser.head_comment[stem_len+1:]...)
+	}
+}
+
 // Parse the productions:
 // block_mapping        ::= BLOCK-MAPPING_START
 //                          *******************
diff --git a/vendor/gopkg.in/yaml.v3/scannerc.go b/vendor/gopkg.in/yaml.v3/scannerc.go
index 2e9cbf97f..d9a539c39 100644
--- a/vendor/gopkg.in/yaml.v3/scannerc.go
+++ b/vendor/gopkg.in/yaml.v3/scannerc.go
@@ -749,6 +749,11 @@ func yaml_parser_fetch_next_token(parser *yaml_parser_t) (ok bool) {
 		if !ok {
 			return
 		}
+		if len(parser.tokens) > 0 && parser.tokens[len(parser.tokens)-1].typ == yaml_BLOCK_ENTRY_TOKEN {
+			// Sequence indicators alone have no line comments. It becomes
+			// a head comment for whatever follows.
+			return
+		}
 		if !yaml_parser_scan_line_comment(parser, comment_mark) {
 			ok = false
 			return
diff --git a/vendor/gopkg.in/yaml.v3/yaml.go b/vendor/gopkg.in/yaml.v3/yaml.go
index b5d35a50d..56e8a8490 100644
--- a/vendor/gopkg.in/yaml.v3/yaml.go
+++ b/vendor/gopkg.in/yaml.v3/yaml.go
@@ -89,7 +89,7 @@ func Unmarshal(in []byte, out interface{}) (err error) {
 	return unmarshal(in, out, false)
 }
 
-// A Decorder reads and decodes YAML values from an input stream.
+// A Decoder reads and decodes YAML values from an input stream.
 type Decoder struct {
 	parser      *parser
 	knownFields bool
@@ -194,7 +194,7 @@ func unmarshal(in []byte, out interface{}, strict bool) (err error) {
 //                  Zero valued structs will be omitted if all their public
 //                  fields are zero, unless they implement an IsZero
 //                  method (see the IsZeroer interface type), in which
-//                  case the field will be included if that method returns true.
+//                  case the field will be excluded if IsZero returns true.
 //
 //     flow         Marshal using a flow style (useful for structs,
 //                  sequences and maps).
@@ -252,6 +252,24 @@ func (e *Encoder) Encode(v interface{}) (err error) {
 	return nil
 }
 
+// Encode encodes value v and stores its representation in n.
+//
+// See the documentation for Marshal for details about the
+// conversion of Go values into YAML.
+func (n *Node) Encode(v interface{}) (err error) {
+	defer handleErr(&err)
+	e := newEncoder()
+	defer e.destroy()
+	e.marshalDoc("", reflect.ValueOf(v))
+	e.finish()
+	p := newParser(e.out)
+	p.textless = true
+	defer p.destroy()
+	doc := p.parse()
+	*n = *doc.Content[0]
+	return nil
+}
+
 // SetIndent changes the used indentation used when encoding.
 func (e *Encoder) SetIndent(spaces int) {
 	if spaces < 0 {
@@ -328,6 +346,12 @@ const (
 // and maps, Node is an intermediate representation that allows detailed
 // control over the content being decoded or encoded.
 //
+// It's worth noting that although Node offers access into details such as
+// line numbers, colums, and comments, the content when re-encoded will not
+// have its original textual representation preserved. An effort is made to
+// render the data plesantly, and to preserve comments near the data they
+// describe, though.
+//
 // Values that make use of the Node type interact with the yaml package in the
 // same way any other type would do, by encoding and decoding yaml data
 // directly or indirectly into them.
@@ -391,6 +415,13 @@ type Node struct {
 	Column int
 }
 
+// IsZero returns whether the node has all of its fields unset.
+func (n *Node) IsZero() bool {
+	return n.Kind == 0 && n.Style == 0 && n.Tag == "" && n.Value == "" && n.Anchor == "" && n.Alias == nil && n.Content == nil &&
+		n.HeadComment == "" && n.LineComment == "" && n.FootComment == "" && n.Line == 0 && n.Column == 0
+}
+
+
 // LongTag returns the long form of the tag that indicates the data type for
 // the node. If the Tag field isn't explicitly defined, one will be computed
 // based on the node properties.
diff --git a/vendor/gopkg.in/yaml.v3/yamlh.go b/vendor/gopkg.in/yaml.v3/yamlh.go
index 2719cfbb0..7c6d00770 100644
--- a/vendor/gopkg.in/yaml.v3/yamlh.go
+++ b/vendor/gopkg.in/yaml.v3/yamlh.go
@@ -787,6 +787,8 @@ type yaml_emitter_t struct {
 	foot_comment []byte
 	tail_comment []byte
 
+	key_line_comment []byte
+
 	// Dumper stuff
 
 	opened bool // If the stream was already opened?
diff --git a/vendor/knative.dev/eventing/pkg/adapter/v2/config.go b/vendor/knative.dev/eventing/pkg/adapter/v2/config.go
index f0cffcca5..ed3e59d19 100644
--- a/vendor/knative.dev/eventing/pkg/adapter/v2/config.go
+++ b/vendor/knative.dev/eventing/pkg/adapter/v2/config.go
@@ -90,7 +90,10 @@ type EnvConfig struct {
 	LeaderElectionConfigJson string `envconfig:"K_LEADER_ELECTION_CONFIG"`
 
 	// Time in seconds to wait for sink to respond
-	EnvSinkTimeout int `envconfig:"K_SINK_TIMEOUT"`
+	EnvSinkTimeout string `envconfig:"K_SINK_TIMEOUT"`
+
+	// cached zap logger
+	logger *zap.SugaredLogger
 }
 
 // EnvConfigAccessor defines accessors for the minimal
@@ -121,7 +124,7 @@ type EnvConfigAccessor interface {
 	// GetLeaderElectionConfig returns leader election configuration.
 	GetLeaderElectionConfig() (*kle.ComponentConfig, error)
 
-	// Get the name of the adapter.
+	// Get the timeout to apply on a request to a sink
 	GetSinktimeout() int
 }
 
@@ -141,18 +144,20 @@ func (e *EnvConfig) GetMetricsConfig() (*metrics.ExporterOptions, error) {
 }
 
 func (e *EnvConfig) GetLogger() *zap.SugaredLogger {
-	loggingConfig, err := logging.JsonToLoggingConfig(e.LoggingConfigJson)
-	if err != nil {
-		// Use default logging config.
-		if loggingConfig, err = logging.NewConfigFromMap(map[string]string{}); err != nil {
-			// If this fails, there is no recovering.
-			panic(err)
+	if e.logger == nil {
+		loggingConfig, err := logging.JsonToLoggingConfig(e.LoggingConfigJson)
+		if err != nil {
+			// Use default logging config.
+			if loggingConfig, err = logging.NewConfigFromMap(map[string]string{}); err != nil {
+				// If this fails, there is no recovering.
+				panic(err)
+			}
 		}
-	}
 
-	logger, _ := logging.NewLoggerFromConfig(loggingConfig, e.Component)
-
-	return logger
+		logger, _ := logging.NewLoggerFromConfig(loggingConfig, e.Component)
+		e.logger = logger
+	}
+	return e.logger
 }
 
 func (e *EnvConfig) GetSink() string {
@@ -168,7 +173,11 @@ func (e *EnvConfig) GetName() string {
 }
 
 func (e *EnvConfig) GetSinktimeout() int {
-	return e.EnvSinkTimeout
+	if duration, err := strconv.Atoi(e.EnvSinkTimeout); err == nil {
+		return duration
+	}
+	e.GetLogger().Warn("Sink timeout configuration is invalid, default to -1 (no timeout)")
+	return -1
 }
 
 func (e *EnvConfig) SetupTracing(logger *zap.SugaredLogger) error {
diff --git a/vendor/knative.dev/eventing/pkg/adapter/v2/main_message.go b/vendor/knative.dev/eventing/pkg/adapter/v2/main_message.go
index 916c5a32c..5cd584d96 100644
--- a/vendor/knative.dev/eventing/pkg/adapter/v2/main_message.go
+++ b/vendor/knative.dev/eventing/pkg/adapter/v2/main_message.go
@@ -39,7 +39,7 @@ type MessageAdapter interface {
 	Start(ctx context.Context) error
 }
 
-type MessageAdapterConstructor func(ctx context.Context, env EnvConfigAccessor, adapter *kncloudevents.HttpMessageSender, reporter source.StatsReporter) MessageAdapter
+type MessageAdapterConstructor func(ctx context.Context, env EnvConfigAccessor, adapter *kncloudevents.HTTPMessageSender, reporter source.StatsReporter) MessageAdapter
 
 func MainMessageAdapter(component string, ector EnvConfigConstructor, ctor MessageAdapterConstructor) {
 	ctx := signals.NewContext()
@@ -108,7 +108,7 @@ func MainMessageAdapterWithContext(ctx context.Context, component string, ector
 		logger.Error("Error setting up trace publishing", zap.Error(err))
 	}
 
-	httpBindingsSender, err := kncloudevents.NewHttpMessageSender(nil, env.GetSink())
+	httpBindingsSender, err := kncloudevents.NewHTTPMessageSender(nil, env.GetSink())
 	if err != nil {
 		logger.Fatal("error building cloud event client", zap.Error(err))
 	}
diff --git a/vendor/knative.dev/eventing/pkg/apis/sources/v1/container_defaults.go b/vendor/knative.dev/eventing/pkg/apis/sources/v1/container_defaults.go
new file mode 100644
index 000000000..18ec5ced2
--- /dev/null
+++ b/vendor/knative.dev/eventing/pkg/apis/sources/v1/container_defaults.go
@@ -0,0 +1,42 @@
+/*
+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 v1
+
+import (
+	"context"
+	"fmt"
+
+	corev1 "k8s.io/api/core/v1"
+	"knative.dev/pkg/apis"
+)
+
+func (s *ContainerSource) SetDefaults(ctx context.Context) {
+	withName := apis.WithinParent(ctx, s.ObjectMeta)
+	s.Spec.SetDefaults(withName)
+}
+
+func (ss *ContainerSourceSpec) SetDefaults(ctx context.Context) {
+	containers := make([]corev1.Container, 0, len(ss.Template.Spec.Containers))
+	for i, c := range ss.Template.Spec.Containers {
+		// If the Container specified has no name, then default to "<source_name>_<i>".
+		if c.Name == "" {
+			c.Name = fmt.Sprintf("%s-%d", apis.ParentMeta(ctx).Name, i)
+		}
+		containers = append(containers, c)
+	}
+	ss.Template.Spec.Containers = containers
+}
diff --git a/vendor/knative.dev/eventing/pkg/apis/sources/v1/container_lifecycle.go b/vendor/knative.dev/eventing/pkg/apis/sources/v1/container_lifecycle.go
new file mode 100644
index 000000000..058738bed
--- /dev/null
+++ b/vendor/knative.dev/eventing/pkg/apis/sources/v1/container_lifecycle.go
@@ -0,0 +1,110 @@
+/*
+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 v1
+
+import (
+	appsv1 "k8s.io/api/apps/v1"
+	corev1 "k8s.io/api/core/v1"
+	"knative.dev/pkg/apis"
+)
+
+const (
+	// ContainerSourceConditionReady has status True when the ContainerSource is ready to send events.
+	ContainerSourceConditionReady = apis.ConditionReady
+
+	// ContainerSourceConditionSinkBindingReady has status True when the ContainerSource's SinkBinding is ready.
+	ContainerSourceConditionSinkBindingReady apis.ConditionType = "SinkBindingReady"
+
+	// ContainerSourceConditionReceiveAdapterReady has status True when the ContainerSource's ReceiveAdapter is ready.
+	ContainerSourceConditionReceiveAdapterReady apis.ConditionType = "ReceiveAdapterReady"
+)
+
+var containerCondSet = apis.NewLivingConditionSet(
+	ContainerSourceConditionSinkBindingReady,
+	ContainerSourceConditionReceiveAdapterReady,
+)
+
+// GetConditionSet retrieves the condition set for this resource. Implements the KRShaped interface.
+func (*ContainerSource) GetConditionSet() apis.ConditionSet {
+	return containerCondSet
+}
+
+// GetCondition returns the condition currently associated with the given type, or nil.
+func (s *ContainerSourceStatus) GetCondition(t apis.ConditionType) *apis.Condition {
+	return containerCondSet.Manage(s).GetCondition(t)
+}
+
+// GetTopLevelCondition returns the top level condition.
+func (s *ContainerSourceStatus) GetTopLevelCondition() *apis.Condition {
+	return containerCondSet.Manage(s).GetTopLevelCondition()
+}
+
+// IsReady returns true if the resource is ready overall.
+func (s *ContainerSourceStatus) IsReady() bool {
+	return containerCondSet.Manage(s).IsHappy()
+}
+
+// InitializeConditions sets relevant unset conditions to Unknown state.
+func (s *ContainerSourceStatus) InitializeConditions() {
+	containerCondSet.Manage(s).InitializeConditions()
+}
+
+// PropagateSinkBindingStatus uses the availability of the provided Deployment to determine if
+// ContainerSourceConditionSinkBindingReady should be marked as true, false or unknown.
+func (s *ContainerSourceStatus) PropagateSinkBindingStatus(status *SinkBindingStatus) {
+	// Do not copy conditions nor observedGeneration
+	conditions := s.Conditions
+	observedGeneration := s.ObservedGeneration
+	s.SourceStatus = status.SourceStatus
+	s.Conditions = conditions
+	s.ObservedGeneration = observedGeneration
+
+	cond := status.GetCondition(apis.ConditionReady)
+	switch {
+	case cond == nil:
+		containerCondSet.Manage(s).MarkUnknown(ContainerSourceConditionSinkBindingReady, "", "")
+	case cond.Status == corev1.ConditionTrue:
+		containerCondSet.Manage(s).MarkTrue(ContainerSourceConditionSinkBindingReady)
+	case cond.Status == corev1.ConditionFalse:
+		containerCondSet.Manage(s).MarkFalse(ContainerSourceConditionSinkBindingReady, cond.Reason, cond.Message)
+	case cond.Status == corev1.ConditionUnknown:
+		containerCondSet.Manage(s).MarkUnknown(ContainerSourceConditionSinkBindingReady, cond.Reason, cond.Message)
+	default:
+		containerCondSet.Manage(s).MarkUnknown(ContainerSourceConditionSinkBindingReady, cond.Reason, cond.Message)
+	}
+}
+
+// PropagateReceiveAdapterStatus uses the availability of the provided Deployment to determine if
+// ContainerSourceConditionReceiveAdapterReady should be marked as true or false.
+func (s *ContainerSourceStatus) PropagateReceiveAdapterStatus(d *appsv1.Deployment) {
+	deploymentAvailableFound := false
+	for _, cond := range d.Status.Conditions {
+		if cond.Type == appsv1.DeploymentAvailable {
+			deploymentAvailableFound = true
+			if cond.Status == corev1.ConditionTrue {
+				containerCondSet.Manage(s).MarkTrue(ContainerSourceConditionReceiveAdapterReady)
+			} else if cond.Status == corev1.ConditionFalse {
+				containerCondSet.Manage(s).MarkFalse(ContainerSourceConditionReceiveAdapterReady, cond.Reason, cond.Message)
+			} else if cond.Status == corev1.ConditionUnknown {
+				containerCondSet.Manage(s).MarkUnknown(ContainerSourceConditionReceiveAdapterReady, cond.Reason, cond.Message)
+			}
+		}
+	}
+	if !deploymentAvailableFound {
+		containerCondSet.Manage(s).MarkUnknown(ContainerSourceConditionReceiveAdapterReady, "DeploymentUnavailable", "The Deployment '%s' is unavailable.", d.Name)
+	}
+}
diff --git a/vendor/knative.dev/eventing/pkg/apis/sources/v1/container_types.go b/vendor/knative.dev/eventing/pkg/apis/sources/v1/container_types.go
new file mode 100644
index 000000000..5dee6312f
--- /dev/null
+++ b/vendor/knative.dev/eventing/pkg/apis/sources/v1/container_types.go
@@ -0,0 +1,98 @@
+/*
+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 v1
+
+import (
+	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/runtime/schema"
+	"knative.dev/pkg/apis"
+	duckv1 "knative.dev/pkg/apis/duck/v1"
+	"knative.dev/pkg/kmeta"
+)
+
+// +genclient
+// +genreconciler
+// +k8s:deepcopy-gen:interfaces=k8s.io/apimachinery/pkg/runtime.Object
+
+// ContainerSource is the Schema for the containersources API
+type ContainerSource struct {
+	metav1.TypeMeta   `json:",inline"`
+	metav1.ObjectMeta `json:"metadata,omitempty"`
+
+	Spec   ContainerSourceSpec   `json:"spec,omitempty"`
+	Status ContainerSourceStatus `json:"status,omitempty"`
+}
+
+var (
+	_ runtime.Object     = (*ContainerSource)(nil)
+	_ kmeta.OwnerRefable = (*ContainerSource)(nil)
+	_ apis.Validatable   = (*ContainerSource)(nil)
+	_ apis.Defaultable   = (*ContainerSource)(nil)
+	_ apis.HasSpec       = (*ContainerSource)(nil)
+	_ duckv1.KRShaped    = (*ContainerSource)(nil)
+)
+
+// ContainerSourceSpec defines the desired state of ContainerSource
+type ContainerSourceSpec struct {
+	// inherits duck/v1 SourceSpec, which currently provides:
+	// * Sink - a reference to an object that will resolve to a domain name or
+	//   a URI directly to use as the sink.
+	// * CloudEventOverrides - defines overrides to control the output format
+	//   and modifications of the event sent to the sink.
+	duckv1.SourceSpec `json:",inline"`
+
+	// Template describes the pods that will be created
+	Template corev1.PodTemplateSpec `json:"template"`
+}
+
+// GetGroupVersionKind returns the GroupVersionKind.
+func (*ContainerSource) GetGroupVersionKind() schema.GroupVersionKind {
+	return SchemeGroupVersion.WithKind("ContainerSource")
+}
+
+// ContainerSourceStatus defines the observed state of ContainerSource
+type ContainerSourceStatus struct {
+	// inherits duck/v1 SourceStatus, which currently provides:
+	// * ObservedGeneration - the 'Generation' of the Service that was last
+	//   processed by the controller.
+	// * Conditions - the latest available observations of a resource's current
+	//   state.
+	// * SinkURI - the current active sink URI that has been configured for the
+	//   Source.
+	duckv1.SourceStatus `json:",inline"`
+}
+
+// +k8s:deepcopy-gen:interfaces=k8s.io/apimachinery/pkg/runtime.Object
+
+// ContainerSourceList contains a list of ContainerSource
+type ContainerSourceList struct {
+	metav1.TypeMeta `json:",inline"`
+	metav1.ListMeta `json:"metadata,omitempty"`
+	Items           []ContainerSource `json:"items"`
+}
+
+// GetUntypedSpec returns the spec of the ContainerSource.
+func (c *ContainerSource) GetUntypedSpec() interface{} {
+	return c.Spec
+}
+
+// GetStatus retrieves the status of the ContainerSource. Implements the KRShaped interface.
+func (c *ContainerSource) GetStatus() *duckv1.Status {
+	return &c.Status.Status
+}
diff --git a/vendor/knative.dev/eventing/pkg/apis/sources/v1/container_validation.go b/vendor/knative.dev/eventing/pkg/apis/sources/v1/container_validation.go
new file mode 100644
index 000000000..63c7f9cf7
--- /dev/null
+++ b/vendor/knative.dev/eventing/pkg/apis/sources/v1/container_validation.go
@@ -0,0 +1,59 @@
+/*
+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 v1
+
+import (
+	"context"
+
+	corev1 "k8s.io/api/core/v1"
+	"knative.dev/pkg/apis"
+)
+
+func (c *ContainerSource) Validate(ctx context.Context) *apis.FieldError {
+	return c.Spec.Validate(ctx).ViaField("spec")
+}
+
+func (cs *ContainerSourceSpec) Validate(ctx context.Context) *apis.FieldError {
+	var errs *apis.FieldError
+	if fe := cs.Sink.Validate(ctx); fe != nil {
+		errs = errs.Also(fe.ViaField("sink"))
+	}
+
+	// Validate there is at least a container
+	if cs.Template.Spec.Containers == nil || len(cs.Template.Spec.Containers) == 0 {
+		fe := apis.ErrMissingField("containers")
+		errs = errs.Also(fe)
+	} else {
+		for i := range cs.Template.Spec.Containers {
+			if ce := isValidContainer(&cs.Template.Spec.Containers[i]); ce != nil {
+				errs = errs.Also(ce.ViaFieldIndex("containers", i))
+			}
+		}
+	}
+	return errs
+}
+
+func isValidContainer(c *corev1.Container) *apis.FieldError {
+	var errs *apis.FieldError
+	if c.Name == "" {
+		errs = errs.Also(apis.ErrMissingField("name"))
+	}
+	if c.Image == "" {
+		errs = errs.Also(apis.ErrMissingField("image"))
+	}
+	return errs
+}
diff --git a/vendor/knative.dev/eventing/pkg/apis/sources/v1/register.go b/vendor/knative.dev/eventing/pkg/apis/sources/v1/register.go
index 2b4616ca8..f0e2c1986 100644
--- a/vendor/knative.dev/eventing/pkg/apis/sources/v1/register.go
+++ b/vendor/knative.dev/eventing/pkg/apis/sources/v1/register.go
@@ -47,6 +47,10 @@ func addKnownTypes(scheme *runtime.Scheme) error {
 	scheme.AddKnownTypes(SchemeGroupVersion,
 		&ApiServerSource{},
 		&ApiServerSourceList{},
+		&SinkBinding{},
+		&SinkBindingList{},
+		&ContainerSource{},
+		&ContainerSourceList{},
 	)
 	metav1.AddToGroupVersion(scheme, SchemeGroupVersion)
 	return nil
diff --git a/vendor/knative.dev/eventing/pkg/apis/sources/v1/sinkbinding_context.go b/vendor/knative.dev/eventing/pkg/apis/sources/v1/sinkbinding_context.go
new file mode 100644
index 000000000..3672076d8
--- /dev/null
+++ b/vendor/knative.dev/eventing/pkg/apis/sources/v1/sinkbinding_context.go
@@ -0,0 +1,43 @@
+/*
+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 v1
+
+import (
+	"context"
+
+	"knative.dev/pkg/apis"
+)
+
+// sinkURIKey is used as the key for associating information
+// with a context.Context.
+type sinkURIKey struct{}
+
+// WithSinkURI notes on the context for binding that the resolved SinkURI
+// is the provided apis.URL.
+func WithSinkURI(ctx context.Context, uri *apis.URL) context.Context {
+	return context.WithValue(ctx, sinkURIKey{}, uri)
+}
+
+// GetSinkURI accesses the apis.URL for the Sink URI that has been associated
+// with this context.
+func GetSinkURI(ctx context.Context) *apis.URL {
+	value := ctx.Value(sinkURIKey{})
+	if value == nil {
+		return nil
+	}
+	return value.(*apis.URL)
+}
diff --git a/vendor/knative.dev/eventing/pkg/apis/sources/v1/sinkbinding_conversion.go b/vendor/knative.dev/eventing/pkg/apis/sources/v1/sinkbinding_conversion.go
new file mode 100644
index 000000000..46edae3ec
--- /dev/null
+++ b/vendor/knative.dev/eventing/pkg/apis/sources/v1/sinkbinding_conversion.go
@@ -0,0 +1,34 @@
+/*
+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 v1
+
+import (
+	"context"
+	"fmt"
+
+	"knative.dev/pkg/apis"
+)
+
+// ConvertTo implements apis.Convertible
+func (source *SinkBinding) ConvertTo(ctx context.Context, sink apis.Convertible) error {
+	return fmt.Errorf("v1 is the highest known version, got: %T", sink)
+}
+
+// ConvertFrom implements apis.Convertible
+func (sink *SinkBinding) ConvertFrom(ctx context.Context, source apis.Convertible) error {
+	return fmt.Errorf("v1 is the highest known version, got: %T", source)
+}
diff --git a/vendor/knative.dev/eventing/pkg/apis/sources/v1/sinkbinding_defaults.go b/vendor/knative.dev/eventing/pkg/apis/sources/v1/sinkbinding_defaults.go
new file mode 100644
index 000000000..1528d61c3
--- /dev/null
+++ b/vendor/knative.dev/eventing/pkg/apis/sources/v1/sinkbinding_defaults.go
@@ -0,0 +1,34 @@
+/*
+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 v1
+
+import (
+	"context"
+
+	"knative.dev/pkg/apis"
+)
+
+// SetDefaults implements apis.Defaultable
+func (fb *SinkBinding) SetDefaults(ctx context.Context) {
+	if fb.Spec.Subject.Namespace == "" {
+		// Default the subject's namespace to our namespace.
+		fb.Spec.Subject.Namespace = fb.Namespace
+	}
+
+	withNS := apis.WithinParent(ctx, fb.ObjectMeta)
+	fb.Spec.Sink.SetDefaults(withNS)
+}
diff --git a/vendor/knative.dev/eventing/pkg/apis/sources/v1/sinkbinding_lifecycle.go b/vendor/knative.dev/eventing/pkg/apis/sources/v1/sinkbinding_lifecycle.go
new file mode 100644
index 000000000..76f2b795f
--- /dev/null
+++ b/vendor/knative.dev/eventing/pkg/apis/sources/v1/sinkbinding_lifecycle.go
@@ -0,0 +1,160 @@
+/*
+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 v1
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+
+	"go.uber.org/zap"
+
+	corev1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/runtime/schema"
+
+	"knative.dev/pkg/apis"
+	"knative.dev/pkg/apis/duck"
+	duckv1 "knative.dev/pkg/apis/duck/v1"
+	"knative.dev/pkg/logging"
+	"knative.dev/pkg/tracker"
+)
+
+var sbCondSet = apis.NewLivingConditionSet()
+
+// GetConditionSet retrieves the condition set for this resource. Implements the KRShaped interface.
+func (*SinkBinding) GetConditionSet() apis.ConditionSet {
+	return sbCondSet
+}
+
+// GetGroupVersionKind returns the GroupVersionKind.
+func (*SinkBinding) GetGroupVersionKind() schema.GroupVersionKind {
+	return SchemeGroupVersion.WithKind("SinkBinding")
+}
+
+// GetUntypedSpec implements apis.HasSpec
+func (s *SinkBinding) GetUntypedSpec() interface{} {
+	return s.Spec
+}
+
+// GetSubject implements psbinding.Bindable
+func (sb *SinkBinding) GetSubject() tracker.Reference {
+	return sb.Spec.Subject
+}
+
+// GetBindingStatus implements psbinding.Bindable
+func (sb *SinkBinding) GetBindingStatus() duck.BindableStatus {
+	return &sb.Status
+}
+
+// SetObservedGeneration implements psbinding.BindableStatus
+func (sbs *SinkBindingStatus) SetObservedGeneration(gen int64) {
+	sbs.ObservedGeneration = gen
+}
+
+// InitializeConditions populates the SinkBindingStatus's conditions field
+// with all of its conditions configured to Unknown.
+func (sbs *SinkBindingStatus) InitializeConditions() {
+	sbCondSet.Manage(sbs).InitializeConditions()
+}
+
+// MarkBindingUnavailable marks the SinkBinding's Ready condition to False with
+// the provided reason and message.
+func (sbs *SinkBindingStatus) MarkBindingUnavailable(reason, message string) {
+	sbCondSet.Manage(sbs).MarkFalse(SinkBindingConditionReady, reason, message)
+}
+
+// MarkBindingAvailable marks the SinkBinding's Ready condition to True.
+func (sbs *SinkBindingStatus) MarkBindingAvailable() {
+	sbCondSet.Manage(sbs).MarkTrue(SinkBindingConditionReady)
+}
+
+// Do implements psbinding.Bindable
+func (sb *SinkBinding) Do(ctx context.Context, ps *duckv1.WithPod) {
+	// First undo so that we can just unconditionally append below.
+	sb.Undo(ctx, ps)
+
+	uri := GetSinkURI(ctx)
+	if uri == nil {
+		logging.FromContext(ctx).Errorf("No sink URI associated with context for %+v", sb)
+		return
+	}
+
+	var ceOverrides string
+	if sb.Spec.CloudEventOverrides != nil {
+		if co, err := json.Marshal(sb.Spec.SourceSpec.CloudEventOverrides); err != nil {
+			logging.FromContext(ctx).Errorw(fmt.Sprintf("Failed to marshal CloudEventOverrides into JSON for %+v", sb), zap.Error(err))
+		} else if len(co) > 0 {
+			ceOverrides = string(co)
+		}
+	}
+
+	spec := ps.Spec.Template.Spec
+	for i := range spec.InitContainers {
+		spec.InitContainers[i].Env = append(spec.InitContainers[i].Env, corev1.EnvVar{
+			Name:  "K_SINK",
+			Value: uri.String(),
+		})
+		spec.InitContainers[i].Env = append(spec.InitContainers[i].Env, corev1.EnvVar{
+			Name:  "K_CE_OVERRIDES",
+			Value: ceOverrides,
+		})
+	}
+	for i := range spec.Containers {
+		spec.Containers[i].Env = append(spec.Containers[i].Env, corev1.EnvVar{
+			Name:  "K_SINK",
+			Value: uri.String(),
+		})
+		spec.Containers[i].Env = append(spec.Containers[i].Env, corev1.EnvVar{
+			Name:  "K_CE_OVERRIDES",
+			Value: ceOverrides,
+		})
+	}
+}
+
+func (sb *SinkBinding) Undo(ctx context.Context, ps *duckv1.WithPod) {
+	spec := ps.Spec.Template.Spec
+	for i, c := range spec.InitContainers {
+		if len(c.Env) == 0 {
+			continue
+		}
+		env := make([]corev1.EnvVar, 0, len(spec.InitContainers[i].Env))
+		for j, ev := range c.Env {
+			switch ev.Name {
+			case "K_SINK", "K_CE_OVERRIDES":
+				continue
+			default:
+				env = append(env, spec.InitContainers[i].Env[j])
+			}
+		}
+		spec.InitContainers[i].Env = env
+	}
+	for i, c := range spec.Containers {
+		if len(c.Env) == 0 {
+			continue
+		}
+		env := make([]corev1.EnvVar, 0, len(spec.Containers[i].Env))
+		for j, ev := range c.Env {
+			switch ev.Name {
+			case "K_SINK", "K_CE_OVERRIDES":
+				continue
+			default:
+				env = append(env, spec.Containers[i].Env[j])
+			}
+		}
+		spec.Containers[i].Env = env
+	}
+}
diff --git a/vendor/knative.dev/eventing/pkg/apis/sources/v1/sinkbinding_types.go b/vendor/knative.dev/eventing/pkg/apis/sources/v1/sinkbinding_types.go
new file mode 100644
index 000000000..8317c2b06
--- /dev/null
+++ b/vendor/knative.dev/eventing/pkg/apis/sources/v1/sinkbinding_types.go
@@ -0,0 +1,100 @@
+/*
+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 v1
+
+import (
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime"
+	"knative.dev/pkg/apis"
+	duckv1 "knative.dev/pkg/apis/duck/v1"
+	"knative.dev/pkg/kmeta"
+)
+
+// +genclient
+// +k8s:deepcopy-gen:interfaces=k8s.io/apimachinery/pkg/runtime.Object
+// +k8s:defaulter-gen=true
+
+// SinkBinding describes a Binding that is also a Source.
+// The `sink` (from the Source duck) is resolved to a URL and
+// then projected into the `subject` by augmenting the runtime
+// contract of the referenced containers to have a `K_SINK`
+// environment variable holding the endpoint to which to send
+// cloud events.
+type SinkBinding struct {
+	metav1.TypeMeta   `json:",inline"`
+	metav1.ObjectMeta `json:"metadata,omitempty"`
+
+	Spec   SinkBindingSpec   `json:"spec"`
+	Status SinkBindingStatus `json:"status"`
+}
+
+// Check the interfaces that SinkBinding should be implementing.
+var (
+	_ runtime.Object     = (*SinkBinding)(nil)
+	_ kmeta.OwnerRefable = (*SinkBinding)(nil)
+	_ apis.Validatable   = (*SinkBinding)(nil)
+	_ apis.Defaultable   = (*SinkBinding)(nil)
+	_ apis.HasSpec       = (*SinkBinding)(nil)
+	_ duckv1.KRShaped    = (*SinkBinding)(nil)
+)
+
+// SinkBindingSpec holds the desired state of the SinkBinding (from the client).
+type SinkBindingSpec struct {
+	// inherits duck/v1 SourceSpec, which currently provides:
+	// * Sink - a reference to an object that will resolve to a domain name or
+	//   a URI directly to use as the sink.
+	// * CloudEventOverrides - defines overrides to control the output format
+	//   and modifications of the event sent to the sink.
+	duckv1.SourceSpec `json:",inline"`
+
+	// inherits duck/v1 BindingSpec, which currently provides:
+	// * Subject - Subject references the resource(s) whose "runtime contract"
+	//   should be augmented by Binding implementations.
+	duckv1.BindingSpec `json:",inline"`
+}
+
+const (
+	// SinkBindingConditionReady is configured to indicate whether the Binding
+	// has been configured for resources subject to its runtime contract.
+	SinkBindingConditionReady = apis.ConditionReady
+)
+
+// SinkBindingStatus communicates the observed state of the SinkBinding (from the controller).
+type SinkBindingStatus struct {
+	// inherits duck/v1 SourceStatus, which currently provides:
+	// * ObservedGeneration - the 'Generation' of the Service that was last
+	//   processed by the controller.
+	// * Conditions - the latest available observations of a resource's current
+	//   state.
+	// * SinkURI - the current active sink URI that has been configured for the
+	//   Source.
+	duckv1.SourceStatus `json:",inline"`
+}
+
+// +k8s:deepcopy-gen:interfaces=k8s.io/apimachinery/pkg/runtime.Object
+
+// SinkBindingList contains a list of SinkBinding
+type SinkBindingList struct {
+	metav1.TypeMeta `json:",inline"`
+	metav1.ListMeta `json:"metadata,omitempty"`
+	Items           []SinkBinding `json:"items"`
+}
+
+// GetStatus retrieves the status of the SinkBinding. Implements the KRShaped interface.
+func (s *SinkBinding) GetStatus() *duckv1.Status {
+	return &s.Status.Status
+}
diff --git a/vendor/knative.dev/eventing/pkg/apis/sources/v1/sinkbinding_validation.go b/vendor/knative.dev/eventing/pkg/apis/sources/v1/sinkbinding_validation.go
new file mode 100644
index 000000000..319742d20
--- /dev/null
+++ b/vendor/knative.dev/eventing/pkg/apis/sources/v1/sinkbinding_validation.go
@@ -0,0 +1,38 @@
+/*
+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 v1
+
+import (
+	"context"
+
+	"knative.dev/pkg/apis"
+)
+
+// Validate implements apis.Validatable
+func (fb *SinkBinding) Validate(ctx context.Context) *apis.FieldError {
+	err := fb.Spec.Validate(ctx).ViaField("spec")
+	if fb.Spec.Subject.Namespace != "" && fb.Namespace != fb.Spec.Subject.Namespace {
+		err = err.Also(apis.ErrInvalidValue(fb.Spec.Subject.Namespace, "spec.subject.namespace"))
+	}
+	return err
+}
+
+// Validate implements apis.Validatable
+func (fbs *SinkBindingSpec) Validate(ctx context.Context) *apis.FieldError {
+	return fbs.Subject.Validate(ctx).ViaField("subject").Also(
+		fbs.Sink.Validate(ctx).ViaField("sink"))
+}
diff --git a/vendor/knative.dev/eventing/pkg/apis/sources/v1/zz_generated.deepcopy.go b/vendor/knative.dev/eventing/pkg/apis/sources/v1/zz_generated.deepcopy.go
index 52693d43a..3435b2959 100644
--- a/vendor/knative.dev/eventing/pkg/apis/sources/v1/zz_generated.deepcopy.go
+++ b/vendor/knative.dev/eventing/pkg/apis/sources/v1/zz_generated.deepcopy.go
@@ -168,3 +168,195 @@ func (in *ApiServerSourceStatus) DeepCopy() *ApiServerSourceStatus {
 	in.DeepCopyInto(out)
 	return out
 }
+
+// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
+func (in *ContainerSource) DeepCopyInto(out *ContainerSource) {
+	*out = *in
+	out.TypeMeta = in.TypeMeta
+	in.ObjectMeta.DeepCopyInto(&out.ObjectMeta)
+	in.Spec.DeepCopyInto(&out.Spec)
+	in.Status.DeepCopyInto(&out.Status)
+	return
+}
+
+// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new ContainerSource.
+func (in *ContainerSource) DeepCopy() *ContainerSource {
+	if in == nil {
+		return nil
+	}
+	out := new(ContainerSource)
+	in.DeepCopyInto(out)
+	return out
+}
+
+// DeepCopyObject is an autogenerated deepcopy function, copying the receiver, creating a new runtime.Object.
+func (in *ContainerSource) DeepCopyObject() runtime.Object {
+	if c := in.DeepCopy(); c != nil {
+		return c
+	}
+	return nil
+}
+
+// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
+func (in *ContainerSourceList) DeepCopyInto(out *ContainerSourceList) {
+	*out = *in
+	out.TypeMeta = in.TypeMeta
+	in.ListMeta.DeepCopyInto(&out.ListMeta)
+	if in.Items != nil {
+		in, out := &in.Items, &out.Items
+		*out = make([]ContainerSource, len(*in))
+		for i := range *in {
+			(*in)[i].DeepCopyInto(&(*out)[i])
+		}
+	}
+	return
+}
+
+// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new ContainerSourceList.
+func (in *ContainerSourceList) DeepCopy() *ContainerSourceList {
+	if in == nil {
+		return nil
+	}
+	out := new(ContainerSourceList)
+	in.DeepCopyInto(out)
+	return out
+}
+
+// DeepCopyObject is an autogenerated deepcopy function, copying the receiver, creating a new runtime.Object.
+func (in *ContainerSourceList) DeepCopyObject() runtime.Object {
+	if c := in.DeepCopy(); c != nil {
+		return c
+	}
+	return nil
+}
+
+// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
+func (in *ContainerSourceSpec) DeepCopyInto(out *ContainerSourceSpec) {
+	*out = *in
+	in.SourceSpec.DeepCopyInto(&out.SourceSpec)
+	in.Template.DeepCopyInto(&out.Template)
+	return
+}
+
+// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new ContainerSourceSpec.
+func (in *ContainerSourceSpec) DeepCopy() *ContainerSourceSpec {
+	if in == nil {
+		return nil
+	}
+	out := new(ContainerSourceSpec)
+	in.DeepCopyInto(out)
+	return out
+}
+
+// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
+func (in *ContainerSourceStatus) DeepCopyInto(out *ContainerSourceStatus) {
+	*out = *in
+	in.SourceStatus.DeepCopyInto(&out.SourceStatus)
+	return
+}
+
+// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new ContainerSourceStatus.
+func (in *ContainerSourceStatus) DeepCopy() *ContainerSourceStatus {
+	if in == nil {
+		return nil
+	}
+	out := new(ContainerSourceStatus)
+	in.DeepCopyInto(out)
+	return out
+}
+
+// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
+func (in *SinkBinding) DeepCopyInto(out *SinkBinding) {
+	*out = *in
+	out.TypeMeta = in.TypeMeta
+	in.ObjectMeta.DeepCopyInto(&out.ObjectMeta)
+	in.Spec.DeepCopyInto(&out.Spec)
+	in.Status.DeepCopyInto(&out.Status)
+	return
+}
+
+// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new SinkBinding.
+func (in *SinkBinding) DeepCopy() *SinkBinding {
+	if in == nil {
+		return nil
+	}
+	out := new(SinkBinding)
+	in.DeepCopyInto(out)
+	return out
+}
+
+// DeepCopyObject is an autogenerated deepcopy function, copying the receiver, creating a new runtime.Object.
+func (in *SinkBinding) DeepCopyObject() runtime.Object {
+	if c := in.DeepCopy(); c != nil {
+		return c
+	}
+	return nil
+}
+
+// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
+func (in *SinkBindingList) DeepCopyInto(out *SinkBindingList) {
+	*out = *in
+	out.TypeMeta = in.TypeMeta
+	in.ListMeta.DeepCopyInto(&out.ListMeta)
+	if in.Items != nil {
+		in, out := &in.Items, &out.Items
+		*out = make([]SinkBinding, len(*in))
+		for i := range *in {
+			(*in)[i].DeepCopyInto(&(*out)[i])
+		}
+	}
+	return
+}
+
+// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new SinkBindingList.
+func (in *SinkBindingList) DeepCopy() *SinkBindingList {
+	if in == nil {
+		return nil
+	}
+	out := new(SinkBindingList)
+	in.DeepCopyInto(out)
+	return out
+}
+
+// DeepCopyObject is an autogenerated deepcopy function, copying the receiver, creating a new runtime.Object.
+func (in *SinkBindingList) DeepCopyObject() runtime.Object {
+	if c := in.DeepCopy(); c != nil {
+		return c
+	}
+	return nil
+}
+
+// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
+func (in *SinkBindingSpec) DeepCopyInto(out *SinkBindingSpec) {
+	*out = *in
+	in.SourceSpec.DeepCopyInto(&out.SourceSpec)
+	in.BindingSpec.DeepCopyInto(&out.BindingSpec)
+	return
+}
+
+// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new SinkBindingSpec.
+func (in *SinkBindingSpec) DeepCopy() *SinkBindingSpec {
+	if in == nil {
+		return nil
+	}
+	out := new(SinkBindingSpec)
+	in.DeepCopyInto(out)
+	return out
+}
+
+// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
+func (in *SinkBindingStatus) DeepCopyInto(out *SinkBindingStatus) {
+	*out = *in
+	in.SourceStatus.DeepCopyInto(&out.SourceStatus)
+	return
+}
+
+// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new SinkBindingStatus.
+func (in *SinkBindingStatus) DeepCopy() *SinkBindingStatus {
+	if in == nil {
+		return nil
+	}
+	out := new(SinkBindingStatus)
+	in.DeepCopyInto(out)
+	return out
+}
diff --git a/vendor/knative.dev/eventing/pkg/apis/sources/v1alpha1/sinkbinding_conversion.go b/vendor/knative.dev/eventing/pkg/apis/sources/v1alpha1/sinkbinding_conversion.go
index 11e173d7e..ba35d5650 100644
--- a/vendor/knative.dev/eventing/pkg/apis/sources/v1alpha1/sinkbinding_conversion.go
+++ b/vendor/knative.dev/eventing/pkg/apis/sources/v1alpha1/sinkbinding_conversion.go
@@ -1,5 +1,5 @@
 /*
-Copyright 2020 The Knative Authors.
+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.
diff --git a/vendor/knative.dev/eventing/pkg/apis/sources/v1alpha2/sinkbinding_conversion.go b/vendor/knative.dev/eventing/pkg/apis/sources/v1alpha2/sinkbinding_conversion.go
index 7b116bc56..7f2621ab7 100644
--- a/vendor/knative.dev/eventing/pkg/apis/sources/v1alpha2/sinkbinding_conversion.go
+++ b/vendor/knative.dev/eventing/pkg/apis/sources/v1alpha2/sinkbinding_conversion.go
@@ -55,6 +55,6 @@ func (sink *SinkBinding) ConvertFrom(ctx context.Context, obj apis.Convertible)
 		sink.Status.SourceStatus = source.Status.SourceStatus
 		return nil
 	default:
-		return apis.ConvertToViaProxy(ctx, source, &v1beta1.SinkBinding{}, sink)
+		return apis.ConvertFromViaProxy(ctx, source, &v1beta1.SinkBinding{}, sink)
 	}
 }
diff --git a/vendor/knative.dev/eventing/pkg/apis/sources/v1beta1/sinkbinding_conversion.go b/vendor/knative.dev/eventing/pkg/apis/sources/v1beta1/sinkbinding_conversion.go
index de89653fa..ed196e250 100644
--- a/vendor/knative.dev/eventing/pkg/apis/sources/v1beta1/sinkbinding_conversion.go
+++ b/vendor/knative.dev/eventing/pkg/apis/sources/v1beta1/sinkbinding_conversion.go
@@ -18,17 +18,43 @@ package v1beta1
 
 import (
 	"context"
-	"fmt"
 
+	v1 "knative.dev/eventing/pkg/apis/sources/v1"
 	"knative.dev/pkg/apis"
+	duckv1 "knative.dev/pkg/apis/duck/v1"
+	duckv1beta1 "knative.dev/pkg/apis/duck/v1beta1"
 )
 
-// ConvertTo implements apis.Convertible
-func (source *SinkBinding) ConvertTo(ctx context.Context, sink apis.Convertible) error {
-	return fmt.Errorf("v1beta1 is the highest known version, got: %T", sink)
+// ConvertTo implements apis.Convertible.
+// Converts source from v1beta1.SinkBinding into a higher version.
+func (source *SinkBinding) ConvertTo(ctx context.Context, obj apis.Convertible) error {
+	switch sink := obj.(type) {
+	case *v1.SinkBinding:
+		sink.ObjectMeta = source.ObjectMeta
+		sink.Spec.SourceSpec = source.Spec.SourceSpec
+		sink.Spec.BindingSpec = duckv1.BindingSpec{
+			Subject: source.Spec.BindingSpec.Subject,
+		}
+		sink.Status.SourceStatus = source.Status.SourceStatus
+		return nil
+	default:
+		return apis.ConvertToViaProxy(ctx, source, &v1.SinkBinding{}, sink)
+	}
 }
 
-// ConvertFrom implements apis.Convertible
-func (sink *SinkBinding) ConvertFrom(ctx context.Context, source apis.Convertible) error {
-	return fmt.Errorf("v1beta1 is the highest known version, got: %T", source)
+// ConvertFrom implements apis.Convertible.
+// Converts obj from a higher version into v1beta1.SinkBinding.
+func (sink *SinkBinding) ConvertFrom(ctx context.Context, obj apis.Convertible) error {
+	switch source := obj.(type) {
+	case *v1.SinkBinding:
+		sink.ObjectMeta = source.ObjectMeta
+		sink.Spec.SourceSpec = source.Spec.SourceSpec
+		sink.Spec.BindingSpec = duckv1beta1.BindingSpec{
+			Subject: source.Spec.BindingSpec.Subject,
+		}
+		sink.Status.SourceStatus = source.Status.SourceStatus
+		return nil
+	default:
+		return apis.ConvertFromViaProxy(ctx, source, &v1.SinkBinding{}, sink)
+	}
 }
diff --git a/vendor/knative.dev/eventing/pkg/channel/message_dispatcher.go b/vendor/knative.dev/eventing/pkg/channel/message_dispatcher.go
index 76c59e808..7e5d70f07 100644
--- a/vendor/knative.dev/eventing/pkg/channel/message_dispatcher.go
+++ b/vendor/knative.dev/eventing/pkg/channel/message_dispatcher.go
@@ -51,7 +51,7 @@ var _ MessageDispatcher = &MessageDispatcherImpl{}
 
 // MessageDispatcherImpl dispatches events to a destination over HTTP.
 type MessageDispatcherImpl struct {
-	sender           *kncloudevents.HttpMessageSender
+	sender           *kncloudevents.HTTPMessageSender
 	supportedSchemes sets.String
 
 	logger *zap.Logger
@@ -65,7 +65,7 @@ func NewMessageDispatcher(logger *zap.Logger) *MessageDispatcherImpl {
 
 // NewMessageDispatcherFromConfig creates a new Message dispatcher based on config.
 func NewMessageDispatcherFromConfig(logger *zap.Logger, config EventDispatcherConfig) *MessageDispatcherImpl {
-	sender, err := kncloudevents.NewHttpMessageSender(&config.ConnectionArgs, "")
+	sender, err := kncloudevents.NewHTTPMessageSender(&config.ConnectionArgs, "")
 	if err != nil {
 		logger.Fatal("Unable to create cloudevents binding sender", zap.Error(err))
 	}
@@ -73,7 +73,7 @@ func NewMessageDispatcherFromConfig(logger *zap.Logger, config EventDispatcherCo
 }
 
 // NewMessageDispatcherFromConfig creates a new event dispatcher.
-func NewMessageDispatcherFromSender(logger *zap.Logger, sender *kncloudevents.HttpMessageSender) *MessageDispatcherImpl {
+func NewMessageDispatcherFromSender(logger *zap.Logger, sender *kncloudevents.HTTPMessageSender) *MessageDispatcherImpl {
 	return &MessageDispatcherImpl{
 		sender:           sender,
 		supportedSchemes: sets.NewString("http", "https"),
@@ -183,9 +183,9 @@ func (d *MessageDispatcherImpl) executeRequest(ctx context.Context, url *url.URL
 	}
 
 	if span.IsRecordingEvents() {
-		err = kncloudevents.WriteHttpRequestWithAdditionalHeaders(ctx, message, req, additionalHeaders, tracing.PopulateSpan(span))
+		err = kncloudevents.WriteHTTPRequestWithAdditionalHeaders(ctx, message, req, additionalHeaders, tracing.PopulateSpan(span))
 	} else {
-		err = kncloudevents.WriteHttpRequestWithAdditionalHeaders(ctx, message, req, additionalHeaders)
+		err = kncloudevents.WriteHTTPRequestWithAdditionalHeaders(ctx, message, req, additionalHeaders)
 	}
 	if err != nil {
 		return ctx, nil, nil, err
diff --git a/vendor/knative.dev/eventing/pkg/channel/message_receiver.go b/vendor/knative.dev/eventing/pkg/channel/message_receiver.go
index c0d7129ca..6546c7b51 100644
--- a/vendor/knative.dev/eventing/pkg/channel/message_receiver.go
+++ b/vendor/knative.dev/eventing/pkg/channel/message_receiver.go
@@ -53,7 +53,7 @@ func (e UnknownHostError) Error() string {
 // MessageReceiver starts a server to receive new events for the channel dispatcher. The new
 // event is emitted via the receiver function.
 type MessageReceiver struct {
-	httpBindingsReceiver *kncloudevents.HttpMessageReceiver
+	httpBindingsReceiver *kncloudevents.HTTPMessageReceiver
 	receiverFunc         UnbufferedMessageReceiverFunc
 	logger               *zap.Logger
 	hostToChannelFunc    ResolveChannelFromHostFunc
@@ -85,7 +85,7 @@ func ResolveMessageChannelFromHostHeader(hostToChannelFunc ResolveChannelFromHos
 // NewMessageReceiver creates an event receiver passing new events to the
 // receiverFunc.
 func NewMessageReceiver(receiverFunc UnbufferedMessageReceiverFunc, logger *zap.Logger, opts ...MessageReceiverOptions) (*MessageReceiver, error) {
-	bindingsReceiver := kncloudevents.NewHttpMessageReceiver(8080)
+	bindingsReceiver := kncloudevents.NewHTTPMessageReceiver(8080)
 	receiver := &MessageReceiver{
 		httpBindingsReceiver: bindingsReceiver,
 		receiverFunc:         receiverFunc,
diff --git a/vendor/knative.dev/eventing/pkg/client/clientset/versioned/typed/sources/v1/containersource.go b/vendor/knative.dev/eventing/pkg/client/clientset/versioned/typed/sources/v1/containersource.go
new file mode 100644
index 000000000..da607fb51
--- /dev/null
+++ b/vendor/knative.dev/eventing/pkg/client/clientset/versioned/typed/sources/v1/containersource.go
@@ -0,0 +1,195 @@
+/*
+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.
+*/
+
+// Code generated by client-gen. DO NOT EDIT.
+
+package v1
+
+import (
+	"context"
+	"time"
+
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	types "k8s.io/apimachinery/pkg/types"
+	watch "k8s.io/apimachinery/pkg/watch"
+	rest "k8s.io/client-go/rest"
+	v1 "knative.dev/eventing/pkg/apis/sources/v1"
+	scheme "knative.dev/eventing/pkg/client/clientset/versioned/scheme"
+)
+
+// ContainerSourcesGetter has a method to return a ContainerSourceInterface.
+// A group's client should implement this interface.
+type ContainerSourcesGetter interface {
+	ContainerSources(namespace string) ContainerSourceInterface
+}
+
+// ContainerSourceInterface has methods to work with ContainerSource resources.
+type ContainerSourceInterface interface {
+	Create(ctx context.Context, containerSource *v1.ContainerSource, opts metav1.CreateOptions) (*v1.ContainerSource, error)
+	Update(ctx context.Context, containerSource *v1.ContainerSource, opts metav1.UpdateOptions) (*v1.ContainerSource, error)
+	UpdateStatus(ctx context.Context, containerSource *v1.ContainerSource, opts metav1.UpdateOptions) (*v1.ContainerSource, error)
+	Delete(ctx context.Context, name string, opts metav1.DeleteOptions) error
+	DeleteCollection(ctx context.Context, opts metav1.DeleteOptions, listOpts metav1.ListOptions) error
+	Get(ctx context.Context, name string, opts metav1.GetOptions) (*v1.ContainerSource, error)
+	List(ctx context.Context, opts metav1.ListOptions) (*v1.ContainerSourceList, error)
+	Watch(ctx context.Context, opts metav1.ListOptions) (watch.Interface, error)
+	Patch(ctx context.Context, name string, pt types.PatchType, data []byte, opts metav1.PatchOptions, subresources ...string) (result *v1.ContainerSource, err error)
+	ContainerSourceExpansion
+}
+
+// containerSources implements ContainerSourceInterface
+type containerSources struct {
+	client rest.Interface
+	ns     string
+}
+
+// newContainerSources returns a ContainerSources
+func newContainerSources(c *SourcesV1Client, namespace string) *containerSources {
+	return &containerSources{
+		client: c.RESTClient(),
+		ns:     namespace,
+	}
+}
+
+// Get takes name of the containerSource, and returns the corresponding containerSource object, and an error if there is any.
+func (c *containerSources) Get(ctx context.Context, name string, options metav1.GetOptions) (result *v1.ContainerSource, err error) {
+	result = &v1.ContainerSource{}
+	err = c.client.Get().
+		Namespace(c.ns).
+		Resource("containersources").
+		Name(name).
+		VersionedParams(&options, scheme.ParameterCodec).
+		Do(ctx).
+		Into(result)
+	return
+}
+
+// List takes label and field selectors, and returns the list of ContainerSources that match those selectors.
+func (c *containerSources) List(ctx context.Context, opts metav1.ListOptions) (result *v1.ContainerSourceList, err error) {
+	var timeout time.Duration
+	if opts.TimeoutSeconds != nil {
+		timeout = time.Duration(*opts.TimeoutSeconds) * time.Second
+	}
+	result = &v1.ContainerSourceList{}
+	err = c.client.Get().
+		Namespace(c.ns).
+		Resource("containersources").
+		VersionedParams(&opts, scheme.ParameterCodec).
+		Timeout(timeout).
+		Do(ctx).
+		Into(result)
+	return
+}
+
+// Watch returns a watch.Interface that watches the requested containerSources.
+func (c *containerSources) Watch(ctx context.Context, opts metav1.ListOptions) (watch.Interface, error) {
+	var timeout time.Duration
+	if opts.TimeoutSeconds != nil {
+		timeout = time.Duration(*opts.TimeoutSeconds) * time.Second
+	}
+	opts.Watch = true
+	return c.client.Get().
+		Namespace(c.ns).
+		Resource("containersources").
+		VersionedParams(&opts, scheme.ParameterCodec).
+		Timeout(timeout).
+		Watch(ctx)
+}
+
+// Create takes the representation of a containerSource and creates it.  Returns the server's representation of the containerSource, and an error, if there is any.
+func (c *containerSources) Create(ctx context.Context, containerSource *v1.ContainerSource, opts metav1.CreateOptions) (result *v1.ContainerSource, err error) {
+	result = &v1.ContainerSource{}
+	err = c.client.Post().
+		Namespace(c.ns).
+		Resource("containersources").
+		VersionedParams(&opts, scheme.ParameterCodec).
+		Body(containerSource).
+		Do(ctx).
+		Into(result)
+	return
+}
+
+// Update takes the representation of a containerSource and updates it. Returns the server's representation of the containerSource, and an error, if there is any.
+func (c *containerSources) Update(ctx context.Context, containerSource *v1.ContainerSource, opts metav1.UpdateOptions) (result *v1.ContainerSource, err error) {
+	result = &v1.ContainerSource{}
+	err = c.client.Put().
+		Namespace(c.ns).
+		Resource("containersources").
+		Name(containerSource.Name).
+		VersionedParams(&opts, scheme.ParameterCodec).
+		Body(containerSource).
+		Do(ctx).
+		Into(result)
+	return
+}
+
+// UpdateStatus was generated because the type contains a Status member.
+// Add a +genclient:noStatus comment above the type to avoid generating UpdateStatus().
+func (c *containerSources) UpdateStatus(ctx context.Context, containerSource *v1.ContainerSource, opts metav1.UpdateOptions) (result *v1.ContainerSource, err error) {
+	result = &v1.ContainerSource{}
+	err = c.client.Put().
+		Namespace(c.ns).
+		Resource("containersources").
+		Name(containerSource.Name).
+		SubResource("status").
+		VersionedParams(&opts, scheme.ParameterCodec).
+		Body(containerSource).
+		Do(ctx).
+		Into(result)
+	return
+}
+
+// Delete takes name of the containerSource and deletes it. Returns an error if one occurs.
+func (c *containerSources) Delete(ctx context.Context, name string, opts metav1.DeleteOptions) error {
+	return c.client.Delete().
+		Namespace(c.ns).
+		Resource("containersources").
+		Name(name).
+		Body(&opts).
+		Do(ctx).
+		Error()
+}
+
+// DeleteCollection deletes a collection of objects.
+func (c *containerSources) DeleteCollection(ctx context.Context, opts metav1.DeleteOptions, listOpts metav1.ListOptions) error {
+	var timeout time.Duration
+	if listOpts.TimeoutSeconds != nil {
+		timeout = time.Duration(*listOpts.TimeoutSeconds) * time.Second
+	}
+	return c.client.Delete().
+		Namespace(c.ns).
+		Resource("containersources").
+		VersionedParams(&listOpts, scheme.ParameterCodec).
+		Timeout(timeout).
+		Body(&opts).
+		Do(ctx).
+		Error()
+}
+
+// Patch applies the patch and returns the patched containerSource.
+func (c *containerSources) Patch(ctx context.Context, name string, pt types.PatchType, data []byte, opts metav1.PatchOptions, subresources ...string) (result *v1.ContainerSource, err error) {
+	result = &v1.ContainerSource{}
+	err = c.client.Patch(pt).
+		Namespace(c.ns).
+		Resource("containersources").
+		Name(name).
+		SubResource(subresources...).
+		VersionedParams(&opts, scheme.ParameterCodec).
+		Body(data).
+		Do(ctx).
+		Into(result)
+	return
+}
diff --git a/vendor/knative.dev/eventing/pkg/client/clientset/versioned/typed/sources/v1/fake/fake_containersource.go b/vendor/knative.dev/eventing/pkg/client/clientset/versioned/typed/sources/v1/fake/fake_containersource.go
new file mode 100644
index 000000000..da23d80cd
--- /dev/null
+++ b/vendor/knative.dev/eventing/pkg/client/clientset/versioned/typed/sources/v1/fake/fake_containersource.go
@@ -0,0 +1,142 @@
+/*
+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.
+*/
+
+// Code generated by client-gen. DO NOT EDIT.
+
+package fake
+
+import (
+	"context"
+
+	v1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	labels "k8s.io/apimachinery/pkg/labels"
+	schema "k8s.io/apimachinery/pkg/runtime/schema"
+	types "k8s.io/apimachinery/pkg/types"
+	watch "k8s.io/apimachinery/pkg/watch"
+	testing "k8s.io/client-go/testing"
+	sourcesv1 "knative.dev/eventing/pkg/apis/sources/v1"
+)
+
+// FakeContainerSources implements ContainerSourceInterface
+type FakeContainerSources struct {
+	Fake *FakeSourcesV1
+	ns   string
+}
+
+var containersourcesResource = schema.GroupVersionResource{Group: "sources.knative.dev", Version: "v1", Resource: "containersources"}
+
+var containersourcesKind = schema.GroupVersionKind{Group: "sources.knative.dev", Version: "v1", Kind: "ContainerSource"}
+
+// Get takes name of the containerSource, and returns the corresponding containerSource object, and an error if there is any.
+func (c *FakeContainerSources) Get(ctx context.Context, name string, options v1.GetOptions) (result *sourcesv1.ContainerSource, err error) {
+	obj, err := c.Fake.
+		Invokes(testing.NewGetAction(containersourcesResource, c.ns, name), &sourcesv1.ContainerSource{})
+
+	if obj == nil {
+		return nil, err
+	}
+	return obj.(*sourcesv1.ContainerSource), err
+}
+
+// List takes label and field selectors, and returns the list of ContainerSources that match those selectors.
+func (c *FakeContainerSources) List(ctx context.Context, opts v1.ListOptions) (result *sourcesv1.ContainerSourceList, err error) {
+	obj, err := c.Fake.
+		Invokes(testing.NewListAction(containersourcesResource, containersourcesKind, c.ns, opts), &sourcesv1.ContainerSourceList{})
+
+	if obj == nil {
+		return nil, err
+	}
+
+	label, _, _ := testing.ExtractFromListOptions(opts)
+	if label == nil {
+		label = labels.Everything()
+	}
+	list := &sourcesv1.ContainerSourceList{ListMeta: obj.(*sourcesv1.ContainerSourceList).ListMeta}
+	for _, item := range obj.(*sourcesv1.ContainerSourceList).Items {
+		if label.Matches(labels.Set(item.Labels)) {
+			list.Items = append(list.Items, item)
+		}
+	}
+	return list, err
+}
+
+// Watch returns a watch.Interface that watches the requested containerSources.
+func (c *FakeContainerSources) Watch(ctx context.Context, opts v1.ListOptions) (watch.Interface, error) {
+	return c.Fake.
+		InvokesWatch(testing.NewWatchAction(containersourcesResource, c.ns, opts))
+
+}
+
+// Create takes the representation of a containerSource and creates it.  Returns the server's representation of the containerSource, and an error, if there is any.
+func (c *FakeContainerSources) Create(ctx context.Context, containerSource *sourcesv1.ContainerSource, opts v1.CreateOptions) (result *sourcesv1.ContainerSource, err error) {
+	obj, err := c.Fake.
+		Invokes(testing.NewCreateAction(containersourcesResource, c.ns, containerSource), &sourcesv1.ContainerSource{})
+
+	if obj == nil {
+		return nil, err
+	}
+	return obj.(*sourcesv1.ContainerSource), err
+}
+
+// Update takes the representation of a containerSource and updates it. Returns the server's representation of the containerSource, and an error, if there is any.
+func (c *FakeContainerSources) Update(ctx context.Context, containerSource *sourcesv1.ContainerSource, opts v1.UpdateOptions) (result *sourcesv1.ContainerSource, err error) {
+	obj, err := c.Fake.
+		Invokes(testing.NewUpdateAction(containersourcesResource, c.ns, containerSource), &sourcesv1.ContainerSource{})
+
+	if obj == nil {
+		return nil, err
+	}
+	return obj.(*sourcesv1.ContainerSource), err
+}
+
+// UpdateStatus was generated because the type contains a Status member.
+// Add a +genclient:noStatus comment above the type to avoid generating UpdateStatus().
+func (c *FakeContainerSources) UpdateStatus(ctx context.Context, containerSource *sourcesv1.ContainerSource, opts v1.UpdateOptions) (*sourcesv1.ContainerSource, error) {
+	obj, err := c.Fake.
+		Invokes(testing.NewUpdateSubresourceAction(containersourcesResource, "status", c.ns, containerSource), &sourcesv1.ContainerSource{})
+
+	if obj == nil {
+		return nil, err
+	}
+	return obj.(*sourcesv1.ContainerSource), err
+}
+
+// Delete takes name of the containerSource and deletes it. Returns an error if one occurs.
+func (c *FakeContainerSources) Delete(ctx context.Context, name string, opts v1.DeleteOptions) error {
+	_, err := c.Fake.
+		Invokes(testing.NewDeleteAction(containersourcesResource, c.ns, name), &sourcesv1.ContainerSource{})
+
+	return err
+}
+
+// DeleteCollection deletes a collection of objects.
+func (c *FakeContainerSources) DeleteCollection(ctx context.Context, opts v1.DeleteOptions, listOpts v1.ListOptions) error {
+	action := testing.NewDeleteCollectionAction(containersourcesResource, c.ns, listOpts)
+
+	_, err := c.Fake.Invokes(action, &sourcesv1.ContainerSourceList{})
+	return err
+}
+
+// Patch applies the patch and returns the patched containerSource.
+func (c *FakeContainerSources) Patch(ctx context.Context, name string, pt types.PatchType, data []byte, opts v1.PatchOptions, subresources ...string) (result *sourcesv1.ContainerSource, err error) {
+	obj, err := c.Fake.
+		Invokes(testing.NewPatchSubresourceAction(containersourcesResource, c.ns, name, pt, data, subresources...), &sourcesv1.ContainerSource{})
+
+	if obj == nil {
+		return nil, err
+	}
+	return obj.(*sourcesv1.ContainerSource), err
+}
diff --git a/vendor/knative.dev/eventing/pkg/client/clientset/versioned/typed/sources/v1/fake/fake_sinkbinding.go b/vendor/knative.dev/eventing/pkg/client/clientset/versioned/typed/sources/v1/fake/fake_sinkbinding.go
new file mode 100644
index 000000000..7cd972381
--- /dev/null
+++ b/vendor/knative.dev/eventing/pkg/client/clientset/versioned/typed/sources/v1/fake/fake_sinkbinding.go
@@ -0,0 +1,142 @@
+/*
+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.
+*/
+
+// Code generated by client-gen. DO NOT EDIT.
+
+package fake
+
+import (
+	"context"
+
+	v1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	labels "k8s.io/apimachinery/pkg/labels"
+	schema "k8s.io/apimachinery/pkg/runtime/schema"
+	types "k8s.io/apimachinery/pkg/types"
+	watch "k8s.io/apimachinery/pkg/watch"
+	testing "k8s.io/client-go/testing"
+	sourcesv1 "knative.dev/eventing/pkg/apis/sources/v1"
+)
+
+// FakeSinkBindings implements SinkBindingInterface
+type FakeSinkBindings struct {
+	Fake *FakeSourcesV1
+	ns   string
+}
+
+var sinkbindingsResource = schema.GroupVersionResource{Group: "sources.knative.dev", Version: "v1", Resource: "sinkbindings"}
+
+var sinkbindingsKind = schema.GroupVersionKind{Group: "sources.knative.dev", Version: "v1", Kind: "SinkBinding"}
+
+// Get takes name of the sinkBinding, and returns the corresponding sinkBinding object, and an error if there is any.
+func (c *FakeSinkBindings) Get(ctx context.Context, name string, options v1.GetOptions) (result *sourcesv1.SinkBinding, err error) {
+	obj, err := c.Fake.
+		Invokes(testing.NewGetAction(sinkbindingsResource, c.ns, name), &sourcesv1.SinkBinding{})
+
+	if obj == nil {
+		return nil, err
+	}
+	return obj.(*sourcesv1.SinkBinding), err
+}
+
+// List takes label and field selectors, and returns the list of SinkBindings that match those selectors.
+func (c *FakeSinkBindings) List(ctx context.Context, opts v1.ListOptions) (result *sourcesv1.SinkBindingList, err error) {
+	obj, err := c.Fake.
+		Invokes(testing.NewListAction(sinkbindingsResource, sinkbindingsKind, c.ns, opts), &sourcesv1.SinkBindingList{})
+
+	if obj == nil {
+		return nil, err
+	}
+
+	label, _, _ := testing.ExtractFromListOptions(opts)
+	if label == nil {
+		label = labels.Everything()
+	}
+	list := &sourcesv1.SinkBindingList{ListMeta: obj.(*sourcesv1.SinkBindingList).ListMeta}
+	for _, item := range obj.(*sourcesv1.SinkBindingList).Items {
+		if label.Matches(labels.Set(item.Labels)) {
+			list.Items = append(list.Items, item)
+		}
+	}
+	return list, err
+}
+
+// Watch returns a watch.Interface that watches the requested sinkBindings.
+func (c *FakeSinkBindings) Watch(ctx context.Context, opts v1.ListOptions) (watch.Interface, error) {
+	return c.Fake.
+		InvokesWatch(testing.NewWatchAction(sinkbindingsResource, c.ns, opts))
+
+}
+
+// Create takes the representation of a sinkBinding and creates it.  Returns the server's representation of the sinkBinding, and an error, if there is any.
+func (c *FakeSinkBindings) Create(ctx context.Context, sinkBinding *sourcesv1.SinkBinding, opts v1.CreateOptions) (result *sourcesv1.SinkBinding, err error) {
+	obj, err := c.Fake.
+		Invokes(testing.NewCreateAction(sinkbindingsResource, c.ns, sinkBinding), &sourcesv1.SinkBinding{})
+
+	if obj == nil {
+		return nil, err
+	}
+	return obj.(*sourcesv1.SinkBinding), err
+}
+
+// Update takes the representation of a sinkBinding and updates it. Returns the server's representation of the sinkBinding, and an error, if there is any.
+func (c *FakeSinkBindings) Update(ctx context.Context, sinkBinding *sourcesv1.SinkBinding, opts v1.UpdateOptions) (result *sourcesv1.SinkBinding, err error) {
+	obj, err := c.Fake.
+		Invokes(testing.NewUpdateAction(sinkbindingsResource, c.ns, sinkBinding), &sourcesv1.SinkBinding{})
+
+	if obj == nil {
+		return nil, err
+	}
+	return obj.(*sourcesv1.SinkBinding), err
+}
+
+// UpdateStatus was generated because the type contains a Status member.
+// Add a +genclient:noStatus comment above the type to avoid generating UpdateStatus().
+func (c *FakeSinkBindings) UpdateStatus(ctx context.Context, sinkBinding *sourcesv1.SinkBinding, opts v1.UpdateOptions) (*sourcesv1.SinkBinding, error) {
+	obj, err := c.Fake.
+		Invokes(testing.NewUpdateSubresourceAction(sinkbindingsResource, "status", c.ns, sinkBinding), &sourcesv1.SinkBinding{})
+
+	if obj == nil {
+		return nil, err
+	}
+	return obj.(*sourcesv1.SinkBinding), err
+}
+
+// Delete takes name of the sinkBinding and deletes it. Returns an error if one occurs.
+func (c *FakeSinkBindings) Delete(ctx context.Context, name string, opts v1.DeleteOptions) error {
+	_, err := c.Fake.
+		Invokes(testing.NewDeleteAction(sinkbindingsResource, c.ns, name), &sourcesv1.SinkBinding{})
+
+	return err
+}
+
+// DeleteCollection deletes a collection of objects.
+func (c *FakeSinkBindings) DeleteCollection(ctx context.Context, opts v1.DeleteOptions, listOpts v1.ListOptions) error {
+	action := testing.NewDeleteCollectionAction(sinkbindingsResource, c.ns, listOpts)
+
+	_, err := c.Fake.Invokes(action, &sourcesv1.SinkBindingList{})
+	return err
+}
+
+// Patch applies the patch and returns the patched sinkBinding.
+func (c *FakeSinkBindings) Patch(ctx context.Context, name string, pt types.PatchType, data []byte, opts v1.PatchOptions, subresources ...string) (result *sourcesv1.SinkBinding, err error) {
+	obj, err := c.Fake.
+		Invokes(testing.NewPatchSubresourceAction(sinkbindingsResource, c.ns, name, pt, data, subresources...), &sourcesv1.SinkBinding{})
+
+	if obj == nil {
+		return nil, err
+	}
+	return obj.(*sourcesv1.SinkBinding), err
+}
diff --git a/vendor/knative.dev/eventing/pkg/client/clientset/versioned/typed/sources/v1/fake/fake_sources_client.go b/vendor/knative.dev/eventing/pkg/client/clientset/versioned/typed/sources/v1/fake/fake_sources_client.go
index 7bec5abb1..d635a1046 100644
--- a/vendor/knative.dev/eventing/pkg/client/clientset/versioned/typed/sources/v1/fake/fake_sources_client.go
+++ b/vendor/knative.dev/eventing/pkg/client/clientset/versioned/typed/sources/v1/fake/fake_sources_client.go
@@ -32,6 +32,14 @@ func (c *FakeSourcesV1) ApiServerSources(namespace string) v1.ApiServerSourceInt
 	return &FakeApiServerSources{c, namespace}
 }
 
+func (c *FakeSourcesV1) ContainerSources(namespace string) v1.ContainerSourceInterface {
+	return &FakeContainerSources{c, namespace}
+}
+
+func (c *FakeSourcesV1) SinkBindings(namespace string) v1.SinkBindingInterface {
+	return &FakeSinkBindings{c, namespace}
+}
+
 // RESTClient returns a RESTClient that is used to communicate
 // with API server by this client implementation.
 func (c *FakeSourcesV1) RESTClient() rest.Interface {
diff --git a/vendor/knative.dev/eventing/pkg/client/clientset/versioned/typed/sources/v1/generated_expansion.go b/vendor/knative.dev/eventing/pkg/client/clientset/versioned/typed/sources/v1/generated_expansion.go
index dfd304c86..fabb07b63 100644
--- a/vendor/knative.dev/eventing/pkg/client/clientset/versioned/typed/sources/v1/generated_expansion.go
+++ b/vendor/knative.dev/eventing/pkg/client/clientset/versioned/typed/sources/v1/generated_expansion.go
@@ -19,3 +19,7 @@ limitations under the License.
 package v1
 
 type ApiServerSourceExpansion interface{}
+
+type ContainerSourceExpansion interface{}
+
+type SinkBindingExpansion interface{}
diff --git a/vendor/knative.dev/eventing/pkg/client/clientset/versioned/typed/sources/v1/sinkbinding.go b/vendor/knative.dev/eventing/pkg/client/clientset/versioned/typed/sources/v1/sinkbinding.go
new file mode 100644
index 000000000..e19e25b38
--- /dev/null
+++ b/vendor/knative.dev/eventing/pkg/client/clientset/versioned/typed/sources/v1/sinkbinding.go
@@ -0,0 +1,195 @@
+/*
+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.
+*/
+
+// Code generated by client-gen. DO NOT EDIT.
+
+package v1
+
+import (
+	"context"
+	"time"
+
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	types "k8s.io/apimachinery/pkg/types"
+	watch "k8s.io/apimachinery/pkg/watch"
+	rest "k8s.io/client-go/rest"
+	v1 "knative.dev/eventing/pkg/apis/sources/v1"
+	scheme "knative.dev/eventing/pkg/client/clientset/versioned/scheme"
+)
+
+// SinkBindingsGetter has a method to return a SinkBindingInterface.
+// A group's client should implement this interface.
+type SinkBindingsGetter interface {
+	SinkBindings(namespace string) SinkBindingInterface
+}
+
+// SinkBindingInterface has methods to work with SinkBinding resources.
+type SinkBindingInterface interface {
+	Create(ctx context.Context, sinkBinding *v1.SinkBinding, opts metav1.CreateOptions) (*v1.SinkBinding, error)
+	Update(ctx context.Context, sinkBinding *v1.SinkBinding, opts metav1.UpdateOptions) (*v1.SinkBinding, error)
+	UpdateStatus(ctx context.Context, sinkBinding *v1.SinkBinding, opts metav1.UpdateOptions) (*v1.SinkBinding, error)
+	Delete(ctx context.Context, name string, opts metav1.DeleteOptions) error
+	DeleteCollection(ctx context.Context, opts metav1.DeleteOptions, listOpts metav1.ListOptions) error
+	Get(ctx context.Context, name string, opts metav1.GetOptions) (*v1.SinkBinding, error)
+	List(ctx context.Context, opts metav1.ListOptions) (*v1.SinkBindingList, error)
+	Watch(ctx context.Context, opts metav1.ListOptions) (watch.Interface, error)
+	Patch(ctx context.Context, name string, pt types.PatchType, data []byte, opts metav1.PatchOptions, subresources ...string) (result *v1.SinkBinding, err error)
+	SinkBindingExpansion
+}
+
+// sinkBindings implements SinkBindingInterface
+type sinkBindings struct {
+	client rest.Interface
+	ns     string
+}
+
+// newSinkBindings returns a SinkBindings
+func newSinkBindings(c *SourcesV1Client, namespace string) *sinkBindings {
+	return &sinkBindings{
+		client: c.RESTClient(),
+		ns:     namespace,
+	}
+}
+
+// Get takes name of the sinkBinding, and returns the corresponding sinkBinding object, and an error if there is any.
+func (c *sinkBindings) Get(ctx context.Context, name string, options metav1.GetOptions) (result *v1.SinkBinding, err error) {
+	result = &v1.SinkBinding{}
+	err = c.client.Get().
+		Namespace(c.ns).
+		Resource("sinkbindings").
+		Name(name).
+		VersionedParams(&options, scheme.ParameterCodec).
+		Do(ctx).
+		Into(result)
+	return
+}
+
+// List takes label and field selectors, and returns the list of SinkBindings that match those selectors.
+func (c *sinkBindings) List(ctx context.Context, opts metav1.ListOptions) (result *v1.SinkBindingList, err error) {
+	var timeout time.Duration
+	if opts.TimeoutSeconds != nil {
+		timeout = time.Duration(*opts.TimeoutSeconds) * time.Second
+	}
+	result = &v1.SinkBindingList{}
+	err = c.client.Get().
+		Namespace(c.ns).
+		Resource("sinkbindings").
+		VersionedParams(&opts, scheme.ParameterCodec).
+		Timeout(timeout).
+		Do(ctx).
+		Into(result)
+	return
+}
+
+// Watch returns a watch.Interface that watches the requested sinkBindings.
+func (c *sinkBindings) Watch(ctx context.Context, opts metav1.ListOptions) (watch.Interface, error) {
+	var timeout time.Duration
+	if opts.TimeoutSeconds != nil {
+		timeout = time.Duration(*opts.TimeoutSeconds) * time.Second
+	}
+	opts.Watch = true
+	return c.client.Get().
+		Namespace(c.ns).
+		Resource("sinkbindings").
+		VersionedParams(&opts, scheme.ParameterCodec).
+		Timeout(timeout).
+		Watch(ctx)
+}
+
+// Create takes the representation of a sinkBinding and creates it.  Returns the server's representation of the sinkBinding, and an error, if there is any.
+func (c *sinkBindings) Create(ctx context.Context, sinkBinding *v1.SinkBinding, opts metav1.CreateOptions) (result *v1.SinkBinding, err error) {
+	result = &v1.SinkBinding{}
+	err = c.client.Post().
+		Namespace(c.ns).
+		Resource("sinkbindings").
+		VersionedParams(&opts, scheme.ParameterCodec).
+		Body(sinkBinding).
+		Do(ctx).
+		Into(result)
+	return
+}
+
+// Update takes the representation of a sinkBinding and updates it. Returns the server's representation of the sinkBinding, and an error, if there is any.
+func (c *sinkBindings) Update(ctx context.Context, sinkBinding *v1.SinkBinding, opts metav1.UpdateOptions) (result *v1.SinkBinding, err error) {
+	result = &v1.SinkBinding{}
+	err = c.client.Put().
+		Namespace(c.ns).
+		Resource("sinkbindings").
+		Name(sinkBinding.Name).
+		VersionedParams(&opts, scheme.ParameterCodec).
+		Body(sinkBinding).
+		Do(ctx).
+		Into(result)
+	return
+}
+
+// UpdateStatus was generated because the type contains a Status member.
+// Add a +genclient:noStatus comment above the type to avoid generating UpdateStatus().
+func (c *sinkBindings) UpdateStatus(ctx context.Context, sinkBinding *v1.SinkBinding, opts metav1.UpdateOptions) (result *v1.SinkBinding, err error) {
+	result = &v1.SinkBinding{}
+	err = c.client.Put().
+		Namespace(c.ns).
+		Resource("sinkbindings").
+		Name(sinkBinding.Name).
+		SubResource("status").
+		VersionedParams(&opts, scheme.ParameterCodec).
+		Body(sinkBinding).
+		Do(ctx).
+		Into(result)
+	return
+}
+
+// Delete takes name of the sinkBinding and deletes it. Returns an error if one occurs.
+func (c *sinkBindings) Delete(ctx context.Context, name string, opts metav1.DeleteOptions) error {
+	return c.client.Delete().
+		Namespace(c.ns).
+		Resource("sinkbindings").
+		Name(name).
+		Body(&opts).
+		Do(ctx).
+		Error()
+}
+
+// DeleteCollection deletes a collection of objects.
+func (c *sinkBindings) DeleteCollection(ctx context.Context, opts metav1.DeleteOptions, listOpts metav1.ListOptions) error {
+	var timeout time.Duration
+	if listOpts.TimeoutSeconds != nil {
+		timeout = time.Duration(*listOpts.TimeoutSeconds) * time.Second
+	}
+	return c.client.Delete().
+		Namespace(c.ns).
+		Resource("sinkbindings").
+		VersionedParams(&listOpts, scheme.ParameterCodec).
+		Timeout(timeout).
+		Body(&opts).
+		Do(ctx).
+		Error()
+}
+
+// Patch applies the patch and returns the patched sinkBinding.
+func (c *sinkBindings) Patch(ctx context.Context, name string, pt types.PatchType, data []byte, opts metav1.PatchOptions, subresources ...string) (result *v1.SinkBinding, err error) {
+	result = &v1.SinkBinding{}
+	err = c.client.Patch(pt).
+		Namespace(c.ns).
+		Resource("sinkbindings").
+		Name(name).
+		SubResource(subresources...).
+		VersionedParams(&opts, scheme.ParameterCodec).
+		Body(data).
+		Do(ctx).
+		Into(result)
+	return
+}
diff --git a/vendor/knative.dev/eventing/pkg/client/clientset/versioned/typed/sources/v1/sources_client.go b/vendor/knative.dev/eventing/pkg/client/clientset/versioned/typed/sources/v1/sources_client.go
index 635c97646..8b02e5fdb 100644
--- a/vendor/knative.dev/eventing/pkg/client/clientset/versioned/typed/sources/v1/sources_client.go
+++ b/vendor/knative.dev/eventing/pkg/client/clientset/versioned/typed/sources/v1/sources_client.go
@@ -27,6 +27,8 @@ import (
 type SourcesV1Interface interface {
 	RESTClient() rest.Interface
 	ApiServerSourcesGetter
+	ContainerSourcesGetter
+	SinkBindingsGetter
 }
 
 // SourcesV1Client is used to interact with features provided by the sources.knative.dev group.
@@ -38,6 +40,14 @@ func (c *SourcesV1Client) ApiServerSources(namespace string) ApiServerSourceInte
 	return newApiServerSources(c, namespace)
 }
 
+func (c *SourcesV1Client) ContainerSources(namespace string) ContainerSourceInterface {
+	return newContainerSources(c, namespace)
+}
+
+func (c *SourcesV1Client) SinkBindings(namespace string) SinkBindingInterface {
+	return newSinkBindings(c, namespace)
+}
+
 // NewForConfig creates a new SourcesV1Client for the given config.
 func NewForConfig(c *rest.Config) (*SourcesV1Client, error) {
 	config := *c
diff --git a/vendor/knative.dev/eventing/pkg/client/informers/externalversions/generic.go b/vendor/knative.dev/eventing/pkg/client/informers/externalversions/generic.go
index a851deb4f..c0d877832 100644
--- a/vendor/knative.dev/eventing/pkg/client/informers/externalversions/generic.go
+++ b/vendor/knative.dev/eventing/pkg/client/informers/externalversions/generic.go
@@ -111,6 +111,10 @@ func (f *sharedInformerFactory) ForResource(resource schema.GroupVersionResource
 		// Group=sources.knative.dev, Version=v1
 	case sourcesv1.SchemeGroupVersion.WithResource("apiserversources"):
 		return &genericInformer{resource: resource.GroupResource(), informer: f.Sources().V1().ApiServerSources().Informer()}, nil
+	case sourcesv1.SchemeGroupVersion.WithResource("containersources"):
+		return &genericInformer{resource: resource.GroupResource(), informer: f.Sources().V1().ContainerSources().Informer()}, nil
+	case sourcesv1.SchemeGroupVersion.WithResource("sinkbindings"):
+		return &genericInformer{resource: resource.GroupResource(), informer: f.Sources().V1().SinkBindings().Informer()}, nil
 
 		// Group=sources.knative.dev, Version=v1alpha1
 	case sourcesv1alpha1.SchemeGroupVersion.WithResource("apiserversources"):
diff --git a/vendor/knative.dev/eventing/pkg/client/informers/externalversions/sources/v1/containersource.go b/vendor/knative.dev/eventing/pkg/client/informers/externalversions/sources/v1/containersource.go
new file mode 100644
index 000000000..8033b9d29
--- /dev/null
+++ b/vendor/knative.dev/eventing/pkg/client/informers/externalversions/sources/v1/containersource.go
@@ -0,0 +1,90 @@
+/*
+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.
+*/
+
+// Code generated by informer-gen. DO NOT EDIT.
+
+package v1
+
+import (
+	"context"
+	time "time"
+
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	runtime "k8s.io/apimachinery/pkg/runtime"
+	watch "k8s.io/apimachinery/pkg/watch"
+	cache "k8s.io/client-go/tools/cache"
+	sourcesv1 "knative.dev/eventing/pkg/apis/sources/v1"
+	versioned "knative.dev/eventing/pkg/client/clientset/versioned"
+	internalinterfaces "knative.dev/eventing/pkg/client/informers/externalversions/internalinterfaces"
+	v1 "knative.dev/eventing/pkg/client/listers/sources/v1"
+)
+
+// ContainerSourceInformer provides access to a shared informer and lister for
+// ContainerSources.
+type ContainerSourceInformer interface {
+	Informer() cache.SharedIndexInformer
+	Lister() v1.ContainerSourceLister
+}
+
+type containerSourceInformer struct {
+	factory          internalinterfaces.SharedInformerFactory
+	tweakListOptions internalinterfaces.TweakListOptionsFunc
+	namespace        string
+}
+
+// NewContainerSourceInformer constructs a new informer for ContainerSource type.
+// Always prefer using an informer factory to get a shared informer instead of getting an independent
+// one. This reduces memory footprint and number of connections to the server.
+func NewContainerSourceInformer(client versioned.Interface, namespace string, resyncPeriod time.Duration, indexers cache.Indexers) cache.SharedIndexInformer {
+	return NewFilteredContainerSourceInformer(client, namespace, resyncPeriod, indexers, nil)
+}
+
+// NewFilteredContainerSourceInformer constructs a new informer for ContainerSource type.
+// Always prefer using an informer factory to get a shared informer instead of getting an independent
+// one. This reduces memory footprint and number of connections to the server.
+func NewFilteredContainerSourceInformer(client versioned.Interface, namespace string, resyncPeriod time.Duration, indexers cache.Indexers, tweakListOptions internalinterfaces.TweakListOptionsFunc) cache.SharedIndexInformer {
+	return cache.NewSharedIndexInformer(
+		&cache.ListWatch{
+			ListFunc: func(options metav1.ListOptions) (runtime.Object, error) {
+				if tweakListOptions != nil {
+					tweakListOptions(&options)
+				}
+				return client.SourcesV1().ContainerSources(namespace).List(context.TODO(), options)
+			},
+			WatchFunc: func(options metav1.ListOptions) (watch.Interface, error) {
+				if tweakListOptions != nil {
+					tweakListOptions(&options)
+				}
+				return client.SourcesV1().ContainerSources(namespace).Watch(context.TODO(), options)
+			},
+		},
+		&sourcesv1.ContainerSource{},
+		resyncPeriod,
+		indexers,
+	)
+}
+
+func (f *containerSourceInformer) defaultInformer(client versioned.Interface, resyncPeriod time.Duration) cache.SharedIndexInformer {
+	return NewFilteredContainerSourceInformer(client, f.namespace, resyncPeriod, cache.Indexers{cache.NamespaceIndex: cache.MetaNamespaceIndexFunc}, f.tweakListOptions)
+}
+
+func (f *containerSourceInformer) Informer() cache.SharedIndexInformer {
+	return f.factory.InformerFor(&sourcesv1.ContainerSource{}, f.defaultInformer)
+}
+
+func (f *containerSourceInformer) Lister() v1.ContainerSourceLister {
+	return v1.NewContainerSourceLister(f.Informer().GetIndexer())
+}
diff --git a/vendor/knative.dev/eventing/pkg/client/informers/externalversions/sources/v1/interface.go b/vendor/knative.dev/eventing/pkg/client/informers/externalversions/sources/v1/interface.go
index ea7baa106..e4e42e620 100644
--- a/vendor/knative.dev/eventing/pkg/client/informers/externalversions/sources/v1/interface.go
+++ b/vendor/knative.dev/eventing/pkg/client/informers/externalversions/sources/v1/interface.go
@@ -26,6 +26,10 @@ import (
 type Interface interface {
 	// ApiServerSources returns a ApiServerSourceInformer.
 	ApiServerSources() ApiServerSourceInformer
+	// ContainerSources returns a ContainerSourceInformer.
+	ContainerSources() ContainerSourceInformer
+	// SinkBindings returns a SinkBindingInformer.
+	SinkBindings() SinkBindingInformer
 }
 
 type version struct {
@@ -43,3 +47,13 @@ func New(f internalinterfaces.SharedInformerFactory, namespace string, tweakList
 func (v *version) ApiServerSources() ApiServerSourceInformer {
 	return &apiServerSourceInformer{factory: v.factory, namespace: v.namespace, tweakListOptions: v.tweakListOptions}
 }
+
+// ContainerSources returns a ContainerSourceInformer.
+func (v *version) ContainerSources() ContainerSourceInformer {
+	return &containerSourceInformer{factory: v.factory, namespace: v.namespace, tweakListOptions: v.tweakListOptions}
+}
+
+// SinkBindings returns a SinkBindingInformer.
+func (v *version) SinkBindings() SinkBindingInformer {
+	return &sinkBindingInformer{factory: v.factory, namespace: v.namespace, tweakListOptions: v.tweakListOptions}
+}
diff --git a/vendor/knative.dev/eventing/pkg/client/informers/externalversions/sources/v1/sinkbinding.go b/vendor/knative.dev/eventing/pkg/client/informers/externalversions/sources/v1/sinkbinding.go
new file mode 100644
index 000000000..8dce7456f
--- /dev/null
+++ b/vendor/knative.dev/eventing/pkg/client/informers/externalversions/sources/v1/sinkbinding.go
@@ -0,0 +1,90 @@
+/*
+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.
+*/
+
+// Code generated by informer-gen. DO NOT EDIT.
+
+package v1
+
+import (
+	"context"
+	time "time"
+
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	runtime "k8s.io/apimachinery/pkg/runtime"
+	watch "k8s.io/apimachinery/pkg/watch"
+	cache "k8s.io/client-go/tools/cache"
+	sourcesv1 "knative.dev/eventing/pkg/apis/sources/v1"
+	versioned "knative.dev/eventing/pkg/client/clientset/versioned"
+	internalinterfaces "knative.dev/eventing/pkg/client/informers/externalversions/internalinterfaces"
+	v1 "knative.dev/eventing/pkg/client/listers/sources/v1"
+)
+
+// SinkBindingInformer provides access to a shared informer and lister for
+// SinkBindings.
+type SinkBindingInformer interface {
+	Informer() cache.SharedIndexInformer
+	Lister() v1.SinkBindingLister
+}
+
+type sinkBindingInformer struct {
+	factory          internalinterfaces.SharedInformerFactory
+	tweakListOptions internalinterfaces.TweakListOptionsFunc
+	namespace        string
+}
+
+// NewSinkBindingInformer constructs a new informer for SinkBinding type.
+// Always prefer using an informer factory to get a shared informer instead of getting an independent
+// one. This reduces memory footprint and number of connections to the server.
+func NewSinkBindingInformer(client versioned.Interface, namespace string, resyncPeriod time.Duration, indexers cache.Indexers) cache.SharedIndexInformer {
+	return NewFilteredSinkBindingInformer(client, namespace, resyncPeriod, indexers, nil)
+}
+
+// NewFilteredSinkBindingInformer constructs a new informer for SinkBinding type.
+// Always prefer using an informer factory to get a shared informer instead of getting an independent
+// one. This reduces memory footprint and number of connections to the server.
+func NewFilteredSinkBindingInformer(client versioned.Interface, namespace string, resyncPeriod time.Duration, indexers cache.Indexers, tweakListOptions internalinterfaces.TweakListOptionsFunc) cache.SharedIndexInformer {
+	return cache.NewSharedIndexInformer(
+		&cache.ListWatch{
+			ListFunc: func(options metav1.ListOptions) (runtime.Object, error) {
+				if tweakListOptions != nil {
+					tweakListOptions(&options)
+				}
+				return client.SourcesV1().SinkBindings(namespace).List(context.TODO(), options)
+			},
+			WatchFunc: func(options metav1.ListOptions) (watch.Interface, error) {
+				if tweakListOptions != nil {
+					tweakListOptions(&options)
+				}
+				return client.SourcesV1().SinkBindings(namespace).Watch(context.TODO(), options)
+			},
+		},
+		&sourcesv1.SinkBinding{},
+		resyncPeriod,
+		indexers,
+	)
+}
+
+func (f *sinkBindingInformer) defaultInformer(client versioned.Interface, resyncPeriod time.Duration) cache.SharedIndexInformer {
+	return NewFilteredSinkBindingInformer(client, f.namespace, resyncPeriod, cache.Indexers{cache.NamespaceIndex: cache.MetaNamespaceIndexFunc}, f.tweakListOptions)
+}
+
+func (f *sinkBindingInformer) Informer() cache.SharedIndexInformer {
+	return f.factory.InformerFor(&sourcesv1.SinkBinding{}, f.defaultInformer)
+}
+
+func (f *sinkBindingInformer) Lister() v1.SinkBindingLister {
+	return v1.NewSinkBindingLister(f.Informer().GetIndexer())
+}
diff --git a/vendor/knative.dev/eventing/pkg/client/listers/sources/v1/containersource.go b/vendor/knative.dev/eventing/pkg/client/listers/sources/v1/containersource.go
new file mode 100644
index 000000000..2219f3fee
--- /dev/null
+++ b/vendor/knative.dev/eventing/pkg/client/listers/sources/v1/containersource.go
@@ -0,0 +1,94 @@
+/*
+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.
+*/
+
+// Code generated by lister-gen. DO NOT EDIT.
+
+package v1
+
+import (
+	"k8s.io/apimachinery/pkg/api/errors"
+	"k8s.io/apimachinery/pkg/labels"
+	"k8s.io/client-go/tools/cache"
+	v1 "knative.dev/eventing/pkg/apis/sources/v1"
+)
+
+// ContainerSourceLister helps list ContainerSources.
+type ContainerSourceLister interface {
+	// List lists all ContainerSources in the indexer.
+	List(selector labels.Selector) (ret []*v1.ContainerSource, err error)
+	// ContainerSources returns an object that can list and get ContainerSources.
+	ContainerSources(namespace string) ContainerSourceNamespaceLister
+	ContainerSourceListerExpansion
+}
+
+// containerSourceLister implements the ContainerSourceLister interface.
+type containerSourceLister struct {
+	indexer cache.Indexer
+}
+
+// NewContainerSourceLister returns a new ContainerSourceLister.
+func NewContainerSourceLister(indexer cache.Indexer) ContainerSourceLister {
+	return &containerSourceLister{indexer: indexer}
+}
+
+// List lists all ContainerSources in the indexer.
+func (s *containerSourceLister) List(selector labels.Selector) (ret []*v1.ContainerSource, err error) {
+	err = cache.ListAll(s.indexer, selector, func(m interface{}) {
+		ret = append(ret, m.(*v1.ContainerSource))
+	})
+	return ret, err
+}
+
+// ContainerSources returns an object that can list and get ContainerSources.
+func (s *containerSourceLister) ContainerSources(namespace string) ContainerSourceNamespaceLister {
+	return containerSourceNamespaceLister{indexer: s.indexer, namespace: namespace}
+}
+
+// ContainerSourceNamespaceLister helps list and get ContainerSources.
+type ContainerSourceNamespaceLister interface {
+	// List lists all ContainerSources in the indexer for a given namespace.
+	List(selector labels.Selector) (ret []*v1.ContainerSource, err error)
+	// Get retrieves the ContainerSource from the indexer for a given namespace and name.
+	Get(name string) (*v1.ContainerSource, error)
+	ContainerSourceNamespaceListerExpansion
+}
+
+// containerSourceNamespaceLister implements the ContainerSourceNamespaceLister
+// interface.
+type containerSourceNamespaceLister struct {
+	indexer   cache.Indexer
+	namespace string
+}
+
+// List lists all ContainerSources in the indexer for a given namespace.
+func (s containerSourceNamespaceLister) List(selector labels.Selector) (ret []*v1.ContainerSource, err error) {
+	err = cache.ListAllByNamespace(s.indexer, s.namespace, selector, func(m interface{}) {
+		ret = append(ret, m.(*v1.ContainerSource))
+	})
+	return ret, err
+}
+
+// Get retrieves the ContainerSource from the indexer for a given namespace and name.
+func (s containerSourceNamespaceLister) Get(name string) (*v1.ContainerSource, error) {
+	obj, exists, err := s.indexer.GetByKey(s.namespace + "/" + name)
+	if err != nil {
+		return nil, err
+	}
+	if !exists {
+		return nil, errors.NewNotFound(v1.Resource("containersource"), name)
+	}
+	return obj.(*v1.ContainerSource), nil
+}
diff --git a/vendor/knative.dev/eventing/pkg/client/listers/sources/v1/expansion_generated.go b/vendor/knative.dev/eventing/pkg/client/listers/sources/v1/expansion_generated.go
index 646cce7e4..31da8a718 100644
--- a/vendor/knative.dev/eventing/pkg/client/listers/sources/v1/expansion_generated.go
+++ b/vendor/knative.dev/eventing/pkg/client/listers/sources/v1/expansion_generated.go
@@ -25,3 +25,19 @@ type ApiServerSourceListerExpansion interface{}
 // ApiServerSourceNamespaceListerExpansion allows custom methods to be added to
 // ApiServerSourceNamespaceLister.
 type ApiServerSourceNamespaceListerExpansion interface{}
+
+// ContainerSourceListerExpansion allows custom methods to be added to
+// ContainerSourceLister.
+type ContainerSourceListerExpansion interface{}
+
+// ContainerSourceNamespaceListerExpansion allows custom methods to be added to
+// ContainerSourceNamespaceLister.
+type ContainerSourceNamespaceListerExpansion interface{}
+
+// SinkBindingListerExpansion allows custom methods to be added to
+// SinkBindingLister.
+type SinkBindingListerExpansion interface{}
+
+// SinkBindingNamespaceListerExpansion allows custom methods to be added to
+// SinkBindingNamespaceLister.
+type SinkBindingNamespaceListerExpansion interface{}
diff --git a/vendor/knative.dev/eventing/pkg/client/listers/sources/v1/sinkbinding.go b/vendor/knative.dev/eventing/pkg/client/listers/sources/v1/sinkbinding.go
new file mode 100644
index 000000000..b4eda9693
--- /dev/null
+++ b/vendor/knative.dev/eventing/pkg/client/listers/sources/v1/sinkbinding.go
@@ -0,0 +1,94 @@
+/*
+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.
+*/
+
+// Code generated by lister-gen. DO NOT EDIT.
+
+package v1
+
+import (
+	"k8s.io/apimachinery/pkg/api/errors"
+	"k8s.io/apimachinery/pkg/labels"
+	"k8s.io/client-go/tools/cache"
+	v1 "knative.dev/eventing/pkg/apis/sources/v1"
+)
+
+// SinkBindingLister helps list SinkBindings.
+type SinkBindingLister interface {
+	// List lists all SinkBindings in the indexer.
+	List(selector labels.Selector) (ret []*v1.SinkBinding, err error)
+	// SinkBindings returns an object that can list and get SinkBindings.
+	SinkBindings(namespace string) SinkBindingNamespaceLister
+	SinkBindingListerExpansion
+}
+
+// sinkBindingLister implements the SinkBindingLister interface.
+type sinkBindingLister struct {
+	indexer cache.Indexer
+}
+
+// NewSinkBindingLister returns a new SinkBindingLister.
+func NewSinkBindingLister(indexer cache.Indexer) SinkBindingLister {
+	return &sinkBindingLister{indexer: indexer}
+}
+
+// List lists all SinkBindings in the indexer.
+func (s *sinkBindingLister) List(selector labels.Selector) (ret []*v1.SinkBinding, err error) {
+	err = cache.ListAll(s.indexer, selector, func(m interface{}) {
+		ret = append(ret, m.(*v1.SinkBinding))
+	})
+	return ret, err
+}
+
+// SinkBindings returns an object that can list and get SinkBindings.
+func (s *sinkBindingLister) SinkBindings(namespace string) SinkBindingNamespaceLister {
+	return sinkBindingNamespaceLister{indexer: s.indexer, namespace: namespace}
+}
+
+// SinkBindingNamespaceLister helps list and get SinkBindings.
+type SinkBindingNamespaceLister interface {
+	// List lists all SinkBindings in the indexer for a given namespace.
+	List(selector labels.Selector) (ret []*v1.SinkBinding, err error)
+	// Get retrieves the SinkBinding from the indexer for a given namespace and name.
+	Get(name string) (*v1.SinkBinding, error)
+	SinkBindingNamespaceListerExpansion
+}
+
+// sinkBindingNamespaceLister implements the SinkBindingNamespaceLister
+// interface.
+type sinkBindingNamespaceLister struct {
+	indexer   cache.Indexer
+	namespace string
+}
+
+// List lists all SinkBindings in the indexer for a given namespace.
+func (s sinkBindingNamespaceLister) List(selector labels.Selector) (ret []*v1.SinkBinding, err error) {
+	err = cache.ListAllByNamespace(s.indexer, s.namespace, selector, func(m interface{}) {
+		ret = append(ret, m.(*v1.SinkBinding))
+	})
+	return ret, err
+}
+
+// Get retrieves the SinkBinding from the indexer for a given namespace and name.
+func (s sinkBindingNamespaceLister) Get(name string) (*v1.SinkBinding, error) {
+	obj, exists, err := s.indexer.GetByKey(s.namespace + "/" + name)
+	if err != nil {
+		return nil, err
+	}
+	if !exists {
+		return nil, errors.NewNotFound(v1.Resource("sinkbinding"), name)
+	}
+	return obj.(*v1.SinkBinding), nil
+}
diff --git a/vendor/knative.dev/eventing/pkg/kncloudevents/message_receiver.go b/vendor/knative.dev/eventing/pkg/kncloudevents/message_receiver.go
index fb22daeeb..2d200e0f9 100644
--- a/vendor/knative.dev/eventing/pkg/kncloudevents/message_receiver.go
+++ b/vendor/knative.dev/eventing/pkg/kncloudevents/message_receiver.go
@@ -19,10 +19,11 @@ import (
 	"context"
 	"fmt"
 	"net"
-	nethttp "net/http"
+	"net/http"
 	"time"
 
 	"go.opencensus.io/plugin/ochttp"
+	"knative.dev/pkg/network/handlers"
 	"knative.dev/pkg/tracing/propagation/tracecontextb3"
 )
 
@@ -30,32 +31,32 @@ const (
 	DefaultShutdownTimeout = time.Minute * 1
 )
 
-type HttpMessageReceiver struct {
+type HTTPMessageReceiver struct {
 	port int
 
-	handler  nethttp.Handler
-	server   *nethttp.Server
+	server   *http.Server
 	listener net.Listener
 }
 
-func NewHttpMessageReceiver(port int) *HttpMessageReceiver {
-	return &HttpMessageReceiver{
+func NewHTTPMessageReceiver(port int) *HTTPMessageReceiver {
+	return &HTTPMessageReceiver{
 		port: port,
 	}
 }
 
 // Blocking
-func (recv *HttpMessageReceiver) StartListen(ctx context.Context, handler nethttp.Handler) error {
+func (recv *HTTPMessageReceiver) StartListen(ctx context.Context, handler http.Handler) error {
 	var err error
 	if recv.listener, err = net.Listen("tcp", fmt.Sprintf(":%d", recv.port)); err != nil {
 		return err
 	}
 
-	recv.handler = CreateHandler(handler)
-
-	recv.server = &nethttp.Server{
+	drainer := &handlers.Drainer{
+		Inner: CreateHandler(handler),
+	}
+	recv.server = &http.Server{
 		Addr:    recv.listener.Addr().String(),
-		Handler: recv.handler,
+		Handler: drainer,
 	}
 
 	errChan := make(chan error, 1)
@@ -66,6 +67,9 @@ func (recv *HttpMessageReceiver) StartListen(ctx context.Context, handler nethtt
 	// wait for the server to return or ctx.Done().
 	select {
 	case <-ctx.Done():
+		// As we start to shutdown, disable keep-alives to avoid clients hanging onto connections.
+		recv.server.SetKeepAlivesEnabled(false)
+		drainer.Drain()
 		ctx, cancel := context.WithTimeout(context.Background(), getShutdownTimeout(ctx))
 		defer cancel()
 		err := recv.server.Shutdown(ctx)
@@ -90,7 +94,7 @@ func WithShutdownTimeout(ctx context.Context, timeout time.Duration) context.Con
 	return context.WithValue(ctx, shutdownTimeoutKey{}, timeout)
 }
 
-func CreateHandler(handler nethttp.Handler) nethttp.Handler {
+func CreateHandler(handler http.Handler) http.Handler {
 	return &ochttp.Handler{
 		Propagation: tracecontextb3.TraceContextEgress,
 		Handler:     handler,
diff --git a/vendor/knative.dev/eventing/pkg/kncloudevents/message_sender.go b/vendor/knative.dev/eventing/pkg/kncloudevents/message_sender.go
index 1a425cd60..49e54f902 100644
--- a/vendor/knative.dev/eventing/pkg/kncloudevents/message_sender.go
+++ b/vendor/knative.dev/eventing/pkg/kncloudevents/message_sender.go
@@ -63,12 +63,12 @@ func (ca *ConnectionArgs) ConfigureTransport(transport *nethttp.Transport) {
 	transport.MaxIdleConnsPerHost = ca.MaxIdleConnsPerHost
 }
 
-type HttpMessageSender struct {
+type HTTPMessageSender struct {
 	Client *nethttp.Client
 	Target string
 }
 
-func NewHttpMessageSender(connectionArgs *ConnectionArgs, target string) (*HttpMessageSender, error) {
+func NewHTTPMessageSender(connectionArgs *ConnectionArgs, target string) (*HTTPMessageSender, error) {
 	// Add connection options to the default transport.
 	var base = nethttp.DefaultTransport.(*nethttp.Transport).Clone()
 	connectionArgs.ConfigureTransport(base)
@@ -80,18 +80,18 @@ func NewHttpMessageSender(connectionArgs *ConnectionArgs, target string) (*HttpM
 		},
 	}
 
-	return &HttpMessageSender{Client: client, Target: target}, nil
+	return &HTTPMessageSender{Client: client, Target: target}, nil
 }
 
-func (s *HttpMessageSender) NewCloudEventRequest(ctx context.Context) (*nethttp.Request, error) {
+func (s *HTTPMessageSender) NewCloudEventRequest(ctx context.Context) (*nethttp.Request, error) {
 	return nethttp.NewRequestWithContext(ctx, "POST", s.Target, nil)
 }
 
-func (s *HttpMessageSender) NewCloudEventRequestWithTarget(ctx context.Context, target string) (*nethttp.Request, error) {
+func (s *HTTPMessageSender) NewCloudEventRequestWithTarget(ctx context.Context, target string) (*nethttp.Request, error) {
 	return nethttp.NewRequestWithContext(ctx, "POST", target, nil)
 }
 
-func (s *HttpMessageSender) Send(req *nethttp.Request) (*nethttp.Response, error) {
+func (s *HTTPMessageSender) Send(req *nethttp.Request) (*nethttp.Response, error) {
 	return s.Client.Do(req)
 }
 
@@ -111,7 +111,6 @@ type CheckRetry func(ctx context.Context, resp *nethttp.Response, err error) (bo
 type Backoff func(attemptNum int, resp *nethttp.Response) time.Duration
 
 type RetryConfig struct {
-
 	// Maximum number of retries
 	RetryMax int
 
@@ -120,7 +119,7 @@ type RetryConfig struct {
 	Backoff Backoff
 }
 
-func (s *HttpMessageSender) SendWithRetries(req *nethttp.Request, config *RetryConfig) (*nethttp.Response, error) {
+func (s *HTTPMessageSender) SendWithRetries(req *nethttp.Request, config *RetryConfig) (*nethttp.Response, error) {
 	if config == nil {
 		return s.Send(req)
 	}
@@ -171,7 +170,7 @@ func RetryConfigFromDeliverySpec(spec duckv1.DeliverySpec) (RetryConfig, error)
 			}
 		case duckv1.BackoffPolicyLinear:
 			retryConfig.Backoff = func(attemptNum int, resp *nethttp.Response) time.Duration {
-				return delayDuration
+				return delayDuration * time.Duration(attemptNum)
 			}
 		}
 	}
diff --git a/vendor/knative.dev/eventing/pkg/kncloudevents/utils.go b/vendor/knative.dev/eventing/pkg/kncloudevents/utils.go
index 4f01d91d7..3fb4fae10 100644
--- a/vendor/knative.dev/eventing/pkg/kncloudevents/utils.go
+++ b/vendor/knative.dev/eventing/pkg/kncloudevents/utils.go
@@ -24,7 +24,8 @@ import (
 	"github.com/cloudevents/sdk-go/v2/protocol/http"
 )
 
-func WriteHttpRequestWithAdditionalHeaders(ctx context.Context, message binding.Message, req *nethttp.Request, additionalHeaders nethttp.Header, transformers ...binding.Transformer) error {
+func WriteHTTPRequestWithAdditionalHeaders(ctx context.Context, message binding.Message, req *nethttp.Request,
+	additionalHeaders nethttp.Header, transformers ...binding.Transformer) error {
 	err := http.WriteRequest(ctx, message, req, transformers...)
 	if err != nil {
 		return err
diff --git a/vendor/knative.dev/pkg/apis/duck/v1/binding_types.go b/vendor/knative.dev/pkg/apis/duck/v1/binding_types.go
new file mode 100644
index 000000000..71b224cf5
--- /dev/null
+++ b/vendor/knative.dev/pkg/apis/duck/v1/binding_types.go
@@ -0,0 +1,92 @@
+/*
+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 v1
+
+import (
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime"
+
+	"knative.dev/pkg/apis"
+	"knative.dev/pkg/apis/duck"
+	"knative.dev/pkg/tracker"
+)
+
+// +genduck
+// +k8s:deepcopy-gen:interfaces=k8s.io/apimachinery/pkg/runtime.Object
+
+// Binding is a duck type that specifies the partial schema to which all
+// Binding implementations should adhere.
+type Binding struct {
+	metav1.TypeMeta   `json:",inline"`
+	metav1.ObjectMeta `json:"metadata,omitempty"`
+
+	Spec BindingSpec `json:"spec"`
+}
+
+// Verify that Binding implements the appropriate interfaces.
+var (
+	_ duck.Implementable = (*Binding)(nil)
+	_ duck.Populatable   = (*Binding)(nil)
+	_ apis.Listable      = (*Binding)(nil)
+)
+
+// BindingSpec specifies the spec portion of the Binding partial-schema.
+type BindingSpec struct {
+	// Subject references the resource(s) whose "runtime contract" should be
+	// augmented by Binding implementations.
+	Subject tracker.Reference `json:"subject"`
+}
+
+// GetFullType implements duck.Implementable
+func (*Binding) GetFullType() duck.Populatable {
+	return &Binding{}
+}
+
+// Populate implements duck.Populatable
+func (t *Binding) Populate() {
+	t.Spec = BindingSpec{
+		Subject: tracker.Reference{
+			APIVersion: "apps/v1",
+			Kind:       "Deployment",
+			Namespace:  "default",
+			// Name and Selector are mutually exclusive,
+			// but we fill them both in for this test.
+			Name: "bazinga",
+			Selector: &metav1.LabelSelector{
+				MatchLabels: map[string]string{
+					"foo": "bar",
+					"baz": "blah",
+				},
+			},
+		},
+	}
+}
+
+// GetListType implements apis.Listable
+func (*Binding) GetListType() runtime.Object {
+	return &BindingList{}
+}
+
+// +k8s:deepcopy-gen:interfaces=k8s.io/apimachinery/pkg/runtime.Object
+
+// BindingList is a list of Binding resources
+type BindingList struct {
+	metav1.TypeMeta `json:",inline"`
+	metav1.ListMeta `json:"metadata"`
+
+	Items []Binding `json:"items"`
+}
diff --git a/vendor/knative.dev/pkg/apis/duck/v1/fuzzer.go b/vendor/knative.dev/pkg/apis/duck/v1/fuzzer.go
new file mode 100644
index 000000000..4ef66e589
--- /dev/null
+++ b/vendor/knative.dev/pkg/apis/duck/v1/fuzzer.go
@@ -0,0 +1,44 @@
+/*
+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 v1
+
+import (
+	fuzz "github.com/google/gofuzz"
+	"k8s.io/apimachinery/pkg/api/apitesting/fuzzer"
+	"k8s.io/apimachinery/pkg/runtime/serializer"
+	"knative.dev/pkg/apis"
+	pkgfuzzer "knative.dev/pkg/apis/testing/fuzzer"
+)
+
+var testConditions = apis.Conditions{{Type: apis.ConditionReady}, {Type: apis.ConditionSucceeded}}
+
+// FuzzerFuncs includes fuzzing funcs for knative.dev/duck v1 types
+// In particular it makes sure that Delivery has only valid BackoffPolicyType in it.
+//
+// For other examples see
+// https://github.com/kubernetes/apimachinery/blob/master/pkg/apis/meta/fuzzer/fuzzer.go
+var FuzzerFuncs = fuzzer.MergeFuzzerFuncs(
+	func(codecs serializer.CodecFactory) []interface{} {
+		return []interface{}{
+			func(status *Status, c fuzz.Continue) {
+				c.FuzzNoCustom(status) // fuzz the Status
+				status.SetConditions(testConditions)
+				pkgfuzzer.FuzzConditions(status, c)
+			},
+		}
+	},
+)
diff --git a/vendor/knative.dev/pkg/apis/duck/v1/register.go b/vendor/knative.dev/pkg/apis/duck/v1/register.go
index 69bb11578..67418b3e9 100644
--- a/vendor/knative.dev/pkg/apis/duck/v1/register.go
+++ b/vendor/knative.dev/pkg/apis/duck/v1/register.go
@@ -54,6 +54,8 @@ func addKnownTypes(scheme *runtime.Scheme) error {
 		(&Source{}).GetListType(),
 		&WithPod{},
 		(&WithPod{}).GetListType(),
+		&Binding{},
+		(&Binding{}).GetListType(),
 	)
 	metav1.AddToGroupVersion(scheme, SchemeGroupVersion)
 	return nil
diff --git a/vendor/knative.dev/pkg/apis/duck/v1/zz_generated.deepcopy.go b/vendor/knative.dev/pkg/apis/duck/v1/zz_generated.deepcopy.go
index 190d773e7..f4da66047 100644
--- a/vendor/knative.dev/pkg/apis/duck/v1/zz_generated.deepcopy.go
+++ b/vendor/knative.dev/pkg/apis/duck/v1/zz_generated.deepcopy.go
@@ -127,6 +127,83 @@ func (in *AddressableTypeList) DeepCopyObject() runtime.Object {
 	return nil
 }
 
+// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
+func (in *Binding) DeepCopyInto(out *Binding) {
+	*out = *in
+	out.TypeMeta = in.TypeMeta
+	in.ObjectMeta.DeepCopyInto(&out.ObjectMeta)
+	in.Spec.DeepCopyInto(&out.Spec)
+	return
+}
+
+// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new Binding.
+func (in *Binding) DeepCopy() *Binding {
+	if in == nil {
+		return nil
+	}
+	out := new(Binding)
+	in.DeepCopyInto(out)
+	return out
+}
+
+// DeepCopyObject is an autogenerated deepcopy function, copying the receiver, creating a new runtime.Object.
+func (in *Binding) DeepCopyObject() runtime.Object {
+	if c := in.DeepCopy(); c != nil {
+		return c
+	}
+	return nil
+}
+
+// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
+func (in *BindingList) DeepCopyInto(out *BindingList) {
+	*out = *in
+	out.TypeMeta = in.TypeMeta
+	in.ListMeta.DeepCopyInto(&out.ListMeta)
+	if in.Items != nil {
+		in, out := &in.Items, &out.Items
+		*out = make([]Binding, len(*in))
+		for i := range *in {
+			(*in)[i].DeepCopyInto(&(*out)[i])
+		}
+	}
+	return
+}
+
+// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new BindingList.
+func (in *BindingList) DeepCopy() *BindingList {
+	if in == nil {
+		return nil
+	}
+	out := new(BindingList)
+	in.DeepCopyInto(out)
+	return out
+}
+
+// DeepCopyObject is an autogenerated deepcopy function, copying the receiver, creating a new runtime.Object.
+func (in *BindingList) DeepCopyObject() runtime.Object {
+	if c := in.DeepCopy(); c != nil {
+		return c
+	}
+	return nil
+}
+
+// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
+func (in *BindingSpec) DeepCopyInto(out *BindingSpec) {
+	*out = *in
+	in.Subject.DeepCopyInto(&out.Subject)
+	return
+}
+
+// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new BindingSpec.
+func (in *BindingSpec) DeepCopy() *BindingSpec {
+	if in == nil {
+		return nil
+	}
+	out := new(BindingSpec)
+	in.DeepCopyInto(out)
+	return out
+}
+
 // DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
 func (in *CloudEventAttributes) DeepCopyInto(out *CloudEventAttributes) {
 	*out = *in
diff --git a/vendor/knative.dev/pkg/apis/duck/v1beta1/fuzzer.go b/vendor/knative.dev/pkg/apis/duck/v1beta1/fuzzer.go
new file mode 100644
index 000000000..efc8be973
--- /dev/null
+++ b/vendor/knative.dev/pkg/apis/duck/v1beta1/fuzzer.go
@@ -0,0 +1,43 @@
+/*
+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 v1beta1
+
+import (
+	fuzz "github.com/google/gofuzz"
+	"k8s.io/apimachinery/pkg/api/apitesting/fuzzer"
+	"k8s.io/apimachinery/pkg/runtime/serializer"
+	"knative.dev/pkg/apis"
+	pkgfuzzer "knative.dev/pkg/apis/testing/fuzzer"
+)
+
+var testConditions = apis.Conditions{{Type: apis.ConditionReady}, {Type: apis.ConditionSucceeded}}
+
+// FuzzerFuncs includes fuzzing funcs for knative.dev/duck v1beta1 Status types
+//
+// For other examples see
+// https://github.com/kubernetes/apimachinery/blob/master/pkg/apis/meta/fuzzer/fuzzer.go
+var FuzzerFuncs = fuzzer.MergeFuzzerFuncs(
+	func(codecs serializer.CodecFactory) []interface{} {
+		return []interface{}{
+			func(status *Status, c fuzz.Continue) {
+				c.FuzzNoCustom(status) // fuzz the Status
+				status.SetConditions(testConditions)
+				pkgfuzzer.FuzzConditions(status, c)
+			},
+		}
+	},
+)
diff --git a/vendor/knative.dev/pkg/controller/controller.go b/vendor/knative.dev/pkg/controller/controller.go
index ee35f37bf..cee68931d 100644
--- a/vendor/knative.dev/pkg/controller/controller.go
+++ b/vendor/knative.dev/pkg/controller/controller.go
@@ -207,7 +207,7 @@ type Impl struct {
 
 // ControllerOptions encapsulates options for creating a new controller,
 // including throttling and stats behavior.
-type ControllerOptions struct {
+type ControllerOptions struct { //nolint for backcompat.
 	WorkQueueName string
 	Logger        *zap.SugaredLogger
 	Reporter      StatsReporter
diff --git a/vendor/knative.dev/pkg/hash/hash.go b/vendor/knative.dev/pkg/hash/hash.go
index b6b642122..1c4c5352f 100644
--- a/vendor/knative.dev/pkg/hash/hash.go
+++ b/vendor/knative.dev/pkg/hash/hash.go
@@ -153,7 +153,7 @@ func ChooseSubset(from sets.String, n int, target string) sets.String {
 		}
 		// Already matched this one. Continue to the next index.
 		for selection.Has(root) {
-			root += 1
+			root++
 			if root == hpl {
 				root = 0
 			}
diff --git a/vendor/knative.dev/pkg/injection/config.go b/vendor/knative.dev/pkg/injection/config.go
new file mode 100644
index 000000000..bae2b2e25
--- /dev/null
+++ b/vendor/knative.dev/pkg/injection/config.go
@@ -0,0 +1,99 @@
+/*
+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 injection
+
+import (
+	"errors"
+	"flag"
+	"log"
+	"math"
+	"os"
+	"os/user"
+	"path/filepath"
+
+	"k8s.io/client-go/rest"
+	"k8s.io/client-go/tools/clientcmd"
+	"k8s.io/klog"
+)
+
+// ParseAndGetRESTConfigOrDie parses the rest config flags and creates a client or
+// dies by calling log.Fatalf.
+func ParseAndGetRESTConfigOrDie() *rest.Config {
+	var (
+		serverURL = flag.String("server", "",
+			"The address of the Kubernetes API server. Overrides any value in kubeconfig. Only required if out-of-cluster.")
+		kubeconfig = flag.String("kubeconfig", "",
+			"Path to a kubeconfig. Only required if out-of-cluster.")
+
+		burst = flag.Int("kube-api-burst", 0, "Maximum burst for throttle.")
+		qps   = flag.Float64("kube-api-qps", 0, "Maximum QPS to the server from the client.")
+	)
+	klog.InitFlags(flag.CommandLine)
+	flag.Parse()
+
+	cfg, err := GetRESTConfig(*serverURL, *kubeconfig)
+	if err != nil {
+		log.Fatal("Error building kubeconfig: ", err)
+	}
+
+	if *burst < 0 {
+		log.Fatal("Invalid burst value", *burst)
+	}
+	if *qps < 0 || *qps > math.MaxFloat32 {
+		log.Fatal("Invalid QPS value", *qps)
+	}
+
+	cfg.Burst = *burst
+	cfg.QPS = float32(*qps)
+
+	return cfg
+}
+
+// GetRESTConfig returns a rest.Config to be used for kubernetes client creation.
+// It does so in the following order:
+//   1. Use the passed kubeconfig/serverURL.
+//   2. Fallback to the KUBECONFIG environment variable.
+//   3. Fallback to in-cluster config.
+//   4. Fallback to the ~/.kube/config.
+func GetRESTConfig(serverURL, kubeconfig string) (*rest.Config, error) {
+	if kubeconfig == "" {
+		kubeconfig = os.Getenv("KUBECONFIG")
+	}
+
+	// If we have an explicit indication of where the kubernetes config lives, read that.
+	if kubeconfig != "" {
+		c, err := clientcmd.BuildConfigFromFlags(serverURL, kubeconfig)
+		if err != nil {
+			return nil, err
+		}
+		return c, nil
+	}
+
+	// If not, try the in-cluster config.
+	if c, err := rest.InClusterConfig(); err == nil {
+		return c, nil
+	}
+
+	// If no in-cluster config, try the default location in the user's home directory.
+	if usr, err := user.Current(); err == nil {
+		if c, err := clientcmd.BuildConfigFromFlags("", filepath.Join(usr.HomeDir, ".kube", "config")); err == nil {
+			return c, nil
+		}
+	}
+
+	return nil, errors.New("could not create a valid kubeconfig")
+}
diff --git a/vendor/knative.dev/pkg/injection/injection.go b/vendor/knative.dev/pkg/injection/injection.go
new file mode 100644
index 000000000..d8cefc9fb
--- /dev/null
+++ b/vendor/knative.dev/pkg/injection/injection.go
@@ -0,0 +1,66 @@
+/*
+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 injection
+
+import (
+	"context"
+
+	"go.uber.org/zap"
+	"k8s.io/client-go/rest"
+
+	"knative.dev/pkg/controller"
+	"knative.dev/pkg/logging"
+	"knative.dev/pkg/signals"
+)
+
+// EnableInjectionOrDie enables Knative Client Injection, and provides a
+// callback to start the informers. Both Context and Config are optional.
+// Returns context with rest config set and a callback to start the informers
+// after watches have been set.
+//
+// Typical integration:
+// ```go
+//   ctx, startInformers := injection.EnableInjectionOrDie(signals.NewContext(), nil)
+//   ... start watches with informers, if required ...
+//   startInformers()
+// ```
+func EnableInjectionOrDie(ctx context.Context, cfg *rest.Config) (context.Context, func()) {
+	if ctx == nil {
+		ctx = signals.NewContext()
+	}
+	if cfg == nil {
+		cfg = ParseAndGetRESTConfigOrDie()
+	}
+
+	// Respect user provided settings, but if omitted customize the default behavior.
+	if cfg.QPS == 0 {
+		cfg.QPS = rest.DefaultQPS
+	}
+	if cfg.Burst == 0 {
+		cfg.Burst = rest.DefaultBurst
+	}
+	ctx = WithConfig(ctx, cfg)
+
+	ctx, informers := Default.SetupInformers(ctx, cfg)
+
+	return ctx, func() {
+		logging.FromContext(ctx).Info("Starting informers...")
+		if err := controller.StartInformers(ctx.Done(), informers...); err != nil {
+			logging.FromContext(ctx).Fatalw("Failed to start informers", zap.Error(err))
+		}
+	}
+}
diff --git a/vendor/knative.dev/pkg/injection/sharedmain/main.go b/vendor/knative.dev/pkg/injection/sharedmain/main.go
index 33d45391e..c9dd034de 100644
--- a/vendor/knative.dev/pkg/injection/sharedmain/main.go
+++ b/vendor/knative.dev/pkg/injection/sharedmain/main.go
@@ -18,16 +18,15 @@ package sharedmain
 
 import (
 	"context"
-	"errors"
 	"flag"
 	"log"
 	"net/http"
 	"os"
-	"os/user"
-	"path/filepath"
 	"time"
 
 	"go.opencensus.io/stats/view"
+	_ "go.uber.org/automaxprocs" // automatically set GOMAXPROCS based on cgroups
+	"go.uber.org/zap"
 	"golang.org/x/sync/errgroup"
 	corev1 "k8s.io/api/core/v1"
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
@@ -35,11 +34,6 @@ import (
 	"k8s.io/apimachinery/pkg/labels"
 	"k8s.io/apimachinery/pkg/util/wait"
 	"k8s.io/client-go/rest"
-	"k8s.io/client-go/tools/clientcmd"
-	"k8s.io/klog"
-
-	_ "go.uber.org/automaxprocs" // automatically set GOMAXPROCS based on cgroups
-	"go.uber.org/zap"
 
 	kubeclient "knative.dev/pkg/client/injection/kube/client"
 	"knative.dev/pkg/configmap"
@@ -63,37 +57,9 @@ import (
 //   2. Fallback to the KUBECONFIG environment variable.
 //   3. Fallback to in-cluster config.
 //   4. Fallback to the ~/.kube/config.
+// Deprecated: use injection.GetRESTConfig
 func GetConfig(serverURL, kubeconfig string) (*rest.Config, error) {
-	if kubeconfig == "" {
-		kubeconfig = os.Getenv("KUBECONFIG")
-	}
-
-	// We produce configs a bunch of ways, this gives us a single place
-	// to "decorate" them with common useful things (e.g. for debugging)
-	decorate := func(cfg *rest.Config) *rest.Config {
-		return cfg
-	}
-
-	// If we have an explicit indication of where the kubernetes config lives, read that.
-	if kubeconfig != "" {
-		c, err := clientcmd.BuildConfigFromFlags(serverURL, kubeconfig)
-		if err != nil {
-			return nil, err
-		}
-		return decorate(c), nil
-	}
-	// If not, try the in-cluster config.
-	if c, err := rest.InClusterConfig(); err == nil {
-		return decorate(c), nil
-	}
-	// If no in-cluster config, try the default location in the user's home directory.
-	if usr, err := user.Current(); err == nil {
-		if c, err := clientcmd.BuildConfigFromFlags("", filepath.Join(usr.HomeDir, ".kube", "config")); err == nil {
-			return decorate(c), nil
-		}
-	}
-
-	return nil, errors.New("could not create a valid kubeconfig")
+	return injection.GetRESTConfig(serverURL, kubeconfig)
 }
 
 // GetLoggingConfig gets the logging config from either the file system if present
@@ -129,40 +95,16 @@ func GetLeaderElectionConfig(ctx context.Context) (*leaderelection.Config, error
 
 // EnableInjectionOrDie enables Knative Injection and starts the informers.
 // Both Context and Config are optional.
+// Deprecated: use injection.EnableInjectionOrDie
 func EnableInjectionOrDie(ctx context.Context, cfg *rest.Config) context.Context {
-	if ctx == nil {
-		ctx = signals.NewContext()
-	}
-	if cfg == nil {
-		cfg = ParseAndGetConfigOrDie()
-	}
-
-	// Respect user provided settings, but if omitted customize the default behavior.
-	if cfg.QPS == 0 {
-		cfg.QPS = rest.DefaultQPS
-	}
-	if cfg.Burst == 0 {
-		cfg.Burst = rest.DefaultBurst
-	}
-	ctx = injection.WithConfig(ctx, cfg)
-
-	ctx, informers := injection.Default.SetupInformers(ctx, cfg)
-
-	// Start the injection clients and informers.
-	logging.FromContext(ctx).Info("Starting informers...")
-	go func(ctx context.Context) {
-		if err := controller.StartInformers(ctx.Done(), informers...); err != nil {
-			logging.FromContext(ctx).Fatalw("Failed to start informers", zap.Error(err))
-		}
-		<-ctx.Done()
-	}(ctx)
-
+	ctx, startInformers := injection.EnableInjectionOrDie(ctx, cfg)
+	go startInformers()
 	return ctx
 }
 
 // Main runs the generic main flow with a new context.
-// If any of the contructed controllers are AdmissionControllers or Conversion webhooks,
-// then a webhook is started to serve them.
+// If any of the constructed controllers are AdmissionControllers or Conversion
+// webhooks, then a webhook is started to serve them.
 func Main(component string, ctors ...injection.ControllerConstructor) {
 	// Set up signals so we handle the first shutdown signal gracefully.
 	MainWithContext(signals.NewContext(), component, ctors...)
@@ -185,7 +127,7 @@ func MainWithContext(ctx context.Context, component string, ctors ...injection.C
 			"issue upstream!")
 
 	// HACK: This parses flags, so the above should be set once this runs.
-	cfg := ParseAndGetConfigOrDie()
+	cfg := injection.ParseAndGetRESTConfigOrDie()
 
 	if *disableHighAvailability {
 		ctx = WithHADisabled(ctx)
@@ -225,16 +167,7 @@ func MainWithConfig(ctx context.Context, component string, cfg *rest.Config, cto
 		cfg.Burst = len(ctors) * rest.DefaultBurst
 	}
 
-	// Respect user provided settings, but if omitted customize the default behavior.
-	if cfg.QPS == 0 {
-		cfg.QPS = rest.DefaultQPS
-	}
-	if cfg.Burst == 0 {
-		cfg.Burst = rest.DefaultBurst
-	}
-	ctx = injection.WithConfig(ctx, cfg)
-
-	ctx, informers := injection.Default.SetupInformers(ctx, cfg)
+	ctx, startInformers := injection.EnableInjectionOrDie(ctx, cfg)
 
 	logger, atomicLevel := SetupLoggerOrDie(ctx, component)
 	defer flush(logger)
@@ -287,11 +220,10 @@ func MainWithConfig(ctx context.Context, component string, cfg *rest.Config, cto
 			return wh.Run(ctx.Done())
 		})
 	}
+
 	// Start the injection clients and informers.
-	logging.FromContext(ctx).Info("Starting informers...")
-	if err := controller.StartInformers(ctx.Done(), informers...); err != nil {
-		logging.FromContext(ctx).Fatalw("Failed to start informers", zap.Error(err))
-	}
+	startInformers()
+
 	// Wait for webhook informers to sync.
 	if wh != nil {
 		wh.InformersHaveSynced()
@@ -317,22 +249,9 @@ func flush(logger *zap.SugaredLogger) {
 
 // ParseAndGetConfigOrDie parses the rest config flags and creates a client or
 // dies by calling log.Fatalf.
+// Deprecated: use injeciton.ParseAndGetRESTConfigOrDie
 func ParseAndGetConfigOrDie() *rest.Config {
-	var (
-		serverURL = flag.String("server", "",
-			"The address of the Kubernetes API server. Overrides any value in kubeconfig. Only required if out-of-cluster.")
-		kubeconfig = flag.String("kubeconfig", "",
-			"Path to a kubeconfig. Only required if out-of-cluster.")
-	)
-	klog.InitFlags(flag.CommandLine)
-	flag.Parse()
-
-	cfg, err := GetConfig(*serverURL, *kubeconfig)
-	if err != nil {
-		log.Fatal("Error building kubeconfig: ", err)
-	}
-
-	return cfg
+	return injection.ParseAndGetRESTConfigOrDie()
 }
 
 // MemStatsOrDie sets up reporting on Go memory usage every 30 seconds or dies
diff --git a/vendor/knative.dev/pkg/logging/config.go b/vendor/knative.dev/pkg/logging/config.go
index 65409032a..d8c4b609b 100644
--- a/vendor/knative.dev/pkg/logging/config.go
+++ b/vendor/knative.dev/pkg/logging/config.go
@@ -240,8 +240,7 @@ func ConfigMapName() string {
 
 // JsonToLoggingConfig converts a json string of a Config.
 // Returns a non-nil Config always.
-// nolint:stylecheck // No rename due to backwards incompatibility.
-func JsonToLoggingConfig(jsonCfg string) (*Config, error) {
+func JsonToLoggingConfig(jsonCfg string) (*Config, error) { //nolint No rename due to backwards incompatibility.
 	if jsonCfg == "" {
 		return nil, errEmptyJSONLogginString
 	}
@@ -260,8 +259,7 @@ func JsonToLoggingConfig(jsonCfg string) (*Config, error) {
 }
 
 // LoggingConfigToJson converts a Config to a json string.
-// nolint:stylecheck // No rename due to backwards incompatibility.
-func LoggingConfigToJson(cfg *Config) (string, error) {
+func LoggingConfigToJson(cfg *Config) (string, error) { //nolint No rename due to backwards incompatibility.
 	if cfg == nil || cfg.LoggingConfig == "" {
 		return "", nil
 	}
diff --git a/vendor/knative.dev/pkg/metrics/config.go b/vendor/knative.dev/pkg/metrics/config.go
index bb55d410b..8eb829775 100644
--- a/vendor/knative.dev/pkg/metrics/config.go
+++ b/vendor/knative.dev/pkg/metrics/config.go
@@ -346,8 +346,7 @@ func prometheusPort() (int, error) {
 
 // JsonToMetricsOptions converts a json string of a
 // ExporterOptions. Returns a non-nil ExporterOptions always.
-// nolint:stylecheck // No rename due to backwards incompatibility.
-func JsonToMetricsOptions(jsonOpts string) (*ExporterOptions, error) {
+func JsonToMetricsOptions(jsonOpts string) (*ExporterOptions, error) { // nolint No rename due to backwards incompatibility.
 	var opts ExporterOptions
 	if jsonOpts == "" {
 		return nil, errors.New("json options string is empty")
@@ -361,8 +360,7 @@ func JsonToMetricsOptions(jsonOpts string) (*ExporterOptions, error) {
 }
 
 // MetricsOptionsToJson converts a ExporterOptions to a json string.
-// nolint:stylecheck // No rename due to backwards incompatibility.
-func MetricsOptionsToJson(opts *ExporterOptions) (string, error) {
+func MetricsOptionsToJson(opts *ExporterOptions) (string, error) { // nolint No rename due to backwards incompatibility.
 	if opts == nil {
 		return "", nil
 	}
diff --git a/vendor/knative.dev/pkg/metrics/stackdriver_exporter.go b/vendor/knative.dev/pkg/metrics/stackdriver_exporter.go
index ec9dba8c2..e3bebe187 100644
--- a/vendor/knative.dev/pkg/metrics/stackdriver_exporter.go
+++ b/vendor/knative.dev/pkg/metrics/stackdriver_exporter.go
@@ -89,7 +89,7 @@ var (
 	// which should be promoted to Stackdriver Resource labels via opencensus resources.
 	metricToResourceLabels = map[string]*resourceTemplate{}
 
-	// A variable for testing to reduce the size (number of metrics) buffered before
+	// TestOverrideBundleCount is a variable for testing to reduce the size (number of metrics) buffered before
 	// Stackdriver will send a bundled metric report. Only applies if non-zero.
 	TestOverrideBundleCount = 0
 )
diff --git a/vendor/knative.dev/pkg/reconciler/events.go b/vendor/knative.dev/pkg/reconciler/events.go
index 6c635643b..262b26222 100644
--- a/vendor/knative.dev/pkg/reconciler/events.go
+++ b/vendor/knative.dev/pkg/reconciler/events.go
@@ -62,7 +62,7 @@ func NewEvent(eventtype, reason, messageFmt string, args ...interface{}) Event {
 
 // ReconcilerEvent wraps the fields required for recorders to create a
 // kubernetes recorder Event.
-type ReconcilerEvent struct {
+type ReconcilerEvent struct { //nolint for backcompat.
 	EventType string
 	Reason    string
 	Format    string
diff --git a/vendor/knative.dev/pkg/tracing/config/tracing.go b/vendor/knative.dev/pkg/tracing/config/tracing.go
index baa3377d1..84128b440 100644
--- a/vendor/knative.dev/pkg/tracing/config/tracing.go
+++ b/vendor/knative.dev/pkg/tracing/config/tracing.go
@@ -132,8 +132,8 @@ func NewTracingConfigFromConfigMap(config *corev1.ConfigMap) (*Config, error) {
 
 // JsonToTracingConfig converts a json string of a Config.
 // Returns a non-nil Config always and an eventual error.
-// nolint:stylecheck // No rename due to backwards incompatibility.
-func JsonToTracingConfig(jsonCfg string) (*Config, error) {
+func JsonToTracingConfig(jsonCfg string) (*Config, error) { //nolint:stylecheck for backcompat.
+
 	if jsonCfg == "" {
 		return defaultConfig(), errors.New("empty json tracing config")
 	}
@@ -151,8 +151,7 @@ func JsonToTracingConfig(jsonCfg string) (*Config, error) {
 }
 
 // TracingConfigToJson converts a Config to a json string.
-// nolint:stylecheck // No rename due to backwards incompatibility.
-func TracingConfigToJson(cfg *Config) (string, error) {
+func TracingConfigToJson(cfg *Config) (string, error) { //nolint:stylecheck for backcompat.
 	if cfg == nil {
 		return "", nil
 	}
diff --git a/vendor/modules.txt b/vendor/modules.txt
index 6dc3508b3..0cff584f8 100644
--- a/vendor/modules.txt
+++ b/vendor/modules.txt
@@ -674,7 +674,7 @@ gopkg.in/jcmturner/rpc.v1/mstypes
 gopkg.in/jcmturner/rpc.v1/ndr
 # gopkg.in/yaml.v2 v2.3.0
 gopkg.in/yaml.v2
-# gopkg.in/yaml.v3 v3.0.0-20200601152816-913338de1bd2
+# gopkg.in/yaml.v3 v3.0.0-20200615113413-eeeca48fe776
 gopkg.in/yaml.v3
 # honnef.co/go/tools v0.0.1-2020.1.5
 ## explicit
@@ -1098,7 +1098,7 @@ k8s.io/utils/buffer
 k8s.io/utils/integer
 k8s.io/utils/pointer
 k8s.io/utils/trace
-# knative.dev/eventing v0.18.1-0.20201006061220-cca58af424d3
+# knative.dev/eventing v0.18.1-0.20201009013021-d37dbc88e6fd
 ## explicit
 knative.dev/eventing/pkg/adapter/v2
 knative.dev/eventing/pkg/adapter/v2/util/crstatusevent
@@ -1210,6 +1210,8 @@ knative.dev/eventing/test/lib/dropevents/dropeventsfibonacci
 knative.dev/eventing/test/lib/dropevents/dropeventsfirst
 knative.dev/eventing/test/lib/duck
 knative.dev/eventing/test/lib/recordevents
+knative.dev/eventing/test/lib/recordevents/observer
+knative.dev/eventing/test/lib/recordevents/recorder_vent
 knative.dev/eventing/test/lib/resources
 knative.dev/eventing/test/lib/sender
 knative.dev/eventing/test/performance/infra
@@ -1227,7 +1229,7 @@ knative.dev/eventing/test/test_images/print
 knative.dev/eventing/test/test_images/recordevents
 knative.dev/eventing/test/test_images/sequencestepper
 knative.dev/eventing/test/test_images/transformevents
-# knative.dev/pkg v0.0.0-20201006044120-247841408e57
+# knative.dev/pkg v0.0.0-20201009023821-73da5142da72
 ## explicit
 knative.dev/pkg/apis
 knative.dev/pkg/apis/duck
@@ -1323,7 +1325,7 @@ knative.dev/pkg/webhook/resourcesemantics
 knative.dev/pkg/webhook/resourcesemantics/conversion
 knative.dev/pkg/webhook/resourcesemantics/defaulting
 knative.dev/pkg/webhook/resourcesemantics/validation
-# knative.dev/test-infra v0.0.0-20201006044420-26457f2cad45
+# knative.dev/test-infra v0.0.0-20201008221921-aeca2ff2371a
 ## explicit
 knative.dev/test-infra/scripts
 # pack.ag/amqp v0.12.4


Test patch:
diff --git a/pkg/channel/consolidated/dispatcher/dispatcher_it_test.go b/pkg/channel/consolidated/dispatcher/dispatcher_it_test.go
index 48c7c64b4..115e53fd5 100644
--- a/pkg/channel/consolidated/dispatcher/dispatcher_it_test.go
+++ b/pkg/channel/consolidated/dispatcher/dispatcher_it_test.go
@@ -205,7 +205,7 @@ func TestDispatcher(t *testing.T) {
 	time.Sleep(5 * time.Second)
 
 	// Ok now everything should be ready to send the event
-	httpsender, err := kncloudevents.NewHttpMessageSender(nil, channelAProxy.URL)
+	httpsender, err := kncloudevents.NewHTTPMessageSender(nil, channelAProxy.URL)
 	if err != nil {
 		t.Fatal(err)
 	}
diff --git a/pkg/source/adapter/adapter_benchmark_test.go b/pkg/source/adapter/adapter_benchmark_test.go
index 1f70f2685..a0f2271db 100644
--- a/pkg/source/adapter/adapter_benchmark_test.go
+++ b/pkg/source/adapter/adapter_benchmark_test.go
@@ -56,7 +56,7 @@ func BenchmarkHandle(b *testing.B) {
 		b.Errorf("unexpected error, %v", err)
 	}
 
-	s := kncloudevents.HttpMessageSender{
+	s := kncloudevents.HTTPMessageSender{
 		Client: NewTestClient(func(req *http.Request) *http.Response {
 			return &http.Response{
 				StatusCode: 202,
diff --git a/pkg/source/adapter/adapter_test.go b/pkg/source/adapter/adapter_test.go
index 0822185f3..47755cf3d 100644
--- a/pkg/source/adapter/adapter_test.go
+++ b/pkg/source/adapter/adapter_test.go
@@ -335,7 +335,7 @@ func TestPostMessage_ServeHTTP_binary_mode(t *testing.T) {
 			//})
 			//defer time.Sleep(1 * time.Second)
 
-			s, err := kncloudevents.NewHttpMessageSender(nil, sinkServer.URL)
+			s, err := kncloudevents.NewHTTPMessageSender(nil, sinkServer.URL)
 			if err != nil {
 				t.Fatal(err)
 			}
diff --git a/vendor/knative.dev/eventing/test/conformance/helpers/broker_control_plane_test_helper.go b/vendor/knative.dev/eventing/test/conformance/helpers/broker_control_plane_test_helper.go
index d82789a57..0121f0471 100644
--- a/vendor/knative.dev/eventing/test/conformance/helpers/broker_control_plane_test_helper.go
+++ b/vendor/knative.dev/eventing/test/conformance/helpers/broker_control_plane_test_helper.go
@@ -25,6 +25,7 @@ import (
 	"knative.dev/pkg/reconciler"
 
 	eventingv1beta1 "knative.dev/eventing/pkg/apis/eventing/v1beta1"
+	"knative.dev/eventing/test/lib/recordevents"
 
 	testlib "knative.dev/eventing/test/lib"
 	"knative.dev/eventing/test/lib/duck"
@@ -82,8 +83,7 @@ func triggerV1Beta1BeforeBrokerHelper(triggerName string, client *testlib.Client
 	const etLogger = "logger"
 	const loggerPodName = "logger-pod"
 
-	logPod := resources.EventRecordPod(loggerPodName)
-	client.CreatePodOrFail(logPod, testlib.WithService(loggerPodName))
+	_ = recordevents.DeployEventRecordOrFail(context.TODO(), client, loggerPodName)
 	client.WaitForAllTestResourcesReadyOrFail(context.Background()) // Can't do this for the trigger because it's not 'ready' yet
 	client.CreateTriggerOrFailV1Beta1(triggerName,
 		resources.WithAttributesTriggerFilterV1Beta1(eventingv1beta1.TriggerAnyFilter, etLogger, map[string]interface{}{}),
diff --git a/vendor/knative.dev/eventing/test/conformance/helpers/broker_tracing_test_helper.go b/vendor/knative.dev/eventing/test/conformance/helpers/broker_tracing_test_helper.go
index 41db7080d..9c09109c9 100644
--- a/vendor/knative.dev/eventing/test/conformance/helpers/broker_tracing_test_helper.go
+++ b/vendor/knative.dev/eventing/test/conformance/helpers/broker_tracing_test_helper.go
@@ -31,6 +31,7 @@ import (
 	"knative.dev/eventing/pkg/apis/eventing/v1beta1"
 	tracinghelper "knative.dev/eventing/test/conformance/helpers/tracing"
 	testlib "knative.dev/eventing/test/lib"
+	"knative.dev/eventing/test/lib/recordevents"
 	"knative.dev/eventing/test/lib/resources"
 	"knative.dev/eventing/test/lib/sender"
 )
@@ -82,8 +83,7 @@ func setupBrokerTracing(ctx context.Context, brokerClass string) SetupTracingTes
 		)
 
 		// Create a logger (EventRecord) Pod and a K8s Service that points to it.
-		logPod := resources.EventRecordPod(loggerPodName)
-		client.CreatePodOrFail(logPod, testlib.WithService(loggerPodName))
+		_ = recordevents.DeployEventRecordOrFail(ctx, client, loggerPodName)
 
 		// Create a Trigger that receives events (type=bar) and sends them to the logger Pod.
 		loggerTrigger := client.CreateTriggerOrFailV1Beta1(
diff --git a/vendor/knative.dev/eventing/test/conformance/helpers/channel_status_subscriber_test_helper.go b/vendor/knative.dev/eventing/test/conformance/helpers/channel_status_subscriber_test_helper.go
index d148e26d6..4a0c2e798 100644
--- a/vendor/knative.dev/eventing/test/conformance/helpers/channel_status_subscriber_test_helper.go
+++ b/vendor/knative.dev/eventing/test/conformance/helpers/channel_status_subscriber_test_helper.go
@@ -24,6 +24,7 @@ import (
 	duckv1beta1 "knative.dev/eventing/pkg/apis/duck/v1beta1"
 	eventingv1beta1 "knative.dev/eventing/pkg/apis/messaging/v1beta1"
 	testlib "knative.dev/eventing/test/lib"
+	"knative.dev/eventing/test/lib/recordevents"
 	"knative.dev/eventing/test/lib/resources"
 
 	corev1 "k8s.io/api/core/v1"
@@ -59,8 +60,7 @@ func channelHasRequiredSubscriberStatus(ctx context.Context, st *testing.T, clie
 	client.CreateChannelOrFail(channelName, &channel)
 	client.WaitForResourceReadyOrFail(channelName, &channel)
 
-	pod := resources.EventRecordPod(subscriberServiceName + "-pod")
-	client.CreatePodOrFail(pod, testlib.WithService(subscriberServiceName))
+	_ = recordevents.DeployEventRecordOrFail(context.TODO(), client, subscriberServiceName+"-pod")
 
 	subscription := client.CreateSubscriptionOrFail(
 		subscriberServiceName,
diff --git a/vendor/knative.dev/eventing/test/conformance/helpers/channel_tracing_test_helper.go b/vendor/knative.dev/eventing/test/conformance/helpers/channel_tracing_test_helper.go
index 38afa62cf..c8d95f567 100644
--- a/vendor/knative.dev/eventing/test/conformance/helpers/channel_tracing_test_helper.go
+++ b/vendor/knative.dev/eventing/test/conformance/helpers/channel_tracing_test_helper.go
@@ -29,6 +29,7 @@ import (
 
 	tracinghelper "knative.dev/eventing/test/conformance/helpers/tracing"
 	testlib "knative.dev/eventing/test/lib"
+	"knative.dev/eventing/test/lib/recordevents"
 	"knative.dev/eventing/test/lib/resources"
 	"knative.dev/eventing/test/lib/sender"
 )
@@ -69,8 +70,7 @@ func setupChannelTracingWithReply(
 	client.CreateChannelOrFail(replyChannelName, channel)
 
 	// Create the 'sink', a LogEvents Pod and a K8s Service that points to it.
-	recordEventsPod := resources.EventRecordPod(recordEventsPodName)
-	client.CreatePodOrFail(recordEventsPod, testlib.WithService(recordEventsPodName))
+	recordEventsPod := recordevents.DeployEventRecordOrFail(ctx, client, recordEventsPodName)
 
 	// Create the subscriber, a Pod that mutates the event.
 	transformerPod := resources.EventTransformationPod(
diff --git a/vendor/knative.dev/eventing/test/conformance/helpers/tracing_test_helper.go b/vendor/knative.dev/eventing/test/conformance/helpers/tracing_test_helper.go
index 1b2f9af79..f3e476f9f 100644
--- a/vendor/knative.dev/eventing/test/conformance/helpers/tracing_test_helper.go
+++ b/vendor/knative.dev/eventing/test/conformance/helpers/tracing_test_helper.go
@@ -67,7 +67,7 @@ func tracingTest(
 	expectedTestSpan, eventMatcher := setupInfrastructure(ctx, t, &channel, client, recordEventsPodName, true)
 
 	// Start the event info store and assert the event was received correctly
-	targetTracker, err := recordevents.NewEventInfoStore(client, recordEventsPodName)
+	targetTracker, err := recordevents.NewEventInfoStore(client, recordEventsPodName, client.Namespace)
 	if err != nil {
 		t.Fatal("Pod tracker failed:", err)
 	}
@@ -107,6 +107,6 @@ func getTraceIDHeader(t *testing.T, evInfos []recordevents.EventInfo) string {
 			}
 		}
 	}
-	t.Fatalf("FAIL: No traceid in %d messages: (%s)", len(evInfos), evInfos)
+	t.Fatalf("FAIL: No traceid in %d messages: (%v)", len(evInfos), evInfos)
 	return ""
 }
diff --git a/vendor/knative.dev/eventing/test/e2e/helpers/trigger_no_broker_test_helper.go b/vendor/knative.dev/eventing/test/e2e/helpers/trigger_no_broker_test_helper.go
index ce8a4d9c4..0303be7e8 100644
--- a/vendor/knative.dev/eventing/test/e2e/helpers/trigger_no_broker_test_helper.go
+++ b/vendor/knative.dev/eventing/test/e2e/helpers/trigger_no_broker_test_helper.go
@@ -27,6 +27,7 @@ import (
 	"k8s.io/apimachinery/pkg/util/wait"
 
 	testlib "knative.dev/eventing/test/lib"
+	"knative.dev/eventing/test/lib/recordevents"
 	"knative.dev/eventing/test/lib/resources"
 )
 
@@ -41,8 +42,8 @@ func TestTriggerNoBroker(ctx context.Context, t *testing.T, channel string, brok
 	brokerName := strings.ToLower(channel)
 
 	subscriberName := "dumper-empty"
-	eventRecordPod := resources.EventRecordPod(subscriberName)
-	client.CreatePodOrFail(eventRecordPod, testlib.WithService(subscriberName))
+	recordevents.DeployEventRecordOrFail(context.TODO(), client, subscriberName)
+
 	client.CreateTriggerOrFailV1Beta1("testtrigger",
 		resources.WithSubscriberServiceRefForTriggerV1Beta1(subscriberName),
 		resources.WithBrokerV1Beta1(brokerName),
diff --git a/vendor/knative.dev/eventing/test/lib/client.go b/vendor/knative.dev/eventing/test/lib/client.go
index 1eaef6ece..d1c5782cc 100644
--- a/vendor/knative.dev/eventing/test/lib/client.go
+++ b/vendor/knative.dev/eventing/test/lib/client.go
@@ -48,6 +48,8 @@ type Client struct {
 	Dynamic       dynamic.Interface
 	Config        *rest.Config
 
+	EventListener *EventListener
+
 	Namespace string
 	T         *testing.T
 	Tracker   *Tracker
@@ -93,6 +95,10 @@ func NewClient(configPath string, clusterName string, namespace string, t *testi
 	client.T = t
 	client.Tracker = NewTracker(t, client.Dynamic)
 
+	// Start informer
+	client.EventListener = NewEventListener(client.Kube.Kube, client.Namespace, client.T.Logf)
+	client.Cleanup(client.EventListener.Stop)
+
 	client.tracingEnv, err = getTracingConfig(client.Kube.Kube)
 	if err != nil {
 		return nil, err
diff --git a/vendor/knative.dev/eventing/test/lib/k8s_events.go b/vendor/knative.dev/eventing/test/lib/k8s_events.go
new file mode 100644
index 000000000..1188d5c23
--- /dev/null
+++ b/vendor/knative.dev/eventing/test/lib/k8s_events.go
@@ -0,0 +1,84 @@
+/*
+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 lib
+
+import (
+	"context"
+	"sync"
+
+	corev1 "k8s.io/api/core/v1"
+	"k8s.io/client-go/informers"
+	"k8s.io/client-go/kubernetes"
+	"k8s.io/client-go/tools/cache"
+)
+
+// EventHandler is the callback type for the EventListener
+type EventHandler func(event *corev1.Event)
+
+// EventListener is a type that broadcasts new k8s events
+type EventListener struct {
+	cancel context.CancelFunc
+
+	lock     sync.Mutex
+	handlers []EventHandler
+}
+
+// NewEventListener creates a new event listener
+func NewEventListener(client kubernetes.Interface, namespace string, logf func(string, ...interface{})) *EventListener {
+	ctx, cancelCtx := context.WithCancel(context.Background())
+	informerFactory := informers.NewSharedInformerFactoryWithOptions(
+		client,
+		0,
+		informers.WithNamespace(namespace),
+	)
+	eventsInformer := informerFactory.Core().V1().Events().Informer()
+
+	el := EventListener{
+		cancel: cancelCtx,
+	}
+
+	eventsInformer.AddEventHandler(cache.ResourceEventHandlerFuncs{
+		AddFunc: func(obj interface{}) {
+			el.handle(obj.(*corev1.Event))
+		},
+	})
+
+	go func() {
+		eventsInformer.Run(ctx.Done())
+		logf("EventListener stopped")
+	}()
+
+	return &el
+}
+
+func (el *EventListener) handle(event *corev1.Event) {
+	el.lock.Lock()
+	defer el.lock.Unlock()
+	for _, handler := range el.handlers {
+		handler(event)
+	}
+}
+
+func (el *EventListener) AddHandler(handler EventHandler) {
+	el.lock.Lock()
+	defer el.lock.Unlock()
+	el.handlers = append(el.handlers, handler)
+}
+
+func (el *EventListener) Stop() {
+	el.cancel()
+}
diff --git a/vendor/knative.dev/eventing/test/lib/recordevents/event_info.go b/vendor/knative.dev/eventing/test/lib/recordevents/event_info.go
index dc0ffd1f0..d9dbbdae8 100644
--- a/vendor/knative.dev/eventing/test/lib/recordevents/event_info.go
+++ b/vendor/knative.dev/eventing/test/lib/recordevents/event_info.go
@@ -17,56 +17,31 @@ limitations under the License.
 package recordevents
 
 import (
-	"context"
-	"encoding/json"
 	"fmt"
-	"io/ioutil"
-	"math/rand"
-	"net/http"
 	"strings"
 	"time"
 
 	cloudevents "github.com/cloudevents/sdk-go/v2"
-	corev1 "k8s.io/api/core/v1"
-	v1 "k8s.io/api/core/v1"
-	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
-	"k8s.io/apimachinery/pkg/util/wait"
-	"k8s.io/client-go/kubernetes"
-	"knative.dev/pkg/test/monitoring"
-
-	"knative.dev/pkg/test/logging"
-
-	testlib "knative.dev/eventing/test/lib"
 )
 
-// Port for the recordevents pod REST listener
-const RecordEventsPort = 8392
-
-// HTTP path for the GetMinMax REST call
-const GetMinMaxPath = "/minmax"
-
-// HTTP path for the GetEntry REST call
-const GetEntryPath = "/entry/"
-
-// HTTP path for the TrimThrough REST call
-const TrimThroughPath = "/trimthrough/"
-
-// On-wire json rest api format for recordevents GetMinMax calls
-// sennt to the recordevents pod.
-type MinMaxResponse struct {
-	MinAvail int
-	MaxSeen  int
-}
+const (
+	// EventReason is the Kubernetes event reason used for observed events.
+	CloudEventObservedReason = "CloudEventObserved"
+)
 
 // Structure to hold information about an event seen by recordevents pod.
 type EventInfo struct {
 	// Set if the http request received by the pod couldn't be decoded or
 	// didn't pass validation
-	Error string
+	Error string `json:"error,omitempty"`
 	// Event received if the cloudevent received by the pod passed validation
-	Event *cloudevents.Event
+	Event *cloudevents.Event `json:"event,omitempty"`
 	// HTTPHeaders of the connection that delivered the event
-	HTTPHeaders map[string][]string
+	HTTPHeaders map[string][]string `json:"httpHeaders,omitempty"`
+	Origin      string              `json:"origin,omitempty"`
+	Observer    string              `json:"observer,omitempty"`
+	Time        time.Time           `json:"time,omitempty"`
+	Sequence    uint64              `json:"sequence"`
 }
 
 // Pretty print the event. Meant for debugging.  This formats the validation error
@@ -95,189 +70,3 @@ func (s *SearchedInfo) String() string {
 	}
 	return sb.String()
 }
-
-// Connection state for a REST connection to a pod
-type eventGetter struct {
-	podName       string
-	podNamespace  string
-	podPort       int
-	kubeClientset kubernetes.Interface
-	logf          logging.FormatLogger
-
-	host       string
-	port       int
-	forwardPID int
-}
-
-// Creates a forwarded port to the specified recordevents pod and waits until
-// it can successfully talk to the REST API.  Times out after timeoutEvRetry
-func newEventGetter(podName string, client *testlib.Client, logf logging.FormatLogger) (eventGetterInterface, error) {
-	egi := &eventGetter{podName: podName, podNamespace: client.Namespace,
-		kubeClientset: client.Kube.Kube, podPort: RecordEventsPort, logf: logf}
-	err := egi.forwardPort()
-	if err != nil {
-		return nil, err
-	}
-
-	err = egi.waitTillUp()
-	if err != nil {
-		return nil, err
-	}
-	return egi, nil
-}
-
-// Get information about the provided podName.  Uses list (rather than get) and
-// returns a pod list for compatibility with the monitoring.PortForward
-// interface
-func (eg *eventGetter) getRunningPodInfo(podName, namespace string) (*v1.PodList, error) {
-	pods, err := eg.kubeClientset.CoreV1().Pods(namespace).List(context.Background(),
-		metav1.ListOptions{FieldSelector: fmt.Sprintf("metadata.name=%s", podName)})
-	if err == nil && len(pods.Items) != 1 {
-		err = fmt.Errorf("no %s Pod found on the cluster", podName)
-	} else if pods.Items[0].Status.Phase != corev1.PodRunning {
-		err = fmt.Errorf("pod %s in state %s, wanted Running", podName,
-			pods.Items[0].Status.Phase)
-	}
-
-	return pods, err
-}
-
-// Try to forward the pod port to a local port somewhere in the range 30000-60000.
-// keeps retrying with random ports in that range, timing out after timeoutEvRetry
-func (eg *eventGetter) forwardPort() error {
-	portRand := rand.New(rand.NewSource(time.Now().UnixNano()))
-	portMin := 30000
-	portMax := 60000
-	var internalErr error
-
-	wait.PollImmediate(minEvRetryInterval, timeoutEvRetry, func() (bool, error) {
-		localPort := portMin + portRand.Intn(portMax-portMin)
-		if err := monitoring.CheckPortAvailability(localPort); err != nil {
-			internalErr = err
-			return false, nil
-		}
-		pods, err := eg.getRunningPodInfo(eg.podName, eg.podNamespace)
-		if err != nil {
-			internalErr = err
-			return false, nil
-		}
-
-		pid, err := monitoring.PortForward(eg.logf, pods, localPort, eg.podPort, eg.podNamespace)
-		if err != nil {
-			internalErr = err
-			return false, nil
-		}
-		internalErr = nil
-
-		eg.forwardPID = pid
-		eg.port = localPort
-		eg.host = "localhost"
-		return true, nil
-	})
-	if internalErr != nil {
-		return fmt.Errorf("timeout forwarding port: %v", internalErr)
-	}
-	return nil
-}
-
-// Return the min available, max seen by the recordevents pod.
-// maxRet is the largest event that has ever been seen (whether it's been trimmed
-// or not).  minRet is the smallest event still available via Get, or 1+maxRet if
-// no events are available.  maxRet starts at 0 when no events have been seen.
-func (eg *eventGetter) getMinMax() (minRet int, maxRet int, errRet error) {
-	resp, err := http.Get(fmt.Sprintf("http://%s:%d%s", eg.host, eg.port, GetMinMaxPath))
-	if err != nil {
-		return -1, -1, fmt.Errorf("http get error: %v", err)
-	}
-	defer resp.Body.Close()
-	bodyContents, err := ioutil.ReadAll(resp.Body)
-	if err != nil {
-		return -1, -1, fmt.Errorf("error reading response body %w", err)
-	}
-	if resp.StatusCode != http.StatusOK {
-		return -1, -1, fmt.Errorf("error %d reading GetMinMax response", resp.StatusCode)
-	}
-	minMaxResponse := MinMaxResponse{}
-	err = json.Unmarshal(bodyContents, &minMaxResponse)
-	if err != nil {
-		return -1, -1, fmt.Errorf("error unmarshalling response %w", err)
-	}
-	if minMaxResponse.MinAvail == 0 {
-		return -1, -1, fmt.Errorf("invalid decoded json: %+v", minMaxResponse)
-	}
-
-	return minMaxResponse.MinAvail, minMaxResponse.MaxSeen, nil
-}
-
-// Return the event with the provided sequence number.  Returns the appropriate
-// EventInfo or an error if no such event is known, or the event has already
-// been trimmed.
-func (eg *eventGetter) getEntry(seqno int) (EventInfo, error) {
-	resp, err := http.Get(fmt.Sprintf("http://%s:%d%s/%d", eg.host, eg.port, GetEntryPath, seqno))
-	if err != nil {
-		return EventInfo{}, fmt.Errorf("http get err %v", err)
-	}
-	defer resp.Body.Close()
-	bodyContents, err := ioutil.ReadAll(resp.Body)
-	if err != nil {
-		return EventInfo{}, fmt.Errorf("error reading response body %w", err)
-	}
-	if resp.StatusCode != http.StatusOK {
-		return EventInfo{}, fmt.Errorf("error %d reading GetEntry response", resp.StatusCode)
-	}
-	entryResponse := EventInfo{}
-	err = json.Unmarshal(bodyContents, &entryResponse)
-	if err != nil {
-		return EventInfo{}, fmt.Errorf("error unmarshalling response %w", err)
-	}
-	if len(entryResponse.Error) == 0 && entryResponse.Event == nil {
-		return EventInfo{}, fmt.Errorf("invalid decoded json: %+v", entryResponse)
-	}
-
-	return entryResponse, nil
-}
-
-// Trim the events up to and including seqno from the recordevents pod.
-// Returns an error if a nonsensical seqno is passed in, but does not return
-// error for trimming already trimmed regions.
-func (eg *eventGetter) trimThrough(seqno int) error {
-	resp, err := http.Post(fmt.Sprintf("http://%s:%d%s/%d", eg.host, eg.port, TrimThroughPath, seqno), "", nil)
-	if err != nil {
-		return fmt.Errorf("http post err %v", err)
-	}
-	defer resp.Body.Close()
-	body, err := ioutil.ReadAll(resp.Body)
-	if err != nil {
-		return fmt.Errorf("error reading response body %w", err)
-	}
-	if resp.StatusCode != http.StatusOK {
-		return fmt.Errorf("error %d reading TrimThrough response: %s", resp.StatusCode, string(body))
-	}
-
-	return nil
-}
-
-// Clean up the getter by tearing down the port forward.
-func (eg *eventGetter) cleanup() {
-	pid := eg.forwardPID
-	eg.forwardPID = 0
-	if pid != 0 {
-		monitoring.Cleanup(pid)
-	}
-}
-
-// Wait (up to timeoutEvRetry) for the pod to RestAPI to answer request.
-func (eg *eventGetter) waitTillUp() error {
-	var internalErr error
-	wait.PollImmediate(minEvRetryInterval, timeoutEvRetry, func() (bool, error) {
-		_, _, internalErr = eg.getMinMax()
-		if internalErr != nil {
-			return false, nil
-		}
-		return true, nil
-	})
-	if internalErr != nil {
-		return fmt.Errorf("timeout waiting for recordevents pod to come up: %v", internalErr)
-	}
-	return nil
-}
diff --git a/vendor/knative.dev/eventing/test/lib/recordevents/event_info_store.go b/vendor/knative.dev/eventing/test/lib/recordevents/event_info_store.go
index 52fcbae3c..007a4c149 100644
--- a/vendor/knative.dev/eventing/test/lib/recordevents/event_info_store.go
+++ b/vendor/knative.dev/eventing/test/lib/recordevents/event_info_store.go
@@ -18,6 +18,7 @@ package recordevents
 
 import (
 	"context"
+	"encoding/json"
 	"fmt"
 	"strconv"
 	"strings"
@@ -27,6 +28,7 @@ import (
 
 	"github.com/pkg/errors"
 	corev1 "k8s.io/api/core/v1"
+	rbacv1 "k8s.io/api/rbac/v1"
 	"k8s.io/apimachinery/pkg/util/wait"
 	pkgTest "knative.dev/pkg/test"
 
@@ -36,169 +38,106 @@ import (
 
 const (
 	// The interval and timeout used for checking events
-	minEvRetryInterval = 4 * time.Second
-	timeoutEvRetry     = 4 * time.Minute
+	retryInterval = 4 * time.Second
+	retryTimeout  = 4 * time.Minute
 )
 
-// Stateful store of events received by the recordevents pod it is pointed at.
-// This pulls events from the pod during any Find or Wait call, storing them
-// locally and triming them from the remote pod store.
-type EventInfoStore struct {
-	tb testing.TB
-
-	podName string
-	getter  eventGetterInterface
-
-	lock          sync.Mutex
-	allEvents     []EventInfo
-	firstID       int
-	closeCh       chan struct{}
-	doRefresh     chan chan error
-	timeout       time.Duration
-	retryInterval time.Duration
-}
-
-// Functions used for getting data from the REST api of the recordevents pod.
-// The interface exists for use with unit tests of this module.
-type eventGetterInterface interface {
-	getMinMax() (minRet int, maxRet int, errRet error)
-	getEntry(seqno int) (EventInfo, error)
-	trimThrough(seqno int) error
-	cleanup()
-}
-
-// Internal function to create an event store.  This is called directly by unit tests of
-// this module.
-func newTestableEventInfoStore(egi eventGetterInterface, retryInterval time.Duration,
-	timeout time.Duration) *EventInfoStore {
-	if timeout == -1 {
-		timeout = timeoutEvRetry
-	}
-	if retryInterval == -1 {
-		retryInterval = minEvRetryInterval
-	}
-	ei := &EventInfoStore{getter: egi, firstID: 1, timeout: timeout, retryInterval: retryInterval}
-	ei.start()
-	return ei
-}
+type EventRecordOption = func(*corev1.Pod, *testlib.Client) error
 
-// Creates an EventInfoStore that is used to iteratively download events recorded by the
-// recordevents pod.  Calling this forwards the recordevents port to the local machine
-// and blocks waiting to connect to that pod.  Fails if it cannot connect within
-// the expected timeout (4 minutes currently)
-func NewEventInfoStore(client *testlib.Client, podName string) (*EventInfoStore, error) {
-	egi, err := newEventGetter(podName, client, client.T.Logf)
+func DeployEventRecordOrFail(ctx context.Context, client *testlib.Client, name string, options ...EventRecordOption) *corev1.Pod {
+	client.CreateServiceAccountOrFail(name)
+	client.CreateRoleOrFail(resources.Role(name,
+		resources.WithRuleForRole(&rbacv1.PolicyRule{
+			APIGroups: []string{""},
+			Resources: []string{"pods"},
+			Verbs:     []string{"get"},
+		}),
+		resources.WithRuleForRole(&rbacv1.PolicyRule{
+			APIGroups: []string{""},
+			Resources: []string{"events"},
+			Verbs:     []string{rbacv1.VerbAll},
+		}),
+	))
+	client.CreateRoleBindingOrFail(name, "Role", name, name, client.Namespace)
+
+	eventRecordPod := EventRecordPod(name, name)
+	client.CreatePodOrFail(eventRecordPod, append(options, testlib.WithService(name))...)
+	err := pkgTest.WaitForPodRunning(ctx, client.Kube, name, client.Namespace)
 	if err != nil {
-		return nil, err
+		client.T.Fatalf("Failed to start the recordevent pod '%s': %v", name, errors.WithStack(err))
 	}
-	ei := newTestableEventInfoStore(egi, -1, -1)
-	ei.podName = podName
-	ei.tb = client.T
-	client.Cleanup(ei.cleanup)
-	return ei, nil
+	client.WaitForServiceEndpointsOrFail(ctx, name, 1)
+	return eventRecordPod
 }
 
-type EventRecordOption = func(*corev1.Pod, *testlib.Client) error
-
 // Deploys a new recordevents pod and start the associated EventInfoStore
 func StartEventRecordOrFail(ctx context.Context, client *testlib.Client, podName string, options ...EventRecordOption) (*EventInfoStore, *corev1.Pod) {
-	eventRecordPod := resources.EventRecordPod(podName)
-	client.CreatePodOrFail(eventRecordPod, append(options, testlib.WithService(podName))...)
-	err := pkgTest.WaitForPodRunning(ctx, client.Kube, podName, client.Namespace)
-	if err != nil {
-		client.T.Fatalf("Failed to start the recordevent pod '%s': %v", podName, errors.WithStack(err))
-	}
-	client.WaitForServiceEndpointsOrFail(ctx, podName, 1)
+	eventRecordPod := DeployEventRecordOrFail(ctx, client, podName, options...)
 
-	eventTracker, err := NewEventInfoStore(client, podName)
+	eventTracker, err := NewEventInfoStore(client, podName, client.Namespace)
 	if err != nil {
 		client.T.Fatalf("Failed to start the EventInfoStore associated to pod '%s': %v", podName, err)
 	}
 	return eventTracker, eventRecordPod
 }
 
-// Starts the single threaded background goroutine used to update local state
-// from the remote REST API.
-func (ei *EventInfoStore) start() {
-	ei.closeCh = make(chan struct{})
-	ei.doRefresh = make(chan chan error)
-	go func() {
-		for {
-			select {
-			case <-ei.closeCh:
-				ei.getter.cleanup()
-				return
-			case replyCh := <-ei.doRefresh:
-				replyCh <- ei.doRetrieveData()
-			}
-		}
-	}()
+// Stateful store of events received by the recordevents pod it is pointed at.
+// This pulls events from the pod during any Find or Wait call, storing them
+// locally and triming them from the remote pod store.
+type EventInfoStore struct {
+	tb testing.TB
+
+	podName      string
+	podNamespace string
+
+	lock      sync.Mutex
+	collected []EventInfo
 }
 
-// The data update thread used by the single threaded background goroutine
-// for updating data from the REST api.
-func (ei *EventInfoStore) doRetrieveData() error {
-	min, max, err := ei.getter.getMinMax()
-	if err != nil {
-		return fmt.Errorf("error getting MinMax %v", err)
-	}
-	ei.lock.Lock()
-	curMin := ei.firstID
-	curMax := curMin + len(ei.allEvents) - 1
-	ei.lock.Unlock()
-	if min == max+1 {
-		// Nothing to read or trim
-		return nil
-	} else {
-		if min > curMax+1 {
-			return fmt.Errorf("mismatched stored max/available min: %d, %d", curMax, min)
-		}
-		min = curMax + 1
-		// We may have data to read, definitely have data to trim.
+// Creates an EventInfoStore that is used to iteratively download events recorded by the
+// recordevents pod.
+func NewEventInfoStore(client *testlib.Client, podName string, podNamespace string) (*EventInfoStore, error) {
+	store := &EventInfoStore{
+		tb:           client.T,
+		podName:      podName,
+		podNamespace: podNamespace,
 	}
-	var newEvents []EventInfo
-	for i := min; i <= max; i++ {
-		e, err := ei.getter.getEntry(i)
-		if err != nil {
-			return fmt.Errorf("error calling getEntry of %d %v", i, err)
-		}
 
-		newEvents = append(newEvents, e)
-	}
-	ei.lock.Lock()
-	ei.allEvents = append(ei.allEvents, newEvents...)
-	ei.lock.Unlock()
-	err = ei.getter.trimThrough(max)
-	return err
+	client.EventListener.AddHandler(store.handle)
 
+	return store, nil
 }
 
-// Clean up any background resources used by the store.  Must be called exactly once after
-// the last use.
-func (ei *EventInfoStore) cleanup() {
-	close(ei.closeCh)
+func (ei *EventInfoStore) getEventInfo() []EventInfo {
+	ei.lock.Lock()
+	defer ei.lock.Unlock()
+	return ei.collected
 }
 
-//TODO remove it, this is not useful anymore
-// Deprecated: you can remove the manual cleanup of the event getter, since now it's done at test tear down automatically
-func (ei *EventInfoStore) Cleanup() {}
-
-// Called internally by functions wanting the current list of all
-// known events.  This calls for an update from the REST server and
-// returns the summary of all locally and remotely known events.
-// Returns an error in case of a connection or protocol error.
-func (ei *EventInfoStore) refreshData() ([]EventInfo, error) {
-	var allEvents []EventInfo
-	replyCh := make(chan error)
-	ei.doRefresh <- replyCh
-	err := <-replyCh
+func (ei *EventInfoStore) handle(event *corev1.Event) {
+	// Filter events
+	if !ei.isMyEvent(event) {
+		return
+	}
+
+	eventInfo := EventInfo{}
+	err := json.Unmarshal([]byte(event.Message), &eventInfo)
 	if err != nil {
-		return nil, err
+		ei.tb.Errorf("Received EventInfo that cannot be unmarshalled! %+v", err)
+		return
 	}
+
 	ei.lock.Lock()
-	allEvents = append(allEvents, ei.allEvents...)
-	ei.lock.Unlock()
-	return allEvents, nil
+	defer ei.lock.Unlock()
+	ei.collected = append(ei.collected, eventInfo)
+}
+
+func (ei *EventInfoStore) isMyEvent(event *corev1.Event) bool {
+	return event.Type == corev1.EventTypeNormal &&
+		event.Reason == CloudEventObservedReason &&
+		event.InvolvedObject.Kind == "Pod" &&
+		event.InvolvedObject.Name == ei.podName &&
+		event.InvolvedObject.Namespace == ei.podNamespace
 }
 
 // Find all events received by the recordevents pod that match the provided matchers,
@@ -216,10 +155,7 @@ func (ei *EventInfoStore) Find(matchers ...EventInfoMatcher) ([]EventInfo, Searc
 	lastEvents := []EventInfo{}
 	var nonMatchingErrors []error
 
-	allEvents, err := ei.refreshData()
-	if err != nil {
-		return nil, sInfo, nonMatchingErrors, fmt.Errorf("error getting events %v", err)
-	}
+	allEvents := ei.getEventInfo()
 	for i := range allEvents {
 		if err := f(allEvents[i]); err == nil {
 			allMatch = append(allMatch, allEvents[i])
@@ -241,27 +177,31 @@ func (ei *EventInfoStore) Find(matchers ...EventInfoMatcher) ([]EventInfo, Searc
 // Assert that there are at least min number of match for the provided matchers.
 // This method fails the test if the assert is not fulfilled.
 func (ei *EventInfoStore) AssertAtLeast(min int, matchers ...EventInfoMatcher) []EventInfo {
+	ei.tb.Helper()
 	events, err := ei.waitAtLeastNMatch(AllOf(matchers...), min)
 	if err != nil {
 		ei.tb.Fatalf("Timeout waiting for at least %d matches.\nError: %+v", min, errors.WithStack(err))
 	}
+	ei.tb.Logf("Assert passed")
 	return events
 }
 
 // Assert that there are at least min number of matches and at most max number of matches for the provided matchers.
 // This method fails the test if the assert is not fulfilled.
 func (ei *EventInfoStore) AssertInRange(min int, max int, matchers ...EventInfoMatcher) []EventInfo {
+	ei.tb.Helper()
 	events := ei.AssertAtLeast(min, matchers...)
 	if max > 0 && len(events) > max {
 		ei.tb.Fatalf("Assert in range failed: %+v", errors.WithStack(fmt.Errorf("expected <= %d events, saw %d", max, len(events))))
 	}
-
+	ei.tb.Logf("Assert passed")
 	return events
 }
 
 // Assert that there aren't any matches for the provided matchers.
 // This method fails the test if the assert is not fulfilled.
 func (ei *EventInfoStore) AssertNot(matchers ...EventInfoMatcher) []EventInfo {
+	ei.tb.Helper()
 	res, recentEvents, _, err := ei.Find(matchers...)
 	if err != nil {
 		ei.tb.Fatalf("Unexpected error during find on recordevents '%s': %+v", ei.podName, errors.WithStack(err))
@@ -272,14 +212,17 @@ func (ei *EventInfoStore) AssertNot(matchers ...EventInfoMatcher) []EventInfo {
 			fmt.Errorf("Unexpected matches on recordevents '%s', found: %v. %s", ei.podName, res, &recentEvents)),
 		)
 	}
-
+	ei.tb.Logf("Assert passed")
 	return res
 }
 
 // Assert that there are exactly n matches for the provided matchers.
 // This method fails the test if the assert is not fulfilled.
 func (ei *EventInfoStore) AssertExact(n int, matchers ...EventInfoMatcher) []EventInfo {
-	return ei.AssertInRange(n, n, matchers...)
+	ei.tb.Helper()
+	events := ei.AssertInRange(n, n, matchers...)
+	ei.tb.Logf("Assert passed")
+	return events
 }
 
 // Wait a long time (currently 4 minutes) until the provided function matches at least
@@ -290,7 +233,7 @@ func (ei *EventInfoStore) waitAtLeastNMatch(f EventInfoMatcher, min int) ([]Even
 	var matchRet []EventInfo
 	var internalErr error
 
-	wait.PollImmediate(ei.retryInterval, ei.timeout, func() (bool, error) {
+	wait.PollImmediate(retryInterval, retryTimeout, func() (bool, error) {
 		allMatch, sInfo, matchErrs, err := ei.Find(f)
 		if err != nil {
 			internalErr = fmt.Errorf("FAIL MATCHING: unexpected error during find: %v", err)
diff --git a/vendor/knative.dev/eventing/test/lib/recordevents/event_log.go b/vendor/knative.dev/eventing/test/lib/recordevents/event_log.go
new file mode 100644
index 000000000..62aba6113
--- /dev/null
+++ b/vendor/knative.dev/eventing/test/lib/recordevents/event_log.go
@@ -0,0 +1,33 @@
+/*
+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 recordevents
+
+// EventLog is the contract for an event logger to vent an event.
+type EventLog interface {
+	Vent(observed EventInfo) error
+}
+
+type EventLogs []EventLog
+
+func (e EventLogs) Vent(observed EventInfo) error {
+	for _, el := range e {
+		if err := el.Vent(observed); err != nil {
+			return err
+		}
+	}
+	return nil
+}
diff --git a/vendor/knative.dev/eventing/test/lib/recordevents/observer/observer.go b/vendor/knative.dev/eventing/test/lib/recordevents/observer/observer.go
new file mode 100644
index 000000000..e9bd4224a
--- /dev/null
+++ b/vendor/knative.dev/eventing/test/lib/recordevents/observer/observer.go
@@ -0,0 +1,116 @@
+/*
+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 observer
+
+import (
+	"context"
+	"net/http"
+	"sync/atomic"
+	"time"
+
+	cloudeventsbindings "github.com/cloudevents/sdk-go/v2/binding"
+	cloudeventshttp "github.com/cloudevents/sdk-go/v2/protocol/http"
+	"github.com/kelseyhightower/envconfig"
+	"github.com/prometheus/common/log"
+	"knative.dev/pkg/logging"
+
+	"knative.dev/eventing/test/lib/recordevents"
+)
+
+// Observer is the entry point for sinking events into the event log.
+type Observer struct {
+	// Name is the name of this Observer, used to filter if multiple observers.
+	Name string
+	// EventLogs is the list of EventLog implementors to vent observed events.
+	EventLogs recordevents.EventLogs
+
+	seq uint64
+}
+
+// New returns an observer that will vent observations to the list of provided
+// EventLog instances. It will listen on :8080.
+func New(name string, eventLogs ...recordevents.EventLog) *Observer {
+	return &Observer{
+		Name:      name,
+		EventLogs: eventLogs,
+	}
+}
+
+type envConfig struct {
+	// ObserverName is used to identify this instance of the observer.
+	ObserverName string `envconfig:"OBSERVER_NAME" default:"observer-default" required:"true"`
+}
+
+func NewFromEnv(eventLogs ...recordevents.EventLog) *Observer {
+	var env envConfig
+	if err := envconfig.Process("", &env); err != nil {
+		log.Fatal("Failed to process env var", err)
+	}
+
+	return New(env.ObserverName, eventLogs...)
+}
+
+// Start will create the CloudEvents client and start listening for inbound
+// HTTP requests. This is a is a blocking call.
+func (o *Observer) Start(ctx context.Context, handlerFuncs ...func(handler http.Handler) http.Handler) error {
+	var handler http.Handler = o
+
+	for _, dec := range handlerFuncs {
+		handler = dec(handler)
+	}
+
+	server := &http.Server{Addr: ":8080", Handler: handler}
+
+	go func() {
+		if err := server.ListenAndServe(); err != nil {
+			logging.FromContext(ctx).Fatal("Error while starting the HTTP server", err)
+		}
+	}()
+
+	<-ctx.Done()
+
+	logging.FromContext(ctx).Info("Closing the HTTP server")
+
+	return server.Close()
+}
+
+func (o *Observer) ServeHTTP(writer http.ResponseWriter, request *http.Request) {
+	m := cloudeventshttp.NewMessageFromHttpRequest(request)
+	defer m.Finish(nil)
+
+	event, eventErr := cloudeventsbindings.ToEvent(context.TODO(), m)
+	header := request.Header
+
+	eventErrStr := ""
+	if eventErr != nil {
+		eventErrStr = eventErr.Error()
+	}
+	err := o.EventLogs.Vent(recordevents.EventInfo{
+		Error:       eventErrStr,
+		Event:       event,
+		HTTPHeaders: header,
+		Origin:      request.RemoteAddr,
+		Observer:    o.Name,
+		Time:        time.Now(),
+		Sequence:    atomic.AddUint64(&o.seq, 1),
+	})
+	if err != nil {
+		log.Warn("Error while venting the recorded event", err)
+	}
+
+	writer.WriteHeader(http.StatusAccepted)
+}
diff --git a/vendor/knative.dev/eventing/test/lib/recordevents/recorder_vent/constructor.go b/vendor/knative.dev/eventing/test/lib/recordevents/recorder_vent/constructor.go
new file mode 100644
index 000000000..cc73a9671
--- /dev/null
+++ b/vendor/knative.dev/eventing/test/lib/recordevents/recorder_vent/constructor.go
@@ -0,0 +1,108 @@
+/*
+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 recorder_vent
+
+import (
+	"context"
+	"log"
+	"strings"
+
+	"github.com/kelseyhightower/envconfig"
+	"knative.dev/pkg/system"
+
+	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/watch"
+	"k8s.io/client-go/kubernetes/scheme"
+	v1 "k8s.io/client-go/kubernetes/typed/core/v1"
+	"k8s.io/client-go/tools/record"
+
+	kubeclient "knative.dev/pkg/client/injection/kube/client"
+	"knative.dev/pkg/controller"
+	"knative.dev/pkg/logging"
+
+	"knative.dev/eventing/test/lib/recordevents"
+)
+
+type envConfig struct {
+	AgentName string `envconfig:"AGENT_NAME" default:"observer-default" required:"true"`
+	PodName   string `envconfig:"POD_NAME" required:"true"`
+	Port      int    `envconfig:"PORT" default:"8080" required:"true"`
+}
+
+func NewFromEnv(ctx context.Context) recordevents.EventLog {
+	var env envConfig
+	if err := envconfig.Process("", &env); err != nil {
+		log.Fatal("Failed to process env var", err)
+	}
+
+	logging.FromContext(ctx).Infof("Environment configuration: %+v", env)
+
+	return NewEventLog(ctx, env.AgentName, env.PodName)
+}
+
+func NewEventLog(ctx context.Context, agentName string, podName string) recordevents.EventLog {
+	on, err := kubeclient.Get(ctx).CoreV1().Pods(system.Namespace()).Get(ctx, podName, metav1.GetOptions{})
+	if err != nil {
+		logging.FromContext(ctx).Fatal("Error while trying to retrieve the pod", err)
+	}
+
+	logging.FromContext(ctx).Infof("Going to send events to pod '%s' in namespace '%s'", on.Name, on.Namespace)
+
+	return &recorder{out: createRecorder(ctx, agentName), on: on}
+}
+
+func createRecorder(ctx context.Context, agentName string) record.EventRecorder {
+	logger := logging.FromContext(ctx)
+
+	recorder := controller.GetEventRecorder(ctx)
+	if recorder == nil {
+		// Create event broadcaster
+		logger.Debug("Creating event broadcaster")
+		eventBroadcaster := record.NewBroadcasterWithCorrelatorOptions(record.CorrelatorOptions{
+			KeyFunc: func(event *corev1.Event) (aggregateKey string, localKey string) {
+				return strings.Join([]string{
+					event.Source.Component,
+					event.Source.Host,
+					event.InvolvedObject.Kind,
+					event.InvolvedObject.Namespace,
+					event.InvolvedObject.Name,
+					string(event.InvolvedObject.UID),
+					event.InvolvedObject.APIVersion,
+					event.Type,
+					event.Reason,
+				}, ""), string(event.UID)
+			},
+		})
+		watches := []watch.Interface{
+			eventBroadcaster.StartLogging(logger.Named("event-broadcaster").Infof),
+			eventBroadcaster.StartRecordingToSink(
+				&v1.EventSinkImpl{Interface: kubeclient.Get(ctx).CoreV1().Events(system.Namespace())},
+			),
+		}
+		recorder = eventBroadcaster.NewRecorder(scheme.Scheme, corev1.EventSource{Component: agentName})
+		go func() {
+			<-ctx.Done()
+			for _, w := range watches {
+				w.Stop()
+			}
+			logging.FromContext(ctx).Debug("Closed event-broadcaster")
+		}()
+	}
+
+	return recorder
+}
diff --git a/vendor/knative.dev/eventing/test/lib/recordevents/recorder_vent/doc.go b/vendor/knative.dev/eventing/test/lib/recordevents/recorder_vent/doc.go
new file mode 100644
index 000000000..1a4bcc290
--- /dev/null
+++ b/vendor/knative.dev/eventing/test/lib/recordevents/recorder_vent/doc.go
@@ -0,0 +1,19 @@
+/*
+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 recorder_vent implements an recordevents.EventLog backed by Kubernetes
+// Events using an event recorder.
+package recorder_vent
diff --git a/vendor/knative.dev/eventing/test/lib/recordevents/recorder_vent/recorder.go b/vendor/knative.dev/eventing/test/lib/recordevents/recorder_vent/recorder.go
new file mode 100644
index 000000000..9669741f9
--- /dev/null
+++ b/vendor/knative.dev/eventing/test/lib/recordevents/recorder_vent/recorder.go
@@ -0,0 +1,43 @@
+/*
+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 recorder_vent
+
+import (
+	"encoding/json"
+
+	corev1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/client-go/tools/record"
+
+	"knative.dev/eventing/test/lib/recordevents"
+)
+
+type recorder struct {
+	out record.EventRecorder
+	on  runtime.Object
+}
+
+func (r *recorder) Vent(observed recordevents.EventInfo) error {
+	b, err := json.Marshal(observed)
+	if err != nil {
+		return err
+	}
+
+	r.out.Eventf(r.on, corev1.EventTypeNormal, recordevents.CloudEventObservedReason, "%s", string(b))
+
+	return nil
+}
diff --git a/vendor/knative.dev/eventing/test/lib/recordevents/resources.go b/vendor/knative.dev/eventing/test/lib/recordevents/resources.go
new file mode 100644
index 000000000..54726bedf
--- /dev/null
+++ b/vendor/knative.dev/eventing/test/lib/recordevents/resources.go
@@ -0,0 +1,61 @@
+/*
+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 recordevents
+
+import (
+	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/util/uuid"
+	"knative.dev/pkg/test"
+)
+
+// EventRecordPod creates a Pod that stores received events for test retrieval.
+func EventRecordPod(name string, serviceAccountName string) *corev1.Pod {
+	return recordEventsPod("recordevents", name, serviceAccountName)
+}
+
+func recordEventsPod(imageName string, name string, serviceAccountName string) *corev1.Pod {
+	return &corev1.Pod{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:   name,
+			Labels: map[string]string{"e2etest": string(uuid.NewUUID())},
+		},
+		Spec: corev1.PodSpec{
+			Containers: []corev1.Container{{
+				Name:            imageName,
+				Image:           test.ImagePath(imageName),
+				ImagePullPolicy: corev1.PullAlways,
+				Env: []corev1.EnvVar{{
+					Name: "SYSTEM_NAMESPACE",
+					ValueFrom: &corev1.EnvVarSource{
+						FieldRef: &corev1.ObjectFieldSelector{FieldPath: "metadata.namespace"},
+					},
+				}, {
+					Name:  "OBSERVER",
+					Value: "recorder-" + name,
+				}, {
+					Name: "POD_NAME",
+					ValueFrom: &corev1.EnvVarSource{
+						FieldRef: &corev1.ObjectFieldSelector{FieldPath: "metadata.name"},
+					},
+				}},
+			}},
+			ServiceAccountName: serviceAccountName,
+			RestartPolicy:      corev1.RestartPolicyAlways,
+		},
+	}
+}
diff --git a/vendor/knative.dev/eventing/test/lib/resources/kube.go b/vendor/knative.dev/eventing/test/lib/resources/kube.go
index 2b154fe2b..bc415d108 100644
--- a/vendor/knative.dev/eventing/test/lib/resources/kube.go
+++ b/vendor/knative.dev/eventing/test/lib/resources/kube.go
@@ -37,28 +37,6 @@ type PodOption func(*corev1.Pod)
 // Option enables further configuration of a Role.
 type RoleOption func(*rbacv1.Role)
 
-// EventRecordPod creates a Pod that stores received events for test retrieval.
-func EventRecordPod(name string) *corev1.Pod {
-	return eventLoggerPod("recordevents", name)
-}
-
-func eventLoggerPod(imageName string, name string) *corev1.Pod {
-	return &corev1.Pod{
-		ObjectMeta: metav1.ObjectMeta{
-			Name:   name,
-			Labels: map[string]string{"e2etest": string(uuid.NewUUID())},
-		},
-		Spec: corev1.PodSpec{
-			Containers: []corev1.Container{{
-				Name:            imageName,
-				Image:           pkgTest.ImagePath(imageName),
-				ImagePullPolicy: corev1.PullAlways,
-			}},
-			RestartPolicy: corev1.RestartPolicyAlways,
-		},
-	}
-}
-
 // EventTransformationPod creates a Pod that transforms events received receiving as arg a cloudevents sdk2 Event
 func EventTransformationPod(name string, newEventType string, newEventSource string, newEventData []byte) *corev1.Pod {
 	const imageName = "transformevents"
diff --git a/vendor/knative.dev/eventing/test/markdown-lint-config.rc b/vendor/knative.dev/eventing/test/markdown-lint-config.rc
index 1d10e7d06..60fac0f65 100644
--- a/vendor/knative.dev/eventing/test/markdown-lint-config.rc
+++ b/vendor/knative.dev/eventing/test/markdown-lint-config.rc
@@ -1,5 +1,5 @@
 # For help, see
-# https://github.com/markdownlint/markdownlint/blob/master/docs/configuration.md 
+# https://github.com/markdownlint/markdownlint/blob/master/docs/configuration.md
 
 # Ignoring some of the available rules. For more, see
 # https://github.com/markdownlint/markdownlint/blob/master/docs/RULES.md
diff --git a/vendor/knative.dev/eventing/test/performance/infra/sender/http_load_generator.go b/vendor/knative.dev/eventing/test/performance/infra/sender/http_load_generator.go
index c45999c68..7b68ed0cc 100644
--- a/vendor/knative.dev/eventing/test/performance/infra/sender/http_load_generator.go
+++ b/vendor/knative.dev/eventing/test/performance/infra/sender/http_load_generator.go
@@ -107,7 +107,7 @@ func (cet CloudEventsTargeter) VegetaTargeter() vegeta.Targeter {
 	}
 }
 
-type HttpLoadGenerator struct {
+type httpLoadGenerator struct {
 	eventSource string
 	sinkUrl     string
 
@@ -119,13 +119,13 @@ type HttpLoadGenerator struct {
 	ceClient       cloudevents.Client
 }
 
-func NewHttpLoadGeneratorFactory(sinkUrl string, minWorkers uint64) LoadGeneratorFactory {
+func NewHTTPLoadGeneratorFactory(sinkUrl string, minWorkers uint64) LoadGeneratorFactory {
 	return func(eventSource string, sentCh chan common.EventTimestamp, acceptedCh chan common.EventTimestamp) (generator LoadGenerator, e error) {
 		if sinkUrl == "" {
 			panic("Missing --sink flag")
 		}
 
-		loadGen := &HttpLoadGenerator{
+		loadGen := &httpLoadGenerator{
 			eventSource: eventSource,
 			sinkUrl:     sinkUrl,
 
@@ -190,27 +190,27 @@ func newCloudEventsClient(sinkUrl string) (cloudevents.Client, error) {
 		cloudevents.WithTarget(sinkUrl),
 	)
 	if err != nil {
-		return nil, fmt.Errorf("failed to create transport: %v", err)
+		return nil, fmt.Errorf("failed to create transport: %w", err)
 	}
 
 	return cloudevents.NewClient(t)
 }
 
-func (h HttpLoadGenerator) Warmup(pace common.PaceSpec, msgSize uint, fixedBody bool) {
+func (h httpLoadGenerator) Warmup(pace common.PaceSpec, msgSize uint, fixedBody bool) {
 	targeter := NewCloudEventsTargeter(h.sinkUrl, msgSize, common.WarmupEventType, defaultEventSource, fixedBody).VegetaTargeter()
 	vegetaResults := h.warmupAttacker.Attack(targeter, vegeta.ConstantPacer{Freq: pace.Rps, Per: time.Second}, pace.Duration, common.WarmupEventType+"-attack")
 	for range vegetaResults {
 	}
 }
 
-func (h HttpLoadGenerator) RunPace(i int, pace common.PaceSpec, msgSize uint, fixedBody bool) {
+func (h httpLoadGenerator) RunPace(i int, pace common.PaceSpec, msgSize uint, fixedBody bool) {
 	targeter := NewCloudEventsTargeter(h.sinkUrl, msgSize, common.MeasureEventType, eventsSource(), fixedBody).VegetaTargeter()
 	res := h.paceAttacker.Attack(targeter, vegeta.ConstantPacer{Freq: pace.Rps, Per: time.Second}, pace.Duration, fmt.Sprintf("%s-attack-%d", h.eventSource, i))
 	for range res {
 	}
 }
 
-func (h HttpLoadGenerator) SendGCEvent() {
+func (h httpLoadGenerator) SendGCEvent() {
 	event := cloudevents.NewEvent(cloudevents.VersionV1)
 	event.SetID(uuid.New().String())
 	event.SetDataContentType(cloudevents.ApplicationJSON)
@@ -220,7 +220,7 @@ func (h HttpLoadGenerator) SendGCEvent() {
 	_ = h.ceClient.Send(context.TODO(), event)
 }
 
-func (h HttpLoadGenerator) SendEndEvent() {
+func (h httpLoadGenerator) SendEndEvent() {
 	event := cloudevents.NewEvent(cloudevents.VersionV1)
 	event.SetID(uuid.New().String())
 	event.SetDataContentType(cloudevents.ApplicationJSON)
diff --git a/vendor/knative.dev/eventing/test/test_images/performance/main.go b/vendor/knative.dev/eventing/test/test_images/performance/main.go
index b1f1312a6..8ef09d27b 100644
--- a/vendor/knative.dev/eventing/test/test_images/performance/main.go
+++ b/vendor/knative.dev/eventing/test/test_images/performance/main.go
@@ -38,5 +38,5 @@ func init() {
 func main() {
 	flag.Parse()
 
-	infra.StartPerformanceImage(sender.NewHttpLoadGeneratorFactory(sinkURL, minWorkers), receiver.EventTypeExtractor, receiver.EventIdExtractor)
+	infra.StartPerformanceImage(sender.NewHTTPLoadGeneratorFactory(sinkURL, minWorkers), receiver.EventTypeExtractor, receiver.EventIdExtractor)
 }
diff --git a/vendor/knative.dev/eventing/test/test_images/performance/mako.config b/vendor/knative.dev/eventing/test/test_images/performance/mako.config
index e106ef5f1..772630692 100644
--- a/vendor/knative.dev/eventing/test/test_images/performance/mako.config
+++ b/vendor/knative.dev/eventing/test/test_images/performance/mako.config
@@ -2,7 +2,7 @@ project_name: "Knative"
 benchmark_name: "Generic Eventing Benchmark"
 description: "Measure latency and throughput of an eventing component."
 
-### ATTENTION: This is a dummy benchmark config. Do not attempt to create or
+### ATTENTION: This is a sample benchmark config. Do not attempt to create or
 # update this benchmark with the mako tool. See
 test/performance/sample-dev.config for an example of a real benchmark config.
 
diff --git a/vendor/knative.dev/eventing/test/test_images/recordevents/eventstore.go b/vendor/knative.dev/eventing/test/test_images/recordevents/eventstore.go
deleted file mode 100644
index 470e5adc2..000000000
--- a/vendor/knative.dev/eventing/test/test_images/recordevents/eventstore.go
+++ /dev/null
@@ -1,191 +0,0 @@
-/*
-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
-
-	https://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 main
-
-import (
-	"encoding/json"
-	"fmt"
-	"sync"
-
-	cloudevents "github.com/cloudevents/sdk-go/v2"
-
-	"knative.dev/eventing/test/lib/recordevents"
-)
-
-// Number of EventInfo per block
-const evBlockSize = 100
-
-// Block of stored EventInfo
-type eventBlock struct {
-	// seqno of [0] evInfoBytes entry
-	firstIndex int
-	// offset inside block for newly appended event
-	firstOffsetFree int
-	// offset inside block of first non-trimmed event
-	firstValid int
-	// serialized EventInfo structures for each seqno.  We enforce
-	// that there is always at least one block.
-	evInfoBytes [evBlockSize][]byte
-}
-
-// All events currently seen
-type eventStore struct {
-	// Blocks of events in increasing sequency number order
-	evBlocks     []*eventBlock
-	evBlocksLock sync.Mutex
-}
-
-// Create a new event store.
-func newEventStore() *eventStore {
-	es := &eventStore{}
-	es.evBlocks = []*eventBlock{{}}
-
-	// One block with no entries starting at sequence number 1
-	es.evBlocks[0].firstIndex = 1
-	es.evBlocks[0].firstOffsetFree = 0
-	es.evBlocks[0].firstValid = 0
-	return es
-}
-
-// See if there's enough room to append to the current last block.  If not,
-// append an extra block.
-func (es *eventStore) checkAppendBlock() {
-	if es.evBlocks[len(es.evBlocks)-1].firstOffsetFree == evBlockSize {
-		newEVBlock := &eventBlock{
-			firstIndex: es.evBlocks[len(es.evBlocks)-1].firstIndex + evBlockSize,
-		}
-		es.evBlocks = append(es.evBlocks, newEVBlock)
-	}
-}
-
-// Store the specified event.
-func (es *eventStore) StoreEvent(event *cloudevents.Event, evErr error, httpHeaders map[string][]string) {
-	var evInfo recordevents.EventInfo
-	var err error
-	var evInfoBytes []byte
-	if evErr != nil {
-		evInfo.HTTPHeaders = httpHeaders
-		evInfo.Error = evErr.Error()
-		if evInfo.Error == "" {
-			evInfo.Error = "Unknown Incoming Error"
-		}
-		evInfoBytes, err = json.Marshal(&evInfo)
-		if err != nil {
-			panic(fmt.Errorf("unexpected marshal error (%v) (%+v)", err, evInfo))
-		}
-	} else {
-		evInfo.Event = event
-		evInfo.HTTPHeaders = httpHeaders
-		evInfoBytes, err = json.Marshal(&evInfo)
-
-		if err != nil {
-			evInfo.Event = nil
-			evInfo.Error = err.Error()
-			if evInfo.Error == "" {
-				evInfo.Error = "Unknown Error"
-			}
-			evInfoBytes, err = json.Marshal(&evInfo)
-			if err != nil {
-				panic(fmt.Errorf("unexpected marshal error (%v) (%+v)", err, evInfo))
-			}
-		}
-	}
-
-	es.evBlocksLock.Lock()
-	// Add a new block if we're out of space
-	es.checkAppendBlock()
-
-	evBlock := es.evBlocks[len(es.evBlocks)-1]
-	if evBlock.firstOffsetFree < evBlockSize {
-		evBlock.evInfoBytes[evBlock.firstOffsetFree] = evInfoBytes
-		evBlock.firstOffsetFree++
-	}
-
-	es.evBlocksLock.Unlock()
-}
-
-// Logically trim all events up to and include the provided
-// sequence number.  Returns error for patently incorrect
-// values (negative sequence number or sequence number larger
-// than the largest event seen).  Trimming already trimmed
-// regions is legal.
-func (es *eventStore) TrimThrough(through int) error {
-	es.evBlocksLock.Lock()
-	defer es.evBlocksLock.Unlock()
-	minAvail, maxSeen := es.minMaxUnlocked()
-	if through > maxSeen {
-		return fmt.Errorf("invalid trim through %d, maxSeen %d", through, maxSeen)
-	} else if through < 0 {
-		return fmt.Errorf("invalid trim less than zero %d", through)
-	} else if through < minAvail {
-		return nil
-	}
-	// Completely remove blocks if they are full and all events in them are less than
-	// the specified value.
-	for len(es.evBlocks) > 1 && (es.evBlocks[0].firstIndex+evBlockSize-1) <= through {
-		es.evBlocks = es.evBlocks[1:]
-	}
-	// Logically trim the block split by through.
-	es.evBlocks[0].firstValid = through - es.evBlocks[0].firstIndex + 1
-	return nil
-}
-
-// return min/max untrimmed value when already holding the lock
-func (es *eventStore) minMaxUnlocked() (minAvail int, maxSeen int) {
-	minBlock := es.evBlocks[0]
-	minAvail = minBlock.firstIndex + (minBlock.firstValid)
-
-	maxBlock := es.evBlocks[len(es.evBlocks)-1]
-	maxSeen = maxBlock.firstIndex + maxBlock.firstOffsetFree - 1
-	return minAvail, maxSeen
-}
-
-// Returns min available value and max seen value for the store.
-// min is the minimum value that can be retrieved via GetEntry, or
-// if no values can be retrieved, min == max+1.  Max starts at 0
-// when no events have been seen.
-func (es *eventStore) MinMax() (minAvail int, maxSeen int) {
-	es.evBlocksLock.Lock()
-	minAvail, maxSeen = es.minMaxUnlocked()
-
-	es.evBlocksLock.Unlock()
-	return minAvail, maxSeen
-}
-
-// Get the already serialized EventInfo structure for the provided sequence
-// number.
-func (es *eventStore) GetEventInfoBytes(seq int) ([]byte, error) {
-	var evInfoBytes []byte
-	found := false
-
-	es.evBlocksLock.Lock()
-	for _, block := range es.evBlocks {
-		if seq < block.firstIndex+block.firstValid {
-			break
-		}
-		if seq < block.firstIndex+block.firstOffsetFree {
-			found = true
-			evInfoBytes = block.evInfoBytes[seq-block.firstIndex]
-			break
-		}
-	}
-	es.evBlocksLock.Unlock()
-	if !found {
-		return evInfoBytes, fmt.Errorf("Invalid sequence number %d", seq)
-	}
-	return evInfoBytes, nil
-}
diff --git a/vendor/knative.dev/eventing/test/test_images/recordevents/main.go b/vendor/knative.dev/eventing/test/test_images/recordevents/main.go
index 8dd741163..51fde078e 100644
--- a/vendor/knative.dev/eventing/test/test_images/recordevents/main.go
+++ b/vendor/knative.dev/eventing/test/test_images/recordevents/main.go
@@ -17,164 +17,60 @@ limitations under the License.
 package main
 
 import (
-	"context"
-	"encoding/json"
-	"fmt"
 	"log"
 	"net/http"
 	"os"
-	"strconv"
-	"strings"
 
-	cloudeventsbindings "github.com/cloudevents/sdk-go/v2/binding"
-	cloudeventshttp "github.com/cloudevents/sdk-go/v2/protocol/http"
-	"go.uber.org/zap"
+	"k8s.io/client-go/rest"
+	"knative.dev/pkg/injection/sharedmain"
+	"knative.dev/pkg/logging"
+	_ "knative.dev/pkg/system/testing"
 
 	"knative.dev/eventing/pkg/kncloudevents"
-	testlib "knative.dev/eventing/test/lib"
 	"knative.dev/eventing/test/lib/dropevents"
-	"knative.dev/eventing/test/lib/recordevents"
+	"knative.dev/eventing/test/lib/recordevents/observer"
+	"knative.dev/eventing/test/lib/recordevents/recorder_vent"
 	"knative.dev/eventing/test/test_images"
 )
 
-type eventRecorder struct {
-	es *eventStore
-}
-
-func newEventRecorder() *eventRecorder {
-	return &eventRecorder{es: newEventStore()}
-}
-
-// Start the recordevents REST api server
-func (er *eventRecorder) StartServer(port int) {
-	http.HandleFunc(recordevents.GetMinMaxPath, er.handleMinMax)
-	http.HandleFunc(recordevents.GetEntryPath, er.handleGetEntry)
-	http.HandleFunc(recordevents.TrimThroughPath, er.handleTrim)
-	go http.ListenAndServe(fmt.Sprintf(":%d", port), nil)
-}
-
-// HTTP handler for GetMinMax requests
-func (er *eventRecorder) handleMinMax(w http.ResponseWriter, r *http.Request) {
-	minAvail, maxSeen := er.es.MinMax()
-	minMax := recordevents.MinMaxResponse{
-		MinAvail: minAvail,
-		MaxSeen:  maxSeen,
-	}
-	respBytes, err := json.Marshal(minMax)
-	if err != nil {
-		log.Panicf("Internal error: json marshal shouldn't fail: (%v) (%+v)", err, minMax)
-	}
-
-	w.Header().Set("Content-Type", "text/json")
-	w.WriteHeader(http.StatusOK)
-	w.Write(respBytes)
-}
-
-// HTTP handler for TrimThrough requests
-func (er *eventRecorder) handleTrim(w http.ResponseWriter, r *http.Request) {
-	// If we extend this much at all we should vendor a better mux(gorilla, etc)
-	path := strings.TrimLeft(r.URL.Path, "/")
-	getPrefix := strings.TrimLeft(recordevents.TrimThroughPath, "/")
-	suffix := strings.TrimLeft(strings.TrimPrefix(path, getPrefix), "/")
-
-	seqNum, err := strconv.ParseInt(suffix, 10, 32)
-	if err != nil {
-		http.Error(w, "Can't parse event sequence number in request", http.StatusBadRequest)
-		return
-	}
-
-	err = er.es.TrimThrough(int(seqNum))
-	if err != nil {
-		http.Error(w, "Invalid sequence number in request to trim", http.StatusNotFound)
-		return
-	}
-
-	w.Header().Set("Content-Type", "text/json")
-	w.WriteHeader(http.StatusOK)
-}
-
-// HTTP handler for GetEntry requests
-func (er *eventRecorder) handleGetEntry(w http.ResponseWriter, r *http.Request) {
-	// If we extend this much at all we should vendor a better mux(gorilla, etc)
-	path := strings.TrimLeft(r.URL.Path, "/")
-	getPrefix := strings.TrimLeft(recordevents.GetEntryPath, "/")
-	suffix := strings.TrimLeft(strings.TrimPrefix(path, getPrefix), "/")
-
-	seqNum, err := strconv.ParseInt(suffix, 10, 32)
+func main() {
+	cfg, err := rest.InClusterConfig()
 	if err != nil {
-		http.Error(w, "Can't parse event sequence number in request", http.StatusBadRequest)
-		return
+		log.Fatal("Error while reading the cfg", err)
 	}
+	//nolint // nil ctx is fine here, look at the code of EnableInjectionOrDie
+	ctx := sharedmain.EnableInjectionOrDie(nil, cfg)
 
-	entryBytes, err := er.es.GetEventInfoBytes(int(seqNum))
-	if err != nil {
-		http.Error(w, "Couldn't find requested event", http.StatusNotFound)
-		return
+	if err := test_images.ConfigureTracing(logging.FromContext(ctx), ""); err != nil {
+		logging.FromContext(ctx).Fatal("Unable to setup trace publishing", err)
 	}
 
-	w.Header().Set("Content-Type", "text/json")
-	w.WriteHeader(http.StatusOK)
-	w.Write(entryBytes)
-}
-
-func (er *eventRecorder) ServeHTTP(writer http.ResponseWriter, request *http.Request) {
-	m := cloudeventshttp.NewMessageFromHttpRequest(request)
-	defer m.Finish(nil)
-
-	event, eventErr := cloudeventsbindings.ToEvent(context.TODO(), m)
-	header := request.Header
-
-	er.es.StoreEvent(event, eventErr, map[string][]string(header))
-
-	headerNameList := testlib.InterestingHeaders()
-	for _, headerName := range headerNameList {
-		if headerValue := header.Get(headerName); headerValue != "" {
-			log.Printf("Header %s: %s\n", headerName, headerValue)
-		}
-	}
-
-	if eventErr != nil {
-		log.Printf("error receiving the event: %v", eventErr)
-	} else {
-		valErr := event.Validate()
-		if valErr == nil {
-			log.Printf("eventdetails:\n%s", event.String())
-		} else {
-			log.Printf("error validating the event: %v", valErr)
-		}
-	}
-
-	writer.WriteHeader(http.StatusAccepted)
-}
-
-func main() {
-	er := newEventRecorder()
-	er.StartServer(recordevents.RecordEventsPort)
-
-	logger, _ := zap.NewDevelopment()
-	if err := test_images.ConfigureTracing(logger.Sugar(), ""); err != nil {
-		log.Fatalf("Unable to setup trace publishing: %v", err)
-	}
+	obs := observer.NewFromEnv(
+		recorder_vent.NewFromEnv(ctx),
+	)
 
 	algorithm, ok := os.LookupEnv(dropevents.SkipAlgorithmKey)
-	handler := kncloudevents.CreateHandler(er)
 	if ok {
 		skipper := dropevents.SkipperAlgorithm(algorithm)
 		counter := dropevents.CounterHandler{
 			Skipper: skipper,
 		}
-		next := handler
-		handler = http.HandlerFunc(func(writer http.ResponseWriter, request *http.Request) {
-			if counter.Skip() {
-				writer.WriteHeader(http.StatusConflict)
-				return
-			}
-			next.ServeHTTP(writer, request)
+		err = obs.Start(ctx, kncloudevents.CreateHandler, func(handler http.Handler) http.Handler {
+			return http.HandlerFunc(func(writer http.ResponseWriter, request *http.Request) {
+				if counter.Skip() {
+					writer.WriteHeader(http.StatusConflict)
+					return
+				}
+				handler.ServeHTTP(writer, request)
+			})
 		})
+	} else {
+		err = obs.Start(ctx, kncloudevents.CreateHandler)
 	}
 
-	err := http.ListenAndServe(":8080", handler)
 	if err != nil {
-		panic(err)
+		logging.FromContext(ctx).Fatal("Error during start", err)
 	}
+
+	logging.FromContext(ctx).Info("Closing the recordevents process")
 }
diff --git a/vendor/knative.dev/pkg/test/gcs/gcs.go b/vendor/knative.dev/pkg/test/gcs/gcs.go
index 13ae206fa..f1b9a695a 100644
--- a/vendor/knative.dev/pkg/test/gcs/gcs.go
+++ b/vendor/knative.dev/pkg/test/gcs/gcs.go
@@ -31,6 +31,7 @@ import (
 	"google.golang.org/api/option"
 )
 
+//nolint — there's also Client so they collide.
 type GCSClient struct {
 	*storage.Client
 }
diff --git a/vendor/knative.dev/pkg/test/helpers/name.go b/vendor/knative.dev/pkg/test/helpers/name.go
index 9cb0d529f..ef5ac4dbd 100644
--- a/vendor/knative.dev/pkg/test/helpers/name.go
+++ b/vendor/knative.dev/pkg/test/helpers/name.go
@@ -71,8 +71,8 @@ func RandomString() string {
 }
 
 // For the same prefix more specific should come first.
-// Note: we expect GRPC.
-var knownNames = []string{"HTTPS", "HTTP2", "HTTP", "GRPC", "TLS", "WS", "H2C"}
+// Note: we expect GRPC vs gRPC.
+var knownNames = []string{"GRPC", "H2C", "HTTPS", "HTTP2", "HTTP", "REST", "TLS", "WS"}
 
 // MakeK8sNamePrefix converts each chunk of non-alphanumeric character into a single dash
 // and also convert camelcase tokens into dash-delimited lowercase tokens.
diff --git a/vendor/knative.dev/pkg/test/logging/error.go b/vendor/knative.dev/pkg/test/logging/error.go
index e34916075..774a60612 100644
--- a/vendor/knative.dev/pkg/test/logging/error.go
+++ b/vendor/knative.dev/pkg/test/logging/error.go
@@ -77,7 +77,7 @@ func (e *structuredError) EnableValuePrinting() {
 
 // Create a StructuredError. Gives a little better logging when given to a TLogger.
 // This may prove to not be useful if users use the logger's WithValues() better.
-func Error(msg string, keysAndValues ...interface{}) *structuredError {
+func Error(msg string, keysAndValues ...interface{}) error {
 	return &structuredError{msg, keysAndValues, true}
 }
 
diff --git a/vendor/knative.dev/pkg/test/logging/tlogger.go b/vendor/knative.dev/pkg/test/logging/tlogger.go
index 6f58f3014..42891bc21 100644
--- a/vendor/knative.dev/pkg/test/logging/tlogger.go
+++ b/vendor/knative.dev/pkg/test/logging/tlogger.go
@@ -137,7 +137,7 @@ func validateKeysAndValues(keysAndValues ...interface{}) bool {
 		_, isField := keysAndValues[i].(zapcore.Field)
 		_, isString := keysAndValues[i].(string)
 		if isField {
-			i += 1
+			i++
 		} else if isString {
 			if i == length-1 {
 				return false
@@ -160,8 +160,7 @@ func (o *TLogger) interfacesToFields(things ...interface{}) []interface{} {
 	return fields
 }
 
-// nolint:stylecheck // Returning the error first is okay and expected here.
-func (o *TLogger) errorWithRuntimeCheck(stringThenKeysAndValues ...interface{}) (error, string, []interface{}) {
+func (o *TLogger) errorWithRuntimeCheck(stringThenKeysAndValues ...interface{}) (error, string, []interface{}) { //nolint Returning the error first is okay and expected here.
 	if len(stringThenKeysAndValues) == 0 {
 		return nil, "", nil
 	}
diff --git a/vendor/knative.dev/pkg/test/mako/analyzer.go b/vendor/knative.dev/pkg/test/mako/analyzer.go
index d581543b7..b945aa4e5 100644
--- a/vendor/knative.dev/pkg/test/mako/analyzer.go
+++ b/vendor/knative.dev/pkg/test/mako/analyzer.go
@@ -17,9 +17,9 @@ limitations under the License.
 package mako
 
 import (
-	"github.com/golang/protobuf/proto"
 	tpb "github.com/google/mako/clients/proto/analyzers/threshold_analyzer_go_proto"
 	mpb "github.com/google/mako/spec/proto/mako_go_proto"
+	"google.golang.org/protobuf/proto"
 )
 
 // NewCrossRunConfig returns a config that can be used in ThresholdAnalyzer.
diff --git a/vendor/knative.dev/pkg/test/mako/config/benchmark.go b/vendor/knative.dev/pkg/test/mako/config/benchmark.go
index 41d046865..8f22a865c 100644
--- a/vendor/knative.dev/pkg/test/mako/config/benchmark.go
+++ b/vendor/knative.dev/pkg/test/mako/config/benchmark.go
@@ -23,7 +23,7 @@ import (
 	"os"
 	"path/filepath"
 
-	"github.com/golang/protobuf/proto"
+	"github.com/golang/protobuf/proto" //nolint apis incompatible with the new version
 	mpb "github.com/google/mako/spec/proto/mako_go_proto"
 )
 
diff --git a/vendor/knative.dev/test-infra/scripts/release.sh b/vendor/knative.dev/test-infra/scripts/release.sh
index a10b842f3..bd05ed8bc 100644
--- a/vendor/knative.dev/test-infra/scripts/release.sh
+++ b/vendor/knative.dev/test-infra/scripts/release.sh
@@ -94,7 +94,7 @@ RELEASE_NOTES=""
 RELEASE_BRANCH=""
 RELEASE_GCS_BUCKET="knative-nightly/${REPO_NAME}"
 RELEASE_DIR=""
-KO_FLAGS="-P"
+KO_FLAGS="-P --platform=all"
 VALIDATION_TESTS="./test/presubmit-tests.sh"
 ARTIFACTS_TO_PUBLISH=""
 FROM_NIGHTLY_RELEASE=""