Skip to content

Add metrics emitted from BulkExecutor responsible for dynamic batch size tuning. #43716

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

Merged
jeet1995 merged 10 commits into from
Jan 9, 2025
Merged

Add metrics emitted from BulkExecutor responsible for dynamic batch size tuning. #43716

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
10 commits
Select commit Hold shift + click to select a range
9a621a0
Comment out failing tests.
jeet1995 Dec 18, 2024
d508f8d
Merge branch 'main' of github.com:jeet1995/azure-sdk-for-java into Bu…
jeet1995 Jan 6, 2025
62ca496
Adding javadoc.
jeet1995 Dec 18, 2024
cbdec3d
Adding targetMaxMicroBatchSize as client telemetry metric.
jeet1995 Dec 18, 2024
ec620ab
Remove unnecessary public API from `CosmosBatchResponse`.
jeet1995 Jan 7, 2025
7ec7824
Remove unnecessary public API from `CosmosBatchResponse`.
jeet1995 Jan 7, 2025
7b34d01
Added javadoc and modified CHANGELOG.md
jeet1995 Jan 7, 2025
f26ab51
Added javadoc.
jeet1995 Jan 7, 2025
a3ea8c7
Fix tests.
jeet1995 Jan 8, 2025
c27114d
Document new metrics in Metrics.md
jeet1995 Jan 9, 2025
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
Prev Previous commit
Next Next commit
Remove unnecessary public API from CosmosBatchResponse.
  • Loading branch information
