stream

Documentation

Index

• func CopyDetachable(dst io.Writer, src io.Reader, keys []byte) (written int64, err error)
• func NewLogWriter(stdout <-chan []byte, logFile string, wg *sync.WaitGroup)
• type DetachError
  • func (DetachError) Error() string
• type Server
  • func NewServer(kubernetesDir, socketPath string, metadataStore metadata.Store) (*Server, error)
  • func (s *Server) Attach(containerID string, inputStream io.Reader, ...) error
  • func (s *Server) GetAttach(req *kubeapi.AttachRequest) (*kubeapi.AttachResponse, error)
  • func (s *Server) GetPortForward(req *kubeapi.PortForwardRequest) (*kubeapi.PortForwardResponse, error)
  • func (s *Server) PortForward(podSandboxID string, port int32, stream io.ReadWriteCloser) error
  • func (s *Server) Start() error
  • func (s *Server) Stop()
• type UnixServer
  • func NewUnixServer(socketPath, kubernetesDir string) *UnixServer
  • func (u *UnixServer) AddOutputReader(containerID string, newChan chan []byte)
  • func (u *UnixServer) Listen()
  • func (u *UnixServer) RemoveOutputReader(containerID string, readerChan chan []byte)
  • func (u *UnixServer) Stop()

Functions

func CopyDetachable

func CopyDetachable(dst io.Writer, src io.Reader, keys []byte) (written int64, err error)

CopyDetachable is similar to io.Copy but support a detach key sequence to break out. based on https://github.com/kubernetes-incubator/cri-o/blob/master/utils/utils.go#L90

func NewLogWriter

func NewLogWriter(stdout <-chan []byte, logFile string, wg *sync.WaitGroup)

NewLogWriter writes the lines from stdout channel to logFile in k8s format

Types

type DetachError

type DetachError struct{}

DetachError is special error which returned in case of container detach.

func (DetachError) Error

func (DetachError) Error() string

type Server

type Server struct {
	DeadlineSeconds int

	streaming.Runtime
	// contains filtered or unexported fields
}

Server implements streaming.Runtime

func NewServer

func NewServer(kubernetesDir, socketPath string, metadataStore metadata.Store) (*Server, error)

NewServer creates a new Server

func (*Server) Attach

func (s *Server) Attach(containerID string, inputStream io.Reader, outputStream, errorStream io.WriteCloser, tty bool, resize <-chan remotecommand.TerminalSize) error

Attach endpoint for streaming.Runtime

func (*Server) GetAttach

func (s *Server) GetAttach(req *kubeapi.AttachRequest) (*kubeapi.AttachResponse, error)

GetAttach returns attach stream request

func (*Server) GetPortForward

func (s *Server) GetPortForward(req *kubeapi.PortForwardRequest) (*kubeapi.PortForwardResponse, error)

GetPortForward returns pofrforward stream request

func (*Server) PortForward

func (s *Server) PortForward(podSandboxID string, port int32, stream io.ReadWriteCloser) error

PortForward endpoint for streaming.Runtime

func (*Server) Start

func (s *Server) Start() error

Start starts streaming server gorutine and unixServer gorutine

func (*Server) Stop

func (s *Server) Stop()

Stop stops all goroutines

type UnixServer

type UnixServer struct {
	SocketPath string

	UnixConnections *syncmap.Map
	// contains filtered or unexported fields
}

UnixServer listens for connections from qemu instances and sends its stdout to registered channels.

func NewUnixServer

func NewUnixServer(socketPath, kubernetesDir string) *UnixServer

NewUnixServer creates new UnixServer. Requires socketPath on which it will listen and kubernetesDir where logs will be written

func (*UnixServer) AddOutputReader

func (u *UnixServer) AddOutputReader(containerID string, newChan chan []byte)

AddOutputReader adds a new channel for containerID to send stdout

func (*UnixServer) Listen

func (u *UnixServer) Listen()

Listen starts listening for connections from qemus

func (*UnixServer) RemoveOutputReader

func (u *UnixServer) RemoveOutputReader(containerID string, readerChan chan []byte)

RemoveOutputReader removes a channel for containerID

func (*UnixServer) Stop

func (u *UnixServer) Stop()

Stop stops listening and waits for all writers to finish