Skip to content

Commit 3e04bde

Browse files
authored
Merge pull request #483 from metrico/bun_fix
fix: throttler -> clickhouse.js cross dependency
2 parents 787a9f9 + 10b8f62 commit 3e04bde

File tree

2 files changed

+43
-28
lines changed

2 files changed

+43
-28
lines changed

lib/db/clickhouse.js

+20-18
Original file line numberDiff line numberDiff line change
@@ -89,25 +89,27 @@ if (isMainThread && !bun()) {
8989
})
9090
} else if (isMainThread && !first) {
9191
first = true
92-
const _throttler = require('./throttler')
93-
throttler = {
94-
on: _throttler.on,
95-
postMessage: _throttler.postMessage,
96-
removeAllListeners: _throttler.removeAllListeners,
97-
terminate: _throttler.terminate
98-
}
99-
_throttler.init()
100-
throttler.on('message', (msg) => {
101-
switch (msg.status) {
102-
case 'ok':
103-
resolvers[msg.id]()
104-
break
105-
case 'err':
106-
rejectors[msg.id](new Error('Database push error'))
107-
break
92+
setTimeout(() => {
93+
const _throttler = require('./throttler')
94+
throttler = {
95+
on: _throttler.on,
96+
postMessage: _throttler.postMessage,
97+
removeAllListeners: _throttler.removeAllListeners,
98+
terminate: _throttler.terminate
10899
}
109-
delete resolvers[msg.id]
110-
delete rejectors[msg.id]
100+
_throttler.init()
101+
throttler.on('message', (msg) => {
102+
switch (msg.status) {
103+
case 'ok':
104+
resolvers[msg.id]()
105+
break
106+
case 'err':
107+
rejectors[msg.id](new Error('Database push error'))
108+
break
109+
}
110+
delete resolvers[msg.id]
111+
delete rejectors[msg.id]
112+
})
111113
})
112114
}
113115
// timeSeriesv2Throttler.start();

lib/db/throttler.js

+23-10
Original file line numberDiff line numberDiff line change
@@ -1,14 +1,20 @@
11
const { isMainThread, parentPort } = require('worker_threads')
2-
const axios = require('axios')
3-
const { getClickhouseUrl, samplesTableName, rawRequest } = require('./clickhouse')
42
const clickhouseOptions = require('./clickhouse_options').databaseOptions
53
const logger = require('../logger')
64
const { DATABASE_NAME } = require('../utils')
75
const clusterName = require('../../common').clusterName
86
const dist = clusterName ? '_dist' : ''
97
const { EventEmitter } = require('events')
108

9+
// variables to be initialized in the init() function due to the './clickhouse.js' cross-dependency & bun
10+
let samplesThrottler
11+
let timeSeriesThrottler
12+
let tracesThottler
13+
let samplesTableName
14+
let rawRequest
15+
1116
const axiosError = async (err) => {
17+
console.log('axiosError', err)
1218
try {
1319
const resp = err.response
1420
if (resp) {
@@ -24,6 +30,7 @@ const axiosError = async (err) => {
2430
(err.responseData ? ' Response data: ' + err.responseData : ''))
2531
}
2632
} catch (e) {
33+
console.log(e)
2734
return err
2835
}
2936
}
@@ -63,14 +70,6 @@ class TimeoutThrottler {
6370
}
6471
}
6572

66-
const samplesThrottler = new TimeoutThrottler(
67-
`INSERT INTO ${clickhouseOptions.queryOptions.database}.${samplesTableName}${dist}(fingerprint, timestamp_ns, value, string, type) FORMAT JSONEachRow`)
68-
const timeSeriesThrottler = new TimeoutThrottler(
69-
`INSERT INTO ${clickhouseOptions.queryOptions.database}.time_series${dist}(date, fingerprint, labels, name, type) FORMAT JSONEachRow`)
70-
const tracesThottler = new TimeoutThrottler(
71-
`INSERT INTO ${clickhouseOptions.queryOptions.database}.traces_input
72-
(trace_id, span_id, parent_id, name, timestamp_ns, duration_ns, service_name, payload_type, payload, tags)
73-
FORMAT JSONEachRow`)
7473

7574
const emitter = new EventEmitter()
7675
let on = true
@@ -111,6 +110,20 @@ const postMessage = message => {
111110
}
112111

113112
const init = () => {
113+
[samplesTableName, rawRequest] = [
114+
require('./clickhouse').samplesTableName,
115+
require('./clickhouse').rawRequest
116+
]
117+
118+
samplesThrottler = new TimeoutThrottler(
119+
`INSERT INTO ${clickhouseOptions.queryOptions.database}.${samplesTableName}${dist}(fingerprint, timestamp_ns, value, string, type) FORMAT JSONEachRow`)
120+
timeSeriesThrottler = new TimeoutThrottler(
121+
`INSERT INTO ${clickhouseOptions.queryOptions.database}.time_series${dist}(date, fingerprint, labels, name, type) FORMAT JSONEachRow`)
122+
tracesThottler = new TimeoutThrottler(
123+
`INSERT INTO ${clickhouseOptions.queryOptions.database}.traces_input
124+
(trace_id, span_id, parent_id, name, timestamp_ns, duration_ns, service_name, payload_type, payload, tags)
125+
FORMAT JSONEachRow`)
126+
114127
setTimeout(async () => {
115128
// eslint-disable-next-line no-unmodified-loop-condition
116129
while (on) {

0 commit comments

Comments
 (0)