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
|
package praefect
import (
"context"
"errors"
"fmt"
"math/rand"
"os"
"os/signal"
"sync"
"syscall"
"time"
gitalyauth "gitlab.com/gitlab-org/gitaly/auth"
gitalyconfig "gitlab.com/gitlab-org/gitaly/internal/config"
"gitlab.com/gitlab-org/gitaly/internal/helper"
"gitlab.com/gitlab-org/gitaly/internal/praefect/models"
"gitlab.com/gitlab-org/gitaly/internal/praefect/protoregistry"
"github.com/golang/protobuf/protoc-gen-go/descriptor"
"github.com/sirupsen/logrus"
"gitlab.com/gitlab-org/gitaly/client"
"gitlab.com/gitlab-org/gitaly/internal/praefect/grpc-proxy/proxy"
"google.golang.org/grpc"
)
// Coordinator takes care of directing client requests to the appropriate
// downstream server. The coordinator is thread safe; concurrent calls to
// register nodes are safe.
type Coordinator struct {
log *logrus.Logger
failoverMutex sync.RWMutex
connMutex sync.RWMutex
datastore ReplicasDatastore
nodes map[string]*grpc.ClientConn
registry *protoregistry.Registry
}
// NewCoordinator returns a new Coordinator that utilizes the provided logger
func NewCoordinator(l *logrus.Logger, datastore ReplicasDatastore, fileDescriptors ...*descriptor.FileDescriptorProto) *Coordinator {
registry := protoregistry.New()
registry.RegisterFiles(fileDescriptors...)
return &Coordinator{
log: l,
datastore: datastore,
nodes: make(map[string]*grpc.ClientConn),
registry: registry,
}
}
// RegisterProtos allows coordinator to register new protos on the fly
func (c *Coordinator) RegisterProtos(protos ...*descriptor.FileDescriptorProto) error {
return c.registry.RegisterFiles(protos...)
}
// streamDirector determines which downstream servers receive requests
func (c *Coordinator) streamDirector(ctx context.Context, fullMethodName string, peeker proxy.StreamModifier) (context.Context, *grpc.ClientConn, error) {
// For phase 1, we need to route messages based on the storage location
// to the appropriate Gitaly node.
c.log.Debugf("Stream director received method %s", fullMethodName)
c.failoverMutex.RLock()
defer c.failoverMutex.RUnlock()
frame, err := peeker.Peek()
if err != nil {
return nil, nil, err
}
mi, err := c.registry.LookupMethod(fullMethodName)
if err != nil {
return nil, nil, err
}
var primary *models.StorageNode
if mi.Scope == protoregistry.ScopeRepository {
m, err := mi.UnmarshalRequestProto(frame)
if err != nil {
return nil, nil, err
}
targetRepo, err := mi.TargetRepo(m)
if err != nil {
return nil, nil, err
}
primary, err = c.datastore.GetPrimary(targetRepo.GetRelativePath())
if err != nil {
if err != ErrPrimaryNotSet {
return nil, nil, err
}
// if there are no primaries for this repository, pick one
nodes, err := c.datastore.GetStorageNodes()
if err != nil {
return nil, nil, err
}
if len(nodes) == 0 {
return nil, nil, fmt.Errorf("no nodes serve storage %s", targetRepo.GetStorageName())
}
newPrimary := nodes[rand.New(rand.NewSource(time.Now().Unix())).Intn(len(nodes))]
//newPrimary := nodes[0]
// set the primary
if err = c.datastore.SetPrimary(targetRepo.GetRelativePath(), newPrimary.ID); err != nil {
return nil, nil, err
}
primary = &newPrimary
}
targetRepo.StorageName = primary.Storage
b, err := proxy.Codec().Marshal(m)
if err != nil {
return nil, nil, err
}
if err = peeker.Modify(b); err != nil {
return nil, nil, err
}
} else {
//TODO: For now we just pick a random storage node for a non repository scoped RPC, but we will need to figure out exactly how to
// proxy requests that are not repository scoped
node, err := c.datastore.GetStorageNodes()
if err != nil {
return nil, nil, err
}
if len(node) == 0 {
return nil, nil, errors.New("no node storages found")
}
primary = &node[0]
}
// We only need the primary node, as there's only one primary storage
// location per praefect at this time
cc, err := c.GetConnection(primary.Storage)
if err != nil {
return nil, nil, fmt.Errorf("unable to find existing client connection for %s", primary.Storage)
}
return helper.IncomingToOutgoing(ctx), cc, nil
}
// RegisterNode will direct traffic to the supplied downstream connection when the storage location
// is encountered.
func (c *Coordinator) RegisterNode(storage, address string) error {
conn, err := client.Dial(address,
[]grpc.DialOption{
grpc.WithDefaultCallOptions(grpc.CallCustomCodec(proxy.Codec())),
grpc.WithPerRPCCredentials(gitalyauth.RPCCredentials(gitalyconfig.Config.Auth.Token)),
},
)
if err != nil {
return err
}
c.setConn(storage, conn)
return nil
}
func (c *Coordinator) setConn(storage string, conn *grpc.ClientConn) {
c.connMutex.Lock()
c.nodes[storage] = conn
c.connMutex.Unlock()
}
// GetConnection gets the grpc client connection based on an address
func (c *Coordinator) GetConnection(storage string) (*grpc.ClientConn, error) {
c.connMutex.RLock()
cc, ok := c.nodes[storage]
c.connMutex.RUnlock()
if !ok {
return nil, errors.New("client connection not found")
}
return cc, nil
}
// FailoverRotation waits for the SIGUSR1 signal, then promotes the next secondary to be primary
func (c *Coordinator) FailoverRotation() {
c.handleSignalAndRotate()
}
func (c *Coordinator) handleSignalAndRotate() {
failoverChan := make(chan os.Signal, 1)
signal.Notify(failoverChan, syscall.SIGUSR1)
for {
<-failoverChan
c.failoverMutex.Lock()
c.log.Info("failover happens")
c.failoverMutex.Unlock()
}
}
|
