Сonfig v2

Makefile

@@ -4,6 +4,7 @@ TIME := $(shell date '+%Y-%m-%d_%H:%M:%S')
 
 MIGRATION_DSN ?= postgresql://localhost/postgres?sslmode=disable&user=postgres&password=postgres
 MIGRATION_DSN_CLICKHOUSE ?= tcp://default@localhost:9000/seq_ui_server
+SOURCE_CONFIG ?= config/config.local.yaml
 
 LOCAL_BIN := $(CURDIR)/bin
 
@@ -58,6 +59,10 @@ push-migration-image: build-migration-image
 run: .check-config
 	go run ./cmd/seq-ui -config=config/config.local.yaml
 
+.PHONY: config-migrate
+config-migrate:
+	go run ./cmd/config-migrate -source=${SOURCE_CONFIG}
+
 .PHONY: .check-config
 .check-config:
 	@if [ ! -f config/config.local.yaml ]; then\

cmd/config-migrate/main.go

@@ -0,0 +1,86 @@
+package main
+
+import (
+	"bytes"
+	"flag"
+	"fmt"
+	"os"
+	"path/filepath"
+	"strings"
+
+	"go.uber.org/zap"
+	"gopkg.in/yaml.v3"
+
+	"github.com/ozontech/seq-ui/internal/app/config/loader"
+	"github.com/ozontech/seq-ui/internal/app/config/migrate"
+	v1 "github.com/ozontech/seq-ui/internal/app/config/v1"
+	"github.com/ozontech/seq-ui/logger"
+)
+
+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"))
+	}
+
+	stat, err := os.Stat(source)
+	if err != nil {
+		logger.Fatal("stat source", zap.String("source", source), zap.Error(err))
+	}
+
+	sourceCfg, err := os.ReadFile(source) //nolint:gosec
+	if err != nil {
+		logger.Fatal("error reading source config", zap.String("source", source), zap.Error(err))
+	}
+
+	version, err := loader.ReadVersion(sourceCfg)
+	if err != nil {
+		logger.Fatal("read config version", zap.Error(err))
+	}
+	if version != loader.V1 {
+		errStr := fmt.Sprintf("source config is not v%d, nothing to migrate", loader.V1)
+		logger.Fatal(errStr, zap.Int("version", version))
+	}
+
+	backupPath := strings.TrimSuffix(source, filepath.Ext(source)) + ".bck" + filepath.Ext(source)
+	if _, err := os.Stat(backupPath); err == nil {
+		logger.Fatal("backup already exists, remove it before re-running", zap.String("backup", backupPath))
+	}
+	if err := os.WriteFile(backupPath, sourceCfg, stat.Mode()); err != nil {
+		logger.Fatal("error writing backup legacy config", zap.String("backup", backupPath), zap.Error(err))
+	}
+
+	var cfg v1.Config
+	decoder := yaml.NewDecoder(bytes.NewReader(sourceCfg))
+	decoder.KnownFields(true)
+	if err := decoder.Decode(&cfg); err != nil {
+		logger.Fatal("error parsing config v1", zap.Error(err))
+	}
+
+	migratedCfg := migrate.V1ToV2(cfg)
+
+	var buf bytes.Buffer
+	encoder := yaml.NewEncoder(&buf)
+	encoder.SetIndent(2)
+	if err := encoder.Encode(&migratedCfg); err != nil {
+		logger.Fatal("error encoding config v2", zap.Error(err))
+	}
+	if err := encoder.Close(); err != nil {
+		logger.Fatal("close encoder", zap.Error(err))
+	}
+
+	if err := os.WriteFile(source, buf.Bytes(), stat.Mode()); err != nil {
+		logger.Fatal("write migrated config", zap.Error(err))
+	}
+
+	logger.Info("config migrated", zap.String("source", source), zap.String("backup", backupPath))
+}

cmd/seq-ui/main.go

@@ -23,7 +23,8 @@ 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"
+	"github.com/ozontech/seq-ui/internal/app/config/loader"
+	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"
@@ -74,7 +75,7 @@ func run(ctx context.Context) {
 		logger.Warn("app uses the default config file, to provide your own config use -config flag")
 	}
 
-	cfg, err := config.FromFile(*configPath)
+	cfg, err := loader.FromFile(*configPath)
 	if err != nil {
 		logger.Fatal("read config file error", zap.Error(err))
 	}
