Implement stream connection for remote write #6580
Conversation
Signed-off-by: Alex Le <[email protected]>
Signed-off-by: Alex Le <[email protected]>
| return nil | ||
| case job := <-c.streamPushChan: | ||
| err = stream.Send(job.req) | ||
| if err == io.EOF { |
There was a problem hiding this comment.
when does the error become EOF? when the stream is closed but we try to send the job anyways?
could we extract this comparison to a well-named helper method to make it more explicit about what EOF means in this case?
There was a problem hiding this comment.
I think so. I checked how stream connection got handled elsewhere. Just handled this in similar way.
| @@ -137,7 +138,7 @@ type Config struct { | |||
| IgnoreSeriesLimitForMetricNames string `yaml:"ignore_series_limit_for_metric_names"` | |||
|
|
|||
| // For testing, you can override the address and ID of this ingester. | |||
| ingesterClientFactory func(addr string, cfg client.Config) (client.HealthAndIngesterClient, error) | |||
| ingesterClientFactory func(addr string, cfg client.Config, useStreamConnection bool) (client.HealthAndIngesterClient, error) | |||
There was a problem hiding this comment.
can we put this in the client.Config instead of adding an extra boolean parameter?
There was a problem hiding this comment.
Right now, I put this flag as distributor config and I try to use only one config. If put this in ingester client config, I would need to add extra attribute for distributor to know if it is using stream connection.
What this PR does:
Which issue(s) this PR fixes:
Fixes #
Checklist
CHANGELOG.mdupdated - the order of entries should be[CHANGE],[FEATURE],[ENHANCEMENT],[BUGFIX]