v3-0003-Add-network-traffic-stats.patch

From 7d3a7ffca211d36c43132280e1c775fb0459572e Mon Sep 17 00:00:00 2001
From: Jacob Burroughs <jburroughs@instructure.com>
Date: Fri, 15 Dec 2023 13:08:55 -0600
Subject: [PATCH v3 3/5] Add network traffic stats

Adds pg_stat_network_traffic to monitor both bytes sent over
the wire and "logical" protocol bytes, allowing for measurement
of compression ratio (and SSL overhead)
---
 doc/src/sgml/monitoring.sgml                | 100 ++++++++++++++++++++
 src/backend/catalog/system_views.sql        |  10 ++
 src/backend/libpq/pqcomm.c                  |   8 ++
 src/backend/utils/activity/backend_status.c |  86 +++++++++++++++++
 src/backend/utils/adt/pgstatfuncs.c         |  50 +++++++++-
 src/include/catalog/pg_proc.dat             |  14 ++-
 src/include/utils/backend_status.h          |  10 ++
 src/test/regress/expected/rules.out         |  15 ++-
 8 files changed, 285 insertions(+), 8 deletions(-)

diff --git a/doc/src/sgml/monitoring.sgml b/doc/src/sgml/monitoring.sgml
index b804eb8b5e..8f4a55d9b2 100644
--- a/doc/src/sgml/monitoring.sgml
+++ b/doc/src/sgml/monitoring.sgml
@@ -350,6 +350,15 @@ postgres   27093  0.0  0.0  30096  2752 ?        Ss   11:34   0:00 postgres: ser
       </entry>
      </row>
 
+     <row>
+      <entry><structname>pg_stat_network_traffic</structname><indexterm><primary>pg_stat_network_traffic</primary></indexterm></entry>
+      <entry>One row per connection (regular and replication), showing information about
+       network traffic of this connection.
+       See <link linkend="monitoring-pg-stat-network-traffic-view">
+       <structname>pg_stat_network_traffic</structname></link> for details.
+      </entry>
+     </row>
+
      <row>
       <entry><structname>pg_stat_ssl</structname><indexterm><primary>pg_stat_ssl</primary></indexterm></entry>
       <entry>One row per connection (regular and replication), showing information about
@@ -2176,6 +2185,97 @@ description | Waiting for a newly initialized WAL file to reach durable storage
 
  </sect2>
 
+ <sect2 id="monitoring-pg-stat-network-traffic-view">
+  <title><structname>pg_stat_network_traffic</structname></title>
+
+  <indexterm>
+   <primary>pg_stat_network_traffic</primary>
+  </indexterm>
+
+  <para>
+   The <structname>pg_stat_network_traffic</structname> view will contain one row per
+   backend or WAL sender process, showing statistics about network traffic on
+   this connection. It can be joined to <structname>pg_stat_activity</structname>
+   or <structname>pg_stat_replication</structname> on the
+   <structfield>pid</structfield> column to get more details about the
+   connection.
+  </para>
+
+  <table id="pg-stat-network-traffic-view" xreflabel="pg_stat_network_traffic">
+   <title><structname>pg_stat_network_traffic</structname> View</title>
+   <tgroup cols="1">
+    <thead>
+     <row>
+      <entry role="catalog_table_entry"><para role="column_definition">
+       Column Type
+      </para>
+      <para>
+       Description
+      </para></entry>
+     </row>
+    </thead>
+
+    <tbody>
+     <row>
+      <entry role="catalog_table_entry"><para role="column_definition">
+       <structfield>pid</structfield> <type>integer</type>
+      </para>
+      <para>
+       Process ID of a backend or WAL sender process
+      </para></entry>
+     </row>
+
+     <row>
+      <entry role="catalog_table_entry"><para role="column_definition">
+       <structfield>rx_socket_bytes</structfield> <type>integer</type>
+      </para>
+      <para>
+       Number of bytes received from the underlying socket of this connection
+      </para></entry>
+     </row>
+
+     <row>
+      <entry role="catalog_table_entry"><para role="column_definition">
+       <structfield>tx_socket_bytes</structfield> <type>integer</type>
+      </para>
+      <para>
+       Number of bytes sent to the underlying socket of this connection
+      </para></entry>
+     </row>
+
+     <row>
+      <entry role="catalog_table_entry"><para role="column_definition">
+       <structfield>rx_pq_bytes</structfield> <type>integer</type>
+      </para>
+      <para>
+       Number of bytes of raw libpq traffic recieved from this connection.
+       This can be used to estimate the overhead of SSL, which will generally
+       make this number smaller than <structfield>rx_socket_bytes</structfield>,
+       and to estimate the compression ratio when using protocol compression,
+       as compression will generally make this number larger than
+       <structfield>rx_socket_bytes</structfield>.
+      </para></entry>
+     </row>
+
+     <row>
+      <entry role="catalog_table_entry"><para role="column_definition">
+       <structfield>tx_pq_bytes</structfield> <type>integer</type>
+      </para>
+      <para>
+       Number of bytes of raw libpq traffic sent through this connection.
+       This can be used to estimate the overhead of SSL, which will generally
+       make this number smaller than <structfield>tx_socket_bytes</structfield>,
+       and to estimate the compression ratio when using protocol compression,
+       as compression will generally make this number larger than
+       <structfield>tx_socket_bytes</structfield>.
+      </para></entry>
+     </row>
+    </tbody>
+   </tgroup>
+  </table>
+
+ </sect2>
+
  <sect2 id="monitoring-pg-stat-ssl-view">
   <title><structname>pg_stat_ssl</structname></title>
 
