diff --git a/.gitignore b/.gitignore index 52d0c93..f2566bb 100644 --- a/.gitignore +++ b/.gitignore @@ -17,3 +17,7 @@ # Local copy of LIKWID headers /collectors/likwid + +# Local development workspace (build against a local cc-lib checkout) +go.work +go.work.sum diff --git a/README.md b/README.md index 3fd06d7..ed38119 100644 --- a/README.md +++ b/README.md @@ -41,6 +41,8 @@ There is a main configuration file with basic settings that point to the other c The `interval` defines how often the metrics should be read and send to the sink(s). The `duration` tells the collectors how long one measurement has to take. This is important for some collectors, like the `likwid` collector. For more information, see [here](./docs/configuration.md). +The optional `channel_buffer_size` sets the capacity of the internal channels between the components. If unset, it defaults to `max(200, 24 * number of CPUs)` so that one interval's burst of per-hwthread metrics fits without back-pressuring the collectors on nodes with many cores. + See the component READMEs for their configuration: * [`collectors`](./collectors/README.md) diff --git a/cc-metric-collector.go b/cc-metric-collector.go index 3188f9c..9865881 100644 --- a/cc-metric-collector.go +++ b/cc-metric-collector.go @@ -13,6 +13,7 @@ import ( "flag" "os" "os/signal" + "runtime" "sync" "syscall" "time" @@ -29,8 +30,9 @@ import ( ) type CentralConfigFile struct { - Interval string `json:"interval"` - Duration string `json:"duration"` + Interval string `json:"interval"` + Duration string `json:"duration"` + ChannelBufferSize int `json:"channel_buffer_size,omitempty"` } type RuntimeConfig struct { @@ -158,6 +160,14 @@ func mainFunc() int { return 1 } + // Size the channels between the managers so that one interval's burst of + // per-hwthread metrics fits without back-pressuring the collectors + chanSize := rcfg.ConfigFile.ChannelBufferSize + if chanSize <= 0 { + chanSize = max(200, 24*runtime.NumCPU()) + } + cclog.ComponentDebug("main", "channel buffer size", chanSize) + routerConf := ccconf.GetPackageConfig("router") if len(routerConf) == 0 { cclog.Error("Metric router configuration file must be set") @@ -194,7 +204,7 @@ func mainFunc() int { } // Connect metric router to sink manager - RouterToSinksChannel := make(chan lp.CCMessage, 200) + RouterToSinksChannel := make(chan lp.CCMessage, chanSize) rcfg.SinkManager.AddInput(RouterToSinksChannel) rcfg.MetricRouter.AddOutput(RouterToSinksChannel) @@ -206,7 +216,7 @@ func mainFunc() int { } // Connect collector manager to metric router - CollectToRouterChannel := make(chan lp.CCMessage, 200) + CollectToRouterChannel := make(chan lp.CCMessage, chanSize) rcfg.CollectManager.AddOutput(CollectToRouterChannel) rcfg.MetricRouter.AddCollectorInput(CollectToRouterChannel) @@ -220,7 +230,7 @@ func mainFunc() int { } // Connect receive manager to metric router - ReceiveToRouterChannel := make(chan lp.CCMessage, 200) + ReceiveToRouterChannel := make(chan lp.CCMessage, chanSize) rcfg.ReceiveManager.AddOutput(ReceiveToRouterChannel) rcfg.MetricRouter.AddReceiverInput(ReceiveToRouterChannel) use_recv = true diff --git a/collectors/collectorManager.go b/collectors/collectorManager.go index 9cce461..b6e00ce 100644 --- a/collectors/collectorManager.go +++ b/collectors/collectorManager.go @@ -12,6 +12,7 @@ import ( "encoding/json" "fmt" "sync" + "sync/atomic" "time" cclog "github.com/ClusterCockpit/cc-lib/v2/ccLogger" @@ -56,16 +57,17 @@ var AvailableCollectors = map[string]MetricCollector{ // Metric collector manager data structure type collectorManager struct { - collectors []MetricCollector // List of metric collectors to read in parallel - serial []MetricCollector // List of metric collectors to read serially - output chan lp.CCMessage // Output channels - done chan bool // channel to finish / stop metric collector manager - ticker mct.MultiChanTicker // periodically ticking once each interval - duration time.Duration // duration (for metrics that measure over a given duration) - wg *sync.WaitGroup // wait group for all goroutines in cc-metric-collector - config map[string]json.RawMessage // json encoded config for collector manager - collector_wg sync.WaitGroup // internally used wait group for the parallel reading of collector - parallel_run bool // Flag whether the collectors are currently read in parallel + collectors []MetricCollector // List of metric collectors to read in parallel + serial []MetricCollector // List of metric collectors to read serially + output chan lp.CCMessage // Output channels + done chan bool // channel to finish / stop metric collector manager + ticker mct.MultiChanTicker // periodically ticking once each interval + duration time.Duration // duration (for metrics that measure over a given duration) + wg *sync.WaitGroup // wait group for all goroutines in cc-metric-collector + config map[string]json.RawMessage // json encoded config for collector manager + collector_wg sync.WaitGroup // internally used wait group for the parallel reading of collector + round_wg sync.WaitGroup // wait group for the currently running collection round + round_running atomic.Bool // Flag whether a collection round is currently running } // Metric collector manager access functions @@ -121,26 +123,54 @@ func (cm *collectorManager) Init(ticker mct.MultiChanTicker, duration time.Durat return nil } +// runRound executes one collection round: first all parallel collectors +// concurrently, then the serial collectors one by one +func (cm *collectorManager) runRound(t time.Time) { + roundStart := time.Now() + for _, c := range cm.collectors { + // Read metrics from collector c via goroutine + cclog.ComponentDebug("CollectorManager", c.Name(), t) + cm.collector_wg.Add(1) + go func(myc MetricCollector) { + start := time.Now() + myc.Read(cm.duration, cm.output) + cclog.ComponentDebug("CollectorManager", myc.Name(), "took", time.Since(start)) + cm.collector_wg.Done() + }(c) + } + cm.collector_wg.Wait() + for _, c := range cm.serial { + // Read metrics from collector c + cclog.ComponentDebug("CollectorManager", c.Name(), t) + start := time.Now() + c.Read(cm.duration, cm.output) + cclog.ComponentDebug("CollectorManager", c.Name(), "took", time.Since(start)) + } + cclog.ComponentDebug("CollectorManager", "collection round took", time.Since(roundStart)) +} + // Start starts the metric collector manager func (cm *collectorManager) Start() { - tick := make(chan time.Time) + tick := make(chan time.Time, 1) cm.ticker.AddChannel(tick) cm.wg.Go(func() { // Collector manager is done done := func() { - // close all metric collectors - if cm.parallel_run { - cm.collector_wg.Wait() - cm.parallel_run = false - } + // wait for a still running collection round, then close all metric collectors + cm.round_wg.Wait() for _, c := range cm.collectors { c.Close() } + for _, c := range cm.serial { + c.Close() + } close(cm.done) cclog.ComponentDebug("CollectorManager", "DONE") } + var roundStart time.Time + // Wait for done signal or timer event for { select { @@ -148,37 +178,20 @@ func (cm *collectorManager) Start() { done() return case t := <-tick: - cm.parallel_run = true - for _, c := range cm.collectors { - // Wait for done signal or execute the collector - select { - case <-cm.done: - done() - return - default: - // Read metrics from collector c via goroutine - cclog.ComponentDebug("CollectorManager", c.Name(), t) - cm.collector_wg.Add(1) - go func(myc MetricCollector) { - myc.Read(cm.duration, cm.output) - cm.collector_wg.Done() - }(c) - } - } - cm.collector_wg.Wait() - cm.parallel_run = false - for _, c := range cm.serial { - // Wait for done signal or execute the collector - select { - case <-cm.done: - done() - return - default: - // Read metrics from collector c - cclog.ComponentDebug("CollectorManager", c.Name(), t) - c.Read(cm.duration, cm.output) - } + // The round runs detached from this loop, so the tick channel + // stays drained even when a round takes longer than the interval + if cm.round_running.Load() { + cclog.ComponentWarn("CollectorManager", "collection round still running after", time.Since(roundStart), "- skipping tick") + continue } + cm.round_running.Store(true) + roundStart = time.Now() + cm.round_wg.Add(1) + go func() { + defer cm.round_wg.Done() + defer cm.round_running.Store(false) + cm.runRound(t) + }() } } }) diff --git a/collectors/collectorManager_test.go b/collectors/collectorManager_test.go new file mode 100644 index 0000000..06af24d --- /dev/null +++ b/collectors/collectorManager_test.go @@ -0,0 +1,121 @@ +package collectors + +import ( + "encoding/json" + "sync" + "sync/atomic" + "testing" + "time" + + lp "github.com/ClusterCockpit/cc-lib/v2/ccMessage" +) + +// Fake ticker that delivers ticks on demand +type fakeTicker struct { + channels []chan time.Time +} + +func (t *fakeTicker) Init(duration time.Duration) {} + +func (t *fakeTicker) AddChannel(c chan time.Time) { + t.channels = append(t.channels, c) +} + +func (t *fakeTicker) Close() {} + +func (t *fakeTicker) tick() { + for _, c := range t.channels { + select { + case c <- time.Now(): + default: + } + } +} + +// Stub collector whose Read blocks until it is released +type stubCollector struct { + metricCollector + readStarted chan struct{} + release chan struct{} + reads atomic.Int32 +} + +func (c *stubCollector) Init(config json.RawMessage) error { + c.name = "teststub" + c.parallel = true + c.init = true + return nil +} + +func (c *stubCollector) Read(duration time.Duration, output chan lp.CCMessage) { + c.reads.Add(1) + c.readStarted <- struct{}{} + <-c.release +} + +func (c *stubCollector) Close() {} + +func TestOverlongCollectionRoundSkipsTick(t *testing.T) { + stub := &stubCollector{ + readStarted: make(chan struct{}, 10), + release: make(chan struct{}), + } + AvailableCollectors["teststub"] = stub + defer delete(AvailableCollectors, "teststub") + + ticker := &fakeTicker{} + var wg sync.WaitGroup + cm, err := New(ticker, time.Second, &wg, json.RawMessage(`{"teststub": {}}`)) + if err != nil { + t.Fatalf("failed to setup collector manager: %s", err.Error()) + } + cm.AddOutput(make(chan lp.CCMessage, 100)) + cm.Start() + + // First tick starts a collection round that blocks in Read + ticker.tick() + select { + case <-stub.readStarted: + case <-time.After(5 * time.Second): + t.Fatal("collection round did not start on tick") + } + + // Further ticks while the round is still running must be skipped, + // not queued up or run concurrently + for range 3 { + ticker.tick() + time.Sleep(20 * time.Millisecond) + } + if got := stub.reads.Load(); got != 1 { + t.Fatalf("expected 1 concurrent collection round, got %d reads", got) + } + + // Finish the round, the next tick must start a new one + stub.release <- struct{}{} + deadline := time.After(5 * time.Second) + for stub.reads.Load() < 2 { + ticker.tick() + select { + case <-stub.readStarted: + case <-time.After(20 * time.Millisecond): + case <-deadline: + t.Fatal("no new collection round started after the previous one finished") + } + } + + // Shutdown must wait for the running round and terminate cleanly. + // Closing the release channel lets any still running or straggler + // round finish immediately + close(stub.release) + closed := make(chan struct{}) + go func() { + cm.Close() + close(closed) + }() + select { + case <-closed: + case <-time.After(5 * time.Second): + t.Fatal("Close() did not terminate") + } + wg.Wait() +} diff --git a/docs/configuration.md b/docs/configuration.md index 88fd669..7265b42 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -19,6 +19,8 @@ The global file contains the paths to the other four files and some global optio Be aware that the paths are relative to the execution folder of the cc-metric-collector binary, so it is recommended to use absolute paths. +The optional `channel_buffer_size` option sets the capacity of the internal channels between the components (collectors → router → sinks). If unset, it defaults to `max(200, 24 * number of CPUs)` so that one interval's burst of per-hwthread metrics fits without back-pressuring the collectors on nodes with many cores. + ## Component configuration The others are mainly list of of subcomponents: the collectors, the receivers, the router and the sinks. Their role is best shown in a picture: diff --git a/internal/metricRouter/metricCache.go b/internal/metricRouter/metricCache.go index 9f22bdf..fdf0998 100644 --- a/internal/metricRouter/metricCache.go +++ b/internal/metricRouter/metricCache.go @@ -79,7 +79,7 @@ func (c *metricCache) Init(output chan lp.CCMessage, ticker mct.MultiChanTicker, // Start starts the metric cache func (c *metricCache) Start() { - c.tickchan = make(chan time.Time) + c.tickchan = make(chan time.Time, 1) c.ticker.AddChannel(c.tickchan) // Router cache is done done := func() { diff --git a/internal/metricRouter/metricRouter.go b/internal/metricRouter/metricRouter.go index 718b6aa..4c6475a 100644 --- a/internal/metricRouter/metricRouter.go +++ b/internal/metricRouter/metricRouter.go @@ -228,11 +228,25 @@ func (r *metricRouter) DoAddTags(point lp.CCMessage) { func (r *metricRouter) Start() { // start timer if configured r.timestamp = time.Now() - timeChan := make(chan time.Time) + timeChan := make(chan time.Time, 1) if r.config.IntervalStamp { r.ticker.AddChannel(timeChan) } + // Drain a pending tick before stamping new metrics, so a new interval's + // metrics never carry the previous interval's timestamp + updateTimestamp := func() { + if !r.config.IntervalStamp { + return + } + select { + case timestamp := <-timeChan: + r.timestamp = timestamp + cclog.ComponentDebug("MetricRouter", "Update timestamp", r.timestamp.UnixNano()) + default: + } + } + // Router manager is done done := func() { close(r.done) @@ -298,14 +312,19 @@ func (r *metricRouter) Start() { case timestamp := <-timeChan: r.timestamp = timestamp cclog.ComponentDebug("MetricRouter", "Update timestamp", r.timestamp.UnixNano()) + if len(r.coll_input) == cap(r.coll_input) { + cclog.ComponentWarn("MetricRouter", "collector input channel full at tick, sinks may be too slow") + } case p := <-r.coll_input: + updateTimestamp() coll_forward(p) for i := 0; len(r.coll_input) > 0 && i < (r.maxForward-1); i++ { coll_forward(<-r.coll_input) } case p := <-r.recv_input: + updateTimestamp() recv_forward(p) for i := 0; len(r.recv_input) > 0 && i < (r.maxForward-1); i++ { recv_forward(<-r.recv_input) diff --git a/internal/metricRouter/metricRouter_test.go b/internal/metricRouter/metricRouter_test.go new file mode 100644 index 0000000..79c61aa --- /dev/null +++ b/internal/metricRouter/metricRouter_test.go @@ -0,0 +1,99 @@ +package metricRouter + +import ( + "encoding/json" + "fmt" + "sync" + "testing" + "time" + + lp "github.com/ClusterCockpit/cc-lib/v2/ccMessage" +) + +// Fake ticker that delivers ticks on demand +type fakeTicker struct { + channels []chan time.Time +} + +func (t *fakeTicker) Init(duration time.Duration) {} + +func (t *fakeTicker) AddChannel(c chan time.Time) { + t.channels = append(t.channels, c) +} + +func (t *fakeTicker) Close() {} + +func (t *fakeTicker) tick(ts time.Time) { + for _, c := range t.channels { + c <- ts + } +} + +func genMessages(t *testing.T, num int) []lp.CCMessage { + t.Helper() + msgs := make([]lp.CCMessage, 0, num) + tags := map[string]string{"type": "node"} + for i := range num { + m, err := lp.NewMetric(fmt.Sprintf("testmetric%d", i), tags, nil, 42.0, time.Unix(1, 0)) + if err != nil { + t.Fatalf("failed to create message: %s", err.Error()) + } + msgs = append(msgs, m) + } + return msgs +} + +// With interval_timestamp enabled, all metrics forwarded after a tick must +// carry that tick's timestamp, never the previous interval's +func TestIntervalTimestamp(t *testing.T) { + ticker := &fakeTicker{} + var wg sync.WaitGroup + r, err := New(ticker, &wg, json.RawMessage(`{"interval_timestamp": true}`)) + if err != nil { + t.Fatalf("failed to setup metric router: %s", err.Error()) + } + + coll := make(chan lp.CCMessage, 100) + out := make(chan lp.CCMessage, 100) + r.AddCollectorInput(coll) + r.AddOutput(out) + r.Start() + + receiveAll := func(num int) []lp.CCMessage { + received := make([]lp.CCMessage, 0, num) + for len(received) < num { + select { + case m := <-out: + received = append(received, m) + case <-time.After(5 * time.Second): + t.Fatalf("received only %d of %d messages", len(received), num) + } + } + return received + } + + for interval, tickTime := range []time.Time{time.Unix(1000, 0), time.Unix(1010, 0)} { + ticker.tick(tickTime) + msgs := genMessages(t, 20) + for _, m := range msgs { + coll <- m + } + for i, m := range receiveAll(len(msgs)) { + if !m.Time().Equal(tickTime) { + t.Errorf("interval %d message %d: got timestamp %v, want %v", interval, i, m.Time(), tickTime) + } + } + } + + closed := make(chan struct{}) + go func() { + r.Close() + close(closed) + }() + select { + case <-closed: + case <-time.After(5 * time.Second): + t.Fatal("Close() did not terminate") + } + wg.Wait() +} diff --git a/pkg/multiChanTicker/README.md b/pkg/multiChanTicker/README.md index 782c240..e39e9fd 100644 --- a/pkg/multiChanTicker/README.md +++ b/pkg/multiChanTicker/README.md @@ -30,8 +30,8 @@ Afterwards, you can add channels: ```golang t := MultiChanTicker(duration) -c1 := make(chan time.Time) -c2 := make(chan time.Time) +c1 := make(chan time.Time, 1) +c2 := make(chan time.Time, 1) t.AddChannel(c1) t.AddChannel(c2) @@ -46,3 +46,5 @@ for { ``` The result should be the same `time.Time` output in both channels, notified "simultaneously". + +Ticks are delivered with a non-blocking send: a consumer that has not yet read the previous tick does not stall the ticker (which would silently drop `time.Ticker` fires for all consumers); instead, the tick for that consumer is skipped and a warning is logged. Register buffered channels (capacity 1) so a consumer that is briefly busy at tick time does not lose the tick. diff --git a/pkg/multiChanTicker/multiChanTicker.go b/pkg/multiChanTicker/multiChanTicker.go index 86f3376..bad75b8 100644 --- a/pkg/multiChanTicker/multiChanTicker.go +++ b/pkg/multiChanTicker/multiChanTicker.go @@ -8,6 +8,8 @@ package multiChanTicker import ( + "fmt" + "sync" "time" cclog "github.com/ClusterCockpit/cc-lib/v2/ccLogger" @@ -15,6 +17,7 @@ import ( type multiChanTicker struct { ticker *time.Ticker + mutex sync.Mutex // protects channels, which is appended to while the tick goroutine iterates it channels []chan time.Time done chan bool } @@ -40,21 +43,30 @@ func (t *multiChanTicker) Init(duration time.Duration) { return case ts := <-t.ticker.C: cclog.ComponentDebug("MultiChanTicker", "Tick", ts) - for _, c := range t.channels { + t.mutex.Lock() + for i, c := range t.channels { + // Non-blocking send: a consumer that has not yet read the + // previous tick must not stall the ticker, otherwise + // time.Ticker silently drops fires for ALL consumers select { - case <-t.done: - done() - return case c <- ts: + default: + cclog.ComponentWarn("MultiChanTicker", fmt.Sprintf("consumer %d did not read previous tick, dropping tick %v", i, ts)) } } + t.mutex.Unlock() } } }() } func (t *multiChanTicker) AddChannel(channel chan time.Time) { + if cap(channel) == 0 { + cclog.ComponentWarn("MultiChanTicker", "unbuffered channel registered, ticks may be dropped if the consumer is not ready") + } + t.mutex.Lock() t.channels = append(t.channels, channel) + t.mutex.Unlock() } func (t *multiChanTicker) Close() { diff --git a/pkg/multiChanTicker/multiChanTicker_test.go b/pkg/multiChanTicker/multiChanTicker_test.go new file mode 100644 index 0000000..1a2da5b --- /dev/null +++ b/pkg/multiChanTicker/multiChanTicker_test.go @@ -0,0 +1,51 @@ +package multiChanTicker + +import ( + "testing" + "time" +) + +// A consumer that never reads its channel must not stall the ticker +// or starve the other consumers +func TestStalledConsumerDoesNotStarveOthers(t *testing.T) { + stalled := make(chan time.Time, 1) // never read + fast := make(chan time.Time, 1) + + ticker := NewTicker(10 * time.Millisecond) + defer ticker.Close() + ticker.AddChannel(stalled) + ticker.AddChannel(fast) + + received := 0 + deadline := time.After(5 * time.Second) + for received < 5 { + select { + case <-fast: + received++ + case <-deadline: + t.Fatalf("received only %d ticks while another consumer stalled", received) + } + } +} + +// Close() must return promptly even if no consumer reads its channel +func TestCloseWithStalledConsumer(t *testing.T) { + stalled := make(chan time.Time, 1) // never read + + ticker := NewTicker(10 * time.Millisecond) + ticker.AddChannel(stalled) + + // Let some ticks fire and be dropped + time.Sleep(50 * time.Millisecond) + + closed := make(chan struct{}) + go func() { + ticker.Close() + close(closed) + }() + select { + case <-closed: + case <-time.After(5 * time.Second): + t.Fatal("Close() blocked with a stalled consumer") + } +}