feat(logfwd): support forwarding logs to Loki

.github/.golangci.yml

@@ -12,7 +12,5 @@ issues:
   # these values ensure that all issues will be surfaced
   max-issues-per-linter: 0
   max-same-issues: 0
-
 run:
-  timeout: 5m
-
+  timeout: 5m

README.md

@@ -403,30 +403,26 @@ $ pebble run --verbose
 ...
 ```
 
-<!--
-TODO: uncomment this section once log forwarding is fully implemented
-TODO: add log targets to the Pebble layer spec below
-
 #### Log forwarding
 
-Pebble supports forwarding its services' logs to a remote Loki server or syslog receiver (via UDP/TCP). In the `log-targets` section of the plan, you can specify destinations for log forwarding, for example:
+Pebble supports forwarding its services' logs to a remote Loki server. In the `log-targets` section of the plan, you can specify destinations for log forwarding, for example:
 ```yaml
 log-targets:
-    loki-example:
+    example1:
         override: merge
         type: loki
         location: http://10.1.77.205:3100/loki/api/v1/push
         services: [all]
-    syslog-example:
+    example2:
         override: merge
-        type: syslog
-        location: tcp://192.168.10.241:1514
+        type: loki
+        location: http://my.loki.server.com/loki/api/v1/push
         services: [svc1, svc2]
 ```
 
-For each log target, use the `services` key to specify a list of services to collect logs from. In the above example, the `syslog-example` target will collect logs from `svc1` and `svc2`.
+For each log target, use the `services` key to specify a list of services to collect logs from. In the above example, the `example2` target will collect logs from `svc1` and `svc2`.
 
-Use the special keyword `all` to match all services, including services that might be added in future layers. In the above example, `loki-example` will collect logs from all services.
+Use the special keyword `all` to match all services, including services that might be added in future layers. In the above example, `example1` will collect logs from all services.
 
 To remove a service from a log target when merging, prefix the service name with a minus `-`. For example, if we have a base layer with
 ```yaml
@@ -455,7 +451,6 @@ my-target:
 ```
 would remove all services and then add `svc1`, so `my-target` would receive logs from only `svc1`.
 
--->
 
 ## Container usage
 
@@ -721,6 +716,37 @@ checks:
             # (Optional) Working directory to run command in. By default, the
             # command is run in the service manager's current directory.
             working-dir: <directory>
+
+# (Optional) A list of remote log receivers, to which service logs can be sent.
+log-targets:
+
+  <log target name>:
+
+    # (Required) Control how this log target definition is combined with
+    # other pre-existing definitions with the same name in the Pebble plan.
+    #
+    # The value 'merge' will ensure that values in this layer specification
+    # are merged over existing definitions, whereas 'replace' will entirely
+    # override the existing target spec in the plan with the same name.
+    override: merge | replace
+
+    # (Required) The type of log target, which determines the format in
+    # which logs will be sent. Currently, the only supported type is 'loki',
+    # but more protocols may be added in the future.
+    type: loki
+
+    # (Required) The URL of the remote log target.
+    # For Loki, this needs to be the fully-qualified URL of the push API,
+    # including the API endpoint, e.g.
+    #     http://<ip-address>:3100/loki/api/v1/push
+    location: <url>
+
+    # (Optional) A list of services whose logs will be sent to this target.
+    # Use the special keyword 'all' to match all services in the plan.
+    # When merging log targets, the 'services' lists are appended. Prefix a
+    # service name with a minus (e.g. '-svc1') to remove a previously added
+    # service. '-all' will remove all services.
+    services: [<service names>]
 ```
 
 ## API and clients
@@ -753,7 +779,8 @@ Here are some of the things coming soon:
   - [x] Automatically restart services that fail
   - [x] Support for custom health checks (HTTP, TCP, command)
   - [x] Terminate all services before exiting run command