@jeet1995
jeet1995 committed Jan 7, 2025
commit 7ec78246b4f69bee3e32acb846995370dbf8629f
48 changes: 24 additions & 24 deletions sdk/cosmos/azure-cosmos-tests/src/test/java/com/azure/cosmos/ClientMetricsTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1345,14 +1345,14 @@ public void meterNameFromStringConversion() {
assertThat(CosmosMetricName.fromString("cosmos.client.req.rntbd.actualITemCount"))
.isSameAs(CosmosMetricName.REQUEST_SUMMARY_DIRECT_ACTUAL_ITEM_COUNT);

assertThat(CosmosMetricName.fromString("cosmos.client.req.rntbd.opCountPerEvaluation"))
.isSameAs(CosmosMetricName.REQUEST_SUMMARY_DIRECT_OP_COUNT_PER_EVALUATION);
assertThat(CosmosMetricName.fromString("cosmos.client.req.rntbd.opRetriedCountPerEvaluation"))
.isSameAs(CosmosMetricName.REQUEST_SUMMARY_DIRECT_OP_RETRIED_COUNT_PER_EVALUATION);
assertThat(CosmosMetricName.fromString("cosmos.client.req.rntbd.globalOpCount"))
.isSameAs(CosmosMetricName.REQUEST_SUMMARY_DIRECT_GLOBAL_OP_COUNT);
assertThat(CosmosMetricName.fromString("cosmos.client.req.rntbd.targetMaxMicroBatchSize"))
.isSameAs(CosmosMetricName.REQUEST_SUMMARY_DIRECT_TARGET_MAX_MICRO_BATCH_SIZE);
assertThat(CosmosMetricName.fromString("cosmos.client.req.rntbd.bulkOpCountPerEvaluation"))
.isSameAs(CosmosMetricName.REQUEST_SUMMARY_DIRECT_BULK_OP_COUNT_PER_EVALUATION);
assertThat(CosmosMetricName.fromString("cosmos.client.req.rntbd.bulkOpRetriedCountPerEvaluation"))
.isSameAs(CosmosMetricName.REQUEST_SUMMARY_DIRECT_BULK_OP_RETRIED_COUNT_PER_EVALUATION);
assertThat(CosmosMetricName.fromString("cosmos.client.req.rntbd.bulkGlobalOpCount"))
.isSameAs(CosmosMetricName.REQUEST_SUMMARY_DIRECT_BULK_GLOBAL_OP_COUNT);
assertThat(CosmosMetricName.fromString("cosmos.client.req.rntbd.bulkTargetMaxMicroBatchSize"))
.isSameAs(CosmosMetricName.REQUEST_SUMMARY_DIRECT_BULK_TARGET_MAX_MICRO_BATCH_SIZE);

assertThat(CosmosMetricName.fromString("cosmos.CLIENT.req.gw.LAtency"))
.isSameAs(CosmosMetricName.REQUEST_SUMMARY_GATEWAY_LATENCY);
Expand All @@ -1367,14 +1367,14 @@ public void meterNameFromStringConversion() {
assertThat(CosmosMetricName.fromString("cosmos.client.req.gw.actualITemCount"))
.isSameAs(CosmosMetricName.REQUEST_SUMMARY_GATEWAY_ACTUAL_ITEM_COUNT);

assertThat(CosmosMetricName.fromString("cosmos.client.req.gw.opCountPerEvaluation"))
.isSameAs(CosmosMetricName.REQUEST_SUMMARY_GATEWAY_OP_COUNT_PER_EVALUATION);
assertThat(CosmosMetricName.fromString("cosmos.client.req.gw.opRetriedCountPerEvaluation"))
.isSameAs(CosmosMetricName.REQUEST_SUMMARY_GATEWAY_OP_RETRIED_COUNT_PER_EVALUATION);
assertThat(CosmosMetricName.fromString("cosmos.client.req.gw.globalOpCount"))
.isSameAs(CosmosMetricName.REQUEST_SUMMARY_GATEWAY_GLOBAL_OP_COUNT);
assertThat(CosmosMetricName.fromString("cosmos.client.req.gw.targetMaxMicroBatchSize"))
.isSameAs(CosmosMetricName.REQUEST_SUMMARY_GATEWAY_TARGET_MAX_MICRO_BATCH_SIZE);
assertThat(CosmosMetricName.fromString("cosmos.client.req.gw.bulkOpCountPerEvaluation"))
.isSameAs(CosmosMetricName.REQUEST_SUMMARY_GATEWAY_BULK_OP_COUNT_PER_EVALUATION);
assertThat(CosmosMetricName.fromString("cosmos.client.req.gw.bulkOpRetriedCountPerEvaluation"))
.isSameAs(CosmosMetricName.REQUEST_SUMMARY_GATEWAY_BULK_OP_RETRIED_COUNT_PER_EVALUATION);
assertThat(CosmosMetricName.fromString("cosmos.client.req.gw.bulkGlobalOpCount"))
.isSameAs(CosmosMetricName.REQUEST_SUMMARY_GATEWAY_BULK_GLOBAL_OP_COUNT);
assertThat(CosmosMetricName.fromString("cosmos.client.req.gw.bulkTargetMaxMicroBatchSize"))
.isSameAs(CosmosMetricName.REQUEST_SUMMARY_GATEWAY_BULK_TARGET_MAX_MICRO_BATCH_SIZE);

assertThat(CosmosMetricName.fromString("cosmos.client.RNTBD.addressResolution.latency"))
.isSameAs(CosmosMetricName.DIRECT_ADDRESS_RESOLUTION_LATENCY);
Expand Down Expand Up @@ -1524,11 +1524,11 @@ private void validateBatchOpCountPerEvaluation(Tag... expectedRequestTags) {
if (this.getEffectiveMetricCategories().contains(MetricCategory.RequestSummary)) {
if (this.client.asyncClient().getConnectionPolicy().getConnectionMode() == ConnectionMode.DIRECT) {
for (Tag expectedRequestTag : expectedRequestTags) {
this.assertMetrics("cosmos.client.req.rntbd.opCountPerEvaluation", true, expectedRequestTag);
this.assertMetrics("cosmos.client.req.rntbd.bulkOpCountPerEvaluation", true, expectedRequestTag);
}
} else {
for (Tag expectedRequestTag : expectedRequestTags) {
this.assertMetrics("cosmos.client.req.gw.opCountPerEvaluation", true, expectedRequestTag);
this.assertMetrics("cosmos.client.req.gw.bulkOpCountPerEvaluation", true, expectedRequestTag);
}
}
}
Expand All @@ -1538,11 +1538,11 @@ private void validateBatchOpRetriedCountPerEvaluation(Tag... expectedRequestTags
if (this.getEffectiveMetricCategories().contains(MetricCategory.RequestSummary)) {
if (this.client.asyncClient().getConnectionPolicy().getConnectionMode() == ConnectionMode.DIRECT) {
for (Tag expectedRequestTag : expectedRequestTags) {
this.assertMetrics("cosmos.client.req.rntbd.opRetriedCountPerEvaluation", true, expectedRequestTag);
this.assertMetrics("cosmos.client.req.rntbd.bulkOpRetriedCountPerEvaluation", true, expectedRequestTag);
}
} else {
for (Tag expectedRequestTag : expectedRequestTags) {
this.assertMetrics("cosmos.client.req.gw.opRetriedCountPerEvaluation", true, expectedRequestTag);
this.assertMetrics("cosmos.client.req.gw.bulkOpRetriedCountPerEvaluation", true, expectedRequestTag);
}
}
}
Expand All @@ -1552,11 +1552,11 @@ private void validateBatchGlobalOpCount(Tag... expectedRequestTags) {
if (this.getEffectiveMetricCategories().contains(MetricCategory.RequestSummary)) {
if (this.client.asyncClient().getConnectionPolicy().getConnectionMode() == ConnectionMode.DIRECT) {
for (Tag expectedRequestTag : expectedRequestTags) {
this.assertMetrics("cosmos.client.req.rntbd.globalOpCount", true, expectedRequestTag);
this.assertMetrics("cosmos.client.req.rntbd.bulkGlobalOpCount", true, expectedRequestTag);
}
} else {
for (Tag expectedRequestTag : expectedRequestTags) {
this.assertMetrics("cosmos.client.req.gw.globalOpCount", true, expectedRequestTag);
this.assertMetrics("cosmos.client.req.gw.bulkGlobalOpCount", true, expectedRequestTag);
}
}
}
Expand All @@ -1566,11 +1566,11 @@ private void validateTargetMaxMicroBatchSize(Tag... expectedRequestTags) {
if (this.getEffectiveMetricCategories().contains(MetricCategory.RequestSummary)) {
if (this.client.asyncClient().getConnectionPolicy().getConnectionMode() == ConnectionMode.DIRECT) {
for (Tag expectedRequestTag : expectedRequestTags) {
this.assertMetrics("cosmos.client.req.rntbd.targetMaxMicroBatchSize", true, expectedRequestTag);
this.assertMetrics("cosmos.client.req.rntbd.bulkTargetMaxMicroBatchSize", true, expectedRequestTag);
}
} else {
for (Tag expectedRequestTag : expectedRequestTags) {
this.assertMetrics("cosmos.client.req.gw.targetMaxMicroBatchSize", true, expectedRequestTag);
this.assertMetrics("cosmos.client.req.gw.bulkTargetMaxMicroBatchSize", true, expectedRequestTag);
}
}
}
Expand Down
10 changes: 6 additions & 4 deletions ...osmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/DiagnosticsProvider.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -79,6 +79,8 @@ public final class DiagnosticsProvider {
private static final
ImplementationBridgeHelpers.CosmosDiagnosticsHelper.CosmosDiagnosticsAccessor diagnosticsAccessor =
ImplementationBridgeHelpers.CosmosDiagnosticsHelper.getCosmosDiagnosticsAccessor();
private static final ImplementationBridgeHelpers.CosmosBatchResponseHelper.CosmosBatchResponseAccessor cosmosBatchResponseAccessor
= ImplementationBridgeHelpers.CosmosBatchResponseHelper.getCosmosBatchResponseAccessor();

private static final Logger LOGGER = LoggerFactory.getLogger(DiagnosticsProvider.class);
private static final ObjectMapper mapper = new ObjectMapper();
Expand Down Expand Up @@ -631,10 +633,10 @@ public <T extends CosmosBatchResponse> Mono<T> traceEnabledBatchResponsePublishe

return diagnostics;
},
CosmosBatchResponse::getOpCountPerEvaluation,
CosmosBatchResponse::getRetriedOpCountPerEvaluation,
CosmosBatchResponse::getGlobalOpCount,
CosmosBatchResponse::getTargetMaxMicroBatchSize,
cosmosBatchResponseAccessor::getOpCountPerEvaluation,
cosmosBatchResponseAccessor::getRetriedOpCountPerEvaluation,
cosmosBatchResponseAccessor::getGlobalOpCount,
cosmosBatchResponseAccessor::getTargetMaxMicroBatchSize,
requestOptions,
null);
}
Expand Down
8 changes: 8 additions & 0 deletions ...ure-cosmos/src/main/java/com/azure/cosmos/implementation/ImplementationBridgeHelpers.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1339,6 +1339,14 @@ public interface CosmosBatchResponseAccessor {
void setRetriedOpCountPerEvaluation(CosmosBatchResponse cosmosBatchResponse, long retriedOpCountPerEvaluation);

void setTargetMaxMicroBatchSize(CosmosBatchResponse cosmosBatchResponse, int targetMaxMicroBatchSize);

long getOpCountPerEvaluation(CosmosBatchResponse cosmosBatchResponse);

long getGlobalOpCount(CosmosBatchResponse cosmosBatchResponse);

long getRetriedOpCountPerEvaluation(CosmosBatchResponse cosmosBatchResponse);

int getTargetMaxMicroBatchSize(CosmosBatchResponse cosmosBatchResponse);
}
}

Expand Down
16 changes: 8 additions & 8 deletions ...src/main/java/com/azure/cosmos/implementation/clienttelemetry/ClientTelemetryMetrics.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1052,7 +1052,7 @@ private void recordStoreResponseStatistics(

CosmosMeterOptions opCountPerEvaluationOptions = clientAccessor.getMeterOptions(
client,
CosmosMetricName.REQUEST_SUMMARY_DIRECT_OP_COUNT_PER_EVALUATION
CosmosMetricName.REQUEST_SUMMARY_DIRECT_BULK_OP_COUNT_PER_EVALUATION
);

if (opCountPerEvaluationOptions.isEnabled()
Expand All @@ -1071,7 +1071,7 @@ private void recordStoreResponseStatistics(

CosmosMeterOptions opRetriedCountPerEvaluationOptions = clientAccessor.getMeterOptions(
client,
CosmosMetricName.REQUEST_SUMMARY_DIRECT_OP_RETRIED_COUNT_PER_EVALUATION
CosmosMetricName.REQUEST_SUMMARY_DIRECT_BULK_OP_RETRIED_COUNT_PER_EVALUATION
);

if (opRetriedCountPerEvaluationOptions.isEnabled()
Expand All @@ -1090,7 +1090,7 @@ private void recordStoreResponseStatistics(

CosmosMeterOptions globalOpCountOptions = clientAccessor.getMeterOptions(
client,
CosmosMetricName.REQUEST_SUMMARY_DIRECT_GLOBAL_OP_COUNT
CosmosMetricName.REQUEST_SUMMARY_DIRECT_BULK_GLOBAL_OP_COUNT
);

if (globalOpCountOptions.isEnabled()
Expand All @@ -1110,7 +1110,7 @@ private void recordStoreResponseStatistics(

CosmosMeterOptions targetMaxMicroBatchSizeOptions = clientAccessor.getMeterOptions(
client,
CosmosMetricName.REQUEST_SUMMARY_DIRECT_TARGET_MAX_MICRO_BATCH_SIZE
CosmosMetricName.REQUEST_SUMMARY_DIRECT_BULK_TARGET_MAX_MICRO_BATCH_SIZE
);

if (targetMaxMicroBatchSizeOptions.isEnabled()
Expand Down Expand Up @@ -1265,7 +1265,7 @@ private void recordGatewayStatistics(

CosmosMeterOptions opCountPerEvaluationOptions = clientAccessor.getMeterOptions(
client,
CosmosMetricName.REQUEST_SUMMARY_GATEWAY_OP_COUNT_PER_EVALUATION
CosmosMetricName.REQUEST_SUMMARY_GATEWAY_BULK_OP_COUNT_PER_EVALUATION
);

if (opCountPerEvaluationOptions.isEnabled()
Expand All @@ -1284,7 +1284,7 @@ private void recordGatewayStatistics(

CosmosMeterOptions opRetriedCountPerEvaluationOptions = clientAccessor.getMeterOptions(
client,
CosmosMetricName.REQUEST_SUMMARY_GATEWAY_OP_RETRIED_COUNT_PER_EVALUATION
CosmosMetricName.REQUEST_SUMMARY_GATEWAY_BULK_OP_RETRIED_COUNT_PER_EVALUATION
);

if (opRetriedCountPerEvaluationOptions.isEnabled()
Expand All @@ -1303,7 +1303,7 @@ private void recordGatewayStatistics(

CosmosMeterOptions globalOpCountOptions = clientAccessor.getMeterOptions(
client,
CosmosMetricName.REQUEST_SUMMARY_GATEWAY_GLOBAL_OP_COUNT
CosmosMetricName.REQUEST_SUMMARY_GATEWAY_BULK_GLOBAL_OP_COUNT
);

if (globalOpCountOptions.isEnabled()
Expand All @@ -1322,7 +1322,7 @@ private void recordGatewayStatistics(

CosmosMeterOptions targetMaxMicroBatchSizeOptions = clientAccessor.getMeterOptions(
client,
CosmosMetricName.REQUEST_SUMMARY_GATEWAY_TARGET_MAX_MICRO_BATCH_SIZE
CosmosMetricName.REQUEST_SUMMARY_GATEWAY_BULK_TARGET_MAX_MICRO_BATCH_SIZE
);

if (targetMaxMicroBatchSizeOptions.isEnabled()
Expand Down
35 changes: 35 additions & 0 deletions sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/models/CosmosBatchResponse.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -201,6 +201,10 @@ void setOpCountPerEvaluation(long opCountPerEvaluation) {
this.opCountPerEvaluation = opCountPerEvaluation;
}

long getOpCountPerEvaluation() {
return this.opCountPerEvaluation;
}

/**
* Set global operation count
* @param globalOpCount Global operation count
Expand All @@ -209,6 +213,10 @@ void setGlobalOpCount(long globalOpCount) {
this.globalOpCount = globalOpCount;
}

long getGlobalOpCount() {
return this.globalOpCount;
}

/**
* Set retried operation count per evaluation
* @param retriedOpCountPerEvaluation retried operation count per evaluation
Expand All @@ -217,6 +225,10 @@ void setRetriedOpCountPerEvaluation(long retriedOpCountPerEvaluation) {
this.retriedOpCountPerEvaluation = retriedOpCountPerEvaluation;
}

long getRetriedOpCountPerEvaluation() {
return this.retriedOpCountPerEvaluation;
}

/**
* Set target max micro batch size
* @param targetMaxMicroBatchSize target max micro batch size
Expand All @@ -225,6 +237,9 @@ void setTargetMaxMicroBatchSize(int targetMaxMicroBatchSize) {
this.targetMaxMicroBatchSize = targetMaxMicroBatchSize;
}

int getTargetMaxMicroBatchSize() {
return this.targetMaxMicroBatchSize;
}

void addAll(List<? extends CosmosBatchOperationResult> collection) {
this.results.addAll(collection);
Expand Down Expand Up @@ -259,6 +274,26 @@ public void setRetriedOpCountPerEvaluation(CosmosBatchResponse cosmosBatchRespon
public void setTargetMaxMicroBatchSize(CosmosBatchResponse cosmosBatchResponse, int targetMaxMicroBatchSize) {
cosmosBatchResponse.setTargetMaxMicroBatchSize(targetMaxMicroBatchSize);
}

@Override
public long getOpCountPerEvaluation(CosmosBatchResponse cosmosBatchResponse) {
return cosmosBatchResponse.getOpCountPerEvaluation();
}

@Override
public long getGlobalOpCount(CosmosBatchResponse cosmosBatchResponse) {
return cosmosBatchResponse.getGlobalOpCount();
}

@Override
public long getRetriedOpCountPerEvaluation(CosmosBatchResponse cosmosBatchResponse) {
return cosmosBatchResponse.getRetriedOpCountPerEvaluation();
}

@Override
public int getTargetMaxMicroBatchSize(CosmosBatchResponse cosmosBatchResponse) {
return cosmosBatchResponse.getTargetMaxMicroBatchSize();
}
});
}

Expand Down
Loading
Loading