Merge pull request #384 from CartoDB/federated_diagnostics_latency

Federated diagnostics latency

scripts-available/CDB_FederatedServerDiagnostics.sql

@@ -98,6 +98,102 @@ $$
 LANGUAGE SQL VOLATILE PARALLEL UNSAFE;
 
 
+--
+-- Get a foreign PG server hostname from the catalog
+--
+CREATE OR REPLACE FUNCTION @extschema@.__CDB_FS_Foreign_Server_Host_PG(server_internal name)
+RETURNS text
+AS $$
+    SELECT option_value FROM (
+        SELECT (pg_options_to_table(srvoptions)).*
+        FROM pg_foreign_server WHERE srvname = server_internal
+    ) AS opt WHERE opt.option_name = 'host';
+$$
+LANGUAGE SQL VOLATILE PARALLEL UNSAFE;
+
+
+--
+-- Get a foreign PG server port from the catalog
+--
+CREATE OR REPLACE FUNCTION @extschema@.__CDB_FS_Foreign_Server_Port_PG(server_internal name)
+RETURNS integer
+AS $$
+    SELECT option_value::integer FROM (
+        SELECT (pg_options_to_table(srvoptions)).*
+        FROM pg_foreign_server WHERE srvname = server_internal
+    ) AS opt WHERE opt.option_name = 'port';
+$$
+LANGUAGE SQL VOLATILE PARALLEL UNSAFE;
+
+--
+-- Get one measure of network latency in ms to a remote TCP server
+--
+CREATE OR REPLACE FUNCTION @extschema@.__CDB_FS_TCP_Foreign_Server_Latency(
+    server_internal name,
+    timeout_seconds float DEFAULT 5.0,
+    n_samples integer DEFAULT 10
+)
+RETURNS jsonb
+AS $$
+    import socket
+    import json
+    import math
+    from timeit import default_timer as timer
+
+    plan = plpy.prepare("SELECT @extschema@.__CDB_FS_Foreign_Server_Host_PG($1) AS host", ['name'])
+    rv = plpy.execute(plan, [server_internal], 1)
+    host = rv[0]['host']
+
+    plan = plpy.prepare("SELECT @extschema@.__CDB_FS_Foreign_Server_Port_PG($1) AS port", ['name'])
+    rv = plpy.execute(plan, [server_internal], 1)
+    port = rv[0]['port'] or 5432
+
+    n_errors = 0
+    samples = []
+
+    for i in range(n_samples):
+        s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+        s.settimeout(timeout_seconds)
+
+        t_start = timer()
+
+        try:
+            s.connect((host, int(port)))
+            t_stop = timer()
+            s.shutdown(socket.SHUT_RD)
+        except (socket.timeout, OSError, socket.error) as ex:
+            plpy.warning('could not connect to server %s:%d, %s' % (host, port, str(ex)))
+            n_errors += 1
+        finally:
+            s.close()
+
+        t_connect = (t_stop - t_start) * 1000.0
+        plpy.debug('TCP connection %s:%d time=%.2f ms' % (host, port, t_connect))
+        samples.append(t_connect)
+
+    stats = {
+        'n_samples': n_samples,
+        'n_errors': n_errors,
+    }
+    n = len(samples)
+    if n >= 1:
+        mean = sum(samples) / n
+        stats.update({
+            'avg': round(mean, 3),
+            'min': round(min(samples), 3),
+            'max': round(max(samples), 3)
+        })
+    if n >= 2:
+        var = sum([ (x - mean)**2 for x in samples ]) / (n-1)
+        stdev = math.sqrt(var)
+        stats.update({
+            'stdev': round(stdev, 3)
+        })
+    return json.dumps(stats)
+$$
+LANGUAGE plpythonu VOLATILE PARALLEL UNSAFE;
+
+
 --
 -- Collect and return diagnostics info from a remote PG into a jsonb
 --
@@ -105,14 +201,16 @@ CREATE OR REPLACE FUNCTION @extschema@.__CDB_FS_Server_Diagnostics_PG(server_int
 RETURNS jsonb
 AS $$
 DECLARE
-    remote_server_version  text := @extschema@.__CDB_FS_Foreign_Server_Version_PG(server_internal);
+    remote_server_version  text    := @extschema@.__CDB_FS_Foreign_Server_Version_PG(server_internal);
-    remote_postgis_version text := @extschema@.__CDB_FS_Foreign_PostGIS_Version_PG(server_internal);
+    remote_postgis_version text    := @extschema@.__CDB_FS_Foreign_PostGIS_Version_PG(server_internal);
-    remote_server_options jsonb := @extschema@.__CDB_FS_Foreign_Server_Options_PG(server_internal);
+    remote_server_options jsonb    := @extschema@.__CDB_FS_Foreign_Server_Options_PG(server_internal);
+    remote_server_latency_ms jsonb := @extschema@.__CDB_FS_TCP_Foreign_Server_Latency(server_internal);
 BEGIN
     RETURN jsonb_build_object(
         'server_version', remote_server_version,
         'postgis_version', remote_postgis_version,
-        'server_options', remote_server_options
+        'server_options', remote_server_options,
+        'server_latency_ms', remote_server_latency_ms
     );
 END
 $$

test/CDB_FederatedServerDiagnostics.sql

@@ -23,6 +23,27 @@ SELECT 'C1', cartodb.CDB_Federated_Server_Register_PG(server => 'loopback'::text
     }
 }'::jsonb);
 
