SCMOD-8730: Support expiry policies

docs/pages/en-us/Architecture.md

@@ -150,7 +150,7 @@ This table stores Ids of dependent jobs i.e. jobs which must be completed before
 
 ### Job Task Data
 
-This table stores information on jobs which have dependent jobs and must wait for execution.  The table contains enough information for the Job Tracking Worker and Job Service Scheduled Executor to forward on the job, once it's dependent jobs have all completed.
+This table stores information on jobs which have dependent jobs and must wait for execution.  The table contains enough information for the Job Tracking Worker and Job Service Scheduled Executor to forward on the job, once its dependent jobs have all completed.
 
 | **Column**           | **Data Type** | **Nullable?** | **Primary Key?** |
 |----------------------|---------------|---------------|------------------|
@@ -163,3 +163,38 @@ This table stores information on jobs which have dependent jobs and must wait fo
 | target_pipe          | varchar(255)  | No            |                  |
 | eligible_to_run_date | timestamp     | Yes           |                  |
 
+### Job Expiration Policy
+
+This table stores information about the expiration policy related to the jobs. The Job Service Scheduled Executor runs 
+at regular interval in order to apply those policies.
+
+| **Column**           | **Data Type**       | **Nullable?** | **Primary Key?** |
+|----------------------|---------------------|---------------|------------------|
+| partition_id         | varchar(40)         | No            | Yes              |
+| job_id               | varchar(48)         | No            | Yes              |
+| job_status           | job_status          | No            | Yes              |
+| operation            | expiration_operation| No            |                  |
+| expiration_time      | varchar(58)         | No            |                  |
+| exact_expiry_time    | timestamp           | Yes           |                  |
+| last_modified_offset | interval            | Yes           |                  |
+
+### Default Job Expiration Policy
+
+This table stores the default expiration_policy related to the jobs. Its role is to provide a policy whenever missing 
+from the job_expiration_policy table
+
+| **Column**           | **Data Type**       | **Nullable?** | **Primary Key?** |
+|----------------------|---------------------|---------------|------------------|
+| job_status           | job_status          | No            | Yes              |
+| operation            | expiration_operation| No            |                  |
+| expiration_time      | varchar(58)         | No            |                  |
+| create_date_offset   | varchar(12)         | Yes           |                  |
+| last_modified_offset | interval            | Yes           |                  |
+
+### Enumeration Types
+
+| **Enumerated Type**  | **Enum labels**                                                |
+|----------------------|----------------------------------------------------------------|
+| expiration_operation | Expire, Delete                                                 |
+| job_status           | Active, Cancelled, Completed, Expired, Failed, Paused, Waiting |
+| expiration_policer   | User, System                                                   |

job-service-acceptance-tests/pom.xml

@@ -184,6 +184,7 @@
                         <datastore.enabled>true</datastore.enabled>
                     </systemPropertyVariables>
                     <environmentVariables>
+                        <CAF_LOG_LEVEL>DEBUG</CAF_LOG_LEVEL>
                         <CAF_APPNAME>test/jobservice-acceptance</CAF_APPNAME>
                         <CAF_CONFIG_PATH>${project.basedir}/test-configs</CAF_CONFIG_PATH>
                         <CAF_JOB_SERVICE_IMAGE>${dockerJobServiceOrg}job-service${dockerProjectVersion}</CAF_JOB_SERVICE_IMAGE>

job-service-contract/src/main/resources/com/hpe/caf/services/job/swagger.yaml

@@ -71,7 +71,7 @@ paths:
         in: query
         type: string
         required: false
-        description: All - no status filter is applied (Default); NotCompleted - only those results with statuses other than Completed will be returned; Completed - only those results with Completed status will be returned; Inactive - only those results with inactive statuses (i.e. Completed, Failed, Cancelled) will be returned; NotFinished - only those results with unfinished statuses (ie. Active, Paused, Waiting) will be returned.
+        description: All - no status filter is applied (Default); NotCompleted - only those results with statuses other than Completed will be returned; Completed - only those results with Completed status will be returned; Inactive - only those results with inactive statuses (i.e. Completed, Failed, Cancelled, Expired) will be returned; NotFinished - only those results with unfinished statuses (ie. Active, Paused, Waiting) will be returned.
         enum:
           - All
           - NotCompleted
@@ -121,6 +121,21 @@ paths:
               $ref: "#/definitions/job"
         503:
           description: The request failed due to a database connection error.
