forked from elastic/beats
-
Notifications
You must be signed in to change notification settings - Fork 0
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
x-pack/filebeat/input/http_endpoint: add ability to remove request tr…
…ace logs This is essentially a replay of elastic#39969, but for the http_endpoint input. The previous configuration system did not allow users to remove trace logs from agents after they are no longer needed. This is potential security risk as it leaves potentially sensitive information on the file system beyond its required lifetime. The mechanism for communicating to the input whether to write logs is not currently powerful enough to indicate that existing logs should be removed without deleting logs from other instances. So add an enabled configuration option to allow the target name to be sent independently of whether the files should be written or removed. The new option is optional, defaulting to the previous behaviour so that it can be opted into via progressive repair in the client integrations.
- Loading branch information
Showing
6 changed files
with
273 additions
and
12 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
221 changes: 221 additions & 0 deletions
221
x-pack/filebeat/input/http_endpoint/0001-review-changes.patch
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,221 @@ | ||
| From 4c19d586bba2ae0f37547f6411572d7da322e62c Mon Sep 17 00:00:00 2001 | ||
| From: Dan Kortschak <[email protected]> | ||
| Date: Wed, 15 May 2024 07:17:13 +0930 | ||
| Subject: [PATCH] review changes | ||
|
|
||
| --- | ||
| x-pack/filebeat/input/cel/input.go | 13 ++--- | ||
| x-pack/filebeat/input/cel/integration_test.go | 48 +++++++------------ | ||
| 2 files changed, 21 insertions(+), 40 deletions(-) | ||
|
|
||
| diff --git a/x-pack/filebeat/input/cel/input.go b/x-pack/filebeat/input/cel/input.go | ||
| index fc4a7e7e3d..89488ec172 100644 | ||
| --- a/x-pack/filebeat/input/cel/input.go | ||
| +++ b/x-pack/filebeat/input/cel/input.go | ||
| @@ -115,13 +115,12 @@ func (input) Run(env v2.Context, src inputcursor.Source, crsr inputcursor.Cursor | ||
| updateStatus(env.StatusReporter, status.Failed, "failed to run: "+err.Error()) | ||
| return err | ||
| } | ||
| - | ||
| updateStatus(env.StatusReporter, status.Stopped, "") | ||
| return nil | ||
| } | ||
|
|
||
| // 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 request.tracer.filename may have ":" when a cel input has cursor config and | ||
| // the macOS Finder will treat this as path-separator and causes to show up strange filepaths. | ||
| func sanitizeFileName(name string) string { | ||
| name = strings.ReplaceAll(name, ":", string(filepath.Separator)) | ||
| @@ -181,6 +180,7 @@ func (i input) run(env v2.Context, src *source, cursor map[string]interface{}, p | ||
| goodURL := cfg.Resource.URL.String() | ||
| state["url"] = goodURL | ||
| metrics.resource.Set(goodURL) | ||
| + updateStatus(rep, status.Running, "") | ||
| // On entry, state is expected to be in the shape: | ||
| // | ||
| // { | ||
| @@ -214,18 +214,15 @@ func (i input) run(env v2.Context, src *source, cursor map[string]interface{}, p | ||
| // In addition to this and the functions and globals available | ||
| // from mito/lib, a global, useragent, is available to use | ||
| // in requests. | ||
| - | ||
| - updateStatus(rep, status.Running, "") | ||
| err = periodically(ctx, cfg.Interval, func() error { | ||
| log.Info("process repeated request") | ||
| var ( | ||
| budget = *cfg.MaxExecutions | ||
| waitUntil time.Time | ||
| ) | ||
| + // Keep track of whether CEL is degraded for this periodic run. | ||
| + var isDegraded bool | ||
| for { | ||
| - // keep track if CEL is degraded for this iteration | ||
| - isDegraded := false | ||
| - | ||
| if wait := time.Until(waitUntil); wait > 0 { | ||
| // We have a special-case wait for when we have a zero limit. | ||
| // x/time/rate allow a burst through even when the limit is zero | ||
| @@ -259,8 +256,8 @@ func (i input) run(env v2.Context, src *source, cursor map[string]interface{}, p | ||
| } | ||
| log.Errorw("failed evaluation", "error", err) | ||
| updateStatus(rep, status.Degraded, "failed evaluation: "+err.Error()) | ||
| - isDegraded = true | ||
| } | ||
| + isDegraded = err != nil | ||
| metrics.celProcessingTime.Update(time.Since(start).Nanoseconds()) | ||
| if trace != nil { | ||
| log.Debugw("final transaction", "transaction.id", trace.TxID()) | ||
| diff --git a/x-pack/filebeat/input/cel/integration_test.go b/x-pack/filebeat/input/cel/integration_test.go | ||
| index 38e48991ee..709bb5d7b9 100644 | ||
| --- a/x-pack/filebeat/input/cel/integration_test.go | ||
| +++ b/x-pack/filebeat/input/cel/integration_test.go | ||
| @@ -56,7 +56,7 @@ func TestCheckinV2(t *testing.T) { | ||
| defer svrTwo.Close() | ||
|
|
||
| // allStreams is an elastic-agent configuration with an ES output and one CEL | ||
| - // input with two streams | ||
| + // input with two streams. | ||
| allStreams := []*proto.UnitExpected{ | ||
| { | ||
| Id: "output-unit", | ||
| @@ -138,7 +138,7 @@ func TestCheckinV2(t *testing.T) { | ||
|
|
||
| // oneStream is an elastic-agent configuration with an ES output and one CEL | ||
| // input with one stream. Effectively this is the same as allStreams with | ||
| - // stream cel-cel.cel-1e8b33de-d54a-45cd-90da-ffffffc482e2 removed | ||
| + // stream cel-cel.cel-1e8b33de-d54a-45cd-90da-ffffffc482e2 removed. | ||
| oneStream := []*proto.UnitExpected{ | ||
| { | ||
| Id: "output-unit", | ||
| @@ -204,7 +204,7 @@ func TestCheckinV2(t *testing.T) { | ||
| }, | ||
| } | ||
|
|
||
| - // noStream is an elastic-agent configuration with just an ES output | ||
| + // noStream is an elastic-agent configuration with just an ES output. | ||
| noStream := []*proto.UnitExpected{ | ||
| { | ||
| Id: "output-unit", | ||
| @@ -253,9 +253,9 @@ func TestCheckinV2(t *testing.T) { | ||
| } | ||
| defer server.Stop() | ||
|
|
||
| + // It's necessary to change os.Args so filebeat.Filebeat() can read the | ||
| + // appropriate args at beat.Execute() | ||
| initialOSArgs := os.Args | ||
| - // necessary to change this so filebeat.Filebeat() can read the appropriate args | ||
| - // at beat.Execute() | ||
| os.Args = []string{ | ||
| "filebeat", | ||
| "-E", fmt.Sprintf(`management.insecure_grpc_url_for_testing="localhost:%d"`, server.Port), | ||
| @@ -278,7 +278,7 @@ func TestCheckinV2(t *testing.T) { | ||
| // of units expected for the server to respond with. | ||
| checks := []func(t *testing.T, observed *proto.CheckinObserved) (bool, []*proto.UnitExpected){ | ||
| func(t *testing.T, observed *proto.CheckinObserved) (bool, []*proto.UnitExpected) { | ||
| - // wait for all healthy | ||
| + // Wait for all healthy. | ||
| unitState, payload := extractStateAndPayload(observed, "input-unit-1") | ||
| if unitState != proto.State_HEALTHY { | ||
| return false, allStreams | ||
| @@ -304,7 +304,7 @@ func TestCheckinV2(t *testing.T) { | ||
| return true, allStreams | ||
| }, | ||
| func(t *testing.T, observed *proto.CheckinObserved) (bool, []*proto.UnitExpected) { | ||
| - // wait for one degraded | ||
| + // Wait for one degraded. | ||
| unitState, payload := extractStateAndPayload(observed, "input-unit-1") | ||
| if unitState != proto.State_DEGRADED { | ||
| return false, allStreams | ||
| @@ -329,7 +329,7 @@ func TestCheckinV2(t *testing.T) { | ||
| return true, allStreams | ||
| }, | ||
| func(t *testing.T, observed *proto.CheckinObserved) (bool, []*proto.UnitExpected) { | ||
| - // wait for all degraded | ||
| + // Wait for all degraded. | ||
| unitState, payload := extractStateAndPayload(observed, "input-unit-1") | ||
| if unitState != proto.State_DEGRADED { | ||
| return false, allStreams | ||
| @@ -355,7 +355,7 @@ func TestCheckinV2(t *testing.T) { | ||
| return true, allStreams | ||
| }, | ||
| func(t *testing.T, observed *proto.CheckinObserved) (bool, []*proto.UnitExpected) { | ||
| - // wait for all healthy | ||
| + // Wait for all healthy. | ||
| unitState, payload := extractStateAndPayload(observed, "input-unit-1") | ||
| if unitState != proto.State_HEALTHY { | ||
| return false, allStreams | ||
| @@ -380,7 +380,7 @@ func TestCheckinV2(t *testing.T) { | ||
| return true, allStreams | ||
| }, | ||
| func(t *testing.T, observed *proto.CheckinObserved) (bool, []*proto.UnitExpected) { | ||
| - // wait for all healthy | ||
| + // Wait for all healthy. | ||
| unitState, payload := extractStateAndPayload(observed, "input-unit-1") | ||
| if unitState != proto.State_DEGRADED { | ||
| return false, allStreams | ||
| @@ -436,7 +436,7 @@ func TestCheckinV2(t *testing.T) { | ||
| return true, allStreams | ||
| }, | ||
| func(t *testing.T, observed *proto.CheckinObserved) (bool, []*proto.UnitExpected) { | ||
| - // wait for all healthy | ||
| + // Wait for all healthy. | ||
| unitState, payload := extractStateAndPayload(observed, "input-unit-1") | ||
| if unitState != proto.State_HEALTHY { | ||
| return false, allStreams | ||
| @@ -473,39 +473,24 @@ func TestCheckinV2(t *testing.T) { | ||
| }, | ||
| } | ||
|
|
||
| - timer := time.NewTimer(3 * time.Minute) | ||
| + const wait = 3 * time.Minute | ||
| + timer := time.NewTimer(wait) | ||
| defer timer.Stop() | ||
| - for { | ||
| + for len(checks) == 0 { | ||
| select { | ||
| case observed := <-observedStates: | ||
| matched, expected := checks[0](t, observed) | ||
| - | ||
| expectedUnits <- expected | ||
| - | ||
| - // if not matched, do not proceed to the next check | ||
| if !matched { | ||
| continue | ||
| } | ||
| - | ||
| - // check returned true, so reset the timer | ||
| - timer.Reset(3 * time.Minute) | ||
| - | ||
| - // proceed to the next check | ||
| - if len(checks) > 0 { | ||
| - checks = checks[1:] | ||
| - } | ||
| - | ||
| - // if no more checks, return | ||
| - if len(checks) == 0 { | ||
| - return | ||
| - } | ||
| + timer.Reset(wait) | ||
| + checks = checks[1:] | ||
| case err := <-beatRunErr: | ||
| if err != nil { | ||
| t.Fatalf("beat run err: %v", err) | ||
| } | ||
| case <-timer.C: | ||
| - // a check hasn't returned true for the whole timeout | ||
| - // so fail | ||
| t.Fatal("timeout waiting for checkin") | ||
| } | ||
| } | ||
| @@ -513,7 +498,6 @@ func TestCheckinV2(t *testing.T) { | ||
| } | ||
|
|
||
| func extractStateAndPayload(observed *proto.CheckinObserved, inputID string) (proto.State, map[string]interface{}) { | ||
| - | ||
| for _, unit := range observed.GetUnits() { | ||
| if unit.Id == inputID { | ||
| return unit.GetState(), unit.Payload.AsMap() | ||
| -- | ||
| 2.34.1 | ||
|
|
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters