GitHub-Proxy/signoz
1
0
Fork 0
mirror of https://git.mirrors.martin98.com/https://github.com/SigNoz/signoz synced 2025-08-12 03:39:02 +08:00
Code Issues Packages Projects Releases Wiki Activity

feat(sqlstore): open a connection to sql once (#6867)

Browse Source
This commit is contained in:
Vibhu Pandey 2025-01-22 12:49:38 +05:30 committed by GitHub
parent 0baf0e9453
commit 837f434fe9
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
24 changed files with 118 additions and 158 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

34
ee/query-service/app/server.go
View File

@ -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
}

14
ee/query-service/dao/factory.go
View File

@ -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)
}

4
ee/query-service/dao/sqlite/modelDao.go
View File

@ -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
}

9
ee/query-service/license/db.go
View File

@ -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) {

4
ee/query-service/license/manager.go
View File

@ -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)

2
ee/query-service/main.go
View File

@ -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")

2
ee/query-service/usage/manager.go
View File

@ -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<hostname>.*):`)
hostNameRegexMatches := hostNameRegex.FindStringSubmatch(os.Getenv("ClickHouseUrl"))

9
pkg/query-service/agentConf/db.go
View File

@ -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(

5
pkg/query-service/agentConf/manager.go
View File

@ -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")
}

38
pkg/query-service/app/dashboards/model.go
View File

@ -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
}

15
pkg/query-service/app/explorer/db.go
View File

@ -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) {

3
pkg/query-service/app/logparsingpipeline/controller.go
View File

@ -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,

9
pkg/query-service/app/logparsingpipeline/db.go
View File

@ -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

12
pkg/query-service/app/preferences/model.go
View File

@ -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())
}

30
pkg/query-service/app/server.go
View File

@ -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,
},

1
pkg/query-service/constants/constants.go
View File

@ -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")

19
pkg/query-service/dao/factory.go
View File

@ -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
}

12
pkg/query-service/dao/sqlite/connection.go
View File

@ -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())
}

2
pkg/query-service/main.go
View File

@ -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")

6
pkg/query-service/migrate/migate.go
View File

@ -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
}

5
pkg/query-service/tests/integration/logparsingpipeline_test.go
View File

@ -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,
}})

6
pkg/query-service/utils/testutils.go
View File

@ -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
}

12
pkg/signoz/config.go
View File

@ -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")
}
}

23
pkg/signoz/signoz.go
View File

@ -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
}