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.
1218 lines
30 KiB
1218 lines
30 KiB
package agent |
|
|
|
import ( |
|
"fmt" |
|
"net/http" |
|
"net/http/httptest" |
|
"net/url" |
|
"testing" |
|
"time" |
|
|
|
"github.com/hashicorp/consul/agent/structs" |
|
"github.com/hashicorp/consul/sdk/testutil/retry" |
|
"github.com/hashicorp/consul/testrpc" |
|
"github.com/hashicorp/serf/coordinate" |
|
"github.com/stretchr/testify/assert" |
|
"github.com/stretchr/testify/require" |
|
) |
|
|
|
func TestCatalogRegister_Service_InvalidAddress(t *testing.T) { |
|
t.Parallel() |
|
a := NewTestAgent(t, t.Name(), "") |
|
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) { |
|
t.Parallel() |
|
a := NewTestAgent(t, t.Name(), "") |
|
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) { |
|
t.Parallel() |
|
a := NewTestAgent(t, t.Name(), "") |
|
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) { |
|
t.Parallel() |
|
a := NewTestAgent(t, t.Name(), "") |
|
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) { |
|
t.Parallel() |
|
a := NewTestAgent(t, t.Name(), "") |
|
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) { |
|
t.Parallel() |
|
a := NewTestAgent(t, t.Name(), "") |
|
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) { |
|
t.Parallel() |
|
a1 := NewTestAgent(t, t.Name(), ` |
|
datacenter = "dc1" |
|
translate_wan_addrs = true |
|
acl_datacenter = "" |
|
`) |
|
defer a1.Shutdown() |
|
testrpc.WaitForTestAgent(t, a1.RPC, "dc1") |
|
|
|
a2 := NewTestAgent(t, t.Name(), ` |
|
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) { |
|
t.Parallel() |
|
a := NewTestAgent(t, t.Name(), "") |
|
defer a.Shutdown() |
|
testrpc.WaitForTestAgent(t, a.RPC, "dc1") |
|
|
|
// Register node |
|
args := &structs.DCSpecificRequest{ |
|
Datacenter: "dc1", |
|
} |
|
|
|
var out structs.IndexedNodes |
|
if err := a.RPC("Catalog.ListNodes", *args, &out); err != nil { |
|
t.Fatalf("err: %v", err) |
|
} |
|
|
|
// t.Fatal must be called from the main go routine |
|
// of the test. Because of this we cannot call |
|
// t.Fatal from within the go routines and use |
|
// an error channel instead. |
|
errch := make(chan error, 2) |
|
go func() { |
|
testrpc.WaitForTestAgent(t, a.RPC, "dc1") |
|
start := time.Now() |
|
|
|
// register a service after the blocking call |
|
// in order to unblock it. |
|
time.AfterFunc(100*time.Millisecond, func() { |
|
args := &structs.RegisterRequest{ |
|
Datacenter: "dc1", |
|
Node: "foo", |
|
Address: "127.0.0.1", |
|
} |
|
var out struct{} |
|
errch <- a.RPC("Catalog.Register", args, &out) |
|
}) |
|
|
|
// now block |
|
req, _ := http.NewRequest("GET", fmt.Sprintf("/v1/catalog/nodes?wait=3s&index=%d", out.Index+1), nil) |
|
resp := httptest.NewRecorder() |
|
obj, err := a.srv.CatalogNodes(resp, req) |
|
if err != nil { |
|
errch <- err |
|
} |
|
|
|
// Should block for a while |
|
if d := time.Since(start); d < 50*time.Millisecond { |
|
errch <- fmt.Errorf("too fast: %v", d) |
|
} |
|
|
|
if idx := getIndex(t, resp); idx <= out.Index { |
|
errch <- fmt.Errorf("bad: %v", idx) |
|
} |
|
|
|
nodes := obj.(structs.Nodes) |
|
if len(nodes) != 2 { |
|
errch <- fmt.Errorf("bad: %v", obj) |
|
} |
|
errch <- nil |
|
}() |
|
|
|
// wait for both go routines to return |
|
if err := <-errch; err != nil { |
|
t.Fatal(err) |
|
} |
|
if err := <-errch; err != nil { |
|
t.Fatal(err) |
|
} |
|
} |
|
|
|
func TestCatalogNodes_DistanceSort(t *testing.T) { |
|
t.Parallel() |
|
a := NewTestAgent(t, t.Name(), "") |
|
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{} |
|
if err := a.RPC("Catalog.Register", args, &out); err != nil { |
|
t.Fatalf("err: %v", err) |
|
} |
|
|
|
args = &structs.RegisterRequest{ |
|
Datacenter: "dc1", |
|
Node: "bar", |
|
Address: "127.0.0.2", |
|
} |
|
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/nodes?dc=dc1&near=foo", nil) |
|
resp := httptest.NewRecorder() |
|
obj, err := a.srv.CatalogNodes(resp, req) |
|
if err != nil { |
|
t.Fatalf("err: %v", err) |
|
} |
|
|
|
assertIndex(t, resp) |
|
nodes := obj.(structs.Nodes) |
|
if len(nodes) != 3 { |
|
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) |
|
} |
|
if nodes[2].Node != a.Config.NodeName { |
|
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) |
|
} |
|
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) |
|
if err != nil { |
|
t.Fatalf("err: %v", err) |
|
} |
|
|
|
assertIndex(t, resp) |
|
nodes = obj.(structs.Nodes) |
|
if len(nodes) != 3 { |
|
t.Fatalf("bad: %v", obj) |
|
} |
|
if nodes[0].Node != "foo" { |
|
t.Fatalf("bad: %v", nodes) |
|
} |
|
if nodes[1].Node != "bar" { |
|
t.Fatalf("bad: %v", nodes) |
|
} |
|
if nodes[2].Node != a.Config.NodeName { |
|
t.Fatalf("bad: %v", nodes) |
|
} |
|
} |
|
|
|
func TestCatalogServices(t *testing.T) { |
|
t.Parallel() |
|
a := NewTestAgent(t, t.Name(), "") |
|
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) { |
|
t.Parallel() |
|
a := NewTestAgent(t, t.Name(), "") |
|
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 TestCatalogServiceNodes(t *testing.T) { |
|
t.Parallel() |
|
a := NewTestAgent(t, t.Name(), "") |
|
defer a.Shutdown() |
|
|
|
assert := assert.New(t) |
|
require := require.New(t) |
|
|
|
// 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(err) |
|
nodes := obj.(structs.ServiceNodes) |
|
assert.Len(nodes, 1) |
|
|
|
// Should be a cache miss |
|
assert.Equal("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(err) |
|
nodes := obj.(structs.ServiceNodes) |
|
assert.Len(nodes, 1) |
|
|
|
// Should be a cache HIT now! |
|
assert.Equal("HIT", resp.Header().Get("X-Cache")) |
|
assert.Equal("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(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) { |
|
t.Parallel() |
|
a := NewTestAgent(t, t.Name(), "") |
|
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) { |
|
t.Parallel() |
|
a := NewTestAgent(t, t.Name(), "") |
|
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) { |
|
t.Parallel() |
|
a1 := NewTestAgent(t, t.Name(), ` |
|
datacenter = "dc1" |
|
translate_wan_addrs = true |
|
acl_datacenter = "" |
|
`) |
|
defer a1.Shutdown() |
|
|
|
a2 := NewTestAgent(t, t.Name(), ` |
|
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) |
|
if _, err := a2.srv.agent.JoinWAN([]string{addr}); err != nil { |
|
t.Fatalf("err: %v", err) |
|
} |
|
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: "foo", |
|
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 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) |
|
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.ServiceNodes) |
|
if len(nodes1) != 1 { |
|
t.Fatalf("bad: %v", obj1) |
|
} |
|
node1 := nodes1[0] |
|
if node1.Address != "127.0.0.2" { |
|
t.Fatalf("bad: %v", node1) |
|
} |
|
|
|
// Query DC2 from DC2. |
|
resp2 := httptest.NewRecorder() |
|
obj2, err2 := a2.srv.CatalogServiceNodes(resp2, req) |
|
if err2 != nil { |
|
t.Fatalf("err: %v", err2) |
|
} |
|
assertIndex(t, resp2) |
|
|
|
// Expect that DC2 gives us a local address (since the node is in DC2). |
|
nodes2 := obj2.(structs.ServiceNodes) |
|
if len(nodes2) != 1 { |
|
t.Fatalf("bad: %v", obj2) |
|
} |
|
node2 := nodes2[0] |
|
if node2.Address != "127.0.0.1" { |
|
t.Fatalf("bad: %v", node2) |
|
} |
|
} |
|
|
|
func TestCatalogServiceNodes_DistanceSort(t *testing.T) { |
|
t.Parallel() |
|
a := NewTestAgent(t, t.Name(), "") |
|
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) |
|
} |
|
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/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 != "foo" { |
|
t.Fatalf("bad: %v", nodes) |
|
} |
|
if nodes[1].Node != "bar" { |
|
t.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) { |
|
t.Parallel() |
|
|
|
assert := assert.New(t) |
|
a := NewTestAgent(t, t.Name(), "") |
|
defer a.Shutdown() |
|
testrpc.WaitForLeader(t, a.RPC, "dc1") |
|
|
|
// Register |
|
args := structs.TestRegisterRequestProxy(t) |
|
var out struct{} |
|
assert.Nil(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(err) |
|
assertIndex(t, resp) |
|
|
|
nodes := obj.(structs.ServiceNodes) |
|
assert.Len(nodes, 1) |
|
assert.Equal(structs.ServiceKindConnectProxy, nodes[0].ServiceKind) |
|
assert.Equal(args.Service.Proxy, nodes[0].ServiceProxy) |
|
// DEPRECATED (ProxyDestination) - remove this when removing ProxyDestination |
|
assert.Equal(args.Service.Proxy.DestinationServiceName, nodes[0].ServiceProxyDestination) |
|
} |
|
|
|
// Test that the Connect-compatible endpoints can be queried for a |
|
// service via /v1/catalog/connect/:service. |
|
func TestCatalogConnectServiceNodes_good(t *testing.T) { |
|
t.Parallel() |
|
|
|
assert := assert.New(t) |
|
a := NewTestAgent(t, t.Name(), "") |
|
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(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(err) |
|
assertIndex(t, resp) |
|
|
|
nodes := obj.(structs.ServiceNodes) |
|
assert.Len(nodes, 1) |
|
assert.Equal(structs.ServiceKindConnectProxy, nodes[0].ServiceKind) |
|
assert.Equal(args.Service.Address, nodes[0].ServiceAddress) |
|
assert.Equal(args.Service.Proxy, nodes[0].ServiceProxy) |
|
} |
|
|
|
func TestCatalogConnectServiceNodes_Filter(t *testing.T) { |
|
t.Parallel() |
|
|
|
a := NewTestAgent(t, t.Name(), "") |
|
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) { |
|
t.Parallel() |
|
a := NewTestAgent(t, t.Name(), "") |
|
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 TestCatalogNodeServices_Filter(t *testing.T) { |
|
t.Parallel() |
|
a := NewTestAgent(t, t.Name(), "") |
|
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) { |
|
t.Parallel() |
|
|
|
assert := assert.New(t) |
|
a := NewTestAgent(t, t.Name(), "") |
|
defer a.Shutdown() |
|
testrpc.WaitForTestAgent(t, a.RPC, "dc1") |
|
|
|
// Register |
|
args := structs.TestRegisterRequestProxy(t) |
|
var out struct{} |
|
assert.Nil(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(err) |
|
assertIndex(t, resp) |
|
|
|
ns := obj.(*structs.NodeServices) |
|
assert.Len(ns.Services, 1) |
|
v := ns.Services[args.Service.Service] |
|
assert.Equal(structs.ServiceKindConnectProxy, v.Kind) |
|
} |
|
|
|
func TestCatalogNodeServices_WanTranslation(t *testing.T) { |
|
t.Parallel() |
|
a1 := NewTestAgent(t, t.Name(), ` |
|
datacenter = "dc1" |
|
translate_wan_addrs = true |
|
acl_datacenter = "" |
|
`) |
|
defer a1.Shutdown() |
|
testrpc.WaitForTestAgent(t, a1.RPC, "dc1") |
|
|
|
a2 := NewTestAgent(t, t.Name(), ` |
|
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.srv.agent.JoinWAN([]string{addr}); err != nil { |
|
t.Fatalf("err: %v", err) |
|
} |
|
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: "foo", |
|
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 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) |
|
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). |
|
services1 := obj1.(*structs.NodeServices) |
|
if len(services1.Services) != 1 { |
|
t.Fatalf("bad: %v", obj1) |
|
} |
|
service1 := services1.Node |
|
if service1.Address != "127.0.0.2" { |
|
t.Fatalf("bad: %v", service1) |
|
} |
|
|
|
// Query DC2 from DC2. |
|
resp2 := httptest.NewRecorder() |
|
obj2, err2 := a2.srv.CatalogNodeServices(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). |
|
services2 := obj2.(*structs.NodeServices) |
|
if len(services2.Services) != 1 { |
|
t.Fatalf("bad: %v", obj2) |
|
} |
|
service2 := services2.Node |
|
if service2.Address != "127.0.0.1" { |
|
t.Fatalf("bad: %v", service2) |
|
} |
|
}
|
|
|