diff --git a/celos-common/src/main/java/com/collective/celos/CelosClient.java b/celos-common/src/main/java/com/collective/celos/CelosClient.java index aaa165ea5..19987bf40 100644 --- a/celos-common/src/main/java/com/collective/celos/CelosClient.java +++ b/celos-common/src/main/java/com/collective/celos/CelosClient.java @@ -17,6 +17,7 @@ import com.fasterxml.jackson.databind.node.ObjectNode; import com.google.common.collect.Lists; +import org.apache.commons.io.IOUtils; import org.apache.commons.lang.StringUtils; import org.apache.http.HttpResponse; import org.apache.http.client.HttpClient; @@ -48,6 +49,8 @@ public class CelosClient { private static final String CLEAR_CACHE_PATH = "/clear-cache"; private static final String WORKFLOW_LIST_PATH = "/workflow-list"; private static final String WORKFLOW_SLOTS_PATH = "/workflow-slots"; + private static final String WORKFLOW_FILE_PATH = "/workflow-file"; + private static final String START_TIME_PARAM = "start"; private static final String END_TIME_PARAM = "end"; private static final String TIME_PARAM = "time"; @@ -163,6 +166,17 @@ public void setWorkflowPaused(WorkflowID workflowID, Boolean paused) throws Exce executePost(uriBuilder.build()); } + public String getWorkflowFile(WorkflowID workflowID) throws Exception { + URIBuilder uriBuilder = new URIBuilder(address); + uriBuilder.setPath(WORKFLOW_FILE_PATH); + uriBuilder.addParameter(ID_PARAM, workflowID.toString()); + + HttpGet workflowListGet = new HttpGet(uriBuilder.build()); + HttpResponse getResponse = execute(workflowListGet); + InputStream content = getResponse.getEntity().getContent(); + return IOUtils.toString(content); + } + public void kill(WorkflowID workflowID, ScheduledTime scheduledTime) throws Exception { URIBuilder uriBuilder = new URIBuilder(address); uriBuilder.setPath(uriBuilder.getPath() + KILL_PATH); diff --git a/celos-server/src/main/java/com/collective/celos/WorkflowConfiguration.java b/celos-server/src/main/java/com/collective/celos/WorkflowConfiguration.java index 3e61c9288..5eb3303f1 100644 --- a/celos-server/src/main/java/com/collective/celos/WorkflowConfiguration.java +++ b/celos-server/src/main/java/com/collective/celos/WorkflowConfiguration.java @@ -25,22 +25,31 @@ public class WorkflowConfiguration { private final Map workflows = new HashMap<>(); + private final Map workflowsToFiles = new HashMap<>(); + + public WorkflowConfiguration() { + } public Collection getWorkflows() { return workflows.values(); } - + public Workflow findWorkflow(WorkflowID id) { return workflows.get(Util.requireNonNull(id)); } - public void addWorkflow(Workflow wf) { + public String getWorkflowJSFileName(WorkflowID id) { + return workflowsToFiles.get(Util.requireNonNull(id)); + } + + public void addWorkflow(Workflow wf, String filePath) { Util.requireNonNull(wf); WorkflowID id = wf.getID(); if (findWorkflow(id) != null) { throw new IllegalArgumentException("Workflow with this ID already exists: " + id); } workflows.put(id, wf); + workflowsToFiles.put(id, filePath); } - + } diff --git a/celos-server/src/main/java/com/collective/celos/WorkflowConfigurationParser.java b/celos-server/src/main/java/com/collective/celos/WorkflowConfigurationParser.java index ba9362528..1aba6eadf 100644 --- a/celos-server/src/main/java/com/collective/celos/WorkflowConfigurationParser.java +++ b/celos-server/src/main/java/com/collective/celos/WorkflowConfigurationParser.java @@ -78,7 +78,7 @@ Object evaluateReader(Reader r, String fileName) throws Exception { Object wrappedThis = Context.javaToJS(this, scope); Map jsProperties = Maps.newHashMap(additionalJsVariables); jsProperties.put("celosWorkflowConfigurationParser", wrappedThis); - + jsProperties.put("celosWorkflowFilePath", fileName); jsConfigParser.putPropertiesInScope(jsProperties, scope); InputStream scripts = WorkflowConfigurationParser.class.getResourceAsStream("celos-scripts.js"); @@ -99,8 +99,10 @@ public void importDefaultsIntoScope(String label, Global scope) throws IOExcepti jsConfigParser.evaluateReader(scope, fileReader, fileName); } - public void addWorkflow(Workflow wf) { - cfg.addWorkflow(wf); + public void addWorkflow(Workflow wf, String workflowFilePath) { + cfg.addWorkflow(wf, workflowFilePath); } + + } diff --git a/celos-server/src/main/java/com/collective/celos/servlet/WorkflowFileServlet.java b/celos-server/src/main/java/com/collective/celos/servlet/WorkflowFileServlet.java new file mode 100644 index 000000000..d8fdc44d3 --- /dev/null +++ b/celos-server/src/main/java/com/collective/celos/servlet/WorkflowFileServlet.java @@ -0,0 +1,62 @@ +package com.collective.celos.servlet; + +import java.io.File; + +import javax.servlet.ServletException; +import javax.servlet.http.HttpServletRequest; +import javax.servlet.http.HttpServletResponse; + +import org.apache.commons.io.FileUtils; + +import com.collective.celos.WorkflowID; + +/** + * Returns JS-config file contents for particular workflow-id + * + * GET /workflow-file?id=workflow-1 + * ==> + * + * addWorkflow({ + * "id": "workflow-1", + * "schedule": hourlySchedule(), + * "schedulingStrategy": serialSchedulingStrategy(), + * "trigger": hdfsCheckTrigger("foo", "file:///"), + * "externalService": oozieExternalService({}, "oj01/oozie"), + * "maxRetryCount": 0 + * }); + * + * + */ +@SuppressWarnings("serial") +public class WorkflowFileServlet extends AbstractJSONServlet { + + private static final String ID_PARAM = "id"; + + protected void doGet(HttpServletRequest req, HttpServletResponse res) throws ServletException { + String id = req.getParameter(ID_PARAM); + try { + if (id == null) { + res.sendError(HttpServletResponse.SC_BAD_REQUEST, ID_PARAM + " parameter missing."); + return; + } + String contents = getWorkflowConfigurationFileContents(id); + if (contents == null) { + res.sendError(HttpServletResponse.SC_NOT_FOUND, "JS config for workflow not found: " + id); + } else { + res.getOutputStream().write(contents.getBytes()); + } + } catch (Exception e) { + throw new ServletException(e); + } + } + + public String getWorkflowConfigurationFileContents(String workflowId) throws Exception { + String filePath = getOrCreateCachedScheduler().getWorkflowConfiguration().getWorkflowJSFileName(new WorkflowID(workflowId)); + if (filePath == null) { + return null; + } else { + return FileUtils.readFileToString(new File(filePath)); + } + } + +} diff --git a/celos-server/src/main/resources/com/collective/celos/celos-scripts.js b/celos-server/src/main/resources/com/collective/celos/celos-scripts.js index 4a662dd63..318e729c1 100644 --- a/celos-server/src/main/resources/com/collective/celos/celos-scripts.js +++ b/celos-server/src/main/resources/com/collective/celos/celos-scripts.js @@ -70,7 +70,7 @@ celos.defineWorkflow = function (json) { workflowInfo ); - celosWorkflowConfigurationParser.addWorkflow(workflow); + celosWorkflowConfigurationParser.addWorkflow(workflow, celosWorkflowFilePath); } diff --git a/celos-server/src/main/webapp/WEB-INF/web.xml b/celos-server/src/main/webapp/WEB-INF/web.xml index 7efb36931..74bb40274 100644 --- a/celos-server/src/main/webapp/WEB-INF/web.xml +++ b/celos-server/src/main/webapp/WEB-INF/web.xml @@ -86,6 +86,15 @@ /pause + + WorkflowFileServlet + com.collective.celos.servlet.WorkflowFileServlet + + + WorkflowFileServlet + /workflow-file + + KillServlet com.collective.celos.servlet.KillServlet diff --git a/celos-server/src/test/java/com/collective/celos/CelosClientServerTest.java b/celos-server/src/test/java/com/collective/celos/CelosClientServerTest.java index ac487b2df..df5d09b7f 100644 --- a/celos-server/src/test/java/com/collective/celos/CelosClientServerTest.java +++ b/celos-server/src/test/java/com/collective/celos/CelosClientServerTest.java @@ -698,6 +698,31 @@ public void testPauseWorkflow() throws Exception { } @Test + public void testGetWorkflowFile() throws Exception { + File src = new File(Thread.currentThread().getContextClassLoader().getResource("com/collective/celos/client/wf-list").toURI()); + FileUtils.copyDirectory(src, workflowsDir); + + String workflowFile1 = celosClient.getWorkflowFile(new WorkflowID("workflow-1")); + String workflowFile2 = celosClient.getWorkflowFile(new WorkflowID("workflow-2")); + + File wfFile1 = new File(Thread.currentThread().getContextClassLoader().getResource("com/collective/celos/client/wf-list/workflow-1.js").toURI()); + File wfFile2 = new File(Thread.currentThread().getContextClassLoader().getResource("com/collective/celos/client/wf-list/workflow-2.js").toURI()); + + String expectedWfFile1 = FileUtils.readFileToString(wfFile1); + String expectedWfFile2 = FileUtils.readFileToString(wfFile2); + + Assert.assertEquals(expectedWfFile1, workflowFile1); + Assert.assertEquals(expectedWfFile2, workflowFile2); + } + + @Test (expected = IOException.class) + public void testGetWorkflowFileFailsNoWF() throws Exception { + File src = new File(Thread.currentThread().getContextClassLoader().getResource("com/collective/celos/client/wf-list").toURI()); + FileUtils.copyDirectory(src, workflowsDir); + + celosClient.getWorkflowFile(new WorkflowID("thereisnosuchwf")); + } + public void testKill() throws Exception { File src = new File(Thread.currentThread().getContextClassLoader().getResource("com/collective/celos/client/wf-list").toURI()); FileUtils.copyDirectory(src, workflowsDir); diff --git a/celos-server/src/test/java/com/collective/celos/DependentScheduleTest.java b/celos-server/src/test/java/com/collective/celos/DependentScheduleTest.java index 553859b44..3d9b5aa17 100644 --- a/celos-server/src/test/java/com/collective/celos/DependentScheduleTest.java +++ b/celos-server/src/test/java/com/collective/celos/DependentScheduleTest.java @@ -69,7 +69,7 @@ public void usesScheduleOfOtherWorkflow() { slidingWindowHours, workflowStartTime, Workflow.DEFAULT_WAIT_TIMEOUT_SECONDS, - new WorkflowInfo(null, Collections.emptyList()))); + new WorkflowInfo(null, Collections.emptyList())), ""); DependentSchedule schedule = new DependentSchedule(new WorkflowID(id)); SortedSet times = schedule.getScheduledTimes(scheduler, schedulingStartTime, schedulingEndTime); diff --git a/celos-server/src/test/java/com/collective/celos/SchedulerTest.java b/celos-server/src/test/java/com/collective/celos/SchedulerTest.java index 217406cbd..bf1c2ab8e 100644 --- a/celos-server/src/test/java/com/collective/celos/SchedulerTest.java +++ b/celos-server/src/test/java/com/collective/celos/SchedulerTest.java @@ -344,7 +344,7 @@ public void updatesWaitingSlotsToReady() throws Exception { Workflow wf1 = new Workflow(wfID1, sch1, str1, tr1, srv1, maxRetryCount, Workflow.DEFAULT_START_TIME, Workflow.DEFAULT_WAIT_TIMEOUT_SECONDS, emptyWorkflowInfo); WorkflowConfiguration cfg = new WorkflowConfiguration(); - cfg.addWorkflow(wf1); + cfg.addWorkflow(wf1, ""); MemoryStateDatabase db = new MemoryStateDatabase(); @@ -386,7 +386,7 @@ public void rerunTest() throws Exception { Workflow wf1 = new Workflow(wfID1, sch1, str1, tr1, srv1, maxRetryCount, Workflow.DEFAULT_START_TIME, Workflow.DEFAULT_WAIT_TIMEOUT_SECONDS, emptyWorkflowInfo); WorkflowConfiguration cfg = new WorkflowConfiguration(); - cfg.addWorkflow(wf1); + cfg.addWorkflow(wf1, ""); MemoryStateDatabase db = new MemoryStateDatabase(); @@ -425,7 +425,7 @@ public void doesNotUpdateWaitingSlotsToReadyWhenNoDataAvailability() throws Exce Workflow wf1 = new Workflow(wfID1, sch1, str1, tr1, srv1, maxRetryCount, Workflow.DEFAULT_START_TIME, Workflow.DEFAULT_WAIT_TIMEOUT_SECONDS, emptyWorkflowInfo); WorkflowConfiguration cfg = new WorkflowConfiguration(); - cfg.addWorkflow(wf1); + cfg.addWorkflow(wf1, ""); MemoryStateDatabase db = new MemoryStateDatabase(); @@ -450,7 +450,7 @@ public void updatesWaitingSlotsToWaitTimeout() throws Exception { Workflow wf1 = new Workflow(wfID1, sch1, str1, tr1, srv1, maxRetryCount, Workflow.DEFAULT_START_TIME, waitTimeoutSeconds, emptyWorkflowInfo); WorkflowConfiguration cfg = new WorkflowConfiguration(); - cfg.addWorkflow(wf1); + cfg.addWorkflow(wf1, ""); MemoryStateDatabase db = new MemoryStateDatabase(); @@ -513,7 +513,7 @@ private void runningSlotUtil(ExternalStatus externalStatus, SlotState.Status exp Workflow wf1 = new Workflow(wfID1, sch1, str1, tr1, srv1, maxRetryCount, Workflow.DEFAULT_START_TIME, Workflow.DEFAULT_WAIT_TIMEOUT_SECONDS, emptyWorkflowInfo); WorkflowConfiguration cfg = new WorkflowConfiguration(); - cfg.addWorkflow(wf1); + cfg.addWorkflow(wf1, ""); MemoryStateDatabase db = new MemoryStateDatabase(); @@ -560,7 +560,7 @@ public void updatesReadySlotsToRunningAndSubmitsThemToExternalSystem() throws Ex Workflow wf1 = new Workflow(wfID1, sch1, str1, tr1, srv1, maxRetryCount, Workflow.DEFAULT_START_TIME, Workflow.DEFAULT_WAIT_TIMEOUT_SECONDS, emptyWorkflowInfo); WorkflowConfiguration cfg = new WorkflowConfiguration(); - cfg.addWorkflow(wf1); + cfg.addWorkflow(wf1, ""); MemoryStateDatabase db = new MemoryStateDatabase(); @@ -620,7 +620,7 @@ public void multipleSlotsTest() throws Exception { Workflow wf1 = new Workflow(wfID1, sch1, str1, tr1, srv1, maxRetryCount, Workflow.DEFAULT_START_TIME, Workflow.DEFAULT_WAIT_TIMEOUT_SECONDS, emptyWorkflowInfo); WorkflowConfiguration cfg = new WorkflowConfiguration(); - cfg.addWorkflow(wf1); + cfg.addWorkflow(wf1, ""); MemoryStateDatabase db = new MemoryStateDatabase(); @@ -677,7 +677,7 @@ public void workflowStartTimeTest() throws Exception { Workflow wf1 = new Workflow(wfID1, sch1, str1, tr1, srv1, maxRetryCount, startTime, Workflow.DEFAULT_WAIT_TIMEOUT_SECONDS, emptyWorkflowInfo); WorkflowConfiguration cfg = new WorkflowConfiguration(); - cfg.addWorkflow(wf1); + cfg.addWorkflow(wf1, ""); MemoryStateDatabase db = new MemoryStateDatabase(); @@ -704,7 +704,7 @@ public void workflowStartTimeTest2() throws Exception { Workflow wf1 = new Workflow(wfID1, sch1, str1, tr1, srv1, maxRetryCount, startTime, Workflow.DEFAULT_WAIT_TIMEOUT_SECONDS, emptyWorkflowInfo); WorkflowConfiguration cfg = new WorkflowConfiguration(); - cfg.addWorkflow(wf1); + cfg.addWorkflow(wf1, ""); MemoryStateDatabase db = new MemoryStateDatabase(); @@ -731,7 +731,7 @@ public void workflowStartTimeTest3() throws Exception { Workflow wf1 = new Workflow(wfID1, sch1, str1, tr1, srv1, maxRetryCount, startTime, Workflow.DEFAULT_WAIT_TIMEOUT_SECONDS, emptyWorkflowInfo); WorkflowConfiguration cfg = new WorkflowConfiguration(); - cfg.addWorkflow(wf1); + cfg.addWorkflow(wf1, ""); MemoryStateDatabase db = new MemoryStateDatabase(); @@ -809,7 +809,7 @@ public void retryTest() throws Exception { Workflow wf1 = new Workflow(wfID1, sch1, str1, tr1, srv1, maxRetryCount, Workflow.DEFAULT_START_TIME, Workflow.DEFAULT_WAIT_TIMEOUT_SECONDS, emptyWorkflowInfo); WorkflowConfiguration cfg = new WorkflowConfiguration(); - cfg.addWorkflow(wf1); + cfg.addWorkflow(wf1, ""); MemoryStateDatabase db = new MemoryStateDatabase(); @@ -863,7 +863,7 @@ public void retryTestWithTooSmallMaxRetryCount() throws Exception { Workflow wf1 = new Workflow(wfID1, sch1, str1, tr1, srv1, maxRetryCount, Workflow.DEFAULT_START_TIME, Workflow.DEFAULT_WAIT_TIMEOUT_SECONDS, emptyWorkflowInfo); WorkflowConfiguration cfg = new WorkflowConfiguration(); - cfg.addWorkflow(wf1); + cfg.addWorkflow(wf1, ""); MemoryStateDatabase db = new MemoryStateDatabase(); @@ -917,8 +917,8 @@ public void schedulingOnlySubsetOfWorkflowsWorks() throws Exception { Workflow wf2 = new Workflow(wfID2, sch1, str1, tr1, srv1, maxRetryCount, Workflow.DEFAULT_START_TIME, Workflow.DEFAULT_WAIT_TIMEOUT_SECONDS, emptyWorkflowInfo); WorkflowConfiguration cfg = new WorkflowConfiguration(); - cfg.addWorkflow(wf1); - cfg.addWorkflow(wf2); + cfg.addWorkflow(wf1, ""); + cfg.addWorkflow(wf2, ""); MemoryStateDatabase db = new MemoryStateDatabase();