chstorage

package
v0.48.0 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Jul 20, 2026 License: Apache-2.0 Imports: 87 Imported by: 0

Documentation

Overview

Package chstorage provides Clickhouse-based storage.

Index

Constants

View Source 
const (
	// MigrationUnknown is an unknown migration status.
	MigrationUnknown = MigrationStatus(iota)
	// MigrationDelete indicates that table should be removed in the new version.
	MigrationDelete
	// MigrationCreate indicates that table should be created in the new version.
	MigrationCreate
	// MigrationUpgrade indicates that table should be changed in the new version.
	MigrationUpgrade
	// MigrationOK indicates that table does not require any changes.
	MigrationOK
)

Variables

View Source 
var ErrLogsResultTooLarge = errors.New("sample query result is too large")

ErrLogsResultTooLarge means that ClickHouse aborted a sample query because its result exceeded the configured byte limit.

View Source 
var ErrLogsTooManySamples = errors.New("too many log lines requested for sampling")

ErrLogsTooManySamples means that a LogQL sample query (e.g. count_over_time, rate, bytes_over_time) matched more log rows than allowed.

View Source 
var ErrMetricsTooManySeries = errors.New("too many timeseries requested")

ErrMetricsTooManySeries whether if query requested more timeseries than allowed.

Functions

func DecodeUnicodeLabel 

func DecodeUnicodeLabel(v string) string

DecodeUnicodeLabel tries to decode U__k8s_2e_node_2e_name into k8s.node.name. It decodes any hex-encoded character in the format _XX_ where XX is a two-digit hex value.

Types

type Attributes 

type Attributes struct {
	Name  string
	Value proto.ColumnOf[otelstorage.Attrs]
}

func NewAttributes 

func NewAttributes(name string, opts ...AttributesOption) *Attributes

NewAttributes constructs a new Attributes storage representation.

func (*Attributes) Append 

func (a *Attributes) Append(kv otelstorage.Attrs)

Append adds a new map of attributes.

func (*Attributes) Columns 

func (a *Attributes) Columns() Columns

Columns returns a slice of Columns for this attribute set.

func (*Attributes) DDL 

func (a *Attributes) DDL(table *ddl.Table)

DDL applies the schema changes to the table.

func (*Attributes) Row 

func (a *Attributes) Row(idx int) otelstorage.Attrs

Row returns a new map of attributes for a given row.

type AttributesOption 

type AttributesOption func(*attributesOptions)

func WithLowCardinality 

func WithLowCardinality(v bool) AttributesOption

type Backup 

type Backup struct {
	// contains filtered or unexported fields
}

Backup implements a oteldb backup process.

Backup is stored in a Clickhouse Native format.

func NewBackup 

func NewBackup(client ClickHouseClient, tables Tables, logger *zap.Logger) *Backup

NewBackup creates a new Backup instance.

func (*Backup) Create 

func (b *Backup) Create(ctx context.Context, dir string) error

Create creates a backup in the specified directory.

type BucketedSampleQuery 

type BucketedSampleQuery struct {
	// Start, End define the output step range.
	Start, End time.Time
	// Step is the output resolution. If <= 0 (instant query), a single
	// bucket covering (End-Range, End] is produced.
	Step time.Duration
	// Range is the range-aggregation window, e.g. the `[5m]` in
	// count_over_time({...}[5m]).
	Range time.Duration
	Sel   LogsSelector

	Sampling SamplingOp
	// GroupingLabels must be non-empty: this query only makes sense for the
	// sum/avg/min/max by(...) (...) shape the optimizer already requires
	// before offloading to it (see querier_logs_optimizer.go).
	GroupingLabels []logql.Label
}

BucketedSampleQuery defines a sample query that aggregates samples per output step directly in ClickHouse, instead of fetching one row per raw log line for logqlmetric.RangeAggregation (range_agg.go) to bucket in Go. It implements the same step-bucketing math as the PromQL rate/increase/delta offload (see querier_metrics_rate.go and chsql_stepfanout.go), without the counter-reset detection tier rate needs — log sample values (line counts, byte lengths) are not counters.

