telegraf/plugins/inputs/tail/tail_test.go

271 lines
6.3 KiB
Go
Raw Normal View History

2016-04-22 21:47:26 +00:00
package tail
2016-04-26 16:43:41 +00:00
import (
"bytes"
2016-04-26 16:43:41 +00:00
"io/ioutil"
"log"
2016-04-26 16:43:41 +00:00
"os"
2017-03-24 19:03:36 +00:00
"runtime"
2016-04-26 16:43:41 +00:00
"testing"
"time"
2016-04-26 16:43:41 +00:00
"github.com/influxdata/telegraf"
2016-04-26 16:43:41 +00:00
"github.com/influxdata/telegraf/plugins/parsers"
"github.com/influxdata/telegraf/plugins/parsers/csv"
"github.com/influxdata/telegraf/plugins/parsers/json"
2016-04-26 16:43:41 +00:00
"github.com/influxdata/telegraf/testutil"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
)
func TestTailFromBeginning(t *testing.T) {
if os.Getenv("CIRCLE_PROJECT_REPONAME") != "" {
t.Skip("Skipping CI testing due to race conditions")
}
2016-04-26 16:43:41 +00:00
tmpfile, err := ioutil.TempFile("", "")
require.NoError(t, err)
defer os.Remove(tmpfile.Name())
2016-07-28 13:08:12 +00:00
_, err = tmpfile.WriteString("cpu,mytag=foo usage_idle=100\n")
require.NoError(t, err)
2016-04-26 16:43:41 +00:00
tt := NewTail()
tt.Log = testutil.Logger{}
2016-04-26 16:43:41 +00:00
tt.FromBeginning = true
tt.Files = []string{tmpfile.Name()}
tt.SetParserFunc(parsers.NewInfluxParser)
2016-04-26 16:43:41 +00:00
defer tt.Stop()
defer tmpfile.Close()
acc := testutil.Accumulator{}
require.NoError(t, tt.Start(&acc))
2017-04-24 18:13:26 +00:00
require.NoError(t, acc.GatherError(tt.Gather))
2016-04-26 16:43:41 +00:00
2017-03-24 19:03:36 +00:00
acc.Wait(1)
2016-04-26 16:43:41 +00:00
acc.AssertContainsTaggedFields(t, "cpu",
map[string]interface{}{
"usage_idle": float64(100),
},
map[string]string{
"mytag": "foo",
2018-07-10 00:39:51 +00:00
"path": tmpfile.Name(),
2016-04-26 16:43:41 +00:00
})
}
func TestTailFromEnd(t *testing.T) {
2018-01-13 00:48:45 +00:00
if os.Getenv("CIRCLE_PROJECT_REPONAME") != "" {
t.Skip("Skipping CI testing due to race conditions")
}
2016-04-26 16:43:41 +00:00
tmpfile, err := ioutil.TempFile("", "")
require.NoError(t, err)
defer os.Remove(tmpfile.Name())
_, err = tmpfile.WriteString("cpu,mytag=foo usage_idle=100\n")
require.NoError(t, err)
tt := NewTail()
tt.Log = testutil.Logger{}
2016-04-26 16:43:41 +00:00
tt.Files = []string{tmpfile.Name()}
tt.SetParserFunc(parsers.NewInfluxParser)
2016-04-26 16:43:41 +00:00
defer tt.Stop()
defer tmpfile.Close()
acc := testutil.Accumulator{}
require.NoError(t, tt.Start(&acc))
2017-03-24 19:03:36 +00:00
for _, tailer := range tt.tailers {
for n, err := tailer.Tell(); err == nil && n == 0; n, err = tailer.Tell() {
// wait for tailer to jump to end
runtime.Gosched()
}
}
2016-04-26 16:43:41 +00:00
_, err = tmpfile.WriteString("cpu,othertag=foo usage_idle=100\n")
require.NoError(t, err)
2017-04-24 18:13:26 +00:00
require.NoError(t, acc.GatherError(tt.Gather))
2016-04-26 16:43:41 +00:00
2017-03-24 19:03:36 +00:00
acc.Wait(1)
2016-04-26 16:43:41 +00:00
acc.AssertContainsTaggedFields(t, "cpu",
map[string]interface{}{
"usage_idle": float64(100),
},
map[string]string{
"othertag": "foo",
2018-07-10 00:39:51 +00:00
"path": tmpfile.Name(),
2016-04-26 16:43:41 +00:00
})
assert.Len(t, acc.Metrics, 1)
}
func TestTailBadLine(t *testing.T) {
tmpfile, err := ioutil.TempFile("", "")
require.NoError(t, err)
defer os.Remove(tmpfile.Name())
tt := NewTail()
tt.Log = testutil.Logger{}
2016-04-26 16:43:41 +00:00
tt.FromBeginning = true
tt.Files = []string{tmpfile.Name()}
tt.SetParserFunc(parsers.NewInfluxParser)
2016-04-26 16:43:41 +00:00
defer tt.Stop()
defer tmpfile.Close()
acc := testutil.Accumulator{}
require.NoError(t, tt.Start(&acc))
buf := &bytes.Buffer{}
log.SetOutput(buf)
require.NoError(t, acc.GatherError(tt.Gather))
2016-04-26 16:43:41 +00:00
_, err = tmpfile.WriteString("cpu mytag= foo usage_idle= 100\n")
require.NoError(t, err)
time.Sleep(500 * time.Millisecond)
assert.Contains(t, buf.String(), "Malformed log line")
2016-04-26 16:43:41 +00:00
}
func TestTailDosLineendings(t *testing.T) {
tmpfile, err := ioutil.TempFile("", "")
require.NoError(t, err)
defer os.Remove(tmpfile.Name())
_, err = tmpfile.WriteString("cpu usage_idle=100\r\ncpu2 usage_idle=200\r\n")
require.NoError(t, err)
tt := NewTail()
tt.Log = testutil.Logger{}
tt.FromBeginning = true
tt.Files = []string{tmpfile.Name()}
tt.SetParserFunc(parsers.NewInfluxParser)
defer tt.Stop()
defer tmpfile.Close()
acc := testutil.Accumulator{}
require.NoError(t, tt.Start(&acc))
require.NoError(t, acc.GatherError(tt.Gather))
acc.Wait(2)
acc.AssertContainsFields(t, "cpu",
map[string]interface{}{
"usage_idle": float64(100),
})
acc.AssertContainsFields(t, "cpu2",
map[string]interface{}{
"usage_idle": float64(200),
})
}
// The csv parser should only parse the header line once per file.
func TestCSVHeadersParsedOnce(t *testing.T) {
tmpfile, err := ioutil.TempFile("", "")
require.NoError(t, err)
defer func() {
tmpfile.Close()
os.Remove(tmpfile.Name())
}()
_, err = tmpfile.WriteString(`
measurement,time_idle
cpu,42
cpu,42
`)
require.NoError(t, err)
plugin := NewTail()
plugin.Log = testutil.Logger{}
plugin.FromBeginning = true
plugin.Files = []string{tmpfile.Name()}
plugin.SetParserFunc(func() (parsers.Parser, error) {
return &csv.Parser{
MeasurementColumn: "measurement",
HeaderRowCount: 1,
TimeFunc: func() time.Time { return time.Unix(0, 0) },
}, nil
})
defer plugin.Stop()
acc := testutil.Accumulator{}
err = plugin.Start(&acc)
require.NoError(t, err)
err = plugin.Gather(&acc)
require.NoError(t, err)
acc.Wait(2)
plugin.Stop()
expected := []telegraf.Metric{
testutil.MustMetric("cpu",
map[string]string{
"path": tmpfile.Name(),
},
map[string]interface{}{
"time_idle": 42,
"measurement": "cpu",
},
time.Unix(0, 0)),
testutil.MustMetric("cpu",
map[string]string{
"path": tmpfile.Name(),
},
map[string]interface{}{
"time_idle": 42,
"measurement": "cpu",
},
time.Unix(0, 0)),
}
testutil.RequireMetricsEqual(t, expected, acc.GetTelegrafMetrics())
}
// Ensure that the first line can produce multiple metrics (#6138)
func TestMultipleMetricsOnFirstLine(t *testing.T) {
tmpfile, err := ioutil.TempFile("", "")
require.NoError(t, err)
defer func() {
tmpfile.Close()
os.Remove(tmpfile.Name())
}()
_, err = tmpfile.WriteString(`
[{"time_idle": 42}, {"time_idle": 42}]
`)
require.NoError(t, err)
plugin := NewTail()
plugin.Log = testutil.Logger{}
plugin.FromBeginning = true
plugin.Files = []string{tmpfile.Name()}
plugin.SetParserFunc(func() (parsers.Parser, error) {
return json.New(
&json.Config{
MetricName: "cpu",
})
})
defer plugin.Stop()
acc := testutil.Accumulator{}
err = plugin.Start(&acc)
require.NoError(t, err)
err = plugin.Gather(&acc)
require.NoError(t, err)
acc.Wait(2)
plugin.Stop()
expected := []telegraf.Metric{
testutil.MustMetric("cpu",
map[string]string{
"path": tmpfile.Name(),
},
map[string]interface{}{
"time_idle": 42.0,
},
time.Unix(0, 0)),
testutil.MustMetric("cpu",
map[string]string{
"path": tmpfile.Name(),
},
map[string]interface{}{
"time_idle": 42.0,
},
time.Unix(0, 0)),
}
testutil.RequireMetricsEqual(t, expected, acc.GetTelegrafMetrics(),
testutil.IgnoreTime())
}