Add agent diagnostics action

_meta/config/common.p2.yml.tmpl

@@ -49,6 +49,22 @@ inputs:
 #       port: 6791
 #       # Metrics buffer endpoint
 #       buffer.enabled: false
+#   # Configuration for the diagnostics action handler
+#   diagnostics:
+#       # Rate limit for the action handler. Does not affect diagnostics collected through the CLI.
+#       limit:
+#           # Rate limit interval.
+#           interval: 1m
+#           # Rate limit burst.
+#           burst: 1
+#       # Configuration for the file-upload client. Client may retry failed requests with an exponential backoff.
+#       uploader:
+#           # Max retries allowed when uploading a chunk.
+#           max_retries: 10
+#           # Initial duration of the backoff.
+#           init_dur: 1s
+#           # Max duration of the backoff.
+#           max_dur: 1m
 
 # # Allow fleet to reload its configuration locally on disk.
 # # Notes: Only specific process configuration will be reloaded.

_meta/config/common.reference.p2.yml.tmpl

@@ -125,6 +125,22 @@ inputs:
 #       port: 6791
 #       # Metrics buffer endpoint
 #       buffer.enabled: false
+#   # Configuration for the diagnostics action handler
+#   diagnostics:
+#       # Rate limit for the action handler. Does not affect diagnostics collected through the CLI.
+#       limit:
+#           # Rate limit interval.
+#           interval: 1m
+#           # Rate limit burst.
+#           burst: 1
+#       # Configuration for the file-upload client. Client may retry failed requests with an exponential backoff.
+#       uploader:
+#           # Max retries allowed when uploading a chunk.
+#           max_retries: 10
+#           # Initial duration of the backoff.
+#           init_dur: 1s
+#           # Max duration of the backoff.
+#           max_dur: 1m
 
 # # Allow fleet to reload its configuration locally on disk.
 # # Notes: Only specific process configuration and external input configurations will be reloaded.

changelog/fragments/1674785407-add-diagnostics-action-handler.yaml

@@ -0,0 +1,36 @@
+# Kind can be one of:
+# - breaking-change: a change to previously-documented behavior
+# - deprecation: functionality that is being removed in a later release
+# - bug-fix: fixes a problem in a previous version
+# - enhancement: extends functionality but does not break or fix existing behavior
+# - feature: new functionality
+# - known-issue: problems that we are aware of in a given version
+# - security: impacts on the security of a product or a user’s deployment.
+# - upgrade: important information for someone upgrading from a prior version
+# - other: does not fit into any of the other categories
+kind: feature
+
+# Change summary; a 80ish characters long description of the change.
+summary: add diagnostics action handler
+
+# Long description; in case the summary is not enough to describe the change
+# this field accommodate a description without length limits.
+description: |
+  Add support for the REQUEST_DIAGNOSTICS action.
+  When this action is recieved the agent will collect a diagnostics bundle and
+  uploads it to fleet-server using the file upload APIs.
+  The handler has a configurable rate limit in order to prevent DOS attacks.
+  The uploader may retry failures with a configurable exponential backoff.
+
+# Affected component; a word indicating the component this changeset affects.
+component: diagnostics
+
+# PR number; optional; the PR number that added the changeset.
+# If not present is automatically filled by the tooling finding the PR where this changelog fragment has been added.
+# NOTE: the tooling supports backports, so it's able to fill the original PR number instead of the backport PR number.
+# Please provide it if you are adding a fragment for a different PR.
+pr: 1703
+
+# Issue number; optional; the GitHub issue related to this changeset (either closes or is part of).
+# If not present is automatically filled by the tooling with the issue linked to the PR number.
+issue: 1883

elastic-agent.reference.yml

@@ -131,6 +131,22 @@ inputs:
 #       port: 6791
 #       # Metrics buffer endpoint
 #       buffer.enabled: false
