Move non-web related processing into processor, add "publish" package #1324

roncohen · 2018-08-28T11:16:19Z

This introduces the processor/stream and publish packages and moves relevant code from beater to the new packages.

roncohen · 2018-08-28T11:17:08Z

jenkins, test this please

simitt · 2018-08-28T12:07:24Z

jenkins, test this please

roncohen · 2018-08-28T14:00:53Z

jenkins, test this please

simitt

Appreciate these changes, as they split the v2 handler into several concerns and are more aligned with the current code structure.
I added mainly small comments, except for moving http related information to the processor package, see below.

simitt · 2018-08-28T13:34:01Z

publish/pub.go

+type PendingReq struct {
+	Transformables []transform.Transformable
+	Tcontext       *transform.Context
+	Trace          bool


I'd keep trace private.

utility/request_time.go

simitt · 2018-08-28T14:34:31Z

beater/v2_handler.go

+		if err != nil {
+			sr := stream.Result{}
+			sr.AddWithMessage(stream.ServerError, 1, err.Error())
+			v.sendResponse(logger, w, &sr)


Can you add a test for this, and ensure to return here.

simitt · 2018-08-28T14:50:53Z

beater/v2_handler_test.go

-	}
-	expectedBuf, err := expected.marshal()
+func TestInvalidContentType(t *testing.T) {
+	req, err := http.NewRequest("POST", "/v2/intake", nil)


With the refactoring it becomes more obvious that server Integration tests for v2 are missing. You could add them e.g. to server_test.go, to run current available tests for v1 and v2.

coming up next 👍

simitt · 2018-08-28T14:52:22Z

beater/v2_integration_test.go

@@ -79,7 +82,7 @@ func TestV2IntakeIntegration(t *testing.T) {
 		name := fmt.Sprintf("approved-es-documents/testV2IntakeIntegration%s", test.name)
 		r = r.WithContext(context.WithValue(r.Context(), "name", name))
 		reqTimestamp, err := time.Parse(time.RFC3339, "2018-08-01T10:00:00Z")
-		r = r.WithContext(context.WithValue(r.Context(), requestTimeContextKey, reqTimestamp))
+		r = r.WithContext(utility.ContextWithRequestTime(r.Context(), reqTimestamp))
 		handler.ServeHTTP(w, r)

 		assert.Equal(t, test.status, w.Code)


Can you move this to the processor package. It is a start for what we had in the processor/package tests for v1.

simitt · 2018-08-28T14:52:59Z

decoder/req_decoder.go

-// CompressedRequestReader makes a function that uses information from an http request to construct a Limited ReadCloser
-// from the body of the request, handling any decompression necessary
+// CompressedRequestReader returns a reader that will decompress the body according
+// the supplied Content-Encoding header in the request


I think you are missing a to here.

simitt · 2018-08-28T14:58:24Z

processor/stream/result.go

@@ -131,7 +131,7 @@ func (s *streamResponse) String() string {
 	return strings.Join(errorList, ", ")
 }

-func (s *streamResponse) statusCode() int {
+func (s *Result) StatusCode() int {


The beater package has been the abstraction layer for http so far. I'd avoid moving http related information into the processor package, and would rather define errors here, that are translated to http errors in the beater package (similar to having a ErrFull in the publisher package that gets translated to an http error then in the beater package for v1).

good point. Do you agree that it makes sense to have Result be part of stream but translate it to a http type error in beater ?

yes, that's exactly what I meant!

roncohen · 2018-08-30T09:44:20Z

I moved the v2 integration tests into stream as you suggested and I'm using the approval system for the results now. I also simplified the error response and that made it possible to remove a lot of code 👍

At the moment, the stream result is just json encoded in beater and used as the http response. That means there's still some overlapping concerns, e.g. concerns are not completely separated.

I didn't feel it would makes sense to create a new structure in beater because it would be a copy of Result in stream with some added json tags, so i put the tags directly on the stream.Result. Let me know what you think. @simitt

roncohen · 2018-08-30T10:04:27Z

If it's OK with you, I’m planning to create the v2 server integration tests you asked for in a separate PR. This one is already doing too much.
EDIT: added to the meta issue "Add server-level integration tests to test endpoints"

simitt · 2018-08-31T06:43:20Z

processor/stream/result.go

+	ProcessingTimeoutErrType ErrorType = 2
+	InvalidInputErrType      ErrorType = 3
+	ShuttingDownErrType      ErrorType = 4
+)


If you used a string here e.g. const ( QueuFullErrType = "QueueFullErr" .. ) it would be pretty easy to return an error type to the agents.

In case you want to stick to having int I suggest following:

type StreamError int const( QueueFullErrType StreamError = iota ProcessingTimeoutErrType InvalidInputErrType ShuttingDownErrType )

simitt · 2018-08-31T07:03:30Z

processor/stream/result.go

+}
+
+func (r *Result) LimitedAdd(err error) {
+	if len(r.Errors) < errorsLimit {


This changes the logic of keeping up to n errors for every error type, and which http status is returned more or less gets random, based on which errors come first.

I would internally keep a map[ErrorType][]error or a map[ErrorType]error struct. This would also allow to not looping over all values when figuring out the highest http.status_code to return but check for available keys.

true that it's random now. I considered it, and went with this because it was what we agreed, its simple and likely good enough for a start. Your suggestion would require us to decide on which errors should take precedence.

Looping over 5 items is not really a problem

to be clear, I'm happy to discuss this and i agree it would be better to make sure some specific errors are included, but i think we should move forward with this behavior at this time

simitt · 2018-08-31T07:06:25Z

processor/stream/approved-stream-result/testIntegrationResultInvalidEvent.approved.json

+            "message": "Problem validating JSON document against schema: I[#] S[#] doesn't validate with \"transaction#\"\n  I[#] S[#/allOf/1] allOf failed\n    I[#/id] S[#/allOf/1/properties/id/type] expected string, but got number"
+        }
+    ]
+}


You already mentioned you prefer adding tests later. Please keep on your list to add more exhaustive error returning testing, as here only one invalid json error is tested.
I can imagine it will be hard to have a good overview over what has been tested and what not as the changes are not in the same PR.

simitt · 2018-08-31T07:35:31Z

processor/stream/stream_processor.go

+			return nil, &Error{
+				Type:     InvalidInputErrType,
+				Message:  e.Error(),
+				Document: string(reader.LastLine()),


not a native - but shouldn't this be LatestLine?

simitt · 2018-08-31T07:53:49Z

processor/stream/stream_processor.go

+				response.LimitedAdd(&Error{
+					Type:     InvalidInputErrType,
+					Message:  err.Error(),
+					Document: string(reader.LastLine()),


Why not add the rawModel here as this lead to the error?

rawModel is a map[string]interface{} here, not a string or []byte.

simitt · 2018-08-31T07:55:01Z

processor/stream/stream_processor.go

+}
+
+// readBatch will read up to `batchSize` objects from the ndjson stream
+// it returns a slice of eventables, a serverResponse and a bool that indicates if we're at EOF.


That's actually not true, the serverResponse is not returned. I think it would be better design though if the response were created in readBatch and returned and the caller then adds it to the overall response.

ah yes, that comment didn't get updated :)

Returning the response instead of taking a reference would requires us to merge the two responses every time readBatch returns. There's no merging code atm. Do you think it's worth it?

Errors should be edge cases. Thus, I don't expect a big performance impact one way or the other. I personally find it cleaner, but no strong opinion.

I've updated the comment now. I prefer to leave the arguments for now, but happy to discuss again at a later point.

simitt · 2018-08-31T08:03:39Z

processor/stream/stream_processor.go

+						Type:    ShuttingDownErrType,
+						Message: "server is shutting down",
+					})
+				case publish.ErrFull:


As roncohen#3 has not been merged, I guess this will need to be implemented again after merging this PR.

I needed to refactor this area, so your PR wouldn't apply cleanly anymore. During the refactor i changed the behavior in this PR so that it returns when the queue is full, like in your PR.

roncohen · 2018-08-31T09:12:06Z

thanks for the thorough review @simitt !

…elastic#1324) also simplify stream error handling.

…ckage (elastic#1324) also simplify stream error handling.

…elastic#1324) also simplify stream error handling.

…ckage (elastic#1324) also simplify stream error handling.

…ckage (#1324) also simplify stream error handling.

roncohen force-pushed the v2-move-to-processing branch 2 times, most recently from 33f14da to 4f2d4ff Compare August 28, 2018 12:32

roncohen force-pushed the v2-move-to-processing branch from 4f2d4ff to 3042571 Compare August 28, 2018 14:19

simitt reviewed Aug 28, 2018

View reviewed changes

zube bot added the [zube]: Inbox label Aug 30, 2018

roncohen force-pushed the v2-move-to-processing branch 2 times, most recently from 4a26547 to c41bf75 Compare August 30, 2018 09:38

roncohen force-pushed the v2-move-to-processing branch from c41bf75 to a3c40f1 Compare August 30, 2018 10:01

roncohen force-pushed the v2 branch from 291b5ab to 02437d4 Compare August 30, 2018 11:03

Move non-web related processing into processor, add "publish" package

b86d869

roncohen force-pushed the v2-move-to-processing branch from a3c40f1 to 83bfc53 Compare August 30, 2018 12:20

Fixes according to review

8748d73

roncohen force-pushed the v2-move-to-processing branch from 83bfc53 to 8748d73 Compare August 30, 2018 12:34

roncohen mentioned this pull request Aug 30, 2018

Intake v2 support #1237

Closed

30 tasks

simitt reviewed Aug 31, 2018

View reviewed changes

Ron cohen added 3 commits August 31, 2018 10:47

Another round of fixes

422b0a0

Allow some errors to be added forcefully above the 5 item limit

0b5f841

Update readBatch comment

30d8c62

simitt approved these changes Aug 31, 2018

View reviewed changes

roncohen merged commit c338927 into elastic:v2 Aug 31, 2018

roncohen deleted the v2-move-to-processing branch August 31, 2018 09:12

zube bot added [zube]: Done and removed [zube]: Inbox labels Aug 31, 2018

simitt mentioned this pull request Sep 5, 2018

v2: Investigate "full queue" behavior #1298

Closed

simitt pushed a commit to simitt/apm-server that referenced this pull request Sep 7, 2018

Move non-web related processing into processor, add "publish" package (…

51757b1

…elastic#1324) also simplify stream error handling.

roncohen added a commit to roncohen/apm-server that referenced this pull request Oct 7, 2018

Move non-web related processing into processor, add "publish" package (…

1d6668c

…elastic#1324) also simplify stream error handling.

roncohen added a commit to roncohen/apm-server that referenced this pull request Oct 15, 2018

Move non-web related processing into processor, add "publish" package (…

40d8f35

…elastic#1324) also simplify stream error handling.

roncohen added a commit to roncohen/apm-server that referenced this pull request Oct 15, 2018

[v2] Move non-web related processing into processor, add "publish" pa…

cf0add4

…ckage (elastic#1324) also simplify stream error handling.

roncohen added a commit to roncohen/apm-server that referenced this pull request Oct 15, 2018

[v2] Move non-web related processing into processor, add "publish" pa…

e55c3d6

…ckage (elastic#1324) also simplify stream error handling.

roncohen added a commit to roncohen/apm-server that referenced this pull request Oct 15, 2018

Move non-web related processing into processor, add "publish" package (…

6fd8127

…elastic#1324) also simplify stream error handling.

roncohen added a commit to roncohen/apm-server that referenced this pull request Oct 15, 2018

[v2] Move non-web related processing into processor, add "publish" pa…

75683ba

…ckage (elastic#1324) also simplify stream error handling.

roncohen added a commit to roncohen/apm-server that referenced this pull request Oct 16, 2018

[v2] Move non-web related processing into processor, add "publish" pa…

b9282ed

…ckage (elastic#1324) also simplify stream error handling.

roncohen added a commit that referenced this pull request Oct 16, 2018

[v2] Move non-web related processing into processor, add "publish" pa…

ea98da8

…ckage (#1324) also simplify stream error handling.

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Move non-web related processing into processor, add "publish" package #1324

Move non-web related processing into processor, add "publish" package #1324

roncohen commented Aug 28, 2018 •

edited

Loading

roncohen commented Aug 28, 2018

simitt commented Aug 28, 2018

roncohen commented Aug 28, 2018

simitt left a comment

simitt Aug 28, 2018

simitt Aug 28, 2018

simitt Aug 28, 2018

roncohen Aug 30, 2018

simitt Aug 28, 2018

simitt Aug 28, 2018

simitt Aug 28, 2018

roncohen Aug 28, 2018

simitt Aug 28, 2018

roncohen commented Aug 30, 2018 •

edited

Loading

roncohen commented Aug 30, 2018 •

edited

Loading

simitt Aug 31, 2018

simitt Aug 31, 2018

roncohen Aug 31, 2018 •

edited

Loading

roncohen Aug 31, 2018 •

edited

Loading

simitt Aug 31, 2018

simitt Aug 31, 2018

simitt Aug 31, 2018

roncohen Aug 31, 2018 •

edited

Loading

simitt Aug 31, 2018

roncohen Aug 31, 2018

simitt Aug 31, 2018

roncohen Aug 31, 2018

simitt Aug 31, 2018

roncohen Aug 31, 2018

roncohen commented Aug 31, 2018

Move non-web related processing into processor, add "publish" package #1324

Move non-web related processing into processor, add "publish" package #1324

Conversation

roncohen commented Aug 28, 2018 • edited Loading

roncohen commented Aug 28, 2018

simitt commented Aug 28, 2018

roncohen commented Aug 28, 2018

simitt left a 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

roncohen commented Aug 30, 2018 • edited Loading

roncohen commented Aug 30, 2018 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

roncohen Aug 31, 2018 • edited Loading

Choose a reason for hiding this comment

roncohen Aug 31, 2018 • edited Loading

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

roncohen Aug 31, 2018 • edited Loading

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

roncohen commented Aug 31, 2018

roncohen commented Aug 28, 2018 •

edited

Loading

roncohen commented Aug 30, 2018 •

edited

Loading

roncohen commented Aug 30, 2018 •

edited

Loading

roncohen Aug 31, 2018 •

edited

Loading

roncohen Aug 31, 2018 •

edited

Loading

roncohen Aug 31, 2018 •

edited

Loading