Fix for proxy exiting early

When a proxied core service was accessed before it was ready
to accept a connection, due to a start-up, restart, etc of
a core service, then the proxy exited instead of continuing
to accept new connections.

This meant having to restart faasd and hope the race condition
worked itself out, or that no incoming requests were made.

Tested with Grafana, which seemed to manifest the issue
the most.

Signed-off-by: Alex Ellis (OpenFaaS Ltd) <alexellis2@gmail.com>

LICENSE

@@ -1,6 +1,6 @@
-For usage of faasd, see: EULA.md
+License for faasd contributions from OpenFaaS Ltd - 2017, 2029-2024, see: EULA.md
 
-See below for source code contributions
+Only third-party contributions to source code are licensed MIT:
 
 MIT License
 

README.md

@@ -121,7 +121,42 @@ View sample pages, reviews and testimonials on Gumroad:
 
 ["Serverless For Everyone Else"](https://openfaas.gumroad.com/l/serverless-for-everyone-else)
 
-### Deploy faasd
+### Deploy OpenFaaS Edge (commercial distribution of faasd)
+
+OpenFaaS Edge is a commercial distribution of faasd, with enhancements and additional features from OpenFaaS Pro. The [OpenFaaS Pro EULA applies](https://github.com/openfaas/faas/blob/master/pro/EULA.md).
+
+* Upgraded Pro components from OpenFaaS Standard: Gateway, Cron Connector, JetStream Queue Worker and Classic Scale to Zero
+* Deploy up to 250 functions per installation
+* Configure private DNS servers
+* Airgap-friendly with installation bundled in an OCI image
+* Detailed RAM/CPU metrics for stateful containers, and functions, including Out Of Memory (OOM) events
+* Multiple namespace support
+
+This version is intended for resale as part of a wider solution, and to be deployed both into industrial and on-premises environments.
+
+Individual [GitHub Sponsors of OpenFaaS](https://github.com/sponsors/openfaas) (25 USD / mo and higher) can use OpenFaaS Edge for personal use.
+
+You can install OpenFaaS Edge with the following script:
+
+```bash
+curl -sLSf \
+    https://raw.githubusercontent.com/openfaas/faasd/refs/heads/master/hack/install-edge.sh \
+    -o install-edge.sh && \
+chmod +x install-edge.sh
+sudo -E ./install-edge.sh
+```
+
+*For an offline installation*
+
+Copy the OCI bundle and the install-edge.sh script to the remote server.
+
+Then mirror the various images from docker-compose.yaml into your private registry, and update the references from i.e. `image: ghcr.io/openfaasltd/gateway` to the equivalents in your registry.
+
+If your system is unable to install apt, yum, or pacman packages, due to limited network access, then set the `SKIP_OS` environment to 1. The list of packages is available in the `install_required_packages` section of the script.
+
+### Deploy faasd CE
+
+faasd-ce supports 15 functions and needs a computer with a stable Internet connection to run. There are some restrictions on commercial use, but [individuals and certain small businesses can use it for free](EULA.md).
 
 The easiest way to deploy faasd is with cloud-init, we give several examples below, and post IaaS platforms will accept "user-data" pasted into their UI, or via their API.
 

docker-compose.yaml

@@ -21,7 +21,7 @@ services:
     #    - "127.0.0.1:8222:8222"
 
   prometheus:
-    image: docker.io/prom/prometheus:v2.49.1
+    image: docker.io/prom/prometheus:v3.1.0
 # nobody
     user: "65534"
     volumes:
@@ -39,7 +39,7 @@ services:
        - "127.0.0.1:9090:9090"
 
   gateway:
-    image: ghcr.io/openfaas/gateway:0.27.9
+    image: ghcr.io/openfaas/gateway:0.27.12
     environment:
       - basic_auth=true
       - functions_provider_url=http://faasd-provider:8081/
@@ -69,7 +69,7 @@ services:
        - "8080:8080"
 
   queue-worker:
-    image: ghcr.io/openfaas/queue-worker:0.14.1
+    image: ghcr.io/openfaas/queue-worker:0.14.2
     environment:
       - faas_nats_address=nats
       - faas_nats_port=4222

hack/faasd-provider.service

@@ -2,7 +2,7 @@
 Description=faasd-provider
 
 [Service]
-MemoryLimit=500M
+MemoryMax=500M
 Environment="secret_mount_path={{.SecretMountPath}}"
 Environment="basic_auth=true"
 Environment="hosts_dir=/var/lib/faasd"

hack/faasd.service

@@ -3,7 +3,7 @@ Description=faasd
 After=faasd-provider.service
 
 [Service]
-MemoryLimit=500M
+MemoryMax=500M
 ExecStart=/usr/local/bin/faasd up
 Restart=on-failure
 RestartSec=10s

hack/install-edge.sh

@@ -0,0 +1,99 @@
+#!/bin/bash
+
+set -e # stop on error
+set -o pipefail
+
+if [ "$EUID" -ne 0 ]; then
+    echo "Please run as root or with sudo"
+    exit
+fi
+
+has_yum() {
+  [ -n "$(command -v yum)" ]
+}
+
+has_apt_get() {
+  [ -n "$(command -v apt-get)" ]
+}
+
+has_pacman() {
+  [ -n "$(command -v pacman)" ]
+}
+
+install_required_packages() {
+  if $(has_apt_get); then
+    # Debian bullseye is missing iptables. Added to required packages
+    # to get it working in raspberry pi. No such known issues in
+    # other distros. Hence, adding only to this block.
+    # reference: https://github.com/openfaas/faasd/pull/237
+    apt-get update -y
+    apt-get install -y curl runc bridge-utils iptables
+  elif $(has_yum); then
+    yum check-update -y
+    yum install -y curl runc iptables-services which
+  elif $(has_pacman); then
+    pacman -Syy
+    pacman -Sy curl runc bridge-utils
+  else
+    fatal "Could not find apt-get, yum, or pacman. Cannot install dependencies on this OS."
+    exit 1
+  fi
+}
+
+echo "OpenFaaS Edge (based upon faasd and OpenFaaS Standard)"
+echo ""
+echo ""
+
+echo "1. Installing required OS packages, set SKIP_OS=1 to skip this step"
+echo ""
+
+if [ -z "$SKIP_OS" ]; then
+    install_required_packages
+fi
+
+echo "2. Downloading OCI image, and installing pre-requisites"
+echo ""
+if [ ! -x "$(command -v arkade)" ]; then
+    # For Centos, RHEL, Fedora, Amazon Linux, and Oracle Linux, use BINLOCATION=/usr/bin/
+
+    if $(has_yum); then
+      BINLOCATION=/usr/bin/
+    fi
+
+    curl -sLS https://get.arkade.dev | BINLOCATION=${BINLOCATION} sh
+fi
+
+PATH=$PATH:$HOME/.arkade/bin
+
+tmpdir=$(mktemp -d)
+
+# Ensure all existing services are stopped when installing over an 
+# existing faasd installation
+systemctl stop faasd || :
+systemctl stop faasd-provider || :
+systemctl stop containerd || :
+killall -9 containerd-shim-runc-v2 || :
+killall -9 faasd || :
+
+# crane, or docker can also be used to download the OCI image and to extract it
+
+# Rather than the :latest tag, a specific tag can be given
+# Use "crane ls ghcr.io/openfaasltd/faasd-pro" to see available tags
+
+${BINLOCATION}arkade oci install --path ${tmpdir} \
+  ghcr.io/openfaasltd/faasd-pro:latest
+
+cd ${tmpdir}
+./install.sh ./
+
+echo ""
+echo "3. You now need to activate your license via GitHub"
+echo ""
+echo "sudo -E faasd github login"
+echo "sudo -E faasd activate"
+echo ""
+echo ""
+echo "4. Then perform the final installation steps"
+echo ""
+echo "sudo -E sh -c \"cd ${tmpdir}/var/lib/faasd && faasd install\""
+echo ""

pkg/logs/requestor.go

@@ -58,10 +58,10 @@ func (r *requester) Query(ctx context.Context, req logs.Request) (<-chan logs.Me
 
 // buildCmd reeturns the equivalent of
 //
-// 	journalctl -t <namespace>:<name>  \
-// 		--output=json \
-// 		--since=<timestamp> \
-// 		<--follow> \
+//	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)
@@ -105,12 +105,12 @@ func streamLogs(ctx context.Context, cmd *exec.Cmd, out io.ReadCloser, msgs chan
 
 	// will ensure `out` is closed and all related resources cleaned up
 	go func() {
-		err := cmd.Wait()
-		log.Println("wait result", err)
+		if err := cmd.Wait(); err != nil {
+			log.Printf("journalctl exited with error: %s", err)
+		}
 	}()
 
 	defer func() {
-		log.Println("closing journal stream")
 		close(msgs)
 	}()
 
@@ -176,7 +176,6 @@ func parseEntry(entry map[string]string) (logs.Message, error) {
 }
 
 func logErrOut(out io.ReadCloser) {
-	defer log.Println("stderr closed")
 	defer out.Close()
 
 	io.Copy(log.Writer(), out)

pkg/provider/handlers/function_list.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"errors"
 	"log"
+	"strings"
 	"time"
 
 	"github.com/containerd/containerd"
@@ -51,7 +52,10 @@ func ListFunctions(client *containerd.Client, namespace string) (map[string]*Fun
 		name := c.ID()
 		f, err := GetFunction(client, name, namespace)
 		if err != nil {
-			log.Printf("skipping %s, error: %s", name, err)
+			if !strings.Contains(err.Error(), "unable to get IP address for container") {
+				log.Printf("List functions, skipping: %s, error: %s", name, err)
+			}
+
 		} else {
 			functions[name] = &f
 		}

pkg/provider/handlers/info.go

@@ -12,7 +12,7 @@ const (
 	OrchestrationIdentifier = "containerd"
 
 	// ProviderName name of the provider
-	ProviderName = "faasd"
+	ProviderName = "faasd-ce"
 )
 
 // MakeInfoHandler creates handler for /system/info endpoint

pkg/provider/handlers/scale.go

@@ -96,32 +96,24 @@ func MakeReplicaUpdateHandler(client *containerd.Client, cni gocni.CNI) func(w h
 
 		createNewTask := false
 
-		// Scale to zero
 		if req.Replicas == 0 {
-			// If a task is running, pause it
-			if taskExists && taskStatus.Status == containerd.Running {
-				if pauseErr := task.Pause(ctx); pauseErr != nil {
-					wrappedPauseErr := fmt.Errorf("error pausing task %s, error: %s", name, pauseErr)
-					log.Printf("[Scale] %s\n", wrappedPauseErr.Error())
-					http.Error(w, wrappedPauseErr.Error(), http.StatusNotFound)
-					return
-				}
-			}
+			http.Error(w, "replicas must > 0 for faasd CE", http.StatusBadRequest)
+			return
 		}
 
 		if taskExists {
 			if taskStatus != nil {
 				if taskStatus.Status == containerd.Paused {
-					if resumeErr := task.Resume(ctx); resumeErr != nil {
-						log.Printf("[Scale] error resuming task %s, error: %s\n", name, resumeErr)
-						http.Error(w, resumeErr.Error(), http.StatusBadRequest)
+					if _, err := task.Delete(ctx); err != nil {
+						log.Printf("[Scale] error deleting paused task %s, error: %s\n", name, err)
+						http.Error(w, err.Error(), http.StatusBadRequest)
 						return
 					}
 				} else if taskStatus.Status == containerd.Stopped {
 					// Stopped tasks cannot be restarted, must be removed, and created again
-					if _, delErr := task.Delete(ctx); delErr != nil {
-						log.Printf("[Scale] error deleting stopped task %s, error: %s\n", name, delErr)
-						http.Error(w, delErr.Error(), http.StatusBadRequest)
+					if _, err := task.Delete(ctx); err != nil {
+						log.Printf("[Scale] error deleting stopped task %s, error: %s\n", name, err)
+						http.Error(w, err.Error(), http.StatusBadRequest)
 						return
 					}
 					createNewTask = true

pkg/proxy.go

@@ -86,7 +86,7 @@ func (p *Proxy) Start() error {
 			conn.Close()
 
 			log.Printf("Unable to dial: %s, error: %s", upstreamAddr, err.Error())
-			return err
+			continue
 		}
 
 		go pipe(conn, upstream)

pkg/service/service.go

@@ -6,6 +6,7 @@ import (
 	"log"
 	"os"
 	"path/filepath"
+	"strings"
 	"sync"
 	"time"
 
@@ -45,10 +46,24 @@ func Remove(ctx context.Context, client *containerd.Client, name string) error {
 				log.Printf("Status of %s is: %s\n", name, status.Status)
 			}
 
-			log.Printf("Need to kill task: %s\n", name)
-			if err = killTask(ctx, t); err != nil {
+			var gracePeriod = time.Second * 30
+			spec, err := t.Spec(ctx)
+			if err == nil {
+				for _, p := range spec.Process.Env {
+					k, v, ok := strings.Cut(p, "=")
+					if ok && k == "grace_period" {
+						periodVal, err := time.ParseDuration(v)
+						if err == nil {
+							gracePeriod = periodVal
+						}
+					}
+				}
+			}
+
+			if err = killTask(ctx, t, gracePeriod); err != nil {
 				return fmt.Errorf("error killing task %s, %s, %w", container.ID(), name, err)
 			}
+
 		}
 
 		if err := container.Delete(ctx, containerd.WithSnapshotCleanup); err != nil {
@@ -66,14 +81,13 @@ func Remove(ctx context.Context, client *containerd.Client, name string) error {
 }
 
 // Adapted from Stellar - https://github.com/stellar
-func killTask(ctx context.Context, task containerd.Task) error {
-
-	killTimeout := 30 * time.Second
+func killTask(ctx context.Context, task containerd.Task, gracePeriod time.Duration) error {
 
 	wg := &sync.WaitGroup{}
 	wg.Add(1)
 	var err error
 
+	waited := false
 	go func() {
 		defer wg.Done()
 		if task != nil {
@@ -89,22 +103,39 @@ func killTask(ctx context.Context, task containerd.Task) error {
 
 			select {
 			case <-wait:
-				task.Delete(ctx)
+				waited = true
 				return
-			case <-time.After(killTimeout):
+			case <-time.After(gracePeriod):
+				log.Printf("Sending SIGKILL to: %s after: %s", task.ID(), gracePeriod.Round(time.Second).String())
 				if err := task.Kill(ctx, unix.SIGKILL, containerd.WithKillAll); err != nil {
-					log.Printf("error force killing container task: %s", err)
+					log.Printf("error sending SIGKILL to task: %s", err)
 				}
+
 				return
 			}
 		}
 	}()
 	wg.Wait()
 
+	if task != nil {
+		if !waited {
+			wait, err := task.Wait(ctx)
+			if err != nil {
+				log.Printf("error waiting on task after kill: %s", err)
+			}
+
+			<-wait
+		}
+
+		if _, err := task.Delete(ctx); err != nil {
+			return err
+		}
+	}
+
 	return err
 }
 
-func getResolver(ctx context.Context, configFile *configfile.ConfigFile) (remotes.Resolver, error) {
+func getResolver(configFile *configfile.ConfigFile) (remotes.Resolver, error) {
 	// credsFunc is based on https://github.com/moby/buildkit/blob/0b130cca040246d2ddf55117eeff34f546417e40/session/auth/authprovider/authprovider.go#L35
 	credFunc := func(host string) (string, string, error) {
 		if host == "registry-1.docker.io" {
@@ -139,7 +170,7 @@ func PrepareImage(ctx context.Context, client *containerd.Client, imageName, sna
 		if err != nil {
 			return nil, err
 		}
-		resolver, err = getResolver(ctx, configFile)
+		resolver, err = getResolver(configFile)
 		if err != nil {
 			return empty, err
 		}