Repository: knative/pkg
Pull request #1793 (2020-10-09 20:13:55)

logstream v2
`logstream` in it's current form reaches out for `knative.dev/pkg/test` globals

knative-prow-robot: Skipping CI for Draft Pull Request.
If you want CI signal for your change, please convert it to an actual PR.
You can still manually trigger a test run with `/test all`
knative-prow-robot: [APPROVALNOTIFIER] This PR is **APPROVED**

This pull-request has been approved by: *<a href="https://github.com/knative/pkg/pull/1793#" title="Author self-approved">dprotaso</a>*

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

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/pkg/blob/master/OWNERS)~~ [dprotaso]

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":[]} -->
dprotaso: cc @mattmoor 
dprotaso: > Do you really need separate v2 dir?

Updated PR description 
dprotaso: racey
dprotaso: Example output is here:
https://github.com/knative/serving/pull/9761/checks?check_run_id=1233417401
codecov[bot]: # [Codecov](https://codecov.io/gh/knative/pkg/pull/1793?src=pr&el=h1) Report
> Merging [#1793](https://codecov.io/gh/knative/pkg/pull/1793?src=pr&el=desc) into [master](https://codecov.io/gh/knative/pkg/commit/3eb7d13daebef96e36d8f4beeb492d1514c0c3ff?el=desc) will **decrease** coverage by `0.20%`.
> The diff coverage is `61.70%`.

[![Impacted file tree graph](https://codecov.io/gh/knative/pkg/pull/1793/graphs/tree.svg?width=650&height=150&src=pr&token=rlqpDnfxbO)](https://codecov.io/gh/knative/pkg/pull/1793?src=pr&el=tree)

```diff
@@            Coverage Diff             @@
##           master    #1793      +/-   ##
==========================================
- Coverage   68.20%   68.00%   -0.21%     
==========================================
  Files         220      221       +1     
  Lines        9320     9408      +88     
==========================================
+ Hits         6357     6398      +41     
- Misses       2646     2688      +42     
- Partials      317      322       +5     
```


| [Impacted Files](https://codecov.io/gh/knative/pkg/pull/1793?src=pr&el=tree) | Coverage Δ | |
|---|---|---|
| [test/logstream/v2/stream.go](https://codecov.io/gh/knative/pkg/pull/1793/diff?src=pr&el=tree#diff-dGVzdC9sb2dzdHJlYW0vdjIvc3RyZWFtLmdv) | `48.35% <60.86%> (ø)` | |
| [hash/bucketer.go](https://codecov.io/gh/knative/pkg/pull/1793/diff?src=pr&el=tree#diff-aGFzaC9idWNrZXRlci5nbw==) | `100.00% <100.00%> (ø)` | |

------

[Continue to review full report at Codecov](https://codecov.io/gh/knative/pkg/pull/1793?src=pr&el=continue).
> **Legend** - [Click here to learn more](https://docs.codecov.io/docs/codecov-delta)
> `Δ = absolute <relative> (impact)`, `ø = not affected`, `? = missing data`
> Powered by [Codecov](https://codecov.io/gh/knative/pkg/pull/1793?src=pr&el=footer). Last update [3c4df8c...4368e3a](https://codecov.io/gh/knative/pkg/pull/1793?src=pr&el=lastupdated). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments).

dprotaso: here's a serving run without induced failures

https://github.com/knative/serving/pull/9761

Patch:
diff --git a/go.mod b/go.mod
index fd51f8307..544cc89a3 100644
--- a/go.mod
+++ b/go.mod
@@ -55,7 +55,7 @@ require (
 	k8s.io/code-generator v0.18.8
 	k8s.io/gengo v0.0.0-20200205140755-e0e292d8aa12
 	k8s.io/klog v1.0.0
-	knative.dev/test-infra v0.0.0-20201009190921-622088de6f36
+	knative.dev/test-infra v0.0.0-20201009204121-322fb08edae7
 	sigs.k8s.io/boskos v0.0.0-20200729174948-794df80db9c9
 )
 
diff --git a/go.sum b/go.sum
index 1eb9045da..7ab87f393 100644
--- a/go.sum
+++ b/go.sum
@@ -1851,8 +1851,8 @@ knative.dev/test-infra v0.0.0-20200505052144-5ea2f705bb55/go.mod h1:WqF1Azka+FxP
 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-20201009190921-622088de6f36 h1:iOk31chJYXMs6l6S7CAyviVlzrFOBSmAC2OdiYOL73w=
-knative.dev/test-infra v0.0.0-20201009190921-622088de6f36/go.mod h1:W4EfffNKnIkYuBHdOjw0UP3YtFbK4ImwDiEoIAbG1zk=
+knative.dev/test-infra v0.0.0-20201009204121-322fb08edae7 h1:2peeccxecZH3kBH8g9U+We3kjstgX8LLloO//ndrxnQ=
+knative.dev/test-infra v0.0.0-20201009204121-322fb08edae7/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/vendor/github.com/google/mako/go/quickstore/BUILD b/vendor/github.com/google/mako/go/quickstore/BUILD
deleted file mode 100644
index 4623b68ea..000000000
--- a/vendor/github.com/google/mako/go/quickstore/BUILD
+++ /dev/null
@@ -1,24 +0,0 @@
-# Quickstore library for Go.
-# See quickstore.go and go/mako-quickstore for documentation.
-# See quickstore_example_test.go for example usage.
-package(default_visibility = ["//visibility:public"])
-
-licenses(["notice"])
-
-load("@io_bazel_rules_go//go:def.bzl", "go_library", "go_test")
-
-go_library(
-    name = "quickstore",
-    srcs = ["quickstore.go"],
-    importpath = "github.com/google/mako/go/quickstore",
-    deps = [
-        "//internal/go/common",
-        "//internal/quickstore_microservice/proto:quickstore_go_grpc_pb",
-        "//internal/quickstore_microservice/proto:quickstore_go_proto",
-        "//proto/quickstore:quickstore_go_proto",
-        "//spec/proto:mako_go_proto",
-        "@com_github_golang_glog//:go_default_library",
-        "@com_github_golang_protobuf//proto:go_default_library",
-        "@org_golang_google_grpc//:go_default_library",
-    ],
-)
diff --git a/vendor/github.com/google/mako/internal/go/common/BUILD b/vendor/github.com/google/mako/internal/go/common/BUILD
deleted file mode 100644
index 1f90d9eda..000000000
--- a/vendor/github.com/google/mako/internal/go/common/BUILD
+++ /dev/null
@@ -1,14 +0,0 @@
-# Go libraries for mako
-package(default_visibility = ["//:internal"])
-
-licenses(["notice"])
-
-load("@io_bazel_rules_go//go:def.bzl", "go_library")
-
-go_library(
-    name = "common",
-    srcs = [
-        "common_deps.go",
-    ],
-    importpath = "github.com/google/mako/internal/go/common",
-)
diff --git a/vendor/github.com/grpc-ecosystem/grpc-gateway/internal/BUILD.bazel b/vendor/github.com/grpc-ecosystem/grpc-gateway/internal/BUILD.bazel
deleted file mode 100644
index 5242751fb..000000000
--- a/vendor/github.com/grpc-ecosystem/grpc-gateway/internal/BUILD.bazel
+++ /dev/null
@@ -1,23 +0,0 @@
-load("@rules_proto//proto:defs.bzl", "proto_library")
-load("@io_bazel_rules_go//go:def.bzl", "go_library")
-load("@io_bazel_rules_go//proto:def.bzl", "go_proto_library")
-
-package(default_visibility = ["//visibility:public"])
-
-proto_library(
-    name = "internal_proto",
-    srcs = ["errors.proto"],
-    deps = ["@com_google_protobuf//:any_proto"],
-)
-
-go_proto_library(
-    name = "internal_go_proto",
-    importpath = "github.com/grpc-ecosystem/grpc-gateway/internal",
-    proto = ":internal_proto",
-)
-
-go_library(
-    name = "go_default_library",
-    embed = [":internal_go_proto"],
-    importpath = "github.com/grpc-ecosystem/grpc-gateway/internal",
-)
diff --git a/vendor/github.com/grpc-ecosystem/grpc-gateway/runtime/BUILD.bazel b/vendor/github.com/grpc-ecosystem/grpc-gateway/runtime/BUILD.bazel
deleted file mode 100644
index 58b72b9cf..000000000
--- a/vendor/github.com/grpc-ecosystem/grpc-gateway/runtime/BUILD.bazel
+++ /dev/null
@@ -1,85 +0,0 @@
-load("@io_bazel_rules_go//go:def.bzl", "go_library", "go_test")
-
-package(default_visibility = ["//visibility:public"])
-
-go_library(
-    name = "go_default_library",
-    srcs = [
-        "context.go",
-        "convert.go",
-        "doc.go",
-        "errors.go",
-        "fieldmask.go",
-        "handler.go",
-        "marshal_httpbodyproto.go",
-        "marshal_json.go",
-        "marshal_jsonpb.go",
-        "marshal_proto.go",
-        "marshaler.go",
-        "marshaler_registry.go",
-        "mux.go",
-        "pattern.go",
-        "proto2_convert.go",
-        "proto_errors.go",
-        "query.go",
-    ],
-    importpath = "github.com/grpc-ecosystem/grpc-gateway/runtime",
-    deps = [
-        "//internal:go_default_library",
-        "//utilities:go_default_library",
-        "@com_github_golang_protobuf//descriptor:go_default_library_gen",
-        "@com_github_golang_protobuf//jsonpb:go_default_library_gen",
-        "@com_github_golang_protobuf//proto:go_default_library",
-        "@go_googleapis//google/api:httpbody_go_proto",
-        "@io_bazel_rules_go//proto/wkt:any_go_proto",
-        "@io_bazel_rules_go//proto/wkt:descriptor_go_proto",
-        "@io_bazel_rules_go//proto/wkt:duration_go_proto",
-        "@io_bazel_rules_go//proto/wkt:field_mask_go_proto",
-        "@io_bazel_rules_go//proto/wkt:timestamp_go_proto",
-        "@io_bazel_rules_go//proto/wkt:wrappers_go_proto",
-        "@org_golang_google_grpc//codes:go_default_library",
-        "@org_golang_google_grpc//grpclog:go_default_library",
-        "@org_golang_google_grpc//metadata:go_default_library",
-        "@org_golang_google_grpc//status:go_default_library",
-    ],
-)
-
-go_test(
-    name = "go_default_test",
-    size = "small",
-    srcs = [
-        "context_test.go",
-        "convert_test.go",
-        "errors_test.go",
-        "fieldmask_test.go",
-        "handler_test.go",
-        "marshal_httpbodyproto_test.go",
-        "marshal_json_test.go",
-        "marshal_jsonpb_test.go",
-        "marshal_proto_test.go",
-        "marshaler_registry_test.go",
-        "mux_test.go",
-        "pattern_test.go",
-        "query_test.go",
-    ],
-    embed = [":go_default_library"],
-    deps = [
-        "//internal:go_default_library",
-        "//runtime/internal/examplepb:go_default_library",
-        "//utilities:go_default_library",
-        "@com_github_golang_protobuf//jsonpb:go_default_library_gen",
-        "@com_github_golang_protobuf//proto:go_default_library",
-        "@com_github_golang_protobuf//ptypes:go_default_library_gen",
-        "@go_googleapis//google/api:httpbody_go_proto",
-        "@go_googleapis//google/rpc:errdetails_go_proto",
-        "@io_bazel_rules_go//proto/wkt:duration_go_proto",
-        "@io_bazel_rules_go//proto/wkt:empty_go_proto",
-        "@io_bazel_rules_go//proto/wkt:field_mask_go_proto",
-        "@io_bazel_rules_go//proto/wkt:struct_go_proto",
-        "@io_bazel_rules_go//proto/wkt:timestamp_go_proto",
-        "@io_bazel_rules_go//proto/wkt:wrappers_go_proto",
-        "@org_golang_google_grpc//codes:go_default_library",
-        "@org_golang_google_grpc//metadata:go_default_library",
-        "@org_golang_google_grpc//status:go_default_library",
-    ],
-)
diff --git a/vendor/github.com/grpc-ecosystem/grpc-gateway/utilities/BUILD.bazel b/vendor/github.com/grpc-ecosystem/grpc-gateway/utilities/BUILD.bazel
deleted file mode 100644
index 7109d7932..000000000
--- a/vendor/github.com/grpc-ecosystem/grpc-gateway/utilities/BUILD.bazel
+++ /dev/null
@@ -1,21 +0,0 @@
-load("@io_bazel_rules_go//go:def.bzl", "go_library", "go_test")
-
-package(default_visibility = ["//visibility:public"])
-
-go_library(
-    name = "go_default_library",
-    srcs = [
-        "doc.go",
-        "pattern.go",
-        "readerfactory.go",
-        "trie.go",
-    ],
-    importpath = "github.com/grpc-ecosystem/grpc-gateway/utilities",
-)
-
-go_test(
-    name = "go_default_test",
-    size = "small",
-    srcs = ["trie_test.go"],
-    embed = [":go_default_library"],
-)
diff --git a/vendor/k8s.io/client-go/rest/fake/fake.go b/vendor/k8s.io/client-go/rest/fake/fake.go
new file mode 100644
index 000000000..293e09694
--- /dev/null
+++ b/vendor/k8s.io/client-go/rest/fake/fake.go
@@ -0,0 +1,118 @@
+/*
+Copyright 2014 The Kubernetes 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.
+*/
+
+// This is made a separate package and should only be imported by tests, because
+// it imports testapi
+package fake
+
+import (
+	"net/http"
+	"net/url"
+
+	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/runtime/schema"
+	"k8s.io/apimachinery/pkg/types"
+	restclient "k8s.io/client-go/rest"
+	"k8s.io/client-go/util/flowcontrol"
+)
+
+// CreateHTTPClient creates an http.Client that will invoke the provided roundTripper func
+// when a request is made.
+func CreateHTTPClient(roundTripper func(*http.Request) (*http.Response, error)) *http.Client {
+	return &http.Client{
+		Transport: roundTripperFunc(roundTripper),
+	}
+}
+
+type roundTripperFunc func(*http.Request) (*http.Response, error)
+
+func (f roundTripperFunc) RoundTrip(req *http.Request) (*http.Response, error) {
+	return f(req)
+}
+
+// RESTClient provides a fake RESTClient interface. It is used to mock network
+// interactions via a rest.Request, or to make them via the provided Client to
+// a specific server.
+type RESTClient struct {
+	NegotiatedSerializer runtime.NegotiatedSerializer
+	GroupVersion         schema.GroupVersion
+	VersionedAPIPath     string
+
+	// Err is returned when any request would be made to the server. If Err is set,
+	// Req will not be recorded, Resp will not be returned, and Client will not be
+	// invoked.
+	Err error
+	// Req is set to the last request that was executed (had the methods Do/DoRaw) invoked.
+	Req *http.Request
+	// If Client is specified, the client will be invoked instead of returning Resp if
+	// Err is not set.
+	Client *http.Client
+	// Resp is returned to the caller after Req is recorded, unless Err or Client are set.
+	Resp *http.Response
+}
+
+func (c *RESTClient) Get() *restclient.Request {
+	return c.Verb("GET")
+}
+
+func (c *RESTClient) Put() *restclient.Request {
+	return c.Verb("PUT")
+}
+
+func (c *RESTClient) Patch(pt types.PatchType) *restclient.Request {
+	return c.Verb("PATCH").SetHeader("Content-Type", string(pt))
+}
+
+func (c *RESTClient) Post() *restclient.Request {
+	return c.Verb("POST")
+}
+
+func (c *RESTClient) Delete() *restclient.Request {
+	return c.Verb("DELETE")
+}
+
+func (c *RESTClient) Verb(verb string) *restclient.Request {
+	return c.Request().Verb(verb)
+}
+
+func (c *RESTClient) APIVersion() schema.GroupVersion {
+	return c.GroupVersion
+}
+
+func (c *RESTClient) GetRateLimiter() flowcontrol.RateLimiter {
+	return nil
+}
+
+func (c *RESTClient) Request() *restclient.Request {
+	config := restclient.ClientContentConfig{
+		ContentType:  runtime.ContentTypeJSON,
+		GroupVersion: c.GroupVersion,
+		Negotiator:   runtime.NewClientNegotiator(c.NegotiatedSerializer, c.GroupVersion),
+	}
+	return restclient.NewRequestWithClient(&url.URL{Scheme: "https", Host: "localhost"}, c.VersionedAPIPath, config, CreateHTTPClient(c.do))
+}
+
+// do is invoked when a Request() created by this client is executed.
+func (c *RESTClient) do(req *http.Request) (*http.Response, error) {
+	if c.Err != nil {
+		return nil, c.Err
+	}
+	c.Req = req
+	if c.Client != nil {
+		return c.Client.Do(req)
+	}
+	return c.Resp, nil
+}
diff --git a/vendor/modules.txt b/vendor/modules.txt
index 1a09ec202..af37830ed 100644
--- a/vendor/modules.txt
+++ b/vendor/modules.txt
@@ -881,6 +881,7 @@ k8s.io/client-go/pkg/apis/clientauthentication/v1beta1
 k8s.io/client-go/pkg/version
 k8s.io/client-go/plugin/pkg/client/auth/exec
 k8s.io/client-go/rest
+k8s.io/client-go/rest/fake
 k8s.io/client-go/rest/watch
 k8s.io/client-go/testing
 k8s.io/client-go/tools/auth
@@ -972,7 +973,7 @@ k8s.io/utils/buffer
 k8s.io/utils/integer
 k8s.io/utils/pointer
 k8s.io/utils/trace
-# knative.dev/test-infra v0.0.0-20201009190921-622088de6f36
+# knative.dev/test-infra v0.0.0-20201009204121-322fb08edae7
 ## explicit
 knative.dev/test-infra/scripts
 # sigs.k8s.io/boskos v0.0.0-20200729174948-794df80db9c9
diff --git a/webhook/certificates/resources/certs.go b/webhook/certificates/resources/certs.go
index 9d1bab16a..3b1486469 100644
--- a/webhook/certificates/resources/certs.go
+++ b/webhook/certificates/resources/certs.go
@@ -30,6 +30,7 @@ import (
 	"go.uber.org/zap"
 
 	"knative.dev/pkg/logging"
+	"knative.dev/pkg/network"
 )
 
 const (
@@ -47,11 +48,12 @@ func createCertTemplate(name, namespace string, notAfter time.Time) (*x509.Certi
 
 	serviceName := name + "." + namespace
 	commonName := serviceName + ".svc"
+	serviceHostname := network.GetServiceHostname(name, namespace)
 	serviceNames := []string{
 		name,
 		serviceName,
 		commonName,
-		serviceName + ".svc.cluster.local",
+		serviceHostname,
 	}
 
 	tmpl := x509.Certificate{


Test patch:
diff --git a/pool/pool_test.go b/pool/pool_test.go
index e2db68561..dc274463e 100644
--- a/pool/pool_test.go
+++ b/pool/pool_test.go
@@ -20,9 +20,10 @@ import (
 	"context"
 	"errors"
 	"sync"
-	"sync/atomic"
 	"testing"
 	"time"
+
+	"go.uber.org/atomic"
 )
 
 func TestParallelismNoErrors(t *testing.T) {
@@ -50,7 +51,7 @@ func TestParallelismNoErrors(t *testing.T) {
 				// m guards max.
 				m      sync.Mutex
 				max    int32
-				active int32
+				active atomic.Int32
 			)
 
 			// Use our own waitgroup to ensure that the work
@@ -60,14 +61,14 @@ func TestParallelismNoErrors(t *testing.T) {
 
 			worker := func() error {
 				defer wg.Done()
-				na := atomic.AddInt32(&active, 1)
-				defer atomic.AddInt32(&active, -1)
+				active.Inc()
+				defer active.Dec()
 
 				func() {
 					m.Lock()
 					defer m.Unlock()
-					if max < na {
-						max = na
+					if v := active.Load(); max < v {
+						max = v
 					}
 				}()
 
@@ -129,7 +130,7 @@ func TestParallelismWithErrors(t *testing.T) {
 				// m guards max.
 				m      sync.Mutex
 				max    int32
-				active int32
+				active atomic.Int32
 			)
 
 			// Use our own waitgroup to ensure that the work
@@ -145,14 +146,14 @@ func TestParallelismWithErrors(t *testing.T) {
 			workerFactory := func(err error) func() error {
 				return func() error {
 					defer wg.Done()
-					na := atomic.AddInt32(&active, 1)
-					defer atomic.AddInt32(&active, -1)
+					active.Inc()
+					defer active.Dec()
 
 					func() {
 						m.Lock()
 						defer m.Unlock()
-						if max < na {
-							max = na
+						if v := active.Load(); max < v {
+							max = v
 						}
 					}()
 
diff --git a/test/logstream/interface.go b/test/logstream/interface.go
index bb85de7f6..1c00cdfb5 100644
--- a/test/logstream/interface.go
+++ b/test/logstream/interface.go
@@ -17,16 +17,19 @@ limitations under the License.
 package logstream
 
 import (
+	"context"
 	"os"
 	"sync"
 
 	"knative.dev/pkg/system"
 	"knative.dev/pkg/test"
+	"knative.dev/pkg/test/helpers"
+	logstreamv2 "knative.dev/pkg/test/logstream/v2"
 )
 
 // Canceler is the type of a function returned when a logstream is started to be
 // deferred so that the logstream can be stopped when the test is complete.
-type Canceler func()
+type Canceler = logstreamv2.Canceler
 
 // Start begins streaming the logs from system components with a `key:` matching
 // `test.ObjectNameForTest(t)` to `t.Log`.  It returns a Canceler, which must
@@ -35,8 +38,14 @@ func Start(t test.TLegacy) Canceler {
 	// Do this lazily to make import ordering less important.
 	once.Do(func() {
 		if ns := os.Getenv(system.NamespaceEnvKey); ns != "" {
-			// If SYSTEM_NAMESPACE is set, then start the stream.
-			stream = &kubelogs{namespace: ns}
+			kc, err := test.NewKubeClient(test.Flags.Kubeconfig, test.Flags.Cluster)
+			if err != nil {
+				t.Error("Error loading client config", "error", err)
+				return
+			}
+
+			stream = &shim{logstreamv2.FromNamespace(context.TODO(), kc.Kube, ns)}
+
 		} else {
 			// Otherwise set up a null stream.
 			stream = &null{}
@@ -54,3 +63,18 @@ var (
 	stream streamer
 	once   sync.Once
 )
+
+type shim struct {
+	logstreamv2.Source
+}
+
+func (s *shim) Start(t test.TLegacy) Canceler {
+	name := helpers.ObjectPrefixForTest(t)
+	canceler, err := s.StartStream(name, t.Logf)
+
+	if err != nil {
+		t.Error("Failed to start logstream", "error", err)
+	}
+
+	return canceler
+}
diff --git a/test/logstream/v2/interface.go b/test/logstream/v2/interface.go
new file mode 100644
index 000000000..cf2359a5a
--- /dev/null
+++ b/test/logstream/v2/interface.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 logstream
+
+type (
+	// Canceler is the type of a function returned when a logstream is
+	// started to be deferred so that the logstream can be stopped when
+	// the test is complete.
+	Canceler func()
+
+	// Callback is invoked after pod logs are transformed
+	Callback func(string, ...interface{})
+
+	// Source allows you to create streams for a given resource name
+	Source interface {
+		// Start a log stream for the given resource name and invoke
+		// the callback with the processed log
+		StartStream(name string, l Callback) (Canceler, error)
+	}
+)
diff --git a/test/logstream/kubelogs.go b/test/logstream/v2/stream.go
similarity index 67%
rename from test/logstream/kubelogs.go
rename to test/logstream/v2/stream.go
index ecd217329..eff2f9b3d 100644
--- a/test/logstream/kubelogs.go
+++ b/test/logstream/v2/stream.go
@@ -1,5 +1,5 @@
 /*
-Copyright 2019 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.
@@ -29,29 +29,87 @@ import (
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/util/sets"
 	"k8s.io/apimachinery/pkg/watch"
-
+	"k8s.io/client-go/kubernetes"
 	"knative.dev/pkg/ptr"
-	"knative.dev/pkg/test"
-	"knative.dev/pkg/test/helpers"
 )
 
-type kubelogs struct {
+func FromNamespace(ctx context.Context, c kubernetes.Interface, namespace string) Source {
+	return &namespaceSource{
+		ctx:       ctx,
+		kc:        c,
+		namespace: namespace,
+		keys:      make(map[string]Callback, 1),
+	}
+}
+
+type namespaceSource struct {
 	namespace string
-	kc        *test.KubeClient
+	kc        kubernetes.Interface
+	ctx       context.Context
 
-	once sync.Once
-	m    sync.RWMutex
-	keys map[string]logger
+	m        sync.RWMutex
+	once     sync.Once
+	keys     map[string]Callback
+	watchErr error
 }
 
-type logger func(string, ...interface{})
+func (s *namespaceSource) StartStream(name string, l Callback) (Canceler, error) {
+	s.once.Do(func() { s.watchErr = s.watchPods() })
+	if s.watchErr != nil {
+		return nil, fmt.Errorf("failed to watch pods in namespace %q: %w", s.namespace, s.watchErr)
+	}
 
-var _ streamer = (*kubelogs)(nil)
+	// Register a key
+	s.m.Lock()
+	defer s.m.Unlock()
+	s.keys[name] = l
 
-// timeFormat defines a simple timestamp with millisecond granularity
-const timeFormat = "15:04:05.000"
+	// Return a function that unregisters that key.
+	return func() {
+		s.m.Lock()
+		defer s.m.Unlock()
+		delete(s.keys, name)
+	}, nil
+}
+
+func (s *namespaceSource) watchPods() error {
+	wi, err := s.kc.CoreV1().Pods(s.namespace).Watch(s.ctx, metav1.ListOptions{})
+	if err != nil {
+		return err
+	}
 
-func (k *kubelogs) startForPod(pod *corev1.Pod) {
+	go func() {
+		defer wi.Stop()
+		watchedPods := sets.NewString()
+
+		for {
+			select {
+			case <-s.ctx.Done():
+				return
+			case ev := <-wi.ResultChan():
+				p := ev.Object.(*corev1.Pod)
+				switch ev.Type {
+				case watch.Deleted:
+					watchedPods.Delete(p.Name)
+				case watch.Added, watch.Modified:
+					if watchedPods.Has(p.Name) {
+						continue
+					}
+					if podIsReady(p) {
+						watchedPods.Insert(p.Name)
+						s.startForPod(p)
+						continue
+					}
+				}
+
+			}
+		}
+	}()
+
+	return nil
+}
+
+func (s *namespaceSource) startForPod(pod *corev1.Pod) {
 	// Grab data from all containers in the pods.  We need this in case
 	// an envoy sidecar is injected for mesh installs.  This should be
 	// equivalent to --all-containers.
@@ -59,11 +117,11 @@ func (k *kubelogs) startForPod(pod *corev1.Pod) {
 		// Required for capture below.
 		psn, pn, cn := pod.Namespace, pod.Name, container.Name
 
-		handleLine := k.handleLine
+		handleLine := s.handleLine
 		if cn == "chaosduck" {
 			// Specialcase logs from chaosduck to be able to easily see when pods
 			// have been killed throughout all tests.
-			handleLine = k.handleGenericLine
+			handleLine = s.handleGenericLine
 		}
 
 		go func() {
@@ -75,10 +133,10 @@ func (k *kubelogs) startForPod(pod *corev1.Pod) {
 				SinceSeconds: ptr.Int64(1),
 			}
 
-			req := k.kc.Kube.CoreV1().Pods(psn).GetLogs(pn, options)
+			req := s.kc.CoreV1().Pods(psn).GetLogs(pn, options)
 			stream, err := req.Stream(context.Background())
 			if err != nil {
-				k.handleGenericLine([]byte(err.Error()), pn)
+				s.handleGenericLine([]byte(err.Error()), pn)
 				return
 			}
 			defer stream.Close()
@@ -104,48 +162,10 @@ func podIsReady(p *corev1.Pod) bool {
 	return false
 }
 
-func (k *kubelogs) watchPods(t test.TLegacy) {
-	wi, err := k.kc.Kube.CoreV1().Pods(k.namespace).Watch(context.Background(), metav1.ListOptions{})
-	if err != nil {
-		t.Error("Logstream knative pod watch failed, logs might be missing", "error", err)
-		return
-	}
-	go func() {
-		watchedPods := sets.NewString()
-		for ev := range wi.ResultChan() {
-			p := ev.Object.(*corev1.Pod)
-			switch ev.Type {
-			case watch.Deleted:
-				watchedPods.Delete(p.Name)
-			case watch.Added, watch.Modified:
-				if watchedPods.Has(p.Name) {
-					continue
-				}
-				if podIsReady(p) {
-					watchedPods.Insert(p.Name)
-					k.startForPod(p)
-					continue
-				}
-			}
-		}
-	}()
-}
-
-func (k *kubelogs) init(t test.TLegacy) {
-	k.keys = make(map[string]logger, 1)
-
-	kc, err := test.NewKubeClient(test.Flags.Kubeconfig, test.Flags.Cluster)
-	if err != nil {
-		t.Error("Error loading client config", "error", err)
-		return
-	}
-	k.kc = kc
-
-	// watchPods will start logging for existing pods as well.
-	k.watchPods(t)
-}
+// timeFormat defines a simple timestamp with millisecond granularity
+const timeFormat = "15:04:05.000"
 
-func (k *kubelogs) handleLine(l []byte, pod string) {
+func (s *namespaceSource) handleLine(l []byte, pod string) {
 	// This holds the standard structure of our logs.
 	var line struct {
 		Level      string    `json:"level"`
@@ -166,10 +186,10 @@ func (k *kubelogs) handleLine(l []byte, pod string) {
 		return
 	}
 
-	k.m.RLock()
-	defer k.m.RUnlock()
+	s.m.RLock()
+	defer s.m.RUnlock()
 
-	for name, logf := range k.keys {
+	for name, logf := range s.keys {
 		// TODO(mattmoor): Do a slightly smarter match.
 		if !strings.Contains(line.Key, "/"+name) {
 			continue
@@ -200,31 +220,12 @@ func (k *kubelogs) handleLine(l []byte, pod string) {
 
 // handleGenericLine prints the given logline to all active tests as it cannot be parsed
 // and/or doesn't contain any correlation data (like the chaosduck for example).
-func (k *kubelogs) handleGenericLine(l []byte, pod string) {
-	k.m.RLock()
-	defer k.m.RUnlock()
+func (s *namespaceSource) handleGenericLine(l []byte, pod string) {
+	s.m.RLock()
+	defer s.m.RUnlock()
 
-	for _, logf := range k.keys {
+	for _, logf := range s.keys {
 		// I 15:04:05.000 webhook-699b7b668d-9smk2 this is my message
 		logf("I %s %s %s", time.Now().Format(timeFormat), pod, string(l))
 	}
 }
-
-// Start implements streamer.
-func (k *kubelogs) Start(t test.TLegacy) Canceler {
-	k.once.Do(func() { k.init(t) })
-
-	name := helpers.ObjectPrefixForTest(t)
-
-	// Register a key
-	k.m.Lock()
-	defer k.m.Unlock()
-	k.keys[name] = t.Logf
-
-	// Return a function that unregisters that key.
-	return func() {
-		k.m.Lock()
-		defer k.m.Unlock()
-		delete(k.keys, name)
-	}
-}
diff --git a/test/logstream/v2/stream_test.go b/test/logstream/v2/stream_test.go
new file mode 100644
index 000000000..10d78fad0
--- /dev/null
+++ b/test/logstream/v2/stream_test.go
@@ -0,0 +1,155 @@
+/*
+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 logstream_test
+
+import (
+	"bytes"
+	"context"
+	"fmt"
+	"io/ioutil"
+	"net/http"
+	"sync"
+	"testing"
+	"time"
+
+	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime/schema"
+	"k8s.io/client-go/kubernetes/fake"
+	"k8s.io/client-go/kubernetes/scheme"
+	v1 "k8s.io/client-go/kubernetes/typed/core/v1"
+	fakecorev1 "k8s.io/client-go/kubernetes/typed/core/v1/fake"
+	restclient "k8s.io/client-go/rest"
+	fakerest "k8s.io/client-go/rest/fake"
+	"knative.dev/pkg/test/logstream/v2"
+)
+
+var pod = &corev1.Pod{
+	ObjectMeta: metav1.ObjectMeta{
+		Name:      "chaosduck",
+		Namespace: "default",
+	},
+	Spec: corev1.PodSpec{
+		Containers: []corev1.Container{
+			{Name: "chaosduck"},
+		},
+	},
+}
+
+var readyStatus = corev1.PodStatus{
+	Phase: corev1.PodRunning,
+	Conditions: []corev1.PodCondition{
+		{Type: corev1.PodReady, Status: corev1.ConditionTrue},
+	},
+}
+
+func TestNamespaceStream(t *testing.T) {
+	f := newK8sFake(fake.NewSimpleClientset())
+
+	logFuncInvoked := make(chan string)
+	logFunc := func(format string, args ...interface{}) {
+		close(logFuncInvoked)
+	}
+
+	stream := logstream.FromNamespace(context.TODO(), f, pod.Namespace)
+	stream.StartStream(pod.Name, logFunc)
+
+	podClient := f.CoreV1().Pods(pod.Namespace)
+	if _, err := podClient.Create(context.TODO(), pod, metav1.CreateOptions{}); err != nil {
+		t.Fatal("CreatePod()=", err)
+	}
+
+	pod.Status = readyStatus
+	if _, err := podClient.Update(context.TODO(), pod, metav1.UpdateOptions{}); err != nil {
+		t.Fatal("UpdatePod()=", err)
+	}
+
+	if _, err := f.logBuffer.WriteString("test\n"); err != nil {
+		t.Fatal("WriteString()=", err)
+	}
+
+	select {
+	case <-time.After(time.Second):
+		t.Error("timed out: log message wasn't received")
+	case <-logFuncInvoked:
+	}
+}
+
+func newK8sFake(c *fake.Clientset) *fakeclient {
+	return &fakeclient{
+		Clientset:  c,
+		FakeCoreV1: &fakecorev1.FakeCoreV1{Fake: &c.Fake},
+		logBuffer:  new(buffer),
+	}
+}
+
+type fakeclient struct {
+	*fake.Clientset
+	*fakecorev1.FakeCoreV1
+
+	logBuffer *buffer
+}
+
+type fakePods struct {
+	*fakeclient
+	v1.PodInterface
+	ns string
+}
+
+func (f *fakeclient) CoreV1() v1.CoreV1Interface { return f }
+
+func (f *fakeclient) Pods(ns string) v1.PodInterface {
+	return &fakePods{
+		f,
+		f.FakeCoreV1.Pods(ns),
+		ns,
+	}
+}
+
+func (f *fakePods) GetLogs(name string, opts *corev1.PodLogOptions) *restclient.Request {
+	fakeClient := &fakerest.RESTClient{
+		Client: fakerest.CreateHTTPClient(func(request *http.Request) (*http.Response, error) {
+			resp := &http.Response{
+				StatusCode: http.StatusOK,
+				Body:       ioutil.NopCloser(f.logBuffer),
+			}
+			return resp, nil
+		}),
+		NegotiatedSerializer: scheme.Codecs.WithoutConversion(),
+		GroupVersion:         schema.GroupVersion{Version: "v1"},
+		VersionedAPIPath:     fmt.Sprintf("/api/v1/namespaces/%s/pods/%s/log", f.ns, name),
+	}
+	return fakeClient.Request()
+}
+
+// use a synchronous buffer
+type buffer struct {
+	bytes.Buffer
+	sync.Mutex
+}
+
+func (b *buffer) WriteString(s string) (n int, err error) {
+	b.Lock()
+	defer b.Unlock()
+	return b.Buffer.WriteString(s)
+}
+
+func (b *buffer) Read(p []byte) (n int, err error) {
+	b.Lock()
+	defer b.Unlock()
+	return b.Buffer.Read(p)
+}
diff --git a/vendor/k8s.io/test-infra/prow/config/secret/BUILD.bazel b/vendor/k8s.io/test-infra/prow/config/secret/BUILD.bazel
deleted file mode 100644
index 8e39a801b..000000000
--- a/vendor/k8s.io/test-infra/prow/config/secret/BUILD.bazel
+++ /dev/null
@@ -1,42 +0,0 @@
-package(default_visibility = ["//visibility:public"])
-
-load("@io_bazel_rules_go//go:def.bzl", "go_library", "go_test")
-
-go_library(
-    name = "go_default_library",
-    srcs = [
-        "agent.go",
-        "secret.go",
-    ],
-    importpath = "k8s.io/test-infra/prow/config/secret",
-    visibility = ["//visibility:public"],
-    deps = [
-        "//prow/logrusutil:go_default_library",
-        "@com_github_sirupsen_logrus//:go_default_library",
-        "@io_k8s_apimachinery//pkg/util/sets:go_default_library",
-    ],
-)
-
-filegroup(
-    name = "package-srcs",
-    srcs = glob(["**"]),
-    tags = ["automanaged"],
-    visibility = ["//visibility:private"],
-)
-
-filegroup(
-    name = "all-srcs",
-    srcs = [":package-srcs"],
-    tags = ["automanaged"],
-    visibility = ["//visibility:public"],
-)
-
-go_test(
-    name = "go_default_test",
-    srcs = ["agent_test.go"],
-    embed = [":go_default_library"],
-    deps = [
-        "//prow/logrusutil:go_default_library",
-        "@com_github_sirupsen_logrus//:go_default_library",
-    ],
-)
diff --git a/vendor/k8s.io/test-infra/prow/logrusutil/BUILD.bazel b/vendor/k8s.io/test-infra/prow/logrusutil/BUILD.bazel
deleted file mode 100644
index 558ed3621..000000000
--- a/vendor/k8s.io/test-infra/prow/logrusutil/BUILD.bazel
+++ /dev/null
@@ -1,37 +0,0 @@
-load("@io_bazel_rules_go//go:def.bzl", "go_library", "go_test")
-
-go_library(
-    name = "go_default_library",
-    srcs = ["logrusutil.go"],
-    importpath = "k8s.io/test-infra/prow/logrusutil",
-    visibility = ["//visibility:public"],
-    deps = [
-        "//prow/version:go_default_library",
-        "@com_github_sirupsen_logrus//:go_default_library",
-        "@io_k8s_apimachinery//pkg/util/sets:go_default_library",
-    ],
-)
-
-filegroup(
-    name = "package-srcs",
-    srcs = glob(["**"]),
-    tags = ["automanaged"],
-    visibility = ["//visibility:private"],
-)
-
-filegroup(
-    name = "all-srcs",
-    srcs = [":package-srcs"],
-    tags = ["automanaged"],
-    visibility = ["//visibility:public"],
-)
-
-go_test(
-    name = "go_default_test",
-    srcs = ["logrusutil_test.go"],
-    embed = [":go_default_library"],
-    deps = [
-        "@com_github_sirupsen_logrus//:go_default_library",
-        "@io_k8s_apimachinery//pkg/util/sets:go_default_library",
-    ],
-)
diff --git a/vendor/k8s.io/test-infra/prow/version/BUILD.bazel b/vendor/k8s.io/test-infra/prow/version/BUILD.bazel
deleted file mode 100644
index 1b6c98abb..000000000
--- a/vendor/k8s.io/test-infra/prow/version/BUILD.bazel
+++ /dev/null
@@ -1,23 +0,0 @@
-load("@io_bazel_rules_go//go:def.bzl", "go_library")
-
-go_library(
-    name = "go_default_library",
-    srcs = ["doc.go"],
-    importpath = "k8s.io/test-infra/prow/version",
-    visibility = ["//visibility:public"],
-    x_defs = {"Version": "{DOCKER_TAG}"},
-)
-
-filegroup(
-    name = "package-srcs",
-    srcs = glob(["**"]),
-    tags = ["automanaged"],
-    visibility = ["//visibility:private"],
-)
-
-filegroup(
-    name = "all-srcs",
-    srcs = [":package-srcs"],
-    tags = ["automanaged"],
-    visibility = ["//visibility:public"],
-)
diff --git a/vendor/knative.dev/test-infra/scripts/library.sh b/vendor/knative.dev/test-infra/scripts/library.sh
index 2a5c3cf75..b11cd33be 100644
--- a/vendor/knative.dev/test-infra/scripts/library.sh
+++ b/vendor/knative.dev/test-infra/scripts/library.sh
@@ -561,7 +561,7 @@ function go_update_deps() {
   echo "--- Removing unwanted vendor files"
 
   # Remove unwanted vendor files
-  find vendor/ \( -name "OWNERS" -o -name "*_test.go" \) -print0 | xargs -0 rm -f
+  find vendor/ \( -name "OWNERS" -o -name "OWNERS_ALIASES" -o -name "BUILD" -o -name "BUILD.bazel" -o -name "*_test.go" \) -print0 | xargs -0 rm -f
 
   export GOFLAGS=-mod=vendor