Add support for basic-auth and no-auth for prometheus sink

Signed-off-by: ps48 <pshenoy36@gmail.com>

Author: ps48

data-prepper-plugins/prometheus-sink/README.md

@@ -1,170 +1,159 @@
 # Prometheus Sink
 
-This is the Data Prepper Prometheus sink plugin that sends records to http/https endpoints. You can use the sink to send data to arbitrary HTTP Endpoints which can be backed by prometheus.
-
+This is the Data Prepper Prometheus sink plugin that sends metric data to Prometheus Remote Write endpoints. It supports both open source Prometheus and AWS Managed Prometheus (AMP).
 
 ## Usages
 
-The Prometheus sink should be configured as part of Data Prepper pipeline yaml file.
+The Prometheus sink should be configured as part of a Data Prepper pipeline YAML file.
 
-### Response status
+### Open Source Prometheus (No Auth)
 
-* `200`: the request data has been successfully pushed to http endpoint.
-* `500`: internal server error while process the request data.
-* `400`: bad request error
-* `404`: the http endpoint is not reachable
-* `501`: the server does not recognize the request method and is incapable of supporting it for any resource
+To use with a vanilla Prometheus instance, provide an `http://` or `https://` URL. No `aws` block is needed.
 
-### HTTP Basic authentication
-```
+Prometheus must be started with the `--web.enable-remote-write-receiver` flag.
+
+```yaml
 pipeline:
   ...
   sink:
-  - prometheus:
-      authentication:
-        http_basic:
-          username: my-user
-          password: my_s3cr3t
+    - prometheus:
+        url: "http://localhost:9090/api/v1/write"
+        threshold:
+          max_events: 500
+          flush_interval: 5s
 ```
 
-### HTTP Bearer token authentication
-```
+### Open Source Prometheus with Basic Auth
+
+To authenticate with HTTP Basic credentials (e.g., when Prometheus is behind a reverse proxy with basic auth):
+
+```yaml
 pipeline:
   ...
   sink:
-  - prometheus:
-      authentication:
-         bearer_token:
-            client_id: 0oaafr4j79grYGC5d7
-            client_secret: fFel-3FutCXAOndezEsOVlght6D6DR4OIt7G5D1_oJ6YtgU17JdyXmGf0M
-            token_url: https://localhost/oauth2/default/v1/token
-            grant_type: client_credentials
-            scope: prometheusSink
+    - prometheus:
+        url: "http://localhost:9090/api/v1/write"
+        authentication:
+          http_basic:
+            username: "promuser"
+            password: "prompass"
 ```
 
-## Configuration
-
-- `url` The http/https endpoint url which can bee backed by prometheus.
-
-- `encoding`  Default is snappy
+### AWS Managed Prometheus (AMP)
 
-- `content_type` Default is application/x-protobuf
+To use with AMP, provide the `aws` configuration block. An `https://` URL is required when using AWS authentication.
 
-- `remote_write_version` : Prometheus Remote.Writer version Version, Default is 0.1.0
-
-- `proxy`(optional): A String of the address of a forward HTTP proxy. The format is like "<host-name-or-ip>:\<port\>". Examples: "example.com:8100", "http://example.com:8100", "112.112.112.112:8100". Note: port number cannot be omitted.
+```yaml
+pipeline:
+  ...
+  sink:
+    - prometheus:
+        url: "https://aps-workspaces.us-east-2.amazonaws.com/workspaces/ws-xxxxxxxx-xxxx/api/v1/remote_write"
+        aws:
+          region: "us-east-2"
+          sts_role_arn: "arn:aws:iam::123456789012:role/data-prepper-prometheus-role"
+        threshold:
+          max_events: 500
+          flush_interval: 5s
+```
 
-- `http_method` (Optional) : HttpMethod to be used. Default is POST.
+### Response Status
 
-- `auth_type` (Optional): Authentication type configuration. By default, this runs an unauthenticated server.
+* `200`: The request data has been successfully pushed to the endpoint.
+* `400`: Bad request error.
+* `404`: The endpoint is not reachable.
+* `429`: Too many requests (retried automatically).
+* `500`: Internal server error.
+* `502`, `503`, `504`: Server errors (retried automatically).
 
