feat(logfwd): add log labels

README.md

@@ -405,7 +405,7 @@ $ pebble run --verbose
 ...
 ```
 
-#### Log forwarding
+### Log forwarding
 
 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
@@ -422,6 +422,8 @@ log-targets:
         services: [svc1, svc2]
 ```
 
+#### Specifying services
+
 For each log target, use the `services` key to specify a list of services to collect logs from. In the above example, the `production-logs` 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, `staging-logs` will collect logs from all services.
@@ -453,6 +455,38 @@ my-target:
 ```
 would remove all services and then add `svc1`, so `my-target` would receive logs from only `svc1`.
 
+#### Labels
+
+In the `labels` section, you can specify custom labels to be added to any outgoing logs. These labels may contain `$ENVIRONMENT_VARIABLES` - these will be interpreted in the environment of the corresponding service. Pebble may also add its own default labels (depending on the protocol). For example, given the following plan:
+```yaml
+services:
+  svc1:
+    environment:
+      OWNER: 'alice'
+  svc2:
+    environment:
+      OWNER: 'bob'
+
+log-targets:
+  tgt1:
+    type: loki
+    labels:
+      product: 'juju'
+      owner: 'user-$OWNER'
+```
+the logs from `svc1` will be sent with the following labels:
+```yaml
+product: juju
+owner: user-alice     # env var $OWNER substituted
+pebble_service: svc1  # default label for Loki
+```
+and for svc2, the labels will be
+```yaml
+product: juju
+owner: user-bob       # env var $OWNER substituted
+pebble_service: svc2  # default label for Loki
+```
+
 
 ## Container usage
 
@@ -733,8 +767,10 @@ log-targets:
     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.
+    # which logs will be sent. The supported types are:
+    #
+    # - loki: Use the Grafana Loki protocol. A "pebble_service" label is
+    #   added automatically, with the name of the Pebble service as its value.
     type: loki
 
     # (Required) The URL of the remote log target.
@@ -749,6 +785,13 @@ log-targets:
     # service name with a minus (e.g. '-svc1') to remove a previously added
     # service. '-all' will remove all services.
     services: [<service names>]
+
+    # (Optional) A list of key/value pairs defining labels which should be set
+    # on the outgoing logs.
+    # The label values may contain $ENVIRONMENT_VARIABLES, which will be
+    # substituted using the environment for the corresponding service.
+    labels:
+      <label name>: <label value>
 ```
 
 ## API and clients

internals/overlord/logstate/gatherer.go

