Skip to content

Commit

Permalink
[SDCISA-15833, swisspost#170] Replace stacktraces by identifiers to e…
Browse files Browse the repository at this point in the history 
…nhance performance.
  • Loading branch information
@hiddenalpha
hiddenalpha committed May 16, 2024
1 parent 743463a commit 2bf6fb9
Show file tree
Hide file tree
Showing 5 changed files with 34 additions and 32 deletions.
2 changes: 1 addition & 1 deletion src/main/java/org/swisspush/redisques/QueueStatsService.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -68,7 +68,7 @@ public QueueStatsService(

public <CTX> void getQueueStats(CTX mCtx, GetQueueStatsMentor<CTX> mentor) {
if (!incomingRequestQuota.tryAcquire()) {
var ex = new RuntimeException("Server too busy to handle yet-another-queue-stats-request now");
var ex = new NoStacktraceException("Server too busy to handle yet-another-queue-stats-request now (error_bG8CAJJ3AgCKMwIAPBUC)");
vertx.runOnContext(v -> mentor.onError(ex, mCtx));
return;
}
Expand Down
48 changes: 24 additions & 24 deletions src/main/java/org/swisspush/redisques/RedisQues.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -20,6 +20,7 @@
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.swisspush.redisques.action.QueueAction;
import org.swisspush.redisques.exception.NoStacktraceException;
import org.swisspush.redisques.handler.RedisquesHttpRequestHandler;
import org.swisspush.redisques.performance.UpperBoundParallel;
import org.swisspush.redisques.scheduling.PeriodicSkipScheduler;
Expand Down Expand Up @@ -546,18 +547,18 @@ private void registerActiveQueueRegistrationRefresh() {
log.debug("RedisQues Periodic consumer refresh for active queue {}", queue);
refreshRegistration(queue, ev -> {
if (ev.failed())
log.warn("TODO error handling", new Exception(ev.cause()));
log.warn("TODO_1WoCAD1bAgBRBwIA9yQC error handling", ev.cause());
updateTimestamp(queue, updateTimestampEv -> {
if (updateTimestampEv.failed()) log.warn("TODO error handling",
new RuntimeException(updateTimestampEv.cause()));
if (updateTimestampEv.failed()) log.warn(
"TODO_mWoCAARJAgDTXAIASwUC error handling", updateTimestampEv.cause());
onDone.run();
});
});
} else {
log.debug("RedisQues Removing queue {} from the list", queue);
myQueues.remove(queue);
queueStatisticsCollector.resetQueueFailureStatistics(queue, (Throwable ex, Void v) -> {
if (ex != null) log.warn("TODO error handling", new RuntimeException(ex));
if (ex != null) log.warn("TODO_RGsCAA4CAgDdWAIAVRMC error handling", ex);
onDone.run();
});
}
Expand Down Expand Up @@ -592,7 +593,7 @@ private Handler<Message<JsonObject>> operationsHandler() {
// handle system operations
switch (queueOperation) {
case check:
checkQueues().onFailure(ex -> log.error("TODO error handling", new RuntimeException(ex)));
checkQueues().onFailure(ex -> log.warn("TODO_e2UCAGlXAgBAZQIAlBcC error handling", ex));
return;
case reset:
resetConsumers();
Expand Down Expand Up @@ -643,7 +644,7 @@ private void registerQueueCheck() {
log.info("periodic queue check is triggered now");
return checkQueues();
}).onFailure(ex -> {
log.error("TODO error handling", ex);
log.warn("TODO_WmYCAIxGAgB0aAIA4yoC error handling", ex);
});
});
}
Expand All @@ -664,10 +665,10 @@ public void stop() {

private void gracefulStop(final Handler<Void> doneHandler) {
consumersMessageConsumer.unregister(event -> uidMessageConsumer.unregister(unregisterEvent -> {
if( event.failed() ) log.warn("TODO error handling", new Exception(event.cause()));
if( event.failed() ) log.warn("TODO_FWsCAAE0AgCPRQIA8QgC error handling", event.cause());
unregisterConsumers(false).onComplete(unregisterConsumersEvent -> {
if( unregisterEvent.failed() )
log.warn("TODO error handling", new Exception(unregisterEvent.cause()));
log.warn("TODO_82sCAAczAgCQVgIA6E0C error handling", unregisterEvent.cause());
stoppedHandler = doneHandler;
if (myQueues.keySet().isEmpty()) {
doneHandler.handle(null);
Expand All @@ -687,7 +688,7 @@ private Future<Void> unregisterConsumers(boolean force) {
if (force || entry.getValue() == QueueState.READY) {
log.trace("RedisQues unregister consumers queue: {}", queue);
refreshRegistration(queue, event -> {
if( event.failed() ) log.warn("TODO error handling", new Exception(event.cause()));
if( event.failed() ) log.warn("TODO_12sCABtNAgAqKAIAWg0C error handling", event.cause());
// Make sure that I am still the registered consumer
String consumerKey = consumersPrefix + queue;
log.trace("RedisQues unregister consumers get: {}", consumerKey);
Expand All @@ -713,7 +714,7 @@ private Future<Void> unregisterConsumers(boolean force) {
}
}
CompositeFuture.all(futureList).onComplete(ev -> {
if( ev.failed() ) log.warn("TODO error handling", new Exception(ev.cause()));
if( ev.failed() ) log.warn("TODO_m2sCABpzAgCHbQIA5TYC error handling", ev.cause());
result.complete();
});
return result.future();
Expand Down Expand Up @@ -786,7 +787,7 @@ private Future<Void> consume(final String queueName) {
log.debug("RedisQues Starting to consume queue {}", queueName);
readQueue(queueName).onComplete(readQueueEvent -> {
if( readQueueEvent.failed() )
log.warn("TODO error handling", new Exception(readQueueEvent.cause()));
log.warn("TODO_aWwCACYIAgCpZwIA8RgC error handling", readQueueEvent.cause());
promise.complete();
});
} else {
Expand All @@ -799,7 +800,7 @@ private Future<Void> consume(final String queueName) {
myQueues.remove(queueName);
notifyConsumer(queueName).onComplete(notifyConsumerEvent -> {
if( notifyConsumerEvent.failed() )
log.warn("TODO error handling", notifyConsumerEvent.cause());
log.warn("TODO_Q2wCAHFYAgCgRQIA8yoC error handling", notifyConsumerEvent.cause());
promise.complete();
});
}
Expand Down Expand Up @@ -839,7 +840,7 @@ private Future<Void> readQueue(final String queueName) {

isQueueLocked(queueName).onComplete(lockAnswer -> {
if( lockAnswer.failed() )
throw new UnsupportedOperationException("TODO error handling " + queueName, lockAnswer.cause());
throw new NoStacktraceException("TODO_o2wCACFRAgAAKAIAMUUC error handling " + queueName, lockAnswer.cause());
boolean locked = lockAnswer.result();
if (!locked) {
redisProvider.redis().onSuccess(redisAPI -> redisAPI.lindex(queueKey, "0", answer -> {
Expand Down Expand Up @@ -877,12 +878,12 @@ private Future<Void> readQueue(final String queueName) {
if (answer1.succeeded() && answer1.result() != null && answer1.result().toInteger() > 0) {
notifyConsumer(queueName).onComplete(event1 -> {
if( event1.failed() )
log.warn("TODO error handling", new Exception(event1.cause()));
log.warn("TODO_d20CAD8wAgCbZQIAvSgC error handling", event1.cause());
promise.complete();
});
} else {
if( answer1.failed() )
log.warn("TODO error handling", new Exception(answer1.cause()));
log.warn("TODO_bm0CAEFuAgBcUAIAfQ8C error handling", answer1.cause());
promise.complete();
}
});
Expand All @@ -892,7 +893,7 @@ private Future<Void> readQueue(final String queueName) {
if (stoppedHandler != null) {
unregisterConsumers(false).onComplete(event -> {
if( event.failed() )
log.warn("TODO error handling", new Exception(event.cause()));
log.warn("TODO_GW0CAMUlAgDcJgIAzncC error handling", event.cause());
if (myQueues.isEmpty()) {
stoppedHandler.handle(null);
}
Expand Down Expand Up @@ -951,7 +952,7 @@ private void rescheduleSendMessageAfterFailure(final String queueName, int retry
log.debug("RedisQues re-notify the consumer of queue '{}' at {}", queueName, new Date(System.currentTimeMillis()));
}
notifyConsumer(queueName).onComplete(event -> {
if( event.failed() ) log.warn("TODO error handling", new Exception(event.cause()));
if( event.failed() ) log.warn("TODO_6m0CAExTAgCYSgIAnxkC error handling", event.cause());
// reset the queue state to be consumed by {@link RedisQues#consume(String)}
myQueues.put(queueName, QueueState.READY);
});
Expand Down Expand Up @@ -1099,7 +1100,6 @@ private Future<Void> checkQueues() {
assert ctx.iter == null;
ctx.counter = new AtomicInteger(queues.size());
ctx.iter = queues.iterator();
// MUST pre-initialize ALL futures, so that our 'Future.all()' call knows
log.trace("RedisQues update queues: {}", ctx.counter);
var p = Promise.<Void>promise();
upperBoundParallel.request(checkQueueRequestsQuota, null, new UpperBoundParallel.Mentor<Void>() {
Expand All @@ -1114,11 +1114,11 @@ private Future<Void> checkQueues() {
// Make sure its TTL is correctly set (replaces the previous orphan detection mechanism).
refreshRegistration(queueName, refreshRegistrationEvent -> {
if( refreshRegistrationEvent.failed() )
log.warn("TODO error handling", new Exception(refreshRegistrationEvent.cause()));
log.warn("TODO_V2kCANpoAgAqHwIA3TIC error handling", refreshRegistrationEvent.cause());
// And trigger its consumer.
notifyConsumer(queueName).onComplete(notifyConsumerEvent -> {
if( notifyConsumerEvent.failed() )
log.warn("TODO error handling", new Exception(notifyConsumerEvent.cause()));
log.warn("TODO_3WkCACRTAgDbdwIAAEwC error handling", notifyConsumerEvent.cause());
onDone.accept(null, null);
});
});
Expand All @@ -1141,7 +1141,7 @@ private Future<Void> checkQueues() {
if (ctx.counter.decrementAndGet() == 0) {
removeOldQueues(ctx.limit).onComplete(removeOldQueuesEvent -> {
if (removeOldQueuesEvent.failed())
log.warn("TODO error handling", new Exception(removeOldQueuesEvent.cause()));
log.warn("TODO_i2kCAChDAgDrbAIAIWcC error handling", removeOldQueuesEvent.cause());
refreshRegHandler.handle(null);
});
} else {
Expand All @@ -1162,7 +1162,7 @@ private Future<Void> checkQueues() {
if (ctx.counter.decrementAndGet() == 0) {
removeOldQueues(ctx.limit).onComplete(removeOldQueuesEvent -> {
if( removeOldQueuesEvent.failed() )
log.warn("TODO error handling", new Exception(removeOldQueuesEvent.cause()));
log.warn("TODO_aGoCAOtOAgAnbQIA4zQC error handling", removeOldQueuesEvent.cause());
queueStatisticsCollector.resetQueueFailureStatistics(queueName, onDone);
});
} else {
Expand All @@ -1174,7 +1174,7 @@ private Future<Void> checkQueues() {
return ctx.iter.hasNext();
}
@Override public boolean onError(Throwable ex, Void ctx_) {
log.warn("TODO error handling", new RuntimeException(ex));
log.warn("TODO_LWoCABp8AgAiIQIAlC4C error handling", ex);
return true; // true, keep going with other queues.
}
@Override public void onDone(Void ctx_) {
Expand Down Expand Up @@ -1208,7 +1208,7 @@ private Future<Void> removeOldQueues(long limit) {
redisProvider.redis()
.onSuccess(redisAPI -> {
redisAPI.zremrangebyscore(queuesKey, "-inf", String.valueOf(limit), event -> {
if( event.failed() ) log.warn("TODO error handling", event.cause());
if( event.failed() ) log.warn("TODO_92oCAP5SAgCcOgIAMW8C error handling", event.cause());
promise.complete();
});
})
Expand Down
5 changes: 3 additions & 2 deletions src/main/java/org/swisspush/redisques/handler/GetQueuesItemsCountHandler.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -132,14 +132,15 @@ public void handle(AsyncResult<Response> handleQueues) {
}
var obj = new JsonObject().put(STATUS, OK).put(QUEUES, result);
long jsonCreateDurationMs = currentTimeMillis() - beginEpchMs;
log.info("Creating JSON with {} entries took {}ms", ctx.queueLengths.length, jsonCreateDurationMs);
log.info("Creating JSON with {} entries did block this tread for {}ms",
ctx.queueLengths.length, jsonCreateDurationMs);
workerPromise.complete(obj);
}, false);
}).onSuccess((JsonObject json) -> {
log.trace("call event.reply(json)");
event.reply(json);
}).onFailure(ex -> {
log.warn("Redis: Failed to get queue length.", new Exception(ex));
log.warn("Redis: Failed to get queue length (error_c3gCAEFrAgChbAIAdhwC)", ex);
event.reply(new JsonObject().put(STATUS, ERROR));
});
}
Expand Down
2 changes: 1 addition & 1 deletion src/main/java/org/swisspush/redisques/handler/RedisquesHttpRequestHandler.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -63,7 +63,7 @@ public class RedisquesHttpRequestHandler implements Handler<HttpServerRequest> {
private static final String EMPTY_QUEUES_PARAM = "emptyQueues";
private static final String DELETED = "deleted";

/** @deprecated TODO <a href="https://xkcd.com/1179/">about date formats</a> */
/** @deprecated <a href="https://xkcd.com/1179/">about obsolete date formats</a> */
@Deprecated
private static final SimpleDateFormat DATE_FORMAT = new SimpleDateFormat("dd.MM.yyyy HH:mm:ss");

Expand Down
9 changes: 5 additions & 4 deletions src/main/java/org/swisspush/redisques/util/QueueStatisticsCollector.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -15,6 +15,7 @@
import io.vertx.redis.client.impl.types.NumberType;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.swisspush.redisques.exception.NoStacktraceException;
import org.swisspush.redisques.performance.UpperBoundParallel;

import java.util.HashMap;
Expand Down Expand Up @@ -361,18 +362,18 @@ private void updateStatisticsInRedis(String queueName, BiConsumer<Throwable, Voi
redisProvider.redis()
.onSuccess(redisAPI -> {
redisAPI.hset(List.of(STATSKEY, queueName, obj.toString()), ev -> {
onDone.accept(ev.failed() ? new RuntimeException("stack", ev.cause()) : null, null);
onDone.accept(ev.failed() ? new NoStacktraceException("TODO_Wn0CANwoAgAZDwIA20gC error handling", ev.cause()) : null, null);
});
})
.onFailure(ex -> onDone.accept(new RuntimeException("stack", ex), null));
.onFailure(ex -> onDone.accept(new NoStacktraceException("TODO_H30CACQ6AgAUWwIAoCYC error handling", ex), null));
} else {
redisProvider.redis()
.onSuccess(redisAPI -> {
redisAPI.hdel(List.of(STATSKEY, queueName), ev -> {
onDone.accept(ev.failed() ? new RuntimeException("stack", ev.cause()) : null, null);
onDone.accept(ev.failed() ? new NoStacktraceException("TODO_Vn4CACQIAgDeLAIAUyEC error handling", ev.cause()) : null, null);
});
})
.onFailure(ex -> onDone.accept(new RuntimeException("stack", ex), null));
.onFailure(ex -> onDone.accept(new NoStacktraceException("TODO_Kn4CABJoAgBvaQIA7QQC error handling", ex), null));
}
} catch (RuntimeException ex) {
onDone.accept(ex, null);
Expand Down

0 comments on commit 2bf6fb9

Please sign in to comment.