diff --git a/CHANGELOG.md b/CHANGELOG.md index a458c0ee1b..5b05c92ea1 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -30,6 +30,7 @@ Changelog for NeoFS Node - Policer now uses an in-memory queue for redundant object removal, reducing metabase writes (#4216) - Session token storage now indexes tokens by expiration epoch for efficient cleanup (#4243) - Session key encryption now uses HPKE (#4247) +- SN object GET/HEAD server now traverses nodes strictly by weight, and falls back to local storage (#4232) ### Removed - fstree-decompress utility (#4223) diff --git a/internal/netmap/netmap.go b/internal/netmap/netmap.go index 61e3def1cd..8f7059148e 100644 --- a/internal/netmap/netmap.go +++ b/internal/netmap/netmap.go @@ -11,11 +11,13 @@ import ( // least one element from nodeSets. func NodeSetsContainPublicKeyFunc(nodeSets [][]netmap.NodeInfo, checkPubKeyFn func([]byte) bool) bool { return slices.ContainsFunc(nodeSets, func(nodeSet []netmap.NodeInfo) bool { - return nodeSetContainsPublicKeyFunc(nodeSet, checkPubKeyFn) + return NodeSetContainsPublicKeyFunc(nodeSet, checkPubKeyFn) }) } -func nodeSetContainsPublicKeyFunc(nodeSet []netmap.NodeInfo, checkPubKeyFn func([]byte) bool) bool { +// NodeSetContainsPublicKeyFunc checks whether checkPubKeyFn is true for at +// least one element from nodeSet. +func NodeSetContainsPublicKeyFunc(nodeSet []netmap.NodeInfo, checkPubKeyFn func([]byte) bool) bool { return slices.ContainsFunc(nodeSet, func(node netmap.NodeInfo) bool { return checkPubKeyFn(node.PublicKey()) }) diff --git a/pkg/services/object/get/container.go b/pkg/services/object/get/container.go index 6617505089..7ffcc1dbac 100644 --- a/pkg/services/object/get/container.go +++ b/pkg/services/object/get/container.go @@ -2,16 +2,13 @@ package getsvc import ( "context" + "errors" + inetmap "github.com/nspcc-dev/neofs-node/internal/netmap" "go.uber.org/zap" ) func (exec *execCtx) executeOnContainer() { - if exec.isLocal() { - exec.log.Debug("return result directly") - return - } - exec.log.Debug("trying to execute in container...") addr := exec.address() @@ -25,10 +22,29 @@ func (exec *execCtx) executeOnContainer() { exec.status = statusUndefined exec.err = err exec.log.Debug("failed to list storage nodes for the object", zap.Error(err)) + exec.logOtherError() return } } + var localDone bool + execLocal := func() bool { + if localDone { + return false + } + localDone = true + return exec.executeLocal() + } + + // try local node first if it is a primary holder + for i, n := range primaryCounts { + if inetmap.NodeSetContainsPublicKeyFunc(nodeLists[i][:n], exec.svc.neoFSNet.IsLocalNodePublicKey) { + if execLocal() { + return + } + } + } + ctx, cancel := context.WithCancel(exec.context()) defer cancel() @@ -50,14 +66,24 @@ func (exec *execCtx) executeOnContainer() { exec.log.Debug("interrupt placement iteration by context", zap.Error(ctx.Err()), ) - + if exec.err == nil { + exec.err = ctx.Err() + } + exec.logOtherError() return default: } bKey := nodeLists[i][j].PublicKey() strKey := string(bKey) - if _, ok := mProcessedNodes[strKey]; ok || exec.svc.neoFSNet.IsLocalNodePublicKey(bKey) { + if _, ok := mProcessedNodes[strKey]; ok { + continue + } + + if exec.svc.neoFSNet.IsLocalNodePublicKey(bKey) { + if execLocal() { + return + } continue } @@ -68,7 +94,10 @@ func (exec *execCtx) executeOnContainer() { // we reach the best result - split info with linking object ID. if exec.processNode(nodeLists[i][j]) { exec.log.Debug("completing the operation") - return + exec.analyzeContainerStatus() + if exec.responseStarted || !errors.Is(exec.err, errNoLinkNoLastPart) { + return + } } } @@ -78,4 +107,10 @@ func (exec *execCtx) executeOnContainer() { i = -1 } } + + // Local node is not in the container now. But it can still store the object. + // One scenario is when the container nodes have changed, the local node is no + // longer in it, and the object hasn't migrated yet. To cover such cases, lookup + // local storage as a last-ditch effort to find the object. + execLocal() } diff --git a/pkg/services/object/get/get.go b/pkg/services/object/get/get.go index 89120408f0..2dca37dc77 100644 --- a/pkg/services/object/get/get.go +++ b/pkg/services/object/get/get.go @@ -205,38 +205,62 @@ func (s *Service) get(ctx context.Context, prm commonPrm, opts ...execOption) st func (exec *execCtx) execute() { exec.log.Debug("serving request...") - // perform local operation - exec.executeLocal() + if exec.isLocal() { + exec.executeLocal() + exec.log.Debug("return result directly") + return + } - exec.analyzeStatus(true) + exec.executeOnContainer() } -func (exec *execCtx) analyzeStatus(execCnr bool) { +func (exec *execCtx) analyzeLocalStatus() bool { // analyze local result switch exec.status { case statusOK: - exec.log.Debug("operation finished successfully") + exec.logSuccess() case statusVIRTUAL: - exec.log.Debug("requested object is virtual") - if exec.collectOnly { - return - } - exec.assemble() - if errors.Is(exec.err, errNoLinkNoLastPart) && execCnr { - exec.executeOnContainer() - exec.analyzeStatus(false) - } + exec.handleVirtualObject() + return !errors.Is(exec.err, errNoLinkNoLastPart) case statusAPIResponse: - exec.log.Debug("received api response locally, return directly", zap.Error(exec.err)) - return + exec.logAPIError() default: - exec.log.Debug("operation finished with error", - zap.Error(exec.err), - ) + exec.logOtherError() + return exec.responseStarted + } - if execCnr && !exec.responseStarted { - exec.executeOnContainer() - exec.analyzeStatus(false) - } + return true +} + +func (exec *execCtx) analyzeContainerStatus() { + switch exec.status { + case statusOK: + exec.logSuccess() + case statusVIRTUAL: + exec.handleVirtualObject() + case statusAPIResponse: + exec.logAPIError() + default: + exec.logOtherError() + } +} + +func (exec *execCtx) handleVirtualObject() { + exec.log.Debug("requested object is virtual") + if exec.collectOnly { + return } + exec.assemble() +} + +func (exec *execCtx) logSuccess() { + exec.log.Debug("operation finished successfully") +} + +func (exec *execCtx) logAPIError() { + exec.log.Debug("received api response locally, return directly", zap.Error(exec.err)) +} + +func (exec *execCtx) logOtherError() { + exec.log.Debug("operation finished with error", zap.Error(exec.err)) } diff --git a/pkg/services/object/get/get_test.go b/pkg/services/object/get/get_test.go index f61eab9973..a6d57337da 100644 --- a/pkg/services/object/get/get_test.go +++ b/pkg/services/object/get/get_test.go @@ -92,7 +92,8 @@ func TestGetExtendedRangeOfECPart(t *testing.T) { } type testNeoFS struct { - vectors map[oid.Address][][]netmap.NodeInfo + vectors map[oid.Address][][]netmap.NodeInfo + localPublicKey []byte } type testClientCache struct { @@ -116,7 +117,9 @@ func newTestStorage() *testStorage { } } -func (g *testNeoFS) IsLocalNodePublicKey([]byte) bool { return false } +func (g *testNeoFS) IsLocalNodePublicKey(pubKey []byte) bool { + return bytes.Equal(g.localPublicKey, pubKey) +} func (g *testNeoFS) GetNodesForObject(addr oid.Address) ([][]netmap.NodeInfo, []uint, []iec.Rule, error) { nodeLists, ok := g.vectors[addr] @@ -1404,7 +1407,7 @@ func TestDoubleWriteHeaderOnPayloadReadFailure(t *testing.T) { err: readErr, } - anyNodeLists, nodeStrs := testNodeMatrix(t, []int{1}) + anyNodeLists, nodeStrs := testNodeMatrix(t, []int{2}) clientCache := &testClientCache{ clients: make(map[string]*testClient), @@ -1422,6 +1425,7 @@ func TestDoubleWriteHeaderOnPayloadReadFailure(t *testing.T) { vectors: map[oid.Address][][]netmap.NodeInfo{ addr: anyNodeLists, }, + localPublicKey: anyNodeLists[0][0].PublicKey(), } writer := &trackingWriter{} @@ -1452,7 +1456,7 @@ func TestDoubleWriteHeaderOnChunkWriteFailure(t *testing.T) { storage := newTestStorage() storage.addPhy(addr, obj) - anyNodeLists, nodeStrs := testNodeMatrix(t, []int{1}) + anyNodeLists, nodeStrs := testNodeMatrix(t, []int{2}) clientCache := &testClientCache{ clients: make(map[string]*testClient), @@ -1470,6 +1474,7 @@ func TestDoubleWriteHeaderOnChunkWriteFailure(t *testing.T) { vectors: map[oid.Address][][]netmap.NodeInfo{ addr: anyNodeLists, }, + localPublicKey: anyNodeLists[0][0].PublicKey(), } writeErr := errors.New("simulated chunk write error") diff --git a/pkg/services/object/get/local.go b/pkg/services/object/get/local.go index 438d11e36d..617513c381 100644 --- a/pkg/services/object/get/local.go +++ b/pkg/services/object/get/local.go @@ -13,7 +13,7 @@ import ( "go.uber.org/zap" ) -func (exec *execCtx) executeLocal() { +func (exec *execCtx) executeLocal() bool { var err error exec.collectedHeader, exec.collectedReader, err = exec.svc.localStorage.get(exec) @@ -40,7 +40,7 @@ func (exec *execCtx) executeLocal() { exec.status = statusNotFound exec.err = err - return + break } exec.status = statusAPIResponse @@ -50,6 +50,8 @@ func (exec *execCtx) executeLocal() { mergeSplitInfo(exec.splitInfo(), errSplitInfo.SplitInfo()) exec.err = object.NewSplitInfoError(exec.infoSplit) } + + return exec.analyzeLocalStatus() } func (s *Service) copyLocalObjectHeader(ctx context.Context, dst internal.HeaderWriter, cnr cid.ID, id oid.ID, raw bool) error {