[FLINK-33139] Add Prometheus Sink Table API

Author: darenwkt

.github/PULL_REQUEST_TEMPLATE.md

@@ -32,7 +32,7 @@ This change added tests and can be verified as follows:
 *(example:)*
 - *Added integration tests for end-to-end deployment*
 - *Added unit tests*
-- *Manually verified by running the Kinesis connector on a local Flink cluster.*
+- *Manually verified by running the Prometheus connector on a local Flink cluster.*
 
 ## Significant changes
 *(Please check any boxes [x] if the answer is "yes". You can first publish the PR and check them afterwards, for convenience.)*

docs/data/prometheus.yml

@@ -0,0 +1,22 @@
+################################################################################
+#  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.
+################################################################################
+
+version: 1.1-SNAPSHOT
+variants:
+  - maven: flink-connector-prometheus
+  - maven: flink-connector-prometheus-request-signer-amp

flink-connector-prometheus-request-signer-amp/src/main/java/org/apache/flink/connector/prometheus/sink/aws/AmazonManagedPrometheusWriteRequestSignerFactory.java

@@ -0,0 +1,19 @@
+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;
+
+public class AmazonManagedPrometheusWriteRequestSignerFactory
+        implements PrometheusDynamicRequestSignerFactory {
+    @Override
+    public String requestSignerIdentifer() {
+        return "amazon-managed-prometheus";
+    }
+
+    @Override
+    public PrometheusRequestSigner getRequestSigner(PrometheusConfig config) {
+        return new AmazonManagedPrometheusWriteRequestSigner(
+                config.getRemoteWriteEndpointUrl(), config.getRemoteWriteEndpointRegion());
+    }
+}

flink-connector-prometheus-request-signer-amp/src/main/resources/META-INF/services/org.apache.flink.connector.prometheus.table.PrometheusDynamicRequestSignerFactory

@@ -0,0 +1,16 @@
+# 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.
+
+org.apache.flink.connector.prometheus.sink.aws.AmazonManagedPrometheusWriteRequestSignerFactory

flink-connector-prometheus/pom.xml

@@ -77,6 +77,31 @@ under the License.
             <artifactId>httpcore5</artifactId>
         </dependency>
 
