From dfdf07bb531aa8a397f3c74e5eec851130325971 Mon Sep 17 00:00:00 2001 From: Patrick Ohly Date: Wed, 4 Dec 2024 15:21:11 +0100 Subject: [PATCH] client-go portforward + apiserver proxy: structured, contextual logging When debugging, it helps to keep output from different connections separate. This can be done with contextual logging and using different loggers for each connection. Cancellation is handled separately for requests. Therefore the new APIs only add support for passing a logger instance. --- .../apiserver/pkg/util/proxy/streamtunnel.go | 38 ++++++++------ .../pkg/util/proxy/translatinghandler.go | 2 +- .../tools/portforward/fallback_dialer.go | 1 + .../tools/portforward/fallback_dialer_test.go | 2 +- .../tools/portforward/portforward.go | 52 +++++++++++++------ .../tools/portforward/portforward_test.go | 1 + .../tools/portforward/tunneling_connection.go | 42 ++++++++++----- .../portforward/tunneling_connection_test.go | 9 +++- .../tools/portforward/tunneling_dialer.go | 23 ++++++-- 9 files changed, 116 insertions(+), 54 deletions(-) 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)