Initial journald log provider attempt

**What**
- journald log provider using exec to journalctl
```
journalctl -t <namespace>:<name>  --output=json --since=<timestamp> <--follow> --output-fields=SYSLOG_IDENTIFIER,MESSAGE,_PID,_SOURCE_REALTIME_TIMESTAMP
```
- This can be tested manually using `faas-cli logs` as normal, e.g.
  `faas-cli logs nodeinfo` should tail the last 5 mins of logs.
- Very basic tests ensuring that the `journalctl` comamand is correctly
  construction and that the json log entrys are parsed correctly.
- Add simple e2e test to grep the function logs

Signed-off-by: Lucas Roesler <roesler.lucas@gmail.com>

Gopkg.lock

@@ -175,8 +175,8 @@
   name = "github.com/coreos/go-systemd"
   packages = ["journal"]
   pruneopts = "UT"
-  revision = "2d78030078ef61b3cae27f42ad6d0e46db51b339"
-  version = "v22.0.0"
+  revision = "d3cd4ed1dbcf5835feba465b180436db54f20228"
+  version = "v21"
 
 [[projects]]
   digest = "1:92ebc9c068ab8e3fff03a58694ee33830964f6febd0130069aadce328802de14"
@@ -361,12 +361,13 @@
   version = "0.18.10"
 
 [[projects]]
-  digest = "1:6f21508bd38feec0d440ca862f5adcb4c955713f3eb4e075b9af731e6ef258ba"
+  digest = "1:7a20be0bdfb2c05a4a7b955cb71645fe2983aa3c0bbae10d6bba3e2dd26ddd0d"
   name = "github.com/openfaas/faas-provider"
   packages = [
     ".",
     "auth",
     "httputil",
+    "logs",
     "proxy",
     "types",
   ]
@@ -593,6 +594,7 @@
     "github.com/morikuni/aec",
     "github.com/opencontainers/runtime-spec/specs-go",
     "github.com/openfaas/faas-provider",
+    "github.com/openfaas/faas-provider/logs",
     "github.com/openfaas/faas-provider/proxy",
     "github.com/openfaas/faas-provider/types",
     "github.com/openfaas/faas/gateway/requests",

Makefile

@@ -48,3 +48,5 @@ test-e2e:
 	/usr/local/bin/faas-cli remove figlet
 	sleep 3
 	/usr/local/bin/faas-cli list
+	sleep 1
+	/usr/local/bin/faas-cli logs figlet | grep Forking

README.md

@@ -130,12 +130,12 @@ An active community of almost 3000 users awaits you on Slack. Over 250 of those
 * `faas version`
 * `faas namespace`
 * `faas secret`
+* `faas logs`
 
 Scale from and to zero is also supported. On a Dell XPS with a small, pre-pulled image unpausing an existing task took 0.19s and starting a task for a killed function took 0.39s. There may be further optimizations to be gained.
 
 Other operations are pending development in the provider such as:
 
-* `faas logs` - to stream logs on-demand for a known function, for the time being you can find logs via `journalctl -u faasd-provider`
 * `faas auth` - supported for Basic Authentication, but OAuth2 & OIDC require a patch
 
 ## Todo

cmd/provider.go

