lotus/extern/sector-storage/stores/index.go

454 lines
10 KiB
Go
Raw Normal View History

2020-03-23 11:40:02 +00:00
package stores
import (
"context"
2020-11-12 09:29:42 +00:00
"errors"
2020-03-23 11:40:02 +00:00
"net/url"
gopath "path"
"sort"
"sync"
2020-05-08 16:08:48 +00:00
"time"
2020-03-23 11:40:02 +00:00
"golang.org/x/xerrors"
2020-09-07 03:49:10 +00:00
"github.com/filecoin-project/go-state-types/abi"
"github.com/filecoin-project/go-state-types/big"
2020-09-30 17:32:19 +00:00
"github.com/filecoin-project/lotus/extern/sector-storage/fsutil"
"github.com/filecoin-project/lotus/extern/sector-storage/storiface"
2020-03-23 11:40:02 +00:00
)
var HeartbeatInterval = 10 * time.Second
var SkippedHeartbeatThresh = HeartbeatInterval * 5
2020-05-08 16:08:48 +00:00
2020-03-23 11:40:02 +00:00
// ID identifies sector storage by UUID. One sector storage should map to one
// filesystem, local or networked / shared by multiple machines
type ID string
type StorageInfo struct {
ID ID
URLs []string // TODO: Support non-http transports
Weight uint64
MaxStorage uint64
2020-03-23 11:40:02 +00:00
CanSeal bool
CanStore bool
2020-05-08 16:08:48 +00:00
}
type HealthReport struct {
2020-07-08 14:58:09 +00:00
Stat fsutil.FsStat
2020-11-12 09:29:42 +00:00
Err string
2020-03-23 11:40:02 +00:00
}
type SectorStorageInfo struct {
ID ID
URLs []string // TODO: Support non-http transports
Weight uint64
CanSeal bool
CanStore bool
Primary bool
}
2020-03-23 11:40:02 +00:00
type SectorIndex interface { // part of storage-miner api
2020-07-08 14:58:09 +00:00
StorageAttach(context.Context, StorageInfo, fsutil.FsStat) error
2020-03-23 11:40:02 +00:00
StorageInfo(context.Context, ID) (StorageInfo, error)
2020-05-08 16:08:48 +00:00
StorageReportHealth(context.Context, ID, HealthReport) error
2020-03-23 11:40:02 +00:00
2020-09-06 16:54:00 +00:00
StorageDeclareSector(ctx context.Context, storageID ID, s abi.SectorID, ft storiface.SectorFileType, primary bool) error
StorageDropSector(ctx context.Context, storageID ID, s abi.SectorID, ft storiface.SectorFileType) error
StorageFindSector(ctx context.Context, sector abi.SectorID, ft storiface.SectorFileType, ssize abi.SectorSize, allowFetch bool) ([]SectorStorageInfo, error)
2020-03-23 11:40:02 +00:00
StorageBestAlloc(ctx context.Context, allocate storiface.SectorFileType, ssize abi.SectorSize, pathType storiface.PathType) ([]StorageInfo, error)
// atomically acquire locks on all sector file types. close ctx to unlock
2020-09-06 16:54:00 +00:00
StorageLock(ctx context.Context, sector abi.SectorID, read storiface.SectorFileType, write storiface.SectorFileType) error
StorageTryLock(ctx context.Context, sector abi.SectorID, read storiface.SectorFileType, write storiface.SectorFileType) (bool, error)
2020-03-23 11:40:02 +00:00
}
type Decl struct {
abi.SectorID
2020-09-06 16:54:00 +00:00
storiface.SectorFileType
2020-03-23 11:40:02 +00:00
}
type declMeta struct {
storage ID
primary bool
}
2020-03-23 11:40:02 +00:00
type storageEntry struct {
info *StorageInfo
2020-07-08 14:58:09 +00:00
fsi fsutil.FsStat
2020-05-08 16:08:48 +00:00
lastHeartbeat time.Time
heartbeatErr error
2020-03-23 11:40:02 +00:00
}
type Index struct {
*indexLocks
2020-03-23 11:40:02 +00:00
lk sync.RWMutex
sectors map[Decl][]*declMeta
2020-03-23 11:40:02 +00:00
stores map[ID]*storageEntry
}
func NewIndex() *Index {
return &Index{
indexLocks: &indexLocks{
locks: map[abi.SectorID]*sectorLock{},
},
2020-06-03 20:00:34 +00:00
sectors: map[Decl][]*declMeta{},
stores: map[ID]*storageEntry{},
2020-03-23 11:40:02 +00:00
}
}
func (i *Index) StorageList(ctx context.Context) (map[ID][]Decl, error) {
2020-04-29 12:13:21 +00:00
i.lk.RLock()
defer i.lk.RUnlock()
2020-09-06 16:54:00 +00:00
byID := map[ID]map[abi.SectorID]storiface.SectorFileType{}
2020-03-23 11:40:02 +00:00
for id := range i.stores {
2020-09-06 16:54:00 +00:00
byID[id] = map[abi.SectorID]storiface.SectorFileType{}
2020-03-23 11:40:02 +00:00
}
for decl, ids := range i.sectors {
for _, id := range ids {
byID[id.storage][decl.SectorID] |= decl.SectorFileType
2020-03-23 11:40:02 +00:00
}
}
out := map[ID][]Decl{}
for id, m := range byID {
out[id] = []Decl{}
for sectorID, fileType := range m {
out[id] = append(out[id], Decl{
SectorID: sectorID,
SectorFileType: fileType,
})
}
}
return out, nil
}
2020-07-08 14:58:09 +00:00
func (i *Index) StorageAttach(ctx context.Context, si StorageInfo, st fsutil.FsStat) error {
2020-03-23 11:40:02 +00:00
i.lk.Lock()
defer i.lk.Unlock()
log.Infof("New sector storage: %s", si.ID)
if _, ok := i.stores[si.ID]; ok {
for _, u := range si.URLs {
if _, err := url.Parse(u); err != nil {
return xerrors.Errorf("failed to parse url %s: %w", si.URLs, err)
}
}
uloop:
for _, u := range si.URLs {
for _, l := range i.stores[si.ID].info.URLs {
if u == l {
continue uloop
}
}
i.stores[si.ID].info.URLs = append(i.stores[si.ID].info.URLs, u)
}
i.stores[si.ID].info.Weight = si.Weight
i.stores[si.ID].info.MaxStorage = si.MaxStorage
i.stores[si.ID].info.CanSeal = si.CanSeal
i.stores[si.ID].info.CanStore = si.CanStore
2020-03-23 11:40:02 +00:00
return nil
}
i.stores[si.ID] = &storageEntry{
info: &si,
fsi: st,
2020-05-08 16:08:48 +00:00
lastHeartbeat: time.Now(),
2020-03-23 11:40:02 +00:00
}
return nil
}
2020-05-08 16:08:48 +00:00
func (i *Index) StorageReportHealth(ctx context.Context, id ID, report HealthReport) error {
i.lk.Lock()
defer i.lk.Unlock()
ent, ok := i.stores[id]
if !ok {
return xerrors.Errorf("health report for unknown storage: %s", id)
}
ent.fsi = report.Stat
2020-11-12 09:29:42 +00:00
if report.Err != "" {
ent.heartbeatErr = errors.New(report.Err)
}
2020-05-08 16:08:48 +00:00
ent.lastHeartbeat = time.Now()
return nil
}
2020-09-06 16:54:00 +00:00
func (i *Index) StorageDeclareSector(ctx context.Context, storageID ID, s abi.SectorID, ft storiface.SectorFileType, primary bool) error {
2020-03-23 11:40:02 +00:00
i.lk.Lock()
defer i.lk.Unlock()
loop:
2020-09-06 16:54:00 +00:00
for _, fileType := range storiface.PathTypes {
2020-03-23 11:40:02 +00:00
if fileType&ft == 0 {
continue
}
d := Decl{s, fileType}
for _, sid := range i.sectors[d] {
2020-08-16 10:40:35 +00:00
if sid.storage == storageID {
if !sid.primary && primary {
sid.primary = true
} else {
2020-08-16 10:40:35 +00:00
log.Warnf("sector %v redeclared in %s", s, storageID)
}
continue loop
2020-03-23 11:40:02 +00:00
}
}
i.sectors[d] = append(i.sectors[d], &declMeta{
2020-08-16 10:40:35 +00:00
storage: storageID,
primary: primary,
})
2020-03-23 11:40:02 +00:00
}
return nil
}
2020-09-06 16:54:00 +00:00
func (i *Index) StorageDropSector(ctx context.Context, storageID ID, s abi.SectorID, ft storiface.SectorFileType) error {
2020-03-23 11:40:02 +00:00
i.lk.Lock()
defer i.lk.Unlock()
2020-09-06 16:54:00 +00:00
for _, fileType := range storiface.PathTypes {
2020-03-23 11:40:02 +00:00
if fileType&ft == 0 {
continue
}
d := Decl{s, fileType}
if len(i.sectors[d]) == 0 {
2021-02-18 04:00:28 +00:00
continue
2020-03-23 11:40:02 +00:00
}
rewritten := make([]*declMeta, 0, len(i.sectors[d])-1)
2020-03-23 11:40:02 +00:00
for _, sid := range i.sectors[d] {
2020-08-16 10:40:35 +00:00
if sid.storage == storageID {
2020-03-23 11:40:02 +00:00
continue
}
rewritten = append(rewritten, sid)
}
if len(rewritten) == 0 {
delete(i.sectors, d)
2021-02-18 04:00:28 +00:00
continue
2020-03-23 11:40:02 +00:00
}
i.sectors[d] = rewritten
}
return nil
}
func (i *Index) StorageFindSector(ctx context.Context, s abi.SectorID, ft storiface.SectorFileType, ssize abi.SectorSize, allowFetch bool) ([]SectorStorageInfo, error) {
2020-03-23 11:40:02 +00:00
i.lk.RLock()
defer i.lk.RUnlock()
storageIDs := map[ID]uint64{}
isprimary := map[ID]bool{}
2020-03-23 11:40:02 +00:00
2020-09-06 16:54:00 +00:00
for _, pathType := range storiface.PathTypes {
2020-03-23 11:40:02 +00:00
if ft&pathType == 0 {
continue
}
for _, id := range i.sectors[Decl{s, pathType}] {
storageIDs[id.storage]++
isprimary[id.storage] = isprimary[id.storage] || id.primary
2020-03-23 11:40:02 +00:00
}
}
out := make([]SectorStorageInfo, 0, len(storageIDs))
2020-03-23 11:40:02 +00:00
for id, n := range storageIDs {
st, ok := i.stores[id]
if !ok {
log.Warnf("storage %s is not present in sector index (referenced by sector %v)", id, s)
continue
}
urls := make([]string, len(st.info.URLs))
for k, u := range st.info.URLs {
rl, err := url.Parse(u)
if err != nil {
return nil, xerrors.Errorf("failed to parse url: %w", err)
}
2020-09-06 16:54:00 +00:00
rl.Path = gopath.Join(rl.Path, ft.String(), storiface.SectorName(s))
2020-03-23 11:40:02 +00:00
urls[k] = rl.String()
}
out = append(out, SectorStorageInfo{
2020-05-26 08:25:29 +00:00
ID: id,
URLs: urls,
Weight: st.info.Weight * n, // storage with more sector types is better
2020-03-23 11:40:02 +00:00
CanSeal: st.info.CanSeal,
CanStore: st.info.CanStore,
Primary: isprimary[id],
2020-03-23 11:40:02 +00:00
})
}
if allowFetch {
spaceReq, err := ft.SealSpaceUse(ssize)
2020-08-11 07:27:03 +00:00
if err != nil {
return nil, xerrors.Errorf("estimating required space: %w", err)
}
2020-03-23 11:40:02 +00:00
for id, st := range i.stores {
2020-08-11 07:27:03 +00:00
if !st.info.CanSeal {
continue
}
if spaceReq > uint64(st.fsi.Available) {
log.Debugf("not selecting on %s, out of space (available: %d, need: %d)", st.info.ID, st.fsi.Available, spaceReq)
continue
}
if time.Since(st.lastHeartbeat) > SkippedHeartbeatThresh {
log.Debugf("not selecting on %s, didn't receive heartbeats for %s", st.info.ID, time.Since(st.lastHeartbeat))
continue
}
if st.heartbeatErr != nil {
log.Debugf("not selecting on %s, heartbeat error: %s", st.info.ID, st.heartbeatErr)
continue
}
2020-03-23 11:40:02 +00:00
if _, ok := storageIDs[id]; ok {
continue
}
urls := make([]string, len(st.info.URLs))
for k, u := range st.info.URLs {
rl, err := url.Parse(u)
if err != nil {
return nil, xerrors.Errorf("failed to parse url: %w", err)
}
2020-09-06 16:54:00 +00:00
rl.Path = gopath.Join(rl.Path, ft.String(), storiface.SectorName(s))
2020-03-23 11:40:02 +00:00
urls[k] = rl.String()
}
out = append(out, SectorStorageInfo{
2020-05-26 08:25:29 +00:00
ID: id,
URLs: urls,
Weight: st.info.Weight * 0, // TODO: something better than just '0'
2020-03-23 11:40:02 +00:00
CanSeal: st.info.CanSeal,
CanStore: st.info.CanStore,
Primary: false,
2020-03-23 11:40:02 +00:00
})
}
}
return out, nil
}
func (i *Index) StorageInfo(ctx context.Context, id ID) (StorageInfo, error) {
i.lk.RLock()
defer i.lk.RUnlock()
si, found := i.stores[id]
if !found {
return StorageInfo{}, xerrors.Errorf("sector store not found")
}
return *si.info, nil
}
func (i *Index) StorageBestAlloc(ctx context.Context, allocate storiface.SectorFileType, ssize abi.SectorSize, pathType storiface.PathType) ([]StorageInfo, error) {
2020-03-23 11:40:02 +00:00
i.lk.RLock()
defer i.lk.RUnlock()
var candidates []storageEntry
spaceReq, err := allocate.SealSpaceUse(ssize)
if err != nil {
return nil, xerrors.Errorf("estimating required space: %w", err)
}
2020-03-23 11:40:02 +00:00
for _, p := range i.stores {
2020-09-06 16:54:00 +00:00
if (pathType == storiface.PathSealing) && !p.info.CanSeal {
2020-03-23 11:40:02 +00:00
continue
}
2020-09-06 16:54:00 +00:00
if (pathType == storiface.PathStorage) && !p.info.CanStore {
2020-03-23 11:40:02 +00:00
continue
}
2020-07-06 16:36:44 +00:00
if spaceReq > uint64(p.fsi.Available) {
log.Debugf("not allocating on %s, out of space (available: %d, need: %d)", p.info.ID, p.fsi.Available, spaceReq)
continue
}
if time.Since(p.lastHeartbeat) > SkippedHeartbeatThresh {
log.Debugf("not allocating on %s, didn't receive heartbeats for %s", p.info.ID, time.Since(p.lastHeartbeat))
continue
}
if p.heartbeatErr != nil {
log.Debugf("not allocating on %s, heartbeat error: %s", p.info.ID, p.heartbeatErr)
continue
}
2020-03-23 11:40:02 +00:00
candidates = append(candidates, *p)
}
if len(candidates) == 0 {
return nil, xerrors.New("no good path found")
}
sort.Slice(candidates, func(i, j int) bool {
2020-08-16 10:40:35 +00:00
iw := big.Mul(big.NewInt(candidates[i].fsi.Available), big.NewInt(int64(candidates[i].info.Weight)))
jw := big.Mul(big.NewInt(candidates[j].fsi.Available), big.NewInt(int64(candidates[j].info.Weight)))
2020-03-23 11:40:02 +00:00
return iw.GreaterThan(jw)
})
out := make([]StorageInfo, len(candidates))
for i, candidate := range candidates {
out[i] = *candidate.info
}
return out, nil
}
2020-09-06 16:54:00 +00:00
func (i *Index) FindSector(id abi.SectorID, typ storiface.SectorFileType) ([]ID, error) {
2020-03-23 11:40:02 +00:00
i.lk.RLock()
defer i.lk.RUnlock()
f, ok := i.sectors[Decl{
2020-03-23 11:40:02 +00:00
SectorID: id,
SectorFileType: typ,
}]
if !ok {
return nil, nil
}
out := make([]ID, 0, len(f))
for _, meta := range f {
out = append(out, meta.storage)
}
return out, nil
2020-03-23 11:40:02 +00:00
}
var _ SectorIndex = &Index{}