Merge pull request #438 from metrico/pyroscope-2

Pyroscope api support

lib/db/clickhouse.js

@@ -1360,6 +1360,7 @@ const rawRequest = async (query, data, database, config) => {
     return await axios.post(url, data || query, config)
   } catch (e) {
     logger.error('rawRequest error: ' + query)
+    e.response?.data && logger.error(e.response.data.toString())
     throw e
   }
 }

lib/db/maintain/index.js

@@ -2,6 +2,7 @@ const hb = require('handlebars')
 const client = require('../clickhouse')
 const logger = require('../../logger')
 const { samplesOrderingRule, clusterName } = require('../../../common')
+const scripts = require('./scripts')
 const getEnv = () => {
   return {
     CLICKHOUSE_DB: 'cloki',
@@ -17,16 +18,17 @@ const getEnv = () => {
  * @returns {Promise<void>}
  */
 module.exports.upgrade = async (db) => {
-  const scripts = require('./scripts')
   await upgradeSingle(db.name, 1, scripts.overall, db.storage_policy)
   await upgradeSingle(db.name, 2, scripts.traces, db.storage_policy)
+  await upgradeSingle(db.name, 5, scripts.profiles, db.storage_policy)
   if (db.storage_policy) {
     await client.addSetting('rotate', 'v3_storage_policy', db.storage_policy, db.name)
     await client.addSetting('rotate', 'v1_traces_storage_policy', db.storage_policy, db.name)
   }
   if (clusterName) {
     await upgradeSingle(db.name, 3, scripts.overall_dist, db.storage_policy)
     await upgradeSingle(db.name, 4, scripts.traces_dist, db.storage_policy)
+    await upgradeSingle(db.name, 6, scripts.profiles_dist, db.storage_policy)
   }
 }
 

lib/db/maintain/scripts.js

@@ -230,3 +230,154 @@ module.exports.traces_dist = [
   duration Int64
 ) ENGINE = Distributed('{{CLUSTER}}','{{DB}}', 'tempo_traces_attrs_gin', sipHash64(oid, trace_id));`
 ]
+
+module.exports.profiles = [
+  `CREATE TABLE IF NOT EXISTS profiles_input {{{OnCluster}}} (
+    timestamp_ns UInt64,
+    type LowCardinality(String),
+    service_name LowCardinality(String),
+    sample_types_units Array(Tuple(String, String)),
+    period_type LowCardinality(String),
+    period_unit LowCardinality(String),
+    tags Array(Tuple(String, String)),
+    duration_ns UInt64,
+    payload_type LowCardinality(String),
+    payload String,
+    values_agg Array(Tuple(String, Int64, Int32)) CODEC(ZSTD(1))
+  ) Engine=Null`,
+
+  `CREATE TABLE IF NOT EXISTS profiles {{{OnCluster}}} (
+    timestamp_ns UInt64 CODEC(DoubleDelta, ZSTD(1)),
+    fingerprint UInt64 CODEC(DoubleDelta, ZSTD(1)),
+    type_id LowCardinality(String) CODEC(ZSTD(1)),
+    sample_types_units Array(Tuple(String, String)) CODEC(ZSTD(1)),
+    service_name LowCardinality(String) CODEC(ZSTD(1)),
+    duration_ns UInt64 CODEC(DoubleDelta, ZSTD(1)),
+    payload_type LowCardinality(String) CODEC(ZSTD(1)),
+    payload String CODEC(ZSTD(1)),
+    values_agg Array(Tuple(String, Int64, Int32)) CODEC(ZSTD(1)) 
+  ) Engine {{MergeTree}}() 
+  ORDER BY (type_id, service_name, timestamp_ns)
+  PARTITION BY toDate(FROM_UNIXTIME(intDiv(timestamp_ns, 1000000000)))`,
+
+  `CREATE MATERIALIZED VIEW IF NOT EXISTS profiles_mv {{{OnCluster}}} TO profiles AS
+    SELECT 
+      timestamp_ns, 
+      cityHash64(arraySort(arrayConcat(
+        profiles_input.tags, [
+          ('__type__', concatWithSeparator(':', type, period_type, period_unit) as _type_id), 
+          ('__sample_types_units__', arrayStringConcat(arrayMap(x -> x.1 || ':' || x.2, arraySort(sample_types_units)), ';')),
+          ('service_name', service_name)
+      ])) as _tags) as fingerprint,
+      _type_id as type_id,
+      sample_types_units,
+      service_name,
+      duration_ns,
+      payload_type,
+      payload,
+      values_agg
+    FROM profiles_input`,
+
+  `CREATE TABLE IF NOT EXISTS profiles_series {{{OnCluster}}} (
+    date Date CODEC(ZSTD(1)),
+    type_id LowCardinality(String) CODEC(ZSTD(1)),
+    sample_types_units Array(Tuple(String, String)) CODEC(ZSTD(1)),
+    service_name LowCardinality(String) CODEC(ZSTD(1)),
+    fingerprint UInt64 CODEC(DoubleDelta, ZSTD(1)),    
+    tags Array(Tuple(String, String)) CODEC(ZSTD(1)),
+  ) Engine {{ReplacingMergeTree}}() 
+  ORDER BY (date, type_id, fingerprint)
+  PARTITION BY date`,
+
+  `CREATE MATERIALIZED VIEW IF NOT EXISTS profiles_series_mv {{{OnCluster}}} TO profiles_series AS
+    SELECT 
+      toDate(intDiv(timestamp_ns, 1000000000)) as date,
+      concatWithSeparator(':', type, period_type, period_unit) as type_id,
+      sample_types_units,
+      service_name,
+      cityHash64(arraySort(arrayConcat(
+        profiles_input.tags, [
+          ('__type__', type_id), 
+          ('__sample_types_units__', arrayStringConcat(arrayMap(x -> x.1 || ':' || x.2, arraySort(sample_types_units)), ';')),
+          ('service_name', service_name)
+      ])) as _tags) as fingerprint,
+      arrayConcat(profiles_input.tags, [('service_name', service_name)]) as tags
+    FROM profiles_input`,
+
+  `CREATE TABLE IF NOT EXISTS profiles_series_gin {{{OnCluster}}} (
+    date Date CODEC(ZSTD(1)),
+    key String CODEC(ZSTD(1)),
+    val String CODEC(ZSTD(1)),
+    type_id LowCardinality(String) CODEC(ZSTD(1)),
+    sample_types_units Array(Tuple(String, String)),
+    service_name LowCardinality(String) CODEC(ZSTD(1)),
+    fingerprint UInt64 CODEC(DoubleDelta, ZSTD(1)),
+  ) Engine {{ReplacingMergeTree}}()
+  ORDER BY (date, key, val, type_id, fingerprint)
+  PARTITION BY date`,
+
+  `CREATE MATERIALIZED VIEW IF NOT EXISTS profiles_series_gin_mv {{{OnCluster}}} TO profiles_series_gin AS
+    SELECT 
+      date,
+      kv.1 as key,
+      kv.2 as val,
+      type_id,
+      sample_types_units,
+      service_name,
+      fingerprint,
+    FROM profiles_series ARRAY JOIN tags as kv`,
+
+  `CREATE TABLE IF NOT EXISTS profiles_series_keys {{{OnCluster}}} (
+    date Date,
+    key String,
+    val String,
+    val_id UInt64,
+  ) Engine {{ReplacingMergeTree}}()
+  ORDER BY (date, key, val_id)
+  PARTITION BY date`,
+
+  `CREATE MATERIALIZED VIEW IF NOT EXISTS profiles_series_keys_mv {{{OnCluster}}} TO profiles_series_keys AS
+    SELECT 
+      date,
+      key,
+      val,
+      cityHash64(val) % 50000 as val_id
+    FROM profiles_series_gin`
+]
+
+module.exports.profiles_dist = [
+  `CREATE TABLE IF NOT EXISTS profiles_dist {{{OnCluster}}} (
+    timestamp_ns UInt64,
+    fingerprint UInt64,
+    type_id LowCardinality(String),
+    service_name LowCardinality(String),
+    duration_ns UInt64,
+    payload_type LowCardinality(String),
+    payload String,
+    values_agg Array(Tuple(String, Int64, Int32))
+  ) ENGINE = Distributed('{{CLUSTER}}','{{DB}}','profiles', fingerprint);`,
+
+  `CREATE TABLE IF NOT EXISTS profiles_series_dist {{{OnCluster}}} (
+    date Date,
+    type_id LowCardinality(String),
+    service_name LowCardinality(String),
+    fingerprint UInt64 CODEC(DoubleDelta, ZSTD(1)),
+    tags Array(Tuple(String, String)) CODEC(ZSTD(1)),
+  ) ENGINE = Distributed('{{CLUSTER}}','{{DB}}','profiles_series',fingerprint);`,
+
+  `CREATE TABLE IF NOT EXISTS profiles_series_gin_dist {{{OnCluster}}} (
+    date Date,
+    key String,
+    val String,
+    type_id LowCardinality(String),
+    service_name LowCardinality(String),
+    fingerprint UInt64 CODEC(DoubleDelta, ZSTD(1)),
+  ) ENGINE = Distributed('{{CLUSTER}}','{{DB}}','profiles_series_gin',fingerprint);`,
+
+  `CREATE TABLE IF NOT EXISTS profiles_series_keys_dist {{{OnCluster}}} (
+    date Date,
+    key String,
+    val String,
+    val_id UInt64,
+  ) ENGINE = Distributed('{{CLUSTER}}','{{DB}}','profiles_series_keys', rand());`
+]

package.json

@@ -76,7 +76,9 @@
     "xxhash-wasm": "^0.4.2",
     "yaml": "^1.10.2",
     "@stricjs/utils": "^1.6.1",
-    "basic-auth": "^2.0.1"
+    "basic-auth": "^2.0.1",
+    "google-protobuf": "^3.21.2",
+    "@grpc/grpc-js": "^1.9.14"
   },
   "devDependencies": {
     "@elastic/elasticsearch": "^8.5.0",

pyroscope/google/v1/profile_grpc_pb.js

@@ -0,0 +1 @@
+// GENERATED CODE -- NO SERVICES IN PROTO