Add message bus to tessend

CHANGELOG.md

@@ -10,6 +10,9 @@ 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.
+
 ## [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.