5b801db24b
Peer replication is intended to be between separate Consul installs and effectively should be considered "external". This PR moves the peer stream replication bidirectional RPC endpoint to the external gRPC server and ensures that things continue to function.
428 lines
13 KiB
Go
428 lines
13 KiB
Go
package agent
|
|
|
|
import (
|
|
"bytes"
|
|
"context"
|
|
"encoding/base64"
|
|
"encoding/json"
|
|
"fmt"
|
|
"io"
|
|
"net/http"
|
|
"net/http/httptest"
|
|
"testing"
|
|
"time"
|
|
|
|
"github.com/stretchr/testify/require"
|
|
|
|
"github.com/hashicorp/consul/agent/structs"
|
|
"github.com/hashicorp/consul/api"
|
|
"github.com/hashicorp/consul/proto/pbpeering"
|
|
"github.com/hashicorp/consul/sdk/testutil/retry"
|
|
"github.com/hashicorp/consul/testrpc"
|
|
)
|
|
|
|
var validCA = `
|
|
-----BEGIN CERTIFICATE-----
|
|
MIICmDCCAj6gAwIBAgIBBzAKBggqhkjOPQQDAjAWMRQwEgYDVQQDEwtDb25zdWwg
|
|
Q0EgNzAeFw0xODA1MjExNjMzMjhaFw0yODA1MTgxNjMzMjhaMBYxFDASBgNVBAMT
|
|
C0NvbnN1bCBDQSA3MFkwEwYHKoZIzj0CAQYIKoZIzj0DAQcDQgAER0qlxjnRcMEr
|
|
iSGlH7G7dYU7lzBEmLUSMZkyBbClmyV8+e8WANemjn+PLnCr40If9cmpr7RnC9Qk
|
|
GTaLnLiF16OCAXswggF3MA4GA1UdDwEB/wQEAwIBhjAPBgNVHRMBAf8EBTADAQH/
|
|
MGgGA1UdDgRhBF8xZjo5MTpjYTo0MTo4ZjphYzo2NzpiZjo1OTpjMjpmYTo0ZTo3
|
|
NTo1YzpkODpmMDo1NTpkZTpiZTo3NTpiODozMzozMTpkNToyNDpiMDowNDpiMzpl
|
|
ODo5Nzo1Yjo3ZTBqBgNVHSMEYzBhgF8xZjo5MTpjYTo0MTo4ZjphYzo2NzpiZjo1
|
|
OTpjMjpmYTo0ZTo3NTo1YzpkODpmMDo1NTpkZTpiZTo3NTpiODozMzozMTpkNToy
|
|
NDpiMDowNDpiMzplODo5Nzo1Yjo3ZTA/BgNVHREEODA2hjRzcGlmZmU6Ly8xMjRk
|
|
ZjVhMC05ODIwLTc2YzMtOWFhOS02ZjYyMTY0YmExYzIuY29uc3VsMD0GA1UdHgEB
|
|
/wQzMDGgLzAtgisxMjRkZjVhMC05ODIwLTc2YzMtOWFhOS02ZjYyMTY0YmExYzIu
|
|
Y29uc3VsMAoGCCqGSM49BAMCA0gAMEUCIQDzkkI7R+0U12a+zq2EQhP/n2mHmta+
|
|
fs2hBxWIELGwTAIgLdO7RRw+z9nnxCIA6kNl//mIQb+PGItespiHZKAz74Q=
|
|
-----END CERTIFICATE-----
|
|
`
|
|
|
|
func TestHTTP_Peering_GenerateToken(t *testing.T) {
|
|
if testing.Short() {
|
|
t.Skip("too slow for testing.Short")
|
|
}
|
|
|
|
t.Parallel()
|
|
a := NewTestAgent(t, "")
|
|
|
|
testrpc.WaitForTestAgent(t, a.RPC, "dc1")
|
|
|
|
t.Run("No Body", func(t *testing.T) {
|
|
req, err := http.NewRequest("POST", "/v1/peering/token", nil)
|
|
require.NoError(t, err)
|
|
resp := httptest.NewRecorder()
|
|
a.srv.h.ServeHTTP(resp, req)
|
|
require.Equal(t, http.StatusBadRequest, resp.Code)
|
|
body, _ := io.ReadAll(resp.Body)
|
|
require.Contains(t, string(body), "The peering arguments must be provided in the body")
|
|
})
|
|
|
|
t.Run("Body Invalid", func(t *testing.T) {
|
|
req, err := http.NewRequest("POST", "/v1/peering/token", bytes.NewReader([]byte("abc")))
|
|
require.NoError(t, err)
|
|
resp := httptest.NewRecorder()
|
|
a.srv.h.ServeHTTP(resp, req)
|
|
require.Equal(t, http.StatusBadRequest, resp.Code)
|
|
body, _ := io.ReadAll(resp.Body)
|
|
require.Contains(t, string(body), "Body decoding failed:")
|
|
})
|
|
|
|
t.Run("No Name", func(t *testing.T) {
|
|
req, err := http.NewRequest("POST", "/v1/peering/token",
|
|
bytes.NewReader([]byte(`{}`)))
|
|
require.NoError(t, err)
|
|
resp := httptest.NewRecorder()
|
|
a.srv.h.ServeHTTP(resp, req)
|
|
require.Equal(t, http.StatusBadRequest, resp.Code)
|
|
body, _ := io.ReadAll(resp.Body)
|
|
require.Contains(t, string(body), "PeerName is required")
|
|
})
|
|
|
|
// TODO(peering): add more failure cases
|
|
|
|
t.Run("Success", func(t *testing.T) {
|
|
body := &pbpeering.GenerateTokenRequest{
|
|
PeerName: "peering-a",
|
|
}
|
|
|
|
bodyBytes, err := json.Marshal(body)
|
|
require.NoError(t, err)
|
|
|
|
req, err := http.NewRequest("POST", "/v1/peering/token", bytes.NewReader(bodyBytes))
|
|
require.NoError(t, err)
|
|
resp := httptest.NewRecorder()
|
|
a.srv.h.ServeHTTP(resp, req)
|
|
require.Equal(t, http.StatusOK, resp.Code, "expected 200, got %d: %v", resp.Code, resp.Body.String())
|
|
|
|
var r pbpeering.GenerateTokenResponse
|
|
require.NoError(t, json.NewDecoder(resp.Body).Decode(&r))
|
|
|
|
tokenJSON, err := base64.StdEncoding.DecodeString(r.PeeringToken)
|
|
require.NoError(t, err)
|
|
|
|
var token structs.PeeringToken
|
|
require.NoError(t, json.Unmarshal(tokenJSON, &token))
|
|
|
|
require.Nil(t, token.CA)
|
|
require.Equal(t, []string{fmt.Sprintf("127.0.0.1:%d", a.config.GRPCPort)}, token.ServerAddresses)
|
|
require.Equal(t, "server.dc1.consul", token.ServerName)
|
|
|
|
// The PeerID in the token is randomly generated so we don't assert on its value.
|
|
require.NotEmpty(t, token.PeerID)
|
|
})
|
|
}
|
|
|
|
func TestHTTP_Peering_Establish(t *testing.T) {
|
|
if testing.Short() {
|
|
t.Skip("too slow for testing.Short")
|
|
}
|
|
|
|
t.Parallel()
|
|
a := NewTestAgent(t, "")
|
|
|
|
testrpc.WaitForTestAgent(t, a.RPC, "dc1")
|
|
|
|
t.Run("No Body", func(t *testing.T) {
|
|
req, err := http.NewRequest("POST", "/v1/peering/establish", nil)
|
|
require.NoError(t, err)
|
|
resp := httptest.NewRecorder()
|
|
a.srv.h.ServeHTTP(resp, req)
|
|
require.Equal(t, http.StatusBadRequest, resp.Code)
|
|
body, _ := io.ReadAll(resp.Body)
|
|
require.Contains(t, string(body), "The peering arguments must be provided in the body")
|
|
})
|
|
|
|
t.Run("Body Invalid", func(t *testing.T) {
|
|
req, err := http.NewRequest("POST", "/v1/peering/establish", bytes.NewReader([]byte("abc")))
|
|
require.NoError(t, err)
|
|
resp := httptest.NewRecorder()
|
|
a.srv.h.ServeHTTP(resp, req)
|
|
require.Equal(t, http.StatusBadRequest, resp.Code)
|
|
body, _ := io.ReadAll(resp.Body)
|
|
require.Contains(t, string(body), "Body decoding failed:")
|
|
})
|
|
|
|
t.Run("No Name", func(t *testing.T) {
|
|
req, err := http.NewRequest("POST", "/v1/peering/establish",
|
|
bytes.NewReader([]byte(`{}`)))
|
|
require.NoError(t, err)
|
|
resp := httptest.NewRecorder()
|
|
a.srv.h.ServeHTTP(resp, req)
|
|
require.Equal(t, http.StatusBadRequest, resp.Code)
|
|
body, _ := io.ReadAll(resp.Body)
|
|
require.Contains(t, string(body), "PeerName is required")
|
|
})
|
|
|
|
t.Run("No Token", func(t *testing.T) {
|
|
req, err := http.NewRequest("POST", "/v1/peering/establish",
|
|
bytes.NewReader([]byte(`{"PeerName": "peer1-usw1"}`)))
|
|
require.NoError(t, err)
|
|
resp := httptest.NewRecorder()
|
|
a.srv.h.ServeHTTP(resp, req)
|
|
require.Equal(t, http.StatusBadRequest, resp.Code)
|
|
body, _ := io.ReadAll(resp.Body)
|
|
require.Contains(t, string(body), "PeeringToken is required")
|
|
})
|
|
|
|
// TODO(peering): add more failure cases
|
|
|
|
t.Run("Success", func(t *testing.T) {
|
|
token := structs.PeeringToken{
|
|
CA: []string{validCA},
|
|
ServerName: "server.dc1.consul",
|
|
ServerAddresses: []string{fmt.Sprintf("1.2.3.4:%d", 443)},
|
|
PeerID: "a0affd3e-f1c8-4bb9-9168-90fd902c441d",
|
|
}
|
|
tokenJSON, _ := json.Marshal(&token)
|
|
tokenB64 := base64.StdEncoding.EncodeToString(tokenJSON)
|
|
body := &pbpeering.EstablishRequest{
|
|
PeerName: "peering-a",
|
|
PeeringToken: tokenB64,
|
|
Meta: map[string]string{"foo": "bar"},
|
|
}
|
|
|
|
bodyBytes, err := json.Marshal(body)
|
|
require.NoError(t, err)
|
|
|
|
req, err := http.NewRequest("POST", "/v1/peering/establish", bytes.NewReader(bodyBytes))
|
|
require.NoError(t, err)
|
|
resp := httptest.NewRecorder()
|
|
a.srv.h.ServeHTTP(resp, req)
|
|
require.Equal(t, http.StatusOK, resp.Code, "expected 200, got %d: %v", resp.Code, resp.Body.String())
|
|
|
|
// success response does not currently return a value so {} is correct
|
|
require.Equal(t, "{}", resp.Body.String())
|
|
})
|
|
}
|
|
|
|
func TestHTTP_Peering_MethodNotAllowed(t *testing.T) {
|
|
if testing.Short() {
|
|
t.Skip("too slow for testing.Short")
|
|
}
|
|
|
|
t.Parallel()
|
|
a := NewTestAgent(t, "")
|
|
|
|
testrpc.WaitForTestAgent(t, a.RPC, "dc1")
|
|
|
|
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
|
|
defer cancel()
|
|
|
|
// Insert peerings directly to state store.
|
|
// Note that the state store holds reference to the underlying
|
|
// variables; do not modify them after writing.
|
|
foo := &pbpeering.PeeringWriteRequest{
|
|
Peering: &pbpeering.Peering{
|
|
Name: "foo",
|
|
State: pbpeering.PeeringState_ESTABLISHING,
|
|
PeerCAPems: nil,
|
|
PeerServerName: "fooservername",
|
|
PeerServerAddresses: []string{"addr1"},
|
|
},
|
|
}
|
|
_, err := a.rpcClientPeering.PeeringWrite(ctx, foo)
|
|
require.NoError(t, err)
|
|
|
|
req, err := http.NewRequest("PUT", "/v1/peering/foo", nil)
|
|
require.NoError(t, err)
|
|
resp := httptest.NewRecorder()
|
|
a.srv.h.ServeHTTP(resp, req)
|
|
require.Equal(t, http.StatusMethodNotAllowed, resp.Code)
|
|
}
|
|
|
|
func TestHTTP_Peering_Read(t *testing.T) {
|
|
if testing.Short() {
|
|
t.Skip("too slow for testing.Short")
|
|
}
|
|
|
|
t.Parallel()
|
|
a := NewTestAgent(t, "")
|
|
|
|
testrpc.WaitForTestAgent(t, a.RPC, "dc1")
|
|
|
|
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
|
|
defer cancel()
|
|
|
|
// Insert peerings directly to state store.
|
|
// Note that the state store holds reference to the underlying
|
|
// variables; do not modify them after writing.
|
|
foo := &pbpeering.PeeringWriteRequest{
|
|
Peering: &pbpeering.Peering{
|
|
Name: "foo",
|
|
State: pbpeering.PeeringState_ESTABLISHING,
|
|
PeerCAPems: nil,
|
|
PeerServerName: "fooservername",
|
|
PeerServerAddresses: []string{"addr1"},
|
|
Meta: map[string]string{"foo": "bar"},
|
|
},
|
|
}
|
|
_, err := a.rpcClientPeering.PeeringWrite(ctx, foo)
|
|
require.NoError(t, err)
|
|
bar := &pbpeering.PeeringWriteRequest{
|
|
Peering: &pbpeering.Peering{
|
|
Name: "bar",
|
|
State: pbpeering.PeeringState_ACTIVE,
|
|
PeerCAPems: nil,
|
|
PeerServerName: "barservername",
|
|
PeerServerAddresses: []string{"addr1"},
|
|
},
|
|
}
|
|
_, err = a.rpcClientPeering.PeeringWrite(ctx, bar)
|
|
require.NoError(t, err)
|
|
|
|
t.Run("return foo", func(t *testing.T) {
|
|
req, err := http.NewRequest("GET", "/v1/peering/foo", nil)
|
|
require.NoError(t, err)
|
|
resp := httptest.NewRecorder()
|
|
a.srv.h.ServeHTTP(resp, req)
|
|
require.Equal(t, http.StatusOK, resp.Code)
|
|
|
|
var apiResp api.Peering
|
|
require.NoError(t, json.NewDecoder(resp.Body).Decode(&apiResp))
|
|
|
|
require.Equal(t, foo.Peering.Name, apiResp.Name)
|
|
require.Equal(t, foo.Peering.Meta, apiResp.Meta)
|
|
|
|
require.Equal(t, uint64(0), apiResp.ImportedServiceCount)
|
|
require.Equal(t, uint64(0), apiResp.ExportedServiceCount)
|
|
|
|
})
|
|
|
|
t.Run("not found", func(t *testing.T) {
|
|
req, err := http.NewRequest("GET", "/v1/peering/baz", nil)
|
|
require.NoError(t, err)
|
|
resp := httptest.NewRecorder()
|
|
a.srv.h.ServeHTTP(resp, req)
|
|
require.Equal(t, http.StatusNotFound, resp.Code)
|
|
require.Equal(t, "Peering not found for \"baz\"", resp.Body.String())
|
|
})
|
|
}
|
|
|
|
func TestHTTP_Peering_Delete(t *testing.T) {
|
|
if testing.Short() {
|
|
t.Skip("too slow for testing.Short")
|
|
}
|
|
|
|
t.Parallel()
|
|
a := NewTestAgent(t, "")
|
|
|
|
testrpc.WaitForTestAgent(t, a.RPC, "dc1")
|
|
|
|
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
|
|
defer cancel()
|
|
|
|
foo := &pbpeering.PeeringWriteRequest{
|
|
Peering: &pbpeering.Peering{
|
|
Name: "foo",
|
|
State: pbpeering.PeeringState_ESTABLISHING,
|
|
PeerCAPems: nil,
|
|
PeerServerName: "fooservername",
|
|
PeerServerAddresses: []string{"addr1"},
|
|
},
|
|
}
|
|
_, err := a.rpcClientPeering.PeeringWrite(ctx, foo)
|
|
require.NoError(t, err)
|
|
|
|
t.Run("read existing token before attempting delete", func(t *testing.T) {
|
|
req, err := http.NewRequest("GET", "/v1/peering/foo", nil)
|
|
require.NoError(t, err)
|
|
resp := httptest.NewRecorder()
|
|
a.srv.h.ServeHTTP(resp, req)
|
|
require.Equal(t, http.StatusOK, resp.Code)
|
|
|
|
var apiResp api.Peering
|
|
require.NoError(t, json.NewDecoder(resp.Body).Decode(&apiResp))
|
|
|
|
require.Equal(t, foo.Peering.Name, apiResp.Name)
|
|
})
|
|
|
|
t.Run("delete the existing token we just read", func(t *testing.T) {
|
|
req, err := http.NewRequest("DELETE", "/v1/peering/foo", nil)
|
|
require.NoError(t, err)
|
|
resp := httptest.NewRecorder()
|
|
a.srv.h.ServeHTTP(resp, req)
|
|
require.Equal(t, http.StatusOK, resp.Code)
|
|
require.Equal(t, "", resp.Body.String())
|
|
})
|
|
|
|
t.Run("now the token is deleted and reads should yield a 404", func(t *testing.T) {
|
|
retry.Run(t, func(r *retry.R) {
|
|
req, err := http.NewRequest("GET", "/v1/peering/foo", nil)
|
|
require.NoError(r, err)
|
|
resp := httptest.NewRecorder()
|
|
a.srv.h.ServeHTTP(resp, req)
|
|
require.Equal(r, http.StatusNotFound, resp.Code)
|
|
})
|
|
})
|
|
|
|
t.Run("delete a token that does not exist", func(t *testing.T) {
|
|
req, err := http.NewRequest("DELETE", "/v1/peering/baz", nil)
|
|
require.NoError(t, err)
|
|
resp := httptest.NewRecorder()
|
|
a.srv.h.ServeHTTP(resp, req)
|
|
|
|
require.Equal(t, http.StatusOK, resp.Code)
|
|
})
|
|
}
|
|
|
|
func TestHTTP_Peering_List(t *testing.T) {
|
|
if testing.Short() {
|
|
t.Skip("too slow for testing.Short")
|
|
}
|
|
|
|
t.Parallel()
|
|
a := NewTestAgent(t, "")
|
|
|
|
testrpc.WaitForTestAgent(t, a.RPC, "dc1")
|
|
|
|
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
|
|
defer cancel()
|
|
|
|
// Insert peerings directly to state store.
|
|
// Note that the state store holds reference to the underlying
|
|
// variables; do not modify them after writing.
|
|
foo := &pbpeering.PeeringWriteRequest{
|
|
Peering: &pbpeering.Peering{
|
|
Name: "foo",
|
|
State: pbpeering.PeeringState_ESTABLISHING,
|
|
PeerCAPems: nil,
|
|
PeerServerName: "fooservername",
|
|
PeerServerAddresses: []string{"addr1"},
|
|
},
|
|
}
|
|
_, err := a.rpcClientPeering.PeeringWrite(ctx, foo)
|
|
require.NoError(t, err)
|
|
bar := &pbpeering.PeeringWriteRequest{
|
|
Peering: &pbpeering.Peering{
|
|
Name: "bar",
|
|
State: pbpeering.PeeringState_ACTIVE,
|
|
PeerCAPems: nil,
|
|
PeerServerName: "barservername",
|
|
PeerServerAddresses: []string{"addr1"},
|
|
},
|
|
}
|
|
_, err = a.rpcClientPeering.PeeringWrite(ctx, bar)
|
|
require.NoError(t, err)
|
|
|
|
t.Run("return all", func(t *testing.T) {
|
|
req, err := http.NewRequest("GET", "/v1/peerings", nil)
|
|
require.NoError(t, err)
|
|
resp := httptest.NewRecorder()
|
|
a.srv.h.ServeHTTP(resp, req)
|
|
require.Equal(t, http.StatusOK, resp.Code)
|
|
|
|
var apiResp []*api.Peering
|
|
require.NoError(t, json.NewDecoder(resp.Body).Decode(&apiResp))
|
|
|
|
require.Len(t, apiResp, 2)
|
|
|
|
for _, p := range apiResp {
|
|
require.Equal(t, uint64(0), p.ImportedServiceCount)
|
|
require.Equal(t, uint64(0), p.ExportedServiceCount)
|
|
}
|
|
})
|
|
}
|