andreacavalli/rockserver
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Print internal errors

Browse Source
This commit is contained in:
Andrea Cavalli 2024-10-24 01:32:10 +02:00
parent 4646f24fb3
commit 8fa0a94b62
5 changed files with 39 additions and 33 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

7
src/fatjar/java/module-info.java
View File

@ -29,9 +29,9 @@ module rockserver.core {
requires io.netty.transport.classes.epoll; requires io.netty.transport.classes.epoll;
requires org.reactivestreams; requires org.reactivestreams;
requires io.netty.transport.unix.common; requires io.netty.transport.unix.common;
requires reactor.core; requires reactor.core;
requires reactor.grpc.stub; requires reactor.grpc.stub;
requires java.annotation; requires java.annotation;
exports it.cavallium.rockserver.core.client; exports it.cavallium.rockserver.core.client;
exports it.cavallium.rockserver.core.common; exports it.cavallium.rockserver.core.common;
@ -40,4 +40,5 @@ module rockserver.core {
opens it.cavallium.rockserver.core.config to org.github.gestalt.core, org.github.gestalt.hocon; opens it.cavallium.rockserver.core.config to org.github.gestalt.core, org.github.gestalt.hocon;
exports it.cavallium.rockserver.core.impl.rocksdb; exports it.cavallium.rockserver.core.impl.rocksdb;
exports it.cavallium.rockserver.core.impl; exports it.cavallium.rockserver.core.impl;
exports it.cavallium.rockserver.core.common.api.proto to protobuf.java;
} }

1
src/library/java/module-info.java
View File

@ -38,4 +38,5 @@ module rockserver.core {
opens it.cavallium.rockserver.core.config to org.github.gestalt.core, org.github.gestalt.hocon; opens it.cavallium.rockserver.core.config to org.github.gestalt.core, org.github.gestalt.hocon;
exports it.cavallium.rockserver.core.impl.rocksdb; exports it.cavallium.rockserver.core.impl.rocksdb;
exports it.cavallium.rockserver.core.impl; exports it.cavallium.rockserver.core.impl;
exports it.cavallium.rockserver.core.common.api.proto to protobuf.java;
} }

3
src/main/java/it/cavallium/rockserver/core/common/RocksDBException.java
View File

