You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by "darenwkt (via GitHub)" <gi...@apache.org> on 2025/04/02 09:15:29 UTC
[PR] [FLINK-33139] Add Prometheus Sink Table API [flink-connector-prometheus]
darenwkt opened a new pull request, #22:
URL: https://github.com/apache/flink-connector-prometheus/pull/22
<!--
*Thank you for contributing to the Apache Flink Prometheus Connector - we are happy that you want to help us improve our Flink connectors. To help the community review your contribution in the best possible way, please go through the checklist below, which will get the contribution into a shape in which it can be best reviewed.*
## Contribution Checklist
- The name of the pull request should correspond to a [JIRA issue](https://issues.apache.org/jira/projects/FLINK/issues). Exceptions are made for typos in JavaDoc or documentation files, which need no JIRA issue.
- Commits should be in the form of "[FLINK-XXXX][Connectors / Prometheus] Title of the pull request", where [FLINK-XXXX] should be replaced by the actual issue number.
- Each pull request should only have one JIRA issue.
- Once all items of the checklist are addressed, remove the above text and this checklist, leaving only the filled out template below.
-->
## 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:
```
CREATE TABLE PromTable (
`my_metric_name` BIGINT,
`my_label` BIGINT,
`sample_value` BIGINT,
`sample_ts` TIMESTAMP(3)
)
WITH (
'connector' = 'prometheus',
'aws.region' = 'us-east-1',
'metric.request-signer' = 'amp',
'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',
'metric.sample.key' = 'sample_value',
'metric.sample.timestamp' = 'sample_ts',
'sink.batch.max-size' = '2',
'sink.flush-buffer.size' = '1'
);
CREATE TABLE DataGen (
`my_metric_name` STRING,
`my_label` STRING,
`sample_value` BIGINT,
`sample_ts` TIMESTAMP(3)
) WITH (
'connector' = 'datagen',
'rows-per-second' = '50',
'fields.my_metric_name.kind' = 'random',
'fields.my_metric_name.values' = '["my_metric_1", "my_metric_2"]',
'fields.my_label.kind' = 'random',
'fields.my_label.values' = '["my_label_1", "my_label_2"]'
);
INSERT INTO PromTable SELECT my_metric_name, my_label, sample_value, sample_ts FROM DataGen;
```
- Manually verified that existing Prometheus Sink DataStream API is still working
<img width="399" alt="Screenshot 2025-04-02 at 09 38 23" src="https://github.com/user-attachments/assets/20af1d1c-20c3-46ed-8306-32756c3b0f8a" />
## 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)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-33139] Add Prometheus Sink Table API [flink-connector-prometheus]
Posted by "darenwkt (via GitHub)" <gi...@apache.org>.
darenwkt commented on code in PR #22:
URL: https://github.com/apache/flink-connector-prometheus/pull/22#discussion_r2072389078
##########
flink-connector-prometheus/src/main/java/org/apache/flink/connector/prometheus/table/PrometheusConnectorOption.java:
##########
@@ -0,0 +1,67 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.connector.prometheus.table;
+
+import org.apache.flink.configuration.ConfigOption;
+import org.apache.flink.configuration.ConfigOptions;
+
+import java.util.List;
+
+public class PrometheusConnectorOption {
+
+ // -----------------------------------------------------------------------------------------
+ // Prometheus connector specific options
+ // -----------------------------------------------------------------------------------------
+
+ public static final ConfigOption<String> METRIC_NAME =
+ ConfigOptions.key("metric.name")
Review Comment:
Thanks Lorenzo for the review and apology for the misleading example as there were typo in there, I have updated description and below with an accurate example:
In the example below, a table is created where the metricName, labels are STRING and there are 2 labels (String/String) pairs used as you have correctly pointed out.
```
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',
> 'metric.endpoint-url' = 'https://aps-workspaces.us-east-1.amazonaws.com/workspaces/ws-bb306bbf-e744-42e9-98a4-d90d7b2ed073/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
```
##########
flink-connector-prometheus/src/main/java/org/apache/flink/connector/prometheus/sink/PrometheusSink.java:
##########
@@ -49,11 +49,11 @@ public class PrometheusSink extends AsyncSinkBase<PrometheusTimeSeries, Types.Ti
@SuppressWarnings("checkstyle:RegexpSingleline")
protected PrometheusSink(
- ElementConverter<PrometheusTimeSeries, Types.TimeSeries> elementConverter,
+ ElementConverter<IN, Types.TimeSeries> elementConverter,
Review Comment:
I have taken a look and have found a way to support TableAPI without breaking change.
In summary, the @PublicEvolving API here are:
- PrometheusSink
- PrometheusSinkBuilder
- PrometheusTimeSeriesConverter
The challenge here is that the above 3 classes need to support RowData as inputType for TableAPI implementation.
There are two ways to solve this:
- Option 1 - Make all 3 classes take a generic input type (Causes breaking change)
- Option 2 - Introduce a Base class with generic input type for all 3 classes which can now inherit it. (No breaking change)
From your comment, I agree that breaking change is unnecessary and I have updated the PR to go for Option 2.
##########
flink-connector-prometheus/src/main/java/org/apache/flink/connector/prometheus/sink/PrometheusSink.java:
##########
@@ -49,11 +49,11 @@ public class PrometheusSink extends AsyncSinkBase<PrometheusTimeSeries, Types.Ti
@SuppressWarnings("checkstyle:RegexpSingleline")
protected PrometheusSink(
- ElementConverter<PrometheusTimeSeries, Types.TimeSeries> elementConverter,
+ ElementConverter<IN, Types.TimeSeries> elementConverter,
int maxInFlightRequests,
int maxBufferedRequests,
int maxBatchSizeInSamples,
- int maxRecordSizeInSamples,
+ long maxRecordSizeInSamples,
Review Comment:
It's because I noticed that `maxBatchSizeInBytes` attribute in `AsyncSinkBase` class is Long and we have casted it from int to long in the super constructor of PrometheusSink. So, I thought we should match it to the super class, but I realised this is a breaking change that is not needed, so I have reverted it.
##########
flink-connector-prometheus/src/main/java/org/apache/flink/connector/prometheus/sink/PrometheusTimeSeriesConverter.java:
##########
@@ -22,42 +22,22 @@
import org.apache.flink.connector.base.sink.writer.ElementConverter;
import org.apache.flink.connector.prometheus.sink.prometheus.Types;
+import java.util.UnknownFormatConversionException;
+
/**
* Converts the sink input {@link PrometheusTimeSeries} into the Protobuf {@link Types.TimeSeries}
* that are sent to Prometheus.
*/
@PublicEvolving
-public class PrometheusTimeSeriesConverter
- implements ElementConverter<PrometheusTimeSeries, Types.TimeSeries> {
-
- private static final String METRIC_NAME_LABEL_NAME = "__name__";
+public class PrometheusTimeSeriesConverter<IN> implements ElementConverter<IN, Types.TimeSeries> {
@Override
- public Types.TimeSeries apply(PrometheusTimeSeries element, SinkWriter.Context context) {
- Types.TimeSeries.Builder builder =
- Types.TimeSeries.newBuilder()
- .addLabels(
- Types.Label.newBuilder()
- .setName(METRIC_NAME_LABEL_NAME)
- .setValue(element.getMetricName())
- .build());
-
- for (PrometheusTimeSeries.Label label : element.getLabels()) {
- builder.addLabels(
- Types.Label.newBuilder()
- .setName(label.getName())
- .setValue(label.getValue())
- .build());
- }
-
- for (PrometheusTimeSeries.Sample sample : element.getSamples()) {
- builder.addSamples(
- Types.Sample.newBuilder()
- .setValue(sample.getValue())
- .setTimestamp(sample.getTimestamp())
- .build());
+ public Types.TimeSeries apply(IN element, SinkWriter.Context context) {
Review Comment:
Answered in previous point: we will not introduce breaking change
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-33139] Add Prometheus Sink Table API [flink-connector-prometheus]
Posted by "darenwkt (via GitHub)" <gi...@apache.org>.
darenwkt commented on PR #22:
URL: https://github.com/apache/flink-connector-prometheus/pull/22#issuecomment-2848616292
> 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
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-33139] Add Prometheus Sink Table API [flink-connector-prometheus]
Posted by "georgelza (via GitHub)" <gi...@apache.org>.
georgelza commented on PR #22:
URL: https://github.com/apache/flink-connector-prometheus/pull/22#issuecomment-2931916009
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
> <https://github.com/apache/flink-connector-prometheus/pull/22#discussion_r2121788185>
> :
>
> > + 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
> <https://github.com/apache/flink-connector-prometheus/pull/22#discussion_r2121788185>,
> 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!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-33139] Add Prometheus Sink Table API [flink-connector-prometheus]
Posted by "nicusX (via GitHub)" <gi...@apache.org>.
nicusX commented on code in PR #22:
URL: https://github.com/apache/flink-connector-prometheus/pull/22#discussion_r2111971912
##########
flink-connector-prometheus-request-signer-amp/src/main/java/org/apache/flink/connector/prometheus/sink/aws/AmazonManagedPrometheusWriteRequestSignerFactory.java:
##########
@@ -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
Review Comment:
I'd be interested to see how this is used to set up the request signer declaratively in SQL
##########
flink-connector-prometheus/src/main/java/org/apache/flink/connector/prometheus/table/RowDataToPrometheusTimeSeriesConverter.java:
##########
@@ -0,0 +1,116 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.connector.prometheus.table;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.connector.prometheus.sink.PrometheusTimeSeries;
+import org.apache.flink.table.api.DataTypes;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.data.TimestampData;
+import org.apache.flink.table.types.DataType;
+
+import java.util.Arrays;
+import java.util.List;
+
+import static org.apache.flink.table.data.RowData.createFieldGetter;
+
+/**
+ * Converts from Flink Table API internal type of {@link RowData} to {@link PrometheusTimeSeries}.
+ */
+@Internal
+public class RowDataToPrometheusTimeSeriesConverter {
+
+ private final DataType physicalDataType;
+ private final PrometheusConfig prometheusConfig;
+
+ public RowDataToPrometheusTimeSeriesConverter(
+ DataType physicalDataType, PrometheusConfig prometheusConfig) {
+ this.physicalDataType = physicalDataType;
+ this.prometheusConfig = prometheusConfig;
+ }
+
+ public PrometheusTimeSeries convertRowData(RowData row) {
+ List<DataTypes.Field> fields = DataType.getFields(physicalDataType);
+
+ 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)) {
Review Comment:
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
##########
flink-connector-prometheus/src/main/java/org/apache/flink/connector/prometheus/table/RowDataToPrometheusTimeSeriesConverter.java:
##########
@@ -0,0 +1,116 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.connector.prometheus.table;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.connector.prometheus.sink.PrometheusTimeSeries;
+import org.apache.flink.table.api.DataTypes;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.data.TimestampData;
+import org.apache.flink.table.types.DataType;
+
+import java.util.Arrays;
+import java.util.List;
+
+import static org.apache.flink.table.data.RowData.createFieldGetter;
+
+/**
+ * Converts from Flink Table API internal type of {@link RowData} to {@link PrometheusTimeSeries}.
+ */
+@Internal
+public class RowDataToPrometheusTimeSeriesConverter {
+
+ private final DataType physicalDataType;
+ private final PrometheusConfig prometheusConfig;
+
+ public RowDataToPrometheusTimeSeriesConverter(
+ DataType physicalDataType, PrometheusConfig prometheusConfig) {
+ this.physicalDataType = physicalDataType;
+ this.prometheusConfig = prometheusConfig;
+ }
+
+ public PrometheusTimeSeries convertRowData(RowData row) {
+ List<DataTypes.Field> fields = DataType.getFields(physicalDataType);
+
+ 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)) {
+ builder.addLabel(fieldName, fieldValue.getStringValue());
+ } else if (fieldName.equals(prometheusConfig.getMetricSampleTimestamp())) {
+ sampleTimestamp = fieldValue.getLongValue();
+ } else {
+ throw new IllegalArgumentException(
Review Comment:
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.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-33139] Add Prometheus Sink Table API [flink-connector-prometheus]
Posted by "darenwkt (via GitHub)" <gi...@apache.org>.
darenwkt commented on code in PR #22:
URL: https://github.com/apache/flink-connector-prometheus/pull/22#discussion_r2125913263
##########
flink-connector-prometheus-request-signer-amp/src/main/java/org/apache/flink/connector/prometheus/sink/aws/AmazonManagedPrometheusWriteRequestSignerFactory.java:
##########
@@ -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
+ implements PrometheusDynamicRequestSignerFactory {
+ @Override
+ public String requestSignerIdentifer() {
+ return "amazon-managed-prometheus";
+ }
+
+ @Override
+ public PrometheusRequestSigner getRequestSigner(PrometheusConfig config) {
+ return new AmazonManagedPrometheusWriteRequestSigner(
+ config.getRemoteWriteEndpointUrl(),
+ getAwsRegionFromEndpointUrl(config.getRemoteWriteEndpointUrl()));
+ }
+
+ private String getAwsRegionFromEndpointUrl(String endpointUrl) {
+ String regex = "aps[a-zA-Z0-9-]*\\.([^.]+)\\.";
+ Pattern pattern = Pattern.compile(regex);
Review Comment:
Refactored to remove this method
##########
flink-connector-prometheus-request-signer-amp/src/main/java/org/apache/flink/connector/prometheus/sink/aws/AmazonManagedPrometheusWriteRequestSigner.java:
##########
@@ -98,6 +98,11 @@ private AwsCredentialsProvider getCredentialsProvider() {
return credentialsProvider;
}
+ @VisibleForTesting
+ String getAwsRegion() {
+ return awsRegion;
+ }
+
Review Comment:
Refactored to remove this method
##########
flink-connector-prometheus/src/main/java/org/apache/flink/connector/prometheus/sink/PrometheusSinkBase.java:
##########
@@ -0,0 +1,189 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.connector.prometheus.sink;
+
+import org.apache.flink.connector.base.sink.AsyncSinkBase;
+import org.apache.flink.connector.base.sink.writer.BufferedRequestState;
+import org.apache.flink.connector.base.sink.writer.ElementConverter;
+import org.apache.flink.connector.prometheus.sink.http.PrometheusAsyncHttpClientBuilder;
+import org.apache.flink.connector.prometheus.sink.metrics.SinkMetrics;
+import org.apache.flink.connector.prometheus.sink.metrics.SinkMetricsCallback;
+import org.apache.flink.connector.prometheus.sink.prometheus.Types;
+import org.apache.flink.core.io.SimpleVersionedSerializer;
+import org.apache.flink.util.Preconditions;
+
+import org.apache.commons.lang3.StringUtils;
+import org.apache.hc.client5.http.impl.async.CloseableHttpAsyncClient;
+
+import java.net.MalformedURLException;
+import java.net.URL;
+import java.util.Collection;
+
+public class PrometheusSinkBase<IN> extends AsyncSinkBase<IN, Types.TimeSeries> {
Review Comment:
Removed this class as we are using 1 `PrometheusSink`, verified it already has `@PublicEvolving` annotation
##########
flink-connector-prometheus/src/main/java/org/apache/flink/connector/prometheus/table/PrometheusDynamicSink.java:
##########
@@ -0,0 +1,185 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.connector.prometheus.table;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.connector.base.table.sink.AsyncDynamicTableSink;
+import org.apache.flink.connector.base.table.sink.AsyncDynamicTableSinkBuilder;
+import org.apache.flink.connector.prometheus.sink.PrometheusRequestSigner;
+import org.apache.flink.connector.prometheus.sink.PrometheusSinkBase;
+import org.apache.flink.connector.prometheus.sink.PrometheusSinkBaseBuilder;
+import org.apache.flink.connector.prometheus.sink.PrometheusSinkConfiguration;
+import org.apache.flink.connector.prometheus.sink.prometheus.Types;
+import org.apache.flink.table.connector.ChangelogMode;
+import org.apache.flink.table.connector.sink.DynamicTableSink;
+import org.apache.flink.table.connector.sink.SinkV2Provider;
+import org.apache.flink.table.connector.sink.abilities.SupportsPartitioning;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.types.DataType;
+import org.apache.flink.util.Preconditions;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.annotation.Nullable;
+
+import java.util.Map;
+import java.util.Optional;
+import java.util.ServiceLoader;
+import java.util.stream.Stream;
+
+public class PrometheusDynamicSink extends AsyncDynamicTableSink<Types.TimeSeries>
Review Comment:
I think this should be `@Internal` instead, I have updated it but let me know if I misunderstood it
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-33139] Add Prometheus Sink Table API [flink-connector-prometheus]
Posted by "darenwkt (via GitHub)" <gi...@apache.org>.
darenwkt commented on code in PR #22:
URL: https://github.com/apache/flink-connector-prometheus/pull/22#discussion_r2121085993
##########
flink-connector-prometheus-request-signer-amp/src/main/java/org/apache/flink/connector/prometheus/sink/aws/AmazonManagedPrometheusWriteRequestSignerFactory.java:
##########
@@ -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
Review Comment:
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'
);
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-33139] Add Prometheus Sink Table API [flink-connector-prometheus]
Posted by "darenwkt (via GitHub)" <gi...@apache.org>.
darenwkt commented on PR #22:
URL: https://github.com/apache/flink-connector-prometheus/pull/22#issuecomment-2887021161
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
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-33139] Add Prometheus Sink Table API [flink-connector-prometheus]
Posted by "georgelza (via GitHub)" <gi...@apache.org>.
georgelza commented on PR #22:
URL: https://github.com/apache/flink-connector-prometheus/pull/22#issuecomment-2953716553
@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.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-33139] Add Prometheus Sink Table API [flink-connector-prometheus]
Posted by "leekeiabstraction (via GitHub)" <gi...@apache.org>.
leekeiabstraction commented on code in PR #22:
URL: https://github.com/apache/flink-connector-prometheus/pull/22#discussion_r2042391665
##########
flink-connector-prometheus/src/main/java/org/apache/flink/connector/prometheus/sink/PrometheusSink.java:
##########
@@ -37,7 +37,7 @@
/** Sink implementation accepting {@link PrometheusTimeSeries} as inputs. */
@PublicEvolving
-public class PrometheusSink extends AsyncSinkBase<PrometheusTimeSeries, Types.TimeSeries> {
+public class PrometheusSink<InputT> extends AsyncSinkBase<InputT, Types.TimeSeries> {
Review Comment:
Also on name of type parameter, should we follow convention of using capital letter?
##########
flink-connector-prometheus/src/main/java/org/apache/flink/connector/prometheus/sink/PrometheusSink.java:
##########
@@ -37,7 +37,7 @@
/** Sink implementation accepting {@link PrometheusTimeSeries} as inputs. */
@PublicEvolving
-public class PrometheusSink extends AsyncSinkBase<PrometheusTimeSeries, Types.TimeSeries> {
+public class PrometheusSink<InputT> extends AsyncSinkBase<InputT, Types.TimeSeries> {
Review Comment:
Is this a breaking change? Would current code which has reference to PrometheusSink compile successfully?
##########
flink-connector-prometheus/src/main/java/org/apache/flink/connector/prometheus/sink/PrometheusTimeSeriesConverter.java:
##########
@@ -22,42 +22,23 @@
import org.apache.flink.connector.base.sink.writer.ElementConverter;
import org.apache.flink.connector.prometheus.sink.prometheus.Types;
+import java.util.UnknownFormatConversionException;
+
/**
* Converts the sink input {@link PrometheusTimeSeries} into the Protobuf {@link Types.TimeSeries}
* that are sent to Prometheus.
*/
@PublicEvolving
-public class PrometheusTimeSeriesConverter
- implements ElementConverter<PrometheusTimeSeries, Types.TimeSeries> {
-
- private static final String METRIC_NAME_LABEL_NAME = "__name__";
+public class PrometheusTimeSeriesConverter<InputT>
+ implements ElementConverter<InputT, Types.TimeSeries> {
Review Comment:
Similar comment on breaking change.
##########
flink-connector-prometheus/src/main/java/org/apache/flink/connector/prometheus/sink/PrometheusSinkBuilder.java:
##########
@@ -30,9 +30,8 @@
/** Builder for Sink implementation. */
@PublicEvolving
-public class PrometheusSinkBuilder
- extends AsyncSinkBaseBuilder<
- PrometheusTimeSeries, Types.TimeSeries, PrometheusSinkBuilder> {
+public class PrometheusSinkBuilder<InputT>
+ extends AsyncSinkBaseBuilder<InputT, Types.TimeSeries, PrometheusSinkBuilder<InputT>> {
Review Comment:
Similar comment on breaking change and also type parameter.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-33139] Add Prometheus Sink Table API [flink-connector-prometheus]
Posted by "darenwkt (via GitHub)" <gi...@apache.org>.
darenwkt commented on code in PR #22:
URL: https://github.com/apache/flink-connector-prometheus/pull/22#discussion_r2066049547
##########
flink-connector-prometheus-request-signer-amp/src/test/java/org/apache/flink/connector/prometheus/sink/aws/AmazonManagedPrometheusWriteRequestSignerFactoryTest.java:
##########
@@ -0,0 +1,55 @@
+package org.apache.flink.connector.prometheus.sink.aws;
+
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.connector.prometheus.sink.PrometheusRequestSigner;
+import org.apache.flink.connector.prometheus.table.PrometheusConfig;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import static org.apache.flink.connector.prometheus.table.PrometheusConnectorOption.METRIC_REMOTE_WRITE_URL;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertInstanceOf;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+
+class AmazonManagedPrometheusWriteRequestSignerFactoryTest {
+ private AmazonManagedPrometheusWriteRequestSignerFactory factory;
+
+ @BeforeEach
+ public void setup() {
+ factory = new AmazonManagedPrometheusWriteRequestSignerFactory();
+ }
+
+ @Test
+ void testIdentifier() {
+ assertEquals(factory.requestSignerIdentifer(), "amazon-managed-prometheus");
+ }
+
+ @Test
Review Comment:
Thanks, yes I have extracted all the endpoints from the doc and used them in the unit test
##########
flink-connector-prometheus-request-signer-amp/src/main/java/org/apache/flink/connector/prometheus/sink/aws/AmazonManagedPrometheusWriteRequestSignerFactory.java:
##########
@@ -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
+ implements PrometheusDynamicRequestSignerFactory {
+ @Override
+ public String requestSignerIdentifer() {
+ return "amazon-managed-prometheus";
+ }
+
+ @Override
+ public PrometheusRequestSigner getRequestSigner(PrometheusConfig config) {
+ return new AmazonManagedPrometheusWriteRequestSigner(
+ config.getRemoteWriteEndpointUrl(),
+ getAwsRegionFromEndpointUrl(config.getRemoteWriteEndpointUrl()));
+ }
+
+ private String getAwsRegionFromEndpointUrl(String endpointUrl) {
+ String regex = "aps-workspaces\\.([^.]+)\\.amazonaws\\.com";
Review Comment:
Thanks, this is a good catch, have updated this
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-33139] Add Prometheus Sink Table API [flink-connector-prometheus]
Posted by "darenwkt (via GitHub)" <gi...@apache.org>.
darenwkt commented on code in PR #22:
URL: https://github.com/apache/flink-connector-prometheus/pull/22#discussion_r2121085993
##########
flink-connector-prometheus-request-signer-amp/src/main/java/org/apache/flink/connector/prometheus/sink/aws/AmazonManagedPrometheusWriteRequestSignerFactory.java:
##########
@@ -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
Review Comment:
Request signer can be specified by user in the table config (`metric.request-signer`), for example:
```
CREATE TABLE PromTable (
`my_metric_name` BIGINT,
`my_label` 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',
'metric.sample.key' = 'sample_value',
'metric.sample.timestamp' = 'sample_ts',
'sink.batch.max-size' = '2',
'sink.flush-buffer.size' = '1'
);
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-33139] Add Prometheus Sink Table API [flink-connector-prometheus]
Posted by "darenwkt (via GitHub)" <gi...@apache.org>.
darenwkt commented on code in PR #22:
URL: https://github.com/apache/flink-connector-prometheus/pull/22#discussion_r2121788185
##########
flink-connector-prometheus/src/main/java/org/apache/flink/connector/prometheus/table/RowDataToPrometheusTimeSeriesConverter.java:
##########
@@ -0,0 +1,116 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.connector.prometheus.table;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.connector.prometheus.sink.PrometheusTimeSeries;
+import org.apache.flink.table.api.DataTypes;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.data.TimestampData;
+import org.apache.flink.table.types.DataType;
+
+import java.util.Arrays;
+import java.util.List;
+
+import static org.apache.flink.table.data.RowData.createFieldGetter;
+
+/**
+ * Converts from Flink Table API internal type of {@link RowData} to {@link PrometheusTimeSeries}.
+ */
+@Internal
+public class RowDataToPrometheusTimeSeriesConverter {
+
+ private final DataType physicalDataType;
+ private final PrometheusConfig prometheusConfig;
+
+ public RowDataToPrometheusTimeSeriesConverter(
+ DataType physicalDataType, PrometheusConfig prometheusConfig) {
+ this.physicalDataType = physicalDataType;
+ this.prometheusConfig = prometheusConfig;
+ }
+
+ public PrometheusTimeSeries convertRowData(RowData row) {
+ List<DataTypes.Field> fields = DataType.getFields(physicalDataType);
+
+ 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)) {
+ builder.addLabel(fieldName, fieldValue.getStringValue());
+ } else if (fieldName.equals(prometheusConfig.getMetricSampleTimestamp())) {
+ sampleTimestamp = fieldValue.getLongValue();
+ } else {
+ throw new IllegalArgumentException(
Review Comment:
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
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-33139] Add Prometheus Sink Table API [flink-connector-prometheus]
Posted by "darenwkt (via GitHub)" <gi...@apache.org>.
darenwkt commented on code in PR #22:
URL: https://github.com/apache/flink-connector-prometheus/pull/22#discussion_r2125915339
##########
flink-connector-prometheus/src/main/java/org/apache/flink/connector/prometheus/table/RowDataToPrometheusTimeSeriesConverter.java:
##########
@@ -0,0 +1,116 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.connector.prometheus.table;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.connector.prometheus.sink.PrometheusTimeSeries;
+import org.apache.flink.table.api.DataTypes;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.data.TimestampData;
+import org.apache.flink.table.types.DataType;
+
+import java.util.Arrays;
+import java.util.List;
+
+import static org.apache.flink.table.data.RowData.createFieldGetter;
+
+/**
+ * Converts from Flink Table API internal type of {@link RowData} to {@link PrometheusTimeSeries}.
+ */
+@Internal
+public class RowDataToPrometheusTimeSeriesConverter {
+
+ private final DataType physicalDataType;
+ private final PrometheusConfig prometheusConfig;
+
+ public RowDataToPrometheusTimeSeriesConverter(
+ DataType physicalDataType, PrometheusConfig prometheusConfig) {
+ this.physicalDataType = physicalDataType;
+ this.prometheusConfig = prometheusConfig;
+ }
+
+ public PrometheusTimeSeries convertRowData(RowData row) {
+ List<DataTypes.Field> fields = DataType.getFields(physicalDataType);
+
+ 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)) {
+ builder.addLabel(fieldName, fieldValue.getStringValue());
+ } else if (fieldName.equals(prometheusConfig.getMetricSampleTimestamp())) {
+ sampleTimestamp = fieldValue.getLongValue();
+ } else {
+ throw new IllegalArgumentException(
+ String.format(
+ "%s is not a supported field, valid fields are: %s",
Review Comment:
Thanks, yes I think it's better to ignore as well, that way user could specify columns that are not used by sink as well
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-33139] Add Prometheus Sink Table API [flink-connector-prometheus]
Posted by "darenwkt (via GitHub)" <gi...@apache.org>.
darenwkt commented on code in PR #22:
URL: https://github.com/apache/flink-connector-prometheus/pull/22#discussion_r2054497015
##########
flink-connector-prometheus/src/main/java/org/apache/flink/connector/prometheus/sink/PrometheusSink.java:
##########
@@ -37,7 +37,7 @@
/** Sink implementation accepting {@link PrometheusTimeSeries} as inputs. */
@PublicEvolving
-public class PrometheusSink extends AsyncSinkBase<PrometheusTimeSeries, Types.TimeSeries> {
+public class PrometheusSink<InputT> extends AsyncSinkBase<InputT, Types.TimeSeries> {
Review Comment:
> Is this a breaking change? Would current code which has reference to PrometheusSink compile successfully?
Yes, this is a breaking change, current code will not compile successfully as it is missing the InputType specification, I have updated the PR description with a breaking change summary to explain this in more details
> Also on name of type parameter, should we follow convention of using capital letter?
Thanks, this is a good shout, have renamed from "InputType" to "IN"
##########
flink-connector-prometheus/src/main/java/org/apache/flink/connector/prometheus/sink/PrometheusSinkBuilder.java:
##########
@@ -30,9 +30,8 @@
/** Builder for Sink implementation. */
@PublicEvolving
-public class PrometheusSinkBuilder
- extends AsyncSinkBaseBuilder<
- PrometheusTimeSeries, Types.TimeSeries, PrometheusSinkBuilder> {
+public class PrometheusSinkBuilder<InputT>
+ extends AsyncSinkBaseBuilder<InputT, Types.TimeSeries, PrometheusSinkBuilder<InputT>> {
Review Comment:
Addressed in PR description's breaking change summary
##########
flink-connector-prometheus/src/main/java/org/apache/flink/connector/prometheus/sink/PrometheusTimeSeriesConverter.java:
##########
@@ -22,42 +22,23 @@
import org.apache.flink.connector.base.sink.writer.ElementConverter;
import org.apache.flink.connector.prometheus.sink.prometheus.Types;
+import java.util.UnknownFormatConversionException;
+
/**
* Converts the sink input {@link PrometheusTimeSeries} into the Protobuf {@link Types.TimeSeries}
* that are sent to Prometheus.
*/
@PublicEvolving
-public class PrometheusTimeSeriesConverter
- implements ElementConverter<PrometheusTimeSeries, Types.TimeSeries> {
-
- private static final String METRIC_NAME_LABEL_NAME = "__name__";
+public class PrometheusTimeSeriesConverter<InputT>
+ implements ElementConverter<InputT, Types.TimeSeries> {
Review Comment:
Addressed in PR description's breaking change summary
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-33139] Add Prometheus Sink Table API [flink-connector-prometheus]
Posted by "darenwkt (via GitHub)" <gi...@apache.org>.
darenwkt commented on PR #22:
URL: https://github.com/apache/flink-connector-prometheus/pull/22#issuecomment-2823271088
Thanks @leekeiabstraction @foxus for the review, I have addressed the comments and updated the PR, please take a look
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-33139] Add Prometheus Sink Table API [flink-connector-prometheus]
Posted by "nicusX (via GitHub)" <gi...@apache.org>.
nicusX commented on PR #22:
URL: https://github.com/apache/flink-connector-prometheus/pull/22#issuecomment-2888387432
@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
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-33139] Add Prometheus Sink Table API [flink-connector-prometheus]
Posted by "georgelza (via GitHub)" <gi...@apache.org>.
georgelza commented on PR #22:
URL: https://github.com/apache/flink-connector-prometheus/pull/22#issuecomment-2872047349
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 ?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-33139] Add Prometheus Sink Table API [flink-connector-prometheus]
Posted by "foxus (via GitHub)" <gi...@apache.org>.
foxus commented on code in PR #22:
URL: https://github.com/apache/flink-connector-prometheus/pull/22#discussion_r2065830625
##########
flink-connector-prometheus-request-signer-amp/src/test/java/org/apache/flink/connector/prometheus/sink/aws/AmazonManagedPrometheusWriteRequestSignerFactoryTest.java:
##########
@@ -0,0 +1,55 @@
+package org.apache.flink.connector.prometheus.sink.aws;
+
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.connector.prometheus.sink.PrometheusRequestSigner;
+import org.apache.flink.connector.prometheus.table.PrometheusConfig;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import static org.apache.flink.connector.prometheus.table.PrometheusConnectorOption.METRIC_REMOTE_WRITE_URL;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertInstanceOf;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+
+class AmazonManagedPrometheusWriteRequestSignerFactoryTest {
+ private AmazonManagedPrometheusWriteRequestSignerFactory factory;
+
+ @BeforeEach
+ public void setup() {
+ factory = new AmazonManagedPrometheusWriteRequestSignerFactory();
+ }
+
+ @Test
+ void testIdentifier() {
+ assertEquals(factory.requestSignerIdentifer(), "amazon-managed-prometheus");
+ }
+
+ @Test
Review Comment:
Consider using a `DataProvider` to test all the endpoints listed in https://docs.aws.amazon.com/general/latest/gr/prometheus-service.html.
##########
flink-connector-prometheus-request-signer-amp/src/main/java/org/apache/flink/connector/prometheus/sink/aws/AmazonManagedPrometheusWriteRequestSignerFactory.java:
##########
@@ -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
+ implements PrometheusDynamicRequestSignerFactory {
+ @Override
+ public String requestSignerIdentifer() {
+ return "amazon-managed-prometheus";
+ }
+
+ @Override
+ public PrometheusRequestSigner getRequestSigner(PrometheusConfig config) {
+ return new AmazonManagedPrometheusWriteRequestSigner(
+ config.getRemoteWriteEndpointUrl(),
+ getAwsRegionFromEndpointUrl(config.getRemoteWriteEndpointUrl()));
+ }
+
+ private String getAwsRegionFromEndpointUrl(String endpointUrl) {
+ String regex = "aps-workspaces\\.([^.]+)\\.amazonaws\\.com";
Review Comment:
I believe this is too prescriptive - there are FIPS endpoints which this will not match - consider reviewing https://docs.aws.amazon.com/general/latest/gr/prometheus-service.html and making this regex less specific.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-33139] Add Prometheus Sink Table API [flink-connector-prometheus]
Posted by "darenwkt (via GitHub)" <gi...@apache.org>.
darenwkt commented on PR #22:
URL: https://github.com/apache/flink-connector-prometheus/pull/22#issuecomment-2838314070
> 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
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-33139] Add Prometheus Sink Table API [flink-connector-prometheus]
Posted by "georgelza (via GitHub)" <gi...@apache.org>.
georgelza commented on PR #22:
URL: https://github.com/apache/flink-connector-prometheus/pull/22#issuecomment-2887118000
> 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
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-33139] Add Prometheus Sink Table API [flink-connector-prometheus]
Posted by "darenwkt (via GitHub)" <gi...@apache.org>.
darenwkt commented on code in PR #22:
URL: https://github.com/apache/flink-connector-prometheus/pull/22#discussion_r2121153890
##########
flink-connector-prometheus/src/main/java/org/apache/flink/connector/prometheus/table/RowDataToPrometheusTimeSeriesConverter.java:
##########
@@ -0,0 +1,116 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.connector.prometheus.table;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.connector.prometheus.sink.PrometheusTimeSeries;
+import org.apache.flink.table.api.DataTypes;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.data.TimestampData;
+import org.apache.flink.table.types.DataType;
+
+import java.util.Arrays;
+import java.util.List;
+
+import static org.apache.flink.table.data.RowData.createFieldGetter;
+
+/**
+ * Converts from Flink Table API internal type of {@link RowData} to {@link PrometheusTimeSeries}.
+ */
+@Internal
+public class RowDataToPrometheusTimeSeriesConverter {
+
+ private final DataType physicalDataType;
+ private final PrometheusConfig prometheusConfig;
+
+ public RowDataToPrometheusTimeSeriesConverter(
+ DataType physicalDataType, PrometheusConfig prometheusConfig) {
+ this.physicalDataType = physicalDataType;
+ this.prometheusConfig = prometheusConfig;
+ }
+
+ public PrometheusTimeSeries convertRowData(RowData row) {
+ List<DataTypes.Field> fields = DataType.getFields(physicalDataType);
+
+ 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)) {
Review Comment:
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)]'
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-33139] Add Prometheus Sink Table API [flink-connector-prometheus]
Posted by "darenwkt (via GitHub)" <gi...@apache.org>.
darenwkt commented on code in PR #22:
URL: https://github.com/apache/flink-connector-prometheus/pull/22#discussion_r2125913882
##########
flink-connector-prometheus/src/main/java/org/apache/flink/connector/prometheus/sink/PrometheusSink.java:
##########
@@ -18,36 +18,13 @@
package org.apache.flink.connector.prometheus.sink;
import org.apache.flink.annotation.PublicEvolving;
-import org.apache.flink.connector.base.sink.AsyncSinkBase;
-import org.apache.flink.connector.base.sink.writer.BufferedRequestState;
import org.apache.flink.connector.base.sink.writer.ElementConverter;
import org.apache.flink.connector.prometheus.sink.http.PrometheusAsyncHttpClientBuilder;
-import org.apache.flink.connector.prometheus.sink.metrics.SinkMetrics;
-import org.apache.flink.connector.prometheus.sink.metrics.SinkMetricsCallback;
import org.apache.flink.connector.prometheus.sink.prometheus.Types;
-import org.apache.flink.core.io.SimpleVersionedSerializer;
-import org.apache.flink.util.Preconditions;
-
-import org.apache.commons.lang3.StringUtils;
-import org.apache.hc.client5.http.impl.async.CloseableHttpAsyncClient;
-
-import java.net.MalformedURLException;
-import java.net.URL;
-import java.util.Collection;
/** Sink implementation accepting {@link PrometheusTimeSeries} as inputs. */
@PublicEvolving
-public class PrometheusSink extends AsyncSinkBase<PrometheusTimeSeries, Types.TimeSeries> {
- private final String prometheusRemoteWriteUrl;
- private final PrometheusAsyncHttpClientBuilder clientBuilder;
- private final PrometheusRequestSigner requestSigner;
- private final int maxBatchSizeInSamples;
- private final String httpUserAgent;
- private final PrometheusSinkConfiguration.SinkWriterErrorHandlingBehaviorConfiguration
- errorHandlingBehaviorConfig;
- private final String metricGroupName;
-
- @SuppressWarnings("checkstyle:RegexpSingleline")
+public class PrometheusSink extends PrometheusSinkBase<PrometheusTimeSeries> {
Review Comment:
Yes, agree on a major version bump, the latest revision have removed the coupling, please take a look
##########
flink-connector-prometheus/src/main/java/org/apache/flink/connector/prometheus/sink/PrometheusSinkBase.java:
##########
@@ -0,0 +1,189 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.connector.prometheus.sink;
+
+import org.apache.flink.connector.base.sink.AsyncSinkBase;
+import org.apache.flink.connector.base.sink.writer.BufferedRequestState;
+import org.apache.flink.connector.base.sink.writer.ElementConverter;
+import org.apache.flink.connector.prometheus.sink.http.PrometheusAsyncHttpClientBuilder;
+import org.apache.flink.connector.prometheus.sink.metrics.SinkMetrics;
+import org.apache.flink.connector.prometheus.sink.metrics.SinkMetricsCallback;
+import org.apache.flink.connector.prometheus.sink.prometheus.Types;
+import org.apache.flink.core.io.SimpleVersionedSerializer;
+import org.apache.flink.util.Preconditions;
+
+import org.apache.commons.lang3.StringUtils;
+import org.apache.hc.client5.http.impl.async.CloseableHttpAsyncClient;
+
+import java.net.MalformedURLException;
+import java.net.URL;
+import java.util.Collection;
+
+public class PrometheusSinkBase<IN> extends AsyncSinkBase<IN, Types.TimeSeries> {
+ private final String prometheusRemoteWriteUrl;
+ private final PrometheusAsyncHttpClientBuilder clientBuilder;
+ private final PrometheusRequestSigner requestSigner;
+ private final int maxBatchSizeInSamples;
+ private final int maxRecordSizeInSamples;
+ private final String httpUserAgent;
+ private final PrometheusSinkConfiguration.SinkWriterErrorHandlingBehaviorConfiguration
+ errorHandlingBehaviorConfig;
+ private final String metricGroupName;
+
+ @SuppressWarnings("checkstyle:RegexpSingleline")
+ protected PrometheusSinkBase(
+ ElementConverter<IN, Types.TimeSeries> elementConverter,
+ int maxInFlightRequests,
+ int maxBufferedRequests,
+ int maxBatchSizeInSamples,
+ int maxRecordSizeInSamples,
+ long maxTimeInBufferMS,
+ String prometheusRemoteWriteUrl,
+ PrometheusAsyncHttpClientBuilder clientBuilder,
+ PrometheusRequestSigner requestSigner,
+ String httpUserAgent,
+ PrometheusSinkConfiguration.SinkWriterErrorHandlingBehaviorConfiguration
+ errorHandlingBehaviorConfig,
+ String metricGroupName) {
+ // This sink batches in terms of "samples", because writes to Prometheus are better
+ // optimized in terms of samples. AsyncSinkBase handles batching and does not make any
+ // assumptions about the actual unit of "size", but parameters are named assuming this unit
+ // is "bytes".
+ super(
+ elementConverter,
+ maxBatchSizeInSamples, // maxBatchSize,
+ maxInFlightRequests,
+ maxBufferedRequests,
+ maxBatchSizeInSamples, // maxBatchSizeInBytes,
+ maxTimeInBufferMS,
+ maxRecordSizeInSamples // maxRecordSizeInBytes
+ );
+
+ Preconditions.checkArgument(
+ maxBatchSizeInSamples > 1, "Max batch size (in samples) must be positive");
+ Preconditions.checkArgument(
+ maxRecordSizeInSamples <= maxBatchSizeInSamples,
+ "Max record size (in samples) must be <= Max batch size");
+ Preconditions.checkArgument(maxInFlightRequests == 1, "Max in-flight requests must be 1");
+ Preconditions.checkArgument(
+ StringUtils.isNotBlank(prometheusRemoteWriteUrl),
+ "Missing or blank Prometheus Remote-Write URL");
+ checkValidRemoteWriteUrl(prometheusRemoteWriteUrl);
+ Preconditions.checkArgument(
+ StringUtils.isNotBlank(httpUserAgent), "Missing HTTP User Agent string");
+ Preconditions.checkNotNull(
+ errorHandlingBehaviorConfig, "Missing error handling configuration");
+ Preconditions.checkArgument(
+ StringUtils.isNotBlank(metricGroupName), "Missing metric group name");
+ this.maxBatchSizeInSamples = maxBatchSizeInSamples;
+ this.maxRecordSizeInSamples = maxRecordSizeInSamples;
+ this.requestSigner = requestSigner;
+ this.prometheusRemoteWriteUrl = prometheusRemoteWriteUrl;
+ this.clientBuilder = clientBuilder;
+ this.httpUserAgent = httpUserAgent;
+ this.errorHandlingBehaviorConfig = errorHandlingBehaviorConfig;
+ this.metricGroupName = metricGroupName;
+ }
+
+ @Override
+ public StatefulSinkWriter<IN, BufferedRequestState<Types.TimeSeries>> createWriter(
+ InitContext initContext) {
+ SinkMetricsCallback metricsCallback =
+ new SinkMetricsCallback(
+ SinkMetrics.registerSinkMetrics(
+ initContext.metricGroup().addGroup(metricGroupName)));
+ CloseableHttpAsyncClient asyncHttpClient =
+ clientBuilder.buildAndStartClient(metricsCallback);
+
+ return new PrometheusSinkWriter<>(
+ getElementConverter(),
+ initContext,
+ getMaxInFlightRequests(),
+ getMaxBufferedRequests(),
+ maxBatchSizeInSamples,
+ maxRecordSizeInSamples,
+ getMaxTimeInBufferMS(),
+ prometheusRemoteWriteUrl,
+ asyncHttpClient,
+ metricsCallback,
+ requestSigner,
+ httpUserAgent,
+ errorHandlingBehaviorConfig);
+ }
+
+ @Override
+ public StatefulSinkWriter<IN, BufferedRequestState<Types.TimeSeries>> restoreWriter(
+ InitContext initContext,
+ Collection<BufferedRequestState<Types.TimeSeries>> recoveredState) {
+ SinkMetricsCallback metricsCallback =
+ new SinkMetricsCallback(
+ SinkMetrics.registerSinkMetrics(
+ initContext.metricGroup().addGroup(metricGroupName)));
+ CloseableHttpAsyncClient asyncHttpClient =
+ clientBuilder.buildAndStartClient(metricsCallback);
+ return new PrometheusSinkWriter<>(
+ getElementConverter(),
+ initContext,
+ getMaxInFlightRequests(),
+ getMaxBufferedRequests(),
+ maxBatchSizeInSamples,
+ maxRecordSizeInSamples,
+ getMaxTimeInBufferMS(),
+ prometheusRemoteWriteUrl,
+ asyncHttpClient,
+ metricsCallback,
+ requestSigner,
+ httpUserAgent,
+ errorHandlingBehaviorConfig,
+ recoveredState);
+ }
+
+ public PrometheusSink castToPrometheusSink() {
+ return new PrometheusSink(
+ new PrometheusTimeSeriesConverter(),
+ getMaxInFlightRequests(),
+ getMaxBufferedRequests(),
+ maxBatchSizeInSamples,
+ maxRecordSizeInSamples,
+ getMaxTimeInBufferMS(),
+ prometheusRemoteWriteUrl,
+ clientBuilder,
+ requestSigner,
+ httpUserAgent,
+ errorHandlingBehaviorConfig,
+ metricGroupName);
+ }
Review Comment:
Yes, agree on a major version bump, the latest revision have fixed the interface, please take a look
##########
flink-connector-prometheus/src/main/java/org/apache/flink/connector/prometheus/table/RowDataToPrometheusTimeSeriesConverter.java:
##########
@@ -0,0 +1,116 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.connector.prometheus.table;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.connector.prometheus.sink.PrometheusTimeSeries;
+import org.apache.flink.table.api.DataTypes;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.data.TimestampData;
+import org.apache.flink.table.types.DataType;
+
+import java.util.Arrays;
+import java.util.List;
+
+import static org.apache.flink.table.data.RowData.createFieldGetter;
+
+/**
+ * Converts from Flink Table API internal type of {@link RowData} to {@link PrometheusTimeSeries}.
+ */
+@Internal
+public class RowDataToPrometheusTimeSeriesConverter {
+
+ private final DataType physicalDataType;
+ private final PrometheusConfig prometheusConfig;
+
+ public RowDataToPrometheusTimeSeriesConverter(
+ DataType physicalDataType, PrometheusConfig prometheusConfig) {
+ this.physicalDataType = physicalDataType;
+ this.prometheusConfig = prometheusConfig;
+ }
+
+ public PrometheusTimeSeries convertRowData(RowData row) {
+ List<DataTypes.Field> fields = DataType.getFields(physicalDataType);
+
+ 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)) {
+ builder.addLabel(fieldName, fieldValue.getStringValue());
+ } else if (fieldName.equals(prometheusConfig.getMetricSampleTimestamp())) {
+ sampleTimestamp = fieldValue.getLongValue();
+ } else {
+ throw new IllegalArgumentException(
+ String.format(
+ "%s is not a supported field, valid fields are: %s",
+ fieldName,
+ Arrays.asList(
+ prometheusConfig.getMetricName(),
+ prometheusConfig.getLabelKeys(),
+ prometheusConfig.getMetricSampleKey(),
+ prometheusConfig.getMetricSampleTimestamp())));
+ }
+ }
+
+ if (sampleValue != null && sampleTimestamp != null) {
+ builder.addSample(sampleValue, sampleTimestamp);
+ } else {
+ throw new IllegalArgumentException(
+ String.format(
+ "Row is missing sampleValue field; %s or sampleTimestamp: %s",
+ sampleValue, sampleTimestamp));
+ }
+
+ return builder.build();
+ }
+
+ private static class FieldValue {
+ private final Object value;
+
+ private FieldValue(Object value) {
+ this.value = value;
+ }
+
+ private String getStringValue() {
+ return value.toString();
+ }
Review Comment:
This is a good call out, we only support `org.apache.flink.table.data.StringData` type here, have added validation for this in latest revision
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-33139] Add Prometheus Sink Table API [flink-connector-prometheus]
Posted by "darenwkt (via GitHub)" <gi...@apache.org>.
darenwkt commented on code in PR #22:
URL: https://github.com/apache/flink-connector-prometheus/pull/22#discussion_r2055384980
##########
flink-connector-prometheus/src/main/java/org/apache/flink/connector/prometheus/table/PrometheusDynamicSink.java:
##########
@@ -0,0 +1,193 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.connector.prometheus.table;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.connector.base.table.sink.AsyncDynamicTableSink;
+import org.apache.flink.connector.base.table.sink.AsyncDynamicTableSinkBuilder;
+import org.apache.flink.connector.prometheus.sink.PrometheusRequestSigner;
+import org.apache.flink.connector.prometheus.sink.PrometheusSink;
+import org.apache.flink.connector.prometheus.sink.PrometheusSinkBuilder;
+import org.apache.flink.connector.prometheus.sink.PrometheusSinkConfiguration;
+import org.apache.flink.connector.prometheus.sink.prometheus.Types;
+import org.apache.flink.table.connector.ChangelogMode;
+import org.apache.flink.table.connector.sink.DynamicTableSink;
+import org.apache.flink.table.connector.sink.SinkV2Provider;
+import org.apache.flink.table.connector.sink.abilities.SupportsPartitioning;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.types.DataType;
+import org.apache.flink.util.Preconditions;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.annotation.Nullable;
+
+import java.util.Iterator;
+import java.util.Map;
+import java.util.Optional;
+import java.util.ServiceConfigurationError;
+import java.util.ServiceLoader;
+
+public class PrometheusDynamicSink extends AsyncDynamicTableSink<Types.TimeSeries>
+ implements SupportsPartitioning {
+ private static final Logger LOG = LoggerFactory.getLogger(PrometheusDynamicSink.class);
+
+ /** Consumed data type of the table. */
+ private final DataType physicalDataType;
+
+ private final PrometheusConfig prometheusConfig;
+
+ protected PrometheusDynamicSink(
+ @Nullable Integer maxBatchSize,
+ @Nullable Integer maxInFlightRequests,
+ @Nullable Integer maxBufferedRequests,
+ @Nullable Long maxBufferSizeInBytes,
+ @Nullable Long maxTimeInBufferMS,
+ @Nullable DataType physicalDataType,
+ PrometheusConfig prometheusConfig) {
+
+ super(
+ maxBatchSize, // maxBatchSizeInSamples
+ maxInFlightRequests,
+ maxBufferedRequests,
+ maxBufferSizeInBytes, // maxRecordSizeInSamples
+ maxTimeInBufferMS);
+
+ this.physicalDataType =
+ Preconditions.checkNotNull(physicalDataType, "Consumed data type must not be null");
+ this.prometheusConfig = prometheusConfig;
+ }
+
+ @Override
+ public ChangelogMode getChangelogMode(ChangelogMode changelogMode) {
+ return ChangelogMode.upsert();
+ }
+
+ @Override
+ public SinkRuntimeProvider getSinkRuntimeProvider(Context context) {
+
+ PrometheusSinkBuilder<RowData> builder =
+ new PrometheusSinkBuilder<RowData>()
+ .setRetryConfiguration(
+ PrometheusSinkConfiguration.RetryConfiguration
+ .DEFAULT_RETRY_CONFIGURATION)
+ .setElementConverter(
+ new RowDataElementConverter(physicalDataType, prometheusConfig))
+ .setPrometheusRemoteWriteUrl(prometheusConfig.getRemoteWriteEndpointUrl());
+
+ Optional.ofNullable(prometheusConfig.getRequestSignerIdentifier())
+ .ifPresent(region -> builder.setRequestSigner(getRequestSigner(prometheusConfig)));
Review Comment:
I agree, on top of your point, moving it to getRequestSigner is better here especially since we want to separate any Amazon config away from flink-connector-prometheus package. Updated and added unit test for this
##########
flink-connector-prometheus/src/main/java/org/apache/flink/connector/prometheus/table/PrometheusDynamicSink.java:
##########
@@ -0,0 +1,193 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.connector.prometheus.table;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.connector.base.table.sink.AsyncDynamicTableSink;
+import org.apache.flink.connector.base.table.sink.AsyncDynamicTableSinkBuilder;
+import org.apache.flink.connector.prometheus.sink.PrometheusRequestSigner;
+import org.apache.flink.connector.prometheus.sink.PrometheusSink;
+import org.apache.flink.connector.prometheus.sink.PrometheusSinkBuilder;
+import org.apache.flink.connector.prometheus.sink.PrometheusSinkConfiguration;
+import org.apache.flink.connector.prometheus.sink.prometheus.Types;
+import org.apache.flink.table.connector.ChangelogMode;
+import org.apache.flink.table.connector.sink.DynamicTableSink;
+import org.apache.flink.table.connector.sink.SinkV2Provider;
+import org.apache.flink.table.connector.sink.abilities.SupportsPartitioning;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.types.DataType;
+import org.apache.flink.util.Preconditions;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.annotation.Nullable;
+
+import java.util.Iterator;
+import java.util.Map;
+import java.util.Optional;
+import java.util.ServiceConfigurationError;
+import java.util.ServiceLoader;
+
+public class PrometheusDynamicSink extends AsyncDynamicTableSink<Types.TimeSeries>
+ implements SupportsPartitioning {
+ private static final Logger LOG = LoggerFactory.getLogger(PrometheusDynamicSink.class);
+
+ /** Consumed data type of the table. */
+ private final DataType physicalDataType;
+
+ private final PrometheusConfig prometheusConfig;
+
+ protected PrometheusDynamicSink(
+ @Nullable Integer maxBatchSize,
+ @Nullable Integer maxInFlightRequests,
+ @Nullable Integer maxBufferedRequests,
+ @Nullable Long maxBufferSizeInBytes,
+ @Nullable Long maxTimeInBufferMS,
+ @Nullable DataType physicalDataType,
+ PrometheusConfig prometheusConfig) {
+
+ super(
+ maxBatchSize, // maxBatchSizeInSamples
+ maxInFlightRequests,
+ maxBufferedRequests,
+ maxBufferSizeInBytes, // maxRecordSizeInSamples
+ maxTimeInBufferMS);
+
+ this.physicalDataType =
+ Preconditions.checkNotNull(physicalDataType, "Consumed data type must not be null");
+ this.prometheusConfig = prometheusConfig;
+ }
+
+ @Override
+ public ChangelogMode getChangelogMode(ChangelogMode changelogMode) {
+ return ChangelogMode.upsert();
+ }
+
+ @Override
+ public SinkRuntimeProvider getSinkRuntimeProvider(Context context) {
+
+ PrometheusSinkBuilder<RowData> builder =
+ new PrometheusSinkBuilder<RowData>()
+ .setRetryConfiguration(
+ PrometheusSinkConfiguration.RetryConfiguration
+ .DEFAULT_RETRY_CONFIGURATION)
+ .setElementConverter(
+ new RowDataElementConverter(physicalDataType, prometheusConfig))
+ .setPrometheusRemoteWriteUrl(prometheusConfig.getRemoteWriteEndpointUrl());
+
+ Optional.ofNullable(prometheusConfig.getRequestSignerIdentifier())
+ .ifPresent(region -> builder.setRequestSigner(getRequestSigner(prometheusConfig)));
+ Optional.ofNullable(maxBatchSize).ifPresent(builder::setMaxBatchSizeInSamples);
+ Optional.ofNullable(maxBufferSizeInBytes).ifPresent(builder::setMaxRecordSizeInSamples);
+
+ PrometheusSink<RowData> prometheusSink = builder.build();
+ return SinkV2Provider.of(prometheusSink);
+ }
+
+ @Override
+ public DynamicTableSink copy() {
+ return new PrometheusDynamicSink(
+ maxBatchSize,
+ maxInFlightRequests,
+ maxBufferedRequests,
+ maxBufferSizeInBytes,
+ maxTimeInBufferMS,
+ physicalDataType,
+ prometheusConfig);
+ }
+
+ @Override
+ public String asSummaryString() {
+ return "Prometheus";
+ }
+
+ @Override
+ public void applyStaticPartition(Map<String, String> partitions) {
+ // We don't need to do anything here because the Prometheus sink handles a static partition
+ // just like a normal partition.
+ }
+
+ private PrometheusRequestSigner getRequestSigner(PrometheusConfig prometheusConfig) {
+ ServiceLoader<PrometheusDynamicRequestSignerFactory> loader =
+ ServiceLoader.load(PrometheusDynamicRequestSignerFactory.class);
+ Iterator<PrometheusDynamicRequestSignerFactory> factories = loader.iterator();
+ String requestSignerIdentifier = prometheusConfig.getRequestSignerIdentifier();
+
+ while (true) {
+ try {
+ if (!factories.hasNext()) {
+ break;
+ }
+
+ PrometheusDynamicRequestSignerFactory factory = factories.next();
Review Comment:
I have converted it to a Stream to make this more readable
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-33139] Add Prometheus Sink Table API [flink-connector-prometheus]
Posted by "darenwkt (via GitHub)" <gi...@apache.org>.
darenwkt commented on code in PR #22:
URL: https://github.com/apache/flink-connector-prometheus/pull/22#discussion_r2055385344
##########
flink-connector-prometheus/src/main/java/org/apache/flink/connector/prometheus/table/RowDataToPrometheusTimeSeriesConverter.java:
##########
@@ -0,0 +1,111 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.connector.prometheus.table;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.connector.prometheus.sink.PrometheusTimeSeries;
+import org.apache.flink.table.api.DataTypes;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.data.TimestampData;
+import org.apache.flink.table.types.DataType;
+
+import java.util.List;
+
+import static org.apache.flink.table.data.RowData.createFieldGetter;
+
+/**
+ * Converts from Flink Table API internal type of {@link RowData} to {@link PrometheusTimeSeries}.
+ */
+@Internal
+public class RowDataToPrometheusTimeSeriesConverter {
+
+ private final DataType physicalDataType;
+ private final PrometheusConfig prometheusConfig;
+
+ public RowDataToPrometheusTimeSeriesConverter(
+ DataType physicalDataType, PrometheusConfig prometheusConfig) {
+ this.physicalDataType = physicalDataType;
+ this.prometheusConfig = prometheusConfig;
+ }
+
+ public PrometheusTimeSeries convertRowData(RowData row) {
+ List<DataTypes.Field> fields = DataType.getFields(physicalDataType);
+
+ 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)) {
+ builder.addLabel(fieldName, fieldValue.getStringValue());
+ } else if (fieldName.equals(prometheusConfig.getMetricSampleTimestamp())) {
+ sampleTimestamp = fieldValue.getLongValue();
+ } else {
+ throw new IllegalArgumentException("Unsupported field: " + fieldName);
Review Comment:
Adopted suggestion
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-33139] Add Prometheus Sink Table API [flink-connector-prometheus]
Posted by "hlteoh37 (via GitHub)" <gi...@apache.org>.
hlteoh37 commented on code in PR #22:
URL: https://github.com/apache/flink-connector-prometheus/pull/22#discussion_r2124415562
##########
flink-connector-prometheus-request-signer-amp/src/main/java/org/apache/flink/connector/prometheus/sink/aws/AmazonManagedPrometheusWriteRequestSigner.java:
##########
@@ -98,6 +98,11 @@ private AwsCredentialsProvider getCredentialsProvider() {
return credentialsProvider;
}
+ @VisibleForTesting
+ String getAwsRegion() {
+ return awsRegion;
+ }
+
Review Comment:
This is a bit of a smell, since we are exposing internal object details to test the Factory method. Can we instead test the factory method directly?
##########
flink-connector-prometheus/src/main/java/org/apache/flink/connector/prometheus/sink/PrometheusSink.java:
##########
@@ -18,36 +18,13 @@
package org.apache.flink.connector.prometheus.sink;
import org.apache.flink.annotation.PublicEvolving;
-import org.apache.flink.connector.base.sink.AsyncSinkBase;
-import org.apache.flink.connector.base.sink.writer.BufferedRequestState;
import org.apache.flink.connector.base.sink.writer.ElementConverter;
import org.apache.flink.connector.prometheus.sink.http.PrometheusAsyncHttpClientBuilder;
-import org.apache.flink.connector.prometheus.sink.metrics.SinkMetrics;
-import org.apache.flink.connector.prometheus.sink.metrics.SinkMetricsCallback;
import org.apache.flink.connector.prometheus.sink.prometheus.Types;
-import org.apache.flink.core.io.SimpleVersionedSerializer;
-import org.apache.flink.util.Preconditions;
-
-import org.apache.commons.lang3.StringUtils;
-import org.apache.hc.client5.http.impl.async.CloseableHttpAsyncClient;
-
-import java.net.MalformedURLException;
-import java.net.URL;
-import java.util.Collection;
/** Sink implementation accepting {@link PrometheusTimeSeries} as inputs. */
@PublicEvolving
-public class PrometheusSink extends AsyncSinkBase<PrometheusTimeSeries, Types.TimeSeries> {
- private final String prometheusRemoteWriteUrl;
- private final PrometheusAsyncHttpClientBuilder clientBuilder;
- private final PrometheusRequestSigner requestSigner;
- private final int maxBatchSizeInSamples;
- private final String httpUserAgent;
- private final PrometheusSinkConfiguration.SinkWriterErrorHandlingBehaviorConfiguration
- errorHandlingBehaviorConfig;
- private final String metricGroupName;
-
- @SuppressWarnings("checkstyle:RegexpSingleline")
+public class PrometheusSink extends PrometheusSinkBase<PrometheusTimeSeries> {
Review Comment:
Why don't we take the hit and simply propose a major version bump to implement a generic `PrometheusSink`? That way we remove the need to maintain two coupled public interfaces (we see this coupling from the `castToPrometheusSink()` public method in the `PrometheusSinkBase`... which is a big smell! Since the parent class creates an instance of the child class...)
##########
flink-connector-prometheus/src/main/java/org/apache/flink/connector/prometheus/table/RowDataToPrometheusTimeSeriesConverter.java:
##########
@@ -0,0 +1,116 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.connector.prometheus.table;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.connector.prometheus.sink.PrometheusTimeSeries;
+import org.apache.flink.table.api.DataTypes;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.data.TimestampData;
+import org.apache.flink.table.types.DataType;
+
+import java.util.Arrays;
+import java.util.List;
+
+import static org.apache.flink.table.data.RowData.createFieldGetter;
+
+/**
+ * Converts from Flink Table API internal type of {@link RowData} to {@link PrometheusTimeSeries}.
+ */
+@Internal
+public class RowDataToPrometheusTimeSeriesConverter {
+
+ private final DataType physicalDataType;
+ private final PrometheusConfig prometheusConfig;
+
+ public RowDataToPrometheusTimeSeriesConverter(
+ DataType physicalDataType, PrometheusConfig prometheusConfig) {
+ this.physicalDataType = physicalDataType;
+ this.prometheusConfig = prometheusConfig;
+ }
+
+ public PrometheusTimeSeries convertRowData(RowData row) {
+ List<DataTypes.Field> fields = DataType.getFields(physicalDataType);
+
+ 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)) {
+ builder.addLabel(fieldName, fieldValue.getStringValue());
+ } else if (fieldName.equals(prometheusConfig.getMetricSampleTimestamp())) {
+ sampleTimestamp = fieldValue.getLongValue();
+ } else {
+ throw new IllegalArgumentException(
+ String.format(
+ "%s is not a supported field, valid fields are: %s",
Review Comment:
Is there a reason we don't just ignore the row in the `RowData` instead? e.g. user specifies a row in the table that is not used in the sink
##########
flink-connector-prometheus/src/main/java/org/apache/flink/connector/prometheus/table/RowDataToPrometheusTimeSeriesConverter.java:
##########
@@ -0,0 +1,116 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.connector.prometheus.table;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.connector.prometheus.sink.PrometheusTimeSeries;
+import org.apache.flink.table.api.DataTypes;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.data.TimestampData;
+import org.apache.flink.table.types.DataType;
+
+import java.util.Arrays;
+import java.util.List;
+
+import static org.apache.flink.table.data.RowData.createFieldGetter;
+
+/**
+ * Converts from Flink Table API internal type of {@link RowData} to {@link PrometheusTimeSeries}.
+ */
+@Internal
+public class RowDataToPrometheusTimeSeriesConverter {
+
+ private final DataType physicalDataType;
+ private final PrometheusConfig prometheusConfig;
+
+ public RowDataToPrometheusTimeSeriesConverter(
+ DataType physicalDataType, PrometheusConfig prometheusConfig) {
+ this.physicalDataType = physicalDataType;
+ this.prometheusConfig = prometheusConfig;
+ }
+
+ public PrometheusTimeSeries convertRowData(RowData row) {
+ List<DataTypes.Field> fields = DataType.getFields(physicalDataType);
+
+ 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)) {
+ builder.addLabel(fieldName, fieldValue.getStringValue());
+ } else if (fieldName.equals(prometheusConfig.getMetricSampleTimestamp())) {
+ sampleTimestamp = fieldValue.getLongValue();
+ } else {
+ throw new IllegalArgumentException(
+ String.format(
+ "%s is not a supported field, valid fields are: %s",
+ fieldName,
+ Arrays.asList(
+ prometheusConfig.getMetricName(),
+ prometheusConfig.getLabelKeys(),
+ prometheusConfig.getMetricSampleKey(),
+ prometheusConfig.getMetricSampleTimestamp())));
+ }
+ }
+
+ if (sampleValue != null && sampleTimestamp != null) {
+ builder.addSample(sampleValue, sampleTimestamp);
+ } else {
+ throw new IllegalArgumentException(
+ String.format(
+ "Row is missing sampleValue field; %s or sampleTimestamp: %s",
+ sampleValue, sampleTimestamp));
+ }
+
+ return builder.build();
+ }
+
+ private static class FieldValue {
+ private final Object value;
+
+ private FieldValue(Object value) {
+ this.value = value;
+ }
+
+ private String getStringValue() {
+ return value.toString();
+ }
Review Comment:
Are there any unsupported types? What happens if the field provided is a `MAP` / `ARRAY` ?
https://nightlies.apache.org/flink/flink-docs-master/docs/dev/table/types/#list-of-data-types
##########
flink-connector-prometheus-request-signer-amp/src/main/java/org/apache/flink/connector/prometheus/sink/aws/AmazonManagedPrometheusWriteRequestSignerFactory.java:
##########
@@ -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
+ implements PrometheusDynamicRequestSignerFactory {
+ @Override
+ public String requestSignerIdentifer() {
+ return "amazon-managed-prometheus";
+ }
+
+ @Override
+ public PrometheusRequestSigner getRequestSigner(PrometheusConfig config) {
+ return new AmazonManagedPrometheusWriteRequestSigner(
+ config.getRemoteWriteEndpointUrl(),
+ getAwsRegionFromEndpointUrl(config.getRemoteWriteEndpointUrl()));
+ }
+
+ private String getAwsRegionFromEndpointUrl(String endpointUrl) {
+ String regex = "aps[a-zA-Z0-9-]*\\.([^.]+)\\.";
+ Pattern pattern = Pattern.compile(regex);
Review Comment:
Let's make this static, instead of reinitialising on each request
##########
flink-connector-prometheus/src/main/java/org/apache/flink/connector/prometheus/sink/PrometheusSinkBase.java:
##########
@@ -0,0 +1,189 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.connector.prometheus.sink;
+
+import org.apache.flink.connector.base.sink.AsyncSinkBase;
+import org.apache.flink.connector.base.sink.writer.BufferedRequestState;
+import org.apache.flink.connector.base.sink.writer.ElementConverter;
+import org.apache.flink.connector.prometheus.sink.http.PrometheusAsyncHttpClientBuilder;
+import org.apache.flink.connector.prometheus.sink.metrics.SinkMetrics;
+import org.apache.flink.connector.prometheus.sink.metrics.SinkMetricsCallback;
+import org.apache.flink.connector.prometheus.sink.prometheus.Types;
+import org.apache.flink.core.io.SimpleVersionedSerializer;
+import org.apache.flink.util.Preconditions;
+
+import org.apache.commons.lang3.StringUtils;
+import org.apache.hc.client5.http.impl.async.CloseableHttpAsyncClient;
+
+import java.net.MalformedURLException;
+import java.net.URL;
+import java.util.Collection;
+
+public class PrometheusSinkBase<IN> extends AsyncSinkBase<IN, Types.TimeSeries> {
Review Comment:
This is `@PublicEvolving`. We will be extending the public interface for the sink repository 2x.
##########
flink-connector-prometheus/src/main/java/org/apache/flink/connector/prometheus/table/PrometheusDynamicSink.java:
##########
@@ -0,0 +1,185 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.connector.prometheus.table;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.connector.base.table.sink.AsyncDynamicTableSink;
+import org.apache.flink.connector.base.table.sink.AsyncDynamicTableSinkBuilder;
+import org.apache.flink.connector.prometheus.sink.PrometheusRequestSigner;
+import org.apache.flink.connector.prometheus.sink.PrometheusSinkBase;
+import org.apache.flink.connector.prometheus.sink.PrometheusSinkBaseBuilder;
+import org.apache.flink.connector.prometheus.sink.PrometheusSinkConfiguration;
+import org.apache.flink.connector.prometheus.sink.prometheus.Types;
+import org.apache.flink.table.connector.ChangelogMode;
+import org.apache.flink.table.connector.sink.DynamicTableSink;
+import org.apache.flink.table.connector.sink.SinkV2Provider;
+import org.apache.flink.table.connector.sink.abilities.SupportsPartitioning;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.types.DataType;
+import org.apache.flink.util.Preconditions;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.annotation.Nullable;
+
+import java.util.Map;
+import java.util.Optional;
+import java.util.ServiceLoader;
+import java.util.stream.Stream;
+
+public class PrometheusDynamicSink extends AsyncDynamicTableSink<Types.TimeSeries>
Review Comment:
This is `@PublicEvolving`.
##########
flink-connector-prometheus/src/main/java/org/apache/flink/connector/prometheus/sink/PrometheusSinkBase.java:
##########
@@ -0,0 +1,189 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.connector.prometheus.sink;
+
+import org.apache.flink.connector.base.sink.AsyncSinkBase;
+import org.apache.flink.connector.base.sink.writer.BufferedRequestState;
+import org.apache.flink.connector.base.sink.writer.ElementConverter;
+import org.apache.flink.connector.prometheus.sink.http.PrometheusAsyncHttpClientBuilder;
+import org.apache.flink.connector.prometheus.sink.metrics.SinkMetrics;
+import org.apache.flink.connector.prometheus.sink.metrics.SinkMetricsCallback;
+import org.apache.flink.connector.prometheus.sink.prometheus.Types;
+import org.apache.flink.core.io.SimpleVersionedSerializer;
+import org.apache.flink.util.Preconditions;
+
+import org.apache.commons.lang3.StringUtils;
+import org.apache.hc.client5.http.impl.async.CloseableHttpAsyncClient;
+
+import java.net.MalformedURLException;
+import java.net.URL;
+import java.util.Collection;
+
+public class PrometheusSinkBase<IN> extends AsyncSinkBase<IN, Types.TimeSeries> {
+ private final String prometheusRemoteWriteUrl;
+ private final PrometheusAsyncHttpClientBuilder clientBuilder;
+ private final PrometheusRequestSigner requestSigner;
+ private final int maxBatchSizeInSamples;
+ private final int maxRecordSizeInSamples;
+ private final String httpUserAgent;
+ private final PrometheusSinkConfiguration.SinkWriterErrorHandlingBehaviorConfiguration
+ errorHandlingBehaviorConfig;
+ private final String metricGroupName;
+
+ @SuppressWarnings("checkstyle:RegexpSingleline")
+ protected PrometheusSinkBase(
+ ElementConverter<IN, Types.TimeSeries> elementConverter,
+ int maxInFlightRequests,
+ int maxBufferedRequests,
+ int maxBatchSizeInSamples,
+ int maxRecordSizeInSamples,
+ long maxTimeInBufferMS,
+ String prometheusRemoteWriteUrl,
+ PrometheusAsyncHttpClientBuilder clientBuilder,
+ PrometheusRequestSigner requestSigner,
+ String httpUserAgent,
+ PrometheusSinkConfiguration.SinkWriterErrorHandlingBehaviorConfiguration
+ errorHandlingBehaviorConfig,
+ String metricGroupName) {
+ // This sink batches in terms of "samples", because writes to Prometheus are better
+ // optimized in terms of samples. AsyncSinkBase handles batching and does not make any
+ // assumptions about the actual unit of "size", but parameters are named assuming this unit
+ // is "bytes".
+ super(
+ elementConverter,
+ maxBatchSizeInSamples, // maxBatchSize,
+ maxInFlightRequests,
+ maxBufferedRequests,
+ maxBatchSizeInSamples, // maxBatchSizeInBytes,
+ maxTimeInBufferMS,
+ maxRecordSizeInSamples // maxRecordSizeInBytes
+ );
+
+ Preconditions.checkArgument(
+ maxBatchSizeInSamples > 1, "Max batch size (in samples) must be positive");
+ Preconditions.checkArgument(
+ maxRecordSizeInSamples <= maxBatchSizeInSamples,
+ "Max record size (in samples) must be <= Max batch size");
+ Preconditions.checkArgument(maxInFlightRequests == 1, "Max in-flight requests must be 1");
+ Preconditions.checkArgument(
+ StringUtils.isNotBlank(prometheusRemoteWriteUrl),
+ "Missing or blank Prometheus Remote-Write URL");
+ checkValidRemoteWriteUrl(prometheusRemoteWriteUrl);
+ Preconditions.checkArgument(
+ StringUtils.isNotBlank(httpUserAgent), "Missing HTTP User Agent string");
+ Preconditions.checkNotNull(
+ errorHandlingBehaviorConfig, "Missing error handling configuration");
+ Preconditions.checkArgument(
+ StringUtils.isNotBlank(metricGroupName), "Missing metric group name");
+ this.maxBatchSizeInSamples = maxBatchSizeInSamples;
+ this.maxRecordSizeInSamples = maxRecordSizeInSamples;
+ this.requestSigner = requestSigner;
+ this.prometheusRemoteWriteUrl = prometheusRemoteWriteUrl;
+ this.clientBuilder = clientBuilder;
+ this.httpUserAgent = httpUserAgent;
+ this.errorHandlingBehaviorConfig = errorHandlingBehaviorConfig;
+ this.metricGroupName = metricGroupName;
+ }
+
+ @Override
+ public StatefulSinkWriter<IN, BufferedRequestState<Types.TimeSeries>> createWriter(
+ InitContext initContext) {
+ SinkMetricsCallback metricsCallback =
+ new SinkMetricsCallback(
+ SinkMetrics.registerSinkMetrics(
+ initContext.metricGroup().addGroup(metricGroupName)));
+ CloseableHttpAsyncClient asyncHttpClient =
+ clientBuilder.buildAndStartClient(metricsCallback);
+
+ return new PrometheusSinkWriter<>(
+ getElementConverter(),
+ initContext,
+ getMaxInFlightRequests(),
+ getMaxBufferedRequests(),
+ maxBatchSizeInSamples,
+ maxRecordSizeInSamples,
+ getMaxTimeInBufferMS(),
+ prometheusRemoteWriteUrl,
+ asyncHttpClient,
+ metricsCallback,
+ requestSigner,
+ httpUserAgent,
+ errorHandlingBehaviorConfig);
+ }
+
+ @Override
+ public StatefulSinkWriter<IN, BufferedRequestState<Types.TimeSeries>> restoreWriter(
+ InitContext initContext,
+ Collection<BufferedRequestState<Types.TimeSeries>> recoveredState) {
+ SinkMetricsCallback metricsCallback =
+ new SinkMetricsCallback(
+ SinkMetrics.registerSinkMetrics(
+ initContext.metricGroup().addGroup(metricGroupName)));
+ CloseableHttpAsyncClient asyncHttpClient =
+ clientBuilder.buildAndStartClient(metricsCallback);
+ return new PrometheusSinkWriter<>(
+ getElementConverter(),
+ initContext,
+ getMaxInFlightRequests(),
+ getMaxBufferedRequests(),
+ maxBatchSizeInSamples,
+ maxRecordSizeInSamples,
+ getMaxTimeInBufferMS(),
+ prometheusRemoteWriteUrl,
+ asyncHttpClient,
+ metricsCallback,
+ requestSigner,
+ httpUserAgent,
+ errorHandlingBehaviorConfig,
+ recoveredState);
+ }
+
+ public PrometheusSink castToPrometheusSink() {
+ return new PrometheusSink(
+ new PrometheusTimeSeriesConverter(),
+ getMaxInFlightRequests(),
+ getMaxBufferedRequests(),
+ maxBatchSizeInSamples,
+ maxRecordSizeInSamples,
+ getMaxTimeInBufferMS(),
+ prometheusRemoteWriteUrl,
+ clientBuilder,
+ requestSigner,
+ httpUserAgent,
+ errorHandlingBehaviorConfig,
+ metricGroupName);
+ }
Review Comment:
The fact that we need this is a big code smell, since the higher level abstraction generates the lower level abstraction. Given we have just released the connector, let's consider instead "fixing" the interface, upgrading the major version to remove this problem?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-33139] Add Prometheus Sink Table API [flink-connector-prometheus]
Posted by "darenwkt (via GitHub)" <gi...@apache.org>.
darenwkt commented on code in PR #22:
URL: https://github.com/apache/flink-connector-prometheus/pull/22#discussion_r2121085993
##########
flink-connector-prometheus-request-signer-amp/src/main/java/org/apache/flink/connector/prometheus/sink/aws/AmazonManagedPrometheusWriteRequestSignerFactory.java:
##########
@@ -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
Review Comment:
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` 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'
);
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-33139] Add Prometheus Sink Table API [flink-connector-prometheus]
Posted by "nicusX (via GitHub)" <gi...@apache.org>.
nicusX commented on code in PR #22:
URL: https://github.com/apache/flink-connector-prometheus/pull/22#discussion_r2121707551
##########
flink-connector-prometheus/src/main/java/org/apache/flink/connector/prometheus/table/RowDataToPrometheusTimeSeriesConverter.java:
##########
@@ -0,0 +1,116 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.connector.prometheus.table;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.connector.prometheus.sink.PrometheusTimeSeries;
+import org.apache.flink.table.api.DataTypes;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.data.TimestampData;
+import org.apache.flink.table.types.DataType;
+
+import java.util.Arrays;
+import java.util.List;
+
+import static org.apache.flink.table.data.RowData.createFieldGetter;
+
+/**
+ * Converts from Flink Table API internal type of {@link RowData} to {@link PrometheusTimeSeries}.
+ */
+@Internal
+public class RowDataToPrometheusTimeSeriesConverter {
+
+ private final DataType physicalDataType;
+ private final PrometheusConfig prometheusConfig;
+
+ public RowDataToPrometheusTimeSeriesConverter(
+ DataType physicalDataType, PrometheusConfig prometheusConfig) {
+ this.physicalDataType = physicalDataType;
+ this.prometheusConfig = prometheusConfig;
+ }
+
+ public PrometheusTimeSeries convertRowData(RowData row) {
+ List<DataTypes.Field> fields = DataType.getFields(physicalDataType);
+
+ 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)) {
Review Comment:
"LABEL@123", "123%label" and "label(123)" would be illegal Label's key (names) for Prometheus BTW (see [specs](https://prometheus.io/docs/specs/prw/remote_write_spec/#labels)). 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.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-33139] Add Prometheus Sink Table API [flink-connector-prometheus]
Posted by "foxus (via GitHub)" <gi...@apache.org>.
foxus commented on code in PR #22:
URL: https://github.com/apache/flink-connector-prometheus/pull/22#discussion_r2044253548
##########
flink-connector-prometheus/src/main/java/org/apache/flink/connector/prometheus/table/PrometheusConnectorOption.java:
##########
@@ -0,0 +1,75 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.connector.prometheus.table;
+
+import org.apache.flink.configuration.ConfigOption;
+import org.apache.flink.configuration.ConfigOptions;
+
+public class PrometheusConnectorOption {
+
+ // -----------------------------------------------------------------------------------------
+ // Prometheus connector specific options
+ // -----------------------------------------------------------------------------------------
+
+ public static final ConfigOption<String> METRIC_NAME =
+ ConfigOptions.key("metric.name")
+ .stringType()
+ .noDefaultValue()
+ .withDescription("Prometheus metric name.");
+
+ public static final ConfigOption<String> METRIC_LABEL_KEYS =
Review Comment:
Related to previous comment, `ConfigOption` has native support for CSVs -
```
ConfigOption<String> METRIC_LABEL_KEYS = ConfigOptions
.key("metric.label.keys")
.stringType()
.asList()
.noDefaultValue()
.withDescription("Prometheus metric label key name.");
```
You could then eliminate the streaming and hardcoded delimiter work above.
##########
flink-connector-prometheus/src/main/java/org/apache/flink/connector/prometheus/table/PrometheusDynamicSink.java:
##########
@@ -0,0 +1,193 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.connector.prometheus.table;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.connector.base.table.sink.AsyncDynamicTableSink;
+import org.apache.flink.connector.base.table.sink.AsyncDynamicTableSinkBuilder;
+import org.apache.flink.connector.prometheus.sink.PrometheusRequestSigner;
+import org.apache.flink.connector.prometheus.sink.PrometheusSink;
+import org.apache.flink.connector.prometheus.sink.PrometheusSinkBuilder;
+import org.apache.flink.connector.prometheus.sink.PrometheusSinkConfiguration;
+import org.apache.flink.connector.prometheus.sink.prometheus.Types;
+import org.apache.flink.table.connector.ChangelogMode;
+import org.apache.flink.table.connector.sink.DynamicTableSink;
+import org.apache.flink.table.connector.sink.SinkV2Provider;
+import org.apache.flink.table.connector.sink.abilities.SupportsPartitioning;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.types.DataType;
+import org.apache.flink.util.Preconditions;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.annotation.Nullable;
+
+import java.util.Iterator;
+import java.util.Map;
+import java.util.Optional;
+import java.util.ServiceConfigurationError;
+import java.util.ServiceLoader;
+
+public class PrometheusDynamicSink extends AsyncDynamicTableSink<Types.TimeSeries>
+ implements SupportsPartitioning {
+ private static final Logger LOG = LoggerFactory.getLogger(PrometheusDynamicSink.class);
+
+ /** Consumed data type of the table. */
+ private final DataType physicalDataType;
+
+ private final PrometheusConfig prometheusConfig;
+
+ protected PrometheusDynamicSink(
+ @Nullable Integer maxBatchSize,
+ @Nullable Integer maxInFlightRequests,
+ @Nullable Integer maxBufferedRequests,
+ @Nullable Long maxBufferSizeInBytes,
+ @Nullable Long maxTimeInBufferMS,
+ @Nullable DataType physicalDataType,
+ PrometheusConfig prometheusConfig) {
+
+ super(
+ maxBatchSize, // maxBatchSizeInSamples
+ maxInFlightRequests,
+ maxBufferedRequests,
+ maxBufferSizeInBytes, // maxRecordSizeInSamples
+ maxTimeInBufferMS);
+
+ this.physicalDataType =
+ Preconditions.checkNotNull(physicalDataType, "Consumed data type must not be null");
+ this.prometheusConfig = prometheusConfig;
+ }
+
+ @Override
+ public ChangelogMode getChangelogMode(ChangelogMode changelogMode) {
+ return ChangelogMode.upsert();
+ }
+
+ @Override
+ public SinkRuntimeProvider getSinkRuntimeProvider(Context context) {
+
+ PrometheusSinkBuilder<RowData> builder =
+ new PrometheusSinkBuilder<RowData>()
+ .setRetryConfiguration(
+ PrometheusSinkConfiguration.RetryConfiguration
+ .DEFAULT_RETRY_CONFIGURATION)
+ .setElementConverter(
+ new RowDataElementConverter(physicalDataType, prometheusConfig))
+ .setPrometheusRemoteWriteUrl(prometheusConfig.getRemoteWriteEndpointUrl());
+
+ Optional.ofNullable(prometheusConfig.getRequestSignerIdentifier())
+ .ifPresent(region -> builder.setRequestSigner(getRequestSigner(prometheusConfig)));
+ Optional.ofNullable(maxBatchSize).ifPresent(builder::setMaxBatchSizeInSamples);
+ Optional.ofNullable(maxBufferSizeInBytes).ifPresent(builder::setMaxRecordSizeInSamples);
+
+ PrometheusSink<RowData> prometheusSink = builder.build();
+ return SinkV2Provider.of(prometheusSink);
+ }
+
+ @Override
+ public DynamicTableSink copy() {
+ return new PrometheusDynamicSink(
+ maxBatchSize,
+ maxInFlightRequests,
+ maxBufferedRequests,
+ maxBufferSizeInBytes,
+ maxTimeInBufferMS,
+ physicalDataType,
+ prometheusConfig);
+ }
+
+ @Override
+ public String asSummaryString() {
+ return "Prometheus";
+ }
+
+ @Override
+ public void applyStaticPartition(Map<String, String> partitions) {
+ // We don't need to do anything here because the Prometheus sink handles a static partition
+ // just like a normal partition.
+ }
+
+ private PrometheusRequestSigner getRequestSigner(PrometheusConfig prometheusConfig) {
+ ServiceLoader<PrometheusDynamicRequestSignerFactory> loader =
+ ServiceLoader.load(PrometheusDynamicRequestSignerFactory.class);
+ Iterator<PrometheusDynamicRequestSignerFactory> factories = loader.iterator();
+ String requestSignerIdentifier = prometheusConfig.getRequestSignerIdentifier();
+
+ while (true) {
+ try {
+ if (!factories.hasNext()) {
+ break;
+ }
+
+ PrometheusDynamicRequestSignerFactory factory = factories.next();
Review Comment:
Would this be simpler as a for loop as well as incorporating line 129?
```
for(Iterator<PrometheusDynamicRequestSignerFactory> factories = loader.iterator(); factories.hasNext(); ) {
PrometheusDynamicRequestSignerFactory factory = factories.next();
```
##########
flink-connector-prometheus/src/main/java/org/apache/flink/connector/prometheus/table/RowDataToPrometheusTimeSeriesConverter.java:
##########
@@ -0,0 +1,111 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.connector.prometheus.table;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.connector.prometheus.sink.PrometheusTimeSeries;
+import org.apache.flink.table.api.DataTypes;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.data.TimestampData;
+import org.apache.flink.table.types.DataType;
+
+import java.util.List;
+
+import static org.apache.flink.table.data.RowData.createFieldGetter;
+
+/**
+ * Converts from Flink Table API internal type of {@link RowData} to {@link PrometheusTimeSeries}.
+ */
+@Internal
+public class RowDataToPrometheusTimeSeriesConverter {
+
+ private final DataType physicalDataType;
+ private final PrometheusConfig prometheusConfig;
+
+ public RowDataToPrometheusTimeSeriesConverter(
+ DataType physicalDataType, PrometheusConfig prometheusConfig) {
+ this.physicalDataType = physicalDataType;
+ this.prometheusConfig = prometheusConfig;
+ }
+
+ public PrometheusTimeSeries convertRowData(RowData row) {
+ List<DataTypes.Field> fields = DataType.getFields(physicalDataType);
+
+ 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)) {
+ builder.addLabel(fieldName, fieldValue.getStringValue());
+ } else if (fieldName.equals(prometheusConfig.getMetricSampleTimestamp())) {
+ sampleTimestamp = fieldValue.getLongValue();
+ } else {
+ throw new IllegalArgumentException("Unsupported field: " + fieldName);
Review Comment:
Could we consider providing the list of supported fields in this exception message?
##########
flink-connector-prometheus/src/main/java/org/apache/flink/connector/prometheus/table/PrometheusConfig.java:
##########
@@ -0,0 +1,75 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.connector.prometheus.table;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.configuration.ReadableConfig;
+
+import java.io.Serializable;
+import java.util.Arrays;
+import java.util.List;
+import java.util.stream.Collectors;
+
+import static org.apache.flink.connector.prometheus.table.PrometheusConnectorOption.AWS_REGION;
+import static org.apache.flink.connector.prometheus.table.PrometheusConnectorOption.METRIC_LABEL_KEYS;
+import static org.apache.flink.connector.prometheus.table.PrometheusConnectorOption.METRIC_NAME;
+import static org.apache.flink.connector.prometheus.table.PrometheusConnectorOption.METRIC_REMOTE_WRITE_URL;
+import static org.apache.flink.connector.prometheus.table.PrometheusConnectorOption.METRIC_REQUEST_SIGNER;
+import static org.apache.flink.connector.prometheus.table.PrometheusConnectorOption.METRIC_SAMPLE_KEY;
+import static org.apache.flink.connector.prometheus.table.PrometheusConnectorOption.METRIC_SAMPLE_TIMESTAMP;
+
+/** Prometheus specific configuration. */
+@Internal
+public class PrometheusConfig implements Serializable {
+ private final ReadableConfig options;
+
+ private static final String DELIMITER = ",";
+
+ public PrometheusConfig(ReadableConfig options) {
+ this.options = options;
+ }
+
+ public String getMetricName() {
+ return options.get(METRIC_NAME);
+ }
+
+ public List<String> getLabelKeys() {
Review Comment:
Are `labelKeys` unique? If so, consider returning a set rather than a list to enforce uniqueness, alternatively consider calling unique after the split. This will protect against a user supplied configuration including duplicates and also ensure that a serialized Set is identical regardless of the order of keys provided.
##########
flink-connector-prometheus/src/main/java/org/apache/flink/connector/prometheus/table/PrometheusConnectorOption.java:
##########
@@ -0,0 +1,75 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.connector.prometheus.table;
+
+import org.apache.flink.configuration.ConfigOption;
+import org.apache.flink.configuration.ConfigOptions;
+
+public class PrometheusConnectorOption {
+
+ // -----------------------------------------------------------------------------------------
+ // Prometheus connector specific options
+ // -----------------------------------------------------------------------------------------
+
+ public static final ConfigOption<String> METRIC_NAME =
+ ConfigOptions.key("metric.name")
+ .stringType()
+ .noDefaultValue()
+ .withDescription("Prometheus metric name.");
+
+ public static final ConfigOption<String> METRIC_LABEL_KEYS =
Review Comment:
I would also consider rewording the description to make it clear there can be more than one key name provided.
##########
flink-connector-prometheus/src/main/java/org/apache/flink/connector/prometheus/table/PrometheusDynamicSinkFactory.java:
##########
@@ -0,0 +1,165 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.connector.prometheus.table;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.configuration.ConfigOption;
+import org.apache.flink.configuration.ReadableConfig;
+import org.apache.flink.connector.base.table.AsyncDynamicTableSinkFactory;
+import org.apache.flink.connector.base.table.AsyncSinkConnectorOptions;
+import org.apache.flink.table.api.DataTypes;
+import org.apache.flink.table.api.ValidationException;
+import org.apache.flink.table.catalog.ResolvedCatalogTable;
+import org.apache.flink.table.connector.sink.DynamicTableSink;
+import org.apache.flink.table.factories.FactoryUtil;
+import org.apache.flink.table.types.DataType;
+
+import java.util.HashSet;
+import java.util.List;
+import java.util.Optional;
+import java.util.Properties;
+import java.util.Set;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+
+import static org.apache.flink.connector.prometheus.table.PrometheusConnectorOption.AWS_REGION;
+import static org.apache.flink.connector.prometheus.table.PrometheusConnectorOption.METRIC_LABEL_KEYS;
+import static org.apache.flink.connector.prometheus.table.PrometheusConnectorOption.METRIC_NAME;
+import static org.apache.flink.connector.prometheus.table.PrometheusConnectorOption.METRIC_REMOTE_WRITE_URL;
+import static org.apache.flink.connector.prometheus.table.PrometheusConnectorOption.METRIC_REQUEST_SIGNER;
+import static org.apache.flink.connector.prometheus.table.PrometheusConnectorOption.METRIC_SAMPLE_KEY;
+import static org.apache.flink.connector.prometheus.table.PrometheusConnectorOption.METRIC_SAMPLE_TIMESTAMP;
+
+/** Factory for creating {@link PrometheusDynamicSink}. */
+@Internal
+public class PrometheusDynamicSinkFactory extends AsyncDynamicTableSinkFactory {
+ public static final String FACTORY_IDENTIFIER = "prometheus";
+
+ @Override
+ public DynamicTableSink createDynamicTableSink(Context context) {
+ FactoryUtil.TableFactoryHelper factoryHelper =
+ FactoryUtil.createTableFactoryHelper(this, context);
+ ResolvedCatalogTable catalogTable = context.getCatalogTable();
+
+ FactoryUtil.validateFactoryOptions(this, factoryHelper.getOptions());
+
+ DataType physicalDataType = catalogTable.getResolvedSchema().toPhysicalRowDataType();
+ PrometheusConfig prometheusConfig = new PrometheusConfig(factoryHelper.getOptions());
+ validateMetricConfigKeys(prometheusConfig, physicalDataType);
+
+ PrometheusDynamicSink.PrometheusDynamicSinkBuilder builder =
+ PrometheusDynamicSink.builder()
+ .setPhysicalDataType(physicalDataType)
+ .setPrometheusConfiguration(prometheusConfig);
+
+ addAsyncOptionsToBuilder(getAsyncSinkOptions(factoryHelper.getOptions()), builder);
+
+ return builder.build();
+ }
+
+ @Override
+ public String factoryIdentifier() {
+ return FACTORY_IDENTIFIER;
+ }
+
+ @Override
+ public Set<ConfigOption<?>> requiredOptions() {
+ final Set<ConfigOption<?>> requiredOptions = new HashSet<>();
+
+ requiredOptions.add(METRIC_NAME);
+ requiredOptions.add(METRIC_LABEL_KEYS);
+ requiredOptions.add(METRIC_SAMPLE_KEY);
+ requiredOptions.add(METRIC_SAMPLE_TIMESTAMP);
+ requiredOptions.add(METRIC_REMOTE_WRITE_URL);
+
+ return requiredOptions;
+ }
+
+ @Override
+ public Set<ConfigOption<?>> optionalOptions() {
+ final Set<ConfigOption<?>> optionalOptions = new HashSet<>();
+
+ optionalOptions.add(AWS_REGION);
+ optionalOptions.add(METRIC_REQUEST_SIGNER);
+
+ return optionalOptions;
+ }
+
+ private Properties getAsyncSinkOptions(ReadableConfig config) {
+ Properties properties = new Properties();
+ Optional.ofNullable(config.get(AsyncSinkConnectorOptions.FLUSH_BUFFER_SIZE))
+ .ifPresent(
+ flushBufferSize ->
+ properties.put(
+ AsyncSinkConnectorOptions.FLUSH_BUFFER_SIZE.key(),
+ flushBufferSize));
+ Optional.ofNullable(config.get(AsyncSinkConnectorOptions.MAX_BATCH_SIZE))
+ .ifPresent(
+ maxBatchSize ->
+ properties.put(
+ AsyncSinkConnectorOptions.MAX_BATCH_SIZE.key(),
+ maxBatchSize));
+ Optional.ofNullable(config.get(AsyncSinkConnectorOptions.MAX_IN_FLIGHT_REQUESTS))
+ .ifPresent(
+ maxInflightRequests ->
+ properties.put(
+ AsyncSinkConnectorOptions.MAX_IN_FLIGHT_REQUESTS.key(),
+ maxInflightRequests));
+ Optional.ofNullable(config.get(AsyncSinkConnectorOptions.MAX_BUFFERED_REQUESTS))
+ .ifPresent(
+ maxBufferedRequests ->
+ properties.put(
+ AsyncSinkConnectorOptions.MAX_BUFFERED_REQUESTS.key(),
+ maxBufferedRequests));
+ Optional.ofNullable(config.get(AsyncSinkConnectorOptions.FLUSH_BUFFER_TIMEOUT))
+ .ifPresent(
+ timeout ->
+ properties.put(
+ AsyncSinkConnectorOptions.FLUSH_BUFFER_TIMEOUT.key(),
+ timeout));
+ return properties;
+ }
+
+ private void validateMetricConfigKeys(
+ PrometheusConfig prometheusConfig, DataType physicalDataType) {
+ Set<String> fieldNames =
+ DataType.getFields(physicalDataType).stream()
+ .map(DataTypes.AbstractField::getName)
+ .collect(Collectors.toSet());
+
+ List<String> metricConfigKeys =
+ Stream.concat(
+ Stream.of(
+ prometheusConfig.getMetricName(),
+ prometheusConfig.getMetricSampleKey(),
+ prometheusConfig.getMetricSampleTimestamp()),
+ prometheusConfig.getLabelKeys().stream())
+ .collect(Collectors.toList());
+
+ metricConfigKeys.forEach(
+ metricConfigKey ->
+ Optional.ofNullable(metricConfigKey)
+ .filter(fieldNames::contains)
+ .orElseThrow(
+ () ->
+ new ValidationException(
+ String.format(
+ "%s not found in fields: %s",
Review Comment:
Could we consider something along the lines of `%s is not a valid Prometheus metric config key, valid keys are: %s`, this message is exposing the implementation of the validation rather than what the user should do to fix it.
##########
flink-connector-prometheus/src/main/java/org/apache/flink/connector/prometheus/table/PrometheusDynamicSink.java:
##########
@@ -0,0 +1,193 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.connector.prometheus.table;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.connector.base.table.sink.AsyncDynamicTableSink;
+import org.apache.flink.connector.base.table.sink.AsyncDynamicTableSinkBuilder;
+import org.apache.flink.connector.prometheus.sink.PrometheusRequestSigner;
+import org.apache.flink.connector.prometheus.sink.PrometheusSink;
+import org.apache.flink.connector.prometheus.sink.PrometheusSinkBuilder;
+import org.apache.flink.connector.prometheus.sink.PrometheusSinkConfiguration;
+import org.apache.flink.connector.prometheus.sink.prometheus.Types;
+import org.apache.flink.table.connector.ChangelogMode;
+import org.apache.flink.table.connector.sink.DynamicTableSink;
+import org.apache.flink.table.connector.sink.SinkV2Provider;
+import org.apache.flink.table.connector.sink.abilities.SupportsPartitioning;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.types.DataType;
+import org.apache.flink.util.Preconditions;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.annotation.Nullable;
+
+import java.util.Iterator;
+import java.util.Map;
+import java.util.Optional;
+import java.util.ServiceConfigurationError;
+import java.util.ServiceLoader;
+
+public class PrometheusDynamicSink extends AsyncDynamicTableSink<Types.TimeSeries>
+ implements SupportsPartitioning {
+ private static final Logger LOG = LoggerFactory.getLogger(PrometheusDynamicSink.class);
+
+ /** Consumed data type of the table. */
+ private final DataType physicalDataType;
+
+ private final PrometheusConfig prometheusConfig;
+
+ protected PrometheusDynamicSink(
+ @Nullable Integer maxBatchSize,
+ @Nullable Integer maxInFlightRequests,
+ @Nullable Integer maxBufferedRequests,
+ @Nullable Long maxBufferSizeInBytes,
+ @Nullable Long maxTimeInBufferMS,
+ @Nullable DataType physicalDataType,
+ PrometheusConfig prometheusConfig) {
+
+ super(
+ maxBatchSize, // maxBatchSizeInSamples
+ maxInFlightRequests,
+ maxBufferedRequests,
+ maxBufferSizeInBytes, // maxRecordSizeInSamples
+ maxTimeInBufferMS);
+
+ this.physicalDataType =
+ Preconditions.checkNotNull(physicalDataType, "Consumed data type must not be null");
+ this.prometheusConfig = prometheusConfig;
+ }
+
+ @Override
+ public ChangelogMode getChangelogMode(ChangelogMode changelogMode) {
+ return ChangelogMode.upsert();
+ }
+
+ @Override
+ public SinkRuntimeProvider getSinkRuntimeProvider(Context context) {
+
+ PrometheusSinkBuilder<RowData> builder =
+ new PrometheusSinkBuilder<RowData>()
+ .setRetryConfiguration(
+ PrometheusSinkConfiguration.RetryConfiguration
+ .DEFAULT_RETRY_CONFIGURATION)
+ .setElementConverter(
+ new RowDataElementConverter(physicalDataType, prometheusConfig))
+ .setPrometheusRemoteWriteUrl(prometheusConfig.getRemoteWriteEndpointUrl());
+
+ Optional.ofNullable(prometheusConfig.getRequestSignerIdentifier())
+ .ifPresent(region -> builder.setRequestSigner(getRequestSigner(prometheusConfig)));
Review Comment:
I'm uncertain why `region` is being created within the context of a request signer identifier only not to be used in the subsequent call. Would this be simpler to leave what's required from the `prometheusConfig` to the `getRequestSigner` method which will return null if there isn't one and only call `builder.setRequestSigner` if it isn't?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-33139] Add Prometheus Sink Table API [flink-connector-prometheus]
Posted by "darenwkt (via GitHub)" <gi...@apache.org>.
darenwkt commented on code in PR #22:
URL: https://github.com/apache/flink-connector-prometheus/pull/22#discussion_r2055387254
##########
flink-connector-prometheus/src/main/java/org/apache/flink/connector/prometheus/table/PrometheusDynamicSinkFactory.java:
##########
@@ -0,0 +1,165 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.connector.prometheus.table;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.configuration.ConfigOption;
+import org.apache.flink.configuration.ReadableConfig;
+import org.apache.flink.connector.base.table.AsyncDynamicTableSinkFactory;
+import org.apache.flink.connector.base.table.AsyncSinkConnectorOptions;
+import org.apache.flink.table.api.DataTypes;
+import org.apache.flink.table.api.ValidationException;
+import org.apache.flink.table.catalog.ResolvedCatalogTable;
+import org.apache.flink.table.connector.sink.DynamicTableSink;
+import org.apache.flink.table.factories.FactoryUtil;
+import org.apache.flink.table.types.DataType;
+
+import java.util.HashSet;
+import java.util.List;
+import java.util.Optional;
+import java.util.Properties;
+import java.util.Set;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+
+import static org.apache.flink.connector.prometheus.table.PrometheusConnectorOption.AWS_REGION;
+import static org.apache.flink.connector.prometheus.table.PrometheusConnectorOption.METRIC_LABEL_KEYS;
+import static org.apache.flink.connector.prometheus.table.PrometheusConnectorOption.METRIC_NAME;
+import static org.apache.flink.connector.prometheus.table.PrometheusConnectorOption.METRIC_REMOTE_WRITE_URL;
+import static org.apache.flink.connector.prometheus.table.PrometheusConnectorOption.METRIC_REQUEST_SIGNER;
+import static org.apache.flink.connector.prometheus.table.PrometheusConnectorOption.METRIC_SAMPLE_KEY;
+import static org.apache.flink.connector.prometheus.table.PrometheusConnectorOption.METRIC_SAMPLE_TIMESTAMP;
+
+/** Factory for creating {@link PrometheusDynamicSink}. */
+@Internal
+public class PrometheusDynamicSinkFactory extends AsyncDynamicTableSinkFactory {
+ public static final String FACTORY_IDENTIFIER = "prometheus";
+
+ @Override
+ public DynamicTableSink createDynamicTableSink(Context context) {
+ FactoryUtil.TableFactoryHelper factoryHelper =
+ FactoryUtil.createTableFactoryHelper(this, context);
+ ResolvedCatalogTable catalogTable = context.getCatalogTable();
+
+ FactoryUtil.validateFactoryOptions(this, factoryHelper.getOptions());
+
+ DataType physicalDataType = catalogTable.getResolvedSchema().toPhysicalRowDataType();
+ PrometheusConfig prometheusConfig = new PrometheusConfig(factoryHelper.getOptions());
+ validateMetricConfigKeys(prometheusConfig, physicalDataType);
+
+ PrometheusDynamicSink.PrometheusDynamicSinkBuilder builder =
+ PrometheusDynamicSink.builder()
+ .setPhysicalDataType(physicalDataType)
+ .setPrometheusConfiguration(prometheusConfig);
+
+ addAsyncOptionsToBuilder(getAsyncSinkOptions(factoryHelper.getOptions()), builder);
+
+ return builder.build();
+ }
+
+ @Override
+ public String factoryIdentifier() {
+ return FACTORY_IDENTIFIER;
+ }
+
+ @Override
+ public Set<ConfigOption<?>> requiredOptions() {
+ final Set<ConfigOption<?>> requiredOptions = new HashSet<>();
+
+ requiredOptions.add(METRIC_NAME);
+ requiredOptions.add(METRIC_LABEL_KEYS);
+ requiredOptions.add(METRIC_SAMPLE_KEY);
+ requiredOptions.add(METRIC_SAMPLE_TIMESTAMP);
+ requiredOptions.add(METRIC_REMOTE_WRITE_URL);
+
+ return requiredOptions;
+ }
+
+ @Override
+ public Set<ConfigOption<?>> optionalOptions() {
+ final Set<ConfigOption<?>> optionalOptions = new HashSet<>();
+
+ optionalOptions.add(AWS_REGION);
+ optionalOptions.add(METRIC_REQUEST_SIGNER);
+
+ return optionalOptions;
+ }
+
+ private Properties getAsyncSinkOptions(ReadableConfig config) {
+ Properties properties = new Properties();
+ Optional.ofNullable(config.get(AsyncSinkConnectorOptions.FLUSH_BUFFER_SIZE))
+ .ifPresent(
+ flushBufferSize ->
+ properties.put(
+ AsyncSinkConnectorOptions.FLUSH_BUFFER_SIZE.key(),
+ flushBufferSize));
+ Optional.ofNullable(config.get(AsyncSinkConnectorOptions.MAX_BATCH_SIZE))
+ .ifPresent(
+ maxBatchSize ->
+ properties.put(
+ AsyncSinkConnectorOptions.MAX_BATCH_SIZE.key(),
+ maxBatchSize));
+ Optional.ofNullable(config.get(AsyncSinkConnectorOptions.MAX_IN_FLIGHT_REQUESTS))
+ .ifPresent(
+ maxInflightRequests ->
+ properties.put(
+ AsyncSinkConnectorOptions.MAX_IN_FLIGHT_REQUESTS.key(),
+ maxInflightRequests));
+ Optional.ofNullable(config.get(AsyncSinkConnectorOptions.MAX_BUFFERED_REQUESTS))
+ .ifPresent(
+ maxBufferedRequests ->
+ properties.put(
+ AsyncSinkConnectorOptions.MAX_BUFFERED_REQUESTS.key(),
+ maxBufferedRequests));
+ Optional.ofNullable(config.get(AsyncSinkConnectorOptions.FLUSH_BUFFER_TIMEOUT))
+ .ifPresent(
+ timeout ->
+ properties.put(
+ AsyncSinkConnectorOptions.FLUSH_BUFFER_TIMEOUT.key(),
+ timeout));
+ return properties;
+ }
+
+ private void validateMetricConfigKeys(
+ PrometheusConfig prometheusConfig, DataType physicalDataType) {
+ Set<String> fieldNames =
+ DataType.getFields(physicalDataType).stream()
+ .map(DataTypes.AbstractField::getName)
+ .collect(Collectors.toSet());
+
+ List<String> metricConfigKeys =
+ Stream.concat(
+ Stream.of(
+ prometheusConfig.getMetricName(),
+ prometheusConfig.getMetricSampleKey(),
+ prometheusConfig.getMetricSampleTimestamp()),
+ prometheusConfig.getLabelKeys().stream())
+ .collect(Collectors.toList());
+
+ metricConfigKeys.forEach(
+ metricConfigKey ->
+ Optional.ofNullable(metricConfigKey)
+ .filter(fieldNames::contains)
+ .orElseThrow(
+ () ->
+ new ValidationException(
+ String.format(
+ "%s not found in fields: %s",
Review Comment:
Adopted suggestion
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-33139] Add Prometheus Sink Table API [flink-connector-prometheus]
Posted by "georgelza (via GitHub)" <gi...@apache.org>.
georgelza commented on PR #22:
URL: https://github.com/apache/flink-connector-prometheus/pull/22#issuecomment-2940352567
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
> <https://github.com/apache/flink-connector-prometheus/pull/22#discussion_r2125943268>
> :
>
> > @@ -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
> <https://github.com/apache/flink-connector-prometheus/pull/22#discussion_r2125943268>,
> 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!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-33139] Add Prometheus Sink Table API [flink-connector-prometheus]
Posted by "darenwkt (via GitHub)" <gi...@apache.org>.
darenwkt commented on code in PR #22:
URL: https://github.com/apache/flink-connector-prometheus/pull/22#discussion_r2125943268
##########
flink-connector-prometheus-request-signer-amp/src/main/java/org/apache/flink/connector/prometheus/sink/aws/AmazonManagedPrometheusWriteRequestSignerFactory.java:
##########
@@ -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
Review Comment:
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
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-33139] Add Prometheus Sink Table API [flink-connector-prometheus]
Posted by "georgelza (via GitHub)" <gi...@apache.org>.
georgelza commented on code in PR #22:
URL: https://github.com/apache/flink-connector-prometheus/pull/22#discussion_r2121669497
##########
flink-connector-prometheus-request-signer-amp/src/main/java/org/apache/flink/connector/prometheus/sink/aws/AmazonManagedPrometheusWriteRequestSignerFactory.java:
##########
@@ -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
Review Comment:
please explain "
'sink.batch.max-size' = '2',
'sink.flush-buffer.size' = '1'
have idea what the intend is, just want to confirm...
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-33139] Add Prometheus Sink Table API [flink-connector-prometheus]
Posted by "georgelza (via GitHub)" <gi...@apache.org>.
georgelza commented on PR #22:
URL: https://github.com/apache/flink-connector-prometheus/pull/22#issuecomment-2930733548
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
> <https://github.com/apache/flink-connector-prometheus/pull/22#discussion_r2121153890>
> :
>
> > + 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
> <https://github.com/apache/flink-connector-prometheus/pull/22#discussion_r2121153890>,
> 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!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-33139] Add Prometheus Sink Table API [flink-connector-prometheus]
Posted by "nicusX (via GitHub)" <gi...@apache.org>.
nicusX commented on code in PR #22:
URL: https://github.com/apache/flink-connector-prometheus/pull/22#discussion_r2121683685
##########
flink-connector-prometheus/src/main/java/org/apache/flink/connector/prometheus/table/RowDataToPrometheusTimeSeriesConverter.java:
##########
@@ -0,0 +1,116 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.connector.prometheus.table;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.connector.prometheus.sink.PrometheusTimeSeries;
+import org.apache.flink.table.api.DataTypes;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.data.TimestampData;
+import org.apache.flink.table.types.DataType;
+
+import java.util.Arrays;
+import java.util.List;
+
+import static org.apache.flink.table.data.RowData.createFieldGetter;
+
+/**
+ * Converts from Flink Table API internal type of {@link RowData} to {@link PrometheusTimeSeries}.
+ */
+@Internal
+public class RowDataToPrometheusTimeSeriesConverter {
+
+ private final DataType physicalDataType;
+ private final PrometheusConfig prometheusConfig;
+
+ public RowDataToPrometheusTimeSeriesConverter(
+ DataType physicalDataType, PrometheusConfig prometheusConfig) {
+ this.physicalDataType = physicalDataType;
+ this.prometheusConfig = prometheusConfig;
+ }
+
+ public PrometheusTimeSeries convertRowData(RowData row) {
+ List<DataTypes.Field> fields = DataType.getFields(physicalDataType);
+
+ 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)) {
Review Comment:
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
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-33139] Add Prometheus Sink Table API [flink-connector-prometheus]
Posted by "georgelza (via GitHub)" <gi...@apache.org>.
georgelza commented on PR #22:
URL: https://github.com/apache/flink-connector-prometheus/pull/22#issuecomment-2888283804
@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`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-33139] Add Prometheus Sink Table API [flink-connector-prometheus]
Posted by "darenwkt (via GitHub)" <gi...@apache.org>.
darenwkt commented on code in PR #22:
URL: https://github.com/apache/flink-connector-prometheus/pull/22#discussion_r2125950271
##########
flink-connector-prometheus/src/main/java/org/apache/flink/connector/prometheus/table/RowDataToPrometheusTimeSeriesConverter.java:
##########
@@ -0,0 +1,116 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.connector.prometheus.table;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.connector.prometheus.sink.PrometheusTimeSeries;
+import org.apache.flink.table.api.DataTypes;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.data.TimestampData;
+import org.apache.flink.table.types.DataType;
+
+import java.util.Arrays;
+import java.util.List;
+
+import static org.apache.flink.table.data.RowData.createFieldGetter;
+
+/**
+ * Converts from Flink Table API internal type of {@link RowData} to {@link PrometheusTimeSeries}.
+ */
+@Internal
+public class RowDataToPrometheusTimeSeriesConverter {
+
+ private final DataType physicalDataType;
+ private final PrometheusConfig prometheusConfig;
+
+ public RowDataToPrometheusTimeSeriesConverter(
+ DataType physicalDataType, PrometheusConfig prometheusConfig) {
+ this.physicalDataType = physicalDataType;
+ this.prometheusConfig = prometheusConfig;
+ }
+
+ public PrometheusTimeSeries convertRowData(RowData row) {
+ List<DataTypes.Field> fields = DataType.getFields(physicalDataType);
+
+ 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)) {
Review Comment:
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
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-33139] Add Prometheus Sink Table API [flink-connector-prometheus]
Posted by "georgelza (via GitHub)" <gi...@apache.org>.
georgelza commented on PR #22:
URL: https://github.com/apache/flink-connector-prometheus/pull/22#issuecomment-2871833036
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
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-33139] Add Prometheus Sink Table API [flink-connector-prometheus]
Posted by "darenwkt (via GitHub)" <gi...@apache.org>.
darenwkt commented on code in PR #22:
URL: https://github.com/apache/flink-connector-prometheus/pull/22#discussion_r2055383768
##########
flink-connector-prometheus/src/main/java/org/apache/flink/connector/prometheus/table/PrometheusConfig.java:
##########
@@ -0,0 +1,75 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.connector.prometheus.table;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.configuration.ReadableConfig;
+
+import java.io.Serializable;
+import java.util.Arrays;
+import java.util.List;
+import java.util.stream.Collectors;
+
+import static org.apache.flink.connector.prometheus.table.PrometheusConnectorOption.AWS_REGION;
+import static org.apache.flink.connector.prometheus.table.PrometheusConnectorOption.METRIC_LABEL_KEYS;
+import static org.apache.flink.connector.prometheus.table.PrometheusConnectorOption.METRIC_NAME;
+import static org.apache.flink.connector.prometheus.table.PrometheusConnectorOption.METRIC_REMOTE_WRITE_URL;
+import static org.apache.flink.connector.prometheus.table.PrometheusConnectorOption.METRIC_REQUEST_SIGNER;
+import static org.apache.flink.connector.prometheus.table.PrometheusConnectorOption.METRIC_SAMPLE_KEY;
+import static org.apache.flink.connector.prometheus.table.PrometheusConnectorOption.METRIC_SAMPLE_TIMESTAMP;
+
+/** Prometheus specific configuration. */
+@Internal
+public class PrometheusConfig implements Serializable {
+ private final ReadableConfig options;
+
+ private static final String DELIMITER = ",";
+
+ public PrometheusConfig(ReadableConfig options) {
+ this.options = options;
+ }
+
+ public String getMetricName() {
+ return options.get(METRIC_NAME);
+ }
+
+ public List<String> getLabelKeys() {
Review Comment:
Thanks, this is a good shout, I have updated it to a set
##########
flink-connector-prometheus/src/main/java/org/apache/flink/connector/prometheus/table/PrometheusConnectorOption.java:
##########
@@ -0,0 +1,75 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.connector.prometheus.table;
+
+import org.apache.flink.configuration.ConfigOption;
+import org.apache.flink.configuration.ConfigOptions;
+
+public class PrometheusConnectorOption {
+
+ // -----------------------------------------------------------------------------------------
+ // Prometheus connector specific options
+ // -----------------------------------------------------------------------------------------
+
+ public static final ConfigOption<String> METRIC_NAME =
+ ConfigOptions.key("metric.name")
+ .stringType()
+ .noDefaultValue()
+ .withDescription("Prometheus metric name.");
+
+ public static final ConfigOption<String> METRIC_LABEL_KEYS =
Review Comment:
> Related to previous comment, ConfigOption has native support for CSVs
Thanks, this is a great suggestion, have updated PR with it
> I would also consider rewording the description to make it clear there can be more than one key name provided.
Updated description
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-33139] Add Prometheus Sink Table API [flink-connector-prometheus]
Posted by "nicusX (via GitHub)" <gi...@apache.org>.
nicusX commented on code in PR #22:
URL: https://github.com/apache/flink-connector-prometheus/pull/22#discussion_r2071357779
##########
flink-connector-prometheus/src/main/java/org/apache/flink/connector/prometheus/sink/PrometheusTimeSeriesConverter.java:
##########
@@ -22,42 +22,22 @@
import org.apache.flink.connector.base.sink.writer.ElementConverter;
import org.apache.flink.connector.prometheus.sink.prometheus.Types;
+import java.util.UnknownFormatConversionException;
+
/**
* Converts the sink input {@link PrometheusTimeSeries} into the Protobuf {@link Types.TimeSeries}
* that are sent to Prometheus.
*/
@PublicEvolving
-public class PrometheusTimeSeriesConverter
- implements ElementConverter<PrometheusTimeSeries, Types.TimeSeries> {
-
- private static final String METRIC_NAME_LABEL_NAME = "__name__";
+public class PrometheusTimeSeriesConverter<IN> implements ElementConverter<IN, Types.TimeSeries> {
@Override
- public Types.TimeSeries apply(PrometheusTimeSeries element, SinkWriter.Context context) {
- Types.TimeSeries.Builder builder =
- Types.TimeSeries.newBuilder()
- .addLabels(
- Types.Label.newBuilder()
- .setName(METRIC_NAME_LABEL_NAME)
- .setValue(element.getMetricName())
- .build());
-
- for (PrometheusTimeSeries.Label label : element.getLabels()) {
- builder.addLabels(
- Types.Label.newBuilder()
- .setName(label.getName())
- .setValue(label.getValue())
- .build());
- }
-
- for (PrometheusTimeSeries.Sample sample : element.getSamples()) {
- builder.addSamples(
- Types.Sample.newBuilder()
- .setValue(sample.getValue())
- .setTimestamp(sample.getTimestamp())
- .build());
+ public Types.TimeSeries apply(IN element, SinkWriter.Context context) {
Review Comment:
As my previous point: if we still support PrometheusTimeSeries only, is it worth breaking the public API?
##########
flink-connector-prometheus/src/main/java/org/apache/flink/connector/prometheus/sink/PrometheusSink.java:
##########
@@ -49,11 +49,11 @@ public class PrometheusSink extends AsyncSinkBase<PrometheusTimeSeries, Types.Ti
@SuppressWarnings("checkstyle:RegexpSingleline")
protected PrometheusSink(
- ElementConverter<PrometheusTimeSeries, Types.TimeSeries> elementConverter,
+ ElementConverter<IN, Types.TimeSeries> elementConverter,
int maxInFlightRequests,
int maxBufferedRequests,
int maxBatchSizeInSamples,
- int maxRecordSizeInSamples,
+ long maxRecordSizeInSamples,
Review Comment:
Why changing this? The batch size will never be > Integer.MAX_VALUE samples
##########
flink-connector-prometheus/src/main/java/org/apache/flink/connector/prometheus/table/PrometheusConnectorOption.java:
##########
@@ -0,0 +1,67 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.connector.prometheus.table;
+
+import org.apache.flink.configuration.ConfigOption;
+import org.apache.flink.configuration.ConfigOptions;
+
+import java.util.List;
+
+public class PrometheusConnectorOption {
+
+ // -----------------------------------------------------------------------------------------
+ // Prometheus connector specific options
+ // -----------------------------------------------------------------------------------------
+
+ public static final ConfigOption<String> METRIC_NAME =
+ ConfigOptions.key("metric.name")
Review Comment:
I am not sure I understand the interface.
Is this configuration specifying the metric name as a constant string, or the field of the table that contains the metric name?
In the example, it seems to be the latter.
It should be, because in a realistic use each record can be from a different metric.
However, looking at the example
```
CREATE TABLE PromTable (
`my_metric_name` BIGINT,
`my_label` BIGINT,
...
)
WITH (
'connector' = 'prometheus',
'metric.name' = 'my_metric_name',
'metric.label.keys' = 'my_label'
....
);
```
but the field is a BIGINT, while it should be a VARCHAR.
Also, labels are key/value (String/String) pairs.
Each record (each measurement) can have a different set of Labels (Set of K/V) where both key and value may change
##########
flink-connector-prometheus/src/main/java/org/apache/flink/connector/prometheus/sink/PrometheusSink.java:
##########
@@ -49,11 +49,11 @@ public class PrometheusSink extends AsyncSinkBase<PrometheusTimeSeries, Types.Ti
@SuppressWarnings("checkstyle:RegexpSingleline")
protected PrometheusSink(
- ElementConverter<PrometheusTimeSeries, Types.TimeSeries> elementConverter,
+ ElementConverter<IN, Types.TimeSeries> elementConverter,
Review Comment:
Is this change required for supporting Table API or just a matter of generalizing the API for something we are not really generalizing? The DataStream sink only supports PrometheusTimeSeries as input
Breaking the public interface not only forces a major version, but because it forces existing users to change their code
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-33139] Add Prometheus Sink Table API [flink-connector-prometheus]
Posted by "darenwkt (via GitHub)" <gi...@apache.org>.
darenwkt commented on code in PR #22:
URL: https://github.com/apache/flink-connector-prometheus/pull/22#discussion_r2072389078
##########
flink-connector-prometheus/src/main/java/org/apache/flink/connector/prometheus/table/PrometheusConnectorOption.java:
##########
@@ -0,0 +1,67 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.connector.prometheus.table;
+
+import org.apache.flink.configuration.ConfigOption;
+import org.apache.flink.configuration.ConfigOptions;
+
+import java.util.List;
+
+public class PrometheusConnectorOption {
+
+ // -----------------------------------------------------------------------------------------
+ // Prometheus connector specific options
+ // -----------------------------------------------------------------------------------------
+
+ public static final ConfigOption<String> METRIC_NAME =
+ ConfigOptions.key("metric.name")
Review Comment:
Thanks Lorenzo for the review and apology for the misleading example as there were typo in there, I have updated description and below with an accurate example:
In the example below, a table is created where the metricName, labels are STRING and there are 2 labels (String/String) pairs used as you have correctly pointed out.
```
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',
> '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
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-33139] Add Prometheus Sink Table API [flink-connector-prometheus]
Posted by "darenwkt (via GitHub)" <gi...@apache.org>.
darenwkt commented on PR #22:
URL: https://github.com/apache/flink-connector-prometheus/pull/22#issuecomment-2938983186
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
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-33139] Add Prometheus Sink Table API [flink-connector-prometheus]
Posted by "darenwkt (via GitHub)" <gi...@apache.org>.
darenwkt commented on code in PR #22:
URL: https://github.com/apache/flink-connector-prometheus/pull/22#discussion_r2121781198
##########
flink-connector-prometheus/src/main/java/org/apache/flink/connector/prometheus/table/RowDataToPrometheusTimeSeriesConverter.java:
##########
@@ -0,0 +1,116 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.connector.prometheus.table;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.connector.prometheus.sink.PrometheusTimeSeries;
+import org.apache.flink.table.api.DataTypes;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.data.TimestampData;
+import org.apache.flink.table.types.DataType;
+
+import java.util.Arrays;
+import java.util.List;
+
+import static org.apache.flink.table.data.RowData.createFieldGetter;
+
+/**
+ * Converts from Flink Table API internal type of {@link RowData} to {@link PrometheusTimeSeries}.
+ */
+@Internal
+public class RowDataToPrometheusTimeSeriesConverter {
+
+ private final DataType physicalDataType;
+ private final PrometheusConfig prometheusConfig;
+
+ public RowDataToPrometheusTimeSeriesConverter(
+ DataType physicalDataType, PrometheusConfig prometheusConfig) {
+ this.physicalDataType = physicalDataType;
+ this.prometheusConfig = prometheusConfig;
+ }
+
+ public PrometheusTimeSeries convertRowData(RowData row) {
+ List<DataTypes.Field> fields = DataType.getFields(physicalDataType);
+
+ 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)) {
Review Comment:
Thanks for the clarification, yes I agree, the behaviour does differ from datastream API in that the label key/name is fixed per table.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-33139] Add Prometheus Sink Table API [flink-connector-prometheus]
Posted by "darenwkt (via GitHub)" <gi...@apache.org>.
darenwkt commented on code in PR #22:
URL: https://github.com/apache/flink-connector-prometheus/pull/22#discussion_r2121085993
##########
flink-connector-prometheus-request-signer-amp/src/main/java/org/apache/flink/connector/prometheus/sink/aws/AmazonManagedPrometheusWriteRequestSignerFactory.java:
##########
@@ -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
Review Comment:
Request signer can be specified by user in the table config (`metric.request-signer`), for example:
```
CREATE TABLE PromTable (
`my_metric_name` BIGINT,
`my_label` 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'
);
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-33139] Add Prometheus Sink Table API [flink-connector-prometheus]
Posted by "darenwkt (via GitHub)" <gi...@apache.org>.
darenwkt commented on PR #22:
URL: https://github.com/apache/flink-connector-prometheus/pull/22#issuecomment-2953691241
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
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org