Remove output fields from the logs test case

**What**
- Fix the test to match the new logs command builder, without the output
  fields flag

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

.travis.yml

@@ -10,6 +10,7 @@ addons:
       - runc
 
 script:
+- make test
 - make dist
 - make prepare-test
 - make test-e2e

Gopkg.lock

@@ -175,8 +175,8 @@
   name = "github.com/coreos/go-systemd"
   packages = ["journal"]
   pruneopts = "UT"
-  revision = "2d78030078ef61b3cae27f42ad6d0e46db51b339"
+  revision = "d3cd4ed1dbcf5835feba465b180436db54f20228"
-  version = "v22.0.0"
+  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

@@ -11,6 +11,10 @@ all: local
 local:
 	CGO_ENABLED=0 GOOS=linux go build -o bin/faasd
 
+.PHONY: test
+test:
+	CGO_ENABLED=0 GOOS=linux go test -ldflags $(LDFLAGS) ./...
+
 .PHONY: dist
 dist:
 	CGO_ENABLED=0 GOOS=linux go build -ldflags $(LDFLAGS) -a -installsuffix cgo -o bin/faasd
@@ -37,9 +41,10 @@ prepare-test:
 .PHONY: test-e2e
 test-e2e:
 	sudo cat /var/lib/faasd/secrets/basic-auth-password | /usr/local/bin/faas-cli login --password-stdin
-	/usr/local/bin/faas-cli store deploy figlet --env write_timeout=1s --env read_timeout=1s
+	/usr/local/bin/faas-cli store deploy figlet --env write_timeout=1s --env read_timeout=1s --label testing=true
 	sleep 5
 	/usr/local/bin/faas-cli list -v
+	/usr/local/bin/faas-cli describe figlet | grep testing
 	uname | /usr/local/bin/faas-cli invoke figlet
 	uname | /usr/local/bin/faas-cli invoke figlet --async
 	sleep 10
@@ -47,3 +52,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 --follow=false | grep Forking

README.md

