diff --git a/bookkeeper-benchmark/src/main/java/org/apache/bookkeeper/benchmark/BenchBookie.java b/bookkeeper-benchmark/src/main/java/org/apache/bookkeeper/benchmark/BenchBookie.java index 58e307824ef..4f2e2824355 100644 --- a/bookkeeper-benchmark/src/main/java/org/apache/bookkeeper/benchmark/BenchBookie.java +++ b/bookkeeper-benchmark/src/main/java/org/apache/bookkeeper/benchmark/BenchBookie.java @@ -29,6 +29,7 @@ import java.io.IOException; import java.util.concurrent.Executors; import java.util.concurrent.ScheduledExecutorService; +import lombok.CustomLog; import org.apache.bookkeeper.client.BKException; import org.apache.bookkeeper.client.BookKeeper; import org.apache.bookkeeper.client.LedgerHandle; @@ -51,14 +52,12 @@ import org.apache.commons.cli.PosixParser; import org.apache.commons.lang3.SystemUtils; import org.apache.zookeeper.KeeperException; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; /** * A utility class used for benchmarking the performance of bookies. */ +@CustomLog public class BenchBookie { - static final Logger LOG = LoggerFactory.getLogger(BenchBookie.class); static class LatencyCallback implements WriteCallback { boolean complete; @@ -66,7 +65,7 @@ static class LatencyCallback implements WriteCallback { public synchronized void writeComplete(int rc, long ledgerId, long entryId, BookieId addr, Object ctx) { if (rc != 0) { - LOG.error("Got error " + rc); + log.error().attr("rc", rc).log("Got error"); } complete = true; notifyAll(); @@ -88,7 +87,7 @@ static class ThroughputCallback implements WriteCallback { public synchronized void writeComplete(int rc, long ledgerId, long entryId, BookieId addr, Object ctx) { if (rc != 0) { - LOG.error("Got error " + rc); + log.error().attr("rc", rc).log("Got error"); } count++; if (count >= waitingCount) { @@ -163,7 +162,9 @@ public static void main(String[] args) try { eventLoop = new EpollEventLoopGroup(); } catch (Throwable t) { - LOG.warn("Could not use Netty Epoll event loop for benchmark {}", t.getMessage()); + log.warn() + .attr("exceptionMessage", t.getMessage()) + .log("Could not use Netty Epoll event loop for benchmark"); eventLoop = new NioEventLoopGroup(); } } else { @@ -196,11 +197,11 @@ public static void main(String[] args) entry, ByteBufList.get(toSend), tc, null, BookieProtocol.FLAG_NONE, false, WriteFlag.NONE); } - LOG.info("Waiting for warmup"); + log.info("Waiting for warmup"); tc.waitFor(warmUpCount); ledger = getValidLedgerId(servers); - LOG.info("Benchmarking latency"); + log.info("Benchmarking latency"); long startTime = System.nanoTime(); for (long entry = 0; entry < latencyCount; entry++) { ByteBuf toSend = Unpooled.buffer(size); @@ -216,10 +217,12 @@ public static void main(String[] args) lc.waitForComplete(); } long endTime = System.nanoTime(); - LOG.info("Latency: " + (((double) (endTime - startTime)) / ((double) latencyCount)) / 1000000.0); + log.info() + .attr("latencyMs", (((double) (endTime - startTime)) / ((double) latencyCount)) / 1000000.0) + .log("Latency"); ledger = getValidLedgerId(servers); - LOG.info("Benchmarking throughput"); + log.info("Benchmarking throughput"); startTime = System.currentTimeMillis(); tc = new ThroughputCallback(); for (long entry = 0; entry < throughputCount; entry++) { @@ -235,7 +238,9 @@ public static void main(String[] args) } tc.waitFor(throughputCount); endTime = System.currentTimeMillis(); - LOG.info("Throughput: " + ((long) throughputCount) * 1000 / (endTime - startTime)); + log.info() + .attr("throughput", ((long) throughputCount) * 1000 / (endTime - startTime)) + .log("Throughput"); bc.close(); scheduler.shutdown(); diff --git a/bookkeeper-benchmark/src/main/java/org/apache/bookkeeper/benchmark/BenchReadThroughputLatency.java b/bookkeeper-benchmark/src/main/java/org/apache/bookkeeper/benchmark/BenchReadThroughputLatency.java index bfff2ed53d1..56bfb8ae9d3 100644 --- a/bookkeeper-benchmark/src/main/java/org/apache/bookkeeper/benchmark/BenchReadThroughputLatency.java +++ b/bookkeeper-benchmark/src/main/java/org/apache/bookkeeper/benchmark/BenchReadThroughputLatency.java @@ -32,6 +32,7 @@ import java.util.concurrent.atomic.AtomicInteger; import java.util.regex.Matcher; import java.util.regex.Pattern; +import lombok.CustomLog; import org.apache.bookkeeper.client.BookKeeper; import org.apache.bookkeeper.client.LedgerEntry; import org.apache.bookkeeper.client.LedgerHandle; @@ -45,14 +46,12 @@ import org.apache.commons.cli.PosixParser; import org.apache.zookeeper.WatchedEvent; import org.apache.zookeeper.Watcher; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; /** * A benchmark that benchmarks the read throughput and latency. */ +@CustomLog public class BenchReadThroughputLatency { - static final Logger LOG = LoggerFactory.getLogger(BenchReadThroughputLatency.class); private static final Pattern LEDGER_PATTERN = Pattern.compile("L([0-9]+)$"); @@ -75,7 +74,7 @@ public int compare(String o1, String o2) { }; private static void readLedger(ClientConfiguration conf, long ledgerId, byte[] passwd, int batchEntries) { - LOG.info("Reading ledger {}", ledgerId); + log.info().attr("ledgerId", ledgerId).log("Reading ledger"); BookKeeper bk = null; long time = 0; long entriesRead = 0; @@ -117,7 +116,10 @@ private static void readLedger(ClientConfiguration conf, long ledgerId, byte[] p entriesRead++; lastRead = e.getEntryId(); if ((entriesRead % 10000) == 0) { - LOG.info("{} entries read from ledger {}", entriesRead, ledgerId); + log.info() + .attr("entriesRead", entriesRead) + .attr("ledgerId", ledgerId) + .log("entries read from ledger"); } e.getEntryBuffer().release(); } @@ -132,9 +134,12 @@ private static void readLedger(ClientConfiguration conf, long ledgerId, byte[] p } catch (InterruptedException ie) { Thread.currentThread().interrupt(); } catch (Exception e) { - LOG.error("Exception in reader", e); + log.error().exception(e).log("Exception in reader"); } finally { - LOG.info("Read {} in {}ms", entriesRead, time / 1000 / 1000); + log.info() + .attr("entriesRead", entriesRead) + .attr("timeMs", time / 1000 / 1000) + .log("Read complete"); try { if (lh != null) { @@ -144,7 +149,7 @@ private static void readLedger(ClientConfiguration conf, long ledgerId, byte[] p bk.close(); } } catch (Exception e) { - LOG.error("Exception closing stuff", e); + log.error().exception(e).log("Exception closing stuff"); } } } @@ -185,7 +190,7 @@ public static void main(String[] args) throws Exception { final int sockTimeout = Integer.parseInt(cmd.getOptionValue("sockettimeout", "5")); final int batchentries = Integer.parseInt(cmd.getOptionValue("batchentries", "1000")); if (cmd.hasOption("ledger") && cmd.hasOption("listen")) { - LOG.error("Cannot used -ledger and -listen together"); + log.error("Cannot used -ledger and -listen together"); usage(options); System.exit(-1); } @@ -197,7 +202,7 @@ public static void main(String[] args) throws Exception { } else if (cmd.hasOption("listen")) { numLedgers.set(Integer.parseInt(cmd.getOptionValue("listen"))); } else { - LOG.error("You must use -ledger or -listen"); + log.error("You must use -ledger or -listen"); usage(options); System.exit(-1); } @@ -215,7 +220,9 @@ public static void main(String[] args) throws Exception { } else if ("longHierarchical".equals(ledgerManagerType)) { nodepath = String.format("/ledgers%s", StringUtils.getLongHierarchicalLedgerPath(ledger.get())); } else { - LOG.warn("Unknown ledger manager type: {}, use flat as the value", ledgerManagerType); + log.warn() + .attr("ledgerManagerType", ledgerManagerType) + .log("Unknown ledger manager type, use flat as the value"); nodepath = String.format("/ledgers/L%010d", ledger.get()); } @@ -271,15 +278,17 @@ public void run() { shutdownLatch.countDown(); } } else { - LOG.error("Cant file ledger id in {}", ledger); + log.error() + .attr("ledger", ledger) + .log("Cant file ledger id"); } } } } else { - LOG.warn("Unknown event {}", event); + log.warn().attr("event", event).log("Unknown event"); } } catch (Exception e) { - LOG.error("Exception in watcher", e); + log.error().exception(e).log("Exception in watcher"); } } }); @@ -289,13 +298,13 @@ public void run() { readLedger(conf, ledger.get(), passwd, batchentries); shutdownLatch.countDown(); } else { - LOG.info("Watching for creation of" + nodepath); + log.info().attr("nodepath", nodepath).log("Watching for creation of znode"); } } else { zk.getChildren("/ledgers", true); } shutdownLatch.await(); - LOG.info("Shutting down"); + log.info("Shutting down"); } } } diff --git a/bookkeeper-benchmark/src/main/java/org/apache/bookkeeper/benchmark/BenchThroughputLatency.java b/bookkeeper-benchmark/src/main/java/org/apache/bookkeeper/benchmark/BenchThroughputLatency.java index 9eeae28f1a3..42aed3ce172 100644 --- a/bookkeeper-benchmark/src/main/java/org/apache/bookkeeper/benchmark/BenchThroughputLatency.java +++ b/bookkeeper-benchmark/src/main/java/org/apache/bookkeeper/benchmark/BenchThroughputLatency.java @@ -34,6 +34,7 @@ import java.util.concurrent.Semaphore; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicLong; +import lombok.CustomLog; import org.apache.bookkeeper.client.AsyncCallback.AddCallback; import org.apache.bookkeeper.client.BKException; import org.apache.bookkeeper.client.BookKeeper; @@ -54,14 +55,12 @@ import org.apache.zookeeper.Watcher.Event.KeeperState; import org.apache.zookeeper.ZooDefs; import org.apache.zookeeper.ZooKeeper; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; /** * A benchmark that benchmarks write throughput and latency. */ +@CustomLog public class BenchThroughputLatency implements AddCallback, Runnable { - static final Logger LOG = LoggerFactory.getLogger(BenchThroughputLatency.class); BookKeeper bk; LedgerHandle[] lh; @@ -99,9 +98,8 @@ public BenchThroughputLatency(int ensemble, int writeQuorumSize, int ackQuorumSi ackQuorumSize, BookKeeper.DigestType.CRC32, passwd); - if (LOG.isDebugEnabled()) { - LOG.debug("Ledger Handle: " + lh[i].getId()); - } + final int idx = i; + log.debug(e -> e.attr("ledgerId", lh[idx].getId()).log("Ledger Handle")); } } catch (BKException e) { e.printStackTrace(); @@ -140,7 +138,7 @@ public synchronized long getDuration() { @Override public void run() { - LOG.info("Running..."); + log.info("Running..."); long start = previous = System.currentTimeMillis(); int sent = 0; @@ -151,10 +149,13 @@ public void run() { try { while (true) { Thread.sleep(1000); - LOG.info("ms: {} req: {}", System.currentTimeMillis(), completedRequests.getAndSet(0)); + log.info() + .attr("timeMs", System.currentTimeMillis()) + .attr("requests", completedRequests.getAndSet(0)) + .log("progress"); } } catch (InterruptedException ie) { - LOG.info("Caught interrupted exception, going away"); + log.info("Caught interrupted exception, going away"); Thread.currentThread().interrupt(); } } @@ -168,7 +169,10 @@ public void run() { if (sent == 10000) { long afterSend = System.nanoTime(); long time = afterSend - beforeSend; - LOG.info("Time to send first batch: {}s {}ns ", time / 1000 / 1000 / 1000, time); + log.info() + .attr("timeSec", time / 1000 / 1000 / 1000) + .attr("timeNs", time) + .log("Time to send first batch"); } } catch (InterruptedException e) { Thread.currentThread().interrupt(); @@ -178,7 +182,7 @@ public void run() { final int index = getRandomLedger(); LedgerHandle h = lh[index]; if (h == null) { - LOG.error("Handle " + index + " is null!"); + log.error().attr("index", index).log("Handle is null!"); } else { long nanoTime = System.nanoTime(); lh[index].asyncAddEntry(bytes, this, new Context(sent, nanoTime)); @@ -186,7 +190,7 @@ public void run() { } sent++; } - LOG.info("Sent: " + sent); + log.info().attr("sent", sent).log("Sent"); try { int i = 0; while (this.counter.get() > 0) { @@ -197,7 +201,7 @@ public void run() { } } } catch (InterruptedException e) { - LOG.error("Interrupted while waiting", e); + log.error().exception(e).log("Interrupted while waiting"); Thread.currentThread().interrupt(); } synchronized (this) { @@ -211,7 +215,10 @@ public void run() { } catch (InterruptedException ie) { Thread.currentThread().interrupt(); } - LOG.info("Finished processing in ms: " + getDuration() + " tp = " + throughput); + log.info() + .attr("durationMs", getDuration()) + .attr("throughput", throughput) + .log("Finished processing"); } long throughput = -1; @@ -307,13 +314,16 @@ public void run() { }, timeout); } - LOG.warn("(Parameters received) running time: " + runningTime - + ", entry size: " + entrysize + ", ensemble size: " + ensemble - + ", quorum size: " + quorum - + ", throttle: " + throttle - + ", number of ledgers: " + ledgers - + ", zk servers: " + servers - + ", latency file: " + latencyFile); + log.warn() + .attr("runningTime", runningTime) + .attr("entrySize", entrysize) + .attr("ensembleSize", ensemble) + .attr("quorumSize", quorum) + .attr("throttle", throttle) + .attr("numLedgers", ledgers) + .attr("zkServers", servers) + .attr("latencyFile", latencyFile) + .log("Parameters received"); long totalTime = runningTime * 1000; @@ -332,11 +342,11 @@ public void run() { if (!cmd.hasOption("skipwarmup")) { long throughput; - LOG.info("Starting warmup"); + log.info("Starting warmup"); throughput = warmUp(data, ledgers, ensemble, quorum, passwd, warmupMessages, conf); - LOG.info("Warmup tp: " + throughput); - LOG.info("Warmup phase finished"); + log.info().attr("throughput", throughput).log("Warmup tp"); + log.info("Warmup phase finished"); } @@ -357,13 +367,13 @@ public void process(WatchedEvent event) { } }}); if (!connectLatch.await(10, TimeUnit.SECONDS)) { - LOG.error("Couldn't connect to zookeeper at " + servers); + log.error().attr("servers", servers).log("Couldn't connect to zookeeper"); zk.close(); System.exit(-1); } final CountDownLatch latch = new CountDownLatch(1); - LOG.info("Waiting for " + coordinationZnode); + log.info().attr("coordinationZnode", coordinationZnode).log("Waiting for znode"); if (zk.exists(coordinationZnode, new Watcher() { @Override public void process(WatchedEvent event) { @@ -374,14 +384,14 @@ public void process(WatchedEvent event) { latch.countDown(); } latch.await(); - LOG.info("Coordination znode created"); + log.info("Coordination znode created"); } thread.start(); Thread.sleep(totalTime); thread.interrupt(); thread.join(); - LOG.info("Calculating percentiles"); + log.info("Calculating percentiles"); int numlat = 0; for (int i = 0; i < bench.latencies.length; i++) { @@ -402,7 +412,11 @@ public void process(WatchedEvent event) { long tp = (long) ((double) (numcompletions * 1000.0) / (double) bench.getDuration()); - LOG.info(numcompletions + " completions in " + bench.getDuration() + " milliseconds: " + tp + " ops/sec"); + log.info() + .attr("completions", numcompletions) + .attr("durationMs", bench.getDuration()) + .attr("opsPerSec", tp) + .log("Benchmark complete"); if (zk != null) { zk.create(coordinationZnode + "/worker-", @@ -421,8 +435,8 @@ public void process(WatchedEvent event) { fos.close(); // now get the latencies - LOG.info("99th percentile latency: {}", percentile(latency, 99)); - LOG.info("95th percentile latency: {}", percentile(latency, 95)); + log.info().attr("latency", percentile(latency, 99)).log("99th percentile latency"); + log.info().attr("latency", percentile(latency, 95)).log("95th percentile latency"); bench.close(); timeouter.cancel(); @@ -458,7 +472,7 @@ public void process(WatchedEvent event) { } }}); if (!connectLatch.await(10, TimeUnit.SECONDS)) { - LOG.error("Couldn't connect to zookeeper at " + servers); + log.error().attr("servers", servers).log("Couldn't connect to zookeeper"); throw new IOException("Couldn't connect to zookeeper " + servers); } bookies = zk.getChildren(bookieRegistrationPath, false).size() - 1; diff --git a/bookkeeper-benchmark/src/main/java/org/apache/bookkeeper/benchmark/TestClient.java b/bookkeeper-benchmark/src/main/java/org/apache/bookkeeper/benchmark/TestClient.java index 9f1b225fa20..ca8745b6fab 100644 --- a/bookkeeper-benchmark/src/main/java/org/apache/bookkeeper/benchmark/TestClient.java +++ b/bookkeeper-benchmark/src/main/java/org/apache/bookkeeper/benchmark/TestClient.java @@ -36,6 +36,7 @@ import java.util.concurrent.Future; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicLong; +import lombok.CustomLog; import org.apache.bookkeeper.client.AsyncCallback.AddCallback; import org.apache.bookkeeper.client.BKException; import org.apache.bookkeeper.client.BookKeeper; @@ -48,15 +49,13 @@ import org.apache.commons.cli.Options; import org.apache.commons.cli.ParseException; import org.apache.commons.cli.PosixParser; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; /** * This is a simple test program to compare the performance of writing to * BookKeeper and to the local file system. */ +@CustomLog public class TestClient { - private static final Logger LOG = LoggerFactory.getLogger(TestClient.class); /** * First says if entries should be written to BookKeeper (0) or to the local @@ -148,7 +147,7 @@ public void run() { clients.add(new FileClient(streams, data, runfor)); } } else { - LOG.error("Unknown option: " + target); + log.error().attr("target", target).log("Unknown option"); throw new IllegalArgumentException("Unknown target " + target); } @@ -161,35 +160,38 @@ public void run() { long count = 0; for (Future r : results) { if (!r.isDone()) { - LOG.warn("Job didn't complete"); + log.warn("Job didn't complete"); System.exit(2); } long c = r.get(); if (c == 0) { - LOG.warn("Task didn't complete"); + log.warn("Task didn't complete"); } count += c; } long time = end - start; - LOG.info("Finished processing writes (ms): {} TPT: {} op/s", time, count / ((double) time / 1000)); + log.info() + .attr("timeMs", time) + .attr("throughput", count / ((double) time / 1000)) + .log("Finished processing writes"); executor.shutdown(); } catch (ExecutionException ee) { - LOG.error("Exception in worker", ee); + log.error().exception(ee).log("Exception in worker"); } catch (BKException e) { - LOG.error("Error accessing bookkeeper", e); + log.error().exception(e).log("Error accessing bookkeeper"); } catch (IOException ioe) { - LOG.error("I/O exception during benchmark", ioe); + log.error().exception(ioe).log("I/O exception during benchmark"); } catch (InterruptedException ie) { - LOG.error("Benchmark interrupted", ie); + log.error().exception(ie).log("Benchmark interrupted"); Thread.currentThread().interrupt(); } finally { if (bkc != null) { try { bkc.close(); } catch (BKException bke) { - LOG.error("Error closing bookkeeper client", bke); + log.error().exception(bke).log("Error closing bookkeeper client"); } catch (InterruptedException ie) { - LOG.warn("Interrupted closing bookkeeper client", ie); + log.warn().exception(ie).log("Interrupted closing bookkeeper client"); Thread.currentThread().interrupt(); } } @@ -228,11 +230,13 @@ public Long call() { } long time = (System.currentTimeMillis() - start); - LOG.info("Worker finished processing writes (ms): {} TPT: {} op/s", time, - count / ((double) time / 1000)); + log.info() + .attr("timeMs", time) + .attr("throughput", count / ((double) time / 1000)) + .log("Worker finished processing writes"); return count; } catch (IOException ioe) { - LOG.error("Exception in worker thread", ioe); + log.error().exception(ioe).log("Exception in worker thread"); return 0L; } } @@ -278,15 +282,17 @@ public Long call() { } long time = (System.currentTimeMillis() - start); - LOG.info("Worker finished processing writes (ms): {} TPT: {} op/s", time, - success.get() / ((double) time / 1000)); + log.info() + .attr("timeMs", time) + .attr("throughput", success.get() / ((double) time / 1000)) + .log("Worker finished processing writes"); return success.get(); } catch (BKException e) { - LOG.error("Exception in worker thread", e); + log.error().exception(e).log("Exception in worker thread"); return 0L; } catch (InterruptedException ie) { Thread.currentThread().interrupt(); - LOG.error("Exception in worker thread", ie); + log.error().exception(ie).log("Exception in worker thread"); return 0L; } } diff --git a/bookkeeper-dist/src/main/resources/LICENSE-all.bin.txt b/bookkeeper-dist/src/main/resources/LICENSE-all.bin.txt index e23b4fe4b09..f66fb0c91a7 100644 --- a/bookkeeper-dist/src/main/resources/LICENSE-all.bin.txt +++ b/bookkeeper-dist/src/main/resources/LICENSE-all.bin.txt @@ -216,6 +216,7 @@ Apache Software License, Version 2. - lib/commons-codec-commons-codec-1.18.0.jar [6] - lib/commons-io-commons-io-2.19.0.jar [8] - lib/commons-logging-commons-logging-1.3.5.jar [10] +- lib/io.github.merlimat.slog-slog-0.9.8.jar [64] - lib/io.netty-netty-buffer-4.2.12.Final.jar [11] - lib/io.netty-netty-codec-base-4.2.12.Final.jar [11] - lib/io.netty-netty-codec-compression-4.2.12.Final.jar [11] @@ -422,6 +423,7 @@ Apache Software License, Version 2. [61] Source available at https://github.com/apache/commons-text/tree/rel/commons-text-1.13.1 [62] Source available at https://github.com/apache/commons-beanutils/tree/rel/commons-beanutils-1.11.0 [63] Source available at https://github.com/googleapis/sdk-platform-java/tree/v2.53.0/api-common-java +[64] Source available at https://github.com/merlimat/slog/tree/v0.9.8 ------------------------------------------------------------------------------------ lib/io.netty-netty-codec-base-4.2.12.Final.jar bundles some 3rd party dependencies diff --git a/bookkeeper-dist/src/main/resources/LICENSE-bkctl.bin.txt b/bookkeeper-dist/src/main/resources/LICENSE-bkctl.bin.txt index 6982f75955d..8b39fb9a5dd 100644 --- a/bookkeeper-dist/src/main/resources/LICENSE-bkctl.bin.txt +++ b/bookkeeper-dist/src/main/resources/LICENSE-bkctl.bin.txt @@ -216,6 +216,7 @@ Apache Software License, Version 2. - lib/commons-codec-commons-codec-1.18.0.jar [6] - lib/commons-io-commons-io-2.19.0.jar [8] - lib/commons-logging-commons-logging-1.3.5.jar [10] +- lib/io.github.merlimat.slog-slog-0.9.8.jar [59] - lib/io.netty-netty-buffer-4.2.12.Final.jar [11] - lib/io.netty-netty-codec-base-4.2.12.Final.jar [11] - lib/io.netty-netty-common-4.2.12.Final.jar [11] @@ -355,6 +356,7 @@ Apache Software License, Version 2. [56] Source available at https://github.com/apache/commons-text/tree/rel/commons-text-1.13.1 [57] Source available at https://github.com/apache/commons-beanutils/tree/rel/commons-beanutils-1.11.0 [58] Source available at https://github.com/googleapis/sdk-platform-java/tree/v2.53.0/api-common-java +[59] Source available at https://github.com/merlimat/slog/tree/v0.9.8 ------------------------------------------------------------------------------------ lib/io.netty-netty-codec-base-4.2.12.Final.jar bundles some 3rd party dependencies diff --git a/bookkeeper-dist/src/main/resources/LICENSE-server.bin.txt b/bookkeeper-dist/src/main/resources/LICENSE-server.bin.txt index b4b7a8f44e0..362493a0515 100644 --- a/bookkeeper-dist/src/main/resources/LICENSE-server.bin.txt +++ b/bookkeeper-dist/src/main/resources/LICENSE-server.bin.txt @@ -216,6 +216,7 @@ Apache Software License, Version 2. - lib/commons-codec-commons-codec-1.18.0.jar [6] - lib/commons-io-commons-io-2.19.0.jar [8] - lib/commons-logging-commons-logging-1.3.5.jar [10] +- lib/io.github.merlimat.slog-slog-0.9.8.jar [63] - lib/io.netty-netty-buffer-4.2.12.Final.jar [11] - lib/io.netty-netty-codec-base-4.2.12.Final.jar [11] - lib/io.netty-netty-codec-compression-4.2.12.Final.jar [11] @@ -417,6 +418,7 @@ Apache Software License, Version 2. [60] Source available at https://github.com/apache/commons-text/tree/rel/commons-text-1.13.1 [61] Source available at https://github.com/apache/commons-beanutils/tree/rel/commons-beanutils-1.11.0 [62] Source available at https://github.com/googleapis/sdk-platform-java/tree/v2.53.0/api-common-java +[63] Source available at https://github.com/merlimat/slog/tree/v0.9.8 ------------------------------------------------------------------------------------ lib/io.netty-netty-codec-base-4.2.12.Final.jar bundles some 3rd party dependencies diff --git a/bookkeeper-http/http-server/src/main/java/org/apache/bookkeeper/http/HttpServerLoader.java b/bookkeeper-http/http-server/src/main/java/org/apache/bookkeeper/http/HttpServerLoader.java index 5dceb8c5c45..a7278857b52 100644 --- a/bookkeeper-http/http-server/src/main/java/org/apache/bookkeeper/http/HttpServerLoader.java +++ b/bookkeeper-http/http-server/src/main/java/org/apache/bookkeeper/http/HttpServerLoader.java @@ -22,17 +22,15 @@ import java.lang.reflect.Constructor; import java.lang.reflect.InvocationTargetException; +import lombok.CustomLog; import org.apache.commons.configuration2.Configuration; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; /** * Class to load and instantiate http server from config. */ +@CustomLog public class HttpServerLoader { - static final Logger LOG = LoggerFactory.getLogger(HttpServerLoader.class); - public static final String HTTP_SERVER_CLASS = "httpServerClass"; static HttpServer server = null; @@ -46,16 +44,17 @@ public static void loadHttpServer(Configuration conf) { (Constructor) cls.getDeclaredConstructor(); server = cons.newInstance(); } catch (ClassNotFoundException cnfe) { - LOG.error("Couldn't find configured class(" + className + ")", cnfe); + log.error().exception(cnfe).attr("className", className).log("Couldn't find configured class"); } catch (NoSuchMethodException nsme) { - LOG.error("Couldn't find default constructor for class (" + className + ")", nsme); + log.error().exception(nsme).attr("className", className) + .log("Couldn't find default constructor for class"); } catch (InstantiationException ie) { - LOG.error("Couldn't construct class (" + className + ")", ie); + log.error().exception(ie).attr("className", className).log("Couldn't construct class"); } catch (IllegalAccessException iae) { - LOG.error("Couldn't construct class (" + className + ")," - + " Is the constructor private?", iae); + log.error().exception(iae).attr("className", className) + .log("Couldn't construct class. Is the constructor private?"); } catch (InvocationTargetException ite) { - LOG.error("Constructor threw an exception. It should not have.", ite); + log.error().exception(ite).log("Constructor threw an exception. It should not have."); } } } diff --git a/bookkeeper-http/servlet-http-server/src/main/java/org/apache/bookkeeper/http/servlet/BookieHttpServiceServlet.java b/bookkeeper-http/servlet-http-server/src/main/java/org/apache/bookkeeper/http/servlet/BookieHttpServiceServlet.java index 30fe063786a..18bab1d5ed5 100644 --- a/bookkeeper-http/servlet-http-server/src/main/java/org/apache/bookkeeper/http/servlet/BookieHttpServiceServlet.java +++ b/bookkeeper-http/servlet-http-server/src/main/java/org/apache/bookkeeper/http/servlet/BookieHttpServiceServlet.java @@ -30,6 +30,7 @@ import javax.servlet.http.HttpServlet; import javax.servlet.http.HttpServletRequest; import javax.servlet.http.HttpServletResponse; +import lombok.CustomLog; import org.apache.bookkeeper.http.AbstractHttpHandlerFactory; import org.apache.bookkeeper.http.HttpRouter; import org.apache.bookkeeper.http.HttpServer; @@ -39,15 +40,13 @@ import org.apache.bookkeeper.http.service.HttpServiceRequest; import org.apache.bookkeeper.http.service.HttpServiceResponse; import org.apache.commons.io.IOUtils; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; /** * Bookie http service servlet. * **/ +@CustomLog public class BookieHttpServiceServlet extends HttpServlet { - static final Logger LOG = LoggerFactory.getLogger(BookieHttpServiceServlet.class); // url to api private final Map mappings = new ConcurrentHashMap<>(); @@ -95,7 +94,7 @@ protected void service(HttpServletRequest httpRequest, HttpServletResponse httpR out.write(response.getBody()); } } catch (Throwable e) { - LOG.error("Error while service Bookie API request {}", uri, e); + log.error().exception(e).attr("uri", uri).log("Error while service Bookie API request"); httpResponse.sendError(HttpServletResponse.SC_INTERNAL_SERVER_ERROR); } } diff --git a/bookkeeper-http/servlet-http-server/src/main/java/org/apache/bookkeeper/http/servlet/BookieServletHttpServer.java b/bookkeeper-http/servlet-http-server/src/main/java/org/apache/bookkeeper/http/servlet/BookieServletHttpServer.java index 5af0d83fe2d..d44b66818d4 100644 --- a/bookkeeper-http/servlet-http-server/src/main/java/org/apache/bookkeeper/http/servlet/BookieServletHttpServer.java +++ b/bookkeeper-http/servlet-http-server/src/main/java/org/apache/bookkeeper/http/servlet/BookieServletHttpServer.java @@ -20,17 +20,16 @@ */ package org.apache.bookkeeper.http.servlet; +import lombok.CustomLog; import org.apache.bookkeeper.http.HttpServer; import org.apache.bookkeeper.http.HttpServerConfiguration; import org.apache.bookkeeper.http.HttpServiceProvider; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; /** * Only use for hold Http service provider,not a fully implement bookie http service server. **/ +@CustomLog public class BookieServletHttpServer implements HttpServer { - static final Logger LOG = LoggerFactory.getLogger(BookieServletHttpServer.class); private static HttpServiceProvider bookieHttpServiceProvider; private static int listenPort = -1; private static String listenHost = "0.0.0.0"; @@ -53,7 +52,7 @@ public static String getListenHost() { @Override public void initialize(HttpServiceProvider httpServiceProvider) { setHttpServiceProvider(httpServiceProvider); - LOG.info("Bookie HTTP Server initialized: {}", httpServiceProvider); + log.info().attr("httpServiceProvider", httpServiceProvider).log("Bookie HTTP Server initialized"); } public static synchronized void setHttpServiceProvider(HttpServiceProvider httpServiceProvider){ diff --git a/bookkeeper-http/vertx-http-server/src/main/java/org/apache/bookkeeper/http/vertx/VertxAbstractHandler.java b/bookkeeper-http/vertx-http-server/src/main/java/org/apache/bookkeeper/http/vertx/VertxAbstractHandler.java index 5dc88948cfb..572b6c525f0 100644 --- a/bookkeeper-http/vertx-http-server/src/main/java/org/apache/bookkeeper/http/vertx/VertxAbstractHandler.java +++ b/bookkeeper-http/vertx-http-server/src/main/java/org/apache/bookkeeper/http/vertx/VertxAbstractHandler.java @@ -30,20 +30,18 @@ import java.util.HashMap; import java.util.Iterator; import java.util.Map; +import lombok.CustomLog; import org.apache.bookkeeper.http.HttpServer; import org.apache.bookkeeper.http.service.HttpEndpointService; import org.apache.bookkeeper.http.service.HttpServiceRequest; import org.apache.bookkeeper.http.service.HttpServiceResponse; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; /** * Http Handler for Vertx based Http Server. */ +@CustomLog public abstract class VertxAbstractHandler implements Handler { - private static final Logger LOG = LoggerFactory.getLogger(VertxAbstractHandler.class); - /** * Process the request using the given httpEndpointService. */ @@ -59,10 +57,11 @@ void processRequest(HttpEndpointService httpEndpointService, RoutingContext cont try { response = httpEndpointService.handle(request); } catch (NullPointerException | IllegalArgumentException e) { - LOG.error("Vertx handler failed to process request {}, cause {}", request.toString(), e); + log.error().exception(e).attr("request", request.toString()) + .log("Vertx handler failed to process request"); throw new RuntimeException(e); } catch (Exception e) { - LOG.error("Exception in vertx handler", e); + log.error().exception(e).log("Exception in vertx handler"); throw new RuntimeException(e); } httpResponse.setStatusCode(response.getStatusCode()); diff --git a/bookkeeper-http/vertx-http-server/src/main/java/org/apache/bookkeeper/http/vertx/VertxHttpServer.java b/bookkeeper-http/vertx-http-server/src/main/java/org/apache/bookkeeper/http/vertx/VertxHttpServer.java index c7c6c20113e..ecc67debf59 100644 --- a/bookkeeper-http/vertx-http-server/src/main/java/org/apache/bookkeeper/http/vertx/VertxHttpServer.java +++ b/bookkeeper-http/vertx-http-server/src/main/java/org/apache/bookkeeper/http/vertx/VertxHttpServer.java @@ -32,20 +32,18 @@ import java.util.concurrent.CompletableFuture; import java.util.concurrent.CountDownLatch; import java.util.concurrent.ExecutionException; +import lombok.CustomLog; import org.apache.bookkeeper.http.HttpRouter; import org.apache.bookkeeper.http.HttpServer; import org.apache.bookkeeper.http.HttpServerConfiguration; import org.apache.bookkeeper.http.HttpServiceProvider; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; /** * Vertx.io implementation of Http Server. */ +@CustomLog public class VertxHttpServer implements HttpServer { - static final Logger LOG = LoggerFactory.getLogger(VertxHttpServer.class); - private final Vertx vertx; private boolean isRunning; private HttpServiceProvider httpServiceProvider; @@ -106,25 +104,28 @@ public void start() throws Exception { trustStoreOptions.setPassword(httpServerConfiguration.getTrustStorePassword()); httpServerOptions.setTrustOptions(trustStoreOptions); } - LOG.info("Starting Vertx HTTP server on port {}", port); + log.info().attr("port", port).log("Starting Vertx HTTP server"); vertx.createHttpServer(httpServerOptions).requestHandler(router).listen(port, host, future::complete); } }); try { AsyncResult asyncResult = future.get(); if (asyncResult.succeeded()) { - LOG.info("Vertx Http server started successfully"); + log.info("Vertx Http server started successfully"); listeningPort = asyncResult.result().actualPort(); isRunning = true; return true; } else { - LOG.error("Failed to start org.apache.bookkeeper.http server on port {}", port, asyncResult.cause()); + log.error().exception(asyncResult.cause()).attr("port", port) + .log("Failed to start org.apache.bookkeeper.http server"); } } catch (InterruptedException ie) { Thread.currentThread().interrupt(); - LOG.error("Failed to start org.apache.bookkeeper.http server on port {}", port, ie); + log.error().exception(ie).attr("port", port) + .log("Failed to start org.apache.bookkeeper.http server"); } catch (ExecutionException e) { - LOG.error("Failed to start org.apache.bookkeeper.http server on port {}", port, e); + log.error().exception(e).attr("port", port) + .log("Failed to start org.apache.bookkeeper.http server"); } return false; } @@ -135,18 +136,18 @@ public void stopServer() { try { httpServiceProvider.close(); } catch (IOException ioe) { - LOG.error("Error while close httpServiceProvider", ioe); + log.error().exception(ioe).log("Error while close httpServiceProvider"); } vertx.close(asyncResult -> { isRunning = false; shutdownLatch.countDown(); - LOG.info("HTTP server is shutdown"); + log.info("HTTP server is shutdown"); }); try { shutdownLatch.await(); } catch (InterruptedException e) { Thread.currentThread().interrupt(); - LOG.error("Interrupted while shutting down org.apache.bookkeeper.http server"); + log.error("Interrupted while shutting down org.apache.bookkeeper.http server"); } } diff --git a/lombok.config b/lombok.config new file mode 100644 index 00000000000..bc68b26eebd --- /dev/null +++ b/lombok.config @@ -0,0 +1,24 @@ +# +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. +# + +# this is the top level Lombok configuration file +# see https://projectlombok.org/features/configuration for reference + +config.stopBubbling = true +lombok.log.custom.declaration = io.github.merlimat.slog.Logger io.github.merlimat.slog.Logger.get(TYPE) diff --git a/metadata-drivers/etcd/src/main/java/io/etcd/jetcd/impl/EtcdConnectionManager.java b/metadata-drivers/etcd/src/main/java/io/etcd/jetcd/impl/EtcdConnectionManager.java index db090135c90..54937c39c56 100644 --- a/metadata-drivers/etcd/src/main/java/io/etcd/jetcd/impl/EtcdConnectionManager.java +++ b/metadata-drivers/etcd/src/main/java/io/etcd/jetcd/impl/EtcdConnectionManager.java @@ -20,12 +20,12 @@ import io.etcd.jetcd.Client; import io.etcd.jetcd.api.WatchGrpc; -import lombok.extern.slf4j.Slf4j; +import lombok.CustomLog; /** * Keep a reference to etcd internal connection manager. */ -@Slf4j +@CustomLog public class EtcdConnectionManager { private final ClientImpl client; @@ -42,11 +42,11 @@ public EtcdConnectionManager(Client client) { client, "connectionManager" ); } catch (NoSuchFieldException e) { - log.error("No `connectionManager` field found in etcd client", e); + log.error().exception(e).log("No `connectionManager` field found in etcd client"); throw new RuntimeException( "No `connectionManager` field found in etcd client", e); } catch (IllegalAccessException e) { - log.error("Illegal access to `connectionManager` field in etcd client", e); + log.error().exception(e).log("Illegal access to `connectionManager` field in etcd client"); throw new RuntimeException( "Illegal access to `connectionManager` field in etcd client", e); } diff --git a/metadata-drivers/etcd/src/main/java/org/apache/bookkeeper/metadata/etcd/Etcd64bitIdGenerator.java b/metadata-drivers/etcd/src/main/java/org/apache/bookkeeper/metadata/etcd/Etcd64bitIdGenerator.java index 652033e4e9f..5eee33178ce 100644 --- a/metadata-drivers/etcd/src/main/java/org/apache/bookkeeper/metadata/etcd/Etcd64bitIdGenerator.java +++ b/metadata-drivers/etcd/src/main/java/org/apache/bookkeeper/metadata/etcd/Etcd64bitIdGenerator.java @@ -34,7 +34,7 @@ import java.nio.charset.StandardCharsets; import java.util.concurrent.ThreadLocalRandom; import java.util.concurrent.atomic.AtomicIntegerFieldUpdater; -import lombok.extern.slf4j.Slf4j; +import lombok.CustomLog; import org.apache.bookkeeper.client.BKException.Code; import org.apache.bookkeeper.meta.LedgerIdGenerator; import org.apache.bookkeeper.proto.BookkeeperInternalCallbacks.GenericCallback; @@ -45,7 +45,7 @@ *

The most significant 8 bits is used as bucket id. The remaining 56 bits are * used as the id generated per bucket. */ -@Slf4j +@CustomLog class Etcd64bitIdGenerator implements LedgerIdGenerator { static final long MAX_ID_PER_BUCKET = 0x00ffffffffffffffL; @@ -115,7 +115,9 @@ public void generateLedgerId(GenericCallback cb) { if (resp.getCount() > 0) { KeyValue kv = resp.getKvs().get(0); if (kv.getVersion() > MAX_ID_PER_BUCKET) { - log.warn("Etcd bucket '{}' is overflowed", bucketKey.toString(StandardCharsets.UTF_8)); + log.warn() + .attr("bucketKey", bucketKey.toString(StandardCharsets.UTF_8)) + .log("Etcd bucket is overflowed"); // the bucket is overflowed, moved to next bucket. generateLedgerId(cb); } else { diff --git a/metadata-drivers/etcd/src/main/java/org/apache/bookkeeper/metadata/etcd/EtcdBookieRegister.java b/metadata-drivers/etcd/src/main/java/org/apache/bookkeeper/metadata/etcd/EtcdBookieRegister.java index 2116e575de9..b73e8ce879a 100644 --- a/metadata-drivers/etcd/src/main/java/org/apache/bookkeeper/metadata/etcd/EtcdBookieRegister.java +++ b/metadata-drivers/etcd/src/main/java/org/apache/bookkeeper/metadata/etcd/EtcdBookieRegister.java @@ -34,8 +34,8 @@ import java.util.concurrent.TimeoutException; import java.util.function.Supplier; import lombok.AccessLevel; +import lombok.CustomLog; import lombok.Getter; -import lombok.extern.slf4j.Slf4j; import org.apache.bookkeeper.bookie.BookieException.MetadataStoreException; import org.apache.bookkeeper.common.concurrent.FutureUtils; import org.apache.bookkeeper.discover.RegistrationManager.RegistrationListener; @@ -43,7 +43,7 @@ /** * Register to register a bookie in Etcd. */ -@Slf4j +@CustomLog class EtcdBookieRegister implements AutoCloseable, Runnable, Supplier { private final Lease leaseClient; @@ -103,26 +103,31 @@ private void newLeaseIfNeeded() throws MetadataStoreException { this.kaListener = leaseClient.keepAlive(leaseId, new StreamObserver() { @Override public void onNext(LeaseKeepAliveResponse response) { - log.info("KeepAlive response : lease = {}, ttl = {}", - response.getID(), response.getTTL()); + log.info() + .attr("leaseId", response.getID()) + .attr("ttl", response.getTTL()) + .log("KeepAlive response"); } @Override public void onError(Throwable t) { - log.info("KeepAlive renewal failed, leaseId {}", leaseId, t.fillInStackTrace()); + log.info() + .attr("leaseId", leaseId) + .exception(t.fillInStackTrace()) + .log("KeepAlive renewal failed"); keepAliveFuture.completeExceptionally(t); } @Override public void onCompleted() { - log.info("lease completed! leaseId {}", leaseId); + log.info().attr("leaseId", leaseId).log("lease completed!"); keepAliveFuture.cancel(true); } }); this.leaseId = leaseId; leaseFuture.complete(leaseId); - log.info("New lease '{}' is granted.", leaseId); + log.info().attr("leaseId", leaseId).log("New lease is granted"); } } @@ -132,14 +137,19 @@ private void waitForNewLeaseId() { newLeaseIfNeeded(); nextWaitTimeMs = 100L; } catch (MetadataStoreException e) { - log.error("Failed to grant a new lease for leaseId {}", leaseId, e); + log.error() + .attr("leaseId", leaseId) + .exception(e) + .log("Failed to grant a new lease"); try { TimeUnit.MILLISECONDS.sleep(nextWaitTimeMs); nextWaitTimeMs *= 2; nextWaitTimeMs = Math.min(nextWaitTimeMs, TimeUnit.SECONDS.toMillis(ttlSeconds)); } catch (InterruptedException e1) { Thread.currentThread().interrupt(); - log.warn("Interrupted at backing off granting a new lease for leaseId {}", leaseId); + log.warn() + .attr("leaseId", leaseId) + .log("Interrupted at backing off granting a new lease"); } continue; } @@ -152,15 +162,18 @@ public void run() { waitForNewLeaseId(); // here we get a lease, keep it alive try { - log.info("Keeping Alive at lease = {}", get()); + log.info().attr("leaseId", get()).log("Keeping Alive"); keepAliveFuture.get(); continue; } catch (InterruptedException ie) { Thread.currentThread().interrupt(); - log.warn("Interrupted at keeping lease '{}' alive", leaseId); + log.warn().attr("leaseId", leaseId).log("Interrupted at keeping lease alive"); resetLease(); } catch (ExecutionException ee) { - log.warn("Failed to keep alive lease '{}'", leaseId, ee); + log.warn() + .attr("leaseId", leaseId) + .exception(ee) + .log("Failed to keep alive lease"); resetLease(); } } @@ -207,7 +220,7 @@ public Long get() { try { return leaseFuture.get(100, TimeUnit.MILLISECONDS); } catch (InterruptedException e) { - log.warn("Interrupted at getting lease id", e); + log.warn().exception(e).log("Interrupted at getting lease id"); return -1L; } catch (ExecutionException e) { throw new IllegalArgumentException("Should never reach here"); diff --git a/metadata-drivers/etcd/src/main/java/org/apache/bookkeeper/metadata/etcd/EtcdLayoutManager.java b/metadata-drivers/etcd/src/main/java/org/apache/bookkeeper/metadata/etcd/EtcdLayoutManager.java index 6a7344e2841..cc0e576497f 100644 --- a/metadata-drivers/etcd/src/main/java/org/apache/bookkeeper/metadata/etcd/EtcdLayoutManager.java +++ b/metadata-drivers/etcd/src/main/java/org/apache/bookkeeper/metadata/etcd/EtcdLayoutManager.java @@ -33,15 +33,15 @@ import java.io.IOException; import java.nio.charset.StandardCharsets; import lombok.AccessLevel; +import lombok.CustomLog; import lombok.Getter; -import lombok.extern.slf4j.Slf4j; import org.apache.bookkeeper.meta.LayoutManager; import org.apache.bookkeeper.meta.LedgerLayout; /** * Etcd based layout manager. */ -@Slf4j +@CustomLog @Getter(AccessLevel.PACKAGE) class EtcdLayoutManager implements LayoutManager { @@ -97,9 +97,8 @@ public void storeLedgerLayout(LedgerLayout layout) throws IOException { public void deleteLedgerLayout() throws IOException { DeleteResponse response = ioResult(kvClient.delete(layoutKey)); if (response.getDeleted() > 0) { - if (log.isDebugEnabled()) { - log.debug("Successfully delete layout '{}'", layoutKey.toString(StandardCharsets.UTF_8)); - } + log.debug(e -> e.attr("layoutKey", layoutKey.toString(StandardCharsets.UTF_8)) + .log("Successfully delete layout")); return; } else { throw new IOException("No ledger layout is found under '" + layoutKey.toString(StandardCharsets.UTF_8) diff --git a/metadata-drivers/etcd/src/main/java/org/apache/bookkeeper/metadata/etcd/EtcdLedgerManager.java b/metadata-drivers/etcd/src/main/java/org/apache/bookkeeper/metadata/etcd/EtcdLedgerManager.java index 1f211b28dcd..dcd3f1e2372 100644 --- a/metadata-drivers/etcd/src/main/java/org/apache/bookkeeper/metadata/etcd/EtcdLedgerManager.java +++ b/metadata-drivers/etcd/src/main/java/org/apache/bookkeeper/metadata/etcd/EtcdLedgerManager.java @@ -41,7 +41,7 @@ import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentMap; import java.util.function.Consumer; -import lombok.extern.slf4j.Slf4j; +import lombok.CustomLog; import org.apache.bookkeeper.client.BKException; import org.apache.bookkeeper.client.api.LedgerMetadata; import org.apache.bookkeeper.meta.LedgerManager; @@ -60,7 +60,7 @@ /** * Etcd ledger manager. */ -@Slf4j +@CustomLog class EtcdLedgerManager implements LedgerManager { private final LedgerMetadataSerDe serDe = new LedgerMetadataSerDe(); @@ -97,7 +97,7 @@ public CompletableFuture> createLedgerMetadata(long le LedgerMetadata metadata) { CompletableFuture> promise = new CompletableFuture<>(); String ledgerKey = EtcdUtils.getLedgerKey(scope, ledgerId); - log.info("Create ledger metadata under key {}", ledgerKey); + log.info().attr("ledgerKey", ledgerKey).log("Create ledger metadata"); ByteSequence ledgerKeyBs = ByteSequence.from(ledgerKey, StandardCharsets.UTF_8); final ByteSequence valueBs; @@ -149,12 +149,14 @@ public CompletableFuture removeLedgerMetadata(long ledgerId, Version versi CompletableFuture promise = new CompletableFuture<>(); long revision = -0xabcd; if (Version.NEW == version) { - log.error("Request to delete ledger {} metadata with version set to the initial one", ledgerId); + log.error() + .attr("ledgerId", ledgerId) + .log("Request to delete ledger metadata with version set to the initial one"); promise.completeExceptionally(new BKException.BKMetadataVersionException()); return promise; } else if (Version.ANY != version) { if (!(version instanceof LongVersion)) { - log.info("Not an instance of LongVersion : {}", ledgerId); + log.info().attr("ledgerId", ledgerId).log("Not an instance of LongVersion"); promise.completeExceptionally(new BKException.BKMetadataVersionException()); return promise; } else { @@ -197,7 +199,10 @@ public CompletableFuture removeLedgerMetadata(long ledgerId, Version versi // fail to delete the ledger promise.completeExceptionally(new BKException.BKMetadataVersionException()); } else { - log.warn("Deleting ledger {} failed due to : ledger key {} doesn't exist", ledgerId, ledgerKey); + log.warn() + .attr("ledgerId", ledgerId) + .attr("ledgerKey", ledgerKey) + .log("Deleting ledger failed: ledger key doesn't exist"); promise.completeExceptionally(new BKException.BKNoSuchLedgerExistsException()); } } @@ -214,7 +219,7 @@ public CompletableFuture> readLedgerMetadata(long ledg CompletableFuture> promise = new CompletableFuture<>(); String ledgerKey = EtcdUtils.getLedgerKey(scope, ledgerId); ByteSequence ledgerKeyBs = ByteSequence.from(ledgerKey, StandardCharsets.UTF_8); - log.info("read ledger metadata under key {}", ledgerKey); + log.info().attr("ledgerKey", ledgerKey).log("read ledger metadata"); kvClient.get(ledgerKeyBs) .thenAccept(getResp -> { if (getResp.getCount() > 0) { @@ -224,7 +229,10 @@ public CompletableFuture> readLedgerMetadata(long ledg LedgerMetadata metadata = serDe.parseConfig(data, ledgerId, Optional.empty()); promise.complete(new Versioned<>(metadata, new LongVersion(kv.getModRevision()))); } catch (IOException ioe) { - log.error("Could not parse ledger metadata for ledger : {}", ledgerId, ioe); + log.error() + .attr("ledgerId", ledgerId) + .exception(ioe) + .log("Could not parse ledger metadata for ledger"); promise.completeExceptionally(new BKException.MetaStoreException()); return; } @@ -278,9 +286,10 @@ public CompletableFuture> writeLedgerMetadata(long led } else { GetResponse getResp = resp.getGetResponses().get(0); if (getResp.getCount() > 0) { - log.warn("Conditional update ledger metadata failed :" - + " expected version = {}, actual version = {}", - getResp.getKvs().get(0).getModRevision(), lv); + log.warn() + .attr("expectedVersion", getResp.getKvs().get(0).getModRevision()) + .attr("actualVersion", lv) + .log("Conditional update ledger metadata failed"); promise.completeExceptionally(new BKException.BKMetadataVersionException()); } else { promise.completeExceptionally(new BKException.BKNoSuchLedgerExistsException()); @@ -322,8 +331,10 @@ public void registerLedgerMetadataListener(long ledgerId, LedgerMetadataListener Optional.empty() ); } catch (IOException ioe) { - log.error("Could not parse ledger metadata : {}", - bs.toString(StandardCharsets.UTF_8), ioe); + log.error() + .attr("metadata", bs.toString(StandardCharsets.UTF_8)) + .exception(ioe) + .log("Could not parse ledger metadata"); throw new RuntimeException( "Could not parse ledger metadata : " + bs.toString(StandardCharsets.UTF_8), ioe); @@ -334,7 +345,7 @@ public void registerLedgerMetadataListener(long ledgerId, LedgerMetadataListener LedgerMetadataConsumer lmConsumer = listenerToConsumer(ledgerId, listener, (lid) -> { if (watchers.remove(lid, lmStream)) { - log.info("Closed ledger metadata watcher on ledger {} deletion.", lid); + log.info().attr("ledgerId", lid).log("Closed ledger metadata watcher on ledger deletion"); lmStream.closeAsync(); } }); @@ -366,8 +377,9 @@ private void unregisterLedgerMetadataListener(long ledgerId, LedgerMetadataConsu lmStream.unwatch(lmConsumer).thenAccept(noConsumers -> { if (noConsumers) { if (watchers.remove(ledgerId, lmStream)) { - log.info("Closed ledger metadata watcher on ledger {} since there are no listeners any more.", - ledgerId); + log.info() + .attr("ledgerId", ledgerId) + .log("Closed ledger metadata watcher since there are no listeners any more"); lmStream.closeAsync(); } } diff --git a/metadata-drivers/etcd/src/main/java/org/apache/bookkeeper/metadata/etcd/EtcdMetadataBookieDriver.java b/metadata-drivers/etcd/src/main/java/org/apache/bookkeeper/metadata/etcd/EtcdMetadataBookieDriver.java index 4cbb352fda0..36d08be93f6 100644 --- a/metadata-drivers/etcd/src/main/java/org/apache/bookkeeper/metadata/etcd/EtcdMetadataBookieDriver.java +++ b/metadata-drivers/etcd/src/main/java/org/apache/bookkeeper/metadata/etcd/EtcdMetadataBookieDriver.java @@ -15,7 +15,7 @@ package org.apache.bookkeeper.metadata.etcd; import java.util.concurrent.TimeUnit; -import lombok.extern.slf4j.Slf4j; +import lombok.CustomLog; import org.apache.bookkeeper.conf.ServerConfiguration; import org.apache.bookkeeper.discover.RegistrationManager; import org.apache.bookkeeper.meta.MetadataBookieDriver; @@ -26,7 +26,7 @@ /** * Etcd based metadata bookie driver. */ -@Slf4j +@CustomLog public class EtcdMetadataBookieDriver extends EtcdMetadataDriverBase implements MetadataBookieDriver { // register myself diff --git a/metadata-drivers/etcd/src/main/java/org/apache/bookkeeper/metadata/etcd/EtcdMetadataClientDriver.java b/metadata-drivers/etcd/src/main/java/org/apache/bookkeeper/metadata/etcd/EtcdMetadataClientDriver.java index 07d64186aa3..ac3af27aaf7 100644 --- a/metadata-drivers/etcd/src/main/java/org/apache/bookkeeper/metadata/etcd/EtcdMetadataClientDriver.java +++ b/metadata-drivers/etcd/src/main/java/org/apache/bookkeeper/metadata/etcd/EtcdMetadataClientDriver.java @@ -19,7 +19,7 @@ import java.util.Optional; import java.util.concurrent.ScheduledExecutorService; -import lombok.extern.slf4j.Slf4j; +import lombok.CustomLog; import org.apache.bookkeeper.conf.ClientConfiguration; import org.apache.bookkeeper.discover.RegistrationClient; import org.apache.bookkeeper.meta.MetadataClientDriver; @@ -30,7 +30,7 @@ /** * Etcd based metadata client driver. */ -@Slf4j +@CustomLog public class EtcdMetadataClientDriver extends EtcdMetadataDriverBase implements MetadataClientDriver { // register myself to driver manager diff --git a/metadata-drivers/etcd/src/main/java/org/apache/bookkeeper/metadata/etcd/EtcdMetadataDriverBase.java b/metadata-drivers/etcd/src/main/java/org/apache/bookkeeper/metadata/etcd/EtcdMetadataDriverBase.java index 7ce9bf5934e..5793765ddd9 100644 --- a/metadata-drivers/etcd/src/main/java/org/apache/bookkeeper/metadata/etcd/EtcdMetadataDriverBase.java +++ b/metadata-drivers/etcd/src/main/java/org/apache/bookkeeper/metadata/etcd/EtcdMetadataDriverBase.java @@ -22,7 +22,7 @@ import java.io.IOException; import java.util.List; import java.util.stream.Collectors; -import lombok.extern.slf4j.Slf4j; +import lombok.CustomLog; import org.apache.bookkeeper.common.net.ServiceURI; import org.apache.bookkeeper.conf.AbstractConfiguration; import org.apache.bookkeeper.meta.LayoutManager; @@ -35,7 +35,7 @@ /** * This is a mixin class for supporting etcd based metadata drivers. */ -@Slf4j +@CustomLog class EtcdMetadataDriverBase implements AutoCloseable { static final String SCHEME = "etcd"; @@ -71,7 +71,7 @@ protected void initialize(AbstractConfiguration conf, StatsLogger statsLogger try { metadataServiceUriStr = conf.getMetadataServiceUri(); } catch (ConfigurationException ce) { - log.error("Failed to retrieve metadata service uri from configuration", ce); + log.error().exception(ce).log("Failed to retrieve metadata service uri from configuration"); throw new MetadataException(Code.INVALID_METADATA_SERVICE_URI, ce); } ServiceURI serviceURI = ServiceURI.create(metadataServiceUriStr); @@ -82,8 +82,10 @@ protected void initialize(AbstractConfiguration conf, StatsLogger statsLogger .map(host -> String.format("http://%s", host)) .collect(Collectors.toList()); - log.info("Initializing etcd metadata driver : etcd endpoints = {}, key scope = {}", - etcdEndpoints, keyPrefix); + log.info() + .attr("etcdEndpoints", etcdEndpoints) + .attr("keyScope", keyPrefix) + .log("Initializing etcd metadata driver"); synchronized (this) { this.client = Client.builder() @@ -121,7 +123,7 @@ public synchronized void close() { try { lmFactory.close(); } catch (IOException e) { - log.error("Failed to close ledger manager factory", e); + log.error().exception(e).log("Failed to close ledger manager factory"); } lmFactory = null; } diff --git a/metadata-drivers/etcd/src/main/java/org/apache/bookkeeper/metadata/etcd/EtcdRegistrationClient.java b/metadata-drivers/etcd/src/main/java/org/apache/bookkeeper/metadata/etcd/EtcdRegistrationClient.java index 3dd7808d619..3cc7eb534ef 100644 --- a/metadata-drivers/etcd/src/main/java/org/apache/bookkeeper/metadata/etcd/EtcdRegistrationClient.java +++ b/metadata-drivers/etcd/src/main/java/org/apache/bookkeeper/metadata/etcd/EtcdRegistrationClient.java @@ -27,7 +27,6 @@ import java.util.concurrent.CompletableFuture; import java.util.function.Consumer; import java.util.function.Function; -import lombok.extern.slf4j.Slf4j; import org.apache.bookkeeper.client.BKException; import org.apache.bookkeeper.common.concurrent.FutureUtils; import org.apache.bookkeeper.discover.RegistrationClient; @@ -38,7 +37,6 @@ /** * Etcd based registration client. */ -@Slf4j class EtcdRegistrationClient implements RegistrationClient { private static Function newBookieSocketAddressFunc(String prefix) { diff --git a/metadata-drivers/etcd/src/main/java/org/apache/bookkeeper/metadata/etcd/EtcdRegistrationManager.java b/metadata-drivers/etcd/src/main/java/org/apache/bookkeeper/metadata/etcd/EtcdRegistrationManager.java index 50badb8f07f..9d9322968e4 100644 --- a/metadata-drivers/etcd/src/main/java/org/apache/bookkeeper/metadata/etcd/EtcdRegistrationManager.java +++ b/metadata-drivers/etcd/src/main/java/org/apache/bookkeeper/metadata/etcd/EtcdRegistrationManager.java @@ -61,8 +61,8 @@ import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeoutException; import lombok.AccessLevel; +import lombok.CustomLog; import lombok.Getter; -import lombok.extern.slf4j.Slf4j; import org.apache.bookkeeper.bookie.BookieException; import org.apache.bookkeeper.bookie.BookieException.BookieIllegalOpException; import org.apache.bookkeeper.bookie.BookieException.CookieNotFoundException; @@ -78,7 +78,7 @@ /** * Etcd registration manager. */ -@Slf4j +@CustomLog class EtcdRegistrationManager implements RegistrationManager { private final String scope; @@ -137,10 +137,10 @@ private EtcdRegistrationManager(Client client, @Override public void close() { if (ownClient) { - log.info("Closing registration manager under scope '{}'", scope); + log.info().attr("scope", scope).log("Closing registration manager"); bkRegister.close(); client.close(); - log.info("Successfully closed registration manager under scope '{}'", scope); + log.info().attr("scope", scope).log("Successfully closed registration manager"); } } @@ -186,8 +186,11 @@ private boolean waitUntilRegNodeExpired(String regPath, long leaseId) .build(), response -> { for (WatchEvent event : response.getEvents()) { - log.info("Received watch event on '{}' : EventType = {}, lease {}", - regPath, event.getEventType(), leaseId); + log.info() + .attr("regPath", regPath) + .attr("eventType", event.getEventType()) + .attr("leaseId", leaseId) + .log("Received watch event"); if (EventType.DELETE == event.getEventType()) { watchFuture.complete(null); return; @@ -195,16 +198,23 @@ private boolean waitUntilRegNodeExpired(String regPath, long leaseId) } }, exception -> { - log.warn("Exception in keepAlive for watch event on {}, lease {}", - regPath, leaseId, exception); + log.warn() + .attr("regPath", regPath) + .attr("leaseId", leaseId) + .exception(exception) + .log("Exception in keepAlive for watch event"); watchFuture.completeExceptionally(new UncheckedExecutionException( "Interrupted at waiting previous registration under " + regPath + " (lease = " + kv.getLease() + ") to be expired", exception)); } ); - log.info("Previous bookie registration (lease = {}) still exists at {}, " - + "so new lease '{}' will be waiting previous lease for {} seconds to be expired", - kv.getLease(), regPath, leaseId, bkRegister.getTtlSeconds()); + log.info() + .attr("previousLeaseId", kv.getLease()) + .attr("regPath", regPath) + .attr("newLeaseId", leaseId) + .attr("ttlSeconds", bkRegister.getTtlSeconds()) + .log("Previous bookie registration still exists, new lease will wait for previous lease" + + " to be expired"); try { msResult(watchFuture, 2 * bkRegister.getTtlSeconds(), TimeUnit.SECONDS); @@ -261,7 +271,7 @@ private void doRegisterBookie(String regPath, long leaseId) throws MetadataStore + regPath + "': lease = " + kv.getLease()); } } else { - log.info("Successfully registered bookie at {}", regPath); + log.info().attr("regPath", regPath).log("Successfully registered bookie"); } } @@ -282,9 +292,12 @@ public void unregisterBookie(BookieId bookieId, boolean readOnly) throws BookieE } DeleteResponse delResp = msResult(kvClient.delete(ByteSequence.from(regPath, UTF_8))); if (delResp.getDeleted() > 0) { - log.info("Successfully unregistered bookie {} from {}", bookieId, regPath); + log.info() + .attr("bookieId", bookieId) + .attr("regPath", regPath) + .log("Successfully unregistered bookie"); } else { - log.info("Bookie disappeared from {} before unregistering", regPath); + log.info().attr("regPath", regPath).log("Bookie disappeared before unregistering"); } } @@ -385,8 +398,10 @@ public void removeCookie(BookieId bookieId, Version version) throws BookieExcept throw new CookieNotFoundException(bookieId.toString()); } } else { - log.info("Removed cookie from {} for bookie {}", - cookiePath.toString(UTF_8), bookieId); + log.info() + .attr("cookiePath", cookiePath.toString(UTF_8)) + .attr("bookieId", bookieId) + .log("Removed cookie"); } } @@ -396,8 +411,8 @@ public String getClusterInstanceId() throws BookieException { kvClient.get(ByteSequence.from(getClusterInstanceIdPath(scope), UTF_8))); if (response.getCount() <= 0) { log.error("BookKeeper metadata doesn't exist in Etcd. " - + "Has the cluster been initialized? " - + "Try running bin/bookkeeper shell initNewCluster"); + + "Has the cluster been initialized? " + + "Try running bin/bookkeeper shell initNewCluster"); throw new MetadataStoreException("BookKeeper is not initialized under '" + scope + "' yet"); } else { KeyValue kv = response.getKvs().get(0); @@ -529,8 +544,9 @@ static boolean nukeExistingCluster(KV kvClient, String scope) throws Exception { ByteSequence rootScopeKey = ByteSequence.from(scope, UTF_8); GetResponse resp = msResult(kvClient.get(rootScopeKey)); if (resp.getCount() <= 0) { - log.info("There is no existing cluster with under scope '{}' in Etcd, " - + "so exiting nuke operation", scope); + log.info() + .attr("scope", scope) + .log("There is no existing cluster under scope in Etcd, so exiting nuke operation"); return true; } @@ -559,7 +575,7 @@ static boolean nukeExistingCluster(KV kvClient, String scope) throws Exception { } if (hasBookiesAlive) { log.error("Bookies are still up and connected to this cluster, " - + "stop all bookies before nuking the cluster"); + + "stop all bookies before nuking the cluster"); return false; } DeleteResponse delResp = msResult(kvClient.delete( @@ -567,8 +583,10 @@ static boolean nukeExistingCluster(KV kvClient, String scope) throws Exception { DeleteOption.newBuilder() .withRange(ByteSequence.from(getScopeEndKey(scope), UTF_8)) .build())); - log.info("Successfully nuked cluster under scope '{}' : {} kv pairs deleted", - scope, delResp.getDeleted()); + log.info() + .attr("scope", scope) + .attr("kvPairsDeleted", delResp.getDeleted()) + .log("Successfully nuked cluster"); return true; } } diff --git a/metadata-drivers/etcd/src/main/java/org/apache/bookkeeper/metadata/etcd/EtcdWatchClient.java b/metadata-drivers/etcd/src/main/java/org/apache/bookkeeper/metadata/etcd/EtcdWatchClient.java index 6a046ba4e2f..092ab18a870 100644 --- a/metadata-drivers/etcd/src/main/java/org/apache/bookkeeper/metadata/etcd/EtcdWatchClient.java +++ b/metadata-drivers/etcd/src/main/java/org/apache/bookkeeper/metadata/etcd/EtcdWatchClient.java @@ -47,7 +47,7 @@ import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.TimeUnit; import java.util.function.BiConsumer; -import lombok.extern.slf4j.Slf4j; +import lombok.CustomLog; import org.apache.bookkeeper.common.concurrent.FutureUtils; import org.apache.bookkeeper.common.util.OrderedScheduler; import org.apache.bookkeeper.util.collections.ConcurrentLongHashMap; @@ -56,7 +56,7 @@ /** * An async watch implementation. */ -@Slf4j +@CustomLog public class EtcdWatchClient implements AutoCloseable { private final EtcdConnectionManager connMgr; @@ -170,7 +170,7 @@ public void close() { try { FutureUtils.result(closeAsync()); } catch (Exception e) { - log.warn("Encountered exceptions on closing watch client", e); + log.warn().exception(e).log("Encountered exceptions on closing watch client"); } this.scheduler.forceShutdown(10, TimeUnit.SECONDS); } diff --git a/metadata-drivers/etcd/src/main/java/org/apache/bookkeeper/metadata/etcd/EtcdWatcher.java b/metadata-drivers/etcd/src/main/java/org/apache/bookkeeper/metadata/etcd/EtcdWatcher.java index 203bfb14253..7a74d7d5f4b 100644 --- a/metadata-drivers/etcd/src/main/java/org/apache/bookkeeper/metadata/etcd/EtcdWatcher.java +++ b/metadata-drivers/etcd/src/main/java/org/apache/bookkeeper/metadata/etcd/EtcdWatcher.java @@ -26,15 +26,15 @@ import java.util.concurrent.CopyOnWriteArraySet; import java.util.concurrent.ScheduledExecutorService; import java.util.function.BiConsumer; +import lombok.CustomLog; import lombok.Getter; import lombok.Setter; -import lombok.extern.slf4j.Slf4j; import org.apache.bookkeeper.common.concurrent.FutureUtils; /** * Watcher class holds watcher information. */ -@Slf4j +@CustomLog public class EtcdWatcher implements AutoCloseable { private final ScheduledExecutorService executor; @@ -107,8 +107,10 @@ public void close() { try { FutureUtils.result(closeAsync()); } catch (Exception e) { - log.warn("Encountered error on removing watcher '{}' from watch client : {}", - watchID, e.getMessage()); + log.warn() + .attr("watchId", watchID) + .exceptionMessage(e) + .log("Encountered error on removing watcher from watch client"); } consumers.clear(); } diff --git a/metadata-drivers/etcd/src/main/java/org/apache/bookkeeper/metadata/etcd/helpers/KeySetReader.java b/metadata-drivers/etcd/src/main/java/org/apache/bookkeeper/metadata/etcd/helpers/KeySetReader.java index 5f8afdb6e40..50727dc6615 100644 --- a/metadata-drivers/etcd/src/main/java/org/apache/bookkeeper/metadata/etcd/helpers/KeySetReader.java +++ b/metadata-drivers/etcd/src/main/java/org/apache/bookkeeper/metadata/etcd/helpers/KeySetReader.java @@ -35,7 +35,7 @@ import java.util.function.Consumer; import java.util.function.Function; import java.util.stream.Collectors; -import lombok.extern.slf4j.Slf4j; +import lombok.CustomLog; import org.apache.bookkeeper.common.concurrent.FutureUtils; import org.apache.bookkeeper.metadata.etcd.EtcdWatchClient; import org.apache.bookkeeper.metadata.etcd.EtcdWatcher; @@ -45,7 +45,7 @@ /** * A helper class to read a set of keys and watch them. */ -@Slf4j +@CustomLog public class KeySetReader implements BiConsumer, AutoCloseable { private final Client client; @@ -131,10 +131,10 @@ private synchronized Versioned> processWatchResponse(WatchResponse respon if (null != closeFuture) { return null; } - if (log.isDebugEnabled()) { - log.debug("Received watch response : revision = {}, {} events = {}", - response.getHeader().getRevision(), response.getEvents().size(), response.getEvents()); - } + log.debug(e -> e.attr("revision", response.getHeader().getRevision()) + .attr("eventCount", response.getEvents().size()) + .attr("events", response.getEvents()) + .log("Received watch response")); if (response.getHeader().getRevision() <= revision) { return null; @@ -276,7 +276,7 @@ public void close() { try { FutureUtils.result(closeAsync()); } catch (Exception e) { - log.warn("Encountered exceptions on closing key reader : {}", e.getMessage()); + log.warn().exceptionMessage(e).log("Encountered exceptions on closing key reader"); } } } diff --git a/metadata-drivers/etcd/src/main/java/org/apache/bookkeeper/metadata/etcd/helpers/KeyStream.java b/metadata-drivers/etcd/src/main/java/org/apache/bookkeeper/metadata/etcd/helpers/KeyStream.java index 268daf42e43..2f23ae70378 100644 --- a/metadata-drivers/etcd/src/main/java/org/apache/bookkeeper/metadata/etcd/helpers/KeyStream.java +++ b/metadata-drivers/etcd/src/main/java/org/apache/bookkeeper/metadata/etcd/helpers/KeyStream.java @@ -30,13 +30,13 @@ import java.util.concurrent.CompletableFuture; import java.util.function.Function; import java.util.stream.Collectors; -import lombok.extern.slf4j.Slf4j; +import lombok.CustomLog; import org.apache.bookkeeper.common.concurrent.FutureUtils; /** * Read a range of key/value pairs in a streaming way. */ -@Slf4j +@CustomLog public class KeyStream { private final KV kvClient; @@ -82,9 +82,9 @@ public CompletableFuture> readNext() { batchSize += 1; } } - if (log.isTraceEnabled()) { - log.trace("Read keys between {} and {}", beginKey.toString(UTF_8), endKey.toString(UTF_8)); - } + log.trace(e -> e.attr("beginKey", beginKey.toString(UTF_8)) + .attr("endKey", endKey.toString(UTF_8)) + .log("Read keys")); return kvClient.get( beginKey, GetOption.newBuilder() diff --git a/metadata-drivers/etcd/src/main/java/org/apache/bookkeeper/metadata/etcd/helpers/ValueStream.java b/metadata-drivers/etcd/src/main/java/org/apache/bookkeeper/metadata/etcd/helpers/ValueStream.java index 25e47df3323..d7afc23ba26 100644 --- a/metadata-drivers/etcd/src/main/java/org/apache/bookkeeper/metadata/etcd/helpers/ValueStream.java +++ b/metadata-drivers/etcd/src/main/java/org/apache/bookkeeper/metadata/etcd/helpers/ValueStream.java @@ -28,7 +28,7 @@ import java.util.function.BiConsumer; import java.util.function.Consumer; import java.util.function.Function; -import lombok.extern.slf4j.Slf4j; +import lombok.CustomLog; import org.apache.bookkeeper.common.concurrent.FutureUtils; import org.apache.bookkeeper.metadata.etcd.EtcdWatchClient; import org.apache.bookkeeper.metadata.etcd.EtcdWatcher; @@ -38,7 +38,7 @@ /** * A helper class to read the stream of values of a given key. */ -@Slf4j +@CustomLog public class ValueStream implements BiConsumer, AutoCloseable { private final Client client; @@ -117,10 +117,10 @@ private synchronized Versioned processWatchResponse(WatchResponse response) { if (null != closeFuture) { return null; } - if (log.isDebugEnabled()) { - log.debug("Received watch response : revision = {}, {} events = {}", - response.getHeader().getRevision(), response.getEvents().size(), response.getEvents()); - } + log.debug(e -> e.attr("revision", response.getHeader().getRevision()) + .attr("eventCount", response.getEvents().size()) + .attr("events", response.getEvents()) + .log("Received watch response")); if (response.getHeader().getRevision() <= revision) { return null; @@ -182,9 +182,7 @@ public CompletableFuture waitUntilWatched() { try { TimeUnit.MILLISECONDS.sleep(100); } catch (InterruptedException e) { - if (log.isDebugEnabled()) { - log.debug("Interrupted at waiting until the key is watched", e); - } + log.debug(ev -> ev.exception(e).log("Interrupted at waiting until the key is watched")); } } return wf; @@ -263,12 +261,10 @@ private CompletableFuture closeOrRewatch(boolean rewatch) { @Override public void accept(WatchResponse watchResponse, Throwable throwable) { if (null == throwable) { - if (log.isDebugEnabled()) { - log.debug("Received watch response : revision = {}, {} events = {}", - watchResponse.getHeader().getRevision(), - watchResponse.getEvents().size(), - watchResponse.getEvents()); - } + log.debug(e -> e.attr("revision", watchResponse.getHeader().getRevision()) + .attr("eventCount", watchResponse.getEvents().size()) + .attr("events", watchResponse.getEvents()) + .log("Received watch response")); synchronized (this) { Versioned localValue = processWatchResponse(watchResponse); @@ -304,7 +300,7 @@ public void close() { try { FutureUtils.result(closeAsync()); } catch (Exception e) { - log.warn("Encountered exceptions on closing key reader : {}", e.getMessage()); + log.warn().exceptionMessage(e).log("Encountered exceptions on closing key reader"); } } } diff --git a/pom.xml b/pom.xml index 29d3d24be86..d18f145ed76 100644 --- a/pom.xml +++ b/pom.xml @@ -192,6 +192,7 @@ 9.9.3 3.3.0 2.0.12 + 0.9.8 2.0 4.6.0 2.43.0 @@ -273,6 +274,11 @@ + + io.github.merlimat.slog + slog + ${slog.version} + org.slf4j slf4j-bom @@ -882,6 +888,10 @@ + + io.github.merlimat.slog + slog + org.slf4j slf4j-api diff --git a/tools/framework/src/main/java/org/apache/bookkeeper/tools/common/BKCommand.java b/tools/framework/src/main/java/org/apache/bookkeeper/tools/common/BKCommand.java index c50b7ccd37f..4036c5ed704 100644 --- a/tools/framework/src/main/java/org/apache/bookkeeper/tools/common/BKCommand.java +++ b/tools/framework/src/main/java/org/apache/bookkeeper/tools/common/BKCommand.java @@ -20,7 +20,7 @@ import com.google.common.base.Strings; import java.io.File; -import lombok.extern.slf4j.Slf4j; +import lombok.CustomLog; import org.apache.bookkeeper.common.annotation.InterfaceAudience.Private; import org.apache.bookkeeper.common.conf.ConfigurationUtil; import org.apache.bookkeeper.common.net.ServiceURI; @@ -35,7 +35,7 @@ /** * Base bk command class. */ -@Slf4j +@CustomLog public abstract class BKCommand extends CliCommand { protected BKCommand(CliSpec spec) { @@ -69,8 +69,10 @@ protected boolean apply(BKFlags bkFlags, CommandFlagsT cmdFlags) { if (null != bkFlags.serviceUri) { serviceURI = ServiceURI.create(bkFlags.serviceUri); if (!acceptServiceUri(serviceURI)) { - log.error("Unresolvable service uri by command '{}' : {}", - path(), bkFlags.serviceUri); + log.error() + .attr("command", path()) + .attr("serviceUri", bkFlags.serviceUri) + .log("Unresolvable service uri"); return false; } } @@ -82,7 +84,10 @@ protected boolean apply(BKFlags bkFlags, CommandFlagsT cmdFlags) { ConfigurationUtil.newConfiguration(c -> c.propertiesBuilder(new File(bkFlags.configFile))); conf.addConfiguration(loadedConf); } catch (ConfigurationException e) { - log.error("Malformed configuration file : {}", bkFlags.configFile, e); + log.error() + .attr("configFile", bkFlags.configFile) + .exception(e) + .log("Malformed configuration file"); throw new IllegalArgumentException(e); } } diff --git a/tools/framework/src/main/java/org/apache/bookkeeper/tools/framework/Cli.java b/tools/framework/src/main/java/org/apache/bookkeeper/tools/framework/Cli.java index 0f970713d89..81965db5419 100644 --- a/tools/framework/src/main/java/org/apache/bookkeeper/tools/framework/Cli.java +++ b/tools/framework/src/main/java/org/apache/bookkeeper/tools/framework/Cli.java @@ -27,14 +27,14 @@ import java.util.function.Function; import java.util.stream.IntStream; import lombok.AccessLevel; +import lombok.CustomLog; import lombok.Getter; -import lombok.extern.slf4j.Slf4j; import org.apache.bookkeeper.tools.framework.CliSpec.Builder; /** * Cli to execute {@link CliSpec}. */ -@Slf4j +@CustomLog public class Cli { private final CliSpec spec; diff --git a/tools/ledger/src/main/java/org/apache/bookkeeper/tools/cli/BKCtl.java b/tools/ledger/src/main/java/org/apache/bookkeeper/tools/cli/BKCtl.java index 970f4634d70..bd5ed42b303 100644 --- a/tools/ledger/src/main/java/org/apache/bookkeeper/tools/cli/BKCtl.java +++ b/tools/ledger/src/main/java/org/apache/bookkeeper/tools/cli/BKCtl.java @@ -20,7 +20,7 @@ import java.util.Iterator; import java.util.ServiceLoader; -import lombok.extern.slf4j.Slf4j; +import lombok.CustomLog; import org.apache.bookkeeper.tools.common.BKFlags; import org.apache.bookkeeper.tools.framework.Cli; import org.apache.bookkeeper.tools.framework.CliSpec; @@ -29,7 +29,7 @@ /** * bkctl interacts and operates the Apache BookKeeper cluster. */ -@Slf4j +@CustomLog public class BKCtl { public static final String NAME = "bkctl"; diff --git a/tools/perf/src/main/java/org/apache/bookkeeper/tools/perf/BKPerf.java b/tools/perf/src/main/java/org/apache/bookkeeper/tools/perf/BKPerf.java index e7ad543aaec..efbfd71f317 100644 --- a/tools/perf/src/main/java/org/apache/bookkeeper/tools/perf/BKPerf.java +++ b/tools/perf/src/main/java/org/apache/bookkeeper/tools/perf/BKPerf.java @@ -14,7 +14,7 @@ package org.apache.bookkeeper.tools.perf; -import lombok.extern.slf4j.Slf4j; +import lombok.CustomLog; import org.apache.bookkeeper.tools.common.BKFlags; import org.apache.bookkeeper.tools.framework.Cli; import org.apache.bookkeeper.tools.framework.CliSpec; @@ -22,7 +22,7 @@ /** * bkperf evaluates the performance of Apache BookKeeper cluster. */ -@Slf4j +@CustomLog public class BKPerf { public static final String NAME = "bkperf"; diff --git a/tools/perf/src/main/java/org/apache/bookkeeper/tools/perf/dlog/PerfReader.java b/tools/perf/src/main/java/org/apache/bookkeeper/tools/perf/dlog/PerfReader.java index ba0ea7ba615..cef76f36e16 100644 --- a/tools/perf/src/main/java/org/apache/bookkeeper/tools/perf/dlog/PerfReader.java +++ b/tools/perf/src/main/java/org/apache/bookkeeper/tools/perf/dlog/PerfReader.java @@ -36,7 +36,7 @@ import java.util.concurrent.Executors; import java.util.concurrent.TimeUnit; import java.util.stream.Collectors; -import lombok.extern.slf4j.Slf4j; +import lombok.CustomLog; import org.apache.bookkeeper.common.net.ServiceURI; import org.apache.commons.lang3.tuple.Pair; import org.apache.distributedlog.DLSN; @@ -48,7 +48,7 @@ /** * A perf reader to evaluate read performance. */ -@Slf4j +@CustomLog public class PerfReader extends PerfReaderBase { PerfReader(ServiceURI serviceURI, Flags flags) { @@ -62,7 +62,7 @@ protected void execute(Namespace namespace) throws Exception { String logName = String.format(flags.logName, i); managers.add(Pair.of(i, namespace.openLog(logName))); } - log.info("Successfully open {} logs", managers.size()); + log.info().attr("numLogs", managers.size()).log("Successfully opened logs"); // register shutdown hook to aggregate stats Runtime.getRuntime().addShutdownHook(new Thread(() -> { @@ -83,11 +83,11 @@ protected void execute(Namespace namespace) throws Exception { try { read(logsThisThread); } catch (Exception e) { - log.error("Encountered error at writing records", e); + log.error().exception(e).log("Encountered error at writing records"); } }); } - log.info("Started {} write threads", flags.numThreads); + log.info().attr("numThreads", flags.numThreads).log("Started write threads"); reportStats(); } finally { executor.shutdown(); @@ -99,15 +99,19 @@ protected void execute(Namespace namespace) throws Exception { } void read(List logs) throws Exception { - log.info("Read thread started with : logs = {}", - logs.stream().map(l -> l.getStreamName()).collect(Collectors.toList())); + log.info() + .attr("logs", logs.stream().map(l -> l.getStreamName()).collect(Collectors.toList())) + .log("Read thread started"); List readers = logs.stream() .map(manager -> { try { return manager.openLogReader(DLSN.InitialDLSN); } catch (IOException e) { - log.error("Failed to open reader for log stream {}", manager.getStreamName(), e); + log.error() + .attr("streamName", manager.getStreamName()) + .exception(e) + .log("Failed to open reader for log stream"); throw new UncheckedIOException(e); } }) diff --git a/tools/perf/src/main/java/org/apache/bookkeeper/tools/perf/dlog/PerfReaderBase.java b/tools/perf/src/main/java/org/apache/bookkeeper/tools/perf/dlog/PerfReaderBase.java index 9520ef9cffb..9e9fe2c56db 100644 --- a/tools/perf/src/main/java/org/apache/bookkeeper/tools/perf/dlog/PerfReaderBase.java +++ b/tools/perf/src/main/java/org/apache/bookkeeper/tools/perf/dlog/PerfReaderBase.java @@ -21,7 +21,7 @@ import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.LongAdder; -import lombok.extern.slf4j.Slf4j; +import lombok.CustomLog; import org.HdrHistogram.Histogram; import org.HdrHistogram.Recorder; import org.apache.bookkeeper.common.net.ServiceURI; @@ -31,7 +31,7 @@ import org.apache.distributedlog.api.namespace.Namespace; import org.apache.distributedlog.api.namespace.NamespaceBuilder; -@Slf4j +@CustomLog abstract class PerfReaderBase implements Runnable { /** @@ -107,7 +107,9 @@ public static class Flags extends CliFlags { protected void execute() throws Exception { ObjectMapper m = new ObjectMapper(); ObjectWriter w = m.writerWithDefaultPrettyPrinter(); - log.info("Starting dlog perf reader with config : {}", w.writeValueAsString(flags)); + log.info() + .attr("config", w.writeValueAsString(flags)) + .log("Starting dlog perf reader"); DistributedLogConfiguration conf = newDlogConf(flags); try (Namespace namespace = NamespaceBuilder.newBuilder() @@ -143,16 +145,22 @@ protected void reportStats() { reportHistogram = recorder.getIntervalHistogram(reportHistogram); - log.info("Throughput read : {} records/s --- {} MB/s --- Latency: mean:" - + " {} ms - med: {} - 95pct: {} - 99pct: {} - 99.9pct: {} - 99.99pct: {} - Max: {}", - THROUGHPUT_FORMAT.format(rate), THROUGHPUT_FORMAT.format(throughput), - PADDING_DECIMAL_FORMAT.format(reportHistogram.getMean() / 1000.0), - PADDING_DECIMAL_FORMAT.format(reportHistogram.getValueAtPercentile(50) / 1000.0), - PADDING_DECIMAL_FORMAT.format(reportHistogram.getValueAtPercentile(95) / 1000.0), - PADDING_DECIMAL_FORMAT.format(reportHistogram.getValueAtPercentile(99) / 1000.0), - PADDING_DECIMAL_FORMAT.format(reportHistogram.getValueAtPercentile(99.9) / 1000.0), - PADDING_DECIMAL_FORMAT.format(reportHistogram.getValueAtPercentile(99.99) / 1000.0), - PADDING_DECIMAL_FORMAT.format(reportHistogram.getMaxValue() / 1000.0)); + log.info() + .attr("rate", THROUGHPUT_FORMAT.format(rate)) + .attr("throughputMBs", THROUGHPUT_FORMAT.format(throughput)) + .attr("latencyMeanMs", PADDING_DECIMAL_FORMAT.format(reportHistogram.getMean() / 1000.0)) + .attr("latencyMedMs", + PADDING_DECIMAL_FORMAT.format(reportHistogram.getValueAtPercentile(50) / 1000.0)) + .attr("latency95pctMs", + PADDING_DECIMAL_FORMAT.format(reportHistogram.getValueAtPercentile(95) / 1000.0)) + .attr("latency99pctMs", + PADDING_DECIMAL_FORMAT.format(reportHistogram.getValueAtPercentile(99) / 1000.0)) + .attr("latency999pctMs", + PADDING_DECIMAL_FORMAT.format(reportHistogram.getValueAtPercentile(99.9) / 1000.0)) + .attr("latency9999pctMs", + PADDING_DECIMAL_FORMAT.format(reportHistogram.getValueAtPercentile(99.99) / 1000.0)) + .attr("latencyMaxMs", PADDING_DECIMAL_FORMAT.format(reportHistogram.getMaxValue() / 1000.0)) + .log("Throughput read"); reportHistogram.reset(); @@ -168,7 +176,7 @@ public void run() { try { execute(); } catch (Exception e) { - log.error("Encountered exception at running dlog perf writer", e); + log.error().exception(e).log("Encountered exception at running dlog perf writer"); } } @@ -187,16 +195,22 @@ private static DistributedLogConfiguration newDlogConf(Flags flags) { protected static void printAggregatedStats(Recorder recorder) { Histogram reportHistogram = recorder.getIntervalHistogram(); - log.info("Aggregated latency stats --- Latency: mean: {} ms - med: {} - 95pct: {} - 99pct: {}" - + " - 99.9pct: {} - 99.99pct: {} - 99.999pct: {} - Max: {}", - PADDING_DECIMAL_FORMAT.format(reportHistogram.getMean() / 1000.0), - PADDING_DECIMAL_FORMAT.format(reportHistogram.getValueAtPercentile(50) / 1000.0), - PADDING_DECIMAL_FORMAT.format(reportHistogram.getValueAtPercentile(95) / 1000.0), - PADDING_DECIMAL_FORMAT.format(reportHistogram.getValueAtPercentile(99) / 1000.0), - PADDING_DECIMAL_FORMAT.format(reportHistogram.getValueAtPercentile(99.9) / 1000.0), - PADDING_DECIMAL_FORMAT.format(reportHistogram.getValueAtPercentile(99.99) / 1000.0), - PADDING_DECIMAL_FORMAT.format(reportHistogram.getValueAtPercentile(99.999) / 1000.0), - PADDING_DECIMAL_FORMAT.format(reportHistogram.getMaxValue() / 1000.0)); + log.info() + .attr("latencyMeanMs", PADDING_DECIMAL_FORMAT.format(reportHistogram.getMean() / 1000.0)) + .attr("latencyMedMs", + PADDING_DECIMAL_FORMAT.format(reportHistogram.getValueAtPercentile(50) / 1000.0)) + .attr("latency95pctMs", + PADDING_DECIMAL_FORMAT.format(reportHistogram.getValueAtPercentile(95) / 1000.0)) + .attr("latency99pctMs", + PADDING_DECIMAL_FORMAT.format(reportHistogram.getValueAtPercentile(99) / 1000.0)) + .attr("latency999pctMs", + PADDING_DECIMAL_FORMAT.format(reportHistogram.getValueAtPercentile(99.9) / 1000.0)) + .attr("latency9999pctMs", + PADDING_DECIMAL_FORMAT.format(reportHistogram.getValueAtPercentile(99.99) / 1000.0)) + .attr("latency99999pctMs", + PADDING_DECIMAL_FORMAT.format(reportHistogram.getValueAtPercentile(99.999) / 1000.0)) + .attr("latencyMaxMs", PADDING_DECIMAL_FORMAT.format(reportHistogram.getMaxValue() / 1000.0)) + .log("Aggregated latency stats"); } } diff --git a/tools/perf/src/main/java/org/apache/bookkeeper/tools/perf/dlog/PerfSegmentReader.java b/tools/perf/src/main/java/org/apache/bookkeeper/tools/perf/dlog/PerfSegmentReader.java index 6b8ecc8ca05..65a2f3e3908 100644 --- a/tools/perf/src/main/java/org/apache/bookkeeper/tools/perf/dlog/PerfSegmentReader.java +++ b/tools/perf/src/main/java/org/apache/bookkeeper/tools/perf/dlog/PerfSegmentReader.java @@ -25,8 +25,8 @@ import java.util.concurrent.Executors; import java.util.concurrent.TimeUnit; import java.util.stream.Collectors; +import lombok.CustomLog; import lombok.Data; -import lombok.extern.slf4j.Slf4j; import org.apache.bookkeeper.common.net.ServiceURI; import org.apache.commons.lang3.mutable.MutableBoolean; import org.apache.commons.lang3.tuple.Pair; @@ -42,7 +42,7 @@ /** * A perf writer to evaluate write performance. */ -@Slf4j +@CustomLog public class PerfSegmentReader extends PerfReaderBase { @Data @@ -64,7 +64,7 @@ protected void execute(Namespace namespace) throws Exception { String logName = String.format(flags.logName, i); managers.add(namespace.openLog(logName)); } - log.info("Successfully open {} logs", managers.size()); + log.info().attr("numLogs", managers.size()).log("Successfully opened logs"); // Get all the log segments final List> segments = managers.stream() @@ -99,11 +99,11 @@ protected void execute(Namespace namespace) throws Exception { try { read(splitsThisThread); } catch (Exception e) { - log.error("Encountered error at writing records", e); + log.error().exception(e).log("Encountered error at writing records"); } }); } - log.info("Started {} write threads", flags.numThreads); + log.info().attr("numThreads", flags.numThreads).log("Started write threads"); reportStats(); } finally { executor.shutdown(); @@ -115,11 +115,13 @@ protected void execute(Namespace namespace) throws Exception { } void read(List splits) throws Exception { - log.info("Read thread started with : splits = {}", - splits.stream() - .map(l -> "(log = " + l.manager.getStreamName() + ", segment = " - + l.segment.getLogSegmentSequenceNumber() + " [" + l.startEntryId + ", " + l.endEntryId + "])") - .collect(Collectors.toList())); + log.info() + .attr("splits", splits.stream() + .map(l -> "(log = " + l.manager.getStreamName() + ", segment = " + + l.segment.getLogSegmentSequenceNumber() + + " [" + l.startEntryId + ", " + l.endEntryId + "])") + .collect(Collectors.toList())) + .log("Read thread started"); splits.forEach(entry -> { try { diff --git a/tools/perf/src/main/java/org/apache/bookkeeper/tools/perf/dlog/PerfWriter.java b/tools/perf/src/main/java/org/apache/bookkeeper/tools/perf/dlog/PerfWriter.java index 946616f30b6..7157aeb6652 100644 --- a/tools/perf/src/main/java/org/apache/bookkeeper/tools/perf/dlog/PerfWriter.java +++ b/tools/perf/src/main/java/org/apache/bookkeeper/tools/perf/dlog/PerfWriter.java @@ -34,7 +34,7 @@ import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.LongAdder; import java.util.stream.Collectors; -import lombok.extern.slf4j.Slf4j; +import lombok.CustomLog; import org.HdrHistogram.Histogram; import org.HdrHistogram.Recorder; import org.apache.bookkeeper.common.concurrent.FutureUtils; @@ -52,7 +52,7 @@ /** * A perf writer to evaluate write performance. */ -@Slf4j +@CustomLog public class PerfWriter implements Runnable { /** @@ -174,14 +174,16 @@ public void run() { try { execute(); } catch (Exception e) { - log.error("Encountered exception at running dlog perf writer", e); + log.error().exception(e).log("Encountered exception at running dlog perf writer"); } } void execute() throws Exception { ObjectMapper m = new ObjectMapper(); ObjectWriter w = m.writerWithDefaultPrettyPrinter(); - log.info("Starting dlog perf writer with config : {}", w.writeValueAsString(flags)); + log.info() + .attr("config", w.writeValueAsString(flags)) + .log("Starting dlog perf writer"); DistributedLogConfiguration conf = newDlogConf(flags); try (Namespace namespace = NamespaceBuilder.newBuilder() @@ -198,7 +200,7 @@ void execute(Namespace namespace) throws Exception { String logName = String.format(flags.logName, i); managers.add(Pair.of(i, namespace.openLog(logName))); } - log.info("Successfully open {} logs", managers.size()); + log.info().attr("numLogs", managers.size()).log("Successfully opened logs"); // register shutdown hook to aggregate stats Runtime.getRuntime().addShutdownHook(new Thread(() -> { @@ -228,11 +230,11 @@ void execute(Namespace namespace) throws Exception { numRecordsForThisThread, numBytesForThisThread); } catch (Exception e) { - log.error("Encountered error at writing records", e); + log.error().exception(e).log("Encountered error at writing records"); } }); } - log.info("Started {} write threads", flags.numThreads); + log.info().attr("numThreads", flags.numThreads).log("Started write threads"); reportStats(); } finally { executor.shutdown(); @@ -248,13 +250,13 @@ void write(List logs, int maxOutstandingBytesForThisThread, long numRecordsForThisThread, long numBytesForThisThread) throws Exception { - log.info("Write thread started with : logs = {}, rate = {}," - + " num records = {}, num bytes = {}, max outstanding bytes = {}", - logs.stream().map(l -> l.getStreamName()).collect(Collectors.toList()), - writeRate, - numRecordsForThisThread, - numBytesForThisThread, - maxOutstandingBytesForThisThread); + log.info() + .attr("logs", logs.stream().map(l -> l.getStreamName()).collect(Collectors.toList())) + .attr("rate", writeRate) + .attr("numRecords", numRecordsForThisThread) + .attr("numBytes", numBytesForThisThread) + .attr("maxOutstandingBytes", maxOutstandingBytesForThisThread) + .log("Write thread started"); List> writerFutures = logs.stream() .map(manager -> manager.openAsyncLogWriter()) @@ -325,7 +327,7 @@ void write(List logs, recorder.recordValue(latencyMicros); cumulativeRecorder.recordValue(latencyMicros); }).exceptionally(cause -> { - log.warn("Error at writing records", cause); + log.warn().exception(cause).log("Error at writing records"); System.exit(-1); return null; }); @@ -367,17 +369,17 @@ void reportStats() { reportHistogram = recorder.getIntervalHistogram(reportHistogram); - log.info( - "Throughput written : {} records/s --- {} MB/s --- Latency: mean:" - + " {} ms - med: {} - 95pct: {} - 99pct: {} - 99.9pct: {} - 99.99pct: {} - Max: {}", - throughputFormat.format(rate), throughputFormat.format(throughput), - dec.format(reportHistogram.getMean() / 1000.0), - dec.format(reportHistogram.getValueAtPercentile(50) / 1000.0), - dec.format(reportHistogram.getValueAtPercentile(95) / 1000.0), - dec.format(reportHistogram.getValueAtPercentile(99) / 1000.0), - dec.format(reportHistogram.getValueAtPercentile(99.9) / 1000.0), - dec.format(reportHistogram.getValueAtPercentile(99.99) / 1000.0), - dec.format(reportHistogram.getMaxValue() / 1000.0)); + log.info() + .attr("rate", throughputFormat.format(rate)) + .attr("throughputMBs", throughputFormat.format(throughput)) + .attr("latencyMeanMs", dec.format(reportHistogram.getMean() / 1000.0)) + .attr("latencyMedMs", dec.format(reportHistogram.getValueAtPercentile(50) / 1000.0)) + .attr("latency95pctMs", dec.format(reportHistogram.getValueAtPercentile(95) / 1000.0)) + .attr("latency99pctMs", dec.format(reportHistogram.getValueAtPercentile(99) / 1000.0)) + .attr("latency999pctMs", dec.format(reportHistogram.getValueAtPercentile(99.9) / 1000.0)) + .attr("latency9999pctMs", dec.format(reportHistogram.getValueAtPercentile(99.99) / 1000.0)) + .attr("latencyMaxMs", dec.format(reportHistogram.getMaxValue() / 1000.0)) + .log("Throughput written"); reportHistogram.reset(); @@ -406,16 +408,16 @@ private static DistributedLogConfiguration newDlogConf(Flags flags) { private static void printAggregatedStats(Recorder recorder) { Histogram reportHistogram = recorder.getIntervalHistogram(); - log.info("Aggregated latency stats --- Latency: mean: {} ms - med: {} - 95pct: {} - 99pct: {}" - + " - 99.9pct: {} - 99.99pct: {} - 99.999pct: {} - Max: {}", - dec.format(reportHistogram.getMean() / 1000.0), - dec.format(reportHistogram.getValueAtPercentile(50) / 1000.0), - dec.format(reportHistogram.getValueAtPercentile(95) / 1000.0), - dec.format(reportHistogram.getValueAtPercentile(99) / 1000.0), - dec.format(reportHistogram.getValueAtPercentile(99.9) / 1000.0), - dec.format(reportHistogram.getValueAtPercentile(99.99) / 1000.0), - dec.format(reportHistogram.getValueAtPercentile(99.999) / 1000.0), - dec.format(reportHistogram.getMaxValue() / 1000.0)); + log.info() + .attr("latencyMeanMs", dec.format(reportHistogram.getMean() / 1000.0)) + .attr("latencyMedMs", dec.format(reportHistogram.getValueAtPercentile(50) / 1000.0)) + .attr("latency95pctMs", dec.format(reportHistogram.getValueAtPercentile(95) / 1000.0)) + .attr("latency99pctMs", dec.format(reportHistogram.getValueAtPercentile(99) / 1000.0)) + .attr("latency999pctMs", dec.format(reportHistogram.getValueAtPercentile(99.9) / 1000.0)) + .attr("latency9999pctMs", dec.format(reportHistogram.getValueAtPercentile(99.99) / 1000.0)) + .attr("latency99999pctMs", dec.format(reportHistogram.getValueAtPercentile(99.999) / 1000.0)) + .attr("latencyMaxMs", dec.format(reportHistogram.getMaxValue() / 1000.0)) + .log("Aggregated latency stats"); } } diff --git a/tools/perf/src/main/java/org/apache/bookkeeper/tools/perf/dlog/ReadCommand.java b/tools/perf/src/main/java/org/apache/bookkeeper/tools/perf/dlog/ReadCommand.java index c21b4213c6a..4446a87d210 100644 --- a/tools/perf/src/main/java/org/apache/bookkeeper/tools/perf/dlog/ReadCommand.java +++ b/tools/perf/src/main/java/org/apache/bookkeeper/tools/perf/dlog/ReadCommand.java @@ -14,7 +14,7 @@ package org.apache.bookkeeper.tools.perf.dlog; -import lombok.extern.slf4j.Slf4j; +import lombok.CustomLog; import org.apache.bookkeeper.common.net.ServiceURI; import org.apache.bookkeeper.tools.common.BKCommand; import org.apache.bookkeeper.tools.common.BKFlags; @@ -25,7 +25,7 @@ /** * Command to read log records to distributedlog streams. */ -@Slf4j +@CustomLog public class ReadCommand extends BKCommand { private static final String NAME = "read"; diff --git a/tools/perf/src/main/java/org/apache/bookkeeper/tools/perf/dlog/SegmentReadCommand.java b/tools/perf/src/main/java/org/apache/bookkeeper/tools/perf/dlog/SegmentReadCommand.java index 076938638c9..3a684691372 100644 --- a/tools/perf/src/main/java/org/apache/bookkeeper/tools/perf/dlog/SegmentReadCommand.java +++ b/tools/perf/src/main/java/org/apache/bookkeeper/tools/perf/dlog/SegmentReadCommand.java @@ -14,7 +14,7 @@ package org.apache.bookkeeper.tools.perf.dlog; -import lombok.extern.slf4j.Slf4j; +import lombok.CustomLog; import org.apache.bookkeeper.common.net.ServiceURI; import org.apache.bookkeeper.tools.common.BKCommand; import org.apache.bookkeeper.tools.common.BKFlags; @@ -25,7 +25,7 @@ /** * Command to read log records to bookkeeper segments. */ -@Slf4j +@CustomLog public class SegmentReadCommand extends BKCommand { private static final String NAME = "segread"; diff --git a/tools/perf/src/main/java/org/apache/bookkeeper/tools/perf/dlog/WriteCommand.java b/tools/perf/src/main/java/org/apache/bookkeeper/tools/perf/dlog/WriteCommand.java index 4da1742655c..eef64a120b8 100644 --- a/tools/perf/src/main/java/org/apache/bookkeeper/tools/perf/dlog/WriteCommand.java +++ b/tools/perf/src/main/java/org/apache/bookkeeper/tools/perf/dlog/WriteCommand.java @@ -14,7 +14,7 @@ package org.apache.bookkeeper.tools.perf.dlog; -import lombok.extern.slf4j.Slf4j; +import lombok.CustomLog; import org.apache.bookkeeper.common.net.ServiceURI; import org.apache.bookkeeper.tools.common.BKCommand; import org.apache.bookkeeper.tools.common.BKFlags; @@ -25,7 +25,7 @@ /** * Command to write log records to distributedlog streams. */ -@Slf4j +@CustomLog public class WriteCommand extends BKCommand { private static final String NAME = "write"; diff --git a/tools/perf/src/main/java/org/apache/bookkeeper/tools/perf/journal/AppendCommand.java b/tools/perf/src/main/java/org/apache/bookkeeper/tools/perf/journal/AppendCommand.java index ec02a15e924..dad3a0a9940 100644 --- a/tools/perf/src/main/java/org/apache/bookkeeper/tools/perf/journal/AppendCommand.java +++ b/tools/perf/src/main/java/org/apache/bookkeeper/tools/perf/journal/AppendCommand.java @@ -14,7 +14,7 @@ package org.apache.bookkeeper.tools.perf.journal; -import lombok.extern.slf4j.Slf4j; +import lombok.CustomLog; import org.apache.bookkeeper.common.net.ServiceURI; import org.apache.bookkeeper.tools.common.BKCommand; import org.apache.bookkeeper.tools.common.BKFlags; @@ -25,7 +25,7 @@ /** * Command to write log records to distributedlog streams. */ -@Slf4j +@CustomLog public class AppendCommand extends BKCommand { private static final String NAME = "append"; diff --git a/tools/perf/src/main/java/org/apache/bookkeeper/tools/perf/journal/JournalWriter.java b/tools/perf/src/main/java/org/apache/bookkeeper/tools/perf/journal/JournalWriter.java index a152dac6a0b..1d0655a6b33 100644 --- a/tools/perf/src/main/java/org/apache/bookkeeper/tools/perf/journal/JournalWriter.java +++ b/tools/perf/src/main/java/org/apache/bookkeeper/tools/perf/journal/JournalWriter.java @@ -40,7 +40,7 @@ import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.LongAdder; -import lombok.extern.slf4j.Slf4j; +import lombok.CustomLog; import org.HdrHistogram.Histogram; import org.HdrHistogram.Recorder; import org.apache.bookkeeper.bookie.BookieException; @@ -62,7 +62,7 @@ /** * A perf writer to evaluate write performance. */ -@Slf4j +@CustomLog public class JournalWriter implements Runnable { /** @@ -205,21 +205,24 @@ public void run() { try { execute(); } catch (Exception e) { - log.error("Encountered exception at running dlog perf writer", e); + log.error().exception(e).log("Encountered exception at running dlog perf writer"); } } void execute() throws Exception { ObjectMapper m = new ObjectMapper(); ObjectWriter w = m.writerWithDefaultPrettyPrinter(); - log.info("Starting journal perf writer with config : {}", w.writeValueAsString(flags)); + log.info() + .attr("config", w.writeValueAsString(flags)) + .log("Starting journal perf writer"); checkArgument(flags.journalDirs.size() > 0, "No journal dirs is provided"); updateServerConf(conf, flags); - log.info("Benchmark the journal perf with server config : {}", - conf.asJson()); + log.info() + .attr("serverConfig", conf.asJson()) + .log("Benchmark the journal perf"); Stats.loadStatsProvider(conf); Stats.get().start(conf); @@ -276,7 +279,10 @@ void execute(Journal[] journals) throws Exception { try { journal.checkpointComplete(cp, true); } catch (IOException e) { - log.error("Failed to complete checkpoint {}", cp, e); + log.error() + .attr("checkpoint", cp) + .exception(e) + .log("Failed to complete checkpoint"); } } }, 30L, 30L, TimeUnit.SECONDS); @@ -302,11 +308,13 @@ void execute(Journal[] journals) throws Exception { numRecordsForThisThread, numBytesForThisThread); } catch (Throwable t) { - log.error("Encountered error at writing records", t); + log.error().exception(t).log("Encountered error at writing records"); } }); } - log.info("Started {} write threads", flags.numTestThreads); + log.info() + .attr("numTestThreads", flags.numTestThreads) + .log("Started write threads"); reportStats(); } finally { flushExecutor.shutdown(); @@ -324,13 +332,13 @@ void write(int threadIdx, int maxOutstandingBytesForThisThread, long numRecordsForThisThread, long numBytesForThisThread) throws Exception { - log.info("Write thread {} started with : rate = {}," - + " num records = {}, num bytes = {}, max outstanding bytes = {}", - threadIdx, - writeRate, - numRecordsForThisThread, - numBytesForThisThread, - maxOutstandingBytesForThisThread); + log.info() + .attr("threadIdx", threadIdx) + .attr("rate", writeRate) + .attr("numRecords", numRecordsForThisThread) + .attr("numBytes", numBytesForThisThread) + .attr("maxOutstandingBytes", maxOutstandingBytesForThisThread) + .log("Write thread started"); RateLimiter limiter; if (writeRate > 0) { @@ -405,7 +413,9 @@ void write(int threadIdx, recorder.recordValue(latencyMicros); cumulativeRecorder.recordValue(latencyMicros); } else { - log.warn("Error at writing records : ", BookieException.create(rc)); + log.warn() + .exception(BookieException.create(rc)) + .log("Error at writing records"); Runtime.getRuntime().exit(-1); } }, @@ -449,17 +459,17 @@ void reportStats() { reportHistogram = recorder.getIntervalHistogram(reportHistogram); - log.info( - "Throughput written : {} records/s --- {} MB/s --- Latency: mean:" - + " {} ms - med: {} - 95pct: {} - 99pct: {} - 99.9pct: {} - 99.99pct: {} - Max: {}", - throughputFormat.format(rate), throughputFormat.format(throughput), - dec.format(reportHistogram.getMean() / 1000.0), - dec.format(reportHistogram.getValueAtPercentile(50) / 1000.0), - dec.format(reportHistogram.getValueAtPercentile(95) / 1000.0), - dec.format(reportHistogram.getValueAtPercentile(99) / 1000.0), - dec.format(reportHistogram.getValueAtPercentile(99.9) / 1000.0), - dec.format(reportHistogram.getValueAtPercentile(99.99) / 1000.0), - dec.format(reportHistogram.getMaxValue() / 1000.0)); + log.info() + .attr("rate", throughputFormat.format(rate)) + .attr("throughputMBs", throughputFormat.format(throughput)) + .attr("latencyMeanMs", dec.format(reportHistogram.getMean() / 1000.0)) + .attr("latencyMedMs", dec.format(reportHistogram.getValueAtPercentile(50) / 1000.0)) + .attr("latency95pctMs", dec.format(reportHistogram.getValueAtPercentile(95) / 1000.0)) + .attr("latency99pctMs", dec.format(reportHistogram.getValueAtPercentile(99) / 1000.0)) + .attr("latency999pctMs", dec.format(reportHistogram.getValueAtPercentile(99.9) / 1000.0)) + .attr("latency9999pctMs", dec.format(reportHistogram.getValueAtPercentile(99.99) / 1000.0)) + .attr("latencyMaxMs", dec.format(reportHistogram.getMaxValue() / 1000.0)) + .log("Throughput written"); reportHistogram.reset(); @@ -481,7 +491,7 @@ private static void updateServerConf(ServerConfiguration conf, Flags flags) { File[] currentDirs = BookieImpl.getCurrentDirectories(conf.getLedgerDirs()); for (File dir : currentDirs) { if (dir.mkdirs()) { - log.info("Successfully created dir {}", dir); + log.info().attr("directory", dir).log("Successfully created dir"); } } } @@ -492,7 +502,7 @@ private static ByteBufAllocator getAllocator(ServerConfiguration conf) { .poolingConcurrency(conf.getAllocatorPoolingConcurrency()) .outOfMemoryPolicy(conf.getAllocatorOutOfMemoryPolicy()) .outOfMemoryListener((ex) -> { - log.error("Unable to allocate memory, exiting bookie", ex); + log.error().exception(ex).log("Unable to allocate memory, exiting bookie"); }) .leakDetectionPolicy(conf.getAllocatorLeakDetectionPolicy()) .exitOnOutOfMemory(conf.exitOnOutOfMemory()) @@ -505,16 +515,16 @@ private static ByteBufAllocator getAllocator(ServerConfiguration conf) { private static void printAggregatedStats(Recorder recorder) { Histogram reportHistogram = recorder.getIntervalHistogram(); - log.info("Aggregated latency stats --- Latency: mean: {} ms - med: {} - 95pct: {} - 99pct: {}" - + " - 99.9pct: {} - 99.99pct: {} - 99.999pct: {} - Max: {}", - dec.format(reportHistogram.getMean() / 1000.0), - dec.format(reportHistogram.getValueAtPercentile(50) / 1000.0), - dec.format(reportHistogram.getValueAtPercentile(95) / 1000.0), - dec.format(reportHistogram.getValueAtPercentile(99) / 1000.0), - dec.format(reportHistogram.getValueAtPercentile(99.9) / 1000.0), - dec.format(reportHistogram.getValueAtPercentile(99.99) / 1000.0), - dec.format(reportHistogram.getValueAtPercentile(99.999) / 1000.0), - dec.format(reportHistogram.getMaxValue() / 1000.0)); + log.info() + .attr("latencyMeanMs", dec.format(reportHistogram.getMean() / 1000.0)) + .attr("latencyMedMs", dec.format(reportHistogram.getValueAtPercentile(50) / 1000.0)) + .attr("latency95pctMs", dec.format(reportHistogram.getValueAtPercentile(95) / 1000.0)) + .attr("latency99pctMs", dec.format(reportHistogram.getValueAtPercentile(99) / 1000.0)) + .attr("latency999pctMs", dec.format(reportHistogram.getValueAtPercentile(99.9) / 1000.0)) + .attr("latency9999pctMs", dec.format(reportHistogram.getValueAtPercentile(99.99) / 1000.0)) + .attr("latency99999pctMs", dec.format(reportHistogram.getValueAtPercentile(99.999) / 1000.0)) + .attr("latencyMaxMs", dec.format(reportHistogram.getMaxValue() / 1000.0)) + .log("Aggregated latency stats"); } } diff --git a/tools/perf/src/main/java/org/apache/bookkeeper/tools/perf/table/BenchmarkCommand.java b/tools/perf/src/main/java/org/apache/bookkeeper/tools/perf/table/BenchmarkCommand.java index e2b547e14b7..1cdec2b92d6 100644 --- a/tools/perf/src/main/java/org/apache/bookkeeper/tools/perf/table/BenchmarkCommand.java +++ b/tools/perf/src/main/java/org/apache/bookkeeper/tools/perf/table/BenchmarkCommand.java @@ -19,7 +19,7 @@ package org.apache.bookkeeper.tools.perf.table; -import lombok.extern.slf4j.Slf4j; +import lombok.CustomLog; import org.apache.bookkeeper.common.net.ServiceURI; import org.apache.bookkeeper.tools.common.BKCommand; import org.apache.bookkeeper.tools.common.BKFlags; @@ -30,7 +30,7 @@ /** * Command to benchmark table service. */ -@Slf4j +@CustomLog public class BenchmarkCommand extends BKCommand { private static final String NAME = "benchmark"; diff --git a/tools/perf/src/main/java/org/apache/bookkeeper/tools/perf/table/IncrementTask.java b/tools/perf/src/main/java/org/apache/bookkeeper/tools/perf/table/IncrementTask.java index 121eadf27b4..d77355360d6 100644 --- a/tools/perf/src/main/java/org/apache/bookkeeper/tools/perf/table/IncrementTask.java +++ b/tools/perf/src/main/java/org/apache/bookkeeper/tools/perf/table/IncrementTask.java @@ -25,7 +25,7 @@ import io.netty.util.ReferenceCountUtil; import java.util.concurrent.Semaphore; import java.util.concurrent.TimeUnit; -import lombok.extern.slf4j.Slf4j; +import lombok.CustomLog; import org.apache.bookkeeper.api.kv.Table; import org.apache.bookkeeper.tools.perf.table.PerfClient.Flags; import org.apache.bookkeeper.tools.perf.table.PerfClient.OP; @@ -34,7 +34,7 @@ /** * Write task to inject key/value pairs to the table. */ -@Slf4j +@CustomLog abstract class IncrementTask extends BenchmarkTask { protected final RateLimiter limiter; @@ -83,7 +83,7 @@ void incKey(long i) { semaphore.release(); } if (null != cause) { - log.error("Error at increment key/amount", cause); + log.error().exception(cause).log("Error at increment key/amount"); } else { long latencyMicros = TimeUnit.NANOSECONDS.toMicros( System.nanoTime() - startTime diff --git a/tools/perf/src/main/java/org/apache/bookkeeper/tools/perf/table/PerfClient.java b/tools/perf/src/main/java/org/apache/bookkeeper/tools/perf/table/PerfClient.java index 20b9422f213..c12b6895cb9 100644 --- a/tools/perf/src/main/java/org/apache/bookkeeper/tools/perf/table/PerfClient.java +++ b/tools/perf/src/main/java/org/apache/bookkeeper/tools/perf/table/PerfClient.java @@ -37,7 +37,7 @@ import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.LongAdder; import lombok.Cleanup; -import lombok.extern.slf4j.Slf4j; +import lombok.CustomLog; import org.HdrHistogram.Histogram; import org.HdrHistogram.Recorder; import org.apache.bookkeeper.api.StorageClient; @@ -51,7 +51,7 @@ /** * Perf client to evaluate the performance of table service. */ -@Slf4j +@CustomLog public class PerfClient implements Runnable { enum OP { @@ -176,34 +176,33 @@ void reportStats(long oldTime) { double elapsed = (now - oldTime) / 1e9; double rate = ops.sumThenReset() / elapsed; reportHistogram = recorder.getIntervalHistogram(reportHistogram); - log.info( - "[{}] Throughput: {} ops/s --- Latency: mean:" - + " {} ms - med: {} - 95pct: {} - 99pct: {} - 99.9pct: {} - 99.99pct: {} - Max: {}", - name, - throughputFormat.format(rate), - dec.format(reportHistogram.getMean() / 1000.0), - dec.format(reportHistogram.getValueAtPercentile(50) / 1000.0), - dec.format(reportHistogram.getValueAtPercentile(95) / 1000.0), - dec.format(reportHistogram.getValueAtPercentile(99) / 1000.0), - dec.format(reportHistogram.getValueAtPercentile(99.9) / 1000.0), - dec.format(reportHistogram.getValueAtPercentile(99.99) / 1000.0), - dec.format(reportHistogram.getMaxValue() / 1000.0)); + log.info() + .attr("op", name) + .attr("throughput", throughputFormat.format(rate)) + .attr("latencyMeanMs", dec.format(reportHistogram.getMean() / 1000.0)) + .attr("latencyMedMs", dec.format(reportHistogram.getValueAtPercentile(50) / 1000.0)) + .attr("latency95pctMs", dec.format(reportHistogram.getValueAtPercentile(95) / 1000.0)) + .attr("latency99pctMs", dec.format(reportHistogram.getValueAtPercentile(99) / 1000.0)) + .attr("latency999pctMs", dec.format(reportHistogram.getValueAtPercentile(99.9) / 1000.0)) + .attr("latency9999pctMs", dec.format(reportHistogram.getValueAtPercentile(99.99) / 1000.0)) + .attr("latencyMaxMs", dec.format(reportHistogram.getMaxValue() / 1000.0)) + .log("Op throughput"); reportHistogram.reset(); } void printAggregatedStats() { Histogram reportHistogram = cumulativeRecorder.getIntervalHistogram(); - log.info("[{}] latency stats --- Latency: mean: {} ms - med: {} - 95pct: {} - 99pct: {}" - + " - 99.9pct: {} - 99.99pct: {} - 99.999pct: {} - Max: {}", - name, - dec.format(reportHistogram.getMean() / 1000.0), - dec.format(reportHistogram.getValueAtPercentile(50) / 1000.0), - dec.format(reportHistogram.getValueAtPercentile(95) / 1000.0), - dec.format(reportHistogram.getValueAtPercentile(99) / 1000.0), - dec.format(reportHistogram.getValueAtPercentile(99.9) / 1000.0), - dec.format(reportHistogram.getValueAtPercentile(99.99) / 1000.0), - dec.format(reportHistogram.getValueAtPercentile(99.999) / 1000.0), - dec.format(reportHistogram.getMaxValue() / 1000.0)); + log.info() + .attr("op", name) + .attr("latencyMeanMs", dec.format(reportHistogram.getMean() / 1000.0)) + .attr("latencyMedMs", dec.format(reportHistogram.getValueAtPercentile(50) / 1000.0)) + .attr("latency95pctMs", dec.format(reportHistogram.getValueAtPercentile(95) / 1000.0)) + .attr("latency99pctMs", dec.format(reportHistogram.getValueAtPercentile(99) / 1000.0)) + .attr("latency999pctMs", dec.format(reportHistogram.getValueAtPercentile(99.9) / 1000.0)) + .attr("latency9999pctMs", dec.format(reportHistogram.getValueAtPercentile(99.99) / 1000.0)) + .attr("latency99999pctMs", dec.format(reportHistogram.getValueAtPercentile(99.999) / 1000.0)) + .attr("latencyMaxMs", dec.format(reportHistogram.getMaxValue() / 1000.0)) + .log("Aggregated op latency stats"); } } @@ -220,14 +219,16 @@ public void run() { try { execute(); } catch (Exception e) { - log.error("Encountered exception at running table perf client", e); + log.error().exception(e).log("Encountered exception at running table perf client"); } } void execute() throws Exception { ObjectMapper m = new ObjectMapper(); ObjectWriter w = m.writerWithDefaultPrettyPrinter(); - log.info("Starting table perf client with config : {}", w.writeValueAsString(flags)); + log.info() + .attr("config", w.writeValueAsString(flags)) + .log("Starting table perf client"); runBenchmarkTasks(); } @@ -332,8 +333,10 @@ private void runBenchmarkTasks() throws Exception { try { task.runTask(); } catch (Exception e) { - log.error("Encountered issue at running benchmark task {}", - task.tid, e); + log.error() + .attr("taskId", task.tid) + .exception(e) + .log("Encountered issue at running benchmark task"); } finally { latch.countDown(); } diff --git a/tools/perf/src/main/java/org/apache/bookkeeper/tools/perf/table/WriteTask.java b/tools/perf/src/main/java/org/apache/bookkeeper/tools/perf/table/WriteTask.java index 82b03d115d2..87d390a40da 100644 --- a/tools/perf/src/main/java/org/apache/bookkeeper/tools/perf/table/WriteTask.java +++ b/tools/perf/src/main/java/org/apache/bookkeeper/tools/perf/table/WriteTask.java @@ -27,7 +27,7 @@ import java.util.concurrent.Semaphore; import java.util.concurrent.ThreadLocalRandom; import java.util.concurrent.TimeUnit; -import lombok.extern.slf4j.Slf4j; +import lombok.CustomLog; import org.apache.bookkeeper.api.kv.Table; import org.apache.bookkeeper.tools.perf.table.PerfClient.Flags; import org.apache.bookkeeper.tools.perf.table.PerfClient.OP; @@ -36,7 +36,7 @@ /** * Write task to inject key/value pairs to the table. */ -@Slf4j +@CustomLog abstract class WriteTask extends BenchmarkTask { protected final RateLimiter limiter; @@ -89,7 +89,7 @@ void writeKey(long i, byte[] valueBytes) { semaphore.release(); } if (null != cause) { - log.error("Error at put key/value", cause); + log.error().exception(cause).log("Error at put key/value"); } else { long latencyMicros = TimeUnit.NANOSECONDS.toMicros( System.nanoTime() - startTime diff --git a/tools/stream/src/main/java/org/apache/bookkeeper/stream/cli/commands/AbstractStreamCommand.java b/tools/stream/src/main/java/org/apache/bookkeeper/stream/cli/commands/AbstractStreamCommand.java index e168a00d8c9..8737cc14cc0 100644 --- a/tools/stream/src/main/java/org/apache/bookkeeper/stream/cli/commands/AbstractStreamCommand.java +++ b/tools/stream/src/main/java/org/apache/bookkeeper/stream/cli/commands/AbstractStreamCommand.java @@ -19,7 +19,7 @@ package org.apache.bookkeeper.stream.cli.commands; -import lombok.extern.slf4j.Slf4j; +import lombok.CustomLog; import org.apache.bookkeeper.common.net.ServiceURI; import org.apache.bookkeeper.tools.common.BKCommand; import org.apache.bookkeeper.tools.common.BKFlags; @@ -31,7 +31,7 @@ /** * Abstract stream storage related commands. */ -@Slf4j +@CustomLog abstract class AbstractStreamCommand extends BKCommand { protected AbstractStreamCommand(CliSpec spec) { @@ -50,7 +50,9 @@ protected boolean apply(ServiceURI serviceURI, CommandFlagsT cmdFlags) { if (serviceURI == null) { serviceURI = ServiceURI.DEFAULT_LOCAL_STREAM_STORAGE_SERVICE_URI; - log.info("Service Uri is not specified. Using default service uri : {}", serviceURI); + log.info() + .attr("serviceUri", serviceURI) + .log("Service Uri is not specified, using default"); } return doApply(serviceURI, conf, globalFlags, cmdFlags); } diff --git a/tools/stream/src/main/java/org/apache/bookkeeper/stream/cli/commands/AdminCommand.java b/tools/stream/src/main/java/org/apache/bookkeeper/stream/cli/commands/AdminCommand.java index 333d670c1be..b7649c1aeec 100644 --- a/tools/stream/src/main/java/org/apache/bookkeeper/stream/cli/commands/AdminCommand.java +++ b/tools/stream/src/main/java/org/apache/bookkeeper/stream/cli/commands/AdminCommand.java @@ -17,7 +17,7 @@ */ package org.apache.bookkeeper.stream.cli.commands; -import lombok.extern.slf4j.Slf4j; +import lombok.CustomLog; import org.apache.bookkeeper.clients.StorageClientBuilder; import org.apache.bookkeeper.clients.admin.StorageAdminClient; import org.apache.bookkeeper.clients.config.StorageClientSettings; @@ -30,7 +30,7 @@ /** * An admin command interface provides a run method to execute admin commands. */ -@Slf4j +@CustomLog public abstract class AdminCommand extends AbstractStreamCommand { protected AdminCommand(CliSpec spec) { @@ -53,7 +53,7 @@ protected boolean doApply(ServiceURI serviceURI, run(admin, bkFlags, cmdFlags); return true; } catch (Exception e) { - log.error("Failed to process stream admin command", e); + log.error().exception(e).log("Failed to process stream admin command"); spec.console().println("Failed to process stream admin command"); e.printStackTrace(spec.console()); return false; diff --git a/tools/stream/src/main/java/org/apache/bookkeeper/stream/cli/commands/ClientCommand.java b/tools/stream/src/main/java/org/apache/bookkeeper/stream/cli/commands/ClientCommand.java index 576a1d7213a..f8db8b1355a 100644 --- a/tools/stream/src/main/java/org/apache/bookkeeper/stream/cli/commands/ClientCommand.java +++ b/tools/stream/src/main/java/org/apache/bookkeeper/stream/cli/commands/ClientCommand.java @@ -17,7 +17,7 @@ */ package org.apache.bookkeeper.stream.cli.commands; -import lombok.extern.slf4j.Slf4j; +import lombok.CustomLog; import org.apache.bookkeeper.api.StorageClient; import org.apache.bookkeeper.clients.StorageClientBuilder; import org.apache.bookkeeper.clients.config.StorageClientSettings; @@ -30,7 +30,7 @@ /** * An admin command interface provides a run method to execute admin commands. */ -@Slf4j +@CustomLog public abstract class ClientCommand extends AbstractStreamCommand { protected ClientCommand(CliSpec spec) { @@ -54,8 +54,10 @@ protected boolean doApply(ServiceURI serviceURI, run(client, cmdFlags); return true; } catch (Exception e) { - log.error("Failed to process commands under namespace '{}'", - globalFlags.namespace, e); + log.error() + .attr("namespace", globalFlags.namespace) + .exception(e) + .log("Failed to process commands under namespace"); spec.console().println("Failed to process stream admin command"); e.printStackTrace(spec.console()); return false; diff --git a/tools/stream/src/main/java/org/apache/bookkeeper/stream/cli/commands/cluster/InitClusterCommand.java b/tools/stream/src/main/java/org/apache/bookkeeper/stream/cli/commands/cluster/InitClusterCommand.java index 5e944da61b5..1a46af547c4 100644 --- a/tools/stream/src/main/java/org/apache/bookkeeper/stream/cli/commands/cluster/InitClusterCommand.java +++ b/tools/stream/src/main/java/org/apache/bookkeeper/stream/cli/commands/cluster/InitClusterCommand.java @@ -26,7 +26,7 @@ import com.google.common.util.concurrent.UncheckedExecutionException; import edu.umd.cs.findbugs.annotations.SuppressFBWarnings; import java.net.URI; -import lombok.extern.slf4j.Slf4j; +import lombok.CustomLog; import org.apache.bookkeeper.common.net.ServiceURI; import org.apache.bookkeeper.conf.ServerConfiguration; import org.apache.bookkeeper.meta.MetadataDrivers; @@ -51,7 +51,7 @@ /** * Command to init a cluster. */ -@Slf4j +@CustomLog public class InitClusterCommand extends BKCommand { private static final String NAME = OP_INIT; @@ -162,14 +162,15 @@ protected boolean apply(ServiceURI ignored, String ledgersUri = rootUri + ledgersPath; String dlogUri = rootUri + dlogPath; - log.info("Initializing cluster {} : \n" - + "\tledgers : path = {}, uri = {}\n" - + "\tdlog: path = {}, uri = {}\n" - + "\tstream storage: path = {}, num_storage_containers = {}", - clusterName, - ledgersPath, ledgersUri, - dlogPath, dlogUri, - StorageConstants.ZK_METADATA_ROOT_PATH, cmdFlags.numStorageContainers); + log.info() + .attr("clusterName", clusterName) + .attr("ledgersPath", ledgersPath) + .attr("ledgersUri", ledgersUri) + .attr("dlogPath", dlogPath) + .attr("dlogUri", dlogUri) + .attr("streamStoragePath", StorageConstants.ZK_METADATA_ROOT_PATH) + .attr("numStorageContainers", cmdFlags.numStorageContainers) + .log("Initializing cluster"); // create the cluster root path initializeCluster(client, clusterName); @@ -185,7 +186,7 @@ protected boolean apply(ServiceURI ignored, metadataServiceHosts, ledgersUri, cmdFlags.numStorageContainers); - log.info("Successfully initialized cluster {}", clusterName); + log.info().attr("clusterName", clusterName).log("Successfully initialized cluster"); return true; } catch (Exception e) { throw new RuntimeException(e); @@ -215,7 +216,9 @@ private void initLedgersMetadata(String ledgersUri) throws Exception { rm -> { try { if (rm.initNewCluster()) { - log.info("Successfully initialized ledgers metadata at {}", ledgersUri); + log.info() + .attr("ledgersUri", ledgersUri) + .log("Successfully initialized ledgers metadata"); } } catch (Exception e) { throw new UncheckedExecutionException("Failed to init ledgers metadata at " + ledgersUri, e); @@ -250,9 +253,10 @@ private void initStreamStorageMetadata(String metadataServiceHosts, int numStorageContainers) { ZkClusterInitializer initializer = new ZkClusterInitializer(metadataServiceHosts); if (initializer.initializeCluster(URI.create(ledgersUri), numStorageContainers)) { - log.info("Successfully initialized stream storage metadata at {}:{}", - metadataServiceHosts, - StorageConstants.ZK_METADATA_ROOT_PATH); + log.info() + .attr("metadataServiceHosts", metadataServiceHosts) + .attr("path", StorageConstants.ZK_METADATA_ROOT_PATH) + .log("Successfully initialized stream storage metadata"); } }