Skip to content

Commit 3123b2a

Browse files
muicodermuicoder
committed
Adding association between a Kafka cluster and a notifier
Signed-off-by: muicoder <[email protected]> linkedin#611
1 parent e593345 commit 3123b2a

File tree

9 files changed

+70
-23
lines changed

9 files changed

+70
-23
lines changed

‎config/burrow.toml‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -58,6 +58,7 @@ min-distance=1
5858

5959
[notifier.default]
6060
class-name="http"
61+
cluster="local"
6162
url-open="http://someservice.example.com:1467/v1/event"
6263
interval=60
6364
timeout=5

‎core/internal/helpers/coordinators.go‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -74,6 +74,12 @@ func (m *MockModule) GetName() string {
7474
return args.String(0)
7575
}
7676

77+
// GetCluster mocks the notifier.Module GetCluster func
78+
func (m *MockModule) GetCluster() string {
79+
args := m.Called()
80+
return args.String(0)
81+
}
82+
7783
// GetGroupAllowlist mocks the notifier.Module GetGroupAllowlist func
7884
func (m *MockModule) GetGroupAllowlist() *regexp.Regexp {
7985
args := m.Called()

‎core/internal/httpserver/config.go‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -213,6 +213,7 @@ func (hc *Coordinator) configNotifierHTTP(w http.ResponseWriter, r *http.Request
213213
SendClose: viper.GetBool(configRoot + ".send-close"),
214214
ExtraCa: viper.GetString(configRoot + ".extra-ca"),
215215
NoVerify: viper.GetString(configRoot + ".noverify"),
216+
Cluster: viper.GetString(configRoot + ".cluster"),
216217
},
217218
Request: requestInfo,
218219
})
@@ -265,6 +266,7 @@ func (hc *Coordinator) configNotifierEmail(w http.ResponseWriter, r *http.Reques
265266
To: viper.GetString(configRoot + ".to"),
266267
ExtraCa: viper.GetString(configRoot + ".extra-ca"),
267268
NoVerify: viper.GetString(configRoot + ".noverify"),
269+
Cluster: viper.GetString(configRoot + ".cluster"),
268270
},
269271
Request: requestInfo,
270272
})

‎core/internal/httpserver/structs.go‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -202,6 +202,7 @@ type httpResponseConfigModuleNotifierHTTP struct {
202202
SendClose bool `json:"send-close"`
203203
ExtraCa string `json:"extra-ca"`
204204
NoVerify string `json:"noverify"`
205+
Cluster string `json:"cluster"`
205206
}
206207

