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
|
// Command praefect provides a reverse-proxy server with high-availability
// specific features for Gitaly.
//
// Additionally, praefect has subcommands for common tasks:
//
// SQL Ping
//
// The subcommand "sql-ping" checks if the database configured in the config
// file is reachable:
//
// praefect -config PATH_TO_CONFIG sql-ping
//
// SQL Migrate
//
// The subcommand "sql-migrate" will apply any outstanding SQL migrations.
//
// praefect -config PATH_TO_CONFIG sql-migrate
//
// Dial Nodes
//
// The subcommand "dial-nodes" helps diagnose connection problems to Gitaly or
// Praefect. The subcommand works by sourcing the connection information from
// the config file, and then dialing and health checking the remote nodes.
//
// praefect -config PATH_TO_CONFIG dial-nodes
//
package main
import (
"context"
"errors"
"flag"
"fmt"
"os"
"strings"
"github.com/sirupsen/logrus"
"gitlab.com/gitlab-org/gitaly/internal/bootstrap"
"gitlab.com/gitlab-org/gitaly/internal/bootstrap/starter"
"gitlab.com/gitlab-org/gitaly/internal/config/sentry"
"gitlab.com/gitlab-org/gitaly/internal/log"
"gitlab.com/gitlab-org/gitaly/internal/praefect"
"gitlab.com/gitlab-org/gitaly/internal/praefect/config"
"gitlab.com/gitlab-org/gitaly/internal/praefect/conn"
"gitlab.com/gitlab-org/gitaly/internal/praefect/datastore"
"gitlab.com/gitlab-org/gitaly/internal/praefect/metrics"
"gitlab.com/gitlab-org/gitaly/internal/praefect/protoregistry"
"gitlab.com/gitlab-org/gitaly/internal/version"
"gitlab.com/gitlab-org/labkit/monitoring"
"gitlab.com/gitlab-org/labkit/tracing"
)
var (
flagConfig = flag.String("config", "", "Location for the config.toml")
flagVersion = flag.Bool("version", false, "Print version and exit")
logger = log.Default()
errNoConfigFile = errors.New("the config flag must be passed")
)
const progname = "praefect"
func main() {
flag.Parse()
// If invoked with -version
if *flagVersion {
fmt.Println(praefect.GetVersionString())
os.Exit(0)
}
conf, err := configure()
if err != nil {
logger.Fatal(err)
}
if args := flag.Args(); len(args) > 0 {
os.Exit(subCommand(conf, args[0], args[1:]))
}
logger.WithField("version", praefect.GetVersionString()).Info("Starting Praefect")
starterConfigs, err := getStarterConfigs(conf.SocketPath, conf.ListenAddr)
if err != nil {
logger.Fatalf("%s", err)
}
if err := run(starterConfigs, conf); err != nil {
logger.Fatalf("%v", err)
}
}
func configure() (config.Config, error) {
var conf config.Config
if *flagConfig == "" {
return conf, errNoConfigFile
}
conf, err := config.FromFile(*flagConfig)
if err != nil {
return conf, fmt.Errorf("error reading config file: %v", err)
}
if err := conf.Validate(); err != nil {
return conf, err
}
logger = conf.ConfigureLogger()
tracing.Initialize(tracing.WithServiceName("praefect"))
if conf.PrometheusListenAddr != "" {
logger.WithField("address", conf.PrometheusListenAddr).Info("Starting prometheus listener")
conf.Prometheus.Configure()
go func() {
if err := monitoring.Serve(
monitoring.WithListenerAddress(conf.PrometheusListenAddr),
monitoring.WithBuildInformation(praefect.GetVersion(), praefect.GetBuildTime())); err != nil {
logger.WithError(err).Errorf("Unable to start healthcheck listener: %v", conf.PrometheusListenAddr)
}
}()
}
sentry.ConfigureSentry(version.GetVersion(), conf.Sentry)
return conf, nil
}
func run(cfgs []starter.Config, conf config.Config) error {
clientConnections := conn.NewClientConnections()
for _, virtualStorage := range conf.VirtualStorages {
for _, node := range virtualStorage.Nodes {
if _, err := clientConnections.GetConnection(node.Storage); err == nil {
continue
}
if err := clientConnections.RegisterNode(node.Storage, node.Address, node.Token); err != nil {
return fmt.Errorf("failed to register %s: %s", node.Address, err)
}
logger.WithField("node_address", node.Address).Info("registered gitaly node")
}
}
latencyMetric, err := metrics.RegisterReplicationLatency(conf.Prometheus)
if err != nil {
return err
}
queueMetric, err := metrics.RegisterReplicationJobsInFlight()
if err != nil {
return err
}
var (
// top level server dependencies
ds = datastore.NewInMemory(conf)
coordinator = praefect.NewCoordinator(logger, ds, clientConnections, conf, protoregistry.GitalyProtoFileDescriptors...)
repl = praefect.NewReplMgr(
"default",
logger,
ds,
clientConnections,
praefect.WithLatencyMetric(latencyMetric),
praefect.WithQueueMetric(queueMetric))
srv = praefect.NewServer(coordinator, repl, nil, logger, clientConnections, conf)
serverErrors = make(chan error, 1)
)
testSQLConnection(logger, conf)
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
b, err := bootstrap.New()
if err != nil {
return fmt.Errorf("unable to create a bootstrap: %v", err)
}
srv.RegisterServices()
b.StopAction = srv.GracefulStop
for _, cfg := range cfgs {
b.RegisterStarter(starter.New(cfg, srv))
}
if err := b.Start(); err != nil {
return fmt.Errorf("unable to start the bootstrap: %v", err)
}
go func() { serverErrors <- b.Wait() }()
go func() { serverErrors <- repl.ProcessBacklog(ctx) }()
go coordinator.FailoverRotation()
return <-serverErrors
}
func getStarterConfigs(socketPath, listenAddr string) ([]starter.Config, error) {
var cfgs []starter.Config
if socketPath != "" {
if err := os.RemoveAll(socketPath); err != nil {
return nil, err
}
cleanPath := strings.TrimPrefix(socketPath, "unix:")
cfgs = append(cfgs, starter.Config{Name: starter.Unix, Addr: cleanPath})
logger.WithField("address", socketPath).Info("listening on unix socket")
}
if listenAddr != "" {
cleanAddr := strings.TrimPrefix(listenAddr, "tcp://")
cfgs = append(cfgs, starter.Config{Name: starter.TCP, Addr: cleanAddr})
logger.WithField("address", listenAddr).Info("listening at tcp address")
}
return cfgs, nil
}
// Test Postgres connection, for diagnostic purposes only while we roll
// out Postgres support. https://gitlab.com/gitlab-org/gitaly/issues/1755
func testSQLConnection(logger *logrus.Entry, conf config.Config) {
if err := datastore.CheckPostgresVersion(conf); err != nil {
logger.WithError(err).Error("SQL connection check failed")
} else {
logger.Info("SQL connection check successful")
}
}
|