@@ -11,9 +11,11 @@ import (
 
 	"github.com/containerd/containerd"
 	bootstrap "github.com/openfaas/faas-provider"
+	"github.com/openfaas/faas-provider/logs"
 	"github.com/openfaas/faas-provider/proxy"
 	"github.com/openfaas/faas-provider/types"
 	"github.com/openfaas/faasd/pkg/cninetwork"
+	faasdlogs "github.com/openfaas/faasd/pkg/logs"
 	"github.com/openfaas/faasd/pkg/provider/config"
 	"github.com/openfaas/faasd/pkg/provider/handlers"
 	"github.com/spf13/cobra"
@@ -93,14 +95,7 @@ func makeProviderCmd() *cobra.Command {
 			InfoHandler:          handlers.MakeInfoHandler(Version, GitCommit),
 			ListNamespaceHandler: listNamespaces(),
 			SecretHandler:        handlers.MakeSecretHandler(client, userSecretPath),
-			LogHandler: func(w http.ResponseWriter, r *http.Request) {
-				if r.Body != nil {
-					defer r.Body.Close()
-				}
-
-				w.WriteHeader(http.StatusNotImplemented)
-				w.Write([]byte(`Logs are not implemented for faasd`))
-			},
+			LogHandler:           logs.NewLogHandlerFunc(faasdlogs.New(), config.ReadTimeout),
 		}
 
 		log.Printf("Listening on TCP port: %d\n", *config.TCPPort)

pkg/contants.go

@@ -0,0 +1,6 @@
+package pkg
+
+const (
+	// FunctionNamespace is the default containerd namespace functions are created
+	FunctionNamespace = "openfaas-fn"
+)

pkg/logs/requestor.go

@@ -0,0 +1,185 @@
+package logs
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"io"
+	"log"
+	"os/exec"
+	"strconv"
+	"strings"
+	"time"
+
+	"github.com/openfaas/faas-provider/logs"
+
+	faasd "github.com/openfaas/faasd/pkg"
+)
+
+type requester struct{}
+
+// New returns a new journalctl log Requester
+func New() logs.Requester {
+	return &requester{}
+}
+
+// Query submits a log request to the actual logging system.
+func (r *requester) Query(ctx context.Context, req logs.Request) (<-chan logs.Message, error) {
+	_, err := exec.LookPath("journalctl")
+	if err != nil {
+		return nil, fmt.Errorf("can not find journalctl: %w", err)
+	}
+
+	cmd := buildCmd(ctx, req)
+	stdout, err := cmd.StdoutPipe()
+	if err != nil {
+		return nil, fmt.Errorf("failed to create journalctl pipe: %w", err)
+	}
+
+	stderr, err := cmd.StderrPipe()
+	if err != nil {
+		return nil, fmt.Errorf("failed to create journalctl err pipe: %w", err)
+	}
+
+	err = cmd.Start()
+	if err != nil {
+		return nil, fmt.Errorf("failed to create journalctl: %w", err)
+	}
+
+	// call start and get the stdout prior to streaming so that we can return a meaningful
+	// error for as long as possible. If the cmd starts correctly, we are highly likely to
+	// succeed anyway
+	msgs := make(chan logs.Message, 100)
+	go streamLogs(ctx, cmd, stdout, msgs)
+	go logErrOut(stderr)
+
+	return msgs, nil
+}
+
+// buildCmd reeturns the equivalent of
+//
+// 	journalctl -t <namespace>:<name>  \
+// 		--output=json \
+// 		--since=<timestamp> \
+// 		<--follow> \
+// 		--output-fields=SYSLOG_IDENTIFIER,MESSAGE,_PID,_SOURCE_REALTIME_TIMESTAMP
+func buildCmd(ctx context.Context, req logs.Request) *exec.Cmd {
+	// // set the cursor position based on req, default to 5m
+	since := time.Now().Add(-5 * time.Minute)
+	if req.Since != nil && req.Since.Before(time.Now()) {
+		since = *req.Since
+	}
+
+	namespace := req.Namespace
+	if namespace == "" {
+		namespace = faasd.FunctionNamespace
+	}
+
+	// find the description of the fields here
+	// https://www.freedesktop.org/software/systemd/man/systemd.journal-fields.html
+	// the available fields can vary greatly, the selected fields were detemined by
+	// trial and error with journalctl in an ubuntu VM  (via multipass)
+	args := []string{
+		"--utc",
+		"--no-pager",
+		"--output=json",
+		"--output-fields=SYSLOG_IDENTIFIER,MESSAGE,_PID,_SOURCE_REALTIME_TIMESTAMP",
+		"--identifier=" + namespace + ":" + req.Name,
+		fmt.Sprintf("--since=%s", since.UTC().Format("2006-01-02 15:04:05")),
+	}
+
+	if req.Follow {
+		args = append(args, "--follow")
+	}
+
+	if req.Tail > 0 {
+		args = append(args, fmt.Sprintf("--lines=%d", req.Tail))
+	}
+
+	return exec.CommandContext(ctx, "journalctl", args...)
+}
+
+// streamLogs copies log entries from the journalctl `cmd`/`out` to `msgs`
+// the loop is based on the Decoder example in the docs
+// https://golang.org/pkg/encoding/json/#Decoder.Decode
+func streamLogs(ctx context.Context, cmd *exec.Cmd, out io.ReadCloser, msgs chan logs.Message) {
+	log.Println("starting journal stream using ", cmd.String())
+
+	// will ensure `out` is closed and all related resources cleaned up
+	go func() {
+		err := cmd.Wait()
+		log.Println("wait result", err)
+	}()
+
+	defer func() {
+		log.Println("closing journal stream")
+		close(msgs)
+	}()
+
+	dec := json.NewDecoder(out)
+	for dec.More() {
+		if ctx.Err() != nil {
+			log.Println("log stream context cancelled")
+			return
+		}
+
+		// the journalctl outputs all the values as a string, so a struct with json
+		// tags wont help much
+		entry := map[string]string{}
+		err := dec.Decode(&entry)
+		if err != nil {
+			log.Printf("error decoding journalctl output: %s", err)
+			return
+		}
+
+		msg, err := parseEntry(entry)
+		if err != nil {
+			log.Printf("error parsing journalctl output: %s", err)
+			return
+		}
+
+		msgs <- msg
+	}
+}
+
+// parseEntry reads the deserialized json from journalctl into a log.Message
+//
+// The following fields are parsed from the journal
+// - MESSAGE
+// - _PID
+// - SYSLOG_IDENTIFIER
+// - __REALTIME_TIMESTAMP
+func parseEntry(entry map[string]string) (logs.Message, error) {
+	logMsg := logs.Message{
+		Text:     entry["MESSAGE"],
+		Instance: entry["_PID"],
+	}
+
+	identifier := entry["SYSLOG_IDENTIFIER"]
+	parts := strings.Split(identifier, ":")
+	if len(parts) != 2 {
+		return logMsg, fmt.Errorf("invalid SYSLOG_IDENTIFIER")
+	}
+	logMsg.Namespace = parts[0]
+	logMsg.Name = parts[1]
+
+	ts, ok := entry["__REALTIME_TIMESTAMP"]
+	if !ok {
+		return logMsg, fmt.Errorf("missing required field __REALTIME_TIMESTAMP")
+	}
+
+	ms, err := strconv.ParseInt(ts, 10, 64)
+	if err != nil {
+		return logMsg, fmt.Errorf("invalid timestamp: %w", err)
+	}
+	logMsg.Timestamp = time.Unix(0, ms*1000).UTC()
+
+	return logMsg, nil
+}
+
+func logErrOut(out io.ReadCloser) {
+	defer log.Println("stderr closed")
+	defer out.Close()
+
+	io.Copy(log.Writer(), out)
+}

