[Filebeat][httpjson] Convert httpjson input to a v2 input #20226

marc-gr · 2020-07-24T09:24:38Z

What does this PR do?

Converts httpjson input to the new v2 stateless input.

Split old input in several new objects for easier understanding
Refactored go tests to simplify them and make them work with the new input

Why is it important?

Several modules rely on httpjson input, and would benefit a lot from the new v2 cursor feature that keeps persistence of a cursor between restarts.

This change moves httpjson input to new v2 input, preparing to be able to implement the cursor persistence state in a following change.

Checklist

My code follows the style guidelines of this project
I have commented my code, particularly in hard-to-understand areas
~~- [] I have made corresponding changes to the documentation~~
~~- [] I have made corresponding change to the default configuration files~~
I have added tests that prove my fix is effective or that my feature works
I have added an entry in CHANGELOG.next.asciidoc or CHANGELOG-developer.next.asciidoc.

Related issues

Relates to #19486

elasticmachine · 2020-07-24T09:24:40Z

Pinging @elastic/siem (Team:SIEM)

elasticmachine · 2020-07-24T09:29:04Z

💚 Build Succeeded

Expand to view the summary

Build stats

Build Cause: [Pull request #20226 updated]
Start Time: 2020-08-19T14:39:22.929+0000
Duration: 51 min 9 sec

Test stats 🧪

Test	Results
Failed	0
Passed	2474
Skipped	388
Total	2862

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

andrewkroh · 2020-07-28T12:53:02Z

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

+	}
+}
+
+func (l *retryLogger) Printf(s string, args ...interface{}) {


This is more of an enhancement request. There's a separate logger interface that can be implemented that might integrate more nicely with our logger (not 100% sure). With this interface each message contains a level like [INFO], I wonder if we switch to the leveled interface if the message would look more "native".

I think you were referring to implement the LeveledLogger, I did so, let me know if you meant something different.

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

andrewkroh · 2020-07-28T13:01:36Z

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


-	if mm != nil && in.config.DateCursor.IsEnabled() {
-		in.advanceCursor(common.MapStr(mm))
+	_, err = net.DialTimeout("tcp", fmt.Sprintf("%s:%s", url.Hostname(), port), time.Second)


Should this use net.JoinHostPort to ensure IPv6 addresses are boxed? I think url.Hostname() strips brackets.

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

urso · 2020-07-28T18:41:03Z

x-pack/filebeat/tests/system/test_httpjson.py

+from filebeat import BaseTest
+
+
+class Test(BaseTest):


why replace the go based tests with a python test? This 'complicates', potentially slowing down the CI, plus increases the risk of introducing flaky tests.

If we want to switch to python based testing, maybe we can make this change in a separate PR?

I had some issues trying to spawn the cursor input manager in unit tests, that was the reason why I moved the integration tests to python. Since your following comments made a lot of sense, and I implemented the stateless input instead as a first step, I moved them back to go now, since the issues I faced with the cursor input seemed to not apply with that one.

urso · 2020-07-28T18:48:22Z

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

+			StateStore: store,
+			Type:       inputName,
+			Configure:  configure,
+		},


Using the InputManager like this will store some state always in the statestore. The httpjson input is quite flexible, as it allows configurations that allow inputs to optionally store state or not.

I'd recommend to create a custom input manager that wraps the stateless and the cursor input managers. When configuring the input the wrapper would choose the right input manager instance, based on the user configuration.

Maybe we can split the effort by first converting the input to the v2 api using the stateless input manager?

Make sense, did so 👍

urso · 2020-07-28T19:02:57Z

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

-		workerCtx:    workerCtx,
-		workerCancel: workerCancel,
-	}
+	in := &httpJSONInput{}


It looks the requester does too much. It is a 'source', unpacks from the cursor, is a client, and executes the request. Plus the requester gets only partially initialized here. The source is used to allow to generate the key in the key value store, and to pass the input the address where to collect from. No need to confiure an http client yet. Configure should focus on unpacking the config, do not create helper objects that are required for data collection yet.

Following the code your source is the URL only. Maybe you want to consider to add other metadata to the source as well, e.g. parameters, username (in case services return different information based on the login).

Pass the config to the httpJSONInput, not your source. The requester can be created in the "Run" method.

I agree, tried to separate pagination, cursor and retry logic out of it, lmk if it still confusing.

urso · 2020-07-28T19:10:49Z

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

+	config      config
+	client      *http.Client
+	cursorValue string
+}


