-
Notifications
You must be signed in to change notification settings - Fork 390
/
db.go
274 lines (244 loc) · 12.3 KB
/
db.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
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
// Copyright (C) 2019 Storj Labs, Inc.
// See LICENSE for copying information.
package accounting
import (
"context"
"time"
"storj.io/common/memory"
"storj.io/common/storj"
"storj.io/common/uuid"
"storj.io/storj/satellite/compensation"
"storj.io/storj/satellite/metabase"
"storj.io/storj/satellite/orders"
)
// RollupStats is a convenience alias.
type RollupStats map[time.Time]map[storj.NodeID]*Rollup
// StoragenodeStorageTally mirrors dbx.StoragenodeStorageTally, allowing us to use that struct without leaking dbx.
type StoragenodeStorageTally struct {
ID int64
NodeID storj.NodeID
IntervalEndTime time.Time
DataTotal float64
}
// StoragenodeBandwidthRollup mirrors dbx.StoragenodeBandwidthRollup, allowing us to use the struct without leaking dbx.
type StoragenodeBandwidthRollup struct {
NodeID storj.NodeID
IntervalStart time.Time
Action uint
Settled uint64
}
// Rollup mirrors dbx.AccountingRollup, allowing us to use that struct without leaking dbx.
type Rollup struct {
ID int64
NodeID storj.NodeID
StartTime time.Time
PutTotal int64
GetTotal int64
GetAuditTotal int64
GetRepairTotal int64
PutRepairTotal int64
AtRestTotal float64
}
// StorageNodePeriodUsage represents a statement for a node for a compensation period.
type StorageNodePeriodUsage struct {
NodeID storj.NodeID
AtRestTotal float64
GetTotal int64
PutTotal int64
GetRepairTotal int64
PutRepairTotal int64
GetAuditTotal int64
}
// StorageNodeUsage is node at rest space usage over a period of time.
type StorageNodeUsage struct {
NodeID storj.NodeID
StorageUsed float64
Timestamp time.Time
}
// ProjectUsage consist of period total storage, egress
// and objects count per hour for certain Project in bytes.
type ProjectUsage struct {
Storage float64 `json:"storage"`
Egress int64 `json:"egress"`
SegmentCount float64 `json:"segmentCount"`
ObjectCount float64 `json:"objectCount"`
Since time.Time `json:"since"`
Before time.Time `json:"before"`
}
// ProjectObjectsSegments consist of period total objects and segments count for certain Project.
type ProjectObjectsSegments struct {
SegmentCount int64 `json:"segmentCount"`
ObjectCount int64 `json:"objectCount"`
}
// ProjectLimits contains the storage, bandwidth and segments limits.
type ProjectLimits struct {
Usage *int64
Bandwidth *int64
Segments *int64
}
// BucketUsage consist of total bucket usage for period.
type BucketUsage struct {
ProjectID uuid.UUID
BucketName string
Storage float64
Egress float64
ObjectCount int64
SegmentCount int64
Since time.Time
Before time.Time
}
// BucketUsageCursor holds info for bucket usage
// cursor pagination.
type BucketUsageCursor struct {
Search string
Limit uint
Page uint
}
// BucketUsagePage represents bucket usage page result.
type BucketUsagePage struct {
BucketUsages []BucketUsage
Search string
Limit uint
Offset uint64
PageCount uint
CurrentPage uint
TotalCount uint64
}
// BucketUsageRollup is total bucket usage info
// for certain period.
type BucketUsageRollup struct {
ProjectID uuid.UUID
BucketName []byte
TotalStoredData float64
TotalSegments float64
ObjectCount float64
MetadataSize float64
RepairEgress float64
GetEgress float64
AuditEgress float64
Since time.Time
Before time.Time
}
// StoragenodeAccounting stores information about bandwidth and storage usage for storage nodes.
//
// architecture: Database
type StoragenodeAccounting interface {
// SaveTallies records tallies of data at rest
SaveTallies(ctx context.Context, latestTally time.Time, nodeData map[storj.NodeID]float64) error
// GetTallies retrieves all tallies
GetTallies(ctx context.Context) ([]*StoragenodeStorageTally, error)
// GetTalliesSince retrieves all tallies since latestRollup
GetTalliesSince(ctx context.Context, latestRollup time.Time) ([]*StoragenodeStorageTally, error)
// GetBandwidthSince retrieves all bandwidth rollup entires since latestRollup
GetBandwidthSince(ctx context.Context, latestRollup time.Time, cb func(context.Context, *StoragenodeBandwidthRollup) error) error
// SaveRollup records tally and bandwidth rollup aggregations to the database
SaveRollup(ctx context.Context, latestTally time.Time, stats RollupStats) error
// LastTimestamp records and returns the latest last tallied time.
LastTimestamp(ctx context.Context, timestampType string) (time.Time, error)
// QueryPaymentInfo queries Nodes and Accounting_Rollup on nodeID
QueryPaymentInfo(ctx context.Context, start time.Time, end time.Time) ([]*CSVRow, error)
// QueryStorageNodePeriodUsage returns accounting statements for nodes for a given compensation period
QueryStorageNodePeriodUsage(ctx context.Context, period compensation.Period) ([]StorageNodePeriodUsage, error)
// QueryStorageNodeUsage returns slice of StorageNodeUsage for given period
QueryStorageNodeUsage(ctx context.Context, nodeID storj.NodeID, start time.Time, end time.Time) ([]StorageNodeUsage, error)
// DeleteTalliesBefore deletes all tallies prior to some time
DeleteTalliesBefore(ctx context.Context, latestRollup time.Time) error
// ArchiveRollupsBefore archives rollups older than a given time and returns num storagenode and bucket bandwidth rollups archived.
ArchiveRollupsBefore(ctx context.Context, before time.Time, batchSize int) (numArchivedNodeBW int, err error)
// GetRollupsSince retrieves all archived bandwidth rollup records since a given time. A hard limit batch size is used for results.
GetRollupsSince(ctx context.Context, since time.Time) ([]StoragenodeBandwidthRollup, error)
// GetArchivedRollupsSince retrieves all archived bandwidth rollup records since a given time. A hard limit batch size is used for results.
GetArchivedRollupsSince(ctx context.Context, since time.Time) ([]StoragenodeBandwidthRollup, error)
}
// ProjectAccounting stores information about bandwidth and storage usage for projects.
//
// architecture: Database
type ProjectAccounting interface {
// SaveTallies saves the latest project info
SaveTallies(ctx context.Context, intervalStart time.Time, bucketTallies map[metabase.BucketLocation]*BucketTally) error
// GetTallies retrieves all tallies
GetTallies(ctx context.Context) ([]BucketTally, error)
// CreateStorageTally creates a record for BucketStorageTally in the accounting DB table
CreateStorageTally(ctx context.Context, tally BucketStorageTally) error
// GetAllocatedBandwidthTotal returns the sum of GET bandwidth usage allocated for a projectID in the past time frame
GetAllocatedBandwidthTotal(ctx context.Context, projectID uuid.UUID, from time.Time) (int64, error)
// GetProjectBandwidth returns project allocated bandwidth for the specified year, month and day.
GetProjectBandwidth(ctx context.Context, projectID uuid.UUID, year int, month time.Month, day int, asOfSystemInterval time.Duration) (int64, error)
// GetProjectDailyBandwidth returns bandwidth (allocated and settled) for the specified day.
GetProjectDailyBandwidth(ctx context.Context, projectID uuid.UUID, year int, month time.Month, day int) (int64, int64, int64, error)
// DeleteProjectBandwidthBefore deletes project bandwidth rollups before the given time
DeleteProjectBandwidthBefore(ctx context.Context, before time.Time) error
// UpdateProjectUsageLimit updates project usage limit.
UpdateProjectUsageLimit(ctx context.Context, projectID uuid.UUID, limit memory.Size) error
// UpdateProjectBandwidthLimit updates project bandwidth limit.
UpdateProjectBandwidthLimit(ctx context.Context, projectID uuid.UUID, limit memory.Size) error
// UpdateProjectSegmentLimit updates project segment limit.
UpdateProjectSegmentLimit(ctx context.Context, projectID uuid.UUID, limit int64) error
// GetProjectStorageLimit returns project storage usage limit.
GetProjectStorageLimit(ctx context.Context, projectID uuid.UUID) (*int64, error)
// GetProjectBandwidthLimit returns project bandwidth usage limit.
GetProjectBandwidthLimit(ctx context.Context, projectID uuid.UUID) (*int64, error)
// GetProjectSegmentLimit returns the segment limit for a project ID.
GetProjectSegmentLimit(ctx context.Context, projectID uuid.UUID) (_ *int64, err error)
// GetProjectLimits returns current project limit for both storage and bandwidth.
GetProjectLimits(ctx context.Context, projectID uuid.UUID) (ProjectLimits, error)
// GetProjectTotal returns project usage summary for specified period of time.
GetProjectTotal(ctx context.Context, projectID uuid.UUID, since, before time.Time) (*ProjectUsage, error)
// GetProjectObjectsSegments returns project objects and segments for specified period of time.
GetProjectObjectsSegments(ctx context.Context, projectID uuid.UUID) (*ProjectObjectsSegments, error)
// GetBucketUsageRollups returns usage rollup per each bucket for specified period of time.
GetBucketUsageRollups(ctx context.Context, projectID uuid.UUID, since, before time.Time) ([]BucketUsageRollup, error)
// GetBucketTotals returns per bucket usage summary for specified period of time.
GetBucketTotals(ctx context.Context, projectID uuid.UUID, cursor BucketUsageCursor, since, before time.Time) (*BucketUsagePage, error)
// ArchiveRollupsBefore archives rollups older than a given time and returns number of bucket bandwidth rollups archived.
ArchiveRollupsBefore(ctx context.Context, before time.Time, batchSize int) (numArchivedBucketBW int, err error)
// GetRollupsSince retrieves all archived bandwidth rollup records since a given time. A hard limit batch size is used for results.
GetRollupsSince(ctx context.Context, since time.Time) ([]orders.BucketBandwidthRollup, error)
// GetArchivedRollupsSince retrieves all archived bandwidth rollup records since a given time. A hard limit batch size is used for results.
GetArchivedRollupsSince(ctx context.Context, since time.Time) ([]orders.BucketBandwidthRollup, error)
}
// Cache stores live information about project storage which has not yet been synced to ProjectAccounting.
//
// All the implementations must follow the convention of returning errors of one
// of the classes defined in this package.
//
// All the methods return:
//
// ErrInvalidArgument: an implementation may return if some parameter contain a
// value which isn't accepted, nonetheless, not all the implementations impose
// the same constraints on them.
//
// ErrSystemOrNetError: any method will return this if there is an error with
// the underlining system or the network.
//
// ErrKeyNotFound: returned when a key is not found.
//
// ErrUnexpectedValue: returned when a key or value stored in the underlying
// system isn't of the expected format or type according the business domain.
//
// architecture: Database
type Cache interface {
// GetProjectStorageUsage returns the project's storage usage.
GetProjectStorageUsage(ctx context.Context, projectID uuid.UUID) (totalUsed int64, err error)
// GetProjectBandwidthUsage returns the project's bandwidth usage.
GetProjectBandwidthUsage(ctx context.Context, projectID uuid.UUID, now time.Time) (currentUsed int64, err error)
// GetProjectSegmentUsage returns the project's segment usage.
GetProjectSegmentUsage(ctx context.Context, projectID uuid.UUID) (currentUsed int64, err error)
// UpdateProjectBandthUsage updates the project's bandwidth usage increasing
// it. The projectID is inserted to the increment when it doesn't exists,
// hence this method will never return ErrKeyNotFound error's class.
UpdateProjectBandwidthUsage(ctx context.Context, projectID uuid.UUID, increment int64, ttl time.Duration, now time.Time) error
// UpdateProjectSegmentUsage updates the project's segment usage increasing
// it. The projectID is inserted to the increment when it doesn't exists,
// hence this method will never return ErrKeyNotFound error's class.
UpdateProjectSegmentUsage(ctx context.Context, projectID uuid.UUID, increment int64, ttl time.Duration) error
// AddProjectStorageUsage adds to the projects storage usage the spacedUsed.
// The projectID is inserted to the spaceUsed when it doesn't exists, hence
// this method will never return ErrKeyNotFound.
AddProjectStorageUsage(ctx context.Context, projectID uuid.UUID, spaceUsed int64) error
// GetAllProjectTotals return the total projects' storage used space.
GetAllProjectTotals(ctx context.Context) (map[uuid.UUID]int64, error)
// Close the client, releasing any open resources. Once it's called any other
// method must be called.
Close() error
}