How do I record a Selenium Grid Session (Via VNC, Remote Frame Buffer - RFB) to an H264 encoded MP4

Question

When my integration tests fail, I want to be able to watch the video of the test running. I test a full stack app with docker-compose and I use the single container Selenium-grid. If I save every Secenerio to a file, I can quickly review a failed test.

I'm trying to record a Selenium Grid session and save it as an MP4 file using GStreamer and VNC. I need a solution that captures the VNC session running in the Selenium Grid container and writes the output to an H264 encoded mp4 file.

Here's the context:

I'm using Selenium Grid with a VNC server running in the container. I can access the VNC session using the se:vncLocalAddress capability.

I would like a tool that can record the VNC remote framebuffer (RFB) output.

Can someone provide a reliable way to record a Selenium Grid session to an MP4 file using a tool like GStreamer or ffmpeg?

Answer 1

Ok, I have spent way too much time figuring this out. If I ever forget how I will look here for sure.

Gstreamer can capture and display an RFB, aka VNC source. It cannot directly convert it to an mp4. At least as far as I was able to discern. But... It CAN record RFB to a stream output (.ts), and when the recording is finished, it can convert a .ts to a properly H264 encoded mp4.

Ok, I have extracted this code from my project for example purposes. I have not run this exact set of lines. YMMV but this should be really darn close to a working solution.

Install GStreamer on the container that is probably running your tests and knows when to start and stop recording.

