diff --git a/go.mod b/go.mod index 8ffc0bb28..d74eca47a 100644 --- a/go.mod +++ b/go.mod @@ -36,6 +36,7 @@ require ( github.com/docker/go-events v0.0.0-20190806004212-e31b211e4f1c github.com/docker/go-metrics v0.0.1 github.com/docker/go-units v0.5.0 + github.com/emicklei/go-restful/v3 v3.11.0 github.com/fsnotify/fsnotify v1.9.0 github.com/google/go-cmp v0.7.0 github.com/google/uuid v1.6.0 @@ -72,6 +73,7 @@ require ( go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracehttp v1.35.0 go.opentelemetry.io/otel/sdk v1.35.0 go.opentelemetry.io/otel/trace v1.35.0 + go.uber.org/goleak v1.3.0 golang.org/x/mod v0.24.0 golang.org/x/sync v0.14.0 golang.org/x/sys v0.33.0 @@ -83,7 +85,6 @@ require ( k8s.io/client-go v0.32.3 k8s.io/cri-api v0.32.3 k8s.io/klog/v2 v2.130.1 - k8s.io/kubelet v0.32.3 k8s.io/utils v0.0.0-20241104100929-3ea5e8cea738 tags.cncf.io/container-device-interface v1.0.1 ) @@ -95,7 +96,6 @@ require ( github.com/cilium/ebpf v0.16.0 // indirect github.com/containers/ocicrypt v1.2.1 // indirect github.com/cpuguy83/go-md2man/v2 v2.0.5 // indirect - github.com/emicklei/go-restful/v3 v3.11.0 // indirect github.com/felixge/httpsnoop v1.0.4 // indirect github.com/fxamacker/cbor/v2 v2.7.0 // indirect github.com/go-jose/go-jose/v4 v4.0.5 // indirect @@ -145,7 +145,6 @@ require ( google.golang.org/genproto/googleapis/api v0.0.0-20250218202821-56aae31c358a // indirect gopkg.in/yaml.v3 v3.0.1 // indirect k8s.io/api v0.32.3 // indirect - k8s.io/apiserver v0.32.3 // indirect sigs.k8s.io/json v0.0.0-20241010143419-9aa6b5e7a4b3 // indirect sigs.k8s.io/structured-merge-diff/v4 v4.4.2 // indirect sigs.k8s.io/yaml v1.4.0 // indirect diff --git a/go.sum b/go.sum index b193c0165..6f99c33f3 100644 --- a/go.sum +++ b/go.sum @@ -556,8 +556,6 @@ k8s.io/api v0.32.3 h1:Hw7KqxRusq+6QSplE3NYG4MBxZw1BZnq4aP4cJVINls= k8s.io/api v0.32.3/go.mod h1:2wEDTXADtm/HA7CCMD8D8bK4yuBUptzaRhYcYEEYA3k= k8s.io/apimachinery v0.32.3 h1:JmDuDarhDmA/Li7j3aPrwhpNBA94Nvk5zLeOge9HH1U= k8s.io/apimachinery v0.32.3/go.mod h1:GpHVgxoKlTxClKcteaeuF1Ul/lDVb74KpZcxcmLDElE= -k8s.io/apiserver v0.32.3 h1:kOw2KBuHOA+wetX1MkmrxgBr648ksz653j26ESuWNY8= -k8s.io/apiserver v0.32.3/go.mod h1:q1x9B8E/WzShF49wh3ADOh6muSfpmFL0I2t+TG0Zdgc= k8s.io/client-go v0.32.3 h1:RKPVltzopkSgHS7aS98QdscAgtgah/+zmpAogooIqVU= k8s.io/client-go v0.32.3/go.mod h1:3v0+3k4IcT9bXTc4V2rt+d2ZPPG700Xy6Oi0Gdl2PaY= k8s.io/cri-api v0.32.3 h1:E8VXbXNn4yAgmuKTeNzg0C1MFSxzTdlHSwUvjuYlPTY= @@ -566,8 +564,6 @@ k8s.io/klog/v2 v2.130.1 h1:n9Xl7H1Xvksem4KFG4PYbdQCQxqc/tTUyrgXaOhHSzk= k8s.io/klog/v2 v2.130.1/go.mod h1:3Jpz1GvMt720eyJH1ckRHK1EDfpxISzJ7I9OYgaDtPE= k8s.io/kube-openapi v0.0.0-20241105132330-32ad38e42d3f h1:GA7//TjRY9yWGy1poLzYYJJ4JRdzg3+O6e8I+e+8T5Y= k8s.io/kube-openapi v0.0.0-20241105132330-32ad38e42d3f/go.mod h1:R/HEjbvWI0qdfb8viZUeVZm0X6IZnxAydC7YU42CMw4= -k8s.io/kubelet v0.32.3 h1:B9HzW4yB67flx8tN2FYuDwZvxnmK3v5EjxxFvOYjmc8= -k8s.io/kubelet v0.32.3/go.mod h1:yyAQSCKC+tjSlaFw4HQG7Jein+vo+GeKBGdXdQGvL1U= k8s.io/utils v0.0.0-20241104100929-3ea5e8cea738 h1:M3sRQVHv7vB20Xc2ybTt7ODCeFj6JSWYFzOFnYeS6Ro= k8s.io/utils v0.0.0-20241104100929-3ea5e8cea738/go.mod h1:OLgZIPagt7ERELqWJFomSt595RzquPNLL48iOWgYOg0= sigs.k8s.io/json v0.0.0-20241010143419-9aa6b5e7a4b3 h1:/Rv+M11QRah1itp8VhT6HoVx1Ray9eB4DBr+K+/sCJ8= diff --git a/internal/cri/config/config.go b/internal/cri/config/config.go index ea5267e9e..92777eefa 100644 --- a/internal/cri/config/config.go +++ b/internal/cri/config/config.go @@ -28,13 +28,13 @@ import ( "github.com/containerd/log" "github.com/pelletier/go-toml/v2" runtime "k8s.io/cri-api/pkg/apis/runtime/v1" - "k8s.io/kubelet/pkg/cri/streaming" runhcsoptions "github.com/Microsoft/hcsshim/cmd/containerd-shim-runhcs-v1/options" runcoptions "github.com/containerd/containerd/api/types/runc/options" runtimeoptions "github.com/containerd/containerd/api/types/runtimeoptions/v1" "github.com/containerd/containerd/v2/internal/cri/annotations" "github.com/containerd/containerd/v2/internal/cri/opts" + streaming "github.com/containerd/containerd/v2/internal/cri/streamingserver" "github.com/containerd/containerd/v2/pkg/deprecation" "github.com/containerd/containerd/v2/plugins" ) diff --git a/internal/cri/config/streaming.go b/internal/cri/config/streaming.go index b02e0bd58..51a16309a 100644 --- a/internal/cri/config/streaming.go +++ b/internal/cri/config/streaming.go @@ -27,7 +27,7 @@ import ( k8snet "k8s.io/apimachinery/pkg/util/net" k8scert "k8s.io/client-go/util/cert" - "k8s.io/kubelet/pkg/cri/streaming" + streaming "github.com/containerd/containerd/v2/internal/cri/streamingserver" ) type streamListenerMode int diff --git a/internal/cri/server/service.go b/internal/cri/server/service.go index 8b65b1465..d14b3698f 100644 --- a/internal/cri/server/service.go +++ b/internal/cri/server/service.go @@ -32,7 +32,6 @@ import ( imagespec "github.com/opencontainers/image-spec/specs-go/v1" "github.com/opencontainers/runtime-spec/specs-go/features" runtime "k8s.io/cri-api/pkg/apis/runtime/v1" - "k8s.io/kubelet/pkg/cri/streaming" apitypes "github.com/containerd/containerd/api/types" @@ -49,6 +48,7 @@ import ( "github.com/containerd/containerd/v2/internal/cri/store/label" sandboxstore "github.com/containerd/containerd/v2/internal/cri/store/sandbox" snapshotstore "github.com/containerd/containerd/v2/internal/cri/store/snapshot" + streaming "github.com/containerd/containerd/v2/internal/cri/streamingserver" ctrdutil "github.com/containerd/containerd/v2/internal/cri/util" "github.com/containerd/containerd/v2/internal/eventq" nriservice "github.com/containerd/containerd/v2/internal/nri" diff --git a/internal/cri/server/streaming.go b/internal/cri/server/streaming.go index 9cf1870b4..b37a8976b 100644 --- a/internal/cri/server/streaming.go +++ b/internal/cri/server/streaming.go @@ -26,8 +26,8 @@ import ( "k8s.io/client-go/tools/remotecommand" "k8s.io/utils/exec" + streaming "github.com/containerd/containerd/v2/internal/cri/streamingserver" ctrdutil "github.com/containerd/containerd/v2/internal/cri/util" - "k8s.io/kubelet/pkg/cri/streaming" ) type streamRuntime struct { diff --git a/internal/cri/streamingserver/doc.go b/internal/cri/streamingserver/doc.go new file mode 100644 index 000000000..a0f37d8fa --- /dev/null +++ b/internal/cri/streamingserver/doc.go @@ -0,0 +1,34 @@ +/* + Copyright The containerd 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. +*/ + +/* +Copyright 2016 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. +*/ + +// Package streamingserver contains server functions related the cri streaming api. +package streamingserver // import "k8s.io/kubelet/pkg/cri/streaming" diff --git a/vendor/k8s.io/kubelet/pkg/cri/streaming/errors.go b/internal/cri/streamingserver/errors.go similarity index 73% rename from vendor/k8s.io/kubelet/pkg/cri/streaming/errors.go rename to internal/cri/streamingserver/errors.go index 83e218ddb..e36803a8b 100644 --- a/vendor/k8s.io/kubelet/pkg/cri/streaming/errors.go +++ b/internal/cri/streamingserver/errors.go @@ -1,3 +1,19 @@ +/* + Copyright The containerd 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. +*/ + /* Copyright 2016 The Kubernetes Authors. @@ -14,7 +30,7 @@ See the License for the specific language governing permissions and limitations under the License. */ -package streaming +package streamingserver import ( "net/http" diff --git a/vendor/k8s.io/kubelet/pkg/cri/streaming/portforward/constants.go b/internal/cri/streamingserver/portforward/constants.go similarity index 60% rename from vendor/k8s.io/kubelet/pkg/cri/streaming/portforward/constants.go rename to internal/cri/streamingserver/portforward/constants.go index 62b14f205..217a0b114 100644 --- a/vendor/k8s.io/kubelet/pkg/cri/streaming/portforward/constants.go +++ b/internal/cri/streamingserver/portforward/constants.go @@ -1,3 +1,19 @@ +/* + Copyright The containerd 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. +*/ + /* Copyright 2015 The Kubernetes Authors. diff --git a/internal/cri/streamingserver/portforward/doc.go b/internal/cri/streamingserver/portforward/doc.go new file mode 100644 index 000000000..0d57281aa --- /dev/null +++ b/internal/cri/streamingserver/portforward/doc.go @@ -0,0 +1,34 @@ +/* + Copyright The containerd 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. +*/ + +/* +Copyright 2016 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. +*/ + +// Package portforward contains functions related to pod portwarding over the cri streaming api. +package portforward // import "k8s.io/kubelet/pkg/cri/streaming/portforward" diff --git a/vendor/k8s.io/kubelet/pkg/cri/streaming/portforward/httpstream.go b/internal/cri/streamingserver/portforward/httpstream.go similarity index 83% rename from vendor/k8s.io/kubelet/pkg/cri/streaming/portforward/httpstream.go rename to internal/cri/streamingserver/portforward/httpstream.go index 4d38f454c..afd0b38c1 100644 --- a/vendor/k8s.io/kubelet/pkg/cri/streaming/portforward/httpstream.go +++ b/internal/cri/streamingserver/portforward/httpstream.go @@ -1,3 +1,19 @@ +/* + Copyright The containerd 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. +*/ + /* Copyright 2016 The Kubernetes Authors. @@ -25,13 +41,12 @@ import ( "sync" "time" - api "k8s.io/api/core/v1" + api "github.com/containerd/containerd/v2/internal/cri/streamingserver/v1" + "github.com/containerd/log" "k8s.io/apimachinery/pkg/types" "k8s.io/apimachinery/pkg/util/httpstream" "k8s.io/apimachinery/pkg/util/httpstream/spdy" utilruntime "k8s.io/apimachinery/pkg/util/runtime" - - "k8s.io/klog/v2" ) func handleHTTPStreams(req *http.Request, w http.ResponseWriter, portForwarder PortForwarder, podName string, uid types.UID, supportedPortForwardProtocols []string, idleTimeout, streamCreationTimeout time.Duration) error { @@ -43,7 +58,7 @@ func handleHTTPStreams(req *http.Request, w http.ResponseWriter, portForwarder P } streamChan := make(chan httpstream.Stream, 1) - klog.V(5).InfoS("Upgrading port forward response") + log.L.Info("Upgrading port forward response") upgrader := spdy.NewResponseUpgrader() conn := upgrader.UpgradeResponse(w, req, httpStreamReceived(streamChan)) if conn == nil { @@ -51,7 +66,7 @@ func handleHTTPStreams(req *http.Request, w http.ResponseWriter, portForwarder P } defer conn.Close() - klog.V(5).InfoS("Connection setting port forwarding streaming connection idle timeout", "connection", conn, "idleTimeout", idleTimeout) + log.L.Infof("Connection setting port forwarding streaming connection idle timeout connection idleTimeout %v", idleTimeout) conn.SetIdleTimeout(idleTimeout) h := &httpStreamHandler{ @@ -122,11 +137,11 @@ func (h *httpStreamHandler) getStreamPair(requestID string) (*httpStreamPair, bo defer h.streamPairsLock.Unlock() if p, ok := h.streamPairs[requestID]; ok { - klog.V(5).InfoS("Connection request found existing stream pair", "connection", h.conn, "request", requestID) + log.L.Infof("Connection request found existing stream pair connection request %v", requestID) return p, false } - klog.V(5).InfoS("Connection request creating new stream pair", "connection", h.conn, "request", requestID) + log.L.Infof("Connection request creating new stream pair connection request %v", requestID) p := newPortForwardPair(requestID) h.streamPairs[requestID] = p @@ -144,7 +159,7 @@ func (h *httpStreamHandler) monitorStreamPair(p *httpStreamPair, timeout <-chan utilruntime.HandleError(err) p.printError(err.Error()) case <-p.complete: - klog.V(5).InfoS("Connection request successfully received error and data streams", "connection", h.conn, "request", p.requestID) + log.L.Infof("Connection request successfully received error and data streams connection request %v", p.requestID) } h.removeStreamPair(p.requestID) } @@ -175,7 +190,7 @@ func (h *httpStreamHandler) removeStreamPair(requestID string) { func (h *httpStreamHandler) requestID(stream httpstream.Stream) string { requestID := stream.Headers().Get(api.PortForwardRequestIDHeader) if len(requestID) == 0 { - klog.V(5).InfoS("Connection stream received without requestID header", "connection", h.conn) + log.L.Info("Connection stream received without requestID header connection") // If we get here, it's because the connection came from an older client // that isn't generating the request id header // (https://github.com/kubernetes/kubernetes/blob/843134885e7e0b360eb5441e85b1410a8b1a7a0c/pkg/client/unversioned/portforward/portforward.go#L258-L287) @@ -202,7 +217,7 @@ func (h *httpStreamHandler) requestID(stream httpstream.Stream) string { requestID = strconv.Itoa(int(stream.Identifier()) - 2) } - klog.V(5).InfoS("Connection automatically assigning request ID from stream type and stream ID", "connection", h.conn, "request", requestID, "streamType", streamType, "stream", stream.Identifier()) + log.L.Infof("Connection automatically assigning request ID from stream type and stream ID connection request %v streamType %v stream %v", requestID, streamType, stream.Identifier()) } return requestID } @@ -211,17 +226,17 @@ func (h *httpStreamHandler) requestID(stream httpstream.Stream) string { // streams, invoking portForward for each complete stream pair. The loop exits // when the httpstream.Connection is closed. func (h *httpStreamHandler) run() { - klog.V(5).InfoS("Connection waiting for port forward streams", "connection", h.conn) + log.L.Infof("Connection waiting for port forward streams connection %v", h.conn) Loop: for { select { case <-h.conn.CloseChan(): - klog.V(5).InfoS("Connection upgraded connection closed", "connection", h.conn) + log.L.Info("Connection upgraded connection closed connection") break Loop case stream := <-h.streamChan: requestID := h.requestID(stream) streamType := stream.Headers().Get(api.StreamType) - klog.V(5).InfoS("Connection request received new type of stream", "connection", h.conn, "request", requestID, "streamType", streamType) + log.L.Infof("Connection request received new type of stream connection request %v streamType %v", requestID, streamType) p, created := h.getStreamPair(requestID) if created { @@ -248,9 +263,9 @@ func (h *httpStreamHandler) portForward(p *httpStreamPair) { portString := p.dataStream.Headers().Get(api.PortHeader) port, _ := strconv.ParseInt(portString, 10, 32) - klog.V(5).InfoS("Connection request invoking forwarder.PortForward for port", "connection", h.conn, "request", p.requestID, "port", portString) + log.L.Infof("Connection request invoking forwarder.PortForward for port connection request %v port %v", p.requestID, portString) err := h.forwarder.PortForward(ctx, h.pod, h.uid, int32(port), p.dataStream) - klog.V(5).InfoS("Connection request done invoking forwarder.PortForward for port", "connection", h.conn, "request", p.requestID, "port", portString) + log.L.Infof("Connection request done invoking forwarder.PortForward for port connection reqeust %v port %v", p.requestID, portString) if err != nil { msg := fmt.Errorf("error forwarding port %d to pod %s, uid %v: %v", port, h.pod, h.uid, err) @@ -260,8 +275,8 @@ func (h *httpStreamHandler) portForward(p *httpStreamPair) { // with data stream most probably, thus we want to reset the streams // indicating there was a problem and allow a new set of streams be // created to mitigate the problem - p.dataStream.Reset() // nolint:errcheck - p.errorStream.Reset() // nolint:errcheck + p.dataStream.Reset() //nolint:errcheck + p.errorStream.Reset() //nolint:errcheck } } diff --git a/internal/cri/streamingserver/portforward/httpstream_test.go b/internal/cri/streamingserver/portforward/httpstream_test.go new file mode 100644 index 000000000..90e3108cb --- /dev/null +++ b/internal/cri/streamingserver/portforward/httpstream_test.go @@ -0,0 +1,283 @@ +/* + Copyright The containerd 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. +*/ + +/* +Copyright 2016 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. +*/ + +package portforward + +import ( + "net/http" + "testing" + "time" + + api "github.com/containerd/containerd/v2/internal/cri/streamingserver/v1" + "k8s.io/apimachinery/pkg/util/httpstream" +) + +func TestHTTPStreamReceived(t *testing.T) { + tests := map[string]struct { + port string + streamType string + expectedError string + }{ + "missing port": { + expectedError: `"port" header is required`, + }, + "unable to parse port": { + port: "abc", + expectedError: `unable to parse "abc" as a port: strconv.ParseUint: parsing "abc": invalid syntax`, + }, + "negative port": { + port: "-1", + expectedError: `unable to parse "-1" as a port: strconv.ParseUint: parsing "-1": invalid syntax`, + }, + "missing stream type": { + port: "80", + expectedError: `"streamType" header is required`, + }, + "valid port with error stream": { + port: "80", + streamType: "error", + }, + "valid port with data stream": { + port: "80", + streamType: "data", + }, + "invalid stream type": { + port: "80", + streamType: "foo", + expectedError: `invalid stream type "foo"`, + }, + } + for name, test := range tests { + streams := make(chan httpstream.Stream, 1) + f := httpStreamReceived(streams) + stream := newFakeHTTPStream() + if len(test.port) > 0 { + stream.headers.Set("port", test.port) + } + if len(test.streamType) > 0 { + stream.headers.Set("streamType", test.streamType) + } + replySent := make(chan struct{}) + err := f(stream, replySent) + close(replySent) + if len(test.expectedError) > 0 { + if err == nil { + t.Errorf("%s: expected err=%q, but it was nil", name, test.expectedError) + } + if e, a := test.expectedError, err.Error(); e != a { + t.Errorf("%s: expected err=%q, got %q", name, e, a) + } + continue + } + if err != nil { + t.Errorf("%s: unexpected error %v", name, err) + continue + } + if s := <-streams; s != stream { + t.Errorf("%s: expected stream %#v, got %#v", name, stream, s) + } + } +} + +type fakeConn struct { + removeStreamsCalled bool +} + +func (*fakeConn) CreateStream(headers http.Header) (httpstream.Stream, error) { return nil, nil } +func (*fakeConn) Close() error { return nil } +func (*fakeConn) CloseChan() <-chan bool { return nil } +func (*fakeConn) SetIdleTimeout(timeout time.Duration) {} +func (f *fakeConn) RemoveStreams(streams ...httpstream.Stream) { f.removeStreamsCalled = true } + +func TestGetStreamPair(t *testing.T) { + timeout := make(chan time.Time) + + conn := &fakeConn{} + h := &httpStreamHandler{ + streamPairs: make(map[string]*httpStreamPair), + conn: conn, + } + + // test adding a new entry + p, created := h.getStreamPair("1") + if p == nil { + t.Fatalf("unexpected nil pair") + } + if !created { + t.Fatal("expected created=true") + } + if p.dataStream != nil { + t.Errorf("unexpected non-nil data stream") + } + if p.errorStream != nil { + t.Errorf("unexpected non-nil error stream") + } + + // start the monitor for this pair + monitorDone := make(chan struct{}) + go func() { + h.monitorStreamPair(p, timeout) + close(monitorDone) + }() + + if !h.hasStreamPair("1") { + t.Fatal("This should still be true") + } + + // make sure we can retrieve an existing entry + p2, created := h.getStreamPair("1") + if created { + t.Fatal("expected created=false") + } + if p != p2 { + t.Fatalf("retrieving an existing pair: expected %#v, got %#v", p, p2) + } + + // removed via complete + dataStream := newFakeHTTPStream() + dataStream.headers.Set(api.StreamType, api.StreamTypeData) + complete, err := p.add(dataStream) + if err != nil { + t.Fatalf("unexpected error adding data stream to pair: %v", err) + } + if complete { + t.Fatalf("unexpected complete") + } + + errorStream := newFakeHTTPStream() + errorStream.headers.Set(api.StreamType, api.StreamTypeError) + complete, err = p.add(errorStream) + if err != nil { + t.Fatalf("unexpected error adding error stream to pair: %v", err) + } + if !complete { + t.Fatal("unexpected incomplete") + } + + // make sure monitorStreamPair completed + <-monitorDone + + if !conn.removeStreamsCalled { + t.Fatalf("connection remove stream not called") + } + conn.removeStreamsCalled = false + + // make sure the pair was removed + if h.hasStreamPair("1") { + t.Fatal("expected removal of pair after both data and error streams received") + } + + // removed via timeout + p, created = h.getStreamPair("2") + if !created { + t.Fatal("expected created=true") + } + if p == nil { + t.Fatal("expected p not to be nil") + } + + monitorDone = make(chan struct{}) + go func() { + h.monitorStreamPair(p, timeout) + close(monitorDone) + }() + // cause the timeout + close(timeout) + // make sure monitorStreamPair completed + <-monitorDone + if h.hasStreamPair("2") { + t.Fatal("expected stream pair to be removed") + } + if !conn.removeStreamsCalled { + t.Fatalf("connection remove stream not called") + } +} + +func TestRequestID(t *testing.T) { + h := &httpStreamHandler{} + + s := newFakeHTTPStream() + s.headers.Set(api.StreamType, api.StreamTypeError) + s.id = 1 + if e, a := "1", h.requestID(s); e != a { + t.Errorf("expected %q, got %q", e, a) + } + + s.headers.Set(api.StreamType, api.StreamTypeData) + s.id = 3 + if e, a := "1", h.requestID(s); e != a { + t.Errorf("expected %q, got %q", e, a) + } + + s.id = 7 + s.headers.Set(api.PortForwardRequestIDHeader, "2") + if e, a := "2", h.requestID(s); e != a { + t.Errorf("expected %q, got %q", e, a) + } +} + +type fakeHTTPStream struct { + headers http.Header + id uint32 +} + +func newFakeHTTPStream() *fakeHTTPStream { + return &fakeHTTPStream{ + headers: make(http.Header), + } +} + +var _ httpstream.Stream = &fakeHTTPStream{} + +func (s *fakeHTTPStream) Read(data []byte) (int, error) { + return 0, nil +} + +func (s *fakeHTTPStream) Write(data []byte) (int, error) { + return 0, nil +} + +func (s *fakeHTTPStream) Close() error { + return nil +} + +func (s *fakeHTTPStream) Reset() error { + return nil +} + +func (s *fakeHTTPStream) Headers() http.Header { + return s.headers +} + +func (s *fakeHTTPStream) Identifier() uint32 { + return s.id +} diff --git a/vendor/k8s.io/kubelet/pkg/cri/streaming/portforward/portforward.go b/internal/cri/streamingserver/portforward/portforward.go similarity index 77% rename from vendor/k8s.io/kubelet/pkg/cri/streaming/portforward/portforward.go rename to internal/cri/streamingserver/portforward/portforward.go index 7aa668ca4..6eb34b192 100644 --- a/vendor/k8s.io/kubelet/pkg/cri/streaming/portforward/portforward.go +++ b/internal/cri/streamingserver/portforward/portforward.go @@ -1,3 +1,19 @@ +/* + Copyright The containerd 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. +*/ + /* Copyright 2016 The Kubernetes Authors. diff --git a/vendor/k8s.io/kubelet/pkg/cri/streaming/portforward/websocket.go b/internal/cri/streamingserver/portforward/websocket.go similarity index 85% rename from vendor/k8s.io/kubelet/pkg/cri/streaming/portforward/websocket.go rename to internal/cri/streamingserver/portforward/websocket.go index 3700a7e22..b5ceccb98 100644 --- a/vendor/k8s.io/kubelet/pkg/cri/streaming/portforward/websocket.go +++ b/internal/cri/streamingserver/portforward/websocket.go @@ -1,3 +1,19 @@ +/* + Copyright The containerd 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. +*/ + /* Copyright 2016 The Kubernetes Authors. @@ -27,13 +43,12 @@ import ( "sync" "time" - "k8s.io/klog/v2" - - api "k8s.io/api/core/v1" + "github.com/containerd/containerd/v2/internal/cri/streamingserver/responsewriter" + api "github.com/containerd/containerd/v2/internal/cri/streamingserver/v1" + "github.com/containerd/log" "k8s.io/apimachinery/pkg/types" "k8s.io/apimachinery/pkg/util/httpstream/wsstream" "k8s.io/apimachinery/pkg/util/runtime" - "k8s.io/apiserver/pkg/endpoints/responsewriter" ) const ( @@ -187,9 +202,9 @@ func (h *websocketStreamHandler) portForward(p *websocketStreamPair) { defer p.dataStream.Close() defer p.errorStream.Close() - klog.V(5).InfoS("Connection invoking forwarder.PortForward for port", "connection", h.conn, "port", p.port) + log.L.Infof("Connection invoking forwarder.PortForward for port connection port %v", p.port) err := h.forwarder.PortForward(ctx, h.pod, h.uid, p.port, p.dataStream) - klog.V(5).InfoS("Connection done invoking forwarder.PortForward for port", "connection", h.conn, "port", p.port) + log.L.Infof("Connection done invoking forwarder.PortForward for port connection port %v", p.port) if err != nil { msg := fmt.Errorf("error forwarding port %d to pod %s, uid %v: %v", p.port, h.pod, h.uid, err) diff --git a/internal/cri/streamingserver/portforward/websocket_test.go b/internal/cri/streamingserver/portforward/websocket_test.go new file mode 100644 index 000000000..061f2fe8a --- /dev/null +++ b/internal/cri/streamingserver/portforward/websocket_test.go @@ -0,0 +1,117 @@ +/* + Copyright The containerd 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. +*/ + +/* +Copyright 2016 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. +*/ + +package portforward + +import ( + "net/http" + "reflect" + "testing" +) + +func TestV4Options(t *testing.T) { + tests := map[string]struct { + url string + websocket bool + expectedOpts *V4Options + expectedError string + }{ + "non-ws request": { + url: "http://example.com", + expectedOpts: &V4Options{}, + }, + "missing port": { + url: "http://example.com", + websocket: true, + expectedError: `query parameter "port" is required`, + }, + "unable to parse port": { + url: "http://example.com?port=abc", + websocket: true, + expectedError: `unable to parse "abc" as a port: strconv.ParseUint: parsing "abc": invalid syntax`, + }, + "negative port": { + url: "http://example.com?port=-1", + websocket: true, + expectedError: `unable to parse "-1" as a port: strconv.ParseUint: parsing "-1": invalid syntax`, + }, + "one port": { + url: "http://example.com?port=80", + websocket: true, + expectedOpts: &V4Options{ + Ports: []int32{80}, + }, + }, + "multiple ports": { + url: "http://example.com?port=80,90,100", + websocket: true, + expectedOpts: &V4Options{ + Ports: []int32{80, 90, 100}, + }, + }, + "multiple port": { + url: "http://example.com?port=80&port=90", + websocket: true, + expectedOpts: &V4Options{ + Ports: []int32{80, 90}, + }, + }, + } + for name, test := range tests { + req, err := http.NewRequest(http.MethodGet, test.url, nil) + if err != nil { + t.Errorf("%s: invalid url %q err=%q", name, test.url, err) + continue + } + if test.websocket { + req.Header.Set("Connection", "Upgrade") + req.Header.Set("Upgrade", "websocket") + } + opts, err := NewV4Options(req) + if len(test.expectedError) > 0 { + if err == nil { + t.Errorf("%s: expected err=%q, but it was nil", name, test.expectedError) + } + if e, a := test.expectedError, err.Error(); e != a { + t.Errorf("%s: expected err=%q, got %q", name, e, a) + } + continue + } + if err != nil { + t.Errorf("%s: unexpected error %v", name, err) + continue + } + if !reflect.DeepEqual(test.expectedOpts, opts) { + t.Errorf("%s: expected options %#v, got %#v", name, test.expectedOpts, err) + } + } +} diff --git a/vendor/k8s.io/kubelet/pkg/cri/streaming/remotecommand/attach.go b/internal/cri/streamingserver/remotecommand/attach.go similarity index 79% rename from vendor/k8s.io/kubelet/pkg/cri/streaming/remotecommand/attach.go rename to internal/cri/streamingserver/remotecommand/attach.go index aa638499a..858f871bc 100644 --- a/vendor/k8s.io/kubelet/pkg/cri/streaming/remotecommand/attach.go +++ b/internal/cri/streamingserver/remotecommand/attach.go @@ -1,3 +1,19 @@ +/* + Copyright The containerd 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. +*/ + /* Copyright 2016 The Kubernetes Authors. diff --git a/vendor/k8s.io/kubelet/pkg/cri/streaming/remotecommand/doc.go b/internal/cri/streamingserver/remotecommand/doc.go similarity index 55% rename from vendor/k8s.io/kubelet/pkg/cri/streaming/remotecommand/doc.go rename to internal/cri/streamingserver/remotecommand/doc.go index c3b6a6ba3..3e820ac93 100644 --- a/vendor/k8s.io/kubelet/pkg/cri/streaming/remotecommand/doc.go +++ b/internal/cri/streamingserver/remotecommand/doc.go @@ -1,3 +1,19 @@ +/* + Copyright The containerd 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. +*/ + /* Copyright 2016 The Kubernetes Authors. diff --git a/vendor/k8s.io/kubelet/pkg/cri/streaming/remotecommand/exec.go b/internal/cri/streamingserver/remotecommand/exec.go similarity index 83% rename from vendor/k8s.io/kubelet/pkg/cri/streaming/remotecommand/exec.go rename to internal/cri/streamingserver/remotecommand/exec.go index 5ec6b86a8..339ef7fa4 100644 --- a/vendor/k8s.io/kubelet/pkg/cri/streaming/remotecommand/exec.go +++ b/internal/cri/streamingserver/remotecommand/exec.go @@ -1,3 +1,19 @@ +/* + Copyright The containerd 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. +*/ + /* Copyright 2016 The Kubernetes Authors. diff --git a/vendor/k8s.io/kubelet/pkg/cri/streaming/remotecommand/httpstream.go b/internal/cri/streamingserver/remotecommand/httpstream.go similarity index 92% rename from vendor/k8s.io/kubelet/pkg/cri/streaming/remotecommand/httpstream.go rename to internal/cri/streamingserver/remotecommand/httpstream.go index 92ab045d2..c5856e158 100644 --- a/vendor/k8s.io/kubelet/pkg/cri/streaming/remotecommand/httpstream.go +++ b/internal/cri/streamingserver/remotecommand/httpstream.go @@ -1,3 +1,19 @@ +/* + Copyright The containerd 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. +*/ + /* Copyright 2016 The Kubernetes Authors. @@ -17,6 +33,7 @@ limitations under the License. package remotecommand import ( + "context" "encoding/json" "errors" "fmt" @@ -24,7 +41,8 @@ import ( "net/http" "time" - api "k8s.io/api/core/v1" + api "github.com/containerd/containerd/v2/internal/cri/streamingserver/v1" + "github.com/containerd/log" apierrors "k8s.io/apimachinery/pkg/api/errors" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/util/httpstream" @@ -33,8 +51,6 @@ import ( remotecommandconsts "k8s.io/apimachinery/pkg/util/remotecommand" "k8s.io/apimachinery/pkg/util/runtime" "k8s.io/client-go/tools/remotecommand" - - "k8s.io/klog/v2" ) // Options contains details about which streams are required for @@ -54,7 +70,7 @@ func NewOptions(req *http.Request) (*Options, error) { stderr := req.FormValue(api.ExecStderrParam) == "1" if tty && stderr { // TODO: make this an error before we reach this method - klog.V(4).InfoS("Access to exec with tty and stderr is not supported, bypassing stderr") + log.L.Info("Access to exec with tty and stderr is not supported, bypassing stderr") stderr = false } @@ -116,7 +132,7 @@ func createStreams(req *http.Request, w http.ResponseWriter, opts *Options, supp if ctx.resizeStream != nil { ctx.resizeChan = make(chan remotecommand.TerminalSize) - go handleResizeEvents(ctx.resizeStream, ctx.resizeChan) + go handleResizeEvents(req.Context(), ctx.resizeStream, ctx.resizeChan) } return ctx, true @@ -155,7 +171,7 @@ func createHTTPStreamStreams(req *http.Request, w http.ResponseWriter, opts *Opt case remotecommandconsts.StreamProtocolV2Name: handler = &v2ProtocolHandler{} case "": - klog.V(4).InfoS("Client did not request protocol negotiation. Falling back", "protocol", remotecommandconsts.StreamProtocolV1Name) + log.L.Infof("Client did not request protocol negotiation. Falling back protocol %v", remotecommandconsts.StreamProtocolV1Name) fallthrough case remotecommandconsts.StreamProtocolV1Name: handler = &v1ProtocolHandler{} @@ -409,7 +425,7 @@ WaitForStreams: // supportsTerminalResizing returns false because v1ProtocolHandler doesn't support it. func (*v1ProtocolHandler) supportsTerminalResizing() bool { return false } -func handleResizeEvents(stream io.Reader, channel chan<- remotecommand.TerminalSize) { +func handleResizeEvents(reqctx context.Context, stream io.Reader, channel chan<- remotecommand.TerminalSize) { defer runtime.HandleCrash() defer close(channel) @@ -419,7 +435,12 @@ func handleResizeEvents(stream io.Reader, channel chan<- remotecommand.TerminalS if err := decoder.Decode(&size); err != nil { break } - channel <- size + select { + case channel <- size: + case <-reqctx.Done(): + // To prevent go routine leak. + return + } } } diff --git a/internal/cri/streamingserver/remotecommand/httpstream_test.go b/internal/cri/streamingserver/remotecommand/httpstream_test.go new file mode 100644 index 000000000..bee9120f3 --- /dev/null +++ b/internal/cri/streamingserver/remotecommand/httpstream_test.go @@ -0,0 +1,98 @@ +/* + Copyright The containerd 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. +*/ + +/* +Copyright 2025 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. +*/ + +package remotecommand + +import ( + "bytes" + "context" + "encoding/json" + "io" + "testing" + + "github.com/stretchr/testify/require" + "go.uber.org/goleak" + + "k8s.io/client-go/tools/remotecommand" +) + +func TestHandleResizeEvents(t *testing.T) { + var testTerminalSize remotecommand.TerminalSize + rawTerminalSize, err := json.Marshal(&testTerminalSize) + require.NoError(t, err) + + testCases := []struct { + name string + resizeStreamData []byte + cancelContext bool + readFromChannel bool + }{ + { + name: "data attempted to be sent on the channel; channel not read; context canceled", + resizeStreamData: rawTerminalSize, + cancelContext: true, + }, + { + name: "data attempted to be sent on the channel; channel read; context not canceled", + resizeStreamData: rawTerminalSize, + readFromChannel: true, + }, + { + name: "no data attempted to be sent on the channel; context canceled", + cancelContext: true, + }, + { + name: "no data attempted to be sent on the channel; context not canceled", + }, + } + for _, testCase := range testCases { + t.Run(testCase.name, func(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + connCtx := connectionContext{ + resizeStream: io.NopCloser(bytes.NewReader(testCase.resizeStreamData)), + resizeChan: make(chan remotecommand.TerminalSize), + } + + go handleResizeEvents(ctx, connCtx.resizeStream, connCtx.resizeChan) + if testCase.readFromChannel { + gotTerminalSize := <-connCtx.resizeChan + require.Equal(t, gotTerminalSize, testTerminalSize) + } + if testCase.cancelContext { + cancel() + } + + goleak.VerifyNone(t) + cancel() + }) + } +} diff --git a/vendor/k8s.io/kubelet/pkg/cri/streaming/remotecommand/websocket.go b/internal/cri/streamingserver/remotecommand/websocket.go similarity index 85% rename from vendor/k8s.io/kubelet/pkg/cri/streaming/remotecommand/websocket.go rename to internal/cri/streamingserver/remotecommand/websocket.go index 259008885..64297de6d 100644 --- a/vendor/k8s.io/kubelet/pkg/cri/streaming/remotecommand/websocket.go +++ b/internal/cri/streamingserver/remotecommand/websocket.go @@ -1,3 +1,19 @@ +/* + Copyright The containerd 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. +*/ + /* Copyright 2016 The Kubernetes Authors. @@ -21,9 +37,9 @@ import ( "net/http" "time" + "github.com/containerd/containerd/v2/internal/cri/streamingserver/responsewriter" "k8s.io/apimachinery/pkg/util/httpstream/wsstream" "k8s.io/apimachinery/pkg/util/runtime" - "k8s.io/apiserver/pkg/endpoints/responsewriter" ) const ( diff --git a/vendor/k8s.io/kubelet/pkg/cri/streaming/request_cache.go b/internal/cri/streamingserver/request_cache.go similarity index 85% rename from vendor/k8s.io/kubelet/pkg/cri/streaming/request_cache.go rename to internal/cri/streamingserver/request_cache.go index 136f54931..8c2d80b86 100644 --- a/vendor/k8s.io/kubelet/pkg/cri/streaming/request_cache.go +++ b/internal/cri/streamingserver/request_cache.go @@ -1,3 +1,19 @@ +/* + Copyright The containerd 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. +*/ + /* Copyright 2016 The Kubernetes Authors. @@ -14,7 +30,7 @@ See the License for the specific language governing permissions and limitations under the License. */ -package streaming +package streamingserver import ( "container/list" @@ -25,7 +41,7 @@ import ( "sync" "time" - "k8s.io/utils/clock" + "github.com/containerd/containerd/v2/internal/cri/clock" ) var ( diff --git a/internal/cri/streamingserver/request_cache_test.go b/internal/cri/streamingserver/request_cache_test.go new file mode 100644 index 000000000..3caac6666 --- /dev/null +++ b/internal/cri/streamingserver/request_cache_test.go @@ -0,0 +1,238 @@ +/* + Copyright The containerd 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. +*/ + +/* +Copyright 2016 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. +*/ + +package streamingserver + +import ( + "net/http" + "net/http/httptest" + "strconv" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + testingclock "github.com/containerd/containerd/v2/internal/cri/clock/testing" +) + +func TestInsert(t *testing.T) { + c, _ := newTestCache() + + // Insert normal + oldestTok, err := c.Insert(nextRequest()) + require.NoError(t, err) + assert.Len(t, oldestTok, tokenLen) + assertCacheSize(t, c, 1) + + // Insert until full + for i := 0; i < maxInFlight-2; i++ { + tok, err := c.Insert(nextRequest()) + require.NoError(t, err) + assert.Len(t, tok, tokenLen) + } + assertCacheSize(t, c, maxInFlight-1) + + newestReq := nextRequest() + newestTok, err := c.Insert(newestReq) + require.NoError(t, err) + assert.Len(t, newestTok, tokenLen) + assertCacheSize(t, c, maxInFlight) + require.Contains(t, c.tokens, oldestTok, "oldest request should still be cached") + + // Consume newest token. + req, ok := c.Consume(newestTok) + assert.True(t, ok, "newest request should still be cached") + assert.Equal(t, newestReq, req) + require.Contains(t, c.tokens, oldestTok, "oldest request should still be cached") + + // Insert again (still full) + tok, err := c.Insert(nextRequest()) + require.NoError(t, err) + assert.Len(t, tok, tokenLen) + assertCacheSize(t, c, maxInFlight) + + // Insert again (should evict) + _, err = c.Insert(nextRequest()) + assert.Error(t, err, "should reject further requests") + recorder := httptest.NewRecorder() + require.NoError(t, WriteError(err, recorder)) + errResponse := recorder.Result() + assert.Equal(t, http.StatusTooManyRequests, errResponse.StatusCode) + assert.Equal(t, strconv.Itoa(int(cacheTTL.Seconds())), errResponse.Header.Get("Retry-After")) + + assertCacheSize(t, c, maxInFlight) + _, ok = c.Consume(oldestTok) + assert.True(t, ok, "oldest request should be valid") +} + +func TestConsume(t *testing.T) { + c, clock := newTestCache() + + { // Insert & consume. + req := nextRequest() + tok, err := c.Insert(req) + require.NoError(t, err) + assertCacheSize(t, c, 1) + + cachedReq, ok := c.Consume(tok) + assert.True(t, ok) + assert.Equal(t, req, cachedReq) + assertCacheSize(t, c, 0) + } + + { // Insert & consume out of order + req1 := nextRequest() + tok1, err := c.Insert(req1) + require.NoError(t, err) + assertCacheSize(t, c, 1) + + req2 := nextRequest() + tok2, err := c.Insert(req2) + require.NoError(t, err) + assertCacheSize(t, c, 2) + + cachedReq2, ok := c.Consume(tok2) + assert.True(t, ok) + assert.Equal(t, req2, cachedReq2) + assertCacheSize(t, c, 1) + + cachedReq1, ok := c.Consume(tok1) + assert.True(t, ok) + assert.Equal(t, req1, cachedReq1) + assertCacheSize(t, c, 0) + } + + { // Consume a second time + req := nextRequest() + tok, err := c.Insert(req) + require.NoError(t, err) + assertCacheSize(t, c, 1) + + cachedReq, ok := c.Consume(tok) + assert.True(t, ok) + assert.Equal(t, req, cachedReq) + assertCacheSize(t, c, 0) + + _, ok = c.Consume(tok) + assert.False(t, ok) + assertCacheSize(t, c, 0) + } + + { // Consume without insert + _, ok := c.Consume("fooBAR") + assert.False(t, ok) + assertCacheSize(t, c, 0) + } + + { // Consume expired + tok, err := c.Insert(nextRequest()) + require.NoError(t, err) + assertCacheSize(t, c, 1) + + clock.Step(2 * cacheTTL) + + _, ok := c.Consume(tok) + assert.False(t, ok) + assertCacheSize(t, c, 0) + } +} + +func TestGC(t *testing.T) { + c, clock := newTestCache() + + // When empty + c.gc() + assertCacheSize(t, c, 0) + + tok1, err := c.Insert(nextRequest()) + require.NoError(t, err) + assertCacheSize(t, c, 1) + clock.Step(10 * time.Second) + tok2, err := c.Insert(nextRequest()) + require.NoError(t, err) + assertCacheSize(t, c, 2) + + // expired: tok1, tok2 + // non-expired: tok3, tok4 + clock.Step(2 * cacheTTL) + tok3, err := c.Insert(nextRequest()) + require.NoError(t, err) + assertCacheSize(t, c, 1) + clock.Step(10 * time.Second) + tok4, err := c.Insert(nextRequest()) + require.NoError(t, err) + assertCacheSize(t, c, 2) + + _, ok := c.Consume(tok1) + assert.False(t, ok) + _, ok = c.Consume(tok2) + assert.False(t, ok) + _, ok = c.Consume(tok3) + assert.True(t, ok) + _, ok = c.Consume(tok4) + assert.True(t, ok) + + // When full, nothing is expired. + for i := 0; i < maxInFlight; i++ { + _, err := c.Insert(nextRequest()) + require.NoError(t, err) + } + assertCacheSize(t, c, maxInFlight) + + // When everything is expired + clock.Step(2 * cacheTTL) + _, err = c.Insert(nextRequest()) + require.NoError(t, err) + assertCacheSize(t, c, 1) +} + +func newTestCache() (*requestCache, *testingclock.FakeClock) { + c := newRequestCache() + fakeClock := testingclock.NewFakeClock(time.Now()) + c.clock = fakeClock + return c, fakeClock +} + +func assertCacheSize(t *testing.T, cache *requestCache, expectedSize int) { + tokenLen := len(cache.tokens) + llLen := cache.ll.Len() + assert.Equal(t, tokenLen, llLen, "inconsistent cache size! len(tokens)=%d; len(ll)=%d", tokenLen, llLen) + assert.Equal(t, expectedSize, tokenLen, "unexpected cache size!") +} + +var requestUID = 0 + +func nextRequest() interface{} { + requestUID++ + return requestUID +} diff --git a/internal/cri/streamingserver/responsewriter/wrapper.go b/internal/cri/streamingserver/responsewriter/wrapper.go new file mode 100644 index 000000000..cc0f510e6 --- /dev/null +++ b/internal/cri/streamingserver/responsewriter/wrapper.go @@ -0,0 +1,67 @@ +/* + Copyright The containerd 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. +*/ + +/* +Copyright 2021 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. +*/ + +package responsewriter + +import ( + "net/http" +) + +// UserProvidedDecorator represensts a user (client that uses this package) +// provided decorator that wraps an inner http.ResponseWriter object. +// The user-provided decorator object must return the inner (decorated) +// http.ResponseWriter object via the Unwrap function. +type UserProvidedDecorator interface { + http.ResponseWriter + + // Unwrap returns the inner http.ResponseWriter object associated + // with the user-provided decorator. + Unwrap() http.ResponseWriter +} + +// GetOriginal goes through the chain of wrapped http.ResponseWriter objects +// and returns the original http.ResponseWriter object provided to the first +// request handler in the filter chain. +func GetOriginal(w http.ResponseWriter) http.ResponseWriter { + decorator, ok := w.(UserProvidedDecorator) + if !ok { + return w + } + + inner := decorator.Unwrap() + if inner == w { + // infinite cycle here, we should never be here though. + panic("http.ResponseWriter decorator chain has a cycle") + } + + return GetOriginal(inner) +} diff --git a/vendor/k8s.io/kubelet/pkg/cri/streaming/server.go b/internal/cri/streamingserver/server.go similarity index 92% rename from vendor/k8s.io/kubelet/pkg/cri/streaming/server.go rename to internal/cri/streamingserver/server.go index fe5c22b04..1a603046e 100644 --- a/vendor/k8s.io/kubelet/pkg/cri/streaming/server.go +++ b/internal/cri/streamingserver/server.go @@ -1,3 +1,19 @@ +/* + Copyright The containerd 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. +*/ + /* Copyright 2016 The Kubernetes Authors. @@ -14,7 +30,7 @@ See the License for the specific language governing permissions and limitations under the License. */ -package streaming +package streamingserver import ( "context" @@ -32,12 +48,12 @@ import ( restful "github.com/emicklei/go-restful/v3" + "github.com/containerd/containerd/v2/internal/cri/streamingserver/portforward" + remotecommandserver "github.com/containerd/containerd/v2/internal/cri/streamingserver/remotecommand" "k8s.io/apimachinery/pkg/types" remotecommandconsts "k8s.io/apimachinery/pkg/util/remotecommand" "k8s.io/client-go/tools/remotecommand" runtimeapi "k8s.io/cri-api/pkg/apis/runtime/v1" - "k8s.io/kubelet/pkg/cri/streaming/portforward" - remotecommandserver "k8s.io/kubelet/pkg/cri/streaming/remotecommand" ) // Server is the library interface to serve the stream requests. @@ -146,9 +162,10 @@ func NewServer(config Config, runtime Runtime) (Server, error) { handler.Add(ws) s.handler = handler s.server = &http.Server{ - Addr: s.config.Addr, - Handler: s.handler, - TLSConfig: s.config.TLSConfig, + Addr: s.config.Addr, + Handler: s.handler, + TLSConfig: s.config.TLSConfig, + ReadHeaderTimeout: 0, } return s, nil diff --git a/internal/cri/streamingserver/server_test.go b/internal/cri/streamingserver/server_test.go new file mode 100644 index 000000000..5d7ea2d35 --- /dev/null +++ b/internal/cri/streamingserver/server_test.go @@ -0,0 +1,490 @@ +/* + Copyright The containerd 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. +*/ + +/* +Copyright 2016 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. +*/ + +package streamingserver + +import ( + "context" + "crypto/tls" + "io" + "net/http" + "net/http/httptest" + "net/url" + "strconv" + "strings" + "sync" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + kubeletportforward "github.com/containerd/containerd/v2/internal/cri/streamingserver/portforward" + api "github.com/containerd/containerd/v2/internal/cri/streamingserver/v1" + restclient "k8s.io/client-go/rest" + "k8s.io/client-go/tools/remotecommand" + "k8s.io/client-go/transport/spdy" + runtimeapi "k8s.io/cri-api/pkg/apis/runtime/v1" +) + +const ( + testAddr = "localhost:12345" + testContainerID = "container789" + testPodSandboxID = "pod0987" +) + +func TestGetExec(t *testing.T) { + serv, err := NewServer(Config{ + Addr: testAddr, + }, nil) + assert.NoError(t, err) + + tlsServer, err := NewServer(Config{ + Addr: testAddr, + TLSConfig: &tls.Config{}, + }, nil) + assert.NoError(t, err) + + const pathPrefix = "cri/shim" + prefixServer, err := NewServer(Config{ + Addr: testAddr, + BaseURL: &url.URL{ + Scheme: "http", + Host: testAddr, + Path: "/" + pathPrefix + "/", + }, + }, nil) + assert.NoError(t, err) + + assertRequestToken := func(expectedReq *runtimeapi.ExecRequest, cache *requestCache, token string) { + req, ok := cache.Consume(token) + require.True(t, ok, "token %s not found!", token) + assert.Equal(t, expectedReq, req) + } + request := &runtimeapi.ExecRequest{ + ContainerId: testContainerID, + Cmd: []string{"echo", "foo"}, + Tty: true, + Stdin: true, + } + { // Non-TLS + resp, err := serv.GetExec(request) + assert.NoError(t, err) + expectedURL := "http://" + testAddr + "/exec/" + assert.Contains(t, resp.Url, expectedURL) + token := strings.TrimPrefix(resp.Url, expectedURL) + assertRequestToken(request, serv.(*server).cache, token) + } + + { // TLS + resp, err := tlsServer.GetExec(request) + assert.NoError(t, err) + expectedURL := "https://" + testAddr + "/exec/" + assert.Contains(t, resp.Url, expectedURL) + token := strings.TrimPrefix(resp.Url, expectedURL) + assertRequestToken(request, tlsServer.(*server).cache, token) + } + + { // Path prefix + resp, err := prefixServer.GetExec(request) + assert.NoError(t, err) + expectedURL := "http://" + testAddr + "/" + pathPrefix + "/exec/" + assert.Contains(t, resp.Url, expectedURL) + token := strings.TrimPrefix(resp.Url, expectedURL) + assertRequestToken(request, prefixServer.(*server).cache, token) + } +} + +func TestValidateExecAttachRequest(t *testing.T) { + type config struct { + tty bool + stdin bool + stdout bool + stderr bool + } + for _, tc := range []struct { + desc string + configs []config + expectErr bool + }{ + { + desc: "at least one stream must be true", + expectErr: true, + configs: []config{ + {false, false, false, false}, + {true, false, false, false}}, + }, + { + desc: "tty and stderr cannot both be true", + expectErr: true, + configs: []config{ + {true, false, false, true}, + {true, false, true, true}, + {true, true, false, true}, + {true, true, true, true}, + }, + }, + { + desc: "a valid config should pass", + expectErr: false, + configs: []config{ + {false, false, false, true}, + {false, false, true, false}, + {false, false, true, true}, + {false, true, false, false}, + {false, true, false, true}, + {false, true, true, false}, + {false, true, true, true}, + {true, false, true, false}, + {true, true, false, false}, + {true, true, true, false}, + }, + }, + } { + t.Run(tc.desc, func(t *testing.T) { + for _, c := range tc.configs { + // validate the exec request. + execReq := &runtimeapi.ExecRequest{ + ContainerId: testContainerID, + Cmd: []string{"date"}, + Tty: c.tty, + Stdin: c.stdin, + Stdout: c.stdout, + Stderr: c.stderr, + } + err := validateExecRequest(execReq) + assert.Equal(t, tc.expectErr, err != nil, "config: %v, err: %v", c, err) + + // validate the attach request. + attachReq := &runtimeapi.AttachRequest{ + ContainerId: testContainerID, + Tty: c.tty, + Stdin: c.stdin, + Stdout: c.stdout, + Stderr: c.stderr, + } + err = validateAttachRequest(attachReq) + assert.Equal(t, tc.expectErr, err != nil, "config: %v, err: %v", c, err) + } + }) + } +} + +func TestGetAttach(t *testing.T) { + serv, err := NewServer(Config{ + Addr: testAddr, + }, nil) + require.NoError(t, err) + + tlsServer, err := NewServer(Config{ + Addr: testAddr, + TLSConfig: &tls.Config{}, + }, nil) + require.NoError(t, err) + + assertRequestToken := func(expectedReq *runtimeapi.AttachRequest, cache *requestCache, token string) { + req, ok := cache.Consume(token) + require.True(t, ok, "token %s not found!", token) + assert.Equal(t, expectedReq, req) + } + + request := &runtimeapi.AttachRequest{ + ContainerId: testContainerID, + Stdin: true, + Tty: true, + } + { // Non-TLS + resp, err := serv.GetAttach(request) + assert.NoError(t, err) + expectedURL := "http://" + testAddr + "/attach/" + assert.Contains(t, resp.Url, expectedURL) + token := strings.TrimPrefix(resp.Url, expectedURL) + assertRequestToken(request, serv.(*server).cache, token) + } + + { // TLS + resp, err := tlsServer.GetAttach(request) + assert.NoError(t, err) + expectedURL := "https://" + testAddr + "/attach/" + assert.Contains(t, resp.Url, expectedURL) + token := strings.TrimPrefix(resp.Url, expectedURL) + assertRequestToken(request, tlsServer.(*server).cache, token) + } +} + +func TestGetPortForward(t *testing.T) { + podSandboxID := testPodSandboxID + request := &runtimeapi.PortForwardRequest{ + PodSandboxId: podSandboxID, + Port: []int32{1, 2, 3, 4}, + } + + { // Non-TLS + serv, err := NewServer(Config{ + Addr: testAddr, + }, nil) + assert.NoError(t, err) + resp, err := serv.GetPortForward(request) + assert.NoError(t, err) + expectedURL := "http://" + testAddr + "/portforward/" + assert.True(t, strings.HasPrefix(resp.Url, expectedURL)) + token := strings.TrimPrefix(resp.Url, expectedURL) + req, ok := serv.(*server).cache.Consume(token) + require.True(t, ok, "token %s not found!", token) + assert.Equal(t, testPodSandboxID, req.(*runtimeapi.PortForwardRequest).PodSandboxId) + } + + { // TLS + tlsServer, err := NewServer(Config{ + Addr: testAddr, + TLSConfig: &tls.Config{}, + }, nil) + assert.NoError(t, err) + resp, err := tlsServer.GetPortForward(request) + assert.NoError(t, err) + expectedURL := "https://" + testAddr + "/portforward/" + assert.True(t, strings.HasPrefix(resp.Url, expectedURL)) + token := strings.TrimPrefix(resp.Url, expectedURL) + req, ok := tlsServer.(*server).cache.Consume(token) + require.True(t, ok, "token %s not found!", token) + assert.Equal(t, testPodSandboxID, req.(*runtimeapi.PortForwardRequest).PodSandboxId) + } +} + +func TestServeExec(t *testing.T) { + runRemoteCommandTest(t, "exec") +} + +func TestServeAttach(t *testing.T) { + runRemoteCommandTest(t, "attach") +} + +func TestServePortForward(t *testing.T) { + s, testServer := startTestServer(t) + defer testServer.Close() + + resp, err := s.GetPortForward(&runtimeapi.PortForwardRequest{ + PodSandboxId: testPodSandboxID, + }) + require.NoError(t, err) + reqURL, err := url.Parse(resp.Url) + require.NoError(t, err) + + transport, upgrader, err := spdy.RoundTripperFor(&restclient.Config{}) + require.NoError(t, err) + dialer := spdy.NewDialer(upgrader, &http.Client{Transport: transport}, "POST", reqURL) + streamConn, _, err := dialer.Dial(kubeletportforward.ProtocolV1Name) + require.NoError(t, err) + defer streamConn.Close() + + // Create the streams. + headers := http.Header{} + // Error stream is required, but unused in this test. + headers.Set(api.StreamType, api.StreamTypeError) + headers.Set(api.PortHeader, strconv.Itoa(testPort)) + _, err = streamConn.CreateStream(headers) + require.NoError(t, err) + // Setup the data stream. + headers.Set(api.StreamType, api.StreamTypeData) + headers.Set(api.PortHeader, strconv.Itoa(testPort)) + stream, err := streamConn.CreateStream(headers) + require.NoError(t, err) + + doClientStreams(t, "portforward", stream, stream, nil) +} + +// Run the remote command test. +// commandType is either "exec" or "attach". +func runRemoteCommandTest(t *testing.T, commandType string) { + s, testServer := startTestServer(t) + defer testServer.Close() + + var reqURL *url.URL + stdin, stdout, stderr := true, true, true + containerID := testContainerID + switch commandType { + case "exec": + resp, err := s.GetExec(&runtimeapi.ExecRequest{ + ContainerId: containerID, + Cmd: []string{"echo"}, + Stdin: stdin, + Stdout: stdout, + Stderr: stderr, + }) + require.NoError(t, err) + reqURL, err = url.Parse(resp.Url) + require.NoError(t, err) + case "attach": + resp, err := s.GetAttach(&runtimeapi.AttachRequest{ + ContainerId: containerID, + Stdin: stdin, + Stdout: stdout, + Stderr: stderr, + }) + require.NoError(t, err) + reqURL, err = url.Parse(resp.Url) + require.NoError(t, err) + } + + wg := sync.WaitGroup{} + wg.Add(2) + + stdinR, stdinW := io.Pipe() + stdoutR, stdoutW := io.Pipe() + stderrR, stderrW := io.Pipe() + + go func() { + defer wg.Done() + exec, err := remotecommand.NewSPDYExecutor(&restclient.Config{}, "POST", reqURL) + if err != nil { + t.Errorf("unexpected error %v", err) + return + } + + opts := remotecommand.StreamOptions{ + Stdin: stdinR, + Stdout: stdoutW, + Stderr: stderrW, + Tty: false, + } + if err = exec.StreamWithContext(context.Background(), opts); err != nil { + t.Errorf("unexpected error %v", err) + } + }() + + go func() { + defer wg.Done() + doClientStreams(t, commandType, stdinW, stdoutR, stderrR) + }() + + wg.Wait() + + // Repeat request with the same URL should be a 404. + resp, err := http.Get(reqURL.String()) + require.NoError(t, err) + assert.Equal(t, http.StatusNotFound, resp.StatusCode) +} + +func startTestServer(t *testing.T) (Server, *httptest.Server) { + var s Server + testServer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + s.ServeHTTP(w, r) + })) + cleanup := true + defer func() { + if cleanup { + testServer.Close() + } + }() + + testURL, err := url.Parse(testServer.URL) + require.NoError(t, err) + + rt := newFakeRuntime(t) + config := DefaultConfig + config.BaseURL = testURL + s, err = NewServer(config, rt) + require.NoError(t, err) + + cleanup = false // Caller must close the test server. + return s, testServer +} + +const ( + testInput = "abcdefg" + testOutput = "fooBARbaz" + testErr = "ERROR!!!" + testPort = 12345 +) + +func newFakeRuntime(t *testing.T) *fakeRuntime { + return &fakeRuntime{ + t: t, + } +} + +type fakeRuntime struct { + t *testing.T +} + +func (f *fakeRuntime) Exec(_ context.Context, containerID string, cmd []string, stdin io.Reader, stdout, stderr io.WriteCloser, tty bool, resize <-chan remotecommand.TerminalSize) error { + assert.Equal(f.t, testContainerID, containerID) + doServerStreams(f.t, "exec", stdin, stdout, stderr) + return nil +} + +func (f *fakeRuntime) Attach(_ context.Context, containerID string, stdin io.Reader, stdout, stderr io.WriteCloser, tty bool, resize <-chan remotecommand.TerminalSize) error { + assert.Equal(f.t, testContainerID, containerID) + doServerStreams(f.t, "attach", stdin, stdout, stderr) + return nil +} + +func (f *fakeRuntime) PortForward(_ context.Context, podSandboxID string, port int32, stream io.ReadWriteCloser) error { + assert.Equal(f.t, testPodSandboxID, podSandboxID) + assert.EqualValues(f.t, testPort, port) + doServerStreams(f.t, "portforward", stream, stream, nil) + return nil +} + +// Send & receive expected input/output. Must be the inverse of doClientStreams. +// Function will block until the expected i/o is finished. +func doServerStreams(t *testing.T, prefix string, stdin io.Reader, stdout, stderr io.Writer) { + if stderr != nil { + writeExpected(t, "server stderr", stderr, prefix+testErr) + } + readExpected(t, "server stdin", stdin, prefix+testInput) + writeExpected(t, "server stdout", stdout, prefix+testOutput) +} + +// Send & receive expected input/output. Must be the inverse of doServerStreams. +// Function will block until the expected i/o is finished. +func doClientStreams(t *testing.T, prefix string, stdin io.Writer, stdout, stderr io.Reader) { + if stderr != nil { + readExpected(t, "client stderr", stderr, prefix+testErr) + } + writeExpected(t, "client stdin", stdin, prefix+testInput) + readExpected(t, "client stdout", stdout, prefix+testOutput) +} + +// Read and verify the expected string from the stream. +func readExpected(t *testing.T, streamName string, r io.Reader, expected string) { + result := make([]byte, len(expected)) + _, err := io.ReadAtLeast(r, result, len(expected)) + assert.NoError(t, err, "stream %s", streamName) + assert.Equal(t, expected, string(result), "stream %s", streamName) +} + +// Write and verify success of the data over the stream. +func writeExpected(t *testing.T, streamName string, w io.Writer, data string) { + n, err := io.WriteString(w, data) + assert.NoError(t, err, "stream %s", streamName) + assert.Equal(t, len(data), n, "stream %s", streamName) +} diff --git a/internal/cri/streamingserver/v1/doc.go b/internal/cri/streamingserver/v1/doc.go new file mode 100644 index 000000000..4a2d3c12e --- /dev/null +++ b/internal/cri/streamingserver/v1/doc.go @@ -0,0 +1,34 @@ +/* + Copyright The containerd 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. +*/ + +/* +Copyright 2016 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. +*/ + +// Package v1 contains types/constants for core types used by the streaming server command functions. +package v1 // import "k8s.io/api/core/v1" diff --git a/internal/cri/streamingserver/v1/types.go b/internal/cri/streamingserver/v1/types.go new file mode 100644 index 000000000..dee670a14 --- /dev/null +++ b/internal/cri/streamingserver/v1/types.go @@ -0,0 +1,67 @@ +/* + Copyright The containerd 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. +*/ + +/* +Copyright 2016 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. +*/ + +package v1 + +const ( + // Enable stdin for remote command execution + ExecStdinParam = "input" + // Enable stdout for remote command execution + ExecStdoutParam = "output" + // Enable stderr for remote command execution + ExecStderrParam = "error" + // Enable TTY for remote command execution + ExecTTYParam = "tty" + // Command to run for remote command execution + ExecCommandParam = "command" + + // Name of header that specifies stream type + StreamType = "streamType" + // Value for streamType header for stdin stream + StreamTypeStdin = "stdin" + // Value for streamType header for stdout stream + StreamTypeStdout = "stdout" + // Value for streamType header for stderr stream + StreamTypeStderr = "stderr" + // Value for streamType header for data stream + StreamTypeData = "data" + // Value for streamType header for error stream + StreamTypeError = "error" + // Value for streamType header for terminal resize stream + StreamTypeResize = "resize" + + // Name of header that specifies the port being forwarded + PortHeader = "port" + // Name of header that specifies a request ID used to associate the error + // and data streams for a single forwarded connection + PortForwardRequestIDHeader = "requestID" +) diff --git a/vendor/go.uber.org/goleak/.gitignore b/vendor/go.uber.org/goleak/.gitignore new file mode 100644 index 000000000..0fff519a4 --- /dev/null +++ b/vendor/go.uber.org/goleak/.gitignore @@ -0,0 +1,5 @@ +vendor/ +/bin +/lint.log +/cover.out +/cover.html diff --git a/vendor/go.uber.org/goleak/.golangci.yml b/vendor/go.uber.org/goleak/.golangci.yml new file mode 100644 index 000000000..f84e6da8b --- /dev/null +++ b/vendor/go.uber.org/goleak/.golangci.yml @@ -0,0 +1,28 @@ +output: + # Make output more digestible with quickfix in vim/emacs/etc. + sort-results: true + print-issued-lines: false + +linters: + enable: + - gofumpt + - nolintlint + - revive + +linters-settings: + govet: + # These govet checks are disabled by default, but they're useful. + enable: + - niliness + - reflectvaluecompare + - sortslice + - unusedwrite + +issues: + # Print all issues reported by all linters. + max-issues-per-linter: 0 + max-same-issues: 0 + + # Don't ignore some of the issues that golangci-lint considers okay. + # This includes documenting all exported entities. + exclude-use-default: false diff --git a/vendor/go.uber.org/goleak/CHANGELOG.md b/vendor/go.uber.org/goleak/CHANGELOG.md new file mode 100644 index 000000000..5cd3f88a5 --- /dev/null +++ b/vendor/go.uber.org/goleak/CHANGELOG.md @@ -0,0 +1,74 @@ +# Changelog +All notable changes to this project will be documented in this file. + +The format is based on [Keep a Changelog](http://keepachangelog.com/en/1.0.0/) +and this project adheres to [Semantic Versioning](http://semver.org/spec/v2.0.0.html). + +## [1.3.0] +### Fixed +- Built-in ignores now match function names more accurately. + They will no longer ignore stacks because of file names + that look similar to function names. (#112) +### Added +- Add an `IgnoreAnyFunction` option to ignore stack traces + that have the provided function anywhere in the stack. (#113) +- Ignore `testing.runFuzzing` and `testing.runFuzzTests` alongside + other already-ignored test functions (`testing.RunTests`, etc). (#105) +### Changed +- Miscellaneous CI-related fixes. (#103, #108, #114) + +[1.3.0]: https://github.com/uber-go/goleak/compare/v1.2.1...v1.3.0 + +## [1.2.1] +### Changed +- Drop golang/x/lint dependency. + +[1.2.1]: https://github.com/uber-go/goleak/compare/v1.2.0...v1.2.1 + +## [1.2.0] +### Added +- Add Cleanup option that can be used for registering cleanup callbacks. (#78) + +### Changed +- Mark VerifyNone as a test helper. (#75) + +Thanks to @tallclair for their contribution to this release. + +[1.2.0]: https://github.com/uber-go/goleak/compare/v1.1.12...v1.2.0 + +## [1.1.12] +### Fixed +- Fixed logic for ignoring trace related goroutines on Go versions 1.16 and above. + +[1.1.12]: https://github.com/uber-go/goleak/compare/v1.1.11...v1.1.12 + +## [1.1.11] +### Fixed +- Documentation fix on how to test. +- Update dependency on stretchr/testify to v1.7.0. (#59) +- Update dependency on golang.org/x/tools to address CVE-2020-14040. (#62) + +[1.1.11]: https://github.com/uber-go/goleak/compare/v1.1.10...v1.1.11 + +## [1.1.10] +### Added +- [#49]: Add option to ignore current goroutines, which checks for any additional leaks and allows for incremental adoption of goleak in larger projects. + +Thanks to @denis-tingajkin for their contributions to this release. + +[#49]: https://github.com/uber-go/goleak/pull/49 +[1.1.10]: https://github.com/uber-go/goleak/compare/v1.0.0...v1.1.10 + +## [1.0.0] +### Changed +- Migrate to Go modules. + +### Fixed +- Ignore trace related goroutines that cause false positives with -trace. + +[1.0.0]: https://github.com/uber-go/goleak/compare/v0.10.0...v1.0.0 + +## [0.10.0] +- Initial release. + +[0.10.0]: https://github.com/uber-go/goleak/compare/v0.10.0...HEAD diff --git a/vendor/go.uber.org/goleak/LICENSE b/vendor/go.uber.org/goleak/LICENSE new file mode 100644 index 000000000..6c9bde216 --- /dev/null +++ b/vendor/go.uber.org/goleak/LICENSE @@ -0,0 +1,21 @@ +The MIT License (MIT) + +Copyright (c) 2018 Uber Technologies, Inc. + +Permission is hereby granted, free of charge, to any person obtaining a copy +of this software and associated documentation files (the "Software"), to deal +in the Software without restriction, including without limitation the rights +to use, copy, modify, merge, publish, distribute, sublicense, and/or sell +copies of the Software, and to permit persons to whom the Software is +furnished to do so, subject to the following conditions: + +The above copyright notice and this permission notice shall be included in +all copies or substantial portions of the Software. + +THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN +THE SOFTWARE. diff --git a/vendor/go.uber.org/goleak/Makefile b/vendor/go.uber.org/goleak/Makefile new file mode 100644 index 000000000..eb7154af3 --- /dev/null +++ b/vendor/go.uber.org/goleak/Makefile @@ -0,0 +1,45 @@ +# Directory containing the Makefile. +PROJECT_ROOT = $(dir $(abspath $(lastword $(MAKEFILE_LIST)))) + +export GOBIN = $(PROJECT_ROOT)/bin +export PATH := $(GOBIN):$(PATH) + +GO_FILES = $(shell find . \ + -path '*/.*' -prune -o \ + '(' -type f -a -name '*.go' ')' -print) + +# Additional test flags. +TEST_FLAGS ?= + +.PHONY: all +all: lint build test + +.PHONY: lint +lint: golangci-lint tidy-lint + +.PHONY: build +build: + go build ./... + +.PHONY: test +test: + go test -v -race ./... + go test -v -trace=/dev/null . + +.PHONY: cover +cover: + go test -race -coverprofile=cover.out -coverpkg=./... ./... + go tool cover -html=cover.out -o cover.html + +.PHONY: golangci-lint +golangci-lint: + golangci-lint run + +.PHONY: tidy +tidy: + go mod tidy + +.PHONY: tidy-lint +tidy-lint: + go mod tidy + git diff --exit-code -- go.mod go.sum diff --git a/vendor/go.uber.org/goleak/README.md b/vendor/go.uber.org/goleak/README.md new file mode 100644 index 000000000..de3d7d51d --- /dev/null +++ b/vendor/go.uber.org/goleak/README.md @@ -0,0 +1,74 @@ +# goleak [![GoDoc][doc-img]][doc] [![Build Status][ci-img]][ci] [![Coverage Status][cov-img]][cov] + +Goroutine leak detector to help avoid Goroutine leaks. + +## Installation + +You can use `go get` to get the latest version: + +`go get -u go.uber.org/goleak` + +`goleak` also supports semver releases. + +Note that go-leak only [supports][release] the two most recent minor versions of Go. + +## Quick Start + +To verify that there are no unexpected goroutines running at the end of a test: + +```go +func TestA(t *testing.T) { + defer goleak.VerifyNone(t) + + // test logic here. +} +``` + +Instead of checking for leaks at the end of every test, `goleak` can also be run +at the end of every test package by creating a `TestMain` function for your +package: + +```go +func TestMain(m *testing.M) { + goleak.VerifyTestMain(m) +} +``` + +## Determine Source of Package Leaks + +When verifying leaks using `TestMain`, the leak test is only run once after all tests +have been run. This is typically enough to ensure there's no goroutines leaked from +tests, but when there are leaks, it's hard to determine which test is causing them. + +You can use the following bash script to determine the source of the failing test: + +```sh +# Create a test binary which will be used to run each test individually +$ go test -c -o tests + +# Run each test individually, printing "." for successful tests, or the test name +# for failing tests. +$ for test in $(go test -list . | grep -E "^(Test|Example)"); do ./tests -test.run "^$test\$" &>/dev/null && echo -n "." || echo -e "\n$test failed"; done +``` + +This will only print names of failing tests which can be investigated individually. E.g., + +``` +..... +TestLeakyTest failed +....... +``` + +## Stability + +goleak is v1 and follows [SemVer](http://semver.org/) strictly. + +No breaking changes will be made to exported APIs before 2.0. + +[doc-img]: https://godoc.org/go.uber.org/goleak?status.svg +[doc]: https://godoc.org/go.uber.org/goleak +[ci-img]: https://github.com/uber-go/goleak/actions/workflows/ci.yml/badge.svg +[ci]: https://github.com/uber-go/goleak/actions/workflows/ci.yml +[cov-img]: https://codecov.io/gh/uber-go/goleak/branch/master/graph/badge.svg +[cov]: https://codecov.io/gh/uber-go/goleak +[release]: https://go.dev/doc/devel/release#policy diff --git a/vendor/go.uber.org/goleak/doc.go b/vendor/go.uber.org/goleak/doc.go new file mode 100644 index 000000000..3832f8dbc --- /dev/null +++ b/vendor/go.uber.org/goleak/doc.go @@ -0,0 +1,22 @@ +// Copyright (c) 2018 Uber Technologies, Inc. + +// Permission is hereby granted, free of charge, to any person obtaining a copy +// of this software and associated documentation files (the "Software"), to deal +// in the Software without restriction, including without limitation the rights +// to use, copy, modify, merge, publish, distribute, sublicense, and/or sell +// copies of the Software, and to permit persons to whom the Software is +// furnished to do so, subject to the following conditions: +// +// The above copyright notice and this permission notice shall be included in +// all copies or substantial portions of the Software. +// +// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN +// THE SOFTWARE. + +// Package goleak is a Goroutine leak detector. +package goleak // import "go.uber.org/goleak" diff --git a/vendor/go.uber.org/goleak/internal/stack/doc.go b/vendor/go.uber.org/goleak/internal/stack/doc.go new file mode 100644 index 000000000..9179a5654 --- /dev/null +++ b/vendor/go.uber.org/goleak/internal/stack/doc.go @@ -0,0 +1,22 @@ +// Copyright (c) 2017-2023 Uber Technologies, Inc. +// +// Permission is hereby granted, free of charge, to any person obtaining a copy +// of this software and associated documentation files (the "Software"), to deal +// in the Software without restriction, including without limitation the rights +// to use, copy, modify, merge, publish, distribute, sublicense, and/or sell +// copies of the Software, and to permit persons to whom the Software is +// furnished to do so, subject to the following conditions: +// +// The above copyright notice and this permission notice shall be included in +// all copies or substantial portions of the Software. +// +// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN +// THE SOFTWARE. + +// Package stack is used for parsing stacks from `runtime.Stack`. +package stack diff --git a/vendor/go.uber.org/goleak/internal/stack/scan.go b/vendor/go.uber.org/goleak/internal/stack/scan.go new file mode 100644 index 000000000..4b7ac8423 --- /dev/null +++ b/vendor/go.uber.org/goleak/internal/stack/scan.go @@ -0,0 +1,56 @@ +// Copyright (c) 2023 Uber Technologies, Inc. +// +// Permission is hereby granted, free of charge, to any person obtaining a copy +// of this software and associated documentation files (the "Software"), to deal +// in the Software without restriction, including without limitation the rights +// to use, copy, modify, merge, publish, distribute, sublicense, and/or sell +// copies of the Software, and to permit persons to whom the Software is +// furnished to do so, subject to the following conditions: +// +// The above copyright notice and this permission notice shall be included in +// all copies or substantial portions of the Software. +// +// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN +// THE SOFTWARE. + +package stack + +import ( + "bufio" + "io" +) + +// scanner provides a bufio.Scanner the ability to Unscan, +// which allows the current token to be read again +// after the next Scan. +type scanner struct { + *bufio.Scanner + + unscanned bool +} + +func newScanner(r io.Reader) *scanner { + return &scanner{Scanner: bufio.NewScanner(r)} +} + +func (s *scanner) Scan() bool { + if s.unscanned { + s.unscanned = false + return true + } + return s.Scanner.Scan() +} + +// Unscan stops the scanner from advancing its position +// for the next Scan. +// +// Bytes and Text will return the same token after next Scan +// that they do right now. +func (s *scanner) Unscan() { + s.unscanned = true +} diff --git a/vendor/go.uber.org/goleak/internal/stack/stacks.go b/vendor/go.uber.org/goleak/internal/stack/stacks.go new file mode 100644 index 000000000..241a9b844 --- /dev/null +++ b/vendor/go.uber.org/goleak/internal/stack/stacks.go @@ -0,0 +1,298 @@ +// Copyright (c) 2017-2023 Uber Technologies, Inc. +// +// Permission is hereby granted, free of charge, to any person obtaining a copy +// of this software and associated documentation files (the "Software"), to deal +// in the Software without restriction, including without limitation the rights +// to use, copy, modify, merge, publish, distribute, sublicense, and/or sell +// copies of the Software, and to permit persons to whom the Software is +// furnished to do so, subject to the following conditions: +// +// The above copyright notice and this permission notice shall be included in +// all copies or substantial portions of the Software. +// +// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN +// THE SOFTWARE. + +package stack + +import ( + "bytes" + "errors" + "fmt" + "io" + "runtime" + "strconv" + "strings" +) + +const _defaultBufferSize = 64 * 1024 // 64 KiB + +// Stack represents a single Goroutine's stack. +type Stack struct { + id int + state string // e.g. 'running', 'chan receive' + + // The first function on the stack. + firstFunction string + + // A set of all functions in the stack, + allFunctions map[string]struct{} + + // Full, raw stack trace. + fullStack string +} + +// ID returns the goroutine ID. +func (s Stack) ID() int { + return s.id +} + +// State returns the Goroutine's state. +func (s Stack) State() string { + return s.state +} + +// Full returns the full stack trace for this goroutine. +func (s Stack) Full() string { + return s.fullStack +} + +// FirstFunction returns the name of the first function on the stack. +func (s Stack) FirstFunction() string { + return s.firstFunction +} + +// HasFunction reports whether the stack has the given function +// anywhere in it. +func (s Stack) HasFunction(name string) bool { + _, ok := s.allFunctions[name] + return ok +} + +func (s Stack) String() string { + return fmt.Sprintf( + "Goroutine %v in state %v, with %v on top of the stack:\n%s", + s.id, s.state, s.firstFunction, s.Full()) +} + +func getStacks(all bool) []Stack { + trace := getStackBuffer(all) + stacks, err := newStackParser(bytes.NewReader(trace)).Parse() + if err != nil { + // Well-formed stack traces should never fail to parse. + // If they do, it's a bug in this package. + // Panic so we can fix it. + panic(fmt.Sprintf("Failed to parse stack trace: %v\n%s", err, trace)) + } + return stacks +} + +type stackParser struct { + scan *scanner + stacks []Stack + errors []error +} + +func newStackParser(r io.Reader) *stackParser { + return &stackParser{ + scan: newScanner(r), + } +} + +func (p *stackParser) Parse() ([]Stack, error) { + for p.scan.Scan() { + line := p.scan.Text() + + // If we see the goroutine header, start a new stack. + if strings.HasPrefix(line, "goroutine ") { + stack, err := p.parseStack(line) + if err != nil { + p.errors = append(p.errors, err) + continue + } + p.stacks = append(p.stacks, stack) + } + } + + p.errors = append(p.errors, p.scan.Err()) + return p.stacks, errors.Join(p.errors...) +} + +// parseStack parses a single stack trace from the given scanner. +// line is the first line of the stack trace, which should look like: +// +// goroutine 123 [runnable]: +func (p *stackParser) parseStack(line string) (Stack, error) { + id, state, err := parseGoStackHeader(line) + if err != nil { + return Stack{}, fmt.Errorf("parse header: %w", err) + } + + // Read the rest of the stack trace. + var ( + firstFunction string + fullStack bytes.Buffer + ) + funcs := make(map[string]struct{}) + for p.scan.Scan() { + line := p.scan.Text() + if strings.HasPrefix(line, "goroutine ") { + // If we see the goroutine header, + // it's the end of this stack. + // Unscan so the next Scan sees the same line. + p.scan.Unscan() + break + } + + fullStack.WriteString(line) + fullStack.WriteByte('\n') // scanner trims the newline + + if len(line) == 0 { + // Empty line usually marks the end of the stack + // but we don't want to have to rely on that. + // Just skip it. + continue + } + + funcName, creator, err := parseFuncName(line) + if err != nil { + return Stack{}, fmt.Errorf("parse function: %w", err) + } + if !creator { + // A function is part of a goroutine's stack + // only if it's not a "created by" function. + // + // The creator function is part of a different stack. + // We don't care about it right now. + funcs[funcName] = struct{}{} + if firstFunction == "" { + firstFunction = funcName + } + + } + + // The function name followed by a line in the form: + // + // example.com/path/to/package/file.go:123 +0x123 + // + // We don't care about the position so we can skip this line. + if p.scan.Scan() { + // Be defensive: + // Skip the line only if it starts with a tab. + bs := p.scan.Bytes() + if len(bs) > 0 && bs[0] == '\t' { + fullStack.Write(bs) + fullStack.WriteByte('\n') + } else { + // Put it back and let the next iteration handle it + // if it doesn't start with a tab. + p.scan.Unscan() + } + } + + if creator { + // The "created by" line is the last line of the stack. + // We can stop parsing now. + // + // Note that if tracebackancestors=N is set, + // there may be more a traceback of the creator function + // following the "created by" line, + // but it should not be considered part of this stack. + // e.g., + // + // created by testing.(*T).Run in goroutine 1 + // /usr/lib/go/src/testing/testing.go:1648 +0x3ad + // [originating from goroutine 1]: + // testing.(*T).Run(...) + // /usr/lib/go/src/testing/testing.go:1649 +0x3ad + // + break + } + } + + return Stack{ + id: id, + state: state, + firstFunction: firstFunction, + allFunctions: funcs, + fullStack: fullStack.String(), + }, nil +} + +// All returns the stacks for all running goroutines. +func All() []Stack { + return getStacks(true) +} + +// Current returns the stack for the current goroutine. +func Current() Stack { + return getStacks(false)[0] +} + +func getStackBuffer(all bool) []byte { + for i := _defaultBufferSize; ; i *= 2 { + buf := make([]byte, i) + if n := runtime.Stack(buf, all); n < i { + return buf[:n] + } + } +} + +// Parses a single function from the given line. +// The line is in one of these formats: +// +// example.com/path/to/package.funcName(args...) +// example.com/path/to/package.(*typeName).funcName(args...) +// created by example.com/path/to/package.funcName +// created by example.com/path/to/package.funcName in goroutine [...] +// +// Also reports whether the line was a "created by" line. +func parseFuncName(line string) (name string, creator bool, err error) { + if after, ok := strings.CutPrefix(line, "created by "); ok { + // The function name is the part after "created by " + // and before " in goroutine [...]". + idx := strings.Index(after, " in goroutine") + if idx >= 0 { + after = after[:idx] + } + name = after + creator = true + } else if idx := strings.LastIndexByte(line, '('); idx >= 0 { + // The function name is the part before the last '('. + name = line[:idx] + } + + if name == "" { + return "", false, fmt.Errorf("no function found: %q", line) + } + + return name, creator, nil +} + +// parseGoStackHeader parses a stack header that looks like: +// goroutine 643 [runnable]:\n +// And returns the goroutine ID, and the state. +func parseGoStackHeader(line string) (goroutineID int, state string, err error) { + // The scanner will have already trimmed the "\n", + // but we'll guard against it just in case. + // + // Trimming them separately makes them both optional. + line = strings.TrimSuffix(strings.TrimSuffix(line, ":"), "\n") + parts := strings.SplitN(line, " ", 3) + if len(parts) != 3 { + return 0, "", fmt.Errorf("unexpected format: %q", line) + } + + id, err := strconv.Atoi(parts[1]) + if err != nil { + return 0, "", fmt.Errorf("bad goroutine ID %q in line %q", parts[1], line) + } + + state = strings.TrimSuffix(strings.TrimPrefix(parts[2], "["), "]") + return id, state, nil +} diff --git a/vendor/go.uber.org/goleak/leaks.go b/vendor/go.uber.org/goleak/leaks.go new file mode 100644 index 000000000..cc206f181 --- /dev/null +++ b/vendor/go.uber.org/goleak/leaks.go @@ -0,0 +1,108 @@ +// Copyright (c) 2017 Uber Technologies, Inc. + +// Permission is hereby granted, free of charge, to any person obtaining a copy +// of this software and associated documentation files (the "Software"), to deal +// in the Software without restriction, including without limitation the rights +// to use, copy, modify, merge, publish, distribute, sublicense, and/or sell +// copies of the Software, and to permit persons to whom the Software is +// furnished to do so, subject to the following conditions: +// +// The above copyright notice and this permission notice shall be included in +// all copies or substantial portions of the Software. +// +// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN +// THE SOFTWARE. + +package goleak + +import ( + "errors" + "fmt" + + "go.uber.org/goleak/internal/stack" +) + +// TestingT is the minimal subset of testing.TB that we use. +type TestingT interface { + Error(...interface{}) +} + +// filterStacks will filter any stacks excluded by the given opts. +// filterStacks modifies the passed in stacks slice. +func filterStacks(stacks []stack.Stack, skipID int, opts *opts) []stack.Stack { + filtered := stacks[:0] + for _, stack := range stacks { + // Always skip the running goroutine. + if stack.ID() == skipID { + continue + } + // Run any default or user-specified filters. + if opts.filter(stack) { + continue + } + filtered = append(filtered, stack) + } + return filtered +} + +// Find looks for extra goroutines, and returns a descriptive error if +// any are found. +func Find(options ...Option) error { + cur := stack.Current().ID() + + opts := buildOpts(options...) + if opts.cleanup != nil { + return errors.New("Cleanup can only be passed to VerifyNone or VerifyTestMain") + } + var stacks []stack.Stack + retry := true + for i := 0; retry; i++ { + stacks = filterStacks(stack.All(), cur, opts) + + if len(stacks) == 0 { + return nil + } + retry = opts.retry(i) + } + + return fmt.Errorf("found unexpected goroutines:\n%s", stacks) +} + +type testHelper interface { + Helper() +} + +// VerifyNone marks the given TestingT as failed if any extra goroutines are +// found by Find. This is a helper method to make it easier to integrate in +// tests by doing: +// +// defer VerifyNone(t) +// +// VerifyNone is currently incompatible with t.Parallel because it cannot +// associate specific goroutines with specific tests. Thus, non-leaking +// goroutines from other tests running in parallel could fail this check. +// If you need to run tests in parallel, use [VerifyTestMain] instead, +// which will verify that no leaking goroutines exist after ALL tests finish. +func VerifyNone(t TestingT, options ...Option) { + opts := buildOpts(options...) + var cleanup func(int) + cleanup, opts.cleanup = opts.cleanup, nil + + if h, ok := t.(testHelper); ok { + // Mark this function as a test helper, if available. + h.Helper() + } + + if err := Find(opts); err != nil { + t.Error(err) + } + + if cleanup != nil { + cleanup(0) + } +} diff --git a/vendor/go.uber.org/goleak/options.go b/vendor/go.uber.org/goleak/options.go new file mode 100644 index 000000000..53fc0a1de --- /dev/null +++ b/vendor/go.uber.org/goleak/options.go @@ -0,0 +1,198 @@ +// Copyright (c) 2017-2023 Uber Technologies, Inc. +// +// Permission is hereby granted, free of charge, to any person obtaining a copy +// of this software and associated documentation files (the "Software"), to deal +// in the Software without restriction, including without limitation the rights +// to use, copy, modify, merge, publish, distribute, sublicense, and/or sell +// copies of the Software, and to permit persons to whom the Software is +// furnished to do so, subject to the following conditions: +// +// The above copyright notice and this permission notice shall be included in +// all copies or substantial portions of the Software. +// +// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN +// THE SOFTWARE. + +package goleak + +import ( + "strings" + "time" + + "go.uber.org/goleak/internal/stack" +) + +// Option lets users specify custom verifications. +type Option interface { + apply(*opts) +} + +// We retry up to 20 times if we can't find the goroutine that +// we are looking for. In between each attempt, we will sleep for +// a short while to let any running goroutines complete. +const _defaultRetries = 20 + +type opts struct { + filters []func(stack.Stack) bool + maxRetries int + maxSleep time.Duration + cleanup func(int) +} + +// implement apply so that opts struct itself can be used as +// an Option. +func (o *opts) apply(opts *opts) { + opts.filters = o.filters + opts.maxRetries = o.maxRetries + opts.maxSleep = o.maxSleep + opts.cleanup = o.cleanup +} + +// optionFunc lets us easily write options without a custom type. +type optionFunc func(*opts) + +func (f optionFunc) apply(opts *opts) { f(opts) } + +// IgnoreTopFunction ignores any goroutines where the specified function +// is at the top of the stack. The function name should be fully qualified, +// e.g., go.uber.org/goleak.IgnoreTopFunction +func IgnoreTopFunction(f string) Option { + return addFilter(func(s stack.Stack) bool { + return s.FirstFunction() == f + }) +} + +// IgnoreAnyFunction ignores goroutines where the specified function +// is present anywhere in the stack. +// +// The function name must be fully qualified, e.g., +// +// go.uber.org/goleak.IgnoreAnyFunction +// +// For methods, the fully qualified form looks like: +// +// go.uber.org/goleak.(*MyType).MyMethod +func IgnoreAnyFunction(f string) Option { + return addFilter(func(s stack.Stack) bool { + return s.HasFunction(f) + }) +} + +// Cleanup sets up a cleanup function that will be executed at the +// end of the leak check. +// When passed to [VerifyTestMain], the exit code passed to cleanupFunc +// will be set to the exit code of TestMain. +// When passed to [VerifyNone], the exit code will be set to 0. +// This cannot be passed to [Find]. +func Cleanup(cleanupFunc func(exitCode int)) Option { + return optionFunc(func(opts *opts) { + opts.cleanup = cleanupFunc + }) +} + +// IgnoreCurrent records all current goroutines when the option is created, and ignores +// them in any future Find/Verify calls. +func IgnoreCurrent() Option { + excludeIDSet := map[int]bool{} + for _, s := range stack.All() { + excludeIDSet[s.ID()] = true + } + return addFilter(func(s stack.Stack) bool { + return excludeIDSet[s.ID()] + }) +} + +func maxSleep(d time.Duration) Option { + return optionFunc(func(opts *opts) { + opts.maxSleep = d + }) +} + +func addFilter(f func(stack.Stack) bool) Option { + return optionFunc(func(opts *opts) { + opts.filters = append(opts.filters, f) + }) +} + +func buildOpts(options ...Option) *opts { + opts := &opts{ + maxRetries: _defaultRetries, + maxSleep: 100 * time.Millisecond, + } + opts.filters = append(opts.filters, + isTestStack, + isSyscallStack, + isStdLibStack, + isTraceStack, + ) + for _, option := range options { + option.apply(opts) + } + return opts +} + +func (o *opts) filter(s stack.Stack) bool { + for _, filter := range o.filters { + if filter(s) { + return true + } + } + return false +} + +func (o *opts) retry(i int) bool { + if i >= o.maxRetries { + return false + } + + d := time.Duration(int(time.Microsecond) << uint(i)) + if d > o.maxSleep { + d = o.maxSleep + } + time.Sleep(d) + return true +} + +// isTestStack is a default filter installed to automatically skip goroutines +// that the testing package runs while the user's tests are running. +func isTestStack(s stack.Stack) bool { + // Until go1.7, the main goroutine ran RunTests, which started + // the test in a separate goroutine and waited for that test goroutine + // to end by waiting on a channel. + // Since go1.7, a separate goroutine is started to wait for signals. + // T.Parallel is for parallel tests, which are blocked until all serial + // tests have run with T.Parallel at the top of the stack. + // testing.runFuzzTests is for fuzz testing, it's blocked until the test + // function with all seed corpus have run. + // testing.runFuzzing is for fuzz testing, it's blocked until a failing + // input is found. + switch s.FirstFunction() { + case "testing.RunTests", "testing.(*T).Run", "testing.(*T).Parallel", "testing.runFuzzing", "testing.runFuzzTests": + // In pre1.7 and post-1.7, background goroutines started by the testing + // package are blocked waiting on a channel. + return strings.HasPrefix(s.State(), "chan receive") + } + return false +} + +func isSyscallStack(s stack.Stack) bool { + // Typically runs in the background when code uses CGo: + // https://github.com/golang/go/issues/16714 + return s.HasFunction("runtime.goexit") && strings.HasPrefix(s.State(), "syscall") +} + +func isStdLibStack(s stack.Stack) bool { + // Importing os/signal starts a background goroutine. + // The name of the function at the top has changed between versions. + if f := s.FirstFunction(); f == "os/signal.signal_recv" || f == "os/signal.loop" { + return true + } + + // Using signal.Notify will start a runtime goroutine. + return s.HasFunction("runtime.ensureSigM") +} diff --git a/vendor/go.uber.org/goleak/testmain.go b/vendor/go.uber.org/goleak/testmain.go new file mode 100644 index 000000000..7b1a50b7a --- /dev/null +++ b/vendor/go.uber.org/goleak/testmain.go @@ -0,0 +1,69 @@ +// Copyright (c) 2017 Uber Technologies, Inc. +// +// Permission is hereby granted, free of charge, to any person obtaining a copy +// of this software and associated documentation files (the "Software"), to deal +// in the Software without restriction, including without limitation the rights +// to use, copy, modify, merge, publish, distribute, sublicense, and/or sell +// copies of the Software, and to permit persons to whom the Software is +// furnished to do so, subject to the following conditions: +// +// The above copyright notice and this permission notice shall be included in +// all copies or substantial portions of the Software. +// +// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN +// THE SOFTWARE. + +package goleak + +import ( + "fmt" + "io" + "os" +) + +// Variables for stubbing in unit tests. +var ( + _osExit = os.Exit + _osStderr io.Writer = os.Stderr +) + +// TestingM is the minimal subset of testing.M that we use. +type TestingM interface { + Run() int +} + +// VerifyTestMain can be used in a TestMain function for package tests to +// verify that there were no goroutine leaks. +// To use it, your TestMain function should look like: +// +// func TestMain(m *testing.M) { +// goleak.VerifyTestMain(m) +// } +// +// See https://golang.org/pkg/testing/#hdr-Main for more details. +// +// This will run all tests as per normal, and if they were successful, look +// for any goroutine leaks and fail the tests if any leaks were found. +func VerifyTestMain(m TestingM, options ...Option) { + exitCode := m.Run() + opts := buildOpts(options...) + + var cleanup func(int) + cleanup, opts.cleanup = opts.cleanup, nil + if cleanup == nil { + cleanup = _osExit + } + defer func() { cleanup(exitCode) }() + + if exitCode == 0 { + if err := Find(opts); err != nil { + fmt.Fprintf(_osStderr, "goleak: Errors on successful test run: %v\n", err) + exitCode = 1 + } + } +} diff --git a/vendor/go.uber.org/goleak/tracestack_new.go b/vendor/go.uber.org/goleak/tracestack_new.go new file mode 100644 index 000000000..4fc6cefce --- /dev/null +++ b/vendor/go.uber.org/goleak/tracestack_new.go @@ -0,0 +1,30 @@ +// Copyright (c) 2021-2023 Uber Technologies, Inc. +// +// Permission is hereby granted, free of charge, to any person obtaining a copy +// of this software and associated documentation files (the "Software"), to deal +// in the Software without restriction, including without limitation the rights +// to use, copy, modify, merge, publish, distribute, sublicense, and/or sell +// copies of the Software, and to permit persons to whom the Software is +// furnished to do so, subject to the following conditions: +// +// The above copyright notice and this permission notice shall be included in +// all copies or substantial portions of the Software. +// +// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN +// THE SOFTWARE. + +//go:build go1.16 +// +build go1.16 + +package goleak + +import "go.uber.org/goleak/internal/stack" + +func isTraceStack(s stack.Stack) bool { + return s.HasFunction("runtime.ReadTrace") +} diff --git a/vendor/k8s.io/apiserver/LICENSE b/vendor/k8s.io/apiserver/LICENSE deleted file mode 100644 index d64569567..000000000 --- a/vendor/k8s.io/apiserver/LICENSE +++ /dev/null @@ -1,202 +0,0 @@ - - Apache License - Version 2.0, January 2004 - http://www.apache.org/licenses/ - - TERMS AND CONDITIONS FOR USE, REPRODUCTION, AND DISTRIBUTION - - 1. Definitions. - - "License" shall mean the terms and conditions for use, reproduction, - and distribution as defined by Sections 1 through 9 of this document. - - "Licensor" shall mean the copyright owner or entity authorized by - the copyright owner that is granting the License. - - "Legal Entity" shall mean the union of the acting entity and all - other entities that control, are controlled by, or are under common - control with that entity. For the purposes of this definition, - "control" means (i) the power, direct or indirect, to cause the - direction or management of such entity, whether by contract or - otherwise, or (ii) ownership of fifty percent (50%) or more of the - outstanding shares, or (iii) beneficial ownership of such entity. - - "You" (or "Your") shall mean an individual or Legal Entity - exercising permissions granted by this License. - - "Source" form shall mean the preferred form for making modifications, - including but not limited to software source code, documentation - source, and configuration files. - - "Object" form shall mean any form resulting from mechanical - transformation or translation of a Source form, including but - not limited to compiled object code, generated documentation, - and conversions to other media types. - - "Work" shall mean the work of authorship, whether in Source or - Object form, made available under the License, as indicated by a - copyright notice that is included in or attached to the work - (an example is provided in the Appendix below). - - "Derivative Works" shall mean any work, whether in Source or Object - form, that is based on (or derived from) the Work and for which the - editorial revisions, annotations, elaborations, or other modifications - represent, as a whole, an original work of authorship. For the purposes - of this License, Derivative Works shall not include works that remain - separable from, or merely link (or bind by name) to the interfaces of, - the Work and Derivative Works thereof. - - "Contribution" shall mean any work of authorship, including - the original version of the Work and any modifications or additions - to that Work or Derivative Works thereof, that is intentionally - submitted to Licensor for inclusion in the Work by the copyright owner - or by an individual or Legal Entity authorized to submit on behalf of - the copyright owner. For the purposes of this definition, "submitted" - means any form of electronic, verbal, or written communication sent - to the Licensor or its representatives, including but not limited to - communication on electronic mailing lists, source code control systems, - and issue tracking systems that are managed by, or on behalf of, the - Licensor for the purpose of discussing and improving the Work, but - excluding communication that is conspicuously marked or otherwise - designated in writing by the copyright owner as "Not a Contribution." - - "Contributor" shall mean Licensor and any individual or Legal Entity - on behalf of whom a Contribution has been received by Licensor and - subsequently incorporated within the Work. - - 2. Grant of Copyright License. Subject to the terms and conditions of - this License, each Contributor hereby grants to You a perpetual, - worldwide, non-exclusive, no-charge, royalty-free, irrevocable - copyright license to reproduce, prepare Derivative Works of, - publicly display, publicly perform, sublicense, and distribute the - Work and such Derivative Works in Source or Object form. - - 3. Grant of Patent License. Subject to the terms and conditions of - this License, each Contributor hereby grants to You a perpetual, - worldwide, non-exclusive, no-charge, royalty-free, irrevocable - (except as stated in this section) patent license to make, have made, - use, offer to sell, sell, import, and otherwise transfer the Work, - where such license applies only to those patent claims licensable - by such Contributor that are necessarily infringed by their - Contribution(s) alone or by combination of their Contribution(s) - with the Work to which such Contribution(s) was submitted. If You - institute patent litigation against any entity (including a - cross-claim or counterclaim in a lawsuit) alleging that the Work - or a Contribution incorporated within the Work constitutes direct - or contributory patent infringement, then any patent licenses - granted to You under this License for that Work shall terminate - as of the date such litigation is filed. - - 4. Redistribution. You may reproduce and distribute copies of the - Work or Derivative Works thereof in any medium, with or without - modifications, and in Source or Object form, provided that You - meet the following conditions: - - (a) You must give any other recipients of the Work or - Derivative Works a copy of this License; and - - (b) You must cause any modified files to carry prominent notices - stating that You changed the files; and - - (c) You must retain, in the Source form of any Derivative Works - that You distribute, all copyright, patent, trademark, and - attribution notices from the Source form of the Work, - excluding those notices that do not pertain to any part of - the Derivative Works; and - - (d) If the Work includes a "NOTICE" text file as part of its - distribution, then any Derivative Works that You distribute must - include a readable copy of the attribution notices contained - within such NOTICE file, excluding those notices that do not - pertain to any part of the Derivative Works, in at least one - of the following places: within a NOTICE text file distributed - as part of the Derivative Works; within the Source form or - documentation, if provided along with the Derivative Works; or, - within a display generated by the Derivative Works, if and - wherever such third-party notices normally appear. The contents - of the NOTICE file are for informational purposes only and - do not modify the License. You may add Your own attribution - notices within Derivative Works that You distribute, alongside - or as an addendum to the NOTICE text from the Work, provided - that such additional attribution notices cannot be construed - as modifying the License. - - You may add Your own copyright statement to Your modifications and - may provide additional or different license terms and conditions - for use, reproduction, or distribution of Your modifications, or - for any such Derivative Works as a whole, provided Your use, - reproduction, and distribution of the Work otherwise complies with - the conditions stated in this License. - - 5. Submission of Contributions. Unless You explicitly state otherwise, - any Contribution intentionally submitted for inclusion in the Work - by You to the Licensor shall be under the terms and conditions of - this License, without any additional terms or conditions. - Notwithstanding the above, nothing herein shall supersede or modify - the terms of any separate license agreement you may have executed - with Licensor regarding such Contributions. - - 6. Trademarks. This License does not grant permission to use the trade - names, trademarks, service marks, or product names of the Licensor, - except as required for reasonable and customary use in describing the - origin of the Work and reproducing the content of the NOTICE file. - - 7. Disclaimer of Warranty. Unless required by applicable law or - agreed to in writing, Licensor provides the Work (and each - Contributor provides its Contributions) on an "AS IS" BASIS, - WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or - implied, including, without limitation, any warranties or conditions - of TITLE, NON-INFRINGEMENT, MERCHANTABILITY, or FITNESS FOR A - PARTICULAR PURPOSE. You are solely responsible for determining the - appropriateness of using or redistributing the Work and assume any - risks associated with Your exercise of permissions under this License. - - 8. Limitation of Liability. In no event and under no legal theory, - whether in tort (including negligence), contract, or otherwise, - unless required by applicable law (such as deliberate and grossly - negligent acts) or agreed to in writing, shall any Contributor be - liable to You for damages, including any direct, indirect, special, - incidental, or consequential damages of any character arising as a - result of this License or out of the use or inability to use the - Work (including but not limited to damages for loss of goodwill, - work stoppage, computer failure or malfunction, or any and all - other commercial damages or losses), even if such Contributor - has been advised of the possibility of such damages. - - 9. Accepting Warranty or Additional Liability. While redistributing - the Work or Derivative Works thereof, You may choose to offer, - and charge a fee for, acceptance of support, warranty, indemnity, - or other liability obligations and/or rights consistent with this - License. However, in accepting such obligations, You may act only - on Your own behalf and on Your sole responsibility, not on behalf - of any other Contributor, and only if You agree to indemnify, - defend, and hold each Contributor harmless for any liability - incurred by, or claims asserted against, such Contributor by reason - of your accepting any such warranty or additional liability. - - END OF TERMS AND CONDITIONS - - APPENDIX: How to apply the Apache License to your work. - - To apply the Apache License to your work, attach the following - boilerplate notice, with the fields enclosed by brackets "[]" - replaced with your own identifying information. (Don't include - the brackets!) The text should be enclosed in the appropriate - comment syntax for the file format. We also recommend that a - file or class name and description of purpose be included on the - same "printed page" as the copyright notice for easier - identification within third-party archives. - - Copyright [yyyy] [name of copyright owner] - - 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. diff --git a/vendor/k8s.io/apiserver/pkg/endpoints/responsewriter/fake.go b/vendor/k8s.io/apiserver/pkg/endpoints/responsewriter/fake.go deleted file mode 100644 index 3a8fe7a6a..000000000 --- a/vendor/k8s.io/apiserver/pkg/endpoints/responsewriter/fake.go +++ /dev/null @@ -1,54 +0,0 @@ -/* -Copyright 2021 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. -*/ - -package responsewriter - -import ( - "bufio" - "net" - "net/http" -) - -var _ http.ResponseWriter = &FakeResponseWriter{} - -// FakeResponseWriter implements http.ResponseWriter, -// it is used for testing purpose only -type FakeResponseWriter struct{} - -func (fw *FakeResponseWriter) Header() http.Header { return http.Header{} } -func (fw *FakeResponseWriter) WriteHeader(code int) {} -func (fw *FakeResponseWriter) Write(bs []byte) (int, error) { return len(bs), nil } - -// For HTTP2 an http.ResponseWriter object implements -// http.Flusher and http.CloseNotifier. -// It is used for testing purpose only -type FakeResponseWriterFlusherCloseNotifier struct { - *FakeResponseWriter -} - -func (fw *FakeResponseWriterFlusherCloseNotifier) Flush() {} -func (fw *FakeResponseWriterFlusherCloseNotifier) CloseNotify() <-chan bool { return nil } - -// For HTTP/1.x an http.ResponseWriter object implements -// http.Flusher, http.CloseNotifier and http.Hijacker. -// It is used for testing purpose only -type FakeResponseWriterFlusherCloseNotifierHijacker struct { - *FakeResponseWriterFlusherCloseNotifier -} - -func (fw *FakeResponseWriterFlusherCloseNotifierHijacker) Hijack() (net.Conn, *bufio.ReadWriter, error) { - return nil, nil, nil -} diff --git a/vendor/k8s.io/apiserver/pkg/endpoints/responsewriter/wrapper.go b/vendor/k8s.io/apiserver/pkg/endpoints/responsewriter/wrapper.go deleted file mode 100644 index 893dfa281..000000000 --- a/vendor/k8s.io/apiserver/pkg/endpoints/responsewriter/wrapper.go +++ /dev/null @@ -1,180 +0,0 @@ -/* -Copyright 2021 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. -*/ - -package responsewriter - -import ( - "bufio" - "net" - "net/http" -) - -// UserProvidedDecorator represensts a user (client that uses this package) -// provided decorator that wraps an inner http.ResponseWriter object. -// The user-provided decorator object must return the inner (decorated) -// http.ResponseWriter object via the Unwrap function. -type UserProvidedDecorator interface { - http.ResponseWriter - - // Unwrap returns the inner http.ResponseWriter object associated - // with the user-provided decorator. - Unwrap() http.ResponseWriter -} - -// WrapForHTTP1Or2 accepts a user-provided decorator of an "inner" http.responseWriter -// object and potentially wraps the user-provided decorator with a new http.ResponseWriter -// object that implements http.CloseNotifier, http.Flusher, and/or http.Hijacker by -// delegating to the user-provided decorator (if it implements the relevant method) or -// the inner http.ResponseWriter (otherwise), so that the returned http.ResponseWriter -// object implements the same subset of those interfaces as the inner http.ResponseWriter. -// -// This function handles the following three casses. -// - The inner ResponseWriter implements `http.CloseNotifier`, `http.Flusher`, -// and `http.Hijacker` (an HTTP/1.1 sever provides such a ResponseWriter). -// - The inner ResponseWriter implements `http.CloseNotifier` and `http.Flusher` -// but not `http.Hijacker` (an HTTP/2 server provides such a ResponseWriter). -// - All the other cases collapse to this one, in which the given ResponseWriter is returned. -// -// There are three applicable terms: -// - "outer": this is the ResponseWriter object returned by the WrapForHTTP1Or2 function. -// - "user-provided decorator" or "middle": this is the user-provided decorator -// that decorates an inner ResponseWriter object. A user-provided decorator -// implements the UserProvidedDecorator interface. A user-provided decorator -// may or may not implement http.CloseNotifier, http.Flusher or http.Hijacker. -// - "inner": the ResponseWriter that the user-provided decorator extends. -func WrapForHTTP1Or2(decorator UserProvidedDecorator) http.ResponseWriter { - // from go net/http documentation: - // The default HTTP/1.x and HTTP/2 ResponseWriter implementations support Flusher - // Handlers should always test for this ability at runtime. - // - // The Hijacker interface is implemented by ResponseWriters that allow an HTTP handler - // to take over the connection. - // The default ResponseWriter for HTTP/1.x connections supports Hijacker, but HTTP/2 connections - // intentionally do not. ResponseWriter wrappers may also not support Hijacker. - // Handlers should always test for this ability at runtime - // - // The CloseNotifier interface is implemented by ResponseWriters which allow detecting - // when the underlying connection has gone away. - // Deprecated: the CloseNotifier interface predates Go's context package. - // New code should use Request.Context instead. - inner := decorator.Unwrap() - if innerNotifierFlusher, ok := inner.(CloseNotifierFlusher); ok { - // for HTTP/2 request, the default ResponseWriter object (http2responseWriter) - // implements Flusher and CloseNotifier. - outerHTTP2 := outerWithCloseNotifyAndFlush{ - UserProvidedDecorator: decorator, - InnerCloseNotifierFlusher: innerNotifierFlusher, - } - - if innerHijacker, hijackable := inner.(http.Hijacker); hijackable { - // for HTTP/1.x request the default implementation of ResponseWriter - // also implement CloseNotifier, Flusher and Hijacker - return &outerWithCloseNotifyFlushAndHijack{ - outerWithCloseNotifyAndFlush: outerHTTP2, - InnerHijacker: innerHijacker, - } - } - - return outerHTTP2 - } - - // we should never be here for either http/1.x or http2 request - return decorator -} - -// CloseNotifierFlusher is a combination of http.CloseNotifier and http.Flusher -// This applies to both http/1.x and http2 requests. -type CloseNotifierFlusher interface { - http.CloseNotifier - http.Flusher -} - -// GetOriginal goes through the chain of wrapped http.ResponseWriter objects -// and returns the original http.ResponseWriter object provided to the first -// request handler in the filter chain. -func GetOriginal(w http.ResponseWriter) http.ResponseWriter { - decorator, ok := w.(UserProvidedDecorator) - if !ok { - return w - } - - inner := decorator.Unwrap() - if inner == w { - // infinite cycle here, we should never be here though. - panic("http.ResponseWriter decorator chain has a cycle") - } - - return GetOriginal(inner) -} - -//nolint:staticcheck // SA1019 -var _ http.CloseNotifier = outerWithCloseNotifyAndFlush{} -var _ http.Flusher = outerWithCloseNotifyAndFlush{} -var _ http.ResponseWriter = outerWithCloseNotifyAndFlush{} -var _ UserProvidedDecorator = outerWithCloseNotifyAndFlush{} - -// outerWithCloseNotifyAndFlush is the outer object that extends the -// user provied decorator with http.CloseNotifier and http.Flusher only. -type outerWithCloseNotifyAndFlush struct { - // UserProvidedDecorator is the user-provided object, it decorates - // an inner ResponseWriter object. - UserProvidedDecorator - - // http.CloseNotifier and http.Flusher for the inner object - InnerCloseNotifierFlusher CloseNotifierFlusher -} - -func (wr outerWithCloseNotifyAndFlush) CloseNotify() <-chan bool { - if notifier, ok := wr.UserProvidedDecorator.(http.CloseNotifier); ok { - return notifier.CloseNotify() - } - - return wr.InnerCloseNotifierFlusher.CloseNotify() -} - -func (wr outerWithCloseNotifyAndFlush) Flush() { - if flusher, ok := wr.UserProvidedDecorator.(http.Flusher); ok { - flusher.Flush() - return - } - - wr.InnerCloseNotifierFlusher.Flush() -} - -//lint:file-ignore SA1019 Keep supporting deprecated http.CloseNotifier -var _ http.CloseNotifier = outerWithCloseNotifyFlushAndHijack{} -var _ http.Flusher = outerWithCloseNotifyFlushAndHijack{} -var _ http.Hijacker = outerWithCloseNotifyFlushAndHijack{} -var _ http.ResponseWriter = outerWithCloseNotifyFlushAndHijack{} -var _ UserProvidedDecorator = outerWithCloseNotifyFlushAndHijack{} - -// outerWithCloseNotifyFlushAndHijack is the outer object that extends the -// user-provided decorator with http.CloseNotifier, http.Flusher and http.Hijacker. -// This applies to http/1.x requests only. -type outerWithCloseNotifyFlushAndHijack struct { - outerWithCloseNotifyAndFlush - - // http.Hijacker for the inner object - InnerHijacker http.Hijacker -} - -func (wr outerWithCloseNotifyFlushAndHijack) Hijack() (net.Conn, *bufio.ReadWriter, error) { - if hijacker, ok := wr.UserProvidedDecorator.(http.Hijacker); ok { - return hijacker.Hijack() - } - - return wr.InnerHijacker.Hijack() -} diff --git a/vendor/k8s.io/kubelet/LICENSE b/vendor/k8s.io/kubelet/LICENSE deleted file mode 100644 index d64569567..000000000 --- a/vendor/k8s.io/kubelet/LICENSE +++ /dev/null @@ -1,202 +0,0 @@ - - Apache License - Version 2.0, January 2004 - http://www.apache.org/licenses/ - - TERMS AND CONDITIONS FOR USE, REPRODUCTION, AND DISTRIBUTION - - 1. Definitions. - - "License" shall mean the terms and conditions for use, reproduction, - and distribution as defined by Sections 1 through 9 of this document. - - "Licensor" shall mean the copyright owner or entity authorized by - the copyright owner that is granting the License. - - "Legal Entity" shall mean the union of the acting entity and all - other entities that control, are controlled by, or are under common - control with that entity. For the purposes of this definition, - "control" means (i) the power, direct or indirect, to cause the - direction or management of such entity, whether by contract or - otherwise, or (ii) ownership of fifty percent (50%) or more of the - outstanding shares, or (iii) beneficial ownership of such entity. - - "You" (or "Your") shall mean an individual or Legal Entity - exercising permissions granted by this License. - - "Source" form shall mean the preferred form for making modifications, - including but not limited to software source code, documentation - source, and configuration files. - - "Object" form shall mean any form resulting from mechanical - transformation or translation of a Source form, including but - not limited to compiled object code, generated documentation, - and conversions to other media types. - - "Work" shall mean the work of authorship, whether in Source or - Object form, made available under the License, as indicated by a - copyright notice that is included in or attached to the work - (an example is provided in the Appendix below). - - "Derivative Works" shall mean any work, whether in Source or Object - form, that is based on (or derived from) the Work and for which the - editorial revisions, annotations, elaborations, or other modifications - represent, as a whole, an original work of authorship. For the purposes - of this License, Derivative Works shall not include works that remain - separable from, or merely link (or bind by name) to the interfaces of, - the Work and Derivative Works thereof. - - "Contribution" shall mean any work of authorship, including - the original version of the Work and any modifications or additions - to that Work or Derivative Works thereof, that is intentionally - submitted to Licensor for inclusion in the Work by the copyright owner - or by an individual or Legal Entity authorized to submit on behalf of - the copyright owner. For the purposes of this definition, "submitted" - means any form of electronic, verbal, or written communication sent - to the Licensor or its representatives, including but not limited to - communication on electronic mailing lists, source code control systems, - and issue tracking systems that are managed by, or on behalf of, the - Licensor for the purpose of discussing and improving the Work, but - excluding communication that is conspicuously marked or otherwise - designated in writing by the copyright owner as "Not a Contribution." - - "Contributor" shall mean Licensor and any individual or Legal Entity - on behalf of whom a Contribution has been received by Licensor and - subsequently incorporated within the Work. - - 2. Grant of Copyright License. Subject to the terms and conditions of - this License, each Contributor hereby grants to You a perpetual, - worldwide, non-exclusive, no-charge, royalty-free, irrevocable - copyright license to reproduce, prepare Derivative Works of, - publicly display, publicly perform, sublicense, and distribute the - Work and such Derivative Works in Source or Object form. - - 3. Grant of Patent License. Subject to the terms and conditions of - this License, each Contributor hereby grants to You a perpetual, - worldwide, non-exclusive, no-charge, royalty-free, irrevocable - (except as stated in this section) patent license to make, have made, - use, offer to sell, sell, import, and otherwise transfer the Work, - where such license applies only to those patent claims licensable - by such Contributor that are necessarily infringed by their - Contribution(s) alone or by combination of their Contribution(s) - with the Work to which such Contribution(s) was submitted. If You - institute patent litigation against any entity (including a - cross-claim or counterclaim in a lawsuit) alleging that the Work - or a Contribution incorporated within the Work constitutes direct - or contributory patent infringement, then any patent licenses - granted to You under this License for that Work shall terminate - as of the date such litigation is filed. - - 4. Redistribution. You may reproduce and distribute copies of the - Work or Derivative Works thereof in any medium, with or without - modifications, and in Source or Object form, provided that You - meet the following conditions: - - (a) You must give any other recipients of the Work or - Derivative Works a copy of this License; and - - (b) You must cause any modified files to carry prominent notices - stating that You changed the files; and - - (c) You must retain, in the Source form of any Derivative Works - that You distribute, all copyright, patent, trademark, and - attribution notices from the Source form of the Work, - excluding those notices that do not pertain to any part of - the Derivative Works; and - - (d) If the Work includes a "NOTICE" text file as part of its - distribution, then any Derivative Works that You distribute must - include a readable copy of the attribution notices contained - within such NOTICE file, excluding those notices that do not - pertain to any part of the Derivative Works, in at least one - of the following places: within a NOTICE text file distributed - as part of the Derivative Works; within the Source form or - documentation, if provided along with the Derivative Works; or, - within a display generated by the Derivative Works, if and - wherever such third-party notices normally appear. The contents - of the NOTICE file are for informational purposes only and - do not modify the License. You may add Your own attribution - notices within Derivative Works that You distribute, alongside - or as an addendum to the NOTICE text from the Work, provided - that such additional attribution notices cannot be construed - as modifying the License. - - You may add Your own copyright statement to Your modifications and - may provide additional or different license terms and conditions - for use, reproduction, or distribution of Your modifications, or - for any such Derivative Works as a whole, provided Your use, - reproduction, and distribution of the Work otherwise complies with - the conditions stated in this License. - - 5. Submission of Contributions. Unless You explicitly state otherwise, - any Contribution intentionally submitted for inclusion in the Work - by You to the Licensor shall be under the terms and conditions of - this License, without any additional terms or conditions. - Notwithstanding the above, nothing herein shall supersede or modify - the terms of any separate license agreement you may have executed - with Licensor regarding such Contributions. - - 6. Trademarks. This License does not grant permission to use the trade - names, trademarks, service marks, or product names of the Licensor, - except as required for reasonable and customary use in describing the - origin of the Work and reproducing the content of the NOTICE file. - - 7. Disclaimer of Warranty. Unless required by applicable law or - agreed to in writing, Licensor provides the Work (and each - Contributor provides its Contributions) on an "AS IS" BASIS, - WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or - implied, including, without limitation, any warranties or conditions - of TITLE, NON-INFRINGEMENT, MERCHANTABILITY, or FITNESS FOR A - PARTICULAR PURPOSE. You are solely responsible for determining the - appropriateness of using or redistributing the Work and assume any - risks associated with Your exercise of permissions under this License. - - 8. Limitation of Liability. In no event and under no legal theory, - whether in tort (including negligence), contract, or otherwise, - unless required by applicable law (such as deliberate and grossly - negligent acts) or agreed to in writing, shall any Contributor be - liable to You for damages, including any direct, indirect, special, - incidental, or consequential damages of any character arising as a - result of this License or out of the use or inability to use the - Work (including but not limited to damages for loss of goodwill, - work stoppage, computer failure or malfunction, or any and all - other commercial damages or losses), even if such Contributor - has been advised of the possibility of such damages. - - 9. Accepting Warranty or Additional Liability. While redistributing - the Work or Derivative Works thereof, You may choose to offer, - and charge a fee for, acceptance of support, warranty, indemnity, - or other liability obligations and/or rights consistent with this - License. However, in accepting such obligations, You may act only - on Your own behalf and on Your sole responsibility, not on behalf - of any other Contributor, and only if You agree to indemnify, - defend, and hold each Contributor harmless for any liability - incurred by, or claims asserted against, such Contributor by reason - of your accepting any such warranty or additional liability. - - END OF TERMS AND CONDITIONS - - APPENDIX: How to apply the Apache License to your work. - - To apply the Apache License to your work, attach the following - boilerplate notice, with the fields enclosed by brackets "[]" - replaced with your own identifying information. (Don't include - the brackets!) The text should be enclosed in the appropriate - comment syntax for the file format. We also recommend that a - file or class name and description of purpose be included on the - same "printed page" as the copyright notice for easier - identification within third-party archives. - - Copyright [yyyy] [name of copyright owner] - - 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. diff --git a/vendor/k8s.io/kubelet/pkg/cri/streaming/.import-restrictions b/vendor/k8s.io/kubelet/pkg/cri/streaming/.import-restrictions deleted file mode 100644 index 10215ff92..000000000 --- a/vendor/k8s.io/kubelet/pkg/cri/streaming/.import-restrictions +++ /dev/null @@ -1,5 +0,0 @@ -rules: - # prevent exposing internal api in streaming packages - - selectorRegexp: k8s[.]io/kubernetes - allowedPrefixes: - - k8s.io/kubernetes/pkg/kubelet/cri diff --git a/vendor/modules.txt b/vendor/modules.txt index e80978dd8..0f38f5f62 100644 --- a/vendor/modules.txt +++ b/vendor/modules.txt @@ -582,6 +582,10 @@ go.opentelemetry.io/proto/otlp/collector/trace/v1 go.opentelemetry.io/proto/otlp/common/v1 go.opentelemetry.io/proto/otlp/resource/v1 go.opentelemetry.io/proto/otlp/trace/v1 +# go.uber.org/goleak v1.3.0 +## explicit; go 1.20 +go.uber.org/goleak +go.uber.org/goleak/internal/stack # golang.org/x/crypto v0.36.0 ## explicit; go 1.23.0 golang.org/x/crypto/cast5 @@ -825,9 +829,6 @@ k8s.io/apimachinery/pkg/version k8s.io/apimachinery/pkg/watch k8s.io/apimachinery/third_party/forked/golang/netutil k8s.io/apimachinery/third_party/forked/golang/reflect -# k8s.io/apiserver v0.32.3 -## explicit; go 1.23.0 -k8s.io/apiserver/pkg/endpoints/responsewriter # k8s.io/client-go v0.32.3 ## explicit; go 1.23.0 k8s.io/client-go/features @@ -864,11 +865,6 @@ k8s.io/klog/v2/internal/dbg k8s.io/klog/v2/internal/serialize k8s.io/klog/v2/internal/severity k8s.io/klog/v2/internal/sloghandler -# k8s.io/kubelet v0.32.3 -## explicit; go 1.23.0 -k8s.io/kubelet/pkg/cri/streaming -k8s.io/kubelet/pkg/cri/streaming/portforward -k8s.io/kubelet/pkg/cri/streaming/remotecommand # k8s.io/utils v0.0.0-20241104100929-3ea5e8cea738 ## explicit; go 1.18 k8s.io/utils/clock