951a842fd4
This functionality will be needed in both packages, so here we move it into the more general reputation-code package and export it for use in satellitedb. This also removes the related UpdateAuditHistory() signature from the reputation DB interface, since it doesn't have anything to do with the db. It doesn't need to be a method, either. Finally, this changes the test for addAudit to be a plain test function instead of using testplanet.Run(). It didn't need a whole testplanet setup or any databases. Refs: https://github.com/storj/storj/issues/4601 Change-Id: I90f6a909e5404f03ad776b95cfa2f248308c57c1
196 lines
6.7 KiB
Go
196 lines
6.7 KiB
Go
// Copyright (C) 2021 Storj Labs, Inc.
|
|
// See LICENSE for copying information.
|
|
|
|
package reputation
|
|
|
|
import (
|
|
"context"
|
|
"time"
|
|
|
|
"go.uber.org/zap"
|
|
|
|
"storj.io/common/storj"
|
|
"storj.io/storj/satellite/overlay"
|
|
)
|
|
|
|
// DB is an interface for storing reputation data.
|
|
type DB interface {
|
|
Update(ctx context.Context, request UpdateRequest, now time.Time) (_ *Info, err error)
|
|
Get(ctx context.Context, nodeID storj.NodeID) (*Info, error)
|
|
|
|
// UnsuspendNodeUnknownAudit unsuspends a storage node for unknown audits.
|
|
UnsuspendNodeUnknownAudit(ctx context.Context, nodeID storj.NodeID) (err error)
|
|
// DisqualifyNode disqualifies a storage node.
|
|
DisqualifyNode(ctx context.Context, nodeID storj.NodeID, disqualifiedAt time.Time, reason overlay.DisqualificationReason) (err error)
|
|
// SuspendNodeUnknownAudit suspends a storage node for unknown audits.
|
|
SuspendNodeUnknownAudit(ctx context.Context, nodeID storj.NodeID, suspendedAt time.Time) (err error)
|
|
}
|
|
|
|
// Info contains all reputation data to be stored in DB.
|
|
type Info struct {
|
|
AuditSuccessCount int64
|
|
TotalAuditCount int64
|
|
VettedAt *time.Time
|
|
UnknownAuditSuspended *time.Time
|
|
OfflineSuspended *time.Time
|
|
UnderReview *time.Time
|
|
Disqualified *time.Time
|
|
DisqualificationReason overlay.DisqualificationReason
|
|
OnlineScore float64
|
|
AuditHistory AuditHistory
|
|
AuditReputationAlpha float64
|
|
AuditReputationBeta float64
|
|
UnknownAuditReputationAlpha float64
|
|
UnknownAuditReputationBeta float64
|
|
}
|
|
|
|
// Service handles storing node reputation data and updating
|
|
// the overlay cache when a node's status changes.
|
|
type Service struct {
|
|
log *zap.Logger
|
|
overlay overlay.DB
|
|
db DB
|
|
config Config
|
|
}
|
|
|
|
// NewService creates a new reputation service.
|
|
func NewService(log *zap.Logger, overlay overlay.DB, db DB, config Config) *Service {
|
|
return &Service{
|
|
log: log,
|
|
overlay: overlay,
|
|
db: db,
|
|
config: config,
|
|
}
|
|
}
|
|
|
|
// ApplyAudit receives an audit result and applies it to the relevant node in DB.
|
|
func (service *Service) ApplyAudit(ctx context.Context, nodeID storj.NodeID, reputation overlay.ReputationStatus, result AuditType) (err error) {
|
|
defer mon.Task()(&ctx)(&err)
|
|
|
|
now := time.Now()
|
|
statusUpdate, err := service.db.Update(ctx, UpdateRequest{
|
|
NodeID: nodeID,
|
|
AuditOutcome: result,
|
|
|
|
AuditLambda: service.config.AuditLambda,
|
|
AuditWeight: service.config.AuditWeight,
|
|
AuditDQ: service.config.AuditDQ,
|
|
SuspensionGracePeriod: service.config.SuspensionGracePeriod,
|
|
SuspensionDQEnabled: service.config.SuspensionDQEnabled,
|
|
AuditsRequiredForVetting: service.config.AuditCount,
|
|
AuditHistory: service.config.AuditHistory,
|
|
}, now)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
// only update node if its health status has changed, or it's a newly vetted
|
|
// node.
|
|
// this prevents the need to require caller of ApplyAudit() to always know
|
|
// the VettedAt time for a node.
|
|
// Due to inconsistencies in the precision of time.Now() on different platforms and databases, the time comparison
|
|
// for the VettedAt status is done using time values that are truncated to second precision.
|
|
if hasReputationChanged(*statusUpdate, reputation, now) {
|
|
reputationUpdate := &overlay.ReputationUpdate{
|
|
Disqualified: statusUpdate.Disqualified,
|
|
DisqualificationReason: statusUpdate.DisqualificationReason,
|
|
UnknownAuditSuspended: statusUpdate.UnknownAuditSuspended,
|
|
OfflineSuspended: statusUpdate.OfflineSuspended,
|
|
VettedAt: statusUpdate.VettedAt,
|
|
}
|
|
err = service.overlay.UpdateReputation(ctx, nodeID, *reputationUpdate)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
}
|
|
|
|
return err
|
|
}
|
|
|
|
// Get returns a node's reputation info from DB.
|
|
// If a node is not found in the DB, default reputation information is returned.
|
|
func (service *Service) Get(ctx context.Context, nodeID storj.NodeID) (info *Info, err error) {
|
|
defer mon.Task()(&ctx)(&err)
|
|
|
|
info, err = service.db.Get(ctx, nodeID)
|
|
if err != nil {
|
|
if ErrNodeNotFound.Has(err) {
|
|
// if there is no audit reputation for the node, that's fine and we
|
|
// return default reputation values
|
|
info = &Info{
|
|
UnknownAuditReputationAlpha: 1,
|
|
AuditReputationAlpha: 1,
|
|
OnlineScore: 1,
|
|
}
|
|
|
|
return info, nil
|
|
}
|
|
|
|
return nil, Error.Wrap(err)
|
|
}
|
|
|
|
return info, nil
|
|
}
|
|
|
|
// TestSuspendNodeUnknownAudit suspends a storage node for unknown audits.
|
|
func (service *Service) TestSuspendNodeUnknownAudit(ctx context.Context, nodeID storj.NodeID, suspendedAt time.Time) (err error) {
|
|
err = service.db.SuspendNodeUnknownAudit(ctx, nodeID, suspendedAt)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
return service.overlay.TestSuspendNodeUnknownAudit(ctx, nodeID, suspendedAt)
|
|
}
|
|
|
|
// TestDisqualifyNode disqualifies a storage node.
|
|
func (service *Service) TestDisqualifyNode(ctx context.Context, nodeID storj.NodeID, reason overlay.DisqualificationReason) (err error) {
|
|
disqualifiedAt := time.Now()
|
|
|
|
err = service.db.DisqualifyNode(ctx, nodeID, disqualifiedAt, reason)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
return service.overlay.DisqualifyNode(ctx, nodeID, disqualifiedAt, reason)
|
|
}
|
|
|
|
// TestUnsuspendNodeUnknownAudit unsuspends a storage node for unknown audits.
|
|
func (service *Service) TestUnsuspendNodeUnknownAudit(ctx context.Context, nodeID storj.NodeID) (err error) {
|
|
err = service.db.UnsuspendNodeUnknownAudit(ctx, nodeID)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
return service.overlay.TestUnsuspendNodeUnknownAudit(ctx, nodeID)
|
|
}
|
|
|
|
// Close closes resources.
|
|
func (service *Service) Close() error { return nil }
|
|
|
|
// hasReputationChanged determines if the current node reputation is different from the newly updated reputation. This
|
|
// function will only consider the Disqualified, UnknownAudiSuspended and OfflineSuspended statuses for changes.
|
|
func hasReputationChanged(updated Info, current overlay.ReputationStatus, now time.Time) bool {
|
|
if statusChanged(current.Disqualified, updated.Disqualified) ||
|
|
statusChanged(current.UnknownAuditSuspended, updated.UnknownAuditSuspended) ||
|
|
statusChanged(current.OfflineSuspended, updated.OfflineSuspended) {
|
|
return true
|
|
}
|
|
// check for newly vetted nodes.
|
|
// Due to inconsistencies in the precision of time.Now() on different platforms and databases, the time comparison
|
|
// for the VettedAt status is done using time values that are truncated to second precision.
|
|
if updated.VettedAt != nil && updated.VettedAt.Truncate(time.Second).Equal(now.Truncate(time.Second)) {
|
|
return true
|
|
}
|
|
return false
|
|
}
|
|
|
|
// statusChanged determines if the two given statuses are different.
|
|
func statusChanged(s1, s2 *time.Time) bool {
|
|
if s1 == nil && s2 == nil {
|
|
return false
|
|
} else if s1 != nil && s2 != nil {
|
|
return !s1.Equal(*s1)
|
|
}
|
|
return true
|
|
}
|