#fix: drop alerts for single server; ws live

metrico · Oct 1, 2023 · 5920f66 · 5920f66
1 parent 6e466d6
commit 5920f66
Show file tree

Hide file tree

Showing 3 changed files with 15 additions and 10 deletions.
diff --git a/lib/db/clickhouse_alerting.js b/lib/db/clickhouse_alerting.js
@@ -156,7 +156,8 @@ module.exports.deleteAlertRule = async (ns, group, rule) => {
     `INSERT INTO ${DATABASE_NAME()}.settings (fingerprint, type, name, value, inserted_at) FORMAT JSONEachRow\n` +
     JSON.stringify({ fingerprint: fp, type: 'alert_rule', name: '', value: '', inserted_at: Date.now() })
   )
-  await axios.post(getClickhouseUrl() + '/?allow_nondeterministic_mutations=1&mutations_execute_nondeterministic_on_initiator=1',
+  const settings = clusterName ? '/?allow_nondeterministic_mutations=1&mutations_execute_nondeterministic_on_initiator=1' : ''
+  await axios.post(getClickhouseUrl() + settings,
     `ALTER TABLE ${DATABASE_NAME()}.settings ${onCluster} DELETE WHERE fingerprint=${fp} AND inserted_at <= now64(9, 'UTC')`
   )
 }
@@ -172,9 +173,9 @@ module.exports.deleteGroup = async (ns, group) => {
     `INSERT INTO ${DATABASE_NAME()}.settings${dist} (fingerprint, type, name, value, inserted_at) FORMAT JSONEachRow\n` +
     JSON.stringify({ fingerprint: fp, type: 'alert_group', name: '', value: '', inserted_at: Date.now() })
   )
-  await axios.post(getClickhouseUrl() + '/?allow_nondeterministic_mutations=1&mutations_execute_nondeterministic_on_initiator=1',
-    `ALTER TABLE ${DATABASE_NAME()}.settings ${onCluster} DELETE WHERE fingerprint=${fp} AND inserted_at <= now64(9, 'UTC')
-SETTINGS allow_nondeterministic_mutations=1`
+  const settings = clusterName ? '/?allow_nondeterministic_mutations=1&mutations_execute_nondeterministic_on_initiator=1' : ''
+  await axios.post(getClickhouseUrl() + settings,
+    `ALTER TABLE ${DATABASE_NAME()}.settings ${onCluster} DELETE WHERE fingerprint=${fp} AND inserted_at <= now64(9, 'UTC')`
   )
 }
 

diff --git a/lib/db/watcher.js b/lib/db/watcher.js
@@ -70,13 +70,13 @@ class Watcher extends EventEmitter {
   }
 
   async initQueryCBPoll () {
-    this.from = (Date.now() - 300000) * 1000000
+    this.from = (Date.now() - 300000)
     while (this.working) {
-      this.to = (Date.now() - 5000) * 1000000
+      this.to = (Date.now() - 1000)
       this.query = transpiler.transpile({
         ...this.request,
-        start: this.from,
-        end: this.to
+        start: this.from * 1000000,
+        end: this.to * 1000000
       })
       this.query.step = this.step
       const resp = await queryFingerprintsScan(this.query)
@@ -88,8 +88,12 @@ class Watcher extends EventEmitter {
         resp.once('close', resolve)
         resp.once('error', reject)
       })
-      this.from = this.to
       this.resp = JSON.parse(this.resp).data.result
+      for (const stream of this.resp) {
+        for (const v of stream.values) {
+          this.from = Math.max(this.from, (parseFloat(v[0]) + 1000000) / 1000000)
+        }
+      }
       this.flush()
       this.resp = ''
       await new Promise((resolve) => setTimeout(resolve, 1000))

diff --git a/parser/transpiler.js b/parser/transpiler.js
@@ -293,7 +293,7 @@ module.exports.transpileTail = (request) => {
   }
   query = module.exports.transpileLogStreamSelector(expression.rootToken, query)
   setQueryParam(query, sharedParamNames.timeSeriesTable, `${DATABASE_NAME()}.time_series`)
-  setQueryParam(query, sharedParamNames.samplesTable, `${DATABASE_NAME()}.${samplesTableName}`)
+  setQueryParam(query, sharedParamNames.samplesTable, `${DATABASE_NAME()}.${samplesTableName}${dist}`)
   setQueryParam(query, sharedParamNames.from, new Sql.Raw('(toUnixTimestamp(now()) - 5) * 1000000000'))
   query.order_expressions = []
   query.orderBy(['timestamp_ns', 'asc'])