diff --git a/cmd/config-migrate/main.go b/cmd/config-migrate/main.go new file mode 100644 index 0000000..08fd1b8 --- /dev/null +++ b/cmd/config-migrate/main.go @@ -0,0 +1,48 @@ +package main + +import ( + "errors" + "flag" + "os" + + "github.com/ozontech/seq-ui/internal/app/config" + "github.com/ozontech/seq-ui/logger" + "go.uber.org/zap" +) + +var ( + source = flag.String("source", "", "path to the config file to migrate in place") +) + +func main() { + flag.Parse() + + run(*source) +} + +func run(source string) { + if source == "" { + logger.Fatal("missing required parameter", zap.String("param", "-source")) + } + + legacyCfg, err := os.ReadFile(source) + if err != nil { + logger.Fatal("error reading legacy config", zap.String("source", source), zap.Error(err)) + } + + backupPath := source + ".bak" + + if _, err := os.Stat(backupPath); err == nil { + logger.Fatal("backup already exists, remove it before re-running", zap.String("backup", backupPath)) + } else if !errors.Is(err, os.ErrNotExist) { + logger.Fatal("stat backup", zap.String("backup", backupPath), zap.Error(err)) + } + if err := os.WriteFile(backupPath, legacyCfg, 0644); err != nil { + logger.Fatal("error writing backup legacy config", zap.String("backup", backupPath), zap.Error(err)) + } + + backupCfg, err := config.FromFile(backupPath) + if err != nil { + logger.Fatal("read backup config file error", zap.Error(err)) + } +} diff --git a/cmd/seq-ui/main.go b/cmd/seq-ui/main.go index d97b3a0..6fa2344 100644 --- a/cmd/seq-ui/main.go +++ b/cmd/seq-ui/main.go @@ -23,7 +23,7 @@ import ( massexport_v1 "github.com/ozontech/seq-ui/internal/api/massexport/v1" seqapi_v1 "github.com/ozontech/seq-ui/internal/api/seqapi/v1" userprofile_v1 "github.com/ozontech/seq-ui/internal/api/userprofile/v1" - "github.com/ozontech/seq-ui/internal/app/config" + config "github.com/ozontech/seq-ui/internal/app/config/v2" "github.com/ozontech/seq-ui/internal/app/server" "github.com/ozontech/seq-ui/internal/pkg/cache" "github.com/ozontech/seq-ui/internal/pkg/client/seqdb" @@ -104,7 +104,7 @@ func run(ctx context.Context) { } } -func initApp(ctx context.Context, cfg config.Config) *api.Registrar { +func initApp(ctx context.Context, cfg config_v2.Config) *api.Registrar { logger.Info("initializing seq-db clients") seqDBClients, err := initSeqDBClients(ctx, cfg) if err != nil { @@ -146,7 +146,7 @@ func initApp(ctx context.Context, cfg config.Config) *api.Registrar { } logger.Info("initializing db") - db, err := initDb(ctx, cfg.Server.DB) + db, err := initDb(ctx, cfg.DB) if err != nil { logger.Fatal("failed to init db", zap.Error(err)) } @@ -225,7 +225,7 @@ func createSeqBDClient(ctx context.Context, cfg config.SeqDBClient, seqAPI confi return seqdb.NewGRPCClient(ctx, clientParams) } -func initDb(ctx context.Context, cfg *config.DB) (*pgxpool.Pool, error) { +func initDb(ctx context.Context, cfg *config_v2.DB) (*pgxpool.Pool, error) { if cfg == nil { logger.Warn("db config is nil, running without db") return nil, nil diff --git a/config/config.example.yaml b/config/config.example.yaml index 604c16c..8e0a5fa 100644 --- a/config/config.example.yaml +++ b/config/config.example.yaml @@ -1,3 +1,5 @@ +version: 2 + server: http_addr: ":5555" grpc_addr: ":5556" diff --git a/config/config.multicluster.yaml b/config/config.multicluster.yaml index 94fb312..3d1eb81 100644 --- a/config/config.multicluster.yaml +++ b/config/config.multicluster.yaml @@ -1,3 +1,5 @@ +version: 2 + server: http_addr: ":5555" grpc_addr: ":5556" diff --git a/docker-compose.yml b/docker-compose.yml index 37cf965..ee1ca48 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -1,4 +1,4 @@ -version: '3.8' +version: "3.8" services: postgres: diff --git a/internal/app/config/config.go b/internal/app/config/config.go index 0e2c63e..b861a80 100644 --- a/internal/app/config/config.go +++ b/internal/app/config/config.go @@ -4,470 +4,66 @@ import ( "bytes" "fmt" "os" - "time" + v1 "github.com/ozontech/seq-ui/internal/app/config/v1" + v2 "github.com/ozontech/seq-ui/internal/app/config/v2" "gopkg.in/yaml.v3" ) const ( - DefaultSeqDBClientID = "default" - - ProxyClientModeGRPC = "grpc" - - MaskModeMask = "mask" - MaskModeReplace = "replace" - MaskModeCut = "cut" - - FieldFilterConditionAnd = "and" - FieldFilterConditionOr = "or" - FieldFilterConditionNot = "not" - - FieldFilterModeEqual = "equal" - FieldFilterModeContains = "contains" - FieldFilterModePrefix = "prefix" - FieldFilterModeSuffix = "suffix" - - minGRPCKeepaliveTime = 10 * time.Second - minGRPCKeepaliveTimeout = 1 * time.Second - - defaultAsyncSearchListQueryLengthLimit = 1000 - - defaultMaxSearchTotalLimit = 1000000 - defaultMaxSearchOffsetLimit = 1000000 - defaultMaxExportLimit = 100000 - defaultMaxAggregationsPerRequest = 1 - defaultMaxBucketsPerAggregationTs = 200 - defaultMaxParallelExportRequests = 1 - - defaultInmemCacheNumCounters = 10000000 - defaultInmemCacheMaxCost = 1000000 - defaultInmemCacheBufferItems = 64 - - defaultEventsCacheTTL = 24 * time.Hour - - defaultLogsLifespanCacheKey = "logs_lifespan" - defaultLogsLifespanCacheTTL = 10 * time.Minute - - defaultClickHouseDialTimeout = 5 * time.Second - defaultClickHouseReadTimeout = 30 * time.Second -) - -type Config struct { - Server *Server `yaml:"server"` - Clients *Clients `yaml:"clients"` - Handlers *Handlers `yaml:"handlers"` -} - -type CORS struct { - AllowedOrigins []string `yaml:"allowed_origins"` - AllowedMethods []string `yaml:"allowed_methods"` - AllowedHeaders []string `yaml:"allowed_headers"` - ExposedHeaders []string `yaml:"exposed_headers"` - AllowCredentials bool `yaml:"allow_credentials"` - MaxAge int `yaml:"max_age"` - OptionsPassthrough bool `yaml:"options_passthrough"` -} - -type OIDC struct { - SkipVerify bool `yaml:"skip_verify"` - AuthURLs []string `yaml:"auth_urls"` - RootCA string `yaml:"root_ca"` - CACert string `yaml:"ca_cert"` - PrivateKey string `yaml:"private_key"` - SSLSkipVerify bool `yaml:"ssl_skip_verify"` - AllowedClients []string `yaml:"allowed_clients"` - CacheSecretKey string `yaml:"cache_secret_key"` -} - -type DB struct { - Name string `yaml:"name"` - Host string `yaml:"host"` - Port int64 `yaml:"port"` - Pass string `yaml:"pass"` - User string `yaml:"user"` - RequestTimeout time.Duration `yaml:"request_timeout"` - ConnectionPoolCapacity int64 `yaml:"connection_pool_capacity"` - UsePreparedStatements *bool `yaml:"use_prepared_statements,omitempty"` -} - -func (db *DB) ConnString() string { - return fmt.Sprintf("host=%s port=%d dbname=%s user=%s password=%s pool_max_conns=%d", db.Host, db.Port, db.Name, db.User, db.Pass, db.ConnectionPoolCapacity) -} - -type CH struct { - Addrs []string `yaml:"addrs"` - Database string `yaml:"database"` - Username string `yaml:"username"` - Password string `yaml:"password"` - Sharded bool `yaml:"sharded"` - DialTimeout time.Duration `yaml:"dial_timeout"` - ReadTimeout time.Duration `yaml:"read_timeout"` -} - -type ( - RateLimiter struct { - RatePerSec int `yaml:"rate_per_sec"` - MaxBurst int `yaml:"max_burst"` - StoreMaxKeys int `yaml:"store_max_keys"` - PerHandler bool `yaml:"per_handler"` - } - - UserToRateLimiter map[string]RateLimiter - - ApiRateLimiters struct { - Default RateLimiter `yaml:"default"` - SpecialUsers UserToRateLimiter `yaml:"spec_users"` - } - - ApiToRateLimiters map[string]ApiRateLimiters + configV1Name = "config_v1" + configV2Name = "config_v2" + previousVersion = 1 + currentVersion = 2 ) -type InmemoryCache struct { - NumCounters int64 `yaml:"num_counters"` - MaxCost int64 `yaml:"max_cost"` - BufferItems int64 `yaml:"buffer_items"` -} - -type Redis struct { - Addr string `yaml:"addr"` - Username string `yaml:"username"` - Password string `yaml:"password"` - Timeout time.Duration `yaml:"timeout"` - MaxRetries int `yaml:"max_retries"` - MinRetryBackoff time.Duration `yaml:"min_retry_backoff"` - MaxRetryBackoff time.Duration `yaml:"max_retry_backoff"` -} - -type Cache struct { - Inmemory InmemoryCache `yaml:"inmemory"` - Redis *Redis `yaml:"redis"` -} - -type S3 struct { - Endpoint string `yaml:"endpoint"` - AccessKeyID string `yaml:"access_key_id"` - SecretAccessKey string `yaml:"secret_access_key"` - BucketName string `yaml:"bucket_name"` - EnableSSl bool `yaml:"enable_ssl"` -} - -type SeqProxyDownloader struct { - Delay time.Duration `yaml:"delay"` - InitialRetryBackoff time.Duration `yaml:"initial_retry_backoff"` - MaxRetryBackoff time.Duration `yaml:"max_retry_backoff"` -} - -type SessionStore struct { - Redis Redis `yaml:"redis"` - ExportLifetime time.Duration `yaml:"export_lifetime"` -} - -type FileStore struct { - S3 *S3 `yaml:"s3"` -} - -type MassExport struct { - BatchSize uint64 `yaml:"batch_size"` - WorkersCount int `yaml:"workers_count"` - TasksChannelSize int `yaml:"tasks_channel_size"` - PartLength time.Duration `yaml:"part_length"` - URLPrefix string `yaml:"url_prefix"` - AllowedUsers []string `yaml:"allowed_users"` - FileStore *FileStore `yaml:"file_store"` - SessionStore *SessionStore `yaml:"session_store"` - SeqProxyDownloader *SeqProxyDownloader `yaml:"seq_proxy_downloader"` -} - -type Server struct { - DebugAddr string `yaml:"debug_addr"` - HTTPAddr string `yaml:"http_addr"` - GRPCAddr string `yaml:"grpc_addr"` - CORS *CORS `yaml:"cors"` - OIDC *OIDC `yaml:"oidc"` - GRPCConnectionTimeout time.Duration `yaml:"grpc_connection_timeout"` - HTTPReadHeaderTimeout time.Duration `yaml:"http_read_header_timeout"` - HTTPReadTimeout time.Duration `yaml:"http_read_timeout"` - HTTPWriteTimeout time.Duration `yaml:"http_write_timeout"` - DB *DB `yaml:"db"` - CH *CH `yaml:"clickhouse"` - RateLimiters ApiToRateLimiters `yaml:"rate_limiters"` - Cache Cache `yaml:"cache"` - JWTSecretKey string `yaml:"jwt_secret_key"` -} - -type GRPCKeepaliveParams struct { - // After a duration of this time if the client doesn't see any activity it - // pings the server to see if the transport is still alive. - // If set below 10s, a minimum value of 10s will be used instead. - Time time.Duration `yaml:"time"` - // After having pinged for keepalive check, the client waits for a duration - // of Timeout and if no activity is seen even after that the connection is - // closed. If set below 1s, a minimum value of 1s will be used instead. - Timeout time.Duration `yaml:"timeout"` - // If true, client sends keepalive pings even with no active RPCs. If false, - // when there are no active RPCs, Time and Timeout will be ignored and no - // keepalive pings will be sent. False by default. - PermitWithoutStream bool `yaml:"permit_without_stream"` -} - -type SeqDBClient struct { - ID string `yaml:"id"` - Timeout time.Duration `yaml:"timeout"` - AvgDocSize int `yaml:"avg_doc_size"` - Addrs []string `yaml:"addrs"` - RequestRetries int `yaml:"request_retries"` - InitialRetryBackoff time.Duration `yaml:"initial_retry_backoff"` - MaxRetryBackoff time.Duration `yaml:"max_retry_backoff"` - ClientMode string `yaml:"client_mode"` - GRPCKeepaliveParams *GRPCKeepaliveParams `yaml:"grpc_keepalive_params"` -} - -type Clients struct { - SeqDBTimeout time.Duration `yaml:"seq_db_timeout"` - SeqDBAvgDocSize int `yaml:"seq_db_avg_doc_size"` - SeqDBAddrs []string `yaml:"seq_db_addrs"` - RequestRetries int `yaml:"request_retries"` - InitialRetryBackoff time.Duration `yaml:"initial_retry_backoff"` - MaxRetryBackoff time.Duration `yaml:"max_retry_backoff"` - ProxyClientMode string `yaml:"proxy_client_mode"` - GRPCKeepaliveParams *GRPCKeepaliveParams `yaml:"grpc_keepalive_params"` - SeqDB []SeqDBClient `yaml:"seq_db"` -} - -type Handlers struct { - SeqAPI SeqAPI `yaml:"seq_api"` - ErrorGroups ErrorGroups `yaml:"error_groups"` - MassExport *MassExport `yaml:"mass_export"` - AsyncSearch AsyncSearch `yaml:"async_search"` -} - -type Field struct { - Name string `yaml:"name"` - Type string `yaml:"type"` -} - -type SeqAPI struct { - *SeqAPIOptions `yaml:",inline"` - Envs map[string]SeqAPIEnv `yaml:"envs"` - DefaultEnv string `yaml:"default_env"` -} - -type SeqAPIEnv struct { - SeqDB string `yaml:"seq_db_id"` - Options *SeqAPIOptions `yaml:"options"` -} - -type SeqAPIOptions struct { - MaxSearchLimit int32 `yaml:"max_search_limit"` - MaxSearchTotalLimit int64 `yaml:"max_search_total_limit"` - MaxSearchOffsetLimit int32 `yaml:"max_search_offset_limit"` - MaxExportLimit int32 `yaml:"max_export_limit"` - SeqCLIMaxSearchLimit int `yaml:"seq_cli_max_search_limit"` - MaxParallelExportRequests int `yaml:"max_parallel_export_requests"` - MaxAggregationsPerRequest int `yaml:"max_aggregations_per_request"` - MaxBucketsPerAggregationTs int `yaml:"max_buckets_per_aggregation_ts"` - EventsCacheTTL time.Duration `yaml:"events_cache_ttl"` - PinnedFields []Field `yaml:"pinned_fields"` - SystemFields []Field `yaml:"system_fields"` - LogsLifespanCacheKey string `yaml:"logs_lifespan_cache_key"` - LogsLifespanCacheTTL time.Duration `yaml:"logs_lifespan_cache_ttl"` - FieldsCacheTTL time.Duration `yaml:"fields_cache_ttl"` - Masking *Masking `yaml:"masking"` -} - -type Masking struct { - Masks []Mask `yaml:"masks"` - ProcessFields []string `yaml:"process_fields"` - IgnoreFields []string `yaml:"ignore_fields"` -} - -type Mask struct { - Re string `yaml:"re"` - Groups []int `yaml:"groups"` - Mode string `yaml:"mode"` // "mask" or "replace" or "cut" - ReplaceWord string `yaml:"replace_word"` // for mode:replace - - ProcessFields []string `yaml:"process_fields"` - IgnoreFields []string `yaml:"ignore_fields"` - - FieldFilters *FieldFilterSet `yaml:"field_filters"` -} - -type FieldFilter struct { - Field string `yaml:"field"` - Mode string `yaml:"mode"` // "equal" or "contains" or "prefix" or "suffix" - Values []string `yaml:"values"` -} - -type FieldFilterSet struct { - Condition string `yaml:"condition"` // "and" or "or" or "not" - Filters []FieldFilter `yaml:"filters"` // max 1 if condition:not -} - -type LogTagsMapping struct { - Env []string `yaml:"env"` - Service []string `yaml:"service"` - Release []string `yaml:"release"` -} - -type ErrorGroups struct { - LogTagsMapping LogTagsMapping `yaml:"log_tags_mapping"` - QueryFilter map[string]string `yaml:"query_filter"` -} - -type AsyncSearch struct { - AdminUsers []string `yaml:"admin_users"` - ListQueryLengthLimit int `yaml:"list_query_length_limit"` +type configMeta struct { + Version *int `yaml:"version"` } // FromFile parse config from config path. -func FromFile(cfgPath string) (Config, error) { +func FromFile(cfgPath string) (v2.Config, error) { cfgBytes, err := os.ReadFile(cfgPath) //nolint:gosec if err != nil { - return Config{}, fmt.Errorf("error reading file: %s", err) + return v2.Config{}, fmt.Errorf("error reading file: %s", err) } - cfg, err := parse(cfgBytes) + meta, err := parse[configMeta](cfgBytes, false) if err != nil { - return Config{}, fmt.Errorf("error parsing file: %s", err) - } - - proxyClientMode := cfg.Clients.ProxyClientMode - if proxyClientMode == "" { - cfg.Clients.ProxyClientMode = ProxyClientModeGRPC - } else if proxyClientMode != ProxyClientModeGRPC { - return Config{}, fmt.Errorf( - "invalid value for clients.proxy_client_mode: %q. Allowed values are empty string (defaults to %q) or %q", - proxyClientMode, ProxyClientModeGRPC, ProxyClientModeGRPC, - ) - } - - setSeqAPIOptionsDefaults(cfg.Handlers.SeqAPI.SeqAPIOptions) - - if cfg.Handlers.AsyncSearch.ListQueryLengthLimit <= 0 { - cfg.Handlers.AsyncSearch.ListQueryLengthLimit = defaultAsyncSearchListQueryLengthLimit + return v2.Config{}, err } - if cfg.Server.DB != nil && cfg.Server.DB.UsePreparedStatements == nil { - cfg.Server.DB.UsePreparedStatements = new(bool) - *cfg.Server.DB.UsePreparedStatements = true - } - - if cfg.Server.CH != nil && cfg.Server.CH.DialTimeout <= 0 { - cfg.Server.CH.DialTimeout = defaultClickHouseDialTimeout - } - - if cfg.Server.CH != nil && cfg.Server.CH.ReadTimeout <= 0 { - cfg.Server.CH.ReadTimeout = defaultClickHouseReadTimeout - } - - if cfg.Clients.GRPCKeepaliveParams != nil { - if cfg.Clients.GRPCKeepaliveParams.Time < minGRPCKeepaliveTime { - cfg.Clients.GRPCKeepaliveParams.Time = minGRPCKeepaliveTime + cfg := v2.Config{} + switch { + case meta.Version == nil || *meta.Version == previousVersion: + cfgV1, err := parse[v1.Config](cfgBytes, true) + if err != nil { + return v2.Config{}, err } - if cfg.Clients.GRPCKeepaliveParams.Timeout < minGRPCKeepaliveTimeout { - cfg.Clients.GRPCKeepaliveParams.Timeout = minGRPCKeepaliveTimeout + cfg = migrateV1ToV2(cfgV1) + case *meta.Version == currentVersion: + cfg, err = parse[v2.Config](cfgBytes, true) + if err != nil { + return v2.Config{}, err } + default: + return v2.Config{}, fmt.Errorf("unsupported config version: %d", *meta.Version) } - if cfg.Server.Cache.Inmemory.NumCounters <= 0 { - cfg.Server.Cache.Inmemory.NumCounters = defaultInmemCacheNumCounters - } - if cfg.Server.Cache.Inmemory.MaxCost <= 0 { - cfg.Server.Cache.Inmemory.MaxCost = defaultInmemCacheMaxCost - } - if cfg.Server.Cache.Inmemory.BufferItems <= 0 { - cfg.Server.Cache.Inmemory.BufferItems = defaultInmemCacheBufferItems - } - - if len(cfg.Clients.SeqDB) == 0 { - defaultClient := SeqDBClient{ - ID: DefaultSeqDBClientID, - Timeout: cfg.Clients.SeqDBTimeout, - AvgDocSize: cfg.Clients.SeqDBAvgDocSize, - Addrs: cfg.Clients.SeqDBAddrs, - RequestRetries: cfg.Clients.RequestRetries, - InitialRetryBackoff: cfg.Clients.InitialRetryBackoff, - MaxRetryBackoff: cfg.Clients.MaxRetryBackoff, - ClientMode: cfg.Clients.ProxyClientMode, - GRPCKeepaliveParams: cfg.Clients.GRPCKeepaliveParams, - } - cfg.Clients.SeqDB = []SeqDBClient{defaultClient} - } - - clientIDs := make(map[string]struct{}) - for _, client := range cfg.Clients.SeqDB { - if client.ID == "" { - return Config{}, fmt.Errorf("seq_db client ID cannot be empty") - } - if _, ok := clientIDs[client.ID]; ok { - return Config{}, fmt.Errorf("duplicate seq_db client ID: %s", client.ID) - } - clientIDs[client.ID] = struct{}{} - } - - if len(cfg.Handlers.SeqAPI.Envs) > 0 { - if cfg.Handlers.SeqAPI.DefaultEnv == "" { - return Config{}, fmt.Errorf("default_env must be specified when using envs") - } - - if _, exists := cfg.Handlers.SeqAPI.Envs[cfg.Handlers.SeqAPI.DefaultEnv]; !exists { - return Config{}, fmt.Errorf("default_env '%s' not found in seq_api.envs", cfg.Handlers.SeqAPI.DefaultEnv) - } - - for envName, envConfig := range cfg.Handlers.SeqAPI.Envs { - if _, ok := clientIDs[envConfig.SeqDB]; !ok { - return Config{}, fmt.Errorf("client '%s' for env '%s' not found", envConfig.SeqDB, envName) - } - - if envConfig.Options == nil { - envConfig.Options = cfg.Handlers.SeqAPI.SeqAPIOptions - } else { - setSeqAPIOptionsDefaults(envConfig.Options) - } - - cfg.Handlers.SeqAPI.Envs[envName] = envConfig - } + if err := v2.Normalize(&cfg); err != nil { + return v2.Config{}, fmt.Errorf("normalize config: %w") } return cfg, nil } -func setSeqAPIOptionsDefaults(options *SeqAPIOptions) { - if options.MaxAggregationsPerRequest <= 0 { - options.MaxAggregationsPerRequest = defaultMaxAggregationsPerRequest - } - if options.MaxBucketsPerAggregationTs <= 0 { - options.MaxBucketsPerAggregationTs = defaultMaxBucketsPerAggregationTs - } - if options.MaxParallelExportRequests <= 0 { - options.MaxParallelExportRequests = defaultMaxParallelExportRequests - } - if options.MaxSearchTotalLimit <= 0 { - options.MaxSearchTotalLimit = defaultMaxSearchTotalLimit - } - if options.MaxSearchOffsetLimit <= 0 { - options.MaxSearchOffsetLimit = defaultMaxSearchOffsetLimit - } - if options.MaxExportLimit <= 0 { - options.MaxExportLimit = defaultMaxExportLimit - } - if options.EventsCacheTTL <= 0 { - options.EventsCacheTTL = defaultEventsCacheTTL - } - if options.LogsLifespanCacheKey == "" { - options.LogsLifespanCacheKey = defaultLogsLifespanCacheKey - } - if options.LogsLifespanCacheTTL <= 0 { - options.LogsLifespanCacheTTL = defaultLogsLifespanCacheTTL - } -} - -func parse(cfg []byte) (Config, error) { - result := Config{} +func parse[T any](cfg []byte, strict bool) (T, error) { + var result T decoder := yaml.NewDecoder(bytes.NewReader(cfg)) - decoder.KnownFields(true) + if strict { + decoder.KnownFields(true) + } if err := decoder.Decode(&result); err != nil { return result, fmt.Errorf("error parsing config: %w", err) } diff --git a/internal/app/config/migrate_v1_v2.go b/internal/app/config/migrate_v1_v2.go new file mode 100644 index 0000000..8dc9df8 --- /dev/null +++ b/internal/app/config/migrate_v1_v2.go @@ -0,0 +1,443 @@ +package config + +import ( + v1 "github.com/ozontech/seq-ui/internal/app/config/v1" + v2 "github.com/ozontech/seq-ui/internal/app/config/v2" +) + +func migrateV1ToV2(src v1.Config) v2.Config { + dst := v2.Config{Version: currentVersion} + + dst.Server = migrateServer(src.Server) + dst.Cache = migrateCache(src.Server) + dst.Clients = migrateClients(src) + dst.DB = migrateDB(src.Server) + dst.Handlers = migrateHandlers(src.Handlers, &dst) + + return dst +} + +func migrateServer(src *v1.Server) *v2.Server { + dst := &v2.Server{ + HTTP: v2.HTTP{ + Addr: src.HTTPAddr, + ReadTimeout: src.HTTPReadTimeout, + ReadHeaderTimeout: src.HTTPReadHeaderTimeout, + WriteTimeout: src.HTTPWriteTimeout, + CORS: migrateCORS(src.CORS), + }, + GRPC: v2.GRPC{ + Addr: src.GRPCAddr, + ConnectionTimeout: src.GRPCConnectionTimeout, + }, + Debug: v2.Debug{ + Addr: src.DebugAddr, + }, + RateLimiters: migrateApiRateLimiters(src.RateLimiters), + } + + if src.OIDC != nil { + dst.Auth = &v2.Auth{ + OIDC: &v2.OIDC{ + SkipVerify: src.OIDC.SkipVerify, + AuthURLs: src.OIDC.AuthURLs, + RootCA: src.OIDC.RootCA, + CACert: src.OIDC.CACert, + PrivateKey: src.OIDC.PrivateKey, + SSLSkipVerify: src.OIDC.SSLSkipVerify, + AllowedClients: src.OIDC.AllowedClients, + CacheSecretKey: src.OIDC.CacheSecretKey, + }, + } + } + + if src.JWTSecretKey != "" { + dst.Auth.JWT = &v2.JWT{SecretKey: src.JWTSecretKey} + } + + return dst +} + +func migrateCORS(src *v1.CORS) *v2.CORS { + if src == nil { + return nil + } + + return &v2.CORS{ + AllowedOrigins: src.AllowedOrigins, + AllowedMethods: src.AllowedMethods, + AllowedHeaders: src.AllowedHeaders, + ExposedHeaders: src.ExposedHeaders, + AllowCredentials: src.AllowCredentials, + MaxAge: src.MaxAge, + OptionsPassthrough: src.OptionsPassthrough, + } +} + +func migrateApiRateLimiters(src v1.ApiToRateLimiters) v2.ApiToRateLimiters { + if src == nil { + return nil + } + + dst := make(v2.ApiToRateLimiters, len(src)) + for api, rl := range src { + dst[api] = v2.ApiRateLimiters{ + Default: v2.RateLimiter{ + RatePerSec: rl.Default.RatePerSec, + MaxBurst: rl.Default.MaxBurst, + StoreMaxKeys: rl.Default.StoreMaxKeys, + PerHandler: rl.Default.PerHandler, + }, + SpecialUsers: migrateUserToRateLimiter(rl.SpecialUsers), + } + } + + return dst +} + +func migrateUserToRateLimiter(src v1.UserToRateLimiter) v2.UserToRateLimiter { + if src == nil { + return nil + } + + dst := make(v2.UserToRateLimiter, len(src)) + for k, rl := range src { + dst[k] = v2.RateLimiter{ + RatePerSec: rl.RatePerSec, + MaxBurst: rl.MaxBurst, + StoreMaxKeys: rl.StoreMaxKeys, + PerHandler: rl.PerHandler, + } + } + + return dst +} + +func migrateGRPCKeepaliveParams(src *v1.GRPCKeepaliveParams) *v2.GRPCKeepaliveParams { + if src == nil { + return nil + } + + return &v2.GRPCKeepaliveParams{ + Time: src.Time, + Timeout: src.Timeout, + PermitWithoutStream: src.PermitWithoutStream, + } +} + +func migrateClients(src v1.Config) *v2.Clients { + dst := &v2.Clients{} + + if src.Clients != nil { + if len(src.Clients.SeqDB) > 0 { + dst.SeqDB = make([]v2.SeqDBClient, 0, len(src.Clients.SeqDB)) + for _, s := range src.Clients.SeqDB { + dst.SeqDB = append(dst.SeqDB, migrateSeqDBClient(s)) + } + } else { + dst.SeqDB = []v2.SeqDBClient{{ + ID: v2.DefaultSeqDBClientID, + Timeout: src.Clients.SeqDBTimeout, + AvgDocSize: src.Clients.SeqDBAvgDocSize, + Addrs: src.Clients.SeqDBAddrs, + RequestRetries: src.Clients.RequestRetries, + InitialRetryBackoff: src.Clients.InitialRetryBackoff, + MaxRetryBackoff: src.Clients.MaxRetryBackoff, + ClientMode: src.Clients.ProxyClientMode, + GRPCKeepaliveParams: migrateGRPCKeepaliveParams(src.Clients.GRPCKeepaliveParams), + }} + } + } + + if src.Server != nil && src.Server.CH != nil { + ch := src.Server.CH + dst.ClickHouse = []v2.CHClient{{ + ID: v2.DefaultSeqDBClientID, + Addrs: ch.Addrs, + Database: ch.Database, + Username: ch.Username, + Password: ch.Password, + Sharded: ch.Sharded, + DialTimeout: ch.DialTimeout, + ReadTimeout: ch.ReadTimeout, + }} + } + + return dst +} + +func migrateSeqDBClient(src v1.SeqDBClient) v2.SeqDBClient { + return v2.SeqDBClient{ + ID: src.ID, + Timeout: src.Timeout, + AvgDocSize: src.AvgDocSize, + Addrs: src.Addrs, + RequestRetries: src.RequestRetries, + InitialRetryBackoff: src.InitialRetryBackoff, + MaxRetryBackoff: src.MaxRetryBackoff, + ClientMode: src.ClientMode, + GRPCKeepaliveParams: migrateGRPCKeepaliveParams(src.GRPCKeepaliveParams), + } +} + +func migrateDB(src *v1.Server) *v2.DB { + if src == nil || src.DB == nil { + return nil + } + + return &v2.DB{ + Name: src.DB.Name, + Host: src.DB.Host, + Port: src.DB.Port, + Pass: src.DB.Pass, + User: src.DB.User, + RequestTimeout: src.DB.RequestTimeout, + ConnectionPoolCapacity: src.DB.ConnectionPoolCapacity, + UsePreparedStatements: src.DB.UsePreparedStatements, + } +} + +func migrateCache(src *v1.Server) *v2.Cache { + if src == nil { + return &v2.Cache{} + } + + cache := &v2.Cache{} + cache.Inmemory = append(cache.Inmemory, v2.InmemoryCache{ + ID: v2.DefaultInmemCacheID, + NumCounters: src.Cache.Inmemory.NumCounters, + MaxCost: src.Cache.Inmemory.MaxCost, + BufferItems: src.Cache.Inmemory.BufferItems, + }) + + if src.Cache.Redis != nil { + cache.Redis = append(cache.Redis, migrateRedis(src.Cache.Redis, v2.DefaultRedisID, v2.DefaultInmemCacheID)) + } + + return cache +} + +func migrateRedis(src *v1.Redis, id, withInmemID string) v2.Redis { + return v2.Redis{ + ID: id, + WithInmemID: withInmemID, + Addr: src.Addr, + Username: src.Username, + Password: src.Password, + Timeout: src.Timeout, + MaxRetries: src.MaxRetries, + MinRetryBackoff: src.MinRetryBackoff, + MaxRetryBackoff: src.MaxRetryBackoff, + } +} + +func migrateHandlers(src *v1.Handlers, cfg *v2.Config) *v2.Handlers { + if src == nil { + return &v2.Handlers{} + } + + dst := &v2.Handlers{ + SeqAPI: migrateSeqAPI(src.SeqAPI), + ErrorGroups: migrateErrorGroups(src.ErrorGroups), + AsyncSearch: migrateAsyncSearch(src.AsyncSearch), + } + + if src.MassExport != nil { + dst.MassExport = migrateMassExport(src.MassExport, cfg) + } + + return dst +} + +func migrateSeqAPI(src v1.SeqAPI) v2.SeqAPI { + return v2.SeqAPI{ + SeqAPIOptions: migrateSeqAPIOptions(src.SeqAPIOptions), + Envs: migrateSeqAPIEnvs(src.Envs), + DefaultEnv: src.DefaultEnv, + } +} + +func migrateSeqAPIEnvs(envs map[string]v1.SeqAPIEnv) map[string]v2.SeqAPIEnv { + if len(envs) == 0 { + return nil + } + + dst := make(map[string]v2.SeqAPIEnv, len(envs)) + for name, cfg := range envs { + dst[name] = v2.SeqAPIEnv{ + SeqDB: cfg.SeqDB, + Options: migrateSeqAPIOptions(cfg.Options), + } + } + + return dst +} + +func migrateSeqAPIOptions(options *v1.SeqAPIOptions) *v2.SeqAPIOptions { + if options == nil { + return nil + } + + return &v2.SeqAPIOptions{ + MaxSearchLimit: options.MaxSearchLimit, + MaxSearchTotalLimit: options.MaxSearchTotalLimit, + MaxSearchOffsetLimit: options.MaxSearchOffsetLimit, + MaxExportLimit: options.MaxExportLimit, + SeqCLIMaxSearchLimit: options.SeqCLIMaxSearchLimit, + MaxParallelExportRequests: options.MaxParallelExportRequests, + MaxAggregationsPerRequest: options.MaxAggregationsPerRequest, + MaxBucketsPerAggregationTs: options.MaxBucketsPerAggregationTs, + EventsCacheTTL: options.EventsCacheTTL, + PinnedFields: migrateFields(options.PinnedFields), + SystemFields: migrateFields(options.SystemFields), + LogsLifespanCacheKey: options.LogsLifespanCacheKey, + LogsLifespanCacheTTL: options.LogsLifespanCacheTTL, + FieldsCacheTTL: options.FieldsCacheTTL, + Masking: migrateMasking(options.Masking), + } +} + +func migrateFields(fs []v1.Field) []v2.Field { + if fs == nil { + return nil + } + + dst := make([]v2.Field, len(fs)) + for i, f := range fs { + dst[i] = v2.Field{ + Name: f.Name, + Type: f.Type, + } + } + + return dst +} + +func migrateMasking(src *v1.Masking) *v2.Masking { + if src == nil { + return nil + } + + return &v2.Masking{ + Masks: migrateMasks(src.Masks), + ProcessFields: src.ProcessFields, + IgnoreFields: src.IgnoreFields, + } +} + +func migrateMasks(ms []v1.Mask) []v2.Mask { + if ms == nil { + return nil + } + + dst := make([]v2.Mask, len(ms)) + for i, m := range ms { + dst[i] = v2.Mask{ + Re: m.Re, + Groups: m.Groups, + Mode: m.Mode, + ReplaceWord: m.ReplaceWord, + ProcessFields: m.ProcessFields, + IgnoreFields: m.IgnoreFields, + FieldFilters: migrateFieldFilters(m.FieldFilters), + } + } + + return dst +} + +func migrateFieldFilters(src *v1.FieldFilterSet) *v2.FieldFilterSet { + if src == nil { + return nil + } + + dst := &v2.FieldFilterSet{ + Condition: src.Condition, + } + + if src.Filters != nil { + dst.Filters = make([]v2.FieldFilter, len(src.Filters)) + for i, f := range src.Filters { + dst.Filters[i] = v2.FieldFilter{ + Field: f.Field, + Mode: f.Mode, + Values: f.Values, + } + } + } + + return dst +} + +func migrateErrorGroups(eg v1.ErrorGroups) v2.ErrorGroups { + return v2.ErrorGroups{ + LogTagsMapping: v2.LogTagsMapping{ + Env: eg.LogTagsMapping.Env, + Service: eg.LogTagsMapping.Service, + Release: eg.LogTagsMapping.Release, + }, + QueryFilter: eg.QueryFilter, + } +} + +func migrateAsyncSearch(a v1.AsyncSearch) v2.AsyncSearch { + return v2.AsyncSearch{ + AdminUsers: a.AdminUsers, + ListQueryLengthLimit: a.ListQueryLengthLimit, + } +} + +func migrateMassExport(me *v1.MassExport, cfg *v2.Config) *v2.MassExport { + dst := &v2.MassExport{ + BatchSize: me.BatchSize, + WorkersCount: me.WorkersCount, + TasksChannelSize: me.TasksChannelSize, + PartLength: me.PartLength, + URLPrefix: me.URLPrefix, + AllowedUsers: me.AllowedUsers, + FileStore: migrateFileStore(me.FileStore), + DownloadParams: migrateDownloadParams(me.SeqProxyDownloader), + } + + if me.SessionStore != nil { + cfg.Cache.Redis = append(cfg.Cache.Redis, migrateRedis(&me.SessionStore.Redis, v2.DefaultMassExportRedisID, "")) + + dst.SessionStore = &v2.SessionStore{ + RedisID: v2.DefaultMassExportRedisID, + ExportLifetime: me.SessionStore.ExportLifetime, + } + } + + return dst +} + +func migrateFileStore(fs *v1.FileStore) *v2.FileStore { + if fs == nil { + return nil + } + + dst := &v2.FileStore{} + if fs.S3 != nil { + dst.S3 = &v2.S3{ + Endpoint: fs.S3.Endpoint, + AccessKeyID: fs.S3.AccessKeyID, + SecretAccessKey: fs.S3.SecretAccessKey, + BucketName: fs.S3.BucketName, + EnableSSl: fs.S3.EnableSSl, + } + } + + return dst +} + +func migrateDownloadParams(src *v1.SeqProxyDownloader) *v2.DownloadParams { + if src == nil { + return nil + } + + return &v2.DownloadParams{ + Delay: src.Delay, + InitialRetryBackoff: src.InitialRetryBackoff, + MaxRetryBackoff: src.MaxRetryBackoff, + } +} diff --git a/internal/app/config/v1/config.go b/internal/app/config/v1/config.go new file mode 100644 index 0000000..9851298 --- /dev/null +++ b/internal/app/config/v1/config.go @@ -0,0 +1,270 @@ +package v1 + +import ( + "time" +) + +type Config struct { + Version *int `yaml:"version"` + Server *Server `yaml:"server"` + Clients *Clients `yaml:"clients"` + Handlers *Handlers `yaml:"handlers"` +} + +type CORS struct { + AllowedOrigins []string `yaml:"allowed_origins"` + AllowedMethods []string `yaml:"allowed_methods"` + AllowedHeaders []string `yaml:"allowed_headers"` + ExposedHeaders []string `yaml:"exposed_headers"` + AllowCredentials bool `yaml:"allow_credentials"` + MaxAge int `yaml:"max_age"` + OptionsPassthrough bool `yaml:"options_passthrough"` +} + +type OIDC struct { + SkipVerify bool `yaml:"skip_verify"` + AuthURLs []string `yaml:"auth_urls"` + RootCA string `yaml:"root_ca"` + CACert string `yaml:"ca_cert"` + PrivateKey string `yaml:"private_key"` + SSLSkipVerify bool `yaml:"ssl_skip_verify"` + AllowedClients []string `yaml:"allowed_clients"` + CacheSecretKey string `yaml:"cache_secret_key"` +} + +type DB struct { + Name string `yaml:"name"` + Host string `yaml:"host"` + Port int64 `yaml:"port"` + Pass string `yaml:"pass"` + User string `yaml:"user"` + RequestTimeout time.Duration `yaml:"request_timeout"` + ConnectionPoolCapacity int64 `yaml:"connection_pool_capacity"` + UsePreparedStatements *bool `yaml:"use_prepared_statements,omitempty"` +} + +type CH struct { + Addrs []string `yaml:"addrs"` + Database string `yaml:"database"` + Username string `yaml:"username"` + Password string `yaml:"password"` + Sharded bool `yaml:"sharded"` + DialTimeout time.Duration `yaml:"dial_timeout"` + ReadTimeout time.Duration `yaml:"read_timeout"` +} + +type ( + RateLimiter struct { + RatePerSec int `yaml:"rate_per_sec"` + MaxBurst int `yaml:"max_burst"` + StoreMaxKeys int `yaml:"store_max_keys"` + PerHandler bool `yaml:"per_handler"` + } + + UserToRateLimiter map[string]RateLimiter + + ApiRateLimiters struct { + Default RateLimiter `yaml:"default"` + SpecialUsers UserToRateLimiter `yaml:"spec_users"` + } + + ApiToRateLimiters map[string]ApiRateLimiters +) + +type InmemoryCache struct { + NumCounters int64 `yaml:"num_counters"` + MaxCost int64 `yaml:"max_cost"` + BufferItems int64 `yaml:"buffer_items"` +} + +type Redis struct { + Addr string `yaml:"addr"` + Username string `yaml:"username"` + Password string `yaml:"password"` + Timeout time.Duration `yaml:"timeout"` + MaxRetries int `yaml:"max_retries"` + MinRetryBackoff time.Duration `yaml:"min_retry_backoff"` + MaxRetryBackoff time.Duration `yaml:"max_retry_backoff"` +} + +type Cache struct { + Inmemory InmemoryCache `yaml:"inmemory"` + Redis *Redis `yaml:"redis"` +} + +type S3 struct { + Endpoint string `yaml:"endpoint"` + AccessKeyID string `yaml:"access_key_id"` + SecretAccessKey string `yaml:"secret_access_key"` + BucketName string `yaml:"bucket_name"` + EnableSSl bool `yaml:"enable_ssl"` +} + +type SeqProxyDownloader struct { + Delay time.Duration `yaml:"delay"` + InitialRetryBackoff time.Duration `yaml:"initial_retry_backoff"` + MaxRetryBackoff time.Duration `yaml:"max_retry_backoff"` +} + +type SessionStore struct { + Redis Redis `yaml:"redis"` + ExportLifetime time.Duration `yaml:"export_lifetime"` +} + +type FileStore struct { + S3 *S3 `yaml:"s3"` +} + +type MassExport struct { + BatchSize uint64 `yaml:"batch_size"` + WorkersCount int `yaml:"workers_count"` + TasksChannelSize int `yaml:"tasks_channel_size"` + PartLength time.Duration `yaml:"part_length"` + URLPrefix string `yaml:"url_prefix"` + AllowedUsers []string `yaml:"allowed_users"` + FileStore *FileStore `yaml:"file_store"` + SessionStore *SessionStore `yaml:"session_store"` + SeqProxyDownloader *SeqProxyDownloader `yaml:"seq_proxy_downloader"` +} + +type Server struct { + DebugAddr string `yaml:"debug_addr"` + HTTPAddr string `yaml:"http_addr"` + GRPCAddr string `yaml:"grpc_addr"` + CORS *CORS `yaml:"cors"` + OIDC *OIDC `yaml:"oidc"` + GRPCConnectionTimeout time.Duration `yaml:"grpc_connection_timeout"` + HTTPReadHeaderTimeout time.Duration `yaml:"http_read_header_timeout"` + HTTPReadTimeout time.Duration `yaml:"http_read_timeout"` + HTTPWriteTimeout time.Duration `yaml:"http_write_timeout"` + DB *DB `yaml:"db"` + CH *CH `yaml:"clickhouse"` + RateLimiters ApiToRateLimiters `yaml:"rate_limiters"` + Cache Cache `yaml:"cache"` + JWTSecretKey string `yaml:"jwt_secret_key"` +} + +type GRPCKeepaliveParams struct { + // After a duration of this time if the client doesn't see any activity it + // pings the server to see if the transport is still alive. + // If set below 10s, a minimum value of 10s will be used instead. + Time time.Duration `yaml:"time"` + // After having pinged for keepalive check, the client waits for a duration + // of Timeout and if no activity is seen even after that the connection is + // closed. If set below 1s, a minimum value of 1s will be used instead. + Timeout time.Duration `yaml:"timeout"` + // If true, client sends keepalive pings even with no active RPCs. If false, + // when there are no active RPCs, Time and Timeout will be ignored and no + // keepalive pings will be sent. False by default. + PermitWithoutStream bool `yaml:"permit_without_stream"` +} + +type SeqDBClient struct { + ID string `yaml:"id"` + Timeout time.Duration `yaml:"timeout"` + AvgDocSize int `yaml:"avg_doc_size"` + Addrs []string `yaml:"addrs"` + RequestRetries int `yaml:"request_retries"` + InitialRetryBackoff time.Duration `yaml:"initial_retry_backoff"` + MaxRetryBackoff time.Duration `yaml:"max_retry_backoff"` + ClientMode string `yaml:"client_mode"` + GRPCKeepaliveParams *GRPCKeepaliveParams `yaml:"grpc_keepalive_params"` +} + +type Clients struct { + SeqDBTimeout time.Duration `yaml:"seq_db_timeout"` + SeqDBAvgDocSize int `yaml:"seq_db_avg_doc_size"` + SeqDBAddrs []string `yaml:"seq_db_addrs"` + RequestRetries int `yaml:"request_retries"` + InitialRetryBackoff time.Duration `yaml:"initial_retry_backoff"` + MaxRetryBackoff time.Duration `yaml:"max_retry_backoff"` + ProxyClientMode string `yaml:"proxy_client_mode"` + GRPCKeepaliveParams *GRPCKeepaliveParams `yaml:"grpc_keepalive_params"` + SeqDB []SeqDBClient `yaml:"seq_db"` +} + +type Handlers struct { + SeqAPI SeqAPI `yaml:"seq_api"` + ErrorGroups ErrorGroups `yaml:"error_groups"` + MassExport *MassExport `yaml:"mass_export"` + AsyncSearch AsyncSearch `yaml:"async_search"` +} + +type Field struct { + Name string `yaml:"name"` + Type string `yaml:"type"` +} + +type SeqAPI struct { + *SeqAPIOptions `yaml:",inline"` + Envs map[string]SeqAPIEnv `yaml:"envs"` + DefaultEnv string `yaml:"default_env"` +} + +type SeqAPIEnv struct { + SeqDB string `yaml:"seq_db_id"` + Options *SeqAPIOptions `yaml:"options"` +} + +type SeqAPIOptions struct { + MaxSearchLimit int32 `yaml:"max_search_limit"` + MaxSearchTotalLimit int64 `yaml:"max_search_total_limit"` + MaxSearchOffsetLimit int32 `yaml:"max_search_offset_limit"` + MaxExportLimit int32 `yaml:"max_export_limit"` + SeqCLIMaxSearchLimit int `yaml:"seq_cli_max_search_limit"` + MaxParallelExportRequests int `yaml:"max_parallel_export_requests"` + MaxAggregationsPerRequest int `yaml:"max_aggregations_per_request"` + MaxBucketsPerAggregationTs int `yaml:"max_buckets_per_aggregation_ts"` + EventsCacheTTL time.Duration `yaml:"events_cache_ttl"` + PinnedFields []Field `yaml:"pinned_fields"` + SystemFields []Field `yaml:"system_fields"` + LogsLifespanCacheKey string `yaml:"logs_lifespan_cache_key"` + LogsLifespanCacheTTL time.Duration `yaml:"logs_lifespan_cache_ttl"` + FieldsCacheTTL time.Duration `yaml:"fields_cache_ttl"` + Masking *Masking `yaml:"masking"` +} + +type Masking struct { + Masks []Mask `yaml:"masks"` + ProcessFields []string `yaml:"process_fields"` + IgnoreFields []string `yaml:"ignore_fields"` +} + +type Mask struct { + Re string `yaml:"re"` + Groups []int `yaml:"groups"` + Mode string `yaml:"mode"` // "mask" or "replace" or "cut" + ReplaceWord string `yaml:"replace_word"` // for mode:replace + + ProcessFields []string `yaml:"process_fields"` + IgnoreFields []string `yaml:"ignore_fields"` + + FieldFilters *FieldFilterSet `yaml:"field_filters"` +} + +type FieldFilter struct { + Field string `yaml:"field"` + Mode string `yaml:"mode"` // "equal" or "contains" or "prefix" or "suffix" + Values []string `yaml:"values"` +} + +type FieldFilterSet struct { + Condition string `yaml:"condition"` // "and" or "or" or "not" + Filters []FieldFilter `yaml:"filters"` // max 1 if condition:not +} + +type LogTagsMapping struct { + Env []string `yaml:"env"` + Service []string `yaml:"service"` + Release []string `yaml:"release"` +} + +type ErrorGroups struct { + LogTagsMapping LogTagsMapping `yaml:"log_tags_mapping"` + QueryFilter map[string]string `yaml:"query_filter"` +} + +type AsyncSearch struct { + AdminUsers []string `yaml:"admin_users"` + ListQueryLengthLimit int `yaml:"list_query_length_limit"` +} diff --git a/internal/app/config/v2/config.go b/internal/app/config/v2/config.go new file mode 100644 index 0000000..882c00a --- /dev/null +++ b/internal/app/config/v2/config.go @@ -0,0 +1,510 @@ +package v2 + +import ( + "fmt" + "time" +) + +const ( + DefaultSeqDBClientID = "default" + DefaultInmemCacheID = "seqapi" + DefaultRedisID = "default" + DefaultMassExportRedisID = "mass_export" + + ProxyClientModeGRPC = "grpc" + + MaskModeMask = "mask" + MaskModeReplace = "replace" + MaskModeCut = "cut" + + FieldFilterConditionAnd = "and" + FieldFilterConditionOr = "or" + FieldFilterConditionNot = "not" + + FieldFilterModeEqual = "equal" + FieldFilterModeContains = "contains" + FieldFilterModePrefix = "prefix" + FieldFilterModeSuffix = "suffix" + + minGRPCKeepaliveTime = 10 * time.Second + minGRPCKeepaliveTimeout = 1 * time.Second + + defaultAsyncSearchListQueryLengthLimit = 1000 + + defaultMaxSearchTotalLimit = 1000000 + defaultMaxSearchOffsetLimit = 1000000 + defaultMaxExportLimit = 100000 + defaultMaxAggregationsPerRequest = 1 + defaultMaxBucketsPerAggregationTs = 200 + defaultMaxParallelExportRequests = 1 + + defaultInmemCacheNumCounters = 10000000 + defaultInmemCacheMaxCost = 1000000 + defaultInmemCacheBufferItems = 64 + + defaultEventsCacheTTL = 24 * time.Hour + + defaultLogsLifespanCacheKey = "logs_lifespan" + defaultLogsLifespanCacheTTL = 10 * time.Minute + + defaultClickHouseDialTimeout = 5 * time.Second + defaultClickHouseReadTimeout = 30 * time.Second +) + +type Config struct { + Version int `yaml:"version"` + Server *Server `yaml:"server"` + Clients *Clients `yaml:"clients"` + Handlers *Handlers `yaml:"handlers"` + DB *DB `yaml:"db"` + Cache *Cache `yaml:"cache"` +} + +type CORS struct { + AllowedOrigins []string `yaml:"allowed_origins"` + AllowedMethods []string `yaml:"allowed_methods"` + AllowedHeaders []string `yaml:"allowed_headers"` + ExposedHeaders []string `yaml:"exposed_headers"` + AllowCredentials bool `yaml:"allow_credentials"` + MaxAge int `yaml:"max_age"` + OptionsPassthrough bool `yaml:"options_passthrough"` +} + +type OIDC struct { + SkipVerify bool `yaml:"skip_verify"` + AuthURLs []string `yaml:"auth_urls"` + RootCA string `yaml:"root_ca"` + CACert string `yaml:"ca_cert"` + PrivateKey string `yaml:"private_key"` + SSLSkipVerify bool `yaml:"ssl_skip_verify"` + AllowedClients []string `yaml:"allowed_clients"` + CacheSecretKey string `yaml:"cache_secret_key"` +} + +type DB struct { + Name string `yaml:"name"` + Host string `yaml:"host"` + Port int64 `yaml:"port"` + Pass string `yaml:"pass"` + User string `yaml:"user"` + RequestTimeout time.Duration `yaml:"request_timeout"` + ConnectionPoolCapacity int64 `yaml:"connection_pool_capacity"` + UsePreparedStatements *bool `yaml:"use_prepared_statements,omitempty"` +} + +func (db *DB) ConnString() string { + return fmt.Sprintf("host=%s port=%d dbname=%s user=%s password=%s pool_max_conns=%d", db.Host, db.Port, db.Name, db.User, db.Pass, db.ConnectionPoolCapacity) +} + +type ( + RateLimiter struct { + RatePerSec int `yaml:"rate_per_sec"` + MaxBurst int `yaml:"max_burst"` + StoreMaxKeys int `yaml:"store_max_keys"` + PerHandler bool `yaml:"per_handler"` + } + + UserToRateLimiter map[string]RateLimiter + + ApiRateLimiters struct { + Default RateLimiter `yaml:"default"` + SpecialUsers UserToRateLimiter `yaml:"spec_users"` + } + + ApiToRateLimiters map[string]ApiRateLimiters +) + +type InmemoryCache struct { + ID string `yaml:"id"` + NumCounters int64 `yaml:"num_counters"` + MaxCost int64 `yaml:"max_cost"` + BufferItems int64 `yaml:"buffer_items"` +} + +type Redis struct { + ID string `yaml:"id"` + WithInmemID string `yaml:"with_inmem_id"` + Addr string `yaml:"addr"` + Username string `yaml:"username"` + Password string `yaml:"password"` + Timeout time.Duration `yaml:"timeout"` + MaxRetries int `yaml:"max_retries"` + MinRetryBackoff time.Duration `yaml:"min_retry_backoff"` + MaxRetryBackoff time.Duration `yaml:"max_retry_backoff"` +} + +type Cache struct { + Inmemory []InmemoryCache `yaml:"inmemory"` + Redis []Redis `yaml:"redis"` +} + +type S3 struct { + Endpoint string `yaml:"endpoint"` + AccessKeyID string `yaml:"access_key_id"` + SecretAccessKey string `yaml:"secret_access_key"` + BucketName string `yaml:"bucket_name"` + EnableSSl bool `yaml:"enable_ssl"` +} + +type DownloadParams struct { + Delay time.Duration `yaml:"delay"` + InitialRetryBackoff time.Duration `yaml:"initial_retry_backoff"` + MaxRetryBackoff time.Duration `yaml:"max_retry_backoff"` +} + +type SessionStore struct { + RedisID string `yaml:"redis_id"` + ExportLifetime time.Duration `yaml:"export_lifetime"` +} + +type FileStore struct { + S3 *S3 `yaml:"s3"` +} + +type MassExport struct { + BatchSize uint64 `yaml:"batch_size"` + WorkersCount int `yaml:"workers_count"` + TasksChannelSize int `yaml:"tasks_channel_size"` + PartLength time.Duration `yaml:"part_length"` + URLPrefix string `yaml:"url_prefix"` + AllowedUsers []string `yaml:"allowed_users"` + FileStore *FileStore `yaml:"file_store"` + SessionStore *SessionStore `yaml:"session_store"` + DownloadParams *DownloadParams `yaml:"download_params"` +} + +type HTTP struct { + Addr string `yaml:"addr"` + ReadHeaderTimeout time.Duration `yaml:"read_header_timeout"` + ReadTimeout time.Duration `yaml:"read_timeout"` + WriteTimeout time.Duration `yaml:"write_timeout"` + CORS *CORS `yaml:"cors"` +} + +type GRPC struct { + Addr string `yaml:"addr"` + ConnectionTimeout time.Duration `yaml:"connection_timeout"` +} + +type Debug struct { + Addr string `yaml:"addr"` +} + +type JWT struct { + SecretKey string `yaml:"secret_key"` +} + +type Auth struct { + OIDC *OIDC `yaml:"oidc"` + JWT *JWT `yaml:"jwt"` +} + +type Server struct { + HTTP HTTP `yaml:"http"` + GRPC GRPC `yaml:"grpc"` + Debug Debug `yaml:"debug"` + Auth *Auth `yaml:"auth"` + RateLimiters ApiToRateLimiters `yaml:"rate_limiters"` +} + +type GRPCKeepaliveParams struct { + // After a duration of this time if the client doesn't see any activity it + // pings the server to see if the transport is still alive. + // If set below 10s, a minimum value of 10s will be used instead. + Time time.Duration `yaml:"time"` + // After having pinged for keepalive check, the client waits for a duration + // of Timeout and if no activity is seen even after that the connection is + // closed. If set below 1s, a minimum value of 1s will be used instead. + Timeout time.Duration `yaml:"timeout"` + // If true, client sends keepalive pings even with no active RPCs. If false, + // when there are no active RPCs, Time and Timeout will be ignored and no + // keepalive pings will be sent. False by default. + PermitWithoutStream bool `yaml:"permit_without_stream"` +} + +type SeqDBClient struct { + ID string `yaml:"id"` + Timeout time.Duration `yaml:"timeout"` + AvgDocSize int `yaml:"avg_doc_size"` + Addrs []string `yaml:"addrs"` + RequestRetries int `yaml:"request_retries"` + InitialRetryBackoff time.Duration `yaml:"initial_retry_backoff"` + MaxRetryBackoff time.Duration `yaml:"max_retry_backoff"` + ClientMode string `yaml:"client_mode"` + GRPCKeepaliveParams *GRPCKeepaliveParams `yaml:"grpc_keepalive_params"` + DownloadParams *DownloadParams `yaml:"download_params"` +} + +type CHClient struct { + ID string `yaml:"id"` + Addrs []string `yaml:"addrs"` + Database string `yaml:"database"` + Username string `yaml:"username"` + Password string `yaml:"password"` + Sharded bool `yaml:"sharded"` + DialTimeout time.Duration `yaml:"dial_timeout"` + ReadTimeout time.Duration `yaml:"read_timeout"` +} + +type Clients struct { + SeqDB []SeqDBClient `yaml:"seq_db"` + ClickHouse []CHClient `yaml:"clickhouse"` +} + +type Handlers struct { + SeqAPI SeqAPI `yaml:"seq_api"` + ErrorGroups ErrorGroups `yaml:"error_groups"` + MassExport *MassExport `yaml:"mass_export"` + AsyncSearch AsyncSearch `yaml:"async_search"` +} + +type Field struct { + Name string `yaml:"name"` + Type string `yaml:"type"` +} + +type SeqAPI struct { + *SeqAPIOptions `yaml:",inline"` + Envs map[string]SeqAPIEnv `yaml:"envs"` + DefaultEnv string `yaml:"default_env"` +} + +type SeqAPIEnv struct { + SeqDB string `yaml:"seq_db_id"` + Options *SeqAPIOptions `yaml:"options"` +} + +type SeqAPIOptions struct { + MaxSearchLimit int32 `yaml:"max_search_limit"` + MaxSearchTotalLimit int64 `yaml:"max_search_total_limit"` + MaxSearchOffsetLimit int32 `yaml:"max_search_offset_limit"` + MaxExportLimit int32 `yaml:"max_export_limit"` + SeqCLIMaxSearchLimit int `yaml:"seq_cli_max_search_limit"` + MaxParallelExportRequests int `yaml:"max_parallel_export_requests"` + MaxAggregationsPerRequest int `yaml:"max_aggregations_per_request"` + MaxBucketsPerAggregationTs int `yaml:"max_buckets_per_aggregation_ts"` + EventsCacheTTL time.Duration `yaml:"events_cache_ttl"` + PinnedFields []Field `yaml:"pinned_fields"` + SystemFields []Field `yaml:"system_fields"` + LogsLifespanCacheKey string `yaml:"logs_lifespan_cache_key"` + LogsLifespanCacheTTL time.Duration `yaml:"logs_lifespan_cache_ttl"` + FieldsCacheTTL time.Duration `yaml:"fields_cache_ttl"` + Masking *Masking `yaml:"masking"` +} + +type Masking struct { + Masks []Mask `yaml:"masks"` + ProcessFields []string `yaml:"process_fields"` + IgnoreFields []string `yaml:"ignore_fields"` +} + +type Mask struct { + Re string `yaml:"re"` + Groups []int `yaml:"groups"` + Mode string `yaml:"mode"` // "mask" or "replace" or "cut" + ReplaceWord string `yaml:"replace_word"` // for mode:replace + + ProcessFields []string `yaml:"process_fields"` + IgnoreFields []string `yaml:"ignore_fields"` + + FieldFilters *FieldFilterSet `yaml:"field_filters"` +} + +type FieldFilter struct { + Field string `yaml:"field"` + Mode string `yaml:"mode"` // "equal" or "contains" or "prefix" or "suffix" + Values []string `yaml:"values"` +} + +type FieldFilterSet struct { + Condition string `yaml:"condition"` // "and" or "or" or "not" + Filters []FieldFilter `yaml:"filters"` // max 1 if condition:not +} + +type LogTagsMapping struct { + Release []string `yaml:"release"` + Service []string `yaml:"service"` + Env []string `yaml:"env"` +} + +type ErrorGroups struct { + LogTagsMapping LogTagsMapping `yaml:"log_tags_mapping"` + QueryFilter map[string]string `yaml:"query_filter"` +} + +type AsyncSearch struct { + AdminUsers []string `yaml:"admin_users"` + ListQueryLengthLimit int `yaml:"list_query_length_limit"` +} + +func Normalize(cfg *Config) error { + if len(cfg.Clients.SeqDB) == 0 { + return fmt.Errorf("clients.seq_db must contain at least one client") + } + + seqDBIDs := make(map[string]struct{}, len(cfg.Clients.SeqDB)) + for i := range cfg.Clients.SeqDB { + c := &cfg.Clients.SeqDB[i] + if c.ID == "" { + return fmt.Errorf("seq_db client ID cannot be empty") + } + if _, ok := seqDBIDs[c.ID]; ok { + return fmt.Errorf("duplicate seq_db client ID: %s", c.ID) + } + + seqDBIDs[c.ID] = struct{}{} + + if c.ClientMode == "" { + c.ClientMode = ProxyClientModeGRPC + } else if c.ClientMode != ProxyClientModeGRPC { + return fmt.Errorf("invalid clients.seq_db[%s].client_mode: %q (allowed: %q)", c.ID, c.ClientMode, ProxyClientModeGRPC) + } + + if c.GRPCKeepaliveParams != nil { + if c.GRPCKeepaliveParams.Time < minGRPCKeepaliveTime { + c.GRPCKeepaliveParams.Time = minGRPCKeepaliveTime + } + if c.GRPCKeepaliveParams.Timeout < minGRPCKeepaliveTimeout { + c.GRPCKeepaliveParams.Timeout = minGRPCKeepaliveTimeout + } + } + } + + chIDs := make(map[string]struct{}, len(cfg.Clients.ClickHouse)) + for i := range cfg.Clients.ClickHouse { + ch := &cfg.Clients.ClickHouse[i] + if ch.ID == "" { + return fmt.Errorf("clickhouse client ID cannot be empty") + } + if _, ok := chIDs[ch.ID]; ok { + return fmt.Errorf("duplicate clickhouse client ID: %s", ch.ID) + } + + chIDs[ch.ID] = struct{}{} + + if ch.DialTimeout <= 0 { + ch.DialTimeout = defaultClickHouseDialTimeout + } + if ch.ReadTimeout <= 0 { + ch.ReadTimeout = defaultClickHouseReadTimeout + } + } + + inmemIDs := make(map[string]struct{}, len(cfg.Cache.Inmemory)) + for i := range cfg.Cache.Inmemory { + inm := &cfg.Cache.Inmemory[i] + if inm.ID == "" { + return fmt.Errorf("inmemory cache ID cannot be empty") + } + if _, ok := inmemIDs[inm.ID]; ok { + return fmt.Errorf("duplicate inmemory cache ID: %s", inm.ID) + } + + inmemIDs[inm.ID] = struct{}{} + + if inm.NumCounters <= 0 { + inm.NumCounters = defaultInmemCacheNumCounters + } + if inm.MaxCost <= 0 { + inm.MaxCost = defaultInmemCacheMaxCost + } + if inm.BufferItems <= 0 { + inm.BufferItems = defaultInmemCacheBufferItems + } + } + + redisIDs := make(map[string]struct{}, len(cfg.Cache.Redis)) + for i := range cfg.Cache.Redis { + r := &cfg.Cache.Redis[i] + if r.ID == "" { + return fmt.Errorf("redis cache ID cannot be empty") + } + if _, ok := inmemIDs[r.ID]; ok { + return fmt.Errorf("duplicate redis cache ID: %s", r.ID) + } + + redisIDs[r.ID] = struct{}{} + + if r.WithInmemID != "" { + if _, ok := redisIDs[r.WithInmemID]; !ok { + return fmt.Errorf("redis cache %q references unknown inmem cache id %q", r.ID, r.WithInmemID) + } + } + } + + if cfg.DB != nil && cfg.DB.UsePreparedStatements == nil { + cfg.DB.UsePreparedStatements = new(bool) + *cfg.DB.UsePreparedStatements = true + } + + if cfg.Handlers.AsyncSearch.ListQueryLengthLimit <= 0 { + cfg.Handlers.AsyncSearch.ListQueryLengthLimit = defaultAsyncSearchListQueryLengthLimit + } + + setSeqAPIOptionsDefaults(cfg.Handlers.SeqAPI.SeqAPIOptions) + + if len(cfg.Handlers.SeqAPI.Envs) > 0 { + if cfg.Handlers.SeqAPI.DefaultEnv == "" { + return fmt.Errorf("default_env must be specified when using envs") + } + + if _, exists := cfg.Handlers.SeqAPI.Envs[cfg.Handlers.SeqAPI.DefaultEnv]; !exists { + return fmt.Errorf("default_env '%s' not found in seq_api.envs", cfg.Handlers.SeqAPI.DefaultEnv) + } + + for envName, envConfig := range cfg.Handlers.SeqAPI.Envs { + if _, ok := seqDBIDs[envConfig.SeqDB]; !ok { + return fmt.Errorf("client '%s' for env '%s' not found", envConfig.SeqDB, envName) + } + + if envConfig.Options == nil { + envConfig.Options = cfg.Handlers.SeqAPI.SeqAPIOptions + } else { + setSeqAPIOptionsDefaults(envConfig.Options) + } + + cfg.Handlers.SeqAPI.Envs[envName] = envConfig + } + } + + if cfg.Handlers.MassExport != nil { + if cfg.Handlers.MassExport.SessionStore.RedisID == "" { + return fmt.Errorf("handlers.mass_export.session_store.redis_id cannot be empty") + } + + if _, ok := redisIDs[cfg.Handlers.MassExport.SessionStore.RedisID]; !ok { + return fmt.Errorf("unknown handlers.mass_export.session_store.redis_id %q", cfg.Handlers.MassExport.SessionStore.RedisID) + } + } + return nil +} + +func setSeqAPIOptionsDefaults(options *SeqAPIOptions) { + if options.MaxAggregationsPerRequest <= 0 { + options.MaxAggregationsPerRequest = defaultMaxAggregationsPerRequest + } + if options.MaxBucketsPerAggregationTs <= 0 { + options.MaxBucketsPerAggregationTs = defaultMaxBucketsPerAggregationTs + } + if options.MaxParallelExportRequests <= 0 { + options.MaxParallelExportRequests = defaultMaxParallelExportRequests + } + if options.MaxSearchTotalLimit <= 0 { + options.MaxSearchTotalLimit = defaultMaxSearchTotalLimit + } + if options.MaxSearchOffsetLimit <= 0 { + options.MaxSearchOffsetLimit = defaultMaxSearchOffsetLimit + } + if options.MaxExportLimit <= 0 { + options.MaxExportLimit = defaultMaxExportLimit + } + if options.EventsCacheTTL <= 0 { + options.EventsCacheTTL = defaultEventsCacheTTL + } + if options.LogsLifespanCacheKey == "" { + options.LogsLifespanCacheKey = defaultLogsLifespanCacheKey + } + if options.LogsLifespanCacheTTL <= 0 { + options.LogsLifespanCacheTTL = defaultLogsLifespanCacheTTL + } +}