func (*BucketedSampleQuery) Execute 

func (v *BucketedSampleQuery) Execute(ctx context.Context, q *Querier) (_ logqlmetric.StepIterator, rerr error)

Execute executes the query using given querier.

type ClickHouseClient 

type ClickHouseClient interface {
	Do(ctx context.Context, q ch.Query) error
	Ping(ctx context.Context) error
}

func Dial 

func Dial(ctx context.Context, dsn string, opts DialOptions) (ClickHouseClient, error)

Dial creates new ClickHouseClient using given DSN.

func NewDialingClickhouseClient 

func NewDialingClickhouseClient(options ch.Options) ClickHouseClient

type ClickhouseOptimizer 

type ClickhouseOptimizer struct{}

ClickhouseOptimizer replaces LogQL engine execution nodes with optimzied Clickhouse queries.

func (*ClickhouseOptimizer) Name 

func (o *ClickhouseOptimizer) Name() string

Name returns optimizer name.

func (*ClickhouseOptimizer) Optimize 

func (o *ClickhouseOptimizer) Optimize(ctx context.Context, q logqlengine.Query) (logqlengine.Query, error)

Optimize implements [Optimizer].

type Column 

type Column struct {
	Name string
	Data proto.Column
}

Column is a column with name and data that can be used in INSERT or SELECT query.

type Columns 

type Columns []Column

Columns is a set of Columns.

func MergeColumns 

func MergeColumns(sets ...Columns) Columns

MergeColumns merges multiple sets of columns into one.

func (Columns) All 

func (c Columns) All() string

All returns comma-separated column names for using in SELECT query instead of `SELECT *`.

func (Columns) ChsqlResult 

func (c Columns) ChsqlResult() []chsql.ResultColumn

ChsqlResult returns columns for using in SELECT query.

func (Columns) Input 

func (c Columns) Input() proto.Input

Input returns columns for using in INSERT query.

func (Columns) Names 

func (c Columns) Names() []string

Names returns a slice of column names.

func (Columns) Reset 

func (c Columns) Reset()

Reset columns.

func (Columns) Result 

func (c Columns) Result() proto.Results

Result returns columns for using in SELECT query.

type DialOptions 

type DialOptions struct {
	// MeterProvider provides OpenTelemetry meter for pool.
	MeterProvider metric.MeterProvider
	// TracerProvider provides OpenTelemetry tracer for pool.
	TracerProvider trace.TracerProvider
	// Logger provides logger for pool.
	Logger *zap.Logger
}

DialOptions is Dial function options.

type ExpHistogramsBatchFunc added in v0.48.0 

type ExpHistogramsBatchFunc func(ctx context.Context, points []metricstorage.ExpHistogramPoint) error

ExpHistogramsBatchFunc is called with each decoded batch of exponential histograms.

type IncompatibleSchemaError 

type IncompatibleSchemaError struct {
	Table string
}

IncompatibleSchemaError is returned when there is an existing table with incompatible schema.

func (*IncompatibleSchemaError) Error 

func (e *IncompatibleSchemaError) Error() string

IncompatibleSchemaError implements [error].

type InputNode 

type InputNode struct {
	Sel LogsSelector
	// contains filtered or unexported fields
}

InputNode rebuilds LogQL pipeline in as Clickhouse query.

func (*InputNode) EvalPipeline 

func (n *InputNode) EvalPipeline(ctx context.Context, params logqlengine.EvalParams) (logqlengine.EntryIterator, error)

EvalPipeline implements logqlengine.PipelineNode.

func (*InputNode) Traverse 

func (n *InputNode) Traverse(cb logqlengine.NodeVisitor) error

Traverse implements logqlengine.Node.

type Inserter 

type Inserter struct {
	// contains filtered or unexported fields
}

Inserter implements tracestorage.Inserter using Clickhouse.

func NewInserter 

func NewInserter(c ClickHouseClient, opts InserterOptions) (*Inserter, error)

NewInserter creates new Inserter.

func (*Inserter) ConsumeMetrics 

func (i *Inserter) ConsumeMetrics(ctx context.Context, metrics pmetric.Metrics) error

