From e88795be133d384813b6b0dd2d556565856cb663 Mon Sep 17 00:00:00 2001 From: Mike Tolman Date: Wed, 7 Dec 2016 17:24:40 -0700 Subject: [PATCH] Adding PostgreSQL Support --- weed/filer/postgres_store/postgres_store.go | 365 ++++++++++++++++++++ weed/server/filer_server.go | 5 + 2 files changed, 370 insertions(+) create mode 100644 weed/filer/postgres_store/postgres_store.go diff --git a/weed/filer/postgres_store/postgres_store.go b/weed/filer/postgres_store/postgres_store.go new file mode 100644 index 000000000..9b674aedf --- /dev/null +++ b/weed/filer/postgres_store/postgres_store.go @@ -0,0 +1,365 @@ +package postgres_store + +import ( + "database/sql" + "fmt" + "hash/crc32" + "sync" + "time" + + "github.com/chrislusf/seaweedfs/weed/glog" + + _ "github.com/lib/pq" +) + +const ( + default_maxIdleConnections = 100 + default_maxOpenConnections = 50 + default_maxTableNums = 1024 + tableName = "filer_mapping" +) + +var ( + _init_db sync.Once + _db_connections []*sql.DB +) + +type PostgresConf struct { + User string + Password string + HostName string + Port int + DataBase string + SslMode string + MaxIdleConnections int + MaxOpenConnections int +} + +type ShardingConf struct { + IsSharding bool `json:"isSharding"` + ShardCount int `json:"shardCount"` +} + +type PostgresStore struct { + dbs []*sql.DB + isSharding bool + shardCount int + server string + user string + password string +} + +func databaseExists(db *sql.DB, databaseName string) (bool, error) { + sqlStatement := "SELECT datname from pg_database WHERE datname='%s'" + row := db.QueryRow(fmt.Sprintf(sqlStatement, databaseName)) + + var dbName string + err := row.Scan(&dbName) + if err != nil { + if err == sql.ErrNoRows { + return false, nil + } + return false, err + } + return true, nil +} + +func createDatabase(db *sql.DB, databaseName string) (error) { + sqlStatement := "CREATE DATABASE %s ENCODING='UTF8'"; + _, err := db.Exec(fmt.Sprintf(sqlStatement, databaseName)) + return err +} + +func getDbConnection(confs []PostgresConf) []*sql.DB { + _init_db.Do(func() { + for _, conf := range confs { + + sqlUrl := fmt.Sprintf("host=%s port=%d user=%s password=%s dbname=%s sslmode=%s connect_timeout=30", conf.HostName, conf.Port, conf.User, conf.Password, "postgres", conf.SslMode) + glog.V(3).Infoln("Opening postgres master database") + + var dbErr error + _db_connection, dbErr := sql.Open("postgres", sqlUrl) + if dbErr != nil { + _db_connection.Close() + _db_connection = nil + panic(dbErr) + } + + pingErr := _db_connection.Ping() + if pingErr != nil { + _db_connection.Close() + _db_connection = nil + panic(pingErr) + } + + glog.V(3).Infoln("Checking to see if DB exists: ", conf.DataBase) + var existsErr error + dbExists, existsErr := databaseExists(_db_connection, conf.DataBase) + if existsErr != nil { + _db_connection.Close() + _db_connection = nil + panic(existsErr) + } + + if !dbExists { + glog.V(3).Infoln("Database doesn't exist. Attempting to create one: ", conf.DataBase) + createErr := createDatabase(_db_connection, conf.DataBase) + if createErr != nil { + _db_connection.Close() + _db_connection = nil + panic(createErr) + } + } + + glog.V(3).Infoln("Closing master postgres database and opening configured database: ", conf.DataBase) + _db_connection.Close() + _db_connection = nil + + sqlUrl = fmt.Sprintf("host=%s port=%d user=%s password=%s dbname=%s sslmode=%s connect_timeout=30", conf.HostName, conf.Port, conf.User, conf.Password, conf.DataBase, conf.SslMode) + _db_connection, dbErr = sql.Open("postgres", sqlUrl) + if dbErr != nil { + _db_connection.Close() + _db_connection = nil + panic(dbErr) + } + + pingErr = _db_connection.Ping() + if pingErr != nil { + _db_connection.Close() + _db_connection = nil + panic(pingErr) + } + + var maxIdleConnections, maxOpenConnections int + + if conf.MaxIdleConnections != 0 { + maxIdleConnections = conf.MaxIdleConnections + } else { + maxIdleConnections = default_maxIdleConnections + } + if conf.MaxOpenConnections != 0 { + maxOpenConnections = conf.MaxOpenConnections + } else { + maxOpenConnections = default_maxOpenConnections + } + + _db_connection.SetMaxIdleConns(maxIdleConnections) + _db_connection.SetMaxOpenConns(maxOpenConnections) + _db_connections = append(_db_connections, _db_connection) + } + }) + return _db_connections +} + +func NewPostgresStore(confs []PostgresConf, isSharding bool, shardCount int) *PostgresStore { + pg := &PostgresStore{ + dbs: getDbConnection(confs), + isSharding: isSharding, + shardCount: shardCount, + } + + for _, db := range pg.dbs { + if !isSharding { + pg.shardCount = 1 + } else { + if pg.shardCount == 0 { + pg.shardCount = default_maxTableNums + } + } + for i := 0; i < pg.shardCount; i++ { + if err := pg.createTables(db, tableName, i); err != nil { + fmt.Printf("create table failed %v", err) + } + } + } + + return pg +} + +func (s *PostgresStore) hash(fullFileName string) (instance_offset, table_postfix int) { + hash_value := crc32.ChecksumIEEE([]byte(fullFileName)) + instance_offset = int(hash_value) % len(s.dbs) + table_postfix = int(hash_value) % s.shardCount + return +} + +func (s *PostgresStore) parseFilerMappingInfo(path string) (instanceId int, tableFullName string, err error) { + instance_offset, table_postfix := s.hash(path) + instanceId = instance_offset + if s.isSharding { + tableFullName = fmt.Sprintf("%s_%04d", tableName, table_postfix) + } else { + tableFullName = tableName + } + return +} + +func (s *PostgresStore) Get(fullFilePath string) (fid string, err error) { + instance_offset, tableFullName, err := s.parseFilerMappingInfo(fullFilePath) + if err != nil { + return "", fmt.Errorf("PostgresStore Get operation can not parse file path %s: err is %v", fullFilePath, err) + } + fid, err = s.query(fullFilePath, s.dbs[instance_offset], tableFullName) + + return fid, err +} + +func (s *PostgresStore) Put(fullFilePath string, fid string) (err error) { + var tableFullName string + + instance_offset, tableFullName, err := s.parseFilerMappingInfo(fullFilePath) + if err != nil { + return fmt.Errorf("PostgresStore Put operation can not parse file path %s: err is %v", fullFilePath, err) + } + var old_fid string + if old_fid, err = s.query(fullFilePath, s.dbs[instance_offset], tableFullName); err != nil && err != sql.ErrNoRows { + return fmt.Errorf("PostgresStore Put operation failed when querying path %s: err is %v", fullFilePath, err) + } else { + if len(old_fid) == 0 { + err = s.insert(fullFilePath, fid, s.dbs[instance_offset], tableFullName) + if err != nil { + return fmt.Errorf("PostgresStore Put operation failed when inserting path %s with fid %s : err is %v", fullFilePath, fid, err) + } + } else { + err = s.update(fullFilePath, fid, s.dbs[instance_offset], tableFullName) + if err != nil { + return fmt.Errorf("PostgresStore Put operation failed when updating path %s with fid %s : err is %v", fullFilePath, fid, err) + } + } + } + return +} + +func (s *PostgresStore) Delete(fullFilePath string) (err error) { + var fid string + instance_offset, tableFullName, err := s.parseFilerMappingInfo(fullFilePath) + if err != nil { + return fmt.Errorf("PostgresStore Delete operation can not parse file path %s: err is %v", fullFilePath, err) + } + if fid, err = s.query(fullFilePath, s.dbs[instance_offset], tableFullName); err != nil { + return fmt.Errorf("PostgresStore Delete operation failed when querying path %s: err is %v", fullFilePath, err) + } else if fid == "" { + return nil + } + if err = s.delete(fullFilePath, s.dbs[instance_offset], tableFullName); err != nil { + return fmt.Errorf("PostgresStore Delete operation failed when deleting path %s: err is %v", fullFilePath, err) + } else { + return nil + } +} + +func (s *PostgresStore) Close() { + for _, db := range s.dbs { + db.Close() + } +} + +var createTable = ` + +CREATE TABLE IF NOT EXISTS %s ( + id BIGSERIAL NOT NULL, + uriPath VARCHAR(1024) NOT NULL DEFAULT '', + fid VARCHAR(36) NOT NULL DEFAULT '', + createTime BIGINT NOT NULL DEFAULT 0, + updateTime BIGINT NOT NULL DEFAULT 0, + remark VARCHAR(20) NOT NULL DEFAULT '', + status SMALLINT NOT NULL DEFAULT '1', + PRIMARY KEY (id), + CONSTRAINT %s_index_uriPath UNIQUE (uriPath) +); +` + +func (s *PostgresStore) createTables(db *sql.DB, tableName string, postfix int) error { + var realTableName string + if s.isSharding { + realTableName = fmt.Sprintf("%s_%04d", tableName, postfix) + } else { + realTableName = tableName + } + + glog.V(3).Infoln("Creating postgres table if it doesn't exist: ", realTableName) + + sqlCreate := fmt.Sprintf(createTable, realTableName, realTableName) + + stmt, err := db.Prepare(sqlCreate) + if err != nil { + return err + } + defer stmt.Close() + + _, err = stmt.Exec() + if err != nil { + return err + } + return nil +} + +func (s *PostgresStore) query(uriPath string, db *sql.DB, tableName string) (string, error) { + sqlStatement := fmt.Sprintf("SELECT fid FROM %s WHERE uriPath=$1", tableName) + + row := db.QueryRow(sqlStatement, uriPath) + var fid string + err := row.Scan(&fid) + + glog.V(3).Infof("Postgres query -- looking up path '%s' and found id '%s' ", uriPath, fid) + + if err != nil { + return "", err + } + return fid, nil +} + +func (s *PostgresStore) update(uriPath string, fid string, db *sql.DB, tableName string) error { + sqlStatement := fmt.Sprintf("UPDATE %s SET fid=$1, updateTime=$2 WHERE uriPath=$3", tableName) + + glog.V(3).Infof("Postgres query -- updating path '%s' with id '%s'", uriPath, fid) + + res, err := db.Exec(sqlStatement, fid, time.Now().Unix(), uriPath) + if err != nil { + return err + } + + _, err = res.RowsAffected() + if err != nil { + return err + } + return nil +} + +func (s *PostgresStore) insert(uriPath string, fid string, db *sql.DB, tableName string) error { + sqlStatement := fmt.Sprintf("INSERT INTO %s (uriPath,fid,createTime) VALUES($1, $2, $3)", tableName) + + glog.V(3).Infof("Postgres query -- inserting path '%s' with id '%s'", uriPath, fid) + + res, err := db.Exec(sqlStatement, uriPath, fid, time.Now().Unix()) + + if err != nil { + return err + } + + rows, err := res.RowsAffected() + if rows != 1 { + return fmt.Errorf("Postgres insert -- rows affected = %d. Expecting 1", rows) + } + if err != nil { + return err + } + return nil +} + +func (s *PostgresStore) delete(uriPath string, db *sql.DB, tableName string) error { + sqlStatement := fmt.Sprintf("DELETE FROM %s WHERE uriPath=$1", tableName) + + glog.V(3).Infof("Postgres query -- deleting path '%s'", uriPath) + + res, err := db.Exec(sqlStatement, uriPath) + if err != nil { + return err + } + + _, err = res.RowsAffected() + if err != nil { + return err + } + return nil +} \ No newline at end of file diff --git a/weed/server/filer_server.go b/weed/server/filer_server.go index 959bb92cb..fc7799efc 100644 --- a/weed/server/filer_server.go +++ b/weed/server/filer_server.go @@ -14,6 +14,7 @@ import ( "github.com/chrislusf/seaweedfs/weed/filer/embedded_filer" "github.com/chrislusf/seaweedfs/weed/filer/flat_namespace" "github.com/chrislusf/seaweedfs/weed/filer/mysql_store" + "github.com/chrislusf/seaweedfs/weed/filer/postgres_store" "github.com/chrislusf/seaweedfs/weed/filer/redis_store" "github.com/chrislusf/seaweedfs/weed/glog" "github.com/chrislusf/seaweedfs/weed/security" @@ -24,6 +25,7 @@ import ( type filerConf struct { MysqlConf []mysql_store.MySqlConf `json:"mysql"` mysql_store.ShardingConf + PostgresConf []postgres_store.PostgresConf `json:"postgres"` } func parseConfFile(confPath string) (*filerConf, error) { @@ -86,6 +88,9 @@ func NewFilerServer(r *http.ServeMux, ip string, port int, master string, dir st if setting.MysqlConf != nil && len(setting.MysqlConf) != 0 { mysql_store := mysql_store.NewMysqlStore(setting.MysqlConf, setting.IsSharding, setting.ShardCount) fs.filer = flat_namespace.NewFlatNamespaceFiler(master, mysql_store) + } else if setting.PostgresConf != nil && len(setting.PostgresConf) != 0 { + postgres_store := postgres_store.NewPostgresStore(setting.PostgresConf, setting.IsSharding, setting.ShardCount) + fs.filer = flat_namespace.NewFlatNamespaceFiler(master, postgres_store) } else if cassandra_server != "" { cassandra_store, err := cassandra_store.NewCassandraStore(cassandra_keyspace, cassandra_server) if err != nil {