mirror of https://github.com/hashicorp/consul
You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
1810 lines
46 KiB
1810 lines
46 KiB
package agent |
|
|
|
import ( |
|
"fmt" |
|
"net/http" |
|
"net/http/httptest" |
|
"net/url" |
|
"testing" |
|
"time" |
|
|
|
"github.com/hashicorp/consul/acl" |
|
"github.com/hashicorp/consul/api" |
|
|
|
"github.com/hashicorp/serf/coordinate" |
|
"github.com/stretchr/testify/assert" |
|
"github.com/stretchr/testify/require" |
|
|
|
"github.com/hashicorp/consul/agent/structs" |
|
"github.com/hashicorp/consul/sdk/testutil/retry" |
|
"github.com/hashicorp/consul/testrpc" |
|
) |
|
|
|
func TestCatalogRegister_Service_InvalidAddress(t *testing.T) { |
|
if testing.Short() { |
|
t.Skip("too slow for testing.Short") |
|
} |
|
|
|
t.Parallel() |
|
a := NewTestAgent(t, "") |
|
defer a.Shutdown() |
|
|
|
for _, addr := range []string{"0.0.0.0", "::", "[::]"} { |
|
t.Run("addr "+addr, func(t *testing.T) { |
|
args := &structs.RegisterRequest{ |
|
Node: "foo", |
|
Address: "127.0.0.1", |
|
Service: &structs.NodeService{ |
|
Service: "test", |
|
Address: addr, |
|
Port: 8080, |
|
}, |
|
} |
|
req, _ := http.NewRequest("PUT", "/v1/catalog/register", jsonReader(args)) |
|
_, err := a.srv.CatalogRegister(nil, req) |
|
if err == nil || err.Error() != "Invalid service address" { |
|
t.Fatalf("err: %v", err) |
|
} |
|
}) |
|
} |
|
} |
|
|
|
func TestCatalogDeregister(t *testing.T) { |
|
if testing.Short() { |
|
t.Skip("too slow for testing.Short") |
|
} |
|
|
|
t.Parallel() |
|
a := NewTestAgent(t, "") |
|
defer a.Shutdown() |
|
|
|
// Register node |
|
args := &structs.DeregisterRequest{Node: "foo"} |
|
req, _ := http.NewRequest("PUT", "/v1/catalog/deregister", jsonReader(args)) |
|
obj, err := a.srv.CatalogDeregister(nil, req) |
|
if err != nil { |
|
t.Fatalf("err: %v", err) |
|
} |
|
|
|
res := obj.(bool) |
|
if res != true { |
|
t.Fatalf("bad: %v", res) |
|
} |
|
} |
|
|
|
func TestCatalogDatacenters(t *testing.T) { |
|
if testing.Short() { |
|
t.Skip("too slow for testing.Short") |
|
} |
|
|
|
t.Parallel() |
|
a := NewTestAgent(t, "") |
|
defer a.Shutdown() |
|
|
|
retry.Run(t, func(r *retry.R) { |
|
req, _ := http.NewRequest("GET", "/v1/catalog/datacenters", nil) |
|
obj, err := a.srv.CatalogDatacenters(nil, req) |
|
if err != nil { |
|
r.Fatal(err) |
|
} |
|
|
|
dcs := obj.([]string) |
|
if got, want := len(dcs), 1; got != want { |
|
r.Fatalf("got %d data centers want %d", got, want) |
|
} |
|
}) |
|
} |
|
|
|
func TestCatalogNodes(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") |
|
|
|
// Register node |
|
args := &structs.RegisterRequest{ |
|
Datacenter: "dc1", |
|
Node: "foo", |
|
Address: "127.0.0.1", |
|
} |
|
|
|
var out struct{} |
|
if err := a.RPC("Catalog.Register", args, &out); err != nil { |
|
t.Fatalf("err: %v", err) |
|
} |
|
|
|
req, _ := http.NewRequest("GET", "/v1/catalog/nodes?dc=dc1", nil) |
|
resp := httptest.NewRecorder() |
|
obj, err := a.srv.CatalogNodes(resp, req) |
|
if err != nil { |
|
t.Fatalf("err: %v", err) |
|
} |
|
|
|
// Verify an index is set |
|
assertIndex(t, resp) |
|
|
|
nodes := obj.(structs.Nodes) |
|
if len(nodes) != 2 { |
|
t.Fatalf("bad: %v ; nodes:=%v", obj, nodes) |
|
} |
|
} |
|
|
|
func TestCatalogNodes_MetaFilter(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") |
|
|
|
// Register a node with a meta field |
|
args := &structs.RegisterRequest{ |
|
Datacenter: "dc1", |
|
Node: "foo", |
|
Address: "127.0.0.1", |
|
NodeMeta: map[string]string{ |
|
"somekey": "somevalue", |
|
}, |
|
} |
|
|
|
var out struct{} |
|
if err := a.RPC("Catalog.Register", args, &out); err != nil { |
|
t.Fatalf("err: %v", err) |
|
} |
|
|
|
req, _ := http.NewRequest("GET", "/v1/catalog/nodes?node-meta=somekey:somevalue", nil) |
|
resp := httptest.NewRecorder() |
|
obj, err := a.srv.CatalogNodes(resp, req) |
|
if err != nil { |
|
t.Fatalf("err: %v", err) |
|
} |
|
|
|
// Verify an index is set |
|
assertIndex(t, resp) |
|
|
|
// Verify we only get the node with the correct meta field back |
|
nodes := obj.(structs.Nodes) |
|
if len(nodes) != 1 { |
|
t.Fatalf("bad: %v", obj) |
|
} |
|
if v, ok := nodes[0].Meta["somekey"]; !ok || v != "somevalue" { |
|
t.Fatalf("bad: %v", nodes[0].Meta) |
|
} |
|
} |
|
|
|
func TestCatalogNodes_Filter(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") |
|
|
|
// Register a node with a meta field |
|
args := &structs.RegisterRequest{ |
|
Datacenter: "dc1", |
|
Node: "foo", |
|
Address: "127.0.0.1", |
|
NodeMeta: map[string]string{ |
|
"somekey": "somevalue", |
|
}, |
|
} |
|
|
|
var out struct{} |
|
require.NoError(t, a.RPC("Catalog.Register", args, &out)) |
|
|
|
req, _ := http.NewRequest("GET", "/v1/catalog/nodes?filter="+url.QueryEscape("Meta.somekey == somevalue"), nil) |
|
resp := httptest.NewRecorder() |
|
obj, err := a.srv.CatalogNodes(resp, req) |
|
require.NoError(t, err) |
|
|
|
// Verify an index is set |
|
assertIndex(t, resp) |
|
|
|
// Verify we only get the node with the correct meta field back |
|
nodes := obj.(structs.Nodes) |
|
require.Len(t, nodes, 1) |
|
|
|
v, ok := nodes[0].Meta["somekey"] |
|
require.True(t, ok) |
|
require.Equal(t, v, "somevalue") |
|
} |
|
|
|
func TestCatalogNodes_WanTranslation(t *testing.T) { |
|
if testing.Short() { |
|
t.Skip("too slow for testing.Short") |
|
} |
|
|
|
t.Parallel() |
|
a1 := NewTestAgent(t, ` |
|
datacenter = "dc1" |
|
translate_wan_addrs = true |
|
acl_datacenter = "" |
|
`) |
|
defer a1.Shutdown() |
|
testrpc.WaitForTestAgent(t, a1.RPC, "dc1") |
|
|
|
a2 := NewTestAgent(t, ` |
|
datacenter = "dc2" |
|
translate_wan_addrs = true |
|
acl_datacenter = "" |
|
`) |
|
defer a2.Shutdown() |
|
testrpc.WaitForTestAgent(t, a2.RPC, "dc2") |
|
|
|
// Wait for the WAN join. |
|
addr := fmt.Sprintf("127.0.0.1:%d", a1.Config.SerfPortWAN) |
|
if _, err := a2.JoinWAN([]string{addr}); err != nil { |
|
t.Fatalf("err: %v", err) |
|
} |
|
testrpc.WaitForLeader(t, a1.RPC, "dc1") |
|
testrpc.WaitForLeader(t, a2.RPC, "dc2") |
|
retry.Run(t, func(r *retry.R) { |
|
if got, want := len(a1.WANMembers()), 2; got < want { |
|
r.Fatalf("got %d WAN members want at least %d", got, want) |
|
} |
|
}) |
|
|
|
// Register a node with DC2. |
|
{ |
|
args := &structs.RegisterRequest{ |
|
Datacenter: "dc2", |
|
Node: "wan_translation_test", |
|
Address: "127.0.0.1", |
|
TaggedAddresses: map[string]string{ |
|
"wan": "127.0.0.2", |
|
}, |
|
Service: &structs.NodeService{ |
|
Service: "http_wan_translation_test", |
|
}, |
|
} |
|
|
|
var out struct{} |
|
if err := a2.RPC("Catalog.Register", args, &out); err != nil { |
|
t.Fatalf("err: %v", err) |
|
} |
|
} |
|
|
|
// Query nodes in DC2 from DC1. |
|
req, _ := http.NewRequest("GET", "/v1/catalog/nodes?dc=dc2", nil) |
|
resp1 := httptest.NewRecorder() |
|
obj1, err1 := a1.srv.CatalogNodes(resp1, req) |
|
if err1 != nil { |
|
t.Fatalf("err: %v", err1) |
|
} |
|
assertIndex(t, resp1) |
|
|
|
// Expect that DC1 gives us a WAN address (since the node is in DC2). |
|
nodes1 := obj1.(structs.Nodes) |
|
if len(nodes1) != 2 { |
|
t.Fatalf("bad: %v, nodes:=%v", obj1, nodes1) |
|
} |
|
var address string |
|
for _, node := range nodes1 { |
|
if node.Node == "wan_translation_test" { |
|
address = node.Address |
|
} |
|
} |
|
if address != "127.0.0.2" { |
|
t.Fatalf("bad: %s", address) |
|
} |
|
|
|
// Query DC2 from DC2. |
|
resp2 := httptest.NewRecorder() |
|
obj2, err2 := a2.srv.CatalogNodes(resp2, req) |
|
if err2 != nil { |
|
t.Fatalf("err: %v", err2) |
|
} |
|
assertIndex(t, resp2) |
|
|
|
// Expect that DC2 gives us a private address (since the node is in DC2). |
|
nodes2 := obj2.(structs.Nodes) |
|
if len(nodes2) != 2 { |
|
t.Fatalf("bad: %v", obj2) |
|
} |
|
for _, node := range nodes2 { |
|
if node.Node == "wan_translation_test" { |
|
address = node.Address |
|
} |
|
} |
|
if address != "127.0.0.1" { |
|
t.Fatalf("bad: %s", address) |
|
} |
|
} |
|
|
|
func TestCatalogNodes_Blocking(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", testrpc.WaitForAntiEntropySync()) |
|
|
|
// Run the query |
|
args := &structs.DCSpecificRequest{ |
|
Datacenter: "dc1", |
|
} |
|
var out structs.IndexedNodes |
|
if err := a.RPC("Catalog.ListNodes", *args, &out); err != nil { |
|
t.Fatalf("err: %v", err) |
|
} |
|
|
|
// Async cause a change |
|
waitIndex := out.Index |
|
start := time.Now() |
|
go func() { |
|
time.Sleep(100 * time.Millisecond) |
|
args := &structs.RegisterRequest{ |
|
Datacenter: "dc1", |
|
Node: "foo", |
|
Address: "127.0.0.1", |
|
} |
|
var out struct{} |
|
if err := a.RPC("Catalog.Register", args, &out); err != nil { |
|
t.Errorf("err: %v", err) |
|
} |
|
}() |
|
|
|
const waitDuration = 3 * time.Second |
|
|
|
// Re-run the query, if errantly woken up with no change, resume blocking. |
|
var elapsed time.Duration |
|
RUN_BLOCKING_QUERY: |
|
req, err := http.NewRequest("GET", fmt.Sprintf("/v1/catalog/nodes?wait=%s&index=%d", |
|
waitDuration.String(), |
|
waitIndex), nil) |
|
if err != nil { |
|
t.Fatalf("err: %v", err) |
|
} |
|
resp := httptest.NewRecorder() |
|
obj, err := a.srv.CatalogNodes(resp, req) |
|
if err != nil { |
|
t.Fatalf("err: %v", err) |
|
} |
|
|
|
elapsed = time.Since(start) |
|
|
|
idx := getIndex(t, resp) |
|
if idx < waitIndex { |
|
t.Fatalf("bad: %v", idx) |
|
} else if idx == waitIndex { |
|
if elapsed > waitDuration { |
|
// This should prevent the loop from running longer than the |
|
// waitDuration |
|
t.Fatalf("too slow: %v", elapsed) |
|
} |
|
goto RUN_BLOCKING_QUERY |
|
} |
|
|
|
// Should block at least 100ms before getting the changed results |
|
if elapsed < 100*time.Millisecond { |
|
t.Fatalf("too fast: %v", elapsed) |
|
} |
|
|
|
nodes := obj.(structs.Nodes) |
|
if len(nodes) != 2 { |
|
t.Fatalf("bad: %v", obj) |
|
} |
|
|
|
} |
|
|
|
func TestCatalogNodes_DistanceSort(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") |
|
|
|
// Register nodes. |
|
args := &structs.RegisterRequest{ |
|
Datacenter: "dc1", |
|
Node: "foo", |
|
Address: "127.0.0.1", |
|
} |
|
var out struct{} |
|
require.NoError(t, a.RPC("Catalog.Register", args, &out)) |
|
|
|
args = &structs.RegisterRequest{ |
|
Datacenter: "dc1", |
|
Node: "bar", |
|
Address: "127.0.0.2", |
|
} |
|
require.NoError(t, a.RPC("Catalog.Register", args, &out)) |
|
|
|
// Nobody has coordinates set so this will still return them in the |
|
// order they are indexed. |
|
req, _ := http.NewRequest("GET", "/v1/catalog/nodes?dc=dc1&near=foo", nil) |
|
resp := httptest.NewRecorder() |
|
obj, err := a.srv.CatalogNodes(resp, req) |
|
require.NoError(t, err) |
|
|
|
assertIndex(t, resp) |
|
nodes := obj.(structs.Nodes) |
|
require.Len(t, nodes, 3) |
|
require.Equal(t, "bar", nodes[0].Node) |
|
require.Equal(t, "foo", nodes[1].Node) |
|
require.Equal(t, a.Config.NodeName, nodes[2].Node) |
|
|
|
// Send an update for the node and wait for it to get applied. |
|
arg := structs.CoordinateUpdateRequest{ |
|
Datacenter: "dc1", |
|
Node: "foo", |
|
Coord: coordinate.NewCoordinate(coordinate.DefaultConfig()), |
|
} |
|
require.NoError(t, a.RPC("Coordinate.Update", &arg, &out)) |
|
time.Sleep(300 * time.Millisecond) |
|
|
|
// Query again and now foo should have moved to the front of the line. |
|
req, _ = http.NewRequest("GET", "/v1/catalog/nodes?dc=dc1&near=foo", nil) |
|
resp = httptest.NewRecorder() |
|
obj, err = a.srv.CatalogNodes(resp, req) |
|
require.NoError(t, err) |
|
|
|
assertIndex(t, resp) |
|
nodes = obj.(structs.Nodes) |
|
require.Len(t, nodes, 3) |
|
require.Equal(t, "foo", nodes[0].Node) |
|
require.Equal(t, "bar", nodes[1].Node) |
|
require.Equal(t, a.Config.NodeName, nodes[2].Node) |
|
} |
|
|
|
func TestCatalogServices(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") |
|
|
|
// Register node |
|
args := &structs.RegisterRequest{ |
|
Datacenter: "dc1", |
|
Node: "foo", |
|
Address: "127.0.0.1", |
|
Service: &structs.NodeService{ |
|
Service: "api", |
|
}, |
|
} |
|
|
|
var out struct{} |
|
if err := a.RPC("Catalog.Register", args, &out); err != nil { |
|
t.Fatalf("err: %v", err) |
|
} |
|
|
|
req, _ := http.NewRequest("GET", "/v1/catalog/services?dc=dc1", nil) |
|
resp := httptest.NewRecorder() |
|
obj, err := a.srv.CatalogServices(resp, req) |
|
if err != nil { |
|
t.Fatalf("err: %v", err) |
|
} |
|
|
|
assertIndex(t, resp) |
|
|
|
services := obj.(structs.Services) |
|
if len(services) != 2 { |
|
t.Fatalf("bad: %v", obj) |
|
} |
|
} |
|
|
|
func TestCatalogServices_NodeMetaFilter(t *testing.T) { |
|
if testing.Short() { |
|
t.Skip("too slow for testing.Short") |
|
} |
|
|
|
t.Parallel() |
|
a := NewTestAgent(t, "") |
|
defer a.Shutdown() |
|
|
|
// Register node |
|
args := &structs.RegisterRequest{ |
|
Datacenter: "dc1", |
|
Node: "foo", |
|
Address: "127.0.0.1", |
|
NodeMeta: map[string]string{ |
|
"somekey": "somevalue", |
|
}, |
|
Service: &structs.NodeService{ |
|
Service: "api", |
|
}, |
|
} |
|
|
|
var out struct{} |
|
if err := a.RPC("Catalog.Register", args, &out); err != nil { |
|
t.Fatalf("err: %v", err) |
|
} |
|
|
|
req, _ := http.NewRequest("GET", "/v1/catalog/services?node-meta=somekey:somevalue", nil) |
|
resp := httptest.NewRecorder() |
|
obj, err := a.srv.CatalogServices(resp, req) |
|
if err != nil { |
|
t.Fatalf("err: %v", err) |
|
} |
|
|
|
assertIndex(t, resp) |
|
|
|
services := obj.(structs.Services) |
|
if len(services) != 1 { |
|
t.Fatalf("bad: %v", obj) |
|
} |
|
if _, ok := services[args.Service.Service]; !ok { |
|
t.Fatalf("bad: %v", services) |
|
} |
|
} |
|
|
|
func TestCatalogRegister_checkRegistration(t *testing.T) { |
|
if testing.Short() { |
|
t.Skip("too slow for testing.Short") |
|
} |
|
|
|
t.Parallel() |
|
a := NewTestAgent(t, "") |
|
defer a.Shutdown() |
|
|
|
// Register node with a service and check |
|
check := structs.HealthCheck{ |
|
Node: "foo", |
|
CheckID: "foo-check", |
|
Name: "foo check", |
|
ServiceID: "api", |
|
Definition: structs.HealthCheckDefinition{ |
|
TCP: "localhost:8888", |
|
Interval: 5 * time.Second, |
|
}, |
|
} |
|
|
|
args := &structs.RegisterRequest{ |
|
Datacenter: "dc1", |
|
Node: "foo", |
|
Address: "127.0.0.1", |
|
Service: &structs.NodeService{ |
|
Service: "api", |
|
}, |
|
Check: &check, |
|
} |
|
|
|
var out struct{} |
|
if err := a.RPC("Catalog.Register", args, &out); err != nil { |
|
t.Fatalf("err: %v", err) |
|
} |
|
|
|
retry.Run(t, func(r *retry.R) { |
|
req, _ := http.NewRequest("GET", "/v1/health/checks/api", nil) |
|
resp := httptest.NewRecorder() |
|
obj, err := a.srv.HealthServiceChecks(resp, req) |
|
if err != nil { |
|
r.Fatalf("err: %v", err) |
|
} |
|
|
|
checks := obj.(structs.HealthChecks) |
|
if len(checks) != 1 { |
|
r.Fatalf("expected 1 check, got: %d", len(checks)) |
|
} |
|
if checks[0].CheckID != check.CheckID { |
|
r.Fatalf("expected check id %s, got %s", check.Type, checks[0].Type) |
|
} |
|
if checks[0].Type != "tcp" { |
|
r.Fatalf("expected check type tcp, got %s", checks[0].Type) |
|
} |
|
}) |
|
} |
|
|
|
func TestCatalogServiceNodes(t *testing.T) { |
|
if testing.Short() { |
|
t.Skip("too slow for testing.Short") |
|
} |
|
|
|
t.Parallel() |
|
a := NewTestAgent(t, "") |
|
defer a.Shutdown() |
|
|
|
// Make sure an empty list is returned, not a nil |
|
{ |
|
req, _ := http.NewRequest("GET", "/v1/catalog/service/api?tag=a", nil) |
|
resp := httptest.NewRecorder() |
|
obj, err := a.srv.CatalogServiceNodes(resp, req) |
|
if err != nil { |
|
t.Fatalf("err: %v", err) |
|
} |
|
|
|
assertIndex(t, resp) |
|
|
|
nodes := obj.(structs.ServiceNodes) |
|
if nodes == nil || len(nodes) != 0 { |
|
t.Fatalf("bad: %v", obj) |
|
} |
|
} |
|
|
|
// Register node |
|
args := &structs.RegisterRequest{ |
|
Datacenter: "dc1", |
|
Node: "foo", |
|
Address: "127.0.0.1", |
|
Service: &structs.NodeService{ |
|
Service: "api", |
|
Tags: []string{"a"}, |
|
}, |
|
} |
|
|
|
var out struct{} |
|
if err := a.RPC("Catalog.Register", args, &out); err != nil { |
|
t.Fatalf("err: %v", err) |
|
} |
|
|
|
req, _ := http.NewRequest("GET", "/v1/catalog/service/api?tag=a", nil) |
|
resp := httptest.NewRecorder() |
|
obj, err := a.srv.CatalogServiceNodes(resp, req) |
|
if err != nil { |
|
t.Fatalf("err: %v", err) |
|
} |
|
|
|
assertIndex(t, resp) |
|
|
|
nodes := obj.(structs.ServiceNodes) |
|
if len(nodes) != 1 { |
|
t.Fatalf("bad: %v", obj) |
|
} |
|
|
|
// Test caching |
|
{ |
|
// List instances with cache enabled |
|
req, _ := http.NewRequest("GET", "/v1/catalog/service/api?cached", nil) |
|
resp := httptest.NewRecorder() |
|
obj, err := a.srv.CatalogServiceNodes(resp, req) |
|
require.NoError(t, err) |
|
nodes := obj.(structs.ServiceNodes) |
|
assert.Len(t, nodes, 1) |
|
|
|
// Should be a cache miss |
|
assert.Equal(t, "MISS", resp.Header().Get("X-Cache")) |
|
} |
|
|
|
{ |
|
// List instances with cache enabled |
|
req, _ := http.NewRequest("GET", "/v1/catalog/service/api?cached", nil) |
|
resp := httptest.NewRecorder() |
|
obj, err := a.srv.CatalogServiceNodes(resp, req) |
|
require.NoError(t, err) |
|
nodes := obj.(structs.ServiceNodes) |
|
assert.Len(t, nodes, 1) |
|
|
|
// Should be a cache HIT now! |
|
assert.Equal(t, "HIT", resp.Header().Get("X-Cache")) |
|
assert.Equal(t, "0", resp.Header().Get("Age")) |
|
} |
|
|
|
// Ensure background refresh works |
|
{ |
|
// Register a new instance of the service |
|
args2 := args |
|
args2.Node = "bar" |
|
args2.Address = "127.0.0.2" |
|
require.NoError(t, a.RPC("Catalog.Register", args, &out)) |
|
|
|
retry.Run(t, func(r *retry.R) { |
|
// List it again |
|
req, _ := http.NewRequest("GET", "/v1/catalog/service/api?cached", nil) |
|
resp := httptest.NewRecorder() |
|
obj, err := a.srv.CatalogServiceNodes(resp, req) |
|
r.Check(err) |
|
|
|
nodes := obj.(structs.ServiceNodes) |
|
if len(nodes) != 2 { |
|
r.Fatalf("Want 2 nodes") |
|
} |
|
|
|
// Should be a cache hit! The data should've updated in the cache |
|
// in the background so this should've been fetched directly from |
|
// the cache. |
|
if resp.Header().Get("X-Cache") != "HIT" { |
|
r.Fatalf("should be a cache hit") |
|
} |
|
}) |
|
} |
|
} |
|
|
|
func TestCatalogServiceNodes_NodeMetaFilter(t *testing.T) { |
|
if testing.Short() { |
|
t.Skip("too slow for testing.Short") |
|
} |
|
|
|
t.Parallel() |
|
a := NewTestAgent(t, "") |
|
defer a.Shutdown() |
|
|
|
// Make sure an empty list is returned, not a nil |
|
{ |
|
req, _ := http.NewRequest("GET", "/v1/catalog/service/api?node-meta=somekey:somevalue", nil) |
|
resp := httptest.NewRecorder() |
|
obj, err := a.srv.CatalogServiceNodes(resp, req) |
|
if err != nil { |
|
t.Fatalf("err: %v", err) |
|
} |
|
|
|
assertIndex(t, resp) |
|
|
|
nodes := obj.(structs.ServiceNodes) |
|
if nodes == nil || len(nodes) != 0 { |
|
t.Fatalf("bad: %v", obj) |
|
} |
|
} |
|
|
|
// Register node |
|
args := &structs.RegisterRequest{ |
|
Datacenter: "dc1", |
|
Node: "foo", |
|
Address: "127.0.0.1", |
|
NodeMeta: map[string]string{ |
|
"somekey": "somevalue", |
|
}, |
|
Service: &structs.NodeService{ |
|
Service: "api", |
|
}, |
|
} |
|
|
|
var out struct{} |
|
if err := a.RPC("Catalog.Register", args, &out); err != nil { |
|
t.Fatalf("err: %v", err) |
|
} |
|
|
|
req, _ := http.NewRequest("GET", "/v1/catalog/service/api?node-meta=somekey:somevalue", nil) |
|
resp := httptest.NewRecorder() |
|
obj, err := a.srv.CatalogServiceNodes(resp, req) |
|
if err != nil { |
|
t.Fatalf("err: %v", err) |
|
} |
|
|
|
assertIndex(t, resp) |
|
|
|
nodes := obj.(structs.ServiceNodes) |
|
if len(nodes) != 1 { |
|
t.Fatalf("bad: %v", obj) |
|
} |
|
} |
|
|
|
func TestCatalogServiceNodes_Filter(t *testing.T) { |
|
if testing.Short() { |
|
t.Skip("too slow for testing.Short") |
|
} |
|
|
|
t.Parallel() |
|
a := NewTestAgent(t, "") |
|
defer a.Shutdown() |
|
|
|
queryPath := "/v1/catalog/service/api?filter=" + url.QueryEscape("ServiceMeta.somekey == somevalue") |
|
|
|
// Make sure an empty list is returned, not a nil |
|
{ |
|
req, _ := http.NewRequest("GET", queryPath, nil) |
|
resp := httptest.NewRecorder() |
|
obj, err := a.srv.CatalogServiceNodes(resp, req) |
|
require.NoError(t, err) |
|
|
|
assertIndex(t, resp) |
|
|
|
nodes := obj.(structs.ServiceNodes) |
|
require.Empty(t, nodes) |
|
} |
|
|
|
// Register node |
|
args := &structs.RegisterRequest{ |
|
Datacenter: "dc1", |
|
Node: "foo", |
|
Address: "127.0.0.1", |
|
Service: &structs.NodeService{ |
|
Service: "api", |
|
Meta: map[string]string{ |
|
"somekey": "somevalue", |
|
}, |
|
}, |
|
} |
|
|
|
var out struct{} |
|
require.NoError(t, a.RPC("Catalog.Register", args, &out)) |
|
|
|
// Register a second service for the node |
|
args = &structs.RegisterRequest{ |
|
Datacenter: "dc1", |
|
Node: "foo", |
|
Address: "127.0.0.1", |
|
Service: &structs.NodeService{ |
|
ID: "api2", |
|
Service: "api", |
|
Meta: map[string]string{ |
|
"somekey": "notvalue", |
|
}, |
|
}, |
|
SkipNodeUpdate: true, |
|
} |
|
|
|
require.NoError(t, a.RPC("Catalog.Register", args, &out)) |
|
|
|
req, _ := http.NewRequest("GET", queryPath, nil) |
|
resp := httptest.NewRecorder() |
|
obj, err := a.srv.CatalogServiceNodes(resp, req) |
|
require.NoError(t, err) |
|
assertIndex(t, resp) |
|
|
|
nodes := obj.(structs.ServiceNodes) |
|
require.Len(t, nodes, 1) |
|
} |
|
|
|
func TestCatalogServiceNodes_WanTranslation(t *testing.T) { |
|
if testing.Short() { |
|
t.Skip("too slow for testing.Short") |
|
} |
|
|
|
t.Parallel() |
|
a1 := NewTestAgent(t, ` |
|
datacenter = "dc1" |
|
translate_wan_addrs = true |
|
acl_datacenter = "" |
|
`) |
|
defer a1.Shutdown() |
|
|
|
a2 := NewTestAgent(t, ` |
|
datacenter = "dc2" |
|
translate_wan_addrs = true |
|
acl_datacenter = "" |
|
`) |
|
defer a2.Shutdown() |
|
|
|
// Wait for the WAN join. |
|
addr := fmt.Sprintf("127.0.0.1:%d", a1.Config.SerfPortWAN) |
|
_, err := a2.srv.agent.JoinWAN([]string{addr}) |
|
require.NoError(t, err) |
|
retry.Run(t, func(r *retry.R) { |
|
require.Len(r, a1.WANMembers(), 2) |
|
}) |
|
|
|
// Register a node with DC2. |
|
{ |
|
args := &structs.RegisterRequest{ |
|
Datacenter: "dc2", |
|
Node: "foo", |
|
Address: "127.0.0.1", |
|
TaggedAddresses: map[string]string{ |
|
"wan": "127.0.0.2", |
|
}, |
|
Service: &structs.NodeService{ |
|
Service: "http_wan_translation_test", |
|
Address: "127.0.0.1", |
|
Port: 8080, |
|
TaggedAddresses: map[string]structs.ServiceAddress{ |
|
"wan": { |
|
Address: "1.2.3.4", |
|
Port: 80, |
|
}, |
|
}, |
|
}, |
|
} |
|
|
|
var out struct{} |
|
require.NoError(t, a2.RPC("Catalog.Register", args, &out)) |
|
} |
|
|
|
// Query for the node in DC2 from DC1. |
|
req, _ := http.NewRequest("GET", "/v1/catalog/service/http_wan_translation_test?dc=dc2", nil) |
|
resp1 := httptest.NewRecorder() |
|
obj1, err1 := a1.srv.CatalogServiceNodes(resp1, req) |
|
require.NoError(t, err1) |
|
require.NoError(t, checkIndex(resp1)) |
|
|
|
// Expect that DC1 gives us a WAN address (since the node is in DC2). |
|
nodes1, ok := obj1.(structs.ServiceNodes) |
|
require.True(t, ok, "obj1 is not a structs.ServiceNodes") |
|
require.Len(t, nodes1, 1) |
|
node1 := nodes1[0] |
|
require.Equal(t, node1.Address, "127.0.0.2") |
|
require.Equal(t, node1.ServiceAddress, "1.2.3.4") |
|
require.Equal(t, node1.ServicePort, 80) |
|
|
|
// Query DC2 from DC2. |
|
resp2 := httptest.NewRecorder() |
|
obj2, err2 := a2.srv.CatalogServiceNodes(resp2, req) |
|
require.NoError(t, err2) |
|
require.NoError(t, checkIndex(resp2)) |
|
|
|
// Expect that DC2 gives us a local address (since the node is in DC2). |
|
nodes2, ok := obj2.(structs.ServiceNodes) |
|
require.True(t, ok, "obj2 is not a structs.ServiceNodes") |
|
require.Len(t, nodes2, 1) |
|
node2 := nodes2[0] |
|
require.Equal(t, node2.Address, "127.0.0.1") |
|
require.Equal(t, node2.ServiceAddress, "127.0.0.1") |
|
require.Equal(t, node2.ServicePort, 8080) |
|
} |
|
|
|
func TestCatalogServiceNodes_DistanceSort(t *testing.T) { |
|
if testing.Short() { |
|
t.Skip("too slow for testing.Short") |
|
} |
|
|
|
t.Parallel() |
|
a := NewTestAgent(t, "") |
|
defer a.Shutdown() |
|
testrpc.WaitForLeader(t, a.RPC, "dc1") |
|
|
|
// Register nodes. |
|
args := &structs.RegisterRequest{ |
|
Datacenter: "dc1", |
|
Node: "bar", |
|
Address: "127.0.0.1", |
|
Service: &structs.NodeService{ |
|
Service: "api", |
|
Tags: []string{"a"}, |
|
}, |
|
} |
|
var out struct{} |
|
if err := a.RPC("Catalog.Register", args, &out); err != nil { |
|
t.Fatalf("err: %v", err) |
|
} |
|
|
|
req, _ := http.NewRequest("GET", "/v1/catalog/service/api?tag=a", nil) |
|
args = &structs.RegisterRequest{ |
|
Datacenter: "dc1", |
|
Node: "foo", |
|
Address: "127.0.0.2", |
|
Service: &structs.NodeService{ |
|
Service: "api", |
|
Tags: []string{"a"}, |
|
}, |
|
} |
|
if err := a.RPC("Catalog.Register", args, &out); err != nil { |
|
t.Fatalf("err: %v", err) |
|
} |
|
|
|
// Nobody has coordinates set so this will still return them in the |
|
// order they are indexed. |
|
req, _ = http.NewRequest("GET", "/v1/catalog/service/api?tag=a&near=foo", nil) |
|
resp := httptest.NewRecorder() |
|
obj, err := a.srv.CatalogServiceNodes(resp, req) |
|
if err != nil { |
|
t.Fatalf("err: %v", err) |
|
} |
|
|
|
assertIndex(t, resp) |
|
nodes := obj.(structs.ServiceNodes) |
|
if len(nodes) != 2 { |
|
t.Fatalf("bad: %v", obj) |
|
} |
|
if nodes[0].Node != "bar" { |
|
t.Fatalf("bad: %v", nodes) |
|
} |
|
if nodes[1].Node != "foo" { |
|
t.Fatalf("bad: %v", nodes) |
|
} |
|
|
|
// Send an update for the node and wait for it to get applied. |
|
arg := structs.CoordinateUpdateRequest{ |
|
Datacenter: "dc1", |
|
Node: "foo", |
|
Coord: coordinate.NewCoordinate(coordinate.DefaultConfig()), |
|
} |
|
if err := a.RPC("Coordinate.Update", &arg, &out); err != nil { |
|
t.Fatalf("err: %v", err) |
|
} |
|
|
|
// Eventually foo should move to the front of the line. |
|
retry.Run(t, func(r *retry.R) { |
|
req, _ = http.NewRequest("GET", "/v1/catalog/service/api?tag=a&near=foo", nil) |
|
resp = httptest.NewRecorder() |
|
obj, err = a.srv.CatalogServiceNodes(resp, req) |
|
if err != nil { |
|
r.Fatalf("err: %v", err) |
|
} |
|
|
|
assertIndex(t, resp) |
|
nodes = obj.(structs.ServiceNodes) |
|
if len(nodes) != 2 { |
|
r.Fatalf("bad: %v", obj) |
|
} |
|
if nodes[0].Node != "foo" { |
|
r.Fatalf("bad: %v", nodes) |
|
} |
|
if nodes[1].Node != "bar" { |
|
r.Fatalf("bad: %v", nodes) |
|
} |
|
}) |
|
} |
|
|
|
// Test that connect proxies can be queried via /v1/catalog/service/:service |
|
// directly and that their results contain the proxy fields. |
|
func TestCatalogServiceNodes_ConnectProxy(t *testing.T) { |
|
if testing.Short() { |
|
t.Skip("too slow for testing.Short") |
|
} |
|
|
|
t.Parallel() |
|
|
|
a := NewTestAgent(t, "") |
|
defer a.Shutdown() |
|
testrpc.WaitForLeader(t, a.RPC, "dc1") |
|
|
|
// Register |
|
args := structs.TestRegisterRequestProxy(t) |
|
var out struct{} |
|
assert.Nil(t, a.RPC("Catalog.Register", args, &out)) |
|
|
|
req, _ := http.NewRequest("GET", fmt.Sprintf( |
|
"/v1/catalog/service/%s", args.Service.Service), nil) |
|
resp := httptest.NewRecorder() |
|
obj, err := a.srv.CatalogServiceNodes(resp, req) |
|
assert.Nil(t, err) |
|
assertIndex(t, resp) |
|
|
|
nodes := obj.(structs.ServiceNodes) |
|
assert.Len(t, nodes, 1) |
|
assert.Equal(t, structs.ServiceKindConnectProxy, nodes[0].ServiceKind) |
|
assert.Equal(t, args.Service.Proxy, nodes[0].ServiceProxy) |
|
} |
|
|
|
func registerService(t *testing.T, a *TestAgent) (registerServiceReq *structs.RegisterRequest) { |
|
t.Helper() |
|
entMeta := acl.DefaultEnterpriseMeta() |
|
registerServiceReq = structs.TestRegisterRequestProxy(t) |
|
registerServiceReq.EnterpriseMeta = *entMeta |
|
registerServiceReq.Service.EnterpriseMeta = *entMeta |
|
registerServiceReq.Service.Proxy.Upstreams = structs.TestAddDefaultsToUpstreams(t, registerServiceReq.Service.Proxy.Upstreams, *entMeta) |
|
registerServiceReq.Check = &structs.HealthCheck{ |
|
Node: registerServiceReq.Node, |
|
Name: "check1", |
|
} |
|
|
|
var out struct{} |
|
require.NoError(t, a.RPC("Catalog.Register", registerServiceReq, &out)) |
|
|
|
return |
|
} |
|
|
|
func registerProxyDefaults(t *testing.T, a *TestAgent) (proxyGlobalEntry structs.ProxyConfigEntry) { |
|
t.Helper() |
|
// Register proxy-defaults |
|
proxyGlobalEntry = structs.ProxyConfigEntry{ |
|
Kind: structs.ProxyDefaults, |
|
Name: structs.ProxyConfigGlobal, |
|
Mode: structs.ProxyModeDirect, |
|
Config: map[string]interface{}{ |
|
"local_connect_timeout_ms": uint64(1000), |
|
"handshake_timeout_ms": uint64(1000), |
|
}, |
|
EnterpriseMeta: *acl.DefaultEnterpriseMeta(), |
|
} |
|
proxyDefaultsConfigEntryReq := &structs.ConfigEntryRequest{ |
|
Op: structs.ConfigEntryUpsert, |
|
Datacenter: "dc1", |
|
Entry: &proxyGlobalEntry, |
|
} |
|
var proxyDefaultsConfigEntryResp bool |
|
require.NoError(t, a.RPC("ConfigEntry.Apply", &proxyDefaultsConfigEntryReq, &proxyDefaultsConfigEntryResp)) |
|
return |
|
} |
|
|
|
func registerServiceDefaults(t *testing.T, a *TestAgent, serviceName string) (serviceDefaultsConfigEntry structs.ServiceConfigEntry) { |
|
t.Helper() |
|
limits := 512 |
|
serviceDefaultsConfigEntry = structs.ServiceConfigEntry{ |
|
Kind: structs.ServiceDefaults, |
|
Name: serviceName, |
|
Mode: structs.ProxyModeTransparent, |
|
UpstreamConfig: &structs.UpstreamConfiguration{ |
|
Defaults: &structs.UpstreamConfig{ |
|
MeshGateway: structs.MeshGatewayConfig{ |
|
Mode: structs.MeshGatewayModeLocal, |
|
}, |
|
Limits: &structs.UpstreamLimits{ |
|
MaxConnections: &limits, |
|
}, |
|
}, |
|
}, |
|
EnterpriseMeta: *acl.DefaultEnterpriseMeta(), |
|
} |
|
serviceDefaultsConfigEntryReq := &structs.ConfigEntryRequest{ |
|
Op: structs.ConfigEntryUpsert, |
|
Datacenter: "dc1", |
|
Entry: &serviceDefaultsConfigEntry, |
|
} |
|
var serviceDefaultsConfigEntryResp bool |
|
require.NoError(t, a.RPC("ConfigEntry.Apply", &serviceDefaultsConfigEntryReq, &serviceDefaultsConfigEntryResp)) |
|
return |
|
} |
|
|
|
func validateMergeCentralConfigResponse(t *testing.T, v *structs.ServiceNode, |
|
registerServiceReq *structs.RegisterRequest, |
|
proxyGlobalEntry structs.ProxyConfigEntry, |
|
serviceDefaultsConfigEntry structs.ServiceConfigEntry) { |
|
|
|
t.Helper() |
|
require.Equal(t, registerServiceReq.Service.Service, v.ServiceName) |
|
// validate proxy global defaults are resolved in the merged service config |
|
require.Equal(t, proxyGlobalEntry.Config, v.ServiceProxy.Config) |
|
// validate service defaults override proxy-defaults/global |
|
require.NotEqual(t, proxyGlobalEntry.Mode, v.ServiceProxy.Mode) |
|
require.Equal(t, serviceDefaultsConfigEntry.Mode, v.ServiceProxy.Mode) |
|
// validate service defaults are resolved in the merged service config |
|
// expected number of upstreams = (number of upstreams defined in the register request proxy config + |
|
// 1 centrally configured default from service defaults) |
|
require.Equal(t, len(registerServiceReq.Service.Proxy.Upstreams)+1, len(v.ServiceProxy.Upstreams)) |
|
for _, up := range v.ServiceProxy.Upstreams { |
|
if up.DestinationType != "" && up.DestinationType != structs.UpstreamDestTypeService { |
|
continue |
|
} |
|
require.Contains(t, up.Config, "limits") |
|
upstreamLimits := up.Config["limits"].(*structs.UpstreamLimits) |
|
require.Equal(t, serviceDefaultsConfigEntry.UpstreamConfig.Defaults.Limits.MaxConnections, upstreamLimits.MaxConnections) |
|
require.Equal(t, serviceDefaultsConfigEntry.UpstreamConfig.Defaults.MeshGateway.Mode, up.MeshGateway.Mode) |
|
} |
|
} |
|
|
|
func TestListServiceNodes_MergeCentralConfig(t *testing.T) { |
|
if testing.Short() { |
|
t.Skip("too slow for testing.Short") |
|
} |
|
|
|
t.Parallel() |
|
|
|
a := NewTestAgent(t, "") |
|
defer a.Shutdown() |
|
testrpc.WaitForLeader(t, a.RPC, "dc1") |
|
|
|
// Register the service |
|
registerServiceReq := registerService(t, a) |
|
// Register proxy-defaults |
|
proxyGlobalEntry := registerProxyDefaults(t, a) |
|
// Register service-defaults |
|
serviceDefaultsConfigEntry := registerServiceDefaults(t, a, registerServiceReq.Service.Proxy.DestinationServiceName) |
|
|
|
type testCase struct { |
|
testCaseName string |
|
serviceName string |
|
connect bool |
|
} |
|
|
|
run := func(t *testing.T, tc testCase) { |
|
url := fmt.Sprintf("/v1/catalog/service/%s?merge-central-config", tc.serviceName) |
|
if tc.connect { |
|
url = fmt.Sprintf("/v1/catalog/connect/%s?merge-central-config", tc.serviceName) |
|
} |
|
req, _ := http.NewRequest("GET", url, nil) |
|
resp := httptest.NewRecorder() |
|
var obj interface{} |
|
var err error |
|
if tc.connect { |
|
obj, err = a.srv.CatalogConnectServiceNodes(resp, req) |
|
} else { |
|
obj, err = a.srv.CatalogServiceNodes(resp, req) |
|
} |
|
|
|
require.NoError(t, err) |
|
assertIndex(t, resp) |
|
|
|
serviceNodes := obj.(structs.ServiceNodes) |
|
|
|
// validate response |
|
require.Len(t, serviceNodes, 1) |
|
v := serviceNodes[0] |
|
|
|
validateMergeCentralConfigResponse(t, v, registerServiceReq, proxyGlobalEntry, serviceDefaultsConfigEntry) |
|
} |
|
testCases := []testCase{ |
|
{ |
|
testCaseName: "List service instances with merge-central-config", |
|
serviceName: registerServiceReq.Service.Service, |
|
}, |
|
{ |
|
testCaseName: "List connect capable service instances with merge-central-config", |
|
serviceName: registerServiceReq.Service.Proxy.DestinationServiceName, |
|
connect: true, |
|
}, |
|
} |
|
for _, tc := range testCases { |
|
t.Run(tc.testCaseName, func(t *testing.T) { |
|
run(t, tc) |
|
}) |
|
} |
|
} |
|
|
|
func TestCatalogServiceNodes_MergeCentralConfigBlocking(t *testing.T) { |
|
if testing.Short() { |
|
t.Skip("too slow for testing.Short") |
|
} |
|
|
|
t.Parallel() |
|
a := NewTestAgent(t, "") |
|
defer a.Shutdown() |
|
|
|
testrpc.WaitForLeader(t, a.RPC, "dc1") |
|
|
|
// Register the service |
|
registerServiceReq := registerService(t, a) |
|
// Register proxy-defaults |
|
proxyGlobalEntry := registerProxyDefaults(t, a) |
|
|
|
// Run the query |
|
rpcReq := structs.ServiceSpecificRequest{ |
|
Datacenter: "dc1", |
|
ServiceName: registerServiceReq.Service.Service, |
|
MergeCentralConfig: true, |
|
} |
|
var rpcResp structs.IndexedServiceNodes |
|
require.NoError(t, a.RPC("Catalog.ServiceNodes", &rpcReq, &rpcResp)) |
|
|
|
require.Len(t, rpcResp.ServiceNodes, 1) |
|
serviceNode := rpcResp.ServiceNodes[0] |
|
require.Equal(t, registerServiceReq.Service.Service, serviceNode.ServiceName) |
|
// validate proxy global defaults are resolved in the merged service config |
|
require.Equal(t, proxyGlobalEntry.Config, serviceNode.ServiceProxy.Config) |
|
require.Equal(t, proxyGlobalEntry.Mode, serviceNode.ServiceProxy.Mode) |
|
|
|
// Async cause a change - register service defaults |
|
waitIndex := rpcResp.Index |
|
start := time.Now() |
|
var serviceDefaultsConfigEntry structs.ServiceConfigEntry |
|
go func() { |
|
time.Sleep(100 * time.Millisecond) |
|
// Register service-defaults |
|
serviceDefaultsConfigEntry = registerServiceDefaults(t, a, registerServiceReq.Service.Proxy.DestinationServiceName) |
|
}() |
|
|
|
const waitDuration = 3 * time.Second |
|
RUN_BLOCKING_QUERY: |
|
url := fmt.Sprintf("/v1/catalog/service/%s?merge-central-config&wait=%s&index=%d", |
|
registerServiceReq.Service.Service, waitDuration.String(), waitIndex) |
|
req, _ := http.NewRequest("GET", url, nil) |
|
resp := httptest.NewRecorder() |
|
obj, err := a.srv.CatalogServiceNodes(resp, req) |
|
|
|
require.NoError(t, err) |
|
assertIndex(t, resp) |
|
|
|
elapsed := time.Since(start) |
|
idx := getIndex(t, resp) |
|
if idx < waitIndex { |
|
t.Fatalf("bad index returned: %v", idx) |
|
} else if idx == waitIndex { |
|
if elapsed > waitDuration { |
|
// This should prevent the loop from running longer than the waitDuration |
|
t.Fatalf("too slow: %v", elapsed) |
|
} |
|
goto RUN_BLOCKING_QUERY |
|
} |
|
// Should block at least 100ms before getting the changed results |
|
if elapsed < 100*time.Millisecond { |
|
t.Fatalf("too fast: %v", elapsed) |
|
} |
|
|
|
serviceNodes := obj.(structs.ServiceNodes) |
|
|
|
// validate response |
|
require.Len(t, serviceNodes, 1) |
|
v := serviceNodes[0] |
|
|
|
validateMergeCentralConfigResponse(t, v, registerServiceReq, proxyGlobalEntry, serviceDefaultsConfigEntry) |
|
} |
|
|
|
// Test that the Connect-compatible endpoints can be queried for a |
|
// service via /v1/catalog/connect/:service. |
|
func TestCatalogConnectServiceNodes_good(t *testing.T) { |
|
if testing.Short() { |
|
t.Skip("too slow for testing.Short") |
|
} |
|
|
|
t.Parallel() |
|
|
|
a := NewTestAgent(t, "") |
|
defer a.Shutdown() |
|
testrpc.WaitForLeader(t, a.RPC, "dc1") |
|
|
|
// Register |
|
args := structs.TestRegisterRequestProxy(t) |
|
args.Service.Address = "127.0.0.55" |
|
var out struct{} |
|
assert.Nil(t, a.RPC("Catalog.Register", args, &out)) |
|
|
|
req, _ := http.NewRequest("GET", fmt.Sprintf( |
|
"/v1/catalog/connect/%s", args.Service.Proxy.DestinationServiceName), nil) |
|
resp := httptest.NewRecorder() |
|
obj, err := a.srv.CatalogConnectServiceNodes(resp, req) |
|
assert.Nil(t, err) |
|
assertIndex(t, resp) |
|
|
|
nodes := obj.(structs.ServiceNodes) |
|
assert.Len(t, nodes, 1) |
|
assert.Equal(t, structs.ServiceKindConnectProxy, nodes[0].ServiceKind) |
|
assert.Equal(t, args.Service.Address, nodes[0].ServiceAddress) |
|
assert.Equal(t, args.Service.Proxy, nodes[0].ServiceProxy) |
|
} |
|
|
|
func TestCatalogConnectServiceNodes_Filter(t *testing.T) { |
|
if testing.Short() { |
|
t.Skip("too slow for testing.Short") |
|
} |
|
|
|
t.Parallel() |
|
|
|
a := NewTestAgent(t, "") |
|
defer a.Shutdown() |
|
testrpc.WaitForLeader(t, a.RPC, "dc1") |
|
|
|
// Register |
|
args := structs.TestRegisterRequestProxy(t) |
|
args.Service.Address = "127.0.0.55" |
|
var out struct{} |
|
require.NoError(t, a.RPC("Catalog.Register", args, &out)) |
|
|
|
args = structs.TestRegisterRequestProxy(t) |
|
args.Service.Address = "127.0.0.55" |
|
args.Service.Meta = map[string]string{ |
|
"version": "2", |
|
} |
|
args.Service.ID = "web-proxy2" |
|
args.SkipNodeUpdate = true |
|
require.NoError(t, a.RPC("Catalog.Register", args, &out)) |
|
|
|
req, _ := http.NewRequest("GET", fmt.Sprintf( |
|
"/v1/catalog/connect/%s?filter=%s", |
|
args.Service.Proxy.DestinationServiceName, |
|
url.QueryEscape("ServiceMeta.version == 2")), nil) |
|
resp := httptest.NewRecorder() |
|
obj, err := a.srv.CatalogConnectServiceNodes(resp, req) |
|
require.NoError(t, err) |
|
assertIndex(t, resp) |
|
|
|
nodes := obj.(structs.ServiceNodes) |
|
require.Len(t, nodes, 1) |
|
require.Equal(t, structs.ServiceKindConnectProxy, nodes[0].ServiceKind) |
|
require.Equal(t, args.Service.Address, nodes[0].ServiceAddress) |
|
require.Equal(t, args.Service.Proxy, nodes[0].ServiceProxy) |
|
} |
|
|
|
func TestCatalogNodeServices(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") |
|
|
|
// Register node with a regular service and connect proxy |
|
args := &structs.RegisterRequest{ |
|
Datacenter: "dc1", |
|
Node: "foo", |
|
Address: "127.0.0.1", |
|
Service: &structs.NodeService{ |
|
Service: "api", |
|
Tags: []string{"a"}, |
|
}, |
|
} |
|
|
|
var out struct{} |
|
if err := a.RPC("Catalog.Register", args, &out); err != nil { |
|
t.Fatalf("err: %v", err) |
|
} |
|
|
|
// Register a connect proxy |
|
args.Service = structs.TestNodeServiceProxy(t) |
|
require.NoError(t, a.RPC("Catalog.Register", args, &out)) |
|
|
|
req, _ := http.NewRequest("GET", "/v1/catalog/node/foo?dc=dc1", nil) |
|
resp := httptest.NewRecorder() |
|
obj, err := a.srv.CatalogNodeServices(resp, req) |
|
if err != nil { |
|
t.Fatalf("err: %v", err) |
|
} |
|
assertIndex(t, resp) |
|
|
|
services := obj.(*structs.NodeServices) |
|
if len(services.Services) != 2 { |
|
t.Fatalf("bad: %v", obj) |
|
} |
|
|
|
// Proxy service should have it's config intact |
|
require.Equal(t, args.Service.Proxy, services.Services["web-proxy"].Proxy) |
|
} |
|
|
|
func TestCatalogNodeServiceList(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") |
|
|
|
// Register node with a regular service and connect proxy |
|
args := &structs.RegisterRequest{ |
|
Datacenter: "dc1", |
|
Node: "foo", |
|
Address: "127.0.0.1", |
|
Service: &structs.NodeService{ |
|
Service: "api", |
|
Tags: []string{"a"}, |
|
}, |
|
} |
|
|
|
var out struct{} |
|
if err := a.RPC("Catalog.Register", args, &out); err != nil { |
|
t.Fatalf("err: %v", err) |
|
} |
|
|
|
// Register a connect proxy |
|
args.Service = structs.TestNodeServiceProxy(t) |
|
require.NoError(t, a.RPC("Catalog.Register", args, &out)) |
|
|
|
req, _ := http.NewRequest("GET", "/v1/catalog/node-services/foo?dc=dc1", nil) |
|
resp := httptest.NewRecorder() |
|
obj, err := a.srv.CatalogNodeServiceList(resp, req) |
|
if err != nil { |
|
t.Fatalf("err: %v", err) |
|
} |
|
assertIndex(t, resp) |
|
|
|
services := obj.(*structs.NodeServiceList) |
|
if len(services.Services) != 2 { |
|
t.Fatalf("bad: %v", obj) |
|
} |
|
|
|
var proxySvc *structs.NodeService |
|
for _, svc := range services.Services { |
|
if svc.ID == "web-proxy" { |
|
proxySvc = svc |
|
} |
|
} |
|
require.NotNil(t, proxySvc, "Missing proxy service registration") |
|
// Proxy service should have it's config intact |
|
require.Equal(t, args.Service.Proxy, proxySvc.Proxy) |
|
} |
|
|
|
func TestCatalogNodeServices_Filter(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") |
|
|
|
// Register node with a regular service and connect proxy |
|
args := &structs.RegisterRequest{ |
|
Datacenter: "dc1", |
|
Node: "foo", |
|
Address: "127.0.0.1", |
|
Service: &structs.NodeService{ |
|
Service: "api", |
|
Tags: []string{"a"}, |
|
}, |
|
} |
|
|
|
var out struct{} |
|
require.NoError(t, a.RPC("Catalog.Register", args, &out)) |
|
|
|
// Register a connect proxy |
|
args.Service = structs.TestNodeServiceProxy(t) |
|
require.NoError(t, a.RPC("Catalog.Register", args, &out)) |
|
|
|
req, _ := http.NewRequest("GET", "/v1/catalog/node/foo?dc=dc1&filter="+url.QueryEscape("Kind == `connect-proxy`"), nil) |
|
resp := httptest.NewRecorder() |
|
obj, err := a.srv.CatalogNodeServices(resp, req) |
|
require.NoError(t, err) |
|
assertIndex(t, resp) |
|
|
|
services := obj.(*structs.NodeServices) |
|
require.Len(t, services.Services, 1) |
|
|
|
// Proxy service should have it's config intact |
|
require.Equal(t, args.Service.Proxy, services.Services["web-proxy"].Proxy) |
|
} |
|
|
|
// Test that the services on a node contain all the Connect proxies on |
|
// the node as well with their fields properly populated. |
|
func TestCatalogNodeServices_ConnectProxy(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") |
|
|
|
// Register |
|
args := structs.TestRegisterRequestProxy(t) |
|
var out struct{} |
|
assert.Nil(t, a.RPC("Catalog.Register", args, &out)) |
|
|
|
req, _ := http.NewRequest("GET", fmt.Sprintf( |
|
"/v1/catalog/node/%s", args.Node), nil) |
|
resp := httptest.NewRecorder() |
|
obj, err := a.srv.CatalogNodeServices(resp, req) |
|
assert.Nil(t, err) |
|
assertIndex(t, resp) |
|
|
|
ns := obj.(*structs.NodeServices) |
|
assert.Len(t, ns.Services, 1) |
|
v := ns.Services[args.Service.Service] |
|
assert.Equal(t, structs.ServiceKindConnectProxy, v.Kind) |
|
} |
|
|
|
func TestCatalogNodeServices_WanTranslation(t *testing.T) { |
|
if testing.Short() { |
|
t.Skip("too slow for testing.Short") |
|
} |
|
|
|
t.Parallel() |
|
a1 := NewTestAgent(t, ` |
|
datacenter = "dc1" |
|
translate_wan_addrs = true |
|
acl_datacenter = "" |
|
`) |
|
defer a1.Shutdown() |
|
testrpc.WaitForTestAgent(t, a1.RPC, "dc1") |
|
|
|
a2 := NewTestAgent(t, ` |
|
datacenter = "dc2" |
|
translate_wan_addrs = true |
|
acl_datacenter = "" |
|
`) |
|
defer a2.Shutdown() |
|
testrpc.WaitForTestAgent(t, a2.RPC, "dc2") |
|
|
|
// Wait for the WAN join. |
|
addr := fmt.Sprintf("127.0.0.1:%d", a1.Config.SerfPortWAN) |
|
_, err := a2.srv.agent.JoinWAN([]string{addr}) |
|
require.NoError(t, err) |
|
retry.Run(t, func(r *retry.R) { |
|
require.Len(r, a1.WANMembers(), 2) |
|
}) |
|
|
|
// Register a node with DC2. |
|
{ |
|
args := &structs.RegisterRequest{ |
|
Datacenter: "dc2", |
|
Node: "foo", |
|
Address: "127.0.0.1", |
|
TaggedAddresses: map[string]string{ |
|
"wan": "127.0.0.2", |
|
}, |
|
Service: &structs.NodeService{ |
|
Service: "http_wan_translation_test", |
|
Address: "127.0.0.1", |
|
Port: 8080, |
|
TaggedAddresses: map[string]structs.ServiceAddress{ |
|
"wan": { |
|
Address: "1.2.3.4", |
|
Port: 80, |
|
}, |
|
}, |
|
}, |
|
} |
|
|
|
var out struct{} |
|
require.NoError(t, a2.RPC("Catalog.Register", args, &out)) |
|
} |
|
|
|
// Query for the node in DC2 from DC1. |
|
req, _ := http.NewRequest("GET", "/v1/catalog/node/foo?dc=dc2", nil) |
|
resp1 := httptest.NewRecorder() |
|
obj1, err1 := a1.srv.CatalogNodeServices(resp1, req) |
|
require.NoError(t, err1) |
|
require.NoError(t, checkIndex(resp1)) |
|
|
|
// Expect that DC1 gives us a WAN address (since the node is in DC2). |
|
service1, ok := obj1.(*structs.NodeServices) |
|
require.True(t, ok, "obj1 is not a *structs.NodeServices") |
|
require.NotNil(t, service1.Node) |
|
require.Equal(t, service1.Node.Address, "127.0.0.2") |
|
require.Len(t, service1.Services, 1) |
|
ns1, ok := service1.Services["http_wan_translation_test"] |
|
require.True(t, ok, "Missing service http_wan_translation_test") |
|
require.Equal(t, "1.2.3.4", ns1.Address) |
|
require.Equal(t, 80, ns1.Port) |
|
|
|
// Query DC2 from DC2. |
|
resp2 := httptest.NewRecorder() |
|
obj2, err2 := a2.srv.CatalogNodeServices(resp2, req) |
|
require.NoError(t, err2) |
|
require.NoError(t, checkIndex(resp2)) |
|
|
|
// Expect that DC2 gives us a private address (since the node is in DC2). |
|
service2 := obj2.(*structs.NodeServices) |
|
require.True(t, ok, "obj2 is not a *structs.NodeServices") |
|
require.NotNil(t, service2.Node) |
|
require.Equal(t, service2.Node.Address, "127.0.0.1") |
|
require.Len(t, service2.Services, 1) |
|
ns2, ok := service2.Services["http_wan_translation_test"] |
|
require.True(t, ok, "Missing service http_wan_translation_test") |
|
require.Equal(t, ns2.Address, "127.0.0.1") |
|
require.Equal(t, ns2.Port, 8080) |
|
} |
|
|
|
func TestCatalog_GatewayServices_Terminating(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") |
|
|
|
// Register a service to be covered by the wildcard in the config entry |
|
args := structs.TestRegisterRequest(t) |
|
args.Service.Service = "redis" |
|
args.Check = &structs.HealthCheck{ |
|
Name: "redis", |
|
Status: api.HealthPassing, |
|
ServiceID: args.Service.Service, |
|
} |
|
var out struct{} |
|
assert.NoError(t, a.RPC("Catalog.Register", &args, &out)) |
|
|
|
// Associate the gateway and api/redis services |
|
entryArgs := &structs.ConfigEntryRequest{ |
|
Op: structs.ConfigEntryUpsert, |
|
Datacenter: "dc1", |
|
Entry: &structs.TerminatingGatewayConfigEntry{ |
|
Kind: "terminating-gateway", |
|
Name: "terminating", |
|
Services: []structs.LinkedService{ |
|
{ |
|
Name: "api", |
|
CAFile: "api/ca.crt", |
|
CertFile: "api/client.crt", |
|
KeyFile: "api/client.key", |
|
SNI: "my-domain", |
|
}, |
|
{ |
|
Name: "*", |
|
CAFile: "ca.crt", |
|
CertFile: "client.crt", |
|
KeyFile: "client.key", |
|
SNI: "my-alt-domain", |
|
}, |
|
}, |
|
}, |
|
} |
|
var entryResp bool |
|
assert.NoError(t, a.RPC("ConfigEntry.Apply", &entryArgs, &entryResp)) |
|
|
|
retry.Run(t, func(r *retry.R) { |
|
req, _ := http.NewRequest("GET", "/v1/catalog/gateway-services/terminating", nil) |
|
resp := httptest.NewRecorder() |
|
obj, err := a.srv.CatalogGatewayServices(resp, req) |
|
assert.NoError(r, err) |
|
|
|
header := resp.Header().Get("X-Consul-Index") |
|
if header == "" || header == "0" { |
|
r.Fatalf("Bad: %v", header) |
|
} |
|
|
|
gatewayServices := obj.(structs.GatewayServices) |
|
|
|
expect := structs.GatewayServices{ |
|
{ |
|
Service: structs.NewServiceName("api", nil), |
|
Gateway: structs.NewServiceName("terminating", nil), |
|
GatewayKind: structs.ServiceKindTerminatingGateway, |
|
CAFile: "api/ca.crt", |
|
CertFile: "api/client.crt", |
|
KeyFile: "api/client.key", |
|
SNI: "my-domain", |
|
}, |
|
{ |
|
Service: structs.NewServiceName("redis", nil), |
|
Gateway: structs.NewServiceName("terminating", nil), |
|
GatewayKind: structs.ServiceKindTerminatingGateway, |
|
CAFile: "ca.crt", |
|
CertFile: "client.crt", |
|
KeyFile: "client.key", |
|
SNI: "my-alt-domain", |
|
FromWildcard: true, |
|
}, |
|
} |
|
|
|
// Ignore raft index for equality |
|
for _, s := range gatewayServices { |
|
s.RaftIndex = structs.RaftIndex{} |
|
} |
|
assert.Equal(r, expect, gatewayServices) |
|
}) |
|
} |
|
|
|
func TestCatalog_GatewayServices_Ingress(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") |
|
|
|
// Associate an ingress gateway with api/redis |
|
entryArgs := &structs.ConfigEntryRequest{ |
|
Op: structs.ConfigEntryUpsert, |
|
Datacenter: "dc1", |
|
Entry: &structs.IngressGatewayConfigEntry{ |
|
Kind: "ingress-gateway", |
|
Name: "ingress", |
|
Listeners: []structs.IngressListener{ |
|
{ |
|
Port: 8888, |
|
Services: []structs.IngressService{ |
|
{ |
|
Name: "api", |
|
}, |
|
}, |
|
}, |
|
{ |
|
Port: 9999, |
|
Services: []structs.IngressService{ |
|
{ |
|
Name: "redis", |
|
}, |
|
}, |
|
}, |
|
}, |
|
}, |
|
} |
|
|
|
var entryResp bool |
|
require.NoError(t, a.RPC("ConfigEntry.Apply", &entryArgs, &entryResp)) |
|
|
|
retry.Run(t, func(r *retry.R) { |
|
req, _ := http.NewRequest("GET", "/v1/catalog/gateway-services/ingress", nil) |
|
resp := httptest.NewRecorder() |
|
obj, err := a.srv.CatalogGatewayServices(resp, req) |
|
require.NoError(r, err) |
|
|
|
header := resp.Header().Get("X-Consul-Index") |
|
if header == "" || header == "0" { |
|
r.Fatalf("Bad: %v", header) |
|
} |
|
|
|
gatewayServices := obj.(structs.GatewayServices) |
|
|
|
expect := structs.GatewayServices{ |
|
{ |
|
Service: structs.NewServiceName("api", nil), |
|
Gateway: structs.NewServiceName("ingress", nil), |
|
GatewayKind: structs.ServiceKindIngressGateway, |
|
Protocol: "tcp", |
|
Port: 8888, |
|
}, |
|
{ |
|
Service: structs.NewServiceName("redis", nil), |
|
Gateway: structs.NewServiceName("ingress", nil), |
|
GatewayKind: structs.ServiceKindIngressGateway, |
|
Protocol: "tcp", |
|
Port: 9999, |
|
}, |
|
} |
|
|
|
// Ignore raft index for equality |
|
for _, s := range gatewayServices { |
|
s.RaftIndex = structs.RaftIndex{} |
|
} |
|
require.Equal(r, expect, gatewayServices) |
|
}) |
|
}
|
|
|