+        <!--Table API dependencies-->
+        <dependency>
+            <groupId>org.apache.flink</groupId>
+            <artifactId>flink-table-common</artifactId>
+            <version>${flink.version}</version>
+            <type>test-jar</type>
+            <scope>test</scope>
+        </dependency>
+
+        <dependency>
+            <groupId>org.apache.flink</groupId>
+            <artifactId>flink-table-api-java-bridge</artifactId>
+            <version>${flink.version}</version>
+            <scope>provided</scope>
+            <optional>true</optional>
+        </dependency>
+
+        <dependency>
+            <groupId>org.apache.flink</groupId>
+            <artifactId>flink-table-runtime</artifactId>
+            <version>${flink.version}</version>
+            <scope>provided</scope>
+            <optional>true</optional>
+        </dependency>
+
         <!-- test -->
         <dependency>
             <groupId>org.apache.flink</groupId>

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> {
     private final String prometheusRemoteWriteUrl;
     private final PrometheusAsyncHttpClientBuilder clientBuilder;
     private final PrometheusRequestSigner requestSigner;
@@ -49,11 +49,11 @@ public class PrometheusSink extends AsyncSinkBase<PrometheusTimeSeries, Types.Ti
 
     @SuppressWarnings("checkstyle:RegexpSingleline")
     protected PrometheusSink(
-            ElementConverter<PrometheusTimeSeries, Types.TimeSeries> elementConverter,
+            ElementConverter<InputT, Types.TimeSeries> elementConverter,
             int maxInFlightRequests,
             int maxBufferedRequests,
             int maxBatchSizeInSamples,
-            int maxRecordSizeInSamples,
+            long maxRecordSizeInSamples,
             long maxTimeInBufferMS,
             String prometheusRemoteWriteUrl,
             PrometheusAsyncHttpClientBuilder clientBuilder,
@@ -102,16 +102,16 @@ protected PrometheusSink(
     }
 
     @Override
-    public StatefulSinkWriter<PrometheusTimeSeries, BufferedRequestState<Types.TimeSeries>>
-            createWriter(InitContext initContext) {
+    public StatefulSinkWriter<InputT, BufferedRequestState<Types.TimeSeries>> createWriter(
+            InitContext initContext) {
         SinkMetricsCallback metricsCallback =
                 new SinkMetricsCallback(
                         SinkMetrics.registerSinkMetrics(
                                 initContext.metricGroup().addGroup(metricGroupName)));
         CloseableHttpAsyncClient asyncHttpClient =
                 clientBuilder.buildAndStartClient(metricsCallback);
 
-        return new PrometheusSinkWriter(
+        return new PrometheusSinkWriter<>(
                 getElementConverter(),
                 initContext,
                 getMaxInFlightRequests(),
@@ -128,17 +128,16 @@ protected PrometheusSink(
     }
 
     @Override
-    public StatefulSinkWriter<PrometheusTimeSeries, BufferedRequestState<Types.TimeSeries>>
-            restoreWriter(
-                    InitContext initContext,
-                    Collection<BufferedRequestState<Types.TimeSeries>> recoveredState) {
+    public StatefulSinkWriter<InputT, 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(
+        return new PrometheusSinkWriter<>(
                 getElementConverter(),
                 initContext,
                 getMaxInFlightRequests(),
@@ -155,8 +154,8 @@ protected PrometheusSink(
                 recoveredState);
     }
 
-    public static PrometheusSinkBuilder builder() {
-        return new PrometheusSinkBuilder();
+    public static <InputT> PrometheusSinkBuilder<InputT> builder() {
+        return new PrometheusSinkBuilder<>();
     }
 
     @Override

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

@@ -18,8 +18,8 @@
 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.AsyncSinkBaseBuilder;
+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.prometheus.Types;
 
@@ -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>> {
     private static final Logger LOG = LoggerFactory.getLogger(PrometheusSinkBuilder.class);
 
     // Max batch size, in number of samples
@@ -52,14 +51,21 @@ public class PrometheusSinkBuilder
     private Integer socketTimeoutMs;
     private PrometheusRequestSigner requestSigner = null;
     private Integer maxBatchSizeInSamples;
-    private Integer maxRecordSizeInSamples;
+    private Long maxRecordSizeInSamples;
     private String httpUserAgent = null;
     private PrometheusSinkConfiguration.SinkWriterErrorHandlingBehaviorConfiguration
             errorHandlingBehaviorConfig = null;
+    private ElementConverter<InputT, Types.TimeSeries> elementConverter;
     private String metricGroupName = null;
 
+    public PrometheusSinkBuilder<InputT> setElementConverter(
+            ElementConverter<InputT, Types.TimeSeries> elementConverter) {
+        this.elementConverter = elementConverter;
+        return this;
+    }
+
     @Override
-    public AsyncSinkBase<PrometheusTimeSeries, Types.TimeSeries> build() {
+    public PrometheusSink<InputT> build() {
 
         int actualMaxBatchSizeInSamples =
                 Optional.ofNullable(maxBatchSizeInSamples)
@@ -69,8 +75,9 @@ public AsyncSinkBase<PrometheusTimeSeries, Types.TimeSeries> build() {
         long actualMaxTimeInBufferMS =
                 Optional.ofNullable(getMaxTimeInBufferMS()).orElse(DEFAULT_MAX_TIME_IN_BUFFER_MS);
 
-        int actualMaxRecordSizeInSamples =
-                Optional.ofNullable(maxRecordSizeInSamples).orElse(actualMaxBatchSizeInSamples);
+        long actualMaxRecordSizeInSamples =
+                Optional.ofNullable(maxRecordSizeInSamples)
+                        .orElse((long) actualMaxBatchSizeInSamples);
 
         int actualSocketTimeoutMs =
                 Optional.ofNullable(socketTimeoutMs)
@@ -116,8 +123,8 @@ public AsyncSinkBase<PrometheusTimeSeries, Types.TimeSeries> build() {
                 actualErrorHandlingBehaviorConfig.getOnMaxRetryExceeded(),
                 actualErrorHandlingBehaviorConfig.getOnPrometheusNonRetryableError());
 
-        return new PrometheusSink(
-                new PrometheusTimeSeriesConverter(),
+        return new PrometheusSink<>(
+                Optional.ofNullable(elementConverter).orElse(new PrometheusTimeSeriesConverter<>()),
                 MAX_IN_FLIGHT_REQUESTS,
                 actualMaxBufferedRequests,
                 actualMaxBatchSizeInSamples,
@@ -132,50 +139,51 @@ public AsyncSinkBase<PrometheusTimeSeries, Types.TimeSeries> build() {
                 actualMetricGroupName);
     }
 
-    public PrometheusSinkBuilder setPrometheusRemoteWriteUrl(String prometheusRemoteWriteUrl) {
+    public PrometheusSinkBuilder<InputT> setPrometheusRemoteWriteUrl(
+            String prometheusRemoteWriteUrl) {
         this.prometheusRemoteWriteUrl = prometheusRemoteWriteUrl;
         return this;
     }
 
-    public PrometheusSinkBuilder setRequestSigner(PrometheusRequestSigner requestSigner) {
+    public PrometheusSinkBuilder<InputT> setRequestSigner(PrometheusRequestSigner requestSigner) {
         this.requestSigner = requestSigner;
         return this;
     }
 
-    public PrometheusSinkBuilder setMaxBatchSizeInSamples(int maxBatchSizeInSamples) {
+    public PrometheusSinkBuilder<InputT> setMaxBatchSizeInSamples(int maxBatchSizeInSamples) {
         this.maxBatchSizeInSamples = maxBatchSizeInSamples;
         return this;
     }
 
-    public PrometheusSinkBuilder setMaxRecordSizeInSamples(int maxRecordSizeInSamples) {
+    public PrometheusSinkBuilder<InputT> setMaxRecordSizeInSamples(long maxRecordSizeInSamples) {
         this.maxRecordSizeInSamples = maxRecordSizeInSamples;
         return this;
     }
 
-    public PrometheusSinkBuilder setRetryConfiguration(
+    public PrometheusSinkBuilder<InputT> setRetryConfiguration(
             PrometheusSinkConfiguration.RetryConfiguration retryConfiguration) {
         this.retryConfiguration = retryConfiguration;
         return this;
     }
 
-    public PrometheusSinkBuilder setSocketTimeoutMs(int socketTimeoutMs) {
+    public PrometheusSinkBuilder<InputT> setSocketTimeoutMs(int socketTimeoutMs) {
         this.socketTimeoutMs = socketTimeoutMs;
         return this;
     }
 
-    public PrometheusSinkBuilder setHttpUserAgent(String httpUserAgent) {
+    public PrometheusSinkBuilder<InputT> setHttpUserAgent(String httpUserAgent) {
         this.httpUserAgent = httpUserAgent;
         return this;
     }
 
-    public PrometheusSinkBuilder setErrorHandlingBehaviorConfiguration(
+    public PrometheusSinkBuilder<InputT> setErrorHandlingBehaviorConfiguration(
             PrometheusSinkConfiguration.SinkWriterErrorHandlingBehaviorConfiguration
                     errorHandlingBehaviorConfig) {
         this.errorHandlingBehaviorConfig = errorHandlingBehaviorConfig;
         return this;
     }
 
-    public PrometheusSinkBuilder setMetricGroupName(String metricGroupName) {
+    public PrometheusSinkBuilder<InputT> setMetricGroupName(String metricGroupName) {
         this.metricGroupName = metricGroupName;
         return this;
     }
@@ -184,20 +192,20 @@ public PrometheusSinkBuilder setMetricGroupName(String metricGroupName) {
 
     /** Not supported. Use setMaxBatchSizeInSamples(int) instead */
     @Override
-    public PrometheusSinkBuilder setMaxBatchSize(int maxBatchSize) {
+    public PrometheusSinkBuilder<InputT> setMaxBatchSize(int maxBatchSize) {
         throw new UnsupportedOperationException("maxBatchSize is not supported by this sink");
     }
 
     /** Not supported. Use setMaxBatchSizeInSamples(int) instead */
     @Override
-    public PrometheusSinkBuilder setMaxBatchSizeInBytes(long maxBatchSizeInBytes) {
+    public PrometheusSinkBuilder<InputT> setMaxBatchSizeInBytes(long maxBatchSizeInBytes) {
         throw new UnsupportedOperationException(
                 "maxBatchSizeInBytes is not supported by this sink");
     }
 
     /** Not supported. Use setMaxRecordSizeInSamples(int) instead */
     @Override
-    public PrometheusSinkBuilder setMaxRecordSizeInBytes(long maxRecordSizeInBytes) {
+    public PrometheusSinkBuilder<InputT> setMaxRecordSizeInBytes(long maxRecordSizeInBytes) {
         throw new UnsupportedOperationException(
                 "maxRecordSizeInBytes is not supported by this sink");
     }

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

@@ -41,7 +41,7 @@
 import java.util.function.Consumer;
 
 /**
- * Writer, taking care of batching the {@link PrometheusTimeSeries} and handling retries.
+ * Writer, taking care of batching the {@link InputT} and handling retries.
  *
  * <p>The batching of this sink is in terms of Samples, not bytes. The goal is adaptively increase
  * the number of Samples in each batch, a WriteRequest sent to Prometheus, to a configurable number.
@@ -59,7 +59,7 @@
  * maxBatchSizeInBytes.
  */
 @Internal
-public class PrometheusSinkWriter extends AsyncSinkWriter<PrometheusTimeSeries, Types.TimeSeries> {
+public class PrometheusSinkWriter<InputT> extends AsyncSinkWriter<InputT, Types.TimeSeries> {
     private static final Logger LOG = LoggerFactory.getLogger(PrometheusSinkWriter.class);
 
     private final SinkMetricsCallback metricsCallback;
@@ -69,7 +69,7 @@ public class PrometheusSinkWriter extends AsyncSinkWriter<PrometheusTimeSeries,
             errorHandlingBehaviorConfig;
 
     public PrometheusSinkWriter(
-            ElementConverter<PrometheusTimeSeries, Types.TimeSeries> elementConverter,
+            ElementConverter<InputT, Types.TimeSeries> elementConverter,
             Sink.InitContext context,
             int maxInFlightRequests,
             int maxBufferedRequests,
@@ -101,7 +101,7 @@ public PrometheusSinkWriter(
     }
 
     public PrometheusSinkWriter(
-            ElementConverter<PrometheusTimeSeries, Types.TimeSeries> elementConverter,
+            ElementConverter<InputT, Types.TimeSeries> elementConverter,
             Sink.InitContext context,
             int maxInFlightRequests,
             int maxBufferedRequests,