[FLINK-33139] Add Prometheus Sink Table API #22

darenwkt · 2025-04-02T09:15:27Z

Purpose of the change

https://issues.apache.org/jira/browse/FLINK-33139 Implements Table API for Prometheus Sink

Verifying this change

Please make sure both new and modified tests in this PR follows the conventions defined in our code quality guide: https://flink.apache.org/contributing/code-style-and-quality-common.html#testing

This change added tests and can be verified as follows:

Added unit tests for RowDataElementConverter and DynamicSinkTableFactory
Manually verified by running the Prometheus Sink TableAPI to submit data to Amazon Managed Prometheus
- SQL Code ran for the test:

In the example below, a table is created where the metricName, labels are STRING and there are 2 labels (String/String) pairs.

Flink SQL> CREATE TABLE PromTable (
>   `my_metric_name` STRING,
>   `my_label_1` STRING,
>   `my_label_2` STRING,
>   `sample_value` BIGINT,
>   `sample_ts` TIMESTAMP(3)
> )
> WITH (
>   'connector' = 'prometheus',
>   'aws.region' = 'us-east-1',
>   'metric.request-signer' = 'amazon-managed-prometheus',
>   'metric.endpoint-url' = 'https://aps-workspaces.us-east-1.amazonaws.com/workspaces/redacted/api/v1/remote_write',
>   'metric.name' = 'my_metric_name',
>   'metric.label.keys' = '[my_label_1,my_label_2]',
>   'metric.sample.key' = 'sample_value',
>   'metric.sample.timestamp' = 'sample_ts'
> );
[INFO] Execute statement succeed.

Next, a row of sample data is inserted into the table where metricName="sample_name", and the label pairs are ("my_label_1":"sample_label_1", "my_label_2":"sample_label_2")

Flink SQL> INSERT INTO PromTable VALUES ('sample_name', 'sample_label_1', 'sample_label_2', 123, TIMESTAMP '2025-04-30 10:00:00.000');
[INFO] Submitting SQL update statement to the cluster...
[INFO] SQL update statement has been successfully submitted to the cluster:
Job ID: 596c86fb5485ddd3c1a3594e3c837d02

DataStream API

PrometheusSink has been changed to support generic InputType, this is because in TableAPI, the inputType needs to be generic so that it can support RowData inputType.

However, the existing PrometheusSink is still supported with the default elementConverter remaining the same, which supports PrometheusTimeSeries -> Type.TimeSeries conversion. In addition, user can also define their own elementConverter to convert their InputType -> Type.TimeSeries.

Manually verified that existing Prometheus Sink DataStream API is still working

Significant changes

(Please check any boxes [x] if the answer is "yes". You can first publish the PR and check them afterwards, for convenience.)

[yes] Dependencies have been added or upgraded
[no] Public API has been changed (Public API is any class annotated with @Public(Evolving))
[no] Serializers have been changed
[yes] New feature has been introduced
- If yes, how is this documented? (JavaDocs)

leekeiabstraction

Left some early review comments

...ctor-prometheus/src/main/java/org/apache/flink/connector/prometheus/sink/PrometheusSink.java

...ometheus/src/main/java/org/apache/flink/connector/prometheus/sink/PrometheusSinkBuilder.java

.../src/main/java/org/apache/flink/connector/prometheus/sink/PrometheusTimeSeriesConverter.java

foxus

Many thanks for your efforts here, I've added a few comments.

...r-prometheus/src/main/java/org/apache/flink/connector/prometheus/table/PrometheusConfig.java

...eus/src/main/java/org/apache/flink/connector/prometheus/table/PrometheusConnectorOption.java

...metheus/src/main/java/org/apache/flink/connector/prometheus/table/PrometheusDynamicSink.java

.../src/main/java/org/apache/flink/connector/prometheus/table/PrometheusDynamicSinkFactory.java

...java/org/apache/flink/connector/prometheus/table/RowDataToPrometheusTimeSeriesConverter.java

darenwkt · 2025-04-23T07:02:21Z

Thanks @leekeiabstraction @foxus for the review, I have addressed the comments and updated the PR, please take a look

foxus

Thanks for the updates, loving the work to migrate the AWS specific Prometheus behaviour out. I've added additional comments on the updates.

