/*
* Copyright 2018 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.viewer.reportal;
import azkaban.executor.ExecutableFlow;
import azkaban.executor.ExecutableNode;
import azkaban.executor.ExecutionOptions;
import azkaban.executor.ExecutorManagerAdapter;
import azkaban.executor.ExecutorManagerException;
import azkaban.flow.Flow;
import azkaban.project.Project;
import azkaban.project.ProjectManager;
import azkaban.project.ProjectManagerException;
import azkaban.reportal.util.IStreamProvider;
import azkaban.reportal.util.Reportal;
import azkaban.reportal.util.Reportal.Query;
import azkaban.reportal.util.Reportal.Variable;
import azkaban.reportal.util.ReportalHelper;
import azkaban.reportal.util.ReportalUtil;
import azkaban.reportal.util.StreamProviderHDFS;
import azkaban.scheduler.ScheduleManager;
import azkaban.scheduler.ScheduleManagerException;
import azkaban.security.commons.HadoopSecurityManager;
import azkaban.server.session.Session;
import azkaban.user.Permission;
import azkaban.user.Permission.Type;
import azkaban.user.User;
import azkaban.user.UserManager;
import azkaban.utils.FileIOUtils.LogData;
import azkaban.utils.Pair;
import azkaban.utils.Props;
import azkaban.webapp.AzkabanWebServer;
import azkaban.webapp.servlet.LoginAbstractAzkabanServlet;
import azkaban.webapp.servlet.Page;
import java.io.File;
import java.io.FileFilter;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Scanner;
import java.util.Set;
import javax.servlet.ServletConfig;
import javax.servlet.ServletException;
import javax.servlet.http.HttpServletRequest;
import javax.servlet.http.HttpServletResponse;
import org.apache.commons.io.FileUtils;
import org.apache.commons.io.IOUtils;
import org.apache.commons.lang.StringEscapeUtils;
import org.apache.commons.lang.StringUtils;
import org.apache.commons.lang.math.NumberUtils;
import org.apache.log4j.Logger;
import org.apache.velocity.tools.generic.EscapeTool;
import org.joda.time.DateTime;
public class ReportalServlet extends LoginAbstractAzkabanServlet {
private static final String REPORTAL_VARIABLE_PREFIX = "reportal.variable.";
private static final String HADOOP_SECURITY_MANAGER_CLASS_PARAM =
"hadoop.security.manager.class";
private static final long serialVersionUID = 1L;
private static final Logger logger = Logger.getLogger(ReportalServlet.class);
private final File reportalMailTempDirectory;
private final Props props;
private final String viewerName;
private final String reportalStorageUser;
private final File webResourcesFolder;
private final int max_allowed_schedule_dates;
private final int default_schedule_dates;
private final boolean showNav;
private CleanerThread cleanerThread;
/**
* A whitelist of allowed email domains (e.g.: example.com). If null, all
* email domains are allowed.
*/
private Set<String> allowedEmailDomains = null;
private AzkabanWebServer server;
private boolean shouldProxy;
private int itemsPerPage = 20;
private HadoopSecurityManager hadoopSecurityManager;
public ReportalServlet(final Props props) {
this.props = props;
this.viewerName = props.getString("viewer.name");
this.reportalStorageUser = props.getString("reportal.storage.user", "reportal");
this.itemsPerPage = props.getInt("reportal.items_per_page", 20);
this.showNav = props.getBoolean("reportal.show.navigation", false);
this.max_allowed_schedule_dates = props.getInt("reportal.max.allowed.schedule.dates", 180);
this.default_schedule_dates = props.getInt("reportal.default.schedule.dates", 30);
this.reportalMailTempDirectory =
new File(props.getString("reportal.mail.temp.dir", "/tmp/reportal"));
this.reportalMailTempDirectory.mkdirs();
ReportalMailCreator.reportalMailTempDirectory = this.reportalMailTempDirectory;
final List<String> allowedDomains =
props.getStringList("reportal.allowed.email.domains",
(List<String>) null);
if (allowedDomains != null) {
this.allowedEmailDomains = new HashSet<>(allowedDomains);
}
ReportalMailCreator.outputLocation =
props.getString("reportal.output.dir", "/tmp/reportal");
ReportalMailCreator.outputFileSystem =
props.getString("reportal.output.filesystem", "local");
ReportalMailCreator.reportalStorageUser = this.reportalStorageUser;
this.webResourcesFolder =
new File(new File(props.getSource()).getParentFile().getParentFile(),
"web");
this.webResourcesFolder.mkdirs();
setResourceDirectory(this.webResourcesFolder);
System.out.println("Reportal web resources: "
+ this.webResourcesFolder.getAbsolutePath());
}
@Override
public void init(final ServletConfig config) throws ServletException {
super.init(config);
this.server = (AzkabanWebServer) getApplication();
ReportalMailCreator.azkaban = this.server;
this.shouldProxy = this.props.getBoolean("azkaban.should.proxy", false);
logger.info("Hdfs browser should proxy: " + this.shouldProxy);
try {
this.hadoopSecurityManager = loadHadoopSecurityManager(this.props, logger);
ReportalMailCreator.hadoopSecurityManager = this.hadoopSecurityManager;
} catch (final RuntimeException e) {
e.printStackTrace();
throw new RuntimeException("Failed to get hadoop security manager!"
+ e.getCause());
}
this.cleanerThread = new CleanerThread();
this.cleanerThread.start();
}
private HadoopSecurityManager loadHadoopSecurityManager(final Props props,
final Logger logger) throws RuntimeException {
final Class<?> hadoopSecurityManagerClass =
props.getClass(HADOOP_SECURITY_MANAGER_CLASS_PARAM, true,
ReportalServlet.class.getClassLoader());
logger.info("Initializing hadoop security manager "
+ hadoopSecurityManagerClass.getName());
HadoopSecurityManager hadoopSecurityManager = null;
try {
final Method getInstanceMethod =
hadoopSecurityManagerClass.getMethod("getInstance", Props.class);
hadoopSecurityManager =
(HadoopSecurityManager) getInstanceMethod.invoke(
hadoopSecurityManagerClass, props);
} catch (final InvocationTargetException e) {
logger.error("Could not instantiate Hadoop Security Manager "
+ hadoopSecurityManagerClass.getName() + e.getCause());
throw new RuntimeException(e.getCause());
} catch (final Exception e) {
e.printStackTrace();
throw new RuntimeException(e.getCause());
}
return hadoopSecurityManager;
}
@Override
protected void handleGet(final HttpServletRequest req, final HttpServletResponse resp,
final Session session) throws ServletException, IOException {
if (hasParam(req, "ajax")) {
handleAJAXAction(req, resp, session);
} else {
if (hasParam(req, "view")) {
try {
handleViewReportal(req, resp, session);
} catch (final Exception e) {
e.printStackTrace();
}
} else if (hasParam(req, "new")) {
handleNewReportal(req, resp, session);
} else if (hasParam(req, "edit")) {
handleEditReportal(req, resp, session);
} else if (hasParam(req, "run")) {
handleRunReportal(req, resp, session);
} else {
handleListReportal(req, resp, session);
}
}
}
private void handleAJAXAction(final HttpServletRequest req,
final HttpServletResponse resp, final Session session) throws ServletException,
IOException {
final HashMap<String, Object> ret = new HashMap<>();
final String ajaxName = getParam(req, "ajax");
final User user = session.getUser();
final int id = getIntParam(req, "id");
final ProjectManager projectManager = this.server.getProjectManager();
final Project project = projectManager.getProject(id);
final Reportal reportal = Reportal.loadFromProject(project);
// Delete report
if (ajaxName.equals("delete")) {
if (!project.hasPermission(user, Type.ADMIN)) {
ret.put("error", "You do not have permissions to delete this reportal.");
} else {
try {
final ScheduleManager scheduleManager = this.server.getScheduleManager();
reportal.removeSchedules(scheduleManager);
projectManager.removeProject(project, user);
} catch (final Exception e) {
e.printStackTrace();
ret.put("error", "An exception occured while deleting this reportal.");
}
ret.put("result", "success");
}
}
// Bookmark report
else if (ajaxName.equals("bookmark")) {
final boolean wasBookmarked = ReportalHelper.isBookmarkProject(project, user);
try {
if (wasBookmarked) {
ReportalHelper.unBookmarkProject(this.server, project, user);
ret.put("result", "success");
ret.put("bookmark", false);
} else {
ReportalHelper.bookmarkProject(this.server, project, user);
ret.put("result", "success");
ret.put("bookmark", true);
}
} catch (final ProjectManagerException e) {
e.printStackTrace();
ret.put("error", "Error bookmarking reportal. " + e.getMessage());
}
}
// Subscribe to report
else if (ajaxName.equals("subscribe")) {
final boolean wasSubscribed = ReportalHelper.isSubscribeProject(project, user);
if (!wasSubscribed && reportal.getAccessViewers().size() > 0
&& !hasPermission(project, user, Type.READ)) {
ret.put("error", "You do not have permissions to view this reportal.");
} else {
try {
if (wasSubscribed) {
ReportalHelper.unSubscribeProject(this.server, project, user);
ret.put("result", "success");
ret.put("subscribe", false);
} else {
ReportalHelper.subscribeProject(this.server, project, user,
user.getEmail());
ret.put("result", "success");
ret.put("subscribe", true);
}
} catch (final ProjectManagerException e) {
e.printStackTrace();
ret.put("error", "Error subscribing to reportal. " + e.getMessage());
}
}
}
// Get a portion of logs
else if (ajaxName.equals("log")) {
final int execId = getIntParam(req, "execId");
final String jobId = getParam(req, "jobId");
final int offset = getIntParam(req, "offset");
final int length = getIntParam(req, "length");
final ExecutableFlow exec;
final ExecutorManagerAdapter executorManagerAdapter = this.server.getExecutorManager();
try {
exec = executorManagerAdapter.getExecutableFlow(execId);
} catch (final Exception e) {
ret.put("error", "Log does not exist or isn't created yet.");
return;
}
final LogData data;
try {
data =
executorManagerAdapter.getExecutionJobLog(exec, jobId, offset, length,
exec.getExecutableNode(jobId).getAttempt());
} catch (final Exception e) {
e.printStackTrace();
ret.put("error", "Log does not exist or isn't created yet.");
return;
}
if (data != null) {
ret.put("result", "success");
ret.put("log", data.getData());
ret.put("offset", data.getOffset());
ret.put("length", data.getLength());
ret.put("completed", exec.getEndTime() != -1);
} else {
// Return an empty result to indicate the end
ret.put("result", "success");
ret.put("log", "");
ret.put("offset", offset);
ret.put("length", 0);
ret.put("completed", exec.getEndTime() != -1);
}
}
if (ret != null) {
this.writeJSON(resp, ret);
}
}
private void handleListReportal(final HttpServletRequest req,
final HttpServletResponse resp, final Session session) throws ServletException,
IOException {
final Page page =
newPage(req, resp, session,
"azkaban/viewer/reportal/reportallistpage.vm");
preparePage(page, session);
final List<Project> projects = ReportalHelper.getReportalProjects(this.server);
page.add("ReportalHelper", ReportalHelper.class);
page.add("user", session.getUser());
final String startDate = DateTime.now().minusWeeks(1).toString("yyyy-MM-dd");
final String endDate = DateTime.now().toString("yyyy-MM-dd");
page.add("startDate", startDate);
page.add("endDate", endDate);
if (!projects.isEmpty()) {
page.add("projects", projects);
} else {
page.add("projects", false);
}
page.render();
}
private void handleViewReportal(final HttpServletRequest req,
final HttpServletResponse resp, final Session session) throws ServletException,
Exception {
final int id = getIntParam(req, "id");
final Page page =
newPage(req, resp, session,
"azkaban/viewer/reportal/reportaldatapage.vm");
preparePage(page, session);
final ProjectManager projectManager = this.server.getProjectManager();
final ExecutorManagerAdapter executorManagerAdapter = this.server.getExecutorManager();
final Project project = projectManager.getProject(id);
final Reportal reportal = Reportal.loadFromProject(project);
if (reportal == null) {
page.add("errorMsg", "Report not found.");
page.render();
return;
}
if (reportal.getAccessViewers().size() > 0
&& !hasPermission(project, session.getUser(), Type.READ)) {
page.add("errorMsg", "You are not allowed to view this report.");
page.render();
return;
}
page.add("project", project);
page.add("title", project.getMetadata().get("title"));
if (hasParam(req, "execid")) {
final int execId = getIntParam(req, "execid");
page.add("execid", execId);
// Show logs
if (hasParam(req, "logs")) {
final ExecutableFlow exec;
try {
exec = executorManagerAdapter.getExecutableFlow(execId);
} catch (final ExecutorManagerException e) {
e.printStackTrace();
page.add("errorMsg", "ExecutableFlow not found. " + e.getMessage());
page.render();
return;
}
// View single log
if (hasParam(req, "log")) {
page.add("view-log", true);
final String jobId = getParam(req, "log");
page.add("execid", execId);
page.add("jobId", jobId);
}
// List files
else {
page.add("view-logs", true);
final List<ExecutableNode> jobLogs = ReportalUtil.sortExecutableNodes(exec);
final boolean showDataCollector = hasParam(req, "debug");
if (!showDataCollector) {
jobLogs.remove(jobLogs.size() - 1);
}
if (jobLogs.size() == 1) {
resp.sendRedirect("/reportal?view&logs&id=" + project.getId()
+ "&execid=" + execId + "&log=" + jobLogs.get(0).getId());
}
page.add("logs", jobLogs);
}
}
// Show data files
else {
final String outputFileSystem = ReportalMailCreator.outputFileSystem;
final String outputBase = ReportalMailCreator.outputLocation;
final String locationFull = (outputBase + "/" + execId).replace("//", "/");
final IStreamProvider streamProvider =
ReportalUtil.getStreamProvider(outputFileSystem);
if (streamProvider instanceof StreamProviderHDFS) {
final StreamProviderHDFS hdfsStreamProvider =
(StreamProviderHDFS) streamProvider;
hdfsStreamProvider.setHadoopSecurityManager(this.hadoopSecurityManager);
hdfsStreamProvider.setUser(this.reportalStorageUser);
}
try {
if (hasParam(req, "download")) {
final String fileName = getParam(req, "download");
final String filePath = locationFull + "/" + fileName;
InputStream csvInputStream = null;
OutputStream out = null;
try {
csvInputStream = streamProvider.getFileInputStream(filePath);
resp.setContentType("application/octet-stream");
out = resp.getOutputStream();
IOUtils.copy(csvInputStream, out);
} finally {
IOUtils.closeQuietly(out);
IOUtils.closeQuietly(csvInputStream);
}
return;
}
// Show file previews
else {
page.add("view-preview", true);
try {
String[] fileList = streamProvider.getFileList(locationFull);
fileList = ReportalHelper.filterCSVFile(fileList);
Arrays.sort(fileList);
final List<Object> files =
getFilePreviews(fileList, locationFull, streamProvider,
reportal.renderResultsAsHtml);
page.add("files", files);
} catch (final Exception e) {
logger.error("Error encountered while processing files in "
+ locationFull, e);
}
}
} finally {
try {
streamProvider.cleanUp();
} catch (final IOException e) {
e.printStackTrace();
}
}
}
}
// List executions and their data
else {
page.add("view-executions", true);
final ArrayList<ExecutableFlow> exFlows = new ArrayList<>();
int pageNumber = 0;
boolean hasNextPage = false;
if (hasParam(req, "page")) {
pageNumber = getIntParam(req, "page") - 1;
}
if (pageNumber < 0) {
pageNumber = 0;
}
try {
final Flow flow = project.getFlows().get(0);
executorManagerAdapter.getExecutableFlows(project.getId(), flow.getId(),
pageNumber * this.itemsPerPage, this.itemsPerPage, exFlows);
final ArrayList<ExecutableFlow> tmp = new ArrayList<>();
executorManagerAdapter.getExecutableFlows(project.getId(), flow.getId(),
(pageNumber + 1) * this.itemsPerPage, 1, tmp);
if (!tmp.isEmpty()) {
hasNextPage = true;
}
} catch (final ExecutorManagerException e) {
page.add("error", "Error retrieving executable flows");
}
if (!exFlows.isEmpty()) {
final ArrayList<Object> history = new ArrayList<>();
for (final ExecutableFlow exFlow : exFlows) {
final HashMap<String, Object> flowInfo = new HashMap<>();
flowInfo.put("execId", exFlow.getExecutionId());
flowInfo.put("status", exFlow.getStatus().toString());
flowInfo.put("startTime", exFlow.getStartTime());
history.add(flowInfo);
}
page.add("executions", history);
}
if (pageNumber > 0) {
page.add("pagePrev", pageNumber);
}
page.add("page", pageNumber + 1);
if (hasNextPage) {
page.add("pageNext", pageNumber + 2);
}
}
page.render();
}
/**
* Returns a list of file Objects that contain a "name" property with the file
* name, a "content" property with the lines in the file, and a "hasMore"
* property if the file contains more than NUM_PREVIEW_ROWS lines.
*/
private List<Object> getFilePreviews(final String[] fileList, final String locationFull,
final IStreamProvider streamProvider, final boolean renderResultsAsHtml) {
final List<Object> files = new ArrayList<>();
InputStream csvInputStream = null;
try {
for (final String fileName : fileList) {
final Map<String, Object> file = new HashMap<>();
file.put("name", fileName);
final String filePath = locationFull + "/" + fileName;
csvInputStream = streamProvider.getFileInputStream(filePath);
final Scanner rowScanner = new Scanner(csvInputStream, StandardCharsets.UTF_8.toString());
final List<Object> lines = new ArrayList<>();
int lineNumber = 0;
while (rowScanner.hasNextLine()
&& lineNumber < ReportalMailCreator.NUM_PREVIEW_ROWS) {
final String csvLine = rowScanner.nextLine();
final String[] data = csvLine.split("\",\"");
final List<String> line = new ArrayList<>();
for (final String item : data) {
String column = item.replace("\"", "");
if (!renderResultsAsHtml) {
column = StringEscapeUtils.escapeHtml(column);
}
line.add(column);
}
lines.add(line);
lineNumber++;
}
file.put("content", lines);
if (rowScanner.hasNextLine()) {
file.put("hasMore", true);
}
files.add(file);
rowScanner.close();
}
} catch (final Exception e) {
logger.debug("Error encountered while processing files in "
+ locationFull, e);
} finally {
IOUtils.closeQuietly(csvInputStream);
}
return files;
}
private void handleRunReportal(final HttpServletRequest req,
final HttpServletResponse resp, final Session session) throws ServletException,
IOException {
final int id = getIntParam(req, "id");
final ProjectManager projectManager = this.server.getProjectManager();
final Page page =
newPage(req, resp, session,
"azkaban/viewer/reportal/reportalrunpage.vm");
preparePage(page, session);
final Project project = projectManager.getProject(id);
final Reportal reportal = Reportal.loadFromProject(project);
if (reportal == null) {
page.add("errorMsg", "Report not found");
page.render();
return;
}
if (reportal.getAccessExecutors().size() > 0
&& !hasPermission(project, session.getUser(), Type.EXECUTE)) {
page.add("errorMsg", "You are not allowed to run this report.");
page.render();
return;
}
page.add("projectId", id);
page.add("title", reportal.title);
page.add("description", reportal.description);
final List<Variable> runtimeVariables =
ReportalUtil.getRunTimeVariables(reportal.variables);
if (runtimeVariables.size() > 0) {
page.add("variableNumber", runtimeVariables.size());
page.add("variables", runtimeVariables);
}
page.render();
}
private void handleNewReportal(final HttpServletRequest req,
final HttpServletResponse resp, final Session session) throws ServletException,
IOException {
final Page page =
newPage(req, resp, session,
"azkaban/viewer/reportal/reportaleditpage.vm");
preparePage(page, session);
page.add("title", "");
page.add("description", "");
page.add("queryNumber", 1);
final List<Map<String, Object>> queryList = new ArrayList<>();
page.add("queries", queryList);
final Map<String, Object> query = new HashMap<>();
queryList.add(query);
query.put("title", "");
query.put("type", "");
query.put("script", "");
page.add("accessViewer", "");
page.add("accessExecutor", "");
page.add("accessOwner", "");
page.add("notifications", "");
page.add("failureNotifications", "");
page.add("max_allowed_schedule_dates", this.max_allowed_schedule_dates);
page.add("default_schedule_dates", this.default_schedule_dates);
page.render();
}
private void handleEditReportal(final HttpServletRequest req,
final HttpServletResponse resp, final Session session) throws ServletException,
IOException {
final int id = getIntParam(req, "id");
final ProjectManager projectManager = this.server.getProjectManager();
final Page page =
newPage(req, resp, session,
"azkaban/viewer/reportal/reportaleditpage.vm");
preparePage(page, session);
page.add("ReportalHelper", ReportalHelper.class);
final Project project = projectManager.getProject(id);
final Reportal reportal = Reportal.loadFromProject(project);
final List<String> errors = new ArrayList<>();
if (reportal == null) {
errors.add("Report not found");
page.add("errorMsgs", errors);
page.render();
return;
}
if (!hasPermission(project, session.getUser(), Type.ADMIN)) {
errors.add("You are not allowed to edit this report.");
page.add("errorMsgs", errors);
page.render();
return;
}
page.add("projectId", id);
page.add("title", reportal.title);
page.add("description", reportal.description);
page.add("queryNumber", reportal.queries.size());
page.add("queries", reportal.queries);
page.add("variableNumber", reportal.variables.size());
page.add("variables", reportal.variables);
page.add("schedule", reportal.schedule);
page.add("scheduleHour", reportal.scheduleHour);
page.add("scheduleMinute", reportal.scheduleMinute);
page.add("scheduleAmPm", reportal.scheduleAmPm);
page.add("scheduleTimeZone", reportal.scheduleTimeZone);
page.add("scheduleDate", reportal.scheduleDate);
page.add("endScheduleDate", reportal.endSchedule);
page.add("scheduleRepeat", reportal.scheduleRepeat);
page.add("scheduleIntervalQuantity", reportal.scheduleIntervalQuantity);
page.add("scheduleInterval", reportal.scheduleInterval);
page.add("renderResultsAsHtml", reportal.renderResultsAsHtml);
page.add("notifications", reportal.notifications);
page.add("failureNotifications", reportal.failureNotifications);
page.add("accessViewer", reportal.accessViewer);
page.add("accessExecutor", reportal.accessExecutor);
page.add("accessOwner", reportal.accessOwner);
page.add("max_allowed_schedule_dates", this.max_allowed_schedule_dates);
page.add("default_schedule_dates", this.default_schedule_dates);
page.render();
}
@Override
protected void handlePost(final HttpServletRequest req, final HttpServletResponse resp,
final Session session) throws ServletException, IOException {
if (hasParam(req, "ajax")) {
final HashMap<String, Object> ret = new HashMap<>();
handleRunReportalWithVariables(req, ret, session);
if (ret != null) {
this.writeJSON(resp, ret);
}
} else {
handleSaveReportal(req, resp, session);
}
}
private void handleSaveReportal(final HttpServletRequest req,
final HttpServletResponse resp, final Session session) throws ServletException,
IOException {
final String projectId = validateAndSaveReport(req, resp, session);
if (projectId != null) {
this.setSuccessMessageInCookie(resp, "Report Saved.");
final String submitType = getParam(req, "submit");
if (submitType.equals("Save")) {
resp.sendRedirect(req.getRequestURI() + "?edit&id=" + projectId);
} else {
resp.sendRedirect(req.getRequestURI() + "?run&id=" + projectId);
}
}
}
/**
* Validates and saves a report, returning the project id of the saved report
* if successful, and null otherwise.
*
* @return The project id of the saved report if successful, and null
* otherwise
*/
private String validateAndSaveReport(final HttpServletRequest req,
final HttpServletResponse resp, final Session session) throws ServletException,
IOException {
final ProjectManager projectManager = this.server.getProjectManager();
final User user = session.getUser();
final Page page =
newPage(req, resp, session,
"azkaban/viewer/reportal/reportaleditpage.vm");
preparePage(page, session);
page.add("ReportalHelper", ReportalHelper.class);
final boolean isEdit = hasParam(req, "id");
if (isEdit) {
page.add("projectId", getIntParam(req, "id"));
}
Project project = null;
final Reportal report = new Reportal();
report.title = getParam(req, "title");
report.description = getParam(req, "description");
page.add("title", report.title);
page.add("description", report.description);
report.schedule = hasParam(req, "schedule");
report.scheduleHour = getParam(req, "schedule-hour");
report.scheduleMinute = getParam(req, "schedule-minute");
report.scheduleAmPm = getParam(req, "schedule-am_pm");
report.scheduleTimeZone = getParam(req, "schedule-timezone");
report.scheduleDate = getParam(req, "schedule-date");
report.scheduleRepeat = hasParam(req, "schedule-repeat");
report.scheduleIntervalQuantity =
getParam(req, "schedule-interval-quantity");
report.scheduleInterval = getParam(req, "schedule-interval");
report.renderResultsAsHtml = hasParam(req, "render-results-as-html");
final boolean isEndSchedule = hasParam(req, "end-schedule-date");
if (isEndSchedule) {
report.endSchedule = getParam(req, "end-schedule-date");
}
page.add("schedule", report.schedule);
page.add("scheduleHour", report.scheduleHour);
page.add("scheduleMinute", report.scheduleMinute);
page.add("scheduleAmPm", report.scheduleAmPm);
page.add("scheduleTimeZone", report.scheduleTimeZone);
page.add("scheduleDate", report.scheduleDate);
page.add("scheduleRepeat", report.scheduleRepeat);
page.add("scheduleIntervalQuantity", report.scheduleIntervalQuantity);
page.add("scheduleInterval", report.scheduleInterval);
page.add("renderResultsAsHtml", report.renderResultsAsHtml);
page.add("endSchedule", report.endSchedule);
page.add("max_allowed_schedule_dates", this.max_allowed_schedule_dates);
page.add("default_schedule_dates", this.default_schedule_dates);
report.accessViewer = getParam(req, "access-viewer");
report.accessExecutor = getParam(req, "access-executor");
report.accessOwner = getParam(req, "access-owner");
page.add("accessViewer", report.accessViewer);
page.add("accessExecutor", report.accessExecutor);
// Adding report creator as explicit owner, if not present already
if (report.accessOwner == null || report.accessOwner.isEmpty()) {
report.accessOwner = user.getUserId();
} else {
final String[] splittedOwners = report.accessOwner.toLowerCase()
.split(Reportal.ACCESS_LIST_SPLIT_REGEX);
if (!Arrays.asList(splittedOwners).contains(user.getUserId())) {
report.accessOwner = String.format("%s,%s", user.getUserId(),
StringUtils.join(splittedOwners, ','));
} else {
report.accessOwner = StringUtils.join(splittedOwners, ',');
}
}
page.add("accessOwner", report.accessOwner);
report.notifications = getParam(req, "notifications");
report.failureNotifications = getParam(req, "failure-notifications");
page.add("notifications", report.notifications);
page.add("failureNotifications", report.failureNotifications);
final int numQueries = getIntParam(req, "queryNumber");
page.add("queryNumber", numQueries);
final List<Query> queryList = new ArrayList<>(numQueries);
page.add("queries", queryList);
report.queries = queryList;
final List<String> errors = new ArrayList<>();
for (int i = 0; i < numQueries; i++) {
final Query query = new Query();
query.title = getParam(req, "query" + i + "title");
query.type = getParam(req, "query" + i + "type");
query.script = getParam(req, "query" + i + "script");
// Type check
final ReportalType type = ReportalType.getTypeByName(query.type);
if (type == null) {
errors.add("Type " + query.type + " is invalid.");
}
if (!type.checkPermission(user) && report.schedule) {
errors.add("You do not have permission to schedule Type " + query.type + ".");
}
queryList.add(query);
}
final int variables = getIntParam(req, "variableNumber");
page.add("variableNumber", variables);
final List<Variable> variableList = new ArrayList<>(variables);
page.add("variables", variableList);
report.variables = variableList;
String proxyUser = null;
for (int i = 0; i < variables; i++) {
final Variable variable =
new Variable(getParam(req, "variable" + i + "title"), getParam(req,
"variable" + i + "name"));
if (variable.title.isEmpty() || variable.name.isEmpty()) {
errors.add("Variable title and name cannot be empty.");
}
if (variable.title.equals("reportal.config.reportal.execution.user")) {
proxyUser = variable.name;
}
// Validate if the session user (who interact with UI) is part of specified user.to.proxy
// user. If not, reportal can not be saved and warn users.
if (variable.title.equals("reportal.config.user.to.proxy")) {
final String userToProxy = variable.name;
final UserManager userManager = getApplication().getUserManager();
if (!userManager.validateProxyUser(userToProxy, user)) {
errors.add("User " + user.getUserId() + " has no permission to add " + userToProxy
+ " as proxy user.");
}
}
variableList.add(variable);
}
// Make sure title isn't empty
if (report.title.isEmpty()) {
errors.add("Title must not be empty.");
}
// Make sure description isn't empty
if (report.description.isEmpty()) {
errors.add("Description must not be empty.");
}
// Verify schedule and repeat
if (report.schedule) {
// Verify schedule time
if (!NumberUtils.isDigits(report.scheduleHour)
|| !NumberUtils.isDigits(report.scheduleMinute)) {
errors.add("Schedule time is invalid.");
}
// Verify schedule date is not empty
if (report.scheduleDate.isEmpty()) {
errors.add("Schedule date must not be empty.");
}
if (report.scheduleRepeat) {
// Verify repeat interval
if (!NumberUtils.isDigits(report.scheduleIntervalQuantity)) {
errors.add("Repeat interval quantity is invalid.");
}
}
}
// Empty query check
if (numQueries <= 0) {
errors.add("There needs to have at least one query.");
}
// Validate access users
final UserManager userManager = getApplication().getUserManager();
final String[] accessLists =
new String[]{report.accessViewer, report.accessExecutor,
report.accessOwner};
for (String accessList : accessLists) {
if (accessList == null) {
continue;
}
accessList = accessList.trim();
if (!accessList.isEmpty()) {
final String[] users = accessList.split(Reportal.ACCESS_LIST_SPLIT_REGEX);
for (final String accessUser : users) {
if (!userManager.validateUser(accessUser)) {
errors.add("User " + accessUser + " in access list is invalid.");
}
}
}
}
// Validate proxy user
if (proxyUser != null) {
if (!userManager.validateProxyUser(proxyUser, user)) {
errors.add("User " + user.getUserId() + " has no permission to add " + proxyUser
+ " as proxy user.");
}
proxyUser = null;
}
// Validate email addresses
final Set<String> emails =
ReportalHelper.parseUniqueEmails(report.notifications + ","
+ report.failureNotifications, Reportal.ACCESS_LIST_SPLIT_REGEX);
for (final String email : emails) {
if (!ReportalHelper.isValidEmailAddress(email)) {
errors.add("Invalid email address: " + email);
continue;
}
final String domain = ReportalHelper.getEmailDomain(email);
if (this.allowedEmailDomains != null && !this.allowedEmailDomains.contains(domain)) {
errors.add("Email address '" + email + "' has an invalid domain '"
+ domain + "'. " + "Valid domains are: " + this.allowedEmailDomains);
}
}
if (errors.size() > 0) {
page.add("errorMsgs", errors);
page.render();
return null;
}
// Attempt to get a project object
if (isEdit) {
// Editing mode, load project
final int projectId = getIntParam(req, "id");
project = projectManager.getProject(projectId);
report.loadImmutableFromProject(project);
} else {
// Creation mode, create project
try {
project =
ReportalHelper.createReportalProject(this.server, report.title,
report.description, user);
report.reportalUser = user.getUserId();
report.ownerEmail = user.getEmail();
} catch (final Exception e) {
e.printStackTrace();
errors.add("Error while creating report. " + e.getMessage());
page.add("errorMsgs", errors);
page.render();
return null;
}
// Project already exists
if (project == null) {
errors.add("A Report with the same name already exists.");
page.add("errorMsgs", errors);
page.render();
return null;
}
}
if (project == null) {
errors.add("Internal Error: Report not found");
page.add("errorMsgs", errors);
page.render();
return null;
}
report.project = project;
page.add("projectId", project.getId());
try {
report.createZipAndUpload(projectManager, user, this.reportalStorageUser);
} catch (final Exception e) {
e.printStackTrace();
errors.add("Error while creating Azkaban jobs. " + e.getMessage());
page.add("errorMsgs", errors);
page.render();
if (!isEdit) {
try {
projectManager.removeProject(project, user);
} catch (final ProjectManagerException e1) {
e1.printStackTrace();
}
}
return null;
}
// Prepare flow
final Flow flow = project.getFlows().get(0);
project.getMetadata().put("flowName", flow.getId());
// Set Reportal mailer
flow.setMailCreator(ReportalMailCreator.REPORTAL_MAIL_CREATOR);
// Create/Save schedule
final ScheduleManager scheduleManager = this.server.getScheduleManager();
try {
report.updateSchedules(report, scheduleManager, user, flow);
} catch (final ScheduleManagerException e) {
e.printStackTrace();
errors.add(e.getMessage());
page.add("errorMsgs", errors);
page.render();
return null;
}
report.saveToProject(project);
try {
ReportalHelper.updateProjectNotifications(project, projectManager);
projectManager.updateProjectSetting(project);
projectManager
.updateProjectDescription(project, report.description, user);
updateProjectPermissions(project, projectManager, report, user);
projectManager.updateFlow(project, flow);
} catch (final ProjectManagerException e) {
e.printStackTrace();
errors.add("Error while updating report. " + e.getMessage());
page.add("errorMsgs", errors);
page.render();
if (!isEdit) {
try {
projectManager.removeProject(project, user);
} catch (final ProjectManagerException e1) {
e1.printStackTrace();
}
}
return null;
}
return Integer.toString(project.getId());
}
private void updateProjectPermissions(final Project project,
final ProjectManager projectManager, final Reportal report, final User currentUser)
throws ProjectManagerException {
// Old permissions and users
final List<Pair<String, Permission>> oldPermissions =
project.getUserPermissions();
final Set<String> oldUsers = new HashSet<>();
for (final Pair<String, Permission> userPermission : oldPermissions) {
oldUsers.add(userPermission.getFirst());
}
// Update permissions
report.updatePermissions();
// New permissions and users
final List<Pair<String, Permission>> newPermissions =
project.getUserPermissions();
final Set<String> newUsers = new HashSet<>();
for (final Pair<String, Permission> userPermission : newPermissions) {
newUsers.add(userPermission.getFirst());
}
// Save all new permissions
for (final Pair<String, Permission> userPermission : newPermissions) {
if (!oldPermissions.contains(userPermission)) {
projectManager.updateProjectPermission(project,
userPermission.getFirst(), userPermission.getSecond(), false,
currentUser);
}
}
// Remove permissions for any old users no longer in the new users
for (final String oldUser : oldUsers) {
if (!newUsers.contains(oldUser)) {
projectManager.removeProjectPermission(project, oldUser, false,
currentUser);
}
}
}
private void handleRunReportalWithVariables(final HttpServletRequest req,
final HashMap<String, Object> ret, final Session session) throws ServletException,
IOException {
final boolean isTestRun = hasParam(req, "testRun");
final int id = getIntParam(req, "id");
final ProjectManager projectManager = this.server.getProjectManager();
final Project project = projectManager.getProject(id);
final Reportal report = Reportal.loadFromProject(project);
final User user = session.getUser();
if (report.getAccessExecutors().size() > 0
&& !hasPermission(project, user, Type.EXECUTE)) {
ret.put("error", "You are not allowed to run this report.");
return;
}
for (final Query query : report.queries) {
final String jobType = query.type;
final ReportalType type = ReportalType.getTypeByName(jobType);
if (!type.checkPermission(user)) {
ret.put(
"error",
"You are not allowed to run this report as you don't have permission to run job type "
+ type.toString() + ".");
return;
}
}
final Flow flow = project.getFlows().get(0);
final ExecutableFlow exflow = new ExecutableFlow(project, flow);
exflow.setSubmitUser(user.getUserId());
exflow.addAllProxyUsers(project.getProxyUsers());
final ExecutionOptions options = exflow.getExecutionOptions();
int i = 0;
for (final Variable variable : ReportalUtil.getRunTimeVariables(report.variables)) {
options.getFlowParameters().put(REPORTAL_VARIABLE_PREFIX + i + ".from",
variable.name);
options.getFlowParameters().put(REPORTAL_VARIABLE_PREFIX + i + ".to",
getParam(req, "variable" + i));
i++;
}
options.getFlowParameters()
.put("reportal.execution.user", user.getUserId());
// Add the execution user's email to the list of success and failure emails.
final String email = user.getEmail();
if (email != null && !email.isEmpty()) {
if (isTestRun) { // Only email the executor
final List<String> emails = new ArrayList<>();
emails.add(email);
options.setSuccessEmails(emails);
options.setFailureEmails(emails);
} else {
options.getSuccessEmails().add(email);
options.getFailureEmails().add(email);
}
}
options.getFlowParameters().put("reportal.title", report.title);
options.getFlowParameters().put("reportal.render.results.as.html",
report.renderResultsAsHtml ? "true" : "false");
options.getFlowParameters().put("reportal.unscheduled.run", "true");
try {
final String message =
this.server.getExecutorManager().submitExecutableFlow(exflow,
session.getUser().getUserId())
+ ".";
ret.put("message", message);
ret.put("result", "success");
ret.put("redirect", "/reportal?view&logs&id=" + project.getId()
+ "&execid=" + exflow.getExecutionId());
} catch (final ExecutorManagerException e) {
e.printStackTrace();
ret.put("error",
"Error running report " + report.title + ". " + e.getMessage());
}
}
private void preparePage(final Page page, final Session session) {
page.add("viewerName", this.viewerName);
page.add("hideNavigation", !this.showNav);
page.add("userid", session.getUser().getUserId());
page.add("esc", new EscapeTool());
}
private class CleanerThread extends Thread {
private static final long DEFAULT_CLEAN_INTERVAL_MS = 24 * 60 * 60 * 1000;
private static final long DEFAULT_OUTPUT_DIR_RETENTION_MS = 7 * 24 * 60
* 60 * 1000;
private static final long DEFAULT_MAIL_TEMP_DIR_RETENTION_MS =
24 * 60 * 60 * 1000;
// The frequency, in milliseconds, that the Reportal output
// and mail temp directories should be cleaned
private final long CLEAN_INTERVAL_MS;
// The duration, in milliseconds, that Reportal output should be retained
// for
private final long OUTPUT_DIR_RETENTION_MS;
// The duration, in milliseconds, that Reportal mail temp files should be
// retained for
private final long MAIL_TEMP_DIR_RETENTION_MS;
private boolean shutdown = false;
public CleanerThread() {
this.setName("Reportal-Cleaner-Thread");
this.CLEAN_INTERVAL_MS =
ReportalServlet.this.props
.getLong("reportal.clean.interval.ms", DEFAULT_CLEAN_INTERVAL_MS);
this.OUTPUT_DIR_RETENTION_MS =
ReportalServlet.this.props.getLong("reportal.output.dir.retention.ms",
DEFAULT_OUTPUT_DIR_RETENTION_MS);
this.MAIL_TEMP_DIR_RETENTION_MS =
ReportalServlet.this.props.getLong("reportal.mail.temp.dir.retention.ms",
DEFAULT_MAIL_TEMP_DIR_RETENTION_MS);
}
@SuppressWarnings("unused")
public void shutdown() {
this.shutdown = true;
this.interrupt();
}
@Override
public void run() {
while (!this.shutdown) {
synchronized (this) {
logger.info("Cleaning old execution output dirs");
cleanOldReportalOutputDirs();
logger.info("Cleaning Reportal mail temp directory");
cleanReportalMailTempDir();
}
try {
Thread.sleep(this.CLEAN_INTERVAL_MS);
} catch (final InterruptedException e) {
logger.error("CleanerThread's sleep was interrupted.", e);
}
}
}
private void cleanOldReportalOutputDirs() {
final IStreamProvider streamProvider =
ReportalUtil.getStreamProvider(ReportalMailCreator.outputFileSystem);
if (streamProvider instanceof StreamProviderHDFS) {
final StreamProviderHDFS hdfsStreamProvider =
(StreamProviderHDFS) streamProvider;
hdfsStreamProvider.setHadoopSecurityManager(ReportalServlet.this.hadoopSecurityManager);
hdfsStreamProvider.setUser(ReportalServlet.this.reportalStorageUser);
}
final long pastTimeThreshold =
System.currentTimeMillis() - this.OUTPUT_DIR_RETENTION_MS;
String[] oldFiles = null;
try {
oldFiles =
streamProvider.getOldFiles(ReportalMailCreator.outputLocation,
pastTimeThreshold);
} catch (final Exception e) {
logger.error("Error getting old files from "
+ ReportalMailCreator.outputLocation + " on "
+ ReportalMailCreator.outputFileSystem + " file system.", e);
}
if (oldFiles != null) {
for (final String file : oldFiles) {
final String filePath = ReportalMailCreator.outputLocation + "/" + file;
try {
streamProvider.deleteFile(filePath);
} catch (final Exception e) {
logger.error("Error deleting file " + filePath + " from "
+ ReportalMailCreator.outputFileSystem + " file system.", e);
}
}
}
}
private void cleanReportalMailTempDir() {
final File dir = ReportalServlet.this.reportalMailTempDirectory;
final long pastTimeThreshold =
System.currentTimeMillis() - this.MAIL_TEMP_DIR_RETENTION_MS;
final File[] oldMailTempDirs = dir.listFiles(new FileFilter() {
@Override
public boolean accept(final File path) {
if (path.isDirectory() && path.lastModified() < pastTimeThreshold) {
return true;
}
return false;
}
});
for (final File tempDir : oldMailTempDirs) {
try {
FileUtils.deleteDirectory(tempDir);
} catch (final IOException e) {
logger.error(
"Error cleaning Reportal mail temp dir " + tempDir.getPath(), e);
}
}
}
}
}