@@ -90,9 +91,20 @@ func run(ctx context.Context) {
 			zap.Float64("sampler_param", tracingCfg.SamplerParam))
 	}
 
-	registrar := initApp(ctx, cfg)
+	logger.Info("initializing caches")
+	caches, err := cache.FromConfig(ctx, cfg.Cache)
+	if err != nil {
+		logger.Fatal("failed to init caches", zap.Error(err))
+	}
+
+	var oidcCache cache.Cache
+	if cfg.Server.Auth != nil && cfg.Server.Auth.OIDC != nil {
+		oidcCache = caches[cfg.Server.Auth.OIDC.CacheID]
+	}
+
+	registrar := initApp(ctx, cfg, caches)
 
-	serv, err := server.New(ctx, cfg.Server, registrar)
+	serv, err := server.New(ctx, cfg.Server, registrar, oidcCache)
 	if err != nil {
 		logger.Fatal("app init error", zap.Error(err))
 	}
@@ -104,93 +116,74 @@ func run(ctx context.Context) {
 	}
 }
 
-func initApp(ctx context.Context, cfg config.Config) *api.Registrar {
+func initApp(ctx context.Context, cfg config.Config, caches map[string]cache.Cache) *api.Registrar {
 	logger.Info("initializing seq-db clients")
 	seqDBClients, err := initSeqDBClients(ctx, cfg)
 	if err != nil {
-		logger.Fatal("failed to init seq-db client", zap.Error(err))
+		logger.Fatal("failed to init seq-db clients", zap.Error(err))
 	}
 
-	defaultClientID := config.DefaultSeqDBClientID
-	if len(cfg.Handlers.SeqAPI.Envs) > 0 {
-		defaultClientID = cfg.Handlers.SeqAPI.Envs[cfg.Handlers.SeqAPI.DefaultEnv].SeqDB
+	logger.Info("initializing clickhouse clients")
+	chClients, err := initClickHouseClients(ctx, cfg)
+	if err != nil {
+		logger.Fatal("failed to init clickhouse clients", zap.Error(err))
 	}
 
-	defaultClient, exists := seqDBClients[defaultClientID]
-	if !exists {
-		logger.Fatal("default seq-db client not found",
-			zap.String("clientID", defaultClientID),
-		)
+	logger.Info("initializing db")
+	db, err := initDb(ctx, cfg.DB)
+	if err != nil {
+		logger.Fatal("failed to init db", zap.Error(err))
 	}
 
 	var massExportV1 *massexport_v1.MassExport
 	if cfg.Handlers.MassExport != nil {
-		exportServer, err := initExportService(ctx, *cfg.Handlers.MassExport, defaultClient)
+		me := cfg.Handlers.MassExport
+		redisCfg := cfg.Cache.RedisByID(me.SessionStore.RedisID)
+
+		exportServer, err := initExportService(ctx, *me, redisCfg, seqDBClients[cfg.Handlers.MassExport.SeqDBID])
 		if err != nil {
 			logger.Fatal("can't init export server", zap.Error(err))
 		}
-
 		massExportV1 = massexport_v1.New(exportServer)
 	}
 
-	logger.Info("initializing inmemory with redis seqapi cache")
-	inmemWithRedisCache, err := cache.NewInmemoryWithRedisOrInmemory(ctx, cfg.Server.Cache)
-	if err != nil {
-		logger.Fatal("failed to init inmemory with redis seqapi cache", zap.Error(err))
-	}
-
-	logger.Info("initializing redis seqapi cache")
-	redisCache, err := cache.NewRedisOrInmemory(ctx, cfg.Server.Cache)
-	if err != nil {
-		logger.Fatal("failed to init redis seqapi cache", zap.Error(err))
-	}
-
-	logger.Info("initializing db")
-	db, err := initDb(ctx, cfg.Server.DB)
-	if err != nil {
-		logger.Fatal("failed to init db", zap.Error(err))
-	}
-
 	var (
 		asyncSearchesService asyncsearches.Service
 		userProfileV1        *userprofile_v1.UserProfile
 		dashboardsV1         *dashboards_v1.Dashboards
 	)
 	if db != nil {
-		repo := repository.New(db, cfg.Server.DB.RequestTimeout)
+		repo := repository.New(db, cfg.DB.RequestTimeout)
 		userProfilesSvc := userprofile.New(repo.UserProfiles, repo.FavoriteQueries)
 		dashboardsSvc := dashboards.New(repo.Dashboards)
 		profiles.InitProfiles(repo.UserProfiles.GetOrCreate)
 
 		userProfileV1 = userprofile_v1.New(userProfilesSvc)
 		dashboardsV1 = dashboards_v1.New(dashboardsSvc)
 
-		asyncSearchesService = asyncsearches.New(ctx, repo, defaultClient, cfg.Handlers.AsyncSearch)
+		asyncSearchesService = asyncsearches.New(ctx, repo, seqDBClients[cfg.Handlers.AsyncSearch.SeqDBID], cfg.Handlers.AsyncSearch)
 	}
 
-	seqApiV1 := seqapi_v1.New(cfg.Handlers.SeqAPI, seqDBClients, inmemWithRedisCache, redisCache, asyncSearchesService)
-
-	logger.Info("initializing clickhouse")
-	ch, err := initClickHouse(ctx, cfg.Server.CH)
-	if err != nil {
-		logger.Fatal("failed to init clickhouse", zap.Error(err))
-	}
+	seqApiCache := caches[cfg.Handlers.SeqAPI.CacheID]
+	seqApiRedisCache := caches[cfg.Handlers.SeqAPI.RedisID]
+	seqApiV1 := seqapi_v1.New(&cfg.Handlers.SeqAPI, seqDBClients, seqApiCache, seqApiRedisCache, asyncSearchesService)
 
 	var errorGroupsV1 *errorgroups_v1.ErrorGroups
-	if ch != nil {
-		repo := repositorych.New(ch, cfg.Server.CH.Sharded, cfg.Handlers.ErrorGroups.QueryFilter)
+	if cfg.Handlers.ErrorGroups != nil {
+		chCfg := cfg.Clients.ClickHouseByID(cfg.Handlers.ErrorGroups.CHID)
+		repo := repositorych.New(chClients[cfg.Handlers.ErrorGroups.CHID], chCfg.Sharded, cfg.Handlers.ErrorGroups.QueryFilter)
 		svc := errorgroups.New(repo, cfg.Handlers.ErrorGroups.LogTagsMapping)
 
-		errorGroupsV1 = errorgroups_v1.New(svc)
+		errorGroupsV1 = errorgroups_v1.New(svc, chClients)
 	}
 
 	return api.NewRegistrar(seqApiV1, userProfileV1, dashboardsV1, massExportV1, errorGroupsV1)
 }
 
 func initSeqDBClients(ctx context.Context, cfg config.Config) (map[string]seqdb.Client, error) {
-	clients := make(map[string]seqdb.Client)
+	clients := make(map[string]seqdb.Client, len(cfg.Clients.SeqDB))
 	for _, clientCfg := range cfg.Clients.SeqDB {
-		client, err := createSeqBDClient(ctx, clientCfg, cfg.Handlers.SeqAPI)
+		client, err := createSeqBDClient(ctx, &clientCfg, &cfg.Handlers.SeqAPI)
 		if err != nil {
 			return nil, fmt.Errorf("failed to create seq_db client %s: %w", clientCfg.ID, err)
 		}
@@ -200,12 +193,20 @@ func initSeqDBClients(ctx context.Context, cfg config.Config) (map[string]seqdb.
 	return clients, nil
 }
 
-func createSeqBDClient(ctx context.Context, cfg config.SeqDBClient, seqAPI config.SeqAPI) (seqdb.Client, error) {
-	clientMaxRecvMsgSize := cfg.AvgDocSize * 1024 * int(seqAPI.MaxSearchLimit)
+func createSeqBDClient(ctx context.Context, cfg *config.SeqDBClient, seqAPI *config.SeqAPI) (seqdb.Client, error) {
+	maxSearchLimit := int(seqAPI.Options.MaxSearchLimit)
+	for _, env := range seqAPI.Envs {
+		if env.SeqDBID != cfg.ID {
+			continue
+		}
+		if l := int(env.Options.MaxSearchLimit); l > maxSearchLimit {
+			maxSearchLimit = l
+		}
+	}
+	clientMaxRecvMsgSize := cfg.AvgDocSize * 1024 * maxSearchLimit
 	if clientMaxRecvMsgSize < defaultClientMaxRecvMsgSize {
 		clientMaxRecvMsgSize = defaultClientMaxRecvMsgSize
 	}
-
 	clientParams := seqdb.ClientParams{
 		Addrs:               cfg.Addrs,
 		Timeout:             cfg.Timeout,
@@ -224,18 +225,15 @@ 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) {
 	if cfg == nil {
 		logger.Warn("db config is nil, running without db")
 		return nil, nil
 	}
-
 	pgxCfg, err := pgxpool.ParseConfig(cfg.ConnString())
 	if err != nil {
 		return nil, fmt.Errorf("can't parse connection string: %w", err)
 	}
-
 	if !*cfg.UsePreparedStatements {
 		// By default, pgx uses the QueryExecModeCacheStatement and automatically prepares and caches prepared statements.
 		// However, this may be incompatible with proxies such as PGBouncer.
@@ -251,8 +249,8 @@ func initDb(ctx context.Context, cfg *config.DB) (*pgxpool.Pool, error) {
 	return pool, nil
 }
 
-func initExportService(ctx context.Context, cfg config.MassExport, client seqdb.Client) (massexport.Service, error) {
-	sessionStore, err := sessionstore.NewRedisSessionStore(ctx, cfg.SessionStore)
+func initExportService(ctx context.Context, cfg config.MassExport, redisCfg *config.Redis, client seqdb.Client) (massexport.Service, error) {
+	sessionStore, err := sessionstore.NewRedisSessionStore(ctx, redisCfg, cfg.SessionStore.ExportLifetime)
 	if err != nil {
 		return nil, fmt.Errorf("init session store: %w", err)
 	}
@@ -267,12 +265,20 @@ func initExportService(ctx context.Context, cfg config.MassExport, client seqdb.
 	return massexport.NewService(ctx, cfg, sessionStore, fileStore, client)
 }
 
-func initClickHouse(ctx context.Context, cfg *config.CH) (driver.Conn, error) {
-	if cfg == nil {
-		logger.Warn("clickhouse config is nil, running without clickhouse")
-		return nil, nil
+func initClickHouseClients(ctx context.Context, cfg config.Config) (map[string]driver.Conn, error) {
+	clients := make(map[string]driver.Conn, len(cfg.Clients.ClickHouse))
+	for _, clientCfg := range cfg.Clients.ClickHouse {
+		client, err := createClickHouseClient(ctx, &clientCfg)
+		if err != nil {
+			return nil, fmt.Errorf("failed to create clickhouse client %s: %w", clientCfg.ID, err)
+		}
+		clients[clientCfg.ID] = client
 	}
 
+	return clients, nil
+}
+
+func createClickHouseClient(ctx context.Context, cfg *config.CHClient) (driver.Conn, error) {
 	conn, err := clickhouse.Open(&clickhouse.Options{
 		Addr: cfg.Addrs,
 		Auth: clickhouse.Auth{

docker-compose.yml

@@ -1,4 +1,4 @@
-version: '3.8'
+version: "3.8"
 
 services:
   postgres:

internal/api/errorgroups/v1/errorgroups.go

@@ -1,6 +1,7 @@
 package errorgroups_v1
 
 import (
+	"github.com/ClickHouse/clickhouse-go/v2/lib/driver"
 	"github.com/go-chi/chi/v5"
 
 	grpc_api "github.com/ozontech/seq-ui/internal/api/errorgroups/v1/grpc"
@@ -13,7 +14,7 @@ type ErrorGroups struct {
 	httpAPI *http_api.API
 }
 
-func New(svc errorgroups.Service) *ErrorGroups {
+func New(svc errorgroups.Service, chClients map[string]driver.Conn) *ErrorGroups {
 	return &ErrorGroups{
 		grpcAPI: grpc_api.New(svc),
 		httpAPI: http_api.New(svc),

internal/api/seqapi/v1/grpc/aggregation.go

@@ -80,7 +80,7 @@ func (a *API) GetAggregation(ctx context.Context, req *seqapi.GetAggregationRequ
 		return nil, err
 	}
 
-	if params.masker != nil {
+	if a.globalParams.masker != nil {
 		buf := make([]string, 0)
 		for i, agg := range resp.Aggregations {
 			if agg == nil {
@@ -98,7 +98,7 @@ func (a *API) GetAggregation(ctx context.Context, req *seqapi.GetAggregationRequ
 				field = aggReq.GroupBy
 			}
 
-			buf = params.masker.MaskAgg(field, buf)
+			buf = a.globalParams.masker.MaskAgg(field, buf)
 
 			for j, key := range buf {
 				if agg.Buckets[j] != nil {