On week improvements (#4156)

Add some comments/simplify the fleet-server startup process

Author: michel-laterman

internal/pkg/checkin/bulk.go

@@ -18,9 +18,10 @@ import (
 	"github.com/elastic/fleet-server/v7/internal/pkg/dl"
 	"github.com/elastic/fleet-server/v7/internal/pkg/sqn"
 
+	"github.com/rs/zerolog"
+
 	estypes "github.com/elastic/go-elasticsearch/v8/typedapi/types"
 	"github.com/elastic/go-elasticsearch/v8/typedapi/types/enums/scriptlanguage"
-	"github.com/rs/zerolog"
 )
 
 const defaultFlushInterval = 10 * time.Second
@@ -142,26 +143,20 @@ func (bc *Bulk) CheckIn(id string, status string, message string, meta []byte, c
 
 // Run starts the flush timer and exit only when the context is cancelled.
 func (bc *Bulk) Run(ctx context.Context) error {
-
 	tick := time.NewTicker(bc.opts.flushInterval)
 	defer tick.Stop()
 
-	var err error
-LOOP:
 	for {
 		select {
 		case <-tick.C:
-			if err = bc.flush(ctx); err != nil {
+			if err := bc.flush(ctx); err != nil {
 				zerolog.Ctx(ctx).Error().Err(err).Msg("Eat bulk checkin error; Keep on truckin'")
 			}
 
 		case <-ctx.Done():
-			err = ctx.Err()
-			break LOOP
+			return ctx.Err()
 		}
 	}
-
-	return err
 }
 
 // flush sends the minium data needed to update records in elasticsearch.

internal/pkg/server/agent.go

@@ -11,22 +11,23 @@ import (
 	"io"
 	"strconv"
 	"strings"
-	"sync"
 	"time"
 
+	"github.com/rs/zerolog"
+
 	"github.com/elastic/fleet-server/v7/internal/pkg/build"
 	"github.com/elastic/fleet-server/v7/internal/pkg/config"
 	"github.com/elastic/fleet-server/v7/internal/pkg/es"
 	"github.com/elastic/fleet-server/v7/internal/pkg/reload"
 	"github.com/elastic/fleet-server/v7/internal/pkg/sleep"
 	"github.com/elastic/fleet-server/v7/internal/pkg/state"
 	"github.com/elastic/fleet-server/v7/internal/pkg/ver"
-	"github.com/rs/zerolog"
+
+	"gopkg.in/yaml.v3"
 
 	"github.com/elastic/elastic-agent-client/v7/pkg/client"
 	"github.com/elastic/elastic-agent-client/v7/pkg/proto"
 	"github.com/elastic/go-ucfg"
-	"gopkg.in/yaml.v3"
 )
 
 const (
@@ -65,7 +66,7 @@ type Agent struct {
 	srv          *Fleet
 	srvCtx       context.Context
 	srvCanceller context.CancelFunc
-	srvDone      chan bool
+	srvDone      chan struct{}
 
 	outputCheckCanceller context.CancelFunc
 	chReconfigure        chan struct{}
@@ -97,6 +98,7 @@ func NewAgent(cliCfg *ucfg.Config, reader io.Reader, bi build.Info, reloadables
 
 // Run starts a Server instance using config from the configured client.
 func (a *Agent) Run(ctx context.Context) error {
+	// ctx is cancelled when a SIGTERM or SIGINT is received.
 	log := zerolog.Ctx(ctx)
 	a.agent.RegisterDiagnosticHook("fleet-server config", "fleet-server's current configuration", "fleet-server.yml", "application/yml", func() []byte {
 		if a.srv == nil {
@@ -150,24 +152,21 @@ func (a *Agent) Run(ctx context.Context) error {
 			log.Warn().Msg("Diagnostics hook failure config is nil.")
 			return []byte(`Diagnostics hook failure config is nil`)
 		}
-		ctx, cancel := context.WithTimeout(ctx, time.Second*30) // TODO(michel-laterman): duration/timeout should be part of the diagnostics action from fleet-server (https://github.com/elastic/fleet-server/issues/3648) and the control protocol (https://github.com/elastic/elastic-agent-client/issues/113)
+		ctx, cancel := context.WithTimeout(ctx, time.Second*30) // diag specific context, has a timeout  // TODO(michel-laterman): duration/timeout should be part of the diagnostics action from fleet-server (https://github.com/elastic/fleet-server/issues/3648) and the control protocol (https://github.com/elastic/elastic-agent-client/issues/113)
 		defer cancel()
 		return cfg.Output.Elasticsearch.DiagRequests(ctx)
 	})
 
-	subCtx, subCanceller := context.WithCancel(ctx)
-	defer subCanceller()
-
-	var wg sync.WaitGroup
-	wg.Add(1)
+	// doneCh is used to track when agent wrapper run loop returns
+	doneCh := make(chan struct{})
 	go func() {
-		defer wg.Done()
+		defer close(doneCh)
 
 		t := time.NewTicker(1 * time.Second)
 		defer t.Stop()
 		for {
 			select {
-			case <-subCtx.Done():
+			case <-ctx.Done():
 				return
 			case err := <-a.agent.Errors():
 				if err != nil && !errors.Is(err, context.Canceled) && !errors.Is(err, io.EOF) {
@@ -176,13 +175,13 @@ func (a *Agent) Run(ctx context.Context) error {
 			case change := <-a.agent.UnitChanges():
 				switch change.Type {
 				case client.UnitChangedAdded:
-					err := a.unitAdded(subCtx, change.Unit)
+					err := a.unitAdded(ctx, change.Unit)
 					if err != nil {
 						log.Error().Str("unit", change.Unit.ID()).Err(err)
 						_ = change.Unit.UpdateState(client.UnitStateFailed, err.Error(), nil)
 					}
 				case client.UnitChangedModified:
-					err := a.unitModified(subCtx, change.Unit)
+					err := a.unitModified(ctx, change.Unit)
 					if err != nil {
 						log.Error().Str("unit", change.Unit.ID()).Err(err)
 						_ = change.Unit.UpdateState(client.UnitStateFailed, err.Error(), nil)
@@ -202,7 +201,7 @@ func (a *Agent) Run(ctx context.Context) error {
 				if agentInfo != nil && agentInfo.ID != "" {
 					// Agent ID is not set for the component.
 					t.Stop()
-					err := a.reconfigure(subCtx)
+					err := a.reconfigure(ctx)
 					if err != nil && !errors.Is(err, context.Canceled) {
 						log.Error().Err(err).Msg("Bootstrap error when reconfiguring")
 					}
@@ -212,13 +211,13 @@ func (a *Agent) Run(ctx context.Context) error {
 	}()
 
 	log.Info().Msg("starting communication connection back to Elastic Agent")
-	err := a.agent.Start(subCtx)
-	if err != nil {
+	err := a.agent.Start(ctx)
+	if err != nil && !errors.Is(err, context.Canceled) {
 		return err
 	}
 
-	<-subCtx.Done()
-	wg.Wait()
+	<-ctx.Done() // wait for a termination signal
+	<-doneCh     // wait for agent wrapper goroutine to terminate
 
 	return nil
 }
@@ -355,7 +354,7 @@ func (a *Agent) start(ctx context.Context) error {
 		}
 	}
 
-	srvDone := make(chan bool)
+	srvDone := make(chan struct{})
 	srvCtx, srvCanceller := context.WithCancel(ctx)
 	srv, err := NewFleet(a.bi, state.NewChained(state.NewLog(), a), false)
 	if err != nil {
@@ -394,6 +393,7 @@ func (a *Agent) reconfigure(ctx context.Context) error {
 	}
 
 	// reload the generic reloadables
+	// Currently logger is the only reloadable
 	for _, r := range a.reloadables {
 		err = r.Reload(ctx, cfg)
 		if err != nil {
@@ -405,6 +405,7 @@ func (a *Agent) reconfigure(ctx context.Context) error {
 }
 
 func (a *Agent) stop() {
+	// stop is called when expected config state indicates an input or output should stop
 	if a.srvCanceller == nil {
 		return
 	}
@@ -418,7 +419,7 @@ func (a *Agent) stop() {
 	a.srvCtx = nil
 	a.srv = nil
 	canceller()
-	<-a.srvDone
+	<-a.srvDone // wait for srv.Run loop to terminate either because root-context received a signal, or stop has been called
 	a.srvDone = nil
 
 	if a.inputUnit != nil {
@@ -460,6 +461,7 @@ func (a *Agent) configFromUnits(ctx context.Context) (*config.Config, error) {
 
 	// elastic-agent should be setting bootstrap with config provided through enrollment flags
 	if bootstrapCfg, ok := outMap["bootstrap"]; ok {
+		// Check if an output check loop is running, cancel if it is.
 		if a.outputCheckCanceller != nil {
 			a.outputCheckCanceller()
 			a.outputCheckCanceller = nil

internal/pkg/server/fleet.go

@@ -86,6 +86,14 @@ func (f *Fleet) GetConfig() *config.Config {
 
 // Run runs the fleet server
 func (f *Fleet) Run(ctx context.Context, initCfg *config.Config) error {
+	// ctx is cancelled when a SIGTERM or SIGINT is received or when the agent wrappers calls stop because a unit is being stopped/removed.
+
+	// Replace context with cancellable ctx
+	// in order to automatically cancel all the go routines
+	// that were started in the scope of this function on function exit
+	ctx, cn := context.WithCancel(ctx)
+	defer cn()
+
 	log := zerolog.Ctx(ctx)
 	err := initCfg.LoadServerLimits()
 	if err != nil {
@@ -102,12 +110,6 @@ func (f *Fleet) Run(ctx context.Context, initCfg *config.Config) error {
 	var curCfg *config.Config
 	newCfg := initCfg
 
-	// Replace context with cancellable ctx
-	// in order to automatically cancel all the go routines
-	// that were started in the scope of this function on function exit
-	ctx, cn := context.WithCancel(ctx)
-	defer cn()
-
 	stop := func(cn context.CancelFunc, g *errgroup.Group) {
 		if cn != nil {
 			cn()
@@ -320,7 +322,6 @@ func initRuntime(cfg *config.Config) {
 	gcPercent := cfg.Inputs[0].Server.Runtime.GCPercent
 	if gcPercent != 0 {
 		old := debug.SetGCPercent(gcPercent)
-
 		zerolog.Ctx(context.TODO()).Info().
 			Int("old", old).
 			Int("new", gcPercent).
@@ -329,7 +330,6 @@ func initRuntime(cfg *config.Config) {
 	memoryLimit := cfg.Inputs[0].Server.Runtime.MemoryLimit
 	if memoryLimit != 0 {
 		old := debug.SetMemoryLimit(memoryLimit)
-
 		zerolog.Ctx(context.TODO()).Info().
 			Int64("old", old).
 			Int64("new", memoryLimit).
@@ -433,6 +433,19 @@ func (f *Fleet) runServer(ctx context.Context, cfg *config.Config) (err error) {
 	return g.Wait()
 }
 
+// runSubsystems starts  all other subsystems for fleet-server
+// we assume bulker.Run is called in another goroutine, it's ctx is not the same ctx passed into runSubsystems and used with the passed errgroup.
+// however if the bulker returns an error, the passed errgroup is canceled.
+// runSubsystems will also do an ES version check and run migrations if started in agent-mode
+// The started subsystems are:
+// - Elasticsearch GC - cleanup expired fleet actions
+// - Policy Index Monitor - track new documents in the .fleet-policies index
+// - Policy Monitor - parse .fleet-policies docuuments into usable policies
+// - Policy Self Monitor - report fleet-server health status based on .fleet-policies index
+// - Action Monitor - track new documents in the .fleet-actions index
+// - Action Dispatcher - send actions from the .fleet-actions index to agents that check in
+// - Bulk Checkin handler - batches agent checkin messages to _bulk endpoint, minimizes changed attributes
+// - HTTP APIs - start http server on 8220 (default) for external agents, and on 8221 (default) for managing agent in agent-mode or local communications.
 func (f *Fleet) runSubsystems(ctx context.Context, cfg *config.Config, g *errgroup.Group, bulker bulk.Bulk, tracer *apm.Tracer) (err error) {
 	esCli := bulker.Client()
 
@@ -448,12 +461,9 @@ func (f *Fleet) runSubsystems(ctx context.Context, cfg *config.Config, g *errgro
 			}
 			return fmt.Errorf("failed version compatibility check with elasticsearch: %w", err)
 		}
-	}
 
-	// Migrations are not executed in standalone mode. When needed, they will be executed
-	// by some external process.
-	if !f.standAlone {
-		// Run migrations
+		// Migrations are not executed in standalone mode. When needed, they will be executed
+		// by some external process.
 		loggedMigration := loggedRunFunc(ctx, "Migrations", func(ctx context.Context) error {
 			return dl.Migrate(ctx, bulker)
 		})