Merge lp:~thumper/juju-core/fix-backlog-limits into lp:~go-bot/juju-core/trunk

[Tim Penhey]

Fix the limit for the debug-log calls.

Unbeknownst to me, the filter method was being called
during the backlog iteration as well. This fix introduces
a callback that the tailer calls when it starts the forward
filtering.

This allows us to count the filter calls only when the
filter is being called to write out the results. We cannot
count the write calls because the tailer uses buffered i/o
there.

https://codereview.appspot.com/85570045/

Fix the limit for the debug-log calls.

Unbeknownst to me, the filter method was being called
during the backlog iteration as well. This fix introduces
a callback that the tailer calls when it starts the forward
filtering.

This allows us to count the filter calls only when the
filter is being called to write out the results. We cannot
count the write calls because the tailer uses buffered i/o
there.

https://codereview.appspot.com/85570045/

[Tim Penhey (thumper)]

Reviewers: mp+215333_code.launchpad.net,

Message:
Please take a look.

Description:
Fix the limit for the debug-log calls.

Unbeknownst to me, the filter method was being called
during the backlog iteration as well. This fix introduces
a callback that the tailer calls when it starts the forward
filtering.

This allows us to count the filter calls only when the
filter is being called to write out the results. We cannot
count the write calls because the tailer uses buffered i/o
there.

https://code.launchpad.net/~thumper/juju-core/fix-backlog-limits/+merge/215333

Requires:
https://code.launchpad.net/~thumper/juju-core/debug-log-api/+merge/215323

(do not edit description out of merge proposal)

Please review this at https://codereview.appspot.com/85570045/

Affected files (+42, -10 lines):
   A [revision details]
   M container/lxc/clonetemplate.go
   M state/apiserver/debuglog.go
   M state/apiserver/debuglog_internal_test.go
   M state/apiserver/debuglog_test.go
   M utils/tailer/tailer.go
   M utils/tailer/tailer_test.go

Index: [revision details]
=== added file '[revision details]'
--- [revision details] 2012-01-01 00:00:00 +0000
+++ [revision details] 2012-01-01 00:00:00 +0000
@@ -0,0 +1,2 @@
+Old revision: <email address hidden>
+New revision: <email address hidden>

Index: container/lxc/clonetemplate.go
=== modified file 'container/lxc/clonetemplate.go'
--- container/lxc/clonetemplate.go 2014-04-04 03:46:13 +0000
+++ container/lxc/clonetemplate.go 2014-04-11 01:57:45 +0000
@@ -179,7 +179,7 @@
   }

   tailWriter := &logTail{tick: time.Now()}
- consoleTailer := tailer.NewTailer(console, tailWriter, nil)
+ consoleTailer := tailer.NewTailer(console, tailWriter, nil, nil)
   defer consoleTailer.Stop()

   // We should wait maybe 1 minute between output?

Index: state/apiserver/debuglog.go
=== modified file 'state/apiserver/debuglog.go'
--- state/apiserver/debuglog.go 2014-04-10 09:43:51 +0000
+++ state/apiserver/debuglog.go 2014-04-11 01:54:49 +0000
@@ -209,18 +209,23 @@
   maxLines uint
   lineCount uint
   fromTheStart bool
+ started bool
  }

  // start the tailer listening to the logFile, and sending the matching
  // lines to the writer.
  func (stream *logStream) start(logFile io.ReadSeeker, writer io.Writer) {
   if stream.fromTheStart {
- stream.logTailer = tailer.NewTailer(logFile, writer, stream.filterLine)
+ stream.logTailer = tailer.NewTailer(logFile, writer, stream.filterLine,
stream.tailStarted)
   } else {
- stream.logTailer = tailer.NewTailerBacktrack(logFile, writer,
stream.backlog, stream.filterLine)
+ stream.logTailer = tailer.NewTailerBacktrack(logFile, writer,
stream.backlog, stream.filterLine, stream.tailStarted)
   }
  }

