From 7435ff1fd0bdca69f5ad5d6a9d22f4b31124b2f7 Mon Sep 17 00:00:00 2001 From: 81reap Date: Tue, 25 Aug 2026 22:58:40 -0400 Subject: [PATCH 1/4] fix(outputs.parquet): Sanitize measurement names before using as file name Since measurement names are used as file names, an invalid file name like `../../../../tmp/evil` wrote its file to `/tmp` instead. A name that was too long or had a `NUL` byte made `os.Create` fail into a retry storm blocking telegraf from writing metrics until the bad metric aged out of the buffer. Now measurement names are sanitized if they are used as a file name. Renamed measurements are logged to console simmilar to how `outputs.bigquery` handles table names. --- plugins/outputs/parquet/README.md | 5 ++ plugins/outputs/parquet/parquet.go | 39 ++++++++++++- plugins/outputs/parquet/parquet_test.go | 77 +++++++++++++++++++++++++ 3 files changed, 119 insertions(+), 2 deletions(-) diff --git a/plugins/outputs/parquet/README.md b/plugins/outputs/parquet/README.md index 19627cd5da2c4..fb1b1647acfd6 100644 --- a/plugins/outputs/parquet/README.md +++ b/plugins/outputs/parquet/README.md @@ -82,6 +82,11 @@ of this occurring. ## File Rotation +Measurement names determine the file name and can be customized with +the [rename processor][rename]. + +[rename]: /plugins/processors/rename/README.md + If a file with the same target name exists at start, the existing file is rotated to avoid over-writing it or conflicting schema. diff --git a/plugins/outputs/parquet/parquet.go b/plugins/outputs/parquet/parquet.go index a4cf876d0c45b..829625711d48a 100644 --- a/plugins/outputs/parquet/parquet.go +++ b/plugins/outputs/parquet/parquet.go @@ -6,8 +6,11 @@ import ( "errors" "fmt" "os" + "runtime" "strconv" + "strings" "time" + "unicode" "github.com/apache/arrow-go/v18/arrow" "github.com/apache/arrow-go/v18/arrow/array" @@ -38,7 +41,8 @@ type Parquet struct { TimestampFieldName string `toml:"timestamp_field_name"` Log telegraf.Logger `toml:"-"` - metricGroups map[string]*metricGroup + metricGroups map[string]*metricGroup + warnedOnFilename bool } func (*Parquet) SampleConfig() string { @@ -88,7 +92,8 @@ func (p *Parquet) Close() error { func (p *Parquet) Write(metrics []telegraf.Metric) error { groupedMetrics := make(map[string][]telegraf.Metric) for _, metric := range metrics { - groupedMetrics[metric.Name()] = append(groupedMetrics[metric.Name()], metric) + name := p.metricToFile(metric.Name()) + groupedMetrics[name] = append(groupedMetrics[name], metric) } now := time.Now() @@ -130,6 +135,36 @@ func (p *Parquet) Write(metrics []telegraf.Metric) error { return nil } +const maxMeasurementLen = 255 - len("-2006-01-02-1234567890.parquet") + +func (p *Parquet) metricToFile(name string) string { + safe := strings.Map(func(r rune) rune { + if reservedInFilename(r) { + return '_' + } + return r + }, name) + + if len(safe) > maxMeasurementLen { + safe = strings.ToValidUTF8(safe[:maxMeasurementLen], "") + } + + if safe != name && !p.warnedOnFilename { + p.warnedOnFilename = true + p.Log.Warnf("Metric %q is not usable as a file name, writing to %q instead; use the rename processor to choose the name", name, safe) + } + + return safe +} + +func reservedInFilename(r rune) bool { + if r == '/' || r == '\\' || unicode.IsControl(r) { + return true + } + + return runtime.GOOS == "windows" && strings.ContainsRune(`<>:"|?*`, r) +} + func (p *Parquet) rotateIfNeeded(name string) error { fileInfo, err := os.Stat(p.metricGroups[name].filename) if err != nil { diff --git a/plugins/outputs/parquet/parquet_test.go b/plugins/outputs/parquet/parquet_test.go index b063ce0997605..09dfd4ca778a1 100644 --- a/plugins/outputs/parquet/parquet_test.go +++ b/plugins/outputs/parquet/parquet_test.go @@ -3,6 +3,8 @@ package parquet import ( "os" "path/filepath" + "runtime" + "strings" "testing" "time" @@ -276,3 +278,78 @@ func TestMissingValuesReadBackAsNull(t *testing.T) { require.Equalf(t, int16(1), column.MaxDefinitionLevel(), "column %q is required, nulls would be written as zero", column.Name()) } } + +func TestMetricToFile(t *testing.T) { + tests := []struct { + name string + expected string + }{ + {"cpu", "cpu"}, + {"../../etc/passwd", ".._.._etc_passwd"}, + {`a/b\c`, "a_b_c"}, + {"nul\x00byte", "nul_byte"}, + {"tab\tnewline\n", "tab_newline_"}, + {strings.Repeat("a", 300), strings.Repeat("a", maxMeasurementLen)}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + p := &Parquet{Log: testutil.Logger{}} + require.Equal(t, tt.expected, p.metricToFile(tt.name)) + }) + } +} + +func TestWindowsReservedCharacters(t *testing.T) { + if runtime.GOOS != "windows" { + t.Skip("windows only") + } + + p := &Parquet{Log: testutil.Logger{}} + require.Equal(t, "a_b_c", p.metricToFile(`ac`)) +} + +func TestMeasurementNamesCannotEscapeDirectory(t *testing.T) { + dir := t.TempDir() + outside := filepath.Dir(dir) + before, err := filepath.Glob(filepath.Join(outside, "*.parquet")) + require.NoError(t, err) + + p := &Parquet{Directory: dir, TimestampFieldName: "timestamp", Log: testutil.Logger{}} + require.NoError(t, p.Init()) + + for _, name := range []string{"../../../../tmp/evil", "..", "../evil", `a\..\..\b`} { + m := metric.New(name, nil, map[string]interface{}{"value": int64(1)}, time.Now()) + require.NoError(t, p.Write([]telegraf.Metric{m})) + } + require.NoError(t, p.Close()) + + after, err := filepath.Glob(filepath.Join(outside, "*.parquet")) + require.NoError(t, err) + require.Equal(t, before, after) + + written, err := filepath.Glob(filepath.Join(dir, "*.parquet")) + require.NoError(t, err) + require.Len(t, written, 4) +} + +func TestLongMeasurementNameKeepsOutputWriting(t *testing.T) { + dir := t.TempDir() + p := &Parquet{Directory: dir, TimestampFieldName: "timestamp", Log: testutil.Logger{}} + require.NoError(t, p.Init()) + + require.NoError(t, p.Write([]telegraf.Metric{ + metric.New(strings.Repeat("a", 300), nil, map[string]interface{}{"value": int64(1)}, time.Now()), + })) + require.NoError(t, p.Write([]telegraf.Metric{ + metric.New("good", nil, map[string]interface{}{"value": int64(2)}, time.Now()), + })) + require.NoError(t, p.Close()) + + written, err := filepath.Glob(filepath.Join(dir, "*.parquet")) + require.NoError(t, err) + require.Len(t, written, 2) + for _, name := range written { + require.LessOrEqual(t, len(filepath.Base(name)), 255) + } +} From baccc5373cdedbc25ac6a3be4a468723e63f0843 Mon Sep 17 00:00:00 2001 From: 81reap Date: Wed, 26 Aug 2026 11:19:53 -0400 Subject: [PATCH 2/4] rev2 (please sqash + merge) --- plugins/outputs/parquet/parquet.go | 42 ++++++------- plugins/outputs/parquet/parquet_test.go | 81 +++++++++++++++++-------- 2 files changed, 76 insertions(+), 47 deletions(-) diff --git a/plugins/outputs/parquet/parquet.go b/plugins/outputs/parquet/parquet.go index 829625711d48a..f87c000931d90 100644 --- a/plugins/outputs/parquet/parquet.go +++ b/plugins/outputs/parquet/parquet.go @@ -20,6 +20,7 @@ import ( "github.com/influxdata/telegraf" "github.com/influxdata/telegraf/config" + "github.com/influxdata/telegraf/internal" "github.com/influxdata/telegraf/plugins/outputs" ) @@ -41,8 +42,7 @@ type Parquet struct { TimestampFieldName string `toml:"timestamp_field_name"` Log telegraf.Logger `toml:"-"` - metricGroups map[string]*metricGroup - warnedOnFilename bool + metricGroups map[string]*metricGroup } func (*Parquet) SampleConfig() string { @@ -91,8 +91,15 @@ func (p *Parquet) Close() error { func (p *Parquet) Write(metrics []telegraf.Metric) error { groupedMetrics := make(map[string][]telegraf.Metric) - for _, metric := range metrics { - name := p.metricToFile(metric.Name()) + var writeErr internal.PartialWriteError + for i, metric := range metrics { + name := metric.Name() + if !usableAsFilename(name) { + writeErr.MetricsReject = append(writeErr.MetricsReject, i) + writeErr.MetricsRejectErrors = append(writeErr.MetricsRejectErrors, fmt.Errorf("measurement %q cannot be used as a file name", name)) + continue + } + writeErr.MetricsAccept = append(writeErr.MetricsAccept, i) groupedMetrics[name] = append(groupedMetrics[name], metric) } @@ -132,29 +139,18 @@ func (p *Parquet) Write(metrics []telegraf.Metric) error { record.Release() } - return nil + if len(writeErr.MetricsReject) == 0 { + return nil + } + writeErr.Err = fmt.Errorf("rejected %d metric(s): %w", len(writeErr.MetricsReject), errors.Join(writeErr.MetricsRejectErrors...)) + + return &writeErr } const maxMeasurementLen = 255 - len("-2006-01-02-1234567890.parquet") -func (p *Parquet) metricToFile(name string) string { - safe := strings.Map(func(r rune) rune { - if reservedInFilename(r) { - return '_' - } - return r - }, name) - - if len(safe) > maxMeasurementLen { - safe = strings.ToValidUTF8(safe[:maxMeasurementLen], "") - } - - if safe != name && !p.warnedOnFilename { - p.warnedOnFilename = true - p.Log.Warnf("Metric %q is not usable as a file name, writing to %q instead; use the rename processor to choose the name", name, safe) - } - - return safe +func usableAsFilename(name string) bool { + return name != "" && len(name) <= maxMeasurementLen && !strings.ContainsFunc(name, reservedInFilename) } func reservedInFilename(r rune) bool { diff --git a/plugins/outputs/parquet/parquet_test.go b/plugins/outputs/parquet/parquet_test.go index 09dfd4ca778a1..f87327e55e4a5 100644 --- a/plugins/outputs/parquet/parquet_test.go +++ b/plugins/outputs/parquet/parquet_test.go @@ -4,6 +4,7 @@ import ( "os" "path/filepath" "runtime" + "slices" "strings" "testing" "time" @@ -13,6 +14,7 @@ import ( "github.com/influxdata/telegraf" "github.com/influxdata/telegraf/config" + "github.com/influxdata/telegraf/internal" "github.com/influxdata/telegraf/metric" "github.com/influxdata/telegraf/testutil" ) @@ -279,23 +281,26 @@ func TestMissingValuesReadBackAsNull(t *testing.T) { } } -func TestMetricToFile(t *testing.T) { +func TestUsableAsFilename(t *testing.T) { tests := []struct { - name string - expected string + name string + usable bool }{ - {"cpu", "cpu"}, - {"../../etc/passwd", ".._.._etc_passwd"}, - {`a/b\c`, "a_b_c"}, - {"nul\x00byte", "nul_byte"}, - {"tab\tnewline\n", "tab_newline_"}, - {strings.Repeat("a", 300), strings.Repeat("a", maxMeasurementLen)}, + {"cpu", true}, + {"disk.io", true}, + {"..", true}, + {"", false}, + {"../../etc/passwd", false}, + {`a\b`, false}, + {"nul\x00byte", false}, + {"tab\tnewline\n", false}, + {strings.Repeat("a", maxMeasurementLen), true}, + {strings.Repeat("a", maxMeasurementLen+1), false}, } for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { - p := &Parquet{Log: testutil.Logger{}} - require.Equal(t, tt.expected, p.metricToFile(tt.name)) + require.Equal(t, tt.usable, usableAsFilename(tt.name)) }) } } @@ -305,8 +310,7 @@ func TestWindowsReservedCharacters(t *testing.T) { t.Skip("windows only") } - p := &Parquet{Log: testutil.Logger{}} - require.Equal(t, "a_b_c", p.metricToFile(`ac`)) + require.False(t, usableAsFilename(`ac`)) } func TestMeasurementNamesCannotEscapeDirectory(t *testing.T) { @@ -318,9 +322,9 @@ func TestMeasurementNamesCannotEscapeDirectory(t *testing.T) { p := &Parquet{Directory: dir, TimestampFieldName: "timestamp", Log: testutil.Logger{}} require.NoError(t, p.Init()) - for _, name := range []string{"../../../../tmp/evil", "..", "../evil", `a\..\..\b`} { + for _, name := range []string{"../../../../tmp/evil", "../evil", `a\..\..\b`, "nul\x00byte"} { m := metric.New(name, nil, map[string]interface{}{"value": int64(1)}, time.Now()) - require.NoError(t, p.Write([]telegraf.Metric{m})) + require.ErrorAs(t, p.Write([]telegraf.Metric{m}), new(*internal.PartialWriteError)) } require.NoError(t, p.Close()) @@ -330,26 +334,55 @@ func TestMeasurementNamesCannotEscapeDirectory(t *testing.T) { written, err := filepath.Glob(filepath.Join(dir, "*.parquet")) require.NoError(t, err) - require.Len(t, written, 4) + require.Empty(t, written) } -func TestLongMeasurementNameKeepsOutputWriting(t *testing.T) { +func TestUnusableMeasurementNameKeepsOutputWriting(t *testing.T) { dir := t.TempDir() p := &Parquet{Directory: dir, TimestampFieldName: "timestamp", Log: testutil.Logger{}} require.NoError(t, p.Init()) - require.NoError(t, p.Write([]telegraf.Metric{ + require.ErrorAs(t, p.Write([]telegraf.Metric{ metric.New(strings.Repeat("a", 300), nil, map[string]interface{}{"value": int64(1)}, time.Now()), - })) - require.NoError(t, p.Write([]telegraf.Metric{ + }), new(*internal.PartialWriteError)) + + for i := 0; i < 8; i++ { + require.NoError(t, p.Write([]telegraf.Metric{ + metric.New("good", nil, map[string]interface{}{"value": int64(i)}, time.Now()), + })) + } + require.NoError(t, p.Close()) + + written, err := filepath.Glob(filepath.Join(dir, "*.parquet")) + require.NoError(t, err) + require.Len(t, written, 1) + require.Contains(t, filepath.Base(written[0]), "good") +} + +func TestRejectedMetricsReportEveryIndexExactlyOnce(t *testing.T) { + dir := t.TempDir() + p := &Parquet{Directory: dir, TimestampFieldName: "timestamp", Log: testutil.Logger{}} + require.NoError(t, p.Init()) + + metrics := []telegraf.Metric{ + metric.New("../evil", nil, map[string]interface{}{"value": int64(1)}, time.Now()), metric.New("good", nil, map[string]interface{}{"value": int64(2)}, time.Now()), - })) + metric.New("nul\x00byte", nil, map[string]interface{}{"value": int64(3)}, time.Now()), + metric.New("alsogood", nil, map[string]interface{}{"value": int64(4)}, time.Now()), + } + + var writeErr *internal.PartialWriteError + require.ErrorAs(t, p.Write(metrics), &writeErr) + require.Equal(t, []int{0, 2}, writeErr.MetricsReject) + require.Equal(t, []int{1, 3}, writeErr.MetricsAccept) + require.Len(t, writeErr.MetricsRejectErrors, 2) + require.ElementsMatch(t, + []int{0, 1, 2, 3}, + append(slices.Clone(writeErr.MetricsAccept), writeErr.MetricsReject...), + ) require.NoError(t, p.Close()) written, err := filepath.Glob(filepath.Join(dir, "*.parquet")) require.NoError(t, err) require.Len(t, written, 2) - for _, name := range written { - require.LessOrEqual(t, len(filepath.Base(name)), 255) - } } From 78cc3296b39fd56939fb6af55c01f4b051eec2a9 Mon Sep 17 00:00:00 2001 From: 81reap Date: Wed, 26 Aug 2026 15:38:12 -0400 Subject: [PATCH 3/4] rev3 (please sqash + merge) --- plugins/outputs/parquet/README.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/plugins/outputs/parquet/README.md b/plugins/outputs/parquet/README.md index fb1b1647acfd6..b5e190f00369e 100644 --- a/plugins/outputs/parquet/README.md +++ b/plugins/outputs/parquet/README.md @@ -82,7 +82,7 @@ of this occurring. ## File Rotation -Measurement names determine the file name and can be customized with +Measurement names determine the file name and can be customized with the [rename processor][rename]. [rename]: /plugins/processors/rename/README.md From a0b34d96c51394e5c8dc3fdd006227d4e398acf9 Mon Sep 17 00:00:00 2001 From: 81reap Date: Wed, 26 Aug 2026 15:57:02 -0400 Subject: [PATCH 4/4] rev4 (please sqash + merge) --- plugins/outputs/parquet/parquet.go | 127 +++++++++++++----------- plugins/outputs/parquet/parquet_test.go | 88 +++++++++------- 2 files changed, 119 insertions(+), 96 deletions(-) diff --git a/plugins/outputs/parquet/parquet.go b/plugins/outputs/parquet/parquet.go index f87c000931d90..1f230b5acd8ff 100644 --- a/plugins/outputs/parquet/parquet.go +++ b/plugins/outputs/parquet/parquet.go @@ -6,11 +6,8 @@ import ( "errors" "fmt" "os" - "runtime" "strconv" - "strings" "time" - "unicode" "github.com/apache/arrow-go/v18/arrow" "github.com/apache/arrow-go/v18/arrow/array" @@ -42,6 +39,7 @@ type Parquet struct { TimestampFieldName string `toml:"timestamp_field_name"` Log telegraf.Logger `toml:"-"` + root *os.Root metricGroups map[string]*metricGroup } @@ -54,15 +52,16 @@ func (p *Parquet) Init() error { p.Directory = "." } - stat, err := os.Stat(p.Directory) - if os.IsNotExist(err) { - if err := os.MkdirAll(p.Directory, 0750); err != nil { - return fmt.Errorf("failed to create directory %q: %w", p.Directory, err) - } - } else if !stat.IsDir() { - return fmt.Errorf("provided directory %q is not a directory", p.Directory) + if err := os.MkdirAll(p.Directory, 0750); err != nil { + return fmt.Errorf("failed to create directory %q: %w", p.Directory, err) + } + + root, err := os.OpenRoot(p.Directory) + if err != nil { + return fmt.Errorf("failed to open directory %q: %w", p.Directory, err) } + p.root = root p.metricGroups = make(map[string]*metricGroup) return nil @@ -82,6 +81,11 @@ func (p *Parquet) Close() error { } } + if err := p.root.Close(); err != nil { + p.Log.Errorf("failed to close directory %q: %v", p.Directory, err) + errorOccurred = true + } + if errorOccurred { return errors.New("failed closing one or more parquet files") } @@ -90,53 +94,43 @@ func (p *Parquet) Close() error { } func (p *Parquet) Write(metrics []telegraf.Metric) error { - groupedMetrics := make(map[string][]telegraf.Metric) - var writeErr internal.PartialWriteError - for i, metric := range metrics { - name := metric.Name() - if !usableAsFilename(name) { - writeErr.MetricsReject = append(writeErr.MetricsReject, i) - writeErr.MetricsRejectErrors = append(writeErr.MetricsRejectErrors, fmt.Errorf("measurement %q cannot be used as a file name", name)) - continue - } - writeErr.MetricsAccept = append(writeErr.MetricsAccept, i) - groupedMetrics[name] = append(groupedMetrics[name], metric) + grouped := make(map[string][]int) + for i, m := range metrics { + grouped[m.Name()] = append(grouped[m.Name()], i) } + var writeErr internal.PartialWriteError + now := time.Now() - for name, metrics := range groupedMetrics { - if _, ok := p.metricGroups[name]; !ok { - filename := fmt.Sprintf("%s/%s-%s-%s.parquet", p.Directory, name, now.Format("2006-01-02"), strconv.FormatInt(now.Unix(), 10)) - schema, err := p.createSchema(metrics) - if err != nil { - return fmt.Errorf("failed to create schema for file %q: %w", name, err) - } - writer, err := p.createWriter(name, filename, schema) - if err != nil { - return fmt.Errorf("failed to create writer for file %q: %w", name, err) - } - p.metricGroups[name] = &metricGroup{ - builder: array.NewRecordBuilder(memory.DefaultAllocator, schema), - filename: filename, - schema: schema, - writer: writer, - } + for name, indices := range grouped { + batch := make([]telegraf.Metric, len(indices)) + for j, i := range indices { + batch[j] = metrics[i] + } + + group, err := p.groupFor(name, batch, now) + if err != nil { + writeErr.MetricsReject = append(writeErr.MetricsReject, indices...) + writeErr.MetricsRejectErrors = append(writeErr.MetricsRejectErrors, err) + continue } if p.RotationInterval != 0 { if err := p.rotateIfNeeded(name); err != nil { - return fmt.Errorf("failed to rotate file %q: %w", p.metricGroups[name].filename, err) + return fmt.Errorf("failed to rotate file %q: %w", group.filename, err) } } - record, err := p.createRecordBatch(metrics, p.metricGroups[name].builder, p.metricGroups[name].schema) + record, err := p.createRecordBatch(batch, group.builder, group.schema) if err != nil { - return fmt.Errorf("failed to create record for file %q: %w", p.metricGroups[name].filename, err) - } - if err = p.metricGroups[name].writer.WriteBuffered(record); err != nil { - return fmt.Errorf("failed to write to file %q: %w", p.metricGroups[name].filename, err) + return fmt.Errorf("failed to create record for file %q: %w", group.filename, err) } + err = group.writer.WriteBuffered(record) record.Release() + if err != nil { + return fmt.Errorf("failed to write to file %q: %w", group.filename, err) + } + writeErr.MetricsAccept = append(writeErr.MetricsAccept, indices...) } if len(writeErr.MetricsReject) == 0 { @@ -147,22 +141,39 @@ func (p *Parquet) Write(metrics []telegraf.Metric) error { return &writeErr } -const maxMeasurementLen = 255 - len("-2006-01-02-1234567890.parquet") +func (p *Parquet) groupFor(name string, metrics []telegraf.Metric, now time.Time) (*metricGroup, error) { + if group, found := p.metricGroups[name]; found { + return group, nil + } -func usableAsFilename(name string) bool { - return name != "" && len(name) <= maxMeasurementLen && !strings.ContainsFunc(name, reservedInFilename) -} + schema, err := p.createSchema(metrics) + if err != nil { + return nil, fmt.Errorf("failed to create schema for file %q: %w", name, err) + } + + filename := parquetFilename(name, now) + writer, err := p.createWriter(name, filename, schema) + if err != nil { + return nil, err + } -func reservedInFilename(r rune) bool { - if r == '/' || r == '\\' || unicode.IsControl(r) { - return true + group := &metricGroup{ + builder: array.NewRecordBuilder(memory.DefaultAllocator, schema), + filename: filename, + schema: schema, + writer: writer, } + p.metricGroups[name] = group - return runtime.GOOS == "windows" && strings.ContainsRune(`<>:"|?*`, r) + return group, nil +} + +func parquetFilename(name string, at time.Time) string { + return fmt.Sprintf("%s-%s-%s.parquet", name, at.Format("2006-01-02"), strconv.FormatInt(at.Unix(), 10)) } func (p *Parquet) rotateIfNeeded(name string) error { - fileInfo, err := os.Stat(p.metricGroups[name].filename) + fileInfo, err := p.root.Stat(p.metricGroups[name].filename) if err != nil { return fmt.Errorf("failed to stat file %q: %w", p.metricGroups[name].filename, err) } @@ -309,20 +320,20 @@ func (p *Parquet) createSchema(metrics []telegraf.Metric) (*arrow.Schema, error) } func (p *Parquet) createWriter(name, filename string, schema *arrow.Schema) (*pqarrow.FileWriter, error) { - if _, err := os.Stat(filename); err == nil { - now := time.Now() - rotatedFilename := fmt.Sprintf("%s/%s-%s-%s.parquet", p.Directory, name, now.Format("2006-01-02"), strconv.FormatInt(now.Unix(), 10)) - if err := os.Rename(filename, rotatedFilename); err != nil { + if _, err := p.root.Stat(filename); err == nil { + if err := p.root.Rename(filename, parquetFilename(name, time.Now())); err != nil { return nil, fmt.Errorf("failed to rename file %q: %w", filename, err) } } - file, err := os.Create(filename) + + file, err := p.root.Create(filename) if err != nil { return nil, fmt.Errorf("failed to create file %q: %w", filename, err) } writer, err := pqarrow.NewFileWriter(schema, file, parquet.NewWriterProperties(), pqarrow.DefaultWriterProps()) if err != nil { + file.Close() return nil, fmt.Errorf("failed to create parquet writer for file %q: %w", filename, err) } diff --git a/plugins/outputs/parquet/parquet_test.go b/plugins/outputs/parquet/parquet_test.go index f87327e55e4a5..f26993fa92dd7 100644 --- a/plugins/outputs/parquet/parquet_test.go +++ b/plugins/outputs/parquet/parquet_test.go @@ -3,7 +3,6 @@ package parquet import ( "os" "path/filepath" - "runtime" "slices" "strings" "testing" @@ -281,60 +280,73 @@ func TestMissingValuesReadBackAsNull(t *testing.T) { } } -func TestUsableAsFilename(t *testing.T) { +func TestUnusableMeasurementNamesAreRejected(t *testing.T) { tests := []struct { - name string - usable bool + name string + rejected bool }{ - {"cpu", true}, - {"disk.io", true}, - {"..", true}, + {"cpu", false}, + {"disk.io", false}, + {"..", false}, + {"../evil", true}, + {"../../../../tmp/evil", true}, + {"/etc/evil", true}, {"", false}, - {"../../etc/passwd", false}, - {`a\b`, false}, - {"nul\x00byte", false}, - {"tab\tnewline\n", false}, - {strings.Repeat("a", maxMeasurementLen), true}, - {strings.Repeat("a", maxMeasurementLen+1), false}, + {"nul\x00byte", true}, + {strings.Repeat("a", 300), true}, } for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { - require.Equal(t, tt.usable, usableAsFilename(tt.name)) - }) - } -} + dir := t.TempDir() + p := &Parquet{Directory: dir, TimestampFieldName: "timestamp", Log: testutil.Logger{}} + require.NoError(t, p.Init()) + + err := p.Write([]telegraf.Metric{ + metric.New(tt.name, nil, map[string]interface{}{"value": int64(1)}, time.Now()), + }) + require.NoError(t, p.Close()) + + written, globErr := filepath.Glob(filepath.Join(dir, "*.parquet")) + require.NoError(t, globErr) + + escaped, globErr := filepath.Glob(filepath.Join(filepath.Dir(dir), "*.parquet")) + require.NoError(t, globErr) + require.Empty(t, escaped) + + if !tt.rejected { + require.NoError(t, err) + require.Len(t, written, 1) + return + } -func TestWindowsReservedCharacters(t *testing.T) { - if runtime.GOOS != "windows" { - t.Skip("windows only") + var writeErr *internal.PartialWriteError + require.ErrorAs(t, err, &writeErr) + require.Equal(t, []int{0}, writeErr.MetricsReject) + require.Empty(t, writeErr.MetricsAccept) + require.Empty(t, written) + }) } - - require.False(t, usableAsFilename(`ac`)) } -func TestMeasurementNamesCannotEscapeDirectory(t *testing.T) { +func TestSymlinkedMeasurementNameCannotEscapeDirectory(t *testing.T) { dir := t.TempDir() - outside := filepath.Dir(dir) - before, err := filepath.Glob(filepath.Join(outside, "*.parquet")) - require.NoError(t, err) + outside := t.TempDir() + require.NoError(t, os.Symlink(outside, filepath.Join(dir, "link"))) p := &Parquet{Directory: dir, TimestampFieldName: "timestamp", Log: testutil.Logger{}} require.NoError(t, p.Init()) - for _, name := range []string{"../../../../tmp/evil", "../evil", `a\..\..\b`, "nul\x00byte"} { - m := metric.New(name, nil, map[string]interface{}{"value": int64(1)}, time.Now()) - require.ErrorAs(t, p.Write([]telegraf.Metric{m}), new(*internal.PartialWriteError)) - } + err := p.Write([]telegraf.Metric{ + metric.New("link/escaped", nil, map[string]interface{}{"value": int64(1)}, time.Now()), + }) require.NoError(t, p.Close()) - after, err := filepath.Glob(filepath.Join(outside, "*.parquet")) - require.NoError(t, err) - require.Equal(t, before, after) + require.ErrorAs(t, err, new(*internal.PartialWriteError)) - written, err := filepath.Glob(filepath.Join(dir, "*.parquet")) - require.NoError(t, err) - require.Empty(t, written) + escaped, globErr := filepath.Glob(filepath.Join(outside, "*")) + require.NoError(t, globErr) + require.Empty(t, escaped) } func TestUnusableMeasurementNameKeepsOutputWriting(t *testing.T) { @@ -373,8 +385,8 @@ func TestRejectedMetricsReportEveryIndexExactlyOnce(t *testing.T) { var writeErr *internal.PartialWriteError require.ErrorAs(t, p.Write(metrics), &writeErr) - require.Equal(t, []int{0, 2}, writeErr.MetricsReject) - require.Equal(t, []int{1, 3}, writeErr.MetricsAccept) + require.ElementsMatch(t, []int{0, 2}, writeErr.MetricsReject) + require.ElementsMatch(t, []int{1, 3}, writeErr.MetricsAccept) require.Len(t, writeErr.MetricsRejectErrors, 2) require.ElementsMatch(t, []int{0, 1, 2, 3},