audit

type ByStreamIDAndPosition ¶ added in v1.68.2

type ByStreamIDAndPosition []Segment

ByStreamIDAndPosition allows sorting of a slice of segments by stream ID and position.

func (b ByStreamIDAndPosition) Len() int

func (b ByStreamIDAndPosition) Less(i, j int) bool

func (b ByStreamIDAndPosition) Swap(i, j int)

type Config ¶

type Config struct {
	MaxRetriesStatDB   int           `help:"max number of times to attempt updating a statdb batch" default:"3"`
	MinBytesPerSecond  memory.Size   `` /* 133-byte string literal not displayed */
	MinDownloadTimeout time.Duration `help:"the minimum duration for downloading a share from storage nodes before timing out" default:"15s" testDefault:"5s"`
	MaxReverifyCount   int           `help:"limit above which we consider an audit is failed" default:"3"`

	ChoreInterval             time.Duration `help:"how often to run the reservoir chore" releaseDefault:"24h" devDefault:"1m" testDefault:"$TESTINTERVAL"`
	QueueInterval             time.Duration `help:"how often to recheck an empty audit queue" releaseDefault:"1h" devDefault:"1m" testDefault:"$TESTINTERVAL"`
	Slots                     int           `help:"number of reservoir slots allotted for nodes, currently capped at 3" default:"3"`
	VerificationPushBatchSize int           `help:"number of audit jobs to push at once to the verification queue" devDefault:"10" releaseDefault:"4096"`
	WorkerConcurrency         int           `help:"number of workers to run audits on segments" default:"2"`
	UseRangedLoop             bool          `help:"whether use Audit observer with ranged loop." default:"true"`

	ReverifyWorkerConcurrency   int           `help:"number of workers to run reverify audits on pieces" default:"2"`
	ReverificationRetryInterval time.Duration `` /* 135-byte string literal not displayed */

	ContainmentSyncChoreInterval time.Duration `help:"how often to run the containment-sync chore" releaseDefault:"2h" devDefault:"2m" testDefault:"$TESTINTERVAL"`
}

Config contains configurable values for audit chore and workers.

type Containment ¶

type Containment interface {
	Get(ctx context.Context, nodeID pb.NodeID) (*ReverificationJob, error)
	Insert(ctx context.Context, job *PieceLocator) error
	Delete(ctx context.Context, job *PieceLocator) (wasDeleted, nodeStillContained bool, err error)
	GetAllContainedNodes(ctx context.Context) ([]pb.NodeID, error)
}

Containment holds information about pending audits for contained nodes.

architecture: Database

type ContainmentSyncChore ¶ added in v1.73.4

type ContainmentSyncChore struct {
	Loop *sync2.Cycle
	// contains filtered or unexported fields
}

