Repository navigation
Expand file tree
/
Copy pathcomponent.go
More file actions
155 lines (137 loc) · 4.76 KB
/
Copy pathcomponent.go
File metadata and controls
155 lines (137 loc) · 4.76 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
// SPDX-License-Identifier: Apache-2.0
package supervisor
import (
"context"
"fmt"
"math"
"math/rand/v2"
"time"
)
// Component is a long-running unit of work managed by a [Supervisor]:
// an HTTP server, a queue worker pool, a pub/sub listener, the scheduler, or
// an ad-hoc background task.
//
// Run must block until the work is finished or ctx is canceled. When ctx is
// canceled, Run should stop taking new work, finish or hand back in-flight
// work, and return promptly. Returning nil means the component completed
// normally; returning an error (or panicking) is a failure handled according
// to the component's [Restart] policy.
type Component interface {
// Name identifies the component in logs and status; unique per app.
Name() string
// Run does the work until ctx is canceled or the work is done.
Run(ctx context.Context) error
}
// Readier is optionally implemented by components that need time to become
// ready (for example, an HTTP server that must bind its port). Until Ready
// returns true, [Supervisor.Ready] reports false.
type Readier interface {
// Ready reports whether the component can take work.
Ready() bool
}
// Func adapts a function to a [Component].
func Func(name string, run func(ctx context.Context) error) Component {
return funcComponent{name: name, run: run}
}
type funcComponent struct {
name string
run func(ctx context.Context) error
}
func (f funcComponent) Name() string { return f.name }
func (f funcComponent) Run(ctx context.Context) error { return f.run(ctx) }
// Restart decides what happens when a component fails (returns an error or
// panics) while the supervisor is running.
type Restart int
const (
// RestartNever logs the failure and leaves the component stopped. The
// rest of the application keeps running.
RestartNever Restart = iota
// RestartOnFailure restarts the component after an exponential backoff.
// If Backoff.MaxRestarts is exceeded, the failure escalates as if the
// policy were StopOnFailure.
RestartOnFailure
// StopOnFailure shuts the whole application down; [Supervisor.Run]
// returns the component's error. Use it for components the app cannot
// work without, such as the HTTP server.
StopOnFailure
)
// String returns the policy's name, as in logs.
func (r Restart) String() string {
switch r {
case RestartNever:
return "never"
case RestartOnFailure:
return "on-failure"
case StopOnFailure:
return "stop-on-failure"
}
return fmt.Sprintf("Restart(%d)", int(r))
}
// Stage orders graceful shutdown. On shutdown the supervisor cancels stages
// from lowest to highest, waiting for every component in a stage to return
// before canceling the next. Producers of work stop before its consumers, so
// consumers can finish what was already produced.
type Stage int
// Built-in stages, in shutdown order. Custom components may use any value;
// equal values stop together.
const (
StageIngress Stage = 0 // HTTP servers: stop accepting requests first
StageScheduler Stage = 10 // scheduler: stop starting new runs
StageListeners Stage = 20 // pub/sub listeners: stop pulling messages
StageWorkers Stage = 30 // queue workers: finish in-flight jobs
StageBackground Stage = 40 // ad-hoc background tasks (app.Go)
)
// Backoff configures restarts for [RestartOnFailure].
type Backoff struct {
// Initial is the delay before the first restart. Default 1s.
Initial time.Duration
// Max caps the delay. Default 30s. A component that ran for longer than
// Max before failing is considered healthy again, and the delay resets.
Max time.Duration
// MaxRestarts limits consecutive restarts; 0 means unlimited.
MaxRestarts int
}
func (b Backoff) withDefaults() Backoff {
if b.Initial <= 0 {
b.Initial = time.Second
}
if b.Max <= 0 {
b.Max = 30 * time.Second
}
if b.Max < b.Initial {
b.Max = b.Initial
}
return b
}
// delay returns the wait before restart number n (1-based), with ±20% jitter.
func (b Backoff) delay(n int) time.Duration {
d := b.Initial
for i := 1; i < n && d < b.Max; i++ {
if d > b.Max/2 { // also prevents overflow for huge Max values
d = b.Max
break
}
d *= 2
}
d = min(d, b.Max)
f := float64(d) * (0.8 + 0.4*rand.Float64())
if f >= math.MaxInt64 {
return time.Duration(math.MaxInt64)
}
return time.Duration(f)
}
// Spec describes how a component is supervised.
type Spec struct {
// Component is what runs.
Component Component
// Roles lists the roles this component belongs to (e.g. "http",
// "workers"). A component with no roles runs in every process.
Roles []string
// Stage orders shutdown: lower stages stop first.
Stage Stage
// Restart says what happens when Run fails (returns an error or
// panics); a nil return is always done.
Restart Restart
// Backoff spaces restarts; the zero value uses the defaults.
Backoff Backoff
}