pkg/logs/requestor_test.go

@@ -0,0 +1,73 @@
+package logs
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/openfaas/faas-provider/logs"
+)
+
+func Test_parseEntry(t *testing.T) {
+	rawEntry := `{ "__CURSOR" : "s=71c4550142d14ace8e2959e3540cc15c;i=133c;b=44864010f0d94baba7b6bf8019f82a56;m=2945cd3;t=5a00d4eb59180;x=8ed47f7f9b3d798", "__REALTIME_TIMESTAMP" : "1583353899094400", "__MONOTONIC_TIMESTAMP" : "43277523", "_BOOT_ID" : "44864010f0d94baba7b6bf8019f82a56", "SYSLOG_IDENTIFIER" : "openfaas-fn:nodeinfo", "_PID" : "2254", "MESSAGE" : "2020/03/04 20:31:39 POST / - 200 OK - ContentLength: 83", "_SOURCE_REALTIME_TIMESTAMP" : "1583353899094372" }`
+	expectedEntry := logs.Message{
+		Name:      "nodeinfo",
+		Namespace: "openfaas-fn",
+		Text:      "2020/03/04 20:31:39 POST / - 200 OK - ContentLength: 83",
+		Timestamp: time.Unix(0, 1583353899094400*1000).UTC(),
+	}
+
+	value := map[string]string{}
+	json.Unmarshal([]byte(rawEntry), &value)
+
+	entry, err := parseEntry(value)
+	if err != nil {
+		t.Fatalf("unexpected error %s", err)
+	}
+
+	if entry.Name != expectedEntry.Name {
+		t.Fatalf("expected Name %s, got %s", expectedEntry.Name, entry.Name)
+	}
+
+	if entry.Namespace != expectedEntry.Namespace {
+		t.Fatalf("expected Namespace %s, got %s", expectedEntry.Namespace, entry.Namespace)
+	}
+
+	if entry.Timestamp != expectedEntry.Timestamp {
+		t.Fatalf("expected Timestamp %s, got %s", expectedEntry.Timestamp, entry.Timestamp)
+	}
+
+	if entry.Text != expectedEntry.Text {
+		t.Fatalf("expected Text %s, got %s", expectedEntry.Text, entry.Text)
+	}
+}
+
+func Test_buildCmd(t *testing.T) {
+	ctx := context.TODO()
+	now := time.Now()
+	req := logs.Request{
+		Name:      "loggyfunc",
+		Namespace: "spacetwo",
+		Follow:    true,
+		Since:     &now,
+		Tail:      5,
+	}
+
+	expectedArgs := fmt.Sprintf(
+		"--utc --no-pager --output=json --output-fields=SYSLOG_IDENTIFIER,MESSAGE,_PID,_SOURCE_REALTIME_TIMESTAMP --identifier=spacetwo:loggyfunc --since=%s --follow --lines=5",
+		now.UTC().Format("2006-01-02 15:04:05"),
+	)
+
+	cmd := buildCmd(ctx, req).String()
+
+	if !strings.Contains(cmd, "journalctl") {
+		t.Fatalf("expected journalctl cmd, got cmd %s", cmd)
+	}
+
+	if !strings.HasSuffix(cmd, expectedArgs) {
+		t.Fatalf("expected arg %s,\ngot cmd %s", expectedArgs, cmd)
+	}
+}

