Skip to content
This repository was archived by the owner on Jul 13, 2026. It is now read-only.

Commit ae04f7f

Browse files
author
Theo van Kraay
committed
simplify
1 parent 9e09a14 commit ae04f7f

2 files changed

Lines changed: 82 additions & 110 deletions

File tree

‎pom.xml‎

Lines changed: 1 addition & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -66,18 +66,13 @@
6666
<artifactId>slf4j-jdk14</artifactId>
6767
<version>1.7.28</version>
6868
</dependency>
69+
6970
<dependency>
7071
<groupId>io.micrometer</groupId>
7172
<artifactId>micrometer-registry-prometheus</artifactId>
7273
<version>1.6.6</version>
7374
</dependency>
7475

75-
<dependency>
76-
<groupId>com.azure</groupId>
77-
<artifactId>azure-cosmos-test</artifactId>
78-
<version>1.0.0-beta.3</version>
79-
</dependency>
80-
8176
<dependency>
8277
<groupId>io.prometheus</groupId>
8378
<artifactId>simpleclient_httpserver</artifactId>

‎src/main/java/com/azure/cosmos/examples/prometheus/async/CosmosClientMetricsQuickStartAsync.java‎

Lines changed: 81 additions & 104 deletions
Original file line numberDiff line numberDiff line change
@@ -3,29 +3,11 @@
33

44
package com.azure.cosmos.examples.prometheus.async;
55

6-
import com.azure.cosmos.ConsistencyLevel;
7-
import com.azure.cosmos.CosmosAsyncClient;
8-
import com.azure.cosmos.CosmosAsyncContainer;
9-
import com.azure.cosmos.CosmosAsyncDatabase;
10-
import com.azure.cosmos.CosmosClientBuilder;
11-
import com.azure.cosmos.CosmosDiagnostics;
6+
import com.azure.cosmos.*;
127
import com.azure.cosmos.examples.common.AccountSettings;
13-
import com.azure.cosmos.models.CosmosClientTelemetryConfig;
14-
import com.azure.cosmos.models.CosmosContainerProperties;
15-
import com.azure.cosmos.models.CosmosContainerResponse;
16-
import com.azure.cosmos.models.CosmosDatabaseRequestOptions;
17-
import com.azure.cosmos.models.CosmosDatabaseResponse;
18-
import com.azure.cosmos.models.CosmosMetricTagName;
19-
import com.azure.cosmos.models.CosmosMicrometerMetricsOptions;
20-
import com.azure.cosmos.models.PartitionKey;
21-
import com.azure.cosmos.models.ThroughputProperties;
22-
import com.azure.cosmos.test.faultinjection.CosmosFaultInjectionHelper;
23-
import com.azure.cosmos.test.faultinjection.FaultInjectionConditionBuilder;
24-
import com.azure.cosmos.test.faultinjection.FaultInjectionOperationType;
25-
import com.azure.cosmos.test.faultinjection.FaultInjectionResultBuilders;
26-
import com.azure.cosmos.test.faultinjection.FaultInjectionRule;
27-
import com.azure.cosmos.test.faultinjection.FaultInjectionRuleBuilder;
28-
import com.azure.cosmos.test.faultinjection.FaultInjectionServerErrorType;
8+
import com.azure.cosmos.examples.common.Family;
9+
import com.azure.cosmos.models.*;
10+
import com.azure.cosmos.util.CosmosPagedFlux;
2911
import com.fasterxml.jackson.databind.JsonNode;
3012
import com.sun.net.httpserver.HttpServer;
3113
import io.micrometer.prometheus.PrometheusConfig;
@@ -40,7 +22,6 @@
4022
import java.net.InetSocketAddress;
4123
import java.time.Duration;
4224
import java.util.ArrayList;
43-
import java.util.Arrays;
4425
import java.util.List;
4526
import java.util.concurrent.atomic.AtomicInteger;
4627

