Class: KubeVirtPortForwarder

Inherits:
Object
  • Object
show all
Defined in:
lib/beaker/hypervisor/port_forward.rb

Overview

KubeVirtPortForwarder acts as a local TCP proxy for a port on a KubeVirt VMI. It handles the entire lifecycle of discovering the VMI, establishing a WebSocket connection via the Kubernetes API, and proxying data.

Each accepted client connection makes a single attempt to open the WebSocket to the VMI. If that attempt fails, the client socket is closed so the caller (e.g. Beaker's SSH client) sees the connection failure promptly and learns the real state of the port, rather than waiting behind an internal retry loop.

See the bottom of this file for a complete usage example.

Constant Summary collapse

STREAM_PROTOCOL =

The subprotocol required by the Kubernetes API for multiplexed streaming. This protocol defines channels for stdin, stdout, stderr, and a special error channel, which allows for out-of-band error reporting.

'v4.channel.k8s.io'
PLAIN_STREAM_PROTOCOL =

A KubeVirt-specific subprotocol for a raw, un-multiplexed data stream.

'plain.kubevirt.io'
DATA_CHANNEL =

The channel byte for the primary data stream (stdin/stdout).

"\x00"
ERROR_CHANNEL =

The channel byte for the error stream from the server.

"\x01"
REACTOR_STARTUP_TIMEOUT =

Upper bound on how long we'll wait for EventMachine.reactor_running? to become true after kicking off EventMachine.run in its own thread. Healthy startup is well under 100ms; five seconds is only reached when something is wrong (native-ext load failure, incompatible libcrypto, etc.).

5
REACTOR_SHUTDOWN_TIMEOUT =

Upper bound on how long we'll wait for the reactor thread to exit during shutdown after EventMachine.stop. If startup timed out, the reactor thread is alive but hung and EventMachine.stop is a no-op, so we must kill it rather than block forever on join.

5
WEBSOCKET_PING_INTERVAL =

Send a WebSocket ping every N seconds. SSH keepalive packets are payload frames that some upstream WS-aware proxies don't count against their idle timer; protocol-level pings always do.

60

Class Attribute Summary collapse

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(kube_client:, namespace:, vmi_name:, target_port:, local_port:, logger: nil, on_error: nil, ssl_options: nil) ⇒ KubeVirtPortForwarder

Returns a new instance of KubeVirtPortForwarder.

Parameters:

  • kube_client (Kubeclient::Client) —

    An initialized kubeclient client.

  • namespace (String) —

    The Kubernetes namespace of the VMI.

  • vmi_name (String) —

    The name of the VirtualMachineInstance.

  • target_port (Integer) —

    The port inside the VMI to connect to (e.g., 22 for SSH).

  • local_port (Integer) —

    The local TCP port to listen on.

  • logger (Logger) (defaults to: nil) —

    An optional logger instance.

  • on_error (Proc) (defaults to: nil) —

    An optional callback (proc or lambda) to handle errors.

  • ssl_options (Hash) (defaults to: nil) —

    Optional SSL options to use instead of kube_client's (to preserve :ca_file)



74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
 
# File 'lib/beaker/hypervisor/port_forward.rb', line 74

def initialize(kube_client:, namespace:, vmi_name:, target_port:, local_port:, logger: nil, on_error: nil, ssl_options: nil)
  @kube_client = kube_client
  @namespace = namespace
  @vmi_name = vmi_name
  @target_port = target_port
  @local_port = local_port
  @on_error = on_error
  @logger = logger || Logger.new($stdout, level: :info)
  @ssl_options = ssl_options # Use provided ssl_options if available

  @state = :new
  @mutex = Mutex.new
  @server_thread = nil
  @reactor_thread = nil
  @connection_threads = []
  @shutdown = false
end

Class Attribute Details

.reactor_owner ⇒ Object

Returns the value of attribute reactor_owner.



63
64
65
 
# File 'lib/beaker/hypervisor/port_forward.rb', line 63

def reactor_owner
  @reactor_owner
end

.reactor_owner_mutex ⇒ Object

Returns the value of attribute reactor_owner_mutex.



63
64
65
 
# File 'lib/beaker/hypervisor/port_forward.rb', line 63

def reactor_owner_mutex
  @reactor_owner_mutex
end

Instance Attribute Details

#local_port ⇒ Object (readonly)

Returns the value of attribute local_port.



23
24
25
 
# File 'lib/beaker/hypervisor/port_forward.rb', line 23

def local_port
  @local_port
end

#state ⇒ Object (readonly)

Returns the value of attribute state.



23
24
25
 
# File 'lib/beaker/hypervisor/port_forward.rb', line 23

def state
  @state
end

Instance Method Details

#start ⇒ Object

Starts the local TCP server and the EventMachine reactor in background threads.



93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
 
# File 'lib/beaker/hypervisor/port_forward.rb', line 93

def start
  return unless state_transition_to(:starting)

  @logger.info("Starting local proxy on 127.0.0.1:#{@local_port} for vmi://#{@namespace}/#{@vmi_name}:#{@target_port}")

  # Reset shutdown flag
  @shutdown = false

  # Start the EventMachine reactor in a dedicated thread to handle WebSocket I/O.
  # EventMachine has a single global reactor, so we need to check if it's already
  # running. Guarding only on EventMachine.reactor_running? would race: a second
  # forwarder entering this block before the first reactor thread has actually
  # become "running" would also see false, set itself as owner, and spawn a
  # second EventMachine.run thread. Guard on reactor_owner.nil? as well so the
  # owner is locked in atomically — a non-owner just waits for the in-progress
  # reactor to become ready.
  start_reactor = false
  self.class.reactor_owner_mutex.synchronize do
    if self.class.reactor_owner.nil? && !EventMachine.reactor_running?
      start_reactor = true
      self.class.reactor_owner = self
    end
  end

  if start_reactor
    @reactor_thread = Thread.new { EventMachine.run }
    wait_for_reactor_ready
    @logger.debug('Started EventMachine reactor (owned by this forwarder)')
  else
    wait_for_existing_reactor_ready
    @logger.debug('Using existing EventMachine reactor (owned by another forwarder)')
  end

  @server = TCPServer.new('127.0.0.1', @local_port)
  state_transition_to(:running)

  @server_thread = Thread.new do
    loop do
      break if @state == :stopping || @shutdown

      begin
        # Accept a connection from a client (e.g., Beaker's SSH).
        client_socket = @server.accept
        peer_ip, peer_port = client_socket.peeraddr(false).values_at(3, 1)
        @logger.debug("Accepted connection from #{peer_ip}:#{peer_port}")

        # Handle the entire KubeVirt connection lifecycle in a new thread.
        conn_thread = Thread.new { handle_connection(client_socket) }
        @mutex.synchronize { @connection_threads << conn_thread }
      rescue IOError
        # This is expected when @server.close is called in stop()
        @logger.debug("Server on port #{@local_port} stopped accepting connections.")
        break
      end
    end
  end
rescue StandardError => e
  report_error(e)
  state_transition_to(:error)
  stop # Attempt a clean shutdown on startup failure
end

#stop ⇒ Object

Stops the server, closes all active connections, and cleans up threads. This method is designed to be idempotent.



157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
 
# File 'lib/beaker/hypervisor/port_forward.rb', line 157

def stop
  return unless state_transition_to(:stopping)

  @logger.info("Stopping port forwarder for vmi://#{@namespace}/#{@vmi_name}:#{@target_port}")

  # Set shutdown flag to signal threads to stop gracefully
  @shutdown = true

  # Close the main server socket to stop accepting new connections.
  @server&.close
  @server = nil

  # Wait for the main server thread to finish.
  @server_thread&.join

  # Clean up any active connection threads.
  threads_to_join = []
  @mutex.synchronize do
    threads_to_join = @connection_threads.dup
    @connection_threads.clear
  end

  # Give threads a chance to terminate gracefully
  threads_to_join.each do |thread|
    # Raise an exception in the thread to interrupt blocking I/O
    thread.raise(IOError, 'Port forwarder shutting down') if thread.alive?
  end

  # Wait for threads to finish with a timeout
  deadline = Time.now + 5
  threads_to_join.each do |thread|
    remaining = deadline - Time.now
    if remaining.positive?
      # Wait for thread to finish, but don't wait longer than the deadline
      begin
        thread.join(remaining)
      rescue StandardError => e
        # Thread may raise an exception when we called thread.raise
        @logger.debug("Thread exited with exception during shutdown: #{e.class}: #{e.message}")
      end
    end

    # If thread is still alive after timeout, kill it as last resort
    next unless thread.alive?

    @logger.warn("Force-killing connection thread that didn't shut down gracefully")
    thread.kill
    begin
      thread.join
    rescue StandardError
      nil
    end
  end

  # Stop the EventMachine reactor only if this instance owns it.
  # EventMachine has a single global reactor, so we must not stop it
  # if other forwarders are still using it.
  should_stop_reactor = false
  self.class.reactor_owner_mutex.synchronize do
    if self.class.reactor_owner == self
      should_stop_reactor = true
      self.class.reactor_owner = nil
    end
  end

  if should_stop_reactor
    EventMachine.stop if EventMachine.reactor_running?
    shut_down_reactor_thread
    @logger.debug('Stopped EventMachine reactor (owned by this forwarder)')
  else
    @logger.debug('Not stopping EventMachine reactor (owned by another forwarder)')
  end

  @logger.info('Port forwarder stopped.')
  state_transition_to(:stopped)
end