ContainmentSyncChore is a chore to update the set of contained nodes in the overlay cache. This is necessary because it is possible for the "contained" field in the nodes table to disagree with whether a node appears in the reverification queue. We make an effort to keep them in sync when making changes to the reverification queue, but this infrequent chore will clean up any inconsistencies that creep in (because we can't maintain perfect consistency while the reverification queue and the nodes table may be in separate databases). Fortunately, it is acceptable for a node's containment status to be out of date for some amount of time.

func NewContainmentSyncChore(log *zap.Logger, queue ReverifyQueue, overlay overlay.DB, interval time.Duration) *ContainmentSyncChore

func (rc *ContainmentSyncChore) Run(ctx context.Context) (err error)

type FailurePhase ¶ added in v1.80.3

type FailurePhase int

FailurePhase indicates during which phase a GET_SHARE operation failed.

const (
	// NoFailure indicates there was no failure during a GET_SHARE operation.
	NoFailure FailurePhase = 0
	// DialFailure indicates a GET_SHARE operation failed during Dial.
	DialFailure FailurePhase = 1
	// RequestFailure indicates a GET_SHARE operation failed to make its RPC request, or the request failed.
	RequestFailure FailurePhase = 2
)

type Observer ¶ added in v1.70.1

type Observer struct {

	// The follow fields are reset on each segment loop cycle.
	Reservoirs map[metabase.NodeAlias]*Reservoir
	// contains filtered or unexported fields
}

Observer populates reservoirs and the audit queue.

architecture: Observer

func NewObserver(log *zap.Logger, queue VerifyQueue, config Config) *Observer

func (obs *Observer) Finish(ctx context.Context) (err error)

func (obs *Observer) Fork(ctx context.Context) (_ rangedloop.Partial, err error)

func (obs *Observer) Join(ctx context.Context, partial rangedloop.Partial) (err error)

func (obs *Observer) Start(ctx context.Context, startTime time.Time) (err error)

type Outcome ¶ added in v1.67.1

type Outcome int

Outcome enumerates the possible results of a piecewise audit.

Note that it is very similar to reputation.AuditType, but it is different in scope and needs a slightly different set of values.

const (
	// OutcomeNotPerformed indicates an audit was not performed, for any of a
	// variety of reasons, but that it should be reattempted later.
	OutcomeNotPerformed Outcome = iota
	// OutcomeNotNecessary indicates that an audit is no longer required,
	// for example because the segment has been updated or no longer exists.
	OutcomeNotNecessary
	// OutcomeSuccess indicates that an audit took place and the piece was
	// fully validated.
	OutcomeSuccess
	// OutcomeFailure indicates that an audit took place but that the node
	// failed the audit, either because it did not have the piece or the
	// data was incorrect.
	OutcomeFailure
	// OutcomeTimedOut indicates the audit could not be completed because
	// it took too long. The audit should be retried later.
	OutcomeTimedOut
	// OutcomeNodeOffline indicates that the audit could not be completed
	// because the node could not be contacted. The audit should be
	// retried later.
	OutcomeNodeOffline
	// OutcomeUnknownError indicates that the audit could not be completed
	// because of an error not otherwise expected or recognized. The
	// audit should be retried later.
	OutcomeUnknownError
)

type PieceAudit ¶ added in v1.40.3

type PieceAudit int

PieceAudit is piece audit status.

const (
	// PieceAuditUnknown is unknown piece audit.
	PieceAuditUnknown PieceAudit = iota
	// PieceAuditFailure is failed piece audit.
	PieceAuditFailure
	// PieceAuditOffline is offline node piece audit.
	PieceAuditOffline
	// PieceAuditContained is online but unresponsive node piece audit.
	PieceAuditContained
	// PieceAuditSuccess is successful piece audit.
	PieceAuditSuccess
)

func PieceAuditFromErr(err error) PieceAudit

type PieceLocator ¶ added in v1.67.1

type PieceLocator struct {
	StreamID uuid.UUID
	Position metabase.SegmentPosition
	NodeID   storj.NodeID
	PieceNum int
}

PieceLocator specifies all information necessary to look up a particular piece on a particular satellite.

type Report ¶

type Report struct {
	Segment *metabase.Segment

	Successes       storj.NodeIDList
	Fails           metabase.Pieces
	Offlines        storj.NodeIDList
	PendingAudits   []*ReverificationJob
	Unknown         storj.NodeIDList
	NodesReputation map[storj.NodeID]overlay.ReputationStatus
}

Report contains audit result. It records whether an audit is able to be completed, the total number of pieces a given audit has conducted for, lists for nodes that succeeded, failed, were offline, have pending audits, or failed for unknown reasons and their current reputation status.

type Reporter ¶

type Reporter interface {
	RecordAudits(ctx context.Context, req Report)
	ReportReverificationNeeded(ctx context.Context, piece *PieceLocator) (err error)
	RecordReverificationResult(ctx context.Context, pendingJob *ReverificationJob, outcome Outcome, reputation overlay.ReputationStatus) (err error)
}

Reporter records audit reports in the overlay and database.

func NewReporter(log *zap.Logger, reputations *reputation.Service, overlay *overlay.Service, metabase *metabase.DB, containment Containment, maxRetries int, maxReverifyCount int32) Reporter

type Reservoir ¶ added in v0.19.0

type Reservoir struct {
	// contains filtered or unexported fields
}

Reservoir holds a certain number of segments to reflect a random sample.

func NewReservoir(size int) *Reservoir

func (reservoir *Reservoir) Keys() []float64

func (reservoir *Reservoir) Merge(operand *Reservoir) error

func (reservoir *Reservoir) Sample(r *rand.Rand, segment rangedloop.Segment)

func (reservoir *Reservoir) Segments() []rangedloop.Segment

type ReverificationJob ¶ added in v1.67.1

type ReverificationJob struct {
	Locator       PieceLocator
	InsertedAt    time.Time
	ReverifyCount int
	LastAttempt   time.Time
}

ReverificationJob represents a job as received from the reverification audit queue.

type Reverifier ¶ added in v1.69.1

type Reverifier struct {
	*Verifier
	// contains filtered or unexported fields
}

Reverifier pulls jobs from the reverification queue and fulfills them by performing the requested reverifications.

architecture: Worker

func NewReverifier(log *zap.Logger, verifier *Verifier, db ReverifyQueue, config Config) *Reverifier

func (reverifier *Reverifier) DoReverifyPiece(ctx context.Context, logger *zap.Logger, locator *PieceLocator) (outcome Outcome, reputation overlay.ReputationStatus, err error)

func (reverifier *Reverifier) GetPiece(ctx context.Context, limit *pb.AddressedOrderLimit, piecePrivateKey storj.PiecePrivateKey, cachedIPAndPort string, pieceSize int32) (pieceData []byte, hash *pb.PieceHash, origLimit *pb.OrderLimit, err error)

func (reverifier *Reverifier) ReverifyPiece(ctx context.Context, logger *zap.Logger, locator *PieceLocator) (outcome Outcome, reputation overlay.ReputationStatus)

type ReverifyQueue ¶ added in v1.67.1

type ReverifyQueue interface {
	Insert(ctx context.Context, piece *PieceLocator) (err error)
	GetNextJob(ctx context.Context, retryInterval time.Duration) (job *ReverificationJob, err error)
	Remove(ctx context.Context, piece *PieceLocator) (wasDeleted bool, err error)
	GetByNodeID(ctx context.Context, nodeID storj.NodeID) (audit *ReverificationJob, err error)
	GetAllContainedNodes(ctx context.Context) ([]storj.NodeID, error)
}

ReverifyQueue controls manipulation of a queue of pieces to be _re_verified; that is, a node timed out when we requested an audit of the piece, and now we need to follow up with that node until we get a proper answer to the audit. (Or until we try too many times, and disqualify the node.)

type ReverifyWorker ¶ added in v1.70.1

type ReverifyWorker struct {
	Loop *sync2.Cycle
	// contains filtered or unexported fields
}

ReverifyWorker processes reverifications (retrying piece audits against nodes that timed out during a Verification).

func NewReverifyWorker(log *zap.Logger, queue ReverifyQueue, reverifier *Reverifier, reporter Reporter, config Config) *ReverifyWorker

func (worker *ReverifyWorker) Close() error

func (worker *ReverifyWorker) Run(ctx context.Context) (err error)

type Segment ¶ added in v1.26.2

type Segment struct {
	StreamID      uuid.UUID
	Position      metabase.SegmentPosition
	ExpiresAt     *time.Time
	EncryptedSize int32 // size of the whole segment (not a piece)
}

Segment is a segment to audit.

func NewSegment(loopSegment rangedloop.Segment) Segment

func (segment *Segment) Expired(now time.Time) bool

type Share ¶

type Share struct {
	Error        error
	FailurePhase FailurePhase
	PieceNum     int
	NodeID       storj.NodeID
	Data         []byte
}

Share represents required information about an audited share.

type Verifier ¶

type Verifier struct {
	OnTestingCheckSegmentAlteredHook func()
	// contains filtered or unexported fields
}

Verifier helps verify the correctness of a given stripe.

architecture: Worker

func NewVerifier(log *zap.Logger, metabase *metabase.DB, dialer rpc.Dialer, overlay *overlay.Service, containment Containment, orders *orders.Service, id *identity.FullIdentity, minBytesPerSecond memory.Size, minDownloadTimeout time.Duration) *Verifier

func (verifier *Verifier) DownloadShares(ctx context.Context, limits []*pb.AddressedOrderLimit, piecePrivateKey storj.PiecePrivateKey, cachedNodesInfo map[storj.NodeID]overlay.NodeReputation, stripeIndex int32, shareSize int32) (shares map[int]Share, err error)

func (verifier *Verifier) GetShare(ctx context.Context, limit *pb.AddressedOrderLimit, piecePrivateKey storj.PiecePrivateKey, cachedIPAndPort string, stripeIndex, shareSize int32, pieceNum int) (share Share)

func (verifier *Verifier) IdentifyContainedNodes(ctx context.Context, segment Segment) (skipList map[storj.NodeID]bool, err error)

func (verifier *Verifier) SetNow(nowFn func() time.Time)

func (verifier *Verifier) Verify(ctx context.Context, segment Segment, skip map[storj.NodeID]bool) (report Report, err error)

type VerifyQueue ¶ added in v1.67.1

type VerifyQueue interface {
	Push(ctx context.Context, segments []Segment, maxBatchSize int) (err error)
	Next(ctx context.Context) (Segment, error)
}

VerifyQueue controls manipulation of a database-based queue of segments to be verified; that is, segments chosen at random from all segments on the satellite, for which workers should perform audits. We will try to download a stripe of data across all pieces in the segment and ensure that all pieces conform to the same polynomial.

type Worker ¶ added in v0.21.0

type Worker struct {
	Loop *sync2.Cycle
	// contains filtered or unexported fields
}

Worker contains information for populating audit queue and processing audits.

func NewWorker(log *zap.Logger, queue VerifyQueue, verifier *Verifier, reverifyQueue ReverifyQueue, reporter Reporter, config Config) *Worker

func (worker *Worker) Close() error

func (worker *Worker) Run(ctx context.Context) (err error)