mirror of
https://github.com/hanzoai/otel-collector.git
synced 2026-11-03 02:20:04 +00:00
zeekay
9a2c38391e
- imports: ClickHouse/ch-go, clickhouse-go/v2, srikanthccv/ClickHouse-go-mock
-> github.com/hanzo-ds/{native,go,mock} (require, clean semver, NO replace)
- identifiers/aliases/type-names/config-keys/type-IDs: clickhouse -> datastore
(receiver dir clickhousesystemtablesreceiver -> datastoresystemtablesreceiver)
- DSN scheme clickhouse:// -> datastore:// (driver treats non-http/https as Native)
- removed the go.opencensus.io version-pin replace (require-pinned instead)
- kept: real external refs (github.com/ClickHouse, clickhouse.com docs) and the
hanzo-ds/mock test API (NewClickHouseNative) pending the mock's own rename
Full ./... build passes. 0 replace directives, 0 signoz.
78 lines
1.9 KiB
Go
78 lines
1.9 KiB
Go
package datastorelogsexporter
|
|
|
|
import (
|
|
"fmt"
|
|
|
|
driver "github.com/hanzo-ds/go/lib/driver"
|
|
"github.com/hanzoai/otel-collector/usage"
|
|
"github.com/google/uuid"
|
|
"github.com/jellydator/ttlcache/v3"
|
|
"go.opentelemetry.io/otel/metric"
|
|
"go.uber.org/zap"
|
|
)
|
|
|
|
type LogExporterOption func(*datastoreLogsExporter)
|
|
|
|
func WithDatastoreClient(client driver.Conn) LogExporterOption {
|
|
return func(e *datastoreLogsExporter) {
|
|
e.db = client
|
|
}
|
|
}
|
|
|
|
func WithLogger(logger *zap.Logger) LogExporterOption {
|
|
return func(e *datastoreLogsExporter) {
|
|
e.logger = logger
|
|
}
|
|
}
|
|
|
|
func WithNewUsageCollector(id uuid.UUID, db driver.Conn) LogExporterOption {
|
|
return func(e *datastoreLogsExporter) {
|
|
e.usageCollector = usage.NewUsageCollector(
|
|
id,
|
|
e.db,
|
|
usage.Options{
|
|
ReportingInterval: usage.DefaultCollectionInterval,
|
|
},
|
|
"o11y_logs",
|
|
UsageExporter,
|
|
e.logger,
|
|
)
|
|
// TODO: handle error
|
|
_ = e.usageCollector.Start()
|
|
e.id = id
|
|
}
|
|
}
|
|
|
|
func WithMeter(meter metric.Meter) LogExporterOption {
|
|
return func(e *datastoreLogsExporter) {
|
|
durationHistogram, err := meter.Float64Histogram(
|
|
"exporter_db_write_latency",
|
|
metric.WithDescription("Time taken to write data to Datastore"),
|
|
metric.WithUnit("ms"),
|
|
metric.WithExplicitBucketBoundaries(250, 500, 750, 1000, 2000, 2500, 3000, 4000, 5000, 6000, 8000, 10000, 15000, 25000, 30000),
|
|
)
|
|
if err != nil {
|
|
panic(fmt.Errorf("error creating duration histogram: %w", err))
|
|
}
|
|
e.durationHistogram = durationHistogram
|
|
}
|
|
}
|
|
|
|
func WithKeysCache(keysCache *ttlcache.Cache[string, struct{}]) LogExporterOption {
|
|
return func(e *datastoreLogsExporter) {
|
|
e.keysCache = keysCache
|
|
}
|
|
}
|
|
|
|
func WithRFCache(rfCache *ttlcache.Cache[string, struct{}]) LogExporterOption {
|
|
return func(e *datastoreLogsExporter) {
|
|
e.rfCache = rfCache
|
|
}
|
|
}
|
|
|
|
func WithConcurrency(concurrency int) LogExporterOption {
|
|
return func(e *datastoreLogsExporter) {
|
|
e.limiter = make(chan struct{}, concurrency)
|
|
}
|
|
}
|