+  /partitions/default-expiry:
+    get:
+      tags:
+        - Jobs
+      summary: Gets the default expiry
+      description: Returns the default expiry  settings.
+      operationId: getDefaultExpiry
+      responses:
+        200:
+          description: Returns the default expiry settings.
+          schema:
+            schema:
+              $ref: "#/definitions/expirationPolicy"
+        503:
+          description: The request failed due to a database connection error.
   /partitions/{partitionId}/jobs/{jobId}:
     parameters:
       - $ref: '#/parameters/partitionId'
@@ -210,8 +225,8 @@ paths:
             type: boolean
           headers:
             CacheableJobStatus:
-                description: A response that includes this header with the value ‘true’ is eligible for caching in a cache specific to job status responses.
-                type: boolean
+              description: A response that includes this header with the value ‘true’ is eligible for caching in a cache specific to job status responses.
+              type: boolean
         400:
           description: The `jobId` parameter contains invalid characters.
         503:
@@ -239,8 +254,8 @@ paths:
             $ref: "#/definitions/job-status"
           headers:
             CacheableJobStatus:
-                description: A response that includes this header with the value ‘true’ is eligible for caching in a cache specific to job status responses.
-                type: boolean
+              description: A response that includes this header with the value ‘true’ is eligible for caching in a cache specific to job status responses.
+              type: boolean
         400:
           description: The `jobId` parameter contains invalid characters.
         404:
@@ -377,6 +392,8 @@ definitions:
       description:
         type: string
         description: The description of the job
+      expiry:
+        $ref: "#/definitions/expirationPolicy"
       externalData:
         # TODO: Should we make this more structured than just string
         #       Name/Value Pairs perhaps
@@ -402,7 +419,7 @@ definitions:
       delay:
         type: integer
         format: int32
-        default: 0		
+        default: 0
         description: The time in seconds after the prerequisite job identifiers have completed before this job is eligible for running.
       labels:
         type: object
@@ -414,6 +431,66 @@ definitions:
         example:
           "tag:4": "4"
           owner: "bob"
+  expirationPolicy:
+    type: object
+    properties:
+      "Active":
+        $ref: "#/definitions/expirablePolicy"
+      "Completed":
+        $ref: "#/definitions/deletePolicy"
+      "Failed":
+        $ref: "#/definitions/deletePolicy"
+      "Cancelled":
+        $ref: "#/definitions/deletePolicy"
+      "Waiting":
+        $ref: "#/definitions/expirablePolicy"
+      "Paused":
+        $ref: "#/definitions/expirablePolicy"
+      "Expired":
+        $ref: "#/definitions/deletePolicy"
+    description: The expiration policy to be applied on the job
+  deletePolicy:
+    type: object
+    properties:
+      expiryTime:
+        type: string
+        pattern: '^(?<fullyear>\d{4})-(?<month>0[1-9]|1[0-2])-(?<mday>0[1-9]|[12][0-9]|3[01])T(?<hour>[01][0-9]|2[0-3]):(?<minute>[0-5][0-9]):(?<second>[0-5][0-9]|60)(?<secfrac>\.[0-9]+)?(Z|(\+|-)(?<offset_hour>[01][0-9]|2[0-3]):(?<offset_minute>[0-5][0-9]))$|^(lastUpdateTime|createTime)\+((P)\d+[DYM])$|^(lastUpdateTime|createTime)\+(PT)?(\d+[HMS])$|^none$'
+        example: 2021-04-12T23:20:50.52Z, lastUpdateTime+P1D, createTime+P90M, createTime+PT9H, none
+        description: The delay before expiration
+      operation:
+        type: string
+        enum:
+          - Delete
+        default: Delete
+        description: The action to apply on expired jobs
+      policer:
+        $ref: "#/definitions/policer"
+    description: The expiration details to be applied on the job
+  expirablePolicy:
+    type: object
+    properties:
+      expiryTime:
+        type: string
+        pattern: '^(?<fullyear>\d{4})-(?<month>0[1-9]|1[0-2])-(?<mday>0[1-9]|[12][0-9]|3[01])T(?<hour>[01][0-9]|2[0-3]):(?<minute>[0-5][0-9]):(?<second>[0-5][0-9]|60)(?<secfrac>\.[0-9]+)?(Z|(\+|-)(?<offset_hour>[01][0-9]|2[0-3]):(?<offset_minute>[0-5][0-9]))$|^(lastUpdateTime|createTime)\+((P)\d+[DYM])$|^(lastUpdateTime|createTime)\+(PT)?(\d+[HMS])$|^none$'
+        example: 2021-04-12T23:20:50.52Z, lastUpdateTime+P1D, createTime+P90M, createTime+PT9H, none
+        description: The delay before expiration
+      operation:
+        type: string
+        enum:
+          - Expire
+          - Delete
+        default: Expire
+        description: The action to apply on expired jobs
+      policer:
+        $ref: "#/definitions/policer"
+    description: The expiration details to be applied on the job
+  policer:
+    type: string
+    enum:
+      - User
+      - System
+    default: User
+    description: The instance defining the policy
   worker-action:
     type: object
     required:
