Add SQL views

Add SQL views for the prom_series.* and prom_metric.* views.
They are described in detail in sql_schema.md. The point
of these views is to make working with prom data in SQL
easier.

This PR also moves all SQL code to 1_base_schema.up.sql since
we aren't supporting upgrades yet.

Author: cevian

pkg/docs/sql_schema.md

@@ -0,0 +1,98 @@
+# Data Model Schema
+
+## Views
+
+We define several views to make working with prometheus data easier.
+
+### Metric Views
+
+Metric views allows access to the full time-series prometheus data for a
+given metric. By default, these views are found in the `prom_metric` schema.
+Each metric has a view named after the metric name (.e.g. the `cpu_usage`
+metric would have a `prom_metric.cpu_usage` view). The view contains a the
+following column:
+
+ - time - The timestamp of the measurement
+ - value - The value of the measurement
+ - series_id - The ID of the series
+ - labels - The array of label ids
+ - plus a column for each label name's id in the metric's label set
+
+For example:
+```
+# \d+ prom_metric.cpu_usage
+                                   View "prom_metric.cpu_usage"
+    Column    |           Type           | Collation | Nullable | Default | Storage  | Description
+--------------+--------------------------+-----------+----------+---------+----------+-------------
+ time         | timestamp with time zone |           |          |         | plain    |
+ value        | double precision         |           |          |         | plain    |
+ series_id    | integer                  |           |          |         | plain    |
+ labels       | integer[]                |           |          |         | extended |
+ namespace_id | integer                  |           |          |         | plain    |
+ node_id      | integer                  |           |          |         | plain    |
+```
+
+Example query for single point with their labels:
+SELECT
+    label_array_to_jsonb(labels) as labels,
+    value
+FROM prom_metric.cpu_usage
+WHERE time < now();
+
+```
+                     labels                   | value
+----------------------------------------------+-------
+ {"node": "brain", "namespace": "production"} |   0.5
+ {"node": "brain", "namespace": "production"} |   0.6
+ {"node": "pinky", "namespace": "dev"}        |   0.1
+ {"node": "pinky", "namespace": "dev"}        |   0.2
+```
+
+Example query for a rollup:
+
+SELECT
+   get_label_value(node_id) as node,
+   avg(value)
+FROM prom_metric.cpu_usage
+WHERE time < now()
+GROUP BY node_id
+
+```
+ node  | avg
+-------+------
+ brain | 0.55
+ pinky | 0.15
+```
+
+### Series Views
+
+The series views allows exploration of the series present for a given metric.
+By default, these views are found in the `prom_series` schema. Each metric
+has a view named after the metric name (.e.g. the `cpu_usage` metric would
+have a `prom_series.cpu_usage` view). The view contains a the following
+column:
+
+- series_id
+- labels
+- plus a column for each label name's value in the metric's label set
+
+For example:
+```
+# \d+ prom_series.cpu_usage
+                          View "prom_series.cpu_usage"
+  Column   |   Type    | Collation | Nullable | Default | Storage  | Description
+-----------+-----------+-----------+----------+---------+----------+-------------
+ series_id | bigint    |           |          |         | plain    |
+ labels    | integer[] |           |          |         | extended |
+ namespace | text      |           |          |         | extended |
+ node      | text      |           |          |         | extended |
+```
+
+Example query to look at all the series:
+```
+# SELECT * FROM prom_series.cpu_usage;
+ series_id | labels  | namespace  | node
+-----------+---------+------------+-------
+         4 | {3,4}   | dev        | pinky
+         5 | {5,6}   | production | brain
+```

pkg/pgmodel/migrate.go

@@ -36,6 +36,8 @@ func (t *mySrc) replaceSchemaNames(r io.ReadCloser) (io.ReadCloser, error) {
 	s := buf.String()
 	s = strings.ReplaceAll(s, "SCHEMA_CATALOG", catalogSchema)
 	s = strings.ReplaceAll(s, "SCHEMA_PROM", promSchema)
+	s = strings.ReplaceAll(s, "SCHEMA_SERIES", seriesViewSchema)
+	s = strings.ReplaceAll(s, "SCHEMA_METRIC", metricViewSchema)
 	r = ioutil.NopCloser(strings.NewReader(s))
 	return r, err
 }

pkg/pgmodel/migrate_test.go

