Implement apply_downsample_config and add E2E tests.

Signed-off-by: Harkishen-Singh <[email protected]>

Author: Harkishen-Singh

docs/sql-api.md

@@ -62,10 +62,10 @@ function interval **prom_api.get_default_metric_retention_period**()
 ```
 function boolean **prom_api.get_downsample_old_data**()
 ```
-### prom_api.get_downsampling_state
+### prom_api.get_global_downsampling_state
 Get automatic downsample state
 ```
-function boolean **prom_api.get_downsampling_state**()
+function boolean **prom_api.get_global_downsampling_state**()
 ```
 ### prom_api.get_metric_chunk_interval
 Get the chunk interval for a specific metric, or the default chunk interval if not explicitly set
@@ -192,10 +192,10 @@ function boolean **prom_api.set_default_retention_period**(retention_period inte
 ```
 function void **prom_api.set_downsample_old_data**(_state boolean)
 ```
-### prom_api.set_downsampling_state
+### prom_api.set_global_downsampling_state
 Set automatic-downsampling state for metrics. Downsampled data will be created only if this returns true
 ```
-function void **prom_api.set_downsampling_state**(_state boolean)
+function void **prom_api.set_global_downsampling_state**(_state boolean)
 ```
 ### prom_api.set_metric_chunk_interval
 set a chunk interval for a specific metric (this overrides the default)
@@ -512,11 +512,21 @@ the specified span_id.
 ```
 function TABLE(trace_id trace_id, parent_span_id bigint, span_id bigint, dist integer, path bigint[]) **ps_trace.upstream_spans**(_trace_id trace_id, _span_id bigint, _max_dist integer DEFAULT NULL::integer)
 ```
+### _prom_catalog.add_compression_clause_to_downsample_view
+
+```
+procedure void **_prom_catalog.add_compression_clause_to_downsample_view**(IN _schema text, IN _table_name text)
+```
 ### _prom_catalog.add_job
 A wrapper around public.add_job that introduces jitter to job start times and schedules.
 ```
 function integer **_prom_catalog.add_job**(proc regproc, schedule_interval interval, config jsonb DEFAULT NULL::jsonb)
 ```
+### _prom_catalog.apply_downsample_config
+
+```
+function void **_prom_catalog.apply_downsample_config**(config jsonb)
+```
 ### _prom_catalog.attach_series_partition
 
 ```
@@ -581,7 +591,7 @@ function record **_prom_catalog.create_label_key**(new_key text, OUT id integer,
 ### _prom_catalog.create_metric_downsampling_view
 
 ```
-function boolean **_prom_catalog.create_metric_downsampling_view**(_schema text, _metric_name text, _table_name text, _resolution interval)
+function boolean **_prom_catalog.create_metric_downsampling_view**(_schema text, _metric_name text, _table_name text, _interval interval)
 ```
 ### _prom_catalog.create_metric_table
 
@@ -593,11 +603,6 @@ function record **_prom_catalog.create_metric_table**(metric_name_arg text, OUT
 ```
 function boolean **_prom_catalog.create_metric_view**(metric_name text)
 ```
-### _prom_catalog.create_or_update_downsampling
-
-```
-procedure void **_prom_catalog.create_or_update_downsampling**(IN _schema_name text, IN _resolution interval, IN _retention interval)
-```
 ### _prom_catalog.create_series
 
 ```
@@ -651,17 +656,17 @@ procedure void **_prom_catalog.do_decompress_chunks_after**(IN metric_table text
 ### _prom_catalog.downsample_counter
 
 ```
-procedure void **_prom_catalog.downsample_counter**(IN _schema text, IN _table_name text, IN _resolution interval)
+procedure void **_prom_catalog.downsample_counter**(IN _schema text, IN _table_name text, IN _interval interval)
 ```
 ### _prom_catalog.downsample_gauge
 
 ```
-procedure void **_prom_catalog.downsample_gauge**(IN _schema text, IN _table_name text, IN _resolution interval)
+procedure void **_prom_catalog.downsample_gauge**(IN _schema text, IN _table_name text, IN _interval interval)
 ```
 ### _prom_catalog.downsample_summary
 
 ```
-procedure void **_prom_catalog.downsample_summary**(IN _schema text, IN _table_name text, IN _resolution interval)
+procedure void **_prom_catalog.downsample_summary**(IN _schema text, IN _table_name text, IN _interval interval)
 ```
 ### _prom_catalog.drop_metric_chunk_data
 drop chunks from schema_name.metric_name containing data older than older_than.
@@ -1126,11 +1131,6 @@ function void **_prom_catalog.unlock_for_vacuum_engine**()
 ```
 function void **_prom_catalog.unlock_metric_for_maintenance**(metric_id integer)
 ```
-### _prom_catalog.update_downsampling_state
-
-```
-function void **_prom_catalog.update_downsampling_state**(_schema_name text, _should_refresh boolean)
-```
 ### _prom_catalog.update_execute_everywhere_entry
 
 ```

migration/idempotent/010-telemetry.sql

@@ -101,11 +101,11 @@ $$
         PERFORM _ps_catalog.apply_telemetry('metrics_default_chunk_interval', result);
 
         -- Metric downsampling.
-        SELECT prom_api.get_automatic_downsample()::TEXT INTO result;
-        PERFORM _ps_catalog.apply_telemetry('metrics_automatic_downsampling_enabled', result);
+        SELECT prom_api.get_global_downsampling_state()::TEXT INTO result;
+        PERFORM _ps_catalog.apply_telemetry('metrics_downsampling_enabled', result);
 
-        SELECT array_agg(resolution || ':' || retention)::TEXT INTO result FROM _prom_catalog.rollup; -- Example: {00:05:00:720:00:00,01:00:00:8760:00:00} => {HH:MM:SS}
-        PERFORM _ps_catalog.apply_telemetry('metrics_automatic_downsampling_resolutions', result);
+        SELECT array_agg(ds_interval || ':' || retention)::TEXT INTO result FROM _prom_catalog.downsample; -- Example: {00:05:00:720:00:00,01:00:00:8760:00:00} => {HH:MM:SS}
+        PERFORM _ps_catalog.apply_telemetry('metrics_downsampling_configs', result);
 
         IF ( SELECT count(*)>0 FROM _prom_catalog.metric WHERE metric_name = 'prometheus_tsdb_head_series' ) THEN
             -- Calculate active series in Promscale. This is done by taking the help of the Prometheus metric 'prometheus_tsdb_head_series'.

migration/idempotent/017-metric-downsampling.sql

@@ -1,5 +1,5 @@
 -- Metric downsampling by metric type
-CREATE OR REPLACE PROCEDURE _prom_catalog.downsample_gauge(_schema TEXT, _table_name TEXT, _resolution INTERVAL)
+CREATE OR REPLACE PROCEDURE _prom_catalog.downsample_gauge(_schema TEXT, _table_name TEXT, _interval INTERVAL)
     SET search_path = pg_catalog, pg_temp
 AS
 $$
@@ -18,12 +18,12 @@ BEGIN
                 max(value) as max
             FROM prom_data.%2$I
             GROUP BY public.time_bucket(%4$L, time), series_id WITH NO DATA
-    ', _schema, _table_name, 'UTC', _resolution::text);
+    ', _schema, _table_name, 'UTC', _interval::text);
 END;
 $$
 LANGUAGE plpgsql;
 
-CREATE OR REPLACE PROCEDURE _prom_catalog.downsample_counter(_schema TEXT, _table_name TEXT, _resolution INTERVAL)
+CREATE OR REPLACE PROCEDURE _prom_catalog.downsample_counter(_schema TEXT, _table_name TEXT, _interval INTERVAL)
     SET search_path = pg_catalog, pg_temp
 AS
 $$
@@ -41,12 +41,12 @@ BEGIN
                 _prom_catalog.irate(array_agg(value)) irate
             FROM prom_data.%2$I
             GROUP BY public.time_bucket(%4$L, time), series_id WITH NO DATA
-    ', _schema, _table_name, 'UTC', _resolution::text);
+    ', _schema, _table_name, 'UTC', _interval::text);
 END;
 $$
 LANGUAGE PLPGSQL;
 
-CREATE OR REPLACE PROCEDURE _prom_catalog.downsample_summary(_schema TEXT, _table_name TEXT, _resolution INTERVAL)
+CREATE OR REPLACE PROCEDURE _prom_catalog.downsample_summary(_schema TEXT, _table_name TEXT, _interval INTERVAL)
     SET search_path = pg_catalog, pg_temp
 AS
 $$
@@ -63,7 +63,7 @@ BEGIN
                 count(value) as count
             FROM prom_data.%2$I
             GROUP BY public.time_bucket(%4$L, time), series_id WITH NO DATA
-    ', _schema, _table_name, 'UTC', _resolution::text);
+    ', _schema, _table_name, 'UTC', _interval::text);
 END;
 $$
 LANGUAGE PLPGSQL;
@@ -104,7 +104,7 @@ LANGUAGE PLPGSQL;
 
 -- create_metric_downsampling_view decides which query should be used for downsampling the given metric depending of metric type
 -- and calls the respective creation function. It returns true if metric downsampling view was created.
-CREATE OR REPLACE FUNCTION _prom_catalog.create_metric_downsampling_view(_schema TEXT, _metric_name TEXT, _table_name TEXT, _resolution INTERVAL)
+CREATE OR REPLACE FUNCTION _prom_catalog.create_metric_downsampling_view(_schema TEXT, _metric_name TEXT, _table_name TEXT, _interval INTERVAL)
     RETURNS BOOLEAN
     SET search_path = pg_catalog, pg_temp
 AS
@@ -129,12 +129,12 @@ BEGIN
 
     CASE
         WHEN _metric_type = 'GAUGE' THEN
-            CALL _prom_catalog.downsample_gauge(_schema, _table_name, _resolution);
+            CALL _prom_catalog.downsample_gauge(_schema, _table_name, _interval);
         WHEN _metric_type = 'COUNTER' OR _metric_type = 'HISTOGRAM' THEN
-            CALL _prom_catalog.downsample_counter(_schema, _table_name, _resolution);
+            CALL _prom_catalog.downsample_counter(_schema, _table_name, _interval);
             _default_query_column := 'last';
         WHEN _metric_type = 'SUMMARY' THEN
-            CALL _prom_catalog.downsample_summary(_schema, _table_name, _resolution);
+            CALL _prom_catalog.downsample_summary(_schema, _table_name, _interval);
         ELSE
             RAISE WARNING '[Downsampling] Skipping creation of metric downsampling for %. REASON: invalid metric_type. Wanted {GAUGE, COUNTER, HISTOGRAM, SUMMARY}, received %', _metric_name, _metric_type;
             RETURN FALSE;
@@ -146,7 +146,7 @@ END;
 $$
 LANGUAGE PLPGSQL;
 
--- scan_for_new_downsampling_views is called in regular intervals to scan for either new metrics or for new downsampling resolutions
+-- scan_for_new_downsampling_views is called in regular intervals to scan for either new metrics or for new downsampling configs
 -- and create metric downsampling for them.
 CREATE OR REPLACE PROCEDURE _prom_catalog.scan_for_new_downsampling_views(job_id int, config jsonb)
 AS
@@ -162,7 +162,7 @@ BEGIN
     -- Note: We cannot use SET in the procedure declaration because we do transaction control
     -- and we can _only_ use SET LOCAL in a procedure which _does_ transaction control
     SET LOCAL search_path = pg_catalog, pg_temp;
-    IF ( SELECT prom_api.get_downsampling_state()::BOOLEAN IS FALSE ) THEN
+    IF ( SELECT prom_api.get_global_downsampling_state()::BOOLEAN IS FALSE ) THEN
         RETURN;
     END IF;
 
@@ -176,7 +176,7 @@ BEGIN
                 SELECT 1 FROM _prom_catalog.metric_downsample md WHERE md.downsample_id = d.id AND md.metric_id = mt.id
             ) ORDER BY mt.id -- Get metric names that have a pending downsampling creation.
         LOOP
-            SELECT INTO downsampling_view_created _prom_catalog.create_metric_downsampling_view(d.schema_name, m.metric_name, m.table_name, d.resolution);
+            SELECT INTO downsampling_view_created _prom_catalog.create_metric_downsampling_view(d.schema_name, m.metric_name, m.table_name, d.ds_interval);
             IF downsampling_view_created THEN
                 INSERT INTO _prom_catalog.metric_downsample(downsample_id, metric_id, refresh_pending) VALUES (d.id, m.id, TRUE);
                 downsampling_view_count := downsampling_view_count + 1;
@@ -199,7 +199,7 @@ BEGIN
             md.id as metric_downsample_id,
             dwn.schema_name as schema_name,
             mt.table_name as table_name,
-            dwn.resolution as resolution,
+            dwn.ds_interval as ds_interval,
             dwn.id as downsample_id
         FROM _prom_catalog.metric_downsample md
                  INNER JOIN _prom_catalog.downsample dwn ON (md.downsample_id = dwn.id)
@@ -219,7 +219,7 @@ BEGIN
         -- 2. In case of non-graceful shutdown, we will successfully register the non-registered downsampling viws the
         --      next time this procedure is called. Earlier, this was done using temp table, which did not
         --      provide this guarantee
-        PERFORM prom_api.register_metric_view(d.schema_name, d.table_name, d.resolution, false, d.downsample_id);
+        PERFORM prom_api.register_metric_view(d.schema_name, d.table_name, d.ds_interval, false, d.downsample_id);
         COMMIT;
         SET LOCAL search_path = pg_catalog, pg_temp;
     END LOOP;
@@ -230,6 +230,7 @@ LANGUAGE PLPGSQL;
 -- apply_downsample_config does 2 things:
 -- 1. Create new downsampling configurations based on the given config
 -- 2. Update the existing downsampling config in terms of retention, enabling/disabling downsampling config
+-- The given config must be an array of <schema_name, ds_interval, retention>
 CREATE OR REPLACE FUNCTION _prom_catalog.apply_downsample_config(config jsonb)
 RETURNS VOID
     SET search_path = pg_catalog, pg_temp
@@ -251,8 +252,8 @@ BEGIN
     END LOOP;
 
     -- Insert or update the retention duration and enable the downsampling if disabled.
-    INSERT INTO _prom_catalog.downsample (schema_name, resolution, retention, should_refresh)
-        SELECT a.schema_name, a.resolution, a.retention, TRUE FROM unnest(_input) a
+    INSERT INTO _prom_catalog.downsample (schema_name, ds_interval, retention, should_refresh)
+        SELECT a.schema_name, a.ds_interval, a.retention, TRUE FROM unnest(_input) a
             ON CONFLICT (schema_name) DO UPDATE
                 SET
                     retention = excluded.retention,
@@ -265,6 +266,7 @@ BEGIN
 END;
 $$
 LANGUAGE PLPGSQL;
+GRANT EXECUTE ON FUNCTION _prom_catalog.apply_downsample_config(jsonb) TO prom_writer;
 
 -- delete_downsampling deletes everything related to the given downsampling schema name. It does the following in order:
 -- 1. Delete the entries in _prom_catalog.downsample
@@ -291,42 +293,27 @@ $$
 LANGUAGE PLPGSQL;
 GRANT EXECUTE ON PROCEDURE _prom_catalog.delete_downsampling(text) TO prom_writer;
 
-CREATE OR REPLACE FUNCTION _prom_catalog.update_downsampling_state(_schema_name TEXT, _should_refresh BOOLEAN)
-    RETURNS VOID
-    SET search_path = pg_catalog, pg_temp
-AS
-$$
-BEGIN
-    UPDATE _prom_catalog.downsample SET should_refresh = _should_refresh WHERE schema_name = _schema_name;
-    IF NOT FOUND THEN
-        RAISE EXCEPTION 'No entry found for schema %', _schema_name;
-    END IF;
-END;
-$$
-LANGUAGE PLPGSQL;
-GRANT EXECUTE ON FUNCTION _prom_catalog.update_downsampling_state(TEXT, BOOLEAN) TO prom_admin;
-
-CREATE OR REPLACE FUNCTION prom_api.set_downsampling_state(_state BOOLEAN)
+CREATE OR REPLACE FUNCTION prom_api.set_global_downsampling_state(_state BOOLEAN)
 RETURNS VOID
 SET search_path = pg_catalog, pg_temp
 AS $$
     SELECT _prom_catalog.set_default_value('downsample', _state::text);
 $$
 LANGUAGE SQL;
-COMMENT ON FUNCTION prom_api.set_downsampling_state(BOOLEAN)
+COMMENT ON FUNCTION prom_api.set_global_downsampling_state(BOOLEAN)
     IS 'Set automatic-downsampling state for metrics. Downsampled data will be created only if this returns true';
-GRANT EXECUTE ON FUNCTION prom_api.set_downsampling_state(BOOLEAN) TO prom_admin;
+GRANT EXECUTE ON FUNCTION prom_api.set_global_downsampling_state(BOOLEAN) TO prom_admin;
 
-CREATE OR REPLACE FUNCTION prom_api.get_downsampling_state()
+CREATE OR REPLACE FUNCTION prom_api.get_global_downsampling_state()
 RETURNS BOOLEAN
 SET search_path = pg_catalog, pg_temp
 AS $$
     SELECT _prom_catalog.get_default_value('downsample')::boolean;
 $$
 LANGUAGE SQL;
-COMMENT ON FUNCTION prom_api.get_downsampling_state()
+COMMENT ON FUNCTION prom_api.get_global_downsampling_state()
     IS 'Get automatic downsample state';
-GRANT EXECUTE ON FUNCTION prom_api.get_downsampling_state() TO prom_admin;
+GRANT EXECUTE ON FUNCTION prom_api.get_global_downsampling_state() TO prom_admin;
 
 CREATE OR REPLACE FUNCTION prom_api.set_downsample_old_data(_state BOOLEAN)
 RETURNS VOID

migration/idempotent/018-metric-downsampling-maintenance.sql

@@ -5,7 +5,7 @@ DECLARE
     _refresh_interval INTERVAL;
     _active_chunk_interval INTERVAL;
     _ignore_downsampling_clause TEXT := '';
-    _refresh_downsampled_data BOOLEAN := (SELECT prom_api.get_downsampling_state()::BOOLEAN);
+    _refresh_downsampled_data BOOLEAN := (SELECT prom_api.get_global_downsampling_state()::BOOLEAN);
     _should_refresh_clause TEXT := 'AND (SELECT should_refresh FROM _prom_catalog.downsample WHERE id = m.downsample_id) = TRUE';
     _safety_refresh_start_buffer INTERVAL := INTERVAL '0 minutes';
     r RECORD;
@@ -16,7 +16,7 @@ BEGIN
     _active_chunk_interval := (SELECT _prom_catalog.get_default_chunk_interval()::INTERVAL * 2);
     _refresh_interval := (config ->> 'refresh_interval')::INTERVAL;
     IF _refresh_interval IS NULL THEN
-        RAISE EXCEPTION 'refresh_interval cannot be null';
+        RAISE EXCEPTION 'Caggs refresh policy: refresh_interval cannot be null';
     END IF;
 
     IF ( SELECT _refresh_interval < INTERVAL '30 minutes' ) THEN
@@ -25,7 +25,7 @@ BEGIN
         -- of refreshing data, a system with a lot of metrics and series might take > 10 minutes to refresh a 5 minute downsampling.
         -- This will lead to data loss. Hence, we give a refresh buffer of 30 minutes early to start.
         --
-        -- This problem is not seen in large downsampling resolutions like 1 hour, since the 2 buckets duration gives 2 hours
+        -- This problem is not seen in large downsampling intervals like 1 hour, since the 2 buckets duration gives 2 hours
         -- of time to complete refresh, which is usually sufficient.
         _safety_refresh_start_buffer := INTERVAL '30 minutes';
     END IF;
@@ -99,6 +99,11 @@ DECLARE
     r RECORD;
 
 BEGIN
+    IF NOT (SELECT prom_api.get_global_downsampling_state()::BOOLEAN) THEN
+        -- Do not compress any Cagg if downsampling state is false.
+        RETURN;
+    END IF;
+
     FOR r IN
         SELECT
             format('%I.%I', m.table_schema, m.table_name) AS compressible_cagg,

migration/incremental/036-metric-downsample.sql

@@ -1,11 +1,11 @@
 CREATE TABLE _prom_catalog.downsample (
     id             SERIAL PRIMARY KEY,
     schema_name    TEXT NOT NULL,
-    resolution     INTERVAL NOT NULL,
+    ds_interval     INTERVAL NOT NULL,
     retention      INTERVAL NOT NULL,
     should_refresh BOOLEAN DEFAULT FALSE,
     UNIQUE(schema_name),
-    UNIQUE(resolution) -- To avoid ds_1m and ds_60s, though technically, both are the same.
+    UNIQUE(ds_interval) -- To avoid ds_1m and ds_60s, as technically, both are the same.
 );
 GRANT SELECT ON TABLE _prom_catalog.downsample TO prom_reader;
 GRANT SELECT, INSERT, UPDATE, DELETE ON TABLE _prom_catalog.downsample TO prom_writer;

sql-tests/testdata/custom_downsample.sql

@@ -29,7 +29,7 @@ SELECT register_metric_view('custom', 'test', INTERVAL '30 minutes');
 SELECT ok(count(*) = 337, 'samples in custom downsample') FROM custom.test;
 
 
--- # Test compressing metric-rollups.
+-- # Test compressing metric downsampling.
 
 \i 'testdata/scripts/test-helpers.sql'
 
@@ -46,7 +46,7 @@ ALTER MATERIALIZED VIEW custom.test SET (timescaledb.compress = true);
 CALL _prom_catalog.execute_caggs_compression_policy(2, '{}'::jsonb);
 SELECT ok(compressed_chunks_exist('custom.test') = true);
 
--- # Test retaining metric-rollups.
+-- # Test retaining metric downsampling.
 
 SELECT ok(count(*) > 0) FROM (SELECT show_chunks('custom.test')) a;
 

sql-tests/testdata/custom_downsample_refresh.sql

@@ -45,7 +45,7 @@ FROM generate_series(
 -- # TEST refreshing custom downsample.
 
 -- Check samples before calling the refresh func.
-SELECT ok(count(*) = 344, 'samples in custom downsample before refresh') FROM custom.test; -- Samples increased even before calling refresh func. My guess is that this is due to timescaledb.materalized behaviour.
+SELECT ok(count(*) = 344, 'samples in custom downsample before refresh') FROM custom.test; -- Samples increased even before calling refresh func. My guess is that this is due to timescaledb.materialized behaviour.
 
 CALL _prom_catalog.execute_caggs_refresh_policy(1, json_build_object('refresh_interval', interval '30 minutes')::jsonb);
 

sql-tests/testdata/maintenance_jobs_separation.sql

@@ -52,7 +52,7 @@ WHERE proc_schema = '_prom_catalog'
   AND NOT (proc_name = 'scan_for_new_downsampling_views' OR proc_name = 'execute_caggs_compression_policy' OR proc_name = 'execute_caggs_retention_policy')
   AND NOT coalesce(config, '{}'::jsonb) ?& ARRAY ['signal', 'type'];
 
-SELECT ok(COUNT(*) = 3, 'Only the new-style configurations should be present')
+SELECT ok(COUNT(*) = 6, 'Only the new-style configurations should be present')
 FROM timescaledb_information.jobs
 WHERE proc_schema = '_prom_catalog'
   AND (schedule_interval >= '10 min' OR schedule_interval < '15 min');