From 2d3c904781f128183e55de94db79fa0d10818302 Mon Sep 17 00:00:00 2001 From: "Han Verstraete (OpenFaaS Ltd)" Date: Wed, 2 Sep 2026 13:39:02 +0200 Subject: [PATCH] Add OpenFaaS function pattern examples in Go, Python, and Node.js Signed-off-by: Han Verstraete (OpenFaaS Ltd) --- .gitignore | 10 + README.md | 36 +++ go/director/.gitignore | 3 + go/director/README.md | 56 +++++ go/director/battery-check/go.mod | 3 + go/director/battery-check/handler.go | 35 +++ go/director/battery-check/handler_test.go | 44 ++++ go/director/stack.yaml | 29 +++ go/director/telemetry-workflow/go.mod | 3 + go/director/telemetry-workflow/handler.go | 226 ++++++++++++++++++ .../telemetry-workflow/handler_test.go | 176 ++++++++++++++ go/director/temperature-check/go.mod | 3 + go/director/temperature-check/handler.go | 35 +++ go/director/temperature-check/handler_test.go | 44 ++++ go/director/validate-reading/go.mod | 3 + go/director/validate-reading/handler.go | 48 ++++ go/director/validate-reading/handler_test.go | 41 ++++ go/fan-out/.gitignore | 3 + go/fan-out/README.md | 66 +++++ go/fan-out/fan-out/go.mod | 3 + go/fan-out/fan-out/handler.go | 160 +++++++++++++ go/fan-out/fan-out/handler_test.go | 98 ++++++++ go/fan-out/stack.yaml | 16 ++ go/fan-out/url-check/go.mod | 3 + go/fan-out/url-check/handler.go | 116 +++++++++ go/fan-out/url-check/handler_test.go | 83 +++++++ go/singleton/.gitignore | 3 + go/singleton/README.md | 41 ++++ go/singleton/notification-hub/go.mod | 3 + go/singleton/notification-hub/handler.go | 103 ++++++++ go/singleton/notification-hub/handler_test.go | 45 ++++ go/singleton/stack.yaml | 16 ++ node/director/README.md | 19 ++ node/director/battery-check/handler.js | 40 ++++ node/director/battery-check/handler.test.js | 35 +++ node/director/battery-check/package.json | 8 + node/director/stack.yaml | 29 +++ node/director/telemetry-workflow/handler.js | 125 ++++++++++ .../telemetry-workflow/handler.test.js | 112 +++++++++ node/director/telemetry-workflow/package.json | 8 + node/director/temperature-check/handler.js | 40 ++++ .../temperature-check/handler.test.js | 35 +++ node/director/temperature-check/package.json | 8 + node/director/validate-reading/handler.js | 57 +++++ .../director/validate-reading/handler.test.js | 63 +++++ node/director/validate-reading/package.json | 8 + node/fan-out/README.md | 12 + node/fan-out/fan-out/handler.js | 101 ++++++++ node/fan-out/fan-out/handler.test.js | 85 +++++++ node/fan-out/fan-out/package.json | 8 + node/fan-out/stack.yaml | 16 ++ node/fan-out/url-check/handler.js | 84 +++++++ node/fan-out/url-check/handler.test.js | 74 ++++++ node/fan-out/url-check/package.json | 8 + python/director/README.md | 19 ++ python/director/battery-check/__init__.py | 0 python/director/battery-check/handler.py | 21 ++ python/director/battery-check/handler_test.py | 16 ++ .../director/battery-check/requirements.txt | 1 + python/director/battery-check/tox.ini | 22 ++ python/director/stack.yaml | 37 +++ .../director/telemetry-workflow/__init__.py | 0 python/director/telemetry-workflow/handler.py | 87 +++++++ .../telemetry-workflow/handler_test.py | 77 ++++++ .../telemetry-workflow/requirements.txt | 1 + python/director/telemetry-workflow/tox.ini | 22 ++ python/director/temperature-check/__init__.py | 0 python/director/temperature-check/handler.py | 21 ++ .../temperature-check/handler_test.py | 16 ++ .../temperature-check/requirements.txt | 1 + python/director/temperature-check/tox.ini | 22 ++ python/director/validate-reading/__init__.py | 0 python/director/validate-reading/handler.py | 41 ++++ .../director/validate-reading/handler_test.py | 36 +++ .../validate-reading/requirements.txt | 1 + python/director/validate-reading/tox.ini | 22 ++ python/fan-out/README.md | 12 + python/fan-out/fan-out/__init__.py | 0 python/fan-out/fan-out/handler.py | 73 ++++++ python/fan-out/fan-out/handler_test.py | 55 +++++ python/fan-out/fan-out/requirements.txt | 1 + python/fan-out/fan-out/tox.ini | 22 ++ python/fan-out/stack.yaml | 20 ++ python/fan-out/url-check/__init__.py | 0 python/fan-out/url-check/handler.py | 61 +++++ python/fan-out/url-check/handler_test.py | 63 +++++ python/fan-out/url-check/requirements.txt | 1 + python/fan-out/url-check/tox.ini | 22 ++ python/singleton/README.md | 12 + python/singleton/notification-hub/__init__.py | 0 python/singleton/notification-hub/handler.py | 73 ++++++ .../notification-hub/handler_test.py | 63 +++++ .../notification-hub/requirements.txt | 1 + python/singleton/notification-hub/tox.ini | 22 ++ python/singleton/stack.yaml | 18 ++ 95 files changed, 3511 insertions(+) create mode 100644 .gitignore create mode 100644 go/director/.gitignore create mode 100644 go/director/README.md create mode 100644 go/director/battery-check/go.mod create mode 100644 go/director/battery-check/handler.go create mode 100644 go/director/battery-check/handler_test.go create mode 100644 go/director/stack.yaml create mode 100644 go/director/telemetry-workflow/go.mod create mode 100644 go/director/telemetry-workflow/handler.go create mode 100644 go/director/telemetry-workflow/handler_test.go create mode 100644 go/director/temperature-check/go.mod create mode 100644 go/director/temperature-check/handler.go create mode 100644 go/director/temperature-check/handler_test.go create mode 100644 go/director/validate-reading/go.mod create mode 100644 go/director/validate-reading/handler.go create mode 100644 go/director/validate-reading/handler_test.go create mode 100644 go/fan-out/.gitignore create mode 100644 go/fan-out/README.md create mode 100644 go/fan-out/fan-out/go.mod create mode 100644 go/fan-out/fan-out/handler.go create mode 100644 go/fan-out/fan-out/handler_test.go create mode 100644 go/fan-out/stack.yaml create mode 100644 go/fan-out/url-check/go.mod create mode 100644 go/fan-out/url-check/handler.go create mode 100644 go/fan-out/url-check/handler_test.go create mode 100644 go/singleton/.gitignore create mode 100644 go/singleton/README.md create mode 100644 go/singleton/notification-hub/go.mod create mode 100644 go/singleton/notification-hub/handler.go create mode 100644 go/singleton/notification-hub/handler_test.go create mode 100644 go/singleton/stack.yaml create mode 100644 node/director/README.md create mode 100644 node/director/battery-check/handler.js create mode 100644 node/director/battery-check/handler.test.js create mode 100644 node/director/battery-check/package.json create mode 100644 node/director/stack.yaml create mode 100644 node/director/telemetry-workflow/handler.js create mode 100644 node/director/telemetry-workflow/handler.test.js create mode 100644 node/director/telemetry-workflow/package.json create mode 100644 node/director/temperature-check/handler.js create mode 100644 node/director/temperature-check/handler.test.js create mode 100644 node/director/temperature-check/package.json create mode 100644 node/director/validate-reading/handler.js create mode 100644 node/director/validate-reading/handler.test.js create mode 100644 node/director/validate-reading/package.json create mode 100644 node/fan-out/README.md create mode 100644 node/fan-out/fan-out/handler.js create mode 100644 node/fan-out/fan-out/handler.test.js create mode 100644 node/fan-out/fan-out/package.json create mode 100644 node/fan-out/stack.yaml create mode 100644 node/fan-out/url-check/handler.js create mode 100644 node/fan-out/url-check/handler.test.js create mode 100644 node/fan-out/url-check/package.json create mode 100644 python/director/README.md create mode 100644 python/director/battery-check/__init__.py create mode 100644 python/director/battery-check/handler.py create mode 100644 python/director/battery-check/handler_test.py create mode 100644 python/director/battery-check/requirements.txt create mode 100644 python/director/battery-check/tox.ini create mode 100644 python/director/stack.yaml create mode 100644 python/director/telemetry-workflow/__init__.py create mode 100644 python/director/telemetry-workflow/handler.py create mode 100644 python/director/telemetry-workflow/handler_test.py create mode 100644 python/director/telemetry-workflow/requirements.txt create mode 100644 python/director/telemetry-workflow/tox.ini create mode 100644 python/director/temperature-check/__init__.py create mode 100644 python/director/temperature-check/handler.py create mode 100644 python/director/temperature-check/handler_test.py create mode 100644 python/director/temperature-check/requirements.txt create mode 100644 python/director/temperature-check/tox.ini create mode 100644 python/director/validate-reading/__init__.py create mode 100644 python/director/validate-reading/handler.py create mode 100644 python/director/validate-reading/handler_test.py create mode 100644 python/director/validate-reading/requirements.txt create mode 100644 python/director/validate-reading/tox.ini create mode 100644 python/fan-out/README.md create mode 100644 python/fan-out/fan-out/__init__.py create mode 100644 python/fan-out/fan-out/handler.py create mode 100644 python/fan-out/fan-out/handler_test.py create mode 100644 python/fan-out/fan-out/requirements.txt create mode 100644 python/fan-out/fan-out/tox.ini create mode 100644 python/fan-out/stack.yaml create mode 100644 python/fan-out/url-check/__init__.py create mode 100644 python/fan-out/url-check/handler.py create mode 100644 python/fan-out/url-check/handler_test.py create mode 100644 python/fan-out/url-check/requirements.txt create mode 100644 python/fan-out/url-check/tox.ini create mode 100644 python/singleton/README.md create mode 100644 python/singleton/notification-hub/__init__.py create mode 100644 python/singleton/notification-hub/handler.py create mode 100644 python/singleton/notification-hub/handler_test.py create mode 100644 python/singleton/notification-hub/requirements.txt create mode 100644 python/singleton/notification-hub/tox.ini create mode 100644 python/singleton/stack.yaml diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..7babaf6 --- /dev/null +++ b/.gitignore @@ -0,0 +1,10 @@ +template/ +build/ +context/ +.secrets/ +*.tar.gz +__pycache__/ +.pytest_cache/ +.tox/ +*.py[cod] +node_modules/ diff --git a/README.md b/README.md index f9e8db1..6c40ba0 100644 --- a/README.md +++ b/README.md @@ -1,2 +1,38 @@ # function-patterns Patterns for OpenFaaS Functions + +Examples for the OpenFaaS documentation. Each example is +self-contained: it has its own `stack.yaml` and function directories, so it +can be built and deployed on its own. + +## Examples + +- [**Director**](https://docs.openfaas.com/languages/patterns/director/) — [Go](go/director/) · [Python](python/director/) · [Node.js](node/director/) +- [**Fan-out**](https://docs.openfaas.com/languages/patterns/fan-out/) — [Go](go/fan-out/) · [Python](python/fan-out/) · [Node.js](node/fan-out/) +- [**Singleton**](https://docs.openfaas.com/languages/patterns/singleton/) — [Go](go/singleton/) · [Python](python/singleton/) + +## Prerequisites + +- [faas-cli](https://docs.openfaas.com/cli/install/) +- Docker, for building images +- A running OpenFaaS cluster, for e2e testing — set `OPENFAAS_URL` and + provide credentials, or point the `gateway` in each `stack.yaml` at a + local gateway + +## Build and test an example + +```bash +cd go/director +faas-cli up --tag=digest +``` + +`faas-cli up` builds each function image, pushes it to the registry, and +deploys it to the gateway. + +The examples default to the public [ttl.sh](https://ttl.sh) registry so they +can be pushed without credentials. The registry and owner are overridable +through environment variables: + +```bash +REGISTRY=ghcr.io OWNER=welteki faas-cli up +``` diff --git a/go/director/.gitignore b/go/director/.gitignore new file mode 100644 index 0000000..4f03565 --- /dev/null +++ b/go/director/.gitignore @@ -0,0 +1,3 @@ +template +build +.secrets diff --git a/go/director/README.md b/go/director/README.md new file mode 100644 index 0000000..7a75302 --- /dev/null +++ b/go/director/README.md @@ -0,0 +1,56 @@ +# Director pattern — telemetry triage + +The director function `telemetry-workflow` validates a sensor reading, invokes +temperature and battery checks in parallel, and combines their results into an +`ok` or `alert` response: + +``` +telemetry-workflow ◄── director: owns the workflow and handles errors + ├── 1. invoke ──► validate-reading + ├── 2. invoke in parallel + │ ├──► temperature-check ──┐ + │ └──► battery-check ──────┤ + │◄──── check results ────────────┘ + └── 3. combine results and return ok or alert +``` + +Documentation: [Director pattern](https://docs.openfaas.com/languages/patterns/director/) + +## Build and deploy + +```bash +faas-cli up --tag=digest +``` + +The stage functions are invoked through the gateway. The `stage_timeout` +environment variable limits each downstream call, while the watchdog timeout +variables limit the complete director invocation. + +## Invoke + +```bash +curl -s http://127.0.0.1:8080/function/telemetry-workflow \ + -H "Content-Type: application/json" \ + -d '{"device_id":"pump-17","temperature_c":82.4,"battery_percent":12}' | \ + jq +``` + +Example output: + +```json +{ + "device_id": "pump-17", + "status": "alert", + "temperature": { + "value_c": 82.4, + "threshold_c": 75, + "alert": true + }, + "battery": { + "value_percent": 12, + "threshold_percent": 20, + "alert": true + }, + "duration_ms": 4 +} +``` diff --git a/go/director/battery-check/go.mod b/go/director/battery-check/go.mod new file mode 100644 index 0000000..1437d2b --- /dev/null +++ b/go/director/battery-check/go.mod @@ -0,0 +1,3 @@ +module handler/function + +go 1.25.0 diff --git a/go/director/battery-check/handler.go b/go/director/battery-check/handler.go new file mode 100644 index 0000000..7a37b84 --- /dev/null +++ b/go/director/battery-check/handler.go @@ -0,0 +1,35 @@ +package function + +import ( + "encoding/json" + "net/http" +) + +const thresholdPercent = 20 + +type Reading struct { + BatteryPercent int `json:"battery_percent"` +} + +type Response struct { + ValuePercent int `json:"value_percent"` + ThresholdPercent int `json:"threshold_percent"` + Alert bool `json:"alert"` +} + +func Handle(w http.ResponseWriter, r *http.Request) { + defer r.Body.Close() + + var reading Reading + if err := json.NewDecoder(r.Body).Decode(&reading); err != nil { + http.Error(w, "expected a JSON sensor reading", http.StatusBadRequest) + return + } + + w.Header().Set("Content-Type", "application/json") + json.NewEncoder(w).Encode(Response{ + ValuePercent: reading.BatteryPercent, + ThresholdPercent: thresholdPercent, + Alert: reading.BatteryPercent < thresholdPercent, + }) +} diff --git a/go/director/battery-check/handler_test.go b/go/director/battery-check/handler_test.go new file mode 100644 index 0000000..324f363 --- /dev/null +++ b/go/director/battery-check/handler_test.go @@ -0,0 +1,44 @@ +package function + +import ( + "encoding/json" + "net/http" + "net/http/httptest" + "strings" + "testing" +) + +func TestHandleBatteryThreshold(t *testing.T) { + tests := []struct { + name string + value int + alert bool + }{ + {name: "below", value: 19, alert: true}, + {name: "at threshold", value: 20, alert: false}, + {name: "above", value: 21, alert: false}, + } + + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + req := httptest.NewRequest(http.MethodPost, "/", strings.NewReader( + `{"battery_percent":`+jsonNumber(test.value)+`}`, + )) + recorder := httptest.NewRecorder() + Handle(recorder, req) + + var response Response + if err := json.Unmarshal(recorder.Body.Bytes(), &response); err != nil { + t.Fatal(err) + } + if response.Alert != test.alert { + t.Fatalf("want alert %t, got %t", test.alert, response.Alert) + } + }) + } +} + +func jsonNumber(value int) string { + data, _ := json.Marshal(value) + return string(data) +} diff --git a/go/director/stack.yaml b/go/director/stack.yaml new file mode 100644 index 0000000..5735773 --- /dev/null +++ b/go/director/stack.yaml @@ -0,0 +1,29 @@ +version: 1.0 +provider: + name: openfaas + gateway: http://127.0.0.1:8080 +functions: + telemetry-workflow: + lang: golang-middleware + handler: ./telemetry-workflow + image: ${REGISTRY:-ttl.sh}/${OWNER:-openfaas-examples}/telemetry-workflow:latest + environment: + stage_timeout: 5s + exec_timeout: 15s + read_timeout: 16s + write_timeout: 16s + + validate-reading: + lang: golang-middleware + handler: ./validate-reading + image: ${REGISTRY:-ttl.sh}/${OWNER:-openfaas-examples}/validate-reading:latest + + temperature-check: + lang: golang-middleware + handler: ./temperature-check + image: ${REGISTRY:-ttl.sh}/${OWNER:-openfaas-examples}/temperature-check:latest + + battery-check: + lang: golang-middleware + handler: ./battery-check + image: ${REGISTRY:-ttl.sh}/${OWNER:-openfaas-examples}/battery-check:latest diff --git a/go/director/telemetry-workflow/go.mod b/go/director/telemetry-workflow/go.mod new file mode 100644 index 0000000..1437d2b --- /dev/null +++ b/go/director/telemetry-workflow/go.mod @@ -0,0 +1,3 @@ +module handler/function + +go 1.25.0 diff --git a/go/director/telemetry-workflow/handler.go b/go/director/telemetry-workflow/handler.go new file mode 100644 index 0000000..bfe9eb0 --- /dev/null +++ b/go/director/telemetry-workflow/handler.go @@ -0,0 +1,226 @@ +package function + +import ( + "bytes" + "context" + "encoding/json" + "fmt" + "io" + "net/http" + "os" + "strings" + "time" +) + +const defaultStageTimeout = 5 * time.Second + +type Reading struct { + DeviceID string `json:"device_id"` + TemperatureC float64 `json:"temperature_c"` + BatteryPercent int `json:"battery_percent"` +} + +type TemperatureResult struct { + ValueC float64 `json:"value_c"` + ThresholdC float64 `json:"threshold_c"` + Alert bool `json:"alert"` +} + +type BatteryResult struct { + ValuePercent int `json:"value_percent"` + ThresholdPercent int `json:"threshold_percent"` + Alert bool `json:"alert"` +} + +type Response struct { + DeviceID string `json:"device_id"` + Status string `json:"status"` + Temperature TemperatureResult `json:"temperature"` + Battery BatteryResult `json:"battery"` + DurationMs int64 `json:"duration_ms"` +} + +type callResult struct { + function string + body []byte + status int + err error +} + +func Handle(w http.ResponseWriter, r *http.Request) { + start := time.Now() + + input, err := io.ReadAll(r.Body) + if err != nil { + http.Error(w, "unable to read request body", http.StatusBadRequest) + return + } + defer r.Body.Close() + + timeout, err := configuredStageTimeout() + if err != nil { + http.Error(w, err.Error(), http.StatusInternalServerError) + return + } + + gateway := os.Getenv("gateway_url") + if gateway == "" { + gateway = "http://gateway.openfaas:8080" + } + + client := &http.Client{Timeout: timeout} + + validated, status, err := invoke( + r.Context(), client, gateway, "validate-reading", input, + ) + if err != nil { + message := fmt.Sprintf("failed to invoke validate-reading: %s", err) + http.Error(w, message, http.StatusBadGateway) + return + } + if status != http.StatusOK { + message := fmt.Sprintf("validate-reading failed: %s", validated) + http.Error(w, message, status) + return + } + + var reading Reading + if err := json.Unmarshal(validated, &reading); err != nil { + message := fmt.Sprintf( + "unexpected response from validate-reading: %s", + err, + ) + http.Error(w, message, http.StatusBadGateway) + return + } + + functions := []string{"temperature-check", "battery-check"} + results := make(chan callResult, len(functions)) + + for _, function := range functions { + go func(name string) { + body, status, err := invoke( + r.Context(), client, gateway, name, validated, + ) + results <- callResult{ + function: name, + body: body, + status: status, + err: err, + } + }(function) + } + + completed := make(map[string]callResult, len(functions)) + for range functions { + result := <-results + completed[result.function] = result + } + + for _, function := range functions { + result := completed[function] + if result.err != nil { + message := fmt.Sprintf( + "failed to invoke %s: %s", + function, + result.err, + ) + http.Error(w, message, http.StatusBadGateway) + return + } + if result.status != http.StatusOK { + message := fmt.Sprintf("%s failed: %s", function, result.body) + http.Error(w, message, result.status) + return + } + } + + var temperature TemperatureResult + if err := json.Unmarshal( + completed["temperature-check"].body, + &temperature, + ); err != nil { + message := fmt.Sprintf( + "unexpected response from temperature-check: %s", + err, + ) + http.Error(w, message, http.StatusBadGateway) + return + } + + var battery BatteryResult + if err := json.Unmarshal( + completed["battery-check"].body, + &battery, + ); err != nil { + message := fmt.Sprintf( + "unexpected response from battery-check: %s", + err, + ) + http.Error(w, message, http.StatusBadGateway) + return + } + + workflowStatus := "ok" + if temperature.Alert || battery.Alert { + workflowStatus = "alert" + } + + response := Response{ + DeviceID: reading.DeviceID, + Status: workflowStatus, + Temperature: temperature, + Battery: battery, + DurationMs: time.Since(start).Milliseconds(), + } + + w.Header().Set("Content-Type", "application/json") + json.NewEncoder(w).Encode(response) +} + +func configuredStageTimeout() (time.Duration, error) { + value := os.Getenv("stage_timeout") + if value == "" { + return defaultStageTimeout, nil + } + + timeout, err := time.ParseDuration(value) + if err != nil || timeout <= 0 { + return 0, fmt.Errorf("invalid stage_timeout %q", value) + } + + return timeout, nil +} + +func invoke( + ctx context.Context, + client *http.Client, + gateway string, + function string, + body []byte, +) ([]byte, int, error) { + url := strings.TrimRight(gateway, "/") + "/function/" + function + req, err := http.NewRequestWithContext( + ctx, + http.MethodPost, + url, + bytes.NewReader(body), + ) + if err != nil { + return nil, 0, fmt.Errorf("create request for %s: %w", function, err) + } + req.Header.Set("Content-Type", "application/json") + + res, err := client.Do(req) + if err != nil { + return nil, 0, fmt.Errorf("invoke %s: %w", function, err) + } + defer res.Body.Close() + + out, err := io.ReadAll(res.Body) + if err != nil { + return nil, 0, fmt.Errorf("read response from %s: %w", function, err) + } + + return out, res.StatusCode, nil +} diff --git a/go/director/telemetry-workflow/handler_test.go b/go/director/telemetry-workflow/handler_test.go new file mode 100644 index 0000000..31aa744 --- /dev/null +++ b/go/director/telemetry-workflow/handler_test.go @@ -0,0 +1,176 @@ +package function + +import ( + "encoding/json" + "net/http" + "net/http/httptest" + "strings" + "sync" + "sync/atomic" + "testing" + "time" +) + +func TestHandleCombinesParallelChecks(t *testing.T) { + started := make(chan string, 2) + release := make(chan struct{}) + var releaseOnce sync.Once + defer releaseOnce.Do(func() { close(release) }) + + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + switch r.URL.Path { + case "/function/validate-reading": + w.Header().Set("Content-Type", "application/json") + w.Write([]byte(`{"device_id":"pump-17","temperature_c":82.4,"battery_percent":12}`)) + case "/function/temperature-check": + started <- "temperature-check" + <-release + w.Header().Set("Content-Type", "application/json") + w.Write([]byte(`{"value_c":82.4,"threshold_c":75,"alert":true}`)) + case "/function/battery-check": + started <- "battery-check" + <-release + w.Header().Set("Content-Type", "application/json") + w.Write([]byte(`{"value_percent":12,"threshold_percent":20,"alert":true}`)) + default: + http.NotFound(w, r) + } + })) + defer server.Close() + + t.Setenv("gateway_url", server.URL) + t.Setenv("stage_timeout", "1s") + + req := httptest.NewRequest(http.MethodPost, "/", strings.NewReader( + `{"device_id":"pump-17","temperature_c":82.4,"battery_percent":12}`, + )) + recorder := httptest.NewRecorder() + done := make(chan struct{}) + + go func() { + Handle(recorder, req) + close(done) + }() + + seen := map[string]bool{} + for len(seen) < 2 { + select { + case function := <-started: + seen[function] = true + case <-time.After(time.Second): + t.Fatal("parallel checks did not start together") + } + } + + releaseOnce.Do(func() { close(release) }) + <-done + + if recorder.Code != http.StatusOK { + t.Fatalf("want status 200, got %d: %s", recorder.Code, recorder.Body.String()) + } + + var response Response + if err := json.Unmarshal(recorder.Body.Bytes(), &response); err != nil { + t.Fatal(err) + } + if response.Status != "alert" { + t.Fatalf("want alert status, got %q", response.Status) + } + if !response.Temperature.Alert || !response.Battery.Alert { + t.Fatalf("want both checks to alert: %+v", response) + } +} + +func TestHandleReturnsOK(t *testing.T) { + server := newStageServer(t, false) + defer server.Close() + + t.Setenv("gateway_url", server.URL) + t.Setenv("stage_timeout", "1s") + + req := httptest.NewRequest(http.MethodPost, "/", strings.NewReader( + `{"device_id":"pump-17","temperature_c":48.2,"battery_percent":78}`, + )) + recorder := httptest.NewRecorder() + Handle(recorder, req) + + var response Response + if err := json.Unmarshal(recorder.Body.Bytes(), &response); err != nil { + t.Fatal(err) + } + if response.Status != "ok" { + t.Fatalf("want ok status, got %q", response.Status) + } +} + +func TestHandleStopsAfterValidationFailure(t *testing.T) { + var checks atomic.Int32 + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path == "/function/validate-reading" { + http.Error(w, "device_id is required", http.StatusBadRequest) + return + } + checks.Add(1) + })) + defer server.Close() + + t.Setenv("gateway_url", server.URL) + t.Setenv("stage_timeout", "1s") + + req := httptest.NewRequest(http.MethodPost, "/", strings.NewReader(`{"temperature_c":20}`)) + recorder := httptest.NewRecorder() + Handle(recorder, req) + + if recorder.Code != http.StatusBadRequest { + t.Fatalf("want status 400, got %d", recorder.Code) + } + if checks.Load() != 0 { + t.Fatalf("want no checks after validation failure, got %d", checks.Load()) + } +} + +func TestHandleTimesOutDownstreamCalls(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path == "/function/validate-reading" { + w.Write([]byte(`{"device_id":"pump-17","temperature_c":48.2,"battery_percent":78}`)) + return + } + time.Sleep(50 * time.Millisecond) + w.Write([]byte(`{"alert":false}`)) + })) + defer server.Close() + + t.Setenv("gateway_url", server.URL) + t.Setenv("stage_timeout", "5ms") + + req := httptest.NewRequest(http.MethodPost, "/", strings.NewReader( + `{"device_id":"pump-17","temperature_c":48.2,"battery_percent":78}`, + )) + recorder := httptest.NewRecorder() + Handle(recorder, req) + + if recorder.Code != http.StatusBadGateway { + t.Fatalf("want status 502, got %d: %s", recorder.Code, recorder.Body.String()) + } + if !strings.Contains(recorder.Body.String(), "Client.Timeout") { + t.Fatalf("want timeout error, got %q", recorder.Body.String()) + } +} + +func newStageServer(t *testing.T, alert bool) *httptest.Server { + t.Helper() + + return httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "application/json") + switch r.URL.Path { + case "/function/validate-reading": + w.Write([]byte(`{"device_id":"pump-17","temperature_c":48.2,"battery_percent":78}`)) + case "/function/temperature-check": + json.NewEncoder(w).Encode(TemperatureResult{ValueC: 48.2, ThresholdC: 75, Alert: alert}) + case "/function/battery-check": + json.NewEncoder(w).Encode(BatteryResult{ValuePercent: 78, ThresholdPercent: 20, Alert: alert}) + default: + http.NotFound(w, r) + } + })) +} diff --git a/go/director/temperature-check/go.mod b/go/director/temperature-check/go.mod new file mode 100644 index 0000000..1437d2b --- /dev/null +++ b/go/director/temperature-check/go.mod @@ -0,0 +1,3 @@ +module handler/function + +go 1.25.0 diff --git a/go/director/temperature-check/handler.go b/go/director/temperature-check/handler.go new file mode 100644 index 0000000..a8d4dab --- /dev/null +++ b/go/director/temperature-check/handler.go @@ -0,0 +1,35 @@ +package function + +import ( + "encoding/json" + "net/http" +) + +const thresholdC = 75.0 + +type Reading struct { + TemperatureC float64 `json:"temperature_c"` +} + +type Response struct { + ValueC float64 `json:"value_c"` + ThresholdC float64 `json:"threshold_c"` + Alert bool `json:"alert"` +} + +func Handle(w http.ResponseWriter, r *http.Request) { + defer r.Body.Close() + + var reading Reading + if err := json.NewDecoder(r.Body).Decode(&reading); err != nil { + http.Error(w, "expected a JSON sensor reading", http.StatusBadRequest) + return + } + + w.Header().Set("Content-Type", "application/json") + json.NewEncoder(w).Encode(Response{ + ValueC: reading.TemperatureC, + ThresholdC: thresholdC, + Alert: reading.TemperatureC > thresholdC, + }) +} diff --git a/go/director/temperature-check/handler_test.go b/go/director/temperature-check/handler_test.go new file mode 100644 index 0000000..66f9ffb --- /dev/null +++ b/go/director/temperature-check/handler_test.go @@ -0,0 +1,44 @@ +package function + +import ( + "encoding/json" + "net/http" + "net/http/httptest" + "strings" + "testing" +) + +func TestHandleTemperatureThreshold(t *testing.T) { + tests := []struct { + name string + value float64 + alert bool + }{ + {name: "below", value: 74.9, alert: false}, + {name: "at threshold", value: 75, alert: false}, + {name: "above", value: 75.1, alert: true}, + } + + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + req := httptest.NewRequest(http.MethodPost, "/", strings.NewReader( + `{"temperature_c":`+jsonNumber(test.value)+`}`, + )) + recorder := httptest.NewRecorder() + Handle(recorder, req) + + var response Response + if err := json.Unmarshal(recorder.Body.Bytes(), &response); err != nil { + t.Fatal(err) + } + if response.Alert != test.alert { + t.Fatalf("want alert %t, got %t", test.alert, response.Alert) + } + }) + } +} + +func jsonNumber(value float64) string { + data, _ := json.Marshal(value) + return string(data) +} diff --git a/go/director/validate-reading/go.mod b/go/director/validate-reading/go.mod new file mode 100644 index 0000000..1437d2b --- /dev/null +++ b/go/director/validate-reading/go.mod @@ -0,0 +1,3 @@ +module handler/function + +go 1.25.0 diff --git a/go/director/validate-reading/handler.go b/go/director/validate-reading/handler.go new file mode 100644 index 0000000..9924af2 --- /dev/null +++ b/go/director/validate-reading/handler.go @@ -0,0 +1,48 @@ +package function + +import ( + "encoding/json" + "net/http" + "strings" +) + +type Reading struct { + DeviceID string `json:"device_id"` + TemperatureC float64 `json:"temperature_c"` + BatteryPercent int `json:"battery_percent"` +} + +func Handle(w http.ResponseWriter, r *http.Request) { + defer r.Body.Close() + + var reading Reading + if err := json.NewDecoder(r.Body).Decode(&reading); err != nil { + http.Error(w, "expected a JSON sensor reading", http.StatusBadRequest) + return + } + + reading.DeviceID = strings.TrimSpace(reading.DeviceID) + if reading.DeviceID == "" { + http.Error(w, "device_id is required", http.StatusBadRequest) + return + } + if reading.TemperatureC < -100 || reading.TemperatureC > 200 { + http.Error( + w, + "temperature_c must be between -100 and 200", + http.StatusBadRequest, + ) + return + } + if reading.BatteryPercent < 0 || reading.BatteryPercent > 100 { + http.Error( + w, + "battery_percent must be between 0 and 100", + http.StatusBadRequest, + ) + return + } + + w.Header().Set("Content-Type", "application/json") + json.NewEncoder(w).Encode(reading) +} diff --git a/go/director/validate-reading/handler_test.go b/go/director/validate-reading/handler_test.go new file mode 100644 index 0000000..095e650 --- /dev/null +++ b/go/director/validate-reading/handler_test.go @@ -0,0 +1,41 @@ +package function + +import ( + "encoding/json" + "net/http" + "net/http/httptest" + "strings" + "testing" +) + +func TestHandleValidReading(t *testing.T) { + req := httptest.NewRequest(http.MethodPost, "/", strings.NewReader( + `{"device_id":" pump-17 ","temperature_c":48.2,"battery_percent":78}`, + )) + recorder := httptest.NewRecorder() + Handle(recorder, req) + + if recorder.Code != http.StatusOK { + t.Fatalf("want status 200, got %d: %s", recorder.Code, recorder.Body.String()) + } + + var reading Reading + if err := json.Unmarshal(recorder.Body.Bytes(), &reading); err != nil { + t.Fatal(err) + } + if reading.DeviceID != "pump-17" { + t.Fatalf("want trimmed device ID, got %q", reading.DeviceID) + } +} + +func TestHandleRejectsInvalidBattery(t *testing.T) { + req := httptest.NewRequest(http.MethodPost, "/", strings.NewReader( + `{"device_id":"pump-17","temperature_c":48.2,"battery_percent":101}`, + )) + recorder := httptest.NewRecorder() + Handle(recorder, req) + + if recorder.Code != http.StatusBadRequest { + t.Fatalf("want status 400, got %d", recorder.Code) + } +} diff --git a/go/fan-out/.gitignore b/go/fan-out/.gitignore new file mode 100644 index 0000000..4f03565 --- /dev/null +++ b/go/fan-out/.gitignore @@ -0,0 +1,3 @@ +template +build +.secrets diff --git a/go/fan-out/README.md b/go/fan-out/README.md new file mode 100644 index 0000000..25d50de --- /dev/null +++ b/go/fan-out/README.md @@ -0,0 +1,66 @@ +# Fan-out pattern — URL health checks + +The `fan-out` function accepts one trusted URL per line and submits each URL as +an asynchronous invocation of the `url-check` function. It returns a summary +immediately without waiting for the checks to complete. + +Documentation: [Fan-out pattern](https://docs.openfaas.com/languages/patterns/fan-out/) + +## Build and deploy + +```bash +faas-cli up --tag=digest +``` + +Environment variables consumed by `fan-out`: + +| Variable | Default | Description | +|---|---|---| +| `gateway_url` | `http://gateway.openfaas:8080` | In-cluster gateway address | +| `callback_url` | (none) | Optional `X-Callback-Url` for function results | + +## Invoke + +Newlines matter, so use `--data-binary`: + +```bash +printf 'https://www.openfaas.com/\nhttps://docs.openfaas.com/\n' | \ + curl -s --data-binary @- http://127.0.0.1:8080/function/fan-out | jq +``` + +Example output: + +```json +{ + "submitted": 2, + "function": "url-check", + "callback": false, + "call_ids": [ + "9c0b1a12-fdea-4f01-baff-c5d9f50435ea", + "4111d512-cdf3-4b8f-96b3-1b7f1f376bd7" + ] +} +``` + +Each URL is queued for the `url-check` function. Track the number of function +invocations with `faas-cli list`, or cancel an individual check with its call +ID via `DELETE /async-function/`. + +## Callbacks + +Deploy the `printer` function, then pass `X-Callback-Url` with the batch to +receive every health-check result: + +```bash +faas-cli store deploy printer + +printf 'https://www.openfaas.com/\nhttps://docs.openfaas.com/\n' | \ + curl -s --data-binary @- \ + -H "X-Callback-Url: http://gateway.openfaas:8080/function/printer" \ + http://127.0.0.1:8080/function/fan-out +``` + +Inspect the individual callback bodies with `faas-cli logs printer`. This +demonstrates result delivery only; it does not wait for all checks or combine +their results. On OpenFaaS Pro, callbacks may be restricted by the +queue-worker's `allowedCallbackURLs` setting. diff --git a/go/fan-out/fan-out/go.mod b/go/fan-out/fan-out/go.mod new file mode 100644 index 0000000..1437d2b --- /dev/null +++ b/go/fan-out/fan-out/go.mod @@ -0,0 +1,3 @@ +module handler/function + +go 1.25.0 diff --git a/go/fan-out/fan-out/handler.go b/go/fan-out/fan-out/handler.go new file mode 100644 index 0000000..fd885c5 --- /dev/null +++ b/go/fan-out/fan-out/handler.go @@ -0,0 +1,160 @@ +package function + +import ( + "bytes" + "context" + "encoding/json" + "fmt" + "io" + "net/http" + "os" + "strings" + "time" +) + +const ( + targetFunction = "url-check" + submitTimeout = 30 * time.Second +) + +type Response struct { + Submitted int `json:"submitted"` + Function string `json:"function"` + Callback bool `json:"callback"` + CallIDs []string `json:"call_ids,omitempty"` +} + +// Handle takes a HTTP request body and splits it into one record per line. +// Each record is submitted as an asynchronous invocation of the target +// function, then a summary is returned to the caller without waiting for +// the function invocations to complete. +func Handle(w http.ResponseWriter, r *http.Request) { + input, err := io.ReadAll(r.Body) + if err != nil { + http.Error(w, "unable to read request body", http.StatusBadRequest) + return + } + defer r.Body.Close() + + gateway := os.Getenv("gateway_url") + if gateway == "" { + gateway = "http://gateway.openfaas:8080" + } + + // Forward the callback URL to every asynchronous invocation. A header on + // the batch request overrides the environment variable. + callback := strings.TrimSpace(r.Header.Get("X-Callback-Url")) + if callback == "" { + callback = strings.TrimSpace(os.Getenv("callback_url")) + } + + records := recordsFromInput(string(input)) + if len(records) == 0 { + http.Error( + w, + "expected one record per line in the request body", + http.StatusBadRequest, + ) + return + } + + submitted := 0 + var callIDs []string + + for i, record := range records { + callID, err := submit( + r.Context(), gateway, targetFunction, record, callback, + ) + if err != nil { + message := fmt.Sprintf( + "record %d of %d: %s", + i+1, + len(records), + err, + ) + http.Error(w, message, http.StatusBadGateway) + return + } + + submitted++ + if callID != "" { + callIDs = append(callIDs, callID) + } + } + + res := Response{ + Submitted: submitted, + Function: targetFunction, + Callback: callback != "", + CallIDs: callIDs, + } + + w.Header().Set("Content-Type", "application/json") + json.NewEncoder(w).Encode(res) +} + +func recordsFromInput(input string) []string { + var records []string + + for _, record := range strings.Split(strings.TrimSpace(input), "\n") { + record = strings.TrimSpace(record) + if record != "" { + records = append(records, record) + } + } + + return records +} + +func submit( + ctx context.Context, + gateway string, + targetFunction string, + record string, + callback string, +) (string, error) { + ctx, cancel := context.WithTimeout(ctx, submitTimeout) + defer cancel() + + url := strings.TrimRight(gateway, "/") + "/async-function/" + targetFunction + req, err := http.NewRequestWithContext( + ctx, + http.MethodPost, + url, + bytes.NewReader([]byte(record)), + ) + if err != nil { + return "", fmt.Errorf("unable to invoke %s: %w", targetFunction, err) + } + req.Header.Set("Content-Type", "text/plain") + if callback != "" { + req.Header.Set("X-Callback-Url", callback) + } + + res, err := http.DefaultClient.Do(req) + if err != nil { + return "", fmt.Errorf("error invoking %s: %w", targetFunction, err) + } + defer res.Body.Close() + + if res.StatusCode != http.StatusAccepted { + out, err := io.ReadAll(res.Body) + if err != nil { + return "", fmt.Errorf( + "unexpected status %d from %s", + res.StatusCode, + targetFunction, + ) + } + + return "", fmt.Errorf( + "unexpected status %d from %s: %s", + res.StatusCode, + targetFunction, + string(out), + ) + } + + // the X-Call-Id header can be used to track or cancel the record + return res.Header.Get("X-Call-Id"), nil +} diff --git a/go/fan-out/fan-out/handler_test.go b/go/fan-out/fan-out/handler_test.go new file mode 100644 index 0000000..1c6b2b1 --- /dev/null +++ b/go/fan-out/fan-out/handler_test.go @@ -0,0 +1,98 @@ +package function + +import ( + "encoding/json" + "io" + "net/http" + "net/http/httptest" + "strings" + "sync" + "testing" +) + +func TestRecordsFromInput(t *testing.T) { + records := recordsFromInput("https://one.example\n\n https://two.example \n") + if len(records) != 2 { + t.Fatalf("want 2 records, got %d", len(records)) + } + if records[0] != "https://one.example" || records[1] != "https://two.example" { + t.Fatalf("unexpected records: %#v", records) + } +} + +func TestHandleSubmitsEachURLWithCallback(t *testing.T) { + var mu sync.Mutex + var bodies []string + var callbacks []string + + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path != "/async-function/url-check" { + http.NotFound(w, r) + return + } + + body, _ := io.ReadAll(r.Body) + mu.Lock() + bodies = append(bodies, string(body)) + callbacks = append(callbacks, r.Header.Get("X-Callback-Url")) + callID := "call-" + string(rune('0'+len(bodies))) + mu.Unlock() + + w.Header().Set("X-Call-Id", callID) + w.WriteHeader(http.StatusAccepted) + })) + defer server.Close() + + t.Setenv("gateway_url", server.URL) + t.Setenv("callback_url", "") + + req := httptest.NewRequest(http.MethodPost, "/", strings.NewReader( + "https://one.example\nhttps://two.example\n", + )) + req.Header.Set("X-Callback-Url", "http://gateway.openfaas:8080/function/printer") + recorder := httptest.NewRecorder() + Handle(recorder, req) + + if recorder.Code != http.StatusOK { + t.Fatalf("want status 200, got %d: %s", recorder.Code, recorder.Body.String()) + } + + var response Response + if err := json.Unmarshal(recorder.Body.Bytes(), &response); err != nil { + t.Fatal(err) + } + if response.Submitted != 2 || response.Function != "url-check" || !response.Callback { + t.Fatalf("unexpected response: %+v", response) + } + if len(response.CallIDs) != 2 { + t.Fatalf("want 2 call IDs, got %#v", response.CallIDs) + } + + mu.Lock() + defer mu.Unlock() + if len(bodies) != 2 || bodies[0] != "https://one.example" || bodies[1] != "https://two.example" { + t.Fatalf("unexpected bodies: %#v", bodies) + } + for _, callback := range callbacks { + if callback != "http://gateway.openfaas:8080/function/printer" { + t.Fatalf("callback was not forwarded: %q", callback) + } + } +} + +func TestHandleReturnsSubmissionFailure(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + http.Error(w, "queue unavailable", http.StatusServiceUnavailable) + })) + defer server.Close() + + t.Setenv("gateway_url", server.URL) + + req := httptest.NewRequest(http.MethodPost, "/", strings.NewReader("https://one.example\n")) + recorder := httptest.NewRecorder() + Handle(recorder, req) + + if recorder.Code != http.StatusBadGateway { + t.Fatalf("want status 502, got %d", recorder.Code) + } +} diff --git a/go/fan-out/stack.yaml b/go/fan-out/stack.yaml new file mode 100644 index 0000000..2377c7c --- /dev/null +++ b/go/fan-out/stack.yaml @@ -0,0 +1,16 @@ +version: 1.0 +provider: + name: openfaas + gateway: http://127.0.0.1:8080 +functions: + fan-out: + lang: golang-middleware + handler: ./fan-out + image: ${REGISTRY:-ttl.sh}/${OWNER:-openfaas-examples}/fan-out:latest + + url-check: + lang: golang-middleware + handler: ./url-check + image: ${REGISTRY:-ttl.sh}/${OWNER:-openfaas-examples}/url-check:latest + environment: + request_timeout: 5s diff --git a/go/fan-out/url-check/go.mod b/go/fan-out/url-check/go.mod new file mode 100644 index 0000000..1437d2b --- /dev/null +++ b/go/fan-out/url-check/go.mod @@ -0,0 +1,3 @@ +module handler/function + +go 1.25.0 diff --git a/go/fan-out/url-check/handler.go b/go/fan-out/url-check/handler.go new file mode 100644 index 0000000..30a2074 --- /dev/null +++ b/go/fan-out/url-check/handler.go @@ -0,0 +1,116 @@ +package function + +import ( + "context" + "encoding/json" + "fmt" + "io" + "net/http" + "net/url" + "os" + "strings" + "time" +) + +const ( + defaultRequestTimeout = 5 * time.Second + maxURLLength = 4096 +) + +var requestTimeout = defaultRequestTimeout + +func init() { + value := os.Getenv("request_timeout") + if value == "" { + return + } + + timeout, err := time.ParseDuration(value) + if err != nil || timeout <= 0 { + panic(fmt.Sprintf("invalid request_timeout %q", value)) + } + + requestTimeout = timeout +} + +type Response struct { + URL string `json:"url"` + Reachable bool `json:"reachable"` + Healthy bool `json:"healthy"` + StatusCode int `json:"status_code,omitempty"` + ContentType string `json:"content_type,omitempty"` + DurationMs int64 `json:"duration_ms"` + Error string `json:"error,omitempty"` +} + +func Handle(w http.ResponseWriter, r *http.Request) { + defer r.Body.Close() + + input, err := io.ReadAll(io.LimitReader(r.Body, maxURLLength+1)) + if err != nil { + http.Error(w, "unable to read request body", http.StatusBadRequest) + return + } + if len(input) > maxURLLength { + http.Error(w, "URL is too long", http.StatusBadRequest) + return + } + + target := strings.TrimSpace(string(input)) + parsed, err := url.ParseRequestURI(target) + if err != nil || parsed.Host == "" { + http.Error( + w, + "expected an absolute HTTP or HTTPS URL", + http.StatusBadRequest, + ) + return + } + if parsed.Scheme != "http" && parsed.Scheme != "https" { + http.Error( + w, + "expected an absolute HTTP or HTTPS URL", + http.StatusBadRequest, + ) + return + } + + start := time.Now() + ctx, cancel := context.WithTimeout(r.Context(), requestTimeout) + defer cancel() + + req, err := http.NewRequestWithContext(ctx, http.MethodGet, target, nil) + if err != nil { + http.Error( + w, + "unable to create health-check request", + http.StatusBadRequest, + ) + return + } + req.Header.Set("User-Agent", "OpenFaaS URL health check") + + res, requestErr := http.DefaultClient.Do(req) + result := Response{ + URL: target, + DurationMs: time.Since(start).Milliseconds(), + } + if requestErr != nil { + result.Error = requestErr.Error() + writeJSON(w, result) + return + } + defer res.Body.Close() + io.Copy(io.Discard, io.LimitReader(res.Body, 1024)) + + result.Reachable = true + result.Healthy = res.StatusCode >= 200 && res.StatusCode < 400 + result.StatusCode = res.StatusCode + result.ContentType = res.Header.Get("Content-Type") + writeJSON(w, result) +} + +func writeJSON(w http.ResponseWriter, result Response) { + w.Header().Set("Content-Type", "application/json") + json.NewEncoder(w).Encode(result) +} diff --git a/go/fan-out/url-check/handler_test.go b/go/fan-out/url-check/handler_test.go new file mode 100644 index 0000000..78cad5d --- /dev/null +++ b/go/fan-out/url-check/handler_test.go @@ -0,0 +1,83 @@ +package function + +import ( + "encoding/json" + "net/http" + "net/http/httptest" + "strings" + "testing" + "time" +) + +func TestHandleHealthyURL(t *testing.T) { + target := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "text/plain") + w.WriteHeader(http.StatusNoContent) + })) + defer target.Close() + + recorder := invokeURLCheck(t, target.URL) + + var response Response + if err := json.Unmarshal(recorder.Body.Bytes(), &response); err != nil { + t.Fatal(err) + } + if !response.Reachable || !response.Healthy || response.StatusCode != http.StatusNoContent { + t.Fatalf("unexpected response: %+v", response) + } +} + +func TestHandleUnhealthyStatus(t *testing.T) { + target := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.WriteHeader(http.StatusServiceUnavailable) + })) + defer target.Close() + + recorder := invokeURLCheck(t, target.URL) + + var response Response + if err := json.Unmarshal(recorder.Body.Bytes(), &response); err != nil { + t.Fatal(err) + } + if !response.Reachable || response.Healthy || response.StatusCode != http.StatusServiceUnavailable { + t.Fatalf("unexpected response: %+v", response) + } +} + +func TestHandleRejectsInvalidURL(t *testing.T) { + recorder := invokeURLCheck(t, "file:///etc/passwd") + if recorder.Code != http.StatusBadRequest { + t.Fatalf("want status 400, got %d", recorder.Code) + } +} + +func TestHandleReportsRequestTimeout(t *testing.T) { + target := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + time.Sleep(50 * time.Millisecond) + w.WriteHeader(http.StatusOK) + })) + defer target.Close() + + originalTimeout := requestTimeout + requestTimeout = 5 * time.Millisecond + t.Cleanup(func() { requestTimeout = originalTimeout }) + + recorder := invokeURLCheck(t, target.URL) + + var response Response + if err := json.Unmarshal(recorder.Body.Bytes(), &response); err != nil { + t.Fatal(err) + } + if response.Reachable || response.Error == "" { + t.Fatalf("want unreachable result with an error: %+v", response) + } +} + +func invokeURLCheck(t *testing.T, target string) *httptest.ResponseRecorder { + t.Helper() + + req := httptest.NewRequest(http.MethodPost, "/", strings.NewReader(target)) + recorder := httptest.NewRecorder() + Handle(recorder, req) + return recorder +} diff --git a/go/singleton/.gitignore b/go/singleton/.gitignore new file mode 100644 index 0000000..4f03565 --- /dev/null +++ b/go/singleton/.gitignore @@ -0,0 +1,3 @@ +template +build +.secrets diff --git a/go/singleton/README.md b/go/singleton/README.md new file mode 100644 index 0000000..9bff999 --- /dev/null +++ b/go/singleton/README.md @@ -0,0 +1,41 @@ +# Singleton pattern — live notification hub + +The `notification-hub` function broadcasts notifications to every client with +an open Server-Sent Events (SSE) connection. Its scaling labels keep exactly +one replica running, so publishers and subscribers share the same in-memory +subscriber list. + +WebSockets have the same connection-local state concern and provide +bidirectional communication. See the OpenFaaS article +[How to Integrate WebSockets with Serverless Functions and OpenFaaS](https://www.openfaas.com/blog/serverless-websockets/). + +The subscriber list is deliberately connection-local. Clients reconnect when +the function is restarted, rescheduled, or redeployed. To scale the hub across +multiple replicas, replace the in-memory broadcast with an external message +bus. + +Documentation: [Singleton pattern](https://docs.openfaas.com/languages/patterns/singleton/) + +## Build and deploy + +```bash +faas-cli up --tag=digest +``` + +## Invoke + +In one terminal, subscribe to notifications: + +```bash +curl -N -H "Accept: text/event-stream" \ + http://127.0.0.1:8080/function/notification-hub +``` + +In another terminal, publish a notification: + +```bash +curl -s -d "deployment complete" \ + http://127.0.0.1:8080/function/notification-hub +``` + +The subscriber receives `data: deployment complete` immediately. diff --git a/go/singleton/notification-hub/go.mod b/go/singleton/notification-hub/go.mod new file mode 100644 index 0000000..1437d2b --- /dev/null +++ b/go/singleton/notification-hub/go.mod @@ -0,0 +1,3 @@ +module handler/function + +go 1.25.0 diff --git a/go/singleton/notification-hub/handler.go b/go/singleton/notification-hub/handler.go new file mode 100644 index 0000000..6ed5160 --- /dev/null +++ b/go/singleton/notification-hub/handler.go @@ -0,0 +1,103 @@ +package function + +import ( + "fmt" + "io" + "net/http" + "strings" + "sync" +) + +var ( + subscribersMu sync.Mutex + subscribers = make(map[chan string]struct{}) +) + +func Handle(w http.ResponseWriter, r *http.Request) { + switch r.Method { + case http.MethodGet: + subscribe(w, r) + case http.MethodPost: + publish(w, r) + default: + w.Header().Set("Allow", "GET, POST") + http.Error(w, "method not allowed", http.StatusMethodNotAllowed) + } +} + +func subscribe(w http.ResponseWriter, r *http.Request) { + flusher, ok := w.(http.Flusher) + if !ok { + http.Error( + w, + "streaming is not supported", + http.StatusInternalServerError, + ) + return + } + + messages := make(chan string, 1) + subscribersMu.Lock() + subscribers[messages] = struct{}{} + subscribersMu.Unlock() + + defer func() { + subscribersMu.Lock() + delete(subscribers, messages) + subscribersMu.Unlock() + }() + + w.Header().Set("Content-Type", "text/event-stream") + w.Header().Set("Cache-Control", "no-cache") + fmt.Fprint(w, ": connected\n\n") + flusher.Flush() + + for { + select { + case message := <-messages: + fmt.Fprintf(w, "data: %s\n\n", message) + flusher.Flush() + case <-r.Context().Done(): + return + } + } +} + +func publish(w http.ResponseWriter, r *http.Request) { + defer r.Body.Close() + + body, err := io.ReadAll(r.Body) + if err != nil { + http.Error(w, "unable to read request body", http.StatusBadRequest) + return + } + + message := strings.TrimSpace(string(body)) + if message == "" { + http.Error(w, "notification must not be empty", http.StatusBadRequest) + return + } + + message = strings.NewReplacer("\r", " ", "\n", " ").Replace(message) + delivered := broadcast(message) + + w.Header().Set("Content-Type", "application/json") + fmt.Fprintf(w, "{\"delivered\":%d}\n", delivered) +} + +func broadcast(message string) int { + subscribersMu.Lock() + defer subscribersMu.Unlock() + + delivered := 0 + for subscriber := range subscribers { + select { + case subscriber <- message: + delivered++ + default: + // Do not let a slow subscriber block every other client. + } + } + + return delivered +} diff --git a/go/singleton/notification-hub/handler_test.go b/go/singleton/notification-hub/handler_test.go new file mode 100644 index 0000000..d4780e1 --- /dev/null +++ b/go/singleton/notification-hub/handler_test.go @@ -0,0 +1,45 @@ +package function + +import ( + "net/http" + "net/http/httptest" + "strings" + "testing" +) + +func TestPublishBroadcastsNotification(t *testing.T) { + resetSubscribers() + + messages := make(chan string, 1) + subscribers[messages] = struct{}{} + + req := httptest.NewRequest(http.MethodPost, "/", strings.NewReader("deployment complete")) + recorder := httptest.NewRecorder() + Handle(recorder, req) + + if recorder.Code != http.StatusOK { + t.Fatalf("want status 200, got %d", recorder.Code) + } + if got := recorder.Body.String(); got != "{\"delivered\":1}\n" { + t.Fatalf("unexpected response: %q", got) + } + if got := <-messages; got != "deployment complete" { + t.Fatalf("unexpected notification: %q", got) + } +} + +func TestPublishRejectsEmptyNotification(t *testing.T) { + req := httptest.NewRequest(http.MethodPost, "/", strings.NewReader(" ")) + recorder := httptest.NewRecorder() + Handle(recorder, req) + + if recorder.Code != http.StatusBadRequest { + t.Fatalf("want status 400, got %d", recorder.Code) + } +} + +func resetSubscribers() { + subscribersMu.Lock() + defer subscribersMu.Unlock() + subscribers = make(map[chan string]struct{}) +} diff --git a/go/singleton/stack.yaml b/go/singleton/stack.yaml new file mode 100644 index 0000000..bf7caf8 --- /dev/null +++ b/go/singleton/stack.yaml @@ -0,0 +1,16 @@ +version: 1.0 +provider: + name: openfaas + gateway: http://127.0.0.1:8080 +functions: + notification-hub: + lang: golang-middleware + handler: ./notification-hub + image: ${REGISTRY:-ttl.sh}/${OWNER:-openfaas-examples}/notification-hub:latest + labels: + com.openfaas.scale.min: "1" + com.openfaas.scale.max: "1" + environment: + exec_timeout: "1h" + read_timeout: "1h1s" + write_timeout: "1h1s" diff --git a/node/director/README.md b/node/director/README.md new file mode 100644 index 0000000..200b7e5 --- /dev/null +++ b/node/director/README.md @@ -0,0 +1,19 @@ +# Director pattern — telemetry triage + +This is the Node.js implementation of the telemetry workflow documented in +the OpenFaaS function patterns guide. It uses `node24` for all four functions +and runs the temperature and battery checks concurrently. + +Documentation: [Director pattern](https://docs.openfaas.com/languages/patterns/director/) + +```bash +faas-cli template store pull node24 +faas-cli up --tag=digest +``` + +```bash +curl -s http://127.0.0.1:8080/function/telemetry-workflow \ + -H "Content-Type: application/json" \ + -d '{"device_id":"pump-17","temperature_c":82.4,"battery_percent":12}' | \ + jq +``` diff --git a/node/director/battery-check/handler.js b/node/director/battery-check/handler.js new file mode 100644 index 0000000..bbd1535 --- /dev/null +++ b/node/director/battery-check/handler.js @@ -0,0 +1,40 @@ +'use strict' + +const thresholdPercent = 20 + +module.exports = async (event, context) => { + let reading + try { + reading = parseBody(event.body) + } catch (error) { + return fail(context) + } + + const value = reading.battery_percent + if (!Number.isInteger(value)) { + return fail(context) + } + + return context + .status(200) + .headers({ 'Content-Type': 'application/json' }) + .succeed({ + value_percent: value, + threshold_percent: thresholdPercent, + alert: value < thresholdPercent + }) +} + +function parseBody (body) { + if (Buffer.isBuffer(body)) { + return JSON.parse(body.toString()) + } + return typeof body === 'string' ? JSON.parse(body) : body +} + +function fail (context) { + return context + .status(400) + .headers({ 'Content-Type': 'text/plain' }) + .succeed('expected a JSON sensor reading') +} diff --git a/node/director/battery-check/handler.test.js b/node/director/battery-check/handler.test.js new file mode 100644 index 0000000..aa155d1 --- /dev/null +++ b/node/director/battery-check/handler.test.js @@ -0,0 +1,35 @@ +'use strict' + +const assert = require('node:assert/strict') +const { test } = require('node:test') + +const handler = require('./handler') + +test('raises an alert below the battery threshold', async () => { + const context = new TestContext() + await handler({ body: { battery_percent: 12 } }, context) + + assert.equal(context.statusCode, 200) + assert.deepEqual(context.body, { + value_percent: 12, + threshold_percent: 20, + alert: true + }) +}) + +class TestContext { + status (value) { + this.statusCode = value + return this + } + + headers (value) { + this.headerValues = value + return this + } + + succeed (value) { + this.body = value + return value + } +} diff --git a/node/director/battery-check/package.json b/node/director/battery-check/package.json new file mode 100644 index 0000000..7e8bea1 --- /dev/null +++ b/node/director/battery-check/package.json @@ -0,0 +1,8 @@ +{ + "name": "battery-check", + "version": "1.0.0", + "private": true, + "scripts": { + "test": "node --test" + } +} diff --git a/node/director/stack.yaml b/node/director/stack.yaml new file mode 100644 index 0000000..71621ed --- /dev/null +++ b/node/director/stack.yaml @@ -0,0 +1,29 @@ +version: 1.0 +provider: + name: openfaas + gateway: http://127.0.0.1:8080 +functions: + telemetry-workflow: + lang: node24 + handler: ./telemetry-workflow + image: ${REGISTRY:-ttl.sh}/${OWNER:-openfaas-examples}/node-telemetry-workflow:latest + environment: + stage_timeout: "5" + exec_timeout: 15s + read_timeout: 16s + write_timeout: 16s + + validate-reading: + lang: node24 + handler: ./validate-reading + image: ${REGISTRY:-ttl.sh}/${OWNER:-openfaas-examples}/node-validate-reading:latest + + temperature-check: + lang: node24 + handler: ./temperature-check + image: ${REGISTRY:-ttl.sh}/${OWNER:-openfaas-examples}/node-temperature-check:latest + + battery-check: + lang: node24 + handler: ./battery-check + image: ${REGISTRY:-ttl.sh}/${OWNER:-openfaas-examples}/node-battery-check:latest diff --git a/node/director/telemetry-workflow/handler.js b/node/director/telemetry-workflow/handler.js new file mode 100644 index 0000000..18f6b20 --- /dev/null +++ b/node/director/telemetry-workflow/handler.js @@ -0,0 +1,125 @@ +'use strict' + +const { performance } = require('node:perf_hooks') + +const gatewayURL = process.env.gateway_url || + 'http://gateway.openfaas:8080' +const stageTimeout = configuredTimeout( + process.env.stage_timeout || '5', + 'stage_timeout' +) + +module.exports = async (event, context) => { + const started = performance.now() + const input = requestBody(event.body) + + let validated + try { + validated = await invoke('validate-reading', input) + } catch (error) { + return fail( + context, + 502, + `failed to invoke validate-reading: ${error.message}` + ) + } + + if (validated.status !== 200) { + return fail( + context, + validated.status, + `validate-reading failed: ${await validated.text()}` + ) + } + + let reading + try { + reading = await validated.json() + } catch (error) { + return fail( + context, + 502, + `unexpected response from validate-reading: ${error.message}` + ) + } + + const body = JSON.stringify(reading) + const names = ['temperature-check', 'battery-check'] + let responses + try { + responses = await Promise.all( + names.map(async (name) => [name, await invoke(name, body)]) + ) + } catch (error) { + return fail(context, 502, `failed to invoke check: ${error.message}`) + } + + const completed = Object.fromEntries(responses) + for (const name of names) { + const response = completed[name] + if (response.status !== 200) { + return fail( + context, + response.status, + `${name} failed: ${await response.text()}` + ) + } + } + + let temperature + let battery + try { + temperature = await completed['temperature-check'].json() + battery = await completed['battery-check'].json() + } catch (error) { + return fail( + context, + 502, + `unexpected response from check function: ${error.message}` + ) + } + + const status = temperature.alert || battery.alert ? 'alert' : 'ok' + return context + .status(200) + .headers({ 'Content-Type': 'application/json' }) + .succeed({ + device_id: reading.device_id, + status, + temperature, + battery, + duration_ms: Math.round(performance.now() - started) + }) +} + +function invoke (name, body) { + const gateway = gatewayURL.replace(/\/$/, '') + return fetch(`${gateway}/function/${name}`, { + method: 'POST', + headers: { 'Content-Type': 'application/json' }, + body, + signal: AbortSignal.timeout(stageTimeout) + }) +} + +function requestBody (body) { + if (Buffer.isBuffer(body)) { + return body + } + return typeof body === 'string' ? body : JSON.stringify(body) +} + +function configuredTimeout (value, name) { + const seconds = Number(value) + if (!Number.isFinite(seconds) || seconds <= 0) { + throw new Error(`${name} must be greater than zero`) + } + return seconds * 1000 +} + +function fail (context, status, message) { + return context + .status(status) + .headers({ 'Content-Type': 'text/plain' }) + .succeed(message.trim()) +} diff --git a/node/director/telemetry-workflow/handler.test.js b/node/director/telemetry-workflow/handler.test.js new file mode 100644 index 0000000..70ce369 --- /dev/null +++ b/node/director/telemetry-workflow/handler.test.js @@ -0,0 +1,112 @@ +'use strict' + +const assert = require('node:assert/strict') +const { afterEach, test } = require('node:test') + +const handler = require('./handler') + +const originalFetch = global.fetch + +afterEach(() => { + global.fetch = originalFetch +}) + +test('runs validation before both checks and combines results', async () => { + const calls = [] + global.fetch = async (url) => { + const name = url.split('/').pop() + calls.push(name) + if (name === 'validate-reading') { + return jsonResponse({ + device_id: 'pump-17', + temperature_c: 82.4, + battery_percent: 12 + }) + } + if (name === 'temperature-check') { + return jsonResponse({ + value_c: 82.4, + threshold_c: 75, + alert: true + }) + } + return jsonResponse({ + value_percent: 12, + threshold_percent: 20, + alert: true + }) + } + + const context = new TestContext() + await handler({ body: inputReading() }, context) + + assert.equal(context.statusCode, 200) + assert.equal(context.body.status, 'alert') + assert.deepEqual(calls, [ + 'validate-reading', + 'temperature-check', + 'battery-check' + ]) +}) + +test('passes through a validation error', async () => { + global.fetch = async () => new Response('device_id is required', { + status: 400 + }) + + const context = new TestContext() + await handler({ body: inputReading() }, context) + + assert.equal(context.statusCode, 400) + assert.match(context.body, /device_id is required/) +}) + +test('returns bad gateway when a function cannot be invoked', async () => { + global.fetch = async () => { + throw new Error('connection refused') + } + + const context = new TestContext() + await handler({ body: inputReading() }, context) + + assert.equal(context.statusCode, 502) + assert.match(context.body, /connection refused/) +}) + +function inputReading () { + return { + device_id: 'pump-17', + temperature_c: 82.4, + battery_percent: 12 + } +} + +function jsonResponse (body) { + return new Response(JSON.stringify(body), { + status: 200, + headers: { 'Content-Type': 'application/json' } + }) +} + +class TestContext { + constructor () { + this.statusCode = 200 + this.headerValues = {} + this.body = undefined + } + + status (value) { + this.statusCode = value + return this + } + + headers (value) { + this.headerValues = value + return this + } + + succeed (value) { + this.body = value + return value + } +} diff --git a/node/director/telemetry-workflow/package.json b/node/director/telemetry-workflow/package.json new file mode 100644 index 0000000..7987483 --- /dev/null +++ b/node/director/telemetry-workflow/package.json @@ -0,0 +1,8 @@ +{ + "name": "telemetry-workflow", + "version": "1.0.0", + "private": true, + "scripts": { + "test": "node --test" + } +} diff --git a/node/director/temperature-check/handler.js b/node/director/temperature-check/handler.js new file mode 100644 index 0000000..2d88b3f --- /dev/null +++ b/node/director/temperature-check/handler.js @@ -0,0 +1,40 @@ +'use strict' + +const thresholdC = 75.0 + +module.exports = async (event, context) => { + let reading + try { + reading = parseBody(event.body) + } catch (error) { + return fail(context) + } + + const value = reading.temperature_c + if (!Number.isFinite(value)) { + return fail(context) + } + + return context + .status(200) + .headers({ 'Content-Type': 'application/json' }) + .succeed({ + value_c: value, + threshold_c: thresholdC, + alert: value > thresholdC + }) +} + +function parseBody (body) { + if (Buffer.isBuffer(body)) { + return JSON.parse(body.toString()) + } + return typeof body === 'string' ? JSON.parse(body) : body +} + +function fail (context) { + return context + .status(400) + .headers({ 'Content-Type': 'text/plain' }) + .succeed('expected a JSON sensor reading') +} diff --git a/node/director/temperature-check/handler.test.js b/node/director/temperature-check/handler.test.js new file mode 100644 index 0000000..c6b2e6f --- /dev/null +++ b/node/director/temperature-check/handler.test.js @@ -0,0 +1,35 @@ +'use strict' + +const assert = require('node:assert/strict') +const { test } = require('node:test') + +const handler = require('./handler') + +test('raises an alert above the temperature threshold', async () => { + const context = new TestContext() + await handler({ body: { temperature_c: 82.4 } }, context) + + assert.equal(context.statusCode, 200) + assert.deepEqual(context.body, { + value_c: 82.4, + threshold_c: 75, + alert: true + }) +}) + +class TestContext { + status (value) { + this.statusCode = value + return this + } + + headers (value) { + this.headerValues = value + return this + } + + succeed (value) { + this.body = value + return value + } +} diff --git a/node/director/temperature-check/package.json b/node/director/temperature-check/package.json new file mode 100644 index 0000000..9b7d7ee --- /dev/null +++ b/node/director/temperature-check/package.json @@ -0,0 +1,8 @@ +{ + "name": "temperature-check", + "version": "1.0.0", + "private": true, + "scripts": { + "test": "node --test" + } +} diff --git a/node/director/validate-reading/handler.js b/node/director/validate-reading/handler.js new file mode 100644 index 0000000..2859154 --- /dev/null +++ b/node/director/validate-reading/handler.js @@ -0,0 +1,57 @@ +'use strict' + +module.exports = async (event, context) => { + let reading + try { + reading = parseBody(event.body) + } catch (error) { + return fail(context, 'expected a JSON sensor reading') + } + + const deviceID = String(reading.device_id || '').trim() + const temperature = reading.temperature_c ?? 0 + const battery = reading.battery_percent ?? 0 + + if (!deviceID) { + return fail(context, 'device_id is required') + } + if ( + !Number.isFinite(temperature) || + temperature < -100 || + temperature > 200 + ) { + return fail( + context, + 'temperature_c must be between -100 and 200' + ) + } + if (!Number.isInteger(battery) || battery < 0 || battery > 100) { + return fail( + context, + 'battery_percent must be between 0 and 100' + ) + } + + return context + .status(200) + .headers({ 'Content-Type': 'application/json' }) + .succeed({ + device_id: deviceID, + temperature_c: temperature, + battery_percent: battery + }) +} + +function parseBody (body) { + if (Buffer.isBuffer(body)) { + return JSON.parse(body.toString()) + } + return typeof body === 'string' ? JSON.parse(body) : body +} + +function fail (context, message) { + return context + .status(400) + .headers({ 'Content-Type': 'text/plain' }) + .succeed(message) +} diff --git a/node/director/validate-reading/handler.test.js b/node/director/validate-reading/handler.test.js new file mode 100644 index 0000000..cba8b11 --- /dev/null +++ b/node/director/validate-reading/handler.test.js @@ -0,0 +1,63 @@ +'use strict' + +const assert = require('node:assert/strict') +const { test } = require('node:test') + +const handler = require('./handler') + +test('normalizes a valid reading', async () => { + const context = new TestContext() + await handler({ + body: { + device_id: ' pump-17 ', + temperature_c: 42.5, + battery_percent: 80 + } + }, context) + + assert.equal(context.statusCode, 200) + assert.deepEqual(context.body, { + device_id: 'pump-17', + temperature_c: 42.5, + battery_percent: 80 + }) +}) + +test('rejects a missing device ID', async () => { + const context = new TestContext() + await handler({ body: { temperature_c: 20 } }, context) + + assert.equal(context.statusCode, 400) + assert.match(context.body, /device_id is required/) +}) + +test('rejects values outside the accepted ranges', async () => { + const context = new TestContext() + await handler({ + body: { + device_id: 'pump-17', + temperature_c: 250, + battery_percent: 80 + } + }, context) + + assert.equal(context.statusCode, 400) + assert.match(context.body, /temperature_c/) +}) + +class TestContext { + status (value) { + this.statusCode = value + return this + } + + headers (value) { + this.headerValues = value + return this + } + + succeed (value) { + this.body = value + return value + } +} diff --git a/node/director/validate-reading/package.json b/node/director/validate-reading/package.json new file mode 100644 index 0000000..8ef5426 --- /dev/null +++ b/node/director/validate-reading/package.json @@ -0,0 +1,8 @@ +{ + "name": "validate-reading", + "version": "1.0.0", + "private": true, + "scripts": { + "test": "node --test" + } +} diff --git a/node/fan-out/README.md b/node/fan-out/README.md new file mode 100644 index 0000000..8c3b092 --- /dev/null +++ b/node/fan-out/README.md @@ -0,0 +1,12 @@ +# Fan-out pattern — URL health checks + +This is the Node.js implementation of the fan-out example documented in the +OpenFaaS function patterns guide. It submits each URL asynchronously and can +forward an optional callback URL for individual results. + +Documentation: [Fan-out pattern](https://docs.openfaas.com/languages/patterns/fan-out/) + +```bash +faas-cli template store pull node24 +faas-cli up --tag=digest +``` diff --git a/node/fan-out/fan-out/handler.js b/node/fan-out/fan-out/handler.js new file mode 100644 index 0000000..1c3eb38 --- /dev/null +++ b/node/fan-out/fan-out/handler.js @@ -0,0 +1,101 @@ +'use strict' + +const targetFunction = 'url-check' +const submitTimeout = 30000 + +module.exports = async (event, context) => { + const input = requestBody(event.body) + const records = input + .trim() + .split('\n') + .map((record) => record.trim()) + .filter(Boolean) + + if (records.length === 0) { + return fail( + context, + 400, + 'expected one record per line in the request body' + ) + } + + const gateway = process.env.gateway_url || + 'http://gateway.openfaas:8080' + const headers = event.headers || {} + const callback = String( + headers['x-callback-url'] || process.env.callback_url || '' + ).trim() + + const callIDs = [] + for (const [index, record] of records.entries()) { + let callID + try { + callID = await submit(gateway, record, callback) + } catch (error) { + return fail( + context, + 502, + `record ${index + 1} of ${records.length}: ${error.message}` + ) + } + if (callID) { + callIDs.push(callID) + } + } + + const response = { + submitted: records.length, + function: targetFunction, + callback: Boolean(callback) + } + if (callIDs.length > 0) { + response.call_ids = callIDs + } + + return context + .status(200) + .headers({ 'Content-Type': 'application/json' }) + .succeed(response) +} + +async function submit (gateway, record, callback) { + const headers = { 'Content-Type': 'text/plain' } + if (callback) { + headers['X-Callback-Url'] = callback + } + + const baseURL = gateway.replace(/\/$/, '') + const response = await fetch( + `${baseURL}/async-function/${targetFunction}`, + { + method: 'POST', + headers, + body: record, + signal: AbortSignal.timeout(submitTimeout) + } + ) + + if (response.status !== 202) { + const body = await response.text() + throw new Error( + `unexpected status ${response.status} ` + + `from ${targetFunction}: ${body}` + ) + } + + return response.headers.get('X-Call-Id') || '' +} + +function requestBody (body) { + if (Buffer.isBuffer(body)) { + return body.toString() + } + return String(body || '') +} + +function fail (context, status, message) { + return context + .status(status) + .headers({ 'Content-Type': 'text/plain' }) + .succeed(message) +} diff --git a/node/fan-out/fan-out/handler.test.js b/node/fan-out/fan-out/handler.test.js new file mode 100644 index 0000000..22e6075 --- /dev/null +++ b/node/fan-out/fan-out/handler.test.js @@ -0,0 +1,85 @@ +'use strict' + +const assert = require('node:assert/strict') +const { afterEach, test } = require('node:test') + +const handler = require('./handler') + +const originalFetch = global.fetch + +afterEach(() => { + global.fetch = originalFetch +}) + +test('submits one asynchronous invocation per URL', async () => { + const requests = [] + global.fetch = async (url, options) => { + requests.push({ url, options }) + return new Response('', { + status: 202, + headers: { 'X-Call-Id': `call-${requests.length}` } + }) + } + + const context = new TestContext() + await handler({ + body: 'https://one.example\nhttps://two.example\n', + headers: {} + }, context) + + assert.equal(context.statusCode, 200) + assert.equal(context.body.submitted, 2) + assert.deepEqual(context.body.call_ids, ['call-1', 'call-2']) + assert.equal(requests[0].options.body, 'https://one.example') +}) + +test('forwards the callback URL to every invocation', async () => { + const requests = [] + global.fetch = async (url, options) => { + requests.push(options) + return new Response('', { status: 202 }) + } + + const context = new TestContext() + await handler({ + body: 'https://one.example', + headers: { + 'x-callback-url': 'https://results.example/callback' + } + }, context) + + assert.equal(context.body.callback, true) + assert.equal( + requests[0].headers['X-Callback-Url'], + 'https://results.example/callback' + ) +}) + +test('returns bad gateway when a submission is rejected', async () => { + global.fetch = async () => new Response('queue unavailable', { + status: 503 + }) + + const context = new TestContext() + await handler({ body: 'https://one.example', headers: {} }, context) + + assert.equal(context.statusCode, 502) + assert.match(context.body, /queue unavailable/) +}) + +class TestContext { + status (value) { + this.statusCode = value + return this + } + + headers (value) { + this.headerValues = value + return this + } + + succeed (value) { + this.body = value + return value + } +} diff --git a/node/fan-out/fan-out/package.json b/node/fan-out/fan-out/package.json new file mode 100644 index 0000000..b4e878c --- /dev/null +++ b/node/fan-out/fan-out/package.json @@ -0,0 +1,8 @@ +{ + "name": "fan-out", + "version": "1.0.0", + "private": true, + "scripts": { + "test": "node --test" + } +} diff --git a/node/fan-out/stack.yaml b/node/fan-out/stack.yaml new file mode 100644 index 0000000..e13e8a4 --- /dev/null +++ b/node/fan-out/stack.yaml @@ -0,0 +1,16 @@ +version: 1.0 +provider: + name: openfaas + gateway: http://127.0.0.1:8080 +functions: + fan-out: + lang: node24 + handler: ./fan-out + image: ${REGISTRY:-ttl.sh}/${OWNER:-openfaas-examples}/node-fan-out:latest + + url-check: + lang: node24 + handler: ./url-check + image: ${REGISTRY:-ttl.sh}/${OWNER:-openfaas-examples}/node-url-check:latest + environment: + request_timeout: "5" diff --git a/node/fan-out/url-check/handler.js b/node/fan-out/url-check/handler.js new file mode 100644 index 0000000..d70d76f --- /dev/null +++ b/node/fan-out/url-check/handler.js @@ -0,0 +1,84 @@ +'use strict' + +const { performance } = require('node:perf_hooks') + +const maxURLLength = 4096 +const requestTimeout = configuredTimeout( + process.env.request_timeout || '5', + 'request_timeout' +) + +module.exports = async (event, context) => { + const target = requestBody(event.body).trim() + if (Buffer.byteLength(target) > maxURLLength) { + return fail(context, 'URL is too long') + } + + let parsed + try { + parsed = new URL(target) + } catch (error) { + return fail(context, 'expected an absolute HTTP or HTTPS URL') + } + if (parsed.protocol !== 'http:' && parsed.protocol !== 'https:') { + return fail(context, 'expected an absolute HTTP or HTTPS URL') + } + + const started = performance.now() + const result = { + url: target, + reachable: false, + healthy: false + } + + let response + try { + response = await fetch(target, { + headers: { 'User-Agent': 'OpenFaaS URL health check' }, + signal: AbortSignal.timeout(requestTimeout) + }) + } catch (error) { + result.duration_ms = Math.round(performance.now() - started) + result.error = error.message + return succeed(context, result) + } + + if (response.body) { + await response.body.cancel() + } + result.reachable = true + result.healthy = response.status >= 200 && response.status < 400 + result.status_code = response.status + result.content_type = response.headers.get('Content-Type') || '' + result.duration_ms = Math.round(performance.now() - started) + return succeed(context, result) +} + +function requestBody (body) { + if (Buffer.isBuffer(body)) { + return body.toString() + } + return String(body || '') +} + +function configuredTimeout (value, name) { + const seconds = Number(value) + if (!Number.isFinite(seconds) || seconds <= 0) { + throw new Error(`${name} must be greater than zero`) + } + return seconds * 1000 +} + +function succeed (context, body) { + return context + .status(200) + .headers({ 'Content-Type': 'application/json' }) + .succeed(body) +} + +function fail (context, message) { + return context + .status(400) + .headers({ 'Content-Type': 'text/plain' }) + .succeed(message) +} diff --git a/node/fan-out/url-check/handler.test.js b/node/fan-out/url-check/handler.test.js new file mode 100644 index 0000000..79838dc --- /dev/null +++ b/node/fan-out/url-check/handler.test.js @@ -0,0 +1,74 @@ +'use strict' + +const assert = require('node:assert/strict') +const { afterEach, test } = require('node:test') + +const handler = require('./handler') + +const originalFetch = global.fetch + +afterEach(() => { + global.fetch = originalFetch +}) + +test('reports a healthy URL', async () => { + global.fetch = async () => new Response(null, { + status: 204, + headers: { 'Content-Type': 'text/plain' } + }) + + const context = new TestContext() + await handler({ body: 'https://www.openfaas.com/' }, context) + + assert.equal(context.statusCode, 200) + assert.equal(context.body.reachable, true) + assert.equal(context.body.healthy, true) +}) + +test('reports an unhealthy HTTP status', async () => { + global.fetch = async () => new Response('unavailable', { status: 503 }) + + const context = new TestContext() + await handler({ body: 'https://one.example' }, context) + + assert.equal(context.body.reachable, true) + assert.equal(context.body.healthy, false) + assert.equal(context.body.status_code, 503) +}) + +test('returns network failures as a health result', async () => { + global.fetch = async () => { + throw new Error('request timed out') + } + + const context = new TestContext() + await handler({ body: 'https://one.example' }, context) + + assert.equal(context.statusCode, 200) + assert.equal(context.body.reachable, false) + assert.match(context.body.error, /timed out/) +}) + +test('rejects a non-HTTP URL', async () => { + const context = new TestContext() + await handler({ body: 'file:///etc/passwd' }, context) + + assert.equal(context.statusCode, 400) +}) + +class TestContext { + status (value) { + this.statusCode = value + return this + } + + headers (value) { + this.headerValues = value + return this + } + + succeed (value) { + this.body = value + return value + } +} diff --git a/node/fan-out/url-check/package.json b/node/fan-out/url-check/package.json new file mode 100644 index 0000000..4c75caa --- /dev/null +++ b/node/fan-out/url-check/package.json @@ -0,0 +1,8 @@ +{ + "name": "url-check", + "version": "1.0.0", + "private": true, + "scripts": { + "test": "node --test" + } +} diff --git a/python/director/README.md b/python/director/README.md new file mode 100644 index 0000000..7cf0526 --- /dev/null +++ b/python/director/README.md @@ -0,0 +1,19 @@ +# Director pattern — telemetry triage + +This is the Python implementation of the telemetry workflow documented in the +OpenFaaS function patterns guide. It uses `python3-http` for all four +functions and runs the temperature and battery checks concurrently. + +Documentation: [Director pattern](https://docs.openfaas.com/languages/patterns/director/) + +```bash +faas-cli template store pull python3-http +faas-cli up --tag=digest +``` + +```bash +curl -s http://127.0.0.1:8080/function/telemetry-workflow \ + -H "Content-Type: application/json" \ + -d '{"device_id":"pump-17","temperature_c":82.4,"battery_percent":12}' | \ + jq +``` diff --git a/python/director/battery-check/__init__.py b/python/director/battery-check/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/python/director/battery-check/handler.py b/python/director/battery-check/handler.py new file mode 100644 index 0000000..46103dc --- /dev/null +++ b/python/director/battery-check/handler.py @@ -0,0 +1,21 @@ +import json + + +THRESHOLD_PERCENT = 20 + + +def handle(event, context): + try: + reading = json.loads(event.body) + value = reading["battery_percent"] + except (KeyError, TypeError, ValueError): + return {"statusCode": 400, "body": "expected a JSON sensor reading"} + + return { + "statusCode": 200, + "body": { + "value_percent": value, + "threshold_percent": THRESHOLD_PERCENT, + "alert": value < THRESHOLD_PERCENT, + }, + } diff --git a/python/director/battery-check/handler_test.py b/python/director/battery-check/handler_test.py new file mode 100644 index 0000000..ccb24c1 --- /dev/null +++ b/python/director/battery-check/handler_test.py @@ -0,0 +1,16 @@ +import json +from types import SimpleNamespace + +try: + from . import handler as h +except ImportError: + import handler as h + + +def test_sets_alert_below_threshold(): + event = SimpleNamespace(body=json.dumps({"battery_percent": 12}).encode()) + response = h.handle(event, {}) + + assert response["statusCode"] == 200 + assert response["body"]["alert"] is True + assert response["body"]["threshold_percent"] == 20 diff --git a/python/director/battery-check/requirements.txt b/python/director/battery-check/requirements.txt new file mode 100644 index 0000000..442bb69 --- /dev/null +++ b/python/director/battery-check/requirements.txt @@ -0,0 +1 @@ +# No additional dependencies. diff --git a/python/director/battery-check/tox.ini b/python/director/battery-check/tox.ini new file mode 100644 index 0000000..93559a5 --- /dev/null +++ b/python/director/battery-check/tox.ini @@ -0,0 +1,22 @@ +[tox] +envlist = lint,test +skipsdist = true + +[testenv:test] +deps = + flask + pytest + -rrequirements.txt +commands = pytest + +[testenv:lint] +deps = flake8 +commands = flake8 . + +[flake8] +count = true +max-line-length = 127 +max-complexity = 10 +statistics = true +select = E9,F63,F7,F82 +show-source = true diff --git a/python/director/stack.yaml b/python/director/stack.yaml new file mode 100644 index 0000000..59e1640 --- /dev/null +++ b/python/director/stack.yaml @@ -0,0 +1,37 @@ +version: 1.0 +provider: + name: openfaas + gateway: http://127.0.0.1:8080 +functions: + telemetry-workflow: + lang: python3-http + handler: ./telemetry-workflow + image: ${REGISTRY:-ttl.sh}/${OWNER:-openfaas-examples}/python-telemetry-workflow:latest + build_args: + TEST_ENABLED: "true" + environment: + stage_timeout: "5" + exec_timeout: 15s + read_timeout: 16s + write_timeout: 16s + + validate-reading: + lang: python3-http + handler: ./validate-reading + image: ${REGISTRY:-ttl.sh}/${OWNER:-openfaas-examples}/python-validate-reading:latest + build_args: + TEST_ENABLED: "true" + + temperature-check: + lang: python3-http + handler: ./temperature-check + image: ${REGISTRY:-ttl.sh}/${OWNER:-openfaas-examples}/python-temperature-check:latest + build_args: + TEST_ENABLED: "true" + + battery-check: + lang: python3-http + handler: ./battery-check + image: ${REGISTRY:-ttl.sh}/${OWNER:-openfaas-examples}/python-battery-check:latest + build_args: + TEST_ENABLED: "true" diff --git a/python/director/telemetry-workflow/__init__.py b/python/director/telemetry-workflow/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/python/director/telemetry-workflow/handler.py b/python/director/telemetry-workflow/handler.py new file mode 100644 index 0000000..eedcd95 --- /dev/null +++ b/python/director/telemetry-workflow/handler.py @@ -0,0 +1,87 @@ +import concurrent.futures +import os +import time + +import requests + + +GATEWAY_URL = os.getenv( + "gateway_url", "http://gateway.openfaas:8080" +).rstrip("/") +STAGE_TIMEOUT = float(os.getenv("stage_timeout", "5")) +if STAGE_TIMEOUT <= 0: + raise ValueError("stage_timeout must be greater than zero") + + +def handle(event, context): + started = time.monotonic() + body = event.body + + try: + validated = invoke("validate-reading", body) + except requests.RequestException as err: + return error(502, f"failed to invoke validate-reading: {err}") + + if validated.status_code != 200: + return error( + validated.status_code, + f"validate-reading failed: {validated.text}", + ) + + try: + reading = validated.json() + except ValueError as err: + return error(502, f"unexpected response from validate-reading: {err}") + + functions = ("temperature-check", "battery-check") + completed = {} + with concurrent.futures.ThreadPoolExecutor(max_workers=2) as executor: + futures = { + name: executor.submit(invoke, name, validated.content) + for name in functions + } + for name, future in futures.items(): + try: + completed[name] = future.result() + except requests.RequestException as err: + return error(502, f"failed to invoke {name}: {err}") + + for name in functions: + response = completed[name] + if response.status_code != 200: + return error( + response.status_code, + f"{name} failed: {response.text}", + ) + + try: + temperature = completed["temperature-check"].json() + battery = completed["battery-check"].json() + except ValueError as err: + return error(502, f"unexpected response from check function: {err}") + + has_alert = temperature["alert"] or battery["alert"] + workflow_status = "alert" if has_alert else "ok" + return { + "statusCode": 200, + "body": { + "device_id": reading["device_id"], + "status": workflow_status, + "temperature": temperature, + "battery": battery, + "duration_ms": int((time.monotonic() - started) * 1000), + }, + } + + +def invoke(function, body): + return requests.post( + f"{GATEWAY_URL}/function/{function}", + data=body, + headers={"Content-Type": "application/json"}, + timeout=STAGE_TIMEOUT, + ) + + +def error(status_code, message): + return {"statusCode": status_code, "body": message.strip()} diff --git a/python/director/telemetry-workflow/handler_test.py b/python/director/telemetry-workflow/handler_test.py new file mode 100644 index 0000000..448cd41 --- /dev/null +++ b/python/director/telemetry-workflow/handler_test.py @@ -0,0 +1,77 @@ +import json +from types import SimpleNamespace + +import requests + +try: + from . import handler as h +except ImportError: + import handler as h + + +class FakeResponse: + def __init__(self, status_code, body): + self.status_code = status_code + self._body = body + self.content = json.dumps(body).encode() if isinstance(body, dict) else body.encode() + self.text = self.content.decode() + + def json(self): + return json.loads(self.content) + + +def event(body): + return SimpleNamespace(body=json.dumps(body).encode()) + + +def test_combines_parallel_results(monkeypatch): + responses = { + "validate-reading": FakeResponse( + 200, + {"device_id": "pump-17", "temperature_c": 82.4, "battery_percent": 12}, + ), + "temperature-check": FakeResponse( + 200, + {"value_c": 82.4, "threshold_c": 75.0, "alert": True}, + ), + "battery-check": FakeResponse( + 200, + {"value_percent": 12, "threshold_percent": 20, "alert": True}, + ), + } + monkeypatch.setattr(h, "invoke", lambda name, body: responses[name]) + + response = h.handle( + event({"device_id": "pump-17", "temperature_c": 82.4, "battery_percent": 12}), + {}, + ) + + assert response["statusCode"] == 200 + assert response["body"]["status"] == "alert" + assert response["body"]["temperature"]["alert"] is True + assert response["body"]["battery"]["alert"] is True + + +def test_stops_after_validation_failure(monkeypatch): + calls = [] + + def fake_invoke(name, body): + calls.append(name) + return FakeResponse(400, "device_id is required") + + monkeypatch.setattr(h, "invoke", fake_invoke) + response = h.handle(event({}), {}) + + assert response["statusCode"] == 400 + assert calls == ["validate-reading"] + + +def test_reports_transport_failure(monkeypatch): + def fake_invoke(name, body): + raise requests.ConnectionError("gateway unavailable") + + monkeypatch.setattr(h, "invoke", fake_invoke) + response = h.handle(event({"device_id": "pump-17"}), {}) + + assert response["statusCode"] == 502 + assert "validate-reading" in response["body"] diff --git a/python/director/telemetry-workflow/requirements.txt b/python/director/telemetry-workflow/requirements.txt new file mode 100644 index 0000000..f229360 --- /dev/null +++ b/python/director/telemetry-workflow/requirements.txt @@ -0,0 +1 @@ +requests diff --git a/python/director/telemetry-workflow/tox.ini b/python/director/telemetry-workflow/tox.ini new file mode 100644 index 0000000..93559a5 --- /dev/null +++ b/python/director/telemetry-workflow/tox.ini @@ -0,0 +1,22 @@ +[tox] +envlist = lint,test +skipsdist = true + +[testenv:test] +deps = + flask + pytest + -rrequirements.txt +commands = pytest + +[testenv:lint] +deps = flake8 +commands = flake8 . + +[flake8] +count = true +max-line-length = 127 +max-complexity = 10 +statistics = true +select = E9,F63,F7,F82 +show-source = true diff --git a/python/director/temperature-check/__init__.py b/python/director/temperature-check/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/python/director/temperature-check/handler.py b/python/director/temperature-check/handler.py new file mode 100644 index 0000000..ffe52d5 --- /dev/null +++ b/python/director/temperature-check/handler.py @@ -0,0 +1,21 @@ +import json + + +THRESHOLD_C = 75.0 + + +def handle(event, context): + try: + reading = json.loads(event.body) + value = reading["temperature_c"] + except (KeyError, TypeError, ValueError): + return {"statusCode": 400, "body": "expected a JSON sensor reading"} + + return { + "statusCode": 200, + "body": { + "value_c": value, + "threshold_c": THRESHOLD_C, + "alert": value > THRESHOLD_C, + }, + } diff --git a/python/director/temperature-check/handler_test.py b/python/director/temperature-check/handler_test.py new file mode 100644 index 0000000..e6912c9 --- /dev/null +++ b/python/director/temperature-check/handler_test.py @@ -0,0 +1,16 @@ +import json +from types import SimpleNamespace + +try: + from . import handler as h +except ImportError: + import handler as h + + +def test_sets_alert_above_threshold(): + event = SimpleNamespace(body=json.dumps({"temperature_c": 82.4}).encode()) + response = h.handle(event, {}) + + assert response["statusCode"] == 200 + assert response["body"]["alert"] is True + assert response["body"]["threshold_c"] == 75.0 diff --git a/python/director/temperature-check/requirements.txt b/python/director/temperature-check/requirements.txt new file mode 100644 index 0000000..442bb69 --- /dev/null +++ b/python/director/temperature-check/requirements.txt @@ -0,0 +1 @@ +# No additional dependencies. diff --git a/python/director/temperature-check/tox.ini b/python/director/temperature-check/tox.ini new file mode 100644 index 0000000..93559a5 --- /dev/null +++ b/python/director/temperature-check/tox.ini @@ -0,0 +1,22 @@ +[tox] +envlist = lint,test +skipsdist = true + +[testenv:test] +deps = + flask + pytest + -rrequirements.txt +commands = pytest + +[testenv:lint] +deps = flake8 +commands = flake8 . + +[flake8] +count = true +max-line-length = 127 +max-complexity = 10 +statistics = true +select = E9,F63,F7,F82 +show-source = true diff --git a/python/director/validate-reading/__init__.py b/python/director/validate-reading/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/python/director/validate-reading/handler.py b/python/director/validate-reading/handler.py new file mode 100644 index 0000000..3d5ce96 --- /dev/null +++ b/python/director/validate-reading/handler.py @@ -0,0 +1,41 @@ +import json + + +def handle(event, context): + try: + reading = json.loads(event.body) + except (TypeError, ValueError): + return error("expected a JSON sensor reading") + + device_id = str(reading.get("device_id", "")).strip() + temperature = reading.get("temperature_c", 0) + battery = reading.get("battery_percent", 0) + + if not device_id: + return error("device_id is required") + if not is_number(temperature) or temperature < -100 or temperature > 200: + return error("temperature_c must be between -100 and 200") + if ( + not isinstance(battery, int) + or isinstance(battery, bool) + or battery < 0 + or battery > 100 + ): + return error("battery_percent must be between 0 and 100") + + return { + "statusCode": 200, + "body": { + "device_id": device_id, + "temperature_c": temperature, + "battery_percent": battery, + }, + } + + +def is_number(value): + return isinstance(value, (int, float)) and not isinstance(value, bool) + + +def error(message): + return {"statusCode": 400, "body": message} diff --git a/python/director/validate-reading/handler_test.py b/python/director/validate-reading/handler_test.py new file mode 100644 index 0000000..21f32d0 --- /dev/null +++ b/python/director/validate-reading/handler_test.py @@ -0,0 +1,36 @@ +import json +from types import SimpleNamespace + +try: + from . import handler as h +except ImportError: + import handler as h + + +def event(body): + return SimpleNamespace(body=json.dumps(body).encode()) + + +def test_normalizes_valid_reading(): + response = h.handle( + event({"device_id": " pump-17 ", "temperature_c": 48.2, "battery_percent": 78}), + {}, + ) + + assert response["statusCode"] == 200 + assert response["body"]["device_id"] == "pump-17" + + +def test_rejects_invalid_values(): + response = h.handle( + event({"device_id": "pump-17", "temperature_c": 250, "battery_percent": 78}), + {}, + ) + + assert response["statusCode"] == 400 + assert "temperature_c" in response["body"] + + +def test_rejects_invalid_json(): + response = h.handle(SimpleNamespace(body=b"not-json"), {}) + assert response["statusCode"] == 400 diff --git a/python/director/validate-reading/requirements.txt b/python/director/validate-reading/requirements.txt new file mode 100644 index 0000000..442bb69 --- /dev/null +++ b/python/director/validate-reading/requirements.txt @@ -0,0 +1 @@ +# No additional dependencies. diff --git a/python/director/validate-reading/tox.ini b/python/director/validate-reading/tox.ini new file mode 100644 index 0000000..93559a5 --- /dev/null +++ b/python/director/validate-reading/tox.ini @@ -0,0 +1,22 @@ +[tox] +envlist = lint,test +skipsdist = true + +[testenv:test] +deps = + flask + pytest + -rrequirements.txt +commands = pytest + +[testenv:lint] +deps = flake8 +commands = flake8 . + +[flake8] +count = true +max-line-length = 127 +max-complexity = 10 +statistics = true +select = E9,F63,F7,F82 +show-source = true diff --git a/python/fan-out/README.md b/python/fan-out/README.md new file mode 100644 index 0000000..ad04fc0 --- /dev/null +++ b/python/fan-out/README.md @@ -0,0 +1,12 @@ +# Fan-out pattern — URL health checks + +This is the Python implementation of the fan-out example documented in the +OpenFaaS function patterns guide. It submits each URL asynchronously and can +forward an optional callback URL for individual results. + +Documentation: [Fan-out pattern](https://docs.openfaas.com/languages/patterns/fan-out/) + +```bash +faas-cli template store pull python3-http +faas-cli up --tag=digest +``` diff --git a/python/fan-out/fan-out/__init__.py b/python/fan-out/fan-out/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/python/fan-out/fan-out/handler.py b/python/fan-out/fan-out/handler.py new file mode 100644 index 0000000..cd212e9 --- /dev/null +++ b/python/fan-out/fan-out/handler.py @@ -0,0 +1,73 @@ +import os + +import requests + + +TARGET_FUNCTION = "url-check" +SUBMIT_TIMEOUT = 30 + + +def handle(event, context): + body = ( + event.body.decode() + if isinstance(event.body, bytes) + else str(event.body) + ) + records = [ + record.strip() + for record in body.strip().splitlines() + if record.strip() + ] + if not records: + return error(400, "expected one record per line in the request body") + + gateway = os.getenv("gateway_url", "http://gateway.openfaas:8080") + callback = event.headers.get("X-Callback-Url", "").strip() + if not callback: + callback = os.getenv("callback_url", "").strip() + + call_ids = [] + for index, record in enumerate(records): + try: + call_id = submit(gateway, record, callback) + except requests.RequestException as err: + return error(502, f"record {index + 1} of {len(records)}: {err}") + except RuntimeError as err: + return error(502, f"record {index + 1} of {len(records)}: {err}") + + if call_id: + call_ids.append(call_id) + + response = { + "submitted": len(records), + "function": TARGET_FUNCTION, + "callback": bool(callback), + } + if call_ids: + response["call_ids"] = call_ids + + return {"statusCode": 200, "body": response} + + +def submit(gateway, record, callback): + headers = {"Content-Type": "text/plain"} + if callback: + headers["X-Callback-Url"] = callback + + response = requests.post( + f"{gateway.rstrip('/')}/async-function/{TARGET_FUNCTION}", + data=record.encode(), + headers=headers, + timeout=SUBMIT_TIMEOUT, + ) + if response.status_code != 202: + raise RuntimeError( + f"unexpected status {response.status_code} " + f"from {TARGET_FUNCTION}: {response.text}" + ) + + return response.headers.get("X-Call-Id", "") + + +def error(status_code, message): + return {"statusCode": status_code, "body": message} diff --git a/python/fan-out/fan-out/handler_test.py b/python/fan-out/fan-out/handler_test.py new file mode 100644 index 0000000..2d63133 --- /dev/null +++ b/python/fan-out/fan-out/handler_test.py @@ -0,0 +1,55 @@ +from types import SimpleNamespace + +try: + from . import handler as h +except ImportError: + import handler as h + + +class FakeResponse: + status_code = 202 + text = "" + headers = {"X-Call-Id": "call-1"} + + +def event(body, callback=""): + headers = {"X-Callback-Url": callback} if callback else {} + return SimpleNamespace(body=body.encode(), headers=headers) + + +def test_submits_each_url_with_callback(monkeypatch): + submitted = [] + + def fake_post(url, data, headers, timeout): + submitted.append((url, data.decode(), headers, timeout)) + return FakeResponse() + + monkeypatch.setattr(h.requests, "post", fake_post) + response = h.handle( + event( + "https://one.example\nhttps://two.example\n", + "http://gateway.openfaas:8080/function/printer", + ), + {}, + ) + + assert response["statusCode"] == 200 + assert response["body"]["submitted"] == 2 + assert response["body"]["callback"] is True + assert len(submitted) == 2 + assert submitted[0][2]["X-Callback-Url"].endswith("/function/printer") + + +def test_rejects_empty_batch(): + response = h.handle(event("\n\n"), {}) + assert response["statusCode"] == 400 + + +def test_reports_submission_failure(monkeypatch): + failed = FakeResponse() + failed.status_code = 503 + failed.text = "queue unavailable" + monkeypatch.setattr(h.requests, "post", lambda *args, **kwargs: failed) + + response = h.handle(event("https://one.example\n"), {}) + assert response["statusCode"] == 502 diff --git a/python/fan-out/fan-out/requirements.txt b/python/fan-out/fan-out/requirements.txt new file mode 100644 index 0000000..f229360 --- /dev/null +++ b/python/fan-out/fan-out/requirements.txt @@ -0,0 +1 @@ +requests diff --git a/python/fan-out/fan-out/tox.ini b/python/fan-out/fan-out/tox.ini new file mode 100644 index 0000000..93559a5 --- /dev/null +++ b/python/fan-out/fan-out/tox.ini @@ -0,0 +1,22 @@ +[tox] +envlist = lint,test +skipsdist = true + +[testenv:test] +deps = + flask + pytest + -rrequirements.txt +commands = pytest + +[testenv:lint] +deps = flake8 +commands = flake8 . + +[flake8] +count = true +max-line-length = 127 +max-complexity = 10 +statistics = true +select = E9,F63,F7,F82 +show-source = true diff --git a/python/fan-out/stack.yaml b/python/fan-out/stack.yaml new file mode 100644 index 0000000..4b1f176 --- /dev/null +++ b/python/fan-out/stack.yaml @@ -0,0 +1,20 @@ +version: 1.0 +provider: + name: openfaas + gateway: http://127.0.0.1:8080 +functions: + fan-out: + lang: python3-http + handler: ./fan-out + image: ${REGISTRY:-ttl.sh}/${OWNER:-openfaas-examples}/python-fan-out:latest + build_args: + TEST_ENABLED: "true" + + url-check: + lang: python3-http + handler: ./url-check + image: ${REGISTRY:-ttl.sh}/${OWNER:-openfaas-examples}/python-url-check:latest + build_args: + TEST_ENABLED: "true" + environment: + request_timeout: "5" diff --git a/python/fan-out/url-check/__init__.py b/python/fan-out/url-check/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/python/fan-out/url-check/handler.py b/python/fan-out/url-check/handler.py new file mode 100644 index 0000000..5ab8764 --- /dev/null +++ b/python/fan-out/url-check/handler.py @@ -0,0 +1,61 @@ +import os +import time +from urllib.parse import urlparse + +import requests + + +MAX_URL_LENGTH = 4096 +REQUEST_TIMEOUT = float(os.getenv("request_timeout", "5")) +if REQUEST_TIMEOUT <= 0: + raise ValueError("request_timeout must be greater than zero") + + +def handle(event, context): + body = ( + event.body + if isinstance(event.body, bytes) + else str(event.body).encode() + ) + if len(body) > MAX_URL_LENGTH: + return error("URL is too long") + + target = body.decode().strip() + parsed = urlparse(target) + if parsed.scheme not in ("http", "https") or not parsed.netloc: + return error("expected an absolute HTTP or HTTPS URL") + + started = time.monotonic() + result = { + "url": target, + "reachable": False, + "healthy": False, + } + + try: + with requests.get( + target, + headers={"User-Agent": "OpenFaaS URL health check"}, + timeout=REQUEST_TIMEOUT, + stream=True, + ) as response: + response.raw.read(1024) + result.update( + { + "reachable": True, + "healthy": 200 <= response.status_code < 400, + "status_code": response.status_code, + "content_type": response.headers.get("Content-Type", ""), + "duration_ms": int((time.monotonic() - started) * 1000), + } + ) + except requests.RequestException as err: + result["duration_ms"] = int((time.monotonic() - started) * 1000) + result["error"] = str(err) + return {"statusCode": 200, "body": result} + + return {"statusCode": 200, "body": result} + + +def error(message): + return {"statusCode": 400, "body": message} diff --git a/python/fan-out/url-check/handler_test.py b/python/fan-out/url-check/handler_test.py new file mode 100644 index 0000000..f38234a --- /dev/null +++ b/python/fan-out/url-check/handler_test.py @@ -0,0 +1,63 @@ +from io import BytesIO +from types import SimpleNamespace + +import requests + +try: + from . import handler as h +except ImportError: + import handler as h + + +class FakeResponse: + def __init__(self, status_code, content_type="text/plain"): + self.status_code = status_code + self.headers = {"Content-Type": content_type} + self.raw = BytesIO(b"response body") + self.closed = False + + def __enter__(self): + return self + + def __exit__(self, *args): + self.closed = True + + +def event(target): + return SimpleNamespace(body=target.encode()) + + +def test_reports_healthy_url(monkeypatch): + fetched = FakeResponse(204) + monkeypatch.setattr(h.requests, "get", lambda *args, **kwargs: fetched) + response = h.handle(event("https://www.openfaas.com/"), {}) + + assert response["statusCode"] == 200 + assert response["body"]["reachable"] is True + assert response["body"]["healthy"] is True + assert fetched.closed is True + + +def test_reports_unhealthy_status(monkeypatch): + monkeypatch.setattr(h.requests, "get", lambda *args, **kwargs: FakeResponse(503)) + response = h.handle(event("https://one.example"), {}) + + assert response["body"]["reachable"] is True + assert response["body"]["healthy"] is False + + +def test_reports_network_error(monkeypatch): + def fail(*args, **kwargs): + raise requests.Timeout("request timed out") + + monkeypatch.setattr(h.requests, "get", fail) + response = h.handle(event("https://one.example"), {}) + + assert response["statusCode"] == 200 + assert response["body"]["reachable"] is False + assert "timed out" in response["body"]["error"] + + +def test_rejects_invalid_url(): + response = h.handle(event("file:///etc/passwd"), {}) + assert response["statusCode"] == 400 diff --git a/python/fan-out/url-check/requirements.txt b/python/fan-out/url-check/requirements.txt new file mode 100644 index 0000000..f229360 --- /dev/null +++ b/python/fan-out/url-check/requirements.txt @@ -0,0 +1 @@ +requests diff --git a/python/fan-out/url-check/tox.ini b/python/fan-out/url-check/tox.ini new file mode 100644 index 0000000..93559a5 --- /dev/null +++ b/python/fan-out/url-check/tox.ini @@ -0,0 +1,22 @@ +[tox] +envlist = lint,test +skipsdist = true + +[testenv:test] +deps = + flask + pytest + -rrequirements.txt +commands = pytest + +[testenv:lint] +deps = flake8 +commands = flake8 . + +[flake8] +count = true +max-line-length = 127 +max-complexity = 10 +statistics = true +select = E9,F63,F7,F82 +show-source = true diff --git a/python/singleton/README.md b/python/singleton/README.md new file mode 100644 index 0000000..8ea9107 --- /dev/null +++ b/python/singleton/README.md @@ -0,0 +1,12 @@ +# Singleton pattern — notification hub + +This is the Python implementation of the singleton SSE example documented in +the OpenFaaS function patterns guide. It uses the `python3-flask` template so +the handler can return a streaming Flask response. + +Documentation: [Singleton pattern](https://docs.openfaas.com/languages/patterns/singleton/) + +```bash +faas-cli template store pull python3-flask +faas-cli up --tag=digest +``` diff --git a/python/singleton/notification-hub/__init__.py b/python/singleton/notification-hub/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/python/singleton/notification-hub/handler.py b/python/singleton/notification-hub/handler.py new file mode 100644 index 0000000..ab21acb --- /dev/null +++ b/python/singleton/notification-hub/handler.py @@ -0,0 +1,73 @@ +import json +import queue +import threading + +from flask import Response, request + + +subscribers = set() +subscribers_lock = threading.Lock() +HEARTBEAT_INTERVAL = 15 + + +def handle(req): + if request.method == "GET": + return subscribe() + if request.method == "POST": + return publish(req) + + return "method not allowed", 405, {"Allow": "GET, POST"} + + +def subscribe(): + messages = queue.Queue(maxsize=1) + with subscribers_lock: + subscribers.add(messages) + + def stream(): + try: + yield ": connected\n\n" + while True: + try: + message = messages.get(timeout=HEARTBEAT_INTERVAL) + yield f"data: {message}\n\n" + except queue.Empty: + # Periodic writes let the server detect idle disconnects. + yield ": keep-alive\n\n" + finally: + with subscribers_lock: + subscribers.discard(messages) + + return Response( + stream(), + mimetype="text/event-stream", + headers={"Cache-Control": "no-cache"}, + ) + + +def publish(req): + body = req.decode() if isinstance(req, bytes) else str(req) + message = " ".join(body.splitlines()).strip() + if not message: + return "notification must not be empty", 400 + + delivered = broadcast(message) + return ( + json.dumps({"delivered": delivered}) + "\n", + 200, + {"Content-Type": "application/json"}, + ) + + +def broadcast(message): + delivered = 0 + with subscribers_lock: + for subscriber in subscribers: + try: + subscriber.put_nowait(message) + delivered += 1 + except queue.Full: + # Do not let a slow subscriber block every other client. + pass + + return delivered diff --git a/python/singleton/notification-hub/handler_test.py b/python/singleton/notification-hub/handler_test.py new file mode 100644 index 0000000..8c609a1 --- /dev/null +++ b/python/singleton/notification-hub/handler_test.py @@ -0,0 +1,63 @@ +import json +import queue + +from flask import Flask + +try: + from . import handler as h +except ImportError: + import handler as h + + +app = Flask(__name__) + + +def setup_function(): + with h.subscribers_lock: + h.subscribers.clear() + + +def test_publish_broadcasts_to_subscribers(): + messages = queue.Queue(maxsize=1) + with h.subscribers_lock: + h.subscribers.add(messages) + + with app.test_request_context(method="POST"): + body, status, headers = h.handle("deployment complete") + + assert status == 200 + assert headers["Content-Type"] == "application/json" + assert json.loads(body)["delivered"] == 1 + assert messages.get_nowait() == "deployment complete" + + +def test_subscribe_registers_and_removes_client(): + with app.test_request_context(method="GET"): + response = h.handle("") + stream = response.response + assert next(stream) == ": connected\n\n" + assert len(h.subscribers) == 1 + stream.close() + + assert len(h.subscribers) == 0 + + +def test_subscribe_sends_heartbeat_while_idle(monkeypatch): + monkeypatch.setattr(h, "HEARTBEAT_INTERVAL", 0) + + with app.test_request_context(method="GET"): + response = h.handle("") + stream = response.response + assert next(stream) == ": connected\n\n" + assert next(stream) == ": keep-alive\n\n" + stream.close() + + assert len(h.subscribers) == 0 + + +def test_rejects_empty_notification(): + with app.test_request_context(method="POST"): + body, status = h.handle("\n") + + assert status == 400 + assert "must not be empty" in body diff --git a/python/singleton/notification-hub/requirements.txt b/python/singleton/notification-hub/requirements.txt new file mode 100644 index 0000000..48f1624 --- /dev/null +++ b/python/singleton/notification-hub/requirements.txt @@ -0,0 +1 @@ +# Flask is provided by the python3-flask template. diff --git a/python/singleton/notification-hub/tox.ini b/python/singleton/notification-hub/tox.ini new file mode 100644 index 0000000..93559a5 --- /dev/null +++ b/python/singleton/notification-hub/tox.ini @@ -0,0 +1,22 @@ +[tox] +envlist = lint,test +skipsdist = true + +[testenv:test] +deps = + flask + pytest + -rrequirements.txt +commands = pytest + +[testenv:lint] +deps = flake8 +commands = flake8 . + +[flake8] +count = true +max-line-length = 127 +max-complexity = 10 +statistics = true +select = E9,F63,F7,F82 +show-source = true diff --git a/python/singleton/stack.yaml b/python/singleton/stack.yaml new file mode 100644 index 0000000..b6ffcd5 --- /dev/null +++ b/python/singleton/stack.yaml @@ -0,0 +1,18 @@ +version: 1.0 +provider: + name: openfaas + gateway: http://127.0.0.1:8080 +functions: + notification-hub: + lang: python3-flask + handler: ./notification-hub + image: ${REGISTRY:-ttl.sh}/${OWNER:-openfaas-examples}/python-notification-hub:latest + build_args: + TEST_ENABLED: "true" + labels: + com.openfaas.scale.min: "1" + com.openfaas.scale.max: "1" + environment: + exec_timeout: "1h" + read_timeout: "1h1s" + write_timeout: "1h1s"