github
/
consul
mirror of https://github.com/hashicorp/consul
1
0
Fork 0
Code Issues Projects Releases Wiki Activity
Consul is a distributed, highly available, and data center aware solution to connect and configure applications across dynamic, distributed infrastructure.
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.
20900 Commits
1702 Branches
457 Tags
697 MiB
 
 
 
 
 
 
Tree: 053367a3b2
Branches Tags
${ item.name }
Create tag ${ searchTerm }
Create branch ${ searchTerm }
from '053367a3b2'
${ noResults }
consul/agent/txn_endpoint_test.go

854 lines
20 KiB
Raw Blame History

// Copyright (c) HashiCorp, Inc.
// SPDX-License-Identifier: BUSL-1.1
package agent
import (
"bytes"
"encoding/base64"
"fmt"
"net/http"
"net/http/httptest"
"strings"
"testing"
"time"
"github.com/hashicorp/raft"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
"github.com/hashicorp/consul/agent/structs"
"github.com/hashicorp/consul/api"
"github.com/hashicorp/consul/testrpc"
)
func TestTxnEndpoint_Bad_JSON(t *testing.T) {
if testing.Short() {
t.Skip("too slow for testing.Short")
}
t.Parallel()
a := NewTestAgent(t, "")
defer a.Shutdown()
buf := bytes.NewBuffer([]byte("{"))
req, _ := http.NewRequest("PUT", "/v1/txn", buf)
resp := httptest.NewRecorder()
_, err := a.srv.Txn(resp, req)
require.True(t, isHTTPBadRequest(err), fmt.Sprintf("Expected bad request HTTP error but got %v", err))
if !strings.Contains(err.Error(), "Failed to parse") {
t.Fatalf("expected conflicting args error")
}
}
func TestTxnEndpoint_Bad_Size_Item(t *testing.T) {
if testing.Short() {
t.Skip("too slow for testing.Short")
}
t.Parallel()
testIt := func(t *testing.T, agent *TestAgent, wantPass bool) {
value := strings.Repeat("X", 3*raft.SuggestedMaxDataSize)
value = base64.StdEncoding.EncodeToString([]byte(value))
buf := bytes.NewBuffer([]byte(fmt.Sprintf(`
[
{
"KV": {
"Verb": "set",
"Key": "key",
"Value": %q
}
}
]
`, value)))
req, _ := http.NewRequest("PUT", "/v1/txn", buf)
resp := httptest.NewRecorder()
_, err := agent.srv.Txn(resp, req)
if wantPass {
if err != nil {
t.Fatalf("err: %v", err)
}
} else {
if httpErr, ok := err.(HTTPError); ok {
if httpErr.StatusCode != 413 {
t.Fatalf("expected 413 but got %d", httpErr.StatusCode)
}
} else {
t.Fatalf("excected HTTP error but got %v", err)
}
}
}
t.Run("exceeds default limits", func(t *testing.T) {
a := NewTestAgent(t, "")
testIt(t, a, false)
a.Shutdown()
})
t.Run("exceeds configured max txn len", func(t *testing.T) {
a := NewTestAgent(t, "limits = { txn_max_req_len = 700000 }")
testIt(t, a, false)
a.Shutdown()
})
t.Run("exceeds default max kv value size", func(t *testing.T) {
a := NewTestAgent(t, "limits = { txn_max_req_len = 123456789 }")
testIt(t, a, false)
a.Shutdown()
})
t.Run("allowed", func(t *testing.T) {
a := NewTestAgent(t, `
limits = {
txn_max_req_len = 123456789
kv_max_value_size = 123456789
}`)
testIt(t, a, true)
a.Shutdown()
})
}
func TestTxnEndpoint_Bad_Size_Net(t *testing.T) {
if testing.Short() {
t.Skip("too slow for testing.Short")
}
t.Parallel()
testIt := func(agent *TestAgent, wantPass bool) {
value := strings.Repeat("X", 3*raft.SuggestedMaxDataSize)
value = base64.StdEncoding.EncodeToString([]byte(value))
buf := bytes.NewBuffer([]byte(fmt.Sprintf(`
[
{
"KV": {
"Verb": "set",
"Key": "key1",
"Value": %q
}
},
{
"KV": {
"Verb": "set",
"Key": "key1",
"Value": %q
}
},
{
"KV": {
"Verb": "set",
"Key": "key1",
"Value": %q
}
}
]
`, value, value, value)))
req, _ := http.NewRequest("PUT", "/v1/txn", buf)
resp := httptest.NewRecorder()
_, err := agent.srv.Txn(resp, req)
if wantPass {
if err != nil {
t.Fatalf("err: %v", err)
}
} else {
if httpErr, ok := err.(HTTPError); ok {
if httpErr.StatusCode != 413 {
t.Fatalf("expected 413 but got %d", httpErr.StatusCode)
}
} else {
t.Fatalf("excected HTTP error but got %v", err)
}
}
}
t.Run("exceeds default limits", func(t *testing.T) {
a := NewTestAgent(t, "")
testIt(a, false)
a.Shutdown()
})
t.Run("exceeds configured max txn len", func(t *testing.T) {
a := NewTestAgent(t, "limits = { txn_max_req_len = 700000 }")
testIt(a, false)
a.Shutdown()
})
t.Run("exceeds default max kv value size", func(t *testing.T) {
a := NewTestAgent(t, "limits = { txn_max_req_len = 123456789 }")
testIt(a, false)
a.Shutdown()
})
t.Run("allowed", func(t *testing.T) {
a := NewTestAgent(t, `
limits = {
txn_max_req_len = 123456789
kv_max_value_size = 123456789
}`)
testIt(a, true)
a.Shutdown()
})
t.Run("allowed kv max backward compatible", func(t *testing.T) {
a := NewTestAgent(t, "limits = { kv_max_value_size = 123456789 }")
testIt(a, true)
a.Shutdown()
})
}
func TestTxnEndpoint_Bad_Size_Ops(t *testing.T) {
if testing.Short() {
t.Skip("too slow for testing.Short")
}
t.Parallel()
a := NewTestAgent(t, "")
defer a.Shutdown()
buf := bytes.NewBuffer([]byte(fmt.Sprintf(`
[
%s
{
"KV": {
"Verb": "set",
"Key": "key",
"Value": ""
}
}
]
`, strings.Repeat(`{ "KV": { "Verb": "get", "Key": "key" } },`, 2*maxTxnOps))))
req, _ := http.NewRequest("PUT", "/v1/txn", buf)
resp := httptest.NewRecorder()
_, err := a.srv.Txn(resp, req)
if httpErr, ok := err.(HTTPError); ok {
if httpErr.StatusCode != 413 {
t.Fatalf("expected 413 but got %d", httpErr.StatusCode)
}
} else {
t.Fatalf("expected HTTP error but got %v", err)
}
}
func TestTxnEndpoint_KV_Actions(t *testing.T) {
if testing.Short() {
t.Skip("too slow for testing.Short")
}
t.Parallel()
t.Run("", func(t *testing.T) {
a := NewTestAgent(t, "")
defer a.Shutdown()
testrpc.WaitForTestAgent(t, a.RPC, "dc1")
// Make sure all incoming fields get converted properly to the internal
// RPC format.
var index uint64
id := makeTestSession(t, a.srv)
{
buf := bytes.NewBuffer([]byte(fmt.Sprintf(`
[
{
"KV": {
"Verb": "lock",
"Key": "key",
"Value": "aGVsbG8gd29ybGQ=",
"Flags": 23,
"Session": %q
}
},
{
"KV": {
"Verb": "get",
"Key": "key"
}
}
]
`, id)))
req, _ := http.NewRequest("PUT", "/v1/txn", buf)
resp := httptest.NewRecorder()
obj, err := a.srv.Txn(resp, req)
if err != nil {
t.Fatalf("err: %v", err)
}
if resp.Code != 200 {
t.Fatalf("expected 200, got %d", resp.Code)
}
txnResp, ok := obj.(structs.TxnResponse)
if !ok {
t.Fatalf("bad type: %T", obj)
}
if len(txnResp.Results) != 2 {
t.Fatalf("bad: %v", txnResp)
}
index = txnResp.Results[0].KV.ModifyIndex
entMeta := txnResp.Results[0].KV.EnterpriseMeta
expected := structs.TxnResponse{
Results: structs.TxnResults{
&structs.TxnResult{
KV: &structs.DirEntry{
Key: "key",
Value: nil,
Flags: 23,
Session: id,
LockIndex: 1,
RaftIndex: structs.RaftIndex{
CreateIndex: index,
ModifyIndex: index,
},
EnterpriseMeta: entMeta,
},
},
&structs.TxnResult{
KV: &structs.DirEntry{
Key: "key",
Value: []byte("hello world"),
Flags: 23,
Session: id,
LockIndex: 1,
RaftIndex: structs.RaftIndex{
CreateIndex: index,
ModifyIndex: index,
},
EnterpriseMeta: entMeta,
},
},
},
}
assert.Equal(t, expected, txnResp)
}
// Do a read-only transaction that should get routed to the
// fast-path endpoint.
{
buf := bytes.NewBuffer([]byte(`
[
{
"KV": {
"Verb": "get",
"Key": "key"
}
},
{
"KV": {
"Verb": "get-tree",
"Key": "key"
}
}
]
`))
req, _ := http.NewRequest("PUT", "/v1/txn", buf)
resp := httptest.NewRecorder()
obj, err := a.srv.Txn(resp, req)
if err != nil {
t.Fatalf("err: %v", err)
}
if resp.Code != 200 {
t.Fatalf("expected 200, got %d", resp.Code)
}
header := resp.Header().Get("X-Consul-KnownLeader")
if header != "true" {
t.Fatalf("bad: %v", header)
}
header = resp.Header().Get("X-Consul-LastContact")
if header != "0" {
t.Fatalf("bad: %v", header)
}
txnResp, ok := obj.(structs.TxnReadResponse)
if !ok {
t.Fatalf("bad type: %T", obj)
}
entMeta := txnResp.Results[0].KV.EnterpriseMeta
expected := structs.TxnReadResponse{
TxnResponse: structs.TxnResponse{
Results: structs.TxnResults{
&structs.TxnResult{
KV: &structs.DirEntry{
Key: "key",
Value: []byte("hello world"),
Flags: 23,
Session: id,
LockIndex: 1,
RaftIndex: structs.RaftIndex{
CreateIndex: index,
ModifyIndex: index,
},
EnterpriseMeta: entMeta,
},
},
&structs.TxnResult{
KV: &structs.DirEntry{
Key: "key",
Value: []byte("hello world"),
Flags: 23,
Session: id,
LockIndex: 1,
RaftIndex: structs.RaftIndex{
CreateIndex: index,
ModifyIndex: index,
},
EnterpriseMeta: entMeta,
},
},
},
},
QueryMeta: structs.QueryMeta{
KnownLeader: true,
Index: 1,
},
}
assert.Equal(t, expected, txnResp)
}
// Now that we have an index we can do a CAS to make sure the
// index field gets translated to the RPC format.
{
buf := bytes.NewBuffer([]byte(fmt.Sprintf(`
[
{
"KV": {
"Verb": "cas",
"Key": "key",
"Value": "Z29vZGJ5ZSB3b3JsZA==",
"Index": %d
}
},
{
"KV": {
"Verb": "get",
"Key": "key"
}
}
]
`, index)))
req, _ := http.NewRequest("PUT", "/v1/txn", buf)
resp := httptest.NewRecorder()
obj, err := a.srv.Txn(resp, req)
if err != nil {
t.Fatalf("err: %v", err)
}
if resp.Code != 200 {
t.Fatalf("expected 200, got %d", resp.Code)
}
txnResp, ok := obj.(structs.TxnResponse)
if !ok {
t.Fatalf("bad type: %T", obj)
}
if len(txnResp.Results) != 2 {
t.Fatalf("bad: %v", txnResp)
}
modIndex := txnResp.Results[0].KV.ModifyIndex
entMeta := txnResp.Results[0].KV.EnterpriseMeta
expected := structs.TxnResponse{
Results: structs.TxnResults{
&structs.TxnResult{
KV: &structs.DirEntry{
Key: "key",
Value: nil,
Session: id,
RaftIndex: structs.RaftIndex{
CreateIndex: index,
ModifyIndex: modIndex,
},
EnterpriseMeta: entMeta,
},
},
&structs.TxnResult{
KV: &structs.DirEntry{
Key: "key",
Value: []byte("goodbye world"),
Session: id,
RaftIndex: structs.RaftIndex{
CreateIndex: index,
ModifyIndex: modIndex,
},
EnterpriseMeta: entMeta,
},
},
},
}
assert.Equal(t, expected, txnResp)
}
})
// Verify an error inside a transaction.
t.Run("", func(t *testing.T) {
a := NewTestAgent(t, "")
defer a.Shutdown()
buf := bytes.NewBuffer([]byte(`
[
{
"KV": {
"Verb": "lock",
"Key": "key",
"Value": "aGVsbG8gd29ybGQ=",
"Session": "nope"
}
},
{
"KV": {
"Verb": "get",
"Key": "key"
}
}
]
`))
req, _ := http.NewRequest("PUT", "/v1/txn", buf)
resp := httptest.NewRecorder()
if _, err := a.srv.Txn(resp, req); err != nil {
t.Fatalf("err: %v", err)
}
if resp.Code != 409 {
t.Fatalf("expected 409, got %d", resp.Code)
}
if !bytes.Contains(resp.Body.Bytes(), []byte("invalid session")) {
t.Fatalf("bad: %s", resp.Body.String())
}
})
}
func TestTxnEndpoint_UpdateCheck(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")
// Make sure the fields of a check are handled correctly when both creating and
// updating, and test both sets of duration fields to ensure backwards compatibility.
buf := bytes.NewBuffer([]byte(fmt.Sprintf(`
[
{
"Check": {
"Verb": "set",
"Check": {
"Node": "%s",
"CheckID": "nodecheck",
"Name": "Node http check",
"Status": "critical",
"Notes": "Http based health check",
"Output": "",
"ServiceID": "",
"ServiceName": "",
"Definition": {
"Interval": "6s",
"Timeout": "6s",
"DeregisterCriticalServiceAfter": "6s",
"HTTP": "http://localhost:8000",
"TLSSkipVerify": true
}
}
}
},
{
"Check": {
"Verb": "set",
"Check": {
"Node": "%s",
"CheckID": "nodecheck",
"Name": "Node http check",
"Status": "passing",
"Notes": "Http based health check",
"Output": "success",
"ServiceID": "",
"ServiceName": "",
"Definition": {
"Interval": "10s",
"Timeout": "10s",
"DeregisterCriticalServiceAfter": "15m",
"HTTP": "http://localhost:9000",
"TLSSkipVerify": false
}
}
}
},
{
"Check": {
"Verb": "set",
"Check": {
"Node": "%s",
"CheckID": "nodecheck",
"Name": "Node http check",
"Status": "passing",
"Notes": "Http based health check",
"Output": "success",
"ServiceID": "",
"ServiceName": "",
"ExposedPort": 5678,
"Definition": {
"IntervalDuration": "15s",
"TimeoutDuration": "15s",
"DeregisterCriticalServiceAfterDuration": "30m",
"HTTP": "http://localhost:9000",
"TLSSkipVerify": false
}
}
}
}
]
`, a.config.NodeName, a.config.NodeName, a.config.NodeName)))
req, _ := http.NewRequest("PUT", "/v1/txn", buf)
resp := httptest.NewRecorder()
obj, err := a.srv.Txn(resp, req)
require.NoError(t, err)
require.Equal(t, 200, resp.Code, resp.Body)
txnResp, ok := obj.(structs.TxnResponse)
if !ok {
t.Fatalf("bad type: %T", obj)
}
if len(txnResp.Results) != 3 {
t.Fatalf("bad: %v", txnResp)
}
index := txnResp.Results[0].Check.ModifyIndex
expected := structs.TxnResponse{
Results: structs.TxnResults{
&structs.TxnResult{
Check: &structs.HealthCheck{
Node: a.config.NodeName,
CheckID: "nodecheck",
Name: "Node http check",
Status: api.HealthCritical,
Notes: "Http based health check",
Definition: structs.HealthCheckDefinition{
Interval: 6 * time.Second,
Timeout: 6 * time.Second,
DeregisterCriticalServiceAfter: 6 * time.Second,
HTTP: "http://localhost:8000",
TLSSkipVerify: true,
},
RaftIndex: structs.RaftIndex{
CreateIndex: index,
ModifyIndex: index,
},
EnterpriseMeta: *structs.DefaultEnterpriseMetaInDefaultPartition(),
},
},
&structs.TxnResult{
Check: &structs.HealthCheck{
Node: a.config.NodeName,
CheckID: "nodecheck",
Name: "Node http check",
Status: api.HealthPassing,
Notes: "Http based health check",
Output: "success",
Definition: structs.HealthCheckDefinition{
Interval: 10 * time.Second,
Timeout: 10 * time.Second,
DeregisterCriticalServiceAfter: 15 * time.Minute,
HTTP: "http://localhost:9000",
TLSSkipVerify: false,
},
RaftIndex: structs.RaftIndex{
CreateIndex: index,
ModifyIndex: index,
},
EnterpriseMeta: *structs.DefaultEnterpriseMetaInDefaultPartition(),
},
},
&structs.TxnResult{
Check: &structs.HealthCheck{
Node: a.config.NodeName,
CheckID: "nodecheck",
Name: "Node http check",
Status: api.HealthPassing,
Notes: "Http based health check",
Output: "success",
ExposedPort: 5678,
Definition: structs.HealthCheckDefinition{
Interval: 15 * time.Second,
Timeout: 15 * time.Second,
DeregisterCriticalServiceAfter: 30 * time.Minute,
HTTP: "http://localhost:9000",
TLSSkipVerify: false,
},
RaftIndex: structs.RaftIndex{
CreateIndex: index,
ModifyIndex: index,
},
EnterpriseMeta: *structs.DefaultEnterpriseMetaInDefaultPartition(),
},
},
},
}
assert.Equal(t, expected, txnResp)
}
func TestTxnEndpoint_NodeService(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")
// Make sure the fields of a check are handled correctly when both creating and
// updating, and test both sets of duration fields to ensure backwards compatibility.
buf := bytes.NewBuffer([]byte(fmt.Sprintf(`
[
{
"Service": {
"Verb": "set",
"Node": "%s",
"Service": {
"Service": "test",
"Port": 4444
}
}
},
{
"Service": {
"Verb": "set",
"Node": "%s",
"Service": {
"Service": "test-sidecar-proxy",
"Port": 20000,
"Kind": "connect-proxy",
"Proxy": {
"DestinationServiceName": "test",
"DestinationServiceID": "test",
"LocalServiceAddress": "127.0.0.1",
"LocalServicePort": 4444,
"upstreams": [
{
"DestinationName": "fake-backend",
"LocalBindPort": 25001
}
]
}
}
}
}
]
`, a.config.NodeName, a.config.NodeName)))
req, _ := http.NewRequest("PUT", "/v1/txn", buf)
resp := httptest.NewRecorder()
obj, err := a.srv.Txn(resp, req)
require.NoError(t, err)
require.Equal(t, 200, resp.Code)
txnResp, ok := obj.(structs.TxnResponse)
if !ok {
t.Fatalf("bad type: %T", obj)
}
require.Equal(t, 2, len(txnResp.Results))
index := txnResp.Results[0].Service.ModifyIndex
expected := structs.TxnResponse{
Results: structs.TxnResults{
&structs.TxnResult{
Service: &structs.NodeService{
Service: "test",
ID: "test",
Port: 4444,
Weights: &structs.Weights{
Passing: 1,
Warning: 1,
},
RaftIndex: structs.RaftIndex{
CreateIndex: index,
ModifyIndex: index,
},
EnterpriseMeta: *structs.DefaultEnterpriseMetaInDefaultPartition(),
},
},
&structs.TxnResult{
Service: &structs.NodeService{
Service: "test-sidecar-proxy",
ID: "test-sidecar-proxy",
Port: 20000,
Kind: "connect-proxy",
Weights: &structs.Weights{
Passing: 1,
Warning: 1,
},
Proxy: structs.ConnectProxyConfig{
DestinationServiceName: "test",
DestinationServiceID: "test",
LocalServiceAddress: "127.0.0.1",
LocalServicePort: 4444,
},
TaggedAddresses: map[string]structs.ServiceAddress{
"consul-virtual": {
Address: "240.0.0.1",
Port: 20000,
},
},
RaftIndex: structs.RaftIndex{
CreateIndex: index,
ModifyIndex: index,
},
EnterpriseMeta: *structs.DefaultEnterpriseMetaInDefaultPartition(),
},
},
},
}
assert.Equal(t, expected, txnResp)
}
func TestTxnEndpoint_OperationsSize(t *testing.T) {
if testing.Short() {
t.Skip("too slow for testing.Short")
}
t.Run("too-many-operations", func(t *testing.T) {
var ops []api.TxnOp
agent := NewTestAgent(t, "limits = { txn_max_req_len = 700000 }")
for i := 0; i < 130; i++ {
ops = append(ops, api.TxnOp{
KV: &api.KVTxnOp{
Verb: api.KVSet,
Key: "key",
Value: []byte("test"),
},
})
}
req, _ := http.NewRequest("PUT", "/v1/txn", jsonBody(ops))
resp := httptest.NewRecorder()
raw, err := agent.srv.Txn(resp, req)
require.Error(t, err)
require.Contains(t, err.Error(), "Transaction contains too many operations")
require.Nil(t, raw)
agent.Shutdown()
})
t.Run("allowed", func(t *testing.T) {
var ops []api.TxnOp
agent := NewTestAgent(t, "limits = { txn_max_req_len = 700000 }")
for i := 0; i < 128; i++ {
ops = append(ops, api.TxnOp{
KV: &api.KVTxnOp{
Verb: api.KVSet,
Key: "key",
Value: []byte("test"),
},
})
}
req, _ := http.NewRequest("PUT", "/v1/txn", jsonBody(ops))
resp := httptest.NewRecorder()
raw, err := agent.srv.Txn(resp, req)
require.NoError(t, err)
require.NotNil(t, raw)
agent.Shutdown()
})
}