forked from open-telemetry/opentelemetry-collector
-
Notifications
You must be signed in to change notification settings - Fork 0
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
[configgrpc] Use own compressors for zstd (open-telemetry#10323) (ope…
…n-telemetry#10324) Backport of open-telemetry#10323 Signed-off-by: Juraci Paixão Kröhling <juraci@kroehling.de> Co-authored-by: Juraci Paixão Kröhling <juraci@kroehling.de>
- Loading branch information
1 parent
0ab388b
commit 10e89bd
Showing
7 changed files
with
144 additions
and
6 deletions.
There are no files selected for viewing
13 changes: 13 additions & 0 deletions
13
.chloggen/jpkroehling-configgrpc-use-own-compressors-for-zstd.yaml
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,13 @@ | ||
| # Use this changelog template to create an entry for release notes. | ||
|
|
||
| # One of 'breaking', 'deprecation', 'new_component', 'enhancement', 'bug_fix' | ||
| change_type: 'bug_fix' | ||
|
|
||
| # The name of the component, or a single word describing the area of concern, (e.g. otlpreceiver) | ||
| component: Use own compressors for zstd | ||
|
|
||
| # A brief description of the change. Surround your text with quotes ("") if it needs to start with a backtick (`). | ||
| note: Before this change, the zstd compressor we used didn't respect the max message size. | ||
|
|
||
| # One or more tracking issues or pull requests related to the change | ||
| issues: [10323] |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,83 @@ | ||
| // Copyright The OpenTelemetry Authors | ||
| // Copyright 2017 gRPC authors | ||
| // SPDX-License-Identifier: Apache-2.0 | ||
|
|
||
| package internal // import "go.opentelemetry.io/collector/config/configgrpc/internal" | ||
|
|
||
| import ( | ||
| "errors" | ||
| "io" | ||
| "sync" | ||
|
|
||
| "github.com/klauspost/compress/zstd" | ||
| "google.golang.org/grpc/encoding" | ||
| ) | ||
|
|
||
| const ZstdName = "zstd" | ||
|
|
||
| func init() { | ||
| encoding.RegisterCompressor(NewZstdCodec()) | ||
| } | ||
|
|
||
| type writer struct { | ||
| *zstd.Encoder | ||
| pool *sync.Pool | ||
| } | ||
|
|
||
| func NewZstdCodec() encoding.Compressor { | ||
| c := &compressor{} | ||
| c.poolCompressor.New = func() any { | ||
| zw, _ := zstd.NewWriter(nil, zstd.WithEncoderConcurrency(1), zstd.WithWindowSize(512*1024)) | ||
| return &writer{Encoder: zw, pool: &c.poolCompressor} | ||
| } | ||
| return c | ||
| } | ||
|
|
||
| func (c *compressor) Compress(w io.Writer) (io.WriteCloser, error) { | ||
| z := c.poolCompressor.Get().(*writer) | ||
| z.Encoder.Reset(w) | ||
| return z, nil | ||
| } | ||
|
|
||
| func (z *writer) Close() error { | ||
| defer z.pool.Put(z) | ||
| return z.Encoder.Close() | ||
| } | ||
|
|
||
| type reader struct { | ||
| *zstd.Decoder | ||
| pool *sync.Pool | ||
| } | ||
|
|
||
| func (c *compressor) Decompress(r io.Reader) (io.Reader, error) { | ||
| z, inPool := c.poolDecompressor.Get().(*reader) | ||
| if !inPool { | ||
| newZ, err := zstd.NewReader(r) | ||
| if err != nil { | ||
| return nil, err | ||
| } | ||
| return &reader{Decoder: newZ, pool: &c.poolDecompressor}, nil | ||
| } | ||
| if err := z.Reset(r); err != nil { | ||
| c.poolDecompressor.Put(z) | ||
| return nil, err | ||
| } | ||
| return z, nil | ||
| } | ||
|
|
||
| func (z *reader) Read(p []byte) (n int, err error) { | ||
| n, err = z.Decoder.Read(p) | ||
| if errors.Is(err, io.EOF) { | ||
| z.pool.Put(z) | ||
| } | ||
| return n, err | ||
| } | ||
|
|
||
| func (c *compressor) Name() string { | ||
| return ZstdName | ||
| } | ||
|
|
||
| type compressor struct { | ||
| poolCompressor sync.Pool | ||
| poolDecompressor sync.Pool | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,41 @@ | ||
| // Copyright The OpenTelemetry Authors | ||
| // SPDX-License-Identifier: Apache-2.0 | ||
|
|
||
| package internal | ||
|
|
||
| import ( | ||
| "bytes" | ||
| "io" | ||
| "testing" | ||
|
|
||
| "github.com/stretchr/testify/require" | ||
| ) | ||
|
|
||
| func Test_zstdCodec_CompressDecompress(t *testing.T) { | ||
| // prepare | ||
| msg := []byte("Hello world.") | ||
| compressed := &bytes.Buffer{} | ||
|
|
||
| // zstd header, for sanity checking | ||
| header := []byte{40, 181, 47, 253} | ||
|
|
||
| c := NewZstdCodec() | ||
| cWriter, err := c.Compress(compressed) | ||
| require.NoError(t, err) | ||
| require.NotNil(t, cWriter) | ||
|
|
||
| _, err = cWriter.Write(msg) | ||
| require.NoError(t, err) | ||
| cWriter.Close() | ||
|
|
||
| cReader, err := c.Decompress(compressed) | ||
| require.NoError(t, err) | ||
| require.NotNil(t, cReader) | ||
|
|
||
| uncompressed, err := io.ReadAll(cReader) | ||
| require.NoError(t, err) | ||
| require.Equal(t, msg, uncompressed) | ||
|
|
||
| // test header | ||
| require.Equal(t, header, compressed.Bytes()[:4]) | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters