Add insecure flag to Prometheus sink to require HTTPS by default (#6688)

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

Author: ps48

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

@@ -8,7 +8,7 @@ The Prometheus sink should be configured as part of a Data Prepper pipeline YAML
 
 ### Open Source Prometheus (No Auth)
 
-To use with a vanilla Prometheus instance, provide an `http://` or `https://` URL. No `aws` block is needed.
+To use with a vanilla Prometheus instance, provide an `https://` URL. If using `http://`, set `insecure: true`. No `aws` block is needed.
 
 Prometheus must be started with the `--web.enable-remote-write-receiver` flag.
 
@@ -18,8 +18,9 @@ pipeline:
   sink:
     - prometheus:
         url: "http://localhost:9090/api/v1/write"
+        insecure: true
         threshold:
-          max_events: 500
+          max_events: 1000
           flush_interval: 5s
 ```
 
@@ -32,7 +33,7 @@ pipeline:
   ...
   sink:
     - prometheus:
-        url: "http://localhost:9090/api/v1/write"
+        url: "https://localhost:9090/api/v1/write"
         authentication:
           http_basic:
             username: "promuser"
@@ -53,7 +54,7 @@ pipeline:
           region: "us-east-2"
           sts_role_arn: "arn:aws:iam::123456789012:role/data-prepper-prometheus-role"
         threshold:
-          max_events: 500
+          max_events: 1000
           flush_interval: 5s
 ```
 
@@ -72,12 +73,13 @@ pipeline:
 
 | Option | Description |
 |--------|-------------|
-| `url` | The Prometheus Remote Write endpoint URL. Supports `http://` and `https://` schemes. When `aws` is configured, `https://` is required. |
+| `url` | The Prometheus Remote Write endpoint URL. Supports `https://` by default. To use `http://`, set `insecure` to `true`. When `aws` is configured, `https://` is required. |
 
 ### Optional
 
 | Option | Default | Description |
 |--------|---------|-------------|
+| `insecure` | `false` | When `true`, allows `http://` URLs. By default, only `https://` URLs are permitted. |
 | `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. |
@@ -95,7 +97,7 @@ pipeline:
 
 | Option | Default | Description |
 |--------|---------|-------------|
-| `max_events` | `500` | Maximum number of events to buffer before flushing. |
+| `max_events` | `1000` | 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. |
 

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

@@ -81,9 +81,16 @@ public class PrometheusSinkConfiguration {
     @DurationMax(seconds = 600)
     private Duration idleTimeout = DEFAULT_IDLE_TIMEOUT;
 
+    @JsonProperty("insecure")
+    private boolean insecure = false;
+
     @JsonProperty("sanitize_names")
     private boolean sanitizeNames = true;
 
+    public boolean isInsecure() {
+        return insecure;
+    }
+
     public boolean getSanitizeNames() {
         return sanitizeNames;
     }
@@ -139,6 +146,17 @@ public Duration getIdleTimeout() {
         return idleTimeout;
     }
 
+    @AssertTrue(message = "url must be https when insecure is not set to true.")
+    boolean isHttpsOrInsecure() {
+        if (url == null) {
+            return true;
+        }
+        if (insecure) {
+            return true;
+        }
+        return url.startsWith("https://");
+    }
+
     @AssertTrue(message = "Cannot use both AWS SigV4 and authentication options. Choose one.")
     boolean isValidAuthConfig() {
         return !(awsConfig != null && authentication != null);

data-prepper-plugins/prometheus-sink/src/test/java/org/opensearch/dataprepper/plugins/sink/prometheus/configuration/PrometheusSinkConfigurationTest.java

@@ -123,14 +123,16 @@ void prometheus_sink_config_test_with_invalid_url() throws JsonProcessingExcepti
     }
 
     @Test
-    void prometheus_sink_config_test_with_http_url_is_valid() throws JsonProcessingException {
+    void prometheus_sink_config_test_with_http_url_and_insecure_is_valid() throws JsonProcessingException {
         final String HTTP_SINK_YAML =
             " url: \"http://localhost:8080/test\"\n" +
+                    " insecure: true\n" +
                     " encoding: \"snappy\" \n" +
                     " remote_write_version: \"0.1.0\" \n" +
                     " content_type: \"application/x-protobuf\" \n";
         final PrometheusSinkConfiguration prometheusSinkConfiguration = objectMapper.readValue(HTTP_SINK_YAML, PrometheusSinkConfiguration.class);
         assertTrue(prometheusSinkConfiguration.isValidConfig());
+        assertTrue(prometheusSinkConfiguration.isHttpsOrInsecure());
     }
 
     @Test
@@ -148,6 +150,7 @@ void prometheus_sink_config_test_without_aws_config_returns_null() throws JsonPr
     void prometheus_sink_config_test_with_basic_auth() throws JsonProcessingException {
         final String AUTH_SINK_YAML =
             " url: \"http://localhost:9090/api/v1/write\"\n" +
+                    " insecure: true\n" +
                     " encoding: \"snappy\" \n" +
                     " remote_write_version: \"0.1.0\" \n" +
                     " content_type: \"application/x-protobuf\" \n" +
@@ -164,6 +167,7 @@ void prometheus_sink_config_test_with_basic_auth() throws JsonProcessingExceptio
     void prometheus_sink_config_test_without_auth_returns_null() throws JsonProcessingException {
         final String NO_AUTH_YAML =
             " url: \"http://localhost:9090/api/v1/write\"\n" +
+                    " insecure: true\n" +
                     " encoding: \"snappy\" \n" +
                     " remote_write_version: \"0.1.0\" \n" +
                     " content_type: \"application/x-protobuf\" \n";
@@ -192,6 +196,7 @@ void prometheus_sink_config_test_aws_and_auth_is_invalid() throws JsonProcessing
     void prometheus_sink_config_test_basic_auth_without_aws_is_valid() throws JsonProcessingException {
         final String AUTH_ONLY_YAML =
             " url: \"http://localhost:9090/api/v1/write\"\n" +
+                    " insecure: true\n" +
                     " encoding: \"snappy\" \n" +
                     " remote_write_version: \"0.1.0\" \n" +
                     " content_type: \"application/x-protobuf\" \n" +
@@ -207,6 +212,7 @@ void prometheus_sink_config_test_basic_auth_without_aws_is_valid() throws JsonPr
     void prometheus_sink_config_test_bearer_token_is_rejected() throws JsonProcessingException {
         final String BEARER_TOKEN_YAML =
             " url: \"http://localhost:9090/api/v1/write\"\n" +
+                    " insecure: true\n" +
                     " encoding: \"snappy\" \n" +
                     " remote_write_version: \"0.1.0\" \n" +
                     " content_type: \"application/x-protobuf\" \n" +
@@ -221,13 +227,66 @@ void prometheus_sink_config_test_bearer_token_is_rejected() throws JsonProcessin
     void prometheus_sink_config_test_without_bearer_token_is_valid() throws JsonProcessingException {
         final String NO_BEARER_YAML =
             " url: \"http://localhost:9090/api/v1/write\"\n" +
+                    " insecure: true\n" +
                     " encoding: \"snappy\" \n" +
                     " remote_write_version: \"0.1.0\" \n" +
                     " content_type: \"application/x-protobuf\" \n";
         final PrometheusSinkConfiguration config = objectMapper.readValue(NO_BEARER_YAML, PrometheusSinkConfiguration.class);
         assertTrue(config.isValidBearerTokenConfig());
     }
 
+    @Test
+    void prometheus_sink_config_insecure_defaults_to_false() {
+        final PrometheusSinkConfiguration config = new PrometheusSinkConfiguration();
+        assertFalse(config.isInsecure());
+    }
+
+    @Test
+    void prometheus_sink_config_http_url_without_insecure_is_invalid() throws JsonProcessingException {
+        final String HTTP_YAML =
+            " url: \"http://localhost:9090/api/v1/write\"\n" +
+                    " encoding: \"snappy\" \n" +
+                    " remote_write_version: \"0.1.0\" \n" +
+                    " content_type: \"application/x-protobuf\" \n";
+        final PrometheusSinkConfiguration config = objectMapper.readValue(HTTP_YAML, PrometheusSinkConfiguration.class);
+        assertFalse(config.isHttpsOrInsecure());
+    }
+
+    @Test
+    void prometheus_sink_config_http_url_with_insecure_true_is_valid() throws JsonProcessingException {
+        final String HTTP_YAML =
+            " url: \"http://localhost:9090/api/v1/write\"\n" +
+                    " insecure: true\n" +
+                    " encoding: \"snappy\" \n" +
+                    " remote_write_version: \"0.1.0\" \n" +
+                    " content_type: \"application/x-protobuf\" \n";
+        final PrometheusSinkConfiguration config = objectMapper.readValue(HTTP_YAML, PrometheusSinkConfiguration.class);
+        assertTrue(config.isHttpsOrInsecure());
+    }
+
+    @Test
+    void prometheus_sink_config_https_url_without_insecure_is_valid() throws JsonProcessingException {
+        final String HTTPS_YAML =
+            " url: \"https://localhost:9090/api/v1/write\"\n" +
+                    " encoding: \"snappy\" \n" +
+                    " remote_write_version: \"0.1.0\" \n" +
+                    " content_type: \"application/x-protobuf\" \n";
+        final PrometheusSinkConfiguration config = objectMapper.readValue(HTTPS_YAML, PrometheusSinkConfiguration.class);
+        assertTrue(config.isHttpsOrInsecure());
+    }
+
+    @Test
+    void prometheus_sink_config_https_url_with_insecure_true_is_valid() throws JsonProcessingException {
+        final String HTTPS_YAML =
+            " url: \"https://localhost:9090/api/v1/write\"\n" +
+                    " insecure: true\n" +
+                    " encoding: \"snappy\" \n" +
+                    " remote_write_version: \"0.1.0\" \n" +
+                    " content_type: \"application/x-protobuf\" \n";
+        final PrometheusSinkConfiguration config = objectMapper.readValue(HTTPS_YAML, PrometheusSinkConfiguration.class);
+        assertTrue(config.isHttpsOrInsecure());
+    }
+
     @Test
     void prometheus_sink_config_test_aws_with_http_url_is_invalid() throws JsonProcessingException {
         final String AWS_HTTP_SINK_YAML =