mysql_store.go 2.3 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384
  1. package mysql
  2. import (
  3. "database/sql"
  4. "fmt"
  5. "time"
  6. "github.com/seaweedfs/seaweedfs/weed/filer"
  7. _ "github.com/go-sql-driver/mysql"
  8. "github.com/seaweedfs/seaweedfs/weed/filer/abstract_sql"
  9. "github.com/seaweedfs/seaweedfs/weed/util"
  10. )
  11. const (
  12. CONNECTION_URL_PATTERN = "%s:%s@tcp(%s:%d)/%s?collation=utf8mb4_bin"
  13. )
  14. func init() {
  15. filer.Stores = append(filer.Stores, &MysqlStore{})
  16. }
  17. type MysqlStore struct {
  18. abstract_sql.AbstractSqlStore
  19. }
  20. func (store *MysqlStore) GetName() string {
  21. return "mysql"
  22. }
  23. func (store *MysqlStore) Initialize(configuration util.Configuration, prefix string) (err error) {
  24. return store.initialize(
  25. configuration.GetString(prefix+"upsertQuery"),
  26. configuration.GetBool(prefix+"enableUpsert"),
  27. configuration.GetString(prefix+"username"),
  28. configuration.GetString(prefix+"password"),
  29. configuration.GetString(prefix+"hostname"),
  30. configuration.GetInt(prefix+"port"),
  31. configuration.GetString(prefix+"database"),
  32. configuration.GetInt(prefix+"connection_max_idle"),
  33. configuration.GetInt(prefix+"connection_max_open"),
  34. configuration.GetInt(prefix+"connection_max_lifetime_seconds"),
  35. configuration.GetBool(prefix+"interpolateParams"),
  36. )
  37. }
  38. func (store *MysqlStore) initialize(upsertQuery string, enableUpsert bool, user, password, hostname string, port int, database string, maxIdle, maxOpen,
  39. maxLifetimeSeconds int, interpolateParams bool) (err error) {
  40. store.SupportBucketTable = false
  41. if !enableUpsert {
  42. upsertQuery = ""
  43. }
  44. store.SqlGenerator = &SqlGenMysql{
  45. CreateTableSqlTemplate: "",
  46. DropTableSqlTemplate: "DROP TABLE `%s`",
  47. UpsertQueryTemplate: upsertQuery,
  48. }
  49. sqlUrl := fmt.Sprintf(CONNECTION_URL_PATTERN, user, password, hostname, port, database)
  50. adaptedSqlUrl := fmt.Sprintf(CONNECTION_URL_PATTERN, user, "<ADAPTED>", hostname, port, database)
  51. if interpolateParams {
  52. sqlUrl += "&interpolateParams=true"
  53. adaptedSqlUrl += "&interpolateParams=true"
  54. }
  55. var dbErr error
  56. store.DB, dbErr = sql.Open("mysql", sqlUrl)
  57. if dbErr != nil {
  58. store.DB.Close()
  59. store.DB = nil
  60. return fmt.Errorf("can not connect to %s error:%v", adaptedSqlUrl, err)
  61. }
  62. store.DB.SetMaxIdleConns(maxIdle)
  63. store.DB.SetMaxOpenConns(maxOpen)
  64. store.DB.SetConnMaxLifetime(time.Duration(maxLifetimeSeconds) * time.Second)
  65. if err = store.DB.Ping(); err != nil {
  66. return fmt.Errorf("connect to %s error:%v", sqlUrl, err)
  67. }
  68. return nil
  69. }