-  - [ ] Log forwarding (syslog and Loki)
+  - [x] Log forwarding to Loki
+  - [ ] Log forwarding to syslog
   - [ ] [Other in-progress PRs](https://github.com/canonical/pebble/pulls)
   - [ ] [Other requested features](https://github.com/canonical/pebble/issues)
 

internals/overlord/logstate/clienterr/clienterr.go

@@ -0,0 +1,83 @@
+// 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 clienterr contains common error types which are recognised by the log
+gatherer. logstate.logClient implementations should return these error types
+to communicate with the gatherer.
+
+Errors in this package should be pattern-matched using errors.As:
+
+		err := client.Flush(ctx)
+		var backoff *clienterr.Backoff
+		if errors.As(err, &backoff) {
+	      ...
+		}
+*/
+package clienterr
+
+import (
+	"bytes"
+	"fmt"
+	"io"
+	"net/http"
+	"time"
+)
+
+// Backoff should be returned if the server indicates we are sending too many
+// requests (e.g. an HTTP 429 response).
+type Backoff struct {
+	RetryAfter *time.Time
+}
+
+func (e *Backoff) Error() string {
+	errStr := "too many requests"
+	if e.RetryAfter != nil {
+		errStr += ", retry after " + e.RetryAfter.String()
+	}
+	return errStr
+}
+
+// ErrorResponse represents an HTTP error response from the server
+// (4xx or 5xx).
+// This is a generic, catch-all error type - clients should use a more refined
+// type when appropriate (e.g. Backoff for a 429 response).
+type ErrorResponse struct {
+	StatusCode int
+	Body       bytes.Buffer
+	ReadErr    error
+}
+
+func (e *ErrorResponse) Error() string {
+	errStr := fmt.Sprintf("server returned HTTP %d\n", e.StatusCode)
+	if e.Body.Len() > 0 {
+		errStr += fmt.Sprintf(`response body:
+%s
+`, e.Body.String())
+	}
+	if e.ReadErr != nil {
+		errStr += "cannot read response body: " + e.ReadErr.Error()
+	}
+	return errStr
+}
+
+// ErrorFromResponse generates a *ErrorResponse from a failed *http.Response.
+// NB: this function reads the response body.
+func ErrorFromResponse(resp *http.Response) *ErrorResponse {
+	err := &ErrorResponse{}
+	err.StatusCode = resp.StatusCode
+	_, readErr := io.CopyN(&err.Body, resp.Body, 1024)
+	err.ReadErr = readErr
+	return err
+}

internals/overlord/logstate/gatherer.go

@@ -16,12 +16,15 @@ package logstate
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"time"
 
 	"gopkg.in/tomb.v2"
 
 	"github.com/canonical/pebble/internals/logger"
+	"github.com/canonical/pebble/internals/overlord/logstate/clienterr"
+	"github.com/canonical/pebble/internals/overlord/logstate/loki"
 	"github.com/canonical/pebble/internals/plan"
 	"github.com/canonical/pebble/internals/servicelog"
 )
@@ -71,6 +74,8 @@ type logGatherer struct {
 	pullers *pullerGroup
 	// All pullers send logs on this channel, received by main loop
 	entryCh chan servicelog.Entry
+
+	timer timer
 }
 
 // logGathererArgs allows overriding the newLogClient method and time values
@@ -103,6 +108,7 @@ func newLogGathererInternal(target *plan.LogTarget, args logGathererArgs) (*logG
 		client:     client,
 		entryCh:    make(chan servicelog.Entry),
 		pullers:    newPullerGroup(target.Name),
+		timer:      newTimer(),
 	}
 	g.clientCtx, g.clientCancel = context.WithCancel(context.Background())
 	g.tomb.Go(g.loop)
@@ -169,43 +175,48 @@ func (g *logGatherer) ServiceStarted(service *plan.Service, buffer *servicelog.R
 // pullers on entryCh, and writes them to the client. It also flushes the
 // client periodically, and exits when the gatherer's tomb is killed.
 func (g *logGatherer) loop() error {
-	timer := newTimer()
-	defer timer.Stop()
+	defer g.timer.Stop()
 
 mainLoop:
 	for {
 		select {
 		case <-g.tomb.Dying():
 			break mainLoop
 
-		case <-timer.Expired():
+		case <-g.timer.Expired():
 			// Mark timer as unset
-			timer.Stop()
-			err := g.client.Flush(g.clientCtx)
-			if err != nil {
-				logger.Noticef("Cannot flush logs to target %q: %v", g.targetName, err)
-			}
+			g.timer.Stop()
+			g.handleClientErr(g.client.Flush(g.clientCtx))
 
 		case entry := <-g.entryCh:
-			err := g.client.Write(g.clientCtx, entry)
-			if err != nil {
-				logger.Noticef("Cannot write logs to target %q: %v", g.targetName, err)
-			}
-			timer.EnsureSet(g.bufferTimeout)
+			g.handleClientErr(g.client.Write(g.clientCtx, entry))
+			g.timer.EnsureSet(g.bufferTimeout)
 		}
 	}
 
 	// Final flush to send any remaining logs buffered in the client
 	// We need to create a new context, as the previous one may have been cancelled.
 	ctx, cancel := context.WithTimeout(context.Background(), g.timeoutFinalFlush)
 	defer cancel()
-	err := g.client.Flush(ctx)
-	if err != nil {
-		logger.Noticef("Cannot flush logs to target %q: %v", g.targetName, err)
-	}
+	g.handleClientErr(g.client.Flush(ctx))
 	return nil
 }
 
+func (g *logGatherer) handleClientErr(err error) {
+	if err == nil {
+		return
+	}
+
+	var backoff *clienterr.Backoff
+	if errors.As(err, &backoff) {
+		logger.Noticef("Target %q: %v", g.targetName, err)
+		g.timer.retryAfter = backoff.RetryAfter
+		return
+	}
+
+	logger.Noticef("Cannot flush logs to target %q: %v", g.targetName, err)
+}
+
 // Stop tears down the gatherer and associated resources (pullers, client).
 // This method will block until gatherer teardown is complete.
 //
@@ -250,6 +261,8 @@ func (g *logGatherer) Stop() {
 type timer struct {
 	timer *time.Timer
 	set   bool
+	// If non-nil, the timer won't expire until after this time
+	retryAfter *time.Time
 }
 
 func newTimer() timer {
@@ -279,6 +292,15 @@ func (t *timer) EnsureSet(timeout time.Duration) {
 		return
 	}
 
+	if t.retryAfter != nil {
+		// We've been told to wait before retrying
+		retryTime := time.Until(*t.retryAfter)
+		if retryTime > timeout {
+			timeout = retryTime
+		}
+		t.retryAfter = nil
+	}
+
 	t.timer.Reset(timeout)
 	t.set = true
 }
@@ -311,7 +333,8 @@ type logClient interface {
 
 func newLogClient(target *plan.LogTarget) (logClient, error) {
 	switch target.Type {
-	//case plan.LokiTarget: TODO
+	case plan.LokiTarget:
+		return loki.NewClient(target), nil
 	//case plan.SyslogTarget: TODO
 	default:
 		return nil, fmt.Errorf("unknown type %q for log target %q", target.Type, target.Name)

internals/overlord/logstate/loki/export_test.go

@@ -0,0 +1,25 @@
+// 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 loki
+
+import "time"
+
+func SetRequestTimeout(new time.Duration) (restore func()) {
+	oldRequestTimeout := requestTimeout
+	requestTimeout = new
+	return func() {
+		requestTimeout = oldRequestTimeout
+	}
+}