+#   # Configuration for the diagnostics action handler
+#   diagnostics:
+#       # Rate limit for the action handler. Does not affect diagnostics collected through the CLI.
+#       limit:
+#           # Rate limit interval.
+#           interval: 1m
+#           # Rate limit burst.
+#           burst: 1
+#       # Configuration for the file-upload client. Client may retry failed requests with an exponential backoff.
+#       uploader:
+#           # Max retries allowed when uploading a chunk.
+#           max_retries: 10
+#           # Initial duration of the backoff.
+#           init_dur: 1s
+#           # Max duration of the backoff.
+#           max_dur: 1m
 
 # # Allow fleet to reload its configuration locally on disk.
 # # Notes: Only specific process configuration and external input configurations will be reloaded.

elastic-agent.yml

@@ -55,6 +55,22 @@ inputs:
 #       port: 6791
 #       # Metrics buffer endpoint
 #       buffer.enabled: false
+#   # Configuration for the diagnostics action handler
+#   diagnostics:
+#       # Rate limit for the action handler. Does not affect diagnostics collected through the CLI.
+#       limit:
+#           # Rate limit interval.
+#           interval: 1m
+#           # Rate limit burst.
+#           burst: 1
+#       # Configuration for the file-upload client. Client may retry failed requests with an exponential backoff.
+#       uploader:
+#           # Max retries allowed when uploading a chunk.
+#           max_retries: 10
+#           # Initial duration of the backoff.
+#           init_dur: 1s
+#           # Max duration of the backoff.
+#           max_dur: 1m
 
 # # Allow fleet to reload its configuration locally on disk.
 # # Notes: Only specific process configuration will be reloaded.

internal/pkg/agent/application/actions/handlers/handler_action_diagnostics.go