ConsumeMetrics inserts given metrics.

func (*Inserter) RecordWriter 

func (i *Inserter) RecordWriter(ctx context.Context) (logstorage.RecordWriter, error)

RecordWriter returns a new logstorage.RecordWriter

func (*Inserter) SpanWriter 

func (i *Inserter) SpanWriter(ctx context.Context) (tracestorage.SpanWriter, error)

SpanWriter returns a new tracestorage.SpanWriter

type InserterOptions 

type InserterOptions struct {
	// Tables provides table paths to query.
	Tables Tables
	// CHLogLevel sets log level for ch-go.
	CHLogLevel zapcore.LevelEnabler
	// MeterProvider provides OpenTelemetry meter for this querier.
	MeterProvider metric.MeterProvider
	// TracerProvider provides OpenTelemetry tracer for this querier.
	TracerProvider trace.TracerProvider
	// Tracker provides global metric tracker.
	Tracker globalmetric.Tracker
}

InserterOptions is Inserter's options.

type LogsBatchFunc added in v0.48.0 

type LogsBatchFunc func(ctx context.Context, records []logstorage.Record) error

LogsBatchFunc is called with each decoded batch of records read by LogsSource.Do.

type LogsQuery 

type LogsQuery[E any] struct {
	Start, End time.Time
	Sel        LogsSelector
	Direction  logqlengine.Direction
	Limit      int

	Mapper func(logstorage.Record) (E, error)
}

LogsQuery defines a logs query.

func (*LogsQuery[E]) Execute 

func (v *LogsQuery[E]) Execute(ctx context.Context, q *Querier) (_ iterators.Iterator[E], rerr error)

Execute executes the query using given querier.

type LogsSelector 

type LogsSelector struct {
	Labels         []logql.LabelMatcher
	Line           []logql.LineFilter
	PipelineLabels []logql.LabelPredicate
}

LogsSelector defines common parameters for logs selection.

type LogsSource added in v0.48.0 

type LogsSource struct {
	// contains filtered or unexported fields
}

LogsSource reads every log record stored in ClickHouse, decoded as logstorage.Record, for migrating them into another storage engine. Unlike Querier, it performs a full table scan with no selector pushdown, day-bucketed like Backup so a single scan never buffers more than one day's worth of rows in ClickHouse at a time.

func NewLogsSource added in v0.48.0 

func NewLogsSource(client ClickHouseClient, tables Tables, logger *zap.Logger) *LogsSource

NewLogsSource creates a new LogsSource.

func (*LogsSource) Do added in v0.48.0 

func (s *LogsSource) Do(ctx context.Context, since time.Duration, batchSize int, batchFn LogsBatchFunc) error

Do scans every log record in the table, in day-bucket then timestamp order, invoking batchFn with batches of up to batchSize records. When since is positive, the scan is restricted to the last since of data, relative to the table's most recent timestamp, instead of the full table.

type MetricsCacheOptions 

type MetricsCacheOptions = metricscache.Options

MetricsCacheOptions is an alias for metricscache.Options.

type MetricsSource added in v0.48.0 

type MetricsSource struct {
	// contains filtered or unexported fields
}

MetricsSource reads metrics stored in ClickHouse for migration into another storage engine. It first loads the series set from metrics_timeseries (small relative to the point volume) into an in-memory hash→[seriesMeta] map, then day-bucket scans metrics_points and metrics_exp_histograms (mirroring Backup) and resolves each row's series by hash. Exemplars are not read (the target engine drops them). metrics_labels is not read (it is an autocomplete index, deriving nothing the timeseries rows do not already carry).

func NewMetricsSource added in v0.48.0 

func NewMetricsSource(client ClickHouseClient, tables Tables, logger *zap.Logger) *MetricsSource

NewMetricsSource creates a new MetricsSource.

func (*MetricsSource) Do added in v0.48.0 

func (s *MetricsSource) Do(
	ctx context.Context,
	since time.Duration,
	batchSize int,
	numberFn NumberPointsBatchFunc,
	expFn ExpHistogramsBatchFunc,
) error