@ -34,7 +34,8 @@ public class RocksDBException extends RuntimeException {
SST_GET_SIZE_FAILED, SST_GET_SIZE_FAILED,
UNSUPPORTED_COLUMN_TYPE, UNSUPPORTED_COLUMN_TYPE,
NOT_IMPLEMENTED, NOT_IMPLEMENTED,
GET_PROPERTY_ERROR GET_PROPERTY_ERROR,
INTERNAL_ERROR
} }
public static RocksDBException of(RocksDBErrorType errorUniqueId, String message) { public static RocksDBException of(RocksDBErrorType errorUniqueId, String message) {

60
src/main/java/it/cavallium/rockserver/core/server/GrpcServer.java
View File

@ -34,6 +34,7 @@ import it.cavallium.rockserver.core.common.RequestType.RequestMulti;
import it.cavallium.rockserver.core.common.RequestType.RequestNothing; import it.cavallium.rockserver.core.common.RequestType.RequestNothing;
import it.cavallium.rockserver.core.common.RequestType.RequestPrevious; import it.cavallium.rockserver.core.common.RequestType.RequestPrevious;
import it.cavallium.rockserver.core.common.RequestType.RequestPreviousPresence; import it.cavallium.rockserver.core.common.RequestType.RequestPreviousPresence;
import it.cavallium.rockserver.core.common.RocksDBException.RocksDBErrorType;
import it.cavallium.rockserver.core.common.api.proto.*; import it.cavallium.rockserver.core.common.api.proto.*;
import it.cavallium.rockserver.core.common.api.proto.Delta; import it.cavallium.rockserver.core.common.api.proto.Delta;
import it.cavallium.rockserver.core.common.api.proto.FirstAndLast; import it.cavallium.rockserver.core.common.api.proto.FirstAndLast;
@ -124,7 +125,7 @@ public class GrpcServer extends Server {
return executeSync(() -> { return executeSync(() -> {
var txId = api.openTransaction(request.getTimeoutMs()); var txId = api.openTransaction(request.getTimeoutMs());
return OpenTransactionResponse.newBuilder().setTransactionId(txId).build(); return OpenTransactionResponse.newBuilder().setTransactionId(txId).build();
}).transform(this.onErrorMapMonoWithRequestInfo(request)); }).transform(this.onErrorMapMonoWithRequestInfo("openTransaction", request));
} }
@Override @Override
@ -132,7 +133,7 @@ public class GrpcServer extends Server {
return executeSync(() -> { return executeSync(() -> {
var committed = api.closeTransaction(request.getTransactionId(), request.getCommit()); var committed = api.closeTransaction(request.getTransactionId(), request.getCommit());
return CloseTransactionResponse.newBuilder().setSuccessful(committed).build(); return CloseTransactionResponse.newBuilder().setSuccessful(committed).build();
}).transform(this.onErrorMapMonoWithRequestInfo(request)); }).transform(this.onErrorMapMonoWithRequestInfo("closeTransaction", request));
} }
@Override @Override
@ -140,7 +141,7 @@ public class GrpcServer extends Server {
return executeSync(() -> { return executeSync(() -> {
api.closeFailedUpdate(request.getUpdateId()); api.closeFailedUpdate(request.getUpdateId());
return Empty.getDefaultInstance(); return Empty.getDefaultInstance();
}).transform(this.onErrorMapMonoWithRequestInfo(request)); }).transform(this.onErrorMapMonoWithRequestInfo("closeFailedUpdate", request));
} }
@Override @Override
@ -148,7 +149,7 @@ public class GrpcServer extends Server {
return executeSync(() -> { return executeSync(() -> {
var colId = api.createColumn(request.getName(), mapColumnSchema(request.getSchema())); var colId = api.createColumn(request.getName(), mapColumnSchema(request.getSchema()));
return CreateColumnResponse.newBuilder().setColumnId(colId).build(); return CreateColumnResponse.newBuilder().setColumnId(colId).build();
}).transform(this.onErrorMapMonoWithRequestInfo(request)); }).transform(this.onErrorMapMonoWithRequestInfo("createColumn", request));
} }
@Override @Override
@ -156,7 +157,7 @@ public class GrpcServer extends Server {
return executeSync(() -> { return executeSync(() -> {
api.deleteColumn(request.getColumnId()); api.deleteColumn(request.getColumnId());
return Empty.getDefaultInstance(); return Empty.getDefaultInstance();
}).transform(this.onErrorMapMonoWithRequestInfo(request)); }).transform(this.onErrorMapMonoWithRequestInfo("deleteColumn", request));
} }
@Override @Override
@ -164,7 +165,7 @@ public class GrpcServer extends Server {
return executeSync(() -> { return executeSync(() -> {
var colId = api.getColumnId(request.getName()); var colId = api.getColumnId(request.getName());
return GetColumnIdResponse.newBuilder().setColumnId(colId).build(); return GetColumnIdResponse.newBuilder().setColumnId(colId).build();
}).transform(this.onErrorMapMonoWithRequestInfo(request)); }).transform(this.onErrorMapMonoWithRequestInfo("getColumnId", request));
} }
@Override @Override
@ -180,7 +181,7 @@ public class GrpcServer extends Server {
); );
} }
return Empty.getDefaultInstance(); return Empty.getDefaultInstance();
}).transform(this.onErrorMapMonoWithRequestInfo(request)); }).transform(this.onErrorMapMonoWithRequestInfo("put", request));
} }
@Override @Override
@ -209,7 +210,7 @@ public class GrpcServer extends Server {
return Mono return Mono
.fromFuture(() -> asyncApi.putBatchAsync(initialRequest.getColumnId(), batches, mode)) .fromFuture(() -> asyncApi.putBatchAsync(initialRequest.getColumnId(), batches, mode))
.transform(this.onErrorMapMonoWithRequestInfo(initialRequest)); .transform(this.onErrorMapMonoWithRequestInfo("putBatch", initialRequest));
} else if (firstSignal.isOnComplete()) { } else if (firstSignal.isOnComplete()) {
return Mono.just(RocksDBException.of( return Mono.just(RocksDBException.of(
RocksDBException.RocksDBErrorType.PUT_INVALID_REQUEST, "No initial request")); RocksDBException.RocksDBErrorType.PUT_INVALID_REQUEST, "No initial request"));
@ -244,7 +245,7 @@ public class GrpcServer extends Server {
new RequestNothing<>()); new RequestNothing<>());
} }
}) })
.transform(this.onErrorMapFluxWithRequestInfo(initialRequest)); .transform(this.onErrorMapFluxWithRequestInfo("putMulti", initialRequest));
} else if (firstSignal.isOnComplete()) { } else if (firstSignal.isOnComplete()) {
return Mono.just(RocksDBException.of( return Mono.just(RocksDBException.of(
RocksDBException.RocksDBErrorType.PUT_INVALID_REQUEST, "No initial request")); RocksDBException.RocksDBErrorType.PUT_INVALID_REQUEST, "No initial request"));
@ -271,7 +272,7 @@ public class GrpcServer extends Server {
} }
return prevBuilder.build(); return prevBuilder.build();
} }
}).transform(this.onErrorMapMonoWithRequestInfo(request)); }).transform(this.onErrorMapMonoWithRequestInfo("putGetPrevious", request));
} }
@Override @Override
@ -294,7 +295,7 @@ public class GrpcServer extends Server {
} }
return deltaBuilder.build(); return deltaBuilder.build();
} }
}).transform(this.onErrorMapMonoWithRequestInfo(request)); }).transform(this.onErrorMapMonoWithRequestInfo("putGetDelta", request));
} }
@Override @Override
@ -310,7 +311,7 @@ public class GrpcServer extends Server {
); );
return Changed.newBuilder().setChanged(changed).build(); return Changed.newBuilder().setChanged(changed).build();
} }
}).transform(this.onErrorMapMonoWithRequestInfo(request)); }).transform(this.onErrorMapMonoWithRequestInfo("putGetChanged", request));
} }
@Override @Override
@ -326,7 +327,7 @@ public class GrpcServer extends Server {
); );
return PreviousPresence.newBuilder().setPresent(present).build(); return PreviousPresence.newBuilder().setPresent(present).build();
} }
}).transform(this.onErrorMapMonoWithRequestInfo(request)); }).transform(this.onErrorMapMonoWithRequestInfo("putGetPreviousPresence", request));
} }
@Override @Override
@ -345,7 +346,7 @@ public class GrpcServer extends Server {
} }
return responseBuilder.build(); return responseBuilder.build();
} }
}).transform(this.onErrorMapMonoWithRequestInfo(request)); }).transform(this.onErrorMapMonoWithRequestInfo("get", request));
} }
@Override @Override
@ -365,7 +366,7 @@ public class GrpcServer extends Server {
} }
return responseBuilder.build(); return responseBuilder.build();
} }
}).transform(this.onErrorMapMonoWithRequestInfo(request)); }).transform(this.onErrorMapMonoWithRequestInfo("getForUpdate", request));
} }
@Override @Override
@ -380,7 +381,7 @@ public class GrpcServer extends Server {
); );
return PreviousPresence.newBuilder().setPresent(exists).build(); return PreviousPresence.newBuilder().setPresent(exists).build();
} }
}).transform(this.onErrorMapMonoWithRequestInfo(request)); }).transform(this.onErrorMapMonoWithRequestInfo("exists", request));
} }
@Override @Override
@ -397,7 +398,7 @@ public class GrpcServer extends Server {
); );
return OpenIteratorResponse.newBuilder().setIteratorId(iteratorId).build(); return OpenIteratorResponse.newBuilder().setIteratorId(iteratorId).build();
} }
}).transform(this.onErrorMapMonoWithRequestInfo(request)); }).transform(this.onErrorMapMonoWithRequestInfo("openIterator", request));
} }
@Override @Override
@ -405,7 +406,7 @@ public class GrpcServer extends Server {
return executeSync(() -> { return executeSync(() -> {
api.closeIterator(request.getIteratorId()); api.closeIterator(request.getIteratorId());
return Empty.getDefaultInstance(); return Empty.getDefaultInstance();
}).transform(this.onErrorMapMonoWithRequestInfo(request)); }).transform(this.onErrorMapMonoWithRequestInfo("closeIterator", request));
} }
@Override @Override
@ -415,7 +416,7 @@ public class GrpcServer extends Server {
api.seekTo(arena, request.getIterationId(), mapKeys(arena, request.getKeysCount(), request::getKeys)); api.seekTo(arena, request.getIterationId(), mapKeys(arena, request.getKeysCount(), request::getKeys));
} }
return Empty.getDefaultInstance(); return Empty.getDefaultInstance();
}).transform(this.onErrorMapMonoWithRequestInfo(request)); }).transform(this.onErrorMapMonoWithRequestInfo("seekTo", request));
} }
@Override @Override
@ -428,7 +429,7 @@ public class GrpcServer extends Server {
new RequestNothing<>()); new RequestNothing<>());
} }
return Empty.getDefaultInstance(); return Empty.getDefaultInstance();
}).transform(this.onErrorMapMonoWithRequestInfo(request)); }).transform(this.onErrorMapMonoWithRequestInfo("subsequent", request));
} }
@Override @Override
@ -441,7 +442,7 @@ public class GrpcServer extends Server {
new RequestExists<>()); new RequestExists<>());
return PreviousPresence.newBuilder().setPresent(exists).build(); return PreviousPresence.newBuilder().setPresent(exists).build();
} }
}).transform(this.onErrorMapMonoWithRequestInfo(request)); }).transform(this.onErrorMapMonoWithRequestInfo("subsequentExists", request));
} }
@Override @Override
@ -469,7 +470,7 @@ public class GrpcServer extends Server {
} }
} }
emitter.complete(); emitter.complete();
}, FluxSink.OverflowStrategy.BUFFER).transform(this.onErrorMapFluxWithRequestInfo(request)); }, FluxSink.OverflowStrategy.BUFFER).transform(this.onErrorMapFluxWithRequestInfo("subsequentMultiGet", request));
} }
@Override @Override
@ -490,7 +491,7 @@ public class GrpcServer extends Server {
resultBuilder.setLast(unmapKV(firstAndLast.last())); resultBuilder.setLast(unmapKV(firstAndLast.last()));
} }
return resultBuilder.build(); return resultBuilder.build();
}).transform(this.onErrorMapMonoWithRequestInfo(request)); }).transform(this.onErrorMapMonoWithRequestInfo("reduceRangeFirstAndLast", request));
} }
@Override @Override
@ -509,7 +510,7 @@ public class GrpcServer extends Server {
); );
return EntriesCount.newBuilder().setCount(entriesCount).build(); return EntriesCount.newBuilder().setCount(entriesCount).build();
} }
}).transform(this.onErrorMapMonoWithRequestInfo(request)); }).transform(this.onErrorMapMonoWithRequestInfo("reduceRangeEntriesCount", request));
} }
@Override @Override
@ -526,7 +527,7 @@ public class GrpcServer extends Server {
request.getTimeoutMs() request.getTimeoutMs()
)) ))
.map(GrpcServerImpl::unmapKV) .map(GrpcServerImpl::unmapKV)
.transform(this.onErrorMapFluxWithRequestInfo(request)); .transform(this.onErrorMapFluxWithRequestInfo("getAllInRange", request));
} }
@ -538,7 +539,7 @@ public class GrpcServer extends Server {
// mappers // mappers
private <T> Function<Flux<T>, Flux<T>> onErrorMapFluxWithRequestInfo(Message request) { private <T> Function<Flux<T>, Flux<T>> onErrorMapFluxWithRequestInfo(String requestName, Message request) {
return flux -> flux.onErrorResume(throwable -> { return flux -> flux.onErrorResume(throwable -> {
var ex = handleError(throwable).asException(); var ex = handleError(throwable).asException();
if (ex.getStatus().getCode() == Code.INTERNAL) { if (ex.getStatus().getCode() == Code.INTERNAL) {
@ -548,11 +549,12 @@ public class GrpcServer extends Server {
}); });
} }
private <T> Function<Mono<T>, Mono<T>> onErrorMapMonoWithRequestInfo(Message request) { private <T> Function<Mono<T>, Mono<T>> onErrorMapMonoWithRequestInfo(String requestName, Message request) {
return flux -> flux.onErrorResume(throwable -> { return flux -> flux.onErrorResume(throwable -> {
var ex = handleError(throwable).asException(); var ex = handleError(throwable).asException();
if (ex.getStatus().getCode() == Code.INTERNAL) { if (ex.getStatus().getCode() == Code.INTERNAL) {
LOG.error("Unexpected internal error during request: {}", request, ex); LOG.error("Unexpected internal error during request \"{}\": {}", requestName, request.toString(), ex);
return Mono.error(RocksDBException.of(RocksDBErrorType.INTERNAL_ERROR, ex.getCause()));
} }
return Mono.error(ex); return Mono.error(ex);
}); });
@ -561,7 +563,7 @@ public class GrpcServer extends Server {
@Override @Override
protected Throwable onErrorMap(Throwable throwable) { protected Throwable onErrorMap(Throwable throwable) {
var ex = handleError(throwable).asException(); var ex = handleError(throwable).asException();
if (ex.getStatus().getCode() == Code.INTERNAL) { if (ex.getStatus().getCode() == Code.INTERNAL && !(throwable instanceof RocksDBException)) {
LOG.error("Unexpected internal error during request", ex); LOG.error("Unexpected internal error during request", ex);
} }
return ex; return ex;

1
src/native/java/module-info.java
View File

@ -40,4 +40,5 @@ module rockserver.core {
opens it.cavallium.rockserver.core.config to org.github.gestalt.core, org.github.gestalt.hocon; opens it.cavallium.rockserver.core.config to org.github.gestalt.core, org.github.gestalt.hocon;
exports it.cavallium.rockserver.core.impl.rocksdb; exports it.cavallium.rockserver.core.impl.rocksdb;
exports it.cavallium.rockserver.core.impl; exports it.cavallium.rockserver.core.impl;
exports it.cavallium.rockserver.core.common.api.proto to protobuf.java;
} }