mirror of
https://github.com/hanzoai/otel-collector.git
synced 2026-11-03 02:20:04 +00:00
Raj Kamal SinghandGitHub
5e4c9d8380
* feat: get signozclickhousesystemlogsreceiver started * chore: put together receiver scaffolding for clickhousesystemtablesreceiver * chore: add clickhousesystemstablereceiver to list of available otel collector receivers * feat: add impl for unixTsNowAtClickhouse * feat: get query_log querying logic started * feat: flesh out query_log scraping loop * feat: finish up with e2e tracer for query log receiver * chore: add example config for clickhousesystemtablesreceiver * chore: get receiver_test started * chore: refactor clickhouse calls behind an interface for easier mocking in tests * chore: flesh out receiver happy path test * chore: wrap up receiver lifecycle test * chore: add test for receiver start and shutdown * chore: add config validation tests * chore: some cleanup * chore: hook log emitted telemetry on receiver * chore: some more cleanup * chore: some more cleanup * chore: some more cleanup * chore: some more cleanup * chore: some more cleanup * chore: some more cleanup * chore: some more cleanup * chore: some more cleanup * chore: address some review feedback * chore: address some more review feedback * feat: add support for specifying a cluster name when scraping clustered deployments * chore: some more cleanup
57 lines
1.5 KiB
Go
57 lines
1.5 KiB
Go
package clickhousesystemtablesreceiver
|
|
|
|
import (
|
|
"context"
|
|
"fmt"
|
|
|
|
"github.com/ClickHouse/clickhouse-go/v2/lib/driver"
|
|
)
|
|
|
|
// Used by the receiver for working with clickhouse.
|
|
// Helps mock things for tests
|
|
type clickhouseQuerier interface {
|
|
// Scrape query_log table for rows with minEventTs <= event_time < maxEventTs
|
|
scrapeQueryLog(
|
|
ctx context.Context, minEventTs uint32, maxEventTs uint32,
|
|
) ([]QueryLog, error)
|
|
|
|
// Get unix epoch time now at the server (seconds)
|
|
unixTsNow(context.Context) (uint32, error)
|
|
}
|
|
|
|
type querrierImpl struct {
|
|
db driver.Conn
|
|
|
|
// Cluster name to use for scraping query log from clustered deployments.
|
|
// If ClusterName is specified, the scrape will target `clusterAllReplicas(clusterName, system.query_log)`
|
|
clusterName string
|
|
}
|
|
|
|
var _ clickhouseQuerier = (*querrierImpl)(nil)
|
|
|
|
func newClickhouseQuerrier(db driver.Conn, clusterName string) *querrierImpl {
|
|
return &querrierImpl{
|
|
db: db,
|
|
clusterName: clusterName,
|
|
}
|
|
}
|
|
|
|
func (q *querrierImpl) scrapeQueryLog(
|
|
ctx context.Context, minTs uint32, maxTs uint32,
|
|
) ([]QueryLog, error) {
|
|
return scrapeQueryLogTable(ctx, q.db, q.clusterName, minTs, maxTs)
|
|
}
|
|
|
|
func (q *querrierImpl) unixTsNow(ctx context.Context) (
|
|
uint32, error,
|
|
) {
|
|
var serverTsNow uint32
|
|
row := q.db.QueryRow(ctx, `select toUnixTimestamp(now())`)
|
|
if err := row.Scan(&serverTsNow); err != nil {
|
|
return 0, fmt.Errorf(
|
|
"couldn't query current timestamp at clickhouse server: %w", err,
|
|
)
|
|
}
|
|
return serverTsNow, nil
|
|
}
|