dilim/containerd
1
0
Fork 0
mirror of https://github.com/containerd/containerd.git synced 2026-08-13 17:07:26 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #11863 from mikebrow/follow-up-reducing-k8s-vendor-footprint-streaming

Browse Source 
follow-up changes discussed at end of review creating these packages
This commit is contained in:
Samuel Karp
2025-05-15 17:19:16 +00:00
committed by GitHub
parent 30be5e1793 6c0d36b24e
commit 80ead0bf3d
8 changed files with 21 additions and 17 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
internal/cri/streamingserver/doc.go
View File

@@ -31,4 +31,4 @@ limitations under the License.
*/
// Package streamingserver contains server functions related the cri streaming api.
package streamingserver // import "k8s.io/kubelet/pkg/cri/streaming"
package streamingserver // this package was cloned from "k8s.io/kubelet/pkg/cri/streaming"

2
internal/cri/streamingserver/portforward/doc.go
View File

@@ -31,4 +31,4 @@ 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"
package portforward // this package was cloned from "k8s.io/kubelet/pkg/cri/streaming/portforward"

22
internal/cri/streamingserver/portforward/httpstream.go
View File

@@ -66,7 +66,7 @@ func handleHTTPStreams(req *http.Request, w http.ResponseWriter, portForwarder P
}
defer conn.Close()
log.L.Infof("Connection setting port forwarding streaming connection idle timeout connection idleTimeout %v", idleTimeout)
log.L.WithField("idleTimeout", idleTimeout).Info("Connection setting port forwarding streaming connection idle timeout")
conn.SetIdleTimeout(idleTimeout)
h := &httpStreamHandler{
@@ -137,11 +137,11 @@ func (h *httpStreamHandler) getStreamPair(requestID string) (*httpStreamPair, bo
defer h.streamPairsLock.Unlock()
if p, ok := h.streamPairs[requestID]; ok {
log.L.Infof("Connection request found existing stream pair connection request %v", requestID)
log.L.WithField("requestID", requestID).Info("Connection request found existing stream pair connection")
return p, false
}
log.L.Infof("Connection request creating new stream pair connection request %v", requestID)
log.L.WithField("requestID", requestID).Info("Connection request creating new stream pair connection")
p := newPortForwardPair(requestID)
h.streamPairs[requestID] = p
@@ -159,7 +159,7 @@ func (h *httpStreamHandler) monitorStreamPair(p *httpStreamPair, timeout <-chan
utilruntime.HandleError(err)
p.printError(err.Error())
case <-p.complete:
log.L.Infof("Connection request successfully received error and data streams connection request %v", p.requestID)
log.L.WithField("requestID", p.requestID).Info("Connection request successfully received error and data streams connection")
}
h.removeStreamPair(p.requestID)
}
@@ -217,7 +217,8 @@ func (h *httpStreamHandler) requestID(stream httpstream.Stream) string {
requestID = strconv.Itoa(int(stream.Identifier()) - 2)
}
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())
log.L.WithFields(log.Fields{"request": requestID, "streamType": streamType, "stream": stream.Identifier()}).Infof(
"Connection automatically assigning request ID from stream type and stream ID connection")
}
return requestID
}
@@ -226,7 +227,7 @@ 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() {
log.L.Infof("Connection waiting for port forward streams connection %v", h.conn)
log.L.Info("Connection waiting for port forward streams connection")
Loop:
for {
select {
@@ -236,7 +237,8 @@ Loop:
case stream := <-h.streamChan:
requestID := h.requestID(stream)
streamType := stream.Headers().Get(api.StreamType)
log.L.Infof("Connection request received new type of stream connection request %v streamType %v", requestID, streamType)
log.L.WithFields(log.Fields{"request": requestID, "streamType": streamType}).Info(
"Connection request received new type of stream connection")
p, created := h.getStreamPair(requestID)
if created {
@@ -263,9 +265,11 @@ func (h *httpStreamHandler) portForward(p *httpStreamPair) {
portString := p.dataStream.Headers().Get(api.PortHeader)
port, _ := strconv.ParseInt(portString, 10, 32)
log.L.Infof("Connection request invoking forwarder.PortForward for port connection request %v port %v", p.requestID, portString)
log.L.WithFields(log.Fields{"request": p.requestID, "port": portString}).Info(
"Connection request invoking forwarder.PortForward for port connection")
err := h.forwarder.PortForward(ctx, h.pod, h.uid, int32(port), p.dataStream)
log.L.Infof("Connection request done invoking forwarder.PortForward for port connection reqeust %v port %v", p.requestID, portString)
log.L.WithFields(log.Fields{"request": p.requestID, "port": portString}).Info(
"Connection request done invoking forwarder.PortForward for port connection")
if err != nil {
msg := fmt.Errorf("error forwarding port %d to pod %s, uid %v: %v", port, h.pod, h.uid, err)

4
internal/cri/streamingserver/portforward/websocket.go
View File

@@ -202,9 +202,9 @@ func (h *websocketStreamHandler) portForward(p *websocketStreamPair) {
defer p.dataStream.Close()
defer p.errorStream.Close()
log.L.Infof("Connection invoking forwarder.PortForward for port connection port %v", p.port)
log.L.WithField("port", p.port).Info("Connection invoking forwarder.PortForward for port connection")
err := h.forwarder.PortForward(ctx, h.pod, h.uid, p.port, p.dataStream)
log.L.Infof("Connection done invoking forwarder.PortForward for port connection port %v", p.port)
log.L.WithField("port", p.port).Info("Connection done invoking forwarder.PortForward for port connection")
if err != nil {
msg := fmt.Errorf("error forwarding port %d to pod %s, uid %v: %v", p.port, h.pod, h.uid, err)

2
internal/cri/streamingserver/remotecommand/doc.go
View File

@@ -31,4 +31,4 @@ limitations under the License.
*/
// Package remotecommand contains functions related to executing commands in and attaching to pods.
package remotecommand // import "k8s.io/kubelet/pkg/cri/streaming/remotecommand"
package remotecommand // This package was cloned from "k8s.io/kubelet/pkg/cri/streaming/remotecommand"

2
internal/cri/streamingserver/remotecommand/httpstream.go
View File

@@ -171,7 +171,7 @@ func createHTTPStreamStreams(req *http.Request, w http.ResponseWriter, opts *Opt
case remotecommandconsts.StreamProtocolV2Name:
handler = &v2ProtocolHandler{}
case "":
log.L.Infof("Client did not request protocol negotiation. Falling back protocol %v", remotecommandconsts.StreamProtocolV1Name)
log.L.WithField("protocol", remotecommandconsts.StreamProtocolV1Name).Info("Client did not request protocol negotiation. Falling back")
fallthrough
case remotecommandconsts.StreamProtocolV1Name:
handler = &v1ProtocolHandler{}

2
internal/cri/streamingserver/responsewriter/wrapper.go
View File

@@ -30,7 +30,7 @@ See the License for the specific language governing permissions and
limitations under the License.
*/
package responsewriter
package responsewriter // This package was cloned from k8s.io/apiserver/pkg/endpoints/responsewriter
import (
"net/http"

2
internal/cri/streamingserver/v1/doc.go
View File

@@ -31,4 +31,4 @@ 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"
package v1 // this package was cloned from "k8s.io/api/core/v1"