diff --git a/staging/src/k8s.io/apiserver/pkg/util/proxy/streamtunnel.go b/staging/src/k8s.io/apiserver/pkg/util/proxy/streamtunnel.go index 7a7b92adaf7..fa937f22fbc 100644 --- a/staging/src/k8s.io/apiserver/pkg/util/proxy/streamtunnel.go +++ b/staging/src/k8s.io/apiserver/pkg/util/proxy/streamtunnel.go @@ -60,7 +60,8 @@ func NewTunnelingHandler(upgradeHandler http.Handler) *TunnelingHandler { // case the upstream upgrade fails, we delegate communication to the passed // in "w" ResponseWriter. func (h *TunnelingHandler) ServeHTTP(w http.ResponseWriter, req *http.Request) { - klog.V(4).Infoln("TunnelingHandler ServeHTTP") + logger := klog.FromContext(req.Context()) + logger.V(4).Info("TunnelingHandler ServeHTTP") spdyProtocols := spdyProtocolsFromWebsocketProtocols(req) if len(spdyProtocols) == 0 { @@ -75,10 +76,12 @@ func (h *TunnelingHandler) ServeHTTP(w http.ResponseWriter, req *http.Request) { // and the "conn" is hijacked and used in the subsequent upgradeHandler, or // the upgrade failed, and "w" is the delegate used for the non-upgrade response. writer := &tunnelingResponseWriter{ + logger: logger, // "w" is used in the non-upgrade error cases called in the upgradeHandler. w: w, // "conn" is returned in the successful upgrade case when hijacked in the upgradeHandler. conn: &headerInterceptingConn{ + logger: logger, initializableConn: &tunnelingWebsocketUpgraderConn{ w: w, req: req, @@ -86,7 +89,7 @@ func (h *TunnelingHandler) ServeHTTP(w http.ResponseWriter, req *http.Request) { }, } - klog.V(4).Infoln("Tunnel spdy through websockets using the UpgradeAwareProxy") + logger.V(4).Info("Tunnel spdy through websockets using the UpgradeAwareProxy") h.upgradeHandler.ServeHTTP(writer, spdyRequest) } @@ -131,6 +134,7 @@ var _ http.Hijacker = &tunnelingResponseWriter{} // Once Write or WriteHeader is called, Hijack returns an error. // Once Hijack is called, Write, WriteHeader, and Hijack return errors. type tunnelingResponseWriter struct { + logger klog.Logger // w is used to delegate Header(), WriteHeader(), and Write() calls w http.ResponseWriter // conn is returned from Hijack() @@ -150,15 +154,15 @@ func (w *tunnelingResponseWriter) Hijack() (net.Conn, *bufio.ReadWriter, error) w.mu.Lock() defer w.mu.Unlock() if w.written { - klog.Errorf("Hijack called after write") + w.logger.Error(nil, "Hijack called after write") return nil, nil, errors.New("connection has already been written to") } if w.hijacked { - klog.Errorf("Hijack called after hijack") + w.logger.Error(nil, "Hijack called after hijack") return nil, nil, errors.New("connection has already been hijacked") } w.hijacked = true - klog.V(6).Infof("Hijack returning websocket tunneling net.Conn") + w.logger.V(6).Info("Hijack returning websocket tunneling net.Conn") return w.conn, nil, nil } @@ -172,7 +176,7 @@ func (w *tunnelingResponseWriter) Write(p []byte) (int, error) { w.mu.Lock() defer w.mu.Unlock() if w.hijacked { - klog.Errorf("Write called after hijack") + w.logger.Error(nil, "Write called after hijack") return 0, http.ErrHijacked } w.written = true @@ -184,18 +188,18 @@ func (w *tunnelingResponseWriter) WriteHeader(statusCode int) { w.mu.Lock() defer w.mu.Unlock() if w.written { - klog.Errorf("WriteHeader called after write") + w.logger.Error(nil, "WriteHeader called after write") return } if w.hijacked { - klog.Errorf("WriteHeader called after hijack") + w.logger.Error(nil, "WriteHeader called after hijack") return } w.written = true if statusCode == http.StatusSwitchingProtocols { // 101 upgrade responses must come via the hijacked connection, not WriteHeader - klog.Errorf("WriteHeader called with 101 upgrade") + w.logger.Error(nil, "WriteHeader called with 101 upgrade") http.Error(w.w, "unexpected upgrade", http.StatusInternalServerError) return } @@ -208,6 +212,7 @@ func (w *tunnelingResponseWriter) WriteHeader(statusCode int) { // HTTP response status/headers from the upstream SPDY connection, then use // that to decide how to initialize the delegate connection for writes. type headerInterceptingConn struct { + logger klog.Logger // initializableConn is delegated to for all net.Conn methods. // initializableConn.Write() is not called until response headers have been read // and initializableConn#InitializeWrite() has been called with the result. @@ -274,7 +279,7 @@ func (h *headerInterceptingConn) Write(b []byte) (int, error) { } resp, err := http.ReadResponse(bufio.NewReader(bytes.NewReader(headerBytes)), nil) if err != nil { - klog.Errorf("invalid headers: %v", err) + h.logger.Error(err, "Invalid headers") h.initializeErr = err return len(b), err } @@ -324,11 +329,12 @@ func (u *tunnelingWebsocketUpgraderConn) InitializeWrite(backendResponse *http.R return u.err } + logger := klog.FromContext(u.req.Context()) if backendResponse.StatusCode == http.StatusSwitchingProtocols { connectionHeader := strings.ToLower(backendResponse.Header.Get(httpstream.HeaderConnection)) upgradeHeader := strings.ToLower(backendResponse.Header.Get(httpstream.HeaderUpgrade)) if !strings.Contains(connectionHeader, strings.ToLower(httpstream.HeaderUpgrade)) || !strings.Contains(upgradeHeader, strings.ToLower(spdy.HeaderSpdy31)) { - klog.Errorf("unable to upgrade: missing upgrade headers in response: %#v", backendResponse.Header) + logger.Error(nil, "Unable to upgrade: missing upgrade headers in response", "headers", backendResponse.Header) u.err = fmt.Errorf("unable to upgrade: missing upgrade headers in response") metrics.IncStreamTunnelRequest(context.Background(), strconv.Itoa(http.StatusInternalServerError)) http.Error(u.w, u.err.Error(), http.StatusInternalServerError) @@ -351,26 +357,26 @@ func (u *tunnelingWebsocketUpgraderConn) InitializeWrite(backendResponse *http.R } conn, err := upgrader.Upgrade(u.w, u.req, nil) if err != nil { - klog.Errorf("error upgrading websocket connection: %v", err) + logger.Error(err, "Error upgrading websocket connection") metrics.IncStreamTunnelRequest(context.Background(), strconv.Itoa(http.StatusInternalServerError)) u.err = err return u.err } - klog.V(4).Infof("websocket connection created: %s", conn.Subprotocol()) + logger.V(4).Info("Websocket connection created", "protocol", conn.Subprotocol()) metrics.IncStreamTunnelRequest(context.Background(), strconv.Itoa(http.StatusSwitchingProtocols)) - u.conn = portforward.NewTunnelingConnection("server", conn) + u.conn = portforward.NewTunnelingConnectionWithLogger(klog.LoggerWithName(logger, "server"), conn) return nil } // anything other than an upgrade should pass through the backend response - klog.Errorf("SPDY upgrade failed: %s", backendResponse.Status) + logger.Error(nil, "SPDY upgrade failed", "status", backendResponse.Status) metrics.IncStreamTunnelRequest(context.Background(), strconv.Itoa(backendResponse.StatusCode)) // try to hijack conn, _, err = u.w.(http.Hijacker).Hijack() if err != nil { - klog.Errorf("Unable to hijack response: %v", err) + logger.Error(err, "Unable to hijack response") u.err = err return u.err } diff --git a/staging/src/k8s.io/apiserver/pkg/util/proxy/translatinghandler.go b/staging/src/k8s.io/apiserver/pkg/util/proxy/translatinghandler.go index 6f6c0088241..cf38875a4fc 100644 --- a/staging/src/k8s.io/apiserver/pkg/util/proxy/translatinghandler.go +++ b/staging/src/k8s.io/apiserver/pkg/util/proxy/translatinghandler.go @@ -43,7 +43,7 @@ func NewTranslatingHandler(delegate http.Handler, translator http.Handler, shoul func (t *translatingHandler) ServeHTTP(w http.ResponseWriter, req *http.Request) { if t.shouldTranslate(req) { - klog.V(4).Infof("request handled by translator proxy") + klog.FromContext(req.Context()).V(4).Info("Request handled by translator proxy") t.translator.ServeHTTP(w, req) return } diff --git a/staging/src/k8s.io/client-go/tools/portforward/fallback_dialer.go b/staging/src/k8s.io/client-go/tools/portforward/fallback_dialer.go index 7fcc2492bfe..e5e0f0866dd 100644 --- a/staging/src/k8s.io/client-go/tools/portforward/fallback_dialer.go +++ b/staging/src/k8s.io/client-go/tools/portforward/fallback_dialer.go @@ -50,6 +50,7 @@ func NewFallbackDialer(primary, secondary httpstream.Dialer, shouldFallback func func (f *FallbackDialer) Dial(protocols ...string) (httpstream.Connection, string, error) { conn, version, err := f.primary.Dial(protocols...) if err != nil && f.shouldFallback(err) { + //nolint:logcheck // This code is only used by kubectl where contextual logging is not that useful. klog.V(4).Infof("fallback to secondary dialer from primary dialer err: %v", err) return f.secondary.Dial(protocols...) } diff --git a/staging/src/k8s.io/client-go/tools/portforward/fallback_dialer_test.go b/staging/src/k8s.io/client-go/tools/portforward/fallback_dialer_test.go index 70583958ea4..3e3b27204ce 100644 --- a/staging/src/k8s.io/client-go/tools/portforward/fallback_dialer_test.go +++ b/staging/src/k8s.io/client-go/tools/portforward/fallback_dialer_test.go @@ -40,7 +40,7 @@ func TestFallbackDialer(t *testing.T) { assert.Equal(t, primaryProtocol, negotiated, "primary negotiated protocol returned") require.NoError(t, err, "error from primary dialer should be nil") // If primary dialer error is upgrade error, then fallback returning secondary dial response. - primary = &fakeDialer{dialed: false, negotiatedProtocol: primaryProtocol, err: &httpstream.UpgradeFailureError{}} + primary = &fakeDialer{dialed: false, negotiatedProtocol: primaryProtocol, err: &httpstream.UpgradeFailureError{Cause: fmt.Errorf("fake error")}} secondary = &fakeDialer{dialed: false, negotiatedProtocol: secondaryProtocol} fallbackDialer = NewFallbackDialer(primary, secondary, httpstream.IsUpgradeFailure) _, negotiated, err = fallbackDialer.Dial(protocols...) diff --git a/staging/src/k8s.io/client-go/tools/portforward/portforward.go b/staging/src/k8s.io/client-go/tools/portforward/portforward.go index 126c14e8fa0..c1e89a92b68 100644 --- a/staging/src/k8s.io/client-go/tools/portforward/portforward.go +++ b/staging/src/k8s.io/client-go/tools/portforward/portforward.go @@ -17,6 +17,7 @@ limitations under the License. package portforward import ( + "context" "errors" "fmt" "io" @@ -30,6 +31,8 @@ import ( v1 "k8s.io/api/core/v1" "k8s.io/apimachinery/pkg/util/httpstream" "k8s.io/apimachinery/pkg/util/runtime" + "k8s.io/apimachinery/pkg/util/wait" + "k8s.io/klog/v2" netutils "k8s.io/utils/net" ) @@ -52,6 +55,7 @@ type PortForwarder struct { ports []ForwardedPort stopChan <-chan struct{} + logger klog.Logger dialer httpstream.Dialer streamConn httpstream.Connection listeners []io.Closer @@ -165,7 +169,14 @@ func New(dialer httpstream.Dialer, ports []string, stopChan <-chan struct{}, rea } // NewOnAddresses creates a new PortForwarder with custom listen addresses. +// +//logcheck:context // NewOnAddressesWithContext should be used instead of NewOnAddresses in code which supports contextual logging. func NewOnAddresses(dialer httpstream.Dialer, addresses []string, ports []string, stopChan <-chan struct{}, readyChan chan struct{}, out, errOut io.Writer) (*PortForwarder, error) { + return NewOnAddressesWithContext(wait.ContextForChannel(stopChan), dialer, addresses, ports, readyChan, out, errOut) +} + +// NewOnAddressesWithContext creates a new PortForwarder with custom listen addresses. +func NewOnAddressesWithContext(ctx context.Context, dialer httpstream.Dialer, addresses []string, ports []string, readyChan chan struct{}, out, errOut io.Writer) (*PortForwarder, error) { if len(addresses) == 0 { return nil, errors.New("you must specify at least 1 address") } @@ -181,10 +192,11 @@ func NewOnAddresses(dialer httpstream.Dialer, addresses []string, ports []string return nil, err } return &PortForwarder{ + logger: klog.FromContext(ctx), dialer: dialer, addresses: parsedAddresses, ports: parsedPorts, - stopChan: stopChan, + stopChan: ctx.Done(), Ready: readyChan, out: out, errOut: errOut, @@ -319,7 +331,7 @@ func (pf *PortForwarder) waitForConnection(listener net.Listener, port Forwarded if err != nil { // TODO consider using something like https://github.com/hydrogen18/stoppableListener? if !strings.Contains(strings.ToLower(err.Error()), networkClosedError) { - runtime.HandleError(fmt.Errorf("error accepting connection on port %d: %v", port.Local, err)) + runtime.HandleErrorWithLogger(pf.logger, err, "Error accepting connection", "localPort", port.Local) } return } @@ -354,21 +366,23 @@ func (pf *PortForwarder) handleConnection(conn net.Conn, port ForwardedPort) { headers.Set(v1.PortForwardRequestIDHeader, strconv.Itoa(requestID)) errorStream, err := pf.streamConn.CreateStream(headers) if err != nil { - runtime.HandleError(fmt.Errorf("error creating error stream for port %d -> %d: %v", port.Local, port.Remote, err)) + runtime.HandleErrorWithLogger(pf.logger, err, "Error creating error stream", "localPort", port.Local, "remotePort", port.Remote) return } // we're not writing to this stream errorStream.Close() defer pf.streamConn.RemoveStreams(errorStream) - errorChan := make(chan error) + type readAllResult struct { + message []byte + err error + } + errorChan := make(chan readAllResult) go func() { message, err := io.ReadAll(errorStream) - switch { - case err != nil: - errorChan <- fmt.Errorf("error reading from error stream for port %d -> %d: %v", port.Local, port.Remote, err) - case len(message) > 0: - errorChan <- fmt.Errorf("an error occurred forwarding %d -> %d: %v", port.Local, port.Remote, string(message)) + errorChan <- readAllResult{ + message: message, + err: err, } close(errorChan) }() @@ -377,7 +391,7 @@ func (pf *PortForwarder) handleConnection(conn net.Conn, port ForwardedPort) { headers.Set(v1.StreamType, v1.StreamTypeData) dataStream, err := pf.streamConn.CreateStream(headers) if err != nil { - runtime.HandleError(fmt.Errorf("error creating forwarding stream for port %d -> %d: %v", port.Local, port.Remote, err)) + runtime.HandleErrorWithLogger(pf.logger, err, "Error creating forwarding stream", "localPort", port.Local, "remotePort", port.Remote) return } defer pf.streamConn.RemoveStreams(dataStream) @@ -388,7 +402,7 @@ func (pf *PortForwarder) handleConnection(conn net.Conn, port ForwardedPort) { go func() { // Copy from the remote side to the local port. if _, err := io.Copy(conn, dataStream); err != nil && !strings.Contains(strings.ToLower(err.Error()), networkClosedError) { - runtime.HandleError(fmt.Errorf("error copying from remote stream to local connection: %v", err)) + runtime.HandleErrorWithLogger(pf.logger, err, "Error copying from remote stream to local connection", "localPort", port.Local, "remotePort", port.Remote) } // inform the select below that the remote copy is done @@ -401,7 +415,7 @@ func (pf *PortForwarder) handleConnection(conn net.Conn, port ForwardedPort) { // Copy from the local port to the remote side. if _, err := io.Copy(dataStream, conn); err != nil && !strings.Contains(strings.ToLower(err.Error()), networkClosedError) { - runtime.HandleError(fmt.Errorf("error copying from local connection to remote stream: %v", err)) + runtime.HandleErrorWithLogger(pf.logger, err, "Error copying from local connection to remote stream", "localPort", port.Local, "remotePort", port.Remote) // break out of the select below without waiting for the other copy to finish close(localError) } @@ -418,10 +432,14 @@ func (pf *PortForwarder) handleConnection(conn net.Conn, port ForwardedPort) { // the blocking data will affect errorStream and cause <-errorChan to block indefinitely. _ = dataStream.Reset() - // always expect something on errorChan (it may be nil) - err = <-errorChan - if err != nil { - runtime.HandleError(err) + // always expect something on errorChan (it may be empty) + errResult := <-errorChan + switch { + case errResult.err != nil: + runtime.HandleErrorWithLogger(pf.logger, errResult.err, "Error reading from error stream", "localPort", port.Local, "remotePort", port.Remote) + pf.streamConn.Close() + case len(errResult.message) > 0: + runtime.HandleErrorWithLogger(pf.logger, errors.New(string(errResult.message)), "An error occurred forwarding", "localPort", port.Local, "remotePort", port.Remote) pf.streamConn.Close() } } @@ -431,7 +449,7 @@ func (pf *PortForwarder) Close() { // stop all listeners for _, l := range pf.listeners { if err := l.Close(); err != nil { - runtime.HandleError(fmt.Errorf("error closing listener: %v", err)) + runtime.HandleErrorWithLogger(pf.logger, err, "Error closing listener") } } } diff --git a/staging/src/k8s.io/client-go/tools/portforward/portforward_test.go b/staging/src/k8s.io/client-go/tools/portforward/portforward_test.go index 075a22e62a1..96db9dff122 100644 --- a/staging/src/k8s.io/client-go/tools/portforward/portforward_test.go +++ b/staging/src/k8s.io/client-go/tools/portforward/portforward_test.go @@ -299,6 +299,7 @@ func TestParsePortsAndNew(t *testing.T) { var pf *PortForwarder if len(test.addresses) > 0 { + //nolint:logcheck // Testing the original function. pf, err = NewOnAddresses(dialer, test.addresses, test.input, expectedStopChan, readyChan, os.Stdout, os.Stderr) } else { pf, err = New(dialer, test.input, expectedStopChan, readyChan, os.Stdout, os.Stderr) diff --git a/staging/src/k8s.io/client-go/tools/portforward/tunneling_connection.go b/staging/src/k8s.io/client-go/tools/portforward/tunneling_connection.go index a9c9b18fd60..b9030691966 100644 --- a/staging/src/k8s.io/client-go/tools/portforward/tunneling_connection.go +++ b/staging/src/k8s.io/client-go/tools/portforward/tunneling_connection.go @@ -34,7 +34,7 @@ var _ net.Conn = &TunnelingConnection{} // TunnelingConnection implements the "httpstream.Connection" interface, wrapping // a websocket connection that tunnels SPDY. type TunnelingConnection struct { - name string + logger klog.Logger conn *gwebsocket.Conn inProgressMessage io.Reader closeOnce sync.Once @@ -42,29 +42,46 @@ type TunnelingConnection struct { // NewTunnelingConnection wraps the passed gorilla/websockets connection // with the TunnelingConnection struct (implementing net.Conn). +// The name is added to all log entries with [klog.LoggerWithName]. +// +//logcheck:context // NewTunnelingConnectionWithLogger should be used instead of NewTunnelingConnection in code which supports contextual logging. func NewTunnelingConnection(name string, conn *gwebsocket.Conn) *TunnelingConnection { + logger := klog.LoggerWithName(klog.Background(), name) + return NewTunnelingConnectionWithLogger(logger, conn) +} + +// NewTunnelingConnectionWithLogger is a variant of NewTunnelingConnection where +// the caller is in control of logging. For example, [klog.LoggerWithName] can be used +// to add a common name for all log entries to identify the connection. +func NewTunnelingConnectionWithLogger(logger klog.Logger, conn *gwebsocket.Conn) *TunnelingConnection { return &TunnelingConnection{ - name: name, - conn: conn, + logger: logger, + conn: conn, } } // Read implements "io.Reader" interface, reading from the stored connection // into the passed buffer "p". Returns the number of bytes read and an error. // Can keep track of the "inProgress" messsage from the tunneled connection. -func (c *TunnelingConnection) Read(p []byte) (int, error) { - klog.V(7).Infof("%s: tunneling connection read...", c.name) - defer klog.V(7).Infof("%s: tunneling connection read...complete", c.name) +func (c *TunnelingConnection) Read(p []byte) (len int, err error) { + c.logger.V(7).Info("Tunneling connection read...") + defer func() { + if loggerV := c.logger.V(8); loggerV.Enabled() { + loggerV.Info("Tunneling connection read...complete", "length", len, "data", p[:len], "err", err) + } else { + c.logger.V(7).Info("Tunneling connection read...complete") + } + }() for { if c.inProgressMessage == nil { - klog.V(8).Infof("%s: tunneling connection read before NextReader()...", c.name) + c.logger.V(8).Info("Tunneling connection read before NextReader()...") messageType, nextReader, err := c.conn.NextReader() if err != nil { closeError := &gwebsocket.CloseError{} if errors.As(err, &closeError) && closeError.Code == gwebsocket.CloseNormalClosure { return 0, io.EOF } - klog.V(4).Infof("%s:tunneling connection NextReader() error: %v", c.name, err) + c.logger.V(4).Info("Tunneling connection NextReader() failed", "err", err) return 0, err } if messageType != gwebsocket.BinaryMessage { @@ -72,12 +89,11 @@ func (c *TunnelingConnection) Read(p []byte) (int, error) { } c.inProgressMessage = nextReader } - klog.V(8).Infof("%s: tunneling connection read in progress message...", c.name) + c.logger.V(8).Info("Tunneling connection read in progress...") i, err := c.inProgressMessage.Read(p) if i == 0 && err == io.EOF { c.inProgressMessage = nil } else { - klog.V(8).Infof("%s: read %d bytes, error=%v, bytes=% X", c.name, i, err, p[:i]) return i, err } } @@ -87,8 +103,8 @@ func (c *TunnelingConnection) Read(p []byte) (int, error) { // byte array "p" into the stored tunneled connection. Returns the number // of bytes written and an error. func (c *TunnelingConnection) Write(p []byte) (n int, err error) { - klog.V(7).Infof("%s: write: %d bytes, bytes=% X", c.name, len(p), p) - defer klog.V(7).Infof("%s: tunneling connection write...complete", c.name) + c.logger.V(7).Info("Tunneling connection write", "length", len(p), "data", p) + defer c.logger.V(7).Info("Tunneling connection write...complete") w, err := c.conn.NextWriter(gwebsocket.BinaryMessage) if err != nil { return 0, err @@ -111,7 +127,7 @@ func (c *TunnelingConnection) Write(p []byte) (n int, err error) { func (c *TunnelingConnection) Close() error { var err error c.closeOnce.Do(func() { - klog.V(7).Infof("%s: tunneling connection Close()...", c.name) + c.logger.V(7).Info("Tunneling connection Close()...") // Signal other endpoint that websocket connection is closing; ignore error. normalCloseMsg := gwebsocket.FormatCloseMessage(gwebsocket.CloseNormalClosure, "") writeControlErr := c.conn.WriteControl(gwebsocket.CloseMessage, normalCloseMsg, time.Now().Add(time.Second)) diff --git a/staging/src/k8s.io/client-go/tools/portforward/tunneling_connection_test.go b/staging/src/k8s.io/client-go/tools/portforward/tunneling_connection_test.go index afbdb6b0cec..0852404c67d 100644 --- a/staging/src/k8s.io/client-go/tools/portforward/tunneling_connection_test.go +++ b/staging/src/k8s.io/client-go/tools/portforward/tunneling_connection_test.go @@ -36,8 +36,13 @@ import ( "k8s.io/apimachinery/pkg/util/wait" "k8s.io/client-go/rest" "k8s.io/client-go/transport/websocket" + "k8s.io/klog/v2" ) +func init() { + klog.InitFlags(nil) +} + func TestTunnelingConnection_ReadWriteClose(t *testing.T) { // Stream channel that will receive streams created on upstream SPDY server. streamChan := make(chan httpstream.Stream) @@ -60,7 +65,7 @@ func TestTunnelingConnection_ReadWriteClose(t *testing.T) { t.Errorf("Not acceptable agreement Subprotocol: %v", conn.Subprotocol()) return } - tunnelingConn := NewTunnelingConnection("server", conn) + tunnelingConn := NewTunnelingConnectionWithLogger(klog.LoggerWithName(klog.Background(), "server"), conn) spdyConn, err := spdy.NewServerConnection(tunnelingConn, justQueueStream(streamChan)) if err != nil { t.Errorf("unexpected error %v", err) @@ -73,6 +78,7 @@ func TestTunnelingConnection_ReadWriteClose(t *testing.T) { // Dial the client tunneling connection to the tunneling server. url, err := url.Parse(tunnelingServer.URL) require.NoError(t, err) + //nolint:logcheck // Intentionally uses the old API. dialer, err := NewSPDYOverWebsocketDialer(url, &rest.Config{Host: url.Host}) require.NoError(t, err) spdyClient, protocol, err := dialer.Dial(constants.PortForwardV1Name) @@ -205,6 +211,7 @@ func dialForTunnelingConnection(url *url.URL) (*TunnelingConnection, error) { if err != nil { return nil, err } + //nolint:logcheck // Intentionally uses the old API. return NewTunnelingConnection("client", conn), nil } diff --git a/staging/src/k8s.io/client-go/tools/portforward/tunneling_dialer.go b/staging/src/k8s.io/client-go/tools/portforward/tunneling_dialer.go index 2bef5ecd720..ce3f9c503f2 100644 --- a/staging/src/k8s.io/client-go/tools/portforward/tunneling_dialer.go +++ b/staging/src/k8s.io/client-go/tools/portforward/tunneling_dialer.go @@ -17,6 +17,7 @@ limitations under the License. package portforward import ( + "context" "fmt" "net/http" "net/url" @@ -35,6 +36,7 @@ const PingPeriod = 10 * time.Second // tunnelingDialer implements "httpstream.Dial" interface type tunnelingDialer struct { + logger klog.Logger url *url.URL transport http.RoundTripper holder websocket.ConnectionHolder @@ -43,12 +45,22 @@ type tunnelingDialer struct { // NewTunnelingDialer creates and returns the tunnelingDialer structure which implemements the "httpstream.Dialer" // interface. The dialer can upgrade a websocket request, creating a websocket connection. This function // returns an error if one occurs. +// +//logcheck:context // NewSPDYOverWebsocketDialerWithLogger should be used instead of NewSPDYOverWebsocketDialer in code which supports contextual logging. func NewSPDYOverWebsocketDialer(url *url.URL, config *restclient.Config) (httpstream.Dialer, error) { + return NewSPDYOverWebsocketDialerWithLogger(klog.Background(), url, config) +} + +// NewTunnelingDialer creates and returns the tunnelingDialer structure which implemements the "httpstream.Dialer" +// interface. The dialer can upgrade a websocket request, creating a websocket connection. This function +// returns an error if one occurs. +func NewSPDYOverWebsocketDialerWithLogger(logger klog.Logger, url *url.URL, config *restclient.Config) (httpstream.Dialer, error) { transport, holder, err := websocket.RoundTripperFor(config) if err != nil { return nil, err } return &tunnelingDialer{ + logger: logger, url: url, transport: transport, holder: holder, @@ -59,9 +71,10 @@ func NewSPDYOverWebsocketDialer(url *url.URL, config *restclient.Config) (httpst // containing a WebSockets connection (which implements "net.Conn"). Also // returns the protocol negotiated, or an error. func (d *tunnelingDialer) Dial(protocols ...string) (httpstream.Connection, string, error) { - // There is no passed context, so skip the context when creating request for now. + // There is no passed context, so use the background context when creating request for now. + ctx := klog.NewContext(context.Background(), d.logger) // Websockets requires "GET" method: RFC 6455 Sec. 4.1 (page 17). - req, err := http.NewRequest("GET", d.url.String(), nil) + req, err := http.NewRequestWithContext(ctx, "GET", d.url.String(), nil) if err != nil { return nil, "", err } @@ -72,7 +85,7 @@ func (d *tunnelingDialer) Dial(protocols ...string) (httpstream.Connection, stri tunnelingProtocol := constants.WebsocketsSPDYTunnelingPrefix + protocol tunnelingProtocols = append(tunnelingProtocols, tunnelingProtocol) } - klog.V(4).Infoln("Before WebSocket Upgrade Connection...") + d.logger.V(4).Info("Before WebSocket Upgrade Connection...") conn, err := websocket.Negotiate(d.transport, d.holder, req, tunnelingProtocols...) if err != nil { return nil, "", err @@ -82,10 +95,10 @@ func (d *tunnelingDialer) Dial(protocols ...string) (httpstream.Connection, stri } protocol := conn.Subprotocol() protocol = strings.TrimPrefix(protocol, constants.WebsocketsSPDYTunnelingPrefix) - klog.V(4).Infof("negotiated protocol: %s", protocol) + d.logger.V(4).Info("Negotiation complete", "protocol", protocol) // Wrap the websocket connection which implements "net.Conn". - tConn := NewTunnelingConnection("client", conn) + tConn := NewTunnelingConnectionWithLogger(klog.LoggerWithName(d.logger, "client"), conn) // Create SPDY connection injecting the previously created tunneling connection. spdyConn, err := spdy.NewClientConnectionWithPings(tConn, PingPeriod)