rsc/src/main/java/org/apache/livy/rsc/RSCConf.java - incubator-livy - Git at Google

 /*
  * Licensed to the Apache Software Foundation (ASF) under one or more
  * contributor license agreements.  See the NOTICE file distributed with
  * this work for additional information regarding copyright ownership.
  * The ASF licenses this file to You 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 org.apache.livy.rsc;

 import java.io.IOException;
 import java.net.Inet4Address;
 import java.net.InetAddress;
 import java.net.NetworkInterface;
 import java.util.Collections;
 import java.util.Enumeration;
 import java.util.HashMap;
 import java.util.Map;
 import java.util.Properties;
 import javax.security.sasl.Sasl;

 import org.apache.livy.client.common.ClientConf;

 public class RSCConf extends ClientConf<RSCConf> {

   public static final String SPARK_CONF_PREFIX = "spark.";
   public static final String LIVY_SPARK_PREFIX = SPARK_CONF_PREFIX + "__livy__.";

   public static final String RSC_CONF_PREFIX = "livy.rsc.";

   public static enum Entry implements ConfEntry {
     CLIENT_ID("client.auth.id", null),
     CLIENT_SECRET("client.auth.secret", null),
     CLIENT_IN_PROCESS("client.do-not-use.run-driver-in-process", false),
     CLIENT_SHUTDOWN_TIMEOUT("client.shutdown-timeout", "10s"),
     DRIVER_CLASS("driver-class", null),
     SESSION_KIND("session.kind", null),

     LIVY_JARS("jars", null),
     SPARKR_PACKAGE("sparkr.package", null),
     PYSPARK_ARCHIVES("pyspark.archives", null),

     // Address for the RSC driver to connect back with it's connection info.
     LAUNCHER_ADDRESS("launcher.address", null),
     LAUNCHER_PORT_RANGE("launcher.port.range", "10000~10010"),
     // Setting up of this property by user has no benefit. It is currently being used
     // to pass  port information from ContextLauncher to RSCDriver
     LAUNCHER_PORT("launcher.port", -1),
     // How long will the RSC wait for a connection for a Livy server before shutting itself down.
     SERVER_IDLE_TIMEOUT("server.idle-timeout", "10m"),

     PROXY_USER("proxy-user", null),

     RPC_SERVER_ADDRESS("rpc.server.address", null),
     RPC_CLIENT_HANDSHAKE_TIMEOUT("server.connect.timeout", "90s"),
     RPC_CLIENT_CONNECT_TIMEOUT("client.connect.timeout", "10s"),
     RPC_CHANNEL_LOG_LEVEL("channel.log.level", null),
     RPC_MAX_MESSAGE_SIZE("rpc.max.size", 50 * 1024 * 1024),
     RPC_MAX_THREADS("rpc.threads", 8),
     RPC_SECRET_RANDOM_BITS("secret.bits", 256),

     SASL_MECHANISMS("rpc.sasl.mechanisms", "DIGEST-MD5"),
     SASL_QOP("rpc.sasl.qop", null),

     TEST_NO_CODE_COVERAGE_ANALYSIS("test.do-not-use.no-code-coverage-analysis", false),
     TEST_STUCK_END_SESSION("test.do-not-use.stuck-end-session", false),
     TEST_STUCK_START_DRIVER("test.do-not-use.stuck-start-driver", false),

     JOB_CANCEL_TRIGGER_INTERVAL("job-cancel.trigger-interval", "100ms"),
     JOB_CANCEL_TIMEOUT("job-cancel.timeout", "30s"),

     RETAINED_STATEMENTS("retained-statements", 100),
     RETAINED_SHARE_VARIABLES("retained.share-variables", 100),

     // Number of result rows to get for SQL Interpreters.
     SQL_NUM_ROWS("sql.num-rows", 1000);

     private final String key;
     private final Object dflt;

     private Entry(String key, Object dflt) {
       this.key = RSC_CONF_PREFIX + key;
       this.dflt = dflt;
     }

     @Override
     public String key() { return key; }

     @Override
     public Object dflt() { return dflt; }
   }

   public RSCConf() {
     this(new Properties());
   }

   public RSCConf(Properties config) {
     super(config);
   }

   public Map<String, String> getSaslOptions() {
     Map<String, String> opts = new HashMap<>();

     // TODO: add more options?
     String qop = get(Entry.SASL_QOP);
     if (qop != null) {
       opts.put(Sasl.QOP, qop);
     }

     return opts;
   }

   public String findLocalAddress() throws IOException {
     InetAddress address = InetAddress.getLocalHost();
     if (address.isLoopbackAddress()) {
       // Address resolves to something like 127.0.1.1, which happens on Debian;
       // try to find a better address using the local network interfaces
       Enumeration<NetworkInterface> ifaces = NetworkInterface.getNetworkInterfaces();
       while (ifaces.hasMoreElements()) {
         NetworkInterface ni = ifaces.nextElement();
         Enumeration<InetAddress> addrs = ni.getInetAddresses();
         while (addrs.hasMoreElements()) {
           InetAddress addr = addrs.nextElement();
           if (!addr.isLinkLocalAddress() && !addr.isLoopbackAddress()
               && addr instanceof Inet4Address) {
             // We've found an address that looks reasonable!
             LOG.warn("Your hostname, {}, resolves to a loopback address; using {} "
                 + " instead (on interface {})", address.getHostName(), addr.getHostAddress(),
                 ni.getName());
             LOG.warn("Set '{}' if you need to bind to another address.",
               Entry.RPC_SERVER_ADDRESS.key);
             return addr.getHostAddress();
           }
         }
       }
     }

     LOG.warn("Your hostname, {}, resolves to a loopback address, but we couldn't find "
         + "any external IP address!", address.getCanonicalHostName());
     LOG.warn("Set {} if you need to bind to another address.",
       Entry.RPC_SERVER_ADDRESS.key);
     return address.getCanonicalHostName();
   }

   private static final Map<String, DeprecatedConf> configsWithAlternatives
     = Collections.unmodifiableMap(new HashMap<String, DeprecatedConf>() {{
       put(RSCConf.Entry.CLIENT_IN_PROCESS.key, DepConf.CLIENT_IN_PROCESS);
       put(RSCConf.Entry.CLIENT_SHUTDOWN_TIMEOUT.key, DepConf.CLIENT_SHUTDOWN_TIMEOUT);
       put(RSCConf.Entry.DRIVER_CLASS.key, DepConf.DRIVER_CLASS);
       put(RSCConf.Entry.SERVER_IDLE_TIMEOUT.key, DepConf.SERVER_IDLE_TIMEOUT);
       put(RSCConf.Entry.PROXY_USER.key, DepConf.PROXY_USER);
       put(RSCConf.Entry.TEST_STUCK_END_SESSION.key, DepConf.TEST_STUCK_END_SESSION);
       put(RSCConf.Entry.TEST_STUCK_START_DRIVER.key, DepConf.TEST_STUCK_START_DRIVER);
       put(RSCConf.Entry.JOB_CANCEL_TRIGGER_INTERVAL.key, DepConf.JOB_CANCEL_TRIGGER_INTERVAL);
       put(RSCConf.Entry.JOB_CANCEL_TIMEOUT.key, DepConf.JOB_CANCEL_TIMEOUT);
       put(RSCConf.Entry.RETAINED_STATEMENTS.key, DepConf.RETAINED_STATEMENTS);
   }});

   // Maps deprecated key to DeprecatedConf with the same key.
   // There are no deprecated configs without alternatives currently.
   private static final Map<String, DeprecatedConf> deprecatedConfigs
     = Collections.unmodifiableMap(new HashMap<String, DeprecatedConf>());

   protected Map<String, DeprecatedConf> getConfigsWithAlternatives() {
     return configsWithAlternatives;
   }

   protected Map<String, DeprecatedConf> getDeprecatedConfigs() {
     return deprecatedConfigs;
   }

   static enum DepConf implements DeprecatedConf {
     CLIENT_IN_PROCESS("client.do_not_use.run_driver_in_process", "0.4"),
     CLIENT_SHUTDOWN_TIMEOUT("client.shutdown_timeout", "0.4"),
     DRIVER_CLASS("driver_class", "0.4"),
     SERVER_IDLE_TIMEOUT("server.idle_timeout", "0.4"),
     PROXY_USER("proxy_user", "0.4"),
     TEST_STUCK_END_SESSION("test.do_not_use.stuck_end_session", "0.4"),
     TEST_STUCK_START_DRIVER("test.do_not_use.stuck_start_driver", "0.4"),
     JOB_CANCEL_TRIGGER_INTERVAL("job_cancel.trigger_interval", "0.4"),
     JOB_CANCEL_TIMEOUT("job_cancel.timeout", "0.4"),
     RETAINED_STATEMENTS("retained_statements", "0.4");

     private final String key;
     private final String version;
     private final String deprecationMessage;

     private DepConf(String key, String version) {
       this(key, version, "");
     }

     private DepConf(String key, String version, String deprecationMessage) {
       this.key = RSC_CONF_PREFIX + key;
       this.version = version;
       this.deprecationMessage = deprecationMessage;
     }

     @Override
     public String key() { return key; }

     @Override
     public String version() { return version; }

     @Override
     public String deprecationMessage() { return deprecationMessage; }
   }

 }
	/*
	* Licensed to the Apache Software Foundation (ASF) under one or more
	* contributor license agreements. See the NOTICE file distributed with
	* this work for additional information regarding copyright ownership.
	* The ASF licenses this file to You 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 org.apache.livy.rsc;

	import java.io.IOException;
	import java.net.Inet4Address;
	import java.net.InetAddress;
	import java.net.NetworkInterface;
	import java.util.Collections;
	import java.util.Enumeration;
	import java.util.HashMap;
	import java.util.Map;
	import java.util.Properties;
	import javax.security.sasl.Sasl;

	import org.apache.livy.client.common.ClientConf;

	public class RSCConf extends ClientConf<RSCConf> {

	public static final String SPARK_CONF_PREFIX = "spark.";
	public static final String LIVY_SPARK_PREFIX = SPARK_CONF_PREFIX + "__livy__.";

	public static final String RSC_CONF_PREFIX = "livy.rsc.";

	public static enum Entry implements ConfEntry {
	CLIENT_ID("client.auth.id", null),
	CLIENT_SECRET("client.auth.secret", null),
	CLIENT_IN_PROCESS("client.do-not-use.run-driver-in-process", false),
	CLIENT_SHUTDOWN_TIMEOUT("client.shutdown-timeout", "10s"),
	DRIVER_CLASS("driver-class", null),
	SESSION_KIND("session.kind", null),

	LIVY_JARS("jars", null),
	SPARKR_PACKAGE("sparkr.package", null),
	PYSPARK_ARCHIVES("pyspark.archives", null),

	// Address for the RSC driver to connect back with it's connection info.
	LAUNCHER_ADDRESS("launcher.address", null),
	LAUNCHER_PORT_RANGE("launcher.port.range", "10000~10010"),
	// Setting up of this property by user has no benefit. It is currently being used
	// to pass port information from ContextLauncher to RSCDriver
	LAUNCHER_PORT("launcher.port", -1),
	// How long will the RSC wait for a connection for a Livy server before shutting itself down.
	SERVER_IDLE_TIMEOUT("server.idle-timeout", "10m"),

	PROXY_USER("proxy-user", null),

	RPC_SERVER_ADDRESS("rpc.server.address", null),
	RPC_CLIENT_HANDSHAKE_TIMEOUT("server.connect.timeout", "90s"),
	RPC_CLIENT_CONNECT_TIMEOUT("client.connect.timeout", "10s"),
	RPC_CHANNEL_LOG_LEVEL("channel.log.level", null),
	RPC_MAX_MESSAGE_SIZE("rpc.max.size", 50 * 1024 * 1024),
	RPC_MAX_THREADS("rpc.threads", 8),
	RPC_SECRET_RANDOM_BITS("secret.bits", 256),

	SASL_MECHANISMS("rpc.sasl.mechanisms", "DIGEST-MD5"),
	SASL_QOP("rpc.sasl.qop", null),

	TEST_NO_CODE_COVERAGE_ANALYSIS("test.do-not-use.no-code-coverage-analysis", false),
	TEST_STUCK_END_SESSION("test.do-not-use.stuck-end-session", false),
	TEST_STUCK_START_DRIVER("test.do-not-use.stuck-start-driver", false),

	JOB_CANCEL_TRIGGER_INTERVAL("job-cancel.trigger-interval", "100ms"),
	JOB_CANCEL_TIMEOUT("job-cancel.timeout", "30s"),

	RETAINED_STATEMENTS("retained-statements", 100),
	RETAINED_SHARE_VARIABLES("retained.share-variables", 100),

	// Number of result rows to get for SQL Interpreters.
	SQL_NUM_ROWS("sql.num-rows", 1000);

	private final String key;
	private final Object dflt;

	private Entry(String key, Object dflt) {
	this.key = RSC_CONF_PREFIX + key;
	this.dflt = dflt;
	}

	@Override
	public String key() { return key; }

	@Override
	public Object dflt() { return dflt; }
	}

	public RSCConf() {
	this(new Properties());
	}

	public RSCConf(Properties config) {
	super(config);
	}

	public Map<String, String> getSaslOptions() {
	Map<String, String> opts = new HashMap<>();

	// TODO: add more options?
	String qop = get(Entry.SASL_QOP);
	if (qop != null) {
	opts.put(Sasl.QOP, qop);
	}

	return opts;
	}

	public String findLocalAddress() throws IOException {
	InetAddress address = InetAddress.getLocalHost();
	if (address.isLoopbackAddress()) {
	// Address resolves to something like 127.0.1.1, which happens on Debian;
	// try to find a better address using the local network interfaces
	Enumeration<NetworkInterface> ifaces = NetworkInterface.getNetworkInterfaces();
	while (ifaces.hasMoreElements()) {
	NetworkInterface ni = ifaces.nextElement();
	Enumeration<InetAddress> addrs = ni.getInetAddresses();
	while (addrs.hasMoreElements()) {
	InetAddress addr = addrs.nextElement();
	if (!addr.isLinkLocalAddress() && !addr.isLoopbackAddress()
	&& addr instanceof Inet4Address) {
	// We've found an address that looks reasonable!
	LOG.warn("Your hostname, {}, resolves to a loopback address; using {} "
	+ " instead (on interface {})", address.getHostName(), addr.getHostAddress(),
	ni.getName());
	LOG.warn("Set '{}' if you need to bind to another address.",
	Entry.RPC_SERVER_ADDRESS.key);
	return addr.getHostAddress();
	}
	}
	}
	}

	LOG.warn("Your hostname, {}, resolves to a loopback address, but we couldn't find "
	+ "any external IP address!", address.getCanonicalHostName());
	LOG.warn("Set {} if you need to bind to another address.",
	Entry.RPC_SERVER_ADDRESS.key);
	return address.getCanonicalHostName();
	}

	private static final Map<String, DeprecatedConf> configsWithAlternatives
	= Collections.unmodifiableMap(new HashMap<String, DeprecatedConf>() {{
	put(RSCConf.Entry.CLIENT_IN_PROCESS.key, DepConf.CLIENT_IN_PROCESS);
	put(RSCConf.Entry.CLIENT_SHUTDOWN_TIMEOUT.key, DepConf.CLIENT_SHUTDOWN_TIMEOUT);
	put(RSCConf.Entry.DRIVER_CLASS.key, DepConf.DRIVER_CLASS);
	put(RSCConf.Entry.SERVER_IDLE_TIMEOUT.key, DepConf.SERVER_IDLE_TIMEOUT);
	put(RSCConf.Entry.PROXY_USER.key, DepConf.PROXY_USER);
	put(RSCConf.Entry.TEST_STUCK_END_SESSION.key, DepConf.TEST_STUCK_END_SESSION);
	put(RSCConf.Entry.TEST_STUCK_START_DRIVER.key, DepConf.TEST_STUCK_START_DRIVER);
	put(RSCConf.Entry.JOB_CANCEL_TRIGGER_INTERVAL.key, DepConf.JOB_CANCEL_TRIGGER_INTERVAL);
	put(RSCConf.Entry.JOB_CANCEL_TIMEOUT.key, DepConf.JOB_CANCEL_TIMEOUT);
	put(RSCConf.Entry.RETAINED_STATEMENTS.key, DepConf.RETAINED_STATEMENTS);
	}});

	// Maps deprecated key to DeprecatedConf with the same key.
	// There are no deprecated configs without alternatives currently.
	private static final Map<String, DeprecatedConf> deprecatedConfigs
	= Collections.unmodifiableMap(new HashMap<String, DeprecatedConf>());

	protected Map<String, DeprecatedConf> getConfigsWithAlternatives() {
	return configsWithAlternatives;
	}

	protected Map<String, DeprecatedConf> getDeprecatedConfigs() {
	return deprecatedConfigs;
	}

	static enum DepConf implements DeprecatedConf {
	CLIENT_IN_PROCESS("client.do_not_use.run_driver_in_process", "0.4"),
	CLIENT_SHUTDOWN_TIMEOUT("client.shutdown_timeout", "0.4"),
	DRIVER_CLASS("driver_class", "0.4"),
	SERVER_IDLE_TIMEOUT("server.idle_timeout", "0.4"),
	PROXY_USER("proxy_user", "0.4"),
	TEST_STUCK_END_SESSION("test.do_not_use.stuck_end_session", "0.4"),
	TEST_STUCK_START_DRIVER("test.do_not_use.stuck_start_driver", "0.4"),
	JOB_CANCEL_TRIGGER_INTERVAL("job_cancel.trigger_interval", "0.4"),
	JOB_CANCEL_TIMEOUT("job_cancel.timeout", "0.4"),
	RETAINED_STATEMENTS("retained_statements", "0.4");

	private final String key;
	private final String version;
	private final String deprecationMessage;

	private DepConf(String key, String version) {
	this(key, version, "");
	}

	private DepConf(String key, String version, String deprecationMessage) {
	this.key = RSC_CONF_PREFIX + key;
	this.version = version;
	this.deprecationMessage = deprecationMessage;
	}

	@Override
	public String key() { return key; }

	@Override
	public String version() { return version; }

	@Override
	public String deprecationMessage() { return deprecationMessage; }
	}

	}