From 244eedb4e3955a0fe19bb477263c7c8f4cd3e787 Mon Sep 17 00:00:00 2001 From: Will Jordan Date: Fri, 5 Nov 2021 09:25:25 -0700 Subject: [PATCH] add deadline to DirectRPC interface --- agent/auto-config/auto_config.go | 2 +- agent/auto-config/auto_config_test.go | 37 +++++++++++++++++++------- agent/auto-config/auto_encrypt.go | 3 ++- agent/auto-config/auto_encrypt_test.go | 9 +++++-- agent/auto-config/config.go | 2 +- agent/auto-config/mock_test.go | 9 ++++--- 6 files changed, 43 insertions(+), 19 deletions(-) diff --git a/agent/auto-config/auto_config.go b/agent/auto-config/auto_config.go index f3eedb7eb7ab..bc0d3f54dc37 100644 --- a/agent/auto-config/auto_config.go +++ b/agent/auto-config/auto_config.go @@ -298,7 +298,7 @@ func (ac *AutoConfig) getInitialConfigurationOnce(ctx context.Context, csr strin } ac.logger.Debug("making AutoConfig.InitialConfiguration RPC", "addr", addr.String()) - if err = ac.acConfig.DirectRPC.RPC(ac.config.Datacenter, ac.config.NodeName, &addr, "AutoConfig.InitialConfiguration", &request, &resp); err != nil { + if err = ac.acConfig.DirectRPC.RPC(ac.config.Datacenter, ac.config.NodeName, &addr, "AutoConfig.InitialConfiguration", &request, &resp, time.Time{}); err != nil { ac.logger.Error("AutoConfig.InitialConfiguration RPC failed", "addr", addr.String(), "error", err) continue } diff --git a/agent/auto-config/auto_config_test.go b/agent/auto-config/auto_config_test.go index 37e9f67ab3db..f906c142fda2 100644 --- a/agent/auto-config/auto_config_test.go +++ b/agent/auto-config/auto_config_test.go @@ -257,7 +257,15 @@ func TestInitialConfiguration_cancelled(t *testing.T) { JWT: "blarg", } - mcfg.directRPC.On("RPC", "dc1", "autoconf", &net.TCPAddr{IP: net.IPv4(127, 0, 0, 1), Port: 8300}, "AutoConfig.InitialConfiguration", &expectedRequest, mock.Anything).Return(fmt.Errorf("injected error")).Times(0).Maybe() + mcfg.directRPC.On( + "RPC", + "dc1", + "autoconf", + &net.TCPAddr{IP: net.IPv4(127, 0, 0, 1), Port: 8300}, + "AutoConfig.InitialConfiguration", + &expectedRequest, + mock.Anything, + time.Time{}).Return(fmt.Errorf("injected error")).Times(0).Maybe() mcfg.serverProvider.On("FindLANServer").Return(nil).Times(0).Maybe() ac, err := New(mcfg.Config) @@ -386,7 +394,8 @@ func TestInitialConfiguration_success(t *testing.T) { &net.TCPAddr{IP: net.IPv4(127, 0, 0, 1), Port: 8300}, "AutoConfig.InitialConfiguration", &expectedRequest, - &pbautoconf.AutoConfigResponse{}).Return(nil).Run(populateResponse) + &pbautoconf.AutoConfigResponse{}, + time.Time{}).Return(nil).Run(populateResponse) ac, err := New(mcfg.Config) require.NoError(t, err) @@ -471,7 +480,8 @@ func TestInitialConfiguration_retries(t *testing.T) { &net.TCPAddr{IP: net.IPv4(198, 18, 0, 1), Port: 8300}, "AutoConfig.InitialConfiguration", &expectedRequest, - &pbautoconf.AutoConfigResponse{}).Return(fmt.Errorf("injected failure")).Times(0) + &pbautoconf.AutoConfigResponse{}, + time.Time{}).Return(fmt.Errorf("injected failure")).Times(0) mcfg.directRPC.On( "RPC", "dc1", @@ -479,7 +489,8 @@ func TestInitialConfiguration_retries(t *testing.T) { &net.TCPAddr{IP: net.IPv4(198, 18, 0, 2), Port: 8398}, "AutoConfig.InitialConfiguration", &expectedRequest, - &pbautoconf.AutoConfigResponse{}).Return(fmt.Errorf("injected failure")).Times(0) + &pbautoconf.AutoConfigResponse{}, + time.Time{}).Return(fmt.Errorf("injected failure")).Times(0) mcfg.directRPC.On( "RPC", "dc1", @@ -487,7 +498,8 @@ func TestInitialConfiguration_retries(t *testing.T) { &net.TCPAddr{IP: net.IPv4(198, 18, 0, 3), Port: 8399}, "AutoConfig.InitialConfiguration", &expectedRequest, - &pbautoconf.AutoConfigResponse{}).Return(fmt.Errorf("injected failure")).Times(0) + &pbautoconf.AutoConfigResponse{}, + time.Time{}).Return(fmt.Errorf("injected failure")).Times(0) mcfg.directRPC.On( "RPC", "dc1", @@ -495,7 +507,8 @@ func TestInitialConfiguration_retries(t *testing.T) { &net.TCPAddr{IP: net.IPv4(127, 0, 0, 1), Port: 1234}, "AutoConfig.InitialConfiguration", &expectedRequest, - &pbautoconf.AutoConfigResponse{}).Return(fmt.Errorf("injected failure")).Once() + &pbautoconf.AutoConfigResponse{}, + time.Time{}).Return(fmt.Errorf("injected failure")).Once() mcfg.directRPC.On( "RPC", "dc1", @@ -503,7 +516,8 @@ func TestInitialConfiguration_retries(t *testing.T) { &net.TCPAddr{IP: net.IPv4(127, 0, 0, 1), Port: 1234}, "AutoConfig.InitialConfiguration", &expectedRequest, - &pbautoconf.AutoConfigResponse{}).Return(nil).Run(populateResponse).Once() + &pbautoconf.AutoConfigResponse{}, + time.Time{}).Return(nil).Run(populateResponse).Once() ac, err := New(mcfg.Config) require.NoError(t, err) @@ -789,7 +803,8 @@ func startedAutoConfig(t *testing.T, autoEncrypt bool) testAutoConfig { &net.TCPAddr{IP: net.IPv4(198, 18, 0, 1), Port: 8300}, "AutoConfig.InitialConfiguration", &expectedRequest, - &pbautoconf.AutoConfigResponse{}).Return(nil).Run(populateResponse).Once() + &pbautoconf.AutoConfigResponse{}, + time.Time{}).Return(nil).Run(populateResponse).Once() } else { expectedRequest := structs.CASignRequest{ WriteRequest: structs.WriteRequest{Token: originalToken}, @@ -809,7 +824,8 @@ func startedAutoConfig(t *testing.T, autoEncrypt bool) testAutoConfig { &net.TCPAddr{IP: net.IPv4(198, 18, 0, 1), Port: 8300}, "AutoEncrypt.Sign", &expectedRequest, - &structs.SignedResponse{}).Return(nil).Run(populateResponse) + &structs.SignedResponse{}, + time.Time{}).Return(nil).Run(populateResponse) } ac, err := New(mcfg.Config) @@ -1083,7 +1099,8 @@ func TestFallback(t *testing.T) { &net.TCPAddr{IP: net.IPv4(198, 18, 23, 2), Port: 8300}, "AutoConfig.InitialConfiguration", &expectedRequest, - &pbautoconf.AutoConfigResponse{}).Return(nil).Run(populateResponse).Once() + &pbautoconf.AutoConfigResponse{}, + time.Time{}).Return(nil).Run(populateResponse).Once() // this gets called when InitialConfiguration is invoked to record the token from the // auto-config response which is how the Fallback for auto-config works diff --git a/agent/auto-config/auto_encrypt.go b/agent/auto-config/auto_encrypt.go index d42e25d47bf2..3eb7b8a3501e 100644 --- a/agent/auto-config/auto_encrypt.go +++ b/agent/auto-config/auto_encrypt.go @@ -5,6 +5,7 @@ import ( "fmt" "net" "strings" + "time" "github.com/hashicorp/consul/agent/structs" ) @@ -56,7 +57,7 @@ func (ac *AutoConfig) autoEncryptInitialCertsOnce(ctx context.Context, csr, key } ac.logger.Debug("making AutoEncrypt.Sign RPC", "addr", addr.String()) - err = ac.acConfig.DirectRPC.RPC(ac.config.Datacenter, ac.config.NodeName, &addr, "AutoEncrypt.Sign", &request, &resp) + err = ac.acConfig.DirectRPC.RPC(ac.config.Datacenter, ac.config.NodeName, &addr, "AutoEncrypt.Sign", &request, &resp, time.Time{}) if err != nil { ac.logger.Error("AutoEncrypt.Sign RPC failed", "addr", addr.String(), "error", err) continue diff --git a/agent/auto-config/auto_encrypt_test.go b/agent/auto-config/auto_encrypt_test.go index 1bcd6bf9abe5..6bef134a4c3f 100644 --- a/agent/auto-config/auto_encrypt_test.go +++ b/agent/auto-config/auto_encrypt_test.go @@ -203,6 +203,7 @@ func TestAutoEncrypt_InitialCerts(t *testing.T) { "AutoEncrypt.Sign", &request, &structs.SignedResponse{}, + time.Time{}, ).Once().Return(fmt.Errorf("injected error")) // second failure mcfg.directRPC.On("RPC", @@ -212,6 +213,7 @@ func TestAutoEncrypt_InitialCerts(t *testing.T) { "AutoEncrypt.Sign", &request, &structs.SignedResponse{}, + time.Time{}, ).Once().Return(fmt.Errorf("injected error")) // third times is successfuly (second attempt to first server) mcfg.directRPC.On("RPC", @@ -221,6 +223,7 @@ func TestAutoEncrypt_InitialCerts(t *testing.T) { "AutoEncrypt.Sign", &request, &structs.SignedResponse{}, + time.Time{}, ).Once().Return(nil).Run(func(args mock.Arguments) { resp, ok := args.Get(5).(*structs.SignedResponse) require.True(t, ok) @@ -308,7 +311,8 @@ func TestAutoEncrypt_InitialConfiguration(t *testing.T) { &net.TCPAddr{IP: net.IPv4(127, 0, 0, 1), Port: 8300}, "AutoEncrypt.Sign", &expectedRequest, - &structs.SignedResponse{}).Return(nil).Run(populateResponse) + &structs.SignedResponse{}, + time.Time{}).Return(nil).Run(populateResponse) ac, err := New(mcfg.Config) require.NoError(t, err) @@ -517,7 +521,8 @@ func TestAutoEncrypt_Fallback(t *testing.T) { &net.TCPAddr{IP: net.IPv4(198, 18, 23, 2), Port: 8300}, "AutoEncrypt.Sign", &expectedRequest, - &structs.SignedResponse{}).Return(nil).Run(populateResponse).Once() + &structs.SignedResponse{}, + time.Time{}).Return(nil).Run(populateResponse).Once() testAC.mcfg.expectInitialTLS(t, "autoconf", "dc1", testAC.originalToken, secondCA, &secondRoots, thirdCert, testAC.extraCerts) diff --git a/agent/auto-config/config.go b/agent/auto-config/config.go index a20121fb9aab..7c4c2d4f37d9 100644 --- a/agent/auto-config/config.go +++ b/agent/auto-config/config.go @@ -20,7 +20,7 @@ import ( // the agent gets configured, it can go through the normal RPC means of selecting a available // server automatically. type DirectRPC interface { - RPC(dc string, node string, addr net.Addr, method string, args interface{}, reply interface{}) error + RPC(dc string, node string, addr net.Addr, method string, args interface{}, reply interface{}, deadline time.Time) error } // Cache is an interface to represent the methods of the diff --git a/agent/auto-config/mock_test.go b/agent/auto-config/mock_test.go index 49d3ed29e246..1b1b3380391d 100644 --- a/agent/auto-config/mock_test.go +++ b/agent/auto-config/mock_test.go @@ -6,6 +6,7 @@ import ( "net" "sync" "testing" + "time" "github.com/stretchr/testify/mock" @@ -29,22 +30,22 @@ func newMockDirectRPC(t *testing.T) *mockDirectRPC { return &m } -func (m *mockDirectRPC) RPC(dc string, node string, addr net.Addr, method string, args interface{}, reply interface{}) error { +func (m *mockDirectRPC) RPC(dc string, node string, addr net.Addr, method string, args interface{}, reply interface{}, deadline time.Time) error { var retValues mock.Arguments if method == "AutoConfig.InitialConfiguration" { req := args.(*pbautoconf.AutoConfigRequest) csr := req.CSR req.CSR = "" - retValues = m.Called(dc, node, addr, method, args, reply) + retValues = m.Called(dc, node, addr, method, args, reply, deadline) req.CSR = csr } else if method == "AutoEncrypt.Sign" { req := args.(*structs.CASignRequest) csr := req.CSR req.CSR = "" - retValues = m.Called(dc, node, addr, method, args, reply) + retValues = m.Called(dc, node, addr, method, args, reply, deadline) req.CSR = csr } else { - retValues = m.Called(dc, node, addr, method, args, reply) + retValues = m.Called(dc, node, addr, method, args, reply, deadline) } return retValues.Error(0)