mysql2_store.go 2.8 KB

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