github/containerd
2
0
Fork 0
Code Issues Pull Requests Packages Projects Releases 2 Wiki Activity

modify streaming io url form

Browse Source 
sandbox address should be in the form of
<ttrpc|grpc>+<unix|vsock|hvsock>://<uds-path|vsock-cid:vsock-port|uds-path:hvsock-port>
for example: ttrpc+hvsock:///run/test.hvsock:1024
or: grpc+vsock://1111111:1024
and the Stdin/Stdout/Stderr will add a `streaming_id` as a parameter of the url
result form is:
<ttrpc|grpc>+<unix|vsock|hvsock>://<uds-path|vsock-cid:vsock-port|uds-path:hvsock-port>?streaming_id=<stream-id>
for example ttrpc+hvsock:///run/test.hvsock:1024?streaming_id=111111
or grpc+vsock://1111111:1024?streaming_id=222222

Signed-off-by: Abel Feng <fshb1988@gmail.com>
This commit is contained in:
Abel Feng 2024-05-08 14:26:36 +08:00
parent bfdc224861
commit 42f778fc14
3 changed files with 28 additions and 9 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

11
internal/cri/io/container_io.go
View File

@ -73,6 +73,17 @@ func WithNewFIFOs(root string, tty, stdin bool) ContainerIOOpts {
} }
// WithStreams creates new streams for the container io. // WithStreams creates new streams for the container io.
// The stream address is in format of `protocol://address?stream_id=xyz`.
// It allocates ContainerID-stdin, ContainerID-stdout and ContainerID-stderr as streaming IDs.
// For example, that advertiser address of shim is `ttrpc+unix:///run/demo.sock` and container ID is `app`.
// There are three streams if stdin is enabled and TTY is disabled.
//
// - Stdin: ttrpc+unix:///run/demo.sock?stream_id=app-stdin
// - Stdout: ttrpc+unix:///run/demo.sock?stream_id=app-stdout
// - stderr: ttrpc+unix:///run/demo.sock?stream_id=app-stderr
//
// The streaming IDs will be used as unique key to establish stream tunnel.
// And it should support reconnection with the same streaming ID if containerd restarts.
func WithStreams(address string, tty, stdin bool) ContainerIOOpts { func WithStreams(address string, tty, stdin bool) ContainerIOOpts {
return func(c *ContainerIO) error { return func(c *ContainerIO) error {
if address == "" { if address == "" {

11
internal/cri/io/exec_io.go
View File

@ -55,6 +55,17 @@ func NewFifoExecIO(id, root string, tty, stdin bool) (*ExecIO, error) {
} }
// NewStreamExecIO creates exec io with streaming. // NewStreamExecIO creates exec io with streaming.
// The stream address is in format of `protocol://address?stream_id=xyz`.
// It allocates ExecID-stdin, ExecID-stdout and ExecID-stderr as streaming IDs.
// For example, that advertiser address of shim is `ttrpc+unix:///run/demo.sock` and exec ID is `app`.
// There are three streams if stdin is enabled and TTY is disabled.
//
// - Stdin: ttrpc+unix:///run/demo.sock?stream_id=app-stdin
// - Stdout: ttrpc+unix:///run/demo.sock?stream_id=app-stdout
// - stderr: ttrpc+unix:///run/demo.sock?stream_id=app-stderr
//
// The streaming IDs will be used as unique key to establish stream tunnel.
// And it should support reconnection with the same streaming ID if containerd restarts.
func NewStreamExecIO(id, address string, tty, stdin bool) (*ExecIO, error) { func NewStreamExecIO(id, address string, tty, stdin bool) (*ExecIO, error) {
fifos, err := newStreams(address, id, tty, stdin) fifos, err := newStreams(address, id, tty, stdin)
if err != nil { if err != nil {

15
internal/cri/io/helpers.go
View File

@ -105,13 +105,13 @@ func newStreams(address, id string, tty, stdin bool) (*cio.FIFOSet, error) {
fifos := cio.NewFIFOSet(cio.Config{}, func() error { return nil }) fifos := cio.NewFIFOSet(cio.Config{}, func() error { return nil })
if stdin { if stdin {
streamID := id + "-stdin" streamID := id + "-stdin"
fifos.Stdin = fmt.Sprintf("%s/streaming?id=%s", address, streamID) fifos.Stdin = fmt.Sprintf("%s?streaming_id=%s", address, streamID)
} }
stdoutStreamID := id + "-stdout" stdoutStreamID := id + "-stdout"
fifos.Stdout = fmt.Sprintf("%s/streaming?id=%s", address, stdoutStreamID) fifos.Stdout = fmt.Sprintf("%s?streaming_id=%s", address, stdoutStreamID)
if !tty { if !tty {
stderrStreamID := id + "-stderr" stderrStreamID := id + "-stderr"
fifos.Stderr = fmt.Sprintf("%s/streaming?id=%s", address, stderrStreamID) fifos.Stderr = fmt.Sprintf("%s?streaming_id=%s", address, stderrStreamID)
} }
fifos.Terminal = tty fifos.Terminal = tty
return fifos, nil return fifos, nil
@ -209,6 +209,8 @@ func openOutputStream(ctx context.Context, url string) (io.ReadCloser, error) {
} }
func openStream(ctx context.Context, urlStr string) (streamingapi.Stream, error) { func openStream(ctx context.Context, urlStr string) (streamingapi.Stream, error) {
// urlStr should be in the form of:
// <ttrpc|grpc>+<unix|vsock|hvsock>://<uds-path|vsock-cid:vsock-port|uds-path:hvsock-port>?streaming_id=<stream-id>
u, err := url.Parse(urlStr) u, err := url.Parse(urlStr)
if err != nil { if err != nil {
return nil, fmt.Errorf("address url parse error: %v", err) return nil, fmt.Errorf("address url parse error: %v", err)
@ -221,12 +223,7 @@ func openStream(ctx context.Context, urlStr string) (streamingapi.Stream, error)
" the form of <protocol>+<unix|vsock|tcp>, i.e. ttrpc+unix or grpc+vsock") " the form of <protocol>+<unix|vsock|tcp>, i.e. ttrpc+unix or grpc+vsock")
} }
if u.Path != "streaming" { id := u.Query().Get("streaming_id")
// TODO, support connect stream other than streaming api
return nil, fmt.Errorf("only <address>/streaming?id=xxx supported")
}
id := u.Query().Get("id")
if id == "" { if id == "" {
return nil, fmt.Errorf("no stream id in url queries") return nil, fmt.Errorf("no stream id in url queries")
} }