Add message bus to tessend

CHANGELOG.md

@@ -10,6 +10,10 @@ Versioning](http://semver.org/spec/v2.0.0.html).
 ### Fixed
 - Fixed the agent `--annotations` and `--labels` flags.
 
+### Added
+- Added the message bus to Tessend in order to track Tessen configuration changes from the API.
+- Added a performance optimizing `Count()` function to the generic store.
+
 ## [5.5.1] - 2019-04-15
 
 ### Changed

backend/apid/actions/tessen.go

@@ -2,19 +2,22 @@ package actions
 
 import (
 	corev2 "github.com/sensu/sensu-go/api/core/v2"
+	"github.com/sensu/sensu-go/backend/messaging"
 	"github.com/sensu/sensu-go/backend/store"
 	"golang.org/x/net/context"
 )
 
 // TessenController exposes actions which a viewer can perform
 type TessenController struct {
 	store store.TessenConfigStore
+	bus   messaging.MessageBus
 }
 
 // NewTessenController returns a new TessenController
-func NewTessenController(store store.TessenConfigStore) TessenController {
+func NewTessenController(store store.TessenConfigStore, bus messaging.MessageBus) TessenController {
 	return TessenController{
 		store: store,
+		bus:   bus,
 	}
 }
 
@@ -29,6 +32,11 @@ func (c TessenController) CreateOrUpdate(ctx context.Context, config *corev2.Tes
 		}
 	}
 
+	// Publish to Tessend
+	if err := c.bus.Publish(messaging.TopicTessen, config); err != nil {
+		return NewError(InternalErr, err)
+	}
+
 	return nil
 }
 

backend/apid/actions/tessen_test.go