-- `username`(optional): A string of username required for basic authentication
+## Configuration
 
-- `password`(optional): A string of password required for basic authentication
+### Required
 
-- `client_id`: It is the client id is the public identifier of your authorization server.
+| Option | Description |
+|--------|-------------|
+| `url` | The Prometheus Remote Write endpoint URL. Supports `http://` and `https://` schemes. When `aws` is configured, `https://` is required. |
 
-- `client_secret` : It is a secret known only to the application and the authorization server.
+### Optional
 
-- `token_url`: The End point URL of the OAuth server.(Eg: /oauth2/default/v1/token)
+| Option | Default | Description |
+|--------|---------|-------------|
+| `aws` | `null` | AWS configuration for SigV4 signing. When present, requests are signed with AWS credentials. See [AWS Configuration](#aws-configuration). |
+| `authentication` | `null` | HTTP Basic authentication credentials. See [Authentication](#authentication). Cannot be used with `aws`. |
+| `encoding` | `snappy` | Compression encoding. Currently only `snappy` is supported. |
+| `content_type` | `application/x-protobuf` | Content type of the request body. Currently only `application/x-protobuf` is supported. |
+| `remote_write_version` | `0.1.0` | Prometheus Remote Write protocol version. Currently only `0.1.0` is supported. |
+| `max_retries` | `5` | Maximum number of retry attempts for failed requests. Uses exponential backoff with jitter on retryable status codes (429, 502, 503, 504). |
+| `request_timeout` | `60s` | HTTP request timeout. Must be between 1s and 600s. |
+| `connection_timeout` | `60s` | TCP connection timeout. Must be between 1s and 600s. |
+| `idle_timeout` | `60s` | Connection idle timeout. Must be between 1s and 600s. |
+| `out_of_order_time_window` | `10s` | Time window for handling out-of-order metric samples. |
+| `sanitize_names` | `true` | Whether to sanitize metric names to be Prometheus-compliant. |
+| `threshold` | See below | Buffer threshold configuration. See [Threshold Configuration](#threshold-configuration). |
 
-- `grant_type` (Optional) : This grant type refers to the way an application gets an access token. Example: client_credentials/refresh_token
+### <a name="threshold-configuration">Threshold Configuration</a>
 
-- `scope` (Optional) : This scope limit an application's access to a user's account.
+| Option | Default | Description |
+|--------|---------|-------------|
+| `max_events` | `500` | Maximum number of events to buffer before flushing. |
+| `max_request_size` | `1048576` (1 MB) | Maximum request size in bytes before flushing. |
+| `flush_interval` | `10000` (ms) | Maximum time in milliseconds to wait before flushing the buffer. |
 
-- `aws` (Optional) : AWS configurations. See [AWS Configuration](#aws_configuration) for details. SigV4 is enabled by default when this option is used. If this option is present, `aws_` options are not expected to be present. If any of `aws_` options are present along with this, error is thrown.
+### <a name="aws-configuration">AWS Configuration</a>
 
-- `custom_header` (Optional) : A Map<String, List<String> for custom headers such as AWS Sagemaker etc
+When the `aws` block is present, requests are automatically signed with AWS SigV4. An `https://` URL is required.
 
-- `dlq_file`(optional): A String of absolute file path for DLQ failed output records. Defaults to null.
-  If not provided, failed records will be written into the default data-prepper log file (`logs/Data-Prepper.log`). If the `dlq` option is present along with this, an error is thrown.
+| Option | Required | Description |
+|--------|----------|-------------|
+| `region` | No | The AWS region to use for credentials. Defaults to [standard SDK behavior to determine the region](https://docs.aws.amazon.com/sdk-for-java/latest/developer-guide/region-selection.html). |
+| `sts_role_arn` | No | The STS role to assume for requests to AWS. Defaults to null, which uses [standard SDK credential behavior](https://docs.aws.amazon.com/sdk-for-java/latest/developer-guide/credentials.html). |
+| `sts_header_overrides` | No | A map of header overrides to make when assuming the IAM role. |
+| `sts_external_id` | No | An optional external ID to use when assuming the IAM role. |
 
-- `dlq` (optional): DLQ configurations. See [DLQ](https://github.com/opensearch-project/data-prepper/tree/main/data-prepper-plugins/failures-common/src/main/java/org/opensearch/dataprepper/plugins/dlq/README.md) for details. If the `dlq_file` option is present along with this, an error is thrown.
+### <a name="authentication">Authentication</a>
 
-- `max_retries`(optional): A number indicating the maximum number of times Prometheus Sink should try to push the data to the Http arbitrary endpoint before considering it as failure. Defaults to `Integer.MAX_VALUE`.
+The `authentication` block supports HTTP Basic authentication. It cannot be used together with `aws` (SigV4 signing).
 
-- `request_timout`(optional): A duration that represents the request timeout. Example: 1000ms, 5s etc
-### Prometheus Sink full pipeline
-```
-  sink:
-    - prometheus:
-        url: http/https arbitrary endpoint url
-        encoding: snappy
-        content-type: application/x-protobuf
-        remote-write-version: 0.1.0
-        proxy: proxy url
-        http_method: "POST"
-        auth_type: "unauthenticated"
-        authentication:
-          http_basic:
-            username: "username"
-            password: "password"
-          bearer_token:
-            client_id: 0oaafr4j79segd7
-            client_secret: fFel-3FutCXAOndezEsOVlghoJ6w0wNoaYtgU17JdyXmGf0M
-            token_url: token url
-            grant_type: client_credentials
-            scope:
-        insecure: false
-        insecure_skip_verify: false
-        ssl_certificate_file: "/full/path/to/certfile.crt"
-        buffer_type: "in_memory"
-        use_acm_cert_for_ssl: false
-        acm_certificate_arn:
-        custom_header:
-          header: ["value"]
-        dlq_file : <dlq file with full path>
-        dlq:
-          s3:
-            bucket: 
-            key_path_prefix:
-        aws:
-          region: "us-east-2"
-          sts_role_arn: "arn:aws:iam::1234567890:role/data-prepper-s3source-execution-role"
-          sigv4: false
-        max_retries: 5
-        request_timout: 20s
-```
+| Option | Description |
+|--------|-------------|
+| `http_basic.username` | The username for HTTP Basic authentication. |
+| `http_basic.password` | The password for HTTP Basic authentication. |
 
-### SSL
+### End-to-End Acknowledgements
 
-* insecure_skip_verify(Optional) => A `boolean` that enables mTLS/SSL. Default is ```false```.
-  * If set to false then the user has two options:
-    * Use default trust. This can allow for reaching many endpoints. The user does not need to provide any .crt/.key files.
-    * Allow the user to specify a .crt file for a certificate (no .key is required because this is the client). By the user providing the .crt file, the user is stating he trusts that certificate. We will still verify the signature match.
-  * If set to true, then skip any verification of the certificate. The user does not need to provide a .crt or .key file.
-* insecure (Optional) => A `boolean` that allows http/https endpoints. Default is ```false```.
-  * If set to false, then only https:// URLs are permitted. Throw an InvalidPluginConfigurationException if the URL is configured with an http:// scheme in the URL.
-  * If set to true, then the user can provide both http:// https:// as the scheme.
-* ssl_certificate_file(Optional) => A `String` that represents the SSL certificate chain file path or AWS S3 path. S3 path example `s3://<bucketName>/<path>`. Required if `ssl` is set to `true` and `use_acm_certificate_for_ssl` is set to `false`.
-* ssl_key_file(Optional) => A `String` that represents the SSL key file path or AWS S3 path. S3 path example `s3://<bucketName>/<path>`. Only decrypted key file is supported. Required if `ssl` is set to `true` and `use_acm_certificate_for_ssl` is set to `false`.
-* use_acm_certificate_for_ssl(Optional) : A `boolean` that enables mTLS/SSL using certificate and private key from AWS Certificate Manager (ACM). Default is `false`.
-* acm_certificate_arn(Optional) : A `String` that represents the ACM certificate ARN. ACM certificate take preference over S3 or local file system certificate. Required if `use_acm_certificate_for_ssl` is set to `true`.
-* acm_private_key_password(Optional): A `String` that represents the ACM private key password which that will be used to decrypt the private key. If it's not provided, a random password will be generated.
-* acm_certificate_timeout_millis(Optional) : An `int` that represents the timeout in milliseconds for ACM to get certificates. Default value is `120000`.
+If the events received by the Prometheus sink have end-to-end acknowledgements enabled (tracked via EventHandle), then upon successful posting a positive acknowledgement is sent, otherwise a negative acknowledgement is sent.
 
-### <a name="aws_configuration">AWS Configuration</a>
+## Not Yet Implemented
 
-* `region` (Optional) : The AWS region to use for credentials. Defaults to [standard SDK behavior to determine the region](https://docs.aws.amazon.com/sdk-for-java/latest/developer-guide/region-selection.html).
-* `sts_role_arn` (Optional) : The STS role to assume for requests to AWS. Defaults to null, which will use the [standard SDK behavior for credentials](https://docs.aws.amazon.com/sdk-for-java/latest/developer-guide/credentials.html).
-* `sts_header_overrides` (Optional): A map of header overrides to make when assuming the IAM role for the sink plugin.
-* `sts_external_id` (Optional): An optional external ID to use when assuming an IAM role.
-* `sigv4`: A boolean flag to sign the HTTP request with AWS credentials. Default to `false`. For aws_sigv4, we don't need any auth_type or ssl
+The following features have configuration classes defined but are **not currently wired up**:
 
-### End-to-End acknowledgements
+- Bearer token / OAuth authentication
+- Proxy support
+- SSL/TLS certificate configuration
+- Custom headers
+- DLQ (Dead Letter Queue) file/plugin support
 
-If the events received by the Prometheus Sink have end-to-end acknowledgements enabled (which is tracked using the presence of EventHandle in the event received for processing), then upon successful posting to OpenSearch or upon successful write to DLQ, a positive acknowledgement is sent to the acknowledgementSetManager, otherwise a negative acknowledgement is sent.
+These are planned for future development.
 
 ## Developer Guide
 
-This plugin is compatible with Java 8. See
+This plugin is compatible with Java 11. See:
 
 - [CONTRIBUTING](https://github.com/opensearch-project/data-prepper/blob/main/CONTRIBUTING.md)
-- [monitoring](https://github.com/opensearch-project/data-prepper/blob/main/docs/monitoring.md)
-  The integration tests for this plugin do not run as part of the Data Prepper build.
+- [Monitoring](https://github.com/opensearch-project/data-prepper/blob/main/docs/monitoring.md)
 
-The following command runs the integration tests:
+### Build and Test
 
-```
+```bash
+# Run unit tests
+./gradlew :data-prepper-plugins:prometheus-sink:test
+
+# Run integration tests
 ./gradlew :data-prepper-plugins:prometheus-sink:integrationTest -Dtests.prometheus.sink.http.endpoint=<http-endpoint>
+
+# Code formatting
+./gradlew :data-prepper-plugins:prometheus-sink:spotlessApply
+
+# Checkstyle
+./gradlew :data-prepper-plugins:prometheus-sink:checkstyleMain :data-prepper-plugins:prometheus-sink:checkstyleTest
 ```

data-prepper-plugins/prometheus-sink/src/main/java/org/opensearch/dataprepper/plugins/sink/prometheus/PrometheusHttpSender.java

@@ -39,6 +39,7 @@
 import javax.annotation.Nonnull;
 import java.nio.charset.StandardCharsets;
 import java.net.URI;
+import java.util.Base64;
 import java.util.Set;
 
 /**
@@ -59,6 +60,7 @@ public class PrometheusHttpSender {
     private final long idleTimeoutMillis;
     private final CompressionEngine compressionEngine;
     private final PrometheusSinkConfiguration config;
+    private final String authHeader;
 
     /**
      * Constructor for the PrometheusHttpSender.
@@ -80,7 +82,16 @@ public PrometheusHttpSender(@Nonnull final AwsCredentialsSupplier awsCredentials
         this.config = config;
         this.connectionTimeoutMillis = config.getConnectionTimeout().toMillis();
         this.idleTimeoutMillis = config.getIdleTimeout().toMillis();
+        this.authHeader = buildAuthHeader(config);
+    }
 
+    private static String buildAuthHeader(final PrometheusSinkConfiguration config) {
+        if (config.getAuthentication() != null && config.getAuthentication().getHttpBasic() != null) {
+            final String credentials = config.getAuthentication().getHttpBasic().getUsername()
+                    + ":" + config.getAuthentication().getHttpBasic().getPassword();
+            return "Basic " + Base64.getEncoder().encodeToString(credentials.getBytes(StandardCharsets.UTF_8));
+        }
+        return null;
     }
 
     /**
@@ -160,16 +171,20 @@ public PrometheusPushResult pushToEndpoint(final byte[] payload) {
     }
 
     private SdkHttpFullRequest createSdkHttpRequest(final String url, @Nonnull final byte[] payload) {
-        return SdkHttpFullRequest.builder()
+        final SdkHttpFullRequest.Builder builder = SdkHttpFullRequest.builder()
                 .method(SdkHttpMethod.POST)
                 .uri(URI.create(url))
-                .putHeader("Content-Encoding", config.getEncoding().toString())
+                .putHeader("Content-Encoding", config.getEncoding().name().toLowerCase())
                 .putHeader("Content-Type", config.getContentType())
                 .putHeader("X-Prometheus-Remote-Write-Version", config.getRemoteWriteVersion())
-                .putHeader("x-amz-content-sha256","required")
-                .contentStreamProvider(() -> SdkBytes.fromByteArray(payload).asInputStream())
-                .build();
-
+                .contentStreamProvider(() -> SdkBytes.fromByteArray(payload).asInputStream());
+        if (authHeader != null) {
+            builder.putHeader("Authorization", authHeader);
+        }
+        if (signer != null) {
+            builder.putHeader("x-amz-content-sha256", "required");
+        }
+        return builder.build();
     }
 
     private HttpRequest buildHttpRequest(final byte[] payload) {

data-prepper-plugins/prometheus-sink/src/main/java/org/opensearch/dataprepper/plugins/sink/prometheus/configuration/PrometheusSinkConfiguration.java

@@ -37,10 +37,13 @@ public class PrometheusSinkConfiguration {
     private static final Duration DEFAULT_OUT_OF_ORDER_TIME_WINDOW = Duration.ofSeconds(10);
 
     @JsonProperty("aws")
-    @NotNull
     @Valid
     private AwsConfig awsConfig;
 
+    @JsonProperty("authentication")
+    @Valid
+    private AuthenticationOptions authentication;
+
     @NotNull
     @JsonProperty("url")
     private String url;
@@ -100,6 +103,10 @@ public AwsConfig getAwsConfig() {
         return awsConfig;
     }
 
+    public AuthenticationOptions getAuthentication() {
+        return authentication;
+    }
+
     public int getMaxRetries() {
         return maxRetries;
     }
@@ -132,11 +139,25 @@ public Duration getIdleTimeout() {
         return idleTimeout;
     }
 
-    @AssertTrue(message = "encoding or content_type or remote_write_version is incorrect.")
+    @AssertTrue(message = "Cannot use both AWS SigV4 and authentication options. Choose one.")
+    boolean isValidAuthConfig() {
+        return !(awsConfig != null && authentication != null);
+    }
+
+    @AssertTrue(message = "encoding or content_type or remote_write_version or url is incorrect.")
     boolean isValidConfig() {
-        return  url.startsWith("https://") &&
+        final boolean validUrl = url.startsWith("https://") || url.startsWith("http://");
+        return  validUrl &&
                 encoding == CompressionOption.SNAPPY &&
                 contentType.equals(X_PROTOBUF) &&
                 remoteWriteVersion.equals(DEFAULT_REMOTE_WRITE_VERSION);
     }
+
+    @AssertTrue(message = "AWS configuration requires an https:// URL.")
+    boolean isValidAwsConfig() {
+        if (awsConfig != null) {
+            return url != null && url.startsWith("https://");
+        }
+        return true;
+    }
 }