Add file rotation based on file age to file output plugin (#5547)

This commit is contained in:
Kristoffer Berdal 2019-06-02 02:11:47 +02:00 committed by Daniel Nelson
parent d0beb19204
commit 1c0d3a0eb9
2 changed files with 35 additions and 11 deletions

View File

@ -8,6 +8,9 @@ This plugin writes telegraf metrics to files
## Files to write to, "stdout" is a specially handled file.
files = ["stdout", "/tmp/metrics.out"]
## If this is defined, files will be rotated by the time.Duration specified
# rotate_max_age = "1m"
## Data format to output.
## Each data format has its own unique set of configuration options, read
## more about them here:

View File

@ -4,16 +4,19 @@ import (
"fmt"
"io"
"os"
"time"
"github.com/influxdata/telegraf"
"github.com/influxdata/telegraf/internal/rotate"
"github.com/influxdata/telegraf/plugins/outputs"
"github.com/influxdata/telegraf/plugins/serializers"
)
type File struct {
Files []string
RotateMaxAge string
writers []io.Writer
writer io.Writer
closers []io.Closer
serializer serializers.Serializer
@ -23,6 +26,9 @@ var sampleConfig = `
## Files to write to, "stdout" is a specially handled file.
files = ["stdout", "/tmp/metrics.out"]
## If this is defined, files will be rotated by the time.Duration specified
# rotate_max_age = "1m"
## Data format to output.
## Each data format has its own unique set of configuration options, read
## more about them here:
@ -35,23 +41,38 @@ func (f *File) SetSerializer(serializer serializers.Serializer) {
}
func (f *File) Connect() error {
writers := []io.Writer{}
if len(f.Files) == 0 {
f.Files = []string{"stdout"}
}
for _, file := range f.Files {
if file == "stdout" {
f.writers = append(f.writers, os.Stdout)
writers = append(writers, os.Stdout)
} else {
of, err := os.OpenFile(file, os.O_CREATE|os.O_APPEND|os.O_WRONLY, os.ModeAppend|0644)
var of io.WriteCloser
var err error
if f.RotateMaxAge != "" {
maxAge, err := time.ParseDuration(f.RotateMaxAge)
if err != nil {
return err
}
f.writers = append(f.writers, of)
// Only rotate by file age for now, keep no archives.
of, err = rotate.NewFileWriter(file, maxAge, 0, -1)
} else {
// Just open a normal file
of, err = rotate.NewFileWriter(file, 0, 0, -1)
}
if err != nil {
return err
}
writers = append(writers, of)
f.closers = append(f.closers, of)
}
}
f.writer = io.MultiWriter(writers...)
return nil
}
@ -76,19 +97,19 @@ func (f *File) Description() string {
func (f *File) Write(metrics []telegraf.Metric) error {
var writeErr error = nil
for _, metric := range metrics {
b, err := f.serializer.Serialize(metric)
if err != nil {
return fmt.Errorf("failed to serialize message: %s", err)
}
for _, writer := range f.writers {
_, err = writer.Write(b)
if err != nil && writer != os.Stdout {
_, err = f.writer.Write(b)
if err != nil {
writeErr = fmt.Errorf("E! failed to write message: %s, %s", b, err)
}
}
}
return writeErr
}