package config import ( "crypto/tls" "errors" "fmt" "time" "github.com/ClickHouse/clickhouse-go/v2" "github.com/spf13/viper" ) type AggregationConfiguration struct { ClickHouse ClickHouseAggregationConfiguration EventsTableName string // Set true for ClickHouse first store the incoming inserts into an in-memory buffer // before flushing them regularly to disk. // See https://clickhouse.com/docs/en/cloud/bestpractices/asynchronous-inserts AsyncInsert bool // Set true if you want an insert statement to return with an acknowledgment immediately // without waiting for the data got inserted into the buffer. // Setting true can cause silent errors that you need to monitor separately. AsyncInsertWait bool // See https://clickhouse.com/docs/en/operations/settings/settings // For example, you can set the `max_insert_threads` setting to control the number of threads // or the `parallel_view_processing` setting to enable pushing to attached views concurrently. InsertQuerySettings map[string]string // MeterQuerySettings is the settings for the meter query // For example, you can set the `enable_parallel_replicas` and `max_parallel_replicas` settings. // See https://clickhouse.com/docs/en/operations/settings/settings MeterQuerySettings map[string]string // EnablePrewhere is the setting to enable prewhere for the meter query. EnablePrewhere bool // EnableDecimalPrecision enables high precision decimal calculations for meter queries. // When enabled, values are calculated using Decimal128 instead of Float64, providing // higher precision for financial calculations at the cost of some performance. EnableDecimalPrecision bool } // Validate validates the configuration. func (c AggregationConfiguration) Validate() error { if err := c.ClickHouse.Validate(); err != nil { return fmt.Errorf("clickhouse: %w", err) } if c.EventsTableName == "" { return errors.New("events table is required") } if c.AsyncInsertWait && !c.AsyncInsert { return errors.New("async insert wait is set but async insert is not") } return nil } // ClickHouseAggregationConfiguration is the configuration for the ClickHouse aggregation engine type ClickHouseAggregationConfiguration struct { Address string TLS bool Username string Password string Database string // ClickHouse connection options DialTimeout time.Duration MaxOpenConns int MaxIdleConns int ConnMaxLifetime time.Duration BlockBufferSize uint8 Tracing bool PoolMetrics ClickhousePoolMetricsConfig Retry ClickhouseQueryRetryConfig } // Validate validates the configuration. func (c ClickHouseAggregationConfiguration) Validate() error { var errs []error if c.Address == "" { errs = append(errs, errors.New("address is required")) } if c.DialTimeout <= 0 { errs = append(errs, errors.New("dial timeout must be greater than 0")) } if c.MaxOpenConns <= 0 { errs = append(errs, errors.New("max open connections must be greater than 0")) } if c.MaxIdleConns <= 0 { errs = append(errs, errors.New("max idle connections must be greater than 0")) } if c.ConnMaxLifetime <= 0 { errs = append(errs, errors.New("connection max lifetime must be greater than 0")) } if c.BlockBufferSize <= 0 { errs = append(errs, errors.New("block buffer size must be greater than 0")) } if err := c.Retry.Validate(); err != nil { errs = append(errs, fmt.Errorf("retry: %w", err)) } if err := c.PoolMetrics.Validate(); err != nil { errs = append(errs, fmt.Errorf("pool metrics: %w", err)) } return errors.Join(errs...) } func (c ClickHouseAggregationConfiguration) GetClientOptions() *clickhouse.Options { options := &clickhouse.Options{ Addr: []string{c.Address}, Auth: clickhouse.Auth{ Database: c.Database, Username: c.Username, Password: c.Password, }, DialTimeout: c.DialTimeout, MaxOpenConns: c.MaxOpenConns, MaxIdleConns: c.MaxIdleConns, ConnMaxLifetime: c.ConnMaxLifetime, ConnOpenStrategy: clickhouse.ConnOpenInOrder, BlockBufferSize: c.BlockBufferSize, } // This minimal TLS.Config is normally sufficient to connect to the secure native port (normally 9440) on a ClickHouse server. // See: https://clickhouse.com/docs/en/integrations/go#using-tls if c.TLS { options.TLS = &tls.Config{ MinVersion: tls.VersionTLS13, } } return options } type ClickhouseQueryRetryConfig struct { Enabled bool MaxTries int RetryWaitDuration time.Duration MaxDelay time.Duration } func (c ClickhouseQueryRetryConfig) Validate() error { var errs []error if !c.Enabled { return nil } if c.MaxTries < 1 { errs = append(errs, errors.New("max retries must be greater than or equal to 1")) } if c.RetryWaitDuration <= 0 { errs = append(errs, errors.New("retry wait duration must be greater than 0")) } if c.MaxDelay < 0 { errs = append(errs, errors.New("max delay must not be negative")) } return errors.Join(errs...) } type ClickhousePoolMetricsConfig struct { Enabled bool PollInterval time.Duration } func (c ClickhousePoolMetricsConfig) Validate() error { var errs []error if !c.Enabled { return nil } if c.PollInterval <= 0 { errs = append(errs, errors.New("poll interval must be greater than 0")) } return errors.Join(errs...) } // ConfigureAggregation configures some defaults in the Viper instance. func ConfigureAggregation(v *viper.Viper) { v.SetDefault("aggregation.eventsTableName", "om_events") v.SetDefault("aggregation.asyncInsert", false) v.SetDefault("aggregation.asyncInsertWait", false) v.SetDefault("aggregation.clickhouse.address", "127.0.0.1:9000") v.SetDefault("aggregation.clickhouse.tls", false) v.SetDefault("aggregation.clickhouse.database", "openmeter") v.SetDefault("aggregation.clickhouse.username", "default") v.SetDefault("aggregation.clickhouse.password", "default") v.SetDefault("aggregation.clickhouse.tracing", false) // ClickHouse connection options v.SetDefault("aggregation.clickhouse.dialTimeout", "10s") v.SetDefault("aggregation.clickhouse.maxOpenConns", 5) v.SetDefault("aggregation.clickhouse.maxIdleConns", 5) v.SetDefault("aggregation.clickhouse.connMaxLifetime", "10m") v.SetDefault("aggregation.clickhouse.blockBufferSize", 10) // Retry v.SetDefault("aggregation.clickhouse.retry.enabled", false) v.SetDefault("aggregation.clickhouse.retry.maxTries", 3) v.SetDefault("aggregation.clickhouse.retry.retryWaitDuration", "20ms") v.SetDefault("aggregation.clickhouse.retry.maxDelay", "5s") // Pool metrics v.SetDefault("aggregation.clickhouse.poolMetrics.enabled", true) v.SetDefault("aggregation.clickhouse.poolMetrics.pollInterval", "5s") // Decimal precision v.SetDefault("aggregation.enableDecimalPrecision", false) }