more efficient metrics_get_all()

.github/actions/pgx-init/action.yml

@@ -6,7 +6,7 @@ inputs:
     required: true
   pg_version:
     description: 'The version of postgres'
-    default: "15.4"
+    default: "15.5"
   pgrx_pg_version:
     description: 'Pgrx pg version alias'
     default: "pg15"

.github/workflows/extension_ci.yml

@@ -51,11 +51,11 @@ jobs:
       fail-fast: false
       matrix:
         pg: [
-          {version: "12.16", pgrx_version: "pg12"},
-          {version: "13.12", pgrx_version: "pg13"},
-          {version: "14.9", pgrx_version: "pg14"},
-          {version: "15.4", pgrx_version: "pg15"},
-          {version: "16.0", pgrx_version: "pg16"}
+          {version: "12.17", pgrx_version: "pg12"},
+          {version: "13.13", pgrx_version: "pg13"},
+          {version: "14.10", pgrx_version: "pg14"},
+          {version: "15.5", pgrx_version: "pg15"},
+          {version: "16.1", pgrx_version: "pg16"}
         ]
     steps:
       - uses: actions/checkout@v2
@@ -91,9 +91,9 @@ jobs:
           git clone https://github.com/pgpartman/pg_partman.git && \
           cd pg_partman && \
           git checkout v4.7.4 && \
-          sudo make install && cd ../
-          cp /usr/share/postgresql/14/extension/pg_partman* ~/.pgrx/${{ matrix.pg.version }}/pgrx-install/share/postgresql/extension/
-          cp /usr/lib/postgresql/14/lib/pg_partman_bgw.so ~/.pgrx/${{ matrix.pg.version }}/pgrx-install/lib/postgresql/
+          sed -i 's|PG_CONFIG = pg_config|PG_CONFIG = ~/.pgrx/${{ matrix.pg.version }}/pgrx-install/bin/pg_config|' Makefile && \
+          make && \
+          make install 
       - name: Run tests
         run: |
           rm -rf ./target/pgrx-test-data-* || true

.github/workflows/extension_upgrade.yml

@@ -56,9 +56,10 @@ jobs:
           git clone https://github.com/pgpartman/pg_partman.git && \
           cd pg_partman && \
           git checkout v4.7.4 && \
-          sudo make install && cd ../
-          cp /usr/share/postgresql/14/extension/pg_partman* ~/.pgrx/15.4/pgrx-install/share/postgresql/extension/
-          cp /usr/lib/postgresql/14/lib/pg_partman_bgw.so ~/.pgrx/15.4/pgrx-install/lib/postgresql/
+          sed -i 's|PG_CONFIG = pg_config|PG_CONFIG = ~/.pgrx/15.5/pgrx-install/bin/pg_config|' Makefile && \
+          make && \
+          make install && \
+          ls -alh /home/runner/.pgrx/15.5/pgrx-install/share/postgresql/extension/pg_partman*
       - name: Run old version (0.14.2)
         run: |
           rm -rf ./target/pgrx-test-data-* || true
@@ -71,6 +72,7 @@ jobs:
           psql $DATABASE_URL -c "CREATE EXTENSION pgmq;"
           psql $DATABASE_URL -c "select * from pgmq_create('test_queue_1')"
           psql $DATABASE_URL -c "select * from pgmq_create_partitioned('test_partitioned_queue_1');"
+          ls -alh /home/runner/.pgrx/15.5/pgrx-install/share/postgresql/extension/pg_partman*
       - name: Checkout branch's version
         env:
           CI_BRANCH: ${{ steps.current-version.outputs.CI_BRANCH }}
@@ -82,6 +84,7 @@ jobs:
           force: true
       - name: Upgrade and run tests
         run: |
+          ls -alh /home/runner/.pgrx/15.5/pgrx-install/share/postgresql/extension/pg_partman*
           export DATABASE_URL=postgres://$USER:postgres@localhost:28815/pgmq
           SQLX_OFFLINE=true cargo pgrx run ${pg_version} --pgcli || true
           psql $DATABASE_URL -c "ALTER EXTENSION pgmq UPDATE;"

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "pgmq"
-version = "0.33.1"
+version = "0.33.2"
 edition = "2021"
 authors = ["Tembo.io"]
 description = "Postgres extension for PGMQ"

sql/pgmq--0.33.1--0.33.2.sql

