stream

Documentation

Index

• Constants
• Variables
• func ErrorStreamingDisabled(method string) error
• func ErrorTooManyInFlight() error
• func WriteError(err error, w http.ResponseWriter) error
• type Config
• type Request
• type RequestCache
  • func NewRequestCache() *RequestCache
  • func (c *RequestCache) Consume(token string) (req Request, found bool)
  • func (c *RequestCache) Insert(req Request) (token string, err error)
• type Router
• type Runtime
  • func NewStreamRuntime(ctrMgr mgr.ContainerMgr) Runtime

Constants

const (
	// DefaultStreamIdleTimeout is the timeout for idle stream.
	DefaultStreamIdleTimeout = 4 * time.Hour

	// DefaultStreamCreationTimeout is the timeout for stream creation.
	DefaultStreamCreationTimeout = 30 * time.Second
)

Keep these constants consistent with the peers in official package: k8s.io/kubernetes/pkg/kubelet/server.

Variables

var (
	// CacheTTL is timeout after which tokens become invalid.
	CacheTTL = 1 * time.Minute
	// MaxInFlight is the maximum number of in-flight requests to allow.
	MaxInFlight = 1000
	// TokenLen is the length of the random base64 encoded token identifying the request.
	TokenLen = 8
)

var DefaultConfig = Config{
	StreamIdleTimeout:               DefaultStreamIdleTimeout,
	StreamCreationTimeout:           DefaultStreamCreationTimeout,
	SupportedRemoteCommandProtocols: SupportedStreamingProtocols,
	SupportedPortForwardProtocols:   SupportedPortForwardProtocols,
}

DefaultConfig provides default values for server Config.

var SupportedPortForwardProtocols = []string{constant.PortForwardProtocolV1Name}

SupportedPortForwardProtocols is the portforward protocols which server supports.

var SupportedStreamingProtocols = []string{
	constant.StreamProtocolV1Name,
	constant.StreamProtocolV2Name,
	constant.StreamProtocolV3Name,
}

SupportedStreamingProtocols is the streaming protocols which server supports.

Functions

func ErrorStreamingDisabled

func ErrorStreamingDisabled(method string) error

ErrorStreamingDisabled returns error when the streaming method is disabled.

func ErrorTooManyInFlight

func ErrorTooManyInFlight() error

ErrorTooManyInFlight returns error when the maximum number of in-flight requests is exceeded.

func WriteError

func WriteError(err error, w http.ResponseWriter) error

WriteError translates a CRI streaming error into an appropriate HTTP response.

Types

type Config

type Config struct {
	// Address is the addr:port address the server will listen on.
	Address string

	// BaseURL is the optional base URL for constructing streaming URLs.
	// If empty, the baseURL will be constructed from the serve address.
	BaseURL *url.URL

	// StreamIdleTimeout is how long to leave idle connections open for.
	StreamIdleTimeout time.Duration
	// StreamCreationTimeout is how long to wait for clients to create streams. Only used for SPDY streaming.
	StreamCreationTimeout time.Duration

	// SupportedStreamingProtocols is the streaming protocols which server supports.
	SupportedRemoteCommandProtocols []string
	// SupportedPortForwardProtocol is the portforward protocols which server supports.
	SupportedPortForwardProtocols []string
}

Config defines the options used for running the stream server.

type Request

type Request interface{}

Request representing an *ExecRequest, *AttachRequest, or *PortForwardRequest Type.

type RequestCache

type RequestCache struct {
	// contains filtered or unexported fields
}

RequestCache caches streaming (exec/attach/port-forward) requests and generates a single-use random token for their retrieval. The requestCache is used for building streaming URLs without the need to encode every request parameter in the URL.

func NewRequestCache

func NewRequestCache() *RequestCache

NewRequestCache return a RequestCache

func (*RequestCache) Consume

func (c *RequestCache) Consume(token string) (req Request, found bool)

Consume the token (remove it from the cache) and return the cached request, if found.

func (*RequestCache) Insert

func (c *RequestCache) Insert(req Request) (token string, err error)

Insert the given request into the cache and returns the token used for fetching it out.

type Router

type Router interface {
	ServeExec(w http.ResponseWriter, r *http.Request)
	ServeAttach(w http.ResponseWriter, r *http.Request)
	ServePortForward(w http.ResponseWriter, r *http.Request)
}

Router exports a set of CRI Stream Server's handlers. We could reuse the pouchd's http server to handle the Stream Server's requests, so pouchd only has to export one port.

type Runtime

type Runtime interface {
	// Exec executes the command in pod.
	Exec(ctx context.Context, containerID string, cmd []string, resizeChan <-chan apitypes.ResizeOptions, streamOpts *remotecommand.Options, streams *remotecommand.Streams) (uint32, error)

	// Attach attaches to pod.
	Attach(ctx context.Context, containerID string, streamOpts *remotecommand.Options, streams *remotecommand.Streams) error

	// PortForward forward port to pod.
	PortForward(ctx context.Context, name string, port int32, stream io.ReadWriteCloser) error
}

Runtime is the interface to execute the commands and provide the streams.

func NewStreamRuntime

func NewStreamRuntime(ctrMgr mgr.ContainerMgr) Runtime

NewStreamRuntime creates a brand new stream runtime.