Di server

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

@@ -5,6 +5,7 @@
 
 package com.amazon.dataprepper.parser.config;
 
+import com.amazon.dataprepper.model.configuration.PluginModel;
 import com.amazon.dataprepper.parser.model.DataPrepperConfiguration;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import org.slf4j.Logger;
@@ -29,7 +30,7 @@ public DataPrepperArgs dataPrepperArgs(final Environment environment) {
         LOG.info("Command line args: {}", commandLineArgs);
 
         if (commandLineArgs != null) {
-            String[] args = commandLineArgs.split(COMMAND_LINE_ARG_DELIMITER);
+            final String[] args = commandLineArgs.split(COMMAND_LINE_ARG_DELIMITER);
             return new DataPrepperArgs(args);
         }
         else {
@@ -55,4 +56,9 @@ public DataPrepperConfiguration dataPrepperConfiguration(
             return new DataPrepperConfiguration();
         }
     }
+
+    @Bean
+    public PluginModel authentication(final DataPrepperConfiguration dataPrepperConfiguration) {
+        return dataPrepperConfiguration.getAuthentication();
+    }
 }

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

@@ -31,7 +31,7 @@ private static String checkForLogstashConfigurationAndConvert(String configurati
             try {
                 configurationFileLocation = logstashConfigConverter.convertLogstashConfigurationToPipeline(
                         configurationFileLocation, String.valueOf(configurationDirectory));
-            } catch (IOException e) {
+            } catch (final IOException e) {
                 LOG.warn("Unable to read the Logstash configuration file", e);
                 throw new IllegalArgumentException("Invalid Logstash configuration file", e);
             }
@@ -51,7 +51,7 @@ else if (args.length > MAXIMUM_SUPPORTED_NUMBER_OF_ARGS) {
                     "Data Prepper supports a maximum of " + MAXIMUM_SUPPORTED_NUMBER_OF_ARGS + " command line arguments");
         }
 
-        String configurationFileLocation = args[DATA_PREPPER_PIPELINE_CONFIG_POSITON];
+        final String configurationFileLocation = args[DATA_PREPPER_PIPELINE_CONFIG_POSITON];
         LOG.info("Using {} configuration file", configurationFileLocation);
 
         this.pipelineConfigFileLocation = DataPrepperArgs.checkForLogstashConfigurationAndConvert(configurationFileLocation);

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

@@ -7,24 +7,41 @@
 
 import com.amazon.dataprepper.parser.model.DataPrepperConfiguration;
 import com.amazon.dataprepper.parser.model.MetricRegistryType;
+import com.amazon.dataprepper.pipeline.server.CloudWatchMeterRegistryProvider;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.fasterxml.jackson.dataformat.yaml.YAMLFactory;
+import io.micrometer.cloudwatch2.CloudWatchMeterRegistry;
 import io.micrometer.core.instrument.MeterRegistry;
 import io.micrometer.core.instrument.Metrics;
+import io.micrometer.core.instrument.Tag;
 import io.micrometer.core.instrument.binder.MeterBinder;
 import io.micrometer.core.instrument.binder.jvm.ClassLoaderMetrics;
 import io.micrometer.core.instrument.binder.jvm.JvmGcMetrics;
 import io.micrometer.core.instrument.binder.jvm.JvmMemoryMetrics;
 import io.micrometer.core.instrument.binder.jvm.JvmThreadMetrics;
 import io.micrometer.core.instrument.binder.system.ProcessorMetrics;
 import io.micrometer.core.instrument.composite.CompositeMeterRegistry;
+import io.micrometer.prometheus.PrometheusConfig;
+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;
 
