FazBrowse GitHub Viewer
|
Trending
|
URL:
|
Home
Tools:
[Download Repo ZIP]
[View Raw Code]
[Original HTTPS Page]
cloudflared/orchestration/orchestrator.go at master · codkeep/cloudflared · GitHub
codkeep
/
cloudflared
Public
forked from
cloudflare/cloudflared
Notifications
You must be signed in to change notification settings
Fork
0
Star
0
Code
Pull requests
0
Actions
Projects
Security and quality
0
Insights
Additional navigation options
Code
Pull requests
Actions
Projects
Security and quality
Insights
Expand file tree
Breadcrumbs
cloudflared
/
orchestration
/
orchestrator.go
Copy path
More file actions
More file actions
Latest commit
History
History
History
277 lines (246 loc) · 9.84 KB
Breadcrumbs
cloudflared
/
orchestration
/
orchestrator.go
Copy path
File metadata and controls
277 lines (246 loc) · 9.84 KB
Raw
Copy raw file
Download raw file
Open symbols panel
Edit and raw actions
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
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
package
orchestration
import
(
"context"
"encoding/json"
"fmt"
"strconv"
"sync"
"sync/atomic"
pkgerrors
"github.com/pkg/errors"
"github.com/rs/zerolog"
"github.com/cloudflare/cloudflared/cmd/cloudflared/flags"
"github.com/cloudflare/cloudflared/config"
"github.com/cloudflare/cloudflared/connection"
cfdflow
"github.com/cloudflare/cloudflared/flow"
"github.com/cloudflare/cloudflared/ingress"
"github.com/cloudflare/cloudflared/proxy"
"github.com/cloudflare/cloudflared/tunnelrpc/pogs"
)
// Orchestrator manages configurations, so they can be updatable during runtime
// properties are static, so it can be read without lock
// currentVersion and config are read/write infrequently, so their access are synchronized with RWMutex
// access to proxy is synchronized with atomic.Value, because it uses copy-on-write to provide scalable frequently
// read when update is infrequent
type
Orchestrator
struct
{
currentVersion
int32
// Used by UpdateConfig to make sure one update at a time
lock
sync.
RWMutex
// Underlying value is proxy.Proxy, can be read without the lock, but still needs the lock to update
proxy
atomic.
Value
// Set of internal ingress rules defined at cloudflared startup (separate from user-defined ingress rules)
internalRules
[]ingress.
Rule
// cloudflared Configuration
config
*
Config
tags
[]pogs.
Tag
// flowLimiter tracks active sessions across the tunnel and limits new sessions if they are above the limit.
flowLimiter
cfdflow.
Limiter
// Origin dialer service to manage egress socket dialing.
originDialerService
*
ingress.
OriginDialerService
log
*
zerolog.
Logger
// orchestrator must not handle any more updates after shutdownC is closed
shutdownC
<-
chan
struct
{}
// Closing proxyShutdownC will close the previous proxy
proxyShutdownC
chan
<-
struct
{}
}
func
NewOrchestrator
(
ctx
context.
Context
,
config
*
Config
,
tags
[]pogs.
Tag
,
internalRules
[]ingress.
Rule
,
log
*
zerolog.
Logger
,
) (
*
Orchestrator
,
error
) {
o
:=
&
Orchestrator
{
// Lowest possible version, any remote configuration will have version higher than this
// Starting at -1 allows a configuration migration (local to remote) to override the current configuration as it
// will start at version 0.
currentVersion
:
-
1
,
internalRules
:
internalRules
,
config
:
config
,
tags
:
tags
,
flowLimiter
:
cfdflow
.
NewLimiter
(
config
.
WarpRouting
.
MaxActiveFlows
),
originDialerService
:
config
.
OriginDialerService
,
log
:
log
,
shutdownC
:
ctx
.
Done
(),
}
if
err
:=
o
.
updateIngress
(
*
config
.
Ingress
,
config
.
WarpRouting
);
err
!=
nil
{
return
nil
,
err
}
go
o
.
waitToCloseLastProxy
()
return
o
,
nil
}
// UpdateConfig creates a new proxy with the new ingress rules
func
(
o
*
Orchestrator
)
UpdateConfig
(
version
int32
,
config
[]
byte
)
*
pogs.
UpdateConfigurationResponse
{
o
.
lock
.
Lock
()
defer
o
.
lock
.
Unlock
()
if
o
.
currentVersion
>=
version
{
o
.
log
.
Debug
().
Int32
(
"current_version"
,
o
.
currentVersion
).
Int32
(
"received_version"
,
version
).
Msg
(
"Current version is equal or newer than received version"
)
return
&
pogs.
UpdateConfigurationResponse
{
LastAppliedVersion
:
o
.
currentVersion
,
}
}
var
newConf
newRemoteConfig
if
err
:=
json
.
Unmarshal
(
config
,
&
newConf
);
err
!=
nil
{
o
.
log
.
Err
(
err
).
Int32
(
"version"
,
version
).
Str
(
"config"
,
string
(
config
)).
Msgf
(
"Failed to deserialize new configuration"
)
return
&
pogs.
UpdateConfigurationResponse
{
LastAppliedVersion
:
o
.
currentVersion
,
Err
:
err
,
}
}
if
err
:=
o
.
updateIngress
(
newConf
.
Ingress
,
newConf
.
WarpRouting
);
err
!=
nil
{
o
.
log
.
Err
(
err
).
Int32
(
"version"
,
version
).
Str
(
"config"
,
string
(
config
)).
Msgf
(
"Failed to update ingress"
)
return
&
pogs.
UpdateConfigurationResponse
{
LastAppliedVersion
:
o
.
currentVersion
,
Err
:
err
,
}
}
o
.
currentVersion
=
version
o
.
log
.
Info
().
Int32
(
"version"
,
version
).
Str
(
"config"
,
string
(
config
)).
Msg
(
"Updated to new configuration"
)
configVersion
.
Set
(
float64
(
version
))
return
&
pogs.
UpdateConfigurationResponse
{
LastAppliedVersion
:
o
.
currentVersion
,
}
}
// overrideRemoteWarpRoutingWithLocalValues overrides the ingress.WarpRoutingConfig that comes from the remote with
// the local values if there is any.
func
(
o
*
Orchestrator
)
overrideRemoteWarpRoutingWithLocalValues
(
remoteWarpRouting
*
ingress.
WarpRoutingConfig
)
error
{
return
o
.
overrideMaxActiveFlows
(
o
.
config
.
ConfigurationFlags
[
flags
.
MaxActiveFlows
],
remoteWarpRouting
)
}
// overrideMaxActiveFlows checks the local configuration flags, and if a value is found for the flags.MaxActiveFlows
// overrides the value that comes on the remote ingress.WarpRoutingConfig with the local value.
func
(
o
*
Orchestrator
)
overrideMaxActiveFlows
(
maxActiveFlowsLocalConfig
string
,
remoteWarpRouting
*
ingress.
WarpRoutingConfig
)
error
{
// If max active flows isn't defined locally just use the remote value
if
maxActiveFlowsLocalConfig
==
""
{
return
nil
}
maxActiveFlowsLocalOverride
,
err
:=
strconv
.
ParseUint
(
maxActiveFlowsLocalConfig
,
10
,
64
)
if
err
!=
nil
{
return
pkgerrors
.
Wrapf
(
err
,
"failed to parse %s"
,
flags
.
MaxActiveFlows
)
}
// Override the value that comes from the remote with the local value
remoteWarpRouting
.
MaxActiveFlows
=
maxActiveFlowsLocalOverride
return
nil
}
// The caller is responsible to make sure there is no concurrent access
func
(
o
*
Orchestrator
)
updateIngress
(
ingressRules
ingress.
Ingress
,
warpRouting
ingress.
WarpRoutingConfig
)
error
{
select
{
case
<-
o
.
shutdownC
:
return
fmt
.
Errorf
(
"cloudflared already shutdown"
)
default
:
}
// Overrides the local values, onto the remote values of the warp routing configuration
if
err
:=
o
.
overrideRemoteWarpRoutingWithLocalValues
(
&
warpRouting
);
err
!=
nil
{
return
pkgerrors
.
Wrap
(
err
,
"failed to merge local overrides into warp routing configuration"
)
}
// Assign the internal ingress rules to the parsed ingress
ingressRules
.
InternalRules
=
o
.
internalRules
// Check if ingress rules are empty, and add the default route if so.
if
ingressRules
.
IsEmpty
() {
ingressRules
.
Rules
=
ingress
.
GetDefaultIngressRules
(
o
.
log
)
}
// Start new proxy before closing the ones from last version.
// The upside is we don't need to restart proxy from last version, which can fail
// The downside is new version might have ingress rule that require previous version to be shutdown first
// The downside is minimized because none of the ingress.OriginService implementation have that requirement
proxyShutdownC
:=
make
(
chan
struct
{})
if
err
:=
ingressRules
.
StartOrigins
(
o
.
log
,
proxyShutdownC
);
err
!=
nil
{
return
pkgerrors
.
Wrap
(
err
,
"failed to start origin"
)
}
// Update the flow limit since the configuration might have changed
o
.
flowLimiter
.
SetLimit
(
warpRouting
.
MaxActiveFlows
)
// Update the origin dialer service with the new dialer settings
// We need to update the dialer here instead of creating a new instance of OriginDialerService because it has
// its own references and go routines. Specifically, the UDP dialer is a reference to this same service all the
// way into the datagram manager. Reconstructing the datagram manager is not something we currently provide during
// runtime in response to a configuration push except when starting a tunnel connection.
o
.
originDialerService
.
UpdateDefaultDialer
(
ingress
.
NewDialer
(
warpRouting
))
// Create and replace the origin proxy with a new instance
proxy
:=
proxy
.
NewOriginProxy
(
ingressRules
,
o
.
originDialerService
,
o
.
tags
,
o
.
flowLimiter
,
o
.
log
)
o
.
proxy
.
Store
(
proxy
)
o
.
config
.
Ingress
=
&
ingressRules
o
.
config
.
WarpRouting
=
warpRouting
// If proxyShutdownC is nil, there is no previous running proxy
if
o
.
proxyShutdownC
!=
nil
{
close
(
o
.
proxyShutdownC
)
}
o
.
proxyShutdownC
=
proxyShutdownC
return
nil
}
// GetConfigJSON returns the current json serialization of the config as the edge understands it
func
(
o
*
Orchestrator
)
GetConfigJSON
() ([]
byte
,
error
) {
o
.
lock
.
RLock
()
defer
o
.
lock
.
RUnlock
()
c
:=
&
newLocalConfig
{
RemoteConfig
: ingress.
RemoteConfig
{
Ingress
:
*
o
.
config
.
Ingress
,
WarpRouting
:
o
.
config
.
WarpRouting
,
},
ConfigurationFlags
:
o
.
config
.
ConfigurationFlags
,
}
return
json
.
Marshal
(
c
)
}
// GetVersionedConfigJSON returns the current version and configuration as JSON
func
(
o
*
Orchestrator
)
GetVersionedConfigJSON
() ([]
byte
,
error
) {
o
.
lock
.
RLock
()
defer
o
.
lock
.
RUnlock
()
var
currentConfiguration
=
struct
{
Version
int32
`json:"version"`
Config
struct
{
Ingress
[]ingress.
Rule
`json:"ingress"`
WarpRouting
config.
WarpRoutingConfig
`json:"warp-routing"`
OriginRequest
ingress.
OriginRequestConfig
`json:"originRequest"`
}
`json:"config"`
}{
Version
:
o
.
currentVersion
,
Config
:
struct
{
Ingress
[]ingress.
Rule
`json:"ingress"`
WarpRouting
config.
WarpRoutingConfig
`json:"warp-routing"`
OriginRequest
ingress.
OriginRequestConfig
`json:"originRequest"`
}{
Ingress
:
o
.
config
.
Ingress
.
Rules
,
WarpRouting
:
o
.
config
.
WarpRouting
.
RawConfig
(),
OriginRequest
:
o
.
config
.
Ingress
.
Defaults
,
},
}
return
json
.
Marshal
(
currentConfiguration
)
}
// GetOriginProxy returns an interface to proxy to origin. It satisfies connection.ConfigManager interface
func
(
o
*
Orchestrator
)
GetOriginProxy
() (connection.
OriginProxy
,
error
) {
val
:=
o
.
proxy
.
Load
()
if
val
==
nil
{
err
:=
fmt
.
Errorf
(
"origin proxy not configured"
)
o
.
log
.
Error
().
Msg
(
err
.
Error
())
return
nil
,
err
}
proxy
,
ok
:=
val
.(connection.
OriginProxy
)
if
!
ok
{
err
:=
fmt
.
Errorf
(
"origin proxy has unexpected value %+v"
,
val
)
o
.
log
.
Error
().
Msg
(
err
.
Error
())
return
nil
,
err
}
return
proxy
,
nil
}
// GetFlowLimiter returns the flow limiter used across cloudflared, that can be hot reload when
// the configuration changes.
func
(
o
*
Orchestrator
)
GetFlowLimiter
() cfdflow.
Limiter
{
return
o
.
flowLimiter
}
func
(
o
*
Orchestrator
)
waitToCloseLastProxy
() {
<-
o
.
shutdownC
o
.
lock
.
Lock
()
defer
o
.
lock
.
Unlock
()
if
o
.
proxyShutdownC
!=
nil
{
close
(
o
.
proxyShutdownC
)
o
.
proxyShutdownC
=
nil
}
}
Back
|
FazBrowse Home
|
New Git URL