this type does a lot. It implements rate limiting, request execution, response parsing+event creation, and publishing. It looks like it is very difficult to test. Can we split some functionality into a dedicated type and introduce unit testing? At least for rate limiting consider to use/introduce a third party library.

For the rate limit bit, it was not in the scope of this change, but it makes sense, will open a new issue to make it separately if that makes sense.

marc-gr · 2020-07-31T10:15:45Z

Following @urso advice I pulled back and implemented stateless input instead. Will let the work of moving to a custom input manager to be able to use also the cursor input for a following pull request. This allowed to go back to go tests since the issues I faced when writing them for the cursor input do not apply for this one.

At the same time did some refactoring to simplify input/requester logic. More improvements can be done but I think would be nice to save them for the following refactor of the input that we plan to do next.

Updated the description to make it describe what it currently does.

urso · 2020-07-31T13:59:42Z

This allowed to go back to go tests since the issues I faced when writing them for the cursor input do not apply for this one.

Maybe let's discuss this offline, but it would be nice to learn where you did struggle with testing. In libbeat we have helpers for mocking beat.Pipeline, and beat.Client. The libbeat/statestore/storetest package provides helpers to construct an in memory store that can be checked (state is stored in a map[string]interface{}). For example check out filebeat/input/v2/input-cursor/store_test.go. We got many expensive python testing for functionality that should be unit or integration tested only. If you find it difficult with existing helpers to write test, let's talk about it to find what is missing. All in all I'd like to encourage devs to write go based tests.

urso · 2020-07-31T14:02:59Z

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

+
+	url.RawQuery = q.Encode()
+
+	return url.String()


why return the string instead of the already initialized URL struct?

urso · 2020-07-31T14:09:59Z

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

+			teardown: func(i interface{}) {
+				server := i.(*httptest.Server)
+				server.Close()
+			},


do we ever return some other type but an http server?

teardown as is can be removed. Better pass testing.T to setup. the close can be registered with the test via t.Cleanup.

For this tests we end up with 2 active servers. how about only providing a setupServer method? then ssl, setup, and teardown can be replaced by a single replacable function.

urso · 2020-07-31T14:12:57Z

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

-			t.Fatal(err)
-		}
-	})
+type publisher struct {


The libbeat/publisher/testing might already contain similar helpers.

urso · 2020-07-31T14:26:33Z

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

-				in.log.Info("Process another repeated request.")
-				err = in.processHTTPRequest(ctx, client, ri)
+				log.Info("Process another repeated request.")
+				err = requester.processHTTPRequest(stdCtx, publisher)
 				if err != nil {
 					return err
 				}
 			}
 		}
 	}


This loop is quite a common pattern. Better use timed.Periodic(stdCtx, in.Config.Interval, func() { ... }). Using this pattern to check for cancellation in a loop one should add an extra select on the done channel only in order to confirm done has not been closed before doing the actual select statement.

Periodic is defined here: https://github.com/elastic/go-concert/blob/master/timed/timed.go#L57

urso · 2020-07-31T14:28:35Z

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