207208
type httpResponseConfigModuleNotifierSlack struct {
@@ -238,6 +239,7 @@ type httpResponseConfigModuleNotifierEmail struct {
238239
To string `json:"to"`
239240
ExtraCa string `json:"extra-ca"`
240241
NoVerify string `json:"noverify"`
242+
Cluster string `json:"cluster"`
241243
}
242244

243245
type httpResponseConfigModuleNotifierNull struct {

‎core/internal/notifier/coordinator.go‎

Lines changed: 11 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -48,6 +48,7 @@ import (
4848
type Module interface {
4949
protocol.Module
5050
GetName() string
51+
GetCluster() string
5152
GetGroupAllowlist() *regexp.Regexp
5253
GetGroupDenylist() *regexp.Regexp
5354
GetLogger() *zap.Logger
@@ -95,7 +96,7 @@ type Coordinator struct {
9596

9697
// getModuleForClass returns the correct module based on the passed className. As part of the Configure steps, if there
9798
// is any error, it will panic with an appropriate message describing the problem.
98-
func getModuleForClass(app *protocol.ApplicationContext, moduleName, className string, groupAllowlist, groupDenylist *regexp.Regexp, extras map[string]string, templateOpen, templateClose *template.Template) protocol.Module {
99+
func getModuleForClass(app *protocol.ApplicationContext, moduleName, className string, groupAllowlist, groupDenylist *regexp.Regexp, extras map[string]string, templateOpen, templateClose *template.Template, cluster string) protocol.Module {
99100
logger := app.Logger.With(
100101
zap.String("type", "module"),
101102
zap.String("coordinator", "notifier"),
@@ -113,6 +114,7 @@ func getModuleForClass(app *protocol.ApplicationContext, moduleName, className s
113114
extras: extras,
114115
templateOpen: templateOpen,
115116
templateClose: templateClose,
117+
cluster: cluster,
116118
}
117119
case "email":
118120
return &EmailNotifier{
@@ -123,6 +125,7 @@ func getModuleForClass(app *protocol.ApplicationContext, moduleName, className s
123125
extras: extras,
124126
templateOpen: templateOpen,
125127
templateClose: templateClose,
128+
cluster: cluster,
126129
}
127130
case "null":
128131
return &NullNotifier{
@@ -133,6 +136,7 @@ func getModuleForClass(app *protocol.ApplicationContext, moduleName, className s
133136
extras: extras,
134137
templateOpen: templateOpen,
135138
templateClose: templateClose,
139+
cluster: cluster,
136140
}
137141
default:
138142
panic("Unknown notifier className provided: " + className)
@@ -194,6 +198,8 @@ func (nc *Coordinator) Configure() {
194198
groupAllowlist = re
195199
}
196200

201+
cluster := viper.GetString(configRoot + ".cluster")
202+
197203
// Compile the denylist for the consumer groups to not notify for
198204
var groupDenylist *regexp.Regexp
199205
denylist := viper.GetString(configRoot + ".group-denylist")
@@ -227,7 +233,7 @@ func (nc *Coordinator) Configure() {
227233
templateClose = tmpl.Templates()[0]
228234
}
229235

230-
module := getModuleForClass(nc.App, name, viper.GetString(configRoot+".class-name"), groupAllowlist, groupDenylist, extras, templateOpen, templateClose)
236+
module := getModuleForClass(nc.App, name, viper.GetString(configRoot+".class-name"), groupAllowlist, groupDenylist, extras, templateOpen, templateClose, cluster)
231237
module.Configure(name, configRoot)
232238
nc.modules[name] = module
233239
interval := viper.GetInt64(configRoot + ".interval")
@@ -436,6 +442,9 @@ func (nc *Coordinator) checkAndSendResponseToModules(response *protocol.Consumer
436442
for _, genericModule := range nc.modules {
437443
module := genericModule.(Module)
438444

445+
if module.GetCluster() != "" && response.Cluster != module.GetCluster() {
446+
continue
447+
}
439448
// No allowlist means everything passes
440449
groupAllowlist := module.GetGroupAllowlist()
441450
groupDenylist := module.GetGroupDenylist()

‎core/internal/notifier/coordinator_test.go‎

Lines changed: 31 additions & 21 deletions
Original file line numberDiff line numberDiff line change
@@ -472,34 +472,39 @@ var notifyModuleTests = []struct {
472472
ExpectClose bool
473473
ExpectID bool
474474
SendOnce bool
475+
Cluster string
475476
}{
476477
// {1, 0, false, false, false, false, false},
477478
// {2, 0, false, false, false, false, false},
478479
// {1, 0, true, false, false, false, false},
479480
// {1, 0, false, true, false, false, false},
480481
// {1, 0, true, true, false, false, false},
481482

482-
{1, 1, false, false, true, false, false, false},
483-
{1, 1, false, true, true, false, false, false},
484-
{1, 1, true, false, true, false, false, false},
485-
{1, 1, true, true, true, true, false, true},
486-
487-
{1, 2, false, false, true, false, true, false},
488-
{1, 2, false, true, true, false, true, false},
489-
{1, 2, true, false, true, false, true, false},
490-
{1, 2, true, true, true, false, true, false},
491-
{1, 2, true, true, false, false, true, true},
492-
{1, 2, false, true, true, false, true, true},
493-
494-
{3, 2, false, false, false, false, true, false},
495-
{3, 2, false, true, false, false, true, false},
496-
{3, 2, true, false, false, false, true, false},
497-
{3, 2, true, true, false, false, true, false},
498-
499-
{2, 1, false, false, false, false, false, false},
500-
{2, 1, false, true, false, false, false, false},
501-
{2, 1, true, false, false, false, false, false},
502-
{2, 1, true, true, true, true, false, false},
483+
{1, 1, false, false, true, false, false, false, ""},
484+
{1, 1, false, true, true, false, false, false, "testcluster"},
485+
{1, 1, true, false, true, false, false, false, "unmatchedCluster"},
486+
{1, 1, true, true, true, true, false, true, ""},
487+
488+
{1, 2, false, false, true, false, true, false, ""},
489+
{1, 2, false, true, true, false, true, false, ""},
490+
{1, 2, true, false, true, false, true, false, ""},
491+
{1, 2, true, true, true, false, true, false, ""},
492+
{1, 2, true, true, false, false, true, true, ""},
493+
{1, 2, false, true, true, false, true, true, ""},
494+
495+
{3, 2, false, false, false, false, true, false, ""},
496+
{3, 2, false, true, false, false, true, false, ""},
497+
{3, 2, true, false, false, false, true, false, ""},
498+
{3, 2, true, true, false, false, true, false, ""},
499+
500+
{2, 1, false, false, false, false, false, false, ""},
501+
{2, 1, false, true, false, false, false, false, ""},
502+
{2, 1, true, false, false, false, false, false, ""},
503+
{2, 1, true, true, true, true, false, false, ""},
504+
}
505+
506+
func checkNotifierClusterMatch(cluster string) bool {
507+
return cluster == "" || cluster == "testcluster"
503508
}
504509

505510
func TestCoordinator_checkAndSendResponseToModules(t *testing.T) {
@@ -558,10 +563,15 @@ func TestCoordinator_checkAndSendResponseToModules(t *testing.T) {
558563
// Set up the mock module and expected calls
559564
mockModule := &helpers.MockModule{}
560565
coordinator.modules["test"] = mockModule
566+
mockModule.On("GetCluster").Return(testSet.Cluster)
567+
568+
if checkNotifierClusterMatch(testSet.Cluster) {
561569
mockModule.On("GetName").Return("test")
562570
mockModule.On("GetGroupAllowlist").Return((*regexp.Regexp)(nil))
563571
mockModule.On("GetGroupDenylist").Return((*regexp.Regexp)(nil))
564572
mockModule.On("AcceptConsumerGroup", response).Return(true)
573+
}
574+
565575
if testSet.ExpectSend {
566576
mockModule.On("Notify", response, mock.MatchedBy(func(s string) bool { return true }), mock.MatchedBy(func(t time.Time) bool { return true }), testSet.ExpectClose).Return()
567577
}

‎core/internal/notifier/email.go‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -38,6 +38,7 @@ type EmailNotifier struct {
3838
Log *zap.Logger
3939

4040
name string
41+
cluster string
4142
groupAllowlist *regexp.Regexp
4243
groupDenylist *regexp.Regexp
4344
extras map[string]string
@@ -139,6 +140,11 @@ func (module *EmailNotifier) GetName() string {
139140
return module.name
140141
}
141142

143+
// GetCluster returns the configured name of this module
144+
func (module *EmailNotifier) GetCluster() string {
145+
return module.cluster
146+
}
147+
142148
// GetGroupAllowlist returns the compiled group allowlist (or nil, if there is not one)
143149
func (module *EmailNotifier) GetGroupAllowlist() *regexp.Regexp {
144150
return module.groupAllowlist

‎core/internal/notifier/http.go‎

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -39,6 +39,7 @@ type HTTPNotifier struct {
3939
Log *zap.Logger
4040

4141
name string
42+
cluster string
4243
groupAllowlist *regexp.Regexp
4344
groupDenylist *regexp.Regexp
4445
extras map[string]string
@@ -124,6 +125,10 @@ func (module *HTTPNotifier) GetName() string {
124125
return module.name
125126
}
126127

128+
func (module *HTTPNotifier) GetCluster() string {
129+
return module.cluster
130+
}
131+
127132
// GetGroupAllowlist returns the compiled group allowlist (or nil, if there is not one)
128133
func (module *HTTPNotifier) GetGroupAllowlist() *regexp.Regexp {
129134
return module.groupAllowlist

‎core/internal/notifier/null.go‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -30,6 +30,7 @@ type NullNotifier struct {
3030
Log *zap.Logger
3131

3232
name string
33+
cluster string
3334
groupAllowlist *regexp.Regexp
3435
groupDenylist *regexp.Regexp
3536
extras map[string]string
@@ -75,6 +76,11 @@ func (module *NullNotifier) GetName() string {
7576
return module.name
7677
}
7778

79+
// GetCluster returns the configured name of this module
80+
func (module *NullNotifier) GetCluster() string {
81+
return module.cluster
82+
}
83+
7884
// GetGroupAllowlist returns the compiled group allowlist (or nil, if there is not one)
7985
func (module *NullNotifier) GetGroupAllowlist() *regexp.Regexp {
8086
return module.groupAllowlist

0 commit comments

Comments
 (0)