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

Commit c1dcd05

Browse files
TheovanKraayTheovanKraay
authored
Merge pull request #63 from TheovanKraay/prometheus-metrics-sample
Prometheus metrics sample
2 parents cd25029 + bad456f commit c1dcd05

2 files changed

Lines changed: 303 additions & 0 deletions

File tree

‎pom.xml‎

Lines changed: 13 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -71,6 +71,19 @@
7171
<version>2.17.2</version>
7272
</dependency>
7373

74+
<dependency>
75+
<groupId>io.micrometer</groupId>
76+
<artifactId>micrometer-registry-prometheus</artifactId>
77+
<version>1.6.6</version>
78+
</dependency>
79+
80+
<dependency>
81+
<groupId>io.prometheus</groupId>
82+
<artifactId>simpleclient_httpserver</artifactId>
83+
<version>0.5.0</version>
84+
</dependency>
85+
<!-- Thanks for using https://jar-download.com -->
86+
7487
<dependency>
7588
<groupId>org.apache.commons</groupId>
7689
<artifactId>commons-lang3</artifactId>

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

Lines changed: 290 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,290 @@
1+
// Copyright (c) Microsoft Corporation. All rights reserved.
2+
// Licensed under the MIT License.
3+
4+
package com.azure.cosmos.examples.prometheus.async;
5+
6+
import com.azure.cosmos.*;
7+
import com.azure.cosmos.examples.common.AccountSettings;
8+
import com.azure.cosmos.examples.common.Family;
9+
import com.azure.cosmos.models.*;
10+
import com.azure.cosmos.util.CosmosPagedFlux;
11+
import com.fasterxml.jackson.databind.JsonNode;
12+
import com.sun.net.httpserver.HttpServer;
13+
import io.micrometer.prometheus.PrometheusConfig;
14+
import io.micrometer.prometheus.PrometheusMeterRegistry;
15+
import org.slf4j.Logger;
16+
import org.slf4j.LoggerFactory;
17+
import reactor.core.publisher.Flux;
18+
import reactor.core.publisher.Mono;
19+
20+
import java.io.IOException;
21+
import java.io.OutputStream;
22+
import java.net.InetSocketAddress;
23+
import java.time.Duration;
24+
import java.util.ArrayList;
25+
import java.util.List;
26+
import java.util.concurrent.atomic.AtomicInteger;
27+
28+
import static com.azure.cosmos.examples.common.Profile.generateDocs;
29+
30+
public class CosmosClientMetricsQuickStartAsync {
31+
32+
private CosmosAsyncClient client;
33+
34+
private final String databaseName = "ClientMetricsPrometheusDB";
35+
private final String containerName = "Container";
36+
37+
private CosmosAsyncDatabase database;
38+
private CosmosAsyncContainer container;
39+
40+
private static AtomicInteger number_docs_inserted = new AtomicInteger(0);
41+
private static AtomicInteger write_request_count = new AtomicInteger(0);
42+
private static AtomicInteger read_request_count = new AtomicInteger(0);
43+
public static final int NUMBER_OF_DOCS = 5000;
44+
45+
public ArrayList<JsonNode> docs;
46+
private final static Logger logger = LoggerFactory.getLogger(CosmosClientMetricsQuickStartAsync.class);
47+
48+
public void close() {
49+
client.close();
50+
}
51+
52+
public static void main(String[] args) {
53+
CosmosClientMetricsQuickStartAsync quickStart = new CosmosClientMetricsQuickStartAsync();
54+
55+
try {
56+
logger.info("Starting ASYNC main");
57+
quickStart.clientMetricsPrometheusDemo();
58+
logger.info("Demo complete, please hold while resources are released");
59+
} finally {
60+
logger.info("Shutting down");
61+
quickStart.shutdown();
62+
}
63+
}
64+
65+
private void clientMetricsPrometheusDemo() {
66+
67+
logger.info("Using Azure Cosmos DB endpoint: {}", AccountSettings.HOST);
68+
69+
70+
// <ClientMetricsConfig>
71+
//prometheus meter registry
72+
PrometheusMeterRegistry prometheusRegistry = new PrometheusMeterRegistry(PrometheusConfig.DEFAULT);
73+
74+
//provide the prometheus registry to the telemetry config
75+
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+
)
86+
// Uncomment below to apply sampling to help further tune client-side resource consumption related to metrics.
87+
// The sampling rate can be modified after Azure Cosmos DB Client initialization – so the sampling rate can be
88+
// modified without any restarts being necessary.
89+
//.sampleDiagnostics(0.25)
90+
.clientCorrelationId("samplePrometheusMetrics001")
91+
.metricsOptions(new CosmosMicrometerMetricsOptions().meterRegistry(prometheusRegistry)
92+
//.configureDefaultTagNames(CosmosMetricTagName.PARTITION_KEY_RANGE_ID)
93+
.applyDiagnosticThresholdsForTransportLevelMeters(true)
94+
);
95+
// </ClientMetricsConfig>
96+
97+
// Start local HttpServer server to expose the meter registry metrics to Prometheus.
98+
// When adding this endpoint to prometheus.yml, add the domain name and port to "targets".
99+
// For example, if prometheus is running on the same server as this app, you can add localhost:8080:
100+
// - targets: ["localhost:9090", "localhost:8080"]
101+
// download and install prometheus from here: https://prometheus.io/download/
102+
103+
// <PrometheusTargetServer>
104+
try {
105+
HttpServer server = HttpServer.create(new InetSocketAddress(8080), 0);
106+
server.createContext("/metrics", httpExchange -> {
107+
String response = prometheusRegistry.scrape();
108+
int i = 1;
109+
httpExchange.sendResponseHeaders(200, response.getBytes().length);
110+
try (OutputStream os = httpExchange.getResponseBody()) {
111+
os.write(response.getBytes());
112+
}
113+
});
114+
new Thread(server::start).start();
115+
} catch (IOException e) {
116+
throw new RuntimeException(e);
117+
}
118+
// </PrometheusTargetServer>
119+
120+
// Create sync client
121+
client = new CosmosClientBuilder()
122+
.endpoint(AccountSettings.HOST)
123+
.key(AccountSettings.MASTER_KEY)
124+
.clientTelemetryConfig(telemetryConfig)
125+
.consistencyLevel(ConsistencyLevel.SESSION) //make sure we can read our own writes
126+
.contentResponseOnWriteEnabled(true)
127+
.buildAsyncClient();
128+
129+
try {
130+
createDatabaseIfNotExists();
131+
} catch (Exception e) {
132+
throw new RuntimeException(e);
133+
}
134+
try {
135+
createContainerIfNotExists();
136+
} catch (Exception e) {
137+
throw new RuntimeException(e);
138+
}
139+
140+
docs = generateDocs(NUMBER_OF_DOCS);
141+
// None of the rntbd / request-level metrics for point operations will show as they violate one the thresholds set (minimum 10 RUs).
142+
createManyDocuments();
143+
readManyDocuments();
144+
// The rntbd / request-level metrics for the below query will show as it exceeds 10 RUs.
145+
// If you comment out the below, no rntbd / request-level metrics at all will be collected due to the thresholds set.
146+
queryAllDocuments();
147+
}
148+
149+
150+
// Database create
151+
private void createDatabaseIfNotExists() throws Exception {
152+
logger.info("Creating database {} if not exists", databaseName);
153+
154+
// Create database if not exists
155+
Mono<CosmosDatabaseResponse> databaseResponseMono = client.createDatabaseIfNotExists(databaseName);
156+
CosmosDatabaseResponse cosmosDatabaseResponse = databaseResponseMono.block();
157+
158+
CosmosDiagnostics diagnostics = cosmosDatabaseResponse.getDiagnostics();
159+
logger.info("Create database diagnostics : {}", diagnostics);
160+
161+
database = client.getDatabase(cosmosDatabaseResponse.getProperties().getId());
162+
163+
logger.info("Done.");
164+
}
165+
166+
// Container create
167+
private void createContainerIfNotExists() throws Exception {
168+
logger.info("Creating container {} if not exists", containerName);
169+
170+
// Create container if not exists
171+
CosmosContainerProperties containerProperties =
172+
new CosmosContainerProperties(containerName, "/id");
173+
174+
// Provision throughput
175+
ThroughputProperties throughputProperties = ThroughputProperties.createManualThroughput(10000);
176+
177+
// Create container
178+
Mono<CosmosContainerResponse> containerResponseMono = database.createContainerIfNotExists(containerProperties,
179+
throughputProperties);
180+
CosmosContainerResponse cosmosContainerResponse = containerResponseMono.block();
181+
CosmosDiagnostics diagnostics = cosmosContainerResponse.getDiagnostics();
182+
logger.info("Create container diagnostics : {}", diagnostics);
183+
container = database.getContainer(cosmosContainerResponse.getProperties().getId());
184+
185+
logger.info("Done.");
186+
}
187+
188+
private void createManyDocuments() {
189+
Flux.fromIterable(docs).flatMap(doc -> container.createItem(doc)
190+
)
191+
.flatMap(itemResponse -> {
192+
if (itemResponse.getStatusCode() == 201) {
193+
number_docs_inserted.getAndIncrement();
194+
write_request_count.incrementAndGet();
195+
} else
196+
logger.info("WARNING insert status code {} != 201" + itemResponse.getStatusCode());
197+
return Mono.empty();
198+
})
199+
.onErrorContinue((throwable, o) -> {
200+
logger.info(
201+
"Exception in create docs. e: {}", throwable.getMessage(), throwable
202+
);
203+
}).blockLast();
204+
logger.info("Number of successful write requests: " + write_request_count);
205+
}
206+
207+
private void readManyDocuments() {
208+
// collect the ids that were generated when writing the data.
209+
List<String> list = new ArrayList<String>();
210+
for (final JsonNode doc : docs) {
211+
list.add(doc.get("id").asText());
212+
}
213+
214+
final long startTime = System.currentTimeMillis();
215+
Flux.fromIterable(list)
216+
.flatMap(id -> container.readItem(id, new PartitionKey(id), JsonNode.class))
217+
.flatMap(itemResponse -> {
218+
if (itemResponse.getStatusCode() == 200) {
219+
read_request_count.getAndIncrement();
220+
} else
221+
logger.info("WARNING insert status code {} != 200" + itemResponse.getStatusCode());
222+
return Mono.empty();
223+
})
224+
.onErrorContinue((throwable, o) -> {
225+
logger.info(
226+
"Exception in create docs. e: {}", throwable.getMessage(), throwable
227+
);
228+
}).blockLast();
229+
logger.info("Number of successful read requests: " + read_request_count);
230+
}
231+
232+
private void queryAllDocuments() {
233+
234+
int preferredPageSize = number_docs_inserted.get(); // We'll use this later
235+
236+
CosmosQueryRequestOptions queryOptions = new CosmosQueryRequestOptions();
237+
238+
// Set populate query metrics to get metrics around query executions
239+
queryOptions.setQueryMetricsEnabled(true);
240+
241+
CosmosPagedFlux<Family> pagedFluxResponse = container.queryItems(
242+
"SELECT * FROM c", queryOptions, Family.class);
243+
244+
try {
245+
246+
pagedFluxResponse.byPage(preferredPageSize).flatMap(fluxResponse -> {
247+
logger.info("Got a page of query result with " +
248+
fluxResponse.getResults().size() + " items(s)"
249+
+ " and request charge of " + fluxResponse.getRequestCharge());
250+
251+
return Flux.empty();
252+
}).blockLast();
253+
254+
} catch(Exception err) {
255+
if (err instanceof CosmosException) {
256+
//Client-specific errors
257+
CosmosException cerr = (CosmosException) err;
258+
cerr.printStackTrace();
259+
logger.error(String.format("Query failed with %s\n", cerr));
260+
} else {
261+
//General errors
262+
err.printStackTrace();
263+
}
264+
}
265+
}
266+
267+
// Database delete
268+
private void deleteDatabase() throws Exception {
269+
logger.info("Last step: delete database {} by ID", databaseName);
270+
271+
// Delete database
272+
CosmosDatabaseResponse dbResp =
273+
client.getDatabase(databaseName).delete(new CosmosDatabaseRequestOptions()).block();
274+
logger.info("Status code for database delete: {}", dbResp.getStatusCode());
275+
276+
logger.info("Done.");
277+
}
278+
279+
// Cleanup before close
280+
private void shutdown() {
281+
try {
282+
//Clean shutdown
283+
deleteDatabase();
284+
} catch (Exception err) {
285+
logger.error("Deleting Cosmos DB resources failed, will still attempt to close the client", err);
286+
}
287+
//client.close();
288+
logger.info("Done with sample.");
289+
}
290+
}

0 commit comments

Comments
 (0)