diff --git a/router/config.go b/router/config.go index 3aa6adc34..702296a08 100644 --- a/router/config.go +++ b/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 } diff --git a/router/router.go b/router/router.go index 1a2b85808..7eb93aa72 100644 --- a/router/router.go +++ b/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 } diff --git a/router/xgress/ordering_test.go b/router/xgress/ordering_test.go index 1f02df7ac..253b81aee 100644 --- a/router/xgress/ordering_test.go +++ b/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)