Struct StreamingServer 

pub(crate) struct StreamingServer {
    port: OnceCell<u16>,
    state: Arc<Mutex<State>>,
}

Serves the exec, attach and port forward sessions over HTTP on the loopback address, each under a one time URL, see StreamingServer::exec_url.

Fields

port: OnceCell<u16>

The port the server listens on, once it got started for the first session.

state: Arc<Mutex<State>>

Implementations

impl StreamingServer

pub(super) async fn attach_loop( session: AttachSession, sender: &mut Output, client: ClientRx, status: &mut Option<Status>, ) -> Result<()>

The attach specific read/write loop. status gets the success once the session ended, unless the client went away, which goes out after the loop.

async fn forward_output( message: IOMessage, stdout_enabled: bool, stderr_enabled: bool, sender: &mut Output, ) -> Result<()>

Forward a single output message to the client.

async fn drain_attach( attach: &SharedContainerAttach, attach_rx: &mut Receiver<IOMessage>, lag: &mut Lag, stdout_enabled: bool, stderr_enabled: bool, sender: &mut Output, ) -> Result<()>

Forward the output which is still queued or arrives shortly after the container exited, until the output ended or the Drain gives up, see Drain::recv_attach.

impl StreamingServer

pub(super) async fn start_exec( session: ExecSession, resize: Receiver<Option<Bytes>>, ) -> Result<StartedExec>

Run the runtime of an exec session and register the process it started. A terminal gets the sizes of the client from resize meanwhile.

pub(super) async fn exec_loop( started: StartedExec, sender: &mut Output, client: ClientRx, status: &mut Option<Status>, ) -> Result<()>

The exec specific read/write loop. status gets the exit status once the process exited, which goes out after the loop.

fn exit_status( res: Result<ExitChannelData, RecvError>, truncated: bool, ) -> Status

The status of the exit the watcher of the process sent. It sends it once, and only goes away without it if its task got aborted, the session must not wait for the exit forever then. A panic aborts the whole server in release builds. A session whose output got truncated while the runtime ran fails, the client must not take output with a gap for the complete one, see ContainerIO::collect_stdio.

async fn resizing<F: Future>( run: F, resize: Receiver<Option<Bytes>>, resizer: Option<TerminalResizer>, ) -> F::Output

Run the runtime and give the terminal the sizes of the client meanwhile, which it keeps until the runtime connected it, see TerminalResizer::resize. The loop of the session takes over the resizing once the runtime ran.

fn early_frames(data: Vec<u8>) -> VecDeque<Bytes>

Split the output collected while the runtime ran into frames, without copying it.

async fn drain_stdio( rx: &IOReceiver<IOMessage>, pipe: Pipe, sender: &mut Output, ) -> Result<()>

Forward the remaining output of the pipe of an exec after the process exited, until the read loop signals that it is done or the Drain gives up.

impl StreamingServer

pub(super) fn handle( kind: &str, token: RandomId, state: &Mutex<State>, request: Request<Incoming>, ) -> Response<Body>

Handle an exec or attach request, kind is the path of its URL, which the client upgrades to a WebSocket or SPDY.

fn upgrade_websocket<S, F>( request: Request<Incoming>, subprotocol: Option<&'static str>, max_message_size: usize, take_session: impl FnOnce() -> Option<S>, serve: impl FnOnce(WebSocketStream<ClientIo<TokioIo<Upgraded>>>, S) -> F + Send + 'static, ) -> Response<Body>

where S: Send + 'static, F: Future<Output = ()> + Send + 'static,

Upgrade a request to a WebSocket with the subprotocol the server picked, and serve it with the session once it got upgraded. The session gets taken only for a valid request. The client may send messages of up to max_message_size bytes.

