Stream logs via podman api

.github/workflows/build.yml

@@ -1,6 +1,6 @@
 name: build
 
-on:
+on: 
   pull_request:
     branches:
       - main

CHANGELOG.md

@@ -2,6 +2,7 @@
 
 FEATURES:
 * config: Map host devices into container. [[GH-41](https://github.com/hashicorp/nomad-driver-podman/pull/41)]
+* config: Stream logs via API, support journald log driver. [[GH-99](https://github.com/hashicorp/nomad-driver-podman/pull/99)]
 
 BUG FIXES:
 * log: Use error key context to log errors rather than Go err style. [[GH-126](https://github.com/hashicorp/nomad-driver-podman/pull/126)]
@@ -14,6 +15,7 @@ BUG FIXES:
 * config: Fixed a bug where we always pulled an image if image name has a transport prefix [[GH-88](https://github.com/hashicorp/nomad-driver-podman/pull/88)]
 * config: Added labels option
 * config: Add force_pull option
+* config: Added logging options
 
 BUG FIXES:
 * [[GH-93](https://github.com/hashicorp/nomad-driver-podman/issues/93)] use slirp4netns as default network mode if running rootless

README.md

@@ -154,6 +154,18 @@ plugin "nomad-driver-podman" {
   }
 }
 ```
+
+* disable_log_collection (string) Defaults to `false`. Setting this to true will disable Nomad logs collection of Podman tasks. If you don't rely on nomad log capabilities and exclusively use host based log aggregation, you may consider this option to disable nomad log collection overhead. Beware to you also loose automatic log rotation.
+
+
+```
+plugin "nomad-driver-podman" {
+  config {
+    disable_log_collection = false
+  }
+}
+```
+
 ## Task Configuration
 
 * **image** - The image to run. Accepted transports are `docker` (default if missing), `oci-archive` and `docker-archive`. Images reference as [short-names](https://github.com/containers/image/blob/master/docs/containers-registries.conf.5.md#short-name-aliasing) will be treated according to user-configured preferences.
@@ -273,6 +285,38 @@ config {
 
 ```
 
+* **logging** - Configure logging. See also plugin option **disable_log_collection**
+
+`driver = "nomad"` (default) Podman redirects its combined stdout/stderr logstream directly to a nomad fifo.
+Benefits of this mode are: zero overhead, don't have to worry about log rotation at system or Podman level. Downside: you can not easily ship the logstream to a log aggregator plus stdout/stderr is multiplexed into a single stream..
+
+```
+config {
+  logging = {
+    driver = "nomad"
+  }
+}
+```
+
+`driver = "journald"` The container log is forwarded from Podman to the journald on your host. Next, it's pulled by the Podman API back from the journal into the Nomad fifo (controllable by **disable_log_collection**)
+Benefits: all containers can log into the host journal, you can ship a structured stream incl. metadata to your log aggregator. No log rotation at Podman level. You can add additional tags to the journal.
+Drawbacks: a bit more overhead, depends on Journal (will not work on WSL2). You should configure some rotation policy for your Journal.
+Ensure you're running Podman 3.1.0 or higher because of bugs in older versions.
+
+```
+config {
+  logging = {
+    driver = "journald"
+    options = [
+      {
+        "tag" = "redis"
+      }
+    ]
+  }
+}
+```
+
+
 * **memory_reservation** - Memory soft limit (nit = b (bytes), k (kilobytes), m (megabytes), or g (gigabytes))
 
 After setting memory reservation, when the system detects memory contention or low memory, containers are forced to restrict their consumption to their reservation. So you should always set the value below --memory, otherwise the hard limit will take precedence. By default, memory reservation will be the same as memory limit.

api/api.go

@@ -16,9 +16,10 @@ import (
 )
 
 type API struct {
-	baseUrl    string
-	httpClient *http.Client
-	logger     hclog.Logger
+	baseUrl          string
+	httpClient       *http.Client
+	httpStreamClient *http.Client
+	logger           hclog.Logger
 }
 
 type ClientConfig struct {
@@ -56,6 +57,8 @@ func NewClient(logger hclog.Logger, config ClientConfig) *API {
 	ac.httpClient = &http.Client{
 		Timeout: config.HttpTimeout,
 	}
+	// we do not want a timeout for streaming requests.
+	ac.httpStreamClient = &http.Client{}
 	if strings.HasPrefix(baseUrl, "unix:") {
 		ac.baseUrl = "http://u"
 		path := strings.TrimPrefix(baseUrl, "unix:")
@@ -64,6 +67,7 @@ func NewClient(logger hclog.Logger, config ClientConfig) *API {
 				return net.Dial("unix", path)
 			},
 		}
+		ac.httpStreamClient.Transport = ac.httpClient.Transport
 	} else {
 		ac.baseUrl = baseUrl
 	}
@@ -84,6 +88,14 @@ func (c *API) Get(ctx context.Context, path string) (*http.Response, error) {
 	return c.Do(req)
 }
 
+func (c *API) GetStream(ctx context.Context, path string) (*http.Response, error) {
+	req, err := http.NewRequestWithContext(ctx, "GET", c.baseUrl+path, nil)
+	if err != nil {
+		return nil, err
+	}
+	return c.httpStreamClient.Do(req)
+}
+
 func (c *API) Post(ctx context.Context, path string, body io.Reader) (*http.Response, error) {
 	return c.PostWithHeaders(ctx, path, body, map[string]string{})
 }

api/container_logs.go

@@ -0,0 +1,62 @@
+package api
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"io"
+	"net/http"
+	"time"
+)
+
+// ContainerLogs gets stdout and stderr logs from a container.
+func (c *API) ContainerLogs(ctx context.Context, name string, since time.Time, stdout io.Writer, stderr io.Writer) error {
+
+	res, err := c.GetStream(ctx, fmt.Sprintf("/v1.0.0/libpod/containers/%s/logs?follow=true&since=%d&stdout=true&stderr=true", name, since.Unix()))
+	if err != nil {
+		return err
+	}
+
+	if res.StatusCode != http.StatusOK {
+		return fmt.Errorf("unknown error, status code: %d", res.StatusCode)
+	}
+
+	c.logger.Debug("Running log stream", "container", name)
+	defer func() {
+		res.Body.Close()
+		c.logger.Debug("Stopped log stream", "container", name)
+	}()
+	buffer := make([]byte, 1024)
+	for {
+		fd, l, err := DemuxHeader(res.Body, buffer)
+		if err != nil {
+			if errors.Is(err, io.EOF) {
+				return nil
+			}
+			return err
+		}
+		frame, err := DemuxFrame(res.Body, buffer, l)
+		if err != nil {
+			return err
+		}
+
+		c.logger.Trace("logger", "frame", string(frame), "fd", fd)
+
+		switch fd {
+		case 0:
+			stdout.Write(frame)
+			stdout.Write([]byte("\n"))
+		case 1:
+			stdout.Write(frame)
+			stdout.Write([]byte("\n"))
+		case 2:
+			stderr.Write(frame)
+			stderr.Write([]byte("\n"))
+		case 3:
+			return fmt.Errorf("Error from log service: %s", string(frame))
+		default:
+			return fmt.Errorf("Unknown log stream identifier: %d", fd)
+		}
+	}
+
+}

api/demux.go

@@ -0,0 +1,46 @@
+package api
+
+import (
+	"encoding/binary"
+	"fmt"
+	"io"
+)
+
+// DemuxHeader reads header for stream from server multiplexed stdin/stdout/stderr/2nd error channel
+func DemuxHeader(r io.Reader, buffer []byte) (fd, sz int, err error) {
+	n, err := io.ReadFull(r, buffer[0:8])
+	if err != nil {
+		return
+	}
+	if n < 8 {
+		err = io.ErrUnexpectedEOF
+		return
+	}
+
+	fd = int(buffer[0])
+	if fd < 0 || fd > 3 {
+		err = fmt.Errorf(`channel "%d" found, 0-3 supported`, fd)
+		return
+	}
+
+	sz = int(binary.BigEndian.Uint32(buffer[4:8]))
+	return
+}
+
+// DemuxFrame reads contents for frame from server multiplexed stdin/stdout/stderr/2nd error channel
+func DemuxFrame(r io.Reader, buffer []byte, length int) (frame []byte, err error) {
+	if len(buffer) < length {
+		buffer = append(buffer, make([]byte, length-len(buffer)+1)...)
+	}
+
+	n, err := io.ReadFull(r, buffer[0:length])
+	if err != nil {
+		return nil, nil
+	}
+	if n < length {
+		err = io.ErrUnexpectedEOF
+		return
+	}
+
+	return buffer[0:length], nil
+}

api/exec_start.go

@@ -3,7 +3,6 @@ package api
 import (
 	"bytes"
 	"context"
-	"encoding/binary"
 	"encoding/json"
 	"errors"
 	"fmt"
@@ -40,45 +39,6 @@ type ExecStartRequest struct {
 	AttachInput bool
 }
 
-// DemuxHeader reads header for stream from server multiplexed stdin/stdout/stderr/2nd error channel
-func DemuxHeader(r io.Reader, buffer []byte) (fd, sz int, err error) {
-	n, err := io.ReadFull(r, buffer[0:8])
-	if err != nil {
-		return
-	}
-	if n < 8 {
-		err = io.ErrUnexpectedEOF
-		return
-	}
-
-	fd = int(buffer[0])
-	if fd < 0 || fd > 3 {
-		err = fmt.Errorf(`channel "%d" found, 0-3 supported`, fd)
-		return
-	}
-
-	sz = int(binary.BigEndian.Uint32(buffer[4:8]))
-	return
-}
-
-// DemuxFrame reads contents for frame from server multiplexed stdin/stdout/stderr/2nd error channel
-func DemuxFrame(r io.Reader, buffer []byte, length int) (frame []byte, err error) {
-	if len(buffer) < length {
-		buffer = append(buffer, make([]byte, length-len(buffer)+1)...)
-	}
-
-	n, err := io.ReadFull(r, buffer[0:length])
-	if err != nil {
-		return nil, nil
-	}
-	if n < length {
-		err = io.ErrUnexpectedEOF
-		return
-	}
-
-	return buffer[0:length], nil
-}
-
 // This is intended to be run as a goroutine, handling resizing for a container
 // or exec session.
 func (c *API) attachHandleResize(ctx context.Context, resizeChannel <-chan drivers.TerminalSize, sessionId string) {

config.go

@@ -34,6 +34,9 @@ var (
 		),
 		// the path to the podman api socket
 		"socket_path": hclspec.NewAttr("socket_path", "string", false),
+		// disable_log_collection indicates whether nomad should collect logs of podman
+		// task containers.  If true, logs are not forwarded to nomad.
+		"disable_log_collection": hclspec.NewAttr("disable_log_collection", "bool", false),
 	})
 
 	// taskConfigSpec is the hcl specification for the driver config section of
@@ -44,17 +47,21 @@ var (
 			"username": hclspec.NewAttr("username", "string", false),
 			"password": hclspec.NewAttr("password", "string", false),
 		})),
-		"command":            hclspec.NewAttr("command", "string", false),
-		"cap_add":            hclspec.NewAttr("cap_add", "list(string)", false),
-		"cap_drop":           hclspec.NewAttr("cap_drop", "list(string)", false),
-		"devices":            hclspec.NewAttr("devices", "list(string)", false),
-		"entrypoint":         hclspec.NewAttr("entrypoint", "string", false),
-		"working_dir":        hclspec.NewAttr("working_dir", "string", false),
-		"hostname":           hclspec.NewAttr("hostname", "string", false),
-		"image":              hclspec.NewAttr("image", "string", true),
-		"init":               hclspec.NewAttr("init", "bool", false),
-		"init_path":          hclspec.NewAttr("init_path", "string", false),
-		"labels":             hclspec.NewAttr("labels", "list(map(string))", false),
+		"command":     hclspec.NewAttr("command", "string", false),
+		"cap_add":     hclspec.NewAttr("cap_add", "list(string)", false),
+		"cap_drop":    hclspec.NewAttr("cap_drop", "list(string)", false),
+		"devices":     hclspec.NewAttr("devices", "list(string)", false),
+		"entrypoint":  hclspec.NewAttr("entrypoint", "string", false),
+		"working_dir": hclspec.NewAttr("working_dir", "string", false),
+		"hostname":    hclspec.NewAttr("hostname", "string", false),
+		"image":       hclspec.NewAttr("image", "string", true),
+		"init":        hclspec.NewAttr("init", "bool", false),
+		"init_path":   hclspec.NewAttr("init_path", "string", false),
+		"labels":      hclspec.NewAttr("labels", "list(map(string))", false),
+		"logging": hclspec.NewBlock("logging", false, hclspec.NewObject(map[string]*hclspec.Spec{
+			"driver":  hclspec.NewAttr("driver", "string", false),
+			"options": hclspec.NewAttr("options", "list(map(string))", false),
+		})),
 		"memory_reservation": hclspec.NewAttr("memory_reservation", "string", false),
 		"memory_swap":        hclspec.NewAttr("memory_swap", "string", false),
 		"memory_swappiness":  hclspec.NewAttr("memory_swappiness", "number", false),
@@ -80,6 +87,12 @@ type GCConfig struct {
 	Container bool `codec:"container"`
 }
 
+// LoggingConfig is the tasks logging configuration
+type LoggingConfig struct {
+	Driver  string             `codec:"driver"`
+	Options hclutils.MapStrStr `codec:"options"`
+}
+
 // VolumeConfig is the drivers volume specific configuration
 type VolumeConfig struct {
 	Enabled      bool   `codec:"enabled"`
@@ -88,10 +101,11 @@ type VolumeConfig struct {
 
 // PluginConfig is the driver configuration set by the SetConfig RPC call
 type PluginConfig struct {
-	Volumes        VolumeConfig `codec:"volumes"`
-	GC             GCConfig     `codec:"gc"`
-	RecoverStopped bool         `codec:"recover_stopped"`
-	SocketPath     string       `codec:"socket_path"`
+	Volumes              VolumeConfig `codec:"volumes"`
+	GC                   GCConfig     `codec:"gc"`
+	RecoverStopped       bool         `codec:"recover_stopped"`
+	DisableLogCollection bool         `codec:"disable_log_collection"`
+	SocketPath           string       `codec:"socket_path"`
 }
 
 // TaskConfig is the driver configuration of a task within a job
@@ -110,6 +124,7 @@ type TaskConfig struct {
 	Hostname          string             `codec:"hostname"`
 	Image             string             `codec:"image"`
 	InitPath          string             `codec:"init_path"`
+	Logging           LoggingConfig      `codec:"logging"`
 	Labels            hclutils.MapStrStr `codec:"labels"`
 	MemoryReservation string             `codec:"memory_reservation"`
 	MemorySwap        string             `codec:"memory_swap"`