More commit log logs (#664)

Co-authored-by: ines potier <[email protected]>

Author: inespot

src/java/org/apache/cassandra/db/commitlog/CommitLog.java

@@ -21,12 +21,14 @@
 import java.nio.ByteBuffer;
 import java.nio.file.Paths;
 import java.util.*;
+import java.util.stream.Collectors;
 
 import javax.annotation.Nullable;
 
 import com.google.common.annotations.VisibleForTesting;
 import com.google.common.collect.ImmutableMap;
 
+import com.palantir.logsafe.SafeArg;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -163,7 +165,8 @@ public boolean accept(File dir, String name)
         else
         {
             Arrays.sort(files, new CommitLogSegmentFileComparator());
-            logger.info("Replaying {}", StringUtils.join(files, ", "));
+            logger.info("Replaying {}",
+                        SafeArg.of("commitLogFiles", Arrays.stream(files).map(File::getName).collect(Collectors.toSet())));
             replayed = recover(files);
             logger.info("Log replay complete, {} replayed mutations", replayed);
 

src/java/org/apache/cassandra/db/commitlog/CommitLogReplayer.java

@@ -31,6 +31,7 @@
 import java.util.concurrent.atomic.AtomicInteger;
 
 import com.palantir.logsafe.SafeArg;
+import com.palantir.logsafe.UnsafeArg;
 import com.google.common.base.Predicate;
 import com.google.common.collect.HashMultimap;
 import com.google.common.collect.Iterables;
@@ -103,6 +104,7 @@ public static CommitLogReplayer construct(CommitLog commitLog)
     {
         // compute per-CF and global replay positions
         Map<UUID, ReplayPosition.ReplayFilter> cfPersisted = new HashMap<>();
+        Set<UUID> cfWithoutFilter = new HashSet<>();
         ReplayFilter replayFilter = ReplayFilter.create();
         ReplayPosition globalPosition = null;
         for (ColumnFamilyStore cfs : ColumnFamilyStore.all())
@@ -133,13 +135,18 @@ public static CommitLogReplayer construct(CommitLog commitLog)
             if (!filter.isEmpty())
                 cfPersisted.put(cfs.metadata.cfId, filter);
             else
+            {
                 globalPosition = ReplayPosition.NONE; // if we have no ranges for this CF, we must replay everything and filter
+                cfWithoutFilter.add(cfs.metadata.cfId);
+            }
         }
         if (globalPosition == null)
             globalPosition = ReplayPosition.firstNotCovered(cfPersisted.values());
+
         logger.debug("Global replay position {} is from columnfamilies {}",
                      SafeArg.of("globalPosition", globalPosition),
-                     SafeArg.of("columnFamilies", FBUtilities.toString(cfPersisted)));
+                     SafeArg.of("columnFamiliesWithReplayFilters", cfPersisted.keySet()),
+                     SafeArg.of("columnFamiliesWithoutReplayFilters", cfWithoutFilter));
         return new CommitLogReplayer(commitLog, globalPosition, cfPersisted, replayFilter);
     }
 
@@ -464,17 +471,18 @@ public void recover(File file, boolean tolerateTruncation) throws IOException
         finally
         {
             FileUtils.closeQuietly(reader);
-            logger.debug("Finished reading {}", file);
+            logger.debug("Finished reading {}",
+                         SafeArg.of("file", file.getName()));
         }
     }
 
     public boolean logAndCheckIfShouldSkip(File file, CommitLogDescriptor desc)
     {
         logger.debug("Replaying {} (CL version {}, messaging version {}, compression {})",
-                    file.getPath(),
-                    desc.version,
-                    desc.getMessagingVersion(),
-                    desc.compression);
+                     SafeArg.of("file", file.getName()),
+                     SafeArg.of("version", desc.version),
+                     SafeArg.of("messagingVersion", desc.getMessagingVersion()),
+                     UnsafeArg.of("compression", desc.compression));
 
         if (globalPosition.segment > desc.id)
         {

src/java/org/apache/cassandra/db/commitlog/CommitLogSegmentManager.java

@@ -122,7 +122,7 @@ public void runMayThrow() throws Exception
                                 // TODO : some error handling in case we fail to create a new segment
                                 CommitLogSegment newSegment = CommitLogSegment.createSegment(commitLog);
                                 availableSegments.add(newSegment);
-                                logger.info("No segments in reserve; created a fresh one",
+                                logger.debug("No segments in reserve; created a fresh one",
                                             SafeArg.of("segment", newSegment.id));
                                 hasAvailableSegments.signalAll();
                             }
@@ -228,6 +228,9 @@ private void advanceAllocatingFrom(CommitLogSegment old)
                 {
                     allocatingFrom = next;
                     activeSegments.add(next);
+                    logger.debug("Advanced allocating segment from {} to {}",
+                                 SafeArg.of("old", old == null ? "null" : old.id),
+                                 SafeArg.of("new", next.id));
                 }
             }
 
@@ -353,7 +356,8 @@ void recycleSegment(final CommitLogSegment segment)
         }
         else
         {
-            logger.warn("segment {} not found in activeSegments queue", segment);
+            logger.warn("Segment {} not found in activeSegments queue",
+                        SafeArg.of("segment", segment.id));
         }
     }