fix(worker): set session.CloseCallTimeout to non-zero value

hashicorp · Feb 4, 2025 · 6e8d6a1 · 6e8d6a1
1 parent 43338db
commit 6e8d6a1
Show file tree

Hide file tree

Showing 2 changed files with 94 additions and 1 deletion.
diff --git a/internal/daemon/worker/worker.go b/internal/daemon/worker/worker.go
@@ -485,7 +485,7 @@ func (w *Worker) Reload(ctx context.Context, newConf *config.Config) {
 
 	w.parseAndStoreTags(newConf.Worker.Tags)
 
-	switch newConf.Worker.SuccessfulControllerRPCGracePeriod {
+	switch newConf.Worker.SuccessfulControllerRPCGracePeriodDuration {
 	case 0:
 		w.successfulRoutingInfoGracePeriod.Store(int64(server.DefaultLiveness))
 		w.successfulSessionInfoGracePeriod.Store(int64(server.DefaultLiveness))

diff --git a/internal/daemon/worker/worker_test.go b/internal/daemon/worker/worker_test.go
@@ -15,6 +15,7 @@ import (
 
 	"github.com/hashicorp/boundary/internal/cmd/base"
 	"github.com/hashicorp/boundary/internal/cmd/config"
+	"github.com/hashicorp/boundary/internal/daemon/worker/common"
 	"github.com/hashicorp/boundary/internal/daemon/worker/session"
 	"github.com/hashicorp/boundary/internal/db"
 	"github.com/hashicorp/boundary/internal/event"
@@ -223,6 +224,98 @@ func TestWorkerNew(t *testing.T) {
 	}
 }
 
+func TestWorkerReload(t *testing.T) {
+	t.Run("default config is the same as the reload config", func(t *testing.T) {
+		require, assert := require.New(t), assert.New(t)
+		cfg := &Config{
+			Server: &base.Server{
+				Logger:  hclog.Default(),
+				Eventer: &event.Eventer{},
+				Listeners: []*base.ServerListener{
+					{Config: &listenerutil.ListenerConfig{Purpose: []string{"api"}}},
+					{Config: &listenerutil.ListenerConfig{Purpose: []string{"proxy"}}},
+					{Config: &listenerutil.ListenerConfig{Purpose: []string{"cluster"}}},
+				},
+			},
+			RawConfig: &config.Config{
+				SharedConfig: &configutil.SharedConfig{DisableMlock: true},
+				Worker:       &config.Worker{},
+			},
+		}
+		w, err := New(context.Background(), cfg)
+		require.NoError(err)
+
+		assert.Equal(int64(server.DefaultLiveness), w.successfulRoutingInfoGracePeriod.Load())
+		assert.Equal(int64(server.DefaultLiveness), w.successfulSessionInfoGracePeriod.Load())
+		assert.Equal(int64(server.DefaultLiveness), session.CloseCallTimeout.Load())
+
+		assert.Equal(int64(common.DefaultRoutingInfoTimeout), w.routingInfoCallTimeoutDuration.Load())
+		assert.Equal(int64(common.DefaultStatisticsTimeout), w.statisticsCallTimeoutDuration.Load())
+		assert.Equal(int64(common.DefaultSessionInfoTimeout), w.sessionInfoCallTimeoutDuration.Load())
+
+		assert.Equal(int64(server.DefaultLiveness), w.getDownstreamWorkersTimeoutDuration.Load())
+
+		w.Reload(context.Background(), cfg.RawConfig)
+
+		assert.Equal(int64(server.DefaultLiveness), w.successfulRoutingInfoGracePeriod.Load())
+		assert.Equal(int64(server.DefaultLiveness), w.successfulSessionInfoGracePeriod.Load())
+		assert.Equal(int64(server.DefaultLiveness), session.CloseCallTimeout.Load())
+
+		assert.Equal(int64(common.DefaultRoutingInfoTimeout), w.routingInfoCallTimeoutDuration.Load())
+		assert.Equal(int64(common.DefaultStatisticsTimeout), w.statisticsCallTimeoutDuration.Load())
+		assert.Equal(int64(common.DefaultSessionInfoTimeout), w.sessionInfoCallTimeoutDuration.Load())
+
+		assert.Equal(int64(server.DefaultLiveness), w.getDownstreamWorkersTimeoutDuration.Load())
+	})
+
+	t.Run("new config is the same as the reload config", func(t *testing.T) {
+		require, assert := require.New(t), assert.New(t)
+		cfg := &Config{
+			Server: &base.Server{
+				Logger:  hclog.Default(),
+				Eventer: &event.Eventer{},
+				Listeners: []*base.ServerListener{
+					{Config: &listenerutil.ListenerConfig{Purpose: []string{"api"}}},
+					{Config: &listenerutil.ListenerConfig{Purpose: []string{"proxy"}}},
+					{Config: &listenerutil.ListenerConfig{Purpose: []string{"cluster"}}},
+				},
+			},
+			RawConfig: &config.Config{
+				SharedConfig: &configutil.SharedConfig{DisableMlock: true},
+				Worker: &config.Worker{
+					SuccessfulControllerRPCGracePeriodDuration: 5 * time.Second,
+					ControllerRPCCallTimeoutDuration:           10 * time.Second,
+					GetDownstreamWorkersTimeoutDuration:        20 * time.Second,
+				},
+			},
+		}
+		w, err := New(context.Background(), cfg)
+		require.NoError(err)
+
+		assert.Equal(int64(5*time.Second), w.successfulRoutingInfoGracePeriod.Load())
+		assert.Equal(int64(5*time.Second), w.successfulSessionInfoGracePeriod.Load())
+		assert.Equal(w.successfulRoutingInfoGracePeriod.Load(), session.CloseCallTimeout.Load())
+
+		assert.Equal(int64(10*time.Second), w.routingInfoCallTimeoutDuration.Load())
+		assert.Equal(int64(10*time.Second), w.statisticsCallTimeoutDuration.Load())
+		assert.Equal(int64(10*time.Second), w.sessionInfoCallTimeoutDuration.Load())
+
+		assert.Equal(int64(20*time.Second), w.getDownstreamWorkersTimeoutDuration.Load())
+
+		w.Reload(context.Background(), cfg.RawConfig)
+
+		assert.Equal(int64(5*time.Second), w.successfulRoutingInfoGracePeriod.Load())
+		assert.Equal(int64(5*time.Second), w.successfulSessionInfoGracePeriod.Load())
+		assert.Equal(w.successfulRoutingInfoGracePeriod.Load(), session.CloseCallTimeout.Load())
+
+		assert.Equal(int64(10*time.Second), w.routingInfoCallTimeoutDuration.Load())
+		assert.Equal(int64(10*time.Second), w.statisticsCallTimeoutDuration.Load())
+		assert.Equal(int64(10*time.Second), w.sessionInfoCallTimeoutDuration.Load())
+
+		assert.Equal(int64(20*time.Second), w.getDownstreamWorkersTimeoutDuration.Load())
+	})
+}
+
 func TestSetupWorkerAuthStorage(t *testing.T) {
 	ctx := context.Background()