Add more meta information & error handling

asyncwrapper/src/main/kotlin/org/n/riesgos/asyncwrapper/datamanagement/DatamanagementRepo.kt

@@ -408,6 +408,19 @@ class DatamanagementRepo (
         jdbcTemplate.update(sql, status, jobId)
     }
 
+    /**
+     * Set the exception report field for the job.
+     */
+    fun setJobExceptionReport (jobId: Long, exceptionReport: String) {
+        val sql = """
+            update jobs
+            set exception_report = ?
+            where id = ?
+        """.trimIndent()
+
+        jdbcTemplate.update(sql, exceptionReport, jobId)
+    }
+
     /**
      * Insert a complex output data in the database.
      */

asyncwrapper/src/main/kotlin/org/n/riesgos/asyncwrapper/dummy/AbstractWrapper.kt

@@ -20,6 +20,8 @@ import org.n.riesgos.asyncwrapper.pulsar.PulsarPublisher
 import org.n.riesgos.asyncwrapper.utils.retry
 import org.n52.geoprocessing.wps.client.WPSClientException
 import java.io.IOException
+import java.io.PrintWriter
+import java.io.StringWriter
 import java.net.URI
 import java.net.http.HttpClient
 import java.net.http.HttpRequest
@@ -312,7 +314,14 @@ abstract class AbstractWrapper(val publisher : PulsarPublisher, val wpsConfigura
             LOGGER.info("WPS call failed")
             LOGGER.info("WPS call failed because of: ${e.message}")
             e.printStackTrace()
+
             datamanagementRepo().updateJobStatus(jobId, WPS_JOB_STATUS_FAILED)
+
+            val sw = StringWriter()
+            val pw = PrintWriter(sw)
+            e.printStackTrace(pw)
+            val exceptionReport = sw.toString()
+            datamanagementRepo().setJobExceptionReport(jobId, exceptionReport)
             // => without the re-raise we would be able to run the loop & don't
             // have to stop when one call has an exception from the WPS itself.
         }

asyncwrapper/src/test/kotlin/org/n/riesgos/asyncwrapper/datamanagement/DatamanagementRepoTest.kt

@@ -2,12 +2,10 @@ package org.n.riesgos.asyncwrapper.datamanagement
 
 import org.json.JSONObject
 import org.junit.jupiter.api.Assertions.*
-import org.junit.jupiter.api.BeforeEach
 import org.junit.jupiter.api.Test
 import org.n.riesgos.asyncwrapper.datamanagement.mapper.*
 import org.n.riesgos.asyncwrapper.datamanagement.models.*
 import org.n.riesgos.asyncwrapper.datamanagement.repos.*
-import org.springframework.jdbc.core.JdbcTemplate
 import org.springframework.test.annotation.DirtiesContext
 import java.util.*
 
@@ -1002,6 +1000,50 @@ class DatamanagementRepoTest {
         assertEquals("failed", jobStatus)
     }
 
