dd0e8eec14
* copyright headers for agent folder * Ignore test data files * fix proto files and remove headers in agent/uiserver folder * ignore deep-copy files
1871 lines
52 KiB
Go
1871 lines
52 KiB
Go
// Copyright (c) HashiCorp, Inc.
|
|
// SPDX-License-Identifier: MPL-2.0
|
|
|
|
package consul
|
|
|
|
import (
|
|
"bufio"
|
|
"bytes"
|
|
"context"
|
|
"crypto/x509"
|
|
"encoding/binary"
|
|
"errors"
|
|
"fmt"
|
|
"io"
|
|
"math"
|
|
"net"
|
|
"os"
|
|
"path/filepath"
|
|
"strings"
|
|
"sync"
|
|
"testing"
|
|
"time"
|
|
|
|
"github.com/hashicorp/go-hclog"
|
|
"github.com/hashicorp/go-memdb"
|
|
"github.com/hashicorp/raft"
|
|
"github.com/stretchr/testify/assert"
|
|
"github.com/stretchr/testify/require"
|
|
"google.golang.org/grpc"
|
|
|
|
"github.com/hashicorp/consul-net-rpc/go-msgpack/codec"
|
|
msgpackrpc "github.com/hashicorp/consul-net-rpc/net-rpc-msgpackrpc"
|
|
|
|
"github.com/hashicorp/consul/acl"
|
|
"github.com/hashicorp/consul/agent/connect"
|
|
"github.com/hashicorp/consul/agent/consul/rate"
|
|
rpcRate "github.com/hashicorp/consul/agent/consul/rate"
|
|
"github.com/hashicorp/consul/agent/consul/state"
|
|
agent_grpc "github.com/hashicorp/consul/agent/grpc-internal"
|
|
"github.com/hashicorp/consul/agent/pool"
|
|
"github.com/hashicorp/consul/agent/structs"
|
|
tokenStore "github.com/hashicorp/consul/agent/token"
|
|
"github.com/hashicorp/consul/api"
|
|
"github.com/hashicorp/consul/lib"
|
|
"github.com/hashicorp/consul/proto/private/pbsubscribe"
|
|
"github.com/hashicorp/consul/sdk/testutil"
|
|
"github.com/hashicorp/consul/sdk/testutil/retry"
|
|
"github.com/hashicorp/consul/testrpc"
|
|
"github.com/hashicorp/consul/tlsutil"
|
|
)
|
|
|
|
func TestRPC_NoLeader_Fail(t *testing.T) {
|
|
if testing.Short() {
|
|
t.Skip("too slow for testing.Short")
|
|
}
|
|
|
|
t.Parallel()
|
|
dir1, s1 := testServerWithConfig(t, func(c *Config) {
|
|
c.RPCHoldTimeout = 1 * time.Millisecond
|
|
})
|
|
defer os.RemoveAll(dir1)
|
|
defer s1.Shutdown()
|
|
codec := rpcClient(t, s1)
|
|
defer codec.Close()
|
|
|
|
arg := structs.RegisterRequest{
|
|
Datacenter: "dc1",
|
|
Node: "foo",
|
|
Address: "127.0.0.1",
|
|
}
|
|
var out struct{}
|
|
|
|
// Make sure we eventually fail with a no leader error, which we should
|
|
// see given the short timeout.
|
|
err := msgpackrpc.CallWithCodec(codec, "Catalog.Register", &arg, &out)
|
|
if err == nil || err.Error() != structs.ErrNoLeader.Error() {
|
|
t.Fatalf("bad: %v", err)
|
|
}
|
|
|
|
// Now make sure it goes through.
|
|
testrpc.WaitForTestAgent(t, s1.RPC, "dc1")
|
|
err = msgpackrpc.CallWithCodec(codec, "Catalog.Register", &arg, &out)
|
|
if err != nil {
|
|
t.Fatalf("bad: %v", err)
|
|
}
|
|
}
|
|
|
|
func TestRPC_NoLeader_Fail_on_stale_read(t *testing.T) {
|
|
if testing.Short() {
|
|
t.Skip("too slow for testing.Short")
|
|
}
|
|
|
|
t.Parallel()
|
|
dir1, s1 := testServerWithConfig(t, func(c *Config) {
|
|
c.RPCHoldTimeout = 1 * time.Millisecond
|
|
})
|
|
defer os.RemoveAll(dir1)
|
|
defer s1.Shutdown()
|
|
codec := rpcClient(t, s1)
|
|
defer codec.Close()
|
|
|
|
arg := structs.RegisterRequest{
|
|
Datacenter: "dc1",
|
|
Node: "foo",
|
|
Address: "127.0.0.1",
|
|
}
|
|
var out struct{}
|
|
|
|
// Make sure we eventually fail with a no leader error, which we should
|
|
// see given the short timeout.
|
|
err := msgpackrpc.CallWithCodec(codec, "Catalog.Register", &arg, &out)
|
|
if err == nil || err.Error() != structs.ErrNoLeader.Error() {
|
|
t.Fatalf("bad: %v", err)
|
|
}
|
|
|
|
// Until leader has never been known, stale should fail
|
|
getKeysReq := structs.KeyListRequest{
|
|
Datacenter: "dc1",
|
|
Prefix: "",
|
|
Seperator: "/",
|
|
QueryOptions: structs.QueryOptions{AllowStale: true},
|
|
}
|
|
var keyList structs.IndexedKeyList
|
|
if err := msgpackrpc.CallWithCodec(codec, "KVS.ListKeys", &getKeysReq, &keyList); err.Error() != structs.ErrNoLeader.Error() {
|
|
t.Fatalf("expected %v but got err: %v", structs.ErrNoLeader, err)
|
|
}
|
|
|
|
testrpc.WaitForTestAgent(t, s1.RPC, "dc1")
|
|
if err := msgpackrpc.CallWithCodec(codec, "KVS.ListKeys", &getKeysReq, &keyList); err != nil {
|
|
t.Fatalf("Did not expect any error but got err: %v", err)
|
|
}
|
|
}
|
|
|
|
func TestRPC_NoLeader_Retry(t *testing.T) {
|
|
if testing.Short() {
|
|
t.Skip("too slow for testing.Short")
|
|
}
|
|
|
|
t.Parallel()
|
|
dir1, s1 := testServerWithConfig(t, func(c *Config) {
|
|
c.RPCHoldTimeout = 10 * time.Second
|
|
})
|
|
defer os.RemoveAll(dir1)
|
|
defer s1.Shutdown()
|
|
codec := rpcClient(t, s1)
|
|
defer codec.Close()
|
|
|
|
arg := structs.RegisterRequest{
|
|
Datacenter: "dc1",
|
|
Node: "foo",
|
|
Address: "127.0.0.1",
|
|
}
|
|
var out struct{}
|
|
|
|
// This isn't sure-fire but tries to check that we don't have a
|
|
// leader going into the RPC, so we exercise the retry logic.
|
|
if ok, _, _ := s1.getLeader(); ok {
|
|
t.Fatalf("should not have a leader yet")
|
|
}
|
|
|
|
// The timeout is long enough to ride out any reasonable leader
|
|
// election.
|
|
err := msgpackrpc.CallWithCodec(codec, "Catalog.Register", &arg, &out)
|
|
if err != nil {
|
|
t.Fatalf("bad: %v", err)
|
|
}
|
|
}
|
|
|
|
func TestRPC_getLeader_ErrLeaderNotTracked(t *testing.T) {
|
|
if testing.Short() {
|
|
t.Skip("too slow for testing.Short")
|
|
}
|
|
|
|
cluster := newTestCluster(t, &testClusterConfig{
|
|
Datacenter: "dc1",
|
|
Servers: 3,
|
|
ServerWait: func(t *testing.T, srv *Server) {
|
|
// The test cluster waits for a leader to be established
|
|
// but not for all the RPC tracking of all servers to be updated
|
|
// so we also want to wait for that here
|
|
retry.Run(t, func(r *retry.R) {
|
|
if !srv.IsLeader() {
|
|
_, _, err := srv.getLeader()
|
|
require.NoError(r, err)
|
|
}
|
|
})
|
|
|
|
},
|
|
})
|
|
|
|
// At this point we know we have a cluster with a leader and all followers are tracking that
|
|
// leader in the serverLookup struct. We need to find a follower to hack its server lookup
|
|
// to force the error we desire
|
|
|
|
var follower *Server
|
|
for _, srv := range cluster.Servers {
|
|
if !srv.IsLeader() {
|
|
follower = srv
|
|
break
|
|
}
|
|
}
|
|
|
|
_, leaderMeta, err := follower.getLeader()
|
|
require.NoError(t, err)
|
|
|
|
// now do some behind the scenes trickery on the followers server lookup
|
|
// to remove the leader from it so that we can force a ErrLeaderNotTracked error
|
|
follower.serverLookup.RemoveServer(leaderMeta)
|
|
|
|
isLeader, meta, err := follower.getLeader()
|
|
require.Error(t, err)
|
|
require.True(t, errors.Is(err, structs.ErrLeaderNotTracked))
|
|
require.Nil(t, meta)
|
|
require.False(t, isLeader)
|
|
}
|
|
|
|
type MockSink struct {
|
|
*bytes.Buffer
|
|
cancel bool
|
|
}
|
|
|
|
func (m *MockSink) ID() string {
|
|
return "Mock"
|
|
}
|
|
|
|
func (m *MockSink) Cancel() error {
|
|
m.cancel = true
|
|
return nil
|
|
}
|
|
|
|
func (m *MockSink) Close() error {
|
|
return nil
|
|
}
|
|
|
|
func TestServer_blockingQuery(t *testing.T) {
|
|
t.Parallel()
|
|
_, s := testServerWithConfig(t)
|
|
|
|
// Perform a non-blocking query. Note that it's significant that the meta has
|
|
// a zero index in response - the implied opts.MinQueryIndex is also zero but
|
|
// this should not block still.
|
|
t.Run("non-blocking query", func(t *testing.T) {
|
|
var opts structs.QueryOptions
|
|
var meta structs.QueryMeta
|
|
var calls int
|
|
fn := func(_ memdb.WatchSet, _ *state.Store) error {
|
|
calls++
|
|
return nil
|
|
}
|
|
err := s.blockingQuery(&opts, &meta, fn)
|
|
require.NoError(t, err)
|
|
require.Equal(t, 1, calls)
|
|
})
|
|
|
|
// Perform a blocking query that gets woken up and loops around once.
|
|
t.Run("blocking query - single loop", func(t *testing.T) {
|
|
opts := structs.QueryOptions{
|
|
MinQueryIndex: 3,
|
|
}
|
|
var meta structs.QueryMeta
|
|
var calls int
|
|
fn := func(ws memdb.WatchSet, _ *state.Store) error {
|
|
if calls == 0 {
|
|
meta.Index = 3
|
|
|
|
fakeCh := make(chan struct{})
|
|
close(fakeCh)
|
|
ws.Add(fakeCh)
|
|
} else {
|
|
meta.Index = 4
|
|
}
|
|
calls++
|
|
return nil
|
|
}
|
|
err := s.blockingQuery(&opts, &meta, fn)
|
|
require.NoError(t, err)
|
|
require.Equal(t, 2, calls)
|
|
})
|
|
|
|
// Perform a blocking query that returns a zero index from blocking func (e.g.
|
|
// no state yet). This should still return an empty response immediately, but
|
|
// with index of 1 and then block on the next attempt. In one sense zero index
|
|
// is not really a valid response from a state method that is not an error but
|
|
// in practice a lot of state store operations do return it unless they
|
|
// explicitly special checks to turn 0 into 1. Often this is not caught or
|
|
// covered by tests but eventually when hit in the wild causes blocking
|
|
// clients to busy loop and burn CPU. This test ensure that blockingQuery
|
|
// systematically does the right thing to prevent future bugs like that.
|
|
t.Run("blocking query with 0 modifyIndex from state func", func(t *testing.T) {
|
|
opts := structs.QueryOptions{
|
|
MinQueryIndex: 0,
|
|
}
|
|
var meta structs.QueryMeta
|
|
var calls int
|
|
fn := func(ws memdb.WatchSet, _ *state.Store) error {
|
|
if opts.MinQueryIndex > 0 {
|
|
// If client requested blocking, block forever. This is simulating
|
|
// waiting for the watched resource to be initialized/written to giving
|
|
// it a non-zero index. Note the timeout on the query options is relied
|
|
// on to stop the test taking forever.
|
|
fakeCh := make(chan struct{})
|
|
ws.Add(fakeCh)
|
|
}
|
|
meta.Index = 0
|
|
calls++
|
|
return nil
|
|
}
|
|
require.NoError(t, s.blockingQuery(&opts, &meta, fn))
|
|
assert.Equal(t, 1, calls)
|
|
assert.Equal(t, uint64(1), meta.Index,
|
|
"expect fake index of 1 to force client to block on next update")
|
|
|
|
// Simulate client making next request
|
|
opts.MinQueryIndex = 1
|
|
opts.MaxQueryTime = 20 * time.Millisecond // Don't wait too long
|
|
|
|
// This time we should block even though the func returns index 0 still
|
|
t0 := time.Now()
|
|
require.NoError(t, s.blockingQuery(&opts, &meta, fn))
|
|
t1 := time.Now()
|
|
assert.Equal(t, 2, calls)
|
|
assert.Equal(t, uint64(1), meta.Index,
|
|
"expect fake index of 1 to force client to block on next update")
|
|
assert.True(t, t1.Sub(t0) > 20*time.Millisecond,
|
|
"should have actually blocked waiting for timeout")
|
|
|
|
})
|
|
|
|
// Perform a query that blocks and gets interrupted when the state store
|
|
// is abandoned.
|
|
t.Run("blocking query interrupted by abandonCh", func(t *testing.T) {
|
|
opts := structs.QueryOptions{
|
|
MinQueryIndex: 3,
|
|
}
|
|
var meta structs.QueryMeta
|
|
var calls int
|
|
fn := func(_ memdb.WatchSet, _ *state.Store) error {
|
|
if calls == 0 {
|
|
meta.Index = 3
|
|
|
|
snap, err := s.fsm.Snapshot()
|
|
if err != nil {
|
|
t.Fatalf("err: %v", err)
|
|
}
|
|
defer snap.Release()
|
|
|
|
buf := bytes.NewBuffer(nil)
|
|
sink := &MockSink{buf, false}
|
|
if err := snap.Persist(sink); err != nil {
|
|
t.Fatalf("err: %v", err)
|
|
}
|
|
|
|
if err := s.fsm.Restore(sink); err != nil {
|
|
t.Fatalf("err: %v", err)
|
|
}
|
|
}
|
|
calls++
|
|
return nil
|
|
}
|
|
err := s.blockingQuery(&opts, &meta, fn)
|
|
require.NoError(t, err)
|
|
require.Equal(t, 1, calls)
|
|
})
|
|
|
|
t.Run("ResultsFilteredByACLs is reset for unauthenticated calls", func(t *testing.T) {
|
|
opts := structs.QueryOptions{
|
|
Token: "",
|
|
}
|
|
var meta structs.QueryMeta
|
|
fn := func(_ memdb.WatchSet, _ *state.Store) error {
|
|
meta.ResultsFilteredByACLs = true
|
|
return nil
|
|
}
|
|
|
|
err := s.blockingQuery(&opts, &meta, fn)
|
|
require.NoError(t, err)
|
|
require.False(t, meta.ResultsFilteredByACLs, "ResultsFilteredByACLs should be reset for unauthenticated calls")
|
|
})
|
|
|
|
t.Run("ResultsFilteredByACLs is honored for authenticated calls", func(t *testing.T) {
|
|
token, err := lib.GenerateUUID(nil)
|
|
require.NoError(t, err)
|
|
|
|
opts := structs.QueryOptions{
|
|
Token: token,
|
|
}
|
|
var meta structs.QueryMeta
|
|
fn := func(_ memdb.WatchSet, _ *state.Store) error {
|
|
meta.ResultsFilteredByACLs = true
|
|
return nil
|
|
}
|
|
|
|
err = s.blockingQuery(&opts, &meta, fn)
|
|
require.NoError(t, err)
|
|
require.True(t, meta.ResultsFilteredByACLs, "ResultsFilteredByACLs should be honored for authenticated calls")
|
|
})
|
|
|
|
t.Run("non-blocking query for item that does not exist", func(t *testing.T) {
|
|
opts := structs.QueryOptions{}
|
|
meta := structs.QueryMeta{}
|
|
calls := 0
|
|
fn := func(_ memdb.WatchSet, _ *state.Store) error {
|
|
calls++
|
|
return errNotFound
|
|
}
|
|
|
|
err := s.blockingQuery(&opts, &meta, fn)
|
|
require.NoError(t, err)
|
|
require.Equal(t, 1, calls)
|
|
})
|
|
|
|
t.Run("blocking query for item that does not exist", func(t *testing.T) {
|
|
opts := structs.QueryOptions{MinQueryIndex: 3, MaxQueryTime: 100 * time.Millisecond}
|
|
meta := structs.QueryMeta{}
|
|
calls := 0
|
|
fn := func(ws memdb.WatchSet, _ *state.Store) error {
|
|
calls++
|
|
if calls == 1 {
|
|
meta.Index = 3
|
|
|
|
ch := make(chan struct{})
|
|
close(ch)
|
|
ws.Add(ch)
|
|
return errNotFound
|
|
}
|
|
meta.Index = 5
|
|
return errNotFound
|
|
}
|
|
|
|
err := s.blockingQuery(&opts, &meta, fn)
|
|
require.NoError(t, err)
|
|
require.Equal(t, 2, calls)
|
|
})
|
|
|
|
t.Run("blocking query for item that existed and is removed", func(t *testing.T) {
|
|
opts := structs.QueryOptions{MinQueryIndex: 3, MaxQueryTime: 100 * time.Millisecond}
|
|
meta := structs.QueryMeta{}
|
|
calls := 0
|
|
fn := func(ws memdb.WatchSet, _ *state.Store) error {
|
|
calls++
|
|
if calls == 1 {
|
|
meta.Index = 3
|
|
|
|
ch := make(chan struct{})
|
|
close(ch)
|
|
ws.Add(ch)
|
|
return nil
|
|
}
|
|
meta.Index = 5
|
|
return errNotFound
|
|
}
|
|
|
|
start := time.Now()
|
|
err := s.blockingQuery(&opts, &meta, fn)
|
|
require.True(t, time.Since(start) < opts.MaxQueryTime, "query timed out")
|
|
require.NoError(t, err)
|
|
require.Equal(t, 2, calls)
|
|
})
|
|
|
|
t.Run("blocking query for non-existent item that is created", func(t *testing.T) {
|
|
opts := structs.QueryOptions{MinQueryIndex: 3, MaxQueryTime: 100 * time.Millisecond}
|
|
meta := structs.QueryMeta{}
|
|
calls := 0
|
|
fn := func(ws memdb.WatchSet, _ *state.Store) error {
|
|
calls++
|
|
if calls == 1 {
|
|
meta.Index = 3
|
|
|
|
ch := make(chan struct{})
|
|
close(ch)
|
|
ws.Add(ch)
|
|
return errNotFound
|
|
}
|
|
meta.Index = 5
|
|
return nil
|
|
}
|
|
|
|
start := time.Now()
|
|
err := s.blockingQuery(&opts, &meta, fn)
|
|
require.True(t, time.Since(start) < opts.MaxQueryTime, "query timed out")
|
|
require.NoError(t, err)
|
|
require.Equal(t, 2, calls)
|
|
})
|
|
}
|
|
|
|
func TestRPC_ReadyForConsistentReads(t *testing.T) {
|
|
if testing.Short() {
|
|
t.Skip("too slow for testing.Short")
|
|
}
|
|
|
|
t.Parallel()
|
|
dir, s := testServerWithConfig(t, func(c *Config) {
|
|
c.RPCHoldTimeout = 2 * time.Millisecond
|
|
})
|
|
defer os.RemoveAll(dir)
|
|
defer s.Shutdown()
|
|
|
|
testrpc.WaitForLeader(t, s.RPC, "dc1")
|
|
|
|
if !s.isReadyForConsistentReads() {
|
|
t.Fatal("Server should be ready for consistent reads")
|
|
}
|
|
|
|
s.resetConsistentReadReady()
|
|
err := s.consistentRead()
|
|
if err.Error() != "Not ready to serve consistent reads" {
|
|
t.Fatal("Server should NOT be ready for consistent reads")
|
|
}
|
|
|
|
go func() {
|
|
time.Sleep(100 * time.Millisecond)
|
|
s.setConsistentReadReady()
|
|
}()
|
|
|
|
retry.Run(t, func(r *retry.R) {
|
|
if err := s.consistentRead(); err != nil {
|
|
r.Fatalf("Expected server to be ready for consistent reads, got error %v", err)
|
|
}
|
|
})
|
|
}
|
|
|
|
func TestRPC_MagicByteTimeout(t *testing.T) {
|
|
if testing.Short() {
|
|
t.Skip("too slow for testing.Short")
|
|
}
|
|
|
|
t.Parallel()
|
|
dir1, s1 := testServerWithConfig(t, func(c *Config) {
|
|
c.RPCHandshakeTimeout = 10 * time.Millisecond
|
|
})
|
|
defer os.RemoveAll(dir1)
|
|
defer s1.Shutdown()
|
|
|
|
// Connect to the server with bare TCP to simulate a malicious client trying
|
|
// to hold open resources.
|
|
addr := s1.config.RPCAdvertise
|
|
conn, err := net.DialTimeout("tcp", addr.String(), time.Second)
|
|
require.NoError(t, err)
|
|
defer conn.Close()
|
|
|
|
// Wait for more than the timeout. This is timing dependent so could fail if
|
|
// the CPU is super overloaded so the handler goroutine so I'm using a retry
|
|
// loop below to be sure but this feels like a pretty generous margin for
|
|
// error (10x the timeout and 100ms of scheduling time).
|
|
time.Sleep(100 * time.Millisecond)
|
|
|
|
// Set a read deadline on the Conn in case the timeout is not working we don't
|
|
// want the read below to block forever. Needs to be much longer than what we
|
|
// expect and the error should be different too.
|
|
conn.SetReadDeadline(time.Now().Add(3 * time.Second))
|
|
|
|
retry.Run(t, func(r *retry.R) {
|
|
// Sanity check the conn was closed by attempting to read from it (a write
|
|
// might not detect the close).
|
|
buf := make([]byte, 10)
|
|
_, err = conn.Read(buf)
|
|
require.Error(r, err)
|
|
require.Contains(r, err.Error(), "EOF")
|
|
})
|
|
}
|
|
|
|
func TestRPC_TLSHandshakeTimeout(t *testing.T) {
|
|
// if this test is failing because of expired certificates
|
|
// use the procedure in test/CA-GENERATION.md
|
|
if testing.Short() {
|
|
t.Skip("too slow for testing.Short")
|
|
}
|
|
|
|
t.Parallel()
|
|
|
|
dir1, s1 := testServerWithConfig(t, func(c *Config) {
|
|
c.RPCHandshakeTimeout = 10 * time.Millisecond
|
|
c.TLSConfig.InternalRPC.CAFile = "../../test/hostname/CertAuth.crt"
|
|
c.TLSConfig.InternalRPC.CertFile = "../../test/hostname/Alice.crt"
|
|
c.TLSConfig.InternalRPC.KeyFile = "../../test/hostname/Alice.key"
|
|
c.TLSConfig.InternalRPC.VerifyServerHostname = true
|
|
c.TLSConfig.InternalRPC.VerifyOutgoing = true
|
|
c.TLSConfig.InternalRPC.VerifyIncoming = true
|
|
})
|
|
defer os.RemoveAll(dir1)
|
|
defer s1.Shutdown()
|
|
|
|
// Connect to the server with TLS magic byte delivered on time
|
|
addr := s1.config.RPCAdvertise
|
|
conn, err := net.DialTimeout("tcp", addr.String(), time.Second)
|
|
require.NoError(t, err)
|
|
defer conn.Close()
|
|
|
|
// Write TLS byte to avoid being closed by either the (outer) first byte
|
|
// timeout or the fact that server requires TLS
|
|
_, err = conn.Write([]byte{byte(pool.RPCTLS)})
|
|
require.NoError(t, err)
|
|
|
|
// Wait for more than the timeout before we start a TLS handshake. This is
|
|
// timing dependent so could fail if the CPU is super overloaded so the
|
|
// handler goroutine so I'm using a retry loop below to be sure but this feels
|
|
// like a pretty generous margin for error (10x the timeout and 100ms of
|
|
// scheduling time).
|
|
time.Sleep(100 * time.Millisecond)
|
|
|
|
// Set a read deadline on the Conn in case the timeout is not working we don't
|
|
// want the read below to block forever. Needs to be much longer than what we
|
|
// expect and the error should be different too.
|
|
conn.SetReadDeadline(time.Now().Add(3 * time.Second))
|
|
|
|
retry.Run(t, func(r *retry.R) {
|
|
// Sanity check the conn was closed by attempting to read from it (a write
|
|
// might not detect the close).
|
|
buf := make([]byte, 10)
|
|
_, err = conn.Read(buf)
|
|
require.Error(r, err)
|
|
require.Contains(r, err.Error(), "EOF")
|
|
})
|
|
}
|
|
|
|
func TestRPC_PreventsTLSNesting(t *testing.T) {
|
|
if testing.Short() {
|
|
t.Skip("too slow for testing.Short")
|
|
}
|
|
|
|
t.Parallel()
|
|
|
|
cases := []struct {
|
|
name string
|
|
outerByte pool.RPCType
|
|
innerByte pool.RPCType
|
|
wantClose bool
|
|
}{
|
|
{
|
|
// Base case, sanity check normal RPC in TLS works
|
|
name: "RPC in TLS",
|
|
outerByte: pool.RPCTLS,
|
|
innerByte: pool.RPCConsul,
|
|
wantClose: false,
|
|
},
|
|
{
|
|
// Nested TLS-in-TLS
|
|
name: "TLS in TLS",
|
|
outerByte: pool.RPCTLS,
|
|
innerByte: pool.RPCTLS,
|
|
wantClose: true,
|
|
},
|
|
{
|
|
// Nested TLS-in-TLS
|
|
name: "TLS in Insecure TLS",
|
|
outerByte: pool.RPCTLSInsecure,
|
|
innerByte: pool.RPCTLS,
|
|
wantClose: true,
|
|
},
|
|
{
|
|
// Nested TLS-in-TLS
|
|
name: "Insecure TLS in TLS",
|
|
outerByte: pool.RPCTLS,
|
|
innerByte: pool.RPCTLSInsecure,
|
|
wantClose: true,
|
|
},
|
|
{
|
|
// Nested TLS-in-TLS
|
|
name: "Insecure TLS in Insecure TLS",
|
|
outerByte: pool.RPCTLSInsecure,
|
|
innerByte: pool.RPCTLSInsecure,
|
|
wantClose: true,
|
|
},
|
|
}
|
|
|
|
for _, tc := range cases {
|
|
t.Run(tc.name, func(t *testing.T) {
|
|
dir1, s1 := testServerWithConfig(t, func(c *Config) {
|
|
c.TLSConfig.InternalRPC.CAFile = "../../test/hostname/CertAuth.crt"
|
|
c.TLSConfig.InternalRPC.CertFile = "../../test/hostname/Alice.crt"
|
|
c.TLSConfig.InternalRPC.KeyFile = "../../test/hostname/Alice.key"
|
|
c.TLSConfig.InternalRPC.VerifyServerHostname = true
|
|
c.TLSConfig.InternalRPC.VerifyOutgoing = true
|
|
c.TLSConfig.InternalRPC.VerifyIncoming = false // saves us getting client cert setup
|
|
c.TLSConfig.Domain = "consul"
|
|
})
|
|
defer os.RemoveAll(dir1)
|
|
defer s1.Shutdown()
|
|
|
|
// Connect to the server with TLS magic byte delivered on time
|
|
addr := s1.config.RPCAdvertise
|
|
conn, err := net.DialTimeout("tcp", addr.String(), time.Second)
|
|
require.NoError(t, err)
|
|
defer conn.Close()
|
|
|
|
// Write Outer magic byte
|
|
_, err = conn.Write([]byte{byte(tc.outerByte)})
|
|
require.NoError(t, err)
|
|
|
|
// Start tls client
|
|
tlsWrap := s1.tlsConfigurator.OutgoingRPCWrapper()
|
|
tlsConn, err := tlsWrap("dc1", conn)
|
|
require.NoError(t, err)
|
|
|
|
// Write Inner magic byte
|
|
_, err = tlsConn.Write([]byte{byte(tc.innerByte)})
|
|
require.NoError(t, err)
|
|
|
|
if tc.wantClose {
|
|
// Allow up to a second for a read failure to indicate conn was closed by
|
|
// server.
|
|
conn.SetReadDeadline(time.Now().Add(1 * time.Second))
|
|
|
|
retry.Run(t, func(r *retry.R) {
|
|
// Sanity check the conn was closed by attempting to read from it (a
|
|
// write might not detect the close).
|
|
buf := make([]byte, 10)
|
|
_, err = tlsConn.Read(buf)
|
|
require.Error(r, err)
|
|
require.Contains(r, err.Error(), "EOF")
|
|
})
|
|
} else {
|
|
// Set a shorter read deadline that should typically be enough to detect
|
|
// immediate close but will also not make test hang forever. This
|
|
// positive case is mostly just a sanity check that the test code here
|
|
// is actually not failing just due to some other error in the way we
|
|
// setup TLS. It also sanity checks that we still allow valid TLS conns
|
|
// but if it produces possible false-positives in CI sometimes that's
|
|
// not such a huge deal - CI won't be brittle and it will have done it's
|
|
// job as a sanity check most of the time.
|
|
conn.SetReadDeadline(time.Now().Add(50 * time.Millisecond))
|
|
buf := make([]byte, 10)
|
|
_, err = tlsConn.Read(buf)
|
|
require.Error(t, err)
|
|
require.Contains(t, err.Error(), "i/o timeout")
|
|
}
|
|
})
|
|
}
|
|
}
|
|
|
|
func connectClient(t *testing.T, s1 *Server, mb pool.RPCType, useTLS, wantOpen bool, message string) net.Conn {
|
|
t.Helper()
|
|
|
|
addr := s1.config.RPCAdvertise
|
|
tlsWrap := s1.tlsConfigurator.OutgoingRPCWrapper()
|
|
|
|
conn, err := net.DialTimeout("tcp", addr.String(), time.Second)
|
|
require.NoError(t, err)
|
|
|
|
// Write magic byte so we aren't timed out
|
|
outerByte := mb
|
|
if useTLS {
|
|
outerByte = pool.RPCTLS
|
|
}
|
|
_, err = conn.Write([]byte{byte(outerByte)})
|
|
require.NoError(t, err)
|
|
|
|
if useTLS {
|
|
tlsConn, err := tlsWrap(s1.config.Datacenter, conn)
|
|
// Subtly, tlsWrap will NOT actually do a handshake in this case - it only
|
|
// does so for some configs, so even if the server closed the conn before
|
|
// handshake this won't fail and it's only when we attempt to read or write
|
|
// that we'll see the broken pipe.
|
|
require.NoError(t, err, "%s: wanted open conn, failed TLS handshake: %s",
|
|
message, err)
|
|
conn = tlsConn
|
|
|
|
// Write Inner magic byte
|
|
_, err = conn.Write([]byte{byte(mb)})
|
|
if !wantOpen {
|
|
// TLS Handshake will be done on this attempt to write and should fail
|
|
require.Error(t, err, "%s: wanted closed conn, TLS Handshake succeeded", message)
|
|
} else {
|
|
require.NoError(t, err, "%s: wanted open conn, failed writing inner magic byte: %s",
|
|
message, err)
|
|
}
|
|
}
|
|
|
|
// Check if the conn is in the state we want.
|
|
retry.Run(t, func(r *retry.R) {
|
|
// Don't wait around as server won't be sending data but the read will fail
|
|
// immediately if the conn is closed.
|
|
conn.SetReadDeadline(time.Now().Add(1 * time.Millisecond))
|
|
buf := make([]byte, 10)
|
|
_, err := conn.Read(buf)
|
|
require.Error(r, err)
|
|
if wantOpen {
|
|
require.Contains(r, err.Error(), "i/o timeout",
|
|
"%s: wanted an open conn (read timeout)", message)
|
|
} else {
|
|
if useTLS {
|
|
require.Error(r, err)
|
|
// TLS may fail during either read or write of the handshake so there
|
|
// are a few different errors that come up.
|
|
if !strings.Contains(err.Error(), "read: connection reset by peer") &&
|
|
!strings.Contains(err.Error(), "write: connection reset by peer") &&
|
|
!strings.Contains(err.Error(), "write: broken pipe") {
|
|
r.Fatalf("%s: wanted closed conn got err: %s", message, err)
|
|
}
|
|
} else {
|
|
require.Contains(r, err.Error(), "EOF", "%s: wanted a closed conn",
|
|
message)
|
|
}
|
|
}
|
|
})
|
|
|
|
return conn
|
|
}
|
|
|
|
func TestRPC_RPCMaxConnsPerClient(t *testing.T) {
|
|
// if this test is failing because of expired certificates
|
|
// use the procedure in test/CA-GENERATION.md
|
|
if testing.Short() {
|
|
t.Skip("too slow for testing.Short")
|
|
}
|
|
|
|
t.Parallel()
|
|
|
|
cases := []struct {
|
|
name string
|
|
magicByte pool.RPCType
|
|
tlsEnabled bool
|
|
}{
|
|
{"RPC v2", pool.RPCMultiplexV2, false},
|
|
{"RPC v2 TLS", pool.RPCMultiplexV2, true},
|
|
{"RPC", pool.RPCConsul, false},
|
|
{"RPC TLS", pool.RPCConsul, true},
|
|
}
|
|
|
|
for _, tc := range cases {
|
|
tc := tc
|
|
t.Run(tc.name, func(t *testing.T) {
|
|
dir1, s1 := testServerWithConfig(t, func(c *Config) {
|
|
// we have to set this to 3 because autopilot is going to keep a connection open
|
|
c.RPCMaxConnsPerClient = 3
|
|
if tc.tlsEnabled {
|
|
c.TLSConfig.InternalRPC.CAFile = "../../test/hostname/CertAuth.crt"
|
|
c.TLSConfig.InternalRPC.CertFile = "../../test/hostname/Alice.crt"
|
|
c.TLSConfig.InternalRPC.KeyFile = "../../test/hostname/Alice.key"
|
|
c.TLSConfig.InternalRPC.VerifyServerHostname = true
|
|
c.TLSConfig.InternalRPC.VerifyOutgoing = true
|
|
c.TLSConfig.InternalRPC.VerifyIncoming = false // saves us getting client cert setup
|
|
c.TLSConfig.Domain = "consul"
|
|
}
|
|
})
|
|
defer os.RemoveAll(dir1)
|
|
defer s1.Shutdown()
|
|
|
|
waitForLeaderEstablishment(t, s1)
|
|
|
|
// Connect to the server with bare TCP
|
|
conn1 := connectClient(t, s1, tc.magicByte, tc.tlsEnabled, true, "conn1")
|
|
defer conn1.Close()
|
|
|
|
// Two conns should succeed
|
|
conn2 := connectClient(t, s1, tc.magicByte, tc.tlsEnabled, true, "conn2")
|
|
defer conn2.Close()
|
|
|
|
// Third should be closed byt the limiter
|
|
conn3 := connectClient(t, s1, tc.magicByte, tc.tlsEnabled, false, "conn3")
|
|
defer conn3.Close()
|
|
|
|
// If we close one of the earlier ones, we should be able to open another
|
|
addr := conn1.RemoteAddr()
|
|
conn1.Close()
|
|
retry.Run(t, func(r *retry.R) {
|
|
if n := s1.rpcConnLimiter.NumOpen(addr); n >= 3 {
|
|
r.Fatal("waiting for open conns to drop")
|
|
}
|
|
})
|
|
conn4 := connectClient(t, s1, tc.magicByte, tc.tlsEnabled, true, "conn4")
|
|
defer conn4.Close()
|
|
|
|
// Reload config with higher limit
|
|
rc := ReloadableConfig{
|
|
RPCRateLimit: s1.config.RPCRateLimit,
|
|
RPCMaxBurst: s1.config.RPCMaxBurst,
|
|
RPCMaxConnsPerClient: 10,
|
|
}
|
|
require.NoError(t, s1.ReloadConfig(rc))
|
|
|
|
// Now another conn should be allowed
|
|
conn5 := connectClient(t, s1, tc.magicByte, tc.tlsEnabled, true, "conn5")
|
|
defer conn5.Close()
|
|
})
|
|
}
|
|
}
|
|
|
|
func TestRPC_readUint32(t *testing.T) {
|
|
cases := []struct {
|
|
name string
|
|
writeFn func(net.Conn)
|
|
readFn func(*testing.T, net.Conn)
|
|
}{
|
|
{
|
|
name: "timeouts irrelevant",
|
|
writeFn: func(conn net.Conn) {
|
|
_ = binary.Write(conn, binary.BigEndian, uint32(42))
|
|
_ = binary.Write(conn, binary.BigEndian, uint32(math.MaxUint32))
|
|
_ = binary.Write(conn, binary.BigEndian, uint32(1))
|
|
},
|
|
readFn: func(t *testing.T, conn net.Conn) {
|
|
t.Helper()
|
|
v, err := readUint32(conn, 5*time.Second)
|
|
require.NoError(t, err)
|
|
require.Equal(t, uint32(42), v)
|
|
|
|
v, err = readUint32(conn, 5*time.Second)
|
|
require.NoError(t, err)
|
|
require.Equal(t, uint32(math.MaxUint32), v)
|
|
|
|
v, err = readUint32(conn, 5*time.Second)
|
|
require.NoError(t, err)
|
|
require.Equal(t, uint32(1), v)
|
|
},
|
|
},
|
|
{
|
|
name: "triggers timeout on last read",
|
|
writeFn: func(conn net.Conn) {
|
|
_ = binary.Write(conn, binary.BigEndian, uint32(42))
|
|
_ = binary.Write(conn, binary.BigEndian, uint32(math.MaxUint32))
|
|
_ = binary.Write(conn, binary.BigEndian, uint16(1)) // half as many bytes as expected
|
|
},
|
|
readFn: func(t *testing.T, conn net.Conn) {
|
|
t.Helper()
|
|
v, err := readUint32(conn, 5*time.Second)
|
|
require.NoError(t, err)
|
|
require.Equal(t, uint32(42), v)
|
|
|
|
v, err = readUint32(conn, 5*time.Second)
|
|
require.NoError(t, err)
|
|
require.Equal(t, uint32(math.MaxUint32), v)
|
|
|
|
_, err = readUint32(conn, 50*time.Millisecond)
|
|
require.Error(t, err)
|
|
nerr, ok := err.(net.Error)
|
|
require.True(t, ok)
|
|
require.True(t, nerr.Timeout())
|
|
},
|
|
},
|
|
}
|
|
|
|
for _, tc := range cases {
|
|
tc := tc
|
|
t.Run(tc.name, func(t *testing.T) {
|
|
var doneWg sync.WaitGroup
|
|
defer doneWg.Wait()
|
|
|
|
client, server := net.Pipe()
|
|
defer client.Close()
|
|
defer server.Close()
|
|
|
|
// Client pushes some data.
|
|
doneWg.Add(1)
|
|
go func() {
|
|
doneWg.Done()
|
|
tc.writeFn(client)
|
|
}()
|
|
|
|
// The server tests the function for us.
|
|
tc.readFn(t, server)
|
|
})
|
|
}
|
|
}
|
|
|
|
func TestRPC_LocalTokenStrippedOnForward(t *testing.T) {
|
|
if testing.Short() {
|
|
t.Skip("too slow for testing.Short")
|
|
}
|
|
|
|
t.Parallel()
|
|
dir1, s1 := testServerWithConfig(t, func(c *Config) {
|
|
c.PrimaryDatacenter = "dc1"
|
|
c.ACLsEnabled = true
|
|
c.ACLResolverSettings.ACLDefaultPolicy = "deny"
|
|
c.ACLInitialManagementToken = "root"
|
|
})
|
|
defer os.RemoveAll(dir1)
|
|
defer s1.Shutdown()
|
|
testrpc.WaitForLeader(t, s1.RPC, "dc1")
|
|
codec := rpcClient(t, s1)
|
|
defer codec.Close()
|
|
|
|
dir2, s2 := testServerWithConfig(t, func(c *Config) {
|
|
c.Datacenter = "dc2"
|
|
c.PrimaryDatacenter = "dc1"
|
|
c.ACLsEnabled = true
|
|
c.ACLResolverSettings.ACLDefaultPolicy = "deny"
|
|
c.ACLTokenReplication = true
|
|
c.ACLReplicationRate = 100
|
|
c.ACLReplicationBurst = 100
|
|
c.ACLReplicationApplyLimit = 1000000
|
|
})
|
|
s2.tokens.UpdateReplicationToken("root", tokenStore.TokenSourceConfig)
|
|
testrpc.WaitForLeader(t, s2.RPC, "dc2")
|
|
defer os.RemoveAll(dir2)
|
|
defer s2.Shutdown()
|
|
codec2 := rpcClient(t, s2)
|
|
defer codec2.Close()
|
|
|
|
// Try to join.
|
|
joinWAN(t, s2, s1)
|
|
testrpc.WaitForLeader(t, s1.RPC, "dc1")
|
|
testrpc.WaitForLeader(t, s1.RPC, "dc2")
|
|
waitForNewACLReplication(t, s2, structs.ACLReplicateTokens, 1, 1, 0)
|
|
|
|
// create simple kv policy
|
|
kvPolicy, err := upsertTestPolicyWithRules(codec, "root", "dc1", `
|
|
key_prefix "" { policy = "write" }
|
|
`)
|
|
require.NoError(t, err)
|
|
|
|
// Wait for it to replicate
|
|
retry.Run(t, func(r *retry.R) {
|
|
_, p, err := s2.fsm.State().ACLPolicyGetByID(nil, kvPolicy.ID, &acl.EnterpriseMeta{})
|
|
require.Nil(r, err)
|
|
require.NotNil(r, p)
|
|
})
|
|
|
|
// create local token that only works in DC2
|
|
localToken2, err := upsertTestToken(codec, "root", "dc2", func(token *structs.ACLToken) {
|
|
token.Local = true
|
|
token.Policies = []structs.ACLTokenPolicyLink{
|
|
{ID: kvPolicy.ID},
|
|
}
|
|
})
|
|
require.NoError(t, err)
|
|
|
|
// Try to use it locally (it should work)
|
|
arg := structs.KVSRequest{
|
|
Datacenter: "dc2",
|
|
Op: api.KVSet,
|
|
DirEnt: structs.DirEntry{
|
|
Key: "foo",
|
|
Value: []byte("bar"),
|
|
},
|
|
WriteRequest: structs.WriteRequest{Token: localToken2.SecretID},
|
|
}
|
|
var out bool
|
|
err = msgpackrpc.CallWithCodec(codec2, "KVS.Apply", &arg, &out)
|
|
require.NoError(t, err)
|
|
require.Equal(t, localToken2.SecretID, arg.WriteRequest.Token, "token should not be stripped")
|
|
|
|
// Try to use it remotely
|
|
arg = structs.KVSRequest{
|
|
Datacenter: "dc1",
|
|
Op: api.KVSet,
|
|
DirEnt: structs.DirEntry{
|
|
Key: "foo",
|
|
Value: []byte("bar"),
|
|
},
|
|
WriteRequest: structs.WriteRequest{Token: localToken2.SecretID},
|
|
}
|
|
err = msgpackrpc.CallWithCodec(codec2, "KVS.Apply", &arg, &out)
|
|
if !acl.IsErrPermissionDenied(err) {
|
|
t.Fatalf("err: %v", err)
|
|
}
|
|
|
|
// Update the anon token to also be able to write to kv
|
|
{
|
|
tokenUpsertReq := structs.ACLTokenSetRequest{
|
|
Datacenter: "dc1",
|
|
ACLToken: structs.ACLToken{
|
|
AccessorID: acl.AnonymousTokenID,
|
|
Policies: []structs.ACLTokenPolicyLink{
|
|
{
|
|
ID: kvPolicy.ID,
|
|
},
|
|
},
|
|
},
|
|
WriteRequest: structs.WriteRequest{Token: "root"},
|
|
}
|
|
token := structs.ACLToken{}
|
|
err = msgpackrpc.CallWithCodec(codec, "ACL.TokenSet", &tokenUpsertReq, &token)
|
|
require.NoError(t, err)
|
|
require.NotEmpty(t, token.SecretID)
|
|
}
|
|
|
|
// Try to use it remotely again, but this time it should fallback to anon
|
|
arg = structs.KVSRequest{
|
|
Datacenter: "dc1",
|
|
Op: api.KVSet,
|
|
DirEnt: structs.DirEntry{
|
|
Key: "foo",
|
|
Value: []byte("bar"),
|
|
},
|
|
WriteRequest: structs.WriteRequest{Token: localToken2.SecretID},
|
|
}
|
|
err = msgpackrpc.CallWithCodec(codec2, "KVS.Apply", &arg, &out)
|
|
require.NoError(t, err)
|
|
require.Equal(t, localToken2.SecretID, arg.WriteRequest.Token, "token should not be stripped")
|
|
}
|
|
|
|
func TestRPC_LocalTokenStrippedOnForward_GRPC(t *testing.T) {
|
|
if testing.Short() {
|
|
t.Skip("too slow for testing.Short")
|
|
}
|
|
|
|
t.Parallel()
|
|
dir1, s1 := testServerWithConfig(t, func(c *Config) {
|
|
c.PrimaryDatacenter = "dc1"
|
|
c.ACLsEnabled = true
|
|
c.ACLResolverSettings.ACLDefaultPolicy = "deny"
|
|
c.ACLInitialManagementToken = "root"
|
|
c.RPCConfig.EnableStreaming = true
|
|
})
|
|
s1.tokens.UpdateAgentToken("root", tokenStore.TokenSourceConfig)
|
|
defer os.RemoveAll(dir1)
|
|
defer s1.Shutdown()
|
|
testrpc.WaitForLeader(t, s1.RPC, "dc1")
|
|
codec := rpcClient(t, s1)
|
|
defer codec.Close()
|
|
|
|
dir2, s2 := testServerWithConfig(t, func(c *Config) {
|
|
c.Datacenter = "dc2"
|
|
c.PrimaryDatacenter = "dc1"
|
|
c.ACLsEnabled = true
|
|
c.ACLResolverSettings.ACLDefaultPolicy = "deny"
|
|
c.ACLTokenReplication = true
|
|
c.ACLReplicationRate = 100
|
|
c.ACLReplicationBurst = 100
|
|
c.ACLReplicationApplyLimit = 1000000
|
|
c.RPCConfig.EnableStreaming = true
|
|
})
|
|
s2.tokens.UpdateReplicationToken("root", tokenStore.TokenSourceConfig)
|
|
s2.tokens.UpdateAgentToken("root", tokenStore.TokenSourceConfig)
|
|
testrpc.WaitForLeader(t, s2.RPC, "dc2")
|
|
defer os.RemoveAll(dir2)
|
|
defer s2.Shutdown()
|
|
codec2 := rpcClient(t, s2)
|
|
defer codec2.Close()
|
|
|
|
// Try to join.
|
|
joinWAN(t, s2, s1)
|
|
testrpc.WaitForLeader(t, s1.RPC, "dc1")
|
|
testrpc.WaitForLeader(t, s1.RPC, "dc2")
|
|
waitForNewACLReplication(t, s2, structs.ACLReplicateTokens, 1, 1, 0)
|
|
|
|
// create simple service policy
|
|
policy, err := upsertTestPolicyWithRules(codec, "root", "dc1", `
|
|
node_prefix "" { policy = "read" }
|
|
service_prefix "" { policy = "read" }
|
|
`)
|
|
require.NoError(t, err)
|
|
|
|
// Wait for it to replicate
|
|
retry.Run(t, func(r *retry.R) {
|
|
_, p, err := s2.fsm.State().ACLPolicyGetByID(nil, policy.ID, &acl.EnterpriseMeta{})
|
|
require.Nil(r, err)
|
|
require.NotNil(r, p)
|
|
})
|
|
|
|
// create local token that only works in DC2
|
|
localToken2, err := upsertTestToken(codec, "root", "dc2", func(token *structs.ACLToken) {
|
|
token.Local = true
|
|
token.Policies = []structs.ACLTokenPolicyLink{
|
|
{ID: policy.ID},
|
|
}
|
|
})
|
|
require.NoError(t, err)
|
|
|
|
testutil.RunStep(t, "Register a dummy node with a service", func(t *testing.T) {
|
|
req := &structs.RegisterRequest{
|
|
Node: "node1",
|
|
Address: "3.4.5.6",
|
|
Datacenter: "dc1",
|
|
Service: &structs.NodeService{
|
|
ID: "redis1",
|
|
Service: "redis",
|
|
Address: "3.4.5.6",
|
|
Port: 8080,
|
|
},
|
|
WriteRequest: structs.WriteRequest{Token: "root"},
|
|
}
|
|
var out struct{}
|
|
require.NoError(t, s1.RPC(context.Background(), "Catalog.Register", &req, &out))
|
|
})
|
|
|
|
var conn *grpc.ClientConn
|
|
{
|
|
client, resolverBuilder := newClientWithGRPCPlumbing(t, func(c *Config) {
|
|
c.Datacenter = "dc2"
|
|
c.PrimaryDatacenter = "dc1"
|
|
c.RPCConfig.EnableStreaming = true
|
|
})
|
|
joinLAN(t, client, s2)
|
|
testrpc.WaitForTestAgent(t, client.RPC, "dc2", testrpc.WithToken("root"))
|
|
|
|
pool := agent_grpc.NewClientConnPool(agent_grpc.ClientConnPoolConfig{
|
|
Servers: resolverBuilder,
|
|
DialingFromServer: false,
|
|
DialingFromDatacenter: "dc2",
|
|
})
|
|
|
|
conn, err = pool.ClientConn("dc2")
|
|
require.NoError(t, err)
|
|
}
|
|
|
|
// Try to use it locally (it should work)
|
|
testutil.RunStep(t, "token used locally should work", func(t *testing.T) {
|
|
arg := &pbsubscribe.SubscribeRequest{
|
|
Topic: pbsubscribe.Topic_ServiceHealth,
|
|
Key: "redis",
|
|
Token: localToken2.SecretID,
|
|
Datacenter: "dc2",
|
|
}
|
|
event, err := getFirstSubscribeEventOrError(conn, arg)
|
|
require.NoError(t, err)
|
|
require.NotNil(t, event)
|
|
|
|
// make sure that token restore defer works
|
|
require.Equal(t, localToken2.SecretID, arg.Token, "token should not be stripped")
|
|
})
|
|
|
|
testutil.RunStep(t, "token used remotely should not work", func(t *testing.T) {
|
|
arg := &pbsubscribe.SubscribeRequest{
|
|
Topic: pbsubscribe.Topic_ServiceHealth,
|
|
Subject: &pbsubscribe.SubscribeRequest_NamedSubject{
|
|
NamedSubject: &pbsubscribe.NamedSubject{
|
|
Key: "redis",
|
|
},
|
|
},
|
|
Token: localToken2.SecretID,
|
|
Datacenter: "dc1",
|
|
}
|
|
|
|
event, err := getFirstSubscribeEventOrError(conn, arg)
|
|
|
|
// NOTE: the subscription endpoint is a filtering style instead of a
|
|
// hard-fail style so when the token isn't present 100% of the data is
|
|
// filtered out leading to a stream with an empty snapshot.
|
|
require.NoError(t, err)
|
|
require.IsType(t, &pbsubscribe.Event_EndOfSnapshot{}, event.Payload)
|
|
require.True(t, event.Payload.(*pbsubscribe.Event_EndOfSnapshot).EndOfSnapshot)
|
|
})
|
|
|
|
testutil.RunStep(t, "update anonymous token to read services", func(t *testing.T) {
|
|
tokenUpsertReq := structs.ACLTokenSetRequest{
|
|
Datacenter: "dc1",
|
|
ACLToken: structs.ACLToken{
|
|
AccessorID: acl.AnonymousTokenID,
|
|
Policies: []structs.ACLTokenPolicyLink{
|
|
{ID: policy.ID},
|
|
},
|
|
},
|
|
WriteRequest: structs.WriteRequest{Token: "root"},
|
|
}
|
|
token := structs.ACLToken{}
|
|
err = msgpackrpc.CallWithCodec(codec, "ACL.TokenSet", &tokenUpsertReq, &token)
|
|
require.NoError(t, err)
|
|
require.NotEmpty(t, token.SecretID)
|
|
})
|
|
|
|
testutil.RunStep(t, "token used remotely should fallback on anonymous token now", func(t *testing.T) {
|
|
arg := &pbsubscribe.SubscribeRequest{
|
|
Topic: pbsubscribe.Topic_ServiceHealth,
|
|
Subject: &pbsubscribe.SubscribeRequest_NamedSubject{
|
|
NamedSubject: &pbsubscribe.NamedSubject{
|
|
Key: "redis",
|
|
},
|
|
},
|
|
Token: localToken2.SecretID,
|
|
Datacenter: "dc1",
|
|
}
|
|
|
|
event, err := getFirstSubscribeEventOrError(conn, arg)
|
|
require.NoError(t, err)
|
|
require.NotNil(t, event)
|
|
|
|
// So now that we can read data, we should get a snapshot with just instances of the "consul" service.
|
|
require.NoError(t, err)
|
|
|
|
require.IsType(t, &pbsubscribe.Event_ServiceHealth{}, event.Payload)
|
|
esh := event.Payload.(*pbsubscribe.Event_ServiceHealth)
|
|
|
|
require.Equal(t, pbsubscribe.CatalogOp_Register, esh.ServiceHealth.Op)
|
|
csn := esh.ServiceHealth.CheckServiceNode
|
|
|
|
require.NotNil(t, csn)
|
|
require.NotNil(t, csn.Node)
|
|
require.Equal(t, "node1", csn.Node.Node)
|
|
require.Equal(t, "3.4.5.6", csn.Node.Address)
|
|
require.NotNil(t, csn.Service)
|
|
require.Equal(t, "redis1", csn.Service.ID)
|
|
require.Equal(t, "redis", csn.Service.Service)
|
|
|
|
// make sure that token restore defer works
|
|
require.Equal(t, localToken2.SecretID, arg.Token, "token should not be stripped")
|
|
})
|
|
}
|
|
|
|
func TestCanRetry(t *testing.T) {
|
|
type testCase struct {
|
|
name string
|
|
req structs.RPCInfo
|
|
err error
|
|
expected bool
|
|
timeout time.Time
|
|
}
|
|
config := DefaultConfig()
|
|
now := time.Now()
|
|
config.RPCHoldTimeout = 7 * time.Second
|
|
retryableMessages := []error{
|
|
ErrChunkingResubmit,
|
|
rpcRate.ErrRetryElsewhere,
|
|
}
|
|
run := func(t *testing.T, tc testCase) {
|
|
timeOutValue := tc.timeout
|
|
if timeOutValue.IsZero() {
|
|
timeOutValue = now
|
|
}
|
|
require.Equal(t, tc.expected, canRetry(tc.req, tc.err, timeOutValue, config, retryableMessages))
|
|
}
|
|
|
|
var testCases = []testCase{
|
|
{
|
|
name: "unexpected error",
|
|
err: fmt.Errorf("some arbitrary error"),
|
|
expected: false,
|
|
},
|
|
{
|
|
name: "checking error",
|
|
err: fmt.Errorf("some wrapping :%w", ErrChunkingResubmit),
|
|
expected: true,
|
|
},
|
|
{
|
|
name: "no leader error",
|
|
err: fmt.Errorf("some wrapping: %w", structs.ErrNoLeader),
|
|
expected: true,
|
|
},
|
|
{
|
|
name: "ErrRetryElsewhere",
|
|
err: fmt.Errorf("some wrapping: %w", rate.ErrRetryElsewhere),
|
|
expected: true,
|
|
},
|
|
{
|
|
name: "ErrRetryLater",
|
|
err: fmt.Errorf("some wrapping: %w", rate.ErrRetryLater),
|
|
expected: false,
|
|
},
|
|
{
|
|
name: "EOF on read request",
|
|
req: isReadRequest{},
|
|
err: io.EOF,
|
|
expected: true,
|
|
},
|
|
{
|
|
name: "EOF error",
|
|
req: &structs.DCSpecificRequest{},
|
|
err: io.EOF,
|
|
expected: true,
|
|
},
|
|
{
|
|
name: "HasTimedOut implementation with no error",
|
|
req: &structs.DCSpecificRequest{},
|
|
err: nil,
|
|
expected: false,
|
|
},
|
|
{
|
|
name: "HasTimedOut implementation timedOut with no error",
|
|
req: &structs.DCSpecificRequest{},
|
|
err: nil,
|
|
expected: false,
|
|
timeout: now.Add(-(config.RPCHoldTimeout + time.Second)),
|
|
},
|
|
{
|
|
name: "HasTimedOut implementation timedOut (with EOF error)",
|
|
req: &structs.DCSpecificRequest{},
|
|
err: io.EOF,
|
|
expected: false,
|
|
timeout: now.Add(-(config.RPCHoldTimeout + time.Second)),
|
|
},
|
|
{
|
|
name: "HasTimedOut implementation timedOut blocking call",
|
|
req: &structs.DCSpecificRequest{QueryOptions: structs.QueryOptions{MaxQueryTime: 300, MinQueryIndex: 1}},
|
|
err: nil,
|
|
expected: false,
|
|
timeout: now.Add(-(config.RPCHoldTimeout + config.MaxQueryTime + time.Second)),
|
|
},
|
|
{
|
|
name: "HasTimedOut implementation timedOut blocking call (MaxQueryTime not set)",
|
|
req: &structs.DCSpecificRequest{QueryOptions: structs.QueryOptions{MinQueryIndex: 1}},
|
|
err: nil,
|
|
expected: false,
|
|
timeout: now.Add(-(config.RPCHoldTimeout + config.MaxQueryTime + time.Second)),
|
|
},
|
|
{
|
|
name: "EOF on write request",
|
|
err: io.EOF,
|
|
expected: false,
|
|
},
|
|
}
|
|
|
|
for _, tc := range testCases {
|
|
t.Run(tc.name, func(t *testing.T) {
|
|
run(t, tc)
|
|
})
|
|
}
|
|
}
|
|
|
|
type isReadRequest struct {
|
|
structs.RPCInfo
|
|
}
|
|
|
|
func (r isReadRequest) IsRead() bool {
|
|
return true
|
|
}
|
|
|
|
func (r isReadRequest) HasTimedOut(_ time.Time, _, _, _ time.Duration) (bool, error) {
|
|
return false, nil
|
|
}
|
|
|
|
func TestRPC_AuthorizeRaftRPC(t *testing.T) {
|
|
caPEM, caPK, err := tlsutil.GenerateCA(tlsutil.CAOpts{Days: 5, Domain: "consul"})
|
|
require.NoError(t, err)
|
|
|
|
caSigner, err := tlsutil.ParseSigner(caPK)
|
|
require.NoError(t, err)
|
|
|
|
dir := testutil.TempDir(t, "certs")
|
|
err = os.WriteFile(filepath.Join(dir, "ca.pem"), []byte(caPEM), 0600)
|
|
require.NoError(t, err)
|
|
|
|
intermediatePEM, intermediatePK, err := tlsutil.GenerateCert(tlsutil.CertOpts{IsCA: true, CA: caPEM, Signer: caSigner, Days: 5})
|
|
require.NoError(t, err)
|
|
|
|
err = os.WriteFile(filepath.Join(dir, "intermediate.pem"), []byte(intermediatePEM), 0600)
|
|
require.NoError(t, err)
|
|
|
|
newCert := func(t *testing.T, caPEM, pk, node, name string) {
|
|
t.Helper()
|
|
|
|
signer, err := tlsutil.ParseSigner(pk)
|
|
require.NoError(t, err)
|
|
|
|
pem, key, err := tlsutil.GenerateCert(tlsutil.CertOpts{
|
|
Signer: signer,
|
|
CA: caPEM,
|
|
Name: name,
|
|
Days: 5,
|
|
DNSNames: []string{node + "." + name, name, "localhost"},
|
|
ExtKeyUsage: []x509.ExtKeyUsage{x509.ExtKeyUsageServerAuth, x509.ExtKeyUsageClientAuth},
|
|
})
|
|
require.NoError(t, err)
|
|
|
|
err = os.WriteFile(filepath.Join(dir, node+"-"+name+".pem"), []byte(pem), 0600)
|
|
require.NoError(t, err)
|
|
err = os.WriteFile(filepath.Join(dir, node+"-"+name+".key"), []byte(key), 0600)
|
|
require.NoError(t, err)
|
|
}
|
|
|
|
newCert(t, caPEM, caPK, "srv1", "server.dc1.consul")
|
|
|
|
_, connectCApk, err := connect.GeneratePrivateKey()
|
|
require.NoError(t, err)
|
|
|
|
_, srv := testServerWithConfig(t, func(c *Config) {
|
|
c.TLSConfig.Domain = "consul." // consul. is the default value in agent/config
|
|
c.TLSConfig.InternalRPC.CAFile = filepath.Join(dir, "ca.pem")
|
|
c.TLSConfig.InternalRPC.CertFile = filepath.Join(dir, "srv1-server.dc1.consul.pem")
|
|
c.TLSConfig.InternalRPC.KeyFile = filepath.Join(dir, "srv1-server.dc1.consul.key")
|
|
c.TLSConfig.InternalRPC.VerifyIncoming = true
|
|
c.TLSConfig.InternalRPC.VerifyServerHostname = true
|
|
// Enable Auto-Encrypt so that Connect CA roots are added to the
|
|
// tlsutil.Configurator.
|
|
c.AutoEncryptAllowTLS = true
|
|
c.CAConfig = &structs.CAConfiguration{
|
|
ClusterID: connect.TestClusterID,
|
|
Provider: structs.ConsulCAProvider,
|
|
Config: map[string]interface{}{"PrivateKey": connectCApk},
|
|
}
|
|
|
|
})
|
|
defer srv.Shutdown()
|
|
|
|
// Wait for ConnectCA initiation to complete.
|
|
retry.Run(t, func(r *retry.R) {
|
|
_, root := srv.caManager.getCAProvider()
|
|
if root == nil {
|
|
r.Fatal("ConnectCA root is still nil")
|
|
}
|
|
})
|
|
|
|
useTLSByte := func(t *testing.T, c *tlsutil.Configurator) net.Conn {
|
|
wrapper := tlsutil.SpecificDC("dc1", c.OutgoingRPCWrapper())
|
|
tlsEnabled := func(_ raft.ServerAddress) bool {
|
|
return true
|
|
}
|
|
|
|
rl := NewRaftLayer(nil, nil, wrapper, tlsEnabled)
|
|
conn, err := rl.Dial(raft.ServerAddress(srv.Listener.Addr().String()), 100*time.Millisecond)
|
|
require.NoError(t, err)
|
|
return conn
|
|
}
|
|
|
|
useNativeTLS := func(t *testing.T, c *tlsutil.Configurator) net.Conn {
|
|
wrapper := c.OutgoingALPNRPCWrapper()
|
|
dialer := &net.Dialer{Timeout: 100 * time.Millisecond}
|
|
|
|
rawConn, err := dialer.Dial("tcp", srv.Listener.Addr().String())
|
|
require.NoError(t, err)
|
|
|
|
tlsConn, err := wrapper("dc1", "srv1", pool.ALPN_RPCRaft, rawConn)
|
|
require.NoError(t, err)
|
|
return tlsConn
|
|
}
|
|
|
|
setupAgentTLSCert := func(name string) func(t *testing.T) string {
|
|
return func(t *testing.T) string {
|
|
newCert(t, caPEM, caPK, "node1", name)
|
|
return filepath.Join(dir, "node1-"+name)
|
|
}
|
|
}
|
|
|
|
setupAgentTLSCertWithIntermediate := func(name string) func(t *testing.T) string {
|
|
return func(t *testing.T) string {
|
|
newCert(t, intermediatePEM, intermediatePK, "node1", name)
|
|
certPrefix := filepath.Join(dir, "node1-"+name)
|
|
f, err := os.OpenFile(certPrefix+".pem", os.O_APPEND|os.O_CREATE|os.O_WRONLY, 0644)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if _, err := f.Write([]byte(intermediatePEM)); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if err := f.Close(); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
return certPrefix
|
|
}
|
|
}
|
|
|
|
setupConnectCACert := func(name string) func(t *testing.T) string {
|
|
return func(t *testing.T) string {
|
|
_, caRoot := srv.caManager.getCAProvider()
|
|
newCert(t, caRoot.RootCert, connectCApk, "node1", name)
|
|
return filepath.Join(dir, "node1-"+name)
|
|
}
|
|
}
|
|
|
|
type testCase struct {
|
|
name string
|
|
conn func(t *testing.T, c *tlsutil.Configurator) net.Conn
|
|
setupCert func(t *testing.T) string
|
|
expectError bool
|
|
}
|
|
|
|
run := func(t *testing.T, tc testCase) {
|
|
certPath := tc.setupCert(t)
|
|
|
|
cfg := tlsutil.Config{
|
|
InternalRPC: tlsutil.ProtocolConfig{
|
|
VerifyOutgoing: true,
|
|
VerifyServerHostname: true,
|
|
CAFile: filepath.Join(dir, "ca.pem"),
|
|
CertFile: certPath + ".pem",
|
|
KeyFile: certPath + ".key",
|
|
},
|
|
Domain: "consul",
|
|
}
|
|
c, err := tlsutil.NewConfigurator(cfg, hclog.New(nil))
|
|
require.NoError(t, err)
|
|
|
|
_, err = doRaftRPC(tc.conn(t, c), srv.config.NodeName)
|
|
if tc.expectError {
|
|
if !isConnectionClosedError(err) {
|
|
t.Fatalf("expected a connection closed error, got: %v", err)
|
|
}
|
|
return
|
|
}
|
|
require.NoError(t, err)
|
|
}
|
|
|
|
var testCases = []testCase{
|
|
{
|
|
name: "TLS byte with client cert",
|
|
setupCert: setupAgentTLSCert("client.dc1.consul"),
|
|
conn: useTLSByte,
|
|
expectError: true,
|
|
},
|
|
{
|
|
name: "TLS byte with server cert in different DC",
|
|
setupCert: setupAgentTLSCert("server.dc2.consul"),
|
|
conn: useTLSByte,
|
|
expectError: true,
|
|
},
|
|
{
|
|
name: "TLS byte with server cert in same DC",
|
|
setupCert: setupAgentTLSCert("server.dc1.consul"),
|
|
conn: useTLSByte,
|
|
},
|
|
{
|
|
name: "TLS byte with server cert in same DC and with unknown intermediate",
|
|
setupCert: setupAgentTLSCertWithIntermediate("server.dc1.consul"),
|
|
conn: useTLSByte,
|
|
},
|
|
{
|
|
name: "TLS byte with ConnectCA leaf cert",
|
|
setupCert: setupConnectCACert("server.dc1.consul"),
|
|
conn: useTLSByte,
|
|
expectError: true,
|
|
},
|
|
{
|
|
name: "native TLS with client cert",
|
|
setupCert: setupAgentTLSCert("client.dc1.consul"),
|
|
conn: useNativeTLS,
|
|
expectError: true,
|
|
},
|
|
{
|
|
name: "native TLS with server cert in different DC",
|
|
setupCert: setupAgentTLSCert("server.dc2.consul"),
|
|
conn: useNativeTLS,
|
|
expectError: true,
|
|
},
|
|
{
|
|
name: "native TLS with server cert in same DC",
|
|
setupCert: setupAgentTLSCert("server.dc1.consul"),
|
|
conn: useNativeTLS,
|
|
},
|
|
{
|
|
name: "native TLS with ConnectCA leaf cert",
|
|
setupCert: setupConnectCACert("server.dc1.consul"),
|
|
conn: useNativeTLS,
|
|
expectError: true,
|
|
},
|
|
}
|
|
|
|
for _, tc := range testCases {
|
|
t.Run(tc.name, func(t *testing.T) {
|
|
run(t, tc)
|
|
})
|
|
}
|
|
}
|
|
|
|
func TestGetWaitTime(t *testing.T) {
|
|
type testCase struct {
|
|
name string
|
|
RPCHoldTimeout time.Duration
|
|
expected time.Duration
|
|
retryCount int
|
|
}
|
|
config := DefaultConfig()
|
|
|
|
run := func(t *testing.T, tc testCase) {
|
|
config.RPCHoldTimeout = tc.RPCHoldTimeout
|
|
require.Equal(t, tc.expected, getWaitTime(config.RPCHoldTimeout, tc.retryCount))
|
|
}
|
|
|
|
var testCases = []testCase{
|
|
{
|
|
name: "init backoff small",
|
|
RPCHoldTimeout: 7 * time.Millisecond,
|
|
retryCount: 1,
|
|
expected: 1 * time.Millisecond,
|
|
},
|
|
{
|
|
name: "first attempt",
|
|
RPCHoldTimeout: 7 * time.Second,
|
|
retryCount: 1,
|
|
expected: 437 * time.Millisecond,
|
|
},
|
|
{
|
|
name: "second attempt",
|
|
RPCHoldTimeout: 7 * time.Second,
|
|
retryCount: 2,
|
|
expected: 874 * time.Millisecond,
|
|
},
|
|
{
|
|
name: "third attempt",
|
|
RPCHoldTimeout: 7 * time.Second,
|
|
retryCount: 3,
|
|
expected: 1748 * time.Millisecond,
|
|
},
|
|
{
|
|
name: "fourth attempt",
|
|
RPCHoldTimeout: 7 * time.Second,
|
|
retryCount: 4,
|
|
expected: 3496 * time.Millisecond,
|
|
},
|
|
}
|
|
|
|
for _, tc := range testCases {
|
|
t.Run(tc.name, func(t *testing.T) {
|
|
run(t, tc)
|
|
})
|
|
}
|
|
}
|
|
|
|
func doRaftRPC(conn net.Conn, leader string) (raft.AppendEntriesResponse, error) {
|
|
var resp raft.AppendEntriesResponse
|
|
|
|
var term uint64 = 0xc
|
|
a := raft.AppendEntriesRequest{
|
|
RPCHeader: raft.RPCHeader{ProtocolVersion: 3},
|
|
Term: 0,
|
|
Leader: []byte(leader),
|
|
PrevLogEntry: 0,
|
|
PrevLogTerm: term,
|
|
LeaderCommitIndex: 50,
|
|
}
|
|
|
|
if err := appendEntries(conn, a, &resp); err != nil {
|
|
return resp, err
|
|
}
|
|
return resp, nil
|
|
}
|
|
|
|
func appendEntries(conn net.Conn, req raft.AppendEntriesRequest, resp *raft.AppendEntriesResponse) error {
|
|
w := bufio.NewWriter(conn)
|
|
enc := codec.NewEncoder(w, &codec.MsgpackHandle{})
|
|
|
|
const rpcAppendEntries = 0
|
|
if err := w.WriteByte(rpcAppendEntries); err != nil {
|
|
return fmt.Errorf("failed to write raft-RPC byte: %w", err)
|
|
}
|
|
|
|
if err := enc.Encode(req); err != nil {
|
|
return fmt.Errorf("failed to send append entries RPC: %w", err)
|
|
}
|
|
if err := w.Flush(); err != nil {
|
|
return fmt.Errorf("failed to flush RPC: %w", err)
|
|
}
|
|
|
|
if err := decodeRaftRPCResponse(conn, resp); err != nil {
|
|
return fmt.Errorf("response error: %w", err)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// copied and modified from raft/net_transport.go
|
|
func decodeRaftRPCResponse(conn net.Conn, resp *raft.AppendEntriesResponse) error {
|
|
r := bufio.NewReader(conn)
|
|
dec := codec.NewDecoder(r, &codec.MsgpackHandle{})
|
|
|
|
var rpcError string
|
|
if err := dec.Decode(&rpcError); err != nil {
|
|
return fmt.Errorf("failed to decode response error: %w", err)
|
|
}
|
|
if err := dec.Decode(resp); err != nil {
|
|
return fmt.Errorf("failed to decode response: %w", err)
|
|
}
|
|
if rpcError != "" {
|
|
return fmt.Errorf("rpc error: %v", rpcError)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func isConnectionClosedError(err error) bool {
|
|
switch {
|
|
case err == nil:
|
|
return false
|
|
case errors.Is(err, io.EOF):
|
|
return true
|
|
case strings.Contains(err.Error(), "connection reset by peer"):
|
|
return true
|
|
default:
|
|
return false
|
|
}
|
|
}
|
|
|
|
func getFirstSubscribeEventOrError(conn *grpc.ClientConn, req *pbsubscribe.SubscribeRequest) (*pbsubscribe.Event, error) {
|
|
streamClient := pbsubscribe.NewStateChangeSubscriptionClient(conn)
|
|
ctx, cancel := context.WithCancel(context.Background())
|
|
defer cancel()
|
|
|
|
handle, err := streamClient.Subscribe(ctx, req)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
event, err := handle.Recv()
|
|
if err == io.EOF {
|
|
return nil, nil
|
|
}
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
return event, nil
|
|
}
|
|
|
|
// channelCallRPC lets you execute an RPC async. Helpful in some
|
|
// tests.
|
|
func channelCallRPC(
|
|
srv *Server,
|
|
method string,
|
|
args interface{},
|
|
resp interface{},
|
|
responseInterceptor func() error,
|
|
) <-chan error {
|
|
errCh := make(chan error, 1)
|
|
go func() {
|
|
codec, err := rpcClientNoClose(srv)
|
|
if err != nil {
|
|
errCh <- err
|
|
return
|
|
}
|
|
defer codec.Close()
|
|
|
|
err = msgpackrpc.CallWithCodec(codec, method, args, resp)
|
|
if err == nil && responseInterceptor != nil {
|
|
err = responseInterceptor()
|
|
}
|
|
errCh <- err
|
|
}()
|
|
return errCh
|
|
}
|
|
|
|
// rpcBlockingQueryTestHarness is specifically meant to test the
|
|
// errNotFound and errNotChanged mechanisms in blockingQuery()
|
|
func rpcBlockingQueryTestHarness(
|
|
t *testing.T,
|
|
readQueryFn func(minQueryIndex uint64) (*structs.QueryMeta, <-chan error),
|
|
noisyWriteFn func(i int) <-chan error,
|
|
) {
|
|
t.Helper()
|
|
|
|
ctx, cancel := context.WithCancel(context.Background())
|
|
defer cancel()
|
|
|
|
launchWriters := func() {
|
|
defer cancel()
|
|
|
|
for i := 0; i < 200; i++ {
|
|
time.Sleep(5 * time.Millisecond)
|
|
|
|
errCh := noisyWriteFn(i)
|
|
select {
|
|
case <-ctx.Done():
|
|
return
|
|
case err := <-errCh:
|
|
if err != nil {
|
|
t.Errorf("[%d] unexpected error: %v", i, err)
|
|
return
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
var (
|
|
count int
|
|
minQueryIndex uint64
|
|
)
|
|
|
|
for ctx.Err() == nil {
|
|
// The first iteration is an orientation iteration, as we don't pass an
|
|
// index value so there is no actual blocking that will happen.
|
|
//
|
|
// Since the data is not changing, we don't expect the second iteration
|
|
// to return soon, so we wait a bit after kicking it off before
|
|
// launching the write-storm.
|
|
var timerCh <-chan time.Time
|
|
if count == 1 {
|
|
timerCh = time.After(50 * time.Millisecond)
|
|
}
|
|
|
|
qm, errCh := readQueryFn(minQueryIndex)
|
|
|
|
RESUME:
|
|
select {
|
|
case err := <-errCh:
|
|
if err != nil {
|
|
require.NoError(t, err)
|
|
}
|
|
|
|
t.Log("blocking query index", qm.Index)
|
|
count++
|
|
minQueryIndex = qm.Index
|
|
|
|
case <-timerCh:
|
|
timerCh = nil
|
|
go launchWriters()
|
|
goto RESUME
|
|
|
|
case <-ctx.Done():
|
|
break
|
|
}
|
|
}
|
|
|
|
require.Equal(t, 1, count, "if this fails, then the timer likely needs to be increased above")
|
|
}
|