# Install system dependencies
RUN apt-get update && apt-get install -y \
    python3 \
    python3-pip \
    gstreamer1.0-tools \
    gstreamer1.0-plugins-base \
    gstreamer1.0-plugins-good \
    gstreamer1.0-plugins-ugly \
    gstreamer1.0-libav \
    python3-gi \
    gir1.2-gstreamer-1.0 \
    && rm -rf /var/lib/apt/lists/*

This is a Python example; this is what you need for the Python bindings. For other languages, just use Python as a guide. GStreamer has many bindings.

Add to requirments.txt

PyGObject

import threading
import time
import os
import subprocess
from gi.repository import Gst, GLib

If you are recording from Selenium, it will not give you the RFB port but it will give you the VNC web socket. However Selenium has known defaults for both ports and they both inc by 1 for each additional display (session) so you can do a little math to get the correct RFB port.

    vnc_address = driver.capabilities.get("se:vncLocalAddress")
    if not vnc_address:
        print("No VNC session detected, skipping recording.")
        return

    vnc_host, no_vnc_port = vnc_address.split("://")[1].split(":")
    VNCBASEPORT = 5900
    NOVNCBASE = 7900
    vnc_port = VNCBASEPORT + (NOVNCBASE - int(no_vnc_port))

start_recording Function

This function sets up and starts the GStreamer pipeline to record a VNC session to a TS file.

def start_recording(vnc_host, vnc_port, ts_output_file):
    """
    Start recording a VNC session to a TS file using GStreamer.

    :param vnc_host: The host address of the VNC server.
    :param vnc_port: The port of the VNC server.
    :param ts_output_file: The path where the TS file will be saved.
    :return: A tuple containing the pipeline, loop, thread, and data_written event.
    """
    # Construct the GStreamer pipeline string:
    # - rfbsrc: Captures raw video from the VNC server.
    # - videoconvert: Ensures compatibility with downstream elements.
    # - videoscale: Scales the video to 1280x720 for consistency.
    # - videorate: Sets the framerate to 30 FPS for smooth playback.
    # - x264enc: Encodes the video using H.264 with zero-latency tuning.
    # - mpegtsmux: Muxes the encoded video into a TS container.
    # - filesink: Saves the output to the specified TS file.
    pipeline_str = (
        f"rfbsrc host={vnc_host} port={vnc_port} ! "
        f"videoconvert ! video/x-raw,format=I420 ! "
        f"videoscale ! video/x-raw,width=1280,height=720 ! "
        f"videorate ! video/x-raw,framerate=30/1 ! "
        f"x264enc key-int-max=30 tune=zerolatency ! "
        f"mpegtsmux ! filesink name=filesink0 location=\"{ts_output_file}\""
    )

    # Create the GStreamer pipeline from the pipeline string.
    pipeline = Gst.parse_launch(pipeline_str)

    # Get the filesink element to add a pad probe.
    filesink = pipeline.get_by_name("filesink0")

    # Add a probe to the sink pad to detect when data starts flowing:
    # - This is crucial to confirm that recording has started successfully.
    # - Once a buffer is received, we set a threading.Event to signal data flow.
    pad = filesink.get_static_pad("sink")
    data_written = threading.Event()

    def probe_callback(pad, info, data_written):
        # Callback triggered when a buffer reaches the filesink.
        print("Data buffer detected at filesink.")
        data_written.set()
        return Gst.PadProbeReturn.OK

    pad.add_probe(Gst.PadProbeType.BUFFER, probe_callback, data_written)

    # Set up the bus and main loop for handling pipeline messages:
    # - The bus receives messages like EOS (end of stream) and errors.
    # - The main loop processes these messages asynchronously.
    bus = pipeline.get_bus()
    loop = GLib.MainLoop()

    def on_message(bus, message, loop):
        # Handle pipeline messages:
        # - EOS: Indicates the pipeline has finished processing.
        # - ERROR: Indicates an error occurred, which stops the loop.
        if message.type == Gst.MessageType.EOS:
            print("Recording pipeline: End of stream reached.")
            loop.quit()
        elif message.type == Gst.MessageType.ERROR:
            err, debug = message.parse_error()
            print(f"Recording pipeline Error: {err}, Debug: {debug}")
            loop.quit()

    bus.add_signal_watch()
    bus.connect("message", on_message, loop)

    # Start the pipeline to begin capturing and recording.
    pipeline.set_state(Gst.State.PLAYING)

    # Run the main loop in a separate thread to keep it non-blocking:
    # - This allows the calling code to continue executing while recording.
    def run_loop():
        loop.run()

    thread = threading.Thread(target=run_loop)
    thread.start()

    # Return objects needed for stopping the recording later.
    return pipeline, loop, thread, data_written

def end_recording(pipeline, loop, thread, data_written, ts_output_file, mp4_output_file, log_file_path):
    """
    Stop the GStreamer recording and convert the TS file to MP4.

    :param pipeline: The GStreamer pipeline.
    :param loop: The GLib main loop.
    :param thread: The thread running the main loop.
    :param data_written: The threading event indicating data has been written.
    :param ts_output_file: The path to the TS file.
    :param mp4_output_file: The path where the MP4 file will be saved.
    :param log_file_path: The path to the log file for conversion output.
    """
    print("Stopping GStreamer recording...")

    # Wait until data has been written to the file:
    # - This ensures we don't stop prematurely and lose data.
    # - Timeout after 10 seconds and proceed anyway, logging a warning.
    if not data_written.wait(timeout=10):
        print("Timeout waiting for data to be written! Proceeding with shutdown anyway.")
    else:
        print("Confirmed: Data has been written to the file.")

    # Send EOS (end of stream) event to flush buffers and stop the pipeline:
    # - This ensures all data is written to the file.
    pipeline.send_event(Gst.Event.new_eos())
    print("EOS sent to pipeline.")

    # Wait for the EOS to be processed or an error to occur:
    # - We poll the bus for messages to confirm shutdown completion.
    # - Timeout after 5 seconds to prevent hanging.
    def wait_for_shutdown(bus, timeout_seconds=5):
        start_time = time.time()
        while time.time() - start_time < timeout_seconds:
            message = bus.timed_pop(int(0.1 * Gst.SECOND))
            if message:
                if message.type == Gst.MessageType.EOS:
                    print("EOS received, pipeline shutdown complete.")
                    return True
                elif message.type == Gst.MessageType.ERROR:
                    err, debug = message.parse_error()
                    print(f"Error during shutdown: {err}, Debug: {debug}")
                    return False
            time.sleep(0.01)  # Small sleep to avoid busy-waiting
        print("Timeout waiting for EOS!")
        return False

    bus = pipeline.get_bus()
    shutdown_success = wait_for_shutdown(bus)

    # Stop the pipeline and quit the main loop:
    # - Set pipeline state to NULL to release resources.
    # - Quit the loop and join the thread to clean up.
    pipeline.set_state(Gst.State.NULL)
    loop.quit()
    thread.join(timeout=5)

    if not shutdown_success:
        print("Shutdown was not successful, check for errors.")

    # Check .ts file size before conversion for debugging.
    ts_size = os.path.getsize(ts_output_file)
    print(f".ts file size before conversion: {ts_size / (1024 * 1024):.2f} MB")

    # GStreamer command to convert .ts to .mp4 using subprocess:
    # - tsdemux: Demuxes the TS file.
    # - h264parse: Parses the H.264 stream.
    # - mp4mux: Muxes into an MP4 container.
    # - filesink: Saves to the MP4 file.
    gst_convert_cmd = [
        "gst-launch-1.0", "-v", "-e",
        "filesrc", f"location={ts_output_file}",
        "!", "tsdemux",
        "!", "h264parse",
        "!", "mp4mux",
        "!", "filesink", f"location={mp4_output_file}"
    ]

    # Run the conversion and log output for debugging:
    # - Logs are written to the specified log file for troubleshooting.
    with open(log_file_path, "a") as log:
        log.write("\n----------------------------\n")
        log.write(f"Command: {' '.join(gst_convert_cmd)}\n")
        log.write("----------------------------\n")
        log.flush()
        convert_process = subprocess.run(gst_convert_cmd, stdout=log, stderr=log)

    # Check if conversion was successful:
    # - Verify the MP4 file exists and is not empty.
    if convert_process.returncode == 0 and os.path.exists(mp4_output_file):
        file_size = os.path.getsize(mp4_output_file)
        if file_size == 0:
            print("WARNING: MP4 file exists but is empty!")
        else:
            print(f"Recording size: {file_size / (1024 * 1024):.2f} MB")
    else:
        print("ERROR: MP4 file was not created!")

    # Print the conversion logs for review.
    print("\n=== Final GStreamer Logs ===")
    with open(log_file_path, "r") as log:
        print(log.read())

If this saves someone at least half the hair-pulling I did, it will be worth this write-up. If you have an improvement LMK and I will update this.