...he/flink/connector/prometheus/sink/aws/AmazonManagedPrometheusWriteRequestSignerFactory.java

...link/connector/prometheus/sink/aws/AmazonManagedPrometheusWriteRequestSignerFactoryTest.java

darenwkt · 2025-04-29T10:54:50Z

Thanks for the updates, loving the work to migrate the AWS specific Prometheus behaviour out. I've added additional comments on the updates.

Thanks for the review, have updated PR to address them, please take another look

foxus

LGTM - thank you for the timely updates!

nicusX

Thanks for the PR
I am not sure I fully understand the Table API interface and whether it matches the use case this connector is supposed to be used for. I left some specific comments about the API. Would you please clarify?

Also, I am concerned of breaking the public interface of DataStream API. Is that required?

...eus/src/main/java/org/apache/flink/connector/prometheus/table/PrometheusConnectorOption.java

...ctor-prometheus/src/main/java/org/apache/flink/connector/prometheus/sink/PrometheusSink.java

.../src/main/java/org/apache/flink/connector/prometheus/sink/PrometheusTimeSeriesConverter.java

darenwkt · 2025-05-03T12:54:14Z

Thanks for the PR I am not sure I fully understand the Table API interface and whether it matches the use case this connector is supposed to be used for. I left some specific comments about the API. Would you please clarify?

Also, I am concerned of breaking the public interface of DataStream API. Is that required?

Thanks Lorenzo for the review, I have addressed comments on the API use case and example and I hope that it clarifies it better now.

I have also found a way to support TableAPI without breaking changes, appreciate if you can take another look at the updated PR, thank you

georgelza · 2025-05-12T09:49:06Z

hi guys...
is there a jar file available for this connector... looks like exactly what i've been looking for.
curious, looking at the Create table, don't see mention of username/password... credentials... just asking,
G

georgelza · 2025-05-12T10:53:44Z

excuse the noob question, so I did the git clone, and the build, the project folder has a couple of directories with target sub folders. which jar from which sub directory do i copy to my flink cluster/lib folder for the above to work ?

darenwkt · 2025-05-16T15:16:46Z

Hi @georgelza.

hi guys... is there a jar file available for this connector... looks like exactly what i've been looking for. curious, looking at the Create table, don't see mention of username/password... credentials... just asking, G

You can use request signer to provide authentication/credentials, you can see the Flink doc for more info on this https://nightlies.apache.org/flink/flink-docs-master/docs/connectors/datastream/prometheus/#request-signer

excuse the noob question, so I did the git clone, and the build, the project folder has a couple of directories with target sub folders. which jar from which sub directory do i copy to my flink cluster/lib folder for the above to work ?

You should not need to build it, the jar is available in maven https://mvnrepository.com/artifact/org.apache.flink/flink-connector-prometheus/1.0.0-1.20

georgelza · 2025-05-16T15:57:34Z

Hi @georgelza.

hi guys... is there a jar file available for this connector... looks like exactly what i've been looking for. curious, looking at the Create table, don't see mention of username/password... credentials... just asking, G

You can use request signer to provide authentication/credentials, you can see the Flink doc for more info on this https://nightlies.apache.org/flink/flink-docs-master/docs/connectors/datastream/prometheus/#request-signer

excuse the noob question, so I did the git clone, and the build, the project folder has a couple of directories with target sub folders. which jar from which sub directory do i copy to my flink cluster/lib folder for the above to work ?

You should not need to build it, the jar is available in maven https://mvnrepository.com/artifact/org.apache.flink/flink-connector-prometheus/1.0.0-1.20

thanks, will pull the jar from maven using the above link, just want to double check does the jar thats available include the create table ... (connector=prometheus...) functionality... as per above. I'm asking as I build as per the instructions and even though the jar file was in the correct place it seems flink was not able to find the prometheus connector functionality.
would appreciate if you can ping me privately to discuss further. georgelza@gmail.com
G

georgelza · 2025-05-17T10:37:15Z

@darenwkt No luck. does it make sense to to this rather off here...

