f3c80c4eef
Protobuf Refactoring for Multi-Module Cleanliness This commit includes the following: Moves all packages that were within proto/ to proto/private Rewrites imports to account for the packages being moved Adds in buf.work.yaml to enable buf workspaces Names the proto-public buf module so that we can override the Go package imports within proto/buf.yaml Bumps the buf version dependency to 1.14.0 (I was trying out the version to see if it would get around an issue - it didn't but it also doesn't break things and it seemed best to keep up with the toolchain changes) Why: In the future we will need to consume other protobuf dependencies such as the Google HTTP annotations for openapi generation or grpc-gateway usage. There were some recent changes to have our own ratelimiting annotations. The two combined were not working when I was trying to use them together (attempting to rebase another branch) Buf workspaces should be the solution to the problem Buf workspaces means that each module will have generated Go code that embeds proto file names relative to the proto dir and not the top level repo root. This resulted in proto file name conflicts in the Go global protobuf type registry. The solution to that was to add in a private/ directory into the path within the proto/ directory. That then required rewriting all the imports. Is this safe? AFAICT yes The gRPC wire protocol doesn't seem to care about the proto file names (although the Go grpc code does tack on the proto file name as Metadata in the ServiceDesc) Other than imports, there were no changes to any generated code as a result of this.
489 lines
13 KiB
Go
489 lines
13 KiB
Go
package state
|
|
|
|
import (
|
|
"context"
|
|
"testing"
|
|
"time"
|
|
|
|
"github.com/stretchr/testify/require"
|
|
|
|
"github.com/hashicorp/consul/acl"
|
|
"github.com/hashicorp/consul/agent/consul/stream"
|
|
"github.com/hashicorp/consul/agent/structs"
|
|
"github.com/hashicorp/consul/proto/private/pbsubscribe"
|
|
)
|
|
|
|
func TestStore_IntegrationWithEventPublisher_ACLTokenUpdate(t *testing.T) {
|
|
if testing.Short() {
|
|
t.Skip("too slow for testing.Short")
|
|
}
|
|
|
|
t.Parallel()
|
|
s := testACLTokensStateStore(t)
|
|
|
|
// Setup token and wait for good state
|
|
token := createTokenAndWaitForACLEventPublish(t, s)
|
|
|
|
// Register the subscription.
|
|
subscription := &stream.SubscribeRequest{
|
|
Topic: topicService,
|
|
Subject: stream.StringSubject("nope"),
|
|
Token: token.SecretID,
|
|
}
|
|
ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
|
|
defer cancel()
|
|
|
|
publisher := stream.NewEventPublisher(0)
|
|
registerTestSnapshotHandlers(t, s, publisher)
|
|
go publisher.Run(ctx)
|
|
s.db.publisher = publisher
|
|
sub, err := publisher.Subscribe(subscription)
|
|
require.NoError(t, err)
|
|
defer sub.Unsubscribe()
|
|
|
|
eventCh := testRunSub(sub)
|
|
|
|
// Stream should get EndOfSnapshot
|
|
e := assertEvent(t, eventCh)
|
|
require.True(t, e.IsEndOfSnapshot())
|
|
|
|
// Update an unrelated token.
|
|
token2 := &structs.ACLToken{
|
|
AccessorID: "a7bbf480-8440-4f55-acfc-6fdca25cb13e",
|
|
SecretID: "72e81982-7a0f-491f-a60e-c9c802ac1402",
|
|
}
|
|
token2.SetHash(false)
|
|
require.NoError(t, s.ACLTokenSet(3, token2.Clone()))
|
|
|
|
// Ensure there's no reset event.
|
|
assertNoEvent(t, eventCh)
|
|
|
|
// Now update the token used in the subscriber.
|
|
token3 := &structs.ACLToken{
|
|
AccessorID: "3af117a9-2233-4cf4-8ff8-3c749c9906b4",
|
|
SecretID: "4268ce0d-d7ae-4718-8613-42eba9036020",
|
|
Description: "something else",
|
|
}
|
|
token3.SetHash(false)
|
|
require.NoError(t, s.ACLTokenSet(4, token3.Clone()))
|
|
|
|
// Ensure the reset event was sent.
|
|
err = assertErr(t, eventCh)
|
|
require.Equal(t, stream.ErrSubForceClosed, err)
|
|
|
|
// Register another subscription.
|
|
subscription2 := &stream.SubscribeRequest{
|
|
Topic: topicService,
|
|
Subject: stream.StringSubject("nope"),
|
|
Token: token.SecretID,
|
|
}
|
|
sub2, err := publisher.Subscribe(subscription2)
|
|
require.NoError(t, err)
|
|
defer sub2.Unsubscribe()
|
|
|
|
eventCh2 := testRunSub(sub2)
|
|
|
|
// Expect initial EoS
|
|
e = assertEvent(t, eventCh2)
|
|
require.True(t, e.IsEndOfSnapshot())
|
|
|
|
// Delete the unrelated token.
|
|
require.NoError(t, s.ACLTokenDeleteByAccessor(5, token2.AccessorID, nil))
|
|
|
|
// Ensure there's no reset event.
|
|
assertNoEvent(t, eventCh2)
|
|
|
|
// Delete the token used by the subscriber.
|
|
require.NoError(t, s.ACLTokenDeleteByAccessor(6, token.AccessorID, nil))
|
|
|
|
// Ensure the reset event was sent.
|
|
err = assertErr(t, eventCh2)
|
|
require.Equal(t, stream.ErrSubForceClosed, err)
|
|
}
|
|
|
|
func TestStore_IntegrationWithEventPublisher_ACLPolicyUpdate(t *testing.T) {
|
|
if testing.Short() {
|
|
t.Skip("too slow for testing.Short")
|
|
}
|
|
|
|
t.Parallel()
|
|
s := testACLTokensStateStore(t)
|
|
|
|
// Create token and wait for good state
|
|
token := createTokenAndWaitForACLEventPublish(t, s)
|
|
|
|
// Register the subscription.
|
|
subscription := &stream.SubscribeRequest{
|
|
Topic: topicService,
|
|
Subject: stream.StringSubject("nope"),
|
|
Token: token.SecretID,
|
|
}
|
|
ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
|
|
defer cancel()
|
|
|
|
publisher := stream.NewEventPublisher(0)
|
|
registerTestSnapshotHandlers(t, s, publisher)
|
|
go publisher.Run(ctx)
|
|
s.db.publisher = publisher
|
|
sub, err := publisher.Subscribe(subscription)
|
|
require.NoError(t, err)
|
|
defer sub.Unsubscribe()
|
|
|
|
eventCh := testRunSub(sub)
|
|
|
|
// Ignore the end of snapshot event
|
|
e := assertEvent(t, eventCh)
|
|
require.True(t, e.IsEndOfSnapshot(), "event should be a EoS got %v", e)
|
|
|
|
// Update an unrelated policy.
|
|
policy2 := structs.ACLPolicy{
|
|
ID: testPolicyID_C,
|
|
Name: "foo-read",
|
|
Rules: `node "foo" { policy = "read" }`,
|
|
Datacenters: []string{"dc1"},
|
|
}
|
|
policy2.SetHash(false)
|
|
require.NoError(t, s.ACLPolicySet(3, &policy2))
|
|
|
|
// Ensure there's no reset event.
|
|
assertNoEvent(t, eventCh)
|
|
|
|
// Now update the policy used in the subscriber.
|
|
policy3 := structs.ACLPolicy{
|
|
ID: testPolicyID_A,
|
|
Name: "node-read",
|
|
Rules: `node_prefix "" { policy = "write" }`,
|
|
Datacenters: []string{"dc1"},
|
|
}
|
|
policy3.SetHash(false)
|
|
require.NoError(t, s.ACLPolicySet(4, &policy3))
|
|
|
|
// Ensure the reset event was sent.
|
|
assertReset(t, eventCh, true)
|
|
|
|
// Register another subscription.
|
|
subscription2 := &stream.SubscribeRequest{
|
|
Topic: topicService,
|
|
Subject: stream.StringSubject("nope"),
|
|
Token: token.SecretID,
|
|
}
|
|
sub, err = publisher.Subscribe(subscription2)
|
|
require.NoError(t, err)
|
|
defer sub.Unsubscribe()
|
|
|
|
eventCh = testRunSub(sub)
|
|
|
|
// Ignore the end of snapshot event
|
|
e = assertEvent(t, eventCh)
|
|
require.True(t, e.IsEndOfSnapshot(), "event should be a EoS got %v", e)
|
|
|
|
// Delete the unrelated policy.
|
|
require.NoError(t, s.ACLPolicyDeleteByID(5, testPolicyID_C, nil))
|
|
|
|
// Ensure there's no reload event.
|
|
assertNoEvent(t, eventCh)
|
|
|
|
// Delete the policy used by the subscriber.
|
|
require.NoError(t, s.ACLPolicyDeleteByID(6, testPolicyID_A, nil))
|
|
|
|
// Ensure the reload event was sent.
|
|
err = assertErr(t, eventCh)
|
|
require.Equal(t, stream.ErrSubForceClosed, err)
|
|
|
|
// Register another subscription.
|
|
subscription3 := &stream.SubscribeRequest{
|
|
Topic: topicService,
|
|
Subject: stream.StringSubject("nope"),
|
|
Token: token.SecretID,
|
|
}
|
|
sub, err = publisher.Subscribe(subscription3)
|
|
require.NoError(t, err)
|
|
defer sub.Unsubscribe()
|
|
|
|
eventCh = testRunSub(sub)
|
|
|
|
// Ignore the end of snapshot event
|
|
e = assertEvent(t, eventCh)
|
|
require.True(t, e.IsEndOfSnapshot(), "event should be a EoS got %v", e)
|
|
|
|
// Now update the policy used in role B, but not directly in the token.
|
|
policy4 := structs.ACLPolicy{
|
|
ID: testPolicyID_B,
|
|
Name: "node-read",
|
|
Rules: `node_prefix "foo" { policy = "read" }`,
|
|
Datacenters: []string{"dc1"},
|
|
}
|
|
policy4.SetHash(false)
|
|
require.NoError(t, s.ACLPolicySet(7, &policy4))
|
|
|
|
// Ensure the reset event was sent.
|
|
assertReset(t, eventCh, true)
|
|
}
|
|
|
|
func TestStore_IntegrationWithEventPublisher_ACLRoleUpdate(t *testing.T) {
|
|
if testing.Short() {
|
|
t.Skip("too slow for testing.Short")
|
|
}
|
|
|
|
t.Parallel()
|
|
s := testACLTokensStateStore(t)
|
|
|
|
// Create token and wait for good state
|
|
token := createTokenAndWaitForACLEventPublish(t, s)
|
|
|
|
// Register the subscription.
|
|
subscription := &stream.SubscribeRequest{
|
|
Topic: topicService,
|
|
Subject: stream.StringSubject("nope"),
|
|
Token: token.SecretID,
|
|
}
|
|
ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
|
|
defer cancel()
|
|
|
|
publisher := stream.NewEventPublisher(0)
|
|
registerTestSnapshotHandlers(t, s, publisher)
|
|
go publisher.Run(ctx)
|
|
s.db.publisher = publisher
|
|
sub, err := publisher.Subscribe(subscription)
|
|
require.NoError(t, err)
|
|
|
|
eventCh := testRunSub(sub)
|
|
|
|
// Stream should get EndOfSnapshot
|
|
e := assertEvent(t, eventCh)
|
|
require.True(t, e.IsEndOfSnapshot())
|
|
|
|
// Update an unrelated role (the token has role testRoleID_B).
|
|
role := structs.ACLRole{
|
|
ID: testRoleID_A,
|
|
Name: "unrelated-role",
|
|
Description: "test",
|
|
}
|
|
role.SetHash(false)
|
|
require.NoError(t, s.ACLRoleSet(3, &role))
|
|
|
|
// Ensure there's no reload event.
|
|
assertNoEvent(t, eventCh)
|
|
|
|
// Now update the role used by the token in the subscriber.
|
|
role2 := structs.ACLRole{
|
|
ID: testRoleID_B,
|
|
Name: "my-new-role",
|
|
Description: "changed",
|
|
}
|
|
role2.SetHash(false)
|
|
require.NoError(t, s.ACLRoleSet(4, &role2))
|
|
|
|
// Ensure the reload event was sent.
|
|
assertReset(t, eventCh, false)
|
|
|
|
// Register another subscription.
|
|
subscription2 := &stream.SubscribeRequest{
|
|
Topic: topicService,
|
|
Subject: stream.StringSubject("nope"),
|
|
Token: token.SecretID,
|
|
}
|
|
sub, err = publisher.Subscribe(subscription2)
|
|
require.NoError(t, err)
|
|
|
|
eventCh = testRunSub(sub)
|
|
|
|
// Ignore the end of snapshot event
|
|
e = assertEvent(t, eventCh)
|
|
require.True(t, e.IsEndOfSnapshot(), "event should be a EoS got %v", e)
|
|
|
|
// Delete the unrelated policy.
|
|
require.NoError(t, s.ACLRoleDeleteByID(5, testRoleID_A, nil))
|
|
|
|
// Ensure there's no reload event.
|
|
assertNoEvent(t, eventCh)
|
|
|
|
// Delete the policy used by the subscriber.
|
|
require.NoError(t, s.ACLRoleDeleteByID(6, testRoleID_B, nil))
|
|
|
|
// Ensure the reload event was sent.
|
|
assertReset(t, eventCh, false)
|
|
}
|
|
|
|
type nextResult struct {
|
|
Event stream.Event
|
|
Err error
|
|
}
|
|
|
|
func testRunSub(sub *stream.Subscription) <-chan nextResult {
|
|
eventCh := make(chan nextResult, 1)
|
|
go func() {
|
|
for {
|
|
es, err := sub.Next(context.TODO())
|
|
eventCh <- nextResult{
|
|
Event: es,
|
|
Err: err,
|
|
}
|
|
if err != nil {
|
|
return
|
|
}
|
|
}
|
|
}()
|
|
return eventCh
|
|
}
|
|
|
|
func assertNoEvent(t *testing.T, eventCh <-chan nextResult) {
|
|
t.Helper()
|
|
select {
|
|
case next := <-eventCh:
|
|
require.NoError(t, next.Err)
|
|
require.Len(t, next.Event, 1)
|
|
t.Fatalf("got unwanted event: %#v", next.Event.Payload)
|
|
case <-time.After(100 * time.Millisecond):
|
|
}
|
|
}
|
|
|
|
func assertEvent(t *testing.T, eventCh <-chan nextResult) *stream.Event {
|
|
t.Helper()
|
|
select {
|
|
case next := <-eventCh:
|
|
require.NoError(t, next.Err)
|
|
return &next.Event
|
|
case <-time.After(100 * time.Millisecond):
|
|
t.Fatalf("no event after 100ms")
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func assertErr(t *testing.T, eventCh <-chan nextResult) error {
|
|
t.Helper()
|
|
select {
|
|
case next := <-eventCh:
|
|
require.Error(t, next.Err)
|
|
return next.Err
|
|
case <-time.After(100 * time.Millisecond):
|
|
t.Fatalf("no err after 100ms")
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// assertReset checks that a ResetStream event is send to the subscription
|
|
// within 100ms. If allowEOS is true it will ignore any intermediate events that
|
|
// come before the reset provided they are EndOfSnapshot events because in many
|
|
// cases it's non-deterministic whether the snapshot will complete before the
|
|
// acl reset is handled.
|
|
func assertReset(t *testing.T, eventCh <-chan nextResult, allowEOS bool) {
|
|
t.Helper()
|
|
for {
|
|
select {
|
|
case next := <-eventCh:
|
|
if allowEOS {
|
|
if next.Err == nil && next.Event.IsEndOfSnapshot() {
|
|
continue
|
|
}
|
|
}
|
|
require.Error(t, next.Err)
|
|
require.Equal(t, stream.ErrSubForceClosed, next.Err)
|
|
return
|
|
case <-time.After(100 * time.Millisecond):
|
|
t.Fatalf("no err after 100ms")
|
|
}
|
|
}
|
|
}
|
|
|
|
type topic string
|
|
|
|
func (t topic) String() string {
|
|
return string(t)
|
|
}
|
|
|
|
var topicService topic = "test-topic-service"
|
|
|
|
func (s *Store) topicServiceTestHandler(req stream.SubscribeRequest, snap stream.SnapshotAppender) (uint64, error) {
|
|
key := req.Subject.String()
|
|
|
|
idx, nodes, err := s.ServiceNodes(nil, key, nil, structs.TODOPeerKeyword)
|
|
if err != nil {
|
|
return idx, err
|
|
}
|
|
|
|
for _, node := range nodes {
|
|
event := stream.Event{
|
|
Topic: req.Topic,
|
|
Index: node.ModifyIndex,
|
|
Payload: nodePayload{node: node, key: key},
|
|
}
|
|
snap.Append([]stream.Event{event})
|
|
}
|
|
return idx, nil
|
|
}
|
|
|
|
func registerTestSnapshotHandlers(t *testing.T, s *Store, publisher EventPublisher) {
|
|
t.Helper()
|
|
err := publisher.RegisterHandler(topicService, s.topicServiceTestHandler, false)
|
|
require.NoError(t, err)
|
|
}
|
|
|
|
type nodePayload struct {
|
|
key string
|
|
node *structs.ServiceNode
|
|
}
|
|
|
|
func (p nodePayload) HasReadPermission(acl.Authorizer) bool {
|
|
return true
|
|
}
|
|
|
|
func (p nodePayload) Subject() stream.Subject {
|
|
return stream.StringSubject(p.key)
|
|
}
|
|
|
|
func (e nodePayload) ToSubscriptionEvent(idx uint64) *pbsubscribe.Event {
|
|
panic("EventPayloadCARoots does not implement ToSubscriptionEvent")
|
|
}
|
|
|
|
func createTokenAndWaitForACLEventPublish(t *testing.T, s *Store) *structs.ACLToken {
|
|
token := &structs.ACLToken{
|
|
AccessorID: "3af117a9-2233-4cf4-8ff8-3c749c9906b4",
|
|
SecretID: "4268ce0d-d7ae-4718-8613-42eba9036020",
|
|
Description: "something",
|
|
Policies: []structs.ACLTokenPolicyLink{
|
|
{ID: testPolicyID_A},
|
|
},
|
|
Roles: []structs.ACLTokenRoleLink{
|
|
{ID: testRoleID_B},
|
|
},
|
|
}
|
|
token.SetHash(false)
|
|
|
|
// If we subscribe immediately after we create a token we race with the
|
|
// publisher that is publishing the ACL token event for the token we just
|
|
// created. That means that the subscription we create right after will often
|
|
// be immediately reset. The most reliable way to avoid that without just
|
|
// sleeping for some arbitrary time is to pre-subscribe using the token before
|
|
// it actually exists (which works because the publisher doesn't check tokens
|
|
// it assumes something lower down did that) and then wait for it to be reset
|
|
// so we know the initial token write event has been sent out before
|
|
// continuing...
|
|
req := &stream.SubscribeRequest{
|
|
Topic: topicService,
|
|
Subject: stream.StringSubject("nope"),
|
|
Token: token.SecretID,
|
|
}
|
|
ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
|
|
defer cancel()
|
|
|
|
publisher := stream.NewEventPublisher(0)
|
|
registerTestSnapshotHandlers(t, s, publisher)
|
|
go publisher.Run(ctx)
|
|
|
|
s.db.publisher = publisher
|
|
sub, err := publisher.Subscribe(req)
|
|
require.NoError(t, err)
|
|
defer sub.Unsubscribe()
|
|
|
|
eventCh := testRunSub(sub)
|
|
|
|
// Create the ACL token to be used in the subscription.
|
|
require.NoError(t, s.ACLTokenSet(2, token.Clone()))
|
|
|
|
// Wait for the pre-subscription to be reset
|
|
assertReset(t, eventCh, true)
|
|
|
|
return token
|
|
}
|