JakeHillion/storj
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

satellite/metabase: remove metaloop package

Browse Source 
We moved everything to segment loop so we can now
remove metaloop from code.

Change-Id: I9bd8d2349e5638d7cdad50f2f313f9bd89a8165c
This commit is contained in:
Michał Niewrzał 2021-07-21 18:47:12 +02:00 committed by Michal Niewrzal
parent 1535bbe673
commit b12d29935a
14 changed files with 32 additions and 1551 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

25
cmd/metabase-verify/verify/progress.go
View File

@ -10,7 +10,7 @@ import (
"go.uber.org/zap" "go.uber.org/zap"
"storj.io/common/memory" "storj.io/common/memory"
"storj.io/storj/satellite/metabase/metaloop" "storj.io/storj/satellite/metabase/segmentloop"
) )
// ProgressObserver counts and prints progress of metabase loop. // ProgressObserver counts and prints progress of metabase loop.
@ -19,7 +19,6 @@ type ProgressObserver struct {
ProgressPrintFrequency int64 ProgressPrintFrequency int64
ObjectCount int64
RemoteSegmentCount int64 RemoteSegmentCount int64
InlineSegmentCount int64 InlineSegmentCount int64
} }
@ -27,7 +26,6 @@ type ProgressObserver struct {
// Report reports the current progress. // Report reports the current progress.
func (progress *ProgressObserver) Report() { func (progress *ProgressObserver) Report() {
progress.Log.Debug("progress", progress.Log.Debug("progress",
zap.Int64("objects", progress.ObjectCount),
zap.Int64("remote segments", progress.RemoteSegmentCount), zap.Int64("remote segments", progress.RemoteSegmentCount),
zap.Int64("inline segments", progress.InlineSegmentCount), zap.Int64("inline segments", progress.InlineSegmentCount),
) )
@ -42,28 +40,25 @@ func (progress *ProgressObserver) Report() {
) )
} }
// Object implements the Observer interface. // RemoteSegment implements the Observer interface.
func (progress *ProgressObserver) Object(context.Context, *metaloop.Object) error { func (progress *ProgressObserver) RemoteSegment(context.Context, *segmentloop.Segment) error {
progress.ObjectCount++ progress.RemoteSegmentCount++
if progress.ObjectCount%progress.ProgressPrintFrequency == 0 { if (progress.RemoteSegmentCount+progress.InlineSegmentCount)%progress.ProgressPrintFrequency == 0 {
progress.Report() progress.Report()
} }
return nil return nil
} }
// RemoteSegment implements the Observer interface.
func (progress *ProgressObserver) RemoteSegment(context.Context, *metaloop.Segment) error {
progress.RemoteSegmentCount++
return nil
}
// InlineSegment implements the Observer interface. // InlineSegment implements the Observer interface.
func (progress *ProgressObserver) InlineSegment(context.Context, *metaloop.Segment) error { func (progress *ProgressObserver) InlineSegment(context.Context, *segmentloop.Segment) error {
progress.InlineSegmentCount++ progress.InlineSegmentCount++
if (progress.RemoteSegmentCount+progress.InlineSegmentCount)%progress.ProgressPrintFrequency == 0 {
progress.Report()
}
return nil return nil
} }
// LoopStarted is called at each start of a loop. // LoopStarted is called at each start of a loop.
func (progress *ProgressObserver) LoopStarted(ctx context.Context, info metaloop.LoopInfo) (err error) { func (progress *ProgressObserver) LoopStarted(ctx context.Context, info segmentloop.LoopInfo) (err error) {
return nil return nil
} }

38
cmd/metabase-verify/verify/segmentsizes.go
View File

