Spring/spring-boot
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Support Pushgateway with new Prometheus client

Browse Source 
Closes gh-43923
This commit is contained in:
Andy Wilkinson
2025-02-04 11:06:28 +00:00
parent 465f05ba4a
commit 63ecfac40d
10 changed files with 330 additions and 113 deletions
Download Patch File Download Diff File Expand all files Collapse all files

51
spring-boot-project/spring-boot-actuator/src/main/java/org/springframework/boot/actuate/metrics/export/prometheus/PrometheusPushGatewayManager.java
View File

@@ -17,13 +17,11 @@
package org.springframework.boot.actuate.metrics.export.prometheus;
import java.time.Duration;
import java.util.Map;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledFuture;
import io.prometheus.client.CollectorRegistry;
import io.prometheus.client.exporter.PushGateway;
import io.prometheus.metrics.exporter.pushgateway.PushGateway;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
@@ -46,56 +44,31 @@ public class PrometheusPushGatewayManager {
private final PushGateway pushGateway;
private final CollectorRegistry registry;
private final String job;
private final Map<String, String> groupingKey;
private final ShutdownOperation shutdownOperation;
private final TaskScheduler scheduler;
private final ScheduledFuture<?> scheduled;
/**
* Create a new {@link PrometheusPushGatewayManager} instance using a single threaded
* {@link TaskScheduler}.
* @param pushGateway the source push gateway
* @param registry the collector registry to push
* @param pushRate the rate at which push operations occur
* @param job the job ID for the operation
* @param groupingKeys an optional set of grouping keys for the operation
* @param shutdownOperation the shutdown operation that should be performed when
* context is closed.
*/
public PrometheusPushGatewayManager(PushGateway pushGateway, CollectorRegistry registry, Duration pushRate,
String job, Map<String, String> groupingKeys, ShutdownOperation shutdownOperation) {
this(pushGateway, registry, new PushGatewayTaskScheduler(), pushRate, job, groupingKeys, shutdownOperation);
}
/**
* Create a new {@link PrometheusPushGatewayManager} instance.
* @param pushGateway the source push gateway
* @param registry the collector registry to push
* @param scheduler the scheduler used for operations
* @param pushRate the rate at which push operations occur
* @param job the job ID for the operation
* @param groupingKey an optional set of grouping keys for the operation
* @param shutdownOperation the shutdown operation that should be performed when
* context is closed.
* @since 3.5.0
*/
public PrometheusPushGatewayManager(PushGateway pushGateway, CollectorRegistry registry, TaskScheduler scheduler,
Duration pushRate, String job, Map<String, String> groupingKey, ShutdownOperation shutdownOperation) {
public PrometheusPushGatewayManager(PushGateway pushGateway, Duration pushRate,
ShutdownOperation shutdownOperation) {
this(pushGateway, new PushGatewayTaskScheduler(), pushRate, shutdownOperation);
}
PrometheusPushGatewayManager(PushGateway pushGateway, TaskScheduler scheduler, Duration pushRate,
ShutdownOperation shutdownOperation) {
Assert.notNull(pushGateway, "'pushGateway' must not be null");
Assert.notNull(registry, "'registry' must not be null");
Assert.notNull(scheduler, "'scheduler' must not be null");
Assert.notNull(pushRate, "'pushRate' must not be null");
Assert.hasLength(job, "'job' must not be empty");
this.pushGateway = pushGateway;
this.registry = registry;
this.job = job;
this.groupingKey = groupingKey;
this.shutdownOperation = (shutdownOperation != null) ? shutdownOperation : ShutdownOperation.NONE;
this.scheduler = scheduler;
this.scheduled = this.scheduler.scheduleAtFixedRate(this::post, pushRate);
@@ -103,7 +76,7 @@ public class PrometheusPushGatewayManager {
private void post() {
try {
this.pushGateway.pushAdd(this.registry, this.job, this.groupingKey);
this.pushGateway.pushAdd();
}
catch (Throwable ex) {
logger.warn("Unexpected exception thrown by POST of metrics to Prometheus Pushgateway", ex);
@@ -112,7 +85,7 @@ public class PrometheusPushGatewayManager {
private void put() {
try {
this.pushGateway.push(this.registry, this.job, this.groupingKey);
this.pushGateway.push();
}
catch (Throwable ex) {
logger.warn("Unexpected exception thrown by PUT of metrics to Prometheus Pushgateway", ex);
@@ -121,7 +94,7 @@ public class PrometheusPushGatewayManager {
private void delete() {
try {
this.pushGateway.delete(this.job, this.groupingKey);
this.pushGateway.delete();
}
catch (Throwable ex) {
logger.warn("Unexpected exception thrown by DELETE of metrics from Prometheus Pushgateway", ex);

79
spring-boot-project/spring-boot-actuator/src/test/java/org/springframework/boot/actuate/metrics/export/prometheus/PrometheusPushGatewayManagerTests.java
View File

@@ -17,12 +17,9 @@
package org.springframework.boot.actuate.metrics.export.prometheus;
import java.time.Duration;
import java.util.Collections;
import java.util.Map;
import java.util.concurrent.ScheduledFuture;
import io.prometheus.client.CollectorRegistry;
import io.prometheus.client.exporter.PushGateway;
import io.prometheus.metrics.exporter.pushgateway.PushGateway;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.ExtendWith;
import org.mockito.ArgumentCaptor;
@@ -55,16 +52,11 @@ class PrometheusPushGatewayManagerTests {
@Mock
private PushGateway pushGateway;
@Mock
private CollectorRegistry registry;
@Mock
private TaskScheduler scheduler;
private final Duration pushRate = Duration.ofSeconds(1);
private final Map<String, String> groupingKey = Collections.singletonMap("foo", "bar");
@Captor
private ArgumentCaptor<Runnable> task;
@@ -74,68 +66,48 @@ class PrometheusPushGatewayManagerTests {
@Test
void createWhenPushGatewayIsNullThrowsException() {
assertThatIllegalArgumentException()
.isThrownBy(() -> new PrometheusPushGatewayManager(null, this.registry, this.scheduler, this.pushRate,
"job", this.groupingKey, null))
.isThrownBy(() -> new PrometheusPushGatewayManager(null, this.scheduler, this.pushRate, null))
.withMessage("'pushGateway' must not be null");
}
@Test
void createWhenCollectorRegistryIsNullThrowsException() {
assertThatIllegalArgumentException()
.isThrownBy(() -> new PrometheusPushGatewayManager(this.pushGateway, null, this.scheduler, this.pushRate,
"job", this.groupingKey, null))
.withMessage("'registry' must not be null");
}
@Test
void createWhenSchedulerIsNullThrowsException() {
assertThatIllegalArgumentException()
.isThrownBy(() -> new PrometheusPushGatewayManager(this.pushGateway, this.registry, null, this.pushRate,
"job", this.groupingKey, null))
.isThrownBy(() -> new PrometheusPushGatewayManager(this.pushGateway, null, this.pushRate, null))
.withMessage("'scheduler' must not be null");
}
@Test
void createWhenPushRateIsNullThrowsException() {
assertThatIllegalArgumentException()
.isThrownBy(() -> new PrometheusPushGatewayManager(this.pushGateway, this.registry, this.scheduler, null,
"job", this.groupingKey, null))
.isThrownBy(() -> new PrometheusPushGatewayManager(this.pushGateway, this.scheduler, null, null))
.withMessage("'pushRate' must not be null");
}
@Test
void createWhenJobIsEmptyThrowsException() {
assertThatIllegalArgumentException()
.isThrownBy(() -> new PrometheusPushGatewayManager(this.pushGateway, this.registry, this.scheduler,
this.pushRate, "", this.groupingKey, null))
.withMessage("'job' must not be empty");
}
@Test
void createShouldSchedulePushAsFixedRate() throws Exception {
new PrometheusPushGatewayManager(this.pushGateway, this.registry, this.scheduler, this.pushRate, "job",
this.groupingKey, null);
new PrometheusPushGatewayManager(this.pushGateway, this.scheduler, this.pushRate, null);
then(this.scheduler).should().scheduleAtFixedRate(this.task.capture(), eq(this.pushRate));
this.task.getValue().run();
then(this.pushGateway).should().pushAdd(this.registry, "job", this.groupingKey);
then(this.pushGateway).should().pushAdd();
}
@Test
void shutdownWhenOwnsSchedulerDoesShutdownScheduler() {
void shutdownWhenOwnsSchedulerDoesShutDownScheduler() {
PushGatewayTaskScheduler ownedScheduler = givenScheduleAtFixedRateWillReturnFuture(
mock(PushGatewayTaskScheduler.class));
PrometheusPushGatewayManager manager = new PrometheusPushGatewayManager(this.pushGateway, this.registry,
ownedScheduler, this.pushRate, "job", this.groupingKey, null);
PrometheusPushGatewayManager manager = new PrometheusPushGatewayManager(this.pushGateway, ownedScheduler,
this.pushRate, null);
manager.shutdown();
then(ownedScheduler).should().shutdown();
}
@Test
void shutdownWhenDoesNotOwnSchedulerDoesNotShutdownScheduler() {
void shutdownWhenDoesNotOwnSchedulerDoesNotShutDownScheduler() {
ThreadPoolTaskScheduler otherScheduler = givenScheduleAtFixedRateWillReturnFuture(
mock(ThreadPoolTaskScheduler.class));
PrometheusPushGatewayManager manager = new PrometheusPushGatewayManager(this.pushGateway, this.registry,
otherScheduler, this.pushRate, "job", this.groupingKey, null);
PrometheusPushGatewayManager manager = new PrometheusPushGatewayManager(this.pushGateway, otherScheduler,
this.pushRate, null);
manager.shutdown();
then(otherScheduler).should(never()).shutdown();
}
@@ -143,38 +115,38 @@ class PrometheusPushGatewayManagerTests {
@Test
void shutdownWhenShutdownOperationIsPostPerformsPushAddOnShutdown() throws Exception {
givenScheduleAtFixedRateWithReturnFuture();
PrometheusPushGatewayManager manager = new PrometheusPushGatewayManager(this.pushGateway, this.registry,
this.scheduler, this.pushRate, "job", this.groupingKey, ShutdownOperation.POST);
PrometheusPushGatewayManager manager = new PrometheusPushGatewayManager(this.pushGateway, this.scheduler,
this.pushRate, ShutdownOperation.POST);
manager.shutdown();
then(this.future).should().cancel(false);
then(this.pushGateway).should().pushAdd(this.registry, "job", this.groupingKey);
then(this.pushGateway).should().pushAdd();
}
@Test
void shutdownWhenShutdownOperationIsPutPerformsPushOnShutdown() throws Exception {
givenScheduleAtFixedRateWithReturnFuture();
PrometheusPushGatewayManager manager = new PrometheusPushGatewayManager(this.pushGateway, this.registry,
this.scheduler, this.pushRate, "job", this.groupingKey, ShutdownOperation.PUT);
PrometheusPushGatewayManager manager = new PrometheusPushGatewayManager(this.pushGateway, this.scheduler,
this.pushRate, ShutdownOperation.PUT);
manager.shutdown();
then(this.future).should().cancel(false);
then(this.pushGateway).should().push(this.registry, "job", this.groupingKey);
then(this.pushGateway).should().push();
}
@Test
void shutdownWhenShutdownOperationIsDeletePerformsDeleteOnShutdown() throws Exception {
givenScheduleAtFixedRateWithReturnFuture();
PrometheusPushGatewayManager manager = new PrometheusPushGatewayManager(this.pushGateway, this.registry,
this.scheduler, this.pushRate, "job", this.groupingKey, ShutdownOperation.DELETE);
PrometheusPushGatewayManager manager = new PrometheusPushGatewayManager(this.pushGateway, this.scheduler,
this.pushRate, ShutdownOperation.DELETE);
manager.shutdown();
then(this.future).should().cancel(false);
then(this.pushGateway).should().delete("job", this.groupingKey);
then(this.pushGateway).should().delete();
}
@Test
void shutdownWhenShutdownOperationIsNoneDoesNothing() {
givenScheduleAtFixedRateWithReturnFuture();
PrometheusPushGatewayManager manager = new PrometheusPushGatewayManager(this.pushGateway, this.registry,
this.scheduler, this.pushRate, "job", this.groupingKey, ShutdownOperation.NONE);
PrometheusPushGatewayManager manager = new PrometheusPushGatewayManager(this.pushGateway, this.scheduler,
this.pushRate, ShutdownOperation.NONE);
manager.shutdown();
then(this.future).should().cancel(false);
then(this.pushGateway).shouldHaveNoInteractions();
@@ -182,10 +154,9 @@ class PrometheusPushGatewayManagerTests {
@Test
void pushDoesNotThrowException() throws Exception {
new PrometheusPushGatewayManager(this.pushGateway, this.registry, this.scheduler, this.pushRate, "job",
this.groupingKey, null);
new PrometheusPushGatewayManager(this.pushGateway, this.scheduler, this.pushRate, null);
then(this.scheduler).should().scheduleAtFixedRate(this.task.capture(), eq(this.pushRate));
willThrow(RuntimeException.class).given(this.pushGateway).pushAdd(this.registry, "job", this.groupingKey);
willThrow(RuntimeException.class).given(this.pushGateway).pushAdd();
this.task.getValue().run();
}