opensource/coder
3
0
Fork 0
mirror of https://github.com/coder/coder.git synced 2026-09-24 15:04:27 +08:00
Code Issues Packages Projects Releases Wiki Activity

fix: Debounce AcquireJob when no jobs are available (#5017)

Browse Source 
This prevents constant database spam at scale to a maximum
of 60 queries/s per coderd instance.
This commit is contained in:
Kyle Carberry
2022-11-10 22:37:33 +00:00
committed by GitHub
parent f32748c929
commit 927c241995
5 changed files with 75 additions and 15 deletions
Download Patch File Download Diff File Expand all files Collapse all files
cli/server.go
+9 -4
View File
@@ -68,8 +68,9 @@ import (
"github.com/coder/coder/provisioner/echo"
"github.com/coder/coder/provisioner/terraform"
"github.com/coder/coder/provisionerd"
"github.com/coder/coder/provisionerd/proto"
"github.com/coder/coder/provisionersdk"
"github.com/coder/coder/provisionersdk/proto"
sdkproto "github.com/coder/coder/provisionersdk/proto"
"github.com/coder/coder/tailnet"
)
@@ -900,7 +901,7 @@ func newProvisionerDaemon(
}
provisioners := provisionerd.Provisioners{
string(database.ProvisionerTypeTerraform): proto.NewDRPCProvisionerClient(provisionersdk.Conn(terraformClient)),
string(database.ProvisionerTypeTerraform): sdkproto.NewDRPCProvisionerClient(provisionersdk.Conn(terraformClient)),
}
// include echo provisioner when in dev mode
if dev {
@@ -921,9 +922,13 @@ func newProvisionerDaemon(
}
}
}()
provisioners[string(database.ProvisionerTypeEcho)] = proto.NewDRPCProvisionerClient(provisionersdk.Conn(echoClient))
provisioners[string(database.ProvisionerTypeEcho)] = sdkproto.NewDRPCProvisionerClient(provisionersdk.Conn(echoClient))
}
return provisionerd.New(coderAPI.ListenProvisionerDaemon, &provisionerd.Options{
return provisionerd.New(func(ctx context.Context) (proto.DRPCProvisionerDaemonClient, error) {
// This debounces calls to listen every second. Read the comment
// in provisionerdserver.go to learn more!
return coderAPI.ListenProvisionerDaemon(ctx, time.Second)
}, &provisionerd.Options{
Logger: logger,
PollInterval: 500 * time.Millisecond,
UpdateInterval: 500 * time.Millisecond,
coderd/coderdtest/coderdtest.go
+6 -3
View File
@@ -69,8 +69,9 @@ import (
"github.com/coder/coder/cryptorand"
"github.com/coder/coder/provisioner/echo"
"github.com/coder/coder/provisionerd"
"github.com/coder/coder/provisionerd/proto"
"github.com/coder/coder/provisionersdk"
"github.com/coder/coder/provisionersdk/proto"
sdkproto "github.com/coder/coder/provisionersdk/proto"
"github.com/coder/coder/tailnet"
"github.com/coder/coder/testutil"
)
@@ -325,14 +326,16 @@ func NewProvisionerDaemon(t *testing.T, coderAPI *coderd.API) io.Closer {
assert.NoError(t, err)
}()
closer := provisionerd.New(coderAPI.ListenProvisionerDaemon, &provisionerd.Options{
closer := provisionerd.New(func(ctx context.Context) (proto.DRPCProvisionerDaemonClient, error) {
return coderAPI.ListenProvisionerDaemon(ctx, 0)
}, &provisionerd.Options{
Filesystem: fs,
Logger: slogtest.Make(t, nil).Named("provisionerd").Leveled(slog.LevelDebug),
PollInterval: 50 * time.Millisecond,
UpdateInterval: 250 * time.Millisecond,
ForceCancelInterval: time.Second,
Provisioners: provisionerd.Provisioners{
string(database.ProvisionerTypeEcho): proto.NewDRPCProvisionerClient(provisionersdk.Conn(echoClient)),
string(database.ProvisionerTypeEcho): sdkproto.NewDRPCProvisionerClient(provisionersdk.Conn(echoClient)),
},
WorkDirectory: t.TempDir(),
})
coderd/provisionerdaemons.go
+10 -8
View File
@@ -7,6 +7,7 @@ import (
"fmt"
"io"
"net/http"
"time"
"github.com/google/uuid"
"github.com/moby/moby/pkg/namesgenerator"
@@ -55,7 +56,7 @@ func (api *API) provisionerDaemons(rw http.ResponseWriter, r *http.Request) {
// ListenProvisionerDaemon is an in-memory connection to a provisionerd. Useful when starting coderd and provisionerd
// in the same process.
func (api *API) ListenProvisionerDaemon(ctx context.Context) (client proto.DRPCProvisionerDaemonClient, err error) {
func (api *API) ListenProvisionerDaemon(ctx context.Context, acquireJobDebounce time.Duration) (client proto.DRPCProvisionerDaemonClient, err error) {
clientSession, serverSession := provisionersdk.TransportPipe()
defer func() {
if err != nil {
@@ -77,13 +78,14 @@ func (api *API) ListenProvisionerDaemon(ctx context.Context) (client proto.DRPCP
mux := drpcmux.New()
err = proto.DRPCRegisterProvisionerDaemon(mux, &provisionerdserver.Server{
AccessURL: api.AccessURL,
ID: daemon.ID,
Database: api.Database,
Pubsub: api.Pubsub,
Provisioners: daemon.Provisioners,
Telemetry: api.Telemetry,
Logger: api.Logger.Named(fmt.Sprintf("provisionerd-%s", daemon.Name)),
AccessURL: api.AccessURL,
ID: daemon.ID,
Database: api.Database,
Pubsub: api.Pubsub,
Provisioners: daemon.Provisioners,
Telemetry: api.Telemetry,
Logger: api.Logger.Named(fmt.Sprintf("provisionerd-%s", daemon.Name)),
AcquireJobDebounce: acquireJobDebounce,
})
if err != nil {
return nil, err
coderd/provisionerdserver/provisionerdserver.go
+22
View File
@@ -8,6 +8,7 @@ import (
"fmt"
"net/url"
"reflect"
"sync"
"time"
"github.com/google/uuid"
@@ -27,6 +28,11 @@ import (
sdkproto "github.com/coder/coder/provisionersdk/proto"
)
var (
lastAcquire time.Time
lastAcquireMutex sync.RWMutex
)
type Server struct {
AccessURL *url.URL
ID uuid.UUID
@@ -35,10 +41,23 @@ type Server struct {
Database database.Store
Pubsub database.Pubsub
Telemetry telemetry.Reporter
AcquireJobDebounce time.Duration
}
// AcquireJob queries the database to lock a job.
func (server *Server) AcquireJob(ctx context.Context, _ *proto.Empty) (*proto.AcquiredJob, error) {
// This prevents loads of provisioner daemons from consistently
// querying the database when no jobs are available.
//
// The debounce only occurs when no job is returned, so if loads of
// jobs are added at once, they will start after at most this duration.
lastAcquireMutex.RLock()
if !lastAcquire.IsZero() && time.Since(lastAcquire) < server.AcquireJobDebounce {
lastAcquireMutex.RUnlock()
return &proto.AcquiredJob{}, nil
}
lastAcquireMutex.RUnlock()
// This marks the job as locked in the database.
job, err := server.Database.AcquireProvisionerJob(ctx, database.AcquireProvisionerJobParams{
StartedAt: sql.NullTime{
@@ -54,6 +73,9 @@ func (server *Server) AcquireJob(ctx context.Context, _ *proto.Empty) (*proto.Ac
if errors.Is(err, sql.ErrNoRows) {
// The provisioner daemon assumes no jobs are available if
// an empty struct is returned.
lastAcquireMutex.Lock()
lastAcquire = time.Now()
lastAcquireMutex.Unlock()
return &proto.AcquiredJob{}, nil
}
if err != nil {
coderd/provisionerdserver/provisionerdserver_test.go
+28
View File
@@ -6,6 +6,7 @@ import (
"encoding/json"
"net/url"
"testing"
"time"
"github.com/google/uuid"
"github.com/stretchr/testify/require"
@@ -22,6 +23,33 @@ import (
func TestAcquireJob(t *testing.T) {
t.Parallel()
t.Run("Debounce", func(t *testing.T) {
t.Parallel()
db := databasefake.New()
pubsub := database.NewPubsubInMemory()
srv := &provisionerdserver.Server{
ID: uuid.New(),
Logger: slogtest.Make(t, nil),
AccessURL: &url.URL{},
Provisioners: []database.ProvisionerType{database.ProvisionerTypeEcho},
Database: db,
Pubsub: pubsub,
Telemetry: telemetry.NewNoop(),
AcquireJobDebounce: time.Hour,
}
job, err := srv.AcquireJob(context.Background(), nil)
require.NoError(t, err)
require.Equal(t, &proto.AcquiredJob{}, job)
_, err = srv.Database.InsertProvisionerJob(context.Background(), database.InsertProvisionerJobParams{
ID: uuid.New(),
InitiatorID: uuid.New(),
Provisioner: database.ProvisionerTypeEcho,
})
require.NoError(t, err)
job, err = srv.AcquireJob(context.Background(), nil)
require.NoError(t, err)
require.Equal(t, &proto.AcquiredJob{}, job)
})
t.Run("NoJobs", func(t *testing.T) {
t.Parallel()
srv := setup(t)