public final class EnvironmentConstants {
	public final static Log LOG = LogFactory.getLog(EnvironmentConstants.class);

	public final static SimpleDateFormat DATE_FORMAT = new SimpleDateFormat(
			"yyyy-MM-dd HH:mm:ss");
	public static String DATA_FILE_EXTENSION = ".gz";
	public static String WORKING_DIRECTORY_ROOT = "/tmp";
	public static int DEFAULT_RESULT_DATA_NUMBER = 500;
	public static int MAX_RESULT_DATA_NUMBER = 1000000;
	// result data will be stored in this location
	public static String HDFS_DATA_ROOT_PATH = "hdfs://10.1.77.86/data/polestar";

	// hadoop security
	public static String HADOOP_PRINCIPAL = "hadoop@DIANPING.COM";
	public static String HADOOP_KEYTAB_FILE = "/home/hadoop/.keytab";

	public static String LOCAL_HOST_ADDRESS;

	static {
		try {
			LOCAL_HOST_ADDRESS = InetAddress.getLocalHost().getHostAddress();
			LOG.info("local host address: " + LOCAL_HOST_ADDRESS);
		} catch (UnknownHostException uhe) {
			LOG.error("get local host address failed, " + uhe);
		}

		InputStream is = ClassUtils.getResourceAsStream("polestar.properties");
		Properties pros = new Properties();
		try {
			pros.load(is);

			DATA_FILE_EXTENSION = pros.getProperty("DATA_FILE_EXTENSION",
					DATA_FILE_EXTENSION);

			WORKING_DIRECTORY_ROOT = pros.getProperty("WORKING_DIRECTORY_ROOT",
					WORKING_DIRECTORY_ROOT);

			DEFAULT_RESULT_DATA_NUMBER = Integer.valueOf(pros.getProperty(
					"DEFAULT_RESULT_DATA_NUMBER",
					String.valueOf(DEFAULT_RESULT_DATA_NUMBER)));

			MAX_RESULT_DATA_NUMBER = Integer.valueOf(pros.getProperty(
					"MAX_RESULT_DATA_NUMBER",
					String.valueOf(MAX_RESULT_DATA_NUMBER)));

			HDFS_DATA_ROOT_PATH = pros.getProperty("HDFS_DATA_ROOT_PATH",
					HDFS_DATA_ROOT_PATH);

			HADOOP_PRINCIPAL = pros.getProperty(
					pros.getProperty("HADOOP_PRINCIPAL"), HADOOP_PRINCIPAL);
		} catch (IOException e) {
			e.printStackTrace();
		}
	}
} public class PolestarException extends RuntimeException {
	private static final long serialVersionUID = 6525L;

	public PolestarException() {
		super();
	}

	public PolestarException(String message) {
		super(message);
	}

	public PolestarException(Throwable e) {
		super(e);
	}

	public PolestarException(String msg, Throwable e) {
		super(msg, e);
	}
} public class CommandQueryEngine implements IQueryEngine {
	public static final Log LOG = LogFactory.getLog(CommandQueryEngine.class);

	private static final MessageFormat SHELL_COMMAND_FORMAT = new MessageFormat(
			" sudo -u {0} -i \"source /etc/profile; {1} -e {2}\" ");
	private static final MessageFormat HIVE_COMMAND_FORMAT = new MessageFormat(
			" set hive.cli.print.header=true; use {0}; {1} ");
	private static final IQueryEngine INSTANCE = new CommandQueryEngine();

	public static IQueryEngine getInstance() {
		return INSTANCE;
	}

	@Override
	public QueryResult postQuery(Query query) {
		JobContext jobCtx = new JobContext();
		jobCtx.setId(query.getId());
		jobCtx.setUsername(query.getUsername());
		jobCtx.setPasswd(query.getPassword());
		jobCtx.setStoreResult(query.isStoreResult());
		jobCtx.setCommands(new String[] {
				"bash",
				"-c",
				buildExecCommand(query.getUsername(), query.getMode(),
						query.getDatabase(), query.getSql()) });
		jobCtx.setWorkDir(EnvironmentConstants.WORKING_DIRECTORY_ROOT
				+ File.separator + query.getId());

		Job job = JobAdapter.createJob(query.getMode(), jobCtx);
		JobManager.putJob(query.getId(), job, jobCtx);

		QueryResult queryRes = new QueryResult();
		queryRes.setId(query.getId());
		try {
			int exitCode = job.run();
			LOG.info("exitcode:" + exitCode + " ,id:" + jobCtx.getId());
			if (0 == exitCode) {
				queryRes.setSuccess(true);
				if (jobCtx.isStoreResult()) {
					String hdfsDataFileAbsolutePath = EnvironmentConstants.HDFS_DATA_ROOT_PATH
							+ File.separator
							+ Utilities.getLastPartFileName(jobCtx
									.getLocalDataPath());
					HDFSManager.putFileToHDFS(jobCtx.getLocalDataPath(),
							hdfsDataFileAbsolutePath);
					queryRes.setResultFilePath(hdfsDataFileAbsolutePath);
				}
				Utilities.fillInColumnsAndData(jobCtx.getStdout().toString(),
						queryRes);
			} else {
				queryRes.setErrorMsg(jobCtx.getStderr().toString());
			}
		} catch (Exception e) {
			LOG.error(e);
		} finally {
			FileUtils.deleteQuietly(new File(jobCtx.getWorkDir()));
			JobManager.removeJob(query.getId());
		}
		return queryRes;
	}

	private String getEngineCommandByMode(String mode) {
		if ("hive".equalsIgnoreCase(mode)) {
			return "hive";
		} else if ("shark".equalsIgnoreCase(mode)) {
			return "shark-witherror";
		}
		return mode;
	}

	private String buildExecCommand(String username, String mode,
			String database, String sql) {
		String hiveCommand = HIVE_COMMAND_FORMAT.format(new String[] {
				database, sql });
		String shellCommand = SHELL_COMMAND_FORMAT.format(new String[] {
				username, getEngineCommandByMode(mode), hiveCommand });
		if (LOG.isDebugEnabled()) {
			LOG.debug("execute command: " + shellCommand);
		}
		return shellCommand;
	}
} public interface IQueryEngine {
	
	QueryResult postQuery(Query query);
} public class Query extends Entity {
	private String sql;
	private String mode;
	private String database;
	private String username;
	private String password;
	private boolean storeResult;

	public String getSql() {
		return sql;
	}

	public void setSql(String sql) {
		this.sql = sql;
	}

	public String getMode() {
		return mode;
	}

	public void setMode(String mode) {
		this.mode = mode;
	}

	public String getDatabase() {
		return database;
	}

	public void setDatabase(String database) {
		this.database = database;
	}

	public String getUsername() {
		return username;
	}

	public void setUsername(String username) {
		this.username = username;
	}

	public String getPassword() {
		return password;
	}

	public void setPassword(String password) {
		this.password = password;
	}

	public boolean isStoreResult() {
		return storeResult;
	}

	public void setStoreResult(boolean storeResult) {
		this.storeResult = storeResult;
	}
} public final class QueryResult extends Entity {
	private String[] columnNames = new String[]{};
	private List<String[]> data = new ArrayList<String[]>();
	private long execTime = -1L;
	private String errorMsg = "";
	private String resultFilePath = "";
	private Boolean success = false;
	
	public String[] getColumnNames() {
		return columnNames;
	}
	public void setColumnNames(String[] columnNames) {
		this.columnNames = columnNames;
	}
	public List<String[]> getData() {
		return data;
	}
	public void setData(List<String[]> data) {
		this.data = data;
	}
	public long getExecTime() {
		return execTime;
	}
	public void setExecTime(long execTime) {
		this.execTime = execTime;
	}
	public String getErrorMsg() {
		return errorMsg;
	}
	public void setErrorMsg(String errorMsg) {
		this.errorMsg = errorMsg;
	}
	public String getResultFilePath() {
		return resultFilePath;
	}
	public void setResultFilePath(String resultFilePath) {
		this.resultFilePath = resultFilePath;
	}
	public Boolean getSuccess() {
		return success;
	}
	public void setSuccess(Boolean success) {
		this.success = success;
	}
	
	@Override
	public String toString() {
		return "QueryResult [columnNames=" + Arrays.toString(columnNames)
				+ ", data=" + data + ", execTime=" + execTime + ", errorMsg="
				+ errorMsg + ", resultFilePath=" + resultFilePath
				+ ", success=" + success + "]";
	}
} public class QueryStatus extends Entity {
	private String message;
	private Boolean success;
	
	public String getMessage() {
		return message;
	}
	public void setMessage(String message) {
		this.message = message;
	}
	public Boolean getSuccess() {
		return success;
	}
	public void setSuccess(Boolean success) {
		this.success = success;
	}
	
	@Override
	public String toString() {
		return "QueryStatus [message=" + message + ", success=" + success + "]";
	}

} public interface Job {

	Integer run() throws Exception;
	
	void cancel();

	boolean isCanceled();

} public final class JobContext {
	private String workDir;
	private String localDataPath;
	private Properties properties = new Properties();
	private String[] commands;
	private String username;
	private String passwd;
	private boolean storeResult;
	private String id;
	private int exitCode;
	private boolean isDone = false;
	private StringBuffer stderr = new StringBuffer(2000);
	private StringBuffer stdout = new StringBuffer(2000);

	public String getWorkDir() {
		return workDir;
	}

	public void setWorkDir(String workDir) {
		this.workDir = workDir;
	}

	public Properties getProperties() {
		return properties;
	}

	public void setProperties(Properties properties) {
		this.properties = properties;
	}

	public String[] getCommands() {
		return commands;
	}

	public void setCommands(String[] commands) {
		this.commands = commands;
	}

	public String getUsername() {
		return username;
	}

	public void setUsername(String username) {
		this.username = username;
	}

	public String getPasswd() {
		return passwd;
	}

	public void setPasswd(String passwd) {
		this.passwd = passwd;
	}

	public boolean isStoreResult() {
		return storeResult;
	}

	public void setStoreResult(boolean storeResult) {
		this.storeResult = storeResult;
	}

	public String getId() {
		return id;
	}

	public void setId(String id) {
		this.id = id;
	}

	public int getExitCode() {
		return exitCode;
	}

	public void setExitCode(int exitCode) {
		this.exitCode = exitCode;
	}

	public String getLocalDataPath() {
		return localDataPath;
	}

	public void setLocalDataPath(String localDataPath) {
		this.localDataPath = localDataPath;
	}

	public StringBuffer getStderr() {
		return stderr;
	}

	public void setStderr(StringBuffer stderr) {
		this.stderr = stderr;
	}

	public StringBuffer getStdout() {
		return stdout;
	}

	public void setStdout(StringBuffer stdout) {
		this.stdout = stdout;
	}

	public boolean isDone() {
		return isDone;
	}

	public void setDone() {
		isDone = true;
	}
} public class JobManager {
	private final static Map<String, Job> idToJob = new ConcurrentHashMap<String, Job>(
			100);
	private final static Map<String, JobContext> idToJobContext = new ConcurrentHashMap<String, JobContext>(
			100);

	public static Job getJobById(String id) {
		return idToJob.get(id);
	}

	public static JobContext getJobContextById(String id) {
		return idToJobContext.get(id);
	}

	public static void putJob(String id, Job job, JobContext jobctx) {
		synchronized (JobManager.class) {
			idToJob.put(id, job);
			idToJobContext.put(id, jobctx);
		}
	}

	public static void removeJob(String id) {
		synchronized (JobManager.class) {
			idToJob.remove(id);
			idToJobContext.remove(id);
		}
	}
} public class Utilities {
	private static final Log LOG = LogFactory.getLog(Utilities.class);

	public static final char LINE_SEPARATOR = '\n';
	public static final char FIELD_SEPARATOR = '\t';
	public static final int COMPRESSION_BUFFER_SIZE = 4096;
	public static final int COMPRESSION_LEVEL = Deflater.BEST_COMPRESSION;

	public static void hadoopKerberosLogin(String username, String passwd,
			String ticketCachePath) throws PolestarException {
		LOG.info("start to kinit, username:" + username
				+ ",ticketcache location:" + ticketCachePath);
		try {
			hadoopKerberosLogin.loginFromPassword(username, passwd,
					ticketCachePath);
		} catch (IOException e) {
			LOG.error("kinit failed " + e);
			throw new PolestarException("kinit login failed, username:"
					+ username + " passwd:" + passwd, e);
		}
	}

	public static OutputStream openOutputStream(File file, boolean append,
			boolean gzip) throws IOException {
		if (file.exists()) {
			if (file.isDirectory()) {
				throw new IOException("File '" + file
						+ "' exists but is a directory");
			}
			if (!file.canWrite()) {
				throw new IOException("File '" + file
						+ "' cannot be written to");
			}
		} else {
			File parent = file.getParentFile();
			if (parent != null) {
				if (!parent.mkdirs() && !parent.isDirectory()) {
					throw new IOException("Directory '" + parent
							+ "' could not be created");
				}
			}
		}
		if (gzip) {
			return new GZIPOutputStream(new FileOutputStream(file, append),
					COMPRESSION_BUFFER_SIZE) {
				{
					def.setLevel(COMPRESSION_LEVEL);
				};
			};
		}
		return new FileOutputStream(file, append);
	}

	public static void fillInColumnsAndData(String content, QueryResult jobctx) {
		if (!StringUtils.isEmpty(content)) {
			String[] lines = StringUtils.split(content, LINE_SEPARATOR);
			Preconditions.checkArgument(lines.length >= 1,
					"query standout lines should not be under 1, content:"
							+ content);
			String[] columnNames = StringUtils.splitPreserveAllTokens(lines[0],
					FIELD_SEPARATOR);
			jobctx.setColumnNames(columnNames);
			for (int i = 1; i < lines.length; i++) {
				String[] columns = StringUtils.splitPreserveAllTokens(lines[i],
						FIELD_SEPARATOR);
				if (columns.length == columnNames.length) {
					jobctx.getData().add(columns);
				} else {
					LOG.warn("unformatted column data:" + lines[i]
							+ ",discard it");
				}
			}
		}
	}

	public static String genUniqueDataFileName() {
		return genUniqueID() + EnvironmentConstants.DATA_FILE_EXTENSION;
	}

	public static String genUniqueID() {
		return StringUtils.replaceChars(UUID.randomUUID().toString(), "-", "");
	}

	public static String getLastPartFileName(String absolutePath) {
		File file = new File(absolutePath);
		return file.getName();
	}

	public static long getCurrentTime() {
		return System.currentTimeMillis();
	}
} public class BadParamException extends SimpleWebException {
	private static final long serialVersionUID = 6788167073159708825L;

	public BadParamException(String msg) {
		super(Status.BAD_REQUEST.getStatusCode(), msg);
	}
	
	public BadParamException(Status status, String msg) {
		super(status.getStatusCode(), msg);
	}
} public class HDFSManager {
	private static Log LOG = LogFactory.getLog(HDFSManager.class);

	private static FileSystem fs = null;

	private static FileSystem getFS() {
		if (fs == null) {
			try {
				fs = FileSystem.get(getDefaultConfiguration());
			} catch (IOException e) {
				LOG.error("filesystem is not initialized correctly!" + e);
				e.printStackTrace();
			}
		}
		return fs;
	}

	public static void putFileToHDFS(String src, String dest) {
		try {
			getFS().copyFromLocalFile(false, true, new Path(src),
					new Path(dest));
			LOG.info("put file " + src + " to " + dest + " succeed !");
		} catch (IOException e) {
			LOG.error("put file " + src + " to " + dest + " failed !", e);
		}
	}

	public static Configuration getDefaultConfiguration() {
		Configuration conf = null;
		File coreSite = new File(getHadoopConfDir(), "core-site.xml");
		File hdfsSite = new File(getHadoopConfDir(), "hdfs-site.xml");
		if (coreSite.exists() && hdfsSite.exists()) {
			conf = new Configuration(true);
			conf.addResource(new Path(coreSite.getAbsolutePath()));
			conf.addResource(new Path(hdfsSite.getAbsolutePath()));

			try {
				// 设置服务申请者的Principle
				conf.set("hadoop.principal",
						EnvironmentConstants.HADOOP_PRINCIPAL);
				// 设置keytab file的路径
				conf.set("hadoop.keytab.file",
						EnvironmentConstants.HADOOP_KEYTAB_FILE);
				// Kerberos Authentication
				UserGroupInformation.setConfiguration(conf);
				SecurityUtil.login(conf, "hadoop.keytab.file",
						"hadoop.principal");
			} catch (IOException e) {
				LOG.error(e);
				e.printStackTrace();
			}
		} else {
			LOG.error("core-site.xml or hdfs-site.xml is not found in the system environment!");
		}
		return conf;
	}

	public static String getHadoopConfDir() {
		String confDir = System.getenv("HADOOP_CONF_DIR");
		if (StringUtils.isBlank(confDir)) {
			confDir = getHadoopHome() + File.separator + "conf";
		}
		return confDir;
	}

	public static String getHadoopHome() {
		return System.getenv("HADOOP_HOME");
	}

	public static String getHiveHome() {
		return System.getenv("HIVE_HOME");
	}

	public static InputStream openFSDataInputStream(String absolutePath)
			throws BadParamException {
		Path p = new Path(absolutePath);
		try {
			if (!getFS().exists(p)) {
				throw new BadParamException(p.toString() + " doesn't exist !");
			}
			if (!getFS().isFile(p)) {
				throw new BadParamException(p.toString() + "  is not a file !");
			}
			return getFS().open(p);
		} catch (IOException e) {
			throw new BadParamException(e.toString());
		}
	}
} public interface QueryDAO {

	public void insertQueryInfo(QueryInfo queryInfo);

	public List<QueryInfo> findQueryInfoByUsername(String username);

	public void insertQueryProgress(QueryProgress queryProgress);

	public QueryProgress findQueryProgressById(String id);

	public void insertQueryCancel(QueryCancel qc);

	public List<QueryCancel> findQueryCancelWithouHost(String host);

	public void deleteQueryCancelById(String id);
} public final class QueryDAOFactory {

	public static QueryDAO queryDao;

	public static QueryDAO getInstance() {
		if (queryDao == null) {
			synchronized (QueryDAOFactory.class) {
				if (queryDao == null) {
					AbstractApplicationContext cxt = new ClassPathXmlApplicationContext(
							"applicationContext.xml");
					queryDao = (QueryDAO) cxt.getBean("queryDaoImpl");
				}
			}
		}
		return queryDao;
	}
} public class QueryCancel {
	private String id;
	private String host;

	public QueryCancel() {
	}

	public QueryCancel(String id, String host) {
		this.id = id;
		this.host = host;
	}

	public String getId() {
		return id;
	}

	public void setId(String id) {
		this.id = id;
	}

	public String getHost() {
		return host;
	}

	public void setHost(String host) {
		this.host = host;
	}
} public class QueryInfo {
	private String sql;
	private String mode;
	private String username;
	private String resultFilePath;
	private long execTime;
	private String addtime;

	public QueryInfo() {
	}

	public QueryInfo(Query q, QueryResult rs, long starttime) {
		setSql(q.getSql());
		setMode(q.getMode());
		setUsername(q.getUsername());
		setResultFilePath(rs.getResultFilePath());
		setExecTime(rs.getExecTime());
		setAddtime(EnvironmentConstants.DATE_FORMAT.format(new Date(starttime)));
	}

	public String getSql() {
		return sql;
	}

	public void setSql(String sql) {
		this.sql = sql;
	}

	public String getMode() {
		return mode;
	}

	public void setMode(String mode) {
		this.mode = mode;
	}

	public String getUsername() {
		return username;
	}

	public void setUsername(String username) {
		this.username = username;
	}

	public String getResultFilePath() {
		return resultFilePath;
	}

	public void setResultFilePath(String resultFilePath) {
		this.resultFilePath = resultFilePath;
	}

	public long getExecTime() {
		return execTime;
	}

	public void setExecTime(long execTime) {
		this.execTime = execTime;
	}

	public String getAddtime() {
		return addtime;
	}

	public void setAddtime(String addtime) {
		this.addtime = addtime;
	}

	@Override
	public String toString() {
		return "QueryInfo [sql=" + sql + ", mode=" + mode + ", username="
				+ username + ", resultFilePath=" + resultFilePath
				+ ", execTime=" + execTime + ", addtime=" + addtime + "]";
	}
} public class QueryProgress {
	private String id;
	private String progressInfo;

	public QueryProgress() {
	}

	public QueryProgress(String id, String progressInfo) {
		this.id = id;
		this.progressInfo = progressInfo;
	}

	public String getId() {
		return id;
	}

	public void setId(String id) {
		this.id = id;
	}

	public String getProgressInfo() {
		return progressInfo;
	}

	public void setProgressInfo(String progressInfo) {
		this.progressInfo = progressInfo;
	}

	@Override
	public String toString() {
		return "QueryProgress [id=" + id + ", progressInfo=" + progressInfo
				+ "]";
	}
}
import java.io.File;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import javax.ws.rs.WebApplicationException;
import javax.ws.rs.core.StreamingOutput;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import com.dianping.polestar.EnvironmentConstants;
import com.dianping.polestar.PolestarException;
import com.dianping.polestar.engine.CommandQueryEngine;
import com.dianping.polestar.engine.IQueryEngine;
import com.dianping.polestar.entity.Query;
import com.dianping.polestar.entity.QueryResult;
import com.dianping.polestar.entity.QueryStatus;
import com.dianping.polestar.jobs.Job;
import com.dianping.polestar.jobs.JobContext;
import com.dianping.polestar.jobs.JobManager;
import com.dianping.polestar.jobs.Utilities;
import com.dianping.polestar.rest.BadParamException;
import com.dianping.polestar.store.HDFSManager;
import com.dianping.polestar.store.mysql.dao.QueryDAO;
import com.dianping.polestar.store.mysql.dao.impl.QueryDAOFactory;
import com.dianping.polestar.store.mysql.domain.QueryCancel;
import com.dianping.polestar.store.mysql.domain.QueryInfo;
import com.dianping.polestar.store.mysql.domain.QueryProgress;
package com.dianping.polestar.service;





public class DefaultQueryService implements IQueryService {
	private static final Log LOG = LogFactory.getLog(DefaultQueryService.class);
	private static final int EOF = -1;

	private static final DefaultQueryService INSTANCE = new DefaultQueryService();

	private IQueryEngine cmdEngine = CommandQueryEngine.getInstance();
	private QueryDAO queryDao = QueryDAOFactory.getInstance();

	public static IQueryService getInstance() {
		return INSTANCE;
	}

	@Override
	public QueryResult postQuery(Query query) throws BadParamException {