fix: add time-based filtering and retention to /query_info_metrics export

PostgREST generates unique queryids at ~7K/min, causing pgss_queryid_queries
to grow unboundedly (520K+ rows/hour). The /query_info_metrics endpoint scans
all rows with DISTINCT ON, exceeding the 30s VictoriaMetrics scrape timeout
and breaking Dashboard 2's group_left() join.

Three-part fix:
- Trigger: UPDATE time on duplicate instead of silently skipping, so active
  queryids (still in pg_stat_statements) have recent timestamps
- Time filter: only export queryids seen in the last 10 minutes (~5K rows
  instead of 520K), well within the 30s scrape timeout
- Retention cleanup: delete entries older than 24h in batches of 10K per
  scrape to prevent unbounded table growth

Also applies trigger migration at Flask startup (idempotent) for existing
deployments, and adds time filtering to get_query_texts_from_sink().

https://claude.ai/code/session_01SzJxzZNQjDQphaHyaX3RU7

Author: claude

config/sink-postgres/init.sql

@@ -55,29 +55,31 @@ declare
 begin
   -- Extract queryid from the data JSONB
   queryid_value := new.data->>'queryid';
-  
+
   -- Allow NULL queryids through
   if queryid_value is null then
     return new;
   end if;
-  
-  -- Silently skip if duplicate exists
-  if exists (
-    select
-    from pgss_queryid_queries
-    where
-      dbname = new.dbname
-      and data->>'queryid' = queryid_value
-    limit 1
-  ) then
-    return null;  -- Cancels INSERT silently
+
+  -- If duplicate exists, refresh its timestamp (marks queryid as still active
+  -- in pg_stat_statements) and skip the INSERT
+  update pgss_queryid_queries
+  set time = new.time
+  where dbname = new.dbname
+    and data->>'queryid' = queryid_value;
+
+  if found then
+    return null;  -- Duplicate updated, cancel INSERT
   end if;
-  
-  return new;
+
+  return new;  -- New queryid, proceed with INSERT
 end;
 $$ language plpgsql;
 
 
+-- Allow pgwatch to update this function (Flask app applies trigger migrations at startup)
+alter function enforce_queryid_uniqueness() owner to pgwatch;
+
 create or replace trigger enforce_queryid_uniqueness_trigger
   before insert
   on pgss_queryid_queries

monitoring_flask_backend/app.py

@@ -187,6 +187,68 @@ def smart_truncate_query(query: str, max_length: int = 40) -> str:
 # Prometheus connection - use environment variable with fallback
 PROMETHEUS_URL = os.environ.get('PROMETHEUS_URL', 'http://localhost:8428')
 
