restore workflow-file servlet

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);

celos-server/src/main/java/com/collective/celos/WorkflowConfiguration.java

@@ -25,22 +25,31 @@
 public class WorkflowConfiguration {
 
     private final Map<WorkflowID, Workflow> workflows = new HashMap<>();
+    private final Map<WorkflowID, String> workflowsToFiles = new HashMap<>();
+
+    public WorkflowConfiguration() {
+    }
 
     public Collection<Workflow> 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);
     }
-    
+
 }

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);
     }
 
+
+
 }

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));
+        }
+    }
+
+}

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);
 }
 
 

celos-server/src/main/webapp/WEB-INF/web.xml

@@ -86,6 +86,15 @@
         <url-pattern>/pause</url-pattern>
     </servlet-mapping>
 
+    <servlet>
+        <servlet-name>WorkflowFileServlet</servlet-name>
+        <servlet-class>com.collective.celos.servlet.WorkflowFileServlet</servlet-class>
+    </servlet>
+    <servlet-mapping>
+        <servlet-name>WorkflowFileServlet</servlet-name>
+        <url-pattern>/workflow-file</url-pattern>
+    </servlet-mapping>
+
     <servlet>
         <servlet-name>KillServlet</servlet-name>
         <servlet-class>com.collective.celos.servlet.KillServlet</servlet-class>

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);

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.<ContactsInfo>emptyList())));
+                        new WorkflowInfo(null, Collections.<ContactsInfo>emptyList())), "");
 
         DependentSchedule schedule = new DependentSchedule(new WorkflowID(id));
         SortedSet<ScheduledTime> times = schedule.getScheduledTimes(scheduler, schedulingStartTime, schedulingEndTime);

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();