-
Notifications
You must be signed in to change notification settings - Fork 1.6k
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
cmd/k8s-operator,k8s-operator/session-recording: implement support fo…
…r WebSocket protocol Kubernetes currently supports two streaming protocols- SPDY and WebSockets. WebSockets are replacing SPDY, see kubernetes/enhancements#4006 Our 'kubectl exec' session recording was only supporting SPDY. This change: - adds functionality to parse streaming sessions over WebSockets - for sessions that the API server proxy has determined need to be recorded, determines if the session is over SPDY or WebSockets and invoke the relevant parser accordingly - refactors the session recording logic into its own package Updates tailscale/corp#19821 Signed-off-by: Irbe Krumina <irbe@tailscale.com>
- Loading branch information
Showing
15 changed files
with
1,178 additions
and
238 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,117 @@ | ||
| // Copyright (c) Tailscale Inc & AUTHORS | ||
| // SPDX-License-Identifier: BSD-3-Clause | ||
|
|
||
| //go:build !plan9 | ||
|
|
||
| // package fakes contains utils for testing session recording behaviour. | ||
| package fakes | ||
|
|
||
| import ( | ||
| "bytes" | ||
| "encoding/json" | ||
| "net" | ||
| "sync" | ||
| "testing" | ||
|
|
||
| "tailscale.com/k8s-operator/session-recording/tsrecorder" | ||
| "tailscale.com/tstime" | ||
| ) | ||
|
|
||
| func New(conn net.Conn, wb bytes.Buffer, rb bytes.Buffer, closed bool) net.Conn { | ||
| return &TestConn{ | ||
| Conn: conn, | ||
| writeBuf: wb, | ||
| readBuf: rb, | ||
| closed: closed, | ||
| } | ||
| } | ||
|
|
||
| type TestConn struct { | ||
| net.Conn | ||
| // writeBuf contains whatever was send to the conn via Write. | ||
| writeBuf bytes.Buffer | ||
| // readBuf contains whatever was sent to the conn via Read. | ||
| readBuf bytes.Buffer | ||
| sync.RWMutex // protects the following | ||
| closed bool | ||
| } | ||
|
|
||
| var _ net.Conn = &TestConn{} | ||
|
|
||
| func (tc *TestConn) Read(b []byte) (int, error) { | ||
| return tc.readBuf.Read(b) | ||
| } | ||
|
|
||
| func (tc *TestConn) Write(b []byte) (int, error) { | ||
| return tc.writeBuf.Write(b) | ||
| } | ||
|
|
||
| func (tc *TestConn) Close() error { | ||
| tc.Lock() | ||
| defer tc.Unlock() | ||
| tc.closed = true | ||
| return nil | ||
| } | ||
|
|
||
| func (tc *TestConn) IsClosed() bool { | ||
| tc.Lock() | ||
| defer tc.Unlock() | ||
| return tc.closed | ||
| } | ||
|
|
||
| func (tc *TestConn) WriteBufBytes() []byte { | ||
| return tc.writeBuf.Bytes() | ||
| } | ||
|
|
||
| func (tc *TestConn) ResetReadBuf() { | ||
| tc.readBuf.Reset() | ||
| } | ||
|
|
||
| func (tc *TestConn) WriteReadBufBytes(b []byte) error { | ||
| _, err := tc.readBuf.Write(b) | ||
| return err | ||
| } | ||
|
|
||
| type TestSessionRecorder struct { | ||
| // buf holds data that was sent to the session recorder. | ||
| buf bytes.Buffer | ||
| } | ||
|
|
||
| func (t *TestSessionRecorder) Write(b []byte) (int, error) { | ||
| return t.buf.Write(b) | ||
| } | ||
|
|
||
| func (t *TestSessionRecorder) Close() error { | ||
| t.buf.Reset() | ||
| return nil | ||
| } | ||
|
|
||
| func (t *TestSessionRecorder) Bytes() []byte { | ||
| return t.buf.Bytes() | ||
| } | ||
|
|
||
| func CastLine(t *testing.T, p []byte, clock tstime.Clock) []byte { | ||
| t.Helper() | ||
| j, err := json.Marshal([]any{ | ||
| clock.Now().Sub(clock.Now()).Seconds(), | ||
| "o", | ||
| string(p), | ||
| }) | ||
| if err != nil { | ||
| t.Fatalf("error marshalling cast line: %v", err) | ||
| } | ||
| return append(j, '\n') | ||
| } | ||
|
|
||
| func AsciinemaResizeMsg(t *testing.T, width, height int) []byte { | ||
| t.Helper() | ||
| ch := tsrecorder.CastHeader{ | ||
| Width: width, | ||
| Height: height, | ||
| } | ||
| bs, err := json.Marshal(ch) | ||
| if err != nil { | ||
| t.Fatalf("error marshalling CastHeader: %v", err) | ||
| } | ||
| return append(bs, '\n') | ||
| } |
Oops, something went wrong.