aboutsummaryrefslogtreecommitdiff
path: root/weed/topology/topology.go
diff options
context:
space:
mode:
Diffstat (limited to 'weed/topology/topology.go')
-rw-r--r--weed/topology/topology.go123
1 files changed, 105 insertions, 18 deletions
diff --git a/weed/topology/topology.go b/weed/topology/topology.go
index 4242bfa05..993f444a7 100644
--- a/weed/topology/topology.go
+++ b/weed/topology/topology.go
@@ -2,24 +2,33 @@ package topology
import (
"errors"
+ "fmt"
"math/rand"
+ "sync"
"github.com/chrislusf/raft"
+
"github.com/chrislusf/seaweedfs/weed/glog"
"github.com/chrislusf/seaweedfs/weed/pb/master_pb"
"github.com/chrislusf/seaweedfs/weed/sequence"
"github.com/chrislusf/seaweedfs/weed/storage"
+ "github.com/chrislusf/seaweedfs/weed/storage/needle"
+ "github.com/chrislusf/seaweedfs/weed/storage/super_block"
"github.com/chrislusf/seaweedfs/weed/util"
)
type Topology struct {
+ vacuumLockCounter int64
NodeImpl
- collectionMap *util.ConcurrentReadMap
+ collectionMap *util.ConcurrentReadMap
+ ecShardMap map[needle.VolumeId]*EcShardLocations
+ ecShardMapLock sync.RWMutex
pulse int64
- volumeSizeLimit uint64
+ volumeSizeLimit uint64
+ replicationAsMin bool
Sequence sequence.Sequencer
@@ -30,15 +39,17 @@ type Topology struct {
RaftServer raft.Server
}
-func NewTopology(id string, seq sequence.Sequencer, volumeSizeLimit uint64, pulse int) *Topology {
+func NewTopology(id string, seq sequence.Sequencer, volumeSizeLimit uint64, pulse int, replicationAsMin bool) *Topology {
t := &Topology{}
t.id = NodeId(id)
t.nodeType = "Topology"
t.NodeImpl.value = t
t.children = make(map[NodeId]Node)
t.collectionMap = util.NewConcurrentReadMap()
+ t.ecShardMap = make(map[needle.VolumeId]*EcShardLocations)
t.pulse = int64(pulse)
t.volumeSizeLimit = volumeSizeLimit
+ t.replicationAsMin = replicationAsMin
t.Sequence = seq
@@ -50,8 +61,13 @@ func NewTopology(id string, seq sequence.Sequencer, volumeSizeLimit uint64, puls
}
func (t *Topology) IsLeader() bool {
- if leader, e := t.Leader(); e == nil {
- return leader == t.RaftServer.Name()
+ if t.RaftServer != nil {
+ if t.RaftServer.State() == raft.Leader {
+ return true
+ }
+ if t.RaftServer.Leader() == "" {
+ return true
+ }
}
return false
}
@@ -66,13 +82,13 @@ func (t *Topology) Leader() (string, error) {
if l == "" {
// We are a single node cluster, we are the leader
- return t.RaftServer.Name(), errors.New("Raft Server not initialized!")
+ return t.RaftServer.Name(), nil
}
return l, nil
}
-func (t *Topology) Lookup(collection string, vid storage.VolumeId) []*DataNode {
+func (t *Topology) Lookup(collection string, vid needle.VolumeId) (dataNodes []*DataNode) {
//maybe an issue if lots of collections?
if collection == "" {
for _, c := range t.collectionMap.Items() {
@@ -85,14 +101,24 @@ func (t *Topology) Lookup(collection string, vid storage.VolumeId) []*DataNode {
return c.(*Collection).Lookup(vid)
}
}
+
+ if locations, found := t.LookupEcShards(vid); found {
+ for _, loc := range locations.Locations {
+ dataNodes = append(dataNodes, loc...)
+ }
+ return dataNodes
+ }
+
return nil
}
-func (t *Topology) NextVolumeId() storage.VolumeId {
+func (t *Topology) NextVolumeId() (needle.VolumeId, error) {
vid := t.GetMaxVolumeId()
next := vid.Next()
- go t.RaftServer.Do(NewMaxVolumeIdCommand(next))
- return next
+ if _, err := t.RaftServer.Do(NewMaxVolumeIdCommand(next)); err != nil {
+ return 0, err
+ }
+ return next, nil
}
func (t *Topology) HasWritableVolume(option *VolumeGrowOption) bool {
@@ -102,19 +128,43 @@ func (t *Topology) HasWritableVolume(option *VolumeGrowOption) bool {
func (t *Topology) PickForWrite(count uint64, option *VolumeGrowOption) (string, uint64, *DataNode, error) {
vid, count, datanodes, err := t.GetVolumeLayout(option.Collection, option.ReplicaPlacement, option.Ttl).PickForWrite(count, option)
- if err != nil || datanodes.Length() == 0 {
- return "", 0, nil, errors.New("No writable volumes available!")
+ if err != nil {
+ return "", 0, nil, fmt.Errorf("failed to find writable volumes for collection:%s replication:%s ttl:%s error: %v", option.Collection, option.ReplicaPlacement.String(), option.Ttl.String(), err)
+ }
+ if datanodes.Length() == 0 {
+ return "", 0, nil, fmt.Errorf("no writable volumes available for collection:%s replication:%s ttl:%s", option.Collection, option.ReplicaPlacement.String(), option.Ttl.String())
}
- fileId, count := t.Sequence.NextFileId(count)
- return storage.NewFileId(*vid, fileId, rand.Uint32()).String(), count, datanodes.Head(), nil
+ fileId := t.Sequence.NextFileId(count)
+ return needle.NewFileId(*vid, fileId, rand.Uint32()).String(), count, datanodes.Head(), nil
}
-func (t *Topology) GetVolumeLayout(collectionName string, rp *storage.ReplicaPlacement, ttl *storage.TTL) *VolumeLayout {
+func (t *Topology) GetVolumeLayout(collectionName string, rp *super_block.ReplicaPlacement, ttl *needle.TTL) *VolumeLayout {
return t.collectionMap.Get(collectionName, func() interface{} {
- return NewCollection(collectionName, t.volumeSizeLimit)
+ return NewCollection(collectionName, t.volumeSizeLimit, t.replicationAsMin)
}).(*Collection).GetOrCreateVolumeLayout(rp, ttl)
}
+func (t *Topology) ListCollections(includeNormalVolumes, includeEcVolumes bool) (ret []string) {
+
+ mapOfCollections := make(map[string]bool)
+ for _, c := range t.collectionMap.Items() {
+ mapOfCollections[c.(*Collection).Name] = true
+ }
+
+ if includeEcVolumes {
+ t.ecShardMapLock.RLock()
+ for _, ecVolumeLocation := range t.ecShardMap {
+ mapOfCollections[ecVolumeLocation.Collection] = true
+ }
+ t.ecShardMapLock.RUnlock()
+ }
+
+ for k := range mapOfCollections {
+ ret = append(ret, k)
+ }
+ return ret
+}
+
func (t *Topology) FindCollection(collectionName string) (*Collection, bool) {
c, hasCollection := t.collectionMap.Find(collectionName)
if !hasCollection {
@@ -152,6 +202,7 @@ func (t *Topology) GetOrCreateDataCenter(dcName string) *DataCenter {
}
func (t *Topology) SyncDataNodeRegistration(volumes []*master_pb.VolumeInformationMessage, dn *DataNode) (newVolumes, deletedVolumes []storage.VolumeInfo) {
+ // convert into in memory struct storage.VolumeInfo
var volumeInfos []storage.VolumeInfo
for _, v := range volumes {
if vi, err := storage.NewVolumeInfo(v); err == nil {
@@ -160,12 +211,48 @@ func (t *Topology) SyncDataNodeRegistration(volumes []*master_pb.VolumeInformati
glog.V(0).Infof("Fail to convert joined volume information: %v", err)
}
}
- newVolumes, deletedVolumes = dn.UpdateVolumes(volumeInfos)
- for _, v := range volumeInfos {
+ // find out the delta volumes
+ var changedVolumes []storage.VolumeInfo
+ newVolumes, deletedVolumes, changedVolumes = dn.UpdateVolumes(volumeInfos)
+ for _, v := range newVolumes {
t.RegisterVolumeLayout(v, dn)
}
for _, v := range deletedVolumes {
t.UnRegisterVolumeLayout(v, dn)
}
+ for _, v := range changedVolumes {
+ vl := t.GetVolumeLayout(v.Collection, v.ReplicaPlacement, v.Ttl)
+ vl.ensureCorrectWritables(&v)
+ }
+ return
+}
+
+func (t *Topology) IncrementalSyncDataNodeRegistration(newVolumes, deletedVolumes []*master_pb.VolumeShortInformationMessage, dn *DataNode) {
+ var newVis, oldVis []storage.VolumeInfo
+ for _, v := range newVolumes {
+ vi, err := storage.NewVolumeInfoFromShort(v)
+ if err != nil {
+ glog.V(0).Infof("NewVolumeInfoFromShort %v: %v", v, err)
+ continue
+ }
+ newVis = append(newVis, vi)
+ }
+ for _, v := range deletedVolumes {
+ vi, err := storage.NewVolumeInfoFromShort(v)
+ if err != nil {
+ glog.V(0).Infof("NewVolumeInfoFromShort %v: %v", v, err)
+ continue
+ }
+ oldVis = append(oldVis, vi)
+ }
+ dn.DeltaUpdateVolumes(newVis, oldVis)
+
+ for _, vi := range newVis {
+ t.RegisterVolumeLayout(vi, dn)
+ }
+ for _, vi := range oldVis {
+ t.UnRegisterVolumeLayout(vi, dn)
+ }
+
return
}
generated by cgit v1.2.3 (git 2.25.1) at 2025-12-22 11:58:43 +0000