http_endpoint: Allow receiving multiple documents on a single request

[adriansr]

What does this PR do?

Updates Filebeat's http_endpoint to support receiving multiple documents from a single POST request.

Until now it only accepted a single document (JSON object) per request.

With this PR:

• If the body is an array of objects the input will emit each object separately.
• If the body is in NDJSON format, the input will emit each object in the stream.

Why is it important?

Minimizes the number of requests for high-volume ingestion.

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.

[elasticmachine]

Pinging @elastic/security-external-integrations (Team:Security-External Integrations)

[elasticmachine]

💚 Build Succeeded

the below badges are clickable and redirect to their specific view in the CI or DOCS

Expand to view the summary

Build stats

• Build Cause: Pull request #25764 updated

• Start Time: 2021-06-03T10:03:06.062+0000

• Duration: 102 min 33 sec

• Commit: 6e3e4ae

Test stats 🧪

Test	Results
Failed	0
Passed	7311
Skipped	1193
Total	8504

Trends 🧪

💚 Flaky test report

Tests succeeded.

Expand to view the summary

Test stats 🧪

Test	Results
Failed	0
Passed	7311
Skipped	1193
Total	8504

[leehinman]

I might be misreading this, but could we just have one function? I think with json.NewDecoder we could handle:

{"a": 1}
or
{"a": 1}{"b":2}
or
{"a":1}
{"b":2}
or
[{"a":1},{"b":2}]

and then json or ndjson would both work.

[leehinman]

Something like this:

func httpReadJSON(body io.Reader) (objs []common.MapStr, status int, err error) {
	if body == http.NoBody {
		return nil, http.StatusNotAcceptable, errBodyEmpty
	}

	decoder := json.NewDecoder(body)
	for idx := 0; ; idx++ {
		var obj interface{}
		if err := decoder.Decode(&obj); err != nil {
			if err == io.EOF {
				break
			}
			return nil, http.StatusBadRequest, errors.Wrapf(err, "malformed JSON object at stream position %d", idx)
		}
		switch v := obj.(type) {
		case map[string]interface{}:
			objs = append(objs, v)
		case []interface{}:
			for listIdx, listObj := range v {
				asMap, ok := listObj.(map[string]interface{})
				if !ok {
					return nil, http.StatusBadRequest, fmt.Errorf("%v at obj index %d, list index %d", errUnsupportedType, idx, listIdx)
				}
				objs = append(objs, asMap)
			}
		default:
			return nil, http.StatusBadRequest, errUnsupportedType
		}
	}
	return objs, 0, nil
}

I can push it to the PR if you want.

[mergify]

This pull request is now in conflicts. Could you fix it? 🙏
To fixup this pull request, you can check out it locally. See documentation: https://help.github.com/articles/checking-out-pull-requests-locally/

git fetch upstream
git checkout -b http_input_multi_doc upstream/http_input_multi_doc
git merge upstream/master
git push upstream http_input_multi_doc

[leehinman]

In case 2 function approach is best fit.

[leehinman]

LGTM

[adriansr]

/test

[mergify]

This pull request is now in conflicts. Could you fix it? 🙏
To fixup this pull request, you can check out it locally. See documentation: https://help.github.com/articles/checking-out-pull-requests-locally/

git fetch upstream
git checkout -b http_input_multi_doc upstream/http_input_multi_doc
git merge upstream/master
git push upstream http_input_multi_doc