8000 feat: add debouncing to provisionerd rpc calls · coder/coder@5fc0c10 · GitHub
[go: up one dir, main page]
Skip to content

Commit 5fc0c10

Browse files
coadlercoadler
committed
feat: add debouncing to provisionerd rpc calls
1 parent 38bdae7 commit 5fc0c10

File tree

12 files changed

+122
-45
lines changed

12 files changed

+122
-45
lines changed

‎cli/deployment/config.go

Lines changed: 12 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -387,6 +387,18 @@ func newConfig() *codersdk.DeploymentConfig {
387387
Flag: "provisioner-daemons",
388388
Default: 3,
389389
},
390+
DaemonPollInterval: &codersdk.DeploymentConfigField[time.Duration]{
391+
Name: "Poll Interval",
392+
Usage: "Time to wait before polling for a new job.",
393+
Flag: "provisioner-daemon-poll-interval",
394+
Default: time.Second,
395+
},
396+
DaemonPollJitter: &codersdk.DeploymentConfigField[time.Duration]{
397+
Name: "Poll Jitter",
398+
Usage: "Random jitter added to the poll interval.",
399+
Flag: "provisioner-daemon-poll-jitter",
400+
Default: 100 * time.Millisecond,
401+
},
390402
ForceCancelInterval: &codersdk.DeploymentConfigField[time.Duration]{
391403
Name: "Force Cancel Interval",
392404
Usage: "Time to force cancel provisioning tasks that are stuck.",

‎cli/deployment/config_test.go

Lines changed: 18 additions & 13 deletions
Original file line numberDiff line numberDiff line change
@@ -2,6 +2,7 @@ package deployment_test
22

33
import (
44
"testing"
5+
"time"
56

67
"github.com/spf13/pflag"
78
"github.com/stretchr/testify/require"
@@ -25,19 +26,21 @@ func TestConfig(t *testing.T) {
2526
}{{
2627
Name: "Deployment",
2728
Env: map[string]string{
28-
"CODER_ADDRESS": "0.0.0.0:8443",
29-
"CODER_ACCESS_URL": "https://dev.coder.com",
30-
"CODER_PG_CONNECTION_URL": "some-url",
31-
"CODER_PPROF_ADDRESS": "something",
32-
"CODER_PPROF_ENABLE": "true",
33-
"CODER_PROMETHEUS_ADDRESS": "hello-world",
34-
"CODER_PROMETHEUS_ENABLE": "true",
35-
"CODER_PROVISIONER_DAEMONS": "5",
36-
"CODER_SECURE_AUTH_COOKIE": "true",
37-
"CODER_SSH_KEYGEN_ALGORITHM": "potato",
38-
"CODER_TELEMETRY": "false",
39-
"CODER_TELEMETRY_TRACE": "false",
40-
"CODER_WILDCARD_ACCESS_URL": "something-wildcard.com",
29+
"CODER_ADDRESS": "0.0.0.0:8443",
30+
"CODER_ACCESS_URL": "https://dev.coder.com",
31+
"CODER_PG_CONNECTION_URL": "some-url",
32+
"CODER_PPROF_ADDRESS": "something",
33+
"CODER_PPROF_ENABLE": "true",
34+
"CODER_PROMETHEUS_ADDRESS": "hello-world",
35+
"CODER_PROMETHEUS_ENABLE": "true",
36+
"CODER_PROVISIONER_DAEMONS": "5",
37+
"CODER_PROVISIONER_DAEMON_POLL_INTERVAL": "5s",
38+
"CODER_PROVISIONER_DAEMON_POLL_JITTER": "1s",
39+
"CODER_SECURE_AUTH_COOKIE": "true",
40+
"CODER_SSH_KEYGEN_ALGORITHM": "potato",
41+
"CODER_TELEMETRY": "false",
42+
"CODER_TELEMETRY_TRACE": "false",
43+
"CODER_WILDCARD_ACCESS_URL": "something-wildcard.com",
4144
},
4245
Valid: func(config *codersdk.DeploymentConfig) {
4346
require.Equal(t, config.Address.Value, "0.0.0.0:8443")
@@ -48,6 +51,8 @@ func TestConfig(t *testing.T) {
4851
require.Equal(t, config.Prometheus.Address.Value, "hello-world")
4952
require.Equal(t, config.Prometheus.Enable.Value, true)
5053
require.Equal(t, config.Provisioner.Daemons.Value, 5)
54+
require.Equal(t, config.Provisioner.DaemonPollInterval.Value, 5*time.Second)
55+
require.Equal(t, config.Provisioner.DaemonPollJitter.Value, 1*time.Second)
5156
require.Equal(t, config.SecureAuthCookie.Value, true)
5257
require.Equal(t, config.SSHKeygenAlgorithm.Value, "potato")
5358
require.Equal(t, config.Telemetry.Enable.Value, false)

‎cli/server.go

Lines changed: 5 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -950,13 +950,16 @@ func newProvisionerDaemon(
950950
}()
951951
provisioners[string(database.ProvisionerTypeEcho)] = sdkproto.NewDRPCProvisionerClient(echoClient)
952952
}
953+
debounce := time.Second
953954
return provisionerd.New(func(ctx context.Context) (proto.DRPCProvisionerDaemonClient, error) {
954955
// This debounces calls to listen every second. Read the comment
955956
// in provisionerdserver.go to learn more!
956-
return coderAPI.CreateInMemoryProvisionerDaemon(ctx, time.Second)
957+
return coderAPI.CreateInMemoryProvisionerDaemon(ctx, debounce)
957958
}, &provisionerd.Options{
958959
Logger: logger,
959-
PollInterval: 500 * time.Millisecond,
960+
JobPollInterval: cfg.Provisioner.DaemonPollInterval.Value,
961+
JobPollJitter: cfg.Provisioner.DaemonPollJitter.Value,
962+
JobPollDebounce: debounce,
960963
UpdateInterval: 500 * time.Millisecond,
961964
ForceCancelInterval: cfg.Provisioner.ForceCancelInterval.Value,
962965
Provisioners: provisioners,

‎cli/testdata/coder_server_--help.golden

Lines changed: 9 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -128,6 +128,15 @@ Flags:
128128
--prometheus-enable Serve prometheus metrics on the address
129129
defined by prometheus address.
130130
Consumes $CODER_PROMETHEUS_ENABLE
131+
--provisioner-daemon-poll-interval duration Time to wait before polling for a new
132+
job.
133+
Consumes
134+
$CODER_PROVISIONER_DAEMON_POLL_INTERVAL
135+
(default 1s)
136+
--provisioner-daemon-poll-jitter duration Random jitter added to the poll interval.
137+
Consumes
138+
$CODER_PROVISIONER_DAEMON_POLL_JITTER
139+
(default 100ms)
131140
--provisioner-daemons int Number of provisioner daemons to create
132141
on start. If builds are stuck in queued
133142
state for a long time, consider

‎coderd/coderd.go

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -633,8 +633,8 @@ func compressHandler(h http.Handler) http.Handler {
633633
return cmp.Handler(h)
634634
}
635635

636-
// CreateInMemoryProvisionerDaemon is an in-memory connection to a provisionerd. Useful when starting coderd and provisionerd
637-
// in the same process.
636+
// CreateInMemoryProvisionerDaemon is an in-memory connection to a provisionerd.
637+
// Useful when starting coderd and provisionerd in the same process.
638638
func (api *API) CreateInMemoryProvisionerDaemon(ctx context.Context, debounce time.Duration) (client proto.DRPCProvisionerDaemonClient, err error) {
639639
clientSession, serverSession := provisionersdk.MemTransportPipe()
640640
defer func() {

‎coderd/coderdtest/coderdtest.go

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -335,7 +335,7 @@ func NewProvisionerDaemon(t *testing.T, coderAPI *coderd.API) io.Closer {
335335
}, &provisionerd.Options{
336336
Filesystem: fs,
337337
Logger: slogtest.Make(t, nil).Named("provisionerd").Leveled(slog.LevelDebug),
338-
PollInterval: 50 * time.Millisecond,
338+
JobPollInterval: 50 * time.Millisecond,
339339
UpdateInterval: 250 * time.Millisecond,
340340
ForceCancelInterval: time.Second,
341341
Provisioners: provisionerd.Provisioners{
@@ -370,7 +370,7 @@ func NewExternalProvisionerDaemon(t *testing.T, client *codersdk.Client, org uui
370370
}, &provisionerd.Options{
371371
Filesystem: fs,
372372
Logger: slogtest.Make(t, nil).Named("provisionerd").Leveled(slog.LevelDebug),
373-
PollInterval: 50 * time.Millisecond,
373+
JobPollInterval: 50 * time.Millisecond,
374374
UpdateInterval: 250 * time.Millisecond,
375375
ForceCancelInterval: time.Second,
376376
Provisioners: provisionerd.Provisioners{

‎codersdk/deploymentconfig.go

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -134,6 +134,8 @@ type GitAuthConfig struct {
134134

135135
type ProvisionerConfig struct {
136136
Daemons *DeploymentConfigField[int] `json:"daemons" typescript:",notnull"`
137+
DaemonPollInterval *DeploymentConfigField[time.Duration] `json:"daemon_poll_interval" typescript:",notnull"`
138+
DaemonPollJitter *DeploymentConfigField[time.Duration] `json:"daemon_poll_jitter" typescript:",notnull"`
137139
ForceCancelInterval *DeploymentConfigField[time.Duration] `json:"force_cancel_interval" typescript:",notnull"`
138140
}
139141

‎enterprise/cli/provisionerdaemons.go

Lines changed: 17 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -7,6 +7,9 @@ import (
77
"os/signal"
88
"time"
99

10+
"github.com/spf13/cobra"
11+
"golang.org/x/xerrors"
12+
1013
"cdr.dev/slog"
1114
"cdr.dev/slog/sloggers/sloghuman"
1215
agpl "github.com/coder/coder/cli"
@@ -20,9 +23,6 @@ import (
2023
provisionerdproto "github.com/coder/coder/provisionerd/proto"
2124
"github.com/coder/coder/provisionersdk"
2225
"github.com/coder/coder/provisionersdk/proto"
23-
24-
"github.com/spf13/cobra"
25-
"golang.org/x/xerrors"
2626
)
2727

2828
func provisionerDaemons() *cobra.Command {
@@ -37,8 +37,10 @@ func provisionerDaemons() *cobra.Command {
3737

3838
func provisionerDaemonStart() *cobra.Command {
3939
var (
40-
cacheDir string
41-
rawTags []string
40+
cacheDir string
41+
rawTags []string
42+
pollInterval time.Duration
43+
pollJitter time.Duration
4244
)
4345
cmd := &cobra.Command{
4446
Use: "start",
@@ -111,11 +113,12 @@ func provisionerDaemonStart() *cobra.Command {
111113
codersdk.ProvisionerTypeTerraform,
112114
}, tags)
113115
}, &provisionerd.Options{
114-
Logger: logger,
115-
PollInterval: 500 * time.Millisecond,
116-
UpdateInterval: 500 * time.Millisecond,
117-
Provisioners: provisioners,
118-
WorkDirectory: tempDir,
116+
Logger: logger,
117+
JobPollInterval: pollInterval,
118+
JobPollJitter: pollJitter,
119+
UpdateInterval: 500 * time.Millisecond,
120+
Provisioners: provisioners,
121+
WorkDirectory: tempDir,
119122
})
120123

121124
var exitErr error
@@ -150,6 +153,10 @@ func provisionerDaemonStart() *cobra.Command {
150153
"Specify a directory to cache provisioner job files.")
151154
cliflag.StringArrayVarP(cmd.Flags(), &rawTags, "tag", "t", "CODER_PROVISIONERD_TAGS", []string{},
152155
"Specify a list of tags to target provisioner jobs.")
156+
cliflag.DurationVarP(cmd.Flags(), &pollInterval, "poll-interval", "", "CODER_PROVISIONERD_POLL_INTERVAL", time.Second,
157+
"Specify the interval for which the provisioner daemon should poll for jobs.")
158+
cliflag.DurationVarP(cmd.Flags(), &pollJitter, "poll-jitter", "", "CODER_PROVISIONERD_POLL_JITTER", 100*time.Millisecond,
159+
"Random jitter added to the poll interval.")
153160

154161
return cmd
155162
}

‎enterprise/coderd/provisionerdaemons.go

Lines changed: 0 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -18,7 +18,6 @@ import (
1818
"storj.io/drpc/drpcserver"
1919

2020
"cdr.dev/slog"
21-
2221
"github.com/coder/coder/coderd"
2322
"github.com/coder/coder/coderd/database"
2423
"github.com/coder/coder/coderd/httpapi"

‎provisionerd/provisionerd.go

Lines changed: 48 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -23,6 +23,7 @@ import (
2323

2424
"cdr.dev/slog"
2525
"github.com/coder/coder/coderd/tracing"
26+
"github.com/coder/coder/cryptorand"
2627
"github.com/coder/coder/provisionerd/proto"
2728
"github.com/coder/coder/provisionerd/runner"
2829
sdkproto "github.com/coder/coder/provisionersdk/proto"
@@ -52,7 +53,9 @@ type Options struct {
5253
ForceCancelInterval time.Duration
5354
UpdateInterval time.Duration
5455
LogBufferInterval time.Duration
55-
PollInterval time.Duration
56+
JobPollInterval time.Duration
57+
JobPollJitter time.Duration
58+
JobPollDebounce time.Duration
5659
Provisioners Provisioners
5760
WorkDirectory string
5861
}
@@ -62,8 +65,14 @@ func New(clientDialer Dialer, opts *Options) *Server {
6265
if opts == nil {
6366
opts = &Options{}
6467
}
65-
if opts.PollInterval == 0 {
66-
opts.PollInterval = 5 * time.Second
68+
if opts.JobPollInterval == 0 {
69+
opts.JobPollInterval = 5 * time.Second
70+
}
71+
if opts.JobPollJitter == 0 {
72+
opts.JobPollJitter = time.Second
73+
}
74+
if opts.JobPollDebounce == 0 {
75+
opts.JobPollDebounce = time.Second
6776
}
6877
if opts.UpdateInterval == 0 {
6978
opts.UpdateInterval = 5 * time.Second
@@ -205,8 +214,8 @@ func (p *Server) connect(ctx context.Context) {
205214
if p.isClosed() {
206215
return
207216
}
208-
ticker := time.NewTicker(p.opts.PollInterval)
209-
defer ticker.Stop()
217+
timer := time.NewTimer(p.opts.JobPollInterval)
218+
defer timer.Stop()
210219
for {
211220
client, ok := p.client()
212221
if !ok {
@@ -217,13 +226,23 @@ func (p *Server) connect(ctx context.Context) {
217226
return
218227
case <-client.DRPCConn().Closed():
219228
return
220-
case <-ticker.C:
229+
case <-timer.C:
221230
p.acquireJob(ctx)
231+
timer.Reset(p.nextInterval())
222232
}
223233
}
224234
}()
225235
}
226236

237+
func (p *Server) nextInterval() time.Duration {
238+
r, err := cryptorand.Float64()
239+
if err != nil {
240+
panic("get random float:" + err.Error())
241+
}
242+
243+
return p.opts.JobPollInterval + time.Duration(float64(p.opts.JobPollJitter)*r)
244+
}
245+
227246
func (p *Server) client() (proto.DRPCProvisionerDaemonClient, bool) {
228247
rawClient := p.clientValue.Load()
229248
if rawClient == nil {
@@ -246,6 +265,11 @@ func (p *Server) isRunningJob() bool {
246265
}
247266
}
248267

268+
var (
269+
lastAcquire time.Time
270+
lastAcquireMutex sync.RWMutex
271+
)
272+
249273
// Locks a job in the database, and runs it!
250274
func (p *Server) acquireJob(ctx context.Context) {
251275
p.mutex.Lock()
@@ -261,6 +285,18 @@ func (p *Server) acquireJob(ctx context.Context) {
261285
return
262286
}
263287

288+
// This prevents loads of provisioner daemons from consistently sending
289+
// requests when no jobs are available.
290+
//
291+
// The debounce only occurs when no job is returned, so if loads of jobs are
292+
// added at once, they will start after at most this duration.
293+
lastAcquireMutex.RLock()
294+
if !lastAcquire.IsZero() && time.Since(lastAcquire) < p.opts.JobPollDebounce {
295+
lastAcquireMutex.RUnlock()
296+
return
297+
}
298+
lastAcquireMutex.RUnlock()
299+
264300
var err error
265301
client, ok := p.client()
266302
if !ok {
@@ -269,17 +305,19 @@ func (p *Server) acquireJob(ctx context.Context) {
269305

270306
job, err := client.AcquireJob(ctx, &proto.Empty{})
271307
if err != nil {
272-
if errors.Is(err, context.Canceled) {
273-
return
274-
}
275-
if errors.Is(err, yamux.ErrSessionShutdown) {
308+
if errors.Is(err, context.Canceled) ||
309+
errors.Is(err, yamux.ErrSessionShutdown) ||
310+
errors.Is(err, fasthttputil.ErrInmemoryListenerClosed) {
276311
return
277312
}
278313

279314
p.opts.Logger.Warn(ctx, "acquire job", slog.Error(err))
280315
return
281316
}
282317
if job.JobId == "" {
318+
lastAcquireMutex.Lock()
319+
lastAcquire = time.Now()
320+
lastAcquireMutex.Unlock()
283321
return
284322
}
285323

‎provisionerd/provisionerd_test.go

Lines changed: 5 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -1053,11 +1053,11 @@ func createTar(t *testing.T, files map[string]string) []byte {
10531053
// Creates a provisionerd implementation with the provided dialer and provisioners.
10541054
func createProvisionerd(t *testing.T, dialer provisionerd.Dialer, provisioners provisionerd.Provisioners) *provisionerd.Server {
10551055
server := provisionerd.New(dialer, &provisionerd.Options{
1056-
Logger: slogtest.Make(t, nil).Named("provisionerd").Leveled(slog.LevelDebug),
1057-
PollInterval: 50 * time.Millisecond,
1058-
UpdateInterval: 50 * time.Millisecond,
1059-
Provisioners: provisioners,
1060-
WorkDirectory: t.TempDir(),
1056+
Logger: slogtest.Make(t, nil).Named("provisionerd").Leveled(slog.LevelDebug),
1057+
JobPollInterval: 50 * time.Millisecond,
1058+
UpdateInterval: 50 * time.Millisecond,
1059+
Provisioners: provisioners,
1060+
WorkDirectory: t.TempDir(),
10611061
})
10621062
t.Cleanup(func() {
10631063
_ = server.Close()

‎site/src/api/typesGenerated.ts

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -526,6 +526,8 @@ export interface PrometheusConfig {
526526
// From codersdk/deploymentconfig.go
527527
export interface ProvisionerConfig {
528528
readonly daemons: DeploymentConfigField<number>
529+
readonly daemon_poll_interval: DeploymentConfigField<number>
530+
readonly daemon_poll_jitter: DeploymentConfigField<number>
529531
readonly force_cancel_interval: DeploymentConfigField<number>
530532
}
531533

0 commit comments

Comments
 (0)
0