Feat/executor json log (#694)

Signed-off-by: Pavel Okhlopkov <[email protected]>
Co-authored-by: Pavel Okhlopkov <[email protected]>

go.mod

@@ -3,7 +3,7 @@ module github.com/flant/shell-operator
 go 1.22.8
 
 require (
-	github.com/deckhouse/deckhouse/pkg/log v0.0.0-20241106140903-258b93b3334e
+	github.com/deckhouse/deckhouse/pkg/log v0.0.0-20241205040953-7b376bae249c
 	github.com/flant/kube-client v1.2.2
 	github.com/flant/libjq-go v1.6.3-0.20201126171326-c46a40ff22ee // branch: master
 	github.com/go-chi/chi/v5 v5.1.0

go.sum

@@ -24,8 +24,8 @@ github.com/creack/pty v1.1.9/go.mod h1:oKZEueFk5CKHvIhNR5MUki03XCEU+Q6VDXinZuGJ3
 github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
 github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
-github.com/deckhouse/deckhouse/pkg/log v0.0.0-20241106140903-258b93b3334e h1:QUQy+5Bv7/UzhfrytiG3c5gfLGhPppepVbRpbMisVIw=
-github.com/deckhouse/deckhouse/pkg/log v0.0.0-20241106140903-258b93b3334e/go.mod h1:Mk5HRzkc5pIcDIZ2JJ6DPuuqnwhXVkb3you8M8Mg+4w=
+github.com/deckhouse/deckhouse/pkg/log v0.0.0-20241205040953-7b376bae249c h1:dK30IW9uGg0DvSy+IcdQ6zwEBRV55R7tEtaruEKYkSA=
+github.com/deckhouse/deckhouse/pkg/log v0.0.0-20241205040953-7b376bae249c/go.mod h1:Mk5HRzkc5pIcDIZ2JJ6DPuuqnwhXVkb3you8M8Mg+4w=
 github.com/docker/go-units v0.4.0/go.mod h1:fgPhTUdO+D/Jk86RDLlptpiXQzgHJF7gydDDbaIK4Dk=
 github.com/emicklei/go-restful/v3 v3.11.0 h1:rAQeMHw1c7zTmncogyy8VvRZwtkmkZ4FxERmMY4rD+g=
 github.com/emicklei/go-restful/v3 v3.11.0/go.mod h1:6n3XBCmQQb25CM2LCACGz8ukIrRry+4bhvbpWn3mrbc=

pkg/config/config.go

@@ -268,7 +268,7 @@ func (c *Config) callOnChange(name string, oldValue string, newValue string) {
 	err := c.params[name].onChange(oldValue, newValue)
 	if err != nil {
 		c.logger.Error("OnChange handler failed for parameter during value change values",
-			slog.String("parameter", name), slog.String("old_value", oldValue), slog.String("new_value", newValue), slog.String("error", err.Error()))
+			slog.String("parameter", name), slog.String("old_value", oldValue), slog.String("new_value", newValue), log.Err(err))
 	}
 	c.m.Lock()
 	delete(c.errors, name)

pkg/executor/executor.go

@@ -108,10 +108,10 @@ func (e *Executor) RunAndLogLines(logLabels map[string]string) (*CmdUsage, error
 	err := e.cmd.Run()
 	if err != nil {
 		if len(stdErr.Bytes()) > 0 {
-			return nil, fmt.Errorf("%s", stdErr.String())
+			return nil, fmt.Errorf("stderr: %s", stdErr.String())
 		}
 
-		return nil, err
+		return nil, fmt.Errorf("cmd run: %w", err)
 	}
 
 	var usage *CmdUsage
@@ -129,7 +129,7 @@ func (e *Executor) RunAndLogLines(logLabels map[string]string) (*CmdUsage, error
 		}
 	}
 
-	return usage, err
+	return usage, nil
 }
 
 type proxyLogger struct {
@@ -158,7 +158,10 @@ func (pl *proxyLogger) Write(p []byte) (int, error) {
 			return len(p), nil
 		}
 
-		return len(p), err
+		pl.logger.Debug("output is not json", log.Err(err))
+		pl.writerScanner(p)
+
+		return len(p), nil
 	}
 
 	logMap, ok := line.(map[string]interface{})
@@ -172,26 +175,11 @@ func (pl *proxyLogger) Write(p []byte) (int, error) {
 		// fall back to using the logger
 		pl.logger.Info(string(p))
 
-		return len(p), err
-	}
-
-	logger := pl.logger.With(pl.proxyJsonLogKey, true)
-
-	logLineRaw, _ := json.Marshal(logMap)
-	logLine := string(logLineRaw)
-
-	if len(logLine) > 10000 {
-		logLine = fmt.Sprintf("%s:truncated", logLine[:10000])
-
-		logger.Log(context.Background(), log.LevelFatal.Level(), "hook result", slog.Any("hook", map[string]any{
-			"truncated": logLine,
-		}))
-
 		return len(p), nil
 	}
 
 	// logEntry.Log(log.FatalLevel, string(logLine))
-	logger.Log(context.Background(), log.LevelFatal.Level(), "hook result", slog.Any("hook", logMap))
+	pl.mergeAndLogInputLog(context.TODO(), logMap, "hook")
 
 	return len(p), nil
 }
