Details
diff --git a/azkaban-common/src/main/java/azkaban/AzkabanCommonModule.java b/azkaban-common/src/main/java/azkaban/AzkabanCommonModule.java
index 11424f6..3c5722a 100644
--- a/azkaban-common/src/main/java/azkaban/AzkabanCommonModule.java
+++ b/azkaban-common/src/main/java/azkaban/AzkabanCommonModule.java
@@ -33,7 +33,11 @@ import java.io.File;
import static azkaban.storage.StorageImplementationType.*;
-
+/**
+ * This Guice module is currently a one place container for all bindings in the current module. This is intended to
+ * help during the migration process to Guice. Once this class starts growing we can move towards more modular
+ * structuring of Guice components.
+ */
public class AzkabanCommonModule extends AbstractModule {
private final Props props;
/**
@@ -68,6 +72,7 @@ public class AzkabanCommonModule extends AbstractModule {
}
}
+ @SuppressWarnings("unchecked")
private Class<? extends Storage> loadCustomStorageClass(String storageImplementation) {
try {
return (Class<? extends Storage>) Class.forName(storageImplementation);
diff --git a/azkaban-common/src/main/java/azkaban/Constants.java b/azkaban-common/src/main/java/azkaban/Constants.java
index bc63436..663475b 100644
--- a/azkaban-common/src/main/java/azkaban/Constants.java
+++ b/azkaban-common/src/main/java/azkaban/Constants.java
@@ -45,6 +45,7 @@ public class Constants {
public static final int MEMORY_CHECK_RETRY_LIMIT = 720;
public static final int DEFAULT_PORT_NUMBER = 8081;
public static final int DEFAULT_SSL_PORT_NUMBER = 8443;
+ public static final int DEFAULT_JETTY_MAX_THREAD_COUNT = 20;
public static class ConfigurationKeys {
// These properties are configurable through azkaban.properties
diff --git a/azkaban-exec-server/src/main/java/azkaban/execapp/AzkabanExecServerModule.java b/azkaban-exec-server/src/main/java/azkaban/execapp/AzkabanExecServerModule.java
new file mode 100644
index 0000000..c8803f9
--- /dev/null
+++ b/azkaban-exec-server/src/main/java/azkaban/execapp/AzkabanExecServerModule.java
@@ -0,0 +1,34 @@
+/*
+ * Copyright 2017 LinkedIn Corp.
+ *
+ * Licensed 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.
+ *
+ */
+
+package azkaban.execapp;
+
+import com.google.inject.AbstractModule;
+import com.google.inject.Scopes;
+
+
+/**
+ * This Guice module is currently a one place container for all bindings in the current module. This is intended to
+ * help during the migration process to Guice. Once this class starts growing we can move towards more modular
+ * structuring of Guice components.
+ */
+public class AzkabanExecServerModule extends AbstractModule {
+ @Override
+ protected void configure() {
+ bind(AzkabanExecutorServer.class).in(Scopes.SINGLETON);
+ }
+}
diff --git a/azkaban-exec-server/src/main/java/azkaban/execapp/AzkabanExecutorServer.java b/azkaban-exec-server/src/main/java/azkaban/execapp/AzkabanExecutorServer.java
index a6f6bff..54068a9 100644
--- a/azkaban-exec-server/src/main/java/azkaban/execapp/AzkabanExecutorServer.java
+++ b/azkaban-exec-server/src/main/java/azkaban/execapp/AzkabanExecutorServer.java
@@ -16,8 +16,11 @@
package azkaban.execapp;
+import azkaban.AzkabanCommonModule;
import com.google.common.base.Throwables;
+import com.google.inject.Guice;
+import com.google.inject.Inject;
import org.apache.commons.lang.StringUtils;
import org.apache.log4j.Logger;
import org.joda.time.DateTimeZone;
@@ -73,6 +76,7 @@ import azkaban.utils.Utils;
import azkaban.metrics.MetricsManager;
import static azkaban.Constants.AZKABAN_EXECUTOR_PORT_FILENAME;
+import static azkaban.ServiceProvider.*;
import static com.google.common.base.Preconditions.checkState;
import static java.util.Objects.requireNonNull;
@@ -104,6 +108,7 @@ public class AzkabanExecutorServer {
*
* @throws Exception
*/
+ @Inject
public AzkabanExecutorServer(Props props) throws Exception {
this.props = props;
server = createJettyServer(props);
@@ -347,14 +352,25 @@ public class AzkabanExecutorServer {
StdOutErrRedirect.redirectOutAndErrToLog();
logger.info("Starting Jetty Azkaban Executor...");
- Props azkabanSettings = AzkabanServer.loadProps(args);
+ Props props = AzkabanServer.loadProps(args);
- if (azkabanSettings == null) {
+ if (props == null) {
logger.error("Azkaban Properties not loaded.");
logger.error("Exiting Azkaban Executor Server...");
return;
}
+
+ /* Initialize Guice Injector */
+ SERVICE_PROVIDER.setInjector(Guice.createInjector(
+ new AzkabanCommonModule(props),
+ new AzkabanExecServerModule()
+ ));
+
+ launch(props);
+ }
+
+ public static void launch(Props azkabanSettings) throws Exception {
// Setup time zone
if (azkabanSettings.containsKey(DEFAULT_TIMEZONE_ID)) {
String timezone = azkabanSettings.getString(DEFAULT_TIMEZONE_ID);
@@ -365,7 +381,7 @@ public class AzkabanExecutorServer {
logger.info("Setting timezone to " + timezone);
}
- app = new AzkabanExecutorServer(azkabanSettings);
+ app = SERVICE_PROVIDER.getInstance(AzkabanExecutorServer.class);
Runtime.getRuntime().addShutdownHook(new Thread() {
diff --git a/azkaban-solo-server/src/main/java/azkaban/soloserver/AzkabanSingleServer.java b/azkaban-solo-server/src/main/java/azkaban/soloserver/AzkabanSingleServer.java
index 37fc8d7..6d0b6ca 100644
--- a/azkaban-solo-server/src/main/java/azkaban/soloserver/AzkabanSingleServer.java
+++ b/azkaban-solo-server/src/main/java/azkaban/soloserver/AzkabanSingleServer.java
@@ -16,6 +16,10 @@
package azkaban.soloserver;
+import azkaban.AzkabanCommonModule;
+import azkaban.execapp.AzkabanExecServerModule;
+import azkaban.webapp.AzkabanWebServerModule;
+import com.google.inject.Guice;
import org.apache.log4j.Logger;
import azkaban.database.AzkabanDatabaseSetup;
@@ -25,6 +29,9 @@ import azkaban.server.AzkabanServer;
import azkaban.webapp.AzkabanWebServer;
import azkaban.utils.Props;
+import static azkaban.ServiceProvider.*;
+
+
public class AzkabanSingleServer {
private static final Logger logger = Logger.getLogger(AzkabanWebServer.class);
@@ -38,21 +45,23 @@ public class AzkabanSingleServer {
return;
}
- boolean checkversion =
- props.getBoolean(AzkabanDatabaseSetup.DATABASE_CHECK_VERSION, true);
-
- if (checkversion) {
- boolean updateDB =
- props.getBoolean(AzkabanDatabaseSetup.DATABASE_AUTO_UPDATE_TABLES,
- true);
- String scriptDir =
- props.getString(AzkabanDatabaseSetup.DATABASE_SQL_SCRIPT_DIR, "sql");
+ if (props.getBoolean(AzkabanDatabaseSetup.DATABASE_CHECK_VERSION, true)) {
+ boolean updateDB = props.getBoolean(AzkabanDatabaseSetup.DATABASE_AUTO_UPDATE_TABLES, true);
+ String scriptDir = props.getString(AzkabanDatabaseSetup.DATABASE_SQL_SCRIPT_DIR, "sql");
AzkabanDatabaseUpdater.runDatabaseUpdater(props, scriptDir, updateDB);
}
- AzkabanWebServer.main(args);
+ /* Initialize Guice Injector */
+ SERVICE_PROVIDER.setInjector(Guice.createInjector(
+ new AzkabanCommonModule(props),
+ new AzkabanWebServerModule(),
+ new AzkabanExecServerModule()
+ ));
+
+ AzkabanWebServer.launch(props);
logger.info("Azkaban Web Server started...");
- AzkabanExecutorServer.main(args);
+
+ AzkabanExecutorServer.launch(props);
logger.info("Azkaban Exec Server started...");
}
}
diff --git a/azkaban-web-server/src/main/java/azkaban/webapp/AzkabanWebServer.java b/azkaban-web-server/src/main/java/azkaban/webapp/AzkabanWebServer.java
index 2ec0fdc..23cfd00 100644
--- a/azkaban-web-server/src/main/java/azkaban/webapp/AzkabanWebServer.java
+++ b/azkaban-web-server/src/main/java/azkaban/webapp/AzkabanWebServer.java
@@ -20,6 +20,7 @@ import azkaban.AzkabanCommonModule;
import com.codahale.metrics.MetricRegistry;
import com.google.inject.Guice;
+import com.google.inject.Inject;
import java.io.File;
import java.io.IOException;
import java.io.InputStream;
@@ -49,10 +50,7 @@ import org.apache.velocity.runtime.log.Log4JLogChute;
import org.apache.velocity.runtime.resource.loader.ClasspathResourceLoader;
import org.apache.velocity.runtime.resource.loader.JarResourceLoader;
import org.joda.time.DateTimeZone;
-import org.mortbay.jetty.Connector;
import org.mortbay.jetty.Server;
-import org.mortbay.jetty.bio.SocketConnector;
-import org.mortbay.jetty.security.SslSocketConnector;
import org.mortbay.jetty.servlet.Context;
import org.mortbay.jetty.servlet.DefaultServlet;
import org.mortbay.jetty.servlet.ServletHolder;
@@ -141,11 +139,9 @@ public class AzkabanWebServer extends AzkabanServer {
public static final String DEFAULT_CONF_PATH = "conf";
private static final int MAX_FORM_CONTENT_SIZE = 10 * 1024 * 1024;
- private static final int MAX_HEADER_BUFFER_SIZE = 10 * 1024 * 1024;
private static AzkabanWebServer app;
private static final String DEFAULT_TIMEZONE_ID = "default.timezone.id";
- private static final int DEFAULT_THREAD_NUMBER = 20;
private static final String VELOCITY_DEV_MODE_PARAM = "velocity.dev.mode";
private static final String USER_MANAGER_CLASS_PARAM = "user.manager.class";
private static final String DEFAULT_STATIC_DIR = "";
@@ -185,17 +181,11 @@ public class AzkabanWebServer extends AzkabanServer {
this(null, loadConfigurationFromAzkabanHome());
}
- /**
- * Constructor
- */
+ @Inject
public AzkabanWebServer(Server server, Props props) throws Exception {
this.props = requireNonNull(props);
this.server = server;
- /* Initialize Guice Injector */
- // TODO move this to a common static context.
- SERVICE_PROVIDER.setInjector(Guice.createInjector(new AzkabanCommonModule(props)));
-
velocityEngine = configureVelocityEngine(props.getBoolean(VELOCITY_DEV_MODE_PARAM, false));
sessionCache = new SessionCache(props);
userManager = loadUserManager(props);
@@ -681,18 +671,28 @@ public class AzkabanWebServer extends AzkabanServer {
StdOutErrRedirect.redirectOutAndErrToLog();
logger.info("Starting Jetty Azkaban Web Server...");
- Props azkabanSettings = AzkabanServer.loadProps(args);
+ Props props = AzkabanServer.loadProps(args);
- if (azkabanSettings == null) {
+ if (props == null) {
logger.error("Azkaban Properties not loaded. Exiting..");
System.exit(1);
}
- final Server server = createServer(azkabanSettings);
+ /* Initialize Guice Injector */
+ SERVICE_PROVIDER.setInjector(Guice.createInjector(
+ new AzkabanCommonModule(props),
+ new AzkabanWebServerModule()
+ ));
- app = new AzkabanWebServer(server, azkabanSettings);
+ launch(props);
+ }
+
+ public static void launch(Props azkabanSettings) throws Exception {
+ /* This creates the Web Server instance */
+ app = SERVICE_PROVIDER.getInstance(AzkabanWebServer.class);
- prepareAndStartServer(azkabanSettings, server);
+ // TODO refactor code into ServerProvider
+ prepareAndStartServer(azkabanSettings, app.server);
Runtime.getRuntime().addShutdownHook(new Thread() {
@@ -706,8 +706,6 @@ public class AzkabanWebServer extends AzkabanServer {
logger.info("Shutting down http server...");
try {
app.close();
- server.stop();
- server.destroy();
} catch (Exception e) {
logger.error("Error while shutting down http server.", e);
}
@@ -738,60 +736,6 @@ public class AzkabanWebServer extends AzkabanServer {
});
}
- private static Server createServer(Props azkabanSettings) {
- final int maxThreads = azkabanSettings.getInt("jetty.maxThreads", DEFAULT_THREAD_NUMBER);
- boolean isStatsOn = azkabanSettings.getBoolean("jetty.connector.stats", true);
- logger.info("Setting up connector with stats on: " + isStatsOn);
-
- boolean ssl;
- int port;
- final Server server = new Server();
- if (azkabanSettings.getBoolean("jetty.use.ssl", true)) {
- int sslPortNumber =
- azkabanSettings.getInt("jetty.ssl.port", Constants.DEFAULT_SSL_PORT_NUMBER);
- port = sslPortNumber;
- ssl = true;
- logger.info("Setting up Jetty Https Server with port:" + sslPortNumber
- + " and numThreads:" + maxThreads);
-
- SslSocketConnector secureConnector = new SslSocketConnector();
- secureConnector.setPort(sslPortNumber);
- secureConnector.setKeystore(azkabanSettings.getString("jetty.keystore"));
- secureConnector.setPassword(azkabanSettings.getString("jetty.password"));
- secureConnector.setKeyPassword(azkabanSettings
- .getString("jetty.keypassword"));
- secureConnector.setTruststore(azkabanSettings
- .getString("jetty.truststore"));
- secureConnector.setTrustPassword(azkabanSettings
- .getString("jetty.trustpassword"));
- secureConnector.setHeaderBufferSize(MAX_HEADER_BUFFER_SIZE);
-
- // set up vulnerable cipher suites to exclude
- List<String> cipherSuitesToExclude = azkabanSettings.getStringList("jetty.excludeCipherSuites");
- logger.info("Excluded Cipher Suites: " + String.valueOf(cipherSuitesToExclude));
- if (cipherSuitesToExclude != null && !cipherSuitesToExclude.isEmpty()) {
- secureConnector.setExcludeCipherSuites(cipherSuitesToExclude.toArray(new String[0]));
- }
-
- server.addConnector(secureConnector);
- } else {
- ssl = false;
- port = azkabanSettings.getInt("jetty.port", Constants.DEFAULT_PORT_NUMBER);
- SocketConnector connector = new SocketConnector();
- connector.setPort(port);
- connector.setHeaderBufferSize(MAX_HEADER_BUFFER_SIZE);
- server.addConnector(connector);
- }
-
- // setting stats configuration for connectors
- for (Connector connector : server.getConnectors()) {
- connector.setStatsOn(isStatsOn);
- }
-
- logger.info(String.format("Starting %sserver on port: %d", ssl ? "SSL " : "", port));
- return server;
- }
-
private static void prepareAndStartServer(Props azkabanSettings, Server server) throws Exception {
validateDatabaseVersion(azkabanSettings);
configureRoutes(server, azkabanSettings);
@@ -824,7 +768,7 @@ public class AzkabanWebServer extends AzkabanServer {
}
private static void configureRoutes(Server server, Props azkabanSettings) throws TriggerManagerException {
- final int maxThreads = azkabanSettings.getInt("jetty.maxThreads", DEFAULT_THREAD_NUMBER);
+ final int maxThreads = azkabanSettings.getInt("jetty.maxThreads", Constants.DEFAULT_JETTY_MAX_THREAD_COUNT);
QueuedThreadPool httpThreadPool = new QueuedThreadPool(maxThreads);
app.setThreadPool(httpThreadPool);
@@ -1261,6 +1205,13 @@ public class AzkabanWebServer extends AzkabanServer {
}
scheduleManager.shutdown();
executorManager.shutdown();
+ try {
+ server.stop();
+ } catch (Throwable t) {
+ // Catch all while closing server
+ logger.error(t);
+ }
+ server.destroy();
}
private void registerMbean(String name, Object mbean) {
diff --git a/azkaban-web-server/src/main/java/azkaban/webapp/AzkabanWebServerModule.java b/azkaban-web-server/src/main/java/azkaban/webapp/AzkabanWebServerModule.java
new file mode 100644
index 0000000..7a244a9
--- /dev/null
+++ b/azkaban-web-server/src/main/java/azkaban/webapp/AzkabanWebServerModule.java
@@ -0,0 +1,35 @@
+/*
+ * Copyright 2017 LinkedIn Corp.
+ *
+ * Licensed 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.
+ *
+ */
+
+package azkaban.webapp;
+
+import com.google.inject.AbstractModule;
+import com.google.inject.Scopes;
+import org.mortbay.jetty.Server;
+
+/**
+ * This Guice module is currently a one place container for all bindings in the current module. This is intended to
+ * help during the migration process to Guice. Once this class starts growing we can move towards more modular
+ * structuring of Guice components.
+ */
+public class AzkabanWebServerModule extends AbstractModule {
+ @Override
+ protected void configure() {
+ bind(Server.class).toProvider(WebServerProvider.class);
+ bind(AzkabanWebServer.class).in(Scopes.SINGLETON);
+ }
+}
diff --git a/azkaban-web-server/src/main/java/azkaban/webapp/WebServerProvider.java b/azkaban-web-server/src/main/java/azkaban/webapp/WebServerProvider.java
new file mode 100644
index 0000000..1ffb6a5
--- /dev/null
+++ b/azkaban-web-server/src/main/java/azkaban/webapp/WebServerProvider.java
@@ -0,0 +1,93 @@
+/*
+ * Copyright 2017 LinkedIn Corp.
+ *
+ * Licensed 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.
+ *
+ */
+
+package azkaban.webapp;
+
+import azkaban.Constants;
+import azkaban.utils.Props;
+import com.google.inject.Inject;
+import com.google.inject.Provider;
+import java.util.List;
+import org.apache.log4j.Logger;
+import org.mortbay.jetty.Connector;
+import org.mortbay.jetty.Server;
+import org.mortbay.jetty.bio.SocketConnector;
+import org.mortbay.jetty.security.SslSocketConnector;
+
+import static java.util.Objects.*;
+
+
+public class WebServerProvider implements Provider<Server> {
+ private static final Logger logger = Logger.getLogger(WebServerProvider.class);
+ private static final int MAX_HEADER_BUFFER_SIZE = 10 * 1024 * 1024;
+
+ @Inject
+ private Props props;
+
+ @Override
+ public Server get() {
+ requireNonNull(props);
+
+ final int maxThreads = props.getInt("jetty.maxThreads", Constants.DEFAULT_JETTY_MAX_THREAD_COUNT);
+ boolean isStatsOn = props.getBoolean("jetty.connector.stats", true);
+ logger.info("Setting up connector with stats on: " + isStatsOn);
+
+ boolean ssl;
+ int port;
+ final Server server = new Server();
+ if (props.getBoolean("jetty.use.ssl", true)) {
+ int sslPortNumber = props.getInt("jetty.ssl.port", Constants.DEFAULT_SSL_PORT_NUMBER);
+ port = sslPortNumber;
+ ssl = true;
+ logger.info("Setting up Jetty Https Server with port:" + sslPortNumber
+ + " and numThreads:" + maxThreads);
+
+ SslSocketConnector secureConnector = new SslSocketConnector();
+ secureConnector.setPort(sslPortNumber);
+ secureConnector.setKeystore(props.getString("jetty.keystore"));
+ secureConnector.setPassword(props.getString("jetty.password"));
+ secureConnector.setKeyPassword(props.getString("jetty.keypassword"));
+ secureConnector.setTruststore(props.getString("jetty.truststore"));
+ secureConnector.setTrustPassword(props.getString("jetty.trustpassword"));
+ secureConnector.setHeaderBufferSize(MAX_HEADER_BUFFER_SIZE);
+
+ // set up vulnerable cipher suites to exclude
+ List<String> cipherSuitesToExclude = props.getStringList("jetty.excludeCipherSuites");
+ logger.info("Excluded Cipher Suites: " + String.valueOf(cipherSuitesToExclude));
+ if (cipherSuitesToExclude != null && !cipherSuitesToExclude.isEmpty()) {
+ secureConnector.setExcludeCipherSuites(cipherSuitesToExclude.toArray(new String[0]));
+ }
+
+ server.addConnector(secureConnector);
+ } else {
+ ssl = false;
+ port = props.getInt("jetty.port", Constants.DEFAULT_PORT_NUMBER);
+ SocketConnector connector = new SocketConnector();
+ connector.setPort(port);
+ connector.setHeaderBufferSize(MAX_HEADER_BUFFER_SIZE);
+ server.addConnector(connector);
+ }
+
+ // setting stats configuration for connectors
+ for (Connector connector : server.getConnectors()) {
+ connector.setStatsOn(isStatsOn);
+ }
+
+ logger.info(String.format("Starting %sserver on port: %d", ssl ? "SSL " : "", port));
+ return server;
+ }
+}
