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

Commit 9e09a14

Browse files
author
Theo van Kraay
committed
updates
1 parent a649b7e commit 9e09a14

2 files changed

Lines changed: 85 additions & 71 deletions

File tree

‎pom.xml‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -75,7 +75,7 @@
7575
<dependency>
7676
<groupId>com.azure</groupId>
7777
<artifactId>azure-cosmos-test</artifactId>
78-
<version>1.0.0-beta.1</version>
78+
<version>1.0.0-beta.3</version>
7979
</dependency>
8080

8181
<dependency>

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

Lines changed: 84 additions & 70 deletions
Original file line numberDiff line numberDiff line change
@@ -10,17 +10,16 @@
1010
import com.azure.cosmos.CosmosClientBuilder;
1111
import com.azure.cosmos.CosmosDiagnostics;
1212
import com.azure.cosmos.examples.common.AccountSettings;
13-
import com.azure.cosmos.examples.common.Family;
1413
import com.azure.cosmos.models.CosmosClientTelemetryConfig;
1514
import com.azure.cosmos.models.CosmosContainerProperties;
1615
import com.azure.cosmos.models.CosmosContainerResponse;
1716
import com.azure.cosmos.models.CosmosDatabaseRequestOptions;
1817
import com.azure.cosmos.models.CosmosDatabaseResponse;
19-
import com.azure.cosmos.models.CosmosItemRequestOptions;
20-
import com.azure.cosmos.models.CosmosItemResponse;
18+
import com.azure.cosmos.models.CosmosMetricTagName;
2119
import com.azure.cosmos.models.CosmosMicrometerMetricsOptions;
2220
import com.azure.cosmos.models.PartitionKey;
2321
import com.azure.cosmos.models.ThroughputProperties;
22+
import com.azure.cosmos.test.faultinjection.CosmosFaultInjectionHelper;
2423
import com.azure.cosmos.test.faultinjection.FaultInjectionConditionBuilder;
2524
import com.azure.cosmos.test.faultinjection.FaultInjectionOperationType;
2625
import com.azure.cosmos.test.faultinjection.FaultInjectionResultBuilders;
@@ -41,8 +40,8 @@
4140
import java.net.InetSocketAddress;
4241
import java.time.Duration;
4342
import java.util.ArrayList;
43+
import java.util.Arrays;
4444
import java.util.List;
45-
import java.util.UUID;
4645
import java.util.concurrent.atomic.AtomicInteger;
4746

