Add more metadata for pipelines

s-pipes-core/src/main/java/cz/cvut/spipes/constants/SPIPES.java

@@ -43,4 +43,6 @@ protected static Property property(String local)
     public static final Property has_output_content = property("has-output-content");
     public static final Property has_script = property("has-script");
 
+    public static final Property has_executed_function = property("has-executed-function");
+    public static final Property has_executed_function_script_path = property("has-executed-function-script-path");
 }

s-pipes-core/src/main/java/cz/cvut/spipes/engine/ExecutionEngineImpl.java

@@ -26,10 +26,27 @@ public ExecutionContext executePipeline(final Module module, final ExecutionCont
         log.info("Executing script {} with context {}.", module.getResource(), inputContext.toSimpleString());
         final long pipelineExecutionId = Instant.now().toEpochMilli()*1000+(i++);
 
-        fire((l) -> {l.pipelineExecutionStarted(pipelineExecutionId); return null;});
-        ExecutionContext outputContext = _executePipeline(pipelineExecutionId, module, inputContext, null);
-        fire((l) -> {l.pipelineExecutionFinished(pipelineExecutionId); return null;});
-        return outputContext;
+        String functionName = inputContext.getId();
+        String scriptPath;
+        if (inputContext.getValue(ExecutionContext.P_SCRIPT_URI) != null) {
+            scriptPath = inputContext.getScriptFile().toString();
+        } else {
+            scriptPath = module.getScriptPath();
+        }
+        String script = module.getResource().toString().replaceAll("\\/[^.]*$", "");
+        fire((l) -> {l.pipelineExecutionStarted(pipelineExecutionId, functionName, scriptPath, script); return null;});
+        try {
+            ExecutionContext outputContext = _executePipeline(pipelineExecutionId, module, inputContext, null);
+            fire((l) -> {l.pipelineExecutionFinished(pipelineExecutionId); return null;});
+            return outputContext;
+        } catch (Exception e) {
+            log.error("Pipeline execution failed", e);
+            fire((l) -> {
+                l.pipelineExecutionFailed(pipelineExecutionId);
+                return null;
+            });
+            throw e;
+        }
     }
 
     private void fire(final Function<ProgressListener,Void> function) {

s-pipes-core/src/main/java/cz/cvut/spipes/engine/LoggingProgressListener.java

@@ -8,7 +8,7 @@ public class LoggingProgressListener implements ProgressListener {
     private static final Logger LOG = LoggerFactory.getLogger(LoggingProgressListener.class);
 
     @Override
-    public void pipelineExecutionStarted(long pipelineExecutionId) {
+    public void pipelineExecutionStarted(long pipelineExecutionId, final String functionName, final String scriptPath, final String script) {
         LOG.debug("pipelineExecutionStarted - pipelineExecutionId: {}", pipelineExecutionId);
     }
 
@@ -17,6 +17,11 @@ public void pipelineExecutionFinished(long pipelineExecutionId) {
         LOG.debug("pipelineExecutionFinished - pipelineExecutionId: {}", pipelineExecutionId);
     }
 
+    @Override
+    public void pipelineExecutionFailed(long pipelineExecutionId) {
+        LOG.debug("pipelineExecutionFailed - pipelineExecutionId: {}", pipelineExecutionId);
+    }
+
     @Override
     public void moduleExecutionStarted(long pipelineExecutionId, String moduleExecutionId, Module outputModule, ExecutionContext inputContext, String predecessorModuleExecutionId) {
         LOG.debug("moduleExecutionStarted - pipelineExecutionId: {}, moduleExecutionId: {}, inputContext: {}, predecessorModuleExecutionId: {}", pipelineExecutionId,

s-pipes-core/src/main/java/cz/cvut/spipes/engine/ProgressListener.java

@@ -16,7 +16,7 @@ public interface ProgressListener {
      *
      * @param pipelineExecutionId execution id of the pipeline
      */
-    void pipelineExecutionStarted(long pipelineExecutionId);
+    void pipelineExecutionStarted(long pipelineExecutionId, final String functionName, final String scriptPath, final String script);
 
     /**
      * Triggers when execution of a pipeline finishes.
@@ -25,6 +25,8 @@ public interface ProgressListener {
      */
     void pipelineExecutionFinished(long pipelineExecutionId);
 
+    void pipelineExecutionFailed(long pipelineExecutionId);
+
     /**
      * Triggers when execution of a module within a pipeline starts.
      *

s-pipes-core/src/main/java/cz/cvut/spipes/logging/AdvancedLoggingProgressListener.java

@@ -40,6 +40,7 @@
 import java.io.*;
 import java.net.MalformedURLException;
 import java.net.URI;
+import java.net.URISyntaxException;
 import java.nio.file.Files;
 import java.nio.file.Path;
 import java.nio.file.Paths;
@@ -103,7 +104,7 @@ private static Property getParameter(final String name) {
     }
 
     @Override
-    public void pipelineExecutionStarted(final long pipelineExecutionId) {
+    public void pipelineExecutionStarted(final long pipelineExecutionId, String functionName, String scriptPath, String script) {
         PipelineExecution pipelineExecution = new PipelineExecution();
         pipelineExecution.setId(getPipelineExecutionIri(pipelineExecutionId));
         pipelineExecution.setTypes(Collections.singleton(Vocabulary.s_c_transformation));
@@ -113,6 +114,11 @@ public void pipelineExecutionStarted(final long pipelineExecutionId) {
         pipelineExecutionDir.toFile().mkdir();
         logDir.put(pipelineExecutionId, pipelineExecutionDir);
 
+        metadataMap.clear();
+        metadataMap.put(SPIPES.has_executed_function.toString(), URI.create(functionName));
+        metadataMap.put(SPIPES.has_executed_function_script_path.toString(), scriptPath);
+        metadataMap.put(SPIPES.has_script.toString(), URI.create(script));
+
         final EntityManager metadataEM = getMetadataEmf().createEntityManager();
         synchronized (metadataEM) {
             persistPipelineExecutionStarted(metadataEM, pipelineExecutionId, pipelineExecution);
@@ -133,7 +139,8 @@ private void persistPipelineExecutionStarted(final EntityManager em, long pipeli
         Date startDate = new Date();
         addProperty(pipelineExecution, SPIPES.has_pipeline_execution_start_date, startDate);
         addProperty(pipelineExecution, SPIPES.has_pipeline_execution_start_date_unix, startDate.getTime());
-
+        addProperty(pipelineExecution, SPIPES.has_executed_function, getURIFromMetadataMap(SPIPES.has_executed_function));
+        addProperty(pipelineExecution, SPIPES.has_executed_function_script_path, metadataMap.get(SPIPES.has_executed_function_script_path.toString()));
         if (pipelineExecutionGroupId != null) {
             addProperty(pipelineExecution, PIPELINE_EXECUTION_GROUP_ID, pipelineExecutionGroupId);
         }
@@ -182,14 +189,29 @@ private void persistPipelineExecutionFinished(final EntityManager em, final long
             final PipelineExecution pipelineExecution =
                     em.find(PipelineExecution.class, pipelineExecutionIri, pd);
 
-            String pipelineName = metadataMap.get(SPIPES.has_pipeline_name.toString()).toString();
+            addProperty(pipelineExecution, SPIPES.has_script, getURIFromMetadataMap(SPIPES.has_script));
             // new
             Date startDate = pipelineExecution.getHas_pipepline_execution_date();
             addProperty(pipelineExecution, SPIPES.has_pipeline_execution_finish_date, finishDate);
             addProperty(pipelineExecution, SPIPES.has_pipeline_execution_finish_date_unix, finishDate.getTime());
             addProperty(pipelineExecution, SPIPES.has_pipeline_execution_duration, computeDuration(startDate, finishDate));
-            addProperty(pipelineExecution, SPIPES.has_pipeline_name, pipelineName);
-//            addScript(pipelineExecution, scriptManager.getScriptByContextId(pipelineName));
+            addProperty(pipelineExecution, ResourceFactory.createProperty(Vocabulary.s_p_has_pipeline_execution_status), URI.create(Vocabulary.s_p_finished_pipeline_execution));
+//            addScript(pipelineExecution, scriptManager.getScriptByContextId(script));
+            em.getTransaction().commit();
+            em.close();
+        }
+    }
+
+    private void persistPipelineExecutionFailed(final EntityManager em, final long pipelineExecutionId) {
+        if (em.isOpen()) {
+            em.getTransaction().begin();
+
+            String pipelineExecutionIri = getPipelineExecutionIri(pipelineExecutionId);
+            final EntityDescriptor pd = new EntityDescriptor(URI.create(pipelineExecutionIri));
+            final PipelineExecution pipelineExecution =
+                    em.find(PipelineExecution.class, pipelineExecutionIri, pd);
+            addProperty(pipelineExecution, SPIPES.has_script, getURIFromMetadataMap(SPIPES.has_script));
+            addProperty(pipelineExecution, ResourceFactory.createProperty(Vocabulary.s_p_has_pipeline_execution_status), URI.create(Vocabulary.s_p_failed_pipeline_execution));
             em.getTransaction().commit();
             em.close();
         }
@@ -220,6 +242,16 @@ public void pipelineExecutionFinished(final long pipelineExecutionId) {
         }
     }
 
+    public void pipelineExecutionFailed(final long pipelineExecutionId) {
+        final EntityManager em = entityManagerMap.get(getPipelineExecutionIri(pipelineExecutionId));
+
+        synchronized (em) {
+            persistPipelineExecutionFailed(em, pipelineExecutionId);
+            entityManagerMap.remove(em);
+            executionMap.remove(getPipelineExecutionIri(pipelineExecutionId));
+        }
+    }
+
     @Override
     public void moduleExecutionStarted(final long pipelineExecutionId, final String moduleExecutionId,
                                        final Module outputModule,
@@ -314,9 +346,10 @@ public void moduleExecutionFinished(long pipelineExecutionId, final String modul
                 addProperty(moduleExecution, SPIPES.has_module_execution_duration, computeDuration(startDate, finishDate));
                 addProperty(moduleExecution, SPIPES.has_output_model_triple_count, module.getOutputContext().getDefaultModel().size());
                 addContentProperty(moduleExecution, module.getOutputContext(), "output");
-                addProperty(moduleExecution, SPIPES.has_pipeline_name, module.getResource().toString().replaceAll("\\/[^.]*$", ""));
-                if(!metadataMap.containsKey(SPIPES.has_pipeline_name.toString())){
-                    metadataMap.put(SPIPES.has_pipeline_name.toString(), module.getResource().toString().replaceAll("\\/[^.]*$", ""));
+                String script = module.getResource().toString().replaceAll("\\/[^.]*$", "");
+                addProperty(moduleExecution, SPIPES.has_script, URI.create(script));
+                if(!metadataMap.containsKey(SPIPES.has_script.toString())){
+                    metadataMap.put(SPIPES.has_script.toString(), URI.create(script));
                 }
 
                 // input binding
@@ -567,5 +600,16 @@ public String toString() {
         }
     }
 
-
+    private URI getURIFromMetadataMap(Property property){
+        URI uri;
+        try {
+            if (!metadataMap.containsKey(property.toString())) {
+                throw new IllegalStateException("Metadata map does not contain property: " + property);
+            }
+            uri = URI.create((metadataMap.get(property.toString()).toString()));
+            return uri;
+        } catch (NullPointerException e) {
+            throw new IllegalStateException("Metadata map is null or does not contain property: " + property, e);
+        }
+    }
 }

s-pipes-core/src/main/java/cz/cvut/spipes/logging/SemanticLoggingProgressListener.java

@@ -68,7 +68,7 @@ public SemanticLoggingProgressListener() {
     public SemanticLoggingProgressListener(Resource configResource) {
     }
 
-    @Override public void pipelineExecutionStarted(final long pipelineExecutionId) {
+    @Override public void pipelineExecutionStarted(final long pipelineExecutionId, final String functionName, final String scriptPath, final String script) {
         Thing pipelineExecution = new Thing();
         pipelineExecution.setId(getPipelineExecutionIri(pipelineExecutionId));
         pipelineExecution.setTypes(Collections.singleton(Vocabulary.s_c_pipeline_execution));
@@ -115,6 +115,34 @@ public SemanticLoggingProgressListener(Resource configResource) {
         }
     }
 
+    @Override
+    public void pipelineExecutionFailed(final long pipelineExecutionId) {
+        final EntityManager em = entityManagerMap.get(getPipelineExecutionIri(pipelineExecutionId));
+        synchronized (em) {
+            if (em.isOpen()) {
+                final TurtleWriterFactory factory = new TurtleWriterFactory();
+                try (FileOutputStream fos = new FileOutputStream(
+                        Files.createFile(getDir(pipelineExecutionId).resolve("log.ttl")).toFile())) {
+                    final RDFWriter writer = factory.getWriter(fos);
+                    writer.startRDF();
+                    RepositoryConnection con = em.unwrap(SailRepository.class).getConnection();
+                    final ValueFactory f = con.getValueFactory();
+                    final RepositoryResult<Statement> res = con
+                            .getStatements(null, null, null, true, f.createIRI(getPipelineExecutionIri(pipelineExecutionId)));
+                    while (res.hasNext()) {
+                        writer.handleStatement(res.next());
+                    }
+                    writer.endRDF();
+                } catch (IOException e) {
+                    log.error("Error during failed pipeline execution logging.", e);
+                }
+                entityManagerMap.remove(em);
+                em.close();
+                logDir.remove(pipelineExecutionId);
+            }
+        }
+    }
+
     @Override public void moduleExecutionStarted(final long pipelineExecutionId, final String moduleExecutionId,
                                                  final Module outputModule,
                                                  final ExecutionContext inputContext,

s-pipes-core/src/main/java/cz/cvut/spipes/modules/AbstractModule.java

@@ -484,6 +484,26 @@ private List<Resource> sortConstraintQueries(List<Resource> constraintQueries) {
         }).collect(Collectors.toList());
     }
 
+    private String functionName;
+
+    public void setFunctionName(String functionName) {
+        this.functionName = functionName;
+    }
+
+    public String getFunctionName() {
+        return functionName;
+    }
+
+    private String scriptPath;
+
+    public void setScriptPath(String scriptPath) {
+        this.scriptPath = scriptPath;
+    }
+
+    public String getScriptPath() {
+        return scriptPath;
+    }
+
 //    @Override
 //    public String toString() {
 //        String resourceId = (resource  != null) ? ( " (" + resource.getId() + ")") : "";

s-pipes-core/src/main/java/cz/cvut/spipes/modules/Module.java

@@ -54,4 +54,12 @@ public interface Module {
 
     // TODO should not be here !!!! but rather generalized
     void addOutputBindings(VariablesBinding variablesBinding);
+
+    void setFunctionName (String functionName);
+
+    String getFunctionName();
+
+    void setScriptPath (String functionName);
+
+    String getScriptPath();
 }

s-pipes-debug/src/main/java/cz/cvut/spipes/debug/dto/PipelineExecutionDto.java

@@ -1,5 +1,6 @@
 package cz.cvut.spipes.debug.dto;
 
+import java.net.URI;
 import java.util.Date;
 import java.util.List;
 
@@ -26,10 +27,60 @@ public Date getHas_pipepline_execution_date() {
         return has_pipepline_execution_date;
     }
 
+    private Date has_pipeline_execution_finish_date;
+
+    public Date getHas_pipeline_execution_finish_date() {
+        return has_pipeline_execution_finish_date;
+    }
+
+    public void setHas_pipeline_execution_finish_date(Date has_pipeline_execution_finish_date) {
+        this.has_pipeline_execution_finish_date = has_pipeline_execution_finish_date;
+    }
+
     public void setHas_pipepline_execution_date(Date has_pipepline_execution_date) {
         this.has_pipepline_execution_date = has_pipepline_execution_date;
     }
 
+    private URI has_pipeline_execution_status;
+
+    public URI getHas_pipeline_execution_status() {
+        return has_pipeline_execution_status;
+    }
+
+    public void setHas_pipeline_execution_status(URI has_pipeline_execution_status) {
+        this.has_pipeline_execution_status = has_pipeline_execution_status;
+    }
+
+    private URI has_script;
+
+    public URI getHas_script() {
+        return has_script;
+    }
+
+    public void setHas_script(URI has_script) {
+        this.has_script = has_script;
+    }
+
+    private URI has_executed_function;
+
+    public URI getHas_executed_function() {
+        return has_executed_function;
+    }
+
+    public void setHas_executed_function(URI has_executed_function) {
+        this.has_executed_function = has_executed_function;
+    }
+
+    private String has_executed_function_script_path;
+
+    public String getHas_executed_function_script_path() {
+        return has_executed_function_script_path;
+    }
+
+    public void setHas_executed_function_script_path(String has_executed_function_script_path) {
+        this.has_executed_function_script_path = has_executed_function_script_path;
+    }
+
     public List<ModuleExecutionDto> getHas_module_executions() {
         return has_module_executions;
     }

s-pipes-debug/src/main/java/cz/cvut/spipes/debug/mapper/PipelineExecutionMapper.java

@@ -10,5 +10,10 @@
 public interface PipelineExecutionMapper {
 
     @Mapping(source = "has_part", target = "has_module_executions")
+    @Mapping(source = "has_pipeline_execution_finish_date", target = "has_pipeline_execution_finish_date")
+    @Mapping(source = "has_pipeline_execution_status", target = "has_pipeline_execution_status")
+    @Mapping(source = "has_script", target = "has_script")
+    @Mapping(source = "has_executed_function", target = "has_executed_function")
+    @Mapping(source = "has_executed_function_script_path", target = "has_executed_function_script_path")
     PipelineExecutionDto toDto(PipelineExecution pipelineExecution);
 }

s-pipes-model/src/main/java/cz/cvut/spipes/Vocabulary.java

@@ -30,6 +30,9 @@ public class Vocabulary {
     public final static String s_c_description = "http://onto.fel.cvut.cz/ontologies/dataset-descriptor/description";
     public final static String s_c_entity_partition = "http://onto.fel.cvut.cz/ontologies/dataset-descriptor/entity-partition";
     public final static String s_c_execution_context_dataset_source = "http://onto.fel.cvut.cz/ontologies/dataset-descriptor/execution-context-dataset-source";
+    public final static String s_c_failed_pipeline_execution = "http://onto.fel.cvut.cz/ontologies/dataset-descriptor/failed-pipeline-execution";
+    public final static String s_c_finished_pipeline_execution = "http://onto.fel.cvut.cz/ontologies/dataset-descriptor/finished-pipeline-execution";
+    public final static String s_c_has_pipeline_execution_status = "http://onto.fel.cvut.cz/ontologies/dataset-descriptor/has-pipeline-execution-status";
     public final static String s_c_human_interpretation = "http://onto.fel.cvut.cz/ontologies/dataset-descriptor/human-interpretation";
     public final static String s_c_intent = "http://onto.fel.cvut.cz/ontologies/dataset-descriptor/intent";
     public final static String s_c_module_execution = "http://onto.fel.cvut.cz/ontologies/dataset-descriptor/module-execution";
@@ -127,6 +130,9 @@ public class Vocabulary {
     public final static String s_p_has_pipeline_execution_start_date = "http://onto.fel.cvut.cz/ontologies/s-pipes/has-pipeline-execution-start-date";
     public final static String s_p_s_pipes_model_name = "http://onto.fel.cvut.cz/ontologies/s-pipes/name";
     public final static String s_p_value = "http://onto.fel.cvut.cz/ontologies/s-pipes/related-resource/value";
+    public final static String s_p_failed_pipeline_execution = "http://onto.fel.cvut.cz/ontologies/dataset-descriptor/failed-pipeline-execution";
+    public final static String s_p_finished_pipeline_execution = "http://onto.fel.cvut.cz/ontologies/dataset-descriptor/finished-pipeline-execution";
+    public final static String s_p_has_pipeline_execution_status = "http://onto.fel.cvut.cz/ontologies/dataset-descriptor/has-pipeline-execution-status";
     public final static String s_p_label = "http://www.w3.org/2000/01/rdf-schema#label";
 
 }