/
walk.go
126 lines (106 loc) · 3.08 KB
/
walk.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
package job
import (
"fmt"
"os"
"github.com/urfave/cli/v2"
"github.com/filecoin-project/lily/chain/indexer/tasktype"
"github.com/filecoin-project/lily/commands"
"github.com/filecoin-project/lily/lens/lily"
lotuscli "github.com/filecoin-project/lotus/cli"
)
type walkOps struct {
interval int `zap:"interval"`
}
var walkFlags walkOps
var WalkIntervalFlag = &cli.IntFlag{
Name: "interval",
Usage: "The interval for specific task",
Value: 120,
Destination: &walkFlags.interval,
}
//revive:disable
var WalkCmd = &cli.Command{
Name: "walk",
Usage: "walk and index a range of the filecoin blockchain.",
Description: `
The walk command will index state based on the list of tasks (--tasks) provided over the specified range (--from --to).
Each epoch will be indexed serially, starting from the heaviest tipset at the upper height (--to) to the lower height (--from).
As and example, the below command:
$ lily job run --tasks=block_header,messages walk --from=10 --to=20
walks epochs 20 through 10 (inclusive) executing the block_header and messages task for each epoch.
The status of each epoch and its set of tasks can be observed in the visor_processing_reports table.
`,
Flags: []cli.Flag{
RangeFromFlag,
RangeToFlag,
WalkIntervalFlag,
},
Subcommands: []*cli.Command{
WalkNotifyCmd,
},
Before: func(_ *cli.Context) error {
tasks := RunFlags.Tasks.Value()
for _, taskName := range tasks {
if _, found := tasktype.TaskLookup[taskName]; found {
continue
} else if _, found := tasktype.TableLookup[taskName]; found {
continue
} else {
return fmt.Errorf("unknown task: %s", taskName)
}
}
return rangeFlags.validate()
},
Action: func(cctx *cli.Context) error {
ctx := lotuscli.ReqContext(cctx)
api, closer, err := commands.GetAPI(ctx)
if err != nil {
return err
}
defer closer()
cfg := &lily.LilyWalkConfig{
JobConfig: RunFlags.ParseJobConfig("walk"),
From: rangeFlags.from,
To: rangeFlags.to,
Interval: walkFlags.interval,
}
res, err := api.LilyWalk(ctx, cfg)
if err != nil {
return err
}
err = commands.PrintNewJob(os.Stdout, res)
return err
},
}
var WalkNotifyCmd = &cli.Command{
Name: "notify",
Usage: "notify the provided queueing system of epochs to index allowing tipset-workers to perform the indexing.",
Description: `
The notify command will insert tasks into the provided queueing system for consumption by tipset-workers.
This command should be used when lily is configured to perform distributed indexing.
`,
Flags: []cli.Flag{
NotifyQueueFlag,
},
Action: func(cctx *cli.Context) error {
ctx := lotuscli.ReqContext(cctx)
api, closer, err := commands.GetAPI(ctx)
if err != nil {
return err
}
defer closer()
cfg := &lily.LilyWalkNotifyConfig{
WalkConfig: lily.LilyWalkConfig{
JobConfig: RunFlags.ParseJobConfig("walk-notify"),
From: rangeFlags.from,
To: rangeFlags.to,
},
Queue: notifyFlags.queue,
}
res, err := api.LilyWalkNotify(ctx, cfg)
if err != nil {
return err
}
return commands.PrintNewJob(os.Stdout, res)
},
}