`Flink SQL> CREATE TABLE hive_catalog.prometheus.PromTable (

sensorid STRING,
siteId STRING,
deviceId STRING,
measurement BIGINT,
ts TIMESTAMP(3)
)
WITH (
'connector' = 'prometheus',
'metric.endpoint-url' = 'prometheus:9090/api/v1/write',
'metric.name' = 'sensorid',
'metric.label.keys' = '[siteId,deviceId]',
'metric.sample.key' = 'measurement',
'metric.sample.timestamp' = 'ts'
);
[INFO] Execute statement succeeded.

Flink SQL> INSERT INTO hive_catalog.prometheus.PromTable VALUES ('10221', '106', '1031', 60, TIMESTAMP '2025-05-15 08:21:00.000');
[ERROR] Could not execute SQL statement. Reason:
org.apache.flink.table.api.ValidationException: Could not find any factory for identifier 'prometheus' that implements 'org.apache.flink.table.factories.DynamicTableFactory' in the classpath.

Available factory identifiers are:

blackhole
datagen
filesystem
kafka
paimon
print
python-input-format
upsert-kafka`

nicusX · 2025-05-17T13:16:48Z

@georgelza the released version DOES NOT support SQL. The support for SQL is still under development
Please do not use the PR discussion as a chat

nicusX

Very nice refactoring.
I have two suggestions, but nothing blocking:

It would be good validating whether table column and configuration are consistent at sink initialization rather than in the converter. Making the job failing on initialization when there is a misconfiguration is better than making it fail (and restart) on processing.
It would be nice to add an example of SQL, in form of a fake test similar to DataStreamExample. It would be SQL embedded in Java but still meaningful for an end user

Also, I am unsure how the initialization of the request-signer works in SQL. Maybe another example class for that would be helpful

nicusX · 2025-05-28T13:48:08Z

...he/flink/connector/prometheus/sink/aws/AmazonManagedPrometheusWriteRequestSignerFactory.java

+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+
+public class AmazonManagedPrometheusWriteRequestSignerFactory


I'd be interested to see how this is used to set up the request signer declaratively in SQL

Request signer is now a plugin (through Java Service Provider Interface (SPI)), and can be specified by user in the table config (metric.request-signer), for example:

CREATE TABLE PromTable ( `my_metric_name` BIGINT, `my_label_1` BIGINT, `my_label_2` BIGINT, `sample_value` BIGINT, `sample_ts` TIMESTAMP(3) ) WITH ( 'connector' = 'prometheus', 'aws.region' = 'us-east-1', 'metric.request-signer' = 'amazon-managed-prometheus', 'metric.endpoint-url' = 'https://aps-workspaces.us-east-1.amazonaws.com/workspaces/abc', 'metric.name' = 'my_metric_name', 'metric.label.keys' = '[my_label_1,my_label_2]', 'metric.sample.key' = 'sample_value', 'metric.sample.timestamp' = 'sample_ts', 'sink.batch.max-size' = '2', 'sink.flush-buffer.size' = '1' );

please explain "
'sink.batch.max-size' = '2',
'sink.flush-buffer.size' = '1'
have idea what the intend is, just want to confirm...

Hi @georgelza,

'sink.batch.max-size' determines the max size of batch records before flushing it to sink. I would recommend checking the asyncSink FLIP for more details on how it works https://cwiki.apache.org/confluence/display/FLINK/FLIP-171%3A+Async+Sink

Another helpful resource is to check other similar connector and how these config are used https://nightlies.apache.org/flink/flink-docs-master/docs/connectors/table/kinesis/#sink-batch-max-size

When this PR is merged and released, we should have equivalent docs on what each of these config mean as well

nicusX · 2025-05-28T14:11:50Z

...java/org/apache/flink/connector/prometheus/table/RowDataToPrometheusTimeSeriesConverter.java

