Allow configuring metrics report interval and queue depth

router/config.go

@@ -85,7 +85,11 @@ type Config struct {
 	Dialers   map[string]xgress.OptionsData
 	Listeners []listenerBinding
 	Transport map[interface{}]interface{}
-	src       map[interface{}]interface{}
+	Metrics   struct {
+		ReportInterval   time.Duration
+		MessageQueueSize int
+	}
+	src map[interface{}]interface{}
 }
 
 func (config *Config) Configure(sub config.Subconfig) error {
@@ -309,6 +313,24 @@ func LoadConfig(path string) (*Config, error) {
 		}
 	}
 
+	cfg.Metrics.ReportInterval = 15 * time.Second
+	cfg.Metrics.MessageQueueSize = 10
+	if value, found := cfgmap["metrics"]; found {
+		if submap, ok := value.(map[interface{}]interface{}); ok {
+			if value, found := submap["reportInterval"]; found {
+				if cfg.Metrics.ReportInterval, err = time.ParseDuration(value.(string)); err != nil {
+					return nil, errors.Wrap(err, "invalid value for metrics.reportInterval")
+				}
+			}
+			if value, found := submap["messageQueueSize"]; found {
+				if intVal, ok := value.(int); ok {
+					cfg.Metrics.MessageQueueSize = intVal
+				} else {
+					return nil, errors.Wrap(err, "invalid value for metrics.messageQueueSize")
+				}
+			}
+		}
+	}
 	return cfg, nil
 }
 

router/router.go

@@ -89,12 +89,7 @@ func Create(config *Config, versionProvider common.VersionProvider) *Router {
 	closeNotify := make(chan struct{})
 
 	eventDispatcher := event.NewDispatcher(closeNotify)
-	metricsConfig := &metrics.Config{
-		Source:         config.Id.Token,
-		ReportInterval: time.Second * 15,
-		EventSink:      metrics.NilHandler{},
-	}
-	metricsRegistry := metrics.NewUsageRegistryFromConfig(metricsConfig, closeNotify)
+	metricsRegistry := metrics.NewUsageRegistry(config.Id.Token, map[string]string{}, closeNotify)
 	xgress.InitMetrics(metricsRegistry)
 
 	faulter := forwarder.NewFaulter(config.Forwarder.FaultTxInterval, closeNotify)
@@ -190,8 +185,14 @@ func (self *Router) Run() error {
 }
 
 func (self *Router) showOptions() {
-	if ctrl, err := json.Marshal(self.config.Ctrl.Options); err == nil {
-		pfxlog.Logger().Infof("ctrl = %s", string(ctrl))
+	if output, err := json.Marshal(self.config.Ctrl.Options); err == nil {
+		pfxlog.Logger().Infof("ctrl = %s", string(output))
+	} else {
+		logrus.Fatalf("unable to display options (%v)", err)
+	}
+
+	if output, err := json.Marshal(self.config.Metrics); err == nil {
+		pfxlog.Logger().Infof("metrics = %s", string(output))
 	} else {
 		logrus.Fatalf("unable to display options (%v)", err)
 	}
@@ -338,7 +339,7 @@ func (self *Router) startControlPlane() error {
 	}
 
 	self.metricsReporter = metrics.NewChannelReporter(self.ctrl)
-	self.metricsRegistry.SetEventSink(self.metricsReporter)
+	self.metricsRegistry.StartReporting(self.metricsReporter, self.config.Metrics.ReportInterval, self.config.Metrics.MessageQueueSize)
 
 	return nil
 }

router/xgress/ordering_test.go

@@ -56,8 +56,8 @@ func (n noopReceiveHandler) HandleXgressReceive(payload *Payload, x *Xgress) {
 }
 
 func Test_Ordering(t *testing.T) {
-	metricsRegistry := metrics.NewUsageRegistry("test", map[string]string{}, time.Minute, noopMetricsHandler{}, nil)
 	closeNotify := make(chan struct{})
+	metricsRegistry := metrics.NewUsageRegistry("test", map[string]string{}, closeNotify)
 	InitPayloadIngester(closeNotify)
 	InitMetrics(metricsRegistry)
 	InitAcker(&noopForwarder{}, metricsRegistry, closeNotify)