@@ -30,7 +30,7 @@ var (
 )
 
 const (
-	expectedVersion = 2
+	expectedVersion = 1
 )
 
 func TestMigrate(t *testing.T) {

pkg/pgmodel/migrations/migration_files_generated.go

@@ -3,6 +3,9 @@
 
 CREATE SCHEMA IF NOT EXISTS SCHEMA_CATALOG; -- catalog tables + internal functions
 CREATE SCHEMA IF NOT EXISTS SCHEMA_PROM; -- data tables + public functions
+CREATE SCHEMA IF NOT EXISTS SCHEMA_SERIES;
+CREATE SCHEMA IF NOT EXISTS SCHEMA_METRIC;
+
 
 CREATE EXTENSION IF NOT EXISTS timescaledb WITH SCHEMA public;
 -----------------------
@@ -105,35 +108,34 @@ CREATE TRIGGER make_metric_table_trigger
 -- Internal functions --
 ------------------------
 
--- Return a table name built from a metric_name and a suffix.
--- The metric name is truncated so that the suffix could fit in full.
-CREATE OR REPLACE FUNCTION SCHEMA_CATALOG.metric_table_name_with_suffix(
-        metric_name text, suffix text)
+-- Return a table name built from a full_name and a suffix.
+-- The full name is truncated so that the suffix could fit in full.
+-- name size will always be 63 chars.
+CREATE OR REPLACE FUNCTION SCHEMA_CATALOG.pg_name_with_suffix(
+        full_name text, suffix text)
     RETURNS name
 AS $func$
-    SELECT (substring(metric_name for 63-(char_length(suffix)+1)) || '_' || suffix)::name
+    SELECT (substring(full_name for 63-(char_length(suffix)+1)) || '_' || suffix)::name
 $func$
 LANGUAGE sql IMMUTABLE PARALLEL SAFE;
 
--- Return a new name for a metric table.
--- This tries to use the metric table in full. But if the
--- metric name doesn't fit, generates a new unique name.
--- Note that this can use up the next val of SCHEMA_CATALOG.metric_name_suffx
--- so it should be called only if a table does not yet exist.
-CREATE OR REPLACE FUNCTION SCHEMA_CATALOG.new_metric_table_name(
-        metric_name_arg text, metric_id int)
+-- Return a new unique name from a name and id.
+-- This tries to use the full_name in full. But if the
+-- full name doesn't fit, generates a new unique name.
+CREATE OR REPLACE FUNCTION SCHEMA_CATALOG.pg_name_unique(
+        full_name_arg text, id int)
     RETURNS name
 AS $func$
     SELECT CASE
-        WHEN char_length(metric_name_arg) < 63 THEN
-            metric_name_arg::name
+        WHEN char_length(full_name_arg) < 63 THEN
+            full_name_arg::name
         ELSE
-            SCHEMA_CATALOG.metric_table_name_with_suffix(
-                metric_name_arg, metric_id::text
+            SCHEMA_CATALOG.pg_name_with_suffix(
+                full_name_arg, id::text
             )
         END
 $func$
-LANGUAGE sql VOLATILE PARALLEL SAFE;
+LANGUAGE sql IMMUTABLE PARALLEL SAFE;
 
 --Creates a new table for a given metric name.
 --This uses up some sequences so should only be called
@@ -149,7 +151,7 @@ LOOP
     INSERT INTO SCHEMA_CATALOG.metric (id, metric_name, table_name)
         SELECT  new_id,
                 metric_name_arg,
-                SCHEMA_CATALOG.new_metric_table_name(metric_name_arg, new_id)
+                SCHEMA_CATALOG.pg_name_unique(metric_name_arg, new_id)
     ON CONFLICT DO NOTHING
     RETURNING SCHEMA_CATALOG.metric.id, SCHEMA_CATALOG.metric.table_name
     INTO id, table_name;
@@ -235,6 +237,10 @@ BEGIN
     IF NOT FOUND THEN
         RAISE 'Could not find a new position';
     END IF;
+
+   PERFORM SCHEMA_CATALOG.create_series_view(metric_name);
+   PERFORM SCHEMA_CATALOG.create_metric_view(metric_name);
+
     RETURN position;
 END
 $func$
@@ -755,3 +761,124 @@ BEGIN
     END LOOP;
 END;
 $$;
+
+CREATE OR REPLACE FUNCTION SCHEMA_PROM.is_stale_marker(value double precision)
+RETURNS BOOLEAN
+AS $func$
+    SELECT float8send(value) = '\x7ff0000000000002'
+$func$
+LANGUAGE SQL IMMUTABLE PARALLEL SAFE;
+
+CREATE OR REPLACE FUNCTION SCHEMA_PROM.is_normal_nan(value double precision)
+RETURNS BOOLEAN
+AS $func$
+    SELECT float8send(value) = '\x7ff8000000000001'
+$func$
+LANGUAGE SQL IMMUTABLE PARALLEL SAFE;
+
+CREATE OR REPLACE FUNCTION SCHEMA_PROM.get_label_value(
+        label_id INT)
+    RETURNS TEXT
+AS $$
+    SELECT
+        value
+    FROM SCHEMA_CATALOG.label
+    WHERE
+        id = label_id
+$$
+LANGUAGE SQL STABLE PARALLEL SAFE;
+
+CREATE OR REPLACE FUNCTION SCHEMA_CATALOG.get_label_key_column_name(label_key text, id int)
+    returns NAME
+AS $func$
+DECLARE
+  is_reserved boolean;
+BEGIN
+  SELECT label_key = ANY(ARRAY['time', 'value', 'series_id', 'labels'])
+  INTO STRICT is_reserved;
+
+  IF is_reserved THEN
+    label_key := label_key || '_label';
+  END IF;
+
+  RETURN SCHEMA_CATALOG.pg_name_unique(label_key, id);
+END
+$func$
+LANGUAGE PLPGSQL;
+
+CREATE OR REPLACE FUNCTION SCHEMA_CATALOG.create_series_view(
+        metric_name text)
+    RETURNS BOOLEAN
+AS $func$
+DECLARE
+   label_value_cols text;
+   view_name text;
+   metric_id int;
+BEGIN
+    SELECT
+        ',' || string_agg(
+            format ('SCHEMA_PROM.get_label_value(series.labels[%s]) AS %I',pos::int, SCHEMA_CATALOG.get_label_key_column_name(key, pos))
+        , ', ' ORDER BY pos)
+    INTO STRICT label_value_cols
+    FROM SCHEMA_CATALOG.label_key_position lkp
+    WHERE lkp.metric = metric_name and key != '__name__';
+
+    SELECT m.table_name, m.id
+    INTO STRICT view_name, metric_id
+    FROM SCHEMA_CATALOG.metric m
+    WHERE m.metric_name = create_series_view.metric_name;
+
+    EXECUTE FORMAT($$
+        CREATE OR REPLACE VIEW SCHEMA_SERIES.%I AS
+        SELECT
+            id AS series_id,
+            labels
+            %s
+        FROM
+            SCHEMA_CATALOG.series
+        WHERE metric_id = %L
+    $$, view_name, label_value_cols, metric_id);
+    RETURN true;
+END
+$func$
+LANGUAGE PLPGSQL;
+
+
+CREATE OR REPLACE FUNCTION SCHEMA_CATALOG.create_metric_view(
+        metric_name text)
+    RETURNS BOOLEAN
+AS $func$
+DECLARE
+   label_value_cols text;
+   table_name text;
+   metric_id int;
+BEGIN
+    SELECT
+        ',' || string_agg(
+            format ('series.labels[%s] AS %I',pos::int, SCHEMA_CATALOG.get_label_key_column_name(key||'_id', pos))
+        , ', ' ORDER BY pos)
+    INTO STRICT label_value_cols
+    FROM SCHEMA_CATALOG.label_key_position lkp
+    WHERE lkp.metric = metric_name and key != '__name__';
+
+    SELECT m.table_name, m.id
+    INTO STRICT table_name, metric_id
+    FROM SCHEMA_CATALOG.metric m
+    WHERE m.metric_name = create_metric_view.metric_name;
+
+    EXECUTE FORMAT($$
+        CREATE OR REPLACE VIEW SCHEMA_METRIC.%1$I AS
+        SELECT
+            data.time as time,
+            data.value as value,
+            data.series_id AS series_id,
+            series.labels
+            %2$s
+        FROM
+            SCHEMA_PROM.%1$I AS data
+            LEFT JOIN SCHEMA_CATALOG.series AS series ON (series.id = data.series_id AND series.metric_id = %3$L)
+    $$, table_name, label_value_cols, metric_id);
+    RETURN true;
+END
+$func$
+LANGUAGE PLPGSQL;

pkg/pgmodel/migrations/sql/1_base_schema.up.sql

@@ -24,8 +24,10 @@ import (
 )
 
 const (
-	promSchema    = "prom"
-	catalogSchema = "_prom_catalog"
+	promSchema       = "prom"
+	seriesViewSchema = "prom_series"
+	metricViewSchema = "prom_metric"
+	catalogSchema    = "_prom_catalog"
 
 	getMetricsTableSQL       = "SELECT table_name FROM " + promSchema + ".get_metric_table_name_if_exists($1)"
 	getCreateMetricsTableSQL = "SELECT table_name FROM " + promSchema + ".get_or_create_metric_table_name($1)"