@@ -3,6 +3,7 @@
 [![Build Status](https://travis-ci.com/openfaas/faasd.svg?branch=master)](https://travis-ci.com/openfaas/faasd)
 [![License: MIT](https://img.shields.io/badge/License-MIT-yellow.svg)](https://opensource.org/licenses/MIT)
 [![OpenFaaS](https://img.shields.io/badge/openfaas-serverless-blue.svg)](https://www.openfaas.com)
+![Downloads](https://img.shields.io/github/downloads/openfaas/faasd/total)
 
 faasd is the same OpenFaaS experience and ecosystem, but without Kubernetes. Functions and microservices can be deployed anywhere with reduced overheads whilst retaining the portability of containers and cloud-native tooling.
 
@@ -130,12 +131,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

cloud-config.txt

@@ -17,7 +17,7 @@ runcmd:
 - curl -sSL https://github.com/containernetworking/plugins/releases/download/v0.8.5/cni-plugins-linux-amd64-v0.8.5.tgz | tar -xz -C /opt/cni/bin
 - mkdir -p /go/src/github.com/openfaas/
 - cd /go/src/github.com/openfaas/ && git clone https://github.com/openfaas/faasd
-- curl -fSLs "https://github.com/openfaas/faasd/releases/download/0.7.7/faasd" --output "/usr/local/bin/faasd" && chmod a+x "/usr/local/bin/faasd"
+- curl -fSLs "https://github.com/openfaas/faasd/releases/download/0.8.1/faasd" --output "/usr/local/bin/faasd" && chmod a+x "/usr/local/bin/faasd"
 - cd /go/src/github.com/openfaas/faasd/ && /usr/local/bin/faasd install
 - systemctl status -l containerd --no-pager
 - journalctl -u faasd-provider --no-pager

cmd/provider.go

@@ -11,21 +11,35 @@ 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"
 )
 
-var providerCmd = &cobra.Command{
+func makeProviderCmd() *cobra.Command {
+	var command = &cobra.Command{
 		Use:   "provider",
 		Short: "Run the faasd-provider",
-	RunE:  runProvider,
 	}
 
-func runProvider(_ *cobra.Command, _ []string) error {
+	command.Flags().String("pull-policy", "Always", `Set to "Always" to force a pull of images upon deployment, or "IfNotPresent" to try to use a cached image.`)
+
+	command.RunE = func(_ *cobra.Command, _ []string) error {
+
+		pullPolicy, flagErr := command.Flags().GetString("pull-policy")
+		if flagErr != nil {
+			return flagErr
+		}
+
+		alwaysPull := false
+		if pullPolicy == "Always" {
+			alwaysPull = true
+		}
 
 		config, providerConfig, err := config.ReadFromEnv(types.OsEnv{})
 		if err != nil {
@@ -72,31 +86,26 @@ func runProvider(_ *cobra.Command, _ []string) error {
 		bootstrapHandlers := types.FaaSHandlers{
 			FunctionProxy:        proxy.NewHandlerFunc(*config, invokeResolver),
 			DeleteHandler:        handlers.MakeDeleteHandler(client, cni),
-		DeployHandler:        handlers.MakeDeployHandler(client, cni, userSecretPath),
+			DeployHandler:        handlers.MakeDeployHandler(client, cni, userSecretPath, alwaysPull),
 			FunctionReader:       handlers.MakeReadHandler(client),
 			ReplicaReader:        handlers.MakeReplicaReaderHandler(client),
 			ReplicaUpdater:       handlers.MakeReplicaUpdateHandler(client, cni),
-		UpdateHandler:        handlers.MakeUpdateHandler(client, cni, userSecretPath),
+			UpdateHandler:        handlers.MakeUpdateHandler(client, cni, userSecretPath, alwaysPull),
 			HealthHandler:        func(w http.ResponseWriter, r *http.Request) {},
 			InfoHandler:          handlers.MakeInfoHandler(Version, GitCommit),
 			ListNamespaceHandler: listNamespaces(),
 			SecretHandler:        handlers.MakeSecretHandler(client, userSecretPath),
-		LogHandler: func(w http.ResponseWriter, r *http.Request) {
+			LogHandler:           logs.NewLogHandlerFunc(faasdlogs.New(), config.ReadTimeout),
-			if r.Body != nil {
-				defer r.Body.Close()
-			}
-
-			w.WriteHeader(http.StatusNotImplemented)
-			w.Write([]byte(`Logs are not implemented for faasd`))
-		},
 		}
 
 		log.Printf("Listening on TCP port: %d\n", *config.TCPPort)
 		bootstrap.Serve(&bootstrapHandlers, config)
-
 		return nil
 	}
 
+	return command
+}
+
 func listNamespaces() func(w http.ResponseWriter, r *http.Request) {
 	return func(w http.ResponseWriter, r *http.Request) {
 		list := []string{""}

cmd/root.go

@@ -14,7 +14,7 @@ func init() {
 	rootCommand.AddCommand(versionCmd)
 	rootCommand.AddCommand(upCmd)
 	rootCommand.AddCommand(installCmd)
-	rootCommand.AddCommand(providerCmd)
+	rootCommand.AddCommand(makeProviderCmd())
 	rootCommand.AddCommand(collectCmd)
 }
 

cmd/up.go

@@ -14,8 +14,8 @@ import (
 
 	"github.com/pkg/errors"
 
-	"github.com/openfaas/faasd/pkg"
 	"github.com/alexellis/k3sup/pkg/env"
+	"github.com/openfaas/faasd/pkg"
 	"github.com/sethvargo/go-password/password"
 	"github.com/spf13/cobra"
 )
@@ -116,6 +116,7 @@ func runUp(_ *cobra.Command, _ []string) error {
 			log.Println(fileErr)
 			return
 		}
+
 		host := ""
 		lines := strings.Split(string(fileData), "\n")
 		for _, line := range lines {
@@ -172,7 +173,7 @@ func makeServiceDefinitions(archSuffix string) []pkg.Service {
 	wd, _ := os.Getwd()
 
 	return []pkg.Service{
-		pkg.Service{
+		{
 			Name:  "basic-auth-plugin",
 			Image: "docker.io/openfaas/basic-auth-plugin:0.18.10" + archSuffix,
 			Env: []string{
@@ -182,11 +183,11 @@ func makeServiceDefinitions(archSuffix string) []pkg.Service {
 				"pass_filename=basic-auth-password",
 			},
 			Mounts: []pkg.Mount{
-				pkg.Mount{
+				{
 					Src:  path.Join(path.Join(wd, "secrets"), "basic-auth-password"),
 					Dest: path.Join(containerSecretMountDir, "basic-auth-password"),
 				},
-				pkg.Mount{
+				{
 					Src:  path.Join(path.Join(wd, "secrets"), "basic-auth-user"),
 					Dest: path.Join(containerSecretMountDir, "basic-auth-user"),
 				},
@@ -194,26 +195,26 @@ func makeServiceDefinitions(archSuffix string) []pkg.Service {
 			Caps: []string{"CAP_NET_RAW"},
 			Args: nil,
 		},
-		pkg.Service{
+		{
 			Name:  "nats",
 			Env:   []string{""},
 			Image: "docker.io/library/nats-streaming:0.11.2",
 			Caps:  []string{},
 			Args:  []string{"/nats-streaming-server", "-m", "8222", "--store=memory", "--cluster_id=faas-cluster"},
 		},
-		pkg.Service{
+		{
 			Name:  "prometheus",
 			Env:   []string{},
 			Image: "docker.io/prom/prometheus:v2.14.0",
 			Mounts: []pkg.Mount{
-				pkg.Mount{
+				{
 					Src:  path.Join(wd, "prometheus.yml"),
 					Dest: "/etc/prometheus/prometheus.yml",
 				},
 			},
 			Caps: []string{"CAP_NET_RAW"},
 		},
-		pkg.Service{
+		{
 			Name: "gateway",
 			Env: []string{
 				"basic_auth=true",
@@ -231,18 +232,18 @@ func makeServiceDefinitions(archSuffix string) []pkg.Service {
 			},
 			Image: "docker.io/openfaas/gateway:0.18.8" + archSuffix,
 			Mounts: []pkg.Mount{
-				pkg.Mount{
+				{
 					Src:  path.Join(path.Join(wd, "secrets"), "basic-auth-password"),
 					Dest: path.Join(containerSecretMountDir, "basic-auth-password"),
 				},
-				pkg.Mount{
+				{
 					Src:  path.Join(path.Join(wd, "secrets"), "basic-auth-user"),
 					Dest: path.Join(containerSecretMountDir, "basic-auth-user"),
 				},
 			},
 			Caps: []string{"CAP_NET_RAW"},
 		},
-		pkg.Service{
+		{
 			Name: "queue-worker",
 			Env: []string{
 				"faas_nats_address=nats",
@@ -257,11 +258,11 @@ func makeServiceDefinitions(archSuffix string) []pkg.Service {
 			},
 			Image: "docker.io/openfaas/queue-worker:0.9.0",
 			Mounts: []pkg.Mount{
-				pkg.Mount{
+				{
 					Src:  path.Join(path.Join(wd, "secrets"), "basic-auth-password"),
 					Dest: path.Join(containerSecretMountDir, "basic-auth-password"),
 				},
-				pkg.Mount{
+				{
 					Src:  path.Join(path.Join(wd, "secrets"), "basic-auth-user"),
 					Dest: path.Join(containerSecretMountDir, "basic-auth-user"),
 				},

docs/DEV.md

@@ -174,17 +174,17 @@ make local
 
 ```sh
 # For x86_64
-sudo curl -fSLs "https://github.com/openfaas/faasd/releases/download/0.7.4/faasd" \
+sudo curl -fSLs "https://github.com/openfaas/faasd/releases/download/0.8.0/faasd" \
     -o "/usr/local/bin/faasd" \
     && sudo chmod a+x "/usr/local/bin/faasd"
 
 # armhf
-sudo curl -fSLs "https://github.com/openfaas/faasd/releases/download/0.7.4/faasd-armhf" \
+sudo curl -fSLs "https://github.com/openfaas/faasd/releases/download/0.8.0/faasd-armhf" \
     -o "/usr/local/bin/faasd" \
     && sudo chmod a+x "/usr/local/bin/faasd"
 
 # arm64
-sudo curl -fSLs "https://github.com/openfaas/faasd/releases/download/0.7.4/faasd-arm64" \
+sudo curl -fSLs "https://github.com/openfaas/faasd/releases/download/0.8.0/faasd-arm64" \
     -o "/usr/local/bin/faasd" \
     && sudo chmod a+x "/usr/local/bin/faasd"
 ```
@@ -225,6 +225,54 @@ To get the CLI for the command above run:
 curl -sSLf https://cli.openfaas.com | sudo sh
 ```
 
+#### Make a change to `faasd`
+
+There are two components you can hack on:
+
+For function CRUD you will work on `faasd provider` which is started from `cmd/provider.go`
+
+For faasd itself, you will work on the code from `faasd up`, which is started from `cmd/up.go`
+
+Before working on either, stop the systemd services:
+
+```
+sudo systemctl stop faasd &    # up command
+sudo systemctl stop faasd-provider   # provider command
+```
+
+Here is a workflow you can use for each code change:
+
+Enter the directory of the source code, and build a new binary:
+
+```bash
+cd $GOPATH/src/github.com/openfaas/faasd
+go build
+```
+
+Copy that binary to `/usr/local/bin/`
+
+```bash
+cp faasd /usr/local/bin/
+```
+
+To run `faasd up`, run it from its working directory as root
+
+```bash
+sudo -i
+cd /var/lib/faasd
+
+faasd up
+```
+
+Now to run `faasd provider`, run it from its working directory:
+
+```bash
+sudo -i
+cd /var/lib/faasd-provider
+
+faasd provider
+```
+
 #### At run-time
 
 Look in `hosts` in the current working folder or in `/var/lib/faasd/` to get the IP for the gateway or Prometheus

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,183 @@
+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)
+	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> \
+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",
+		"--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("want Name: %q, got %q", expectedEntry.Name, entry.Name)
+	}
+
+	if entry.Namespace != expectedEntry.Namespace {
+		t.Fatalf("want Namespace: %q, got %q", expectedEntry.Namespace, entry.Namespace)
+	}
+
+	if entry.Timestamp != expectedEntry.Timestamp {
+		t.Fatalf("want Timestamp: %q, got %q", expectedEntry.Timestamp, entry.Timestamp)
+	}
+
+	if entry.Text != expectedEntry.Text {
+		t.Fatalf("want Text: %q, got %q", 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 --identifier=spacetwo:loggyfunc --since=%s --follow --lines=5",
+		now.UTC().Format("2006-01-02 15:04:05"),
+	)
+
+	cmd := buildCmd(ctx, req).String()
+	wantCmd := "journalctl"
+	if !strings.Contains(cmd, wantCmd) {
+		t.Fatalf("cmd want: %q, got: %q", wantCmd, cmd)
+	}
+
+	if !strings.HasSuffix(cmd, expectedArgs) {
+		t.Fatalf("arg want: %q\ngot: %q", expectedArgs, cmd)
+	}
+}

pkg/provider/handlers/delete.go

@@ -8,12 +8,14 @@ import (
 	"log"
 	"net/http"
 
-	cninetwork "github.com/openfaas/faasd/pkg/cninetwork"
-	"github.com/openfaas/faasd/pkg/service"
 	"github.com/containerd/containerd"
 	"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"
 )
 
 func MakeDeleteHandler(client *containerd.Client, cni gocni.CNI) func(w http.ResponseWriter, r *http.Request) {
@@ -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,12 +18,13 @@ 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"
 )
 
-func MakeDeployHandler(client *containerd.Client, cni gocni.CNI, secretMountPath string) func(w http.ResponseWriter, r *http.Request) {
+func MakeDeployHandler(client *containerd.Client, cni gocni.CNI, secretMountPath string, alwaysPull bool) func(w http.ResponseWriter, r *http.Request) {
 
 	return func(w http.ResponseWriter, r *http.Request) {
 
@@ -52,9 +53,9 @@ 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)
+		deployErr := deploy(ctx, req, client, cni, secretMountPath, alwaysPull)
 		if deployErr != nil {
 			log.Printf("[Deploy] error deploying %s, error: %s\n", name, deployErr)
 			http.Error(w, deployErr.Error(), http.StatusBadRequest)
@@ -63,7 +64,7 @@ func MakeDeployHandler(client *containerd.Client, cni gocni.CNI, secretMountPath
 	}
 }
 
-func deploy(ctx context.Context, req types.FunctionDeployment, client *containerd.Client, cni gocni.CNI, secretMountPath string) error {
+func deploy(ctx context.Context, req types.FunctionDeployment, client *containerd.Client, cni gocni.CNI, secretMountPath string, alwaysPull bool) error {
 	r, err := reference.ParseNormalizedNamed(req.Image)
 	if err != nil {
 		return err
@@ -75,7 +76,7 @@ func deploy(ctx context.Context, req types.FunctionDeployment, client *container
 		snapshotter = val
 	}
 
-	image, err := service.PrepareImage(ctx, client, imgRef, snapshotter)
+	image, err := service.PrepareImage(ctx, client, imgRef, snapshotter, alwaysPull)
 	if err != nil {
 		return errors.Wrapf(err, "unable to pull image %s", imgRef)
 	}

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,31 +22,29 @@ 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)
 	for _, k := range containers {
 		name := k.ID()
-		functions[name], _ = GetFunction(client, name)
+		f, err := GetFunction(client, name)
+		if err != nil {
+			continue
+		}
+		functions[name] = f
 	}
 	return functions, nil
 }
 
 // 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 {
-
 		image, _ := c.Image(ctx)
 
 		containerName := c.ID()
@@ -55,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/secret.go

@@ -2,11 +2,13 @@ package handlers
 
 import (
 	"encoding/json"
+	"fmt"
 	"io/ioutil"
 	"log"
 	"net/http"
 	"os"
 	"path"
+	"strings"
 
 	"github.com/containerd/containerd"
 	"github.com/openfaas/faas-provider/types"
@@ -76,17 +78,6 @@ func createSecret(c *containerd.Client, w http.ResponseWriter, r *http.Request,
 	}
 }
 
-func parseSecret(r *http.Request) (types.Secret, error) {
-	secret := types.Secret{}
-	bytesOut, err := ioutil.ReadAll(r.Body)
-	if err != nil {
-		return secret, err
-	}
-
-	err = json.Unmarshal(bytesOut, &secret)
-	return secret, err
-}
-
 func deleteSecret(c *containerd.Client, w http.ResponseWriter, r *http.Request, mountPath string) {
 	secret, err := parseSecret(r)
 	if err != nil {
@@ -103,3 +94,29 @@ func deleteSecret(c *containerd.Client, w http.ResponseWriter, r *http.Request,
 		return
 	}
 }
+
+func parseSecret(r *http.Request) (types.Secret, error) {
+	secret := types.Secret{}
+	bytesOut, err := ioutil.ReadAll(r.Body)
+	if err != nil {
+		return secret, err
+	}
+
+	err = json.Unmarshal(bytesOut, &secret)
+	if err != nil {
+		return secret, err
+	}
+
+	if isTraversal(secret.Name) {
+		return secret, fmt.Errorf(traverseErrorSt)
+	}
+
+	return secret, err
+}
+
+const traverseErrorSt = "directory traversal found in name"
+
+func isTraversal(name string) bool {
+	return strings.Contains(name, fmt.Sprintf("%s", string(os.PathSeparator))) ||
+		strings.Contains(name, "..")
+}

pkg/provider/handlers/secret_test.go

@@ -0,0 +1,63 @@
+package handlers
+
+import (
+	"bytes"
+	"encoding/json"
+	"net/http"
+	"net/http/httptest"
+	"testing"
+
+	"github.com/openfaas/faas-provider/types"
+)
+
+func Test_parseSecretValidName(t *testing.T) {
+
+	s := types.Secret{Name: "authorized_keys"}
+	body, _ := json.Marshal(s)
+	reader := bytes.NewReader(body)
+	r := httptest.NewRequest(http.MethodPost, "/", reader)
+	_, err := parseSecret(r)
+
+	if err != nil {
+		t.Fatalf("secret name is valid with no traversal characters")
+	}
+}
+
+func Test_parseSecretValidNameWithDot(t *testing.T) {
+
+	s := types.Secret{Name: "authorized.keys"}
+	body, _ := json.Marshal(s)
+	reader := bytes.NewReader(body)
+	r := httptest.NewRequest(http.MethodPost, "/", reader)
+	_, err := parseSecret(r)
+
+	if err != nil {
+		t.Fatalf("secret name is valid with no traversal characters")
+	}
+}
+
+func Test_parseSecretWithTraversalWithSlash(t *testing.T) {
+
+	s := types.Secret{Name: "/root/.ssh/authorized_keys"}
+	body, _ := json.Marshal(s)
+	reader := bytes.NewReader(body)
+	r := httptest.NewRequest(http.MethodPost, "/", reader)
+	_, err := parseSecret(r)
+
+	if err == nil {
+		t.Fatalf("secret name should fail due to path traversal")
+	}
+}
+
+func Test_parseSecretWithTraversalWithDoubleDot(t *testing.T) {
+
+	s := types.Secret{Name: ".."}
+	body, _ := json.Marshal(s)
+	reader := bytes.NewReader(body)
+	r := httptest.NewRequest(http.MethodPost, "/", reader)
+	_, err := parseSecret(r)
+
+	if err == nil {
+		t.Fatalf("secret name should fail due to path traversal")
+	}
+}

pkg/provider/handlers/update.go

@@ -8,15 +8,17 @@ import (
 	"log"
 	"net/http"
 
-	"github.com/openfaas/faasd/pkg/cninetwork"
-	"github.com/openfaas/faasd/pkg/service"
 	"github.com/containerd/containerd"
 	"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"
 )
 
-func MakeUpdateHandler(client *containerd.Client, cni gocni.CNI, secretMountPath string) func(w http.ResponseWriter, r *http.Request) {
+func MakeUpdateHandler(client *containerd.Client, cni gocni.CNI, secretMountPath string, alwaysPull bool) func(w http.ResponseWriter, r *http.Request) {
 
 	return func(w http.ResponseWriter, r *http.Request) {
 
@@ -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 {
@@ -68,7 +70,7 @@ func MakeUpdateHandler(client *containerd.Client, cni gocni.CNI, secretMountPath
 			return
 		}
 
-		deployErr := deploy(ctx, req, client, cni, secretMountPath)
+		deployErr := deploy(ctx, req, client, cni, secretMountPath, alwaysPull)
 		if deployErr != nil {
 			log.Printf("[Update] error deploying %s, error: %s\n", name, deployErr)
 			http.Error(w, deployErr.Error(), http.StatusBadRequest)

pkg/service/service.go

@@ -122,11 +122,12 @@ func getResolver(ctx context.Context, configFile *configfile.ConfigFile) (remote
 	return docker.NewResolver(opts), nil
 }
 
-func PrepareImage(ctx context.Context, client *containerd.Client, imageName, snapshotter string) (containerd.Image, error) {
+func PrepareImage(ctx context.Context, client *containerd.Client, imageName, snapshotter string, pullAlways bool) (containerd.Image, error) {
 	var (
 		empty    containerd.Image
 		resolver remotes.Resolver
 	)
+
 	if _, stErr := os.Stat(filepath.Join(dockerConfigDir, config.ConfigFileName)); stErr == nil {
 		configFile, err := config.Load(dockerConfigDir)
 		if err != nil {
@@ -140,22 +141,29 @@ func PrepareImage(ctx context.Context, client *containerd.Client, imageName, sna
 		return empty, stErr
 	}
 
-	image, err := client.GetImage(ctx, imageName)
+	var image containerd.Image
+	if pullAlways {
+		img, err := pullImage(ctx, client, resolver, imageName)
+		if err != nil {
+			return empty, err
+		}
+
+		image = img
+	} else {
+
+		img, err := client.GetImage(ctx, imageName)
 		if err != nil {
 			if !errdefs.IsNotFound(err) {
 				return empty, err
 			}
-		rOpts := []containerd.RemoteOpt{
+			img, err := pullImage(ctx, client, resolver, imageName)
-			containerd.WithPullUnpack,
-		}
-		if resolver != nil {
-			rOpts = append(rOpts, containerd.WithResolver(resolver))
-		}
-		img, err := client.Pull(ctx, imageName, rOpts...)
 			if err != nil {
-			return empty, fmt.Errorf("cannot pull: %s", err)
+				return empty, err
 			}
 			image = img
+		} else {
+			image = img
+		}
 	}
 
 	unpacked, err := image.IsUnpacked(ctx, snapshotter)
@@ -171,3 +179,21 @@ func PrepareImage(ctx context.Context, client *containerd.Client, imageName, sna
 
 	return image, nil
 }
+
+func pullImage(ctx context.Context, client *containerd.Client, resolver remotes.Resolver, imageName string) (containerd.Image, error) {
+
+	var empty containerd.Image
+
+	rOpts := []containerd.RemoteOpt{
+		containerd.WithPullUnpack,
+	}
+	if resolver != nil {
+		rOpts = append(rOpts, containerd.WithResolver(resolver))
+	}
+	img, err := client.Pull(ctx, imageName, rOpts...)
+	if err != nil {
+		return empty, fmt.Errorf("cannot pull: %s", err)
+	}
+
+	return img, nil
+}

pkg/supervisor.go

@@ -8,13 +8,13 @@ import (
 	"os"
 	"path"
 
-	"github.com/openfaas/faasd/pkg/cninetwork"
-	"github.com/openfaas/faasd/pkg/service"
 	"github.com/containerd/containerd"
 	"github.com/containerd/containerd/cio"
 	"github.com/containerd/containerd/containers"
 	"github.com/containerd/containerd/oci"
 	gocni "github.com/containerd/go-cni"
+	"github.com/openfaas/faasd/pkg/cninetwork"
+	"github.com/openfaas/faasd/pkg/service"
 
 	"github.com/containerd/containerd/namespaces"
 	"github.com/opencontainers/runtime-spec/specs-go"
@@ -25,6 +25,7 @@ const (
 	workingDirectoryPermission = 0644
 	// faasdNamespace is the containerd namespace services are created
 	faasdNamespace         = "default"
+	faasServicesPullAlways = false
 )
 
 type Service struct {
@@ -88,7 +89,7 @@ func (s *Supervisor) Start(svcs []Service) error {
 	for _, svc := range svcs {
 		fmt.Printf("Preparing: %s with image: %s\n", svc.Name, svc.Image)
 
-		img, err := service.PrepareImage(ctx, s.client, svc.Image, defaultSnapshotter)
+		img, err := service.PrepareImage(ctx, s.client, svc.Image, defaultSnapshotter, faasServicesPullAlways)
 		if err != nil {
 			return err
 		}

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,
+	)
+}