+# Retention: how long to keep stale queryid entries (queryids no longer in pg_stat_statements)
+QUERYID_RETENTION_HOURS = int(os.environ.get('QUERYID_RETENTION_HOURS', '24'))
+
+# How recent a queryid must be to appear in /query_info_metrics (must be > pgwatch collection interval)
+QUERYID_ACTIVE_MINUTES = int(os.environ.get('QUERYID_ACTIVE_MINUTES', '10'))
+
+
+def _apply_trigger_migration():
+    """
+    Apply trigger migration: update queryid timestamps on duplicate instead of
+    silently skipping. This lets us distinguish active queryids (recently seen in
+    pg_stat_statements) from stale ones, enabling time-based filtering and retention.
+
+    Safe to call multiple times (CREATE OR REPLACE FUNCTION is idempotent).
+    """
+    conn = None
+    try:
+        conn = psycopg2.connect(POSTGRES_SINK_URL)
+        conn.autocommit = True
+        with conn.cursor() as cursor:
+            cursor.execute("""
+                CREATE OR REPLACE FUNCTION enforce_queryid_uniqueness()
+                RETURNS trigger AS $$
+                DECLARE
+                  queryid_value text;
+                BEGIN
+                  queryid_value := new.data->>'queryid';
+
+                  IF queryid_value IS NULL THEN
+                    RETURN new;
+                  END IF;
+
+                  -- If duplicate exists, refresh its timestamp (marks queryid as
+                  -- still active in pg_stat_statements) and skip the INSERT
+                  UPDATE pgss_queryid_queries
+                  SET time = new.time
+                  WHERE dbname = new.dbname
+                    AND data->>'queryid' = queryid_value;
+
+                  IF FOUND THEN
+                    RETURN NULL;
+                  END IF;
+
+                  RETURN new;
+                END;
+                $$ LANGUAGE plpgsql;
+            """)
+        logger.info("Applied trigger migration: enforce_queryid_uniqueness now refreshes timestamps")
+    except Exception as e:
+        logger.warning(
+            f"Failed to apply trigger migration: {e}. "
+            "If 'must be owner' error, run with postgres superuser: "
+            "ALTER FUNCTION enforce_queryid_uniqueness() OWNER TO pgwatch;"
+        )
+    finally:
+        if conn:
+            conn.close()
+
+
+# Apply trigger migration at startup (idempotent)
+_apply_trigger_migration()
+
 # Metric name mapping for cleaner CSV output
 METRIC_NAME_MAPPING = {
     'calls': 'calls',
@@ -230,17 +292,22 @@ def get_prometheus_client():
         raise
 
 
-def get_query_texts_from_sink(db_name: str = None, truncation_mode: str = 'smart') -> dict:
+def get_query_texts_from_sink(db_name: str = None, truncation_mode: str = 'smart',
+                              max_age_hours: int = None) -> dict:
     """
     Fetch queryid-to-query text mappings from the PostgreSQL sink database.
 
     Args:
         db_name: Optional database name to filter results
         truncation_mode: 'smart' for smart truncation, 'raw' for simple truncation
+        max_age_hours: Only return queryids seen within this many hours (None = use retention window)
 
     Returns:
         Dictionary mapping queryid to query text
     """
+    if max_age_hours is None:
+        max_age_hours = QUERYID_RETENTION_HOURS
+
     query_texts = {}
 
     conn = None
@@ -257,23 +324,25 @@ def get_query_texts_from_sink(db_name: str = None, truncation_mode: str = 'smart
                     FROM public.pgss_queryid_queries
                     WHERE
                         dbname = %s
+                        AND time > now() - interval '%s hours'
                         AND data->>'queryid' IS NOT NULL
                         AND data->>'query' IS NOT NULL
                     ORDER BY data->>'queryid', time DESC
                 """
-                cursor.execute(query, (db_name,))
+                cursor.execute(query, (db_name, max_age_hours))
             else:
                 query = """
                     SELECT DISTINCT ON (data->>'queryid')
                         data->>'queryid' as queryid,
                         data->>'query' as query
                     FROM public.pgss_queryid_queries
                     WHERE
-                        data->>'queryid' IS NOT NULL
+                        time > now() - interval '%s hours'
+                        AND data->>'queryid' IS NOT NULL
                         AND data->>'query' IS NOT NULL
                     ORDER BY data->>'queryid', time DESC
                 """
-                cursor.execute(query)
+                cursor.execute(query, (max_age_hours,))
 
             for row in cursor:
                 queryid = row['queryid']
@@ -1284,7 +1353,28 @@ def get_query_info_metrics():
         try:
             conn = psycopg2.connect(POSTGRES_SINK_URL)
             with conn.cursor(cursor_factory=psycopg2.extras.DictCursor) as cursor:
-                # Skip db_name filter if it's empty, "All", or contains special chars
+                # Retention cleanup: delete stale entries in batches to prevent
+                # unbounded table growth from high-churn queryid sources (e.g. PostgREST)
+                try:
+                    cursor.execute("""
+                        DELETE FROM public.pgss_queryid_queries
+                        WHERE ctid IN (
+                            SELECT ctid FROM public.pgss_queryid_queries
+                            WHERE time < now() - interval '%s hours'
+                            LIMIT 10000
+                        )
+                    """, (QUERYID_RETENTION_HOURS,))
+                    deleted = cursor.rowcount
+                    conn.commit()
+                    if deleted > 0:
+                        logger.info(f"Retention cleanup: deleted {deleted} stale queryid entries")
+                except Exception as e:
+                    logger.warning(f"Retention cleanup failed: {e}")
+                    conn.rollback()
+
+                # Only export queryids recently seen in pg_stat_statements.
+                # The dedup trigger refreshes the timestamp on each collection,
+                # so active queryids have time within the last few minutes.
                 use_db_filter = db_name and db_name.lower() not in ('all', '') and not db_name.startswith('$')
                 if use_db_filter:
                     query = """
@@ -1294,23 +1384,25 @@ def get_query_info_metrics():
                         FROM public.pgss_queryid_queries
                         WHERE
                             dbname = %s
+                            AND time > now() - interval '%s minutes'
                             AND data->>'queryid' IS NOT NULL
                             AND data->>'query' IS NOT NULL
                         ORDER BY data->>'queryid', time DESC
                     """
-                    cursor.execute(query, (db_name,))
+                    cursor.execute(query, (db_name, QUERYID_ACTIVE_MINUTES))
                 else:
                     query = """
                         SELECT DISTINCT ON (data->>'queryid')
                             data->>'queryid' as queryid,
                             data->>'query' as query
                         FROM public.pgss_queryid_queries
                         WHERE
-                            data->>'queryid' IS NOT NULL
+                            time > now() - interval '%s minutes'
+                            AND data->>'queryid' IS NOT NULL
                             AND data->>'query' IS NOT NULL
                         ORDER BY data->>'queryid', time DESC
                     """
-                    cursor.execute(query)
+                    cursor.execute(query, (QUERYID_ACTIVE_MINUTES,))
 
                 for row in cursor:
                     queryid = row['queryid']
@@ -1355,6 +1447,8 @@ def get_query_info_metrics():
                             'full': full,
                             'queryid_only': queryid_only,
                         }
+
+                logger.info(f"Exported {len(query_data)} active queryids for metrics")
         except Exception as e:
             logger.warning(f"Failed to fetch query texts from sink database: {e}")
         finally:

monitoring_flask_backend/test_app.py

@@ -523,7 +523,7 @@ def test_db_name_filter(self, mock_connect, client):
         response = client.get('/query_info_metrics?db_name=mydb')
         assert response.status_code == 200
 
-        # Verify the query was called with db_name filter
+        # Verify the main query (last execute call) was called with db_name filter
         call_args = mock_cursor.execute.call_args
         assert call_args is not None
         query = call_args[0][0]
@@ -538,7 +538,7 @@ def test_db_name_all_skips_filter(self, mock_connect, client):
         response = client.get('/query_info_metrics?db_name=all')
         assert response.status_code == 200
 
-        # Verify the query was called without db_name filter
+        # Verify the main query (last execute call) was called without db_name filter
         call_args = mock_cursor.execute.call_args
         query = call_args[0][0]
         assert 'dbname = %s' not in query
@@ -552,11 +552,42 @@ def test_db_name_variable_skips_filter(self, mock_connect, client):
         response = client.get('/query_info_metrics?db_name=$db_name')
         assert response.status_code == 200
 
-        # Verify the query was called without db_name filter
+        # Verify the main query (last execute call) was called without db_name filter
         call_args = mock_cursor.execute.call_args
         query = call_args[0][0]
         assert 'dbname = %s' not in query
 
+    @patch('app.psycopg2.connect')
+    def test_time_filter_applied(self, mock_connect, client):
+        """Test that the time filter is applied to only export active queryids."""
+        mock_cursor = mock_connect.return_value.cursor.return_value.__enter__.return_value
+        mock_cursor.__iter__ = lambda self: iter([])
+
+        response = client.get('/query_info_metrics')
+        assert response.status_code == 200
+
+        # Verify the main query includes a time filter
+        call_args = mock_cursor.execute.call_args
+        query = call_args[0][0]
+        assert "interval" in query
+        assert "minutes" in query
+
+    @patch('app.psycopg2.connect')
+    def test_retention_cleanup_runs(self, mock_connect, client):
+        """Test that retention cleanup DELETE runs before the main query."""
+        mock_cursor = mock_connect.return_value.cursor.return_value.__enter__.return_value
+        mock_cursor.__iter__ = lambda self: iter([])
+        mock_cursor.rowcount = 0
+
+        response = client.get('/query_info_metrics')
+        assert response.status_code == 200
+
+        # Verify that execute was called at least twice (cleanup + main query)
+        assert mock_cursor.execute.call_count >= 2
+        # First call should be the retention DELETE
+        first_call_query = mock_cursor.execute.call_args_list[0][0][0]
+        assert 'DELETE' in first_call_query
+
     @patch('app.psycopg2.connect')
     def test_handles_db_connection_error(self, mock_connect, client):
         """Test endpoint handles database connection errors gracefully."""