+    @Test
+    fun testSetJobExceptionReport() {
+        val template = H2DbFixture().getJdbcTemplate()
+        val repo = DatamanagementRepo(
+                template,
+                LiteralInputRepo(template),
+                ComplexInputRepo(template),
+                ComplexOutputAsInputRepo(template),
+                ComplexInputAsValueRepo(template),
+                BboxInputRepo(template),
+                OrderJobRefRepo(template),
+                ComplexOutputRepo(template),
+                OrderRepo(template),
+                StoredLinkRepo(template)
+        )
+
+        val gfzWpsUrl = "https://rz-vm140.gfz-potsdam.de/wps"
+        val assetmasterProcessIdentifier = "org.n52.gfz.riesgos.algorithm.AssetmasterProcess"
+        val deusProcessIdentifier = "org.n52.gfz.riesgos.algorithm.DeusProcess"
+
+        template.execute(
+                """
+                    insert into processes (id, wps_url, wps_identifier)
+                    values (1, '${gfzWpsUrl}', '${assetmasterProcessIdentifier}')
+                """.trimIndent()
+        )
+        template.execute(
+                """
+                    insert into processes (id, wps_url, wps_identifier)
+                    values (2, '${gfzWpsUrl}', '${deusProcessIdentifier}')
+                """.trimIndent()
+        )
+        template.execute(
+                """
+                    insert into jobs (id, process_id, status)
+                    values (1, 1, 'pending')
+                """.trimIndent()
+        )
+        repo.setJobExceptionReport(1L, "Something bad happened")
+
+        val exceptionReport = template.queryForObject("select exception_report from jobs where id = 1", String::class.javaObjectType)
+        assertEquals("Something bad happened", exceptionReport)
+    }
+
     @Test
     fun testInsertComplexOutput() {
         val template = H2DbFixture().getJdbcTemplate()

backend/Dockerfile

@@ -6,7 +6,7 @@ RUN apk add python3-dev gcc libc-dev g++
 RUN /usr/local/bin/python -m pip install --upgrade pip
 
 
-RUN pip install fastapi uvicorn psycopg2-binary sqlalchemy pytest requests httpx
+RUN pip install fastapi uvicorn psycopg2-binary sqlalchemy pytest requests httpx pytz
 RUN mkdir -p /usr/src/app
 WORKDIR /usr/src/app
 

backend/catalog_app/database.py

@@ -1,7 +1,6 @@
 from .config import Config
 from sqlalchemy import create_engine
-from sqlalchemy.ext.declarative import declarative_base
-from sqlalchemy.orm import sessionmaker
+from sqlalchemy.orm import sessionmaker, declarative_base
 
 config = Config()
 engine = create_engine(config.sqlalchemy_database_url)

backend/catalog_app/models.py

@@ -1,18 +1,31 @@
+import base64
 import binascii
-import os
 import hashlib
-import base64
-from sqlalchemy import JSON, Boolean, Column, Float, ForeignKey, Integer, String, Text
+import os
+
+from sqlalchemy import (
+    JSON,
+    Boolean,
+    Column,
+    DateTime,
+    Float,
+    ForeignKey,
+    Integer,
+    String,
+    Text,
+)
 from sqlalchemy.orm import relationship
 
 from .database import Base
+from .utils import utc_now
 
 
 class Process(Base):
     __tablename__ = "processes"
     id = Column(Integer, primary_key=True)
     wps_url = Column(String(256))
     wps_identifier = Column(String(256))
+    created_at = Column(DateTime(timezone=True), default=utc_now, nullable=False)
     jobs = relationship("Job", back_populates="process")
 
 
@@ -23,6 +36,7 @@ class User(Base):
     password_hash = Column(String(256))
     apikey = Column(String(256))
     superuser = Column(Boolean, default=False)
+    created_at = Column(DateTime(timezone=True), default=utc_now, nullable=False)
     orders = relationship("Order", back_populates="user")
 
     @staticmethod
@@ -59,6 +73,7 @@ class Order(Base):
     user_id = Column(Integer, ForeignKey("users.id"))
     user = relationship("User", back_populates="orders")
     order_constraints = Column(JSON)
+    created_at = Column(DateTime(timezone=True), default=utc_now, nullable=False)
     order_job_refs = relationship("OrderJobRef", back_populates="order")
 
 
@@ -68,6 +83,8 @@ class Job(Base):
     process_id = Column(Integer, ForeignKey("processes.id"))
     process = relationship("Process", back_populates="jobs")
     status = Column(String(16))
+    created_at = Column(DateTime(timezone=True), default=utc_now, nullable=False)
+    exception_report = Column(Text, nullable=True)
 
     order_job_refs = relationship("OrderJobRef", back_populates="job")
     complex_outputs = relationship("ComplexOutput", back_populates="job")
@@ -87,6 +104,7 @@ class OrderJobRef(Base):
     order = relationship("Order", back_populates="order_job_refs")
     job_id = Column(Integer, ForeignKey("jobs.id"))
     job = relationship("Job", back_populates="order_job_refs")
+    created_at = Column(DateTime(timezone=True), default=utc_now, nullable=False)
 
 
 class ComplexOutput(Base):
@@ -99,6 +117,7 @@ class ComplexOutput(Base):
     mime_type = Column(String(64))
     xmlschema = Column(String(256))
     encoding = Column(String(16))
+    created_at = Column(DateTime(timezone=True), default=utc_now, nullable=False)
     inputs = relationship("ComplexOutputAsInput", back_populates="complex_output")
 
 
@@ -110,6 +129,7 @@ class ComplexOutputAsInput(Base):
     wps_identifier = Column(String(256))
     complex_output_id = Column(Integer, ForeignKey("complex_outputs.id"))
     complex_output = relationship("ComplexOutput", back_populates="inputs")
+    created_at = Column(DateTime(timezone=True), default=utc_now, nullable=False)
 
 
 class ComplexInput(Base):
@@ -122,6 +142,7 @@ class ComplexInput(Base):
     mime_type = Column(String(64))
     xmlschema = Column(String(256))
     encoding = Column(String(16))
+    created_at = Column(DateTime(timezone=True), default=utc_now, nullable=False)
 
 
 class ComplexInputAsValue(Base):
@@ -134,6 +155,7 @@ class ComplexInputAsValue(Base):
     mime_type = Column(String(64))
     xmlschema = Column(String(256))
     encoding = Column(String(16))
+    created_at = Column(DateTime(timezone=True), default=utc_now, nullable=False)
 
 
 class LiteralInput(Base):
@@ -143,6 +165,7 @@ class LiteralInput(Base):
     job = relationship("Job", back_populates="literal_inputs")
     wps_identifier = Column(String(256))
     input_value = Column(Text)
+    created_at = Column(DateTime(timezone=True), default=utc_now, nullable=False)
 
 
 class BboxInput(Base):
@@ -156,3 +179,4 @@ class BboxInput(Base):
     upper_corner_x = Column(Float)
     upper_corner_y = Column(Float)
     crs = Column(String(32))
+    created_at = Column(DateTime(timezone=True), default=utc_now, nullable=False)

backend/catalog_app/schemas.py

@@ -1,3 +1,4 @@
+import datetime
 from typing import Optional
 
 from pydantic import BaseModel
@@ -15,6 +16,7 @@ class BboxInputBase(BaseModel):
 
 class BboxInput(BboxInputBase):
     id: int
+    created_at: datetime.datetime
 
     class Config:
         orm_mode = True
@@ -31,6 +33,7 @@ class ComplexInputBase(BaseModel):
 
 class ComplexInput(ComplexInputBase):
     id: int
+    created_at: datetime.datetime
 
     class Config:
         orm_mode = True
@@ -47,6 +50,7 @@ class ComplexInputAsValueBase(BaseModel):
 
 class ComplexInputAsValue(ComplexInputAsValueBase):
     id: int
+    created_at: datetime.datetime
 
     class Config:
         orm_mode = True
@@ -63,6 +67,7 @@ class ComplexOutputBase(BaseModel):
 
 class ComplexOutput(ComplexOutputBase):
     id: int
+    created_at: datetime.datetime
 
     class Config:
         orm_mode = True
@@ -76,6 +81,7 @@ class ComplexOutputAsInputBase(BaseModel):
 
 class ComplexOutputAsInput(ComplexOutputAsInputBase):
     id: int
+    created_at: datetime.datetime
 
     class Config:
         orm_mode = True
@@ -84,10 +90,12 @@ class Config:
 class JobBase(BaseModel):
     process_id: int
     status: str
+    exception_report: Optional[str]
 
 
 class Job(JobBase):
     id: int
+    created_at: datetime.datetime
 
     class Config:
         orm_mode = True
@@ -101,6 +109,7 @@ class LiteralInputBase(BaseModel):
 
 class LiteralInput(LiteralInputBase):
     id: int
+    created_at: datetime.datetime
 
     class Config:
         orm_mode = True
@@ -113,6 +122,7 @@ class OrderBase(BaseModel):
 
 class Order(OrderBase):
     id: int
+    created_at: datetime.datetime
 
     class Config:
         orm_mode = True
@@ -129,6 +139,7 @@ class OrderJobRefBase(BaseModel):
 
 class OrderJobRef(OrderJobRefBase):
     id: int
+    created_at: datetime.datetime
 
     class Config:
         orm_mode = True
@@ -141,6 +152,7 @@ class ProcessBase(BaseModel):
 
 class Process(ProcessBase):
     id: int
+    created_at: datetime.datetime
 
     class Config:
         orm_mode = True

backend/catalog_app/utils.py

@@ -0,0 +1,7 @@
+import datetime
+
+import pytz
+
+
+def utc_now():
+    return pytz.utc.localize(datetime.datetime.now())

backend/migrations/V4__Add_created_at.sql

@@ -0,0 +1,12 @@
+alter table processes add column created_at timestamp with time zone not null default now();
+alter table users add column created_at timestamp with time zone not null default now();
+alter table orders add column created_at timestamp with time zone not null default now();
+alter table jobs add column created_at timestamp with time zone not null default now();
+alter table order_job_refs add column created_at timestamp with time zone not null default now();
+alter table complex_outputs add column created_at timestamp with time zone not null default now();
+alter table complex_outputs_as_inputs add column created_at timestamp with time zone not null default now();
+alter table complex_inputs add column created_at timestamp with time zone not null default now();
+alter table complex_inputs_as_values add column created_at timestamp with time zone not null default now();
+alter table literal_inputs add column created_at timestamp with time zone not null default now();
+alter table bbox_inputs add column created_at timestamp with time zone not null default now();
+alter table stored_links add column created_at timestamp with time zone not null default now();

backend/migrations/V5__Add_exception_column.sql

@@ -0,0 +1 @@
+alter table jobs add column exception_report text;