diff --git a/internal/logs/logs.go b/internal/logs/logs.go index a72609b4..4b8305e4 100644 --- a/internal/logs/logs.go +++ b/internal/logs/logs.go @@ -519,4 +519,8 @@ const ( FailedToCountWritecacheItems = "failed to count writecache items" AttemtToCloseAlreadyClosedBlobovnicza = "attempt to close an already closed blobovnicza" FailedToGetContainerCounters = "failed to get container counters values" + FailedToRebuildBlobstore = "failed to rebuild blobstore" + BlobstoreRebuildStarted = "blobstore rebuild started" + BlobstoreRebuildCompletedSuccessfully = "blobstore rebuild completed successfully" + BlobstoreRebuildStopped = "blobstore rebuild stopped" ) diff --git a/pkg/local_object_storage/blobstor/rebuild.go b/pkg/local_object_storage/blobstor/rebuild.go new file mode 100644 index 00000000..110735fa --- /dev/null +++ b/pkg/local_object_storage/blobstor/rebuild.go @@ -0,0 +1,15 @@ +package blobstor + +import ( + "context" + + objectSDK "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/object" +) + +type StorageIDUpdate interface { + UpdateStorageID(ctx context.Context, obj *objectSDK.Object, storageID []byte) error +} + +func (b *BlobStor) Rebuild(_ context.Context, _ StorageIDUpdate) error { + return nil +} diff --git a/pkg/local_object_storage/shard/control.go b/pkg/local_object_storage/shard/control.go index 257498b3..a988683a 100644 --- a/pkg/local_object_storage/shard/control.go +++ b/pkg/local_object_storage/shard/control.go @@ -162,6 +162,9 @@ func (s *Shard) Init(ctx context.Context) error { s.gc.init(ctx) + s.rb = newRebuilder() + s.rb.Start(ctx, s.blobStor, s.metaBase, s.log) + return nil } @@ -266,6 +269,9 @@ func (s *Shard) refillTombstoneObject(ctx context.Context, obj *objectSDK.Object // Close releases all Shard's components. func (s *Shard) Close() error { + if s.rb != nil { + s.rb.Stop(s.log) + } components := []interface{ Close() error }{} if s.pilorama != nil { @@ -310,6 +316,11 @@ func (s *Shard) Reload(ctx context.Context, opts ...Option) error { unlock := s.lockExclusive() defer unlock() + s.rb.Stop(s.log) + defer func() { + s.rb.Start(ctx, s.blobStor, s.metaBase, s.log) + }() + ok, err := s.metaBase.Reload(c.metaOpts...) if err != nil { if errors.Is(err, meta.ErrDegradedMode) { diff --git a/pkg/local_object_storage/shard/rebuilder.go b/pkg/local_object_storage/shard/rebuilder.go new file mode 100644 index 00000000..3e0c66ab --- /dev/null +++ b/pkg/local_object_storage/shard/rebuilder.go @@ -0,0 +1,89 @@ +package shard + +import ( + "context" + "errors" + "sync" + + "git.frostfs.info/TrueCloudLab/frostfs-node/internal/logs" + "git.frostfs.info/TrueCloudLab/frostfs-node/pkg/local_object_storage/blobstor" + meta "git.frostfs.info/TrueCloudLab/frostfs-node/pkg/local_object_storage/metabase" + "git.frostfs.info/TrueCloudLab/frostfs-node/pkg/util/logger" + objectSDK "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/object" + "go.uber.org/zap" +) + +type rebuilder struct { + mtx *sync.Mutex + wg *sync.WaitGroup + cancel func() +} + +func newRebuilder() *rebuilder { + return &rebuilder{ + mtx: &sync.Mutex{}, + wg: &sync.WaitGroup{}, + cancel: nil, + } +} + +func (r *rebuilder) Start(ctx context.Context, bs *blobstor.BlobStor, mb *meta.DB, log *logger.Logger) { + r.mtx.Lock() + defer r.mtx.Unlock() + + r.start(ctx, bs, mb, log) +} + +func (r *rebuilder) start(ctx context.Context, bs *blobstor.BlobStor, mb *meta.DB, log *logger.Logger) { + if r.cancel != nil { + r.stop(log) + } + ctx, cancel := context.WithCancel(ctx) + r.cancel = cancel + r.wg.Add(1) + go func() { + defer r.wg.Done() + + log.Info(logs.BlobstoreRebuildStarted) + if err := bs.Rebuild(ctx, &mbStorageIDUpdate{mb: mb}); err != nil { + log.Warn(logs.FailedToRebuildBlobstore, zap.Error(err)) + } else { + log.Info(logs.BlobstoreRebuildCompletedSuccessfully) + } + }() +} + +func (r *rebuilder) Stop(log *logger.Logger) { + r.mtx.Lock() + defer r.mtx.Unlock() + + r.stop(log) +} + +func (r *rebuilder) stop(log *logger.Logger) { + if r.cancel == nil { + return + } + + r.cancel() + r.wg.Wait() + r.cancel = nil + log.Info(logs.BlobstoreRebuildStopped) +} + +var errMBIsNotAvailable = errors.New("metabase is not available") + +type mbStorageIDUpdate struct { + mb *meta.DB +} + +func (u *mbStorageIDUpdate) UpdateStorageID(ctx context.Context, obj *objectSDK.Object, storageID []byte) error { + if u.mb == nil { + return errMBIsNotAvailable + } + var prm meta.PutPrm + prm.SetObject(obj) + prm.SetStorageID(storageID) + _, err := u.mb.Put(ctx, prm) + return err +} diff --git a/pkg/local_object_storage/shard/shard.go b/pkg/local_object_storage/shard/shard.go index 2c4ce91d..915feccb 100644 --- a/pkg/local_object_storage/shard/shard.go +++ b/pkg/local_object_storage/shard/shard.go @@ -39,6 +39,8 @@ type Shard struct { tsSource TombstoneSource + rb *rebuilder + gcCancel atomic.Value setModeRequested atomic.Bool }