Add Prometheus Sink by kkondaka · Pull Request #6229 · opensearch-project/data-prepper

kkondaka · 2025-11-01T05:22:32Z

Description

Add Prometheus Sink.
Only supports Amazon Promethus Sink as the destination with AWS credentials.

Issues Resolved

Resolves #3028

Check List

New functionality includes testing.
New functionality has a documentation issue. Please link to it in this PR.
New functionality has javadoc added
Commits are signed with a real name per the DCO

By submitting this pull request, I confirm that my contribution is made under the terms of the Apache 2.0 license.
For more information on following Developer Certificate of Origin and signing off your commits, please check here.

Signed-off-by: Kondaka <krishkdk@amazon.com>

KarstenSchnitter · 2025-11-04T13:17:51Z

How much effort would be required to make this compatible with a standard Prometheus instance, e.g. prom/prometheus?

kkondaka · 2025-11-04T19:09:56Z

@KarstenSchnitter I think it should work with any prometheus server once we support different auth mechanisms. May be the simplest one maybe username/password auth support

graytaylor0 · 2025-11-05T00:40:32Z

+                .with("statusCode", statusCode)
+                .with("pluginName", pluginSetting.getName())
+                .with("pipelineName", pluginSetting.getPipelineName());


Should this metadata be made into variables? Seems like common metadata all sinks may use

We could. Is it OK to do this as a separate followup PR?

KarstenSchnitter · 2025-11-05T08:29:34Z

@kkondaka can you add basic auth support? Other authentications can be added on request.

dlvenable

Thanks @kkondaka ! This is a great improvement. I took an initial look and left some comments.

dlvenable · 2025-11-05T16:33:58Z

+        Region region = (awsConfig != null) ? awsConfig.getAwsRegion() : awsCredentialsSupplier.getDefaultRegion().get();
+
+        sinkMetrics = new DefaultSinkMetrics(pluginMetrics, "metric");
+        httpSender = new PrometheusHttpSender(awsCredentialsSupplier, prometheusSinkConfiguration, sinkMetrics, prometheusSinkConfiguration.getConnectionTimeoutMs(), prometheusSinkConfiguration.getIdleTimeoutMs());


Let's start moving toward dependency injection. This is support now. You can see an example of this in PR #6190. I added DI to the DynamoDB source coordinator.

I will do this in a fast-followup PR

dlvenable · 2025-11-05T16:34:59Z

