diff --git a/README.md b/README.md
index 44ab31cf5..b182431f8 100644
--- a/README.md
+++ b/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,12 @@ 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 $ENV_VARS, which will
+    # be substituted using the environment for the corresponding service.
+    labels:
+      <label name>: <label value>
 ```
 
 ## API and clients
diff --git a/client/files.go b/client/files.go
index cce33414d..14e5623d8 100644
--- a/client/files.go
+++ b/client/files.go
@@ -16,11 +16,16 @@ package client
 
 import (
 	"bytes"
+	"context"
 	"encoding/json"
 	"fmt"
+	"io"
+	"mime/multipart"
+	"net/textproto"
 	"net/url"
 	"os"
 	"strconv"
+	"strings"
 	"time"
 )
 
@@ -364,3 +369,146 @@ func (client *Client) RemovePath(opts *RemovePathOptions) error {
 
 	return nil
 }
+
+type PushOptions struct {
+	// Source is the source of data to write (required).
+	Source io.Reader
+
+	// Path indicates the absolute path of the file in the destination
+	// machine (required).
+	Path string
+
+	// MakeDirs, if true, will create any non-existing directories in the path
+	// to the remote file. If false (the default) the call to Push will
+	// fail if any of the parent directories of path do not exist.
+	MakeDirs bool
+
+	// Permissions indicates the mode of the file on the destination machine.
+	// If 0 or unset, defaults to 0644. Note that, when used together with MakeDirs,
+	// the directories that are created will not use this mode, but 0755.
+	Permissions os.FileMode
+
+	// UserID indicates the user ID of the owner for the file on the destination
+	// machine. When used together with MakeDirs, the directories that are
+	// created will also be owned by this user.
+	UserID *int
+
+	// User indicates the name of the owner user for the file on the destination
+	// machine. When used together with MakeDirs, the directories that are
+	// created will also be owned by this user.
+	User string
+
+	// GroupID indicates the ID of the owner group for the file on the destination
+	// machine. When used together with MakeDirs, the directories that are
+	// created will also be owned by this user.
+	GroupID *int
+
+	// Group indicates the name of the owner group for the file on the
+	// machine. When used together with MakeDirs, the directories that are
+	// created will also be owned by this user.
+	Group string
+}
+
+type writeFilesPayload struct {
+	Action string           `json:"action"`
+	Files  []writeFilesItem `json:"files"`
+}
+
+type writeFilesItem struct {
+	Path        string `json:"path"`
+	MakeDirs    bool   `json:"make-dirs"`
+	Permissions string `json:"permissions"`
+	UserID      *int   `json:"user-id"`
+	User        string `json:"user"`
+	GroupID     *int   `json:"group-id"`
+	Group       string `json:"group"`
+}
+
+var quoteEscaper = strings.NewReplacer("\\", "\\\\", `"`, "\\\"")
+
+func escapeQuotes(s string) string {
+	return quoteEscaper.Replace(s)
+}
+
+// Push writes content to a path on the remote system.
+func (client *Client) Push(opts *PushOptions) error {
+	var permissions string
+	if opts.Permissions != 0 {
+		permissions = fmt.Sprintf("%03o", opts.Permissions)
+	}
+
+	payload := writeFilesPayload{
+		Action: "write",
+		Files: []writeFilesItem{{
+			Path:        opts.Path,
+			MakeDirs:    opts.MakeDirs,
+			Permissions: permissions,
+			UserID:      opts.UserID,
+			User:        opts.User,
+			GroupID:     opts.GroupID,
+			Group:       opts.Group,
+		}},
+	}
+
+	var body bytes.Buffer
+	mw := multipart.NewWriter(&body)
+
+	// Encode metadata part of the header
+	part, err := mw.CreatePart(textproto.MIMEHeader{
+		"Content-Type":        {"application/json"},
+		"Content-Disposition": {`form-data; name="request"`},
+	})
+	if err != nil {
+		return fmt.Errorf("cannot encode metadata in request payload: %w", err)
+	}
+
+	// Buffer for multipart header/footer
+	if err := json.NewEncoder(part).Encode(&payload); err != nil {
+		return err
+	}
+
+	// Encode file part of the header
+	escapedPath := escapeQuotes(opts.Path)
+	_, err = mw.CreatePart(textproto.MIMEHeader{
+		"Content-Type":        {"application/octet-stream"},
+		"Content-Disposition": {fmt.Sprintf(`form-data; name="files"; filename="%s"`, escapedPath)},
+	})
+	if err != nil {
+		return fmt.Errorf("cannot encode file in request payload: %w", err)
+	}
+
+	header := body.String()
+
+	// Encode multipart footer
+	body.Reset()
+	mw.Close()
+	footer := body.String()
+
+	resp, err := client.Requester().Do(context.Background(), &RequestOptions{
+		Type:    SyncRequest,
+		Method:  "POST",
+		Path:    "/v1/files",
+		Headers: map[string]string{"Content-Type": mw.FormDataContentType()},
+		Body:    io.MultiReader(strings.NewReader(header), opts.Source, strings.NewReader(footer)),
+	})
+	if err != nil {
+		return err
+	}
+
+	var result []fileResult
+	if err = resp.DecodeResult(&result); err != nil {
+		return err
+	}
+	if len(result) != 1 {
+		return fmt.Errorf("expected exactly one result from API, got %d", len(result))
+	}
+	if result[0].Error != nil {
+		return &Error{
+			Kind:    result[0].Error.Kind,
+			Value:   result[0].Error.Value,
+			Message: result[0].Error.Message,
+		}
+	}
+
+	return nil
+}
diff --git a/client/files_test.go b/client/files_test.go
index bc808ade9..4f3da5a06 100644
--- a/client/files_test.go
+++ b/client/files_test.go
@@ -15,8 +15,12 @@
 package client_test
 
 import (
+	"bytes"
 	"encoding/json"
+	"io"
 	"os"
+	"path"
+	"strings"
 	"time"
 
 	. "gopkg.in/check.v1"
@@ -574,3 +578,130 @@ func (cs *clientSuite) TestRemovePathFailsWithMultipleAPIResults(c *C) {
 		}},
 	})
 }