-		return nil, nil
+func newHTTPJSONInput(config config) (*httpJSONInput, error) {
+	if err := config.Validate(); err != nil {
+		return nil, err
 	}


note: Validate is already called by Unpack. But I like it to be called here again, as the constructor should be independent of configure.

urso · 2020-08-13T14:54:27Z

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

+			log.Info("Process another repeated request.")
+			err = requester.processHTTPRequest(stdCtx, publisher)
+			if err != nil {
+				periodCancel()


the Periodic function will support error returns to stop the loop once #20590 is merged.

- Paginator takes care of requesting next page info - Rate limiter takes care of rate limiting requests - Date cursor takes care of keeping track of cursor state

urso · 2020-08-19T12:27:40Z

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

-		return in.config.URL
-	}
+// Run starts the input worker then returns. Only the first invocation
+// will ever start the worker.


This comment seems to be wrong.

urso · 2020-08-19T12:30:13Z

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


-	url, err := url.Parse(in.config.URL)
+	httpClient, err := in.newHTTPClient(stdCtx)


how can this fail? Does the client make a request to init oauth support? Are there recoverable error types (e.g. io timeout)? In the later case we might want to put the initialization into a loop with exponential backoff.

urso · 2020-08-19T12:31:53Z

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

-		in.log.Warn("date_cursor field must be a string, cursor will not advance")
-		return
+	if in.config.Interval == 0 {
+		return nil


Is Interval==0 a configuration error? Maybe this is exactly what some users want, (long-)poll the server.

urso · 2020-08-19T12:33:11Z

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

-				}
-			}
-		}
+	switch errors.Cause(err) {


Cause is used by some legacy error libs (e.g. pkg/errors which is not maintained anymore). Better use Unpack or errors.Is(context.X).

urso · 2020-08-19T12:34:31Z

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

-			}
-		}
+	switch errors.Cause(err) {
+	case context.Canceled, context.DeadlineExceeded:


is DeadlineExceeded a signal that might be triggered internally by processHTTPRequest, or do we expect this to be triggered via the inputs parent context?

urso · 2020-08-19T12:40:40Z

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

+				}
+				msg, err := r.client.Do(req)
+				if err != nil {
+					return nil, errors.Wrapf(err, "failed to execute http client.Do")


It looks like this error is passed through to Run. If so the input shuts down after an IO error. Timeouts/disconnects or other network errors should not shut down the input. These are recoverable errors in a sense of "let's try again".

errors.Wrap and similar make it difficult to "judge" the error (errors.Is can help to check for a cause). Consider to either log recoverable errors or create special error types that allow you to see what's going on.

About errors.Wrap: We should stop using the pkg/errors package. It is discontinued. At the same time go 1.14 introduced something similar via fmt.Errorf with %w. Better use fmt.Errorf("failed to execute http request: %w", err)

urso · 2020-08-19T12:41:10Z

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

+
+		responseData, err := ioutil.ReadAll(resp.Body)
+		if err != nil {
+			return errors.Wrapf(err, "failed to read http response")


IO error here should not force the input to shutdown.

) * Convert httpjson input to a v2 cursor input * Add CHANGELOG entry * Fix format errors * Convert to stateless input and refactor: - Paginator takes care of requesting next page info - Rate limiter takes care of rate limiting requests - Date cursor takes care of keeping track of cursor state * Remove python tests * Do not fail if there is no next page * Refactor go integration tests to work with v2 input * Do suggested changes to input and tests * Update time.Periodic call with error return * Change test duration values * Sepparate sync test case * Create custon url config type * Change input.Run comment * Change input.Run to only return on context cancellation * Remove all usages of pkg/errors (cherry picked from commit 929e838)

…20702) * Convert httpjson input to a v2 cursor input * Add CHANGELOG entry * Fix format errors * Convert to stateless input and refactor: - Paginator takes care of requesting next page info - Rate limiter takes care of rate limiting requests - Date cursor takes care of keeping track of cursor state * Remove python tests * Do not fail if there is no next page * Refactor go integration tests to work with v2 input * Do suggested changes to input and tests * Update time.Periodic call with error return * Change test duration values * Sepparate sync test case * Create custon url config type * Change input.Run comment * Change input.Run to only return on context cancellation * Remove all usages of pkg/errors (cherry picked from commit 929e838)

) * Convert httpjson input to a v2 cursor input * Add CHANGELOG entry * Fix format errors * Convert to stateless input and refactor: - Paginator takes care of requesting next page info - Rate limiter takes care of rate limiting requests - Date cursor takes care of keeping track of cursor state * Remove python tests * Do not fail if there is no next page * Refactor go integration tests to work with v2 input * Do suggested changes to input and tests * Update time.Periodic call with error return * Change test duration values * Sepparate sync test case * Create custon url config type * Change input.Run comment * Change input.Run to only return on context cancellation * Remove all usages of pkg/errors

marc-gr added enhancement review Filebeat Filebeat Team:SIEM v7.10.0 labels Jul 24, 2020

botelastic bot added needs_team Indicates that the issue/PR needs a Team:* label and removed needs_team Indicates that the issue/PR needs a Team:* label labels Jul 24, 2020

marc-gr force-pushed the httpjson_filebeat_v2 branch from 4f05073 to 01e904e Compare July 27, 2020 13:05

marc-gr requested a review from a team July 27, 2020 13:06

marc-gr force-pushed the httpjson_filebeat_v2 branch from 01e904e to 995608a Compare July 28, 2020 06:54

andrewkroh added the Project:Filebeat-Input-v2 label Jul 28, 2020

andrewkroh requested a review from urso July 28, 2020 12:29

andrewkroh reviewed Jul 28, 2020

View reviewed changes

urso reviewed Jul 28, 2020

View reviewed changes

marc-gr force-pushed the httpjson_filebeat_v2 branch from 995608a to 7b92ab9 Compare July 31, 2020 10:06

marc-gr changed the title ~~[Filebeat][httpjson] Convert httpjson input to a v2 cursor input~~ [Filebeat][httpjson] Convert httpjson input to a v2 input Jul 31, 2020

