ansible · arrestle · May 8, 2025 · May 5, 2025 · May 5, 2025 · May 6, 2025
diff --git a/pkg/backends/tcp.go b/pkg/backends/tcp.go
@@ -62,6 +62,10 @@ func (b *TCPDialer) Start(ctx context.Context, wg *sync.WaitGroup) (chan netcept
 				return nil, err
 			}
 
+			if b.logger != nil {
+				b.logger.Debug("TCPDialer connected to TCP %s Address %s\n", b.address, conn.RemoteAddr().String())
+			}
+
 			return newTCPSession(conn, closeChan), nil
 		})
 }

diff --git a/pkg/logger/logger.go b/pkg/logger/logger.go
@@ -88,6 +88,7 @@ type ReceptorLogger struct {
 	log.Logger
 	Prefix string
 	m      sync.Mutex
+	Suffix map[string]string
 }
 
 // NewReceptorLogger to instantiate a new logger object.
@@ -98,6 +99,36 @@ func NewReceptorLogger(prefix string) *ReceptorLogger {
 	}
 }
 
+// NewReceptorLoggerWithSuffix to instantiate a new logger object with a new Suffix.
+func NewReceptorLoggerWithSuffix(prefix string, suffix map[string]string) *ReceptorLogger {
+	logger := NewReceptorLogger(prefix)
+	logger.SetSuffix(suffix)
+
+	return logger
+}
+
+// SetSuffix sets the suffix for the logger, overwriting any existing suffix.
+func (rl *ReceptorLogger) SetSuffix(suffix map[string]string) {
+	rl.m.Lock()
+	defer rl.m.Unlock()
+	rl.Suffix = suffix
+}
+
+// UpdateSuffix allows adding suffix key/value pairs to an existing suffix.
+// If the key already exists, the value will be overwritten.
+// If the suffix is nil, it will be initialized.
+// This is useful for adding additional context to log messages when you want to keep the existing suffix key/value pairs.
+func (rl *ReceptorLogger) UpdateSuffix(suffix map[string]string) {
+	rl.m.Lock()
+	defer rl.m.Unlock()
+	if rl.Suffix == nil {
+		rl.Suffix = make(map[string]string)
+	}
+	for k, v := range suffix {
+		rl.Suffix[k] = v
+	}
+}
+
 // SetOutput sets the output destination for the logger.
 func (rl *ReceptorLogger) SetOutput(w io.Writer) {
 	rl.Logger.SetOutput(w)
@@ -218,10 +249,39 @@ func (rl *ReceptorLogger) Log(level int, format string, v ...interface{}) {
 
 	if logLevel >= level {
 		rl.Logger.SetPrefix(prefix)
+		format, v = rl.appendSuffix(format, v)
 		rl.Logger.Printf(format, v...)
 	}
 }
 
+func (rl *ReceptorLogger) appendSuffix(format string, v []interface{}) (string, []interface{}) {
+	if rl.Suffix != nil {
+		rl.m.Lock()
+		defer rl.m.Unlock()
+
+		var b strings.Builder
+		b.WriteString("{")
+		first := true
+		for k, val := range rl.Suffix {
+			if !first {
+				b.WriteString(",")
+			}
+			first = false
+			b.WriteString(`"`)
+			b.WriteString(k)
+			b.WriteString(`":"`)
+			b.WriteString(val)
+			b.WriteString(`"`)
+		}
+		b.WriteString("}")
+
+		format = strings.ReplaceAll(format, "\n", "") + " %s"
+		v = append(v, b.String())
+	}
+
+	return format, v
+}
+
 // SanitizedLog adds a prefix and prints a given log message.
 func (rl *ReceptorLogger) SanitizedLog(level int, format string, v ...interface{}) {
 	if logger != nil {
@@ -245,7 +305,8 @@ func (rl *ReceptorLogger) SanitizedLog(level int, format string, v ...interface{
 	}
 
 	if logLevel >= level {
-		message := fmt.Sprintf(format, v...)
+		message, v := rl.appendSuffix(format, v)
+		message = fmt.Sprintf(message, v...)
 		sanMessage := strings.ReplaceAll(message, "\n", "")
 		rl.Logger.SetPrefix(prefix)
 		rl.Logger.Print(sanMessage)

diff --git a/pkg/logger/logger_test.go b/pkg/logger/logger_test.go
@@ -3,7 +3,7 @@ package logger_test
 import (
 	"bytes"
 	"fmt"
-	"os"
+	"strings"
 	"testing"
 
 	"github.com/ansible/receptor/pkg/logger"
@@ -72,14 +72,10 @@ func TestLogLevelToNameWithError(t *testing.T) {
 }
 
 func TestDebugPayload(t *testing.T) {
-	logFilePath := "/tmp/test-output"
+	var logBuffer bytes.Buffer
 	logger.SetGlobalLogLevel(4)
 	receptorLogger := logger.NewReceptorLogger("testDebugPayload")
-	logFile, err := os.OpenFile(logFilePath, os.O_WRONLY|os.O_CREATE|os.O_APPEND, 0o600)
-	if err != nil {
-		t.Error("error creating test-output file")
-	}
-
+	receptorLogger.SetOutput(&logBuffer)
 	payload := "Testing debugPayload"
 	workUnitID := "1234"
 	connectionType := "unix socket"
@@ -104,19 +100,67 @@ func TestDebugPayload(t *testing.T) {
 
 	for _, testCase := range debugPayloadTestCases {
 		t.Run(testCase.name, func(t *testing.T) {
-			receptorLogger.SetOutput(logFile)
 			receptorLogger.DebugPayload(testCase.debugPayload, testCase.payload, testCase.workUnitID, testCase.connectionType)
+			testOutput := logBuffer.Bytes()
 
-			testOutput, err := os.ReadFile(logFilePath)
-			if err != nil {
-				t.Error("error reading test-output file")
-			}
-			if !bytes.Contains(testOutput, []byte(testCase.expectedLog)) {
+			if !strings.Contains(string(testOutput), testCase.expectedLog) {
 				t.Errorf("failed to log correctly, expected: %v got %v", testCase.expectedLog, string(testOutput))
 			}
-			if err := os.Truncate(logFilePath, 0); err != nil {
-				t.Errorf("failed to truncate: %v", err)
-			}
+			logBuffer.Reset()
 		})
 	}
 }
+
+func assertSuffixFieldsPresent(t *testing.T, logLine string, expected map[string]string) {
+	t.Helper()
+	for k, v := range expected {
+		needle := `"` + k + `":"` + v + `"`
+		if !strings.Contains(logLine, needle) {
+			t.Errorf("expected key-value pair %s not found in log: %s", needle, logLine)
+		}
+	}
+}
+
+func TestGetLoggerWithSuffix(t *testing.T) {
+	logger.SetGlobalLogLevel(4)
+
+	t.Run("initial suffix", func(t *testing.T) {
+		var logBuffer bytes.Buffer
+		testname := "Initial Suffix Example"
+		suffix := map[string]string{
+			"node_id":   "controller",
+			"remote_id": "hop",
+		}
+		receptorLogger := logger.NewReceptorLoggerWithSuffix("", suffix)
+		receptorLogger.SetOutput(&logBuffer)
+
+		receptorLogger.Error("%s", testname)
+		if !strings.Contains(logBuffer.String(), testname) {
+			t.Errorf("expected log message %s not found in log: %s", testname, logBuffer.String())
+		}
+		assertSuffixFieldsPresent(t, logBuffer.String(), suffix)
+	})
+	t.Run("updated suffix", func(t *testing.T) {
+		var logBuffer bytes.Buffer
+		testname := "Updated Suffix Example"
+		suffix := map[string]string{
+			"node_id":   "controller",
+			"remote_id": "hop",
+		}
+		receptorLogger := logger.NewReceptorLoggerWithSuffix("", suffix)
+		receptorLogger.SetOutput(&logBuffer)
+
+		updated := map[string]string{
+			"cost": "12",
+		}
+		receptorLogger.UpdateSuffix(updated)
+		receptorLogger.SanitizedError("%s", testname)
+
+		if !strings.Contains(logBuffer.String(), testname) {
+			t.Errorf("expected log message %s not found in log: %s", testname, logBuffer.String())
+		}
+
+		assertSuffixFieldsPresent(t, logBuffer.String(), suffix)
+		assertSuffixFieldsPresent(t, logBuffer.String(), updated)
+	})
+}
diff --git a/pkg/netceptor/external_backend.go b/pkg/netceptor/external_backend.go
@@ -204,3 +204,7 @@ func (es *ExternalSession) Close() error {
 
 	return err
 }
+
+func (mc *netMessageConn) RemoteAddr() net.Addr {
+	return mc.conn.RemoteAddr()
+}
diff --git a/pkg/netceptor/netceptor.go b/pkg/netceptor/netceptor.go
@@ -355,6 +355,7 @@ func NewWithConsts(ctx context.Context, nodeID string,
 		MinVersion: tls.VersionTLS12,
 	}
 	s.AddNameHash(nodeID)
+	s.GetLogger().SetSuffix(map[string]string{"node_id": nodeID})
 	s.context, s.cancelFunc = context.WithCancel(ctx)
 	s.unreachableBroker = utils.NewBroker(s.context, reflect.TypeOf(UnreachableNotification{}))
 	s.routingUpdateBroker = utils.NewBroker(s.context, reflect.TypeOf(map[string]string{}))
@@ -1987,6 +1988,8 @@ func (s *Netceptor) runProtocol(ctx context.Context, sess BackendSession, bi *Ba
 					if remoteNodeID == s.nodeID {
 						return s.sendAndLogConnectionRejection(remoteNodeID, ci, "it tried to connect using our own node ID")
 					}
+					suffix := map[string]string{"remote_id": remoteNodeID}
+					s.GetLogger().UpdateSuffix(suffix)
 					remoteNodeAccepted := true
 					if bi.allowedPeers != nil {
 						remoteNodeAccepted = false
@@ -2002,6 +2005,7 @@ func (s *Netceptor) runProtocol(ctx context.Context, sess BackendSession, bi *Ba
 						return s.sendAndLogConnectionRejection(remoteNodeID, ci, "it is not in the allowed peers list")
 					}
 
+					// Check if there is connection cost for this remoteNodeID
 					// Check if there is connection cost for this remoteNodeID
 					remoteNodeCost, ok := bi.nodeCost[remoteNodeID]
 					if ok {
@@ -2010,11 +2014,21 @@ func (s *Netceptor) runProtocol(ctx context.Context, sess BackendSession, bi *Ba
 					}
 					s.connLock.Lock()
 
-					// Check if there connInfo for this remoteNodeID
-					_, ok = s.connections[remoteNodeID]
+					// Check if there is already connInfo for this remoteNodeID
+					existingConn, ok := s.connections[remoteNodeID]
 					if ok {
 						remoteNodeAccepted = false
 					}
+					var connError error
+					connError = nil
+
+					// Verify that the existing connection is valid
+					if ok && existingConn != nil {
+						connError = existingConn.Context.Err()
+					}
+					if ok && connError != nil {
+						s.Logger.Error("Context for existing connection error: %s", connError)
+					}
 
 					if !remoteNodeAccepted {
 						s.connLock.Unlock()