From 68bcaff14d6630c4549474125f631d9093502828 Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Sat, 26 May 2018 05:32:15 -0700 Subject: [PATCH] mysql can compile, not yet tested! --- .../filer2/abstract_sql/abstract_sql_store.go | 121 ++++++++++++++++++ weed/filer2/configuration.go | 15 ++- weed/filer2/memdb/memdb_store.go | 3 - weed/filer2/mysql/mysql_store.go | 54 ++++++++ weed/server/filer_server.go | 11 +- 5 files changed, 188 insertions(+), 16 deletions(-) create mode 100644 weed/filer2/abstract_sql/abstract_sql_store.go create mode 100644 weed/filer2/mysql/mysql_store.go diff --git a/weed/filer2/abstract_sql/abstract_sql_store.go b/weed/filer2/abstract_sql/abstract_sql_store.go new file mode 100644 index 000000000..252a3c549 --- /dev/null +++ b/weed/filer2/abstract_sql/abstract_sql_store.go @@ -0,0 +1,121 @@ +package abstract_sql + +import ( + "fmt" + "database/sql" + + "github.com/chrislusf/seaweedfs/weed/filer2" + "github.com/chrislusf/seaweedfs/weed/glog" +) + +type AbstractSqlStore struct { + DB *sql.DB +} + +func (store *AbstractSqlStore) InsertEntry(entry *filer2.Entry) (err error) { + + dir, name := entry.FullPath.DirAndName() + meta, err := entry.EncodeAttributesAndChunks() + if err != nil { + return fmt.Errorf("mysql encode %s: %s", entry.FullPath, err) + } + + res, err := store.DB.Exec("INSERT INTO seaweedfs (directory,name,meta) VALUES(?,?,?)", dir, name, meta) + if err != nil { + return fmt.Errorf("mysql insert %s: %s", entry.FullPath, err) + } + + _, err = res.RowsAffected() + if err != nil { + return fmt.Errorf("mysql insert %s but no rows affected: %s", entry.FullPath, err) + } + return nil +} + +func (store *AbstractSqlStore) UpdateEntry(entry *filer2.Entry) (err error) { + + dir, name := entry.FullPath.DirAndName() + meta, err := entry.EncodeAttributesAndChunks() + if err != nil { + return fmt.Errorf("mysql encode %s: %s", entry.FullPath, err) + } + + res, err := store.DB.Exec("UPDATE seaweedfs SET meta=? WHERE directory=? and name=?", dir, name, meta) + if err != nil { + return fmt.Errorf("mysql update %s: %s", entry.FullPath, err) + } + + _, err = res.RowsAffected() + if err != nil { + return fmt.Errorf("mysql update %s but no rows affected: %s", entry.FullPath, err) + } + return nil +} + +func (store *AbstractSqlStore) FindEntry(fullpath filer2.FullPath) (*filer2.Entry, error) { + + dir, name := fullpath.DirAndName() + row := store.DB.QueryRow("SELECT meta FROM seaweedfs WHERE directory=? and name=?", dir, name) + var data []byte + if err := row.Scan(&data); err != nil { + return nil, fmt.Errorf("mysql read entry %s: %v", fullpath, err) + } + + entry := &filer2.Entry{ + FullPath: fullpath, + } + if err := entry.DecodeAttributesAndChunks(data); err != nil { + return entry, fmt.Errorf("mysql decode %s : %v", entry.FullPath, err) + } + + return entry, nil +} + +func (store *AbstractSqlStore) DeleteEntry(fullpath filer2.FullPath) (*filer2.Entry, error) { + + entry, _ := store.FindEntry(fullpath) + + dir, name := fullpath.DirAndName() + + res, err := store.DB.Exec("DELETE FROM seaweedfs WHERE directory=? and name=?", dir, name) + if err != nil { + return nil, fmt.Errorf("mysql delete %s: %s", fullpath, err) + } + + _, err = res.RowsAffected() + if err != nil { + return nil, fmt.Errorf("mysql delete %s but no rows affected: %s", fullpath, err) + } + + return entry, nil +} + +func (store *AbstractSqlStore) ListDirectoryEntries(fullpath filer2.FullPath, startFileName string, inclusive bool, limit int) (entries []*filer2.Entry, err error) { + + rows, err := store.DB.Query("SELECT NAME, meta FROM seaweedfs WHERE directory=? and name>?", fullpath, startFileName) + if err != nil { + return nil, fmt.Errorf("mysql list %s : %v", fullpath, err) + } + defer rows.Close() + + for rows.Next() { + var name string + var data []byte + if err = rows.Scan(&name, &data); err != nil { + glog.V(0).Infof("mysql scan %s : %v", fullpath, err) + return nil, fmt.Errorf("mysql scan %s: %v", fullpath, err) + } + + entry := &filer2.Entry{ + FullPath: fullpath, + } + if err = entry.DecodeAttributesAndChunks(data); err != nil { + glog.V(0).Infof("mysql scan decode %s : %v", entry.FullPath, err) + return nil, fmt.Errorf("mysql scan decode %s : %v", entry.FullPath, err) + } + + entries = append(entries, entry) + } + + return entries, nil +} diff --git a/weed/filer2/configuration.go b/weed/filer2/configuration.go index 6dabf10b4..caca0a49c 100644 --- a/weed/filer2/configuration.go +++ b/weed/filer2/configuration.go @@ -20,13 +20,20 @@ enabled = false dir = "." # directory to store level db files [mysql] +# also need to manually create seaweedfs table. +# CREATE TABLE IF NOT EXISTS seaweedfs ( +# directory VARCHAR(4096) NOT NULL DEFAULT "" COMMENT 'full path to parent directory', +# name VARCHAR(1024) NOT NULL DEFAULT "" COMMENT 'directory or file name', +# meta BLOB, +# PRIMARY KEY (directory, name), +# ) DEFAULT CHARSET=utf8; enabled = true -server = "192.168.1.1" -port = 8080 -username = "" +server = "localhost" +port = 3306 +username = "root" password = "" database = "" -connection_max_idle = 100 +connection_max_idle = 2 connection_max_open = 100 [postgres] diff --git a/weed/filer2/memdb/memdb_store.go b/weed/filer2/memdb/memdb_store.go index e5b06ce47..452e4970c 100644 --- a/weed/filer2/memdb/memdb_store.go +++ b/weed/filer2/memdb/memdb_store.go @@ -5,7 +5,6 @@ import ( "github.com/google/btree" "strings" "fmt" - "time" "github.com/spf13/viper" ) @@ -36,7 +35,6 @@ func (store *MemDbStore) Initialize(viper *viper.Viper) (err error) { func (store *MemDbStore) InsertEntry(entry *filer2.Entry) (err error) { // println("inserting", entry.FullPath) - entry.Crtime = time.Now() store.tree.ReplaceOrInsert(entryItem{entry}) return nil } @@ -45,7 +43,6 @@ func (store *MemDbStore) UpdateEntry(entry *filer2.Entry) (err error) { if _, err = store.FindEntry(entry.FullPath); err != nil { return fmt.Errorf("no such file %s : %v", entry.FullPath, err) } - entry.Mtime = time.Now() store.tree.ReplaceOrInsert(entryItem{entry}) return nil } diff --git a/weed/filer2/mysql/mysql_store.go b/weed/filer2/mysql/mysql_store.go new file mode 100644 index 000000000..4accc9a9f --- /dev/null +++ b/weed/filer2/mysql/mysql_store.go @@ -0,0 +1,54 @@ +package mysql + +import ( + "fmt" + "database/sql" + + "github.com/chrislusf/seaweedfs/weed/filer2" + "github.com/spf13/viper" + _ "github.com/go-sql-driver/mysql" + "github.com/chrislusf/seaweedfs/weed/filer2/abstract_sql" +) + +const ( + CONNECTION_URL_PATTERN = "%s:%s@tcp(%s:%d)/%s?charset=utf8" +) + +func init() { + filer2.Stores = append(filer2.Stores, &MysqlStore{}) +} + +type MysqlStore struct { + abstract_sql.AbstractSqlStore +} + +func (store *MysqlStore) GetName() string { + return "mysql" +} + +func (store *MysqlStore) Initialize(viper *viper.Viper) (err error) { + return store.initialize( + viper.GetString("username"), + viper.GetString("password"), + viper.GetString("hostname"), + viper.GetString("port"), + viper.GetString("database"), + viper.GetInt("connection_max_idle"), + viper.GetInt("connection_max_open"), + ) +} + +func (store *MysqlStore) initialize(user, password, hostname, port, database string, maxIdle, maxOpen int) (err error) { + sqlUrl := fmt.Sprintf(CONNECTION_URL_PATTERN, user, password, hostname, port, database) + var dbErr error + store.DB, dbErr = sql.Open("mysql", sqlUrl) + if dbErr != nil { + store.DB.Close() + store.DB = nil + return fmt.Errorf("can not connect to %s error:%v", sqlUrl, err) + } + + store.DB.SetMaxIdleConns(maxIdle) + store.DB.SetMaxOpenConns(maxOpen) + return nil +} diff --git a/weed/server/filer_server.go b/weed/server/filer_server.go index 8687edf6d..2e88252e9 100644 --- a/weed/server/filer_server.go +++ b/weed/server/filer_server.go @@ -7,23 +7,16 @@ import ( "sync" "time" - "github.com/chrislusf/seaweedfs/weed/filer/mysql_store" - "github.com/chrislusf/seaweedfs/weed/filer/postgres_store" "github.com/chrislusf/seaweedfs/weed/glog" "github.com/chrislusf/seaweedfs/weed/security" "github.com/chrislusf/seaweedfs/weed/storage" "github.com/chrislusf/seaweedfs/weed/util" "github.com/chrislusf/seaweedfs/weed/filer2" - _ "github.com/chrislusf/seaweedfs/weed/filer2/memdb" _ "github.com/chrislusf/seaweedfs/weed/filer2/leveldb" + _ "github.com/chrislusf/seaweedfs/weed/filer2/memdb" + _ "github.com/chrislusf/seaweedfs/weed/filer2/mysql" ) -type filerConf struct { - MysqlConf []mysql_store.MySqlConf `json:"mysql"` - mysql_store.ShardingConf - PostgresConf *postgres_store.PostgresConf `json:"postgres"` -} - type FilerServer struct { port string master string