Do migrates metrics: it loads the series set (restricted to the scan window when since is positive), then scans number points and exponential histograms, invoking numberFn and expFn with batches of up to batchSize decoded points. Rows whose hash is absent from the series set are skipped and counted (logged at the end).

type MigrationDiff 

type MigrationDiff struct {
	Table  string
	Diff   string
	Status MigrationStatus
}

MigrationDiff is result of comparison of existing table and latest schema version.

type MigrationStatus 

type MigrationStatus uint8

MigrationStatus defines status of table.

func (MigrationStatus) ColorString 

func (s MigrationStatus) ColorString() string

ColorString returns colored string representation.

func (MigrationStatus) String 

func (s MigrationStatus) String() string

String implements fmt.Stringer.

type Migrator 

type Migrator struct {
	// contains filtered or unexported fields
}

Migrator provides migration tool for oteldb.

func NewMigrator 

func NewMigrator(client ClickHouseClient, opts MigratorOptions) *Migrator

NewMigrator creates new Migrator.

func (*Migrator) Create 

func (m *Migrator) Create(ctx context.Context) error

Create creates schema.

func (*Migrator) Diff 

func (m *Migrator) Diff(ctx context.Context) ([]MigrationDiff, error)

Diff returns a difference of current schema and latest schema.

func (*Migrator) Drop 

func (m *Migrator) Drop(ctx context.Context, log func(database, table string)) error

Drop drops all known tables. This will remove data.

func (*Migrator) DropIfExists 

func (m *Migrator) DropIfExists(ctx context.Context, log func(database, table string)) error

DropIfExists drops all known existing tables. This will remove data.

func (*Migrator) Validate 

func (m *Migrator) Validate(ctx context.Context) error

Validate performs validation of existing schema without applying any changes.

This can be used in health checks to verify that schema is compatible with the current version of oteldb.

type MigratorOptions 

type MigratorOptions struct {
	Tables           Tables
	Cluster          string
	KeeperPathPrefix string
	TTL              time.Duration
	Replicated       bool
}

MigratorOptions defines migration options.

func (*MigratorOptions) AddFlags 

func (o *MigratorOptions) AddFlags(fs *pflag.FlagSet)

AddFlags registers command-line flags for migrator options.

type NumberPointsBatchFunc added in v0.48.0 

type NumberPointsBatchFunc func(ctx context.Context, points []metricstorage.NumberPoint) error

NumberPointsBatchFunc is called with each decoded batch of number points read by MetricsSource.

type Querier 

type Querier struct {
	// contains filtered or unexported fields
}

Querier implements tracestorage.Querier using Clickhouse.

func NewQuerier 

func NewQuerier(c ClickHouseClient, opts QuerierOptions) (*Querier, error)

NewQuerier creates new Querier.

func (*Querier) Capabilities 

func (q *Querier) Capabilities() (caps logqlengine.QuerierCapabilities)

Capabilities implements logqlengine.Querier.

func (*Querier) DetectedFields 

func (q *Querier) DetectedFields(ctx context.Context, opts logstorage.LabelsOptions) (values []logstorage.DetectedField, rerr error)

DetectedFields implements logstorage.Querier.

func (*Querier) DetectedLabels 

func (q *Querier) DetectedLabels(ctx context.Context, opts logstorage.LabelsOptions) (values []logstorage.DetectedLabel, rerr error)

DetectedLabels implements logstorage.Querier.

func (*Querier) ExemplarQuerier 

func (q *Querier) ExemplarQuerier(ctx context.Context) (storage.ExemplarQuerier, error)

Querier returns a new Querier on the storage.

func (*Querier) LabelNames 

func (q *Querier) LabelNames(ctx context.Context, opts logstorage.LabelsOptions) (result []string, rerr error)

LabelNames implements logstorage.Querier.

func (*Querier) LabelValues 

func (q *Querier) LabelValues(ctx context.Context, labelName string, opts logstorage.LabelsOptions) (riter iterators.Iterator[logstorage.Label], rerr error)

LabelValues implements logstorage.Querier.

func (*Querier) MetricMetadata 