@ -8,8 +8,8 @@ import (
"go.uber.org/zap" "go.uber.org/zap"
"storj.io/storj/satellite/metabase" "storj.io/common/uuid"
"storj.io/storj/satellite/metabase/metaloop" "storj.io/storj/satellite/metabase/segmentloop"
) )
// SegmentSizes verifies segments table plain_offset and plain_size. // SegmentSizes verifies segments table plain_offset and plain_size.
@ -20,53 +20,45 @@ type SegmentSizes struct {
} }
type segmentState struct { type segmentState struct {
Current metabase.ObjectStream StreamID uuid.UUID
Status metabase.ObjectStatus
ExpectedOffset int64 ExpectedOffset int64
} }
// Object implements the Observer interface.
func (verify *SegmentSizes) Object(ctx context.Context, obj *metaloop.Object) error {
verify.segmentState = segmentState{
Current: obj.ObjectStream,
Status: obj.Status,
}
return nil
}
// LoopStarted is called at each start of a loop. // LoopStarted is called at each start of a loop.
func (verify *SegmentSizes) LoopStarted(ctx context.Context, info metaloop.LoopInfo) (err error) { func (verify *SegmentSizes) LoopStarted(ctx context.Context, info segmentloop.LoopInfo) (err error) {
return nil return nil
} }
// RemoteSegment implements the Observer interface. // RemoteSegment implements the Observer interface.
func (verify *SegmentSizes) RemoteSegment(ctx context.Context, seg *metaloop.Segment) error { func (verify *SegmentSizes) RemoteSegment(ctx context.Context, seg *segmentloop.Segment) error {
return verify.advanceSegment(ctx, seg) return verify.advanceSegment(ctx, seg)
} }
// InlineSegment implements the Observer interface. // InlineSegment implements the Observer interface.
func (verify *SegmentSizes) InlineSegment(ctx context.Context, seg *metaloop.Segment) error { func (verify *SegmentSizes) InlineSegment(ctx context.Context, seg *segmentloop.Segment) error {
return verify.advanceSegment(ctx, seg) return verify.advanceSegment(ctx, seg)
} }
func (verify *SegmentSizes) advanceSegment(ctx context.Context, seg *metaloop.Segment) error { func (verify *SegmentSizes) advanceSegment(ctx context.Context, seg *segmentloop.Segment) error {
if verify.segmentState.StreamID != seg.StreamID {
verify.segmentState = segmentState{
StreamID: seg.StreamID,
}
}
if seg.PlainSize > seg.EncryptedSize { if seg.PlainSize > seg.EncryptedSize {
verify.Log.Error("plain size larger than encrypted size", verify.Log.Error("plain size larger than encrypted size",
zap.Any("object", formatObject(verify.Current)), zap.Any("stream_id", seg.StreamID.String()),
zap.Any("position", seg.Position), zap.Any("position", seg.Position),
zap.Int32("plain size", seg.PlainSize), zap.Int32("plain size", seg.PlainSize),
zap.Int32("encrypted size", seg.EncryptedSize)) zap.Int32("encrypted size", seg.EncryptedSize))
} }
if verify.Status != metabase.Committed {
return nil
}
if verify.ExpectedOffset != seg.PlainOffset { if verify.ExpectedOffset != seg.PlainOffset {
verify.Log.Error("invalid offset", verify.Log.Error("invalid offset",
zap.Any("object", formatObject(verify.Current)), zap.Any("stream_id", seg.StreamID.String()),
zap.Any("position", seg.Position), zap.Any("position", seg.Position),
zap.Int64("expected", verify.ExpectedOffset), zap.Int64("expected", verify.ExpectedOffset),

18
cmd/metabase-verify/verify/verify.go
View File

@ -5,14 +5,12 @@ package verify
import ( import (
"context" "context"
"fmt"
"github.com/zeebo/errs" "github.com/zeebo/errs"
"go.uber.org/zap" "go.uber.org/zap"
"storj.io/common/errs2" "storj.io/common/errs2"
"storj.io/storj/satellite/metabase" "storj.io/storj/satellite/metabase/segmentloop"
"storj.io/storj/satellite/metabase/metaloop"
) )
// Error is the default error class for the package. // Error is the default error class for the package.
@ -24,17 +22,17 @@ type Chore struct {
Config Config Config Config
DB metaloop.MetabaseDB DB segmentloop.MetabaseDB
} }
// Config contains configuration for all the services. // Config contains configuration for all the services.
type Config struct { type Config struct {
ProgressPrintFrequency int64 ProgressPrintFrequency int64
Loop metaloop.Config Loop segmentloop.Config
} }
// New creates new verification. // New creates new verification.
func New(log *zap.Logger, mdb metaloop.MetabaseDB, config Config) *Chore { func New(log *zap.Logger, mdb segmentloop.MetabaseDB, config Config) *Chore {
return &Chore{ return &Chore{
Log: log, Log: log,
Config: config, Config: config,
@ -42,9 +40,9 @@ func New(log *zap.Logger, mdb metaloop.MetabaseDB, config Config) *Chore {
} }
} }
// RunOnce creates a new metaloop and runs the verifications. // RunOnce creates a new segmentloop and runs the verifications.
func (chore *Chore) RunOnce(ctx context.Context) error { func (chore *Chore) RunOnce(ctx context.Context) error {
loop := metaloop.New(chore.Config.Loop, chore.DB) loop := segmentloop.New(chore.Config.Loop, chore.DB)
var group errs2.Group var group errs2.Group
group.Go(func() error { group.Go(func() error {
@ -69,7 +67,3 @@ func (chore *Chore) RunOnce(ctx context.Context) error {
}) })
return Error.Wrap(errs.Combine(group.Wait()...)) return Error.Wrap(errs.Combine(group.Wait()...))
} }
func formatObject(obj metabase.ObjectStream) string {
return fmt.Sprintf("project:%q bucket:%q key:%q stream:\"%x\"", obj.ProjectID, obj.BucketName, obj.ObjectKey, obj.StreamID[:])
}

182
cmd/metainfo-loop-benchmark/bench.go
View File

@ -1,182 +0,0 @@
// Copyright (C) 2021 Storj Labs, Inc.
// See LICENSE for copying information.
package main
import (
"context"
"errors"
"os"
"runtime"
"runtime/pprof"
"time"
"github.com/spacemonkeygo/monkit/v3"
flag "github.com/spf13/pflag"
"github.com/zeebo/errs"
"go.uber.org/zap"
"storj.io/common/errs2"
"storj.io/common/memory"
"storj.io/storj/satellite/metabase"
"storj.io/storj/satellite/metabase/metaloop"
)
var mon = monkit.Package()
// Error is the default error class for the package.
var Error = errs.Class("metaloop-benchmark")
// Bench benchmarks metainfo loop performance.
type Bench struct {
CPUProfile string
MetabaseDB string
IgnoreVersionMismatch bool
ProgressPrintFrequency int64
Loop metaloop.Config
}
// BindFlags adds bench flags to the the flagset.
func (bench *Bench) BindFlags(flag *flag.FlagSet) {
flag.StringVar(&bench.CPUProfile, "cpuprofile", "", "write cpu profile to file")
flag.StringVar(&bench.MetabaseDB, "metabasedb", "", "connection URL for MetabaseDB")
flag.BoolVar(&bench.IgnoreVersionMismatch, "ignore-version-mismatch", false, "ignore version mismatch")
flag.Int64Var(&bench.ProgressPrintFrequency, "progress.frequency", 1000000, "how often should we print progress (every object)")
flag.DurationVar(&bench.Loop.CoalesceDuration, "loop.coalesce-duration", 5*time.Second, "how long to wait for new observers before starting iteration")
flag.Float64Var(&bench.Loop.RateLimit, "loop.rate-limit", 0, "rate limit (default is 0 which is unlimited segments per second)")
flag.IntVar(&bench.Loop.ListLimit, "loop.list-limit", 2500, "how many items to query in a batch")
}
// VerifyFlags verifies whether the values provided are valid.
func (bench *Bench) VerifyFlags() error {
var errlist errs.Group
if bench.MetabaseDB == "" {
errlist.Add(errors.New("flag '--metabasedb' is not set"))
}
return errlist.Err()
}
// Run runs the benchmark.
func (bench *Bench) Run(ctx context.Context, log *zap.Logger) (err error) {
defer mon.Task()(&ctx)(&err)
// setup profiling
if bench.CPUProfile != "" {
f, err := os.Create(bench.CPUProfile)
if err != nil {
return err
}
err = pprof.StartCPUProfile(f)
if err != nil {
return err
}
defer pprof.StopCPUProfile()
}
// setup databases
mdb, err := metabase.Open(ctx, log.Named("mdb"), bench.MetabaseDB)
if err != nil {
return Error.Wrap(err)
}
defer func() { _ = mdb.Close() }()
checkMetabase := mdb.CheckVersion(ctx)
if checkMetabase != nil {
log.Error("versions skewed", zap.Any("metabase version", checkMetabase))
if !bench.IgnoreVersionMismatch {
return checkMetabase
}
}
// setup metainfo loop
var group errs2.Group
loop := metaloop.New(bench.Loop, mdb)
group.Go(func() error {
progress := &ProgressObserver{
Log: log.Named("progress"),
ProgressPrintFrequency: bench.ProgressPrintFrequency,
}
err := loop.Join(ctx, progress)
progress.Report()
return Error.Wrap(err)
})
group.Go(func() error {
err := loop.RunOnce(ctx)
return Error.Wrap(err)
})
// wait for loop to finish
if allErrors := group.Wait(); len(allErrors) > 0 {
return Error.Wrap(errs.Combine(allErrors...))
}
return nil
}
// ProgressObserver counts and prints progress of metainfo loop.
type ProgressObserver struct {
Log *zap.Logger
ProgressPrintFrequency int64
ObjectCount int64
RemoteSegmentCount int64
InlineSegmentCount int64
}
// Report reports the current progress.
func (progress *ProgressObserver) Report() {
progress.Log.Debug("progress",
zap.Int64("objects", progress.ObjectCount),
zap.Int64("remote segments", progress.RemoteSegmentCount),
zap.Int64("inline segments", progress.InlineSegmentCount),
)
var m runtime.MemStats
runtime.ReadMemStats(&m)
progress.Log.Debug("memory",
zap.String("Alloc", memory.Size(int64(m.Alloc)).String()),
zap.String("TotalAlloc", memory.Size(int64(m.TotalAlloc)).String()),
zap.String("Sys", memory.Size(int64(m.Sys)).String()),
zap.Uint32("NumGC", m.NumGC),
)
}
// Object implements the Observer interface.
func (progress *ProgressObserver) Object(context.Context, *metaloop.Object) error {
progress.ObjectCount++
if progress.ObjectCount%progress.ProgressPrintFrequency == 0 {
progress.Report()
}
return nil
}
// RemoteSegment implements the Observer interface.
func (progress *ProgressObserver) RemoteSegment(context.Context, *metaloop.Segment) error {
progress.RemoteSegmentCount++
return nil
}
// InlineSegment implements the Observer interface.
func (progress *ProgressObserver) InlineSegment(context.Context, *metaloop.Segment) error {
progress.InlineSegmentCount++
return nil
}
// LoopStarted is called at each start of a loop.
func (progress *ProgressObserver) LoopStarted(ctx context.Context, info metaloop.LoopInfo) (err error) {
return nil
}

46
cmd/metainfo-loop-benchmark/main.go
View File

@ -1,46 +0,0 @@
// Copyright (C) 2021 Storj Labs, Inc.
// See LICENSE for copying information.
package main
import (
"github.com/spf13/cobra"
"go.uber.org/zap"
"storj.io/private/process"
)
var (
rootCmd = &cobra.Command{
Use: "metainfo-loop-benchmark",
Short: "metainfo-loop-benchmark",
}
runCmd = &cobra.Command{
Use: "run",
Short: "run metainfo-loop-benchmark",
RunE: run,
}
bench Bench
)
func init() {
rootCmd.AddCommand(runCmd)
bench.BindFlags(runCmd.Flags())
}
func run(cmd *cobra.Command, args []string) error {
if err := bench.VerifyFlags(); err != nil {
return err
}
ctx, _ := process.Ctx(cmd)
log := zap.L()
return bench.Run(ctx, log)
}
func main() {
process.Exec(rootCmd)
}

4
monkit.lock
View File

@ -58,10 +58,6 @@ storj.io/storj/satellite/gracefulexit."graceful_exit_transfer_piece_success" Met
storj.io/storj/satellite/metabase/segmentloop."segmentsProcessed" IntVal storj.io/storj/satellite/metabase/segmentloop."segmentsProcessed" IntVal
storj.io/storj/satellite/metabase/segmentloop.*Service.RunOnce Task storj.io/storj/satellite/metabase/segmentloop.*Service.RunOnce Task
storj.io/storj/satellite/metainfo."metainfo_rate_limit_exceeded" Event storj.io/storj/satellite/metainfo."metainfo_rate_limit_exceeded" Event
storj.io/storj/satellite/metainfo/metaloop."objectsIterated" IntVal
storj.io/storj/satellite/metainfo/metaloop."objectsProcessed" IntVal
storj.io/storj/satellite/metainfo/metaloop."segmentsProcessed" IntVal
storj.io/storj/satellite/metainfo/metaloop.*Service.RunOnce Task
storj.io/storj/satellite/metainfo/piecedeletion."delete_batch_size" IntVal storj.io/storj/satellite/metainfo/piecedeletion."delete_batch_size" IntVal
storj.io/storj/satellite/metainfo/piecedeletion."deletion_pieces_unhandled_count" IntVal storj.io/storj/satellite/metainfo/piecedeletion."deletion_pieces_unhandled_count" IntVal
storj.io/storj/satellite/orders."download_failed_not_enough_pieces_uplink" Meter storj.io/storj/satellite/orders."download_failed_not_enough_pieces_uplink" Meter

3
private/testplanet/satellite.go
View File

@ -46,7 +46,6 @@ import (
"storj.io/storj/satellite/inspector" "storj.io/storj/satellite/inspector"
"storj.io/storj/satellite/mailservice" "storj.io/storj/satellite/mailservice"
"storj.io/storj/satellite/metabase" "storj.io/storj/satellite/metabase"
"storj.io/storj/satellite/metabase/metaloop"
"storj.io/storj/satellite/metabase/segmentloop" "storj.io/storj/satellite/metabase/segmentloop"
"storj.io/storj/satellite/metainfo" "storj.io/storj/satellite/metainfo"
"storj.io/storj/satellite/metainfo/expireddeletion" "storj.io/storj/satellite/metainfo/expireddeletion"
@ -96,7 +95,6 @@ type Satellite struct {
Metabase *metabase.DB Metabase *metabase.DB
Service *metainfo.Service Service *metainfo.Service
Endpoint *metainfo.Endpoint Endpoint *metainfo.Endpoint
Loop *metaloop.Service
SegmentLoop *segmentloop.Service SegmentLoop *segmentloop.Service
} }
@ -545,7 +543,6 @@ func createNewSystem(name string, log *zap.Logger, config satellite.Config, peer
system.Metainfo.Metabase = api.Metainfo.Metabase system.Metainfo.Metabase = api.Metainfo.Metabase
system.Metainfo.Service = peer.Metainfo.Service system.Metainfo.Service = peer.Metainfo.Service
system.Metainfo.Endpoint = api.Metainfo.Endpoint system.Metainfo.Endpoint = api.Metainfo.Endpoint
system.Metainfo.Loop = peer.Metainfo.Loop
system.Metainfo.SegmentLoop = peer.Metainfo.SegmentLoop system.Metainfo.SegmentLoop = peer.Metainfo.SegmentLoop
system.Inspector.Endpoint = api.Inspector.Endpoint system.Inspector.Endpoint = api.Inspector.Endpoint

11
satellite/core.go
View File

@ -32,7 +32,6 @@ import (
"storj.io/storj/satellite/audit" "storj.io/storj/satellite/audit"
"storj.io/storj/satellite/gracefulexit" "storj.io/storj/satellite/gracefulexit"
"storj.io/storj/satellite/metabase" "storj.io/storj/satellite/metabase"
"storj.io/storj/satellite/metabase/metaloop"
"storj.io/storj/satellite/metabase/segmentloop" "storj.io/storj/satellite/metabase/segmentloop"
"storj.io/storj/satellite/metainfo" "storj.io/storj/satellite/metainfo"
"storj.io/storj/satellite/metainfo/expireddeletion" "storj.io/storj/satellite/metainfo/expireddeletion"
@ -79,7 +78,6 @@ type Core struct {
Metainfo struct { Metainfo struct {
Metabase *metabase.DB Metabase *metabase.DB
Service *metainfo.Service Service *metainfo.Service
Loop *metaloop.Service
SegmentLoop *segmentloop.Service SegmentLoop *segmentloop.Service
} }
@ -248,15 +246,6 @@ func New(log *zap.Logger, full *identity.FullIdentity, db DB,
peer.DB.Buckets(), peer.DB.Buckets(),
peer.Metainfo.Metabase, peer.Metainfo.Metabase,
) )
peer.Metainfo.Loop = metaloop.New(
config.Metainfo.Loop,
peer.Metainfo.Metabase,
)
peer.Services.Add(lifecycle.Item{
Name: "metainfo:loop",
Run: peer.Metainfo.Loop.Run,
Close: peer.Metainfo.Loop.Close,
})
peer.Metainfo.SegmentLoop = segmentloop.New( peer.Metainfo.SegmentLoop = segmentloop.New(
config.Metainfo.SegmentLoop, config.Metainfo.SegmentLoop,
peer.Metainfo.Metabase, peer.Metainfo.Metabase,

663
satellite/metabase/metaloop/service.go
View File

@ -1,663 +0,0 @@
// Copyright (C) 2019 Storj Labs, Inc.
// See LICENSE for copying information.
package metaloop
import (
"context"
"errors"
"fmt"
"time"
"github.com/spacemonkeygo/monkit/v3"
"github.com/zeebo/errs"
"golang.org/x/time/rate"
"storj.io/common/uuid"
"storj.io/storj/satellite/metabase"
)
const batchsizeLimit = 2500
var (
mon = monkit.Package()
// Error is a standard error class for this component.
Error = errs.Class("metainfo loop")
// ErrClosed is a loop closed error.
ErrClosed = Error.New("loop closed")
)
// Object is the object info passed to Observer by metainfo loop.
type Object metabase.LoopObjectEntry
// Expired checks if object expired relative to now.
func (object *Object) Expired(now time.Time) bool {
return object.ExpiresAt != nil && object.ExpiresAt.Before(now)
}
// Segment is the segment info passed to Observer by metainfo loop.
type Segment struct {
Location metabase.SegmentLocation // tally, repair, graceful exit, audit
ExpirationDate time.Time // tally, repair
metabase.LoopSegmentEntry
}
// Expired checks if segment is expired relative to now.
func (segment *Segment) Expired(now time.Time) bool {
return !segment.ExpirationDate.IsZero() && segment.ExpirationDate.Before(now)
}
// Observer is an interface defining an observer that can subscribe to the metainfo loop.
//
// architecture: Observer
type Observer interface {
Object(context.Context, *Object) error
RemoteSegment(context.Context, *Segment) error
InlineSegment(context.Context, *Segment) error
LoopStarted(context.Context, LoopInfo) error
}
// LoopInfo contains information about the current loop.
type LoopInfo struct {
Started time.Time
}
// NullObserver is an observer that does nothing. This is useful for joining
// and ensuring the metainfo loop runs once before you use a real observer.
type NullObserver struct{}
// Object implements the Observer interface.
func (NullObserver) Object(context.Context, *Object) error {
return nil
}
// RemoteSegment implements the Observer interface.
func (NullObserver) RemoteSegment(context.Context, *Segment) error {
return nil
}
// InlineSegment implements the Observer interface.
func (NullObserver) InlineSegment(context.Context, *Segment) error {
return nil
}
// LoopStarted is called at each loop start.
func (NullObserver) LoopStarted(context.Context, LoopInfo) error {
return nil
}
type observerContext struct {
trigger bool
observer Observer
ctx context.Context
done chan error
object *monkit.DurationDist
remote *monkit.DurationDist
inline *monkit.DurationDist
}
func newObserverContext(ctx context.Context, obs Observer) *observerContext {
name := fmt.Sprintf("%T", obs)
key := monkit.NewSeriesKey("observer").WithTag("name", name)
return &observerContext{
observer: obs,
ctx: ctx,
done: make(chan error),
object: monkit.NewDurationDist(key.WithTag("pointer_type", "object")),
inline: monkit.NewDurationDist(key.WithTag("pointer_type", "inline")),
remote: monkit.NewDurationDist(key.WithTag("pointer_type", "remote")),
}
}
func (observer *observerContext) Object(ctx context.Context, object *Object) error {
start := time.Now()
defer func() { observer.object.Insert(time.Since(start)) }()
return observer.observer.Object(ctx, object)
}
func (observer *observerContext) RemoteSegment(ctx context.Context, segment *Segment) error {
start := time.Now()
defer func() { observer.remote.Insert(time.Since(start)) }()
return observer.observer.RemoteSegment(ctx, segment)
}
func (observer *observerContext) InlineSegment(ctx context.Context, segment *Segment) error {
start := time.Now()
defer func() { observer.inline.Insert(time.Since(start)) }()
return observer.observer.InlineSegment(ctx, segment)
}
func (observer *observerContext) HandleError(err error) bool {
if err != nil {
observer.done <- err
observer.Finish()
return true
}
return false
}
func (observer *observerContext) Finish() {
close(observer.done)
name := fmt.Sprintf("%T", observer.observer)
stats := allObserverStatsCollectors.GetStats(name)
stats.Observe(observer)
}
func (observer *observerContext) Wait() error {
return <-observer.done
}
// Config contains configurable values for the metainfo loop.
type Config struct {
CoalesceDuration time.Duration `help:"how long to wait for new observers before starting iteration" releaseDefault:"5s" devDefault:"5s" testDefault:"1s"`
RateLimit float64 `help:"rate limit (default is 0 which is unlimited segments per second)" default:"0"`
ListLimit int `help:"how many items to query in a batch" default:"2500" testDefault:"10000"`
AsOfSystemInterval time.Duration `help:"as of system interval" releaseDefault:"-5m" devDefault:"-1us" testDefault:"-1us"`
SuspiciousProcessedRatio float64 `help:"ratio where to consider processed count as supicious" default:"0.03"`
}
// MetabaseDB contains iterators for the metabase data.
type MetabaseDB interface {
// Now returns the time on the database.
Now(ctx context.Context) (time.Time, error)
// IterateLoopObjects iterates through all objects in metabase for metainfo loop purpose.
IterateLoopObjects(ctx context.Context, opts metabase.IterateLoopObjects, fn func(context.Context, metabase.LoopObjectsIterator) error) (err error)
// IterateLoopStreams iterates through all streams passed in as arguments.
IterateLoopStreams(ctx context.Context, opts metabase.IterateLoopStreams, handleStream func(ctx context.Context, streamID uuid.UUID, next metabase.SegmentIterator) error) (err error)
// GetTableStats gathers statistics about the tables.
GetTableStats(context.Context, metabase.GetTableStats) (metabase.TableStats, error)
}
// Service is a metainfo loop service.
//
// architecture: Service
type Service struct {
config Config
metabaseDB MetabaseDB
join chan *observerContext
done chan struct{}
}
// New creates a new metainfo loop service.
func New(config Config, metabaseDB MetabaseDB) *Service {
return &Service{
metabaseDB: metabaseDB,
config: config,
join: make(chan *observerContext),
done: make(chan struct{}),
}
}
// Join will join the looper for one full cycle until completion and then returns.
// Joining will trigger a new iteration after coalesce duration.
// On ctx cancel the observer will return without completely finishing.
// Only on full complete iteration it will return nil.
// Safe to be called concurrently.
func (loop *Service) Join(ctx context.Context, observer Observer) (err error) {
return loop.joinObserver(ctx, true, observer)
}
// Monitor will join the looper for one full cycle until completion and then returns.
// Joining with monitoring won't trigger after coalesce duration.
// On ctx cancel the observer will return without completely finishing.
// Only on full complete iteration it will return nil.
// Safe to be called concurrently.
func (loop *Service) Monitor(ctx context.Context, observer Observer) (err error) {
return loop.joinObserver(ctx, false, observer)
}
// joinObserver will join the looper for one full cycle until completion and then returns.
// On ctx cancel the observer will return without completely finishing.
// Only on full complete iteration it will return nil.
// Safe to be called concurrently.
func (loop *Service) joinObserver(ctx context.Context, trigger bool, obs Observer) (err error) {
defer mon.Task()(&ctx)(&err)
obsctx := newObserverContext(ctx, obs)
obsctx.trigger = trigger
select {
case loop.join <- obsctx:
case <-ctx.Done():
return ctx.Err()
case <-loop.done:
return ErrClosed
}
return obsctx.Wait()
}
// Run starts the looping service.
// It can only be called once, otherwise a panic will occur.
func (loop *Service) Run(ctx context.Context) (err error) {
defer mon.Task()(&ctx)(&err)
for {
err := loop.RunOnce(ctx)
if err != nil {
return err
}
}
}
// Close closes the looping services.
func (loop *Service) Close() (err error) {
close(loop.done)
return nil
}
// monMetainfo is to preserve the monitoring names.
var monMetainfo = monkit.ScopeNamed("storj.io/storj/satellite/metainfo/metaloop")
// RunOnce goes through metainfo one time and sends information to observers.
//
// It is not safe to call this concurrently with Run.
func (loop *Service) RunOnce(ctx context.Context) (err error) {
defer monMetainfo.Task()(&ctx)(&err) //mon:locked
coalesceTimer := time.NewTimer(loop.config.CoalesceDuration)
defer coalesceTimer.Stop()
stopTimer(coalesceTimer)
earlyExit := make(chan *observerContext)
earlyExitDone := make(chan struct{})
monitorEarlyExit := func(obs *observerContext) {
select {
case <-obs.ctx.Done():
select {
case <-earlyExitDone:
case earlyExit <- obs:
}
case <-earlyExitDone:
}
}
timerStarted := false
observers := []*observerContext{}
waitformore:
for {
select {
// when the coalesce timer hits, we have waited enough for observers to join.
case <-coalesceTimer.C:
break waitformore
// wait for a new observer to join.
case obsctx := <-loop.join:
// when the observer triggers the loop and it's the first one,
// then start the coalescing timer.
if obsctx.trigger {
if !timerStarted {
coalesceTimer.Reset(loop.config.CoalesceDuration)
timerStarted = true
}
}
observers = append(observers, obsctx)
go monitorEarlyExit(obsctx)
// remove an observer from waiting when it's canceled before the loop starts.
case obsctx := <-earlyExit:
for i, obs := range observers {
if obs == obsctx {
observers = append(observers[:i], observers[i+1:]...)
break
}
}
obsctx.HandleError(obsctx.ctx.Err())
// reevalute, whether we acually need to start the loop.
timerShouldRun := false
for _, obs := range observers {
timerShouldRun = timerShouldRun || obs.trigger
}
if !timerShouldRun && timerStarted {
stopTimer(coalesceTimer)
}
// when ctx done happens we can finish all the waiting observers.
case <-ctx.Done():
close(earlyExitDone)
errorObservers(observers, ctx.Err())
return ctx.Err()
}
}
close(earlyExitDone)
return loop.iterateDatabase(ctx, observers)
}
func stopTimer(t *time.Timer) {
t.Stop()
// drain if it contains something
select {
case <-t.C:
default:
}
}
// Wait waits for run to be finished.
// Safe to be called concurrently.
func (loop *Service) Wait() {
<-loop.done
}
var errNoObservers = errs.New("no observers")
func (loop *Service) iterateDatabase(ctx context.Context, observers []*observerContext) (err error) {
defer func() {
if err != nil {
errorObservers(observers, err)
return
}
finishObservers(observers)
}()
before, err := loop.metabaseDB.GetTableStats(ctx, metabase.GetTableStats{
AsOfSystemInterval: loop.config.AsOfSystemInterval,
})
if err != nil {
return Error.Wrap(err)
}
var processed processedStats
processed, observers, err = loop.iterateObjects(ctx, observers)
if errors.Is(err, errNoObservers) {
return nil
}
if err != nil {
return Error.Wrap(err)
}
after, err := loop.metabaseDB.GetTableStats(ctx, metabase.GetTableStats{
AsOfSystemInterval: loop.config.AsOfSystemInterval,
})
if err != nil {
return Error.Wrap(err)
}
if err := loop.verifyProcessedCount(before, after, processed); err != nil {
return Error.Wrap(err)
}
return nil
}
func (loop *Service) verifyProcessedCount(before, after metabase.TableStats, processed processedStats) error {
return errs.Combine(
loop.verifyCount("object", before.ObjectCount, after.ObjectCount, processed.objects),
loop.verifyCount("segment", before.SegmentCount, after.SegmentCount, processed.segments),
)
}
func (loop *Service) verifyCount(kind string, before, after, processed int64) error {
low, high := before, after
if low > high {
low, high = high, low
}
var deltaFromBounds int64
var ratio float64
if processed < low {
deltaFromBounds = low - processed
// +1 to avoid division by zero
ratio = float64(deltaFromBounds) / float64(low+1)
} else if processed > high {
deltaFromBounds = processed - high
// +1 to avoid division by zero
ratio = float64(deltaFromBounds) / float64(high+1)
}
mon.IntVal("metaloop_verify_" + kind + "_before").Observe(before)
mon.IntVal("metaloop_verify_" + kind + "_after").Observe(after)
mon.IntVal("metaloop_verify_" + kind + "_processed").Observe(processed)
mon.IntVal("metaloop_verify_" + kind + "_outside").Observe(deltaFromBounds)
mon.FloatVal("metaloop_verify_" + kind + "_outside_ratio").Observe(ratio)
// If we have very few items from the bounds, then it's expected and the ratio does not capture it well.
const minimumDeltaThreshold = 100
if deltaFromBounds < minimumDeltaThreshold {
return nil
}
if ratio > loop.config.SuspiciousProcessedRatio {
return Error.New("%s processed count looks suspicious: before:%v after:%v processed:%v ratio:%v threshold:%v", kind, before, after, processed, ratio, loop.config.SuspiciousProcessedRatio)
}
return nil
}
type processedStats struct {
objects int64
segments int64
}
func (loop *Service) iterateObjects(ctx context.Context, observers []*observerContext) (processed processedStats, _ []*observerContext, err error) {
defer mon.Task()(&ctx)(&err)
limit := loop.config.ListLimit
if limit <= 0 || limit > batchsizeLimit {
limit = batchsizeLimit
}
rateLimiter := rate.NewLimiter(rate.Limit(loop.config.RateLimit), 1)
startingTime, err := loop.metabaseDB.Now(ctx)
if err != nil {
return processed, observers, Error.Wrap(err)
}
observers = withObservers(ctx, observers, func(ctx context.Context, observer *observerContext) bool {
err := observer.observer.LoopStarted(ctx, LoopInfo{Started: startingTime})
return !observer.HandleError(err)
})
if len(observers) == 0 {
return processed, observers, errNoObservers
}
// TODO we may consider keeping only expiration time as its
// only thing we need to handle segments
objectsMap := make(map[uuid.UUID]metabase.LoopObjectEntry)
ids := make([]uuid.UUID, 0, limit)
processBatch := func(ctx context.Context) (err error) {
defer mon.TaskNamed("processBatch")(&ctx)(&err)
if len(objectsMap) == 0 {
return nil
}
err = loop.metabaseDB.IterateLoopStreams(ctx, metabase.IterateLoopStreams{
StreamIDs: ids,
AsOfSystemTime: startingTime,
AsOfSystemInterval: loop.config.AsOfSystemInterval,
}, func(ctx context.Context, streamID uuid.UUID, next metabase.SegmentIterator) (err error) {
defer mon.TaskNamed("iterateLoopStreamsCB")(&ctx, "objs", processed.objects, "segs", processed.segments)(&err)
if err := ctx.Err(); err != nil {
return err
}
obj, ok := objectsMap[streamID]
if !ok {
return Error.New("unable to find corresponding object: %v", streamID)
}
delete(objectsMap, streamID)
observers = withObservers(ctx, observers, func(ctx context.Context, observer *observerContext) bool {
object := Object(obj)
return !observer.HandleError(handleObject(ctx, observer, &object))
})
if len(observers) == 0 {
return errNoObservers
}
processed.objects++
monMetainfo.IntVal("objectsProcessed").Observe(processed.objects) //mon:locked
for {
// if context has been canceled exit. Otherwise, continue
if err := ctx.Err(); err != nil {
return err
}
var segment metabase.LoopSegmentEntry
if !next(ctx, &segment) {
break
}
location := metabase.SegmentLocation{
ProjectID: obj.ProjectID,
BucketName: obj.BucketName,
ObjectKey: obj.ObjectKey,
Position: segment.Position,
}
observers = withObservers(ctx, observers, func(ctx context.Context, observer *observerContext) bool {
return !observer.HandleError(handleSegment(ctx, observer, location, segment, obj.ExpiresAt))
})
if len(observers) == 0 {
return errNoObservers
}
processed.segments++
monMetainfo.IntVal("segmentsProcessed").Observe(processed.segments) //mon:locked
}
return nil
})
if err != nil {
return Error.Wrap(err)
}
if len(objectsMap) > 0 {
return Error.New("unhandled objects %#v", objectsMap)
}
return nil
}
var objectsIterated int64
segmentsInBatch := int32(0)
err = loop.metabaseDB.IterateLoopObjects(ctx, metabase.IterateLoopObjects{
BatchSize: limit,
AsOfSystemTime: startingTime,
AsOfSystemInterval: loop.config.AsOfSystemInterval,
}, func(ctx context.Context, it metabase.LoopObjectsIterator) (err error) {
defer mon.TaskNamed("iterateLoopObjectsCB")(&ctx)(&err)
var entry metabase.LoopObjectEntry
for it.Next(ctx, &entry) {
timer := mon.Timer("iterateLoopObjectsRateLimit").Start()
if err := rateLimiter.Wait(ctx); err != nil {
// We don't really execute concurrent batches so we should never
// exceed the burst size of 1 and this should never happen.
// We can also enter here if the context is cancelled.
timer.Stop()
return err
}
timer.Stop()
monMetainfo.IntVal("objectsIterated").Observe(objectsIterated) //mon:locked
objectsIterated++
objectsMap[entry.StreamID] = entry
ids = append(ids, entry.StreamID)
// add +1 to reduce risk of crossing limit
segmentsInBatch += entry.SegmentCount + 1
if segmentsInBatch >= int32(limit) {
err := processBatch(ctx)
if err != nil {
return err
}
if len(objectsMap) > 0 {
return errs.New("objects map is not empty")
}
ids = ids[:0]
segmentsInBatch = 0
}
}
return processBatch(ctx)
})
return processed, observers, err
}
func withObservers(ctx context.Context, observers []*observerContext, handleObserver func(ctx context.Context, observer *observerContext) bool) []*observerContext {
defer mon.Task()(&ctx)(nil)
nextObservers := observers[:0]
for _, observer := range observers {
keepObserver := handleObserver(ctx, observer)
if keepObserver {
nextObservers = append(nextObservers, observer)
}
}
return nextObservers
}
func handleObject(ctx context.Context, observer *observerContext, object *Object) (err error) {
defer mon.Task()(&ctx)(&err)
if err := observer.Object(ctx, object); err != nil {
return err
}
return observer.ctx.Err()
}
func handleSegment(ctx context.Context, observer *observerContext, location metabase.SegmentLocation, segment metabase.LoopSegmentEntry, expirationDate *time.Time) (err error) {
defer mon.Task()(&ctx)(&err)
loopSegment := &Segment{
Location: location,
LoopSegmentEntry: segment,
}
if expirationDate != nil {
loopSegment.ExpirationDate = *expirationDate
}
if loopSegment.Inline() {
if err := observer.InlineSegment(ctx, loopSegment); err != nil {
return err
}
} else {
if err := observer.RemoteSegment(ctx, loopSegment); err != nil {
return err
}
}
return observer.ctx.Err()
}
func finishObservers(observers []*observerContext) {
for _, observer := range observers {
observer.Finish()
}
}
func errorObservers(observers []*observerContext, err error) {
for _, observer := range observers {
observer.HandleError(err)
}
}

489
satellite/metabase/metaloop/service_test.go
View File

@ -1,489 +0,0 @@
// Copyright (C) 2019 Storj Labs, Inc.
// See LICENSE for copying information.
package metaloop_test
import (
"context"
"errors"
"strconv"
"strings"
"sync/atomic"
"testing"
"time"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
"go.uber.org/zap"
"golang.org/x/sync/errgroup"
"storj.io/common/errs2"
"storj.io/common/memory"
"storj.io/common/testcontext"
"storj.io/common/testrand"
"storj.io/storj/private/testplanet"
"storj.io/storj/satellite"
"storj.io/storj/satellite/metabase"
"storj.io/storj/satellite/metabase/metaloop"
)
// TestLoop does the following
// * upload 5 remote files with 1 segment
// * (TODO) upload 3 remote files with 2 segments
// * upload 2 inline files
// * connect two observers to the metainfo loop
// * run the metainfo loop
// * expect that each observer has seen:
// - 5 remote files
// - 5 remote segments
// - 2 inline files/segments
// - 7 unique path items
func TestLoop(t *testing.T) {
// TODO: figure out how to configure testplanet so we can upload 2*segmentSize to get two segments
segmentSize := 8 * memory.KiB
testplanet.Run(t, testplanet.Config{
SatelliteCount: 1,
StorageNodeCount: 4,
UplinkCount: 1,
Reconfigure: testplanet.Reconfigure{
Satellite: func(log *zap.Logger, index int, config *satellite.Config) {
config.Metainfo.Loop.CoalesceDuration = 1 * time.Second
},
},
}, func(t *testing.T, ctx *testcontext.Context, planet *testplanet.Planet) {
ul := planet.Uplinks[0]
satellite := planet.Satellites[0]
metaLoop := satellite.Metainfo.Loop
// upload 5 remote files with 1 segment
for i := 0; i < 5; i++ {
testData := testrand.Bytes(segmentSize)
path := "/some/remote/path/" + strconv.Itoa(i)
err := ul.Upload(ctx, satellite, "bucket", path, testData)
require.NoError(t, err)
}
// (TODO) upload 3 remote files with 2 segments
// for i := 0; i < 3; i++ {
// testData := testrand.Bytes(2 * segmentSize)
// path := "/some/other/remote/path/" + strconv.Itoa(i)
// err := ul.Upload(ctx, satellite, "bucket", path, testData)
// require.NoError(t, err)
// }
// upload 2 inline files
for i := 0; i < 2; i++ {
testData := testrand.Bytes(segmentSize / 8)
path := "/some/inline/path/" + strconv.Itoa(i)
err := ul.Upload(ctx, satellite, "bucket", path, testData)
require.NoError(t, err)
}
// create 2 observers
obs1 := newTestObserver(nil)
obs2 := newTestObserver(nil)
var group errgroup.Group
group.Go(func() error {
return metaLoop.Join(ctx, obs1)
})
group.Go(func() error {
return metaLoop.Join(ctx, obs2)
})
err := group.Wait()
require.NoError(t, err)
projectID := ul.Projects[0].ID
for _, obs := range []*testObserver{obs1, obs2} {
assert.EqualValues(t, 7, obs.objectCount)
assert.EqualValues(t, 5, obs.remoteSegCount)
assert.EqualValues(t, 2, obs.inlineSegCount)
assert.EqualValues(t, 7, len(obs.uniquePaths))
for _, path := range obs.uniquePaths {
assert.EqualValues(t, path.BucketName, "bucket")
assert.EqualValues(t, path.ProjectID, projectID)
}
// TODO we need better calulation
assert.NotZero(t, obs.totalMetadataSize)
}
})
}
func TestLoop_AllData(t *testing.T) {
segmentSize := 8 * memory.KiB
testplanet.Run(t, testplanet.Config{
SatelliteCount: 1,
StorageNodeCount: 4,
UplinkCount: 3,
Reconfigure: testplanet.Reconfigure{
Satellite: func(log *zap.Logger, index int, config *satellite.Config) {
config.Metainfo.Loop.CoalesceDuration = 1 * time.Second
config.Metainfo.Loop.ListLimit = 2
},
},
}, func(t *testing.T, ctx *testcontext.Context, planet *testplanet.Planet) {
bucketNames := strings.Split("abc", "")
data := testrand.Bytes(segmentSize)
for _, up := range planet.Uplinks {
for _, bucketName := range bucketNames {
err := up.Upload(ctx, planet.Satellites[0], "zzz"+bucketName, "1", data)
require.NoError(t, err)
}
}
metaLoop := planet.Satellites[0].Metainfo.Loop
obs := newTestObserver(nil)
err := metaLoop.Join(ctx, obs)
require.NoError(t, err)
gotItems := len(obs.uniquePaths)
require.Equal(t, len(bucketNames)*len(planet.Uplinks), gotItems)
})
}
func TestLoop_ObjectNoSegments(t *testing.T) {
testplanet.Run(t, testplanet.Config{
SatelliteCount: 1,
StorageNodeCount: 4,
UplinkCount: 1,
Reconfigure: testplanet.Reconfigure{
Satellite: func(log *zap.Logger, index int, config *satellite.Config) {
config.Metainfo.Loop.CoalesceDuration = 1 * time.Second
config.Metainfo.Loop.ListLimit = 2
},
},
}, func(t *testing.T, ctx *testcontext.Context, planet *testplanet.Planet) {
err := planet.Uplinks[0].CreateBucket(ctx, planet.Satellites[0], "abcd")
require.NoError(t, err)
project, err := planet.Uplinks[0].OpenProject(ctx, planet.Satellites[0])
require.NoError(t, err)
defer ctx.Check(project.Close)
expectedNumberOfObjects := 5
for i := 0; i < expectedNumberOfObjects; i++ {
info, err := project.BeginUpload(ctx, "abcd", "t"+strconv.Itoa(i), nil)
require.NoError(t, err)
_, err = project.CommitUpload(ctx, "abcd", "t"+strconv.Itoa(i), info.UploadID, nil)
require.NoError(t, err)
}
metaLoop := planet.Satellites[0].Metainfo.Loop
obs := newTestObserver(nil)
err = metaLoop.Join(ctx, obs)
require.NoError(t, err)
require.Equal(t, expectedNumberOfObjects, obs.objectCount)
require.Zero(t, obs.inlineSegCount)
require.Zero(t, obs.remoteSegCount)
// add object with single segment
data := testrand.Bytes(8 * memory.KiB)
err = planet.Uplinks[0].Upload(ctx, planet.Satellites[0], "dcba", "1", data)
require.NoError(t, err)
obs = newTestObserver(nil)
err = metaLoop.Join(ctx, obs)
require.NoError(t, err)
require.Equal(t, expectedNumberOfObjects+1, obs.objectCount)
require.Zero(t, obs.inlineSegCount)
require.Equal(t, 1, obs.remoteSegCount)
})
}
// TestLoopObserverCancel does the following:
// * upload 3 remote segments
// * hook three observers up to metainfo loop
// * let observer 1 run normally
// * let observer 2 return an error from one of its handlers
// * let observer 3's context be canceled
// * expect observer 1 to see all segments
// * expect observers 2 and 3 to finish with errors.
func TestLoopObserverCancel(t *testing.T) {
segmentSize := 8 * memory.KiB
testplanet.Run(t, testplanet.Config{
SatelliteCount: 1,
StorageNodeCount: 4,
UplinkCount: 1,
Reconfigure: testplanet.Reconfigure{
Satellite: func(log *zap.Logger, index int, config *satellite.Config) {
config.Metainfo.Loop.CoalesceDuration = 1 * time.Second
},
},
}, func(t *testing.T, ctx *testcontext.Context, planet *testplanet.Planet) {
ul := planet.Uplinks[0]
satellite := planet.Satellites[0]
metaLoop := satellite.Metainfo.Loop
// upload 3 remote files with 1 segment
for i := 0; i < 3; i++ {
testData := testrand.Bytes(segmentSize)
path := "/some/remote/path/" + strconv.Itoa(i)
err := ul.Upload(ctx, satellite, "bucket", path, testData)
require.NoError(t, err)
}
// create 1 "good" observer
obs1 := newTestObserver(nil)
mon1 := newTestObserver(nil)
// create observer that will return an error from RemoteSegment
obs2 := newTestObserver(func(ctx context.Context) error {
return errors.New("test error")
})
// create observer that will cancel its own context from RemoteSegment
obs3Ctx, cancel := context.WithCancel(ctx)
var once int64
obs3 := newTestObserver(func(ctx context.Context) error {
if atomic.AddInt64(&once, 1) == 1 {
cancel()
<-obs3Ctx.Done() // ensure we wait for cancellation to propagate
} else {
panic("multiple calls to observer after loop cancel")
}
return nil
})
var group errgroup.Group
group.Go(func() error {
return metaLoop.Join(ctx, obs1)
})
group.Go(func() error {
return metaLoop.Monitor(ctx, mon1)
})
group.Go(func() error {
err := metaLoop.Join(ctx, obs2)
if err == nil {
return errors.New("got no error")
}
if !strings.Contains(err.Error(), "test error") {
return errors.New("expected to find error")
}
return nil
})
group.Go(func() error {
err := metaLoop.Join(obs3Ctx, obs3)
if !errs2.IsCanceled(err) {
return errors.New("expected canceled")
}
return nil
})
err := group.Wait()
require.NoError(t, err)
// expect that obs1 saw all three segments, but obs2 and obs3 only saw the first one
assert.EqualValues(t, 3, obs1.remoteSegCount)
assert.EqualValues(t, 3, mon1.remoteSegCount)
assert.EqualValues(t, 1, obs2.remoteSegCount)
assert.EqualValues(t, 1, obs3.remoteSegCount)
})
}
// TestLoopCancel does the following:
// * upload 3 remote segments
// * hook two observers up to metainfo loop
// * cancel loop context partway through
// * expect both observers to exit with an error and see fewer than 3 remote segments
// * expect that a new observer attempting to join at this point receives a loop closed error.
func TestLoopCancel(t *testing.T) {
segmentSize := 8 * memory.KiB
testplanet.Run(t, testplanet.Config{
SatelliteCount: 1,
StorageNodeCount: 4,
UplinkCount: 1,
}, func(t *testing.T, ctx *testcontext.Context, planet *testplanet.Planet) {
ul := planet.Uplinks[0]
satellite := planet.Satellites[0]
// upload 3 remote files with 1 segment
for i := 0; i < 3; i++ {
testData := testrand.Bytes(segmentSize)
path := "/some/remote/path/" + strconv.Itoa(i)
err := ul.Upload(ctx, satellite, "bucket", path, testData)
require.NoError(t, err)
}
// create a new metainfo loop
metaLoop := metaloop.New(metaloop.Config{
CoalesceDuration: 1 * time.Second,
ListLimit: 10000,
}, satellite.Metainfo.Metabase)
// create a cancelable context to pass into metaLoop.Run
loopCtx, cancel := context.WithCancel(ctx)
// create 1 normal observer
obs1 := newTestObserver(nil)
var once int64
// create another normal observer that will wait before returning during RemoteSegment so we can sync with context cancelation
obs2 := newTestObserver(func(ctx context.Context) error {
// cancel context during call to obs2.RemoteSegment inside loop
if atomic.AddInt64(&once, 1) == 1 {
cancel()
<-ctx.Done() // ensure we wait for cancellation to propagate
} else {
panic("multiple calls to observer after loop cancel")
}
return nil
})
var group errgroup.Group
// start loop with cancelable context
group.Go(func() error {
err := metaLoop.Run(loopCtx)
if !errs2.IsCanceled(err) {
return errors.New("expected context canceled")
}
return nil
})
group.Go(func() error {
err := metaLoop.Join(ctx, obs1)
if !errs2.IsCanceled(err) {
return errors.New("expected context canceled")
}
return nil
})
group.Go(func() error {
err := metaLoop.Join(ctx, obs2)
if !errs2.IsCanceled(err) {
return errors.New("expected context canceled")
}
return nil
})
err := group.Wait()
require.NoError(t, err)
err = metaLoop.Close()
require.NoError(t, err)
obs3 := newTestObserver(nil)
err = metaLoop.Join(ctx, obs3)
require.Error(t, err)
assert.Contains(t, err.Error(), "loop closed")
// expect that obs1 and obs2 each saw fewer than three remote segments
assert.True(t, obs1.remoteSegCount < 3)
assert.True(t, obs2.remoteSegCount < 3)
})
}
func TestLoop_MonitorCancel(t *testing.T) {
testplanet.Run(t, testplanet.Config{
SatelliteCount: 1,
}, func(t *testing.T, ctx *testcontext.Context, planet *testplanet.Planet) {
satellite := planet.Satellites[0]
metaLoop := metaloop.New(metaloop.Config{
CoalesceDuration: time.Nanosecond,
ListLimit: 10000,
}, satellite.Metainfo.Metabase)
obs1 := newTestObserver(func(ctx context.Context) error {
return errors.New("test error")
})
var group errgroup.Group
loopCtx, loopCancel := context.WithCancel(ctx)
group.Go(func() error {
err := metaLoop.Run(loopCtx)
t.Log("metaloop stopped")
if !errs2.IsCanceled(err) {
return errors.New("expected context canceled")
}
return nil
})
obsCtx, obsCancel := context.WithCancel(ctx)
group.Go(func() error {
defer loopCancel()
err := metaLoop.Monitor(obsCtx, obs1)
t.Log("observer stopped")
if !errs2.IsCanceled(err) {
return errors.New("expected context canceled")
}
return nil
})
obsCancel()
err := group.Wait()
require.NoError(t, err)
err = metaLoop.Close()
require.NoError(t, err)
})
}
type testObserver struct {
objectCount int
remoteSegCount int
inlineSegCount int
totalMetadataSize int
uniquePaths map[string]metabase.SegmentLocation
onSegment func(context.Context) error // if set, run this during RemoteSegment()
}
func newTestObserver(onSegment func(context.Context) error) *testObserver {
return &testObserver{
objectCount: 0,
remoteSegCount: 0,
inlineSegCount: 0,
totalMetadataSize: 0,
uniquePaths: make(map[string]metabase.SegmentLocation),
onSegment: onSegment,
}
}
// LoopStarted is called at each start of a loop.
func (obs *testObserver) LoopStarted(ctx context.Context, info metaloop.LoopInfo) (err error) {
return nil
}
func (obs *testObserver) RemoteSegment(ctx context.Context, segment *metaloop.Segment) error {
obs.remoteSegCount++
key := segment.Location.Encode()
if _, ok := obs.uniquePaths[string(key)]; ok {
// TODO: collect the errors and check in test
panic("Expected unique path in observer.RemoteSegment")
}
obs.uniquePaths[string(key)] = segment.Location
if obs.onSegment != nil {
return obs.onSegment(ctx)
}
return nil
}
func (obs *testObserver) Object(ctx context.Context, object *metaloop.Object) error {
obs.objectCount++
obs.totalMetadataSize += object.EncryptedMetadataSize
return nil
}
func (obs *testObserver) InlineSegment(ctx context.Context, segment *metaloop.Segment) error {
obs.inlineSegCount++
key := segment.Location.Encode()
if _, ok := obs.uniquePaths[string(key)]; ok {
// TODO: collect the errors and check in test
panic("Expected unique path in observer.InlineSegment")
}
obs.uniquePaths[string(key)] = segment.Location
return nil
}

85
satellite/metabase/metaloop/stats.go
View File

@ -1,85 +0,0 @@
// Copyright (C) 2020 Storj Labs, Inc.
// See LICENSE for copying information.
package metaloop
import (
"sync"
"time"
"github.com/spacemonkeygo/monkit/v3"
)
var allObserverStatsCollectors = newObserverStatsCollectors()
type observerStatsCollectors struct {
mu sync.Mutex
observer map[string]*observerStats
}
func newObserverStatsCollectors() *observerStatsCollectors {
return &observerStatsCollectors{
observer: make(map[string]*observerStats),
}
}
func (list *observerStatsCollectors) GetStats(name string) *observerStats {
list.mu.Lock()
defer list.mu.Unlock()
stats, ok := list.observer[name]
if !ok {
stats = newObserverStats(name)
mon.Chain(stats)
list.observer[name] = stats
}
return stats
}
// observerStats tracks the most recent observer stats.
type observerStats struct {
mu sync.Mutex
key monkit.SeriesKey
total time.Duration
object *monkit.DurationDist
inline *monkit.DurationDist
remote *monkit.DurationDist
}
func newObserverStats(name string) *observerStats {
return &observerStats{
key: monkit.NewSeriesKey("observer").WithTag("name", name),
total: 0,
object: nil,
inline: nil,
remote: nil,
}
}
func (stats *observerStats) Observe(observer *observerContext) {
stats.mu.Lock()
defer stats.mu.Unlock()
stats.total = observer.object.Sum + observer.inline.Sum + observer.remote.Sum
stats.object = observer.object
stats.inline = observer.inline
stats.remote = observer.remote
}
func (stats *observerStats) Stats(cb func(key monkit.SeriesKey, field string, val float64)) {
stats.mu.Lock()
defer stats.mu.Unlock()
cb(stats.key, "sum", stats.total.Seconds())
if stats.object != nil {
stats.object.Stats(cb)
}
if stats.inline != nil {
stats.inline.Stats(cb)
}
if stats.remote != nil {
stats.remote.Stats(cb)
}
}

2
satellite/metabase/segmentloop/service_test.go
View File

@ -151,7 +151,7 @@ func TestSegmentsLoopObserverCancel(t *testing.T) {
UplinkCount: 1, UplinkCount: 1,
Reconfigure: testplanet.Reconfigure{ Reconfigure: testplanet.Reconfigure{
Satellite: func(log *zap.Logger, index int, config *satellite.Config) { Satellite: func(log *zap.Logger, index int, config *satellite.Config) {
config.Metainfo.Loop.CoalesceDuration = 1 * time.Second config.Metainfo.SegmentLoop.CoalesceDuration = 1 * time.Second
}, },
}, },
}, func(t *testing.T, ctx *testcontext.Context, planet *testplanet.Planet) { }, func(t *testing.T, ctx *testcontext.Context, planet *testplanet.Planet) {

2
satellite/metainfo/config.go
View File

@ -10,7 +10,6 @@ import (
"time" "time"
"storj.io/common/memory" "storj.io/common/memory"
"storj.io/storj/satellite/metabase/metaloop"
"storj.io/storj/satellite/metabase/segmentloop" "storj.io/storj/satellite/metabase/segmentloop"
"storj.io/storj/satellite/metainfo/piecedeletion" "storj.io/storj/satellite/metainfo/piecedeletion"
) )
@ -114,7 +113,6 @@ type Config struct {
MaxCommitInterval time.Duration `default:"48h" testDefault:"1h" help:"maximum time allowed to pass between creating and committing a segment"` MaxCommitInterval time.Duration `default:"48h" testDefault:"1h" help:"maximum time allowed to pass between creating and committing a segment"`
Overlay bool `default:"true" help:"toggle flag if overlay is enabled"` Overlay bool `default:"true" help:"toggle flag if overlay is enabled"`
RS RSConfig `releaseDefault:"29/35/80/110-256B" devDefault:"4/6/8/10-256B" help:"redundancy scheme configuration in the format k/m/o/n-sharesize"` RS RSConfig `releaseDefault:"29/35/80/110-256B" devDefault:"4/6/8/10-256B" help:"redundancy scheme configuration in the format k/m/o/n-sharesize"`
Loop metaloop.Config `help:"loop configuration"`
SegmentLoop segmentloop.Config `help:"segment loop configuration"` SegmentLoop segmentloop.Config `help:"segment loop configuration"`
RateLimiter RateLimiterConfig `help:"rate limiter configuration"` RateLimiter RateLimiterConfig `help:"rate limiter configuration"`
ProjectLimits ProjectLimitConfig `help:"project limit configuration"` ProjectLimits ProjectLimitConfig `help:"project limit configuration"`

15
scripts/testdata/satellite-config.yaml.lock vendored
View File

@ -367,21 +367,6 @@ identity.key-path: /root/.local/share/storj/identity/satellite/identity.key
# the database connection string to use # the database connection string to use
# metainfo.database-url: postgres:// # metainfo.database-url: postgres://
# as of system interval
# metainfo.loop.as-of-system-interval: -5m0s
# how long to wait for new observers before starting iteration
# metainfo.loop.coalesce-duration: 5s
# how many items to query in a batch
# metainfo.loop.list-limit: 2500
# rate limit (default is 0 which is unlimited segments per second)
# metainfo.loop.rate-limit: 0
# ratio where to consider processed count as supicious
# metainfo.loop.suspicious-processed-ratio: 0.03
# maximum time allowed to pass between creating and committing a segment # maximum time allowed to pass between creating and committing a segment
# metainfo.max-commit-interval: 48h0m0s # metainfo.max-commit-interval: 48h0m0s