postgres_store.go 2.6 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899
  1. package postgres
  2. import (
  3. "database/sql"
  4. "fmt"
  5. "strconv"
  6. "time"
  7. _ "github.com/lib/pq"
  8. "github.com/seaweedfs/seaweedfs/weed/filer"
  9. "github.com/seaweedfs/seaweedfs/weed/filer/abstract_sql"
  10. "github.com/seaweedfs/seaweedfs/weed/util"
  11. )
  12. func init() {
  13. filer.Stores = append(filer.Stores, &PostgresStore{})
  14. }
  15. type PostgresStore struct {
  16. abstract_sql.AbstractSqlStore
  17. }
  18. func (store *PostgresStore) GetName() string {
  19. return "postgres"
  20. }
  21. func (store *PostgresStore) Initialize(configuration util.Configuration, prefix string) (err error) {
  22. return store.initialize(
  23. configuration.GetString(prefix+"upsertQuery"),
  24. configuration.GetBool(prefix+"enableUpsert"),
  25. configuration.GetString(prefix+"username"),
  26. configuration.GetString(prefix+"password"),
  27. configuration.GetString(prefix+"hostname"),
  28. configuration.GetInt(prefix+"port"),
  29. configuration.GetString(prefix+"database"),
  30. configuration.GetString(prefix+"schema"),
  31. configuration.GetString(prefix+"sslmode"),
  32. configuration.GetInt(prefix+"connection_max_idle"),
  33. configuration.GetInt(prefix+"connection_max_open"),
  34. configuration.GetInt(prefix+"connection_max_lifetime_seconds"),
  35. )
  36. }
  37. func (store *PostgresStore) initialize(upsertQuery string, enableUpsert bool, user, password, hostname string, port int, database, schema, sslmode string, maxIdle, maxOpen, maxLifetimeSeconds int) (err error) {
  38. store.SupportBucketTable = false
  39. if !enableUpsert {
  40. upsertQuery = ""
  41. }
  42. store.SqlGenerator = &SqlGenPostgres{
  43. CreateTableSqlTemplate: "",
  44. DropTableSqlTemplate: `drop table "%s"`,
  45. UpsertQueryTemplate: upsertQuery,
  46. }
  47. sqlUrl := "connect_timeout=30"
  48. if hostname != "" {
  49. sqlUrl += " host=" + hostname
  50. }
  51. if port != 0 {
  52. sqlUrl += " port=" + strconv.Itoa(port)
  53. }
  54. if sslmode != "" {
  55. sqlUrl += " sslmode=" + sslmode
  56. }
  57. if user != "" {
  58. sqlUrl += " user=" + user
  59. }
  60. adaptedSqlUrl := sqlUrl
  61. if password != "" {
  62. sqlUrl += " password=" + password
  63. adaptedSqlUrl += " password=ADAPTED"
  64. }
  65. if database != "" {
  66. sqlUrl += " dbname=" + database
  67. adaptedSqlUrl += " dbname=" + database
  68. }
  69. if schema != "" {
  70. sqlUrl += " search_path=" + schema
  71. adaptedSqlUrl += " search_path=" + schema
  72. }
  73. var dbErr error
  74. store.DB, dbErr = sql.Open("postgres", sqlUrl)
  75. if dbErr != nil {
  76. store.DB.Close()
  77. store.DB = nil
  78. return fmt.Errorf("can not connect to %s error:%v", adaptedSqlUrl, err)
  79. }
  80. store.DB.SetMaxIdleConns(maxIdle)
  81. store.DB.SetMaxOpenConns(maxOpen)
  82. store.DB.SetConnMaxLifetime(time.Duration(maxLifetimeSeconds) * time.Second)
  83. if err = store.DB.Ping(); err != nil {
  84. return fmt.Errorf("connect to %s error:%v", adaptedSqlUrl, err)
  85. }
  86. return nil
  87. }