mirror of https://github.com/hashicorp/consul
Browse Source
Receiving an "acl not found" error from an RPC in the agent cache and the streaming/event components will cause any request loops to cease under the assumption that they will never work again if the token was destroyed. This prevents log spam (#14144, #9738). Unfortunately due to things like: - authz requests going to stale servers that may not have witnessed the token creation yet - authz requests in a secondary datacenter happening before the tokens get replicated to that datacenter - authz requests from a primary TO a secondary datacenter happening before the tokens get replicated to that datacenter The caller will get an "acl not found" *before* the token exists, rather than just after. The machinery added above in the linked PRs will kick in and prevent the request loop from looping around again once the tokens actually exist. For `consul-dataplane` usages, where xDS is served by the Consul servers rather than the clients ultimately this is not a problem because in that scenario the `agent/proxycfg` machinery is on-demand and launched by a new xDS stream needing data for a specific service in the catalog. If the watching goroutines are terminated it ripples down and terminates the xDS stream, which CDP will eventually re-establish and restart everything. For Consul client usages, the `agent/proxycfg` machinery is ahead-of-time launched at service registration time (called "local" in some of the proxycfg machinery) so when the xDS stream comes in the data is already ready to go. If the watching goroutines terminate it should terminate the xDS stream, but there's no mechanism to re-spawn the watching goroutines. If the xDS stream reconnects it will see no `ConfigSnapshot` and will not get one again until the client agent is restarted, or the service is re-registered with something changed in it. This PR fixes a few things in the machinery: - there was an inadvertent deadlock in fetching snapshot from the proxycfg machinery by xDS, such that when the watching goroutine terminated the snapshots would never be fetched. This caused some of the xDS machinery to get indefinitely paused and not finish the teardown properly. - Every 30s we now attempt to re-insert all locally registered services into the proxycfg machinery. - When services are re-inserted into the proxycfg machinery we special case "dead" ones such that we unilaterally replace them rather that doing that conditionally.pull/16512/head
R.B. Boyer
2 years ago
committed by
GitHub
11 changed files with 208 additions and 14 deletions
@ -0,0 +1,3 @@
|
||||
```release-note:bug |
||||
proxycfg: ensure that an irrecoverable error in proxycfg closes the xds session and triggers a replacement proxycfg watcher |
||||
``` |
||||
@ -0,0 +1,138 @@
|
||||
package agent |
||||
|
||||
import ( |
||||
"encoding/json" |
||||
"net/http" |
||||
"net/http/httptest" |
||||
"testing" |
||||
"time" |
||||
|
||||
"github.com/stretchr/testify/require" |
||||
|
||||
"github.com/hashicorp/consul/agent/grpc-external/limiter" |
||||
"github.com/hashicorp/consul/agent/proxycfg" |
||||
"github.com/hashicorp/consul/agent/structs" |
||||
"github.com/hashicorp/consul/api" |
||||
"github.com/hashicorp/consul/testrpc" |
||||
) |
||||
|
||||
func TestAgent_local_proxycfg(t *testing.T) { |
||||
a := NewTestAgent(t, TestACLConfig()) |
||||
defer a.Shutdown() |
||||
|
||||
testrpc.WaitForLeader(t, a.RPC, "dc1") |
||||
|
||||
token := generateUUID() |
||||
|
||||
svc := &structs.NodeService{ |
||||
ID: "db", |
||||
Service: "db", |
||||
Port: 5000, |
||||
EnterpriseMeta: *structs.DefaultEnterpriseMetaInDefaultPartition(), |
||||
} |
||||
require.NoError(t, a.State.AddServiceWithChecks(svc, nil, token, true)) |
||||
|
||||
proxy := &structs.NodeService{ |
||||
Kind: structs.ServiceKindConnectProxy, |
||||
ID: "db-sidecar-proxy", |
||||
Service: "db-sidecar-proxy", |
||||
Port: 5000, |
||||
// Set this internal state that we expect sidecar registrations to have.
|
||||
LocallyRegisteredAsSidecar: true, |
||||
Proxy: structs.ConnectProxyConfig{ |
||||
DestinationServiceName: "db", |
||||
Upstreams: structs.TestUpstreams(t), |
||||
}, |
||||
EnterpriseMeta: *structs.DefaultEnterpriseMetaInDefaultPartition(), |
||||
} |
||||
require.NoError(t, a.State.AddServiceWithChecks(proxy, nil, token, true)) |
||||
|
||||
// This is a little gross, but this gives us the layered pair of
|
||||
// local/catalog sources for now.
|
||||
cfg := a.xdsServer.CfgSrc |
||||
|
||||
var ( |
||||
timer = time.After(100 * time.Millisecond) |
||||
timerFired = false |
||||
finalTimer <-chan time.Time |
||||
) |
||||
|
||||
var ( |
||||
firstTime = true |
||||
ch <-chan *proxycfg.ConfigSnapshot |
||||
stc limiter.SessionTerminatedChan |
||||
cancel proxycfg.CancelFunc |
||||
) |
||||
defer func() { |
||||
if cancel != nil { |
||||
cancel() |
||||
} |
||||
}() |
||||
for { |
||||
if ch == nil { |
||||
// Sign up for a stream of config snapshots, in the same manner as the xds server.
|
||||
sid := proxy.CompoundServiceID() |
||||
|
||||
if firstTime { |
||||
firstTime = false |
||||
} else { |
||||
t.Logf("re-creating watch") |
||||
} |
||||
|
||||
// Prior to fixes in https://github.com/hashicorp/consul/pull/16497
|
||||
// this call to Watch() would deadlock.
|
||||
var err error |
||||
ch, stc, cancel, err = cfg.Watch(sid, a.config.NodeName, token) |
||||
require.NoError(t, err) |
||||
} |
||||
select { |
||||
case <-stc: |
||||
t.Fatal("session unexpectedly terminated") |
||||
case snap, ok := <-ch: |
||||
if !ok { |
||||
t.Logf("channel is closed") |
||||
cancel() |
||||
ch, stc, cancel = nil, nil, nil |
||||
continue |
||||
} |
||||
require.NotNil(t, snap) |
||||
if !timerFired { |
||||
t.Fatal("should not have gotten snapshot until after we manifested the token") |
||||
} |
||||
return |
||||
case <-timer: |
||||
timerFired = true |
||||
finalTimer = time.After(1 * time.Second) |
||||
|
||||
// This simulates the eventual consistency of a token
|
||||
// showing up on a server after it's creation by
|
||||
// pre-creating the UUID and later using that as the
|
||||
// initial SecretID for a real token.
|
||||
gotToken := testWriteToken(t, a, &api.ACLToken{ |
||||
AccessorID: generateUUID(), |
||||
SecretID: token, |
||||
Description: "my token", |
||||
ServiceIdentities: []*api.ACLServiceIdentity{{ |
||||
ServiceName: "db", |
||||
}}, |
||||
}) |
||||
require.Equal(t, token, gotToken) |
||||
case <-finalTimer: |
||||
t.Fatal("did not receive a snapshot after the token manifested") |
||||
} |
||||
} |
||||
|
||||
} |
||||
|
||||
func testWriteToken(t *testing.T, a *TestAgent, tok *api.ACLToken) string { |
||||
req, _ := http.NewRequest("PUT", "/v1/acl/token", jsonReader(tok)) |
||||
req.Header.Add("X-Consul-Token", "root") |
||||
resp := httptest.NewRecorder() |
||||
a.srv.h.ServeHTTP(resp, req) |
||||
require.Equal(t, http.StatusOK, resp.Code) |
||||
|
||||
dec := json.NewDecoder(resp.Body) |
||||
aclResp := &structs.ACLToken{} |
||||
require.NoError(t, dec.Decode(aclResp)) |
||||
return aclResp.SecretID |
||||
} |
||||
Loading…
Reference in new issue