Skip to content

Commit 82d0cc7

Browse files
Make console logging async and memoize .env lookups on the request hot path
Both were found during a round-trip latency audit of the SDK's request path (same audit that flagged the retry-interceptor duplication this branch fixes). - LogUtil's ConsoleHandler writes and flushes to the console synchronously, under a lock shared by every calling thread. LogUtil.printInfoLog/etc. fire several times per request (setBearerToken reuse/expiry, request validation, per-batch triggers, request-resolved), so enabling INFO/DEBUG logging turned console output into a per-request blocking-I/O + contention point. A new AsyncConsoleHandler now hands each LogRecord to a single background daemon thread via a bounded, non-blocking queue; the calling thread only enqueues. Applied to both the common/flowvault LogUtil and skyvault's separate copy. - Several settings/credentials lookups fell back to Dotenv.load() when an env var wasn't set: VaultController.resolveSettingFromEnvironment did this twice per bulk call (batch size + concurrency limit), unconditionally, on every single insert/detokenize/tokenize/deleteTokens call -- Dotenv.load() re-reads the .env file from disk every time it's invoked, so this was uncached, blocking disk I/O on the hot path. The same pattern existed in BaseVaultClient/ConnectionClient's credential fallback and Utils.getEnvVaultUrl. BaseUtils.resolveEnvOrDotenv now memoizes the loaded (or absent) .env for the life of the JVM -- a project's .env doesn't change while the process runs, so there's no reason to keep re-reading it -- with a resetDotenvCacheForTests() escape hatch for the handful of tests that intentionally rewrite .env mid-run to exercise both branches. Testing: full common + skyvault + flowvault suites, 1450/1450 passing (165+575+710), BUILD SUCCESS. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
1 parent 8746ebd commit 82d0cc7

12 files changed

Lines changed: 237 additions & 29 deletions

File tree

‎common/src/main/java/com/skyflow/BaseVaultClient.java‎

