[aggregator] Add compression for data sent to aggregator #2082

robskillington · 2019-12-27T01:04:12Z

What this PR does / why we need it:

Fixes #

Special notes for your reviewer:

Does this PR introduce a user-facing and/or backwards incompatible change?:

Does this PR require updating code package or user-facing documentation?:

codecov · 2019-12-27T02:40:43Z

Codecov Report

Merging #2082 (78867a8) into master (01958a4) will decrease coverage by 0.2%.
The diff coverage is 53.6%.

@@           Coverage Diff            @@
##           master   #2082     +/-   ##
========================================
- Coverage    72.3%   72.2%   -0.2%     
========================================
  Files        1015    1014      -1     
  Lines       87961   87492    -469     
========================================
- Hits        63683   63252    -431     
+ Misses      20029   20003     -26     
+ Partials     4249    4237     -12

Flag	Coverage Δ
aggregator	`81.7% <42.8%> (-0.4%)`	⬇️
cluster	`85.6% <ø> (ø)`
collector	`64.8% <ø> (ø)`
dbnode	`79.7% <ø> (+<0.1%)`	⬆️
m3em	`73.2% <ø> (ø)`
m3ninx	`73.9% <ø> (ø)`
m3nsch	`51.1% <ø> (ø)`
metrics	`17.7% <ø> (ø)`
msg	`74.3% <65.2%> (-0.7%)`	⬇️
query	`68.2% <ø> (-0.5%)`	⬇️
x	`83.3% <ø> (ø)`

Flags with carried forward coverage won't be shown. Click here to find out more.

Continue to review full report in Codecov by Sentry.

Legend - Click here to learn more
Δ = absolute <relative> (impact), ø = not affected, ? = missing data
Powered by Codecov. Last update 01958a4...78867a8. Read the comment docs.

arnikola · 2019-12-30T21:31:49Z

src/aggregator/client/config.go