+import static com.amazon.dataprepper.DataPrepper.getServiceNameForMetrics;
+import static com.amazon.dataprepper.metrics.MetricNames.SERVICE_NAME;
+
 @Configuration
 public class MetricsConfig {
+    private static final Logger LOG = LoggerFactory.getLogger(MetricsConfig.class);
+    private static final String METRICS_CONTEXT_PREFIX = "/metrics";
+
     @Bean
     public YAMLFactory yamlFactory() {
         return new YAMLFactory();
@@ -60,21 +77,80 @@ public JvmThreadMetrics jvmThreadMetrics() {
         return new JvmThreadMetrics();
     }
 
+    private void configureMetricRegistry(final MeterRegistry meterRegistry) {
+        meterRegistry.config()
+                .commonTags(Collections.singletonList(
+                        Tag.of(SERVICE_NAME, getServiceNameForMetrics())
+                ));
+
+    }
+
+    @Bean
+    public MeterRegistry prometheusMeterRegistry(final DataPrepperConfiguration dataPrepperConfiguration) {
+        if (dataPrepperConfiguration.getMetricRegistryTypes().contains(MetricRegistryType.Prometheus)) {
+            final PrometheusMeterRegistry meterRegistry = new PrometheusMeterRegistry(PrometheusConfig.DEFAULT);
+            configureMetricRegistry(meterRegistry);
+
+            return meterRegistry;
+        }
+        else {
+            return null;
+        }
+    }
+
+    @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,
+            @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);
+
+                return meterRegistry;
+            } catch (final SdkClientException e) {
+                LOG.warn("Unable to configure Cloud Watch Meter Registry but Meter Registry was requested in Data Prepper Configuration");
+                throw new RuntimeException("Unable to initialize Cloud Watch Meter Registry", e);
+            }
+        }
+        else {
+            return null;
+        }
+    }
+
     @Bean
     public CompositeMeterRegistry systemMeterRegistry(
             final List<MeterBinder> meterBinders,
-            final DataPrepperConfiguration dataPrepperConfiguration
+            final List<MeterRegistry> meterRegistries
     ) {
-        final CompositeMeterRegistry meterRegistry = new CompositeMeterRegistry();
+        final CompositeMeterRegistry compositeMeterRegistry = new CompositeMeterRegistry();
 
-        meterBinders.forEach(binder -> binder.bindTo(meterRegistry));
+        LOG.debug("{} Meter Binder beans registered.", meterBinders.size());
+        meterBinders.forEach(binder -> binder.bindTo(compositeMeterRegistry));
 
-        dataPrepperConfiguration.getMetricRegistryTypes().forEach(metricRegistryType -> {
-            MeterRegistry registryForType = MetricRegistryType.getDefaultMeterRegistryForType(metricRegistryType);
-            meterRegistry.add(registryForType);
-            Metrics.addRegistry(registryForType);
+        meterRegistries.forEach(meterRegistry -> {
+            compositeMeterRegistry.add(meterRegistry);
+            Metrics.addRegistry(meterRegistry);
         });
 
-        return meterRegistry;
+        return compositeMeterRegistry;
     }
 }

data-prepper-core/src/main/java/com/amazon/dataprepper/parser/model/MetricRegistryType.java

@@ -5,35 +5,7 @@
 
 package com.amazon.dataprepper.parser.model;
 
-import com.amazon.dataprepper.pipeline.server.CloudWatchMeterRegistryProvider;
-import io.micrometer.core.instrument.MeterRegistry;
-import io.micrometer.core.instrument.Tag;
-import io.micrometer.prometheus.PrometheusConfig;
-import io.micrometer.prometheus.PrometheusMeterRegistry;
-
-import java.util.Arrays;
-
-import static com.amazon.dataprepper.DataPrepper.getServiceNameForMetrics;
-import static com.amazon.dataprepper.metrics.MetricNames.SERVICE_NAME;
-import static java.lang.String.format;
-
 public enum MetricRegistryType {
     Prometheus,
     CloudWatch;
-
-    public static MeterRegistry getDefaultMeterRegistryForType(final MetricRegistryType metricRegistryType) {
-        MeterRegistry meterRegistry = null;
-        switch (metricRegistryType) {
-            case Prometheus:
-                meterRegistry = new PrometheusMeterRegistry(PrometheusConfig.DEFAULT);
-                break;
-            case CloudWatch:
-                meterRegistry = new CloudWatchMeterRegistryProvider().getCloudWatchMeterRegistry();
-                break;
-            default:
-                throw new IllegalArgumentException(format("Invalid metricRegistryType %s", metricRegistryType));
-        }
-        meterRegistry.config().commonTags(Arrays.asList(Tag.of(SERVICE_NAME, getServiceNameForMetrics())));
-        return meterRegistry;
-    }
 }

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

@@ -55,22 +55,17 @@ public CloudWatchMeterRegistry getCloudWatchMeterRegistry() {
      * Returns CloudWatchConfig using the properties from {@link #CLOUDWATCH_PROPERTIES}
      */
     private CloudWatchConfig createCloudWatchConfig(final String cloudWatchPropertiesFilePath) {
-        CloudWatchConfig cloudWatchConfig = null;
         try (final InputStream inputStream = requireNonNull(getClass().getClassLoader()
                 .getResourceAsStream(cloudWatchPropertiesFilePath))) {
             final Properties cloudwatchProperties = new Properties();
             cloudwatchProperties.load(inputStream);
-            cloudWatchConfig = new CloudWatchConfig() {
-                @Override
-                public String get(final String key) {
-                    return cloudwatchProperties.getProperty(key);
-                }
-            };
-        } catch (IOException ex) {
+            return cloudwatchProperties::getProperty;
+        } catch (final IOException ex) {
             LOG.error("Encountered exception in creating CloudWatchConfig for CloudWatchMeterRegistry, " +
                     "Proceeding without metrics", ex);
+
             //If there is no registry attached, micrometer will make NoopMeters which are discarded.
+            return null;
         }
-        return cloudWatchConfig;
     }
 }

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

@@ -5,32 +5,12 @@
 
 package com.amazon.dataprepper.pipeline.server;
 
-import com.amazon.dataprepper.DataPrepper;
-import com.amazon.dataprepper.model.configuration.PluginModel;
-import com.amazon.dataprepper.model.configuration.PluginSetting;
-import com.amazon.dataprepper.model.plugin.PluginFactory;
-import com.amazon.dataprepper.parser.model.DataPrepperConfiguration;
-import com.sun.net.httpserver.Authenticator;
 import com.sun.net.httpserver.HttpServer;
-import com.sun.net.httpserver.HttpsConfigurator;
-import com.sun.net.httpserver.HttpsParameters;
-import com.sun.net.httpserver.HttpsServer;
-import io.micrometer.core.instrument.MeterRegistry;
-import io.micrometer.core.instrument.Metrics;
-import io.micrometer.core.instrument.composite.CompositeMeterRegistry;
-import io.micrometer.prometheus.PrometheusMeterRegistry;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import javax.inject.Inject;
 import javax.inject.Named;
-import javax.net.ssl.SSLContext;
-import javax.net.ssl.SSLParameters;
-import java.io.IOException;
-import java.net.InetSocketAddress;
-import java.util.Collections;
-import java.util.Optional;
-import java.util.Set;
 
 
 
@@ -45,70 +25,9 @@ public class DataPrepperServer {
 
     @Inject
     public DataPrepperServer(
-            final DataPrepperConfiguration dataPrepperConfiguration,
-            final PluginFactory pluginFactory,
-            final DataPrepper dataPrepper,
-            final CompositeMeterRegistry systemMeterRegistry
+            final HttpServer server
     ) {
-        final int port = dataPrepperConfiguration.getServerPort();
-        final boolean ssl = dataPrepperConfiguration.ssl();
-        final String keyStoreFilePath = dataPrepperConfiguration.getKeyStoreFilePath();
-        final String keyStorePassword = dataPrepperConfiguration.getKeyStorePassword();
-        final String privateKeyPassword = dataPrepperConfiguration.getPrivateKeyPassword();
-
-        final PluginModel authenticationConfiguration = dataPrepperConfiguration.getAuthentication();
-        final PluginSetting authenticationPluginSetting;
-
-        if (authenticationConfiguration == null || authenticationConfiguration.getPluginName().equals(DataPrepperCoreAuthenticationProvider.UNAUTHENTICATED_PLUGIN_NAME)) {
-            LOG.warn("Creating data prepper server without authentication. This is not secure.");
-            LOG.warn("In order to set up Http Basic authentication for the data prepper server, go here: https://github.com/opensearch-project/data-prepper/blob/main/docs/core_apis.md#authentication");
-        }
-
-        if(authenticationConfiguration != null) {
-            authenticationPluginSetting =
-                    new PluginSetting(authenticationConfiguration.getPluginName(), authenticationConfiguration.getPluginSettings());
-        } else {
-            authenticationPluginSetting =
-                    new PluginSetting(DataPrepperCoreAuthenticationProvider.UNAUTHENTICATED_PLUGIN_NAME, Collections.emptyMap());
-        }
-
-        final DataPrepperCoreAuthenticationProvider authenticationProvider = pluginFactory.loadPlugin(DataPrepperCoreAuthenticationProvider.class, authenticationPluginSetting);
-        final Authenticator authenticator = authenticationProvider.getAuthenticator();
-
-        try {
-            if (ssl) {
-                LOG.info("Creating Data Prepper server with TLS");
-                this.server = createHttpsServer(port, keyStoreFilePath, keyStorePassword, privateKeyPassword);
-            } 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");
-                this.server = createHttpServer(port);
-            }
-        } catch (final IOException ex) {
-            throw new RuntimeException("Failed to create server", ex);
-        }
-
-        getPrometheusMeterRegistryFromRegistries(Metrics.globalRegistry.getRegistries()).ifPresent(meterRegistry -> {
-            final PrometheusMeterRegistry prometheusMeterRegistryForDataPrepper = (PrometheusMeterRegistry) meterRegistry;
-            server.createContext("/metrics/prometheus", new PrometheusMetricsHandler(prometheusMeterRegistryForDataPrepper))
-                    .setAuthenticator(authenticator);
-        });
-
-        getPrometheusMeterRegistryFromRegistries(systemMeterRegistry.getRegistries()).ifPresent(
-                meterRegistry -> {
-                    final PrometheusMeterRegistry prometheusMeterRegistryForSystem = (PrometheusMeterRegistry) meterRegistry;
-                    server.createContext("/metrics/sys", new PrometheusMetricsHandler(prometheusMeterRegistryForSystem))
-                            .setAuthenticator(authenticator);
-                });
-        server.createContext("/list", new ListPipelinesHandler(dataPrepper))
-                .setAuthenticator(authenticator);
-        server.createContext("/shutdown", new ShutdownHandler(dataPrepper))
-                .setAuthenticator(authenticator);
-    }
-
-    private Optional<MeterRegistry> getPrometheusMeterRegistryFromRegistries(final Set<MeterRegistry> meterRegistries) {
-        return meterRegistries.stream().filter(meterRegistry ->
-                meterRegistry instanceof PrometheusMeterRegistry).findFirst();
+        this.server = server;
     }
 
     /**
@@ -117,7 +36,6 @@ private Optional<MeterRegistry> getPrometheusMeterRegistryFromRegistries(final S
     public void start() {
         server.start();
         LOG.info("Data Prepper server running at :{}", server.getAddress().getPort());
-
     }
 
     /**
@@ -127,26 +45,4 @@ public void stop() {
         server.stop(0);
         LOG.info("Data Prepper server stopped");
     }
-
-    private HttpServer createHttpServer(final int port) throws IOException {
-        return HttpServer.create(new InetSocketAddress(port), 0);
-    }
-
-    private HttpServer createHttpsServer(final int port,
-                                         final String keyStoreFilePath,
-                                         final String keyStorePassword,
-                                         final String privateKeyPassword) throws IOException {
-        final SSLContext sslContext = SslUtil.createSslContext(keyStoreFilePath, keyStorePassword, privateKeyPassword);
-
-        final HttpsServer server = HttpsServer.create(new InetSocketAddress(port), 0);
-        server.setHttpsConfigurator(new HttpsConfigurator(sslContext) {
-            public void configure(HttpsParameters params) {
-                SSLContext context = getSSLContext();
-                SSLParameters sslparams = context.getDefaultSSLParameters();
-                params.setSSLParameters(sslparams);
-            }
-        });
-
-        return server;
-    }
 }