package tail import ( "io/ioutil" "os" "runtime" "testing" "time" "github.com/influxdata/telegraf" "github.com/influxdata/telegraf/plugins/parsers" "github.com/influxdata/telegraf/plugins/parsers/csv" "github.com/influxdata/telegraf/plugins/parsers/json" "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") } 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.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(1) acc.AssertContainsTaggedFields(t, "cpu", map[string]interface{}{ "usage_idle": float64(100), }, map[string]string{ "mytag": "foo", "path": tmpfile.Name(), }) } func TestTailFromEnd(t *testing.T) { if os.Getenv("CIRCLE_PROJECT_REPONAME") != "" { t.Skip("Skipping CI testing due to race conditions") } 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.Files = []string{tmpfile.Name()} tt.SetParserFunc(parsers.NewInfluxParser) defer tt.Stop() defer tmpfile.Close() acc := testutil.Accumulator{} require.NoError(t, tt.Start(&acc)) 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() } } _, err = tmpfile.WriteString("cpu,othertag=foo usage_idle=100\n") require.NoError(t, err) require.NoError(t, acc.GatherError(tt.Gather)) acc.Wait(1) acc.AssertContainsTaggedFields(t, "cpu", map[string]interface{}{ "usage_idle": float64(100), }, map[string]string{ "othertag": "foo", "path": tmpfile.Name(), }) 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.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)) _, err = tmpfile.WriteString("cpu mytag= foo usage_idle= 100\n") require.NoError(t, err) acc.WaitError(1) assert.Contains(t, acc.Errors[0].Error(), "malformed log line") } 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.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.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.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()) }