@@ -232,6 +220,49 @@ func (pl *proxyLogger) writerScanner(p []byte) {
 
 	// If there was an error while scanning the input, log an error
 	if err := scanner.Err(); err != nil {
-		pl.logger.Error("reading from scanner", slog.String("error", err.Error()))
+		pl.logger.Error("reading from scanner", log.Err(err))
+	}
+}
+
+// level = level
+// msg = msg
+// prefix for all fields hook_
+// source = hook_source
+// stacktrace = hook_stacktrace
+func (pl *proxyLogger) mergeAndLogInputLog(ctx context.Context, inputLog map[string]interface{}, prefix string) {
+	var lvl log.Level
+
+	lvlRaw, ok := inputLog[slog.LevelKey].(string)
+	if ok {
+		lvl = log.LogLevelFromStr(lvlRaw)
+		delete(inputLog, slog.LevelKey)
 	}
+
+	msg, ok := inputLog[slog.MessageKey].(string)
+	if !ok {
+		msg = "hook result"
+	}
+	delete(inputLog, slog.MessageKey)
+	delete(inputLog, slog.TimeKey)
+
+	logLineRaw, _ := json.Marshal(inputLog)
+	logLine := string(logLineRaw)
+
+	logger := pl.logger.With(pl.proxyJsonLogKey, true)
+
+	if len(logLine) > 10000 {
+		logLine = fmt.Sprintf("%s:truncated", logLine[:10000])
+
+		logger.Log(ctx, lvl.Level(), msg, slog.Any("hook", map[string]any{
+			"truncated": logLine,
+		}))
+
+		return
+	}
+
+	for key, val := range inputLog {
+		logger = logger.With(slog.Any(prefix+"_"+key, val))
+	}
+
+	logger.Log(ctx, lvl.Level(), msg)
 }

pkg/executor/executor_test.go

@@ -39,7 +39,7 @@ func TestRunAndLogLines(t *testing.T) {
 		_, err := ex.RunAndLogLines(map[string]string{"a": "b"})
 		assert.NoError(t, err)
 
-		assert.Equal(t, buf.String(), `{"level":"fatal","msg":"hook result","a":"b","hook":{"foo":"baz"},"output":"stdout","proxyJsonLog":true,"time":"2006-01-02T15:04:05Z"}`+"\n")
+		assert.Equal(t, buf.String(), `{"level":"info","msg":"hook result","a":"b","hook_foo":"baz","output":"stdout","proxyJsonLog":true,"time":"2006-01-02T15:04:05Z"}`+"\n")
 
 		buf.Reset()
 	})
