src/test/regress/sql/motion_socket.sql

-- The following test checks if the correct number and type of sockets are
-- created for motion connections both on QD and QE backends for the same
-- gp_session_id. Additionally we check if the source address used for creating
-- the motion sockets is equal to gp_segment_configuration.address.


-- start_matchignore
-- m/^INFO:  Checking postgres backend.*/
-- m/b.*COMMAND.*NODE.*NAME.*/
-- m/b.*postgres.*TCP.*/
-- m/b.*postgres.*UDP.*/
-- end_matchignore
CREATE FUNCTION check_motion_sockets()
    RETURNS VOID as $$
import subprocess, re, os, socket

# Create a temporary table to create a gang
plpy.execute("CREATE TEMP TABLE motion_socket_force_create_gang(i int);")

# We expect different number of sockets to be created for different
# interconnect types
# UDP: See calls to setupUDPListeningSocket in InitMotionUDPIFC
# TCP/PROXY: See call to setupTCPListeningSocket in InitMotionTCP
res = plpy.execute("SELECT current_setting('gp_interconnect_type');", 1)
ic_type = res[0]['current_setting']
if ic_type in ['tcp', 'proxy']:
    expected_socket_count_per_segment = 1
    expected_socket_kind = "TCP"
elif ic_type=='udpifc':
    expected_socket_count_per_segment = 2
    expected_socket_kind = "UDP"
else:
    plpy.error('Unrecognized gp_interconnect_type {}.'.format(ic_type))

# Since this test is run on a single physical host we assume that all segments
# have the same gp_segment_configuration.address
res = plpy.execute("SELECT address FROM gp_segment_configuration;", 1)
hostip = socket.gethostbyname(res[0]['address'])

res = plpy.execute("SELECT pid from gp_backend_info();")
pids_to_check = [r['pid'] for r in res]

for pid in pids_to_check:
    # We iterate through all backends related to current session
    motion_socket_count = 0
    # Typical 'lsof -i -nP -a -p <pid>' output:
    # COMMAND     PID    USER   FD   TYPE  DEVICE SIZE/OFF NODE NAME
    # postgres 271688 cmulong    4u  IPv4 1087560      0t0  TCP 127.0.0.1:51327 (LISTEN)
    # postgres 271688 cmulong    9u  IPv6 1077090      0t0  UDP [::1]:42130->[::1]:42130
    # We check count of those connections which have not been established.
    # Use the regex for example: "TCP <ip>:\d+ .*" (without '->')
    lsof_ret = subprocess.run(["lsof", "-i", "-nP", "-a", "-p", str(pid)],
        capture_output=True, check=True).stdout
    plpy.info(
        f'Checking postgres backend {pid}, ' \
        f'lsof output:\n{os.linesep.join(map(str, lsof_ret.splitlines()))}')
    # The first row are the column names
    r = lsof_ret.splitlines()[1:]
    for line in r:
        # No '->' follows he port. The connection has not be setup yet
        match=re.match(f".*{expected_socket_kind} {hostip}:\d+ .*", str(line))
        if match is not None:
            motion_socket_count += 1

    if motion_socket_count != expected_socket_count_per_segment:
        plpy.error('Expected {} motion sockets but found {}. '\
        .format(expected_socket_count_per_segment, motion_socket_count))


$$ LANGUAGE plpython3u;
SELECT check_motion_sockets();