+                builder.withMetricName(fieldValue.getStringValue());
+            } else if (fieldName.equals(prometheusConfig.getMetricSampleKey())) {
+                sampleValue = fieldValue.getDoubleValue();
+            } else if (prometheusConfig.getLabelKeys().contains(fieldName)) {


LGTM
Just thinking out loud: the Label name is the field name and the value is the content.
No issue as label names must follow [a-zA-Z_]([a-zA-Z0-9_])* which is a subset of possible column names.
This limits the usage a bit compared to DataStream because all metrics represented by a table must have the same set of Label names.
But I do not think anything different is possible with a table abstraction

I may require further clarification on the question here, but my understanding is that this approach allows setting any label name actually, for example, the following label names are all valid:

'metric.label.keys' = '[my_label_1, my_label_2, LABEL@123, 123%label, label(123)]'

Labels in Prometheus are {key, value}
you are adding the fieldName as key, and the content of the field as value

builder.addLabel(fieldName, fieldValue.getStringValue())

Which is fine. But it means all records will have the same set of Label keys, because in a table abstraction all records have the same fields by definition. Label values will be different, being the content of the fields.
As I mentioned, it's limited compared to DataStream but acceptable

"LABEL@123", "123%label" and "label(123)" would be illegal Label's key (names) for Prometheus BTW (see specs). They would be rejected by Prometheus. But the behaviour is not be inconsistent with the DataStream version where illegal label names (which are data in that case, not configurations) are rejected at runtime when you try writing. I am not suggesting a change.

Thanks for the clarification, yes I agree, the behaviour does differ from datastream API in that the label key/name is fixed per table.

Thanks Lorenzo, I have also added TableAPIExample as well, feel free to take a look. It seems like TableAPI groupBy is not the same as keyBy operation, so there might be some limitation around parallelism > 1

nicusX · 2025-05-28T14:25:19Z

...java/org/apache/flink/connector/prometheus/table/RowDataToPrometheusTimeSeriesConverter.java

+            } else if (fieldName.equals(prometheusConfig.getMetricSampleTimestamp())) {
+                sampleTimestamp = fieldValue.getLongValue();
+            } else {
+                throw new IllegalArgumentException(


It would be really nice to move this validation to the builder or the sink constructor. This would cause the job to fail when created, instead of failing when the first record is processed (putting the job in a fail-and-restart loop).
A more user friendly behaviour considering this is a misconfiguration rather than a runtime problem.

Yes this is a good suggestion, and I can confirm that the validation are already in place, please take a look at:

PrometheusDynamicSinkFactory.validateMetricConfigKeys()

PrometheusDynamicSinkFactory.requiredOptions()

They are both called in sink constructor which aligns with your suggestion

nicusX

Approved because my notes in my last review are not blockers

georgelza · 2025-06-02T13:29:09Z

Hi Daren I actually think the way you defining the labels are very elegant. G

…

On Mon, Jun 2, 2025 at 3:27 PM darenwkt ***@***.***> wrote: ***@***.**** commented on this pull request. ------------------------------ In flink-connector-prometheus/src/main/java/org/apache/flink/connector/prometheus/table/RowDataToPrometheusTimeSeriesConverter.java <#22 (comment)> : > + PrometheusTimeSeries.Builder builder = PrometheusTimeSeries.builder(); + Double sampleValue = null; + Long sampleTimestamp = null; + + for (int i = 0; i < fields.size(); i++) { + DataTypes.Field field = fields.get(i); + RowData.FieldGetter fieldGetter = + createFieldGetter(fields.get(i).getDataType().getLogicalType(), i); + FieldValue fieldValue = new FieldValue(fieldGetter.getFieldOrNull(row)); + String fieldName = field.getName(); + + if (fieldName.equals(prometheusConfig.getMetricName())) { + builder.withMetricName(fieldValue.getStringValue()); + } else if (fieldName.equals(prometheusConfig.getMetricSampleKey())) { + sampleValue = fieldValue.getDoubleValue(); + } else if (prometheusConfig.getLabelKeys().contains(fieldName)) { I may require further clarification on the question here, but my understanding is that this approach allows setting any label name actually, for example, the following label names are all valid: 'metric.label.keys' = '[my_label_1, my_label_2, ***@***.***, 123%label, label(123)]' — Reply to this email directly, view it on GitHub <#22 (comment)>, or unsubscribe <https://github.com/notifications/unsubscribe-auth/AGZLFP7W3XYQO2NRRNNXMSL3BRGM3AVCNFSM6AAAAAB2JAJHWCVHI2DSMVQWIX3LMV43YUDVNRWFEZLROVSXG5CSMV3GSZLXHMZDQOBYGQYTKNJUGI> . You are receiving this because you were mentioned.Message ID: ***@***.***>

-- You have the obligation to inform one honestly of the risk, and as a person you are committed to educate yourself to the total risk in any activity! Once informed & totally aware of the risk, every fool has the right to kill or injure themselves as they see fit!

georgelza · 2025-06-02T18:17:58Z

def keen to get access to the jar that implements this sink, have a blog just about ready to implement a IoT ->kafka->flink-prometheus-Grafana flow... G

…

On Mon, Jun 2, 2025 at 7:40 PM darenwkt ***@***.***> wrote: ***@***.**** commented on this pull request. ------------------------------ In flink-connector-prometheus/src/main/java/org/apache/flink/connector/prometheus/table/RowDataToPrometheusTimeSeriesConverter.java <#22 (comment)> : > + DataTypes.Field field = fields.get(i); + RowData.FieldGetter fieldGetter = + createFieldGetter(fields.get(i).getDataType().getLogicalType(), i); + FieldValue fieldValue = new FieldValue(fieldGetter.getFieldOrNull(row)); + String fieldName = field.getName(); + + if (fieldName.equals(prometheusConfig.getMetricName())) { + builder.withMetricName(fieldValue.getStringValue()); + } else if (fieldName.equals(prometheusConfig.getMetricSampleKey())) { + sampleValue = fieldValue.getDoubleValue(); + } else if (prometheusConfig.getLabelKeys().contains(fieldName)) { + builder.addLabel(fieldName, fieldValue.getStringValue()); + } else if (fieldName.equals(prometheusConfig.getMetricSampleTimestamp())) { + sampleTimestamp = fieldValue.getLongValue(); + } else { + throw new IllegalArgumentException( Yes this is a good suggestion, and I can confirm that the validation are already in place, please take a look at: - PrometheusDynamicSinkFactory.validateMetricConfigKeys() - PrometheusDynamicSinkFactory.requiredOptions() They are both called in sink constructor which aligns with your suggestion — Reply to this email directly, view it on GitHub <#22 (comment)>, or unsubscribe <https://github.com/notifications/unsubscribe-auth/AGZLFP4QHMAFJJKZDWJBOE33BSEB5AVCNFSM6AAAAAB2JAJHWCVHI2DSMVQWIX3LMV43YUDVNRWFEZLROVSXG5CSMV3GSZLXHMZDQOBZGM4DKNJZGI> . You are receiving this because you were mentioned.Message ID: ***@***.***>

-- You have the obligation to inform one honestly of the risk, and as a person you are committed to educate yourself to the total risk in any activity! Once informed & totally aware of the risk, every fool has the right to kill or injure themselves as they see fit!

hlteoh37

Thanks @darenwkt for the PR!
I'm not a big fan of the way the PR is currently implemented, but I can see the reasons motivating it (non-generic interfaces). I've mainly put in larger comments, but will progress on essential "formatting" comments (e.g. @PublicEvolving etc.) after we agree on the below:

Main concerns:
1/ There is currently tight coupling between PrometheusSinkBase and PrometheusSink, which is a code smell, of wrongly directed dependency! Given this is a relatively new connector, shall we consider instead taking the hit of making a breaking public interface change, and bumping the major version, to include only 1 PrometheusSink with the generic type?

2/ The current type handling for RowData -> String on metric labels is a little bit too open, in my opinion. https://nightlies.apache.org/flink/flink-docs-master/docs/dev/table/types/#list-of-data-types There is a fixed set of Flink SQL data types. We should consider fixing the list of available types for fields (e.g. CHAR, VARCHAR, etc) and rejecting types that are funky like (RAW, MULTISET etc.)

...he/flink/connector/prometheus/sink/aws/AmazonManagedPrometheusWriteRequestSignerFactory.java

...rg/apache/flink/connector/prometheus/sink/aws/AmazonManagedPrometheusWriteRequestSigner.java

...-prometheus/src/main/java/org/apache/flink/connector/prometheus/sink/PrometheusSinkBase.java

...metheus/src/main/java/org/apache/flink/connector/prometheus/table/PrometheusDynamicSink.java

...ctor-prometheus/src/main/java/org/apache/flink/connector/prometheus/sink/PrometheusSink.java

...-prometheus/src/main/java/org/apache/flink/connector/prometheus/sink/PrometheusSinkBase.java

...java/org/apache/flink/connector/prometheus/table/RowDataToPrometheusTimeSeriesConverter.java

darenwkt · 2025-06-04T07:49:01Z

Thanks @hlteoh37 for the review! I have addressed the comments and appreciate if you could help take another look, thank you

1/ There is currently tight coupling between PrometheusSinkBase and PrometheusSink, which is a code smell, of wrongly directed dependency! Given this is a relatively new connector, shall we consider instead taking the hit of making a breaking public interface change, and bumping the major version, to include only 1 PrometheusSink with the generic type?

Yes, +1 to this, I have updated this in the latest revision

2/ The current type handling for RowData -> String on metric labels is a little bit too open, in my opinion. https://nightlies.apache.org/flink/flink-docs-master/docs/dev/table/types/#list-of-data-types There is a fixed set of Flink SQL data types. We should consider fixing the list of available types for fields (e.g. CHAR, VARCHAR, etc) and rejecting types that are funky like (RAW, MULTISET etc.)

This is a good callout, we should only support (CHAR, VARCHAR, STRING) for metric name and labels, I have updated this in the latest revision

georgelza · 2025-06-04T14:52:35Z

Hi Darren Is sink.batch.max-size and sink.flush-buffer.size part of the current build. I noticed metric.name needs to map to a column in the table, I will argue that this is not desirable... you might have multiple tables being that will push to the prometheus store. based on the table being used the user will want to define a text string as the name of the metrics contained... G

…

On Wed, Jun 4, 2025 at 9:57 AM darenwkt ***@***.***> wrote: ***@***.**** commented on this pull request. ------------------------------ In flink-connector-prometheus-request-signer-amp/src/main/java/org/apache/flink/connector/prometheus/sink/aws/AmazonManagedPrometheusWriteRequestSignerFactory.java <#22 (comment)> : > @@ -0,0 +1,36 @@ +package org.apache.flink.connector.prometheus.sink.aws; + +import org.apache.flink.connector.prometheus.sink.PrometheusRequestSigner; +import org.apache.flink.connector.prometheus.table.PrometheusConfig; +import org.apache.flink.connector.prometheus.table.PrometheusDynamicRequestSignerFactory; + +import java.util.regex.Matcher; +import java.util.regex.Pattern; + +public class AmazonManagedPrometheusWriteRequestSignerFactory Hi @georgelza <https://github.com/georgelza>, 'sink.batch.max-size' determines the max size of batch records before flushing it to sink. I would recommend checking the asyncSink FLIP for more details on how it works https://cwiki.apache.org/confluence/display/FLINK/FLIP-171%3A+Async+Sink Another helpful resource is to check other similar connector and how these config are used https://nightlies.apache.org/flink/flink-docs-master/docs/connectors/table/kinesis/#sink-batch-max-size When this PR is merged and released, we should have equivalent docs on what each of these config mean as well — Reply to this email directly, view it on GitHub <#22 (comment)>, or unsubscribe <https://github.com/notifications/unsubscribe-auth/AGZLFPZT6VNQTEBKLASEMFD3B2RFFAVCNFSM6AAAAAB2JAJHWCVHI2DSMVQWIX3LMV43YUDVNRWFEZLROVSXG5CSMV3GSZLXHMZDQOJVGY4TINRQHE> . You are receiving this because you were mentioned.Message ID: ***@***.***>

-- You have the obligation to inform one honestly of the risk, and as a person you are committed to educate yourself to the total risk in any activity! Once informed & totally aware of the risk, every fool has the right to kill or injure themselves as they see fit!

darenwkt · 2025-06-08T07:47:19Z

Hi @georgelza ,

Is sink.batch.max-size and sink.flush-buffer.size part of the current build.

Can I clarify on your question please? Do you mean if the current TableAPI connector implementation allows user to configure their own sink.batch.max-size and sink.flush-buffer.size?

I noticed metric.name needs to map to a column in the table, I will argue
that this is not desirable... you might have multiple tables being that
will push to the prometheus store. based on the table being used the user
will want to define a text string as the name of the metrics contained...

In the Datastream API connector, we allow PrometheusSink to submit to different MetricName according to what is set in TimeSeries. Hence, the TableAPI connector is consistent in this regard as well. https://nightlies.apache.org/flink/flink-docs-master/docs/connectors/datastream/prometheus/#populating-a-prometheustimeseries

georgelza · 2025-06-08T08:07:49Z

@darenwkt
thanks for the responses.

w.r.t. sink.batch.max-size and sink.flush-buffer.size? using it ran into the inherent rules of the one has to be bigger than the other, but when I followed that had some other errors... so eventually for the blog i simply removed both,

got it working, and it works nicely :) took some figuring out how the connector works, how to use Prometheus remote write and then also the time of the prometheus server and time of my metrics can't be to far in past (historic data) or to Marin future, time offset between prometheus server and data source TZ.

wrote a blog, using it, thats scheduled to come out tomorrow actually.
Will be watching this one, think it will have some good legs on it.
thanks for the work guys, appreciated.

darenwkt force-pushed the FLINK-33139 branch from 4c98907 to e200be8 Compare April 9, 2025 20:49

leekeiabstraction reviewed Apr 14, 2025

View reviewed changes

foxus suggested changes Apr 15, 2025

View reviewed changes

darenwkt force-pushed the FLINK-33139 branch 2 times, most recently from cebd639 to 5644639 Compare April 23, 2025 06:29

darenwkt requested a review from foxus April 23, 2025 07:00

darenwkt requested a review from leekeiabstraction April 23, 2025 09:02

foxus suggested changes Apr 29, 2025

View reviewed changes

...he/flink/connector/prometheus/sink/aws/AmazonManagedPrometheusWriteRequestSignerFactory.java Outdated Show resolved Hide resolved

...link/connector/prometheus/sink/aws/AmazonManagedPrometheusWriteRequestSignerFactoryTest.java Show resolved Hide resolved

darenwkt force-pushed the FLINK-33139 branch from 5644639 to c18248d Compare April 29, 2025 10:52

foxus approved these changes Apr 29, 2025

View reviewed changes

nicusX suggested changes May 2, 2025

View reviewed changes

darenwkt force-pushed the FLINK-33139 branch 3 times, most recently from 63061ee to 006471b Compare May 3, 2025 12:31

darenwkt force-pushed the FLINK-33139 branch from 006471b to 3cabc3c Compare May 3, 2025 12:56

darenwkt requested a review from nicusX May 3, 2025 12:56

darenwkt force-pushed the FLINK-33139 branch from 3cabc3c to a4f0988 Compare May 3, 2025 13:00

nicusX reviewed May 28, 2025

View reviewed changes

nicusX approved these changes May 28, 2025

View reviewed changes

hlteoh37 requested changes Jun 3, 2025

View reviewed changes

darenwkt force-pushed the FLINK-33139 branch 2 times, most recently from 6deca3f to 9de55ca Compare June 4, 2025 07:44

darenwkt requested a review from hlteoh37 June 4, 2025 07:49

[FLINK-33139] Add Prometheus Sink Table API

c847a3a

darenwkt force-pushed the FLINK-33139 branch from 9de55ca to c847a3a Compare June 4, 2025 11:29

[FLINK-33139] Add Prometheus Sink Table API #22

Are you sure you want to change the base?

[FLINK-33139] Add Prometheus Sink Table API #22

Uh oh!

Conversation

darenwkt commented Apr 2, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Purpose of the change

Verifying this change

DataStream API

Significant changes

Uh oh!

leekeiabstraction left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

foxus left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

darenwkt commented Apr 23, 2025

Uh oh!

foxus left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

darenwkt commented Apr 29, 2025

Uh oh!

foxus left a comment

Choose a reason for hiding this comment

Uh oh!

nicusX left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

darenwkt commented May 3, 2025

Uh oh!

georgelza commented May 12, 2025

Uh oh!

georgelza commented May 12, 2025

Uh oh!

darenwkt commented May 16, 2025

Uh oh!

georgelza commented May 16, 2025

Uh oh!

georgelza commented May 17, 2025

Uh oh!

nicusX commented May 17, 2025

Uh oh!

nicusX left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

darenwkt Jun 2, 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

darenwkt commented Apr 2, 2025 •

edited

Loading

darenwkt Jun 2, 2025 •

edited

Loading