@@ -71,7 +71,7 @@ func TestRunAndLogLines(t *testing.T) {
 		_, err = ex.RunAndLogLines(map[string]string{"a": "b"})
 		assert.NoError(t, err)
 
-		reg := regexp.MustCompile(`{"level":"fatal","msg":"hook result","a":"b","hook":{"truncated":".*:truncated"},"output":"stdout","proxyJsonLog":true,"time":"2006-01-02T15:04:05Z"`)
+		reg := regexp.MustCompile(`{"level":"info","msg":"hook result","a":"b","hook":{"truncated":".*:truncated"},"output":"stdout","proxyJsonLog":true,"time":"2006-01-02T15:04:05Z"`)
 		assert.Regexp(t, reg, buf.String())
 
 		buf.Reset()
@@ -108,8 +108,8 @@ func TestRunAndLogLines(t *testing.T) {
 		assert.NoError(t, err)
 
 		assert.Equal(t, buf.String(), `{"level":"debug","msg":"Executing command 'echo [\"a\",\"b\",\"c\"]' in '' dir","source":"executor/executor.go:101","a":"b","time":"2006-01-02T15:04:05Z"}`+"\n"+
-			`{"level":"debug","msg":"json log line not map[string]interface{}","source":"executor/executor.go:170","a":"b","line":["a","b","c"],"output":"stdout","time":"2006-01-02T15:04:05Z"}`+"\n"+
-			`{"level":"info","msg":"[\"a\",\"b\",\"c\"]\n","source":"executor/executor.go:173","a":"b","output":"stdout","time":"2006-01-02T15:04:05Z"}`+"\n")
+			`{"level":"debug","msg":"json log line not map[string]interface{}","source":"executor/executor.go:173","a":"b","line":["a","b","c"],"output":"stdout","time":"2006-01-02T15:04:05Z"}`+"\n"+
+			`{"level":"info","msg":"[\"a\",\"b\",\"c\"]\n","source":"executor/executor.go:176","a":"b","output":"stdout","time":"2006-01-02T15:04:05Z"}`+"\n")
 
 		buf.Reset()
 	})
@@ -126,7 +126,7 @@ func TestRunAndLogLines(t *testing.T) {
 
 		_, err := ex.RunAndLogLines(map[string]string{"foor": "baar"})
 		assert.NoError(t, err)
-		assert.Equal(t, buf.String(), `{"level":"fatal","msg":"hook result","foor":"baar","hook":{"a":"b","c":"d"},"output":"stdout","proxyJsonLog":true,"time":"2006-01-02T15:04:05Z"}`+"\n")
+		assert.Equal(t, buf.String(), `{"level":"info","msg":"hook result","foor":"baar","hook_a":"b","hook_c":"d","output":"stdout","proxyJsonLog":true,"time":"2006-01-02T15:04:05Z"}`+"\n")
 
 		buf.Reset()
 	})
@@ -147,6 +147,23 @@ c d
 		buf.Reset()
 	})
 
