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
|
package praefect
import (
"crypto/tls"
"fmt"
"net"
"sync"
"github.com/sirupsen/logrus"
"gitlab.com/gitlab-org/gitaly/internal/praefect/config"
"gitlab.com/gitlab-org/gitaly/internal/praefect/datastore"
"gitlab.com/gitlab-org/gitaly/internal/praefect/grpc-proxy/proxy"
"gitlab.com/gitlab-org/gitaly/internal/praefect/nodes"
"gitlab.com/gitlab-org/gitaly/internal/praefect/protoregistry"
"gitlab.com/gitlab-org/gitaly/internal/praefect/transactions"
"google.golang.org/grpc"
"google.golang.org/grpc/credentials"
)
// NewServerFactory returns factory object for initialization of praefect gRPC servers.
func NewServerFactory(
conf config.Config,
logger *logrus.Entry,
director proxy.StreamDirector,
nodeMgr nodes.Manager,
txMgr *transactions.Manager,
queue datastore.ReplicationEventQueue,
rs datastore.RepositoryStore,
assignmentStore AssignmentStore,
registry *protoregistry.Registry,
conns Connections,
primaryGetter PrimaryGetter,
) *ServerFactory {
return &ServerFactory{
conf: conf,
logger: logger,
director: director,
nodeMgr: nodeMgr,
txMgr: txMgr,
queue: queue,
rs: rs,
assignmentStore: assignmentStore,
registry: registry,
conns: conns,
primaryGetter: primaryGetter,
}
}
// ServerFactory is a factory of praefect grpc servers
type ServerFactory struct {
mtx sync.Mutex
conf config.Config
logger *logrus.Entry
director proxy.StreamDirector
nodeMgr nodes.Manager
txMgr *transactions.Manager
queue datastore.ReplicationEventQueue
rs datastore.RepositoryStore
assignmentStore AssignmentStore
registry *protoregistry.Registry
secure, insecure []*grpc.Server
conns Connections
primaryGetter PrimaryGetter
}
// Serve starts serving on the provided listener with newly created grpc.Server
func (s *ServerFactory) Serve(l net.Listener, secure bool) error {
srv, err := s.Create(secure)
if err != nil {
return err
}
return srv.Serve(l)
}
// Stop stops all servers created by the factory.
func (s *ServerFactory) Stop() {
for _, srv := range s.all() {
srv.Stop()
}
}
// GracefulStop stops both the secure and insecure servers gracefully.
func (s *ServerFactory) GracefulStop() {
wg := sync.WaitGroup{}
for _, srv := range s.all() {
wg.Add(1)
go func(s *grpc.Server) {
s.GracefulStop()
wg.Done()
}(srv)
}
wg.Wait()
}
// Create returns newly instantiated and initialized with interceptors instance of the gRPC server.
func (s *ServerFactory) Create(secure bool) (*grpc.Server, error) {
s.mtx.Lock()
defer s.mtx.Unlock()
if !secure {
s.insecure = append(s.insecure, s.createGRPC(nil))
return s.insecure[len(s.insecure)-1], nil
}
cert, err := tls.LoadX509KeyPair(s.conf.TLS.CertPath, s.conf.TLS.KeyPath)
if err != nil {
return nil, fmt.Errorf("load certificate key pair: %w", err)
}
s.secure = append(s.secure, s.createGRPC(credentials.NewTLS(&tls.Config{
Certificates: []tls.Certificate{cert},
MinVersion: tls.VersionTLS12,
})))
return s.secure[len(s.secure)-1], nil
}
func (s *ServerFactory) createGRPC(creds credentials.TransportCredentials) *grpc.Server {
return NewGRPCServer(
s.conf,
s.logger,
s.registry,
s.director,
s.txMgr,
s.rs,
s.assignmentStore,
s.conns,
s.primaryGetter,
creds,
)
}
func (s *ServerFactory) all() []*grpc.Server {
s.mtx.Lock()
defer s.mtx.Unlock()
servers := make([]*grpc.Server, 0, len(s.secure)+len(s.insecure))
servers = append(servers, s.secure...)
servers = append(servers, s.insecure...)
return servers
}
|