diff --git a/src/backend/catalog/system_views.sql b/src/backend/catalog/system_views.sql
index 058fc47c91..2cd4d61305 100644
--- a/src/backend/catalog/system_views.sql
+++ b/src/backend/catalog/system_views.sql
@@ -893,6 +893,16 @@ CREATE VIEW pg_stat_activity AS
         LEFT JOIN pg_database AS D ON (S.datid = D.oid)
         LEFT JOIN pg_authid AS U ON (S.usesysid = U.oid);
 
+CREATE VIEW pg_stat_network_traffic AS
+    SELECT
+            S.pid,
+            S.rx_socket_bytes,
+            S.tx_socket_bytes,
+            S.rx_pq_bytes,
+            S.tx_pq_bytes
+    FROM pg_stat_get_activity(NULL) AS S
+    WHERE S.client_port IS NOT NULL;
+
 CREATE VIEW pg_stat_replication AS
     SELECT
             S.pid,
diff --git a/src/backend/libpq/pqcomm.c b/src/backend/libpq/pqcomm.c
index ed4aad57e6..c44840881c 100644
--- a/src/backend/libpq/pqcomm.c
+++ b/src/backend/libpq/pqcomm.c
@@ -966,6 +966,8 @@ socket_read(IoStreamLayer * self, Port *port, void *ptr, size_t len, bool buffer
 #ifdef WIN32
 	pgwin32_noblock = false;
 #endif
+	if (n > 0)
+		pgstat_report_rx_socket_traffic(n);
 
 	return n;
 }