marc-gr requested review from urso and andrewkroh July 31, 2020 10:20

marc-gr mentioned this pull request Jul 31, 2020

[Filebeat][httpjson] Substitute custom rate limit implementation for a third party #20379

Closed

urso reviewed Jul 31, 2020

View reviewed changes

marc-gr force-pushed the httpjson_filebeat_v2 branch from 7b92ab9 to 3ba6bda Compare July 31, 2020 15:46

marc-gr requested a review from urso July 31, 2020 15:48

urso reviewed Aug 13, 2020

View reviewed changes

andrewkroh mentioned this pull request Aug 13, 2020

[Filebeat] Okta Module Crashing Frequently #18530

Closed

marc-gr force-pushed the httpjson_filebeat_v2 branch 2 times, most recently from 317973d to b4a2f08 Compare August 18, 2020 07:27

marc-gr added 10 commits August 19, 2020 11:24

Fix format errors

9cb808b

Convert to stateless input and refactor:

b21108f

- Paginator takes care of requesting next page info - Rate limiter takes care of rate limiting requests - Date cursor takes care of keeping track of cursor state

Remove python tests

5bd77e9

Do not fail if there is no next page

723d87c

Refactor go integration tests to work with v2 input

73a5add

Do suggested changes to input and tests

0b7d9d4

Update time.Periodic call with error return

320610b

Change test duration values

852a344

Sepparate sync test case

200daac

Create custon url config type

25234dd

marc-gr force-pushed the httpjson_filebeat_v2 branch from d23bdc6 to 25234dd Compare August 19, 2020 09:25

urso reviewed Aug 19, 2020

View reviewed changes

Change input.Run comment

f18833d

urso reviewed Aug 19, 2020

View reviewed changes

marc-gr added 2 commits August 19, 2020 16:36

Change input.Run to only return on context cancellation

e5d5be5

Remove all usages of pkg/errors

dbe98a1

urso approved these changes Aug 19, 2020

View reviewed changes

marc-gr merged commit 929e838 into elastic:master Aug 20, 2020

marc-gr deleted the httpjson_filebeat_v2 branch August 20, 2020 07:07

marc-gr mentioned this pull request Aug 20, 2020

Cherry-pick #20226 to 7.x: [Filebeat][httpjson] Convert httpjson input to a v2 input #20702

Merged

4 tasks

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[Filebeat][httpjson] Convert httpjson input to a v2 input #20226

[Filebeat][httpjson] Convert httpjson input to a v2 input #20226

marc-gr commented Jul 24, 2020 •

edited

Loading

elasticmachine commented Jul 24, 2020

elasticmachine commented Jul 24, 2020 •

edited by jenkins-beats-ci bot

Loading

Build stats

Test stats 🧪

andrewkroh Jul 28, 2020

marc-gr Jul 31, 2020

andrewkroh Jul 28, 2020

urso Jul 28, 2020

marc-gr Jul 31, 2020

urso Jul 28, 2020

marc-gr Jul 31, 2020

urso Jul 28, 2020

marc-gr Jul 31, 2020

urso Jul 28, 2020

marc-gr Jul 31, 2020

marc-gr commented Jul 31, 2020 •

edited

Loading

urso commented Jul 31, 2020

urso Jul 31, 2020

urso Jul 31, 2020

urso Jul 31, 2020

urso Jul 31, 2020

urso Jul 31, 2020

urso Aug 13, 2020

urso Aug 19, 2020

urso Aug 19, 2020

urso Aug 19, 2020

urso Aug 19, 2020

urso Aug 19, 2020

urso Aug 19, 2020

urso Aug 19, 2020


		url, err := url.Parse(in.config.URL)
		httpClient, err := in.newHTTPClient(stdCtx)

[Filebeat][httpjson] Convert httpjson input to a v2 input #20226

[Filebeat][httpjson] Convert httpjson input to a v2 input #20226

Conversation

marc-gr commented Jul 24, 2020 • edited Loading

What does this PR do?

Why is it important?

Checklist

Related issues

elasticmachine commented Jul 24, 2020

elasticmachine commented Jul 24, 2020 • edited by jenkins-beats-ci bot Loading

💚 Build Succeeded

Build stats

Test stats 🧪

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

marc-gr commented Jul 31, 2020 • edited Loading

urso commented Jul 31, 2020

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

marc-gr commented Jul 24, 2020 •

edited

Loading

elasticmachine commented Jul 24, 2020 •

edited by jenkins-beats-ci bot

Loading

marc-gr commented Jul 31, 2020 •

edited

Loading