4847
import static com.azure.cosmos.examples.common.Profile.generateDocs;
@@ -51,16 +50,15 @@ public class CosmosClientMetricsQuickStartAsync {
5150

5251
private CosmosAsyncClient client;
5352

54-
private final String databaseName = "ClientMetricsPrometheusTestDB";
55-
private final String containerName = "FamilyContainer";
56-
private final String documentId = UUID.randomUUID().toString();
57-
private final String documentLastName = "Witherspoon";
53+
private final String databaseName = "ClientMetricsPrometheusDB";
54+
private final String containerName = "Container";
5855

5956
private CosmosAsyncDatabase database;
6057
private CosmosAsyncContainer container;
6158

6259
private static AtomicInteger number_docs_inserted = new AtomicInteger(0);
63-
private static AtomicInteger request_count = new AtomicInteger(0);
60+
private static AtomicInteger write_request_count = new AtomicInteger(0);
61+
private static AtomicInteger read_request_count = new AtomicInteger(0);
6462
public static final int NUMBER_OF_DOCS = 1000;
6563

6664
public ArrayList<JsonNode> docs;
@@ -70,27 +68,20 @@ public void close() {
7068
client.close();
7169
}
7270

73-
public static void main(String[] args) {
71+
public static void main(String[] args) {
7472
CosmosClientMetricsQuickStartAsync quickStart = new CosmosClientMetricsQuickStartAsync();
7573

76-
77-
78-
79-
8074
try {
8175
logger.info("Starting ASYNC main");
82-
8376
quickStart.clientMetricsDemo();
8477
logger.info("Demo complete, please hold while resources are released");
85-
} catch (Exception e) {
86-
logger.error("Cosmos getStarted failed with", e);
87-
} finally {
78+
} finally {
8879
logger.info("Shutting down");
8980
quickStart.shutdown();
9081
}
9182
}
9283

93-
private void clientMetricsDemo() throws Exception {
84+
private void clientMetricsDemo() {
9485

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

@@ -108,16 +99,54 @@ private void clientMetricsDemo() throws Exception {
10899
.times(1)
109100
.build()
110101
)
111-
.duration(Duration.ofMillis(10))
112-
.hitLimit(2)
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)
113138
.build();
114139

115140

141+
//prometheus meter registry
116142
PrometheusMeterRegistry prometheusRegistry = new PrometheusMeterRegistry(PrometheusConfig.DEFAULT);
143+
//provide the prometheus registry to the telemetry config
117144
CosmosClientTelemetryConfig telemetryConfig = new CosmosClientTelemetryConfig()
118-
.metricsOptions(new CosmosMicrometerMetricsOptions().meterRegistry(prometheusRegistry));
119-
145+
.clientCorrelationId("samplePrometheusMetrics001")
146+
.metricsOptions(new CosmosMicrometerMetricsOptions().meterRegistry(prometheusRegistry)
147+
.configureDefaultTagNames(CosmosMetricTagName.PARTITION_KEY_RANGE_ID));
120148

149+
//start local HttpServer server to expose the meter registry metrics to Prometheus
121150
try {
122151
HttpServer server = HttpServer.create(new InetSocketAddress(8080), 0);
123152
server.createContext("/metrics", httpExchange -> {
@@ -128,28 +157,34 @@ private void clientMetricsDemo() throws Exception {
128157
os.write(response.getBytes());
129158
}
130159
});
131-
132160
new Thread(server::start).start();
133161
} catch (IOException e) {
134162
throw new RuntimeException(e);
135163
}
136164

137-
138-
139165
// Create sync client
140166
client = new CosmosClientBuilder()
141167
.endpoint(AccountSettings.HOST)
142168
.key(AccountSettings.MASTER_KEY)
143169
.clientTelemetryConfig(telemetryConfig)
144-
.consistencyLevel(ConsistencyLevel.EVENTUAL)
170+
.consistencyLevel(ConsistencyLevel.SESSION) //make sure we can read our own writes
145171
.contentResponseOnWriteEnabled(true)
146-
.buildAsyncClient();
147172

173+
.buildAsyncClient();
148174

149-
createDatabaseIfNotExists();
150-
createContainerIfNotExists();
175+
try {
176+
createDatabaseIfNotExists();
177+
} catch (Exception e) {
178+
throw new RuntimeException(e);
179+
}
180+
try {
181+
createContainerIfNotExists();
182+
} catch (Exception e) {
183+
throw new RuntimeException(e);
184+
}
151185

152-
//CosmosFaultInjectionHelper.configureFaultInjectionRules(container, Arrays.asList(serverConnectionDelayRule)).block();
186+
// inject fault injection rules
187+
CosmosFaultInjectionHelper.configureFaultInjectionRules(container, Arrays.asList(serverConnectionDelayRule)).block();
153188

154189
docs = generateDocs(NUMBER_OF_DOCS);
155190
createManyDocuments();
@@ -168,7 +203,7 @@ private void pressAnyKeyToContinue(String message) {
168203
}
169204

170205

171-
// Database Diagnostics
206+
// Database create
172207
private void createDatabaseIfNotExists() throws Exception {
173208
logger.info("Creating database {} if not exists", databaseName);
174209

@@ -206,46 +241,24 @@ private void createContainerIfNotExists() throws Exception {
206241
logger.info("Done.");
207242
}
208243

209-
private void createDocuments() throws Exception {
210-
logger.info("Create document : {}", documentId);
211-
212-
// Define a document as a POJO (internally this
213-
// is converted to JSON via custom serialization)
214-
Family family = new Family();
215-
family.setLastName(documentLastName);
216-
family.setId(documentId);
217-
218-
// Insert this item as a document
219-
// Explicitly specifying the /pk value improves performance.
220-
Mono<CosmosItemResponse<Family>> itemResponseMono = container.createItem(family,
221-
new PartitionKey(family.getLastName()),
222-
new CosmosItemRequestOptions());
223-
224-
CosmosItemResponse<Family> itemResponse = itemResponseMono.block();
225-
CosmosDiagnostics diagnostics = itemResponse.getDiagnostics();
226-
logger.info("Create item diagnostics : {}", diagnostics);
227-
228-
logger.info("Done.");
229-
}
230-
231-
private void createManyDocuments() throws Exception {
244+
private void createManyDocuments() {
232245
Flux.fromIterable(docs).flatMap(doc -> container.createItem(doc))
233246
.flatMap(itemResponse -> {
234247
if (itemResponse.getStatusCode() == 201) {
235248
number_docs_inserted.getAndIncrement();
236249
} else
237250
logger.info("WARNING insert status code {} != 201" + itemResponse.getStatusCode());
238-
request_count.incrementAndGet();
251+
write_request_count.incrementAndGet();
239252
return Mono.empty();
240-
}).subscribe(); // ...Subscribing to the publisher triggers stream execution.
241-
242-
logger.info("Doing other things until async doc inserts complete...");
243-
while (request_count.get() < NUMBER_OF_DOCS) {
244-
}
245-
253+
})
254+
.onErrorContinue((throwable, o) -> {
255+
logger.error(
256+
"Exception in create docs. e: {}", throwable.getMessage(), throwable
257+
);
258+
}).blockLast();
246259
}
247260

248-
private void readManyDocuments() throws InterruptedException {
261+
private void readManyDocuments() {
249262
// collect the ids that were generated when writing the data.
250263
List<String> list = new ArrayList<String>();
251264
for (final JsonNode doc : docs) {
@@ -260,13 +273,14 @@ private void readManyDocuments() throws InterruptedException {
260273
logger.info("read item with id: " + itemResponse.getItem().get("id"));
261274
} else
262275
logger.info("WARNING insert status code {} != 200" + itemResponse.getStatusCode());
263-
request_count.getAndIncrement();
276+
read_request_count.getAndIncrement();
264277
return Mono.empty();
265-
}).subscribe();
266-
logger.info("Waiting while subscribed async operation completes all threads...");
267-
while (request_count.get() < NUMBER_OF_DOCS) {
268-
// looping while subscribed async operation completes all threads
269-
}
278+
})
279+
.onErrorContinue((throwable, o) -> {
280+
logger.error(
281+
"Exception in create docs. e: {}", throwable.getMessage(), throwable
282+
);
283+
}).blockLast();
270284
}
271285

272286
// Database delete
@@ -285,7 +299,7 @@ private void deleteDatabase() throws Exception {
285299
private void shutdown() {
286300
try {
287301
//Clean shutdown
288-
deleteDatabase();
302+
//deleteDatabase();
289303
} catch (Exception err) {
290304
logger.error("Deleting Cosmos DB resources failed, will still attempt to close the client", err);
291305
}

0 commit comments

Comments
 (0)