func (q *Querier) MetricMetadata(ctx context.Context, opts metricstorage.MetadataParams) (metricstorage.Metadata, error)

MetricMetadata returns metric metadata for the given options.

func (*Querier) MetricsScanners 

func (q *Querier) MetricsScanners() (enginestorage.Scanners, error)

MetricsScanners returns scanners implementation to use with thanos-io PromQL engine.

func (*Querier) Querier 

func (q *Querier) Querier(mint, maxt int64) (storage.Querier, error)

Querier returns a new metrics storage.Querier.

func (*Querier) Query 

func (q *Querier) Query(ctx context.Context, labels []logql.LabelMatcher) (logqlengine.PipelineNode, error)

Query creates new InputNode.

func (*Querier) SearchTags 

func (q *Querier) SearchTags(ctx context.Context, tags map[string]string, opts tracestorage.SearchTagsOptions) (_ iterators.Iterator[tracestorage.Span], rerr error)

SearchTags performs search by given tags.

func (*Querier) SelectSpansets 

func (q *Querier) SelectSpansets(ctx context.Context, params traceqlengine.SelectSpansetsParams) (_ iterators.Iterator[traceqlengine.Trace], rerr error)

SelectSpansets get spansets from storage.

func (*Querier) Series 

func (q *Querier) Series(ctx context.Context, opts logstorage.SeriesOptions) (result logstorage.Series, rerr error)

Series returns all available log series.

func (*Querier) TagNames 

func (q *Querier) TagNames(ctx context.Context, opts tracestorage.TagNamesOptions) (r []tracestorage.TagName, rerr error)

TagNames returns all available tag names.

func (*Querier) TagValues 

func (q *Querier) TagValues(ctx context.Context, tag traceql.Attribute, opts tracestorage.TagValuesOptions) (_ iterators.Iterator[tracestorage.Tag], rerr error)

TagValues returns all available tag values for given tag.

func (*Querier) TraceByID 

func (q *Querier) TraceByID(ctx context.Context, id otelstorage.TraceID, opts tracestorage.TraceByIDOptions) (_ iterators.Iterator[tracestorage.Span], rerr error)

TraceByID returns spans of given trace.

type QuerierOptions 

type QuerierOptions struct {
	// Tables provides table paths to query.
	Tables Tables
	// LabelLimit defines limit for label lookup in the main table.
	LabelLimit int
	// MetricSeriesLimit defines limit for total number of series requested by the query.
	MetricSeriesLimit int
	// MetricExemplarsLimit defines limit for total number of exemplars returned by a single query.
	MetricExemplarsLimit int

	// MaxResultRows defines max number of rows to read from ClickHouse.
	MaxResultRows int
	// MaxResultBytes defines max number of bytes to read from ClickHouse.
	MaxResultBytes int
	// MaxExecutionTime defines max execution time for ClickHouse query.
	MaxExecutionTime time.Duration

	// MaxSampleRows defines max number of log rows a LogQL sample query
	// (e.g. count_over_time, rate, bytes_over_time) is allowed to fetch.
	MaxSampleRows int
	// MaxSampleResultBytes defines max number of result bytes a LogQL sample
	// query is allowed to fetch from ClickHouse (max_result_bytes override).
	MaxSampleResultBytes int

	// DisableRateOffloading disables rate/increase/delta/etc. offloading to ClickHouse.
	DisableRateOffloading bool
	// DisableMetricOffloading disables all metric offloading to ClickHouse.
	DisableMetricOffloading bool

	// MetricsCacheOptions configures metrics cache.
	MetricsCacheOptions MetricsCacheOptions
	// CHLogLevel sets log level for ch-go.
	CHLogLevel zapcore.LevelEnabler
	// MeterProvider provides OpenTelemetry meter for this querier.
	MeterProvider metric.MeterProvider
	// TracerProvider provides OpenTelemetry tracer for this querier.
	TracerProvider trace.TracerProvider
	// Tracker tracks global metrics.
	Tracker globalmetric.Tracker
}

QuerierOptions is Querier's options.

type Restore 

