Skip to content

feat: Add session closed metric #700

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
KirillKurdyukov wants to merge 1 commit into
base: master
Choose a base branch
from
Open

feat: Add session closed metric #700

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion core/src/main/java/tech/ydb/core/impl/Observability.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -12,7 +12,7 @@
*/
public final class Observability {
public static final String TRACING_CHAIN = ";ydb-sdk-tracing/0.1.0";
public static final String METRICS_CHAIN = ";ydb-sdk-metrics/0.1.0";
public static final String METRICS_CHAIN = ";ydb-sdk-metrics/0.2.0";

private static volatile boolean isTracingEnabled = false;
private static volatile boolean isMetricsEnabled = false;
Expand Down
8 changes: 4 additions & 4 deletions core/src/test/java/tech/ydb/core/impl/ObservabilityTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -31,19 +31,19 @@ public void baseTest() {
Assert.assertEquals(BASE, Observability.getDiscoveryBuildInfo(BASE));

Observability.reportMetricsUsage(new Meter() { });
Assert.assertEquals(BASE + ";ydb-sdk-metrics/0.1.0", Observability.getDiscoveryBuildInfo(BASE));
Assert.assertEquals(BASE + ";ydb-sdk-metrics/0.2.0", Observability.getDiscoveryBuildInfo(BASE));

Observability.reportTracingUsage((String spanName, SpanKind spanKind) -> Span.NOOP);
Assert.assertEquals(
BASE + ";ydb-sdk-tracing/0.1.0;ydb-sdk-metrics/0.1.0",
BASE + ";ydb-sdk-tracing/0.1.0;ydb-sdk-metrics/0.2.0",
Observability.getDiscoveryBuildInfo(BASE)
);

Observability.reportMetricsUsage(Meter.NOOP);
Observability.reportTracingUsage(NoopTracer.getInstance());
Assert.assertEquals(
BASE + ";ydb-sdk-tracing/0.1.0;ydb-sdk-metrics/0.1.0",
BASE + ";ydb-sdk-tracing/0.1.0;ydb-sdk-metrics/0.2.0",
Observability.getDiscoveryBuildInfo(BASE)
);
}
}
}
6 changes: 4 additions & 2 deletions query/src/main/java/tech/ydb/query/impl/SessionImpl.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -120,6 +120,8 @@ public QueryTransaction createNewTransaction(TxMode txMode) {

public abstract void updateSessionState(Status status);

abstract void closeSession(String reason);

@Override
public CompletableFuture<Result<QueryTransaction>> beginTransaction(TxMode tx, BeginTransactionSettings settings) {
YdbQuery.BeginTransactionRequest request = YdbQuery.BeginTransactionRequest.newBuilder()
Expand Down Expand Up @@ -175,10 +177,10 @@ public CompletableFuture<Status> start(GrpcReadStream.Observer<Status> observer)
switch (message.getSessionHintCase()) {
case NODE_SHUTDOWN:
pessimizationHook.set(nodeID != 0);
updateSessionState(Status.of(StatusCode.BAD_SESSION));
closeSession("node_shutdown");
break;
case SESSION_SHUTDOWN:
updateSessionState(Status.of(StatusCode.BAD_SESSION));
closeSession("session_shutdown");
break;
default:
break;
Expand Down
66 changes: 51 additions & 15 deletions query/src/main/java/tech/ydb/query/impl/SessionPool.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -9,6 +9,7 @@
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.LongAdder;
import java.util.function.BiConsumer;

Expand Down Expand Up @@ -163,10 +164,10 @@ private boolean tryComplete(CompletableFuture<Result<QuerySession>> future, Pool

private class PooledQuerySession extends SessionImpl {
private final GrpcReadStream<Status> attachStream;
private final AtomicBoolean isBroken = new AtomicBoolean();

private volatile Instant lastActive;
private volatile boolean isStarted = false;
private volatile boolean isBroken = false;
private volatile boolean isStopped = false;

PooledQuerySession(QueryServiceRpc rpc, YdbQuery.CreateSessionResponse response) {
Expand All @@ -179,18 +180,42 @@ private class PooledQuerySession extends SessionImpl {
@Override
public void updateSessionState(Status status) {
this.lastActive = clock.instant();
boolean isStatusBroken =
status.getCode() == StatusCode.BAD_SESSION ||
status.getCode() == StatusCode.SESSION_BUSY ||
status.getCode() == StatusCode.INTERNAL_ERROR ||
status.getCode() == StatusCode.CLIENT_DEADLINE_EXCEEDED ||
status.getCode() == StatusCode.CLIENT_DEADLINE_EXPIRED ||
status.getCode() == StatusCode.CLIENT_CANCELLED ||
status.getCode() == StatusCode.TRANSPORT_UNAVAILABLE;
if (isStatusBroken) {
logger.warn("QuerySession[{}] broken with status {}", getId(), status);
if (!isStarted) {
return;
}

String reason;
switch (status.getCode()) {
case BAD_SESSION:
case SESSION_EXPIRED:
reason = "bad_session";
break;
case SESSION_BUSY:
reason = "session_busy";
break;
case CLIENT_DEADLINE_EXCEEDED:
case CLIENT_DEADLINE_EXPIRED:
reason = "client_timeout";
break;
case CLIENT_CANCELLED:
reason = "client_cancelled";
break;
case TRANSPORT_UNAVAILABLE:
reason = "transport_error";
break;
default:
return;
}

logger.warn("QuerySession[{}] broken with status {}", getId(), status);
closeSession(reason);
}

@Override
void closeSession(String reason) {
if (isBroken.compareAndSet(false, true)) {
metrics.onSessionClosed(reason);
}
isBroken = isBroken || isStatusBroken;
}

public Instant getLastActive() {
Expand All @@ -217,6 +242,12 @@ public CompletableFuture<Result<PooledQuerySession>> start() {

logger.trace("QuerySession[{}] attach message {}", getId(), status);
}).whenComplete((status, th) -> {
if (isStarted && !isStopped) {
closeSession(status != null && status.isSuccess()
? "attach_closed"
: "transport_error");
}

if (th != null) {
logger.debug("QuerySession[{}] finished with exception", getId(), th);
}
Expand Down Expand Up @@ -260,11 +291,11 @@ public void destroy() {

@Override
public void close() {
logger.trace("QuerySession[{}] closed with broken status {}", getId(), isBroken);
logger.trace("QuerySession[{}] closed with broken status {}", getId(), isBroken.get());

stats.released.increment();
metrics.onSessionReleased();
if (isBroken || isStopped) {
if (isBroken.get() || isStopped) {
queue.delete(this);
} else {
queue.release(this);
Expand Down Expand Up @@ -313,10 +344,15 @@ public CompletableFuture<PooledQuerySession> create() {
}
}

@Override
public void destroy(PooledQuerySession session, WaitingQueue.RemovalReason reason) {
session.closeSession(reason.toString());
destroy(session);
}

@Override
public void destroy(PooledQuerySession session) {
stats.deleted.increment();
metrics.onSessionDeleted();

// Execute deleteSession call outside current context to avoid cancellation and deadline propogation
Context ctx = Context.ROOT.fork();
Expand Down
Loading