Lines changed: 1 addition & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -14,7 +14,6 @@
1414
import com.skyflow.utils.BaseUtils;
1515
import com.skyflow.utils.logger.LogUtil;
1616
import com.skyflow.utils.validations.BaseValidations;
17-
import io.github.cdimascio.dotenv.Dotenv;
1817
import io.github.cdimascio.dotenv.DotenvException;
1918
import okhttp3.ConnectionPool;
2019
import okhttp3.OkHttpClient;
@@ -63,11 +62,7 @@ protected synchronized void prioritiseCredentials(BaseCredentials vaultSpecificC
6362
} else if (this.commonCredentials != null) {
6463
this.finalCredentials = this.commonCredentials;
6564
} else {
66-
String sysCredentials = System.getenv(BaseConstants.ENV_CREDENTIALS_KEY_NAME);
67-
if (sysCredentials == null) {
68-
Dotenv dotenv = Dotenv.load();
69-
sysCredentials = dotenv.get(BaseConstants.ENV_CREDENTIALS_KEY_NAME);
70-
}
65+
String sysCredentials = BaseUtils.resolveEnvOrDotenv(BaseConstants.ENV_CREDENTIALS_KEY_NAME);
7166
if (sysCredentials == null) {
7267
throw new SkyflowException(ErrorCode.INVALID_INPUT.getCode(), ErrorMessage.EmptyCredentials.getMessage());
7368
} else {

‎common/src/main/java/com/skyflow/utils/BaseUtils.java‎

Lines changed: 55 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -21,8 +21,63 @@
2121
import com.skyflow.logs.InfoLogs;
2222
import com.skyflow.serviceaccount.util.BearerToken;
2323
import com.skyflow.utils.logger.LogUtil;
24+
import io.github.cdimascio.dotenv.Dotenv;
25+
import io.github.cdimascio.dotenv.DotenvException;
2426

2527
public class BaseUtils {
28+
29+
// Memoized .env: Dotenv.load() does a filesystem read every time it's called, and several
30+
// call sites resolve a setting this way on every single SDK request -- re-reading a file
31+
// whose contents never change for the life of the process turned those into repeated,
32+
// uncached, blocking disk I/O on the hot path. Loaded at most once per JVM; `dotenvAttempted`
33+
// also memoizes the "no .env file present" outcome so a missing file isn't retried either.
34+
private static volatile boolean dotenvAttempted = false;
35+
private static volatile Dotenv cachedDotenv = null;
36+
37+
private static Dotenv memoizedDotenv() {
38+
if (!dotenvAttempted) {
39+
synchronized (BaseUtils.class) {
40+
if (!dotenvAttempted) {
41+
try {
42+
cachedDotenv = Dotenv.load();
43+
} catch (DotenvException e) {
44+
cachedDotenv = null; // no .env file in the working directory
45+
}
46+
dotenvAttempted = true;
47+
}
48+
}
49+
}
50+
return cachedDotenv;
51+
}
52+
53+
/**
54+
* Resolves {@code key} from the process environment first, falling back to the (memoized)
55+
* {@code .env} file if present. Returns null if found in neither.
56+
*/
57+
public static String resolveEnvOrDotenv(String key) {
58+
String value = System.getenv(key);
59+
if (value == null) {
60+
Dotenv dotenv = memoizedDotenv();
61+
if (dotenv != null) {
62+
value = dotenv.get(key);
63+
}
64+
}
65+
return value;
66+
}
67+
68+
/**
69+
* Test-only: forces the next {@link #resolveEnvOrDotenv} call to re-read the {@code .env}
70+
* file from disk instead of reusing the memoized one. Production code always wants the
71+
* memoized behavior (a project's {@code .env} doesn't change while the process is running);
72+
* this exists purely so tests that rewrite {@code .env} mid-run can observe the new content
73+
* without restarting the JVM.
74+
*/
75+
public static void resetDotenvCacheForTests() {
76+
synchronized (BaseUtils.class) {
77+
dotenvAttempted = false;
78+
cachedDotenv = null;
79+
}
80+
}
2681
public static String generateBearerToken(BaseCredentials credentials) throws SkyflowException {
2782
if (credentials.getPath() != null) {
2883
BearerToken.BearerTokenBuilder builder = BearerToken.builder()
Lines changed: 82 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,82 @@
1+
package com.skyflow.utils.logger;
2+
3+
import java.util.concurrent.LinkedBlockingQueue;
4+
import java.util.logging.ErrorManager;
5+
import java.util.logging.Handler;
6+
import java.util.logging.LogRecord;
7+
8+
/**
9+
* Wraps a delegate {@link Handler} (in practice a {@link java.util.logging.ConsoleHandler}) so that
10+
* {@link #publish(LogRecord)} never performs blocking I/O on the calling thread.
11+
* <p>
12+
* {@code ConsoleHandler.publish} writes to and flushes the underlying stream synchronously, and does
13+
* so under a lock shared by every thread using the logger. Since {@code LogUtil.printInfoLog}/etc. are
14+
* called on every SDK request (often several times per call), that turns console logging into a
15+
* per-request blocking-I/O + lock-contention point under concurrent load. This handler hands each
16+
* {@link LogRecord} off to a single background daemon thread instead, which performs the actual write;
17+
* the calling thread only enqueues.
18+
* <p>
19+
* The handoff never blocks or applies backpressure to the caller: if the queue is momentarily full
20+
* (a sustained logging flood, or the writer thread stalled) the record is dropped rather than slowing
21+
* down request-serving threads — logging must never become the bottleneck it was flagged for.
22+
*/
23+
final class AsyncConsoleHandler extends Handler {
24+
25+
/** Bounds worst-case memory use if the writer thread falls behind; excess records are dropped. */
26+
private static final int QUEUE_CAPACITY = 10_000;
27+
28+
private final Handler delegate;
29+
private final LinkedBlockingQueue<LogRecord> queue = new LinkedBlockingQueue<>(QUEUE_CAPACITY);
30+
private final Thread writer;
31+
private volatile boolean closed = false;
32+
33+
AsyncConsoleHandler(Handler delegate) {
34+
this.delegate = delegate;
35+
setLevel(delegate.getLevel());
36+
this.writer = new Thread(this::drain, "skyflow-sdk-log-writer");
37+
this.writer.setDaemon(true);
38+
this.writer.start();
39+
}
40+
41+
@Override
42+
public void publish(LogRecord record) {
43+
if (closed || !isLoggable(record)) {
44+
return;
45+
}
46+
// offer() never blocks: a full queue means "drop", never "wait".
47+
queue.offer(record);
48+
}
49+
50+
private void drain() {
51+
try {
52+
while (true) {
53+
LogRecord record = queue.take();
54+
try {
55+
delegate.publish(record);
56+
} catch (RuntimeException e) {
57+
reportError(null, e, ErrorManager.WRITE_FAILURE);
58+
}
59+
}
60+
} catch (InterruptedException e) {
61+
Thread.currentThread().interrupt();
62+
}
63+
}
64+
65+
@Override
66+
public void flush() {
67+
// Best-effort: drain what's queued right now onto the delegate, then flush it.
68+
LogRecord record;
69+
while ((record = queue.poll()) != null) {
70+
delegate.publish(record);
71+
}
72+
delegate.flush();
73+
}
74+
75+
@Override
76+
public void close() {
77+
closed = true;
78+
writer.interrupt();
79+
flush();
80+
delegate.close();
81+
}
82+
}

‎common/src/main/java/com/skyflow/utils/logger/LogUtil.java‎

Lines changed: 4 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -35,7 +35,10 @@ public synchronized String format(LogRecord logRecord) {
3535
consoleHandler.setFormatter(formatter);
3636
consoleHandler.setLevel(Level.CONFIG);
3737

38-
LOGGER.addHandler(consoleHandler);
38+
// The actual write+flush to the console happens on a background thread, so per-call
39+
// logging never blocks (or lock-contends on) the request-serving thread. See
40+
// AsyncConsoleHandler's class doc for why this matters under concurrent load.
41+
LOGGER.addHandler(new AsyncConsoleHandler(consoleHandler));
3942
LOGGER.setLevel(logLevelToLoggerLevelMap(logLevel));
4043
printInfoLog(InfoLogs.LOGGER_SETUP_DONE.getLog());
4144
}

‎common/src/test/java/com/skyflow/BaseVaultClientTests.java‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -6,6 +6,7 @@
66
import com.skyflow.errors.SkyflowException;
77
import com.skyflow.logs.ErrorLogs;
88
import com.skyflow.utils.BaseConstants;
9+
import com.skyflow.utils.BaseUtils;
910
import okhttp3.Call;
1011
import okhttp3.Connection;
1112
import okhttp3.Interceptor;
@@ -38,6 +39,7 @@ public class BaseVaultClientTests {
3839
public void saveEnvFileState() throws IOException {
3940
File f = new File(ENV_FILE);
4041
originalEnvContent = f.exists() ? Files.readAllBytes(Paths.get(ENV_FILE)) : null;
42+
BaseUtils.resetDotenvCacheForTests(); // see its javadoc: .env is otherwise memoized JVM-wide
4143
}
4244

4345
@After

‎flowvault/src/main/java/com/skyflow/utils/Utils.java‎

Lines changed: 1 addition & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -52,7 +52,6 @@
5252
import com.skyflow.vault.data.TokenGroupRedactions;
5353
import com.skyflow.vault.data.UpsertOptions;
5454

55-
import io.github.cdimascio.dotenv.Dotenv;
5655
import io.github.cdimascio.dotenv.DotenvException;
5756

5857
import java.io.File;
@@ -82,11 +81,7 @@ public static JsonObject getMetrics() {
8281

8382
public static String getEnvVaultUrl() throws SkyflowException {
8483
try {
85-
String vaultUrl = System.getenv("VAULT_URL");
86-
if (vaultUrl == null) {
87-
Dotenv dotenv = Dotenv.load();
88-
vaultUrl = dotenv.get("VAULT_URL");
89-
}
84+
String vaultUrl = resolveEnvOrDotenv("VAULT_URL");
9085
if (vaultUrl != null && vaultUrl.trim().isEmpty()) {
9186
LogUtil.printErrorLog(ErrorLogs.EMPTY_VAULT_URL.getLog());
9287
throw new SkyflowException(ErrorCode.INVALID_INPUT.getCode(), ErrorMessage.EmptyVaultUrl.getMessage());

‎flowvault/src/main/java/com/skyflow/vault/controller/VaultController.java‎

Lines changed: 1 addition & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -58,9 +58,6 @@
5858
import com.skyflow.vault.data.RequestInterceptor;
5959
import com.skyflow.vault.data.TokenizeOptions;
6060

61-
import io.github.cdimascio.dotenv.Dotenv;
62-
import io.github.cdimascio.dotenv.DotenvException;
63-
6461
public final class VaultController extends VaultClient {
6562
private static final Gson gson = new GsonBuilder().serializeNulls().create();
6663
private JsonObject metrics = Utils.getMetrics();
@@ -507,15 +504,7 @@ private ApiClientHttpResponse<V1FlowDeleteTokenResponse> processDeleteTokensBatc
507504
static Function<String, String> settingResolver = VaultController::resolveSettingFromEnvironment;
508505

509506
private static String resolveSettingFromEnvironment(String key) {
510-
String value = System.getenv(key);
511-
if (value == null) {
512-
try {
513-
value = Dotenv.load().get(key);
514-
} catch (DotenvException ignored) {
515-
// no .env available — environment-only
516-
}
517-
}
518-
return value;
507+
return Utils.resolveEnvOrDotenv(key);
519508
}
520509

521510
private BatchConfig configureDeleteTokensConcurrencyAndBatchSize(int totalRequests) {

‎flowvault/src/test/java/com/skyflow/utils/UtilsTests.java‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -68,6 +68,7 @@ public class UtilsTests {
6868
public void saveEnvFileState() throws IOException {
6969
File f = new File(ENV_FILE);
7070
originalEnvContent = f.exists() ? Files.readAllBytes(Paths.get(ENV_FILE)) : null;
71+
BaseUtils.resetDotenvCacheForTests(); // see its javadoc: .env is otherwise memoized JVM-wide
7172
}
7273

7374
@After

‎skyvault/src/main/java/com/skyflow/ConnectionClient.java‎

Lines changed: 2 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -7,11 +7,11 @@
77
import com.skyflow.errors.SkyflowException;
88
import com.skyflow.logs.InfoLogs;
99
import com.skyflow.serviceaccount.util.Token;
10+
import com.skyflow.utils.BaseUtils;
1011
import com.skyflow.utils.Constants;
1112
import com.skyflow.utils.Utils;
1213
import com.skyflow.utils.logger.LogUtil;
1314
import com.skyflow.utils.validations.Validations;
14-
import io.github.cdimascio.dotenv.Dotenv;
1515
import io.github.cdimascio.dotenv.DotenvException;
1616

1717
public class ConnectionClient {
@@ -69,8 +69,7 @@ private void prioritiseCredentials() throws SkyflowException {
6969
} else if (this.commonCredentials != null) {
7070
this.finalCredentials = this.commonCredentials;
7171
} else {
72-
Dotenv dotenv = Dotenv.load();
73-
String sysCredentials = dotenv.get(Constants.ENV_CREDENTIALS_KEY_NAME);
72+
String sysCredentials = BaseUtils.resolveEnvOrDotenv(Constants.ENV_CREDENTIALS_KEY_NAME);
7473
if (sysCredentials == null) {
7574
throw new SkyflowException(ErrorCode.INVALID_INPUT.getCode(),
7675
ErrorMessage.EmptyCredentials.getMessage());
Lines changed: 82 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,82 @@
1+
package com.skyflow.utils.logger;
2+
3+
import java.util.concurrent.LinkedBlockingQueue;
4+
import java.util.logging.ErrorManager;
5+
import java.util.logging.Handler;
6+
import java.util.logging.LogRecord;
7+
8+
/**
9+
* Wraps a delegate {@link Handler} (in practice a {@link java.util.logging.ConsoleHandler}) so that
10+
* {@link #publish(LogRecord)} never performs blocking I/O on the calling thread.
11+
* <p>
12+
* {@code ConsoleHandler.publish} writes to and flushes the underlying stream synchronously, and does
13+
* so under a lock shared by every thread using the logger. Since {@code LogUtil.printInfoLog}/etc. are
14+
* called on every SDK request (often several times per call), that turns console logging into a
15+
* per-request blocking-I/O + lock-contention point under concurrent load. This handler hands each
16+
* {@link LogRecord} off to a single background daemon thread instead, which performs the actual write;
17+
* the calling thread only enqueues.
18+
* <p>
19+
* The handoff never blocks or applies backpressure to the caller: if the queue is momentarily full
20+
* (a sustained logging flood, or the writer thread stalled) the record is dropped rather than slowing
21+
* down request-serving threads — logging must never become the bottleneck it was flagged for.
22+
*/
23+
final class AsyncConsoleHandler extends Handler {
24+
25+
/** Bounds worst-case memory use if the writer thread falls behind; excess records are dropped. */
26+
private static final int QUEUE_CAPACITY = 10_000;
27+
28+
private final Handler delegate;
29+
private final LinkedBlockingQueue<LogRecord> queue = new LinkedBlockingQueue<>(QUEUE_CAPACITY);
30+
private final Thread writer;
31+
private volatile boolean closed = false;
32+
33+
AsyncConsoleHandler(Handler delegate) {
34+
this.delegate = delegate;
35+
setLevel(delegate.getLevel());
36+
this.writer = new Thread(this::drain, "skyflow-sdk-log-writer");
37+
this.writer.setDaemon(true);
38+
this.writer.start();
39+
}
40+
41+
@Override
42+
public void publish(LogRecord record) {
43+
if (closed || !isLoggable(record)) {
44+
return;
45+
}
46+
// offer() never blocks: a full queue means "drop", never "wait".
47+
queue.offer(record);
48+
}
49+
50+
private void drain() {
51+
try {
52+
while (true) {
53+
LogRecord record = queue.take();
54+
try {
55+
delegate.publish(record);
56+
} catch (RuntimeException e) {
57+
reportError(null, e, ErrorManager.WRITE_FAILURE);
58+
}
59+
}
60+
} catch (InterruptedException e) {
61+
Thread.currentThread().interrupt();
62+
}
63+
}
64+
65+
@Override
66+
public void flush() {
67+
// Best-effort: drain what's queued right now onto the delegate, then flush it.
68+
LogRecord record;
69+
while ((record = queue.poll()) != null) {
70+
delegate.publish(record);
71+
}
72+
delegate.flush();
73+
}
74+
75+
@Override
76+
public void close() {
77+
closed = true;
78+
writer.interrupt();
79+
flush();
80+
delegate.close();
81+
}
82+
}

0 commit comments

Comments
 (0)