GitVita
New repositoryImport repositoryNew issue
Your repositories Your dashboard Sign in

cloudflare/cloudflared

Public

mirrored from https://github.com/cloudflare/cloudflaredAvailable

Watch0Fork0Star0
CodeCommitsIssuesPull requestsActionsInsightsSecurity
d50fee4fa0ea358952a83798af3400c5757dff21

Branches

Tags

  • No tags available.
0Branches0Tags
Go to file
Add file
Code

Clone

HTTPS

Download ZIP

cloudflared/origin

origin/supervisor.go

440lines · modecode

RawDownload
Latest commit unavailable.
1package origin
2
3import (
4 "context"
5 "errors"
6 "fmt"
7 "net"
8 "sync"
9 "time"
10
11 "github.com/google/uuid"
12 "github.com/sirupsen/logrus"
13
14 "github.com/cloudflare/cloudflared/buffer"
15 "github.com/cloudflare/cloudflared/connection"
16 "github.com/cloudflare/cloudflared/edgediscovery"
17 "github.com/cloudflare/cloudflared/h2mux"
18 "github.com/cloudflare/cloudflared/signal"
19 tunnelpogs "github.com/cloudflare/cloudflared/tunnelrpc/pogs"
20)
21
22const (
23 // Waiting time before retrying a failed tunnel connection
24 tunnelRetryDuration = time.Second * 10
25 // SRV record resolution TTL
26 resolveTTL = time.Hour
27 // Interval between registering new tunnels
28 registrationInterval = time.Second
29
30 subsystemRefreshAuth = "refresh_auth"
31 // Maximum exponent for 'Authenticate' exponential backoff
32 refreshAuthMaxBackoff = 10
33 // Waiting time before retrying a failed 'Authenticate' connection
34 refreshAuthRetryDuration = time.Second * 10
35 // Maximum time to make an Authenticate RPC
36 authTokenTimeout = time.Second * 30
37)
38
39var (
40 errJWTUnset = errors.New("JWT unset")
41 errEventDigestUnset = errors.New("event digest unset")
42 errConnDigestUnset = errors.New("conn digest unset")
43)
44
45// Supervisor manages non-declarative tunnels. Establishes TCP connections with the edge, and
46// reconnects them if they disconnect.
47type Supervisor struct {
48 cloudflaredUUID uuid.UUID
49 config *TunnelConfig
50 edgeIPs *edgediscovery.Edge
51 lastResolve time.Time
52 resolverC chan resolveResult
53 tunnelErrors chan tunnelError
54 tunnelsConnecting map[int]chan struct{}
55 // nextConnectedIndex and nextConnectedSignal are used to wait for all
56 // currently-connecting tunnels to finish connecting so we can reset backoff timer
57 nextConnectedIndex int
58 nextConnectedSignal chan struct{}
59
60 logger *logrus.Entry
61
62 jwtLock *sync.RWMutex
63 jwt []byte
64
65 eventDigestLock *sync.RWMutex
66 eventDigest []byte
67
68 connDigestLock *sync.RWMutex
69 connDigest []byte
70
71 bufferPool *buffer.Pool
72}
73
74type resolveResult struct {
75 err error
76}
77
78type tunnelError struct {
79 index int
80 addr *net.TCPAddr
81 err error
82}
83
84func NewSupervisor(config *TunnelConfig, u uuid.UUID) (*Supervisor, error) {
85 var (
86 edgeIPs *edgediscovery.Edge
87 err error
88 )
89 if len(config.EdgeAddrs) > 0 {
90 edgeIPs, err = edgediscovery.StaticEdge(config.Logger, config.EdgeAddrs)
91 } else {
92 edgeIPs, err = edgediscovery.ResolveEdge(config.Logger)
93 }
94 if err != nil {
95 return nil, err
96 }
97 return &Supervisor{
98 cloudflaredUUID: u,
99 config: config,
100 edgeIPs: edgeIPs,
101 tunnelErrors: make(chan tunnelError),
102 tunnelsConnecting: map[int]chan struct{}{},
103 logger: config.Logger.WithField("subsystem", "supervisor"),
104 jwtLock: &sync.RWMutex{},
105 eventDigestLock: &sync.RWMutex{},
106 bufferPool: buffer.NewPool(512 * 1024),
107 }, nil
108}
109
110func (s *Supervisor) Run(ctx context.Context, connectedSignal *signal.Signal) error {
111 logger := s.config.Logger
112 if err := s.initialize(ctx, connectedSignal); err != nil {
113 return err
114 }
115 var tunnelsWaiting []int
116 tunnelsActive := s.config.HAConnections
117
118 backoff := BackoffHandler{MaxRetries: s.config.Retries, BaseTime: tunnelRetryDuration, RetryForever: true}
119 var backoffTimer <-chan time.Time
120
121 refreshAuthBackoff := &BackoffHandler{MaxRetries: refreshAuthMaxBackoff, BaseTime: refreshAuthRetryDuration, RetryForever: true}
122 var refreshAuthBackoffTimer <-chan time.Time
123
124 if s.config.UseReconnectToken {
125 if timer, err := s.refreshAuth(ctx, refreshAuthBackoff, s.authenticate); err == nil {
126 refreshAuthBackoffTimer = timer
127 } else {
128 logger.WithError(err).Errorf("initial refreshAuth failed, retrying in %v", refreshAuthRetryDuration)
129 refreshAuthBackoffTimer = time.After(refreshAuthRetryDuration)
130 }
131 }
132
133 for {
134 select {
135 // Context cancelled
136 case <-ctx.Done():
137 for tunnelsActive > 0 {
138 <-s.tunnelErrors
139 tunnelsActive--
140 }
141 return nil
142 // startTunnel returned with error
143 // (note that this may also be caused by context cancellation)
144 case tunnelError := <-s.tunnelErrors:
145 tunnelsActive--
146 if tunnelError.err != nil {
147 logger.WithError(tunnelError.err).Warn("Tunnel disconnected due to error")
148 tunnelsWaiting = append(tunnelsWaiting, tunnelError.index)
149 s.waitForNextTunnel(tunnelError.index)
150
151 if backoffTimer == nil {
152 backoffTimer = backoff.BackoffTimer()
153 }
154
155 // Previously we'd mark the edge address as bad here, but now we'll just silently use
156 // another.
157 }
158 // Backoff was set and its timer expired
159 case <-backoffTimer:
160 backoffTimer = nil
161 for _, index := range tunnelsWaiting {
162 go s.startTunnel(ctx, index, s.newConnectedTunnelSignal(index))
163 }
164 tunnelsActive += len(tunnelsWaiting)
165 tunnelsWaiting = nil
166 // Time to call Authenticate
167 case <-refreshAuthBackoffTimer:
168 newTimer, err := s.refreshAuth(ctx, refreshAuthBackoff, s.authenticate)
169 if err != nil {
170 logger.WithError(err).Error("Authentication failed")
171 // Permanent failure. Leave the `select` without setting the
172 // channel to be non-null, so we'll never hit this case of the `select` again.
173 continue
174 }
175 refreshAuthBackoffTimer = newTimer
176 // Tunnel successfully connected
177 case <-s.nextConnectedSignal:
178 if !s.waitForNextTunnel(s.nextConnectedIndex) && len(tunnelsWaiting) == 0 {
179 // No more tunnels outstanding, clear backoff timer
180 backoff.SetGracePeriod()
181 }
182 // DNS resolution returned
183 case result := <-s.resolverC:
184 s.lastResolve = time.Now()
185 s.resolverC = nil
186 if result.err == nil {
187 logger.Debug("Service discovery refresh complete")
188 } else {
189 logger.WithError(result.err).Error("Service discovery error")
190 }
191 }
192 }
193}
194
195// Returns nil if initialization succeeded, else the initialization error.
196func (s *Supervisor) initialize(ctx context.Context, connectedSignal *signal.Signal) error {
197 logger := s.logger
198
199 s.lastResolve = time.Now()
200 availableAddrs := int(s.edgeIPs.AvailableAddrs())
201 if s.config.HAConnections > availableAddrs {
202 logger.Warnf("You requested %d HA connections but I can give you at most %d.", s.config.HAConnections, availableAddrs)
203 s.config.HAConnections = availableAddrs
204 }
205
206 go s.startFirstTunnel(ctx, connectedSignal)
207 select {
208 case <-ctx.Done():
209 <-s.tunnelErrors
210 return ctx.Err()
211 case tunnelError := <-s.tunnelErrors:
212 return tunnelError.err
213 case <-connectedSignal.Wait():
214 }
215 // At least one successful connection, so start the rest
216 for i := 1; i < s.config.HAConnections; i++ {
217 ch := signal.New(make(chan struct{}))
218 go s.startTunnel(ctx, i, ch)
219 time.Sleep(registrationInterval)
220 }
221 return nil
222}
223
224// startTunnel starts the first tunnel connection. The resulting error will be sent on
225// s.tunnelErrors. It will send a signal via connectedSignal if registration succeed
226func (s *Supervisor) startFirstTunnel(ctx context.Context, connectedSignal *signal.Signal) {
227 var (
228 addr *net.TCPAddr
229 err error
230 )
231 const thisConnID = 0
232 defer func() {
233 s.tunnelErrors <- tunnelError{index: thisConnID, addr: addr, err: err}
234 }()
235
236 addr, err = s.edgeIPs.GetAddr(thisConnID)
237 if err != nil {
238 return
239 }
240
241 err = ServeTunnelLoop(ctx, s, s.config, addr, thisConnID, connectedSignal, s.cloudflaredUUID, s.bufferPool)
242 // If the first tunnel disconnects, keep restarting it.
243 edgeErrors := 0
244 for s.unusedIPs() {
245 if ctx.Err() != nil {
246 return
247 }
248 switch err.(type) {
249 case nil:
250 return
251 // try the next address if it was a dialError(network problem) or
252 // dupConnRegisterTunnelError
253 case connection.DialError, dupConnRegisterTunnelError:
254 edgeErrors++
255 default:
256 return
257 }
258 if edgeErrors >= 2 {
259 addr, err = s.edgeIPs.GetDifferentAddr(thisConnID)
260 if err != nil {
261 return
262 }
263 }
264 err = ServeTunnelLoop(ctx, s, s.config, addr, thisConnID, connectedSignal, s.cloudflaredUUID, s.bufferPool)
265 }
266}
267
268// startTunnel starts a new tunnel connection. The resulting error will be sent on
269// s.tunnelErrors.
270func (s *Supervisor) startTunnel(ctx context.Context, index int, connectedSignal *signal.Signal) {
271 var (
272 addr *net.TCPAddr
273 err error
274 )
275 defer func() {
276 s.tunnelErrors <- tunnelError{index: index, addr: addr, err: err}
277 }()
278
279 addr, err = s.edgeIPs.GetAddr(index)
280 if err != nil {
281 return
282 }
283 err = ServeTunnelLoop(ctx, s, s.config, addr, uint8(index), connectedSignal, s.cloudflaredUUID, s.bufferPool)
284}
285
286func (s *Supervisor) newConnectedTunnelSignal(index int) *signal.Signal {
287 sig := make(chan struct{})
288 s.tunnelsConnecting[index] = sig
289 s.nextConnectedSignal = sig
290 s.nextConnectedIndex = index
291 return signal.New(sig)
292}
293
294func (s *Supervisor) waitForNextTunnel(index int) bool {
295 delete(s.tunnelsConnecting, index)
296 s.nextConnectedSignal = nil
297 for k, v := range s.tunnelsConnecting {
298 s.nextConnectedIndex = k
299 s.nextConnectedSignal = v
300 return true
301 }
302 return false
303}
304
305func (s *Supervisor) unusedIPs() bool {
306 return s.edgeIPs.AvailableAddrs() > s.config.HAConnections
307}
308
309func (s *Supervisor) ReconnectToken() ([]byte, error) {
310 s.jwtLock.RLock()
311 defer s.jwtLock.RUnlock()
312 if s.jwt == nil {
313 return nil, errJWTUnset
314 }
315 return s.jwt, nil
316}
317
318func (s *Supervisor) SetReconnectToken(jwt []byte) {
319 s.jwtLock.Lock()
320 defer s.jwtLock.Unlock()
321 s.jwt = jwt
322}
323
324func (s *Supervisor) EventDigest() ([]byte, error) {
325 s.eventDigestLock.RLock()
326 defer s.eventDigestLock.RUnlock()
327 if s.eventDigest == nil {
328 return nil, errEventDigestUnset
329 }
330 return s.eventDigest, nil
331}
332
333func (s *Supervisor) SetEventDigest(eventDigest []byte) {
334 s.eventDigestLock.Lock()
335 defer s.eventDigestLock.Unlock()
336 s.eventDigest = eventDigest
337}
338
339func (s *Supervisor) ConnDigest() ([]byte, error) {
340 s.connDigestLock.RLock()
341 defer s.connDigestLock.RUnlock()
342 if s.connDigest == nil {
343 return nil, errConnDigestUnset
344 }
345 return s.connDigest, nil
346}
347
348func (s *Supervisor) SetConnDigest(connDigest []byte) {
349 s.connDigestLock.Lock()
350 defer s.connDigestLock.Unlock()
351 s.connDigest = connDigest
352}
353
354func (s *Supervisor) refreshAuth(
355 ctx context.Context,
356 backoff *BackoffHandler,
357 authenticate func(ctx context.Context, numPreviousAttempts int) (tunnelpogs.AuthOutcome, error),
358) (retryTimer <-chan time.Time, err error) {
359 logger := s.config.Logger.WithField("subsystem", subsystemRefreshAuth)
360 authOutcome, err := authenticate(ctx, backoff.Retries())
361 if err != nil {
362 s.config.Metrics.authFail.WithLabelValues(err.Error()).Inc()
363 if duration, ok := backoff.GetBackoffDuration(ctx); ok {
364 logger.WithError(err).Warnf("Retrying in %v", duration)
365 return backoff.BackoffTimer(), nil
366 }
367 return nil, err
368 }
369 // clear backoff timer
370 backoff.SetGracePeriod()
371
372 switch outcome := authOutcome.(type) {
373 case tunnelpogs.AuthSuccess:
374 s.SetReconnectToken(outcome.JWT())
375 s.config.Metrics.authSuccess.Inc()
376 return timeAfter(outcome.RefreshAfter()), nil
377 case tunnelpogs.AuthUnknown:
378 duration := outcome.RefreshAfter()
379 s.config.Metrics.authFail.WithLabelValues(outcome.Error()).Inc()
380 logger.WithError(outcome).Warnf("Retrying in %v", duration)
381 return timeAfter(duration), nil
382 case tunnelpogs.AuthFail:
383 s.config.Metrics.authFail.WithLabelValues(outcome.Error()).Inc()
384 return nil, outcome
385 default:
386 err := fmt.Errorf("Unexpected outcome type %T", authOutcome)
387 s.config.Metrics.authFail.WithLabelValues(err.Error()).Inc()
388 return nil, err
389 }
390}
391
392func (s *Supervisor) authenticate(ctx context.Context, numPreviousAttempts int) (tunnelpogs.AuthOutcome, error) {
393 arbitraryEdgeIP, err := s.edgeIPs.GetAddrForRPC()
394 if err != nil {
395 return nil, err
396 }
397
398 edgeConn, err := connection.DialEdge(ctx, dialTimeout, s.config.TlsConfig, arbitraryEdgeIP)
399 if err != nil {
400 return nil, err
401 }
402 defer edgeConn.Close()
403
404 handler := h2mux.MuxedStreamFunc(func(*h2mux.MuxedStream) error {
405 // This callback is invoked by h2mux when the edge initiates a stream.
406 return nil // noop
407 })
408 muxerConfig := s.config.muxerConfig(handler)
409 muxerConfig.Logger = muxerConfig.Logger.WithField("subsystem", subsystemRefreshAuth)
410 muxer, err := h2mux.Handshake(edgeConn, edgeConn, muxerConfig, s.config.Metrics.activeStreams)
411 if err != nil {
412 return nil, err
413 }
414 go muxer.Serve(ctx)
415 defer func() {
416 // If we don't wait for the muxer shutdown here, edgeConn.Close() runs before the muxer connections are done,
417 // and the user sees log noise: "error writing data", "connection closed unexpectedly"
418 <-muxer.Shutdown()
419 }()
420
421 tunnelServer, err := connection.NewRPCClient(ctx, muxer, s.logger.WithField("subsystem", subsystemRefreshAuth), openStreamTimeout)
422 if err != nil {
423 return nil, err
424 }
425 defer tunnelServer.Close()
426
427 const arbitraryConnectionID = uint8(0)
428 registrationOptions := s.config.RegistrationOptions(arbitraryConnectionID, edgeConn.LocalAddr().String(), s.cloudflaredUUID)
429 registrationOptions.NumPreviousAttempts = uint8(numPreviousAttempts)
430 authResponse, err := tunnelServer.Authenticate(
431 ctx,
432 s.config.OriginCert,
433 s.config.Hostname,
434 registrationOptions,
435 )
436 if err != nil {
437 return nil, err
438 }
439 return authResponse.Outcome(), nil
440}
441