seaweedfs/go/storage/store.go

377 lines
11 KiB
Go
Raw Normal View History

package storage
import (
"encoding/json"
"errors"
2013-07-04 05:14:16 +00:00
"fmt"
2012-09-13 08:33:47 +00:00
"io/ioutil"
"math/rand"
"strconv"
"strings"
"github.com/chrislusf/weed-fs/go/glog"
"github.com/chrislusf/weed-fs/go/operation"
2015-02-07 23:35:28 +00:00
"github.com/chrislusf/weed-fs/go/security"
"github.com/chrislusf/weed-fs/go/util"
"github.com/golang/protobuf/proto"
)
const (
MAX_TTL_VOLUME_REMOVAL_DELAY = 10 // 10 minutes
)
type DiskLocation struct {
2014-03-26 20:22:27 +00:00
Directory string
MaxVolumeCount int
volumes map[VolumeId]*Volume
}
2014-03-26 20:22:27 +00:00
func (mn *DiskLocation) reset() {
}
type MasterNodes struct {
nodes []string
lastNode int
}
func (mn *MasterNodes) String() string {
return fmt.Sprintf("nodes:%v, lastNode:%d", mn.nodes, mn.lastNode)
}
func NewMasterNodes(bootstrapNode string) (mn *MasterNodes) {
mn = &MasterNodes{nodes: []string{bootstrapNode}, lastNode: -1}
return
}
func (mn *MasterNodes) reset() {
if len(mn.nodes) > 1 && mn.lastNode > 0 {
mn.lastNode = -mn.lastNode
}
}
func (mn *MasterNodes) findMaster() (string, error) {
if len(mn.nodes) == 0 {
return "", errors.New("No master node found!")
}
if mn.lastNode < 0 {
for _, m := range mn.nodes {
if masters, e := operation.ListMasters(m); e == nil {
if len(masters) == 0 {
continue
}
mn.nodes = masters
mn.lastNode = rand.Intn(len(mn.nodes))
glog.V(2).Info("current master node is :", mn.nodes[mn.lastNode])
break
}
}
}
if mn.lastNode < 0 {
2015-01-13 10:46:56 +00:00
return "", errors.New("No master node available!")
}
return mn.nodes[mn.lastNode], nil
}
/*
* A VolumeServer contains one Store
*/
type Store struct {
Ip string
Port int
PublicUrl string
2014-03-26 20:22:27 +00:00
Locations []*DiskLocation
dataCenter string //optional informaton, overwriting master setting if exists
rack string //optional information, overwriting master setting if exists
connected bool
volumeSizeLimit uint64 //read from the master
masterNodes *MasterNodes
}
func (s *Store) String() (str string) {
str = fmt.Sprintf("Ip:%s, Port:%d, PublicUrl:%s, dataCenter:%s, rack:%s, connected:%v, volumeSizeLimit:%d, masterNodes:%s", s.Ip, s.Port, s.PublicUrl, s.dataCenter, s.rack, s.connected, s.volumeSizeLimit, s.masterNodes)
return
}
func NewStore(port int, ip, publicUrl string, dirnames []string, maxVolumeCounts []int, useLevelDb bool) (s *Store) {
s = &Store{Port: port, Ip: ip, PublicUrl: publicUrl}
2014-03-26 20:22:27 +00:00
s.Locations = make([]*DiskLocation, 0)
for i := 0; i < len(dirnames); i++ {
2014-03-26 20:22:27 +00:00
location := &DiskLocation{Directory: dirnames[i], MaxVolumeCount: maxVolumeCounts[i]}
location.volumes = make(map[VolumeId]*Volume)
location.loadExistingVolumes(useLevelDb)
2014-03-26 20:22:27 +00:00
s.Locations = append(s.Locations, location)
}
return
}
func (s *Store) AddVolume(volumeListString string, collection string, useLevelDb bool, replicaPlacement string, ttlString string) error {
rt, e := NewReplicaPlacementFromString(replicaPlacement)
2012-09-26 08:55:56 +00:00
if e != nil {
return e
}
ttl, e := ReadTTL(ttlString)
if e != nil {
return e
}
for _, range_string := range strings.Split(volumeListString, ",") {
if strings.Index(range_string, "-") < 0 {
id_string := range_string
2012-11-07 09:51:43 +00:00
id, err := NewVolumeId(id_string)
if err != nil {
2013-07-04 05:14:16 +00:00
return fmt.Errorf("Volume Id %s is not a valid unsigned integer!", id_string)
}
e = s.addVolume(VolumeId(id), collection, useLevelDb, rt, ttl)
} else {
pair := strings.Split(range_string, "-")
start, start_err := strconv.ParseUint(pair[0], 10, 64)
if start_err != nil {
2013-07-04 05:14:16 +00:00
return fmt.Errorf("Volume Start Id %s is not a valid unsigned integer!", pair[0])
}
end, end_err := strconv.ParseUint(pair[1], 10, 64)
if end_err != nil {
return fmt.Errorf("Volume End Id %s is not a valid unsigned integer!", pair[1])
}
for id := start; id <= end; id++ {
if err := s.addVolume(VolumeId(id), collection, useLevelDb, rt, ttl); err != nil {
2012-09-13 08:33:47 +00:00
e = err
2012-09-13 07:04:56 +00:00
}
}
}
}
2012-09-13 07:04:56 +00:00
return e
}
func (s *Store) DeleteCollection(collection string) (e error) {
2014-03-26 20:22:27 +00:00
for _, location := range s.Locations {
for k, v := range location.volumes {
if v.Collection == collection {
e = v.Destroy()
if e != nil {
return
}
delete(location.volumes, k)
}
}
}
return
}
func (s *Store) DeleteVolume(volumes map[VolumeId]*Volume, v *Volume) (e error) {
e = v.Destroy()
if e != nil {
return
}
delete(volumes, v.Id)
return
}
func (s *Store) findVolume(vid VolumeId) *Volume {
2014-03-26 20:22:27 +00:00
for _, location := range s.Locations {
if v, found := location.volumes[vid]; found {
return v
}
}
return nil
}
func (s *Store) findFreeLocation() (ret *DiskLocation) {
max := 0
2014-03-26 20:22:27 +00:00
for _, location := range s.Locations {
currentFreeCount := location.MaxVolumeCount - len(location.volumes)
if currentFreeCount > max {
max = currentFreeCount
ret = location
}
}
return ret
}
func (s *Store) addVolume(vid VolumeId, collection string, useLevelDb bool, replicaPlacement *ReplicaPlacement, ttl *TTL) error {
if s.findVolume(vid) != nil {
return fmt.Errorf("Volume Id %d already exists!", vid)
}
if location := s.findFreeLocation(); location != nil {
glog.V(0).Infof("In dir %s adds volume:%v collection:%s replicaPlacement:%v ttl:%v",
location.Directory, vid, collection, replicaPlacement, ttl)
if volume, err := NewVolume(location.Directory, collection, vid, useLevelDb, replicaPlacement, ttl); err == nil {
location.volumes[vid] = volume
return nil
2013-07-20 03:38:00 +00:00
} else {
return err
2013-07-20 03:38:00 +00:00
}
}
return fmt.Errorf("No more free space left")
}
2012-11-07 09:51:43 +00:00
func (l *DiskLocation) loadExistingVolumes(useLevelDb bool) {
2014-03-26 20:22:27 +00:00
if dirs, err := ioutil.ReadDir(l.Directory); err == nil {
2012-09-13 08:33:47 +00:00
for _, dir := range dirs {
name := dir.Name()
if !dir.IsDir() && strings.HasSuffix(name, ".dat") {
2013-11-12 10:21:22 +00:00
collection := ""
2012-09-13 08:33:47 +00:00
base := name[:len(name)-len(".dat")]
2013-11-12 10:21:22 +00:00
i := strings.Index(base, "_")
if i > 0 {
collection, base = base[0:i], base[i+1:]
}
2012-09-13 08:33:47 +00:00
if vid, err := NewVolumeId(base); err == nil {
if l.volumes[vid] == nil {
if v, e := NewVolume(l.Directory, collection, vid, useLevelDb, nil, nil); e == nil {
l.volumes[vid] = v
glog.V(0).Infof("data file %s, replicaPlacement=%s v=%d size=%d ttl=%s", l.Directory+"/"+name, v.ReplicaPlacement, v.Version(), v.Size(), v.Ttl.String())
2013-01-17 08:56:56 +00:00
}
2012-09-13 08:33:47 +00:00
}
}
}
}
}
2014-03-26 20:22:27 +00:00
glog.V(0).Infoln("Store started on dir:", l.Directory, "with", len(l.volumes), "volumes", "max", l.MaxVolumeCount)
2012-09-13 08:33:47 +00:00
}
func (s *Store) Status() []*VolumeInfo {
var stats []*VolumeInfo
2014-03-26 20:22:27 +00:00
for _, location := range s.Locations {
for k, v := range location.volumes {
s := &VolumeInfo{Id: VolumeId(k), Size: v.ContentSize(),
2013-11-12 10:21:22 +00:00
Collection: v.Collection,
ReplicaPlacement: v.ReplicaPlacement,
2013-11-12 10:21:22 +00:00
Version: v.Version(),
FileCount: v.nm.FileCount(),
DeleteCount: v.nm.DeletedCount(),
DeletedByteCount: v.nm.DeletedSize(),
2015-03-22 19:50:04 +00:00
ReadOnly: v.readOnly,
Ttl: v.Ttl}
stats = append(stats, s)
}
}
return stats
}
func (s *Store) SetDataCenter(dataCenter string) {
s.dataCenter = dataCenter
}
func (s *Store) SetRack(rack string) {
s.rack = rack
}
func (s *Store) SetBootstrapMaster(bootstrapMaster string) {
s.masterNodes = NewMasterNodes(bootstrapMaster)
}
func (s *Store) SendHeartbeatToMaster() (masterNode string, secretKey security.Secret, e error) {
masterNode, e = s.masterNodes.findMaster()
if e != nil {
return
}
var volumeMessages []*operation.VolumeInformationMessage
maxVolumeCount := 0
var maxFileKey uint64
2014-03-26 20:22:27 +00:00
for _, location := range s.Locations {
maxVolumeCount = maxVolumeCount + location.MaxVolumeCount
for k, v := range location.volumes {
if maxFileKey < v.nm.MaxFileKey() {
maxFileKey = v.nm.MaxFileKey()
}
if !v.expired(s.volumeSizeLimit) {
volumeMessage := &operation.VolumeInformationMessage{
Id: proto.Uint32(uint32(k)),
Size: proto.Uint64(uint64(v.Size())),
Collection: proto.String(v.Collection),
FileCount: proto.Uint64(uint64(v.nm.FileCount())),
DeleteCount: proto.Uint64(uint64(v.nm.DeletedCount())),
DeletedByteCount: proto.Uint64(v.nm.DeletedSize()),
ReadOnly: proto.Bool(v.readOnly),
ReplicaPlacement: proto.Uint32(uint32(v.ReplicaPlacement.Byte())),
Version: proto.Uint32(uint32(v.Version())),
Ttl: proto.Uint32(v.Ttl.ToUint32()),
}
volumeMessages = append(volumeMessages, volumeMessage)
} else {
if v.exiredLongEnough(MAX_TTL_VOLUME_REMOVAL_DELAY) {
s.DeleteVolume(location.volumes, v)
glog.V(0).Infoln("volume", v.Id, "is deleted.")
} else {
glog.V(0).Infoln("volume", v.Id, "is expired.")
}
}
}
}
joinMessage := &operation.JoinMessage{
IsInit: proto.Bool(!s.connected),
Ip: proto.String(s.Ip),
Port: proto.Uint32(uint32(s.Port)),
PublicUrl: proto.String(s.PublicUrl),
MaxVolumeCount: proto.Uint32(uint32(maxVolumeCount)),
MaxFileKey: proto.Uint64(maxFileKey),
DataCenter: proto.String(s.dataCenter),
Rack: proto.String(s.rack),
Volumes: volumeMessages,
}
data, err := proto.Marshal(joinMessage)
if err != nil {
2015-02-07 23:35:28 +00:00
return "", "", err
}
joinUrl := "http://" + masterNode + "/dir/join"
jsonBlob, err := util.PostBytes(joinUrl, data)
if err != nil {
s.masterNodes.reset()
2015-02-07 23:35:28 +00:00
return "", "", err
}
2014-04-17 00:29:58 +00:00
var ret operation.JoinResult
if err := json.Unmarshal(jsonBlob, &ret); err != nil {
glog.V(0).Infof("Failed to join %s with response: %s", joinUrl, string(jsonBlob))
2015-02-07 23:35:28 +00:00
return masterNode, "", err
}
2014-04-17 00:29:58 +00:00
if ret.Error != "" {
2015-02-07 23:35:28 +00:00
return masterNode, "", errors.New(ret.Error)
2014-04-17 00:29:58 +00:00
}
s.volumeSizeLimit = ret.VolumeSizeLimit
2015-02-07 23:35:28 +00:00
secretKey = security.Secret(ret.SecretKey)
s.connected = true
return
}
func (s *Store) Close() {
2014-03-26 20:22:27 +00:00
for _, location := range s.Locations {
for _, v := range location.volumes {
v.Close()
}
}
}
func (s *Store) Write(i VolumeId, n *Needle) (size uint32, err error) {
if v := s.findVolume(i); v != nil {
2013-04-15 02:34:37 +00:00
if v.readOnly {
2015-03-10 07:20:31 +00:00
err = fmt.Errorf("Volume %d is read only", i)
return
2015-03-10 07:20:31 +00:00
}
if MaxPossibleVolumeSize >= v.ContentSize()+uint64(size) {
size, err = v.write(n)
2013-04-15 02:34:37 +00:00
} else {
2015-03-10 07:20:31 +00:00
err = fmt.Errorf("Volume Size Limit %d Exceeded! Current size is %d", s.volumeSizeLimit, v.ContentSize())
}
if s.volumeSizeLimit < v.ContentSize()+3*uint64(size) {
glog.V(0).Infoln("volume", i, "size", v.ContentSize(), "will exceed limit", s.volumeSizeLimit)
if _, _, e := s.SendHeartbeatToMaster(); e != nil {
2015-03-10 07:20:31 +00:00
glog.V(0).Infoln("error when reporting size:", e)
2013-02-27 06:54:22 +00:00
}
}
return
2012-09-11 00:08:52 +00:00
}
glog.V(0).Infoln("volume", i, "not found!")
err = fmt.Errorf("Volume %d not found!", i)
return
}
func (s *Store) Delete(i VolumeId, n *Needle) (uint32, error) {
if v := s.findVolume(i); v != nil && !v.readOnly {
2012-09-11 00:08:52 +00:00
return v.delete(n)
}
return 0, nil
}
2012-08-24 05:46:54 +00:00
func (s *Store) Read(i VolumeId, n *Needle) (int, error) {
if v := s.findVolume(i); v != nil {
2012-09-11 00:08:52 +00:00
return v.read(n)
}
2014-04-14 08:00:09 +00:00
return 0, fmt.Errorf("Volume %v not found!", i)
2012-09-11 00:08:52 +00:00
}
2012-09-21 00:58:29 +00:00
func (s *Store) GetVolume(i VolumeId) *Volume {
return s.findVolume(i)
2012-09-21 00:58:29 +00:00
}
2012-09-11 00:08:52 +00:00
func (s *Store) HasVolume(i VolumeId) bool {
v := s.findVolume(i)
return v != nil
}