Utilize new ES fleet polling API for global checkpoint monitoring (#200)

* Utilize fleet polling API for global checkpoint monitoring

* Adjust for API changes, configurable poll timeout

* Update retryDelay to 3 secs

* Update tests for monitor API change

* Adjust to the latest API changes

* Remove fleet indexes bootstrapping for tests, it is done by fleet system index plugin now

* Fix unit tests

Author: aleksmaus

cmd/fleet/handleCheckin.go

@@ -313,8 +313,8 @@ func (ct *CheckinT) fetchAgentPendingActions(ctx context.Context, seqno sqn.SeqN
 	now := time.Now().UTC().Format(time.RFC3339)
 
 	return dl.FindActions(ctx, ct.bulker, dl.QueryAgentActions, map[string]interface{}{
-		dl.FieldSeqNo:      seqno.Get(0),
-		dl.FieldMaxSeqNo:   ct.gcp.GetCheckpoint(),
+		dl.FieldSeqNo:      seqno.Value(),
+		dl.FieldMaxSeqNo:   ct.gcp.GetCheckpoint().Value(),
 		dl.FieldExpiration: now,
 		dl.FieldAgents:     []string{agentId},
 	})

cmd/fleet/main.go

@@ -21,6 +21,7 @@ import (
 	"github.com/elastic/fleet-server/v7/internal/pkg/config"
 	"github.com/elastic/fleet-server/v7/internal/pkg/coordinator"
 	"github.com/elastic/fleet-server/v7/internal/pkg/dl"
+	"github.com/elastic/fleet-server/v7/internal/pkg/es"
 	"github.com/elastic/fleet-server/v7/internal/pkg/logger"
 	"github.com/elastic/fleet-server/v7/internal/pkg/monitor"
 	"github.com/elastic/fleet-server/v7/internal/pkg/policy"
@@ -505,7 +506,13 @@ func (f *FleetServer) runServer(ctx context.Context, cfg *config.Config) (err er
 	// shutdown before the bulker is then cancelled.
 	bulkCtx, bulkCancel := context.WithCancel(context.Background())
 	defer bulkCancel()
-	es, bulker, err := bulk.InitES(bulkCtx, cfg)
+	esCli, bulker, err := bulk.InitES(bulkCtx, cfg)
+	if err != nil {
+		return err
+	}
+
+	// Monitoring es client, longer timeout, no retries
+	monCli, err := es.NewClient(ctx, cfg, true)
 	if err != nil {
 		return err
 	}
@@ -514,7 +521,10 @@ func (f *FleetServer) runServer(ctx context.Context, cfg *config.Config) (err er
 	g, ctx := errgroup.WithContext(ctx)
 
 	// Coordinator policy monitor
-	pim, err := monitor.New(dl.FleetPolicies, es, monitor.WithFetchSize(cfg.Inputs[0].Monitor.FetchSize))
+	pim, err := monitor.New(dl.FleetPolicies, esCli, monCli,
+		monitor.WithFetchSize(cfg.Inputs[0].Monitor.FetchSize),
+		monitor.WithPollTimeout(cfg.Inputs[0].Monitor.PollTimeout),
+	)
 	if err != nil {
 		return err
 	}
@@ -536,7 +546,11 @@ func (f *FleetServer) runServer(ctx context.Context, cfg *config.Config) (err er
 	var ad *action.Dispatcher
 	var tr *action.TokenResolver
 
-	am, err = monitor.NewSimple(dl.FleetActions, es, monitor.WithExpiration(true), monitor.WithFetchSize(cfg.Inputs[0].Monitor.FetchSize))
+	am, err = monitor.NewSimple(dl.FleetActions, esCli, monCli,
+		monitor.WithExpiration(true),
+		monitor.WithFetchSize(cfg.Inputs[0].Monitor.FetchSize),
+		monitor.WithPollTimeout(cfg.Inputs[0].Monitor.PollTimeout),
+	)
 	if err != nil {
 		return err
 	}

dev-tools/integration/main.go

@@ -31,7 +31,7 @@ func main() {
 	checkErr(err)
 
 	ctx := context.Background()
-	es, err := es.NewClient(ctx, cfg)
+	es, err := es.NewClient(ctx, cfg, false)
 	checkErr(err)
 
 	err = esutil.EnsureESIndices(ctx, es)

internal/pkg/bulk/bulk.go

@@ -87,7 +87,7 @@ const (
 
 func InitES(ctx context.Context, cfg *config.Config, opts ...BulkOpt) (*elasticsearch.Client, Bulk, error) {
 
-	es, err := es.NewClient(ctx, cfg)
+	es, err := es.NewClient(ctx, cfg, false)
 	if err != nil {
 		return nil, nil, err
 	}

internal/pkg/config/config_test.go

@@ -93,7 +93,8 @@ func TestConfig(t *testing.T) {
 							MaxCost:     defaultCacheMaxCost,
 						},
 						Monitor: Monitor{
-							FetchSize: defaultFetchSize,
+							FetchSize:   defaultFetchSize,
+							PollTimeout: defaultPollTimeout,
 						},
 					},
 				},
@@ -182,7 +183,8 @@ func TestConfig(t *testing.T) {
 							MaxCost:     defaultCacheMaxCost,
 						},
 						Monitor: Monitor{
-							FetchSize: defaultFetchSize,
+							FetchSize:   defaultFetchSize,
+							PollTimeout: defaultPollTimeout,
 						},
 					},
 				},
@@ -269,7 +271,8 @@ func TestConfig(t *testing.T) {
 							MaxCost:     defaultCacheMaxCost,
 						},
 						Monitor: Monitor{
-							FetchSize: defaultFetchSize,
+							FetchSize:   defaultFetchSize,
+							PollTimeout: defaultPollTimeout,
 						},
 					},
 				},
@@ -356,7 +359,8 @@ func TestConfig(t *testing.T) {
 							MaxCost:     defaultCacheMaxCost,
 						},
 						Monitor: Monitor{
-							FetchSize: defaultFetchSize,
+							FetchSize:   defaultFetchSize,
+							PollTimeout: defaultPollTimeout,
 						},
 					},
 				},

internal/pkg/config/monitor.go

@@ -4,14 +4,19 @@
 
 package config
 
+import "time"
+
 const (
-	defaultFetchSize = 1000
+	defaultFetchSize   = 1000
+	defaultPollTimeout = 5 * time.Minute
 )
 
 type Monitor struct {
-	FetchSize int `config:"fetch_size"`
+	FetchSize   int           `config:"fetch_size"`
+	PollTimeout time.Duration `config:"poll_timeout"`
 }
 
 func (m *Monitor) InitDefaults() {
 	m.FetchSize = defaultFetchSize
+	m.PollTimeout = defaultPollTimeout
 }

internal/pkg/config/output.go

@@ -20,6 +20,10 @@ import (
 	"github.com/elastic/beats/v7/libbeat/common/transport/tlscommon"
 )
 
+// The timeout would be driven by the server for long poll.
+// Giving it some sane long value.
+const httpTransportLongPollTimeout = 10 * time.Minute
+
 var hasScheme = regexp.MustCompile(`^([a-z][a-z0-9+\-.]*)://`)
 
 // Elasticsearch is the configuration for elasticsearch.
@@ -77,7 +81,7 @@ func (c *Elasticsearch) Validate() error {
 }
 
 // ToESConfig converts the configuration object into the config for the elasticsearch client.
-func (c *Elasticsearch) ToESConfig() (elasticsearch.Config, error) {
+func (c *Elasticsearch) ToESConfig(longPoll bool) (elasticsearch.Config, error) {
 	// build the addresses
 	addrs := make([]string, len(c.Hosts))
 	for i, host := range c.Hosts {
@@ -104,6 +108,17 @@ func (c *Elasticsearch) ToESConfig() (elasticsearch.Config, error) {
 		ResponseHeaderTimeout: c.Timeout,
 		ExpectContinueTimeout: 1 * time.Second,
 	}
+
+	disableRetry := false
+
+	if longPoll {
+		httpTransport.IdleConnTimeout = httpTransportLongPollTimeout
+		httpTransport.ResponseHeaderTimeout = httpTransportLongPollTimeout
+
+		// no retries for long poll monitoring
+		disableRetry = true
+	}
+
 	if c.TLS != nil && c.TLS.IsEnabled() {
 		tls, err := tlscommon.LoadTLSConfig(c.TLS)
 		if err != nil {
@@ -136,6 +151,7 @@ func (c *Elasticsearch) ToESConfig() (elasticsearch.Config, error) {
 		Header:       h,
 		Transport:    httpTransport,
 		MaxRetries:   c.MaxRetries,
+		DisableRetry: disableRetry,
 	}, nil
 }
 

internal/pkg/config/output_test.go

@@ -171,7 +171,7 @@ func TestToESConfig(t *testing.T) {
 			cmpopts.IgnoreUnexported(tls.Config{}),
 		}
 		t.Run(name, func(t *testing.T) {
-			res, err := test.cfg.ToESConfig()
+			res, err := test.cfg.ToESConfig(false)
 			require.NoError(t, err)
 			test.result.Header.Set("X-elastic-product-origin", "fleet")
 			if !assert.True(t, cmp.Equal(test.result, res, copts...)) {

internal/pkg/coordinator/monitor_integration_test.go

@@ -36,7 +36,7 @@ func TestMonitorLeadership(t *testing.T) {
 	serversIndex := ftesting.SetupIndex(bulkCtx, t, bulker, es.MappingServer)
 	policiesIndex := ftesting.SetupIndex(bulkCtx, t, bulker, es.MappingPolicy)
 	leadersIndex := ftesting.SetupIndex(bulkCtx, t, bulker, es.MappingPolicyLeader)
-	pim, err := monitor.New(policiesIndex, bulker.Client())
+	pim, err := monitor.New(policiesIndex, bulker.Client(), bulker.Client())
 	if err != nil {
 		t.Fatal(err)
 	}

internal/pkg/es/client.go

@@ -15,8 +15,8 @@ import (
 	"github.com/rs/zerolog/log"
 )
 
-func NewClient(ctx context.Context, cfg *config.Config) (*elasticsearch.Client, error) {
-	escfg, err := cfg.Output.Elasticsearch.ToESConfig()
+func NewClient(ctx context.Context, cfg *config.Config, longPoll bool) (*elasticsearch.Client, error) {
+	escfg, err := cfg.Output.Elasticsearch.ToESConfig(longPoll)
 	if err != nil {
 		return nil, err
 	}