diff --git a/benchmarks/benchmarks_test.go b/benchmarks/benchmarks_test.go index 4dcb2819..75c5a0ab 100644 --- a/benchmarks/benchmarks_test.go +++ b/benchmarks/benchmarks_test.go @@ -24,14 +24,19 @@ import ( "github.com/splunk/stef/go/pkg" ) -var speedEncodings = []encodings.MetricEncoding{ - &otlp.OTLPEncoding{}, +var metricsSpeedEncodings = []encodings.MetricEncoding{ + &otlp.MetricsEncoding{}, &stef.STEFEncoding{Opts: pkg.WriterOptions{Compression: pkg.CompressionNone}}, &stef.STEFUEncoding{Opts: pkg.WriterOptions{Compression: pkg.CompressionNone}}, &parquetenc.Encoding{}, &otelarrow.OtelArrowEncoding{}, } +var logsSpeedEncodings = []encodings.LogEncoding{ + &otlp.LogsEncoding{}, + &stef.LogsSTEFEncoding{Opts: pkg.WriterOptions{Compression: pkg.CompressionNone}}, +} + var benchmarkDataVariations = []struct { generator generators.Generator }{ @@ -49,6 +54,11 @@ var benchmarkDataVariations = []struct { var chart = BarOutput{} +type encodingNamer interface { + Name() string + LongName() string +} + func TestMain(m *testing.M) { // call flag.Parse() here if TestMain uses flags chart.Begin() @@ -58,9 +68,9 @@ func TestMain(m *testing.M) { func addZstdCompressTime( b *testing.B, - encoding encodings.MetricEncoding, + encoding encodingNamer, bodyBytes []byte, - dataPointCount int, + unitCount int, ) { if !chartsEnabled() { return @@ -76,11 +86,12 @@ func addZstdCompressTime( log.Fatal("compression failed") } } + value := float64(b.Elapsed().Nanoseconds()) / float64(b.N*unitCount) chart.RecordStacked( b, encoding.LongName(), "Zstd Compress", - float64(b.Elapsed().Nanoseconds())/float64(b.N*dataPointCount), + value, ) }, ) @@ -88,9 +99,9 @@ func addZstdCompressTime( func addZstdDecompressTime( b *testing.B, - encoding encodings.MetricEncoding, + encoding encodingNamer, bodyBytes []byte, - dataPointCount int, + unitCount int, ) { if !chartsEnabled() { return @@ -110,28 +121,28 @@ func addZstdDecompressTime( log.Fatal(err) } } + value := float64(b.Elapsed().Nanoseconds()) / float64(b.N*unitCount) chart.RecordStacked( b, encoding.LongName(), "Zstd Decompress", - float64(b.Elapsed().Nanoseconds())/float64(b.N*dataPointCount), + value, ) }, ) } -func BenchmarkSerializeNative(b *testing.B) { +func BenchmarkMetricsSerializeNative(b *testing.B) { chart.BeginSection("Speed Benchmarks") - chart.BeginChart("Serialization Speed", b) + chart.BeginChart("Metrics Serialize", "ns/point", b) defer chart.EndChart( - "ns/point", charts.WithColorsOpts(opts.Colors{"#92C5F9", "#12C5F9"}), ) for _, dataVariation := range benchmarkDataVariations { - for _, encoding := range speedEncodings { + for _, encoding := range metricsSpeedEncodings { if _, ok := encoding.(*otelarrow.OtelArrowEncoding); ok { // Skip Arrow, it does not have native serialization continue @@ -167,15 +178,55 @@ func BenchmarkSerializeNative(b *testing.B) { b.ReportAllocs() } -func BenchmarkDeserializeNative(b *testing.B) { - chart.BeginChart("Deserialization Speed", b) +func BenchmarkLogsSerializeNative(b *testing.B) { + chart.BeginChart("Logs Serialize", "ns/log", b) + defer chart.EndChart( + charts.WithColorsOpts(opts.Colors{"#92C5F9", "#12C5F9"}), + ) + + logsData, err := testutils.ReadOTLPLogsFile("testdata/astronomy-otellogs.zst") + require.NoError(b, err) + logCount := logsData.LogRecordCount() + require.Positive(b, logCount) + + for _, encoding := range logsSpeedEncodings { + inmem, err := encoding.FromOTLP(logsData) + require.NoError(b, err) + b.Run( + fmt.Sprintf("%s/serialize", encoding.Name()), + func(b *testing.B) { + b.ResetTimer() + for i := 0; i < b.N; i++ { + bodyBytes, err := encoding.Encode(inmem) + if err != nil || bodyBytes == nil { + b.Fatal(err) + } + } + chart.RecordStacked( + b, + encoding.LongName(), + "Serialize", + float64(b.Elapsed().Nanoseconds())/float64(b.N*logCount), + ) + }, + ) + + bodyBytes, err := encoding.Encode(inmem) + require.NoError(b, err) + require.NotNil(b, bodyBytes) + addZstdCompressTime(b, encoding, bodyBytes, logCount) + } + b.ReportAllocs() +} + +func BenchmarkMetricsDeserializeNative(b *testing.B) { + chart.BeginChart("Metrics Deserialize", "ns/point", b) defer chart.EndChart( - "ns/point", charts.WithColorsOpts(opts.Colors{"#92C5F9", "#12C5F9"}), ) for _, dataVariation := range benchmarkDataVariations { - for _, encoding := range speedEncodings { + for _, encoding := range metricsSpeedEncodings { if _, ok := encoding.(*otelarrow.OtelArrowEncoding); ok { // Skip Arrow, it does not have native serialization continue @@ -224,15 +275,14 @@ func nsPerPoint(b *testing.B, dataPointCount int) float64 { return math.Round(float64(b.Elapsed().Nanoseconds()) / float64(totalDpCount)) } -func BenchmarkSerializeFromPdata(b *testing.B) { - chart.BeginChart("Serialization From pdata Speed", b) +func BenchmarkMetricsSerializeFromPdata(b *testing.B) { + chart.BeginChart("Metrics Serialize From pdata", "ns/point", b) defer chart.EndChart( - "ns/point", charts.WithColorsOpts(opts.Colors{"#92C5F9", "#12C5F9"}), ) for _, dataVariation := range benchmarkDataVariations { - for _, encoding := range speedEncodings { + for _, encoding := range metricsSpeedEncodings { if dataVariation.generator.GetName() == "hostandcollector-otelmetrics.zst" && encoding.Name() == "ARROW" { // Skip due to bug in Arrow encoding @@ -269,15 +319,56 @@ func BenchmarkSerializeFromPdata(b *testing.B) { b.ReportAllocs() } -func BenchmarkDeserializeToPdata(b *testing.B) { - chart.BeginChart("Deserialization To pdata Speed", b) +func BenchmarkLogsSerializeFromPdata(b *testing.B) { + chart.BeginChart("Logs Serialize From pdata", "ns/log", b) + defer chart.EndChart( + charts.WithColorsOpts(opts.Colors{"#92C5F9", "#12C5F9"}), + ) + + logsData, err := testutils.ReadOTLPLogsFile("testdata/astronomy-otellogs.zst") + require.NoError(b, err) + logCount := logsData.LogRecordCount() + require.Positive(b, logCount) + + for _, encoding := range logsSpeedEncodings { + b.Run( + fmt.Sprintf("%s/serialize", encoding.Name()), + func(b *testing.B) { + b.ResetTimer() + for i := 0; i < b.N; i++ { + inmem, err := encoding.FromOTLP(logsData) + require.NoError(b, err) + bodyBytes, err := encoding.Encode(inmem) + require.NoError(b, err) + require.NotNil(b, bodyBytes) + } + chart.Record( + b, + encoding.LongName(), + "CPU time to serialize one log", + nsPerPoint(b, logCount), + ) + }, + ) + + inmem, err := encoding.FromOTLP(logsData) + require.NoError(b, err) + bodyBytes, err := encoding.Encode(inmem) + require.NoError(b, err) + require.NotNil(b, bodyBytes) + addZstdCompressTime(b, encoding, bodyBytes, logCount) + } + b.ReportAllocs() +} + +func BenchmarkMetricsDeserializeToPdata(b *testing.B) { + chart.BeginChart("Metrics Deserialize To pdata", "ns/point", b) defer chart.EndChart( - "ns/point", charts.WithColorsOpts(opts.Colors{"#92C5F9", "#12C5F9"}), ) for _, dataVariation := range benchmarkDataVariations { - for _, encoding := range speedEncodings { + for _, encoding := range metricsSpeedEncodings { if dataVariation.generator.GetName() == "hostandcollector-otelmetrics.zst" && encoding.Name() == "ARROW" { // Skip due to bug in Arrow encoding @@ -316,7 +407,48 @@ func BenchmarkDeserializeToPdata(b *testing.B) { b.ReportAllocs() } -func BenchmarkReaderReadMany(b *testing.B) { +func BenchmarkLogsDeserializeToPdata(b *testing.B) { + chart.BeginChart("Logs Deserialize To pdata", "ns/log", b) + defer chart.EndChart( + charts.WithColorsOpts(opts.Colors{"#92C5F9", "#12C5F9"}), + ) + + logsData, err := testutils.ReadOTLPLogsFile("testdata/astronomy-otellogs.zst") + require.NoError(b, err) + logCount := logsData.LogRecordCount() + require.Positive(b, logCount) + + for _, encoding := range logsSpeedEncodings { + inmem, err := encoding.FromOTLP(logsData) + require.NoError(b, err) + bodyBytes, err := encoding.Encode(inmem) + require.NoError(b, err) + require.NotNil(b, bodyBytes) + + b.Run( + fmt.Sprintf("%s/deserialize", encoding.Name()), + func(b *testing.B) { + b.ResetTimer() + for i := 0; i < b.N; i++ { + _, err := encoding.ToOTLP(bodyBytes) + if err != nil { + b.Fatal(err) + } + } + chart.Record( + b, + encoding.LongName(), + "CPU time to deserialize one log", + nsPerPoint(b, logCount), + ) + }, + ) + addZstdDecompressTime(b, encoding, bodyBytes, logCount) + } + b.ReportAllocs() +} + +func BenchmarkMetricsReaderReadMany(b *testing.B) { generator := &generators.File{ FilePath: "testdata/hipstershop-otelmetrics.zst", } @@ -364,7 +496,7 @@ func BenchmarkReaderReadMany(b *testing.B) { ) } -func BenchmarkSTEFReaderRead(b *testing.B) { +func BenchmarkMetricsSTEFReaderRead(b *testing.B) { generator := &generators.File{ FilePath: "testdata/hipstershop-otelmetrics.zst", } @@ -405,9 +537,9 @@ func BenchmarkSTEFReaderRead(b *testing.B) { var multipartFiles = []string{"astronomy-otelmetrics"} -func BenchmarkSTEFSerializeMultipart(b *testing.B) { +func BenchmarkMetricsSTEFSerializeMultipart(b *testing.B) { for _, file := range multipartFiles { - parts, err := testutils.ReadMultipartOTLPFile("testdata/" + file + ".zst") + parts, err := testutils.ReadMultipartMetricsOTLPFile("testdata/" + file + ".zst") require.NoError(b, err) b.Run( file, @@ -442,9 +574,9 @@ func BenchmarkSTEFSerializeMultipart(b *testing.B) { } } -func BenchmarkSTEFDeserializeMultipart(b *testing.B) { +func BenchmarkMetricsSTEFDeserializeMultipart(b *testing.B) { for _, file := range multipartFiles { - parts, err := testutils.ReadMultipartOTLPFile("testdata/" + file + ".zst") + parts, err := testutils.ReadMultipartMetricsOTLPFile("testdata/" + file + ".zst") require.NoError(b, err) b.Run( file, diff --git a/benchmarks/charts.go b/benchmarks/charts.go index e06dde43..aa2a547f 100644 --- a/benchmarks/charts.go +++ b/benchmarks/charts.go @@ -16,6 +16,7 @@ import ( type BarOutput struct { t testing.TB title string + unit string file *os.File results map[string]map[string]float64 enabled bool @@ -25,13 +26,14 @@ func chartsEnabled() bool { return os.Getenv("UPDATE_BENCH_HTML") != "" } -func (c *BarOutput) BeginChart(title string, t testing.TB) { +func (c *BarOutput) BeginChart(title, unit string, t testing.TB) { c.t = t c.title = title + c.unit = unit c.results = map[string]map[string]float64{} } -func (c *BarOutput) EndChart(unit string, globalopts ...charts.GlobalOpts) { +func (c *BarOutput) EndChart(globalopts ...charts.GlobalOpts) { if !c.enabled { return } @@ -64,7 +66,7 @@ func (c *BarOutput) EndChart(unit string, globalopts ...charts.GlobalOpts) { charts.WithAnimation(false), charts.WithYAxisOpts( opts.YAxis{ - Name: unit, + Name: c.unit, }, ), ) @@ -117,14 +119,14 @@ func roundFloat(val float64, decimals int) float64 { func (c *BarOutput) Record(b *testing.B, encoding string, series string, val float64) { if b != nil { - b.ReportMetric(val, "ns/point") + b.ReportMetric(val, c.unit) } c.results[encoding] = map[string]float64{series: val} } func (c *BarOutput) RecordStacked(b *testing.B, encoding string, series string, val float64) { if b != nil { - b.ReportMetric(val, "ns/point") + b.ReportMetric(val, c.unit) } if c.results[encoding] == nil { c.results[encoding] = map[string]float64{} diff --git a/benchmarks/cmd/otlp2stef/main.go b/benchmarks/cmd/otlp2stef/main.go index f74a2e0a..235da7e8 100644 --- a/benchmarks/cmd/otlp2stef/main.go +++ b/benchmarks/cmd/otlp2stef/main.go @@ -35,7 +35,7 @@ func main() { } func convertOTLPtoSTEF(inputFilePath, outputFilePathNoExt, compressionMethod string) { - all, err := testutils.ReadOTLPFile(inputFilePath) + all, err := testutils.ReadOTLPMetricsFile(inputFilePath) if err != nil { log.Fatal(err) } diff --git a/benchmarks/correctness_test.go b/benchmarks/correctness_test.go index 04b86ebe..5abbde6a 100644 --- a/benchmarks/correctness_test.go +++ b/benchmarks/correctness_test.go @@ -8,17 +8,20 @@ import ( "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" + "go.opentelemetry.io/collector/pdata/plog" "go.opentelemetry.io/collector/pdata/pmetric" "github.com/splunk/stef/benchmarks/encodings/stef" "github.com/splunk/stef/benchmarks/testutils" "github.com/splunk/stef/go/otel/otelstef" + "github.com/splunk/stef/go/pdata/logs" + logtesttools "github.com/splunk/stef/go/pdata/logs/testtools" "github.com/splunk/stef/go/pdata/metrics" "github.com/splunk/stef/go/pdata/metrics/testtools" "github.com/splunk/stef/go/pkg" ) -func TestConvertSTEFFromToOTLP(t *testing.T) { +func TestMetricsConvertSTEFFromToOTLP(t *testing.T) { tests := []struct { file string }{ @@ -33,7 +36,7 @@ func TestConvertSTEFFromToOTLP(t *testing.T) { for _, test := range tests { t.Run( test.file, func(t *testing.T) { - otlpDataSrc, err := testutils.ReadOTLPFile(test.file) + otlpDataSrc, err := testutils.ReadOTLPMetricsFile(test.file) require.NoError(t, err) testtools.NormalizeMetrics(otlpDataSrc) @@ -69,6 +72,33 @@ func TestConvertSTEFFromToOTLP(t *testing.T) { } } +func TestLogsConvertSTEFFromToOTLP(t *testing.T) { + otlpDataSrc, err := testutils.ReadOTLPLogsFile("testdata/astronomy-otellogs.zst") + require.NoError(t, err) + + logtesttools.NormalizeLogs(otlpDataSrc) + srcCount := otlpDataSrc.LogRecordCount() + + buf := &pkg.MemChunkWriter{} + writer, err := otelstef.NewLogsWriter(buf, pkg.WriterOptions{}) + require.NoError(t, err) + + toStef := logs.OtlpToStefUnsorted{Sorted: true} + require.NoError(t, toStef.Convert(otlpDataSrc, writer)) + require.NoError(t, writer.Flush()) + + reader, err := otelstef.NewLogsReader(bytes.NewBuffer(buf.Bytes())) + require.NoError(t, err) + + toOtlp := logs.StefToOtlpUnsorted{} + otlpDataCopy, err := toOtlp.Convert(reader, true) + require.NoError(t, err) + + logtesttools.NormalizeLogs(otlpDataCopy) + assert.EqualValues(t, srcCount, otlpDataCopy.LogRecordCount()) + assert.True(t, bytes.Equal(logsToBytes(t, otlpDataSrc), logsToBytes(t, otlpDataCopy))) +} + func toBytes(t *testing.T, data pmetric.Metrics) []byte { marshaler := pmetric.ProtoMarshaler{} bytes, err := marshaler.MarshalMetrics(data) @@ -76,6 +106,13 @@ func toBytes(t *testing.T, data pmetric.Metrics) []byte { return bytes } +func logsToBytes(t *testing.T, data plog.Logs) []byte { + marshaler := plog.ProtoMarshaler{} + dataBytes, err := marshaler.MarshalLogs(data) + require.NoError(t, err) + return dataBytes +} + func TestTEFMultiPart(t *testing.T) { var testInputOtlpFiles = []string{ "testdata/astronomy-otelmetrics.zst", @@ -88,7 +125,7 @@ func TestTEFMultiPart(t *testing.T) { t.Run( inputFile, func(t *testing.T) { - parts, err := testutils.ReadMultipartOTLPFile(inputFile) + parts, err := testutils.ReadMultipartMetricsOTLPFile(inputFile) require.NoError(t, err) tefStream, err := tefEncoding.StartMultipart("") diff --git a/benchmarks/encodings/log_encoding.go b/benchmarks/encodings/log_encoding.go new file mode 100644 index 00000000..5c22c397 --- /dev/null +++ b/benchmarks/encodings/log_encoding.go @@ -0,0 +1,12 @@ +package encodings + +import "go.opentelemetry.io/collector/pdata/plog" + +type LogEncoding interface { + Name() string + LongName() string + FromOTLP(batch plog.Logs) (InMemoryData, error) + Encode(data InMemoryData) ([]byte, error) + Decode([]byte) (any, error) + ToOTLP(data []byte) (plog.Logs, error) +} diff --git a/benchmarks/encodings/otlp/logs.go b/benchmarks/encodings/otlp/logs.go new file mode 100644 index 00000000..92605b13 --- /dev/null +++ b/benchmarks/encodings/otlp/logs.go @@ -0,0 +1,37 @@ +package otlp + +import ( + "go.opentelemetry.io/collector/pdata/plog" + + "github.com/splunk/stef/benchmarks/encodings" +) + +type LogsEncoding struct{} + +var _ encodings.LogEncoding = (*LogsEncoding)(nil) + +func (*LogsEncoding) FromOTLP(data plog.Logs) (encodings.InMemoryData, error) { + return data, nil +} + +func (*LogsEncoding) Encode(data encodings.InMemoryData) ([]byte, error) { + marshaler := plog.ProtoMarshaler{} + return marshaler.MarshalLogs(data.(plog.Logs)) +} + +func (d *LogsEncoding) Decode(data []byte) (any, error) { + return d.ToOTLP(data) +} + +func (*LogsEncoding) ToOTLP(data []byte) (plog.Logs, error) { + unmarshaler := plog.ProtoUnmarshaler{} + return unmarshaler.UnmarshalLogs(data) +} + +func (*LogsEncoding) Name() string { + return "OTLP" +} + +func (*LogsEncoding) LongName() string { + return "Protobuf OTLP" +} diff --git a/benchmarks/encodings/otlp/otlp.go b/benchmarks/encodings/otlp/metrics.go similarity index 50% rename from benchmarks/encodings/otlp/otlp.go rename to benchmarks/encodings/otlp/metrics.go index 4ab2200d..9288906a 100644 --- a/benchmarks/encodings/otlp/otlp.go +++ b/benchmarks/encodings/otlp/metrics.go @@ -7,40 +7,40 @@ import ( "github.com/splunk/stef/benchmarks/testutils" ) -type OTLPEncoding struct { +type MetricsEncoding struct { } -func (d *OTLPEncoding) FromOTLP(data pmetric.Metrics) (encodings.InMemoryData, error) { +func (d *MetricsEncoding) FromOTLP(data pmetric.Metrics) (encodings.InMemoryData, error) { return data, nil } -func (d *OTLPEncoding) Encode(data encodings.InMemoryData) ([]byte, error) { +func (d *MetricsEncoding) Encode(data encodings.InMemoryData) ([]byte, error) { marshaler := pmetric.ProtoMarshaler{} return marshaler.MarshalMetrics(data.(pmetric.Metrics)) } -func (d *OTLPEncoding) Decode(b []byte) (any, error) { +func (d *MetricsEncoding) Decode(b []byte) (any, error) { return d.ToOTLP(b) } -func (*OTLPEncoding) ToOTLP(data []byte) (pmetric.Metrics, error) { +func (*MetricsEncoding) ToOTLP(data []byte) (pmetric.Metrics, error) { marshaler := pmetric.ProtoUnmarshaler{} return marshaler.UnmarshalMetrics(data) } -func (*OTLPEncoding) Name() string { +func (*MetricsEncoding) Name() string { return "OTLP" } -func (*OTLPEncoding) LongName() string { +func (*MetricsEncoding) LongName() string { return "Protobuf OTLP" } -type otlpMultipart struct { +type metricsMultipart struct { compression string bytes []byte } -func (o *otlpMultipart) AppendPart(part pmetric.Metrics) error { +func (o *metricsMultipart) AppendPart(part pmetric.Metrics) error { marshaler := pmetric.ProtoMarshaler{} b, err := marshaler.MarshalMetrics(part) if err != nil { @@ -55,10 +55,10 @@ func (o *otlpMultipart) AppendPart(part pmetric.Metrics) error { return nil } -func (o *otlpMultipart) FinishStream() ([]byte, error) { +func (o *metricsMultipart) FinishStream() ([]byte, error) { return o.bytes, nil } -func (d *OTLPEncoding) StartMultipart(compression string) (encodings.MetricMultipartStream, error) { - return &otlpMultipart{compression: compression}, nil +func (d *MetricsEncoding) StartMultipart(compression string) (encodings.MetricMultipartStream, error) { + return &metricsMultipart{compression: compression}, nil } diff --git a/benchmarks/encodings/stef/logs.go b/benchmarks/encodings/stef/logs.go new file mode 100644 index 00000000..fd3701e5 --- /dev/null +++ b/benchmarks/encodings/stef/logs.go @@ -0,0 +1,77 @@ +package stef + +import ( + "bytes" + "io" + + "go.opentelemetry.io/collector/pdata/plog" + + "github.com/splunk/stef/benchmarks/encodings" + "github.com/splunk/stef/go/otel/otelstef" + logconvert "github.com/splunk/stef/go/pdata/logs" + "github.com/splunk/stef/go/pdata/logs/sortedbyresource" + "github.com/splunk/stef/go/pkg" +) + +// LogsSTEFEncoding is resource-sorted STEF format for logs. +type LogsSTEFEncoding struct { + Opts pkg.WriterOptions +} + +var _ encodings.LogEncoding = (*LogsSTEFEncoding)(nil) + +func (*LogsSTEFEncoding) FromOTLP(data plog.Logs) (encodings.InMemoryData, error) { + return sortedbyresource.OtlpToSortedTree(data), nil +} + +func (d *LogsSTEFEncoding) Encode(data encodings.InMemoryData) ([]byte, error) { + tree := data.(*sortedbyresource.SortedTree) + output := &pkg.MemChunkWriter{} + writer, err := otelstef.NewLogsWriter(output, d.Opts) + if err != nil { + return nil, err + } + if err := tree.ToStef(writer); err != nil { + return nil, err + } + if err := writer.Flush(); err != nil { + return nil, err + } + return output.Bytes(), nil +} + +func (*LogsSTEFEncoding) Decode(data []byte) (any, error) { + reader, err := otelstef.NewLogsReader(bytes.NewReader(data)) + if err != nil { + return nil, err + } + for { + if err := reader.Read(pkg.ReadOptions{}); err != nil { + if err == io.EOF { + return nil, nil + } + return nil, err + } + } +} + +func (*LogsSTEFEncoding) ToOTLP(data []byte) (plog.Logs, error) { + reader, err := otelstef.NewLogsReader(bytes.NewReader(data)) + if err != nil { + return plog.NewLogs(), err + } + converter := logconvert.StefToOtlpUnsorted{} + return converter.Convert(reader, true) +} + +func (e *LogsSTEFEncoding) Name() string { + name := "STEF" + if e.Opts.Compression != pkg.CompressionNone { + name += "Z" + } + return name +} + +func (*LogsSTEFEncoding) LongName() string { + return "STEF" +} diff --git a/benchmarks/generators/file.go b/benchmarks/generators/file.go index 00869d57..3456c66b 100644 --- a/benchmarks/generators/file.go +++ b/benchmarks/generators/file.go @@ -23,7 +23,7 @@ func (f *File) GetName() string { } func (f *File) Generate() pmetric.Metrics { - all, err := testutils.ReadOTLPFile(f.FilePath) + all, err := testutils.ReadOTLPMetricsFile(f.FilePath) if err != nil { log.Fatalf("Error reading file %s: %v", f.FilePath, err) } diff --git a/benchmarks/otlp2sortedtree_test.go b/benchmarks/otlp2sortedtree_test.go index 4a52ae7b..00c711a4 100644 --- a/benchmarks/otlp2sortedtree_test.go +++ b/benchmarks/otlp2sortedtree_test.go @@ -36,7 +36,7 @@ func mapContains[K any](searchFor, inTree *b.Tree[K, bool]) bool { } func TestConvertFromOTLP(t *testing.T) { - otlpData, err := testutils.ReadOTLPFile("testdata/hipstershop-otelmetrics.zst") + otlpData, err := testutils.ReadOTLPMetricsFile("testdata/hipstershop-otelmetrics.zst") require.NoError(t, err) sorted, err := sortedbymetric.OtlpToSortedTree(otlpData) diff --git a/benchmarks/readwrite_test.go b/benchmarks/readwrite_test.go index b3bc507e..fe0c9629 100644 --- a/benchmarks/readwrite_test.go +++ b/benchmarks/readwrite_test.go @@ -76,7 +76,7 @@ func TestCopy(t *testing.T) { } } -func BenchmarkReadSTEF(b *testing.B) { +func BenchmarkMetricsReadSTEF(b *testing.B) { tefBytes, err := os.ReadFile("testdata/generated/hipstershop-otelmetrics.stefz") require.NoError(b, err) @@ -130,7 +130,7 @@ func BenchmarkReadSTEF(b *testing.B) { b.ReportMetric(float64(b.Elapsed().Nanoseconds())/float64(b.N*recCount), "ns/point") } -func BenchmarkReadSTEFZ(b *testing.B) { +func BenchmarkMetricsReadSTEFZ(b *testing.B) { tefBytes, err := os.ReadFile("testdata/generated/hipstershop-otelmetrics.stefz") require.NoError(b, err) @@ -157,7 +157,7 @@ func BenchmarkReadSTEFZ(b *testing.B) { b.ReportMetric(float64(b.Elapsed().Nanoseconds())/float64(b.N*recCount), "ns/point") } -func BenchmarkReadSTEFZWriteSTEF(b *testing.B) { +func BenchmarkMetricsReadSTEFZWriteSTEF(b *testing.B) { tefBytes, err := os.ReadFile("testdata/generated/hipstershop-otelmetrics.stefz") require.NoError(b, err) diff --git a/benchmarks/size_test.go b/benchmarks/size_test.go index c878527b..e2562a49 100644 --- a/benchmarks/size_test.go +++ b/benchmarks/size_test.go @@ -12,6 +12,7 @@ import ( "github.com/go-echarts/go-echarts/v2/charts" "github.com/go-echarts/go-echarts/v2/opts" "github.com/stretchr/testify/require" + "go.opentelemetry.io/collector/pdata/plog" "go.opentelemetry.io/collector/pdata/ptrace" "github.com/splunk/stef/benchmarks/encodings" @@ -22,6 +23,7 @@ import ( "github.com/splunk/stef/benchmarks/generators" "github.com/splunk/stef/benchmarks/testutils" "github.com/splunk/stef/go/otel/otelstef" + pdatalogs "github.com/splunk/stef/go/pdata/logs" "github.com/splunk/stef/go/pdata/traces" "github.com/splunk/stef/go/pkg" ) @@ -69,7 +71,7 @@ var metricsDataVariations = []struct { } var sizeEncodings = []encodings.MetricEncoding{ - &otlp.OTLPEncoding{}, + &otlp.MetricsEncoding{}, &stef.STEFEncoding{Opts: pkg.WriterOptions{Compression: pkg.CompressionNone}}, &stef.STEFUEncoding{Opts: pkg.WriterOptions{Compression: pkg.CompressionNone}}, &parquetenc.Encoding{}, @@ -98,7 +100,7 @@ func TestMetricsSize(t *testing.T) { wantChart := strings.HasSuffix(dataVariation.generator.GetName(), ".zst") if wantChart { - chart.BeginChart("Dataset: "+dataVariation.generator.GetName(), t) + chart.BeginChart("Dataset: "+dataVariation.generator.GetName(), "Bytes/point", t) } for _, encoding := range sizeEncodings { @@ -179,7 +181,6 @@ func TestMetricsSize(t *testing.T) { if wantChart { chart.EndChart( - "Bytes/point", charts.WithColorsOpts(opts.Colors{"#92C5F9"}), ) } @@ -201,7 +202,7 @@ func TestMetricsMultipart(t *testing.T) { } testMultipartEncodings := []encodings.MetricMultipartEncoding{ - &otlp.OTLPEncoding{}, + &otlp.MetricsEncoding{}, &stef.STEFEncoding{}, &stef.STEFUEncoding{}, &otelarrow.OtelArrowEncoding{}, @@ -213,7 +214,7 @@ func TestMetricsMultipart(t *testing.T) { for _, compression := range compressions { for _, dataset := range datasets { - chart.BeginChart("Dataset: "+dataset.name, t) + chart.BeginChart("Dataset: "+dataset.name, "Bytes/point", t) fmt.Printf("%-30s %4v %9v %4v\n", dataset.name, "Comp", "Bytes", "Ratio") @@ -225,7 +226,7 @@ func TestMetricsMultipart(t *testing.T) { // Encode each part one after another and write to the same STEF stream. // This models more closely the operation of STEF exporter in Collector. - parts, err := testutils.ReadMultipartOTLPFile("testdata/" + dataset.name + ".zst") + parts, err := testutils.ReadMultipartMetricsOTLPFile("testdata/" + dataset.name + ".zst") require.NoError(t, err) pointCount := 0 @@ -258,7 +259,6 @@ func TestMetricsMultipart(t *testing.T) { } chart.EndChart( - "Bytes/point", charts.WithColorsOpts(opts.Colors{"#87BB62"}), ) } @@ -266,7 +266,7 @@ func TestMetricsMultipart(t *testing.T) { } func TestSTEFVeryShortFrames(t *testing.T) { - input, err := testutils.ReadOTLPFile("testdata/hipstershop-otelmetrics.zst") + input, err := testutils.ReadOTLPMetricsFile("testdata/hipstershop-otelmetrics.zst") require.NoError(t, err) compressions := []pkg.Compression{pkg.CompressionNone, pkg.CompressionZstd} @@ -348,7 +348,7 @@ func TestTracesMultipart(t *testing.T) { } fmt.Println(compressionStr) - chart.BeginChart("Dataset: "+fileName, t) + chart.BeginChart("Dataset: "+fileName, "Bytes/span", t) otlpSize := 0 spanCount := 0 @@ -382,8 +382,11 @@ func TestTracesMultipart(t *testing.T) { converter := traces.OtlpToStefUnsorted{Sorted: sorted} - for i := 0; i < len(traceData); i++ { - err = converter.Convert(traceData[i].(ptrace.Traces), writer) + for i := 0; i < len(otlpParts); i++ { + tracesPart, err := u.UnmarshalTraces(otlpParts[i]) + require.NoError(t, err) + + err = converter.Convert(tracesPart, writer) require.NoError(t, err) err = writer.Flush() @@ -412,7 +415,110 @@ func TestTracesMultipart(t *testing.T) { ) } chart.EndChart( - "Bytes/span", + charts.WithColorsOpts(opts.Colors{"#87BB62"}), + ) + } + } +} + +func TestLogsMultipart(t *testing.T) { + u := plog.ProtoUnmarshaler{} + + fileNames := []string{"astronomy-otellogs"} + + chart.BeginSection("Size - Many Batches, Multipart Logs") + + for _, fileName := range fileNames { + fmt.Println("======= " + fileName) + + var otlpParts [][]byte + logsData, err := testutils.ReadMultipartOTLPFileGeneric( + "testdata/"+fileName+".zst", func(data []byte) (any, error) { + otlpParts = append(otlpParts, data) + return u.UnmarshalLogs(data) + }, + ) + require.NoError(t, err) + + compressions := []pkg.Compression{pkg.CompressionNone, pkg.CompressionZstd} + sorteds := []bool{false, true} + + for _, compression := range compressions { + var compressionStr string + if compression == pkg.CompressionZstd { + compressionStr = "zstd" + } else { + compressionStr = "none" + } + fmt.Println(compressionStr) + + chart.BeginChart("Dataset: "+fileName, "Bytes/log", t) + + otlpSize := 0 + logCount := 0 + for i := 0; i < len(logsData); i++ { + if compression == pkg.CompressionNone { + otlpSize += len(otlpParts[i]) + } else { + otlpZstd := testutils.CompressZstd(otlpParts[i]) + otlpSize += len(otlpZstd) + } + logCount += logsData[i].(plog.Logs).LogRecordCount() + } + + chart.Record( + nil, "OTLP", "Bytes/log, compression="+compressionStr, + unitSize(otlpSize, logCount), + ) + + for _, sorted := range sorteds { + var sortedStr string + if sorted { + sortedStr = "Sorted" + } else { + sortedStr = "Unsorted" + } + fmt.Println(sortedStr) + + outputBuf := &pkg.MemChunkWriter{} + writer, err := otelstef.NewLogsWriter(outputBuf, pkg.WriterOptions{Compression: compression}) + require.NoError(t, err) + + converter := pdatalogs.OtlpToStefUnsorted{Sorted: sorted} + + for i := 0; i < len(otlpParts); i++ { + logsPart, err := u.UnmarshalLogs(otlpParts[i]) + require.NoError(t, err) + + err = converter.Convert(logsPart, writer) + require.NoError(t, err) + + err = writer.Flush() + require.NoError(t, err) + } + + stefSize := len(outputBuf.Bytes()) + + fmt.Printf( + "Logs OTLP: %8d (%5.1f bytes/log)\n", + otlpSize, + float64(otlpSize)/float64(logCount), + ) + fmt.Printf( + "Logs STEF: %8d (%5.1f bytes/log)\n", + stefSize, + float64(stefSize)/float64(logCount), + ) + fmt.Printf( + "Ratio: %8.2f\n", float64(otlpSize)/float64(stefSize), + ) + + chart.Record( + nil, "STEF "+sortedStr, "Bytes/log, compression="+compressionStr, + unitSize(stefSize, logCount), + ) + } + chart.EndChart( charts.WithColorsOpts(opts.Colors{"#87BB62"}), ) } diff --git a/benchmarks/testdata/astronomy-otellogs.zst b/benchmarks/testdata/astronomy-otellogs.zst new file mode 100644 index 00000000..198730cc Binary files /dev/null and b/benchmarks/testdata/astronomy-otellogs.zst differ diff --git a/benchmarks/testutils/testutils.go b/benchmarks/testutils/testutils.go index cec5a226..c37f9f6c 100644 --- a/benchmarks/testutils/testutils.go +++ b/benchmarks/testutils/testutils.go @@ -9,6 +9,7 @@ import ( "strings" "github.com/klauspost/compress/zstd" + "go.opentelemetry.io/collector/pdata/plog" "go.opentelemetry.io/collector/pdata/pmetric" ) @@ -43,7 +44,7 @@ func openOTLPFile(filePath string) (io.Reader, func(), error) { return reader, closer, nil } -func ReadMultipartOTLPFile(filePath string) ([]pmetric.Metrics, error) { +func ReadMultipartMetricsOTLPFile(filePath string) ([]pmetric.Metrics, error) { var result []pmetric.Metrics reader, closer, err := openOTLPFile(filePath) @@ -135,9 +136,9 @@ func ReadMultipartOTLPFileGeneric(filePath string, unmarshaler func([]byte) (any return result, nil } -func ReadOTLPFile(filePath string) (pmetric.Metrics, error) { +func ReadOTLPMetricsFile(filePath string) (pmetric.Metrics, error) { combined := pmetric.NewMetrics() - parts, err := ReadMultipartOTLPFile(filePath) + parts, err := ReadMultipartMetricsOTLPFile(filePath) if err != nil { return combined, err } @@ -146,3 +147,20 @@ func ReadOTLPFile(filePath string) (pmetric.Metrics, error) { } return combined, nil } + +func ReadOTLPLogsFile(filePath string) (plog.Logs, error) { + combined := plog.NewLogs() + unmarshaler := plog.ProtoUnmarshaler{} + parts, err := ReadMultipartOTLPFileGeneric( + filePath, func(data []byte) (any, error) { + return unmarshaler.UnmarshalLogs(data) + }, + ) + if err != nil { + return combined, err + } + for _, part := range parts { + part.(plog.Logs).ResourceLogs().MoveAndAppendTo(combined.ResourceLogs()) + } + return combined, nil +} diff --git a/docs/benchmarks.html b/docs/benchmarks.html index 55c373ff..692a1f95 100644 --- a/docs/benchmarks.html +++ b/docs/benchmarks.html @@ -9,186 +9,247 @@