Merge pull request #4714 from sensu/update-changelog

Update changelog for 6.7.1 release

Author: amdprophet

CHANGELOG-6.md

@@ -1,5 +1,3 @@
-
-
 # Changelog
 All notable changes to this project will be documented in this file.
 
@@ -9,6 +7,16 @@ Versioning](http://semver.org/spec/v2.0.0.html).
 
 ## Unreleased
 
+## [6.7.1] - 2022-04-28
+
+### Changed
+- Cron scheduler now logs that it is stopping before it begins the process of
+stopping.
+
+### Fixed
+- Fixed several data races in schedulerd.
+- Mitigate a data race in agentd sessions.
+
 ## [6.7.0] - 2022-04-21
 
 ### Added

backend/agentd/session.go

@@ -577,6 +577,8 @@ func (s *Session) stop() {
 			logger.WithError(err).Error("error closing session")
 		}
 	}()
+	defer close(s.entityConfig.updatesChannel)
+	defer close(s.checkChannel)
 
 	sessionCounter.WithLabelValues(s.cfg.Namespace).Dec()
 
@@ -593,9 +595,6 @@ func (s *Session) stop() {
 		sessionErrorCounter.WithLabelValues("GracePeriodExpired").Inc()
 	}
 
-	close(s.entityConfig.updatesChannel)
-	close(s.checkChannel)
-
 	// Remove the entity config subscriptions
 	for sub := range s.entityConfig.subscriptions {
 		if err := sub.Cancel(); err != nil {

backend/schedulerd/cron_scheduler.go

@@ -2,6 +2,7 @@ package schedulerd
 
 import (
 	"context"
+	"sync"
 
 	"github.com/sensu/sensu-go/backend/messaging"
 	"github.com/sensu/sensu-go/backend/secrets"
@@ -24,6 +25,7 @@ type CronScheduler struct {
 	interrupt              chan *corev2.CheckConfig
 	entityCache            *cachev2.Resource
 	secretsProviderManager *secrets.ProviderManager
+	stopWg                 sync.WaitGroup
 }
 
 // NewCronScheduler initializes a CronScheduler
@@ -65,10 +67,12 @@ func (s *CronScheduler) schedule(timer *CronTimer, executor *CheckExecutor) {
 // Start starts the cron scheduler.
 func (s *CronScheduler) Start() {
 	cronCounter.WithLabelValues(s.check.Namespace).Inc()
+	s.stopWg.Add(1)
 	go s.start()
 }
 
 func (s *CronScheduler) start() {
+	defer s.stopWg.Done()
 	s.logger.Info("starting new cron scheduler")
 	timer := NewCronTimer(s.check.Name, s.check.Cron)
 	executor := NewCheckExecutor(s.bus, s.check.Namespace, s.store, s.entityCache, s.secretsProviderManager)
@@ -101,9 +105,10 @@ func (s *CronScheduler) Interrupt(check *corev2.CheckConfig) {
 
 // Stop stops the cron scheduler.
 func (s *CronScheduler) Stop() error {
-	cronCounter.WithLabelValues(s.check.Namespace).Dec()
 	logger.Info("stopping cron scheduler")
 	s.cancel()
+	s.stopWg.Wait()
+	cronCounter.WithLabelValues(s.check.Namespace).Dec()
 
 	return nil
 }

backend/schedulerd/interval_scheduler.go

@@ -2,6 +2,7 @@ package schedulerd
 
 import (
 	"context"
+	"sync"
 
 	corev2 "github.com/sensu/sensu-go/api/core/v2"
 	"github.com/sensu/sensu-go/backend/messaging"
@@ -23,6 +24,7 @@ type IntervalScheduler struct {
 	interrupt              chan *corev2.CheckConfig
 	entityCache            *cachev2.Resource
 	secretsProviderManager *secrets.ProviderManager
+	stopWg                 sync.WaitGroup
 }
 
 // NewIntervalScheduler initializes an IntervalScheduler
@@ -64,10 +66,12 @@ func (s *IntervalScheduler) schedule(timer CheckTimer, executor *CheckExecutor)
 // Start starts the IntervalScheduler.
 func (s *IntervalScheduler) Start() {
 	intervalCounter.WithLabelValues(s.check.Namespace).Inc()
+	s.stopWg.Add(1)
 	go s.start()
 }
 
 func (s *IntervalScheduler) start() {
+	defer s.stopWg.Done()
 	s.logger.Info("starting new interval scheduler")
 	timer := NewIntervalTimer(s.check.Name, uint(s.check.Interval))
 	executor := NewCheckExecutor(s.bus, s.check.Namespace, s.store, s.entityCache, s.secretsProviderManager)
@@ -101,9 +105,11 @@ func (s *IntervalScheduler) Interrupt(check *corev2.CheckConfig) {
 
 // Stop stops the IntervalScheduler
 func (s *IntervalScheduler) Stop() error {
-	intervalCounter.WithLabelValues(s.check.Namespace).Dec()
 	s.logger.Info("stopping scheduler")
 	s.cancel()
+	s.stopWg.Wait()
+
+	intervalCounter.WithLabelValues(s.check.Namespace).Dec()
 
 	return nil
 }

backend/schedulerd/roundrobin_cron.go

@@ -32,6 +32,7 @@ type RoundRobinCronScheduler struct {
 	entityCache   *cachev2.Resource
 	mu            sync.Mutex
 	proxyEntities []*corev3.EntityConfig
+	stopWg        sync.WaitGroup
 }
 
 // NewRoundRobinCronScheduler creates a new RoundRobinCronScheduler.
@@ -61,6 +62,7 @@ func NewRoundRobinCronScheduler(ctx context.Context, store store.Store, bus mess
 // Start starts the scheduler.
 func (s *RoundRobinCronScheduler) Start() {
 	rrCronCounter.WithLabelValues(s.check.Namespace).Inc()
+	s.stopWg.Add(1)
 	go s.start()
 }
 
@@ -97,6 +99,7 @@ func (s *RoundRobinCronScheduler) handleEvent(executor *CheckExecutor, event rin
 }
 
 func (s *RoundRobinCronScheduler) start() {
+	defer s.stopWg.Done()
 	s.logger.Info("starting new round-robin cron scheduler")
 	s.setLastState()
 	s.updateRings()
@@ -220,9 +223,10 @@ func (s *RoundRobinCronScheduler) Interrupt(check *corev2.CheckConfig) {
 
 // Stop stops the scheduler
 func (s *RoundRobinCronScheduler) Stop() error {
-	rrCronCounter.WithLabelValues(s.check.Namespace).Dec()
 	s.logger.Info("stopping scheduler")
 	s.cancel()
+	s.stopWg.Wait()
+	rrCronCounter.WithLabelValues(s.check.Namespace).Dec()
 	return nil
 }
 

backend/schedulerd/roundrobin_interval.go

@@ -45,6 +45,7 @@ type RoundRobinIntervalScheduler struct {
 	entityCache            *cachev2.Resource
 	mu                     sync.Mutex
 	proxyEntities          []*corev3.EntityConfig
+	stopWg                 sync.WaitGroup
 }
 
 // NewRoundRobinIntervalScheduler initializes a RoundRobinIntervalScheduler
@@ -123,6 +124,7 @@ func (s *RoundRobinIntervalScheduler) updateRings() {
 // Start starts the round robin interval scheduler.
 func (s *RoundRobinIntervalScheduler) Start() {
 	rrIntervalCounter.WithLabelValues(s.check.Namespace).Inc()
+	s.stopWg.Add(1)
 	go s.start()
 }
 
@@ -174,6 +176,7 @@ func (s *RoundRobinIntervalScheduler) handleEvent(executor *CheckExecutor, event
 }
 
 func (s *RoundRobinIntervalScheduler) start() {
+	defer s.stopWg.Done()
 	s.logger.Info("starting new round-robin interval scheduler")
 	s.setLastState()
 	s.updateRings()
@@ -245,9 +248,10 @@ func (s *RoundRobinIntervalScheduler) Interrupt(check *corev2.CheckConfig) {
 
 // Stop stops the scheduler
 func (s *RoundRobinIntervalScheduler) Stop() error {
-	rrIntervalCounter.WithLabelValues(s.check.Namespace).Dec()
 	s.logger.Info("stopping scheduler")
 	s.cancel()
+	s.stopWg.Wait()
+	rrIntervalCounter.WithLabelValues(s.check.Namespace).Dec()
 	return nil
 }