@@ -451,6 +528,8 @@ definitions:
       description:
         type: string
         description: The description of the job
+      expiry:
+        $ref: "#/definitions/expirationPolicy"
       externalData:
         type: string
         description: |
@@ -492,6 +571,7 @@ definitions:
       - Failed
       - Paused
       - Waiting
+      - Expired
   failure:
     type: object
     properties:

job-service-db/src/main/resources/procedures/applyJobExpirationPolicy.sql

@@ -0,0 +1,99 @@
+--
+-- Copyright 2016-2021 Micro Focus or one of its affiliates.
+--
+-- Licensed under the Apache License, Version 2.0 (the "License");
+-- you may not use this file except in compliance with the License.
+-- You may obtain a copy of the License at
+--
+--      http://www.apache.org/licenses/LICENSE-2.0
+--
+-- Unless required by applicable law or agreed to in writing, software
+-- distributed under the License is distributed on an "AS IS" BASIS,
+-- WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+-- See the License for the specific language governing permissions and
+-- limitations under the License.
+--
+
+/*
+ *  Name: apply_job_expiration_policy
+ *
+ *  Description:
+ *  Expires or deletes jobs in accordance with their expiry policies.
+ *  Policies can be found in 2 tables:
+ *      - job_expiration_policy if the user provided it on job's creation ( alias user_p)
+ *      - default_job_expiration_policy ( alias default_p)
+ *  Each of those tables have two columns, respectively related, to create_date and last_update_date such as:
+ *      - default_job_expiration_policy -> create_date_offset(varchar), last_modified_offset(interval)
+ *      - job_expiration_policy         -> exact_expiry_time(timestamp), last_modified_offset(interval)
+ *  Only 1 of those fields maximum, by table, can be filled (the other one should be null)
+ *
+ *  User's policy is used in priority if provided, default is applied otherwise
+ *
+ */
+CREATE OR REPLACE PROCEDURE apply_job_expiration_policy(
+)
+LANGUAGE plpgsql
+AS $$
+BEGIN
+
+    PERFORM NULL
+    FROM (
+             SELECT partition_id,
+                    job_id,
+                    COALESCE(user_p.operation, default_p.operation) AS operation
+             FROM job j
+             CROSS JOIN LATERAL (
+                 -- Getting the latest status
+                 SELECT status AS current_status
+                 FROM
+                     get_job(
+                             partition_id,
+                             job_id
+                         )
+                 LIMIT 1
+                 ) latest_status
+             CROSS JOIN default_job_expiration_policy default_p      -- default expiration policy table
+             LEFT JOIN job_expiration_policy user_p             -- user defined expiration policy table
+                 USING (
+                 partition_id,
+                 job_id,
+                 job_status
+                 )
+             WHERE job_status = status
+               -- Order:
+               --   user_p.exact_expiry_time (1)
+               --   user_p.last_modified_offset (2)
+               --   default_p.last_modified_offset (3)
+               --   default_p.create_date_offset (4)
+               AND COALESCE(
+                           COALESCE(
+                                   user_p.exact_expiry_time, ---------------------------------------(1)
+                                   last_update_date + COALESCE(
+                                           user_p.last_modified_offset, ----------------------------(2)
+                                           CASE
+                                               WHEN
+                                                   default_p.last_modified_offset IS NOT NULL
+                                                   THEN
+                                                   default_p.last_modified_offset::INTERVAL --------(3)
+                                               END
+                                       )
+                               )
+                       ,
+                           CASE --------------------------------------------------------------------(4)
+                               WHEN default_p.create_date_offset = 'infinity'
+                                   THEN 'infinity'::TIMESTAMP
+                               WHEN default_p.create_date_offset IS NULL
+                                   THEN NULL
+                               ELSE
+                                   (create_date + default_p.create_date_offset::INTERVAL)::TIMESTAMP
+                               END
+               ) <= now() -- compare to now to get only the expired jobs
+         ) expired_jobs
+         CROSS JOIN LATERAL
+            delete_or_expire_job(
+                    partition_id,
+                    job_id,
+                    operation
+            );
+END;
+$$;

