-
Notifications
You must be signed in to change notification settings - Fork 4
/
execution.go
380 lines (346 loc) · 11.7 KB
/
execution.go
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
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
package caretakerd
import (
"github.com/echocat/caretakerd/errors"
"github.com/echocat/caretakerd/keyStore"
"github.com/echocat/caretakerd/logger"
"github.com/echocat/caretakerd/service"
"github.com/echocat/caretakerd/values"
ssync "sync"
"time"
)
// Executable indicates an object that could be executed.
type Executable interface {
Services() *service.Services
KeyStore() *keyStore.KeyStore
Logger() *logger.Logger
}
// Execution is an instance of an execution of every service of caretakerd.
type Execution struct {
executable Executable
executions map[*service.Service]*service.Execution
restartRequests map[*service.Service]bool
stopRequests map[*service.Service]bool
masterExitCode *values.ExitCode
masterError error
lock *ssync.RWMutex
wg *ssync.WaitGroup
}
// NewExecution create a new Execution instance of caretakerd.
//
// Hint: A caretakerd Execution instance could only be called once.
func NewExecution(executable Executable) *Execution {
defaultExitCode := values.ExitCode(-1)
return &Execution{
executable: executable,
masterExitCode: &defaultExitCode,
executions: map[*service.Service]*service.Execution{},
restartRequests: map[*service.Service]bool{},
stopRequests: map[*service.Service]bool{},
lock: new(ssync.RWMutex),
wg: new(ssync.WaitGroup),
}
}
// Run starts the caretakerd Execution and start also every service and required resource.
// This method is blocking.
func (instance *Execution) Run() (values.ExitCode, error) {
autoStartableServices := instance.executable.Services().GetAllAutoStartable()
// Start all not master services at first. Because this are the dependencies for the master...
for _, target := range autoStartableServices {
if target.Config().Type != service.Master {
instance.startAndLogProblemsIfNeeded(target)
}
}
// Now start the master.
masterStarted := false
for _, target := range autoStartableServices {
if target.Config().Type == service.Master {
err := instance.Start(target)
if err != nil {
(*instance).masterError = err
}
masterStarted = true
}
}
if !masterStarted {
(*instance).masterError = errors.New("No master was started. There are no master configured?")
}
if (*instance).masterError != nil {
instance.stopOthers()
}
instance.wg.Wait()
return *(*instance).masterExitCode, (*instance).masterError
}
// GetCountOfActiveExecutions return number of active execution of services.
func (instance *Execution) GetCountOfActiveExecutions() int {
instance.doRLock()
defer instance.doRUnlock()
return len(instance.executions)
}
func (instance *Execution) startAndLogProblemsIfNeeded(target *service.Service) {
err := instance.Start(target)
if err != nil {
instance.executable.Logger().LogProblem(err, logger.Error, "Could not start execution of service '%v'.", target)
}
}
// Start start the given service.
// This method is not blocking.
func (instance *Execution) Start(target *service.Service) error {
execution, err := instance.createAndRegisterNotExistingExecutionFor(target)
if err != nil {
if sare, ok := err.(service.AlreadyRunningError); ok {
return sare
}
return errors.New("Could not start service '%v'.", target).CausedBy(err)
}
instance.wg.Add(1)
go instance.drive(execution)
return nil
}
func (instance *Execution) drive(target *service.Execution) {
var exitCode values.ExitCode
var err error
defer instance.doAfterExecution(target, exitCode, err)
respectDelay := true
doRun := true
for run := 1; doRun && target != nil && !instance.isAlreadyStopRequested(target); run++ {
if respectDelay {
if !instance.delayedStartIfNeeded(target, run) {
break
}
} else {
run = 1
}
if !instance.isAlreadyStopRequested(target) {
exitCode, err = target.Run()
doRun, respectDelay = instance.checkAfterExecutionStates(target, exitCode, err)
if doRun && !instance.isAlreadyStopRequested(target) {
newTarget, err := instance.recreateExecution(target)
if err != nil {
instance.executable.Logger().LogProblem(err, logger.Error, "Could not retrigger execution of '%v'.", target)
} else {
target = newTarget
}
}
}
}
}
func (instance *Execution) isAlreadyStopRequested(target *service.Execution) bool {
instance.doRLock()
defer instance.doRUnlock()
stopRequested, ok := instance.stopRequests[target.Service()]
return ok && stopRequested
}
func (instance *Execution) recreateExecution(target *service.Execution) (*service.Execution, error) {
instance.doWLock()
defer instance.doWUnlock()
s := target.Service()
newTarget, err := s.NewExecution(instance.executable.KeyStore())
if err != nil {
delete(instance.executions, s)
} else {
instance.executions[s] = newTarget
}
delete(instance.restartRequests, s)
return newTarget, err
}
func (instance *Execution) checkAfterExecutionStates(target *service.Execution, exitCode values.ExitCode, err error) (doRestart bool, respectDelay bool) {
if _, ok := err.(service.StoppedOrKilledError); ok {
doRestart = false
} else if _, ok := err.(service.UnrecoverableError); ok {
doRestart = target.Service().Config().CronExpression.IsEnabled() && instance.masterExitCode == nil
} else if instance.checkRestartRequestedAndClean(target.Service()) {
doRestart = true
respectDelay = false
} else if target.Service().Config().SuccessExitCodes.Contains(exitCode) {
doRestart = target.Service().Config().CronExpression.IsEnabled() && instance.masterExitCode == nil
} else {
doRestart = true
respectDelay = true
}
return doRestart, respectDelay
}
func (instance *Execution) doAfterExecution(target *service.Execution, exitCode values.ExitCode, err error) {
defer instance.doUnregisterExecution(target)
if target.Service().Config().Type == service.Master {
instance.masterExitCode = &exitCode
instance.masterError = err
instance.stopOthers()
}
}
func (instance *Execution) stopOthers() {
others := instance.allExecutionsButMaster()
if len(others) > 0 {
instance.executable.Logger().Log(logger.Debug, "Master is down. Stopping all left services...")
for _, other := range others {
go instance.Stop(other.Service())
}
}
}
func (instance *Execution) delayedStartIfNeeded(target *service.Execution, currentRun int) bool {
config := target.Service().Config()
if currentRun == 1 {
return instance.delayedStartIfNeededFor(target, config.StartDelayInSeconds, "Wait %d seconds before start...")
}
return instance.delayedStartIfNeededFor(target, config.RestartDelayInSeconds, "Wait %d seconds before restart...")
}
func (instance *Execution) delayedStartIfNeededFor(target *service.Execution, delayInSeconds values.NonNegativeInteger, messagePattern string) bool {
s := target.Service()
if s.Config().StartDelayInSeconds > 0 {
s.Logger().Log(logger.Debug, messagePattern, delayInSeconds)
return target.SyncGroup().Sleep(time.Duration(delayInSeconds)*time.Second) == nil
}
return true
}
func (instance *Execution) checkRestartRequestedAndClean(target *service.Service) bool {
instance.doWLock()
defer instance.doWUnlock()
result := instance.restartRequests[target]
delete(instance.restartRequests, target)
return result
}
func (instance *Execution) registerStopRequestsFor(executions ...*service.Execution) {
instance.doWLock()
defer instance.doWUnlock()
for _, execution := range executions {
instance.stopRequests[execution.Service()] = true
delete(instance.restartRequests, execution.Service())
}
}
// StopAll stop all running serivces.
func (instance *Execution) StopAll() {
for _, execution := range instance.allExecutions() {
if execution.Service().Config().Type == service.Master {
// Hint: Stop of the master should also trigger the shutdown of all other services.
// This is the reason why we only shutdown the master at this point.
instance.Stop(execution.Service())
}
}
instance.wg.Wait()
}
// Restart stops and starts the given service.
func (instance *Execution) Restart(target *service.Service) error {
instance.doRLock()
if stopRequested, ok := instance.stopRequests[target]; ok && stopRequested {
instance.doRUnlock()
return service.AlreadyStoppedError{Name: target.Name()}
}
execution, ok := instance.executions[target]
if !ok {
instance.doRUnlock()
return instance.Start(target)
}
instance.restartRequests[target] = true
instance.doRUnlock()
execution.Stop()
return nil
}
// Stop stops the given service.
func (instance *Execution) Stop(target *service.Service) error {
instance.doRLock()
execution, ok := instance.executions[target]
if !ok {
instance.doRUnlock()
return service.AlreadyStoppedError{Name: target.Name()}
}
instance.doRUnlock()
instance.registerStopRequestsFor(execution)
execution.Stop()
return nil
}
// Kill kills the given service.
func (instance *Execution) Kill(target *service.Service) error {
instance.doRLock()
execution, ok := instance.executions[target]
if !ok {
instance.doRUnlock()
return service.AlreadyStoppedError{Name: target.Name()}
}
instance.doRUnlock()
instance.registerStopRequestsFor(execution)
return execution.Kill()
}
// Signal sends the given signal to the given service.
func (instance *Execution) Signal(target *service.Service, what values.Signal) error {
instance.doRLock()
execution, ok := instance.executions[target]
if !ok {
instance.doRUnlock()
return service.AlreadyStoppedError{Name: target.Name()}
}
instance.doRUnlock()
return execution.Signal(what)
}
func (instance *Execution) createAndRegisterNotExistingExecutionFor(target *service.Service) (*service.Execution, error) {
instance.doWLock()
defer instance.doWUnlock()
result, err := target.NewExecution(instance.executable.KeyStore())
if err != nil {
return nil, err
}
if _, ok := instance.executions[target]; ok {
return nil, service.AlreadyRunningError{Name: target.Name()}
}
instance.executions[target] = result
return result, nil
}
func (instance *Execution) allExecutionsButMaster() []*service.Execution {
instance.doRLock()
defer instance.doRUnlock()
result := []*service.Execution{}
for s, candidate := range instance.executions {
if s.Config().Type != service.Master {
result = append(result, candidate)
}
}
return result
}
func (instance *Execution) allExecutions() []*service.Execution {
instance.doRLock()
defer instance.doRUnlock()
result := []*service.Execution{}
for _, candidate := range instance.executions {
result = append(result, candidate)
}
return result
}
func (instance *Execution) doUnregisterExecution(target *service.Execution) {
instance.doWLock()
defer instance.doWUnlock()
delete(instance.executions, target.Service())
delete(instance.restartRequests, target.Service())
instance.wg.Done()
}
func (instance *Execution) doWLock() {
instance.lock.Lock()
}
func (instance *Execution) doWUnlock() {
instance.lock.Unlock()
}
func (instance *Execution) doRLock() {
instance.lock.RLock()
}
func (instance *Execution) doRUnlock() {
instance.lock.RUnlock()
}
// GetFor get the current active service exectuion for the given service.
// Return false if there is no current matching execution.
func (instance *Execution) GetFor(s *service.Service) (*service.Execution, bool) {
result, ok := instance.executions[s]
return result, ok
}
// Information return an information object that contain an information for every
// configured service.
func (instance *Execution) Information() map[string]service.Information {
result := map[string]service.Information{}
for _, service := range *instance.executable.Services() {
result[service.Name()] = instance.InformationFor(service)
}
return result
}
// InformationFor return an information object for given service.
func (instance *Execution) InformationFor(s *service.Service) service.Information {
if result, ok := instance.GetFor(s); ok {
return service.NewInformationForExecution(result)
}
return service.NewInformationForService(s)
}