mirror of
https://github.com/status-im/consul.git
synced 2025-01-11 14:24:39 +00:00
a70e1886c9
Adds a new gRPC streaming endpoint (WatchRoots) that dataplane clients will use to fetch the current list of active Connect CA roots and receive new lists whenever the roots are rotated.
383 lines
9.0 KiB
Go
383 lines
9.0 KiB
Go
package submatview_test
|
|
|
|
import (
|
|
"context"
|
|
"fmt"
|
|
"math/rand"
|
|
"net"
|
|
"sort"
|
|
"sync"
|
|
"sync/atomic"
|
|
"testing"
|
|
"time"
|
|
|
|
"github.com/google/go-cmp/cmp"
|
|
"github.com/google/go-cmp/cmp/cmpopts"
|
|
"github.com/hashicorp/go-hclog"
|
|
"github.com/stretchr/testify/require"
|
|
"golang.org/x/sync/errgroup"
|
|
"google.golang.org/grpc"
|
|
|
|
"github.com/hashicorp/consul/acl"
|
|
"github.com/hashicorp/consul/agent/cache"
|
|
"github.com/hashicorp/consul/agent/consul/state"
|
|
"github.com/hashicorp/consul/agent/consul/stream"
|
|
"github.com/hashicorp/consul/agent/grpc/private/services/subscribe"
|
|
"github.com/hashicorp/consul/agent/rpcclient/health"
|
|
"github.com/hashicorp/consul/agent/structs"
|
|
"github.com/hashicorp/consul/agent/submatview"
|
|
"github.com/hashicorp/consul/proto/pbsubscribe"
|
|
)
|
|
|
|
func TestStore_IntegrationWithBackend(t *testing.T) {
|
|
if testing.Short() {
|
|
t.Skip("too slow for testing.Short")
|
|
}
|
|
|
|
var maxIndex uint64 = 200
|
|
count := &counter{latest: 3}
|
|
producers := map[string]*eventProducer{
|
|
state.EventSubjectService{Key: "srv1"}.String(): newEventProducer(pbsubscribe.Topic_ServiceHealth, "srv1", count, maxIndex),
|
|
state.EventSubjectService{Key: "srv2"}.String(): newEventProducer(pbsubscribe.Topic_ServiceHealth, "srv2", count, maxIndex),
|
|
state.EventSubjectService{Key: "srv3"}.String(): newEventProducer(pbsubscribe.Topic_ServiceHealth, "srv3", count, maxIndex),
|
|
}
|
|
|
|
sh := snapshotHandler{producers: producers}
|
|
handlers := map[stream.Topic]stream.SnapshotFunc{
|
|
pbsubscribe.Topic_ServiceHealth: sh.Snapshot,
|
|
}
|
|
pub := stream.NewEventPublisher(handlers, 10*time.Millisecond)
|
|
|
|
ctx, cancel := context.WithCancel(context.Background())
|
|
defer cancel()
|
|
go pub.Run(ctx)
|
|
|
|
store := submatview.NewStore(hclog.New(nil))
|
|
go store.Run(ctx)
|
|
|
|
addr := runServer(t, pub)
|
|
|
|
consumers := []*consumer{
|
|
newConsumer(t, addr, store, "srv1"),
|
|
newConsumer(t, addr, store, "srv1"),
|
|
newConsumer(t, addr, store, "srv1"),
|
|
newConsumer(t, addr, store, "srv2"),
|
|
newConsumer(t, addr, store, "srv2"),
|
|
newConsumer(t, addr, store, "srv2"),
|
|
}
|
|
|
|
group, gctx := errgroup.WithContext(ctx)
|
|
for i := range producers {
|
|
producer := producers[i]
|
|
group.Go(func() error {
|
|
producer.Produce(gctx, pub)
|
|
return nil
|
|
})
|
|
}
|
|
|
|
for i := range consumers {
|
|
consumer := consumers[i]
|
|
group.Go(func() error {
|
|
return consumer.Consume(gctx, maxIndex)
|
|
})
|
|
}
|
|
|
|
_ = group.Wait()
|
|
|
|
for i, consumer := range consumers {
|
|
t.Run(fmt.Sprintf("consumer %d", i), func(t *testing.T) {
|
|
require.True(t, len(consumer.states) > 2, "expected more than %d events", len(consumer.states))
|
|
|
|
expected := producers[state.EventSubjectService{Key: consumer.srvName}.String()].nodesByIndex
|
|
for idx, nodes := range consumer.states {
|
|
assertDeepEqual(t, idx, expected[idx], nodes)
|
|
}
|
|
})
|
|
}
|
|
}
|
|
|
|
func assertDeepEqual(t *testing.T, idx uint64, x, y interface{}) {
|
|
t.Helper()
|
|
if diff := cmp.Diff(x, y, cmpopts.EquateEmpty()); diff != "" {
|
|
t.Fatalf("assertion failed: values at index %d are not equal\n--- expected\n+++ actual\n%v", idx, diff)
|
|
}
|
|
}
|
|
|
|
func stateFromUpdates(u cache.UpdateEvent) []string {
|
|
var result []string
|
|
for _, node := range u.Result.(*structs.IndexedCheckServiceNodes).Nodes {
|
|
result = append(result, node.Node.Node)
|
|
}
|
|
|
|
sort.Strings(result)
|
|
return result
|
|
}
|
|
|
|
func runServer(t *testing.T, pub *stream.EventPublisher) net.Addr {
|
|
subSrv := &subscribe.Server{
|
|
Backend: backend{pub: pub},
|
|
Logger: hclog.New(nil),
|
|
}
|
|
srv := grpc.NewServer()
|
|
pbsubscribe.RegisterStateChangeSubscriptionServer(srv, subSrv)
|
|
|
|
lis, err := net.Listen("tcp", "127.0.0.1:0")
|
|
require.NoError(t, err)
|
|
|
|
var g errgroup.Group
|
|
g.Go(func() error {
|
|
return srv.Serve(lis)
|
|
})
|
|
t.Cleanup(func() {
|
|
srv.Stop()
|
|
if err := g.Wait(); err != nil {
|
|
t.Log(err.Error())
|
|
}
|
|
})
|
|
|
|
return lis.Addr()
|
|
}
|
|
|
|
type backend struct {
|
|
pub *stream.EventPublisher
|
|
}
|
|
|
|
func (b backend) ResolveTokenAndDefaultMeta(string, *structs.EnterpriseMeta, *acl.AuthorizerContext) (acl.Authorizer, error) {
|
|
return acl.AllowAll(), nil
|
|
}
|
|
|
|
func (b backend) Forward(structs.RPCInfo, func(*grpc.ClientConn) error) (handled bool, err error) {
|
|
return false, nil
|
|
}
|
|
|
|
func (b backend) Subscribe(req *stream.SubscribeRequest) (*stream.Subscription, error) {
|
|
return b.pub.Subscribe(req)
|
|
}
|
|
|
|
var _ subscribe.Backend = (*backend)(nil)
|
|
|
|
type eventProducer struct {
|
|
rand *rand.Rand
|
|
counter *counter
|
|
topic stream.Topic
|
|
srvName string
|
|
nodesByIndex map[uint64][]string
|
|
nodesLock sync.Mutex
|
|
maxIndex uint64
|
|
}
|
|
|
|
func newEventProducer(
|
|
topic stream.Topic,
|
|
srvName string,
|
|
counter *counter,
|
|
maxIndex uint64,
|
|
) *eventProducer {
|
|
return &eventProducer{
|
|
rand: rand.New(rand.NewSource(time.Now().UnixNano())),
|
|
counter: counter,
|
|
nodesByIndex: map[uint64][]string{},
|
|
topic: topic,
|
|
srvName: srvName,
|
|
maxIndex: maxIndex,
|
|
}
|
|
}
|
|
|
|
var minEventDelay = 5 * time.Millisecond
|
|
|
|
func (e *eventProducer) Produce(ctx context.Context, pub *stream.EventPublisher) {
|
|
var nodes []string
|
|
var nextID int
|
|
|
|
for ctx.Err() == nil {
|
|
var event stream.Event
|
|
|
|
action := e.rand.Intn(3)
|
|
if len(nodes) == 0 {
|
|
action = 1
|
|
}
|
|
|
|
idx := e.counter.Next()
|
|
switch action {
|
|
|
|
case 0: // Deregister
|
|
nodeIdx := e.rand.Intn(len(nodes))
|
|
node := nodes[nodeIdx]
|
|
nodes = append(nodes[:nodeIdx], nodes[nodeIdx+1:]...)
|
|
|
|
event = stream.Event{
|
|
Topic: e.topic,
|
|
Index: idx,
|
|
Payload: state.EventPayloadCheckServiceNode{
|
|
Op: pbsubscribe.CatalogOp_Deregister,
|
|
Value: &structs.CheckServiceNode{
|
|
Node: &structs.Node{Node: node},
|
|
Service: &structs.NodeService{
|
|
ID: e.srvName,
|
|
Service: e.srvName,
|
|
},
|
|
},
|
|
},
|
|
}
|
|
|
|
case 1: // Register new
|
|
node := nodeName(nextID)
|
|
nodes = append(nodes, node)
|
|
nextID++
|
|
|
|
event = stream.Event{
|
|
Topic: e.topic,
|
|
Index: idx,
|
|
Payload: state.EventPayloadCheckServiceNode{
|
|
Op: pbsubscribe.CatalogOp_Register,
|
|
Value: &structs.CheckServiceNode{
|
|
Node: &structs.Node{Node: node},
|
|
Service: &structs.NodeService{
|
|
ID: e.srvName,
|
|
Service: e.srvName,
|
|
},
|
|
},
|
|
},
|
|
}
|
|
|
|
case 2: // Register update
|
|
node := nodes[e.rand.Intn(len(nodes))]
|
|
event = stream.Event{
|
|
Topic: e.topic,
|
|
Index: idx,
|
|
Payload: state.EventPayloadCheckServiceNode{
|
|
Op: pbsubscribe.CatalogOp_Register,
|
|
Value: &structs.CheckServiceNode{
|
|
Node: &structs.Node{Node: node},
|
|
Service: &structs.NodeService{
|
|
ID: e.srvName,
|
|
Service: e.srvName,
|
|
},
|
|
},
|
|
},
|
|
}
|
|
|
|
}
|
|
|
|
e.nodesLock.Lock()
|
|
pub.Publish([]stream.Event{event})
|
|
e.nodesByIndex[idx] = copyNodeList(nodes)
|
|
e.nodesLock.Unlock()
|
|
|
|
if idx > e.maxIndex {
|
|
return
|
|
}
|
|
|
|
delay := time.Duration(rand.Intn(25)) * time.Millisecond
|
|
time.Sleep(minEventDelay + delay)
|
|
}
|
|
}
|
|
|
|
func nodeName(i int) string {
|
|
return fmt.Sprintf("node-%d", i)
|
|
}
|
|
|
|
func copyNodeList(nodes []string) []string {
|
|
result := make([]string, len(nodes))
|
|
copy(result, nodes)
|
|
sort.Strings(result)
|
|
return result
|
|
}
|
|
|
|
type counter struct {
|
|
latest uint64
|
|
}
|
|
|
|
func (c *counter) Next() uint64 {
|
|
return atomic.AddUint64(&c.latest, 1)
|
|
}
|
|
|
|
type consumer struct {
|
|
healthClient *health.Client
|
|
states map[uint64][]string
|
|
srvName string
|
|
}
|
|
|
|
func newConsumer(t *testing.T, addr net.Addr, store *submatview.Store, srv string) *consumer {
|
|
conn, err := grpc.Dial(addr.String(), grpc.WithInsecure())
|
|
require.NoError(t, err)
|
|
|
|
c := &health.Client{
|
|
UseStreamingBackend: true,
|
|
ViewStore: store,
|
|
MaterializerDeps: health.MaterializerDeps{
|
|
Conn: conn,
|
|
Logger: hclog.New(nil),
|
|
},
|
|
}
|
|
|
|
return &consumer{
|
|
healthClient: c,
|
|
states: make(map[uint64][]string),
|
|
srvName: srv,
|
|
}
|
|
}
|
|
|
|
func (c *consumer) Consume(ctx context.Context, maxIndex uint64) error {
|
|
req := structs.ServiceSpecificRequest{ServiceName: c.srvName}
|
|
updateCh := make(chan cache.UpdateEvent, 10)
|
|
|
|
group, cctx := errgroup.WithContext(ctx)
|
|
group.Go(func() error {
|
|
return c.healthClient.Notify(cctx, req, "", updateCh)
|
|
})
|
|
group.Go(func() error {
|
|
var idx uint64
|
|
for {
|
|
if idx >= maxIndex {
|
|
return nil
|
|
}
|
|
select {
|
|
case u := <-updateCh:
|
|
idx = u.Meta.Index
|
|
c.states[u.Meta.Index] = stateFromUpdates(u)
|
|
case <-cctx.Done():
|
|
return nil
|
|
}
|
|
}
|
|
})
|
|
return group.Wait()
|
|
}
|
|
|
|
type snapshotHandler struct {
|
|
producers map[string]*eventProducer
|
|
}
|
|
|
|
func (s *snapshotHandler) Snapshot(req stream.SubscribeRequest, buf stream.SnapshotAppender) (index uint64, err error) {
|
|
producer := s.producers[req.Subject.String()]
|
|
|
|
producer.nodesLock.Lock()
|
|
defer producer.nodesLock.Unlock()
|
|
idx := atomic.LoadUint64(&producer.counter.latest)
|
|
|
|
// look backwards for an index that was used by the producer
|
|
nodes, ok := producer.nodesByIndex[idx]
|
|
for !ok && idx > 0 {
|
|
idx--
|
|
nodes, ok = producer.nodesByIndex[idx]
|
|
}
|
|
|
|
for _, node := range nodes {
|
|
event := stream.Event{
|
|
Topic: producer.topic,
|
|
Index: idx,
|
|
Payload: state.EventPayloadCheckServiceNode{
|
|
Op: pbsubscribe.CatalogOp_Register,
|
|
Value: &structs.CheckServiceNode{
|
|
Node: &structs.Node{Node: node},
|
|
Service: &structs.NodeService{
|
|
ID: producer.srvName,
|
|
Service: producer.srvName,
|
|
},
|
|
},
|
|
},
|
|
}
|
|
buf.Append([]stream.Event{event})
|
|
}
|
|
return idx, nil
|
|
}
|