Added CloudWatchMeterRegistryProvider bean factory method

Signed-off-by: Steven Bayer <[email protected]>

data-prepper-core/src/main/java/com/amazon/dataprepper/parser/config/MetricsConfig.java

@@ -25,10 +25,12 @@
 import io.micrometer.prometheus.PrometheusMeterRegistry;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
+import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
 import software.amazon.awssdk.core.exception.SdkClientException;
 
+import javax.annotation.Nullable;
 import java.util.Collections;
 import java.util.List;
 
@@ -96,12 +98,29 @@ public MeterRegistry prometheusMeterRegistry(final DataPrepperConfiguration data
         }
     }
 
+    @Bean
+    public CloudWatchMeterRegistryProvider cloudWatchMeterRegistryProvider(
+            final DataPrepperConfiguration dataPrepperConfiguration
+    ) {
+        if (dataPrepperConfiguration.getMetricRegistryTypes().contains(MetricRegistryType.CloudWatch)) {
+            return new CloudWatchMeterRegistryProvider();
+        }
+        else {
+            return null;
+        }
+    }
+
     @Bean
     public MeterRegistry cloudWatchMeterRegistry(
             final DataPrepperConfiguration dataPrepperConfiguration,
-            final CloudWatchMeterRegistryProvider cloudWatchMeterRegistryProvider
+            @Autowired(required = false) @Nullable final CloudWatchMeterRegistryProvider cloudWatchMeterRegistryProvider
     ) {
         if (dataPrepperConfiguration.getMetricRegistryTypes().contains(MetricRegistryType.CloudWatch)) {
+            if (cloudWatchMeterRegistryProvider == null) {
+                throw new IllegalStateException(
+                        "configuration required configure cloudwatch meter registry but one could not be configured");
+            }
+
             try {
                 final CloudWatchMeterRegistry meterRegistry = cloudWatchMeterRegistryProvider.getCloudWatchMeterRegistry();
                 configureMetricRegistry(meterRegistry);

data-prepper-core/src/main/java/com/amazon/dataprepper/pipeline/server/CloudWatchMeterRegistryProvider.java

@@ -12,8 +12,6 @@
 import org.slf4j.LoggerFactory;
 import software.amazon.awssdk.services.cloudwatch.CloudWatchAsyncClient;
 
-import javax.inject.Inject;
-import javax.inject.Named;
 import java.io.IOException;
 import java.io.InputStream;
 import java.util.Properties;
@@ -27,14 +25,12 @@
  * {@link CloudWatchMeterRegistryProvider} also has a constructor with {@link CloudWatchAsyncClient} that will be used
  * for communication with Cloudwatch.
  */
-@Named
 public class CloudWatchMeterRegistryProvider {
     private static final String CLOUDWATCH_PROPERTIES = "cloudwatch.properties";
     private static final Logger LOG = LoggerFactory.getLogger(CloudWatchMeterRegistryProvider.class);
 
     private final CloudWatchMeterRegistry cloudWatchMeterRegistry;
 
-    @Inject
     public CloudWatchMeterRegistryProvider() {
         this(CLOUDWATCH_PROPERTIES, CloudWatchAsyncClient.create());
     }

data-prepper-core/src/main/java/com/amazon/dataprepper/pipeline/server/HttpServerProvider.java

@@ -55,6 +55,10 @@ public void configure(final HttpsParameters params) {
 
                 return server;
             } else {
+                LOG.warn("Creating Data Prepper server without TLS. This is not secure.");
+                LOG.warn("In order to set up TLS for the Data Prepper server, go here: " +
+                        "https://github.com/opensearch-project/data-prepper/blob/main/docs/configuration.md#server-configuration");
+
                 return HttpServer.create(socketAddress, 0);
             }
         } catch (final IOException ex) {

data-prepper-core/src/main/java/com/amazon/dataprepper/pipeline/server/config/DataPrepperServerConfiguration.java

@@ -27,7 +27,6 @@
 
 import javax.annotation.Nullable;
 import java.util.Collections;
-import java.util.Optional;
 
 @Configuration
 public class DataPrepperServerConfiguration {
@@ -52,13 +51,15 @@ private void createContext(
     public HttpServer httpServer(
             final HttpServerProvider httpServerProvider,
             final ListPipelinesHandler listPipelinesHandler,
+            final ShutdownHandler shutdownHandler,
             @Autowired(required = false) @Nullable final PrometheusMeterRegistry prometheusMeterRegistry,
             @Autowired(required = false) @Nullable final Authenticator authenticator
             ) {
 
         final HttpServer server = httpServerProvider.get();
 
         createContext(server, listPipelinesHandler, authenticator, "/list");
+        createContext(server, shutdownHandler, authenticator, "/shutdown");
 
         if (prometheusMeterRegistry != null) {
             final PrometheusMetricsHandler prometheusMetricsHandler = new PrometheusMetricsHandler(prometheusMeterRegistry);
@@ -75,14 +76,13 @@ private void printInsecurePluginModelWarning() {
     }
 
     @Bean
-    public PluginSetting pluginSetting(final Optional<PluginModel> optionalPluginModel) {
-        if (optionalPluginModel.isPresent()) {
-            final PluginModel pluginModel = optionalPluginModel.get();
-            final String pluginName = pluginModel.getPluginName();
+    public PluginSetting pluginSetting(@Autowired(required = false) final PluginModel authentication) {
+        if (authentication != null) {
+            final String pluginName = authentication.getPluginName();
             if (pluginName.equals(DataPrepperCoreAuthenticationProvider.UNAUTHENTICATED_PLUGIN_NAME)) {
                 printInsecurePluginModelWarning();
             }
-            return new PluginSetting(pluginName, pluginModel.getPluginSettings());
+            return new PluginSetting(pluginName, authentication.getPluginSettings());
         }
         else {
             printInsecurePluginModelWarning();
@@ -114,16 +114,7 @@ public ListPipelinesHandler listPipelinesHandler(final DataPrepper dataPrepper)
     }
 
     @Bean
-    public ShutdownHandler shutdownHandler(
-            final DataPrepper dataPrepper,
-            final Optional<Authenticator> optionalAuthenticator,
-            final HttpServer server
-    ) {
-        final ShutdownHandler shutdownHandler = new ShutdownHandler(dataPrepper);
-
-        final HttpContext context = server.createContext("/shutdown", shutdownHandler);
-        optionalAuthenticator.ifPresent(context::setAuthenticator);
-
-        return shutdownHandler;
+    public ShutdownHandler shutdownHandler(final DataPrepper dataPrepper) {
+        return new ShutdownHandler(dataPrepper);
     }
 }

data-prepper-core/src/test/java/com/amazon/dataprepper/pipeline/server/config/DataPrepperServerConfigurationTest.java

@@ -26,7 +26,6 @@
 
 import java.util.HashMap;
 import java.util.Map;
-import java.util.Optional;
 
 import static org.hamcrest.MatcherAssert.assertThat;
 import static org.hamcrest.Matchers.is;
@@ -53,17 +52,20 @@ class DataPrepperServerConfigurationTest {
     @Mock
     private ListPipelinesHandler listPipelinesHandler;
 
+    @Mock
+    private ShutdownHandler shutdownHandler;
+
     private final DataPrepperServerConfiguration serverConfiguration = new DataPrepperServerConfiguration();
 
     @Test
     public void testGivenNullPrometheusMeterRegistryAndNullAuthenticatorThenServerIsCreated() {
         when(httpServerProvider.get())
                 .thenReturn(httpServer);
-        final HttpServer server = serverConfiguration.httpServer(httpServerProvider, listPipelinesHandler, null, null);
+        final HttpServer server = serverConfiguration.httpServer(httpServerProvider, listPipelinesHandler, shutdownHandler, null, null);
 
         assertThat(server, is(httpServer));
         verify(server).createContext("/list", listPipelinesHandler);
-
+        verify(server).createContext(eq("/shutdown"), eq(shutdownHandler));
     }
 
     @Test
@@ -75,10 +77,16 @@ public void testGivenPrometheusMeterRegistryAndNullAuthenticatorThenServerIsCrea
         when(httpServer.createContext(any(String.class), any(HttpHandler.class)))
                 .thenReturn(context);
 
-        final HttpServer server = serverConfiguration.httpServer(httpServerProvider, listPipelinesHandler, meterRegistry, null);
+        final HttpServer server = serverConfiguration.httpServer(
+                httpServerProvider,
+                listPipelinesHandler,
+                shutdownHandler,
+                meterRegistry,
+                null);
 
         assertThat(server, is(httpServer));
         verify(server).createContext(eq("/list"), eq(listPipelinesHandler));
+        verify(server).createContext(eq("/shutdown"), eq(shutdownHandler));
         verify(server).createContext(eq("/metrics/prometheus"), any(PrometheusMetricsHandler.class));
         verify(server).createContext(eq("/metrics/sys"), any(PrometheusMetricsHandler.class));
         verifyNoInteractions(context);
@@ -94,18 +102,24 @@ public void testGivenPrometheusMeterRegistryAndAuthenticatorThenServerIsCreated(
         when(httpServer.createContext(any(String.class), any(HttpHandler.class)))
                 .thenReturn(context);
 
-        final HttpServer server = serverConfiguration.httpServer(httpServerProvider, listPipelinesHandler, meterRegistry, authenticator);
+        final HttpServer server = serverConfiguration.httpServer(
+                httpServerProvider,
+                listPipelinesHandler,
+                shutdownHandler,
+                meterRegistry,
+                authenticator);
 
         assertThat(server, is(httpServer));
         verify(server).createContext(eq("/list"), eq(listPipelinesHandler));
+        verify(server).createContext(eq("/shutdown"), eq(shutdownHandler));
         verify(server).createContext(eq("/metrics/prometheus"), any(PrometheusMetricsHandler.class));
         verify(server).createContext(eq("/metrics/sys"), any(PrometheusMetricsHandler.class));
-        verify(context, times(3)).setAuthenticator(eq(authenticator));
+        verify(context, times(4)).setAuthenticator(eq(authenticator));
     }
 
     @Test
     public void testGivingNoConfigThenCreateInsecureSettings() {
-        final PluginSetting pluginSetting = serverConfiguration.pluginSetting(Optional.empty());
+        final PluginSetting pluginSetting = serverConfiguration.pluginSetting(null);
 
         assertThat(pluginSetting.getName(), is("unauthenticated"));
         assertThat(pluginSetting.getSettings().isEmpty(), is(true));
@@ -118,7 +132,7 @@ public void testGivingInsecureConfigThenCreateInsecureSettings() {
         when(pluginModel.getPluginName())
                 .thenReturn("unauthenticated");
 
-        final PluginSetting pluginSetting = serverConfiguration.pluginSetting(Optional.of(pluginModel));
+        final PluginSetting pluginSetting = serverConfiguration.pluginSetting(pluginModel);
 
         assertThat(pluginSetting.getName(), is("unauthenticated"));
         assertThat(pluginSetting.getSettings().isEmpty(), is(true));
@@ -135,7 +149,7 @@ public void testGivingSecureConfigThenCreateInsecureSettings() {
         when(pluginModel.getPluginSettings())
                 .thenReturn(settings);
 
-        final PluginSetting pluginSetting = serverConfiguration.pluginSetting(Optional.of(pluginModel));
+        final PluginSetting pluginSetting = serverConfiguration.pluginSetting(pluginModel);
 
         assertThat(pluginSetting.getName(), is("super secure plugin"));
         assertThat(pluginSetting.getSettings(), is(settings));
@@ -189,34 +203,11 @@ public void testGivenValidInputWithAuthenticatorThenServerListContextCreated() {
     }
 
     @Test
-    public void testGivenValidInputWithNoAuthenticatorThenServerShutdownContextCreated() {
-        final DataPrepper dataPrepper = mock(DataPrepper.class);
-        final HttpServer server = mock(HttpServer.class);
-        final HttpContext context = mock(HttpContext.class);
-
-        when(server.createContext(eq("/shutdown"), any(ShutdownHandler.class)))
-                .thenReturn(context);
-
-        final ShutdownHandler handler = serverConfiguration.shutdownHandler(dataPrepper, Optional.empty(), server);
-
-        assertThat(handler, isA(ShutdownHandler.class));
-        verifyNoInteractions(context);
-    }
-
-    @Test
-    public void testGivenValidInputWithAuthenticatorThenServerShutdownContextCreated() {
+    public void testShutdownHandlerIsCreated() {
         final DataPrepper dataPrepper = mock(DataPrepper.class);
-        final Authenticator authenticator = mock(Authenticator.class);
-        final HttpServer server = mock(HttpServer.class);
-        final HttpContext context = mock(HttpContext.class);
-
-        when(server.createContext(eq("/shutdown"), any(ShutdownHandler.class)))
-                .thenReturn(context);
 
-        final ShutdownHandler handler = serverConfiguration.shutdownHandler(dataPrepper, Optional.of(authenticator), server);
+        final ShutdownHandler handler = serverConfiguration.shutdownHandler(dataPrepper);
 
         assertThat(handler, isA(ShutdownHandler.class));
-        verify(context)
-                .setAuthenticator(eq(authenticator));
     }
 }