lsm: switch to structured logging

Additionally add a logs_to_trace script that can create a trace JSON
file that can be viewed in chrome://tracing from the logs output to
visualize what is going on.

Author: rockwotj

src/v/lsm/db/gc_actor.cc

@@ -17,8 +17,10 @@
 namespace lsm::db {
 
 ss::future<> gc_actor::process(gc_message msg) {
+    vlog(log.trace, "gc_actor_process_start");
     auto gen = _persistence->list_files();
     chunked_hash_set<internal::file_handle> seen_gc_files;
+    int deleted = 0;
     while (auto file_handle_opt = co_await gen()) {
         if (_as.abort_requested()) {
             co_return;
@@ -59,11 +61,13 @@ ss::future<> gc_actor::process(gc_message msg) {
         co_await _table_cache->evict(file_handle);
         co_await _persistence->remove_file(file_handle);
         _pending_deletes.erase(it);
+        ++deleted;
     }
+    vlog(log.trace, "gc_actor_process_end deleted={}", deleted);
 }
 
 void gc_actor::on_error(std::exception_ptr ex) noexcept {
-    vlog(log.warn, "error in LSM tree GC: {}", ex);
+    vlog(log.warn, "gc_actor_process_end error=\"{}\"", ex);
 }
 
 } // namespace lsm::db

src/v/lsm/db/impl.cc

@@ -25,7 +25,6 @@
 #include "lsm/sst/block_cache.h"
 #include "lsm/sst/builder.h"
 
-#include <seastar/core/coroutine.hh>
 #include <seastar/core/sleep.hh>
 #include <seastar/coroutine/as_future.hh>
 
@@ -54,6 +53,7 @@ impl::impl(ctor, io::persistence p, ss::lw_shared_ptr<internal::options> o)
 
 ss::future<std::unique_ptr<impl>> impl::open(
   ss::lw_shared_ptr<internal::options> opts, io::persistence persistence) {
+    vlog(log.trace, "open_start");
     auto db = std::make_unique<impl>(
       ctor{}, std::move(persistence), std::move(opts));
     co_await db->_gc_actor.start();
@@ -65,21 +65,22 @@ ss::future<std::unique_ptr<impl>> impl::open(
     }
     // If we're readonly, we don't need to start any compaction loop.
     if (db->_opts->readonly) {
+        vlog(log.trace, "open_end readonly=true");
         co_return db;
     }
     db->_background_work = ss::with_scheduling_group(
       db->_opts->compaction_scheduling_group, [db = db.get()] {
           return ss::do_until(
             [db] { return db->_as.abort_requested(); },
             [db] {
-                vlog(log.trace, "waiting for background work");
                 return db->_start_background_work_signal.wait(db->_as)
                   .then([db] {
                       db->_background_work_running = true;
-                      vlog(log.trace, "start background compaction");
+                      vlog(log.trace, "compaction_loop_start");
                       return db->run_background_compaction();
                   })
                   .then_wrapped([db](ss::future<> fut) {
+                      vlog(log.trace, "compaction_loop_end");
                       db->_background_work_running = false;
                       db->maybe_schedule_compaction();
                       db->_background_work_finished_signal.broadcast();
@@ -89,18 +90,19 @@ ss::future<std::unique_ptr<impl>> impl::open(
                       } catch (const abort_requested_exception& ex) {
                           vlog(
                             log.debug,
-                            "LSM background loop got abort request: {}",
+                            "compaction_loop_error abort=true error=\"{}\"",
                             ex.what());
                       } catch (const io_error_exception& ex) {
                           vlog(
                             log.warn,
-                            "LSM background loop hit IO error: {}",
+                            "compaction_loop_error io=true error=\"{}\"",
                             ex.what());
                       } catch (...) {
                           auto ep = std::current_exception();
                           vlog(
                             log.error,
-                            "Unexpected error in LSM background loop: {}",
+                            "compaction_loop_error unexpected=true "
+                            "error=\"{}\"",
                             ep);
                       }
                       // Signal so that we immediately retry
@@ -110,6 +112,7 @@ ss::future<std::unique_ptr<impl>> impl::open(
                   });
             });
       });
+    vlog(log.trace, "open_end readonly=false");
     co_return db;
 }
 
@@ -133,7 +136,7 @@ ss::future<> impl::make_room_for_write() {
           && _versions->current()->num_files(0_level)
                > _opts->level_zero_slowdown_writes_trigger) {
             // We're in throttling mode
-            vlog(log.debug, "throttling writes due to number of L0 files");
+            vlog(log.debug, "throttling_writes reason=l0_file_count");
             try {
                 co_await ss::sleep_abortable(std::chrono::seconds(1), _as);
             } catch (...) {
@@ -151,7 +154,7 @@ ss::future<> impl::make_room_for_write() {
             co_return;
         }
         if (_imm) {
-            vlog(log.warn, "blocking writes as in memory buffers are full");
+            vlog(log.warn, "blocking_writes reason=memtable_full");
             // We are over the write buffer limit and we have a pending
             // memtable flush, wait for it to finish.
             co_await _background_work_finished_signal.wait(_as);
@@ -160,16 +163,13 @@ ss::future<> impl::make_room_for_write() {
         if (
           _versions->current()->num_files(0_level)
           > _opts->level_zero_stop_writes_trigger) {
-            vlog(
-              log.warn,
-              "too many L0 files, writing for compaction to finish before "
-              "allowing more writes");
+            vlog(log.warn, "blocking_writes reason=l0_full");
             // We've hit out L0 file limit, wait for compaction to finish.
             co_await _background_work_finished_signal.wait(_as);
             continue;
         }
         // We're over our limit, let's make a new memtable
-        vlog(log.trace, "scheduling memtable flush");
+        vlog(log.trace, "scheduling_memtable_flush");
         _imm = std::exchange(_mem, ss::make_lw_shared<memtable>());
         maybe_schedule_compaction();
     }
@@ -265,6 +265,7 @@ ss::future<> impl::flush() {
 }
 
 ss::future<> impl::close() {
+    vlog(log.trace, "close_start");
     _as.request_abort_ex(abort_requested_exception("database closing"));
     auto fut = std::exchange(_background_work, std::nullopt);
     if (fut) {
@@ -274,12 +275,14 @@ ss::future<> impl::close() {
     co_await _table_cache->close();
     co_await _persistence.data->close();
     co_await _persistence.metadata->close();
+    vlog(log.trace, "close_end");
     if (fut && fut->failed()) {
         std::rethrow_exception(fut->get_exception());
     }
 }
 
 ss::future<> impl::recover() {
+    vlog(log.trace, "recover_start");
     co_await _versions->recover();
     // If requested, then pre-open all the files we know about.
     if (auto max_fibers = _opts->max_pre_open_fibers) {
@@ -291,12 +294,15 @@ ss::future<> impl::recover() {
                 all_files.push_back(std::move(file));
             }
         }
+        vlog(log.trace, "recover_pre_open_start files={}", all_files.size());
         co_await ss::max_concurrent_for_each(
           all_files, max_fibers, [this](ss::lw_shared_ptr<file_meta_data> f) {
               return _table_cache->create_iterator(f->handle, f->file_size)
                 .discard_result();
           });
+        vlog(log.trace, "recover_pre_open_end files={}", all_files.size());
     }
+    vlog(log.trace, "recover_end");
 }
 
 void impl::maybe_schedule_compaction() {
@@ -364,6 +370,17 @@ ss::future<> impl::run_background_compaction() {
         co_return;
     }
     auto compaction = *std::move(maybe_compaction);
+    auto input_level = compaction.level();
+    auto output_level = input_level + 1_level;
+    auto num_input_files
+      = compaction.num_input_files(compaction::which::input_level)
+        + compaction.num_input_files(compaction::which::output_level);
+    vlog(
+      log.trace,
+      "compaction_start input_level={} output_level={} input_files={}",
+      input_level,
+      output_level,
+      num_input_files);
     if (compaction.is_trivial_move()) {
         auto input_level_files = compaction.num_input_files(
           compaction::which::input_level);
@@ -382,7 +399,16 @@ ss::future<> impl::run_background_compaction() {
           .oldest_seqno = file->oldest_seqno,
           .newest_seqno = file->newest_seqno,
         });
+        vlog(log.trace, "manifest_write_start");
         co_await _versions->log_and_apply(std::move(*compaction.edit()));
+        vlog(log.trace, "manifest_write_end");
+        vlog(
+          log.trace,
+          "compaction_end input_level={} output_level={} output_files=1 "
+          "output_bytes={} trivial_move=true",
+          input_level,
+          output_level,
+          file->file_size);
         co_return;
     }
     compaction_state state{
@@ -395,7 +421,6 @@ ss::future<> impl::run_background_compaction() {
       .smallest_snapshot = _snapshots.oldest_seqno().value_or(
         _versions->last_seqno().value()),
     };
-    auto output_level = compaction.level() + 1_level;
     auto max_file_size = _opts->levels[output_level].max_file_size;
     sst::builder::options sst_options{
       .block_size = _opts->sst_block_size,
@@ -493,14 +518,24 @@ ss::future<> impl::run_background_compaction() {
           .newest_seqno = output.newest,
         });
     }
+    vlog(log.trace, "manifest_write_start");
     co_await _versions->log_and_apply(std::move(*edit));
+    vlog(log.trace, "manifest_write_end");
     // Go and cleanup any obsolete files where both the epoch, and file ID is
     // less than what we have committed, and is not in our working set.
     co_await _gc_actor.tell(
       gc_message{
         .highest_used_file_id = _versions->highest_used_file_id(),
         .live_files = _versions->get_live_files(),
       });
+    vlog(
+      log.trace,
+      "compaction_end input_level={} output_level={} output_files={} "
+      "output_bytes={}",
+      input_level,
+      output_level,
+      state.outputs.size(),
+      state.total_bytes);
 }
 
 ss::future<> impl::flush_memtable() {
@@ -512,6 +547,12 @@ ss::future<> impl::flush_memtable() {
                    ? 0_level
                    : v->pick_level_for_memtable_output(
                        imm->min_key().user_key(), imm->max_key().user_key());
+    auto mem_bytes = imm->approximate_memory_usage();
+    vlog(
+      log.trace,
+      "flush_memtable_start level={} mem_bytes={}",
+      level,
+      mem_bytes);
     sst::builder::options sst_options{
       .block_size = _opts->sst_block_size,
       .filter_period = _opts->sst_filter_period,
@@ -525,6 +566,10 @@ ss::future<> impl::flush_memtable() {
       &_as);
     if (!result) {
         _versions->reuse_file_id(id);
+        vlog(
+          log.trace,
+          "flush_memtable_end level={} file_bytes=0 empty=true",
+          level);
         co_return;
     }
     version_edit edit(*_opts);
@@ -538,7 +583,14 @@ ss::future<> impl::flush_memtable() {
       .oldest_seqno = result->oldest_seqno,
       .newest_seqno = result->newest_seqno,
     });
+    vlog(log.trace, "manifest_write_start");
     co_await _versions->log_and_apply(std::move(edit));
+    vlog(log.trace, "manifest_write_end");
+    vlog(
+      log.trace,
+      "flush_memtable_end level={} file_bytes={}",
+      level,
+      result->file_size);
     // Now that the new version has been applied, it's safe to remove the
     // immutable memtable, as readers will pick up the new file instead.
     //

src/v/lsm/db/table_cache.cc

@@ -134,10 +134,7 @@ class table_cache::impl {
       , _cache(compute_cache_config(max_entries), eviction(this))
       , _block_cache(std::move(block_cache))
       , _cleanup_queue([](const std::exception_ptr& ex) {
-          vlog(
-            log.error,
-            "expected exception on table cache cleanup queue: {}",
-            ex);
+          vlog(log.error, "table_cache_cleanup_error error=\"{}\"", ex);
       }) {}
 
     ss::future<std::unique_ptr<internal::iterator>>

src/v/lsm/db/tests/.gitignore

@@ -0,0 +1 @@
+trace.json

src/v/lsm/db/tests/logs_to_trace.py

@@ -0,0 +1,103 @@
+#!/usr/bin/env python3
+"""
+Convert LSM trace logs to Chrome trace JSON format.
+
+Usage:
+    ./logs_to_trace.py < logfile.log > trace.json
+    ./logs_to_trace.py logfile.log > trace.json
+
+Open the output in chrome://tracing or https://ui.perfetto.dev/
+"""
+
+import json
+import re
+import sys
+from datetime import datetime
+
+
+# Log format: LEVEL TIMESTAMP [shard N:CONTEXT] LOGGER - FILE:LINE - MESSAGE
+LOG_PATTERN = re.compile(
+    r"^(?P<level>\w+)\s+"
+    r"(?P<timestamp>\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2},\d{3})\s+"
+    r"\[shard (?P<shard>\d+):\w+\]\s+"
+    r"lsm\s+-\s+"
+    r"(?P<file>\w+\.cc):(?P<line>\d+)\s+-\s+"
+    r"(?P<message>.+)$"
+)
+
+
+def parse_timestamp(ts_str: str) -> int:
+    """Parse seastar log timestamp to microseconds since epoch."""
+    dt = datetime.strptime(ts_str, "%Y-%m-%d %H:%M:%S,%f")
+    return int(dt.timestamp() * 1_000_000)
+
+
+def parse_attrs(message: str) -> dict[str, str]:
+    """Parse key=value attributes from message, supporting quoted values."""
+    attrs = {}
+    # Match key=value or key="quoted value"
+    for match in re.finditer(r'(\w+)=(?:"([^"]*)"|(\S+))', message):
+        k = match.group(1)
+        v = match.group(2) if match.group(2) is not None else match.group(3)
+        try:
+            attrs[k] = int(v)
+        except ValueError:
+            attrs[k] = v
+    return attrs
+
+
+def main():
+    # Read input
+    if len(sys.argv) > 1:
+        with open(sys.argv[1]) as f:
+            lines = f.readlines()
+    else:
+        lines = sys.stdin.readlines()
+
+    events = []
+
+    for line in lines:
+        match = LOG_PATTERN.match(line.strip())
+        if not match:
+            continue
+
+        msg = match.group("message")
+        parts = msg.split()
+        if not parts:
+            continue
+
+        op = parts[0]
+
+        # Determine phase: B=begin, E=end, i=instant
+        if op.endswith("_start"):
+            name = op[:-6]
+            ph = "B"
+        elif op.endswith("_end"):
+            name = op[:-4]
+            ph = "E"
+        else:
+            # Instant event for point-in-time operations
+            name = op
+            ph = "i"
+
+        event = {
+            "name": name,
+            "cat": "lsm",
+            "ph": ph,
+            "ts": parse_timestamp(match.group("timestamp")),
+            "pid": 1,
+            "tid": int(match.group("shard")),
+            "args": parse_attrs(msg),
+        }
+
+        # Instant events need a scope (g=global, p=process, t=thread)
+        if ph == "i":
+            event["s"] = "t"  # Thread-scoped instant event
+
+        events.append(event)
+
+    print(json.dumps({"traceEvents": events}, indent=2))
+
+
+if __name__ == "__main__":
+    main()