Move logic to logger, add logging for kube

pkg/controlsvc/controlsvc.go

@@ -125,9 +125,15 @@ func (s *SockControl) ReadFromConn(message string, out io.Writer, io Copier) err
 		return err
 	}
 	payloadDebug, _ := strconv.Atoi(os.Getenv("RECEPTOR_PAYLOAD_TRACE_LEVEL"))
-	switch {
-	case payloadDebug > 2:
-		var data string
+
+	if payloadDebug != 0 {
+		var connectionType string
+		var payload string
+		if s.conn.LocalAddr().Network() == "unix" {
+			connectionType = "unix socket"
+		} else {
+			connectionType = "network connection"
+		}
 		reader := bufio.NewReader(s.conn)
 
 		for {
@@ -139,23 +145,14 @@ func (s *SockControl) ReadFromConn(message string, out io.Writer, io Copier) err
 
 				break
 			}
-			data += response
-			MainInstance.nc.GetLogger().Debug("Response reading from conn: %v", response)
-		}
-		if _, err := out.Write([]byte(data)); err != nil {
-			return err
+			payload += response
 		}
 
-		fallthrough
-	case payloadDebug > 0:
-		var connectType string
-		if s.conn.LocalAddr().Network() == "unix" {
-			connectType = "unix socket"
-		} else {
-			connectType = "network connection"
+		MainInstance.nc.GetLogger().DebugPayload(payloadDebug, payload, "", connectionType)
+		if _, err := out.Write([]byte(payload)); err != nil {
+			return err
 		}
-		MainInstance.nc.GetLogger().Debug("Reading from %v", connectType)
-	default:
+	} else {
 		if _, err := io.Copy(out, s.conn); err != nil {
 			return err
 		}

pkg/logger/logger.go

@@ -148,6 +148,31 @@ func (rl *ReceptorLogger) Debug(format string, v ...interface{}) {
 	rl.Log(DebugLevel, format, v...)
 }
 
+// Debug payload data.
+func (rl *ReceptorLogger) DebugPayload(payloadDebug int, payload string, workUnitID string, connectionType string) {
+	switch payloadDebug {
+	case 3:
+		if workUnitID != "" {
+			rl.Debug("Work unit %v stdin: %v", workUnitID, payload)
+		} else {
+			rl.Debug("Response reading from conn: %v", payload)
+		}
+
+		fallthrough
+	case 2:
+		if payloadDebug == 2 && workUnitID != "" {
+			rl.Debug("Work unit %v received command\n", workUnitID)
+		}
+
+		fallthrough
+	case 1:
+		if connectionType != "" {
+			rl.Debug("Reading from %v", connectionType)
+		}
+	default:
+	}
+}
+
 // SanitizedDebug contains extra information helpful to developers.
 func (rl *ReceptorLogger) SanitizedDebug(format string, v ...interface{}) {
 	rl.SanitizedLog(DebugLevel, format, v...)

pkg/workceptor/command.go

@@ -116,16 +116,20 @@ func commandRunner(command string, params string, unitdir string) error {
 		return err
 	}
 	payloadDebug, _ := strconv.Atoi(os.Getenv("RECEPTOR_PAYLOAD_TRACE_LEVEL"))
-	splitUnitDir := strings.Split(unitdir, "/")
-	workUnitID := splitUnitDir[len(splitUnitDir)-1]
-	switch payloadDebug {
-	case 3:
-		var data string
-		reader := bufio.NewReader(stdin)
+
+	if payloadDebug != 0 {
+		splitUnitDir := strings.Split(unitdir, "/")
+		workUnitID := splitUnitDir[len(splitUnitDir)-1]
 		stdinStream, err := cmd.StdinPipe()
 		if err != nil {
 			return err
 		}
+		var payload string
+		reader := bufio.NewReader(stdin)
+		if err != nil {
+			return err
+		}
+
 		for {
 			response, err := reader.ReadString('\n')
 			if err != nil {
@@ -135,14 +139,12 @@ func commandRunner(command string, params string, unitdir string) error {
 
 				break
 			}
-			data += response
-			MainInstance.nc.GetLogger().Debug("Work unit %v stdin: %v", workUnitID, response)
+			payload += response
 		}
-		io.WriteString(stdinStream, data)
-	case 2:
-		MainInstance.nc.GetLogger().Debug("Work unit %v received command\n", workUnitID)
-		cmd.Stdin = stdin
-	default:
+
+		MainInstance.nc.GetLogger().DebugPayload(payloadDebug, payload, workUnitID, "")
+		io.WriteString(stdinStream, payload)
+	} else {
 		cmd.Stdin = stdin
 	}
 	stdout, err := os.OpenFile(path.Join(unitdir, "stdout"), os.O_CREATE+os.O_WRONLY+os.O_SYNC, 0o600)

pkg/workceptor/stdio_utils.go

@@ -8,6 +8,8 @@ import (
 	"io"
 	"os"
 	"path"
+	"strconv"
+	"strings"
 	"sync"
 )
 
@@ -111,6 +113,7 @@ func (sw *STDoutWriter) SetWriter(writer FileWriteCloser) {
 // STDinReader reads from a stdin file and provides a Done function.
 type STDinReader struct {
 	reader   FileReadCloser
+	workUnit string
 	lasterr  error
 	doneChan chan struct{}
 	doneOnce sync.Once
@@ -120,6 +123,8 @@ var errFileSizeZero = errors.New("file is empty")
 
 // NewStdinReader allocates a new stdinReader, which reads from a stdin file and provides a Done function.
 func NewStdinReader(fs FileSystemer, unitdir string) (*STDinReader, error) {
+	splitUnitDir := strings.Split(unitdir, "/")
+	workUnitID := splitUnitDir[len(splitUnitDir)-1]
 	stdinpath := path.Join(unitdir, "stdin")
 	stat, err := fs.Stat(stdinpath)
 	if err != nil {
@@ -135,6 +140,7 @@ func NewStdinReader(fs FileSystemer, unitdir string) (*STDinReader, error) {
 
 	return &STDinReader{
 		reader:   reader,
+		workUnit: workUnitID,
 		lasterr:  nil,
 		doneChan: make(chan struct{}),
 		doneOnce: sync.Once{},
@@ -143,6 +149,23 @@ func NewStdinReader(fs FileSystemer, unitdir string) (*STDinReader, error) {
 
 // Read reads data from the stdout file, implementing io.Reader.
 func (sr *STDinReader) Read(p []byte) (n int, err error) {
+	payloadDebug, _ := strconv.Atoi(os.Getenv("RECEPTOR_PAYLOAD_TRACE_LEVEL"))
+
+	if payloadDebug != 0 {
+		isNotEmpty := func() bool {
+			for _, v := range p {
+				if v != 0 {
+					return true
+				}
+			}
+
+			return false
+		}()
+		if isNotEmpty {
+			payload := string(p)
+			MainInstance.nc.GetLogger().DebugPayload(payloadDebug, payload, sr.workUnit, "")
+		}
+	}
 	n, err = sr.reader.Read(p)
 	if err != nil {
 		sr.lasterr = err