Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Fix some data races #10396

Merged
merged 8 commits into from
Jul 16, 2021
Merged

Fix some data races #10396

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
8 commits
Select commit Hold shift + click to select a range
baa2b86
consul: fix data race in leader CA tests
dnephin Jun 2, 2021
970f5d7
agent: fix two data race in agent tests
dnephin Jun 8, 2021
0acfc2c
agent: fix a data race in DNS tests
dnephin Jun 8, 2021
678014d
agent: fix a data race in a test
dnephin Jun 9, 2021
a0ca381
agent: remove deprecated call in a test
dnephin Jun 9, 2021
414ce3f
Update serf
dnephin Jun 9, 2021
fa47c04
Fix a data race in TestACLResolver_Client
dnephin Jun 9, 2021
a77575e
acl: use SetHash consistently in testPolicyForID
dnephin Jul 16, 2021
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
24 changes: 21 additions & 3 deletions agent/agent_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,6 +18,7 @@ import (
"path/filepath"
"strconv"
"strings"
"sync"
"testing"
"time"

Expand Down Expand Up @@ -3729,10 +3730,27 @@ func TestAgent_SecurityChecks(t *testing.T) {
defer a.Shutdown()

data := make([]byte, 0, 8192)
bytesBuffer := bytes.NewBuffer(data)
a.LogOutput = bytesBuffer
buf := &syncBuffer{b: bytes.NewBuffer(data)}
a.LogOutput = buf
assert.NoError(t, a.Start(t))
assert.Contains(t, bytesBuffer.String(), "using enable-script-checks without ACLs and without allow_write_http_from is DANGEROUS")
assert.Contains(t, buf.String(), "using enable-script-checks without ACLs and without allow_write_http_from is DANGEROUS")
}

type syncBuffer struct {
lock sync.RWMutex
b *bytes.Buffer
}

func (b *syncBuffer) Write(data []byte) (int, error) {
b.lock.Lock()
defer b.lock.Unlock()
return b.b.Write(data)
}

func (b *syncBuffer) String() string {
b.lock.Lock()
defer b.lock.Unlock()
return b.b.String()
}

func TestAgent_ReloadConfigOutgoingRPCConfig(t *testing.T) {
Expand Down
36 changes: 24 additions & 12 deletions agent/consul/acl_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -310,61 +310,73 @@ func testIdentityForToken(token string) (bool, structs.ACLIdentity, error) {
func testPolicyForID(policyID string) (bool, *structs.ACLPolicy, error) {
switch policyID {
case "acl-ro":
return true, &structs.ACLPolicy{
p := &structs.ACLPolicy{
ID: "acl-ro",
Name: "acl-ro",
Description: "acl-ro",
Rules: `acl = "read"`,
Syntax: acl.SyntaxCurrent,
RaftIndex: structs.RaftIndex{CreateIndex: 1, ModifyIndex: 2},
}, nil
}
p.SetHash(false)
return true, p, nil
case "acl-wr":
return true, &structs.ACLPolicy{
p := &structs.ACLPolicy{
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

How come this change was only needed for a subset of the cases?

Copy link
Contributor Author

@dnephin dnephin Jul 15, 2021 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I believe it is because only the "Concurrent-Token-Resolve" test cases has a race on SetHash, and only these two cases are used by the "Concurrent-Token-Resolve" test case.

I can add SetHash to the other cases as well, if we think that would better imitate production.

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I think adding it to all of them would be nice for that reason. Would also avoid that confusion when someone looks at this again later.

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Sounds good! Added those in the latest commit

ID: "acl-wr",
Name: "acl-wr",
Description: "acl-wr",
Rules: `acl = "write"`,
Syntax: acl.SyntaxCurrent,
RaftIndex: structs.RaftIndex{CreateIndex: 1, ModifyIndex: 2},
}, nil
}
p.SetHash(false)
return true, p, nil
case "service-ro":
return true, &structs.ACLPolicy{
p := &structs.ACLPolicy{
ID: "service-ro",
Name: "service-ro",
Description: "service-ro",
Rules: `service_prefix "" { policy = "read" }`,
Syntax: acl.SyntaxCurrent,
RaftIndex: structs.RaftIndex{CreateIndex: 1, ModifyIndex: 2},
}, nil
}
p.SetHash(false)
return true, p, nil
case "service-wr":
return true, &structs.ACLPolicy{
p := &structs.ACLPolicy{
ID: "service-wr",
Name: "service-wr",
Description: "service-wr",
Rules: `service_prefix "" { policy = "write" }`,
Syntax: acl.SyntaxCurrent,
RaftIndex: structs.RaftIndex{CreateIndex: 1, ModifyIndex: 2},
}, nil
}
p.SetHash(false)
return true, p, nil
case "node-wr":
return true, &structs.ACLPolicy{
p := &structs.ACLPolicy{
ID: "node-wr",
Name: "node-wr",
Description: "node-wr",
Rules: `node_prefix "" { policy = "write"}`,
Syntax: acl.SyntaxCurrent,
Datacenters: []string{"dc1"},
RaftIndex: structs.RaftIndex{CreateIndex: 1, ModifyIndex: 2},
}, nil
}
p.SetHash(false)
Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I'm wondering about this fix. Why does the ACL resolver need to call SetHash ? Shouldn't that be done when the policy is created instead of when it is read?

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

We do set the hash on creation in the PolicySet endpoint: https://github.com/hashicorp/consul/blob/main/agent/consul/acl_endpoint.go#L1204

testPolicyForID is used by a test resolver delegate, so I think it's necessary to do it on reads here because there is no write beyond the policy existing in this function. It resolves these hard coded policies as if they had been created by some other process. (But maybe I'm misunderstanding your question)

Copy link
Contributor Author

@dnephin dnephin Jul 15, 2021 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Thanks! I think that suggests that this fix is reasonable (because it better matches a production flow), which is reassuring.

I think my broader question is about the call to SetHash that caused the data race. The call in resolveWithCache (here).

That seems like a very strange place to be calling SetHash. From my reading of the code this method is called in read flows, which should already have a hash set from when the policy was written. And if not, why would SetHash be called here, and not in a more core place (like the state store or RPC handler) ? My thinking is that if this SetHash call is misplaced, it has the potential to cause or hide bugs, either now or in the future. I guess maybe it was done defensively because we are about to read the hash, but I think sometimes this kind of defense can actually introduce other problems.

I think answering that question isn't a blocker for this PR, but it may be worth exploring further to see if we can rationalize where SetHash is called.

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Yeah that's a good point about resolveWithCache. I'm not sure why it's done there. At first I thought it could have been a new field, where policies before some version doesn't have it, but the Hash field has existed since the New ACLs PR.

It also doesn't seem to be used for ACL compatibility, since the function to extract policies from embedded tokens sets the hash itself:

consul/agent/structs/acl.go

Line 475 in 0527dcf

policy.SetHash(true)

Same with these other synthetic policy extractors:

consul/agent/structs/acl.go

Line 189 in 0527dcf

policy.SetHash(true)

consul/agent/structs/acl.go

Line 233 in 0527dcf

policy.SetHash(true)

My sense is that it was there for tests like this one that don't set the hash in advance. I don't see a non-test use for it.

Agreed that it doesn't need to be handled in this PR.

return true, p, nil
case "dc2-key-wr":
return true, &structs.ACLPolicy{
p := &structs.ACLPolicy{
ID: "dc2-key-wr",
Name: "dc2-key-wr",
Description: "dc2-key-wr",
Rules: `key_prefix "" { policy = "write"}`,
Syntax: acl.SyntaxCurrent,
Datacenters: []string{"dc2"},
RaftIndex: structs.RaftIndex{CreateIndex: 1, ModifyIndex: 2},
}, nil
}
p.SetHash(false)
return true, p, nil
default:
return testPolicyForIDEnterprise(policyID)
}
Expand Down
15 changes: 12 additions & 3 deletions agent/consul/leader_connect_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -373,7 +373,10 @@ func TestLeader_Vault_PrimaryCA_IntermediateRenew(t *testing.T) {
}
})
defer os.RemoveAll(dir1)
defer s1.Shutdown()
defer func() {
s1.Shutdown()
s1.leaderRoutineManager.Wait()
}()

testrpc.WaitForLeader(t, s1.RPC, "dc1")

Expand Down Expand Up @@ -482,7 +485,10 @@ func TestLeader_SecondaryCA_IntermediateRenew(t *testing.T) {
}
})
defer os.RemoveAll(dir1)
defer s1.Shutdown()
defer func() {
s1.Shutdown()
s1.leaderRoutineManager.Wait()
}()

