Skip to content
Draft
Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
16 commits
Select commit Hold shift + click to select a range
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
Original file line number Diff line number Diff line change
Expand Up @@ -28,7 +28,10 @@
public class KpiSnapshotLedger {

private static final String SNAPSHOT = "SNAPSHOT";
private static final String TERMINAL_OUTCOMES_RATE_VERSION = "terminal-outcomes-v1";
private static final String NULL_FIELD = "-";
private static final int LEGACY_FIELD_COUNT = 12;
private static final int CURRENT_FIELD_COUNT = 13;
private static final Base64.Encoder ENCODER = Base64.getUrlEncoder().withoutPadding();
private static final Base64.Decoder DECODER = Base64.getUrlDecoder();

Expand Down Expand Up @@ -76,7 +79,7 @@ public synchronized void recordSnapshot(TenantContext tenantContext, KpiSnapshot
snapshot.succeededJobs(),
snapshot.failedJobs(),
snapshot.deadLetteredJobs(),
snapshot.conversionSuccessRate(),
terminalSuccessRate(snapshot.succeededJobs(), snapshot.failedJobs()),
snapshot.p95TimeToPreviewMs()
);
snapshots.add(record);
Expand Down Expand Up @@ -110,24 +113,66 @@ private void load() {

private void replayLine(String line) {
String[] fields = line.split("\t", -1);
if (fields.length != 12 || !SNAPSHOT.equals(fields[0])) {
throw invalidLine();
if (fields.length == LEGACY_FIELD_COUNT && SNAPSHOT.equals(fields[0])) {
replayLegacySnapshot(fields);
return;
}
if (fields.length == CURRENT_FIELD_COUNT
&& SNAPSHOT.equals(fields[0])
&& TERMINAL_OUTCOMES_RATE_VERSION.equals(fields[1])) {
replayCurrentSnapshot(fields);
return;
}
throw invalidLine();
}

private void replayLegacySnapshot(String[] fields) {
int totalJobs = integer(fields[4]);
int submittedJobs = integer(fields[5]);
int processingJobs = integer(fields[6]);
int succeededJobs = integer(fields[7]);
int failedJobs = integer(fields[8]);
int deadLetteredJobs = integer(fields[9]);
requireRate(rate(fields[10]), totalSuccessRate(totalJobs, succeededJobs));
snapshots.add(new KpiSnapshotRecord(
requiredValue(fields[1]),
requiredValue(fields[2]),
instant(fields[3]),
integer(fields[4]),
integer(fields[5]),
integer(fields[6]),
integer(fields[7]),
integer(fields[8]),
integer(fields[9]),
rate(fields[10]),
totalJobs,
submittedJobs,
processingJobs,
succeededJobs,
failedJobs,
deadLetteredJobs,
terminalSuccessRate(succeededJobs, failedJobs),
nullableLong(fields[11])
));
}

private void replayCurrentSnapshot(String[] fields) {
int totalJobs = integer(fields[5]);
int submittedJobs = integer(fields[6]);
int processingJobs = integer(fields[7]);
int succeededJobs = integer(fields[8]);
int failedJobs = integer(fields[9]);
int deadLetteredJobs = integer(fields[10]);
double currentRate = terminalSuccessRate(succeededJobs, failedJobs);
requireRate(rate(fields[11]), currentRate);
snapshots.add(new KpiSnapshotRecord(
requiredValue(fields[2]),
requiredValue(fields[3]),
instant(fields[4]),
totalJobs,
submittedJobs,
processingJobs,
succeededJobs,
failedJobs,
deadLetteredJobs,
currentRate,
nullableLong(fields[12])
));
}

private void appendLine(String line) {
if (ledgerPath == null) {
return;
Expand All @@ -150,6 +195,7 @@ private void appendLine(String line) {
private static String serialize(KpiSnapshotRecord record) {
return String.join("\t",
SNAPSHOT,
TERMINAL_OUTCOMES_RATE_VERSION,
field(record.tenantId()),
field(record.subjectId()),
field(record.exportedAt()),
Expand All @@ -164,6 +210,21 @@ private static String serialize(KpiSnapshotRecord record) {
);
}

private static double totalSuccessRate(int totalJobs, int succeededJobs) {
return totalJobs == 0 ? 0.0 : (double) succeededJobs / totalJobs;
}

private static double terminalSuccessRate(int succeededJobs, int failedJobs) {
int terminalJobs = succeededJobs + failedJobs;
return terminalJobs == 0 ? 0.0 : (double) succeededJobs / terminalJobs;
}

private static void requireRate(double storedRate, double expectedRate) {
if (Double.compare(storedRate, expectedRate) != 0) {
throw invalidLine();
}
}

private static String field(String value) {
return ENCODER.encodeToString(value.getBytes(StandardCharsets.UTF_8));
}
Expand Down Expand Up @@ -213,7 +274,11 @@ private static int integer(String field) {

private static double rate(String field) {
try {
return Double.parseDouble(field);
double parsed = Double.parseDouble(field);
if (!Double.isFinite(parsed) || parsed < 0.0 || parsed > 1.0) {
throw invalidLine();
}
return parsed;
} catch (NumberFormatException ex) {
throw invalidLine(ex);
}
Expand All @@ -224,7 +289,11 @@ private static Long nullableLong(String field) {
return null;
}
try {
return Long.parseLong(field);
long parsed = Long.parseLong(field);
if (parsed < 0L) {
throw invalidLine();
}
return parsed;
} catch (NumberFormatException ex) {
throw invalidLine(ex);
}
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -14,7 +14,7 @@
* @param succeededJobs succeeded jobs in the snapshot
* @param failedJobs failed jobs in the snapshot
* @param deadLetteredJobs dead-lettered jobs in the snapshot
* @param conversionSuccessRate succeeded jobs divided by total jobs
* @param conversionSuccessRate succeeded jobs divided by terminal succeeded/failed jobs
* @param p95TimeToPreviewMs p95 time to preview, when available
*/
public record KpiSnapshotRecord(
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -15,7 +15,7 @@
* @param succeededJobs succeeded jobs in the exported snapshot
* @param failedJobs failed jobs in the exported snapshot
* @param deadLetteredJobs dead-lettered jobs in the exported snapshot
* @param conversionSuccessRate succeeded jobs divided by total jobs
* @param conversionSuccessRate succeeded jobs divided by terminal succeeded/failed jobs
* @param p95TimeToPreviewMs p95 time to preview, when available
*/
public record KpiSnapshotExportResponse(
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -17,7 +17,7 @@
* @param succeededJobs jobs with a ready preview artifact
* @param failedJobs jobs in failed state
* @param deadLetteredJobs failed jobs that exhausted retry handling
* @param conversionSuccessRate succeeded jobs divided by total jobs
* @param conversionSuccessRate succeeded jobs divided by terminal succeeded/failed jobs
* @param p95TimeToPreviewMs p95 processing time for succeeded jobs with timestamps
*/
public record KpiSnapshotResponse(
Expand Down Expand Up @@ -64,7 +64,8 @@ public static KpiSnapshotResponse from(List<ConversionJob> jobs) {
}

int total = jobs.size();
double successRate = total == 0 ? 0.0 : (double) succeeded / total;
int terminal = succeeded + failed;
double successRate = terminal == 0 ? 0.0 : (double) succeeded / terminal;
Comment thread
coderabbitai[bot] marked this conversation as resolved.
Resolved
return new KpiSnapshotResponse(
total,
submitted,
Expand Down
Loading
Loading