GitVita
New repositoryImport repositoryNew issue
Your repositories Your dashboard Sign in

cloudflare/cloudflared

Public

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

Watch0Fork0Star0
CodeCommitsIssuesPull requestsActionsInsightsSecurity
2025.2.1

Branches

Tags

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

Clone

HTTPS

Download ZIP

cloudflared/connection

connection/control.go

151lines · modecode

RawDownload
Latest commit unavailable.
1package connection
2
3import (
4 "context"
5 "io"
6 "net"
7 "time"
8
9 "github.com/pkg/errors"
10
11 "github.com/cloudflare/cloudflared/management"
12 "github.com/cloudflare/cloudflared/tunnelrpc"
13 tunnelpogs "github.com/cloudflare/cloudflared/tunnelrpc/pogs"
14)
15
16// registerClient derives a named tunnel rpc client that can then be used to register and unregister connections.
17type registerClientFunc func(context.Context, io.ReadWriteCloser, time.Duration) tunnelrpc.RegistrationClient
18
19type controlStream struct {
20 observer *Observer
21
22 connectedFuse ConnectedFuse
23 tunnelProperties *TunnelProperties
24 connIndex uint8
25 edgeAddress net.IP
26 protocol Protocol
27
28 registerClientFunc registerClientFunc
29 registerTimeout time.Duration
30
31 gracefulShutdownC <-chan struct{}
32 gracePeriod time.Duration
33 stoppedGracefully bool
34}
35
36// ControlStreamHandler registers connections with origintunneld and initiates graceful shutdown.
37type ControlStreamHandler interface {
38 // ServeControlStream handles the control plane of the transport in the current goroutine calling this
39 ServeControlStream(ctx context.Context, rw io.ReadWriteCloser, connOptions *tunnelpogs.ConnectionOptions, tunnelConfigGetter TunnelConfigJSONGetter) error
40 // IsStopped tells whether the method above has finished
41 IsStopped() bool
42}
43
44type TunnelConfigJSONGetter interface {
45 GetConfigJSON() ([]byte, error)
46}
47
48// NewControlStream returns a new instance of ControlStreamHandler
49func NewControlStream(
50 observer *Observer,
51 connectedFuse ConnectedFuse,
52 tunnelProperties *TunnelProperties,
53 connIndex uint8,
54 edgeAddress net.IP,
55 registerClientFunc registerClientFunc,
56 registerTimeout time.Duration,
57 gracefulShutdownC <-chan struct{},
58 gracePeriod time.Duration,
59 protocol Protocol,
60) ControlStreamHandler {
61 if registerClientFunc == nil {
62 registerClientFunc = tunnelrpc.NewRegistrationClient
63 }
64 return &controlStream{
65 observer: observer,
66 connectedFuse: connectedFuse,
67 tunnelProperties: tunnelProperties,
68 registerClientFunc: registerClientFunc,
69 registerTimeout: registerTimeout,
70 connIndex: connIndex,
71 edgeAddress: edgeAddress,
72 gracefulShutdownC: gracefulShutdownC,
73 gracePeriod: gracePeriod,
74 protocol: protocol,
75 }
76}
77
78func (c *controlStream) ServeControlStream(
79 ctx context.Context,
80 rw io.ReadWriteCloser,
81 connOptions *tunnelpogs.ConnectionOptions,
82 tunnelConfigGetter TunnelConfigJSONGetter,
83) error {
84 registrationClient := c.registerClientFunc(ctx, rw, c.registerTimeout)
85
86 registrationDetails, err := registrationClient.RegisterConnection(
87 ctx,
88 c.tunnelProperties.Credentials.Auth(),
89 c.tunnelProperties.Credentials.TunnelID,
90 connOptions,
91 c.connIndex,
92 c.edgeAddress)
93 if err != nil {
94 defer registrationClient.Close()
95 if err.Error() == DuplicateConnectionError {
96 c.observer.metrics.regFail.WithLabelValues("dup_edge_conn", "registerConnection").Inc()
97 return errDuplicationConnection
98 }
99 c.observer.metrics.regFail.WithLabelValues("server_error", "registerConnection").Inc()
100 return serverRegistrationErrorFromRPC(err)
101 }
102 c.observer.metrics.regSuccess.WithLabelValues("registerConnection").Inc()
103
104 c.observer.logConnected(registrationDetails.UUID, c.connIndex, registrationDetails.Location, c.edgeAddress, c.protocol)
105 c.observer.sendConnectedEvent(c.connIndex, c.protocol, registrationDetails.Location, c.edgeAddress)
106 c.connectedFuse.Connected()
107
108 // if conn index is 0 and tunnel is not remotely managed, then send local ingress rules configuration
109 if c.connIndex == 0 && !registrationDetails.TunnelIsRemotelyManaged {
110 if tunnelConfig, err := tunnelConfigGetter.GetConfigJSON(); err == nil {
111 if err := registrationClient.SendLocalConfiguration(ctx, tunnelConfig); err != nil {
112 c.observer.metrics.localConfigMetrics.pushesErrors.Inc()
113 c.observer.log.Err(err).Msg("unable to send local configuration")
114 }
115 c.observer.metrics.localConfigMetrics.pushes.Inc()
116 } else {
117 c.observer.log.Err(err).Msg("failed to obtain current configuration")
118 }
119 }
120
121 return c.waitForUnregister(ctx, registrationClient)
122}
123
124func (c *controlStream) waitForUnregister(ctx context.Context, registrationClient tunnelrpc.RegistrationClient) error {
125 // wait for connection termination or start of graceful shutdown
126 defer registrationClient.Close()
127 var shutdownError error
128 select {
129 case <-ctx.Done():
130 shutdownError = ctx.Err()
131 break
132 case <-c.gracefulShutdownC:
133 c.stoppedGracefully = true
134 }
135
136 c.observer.sendUnregisteringEvent(c.connIndex)
137 err := registrationClient.GracefulShutdown(ctx, c.gracePeriod)
138 if err != nil {
139 return errors.Wrap(err, "Error shutting down control stream")
140 }
141 c.observer.log.Info().
142 Int(management.EventTypeKey, int(management.Cloudflared)).
143 Uint8(LogFieldConnIndex, c.connIndex).
144 IPAddr(LogFieldIPAddress, c.edgeAddress).
145 Msg("Unregistered tunnel connection")
146 return shutdownError
147}
148
149func (c *controlStream) IsStopped() bool {
150 return c.stoppedGracefully
151}