type Restore struct {
	// contains filtered or unexported fields
}

Restore implements a oteldb restore process.

Restore is stored in a Clickhouse native format.

func NewRestore 

func NewRestore(client ClickHouseClient, tables Tables, logger *zap.Logger) *Restore

NewRestore creates a new Restore instance.

func (*Restore) Restore 

func (b *Restore) Restore(ctx context.Context, dir string) error

Restore performs restore from the given directory.

type SampleQuery 

type SampleQuery struct {
	Start, End     time.Time
	Sel            LogsSelector
	Sampling       SamplingOp
	GroupingLabels []logql.Label
}

SampleQuery defines a sample query.

func (*SampleQuery) Execute 

func (v *SampleQuery) Execute(ctx context.Context, q *Querier) (_ logqlengine.SampleIterator, rerr error)

Execute executes the query using given querier.

type SamplingNode 

type SamplingNode struct {
	Sel            LogsSelector
	Sampling       SamplingOp
	GroupingLabels []logql.Label
	// contains filtered or unexported fields
}

SamplingNode is a logqlengine.SampleNode, which offloads sampling to Clickhouse

func (*SamplingNode) EvalBucketedSample 

func (n *SamplingNode) EvalBucketedSample(
	ctx context.Context,
	params logqlengine.EvalParams,
	window time.Duration,
) (logqlengine.StepIterator, error)

EvalBucketedSample implements logqlengine.BucketedSampleNode.

func (*SamplingNode) EvalSample 

func (n *SamplingNode) EvalSample(ctx context.Context, params logqlengine.EvalParams) (logqlengine.SampleIterator, error)

EvalSample implements logqlengine.SampleNode.

func (*SamplingNode) Traverse 

func (n *SamplingNode) Traverse(cb logqlengine.NodeVisitor) error

Traverse implements logqlengine.Node.

type SamplingOp 

type SamplingOp int

SamplingOp defines a sampler operation. 

const (
	// CountSampling counts lines.
	CountSampling SamplingOp = iota + 1
	// BytesSampling counts line lengths in bytes.
	BytesSampling
)

func (SamplingOp) String 

func (s SamplingOp) String() string

String implments fmt.Stringer.

type Tables 

type Tables struct {
	Spans string
	Tags  string

	Points        string
	Timeseries    string
	ExpHistograms string
	Exemplars     string
	Labels        string

	Logs     string
	LogAttrs string

	Migration string
}

Tables define table names.

func DefaultTables 

func DefaultTables() Tables

DefaultTables returns default tables.

func (*Tables) Each 

func (t *Tables) Each(cb func(name *string) error) error

Each calls given callback for each table.

func (*Tables) Validate 

func (t *Tables) Validate() error

Validate checks table names

type TracesBatchFunc added in v0.48.0 

type TracesBatchFunc func(ctx context.Context, spans []tracestorage.Span) error

TracesBatchFunc is called with each decoded batch of spans read by TracesSource.Do.

type TracesSource added in v0.48.0 

type TracesSource struct {
	// contains filtered or unexported fields
}

TracesSource reads every span stored in ClickHouse, decoded as tracestorage.Span, for migrating them into another storage engine. Unlike Querier, it performs a full table scan with no selector pushdown, day-bucketed like Backup so a single scan never buffers more than one day's worth of rows in ClickHouse at a time.

func NewTracesSource added in v0.48.0 

func NewTracesSource(client ClickHouseClient, tables Tables, logger *zap.Logger) *TracesSource

NewTracesSource creates a new TracesSource.

func (*TracesSource) Do added in v0.48.0 

func (s *TracesSource) Do(ctx context.Context, since time.Duration, batchSize int, batchFn TracesBatchFunc) error

Do scans every span in the table, in day-bucket then start-timestamp order, invoking batchFn with batches of up to batchSize spans. When since is positive, the scan is restricted to the last since of data, relative to the table's most recent start timestamp, instead of the full table.

Source Files

View all Source files

Directories

Path Synopsis
Package chsql provides fluent Clickhouse SQL query builder.
 Package chsql provides fluent Clickhouse SQL query builder.

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.