add forwarder test (currently failing)

- add doc comment to logForwarder

internals/overlord/logstate/forwarder.go

@@ -11,6 +11,7 @@
 //
 // You should have received a copy of the GNU General Public License
 // along with this program.  If not, see <http://www.gnu.org/licenses/>.
+
 package logstate
 
 import (
@@ -20,6 +21,12 @@ import (
 	"github.com/canonical/pebble/internals/servicelog"
 )
 
+// logForwarder is responsible for pulling logs from a service's ringbuffer,
+// and distributing each log message to its logGatherers. Its gatherers field
+// holds a reference to the gatherer for each log target that the service is
+// sending logs to.
+// One logForwarder will run per service. Its forward() method should be run
+// in its own goroutine.
 type logForwarder struct {
 	serviceName string
 
@@ -33,7 +40,6 @@ func newLogForwarder(serviceName string) *logForwarder {
 	f := &logForwarder{
 		serviceName: serviceName,
 		cancel:      make(chan struct{}),
-		// TODO
 	}
 
 	return f
@@ -52,22 +58,14 @@ func (f *logForwarder) forward(buffer *servicelog.RingBuffer) {
 			f.mu.Unlock()
 			for _, c := range gatherers {
 				c.addLog(entry)
-				//if err != nil {
-				//	logger.Noticef("Cannot write log entry for service %q to %q: %v",
-				//		f.service, c.Target().Name, err)}
 			}
 		}
 		if err := parser.Err(); err != nil {
-			logger.Noticef(
-				"Cannot read logs from service %q: %v",
-				f.serviceName, err)
+			logger.Noticef("Cannot read logs from service %q: %v", f.serviceName, err)
 		}
 	}
 }
 
 func (f *logForwarder) stop() {
 	close(f.cancel)
 }
-
-// TODO
-//   should we have a forwarder for services w/ no log targets

internals/overlord/logstate/forwarder_test.go

@@ -0,0 +1,66 @@
+// Copyright (c) 2023 Canonical Ltd
+//
+// This program is free software: you can redistribute it and/or modify
+// it under the terms of the GNU General Public License version 3 as
+// published by the Free Software Foundation.
+//
+// This program is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+// GNU General Public License for more details.
+//
+// You should have received a copy of the GNU General Public License
+// along with this program.  If not, see <http://www.gnu.org/licenses/>.
+
+package logstate
+
+import (
+	"fmt"
+	"time"
+
+	"github.com/canonical/pebble/internals/servicelog"
+	. "gopkg.in/check.v1"
+)
+
+type forwarderSuite struct{}
+
+var _ = Suite(&forwarderSuite{})
+
+func (s *forwarderSuite) TestForwarder(c *C) {
+	serviceName := "foobar"
+	ringBuffer := servicelog.NewRingBuffer(1024)
+	logWriter := servicelog.NewFormatWriter(ringBuffer, serviceName)
+
+	recv1, recv2 := make(chan []servicelog.Entry), make(chan []servicelog.Entry)
+	gatherer1 := newLogGathererForTest(nil, 1*time.Microsecond, 5, recv1)
+	gatherer2 := newLogGathererForTest(nil, 1*time.Microsecond, 5, recv2)
+
+	forwarder := newLogForwarder(serviceName)
+	go forwarder.forward(ringBuffer)
+
+	forwarder.mu.Lock()
+	forwarder.gatherers = []*logGatherer{gatherer1, gatherer2}
+	forwarder.mu.Unlock()
+
+	message := "this is a log line"
+	_, err := fmt.Fprintln(logWriter, message)
+	c.Assert(err, IsNil)
+
+	select {
+	case entries := <-recv1:
+		c.Assert(entries, HasLen, 1)
+		c.Check(entries[0].Service, Equals, serviceName)
+		c.Check(entries[0].Message, Equals, message)
+	case <-time.After(10 * time.Millisecond):
+		c.Fatal("timeout waiting to receive logs from gatherer1")
+	}
+
+	select {
+	case entries := <-recv2:
+		c.Assert(entries, HasLen, 1)
+		c.Check(entries[0].Service, Equals, serviceName)
+		c.Check(entries[0].Message, Equals, message)
+	case <-time.After(10 * time.Millisecond):
+		c.Fatal("timeout waiting to receive logs from gatherer2")
+	}
+}

internals/overlord/logstate/gatherer.go

@@ -11,6 +11,7 @@
 //
 // You should have received a copy of the GNU General Public License
 // along with this program.  If not, see <http://www.gnu.org/licenses/>.
+
 package logstate
 
 import (
@@ -30,7 +31,7 @@ import (
 // One logGatherer will run per log target. Its loop() method should be run
 // in its own goroutine, while the addLog() method can be invoked in a
 // separate goroutine by a logForwarder.
-// The logGatherer will "flush" and send a request to the client when:
+// The logGatherer will "flush" and send a request to the client:
 //   - on a regular cadence (e.g. every 1 second)
 //   - when the buffer reaches a certain size
 //   - when it is told to shut down.

internals/overlord/logstate/gatherer_test.go

@@ -11,6 +11,7 @@
 //
 // You should have received a copy of the GNU General Public License
 // along with this program.  If not, see <http://www.gnu.org/licenses/>.
+
 package logstate
 
 import (
@@ -64,10 +65,9 @@ func (s *gathererSuite) TestGathererBufferFull(c *C) {
 	c.Assert(g.buffer.IsFull(), Equals, true)
 
 	select {
-	case entries := <-recv:
-		c.Assert(entries, DeepEquals, entries)
+	case received := <-recv:
+		c.Assert(received, DeepEquals, entries)
 	case <-time.After(10 * time.Millisecond):
-		c.Logf("buffer is full: %v", g.buffer.IsFull())
 		c.Fatal("timeout waiting to receive logs")
 	}
 }

internals/overlord/logstate/manager_test.go

@@ -11,6 +11,7 @@
 //
 // You should have received a copy of the GNU General Public License
 // along with this program.  If not, see <http://www.gnu.org/licenses/>.
+
 package logstate
 
 import (

internals/overlord/logstate/package_test.go

@@ -11,6 +11,7 @@
 //
 // You should have received a copy of the GNU General Public License
 // along with this program.  If not, see <http://www.gnu.org/licenses/>.
+
 package logstate
 
 import (