@@ -0,0 +1,204 @@
+// Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+// or more contributor license agreements. Licensed under the Elastic License;
+// you may not use this file except in compliance with the Elastic License.
+
+package handlers
+
+import (
+	"bytes"
+	"context"
+	"fmt"
+	"io"
+	"os"
+	"time"
+
+	"github.com/elastic/elastic-agent/internal/pkg/agent/application/coordinator"
+	"github.com/elastic/elastic-agent/internal/pkg/agent/control/v2/client"
+	"github.com/elastic/elastic-agent/internal/pkg/agent/control/v2/cproto"
+	"github.com/elastic/elastic-agent/internal/pkg/core/monitoring/config"
+	"github.com/elastic/elastic-agent/internal/pkg/diagnostics"
+	"github.com/elastic/elastic-agent/internal/pkg/fleetapi"
+	"github.com/elastic/elastic-agent/internal/pkg/fleetapi/acker"
+	"github.com/elastic/elastic-agent/pkg/core/logger"
+
+	"golang.org/x/time/rate"
+)
+
+// ErrRateLimit is the rate limit error that is returned if the handler is ran too often.
+// This may occur if the user sends multiple diagnostics actions to an agent in a short duration
+// or if the agent goes offline and retrieves multiple diagnostics actions.
+// In either case the 1st action will succeed and the others will ack with an the error.
+var ErrRateLimit = fmt.Errorf("rate limit exceeded")
+
+// Uploader is the interface used to upload a diagnostics bundle to fleet-server.
+type Uploader interface {
+	UploadDiagnostics(context.Context, string, string, int64, io.Reader) (string, error)
+}
+
+// Diagnostics is the handler to process Diagnostics actions.
+// When a Diagnostics action is received a full diagnostics bundle is taken and uploaded to fleet-server.
+type Diagnostics struct {
+	log      *logger.Logger
+	coord    *coordinator.Coordinator
+	limiter  *rate.Limiter
+	uploader Uploader
+}
+
+// NewDiagnostics returns a new Diagnostics handler.
+func NewDiagnostics(log *logger.Logger, coord *coordinator.Coordinator, cfg config.Limit, uploader Uploader) *Diagnostics {
+	return &Diagnostics{
+		log:      log,
+		coord:    coord,
+		limiter:  rate.NewLimiter(rate.Every(cfg.Interval), cfg.Burst),
+		uploader: uploader,
+	}
+}
+
+// Handle processes the passed Diagnostics action.
+func (h *Diagnostics) Handle(ctx context.Context, a fleetapi.Action, ack acker.Acker) error {
+	h.log.Debugf("handlerDiagnostics: action '%+v' received", a)
+	action, ok := a.(*fleetapi.ActionDiagnostics)
+	if !ok {
+		return fmt.Errorf("invalid type, expected ActionDiagnostics and received %T", a)
+	}
+	ts := time.Now().UTC()
+	defer func() {
+		ack.Ack(ctx, action) //nolint:errcheck // no path for a failed ack
+		ack.Commit(ctx)      //nolint:errcheck //no path for failing a commit
+	}()
+
+	if !h.limiter.Allow() {
+		action.Err = ErrRateLimit
+		return ErrRateLimit
+	}
+
+	h.log.Debug("Gathering agent diagnostics.")
+	aDiag, err := h.runHooks(ctx)
+	if err != nil {
+		action.Err = err
+		return fmt.Errorf("unable to gather agent diagnostics: %w", err)
+	}
+	h.log.Debug("Gathering unit diagnostics.")
+	uDiag := h.diagUnits(ctx)
+
+	var r io.Reader
+	// attempt to create the a temporary diagnostics file on disk in order to avoid loading a
+	// potentially large file in memory.
+	// if on-disk creation fails an in-memory buffer is used.
+	f, s, err := h.diagFile(aDiag, uDiag)
+	if err != nil {
+		var b bytes.Buffer
+		h.log.Warnw("Diagnostics action unable to use tempoary file, using buffer instead.", "err", err)
+		var wBuf bytes.Buffer
+		defer func() {
+			if str := wBuf.String(); str != "" {
+				h.log.Warn(str)
+			}
+		}()
+		err := diagnostics.ZipArchive(&wBuf, &b, aDiag, uDiag)
+		if err != nil {
+			action.Err = err
+			return fmt.Errorf("error creating diagnostics bundle: %w", err)
+		}
+		r = &b
+		s = int64(b.Len())
+	} else {
+		defer func() {
+			f.Close()
+			os.Remove(f.Name())
+		}()
+		r = f
+	}
+	h.log.Debug("Sending diagnostics archive.")
+	uploadID, err := h.uploader.UploadDiagnostics(ctx, action.ActionID, ts.Format("2006-01-02T15-04-05Z07-00"), s, r) // RFC3339 format that uses - instead of : so it works on Windows
+	action.Err = err
+	action.UploadID = uploadID
+	if err != nil {
+		return fmt.Errorf("unable to upload diagnostics: %w", err)
+	}
+	h.log.Debugf("Diagnostics action '%+v' complete.", a)
+	return nil
+}
+
+func (h *Diagnostics) runHooks(ctx context.Context) ([]client.DiagnosticFileResult, error) {
+	hooks := append(h.coord.DiagnosticHooks(), diagnostics.GlobalHooks()...)
+	diags := make([]client.DiagnosticFileResult, 0, len(hooks))
+	for _, hook := range hooks {
+		if ctx.Err() != nil {
+			return diags, ctx.Err()
+		}
+		diags = append(diags, client.DiagnosticFileResult{
+			Name:        hook.Name,
+			Filename:    hook.Filename,
+			Description: hook.Description,
+			ContentType: hook.ContentType,
+			Content:     hook.Hook(ctx),
+			Generated:   time.Now().UTC(),
+		})
+	}
+	return diags, nil
+}
+
+func (h *Diagnostics) diagUnits(ctx context.Context) []client.DiagnosticUnitResult {
+	uDiag := make([]client.DiagnosticUnitResult, 0)
+	rr := h.coord.PerformDiagnostics(ctx)
+	for _, r := range rr {
+		diag := client.DiagnosticUnitResult{
+			ComponentID: r.Component.ID,
+			UnitID:      r.Unit.ID,
+			UnitType:    cproto.UnitType(r.Unit.Type),
+		}
+		if r.Err != nil {
+			diag.Err = r.Err
+		} else {
+			results := make([]client.DiagnosticFileResult, 0, len(r.Results))
+			for _, res := range r.Results {
+				results = append(results, client.DiagnosticFileResult{
+					Name:        res.Name,
+					Filename:    res.Filename,
+					Description: res.Description,
+					ContentType: res.ContentType,
+					Content:     res.Content,
+					Generated:   res.Generated.AsTime(),
+				})
+			}
+			diag.Results = results
+		}
+		uDiag = append(uDiag, diag)
+	}
+	return uDiag
+}
+
+// diagFile will write the diagnostics to a temporary file and return the file ready to be read
+func (h *Diagnostics) diagFile(aDiag []client.DiagnosticFileResult, uDiag []client.DiagnosticUnitResult) (*os.File, int64, error) {
+	f, err := os.CreateTemp("", "elastic-agent-diagnostics")
+	if err != nil {
+		return nil, 0, err
+	}
+
+	name := f.Name()
+	var wBuf bytes.Buffer
+	defer func() {
+		if str := wBuf.String(); str != "" {
+			h.log.Warn(str)
+		}
+	}()
+	if err := diagnostics.ZipArchive(&wBuf, f, aDiag, uDiag); err != nil {
+		os.Remove(name)
+		return nil, 0, err
+	}
+	f.Sync()
+
+	_, err = f.Seek(0, 0)
+	if err != nil {
+		os.Remove(name)
+		return nil, 0, err
+	}
+
+	fi, err := f.Stat()
+	if err != nil {
+		os.Remove(name)
+		return nil, 0, err
+	}
+	return f, fi.Size(), nil
+}