testrpc.WaitForLeader(t, s1.RPC, "dc1")

Expand All @@ -493,7 +499,10 @@ func TestLeader_SecondaryCA_IntermediateRenew(t *testing.T) {
c.Build = "1.6.0"
})
defer os.RemoveAll(dir2)
defer s2.Shutdown()
defer func() {
s2.Shutdown()
s2.leaderRoutineManager.Wait()
}()

// Create the WAN link
joinWAN(t, s2, s1)
Expand Down
11 changes: 6 additions & 5 deletions agent/http_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,17 +20,18 @@ import (
"time"

"github.com/NYTimes/gziphandler"
cleanhttp "github.com/hashicorp/go-cleanhttp"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
"golang.org/x/net/http2"

"github.com/hashicorp/consul/agent/config"
"github.com/hashicorp/consul/agent/structs"
tokenStore "github.com/hashicorp/consul/agent/token"
"github.com/hashicorp/consul/api"
"github.com/hashicorp/consul/sdk/testutil"
"github.com/hashicorp/consul/sdk/testutil/retry"
"github.com/hashicorp/consul/testrpc"
cleanhttp "github.com/hashicorp/go-cleanhttp"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
"golang.org/x/net/http2"
)

func TestHTTPServer_UnixSocket(t *testing.T) {
Expand Down Expand Up @@ -632,7 +633,7 @@ func TestHTTP_wrap_obfuscateLog(t *testing.T) {
}

t.Parallel()
buf := new(bytes.Buffer)
buf := &syncBuffer{b: new(bytes.Buffer)}
a := StartTestAgent(t, TestAgent{LogOutput: buf})
defer a.Shutdown()

Expand Down
52 changes: 31 additions & 21 deletions agent/service_manager_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -8,11 +8,12 @@ import (
"path/filepath"
"testing"

"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"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
)

func TestServiceManager_RegisterService(t *testing.T) {
Expand Down Expand Up @@ -330,26 +331,27 @@ func TestServiceManager_PersistService_API(t *testing.T) {

testrpc.WaitForLeader(t, a.RPC, "dc1")

// Now register a sidecar proxy via the API.
svc := &structs.NodeService{
Kind: structs.ServiceKindConnectProxy,
ID: "web-sidecar-proxy",
Service: "web-sidecar-proxy",
Port: 21000,
Proxy: structs.ConnectProxyConfig{
DestinationServiceName: "web",
DestinationServiceID: "web",
LocalServiceAddress: "127.0.0.1",
LocalServicePort: 8000,
Upstreams: structs.Upstreams{
{
DestinationName: "redis",
DestinationNamespace: "default",
LocalBindPort: 5000,
newNodeService := func() *structs.NodeService {
return &structs.NodeService{
Kind: structs.ServiceKindConnectProxy,
ID: "web-sidecar-proxy",
Service: "web-sidecar-proxy",
Port: 21000,
Proxy: structs.ConnectProxyConfig{
DestinationServiceName: "web",
DestinationServiceID: "web",
LocalServiceAddress: "127.0.0.1",
LocalServicePort: 8000,
Upstreams: structs.Upstreams{
{
DestinationName: "redis",
DestinationNamespace: "default",
LocalBindPort: 5000,
},
},
},
},
EnterpriseMeta: *structs.DefaultEnterpriseMeta(),
EnterpriseMeta: *structs.DefaultEnterpriseMeta(),
}
}

expectState := &structs.NodeService{
Expand Down Expand Up @@ -385,6 +387,7 @@ func TestServiceManager_PersistService_API(t *testing.T) {
EnterpriseMeta: *structs.DefaultEnterpriseMeta(),
}

svc := newNodeService()
dnephin marked this conversation as resolved.
Show resolved Hide resolved
svcID := svc.CompoundServiceID()

svcFile := filepath.Join(a.Config.DataDir, servicesDir, svcID.StringHash())
Expand Down Expand Up @@ -443,8 +446,15 @@ func TestServiceManager_PersistService_API(t *testing.T) {
}

// Updates service definition on disk
svc = newNodeService()
svc.Proxy.LocalServicePort = 8001
require.NoError(a.addServiceFromSource(svc, nil, true, "mytoken", ConfigSourceRemote))
err = a.AddService(AddServiceRequest{
Service: svc,
persist: true,
token: "mytoken",
Source: ConfigSourceRemote,
})
require.NoError(err)
requireFileIsPresent(t, svcFile)
requireFileIsPresent(t, configFile)

Expand Down
6 changes: 4 additions & 2 deletions agent/testagent.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -53,6 +53,8 @@ type TestAgent struct {
Config *config.RuntimeConfig

// LogOutput is the sink for the logs. If nil, logs are written to os.Stderr.
// The io.Writer must allow concurrent reads and writes. Note that
// bytes.Buffer is not safe for concurrent reads and writes.
LogOutput io.Writer

// DataDir may be set to a directory which exists. If is it not set,
Expand Down Expand Up @@ -343,8 +345,8 @@ func (a *TestAgent) Client() *api.Client {
// DNSDisableCompression disables compression for all started DNS servers.
func (a *TestAgent) DNSDisableCompression(b bool) {
for _, srv := range a.dnsServers {
cfg := srv.config.Load().(*dnsConfig)
cfg.DisableCompression = b
a.config.DNSDisableCompression = b
srv.ReloadConfig(a.config)
}
}

Expand Down
2 changes: 1 addition & 1 deletion go.mod
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -55,7 +55,7 @@ require (
github.com/hashicorp/raft v1.3.1
github.com/hashicorp/raft-autopilot v0.1.5
github.com/hashicorp/raft-boltdb v0.0.0-20171010151810-6e5ba93211ea
github.com/hashicorp/serf v0.9.5
github.com/hashicorp/serf v0.9.6-0.20210609195804-2b5dd0cd2de9
github.com/hashicorp/vault/api v1.0.5-0.20200717191844-f687267c8086
github.com/hashicorp/yamux v0.0.0-20200609203250-aecfd211c9ce
github.com/imdario/mergo v0.3.6
Expand Down
3 changes: 2 additions & 1 deletion go.sum
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -291,8 +291,9 @@ github.com/hashicorp/raft-autopilot v0.1.5 h1:onEfMH5uHVdXQqtas36zXUHEZxLdsJVu/n
github.com/hashicorp/raft-autopilot v0.1.5/go.mod h1:Af4jZBwaNOI+tXfIqIdbcAnh/UyyqIMj/pOISIfhArw=
github.com/hashicorp/raft-boltdb v0.0.0-20171010151810-6e5ba93211ea h1:xykPFhrBAS2J0VBzVa5e80b5ZtYuNQtgXjN40qBZlD4=
github.com/hashicorp/raft-boltdb v0.0.0-20171010151810-6e5ba93211ea/go.mod h1:pNv7Wc3ycL6F5oOWn+tPGo2gWD4a5X+yp/ntwdKLjRk=
github.com/hashicorp/serf v0.9.5 h1:EBWvyu9tcRszt3Bxp3KNssBMP1KuHWyO51lz9+786iM=
github.com/hashicorp/serf v0.9.5/go.mod h1:UWDWwZeL5cuWDJdl0C6wrvrUwEqtQ4ZKBKKENpqIUyk=
github.com/hashicorp/serf v0.9.6-0.20210609195804-2b5dd0cd2de9 h1:lCZfMBDn/Puwg9VosHMf/9p9jNDYYkbzVjb4jYjVfqU=
github.com/hashicorp/serf v0.9.6-0.20210609195804-2b5dd0cd2de9/go.mod h1:qapjppkpNXHYTyzx+HqkyWGGkmUxafHjuspm/Bqb2Jc=
github.com/hashicorp/vault/api v1.0.5-0.20200717191844-f687267c8086 h1:OKsyxKi2sNmqm1Gv93adf2AID2FOBFdCbbZn9fGtIdg=
github.com/hashicorp/vault/api v1.0.5-0.20200717191844-f687267c8086/go.mod h1:R3Umvhlxi2TN7Ex2hzOowyeNb+SfbVWI973N+ctaFMk=
github.com/hashicorp/vault/sdk v0.1.14-0.20200519221838-e0cfd64bc267 h1:e1ok06zGrWJW91rzRroyl5nRNqraaBe4d5hiKcVZuHM=
Expand Down
17 changes: 15 additions & 2 deletions lib/routine/routine.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -24,6 +24,10 @@ func (r *routineTracker) running() bool {
}
}

func (r *routineTracker) wait() {
<-r.stoppedCh
}

type Manager struct {
lock sync.RWMutex
logger hclog.Logger
Expand Down Expand Up @@ -131,6 +135,8 @@ func (m *Manager) stopInstance(name string) *routineTracker {
return instance
}

// StopAll goroutines. Once StopAll is called, it is no longer safe to add no
// goroutines to the Manager.
func (m *Manager) StopAll() {
m.lock.Lock()
defer m.lock.Unlock()
Expand All @@ -142,7 +148,14 @@ func (m *Manager) StopAll() {
m.logger.Debug("stopping routine", "routine", name)
routine.cancel()
}
}

// just wipe out the entire map
m.routines = make(map[string]*routineTracker)
// Wait for all goroutines to stop after StopAll is called.
func (m *Manager) Wait() {
m.lock.Lock()
defer m.lock.Unlock()

for _, routine := range m.routines {
routine.wait()
}
}
7 changes: 4 additions & 3 deletions vendor/github.com/hashicorp/serf/serf/serf.go
View file
Open in desktop

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

Loading