open-consul/agent/service_manager_test.go
Daniel Nephin b6d91d59f3 testing: Remove TestAgent.Key and change TestAgent.DataDir
TestAgent.Key was only used by 3 tests. Extracting it from the common helper that is used in hundreds of
tests helps keep the shared part small and more focused.

This required a second change (which I was planning on making anyway), which was to change the behaviour of
DataDir. Now in all cases the TestAgent will use the DataDir, and clean it up once the test is complete.
2020-08-13 17:53:24 -04:00

817 lines
21 KiB
Go

package agent
import (
"encoding/json"
"fmt"
"io/ioutil"
"os"
"path/filepath"
"testing"
"github.com/hashicorp/consul/agent/structs"
"github.com/hashicorp/consul/sdk/testutil/retry"
"github.com/hashicorp/consul/testrpc"
"github.com/stretchr/testify/require"
)
func TestServiceManager_RegisterService(t *testing.T) {
require := require.New(t)
a := NewTestAgent(t, "enable_central_service_config = true")
defer a.Shutdown()
testrpc.WaitForLeader(t, a.RPC, "dc1")
// Register a global proxy and service config
testApplyConfigEntries(t, a,
&structs.ProxyConfigEntry{
Config: map[string]interface{}{
"foo": 1,
},
},
&structs.ServiceConfigEntry{
Kind: structs.ServiceDefaults,
Name: "redis",
Protocol: "tcp",
},
)
// Now register a service locally with no sidecar, it should be a no-op.
svc := &structs.NodeService{
ID: "redis",
Service: "redis",
Port: 8000,
EnterpriseMeta: *structs.DefaultEnterpriseMeta(),
}
require.NoError(a.AddService(svc, nil, false, "", ConfigSourceLocal))
// Verify both the service and sidecar.
redisService := a.State.Service(structs.NewServiceID("redis", nil))
require.NotNil(redisService)
require.Equal(&structs.NodeService{
ID: "redis",
Service: "redis",
Port: 8000,
TaggedAddresses: map[string]structs.ServiceAddress{},
Weights: &structs.Weights{
Passing: 1,
Warning: 1,
},
EnterpriseMeta: *structs.DefaultEnterpriseMeta(),
}, redisService)
}
func TestServiceManager_RegisterSidecar(t *testing.T) {
require := require.New(t)
a := NewTestAgent(t, "enable_central_service_config = true")
defer a.Shutdown()
testrpc.WaitForLeader(t, a.RPC, "dc1")
// Register a global proxy and service config
testApplyConfigEntries(t, a,
&structs.ProxyConfigEntry{
Config: map[string]interface{}{
"foo": 1,
},
},
&structs.ServiceConfigEntry{
Kind: structs.ServiceDefaults,
Name: "web",
Protocol: "http",
},
&structs.ServiceConfigEntry{
Kind: structs.ServiceDefaults,
Name: "redis",
Protocol: "tcp",
},
)
// Now register a sidecar proxy. Note we don't use SidecarService here because
// that gets resolved earlier in config handling than the AddService call
// here.
svc := &structs.NodeService{
Kind: structs.ServiceKindConnectProxy,
ID: "web-sidecar-proxy",
Service: "web-sidecar-proxy",
Port: 21000,
Proxy: structs.ConnectProxyConfig{
DestinationServiceName: "web",
DestinationServiceID: "web",
LocalServiceAddress: "127.0.0.1",
LocalServicePort: 8000,
Upstreams: structs.Upstreams{
{
DestinationName: "redis",
LocalBindPort: 5000,
},
},
},
EnterpriseMeta: *structs.DefaultEnterpriseMeta(),
}
require.NoError(a.AddService(svc, nil, false, "", ConfigSourceLocal))
// Verify sidecar got global config loaded
sidecarService := a.State.Service(structs.NewServiceID("web-sidecar-proxy", nil))
require.NotNil(sidecarService)
require.Equal(&structs.NodeService{
Kind: structs.ServiceKindConnectProxy,
ID: "web-sidecar-proxy",
Service: "web-sidecar-proxy",
Port: 21000,
TaggedAddresses: map[string]structs.ServiceAddress{},
Proxy: structs.ConnectProxyConfig{
DestinationServiceName: "web",
DestinationServiceID: "web",
LocalServiceAddress: "127.0.0.1",
LocalServicePort: 8000,
Config: map[string]interface{}{
"foo": int64(1),
"protocol": "http",
},
Upstreams: structs.Upstreams{
{
DestinationName: "redis",
LocalBindPort: 5000,
Config: map[string]interface{}{
"protocol": "tcp",
},
},
},
},
Weights: &structs.Weights{
Passing: 1,
Warning: 1,
},
EnterpriseMeta: *structs.DefaultEnterpriseMeta(),
}, sidecarService)
}
func TestServiceManager_RegisterMeshGateway(t *testing.T) {
require := require.New(t)
a := NewTestAgent(t, "enable_central_service_config = true")
defer a.Shutdown()
testrpc.WaitForLeader(t, a.RPC, "dc1")
// Register a global proxy and service config
testApplyConfigEntries(t, a,
&structs.ProxyConfigEntry{
Config: map[string]interface{}{
"foo": 1,
},
},
&structs.ServiceConfigEntry{
Kind: structs.ServiceDefaults,
Name: "mesh-gateway",
Protocol: "http",
},
)
// Now register a mesh-gateway.
svc := &structs.NodeService{
Kind: structs.ServiceKindMeshGateway,
ID: "mesh-gateway",
Service: "mesh-gateway",
Port: 443,
EnterpriseMeta: *structs.DefaultEnterpriseMeta(),
}
require.NoError(a.AddService(svc, nil, false, "", ConfigSourceLocal))
// Verify gateway got global config loaded
gateway := a.State.Service(structs.NewServiceID("mesh-gateway", nil))
require.NotNil(gateway)
require.Equal(&structs.NodeService{
Kind: structs.ServiceKindMeshGateway,
ID: "mesh-gateway",
Service: "mesh-gateway",
Port: 443,
TaggedAddresses: map[string]structs.ServiceAddress{},
Proxy: structs.ConnectProxyConfig{
Config: map[string]interface{}{
"foo": int64(1),
"protocol": "http",
},
},
Weights: &structs.Weights{
Passing: 1,
Warning: 1,
},
EnterpriseMeta: *structs.DefaultEnterpriseMeta(),
}, gateway)
}
func TestServiceManager_RegisterTerminatingGateway(t *testing.T) {
require := require.New(t)
a := NewTestAgent(t, "enable_central_service_config = true")
defer a.Shutdown()
testrpc.WaitForLeader(t, a.RPC, "dc1")
// Register a global proxy and service config
testApplyConfigEntries(t, a,
&structs.ProxyConfigEntry{
Config: map[string]interface{}{
"foo": 1,
},
},
&structs.ServiceConfigEntry{
Kind: structs.ServiceDefaults,
Name: "terminating-gateway",
Protocol: "http",
},
)
// Now register a terminating-gateway.
svc := &structs.NodeService{
Kind: structs.ServiceKindTerminatingGateway,
ID: "terminating-gateway",
Service: "terminating-gateway",
Port: 443,
EnterpriseMeta: *structs.DefaultEnterpriseMeta(),
}
require.NoError(a.AddService(svc, nil, false, "", ConfigSourceLocal))
// Verify gateway got global config loaded
gateway := a.State.Service(structs.NewServiceID("terminating-gateway", nil))
require.NotNil(gateway)
require.Equal(&structs.NodeService{
Kind: structs.ServiceKindTerminatingGateway,
ID: "terminating-gateway",
Service: "terminating-gateway",
Port: 443,
TaggedAddresses: map[string]structs.ServiceAddress{},
Proxy: structs.ConnectProxyConfig{
Config: map[string]interface{}{
"foo": int64(1),
"protocol": "http",
},
},
Weights: &structs.Weights{
Passing: 1,
Warning: 1,
},
EnterpriseMeta: *structs.DefaultEnterpriseMeta(),
}, gateway)
}
func TestServiceManager_PersistService_API(t *testing.T) {
// This is the ServiceManager version of TestAgent_PersistService and
// TestAgent_PurgeService.
t.Parallel()
require := require.New(t)
// Launch a server to manage the config entries.
serverAgent := NewTestAgent(t, `enable_central_service_config = true`)
defer serverAgent.Shutdown()
testrpc.WaitForLeader(t, serverAgent.RPC, "dc1")
// Register a global proxy and service config
testApplyConfigEntries(t, serverAgent,
&structs.ProxyConfigEntry{
Config: map[string]interface{}{
"foo": 1,
},
},
&structs.ServiceConfigEntry{
Kind: structs.ServiceDefaults,
Name: "web",
Protocol: "http",
},
&structs.ServiceConfigEntry{
Kind: structs.ServiceDefaults,
Name: "redis",
Protocol: "tcp",
},
)
// Now launch a single client agent
cfg := `
enable_central_service_config = true
server = false
bootstrap = false
`
a := StartTestAgent(t, TestAgent{HCL: cfg})
defer a.Shutdown()
// Join first
_, err := a.JoinLAN([]string{
fmt.Sprintf("127.0.0.1:%d", serverAgent.Config.SerfPortLAN),
})
require.NoError(err)
testrpc.WaitForLeader(t, a.RPC, "dc1")
// Now register a sidecar proxy via the API.
svc := &structs.NodeService{
Kind: structs.ServiceKindConnectProxy,
ID: "web-sidecar-proxy",
Service: "web-sidecar-proxy",
Port: 21000,
Proxy: structs.ConnectProxyConfig{
DestinationServiceName: "web",
DestinationServiceID: "web",
LocalServiceAddress: "127.0.0.1",
LocalServicePort: 8000,
Upstreams: structs.Upstreams{
{
DestinationName: "redis",
LocalBindPort: 5000,
},
},
},
EnterpriseMeta: *structs.DefaultEnterpriseMeta(),
}
expectState := &structs.NodeService{
Kind: structs.ServiceKindConnectProxy,
ID: "web-sidecar-proxy",
Service: "web-sidecar-proxy",
Port: 21000,
TaggedAddresses: map[string]structs.ServiceAddress{},
Proxy: structs.ConnectProxyConfig{
DestinationServiceName: "web",
DestinationServiceID: "web",
LocalServiceAddress: "127.0.0.1",
LocalServicePort: 8000,
Config: map[string]interface{}{
"foo": int64(1),
"protocol": "http",
},
Upstreams: structs.Upstreams{
{
DestinationName: "redis",
LocalBindPort: 5000,
Config: map[string]interface{}{
"protocol": "tcp",
},
},
},
},
Weights: &structs.Weights{
Passing: 1,
Warning: 1,
},
EnterpriseMeta: *structs.DefaultEnterpriseMeta(),
}
svcID := svc.CompoundServiceID()
svcFile := filepath.Join(a.Config.DataDir, servicesDir, svcID.StringHash())
configFile := filepath.Join(a.Config.DataDir, serviceConfigDir, svcID.StringHash())
// Service is not persisted unless requested, but we always persist service configs.
require.NoError(a.AddService(svc, nil, false, "", ConfigSourceRemote))
requireFileIsAbsent(t, svcFile)
requireFileIsPresent(t, configFile)
// Persists to file if requested
require.NoError(a.AddService(svc, nil, true, "mytoken", ConfigSourceRemote))
requireFileIsPresent(t, svcFile)
requireFileIsPresent(t, configFile)
// Service definition file is sane.
expectJSONFile(t, svcFile, persistedService{
Token: "mytoken",
Service: svc,
Source: "remote",
}, nil)
// Service config file is sane.
pcfg := persistedServiceConfig{
ServiceID: "web-sidecar-proxy",
Defaults: &structs.ServiceConfigResponse{
ProxyConfig: map[string]interface{}{
"foo": 1,
"protocol": "http",
},
UpstreamIDConfigs: structs.UpstreamConfigs{
structs.UpstreamConfig{
Upstream: structs.NewServiceID("redis", nil),
Config: map[string]interface{}{
"protocol": "tcp",
},
},
},
},
EnterpriseMeta: *structs.DefaultEnterpriseMeta(),
}
expectJSONFile(t, configFile, pcfg, resetDefaultsQueryMeta)
// Verify in memory state.
{
sidecarService := a.State.Service(structs.NewServiceID("web-sidecar-proxy", nil))
require.NotNil(sidecarService)
require.Equal(expectState, sidecarService)
}
// Updates service definition on disk
svc.Proxy.LocalServicePort = 8001
require.NoError(a.AddService(svc, nil, true, "mytoken", ConfigSourceRemote))
requireFileIsPresent(t, svcFile)
requireFileIsPresent(t, configFile)
// Service definition file is updated.
expectJSONFile(t, svcFile, persistedService{
Token: "mytoken",
Service: svc,
Source: "remote",
}, nil)
// Service config file is the same.
pcfg = persistedServiceConfig{
ServiceID: "web-sidecar-proxy",
Defaults: &structs.ServiceConfigResponse{
ProxyConfig: map[string]interface{}{
"foo": 1,
"protocol": "http",
},
UpstreamIDConfigs: structs.UpstreamConfigs{
structs.UpstreamConfig{
Upstream: structs.NewServiceID("redis", nil),
Config: map[string]interface{}{
"protocol": "tcp",
},
},
},
},
EnterpriseMeta: *structs.DefaultEnterpriseMeta(),
}
expectJSONFile(t, configFile, pcfg, resetDefaultsQueryMeta)
// Verify in memory state.
expectState.Proxy.LocalServicePort = 8001
{
sidecarService := a.State.Service(structs.NewServiceID("web-sidecar-proxy", nil))
require.NotNil(sidecarService)
require.Equal(expectState, sidecarService)
}
// Kill the agent to restart it.
a.Shutdown()
// Kill the server so that it can't phone home and must rely upon the persisted defaults.
serverAgent.Shutdown()
// Should load it back during later start.
a2 := StartTestAgent(t, TestAgent{HCL: cfg, DataDir: a.DataDir})
defer a2.Shutdown()
{
restored := a.State.Service(structs.NewServiceID("web-sidecar-proxy", nil))
require.NotNil(restored)
require.Equal(expectState, restored)
}
// Now remove it.
require.NoError(a2.RemoveService(structs.NewServiceID("web-sidecar-proxy", nil)))
requireFileIsAbsent(t, svcFile)
requireFileIsAbsent(t, configFile)
}
func TestServiceManager_PersistService_ConfigFiles(t *testing.T) {
// This is the ServiceManager version of TestAgent_PersistService and
// TestAgent_PurgeService but for config files.
t.Parallel()
// Launch a server to manage the config entries.
serverAgent := NewTestAgent(t, `enable_central_service_config = true`)
defer serverAgent.Shutdown()
testrpc.WaitForLeader(t, serverAgent.RPC, "dc1")
// Register a global proxy and service config
testApplyConfigEntries(t, serverAgent,
&structs.ProxyConfigEntry{
Config: map[string]interface{}{
"foo": 1,
},
},
&structs.ServiceConfigEntry{
Kind: structs.ServiceDefaults,
Name: "web",
Protocol: "http",
},
&structs.ServiceConfigEntry{
Kind: structs.ServiceDefaults,
Name: "redis",
Protocol: "tcp",
},
)
// Now launch a single client agent
serviceSnippet := `
service = {
kind = "connect-proxy"
id = "web-sidecar-proxy"
name = "web-sidecar-proxy"
port = 21000
token = "mytoken"
proxy {
destination_service_name = "web"
destination_service_id = "web"
local_service_address = "127.0.0.1"
local_service_port = 8000
upstreams = [{
destination_name = "redis"
local_bind_port = 5000
}]
}
}
`
cfg := `
enable_central_service_config = true
server = false
bootstrap = false
` + serviceSnippet
a := StartTestAgent(t, TestAgent{HCL: cfg})
defer a.Shutdown()
// Join first
_, err := a.JoinLAN([]string{
fmt.Sprintf("127.0.0.1:%d", serverAgent.Config.SerfPortLAN),
})
require.NoError(t, err)
testrpc.WaitForLeader(t, a.RPC, "dc1")
// Now register a sidecar proxy via the API.
svcID := "web-sidecar-proxy"
expectState := &structs.NodeService{
Kind: structs.ServiceKindConnectProxy,
ID: "web-sidecar-proxy",
Service: "web-sidecar-proxy",
Port: 21000,
TaggedAddresses: map[string]structs.ServiceAddress{},
Proxy: structs.ConnectProxyConfig{
DestinationServiceName: "web",
DestinationServiceID: "web",
LocalServiceAddress: "127.0.0.1",
LocalServicePort: 8000,
Config: map[string]interface{}{
"foo": int64(1),
"protocol": "http",
},
Upstreams: structs.Upstreams{
{
DestinationType: "service",
DestinationName: "redis",
LocalBindPort: 5000,
Config: map[string]interface{}{
"protocol": "tcp",
},
},
},
},
Weights: &structs.Weights{
Passing: 1,
Warning: 1,
},
EnterpriseMeta: *structs.DefaultEnterpriseMeta(),
}
// Now wait until we've re-registered using central config updated data.
retry.Run(t, func(r *retry.R) {
a.stateLock.Lock()
defer a.stateLock.Unlock()
current := a.State.Service(structs.NewServiceID("web-sidecar-proxy", nil))
if current == nil {
r.Fatalf("service is missing")
}
require.Equal(r, expectState, current)
})
svcFile := filepath.Join(a.Config.DataDir, servicesDir, stringHash(svcID))
configFile := filepath.Join(a.Config.DataDir, serviceConfigDir, stringHash(svcID))
// Service is never persisted, but we always persist service configs.
requireFileIsAbsent(t, svcFile)
requireFileIsPresent(t, configFile)
// Service config file is sane.
expectJSONFile(t, configFile, persistedServiceConfig{
ServiceID: "web-sidecar-proxy",
Defaults: &structs.ServiceConfigResponse{
ProxyConfig: map[string]interface{}{
"foo": 1,
"protocol": "http",
},
UpstreamIDConfigs: structs.UpstreamConfigs{
structs.UpstreamConfig{
Upstream: structs.NewServiceID("redis", nil),
Config: map[string]interface{}{
"protocol": "tcp",
},
},
},
},
EnterpriseMeta: *structs.DefaultEnterpriseMeta(),
}, resetDefaultsQueryMeta)
// Verify in memory state.
{
sidecarService := a.State.Service(structs.NewServiceID("web-sidecar-proxy", nil))
require.NotNil(t, sidecarService)
require.Equal(t, expectState, sidecarService)
}
// Kill the agent to restart it.
a.Shutdown()
// Kill the server so that it can't phone home and must rely upon the persisted defaults.
serverAgent.Shutdown()
// Should load it back during later start.
a2 := StartTestAgent(t, TestAgent{HCL: cfg, DataDir: a.DataDir})
defer a2.Shutdown()
{
restored := a.State.Service(structs.NewServiceID("web-sidecar-proxy", nil))
require.NotNil(t, restored)
require.Equal(t, expectState, restored)
}
// Now remove it.
require.NoError(t, a2.RemoveService(structs.NewServiceID("web-sidecar-proxy", nil)))
requireFileIsAbsent(t, svcFile)
requireFileIsAbsent(t, configFile)
}
func TestServiceManager_Disabled(t *testing.T) {
require := require.New(t)
a := NewTestAgent(t, "enable_central_service_config = false")
defer a.Shutdown()
testrpc.WaitForLeader(t, a.RPC, "dc1")
// Register a global proxy and service config
testApplyConfigEntries(t, a,
&structs.ProxyConfigEntry{
Config: map[string]interface{}{
"foo": 1,
},
},
&structs.ServiceConfigEntry{
Kind: structs.ServiceDefaults,
Name: "web",
Protocol: "http",
},
&structs.ServiceConfigEntry{
Kind: structs.ServiceDefaults,
Name: "redis",
Protocol: "tcp",
},
)
// Now register a sidecar proxy. Note we don't use SidecarService here because
// that gets resolved earlier in config handling than the AddService call
// here.
svc := &structs.NodeService{
Kind: structs.ServiceKindConnectProxy,
ID: "web-sidecar-proxy",
Service: "web-sidecar-proxy",
Port: 21000,
Proxy: structs.ConnectProxyConfig{
DestinationServiceName: "web",
DestinationServiceID: "web",
LocalServiceAddress: "127.0.0.1",
LocalServicePort: 8000,
Upstreams: structs.Upstreams{
{
DestinationName: "redis",
LocalBindPort: 5000,
},
},
},
EnterpriseMeta: *structs.DefaultEnterpriseMeta(),
}
require.NoError(a.AddService(svc, nil, false, "", ConfigSourceLocal))
// Verify sidecar got global config loaded
sidecarService := a.State.Service(structs.NewServiceID("web-sidecar-proxy", nil))
require.NotNil(sidecarService)
require.Equal(&structs.NodeService{
Kind: structs.ServiceKindConnectProxy,
ID: "web-sidecar-proxy",
Service: "web-sidecar-proxy",
Port: 21000,
TaggedAddresses: map[string]structs.ServiceAddress{},
Proxy: structs.ConnectProxyConfig{
DestinationServiceName: "web",
DestinationServiceID: "web",
LocalServiceAddress: "127.0.0.1",
LocalServicePort: 8000,
// No config added
Upstreams: structs.Upstreams{
{
DestinationName: "redis",
LocalBindPort: 5000,
// No config added
},
},
},
Weights: &structs.Weights{
Passing: 1,
Warning: 1,
},
EnterpriseMeta: *structs.DefaultEnterpriseMeta(),
}, sidecarService)
}
func testApplyConfigEntries(t *testing.T, a *TestAgent, entries ...structs.ConfigEntry) {
t.Helper()
for _, entry := range entries {
args := &structs.ConfigEntryRequest{
Datacenter: "dc1",
Entry: entry,
}
var out bool
require.NoError(t, a.RPC("ConfigEntry.Apply", args, &out))
}
}
func requireFileIsAbsent(t *testing.T, file string) {
t.Helper()
if _, err := os.Stat(file); !os.IsNotExist(err) {
t.Fatalf("should not persist")
}
}
func requireFileIsPresent(t *testing.T, file string) {
t.Helper()
if _, err := os.Stat(file); err != nil {
t.Fatalf("err: %v", err)
}
}
func expectJSONFile(t *testing.T, file string, expect interface{}, fixupContentBeforeCompareFn func([]byte) ([]byte, error)) {
t.Helper()
expected, err := json.Marshal(expect)
require.NoError(t, err)
content, err := ioutil.ReadFile(file)
require.NoError(t, err)
if fixupContentBeforeCompareFn != nil {
content, err = fixupContentBeforeCompareFn(content)
require.NoError(t, err)
}
require.JSONEq(t, string(expected), string(content))
}
// resetDefaultsQueryMeta will reset the embedded fields from structs.QueryMeta
// to their zero values in the json object keyed under 'Defaults'.
func resetDefaultsQueryMeta(content []byte) ([]byte, error) {
var raw map[string]interface{}
if err := json.Unmarshal(content, &raw); err != nil {
return nil, err
}
def, ok := raw["Defaults"]
if !ok {
return content, nil
}
rawDef, ok := def.(map[string]interface{})
if !ok {
return nil, fmt.Errorf("unexpected structure found in 'Defaults' key")
}
qmZero, err := convertToMap(structs.QueryMeta{})
if err != nil {
return nil, err
}
for k, v := range qmZero {
rawDef[k] = v
}
raw["Defaults"] = rawDef
return json.Marshal(raw)
}
func convertToMap(v interface{}) (map[string]interface{}, error) {
b, err := json.Marshal(v)
if err != nil {
return nil, err
}
var raw map[string]interface{}
if err := json.Unmarshal(b, &raw); err != nil {
return nil, err
}
return raw, nil
}