Skip to content

Commit

Permalink
Logging improvements (#141)
Browse files Browse the repository at this point in the history
  • Loading branch information
@GoodforGod
GoodforGod authored Jul 14, 2023
1 parent f536f82 commit 033b82a
Show file tree
Hide file tree
Showing 40 changed files with 475 additions and 286 deletions.
28 changes: 14 additions & 14 deletions application-graph/src/main/java/ru/tinkoff/kora/application/graph/GraphImpl.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -95,7 +95,7 @@ public Mono<Void> refresh(Node<?> fromNode) {
}
case ON_COMPLETE -> {
this.semaphore.release();
log.debug("Refreshing Graph completed took {}", Duration.ofNanos(System.nanoTime() - started));
log.debug("Refreshing Graph completed in {}", Duration.ofNanos(System.nanoTime() - started).toString().substring(2).toLowerCase());
}
default -> {}
}
Expand All @@ -110,22 +110,22 @@ public Mono<Void> init() {
root.set(0, this.objects.length());
this.semaphore.acquireUninterruptibly();

log.debug("Initializing Graph...");
log.debug("Graph Initializing...");
final long started = System.nanoTime();
return this.initializeSubgraph(root)
.doOnEach(s -> {
switch (s.getType()) {
case CANCEL -> {
this.semaphore.release();
log.debug("Initializing Graph cancelled");
log.debug("Graph Initializing cancelled");
}
case ON_ERROR -> {
this.semaphore.release();
log.debug("Initializing Graph error", s.getThrowable());
log.debug("Graph Initializing error", s.getThrowable());
}
case ON_COMPLETE -> {
this.semaphore.release();
log.debug("Initializing Graph completed took {}", Duration.ofNanos(System.nanoTime() - started));
log.debug("Graph Initializing completed in {}", Duration.ofNanos(System.nanoTime() - started).toString().substring(2).toLowerCase());
}
default -> {}
}
Expand All @@ -139,22 +139,22 @@ public Mono<Void> release() {
var root = new BitSet(this.objects.length());
root.set(0, this.objects.length());
this.semaphore.acquireUninterruptibly();
log.debug("Releasing Graph...");
log.debug("Graph Releasing...");
final long started = System.nanoTime();
return this.releaseNodes(this.objects, root)
.doOnEach(s -> {
switch (s.getType()) {
case CANCEL -> {
this.semaphore.release();
log.debug("Releasing graph cancelled");
log.debug("Graph Releasing cancelled");
}
case ON_ERROR -> {
this.semaphore.release();
log.debug("Releasing graph error", s.getThrowable());
log.debug("Graph Releasing error", s.getThrowable());
}
case ON_COMPLETE -> {
this.semaphore.release();
log.debug("Releasing graph completed took {}", Duration.ofNanos(System.nanoTime() - started));
log.debug("Graph Releasing completed in {}", Duration.ofNanos(System.nanoTime() - started).toString().substring(2).toLowerCase());
}
default -> {}
}
Expand All @@ -175,7 +175,7 @@ private Mono<Void> initializeSubgraph(BitSet root) {
for (var newPromise : tmpGraph.newPromises) {
newPromise.graph = GraphImpl.this;
}
log.trace("Graph refreshed, calling interceptors");
log.trace("Graph refreshed, calling interceptors...");
for (var refreshListenerNode : this.refreshListenerNodes) {
if (this.objects.get(refreshListenerNode) instanceof RefreshListener refreshListener) {
try {
Expand Down Expand Up @@ -386,14 +386,14 @@ private Mono<Void> initializeNode(Node<?> node, Lifecycle lifecycle) {
return lifecycle.init().then()
.doOnEach(s -> {
switch (s.getType()) {
case CANCEL -> this.rootGraph.log.trace("Initializing node {} of class {} cancelled", index, lifecycle.getClass());
case ON_SUBSCRIBE -> this.rootGraph.log.trace("Initializing node {} of class {}", index, lifecycle.getClass());
case ON_ERROR -> this.rootGraph.log.trace("Initializing node {} of class {} error", index, lifecycle.getClass(), s.getThrowable());
case CANCEL -> this.rootGraph.log.trace("Node Initializing {} of class {} cancelled", index, lifecycle.getClass());
case ON_SUBSCRIBE -> this.rootGraph.log.trace("Node Initializing {} of class {}", index, lifecycle.getClass());
case ON_ERROR -> this.rootGraph.log.trace("Node Initializing {} of class {} error", index, lifecycle.getClass(), s.getThrowable());
case ON_COMPLETE -> {
synchronized (TmpGraph.this) {
this.initialized.set(node.index);
}
this.rootGraph.log.trace("Initializing node {} of class {} complete", index, lifecycle.getClass());
this.rootGraph.log.trace("Node Initializing {} of class {} complete", index, lifecycle.getClass());
}
default -> {}
}
Expand Down
8 changes: 4 additions & 4 deletions application-graph/src/main/java/ru/tinkoff/kora/application/graph/KoraApplication.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -11,22 +11,22 @@ public static RefreshableGraph run(Supplier<ApplicationGraphDraw> supplier) {
var start = System.currentTimeMillis();
var graphDraw = supplier.get();
var log = LoggerFactory.getLogger(graphDraw.getRoot());
log.debug("Initializing Application...");
log.debug("Application initializing...");
try {
var graph = graphDraw.init().block();
var end = System.currentTimeMillis();
try {
var uptime = ManagementFactory.getRuntimeMXBean().getUptime() / 1000.0;
log.info("Initialized Application in {} ms (JVM running for {} s)", end - start, uptime);
log.info("Application initialized in {} ms (JVM running for {} s)", end - start, uptime);
} catch (Throwable ex) {
log.info("Initialized Application in {}ms", end - start);
log.info("Application initialized in {}ms", end - start);
}
var thread = new Thread(() -> graph.release().block());
thread.setName("kora-shutdown");
Runtime.getRuntime().addShutdownHook(thread);
return graph;
} catch (Exception e) {
log.error("Initializing Application failed with error", e);
log.error("Application initializing failed with error", e);
e.printStackTrace();
try {
Thread.sleep(100);// so async logger is able to write exception to log
Expand Down
12 changes: 6 additions & 6 deletions cache/cache-caffeine/src/main/java/ru/tinkoff/kora/cache/caffeine/CaffeineCache.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -29,15 +29,15 @@ String origin() {

@Override
public V get(@Nonnull K key) {
logger.trace("Looking for value in cache '{}' for key: {}", name, key);
logger.trace("Cache '{}' looking for value for key: {}", name, key);
final CacheTelemetry.TelemetryContext telemetryContext = telemetry.create(CacheTelemetry.Operation.Type.GET, name, origin());
telemetryContext.startRecording();

final V v = caffeine.getIfPresent(key);
if (v == null) {
logger.trace("Value NOT found in cache '{}' for key: {}", name, key);
logger.trace("Cache '{}' no value found for key: {}", name, key);
} else {
logger.debug("Value found in cache '{}' for key: {}", name, key);
logger.debug("Cache '{}' found value for key: {}", name, key);
}

telemetryContext.recordSuccess(v);
Expand All @@ -46,7 +46,7 @@ public V get(@Nonnull K key) {

@Nonnull
public V put(@Nonnull K key, @Nonnull V value) {
logger.trace("Putting value in cache '{}' for key: {}", name, key);
logger.trace("Cache '{}' storing for key: {}", name, key);
final CacheTelemetry.TelemetryContext telemetryContext = telemetry.create(CacheTelemetry.Operation.Type.PUT, name, origin());
telemetryContext.startRecording();

Expand All @@ -58,7 +58,7 @@ public V put(@Nonnull K key, @Nonnull V value) {

@Override
public void invalidate(@Nonnull K key) {
logger.trace("Invalidating value in cache '{}' for key: {}", name, key);
logger.trace("Cache '{}' invalidating for key: {}", name, key);
final CacheTelemetry.TelemetryContext telemetryContext = telemetry.create(CacheTelemetry.Operation.Type.INVALIDATE, name, origin());
telemetryContext.startRecording();

Expand All @@ -69,7 +69,7 @@ public void invalidate(@Nonnull K key) {

@Override
public void invalidateAll() {
logger.trace("Invalidating all values in cache '{}'", name);
logger.trace("Cache '{}' invalidating all values", name);
final CacheTelemetry.TelemetryContext telemetryContext = telemetry.create(CacheTelemetry.Operation.Type.INVALIDATE_ALL, name, origin());
telemetryContext.startRecording();

Expand Down
42 changes: 24 additions & 18 deletions cache/cache-redis/src/main/java/ru/tinkoff/kora/cache/redis/RedisCache.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -53,7 +53,7 @@ String origin() {

@Override
public V get(@Nonnull K key) {
logger.trace("Looking for value in cache '{}' for key: {}", name, key);
logger.trace("Cache '{}' looking for value for key: {}", name, key);
final byte[] keyAsBytes = keyMapper.apply(key);
final CacheTelemetry.TelemetryContext telemetryContext = telemetry.create(CacheTelemetry.Operation.Type.GET, name, origin());

Expand All @@ -64,9 +64,9 @@ public V get(@Nonnull K key) {
: syncClient.getExpire(keyAsBytes, expireAfterAccessMillis);

if (jsonAsBytes != null) {
logger.trace("Value NOT found in cache '{}' for key: {}", name, key);
logger.trace("Cache '{}' no value found for key: {}", name, key);
} else {
logger.debug("Value found in cache '{}' for key: {}", name, key);
logger.debug("Cache '{}' found value for key: {}", name, key);
}

final V v = valueMapper.read(jsonAsBytes);
Expand All @@ -82,7 +82,7 @@ public V get(@Nonnull K key) {
@Nonnull
@Override
public V put(@Nonnull K key, @Nonnull V value) {
logger.trace("Putting value in cache '{}' for key: {}", name, key);
logger.trace("Cache '{}' storing for key: {}", name, key);
final byte[] keyAsBytes = keyMapper.apply(key);
final byte[] valueAsBytes = valueMapper.write(value);
final CacheTelemetry.TelemetryContext telemetryContext = telemetry.create(CacheTelemetry.Operation.Type.PUT, name, origin());
Expand All @@ -95,7 +95,7 @@ public V put(@Nonnull K key, @Nonnull V value) {
syncClient.setExpire(keyAsBytes, valueAsBytes, expireAfterWriteMillis);
}
telemetryContext.recordSuccess();
logger.trace("Putted value in cache '{}' for key: {}", name, key);
logger.debug("Cache '{}' stored for key: {}", name, key);
return value;
} catch (Exception e) {
telemetryContext.recordFailure(e);
Expand All @@ -106,15 +106,15 @@ public V put(@Nonnull K key, @Nonnull V value) {

@Override
public void invalidate(@Nonnull K key) {
logger.trace("Invalidating value in cache '{}' for key: {}", name, key);
logger.trace("Cache '{}' invalidating for key: {}", name, key);
final byte[] keyAsBytes = keyMapper.apply(key);
final CacheTelemetry.TelemetryContext telemetryContext = telemetry.create(CacheTelemetry.Operation.Type.INVALIDATE, name, origin());

try {
telemetryContext.startRecording();
syncClient.del(keyAsBytes);
telemetryContext.recordSuccess();
logger.trace("Invalidated value in cache '{}' for key: {}", name, key);
logger.debug("Cache '{}' invalidated for key: {}", name, key);
} catch (Exception e) {
telemetryContext.recordFailure(e);
logger.warn(e.getMessage(), e);
Expand All @@ -123,14 +123,14 @@ public void invalidate(@Nonnull K key) {

@Override
public void invalidateAll() {
logger.trace("Invalidating all values in cache '{}'", name);
logger.trace("Cache '{}' invalidating all", name);
final CacheTelemetry.TelemetryContext telemetryContext = telemetry.create(CacheTelemetry.Operation.Type.INVALIDATE_ALL, name, origin());

try {
telemetryContext.startRecording();
syncClient.flushAll();
telemetryContext.recordSuccess();
logger.trace("Invalidated all values in cache '{}'", name);
logger.debug("Cache '{}' invalidated all", name);
} catch (Exception e) {
telemetryContext.recordFailure(e);
logger.warn(e.getMessage(), e);
Expand All @@ -143,17 +143,17 @@ public Mono<V> getAsync(@Nonnull K key) {
final CacheTelemetry.TelemetryContext telemetryContext = telemetry.create(CacheTelemetry.Operation.Type.GET, name, origin());
return Mono.fromCallable(() -> keyMapper.apply(key))
.flatMap(keyAsBytes -> {
logger.trace("Looking for value in cache '{}' for key: {}", name, key);
logger.trace("Cache '{}' looking for value for key: {}", name, key);
telemetryContext.startRecording();
return (expireAfterAccessMillis == null)
? reactiveClient.get(keyAsBytes)
: reactiveClient.getExpire(keyAsBytes, expireAfterAccessMillis);
})
.map(jsonAsBytes -> {
if (jsonAsBytes != null) {
logger.trace("Value NOT found in cache '{}' for key: {}", name, key);
logger.trace("Cache '{}' no value found for key: {}", name, key);
} else {
logger.debug("Value found in cache '{}' for key: {}", name, key);
logger.debug("Cache '{}' found value for key: {}", name, key);
}
final V v = valueMapper.read(jsonAsBytes);
telemetryContext.recordSuccess(jsonAsBytes);
Expand All @@ -172,7 +172,7 @@ public Mono<V> putAsync(@Nonnull K key, @Nonnull V value) {
final CacheTelemetry.TelemetryContext telemetryContext = telemetry.create(CacheTelemetry.Operation.Type.PUT, name, origin());
return Mono.fromCallable(() -> keyMapper.apply(key))
.flatMap(keyAsBytes -> {
logger.trace("Putting value in cache '{}' for key: {}", name, key);
logger.trace("Cache '{}' storing for key: {}", name, key);
final byte[] valueAsBytes = valueMapper.write(value);
telemetryContext.startRecording();
return (expireAfterWriteMillis == null)
Expand All @@ -182,7 +182,7 @@ public Mono<V> putAsync(@Nonnull K key, @Nonnull V value) {
.map(r -> value)
.switchIfEmpty(Mono.fromCallable(() -> {
telemetryContext.recordSuccess();
logger.trace("Putted value in cache '{}' for key: {}", name, key);
logger.debug("Cache '{}' stored for key: {}", name, key);
return value;
}))
.onErrorResume(e -> {
Expand All @@ -197,11 +197,14 @@ public Mono<V> putAsync(@Nonnull K key, @Nonnull V value) {
public Mono<Void> invalidateAsync(@Nonnull K key) {
final CacheTelemetry.TelemetryContext telemetryContext = telemetry.create(CacheTelemetry.Operation.Type.INVALIDATE, name, origin());
return Mono.fromCallable(() -> {
logger.trace("Invalidating value in cache '{}' for key: {}", name, key);
logger.trace("Cache '{}' invalidating for key: {}", name, key);
return keyMapper.apply(key);
})
.flatMap(reactiveClient::del)
.doOnSuccess(r -> telemetryContext.recordSuccess())
.doOnSuccess(r -> {
telemetryContext.recordSuccess();
logger.debug("Cache '{}' invalidated for key: {}", name, key);
})
.onErrorResume(e -> {
telemetryContext.recordFailure(e);
logger.warn(e.getMessage(), e);
Expand All @@ -213,9 +216,12 @@ public Mono<Void> invalidateAsync(@Nonnull K key) {
@Override
public Mono<Void> invalidateAllAsync() {
final CacheTelemetry.TelemetryContext telemetryContext = telemetry.create(CacheTelemetry.Operation.Type.INVALIDATE_ALL, name, origin());
logger.trace("Invalidating all values in cache '{}'", name);
logger.trace("Cache '{}' invalidating all", name);
return reactiveClient.flushAll()
.doOnSuccess(r -> telemetryContext.recordSuccess())
.doOnSuccess(r -> {
telemetryContext.recordSuccess();
logger.debug("Cache '{}' invalidated all", name);
})
.onErrorResume(e -> {
telemetryContext.recordFailure(e);
logger.warn(e.getMessage(), e);
Expand Down
8 changes: 4 additions & 4 deletions ...cache-redis/src/main/java/ru/tinkoff/kora/cache/redis/client/DefaultLettuceCommander.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -31,7 +31,7 @@ public DefaultLettuceCommander(AbstractRedisClient redisClient) {
@Override
public Mono<?> init() {
return ReactorUtils.ioMono(() -> {
logger.debug("Starting Redis Client (Lettuce)...");
logger.debug("Redis Client (Lettuce) starting...");
final long started = System.nanoTime();

try {
Expand All @@ -56,18 +56,18 @@ public Mono<?> init() {
throw Exceptions.propagate(e);
}

logger.info("Started Redis Client (Lettuce) took {}", Duration.ofNanos(System.nanoTime() - started));
logger.info("Redis Client (Lettuce) started in {}", Duration.ofNanos(System.nanoTime() - started).toString().substring(2).toLowerCase());
});
}

@Override
public Mono<?> release() {
return ReactorUtils.ioMono(() -> {
try {
logger.debug("Stopping Redis Client (Lettuce)...");
logger.debug("Redis Client (Lettuce) stopping...");
final long started = System.nanoTime();
connection.close();
logger.info("Stopping Redis Client (Lettuce) took {}", Duration.ofNanos(System.nanoTime() - started));
logger.info("Redis Client (Lettuce) stopped in {}", Duration.ofNanos(System.nanoTime() - started).toString().substring(2).toLowerCase());
} catch (Exception e) {
throw Exceptions.propagate(e);
}
Expand Down
1 change: 1 addition & 0 deletions ...atabase-cassandra/src/main/java/ru/tinkoff/kora/database/cassandra/CassandraDatabase.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -21,6 +21,7 @@ public CassandraDatabase(CassandraConfig config, DataBaseTelemetryFactory teleme
this.telemetry = telemetryFactory.get(
Objects.requireNonNullElse(config.basic().sessionName(), "cassandra"),
"cassandra",
"cassandra",
Optional.ofNullable(config.auth()).map(CassandraConfig.CassandraCredentials::login).orElse("anonymous")
);
}
Expand Down
2 changes: 1 addition & 1 deletion ...tabase-common/src/main/java/ru/tinkoff/kora/database/common/telemetry/DataBaseLogger.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -9,5 +9,5 @@ public interface DataBaseLogger {

void logQueryBegin(QueryContext queryContext);

void logQueryEnd(long duration, QueryContext queryContext, @Nullable Throwable ex);
void logQueryEnd(long processingTime, QueryContext queryContext, @Nullable Throwable ex);
}
1 change: 1 addition & 0 deletions ...common/src/main/java/ru/tinkoff/kora/database/common/telemetry/DataBaseLoggerFactory.java
View file
Original file line number Diff line number Diff line change
@@ -1,6 +1,7 @@
package ru.tinkoff.kora.database.common.telemetry;

public interface DataBaseLoggerFactory {

DataBaseLogger get(String poolName);

final class DefaultDataBaseLoggerFactory implements DataBaseLoggerFactory {
Expand Down
2 changes: 1 addition & 1 deletion ...mon/src/main/java/ru/tinkoff/kora/database/common/telemetry/DataBaseTelemetryFactory.java
View file
Original file line number Diff line number Diff line change
@@ -1,5 +1,5 @@
package ru.tinkoff.kora.database.common.telemetry;

public interface DataBaseTelemetryFactory {
DataBaseTelemetry get(String name, String dbType, String username);
DataBaseTelemetry get(String name, String driverType, String dbType, String username);
}
Loading

0 comments on commit 033b82a

Please sign in to comment.