Skip to content

Commit aa898b7

Browse files
committed
Make response messages consistent between modes
Makes the response messages consistent between HTTP, streaming and serializing modes. Tested with various processes as a Go process on a Linux host. Signed-off-by: Alex Ellis (OpenFaaS Ltd) <[email protected]>
1 parent 0b79385 commit aa898b7

File tree

16 files changed

+742
-54
lines changed

16 files changed

+742
-54
lines changed

executor/function_runner.go

Lines changed: 4 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -14,7 +14,10 @@ type FunctionRunner interface {
1414

1515
// FunctionRequest stores request for function execution
1616
type FunctionRequest struct {
17-
Path string
17+
RequestURI string
18+
Method string
19+
UserAgent string
20+
1821
Process string
1922
ProcessArgs []string
2023
Environment []string

executor/http_runner.go

Lines changed: 6 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -9,6 +9,9 @@ import (
99
"fmt"
1010
"io"
1111
"io/ioutil"
12+
13+
units "github.com/docker/go-units"
14+
1215
"log"
1316
"net"
1417
"net/http"
@@ -155,8 +158,9 @@ func (f *HTTPFunctionRunner) Run(req FunctionRequest, contentLength int64, r *ht
155158
}
156159

157160
copyHeaders(w.Header(), &res.Header)
161+
done := time.Since(startedTime)
158162

159-
w.Header().Set("X-Duration-Seconds", fmt.Sprintf("%f", time.Since(startedTime).Seconds()))
163+
w.Header().Set("X-Duration-Seconds", fmt.Sprintf("%f", done.Seconds()))
160164

161165
w.WriteHeader(res.StatusCode)
162166
if res.Body != nil {
@@ -172,7 +176,7 @@ func (f *HTTPFunctionRunner) Run(req FunctionRequest, contentLength int64, r *ht
172176
// Exclude logging for health check probes from the kubelet which can spam
173177
// log collection systems.
174178
if !strings.HasPrefix(r.UserAgent(), "kube-probe") {
175-
log.Printf("%s %s - %s - ContentLength: %d", r.Method, r.RequestURI, res.Status, res.ContentLength)
179+
log.Printf("%s %s - %s - ContentLength: %s (%.4fs)", r.Method, r.RequestURI, res.Status, units.HumanSize(float64(res.ContentLength)), done.Seconds())
176180
}
177181

178182
return nil

executor/serializing_fork_runner.go

Lines changed: 35 additions & 21 deletions
Original file line numberDiff line numberDiff line change
@@ -8,9 +8,13 @@ import (
88
"fmt"
99
"io"
1010
"io/ioutil"
11+
12+
units "github.com/docker/go-units"
13+
1114
"log"
1215
"net/http"
1316
"os/exec"
17+
"strings"
1418
"sync"
1519
"time"
1620
)
@@ -27,30 +31,42 @@ func (f *SerializingForkFunctionRunner) Run(req FunctionRequest, w http.Response
2731
body, err := serializeFunction(req, f)
2832
if err != nil {
2933
w.Header().Set("X-Duration-Seconds", fmt.Sprintf("%f", time.Since(start).Seconds()))
30-
w.WriteHeader(500)
34+
w.WriteHeader(http.StatusInternalServerError)
3135
w.Write([]byte(err.Error()))
36+
37+
done := time.Since(start)
38+
39+
if !strings.HasPrefix(req.UserAgent, "kube-probe") {
40+
log.Printf("%s %s - %d - ContentLength: %s (%.4fs)", req.Method, req.RequestURI, http.StatusOK, units.HumanSize(float64(len(err.Error()))), done.Seconds())
41+
}
42+
3243
return err
3344
}
3445

3546
w.Header().Set("X-Duration-Seconds", fmt.Sprintf("%f", time.Since(start).Seconds()))
3647
w.WriteHeader(200)
3748

49+
bodyLen := 0
3850
if body != nil {
3951
_, err = w.Write(*body)
52+
bodyLen = len(*body)
53+
}
54+
55+
done := time.Since(start)
56+
57+
if !strings.HasPrefix(req.UserAgent, "kube-probe") {
58+
log.Printf("%s %s - %d - ContentLength: %s (%.4fs)", req.Method, req.RequestURI, http.StatusOK, units.HumanSize(float64(bodyLen)), done.Seconds())
4059
}
4160

4261
return err
4362
}
4463

4564
func serializeFunction(req FunctionRequest, f *SerializingForkFunctionRunner) (*[]byte, error) {
46-
log.Printf("Running: %s", req.Process)
4765

4866
if req.InputReader != nil {
4967
defer req.InputReader.Close()
5068
}
5169

52-
start := time.Now()
53-
5470
var cmd *exec.Cmd
5571
ctx := context.Background()
5672
if f.ExecTimeout.Nanoseconds() > 0 {
@@ -60,21 +76,25 @@ func serializeFunction(req FunctionRequest, f *SerializingForkFunctionRunner) (*
6076
}
6177

6278
cmd = exec.CommandContext(ctx, req.Process, req.ProcessArgs...)
79+
cmd.Env = req.Environment
6380

6481
var data []byte
6582

66-
reader := req.InputReader.(io.Reader)
83+
if req.InputReader != nil {
84+
reader := req.InputReader.(io.Reader)
6785

68-
// Limit read to the Content-Length header, if provided
69-
if req.ContentLength != nil && *req.ContentLength > 0 {
70-
reader = io.LimitReader(req.InputReader, *req.ContentLength)
71-
}
86+
// Limit read to the Content-Length header, if provided
87+
if req.ContentLength != nil && *req.ContentLength > 0 {
88+
reader = io.LimitReader(req.InputReader, *req.ContentLength)
89+
}
7290

73-
var err error
74-
data, err = ioutil.ReadAll(reader)
91+
var err error
92+
data, err = ioutil.ReadAll(reader)
93+
94+
if err != nil {
95+
return nil, err
96+
}
7597

76-
if err != nil {
77-
return nil, err
7898
}
7999

80100
stdout, _ := cmd.StdoutPipe()
@@ -89,15 +109,9 @@ func serializeFunction(req FunctionRequest, f *SerializingForkFunctionRunner) (*
89109
return nil, errors[0]
90110
}
91111

92-
err = cmd.Wait()
93-
done := time.Since(start)
94-
if err != nil {
95-
return nil, fmt.Errorf("%s exited: after %.2fs, error: %s", req.Process, done.Seconds(), err)
96-
}
97-
98-
log.Printf("%s done: %.2fs secs", req.Process, done.Seconds())
112+
err := cmd.Wait()
99113

100-
return functionRes, nil
114+
return functionRes, err
101115
}
102116

103117
func pipeToProcess(stdin io.WriteCloser, stdout io.Reader, data *[]byte) (*[]byte, []error) {

executor/forking_runner.go renamed to executor/streaming_runner.go

Lines changed: 1 addition & 13 deletions
Original file line numberDiff line numberDiff line change
@@ -5,8 +5,6 @@ package executor
55

66
import (
77
"context"
8-
"fmt"
9-
"log"
108
"os"
119
"os/exec"
1210
"time"
@@ -21,8 +19,6 @@ type StreamingFunctionRunner struct {
2119

2220
// Run run a fork for each invocation
2321
func (f *StreamingFunctionRunner) Run(req FunctionRequest) error {
24-
log.Printf("Running: %s - %s", req.Process, req.Path)
25-
start := time.Now()
2622

2723
var cmd *exec.Cmd
2824
ctx := context.Background()
@@ -50,13 +46,5 @@ func (f *StreamingFunctionRunner) Run(req FunctionRequest) error {
5046
return err
5147
}
5248

53-
err := cmd.Wait()
54-
done := time.Since(start)
55-
if err != nil {
56-
return fmt.Errorf("%s exited: after %.2fs, error: %s", req.Process, done.Seconds(), err)
57-
}
58-
59-
log.Printf("%s done: %.2fs secs", req.Process, done.Seconds())
60-
61-
return nil
49+
return cmd.Wait()
6250
}

go.mod

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -3,6 +3,7 @@ module github.com/openfaas/of-watchdog
33
go 1.18
44

55
require (
6+
github.com/docker/go-units v0.5.0
67
github.com/openfaas/faas-middleware v1.2.2
78
github.com/prometheus/client_golang v1.13.0
89
)

go.sum

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -54,6 +54,8 @@ github.com/cncf/udpa/go v0.0.0-20191209042840-269d4d468f6f/go.mod h1:M8M6+tZqaGX
5454
github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
5555
github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
5656
github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
57+
github.com/docker/go-units v0.5.0 h1:69rxXcBk27SvSaaxTtLh/8llcHD8vYHT7WSdRZ/jvr4=
58+
github.com/docker/go-units v0.5.0/go.mod h1:fgPhTUdO+D/Jk86RDLlptpiXQzgHJF7gydDDbaIK4Dk=
5759
github.com/envoyproxy/go-control-plane v0.9.0/go.mod h1:YTl/9mNaCwkRvm6d1a2C3ymFceY/DCBVvsKhRF0iEA4=
5860
github.com/envoyproxy/go-control-plane v0.9.1-0.20191026205805-5f8ba28d4473/go.mod h1:YTl/9mNaCwkRvm6d1a2C3ymFceY/DCBVvsKhRF0iEA4=
5961
github.com/envoyproxy/go-control-plane v0.9.4/go.mod h1:6rpuAdCZL397s3pYoYcLgu1mIlRU8Am5FuJP05cCM98=

main.go

Lines changed: 47 additions & 17 deletions
Original file line numberDiff line numberDiff line change
@@ -7,6 +7,7 @@ import (
77
"context"
88
"flag"
99
"fmt"
10+
"io"
1011
"io/ioutil"
1112
"log"
1213
"net/http"
@@ -19,6 +20,8 @@ import (
1920
"syscall"
2021
"time"
2122

23+
units "github.com/docker/go-units"
24+
2225
limiter "github.com/openfaas/faas-middleware/concurrency-limiter"
2326
"github.com/openfaas/of-watchdog/config"
2427
"github.com/openfaas/of-watchdog/executor"
@@ -78,7 +81,7 @@ func main() {
7881
limit = requestLimiter
7982
}
8083

81-
log.Printf("Watchdog mode: %s\n", config.WatchdogMode(watchdogConfig.OperationalMode))
84+
log.Printf("Watchdog mode: %s\tfprocess: %q\n", config.WatchdogMode(watchdogConfig.OperationalMode), watchdogConfig.FunctionProcess)
8285

8386
httpMetrics := metrics.NewHttp()
8487
http.HandleFunc("/", metrics.InstrumentHandler(requestHandler, httpMetrics))
@@ -242,10 +245,6 @@ func makeSerializingForkRequestHandler(watchdogConfig config.WatchdogConfig, log
242245
environment = getEnvironment(r)
243246
}
244247

245-
path := "/"
246-
if r.URL != nil {
247-
path = r.URL.Path
248-
}
249248
commandName, arguments := watchdogConfig.Process()
250249
req := executor.FunctionRequest{
251250
Process: commandName,
@@ -254,7 +253,9 @@ func makeSerializingForkRequestHandler(watchdogConfig config.WatchdogConfig, log
254253
ContentLength: &r.ContentLength,
255254
OutputWriter: w,
256255
Environment: environment,
257-
Path: path,
256+
RequestURI: r.RequestURI,
257+
Method: r.Method,
258+
UserAgent: r.UserAgent(),
258259
}
259260

260261
w.Header().Set("Content-Type", watchdogConfig.ContentType)
@@ -280,18 +281,19 @@ func makeStreamingRequestHandler(watchdogConfig config.WatchdogConfig, prefixLog
280281
environment = getEnvironment(r)
281282
}
282283

283-
path := "/"
284-
if r.URL != nil {
285-
path = r.URL.Path
286-
}
284+
ww := WriterCounter{}
285+
ww.setWriter(w)
286+
start := time.Now()
287287
commandName, arguments := watchdogConfig.Process()
288288
req := executor.FunctionRequest{
289289
Process: commandName,
290290
ProcessArgs: arguments,
291291
InputReader: r.Body,
292-
OutputWriter: w,
292+
OutputWriter: &ww,
293293
Environment: environment,
294-
Path: path,
294+
RequestURI: r.RequestURI,
295+
Method: r.Method,
296+
UserAgent: r.UserAgent(),
295297
}
296298

297299
w.Header().Set("Content-Type", watchdogConfig.ContentType)
@@ -301,6 +303,16 @@ func makeStreamingRequestHandler(watchdogConfig config.WatchdogConfig, prefixLog
301303

302304
// Cannot write a status code to the client because we
303305
// already have written a header
306+
done := time.Since(start)
307+
if !strings.HasPrefix(req.UserAgent, "kube-probe") {
308+
log.Printf("%s %s - %d - ContentLength: %s (%.4fs)", req.Method, req.RequestURI, http.StatusInternalServerError, units.HumanSize(float64(ww.Bytes())), done.Seconds())
309+
return
310+
}
311+
}
312+
313+
done := time.Since(start)
314+
if !strings.HasPrefix(req.UserAgent, "kube-probe") {
315+
log.Printf("%s %s - %d - ContentLength: %s (%.4fs)", req.Method, req.RequestURI, http.StatusOK, units.HumanSize(float64(ww.Bytes())), done.Seconds())
304316
}
305317
}
306318
}
@@ -357,16 +369,11 @@ func makeHTTPRequestHandler(watchdogConfig config.WatchdogConfig, prefixLogs boo
357369

358370
return func(w http.ResponseWriter, r *http.Request) {
359371

360-
path := "/"
361-
if r.URL != nil {
362-
path = r.URL.Path
363-
}
364372
req := executor.FunctionRequest{
365373
Process: commandName,
366374
ProcessArgs: arguments,
367375
InputReader: r.Body,
368376
OutputWriter: w,
369-
Path: path,
370377
}
371378

372379
if r.Body != nil {
@@ -424,3 +431,26 @@ func printVersion() {
424431

425432
log.Printf("Version: %v\tSHA: %v\n", BuildVersion(), sha)
426433
}
434+
435+
type WriterCounter struct {
436+
w io.Writer
437+
bytes int64
438+
}
439+
440+
func (nc *WriterCounter) setWriter(w io.Writer) {
441+
nc.w = w
442+
}
443+
444+
func (nc *WriterCounter) Bytes() int64 {
445+
return nc.bytes
446+
}
447+
448+
func (nc *WriterCounter) Write(p []byte) (int, error) {
449+
n, err := nc.w.Write(p)
450+
if err != nil {
451+
return n, err
452+
}
453+
454+
nc.bytes += int64(n)
455+
return n, err
456+
}

0 commit comments

Comments
 (0)