public/grafana
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Sqlstore refactor (#11908)

Browse Source 
* refactor: tracing service refactoring

* refactor: sqlstore to instance service

* refactor: sqlstore & registory priority

* refactor: sqlstore refactor wip

* sqlstore: progress on getting tests to work again

* sqlstore: progress on refactoring and getting tests working

* sqlstore: connection string fix

* fix: not sure why this test is not working and required changing expires

* fix: updated grafana-cli
This commit is contained in:
Torkel Ödegaard
2018-05-18 11:10:10 +02:00
committed by GitHub
parent 78e837cf39
commit 6c0752473a
13 changed files with 228 additions and 201 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pkg/services/sqlstore/sqlstore.go
+163 -142
View File
@@ -13,6 +13,7 @@ import (
"github.com/grafana/grafana/pkg/bus"
"github.com/grafana/grafana/pkg/log"
m "github.com/grafana/grafana/pkg/models"
"github.com/grafana/grafana/pkg/registry"
"github.com/grafana/grafana/pkg/services/annotations"
"github.com/grafana/grafana/pkg/services/sqlstore/migrations"
"github.com/grafana/grafana/pkg/services/sqlstore/migrator"
@@ -27,39 +28,72 @@ import (
_ "github.com/grafana/grafana/pkg/tsdb/mssql"
)
type DatabaseConfig struct {
Type, Host, Name, User, Pwd, Path, SslMode string
CaCertPath string
ClientKeyPath string
ClientCertPath string
ServerCertName string
MaxOpenConn int
MaxIdleConn int
ConnMaxLifetime int
}
var (
x *xorm.Engine
dialect migrator.Dialect
HasEngine bool
DbCfg DatabaseConfig
UseSQLite3 bool
sqlog log.Logger = log.New("sqlstore")
sqlog log.Logger = log.New("sqlstore")
)
func EnsureAdminUser() {
func init() {
registry.Register(&registry.Descriptor{
Name: "SqlStore",
Instance: &SqlStore{},
InitPriority: registry.High,
})
}
type SqlStore struct {
Cfg *setting.Cfg `inject:""`
dbCfg DatabaseConfig
engine *xorm.Engine
log log.Logger
skipEnsureAdmin bool
}
func (ss *SqlStore) Init() error {
ss.log = log.New("sqlstore")
ss.readConfig()
engine, err := ss.getEngine()
if err != nil {
return fmt.Errorf("Fail to connect to database: %v", err)
}
ss.engine = engine
// temporarily still set global var
x = engine
dialect = migrator.NewDialect(x)
migrator := migrator.NewMigrator(x)
migrations.AddMigrations(migrator)
if err := migrator.Start(); err != nil {
return fmt.Errorf("Migration failed err: %v", err)
}
// Init repo instances
annotations.SetRepository(&SqlAnnotationRepo{})
// ensure admin user
if ss.skipEnsureAdmin {
return nil
}
return ss.ensureAdminUser()
}
func (ss *SqlStore) ensureAdminUser() error {
statsQuery := m.GetSystemStatsQuery{}
if err := bus.Dispatch(&statsQuery); err != nil {
log.Fatal(3, "Could not determine if admin user exists: %v", err)
return
fmt.Errorf("Could not determine if admin user exists: %v", err)
}
if statsQuery.Result.Users > 0 {
return
return nil
}
cmd := m.CreateUserCommand{}
@@ -69,109 +103,89 @@ func EnsureAdminUser() {
cmd.IsAdmin = true
if err := bus.Dispatch(&cmd); err != nil {
log.Error(3, "Failed to create default admin user", err)
return
return fmt.Errorf("Failed to create admin user: %v", err)
}
log.Info("Created default admin user: %v", setting.AdminUser)
}
ss.log.Info("Created default admin user: %v", setting.AdminUser)
func NewEngine() *xorm.Engine {
x, err := getEngine()
if err != nil {
sqlog.Crit("Fail to connect to database", "error", err)
os.Exit(1)
}
err = SetEngine(x)
if err != nil {
sqlog.Error("Fail to initialize orm engine", "error", err)
os.Exit(1)
}
return x
}
func SetEngine(engine *xorm.Engine) (err error) {
x = engine
dialect = migrator.NewDialect(x)
migrator := migrator.NewMigrator(x)
migrations.AddMigrations(migrator)
if err := migrator.Start(); err != nil {
return fmt.Errorf("Sqlstore::Migration failed err: %v\n", err)
}
// Init repo instances
annotations.SetRepository(&SqlAnnotationRepo{})
return nil
}
func getEngine() (*xorm.Engine, error) {
LoadConfig()
func (ss *SqlStore) buildConnectionString() (string, error) {
cnnstr := ss.dbCfg.ConnectionString
cnnstr := ""
switch DbCfg.Type {
// special case used by integration tests
if cnnstr != "" {
return cnnstr, nil
}
switch ss.dbCfg.Type {
case migrator.MYSQL:
protocol := "tcp"
if strings.HasPrefix(DbCfg.Host, "/") {
if strings.HasPrefix(ss.dbCfg.Host, "/") {
protocol = "unix"
}
cnnstr = fmt.Sprintf("%s:%s@%s(%s)/%s?collation=utf8mb4_unicode_ci&allowNativePasswords=true",
url.QueryEscape(DbCfg.User), url.QueryEscape(DbCfg.Pwd), protocol, DbCfg.Host, url.PathEscape(DbCfg.Name))
ss.dbCfg.User, ss.dbCfg.Pwd, protocol, ss.dbCfg.Host, ss.dbCfg.Name)
if DbCfg.SslMode == "true" || DbCfg.SslMode == "skip-verify" {
tlsCert, err := makeCert("custom", DbCfg)
if ss.dbCfg.SslMode == "true" || ss.dbCfg.SslMode == "skip-verify" {
tlsCert, err := makeCert("custom", ss.dbCfg)
if err != nil {
return nil, err
return "", err
}
mysql.RegisterTLSConfig("custom", tlsCert)
cnnstr += "&tls=custom"
}
case migrator.POSTGRES:
var host, port = "127.0.0.1", "5432"
fields := strings.Split(DbCfg.Host, ":")
fields := strings.Split(ss.dbCfg.Host, ":")
if len(fields) > 0 && len(strings.TrimSpace(fields[0])) > 0 {
host = fields[0]
}
if len(fields) > 1 && len(strings.TrimSpace(fields[1])) > 0 {
port = fields[1]
}
cnnstr = fmt.Sprintf("user='%s' password='%s' host='%s' port='%s' dbname='%s' sslmode='%s' sslcert='%s' sslkey='%s' sslrootcert='%s'",
strings.Replace(DbCfg.User, `'`, `\'`, -1),
strings.Replace(DbCfg.Pwd, `'`, `\'`, -1),
strings.Replace(host, `'`, `\'`, -1),
strings.Replace(port, `'`, `\'`, -1),
strings.Replace(DbCfg.Name, `'`, `\'`, -1),
strings.Replace(DbCfg.SslMode, `'`, `\'`, -1),
strings.Replace(DbCfg.ClientCertPath, `'`, `\'`, -1),
strings.Replace(DbCfg.ClientKeyPath, `'`, `\'`, -1),
strings.Replace(DbCfg.CaCertPath, `'`, `\'`, -1),
)
case migrator.SQLITE:
if !filepath.IsAbs(DbCfg.Path) {
DbCfg.Path = filepath.Join(setting.DataPath, DbCfg.Path)
if ss.dbCfg.Pwd == "" {
ss.dbCfg.Pwd = "''"
}
os.MkdirAll(path.Dir(DbCfg.Path), os.ModePerm)
cnnstr = "file:" + DbCfg.Path + "?cache=shared&mode=rwc"
if ss.dbCfg.User == "" {
ss.dbCfg.User = "''"
}
cnnstr = fmt.Sprintf("user=%s password=%s host=%s port=%s dbname=%s sslmode=%s sslcert=%s sslkey=%s sslrootcert=%s", ss.dbCfg.User, ss.dbCfg.Pwd, host, port, ss.dbCfg.Name, ss.dbCfg.SslMode, ss.dbCfg.ClientCertPath, ss.dbCfg.ClientKeyPath, ss.dbCfg.CaCertPath)
case migrator.SQLITE:
// special case for tests
if !filepath.IsAbs(ss.dbCfg.Path) {
ss.dbCfg.Path = filepath.Join(setting.DataPath, ss.dbCfg.Path)
}
os.MkdirAll(path.Dir(ss.dbCfg.Path), os.ModePerm)
cnnstr = "file:" + ss.dbCfg.Path + "?cache=shared&mode=rwc"
default:
return nil, fmt.Errorf("Unknown database type: %s", DbCfg.Type)
return "", fmt.Errorf("Unknown database type: %s", ss.dbCfg.Type)
}
sqlog.Info("Initializing DB", "dbtype", DbCfg.Type)
engine, err := xorm.NewEngine(DbCfg.Type, cnnstr)
return cnnstr, nil
}
func (ss *SqlStore) getEngine() (*xorm.Engine, error) {
connectionString, err := ss.buildConnectionString()
if err != nil {
return nil, err
}
engine.SetMaxOpenConns(DbCfg.MaxOpenConn)
engine.SetMaxIdleConns(DbCfg.MaxIdleConn)
engine.SetConnMaxLifetime(time.Second * time.Duration(DbCfg.ConnMaxLifetime))
debugSql := setting.Raw.Section("database").Key("log_queries").MustBool(false)
sqlog.Info("Connecting to DB", "dbtype", ss.dbCfg.Type)
engine, err := xorm.NewEngine(ss.dbCfg.Type, connectionString)
if err != nil {
return nil, err
}
engine.SetMaxOpenConns(ss.dbCfg.MaxOpenConn)
engine.SetMaxIdleConns(ss.dbCfg.MaxIdleConn)
engine.SetConnMaxLifetime(time.Second * time.Duration(ss.dbCfg.ConnMaxLifetime))
// configure sql logging
debugSql := ss.Cfg.Raw.Section("database").Key("log_queries").MustBool(false)
if !debugSql {
engine.SetLogger(&xorm.DiscardLogger{})
} else {
@@ -183,95 +197,90 @@ func getEngine() (*xorm.Engine, error) {
return engine, nil
}
func LoadConfig() {
sec := setting.Raw.Section("database")
func (ss *SqlStore) readConfig() {
sec := ss.Cfg.Raw.Section("database")
cfgURL := sec.Key("url").String()
if len(cfgURL) != 0 {
dbURL, _ := url.Parse(cfgURL)
DbCfg.Type = dbURL.Scheme
DbCfg.Host = dbURL.Host
ss.dbCfg.Type = dbURL.Scheme
ss.dbCfg.Host = dbURL.Host
pathSplit := strings.Split(dbURL.Path, "/")
if len(pathSplit) > 1 {
DbCfg.Name = pathSplit[1]
ss.dbCfg.Name = pathSplit[1]
}
userInfo := dbURL.User
if userInfo != nil {
DbCfg.User = userInfo.Username()
DbCfg.Pwd, _ = userInfo.Password()
ss.dbCfg.User = userInfo.Username()
ss.dbCfg.Pwd, _ = userInfo.Password()
}
} else {
DbCfg.Type = sec.Key("type").String()
DbCfg.Host = sec.Key("host").String()
DbCfg.Name = sec.Key("name").String()
DbCfg.User = sec.Key("user").String()
if len(DbCfg.Pwd) == 0 {
DbCfg.Pwd = sec.Key("password").String()
}
ss.dbCfg.Type = sec.Key("type").String()
ss.dbCfg.Host = sec.Key("host").String()
ss.dbCfg.Name = sec.Key("name").String()
ss.dbCfg.User = sec.Key("user").String()
ss.dbCfg.ConnectionString = sec.Key("connection_string").String()
ss.dbCfg.Pwd = sec.Key("password").String()
}
DbCfg.MaxOpenConn = sec.Key("max_open_conn").MustInt(0)
DbCfg.MaxIdleConn = sec.Key("max_idle_conn").MustInt(0)
DbCfg.ConnMaxLifetime = sec.Key("conn_max_lifetime").MustInt(14400)
if DbCfg.Type == "sqlite3" {
UseSQLite3 = true
// only allow one connection as sqlite3 has multi threading issues that cause table locks
// DbCfg.MaxIdleConn = 1
// DbCfg.MaxOpenConn = 1
}
DbCfg.SslMode = sec.Key("ssl_mode").String()
DbCfg.CaCertPath = sec.Key("ca_cert_path").String()
DbCfg.ClientKeyPath = sec.Key("client_key_path").String()
DbCfg.ClientCertPath = sec.Key("client_cert_path").String()
DbCfg.ServerCertName = sec.Key("server_cert_name").String()
DbCfg.Path = sec.Key("path").MustString("data/grafana.db")
ss.dbCfg.MaxOpenConn = sec.Key("max_open_conn").MustInt(0)
ss.dbCfg.MaxIdleConn = sec.Key("max_idle_conn").MustInt(2)
ss.dbCfg.ConnMaxLifetime = sec.Key("conn_max_lifetime").MustInt(14400)
ss.dbCfg.SslMode = sec.Key("ssl_mode").String()
ss.dbCfg.CaCertPath = sec.Key("ca_cert_path").String()
ss.dbCfg.ClientKeyPath = sec.Key("client_key_path").String()
ss.dbCfg.ClientCertPath = sec.Key("client_cert_path").String()
ss.dbCfg.ServerCertName = sec.Key("server_cert_name").String()
ss.dbCfg.Path = sec.Key("path").MustString("data/grafana.db")
}
func InitTestDB(t *testing.T) *xorm.Engine {
selectedDb := migrator.SQLITE
// selectedDb := migrator.MYSQL
// selectedDb := migrator.POSTGRES
func InitTestDB(t *testing.T) *SqlStore {
sqlstore := &SqlStore{}
sqlstore.skipEnsureAdmin = true
var x *xorm.Engine
var err error
dbType := migrator.SQLITE
// environment variable present for test db?
if db, present := os.LookupEnv("GRAFANA_TEST_DB"); present {
selectedDb = db
dbType = db
}
switch strings.ToLower(selectedDb) {
case migrator.MYSQL:
x, err = xorm.NewEngine(sqlutil.TestDB_Mysql.DriverName, sqlutil.TestDB_Mysql.ConnStr)
case migrator.POSTGRES:
x, err = xorm.NewEngine(sqlutil.TestDB_Postgres.DriverName, sqlutil.TestDB_Postgres.ConnStr)
// set test db config
sqlstore.Cfg = setting.NewCfg()
sec, _ := sqlstore.Cfg.Raw.NewSection("database")
sec.NewKey("type", dbType)
switch dbType {
case "mysql":
sec.NewKey("connection_string", sqlutil.TestDB_Mysql.ConnStr)
case "postgres":
sec.NewKey("connection_string", sqlutil.TestDB_Postgres.ConnStr)
default:
x, err = xorm.NewEngine(sqlutil.TestDB_Sqlite3.DriverName, sqlutil.TestDB_Sqlite3.ConnStr)
sec.NewKey("connection_string", sqlutil.TestDB_Sqlite3.ConnStr)
}
x.DatabaseTZ = time.UTC
x.TZLocation = time.UTC
// need to get engine to clean db before we init
engine, err := xorm.NewEngine(dbType, sec.Key("connection_string").String())
if err != nil {
t.Fatalf("Failed to init test database: %v", err)
}
dialect = migrator.NewDialect(x)
err = dialect.CleanDB()
if err != nil {
dialect = migrator.NewDialect(engine)
if err := dialect.CleanDB(); err != nil {
t.Fatalf("Failed to clean test db %v", err)
}
if err := SetEngine(x); err != nil {
t.Fatal(err)
if err := sqlstore.Init(); err != nil {
t.Fatalf("Failed to init test database: %v", err)
}
// x.ShowSQL()
//// sqlstore.engine.DatabaseTZ = time.UTC
//// sqlstore.engine.TZLocation = time.UTC
return x
return sqlstore
}
func IsTestDbMySql() bool {
@@ -289,3 +298,15 @@ func IsTestDbPostgres() bool {
return false
}
type DatabaseConfig struct {
Type, Host, Name, User, Pwd, Path, SslMode string
CaCertPath string
ClientKeyPath string
ClientCertPath string
ServerCertName string
ConnectionString string
MaxOpenConn int
MaxIdleConn int
ConnMaxLifetime int
}