Introduce telemetry and metrics patterns

[shah]

Integrate telemetry and metrics into pattern/telemetry.ts so that exceptions and telemetry are all available to any information model.

In PgDCP we had a special exception hub with specific satellites. However, we should switch from specific exception hub to generalized telemetry information models that can be reused across services and applications (see this PgDCP idea).

Potential telemetry objects:

CREATE TABLE span (
  span_id UUID PRIMARY KEY,
  trace_id UUID NOT NULL,
  parent_span_id UUID,
  name TEXT NOT NULL,
  start_time TIMESTAMP WITH TIME ZONE NOT NULL,
  end_time TIMESTAMP WITH TIME ZONE,
  status INT NOT NULL DEFAULT 0,
  kind INT NOT NULL DEFAULT 0,
  span_kind TEXT,
  trace_state TEXT,
  resource TEXT,
  FOREIGN KEY (parent_span_id) REFERENCES span(span_id)
);

CREATE INDEX idx_trace_id ON span(trace_id);

CREATE TABLE attribute (
  id SERIAL PRIMARY KEY,
  span_id UUID NOT NULL,
  key TEXT NOT NULL,
  value TEXT NOT NULL,
  FOREIGN KEY (span_id) REFERENCES span(span_id)
);

CREATE UNIQUE INDEX idx_attributes ON attribute(span_id, key);

CREATE TABLE event (
  id SERIAL PRIMARY KEY,
  span_id UUID NOT NULL,
  name TEXT NOT NULL,
  timestamp TIMESTAMP WITH TIME ZONE NOT NULL,
  FOREIGN KEY (span_id) REFERENCES span(span_id)
);

CREATE TABLE link (
  id SERIAL PRIMARY KEY,
  span_id UUID NOT NULL,
  linked_span_id UUID NOT NULL,
  FOREIGN KEY (span_id) REFERENCES span(span_id),
  FOREIGN KEY (linked_span_id) REFERENCES span(span_id)
);

CREATE TABLE baggage (
  id SERIAL PRIMARY KEY,
  span_id UUID NOT NULL,
  key TEXT NOT NULL,
  value TEXT NOT NULL,
  FOREIGN KEY (span_id) REFERENCES span(span_id)
);

-- Metrics table storing measurements related to spans.
CREATE TABLE metric (
  id SERIAL PRIMARY KEY,
  span_id UUID NOT NULL,
  name TEXT NOT NULL,
  value DOUBLE PRECISION NOT NULL,
  timestamp TIMESTAMP WITH TIME ZONE NOT NULL,
  FOREIGN KEY (span_id) REFERENCES span(span_id)
);

-- This stored procedure inserts the provided telemetry data into the appropriate tables. It first inserts the span into the span table and gets the span_id of the inserted row. It then uses this span_id when inserting the associated attributes, events, links, baggage items, and metrics. If arrays for these fields in the composite type are NULL or empty, it won't insert anything into the corresponding tables.
CREATE OR REPLACE PROCEDURE insert_telemetry_data(_telemetry telemetry_type)
LANGUAGE plpgsql
AS $$
DECLARE
  inserted_span_id UUID;
BEGIN
  -- Insert into span table
  INSERT INTO span(span_id, trace_id, parent_span_id, name, start_time, end_time, status, kind, span_kind, trace_state, resource)
  VALUES (_telemetry.span_id, _telemetry.trace_id, _telemetry.parent_span_id, _telemetry.name, _telemetry.start_time, _telemetry.end_time, _telemetry.status, _telemetry.kind, _telemetry.span_kind, _telemetry.trace_state, _telemetry.resource)
  RETURNING span_id INTO inserted_span_id;

  -- Insert into attribute table
  FOREACH item IN ARRAY _telemetry.attributes
  LOOP
    INSERT INTO attribute(span_id, key, value)
    VALUES (inserted_span_id, item.key, item.value);
  END LOOP;

  -- Insert into event table
  FOREACH item IN ARRAY _telemetry.events
  LOOP
    INSERT INTO event(span_id, name, timestamp)
    VALUES (inserted_span_id, item.name, item.timestamp);
  END LOOP;

  -- Insert into link table
  FOREACH item IN ARRAY _telemetry.links
  LOOP
    INSERT INTO link(span_id, linked_span_id)
    VALUES (inserted_span_id, item.linked_span_id);
  END LOOP;

  -- Insert into baggage table
  FOREACH item IN ARRAY _telemetry.baggage_items
  LOOP
    INSERT INTO baggage(span_id, key, value)
    VALUES (inserted_span_id, item.key, item.value);
  END LOOP;

  -- Insert into metric table
  FOREACH item IN ARRAY _telemetry.metrics
  LOOP
    INSERT INTO metric(span_id, name, value, timestamp)
    VALUES (inserted_span_id, item.name, item.value, item.timestamp);
  END LOOP;
END;
$$;

-- sample call..inserts the provided telemetry data into the appropriate tables. It first inserts the span into the span table and gets the span_id of the inserted row. It then uses this span_id when inserting the associated attributes, events, links, baggage items, and metrics. If arrays for these fields in the composite type are NULL or empty, it won't insert anything into the corresponding tables.