@@ -17,6 +17,7 @@ package logstate
 import (
 	"context"
 	"fmt"
+	"os"
 	"time"
 
 	"gopkg.in/tomb.v2"
@@ -59,7 +60,7 @@ const (
 type logGatherer struct {
 	*logGathererOptions
 
-	targetName string
+	target *plan.LogTarget
 	// tomb for the main loop
 	tomb tomb.Tomb
 
@@ -70,6 +71,9 @@ type logGatherer struct {
 	// ensure the client is not blocking subsequent teardown steps.
 	clientCancel context.CancelFunc
 
+	// Channel used to notify the main loop to set the client's labels
+	setLabels chan svcWithLabels
+
 	pullers *pullerGroup
 	// All pullers send logs on this channel, received by main loop
 	entryCh chan servicelog.Entry
@@ -78,9 +82,10 @@ type logGatherer struct {
 // logGathererOptions allows overriding the newLogClient method and time values
 // in testing.
 type logGathererOptions struct {
-	bufferTimeout      time.Duration
-	maxBufferedEntries int
-	timeoutFinalFlush  time.Duration
+	bufferTimeout       time.Duration
+	maxBufferedEntries  int
+	timeoutCurrentFlush time.Duration
+	timeoutFinalFlush   time.Duration
 	// method to get a new client
 	newClient func(*plan.LogTarget) (logClient, error)
 }
@@ -102,10 +107,11 @@ func newLogGathererInternal(target *plan.LogTarget, options *logGathererOptions)
 	g := &logGatherer{
 		logGathererOptions: options,
 
-		targetName: target.Name,
-		client:     client,
-		entryCh:    make(chan servicelog.Entry),
-		pullers:    newPullerGroup(target.Name),
+		target:    target,
+		client:    client,
+		setLabels: make(chan svcWithLabels),
+		entryCh:   make(chan servicelog.Entry),
+		pullers:   newPullerGroup(target.Name),
 	}
 	g.clientCtx, g.clientCancel = context.WithCancel(context.Background())
 	g.tomb.Go(g.loop)
@@ -121,6 +127,9 @@ func fillDefaultOptions(options *logGathererOptions) *logGathererOptions {
 	if options.maxBufferedEntries == 0 {
 		options.maxBufferedEntries = maxBufferedEntries
 	}
+	if options.timeoutCurrentFlush == 0 {
+		options.timeoutCurrentFlush = timeoutCurrentFlush
+	}
 	if options.timeoutFinalFlush == 0 {
 		options.timeoutFinalFlush = timeoutFinalFlush
 	}
@@ -133,44 +142,84 @@ func fillDefaultOptions(options *logGathererOptions) *logGathererOptions {
 // PlanChanged is called by the LogManager when the plan is changed, if this
 // gatherer's target exists in the new plan.
 func (g *logGatherer) PlanChanged(pl *plan.Plan, buffers map[string]*servicelog.RingBuffer) {
+	// Update target config
+	g.target = pl.LogTargets[g.target.Name]
+
 	// Remove old pullers
 	for _, svcName := range g.pullers.Services() {
 		svc, svcExists := pl.Services[svcName]
-		if !svcExists {
-			g.pullers.Remove(svcName)
-			continue
+		if svcExists {
+			if svc.LogsTo(g.target) {
+				// We're still collecting logs from this service, so don't remove it.
+				continue
+			}
 		}
 
-		tgt := pl.LogTargets[g.targetName]
-		if !svc.LogsTo(tgt) {
-			g.pullers.Remove(svcName)
+		// Service no longer forwarding to this log target (or it was removed from
+		// the plan). Remove it from the gatherer.
+		g.pullers.Remove(svcName)
+		select {
+		case g.setLabels <- svcWithLabels{svcName, nil}:
+		case <-g.tomb.Dying():
+			return
 		}
 	}
 
 	// Add new pullers
 	for _, service := range pl.Services {
-		target := pl.LogTargets[g.targetName]
-		if !service.LogsTo(target) {
+		if !service.LogsTo(g.target) {
 			continue
 		}
 
-		buffer, bufferExists := buffers[service.Name]
-		if !bufferExists {
+		buffer, svcStarted := buffers[service.Name]
+		if !svcStarted {
 			// We don't yet have a reference to the service's ring buffer
 			// Need to wait until ServiceStarted
 			continue
 		}
 
+		labels := evaluateLabels(g.target.Labels, service.Environment)
+		select {
+		case g.setLabels <- svcWithLabels{service.Name, labels}:
+		case <-g.tomb.Dying():
+			return
+		}
+
 		g.pullers.Add(service.Name, buffer, g.entryCh)
 	}
 }
 
 // ServiceStarted is called by the LogManager on the start of a service which
 // logs to this gatherer's target.
 func (g *logGatherer) ServiceStarted(service *plan.Service, buffer *servicelog.RingBuffer) {
+	labels := evaluateLabels(g.target.Labels, service.Environment)
+	select {
+	case g.setLabels <- svcWithLabels{service.Name, labels}:
+	case <-g.tomb.Dying():
+		return
+	}
+
 	g.pullers.Add(service.Name, buffer, g.entryCh)
 }
 
+// evaluateLabels interprets the labels defined in the plan, substituting any
+// $env_vars with the corresponding value in the service's environment.
+func evaluateLabels(rawLabels, env map[string]string) map[string]string {
+	substitute := func(k string) string {
+		if v, ok := env[k]; ok {
+			return v
+		}
+		// Undefined variables default to "", just like Bash
+		return ""
+	}
+
+	labels := make(map[string]string, len(rawLabels))
+	for key, rawLabel := range rawLabels {
+		labels[key] = os.Expand(rawLabel, substitute)
+	}
+	return labels
+}
+
 // The main control loop for the logGatherer. loop receives logs from the
 // pullers on entryCh, and writes them to the client. It also flushes the
 // client periodically, and exits when the gatherer's tomb is killed.
@@ -185,7 +234,7 @@ func (g *logGatherer) loop() error {
 		flushTimer.Stop()
 		err := g.client.Flush(ctx)
 		if err != nil {
-			logger.Noticef("Cannot flush logs to target %q: %v", g.targetName, err)
+			logger.Noticef("Cannot flush logs to target %q: %v", g.target.Name, err)
 		}
 		numWritten = 0
 	}
@@ -199,10 +248,16 @@ mainLoop:
 		case <-flushTimer.Expired():
 			flushClient(g.clientCtx)
 
+		case args := <-g.setLabels:
+			// Before we change the labels, flush any logs currently in the buffer,
+			// so that these logs are sent with the correct (old) labels.
+			flushClient(g.clientCtx)
+			g.client.SetLabels(args.service, args.labels)
+
 		case entry := <-g.entryCh:
 			err := g.client.Add(entry)
 			if err != nil {
-				logger.Noticef("Cannot write logs to target %q: %v", g.targetName, err)
+				logger.Noticef("Cannot write logs to target %q: %v", g.target.Name, err)
 				continue
 			}
 			numWritten++
@@ -236,12 +291,12 @@ mainLoop:
 //   - Flush out any final logs buffered in the client.
 func (g *logGatherer) Stop() {
 	// Wait up to timeoutCurrentFlush for the current flush to complete (if any)
-	time.AfterFunc(timeoutCurrentFlush, g.clientCancel)
+	time.AfterFunc(g.timeoutCurrentFlush, g.clientCancel)
 
 	// Wait up to timeoutPullers for the pullers to pull the final logs from the
 	// iterator and send to the main loop.
 	time.AfterFunc(timeoutPullers, func() {
-		logger.Debugf("gatherer %q: force killing log pullers", g.targetName)
+		logger.Debugf("gatherer %q: force killing log pullers", g.target.Name)
 		g.pullers.KillAll()
 	})
 
@@ -251,9 +306,9 @@ func (g *logGatherer) Stop() {
 	g.pullers.tomb.Kill(nil)
 	select {
 	case <-g.pullers.Done():
-		logger.Debugf("gatherer %q: pullers have finished", g.targetName)
+		logger.Debugf("gatherer %q: pullers have finished", g.target.Name)
 	case <-time.After(timeoutMainLoop):
-		logger.Debugf("gatherer %q: force killing main loop", g.targetName)
+		logger.Debugf("gatherer %q: force killing main loop", g.target.Name)
 	}
 
 	g.tomb.Kill(nil)
@@ -264,6 +319,11 @@ func (g *logGatherer) Stop() {
 	}
 }
 
+type svcWithLabels struct {
+	service string
+	labels  map[string]string
+}
+
 // timer wraps time.Timer and provides a better API.
 type timer struct {
 	timer *time.Timer
@@ -312,6 +372,10 @@ type logClient interface {
 
 	// Flush sends buffered logs (if any) to the remote target.
 	Flush(context.Context) error
+
+	// SetLabels sets the log labels for the given services, or releases
+	// previously allocated label resources if the labels parameter is nil.
+	SetLabels(serviceName string, labels map[string]string)
 }
 
 func newLogClient(target *plan.LogTarget) (logClient, error) {