@@ -0,0 +1,66 @@
+-- returned by pgmq.metrics() and pgmq.metrics_all
+CREATE TYPE pgmq.metrics_result AS (
+    queue_name text,
+    queue_length bigint,
+    newest_msg_age_sec int,
+    oldest_msg_age_sec int,
+    total_messages bigint,
+    scrape_time timestamp with time zone
+);
+
+DROP FUNCTION pgmq."metrics"(text);
+
+-- get metrics for a single queue
+CREATE OR REPLACE FUNCTION pgmq.metrics(queue_name TEXT)
+RETURNS pgmq.metrics_result AS $$
+DECLARE
+    result_row pgmq.metrics_result;
+    query TEXT;
+BEGIN
+    query := FORMAT(
+        $QUERY$
+        WITH q_summary AS (
+            SELECT
+                count(*) as queue_length,
+                EXTRACT(epoch FROM (NOW() - max(enqueued_at)))::int as newest_msg_age_sec,
+                EXTRACT(epoch FROM (NOW() - min(enqueued_at)))::int as oldest_msg_age_sec,
+                NOW() as scrape_time
+            FROM pgmq.q_%I
+        ),
+        all_metrics AS (
+            SELECT CASE
+                WHEN is_called THEN last_value ELSE 0
+                END as total_messages
+            FROM pgmq.q_%I_msg_id_seq
+        )
+        SELECT 
+            '%s' as queue_name,
+            q_summary.queue_length,
+            q_summary.newest_msg_age_sec,
+            q_summary.oldest_msg_age_sec,
+            all_metrics.total_messages,
+            q_summary.scrape_time
+        FROM q_summary, all_metrics
+        $QUERY$,
+        queue_name, queue_name, queue_name
+    );
+    EXECUTE query INTO result_row;
+    RETURN result_row;
+END;
+$$ LANGUAGE plpgsql;
+
+DROP FUNCTION pgmq."metrics_all"();
+
+-- get metrics for all queues
+CREATE OR REPLACE FUNCTION pgmq."metrics_all"()
+RETURNS SETOF pgmq.metrics_result AS $$
+DECLARE
+    row_name RECORD;
+    result_row pgmq.metrics_result;
+BEGIN
+    FOR row_name IN SELECT queue_name FROM pgmq.meta LOOP
+        result_row := pgmq.metrics(row_name.queue_name);
+        RETURN NEXT result_row;
+    END LOOP;
+END;
+$$ LANGUAGE plpgsql;

src/lib.rs

@@ -2,13 +2,10 @@ pgrx::pg_module_magic!();
 
 pub mod api;
 pub mod errors;
-pub mod metrics;
 pub mod partition;
 pub mod removed_functions;
 pub mod util;
 
-use crate::errors::PgmqExtError;
-
 pgrx::extension_sql_file!("./sql_src.sql", name = "bootstrap");
 
 #[cfg(test)]

src/removed_functions.rs

@@ -27,3 +27,7 @@ pub fn pgmq_delete_batch() {}
 pub fn pgmq_send() {}
 #[pg_extern(sql = "")]
 pub fn pgmq_send_batch() {}
+#[pg_extern(sql = "")]
+pub fn pgmq_metrics() {}
+#[pg_extern(sql = "")]
+pub fn pgmq_metrics_all() {}

src/sql_src.sql

@@ -274,3 +274,66 @@ BEGIN
     RETURN QUERY EXECUTE sql USING msgs;
 END;
 $$ LANGUAGE plpgsql;
+
+-- returned by pgmq.metrics() and pgmq.metrics_all
+CREATE TYPE pgmq.metrics_result AS (
+    queue_name text,
+    queue_length bigint,
+    newest_msg_age_sec int,
+    oldest_msg_age_sec int,
+    total_messages bigint,
+    scrape_time timestamp with time zone
+);
+
+-- get metrics for a single queue
+CREATE OR REPLACE FUNCTION pgmq.metrics(queue_name TEXT)
+RETURNS pgmq.metrics_result AS $$
+DECLARE
+    result_row pgmq.metrics_result;
+    query TEXT;
+BEGIN
+    query := FORMAT(
+        $QUERY$
+        WITH q_summary AS (
+            SELECT
+                count(*) as queue_length,
+                EXTRACT(epoch FROM (NOW() - max(enqueued_at)))::int as newest_msg_age_sec,
+                EXTRACT(epoch FROM (NOW() - min(enqueued_at)))::int as oldest_msg_age_sec,
+                NOW() as scrape_time
+            FROM pgmq.q_%I
+        ),
+        all_metrics AS (
+            SELECT CASE
+                WHEN is_called THEN last_value ELSE 0
+                END as total_messages
+            FROM pgmq.q_%I_msg_id_seq
+        )
+        SELECT 
+            '%s' as queue_name,
+            q_summary.queue_length,
+            q_summary.newest_msg_age_sec,
+            q_summary.oldest_msg_age_sec,
+            all_metrics.total_messages,
+            q_summary.scrape_time
+        FROM q_summary, all_metrics
+        $QUERY$,
+        queue_name, queue_name, queue_name
+    );
+    EXECUTE query INTO result_row;
+    RETURN result_row;
+END;
+$$ LANGUAGE plpgsql;
+
+-- get metrics for all queues
+CREATE OR REPLACE FUNCTION pgmq."metrics_all"()
+RETURNS SETOF pgmq.metrics_result AS $$
+DECLARE
+    row_name RECORD;
+    result_row pgmq.metrics_result;
+BEGIN
+    FOR row_name IN SELECT queue_name FROM pgmq.meta LOOP
+        result_row := pgmq.metrics(row_name.queue_name);
+        RETURN NEXT result_row;
+    END LOOP;
+END;
+$$ LANGUAGE plpgsql;