-
Notifications
You must be signed in to change notification settings - Fork 402
/
service.go
156 lines (130 loc) · 4.89 KB
/
service.go
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
// 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) (_ *overlay.ReputationStatus, changed bool, err error)
SetNodeStatus(ctx context.Context, id storj.NodeID, status overlay.ReputationStatus) 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) (err error)
// SuspendNodeUnknownAudit suspends a storage node for unknown audits.
SuspendNodeUnknownAudit(ctx context.Context, nodeID storj.NodeID, suspendedAt time.Time) (err error)
// UpdateAuditHistory updates a node's audit history
UpdateAuditHistory(ctx context.Context, oldHistory []byte, updateReq UpdateRequest, auditTime time.Time) (res *UpdateAuditHistoryResponse, err error)
}
// Info contains all reputation data to be stored in DB.
type Info struct {
AuditSuccessCount int64
TotalAuditCount int64
VettedAt *time.Time
Contained bool
Disqualified *time.Time
Suspended *time.Time
UnknownAuditSuspended *time.Time
OfflineSuspended *time.Time
UnderReview *time.Time
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, result AuditType) (err error) {
defer mon.Task()(&ctx)(&err)
statusUpdate, changed, 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,
}, time.Now())
if err != nil {
return err
}
if changed {
err = service.overlay.UpdateReputation(ctx, nodeID, statusUpdate)
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) (err error) {
err = service.db.DisqualifyNode(ctx, nodeID)
if err != nil {
return err
}
return service.overlay.DisqualifyNode(ctx, nodeID)
}
// 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 }