this repo has no description
update rainbow to cli/v3
+31
-31
+31
-31
cmd/rainbow/main.go
+31
-31
cmd/rainbow/main.go
···
18
18
"github.com/bluesky-social/indigo/util/svcutil"
19
19
20
20
"github.com/earthboundkid/versioninfo/v2"
21
-
"github.com/urfave/cli/v2"
21
+
"github.com/urfave/cli/v3"
22
22
"go.opentelemetry.io/otel"
23
23
"go.opentelemetry.io/otel/attribute"
24
24
"go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracehttp"
···
36
36
37
37
func run(args []string) error {
38
38
39
-
app := cli.App{
39
+
app := cli.Command{
40
40
Name: "rainbow",
41
41
Usage: "atproto firehose fan-out daemon",
42
42
Version: versioninfo.Short(),
···
47
47
&cli.StringFlag{
48
48
Name: "log-level",
49
49
Usage: "log verbosity level (eg: warn, info, debug)",
50
-
EnvVars: []string{"RAINBOW_LOG_LEVEL", "GO_LOG_LEVEL", "LOG_LEVEL"},
50
+
Sources: cli.EnvVars("RAINBOW_LOG_LEVEL", "GO_LOG_LEVEL", "LOG_LEVEL"),
51
51
},
52
52
&cli.StringFlag{
53
53
Name: "upstream-host",
54
54
Value: "http://localhost:2470",
55
55
Usage: "URL (schema and hostname, no path) of the upstream host (eg, relay)",
56
-
EnvVars: []string{"ATP_RELAY_HOST", "RAINBOW_RELAY_HOST"},
56
+
Sources: cli.EnvVars("ATP_RELAY_HOST", "RAINBOW_RELAY_HOST"),
57
57
},
58
58
&cli.StringFlag{
59
59
Name: "persist-db",
60
60
Value: "./rainbow.db",
61
61
Usage: "path to persistence db",
62
-
EnvVars: []string{"RAINBOW_DB_PATH"},
62
+
Sources: cli.EnvVars("RAINBOW_DB_PATH"),
63
63
},
64
64
&cli.StringFlag{
65
65
Name: "cursor-file",
66
66
Value: "./rainbow-cursor",
67
67
Usage: "write upstream cursor number to this file",
68
-
EnvVars: []string{"RAINBOW_CURSOR_PATH"},
68
+
Sources: cli.EnvVars("RAINBOW_CURSOR_PATH"),
69
69
},
70
70
&cli.StringFlag{
71
71
Name: "api-listen",
72
72
Value: ":2480",
73
-
EnvVars: []string{"RAINBOW_API_LISTEN"},
73
+
Sources: cli.EnvVars("RAINBOW_API_LISTEN"),
74
74
},
75
75
&cli.StringFlag{
76
76
Name: "metrics-listen",
77
77
Value: ":2481",
78
-
EnvVars: []string{"RAINBOW_METRICS_LISTEN", "SPLITTER_METRICS_LISTEN"},
78
+
Sources: cli.EnvVars("RAINBOW_METRICS_LISTEN", "SPLITTER_METRICS_LISTEN"),
79
79
},
80
80
&cli.Float64Flag{
81
81
Name: "persist-hours",
82
82
Value: 24 * 3,
83
-
EnvVars: []string{"RAINBOW_PERSIST_HOURS", "SPLITTER_PERSIST_HOURS"},
83
+
Sources: cli.EnvVars("RAINBOW_PERSIST_HOURS", "SPLITTER_PERSIST_HOURS"),
84
84
Usage: "hours to buffer (float, may be fractional)",
85
85
},
86
86
&cli.Int64Flag{
87
87
Name: "persist-bytes",
88
88
Value: 0,
89
89
Usage: "max bytes target for event cache, 0 to disable size target trimming",
90
-
EnvVars: []string{"RAINBOW_PERSIST_BYTES", "SPLITTER_PERSIST_BYTES"},
90
+
Sources: cli.EnvVars("RAINBOW_PERSIST_BYTES", "SPLITTER_PERSIST_BYTES"),
91
91
},
92
92
&cli.StringSliceFlag{
93
93
// TODO: better name for this argument
94
94
Name: "next-crawler",
95
95
Usage: "forward POST requestCrawl to these hosts (schema and host, no path) in addition to upstream-host. Comma-separated or multiple flags",
96
-
EnvVars: []string{"RAINBOW_NEXT_CRAWLER", "RELAY_NEXT_CRAWLER"},
96
+
Sources: cli.EnvVars("RAINBOW_NEXT_CRAWLER", "RELAY_NEXT_CRAWLER"),
97
97
},
98
98
&cli.StringFlag{
99
99
Name: "collectiondir-host",
100
100
Value: "http://localhost:2510",
101
101
Usage: "host (schema and hostname, no path) of upstream collectiondir instance, for com.atproto.sync.listReposByCollection",
102
-
EnvVars: []string{"RAINBOW_COLLECTIONDIR_HOST"},
102
+
Sources: cli.EnvVars("RAINBOW_COLLECTIONDIR_HOST"),
103
103
},
104
104
&cli.StringFlag{
105
105
Name: "env",
106
106
Usage: "operating environment (eg, 'prod', 'test')",
107
107
Value: "dev",
108
-
EnvVars: []string{"ENVIRONMENT"},
108
+
Sources: cli.EnvVars("ENVIRONMENT"),
109
109
},
110
110
&cli.BoolFlag{
111
111
Name: "enable-otel-otlp",
112
112
Usage: "enables OTEL OTLP exporter endpoint",
113
-
EnvVars: []string{"RAINBOW_ENABLE_OTEL_OTLP", "ENABLE_OTEL_OTLP"},
113
+
Sources: cli.EnvVars("RAINBOW_ENABLE_OTEL_OTLP", "ENABLE_OTEL_OTLP"),
114
114
},
115
115
&cli.StringFlag{
116
116
Name: "otel-otlp-endpoint",
117
117
Usage: "OTEL traces export endpoint",
118
118
Value: "http://localhost:4318",
119
-
EnvVars: []string{"OTEL_EXPORTER_OTLP_ENDPOINT"},
119
+
Sources: cli.EnvVars("OTEL_EXPORTER_OTLP_ENDPOINT"),
120
120
},
121
121
}
122
122
123
-
return app.Run(args)
123
+
return app.Run(context.Background(), args)
124
124
}
125
125
126
-
func runSplitter(cctx *cli.Context) error {
126
+
func runSplitter(ctx context.Context, cmd *cli.Command) error {
127
127
// Trap SIGINT to trigger a shutdown.
128
128
signals := make(chan os.Signal, 1)
129
129
signal.Notify(signals, syscall.SIGINT, syscall.SIGTERM)
130
130
131
-
logger := svcutil.ConfigLogger(cctx, os.Stdout).With("system", "rainbow")
131
+
logger := svcutil.ConfigLogger(cmd, os.Stdout).With("system", "rainbow")
132
132
133
133
// Enable OTLP HTTP exporter
134
134
// For relevant environment variables:
135
135
// https://pkg.go.dev/go.opentelemetry.io/otel/exporters/otlp/otlptrace#readme-environment-variables
136
-
if cctx.Bool("enable-otel-otlp") {
137
-
ep := cctx.String("otel-otlp-endpoint")
136
+
if cmd.Bool("enable-otel-otlp") {
137
+
ep := cmd.String("otel-otlp-endpoint")
138
138
logger.Info("setting up trace exporter", "endpoint", ep)
139
139
ctx, cancel := context.WithCancel(context.Background())
140
140
defer cancel()
···
152
152
}
153
153
}()
154
154
155
-
env := cctx.String("env")
155
+
env := cmd.String("env")
156
156
tp := tracesdk.NewTracerProvider(
157
157
tracesdk.WithBatcher(exp),
158
158
tracesdk.WithResource(resource.NewWithAttributes(
···
166
166
otel.SetTracerProvider(tp)
167
167
}
168
168
169
-
persistPath := cctx.String("persist-db")
170
-
upstreamHost := cctx.String("upstream-host")
171
-
collectionDirHost := cctx.String("collectiondir-host")
172
-
nextCrawlers := cctx.StringSlice("next-crawler")
169
+
persistPath := cmd.String("persist-db")
170
+
upstreamHost := cmd.String("upstream-host")
171
+
collectionDirHost := cmd.String("collectiondir-host")
172
+
nextCrawlers := cmd.StringSlice("next-crawler")
173
173
174
174
var spl *splitter.Splitter
175
175
var err error
···
177
177
logger.Info("building splitter with storage at", "path", persistPath)
178
178
ppopts := pebblepersist.PebblePersistOptions{
179
179
DbPath: persistPath,
180
-
PersistDuration: time.Duration(float64(time.Hour) * cctx.Float64("persist-hours")),
180
+
PersistDuration: time.Duration(float64(time.Hour) * cmd.Float64("persist-hours")),
181
181
GCPeriod: 5 * time.Minute,
182
-
MaxBytes: uint64(cctx.Int64("persist-bytes")),
182
+
MaxBytes: uint64(cmd.Int64("persist-bytes")),
183
183
}
184
184
conf := splitter.SplitterConfig{
185
185
UpstreamHost: upstreamHost,
186
186
CollectionDirHost: collectionDirHost,
187
-
CursorFile: cctx.String("cursor-file"),
187
+
CursorFile: cmd.String("cursor-file"),
188
188
PebbleOptions: &ppopts,
189
189
UserAgent: fmt.Sprintf("rainbow/%s (atproto-relay)", versioninfo.Short()),
190
190
}
···
194
194
conf := splitter.SplitterConfig{
195
195
UpstreamHost: upstreamHost,
196
196
CollectionDirHost: collectionDirHost,
197
-
CursorFile: cctx.String("cursor-file"),
197
+
CursorFile: cmd.String("cursor-file"),
198
198
}
199
199
spl, err = splitter.NewSplitter(conf, nextCrawlers)
200
200
}
···
205
205
}
206
206
207
207
// set up metrics endpoint
208
-
metricsListen := cctx.String("metrics-listen")
208
+
metricsListen := cmd.String("metrics-listen")
209
209
go func() {
210
210
if err := spl.StartMetrics(metricsListen); err != nil {
211
211
logger.Error("failed to start metrics endpoint", "err", err)
···
216
216
runErr := make(chan error, 1)
217
217
218
218
go func() {
219
-
err := spl.StartAPI(cctx.String("api-listen"))
219
+
err := spl.StartAPI(cmd.String("api-listen"))
220
220
runErr <- err
221
221
}()
222
222