From 88ec67ed3c80b91813cedc841bb09acc2918b2ac Mon Sep 17 00:00:00 2001 From: Alexander Konopko Date: Tue, 22 Sep 2015 15:04:55 +0300 Subject: [PATCH 1/2] restore workflow-file servlet --- .../com/collective/celos/CelosClient.java | 14 +++++ .../celos/WorkflowConfiguration.java | 15 ++++- .../celos/WorkflowConfigurationParser.java | 8 ++- .../celos/servlet/WorkflowFileServlet.java | 62 +++++++++++++++++++ .../com/collective/celos/celos-scripts.js | 2 +- celos-server/src/main/webapp/WEB-INF/web.xml | 9 +++ .../celos/CelosClientServerTest.java | 27 ++++++++ .../celos/DependentScheduleTest.java | 2 +- .../com/collective/celos/SchedulerTest.java | 28 ++++----- 9 files changed, 145 insertions(+), 22 deletions(-) create mode 100644 celos-server/src/main/java/com/collective/celos/servlet/WorkflowFileServlet.java 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 3c0ec9196..1bcf2a745 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; @@ -46,6 +47,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"; @@ -161,6 +164,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); + } + private void executePost(URI request) throws IOException { HttpPost post = new HttpPost(request); HttpResponse postResponse = execute(post); 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..0dde222d2 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("workflowFilePath", 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..920768879 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, workflowFilePath); } diff --git a/celos-server/src/main/webapp/WEB-INF/web.xml b/celos-server/src/main/webapp/WEB-INF/web.xml index 190704f9a..ae0f00156 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 + + ClearCacheServlet com.collective.celos.servlet.ClearCacheServlet 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 e963b050a..b2e5d881a 100644 --- a/celos-server/src/test/java/com/collective/celos/CelosClientServerTest.java +++ b/celos-server/src/test/java/com/collective/celos/CelosClientServerTest.java @@ -696,4 +696,31 @@ public void testPauseWorkflow() throws Exception { Assert.assertEquals(slotStateLast3.getStatus(), SlotState.Status.RUNNING); Assert.assertFalse(workflowStatus3.isPaused()); } + + @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")); + } + } 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 131bd170a..536cf894e 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(); @@ -804,7 +804,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(); @@ -858,7 +858,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(); @@ -912,8 +912,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(); From c0694f6696329dc1b0fbf01fc2fbfc3228634d56 Mon Sep 17 00:00:00 2001 From: Alexander Konopko Date: Tue, 22 Sep 2015 17:11:31 +0300 Subject: [PATCH 2/2] review --- .../java/com/collective/celos/WorkflowConfigurationParser.java | 2 +- .../src/main/resources/com/collective/celos/celos-scripts.js | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) 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 0dde222d2..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("workflowFilePath", fileName); + jsProperties.put("celosWorkflowFilePath", fileName); jsConfigParser.putPropertiesInScope(jsProperties, scope); InputStream scripts = WorkflowConfigurationParser.class.getResourceAsStream("celos-scripts.js"); 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 920768879..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, workflowFilePath); + celosWorkflowConfigurationParser.addWorkflow(workflow, celosWorkflowFilePath); }