[8.19](backport #44365) x-pack/filebeat/input/httpjson: add fleet input status updating (#44941)

* x-pack/filebeat/input/httpjson: add fleet input status updating (#44365)

(cherry picked from commit c203b82)

# Conflicts:
#	docs/reference/filebeat/filebeat-input-httpjson.md
#	x-pack/filebeat/input/httpjson/transform_set_test.go

* remove irrelevant changelog entry
* remove incorrectly cherry-picked markdown docs
* resolve conflicts
* add asciidoc documentation

---------

Co-authored-by: Dan Kortschak <dan.kortschak@elastic.co>

Author: mergify[bot]

CHANGELOG.next.asciidoc

@@ -411,6 +411,7 @@ otherwise no tag is added. {issue}42208[42208] {pull}42403[42403]
 - Introduce lastSync start position to AWS CloudWatch input backed by state registry. {pull}43251[43251]
 - Add Fleet status update functionality to udp input. {issue}44419[44419] {pull}44785[44785]
 - Filestream now logs at level warn the number of files that are too small to be ingested {pull}44751[44751]
+- Add Fleet status updating to HTTP JSON input. {issue}44282[44282] {pull}44365[44365]
 
 *Auditbeat*
 

x-pack/filebeat/docs/inputs/input-httpjson.asciidoc

@@ -572,6 +572,15 @@ e.g. instead of rate-limiting when `remaining` hits `0`, rate-limiting will occu
 
 It is not set by default (by default the rate-limiting as specified in the Response is followed).
 
+[[request-rate-limit-max-non-degraded]]
+[float]
+==== `request.rate_limit.max_non_degraded`
+
+The maximum number of consecutive 429 rate limit responses from the endpoint to accept
+before making an input health status update. If this is not set, rate limit responses
+will not be reported as health updates. Default behavior is to not report rate limiting
+responses.
+
 [[request-transforms]]
 [float]
 ==== `request.transforms`

x-pack/filebeat/input/httpjson/config_request.go

@@ -60,10 +60,11 @@ func (c retryConfig) getWaitMax() time.Duration {
 }
 
 type rateLimitConfig struct {
-	Limit      *valueTpl `config:"limit"`
-	Reset      *valueTpl `config:"reset"`
-	Remaining  *valueTpl `config:"remaining"`
-	EarlyLimit *float64  `config:"early_limit"`
+	Limit          *valueTpl `config:"limit"`
+	Reset          *valueTpl `config:"reset"`
+	Remaining      *valueTpl `config:"remaining"`
+	EarlyLimit     *float64  `config:"early_limit"`
+	MaxNonDegraded *int      `config:"max_non_degraded"`
 }
 
 func (c rateLimitConfig) Validate() error {
@@ -160,7 +161,7 @@ func (c *requestConfig) Validate() error {
 		return fmt.Errorf("unsupported method %q", c.Method)
 	}
 
-	if _, err := newBasicTransformsFromConfig(registeredTransforms, c.Transforms, requestNamespace, nil); err != nil {
+	if _, err := newBasicTransformsFromConfig(registeredTransforms, c.Transforms, requestNamespace, noopReporter{}, nil); err != nil {
 		return err
 	}
 

x-pack/filebeat/input/httpjson/config_response.go

@@ -37,10 +37,10 @@ type splitConfig struct {
 }
 
 func (c *responseConfig) Validate() error {
-	if _, err := newBasicTransformsFromConfig(registeredTransforms, c.Transforms, responseNamespace, nil); err != nil {
+	if _, err := newBasicTransformsFromConfig(registeredTransforms, c.Transforms, responseNamespace, noopReporter{}, nil); err != nil {
 		return err
 	}
-	if _, err := newBasicTransformsFromConfig(registeredTransforms, c.Pagination, paginationNamespace, nil); err != nil {
+	if _, err := newBasicTransformsFromConfig(registeredTransforms, c.Pagination, paginationNamespace, noopReporter{}, nil); err != nil {
 		return err
 	}
 	if c.DecodeAs != "" {
@@ -52,7 +52,7 @@ func (c *responseConfig) Validate() error {
 }
 
 func (c *splitConfig) Validate() error {
-	if _, err := newBasicTransformsFromConfig(registeredTransforms, c.Transforms, responseNamespace, nil); err != nil {
+	if _, err := newBasicTransformsFromConfig(registeredTransforms, c.Transforms, responseNamespace, noopReporter{}, nil); err != nil {
 		return err
 	}
 
@@ -71,7 +71,7 @@ func (c *splitConfig) Validate() error {
 		return fmt.Errorf("invalid split type: %s", c.Type)
 	}
 
-	if _, err := newSplitResponse(c, nil); err != nil {
+	if _, err := newSplitResponse(c, noopReporter{}, nil); err != nil {
 		return err
 	}
 

x-pack/filebeat/input/httpjson/cursor.go

@@ -6,6 +6,7 @@ package httpjson
 
 import (
 	inputcursor "github.com/elastic/beats/v7/filebeat/input/v2/input-cursor"
+	"github.com/elastic/beats/v7/libbeat/management/status"
 	"github.com/elastic/elastic-agent-libs/logp"
 	"github.com/elastic/elastic-agent-libs/mapstr"
 )
@@ -15,11 +16,12 @@ type cursor struct {
 
 	state mapstr.M
 
-	log *logp.Logger
+	status status.StatusReporter
+	log    *logp.Logger
 }
 
-func newCursor(cfg cursorConfig, log *logp.Logger) *cursor {
-	return &cursor{cfg: cfg, log: log}
+func newCursor(cfg cursorConfig, stat status.StatusReporter, log *logp.Logger) *cursor {
+	return &cursor{cfg: cfg, status: stat, log: log}
 }
 
 func (c *cursor) load(cursor *inputcursor.Cursor) {
@@ -34,6 +36,7 @@ func (c *cursor) load(cursor *inputcursor.Cursor) {
 
 	if err := cursor.Unpack(&c.state); err != nil {
 		c.log.Errorf("Reset cursor state. Failed to read from registry: %v", err)
+		c.status.UpdateStatus(status.Degraded, "failed to load cursor: "+err.Error())
 		return
 	}
 
@@ -50,7 +53,7 @@ func (c *cursor) update(trCtx *transformContext) {
 	}
 
 	for k, cfg := range c.cfg {
-		v, _ := cfg.Value.Execute(trCtx, transformable{}, k, cfg.Default, c.log)
+		v, _ := cfg.Value.Execute(trCtx, transformable{}, k, cfg.Default, c.status, c.log)
 		if v != "" || !cfg.mustIgnoreEmptyValue() {
 			_, _ = c.state.Put(k, v)
 			c.log.Debugf("cursor.%s stored with %s", k, v)

x-pack/filebeat/input/httpjson/cursor_test.go

@@ -110,7 +110,7 @@ func TestCursorUpdate(t *testing.T) {
 			conf := cursorConfig{}
 			require.NoError(t, cfg.Unpack(&conf))
 
-			c := newCursor(conf, logp.NewLogger("cursor-test"))
+			c := newCursor(conf, noopReporter{}, logp.NewLogger("cursor-test"))
 			c.state = tc.initialState
 			c.update(tc.trCtx)
 			assert.Equal(t, tc.expectedState, c.state)

x-pack/filebeat/input/httpjson/input.go

@@ -30,6 +30,7 @@ import (
 	inputcursor "github.com/elastic/beats/v7/filebeat/input/v2/input-cursor"
 	"github.com/elastic/beats/v7/libbeat/beat"
 	"github.com/elastic/beats/v7/libbeat/feature"
+	"github.com/elastic/beats/v7/libbeat/management/status"
 	"github.com/elastic/beats/v7/libbeat/monitoring/inputmon"
 	"github.com/elastic/beats/v7/libbeat/statestore"
 	"github.com/elastic/beats/v7/libbeat/version"
@@ -175,8 +176,14 @@ func runWithMetrics(ctx v2.Context, cfg config, pub inputcursor.Publisher, crsr
 }
 
 func run(ctx v2.Context, cfg config, pub inputcursor.Publisher, crsr *inputcursor.Cursor, reg *monitoring.Registry) error {
-	log := ctx.Logger.With("input_url", cfg.Request.URL)
+	stat := ctx.StatusReporter
+	if stat == nil {
+		stat = noopReporter{}
+	}
+	stat.UpdateStatus(status.Starting, "")
+	stat.UpdateStatus(status.Configuring, "")
 
+	log := ctx.Logger.With("input_url", cfg.Request.URL)
 	stdCtx := ctxtool.FromCanceller(ctx.Cancelation)
 
 	if cfg.Request.Tracer != nil {
@@ -196,30 +203,33 @@ func run(ctx v2.Context, cfg config, pub inputcursor.Publisher, crsr *inputcurso
 
 	metrics := newInputMetrics(reg)
 
-	client, err := newHTTPClient(stdCtx, cfg, log, reg)
+	client, err := newHTTPClient(stdCtx, cfg, stat, log, reg)
 	if err != nil {
+		stat.UpdateStatus(status.Failed, "failed to create HTTP client: "+err.Error())
 		return err
 	}
 
-	requestFactory, err := newRequestFactory(stdCtx, cfg, log, metrics, reg)
+	requestFactory, err := newRequestFactory(stdCtx, cfg, stat, log, metrics, reg)
 	if err != nil {
 		log.Errorf("Error while creating requestFactory: %v", err)
+		stat.UpdateStatus(status.Failed, "failed to create request factory: "+err.Error())
 		return err
 	}
 	var xmlDetails map[string]xml.Detail
 	if cfg.Response.XSD != "" {
 		xmlDetails, err = xml.Details([]byte(cfg.Response.XSD))
 		if err != nil {
 			log.Errorf("error while collecting xml decoder type hints: %v", err)
+			stat.UpdateStatus(status.Failed, "error while collecting xml decoder type hints: "+err.Error())
 			return err
 		}
 	}
-	pagination := newPagination(cfg, client, log)
-	responseProcessor := newResponseProcessor(cfg, pagination, xmlDetails, metrics, log)
-	requester := newRequester(client, requestFactory, responseProcessor, metrics, log)
+	pagination := newPagination(cfg, client, stat, log)
+	responseProcessor := newResponseProcessor(cfg, pagination, xmlDetails, metrics, stat, log)
+	requester := newRequester(client, requestFactory, responseProcessor, metrics, stat, log)
 
 	trCtx := emptyTransformContext()
-	trCtx.cursor = newCursor(cfg.Cursor, log)
+	trCtx.cursor = newCursor(cfg.Cursor, stat, log)
 	trCtx.cursor.load(crsr)
 
 	doFunc := func() error {
@@ -241,6 +251,7 @@ func run(ctx v2.Context, cfg config, pub inputcursor.Publisher, crsr *inputcurso
 		metrics.updateIntervalMetrics(err, startTime)
 
 		if err := stdCtx.Err(); err != nil {
+			stat.UpdateStatus(status.Stopping, "")
 			return err
 		}
 
@@ -254,10 +265,14 @@ func run(ctx v2.Context, cfg config, pub inputcursor.Publisher, crsr *inputcurso
 	}
 
 	log.Infof("Input stopped because context was cancelled with: %v", err)
-
+	stat.UpdateStatus(status.Stopped, "")
 	return nil
 }
 
+type noopReporter struct{}
+
+func (noopReporter) UpdateStatus(status.Status, string) {}
+
 // sanitizeFileName returns name with ":" and "/" replaced with "_", removing repeated instances.
 // The request.tracer.filename may have ":" when a httpjson input has cursor config and
 // the macOS Finder will treat this as path-separator and causes to show up strange filepaths.
@@ -267,7 +282,7 @@ func sanitizeFileName(name string) string {
 	return strings.ReplaceAll(name, string(filepath.Separator), "_")
 }
 
-func newHTTPClient(ctx context.Context, config config, log *logp.Logger, reg *monitoring.Registry) (*httpClient, error) {
+func newHTTPClient(ctx context.Context, config config, stat status.StatusReporter, log *logp.Logger, reg *monitoring.Registry) (*httpClient, error) {
 	client, err := newNetHTTPClient(ctx, config.Request, log, reg)
 	if err != nil {
 		return nil, err
@@ -286,7 +301,7 @@ func newHTTPClient(ctx context.Context, config config, log *logp.Logger, reg *mo
 		}).StandardClient()
 	}
 
-	limiter := newRateLimiterFromConfig(config.Request.RateLimit, log)
+	limiter := newRateLimiterFromConfig(config.Request.RateLimit, stat, log)
 
 	if config.Auth.OAuth2.isEnabled() {
 		authClient, err := config.Auth.OAuth2.client(ctx, client)
@@ -356,7 +371,7 @@ func newNetHTTPClient(ctx context.Context, cfg *requestConfig, log *logp.Logger,
 	return netHTTPClient, nil
 }
 
-func newChainHTTPClient(ctx context.Context, authCfg *authConfig, requestCfg *requestConfig, log *logp.Logger, reg *monitoring.Registry, p ...*Policy) (*httpClient, error) {
+func newChainHTTPClient(ctx context.Context, authCfg *authConfig, requestCfg *requestConfig, stat status.StatusReporter, log *logp.Logger, reg *monitoring.Registry, p ...*Policy) (*httpClient, error) {
 	client, err := newNetHTTPClient(ctx, requestCfg, log, reg)
 	if err != nil {
 		return nil, err
@@ -382,7 +397,7 @@ func newChainHTTPClient(ctx context.Context, authCfg *authConfig, requestCfg *re
 		}).StandardClient()
 	}
 
-	limiter := newRateLimiterFromConfig(requestCfg.RateLimit, log)
+	limiter := newRateLimiterFromConfig(requestCfg.RateLimit, stat, log)
 
 	if authCfg != nil && authCfg.OAuth2.isEnabled() {
 		authClient, err := authCfg.OAuth2.client(ctx, client)

x-pack/filebeat/input/httpjson/pagination.go

@@ -11,6 +11,7 @@ import (
 	"net/http"
 	"net/url"
 
+	"github.com/elastic/beats/v7/libbeat/management/status"
 	"github.com/elastic/mito/lib/xml"
 
 	"github.com/elastic/elastic-agent-libs/logp"
@@ -23,11 +24,12 @@ type pagination struct {
 	client         *httpClient
 	requestFactory *requestFactory
 	decoder        decoderFunc
+	status         status.StatusReporter
 	log            *logp.Logger
 }
 
-func newPagination(config config, client *httpClient, log *logp.Logger) *pagination {
-	pagination := &pagination{client: client, log: log}
+func newPagination(config config, client *httpClient, stat status.StatusReporter, log *logp.Logger) *pagination {
+	pagination := &pagination{client: client, status: stat, log: log}
 	if config.Response == nil {
 		return pagination
 	}
@@ -38,8 +40,8 @@ func newPagination(config config, client *httpClient, log *logp.Logger) *paginat
 		return pagination
 	}
 
-	rts, _ := newBasicTransformsFromConfig(registeredTransforms, config.Request.Transforms, requestNamespace, log)
-	pts, _ := newBasicTransformsFromConfig(registeredTransforms, config.Response.Pagination, paginationNamespace, log)
+	rts, _ := newBasicTransformsFromConfig(registeredTransforms, config.Request.Transforms, requestNamespace, stat, log)
+	pts, _ := newBasicTransformsFromConfig(registeredTransforms, config.Response.Pagination, paginationNamespace, stat, log)
 
 	body := func() *mapstr.M {
 		if config.Response.RequestBodyOnPagination {
@@ -55,13 +57,14 @@ func newPagination(config config, client *httpClient, log *logp.Logger) *paginat
 		body,
 		append(rts, pts...),
 		config.Auth,
+		stat,
 		log,
 	)
 	pagination.requestFactory = requestFactory
 	return pagination
 }
 
-func newPaginationRequestFactory(method, encodeAs string, url url.URL, body *mapstr.M, ts []basicTransform, authConfig *authConfig, log *logp.Logger) *requestFactory {
+func newPaginationRequestFactory(method, encodeAs string, url url.URL, body *mapstr.M, ts []basicTransform, authConfig *authConfig, stat status.StatusReporter, log *logp.Logger) *requestFactory {
 	// config validation already checked for errors here
 	rf := &requestFactory{
 		url:        url,
@@ -92,16 +95,19 @@ type pageIterator struct {
 	done    bool
 
 	n int64
+
+	status status.StatusReporter
 }
 
-func (p *pagination) newPageIterator(stdCtx context.Context, trCtx *transformContext, resp *http.Response, xmlDetails map[string]xml.Detail) *pageIterator {
+func (p *pagination) newPageIterator(stdCtx context.Context, trCtx *transformContext, resp *http.Response, xmlDetails map[string]xml.Detail, stat status.StatusReporter) *pageIterator {
 	return &pageIterator{
 		pagination: p,
 		stdCtx:     stdCtx,
 		trCtx:      trCtx,
 		resp:       resp,
 		xmlDetails: xmlDetails,
 		isFirst:    true,
+		status:     stat,
 	}
 }
 
@@ -161,6 +167,7 @@ func (iter *pageIterator) next() (*response, bool, error) {
 func (iter *pageIterator) getPage() (*response, error) {
 	bodyBytes, err := io.ReadAll(iter.resp.Body)
 	if err != nil {
+		iter.status.UpdateStatus(status.Degraded, "failed to read page body: "+err.Error())
 		return nil, err
 	}
 	iter.resp.Body.Close()
@@ -182,6 +189,7 @@ func (iter *pageIterator) getPage() (*response, error) {
 			err = decode(iter.resp.Header.Get("Content-Type"), bodyBytes, &r)
 		}
 		if err != nil {
+			iter.status.UpdateStatus(status.Degraded, "failed to decode page: "+err.Error())
 			return nil, err
 		}
 	}

x-pack/filebeat/input/httpjson/policy.go

@@ -15,11 +15,11 @@ import (
 	"net/url"
 	"regexp"
 
+	"github.com/elastic/beats/v7/libbeat/management/status"
 	"github.com/elastic/elastic-agent-libs/logp"
 )
 
 var (
-
 	// A regular expression to match the error returned by net/http when the
 	// configured number of redirects is exhausted. This error isn't typed
 	// specifically so we resort to matching on the error string.
@@ -36,20 +36,22 @@ var (
 // field value present in data using the defined operator/function in the given expression.
 // Example : [[ eq .last_response.body.status "completed" ]] -- which means here data is a http response
 // containing a field "status" under the field "body" , and value status should be equal to the string "completed"
-type Evaluate func(expression *valueTpl, data []byte, log *logp.Logger) (bool, error)
+type Evaluate func(expression *valueTpl, data []byte, stat status.StatusReporter, log *logp.Logger) (bool, error)
 
 // Policy is responsible for maintaining different http client policies
 // Currently just contains a retry policy function
 type Policy struct {
 	fn         Evaluate
 	expression *valueTpl
+	status     status.StatusReporter
 	log        *logp.Logger
 }
 
-func newHTTPPolicy(fn Evaluate, expression *valueTpl, log *logp.Logger) *Policy {
+func newHTTPPolicy(fn Evaluate, expression *valueTpl, stat status.StatusReporter, log *logp.Logger) *Policy {
 	return &Policy{
 		fn:         fn,
 		expression: expression,
+		status:     stat,
 		log:        log,
 	}
 }
@@ -115,7 +117,7 @@ func (p *Policy) CustomRetryPolicy(ctx context.Context, resp *http.Response, err
 		}
 		resp.Body = io.NopCloser(bytes.NewBuffer(body))
 
-		result, err := p.fn(p.expression, body, p.log)
+		result, err := p.fn(p.expression, body, p.status, p.log)
 		if err != nil {
 			return retry, err
 		}

x-pack/filebeat/input/httpjson/policy_test.go

@@ -136,6 +136,7 @@ func TestPolicy_CustomRetryPolicy(t *testing.T) {
 			p := &Policy{
 				fn:         tt.fields.fn,
 				expression: tt.fields.expression,
+				status:     noopReporter{},
 				log:        tt.fields.log,
 			}
 			got, err := p.CustomRetryPolicy(tt.args.ctx, tt.args.resp, tt.args.err)