internal/pkg/agent/application/coordinator/coordinator.go

@@ -293,7 +293,7 @@ func (c *Coordinator) Upgrade(ctx context.Context, version string, sourceURI str
 	return nil
 }
 
-// AckUpgrade performs acknowledgement for upgrade.
+// AckUpgrade is the method used on startup to ack a previously successful upgrade action.
 func (c *Coordinator) AckUpgrade(ctx context.Context, acker acker.Acker) error {
 	return c.upgradeMgr.Ack(ctx, acker)
 }

internal/pkg/agent/application/dispatcher/dispatcher_test.go

@@ -53,6 +53,10 @@ func (m *mockAction) String() string {
 	args := m.Called()
 	return args.String(0)
 }
+func (m *mockAction) AckEvent() fleetapi.AckEvent {
+	args := m.Called()
+	return args.Get(0).(fleetapi.AckEvent)
+}
 func (m *mockScheduledAction) StartTime() (time.Time, error) {
 	args := m.Called()
 	return args.Get(0).(time.Time), args.Error(1)

internal/pkg/agent/application/managed_mode.go

@@ -25,6 +25,7 @@ import (
 	"github.com/elastic/elastic-agent/internal/pkg/fleetapi/acker/lazy"
 	"github.com/elastic/elastic-agent/internal/pkg/fleetapi/acker/retrier"
 	fleetclient "github.com/elastic/elastic-agent/internal/pkg/fleetapi/client"
+	"github.com/elastic/elastic-agent/internal/pkg/fleetapi/uploader"
 	"github.com/elastic/elastic-agent/internal/pkg/queue"
 	"github.com/elastic/elastic-agent/internal/pkg/remote"
 	"github.com/elastic/elastic-agent/internal/pkg/runner"
@@ -348,6 +349,16 @@ func (m *managedConfigManager) initDispatcher(canceller context.CancelFunc) *han
 		),
 	)
 
+	m.dispatcher.MustRegister(
+		&fleetapi.ActionDiagnostics{},
+		handlers.NewDiagnostics(
+			m.log,
+			m.coord,
+			m.cfg.Settings.MonitoringConfig.Diagnostics.Limit,
+			uploader.New(m.agentInfo.AgentID(), m.client, m.cfg.Settings.MonitoringConfig.Diagnostics.Uploader),
+		),
+	)
+
 	m.dispatcher.MustRegister(
 		&fleetapi.ActionApp{},
 		handlers.NewAppAction(m.log, m.coord),