-
Notifications
You must be signed in to change notification settings - Fork 393
/
main.go
367 lines (309 loc) · 10 KB
/
main.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
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
// Copyright (C) 2019 Storj Labs, Inc.
// See LICENSE for copying information.
package main
import (
"context"
"encoding/base64"
"encoding/csv"
"flag"
"fmt"
"os"
"strconv"
"github.com/spf13/cobra"
"github.com/zeebo/errs"
"storj.io/common/identity"
"storj.io/common/rpc"
"storj.io/common/storj"
"storj.io/private/process"
_ "storj.io/storj/private/version" // This attaches version information during release builds.
"storj.io/storj/satellite/internalpb"
"storj.io/uplink/private/eestream"
)
var (
// Addr is the address of peer from command flags.
Addr = flag.String("address", "127.0.0.1:7778", "address of peer to inspect")
// IdentityPath is the path to the identity the inspector should use for network communication.
IdentityPath = flag.String("identity-path", "", "path to the identity certificate for use on the network")
// CSVPath is the csv path where command output is written.
CSVPath string
// ErrInspectorDial throws when there are errors dialing the inspector server.
ErrInspectorDial = errs.Class("dialing inspector server")
// ErrRequest is for request errors after dialing.
ErrRequest = errs.Class("processing request")
// ErrIdentity is for errors during identity creation for this CLI.
ErrIdentity = errs.Class("creating identity")
// ErrArgs throws when there are errors with CLI args.
ErrArgs = errs.Class("invalid CLI args")
// Commander CLI.
rootCmd = &cobra.Command{
Use: "inspector",
Short: "CLI for interacting with Storj network",
}
statsCmd = &cobra.Command{
Use: "statdb",
Short: "commands for statdb",
}
healthCmd = &cobra.Command{
Use: "health",
Short: "commands for querying health of a stored data",
}
objectHealthCmd = &cobra.Command{
Use: "object <project-id> <bucket> <encrypted-path>",
Short: "Get stats about an object's health",
Args: cobra.MinimumNArgs(3),
RunE: ObjectHealth,
}
segmentHealthCmd = &cobra.Command{
Use: "segment <project-id> <segment-index> <bucket> <encrypted-path>",
Short: "Get stats about a segment's health",
Args: cobra.MinimumNArgs(4),
RunE: SegmentHealth,
}
)
// Inspector gives access to overlay.
type Inspector struct {
conn *rpc.Conn
identity *identity.FullIdentity
healthclient internalpb.DRPCHealthInspectorClient
}
// NewInspector creates a new inspector client for access to overlay.
func NewInspector(ctx context.Context, address, path string) (*Inspector, error) {
id, err := identity.Config{
CertPath: fmt.Sprintf("%s/identity.cert", path),
KeyPath: fmt.Sprintf("%s/identity.key", path),
}.Load()
if err != nil {
return nil, ErrIdentity.Wrap(err)
}
conn, err := rpc.NewDefaultDialer(nil).DialAddressUnencrypted(ctx, address)
if err != nil {
return &Inspector{}, ErrInspectorDial.Wrap(err)
}
return &Inspector{
conn: conn,
identity: id,
healthclient: internalpb.NewDRPCHealthInspectorClient(conn),
}, nil
}
// Close closes the inspector.
func (i *Inspector) Close() error { return i.conn.Close() }
// ObjectHealth gets information about the health of an object on the network.
func ObjectHealth(cmd *cobra.Command, args []string) (err error) {
ctx, _ := process.Ctx(cmd)
i, err := NewInspector(ctx, *Addr, *IdentityPath)
if err != nil {
return ErrArgs.Wrap(err)
}
defer func() { err = errs.Combine(err, i.Close()) }()
startAfterSegment := int64(0) // start from first segment
endBeforeSegment := int64(0) // No end, so we stop when we've hit limit or arrived at the last segment
limit := int64(0) // No limit, so we stop when we've arrived at the last segment
switch len(args) {
case 6:
limit, err = strconv.ParseInt(args[5], 10, 64)
if err != nil {
return ErrRequest.Wrap(err)
}
fallthrough
case 5:
endBeforeSegment, err = strconv.ParseInt(args[4], 10, 64)
if err != nil {
return ErrRequest.Wrap(err)
}
fallthrough
case 4:
startAfterSegment, err = strconv.ParseInt(args[3], 10, 64)
if err != nil {
return ErrRequest.Wrap(err)
}
fallthrough
default:
}
decodedPath, err := base64.URLEncoding.DecodeString(args[2])
if err != nil {
return err
}
req := &internalpb.ObjectHealthRequest{
ProjectId: []byte(args[0]),
Bucket: []byte(args[1]),
EncryptedPath: decodedPath,
StartAfterSegment: startAfterSegment,
EndBeforeSegment: endBeforeSegment,
Limit: int32(limit),
}
resp, err := i.healthclient.ObjectHealth(ctx, req)
if err != nil {
return ErrRequest.Wrap(err)
}
f, err := csvOutput()
if err != nil {
return err
}
defer func() {
err := f.Close()
if err != nil {
fmt.Printf("error closing file: %+v\n", err)
}
}()
w := csv.NewWriter(f)
defer w.Flush()
redundancy, err := eestream.NewRedundancyStrategyFromProto(resp.GetRedundancy())
if err != nil {
return ErrRequest.Wrap(err)
}
if err := printRedundancyTable(w, redundancy); err != nil {
return err
}
if err := printSegmentHealthAndNodeTables(w, redundancy, resp.GetSegments()); err != nil {
return err
}
return nil
}
// SegmentHealth gets information about the health of a segment on the network.
func SegmentHealth(cmd *cobra.Command, args []string) (err error) {
ctx, _ := process.Ctx(cmd)
i, err := NewInspector(ctx, *Addr, *IdentityPath)
if err != nil {
return ErrArgs.Wrap(err)
}
defer func() { err = errs.Combine(err, i.Close()) }()
segmentIndex, err := strconv.ParseInt(args[1], 10, 64)
if err != nil {
return ErrRequest.Wrap(err)
}
req := &internalpb.SegmentHealthRequest{
ProjectId: []byte(args[0]),
SegmentIndex: segmentIndex,
Bucket: []byte(args[2]),
EncryptedPath: []byte(args[3]),
}
resp, err := i.healthclient.SegmentHealth(ctx, req)
if err != nil {
return ErrRequest.Wrap(err)
}
f, err := csvOutput()
if err != nil {
return err
}
defer func() {
err := f.Close()
if err != nil {
fmt.Printf("error closing file: %+v\n", err)
}
}()
w := csv.NewWriter(f)
defer w.Flush()
redundancy, err := eestream.NewRedundancyStrategyFromProto(resp.GetRedundancy())
if err != nil {
return ErrRequest.Wrap(err)
}
if err := printRedundancyTable(w, redundancy); err != nil {
return err
}
if err := printSegmentHealthAndNodeTables(w, redundancy, []*internalpb.SegmentHealth{resp.GetHealth()}); err != nil {
return err
}
return nil
}
func csvOutput() (*os.File, error) {
if CSVPath == "stdout" {
return os.Stdout, nil
}
return os.Create(CSVPath)
}
func printSegmentHealthAndNodeTables(w *csv.Writer, redundancy eestream.RedundancyStrategy, segments []*internalpb.SegmentHealth) error {
segmentTableHeader := []string{
"Segment Index", "Healthy Nodes", "Unhealthy Nodes", "Offline Nodes",
}
if err := w.Write(segmentTableHeader); err != nil {
return fmt.Errorf("error writing record to csv: %w", err)
}
currentNodeIndex := 1 // start at index 1 to leave first column empty
nodeIndices := make(map[storj.NodeID]int) // to keep track of node positions for node table
// Add each segment to the segmentTable
for _, segment := range segments {
healthyNodes := segment.HealthyIds // healthy nodes with pieces currently online
unhealthyNodes := segment.UnhealthyIds // unhealthy nodes with pieces currently online
offlineNodes := segment.OfflineIds // offline nodes
segmentIndexPath := string(segment.GetSegment()) // path formatted Segment Index
row := []string{
segmentIndexPath,
strconv.FormatInt(int64(len(healthyNodes)), 10),
strconv.FormatInt(int64(len(unhealthyNodes)), 10),
strconv.FormatInt(int64(len(offlineNodes)), 10),
}
if err := w.Write(row); err != nil {
return fmt.Errorf("error writing record to csv: %w", err)
}
allNodes := []storj.NodeID{}
allNodes = append(allNodes, healthyNodes...)
allNodes = append(allNodes, unhealthyNodes...)
allNodes = append(allNodes, offlineNodes...)
for _, id := range allNodes {
if nodeIndices[id] == 0 {
nodeIndices[id] = currentNodeIndex
currentNodeIndex++
}
}
}
if err := w.Write([]string{}); err != nil {
return fmt.Errorf("error writing record to csv: %w", err)
}
numNodes := len(nodeIndices)
nodeTableHeader := make([]string, numNodes+1)
for id, i := range nodeIndices {
nodeTableHeader[i] = id.String()
}
if err := w.Write(nodeTableHeader); err != nil {
return fmt.Errorf("error writing record to csv: %w", err)
}
// Add online/offline info to the node table
for _, segment := range segments {
row := make([]string, numNodes+1)
for _, id := range segment.HealthyIds {
i := nodeIndices[id]
row[i] = "healthy"
}
for _, id := range segment.UnhealthyIds {
i := nodeIndices[id]
row[i] = "unhealthy"
}
for _, id := range segment.OfflineIds {
i := nodeIndices[id]
row[i] = "offline"
}
row[0] = string(segment.GetSegment())
if err := w.Write(row); err != nil {
return fmt.Errorf("error writing record to csv: %w", err)
}
}
return nil
}
func printRedundancyTable(w *csv.Writer, redundancy eestream.RedundancyStrategy) error {
total := redundancy.TotalCount() // total amount of pieces we generated (n)
required := redundancy.RequiredCount() // minimum required stripes for reconstruction (k)
optimalThreshold := redundancy.OptimalThreshold() // amount of pieces we need to store to call it a success (o)
repairThreshold := redundancy.RepairThreshold() // amount of pieces we need to drop to before triggering repair (m)
redundancyTable := [][]string{
{"Total Pieces (n)", "Minimum Required (k)", "Optimal Threshold (o)", "Repair Threshold (m)"},
{strconv.Itoa(total), strconv.Itoa(required), strconv.Itoa(optimalThreshold), strconv.Itoa(repairThreshold)},
{},
}
for _, row := range redundancyTable {
if err := w.Write(row); err != nil {
return fmt.Errorf("error writing record to csv: %w", err)
}
}
return nil
}
func init() {
rootCmd.AddCommand(statsCmd)
rootCmd.AddCommand(healthCmd)
healthCmd.AddCommand(objectHealthCmd)
healthCmd.AddCommand(segmentHealthCmd)
objectHealthCmd.Flags().StringVar(&CSVPath, "csv-path", "stdout", "csv path where command output is written")
flag.Parse()
}
func main() {
process.Exec(rootCmd)
}