CALL insert_telemetry_data(row(
  'span_id_value', 
  'trace_id_value', 
  'parent_span_id_value', 
  'name_value', 
  'start_time_value', 
  'end_time_value', 
  1, 
  0, 
  'span_kind_value', 
  'trace_state_value', 
  'resource_value', 
  ARRAY[('key1', 'value1'), ('key2', 'value2')]::attribute[], 
  ARRAY[('name1', 'timestamp1'), ('name2', 'timestamp2')]::event[], 
  ARRAY[('linked_span_id1'), ('linked_span_id2')]::link[], 
  ARRAY[('key1', 'value1'), ('key2', 'value2')]::baggage[], 
  ARRAY[('name1', 1.0, 'timestamp1'), ('name2', 2.0, 'timestamp2')]::metric[]
)::telemetry_type);

-- In OpenTelemetry, errors and exceptions are typically represented as events within a span. Instead of hub_exception we can create an event with a specific name indicating that it represents an exception, and use attributes to store detailed information about the exception.
CALL insert_telemetry_data(row(
  'span_id_value', 
  'trace_id_value', 
  'parent_span_id_value', 
  'process_request',  -- name of the operation
  'start_time_value', 
  'end_time_value', 
  1,  -- status indicating there was an error
  0, 
  'INTERNAL',  -- span kind
  'trace_state_value', 
  'resource_value', 
  ARRAY[('key1', 'value1'), ('key2', 'value2')]::attribute[], 
  ARRAY[('exception', 'timestamp_error', 'error message', 'error type', 'stack trace')]::event[], 
  NULL::link[], 
  ARRAY[('key1', 'value1'), ('key2', 'value2')]::baggage[], 
  ARRAY[('name1', 1.0, 'timestamp1'), ('name2', 2.0, 'timestamp2')]::metric[]
)::telemetry_type);

-- Prometheus-style metrics ... could be linked to hubs, spans, etc.
-- The metric table contains all metric definitions. Each metric can have multiple labels, which are defined in the metric_label table. The metric_value table stores the timestamped values for counters and gauges. Histograms and summaries are more complex: they have buckets and quantiles, respectively. These are stored in the histogram_bucket and summary_quantile tables.
CREATE TABLE metric (
    id SERIAL PRIMARY KEY,
    name VARCHAR(255) NOT NULL,
    description TEXT,
    type VARCHAR(50) NOT NULL,
    unit VARCHAR(50),
    UNIQUE(name)
);

CREATE TABLE metric_label (
    id SERIAL PRIMARY KEY,
    metric_id INT NOT NULL REFERENCES metric(id),
    label_key VARCHAR(255) NOT NULL,
    label_value VARCHAR(255) NOT NULL,
    UNIQUE(metric_id, label_key, label_value)
);

CREATE TABLE metric_value (
    id SERIAL PRIMARY KEY,
    metric_label_id INT NOT NULL REFERENCES metric_label(id),
    timestamp TIMESTAMP NOT NULL,
    value DOUBLE PRECISION NOT NULL
);

CREATE TABLE histogram_bucket (
    id SERIAL PRIMARY KEY,
    metric_label_id INT NOT NULL REFERENCES metric_label(id),
    upper_bound DOUBLE PRECISION NOT NULL,
    bucket_value DOUBLE PRECISION NOT NULL,
    UNIQUE(metric_label_id, upper_bound)
);

CREATE TABLE summary_quantile (
    id SERIAL PRIMARY KEY,
    metric_label_id INT NOT NULL REFERENCES metric_label(id),
    quantile DOUBLE PRECISION NOT NULL,
    quantile_value DOUBLE PRECISION NOT NULL,
    UNIQUE(metric_label_id, quantile)
);

CREATE OR REPLACE PROCEDURE insert_metric_value(metric_name_param TEXT, label_key_param TEXT, label_value_param TEXT, timestamp_param TIMESTAMP, value_param DOUBLE PRECISION)
LANGUAGE plpgsql
AS $$
DECLARE 
  partition_name TEXT;
  partition_start TIMESTAMP;
  partition_end TIMESTAMP;
  metric_id_param INT;
  metric_label_id_param INT;
BEGIN
  partition_name := 'metric_value_' || to_char(timestamp_param, 'YYYY_MM');
  partition_start := date_trunc('month', timestamp_param);
  partition_end := partition_start + INTERVAL '1 month';

  SELECT id INTO metric_id_param FROM metric WHERE name = metric_name_param;

  IF metric_id_param IS NULL THEN
    RAISE EXCEPTION 'Metric % does not exist.', metric_name_param;
  END IF;

  SELECT id INTO metric_label_id_param FROM metric_label WHERE metric_id = metric_id_param AND label_key = label_key_param AND label_value = label_value_param;

  IF metric_label_id_param IS NULL THEN
    INSERT INTO metric_label (metric_id, label_key, label_value) VALUES (metric_id_param, label_key_param, label_value_param)
    RETURNING id INTO metric_label_id_param;
  END IF;

  BEGIN
    EXECUTE format('CREATE TABLE IF NOT EXISTS %I PARTITION OF metric_value FOR VALUES FROM (%L) TO (%L);', partition_name, partition_start, partition_end);
  EXCEPTION WHEN duplicate_table THEN
    -- Do nothing, the partition already exists
  END;

  EXECUTE format('INSERT INTO %I (metric_label_id, timestamp, value) VALUES ($1, $2, $3);', partition_name) USING metric_label_id_param, timestamp_param, value_param;
END;
$$;

-- sample call
CALL insert_metric_value('http_requests_total', 'method', 'GET', '2023-05-27 10:00:00', 5000);

[shah]

@geovlazar please assign this to either Alan or Joby (or any other person who's now learned how to write SQLa tables). Discuss with me if you have questions.