Chris Lu
7 years ago
5 changed files with 188 additions and 16 deletions
-
121weed/filer2/abstract_sql/abstract_sql_store.go
-
15weed/filer2/configuration.go
-
3weed/filer2/memdb/memdb_store.go
-
54weed/filer2/mysql/mysql_store.go
-
11weed/server/filer_server.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 |
|||
} |
@ -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 |
|||
} |
Write
Preview
Loading…
Cancel
Save
Reference in new issue