+SELECT 'C2', cartodb.CDB_Federated_Server_Register_PG(server => 'wrong-port'::text, config => '{
+    "server": {
+        "host": "localhost",
+        "port": "12345"
+    },
+    "credentials": {
+        "username": "cdb_fs_tester",
+        "password": "cdb_fs_passwd"
+    }
+}'::jsonb);
+
+SELECT 'C3', cartodb.CDB_Federated_Server_Register_PG(server => 'loopback-no-port'::text, config => '{
+    "server": {
+        "host": "localhost"
+    },
+    "credentials": {
+        "username": "cdb_fs_tester",
+        "password": "cdb_fs_passwd"
+    }
+}'::jsonb);
+
 \c cdb_fs_tester postgres
 CREATE EXTENSION postgis;
 \c contrib_regression postgres
@@ -57,12 +78,41 @@ SELECT '1.5', cartodb.CDB_Federated_Server_Diagnostics(server => 'loopback') @>
 \echo '%% It returns the remote server options'
 SELECT '1.6', cartodb.CDB_Federated_Server_Diagnostics(server => 'loopback') @> '{"server_options": {"host": "localhost", "port": "@@PGPORT@@", "updatable": "false", "extensions": "postgis", "fetch_size": "1000", "use_remote_estimate": "true"}}'::jsonb;
 
+\echo '%% It returns network latency stats to the remote server: min <= avg <= max'
+WITH latency AS (
+   SELECT CDB_Federated_Server_Diagnostics('loopback')->'server_latency_ms' ms
+) SELECT '2.1', (latency.ms->'min')::text::float <= (latency.ms->'avg')::text::float, (latency.ms->'avg')::text::float <= (latency.ms->'max')::text::float
+FROM latency;
+
+\echo '%% Latency stats: 0 <= min <= max <= 1000 ms (local connection)'
+WITH latency AS (
+   SELECT CDB_Federated_Server_Diagnostics('loopback')->'server_latency_ms' ms
+) SELECT '2.2', 0.0 <= (latency.ms->'min')::text::float, (latency.ms->'max')::text::float <= 1000.0
+FROM latency;
+
+\echo '%% Latency stats: stdev > 0'
+WITH latency AS (
+   SELECT CDB_Federated_Server_Diagnostics('loopback')->'server_latency_ms' ms
+) SELECT '2.3', (latency.ms->'stdev')::text::float >= 0.0
+FROM latency;
+
+\echo '%% It raises an error if the wrong port is provided'
+SELECT '3.0', cartodb.CDB_Federated_Server_Diagnostics(server => 'wrong-port');
+
+\echo '%% Latency stats: can get them on default PG port 5432 when not provided'
+WITH latency AS (
+   SELECT CDB_Federated_Server_Diagnostics('loopback-no-port')->'server_latency_ms' ms
+) SELECT '2.4', 0.0 <= (latency.ms->'min')::text::float, (latency.ms->'max')::text::float <= 1000.0
+FROM latency;
+
 
 -- ===================================================================
 -- Cleanup
 -- ===================================================================
 \set QUIET on
 SELECT 'D1', cartodb.CDB_Federated_Server_Unregister(server => 'loopback'::text);
+SELECT 'D2', cartodb.CDB_Federated_Server_Unregister(server => 'wrong-port'::text);
+SELECT 'D3', cartodb.CDB_Federated_Server_Unregister(server => 'loopback-no-port'::text);
 -- Reconnect, using a new session in order to close FDW connections
 \connect
 DROP DATABASE cdb_fs_tester;

test/CDB_FederatedServerDiagnostics_expect

@@ -1,4 +1,6 @@
 C1|
+C2|
+C3|
 %% It raises an error if the server does not exist
 ERROR:  Server "doesNotExist" does not exist
 %% It returns a jsonb object
@@ -11,4 +13,16 @@ ERROR:  Server "doesNotExist" does not exist
 1.5|t
 %% It returns the remote server options
 1.6|t
+%% It returns network latency stats to the remote server: min <= avg <= max
+2.1|t|t
+%% Latency stats: 0 <= min <= max <= 1000 ms (local connection)
+2.2|t|t
+%% Latency stats: stdev > 0
+2.3|t
+%% It raises an error if the wrong port is provided
+ERROR:  could not connect to server "cdb_fs_wrong-port"
+%% Latency stats: can get them on default PG port 5432 when not provided
+2.4|t|t
 D1|
+D2|
+D3|