seaweedfs/weed/filer/mysql_store/mysql_store.go

package mysql_store

import (
	"database/sql"
	"fmt"
	"hash/crc32"
	"sync"
	"time"

	_ "github.com/go-sql-driver/mysql"
)

const (
	sqlUrl                     = "%s:%s@tcp(%s:%d)/%s?charset=utf8"
	default_maxIdleConnections = 100
	default_maxOpenConnections = 50
	default_maxTableNums       = 1024
	tableName                  = "filer_mapping"
)

var (
	_init_db        sync.Once
	_db_connections []*sql.DB
)

type MySqlConf struct {
	User               string
	Password           string
	HostName           string
	Port               int
	DataBase           string
	MaxIdleConnections int
	MaxOpenConnections int
}

type ShardingConf struct {
	IsSharding  bool `json:"isSharding"`
	ShardingNum int  `json:"shardingNum"`
}

type MySqlStore struct {
	dbs         []*sql.DB
	isSharding  bool
	shardingNum int
}

func getDbConnection(confs []MySqlConf) []*sql.DB {
	_init_db.Do(func() {
		for _, conf := range confs {

			sqlUrl := fmt.Sprintf(sqlUrl, conf.User, conf.Password, conf.HostName, conf.Port, conf.DataBase)
			var dbErr error
			_db_connection, dbErr := sql.Open("mysql", sqlUrl)
			if dbErr != nil {
				_db_connection.Close()
				_db_connection = nil
				panic(dbErr)
			}
			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 NewMysqlStore(confs []MySqlConf, isSharding bool, shardingNum int) *MySqlStore {
	ms := &MySqlStore{
		dbs:         getDbConnection(confs),
		isSharding:  isSharding,
		shardingNum: shardingNum,
	}

	for _, db := range ms.dbs {
		if !isSharding {
			ms.shardingNum = 1
		} else {
			if ms.shardingNum == 0 {
				ms.shardingNum = default_maxTableNums
			}
		}
		for i := 0; i < ms.shardingNum; i++ {
			if err := ms.createTables(db, tableName, i); err != nil {
				fmt.Printf("create table failed %v", err)
			}
		}
	}

	return ms
}

func (s *MySqlStore) 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.shardingNum
	return
}

func (s *MySqlStore) 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 *MySqlStore) Get(fullFilePath string) (fid string, err error) {
	instance_offset, tableFullName, err := s.parseFilerMappingInfo(fullFilePath)
	if err != nil {
		return "", fmt.Errorf("MySqlStore Get operation can not parse file path %s: err is %v", fullFilePath, err)
	}
	fid, err = s.query(fullFilePath, s.dbs[instance_offset], tableFullName)
	if err == sql.ErrNoRows {
		//Could not found
		err = nil
	}
	return fid, err
}

func (s *MySqlStore) Put(fullFilePath string, fid string) (err error) {
	var tableFullName string

	instance_offset, tableFullName, err := s.parseFilerMappingInfo(fullFilePath)
	if err != nil {
		return fmt.Errorf("MySqlStore 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("MySqlStore 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)
			err = fmt.Errorf("MySqlStore 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)
			err = fmt.Errorf("MySqlStore Put operation failed when updating path %s with fid %s : err is %v", fullFilePath, fid, err)
		}
	}
	return
}

func (s *MySqlStore) Delete(fullFilePath string) (err error) {
	var fid string
	instance_offset, tableFullName, err := s.parseFilerMappingInfo(fullFilePath)
	if err != nil {
		return fmt.Errorf("MySqlStore 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("MySqlStore 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("MySqlStore Delete operation failed when deleting path %s: err is %v", fullFilePath, err)
	} else {
		return nil
	}
}

func (s *MySqlStore) Close() {
	for _, db := range s.dbs {
		db.Close()
	}
}

var createTable = `
CREATE TABLE IF NOT EXISTS %s (
  id bigint(20) NOT NULL AUTO_INCREMENT,
  uriPath char(256) NOT NULL DEFAULT "" COMMENT 'http uriPath',
  fid char(36) NOT NULL DEFAULT "" COMMENT 'seaweedfs fid',
  createTime int(10) NOT NULL DEFAULT 0 COMMENT 'createdTime in unix timestamp',
  updateTime int(10) NOT NULL DEFAULT 0 COMMENT 'updatedTime in unix timestamp',
  remark varchar(20) NOT NULL DEFAULT "" COMMENT 'reserverd field',
  status tinyint(2) DEFAULT '1' COMMENT 'resource status',
  PRIMARY KEY (id),
  UNIQUE KEY index_uriPath (uriPath)
) DEFAULT CHARSET=utf8;
`

func (s *MySqlStore) createTables(db *sql.DB, tableName string, postfix int) error {
	var realTableName string
	if s.isSharding {
		realTableName = fmt.Sprintf("%s_%4d", tableName, postfix)
	} else {
		realTableName = tableName
	}

	stmt, err := db.Prepare(fmt.Sprintf(createTable, realTableName))
	if err != nil {
		return err
	}
	defer stmt.Close()

	_, err = stmt.Exec()
	if err != nil {
		return err
	}
	return nil
}

func (s *MySqlStore) query(uriPath string, db *sql.DB, tableName string) (string, error) {
	sqlStatement := "SELECT fid FROM %s WHERE uriPath=?"
	row := db.QueryRow(fmt.Sprintf(sqlStatement, tableName), uriPath)
	var fid string
	err := row.Scan(&fid)
	if err != nil {
		return "", err
	}
	return fid, nil
}

func (s *MySqlStore) update(uriPath string, fid string, db *sql.DB, tableName string) error {
	sqlStatement := "UPDATE %s SET fid=?, updateTime=? WHERE uriPath=?"
	res, err := db.Exec(fmt.Sprintf(sqlStatement, tableName), fid, time.Now().Unix(), uriPath)
	if err != nil {
		return err
	}

	_, err = res.RowsAffected()
	if err != nil {
		return err
	}
	return nil
}

func (s *MySqlStore) insert(uriPath string, fid string, db *sql.DB, tableName string) error {
	sqlStatement := "INSERT INTO %s (uriPath,fid,createTime) VALUES(?,?,?)"
	res, err := db.Exec(fmt.Sprintf(sqlStatement, tableName), uriPath, fid, time.Now().Unix())
	if err != nil {
		return err
	}

	_, err = res.RowsAffected()
	if err != nil {
		return err
	}
	return nil
}

func (s *MySqlStore) delete(uriPath string, db *sql.DB, tableName string) error {
	sqlStatement := "DELETE FROM %s WHERE uriPath=?"
	res, err := db.Exec(fmt.Sprintf(sqlStatement, tableName), uriPath)
	if err != nil {
		return err
	}

	_, err = res.RowsAffected()
	if err != nil {
		return err
	}
	return nil
}