Merge pull request #27 from lovoo/metrics-option

option to provide metrics registry

options.go

@@ -183,6 +183,15 @@ func WithLogger(log logger.Logger) ProcessorOption {
 	}
 }
 
+// WithRegistry sets the metrics registry the processor should use. The registry
+// should not be prefixed, otherwise the monitor web view won't work as it
+// expects metrics of certain name.
+func WithRegistry(r metrics.Registry) ProcessorOption {
+	return func(o *poptions) {
+		o.registry = r
+	}
+}
+
 func (opt *poptions) applyOptions(group string, opts ...ProcessorOption) error {
 	opt.clientID = defaultClientID
 
@@ -207,8 +216,6 @@ func (opt *poptions) applyOptions(group string, opts ...ProcessorOption) error {
 		opt.storageSnapshotInterval = storage.DefaultStorageSnapshotInterval
 	}
 
-	opt.registry = metrics.NewRegistry()
-
 	// prefix registry
 	opt.gokaRegistry = metrics.NewPrefixedChildRegistry(opt.registry, fmt.Sprintf("goka.processor-%s.", group))
 
@@ -253,6 +260,15 @@ type voptions struct {
 	kafkaRegistry metrics.Registry
 }
 
+// WithViewRegistry sets the metrics registry the processor should use. The
+// registry should not be prefixed, otherwise the monitor web view won't work as
+// it expects metrics of certain name.
+func WithViewRegistry(r metrics.Registry) ViewOption {
+	return func(o *voptions) {
+		o.registry = r
+	}
+}
+
 // WithViewLogger sets the logger the view should use. By default, views
 // use the standard library logger.
 func WithViewLogger(log logger.Logger) ViewOption {
@@ -350,8 +366,6 @@ func (opt *voptions) applyOptions(topic Table, opts ...ViewOption) error {
 		opt.builders.topicmgr = defaultTopicManagerBuilder
 	}
 
-	opt.registry = metrics.NewRegistry()
-
 	// Set a default registry to pass it to the kafka consumer builders.
 	if opt.kafkaRegistry == nil {
 		opt.kafkaRegistry = metrics.NewPrefixedChildRegistry(opt.registry, fmt.Sprintf("kafka.view-%s.", topic))

processor.go

@@ -58,6 +58,7 @@ func NewProcessor(brokers []string, gg *GroupGraph, options ...ProcessorOption)
 	options = append(
 		// default options comes first
 		[]ProcessorOption{
+			WithRegistry(metrics.NewRegistry()),
 			WithLogger(logger.Default()),
 			WithUpdateCallback(DefaultUpdate),
 			WithStoragePath("/tmp/goka"),

view.go

@@ -39,6 +39,7 @@ func NewView(brokers []string, topic Table, codec Codec, options ...ViewOption)
 	options = append(
 		// default options comes first
 		[]ViewOption{
+			WithViewRegistry(metrics.NewRegistry()),
 			WithViewLogger(logger.Default()),
 			WithViewCallback(DefaultUpdate),
 			WithViewStoragePath(defaultReaderStoragePath),