added streaming remote read client side functionality #7354

Sudhar287 · 2020-06-06T04:59:48Z

For #5926
Thank you to @YaoZengzeng for his initial design. I've implemented his comments from here.
FYI @bwplotka

bwplotka

This actually looks quite amazing (: I suggested couple of things but with some more work this will work well 💪 Thanks!

bwplotka · 2020-06-10T18:06:53Z

storage/remote/client.go

@@ -168,8 +172,7 @@ func (c *Client) Read(ctx context.Context, query *prompb.Query) (*prompb.QueryRe
 	httpReq.Header.Set("User-Agent", userAgent)
 	httpReq.Header.Set("X-Prometheus-Remote-Read-Version", "0.1.0")

-	ctx, cancel := context.WithTimeout(ctx, c.timeout)
-	defer cancel()
+	httpResp, err := c.client.Do(httpReq)


why we use request before we put context into it? (:

storage/remote/client.go

storage/remote/codec.go

storage/remote/codec_test.go

storage/remote/codec.go

bwplotka

I think you asumption is right, I commented where you can find the answer (just giving you an answer would not teach you anything ❤️ )

One suggestion: Please do not leave dead or unfinished code you don't want to merge in a PR. Add your comments as GitHub comment. Otherwise it's hard to review something which is .. not ready. So there is nothing to approve / review really (: Making sure you keep PR reviewable / ready to go even if blocked, makes sure you will get it merged sooner! (:

I think assumption of @wwformat was totally alright. Did you try that before reaching me on slack? 😉

storage/remote/codec.go

Sudhar287 · 2020-07-10T17:59:12Z

Sorry I didn't try @wwformat suggestion before reaching you on slack because I thought using &prompb.ChunkedSeries is right as no objections about it were raised on my previous PR. But now I think I should have done it. Sorry!

Sudhar287 · 2020-07-11T23:56:49Z

Actively working on the fix and some issues with TestEndpoints. Will try to get the revision within a week. :-)

bwplotka

Wow!

This is actually high-quality code now 💪 Thanks for this. 🎉

I am approving, but let's merge when last comments will be addressed 🤗

storage/remote/client.go

tsdb/tsdbutil/buffer.go

storage/remote/codec_test.go

Sudhar287

Thanks. PTAL.

Sudhar287 · 2020-07-22T03:28:45Z

storage/remote/client.go

+
+	if httpResp.StatusCode/100 != 2 {
+		io.Copy(ioutil.Discard, httpResp.Body)
+		httpResp.Body.Close()


I hope this is what you meant.

Why not having:

defer func() { io.Copy(ioutil.Discard, httpResp.Body) httpResp.Body.Close() }()

here?

Really sorry Bartek. Missed the notification.

Still not addressed (:

still not addressed, in case below with unsupported contentType the response body will be not closed/exhausted leaving the connection alive without good reason.

To solve the problem with the unsupported contentType, is the additional line here enough to solve the issue?

if !strings.HasPrefix(contentType, "application/x-streamed-protobuf; proto=prometheus.ChunkedReadResponse") { --> httpResp.Body.Close() return nil, errors.Errorf("not supported remote read content type: %s", contentType) }

There is one easy fix here. Just defer all (exhaust + close body) once (: and that's it. And don't double close in handleStreamedResponse (: WDYT?

Hmm, I considered the above option but won't it be incorrect?
for eg: We're going to return return &chunkedResponseSeriesSet{stream: stream, httpResp: httpResp}, nil in StartRead() and thus if we have

defer func() { io.Copy(ioutil.Discard, httpResp.Body) httpResp.Body.Close() }()

before the return &chunkedResponseSeriesSet{stream: stream, httpResp: httpResp}, nil we will end up closing the http connection that is needed for stream.NextProto(res).
From my understanding, the most straightforward option seems to be to close the HttpRest individually/separately depending on the use case.

well there is better way. You are right that we cannot do it for the case when we return iterator. So in this case I would do this.:

defer func() { if err != nil { io.Copy(ioutil.Discard, httpResp.Body) httpResp.Body.Close() } }()

Then make sure the function returns (<something>, err error) and it will only close on error (:

I think you mean return streamedResponse, streamedResponse.err This makes sense to me. (I have a small clarifying question though)
But, I can see that won't be returning an error when we see an io.EOF in the chunkedResponseSeriesSet Next(). The connection needs to be closed in this case too, right?
Should the Next() be replaced from

if err := c.stream.NextProto(res); err != nil { if err != io.EOF { c.err = err } return false }

Replaced to

if err := c.stream.NextProto(res); err != nil { c.err = err return false }

This above approach produces some errors like:
api_test.go:1784: Unexpected error: execution: EOF
Or is this the best version:

if err := c.stream.NextProto(res); err != nil { httpResp.Body.Close() if err != io.EOF { c.err = err } return false }

bwplotka

Some important comments not addressed, and still, there is a way this code will leak resources.
):

But super close!

bwplotka · 2020-07-27T14:08:02Z

storage/remote/client.go

+
+	if httpResp.StatusCode/100 != 2 {
+		io.Copy(ioutil.Discard, httpResp.Body)
+		httpResp.Body.Close()


Why not having:

defer func() { io.Copy(ioutil.Discard, httpResp.Body) httpResp.Body.Close() }()

here?

storage/remote/codec_test.go

bwplotka

Stil potential leak place let's fix (:

bwplotka · 2020-08-04T12:38:00Z

storage/remote/client.go

+	}
+
+	if httpResp.StatusCode/100 != 2 {
+		defer func() {


this does not help.

Please look on this function and think about what are the cases when we need to close the body and exhaust? I think you are missing one.

I think because we run functions it's totally ok to defer this in all cases JUST after we get response. and let's make sure nothing closes twice inside functions WDYT?

Thanks for taking a look at my code Bartek.
I think you are referring to closing the httpRest.Body over here:

func (c *client) handleStreamedResponse(httpResp *http.Response) (storage.SeriesSet, error) { stream := NewChunkedReader(httpResp.Body, DefaultChunkedReadLimit, nil) return &chunkedResponseSeriesSet{stream: stream}, nil }

So I have to close() the bufio.Reader in ChunkedReader?

Suggested something here: #7354 (comment)

bwplotka

I see two things missing. Otherwise LGTM!

storage/remote/codec.go

bwplotka

👋

Let's get back to this if you want (: Added a few comments, otherwise good 👍 Also some old comments are still not addressed (again). Just letting you know that it's natural that if maintainers see that their comments are not addressed (nothing done, nothing responded, nothing adjusted, just ignored) it might mean that ther time is being wasted. If you wish to contribute and have PR merged fasted, you need to keep attention to details like this. (:

bwplotka · 2020-09-11T09:23:06Z

storage/remote/client.go

+
+	if httpResp.StatusCode/100 != 2 {
+		io.Copy(ioutil.Discard, httpResp.Body)
+		httpResp.Body.Close()


still not addressed, in case below with unsupported contentType the response body will be not closed/exhausted leaving the connection alive without good reason.

bwplotka · 2020-09-11T09:24:34Z

storage/remote/client.go


-	return resp.Results[0], nil
+func (c *Client) handleStreamedResponse(httpResp *http.Response) (storage.SeriesSet, error) {
+	stream := NewChunkedReader(httpResp.Body, DefaultChunkedReadLimit, nil)


This method might be too shallow I would just inline this.

storage/remote/codec.go

bwplotka

Also it's ok to say that you are busy and cannot complete this PR. Let's just close it and allow others to finish this if they have time 👍 (:

storage/remote/codec.go

Sudhar287

Hey Bartek, I have some small queries. Can you please check them out when you're available? :)

bwplotka

Nice, responded for missed comment, sorry for missing it.

storage/remote/codec.go

bwplotka

Nice, just few comments and LGTM! (: Thanks for patience 👍

storage/remote/client.go

bwplotka · 2020-09-25T09:24:17Z

storage/remote/codec_test.go

@@ -229,3 +234,160 @@ func TestMergeLabels(t *testing.T) {
 		testutil.Equals(t, tc.expected, MergeLabels(tc.primary, tc.secondary))
 	}
 }
+
+type TestSample struct {


Suggested change

type TestSample struct {

type sample struct {

This is not possible as it collides with this

prometheus/storage/remote/queue_manager.go

Line 650 in 90fc6be

type sample struct {

Anyway, I have replaced it to testSample

bwplotka · 2020-09-25T09:25:46Z

Can you also resolve comments which are addressed to keep this PR discussion clear?

bwplotka · 2020-09-25T09:27:40Z

Tests are definitely failing because of this PR. I can see the endpoint tests are failing with context.Cancelled. Maybe we close connection sooner than we should?

Signed-off-by: Sudharshann D <sudhar287@gmail.com>

tomwilkie · 2021-02-27T10:25:54Z

This has been idle for a couple of months now, and @wwformat wants to continue work on it in #8351 - I suggest we close this and continue there?

roidelapluie · 2023-09-26T11:33:26Z

We have looked at this pull request during our bug scrub.

This is now a duplicate of #11379.

Thank you for your contribution.

bwplotka reviewed Jun 10, 2020

View reviewed changes

storage/remote/codec_test.go Outdated Show resolved Hide resolved

Sudhar287 force-pushed the remoteReadClient branch from e1a435f to 88e34ba Compare July 5, 2020 06:33

wycbox reviewed Jul 8, 2020

View reviewed changes

storage/remote/codec.go Outdated Show resolved Hide resolved

bwplotka reviewed Jul 10, 2020

View reviewed changes

storage/remote/codec.go Outdated Show resolved Hide resolved

storage/remote/codec.go Outdated Show resolved Hide resolved

Sudhar287 force-pushed the remoteReadClient branch 7 times, most recently from 752c227 to 81ba12f Compare July 14, 2020 04:51

bwplotka approved these changes Jul 21, 2020

View reviewed changes

storage/remote/client.go Show resolved Hide resolved

tsdb/tsdbutil/buffer.go Outdated Show resolved Hide resolved

Sudhar287 commented Jul 22, 2020

View reviewed changes

storage/remote/codec_test.go Outdated Show resolved Hide resolved

Sudhar287 force-pushed the remoteReadClient branch from 81ba12f to dfaac89 Compare July 22, 2020 03:27

Sudhar287 commented Jul 22, 2020

View reviewed changes

bwplotka reviewed Jul 27, 2020

View reviewed changes

Sudhar287 force-pushed the remoteReadClient branch 2 times, most recently from 32a4803 to e6b4942 Compare August 2, 2020 00:50

bwplotka requested changes Aug 4, 2020

View reviewed changes

Sudhar287 force-pushed the remoteReadClient branch 2 times, most recently from 5b62052 to b846c93 Compare August 9, 2020 09:42

bwplotka approved these changes Aug 20, 2020

View reviewed changes

storage/remote/codec.go Outdated Show resolved Hide resolved

storage/remote/codec.go Outdated Show resolved Hide resolved

Sudhar287 force-pushed the remoteReadClient branch from b846c93 to b766a76 Compare September 5, 2020 21:37

bwplotka requested changes Sep 11, 2020

View reviewed changes

bwplotka reviewed Sep 11, 2020

View reviewed changes

storage/remote/codec.go Outdated Show resolved Hide resolved

4n4nd mentioned this pull request Sep 14, 2020

Implement the Remoteread Package thanos-community/obslytics#10

Merged

Sudhar287 commented Sep 15, 2020

View reviewed changes

bwplotka reviewed Sep 17, 2020

View reviewed changes

storage/remote/codec.go Outdated Show resolved Hide resolved

storage/remote/codec.go Outdated Show resolved Hide resolved

Sudhar287 force-pushed the remoteReadClient branch from 6dffabb to bd27683 Compare September 19, 2020 22:59

bwplotka reviewed Sep 25, 2020

View reviewed changes

Sudhar287 force-pushed the remoteReadClient branch 6 times, most recently from 00a6743 to e453ffd Compare October 3, 2020 06:28

Sudhar287 force-pushed the remoteReadClient branch from e453ffd to fd23df5 Compare October 4, 2020 20:54

Sudhar287 force-pushed the remoteReadClient branch from fd23df5 to 0850f62 Compare October 17, 2020 03:52

streaming remote read

c13c652

Signed-off-by: Sudharshann D <sudhar287@gmail.com>

Sudhar287 force-pushed the remoteReadClient branch from 0850f62 to c13c652 Compare October 17, 2020 04:14

stale bot added the stale label Dec 16, 2020

wycbox mentioned this pull request Jan 8, 2021

added streaming remote read client side functionality (fork) #8351

Closed

Base automatically changed from master to main February 23, 2021 19:36

stale bot removed the stale label Feb 23, 2021

tomwilkie added the component/remote storage label Feb 27, 2021

stale bot added the stale label Apr 30, 2021

roidelapluie closed this Sep 26, 2023

added streaming remote read client side functionality #7354

added streaming remote read client side functionality #7354

Conversation

Sudhar287 commented Jun 6, 2020 • edited

bwplotka left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

bwplotka left a comment

Choose a reason for hiding this comment

Sudhar287 commented Jul 10, 2020

Sudhar287 commented Jul 11, 2020

bwplotka left a comment

Choose a reason for hiding this comment

Sudhar287 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

Sudhar287 Sep 27, 2020 • edited

Choose a reason for hiding this comment

bwplotka left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

bwplotka 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

bwplotka left a comment

Choose a reason for hiding this comment

bwplotka 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

bwplotka left a comment

Choose a reason for hiding this comment

Sudhar287 left a comment

Choose a reason for hiding this comment

bwplotka left a comment

Choose a reason for hiding this comment

bwplotka left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

bwplotka commented Sep 25, 2020

bwplotka commented Sep 25, 2020

tomwilkie commented Feb 27, 2021

roidelapluie commented Sep 26, 2023

Sudhar287 commented Jun 6, 2020 •

edited

Sudhar287 Sep 27, 2020 •

edited