-
Notifications
You must be signed in to change notification settings - Fork 19.2k
Expand file tree
/
Copy pathreload.go
More file actions
334 lines (294 loc) · 12.1 KB
/
Copy pathreload.go
File metadata and controls
334 lines (294 loc) · 12.1 KB
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
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
package daemon
import (
"context"
"encoding/json"
"errors"
"fmt"
"net/netip"
"reflect"
"strconv"
"github.com/containerd/log"
"github.com/mitchellh/copystructure"
"github.com/moby/moby/api/types/events"
"github.com/moby/moby/v2/daemon/config"
"github.com/moby/moby/v2/daemon/pkg/opts"
)
func init() {
// Register a custom copier for netip.Addr. The copystructure library uses
// reflection which cannot access unexported fields in netip.Addr, resulting
// in zero-value copies. Since netip.Addr is an immutable value type, we can
// safely return it as-is.
//
// Note: copystructure is archived (https://github.com/mitchellh/copystructure)
// and won't receive upstream fixes for this limitation.
copystructure.Copiers[reflect.TypeFor[netip.Addr]()] = func(v any) (any, error) {
return v.(netip.Addr), nil
}
}
// reloadTxn is used to defer side effects of a config reload.
type reloadTxn struct {
onCommit, onRollback []func() error
}
// OnCommit defers a function to be called when a config reload is being finalized.
// The error returned from cb is purely informational.
func (tx *reloadTxn) OnCommit(cb func() error) {
tx.onCommit = append(tx.onCommit, cb)
}
// OnRollback defers a function to be called when a config reload is aborted.
// The error returned from cb is purely informational.
func (tx *reloadTxn) OnRollback(cb func() error) {
tx.onRollback = append(tx.onRollback, cb)
}
func (tx *reloadTxn) run(cbs []func() error) error {
tx.onCommit = nil
tx.onRollback = nil
var errs []error
for _, cb := range cbs {
errs = append(errs, cb())
}
return errors.Join(errs...)
}
// Commit calls all functions registered with OnCommit.
// Any errors returned by the functions are collated into a
// *github.com/hashicorp/go-multierror.Error value.
func (tx *reloadTxn) Commit() error {
return tx.run(tx.onCommit)
}
// Rollback calls all functions registered with OnRollback.
// Any errors returned by the functions are collated into a
// *github.com/hashicorp/go-multierror.Error value.
func (tx *reloadTxn) Rollback() error {
return tx.run(tx.onRollback)
}
// Reload modifies the live daemon configuration from conf.
// conf is assumed to be a validated configuration.
//
// These are the settings that Reload changes:
// - Platform runtime
// - Daemon debug log level
// - Daemon max concurrent downloads
// - Daemon max concurrent uploads
// - Daemon max download attempts
// - Daemon shutdown timeout (in seconds)
// - Daemon default container stop timeout (in seconds)
// - Cluster discovery (reconfigure and restart)
// - Daemon labels
// - Insecure registries
// - Registry mirrors
// - Daemon live restore
// - NRI enable and filesystem locations
func (daemon *Daemon) Reload(conf *config.Config) error {
daemon.configReload.Lock()
defer daemon.configReload.Unlock()
copied, err := copystructure.Copy(daemon.config().Config)
if err != nil {
return err
}
newCfg := &configStore{
Config: copied.(config.Config),
}
attributes := map[string]string{}
// Ideally reloading should be transactional: the reload either completes
// successfully, or the daemon config and state are left untouched. We use a
// two-phase commit protocol to achieve this. Any fallible reload operation is
// split into two phases. The first phase performs all the fallible operations
// and mutates the newCfg copy. The second phase atomically swaps newCfg into
// the live daemon configuration and executes any commit functions the first
// phase registered to apply the side effects. If any first-phase returns an
// error, the reload transaction is rolled back by discarding newCfg and
// executing any registered rollback functions.
var txn reloadTxn
for _, reload := range []func(txn *reloadTxn, newCfg *configStore, conf *config.Config, attributes map[string]string) error{
// TODO(thaJeztah): most of these are defined as method, but don't use the daemon receiver; consider making them regular functions.
daemon.reloadPlatform,
daemon.reloadDebug,
daemon.reloadMaxConcurrentDownloadsAndUploads,
daemon.reloadMaxDownloadAttempts,
daemon.reloadShutdownTimeout,
daemon.reloadDefaultStopTimeout,
daemon.reloadFeatures,
daemon.reloadLabels,
daemon.reloadRegistryConfig,
daemon.reloadLiveRestore,
daemon.reloadNetworkDiagnosticPort,
daemon.reloadNRI,
} {
if err := reload(&txn, newCfg, conf, attributes); err != nil {
return errors.Join(err, txn.Rollback())
}
}
daemon.configStore.Store(newCfg)
err = txn.Commit()
daemon.LogDaemonEventWithAttributes(events.ActionReload, attributes)
return err
}
func marshalAttributeSlice(v []string) string {
if v == nil {
return "[]"
}
b, err := json.Marshal(v)
if err != nil {
panic(err) // Should never happen as the input type is fixed.
}
return string(b)
}
// reloadDebug updates configuration with Debug option
// and updates the passed attributes
func (daemon *Daemon) reloadDebug(_ *reloadTxn, newCfg *configStore, conf *config.Config, attributes map[string]string) error {
// update corresponding configuration
if conf.IsValueSet("debug") {
newCfg.Debug = conf.Debug
}
// prepare reload event attributes with updatable configurations
attributes["debug"] = strconv.FormatBool(newCfg.Debug)
return nil
}
// reloadMaxConcurrentDownloadsAndUploads updates configuration with max concurrent
// download and upload options and updates the passed attributes
func (daemon *Daemon) reloadMaxConcurrentDownloadsAndUploads(txn *reloadTxn, newCfg *configStore, conf *config.Config, attributes map[string]string) error {
// We always "reset" as the cost is lightweight and easy to maintain.
newCfg.MaxConcurrentDownloads = config.DefaultMaxConcurrentDownloads
newCfg.MaxConcurrentUploads = config.DefaultMaxConcurrentUploads
if conf.IsValueSet("max-concurrent-downloads") && conf.MaxConcurrentDownloads != 0 {
newCfg.MaxConcurrentDownloads = conf.MaxConcurrentDownloads
}
if conf.IsValueSet("max-concurrent-uploads") && conf.MaxConcurrentUploads != 0 {
newCfg.MaxConcurrentUploads = conf.MaxConcurrentUploads
}
txn.OnCommit(func() error {
if daemon.imageService != nil {
daemon.imageService.UpdateConfig(
newCfg.MaxConcurrentDownloads,
newCfg.MaxConcurrentUploads,
)
}
return nil
})
// prepare reload event attributes with updatable configurations
attributes["max-concurrent-downloads"] = strconv.Itoa(newCfg.MaxConcurrentDownloads)
attributes["max-concurrent-uploads"] = strconv.Itoa(newCfg.MaxConcurrentUploads)
log.G(context.TODO()).Debug("Reset Max Concurrent Downloads: ", attributes["max-concurrent-downloads"])
log.G(context.TODO()).Debug("Reset Max Concurrent Uploads: ", attributes["max-concurrent-uploads"])
return nil
}
// reloadMaxDownloadAttempts updates configuration with max concurrent
// download attempts when a connection is lost and updates the passed attributes
func (daemon *Daemon) reloadMaxDownloadAttempts(_ *reloadTxn, newCfg *configStore, conf *config.Config, attributes map[string]string) error {
// We always "reset" as the cost is lightweight and easy to maintain.
newCfg.MaxDownloadAttempts = config.DefaultDownloadAttempts
if conf.IsValueSet("max-download-attempts") && conf.MaxDownloadAttempts != 0 {
newCfg.MaxDownloadAttempts = conf.MaxDownloadAttempts
}
// prepare reload event attributes with updatable configurations
attributes["max-download-attempts"] = strconv.Itoa(newCfg.MaxDownloadAttempts)
log.G(context.TODO()).Debug("Reset Max Download Attempts: ", attributes["max-download-attempts"])
return nil
}
// reloadShutdownTimeout updates configuration with daemon shutdown timeout option
// and updates the passed attributes
func (daemon *Daemon) reloadShutdownTimeout(_ *reloadTxn, newCfg *configStore, conf *config.Config, attributes map[string]string) error {
// update corresponding configuration
if conf.IsValueSet("shutdown-timeout") {
newCfg.ShutdownTimeout = conf.ShutdownTimeout
log.G(context.TODO()).Debugf("Reset Shutdown Timeout: %d", newCfg.ShutdownTimeout)
}
// prepare reload event attributes with updatable configurations
attributes["shutdown-timeout"] = strconv.Itoa(newCfg.ShutdownTimeout)
return nil
}
// reloadDefaultStopTimeout updates the default container stop timeout
// and updates the passed attributes.
func (daemon *Daemon) reloadDefaultStopTimeout(_ *reloadTxn, newCfg *configStore, conf *config.Config, attributes map[string]string) error {
if conf.IsValueSet("default-stop-timeout") {
newCfg.DefaultStopTimeout = conf.DefaultStopTimeout
log.G(context.TODO()).Debugf("Reset Default Stop Timeout: %d", newCfg.DefaultStopTimeout)
}
attributes["default-stop-timeout"] = strconv.Itoa(newCfg.DefaultStopTimeout)
return nil
}
// reloadLabels updates configuration with engine labels
// and updates the passed attributes
func (daemon *Daemon) reloadLabels(_ *reloadTxn, newCfg *configStore, conf *config.Config, attributes map[string]string) error {
// update corresponding configuration
if conf.IsValueSet("labels") {
newCfg.Labels = conf.Labels
}
// prepare reload event attributes with updatable configurations
attributes["labels"] = marshalAttributeSlice(newCfg.Labels)
return nil
}
// reloadRegistryConfig updates the configuration with registry options
// and updates the passed attributes.
func (daemon *Daemon) reloadRegistryConfig(txn *reloadTxn, newCfg *configStore, conf *config.Config, attributes map[string]string) error {
if conf.IsValueSet("insecure-registries") {
newCfg.ServiceOptions.InsecureRegistries = conf.InsecureRegistries
}
if conf.IsValueSet("registry-mirrors") {
newCfg.ServiceOptions.Mirrors = conf.Mirrors
}
commit, err := daemon.registryService.ReplaceConfig(newCfg.ServiceOptions)
if err != nil {
return err
}
txn.OnCommit(func() error { commit(); return nil })
attributes["insecure-registries"] = marshalAttributeSlice(newCfg.ServiceOptions.InsecureRegistries)
attributes["registry-mirrors"] = marshalAttributeSlice(newCfg.ServiceOptions.Mirrors)
return nil
}
// reloadLiveRestore updates configuration with live restore option
// and updates the passed attributes
func (daemon *Daemon) reloadLiveRestore(_ *reloadTxn, newCfg *configStore, conf *config.Config, attributes map[string]string) error {
// update corresponding configuration
if conf.IsValueSet("live-restore") {
newCfg.LiveRestoreEnabled = conf.LiveRestoreEnabled
}
// prepare reload event attributes with updatable configurations
attributes["live-restore"] = strconv.FormatBool(newCfg.LiveRestoreEnabled)
return nil
}
// reloadNetworkDiagnosticPort updates the network controller starting the diagnostic if the config is valid
func (daemon *Daemon) reloadNetworkDiagnosticPort(txn *reloadTxn, newCfg *configStore, conf *config.Config, attributes map[string]string) error {
txn.OnCommit(func() error {
if conf == nil || daemon.netController == nil || !conf.IsValueSet("network-diagnostic-port") || conf.NetworkDiagnosticPort == 0 {
// If there is no config make sure that the diagnostic is off
if daemon.netController != nil {
daemon.netController.StopDiagnostic()
}
return nil
}
// Enable the network diagnostic if the flag is set with a valid port within the range
daemon.netController.StartDiagnostic(conf.NetworkDiagnosticPort)
return nil
})
return nil
}
// reloadFeatures updates configuration with enabled/disabled features
func (daemon *Daemon) reloadFeatures(_ *reloadTxn, newCfg *configStore, conf *config.Config, attributes map[string]string) error {
// update corresponding configuration
// note that we allow features option to be entirely unset
newCfg.Features = conf.Features
// prepare reload event attributes with updatable configurations
attributes["features"] = fmt.Sprint(newCfg.Features)
return nil
}
// reloadNRI updates NRI configuration
func (daemon *Daemon) reloadNRI(txn *reloadTxn, newCfg *configStore, conf *config.Config, attributes map[string]string) error {
if daemon.nri == nil {
// Daemon not initialised.
return nil
}
if conf.IsValueSet("nri-opts") {
newCfg.Config.NRIOpts = conf.NRIOpts
} else {
newCfg.Config.NRIOpts = opts.NRIOpts{}
}
commit, err := daemon.nri.PrepareReload(newCfg.NRIOpts)
if err != nil {
return err
}
if commit != nil {
txn.OnCommit(commit)
}
attributes["nri-opts"] = fmt.Sprint(newCfg.NRIOpts)
return nil
}