/**
* 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.hive.service.cli.operation;
import com.google.common.base.Charsets;
import org.apache.commons.io.FileUtils;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.io.IOUtils;
import org.apache.hive.service.cli.FetchOrientation;
import org.apache.hive.service.cli.HiveSQLException;
import org.apache.hive.service.cli.OperationHandle;
import java.io.*;
import java.util.ArrayList;
import java.util.List;
/**
* OperationLog wraps the actual operation log file, and provides interface
* for accessing, reading, writing, and removing the file.
*/
public class OperationLog {
private static final Log LOG = LogFactory.getLog(OperationLog.class.getName());
private final String operationName;
private final LogFile logFile;
public OperationLog(String name, File file) throws FileNotFoundException{
operationName = name;
logFile = new LogFile(file);
}
/**
* Singleton OperationLog object per thread.
*/
private static final ThreadLocal<OperationLog> THREAD_LOCAL_OPERATION_LOG = new
ThreadLocal<OperationLog>() {
@Override
protected synchronized OperationLog initialValue() {
return null;
}
};
public static void setCurrentOperationLog(OperationLog operationLog) {
THREAD_LOCAL_OPERATION_LOG.set(operationLog);
}
public static OperationLog getCurrentOperationLog() {
return THREAD_LOCAL_OPERATION_LOG.get();
}
public static void removeCurrentOperationLog() {
THREAD_LOCAL_OPERATION_LOG.remove();
}
/**
* Write operation execution logs into log file
* @param operationLogMessage one line of log emitted from log4j
*/
public void writeOperationLog(String operationLogMessage) {
logFile.write(operationLogMessage);
}
/**
* Read operation execution logs from log file
* @param fetchOrientation one of Enum FetchOrientation values
* @param maxRows the max number of fetched lines from log
* @return
* @throws HiveSQLException
*/
public List<String> readOperationLog(FetchOrientation fetchOrientation, long maxRows)
throws HiveSQLException{
return logFile.read(fetchOrientation, maxRows);
}
/**
* Close this OperationLog when operation is closed. The log file will be removed.
*/
public void close() {
logFile.remove();
}
/**
* Wrapper for read/write the operation log file
*/
private class LogFile {
private File file;
private BufferedReader in;
private PrintStream out;
private volatile boolean isRemoved;
LogFile(File file) throws FileNotFoundException {
this.file = file;
in = new BufferedReader(new InputStreamReader(new FileInputStream(file)));
out = new PrintStream(new FileOutputStream(file));
isRemoved = false;
}
synchronized void write(String msg) {
// write log to the file
out.print(msg);
}
synchronized List<String> read(FetchOrientation fetchOrientation, long maxRows)
throws HiveSQLException{
// reset the BufferReader, if fetching from the beginning of the file
if (fetchOrientation.equals(FetchOrientation.FETCH_FIRST)) {
resetIn();
}
return readResults(maxRows);
}
void remove() {
try {
if (in != null) {
in.close();
}
if (out != null) {
out.close();
}
FileUtils.forceDelete(file);
isRemoved = true;
} catch (Exception e) {
LOG.error("Failed to remove corresponding log file of operation: " + operationName, e);
}
}
private void resetIn() {
if (in != null) {
IOUtils.cleanup(LOG, in);
in = null;
}
}
private List<String> readResults(long nLines) throws HiveSQLException {
if (in == null) {
try {
in = new BufferedReader(new InputStreamReader(new FileInputStream(file)));
} catch (FileNotFoundException e) {
if (isRemoved) {
throw new HiveSQLException("The operation has been closed and its log file " +
file.getAbsolutePath() + " has been removed.", e);
} else {
throw new HiveSQLException("Operation Log file " + file.getAbsolutePath() +
" is not found.", e);
}
}
}
List<String> logs = new ArrayList<String>();
String line = "";
// if nLines <= 0, read all lines in log file.
for (int i = 0; i < nLines || nLines <= 0; i++) {
try {
line = in.readLine();
if (line == null) {
break;
} else {
logs.add(line);
}
} catch (IOException e) {
if (isRemoved) {
throw new HiveSQLException("The operation has been closed and its log file " +
file.getAbsolutePath() + " has been removed.", e);
} else {
throw new HiveSQLException("Reading operation log file encountered an exception: ", e);
}
}
}
return logs;
}
}
}