+
+type writeFilesPayload struct {
+	Action string           `json:"action"`
+	Files  []writeFilesItem `json:"files"`
+}
+
+type writeFilesItem struct {
+	Path        string `json:"path"`
+	MakeDirs    bool   `json:"make-dirs"`
+	Permissions string `json:"permissions"`
+	UserID      *int   `json:"user-id"`
+	User        string `json:"user"`
+	GroupID     *int   `json:"group-id"`
+	Group       string `json:"group"`
+}
+
+func (cs *clientSuite) TestPush(c *C) {
+	cs.rsp = `{"type": "sync", "result": [{"path": "/file.dat"}]}`
+
+	err := cs.cli.Push(&client.PushOptions{
+		Path:   "/file.dat",
+		Source: strings.NewReader("Hello, world!"),
+	})
+	c.Assert(err, IsNil)
+	mr, err := cs.req.MultipartReader()
+	c.Assert(err, IsNil)
+
+	c.Assert(cs.req.URL.Path, Equals, "/v1/files")
+	c.Assert(cs.req.Method, Equals, "POST")
+
+	// Check metadata part
+	metadata, err := mr.NextPart()
+	c.Assert(err, IsNil)
+	c.Assert(metadata.Header.Get("Content-Type"), Equals, "application/json")
+	c.Assert(metadata.FormName(), Equals, "request")
+
+	buf := bytes.NewBuffer(make([]byte, 0))
+	_, err = buf.ReadFrom(metadata)
+	c.Assert(err, IsNil)
+
+	// Decode metadata
+	var payload writeFilesPayload
+	err = json.NewDecoder(buf).Decode(&payload)
+	c.Assert(err, IsNil)
+	c.Assert(payload, DeepEquals, writeFilesPayload{
+		Action: "write",
+		Files: []writeFilesItem{{
+			Path: "/file.dat",
+		}},
+	})
+
+	// Check file part
+	file, err := mr.NextPart()
+	c.Assert(err, IsNil)
+	c.Assert(file.Header.Get("Content-Type"), Equals, "application/octet-stream")
+	c.Assert(file.FormName(), Equals, "files")
+	c.Assert(path.Base(file.FileName()), Equals, "file.dat")
+
+	buf.Reset()
+	_, err = buf.ReadFrom(file)
+	c.Assert(err, IsNil)
+	c.Assert(buf.String(), Equals, "Hello, world!")
+
+	// Check end of multipart request
+	_, err = mr.NextPart()
+	c.Assert(err, Equals, io.EOF)
+}
+
+func (cs *clientSuite) TestPushFails(c *C) {
+	cs.rsp = `{"type": "error", "result": {"message": "could not foo"}}`
+
+	err := cs.cli.Push(&client.PushOptions{
+		Path:   "/file.dat",
+		Source: strings.NewReader("Hello, world!"),
+	})
+	c.Assert(err, ErrorMatches, "could not foo")
+}
+
+func (cs *clientSuite) TestPushFailsOnFile(c *C) {
+	cs.rsp = `{
+		"type": "sync",
+		"result": [{
+			"path": "/file.dat",
+			"error": {
+				"message": "could not bar",
+				"kind": "permission-denied",
+				"value": 42
+			}
+		}]
+	}`
+
+	err := cs.cli.Push(&client.PushOptions{
+		Path:   "/file.dat",
+		Source: strings.NewReader("Hello, world!"),
+	})
+	clientErr, ok := err.(*client.Error)
+	c.Assert(ok, Equals, true)
+	c.Assert(clientErr.Message, Equals, "could not bar")
+	c.Assert(clientErr.Kind, Equals, "permission-denied")
+}
+
+func (cs *clientSuite) TestPushFailsWithMultipleAPIResults(c *C) {
+	cs.rsp = `{
+		"type": "sync",
+		"result": [{
+			"path": "/file.dat",
+			"error": {
+				"message": "could not bar",
+				"kind": "permission-denied",
+				"value": 42
+			}
+		}, {
+			"path": "/file.dat",
+			"error": {
+				"message": "could not baz",
+				"kind": "generic-file-error",
+				"value": 41
+			}
+		}]
+	}`
+
+	err := cs.cli.Push(&client.PushOptions{
+		Path:   "/file.dat",
+		Source: strings.NewReader("Hello, world!"),
+	})
+	c.Assert(err, ErrorMatches, "expected exactly one result from API, got 2")
+}
diff --git a/internals/cli/cmd_help.go b/internals/cli/cmd_help.go
index a322f83aa..21e96a342 100644
--- a/internals/cli/cmd_help.go
+++ b/internals/cli/cmd_help.go
@@ -192,7 +192,7 @@ var HelpCategories = []HelpCategory{{
 }, {
 	Label:       "Files",
 	Description: "work with files and execute commands",
-	Commands:    []string{"ls", "mkdir", "rm", "exec"},
+	Commands:    []string{"push", "ls", "mkdir", "rm", "exec"},
 }, {
 	Label:       "Changes",
 	Description: "manage changes and their tasks",
diff --git a/internals/cli/cmd_mkdir.go b/internals/cli/cmd_mkdir.go
index 3b952727f..0c2770413 100644
--- a/internals/cli/cmd_mkdir.go
+++ b/internals/cli/cmd_mkdir.go
@@ -32,13 +32,13 @@ The mkdir command creates the specified directory.
 type cmdMkdir struct {
 	client *client.Client
 
-	MakeParents bool   `short:"p"`
-	Permissions string `short:"m"`
-	UserID      *int   `long:"uid"`
-	User        string `long:"user"`
-	GroupID     *int   `long:"gid"`
-	Group       string `long:"group"`
-	Positional  struct {
+	Parents    bool   `short:"p"`
+	Mode       string `short:"m"`
+	UserID     *int   `long:"uid"`
+	User       string `long:"user"`
+	GroupID    *int   `long:"gid"`
+	Group      string `long:"group"`
+	Positional struct {
 		Path string `positional-arg-name:"<path>"`
 	} `positional-args:"yes" required:"yes"`
 }
@@ -50,7 +50,7 @@ func init() {
 		Description: cmdMkdirDescription,
 		ArgsHelp: map[string]string{
 			"-p":      "Create parent directories as needed",
-			"-m":      "Set permissions (e.g. 0644)",
+			"-m":      "Override mode bits (3-digit octal)",
 			"--uid":   "Use specified user ID",
 			"--user":  "Use specified username",
 			"--gid":   "Use specified group ID",
@@ -69,17 +69,17 @@ func (cmd *cmdMkdir) Execute(args []string) error {
 
 	opts := client.MakeDirOptions{
 		Path:        cmd.Positional.Path,
-		MakeParents: cmd.MakeParents,
+		MakeParents: cmd.Parents,
 		UserID:      cmd.UserID,
 		User:        cmd.User,
 		GroupID:     cmd.GroupID,
 		Group:       cmd.Group,
 	}
 
-	if cmd.Permissions != "" {
-		p, err := strconv.ParseUint(cmd.Permissions, 8, 32)
+	if cmd.Mode != "" {
+		p, err := strconv.ParseUint(cmd.Mode, 8, 32)
 		if err != nil {
-			return fmt.Errorf("invalid mode for directory: %q", cmd.Permissions)
+			return fmt.Errorf("invalid mode for directory: %q", cmd.Mode)
 		}
 		opts.Permissions = os.FileMode(p)
 	}
diff --git a/internals/cli/cmd_push.go b/internals/cli/cmd_push.go
new file mode 100644
index 000000000..fc222b40c
--- /dev/null
+++ b/internals/cli/cmd_push.go
@@ -0,0 +1,102 @@
+// Copyright (c) 2022 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 cli
+
+import (
+	"fmt"
+	"os"
+	"strconv"
+
+	"github.com/canonical/go-flags"
+
+	"github.com/canonical/pebble/client"
+)
+
+const cmdPushSummary = "Transfer a file to the remote system"
+const cmdPushDescription = `
+The push command transfers a file to the remote system.
+`
+
+type cmdPush struct {
+	client *client.Client
+
+	Parents bool   `short:"p"`
+	Mode    string `short:"m"`
+	UserID  *int   `long:"uid"`
+	User    string `long:"user"`
+	GroupID *int   `long:"gid"`
+	Group   string `long:"group"`
+
+	Positional struct {
+		LocalPath  string `positional-arg-name:"<local-path>" required:"1"`
+		RemotePath string `positional-arg-name:"<remote-path>" required:"1"`
+	} `positional-args:"yes"`
+}
+
+func init() {
+	AddCommand(&CmdInfo{
+		Name:        "push",
+		Summary:     cmdPushSummary,
+		Description: cmdPushDescription,
+		ArgsHelp: map[string]string{
+			"-p":      "Create parent directories for the file",
+			"-m":      "Override mode bits (3-digit octal)",
+			"--uid":   "Use specified user ID",
+			"--user":  "Use specified username",
+			"--gid":   "Use specified group ID",
+			"--group": "Use specified group name",
+		},
+		New: func(opts *CmdOptions) flags.Commander {
+			return &cmdPush{client: opts.Client}
+		},
+	})
+}
+func (cmd *cmdPush) Execute(args []string) error {
+	if len(args) > 0 {
+		return ErrExtraArgs
+	}
+
+	f, err := os.Open(cmd.Positional.LocalPath)
+	if err != nil {
+		return err
+	}
+	defer f.Close()
+
+	var permissions os.FileMode
+	if cmd.Mode != "" {
+		p, err := strconv.ParseUint(cmd.Mode, 8, 32)
+		if err != nil {
+			return fmt.Errorf("invalid mode for directory: %q", cmd.Mode)
+		}
+		permissions = os.FileMode(p)
+	} else {
+		st, err := f.Stat()
+		if err != nil {
+			return err
+		}
+		permissions = st.Mode().Perm()
+	}
+
+	return cmd.client.Push(&client.PushOptions{
+		Source:      f,
+		Path:        cmd.Positional.RemotePath,
+		MakeDirs:    cmd.Parents,
+		Permissions: permissions,
+		UserID:      cmd.UserID,
+		User:        cmd.User,
+		GroupID:     cmd.GroupID,
+		Group:       cmd.Group,
+	})
+}
diff --git a/internals/cli/cmd_push_test.go b/internals/cli/cmd_push_test.go
new file mode 100644
index 000000000..5ef9e16d5
--- /dev/null
+++ b/internals/cli/cmd_push_test.go
@@ -0,0 +1,209 @@
+// Copyright (c) 2022 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 cli_test
+
+import (
+	"bytes"
+	"encoding/json"
+	"fmt"
+	"io"
+	"io/ioutil"
+	"net/http"
+	"os"
+	"path"
+	"path/filepath"
+	"syscall"
+
+	. "gopkg.in/check.v1"
+
+	"github.com/canonical/pebble/client"
+	"github.com/canonical/pebble/internals/cli"
+)
+
+type writeFilesPayload struct {
+	Action string           `json:"action"`
+	Files  []writeFilesItem `json:"files"`
+}
+
+type writeFilesItem struct {
+	Path        string `json:"path"`
+	MakeDirs    bool   `json:"make-dirs"`
+	Permissions string `json:"permissions"`
+	UserID      *int   `json:"user-id"`
+	User        string `json:"user"`
+	GroupID     *int   `json:"group-id"`
+	Group       string `json:"group"`
+}
+
+func (s *PebbleSuite) TestPush(c *C) {
+	s.RedirectClientToTestServer(func(w http.ResponseWriter, r *http.Request) {
+		c.Assert(r.URL.Path, Equals, "/v1/files")
+		c.Assert(r.Method, Equals, "POST")
+
+		mr, err := r.MultipartReader()
+		c.Assert(err, IsNil)
+
+		// Check metadata part
+		metadata, err := mr.NextPart()
+		c.Assert(err, IsNil)
+		c.Assert(metadata.Header.Get("Content-Type"), Equals, "application/json")
+		c.Assert(metadata.FormName(), Equals, "request")
+
+		buf := bytes.NewBuffer(make([]byte, 0))
+		_, err = buf.ReadFrom(metadata)
+		c.Assert(err, IsNil)
+
+		// Decode metadata
+		var payload writeFilesPayload
+		err = json.NewDecoder(buf).Decode(&payload)
+		c.Assert(err, IsNil)
+		c.Assert(payload, DeepEquals, writeFilesPayload{
+			Action: "write",
+			Files: []writeFilesItem{{
+				Path:        "/tmp/file.bin",
+				Permissions: "755",
+			}},
+		})
+
+		// Check file part
+		file, err := mr.NextPart()
+		c.Assert(err, IsNil)
+		c.Assert(file.Header.Get("Content-Type"), Equals, "application/octet-stream")
+		c.Assert(file.FormName(), Equals, "files")
+		c.Assert(path.Base(file.FileName()), Equals, "file.bin")
+
+		buf.Reset()
+		_, err = buf.ReadFrom(file)
+		c.Assert(err, IsNil)
+		c.Assert(buf.String(), Equals, "Hello, world!")
+
+		// Check end of multipart request
+		_, err = mr.NextPart()
+		c.Assert(err, Equals, io.EOF)
+
+		fmt.Fprintln(w, `{"type": "sync", "result": [{"path": "/tmp/file.bin"}]}`)
+	})
+
+	// Create temporary file
+	tempDir := c.MkDir()
+	filePath := filepath.Join(tempDir, "file.dat")
+	err := ioutil.WriteFile(filePath, []byte("Hello, world!"), 0755)
+	c.Assert(err, IsNil)
+
+	args := []string{"push", filePath, "/tmp/file.bin"}
+	rest, err := cli.Parser(cli.Client()).ParseArgs(args)
+	c.Assert(err, IsNil)
+	c.Assert(rest, HasLen, 0)
+	c.Check(s.Stdout(), Equals, "")
+	c.Check(s.Stderr(), Equals, "")
+}
+
+func (s *PebbleSuite) TestPushExtraArgs(c *C) {
+	args := []string{"push", "file.dat", "/tmp/file.bin", "extra", "args"}
+	_, err := cli.Parser(cli.Client()).ParseArgs(args)
+	c.Assert(err, Equals, cli.ErrExtraArgs)
+	c.Check(s.Stdout(), Equals, "")
+	c.Check(s.Stderr(), Equals, "")
+}
+
+func (s *PebbleSuite) TestPushFailsToOpen(c *C) {
+	args := []string{"push", "/non/existing/path", "/tmp/file.bin"}
+	_, err := cli.Parser(cli.Client()).ParseArgs(args)
+	c.Assert(err, Not(IsNil))
+	e, ok := err.(*os.PathError)
+	c.Assert(ok, Equals, true)
+	c.Assert(e.Path, Equals, "/non/existing/path")
+	c.Assert(e.Err, Equals, syscall.ENOENT)
+	c.Check(s.Stdout(), Equals, "")
+	c.Check(s.Stderr(), Equals, "")
+}
+
+func (s *PebbleSuite) TestPushAPIFails(c *C) {
+	s.RedirectClientToTestServer(func(w http.ResponseWriter, r *http.Request) {
+		c.Assert(r.URL.Path, Equals, "/v1/files")
+		c.Assert(r.Method, Equals, "POST")
+
+		mr, err := r.MultipartReader()
+		c.Assert(err, IsNil)
+
+		// Check metadata part
+		metadata, err := mr.NextPart()
+		c.Assert(err, IsNil)
+		c.Assert(metadata.Header.Get("Content-Type"), Equals, "application/json")
+		c.Assert(metadata.FormName(), Equals, "request")
+
+		buf := bytes.NewBuffer(make([]byte, 0))
+		_, err = buf.ReadFrom(metadata)
+		c.Assert(err, IsNil)
+
+		// Decode metadata
+		var payload writeFilesPayload
+		err = json.NewDecoder(buf).Decode(&payload)
+		c.Assert(err, IsNil)
+		c.Assert(payload, DeepEquals, writeFilesPayload{
+			Action: "write",
+			Files: []writeFilesItem{{
+				Path:        "/tmp/file.bin",
+				Permissions: "600",
+			}},
+		})
+
+		// Check file part
+		file, err := mr.NextPart()
+		c.Assert(err, IsNil)
+		c.Assert(file.Header.Get("Content-Type"), Equals, "application/octet-stream")
+		c.Assert(file.FormName(), Equals, "files")
+		c.Assert(path.Base(file.FileName()), Equals, "file.bin")
+
+		buf.Reset()
+		_, err = buf.ReadFrom(file)
+		c.Assert(err, IsNil)
+		c.Assert(buf.String(), Equals, "Hello, world!")
+
+		// Check end of multipart request
+		_, err = mr.NextPart()
+		c.Assert(err, Equals, io.EOF)
+
+		fmt.Fprintln(w, `{
+			"type": "sync",
+			"result": [{
+				"path": "/tmp/file.bin",
+				"error": {
+					"message": "could not bar",
+					"kind": "permission-denied",
+					"value": 42
+				}
+			}]
+		}`)
+	})
+
+	// Create temporary file
+	tempDir := c.MkDir()
+	filePath := filepath.Join(tempDir, "file.dat")
+	err := ioutil.WriteFile(filePath, []byte("Hello, world!"), 0755)
+	c.Assert(err, IsNil)
+
+	args := []string{"push", "-m", "600", filePath, "/tmp/file.bin"}
+	rest, err := cli.Parser(cli.Client()).ParseArgs(args)
+
+	clientErr, ok := err.(*client.Error)
+	c.Assert(ok, Equals, true)
+	c.Assert(clientErr.Message, Equals, "could not bar")
+	c.Assert(clientErr.Kind, Equals, "permission-denied")
+
+	c.Assert(rest, HasLen, 1)
+	c.Check(s.Stdout(), Equals, "")
+	c.Check(s.Stderr(), Equals, "")
+}
diff --git a/internals/overlord/logstate/gatherer.go b/internals/overlord/logstate/gatherer.go
index cde953066..58ebfb501 100644
--- a/internals/overlord/logstate/gatherer.go
+++ b/internals/overlord/logstate/gatherer.go
@@ -17,6 +17,7 @@ package logstate
 import (
 	"context"
 	"fmt"
+	"os"
 	"time"
 
 	"gopkg.in/tomb.v2"
@@ -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)
 }
@@ -104,6 +109,7 @@ func newLogGathererInternal(target *plan.LogTarget, options *logGathererOptions)
 
 		targetName: target.Name,
 		client:     client,
+		setLabels:  make(chan svcWithLabels),
 		entryCh:    make(chan servicelog.Entry),
 		pullers:    newPullerGroup(target.Name),
 	}
@@ -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,35 +142,46 @@ 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) {
+	target := pl.LogTargets[g.targetName]
+
 	// Remove old pullers
 	for _, svcName := range g.pullers.Services() {
 		svc, svcExists := pl.Services[svcName]
-		if !svcExists {
-			g.pullers.Remove(svcName)
+		if svcExists && svc.LogsTo(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) {
 			continue
 		}
 
-		buffer, bufferExists := buffers[service.Name]
-		if !bufferExists {
-			// We don't yet have a reference to the service's ring buffer
-			// Need to wait until ServiceStarted
-			continue
+		labels := evaluateLabels(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)
+		// If the service was just added, it may not be started yet. In this case,
+		// we need to wait until the buffer is created, and then we can update the
+		// pullers inside ServiceStarted.
+		buffer, svcStarted := buffers[service.Name]
+		if svcStarted {
+			g.pullers.Add(service.Name, buffer, g.entryCh)
+		}
 	}
 }
 
@@ -171,6 +191,21 @@ func (g *logGatherer) ServiceStarted(service *plan.Service, buffer *servicelog.R
 	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 {
+		// Undefined variables default to "", just like Bash
+		return env[k]
+	}
+
+	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.
@@ -199,6 +234,12 @@ 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 {
@@ -236,7 +277,7 @@ 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.
@@ -264,6 +305,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 +358,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 service, 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) {
diff --git a/internals/overlord/logstate/gatherer_test.go b/internals/overlord/logstate/gatherer_test.go
index 956f22f33..37cee10f2 100644
--- a/internals/overlord/logstate/gatherer_test.go
+++ b/internals/overlord/logstate/gatherer_test.go
@@ -20,6 +20,7 @@ import (
 	"io"
 	"net/http"
 	"net/http/httptest"
+	"sync"
 	"time"
 
 	. "gopkg.in/check.v1"
@@ -148,11 +149,14 @@ func (s *gathererSuite) TestRetryLoki(c *C) {
 	}))
 	defer server.Close()
 
+	logTarget := &plan.LogTarget{
+		Name:     "tgt1",
+		Location: server.URL,
+		Services: []string{"all"},
+	}
+
 	g, err := newLogGathererInternal(
-		&plan.LogTarget{
-			Name:     "tgt1",
-			Location: server.URL,
-		},
+		logTarget,
 		&logGathererOptions{
 			bufferTimeout:      1 * time.Millisecond,
 			maxBufferedEntries: 5,
@@ -166,6 +170,14 @@ func (s *gathererSuite) TestRetryLoki(c *C) {
 	c.Assert(err, IsNil)
 
 	testSvc := newTestService("svc1")
+	g.PlanChanged(&plan.Plan{
+		Services: map[string]*plan.Service{
+			"svc1": testSvc.config,
+		},
+		LogTargets: map[string]*plan.LogTarget{
+			"tgt1": logTarget,
+		},
+	}, nil)
 	g.ServiceStarted(testSvc.config, testSvc.ringBuffer)
 
 	reqReceived := make(chan struct{})
@@ -218,6 +230,106 @@ func (s *gathererSuite) TestRetryLoki(c *C) {
 	}
 }
 
+// Test to catch race conditions in gatherer
+func (s *gathererSuite) TestConcurrency(c *C) {
+	target := &plan.LogTarget{
+		Name:     "tgt1",
+		Type:     plan.LokiTarget,
+		Services: []string{"all"},
+		Labels:   map[string]string{"foo": "bar-$SECRET-$SECRET2", "baz": "foo"},
+	}
+
+	g, err := newLogGathererInternal(target, &logGathererOptions{
+		maxBufferedEntries: 2,
+	})
+	c.Assert(err, IsNil)
+
+	svc1 := newTestService("svc1")
+	svc2 := newTestService("svc2")
+	fakeEnv := map[string]string{
+		"SECRET":  "pie",
+		"SECRET2": "pizza",
+	}
+	svc1.config.Environment = fakeEnv
+	svc2.config.Environment = fakeEnv
+
+	buffers := map[string]*servicelog.RingBuffer{
+		svc1.name: svc1.ringBuffer,
+		svc2.name: svc2.ringBuffer,
+	}
+
+	// Run a bunch of operations concurrently
+	doConcurrently := func(ops ...func()) {
+		wg := sync.WaitGroup{}
+		wg.Add(len(ops))
+		for _, f := range ops {
+			go func(f func()) {
+				defer wg.Done()
+				f()
+			}(f)
+		}
+		wg.Wait()
+	}
+
+	doConcurrently(
+		// Change plan
+		func() {
+			g.PlanChanged(&plan.Plan{
+				Services: map[string]*plan.Service{
+					svc1.name: svc1.config,
+				},
+				LogTargets: map[string]*plan.LogTarget{
+					target.Name: target,
+				},
+			}, buffers)
+		},
+		// Start new service
+		func() { g.ServiceStarted(svc1.config, svc1.ringBuffer) },
+		// Write some logs
+		func() {
+			svc1.writeLog("hello")
+			svc1.writeLog("goodbye")
+		},
+	)
+
+	doConcurrently(
+		// Write some more logs
+		func() {
+			svc1.writeLog("hello again")
+			svc1.writeLog("goodbye again")
+		},
+		// Simulate a service restart
+		func() { g.ServiceStarted(svc1.config, svc1.ringBuffer) },
+	)
+
+	doConcurrently(
+		// Change plan
+		func() {
+			g.PlanChanged(&plan.Plan{
+				Services: map[string]*plan.Service{
+					svc2.name: svc2.config,
+				},
+				LogTargets: map[string]*plan.LogTarget{
+					target.Name: target,
+				},
+			}, buffers)
+		},
+		// Start new service
+		func() { g.ServiceStarted(svc2.config, svc2.ringBuffer) },
+		// Write some logs
+		func() {
+			svc2.writeLog("hello")
+			go svc2.writeLog("goodbye")
+		},
+	)
+
+	err = svc1.stop()
+	c.Assert(err, IsNil)
+	err = svc2.stop()
+	c.Assert(err, IsNil)
+	g.Stop()
+}
+
 func checkLogs(c *C, received []servicelog.Entry, expected []string) {
 	c.Assert(received, HasLen, len(expected))
 	for i, entry := range received {
@@ -232,6 +344,10 @@ type testClient struct {
 	sendCh     chan []servicelog.Entry
 }
 
+func (c *testClient) SetLabels(serviceName string, labels map[string]string) {
+	// no-op
+}
+
 func (c *testClient) Add(entry servicelog.Entry) error {
 	c.buffered = append(c.buffered, entry)
 	return nil
diff --git a/internals/overlord/logstate/loki/loki.go b/internals/overlord/logstate/loki/loki.go
index d3f37bfd4..1a371deaf 100644
--- a/internals/overlord/logstate/loki/loki.go
+++ b/internals/overlord/logstate/loki/loki.go
@@ -38,16 +38,17 @@ const (
 )
 
 type Client struct {
-	options *ClientOptions
-
-	targetName string
-	remoteURL  string
+	options    *ClientOptions
+	target     *plan.LogTarget
 	httpClient *http.Client
 
 	// To store log entries, keep a buffer of size 2*MaxRequestEntries with a
 	// sliding window 'entries' of size MaxRequestEntries
 	buffer  []lokiEntryWithService
 	entries []lokiEntryWithService
+
+	// store the custom labels for each service
+	labels map[string]json.RawMessage
 }
 
 func NewClient(target *plan.LogTarget) *Client {
@@ -64,13 +65,13 @@ func NewClientWithOptions(target *plan.LogTarget, options *ClientOptions) *Clien
 	options = fillDefaultOptions(options)
 	c := &Client{
 		options:    options,
-		targetName: target.Name,
-		remoteURL:  target.Location,
+		target:     target,
 		httpClient: &http.Client{Timeout: options.RequestTimeout},
 		buffer:     make([]lokiEntryWithService, 2*options.MaxRequestEntries),
+		labels:     make(map[string]json.RawMessage),
 	}
 	// c.entries should be backed by the same array as c.buffer
-	c.entries = c.buffer[0:0:len(c.buffer)]
+	c.entries = c.buffer[:0]
 	return c
 }
 
@@ -84,6 +85,30 @@ func fillDefaultOptions(options *ClientOptions) *ClientOptions {
 	return options
 }
 
+func (c *Client) SetLabels(serviceName string, labels map[string]string) {
+	if labels == nil {
+		delete(c.labels, serviceName)
+		return
+	}
+
+	// Make a copy to avoid altering the original map
+	newLabels := make(map[string]string, len(labels)+1)
+	for k, v := range labels {
+		newLabels[k] = v
+	}
+
+	// Add Loki-specific default labels
+	newLabels["pebble_service"] = serviceName
+
+	// Encode labels now to save time later
+	marshalledLabels, err := json.Marshal(newLabels)
+	if err != nil {
+		// Can't happen as map[string]string will always be marshallable
+		logger.Panicf("Loki client for %q: cannot marshal labels: %v", c.target.Name, err)
+	}
+	c.labels[serviceName] = marshalledLabels
+}
+
 func (c *Client) Add(entry servicelog.Entry) error {
 	if n := len(c.entries); n >= c.options.MaxRequestEntries {
 		// 'entries' is full - remove the first element to make room
@@ -97,7 +122,7 @@ func (c *Client) Add(entry servicelog.Entry) error {
 		copy(c.buffer, c.entries)
 
 		// Reset the view into the buffer
-		c.entries = c.buffer[0:len(c.entries):len(c.buffer)]
+		c.entries = c.buffer[:len(c.entries):len(c.buffer)]
 
 		// Zero removed elements to allow garbage collection
 		for i := len(c.entries); i < len(c.buffer); i++ {
@@ -130,7 +155,7 @@ func (c *Client) Flush(ctx context.Context) error {
 		return fmt.Errorf("encoding request to JSON: %v", err)
 	}
 
-	httpReq, err := http.NewRequestWithContext(ctx, http.MethodPost, c.remoteURL, bytes.NewReader(jsonReq))
+	httpReq, err := http.NewRequestWithContext(ctx, http.MethodPost, c.target.Location, bytes.NewReader(jsonReq))
 	if err != nil {
 		return fmt.Errorf("creating HTTP request: %v", err)
 	}
@@ -148,7 +173,11 @@ func (c *Client) Flush(ctx context.Context) error {
 // resetBuffer drops all buffered logs (in the case of a successful send, or an
 // unrecoverable error).
 func (c *Client) resetBuffer() {
-	c.entries = c.entries[:0]
+	// Zero removed elements to allow garbage collection
+	for i := 0; i < len(c.entries); i++ {
+		c.entries[i] = lokiEntryWithService{}
+	}
+	c.entries = c.buffer[:0]
 }
 
 func (c *Client) buildRequest() lokiRequest {
@@ -169,9 +198,7 @@ func (c *Client) buildRequest() lokiRequest {
 	for _, service := range services {
 		entries := bucketedEntries[service]
 		stream := lokiStream{
-			Labels: map[string]string{
-				"pebble_service": service,
-			},
+			Labels:  c.labels[service],
 			Entries: entries,
 		}
 		req.Streams = append(req.Streams, stream)
@@ -184,8 +211,8 @@ type lokiRequest struct {
 }
 
 type lokiStream struct {
-	Labels  map[string]string `json:"stream"`
-	Entries []lokiEntry       `json:"values"`
+	Labels  json.RawMessage `json:"stream"`
+	Entries []lokiEntry     `json:"values"`
 }
 
 type lokiEntry [2]string
@@ -220,7 +247,7 @@ func (c *Client) handleServerResponse(resp *http.Response) error {
 	case 400 <= code && code < 500:
 		// Other 4xx codes indicate a client problem, so drop the logs (retrying won't help)
 		logger.Noticef("Target %q: request failed with status %d, dropping %d logs",
-			c.targetName, code, len(c.entries))
+			c.target.Name, code, len(c.entries))
 		c.resetBuffer()
 		return errFromResponse(resp)
 
diff --git a/internals/overlord/logstate/loki/loki_test.go b/internals/overlord/logstate/loki/loki_test.go
index 3baef17e7..0bc94df01 100644
--- a/internals/overlord/logstate/loki/loki_test.go
+++ b/internals/overlord/logstate/loki/loki_test.go
@@ -113,11 +113,15 @@ func (*suite) TestRequest(c *C) {
 
 		reqBody, err := io.ReadAll(r.Body)
 		c.Assert(err, IsNil)
-		c.Assert(reqBody, DeepEquals, expected)
+		c.Assert(string(reqBody), DeepEquals, string(expected))
 	}))
 	defer server.Close()
 
 	client := loki.NewClient(&plan.LogTarget{Location: server.URL})
+	client.SetLabels("svc1", map[string]string{})
+	client.SetLabels("svc2", map[string]string{})
+	client.SetLabels("svc3", map[string]string{})
+	client.SetLabels("svc4", map[string]string{})
 	for _, entry := range input {
 		err := client.Add(entry)
 		c.Assert(err, IsNil)
@@ -150,11 +154,11 @@ func (*suite) TestFlushCancelContext(c *C) {
 	flushReturned := make(chan struct{})
 	go func() {
 		// Cancel the Flush context quickly
-		ctx, cancel := context.WithTimeout(context.Background(), 1*time.Microsecond)
+		ctx, cancel := context.WithTimeout(context.Background(), 10*time.Millisecond)
 		defer cancel()
 
 		err := client.Flush(ctx)
-		c.Assert(err, ErrorMatches, ".*context deadline exceeded.*")
+		c.Check(err, ErrorMatches, ".*context deadline exceeded.*")
 		close(flushReturned)
 	}()
 
@@ -245,6 +249,58 @@ func (*suite) TestBufferFull(c *C) {
 	checkBuffer([]any{"5", "6", "7", nil, nil, nil})
 }
 
+func (*suite) TestLabels(c *C) {
+	var expected []byte
+
+	received := make(chan struct{})
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		reqBody, err := io.ReadAll(r.Body)
+		c.Assert(err, IsNil)
+		c.Assert(string(reqBody), Equals, string(expected))
+		close(received)
+	}))
+	defer server.Close()
+
+	client := loki.NewClientWithOptions(
+		&plan.LogTarget{
+			Location: server.URL,
+		},
+		&loki.ClientOptions{},
+	)
+
+	client.SetLabels("svc1", map[string]string{
+		"label1": "val1",
+		"label2": "val2",
+	})
+
+	err := client.Add(servicelog.Entry{
+		Service: "svc1",
+		Time:    time.Date(2023, 10, 3, 4, 20, 33, 0, time.UTC),
+		Message: "hello",
+	})
+	c.Assert(err, IsNil)
+
+	expected = compactJSON(`
+{"streams": [{
+	"stream": {
+		"label1": "val1",
+		"label2": "val2",
+		"pebble_service": "svc1"
+	},
+	"values": [
+		[ "1696306833000000000", "hello" ]
+	]
+}]}`)
+
+	err = client.Flush(context.Background())
+	c.Assert(err, IsNil)
+	select {
+	case <-received:
+	case <-time.After(1 * time.Second):
+		c.Fatal("timed out waiting for request")
+	}
+}
+
 // Strips all extraneous whitespace from JSON
 func compactJSON(s string) []byte {
 	var buf bytes.Buffer
diff --git a/internals/overlord/logstate/manager_test.go b/internals/overlord/logstate/manager_test.go
index 73ef8dfbf..0e480900a 100644
--- a/internals/overlord/logstate/manager_test.go
+++ b/internals/overlord/logstate/manager_test.go
@@ -17,10 +17,13 @@ package logstate
 import (
 	"context"
 	"fmt"
+	"os"
+	"sync"
 	"time"
 
 	. "gopkg.in/check.v1"
 
+	"github.com/canonical/pebble/internals/logger"
 	"github.com/canonical/pebble/internals/plan"
 	"github.com/canonical/pebble/internals/servicelog"
 )
@@ -29,7 +32,13 @@ type managerSuite struct{}
 
 var _ = Suite(&managerSuite{})
 
-func (s *managerSuite) TestPlanChange(c *C) {
+func (*managerSuite) SetUpSuite(c *C) {
+	// Send logs to stderr, so we can see them when debugging
+	l := logger.New(os.Stderr, "[test] ")
+	logger.SetLogger(l)
+}
+
+func (*managerSuite) TestPlanChange(c *C) {
 	gathererOptions := logGathererOptions{
 		newClient: func(target *plan.LogTarget) (logClient, error) {
 			return &testClient{}, nil
@@ -116,12 +125,15 @@ func checkBuffers(c *C, buffers map[string]*servicelog.RingBuffer, expected []st
 }
 
 func (s *managerSuite) TestTimelyShutdown(c *C) {
+	client := &slowFlushingClient{
+		flushTime: 1 * time.Microsecond,
+	}
+
 	gathererOptions := logGathererOptions{
-		timeoutFinalFlush: 5 * time.Millisecond,
-		newClient: func(target *plan.LogTarget) (logClient, error) {
-			return &slowFlushingClient{
-				flushTime: 10 * time.Second,
-			}, nil
+		timeoutCurrentFlush: 5 * time.Millisecond,
+		timeoutFinalFlush:   5 * time.Millisecond,
+		newClient: func(_ *plan.LogTarget) (logClient, error) {
+			return client, nil
 		},
 	}
 
@@ -154,6 +166,7 @@ func (s *managerSuite) TestTimelyShutdown(c *C) {
 	err := svc1.stop()
 	c.Assert(err, IsNil)
 
+	client.SetFlushTime(10 * time.Second)
 	// Stop all gatherers and check this happens quickly
 	done := make(chan struct{})
 	go func() {
@@ -169,6 +182,11 @@ func (s *managerSuite) TestTimelyShutdown(c *C) {
 
 type slowFlushingClient struct {
 	flushTime time.Duration
+	mu        sync.Mutex
+}
+
+func (c *slowFlushingClient) SetLabels(serviceName string, labels map[string]string) {
+	// no-op
 }
 
 func (c *slowFlushingClient) Add(_ servicelog.Entry) error {
@@ -177,10 +195,140 @@ func (c *slowFlushingClient) Add(_ servicelog.Entry) error {
 }
 
 func (c *slowFlushingClient) Flush(ctx context.Context) error {
+	c.mu.Lock()
+	flushTime := c.flushTime
+	c.mu.Unlock()
+
 	select {
 	case <-ctx.Done():
 		return fmt.Errorf("flush timed out")
-	case <-time.After(c.flushTime):
+	case <-time.After(flushTime):
 		return nil
 	}
 }
+
+func (c *slowFlushingClient) SetFlushTime(timeout time.Duration) {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	c.flushTime = timeout
+}
+
+func (s *managerSuite) TestLabels(c *C) {
+	fakeClient := &labelStore{
+		labels:          map[string]map[string]string{},
+		notifySetLabels: make(chan struct{}, 2),
+	}
+
+	m := NewLogManager()
+	m.newGatherer = func(t *plan.LogTarget) (*logGatherer, error) {
+		return newLogGathererInternal(t, &logGathererOptions{
+			newClient: func(_ *plan.LogTarget) (logClient, error) { return fakeClient, nil },
+		})
+	}
+
+	svc1 := newTestService("svc1")
+	svc1.config.Environment = map[string]string{
+		"OWNER": "alice",
+		"IP":    "103.2.51.6",
+		"PORT":  "3456",
+	}
+
+	svc2 := newTestService("svc2")
+	svc2.config.Environment = map[string]string{
+		"IP":   "103.2.52.88",
+		"PORT": "9090",
+	}
+
+	pl := &plan.Plan{
+		Services: map[string]*plan.Service{
+			"svc1": svc1.config,
+			"svc2": svc2.config,
+		},
+		LogTargets: map[string]*plan.LogTarget{
+			"tgt1": {
+				Name:     "tgt1",
+				Type:     plan.LokiTarget,
+				Services: []string{"all"},
+				Labels: map[string]string{
+					"owner":   "user-$OWNER",
+					"address": "http://${IP}:${PORT}",
+				},
+			},
+		},
+	}
+
+	m.PlanChanged(pl)
+	checkGatherers(c, m.gatherers, map[string][]string{
+		"tgt1": nil,
+	})
+
+	m.ServiceStarted(svc1.config, svc1.ringBuffer)
+	m.ServiceStarted(svc2.config, svc2.ringBuffer)
+
+	fakeClient.waitLabels(c)
+	fakeClient.waitLabels(c)
+	c.Assert(fakeClient.labels, DeepEquals, map[string]map[string]string{
+		"svc1": {
+			"owner":   "user-alice",
+			"address": "http://103.2.51.6:3456",
+		},
+		"svc2": {
+			"owner":   "user-", // undefined env vars -> empty string
+			"address": "http://103.2.52.88:9090",
+		},
+	})
+
+	// If we change only the target's labels (with no change to the services),
+	// we still need to recalculate the client's labels.
+	pl.LogTargets["tgt1"].Labels["foo"] = "bar"
+	m.PlanChanged(pl)
+
+	// Wait for labels to be set
+	fakeClient.waitLabels(c)
+	fakeClient.waitLabels(c)
+	c.Assert(fakeClient.labels, DeepEquals, map[string]map[string]string{
+		"svc1": {
+			"owner":   "user-alice",
+			"address": "http://103.2.51.6:3456",
+			"foo":     "bar",
+		},
+		"svc2": {
+			"owner":   "user-",
+			"address": "http://103.2.52.88:9090",
+			"foo":     "bar",
+		},
+	})
+}
+
+// Fake logClient implementation which just stores the passed-in labels
+type labelStore struct {
+	labels map[string]map[string]string
+	// synchronise on this channel to avoid data races
+	notifySetLabels chan struct{}
+}
+
+func (c *labelStore) Add(_ servicelog.Entry) error {
+	return nil // no-op
+}
+
+func (c *labelStore) Flush(_ context.Context) error {
+	return nil // no-op
+}
+
+func (c *labelStore) SetLabels(serviceName string, labels map[string]string) {
+	c.labels[serviceName] = labels
+	select {
+	case c.notifySetLabels <- struct{}{}:
+	case <-time.After(1 * time.Second):
+		panic("timeout waiting for notify for SetLabels")
+	}
+}
+
+// wait for labels to be set
+func (l *labelStore) waitLabels(c *C) {
+	select {
+	case <-l.notifySetLabels:
+	case <-time.After(1 * time.Second):
+		c.Fatal("timed out waiting for labels to be set")
+	}
+}
diff --git a/internals/plan/plan.go b/internals/plan/plan.go
index eccf42a90..4864c1f99 100644
--- a/internals/plan/plan.go
+++ b/internals/plan/plan.go
@@ -488,11 +488,12 @@ func (c *ExecCheck) Merge(other *ExecCheck) {
 
 // LogTarget specifies a remote server to forward logs to.
 type LogTarget struct {
-	Name     string        `yaml:"-"`
-	Type     LogTargetType `yaml:"type"`
-	Location string        `yaml:"location"`
-	Services []string      `yaml:"services"`
-	Override Override      `yaml:"override,omitempty"`
+	Name     string            `yaml:"-"`
+	Type     LogTargetType     `yaml:"type"`
+	Location string            `yaml:"location"`
+	Services []string          `yaml:"services"`
+	Override Override          `yaml:"override,omitempty"`
+	Labels   map[string]string `yaml:"labels,omitempty"`
 }
 
 // LogTargetType defines the protocol to use to forward logs.
@@ -508,6 +509,12 @@ const (
 func (t *LogTarget) Copy() *LogTarget {
 	copied := *t
 	copied.Services = append([]string(nil), t.Services...)
+	if t.Labels != nil {
+		copied.Labels = make(map[string]string)
+		for k, v := range t.Labels {
+			copied.Labels[k] = v
+		}
+	}
 	return &copied
 }
 
@@ -520,6 +527,12 @@ func (t *LogTarget) Merge(other *LogTarget) {
 		t.Location = other.Location
 	}
 	t.Services = append(t.Services, other.Services...)
+	for k, v := range other.Labels {
+		if t.Labels == nil {
+			t.Labels = make(map[string]string)
+		}
+		t.Labels[k] = v
+	}
 }
 
 // FormatError is the error returned when a layer has a format error, such as
@@ -969,6 +982,14 @@ func ParseLayer(order int, label string, data []byte) (*Layer, error) {
 				Message: fmt.Sprintf("log target object cannot be null for log target %q", name),
 			}
 		}
+		for labelName := range target.Labels {
+			// 'pebble_*' labels are reserved
+			if strings.HasPrefix(labelName, "pebble_") {
+				return nil, &FormatError{
+					Message: fmt.Sprintf(`log target %q: label %q uses reserved prefix "pebble_"`, name, labelName),
+				}
+			}
+		}
 		target.Name = name
 	}
 
diff --git a/internals/plan/plan_test.go b/internals/plan/plan_test.go
index c9f999d0d..45160998f 100644
--- a/internals/plan/plan_test.go
+++ b/internals/plan/plan_test.go
@@ -1125,6 +1125,131 @@ var planTests = []planTest{{
 				command: foo
 				override: merge
 `},
+}, {
+	summary: "Log forwarding labels override",
+	input: []string{`
+		log-targets:
+			tgt1:
+				override: merge
+				type: loki
+				location: https://my.loki.server/loki/api/v1/push
+				labels:
+					label1: foo11
+					label2: foo12
+			tgt2:
+				override: merge
+				type: loki
+				location: https://my.loki.server/loki/api/v1/push
+				labels:
+					label1: foo21
+					label2: foo22
+`, `
+		log-targets:
+			tgt1:
+				override: merge
+				labels:
+					label2: bar12
+					label3: bar13
+			tgt2:
+				override: replace
+				type: loki
+				location: https://new.loki.server/loki/api/v1/push
+				labels:
+					label2: bar22
+					label3: bar23
+`},
+	layers: []*plan.Layer{{
+		Order:    0,
+		Label:    "layer-0",
+		Services: map[string]*plan.Service{},
+		Checks:   map[string]*plan.Check{},
+		LogTargets: map[string]*plan.LogTarget{
+			"tgt1": {
+				Name:     "tgt1",
+				Override: plan.MergeOverride,
+				Type:     plan.LokiTarget,
+				Location: "https://my.loki.server/loki/api/v1/push",
+				Labels: map[string]string{
+					"label1": "foo11",
+					"label2": "foo12",
+				},
+			},
+			"tgt2": {
+				Name:     "tgt2",
+				Override: plan.MergeOverride,
+				Type:     plan.LokiTarget,
+				Location: "https://my.loki.server/loki/api/v1/push",
+				Labels: map[string]string{
+					"label1": "foo21",
+					"label2": "foo22",
+				},
+			},
+		},
+	}, {
+		Order:    1,
+		Label:    "layer-1",
+		Services: map[string]*plan.Service{},
+		Checks:   map[string]*plan.Check{},
+		LogTargets: map[string]*plan.LogTarget{
+			"tgt1": {
+				Name:     "tgt1",
+				Override: plan.MergeOverride,
+				Labels: map[string]string{
+					"label2": "bar12",
+					"label3": "bar13",
+				},
+			},
+			"tgt2": {
+				Name:     "tgt2",
+				Override: plan.ReplaceOverride,
+				Type:     plan.LokiTarget,
+				Location: "https://new.loki.server/loki/api/v1/push",
+				Labels: map[string]string{
+					"label2": "bar22",
+					"label3": "bar23",
+				},
+			},
+		},
+	}},
+	result: &plan.Layer{
+		Services: map[string]*plan.Service{},
+		Checks:   map[string]*plan.Check{},
+		LogTargets: map[string]*plan.LogTarget{
+			"tgt1": {
+				Name:     "tgt1",
+				Override: plan.MergeOverride,
+				Type:     plan.LokiTarget,
+				Location: "https://my.loki.server/loki/api/v1/push",
+				Labels: map[string]string{
+					"label1": "foo11",
+					"label2": "bar12",
+					"label3": "bar13",
+				},
+			},
+			"tgt2": {
+				Name:     "tgt2",
+				Override: plan.ReplaceOverride,
+				Type:     plan.LokiTarget,
+				Location: "https://new.loki.server/loki/api/v1/push",
+				Labels: map[string]string{
+					"label2": "bar22",
+					"label3": "bar23",
+				},
+			},
+		},
+	},
+}, {
+	summary: "Reserved log target labels",
+	input: []string{`
+		log-targets:
+			tgt1:
+				override: merge
+				type: loki
+				location: https://my.loki.server/loki/api/v1/push
+				labels:
+					pebble_service: illegal
+`},
+	error: `log target "tgt1": label "pebble_service" uses reserved prefix "pebble_"`,
 }}
 
 func (s *S) TestParseLayer(c *C) {