fn upgrade_spdy<S, T, F>( request: Request<Incoming>, supported: &[(&'static str, T)], take_session: impl FnOnce() -> Option<S>, serve: impl FnOnce(TokioIo<Upgraded>, S, T) -> F + Send + 'static, span: Span, ) -> Response<Body>

where S: Send + 'static, T: Copy + Send + 'static, F: Future<Output = ()> + Send + 'static,

Answer a SPDY upgrade request with one of the supported protocols and serve the connection with the session and what the protocol stands for once it got upgraded. The session gets taken only for a valid request.

fn switching_protocols(protocol: &'static str) -> Response<Body>

The response of a successful SPDY upgrade.

fn websocket_protocols(headers: &HeaderMap) -> Vec<String>

The WebSocket subprotocols a client requested.

fn unsupported_protocols(requested: &[String]) -> Response<Body>

The answer to a WebSocket upgrade without a supported subprotocol.

pub(super) fn handle_port_forward( token: RandomId, state: &Mutex<State>, request: Request<Incoming>, ) -> Response<Body>

Handle a port forward request, which the client upgrades to SPDY.

fn is_websocket_request(headers: &HeaderMap) -> bool

fn upgrade_port_forward_websocket( token: RandomId, state: &Mutex<State>, request: Request<Incoming>, ) -> Response<Body>

Upgrade a port forward request to a WebSocket, which tunnels SPDY or speaks the channel protocol, by the subprotocols of the client.

fn spdy_handshake<T: Copy>( headers: &HeaderMap, supported: &[(&'static str, T)], ) -> Result<(T, &'static str), Box<Response<Body>>>

Check that the request asks for a SPDY/3.1 upgrade and negotiate one of the supported protocols, by the order of the client. The rejections are the ones of the SPDY server of Kubernetes.

impl StreamingServer

pub(super) async fn handle_websocket( socket: WebSocketStream<ClientIo<TokioIo<Upgraded>>>, session: Session, framing: Framing, )

Handle a single WebSocket connection, in the span of its session.

async fn run_websocket( socket: WebSocketStream<ClientIo<TokioIo<Upgraded>>>, session: Session, framing: Framing, )

Run the session of a WebSocket connection.

pub(super) async fn handle_spdy<T>(io: T, session: Session, version: Version)

where T: AsyncRead + AsyncWrite + Send + 'static,

Handle a connection which got upgraded to SPDY, in the span of its session, which the tasks of the connection inherit: a protocol error of the client names the container as well.

async fn run_spdy<T>(io: T, session: Session, version: Version)

where T: AsyncRead + AsyncWrite + Send + 'static,

Wait for the streams of the client of a SPDY connection, then run the session on them.

fn session_span(session: &Session) -> Span

The span of a session and of the tasks it starts, which shows the container ID at every log level, like the span of a request for a container does, see enter_request of the RPC interface.

async fn run_session( output: Output, input: impl ClientInput, session: Session, activity: Activity, )

Run a session until it ended or the client went away. activity tells when the connection was active the last time, for the idle timeout of the session.

fn sent(result: Result<Option<Output>, JoinError>) -> Option<Output>

Log how the write task ended, return its output once the close went out.

fn write_task( sender: Output, client: ClientRx, session: Session, activity: Activity, ) -> JoinHandle<Option<Output>>

Build a common write task based on the session type. The loop of the session only decides its final status, which goes out after it, so that the idle timeout never cuts a status off: a client gets exactly one, or none if it went away. A session whose connection was idle for its timeout fails, like any other session which did not end, so that the client does not take it for a success, unless the process exited already. An exec process gets killed then, nothing is left to end it. The task returns the output once the close went out, see Output::close.

fn final_status( result: Result<()>, status: Option<Status>, operation: Operation, ) -> Option<Status>

The one status a client gets, out of how the loop of its session ended and the status the loop decided on, none for a client which went away.

async fn until_too_large( run: impl Future<Output = Result<()>>, too_large: &CancellationToken, ) -> Result<()>

Run the loop of a session until the client sent a message larger than the server takes, which fails it. The loop may wait for a client which does not read its output then, it gets dropped, so that the status and the close go out right away.

async fn until_idle( run: impl Future<Output = Result<()>>, activity: &Activity, timeout: Option<Duration>, on_idle: impl FnOnce(), ) -> Result<()>

Run the loop of a session until the connection was idle for the timeout, which fails it. on_idle runs before the loop gets dropped.

fn read_task<I>( input: I, stdin_tx: MpscSender<Stdin>, resize_tx: Sender<Option<Bytes>>, peer_closed: Arc<AtomicBool>, too_large: CancellationToken, stdin_taken: Activity, ) -> JoinHandle<InputEnd<I>>

where I: ClientInput,

Build a common read task. Every message of the client it reads counts as activity, also stdin. Once stdin backs up behind a process which does not take it, the client is not read anymore and the connection may go idle, like a spdystream connection whose reader waits. Its pings get no answer meanwhile either, a WebSocket client which fails without pongs, like the one of client-go after about a minute, ends the session once the process does not take stdin for that long. The pings sit behind the stdin in the socket, reading them would mean buffering the stdin without a bound.

fn read_event( event: ClientEvent, pending: &mut PendingStdin, resize_tx: &Sender<Option<Bytes>>, ) -> ControlFlow<EventEnd>

Handle a single event of the client, break once its input ended.

impl StreamingServer

async fn start_if_required(&self) -> Result<u16>

Start the streaming server if not already running and return its port.

async fn serve(listener: TcpListener, state: Arc<Mutex<State>>)

Serve the main streaming server, every connection in a task of its own, at most MAX_CONNECTIONS at once, each for one request.

async fn accept(listener: &TcpListener) -> Result<TcpStream>

Accept the next connection. Like the Go server of CRI-O without Nagle’s algorithm, so that small writes in a row, like the last output and the status of a session, do not wait for a delayed ACK.

async fn sweep(state: Weak<Mutex<State>>)

Drop the sessions whose client did not come in time, otherwise they would stay until the next request. The task ends once no session waits or the server is gone, so that an idle server does not wake up.

fn route(request: Request<Incoming>, state: &Mutex<State>) -> Response<Body>

Route a request and log it with the status of its response.

fn logged_path(path: &str) -> String

The path of a request for the logs, which shows only the start of its token, see RandomId.

fn dispatch(request: Request<Incoming>, state: &Mutex<State>) -> Response<Body>

Hand a request to the handler of its path, /<kind>/<token>. Like a GET, a HEAD gets handled, the response has no body then.

pub(crate) async fn exec_url(&self, session: ExecSession) -> Result<String>

Start the server if required and return the URL of the exec session.

pub(crate) async fn attach_url(&self, session: AttachSession) -> Result<String>

Start the server if required and return the URL of the attach session.

pub(crate) async fn port_forward_url( &self, netns_path: PathBuf, idle_timeout: Option<Duration>, ) -> Result<String>

Start the server if required and return the URL for forwarding ports into the network namespace.

async fn wait_for_client<V>( &self, path: &str, sessions: impl FnOnce(&mut State) -> &mut BoundedMap<RandomId, V>, session: V, ) -> Result<String>

Start the server if required and keep a session in its map until its client takes it, return its URL under path. Too many sessions which wait for their client fail a new one, like for the kubelet, instead of dropping one whose client would come.

fn take_session(state: &Mutex<State>, token: RandomId) -> Option<Session>

Take the exec or attach session of a valid request. Requests which get rejected before leave the token, so that the client can retry.

fn take_port_forward( state: &Mutex<State>, token: RandomId, ) -> Option<PortForward>

Take the session of a valid request. Requests which get rejected before leave the token, so that the client can retry. Another request may have taken it meanwhile.

Trait Implementations

impl Debug for StreamingServer

fn fmt(&self, f: &mut Formatter<'_>) -> Result

Formats the value using the given formatter.

impl Default for StreamingServer

fn default() -> Self

Returns the “default value” for a type.