+	t.Run("multiline non json with json proxy on", func(t *testing.T) {
+		arg := `
+a b
+c d
+`
+		ex := NewExecutor("", "echo", []string{arg}, []string{}).
+			WithLogProxyHookJSON(true).
+			WithLogger(logger)
+
+		_, err := ex.RunAndLogLines(map[string]string{"foor": "baar"})
+		assert.NoError(t, err)
+		assert.Equal(t, buf.String(), `{"level":"info","msg":"a b","foor":"baar","output":"stdout","time":"2006-01-02T15:04:05Z"}`+"\n"+
+			`{"level":"info","msg":"c d","foor":"baar","output":"stdout","time":"2006-01-02T15:04:05Z"}`+"\n")
+
+		buf.Reset()
+	})
+
 	t.Run("multiline json", func(t *testing.T) {
 		arg := `{
 "a":"b",
@@ -158,7 +175,20 @@ c d
 
 		_, err := ex.RunAndLogLines(map[string]string{"foor": "baar"})
 		assert.NoError(t, err)
-		assert.Equal(t, buf.String(), `{"level":"fatal","msg":"hook result","foor":"baar","hook":{"a":"b","c":"d"},"output":"stdout","proxyJsonLog":true,"time":"2006-01-02T15:04:05Z"}`+"\n")
+		assert.Equal(t, buf.String(), `{"level":"info","msg":"hook result","foor":"baar","hook_a":"b","hook_c":"d","output":"stdout","proxyJsonLog":true,"time":"2006-01-02T15:04:05Z"}`+"\n")
+
+		buf.Reset()
+	})
+
+	t.Run("input json nest", func(t *testing.T) {
+		arg := `{"level":"info","msg":"hook result","foor":"baar","a":"b","c":"d","time":"2024-01-02T15:04:05Z"}`
+		ex := NewExecutor("", "echo", []string{arg}, []string{}).
+			WithLogProxyHookJSON(true).
+			WithLogger(logger)
+
+		_, err := ex.RunAndLogLines(map[string]string{"foor": "baar"})
+		assert.NoError(t, err)
+		assert.Equal(t, buf.String(), `{"level":"info","msg":"hook result","foor":"baar","hook_a":"b","hook_c":"d","hook_foor":"baar","output":"stdout","proxyJsonLog":true,"time":"2006-01-02T15:04:05Z"}`+"\n")
 
 		buf.Reset()
 	})

pkg/shell-operator/bootstrap.go

@@ -3,7 +3,6 @@ package shell_operator
 import (
 	"context"
 	"fmt"
-	"log/slog"
 
 	"github.com/deckhouse/deckhouse/pkg/log"
 
@@ -34,13 +33,13 @@ func Init(logger *log.Logger) (*ShellOperator, error) {
 
 	hooksDir, err := utils.RequireExistingDirectory(app.HooksDir)
 	if err != nil {
-		logger.Log(context.TODO(), log.LevelFatal.Level(), "hooks directory is required", slog.String("error", err.Error()))
+		logger.Log(context.TODO(), log.LevelFatal.Level(), "hooks directory is required", log.Err(err))
 		return nil, err
 	}
 
 	tempDir, err := utils.EnsureTempDirectory(app.TempDir)
 	if err != nil {
-		logger.Log(context.TODO(), log.LevelFatal.Level(), "temp directory", slog.String("error", err.Error()))
+		logger.Log(context.TODO(), log.LevelFatal.Level(), "temp directory", log.Err(err))
 		return nil, err
 	}
 
@@ -49,7 +48,7 @@ func Init(logger *log.Logger) (*ShellOperator, error) {
 	// Debug server.
 	debugServer, err := RunDefaultDebugServer(app.DebugUnixSocket, app.DebugHttpServerAddr, op.logger.Named("debug-server"))
 	if err != nil {
-		logger.Log(context.TODO(), log.LevelFatal.Level(), "start Debug server", slog.String("error", err.Error()))
+		logger.Log(context.TODO(), log.LevelFatal.Level(), "start Debug server", log.Err(err))
 		return nil, err
 	}
 
@@ -59,13 +58,13 @@ func Init(logger *log.Logger) (*ShellOperator, error) {
 		"queue":   "",
 	})
 	if err != nil {
-		logger.Log(context.TODO(), log.LevelFatal.Level(), "essemble common operator", slog.String("error", err.Error()))
+		logger.Log(context.TODO(), log.LevelFatal.Level(), "essemble common operator", log.Err(err))
 		return nil, err
 	}
 
 	err = op.assembleShellOperator(hooksDir, tempDir, debugServer, runtimeConfig)
 	if err != nil {
-		logger.Log(context.TODO(), log.LevelFatal.Level(), "essemble shell operator", slog.String("error", err.Error()))
+		logger.Log(context.TODO(), log.LevelFatal.Level(), "essemble shell operator", log.Err(err))
 		return nil, err
 	}
 

pkg/webhook/admission/handler.go

@@ -3,7 +3,6 @@ package admission
 import (
 	"encoding/json"
 	"fmt"
-	"log/slog"
 	"net/http"
 	"strings"
 
@@ -64,7 +63,7 @@ func (h *WebhookHandler) serveReviewRequest(w http.ResponseWriter, r *http.Reque
 
 	admissionResponse, err := h.handleReviewRequest(r.URL.Path, admissionReview.Request)
 	if err != nil {
-		log.Error("validation failed", "request", admissionReview.Request.UID, slog.String("error", err.Error()))
+		log.Error("validation failed", "request", admissionReview.Request.UID, log.Err(err))
 		admissionReview.Response = errored(err)
 	} else {
 		admissionReview.Response = admissionResponse

pkg/webhook/conversion/handler.go

@@ -4,7 +4,6 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
-	"log/slog"
 	"net/http"
 	"strings"
 
@@ -67,7 +66,7 @@ func (h *WebhookHandler) serveReviewRequest(w http.ResponseWriter, r *http.Reque
 
 	conversionResponse, err := h.handleReviewRequest(crdName, convertReview.Request)
 	if err != nil {
-		log.Error("failed to convert", "request", convertReview.Request.UID, slog.String("error", err.Error()))
+		log.Error("failed to convert", "request", convertReview.Request.UID, log.Err(err))
 		convertReview.Response = errored(err)
 	} else {
 		convertReview.Response = conversionResponse