@@ -59,7 +40,7 @@ public class CosmosClientMetricsQuickStartAsync {
5940
private static AtomicInteger number_docs_inserted = new AtomicInteger(0);
6041
private static AtomicInteger write_request_count = new AtomicInteger(0);
6142
private static AtomicInteger read_request_count = new AtomicInteger(0);
62-
public static final int NUMBER_OF_DOCS = 1000;
43+
public static final int NUMBER_OF_DOCS = 5000;
6344

6445
public ArrayList<JsonNode> docs;
6546
private final static Logger logger = LoggerFactory.getLogger(CosmosClientMetricsQuickStartAsync.class);
@@ -73,80 +54,49 @@ public static void main(String[] args) {
7354

7455
try {
7556
logger.info("Starting ASYNC main");
76-
quickStart.clientMetricsDemo();
57+
quickStart.clientMetricsPrometheusDemo();
7758
logger.info("Demo complete, please hold while resources are released");
7859
} finally {
7960
logger.info("Shutting down");
8061
quickStart.shutdown();
8162
}
8263
}
8364

84-
private void clientMetricsDemo() {
65+
private void clientMetricsPrometheusDemo() {
8566

8667
logger.info("Using Azure Cosmos DB endpoint: {}", AccountSettings.HOST);
8768

88-
FaultInjectionRule serverConnectionDelayRule =
89-
new FaultInjectionRuleBuilder("ServerError-ConnectionTimeout")
90-
.condition(
91-
new FaultInjectionConditionBuilder()
92-
.operationType(FaultInjectionOperationType.CREATE_ITEM)
93-
.build()
94-
)
95-
.result(
96-
FaultInjectionResultBuilders
97-
.getResultBuilder(FaultInjectionServerErrorType.RESPONSE_DELAY)
98-
.delay(Duration.ofSeconds(6)) // default connection timeout is 5s
99-
.times(1)
100-
.build()
101-
)
102-
.duration(Duration.ofMillis(500))
103-
.hitLimit(10)
104-
.build();
105-
106-
FaultInjectionRule partitionSplitRule =
107-
new FaultInjectionRuleBuilder("Partition-Split")
108-
.condition(
109-
new FaultInjectionConditionBuilder()
110-
.operationType(FaultInjectionOperationType.CREATE_ITEM)
111-
.build()
112-
)
113-
.result(
114-
FaultInjectionResultBuilders
115-
.getResultBuilder(FaultInjectionServerErrorType.GONE)
116-
.times(1)
117-
.build()
118-
)
119-
.duration(Duration.ofMillis(2000))
120-
.hitLimit(50)
121-
.build();
122-
123-
FaultInjectionRule readTimeouts =
124-
new FaultInjectionRuleBuilder("Read-Timeouts")
125-
.condition(
126-
new FaultInjectionConditionBuilder()
127-
.operationType(FaultInjectionOperationType.READ_ITEM)
128-
.build()
129-
)
130-
.result(
131-
FaultInjectionResultBuilders
132-
.getResultBuilder(FaultInjectionServerErrorType.TIMEOUT)
133-
.times(1)
134-
.build()
135-
)
136-
.duration(Duration.ofMillis(2000))
137-
.hitLimit(50)
138-
.build();
139-
14069

70+
// <ClientMetricsConfig>
14171
//prometheus meter registry
14272
PrometheusMeterRegistry prometheusRegistry = new PrometheusMeterRegistry(PrometheusConfig.DEFAULT);
73+
14374
//provide the prometheus registry to the telemetry config
14475
CosmosClientTelemetryConfig telemetryConfig = new CosmosClientTelemetryConfig()
76+
.diagnosticsThresholds(
77+
new CosmosDiagnosticsThresholds()
78+
// Any requests that violate (are lower than) any of the below thresholds that are set
79+
// will not appear in "request-level" metrics (those with "rntbd" or "gw" in their name).
80+
// The "operation-level" metrics (those with "ops" in their name) will still be collected.
81+
// Use this to reduce noise in the amount of metrics collected.
82+
.setRequestChargeThreshold(10)
83+
.setNonPointOperationLatencyThreshold(Duration.ofDays(10))
84+
.setPointOperationLatencyThreshold(Duration.ofDays(10))
85+
)
14586
.clientCorrelationId("samplePrometheusMetrics001")
14687
.metricsOptions(new CosmosMicrometerMetricsOptions().meterRegistry(prometheusRegistry)
147-
.configureDefaultTagNames(CosmosMetricTagName.PARTITION_KEY_RANGE_ID));
148-
149-
//start local HttpServer server to expose the meter registry metrics to Prometheus
88+
//.configureDefaultTagNames(CosmosMetricTagName.PARTITION_KEY_RANGE_ID)
89+
.applyDiagnosticThresholdsForTransportLevelMeters(true)
90+
);
91+
// </ClientMetricsConfig>
92+
93+
// Start local HttpServer server to expose the meter registry metrics to Prometheus.
94+
// When adding this endpoint to prometheus.yml, add the domain name and port to "targets".
95+
// For example, if prometheus is running on the same server as this app, you can add localhost:8080:
96+
// - targets: ["localhost:9090", "localhost:8080"]
97+
// download and install prometheus from here: https://prometheus.io/download/
98+
99+
// <PrometheusTargetServer>
150100
try {
151101
HttpServer server = HttpServer.create(new InetSocketAddress(8080), 0);
152102
server.createContext("/metrics", httpExchange -> {
@@ -161,6 +111,7 @@ private void clientMetricsDemo() {
161111
} catch (IOException e) {
162112
throw new RuntimeException(e);
163113
}
114+
// </PrometheusTargetServer>
164115

165116
// Create sync client
166117
client = new CosmosClientBuilder()
@@ -169,7 +120,6 @@ private void clientMetricsDemo() {
169120
.clientTelemetryConfig(telemetryConfig)
170121
.consistencyLevel(ConsistencyLevel.SESSION) //make sure we can read our own writes
171122
.contentResponseOnWriteEnabled(true)
172-
173123
.buildAsyncClient();
174124

175125
try {
@@ -183,23 +133,13 @@ private void clientMetricsDemo() {
183133
throw new RuntimeException(e);
184134
}
185135

186-
// inject fault injection rules
187-
CosmosFaultInjectionHelper.configureFaultInjectionRules(container, Arrays.asList(serverConnectionDelayRule)).block();
188-
189136
docs = generateDocs(NUMBER_OF_DOCS);
137+
// None of the rntbd / request-level metrics for point operations will show as they violate one the thresholds set (minimum 10 RUs).
190138
createManyDocuments();
191139
readManyDocuments();
192-
pressAnyKeyToContinue("Press any key to continue ...");
193-
}
194-
195-
private void pressAnyKeyToContinue(String message) {
196-
System.out.println(message);
197-
try {
198-
// noinspection ResultOfMethodCallIgnored
199-
System.in.read();
200-
} catch (Exception e) {
201-
e.printStackTrace();
202-
}
140+
// The rntbd / request-level metrics for the below query will show as it exceeds 10 RUs.
141+
// If you comment out the below, no rntbd / request-level metrics at all will be collected due to the thresholds set.
142+
queryAllDocuments();
203143
}
204144

205145

@@ -230,7 +170,7 @@ private void createContainerIfNotExists() throws Exception {
230170
// Provision throughput
231171
ThroughputProperties throughputProperties = ThroughputProperties.createManualThroughput(10000);
232172

233-
// Create container with 200 RU/s
173+
// Create container
234174
Mono<CosmosContainerResponse> containerResponseMono = database.createContainerIfNotExists(containerProperties,
235175
throughputProperties);
236176
CosmosContainerResponse cosmosContainerResponse = containerResponseMono.block();
@@ -242,20 +182,22 @@ private void createContainerIfNotExists() throws Exception {
242182
}
243183

244184
private void createManyDocuments() {
245-
Flux.fromIterable(docs).flatMap(doc -> container.createItem(doc))
185+
Flux.fromIterable(docs).flatMap(doc -> container.createItem(doc)
186+
)
246187
.flatMap(itemResponse -> {
247188
if (itemResponse.getStatusCode() == 201) {
248189
number_docs_inserted.getAndIncrement();
190+
write_request_count.incrementAndGet();
249191
} else
250192
logger.info("WARNING insert status code {} != 201" + itemResponse.getStatusCode());
251-
write_request_count.incrementAndGet();
252193
return Mono.empty();
253194
})
254195
.onErrorContinue((throwable, o) -> {
255-
logger.error(
196+
logger.info(
256197
"Exception in create docs. e: {}", throwable.getMessage(), throwable
257198
);
258199
}).blockLast();
200+
logger.info("Number of successful write requests: " + write_request_count);
259201
}
260202

261203
private void readManyDocuments() {
@@ -270,17 +212,52 @@ private void readManyDocuments() {
270212
.flatMap(id -> container.readItem(id, new PartitionKey(id), JsonNode.class))
271213
.flatMap(itemResponse -> {
272214
if (itemResponse.getStatusCode() == 200) {
273-
logger.info("read item with id: " + itemResponse.getItem().get("id"));
215+
read_request_count.getAndIncrement();
274216
} else
275217
logger.info("WARNING insert status code {} != 200" + itemResponse.getStatusCode());
276-
read_request_count.getAndIncrement();
277218
return Mono.empty();
278219
})
279220
.onErrorContinue((throwable, o) -> {
280-
logger.error(
221+
logger.info(
281222
"Exception in create docs. e: {}", throwable.getMessage(), throwable
282223
);
283224
}).blockLast();
225+
logger.info("Number of successful read requests: " + read_request_count);
226+
}
227+
228+
private void queryAllDocuments() {
229+
230+
int preferredPageSize = number_docs_inserted.get(); // We'll use this later
231+
232+
CosmosQueryRequestOptions queryOptions = new CosmosQueryRequestOptions();
233+
234+
// Set populate query metrics to get metrics around query executions
235+
queryOptions.setQueryMetricsEnabled(true);
236+
237+
CosmosPagedFlux<Family> pagedFluxResponse = container.queryItems(
238+
"SELECT * FROM c", queryOptions, Family.class);
239+
240+
try {
241+
242+
pagedFluxResponse.byPage(preferredPageSize).flatMap(fluxResponse -> {
243+
logger.info("Got a page of query result with " +
244+
fluxResponse.getResults().size() + " items(s)"
245+
+ " and request charge of " + fluxResponse.getRequestCharge());
246+
247+
return Flux.empty();
248+
}).blockLast();
249+
250+
} catch(Exception err) {
251+
if (err instanceof CosmosException) {
252+
//Client-specific errors
253+
CosmosException cerr = (CosmosException) err;
254+
cerr.printStackTrace();
255+
logger.error(String.format("Query failed with %s\n", cerr));
256+
} else {
257+
//General errors
258+
err.printStackTrace();
259+
}
260+
}
284261
}
285262

286263
// Database delete
@@ -299,7 +276,7 @@ private void deleteDatabase() throws Exception {
299276
private void shutdown() {
300277
try {
301278
//Clean shutdown
302-
//deleteDatabase();
279+
deleteDatabase();
303280
} catch (Exception err) {
304281
logger.error("Deleting Cosmos DB resources failed, will still attempt to close the client", err);
305282
}

0 commit comments

Comments
 (0)