fix: Tests + logging

keboola · Feb 17, 2025 · 6187b36 · 6187b36
1 parent 1bdd23f
commit 6187b36
Show file tree

Hide file tree

Showing 3 changed files with 8 additions and 6 deletions.
diff --git a/internal/pkg/service/stream/storage/level/local/diskwriter/network/connection/connection.go b/internal/pkg/service/stream/storage/level/local/diskwriter/network/connection/connection.go
@@ -144,7 +144,9 @@ func (m *Manager) ConnectionsCount() int {
 
 func (m *Manager) updateConnections(ctx context.Context) {
 	activeNodes := m.writerNodes()
-	m.logger.Infof(ctx, `the list of volumes has changed, updating connections: %v`, activeNodes)
+	for _, node := range activeNodes {
+		m.logger.Infof(ctx, `the list of volumes has changed, updating connections: %q - %q`, node.ID, node.Address)
+	}
 
 	// Detect new nodes - to open connection
 	var toOpen []*nodeData

diff --git a/...l/pkg/service/stream/storage/level/local/diskwriter/network/connection/connection_test.go b/...l/pkg/service/stream/storage/level/local/diskwriter/network/connection/connection_test.go
@@ -41,7 +41,7 @@ func TestConnectionManager(t *testing.T) {
 	// Start source node
 	connManager, s := startSourceNode(t, ctx, etcdCfg, "s1")
 	sourceLogger := s.DebugLogger()
-	waitForLog(t, sourceLogger, `{"level":"info","message":"the list of volumes has changed, updating connections","component":"storage.router.connections"}`)
+	waitForLog(t, sourceLogger, `{"level":"info","message":"the list of volumes has changed, updating connections: \"%s\" - \"%s\"","component":"storage.router.connections"}`)
 	waitForLog(t, sourceLogger, `{"level":"info","message":"disk writer client connected from \"%s\" to \"w1\" - \"l%s\"","component":"storage.router.connections.client.transport"}`)
 	waitForLog(t, sourceLogger, `{"level":"info","message":"disk writer client connected from \"%s\" to \"w2\" - \"l%s\"","component":"storage.router.connections.client.transport"}`)
 	waitForLog(t, w1.DebugLogger(), `{"level":"info","message":"accepted connection from \"%s\" to \"%s\"","component":"storage.node.writer.rpc.transport"}`)
@@ -58,7 +58,7 @@ func TestConnectionManager(t *testing.T) {
 	// Start another 2 writer nodes
 	w3 := startWriterNode(t, ctx, etcdCfg, "w3")
 	w4 := startWriterNode(t, ctx, etcdCfg, "w4")
-	waitForLog(t, sourceLogger, `{"level":"info","message":"the list of volumes has changed, updating connections","component":"storage.router.connections"}`)
+	waitForLog(t, sourceLogger, `{"level":"info","message":"the list of volumes has changed, updating connections: \"%s\" - \"%s\"","component":"storage.router.connections"}`)
 	waitForLog(t, sourceLogger, `{"level":"info","message":"disk writer client connected from \"%s\" to \"w3\" - \"l%s\"","component":"storage.router.connections.client.transport"}`)
 	waitForLog(t, sourceLogger, `{"level":"info","message":"disk writer client connected from \"%s\" to \"w4\" - \"l%s\"","component":"storage.router.connections.client.transport"}`)
 	waitForLog(t, w3.DebugLogger(), `{"level":"info","message":"accepted connection from \"%s\" to \"%s\"","component":"storage.node.writer.rpc.transport"}`)
@@ -83,7 +83,7 @@ func TestConnectionManager(t *testing.T) {
 	w1.Process().WaitForShutdown()
 	w3.Process().Shutdown(ctx, errors.New("bye bye writer 3"))
 	w3.Process().WaitForShutdown()
-	waitForLog(t, sourceLogger, `{"level":"info","message":"the list of volumes has changed, updating connections","component":"storage.router.connections"}`)
+	waitForLog(t, sourceLogger, `{"level":"info","message":"the list of volumes has changed, updating connections: \"%s\" - \"%s\"","component":"storage.router.connections"}`)
 	waitForLog(t, sourceLogger, `{"level":"info","message":"disk writer client disconnected from \"w1\" - \"localhost:%s\": %s","component":"storage.router.connections.client.transport"}`)
 	waitForLog(t, sourceLogger, `{"level":"info","message":"disk writer client disconnected from \"w3\" - \"localhost:%s\": %s","component":"storage.router.connections.client.transport"}`)
 	sourceLogger.Truncate()

diff --git a/internal/pkg/service/stream/storage/level/local/encoding/pipeline_test.go b/internal/pkg/service/stream/storage/level/local/encoding/pipeline_test.go
@@ -101,7 +101,7 @@ func TestEncodingPipeline_FlushError(t *testing.T) {
 		slice.Encoding,
 		slice.LocalStorage,
 		func(ctx context.Context, cause string) {},
-		nil,
+		newDummyOutput(),
 	)
 	require.NoError(t, err)
 	// Test Close method
@@ -133,7 +133,7 @@ func TestEncodingPipeline_CloseError(t *testing.T) {
 		slice.Encoding,
 		slice.LocalStorage,
 		func(ctx context.Context, cause string) {},
-		nil,
+		newDummyOutput(),
 	)
 	require.NoError(t, err)