@@ -982,6 +984,8 @@ socket_write(IoStreamLayer * self, Port *port, const void *ptr, size_t len, size
 #ifdef WIN32
 	pgwin32_noblock = false;
 #endif
+	if (n > 0)
+		pgstat_report_tx_socket_traffic(n);
 
 	if (n >= 0)
 	{
@@ -1009,6 +1013,8 @@ io_read_with_wait(Port *port, void *ptr, size_t len)
 retry:
 	port->waitfor = WL_SOCKET_READABLE;
 	n = io_stream_read(port->io_stream, ptr, len, false);
+	if (n > 0)
+		pgstat_report_rx_pq_traffic(n);
 
 	/* In blocking mode, wait until the socket is ready */
 	if (n < 0 && !port->noblock && (errno == EWOULDBLOCK || errno == EAGAIN))
@@ -1092,6 +1098,8 @@ retry:
 	 */
 	rc = io_stream_write(port->io_stream, ptr + *bytes_processed, len - *bytes_processed, &count);
 	*bytes_processed += count;
+	if (count > 0)
+		pgstat_report_tx_pq_traffic(count);
 
 	if (rc < 0 && !port->noblock && (errno == EWOULDBLOCK || errno == EAGAIN))
 	{
diff --git a/src/backend/utils/activity/backend_status.c b/src/backend/utils/activity/backend_status.c
index 6e734c6caf..c5712d86b8 100644
--- a/src/backend/utils/activity/backend_status.c
+++ b/src/backend/utils/activity/backend_status.c
@@ -338,6 +338,9 @@ pgstat_bestart(void)
 	lbeentry.st_xact_start_timestamp = 0;
 	lbeentry.st_databaseid = MyDatabaseId;
 
+	lbeentry.st_rx_socket_bytes = lbeentry.st_tx_socket_bytes =
+		lbeentry.st_rx_pq_bytes = lbeentry.st_tx_pq_bytes = 0;
+
 	/* We have userid for client-backends, wal-sender and bgworker processes */
 	if (lbeentry.st_backendType == B_BACKEND
 		|| lbeentry.st_backendType == B_WAL_SENDER
@@ -717,6 +720,89 @@ pgstat_report_xact_timestamp(TimestampTz tstamp)
 	PGSTAT_END_WRITE_ACTIVITY(beentry);
 }
 
+/*
+ * Report network raw or compressed tx/rx traffic as the specified values.
+ */
+void
+pgstat_report_rx_socket_traffic(uint64 bytes)
+{
+	volatile PgBackendStatus *beentry = MyBEEntry;
+
+	if (!pgstat_track_activities || !beentry)
+		return;
+
+	/*
+	 * Update my status entry, following the protocol of bumping
+	 * st_changecount before and after.  We use a volatile pointer here to
+	 * ensure the compiler doesn't try to get cute.
+	 */
+	PGSTAT_BEGIN_WRITE_ACTIVITY(beentry);
+
+	beentry->st_rx_socket_bytes += bytes;
+
+	PGSTAT_END_WRITE_ACTIVITY(beentry);
+}
+
+void
+pgstat_report_tx_socket_traffic(uint64 bytes)
+{
+	volatile PgBackendStatus *beentry = MyBEEntry;
+
+	if (!pgstat_track_activities || !beentry)
+		return;
+
+	/*
+	 * Update my status entry, following the protocol of bumping
+	 * st_changecount before and after.  We use a volatile pointer here to
+	 * ensure the compiler doesn't try to get cute.
+	 */
+	PGSTAT_BEGIN_WRITE_ACTIVITY(beentry);
+
+	beentry->st_tx_socket_bytes += bytes;
+
+	PGSTAT_END_WRITE_ACTIVITY(beentry);
+}
+
+void
+pgstat_report_rx_pq_traffic(uint64 bytes)
+{
+	volatile PgBackendStatus *beentry = MyBEEntry;
+
+	if (!pgstat_track_activities || !beentry)
+		return;
+
+	/*
+	 * Update my status entry, following the protocol of bumping
+	 * st_changecount before and after.  We use a volatile pointer here to
+	 * ensure the compiler doesn't try to get cute.
+	 */
+	PGSTAT_BEGIN_WRITE_ACTIVITY(beentry);
+
+	beentry->st_rx_pq_bytes += bytes;
+
+	PGSTAT_END_WRITE_ACTIVITY(beentry);
+}
+
+void
+pgstat_report_tx_pq_traffic(uint64 bytes)
+{
+	volatile PgBackendStatus *beentry = MyBEEntry;
+
+	if (!pgstat_track_activities || !beentry)
+		return;
+
+	/*
+	 * Update my status entry, following the protocol of bumping
+	 * st_changecount before and after.  We use a volatile pointer here to
+	 * ensure the compiler doesn't try to get cute.
+	 */
+	PGSTAT_BEGIN_WRITE_ACTIVITY(beentry);
+
+	beentry->st_tx_pq_bytes += bytes;
+
+	PGSTAT_END_WRITE_ACTIVITY(beentry);
+}
+
 /* ----------
  * pgstat_read_current_status() -
  *
diff --git a/src/backend/utils/adt/pgstatfuncs.c b/src/backend/utils/adt/pgstatfuncs.c
index e65cbf41e9..1273944fd8 100644
--- a/src/backend/utils/adt/pgstatfuncs.c
+++ b/src/backend/utils/adt/pgstatfuncs.c
@@ -304,7 +304,7 @@ pg_stat_get_progress_info(PG_FUNCTION_ARGS)
 Datum
 pg_stat_get_activity(PG_FUNCTION_ARGS)
 {
-#define PG_STAT_GET_ACTIVITY_COLS	31
+#define PG_STAT_GET_ACTIVITY_COLS	35
 	int			num_backends = pgstat_fetch_stat_numbackends();
 	int			curr_backend;
 	int			pid = PG_ARGISNULL(0) ? -1 : PG_GETARG_INT32(0);
@@ -617,6 +617,10 @@ pg_stat_get_activity(PG_FUNCTION_ARGS)
 				nulls[30] = true;
 			else
 				values[30] = UInt64GetDatum(beentry->st_query_id);
+			values[31] = UInt64GetDatum(beentry->st_rx_socket_bytes);
+			values[32] = UInt64GetDatum(beentry->st_tx_socket_bytes);
+			values[33] = UInt64GetDatum(beentry->st_rx_pq_bytes);
+			values[34] = UInt64GetDatum(beentry->st_tx_pq_bytes);
 		}
 		else
 		{
@@ -646,6 +650,10 @@ pg_stat_get_activity(PG_FUNCTION_ARGS)
 			nulls[28] = true;
 			nulls[29] = true;
 			nulls[30] = true;
+			nulls[31] = true;
+			nulls[32] = true;
+			nulls[33] = true;
+			nulls[34] = true;
 		}
 
 		tuplestore_putvalues(rsinfo->setResult, rsinfo->setDesc, values, nulls);
@@ -876,6 +884,46 @@ pg_stat_get_backend_start(PG_FUNCTION_ARGS)
 	PG_RETURN_TIMESTAMPTZ(result);
 }
 
+Datum
+pg_stat_get_network_traffic(PG_FUNCTION_ARGS)
+{
+#define PG_STAT_NETWORK_TRAFFIC_COLS	4
+	TupleDesc	tupdesc;
+	Datum		values[PG_STAT_NETWORK_TRAFFIC_COLS];
+	bool		nulls[PG_STAT_NETWORK_TRAFFIC_COLS];
+	int32		beid = PG_GETARG_INT32(0);
+	PgBackendStatus *beentry;
+
+	if ((beentry = pgstat_get_beentry_by_backend_id(beid)) == NULL)
+		PG_RETURN_NULL();
+	else if (!HAS_PGSTAT_PERMISSIONS(beentry->st_userid))
+		PG_RETURN_NULL();
+
+	/* Initialise values and NULL flags arrays */
+	MemSet(values, 0, sizeof(values));
+	MemSet(nulls, 0, sizeof(nulls));
+
+	/* Initialise attributes information in the tuple descriptor */
+	tupdesc = CreateTemplateTupleDesc(PG_STAT_NETWORK_TRAFFIC_COLS);
+	TupleDescInitEntry(tupdesc, (AttrNumber) 1, "rx_socket_bytes",
+					   INT8OID, -1, 0);
+	TupleDescInitEntry(tupdesc, (AttrNumber) 2, "tx_socket_bytes",
+					   INT8OID, -1, 0);
+	TupleDescInitEntry(tupdesc, (AttrNumber) 3, "rx_pq_bytes",
+					   INT8OID, -1, 0);
+	TupleDescInitEntry(tupdesc, (AttrNumber) 4, "tx_pq_bytes",
+					   INT8OID, -1, 0);
+	BlessTupleDesc(tupdesc);
+
+	/* Fill values and NULLs */
+	values[0] = UInt64GetDatum(beentry->st_rx_socket_bytes);
+	values[1] = UInt64GetDatum(beentry->st_tx_socket_bytes);
+	values[2] = UInt64GetDatum(beentry->st_rx_pq_bytes);
+	values[3] = UInt64GetDatum(beentry->st_tx_pq_bytes);
+
+	/* Returns the record as Datum */
+	PG_RETURN_DATUM(HeapTupleGetDatum(heap_form_tuple(tupdesc, values, nulls)));
+}
 
 Datum
 pg_stat_get_backend_client_addr(PG_FUNCTION_ARGS)
diff --git a/src/include/catalog/pg_proc.dat b/src/include/catalog/pg_proc.dat
index 9052f5262a..cf108da019 100644
--- a/src/include/catalog/pg_proc.dat
+++ b/src/include/catalog/pg_proc.dat
@@ -5440,9 +5440,9 @@
   proname => 'pg_stat_get_activity', prorows => '100', proisstrict => 'f',
   proretset => 't', provolatile => 's', proparallel => 'r',
   prorettype => 'record', proargtypes => 'int4',
-  proallargtypes => '{int4,oid,int4,oid,text,text,text,text,text,timestamptz,timestamptz,timestamptz,timestamptz,inet,text,int4,xid,xid,text,bool,text,text,int4,text,numeric,text,bool,text,bool,bool,int4,int8}',
-  proargmodes => '{i,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o}',
-  proargnames => '{pid,datid,pid,usesysid,application_name,state,query,wait_event_type,wait_event,xact_start,query_start,backend_start,state_change,client_addr,client_hostname,client_port,backend_xid,backend_xmin,backend_type,ssl,sslversion,sslcipher,sslbits,ssl_client_dn,ssl_client_serial,ssl_issuer_dn,gss_auth,gss_princ,gss_enc,gss_delegation,leader_pid,query_id}',
+  proallargtypes => '{int4,oid,int4,oid,text,text,text,text,text,timestamptz,timestamptz,timestamptz,timestamptz,inet,text,int4,xid,xid,text,bool,text,text,int4,text,numeric,text,bool,text,bool,bool,int4,int8,int8,int8,int8,int8}',
+  proargmodes => '{i,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o}',
+  proargnames => '{pid,datid,pid,usesysid,application_name,state,query,wait_event_type,wait_event,xact_start,query_start,backend_start,state_change,client_addr,client_hostname,client_port,backend_xid,backend_xmin,backend_type,ssl,sslversion,sslcipher,sslbits,ssl_client_dn,ssl_client_serial,ssl_issuer_dn,gss_auth,gss_princ,gss_enc,gss_delegation,leader_pid,query_id,rx_socket_bytes,tx_socket_bytes,rx_pq_bytes,tx_pq_bytes}',
   prosrc => 'pg_stat_get_activity' },
 { oid => '8403', descr => 'describe wait events',
   proname => 'pg_get_wait_events', procost => '10', prorows => '250',
@@ -5797,6 +5797,14 @@
   proargnames => '{stats_reset,prefetch,hit,skip_init,skip_new,skip_fpw,skip_rep,wal_distance,block_distance,io_depth}',
   prosrc => 'pg_stat_get_recovery_prefetch' },
 
+{ oid => '9598', descr => 'statistics: information about network traffic',
+  proname => 'pg_stat_get_network_traffic', proisstrict => 'f', provolatile => 's',
+  proparallel => 'r', prorettype => 'record', proargtypes => 'int4',
+  proallargtypes => '{int4,int8,int8,int8,int8}',
+  proargmodes => '{i,o,o,o,o}',
+  proargnames => '{_beid,rx_socket_bytes,tx_socket_bytes,rx_pq_bytes,tx_pq_bytes}',
+  prosrc => 'pg_stat_get_network_traffic' },
+
 { oid => '2306', descr => 'statistics: information about SLRU caches',
   proname => 'pg_stat_get_slru', prorows => '100', proisstrict => 'f',
   proretset => 't', provolatile => 's', proparallel => 'r',
diff --git a/src/include/utils/backend_status.h b/src/include/utils/backend_status.h
index 75fc18c432..5df7bbd797 100644
--- a/src/include/utils/backend_status.h
+++ b/src/include/utils/backend_status.h
@@ -146,6 +146,12 @@ typedef struct PgBackendStatus
 	/* application name; MUST be null-terminated */
 	char	   *st_appname;
 
+	/* client-server traffic information */
+	uint64		st_rx_socket_bytes;
+	uint64		st_tx_socket_bytes;
+	uint64		st_rx_pq_bytes;
+	uint64		st_tx_pq_bytes;
+
 	/*
 	 * Current command string; MUST be null-terminated. Note that this string
 	 * possibly is truncated in the middle of a multi-byte character. As
@@ -321,6 +327,10 @@ extern void pgstat_report_query_id(uint64 query_id, bool force);
 extern void pgstat_report_tempfile(size_t filesize);
 extern void pgstat_report_appname(const char *appname);
 extern void pgstat_report_xact_timestamp(TimestampTz tstamp);
+extern void pgstat_report_rx_socket_traffic(uint64 bytes);
+extern void pgstat_report_tx_socket_traffic(uint64 bytes);
+extern void pgstat_report_rx_pq_traffic(uint64 bytes);
+extern void pgstat_report_tx_pq_traffic(uint64 bytes);
 extern const char *pgstat_get_backend_current_activity(int pid, bool checkUser);
 extern const char *pgstat_get_crashed_backend_activity(int pid, char *buffer,
 													   int buflen);
diff --git a/src/test/regress/expected/rules.out b/src/test/regress/expected/rules.out
index f645e8486b..33e1017109 100644
--- a/src/test/regress/expected/rules.out
+++ b/src/test/regress/expected/rules.out
@@ -1760,7 +1760,7 @@ pg_stat_activity| SELECT s.datid,
     s.query_id,
     s.query,
     s.backend_type
-   FROM ((pg_stat_get_activity(NULL::integer) s(datid, pid, usesysid, application_name, state, query, wait_event_type, wait_event, xact_start, query_start, backend_start, state_change, client_addr, client_hostname, client_port, backend_xid, backend_xmin, backend_type, ssl, sslversion, sslcipher, sslbits, ssl_client_dn, ssl_client_serial, ssl_issuer_dn, gss_auth, gss_princ, gss_enc, gss_delegation, leader_pid, query_id)
+   FROM ((pg_stat_get_activity(NULL::integer) s(datid, pid, usesysid, application_name, state, query, wait_event_type, wait_event, xact_start, query_start, backend_start, state_change, client_addr, client_hostname, client_port, backend_xid, backend_xmin, backend_type, ssl, sslversion, sslcipher, sslbits, ssl_client_dn, ssl_client_serial, ssl_issuer_dn, gss_auth, gss_princ, gss_enc, gss_delegation, leader_pid, query_id, rx_socket_bytes, tx_socket_bytes, rx_pq_bytes, tx_pq_bytes)
      LEFT JOIN pg_database d ON ((s.datid = d.oid)))
      LEFT JOIN pg_authid u ON ((s.usesysid = u.oid)));
 pg_stat_all_indexes| SELECT c.oid AS relid,
@@ -1880,7 +1880,7 @@ pg_stat_gssapi| SELECT pid,
     gss_princ AS principal,
     gss_enc AS encrypted,
     gss_delegation AS credentials_delegated
-   FROM pg_stat_get_activity(NULL::integer) s(datid, pid, usesysid, application_name, state, query, wait_event_type, wait_event, xact_start, query_start, backend_start, state_change, client_addr, client_hostname, client_port, backend_xid, backend_xmin, backend_type, ssl, sslversion, sslcipher, sslbits, ssl_client_dn, ssl_client_serial, ssl_issuer_dn, gss_auth, gss_princ, gss_enc, gss_delegation, leader_pid, query_id)
+   FROM pg_stat_get_activity(NULL::integer) s(datid, pid, usesysid, application_name, state, query, wait_event_type, wait_event, xact_start, query_start, backend_start, state_change, client_addr, client_hostname, client_port, backend_xid, backend_xmin, backend_type, ssl, sslversion, sslcipher, sslbits, ssl_client_dn, ssl_client_serial, ssl_issuer_dn, gss_auth, gss_princ, gss_enc, gss_delegation, leader_pid, query_id, rx_socket_bytes, tx_socket_bytes, rx_pq_bytes, tx_pq_bytes)
   WHERE (client_port IS NOT NULL);
 pg_stat_io| SELECT backend_type,
     object,
@@ -1901,6 +1901,13 @@ pg_stat_io| SELECT backend_type,
     fsync_time,
     stats_reset
    FROM pg_stat_get_io() b(backend_type, object, context, reads, read_time, writes, write_time, writebacks, writeback_time, extends, extend_time, op_bytes, hits, evictions, reuses, fsyncs, fsync_time, stats_reset);
+pg_stat_network_traffic| SELECT pid,
+    rx_socket_bytes,
+    tx_socket_bytes,
+    rx_pq_bytes,
+    tx_pq_bytes
+   FROM pg_stat_get_activity(NULL::integer) s(datid, pid, usesysid, application_name, state, query, wait_event_type, wait_event, xact_start, query_start, backend_start, state_change, client_addr, client_hostname, client_port, backend_xid, backend_xmin, backend_type, ssl, sslversion, sslcipher, sslbits, ssl_client_dn, ssl_client_serial, ssl_issuer_dn, gss_auth, gss_princ, gss_enc, gss_delegation, leader_pid, query_id, rx_socket_bytes, tx_socket_bytes, rx_pq_bytes, tx_pq_bytes)
+  WHERE (client_port IS NOT NULL);
 pg_stat_progress_analyze| SELECT s.pid,
     s.datid,
     d.datname,
@@ -2082,7 +2089,7 @@ pg_stat_replication| SELECT s.pid,
     w.sync_priority,
     w.sync_state,
     w.reply_time
-   FROM ((pg_stat_get_activity(NULL::integer) s(datid, pid, usesysid, application_name, state, query, wait_event_type, wait_event, xact_start, query_start, backend_start, state_change, client_addr, client_hostname, client_port, backend_xid, backend_xmin, backend_type, ssl, sslversion, sslcipher, sslbits, ssl_client_dn, ssl_client_serial, ssl_issuer_dn, gss_auth, gss_princ, gss_enc, gss_delegation, leader_pid, query_id)
+   FROM ((pg_stat_get_activity(NULL::integer) s(datid, pid, usesysid, application_name, state, query, wait_event_type, wait_event, xact_start, query_start, backend_start, state_change, client_addr, client_hostname, client_port, backend_xid, backend_xmin, backend_type, ssl, sslversion, sslcipher, sslbits, ssl_client_dn, ssl_client_serial, ssl_issuer_dn, gss_auth, gss_princ, gss_enc, gss_delegation, leader_pid, query_id, rx_socket_bytes, tx_socket_bytes, rx_pq_bytes, tx_pq_bytes)
      JOIN pg_stat_get_wal_senders() w(pid, state, sent_lsn, write_lsn, flush_lsn, replay_lsn, write_lag, flush_lag, replay_lag, sync_priority, sync_state, reply_time) ON ((s.pid = w.pid)))
      LEFT JOIN pg_authid u ON ((s.usesysid = u.oid)));
 pg_stat_replication_slots| SELECT s.slot_name,
@@ -2116,7 +2123,7 @@ pg_stat_ssl| SELECT pid,
     ssl_client_dn AS client_dn,
     ssl_client_serial AS client_serial,
     ssl_issuer_dn AS issuer_dn
-   FROM pg_stat_get_activity(NULL::integer) s(datid, pid, usesysid, application_name, state, query, wait_event_type, wait_event, xact_start, query_start, backend_start, state_change, client_addr, client_hostname, client_port, backend_xid, backend_xmin, backend_type, ssl, sslversion, sslcipher, sslbits, ssl_client_dn, ssl_client_serial, ssl_issuer_dn, gss_auth, gss_princ, gss_enc, gss_delegation, leader_pid, query_id)
+   FROM pg_stat_get_activity(NULL::integer) s(datid, pid, usesysid, application_name, state, query, wait_event_type, wait_event, xact_start, query_start, backend_start, state_change, client_addr, client_hostname, client_port, backend_xid, backend_xmin, backend_type, ssl, sslversion, sslcipher, sslbits, ssl_client_dn, ssl_client_serial, ssl_issuer_dn, gss_auth, gss_princ, gss_enc, gss_delegation, leader_pid, query_id, rx_socket_bytes, tx_socket_bytes, rx_pq_bytes, tx_pq_bytes)
   WHERE (client_port IS NOT NULL);
 pg_stat_subscription| SELECT su.oid AS subid,
     su.subname,
-- 
2.42.0