pkg/provider/handlers/delete.go

@@ -12,6 +12,8 @@ import (
 	"github.com/containerd/containerd/namespaces"
 	gocni "github.com/containerd/go-cni"
 	"github.com/openfaas/faas/gateway/requests"
+
+	faasd "github.com/openfaas/faasd/pkg"
 	cninetwork "github.com/openfaas/faasd/pkg/cninetwork"
 	"github.com/openfaas/faasd/pkg/service"
 )
@@ -49,7 +51,7 @@ func MakeDeleteHandler(client *containerd.Client, cni gocni.CNI) func(w http.Res
 			return
 		}
 
-		ctx := namespaces.WithNamespace(context.Background(), FunctionNamespace)
+		ctx := namespaces.WithNamespace(context.Background(), faasd.FunctionNamespace)
 
 		// TODO: this needs to still happen if the task is paused
 		if function.replicas != 0 {

pkg/provider/handlers/deploy.go

@@ -18,6 +18,7 @@ import (
 	"github.com/docker/distribution/reference"
 	"github.com/opencontainers/runtime-spec/specs-go"
 	"github.com/openfaas/faas-provider/types"
+	faasd "github.com/openfaas/faasd/pkg"
 	cninetwork "github.com/openfaas/faasd/pkg/cninetwork"
 	"github.com/openfaas/faasd/pkg/service"
 	"github.com/pkg/errors"
@@ -52,7 +53,7 @@ func MakeDeployHandler(client *containerd.Client, cni gocni.CNI, secretMountPath
 		}
 
 		name := req.Service
-		ctx := namespaces.WithNamespace(context.Background(), FunctionNamespace)
+		ctx := namespaces.WithNamespace(context.Background(), faasd.FunctionNamespace)
 
 		deployErr := deploy(ctx, req, client, cni, secretMountPath, alwaysPull)
 		if deployErr != nil {

pkg/provider/handlers/functions.go

@@ -8,6 +8,8 @@ import (
 	"github.com/containerd/containerd"
 	"github.com/containerd/containerd/namespaces"
 	"github.com/openfaas/faasd/pkg/cninetwork"
+
+	faasd "github.com/openfaas/faasd/pkg"
 )
 
 type Function struct {
@@ -20,14 +22,9 @@ type Function struct {
 	labels    map[string]string
 }
 
-const (
-	// FunctionNamespace is the containerd namespace functions are created
-	FunctionNamespace = "openfaas-fn"
-)
-
 // ListFunctions returns a map of all functions with running tasks on namespace
 func ListFunctions(client *containerd.Client) (map[string]Function, error) {
-	ctx := namespaces.WithNamespace(context.Background(), FunctionNamespace)
+	ctx := namespaces.WithNamespace(context.Background(), faasd.FunctionNamespace)
 	functions := make(map[string]Function)
 
 	containers, _ := client.Containers(ctx)
@@ -44,7 +41,7 @@ func ListFunctions(client *containerd.Client) (map[string]Function, error) {
 
 // GetFunction returns a function that matches name
 func GetFunction(client *containerd.Client, name string) (Function, error) {
-	ctx := namespaces.WithNamespace(context.Background(), FunctionNamespace)
+	ctx := namespaces.WithNamespace(context.Background(), faasd.FunctionNamespace)
 	c, err := client.LoadContainer(ctx, name)
 
 	if err == nil {
@@ -58,7 +55,7 @@ func GetFunction(client *containerd.Client, name string) (Function, error) {
 
 		f := Function{
 			name:      containerName,
-			namespace: FunctionNamespace,
+			namespace: faasd.FunctionNamespace,
 			image:     image.Name(),
 			labels:    labels,
 		}

pkg/provider/handlers/scale.go

@@ -12,6 +12,7 @@ import (
 	"github.com/containerd/containerd/namespaces"
 	gocni "github.com/containerd/go-cni"
 	"github.com/openfaas/faas-provider/types"
+	faasd "github.com/openfaas/faasd/pkg"
 )
 
 func MakeReplicaUpdateHandler(client *containerd.Client, cni gocni.CNI) func(w http.ResponseWriter, r *http.Request) {
@@ -47,7 +48,7 @@ func MakeReplicaUpdateHandler(client *containerd.Client, cni gocni.CNI) func(w h
 			return
 		}
 
-		ctx := namespaces.WithNamespace(context.Background(), FunctionNamespace)
+		ctx := namespaces.WithNamespace(context.Background(), faasd.FunctionNamespace)
 
 		ctr, ctrErr := client.LoadContainer(ctx, name)
 		if ctrErr != nil {

pkg/provider/handlers/update.go

@@ -12,6 +12,8 @@ import (
 	"github.com/containerd/containerd/namespaces"
 	gocni "github.com/containerd/go-cni"
 	"github.com/openfaas/faas-provider/types"
+
+	faasd "github.com/openfaas/faasd/pkg"
 	"github.com/openfaas/faasd/pkg/cninetwork"
 	"github.com/openfaas/faasd/pkg/service"
 )
@@ -53,7 +55,7 @@ func MakeUpdateHandler(client *containerd.Client, cni gocni.CNI, secretMountPath
 			http.Error(w, err.Error(), http.StatusBadRequest)
 		}
 
-		ctx := namespaces.WithNamespace(context.Background(), FunctionNamespace)
+		ctx := namespaces.WithNamespace(context.Background(), faasd.FunctionNamespace)
 		if function.replicas != 0 {
 			err = cninetwork.DeleteCNINetwork(ctx, cni, client, name)
 			if err != nil {

vendor/github.com/openfaas/faas-provider/logs/handler.go

@@ -0,0 +1,144 @@
+package logs
+
+import (
+	"context"
+	"encoding/json"
+	"log"
+	"net/http"
+	"net/url"
+	"strconv"
+	"time"
+
+	"github.com/openfaas/faas-provider/httputil"
+)
+
+// Requester submits queries the logging system.
+// This will be passed to the log handler constructor.
+type Requester interface {
+	// Query submits a log request to the actual logging system.
+	Query(context.Context, Request) (<-chan Message, error)
+}
+
+// NewLogHandlerFunc creates an http HandlerFunc from the supplied log Requestor.
+func NewLogHandlerFunc(requestor Requester, timeout time.Duration) http.HandlerFunc {
+	return func(w http.ResponseWriter, r *http.Request) {
+		if r.Body != nil {
+			defer r.Body.Close()
+		}
+
+		cn, ok := w.(http.CloseNotifier)
+		if !ok {
+			log.Println("LogHandler: response is not a CloseNotifier, required for streaming response")
+			http.NotFound(w, r)
+			return
+		}
+		flusher, ok := w.(http.Flusher)
+		if !ok {
+			log.Println("LogHandler: response is not a Flusher, required for streaming response")
+			http.NotFound(w, r)
+			return
+		}
+
+		logRequest, err := parseRequest(r)
+		if err != nil {
+			log.Printf("LogHandler: could not parse request %s", err)
+			httputil.Errorf(w, http.StatusUnprocessableEntity, "could not parse the log request")
+			return
+		}
+
+		ctx, cancelQuery := context.WithTimeout(r.Context(), timeout)
+		defer cancelQuery()
+		messages, err := requestor.Query(ctx, logRequest)
+		if err != nil {
+			// add smarter error handling here
+			httputil.Errorf(w, http.StatusInternalServerError, "function log request failed")
+			return
+		}
+
+		// Send the initial headers saying we're gonna stream the response.
+		w.Header().Set("Connection", "Keep-Alive")
+		w.Header().Set("Transfer-Encoding", "chunked")
+		w.Header().Set(http.CanonicalHeaderKey("Content-Type"), "application/x-ndjson")
+		w.WriteHeader(http.StatusOK)
+		flusher.Flush()
+
+		// ensure that we always try to send the closing chunk, not the inverted order due to how
+		// the defer stack works. We need two flush statements to ensure that the empty slice is
+		// sent as its own chunk
+		defer flusher.Flush()
+		defer w.Write([]byte{})
+		defer flusher.Flush()
+
+		jsonEncoder := json.NewEncoder(w)
+		for messages != nil {
+			select {
+			case <-cn.CloseNotify():
+				log.Println("LogHandler: client stopped listening")
+				return
+			case msg, ok := <-messages:
+				if !ok {
+					log.Println("LogHandler: end of log stream")
+					messages = nil
+					return
+				}
+
+				// serialize and write the msg to the http ResponseWriter
+				err := jsonEncoder.Encode(msg)
+				if err != nil {
+					// can't actually write the status header here so we should json serialize an error
+					// and return that because we have already sent the content type and status code
+					log.Printf("LogHandler: failed to serialize log message: '%s'\n", msg.String())
+					log.Println(err.Error())
+					// write json error message here ?
+					jsonEncoder.Encode(Message{Text: "failed to serialize log message"})
+					flusher.Flush()
+					return
+				}
+
+				flusher.Flush()
+			}
+		}
+
+		return
+	}
+}
+
+// parseRequest extracts the logRequest from the GET variables or from the POST body
+func parseRequest(r *http.Request) (logRequest Request, err error) {
+	query := r.URL.Query()
+	logRequest.Name = getValue(query, "name")
+	logRequest.Namespace = getValue(query, "namespace")
+	logRequest.Instance = getValue(query, "instance")
+	tailStr := getValue(query, "tail")
+	if tailStr != "" {
+		logRequest.Tail, err = strconv.Atoi(tailStr)
+		if err != nil {
+			return logRequest, err
+		}
+	}
+
+	// ignore error because it will default to false if we can't parse it
+	logRequest.Follow, _ = strconv.ParseBool(getValue(query, "follow"))
+
+	sinceStr := getValue(query, "since")
+	if sinceStr != "" {
+		since, err := time.Parse(time.RFC3339, sinceStr)
+		logRequest.Since = &since
+		if err != nil {
+			return logRequest, err
+		}
+	}
+
+	return logRequest, nil
+}
+
+// getValue returns the value for the given key. If the key has more than one value, it returns the
+// last value. if the value does not exist, it returns the empty string.
+func getValue(queryValues url.Values, name string) string {
+	values := queryValues[name]
+	if len(values) == 0 {
+		return ""
+	}
+
+	return values[len(values)-1]
+}

vendor/github.com/openfaas/faas-provider/logs/logs.go

@@ -0,0 +1,62 @@
+// Package logs provides the standard interface and handler for OpenFaaS providers to expose function logs.
+//
+// The package defines the Requester interface that OpenFaaS providers should implement and then expose using
+// the predefined NewLogHandlerFunc. See the example folder for a minimal log provider implementation.
+//
+// The Requester is where the actual specific logic for connecting to and querying the log system should be implemented.
+//
+package logs
+
+import (
+	"fmt"
+	"time"
+)
+
+// Request is the query to return the function logs.
+type Request struct {
+	// Name is the function name and is required
+	Name string `json:"name"`
+	// Namespace is the namespace the function is deployed to, how a namespace is defined
+	// is faas-provider specific
+	Namespace string `json:"namespace"`
+	// Instance is the optional container name, that allows you to request logs from a specific function instance
+	Instance string `json:"instance"`
+	// Since is the optional datetime value to start the logs from
+	Since *time.Time `json:"since"`
+	// Tail sets the maximum number of log messages to return, <=0 means unlimited
+	Tail int `json:"tail"`
+	// Follow is allows the user to request a stream of logs until the timeout
+	Follow bool `json:"follow"`
+}
+
+// String implements that Stringer interface and prints the log Request in a consistent way that
+// allows you to safely compare if two requests have the same value.
+func (r Request) String() string {
+	return fmt.Sprintf(
+		"name:%s namespace: %s instance:%s since:%v tail:%d follow:%v",
+		r.Name, r.Namespace, r.Instance, r.Since, r.Tail, r.Follow,
+	)
+}
+
+// Message is a specific log message from a function container log stream
+type Message struct {
+	// Name is the function name
+	Name string `json:"name"`
+	// Namespace is the namespace the function is deployed to, how a namespace is defined
+	// is faas-provider specific
+	Namespace string `json:"namespace"`
+	// instance is the name/id of the specific function instance
+	Instance string `json:"instance"`
+	// Timestamp is the timestamp of when the log message was recorded
+	Timestamp time.Time `json:"timestamp"`
+	// Text is the raw log message content
+	Text string `json:"text"`
+}
+
+// String implements the Stringer interface and allows for nice and simple string formatting of a log Message.
+func (m Message) String() string {
+	return fmt.Sprintf(
+		"%s %s (%s %s) %s",
+		m.Timestamp.String(), m.Name, m.Namespace, m.Instance, m.Text,
+	)
+}