@@ -7,6 +7,7 @@ import (
 
 	corev2 "github.com/sensu/sensu-go/api/core/v2"
 	"github.com/sensu/sensu-go/backend/store"
+	"github.com/sensu/sensu-go/testing/mockbus"
 	"github.com/sensu/sensu-go/testing/mockstore"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/mock"
@@ -16,10 +17,12 @@ func TestNewTessenController(t *testing.T) {
 	assert := assert.New(t)
 
 	store := &mockstore.MockStore{}
-	actions := NewTessenController(store)
+	bus := &mockbus.MockBus{}
+	actions := NewTessenController(store, bus)
 
 	assert.NotNil(actions)
 	assert.Equal(store, actions.store)
+	assert.Equal(bus, actions.bus)
 }
 
 func TestCreateOrUpdateTessenConfig(t *testing.T) {
@@ -28,6 +31,7 @@ func TestCreateOrUpdateTessenConfig(t *testing.T) {
 		ctx             context.Context
 		argument        *corev2.TessenConfig
 		storeErr        error
+		busErr          error
 		expectedErr     bool
 		expectedErrCode ErrCode
 	}{
@@ -52,11 +56,20 @@ func TestCreateOrUpdateTessenConfig(t *testing.T) {
 			expectedErr:     true,
 			expectedErrCode: InternalErr,
 		},
+		{
+			name:            "Bus Error",
+			ctx:             context.Background(),
+			argument:        corev2.DefaultTessenConfig(),
+			busErr:          errors.New("the bus has a flat tire"),
+			expectedErr:     true,
+			expectedErrCode: InternalErr,
+		},
 	}
 
 	for _, tc := range testCases {
 		store := &mockstore.MockStore{}
-		actions := NewTessenController(store)
+		bus := &mockbus.MockBus{}
+		actions := NewTessenController(store, bus)
 
 		t.Run(tc.name, func(t *testing.T) {
 			assert := assert.New(t)
@@ -65,6 +78,8 @@ func TestCreateOrUpdateTessenConfig(t *testing.T) {
 				On("CreateOrUpdateTessenConfig", mock.Anything, mock.Anything).
 				Return(tc.storeErr)
 
+			bus.On("Publish", mock.Anything, mock.Anything).Return(tc.busErr)
+
 			err := actions.CreateOrUpdate(tc.ctx, tc.argument)
 
 			if tc.expectedErr {
@@ -114,7 +129,8 @@ func TestGetTessenConfig(t *testing.T) {
 
 	for _, tc := range testCases {
 		store := &mockstore.MockStore{}
-		actions := NewTessenController(store)
+		bus := &mockbus.MockBus{}
+		actions := NewTessenController(store, bus)
 
 		t.Run(tc.name, func(t *testing.T) {
 			assert := assert.New(t)

backend/apid/apid.go

@@ -258,7 +258,7 @@ func registerRestrictedResources(router *mux.Router, store store.Store, getter t
 		routers.NewRolesRouter(store),
 		routers.NewRoleBindingsRouter(store),
 		routers.NewSilencedRouter(store),
-		routers.NewTessenRouter(actions.NewTessenController(store)),
+		routers.NewTessenRouter(actions.NewTessenController(store, bus)),
 		routers.NewUsersRouter(store),
 	)
 }

backend/backend.go

@@ -243,6 +243,7 @@ func Initialize(config *Config) (*Backend, error) {
 		Store:    store,
 		RingPool: ringPool,
 		Client:   b.Client,
+		Bus:      bus,
 	})
 	if err != nil {
 		return nil, fmt.Errorf("error initializing %s: %s", tessen.Name(), err)

backend/messaging/message_bus.go

@@ -22,6 +22,9 @@ const (
 
 	// TopicSubscriptions is the topic prefix for each subscription
 	TopicSubscriptions = "sensu:check"
+
+	// TopicTessen is the topic prefix for tessen api events to Tessend.
+	TopicTessen = "sensu:tessen"
 )
 
 // A Subscriber receives messages via a channel.

backend/store/etcd/entity_store.go

@@ -23,7 +23,8 @@ func getEntityPath(entity *corev2.Entity) string {
 	return entityKeyBuilder.WithResource(entity).Build(entity.Name)
 }
 
-func getEntitiesPath(ctx context.Context, name string) string {
+// GetEntitiesPath gets the path of the entity store
+func GetEntitiesPath(ctx context.Context, name string) string {
 	return entityKeyBuilder.WithContext(ctx).Build(name)
 }
 
@@ -42,7 +43,7 @@ func (s *Store) DeleteEntityByName(ctx context.Context, name string) error {
 		return errors.New("must specify name")
 	}
 
-	_, err := s.client.Delete(ctx, getEntitiesPath(ctx, name))
+	_, err := s.client.Delete(ctx, GetEntitiesPath(ctx, name))
 	return err
 }
 
@@ -52,7 +53,7 @@ func (s *Store) GetEntityByName(ctx context.Context, name string) (*corev2.Entit
 		return nil, errors.New("must specify name")
 	}
 
-	resp, err := s.client.Get(ctx, getEntitiesPath(ctx, name), clientv3.WithLimit(1))
+	resp, err := s.client.Get(ctx, GetEntitiesPath(ctx, name), clientv3.WithLimit(1))
 	if err != nil {
 		return nil, err
 	}
@@ -76,7 +77,7 @@ func (s *Store) GetEntityByName(ctx context.Context, name string) (*corev2.Entit
 // GetEntities returns the entities for the namespace in the supplied context.
 func (s *Store) GetEntities(ctx context.Context, pred *store.SelectionPredicate) ([]*corev2.Entity, error) {
 	entities := []*corev2.Entity{}
-	err := List(ctx, s.client, getEntitiesPath, &entities, pred)
+	err := List(ctx, s.client, GetEntitiesPath, &entities, pred)
 	return entities, err
 }
 

backend/store/etcd/store.go

@@ -254,6 +254,22 @@ func Update(ctx context.Context, client *clientv3.Client, key, namespace string,
 	return nil
 }
 
+// Count retrieves the count of all keys from storage under the
+// provided prefix key, while supporting all namespaces.
+func Count(ctx context.Context, client *clientv3.Client, key string) (int64, error) {
+	opts := []clientv3.OpOption{
+		clientv3.WithCountOnly(),
+		clientv3.WithRange(clientv3.GetPrefixRangeEnd(key)),
+	}
+
+	resp, err := client.Get(ctx, key, opts...)
+	if err != nil {
+		return 0, err
+	}
+
+	return resp.Count, nil
+}
+
 func getKey(key string) clientv3.Op {
 	return clientv3.OpGet(key)
 }

backend/store/etcd/store_test.go

@@ -378,3 +378,41 @@ func TestUpdate(t *testing.T) {
 		assert.Equal(t, 2, obj.Revision)
 	})
 }
+
+func TestCount(t *testing.T) {
+	testWithEtcdStore(t, func(s *Store) {
+		// Create a second namespace
+		require.NoError(t, s.CreateNamespace(context.Background(), types.FixtureNamespace("acme")))
+
+		// Create a bunch of keys everywhere
+		obj1 := &genericObject{ObjectMeta: corev2.ObjectMeta{Name: "obj1", Namespace: "default"}}
+		ctx := context.WithValue(context.Background(), types.NamespaceKey, "default")
+		require.NoError(t, Create(ctx, s.client, "/sensu.io/generic/default/obj1", "default", obj1))
+
+		obj2 := &genericObject{ObjectMeta: corev2.ObjectMeta{Name: "obj2", Namespace: "acme"}}
+		ctx = context.WithValue(context.Background(), types.NamespaceKey, "acme")
+		require.NoError(t, Create(ctx, s.client, "/sensu.io/generic/acme/obj2", "acme", obj2))
+
+		obj3 := &genericObject{ObjectMeta: corev2.ObjectMeta{Name: "obj3", Namespace: "acme"}}
+		ctx = context.WithValue(context.Background(), types.NamespaceKey, "acme")
+		require.NoError(t, Create(ctx, s.client, "/sensu.io/generic/acme/obj3", "acme", obj3))
+
+		// We should have 1 object when listing keys under the default namespace
+		ctx = context.WithValue(context.Background(), types.NamespaceKey, "default")
+		count, err := Count(ctx, s.client, getGenericObjectsPath(ctx, ""))
+		require.NoError(t, err)
+		assert.Equal(t, int64(1), count)
+
+		// We should have 2 objects when listing keys under the acme namespace
+		ctx = context.WithValue(context.Background(), types.NamespaceKey, "acme")
+		count, err = Count(ctx, s.client, getGenericObjectsPath(ctx, ""))
+		require.NoError(t, err)
+		assert.Equal(t, int64(2), count)
+
+		// We should have 3 objects when listing through all namespaces
+		ctx = context.WithValue(context.Background(), types.NamespaceKey, "")
+		count, err = Count(ctx, s.client, getGenericObjectsPath(ctx, ""))
+		require.NoError(t, err)
+		assert.Equal(t, int64(3), count)
+	})
+}