diff --git a/ee/query-service/app/server.go b/ee/query-service/app/server.go index c4df57db89..48f7837785 100644 --- a/ee/query-service/app/server.go +++ b/ee/query-service/app/server.go @@ -111,25 +111,22 @@ func (s Server) HealthCheckStatus() chan healthcheck.Status { // NewServer creates and initializes Server func NewServer(serverOptions *ServerOptions) (*Server, error) { - - modelDao, err := dao.InitDao("sqlite", baseconst.RELATIONAL_DATASOURCE_PATH) + modelDao, err := dao.InitDao(serverOptions.SigNoz.SQLStore.SQLxDB()) if err != nil { return nil, err } - baseexplorer.InitWithDSN(baseconst.RELATIONAL_DATASOURCE_PATH) - - if err := preferences.InitDB(baseconst.RELATIONAL_DATASOURCE_PATH); err != nil { + if err := baseexplorer.InitWithDSN(serverOptions.SigNoz.SQLStore.SQLxDB()); err != nil { return nil, err } - localDB, err := dashboards.InitDB(baseconst.RELATIONAL_DATASOURCE_PATH) - - if err != nil { + if err := preferences.InitDB(serverOptions.SigNoz.SQLStore.SQLxDB()); err != nil { return nil, err } - localDB.SetMaxOpenConns(10) + if err := dashboards.InitDB(serverOptions.SigNoz.SQLStore.SQLxDB()); err != nil { + return nil, err + } gatewayProxy, err := gateway.NewProxy(serverOptions.GatewayUrl, gateway.RoutePrefix) if err != nil { @@ -137,7 +134,7 @@ func NewServer(serverOptions *ServerOptions) (*Server, error) { } // initiate license manager - lm, err := licensepkg.StartManager("sqlite", localDB) + lm, err := licensepkg.StartManager(serverOptions.SigNoz.SQLStore.SQLxDB()) if err != nil { return nil, err } @@ -151,7 +148,7 @@ func NewServer(serverOptions *ServerOptions) (*Server, error) { if storage == "clickhouse" { zap.L().Info("Using ClickHouse as datastore ...") qb := db.NewDataConnector( - localDB, + serverOptions.SigNoz.SQLStore.SQLxDB(), serverOptions.PromConfigPath, lm, serverOptions.MaxIdleConns, @@ -187,7 +184,7 @@ func NewServer(serverOptions *ServerOptions) (*Server, error) { rm, err := makeRulesManager(serverOptions.PromConfigPath, baseconst.GetAlertManagerApiPrefix(), serverOptions.RuleRepoURL, - localDB, + serverOptions.SigNoz.SQLStore.SQLxDB(), reader, c, serverOptions.DisableRules, @@ -201,19 +198,19 @@ func NewServer(serverOptions *ServerOptions) (*Server, error) { } // initiate opamp - _, err = opAmpModel.InitDB(localDB) + _, err = opAmpModel.InitDB(serverOptions.SigNoz.SQLStore.SQLxDB()) if err != nil { return nil, err } - integrationsController, err := integrations.NewController(localDB) + integrationsController, err := integrations.NewController(serverOptions.SigNoz.SQLStore.SQLxDB()) if err != nil { return nil, fmt.Errorf( "couldn't create integrations controller: %w", err, ) } - cloudIntegrationsController, err := cloudintegrations.NewController(localDB) + cloudIntegrationsController, err := cloudintegrations.NewController(serverOptions.SigNoz.SQLStore.SQLxDB()) if err != nil { return nil, fmt.Errorf( "couldn't create cloud provider integrations controller: %w", err, @@ -222,7 +219,7 @@ func NewServer(serverOptions *ServerOptions) (*Server, error) { // ingestion pipelines manager logParsingPipelineController, err := logparsingpipeline.NewLogParsingPipelinesController( - localDB, "sqlite", integrationsController.GetPipelinesForInstalledIntegrations, + serverOptions.SigNoz.SQLStore.SQLxDB(), integrationsController.GetPipelinesForInstalledIntegrations, ) if err != nil { return nil, err @@ -230,8 +227,7 @@ func NewServer(serverOptions *ServerOptions) (*Server, error) { // initiate agent config handler agentConfMgr, err := agentConf.Initiate(&agentConf.ManagerOptions{ - DB: localDB, - DBEngine: AppDbEngine, + DB: serverOptions.SigNoz.SQLStore.SQLxDB(), AgentFeatures: []agentConf.AgentFeature{logParsingPipelineController}, }) if err != nil { @@ -239,7 +235,7 @@ func NewServer(serverOptions *ServerOptions) (*Server, error) { } // start the usagemanager - usageManager, err := usage.New("sqlite", modelDao, lm.GetRepo(), reader.GetConn()) + usageManager, err := usage.New(modelDao, lm.GetRepo(), reader.GetConn()) if err != nil { return nil, err } diff --git a/ee/query-service/dao/factory.go b/ee/query-service/dao/factory.go index f623e17783..2db8404bd1 100644 --- a/ee/query-service/dao/factory.go +++ b/ee/query-service/dao/factory.go @@ -1,18 +1,10 @@ package dao import ( - "fmt" - + "github.com/jmoiron/sqlx" "go.signoz.io/signoz/ee/query-service/dao/sqlite" ) -func InitDao(engine, path string) (ModelDao, error) { - - switch engine { - case "sqlite": - return sqlite.InitDB(path) - default: - return nil, fmt.Errorf("qsdb type: %s is not supported in query service", engine) - } - +func InitDao(inputDB *sqlx.DB) (ModelDao, error) { + return sqlite.InitDB(inputDB) } diff --git a/ee/query-service/dao/sqlite/modelDao.go b/ee/query-service/dao/sqlite/modelDao.go index 02b4367da0..64875bc5fe 100644 --- a/ee/query-service/dao/sqlite/modelDao.go +++ b/ee/query-service/dao/sqlite/modelDao.go @@ -65,8 +65,8 @@ func columnExists(db *sqlx.DB, tableName, columnName string) bool { } // InitDB creates and extends base model DB repository -func InitDB(dataSourceName string) (*modelDao, error) { - dao, err := basedsql.InitDB(dataSourceName) +func InitDB(inputDB *sqlx.DB) (*modelDao, error) { + dao, err := basedsql.InitDB(inputDB) if err != nil { return nil, err } diff --git a/ee/query-service/license/db.go b/ee/query-service/license/db.go index 1dba4053d7..b663a5a2ee 100644 --- a/ee/query-service/license/db.go +++ b/ee/query-service/license/db.go @@ -28,13 +28,8 @@ func NewLicenseRepo(db *sqlx.DB) Repo { } } -func (r *Repo) InitDB(engine string) error { - switch engine { - case "sqlite3", "sqlite": - return sqlite.InitDB(r.db) - default: - return fmt.Errorf("unsupported db") - } +func (r *Repo) InitDB(inputDB *sqlx.DB) error { + return sqlite.InitDB(inputDB) } func (r *Repo) GetLicenses(ctx context.Context) ([]model.License, error) { diff --git a/ee/query-service/license/manager.go b/ee/query-service/license/manager.go index c036a01ab5..ccdfa2b651 100644 --- a/ee/query-service/license/manager.go +++ b/ee/query-service/license/manager.go @@ -51,13 +51,13 @@ type Manager struct { activeFeatures basemodel.FeatureSet } -func StartManager(dbType string, db *sqlx.DB, features ...basemodel.Feature) (*Manager, error) { +func StartManager(db *sqlx.DB, features ...basemodel.Feature) (*Manager, error) { if LM != nil { return LM, nil } repo := NewLicenseRepo(db) - err := repo.InitDB(dbType) + err := repo.InitDB(db) if err != nil { return nil, fmt.Errorf("failed to initiate license repo: %v", err) diff --git a/ee/query-service/main.go b/ee/query-service/main.go index b7f23f43b3..0d5e7dd29e 100644 --- a/ee/query-service/main.go +++ b/ee/query-service/main.go @@ -179,7 +179,7 @@ func main() { zap.L().Info("JWT secret key set successfully.") } - if err := migrate.Migrate(baseconst.RELATIONAL_DATASOURCE_PATH); err != nil { + if err := migrate.Migrate(signoz.SQLStore.SQLxDB()); err != nil { zap.L().Error("Failed to migrate", zap.Error(err)) } else { zap.L().Info("Migration successful") diff --git a/ee/query-service/usage/manager.go b/ee/query-service/usage/manager.go index d52d5ad0c2..1014506faa 100644 --- a/ee/query-service/usage/manager.go +++ b/ee/query-service/usage/manager.go @@ -46,7 +46,7 @@ type Manager struct { tenantID string } -func New(dbType string, modelDao dao.ModelDao, licenseRepo *license.Repo, clickhouseConn clickhouse.Conn) (*Manager, error) { +func New(modelDao dao.ModelDao, licenseRepo *license.Repo, clickhouseConn clickhouse.Conn) (*Manager, error) { hostNameRegex := regexp.MustCompile(`tcp://(?P.*):`) hostNameRegexMatches := hostNameRegex.FindStringSubmatch(os.Getenv("ClickHouseUrl")) diff --git a/pkg/query-service/agentConf/db.go b/pkg/query-service/agentConf/db.go index 30ec1caef3..9cae413b4b 100644 --- a/pkg/query-service/agentConf/db.go +++ b/pkg/query-service/agentConf/db.go @@ -19,13 +19,8 @@ type Repo struct { db *sqlx.DB } -func (r *Repo) initDB(engine string) error { - switch engine { - case "sqlite3", "sqlite": - return sqlite.InitDB(r.db) - default: - return fmt.Errorf("unsupported db") - } +func (r *Repo) initDB(inputDB *sqlx.DB) error { + return sqlite.InitDB(inputDB) } func (r *Repo) GetConfigHistory( diff --git a/pkg/query-service/agentConf/manager.go b/pkg/query-service/agentConf/manager.go index 18ad142cab..73c9d27ed7 100644 --- a/pkg/query-service/agentConf/manager.go +++ b/pkg/query-service/agentConf/manager.go @@ -39,8 +39,7 @@ type Manager struct { } type ManagerOptions struct { - DB *sqlx.DB - DBEngine string + DB *sqlx.DB // When acting as opamp.AgentConfigProvider, agent conf recommendations are // applied to the base conf in the order the features have been specified here. @@ -66,7 +65,7 @@ func Initiate(options *ManagerOptions) (*Manager, error) { configSubscribers: map[string]func(){}, } - err := m.initDB(options.DBEngine) + err := m.initDB(options.DB) if err != nil { return nil, errors.Wrap(err, "could not init agentConf db") } diff --git a/pkg/query-service/app/dashboards/model.go b/pkg/query-service/app/dashboards/model.go index b54ba384d3..f52a6d8e8d 100644 --- a/pkg/query-service/app/dashboards/model.go +++ b/pkg/query-service/app/dashboards/model.go @@ -35,14 +35,8 @@ var ( ) // InitDB sets up setting up the connection pool global variable. -func InitDB(dataSourceName string) (*sqlx.DB, error) { - var err error - - db, err = sqlx.Open("sqlite3", dataSourceName) - if err != nil { - return nil, err - } - +func InitDB(inputDB *sqlx.DB) error { + db = inputDB table_schema := `CREATE TABLE IF NOT EXISTS dashboards ( id INTEGER PRIMARY KEY AUTOINCREMENT, uuid TEXT NOT NULL UNIQUE, @@ -51,9 +45,9 @@ func InitDB(dataSourceName string) (*sqlx.DB, error) { data TEXT NOT NULL );` - _, err = db.Exec(table_schema) + _, err := db.Exec(table_schema) if err != nil { - return nil, fmt.Errorf("error in creating dashboard table: %s", err.Error()) + return fmt.Errorf("error in creating dashboard table: %s", err.Error()) } table_schema = `CREATE TABLE IF NOT EXISTS rules ( @@ -65,7 +59,7 @@ func InitDB(dataSourceName string) (*sqlx.DB, error) { _, err = db.Exec(table_schema) if err != nil { - return nil, fmt.Errorf("error in creating rules table: %s", err.Error()) + return fmt.Errorf("error in creating rules table: %s", err.Error()) } table_schema = `CREATE TABLE IF NOT EXISTS notification_channels ( @@ -80,7 +74,7 @@ func InitDB(dataSourceName string) (*sqlx.DB, error) { _, err = db.Exec(table_schema) if err != nil { - return nil, fmt.Errorf("error in creating notification_channles table: %s", err.Error()) + return fmt.Errorf("error in creating notification_channles table: %s", err.Error()) } tableSchema := `CREATE TABLE IF NOT EXISTS planned_maintenance ( @@ -96,7 +90,7 @@ func InitDB(dataSourceName string) (*sqlx.DB, error) { );` _, err = db.Exec(tableSchema) if err != nil { - return nil, fmt.Errorf("error in creating planned_maintenance table: %s", err.Error()) + return fmt.Errorf("error in creating planned_maintenance table: %s", err.Error()) } table_schema = `CREATE TABLE IF NOT EXISTS ttl_status ( @@ -112,49 +106,49 @@ func InitDB(dataSourceName string) (*sqlx.DB, error) { _, err = db.Exec(table_schema) if err != nil { - return nil, fmt.Errorf("error in creating ttl_status table: %s", err.Error()) + return fmt.Errorf("error in creating ttl_status table: %s", err.Error()) } // sqlite does not support "IF NOT EXISTS" createdAt := `ALTER TABLE rules ADD COLUMN created_at datetime;` _, err = db.Exec(createdAt) if err != nil && !strings.Contains(err.Error(), "duplicate column name") { - return nil, fmt.Errorf("error in adding column created_at to rules table: %s", err.Error()) + return fmt.Errorf("error in adding column created_at to rules table: %s", err.Error()) } createdBy := `ALTER TABLE rules ADD COLUMN created_by TEXT;` _, err = db.Exec(createdBy) if err != nil && !strings.Contains(err.Error(), "duplicate column name") { - return nil, fmt.Errorf("error in adding column created_by to rules table: %s", err.Error()) + return fmt.Errorf("error in adding column created_by to rules table: %s", err.Error()) } updatedBy := `ALTER TABLE rules ADD COLUMN updated_by TEXT;` _, err = db.Exec(updatedBy) if err != nil && !strings.Contains(err.Error(), "duplicate column name") { - return nil, fmt.Errorf("error in adding column updated_by to rules table: %s", err.Error()) + return fmt.Errorf("error in adding column updated_by to rules table: %s", err.Error()) } createdBy = `ALTER TABLE dashboards ADD COLUMN created_by TEXT;` _, err = db.Exec(createdBy) if err != nil && !strings.Contains(err.Error(), "duplicate column name") { - return nil, fmt.Errorf("error in adding column created_by to dashboards table: %s", err.Error()) + return fmt.Errorf("error in adding column created_by to dashboards table: %s", err.Error()) } updatedBy = `ALTER TABLE dashboards ADD COLUMN updated_by TEXT;` _, err = db.Exec(updatedBy) if err != nil && !strings.Contains(err.Error(), "duplicate column name") { - return nil, fmt.Errorf("error in adding column updated_by to dashboards table: %s", err.Error()) + return fmt.Errorf("error in adding column updated_by to dashboards table: %s", err.Error()) } locked := `ALTER TABLE dashboards ADD COLUMN locked INTEGER DEFAULT 0;` _, err = db.Exec(locked) if err != nil && !strings.Contains(err.Error(), "duplicate column name") { - return nil, fmt.Errorf("error in adding column locked to dashboards table: %s", err.Error()) + return fmt.Errorf("error in adding column locked to dashboards table: %s", err.Error()) } telemetry.GetInstance().SetDashboardsInfoCallback(GetDashboardsInfo) - return db, nil + return nil } type Dashboard struct { @@ -288,7 +282,7 @@ func GetDashboard(ctx context.Context, uuid string) (*Dashboard, *model.ApiError if err != nil { return nil, &model.ApiError{Typ: model.ErrorNotFound, Err: fmt.Errorf("no dashboard found with uuid: %s", uuid)} } - + return &dashboard, nil } diff --git a/pkg/query-service/app/explorer/db.go b/pkg/query-service/app/explorer/db.go index c53345e65b..158a460802 100644 --- a/pkg/query-service/app/explorer/db.go +++ b/pkg/query-service/app/explorer/db.go @@ -34,13 +34,8 @@ type SavedView struct { } // InitWithDSN sets up setting up the connection pool global variable. -func InitWithDSN(dataSourceName string) (*sqlx.DB, error) { - var err error - - db, err = sqlx.Open("sqlite3", dataSourceName) - if err != nil { - return nil, err - } +func InitWithDSN(inputDB *sqlx.DB) error { + db = inputDB tableSchema := `CREATE TABLE IF NOT EXISTS saved_views ( uuid TEXT PRIMARY KEY, @@ -56,14 +51,14 @@ func InitWithDSN(dataSourceName string) (*sqlx.DB, error) { extra_data TEXT );` - _, err = db.Exec(tableSchema) + _, err := db.Exec(tableSchema) if err != nil { - return nil, fmt.Errorf("error in creating saved views table: %s", err.Error()) + return fmt.Errorf("error in creating saved views table: %s", err.Error()) } telemetry.GetInstance().SetSavedViewsInfoCallback(GetSavedViewsInfo) - return db, nil + return nil } func InitWithDB(sqlDB *sqlx.DB) { diff --git a/pkg/query-service/app/logparsingpipeline/controller.go b/pkg/query-service/app/logparsingpipeline/controller.go index 4929f72f78..9b08632592 100644 --- a/pkg/query-service/app/logparsingpipeline/controller.go +++ b/pkg/query-service/app/logparsingpipeline/controller.go @@ -27,11 +27,10 @@ type LogParsingPipelineController struct { func NewLogParsingPipelinesController( db *sqlx.DB, - engine string, getIntegrationPipelines func(context.Context) ([]Pipeline, *model.ApiError), ) (*LogParsingPipelineController, error) { repo := NewRepo(db) - err := repo.InitDB(engine) + err := repo.InitDB(db) return &LogParsingPipelineController{ Repo: repo, GetIntegrationPipelines: getIntegrationPipelines, diff --git a/pkg/query-service/app/logparsingpipeline/db.go b/pkg/query-service/app/logparsingpipeline/db.go index 318f23a035..54375aa025 100644 --- a/pkg/query-service/app/logparsingpipeline/db.go +++ b/pkg/query-service/app/logparsingpipeline/db.go @@ -29,13 +29,8 @@ func NewRepo(db *sqlx.DB) Repo { } } -func (r *Repo) InitDB(engine string) error { - switch engine { - case "sqlite3", "sqlite": - return sqlite.InitDB(r.db) - default: - return fmt.Errorf("unsupported db") - } +func (r *Repo) InitDB(inputDB *sqlx.DB) error { + return sqlite.InitDB(inputDB) } // insertPipeline stores a given postable pipeline to database diff --git a/pkg/query-service/app/preferences/model.go b/pkg/query-service/app/preferences/model.go index fce34653fb..380487b900 100644 --- a/pkg/query-service/app/preferences/model.go +++ b/pkg/query-service/app/preferences/model.go @@ -203,14 +203,8 @@ type UpdatePreference struct { var db *sqlx.DB -func InitDB(datasourceName string) error { - var err error - db, err = sqlx.Open("sqlite3", datasourceName) - - if err != nil { - return err - } - +func InitDB(inputDB *sqlx.DB) error { + db = inputDB // create the user preference table tableSchema := ` PRAGMA foreign_keys = ON; @@ -225,7 +219,7 @@ func InitDB(datasourceName string) error { ON DELETE CASCADE );` - _, err = db.Exec(tableSchema) + _, err := db.Exec(tableSchema) if err != nil { return fmt.Errorf("error in creating user_preference table: %s", err.Error()) } diff --git a/pkg/query-service/app/server.go b/pkg/query-service/app/server.go index cfddc0e4c3..4d4147f884 100644 --- a/pkg/query-service/app/server.go +++ b/pkg/query-service/app/server.go @@ -99,23 +99,22 @@ func (s Server) HealthCheckStatus() chan healthcheck.Status { // NewServer creates and initializes Server func NewServer(serverOptions *ServerOptions) (*Server, error) { - - if err := dao.InitDao("sqlite", constants.RELATIONAL_DATASOURCE_PATH); err != nil { + var err error + if err := dao.InitDao(serverOptions.SigNoz.SQLStore.SQLxDB()); err != nil { return nil, err } - if err := preferences.InitDB(constants.RELATIONAL_DATASOURCE_PATH); err != nil { + if err := preferences.InitDB(serverOptions.SigNoz.SQLStore.SQLxDB()); err != nil { return nil, err } - localDB, err := dashboards.InitDB(constants.RELATIONAL_DATASOURCE_PATH) - explorer.InitWithDSN(constants.RELATIONAL_DATASOURCE_PATH) - - if err != nil { + if err := dashboards.InitDB(serverOptions.SigNoz.SQLStore.SQLxDB()); err != nil { return nil, err } - localDB.SetMaxOpenConns(10) + if err := explorer.InitWithDSN(serverOptions.SigNoz.SQLStore.SQLxDB()); err != nil { + return nil, err + } // initiate feature manager fm := featureManager.StartManager() @@ -127,7 +126,7 @@ func NewServer(serverOptions *ServerOptions) (*Server, error) { if storage == "clickhouse" { zap.L().Info("Using ClickHouse as datastore ...") clickhouseReader := clickhouseReader.NewReader( - localDB, + serverOptions.SigNoz.SQLStore.SQLxDB(), serverOptions.PromConfigPath, fm, serverOptions.MaxIdleConns, @@ -163,7 +162,7 @@ func NewServer(serverOptions *ServerOptions) (*Server, error) { rm, err := makeRulesManager( serverOptions.PromConfigPath, constants.GetAlertManagerApiPrefix(), - serverOptions.RuleRepoURL, localDB, reader, c, serverOptions.DisableRules, fm, serverOptions.UseLogsNewSchema, serverOptions.UseTraceNewSchema) + serverOptions.RuleRepoURL, serverOptions.SigNoz.SQLStore.SQLxDB(), reader, c, serverOptions.DisableRules, fm, serverOptions.UseLogsNewSchema, serverOptions.UseTraceNewSchema) if err != nil { return nil, err } @@ -173,18 +172,18 @@ func NewServer(serverOptions *ServerOptions) (*Server, error) { return nil, err } - integrationsController, err := integrations.NewController(localDB) + integrationsController, err := integrations.NewController(serverOptions.SigNoz.SQLStore.SQLxDB()) if err != nil { return nil, fmt.Errorf("couldn't create integrations controller: %w", err) } - cloudIntegrationsController, err := cloudintegrations.NewController(localDB) + cloudIntegrationsController, err := cloudintegrations.NewController(serverOptions.SigNoz.SQLStore.SQLxDB()) if err != nil { return nil, fmt.Errorf("couldn't create cloud provider integrations controller: %w", err) } logParsingPipelineController, err := logparsingpipeline.NewLogParsingPipelinesController( - localDB, "sqlite", integrationsController.GetPipelinesForInstalledIntegrations, + serverOptions.SigNoz.SQLStore.SQLxDB(), integrationsController.GetPipelinesForInstalledIntegrations, ) if err != nil { return nil, err @@ -236,14 +235,13 @@ func NewServer(serverOptions *ServerOptions) (*Server, error) { s.privateHTTP = privateServer - _, err = opAmpModel.InitDB(localDB) + _, err = opAmpModel.InitDB(serverOptions.SigNoz.SQLStore.SQLxDB()) if err != nil { return nil, err } agentConfMgr, err := agentConf.Initiate(&agentConf.ManagerOptions{ - DB: localDB, - DBEngine: "sqlite", + DB: serverOptions.SigNoz.SQLStore.SQLxDB(), AgentFeatures: []agentConf.AgentFeature{ logParsingPipelineController, }, diff --git a/pkg/query-service/constants/constants.go b/pkg/query-service/constants/constants.go index 24ae12c4bb..f5999917be 100644 --- a/pkg/query-service/constants/constants.go +++ b/pkg/query-service/constants/constants.go @@ -76,6 +76,7 @@ var AmChannelApiPath = GetOrDefaultEnv("ALERTMANAGER_API_CHANNEL_PATH", "v1/rout var OTLPTarget = GetOrDefaultEnv("OTEL_EXPORTER_OTLP_ENDPOINT", "") var LogExportBatchSize = GetOrDefaultEnv("OTEL_BLRP_MAX_EXPORT_BATCH_SIZE", "512") +// [Deprecated] SIGNOZ_LOCAL_DB_PATH is deprecated and scheduled for removal. Please use SIGNOZ_SQLSTORE_SQLITE_PATH instead. var RELATIONAL_DATASOURCE_PATH = GetOrDefaultEnv("SIGNOZ_LOCAL_DB_PATH", "/var/lib/signoz/signoz.db") var DurationSortFeature = GetOrDefaultEnv("DURATION_SORT_FEATURE", "true") diff --git a/pkg/query-service/dao/factory.go b/pkg/query-service/dao/factory.go index a8b13cd295..02d50158fd 100644 --- a/pkg/query-service/dao/factory.go +++ b/pkg/query-service/dao/factory.go @@ -1,26 +1,19 @@ package dao import ( - "fmt" - - "github.com/pkg/errors" + "github.com/jmoiron/sqlx" "go.signoz.io/signoz/pkg/query-service/dao/sqlite" ) var db ModelDao -func InitDao(engine, path string) error { +func InitDao(inputDB *sqlx.DB) error { var err error - - switch engine { - case "sqlite": - db, err = sqlite.InitDB(path) - if err != nil { - return errors.Wrap(err, "failed to initialize DB") - } - default: - return fmt.Errorf("RelationalDB type: %s is not supported in query service", engine) + db, err = sqlite.InitDB(inputDB) + if err != nil { + return err } + return nil } diff --git a/pkg/query-service/dao/sqlite/connection.go b/pkg/query-service/dao/sqlite/connection.go index a7e77334ef..9babfe175f 100644 --- a/pkg/query-service/dao/sqlite/connection.go +++ b/pkg/query-service/dao/sqlite/connection.go @@ -17,15 +17,7 @@ type ModelDaoSqlite struct { } // InitDB sets up setting up the connection pool global variable. -func InitDB(dataSourceName string) (*ModelDaoSqlite, error) { - var err error - - db, err := sqlx.Open("sqlite3", dataSourceName) - if err != nil { - return nil, errors.Wrap(err, "failed to Open sqlite3 DB") - } - db.SetMaxOpenConns(10) - +func InitDB(db *sqlx.DB) (*ModelDaoSqlite, error) { table_schema := ` PRAGMA foreign_keys = ON; @@ -88,7 +80,7 @@ func InitDB(dataSourceName string) (*ModelDaoSqlite, error) { ); ` - _, err = db.Exec(table_schema) + _, err := db.Exec(table_schema) if err != nil { return nil, fmt.Errorf("error in creating tables: %v", err.Error()) } diff --git a/pkg/query-service/main.go b/pkg/query-service/main.go index ba475a05e3..835e51c10f 100644 --- a/pkg/query-service/main.go +++ b/pkg/query-service/main.go @@ -122,7 +122,7 @@ func main() { zap.L().Info("JWT secret key set successfully.") } - if err := migrate.Migrate(constants.RELATIONAL_DATASOURCE_PATH); err != nil { + if err := migrate.Migrate(signoz.SQLStore.SQLxDB()); err != nil { zap.L().Error("Failed to migrate", zap.Error(err)) } else { zap.L().Info("Migration successful") diff --git a/pkg/query-service/migrate/migate.go b/pkg/query-service/migrate/migate.go index d6d90e2168..7b2d02a77c 100644 --- a/pkg/query-service/migrate/migate.go +++ b/pkg/query-service/migrate/migate.go @@ -41,11 +41,7 @@ func getMigrationVersion(conn *sqlx.DB, version string) (*DataMigration, error) return &migration, nil } -func Migrate(dsn string) error { - conn, err := sqlx.Connect("sqlite3", dsn) - if err != nil { - return err - } +func Migrate(conn *sqlx.DB) error { if err := initSchema(conn); err != nil { return err } diff --git a/pkg/query-service/tests/integration/logparsingpipeline_test.go b/pkg/query-service/tests/integration/logparsingpipeline_test.go index 9e07b604d7..50c577002b 100644 --- a/pkg/query-service/tests/integration/logparsingpipeline_test.go +++ b/pkg/query-service/tests/integration/logparsingpipeline_test.go @@ -461,7 +461,7 @@ func NewTestbedWithoutOpamp(t *testing.T, testDB *sqlx.DB) *LogPipelinesTestBed } controller, err := logparsingpipeline.NewLogParsingPipelinesController( - testDB, "sqlite", ic.GetPipelinesForInstalledIntegrations, + testDB, ic.GetPipelinesForInstalledIntegrations, ) if err != nil { t.Fatalf("could not create a logparsingpipelines controller: %v", err) @@ -485,8 +485,7 @@ func NewTestbedWithoutOpamp(t *testing.T, testDB *sqlx.DB) *LogPipelinesTestBed require.Nil(t, err, "failed to init opamp model") agentConfMgr, err := agentConf.Initiate(&agentConf.ManagerOptions{ - DB: testDB, - DBEngine: "sqlite", + DB: testDB, AgentFeatures: []agentConf.AgentFeature{ apiHandler.LogsParsingPipelineController, }}) diff --git a/pkg/query-service/utils/testutils.go b/pkg/query-service/utils/testutils.go index 69c7a9bf24..475586f820 100644 --- a/pkg/query-service/utils/testutils.go +++ b/pkg/query-service/utils/testutils.go @@ -28,11 +28,11 @@ func NewTestSqliteDB(t *testing.T) (testDB *sqlx.DB, testDBFilePath string) { } func NewQueryServiceDBForTests(t *testing.T) *sqlx.DB { - testDB, testDBFilePath := NewTestSqliteDB(t) + testDB, _ := NewTestSqliteDB(t) // TODO(Raj): This should not require passing in the DB file path - dao.InitDao("sqlite", testDBFilePath) - dashboards.InitDB(testDBFilePath) + dao.InitDao(testDB) + dashboards.InitDB(testDB) return testDB } diff --git a/pkg/signoz/config.go b/pkg/signoz/config.go index 4b0041d271..baf1f09eae 100644 --- a/pkg/signoz/config.go +++ b/pkg/signoz/config.go @@ -2,6 +2,8 @@ package signoz import ( "context" + "fmt" + "os" "go.signoz.io/signoz/pkg/cache" "go.signoz.io/signoz/pkg/config" @@ -49,5 +51,15 @@ func NewConfig(ctx context.Context, resolverConfig config.ResolverConfig) (Confi return Config{}, err } + mergeAndEnsureBackwardCompatibility(&config) + return config, nil } + +func mergeAndEnsureBackwardCompatibility(config *Config) { + // SIGNOZ_LOCAL_DB_PATH + if os.Getenv("SIGNOZ_LOCAL_DB_PATH") != "" { + fmt.Println("[Deprecated] env SIGNOZ_LOCAL_DB_PATH is deprecated and scheduled for removal. Please use SIGNOZ_SQLSTORE_SQLITE_PATH instead.") + config.SQLStore.Sqlite.Path = os.Getenv("SIGNOZ_LOCAL_DB_PATH") + } +} diff --git a/pkg/signoz/signoz.go b/pkg/signoz/signoz.go index ce31a349cf..f01475b213 100644 --- a/pkg/signoz/signoz.go +++ b/pkg/signoz/signoz.go @@ -6,14 +6,16 @@ import ( "go.signoz.io/signoz/pkg/cache" "go.signoz.io/signoz/pkg/factory" "go.signoz.io/signoz/pkg/instrumentation" + "go.signoz.io/signoz/pkg/sqlstore" "go.signoz.io/signoz/pkg/version" "go.signoz.io/signoz/pkg/web" ) type SigNoz struct { - Cache cache.Cache - Web web.Web + Cache cache.Cache + Web web.Web + SQLStore sqlstore.SQLStore } func New( @@ -54,8 +56,21 @@ func New( return nil, err } + // Initialize sqlstore from the available sqlstore provider factories + sqlstore, err := factory.NewProviderFromNamedMap( + ctx, + providerSettings, + config.SQLStore, + providerConfig.SQLStoreProviderFactories, + config.SQLStore.Provider, + ) + if err != nil { + return nil, err + } + return &SigNoz{ - Cache: cache, - Web: web, + Cache: cache, + Web: web, + SQLStore: sqlstore, }, nil }