luxolus
/
open-consul
2
0
Fork 0
Code Issues 1 Pull Requests Packages Releases Wiki Activity
open-consul/agent/agent_test.go

6158 lines
166 KiB
Go
Raw Blame History

package agent
import (
"bytes"
"context"
"crypto/md5"
"crypto/tls"
"crypto/x509"
"encoding/base64"
"encoding/json"
"fmt"
"io/ioutil"
"math/rand"
"net"
"net/http"
"net/http/httptest"
"net/url"
"os"
"path"
"path/filepath"
"strconv"
"strings"
"sync"
"testing"
"time"
"github.com/golang/protobuf/jsonpb"
"github.com/google/go-cmp/cmp"
"github.com/google/go-cmp/cmp/cmpopts"
"github.com/google/tcpproxy"
"github.com/hashicorp/consul/agent/hcp"
"github.com/hashicorp/consul/agent/hcp/scada"
"github.com/hashicorp/go-hclog"
"github.com/hashicorp/hcp-scada-provider/capability"
"github.com/hashicorp/serf/coordinate"
"github.com/hashicorp/serf/serf"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/mock"
"github.com/stretchr/testify/require"
"golang.org/x/sync/errgroup"
"google.golang.org/grpc"
"gopkg.in/square/go-jose.v2/jwt"
"github.com/hashicorp/consul/agent/cache"
cachetype "github.com/hashicorp/consul/agent/cache-types"
"github.com/hashicorp/consul/agent/checks"
"github.com/hashicorp/consul/agent/config"
"github.com/hashicorp/consul/agent/connect"
"github.com/hashicorp/consul/agent/consul"
"github.com/hashicorp/consul/agent/structs"
"github.com/hashicorp/consul/agent/token"
"github.com/hashicorp/consul/api"
"github.com/hashicorp/consul/internal/go-sso/oidcauth/oidcauthtest"
"github.com/hashicorp/consul/ipaddr"
"github.com/hashicorp/consul/lib"
"github.com/hashicorp/consul/proto/pbautoconf"
"github.com/hashicorp/consul/sdk/freeport"
"github.com/hashicorp/consul/sdk/testutil"
"github.com/hashicorp/consul/sdk/testutil/retry"
"github.com/hashicorp/consul/testrpc"
"github.com/hashicorp/consul/tlsutil"
"github.com/hashicorp/consul/types"
)
func getService(a *TestAgent, id string) *structs.NodeService {
return a.State.Service(structs.NewServiceID(id, nil))
}
func getCheck(a *TestAgent, id types.CheckID) *structs.HealthCheck {
return a.State.Check(structs.NewCheckID(id, nil))
}
func requireServiceExists(t *testing.T, a *TestAgent, id string) *structs.NodeService {
t.Helper()
svc := getService(a, id)
require.NotNil(t, svc, "missing service %q", id)
return svc
}
func requireServiceMissing(t *testing.T, a *TestAgent, id string) {
t.Helper()
require.Nil(t, getService(a, id), "have service %q (expected missing)", id)
}
func requireCheckExists(t *testing.T, a *TestAgent, id types.CheckID) *structs.HealthCheck {
t.Helper()
chk := getCheck(a, id)
require.NotNil(t, chk, "missing check %q", id)
return chk
}
func requireCheckMissing(t *testing.T, a *TestAgent, id types.CheckID) {
t.Helper()
require.Nil(t, getCheck(a, id), "have check %q (expected missing)", id)
}
func requireCheckExistsMap(t *testing.T, m interface{}, id types.CheckID) {
t.Helper()
require.Contains(t, m, structs.NewCheckID(id, nil), "missing check %q", id)
}
func requireCheckMissingMap(t *testing.T, m interface{}, id types.CheckID) {
t.Helper()
require.NotContains(t, m, structs.NewCheckID(id, nil), "have check %q (expected missing)", id)
}
func TestAgent_MultiStartStop(t *testing.T) {
if testing.Short() {
t.Skip("too slow for testing.Short")
}
for i := 0; i < 10; i++ {
t.Run("", func(t *testing.T) {
t.Parallel()
a := NewTestAgent(t, "")
time.Sleep(250 * time.Millisecond)
a.Shutdown()
})
}
}
func TestAgent_ConnectClusterIDConfig(t *testing.T) {
if testing.Short() {
t.Skip("too slow for testing.Short")
}
tests := []struct {
name string
hcl string
wantClusterID string
wantErr bool
}{
{
name: "default TestAgent has fixed cluster id",
hcl: "",
wantClusterID: connect.TestClusterID,
},
{
name: "no cluster ID specified sets to test ID",
hcl: "connect { enabled = true }",
wantClusterID: connect.TestClusterID,
},
{
name: "non-UUID cluster_id is fatal",
hcl: `connect {
enabled = true
ca_config {
cluster_id = "fake-id"
}
}`,
wantClusterID: "",
wantErr: true,
},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
a := TestAgent{HCL: tt.hcl}
err := a.Start(t)
if tt.wantErr {
if err == nil {
t.Fatal("expected error, got nil")
}
return // don't run the rest of the test
}
if !tt.wantErr && err != nil {
t.Fatal(err)
}
defer a.Shutdown()
cfg := a.consulConfig()
assert.Equal(t, tt.wantClusterID, cfg.CAConfig.ClusterID)
})
}
}
func TestAgent_StartStop(t *testing.T) {
if testing.Short() {
t.Skip("too slow for testing.Short")
}
t.Parallel()
a := NewTestAgent(t, "")
defer a.Shutdown()
if err := a.Leave(); err != nil {
t.Fatalf("err: %v", err)
}
if err := a.Shutdown(); err != nil {
t.Fatalf("err: %v", err)
}
select {
case <-a.ShutdownCh():
default:
t.Fatalf("should be closed")
}
}
func TestAgent_RPCPing(t *testing.T) {
if testing.Short() {
t.Skip("too slow for testing.Short")
}
t.Parallel()
a := NewTestAgent(t, "")
defer a.Shutdown()
testrpc.WaitForTestAgent(t, a.RPC, "dc1")
var out struct{}
if err := a.RPC("Status.Ping", struct{}{}, &out); err != nil {
t.Fatalf("err: %v", err)
}
}
func TestAgent_TokenStore(t *testing.T) {
if testing.Short() {
t.Skip("too slow for testing.Short")
}
t.Parallel()
a := NewTestAgent(t, `
acl {
tokens {
default = "user"
agent = "agent"
agent_recovery = "recovery"
}
}
`)
defer a.Shutdown()
if got, want := a.tokens.UserToken(), "user"; got != want {
t.Fatalf("got %q want %q", got, want)
}
if got, want := a.tokens.AgentToken(), "agent"; got != want {
t.Fatalf("got %q want %q", got, want)
}
if got, want := a.tokens.IsAgentRecoveryToken("recovery"), true; got != want {
t.Fatalf("got %v want %v", got, want)
}
}
func TestAgent_ReconnectConfigSettings(t *testing.T) {
if testing.Short() {
t.Skip("too slow for testing.Short")
}
t.Parallel()
func() {
a := NewTestAgent(t, "")
defer a.Shutdown()
lan := a.consulConfig().SerfLANConfig.ReconnectTimeout
if lan != 3*24*time.Hour {
t.Fatalf("bad: %s", lan.String())
}
wan := a.consulConfig().SerfWANConfig.ReconnectTimeout
if wan != 3*24*time.Hour {
t.Fatalf("bad: %s", wan.String())
}
}()
func() {
a := NewTestAgent(t, `
reconnect_timeout = "24h"
reconnect_timeout_wan = "36h"
`)
defer a.Shutdown()
lan := a.consulConfig().SerfLANConfig.ReconnectTimeout
if lan != 24*time.Hour {
t.Fatalf("bad: %s", lan.String())
}
wan := a.consulConfig().SerfWANConfig.ReconnectTimeout
if wan != 36*time.Hour {
t.Fatalf("bad: %s", wan.String())
}
}()
}
func TestAgent_HTTPMaxHeaderBytes(t *testing.T) {
tests := []struct {
name string
maxHeaderBytes int
expectedHTTPResponse int
}{
{
"max header bytes 1 returns 431 http response when too large headers are sent",
1,
431,
},
{
"max header bytes 0 returns 200 http response, as the http.DefaultMaxHeaderBytes size of 1MB is used",
0,
200,
},
{
"negative maxHeaderBytes returns 200 http response, as the http.DefaultMaxHeaderBytes size of 1MB is used",
-10,
200,
},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
caConfig := tlsutil.Config{}
tlsConf, err := tlsutil.NewConfigurator(caConfig, hclog.New(nil))
require.NoError(t, err)
bd := BaseDeps{
Deps: consul.Deps{
Logger: hclog.NewInterceptLogger(nil),
Tokens: new(token.Store),
TLSConfigurator: tlsConf,
GRPCConnPool: &fakeGRPCConnPool{},
},
RuntimeConfig: &config.RuntimeConfig{
HTTPAddrs: []net.Addr{
&net.TCPAddr{IP: net.ParseIP("127.0.0.1"), Port: freeport.GetOne(t)},
},
HTTPMaxHeaderBytes: tt.maxHeaderBytes,
},
Cache: cache.New(cache.Options{}),
}
bd, err = initEnterpriseBaseDeps(bd, nil)
require.NoError(t, err)
a, err := New(bd)
require.NoError(t, err)
a.startLicenseManager(testutil.TestContext(t))
srvs, err := a.listenHTTP()
require.NoError(t, err)
require.Equal(t, tt.maxHeaderBytes, a.config.HTTPMaxHeaderBytes)
ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
t.Cleanup(cancel)
g := new(errgroup.Group)
for _, s := range srvs {
g.Go(s.Run)
}
require.Len(t, srvs, 1)
client := &http.Client{}
for _, s := range srvs {
u := url.URL{Scheme: s.Protocol, Host: s.Addr.String()}
req, err := http.NewRequest(http.MethodGet, u.String(), nil)
require.NoError(t, err)
// This is directly pulled from the testing of request limits in the net/http source
// https://github.com/golang/go/blob/go1.15.3/src/net/http/serve_test.go#L2897-L2900
var bytesPerHeader = len("header12345: val12345\r\n")
for i := 0; i < ((tt.maxHeaderBytes+4096)/bytesPerHeader)+1; i++ {
req.Header.Set(fmt.Sprintf("header%05d", i), fmt.Sprintf("val%05d", i))
}
resp, err := client.Do(req.WithContext(ctx))
require.NoError(t, err)
require.Equal(t, tt.expectedHTTPResponse, resp.StatusCode, "expected a '%d' http response, got '%d'", tt.expectedHTTPResponse, resp.StatusCode)
}
})
}
}
type fakeGRPCConnPool struct{}
func (f fakeGRPCConnPool) ClientConn(_ string) (*grpc.ClientConn, error) {
return nil, nil
}
func (f fakeGRPCConnPool) ClientConnLeader() (*grpc.ClientConn, error) {
return nil, nil
}
func (f fakeGRPCConnPool) SetGatewayResolver(_ func(string) string) {
}
func TestAgent_ReconnectConfigWanDisabled(t *testing.T) {
if testing.Short() {
t.Skip("too slow for testing.Short")
}
t.Parallel()
a := NewTestAgent(t, `
ports { serf_wan = -1 }
reconnect_timeout_wan = "36h"
`)
defer a.Shutdown()
// This is also testing that we dont panic like before #4515
require.Nil(t, a.consulConfig().SerfWANConfig)
}
func TestAgent_AddService(t *testing.T) {
if testing.Short() {
t.Skip("too slow for testing.Short")
}
t.Run("normal", func(t *testing.T) {
t.Parallel()
testAgent_AddService(t, "enable_central_service_config = false")
})
t.Run("service manager", func(t *testing.T) {
t.Parallel()
testAgent_AddService(t, "enable_central_service_config = true")
})
}
func testAgent_AddService(t *testing.T, extraHCL string) {
t.Helper()
a := NewTestAgent(t, `
node_name = "node1"
`+extraHCL)
defer a.Shutdown()
duration3s, _ := time.ParseDuration("3s")
duration10s, _ := time.ParseDuration("10s")
tests := []struct {
desc string
srv *structs.NodeService
wantSrv func(ns *structs.NodeService)
chkTypes []*structs.CheckType
healthChks map[string]*structs.HealthCheck
}{
{
"one check",
&structs.NodeService{
ID: "svcid1",
Service: "svcname1",
Tags: []string{"tag1"},
Weights: nil, // nil weights...
Port: 8100,
EnterpriseMeta: *structs.DefaultEnterpriseMetaInDefaultPartition(),
},
// ... should be populated to avoid "IsSame" returning true during AE.
func(ns *structs.NodeService) {
ns.Weights = &structs.Weights{
Passing: 1,
Warning: 1,
}
},
[]*structs.CheckType{
{
CheckID: "check1",
Name: "name1",
TTL: time.Minute,
Notes: "note1",
},
},
map[string]*structs.HealthCheck{
"check1": {
Node: "node1",
CheckID: "check1",
Name: "name1",
Interval: "",
Timeout: "", // these are empty because a TTL was provided
Status: "critical",
Notes: "note1",
ServiceID: "svcid1",
ServiceName: "svcname1",
ServiceTags: []string{"tag1"},
Type: "ttl",
EnterpriseMeta: *structs.DefaultEnterpriseMetaInDefaultPartition(),
},
},
},
{
"one http check with interval and duration",
&structs.NodeService{
ID: "svcid1",
Service: "svcname1",
Tags: []string{"tag1"},
Weights: nil, // nil weights...
Port: 8100,
EnterpriseMeta: *structs.DefaultEnterpriseMetaInDefaultPartition(),
},
// ... should be populated to avoid "IsSame" returning true during AE.
func(ns *structs.NodeService) {
ns.Weights = &structs.Weights{
Passing: 1,
Warning: 1,
}
},
[]*structs.CheckType{
{
CheckID: "check1",
Name: "name1",
HTTP: "http://localhost:8100/",
Interval: duration10s,
Timeout: duration3s,
Notes: "note1",
},
},
map[string]*structs.HealthCheck{
"check1": {
Node: "node1",
CheckID: "check1",
Name: "name1",
Interval: "10s",
Timeout: "3s",
Status: "critical",
Notes: "note1",
ServiceID: "svcid1",
ServiceName: "svcname1",
ServiceTags: []string{"tag1"},
Type: "http",
EnterpriseMeta: *structs.DefaultEnterpriseMetaInDefaultPartition(),
},
},
},
{
"multiple checks",
&structs.NodeService{
ID: "svcid2",
Service: "svcname2",
Weights: &structs.Weights{
Passing: 2,
Warning: 1,
},
Tags: []string{"tag2"},
Port: 8200,
EnterpriseMeta: *structs.DefaultEnterpriseMetaInDefaultPartition(),
},
nil, // No change expected
[]*structs.CheckType{
{
CheckID: "check1",
Name: "name1",
TTL: time.Minute,
Notes: "note1",
},
{
CheckID: "check-noname",
TTL: time.Minute,
},
{
Name: "check-noid",
TTL: time.Minute,
},
{
TTL: time.Minute,
},
},
map[string]*structs.HealthCheck{
"check1": {
Node: "node1",
CheckID: "check1",
Name: "name1",
Interval: "",
Timeout: "", // these are empty bcause a TTL was provided
Status: "critical",
Notes: "note1",
ServiceID: "svcid2",
ServiceName: "svcname2",
ServiceTags: []string{"tag2"},
Type: "ttl",
EnterpriseMeta: *structs.DefaultEnterpriseMetaInDefaultPartition(),
},
"check-noname": {
Node: "node1",
CheckID: "check-noname",
Name: "Service 'svcname2' check",
Interval: "",
Timeout: "", // these are empty because a TTL was provided
Status: "critical",
ServiceID: "svcid2",
ServiceName: "svcname2",
ServiceTags: []string{"tag2"},
Type: "ttl",
EnterpriseMeta: *structs.DefaultEnterpriseMetaInDefaultPartition(),
},
"service:svcid2:3": {
Node: "node1",
CheckID: "service:svcid2:3",
Name: "check-noid",
Interval: "",
Timeout: "", // these are empty becuase a TTL was provided
Status: "critical",
ServiceID: "svcid2",
ServiceName: "svcname2",
ServiceTags: []string{"tag2"},
Type: "ttl",
EnterpriseMeta: *structs.DefaultEnterpriseMetaInDefaultPartition(),
},
"service:svcid2:4": {
Node: "node1",
CheckID: "service:svcid2:4",
Name: "Service 'svcname2' check",
Interval: "",
Timeout: "", // these are empty because a TTL was provided
Status: "critical",
ServiceID: "svcid2",
ServiceName: "svcname2",
ServiceTags: []string{"tag2"},
Type: "ttl",
EnterpriseMeta: *structs.DefaultEnterpriseMetaInDefaultPartition(),
},
},
},
}
for _, tt := range tests {
t.Run(tt.desc, func(t *testing.T) {
// check the service registration
t.Run(tt.srv.ID, func(t *testing.T) {
err := a.addServiceFromSource(tt.srv, tt.chkTypes, false, "", ConfigSourceLocal)
if err != nil {
t.Fatalf("err: %v", err)
}
got := getService(a, tt.srv.ID)
// Make a copy since the tt.srv points to the one in memory in the local
// state still so changing it is a tautology!
want := *tt.srv
if tt.wantSrv != nil {
tt.wantSrv(&want)
}
require.Equal(t, &want, got)
require.True(t, got.IsSame(&want))
})
// check the health checks
for k, v := range tt.healthChks {
t.Run(k, func(t *testing.T) {
got := getCheck(a, types.CheckID(k))
require.Equal(t, v, got)
})
}
// check the ttl checks
for k := range tt.healthChks {
t.Run(k+" ttl", func(t *testing.T) {
chk := a.checkTTLs[structs.NewCheckID(types.CheckID(k), nil)]
if chk == nil {
t.Fatal("got nil want TTL check")
}
if got, want := string(chk.CheckID.ID), k; got != want {
t.Fatalf("got CheckID %v want %v", got, want)
}
if got, want := chk.TTL, time.Minute; got != want {
t.Fatalf("got TTL %v want %v", got, want)
}
})
}
})
}
}
// addServiceFromSource is a test helper that exists to maintain an old function
// signature that was used in many tests.
// Deprecated: use AddService
func (a *Agent) addServiceFromSource(service *structs.NodeService, chkTypes []*structs.CheckType, persist bool, token string, source configSource) error {
return a.AddService(AddServiceRequest{
Service: service,
chkTypes: chkTypes,
persist: persist,
token: token,
replaceExistingChecks: false,
Source: source,
})
}
func TestAgent_AddServices_AliasUpdateCheckNotReverted(t *testing.T) {
if testing.Short() {
t.Skip("too slow for testing.Short")
}
t.Run("normal", func(t *testing.T) {
t.Parallel()
testAgent_AddServices_AliasUpdateCheckNotReverted(t, "enable_central_service_config = false")
})
t.Run("service manager", func(t *testing.T) {
t.Parallel()
testAgent_AddServices_AliasUpdateCheckNotReverted(t, "enable_central_service_config = true")
})
}
func testAgent_AddServices_AliasUpdateCheckNotReverted(t *testing.T, extraHCL string) {
t.Helper()
a := NewTestAgent(t, `
node_name = "node1"
`+extraHCL)
defer a.Shutdown()
// It's tricky to get an UpdateCheck call to be timed properly so it lands
// right in the middle of an addServiceInternal call so we cheat a bit and
// rely upon alias checks to do that work for us. We add enough services
// that probabilistically one of them is going to end up properly in the
// critical section.
//
// The first number I picked here (10) surprisingly failed every time prior
// to PR #6144 solving the underlying problem.
const numServices = 10
services := make([]*structs.ServiceDefinition, numServices)
checkIDs := make([]types.CheckID, numServices)
services[0] = &structs.ServiceDefinition{
ID: "fake",
Name: "fake",
Port: 8080,
Checks: []*structs.CheckType{},
}
for i := 1; i < numServices; i++ {
name := fmt.Sprintf("web-%d", i)
services[i] = &structs.ServiceDefinition{
ID: name,
Name: name,
Port: 8080 + i,
Checks: []*structs.CheckType{
{
Name: "alias-for-fake-service",
AliasService: "fake",
},
},
}
checkIDs[i] = types.CheckID("service:" + name)
}
// Add all of the services quickly as you might do from config file snippets.
for _, service := range services {
ns := service.NodeService()
chkTypes, err := service.CheckTypes()
require.NoError(t, err)
require.NoError(t, a.addServiceFromSource(ns, chkTypes, false, service.Token, ConfigSourceLocal))
}
retry.Run(t, func(r *retry.R) {
gotChecks := a.State.Checks(nil)
for id, check := range gotChecks {
require.Equal(r, "passing", check.Status, "check %q is wrong", id)
require.Equal(r, "No checks found.", check.Output, "check %q is wrong", id)
}
})
}
func test_createAlias(t *testing.T, agent *TestAgent, chk *structs.CheckType, expectedResult string) func(r *retry.R) {
t.Helper()
serviceNum := rand.Int()
srv := &structs.NodeService{
Service: fmt.Sprintf("serviceAlias-%d", serviceNum),
Tags: []string{"tag1"},
Port: 8900 + serviceNum,
}
if srv.ID == "" {
srv.ID = fmt.Sprintf("serviceAlias-%d", serviceNum)
}
chk.Status = api.HealthWarning
if chk.CheckID == "" {
chk.CheckID = types.CheckID(fmt.Sprintf("check-%d", serviceNum))
}
err := agent.addServiceFromSource(srv, []*structs.CheckType{chk}, false, "", ConfigSourceLocal)
assert.NoError(t, err)
return func(r *retry.R) {
t.Helper()
found := false
for _, c := range agent.State.CheckStates(structs.WildcardEnterpriseMetaInDefaultPartition()) {
if c.Check.CheckID == chk.CheckID {
found = true
assert.Equal(t, expectedResult, c.Check.Status, "Check state should be %s, was %s in %#v", expectedResult, c.Check.Status, c.Check)
srvID := structs.NewServiceID(srv.ID, structs.WildcardEnterpriseMetaInDefaultPartition())
if err := agent.Agent.State.RemoveService(srvID); err != nil {
fmt.Println("[DEBUG] Fail to remove service", srvID, ", err:=", err)
}
fmt.Println("[DEBUG] Service Removed", srvID, ", err:=", err)
break
}
}
assert.True(t, found)
}
}
// TestAgent_CheckAliasRPC test the Alias Check to be properly sync remotely
// and locally.
// It contains a few hacks such as unlockIndexOnNode because watch performed
// in CheckAlias.runQuery() waits for 1 min, so Shutdoww the agent might take time
// So, we ensure the agent will update regularilly the index
func TestAgent_CheckAliasRPC(t *testing.T) {
if testing.Short() {
t.Skip("too slow for testing.Short")
}
t.Helper()
a := NewTestAgent(t, `
node_name = "node1"
`)
srv := &structs.NodeService{
ID: "svcid1",
Service: "svcname1",
Tags: []string{"tag1"},
Port: 8100,
}
unlockIndexOnNode := func() {
// We ensure to not block and update Agent's index
srv.Tags = []string{fmt.Sprintf("tag-%s", time.Now())}
assert.NoError(t, a.waitForUp())
err := a.addServiceFromSource(srv, []*structs.CheckType{}, false, "", ConfigSourceLocal)
assert.NoError(t, err)
}
shutdownAgent := func() {
// This is to be sure Alias Checks on remote won't be blocked during 1 min
unlockIndexOnNode()
fmt.Println("[DEBUG] STARTING shutdown for TestAgent_CheckAliasRPC", time.Now())
go a.Shutdown()
unlockIndexOnNode()
fmt.Println("[DEBUG] DONE shutdown for TestAgent_CheckAliasRPC", time.Now())
}
defer shutdownAgent()
testrpc.WaitForTestAgent(t, a.RPC, "dc1")
assert.NoError(t, a.waitForUp())
err := a.addServiceFromSource(srv, []*structs.CheckType{}, false, "", ConfigSourceLocal)
assert.NoError(t, err)
retry.Run(t, func(r *retry.R) {
t.Helper()
var args structs.NodeSpecificRequest
args.Datacenter = "dc1"
args.Node = "node1"
args.AllowStale = true
var out structs.IndexedNodeServices
err := a.RPC("Catalog.NodeServices", &args, &out)
assert.NoError(r, err)
foundService := false
lookup := structs.NewServiceID("svcid1", structs.WildcardEnterpriseMetaInDefaultPartition())
for _, srv := range out.NodeServices.Services {
if lookup.Matches(srv.CompoundServiceID()) {
foundService = true
}
}
assert.True(r, foundService, "could not find svcid1 in %#v", out.NodeServices.Services)
})
checks := make([](func(*retry.R)), 0)
checks = append(checks, test_createAlias(t, a, &structs.CheckType{
Name: "Check_Local_Ok",
AliasService: "svcid1",
}, api.HealthPassing))
checks = append(checks, test_createAlias(t, a, &structs.CheckType{
Name: "Check_Local_Fail",
AliasService: "svcidNoExistingID",
}, api.HealthCritical))
checks = append(checks, test_createAlias(t, a, &structs.CheckType{
Name: "Check_Remote_Host_Ok",
AliasNode: "node1",
AliasService: "svcid1",
}, api.HealthPassing))
checks = append(checks, test_createAlias(t, a, &structs.CheckType{
Name: "Check_Remote_Host_Non_Existing_Service",
AliasNode: "node1",
AliasService: "svcidNoExistingID",
}, api.HealthCritical))
// We wait for max 5s for all checks to be in sync
{
for i := 0; i < 50; i++ {
unlockIndexOnNode()
allNonWarning := true
for _, chk := range a.State.Checks(structs.WildcardEnterpriseMetaInDefaultPartition()) {
if chk.Status == api.HealthWarning {
allNonWarning = false
}
}
if allNonWarning {
break
} else {
time.Sleep(100 * time.Millisecond)
}
}
}
for _, toRun := range checks {
unlockIndexOnNode()
retry.Run(t, toRun)
}
}
func TestAgent_AddServiceWithH2PINGCheck(t *testing.T) {
t.Parallel()
a := NewTestAgent(t, "")
defer a.Shutdown()
check := []*structs.CheckType{
{
CheckID: "test-h2ping-check",
Name: "test-h2ping-check",
H2PING: "localhost:12345",
TLSSkipVerify: true,
Interval: 10 * time.Second,
},
}
nodeService := &structs.NodeService{
ID: "test-h2ping-check-service",
Service: "test-h2ping-check-service",
}
err := a.addServiceFromSource(nodeService, check, false, "", ConfigSourceLocal)
if err != nil {
t.Fatalf("Error registering service: %v", err)
}
requireCheckExists(t, a, "test-h2ping-check")
}
func TestAgent_AddServiceWithH2CPINGCheck(t *testing.T) {
t.Parallel()
a := NewTestAgent(t, "")
defer a.Shutdown()
check := []*structs.CheckType{
{
CheckID: "test-h2cping-check",
Name: "test-h2cping-check",
H2PING: "localhost:12345",
TLSSkipVerify: true,
Interval: 10 * time.Second,
H2PingUseTLS: false,
},
}
nodeService := &structs.NodeService{
ID: "test-h2cping-check-service",
Service: "test-h2cping-check-service",
}
err := a.addServiceFromSource(nodeService, check, false, "", ConfigSourceLocal)
if err != nil {
t.Fatalf("Error registering service: %v", err)
}
requireCheckExists(t, a, "test-h2cping-check")
}
func TestAgent_AddServiceNoExec(t *testing.T) {
if testing.Short() {
t.Skip("too slow for testing.Short")
}
t.Run("normal", func(t *testing.T) {
t.Parallel()
testAgent_AddServiceNoExec(t, "enable_central_service_config = false")
})
t.Run("service manager", func(t *testing.T) {
t.Parallel()
testAgent_AddServiceNoExec(t, "enable_central_service_config = true")
})
}
func testAgent_AddServiceNoExec(t *testing.T, extraHCL string) {
t.Helper()
a := NewTestAgent(t, `
node_name = "node1"
`+extraHCL)
defer a.Shutdown()
testrpc.WaitForTestAgent(t, a.RPC, "dc1")
srv := &structs.NodeService{
ID: "svcid1",
Service: "svcname1",
Tags: []string{"tag1"},
Port: 8100,
}
chk := &structs.CheckType{
ScriptArgs: []string{"exit", "0"},
Interval: 15 * time.Second,
}
err := a.addServiceFromSource(srv, []*structs.CheckType{chk}, false, "", ConfigSourceLocal)
if err == nil || !strings.Contains(err.Error(), "Scripts are disabled on this agent") {
t.Fatalf("err: %v", err)
}
err = a.addServiceFromSource(srv, []*structs.CheckType{chk}, false, "", ConfigSourceRemote)
if err == nil || !strings.Contains(err.Error(), "Scripts are disabled on this agent") {
t.Fatalf("err: %v", err)
}
}
func TestAgent_AddServiceNoRemoteExec(t *testing.T) {
if testing.Short() {
t.Skip("too slow for testing.Short")
}
t.Run("normal", func(t *testing.T) {
t.Parallel()
testAgent_AddServiceNoRemoteExec(t, "enable_central_service_config = false")
})
t.Run("service manager", func(t *testing.T) {
t.Parallel()
testAgent_AddServiceNoRemoteExec(t, "enable_central_service_config = true")
})
}
func testAgent_AddServiceNoRemoteExec(t *testing.T, extraHCL string) {
t.Helper()
a := NewTestAgent(t, `
node_name = "node1"
enable_local_script_checks = true
`+extraHCL)
defer a.Shutdown()
testrpc.WaitForTestAgent(t, a.RPC, "dc1")
srv := &structs.NodeService{
ID: "svcid1",
Service: "svcname1",
Tags: []string{"tag1"},
Port: 8100,
}
chk := &structs.CheckType{
ScriptArgs: []string{"exit", "0"},
Interval: 15 * time.Second,
}
err := a.addServiceFromSource(srv, []*structs.CheckType{chk}, false, "", ConfigSourceRemote)
if err == nil || !strings.Contains(err.Error(), "Scripts are disabled on this agent") {
t.Fatalf("err: %v", err)
}
}
func TestAddServiceIPv4TaggedDefault(t *testing.T) {
if testing.Short() {
t.Skip("too slow for testing.Short")
}
t.Helper()
a := NewTestAgent(t, "")
defer a.Shutdown()
testrpc.WaitForTestAgent(t, a.RPC, "dc1")
srv := &structs.NodeService{
Service: "my_service",
ID: "my_service_id",
Port: 8100,
Address: "10.0.1.2",
}
err := a.addServiceFromSource(srv, []*structs.CheckType{}, false, "", ConfigSourceRemote)
require.Nil(t, err)
ns := a.State.Service(structs.NewServiceID("my_service_id", nil))
require.NotNil(t, ns)
svcAddr := structs.ServiceAddress{Address: srv.Address, Port: srv.Port}
require.Equal(t, svcAddr, ns.TaggedAddresses[structs.TaggedAddressLANIPv4])
require.Equal(t, svcAddr, ns.TaggedAddresses[structs.TaggedAddressWANIPv4])
_, ok := ns.TaggedAddresses[structs.TaggedAddressLANIPv6]
require.False(t, ok)
_, ok = ns.TaggedAddresses[structs.TaggedAddressWANIPv6]
require.False(t, ok)
}
func TestAddServiceIPv6TaggedDefault(t *testing.T) {
if testing.Short() {
t.Skip("too slow for testing.Short")
}
t.Helper()
a := NewTestAgent(t, "")
defer a.Shutdown()
testrpc.WaitForTestAgent(t, a.RPC, "dc1")
srv := &structs.NodeService{
Service: "my_service",
ID: "my_service_id",
Port: 8100,
Address: "::5",
}
err := a.addServiceFromSource(srv, []*structs.CheckType{}, false, "", ConfigSourceRemote)
require.Nil(t, err)
ns := a.State.Service(structs.NewServiceID("my_service_id", nil))
require.NotNil(t, ns)
svcAddr := structs.ServiceAddress{Address: srv.Address, Port: srv.Port}
require.Equal(t, svcAddr, ns.TaggedAddresses[structs.TaggedAddressLANIPv6])
require.Equal(t, svcAddr, ns.TaggedAddresses[structs.TaggedAddressWANIPv6])
_, ok := ns.TaggedAddresses[structs.TaggedAddressLANIPv4]
require.False(t, ok)
_, ok = ns.TaggedAddresses[structs.TaggedAddressWANIPv4]
require.False(t, ok)
}
func TestAddServiceIPv4TaggedSet(t *testing.T) {
if testing.Short() {
t.Skip("too slow for testing.Short")
}
t.Helper()
a := NewTestAgent(t, "")
defer a.Shutdown()
testrpc.WaitForTestAgent(t, a.RPC, "dc1")
srv := &structs.NodeService{
Service: "my_service",
ID: "my_service_id",
Port: 8100,
Address: "10.0.1.2",
TaggedAddresses: map[string]structs.ServiceAddress{
structs.TaggedAddressWANIPv4: {
Address: "10.100.200.5",
Port: 8100,
},
},
}
err := a.addServiceFromSource(srv, []*structs.CheckType{}, false, "", ConfigSourceRemote)
require.Nil(t, err)
ns := a.State.Service(structs.NewServiceID("my_service_id", nil))
require.NotNil(t, ns)
svcAddr := structs.ServiceAddress{Address: srv.Address, Port: srv.Port}
require.Equal(t, svcAddr, ns.TaggedAddresses[structs.TaggedAddressLANIPv4])
require.Equal(t, structs.ServiceAddress{Address: "10.100.200.5", Port: 8100}, ns.TaggedAddresses[structs.TaggedAddressWANIPv4])
_, ok := ns.TaggedAddresses[structs.TaggedAddressLANIPv6]
require.False(t, ok)
_, ok = ns.TaggedAddresses[structs.TaggedAddressWANIPv6]
require.False(t, ok)
}
func TestAddServiceIPv6TaggedSet(t *testing.T) {
if testing.Short() {
t.Skip("too slow for testing.Short")
}
t.Helper()
a := NewTestAgent(t, "")
defer a.Shutdown()
testrpc.WaitForTestAgent(t, a.RPC, "dc1")
srv := &structs.NodeService{
Service: "my_service",
ID: "my_service_id",
Port: 8100,
Address: "::5",
TaggedAddresses: map[string]structs.ServiceAddress{
structs.TaggedAddressWANIPv6: {
Address: "::6",
Port: 8100,
},
},
}
err := a.addServiceFromSource(srv, []*structs.CheckType{}, false, "", ConfigSourceRemote)
require.Nil(t, err)
ns := a.State.Service(structs.NewServiceID("my_service_id", nil))
require.NotNil(t, ns)
svcAddr := structs.ServiceAddress{Address: srv.Address, Port: srv.Port}
require.Equal(t, svcAddr, ns.TaggedAddresses[structs.TaggedAddressLANIPv6])
require.Equal(t, structs.ServiceAddress{Address: "::6", Port: 8100}, ns.TaggedAddresses[structs.TaggedAddressWANIPv6])
_, ok := ns.TaggedAddresses[structs.TaggedAddressLANIPv4]
require.False(t, ok)
_, ok = ns.TaggedAddresses[structs.TaggedAddressWANIPv4]
require.False(t, ok)
}
func TestAgent_RemoveService(t *testing.T) {
if testing.Short() {
t.Skip("too slow for testing.Short")
}
t.Run("normal", func(t *testing.T) {
t.Parallel()
testAgent_RemoveService(t, "enable_central_service_config = false")
})
t.Run("service manager", func(t *testing.T) {
t.Parallel()
testAgent_RemoveService(t, "enable_central_service_config = true")
})
}
func testAgent_RemoveService(t *testing.T, extraHCL string) {
t.Helper()
a := NewTestAgent(t, extraHCL)
defer a.Shutdown()
// Remove a service that doesn't exist
if err := a.RemoveService(structs.NewServiceID("redis", nil)); err != nil {
t.Fatalf("err: %v", err)
}
// Remove without an ID
if err := a.RemoveService(structs.NewServiceID("", nil)); err == nil {
t.Fatalf("should have errored")
}
// Removing a service with a single check works
{
srv := &structs.NodeService{
ID: "memcache",
Service: "memcache",
Port: 8000,
}
chkTypes := []*structs.CheckType{{TTL: time.Minute}}
if err := a.addServiceFromSource(srv, chkTypes, false, "", ConfigSourceLocal); err != nil {
t.Fatalf("err: %v", err)
}
// Add a check after the fact with a specific check ID
check := &structs.CheckDefinition{
ID: "check2",
Name: "check2",
ServiceID: "memcache",
TTL: time.Minute,
}
hc := check.HealthCheck("node1")
if err := a.AddCheck(hc, check.CheckType(), false, "", ConfigSourceLocal); err != nil {
t.Fatalf("err: %s", err)
}
if err := a.RemoveService(structs.NewServiceID("memcache", nil)); err != nil {
t.Fatalf("err: %s", err)
}
require.Nil(t, a.State.Check(structs.NewCheckID("service:memcache", nil)), "have memcache check")
require.Nil(t, a.State.Check(structs.NewCheckID("check2", nil)), "have check2 check")
}
// Removing a service with multiple checks works
{
// add a service to remove
srv := &structs.NodeService{
ID: "redis",
Service: "redis",
Port: 8000,
}
chkTypes := []*structs.CheckType{
{TTL: time.Minute},
{TTL: 30 * time.Second},
}
if err := a.addServiceFromSource(srv, chkTypes, false, "", ConfigSourceLocal); err != nil {
t.Fatalf("err: %v", err)
}
// add another service that wont be affected
srv = &structs.NodeService{
ID: "mysql",
Service: "mysql",
Port: 3306,
}
chkTypes = []*structs.CheckType{
{TTL: time.Minute},
{TTL: 30 * time.Second},
}
if err := a.addServiceFromSource(srv, chkTypes, false, "", ConfigSourceLocal); err != nil {
t.Fatalf("err: %v", err)
}
// Remove the service
if err := a.RemoveService(structs.NewServiceID("redis", nil)); err != nil {
t.Fatalf("err: %v", err)
}
// Ensure we have a state mapping
requireServiceMissing(t, a, "redis")
// Ensure checks were removed
requireCheckMissing(t, a, "service:redis:1")
requireCheckMissing(t, a, "service:redis:2")
requireCheckMissingMap(t, a.checkTTLs, "service:redis:1")
requireCheckMissingMap(t, a.checkTTLs, "service:redis:2")
// check the mysql service is unnafected
requireCheckExistsMap(t, a.checkTTLs, "service:mysql:1")
requireCheckExists(t, a, "service:mysql:1")
requireCheckExistsMap(t, a.checkTTLs, "service:mysql:2")
requireCheckExists(t, a, "service:mysql:2")
}
}
func TestAgent_RemoveServiceRemovesAllChecks(t *testing.T) {
if testing.Short() {
t.Skip("too slow for testing.Short")
}
t.Run("normal", func(t *testing.T) {
t.Parallel()
testAgent_RemoveServiceRemovesAllChecks(t, "enable_central_service_config = false")
})
t.Run("service manager", func(t *testing.T) {
t.Parallel()
testAgent_RemoveServiceRemovesAllChecks(t, "enable_central_service_config = true")
})
}
func testAgent_RemoveServiceRemovesAllChecks(t *testing.T, extraHCL string) {
t.Helper()
a := NewTestAgent(t, `
node_name = "node1"
`+extraHCL)
defer a.Shutdown()
svc := &structs.NodeService{ID: "redis", Service: "redis", Port: 8000, EnterpriseMeta: *structs.DefaultEnterpriseMetaInDefaultPartition()}
chk1 := &structs.CheckType{CheckID: "chk1", Name: "chk1", TTL: time.Minute}
chk2 := &structs.CheckType{CheckID: "chk2", Name: "chk2", TTL: 2 * time.Minute}
hchk1 := &structs.HealthCheck{
Node: "node1",
CheckID: "chk1",
Name: "chk1",
Status: "critical",
ServiceID: "redis",
ServiceName: "redis",
Type: "ttl",
EnterpriseMeta: *structs.DefaultEnterpriseMetaInDefaultPartition(),
}
hchk2 := &structs.HealthCheck{Node: "node1",
CheckID: "chk2",
Name: "chk2",
Status: "critical",
ServiceID: "redis",
ServiceName: "redis",
Type: "ttl",
EnterpriseMeta: *structs.DefaultEnterpriseMetaInDefaultPartition(),
}
// register service with chk1
if err := a.addServiceFromSource(svc, []*structs.CheckType{chk1}, false, "", ConfigSourceLocal); err != nil {
t.Fatal("Failed to register service", err)
}
// verify chk1 exists
requireCheckExists(t, a, "chk1")
// update the service with chk2
if err := a.addServiceFromSource(svc, []*structs.CheckType{chk2}, false, "", ConfigSourceLocal); err != nil {
t.Fatal("Failed to update service", err)
}
// check that both checks are there
require.Equal(t, hchk1, getCheck(a, "chk1"))
require.Equal(t, hchk2, getCheck(a, "chk2"))
// Remove service
if err := a.RemoveService(structs.NewServiceID("redis", nil)); err != nil {
t.Fatal("Failed to remove service", err)
}
// Check that both checks are gone
requireCheckMissing(t, a, "chk1")
requireCheckMissing(t, a, "chk2")
}
// TestAgent_IndexChurn is designed to detect a class of issues where
// we would have unnecessary catalog churn from anti-entropy. See issues
// #3259, #3642, #3845, and #3866.
func TestAgent_IndexChurn(t *testing.T) {
if testing.Short() {
t.Skip("too slow for testing.Short")
}
t.Parallel()
t.Run("no tags", func(t *testing.T) {
verifyIndexChurn(t, nil)
})
t.Run("with tags", func(t *testing.T) {
verifyIndexChurn(t, []string{"foo", "bar"})
})
}
// verifyIndexChurn registers some things and runs anti-entropy a bunch of times
// in a row to make sure there are no index bumps.
func verifyIndexChurn(t *testing.T, tags []string) {
t.Helper()
a := NewTestAgent(t, "")
defer a.Shutdown()
weights := &structs.Weights{
Passing: 1,
Warning: 1,
}
// Ensure we have a leader before we start adding the services
testrpc.WaitForLeader(t, a.RPC, "dc1")
svc := &structs.NodeService{
ID: "redis",
Service: "redis",
Port: 8000,
Tags: tags,
Weights: weights,
}
if err := a.addServiceFromSource(svc, nil, true, "", ConfigSourceLocal); err != nil {
t.Fatalf("err: %v", err)
}
chk := &structs.HealthCheck{
CheckID: "redis-check",
Name: "Service-level check",
ServiceID: "redis",
Status: api.HealthCritical,
}
chkt := &structs.CheckType{
TTL: time.Hour,
}
if err := a.AddCheck(chk, chkt, true, "", ConfigSourceLocal); err != nil {
t.Fatalf("err: %v", err)
}
chk = &structs.HealthCheck{
CheckID: "node-check",
Name: "Node-level check",
Status: api.HealthCritical,
}
chkt = &structs.CheckType{
TTL: time.Hour,
}
if err := a.AddCheck(chk, chkt, true, "", ConfigSourceLocal); err != nil {
t.Fatalf("err: %v", err)
}
if err := a.sync.State.SyncFull(); err != nil {
t.Fatalf("err: %v", err)
}
args := &structs.ServiceSpecificRequest{
Datacenter: "dc1",
ServiceName: "redis",
}
var before structs.IndexedCheckServiceNodes
// This sleep is so that the serfHealth check is added to the agent
// A value of 375ms is sufficient enough time to ensure the serfHealth
// check is added to an agent. 500ms so that we don't see flakiness ever.
time.Sleep(500 * time.Millisecond)
if err := a.RPC("Health.ServiceNodes", args, &before); err != nil {
t.Fatalf("err: %v", err)
}
for _, name := range before.Nodes[0].Checks {
a.logger.Debug("Registered node", "node", name.Name)
}
if got, want := len(before.Nodes), 1; got != want {
t.Fatalf("got %d want %d", got, want)
}
if got, want := len(before.Nodes[0].Checks), 3; /* incl. serfHealth */ got != want {
t.Fatalf("got %d want %d", got, want)
}
for i := 0; i < 10; i++ {
a.logger.Info("Sync in progress", "iteration", i+1)
if err := a.sync.State.SyncFull(); err != nil {
t.Fatalf("err: %v", err)
}
}
// If this test fails here this means that the Consul-X-Index
// has changed for the RPC, which means that idempotent ops
// are not working as intended.
var after structs.IndexedCheckServiceNodes
if err := a.RPC("Health.ServiceNodes", args, &after); err != nil {
t.Fatalf("err: %v", err)
}
require.Equal(t, before, after)
}
func TestAgent_AddCheck(t *testing.T) {
if testing.Short() {
t.Skip("too slow for testing.Short")
}
t.Parallel()
a := NewTestAgent(t, `
enable_script_checks = true
`)
defer a.Shutdown()
health := &structs.HealthCheck{
Node: "foo",
CheckID: "mem",
Name: "memory util",
Status: api.HealthCritical,
}
chk := &structs.CheckType{
ScriptArgs: []string{"exit", "0"},
Interval: 15 * time.Second,
}
err := a.AddCheck(health, chk, false, "", ConfigSourceLocal)
if err != nil {
t.Fatalf("err: %v", err)
}
// Ensure we have a check mapping
sChk := requireCheckExists(t, a, "mem")
// Ensure our check is in the right state
if sChk.Status != api.HealthCritical {
t.Fatalf("check not critical")
}
// Ensure a TTL is setup
requireCheckExistsMap(t, a.checkMonitors, "mem")
}
func TestAgent_AddCheck_StartPassing(t *testing.T) {
if testing.Short() {
t.Skip("too slow for testing.Short")
}
t.Parallel()
a := NewTestAgent(t, `
enable_script_checks = true
`)
defer a.Shutdown()
health := &structs.HealthCheck{
Node: "foo",
CheckID: "mem",
Name: "memory util",
Status: api.HealthPassing,
}
chk := &structs.CheckType{
ScriptArgs: []string{"exit", "0"},
Interval: 15 * time.Second,
}
err := a.AddCheck(health, chk, false, "", ConfigSourceLocal)
if err != nil {
t.Fatalf("err: %v", err)
}
// Ensure we have a check mapping
sChk := requireCheckExists(t, a, "mem")
// Ensure our check is in the right state
if sChk.Status != api.HealthPassing {
t.Fatalf("check not passing")
}
// Ensure a TTL is setup
requireCheckExistsMap(t, a.checkMonitors, "mem")
}
func TestAgent_AddCheck_MinInterval(t *testing.T) {
if testing.Short() {
t.Skip("too slow for testing.Short")
}
t.Parallel()
a := NewTestAgent(t, `
enable_script_checks = true
`)
defer a.Shutdown()
health := &structs.HealthCheck{
Node: "foo",
CheckID: "mem",
Name: "memory util",
Status: api.HealthCritical,
}
chk := &structs.CheckType{
ScriptArgs: []string{"exit", "0"},
Interval: time.Microsecond,
}
err := a.AddCheck(health, chk, false, "", ConfigSourceLocal)
if err != nil {
t.Fatalf("err: %v", err)
}
// Ensure we have a check mapping
requireCheckExists(t, a, "mem")
// Ensure a TTL is setup
if mon, ok := a.checkMonitors[structs.NewCheckID("mem", nil)]; !ok {
t.Fatalf("missing mem monitor")
} else if mon.Interval != checks.MinInterval {
t.Fatalf("bad mem monitor interval")
}
}
func TestAgent_AddCheck_MissingService(t *testing.T) {
if testing.Short() {
t.Skip("too slow for testing.Short")
}
t.Parallel()
a := NewTestAgent(t, `
enable_script_checks = true
`)
defer a.Shutdown()
health := &structs.HealthCheck{
Node: "foo",
CheckID: "baz",
Name: "baz check 1",
ServiceID: "baz",
}
chk := &structs.CheckType{
ScriptArgs: []string{"exit", "0"},
Interval: time.Microsecond,
}
err := a.AddCheck(health, chk, false, "", ConfigSourceLocal)
if err == nil || err.Error() != fmt.Sprintf("ServiceID %q does not exist", structs.ServiceIDString("baz", nil)) {
t.Fatalf("expected service id error, got: %v", err)
}
}
func TestAgent_AddCheck_RestoreState(t *testing.T) {
if testing.Short() {
t.Skip("too slow for testing.Short")
}
t.Parallel()
a := NewTestAgent(t, "")
defer a.Shutdown()
// Create some state and persist it
ttl := &checks.CheckTTL{
CheckID: structs.NewCheckID("baz", nil),
TTL: time.Minute,
}
err := a.persistCheckState(ttl, api.HealthPassing, "yup")
if err != nil {
t.Fatalf("err: %s", err)
}
// Build and register the check definition and initial state
health := &structs.HealthCheck{
Node: "foo",
CheckID: "baz",
Name: "baz check 1",
}
chk := &structs.CheckType{
TTL: time.Minute,
}
err = a.AddCheck(health, chk, false, "", ConfigSourceLocal)
if err != nil {
t.Fatalf("err: %s", err)
}
// Ensure the check status was restored during registration
check := requireCheckExists(t, a, "baz")
if check.Status != api.HealthPassing {
t.Fatalf("bad: %#v", check)
}
if check.Output != "yup" {
t.Fatalf("bad: %#v", check)
}
}
func TestAgent_AddCheck_ExecDisable(t *testing.T) {
if testing.Short() {
t.Skip("too slow for testing.Short")
}
t.Parallel()
a := NewTestAgent(t, "")
defer a.Shutdown()
health := &structs.HealthCheck{
Node: "foo",
CheckID: "mem",
Name: "memory util",
Status: api.HealthCritical,
}
chk := &structs.CheckType{
ScriptArgs: []string{"exit", "0"},
Interval: 15 * time.Second,
}
err := a.AddCheck(health, chk, false, "", ConfigSourceLocal)
if err == nil || !strings.Contains(err.Error(), "Scripts are disabled on this agent") {
t.Fatalf("err: %v", err)
}
// Ensure we don't have a check mapping
requireCheckMissing(t, a, "mem")
err = a.AddCheck(health, chk, false, "", ConfigSourceRemote)
if err == nil || !strings.Contains(err.Error(), "Scripts are disabled on this agent") {
t.Fatalf("err: %v", err)
}
// Ensure we don't have a check mapping
requireCheckMissing(t, a, "mem")
}
func TestAgent_AddCheck_ExecRemoteDisable(t *testing.T) {
if testing.Short() {
t.Skip("too slow for testing.Short")
}
t.Parallel()
a := NewTestAgent(t, `
enable_local_script_checks = true
`)
defer a.Shutdown()
testrpc.WaitForTestAgent(t, a.RPC, "dc1")
health := &structs.HealthCheck{
Node: "foo",
CheckID: "mem",
Name: "memory util",
Status: api.HealthCritical,
}
chk := &structs.CheckType{
ScriptArgs: []string{"exit", "0"},
Interval: 15 * time.Second,
}
err := a.AddCheck(health, chk, false, "", ConfigSourceRemote)
if err == nil || !strings.Contains(err.Error(), "Scripts are disabled on this agent from remote calls") {
t.Fatalf("err: %v", err)
}
// Ensure we don't have a check mapping
requireCheckMissing(t, a, "mem")
}
func TestAgent_AddCheck_GRPC(t *testing.T) {
if testing.Short() {
t.Skip("too slow for testing.Short")
}
t.Parallel()
a := NewTestAgent(t, "")
defer a.Shutdown()
health := &structs.HealthCheck{
Node: "foo",
CheckID: "grpchealth",
Name: "grpc health checking protocol",
Status: api.HealthCritical,
}
chk := &structs.CheckType{
GRPC: "localhost:12345/package.Service",
Interval: 15 * time.Second,
}
err := a.AddCheck(health, chk, false, "", ConfigSourceLocal)
if err != nil {
t.Fatalf("err: %v", err)
}
// Ensure we have a check mapping
sChk := requireCheckExists(t, a, "grpchealth")
// Ensure our check is in the right state
if sChk.Status != api.HealthCritical {
t.Fatalf("check not critical")
}
// Ensure a check is setup
requireCheckExistsMap(t, a.checkGRPCs, "grpchealth")
}
func TestAgent_RestoreServiceWithAliasCheck(t *testing.T) {
// t.Parallel() don't even think about making this parallel
// This test is very contrived and tests for the absence of race conditions
// related to the implementation of alias checks. As such it is slow,
// serial, full of sleeps and retries, and not generally a great test to
// run all of the time.
//
// That said it made it incredibly easy to root out various race conditions
// quite successfully.
//
// The original set of races was between:
//
// - agent startup reloading Services and Checks from disk
// - API requests to also re-register those same Services and Checks
// - the goroutines for the as-yet-to-be-stopped CheckAlias goroutines
if os.Getenv("SLOWTEST") != "1" {
t.Skip("skipping slow test; set SLOWTEST=1 to run")
return
}
// We do this so that the agent logs and the informational messages from
// the test itself are interwoven properly.
logf := func(t *testing.T, a *TestAgent, format string, args ...interface{}) {
a.logger.Info("testharness: " + fmt.Sprintf(format, args...))
}
cfg := `
server = false
bootstrap = false
enable_central_service_config = false
`
a := StartTestAgent(t, TestAgent{HCL: cfg})
defer a.Shutdown()
handler := http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) {
w.WriteHeader(http.StatusOK)
_, _ = w.Write([]byte("OK\n"))
})
testHTTPServer := httptest.NewServer(handler)
t.Cleanup(testHTTPServer.Close)
registerServicesAndChecks := func(t *testing.T, a *TestAgent) {
// add one persistent service with a simple check
require.NoError(t, a.addServiceFromSource(
&structs.NodeService{
ID: "ping",
Service: "ping",
Port: 8000,
},
[]*structs.CheckType{
{
HTTP: testHTTPServer.URL,
Method: "GET",
Interval: 5 * time.Second,
Timeout: 1 * time.Second,
},
},
true, "", ConfigSourceLocal,
))
// add one persistent sidecar service with an alias check in the manner
// of how sidecar_service would add it
require.NoError(t, a.addServiceFromSource(
&structs.NodeService{
ID: "ping-sidecar-proxy",
Service: "ping-sidecar-proxy",
Port: 9000,
},
[]*structs.CheckType{
{
Name: "Connect Sidecar Aliasing ping",
AliasService: "ping",
},
},
true, "", ConfigSourceLocal,
))
}
retryUntilCheckState := func(t *testing.T, a *TestAgent, checkID string, expectedStatus string) {
t.Helper()
retry.Run(t, func(r *retry.R) {
chk := requireCheckExists(t, a, types.CheckID(checkID))
if chk.Status != expectedStatus {
logf(t, a, "check=%q expected status %q but got %q", checkID, expectedStatus, chk.Status)
r.Fatalf("check=%q expected status %q but got %q", checkID, expectedStatus, chk.Status)
}
logf(t, a, "check %q has reached desired status %q", checkID, expectedStatus)
})
}
registerServicesAndChecks(t, a)
time.Sleep(1 * time.Second)
retryUntilCheckState(t, a, "service:ping", api.HealthPassing)
retryUntilCheckState(t, a, "service:ping-sidecar-proxy", api.HealthPassing)
logf(t, a, "==== POWERING DOWN ORIGINAL ====")
require.NoError(t, a.Shutdown())
time.Sleep(1 * time.Second)
futureHCL := cfg + `
node_id = "` + string(a.Config.NodeID) + `"
node_name = "` + a.Config.NodeName + `"
`
restartOnce := func(idx int, t *testing.T) {
t.Helper()
// Reload and retain former NodeID and data directory.
a2 := StartTestAgent(t, TestAgent{HCL: futureHCL, DataDir: a.DataDir})
defer a2.Shutdown()
a = nil
// reregister during standup; we use an adjustable timing to try and force a race
sleepDur := time.Duration(idx+1) * 500 * time.Millisecond
time.Sleep(sleepDur)
logf(t, a2, "re-registering checks and services after a delay of %v", sleepDur)
for i := 0; i < 20; i++ { // RACE RACE RACE!
registerServicesAndChecks(t, a2)
time.Sleep(50 * time.Millisecond)
}
time.Sleep(1 * time.Second)
retryUntilCheckState(t, a2, "service:ping", api.HealthPassing)
logf(t, a2, "giving the alias check a chance to notice...")
time.Sleep(5 * time.Second)
retryUntilCheckState(t, a2, "service:ping-sidecar-proxy", api.HealthPassing)
}
for i := 0; i < 20; i++ {
name := "restart-" + strconv.Itoa(i)
ok := t.Run(name, func(t *testing.T) {
restartOnce(i, t)
})
require.True(t, ok, name+" failed")
}
}
func TestAgent_AddCheck_Alias(t *testing.T) {
if testing.Short() {
t.Skip("too slow for testing.Short")
}
t.Parallel()
a := NewTestAgent(t, "")
defer a.Shutdown()
health := &structs.HealthCheck{
Node: "foo",
CheckID: "aliashealth",
Name: "Alias health check",
Status: api.HealthCritical,
}
chk := &structs.CheckType{
AliasService: "foo",
}
err := a.AddCheck(health, chk, false, "", ConfigSourceLocal)
require.NoError(t, err)
// Ensure we have a check mapping
sChk := requireCheckExists(t, a, "aliashealth")
require.Equal(t, api.HealthCritical, sChk.Status)
chkImpl, ok := a.checkAliases[structs.NewCheckID("aliashealth", nil)]
require.True(t, ok, "missing aliashealth check")
require.Equal(t, "", chkImpl.RPCReq.Token)
cs := a.State.CheckState(structs.NewCheckID("aliashealth", nil))
require.NotNil(t, cs)
require.Equal(t, "", cs.Token)
}
func TestAgent_AddCheck_Alias_setToken(t *testing.T) {
if testing.Short() {
t.Skip("too slow for testing.Short")
}
t.Parallel()
a := NewTestAgent(t, "")
defer a.Shutdown()
health := &structs.HealthCheck{
Node: "foo",
CheckID: "aliashealth",
Name: "Alias health check",
Status: api.HealthCritical,
}
chk := &structs.CheckType{
AliasService: "foo",
}
err := a.AddCheck(health, chk, false, "foo", ConfigSourceLocal)
require.NoError(t, err)
cs := a.State.CheckState(structs.NewCheckID("aliashealth", nil))
require.NotNil(t, cs)
require.Equal(t, "foo", cs.Token)
chkImpl, ok := a.checkAliases[structs.NewCheckID("aliashealth", nil)]
require.True(t, ok, "missing aliashealth check")
require.Equal(t, "foo", chkImpl.RPCReq.Token)
}
func TestAgent_AddCheck_Alias_userToken(t *testing.T) {
if testing.Short() {
t.Skip("too slow for testing.Short")
}
t.Parallel()
a := NewTestAgent(t, `
acl_token = "hello"
`)
defer a.Shutdown()
health := &structs.HealthCheck{
Node: "foo",
CheckID: "aliashealth",
Name: "Alias health check",
Status: api.HealthCritical,
}
chk := &structs.CheckType{
AliasService: "foo",
}
err := a.AddCheck(health, chk, false, "", ConfigSourceLocal)
require.NoError(t, err)
cs := a.State.CheckState(structs.NewCheckID("aliashealth", nil))
require.NotNil(t, cs)
require.Equal(t, "", cs.Token) // State token should still be empty
chkImpl, ok := a.checkAliases[structs.NewCheckID("aliashealth", nil)]
require.True(t, ok, "missing aliashealth check")
require.Equal(t, "hello", chkImpl.RPCReq.Token) // Check should use the token
}
func TestAgent_AddCheck_Alias_userAndSetToken(t *testing.T) {
if testing.Short() {
t.Skip("too slow for testing.Short")
}
t.Parallel()
a := NewTestAgent(t, `
acl_token = "hello"
`)
defer a.Shutdown()
health := &structs.HealthCheck{
Node: "foo",
CheckID: "aliashealth",
Name: "Alias health check",
Status: api.HealthCritical,
}
chk := &structs.CheckType{
AliasService: "foo",
}
err := a.AddCheck(health, chk, false, "goodbye", ConfigSourceLocal)
require.NoError(t, err)
cs := a.State.CheckState(structs.NewCheckID("aliashealth", nil))
require.NotNil(t, cs)
require.Equal(t, "goodbye", cs.Token)
chkImpl, ok := a.checkAliases[structs.NewCheckID("aliashealth", nil)]
require.True(t, ok, "missing aliashealth check")
require.Equal(t, "goodbye", chkImpl.RPCReq.Token)
}
func TestAgent_RemoveCheck(t *testing.T) {
if testing.Short() {
t.Skip("too slow for testing.Short")
}
t.Parallel()
a := NewTestAgent(t, `
enable_script_checks = true
`)
defer a.Shutdown()
// Remove check that doesn't exist
if err := a.RemoveCheck(structs.NewCheckID("mem", nil), false); err != nil {
t.Fatalf("err: %v", err)
}
// Remove without an ID
if err := a.RemoveCheck(structs.NewCheckID("", nil), false); err == nil {
t.Fatalf("should have errored")
}
health := &structs.HealthCheck{
Node: "foo",
CheckID: "mem",
Name: "memory util",
Status: api.HealthCritical,
}
chk := &structs.CheckType{
ScriptArgs: []string{"exit", "0"},
Interval: 15 * time.Second,
}
err := a.AddCheck(health, chk, false, "", ConfigSourceLocal)
if err != nil {
t.Fatalf("err: %v", err)
}
// Remove check
if err := a.RemoveCheck(structs.NewCheckID("mem", nil), false); err != nil {
t.Fatalf("err: %v", err)
}
// Ensure we have a check mapping
requireCheckMissing(t, a, "mem")
// Ensure a TTL is setup
requireCheckMissingMap(t, a.checkMonitors, "mem")
}
func TestAgent_HTTPCheck_TLSSkipVerify(t *testing.T) {
if testing.Short() {
t.Skip("too slow for testing.Short")
}
t.Parallel()
handler := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
fmt.Fprintln(w, "GOOD")
})
server := httptest.NewTLSServer(handler)
defer server.Close()
a := NewTestAgent(t, "")
defer a.Shutdown()
health := &structs.HealthCheck{
Node: "foo",
CheckID: "tls",
Name: "tls check",
Status: api.HealthCritical,
}
chk := &structs.CheckType{
HTTP: server.URL,
Interval: 20 * time.Millisecond,
TLSSkipVerify: true,
}
err := a.AddCheck(health, chk, false, "", ConfigSourceLocal)
if err != nil {
t.Fatalf("err: %v", err)
}
retry.Run(t, func(r *retry.R) {
status := getCheck(a, "tls")
if status.Status != api.HealthPassing {
r.Fatalf("bad: %v", status.Status)
}
if !strings.Contains(status.Output, "GOOD") {
r.Fatalf("bad: %v", status.Output)
}
})
}
func TestAgent_HTTPCheck_EnableAgentTLSForChecks(t *testing.T) {
if testing.Short() {
t.Skip("too slow for testing.Short")
}
t.Parallel()
run := func(t *testing.T, ca string) {
a := StartTestAgent(t, TestAgent{
UseHTTPS: true,
HCL: `
enable_agent_tls_for_checks = true
verify_incoming = true
server_name = "consul.test"
key_file = "../test/client_certs/server.key"
cert_file = "../test/client_certs/server.crt"
` + ca,
})
defer a.Shutdown()
health := &structs.HealthCheck{
Node: "foo",
CheckID: "tls",
Name: "tls check",
Status: api.HealthCritical,
}
addr, err := firstAddr(a.Agent.apiServers, "https")
require.NoError(t, err)
url := fmt.Sprintf("https://%s/v1/agent/self", addr.String())
chk := &structs.CheckType{
HTTP: url,
Interval: 20 * time.Millisecond,
}
err = a.AddCheck(health, chk, false, "", ConfigSourceLocal)
if err != nil {
t.Fatalf("err: %v", err)
}
retry.Run(t, func(r *retry.R) {
status := getCheck(a, "tls")
if status.Status != api.HealthPassing {
r.Fatalf("bad: %v", status.Status)
}
if !strings.Contains(status.Output, "200 OK") {
r.Fatalf("bad: %v", status.Output)
}
})
}
// We need to test both methods of passing the CA info to ensure that
// we propagate all the fields correctly. All the other fields are
// covered by the HCL in the test run function.
tests := []struct {
desc string
config string
}{
{"ca_file", `ca_file = "../test/client_certs/rootca.crt"`},
{"ca_path", `ca_path = "../test/client_certs/path"`},
}
for _, tt := range tests {
t.Run(tt.desc, func(t *testing.T) {
run(t, tt.config)
})
}
}
func TestAgent_updateTTLCheck(t *testing.T) {
if testing.Short() {
t.Skip("too slow for testing.Short")
}
t.Parallel()
a := NewTestAgent(t, "")
defer a.Shutdown()
checkBufSize := 100
health := &structs.HealthCheck{
Node: "foo",
CheckID: "mem",
Name: "memory util",
Status: api.HealthCritical,
}
chk := &structs.CheckType{
TTL: 15 * time.Second,
OutputMaxSize: checkBufSize,
}
// Add check and update it.
err := a.AddCheck(health, chk, false, "", ConfigSourceLocal)
if err != nil {
t.Fatalf("err: %v", err)
}
if err := a.updateTTLCheck(structs.NewCheckID("mem", nil), api.HealthPassing, "foo"); err != nil {
t.Fatalf("err: %v", err)
}
// Ensure we have a check mapping.
status := getCheck(a, "mem")
if status.Status != api.HealthPassing {
t.Fatalf("bad: %v", status)
}
if status.Output != "foo" {
t.Fatalf("bad: %v", status)
}
if err := a.updateTTLCheck(structs.NewCheckID("mem", nil), api.HealthCritical, strings.Repeat("--bad-- ", 5*checkBufSize)); err != nil {
t.Fatalf("err: %v", err)
}
// Ensure we have a check mapping.
status = getCheck(a, "mem")
if status.Status != api.HealthCritical {
t.Fatalf("bad: %v", status)
}
if len(status.Output) > checkBufSize*2 {
t.Fatalf("bad: %v", len(status.Output))
}
}
func TestAgent_PersistService(t *testing.T) {
if testing.Short() {
t.Skip("too slow for testing.Short")
}
t.Run("normal", func(t *testing.T) {
t.Parallel()
testAgent_PersistService(t, "enable_central_service_config = false")
})
t.Run("service manager", func(t *testing.T) {
t.Parallel()
testAgent_PersistService(t, "enable_central_service_config = true")
})
}
func testAgent_PersistService(t *testing.T, extraHCL string) {
t.Helper()
cfg := `
server = false
bootstrap = false
` + extraHCL
a := StartTestAgent(t, TestAgent{HCL: cfg})
defer a.Shutdown()
svc := &structs.NodeService{
ID: "redis",
Service: "redis",
Tags: []string{"foo"},
Port: 8000,
}
file := filepath.Join(a.Config.DataDir, servicesDir, structs.NewServiceID(svc.ID, nil).StringHashSHA256())
// Check is not persisted unless requested
if err := a.addServiceFromSource(svc, nil, false, "", ConfigSourceLocal); err != nil {
t.Fatalf("err: %v", err)
}
if _, err := os.Stat(file); err == nil {
t.Fatalf("should not persist")
}
// Persists to file if requested
if err := a.addServiceFromSource(svc, nil, true, "mytoken", ConfigSourceLocal); err != nil {
t.Fatalf("err: %v", err)
}
if _, err := os.Stat(file); err != nil {
t.Fatalf("err: %s", err)
}
expected, err := json.Marshal(persistedService{
Token: "mytoken",
Service: svc,
Source: "local",
})
if err != nil {
t.Fatalf("err: %s", err)
}
content, err := ioutil.ReadFile(file)
if err != nil {
t.Fatalf("err: %s", err)
}
if !bytes.Equal(expected, content) {
t.Fatalf("bad: %s", string(content))
}
// Updates service definition on disk
svc.Port = 8001
if err := a.addServiceFromSource(svc, nil, true, "mytoken", ConfigSourceLocal); err != nil {
t.Fatalf("err: %v", err)
}
expected, err = json.Marshal(persistedService{
Token: "mytoken",
Service: svc,
Source: "local",
})
if err != nil {
t.Fatalf("err: %s", err)
}
content, err = ioutil.ReadFile(file)
if err != nil {
t.Fatalf("err: %s", err)
}
if !bytes.Equal(expected, content) {
t.Fatalf("bad: %s", string(content))
}
a.Shutdown()
// Should load it back during later start
a2 := StartTestAgent(t, TestAgent{HCL: cfg, DataDir: a.DataDir})
defer a2.Shutdown()
restored := a2.State.ServiceState(structs.NewServiceID(svc.ID, nil))
if restored == nil {
t.Fatalf("service %q missing", svc.ID)
}
if got, want := restored.Token, "mytoken"; got != want {
t.Fatalf("got token %q want %q", got, want)
}
if got, want := restored.Service.Port, 8001; got != want {
t.Fatalf("got port %d want %d", got, want)
}
}
func TestAgent_persistedService_compat(t *testing.T) {
if testing.Short() {
t.Skip("too slow for testing.Short")
}
t.Run("normal", func(t *testing.T) {
t.Parallel()
testAgent_persistedService_compat(t, "enable_central_service_config = false")
})
t.Run("service manager", func(t *testing.T) {
t.Parallel()
testAgent_persistedService_compat(t, "enable_central_service_config = true")
})
}
func testAgent_persistedService_compat(t *testing.T, extraHCL string) {
t.Helper()
// Tests backwards compatibility of persisted services from pre-0.5.1
a := NewTestAgent(t, extraHCL)
defer a.Shutdown()
svc := &structs.NodeService{
ID: "redis",
Service: "redis",
Tags: []string{"foo"},
Port: 8000,
TaggedAddresses: map[string]structs.ServiceAddress{},
Weights: &structs.Weights{Passing: 1, Warning: 1},
EnterpriseMeta: *structs.DefaultEnterpriseMetaInDefaultPartition(),
}
// Encode the NodeService directly. This is what previous versions
// would serialize to the file (without the wrapper)
encoded, err := json.Marshal(svc)
if err != nil {
t.Fatalf("err: %s", err)
}
// Write the content to the file
file := filepath.Join(a.Config.DataDir, servicesDir, structs.NewServiceID(svc.ID, nil).StringHashSHA256())
if err := os.MkdirAll(filepath.Dir(file), 0700); err != nil {
t.Fatalf("err: %s", err)
}
if err := ioutil.WriteFile(file, encoded, 0600); err != nil {
t.Fatalf("err: %s", err)
}
// Load the services
if err := a.loadServices(a.Config, nil); err != nil {
t.Fatalf("err: %s", err)
}
// Ensure the service was restored
result := requireServiceExists(t, a, "redis")
require.Equal(t, svc, result)
}
func TestAgent_persistedService_compat_hash(t *testing.T) {
t.Run("normal", func(t *testing.T) {
t.Parallel()
testAgent_persistedService_compat_hash(t, "enable_central_service_config = false")
})
t.Run("service manager", func(t *testing.T) {
t.Parallel()
testAgent_persistedService_compat_hash(t, "enable_central_service_config = true")
})
}
func testAgent_persistedService_compat_hash(t *testing.T, extraHCL string) {
t.Helper()
// Tests backwards compatibility of persisted services from pre-0.5.1
a := NewTestAgent(t, extraHCL)
defer a.Shutdown()
svc := &structs.NodeService{
ID: "redis",
Service: "redis",
Tags: []string{"foo"},
Port: 8000,
TaggedAddresses: map[string]structs.ServiceAddress{},
Weights: &structs.Weights{Passing: 1, Warning: 1},
EnterpriseMeta: *structs.DefaultEnterpriseMetaInDefaultPartition(),
}
// Encode the NodeService directly. This is what previous versions
// would serialize to the file (without the wrapper)
encoded, err := json.Marshal(svc)
if err != nil {
t.Fatalf("err: %s", err)
}
// Write the content to the file using the old md5 based path
file := filepath.Join(a.Config.DataDir, servicesDir, stringHashMD5(svc.ID))
if err := os.MkdirAll(filepath.Dir(file), 0700); err != nil {
t.Fatalf("err: %s", err)
}
if err := ioutil.WriteFile(file, encoded, 0600); err != nil {
t.Fatalf("err: %s", err)
}
wrapped := persistedServiceConfig{
ServiceID: "redis",
Defaults: &structs.ServiceConfigResponse{},
EnterpriseMeta: *structs.DefaultEnterpriseMetaInDefaultPartition(),
}
encodedConfig, err := json.Marshal(wrapped)
if err != nil {
t.Fatalf("err: %s", err)
}
configFile := filepath.Join(a.Config.DataDir, serviceConfigDir, stringHashMD5(svc.ID))
if err := os.MkdirAll(filepath.Dir(configFile), 0700); err != nil {
t.Fatalf("err: %s", err)
}
if err := ioutil.WriteFile(configFile, encodedConfig, 0600); err != nil {
t.Fatalf("err: %s", err)
}
// Load the services
if err := a.loadServices(a.Config, nil); err != nil {
t.Fatalf("err: %s", err)
}
// Ensure the service was restored
result := requireServiceExists(t, a, "redis")
require.Equal(t, svc, result)
}
// Exists for backwards compatibility testing
func stringHashMD5(s string) string {
return fmt.Sprintf("%x", md5.Sum([]byte(s)))
}
func TestAgent_PurgeService(t *testing.T) {
if testing.Short() {
t.Skip("too slow for testing.Short")
}
t.Run("normal", func(t *testing.T) {
t.Parallel()
testAgent_PurgeService(t, "enable_central_service_config = false")
})
t.Run("service manager", func(t *testing.T) {
t.Parallel()
testAgent_PurgeService(t, "enable_central_service_config = true")
})
}
func testAgent_PurgeService(t *testing.T, extraHCL string) {
t.Helper()
a := NewTestAgent(t, extraHCL)
defer a.Shutdown()
svc := &structs.NodeService{
ID: "redis",
Service: "redis",
Tags: []string{"foo"},
Port: 8000,
}
file := filepath.Join(a.Config.DataDir, servicesDir, structs.NewServiceID(svc.ID, nil).StringHashSHA256())
if err := a.addServiceFromSource(svc, nil, true, "", ConfigSourceLocal); err != nil {
t.Fatalf("err: %v", err)
}
// Exists
if _, err := os.Stat(file); err != nil {
t.Fatalf("err: %s", err)
}
// Not removed
if err := a.removeService(structs.NewServiceID(svc.ID, nil), false); err != nil {
t.Fatalf("err: %s", err)
}
if _, err := os.Stat(file); err != nil {
t.Fatalf("err: %s", err)
}
// Re-add the service
if err := a.addServiceFromSource(svc, nil, true, "", ConfigSourceLocal); err != nil {
t.Fatalf("err: %v", err)
}
// Removed
if err := a.removeService(structs.NewServiceID(svc.ID, nil), true); err != nil {
t.Fatalf("err: %s", err)
}
if _, err := os.Stat(file); !os.IsNotExist(err) {
t.Fatalf("bad: %#v", err)
}
}
func TestAgent_PurgeServiceOnDuplicate(t *testing.T) {
if testing.Short() {
t.Skip("too slow for testing.Short")
}
t.Run("normal", func(t *testing.T) {
t.Parallel()
testAgent_PurgeServiceOnDuplicate(t, "enable_central_service_config = false")
})
t.Run("service manager", func(t *testing.T) {
t.Parallel()
testAgent_PurgeServiceOnDuplicate(t, "enable_central_service_config = true")
})
}
func testAgent_PurgeServiceOnDuplicate(t *testing.T, extraHCL string) {
t.Helper()
cfg := `
server = false
bootstrap = false
` + extraHCL
a := StartTestAgent(t, TestAgent{HCL: cfg})
defer a.Shutdown()
svc1 := &structs.NodeService{
ID: "redis",
Service: "redis",
Tags: []string{"foo"},
Port: 8000,
}
// First persist the service
require.NoError(t, a.addServiceFromSource(svc1, nil, true, "", ConfigSourceLocal))
a.Shutdown()
// Try bringing the agent back up with the service already
// existing in the config
a2 := StartTestAgent(t, TestAgent{Name: "Agent2", HCL: cfg + `
service = {
id = "redis"
name = "redis"
tags = ["bar"]
port = 9000
}
`, DataDir: a.DataDir})
defer a2.Shutdown()
sid := svc1.CompoundServiceID()
file := filepath.Join(a.Config.DataDir, servicesDir, sid.StringHashSHA256())
_, err := os.Stat(file)
require.Error(t, err, "should have removed persisted service")
result := requireServiceExists(t, a, "redis")
require.NotEqual(t, []string{"bar"}, result.Tags)
require.NotEqual(t, 9000, result.Port)
}
func TestAgent_PersistCheck(t *testing.T) {
if testing.Short() {
t.Skip("too slow for testing.Short")
}
t.Parallel()
cfg := `
server = false
bootstrap = false
enable_script_checks = true
`
a := StartTestAgent(t, TestAgent{HCL: cfg})
defer a.Shutdown()
check := &structs.HealthCheck{
Node: a.config.NodeName,
CheckID: "mem",
Name: "memory check",
Status: api.HealthPassing,
}
chkType := &structs.CheckType{
ScriptArgs: []string{"/bin/true"},
Interval: 10 * time.Second,
}
cid := check.CompoundCheckID()
file := filepath.Join(a.Config.DataDir, checksDir, cid.StringHashSHA256())
// Not persisted if not requested
require.NoError(t, a.AddCheck(check, chkType, false, "", ConfigSourceLocal))
_, err := os.Stat(file)
require.Error(t, err, "should not persist")
// Should persist if requested
require.NoError(t, a.AddCheck(check, chkType, true, "mytoken", ConfigSourceLocal))
_, err = os.Stat(file)
require.NoError(t, err)
expected, err := json.Marshal(persistedCheck{
Check: check,
ChkType: chkType,
Token: "mytoken",
Source: "local",
})
require.NoError(t, err)
content, err := ioutil.ReadFile(file)
require.NoError(t, err)
require.Equal(t, expected, content)
// Updates the check definition on disk
check.Name = "mem1"
require.NoError(t, a.AddCheck(check, chkType, true, "mytoken", ConfigSourceLocal))
expected, err = json.Marshal(persistedCheck{
Check: check,
ChkType: chkType,
Token: "mytoken",
Source: "local",
})
require.NoError(t, err)
content, err = ioutil.ReadFile(file)
require.NoError(t, err)
require.Equal(t, expected, content)
a.Shutdown()
// Should load it back during later start
a2 := StartTestAgent(t, TestAgent{Name: "Agent2", HCL: cfg, DataDir: a.DataDir})
defer a2.Shutdown()
result := requireCheckExists(t, a2, check.CheckID)
require.Equal(t, api.HealthCritical, result.Status)
require.Equal(t, "mem1", result.Name)
// Should have restored the monitor
requireCheckExistsMap(t, a2.checkMonitors, check.CheckID)
chkState := a2.State.CheckState(structs.NewCheckID(check.CheckID, nil))
require.NotNil(t, chkState)
require.Equal(t, "mytoken", chkState.Token)
}
func TestAgent_PurgeCheck(t *testing.T) {
if testing.Short() {
t.Skip("too slow for testing.Short")
}
t.Parallel()
a := NewTestAgent(t, "")
defer a.Shutdown()
check := &structs.HealthCheck{
Node: a.Config.NodeName,
CheckID: "mem",
Name: "memory check",
Status: api.HealthPassing,
}
file := filepath.Join(a.Config.DataDir, checksDir, checkIDHash(check.CheckID))
if err := a.AddCheck(check, nil, true, "", ConfigSourceLocal); err != nil {
t.Fatalf("err: %v", err)
}
// Not removed
if err := a.RemoveCheck(structs.NewCheckID(check.CheckID, nil), false); err != nil {
t.Fatalf("err: %s", err)
}
if _, err := os.Stat(file); err != nil {
t.Fatalf("err: %s", err)
}
// Removed
if err := a.RemoveCheck(structs.NewCheckID(check.CheckID, nil), true); err != nil {
t.Fatalf("err: %s", err)
}
if _, err := os.Stat(file); !os.IsNotExist(err) {
t.Fatalf("bad: %#v", err)
}
}
func TestAgent_PurgeCheckOnDuplicate(t *testing.T) {
if testing.Short() {
t.Skip("too slow for testing.Short")
}
t.Parallel()
nodeID := NodeID()
a := StartTestAgent(t, TestAgent{
HCL: `
node_id = "` + nodeID + `"
node_name = "Node ` + nodeID + `"
server = false
bootstrap = false
enable_script_checks = true
`})
defer a.Shutdown()
check1 := &structs.HealthCheck{
Node: a.Config.NodeName,
CheckID: "mem",
Name: "memory check",
Status: api.HealthPassing,
EnterpriseMeta: *structs.DefaultEnterpriseMetaInDefaultPartition(),
}
// First persist the check
if err := a.AddCheck(check1, nil, true, "", ConfigSourceLocal); err != nil {
t.Fatalf("err: %v", err)
}
a.Shutdown()
// Start again with the check registered in config
a2 := StartTestAgent(t, TestAgent{
Name: "Agent2",
DataDir: a.DataDir,
HCL: `
node_id = "` + nodeID + `"
node_name = "Node ` + nodeID + `"
server = false
bootstrap = false
enable_script_checks = true
check = {
id = "mem"
name = "memory check"
notes = "my cool notes"
args = ["/bin/check-redis.py"]
interval = "30s"
timeout = "5s"
}
`})
defer a2.Shutdown()
cid := check1.CompoundCheckID()
file := filepath.Join(a.DataDir, checksDir, cid.StringHashSHA256())
if _, err := os.Stat(file); err == nil {
t.Fatalf("should have removed persisted check")
}
result := requireCheckExists(t, a2, "mem")
expected := &structs.HealthCheck{
Node: a2.Config.NodeName,
CheckID: "mem",
Name: "memory check",
Status: api.HealthCritical,
Notes: "my cool notes",
Interval: "30s",
Timeout: "5s",
EnterpriseMeta: *structs.DefaultEnterpriseMetaInDefaultPartition(),
}
require.Equal(t, expected, result)
}
func TestAgent_DeregisterPersistedSidecarAfterRestart(t *testing.T) {
if testing.Short() {
t.Skip("too slow for testing.Short")
}
t.Parallel()
nodeID := NodeID()
a := StartTestAgent(t, TestAgent{
HCL: `
node_id = "` + nodeID + `"
node_name = "Node ` + nodeID + `"
server = false
bootstrap = false
enable_central_service_config = false
`})
defer a.Shutdown()
srv := &structs.NodeService{
ID: "svc",
Service: "svc",
Weights: &structs.Weights{
Passing: 2,
Warning: 1,
},
Tags: []string{"tag2"},
Port: 8200,
EnterpriseMeta: *structs.DefaultEnterpriseMetaInDefaultPartition(),
Connect: structs.ServiceConnect{
SidecarService: &structs.ServiceDefinition{},
},
}
connectSrv, _, _, err := sidecarServiceFromNodeService(srv, "")
require.NoError(t, err)
// First persist the check
err = a.addServiceFromSource(srv, nil, true, "", ConfigSourceLocal)
require.NoError(t, err)
err = a.addServiceFromSource(connectSrv, nil, true, "", ConfigSourceLocal)
require.NoError(t, err)
// check both services were registered
require.NotNil(t, a.State.Service(srv.CompoundServiceID()))
require.NotNil(t, a.State.Service(connectSrv.CompoundServiceID()))
a.Shutdown()
// Start again with the check registered in config
a2 := StartTestAgent(t, TestAgent{
Name: "Agent2",
DataDir: a.DataDir,
HCL: `
node_id = "` + nodeID + `"
node_name = "Node ` + nodeID + `"
server = false
bootstrap = false
enable_central_service_config = false
`})
defer a2.Shutdown()
// check both services were restored
require.NotNil(t, a2.State.Service(srv.CompoundServiceID()))
require.NotNil(t, a2.State.Service(connectSrv.CompoundServiceID()))
err = a2.RemoveService(srv.CompoundServiceID())
require.NoError(t, err)
// check both services were deregistered
require.Nil(t, a2.State.Service(srv.CompoundServiceID()))
require.Nil(t, a2.State.Service(connectSrv.CompoundServiceID()))
}
func TestAgent_loadChecks_token(t *testing.T) {
if testing.Short() {
t.Skip("too slow for testing.Short")
}
t.Parallel()
a := NewTestAgent(t, `
check = {
id = "rabbitmq"
name = "rabbitmq"
token = "abc123"
ttl = "10s"
}
`)
defer a.Shutdown()
requireCheckExists(t, a, "rabbitmq")
require.Equal(t, "abc123", a.State.CheckToken(structs.NewCheckID("rabbitmq", nil)))
}
func TestAgent_unloadChecks(t *testing.T) {
if testing.Short() {
t.Skip("too slow for testing.Short")
}
t.Parallel()
a := NewTestAgent(t, "")
defer a.Shutdown()
// First register a service
svc := &structs.NodeService{
ID: "redis",
Service: "redis",
Tags: []string{"foo"},
Port: 8000,
}
if err := a.addServiceFromSource(svc, nil, false, "", ConfigSourceLocal); err != nil {
t.Fatalf("err: %v", err)
}
// Register a check
check1 := &structs.HealthCheck{
Node: a.Config.NodeName,
CheckID: "service:redis",
Name: "redischeck",
Status: api.HealthPassing,
ServiceID: "redis",
ServiceName: "redis",
}
if err := a.AddCheck(check1, nil, false, "", ConfigSourceLocal); err != nil {
t.Fatalf("err: %s", err)
}
requireCheckExists(t, a, check1.CheckID)
// Unload all of the checks
if err := a.unloadChecks(); err != nil {
t.Fatalf("err: %s", err)
}
// Make sure it was unloaded
requireCheckMissing(t, a, check1.CheckID)
}
func TestAgent_loadServices_token(t *testing.T) {
if testing.Short() {
t.Skip("too slow for testing.Short")
}
t.Run("normal", func(t *testing.T) {
t.Parallel()
testAgent_loadServices_token(t, "enable_central_service_config = false")
})
t.Run("service manager", func(t *testing.T) {
t.Parallel()
testAgent_loadServices_token(t, "enable_central_service_config = true")
})
}
func testAgent_loadServices_token(t *testing.T, extraHCL string) {
t.Helper()
a := NewTestAgent(t, `
service = {
id = "rabbitmq"
name = "rabbitmq"
port = 5672
token = "abc123"
}
`+extraHCL)
defer a.Shutdown()
requireServiceExists(t, a, "rabbitmq")
if token := a.State.ServiceToken(structs.NewServiceID("rabbitmq", nil)); token != "abc123" {
t.Fatalf("bad: %s", token)
}
}
func TestAgent_loadServices_sidecar(t *testing.T) {
if testing.Short() {
t.Skip("too slow for testing.Short")
}
t.Run("normal", func(t *testing.T) {
t.Parallel()
testAgent_loadServices_sidecar(t, "enable_central_service_config = false")
})
t.Run("service manager", func(t *testing.T) {
t.Parallel()
testAgent_loadServices_sidecar(t, "enable_central_service_config = true")
})
}
func testAgent_loadServices_sidecar(t *testing.T, extraHCL string) {
t.Helper()
a := NewTestAgent(t, `
service = {
id = "rabbitmq"
name = "rabbitmq"
port = 5672
token = "abc123"
connect = {
sidecar_service {}
}
}
`+extraHCL)
defer a.Shutdown()
svc := requireServiceExists(t, a, "rabbitmq")
if token := a.State.ServiceToken(structs.NewServiceID("rabbitmq", nil)); token != "abc123" {
t.Fatalf("bad: %s", token)
}
sidecarSvc := requireServiceExists(t, a, "rabbitmq-sidecar-proxy")
if token := a.State.ServiceToken(structs.NewServiceID("rabbitmq-sidecar-proxy", nil)); token != "abc123" {
t.Fatalf("bad: %s", token)
}
// Verify default checks have been added
wantChecks := sidecarDefaultChecks(sidecarSvc.ID, sidecarSvc.Address, sidecarSvc.Proxy.LocalServiceAddress, sidecarSvc.Port)
gotChecks := a.State.ChecksForService(sidecarSvc.CompoundServiceID(), true)
gotChkNames := make(map[string]types.CheckID)
for _, check := range gotChecks {
requireCheckExists(t, a, check.CheckID)
gotChkNames[check.Name] = check.CheckID
}
for _, check := range wantChecks {
chkName := check.Name
require.NotNil(t, gotChkNames[chkName])
}
// Sanity check rabbitmq service should NOT have sidecar info in state since
// it's done it's job and should be a registration syntax sugar only.
assert.Nil(t, svc.Connect.SidecarService)
}
func TestAgent_loadServices_sidecarSeparateToken(t *testing.T) {
if testing.Short() {
t.Skip("too slow for testing.Short")
}
t.Run("normal", func(t *testing.T) {
t.Parallel()
testAgent_loadServices_sidecarSeparateToken(t, "enable_central_service_config = false")
})
t.Run("service manager", func(t *testing.T) {
t.Parallel()
testAgent_loadServices_sidecarSeparateToken(t, "enable_central_service_config = true")
})
}
func testAgent_loadServices_sidecarSeparateToken(t *testing.T, extraHCL string) {
t.Helper()
a := NewTestAgent(t, `
service = {
id = "rabbitmq"
name = "rabbitmq"
port = 5672
token = "abc123"
connect = {
sidecar_service {
token = "789xyz"
}
}
}
`+extraHCL)
defer a.Shutdown()
requireServiceExists(t, a, "rabbitmq")
if token := a.State.ServiceToken(structs.NewServiceID("rabbitmq", nil)); token != "abc123" {
t.Fatalf("bad: %s", token)
}
requireServiceExists(t, a, "rabbitmq-sidecar-proxy")
if token := a.State.ServiceToken(structs.NewServiceID("rabbitmq-sidecar-proxy", nil)); token != "789xyz" {
t.Fatalf("bad: %s", token)
}
}
func TestAgent_loadServices_sidecarInheritMeta(t *testing.T) {
if testing.Short() {
t.Skip("too slow for testing.Short")
}
t.Run("normal", func(t *testing.T) {
t.Parallel()
testAgent_loadServices_sidecarInheritMeta(t, "enable_central_service_config = false")
})
t.Run("service manager", func(t *testing.T) {
t.Parallel()
testAgent_loadServices_sidecarInheritMeta(t, "enable_central_service_config = true")
})
}
func testAgent_loadServices_sidecarInheritMeta(t *testing.T, extraHCL string) {
t.Helper()
a := NewTestAgent(t, `
service = {
id = "rabbitmq"
name = "rabbitmq"
port = 5672
tags = ["a", "b"],
meta = {
environment = "prod"
}
connect = {
sidecar_service {
}
}
}
`+extraHCL)
defer a.Shutdown()
svc := requireServiceExists(t, a, "rabbitmq")
require.Len(t, svc.Tags, 2)
require.Len(t, svc.Meta, 1)
sidecar := requireServiceExists(t, a, "rabbitmq-sidecar-proxy")
require.ElementsMatch(t, svc.Tags, sidecar.Tags)
require.Len(t, sidecar.Meta, 1)
meta, ok := sidecar.Meta["environment"]
require.True(t, ok, "missing sidecar service meta")
require.Equal(t, "prod", meta)
}
func TestAgent_loadServices_sidecarOverrideMeta(t *testing.T) {
if testing.Short() {
t.Skip("too slow for testing.Short")
}
t.Run("normal", func(t *testing.T) {
t.Parallel()
testAgent_loadServices_sidecarOverrideMeta(t, "enable_central_service_config = false")
})
t.Run("service manager", func(t *testing.T) {
t.Parallel()
testAgent_loadServices_sidecarOverrideMeta(t, "enable_central_service_config = true")
})
}
func testAgent_loadServices_sidecarOverrideMeta(t *testing.T, extraHCL string) {
t.Helper()
a := NewTestAgent(t, `
service = {
id = "rabbitmq"
name = "rabbitmq"
port = 5672
tags = ["a", "b"],
meta = {
environment = "prod"
}
connect = {
sidecar_service {
tags = ["foo"],
meta = {
environment = "qa"
}
}
}
}
`+extraHCL)
defer a.Shutdown()
svc := requireServiceExists(t, a, "rabbitmq")
require.Len(t, svc.Tags, 2)
require.Len(t, svc.Meta, 1)
sidecar := requireServiceExists(t, a, "rabbitmq-sidecar-proxy")
require.Len(t, sidecar.Tags, 1)
require.Equal(t, "foo", sidecar.Tags[0])
require.Len(t, sidecar.Meta, 1)
meta, ok := sidecar.Meta["environment"]
require.True(t, ok, "missing sidecar service meta")
require.Equal(t, "qa", meta)
}
func TestAgent_unloadServices(t *testing.T) {
if testing.Short() {
t.Skip("too slow for testing.Short")
}
t.Run("normal", func(t *testing.T) {
t.Parallel()
testAgent_unloadServices(t, "enable_central_service_config = false")
})
t.Run("service manager", func(t *testing.T) {
t.Parallel()
testAgent_unloadServices(t, "enable_central_service_config = true")
})
}
func testAgent_unloadServices(t *testing.T, extraHCL string) {
t.Helper()
a := NewTestAgent(t, extraHCL)
defer a.Shutdown()
svc := &structs.NodeService{
ID: "redis",
Service: "redis",
Tags: []string{"foo"},
Port: 8000,
}
// Register the service
if err := a.addServiceFromSource(svc, nil, false, "", ConfigSourceLocal); err != nil {
t.Fatalf("err: %v", err)
}
requireServiceExists(t, a, svc.ID)
// Unload all services
if err := a.unloadServices(); err != nil {
t.Fatalf("err: %s", err)
}
if len(a.State.Services(structs.WildcardEnterpriseMetaInDefaultPartition())) != 0 {
t.Fatalf("should have unloaded services")
}
}
func TestAgent_Service_MaintenanceMode(t *testing.T) {
if testing.Short() {
t.Skip("too slow for testing.Short")
}
t.Parallel()
a := NewTestAgent(t, "")
defer a.Shutdown()
svc := &structs.NodeService{
ID: "redis",
Service: "redis",
Tags: []string{"foo"},
Port: 8000,
}
// Register the service
if err := a.addServiceFromSource(svc, nil, false, "", ConfigSourceLocal); err != nil {
t.Fatalf("err: %v", err)
}
sid := structs.NewServiceID("redis", nil)
// Enter maintenance mode for the service
if err := a.EnableServiceMaintenance(sid, "broken", "mytoken"); err != nil {
t.Fatalf("err: %s", err)
}
// Make sure the critical health check was added
checkID := serviceMaintCheckID(sid)
check := a.State.Check(checkID)
if check == nil {
t.Fatalf("should have registered critical maintenance check")
}
// Check that the token was used to register the check
if token := a.State.CheckToken(checkID); token != "mytoken" {
t.Fatalf("expected 'mytoken', got: '%s'", token)
}
// Ensure the reason was set in notes
if check.Notes != "broken" {
t.Fatalf("bad: %#v", check)
}
// Leave maintenance mode
if err := a.DisableServiceMaintenance(sid); err != nil {
t.Fatalf("err: %s", err)
}
// Ensure the check was deregistered
if found := a.State.Check(checkID); found != nil {
t.Fatalf("should have deregistered maintenance check")
}
// Enter service maintenance mode without providing a reason
if err := a.EnableServiceMaintenance(sid, "", ""); err != nil {
t.Fatalf("err: %s", err)
}
// Ensure the check was registered with the default notes
check = a.State.Check(checkID)
if check == nil {
t.Fatalf("should have registered critical check")
}
if check.Notes != defaultServiceMaintReason {
t.Fatalf("bad: %#v", check)
}
}
func TestAgent_Service_Reap(t *testing.T) {
if testing.Short() {
t.Skip("too slow for testing.Short")
}
// t.Parallel() // timing test. no parallel
a := StartTestAgent(t, TestAgent{Overrides: `
check_reap_interval = "50ms"
check_deregister_interval_min = "0s"
`})
defer a.Shutdown()
testrpc.WaitForTestAgent(t, a.RPC, "dc1")
svc := &structs.NodeService{
ID: "redis",
Service: "redis",
Tags: []string{"foo"},
Port: 8000,
}
chkTypes := []*structs.CheckType{
{
Status: api.HealthPassing,
TTL: 25 * time.Millisecond,
DeregisterCriticalServiceAfter: 200 * time.Millisecond,
},
}
// Register the service.
if err := a.addServiceFromSource(svc, chkTypes, false, "", ConfigSourceLocal); err != nil {
t.Fatalf("err: %v", err)
}
// Make sure it's there and there's no critical check yet.
requireServiceExists(t, a, "redis")
require.Len(t, a.State.CriticalCheckStates(structs.WildcardEnterpriseMetaInDefaultPartition()), 0, "should not have critical checks")
// Wait for the check TTL to fail but before the check is reaped.
time.Sleep(100 * time.Millisecond)
requireServiceExists(t, a, "redis")
require.Len(t, a.State.CriticalCheckStates(nil), 1, "should have 1 critical check")
// Pass the TTL.
if err := a.updateTTLCheck(structs.NewCheckID("service:redis", nil), api.HealthPassing, "foo"); err != nil {
t.Fatalf("err: %v", err)
}
requireServiceExists(t, a, "redis")
require.Len(t, a.State.CriticalCheckStates(structs.WildcardEnterpriseMetaInDefaultPartition()), 0, "should not have critical checks")
// Wait for the check TTL to fail again.
time.Sleep(100 * time.Millisecond)
requireServiceExists(t, a, "redis")
require.Len(t, a.State.CriticalCheckStates(structs.WildcardEnterpriseMetaInDefaultPartition()), 1, "should have 1 critical check")
// Wait for the reap.
time.Sleep(400 * time.Millisecond)
requireServiceMissing(t, a, "redis")
require.Len(t, a.State.CriticalCheckStates(structs.WildcardEnterpriseMetaInDefaultPartition()), 0, "should not have critical checks")
}
func TestAgent_Service_NoReap(t *testing.T) {
if testing.Short() {
t.Skip("too slow for testing.Short")
}
// t.Parallel() // timing test. no parallel
a := StartTestAgent(t, TestAgent{Overrides: `
check_reap_interval = "50ms"
check_deregister_interval_min = "0s"
`})
defer a.Shutdown()
svc := &structs.NodeService{
ID: "redis",
Service: "redis",
Tags: []string{"foo"},
Port: 8000,
}
chkTypes := []*structs.CheckType{
{
Status: api.HealthPassing,
TTL: 25 * time.Millisecond,
},
}
// Register the service.
if err := a.addServiceFromSource(svc, chkTypes, false, "", ConfigSourceLocal); err != nil {
t.Fatalf("err: %v", err)
}
// Make sure it's there and there's no critical check yet.
requireServiceExists(t, a, "redis")
require.Len(t, a.State.CriticalCheckStates(structs.WildcardEnterpriseMetaInDefaultPartition()), 0)
// Wait for the check TTL to fail.
time.Sleep(200 * time.Millisecond)
requireServiceExists(t, a, "redis")
require.Len(t, a.State.CriticalCheckStates(structs.WildcardEnterpriseMetaInDefaultPartition()), 1)
// Wait a while and make sure it doesn't reap.
time.Sleep(200 * time.Millisecond)
requireServiceExists(t, a, "redis")
require.Len(t, a.State.CriticalCheckStates(structs.WildcardEnterpriseMetaInDefaultPartition()), 1)
}