@@ -29,7 +37,7 @@

 public class PrometheusSinkTest {


We should have at least one test that uses the test plugin framework to verify behavior. See GrokProcessorIT as an example. At the very least it will get the automatic tests that verify that the plugin can load.

I will do this in a fast-followup PR

dlvenable · 2025-11-10T18:03:30Z

+    }
+
+    public String toString() {
+        String result = "";


Let's not manually build out JSON. Use Jackson.

Actually, this is not needed. Removed it.

dlvenable · 2025-11-10T18:04:58Z

+                .maxTotalAttempts(config.getMaxRetries() + 1);
+
+        final long httpTimeoutMs = Math.min(
+                Math.max(config.getThresholdConfig().getFlushInterval() * 2, 3_000), 10_000


You have magic numbers here. Make these into constants and use a Duration to derive it for readability.

They are copied from XRay Sink :-) And not even used. Removed it.

dlvenable · 2025-11-10T18:07:00Z

+
+        final RequestHeadersBuilder headersBuilder = RequestHeaders.builder()
+                .method(HttpMethod.POST)
+                .scheme(signedRequest.getUri().getScheme())


The logic for the HTTP request should not be tied to the signing. Get the URI independently of AWS SDK signing.

Agreed. Again copied from XRay. And I will work on supporting opensource prometheus server which doesn't use AWS signing. Will address this issue then.

dlvenable · 2025-11-10T18:07:53Z

+                .authority(signedRequest.getUri().getAuthority());
+
+        // Preserve all original headers from the signed request without modification
+        signedRequest.headers().forEach((k, vList) -> {


Adding signing headers should be done by an AWS-specific class. Either the current signer or a new class.

Actually, isn't adding the headers the only thing that the AWS-specific signing class really needs to do?

Agreed. Please see above.

dlvenable · 2025-11-10T18:09:09Z

+     *
+     * @param payload - batch the batch of spans to send
+     */
+    public Pair<Boolean, Integer> pushToEndPoint(final byte[] payload) {


Let's avoid Pair classes on public methods because left and right don't have much meaning outside the pair. Just make a simple POJO model.

Also, we should not require the AWS SDK for general HTTP methods.

dlvenable · 2025-11-10T18:16:44Z

+        }
+        String encodedQuery = URLEncoder.encode(query, StandardCharsets.UTF_8);
+        String getUrlQuery = "query=" + query + "&start="+testStartTime+"&end="+endTime+"&step=1s";
+        String getUrl = queryRangeUrl+"?query=" + encodedQuery + "&start="+testStartTime+"&end="+endTime+"&step=1s";


Use a URL builder instead of direct string manipulation. Maybe Armeria has one. If not, Spring has one that Data Prepper uses in other locations.

For some reason query string URLBuilder() is not working. Since this is test code, I think it is OK to keep it this way for now.

dlvenable · 2025-11-10T18:17:35Z

+                                assertThat(actualMetricValue, equalTo(10+j));
+                            } else if (metric.get("quantile") != null) {
+                                String q = (String)metric.get("quantile");
+                                if (q.equals("0.5")) {


What does this mean? How does one understand these conditions?

Quantile is a well known term in OTEL. - https://opentelemetry.io/docs/specs/otel/metrics/data-model/

dlvenable · 2025-11-10T18:19:54Z

    classpath = sourceSets.integrationTest.runtimeClasspath
-    systemProperty 'tests.prometheus.sink.http.endpoint', System.getProperty('tests.prometheus.sink.http.endpoint')
-
+    systemProperty 'tests.prometheus.url', System.getProperty('tests.prometheus.url')


How does one run these integration tests locally? Please add instructions.

dlvenable · 2025-11-10T18:21:45Z

+     *
+     * @param payload - batch the batch of spans to send
+     */
+    public Pair<Boolean, Integer> pushToEndPoint(final byte[] payload) {


Rename to pushToEndpoint.

dlvenable · 2025-11-10T18:24:50Z


    @NotNull
    @JsonProperty("url")
    private String url;


It looks like you removed the insecure flag. We should require that this url start with https:// if you aren't going to keep that.

We should keep the insecure_skip_verify since this is helpful in many situations where there is not a root CA.

I will work on supporting "insecure" and some auth mechanism to work with opensource prometheus server.

Also, "root CA" is not applicable to AWS, right? Will support other auth mechanisms in a different PR

Signed-off-by: Kondaka <krishkdk@amazon.com>

dlvenable · 2025-11-12T22:40:34Z


 public interface CompressionEngine {
    OutputStream createOutputStream(OutputStream outputStream) throws IOException;
+    default byte[] compress(byte[] payload) throws IOException {


We can give an implementation here. Something like this.

default byte[] compress(byte[] payload) throws IOException { try (ByteArrayOutputStream compressedBytesStream = new ByteArrayOutputStream(); OutputStream compressedOut = createOutputStream(payloadOutputStream)) { compressedOut.write(payload); compressedOut.close(); return compressedBytesStream.toByteArray(); } }

Looks like Stream compression is different from block compression. This implementation won't work for Prometheus.

Signed-off-by: Kondaka <krishkdk@amazon.com>

dlvenable · 2025-11-13T00:10:47Z

@kkondaka , Thank you for this great new feature! Let's address some of the remaining comments in a follow-on PR.

sb2k16 · 2025-11-13T04:36:45Z

+     * @param awsCredentialsSupplier the AWS credentials supplier
+     * @param config The configuration for the Prometheus sink plugin.
+     */
+    public PrometheusHttpSender(@Nonnull final AwsCredentialsSupplier awsCredentialsSupplier, @Nonnull final PrometheusSinkConfiguration config, @Nonnull final SinkMetrics sinkMetrics, final long connectionTimeoutMillis, final long idleTimeoutMillis) {


Is there a unit test for this class?

Will add it in the followup PR

sb2k16 · 2025-11-13T04:37:46Z

+    private static WebClient buildWebClient(final PrometheusSinkConfiguration config) {
+        final RetryRuleWithContent<HttpResponse> retryRule = RetryRuleWithContent.<HttpResponse>builder()
+                .onStatus((ctx, status) -> RETRYABLE_STATUS_CODES.contains(status.code()))
+                .thenBackoff(Backoff.exponential(100, 10_000).withJitter(0.2));


Perhaps, these values could be moved to the Sink configuration.

No need for configuring this for now. I will make them constants.

Add Prometheus Sink

5044610

Signed-off-by: Kondaka <krishkdk@amazon.com>

kkondaka requested review from KarstenSchnitter, chenqi0805, dinujoh, dlvenable, engechas, graytaylor0, oeyh, san81, sb2k16 and srikanthjg as code owners November 1, 2025 05:22

kkondaka added 3 commits November 1, 2025 09:37

Remove debug statements

a5c708e

Signed-off-by: Kondaka <krishkdk@amazon.com>

Make Prometheus sink Experimental

15bb14c

Signed-off-by: Kondaka <krishkdk@amazon.com>

Add sanitize_names config option to sanitize metric/label names

aa845ba

Signed-off-by: Kondaka <krishkdk@amazon.com>

graytaylor0 reviewed Nov 5, 2025

View reviewed changes

dlvenable requested changes Nov 5, 2025

View reviewed changes

graytaylor0 previously approved these changes Nov 10, 2025

View reviewed changes

dlvenable requested changes Nov 10, 2025

View reviewed changes

Addressed review comments

57957a1

Signed-off-by: Kondaka <krishkdk@amazon.com>

kkondaka dismissed graytaylor0’s stale review via 57957a1 November 11, 2025 03:48

Cleaned up HTTP sender and Sigv4Signer

3510093

Signed-off-by: Kondaka <krishkdk@amazon.com>

dlvenable reviewed Nov 12, 2025

View reviewed changes

Added check for https in valid config

405711a

Signed-off-by: Kondaka <krishkdk@amazon.com>

dlvenable approved these changes Nov 13, 2025

View reviewed changes

graytaylor0 approved these changes Nov 13, 2025

View reviewed changes

kkondaka merged commit 3eaf52a into opensearch-project:main Nov 13, 2025
78 of 80 checks passed

sb2k16 reviewed Nov 13, 2025

View reviewed changes

Conversation

kkondaka commented Nov 1, 2025 • edited by KarstenSchnitter Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Description

Issues Resolved

Check List

Uh oh!

KarstenSchnitter commented Nov 4, 2025

Uh oh!

kkondaka commented Nov 4, 2025

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

KarstenSchnitter commented Nov 5, 2025

Uh oh!

dlvenable left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

kkondaka Nov 10, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

kkondaka commented Nov 1, 2025 •

edited by KarstenSchnitter

Loading

kkondaka Nov 10, 2025 •

edited

Loading