frostfs-node/pkg/services/audit/auditor/por.go
Evgenii Stratonikov 0e31c12e63 [#240] logs: Move log messages to constants
Drop duplicate entities.
Format entities.

Signed-off-by: Dmitrii Stepanov <d.stepanov@yadro.com>
Signed-off-by: Evgenii Stratonikov <e.stratonikov@yadro.com>
2023-04-14 05:06:09 +00:00

156 lines
3.7 KiB
Go

package auditor
import (
"bytes"
"context"
"sync"
"git.frostfs.info/TrueCloudLab/frostfs-node/internal/logs"
"git.frostfs.info/TrueCloudLab/frostfs-node/pkg/services/object_manager/placement"
"git.frostfs.info/TrueCloudLab/frostfs-node/pkg/util/rand"
containerSDK "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/container"
"git.frostfs.info/TrueCloudLab/frostfs-sdk-go/netmap"
oid "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/object/id"
storagegroupSDK "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/storagegroup"
"git.frostfs.info/TrueCloudLab/tzhash/tz"
"go.uber.org/zap"
)
func (c *Context) executePoR(ctx context.Context) {
wg := new(sync.WaitGroup)
sgs := c.task.StorageGroupList()
for _, sg := range sgs {
wg.Add(1)
if err := c.porWorkerPool.Submit(func() {
c.checkStorageGroupPoR(ctx, sg.ID(), sg.StorageGroup())
wg.Done()
}); err != nil {
wg.Done()
}
}
wg.Wait()
c.porWorkerPool.Release()
c.report.SetPoRCounters(c.porRequests.Load(), c.porRetries.Load())
}
func (c *Context) checkStorageGroupPoR(ctx context.Context, sgID oid.ID, sg storagegroupSDK.StorageGroup) {
members := sg.Members()
c.updateSGInfo(sgID, members)
var (
tzHash []byte
totalSize uint64
accRequests, accRetries uint32
)
var getHeaderPrm GetHeaderPrm
getHeaderPrm.CID = c.task.ContainerID()
getHeaderPrm.NodeIsRelay = true
homomorphicHashingEnabled := !containerSDK.IsHomomorphicHashingDisabled(c.task.ContainerStructure())
for i := range members {
flat, ok := c.getShuffledNodes(members[i], sgID)
if !ok {
continue
}
getHeaderPrm.OID = members[i]
for j := range flat {
accRequests++
if j > 0 { // in best case audit get object header on first iteration
accRetries++
}
getHeaderPrm.Node = flat[j]
hdr, err := c.cnrCom.GetHeader(ctx, getHeaderPrm)
if err != nil {
c.log.Debug(logs.AuditorCantHeadObject,
zap.String("remote_node", netmap.StringifyPublicKey(flat[j])),
zap.Stringer("oid", members[i]),
)
continue
}
// update cache for PoR and PDP audit checks
c.updateHeadResponses(hdr)
if homomorphicHashingEnabled {
cs, _ := hdr.PayloadHomomorphicHash()
if len(tzHash) == 0 {
tzHash = cs.Value()
} else {
tzHash, err = tz.Concat([][]byte{
tzHash,
cs.Value(),
})
if err != nil {
c.log.Debug(logs.AuditorCantConcatenateTzHash,
zap.String("oid", members[i].String()),
zap.String("error", err.Error()))
break
}
}
}
totalSize += hdr.PayloadSize()
break
}
}
c.porRequests.Add(accRequests)
c.porRetries.Add(accRetries)
sizeCheck := sg.ValidationDataSize() == totalSize
cs, _ := sg.ValidationDataHash()
tzCheck := !homomorphicHashingEnabled || bytes.Equal(tzHash, cs.Value())
c.writeCheckReport(sizeCheck, tzCheck, sgID, sg, totalSize)
}
func (c *Context) writeCheckReport(sizeCheck, tzCheck bool, sgID oid.ID, sg storagegroupSDK.StorageGroup, totalSize uint64) {
if sizeCheck && tzCheck {
c.report.PassedPoR(sgID)
} else {
if !sizeCheck {
c.log.Debug(logs.AuditorStorageGroupSizeCheckFailed,
zap.Uint64("expected", sg.ValidationDataSize()),
zap.Uint64("got", totalSize))
}
if !tzCheck {
c.log.Debug(logs.AuditorStorageGroupTzHashCheckFailed)
}
c.report.FailedPoR(sgID)
}
}
func (c *Context) getShuffledNodes(member oid.ID, sgID oid.ID) ([]netmap.NodeInfo, bool) {
objectPlacement, err := c.buildPlacement(member)
if err != nil {
c.log.Info(logs.AuditorCantBuildPlacementForStorageGroupMember,
zap.Stringer("sg", sgID),
zap.String("member_id", member.String()),
)
return nil, false
}
flat := placement.FlattenNodes(objectPlacement)
rand.Shuffle(len(flat), func(i, j int) {
flat[i], flat[j] = flat[j], flat[i]
})
return flat, true
}