package blobovniczatree import ( "context" "path/filepath" "time" "git.frostfs.info/TrueCloudLab/frostfs-node/internal/logs" "git.frostfs.info/TrueCloudLab/frostfs-node/pkg/local_object_storage/blobovnicza" "git.frostfs.info/TrueCloudLab/frostfs-node/pkg/local_object_storage/blobstor/common" tracingPkg "git.frostfs.info/TrueCloudLab/frostfs-node/pkg/tracing" "git.frostfs.info/TrueCloudLab/frostfs-observability/tracing" "go.opentelemetry.io/otel/attribute" "go.opentelemetry.io/otel/trace" "go.uber.org/zap" ) // Put saves object in the maximum weight blobobnicza. // // returns error if could not save object in any blobovnicza. func (b *Blobovniczas) Put(ctx context.Context, prm common.PutPrm) (common.PutRes, error) { var ( success bool size int startedAt = time.Now() ) defer func() { b.metrics.Put(time.Since(startedAt), size, success) }() _, span := tracing.StartSpanFromContext(ctx, "Blobovniczas.Put", trace.WithAttributes( attribute.String("address", prm.Address.EncodeToString()), attribute.Bool("dont_compress", prm.DontCompress), )) defer span.End() if b.readOnly { return common.PutRes{}, common.ErrReadOnly } if !prm.DontCompress { prm.RawData = b.compression.Compress(prm.RawData) } size = len(prm.RawData) var putPrm blobovnicza.PutPrm putPrm.SetAddress(prm.Address) putPrm.SetMarshaledObject(prm.RawData) it := &putIterator{ B: b, ID: nil, AllFull: true, PutPrm: putPrm, } if err := b.iterateDeepest(ctx, prm.Address, func(s string) (bool, error) { return it.iterate(ctx, s) }); err != nil { return common.PutRes{}, err } else if it.ID == nil { if it.AllFull { return common.PutRes{}, common.ErrNoSpace } return common.PutRes{}, errPutFailed } success = true return common.PutRes{StorageID: it.ID.Bytes()}, nil } type putIterator struct { B *Blobovniczas ID *blobovnicza.ID AllFull bool PutPrm blobovnicza.PutPrm } func (i *putIterator) iterate(ctx context.Context, lvlPath string) (bool, error) { active, err := i.B.activeDBManager.GetOpenedActiveDBForLevel(lvlPath) if err != nil { if !isLogical(err) { i.B.reportError(logs.BlobovniczatreeCouldNotGetActiveBlobovnicza, err) } else { i.B.log.Debug(logs.BlobovniczatreeCouldNotGetActiveBlobovnicza, zap.String("error", err.Error()), zap.String("trace_id", tracingPkg.GetTraceID(ctx))) } return false, nil } if active == nil { i.B.log.Debug(logs.BlobovniczatreeBlobovniczaOverflowed, zap.String("level", lvlPath), zap.String("trace_id", tracingPkg.GetTraceID(ctx))) return false, nil } defer active.Close() i.AllFull = false _, err = active.Blobovnicza().Put(ctx, i.PutPrm) if err != nil { if !isLogical(err) { i.B.reportError(logs.BlobovniczatreeCouldNotPutObjectToActiveBlobovnicza, err) } else { i.B.log.Debug(logs.BlobovniczatreeCouldNotPutObjectToActiveBlobovnicza, zap.String("path", active.Path()), zap.String("error", err.Error()), zap.String("trace_id", tracingPkg.GetTraceID(ctx))) } return false, nil } idx := u64FromHexString(filepath.Base(active.Path())) i.ID = blobovnicza.NewIDFromBytes([]byte(filepath.Join(lvlPath, u64ToHexString(idx)))) return true, nil }