+func (stream *logStream) tailStarted() {
+ stream.started = true
+}
+
  // loop starts the tailer with the log file and the web socket.
  func (stream *logStream) loop() error {
   select {
@@ -239,7 +244,7 @@
    stream.checkIncludeModule(log) &&
    !stream.exclude(log) &&
    stream.checkLevel(log)
- if result && stream.maxLines > 0 {
+ if stream.started && result && stream.maxLines > 0 {
    stream...

[Andrew Wilkins (axwalk)]

https://codereview.appspot.com/85570045/diff/1/utils/tailer/tailer.go
File utils/tailer/tailer.go (right):

https://codereview.appspot.com/85570045/diff/1/utils/tailer/tailer.go#newcode56
utils/tailer/tailer.go:56: filter TailerFilterFunc, callback
TailerFilterStartedFunc) *Tailer {
Rather than adding *another* parameter that's only relevant to
backtracking, I really think it would be best to ditch
NewTailerBacktrack and create a separate function that does the
backtracking.

https://codereview.appspot.com/85570045/diff/1/utils/tailer/tailer.go#newcode65
utils/tailer/tailer.go:65: filter TailerFilterFunc, callback
TailerFilterStartedFunc) *Tailer {
Isn't having a callback fairly pointless here? It's going to be called
immediately.

https://codereview.appspot.com/85570045/

[Roger Peppe (rogpeppe)]

As far as I can see, this makes a nice Tailer API into
a not-so-nice one because you've made a stateful
filter function and you want it to be called in just
the way you expect.

I don't think the filter function should be stateful.

I can see a couple of solutions that might be better:

1) (my preferred option) ditch the server-side maxLines
functionality completely. It could be implemented client-side by
simply closing the connection when the right number
of lines is received, but we've always got head(1),
so I'd suggest just losing it entirely.
2) add maxLines functionality to the Tailer itself.

1) might use a little extra bandwidth,
but I can't see that that would be great problem
in practice. If it is, the second option could
be implemented later.

https://codereview.appspot.com/85570045/

[Tim Penhey (thumper)]

Due to the buffered i/o in the tailer, I'd prefer to keep the line
parsing functionality in the server side.

However I think that axw's approach of breaking out the backtracking is
a good one, and that is the approach I'll use.

https://codereview.appspot.com/85570045/

[Tim Penhey (thumper)]

Please take a look.

https://codereview.appspot.com/85570045/

[Andrew Wilkins (axwalk)]

LGTM with a few suggestions

https://codereview.appspot.com/85570045/diff/20001/state/apiserver/debuglog.go
File state/apiserver/debuglog.go (right):

https://codereview.appspot.com/85570045/diff/20001/state/apiserver/debuglog.go#newcode225
state/apiserver/debuglog.go:225: err := tailer.SeekLastLines(logFile,
stream.backlog, stream.filterLine)
return tailer.SeekLastLines(...

https://codereview.appspot.com/85570045/diff/20001/state/apiserver/debuglog.go#newcode258
state/apiserver/debuglog.go:258: if stream.started && result &&
stream.maxLines > 0 {
It would be better to just split filterLine into something non-counting
and counting, where the latter calls the former and adds the linecount
check. Then you don't need the boolean.

https://codereview.appspot.com/85570045/diff/20001/utils/tailer/tailer.go
File utils/tailer/tailer.go (right):

https://codereview.appspot.com/85570045/diff/20001/utils/tailer/tailer.go#newcode31
utils/tailer/tailer.go:31: // TailerFilterStartedFunc is a callback that
is called when the filtering is
Delete this?

https://codereview.appspot.com/85570045/

[Roger Peppe (rogpeppe)]

> Due to the buffered i/o in the tailer, I'd prefer to keep the line
parsing
> functionality in the server side.

I don't quite understand this remark (the buffer is flushed whenever we
get to the end of the file, so it shouldn't make any significant
difference), but splitting out the backtracking code makes me much
happier.

LGTM with a couple of trivial suggestions below.

https://codereview.appspot.com/85570045/diff/20001/state/apiserver/debuglog.go
File state/apiserver/debuglog.go (right):

https://codereview.appspot.com/85570045/diff/20001/state/apiserver/debuglog.go#newcode224
state/apiserver/debuglog.go:224: if !stream.fromTheStart {
if stream.fromTheStart {
     return nil
}
...

saves a negative and a level of indentation.

https://codereview.appspot.com/85570045/diff/20001/state/apiserver/debuglog.go#newcode258
state/apiserver/debuglog.go:258: if stream.started && result &&
stream.maxLines > 0 {
On 2014/04/14 03:52:40, axw wrote:
> It would be better to just split filterLine into something
non-counting and
> counting, where the latter calls the former and adds the linecount
check. Then
> you don't need the boolean.

+1

https://codereview.appspot.com/85570045/diff/20001/utils/tailer/tailer.go
File utils/tailer/tailer.go (right):

https://codereview.appspot.com/85570045/diff/20001/utils/tailer/tailer.go#newcode31
utils/tailer/tailer.go:31: // TailerFilterStartedFunc is a callback that
is called when the filtering is
On 2014/04/14 03:52:40, axw wrote:
> Delete this?

+1

https://codereview.appspot.com/85570045/diff/20001/utils/tailer/tailer.go#newcode132
utils/tailer/tailer.go:132: // wanted number of filtered lines before
the end.
// If filter is non-nil, only lines for which filter returns
// true will be counted.

?

https://codereview.appspot.com/85570045/

=== modified file 'state/apiserver/debuglog.go'
--- state/apiserver/debuglog.go	2014-04-10 09:43:51 +0000
+++ state/apiserver/debuglog.go	2014-04-14 04:14:21 +0000
@@ -70,6 +70,11 @@
 				return
 			}
 			defer logFile.Close()
+			if err := stream.positionLogFile(logFile); err != nil {
+				h.sendError(socket, fmt.Errorf("cannot position log file: %v", err))
+				socket.Close()
+				return
+			}
 
 			// If we get to here, no more errors to report, so we report a nil
 			// error.  This way the first line of the socket is always a json
@@ -211,14 +216,20 @@
 	fromTheStart  bool
 }
 
+// positionLogFile will update the internal read position of the logFile to be
+// at the end of the file or somewhere in the middle if backlog has been specified.
+func (stream *logStream) positionLogFile(logFile io.ReadSeeker) error {
+	// Seek to the end, or lines back from the end if we need to.
+	if !stream.fromTheStart {
+		return tailer.SeekLastLines(logFile, stream.backlog, stream.filterLine)
+	}
+	return nil
+}
+
 // start the tailer listening to the logFile, and sending the matching
 // lines to the writer.
 func (stream *logStream) start(logFile io.ReadSeeker, writer io.Writer) {
-	if stream.fromTheStart {
-		stream.logTailer = tailer.NewTailer(logFile, writer, stream.filterLine)
-	} else {
-		stream.logTailer = tailer.NewTailerBacktrack(logFile, writer, stream.backlog, stream.filterLine)
-	}
+	stream.logTailer = tailer.NewTailer(logFile, writer, stream.countedFilterLine)
 }
 
 // loop starts the tailer with the log file and the web socket.
@@ -235,10 +246,17 @@
 // filterLine checks the received line for one of the confgured tags.
 func (stream *logStream) filterLine(line []byte) bool {
 	log := parseLogLine(string(line))
-	result := stream.checkIncludeEntity(log) &&
+	return stream.checkIncludeEntity(log) &&
 		stream.checkIncludeModule(log) &&
 		!stream.exclude(log) &&
 		stream.checkLevel(log)
+}
+
+// countedFilterLine checks the received line for one of the confgured tags,
+// and also checks to make sure the stream doesn't send more than the
+// specified number of lines.
+func (stream *logStream) countedFilterLine(line []byte) bool {
+	result := stream.filterLine(line)
 	if result && stream.maxLines > 0 {
 		stream.lineCount++
 		result = stream.lineCount <= stream.maxLines

=== modified file 'state/apiserver/debuglog_internal_test.go'
--- state/apiserver/debuglog_internal_test.go	2014-04-07 05:11:48 +0000
+++ state/apiserver/debuglog_internal_test.go	2014-04-14 04:14:21 +0000
@@ -7,7 +7,6 @@
 
 import (
 	"bytes"
-	"io"
 	"net/url"
 	"os"
 	"path/filepath"
@@ -179,36 +178,22 @@
 		"machine-0: date time WARNING juju.foo.bar")), jc.IsFalse)
 }
 
-func (s *debugInternalSuite) TestFilterLineWithLimit(c *gc.C) {
+func (s *debugInternalSuite) TestCountedFilterLineWithLimit(c *gc.C) {
 	stream := &logStream{
 		filterLevel: loggo.INFO,
 		maxLines:    5,
 	}
 	line := []byte("machine-0: date time WARNING juju")
-	c.Check(stream.filterLine(line), jc.IsTrue)
-	c.Check(stream.filterLine(line), jc.IsTrue)
-	c.Check(stream.filterLine(line), jc.IsTrue)
-	c.Check(stream.filterLine(line), jc.IsTrue)
-	c.Check(stream.filterLine(line), jc.IsTrue)
-	c.Check(stream.filterLine(line), jc.IsFalse)
-	c.Check(stream.filterLine(line), jc.IsFalse)
-}
-
-type seekWaitReader struct {
-	io.ReadSeeker
-	wait chan struct{}
-}
-
-func (w *seekWaitReader) Seek(offset int64, whence int) (int64, error) {
-	pos, err := w.ReadSeeker.Seek(offset, whence)
-	if w.wait != nil {
-		close(w.wait)
-		w.wait = nil
-	}
-	return pos, err
-}
-
-func (s *debugInternalSuite) testStreamInternal(c *gc.C, fromTheStart bool, maxLines uint, expected, errMatch string) {
+	c.Check(stream.countedFilterLine(line), jc.IsTrue)
+	c.Check(stream.countedFilterLine(line), jc.IsTrue)
+	c.Check(stream.countedFilterLine(line), jc.IsTrue)
+	c.Check(stream.countedFilterLine(line), jc.IsTrue)
+	c.Check(stream.countedFilterLine(line), jc.IsTrue)
+	c.Check(stream.countedFilterLine(line), jc.IsFalse)
+	c.Check(stream.countedFilterLine(line), jc.IsFalse)
+}
+
+func (s *debugInternalSuite) testStreamInternal(c *gc.C, fromTheStart bool, backlog, maxLines uint, expected, errMatch string) {
 
 	dir := c.MkDir()
 	logPath := filepath.Join(dir, "logfile.txt")
@@ -223,17 +208,20 @@
 line 2
 line 3
 `)
-	stream := &logStream{fromTheStart: fromTheStart, maxLines: maxLines}
+	stream := &logStream{
+		fromTheStart: fromTheStart,
+		backlog:      backlog,
+		maxLines:     maxLines,
+	}
+	err = stream.positionLogFile(logFileReader)
+	c.Assert(err, gc.IsNil)
 	output := &bytes.Buffer{}
-	waitReader := &seekWaitReader{logFileReader, make(chan struct{})}
-	stream.start(waitReader, output)
+	stream.start(logFileReader, output)
 
 	go func() {
 		defer stream.tomb.Done()
 		stream.tomb.Kill(stream.loop())
 	}()
-	// wait for the tailer to have started tailing before writing more
-	<-waitReader.wait
 
 	logFile.WriteString("line 4\n")
 	logFile.WriteString("line 5\n")
@@ -265,7 +253,7 @@
 line 4
 line 5
 `
-	s.testStreamInternal(c, true, 0, expected, "")
+	s.testStreamInternal(c, true, 0, 0, expected, "")
 }
 
 func (s *debugInternalSuite) TestLogStreamLoopFromTheStartMaxLines(c *gc.C) {
@@ -273,21 +261,28 @@
 line 2
 line 3
 `
-	s.testStreamInternal(c, true, 3, expected, "max lines reached")
+	s.testStreamInternal(c, true, 0, 3, expected, "max lines reached")
 }
 
 func (s *debugInternalSuite) TestLogStreamLoopJustTail(c *gc.C) {
 	expected := `line 4
 line 5
 `
-	s.testStreamInternal(c, false, 0, expected, "")
+	s.testStreamInternal(c, false, 0, 0, expected, "")
+}
+
+func (s *debugInternalSuite) TestLogStreamLoopBackOneLimitTwo(c *gc.C) {
+	expected := `line 3
+line 4
+`
+	s.testStreamInternal(c, false, 1, 2, expected, "max lines reached")
 }
 
 func (s *debugInternalSuite) TestLogStreamLoopTailMaxLinesNotYetReached(c *gc.C) {
 	expected := `line 4
 line 5
 `
-	s.testStreamInternal(c, false, 3, expected, "")
+	s.testStreamInternal(c, false, 0, 3, expected, "")
 }
 
 func assertStreamParams(c *gc.C, obtained, expected *logStream) {

=== modified file 'state/apiserver/debuglog_test.go'
--- state/apiserver/debuglog_test.go	2014-04-10 09:43:51 +0000
+++ state/apiserver/debuglog_test.go	2014-04-14 04:14:21 +0000
@@ -122,6 +122,18 @@
 	s.assertWebsocketClosed(c, reader)
 }
 
+func (s *debugLogSuite) TestBacklogWithMaxLines(c *gc.C) {
+	s.writeLogLines(c, 10)
+
+	reader := s.openWebsocket(c, url.Values{"backlog": {"5"}, "maxLines": {"10"}})
+	s.assertLogFollowing(c, reader)
+	s.writeLogLines(c, logLineCount)
+
+	linesRead := s.readLogLines(c, reader, 10)
+	c.Assert(linesRead, jc.DeepEquals, logLines[5:15])
+	s.assertWebsocketClosed(c, reader)
+}
+
 func (s *debugLogSuite) TestFilter(c *gc.C) {
 	s.ensureLogFile(c)
 

=== modified file 'utils/tailer/export_test.go'
--- utils/tailer/export_test.go	2013-12-11 17:11:30 +0000
+++ utils/tailer/export_test.go	2014-04-14 04:14:21 +0000
@@ -3,4 +3,7 @@
 
 package tailer
 
-var NewTestTailer = newTailer
+var (
+	BufferSize    = &bufferSize
+	NewTestTailer = newTailer
+)

=== modified file 'utils/tailer/tailer.go'
--- utils/tailer/tailer.go	2014-04-04 04:35:44 +0000
+++ utils/tailer/tailer.go	2014-04-14 04:14:21 +0000
@@ -14,12 +14,13 @@
 )
 
 const (
-	bufferSize = 4096
-	polltime   = time.Second
-	delimiter  = '\n'
+	defaultBufferSize = 4096
+	polltime          = time.Second
+	delimiter         = '\n'
 )
 
 var (
+	bufferSize = defaultBufferSize
 	delimiters = []byte{delimiter}
 )
 
@@ -35,20 +36,8 @@
 	reader      *bufio.Reader
 	writeCloser io.WriteCloser
 	writer      *bufio.Writer
-	lines       uint
 	filter      TailerFilterFunc
-	bufferSize  int
 	polltime    time.Duration
-	lookBack    bool
-}
-
-// NewTailerBacktrack starts a Tailer which reads strings from the passed
-// ReadSeeker line by line. If a filter function is specified the read
-// lines are filtered. The matching lines are written to the passed
-// Writer. The reading begins the specified number of matching lines
-// from the end.
-func NewTailerBacktrack(readSeeker io.ReadSeeker, writer io.Writer, lines uint, filter TailerFilterFunc) *Tailer {
-	return newTailer(readSeeker, writer, lines, filter, bufferSize, polltime, true)
 }
 
 // NewTailer starts a Tailer which reads strings from the passed
@@ -56,22 +45,19 @@
 // lines are filtered. The matching lines are written to the passed
 // Writer.
 func NewTailer(readSeeker io.ReadSeeker, writer io.Writer, filter TailerFilterFunc) *Tailer {
-	return newTailer(readSeeker, writer, 0, filter, bufferSize, polltime, false)
+	return newTailer(readSeeker, writer, filter, polltime)
 }
 
 // newTailer starts a Tailer like NewTailer but allows the setting of
 // the read buffer size and the time between pollings for testing.
-func newTailer(readSeeker io.ReadSeeker, writer io.Writer, lines uint, filter TailerFilterFunc,
-	bufferSize int, polltime time.Duration, lookBack bool) *Tailer {
+func newTailer(readSeeker io.ReadSeeker, writer io.Writer,
+	filter TailerFilterFunc, polltime time.Duration) *Tailer {
 	t := &Tailer{
 		readSeeker: readSeeker,
 		reader:     bufio.NewReaderSize(readSeeker, bufferSize),
 		writer:     bufio.NewWriter(writer),
-		lines:      lines,
 		filter:     filter,
-		bufferSize: bufferSize,
 		polltime:   polltime,
-		lookBack:   lookBack,
 	}
 	go func() {
 		defer t.tomb.Done()
@@ -107,12 +93,6 @@
 // writer and then polls for more data to write it to the
 // writer too.
 func (t *Tailer) loop() error {
-	// Position the readSeeker.
-	if t.lookBack {
-		if err := t.seekLastLines(); err != nil {
-			return err
-		}
-	}
 	// Start polling.
 	// TODO(mue) 2013-12-06
 	// Handling of read-seeker/files being truncated during
@@ -144,26 +124,26 @@
 	}
 }
 
-// seekLastLines sets the read position of the ReadSeeker to the
+// SeekLastLines sets the read position of the ReadSeeker to the
 // wanted number of filtered lines before the end.
-func (t *Tailer) seekLastLines() error {
-	offset, err := t.readSeeker.Seek(0, os.SEEK_END)
+func SeekLastLines(readSeeker io.ReadSeeker, lines uint, filter TailerFilterFunc) error {
+	offset, err := readSeeker.Seek(0, os.SEEK_END)
 	if err != nil {
 		return err
 	}
-	if t.lines == 0 {
+	if lines == 0 {
 		// We are done, just seeking to the end is sufficient.
 		return nil
 	}
 	seekPos := int64(0)
 	found := uint(0)
-	buffer := make([]byte, t.bufferSize)
+	buffer := make([]byte, bufferSize)
 SeekLoop:
 	for offset > 0 {
 		// buffer contains the data left over from the
 		// previous iteration.
 		space := cap(buffer) - len(buffer)
-		if space < t.bufferSize {
+		if space < bufferSize {
 			// Grow buffer.
 			newBuffer := make([]byte, len(buffer), cap(buffer)*2)
 			copy(newBuffer, buffer)
@@ -180,11 +160,11 @@
 		copy(buffer[space:cap(buffer)], buffer)
 		buffer = buffer[0 : len(buffer)+space]
 		offset -= int64(space)
-		_, err := t.readSeeker.Seek(offset, os.SEEK_SET)
+		_, err := readSeeker.Seek(offset, os.SEEK_SET)
 		if err != nil {
 			return err
 		}
-		_, err = io.ReadFull(t.readSeeker, buffer[0:space])
+		_, err = io.ReadFull(readSeeker, buffer[0:space])
 		if err != nil {
 			return err
 		}
@@ -207,9 +187,9 @@
 				break
 			}
 			start++
-			if t.isValid(buffer[start:end]) {
+			if filter == nil || filter(buffer[start:end]) {
 				found++
-				if found >= t.lines {
+				if found >= lines {
 					seekPos = offset + int64(start)
 					break SeekLoop
 				}
@@ -221,7 +201,7 @@
 		buffer = buffer[0:end]
 	}
 	// Final positioning.
-	t.readSeeker.Seek(seekPos, os.SEEK_SET)
+	readSeeker.Seek(seekPos, os.SEEK_SET)
 	return nil
 }
 

=== modified file 'utils/tailer/tailer_test.go'
--- utils/tailer/tailer_test.go	2014-04-04 03:46:13 +0000
+++ utils/tailer/tailer_test.go	2014-04-14 04:14:21 +0000
@@ -15,6 +15,7 @@
 	gc "launchpad.net/gocheck"
 
 	"launchpad.net/juju-core/testing"
+	"launchpad.net/juju-core/testing/testbase"
 	"launchpad.net/juju-core/utils/tailer"
 )
 
@@ -22,9 +23,11 @@
 	gc.TestingT(t)
 }
 
-type tailerSuite struct{}
+type tailerSuite struct {
+	testbase.LoggingSuite
+}
 
-var _ = gc.Suite(tailerSuite{})
+var _ = gc.Suite(&tailerSuite{})
 
 var alphabetData = []string{
 	"alpha alpha\n",
@@ -326,7 +329,7 @@
 	},
 }}
 
-func (tailerSuite) TestTailer(c *gc.C) {
+func (s *tailerSuite) TestTailer(c *gc.C) {
 	for i, test := range tests {
 		c.Logf("Test #%d) %s", i, test.description)
 		bufferSize := test.bufferSize
@@ -334,10 +337,15 @@
 			// Default value.
 			bufferSize = 4096
 		}
+		s.PatchValue(tailer.BufferSize, bufferSize)
 		reader, writer := io.Pipe()
 		sigc := make(chan struct{}, 1)
 		rs := startReadSeeker(c, test.data, test.initialLinesWritten, sigc)
-		tailer := tailer.NewTestTailer(rs, writer, test.initialLinesRequested, test.filter, bufferSize, 2*time.Millisecond, !test.fromStart)
+		if !test.fromStart {
+			err := tailer.SeekLastLines(rs, test.initialLinesRequested, test.filter)
+			c.Assert(err, gc.IsNil)
+		}
+		tailer := tailer.NewTestTailer(rs, writer, test.filter, 2*time.Millisecond)
 		linec := startReading(c, tailer, reader, writer)
 
 		// Collect initial data.