@@ -141,6 +141,7 @@ func (c *Configuration) newClientOptions(

 // ConnectionConfiguration contains the connection configuration.
 type ConnectionConfiguration struct {
+	Compress                     CompressType         `yaml:"compress"`


nit: CompressionType? The other names seem to be fairly verbose

Also compression as the yaml key reads a little better imo

+1, compress would be passable if it was a bool (but not future-proof), but for enums this doesn't work at all

arnikola · 2019-12-30T21:42:50Z

src/aggregator/client/conn.go

+	// immediately write once it should be written to connection
+	// so always flush immediately.
+	if err := c.conn.Flush(); err != nil {
+		c.metrics.writeError.Inc(1)


Worth logging too? Or is it likely to be super spammy if it fails?

arnikola · 2019-12-30T21:43:55Z

src/aggregator/client/conn.go

@@ -228,9 +286,10 @@ func (c *connection) resetWithLock() {
 }

 func (c *connection) closeWithLock() {
-	if c.conn != nil {
-		c.conn.Close() // nolint: errcheck
+	if c.tcpConn != nil {


Should this also close c.conn or is tcpConn guaranteed to marshall the Close()?

arnikola · 2019-12-30T21:45:02Z

src/aggregator/client/conn_options.go

@@ -44,6 +45,55 @@ const (
 	defaultWriteRetryJitterEnabled      = true
 )

+// CompressType is a compression type.
+type CompressType uint


nit: CompressionType reads a little better? Up to you

arnikola · 2019-12-30T21:47:30Z

src/aggregator/client/conn_options.go

+	FlateCompressType
+)
+
+var (


meganit: instead of the var(...) block, var validCompressTypes = ... since there's only the one?

arnikola · 2019-12-30T21:49:03Z

src/aggregator/client/conn_options.go

+	if err := unmarshal(&str); err != nil {
+		return err
+	}
+	for _, validType := range validCompressTypes {


Should we do a default case?

if len(validCompressTypes) == 0 { *t = NoCompressType return nil }

arnikola · 2019-12-30T21:49:28Z

src/aggregator/client/conn_options.go

+		}
+	}
+
+	return fmt.Errorf("invalid compress type: %s, valid types are: %v",


nit; invalid compression type

arnikola · 2019-12-30T22:28:52Z

src/msg/consumer/consumer.go

@@ -85,14 +90,19 @@ func newConsumerMetrics(scope tally.Scope) metrics {
 	}
 }

+type connWriter interface {


nit; May be better to expose the client/conn interface from src/aggregator/client/conn.go here?

arnikola · 2019-12-30T22:31:19Z

src/msg/producer/writer/consumer_writer.go

+	var writer connWriter
+	switch connOpts.CompressType() {
+	case client.SnappyCompressType:
+		writer = snappy.NewBufferedWriter(u)


Can probably include these in the switch statement above if not using a common builder function?

arnikola · 2020-01-02T15:15:32Z

src/aggregator/server/rawtcp/server.go

+		connReader = snappy.NewReader(connReader)
+	case client.FlateCompressType:
+		connReader = flate.NewReader(connReader)
+	}


nit: might be worth adding a default: //noop case just for clarity (also in case the linter complains)?

arnikola · 2020-01-02T15:17:07Z

src/cmd/services/m3aggregator/config/server.go

+	Compress client.CompressType `yaml:"compress"`
+


nit: similar case re: naming here, compression over compress? Also comments in this file don't seem to have correct formatting, but happy to punt on that... may also be worth making it a pointer so it's not mandatory

arnikola · 2020-01-02T15:17:38Z

src/msg/consumer/config.go

@@ -37,6 +38,7 @@ type Configuration struct {
 	AckBufferSize             *int                      `yaml:"ackBufferSize"`
 	ConnectionWriteBufferSize *int                      `yaml:"connectionWriteBufferSize"`
 	ConnectionReadBufferSize  *int                      `yaml:"connectionReadBufferSize"`
+	Compress                  client.CompressType       `yaml:"compress"`


nit: same name comment re: compression vs compress, also perhaps make it a pointer so it's not mandatory?

arnikola · 2020-01-02T15:20:15Z

src/msg/consumer/consumer.go

+	var writer connWriter
+	switch opts.CompressType() {
+	case client.SnappyCompressType:
+		writer = snappy.NewBufferedWriter(conn)


If not building a common reader/writer builder that creates these, looks like we can roll both of these into a single switch statement, i.e.

case client.SnappyCompressType: reader = snappy.NewReader(conn) writer = snappy.NewBufferedWriter(conn) case ...

arnikola · 2020-01-02T15:28:02Z

src/aggregator/client/conn.go

+		writer = c.snappyWriter
+	case FlateCompressType:
+		c.flateWriter.Reset(tcpConn)
+		writer = c.flateWriter


Looks like we do this and the the connReader init in a couple of different places; might be worth adding a function that takes in a CompressType and a *net.TCPConn and returns the compression-wrapped connection?

arnikola · 2020-01-02T15:32:41Z

src/msg/producer/config/writer.go

-	FlushInterval   *time.Duration       `yaml:"flushInterval"`
-	WriteBufferSize *int                 `yaml:"writeBufferSize"`
-	ReadBufferSize  *int                 `yaml:"readBufferSize"`
+	Compress        aggclient.CompressType `yaml:"compress"`


Should this be a pointer so it's not mandatory?

arnikola · 2020-01-02T15:38:21Z

src/aggregator/client/conn.go

+type connWriter interface {
+	io.Writer
+	Flush() error
+}
+
+type rawConnWriter struct {
+	writer io.Writer
+}
+
+func newRawConnWriter(w io.Writer) connWriter {
+	return &rawConnWriter{writer: w}
+}
+
+func (w *rawConnWriter) Write(p []byte) (int, error) {
+	return w.writer.Write(p)
+}
+
+func (w *rawConnWriter) Flush() error {
+	return nil
+}


Maybe we should pull these into a common file, as well as flateReader and flateConnReader, then we can add a method that will build the correct readers and writers? Otherwise looks like there's a few places that are doing the same thing

arnikola · 2020-01-02T15:40:27Z

src/aggregator/client/conn.go

+		compressType:   opts.CompressType(),
+		snappyWriter:   snappy.NewBufferedWriter(nil),
+		flateWriter:    flateWriter,


Do these all need to be created here? Doesn't seem to be a way to change the compressType, so maybe better to add a Reset(c *net.TCPConn) method to connWriter interface and just create the appropriate writer here, then call Reset(tcpConn) in connectWithLock()?

vdarulis · 2020-03-02T17:24:26Z

src/aggregator/client/config.go

@@ -141,6 +141,7 @@ func (c *Configuration) newClientOptions(

 // ConnectionConfiguration contains the connection configuration.
 type ConnectionConfiguration struct {
+	Compress                     CompressType         `yaml:"compress"`


+1, compress would be passable if it was a bool (but not future-proof), but for enums this doesn't work at all

vdarulis · 2020-03-02T17:29:43Z

src/msg/producer/writer/consumer_writer.go

+	case client.FlateCompressType:
+		reader = newFlateConnReader()
+	default:
+		reader = bufio.NewReaderSize(u, connOpts.ReadBufferSize())


should document that ReadBufferSize has no effect if using compression

vdarulis · 2020-03-03T17:16:04Z

src/msg/producer/writer/consumer_writer.go

+	connOpts := opts.ConnectionOptions()
+
+	var reader connReader
+	switch connOpts.CompressType() {


it would be really nice in terms of ease of migration to actually prepend a header eg. {0xFF, 0xF0+compressType} and fall back to uncompressed en/decoders in case the header's wrong.

jorgelbg · 2020-03-11T23:27:56Z

Hi team 👋

I know that this PR is not yet merged, but we have started testing it and the results look quite promising.

After we deployed the docker images built with this PR we've seen ~25 write errors/s. From the m3aggregator logs:

{"level":"error","ts":1583968033.1887946,"msg":"decode error","remoteAddress":"10.2.10.207:34966","error":"snappy: corrupt input"}

I thought to provide some feedback here and also check if you have any advice on how to track the underline issue. Some time ago we had a problem with some particular metrics that had a super long value for a label (enough to trigger the id too large error). Wondering if something similar might be affecting the compression?

pavelnikolov · 2020-04-24T02:00:55Z

src/aggregator/client/conn.go

+}
+
+func (w *rawConnWriter) Flush() error {
+	return nil


What if the underlying writer is in some kind of an error state, for example, the connection is closed?

pavelnikolov · 2020-05-06T13:57:05Z

This change is something we need as we are hitting network congestion limits too in many of our m3coordinator pods.

@jorgelbg I believe that the errors that you are facing are because the connection from the coordinator are timing out and get closed immediately. This in turn causes issues when decoding on the m3aggregator side. In order to test this hypothesis you can increase the connection timoeut in your m3coordinator:

downsample:
  remoteAggregator:
    client:
      ...
      connection:
          writeTimeout: 15000ms   # set this value to something higher and monitor in the e2e dashboard

I was having the same issue and after I increased this timeout the errors almost disappeared. See screenshot attached:

pavelnikolov · 2020-05-06T14:50:13Z

Do you need help with this PR? I would be happy to resolve the conflicts and action the feedback comments.

robskillington added 2 commits December 27, 2019 12:03

[aggregator] Add compression for data sent to aggregator

981b52c

Fix bufio connection reader not use conn reader for rawtcp server

3cf2dd3

robskillington added 2 commits December 27, 2019 14:23

Add m3msg compression

29b3f05

Set compress type for consumer

78867a8

arnikola reviewed Dec 30, 2019

View reviewed changes

arnikola reviewed Jan 2, 2020

View reviewed changes

vdarulis reviewed Mar 2, 2020

View reviewed changes

vdarulis reviewed Mar 3, 2020

View reviewed changes

pavelnikolov reviewed Apr 24, 2020

View reviewed changes

tsharju mentioned this pull request Oct 2, 2020

[m3msg] add snappy compression for m3msg connections #2690

Open

robskillington force-pushed the master branch from 834e25e to 99f874f Compare November 3, 2020 17:16

robskillington force-pushed the master branch from e130b7e to 3cf877e Compare November 12, 2020 20:21

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[aggregator] Add compression for data sent to aggregator #2082

[aggregator] Add compression for data sent to aggregator #2082

robskillington commented Dec 27, 2019

codecov bot commented Dec 27, 2019 •

edited

Loading

arnikola Dec 30, 2019 •

edited

Loading

vdarulis Mar 2, 2020

arnikola Dec 30, 2019

arnikola Dec 30, 2019 •

edited

Loading

arnikola Dec 30, 2019

arnikola Dec 30, 2019 •

edited

Loading

arnikola Dec 30, 2019

arnikola Dec 30, 2019

arnikola Dec 30, 2019 •

edited

Loading

arnikola Dec 30, 2019 •

edited

Loading

arnikola Jan 2, 2020

arnikola Jan 2, 2020 •

edited

Loading

arnikola Jan 2, 2020 •

edited

Loading

arnikola Jan 2, 2020 •

edited

Loading

arnikola Jan 2, 2020

arnikola Jan 2, 2020

arnikola Jan 2, 2020

arnikola Jan 2, 2020

vdarulis Mar 2, 2020

vdarulis Mar 2, 2020

vdarulis Mar 3, 2020

jorgelbg commented Mar 11, 2020 •

edited

Loading

pavelnikolov Apr 24, 2020

pavelnikolov commented May 6, 2020

pavelnikolov commented May 6, 2020

[aggregator] Add compression for data sent to aggregator #2082

Are you sure you want to change the base?

[aggregator] Add compression for data sent to aggregator #2082

Conversation

robskillington commented Dec 27, 2019

codecov bot commented Dec 27, 2019 • edited Loading

Codecov Report

arnikola Dec 30, 2019 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

arnikola Dec 30, 2019 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

arnikola Dec 30, 2019 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

arnikola Dec 30, 2019 • edited Loading

Choose a reason for hiding this comment

arnikola Dec 30, 2019 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

arnikola Jan 2, 2020 • edited Loading

Choose a reason for hiding this comment

arnikola Jan 2, 2020 • edited Loading

Choose a reason for hiding this comment

arnikola Jan 2, 2020 • 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

Choose a reason for hiding this comment

jorgelbg commented Mar 11, 2020 • edited Loading

Choose a reason for hiding this comment

pavelnikolov commented May 6, 2020

pavelnikolov commented May 6, 2020

codecov bot commented Dec 27, 2019 •

edited

Loading

arnikola Dec 30, 2019 •

edited

Loading

arnikola Dec 30, 2019 •

edited

Loading

arnikola Dec 30, 2019 •

edited

Loading

arnikola Dec 30, 2019 •

edited

Loading

arnikola Dec 30, 2019 •

edited

Loading

arnikola Jan 2, 2020 •

edited

Loading

arnikola Jan 2, 2020 •

edited

Loading

arnikola Jan 2, 2020 •

edited

Loading

jorgelbg commented Mar 11, 2020 •

edited

Loading