job-service-db/src/main/resources/procedures/cancelJob.sql

@@ -43,7 +43,7 @@ BEGIN
 
     -- Only support Cancel operation on jobs with current status 'Waiting', 'Active' or 'Paused'
     -- And take out an exclusive update lock on the job row
-    SELECT status IN ('Completed', 'Failed') INTO v_is_finished
+    SELECT status IN ('Expired', 'Completed', 'Failed') INTO v_is_finished
     FROM job
     WHERE partition_id = in_partition_id
         AND job_id = in_job_id

job-service-db/src/main/resources/procedures/changelog.xml

@@ -27,6 +27,12 @@
           When adding a new procedure add a corresponding changeSet here with `runOnChange="true"` set.
           If modifying an existing procedure, DO NOT add a new changeSet. The changes will be applied automatically.
      -->
+    <changeSet id="create_apply_job_expiration_policy_proc" runOnChange="true" author="Gamma Team">
+        <createProcedure path="procedures/applyJobExpirationPolicy.sql"
+                         procedureName="apply_job_expiration_policy"
+                         schemaName="public">
+        </createProcedure>
+    </changeSet>
     <changeSet id="create_cancel_job_proc" runOnChange="true" author="Gamma Team">
         <createProcedure path="procedures/cancelJob.sql"
                          procedureName="cancel_job"
@@ -69,6 +75,12 @@
                          schemaName="public">
         </createProcedure>
     </changeSet>
+    <changeSet id="create_delete_or_expire_job_proc" runOnChange="true" author="Gamma Team">
+        <createProcedure path="procedures/deleteOrExpireJob.sql"
+                         procedureName="delete_or_expire_job"
+                         schemaName="public">
+        </createProcedure>
+    </changeSet>
     <changeSet id="create_delete_task_table_proc" runOnChange="true" author="Gamma Team">
         <createProcedure path="procedures/deleteTaskTable.sql"
                          procedureName="internal_delete_task_table"
@@ -91,12 +103,24 @@
                          schemaName="public">
         </createProcedure>
     </changeSet>
+    <changeSet id="create_expire_job_proc" runOnChange="true" author="Gamma Team">
+        <createProcedure path="procedures/expireJob.sql"
+                         procedureName="internal_expire_job"
+                         schemaName="public">
+        </createProcedure>
+    </changeSet>
     <changeSet id="create_get_dependent_jobs_proc" runOnChange="true" author="Gamma Team">
         <createProcedure path="procedures/getDependentJobs.sql"
                          procedureName="get_dependent_jobs"
                          schemaName="public">
         </createProcedure>
     </changeSet>
+    <changeSet id="create_get_default_expiry_proc" runOnChange="true" author="Gamma Team">
+        <createProcedure path="procedures/getDefaultSystemExpiry.sql"
+                         procedureName="get_default_expiry"
+                         schemaName="public">
+        </createProcedure>
+    </changeSet>
     <changeSet id="create_get_job_proc" runOnChange="true" author="Gamma Team">
         <createProcedure path="procedures/getJob.sql"
                          procedureName="get_job"
@@ -295,6 +319,12 @@
                          schemaName="public">
         </createProcedure>
     </changeSet>
+    <changeSet id="create_upsert_job_policy_proc" runOnChange="true" author="Gamma Team">
+        <createProcedure path="procedures/upsertJobPolicy.sql"
+                         procedureName="internal_upsert_job_policy"
+                         schemaName="public">
+        </createProcedure>
+    </changeSet>
     <changeSet id="drop_deleted_task_tables_proc" runOnChange="true" author="Gamma Team">
         <createProcedure path="procedures/dropDeletedTaskTables.sql"
                          procedureName="drop_deleted_task_tables"
@@ -307,4 +337,13 @@
                          schemaName="public">
         </createProcedure>
     </changeSet>
+    <changeSet id="refresh_default_expiry_proc" runOnChange="true" author="Gamma Team">
+        <createProcedure path="procedures/refreshDefaultExpiry.sql"
+                         procedureName="refresh_default_expiry"
+                         schemaName="public">
+        </createProcedure>
+    </changeSet>
+    <changeSet id="run_refresh_default_expiry_proc" runOnChange="true" author="Gamma Team">
+        <sql>CALL refresh_default_expiry();</sql>
+    </changeSet>
 </databaseChangeLog>