pgtask.sql
application/octet-stream
Filename: pgtask.sql
Type: application/octet-stream
Part: 1
CREATE OR REPLACE FUNCTION InitializeServer() RETURNS VOID AS
$$
BEGIN
/* the pgtask server requries postgres 11 */
IF (SELECT setting::INT FROM pg_settings WHERE name = 'server_version_num')
< 110000
THEN
RAISE EXCEPTION 'pg_task server requires postgres 11 or greater';
END IF;
/* the pgtask server requries dblink */
CREATE EXTENSION IF NOT EXISTS dblink;
/* generate a sting that identifies the pgtask server. That way, when we nuke
* tasks, we make sure to only target tasks runnign from *this* pgtask
* instance. The server identification is generated from pgtask databasebase
* name, server ip, and port. (if those values change the id is not updated
* however).
*/
CREATE OR REPLACE FUNCTION ServerIdPhrase() RETURNS TEXT AS
$sip$
SELECT format('%s_%s_%s',
current_database(),
inet_server_addr(),
(SELECT setting FROM pg_settings WHERE Name = 'port'));
$sip$ LANGUAGE SQL;
CREATE TABLE GlobalControl
(
Version TEXT,
MainLoopSleepSecondsIdle NUMERIC DEFAULT 5.0,
MainLoopSleepSecondsActive NUMERIC DEFAULT 0.1,
Debug BOOL DEFAULT false,
DebugQuery TEXT DEFAULT 'SELECT 0',
MainProcessPid INT,
GracefulShutdown BOOL,
TaskMinMessages TEXT DEFAULT 'WARNING',
HasPLSH BOOL DEFAULT FALSE,
ServerId TEXT DEFAULT md5(ServerIdPhrase()),
StartTime TIMESTAMPTZ,
HasTDSFDW BOOL DEFAULT FALSE,
LogMinMessages TEXT DEFAULT 'NOTICE',
LogErrorVerbosity TEXT DEFAULT 'TERSE',
IdleTaskTimeoutInterval INTERVAL DEFAULT '1 minute'::INTERVAL,
IdleTaskTimeoutLastCheck TIMESTAMPTZ,
AuditInterval INTERVAL DEFAULT '5 minutes'::INTERVAL,
LastAudit TIMESTAMPTZ
);
CREATE INDEX ON GlobalControl((1));
INSERT INTO GlobalControl VALUES ('1.0');
DO $q$
BEGIN
PERFORM 1 FROM pg_extension WHERE extname = 'plsh';
IF FOUND
THEN
UPDATE GlobalControl SET HasPLSH = true;
ELSE
CREATE EXTENSION plsh;
END IF;
EXCEPTION WHEN OTHERS THEN
UPDATE GlobalControl SET HasPLSH = false;
END;
$q$;
DO $q$
BEGIN
PERFORM 1 FROM pg_extension WHERE extname = 'tds_fdw';
IF FOUND
THEN
UPDATE GlobalControl SET HasTDSFDW = true;
ELSE
CREATE EXTENSION HasTDSFDW;
END IF;
EXCEPTION WHEN OTHERS THEN
UPDATE GlobalControl SET HasTDSFDW = false;
END;
$q$;
CREATE TABLE Environment
(
Environment TEXT PRIMARY KEY DEFAULT 'Production'
);
CREATE TYPE ServerType_t AS ENUM(
'POSTGRES',
'SQL SERVER',
'WEB SERVICE');
CREATE TABLE Server
(
Server TEXT PRIMARY KEY,
TaskClientThreads INT,
ServerType ServerType_t NOT NULL DEFAULT 'POSTGRES'
);
CREATE TABLE Database
(
Environment TEXT DEFAULT 'Production' REFERENCES Environment
ON UPDATE CASCADE ON DELETE CASCADE,
/* usually the name of the database, doesn't have to be */
Database TEXT,
/* If Database is distributed, each database must have a unique nodeid */
NodeId INT DEFAULT 0,
Server TEXT REFERENCES Server ON DELETE CASCADE ON UPDATE CASCADE,
ConnectionString TEXT UNIQUE,
LogMinMessages TEXT DEFAULT 'NOTICE',
LogErrorVerbosity TEXT DEFAULT 'TERSE',
PRIMARY KEY(Environment, Database, NodeId)
);
CREATE TYPE ConcurrencyModel_t AS ENUM(
/* Task chain may not be created while another is active. */
'BLOCK',
/* Task chain may be created, but the second and later task chains
* will be held so that only one is active at a time.
*/
'QUEUE',
/* Task chain may be created if and only if a task chain is not already
* queued.
*/
'QUEUE_ONCE',
/* The task chain, when created, will merge the created DAG with an
* in progress dag if when is active.
*/
'MERGE',
/* The task chain can always be run regardless of what else is runing */
'NOBLOCK');
CREATE TABLE TaskChainConfiguration
(
TaskChainConfigurationName TEXT PRIMARY KEY,
/* Controls the behavior when a task chain is created while another task
* chain is active with the same name in the same environment. Unnamed
* task chains work as 'QUEUE' with the special dispensation that if
* the task chain does not share any task with another active task
* chains that are pending or in progress, it can be made to run
* run immediately.
*/
DefaultConcurrencyModel ConcurrencyModel_t NOT NULL DEFAULT 'BLOCK'
);
/*
* Defines specific tasks.
*/
CREATE TABLE Task
(
Task TEXT PRIMARY KEY,
/* TaskBody can be null, in which case the task is a NOP. This task is then
* only in place for depndency purposes or as a container for subtasks
*/
TaskBody TEXT,
RunSchedule INT[],
Database TEXT,
StartTime TIME,
EndTime TIME,
MaxRunTime INTERVAL,
IsConcurrent BOOLEAN,
/* Tasks can be a place holder for a task chain. In that case the TaskBody
* would be the name of the task chain. During conversion to
* TaskChainTask, the structure is flattened out to the task level.
*/
IsTaskChainConfiguration BOOL DEFAULT FALSE,
/* gives rough visual ordering */
TaskOrder INT DEFAULT 0,
/* If the task has subtasks, does one subtask failure cause a failure of
* the task? If there are no subtasks, value is moot.
*/
SubTasksMustComplete BOOL DEFAULT TRUE,
LogMinMessages TEXT DEFAULT 'NOTICE',
LogErrorVerbosity TEXT DEFAULT 'TERSE'
);
CREATE INDEX ON Task(Database);
CREATE TABLE TaskChainConfigurationTask
(
TaskChainConfigurationName TEXT REFERENCES TaskChainConfiguration
ON UPDATE CASCADE ON DELETE CASCADE,
Task TEXT REFERENCES Task ON UPDATE CASCADE ON DELETE CASCADE,
PRIMARY KEY (TaskChainConfigurationName, Task)
);
/*
* Subtasks have no dependency except on the parent task. They are all
* activated at the same time.
*/
CREATE TABLE SubTask
(
Task TEXT REFERENCES Task ON DELETE CASCADE,
SubTask TEXT,
TaskBody TEXT,
RunSchedule INT[],
PRIMARY KEY(Task, SubTask)
);
/*
* Holds Dependency details of one tasks on another
*/
CREATE TABLE TaskDependency
(
ChildTask TEXT REFERENCES Task ON UPDATE CASCADE ON DELETE CASCADE,
ParentTask TEXT REFERENCES Task ON UPDATE CASCADE ON DELETE CASCADE,
AllNodes BOOL,
CanRun BOOL,
PRIMARY KEY(ChildTask, ParentTask)
);
/* tracks a chain of tasks that are run */
CREATE TABLE TaskChain
(
TaskChainId BIGSERIAL PRIMARY KEY,
Environment TEXT REFERENCES Environment
ON UPDATE CASCADE ON DELETE CASCADE,
/* set to a named configuration if provided */
TaskChainConfigurationName TEXT,
Entered TIMESTAMPTZ DEFAULT now(),
Started TIMESTAMPTZ DEFAULT now(),
Finished TIMESTAMPTZ,
Paused BOOL DEFAULT false,
PauseStartTime TIMESTAMPTZ,
PauseEndTime TIMESTAMPTZ,
DidSucceed BOOL,
ErrorMessage TEXT,
PendingAction TEXT CHECK (PendingAction IN ('PAUSE', 'CANCEL'))
);
CREATE INDEX ON TaskChain(Environment) WHERE Finished IS NULL;
CREATE TYPE TaskExecutionState_t AS ENUM
(
/* task is still running */
'RUNNING',
/* task has completed, or has failed. */
'FINISHED',
/* the remote server connect reset */
'RESET',
/* task failed to execute */
'FAILED'
);
CREATE TYPE TaskResult_t AS ENUM(
'SUCCESS_COMPLETION',
'SUCCESS_EMPTY',
'SUCCESS_VIA_SUBTASKS',
'FAILED_VIA_SUBTASKS',
'FAILED_VIA_DEPENDENCY',
'FAILED_VIA_PARENT_TASK',
'FAILED_VIA_RESTART',
'FAILED_VIA_CANCEL',
'FAILED_VIA_EXECUTION',
'FAILED_VIA_TIMEOUT',
'FAILED_VIA_CONNECTION_RESET');
/* tracks a specific instance of a task in a chain of tasks */
CREATE TABLE TaskChainTask
(
TaskChainTaskId BIGSERIAL UNIQUE,
TaskChainId BIGINT REFERENCES TaskChain ON DELETE CASCADE,
Environment TEXT,
Database TEXT,
Task TEXT,
SubTask TEXT,
NodeId INT DEFAULT 0,
Eligible BOOL NOT NULL,
Started TIMESTAMPTZ,
Finished TIMESTAMPTZ,
TaskResult TaskResult_t,
ErrorMessage TEXT,
TaskBody TEXT,
Yielded TIMESTAMPTZ,
PRIMARY KEY(TaskChainId, Environment, Database, NodeId, Task, SubTask),
FOREIGN KEY(Environment, Database, NodeId)
REFERENCES Database ON DELETE CASCADE
);
CREATE INDEX ON TaskChainTask(Environment, Database, NodeId);
/* optimizes the question, what tasks are eligible to be run? */
CREATE INDEX ON TaskChainTask(
Environment,
Database,
NodeId,
TaskChainTaskId)
WHERE
Eligible
AND Started IS NULL
AND Finished IS NULL;
/* optimizes the question, what tasks are running? */
CREATE INDEX ON TaskChainTask(TaskChainTaskId)
WHERE
Started IS NOT NULL
AND Finished IS NULL;
CREATE OR REPLACE FUNCTION IsSubTask(_SubTask TEXT) RETURNS BOOL AS
$q$
SELECT _SubTask != '';
$q$ LANGUAGE SQL IMMUTABLE;
/* Tracks dblink connections to various databases.
* The table is unlogged as it's meanliss across server restarts.
*/
CREATE UNLOGGED TABLE ServerConnection
(
Server TEXT,
ConnectionNumber INT,
Environment TEXT,
Database TEXT,
NodeId INT,
ConnectionName TEXT,
Active BOOL,
TaskChainTaskId BIGINT,
Cancelling BOOL,
PRIMARY KEY(Server, ConnectionNumber),
UNIQUE(ConnectionName),
UNIQUE(Environment, Database, NodeId, ConnectionNumber)
);
CREATE OR REPLACE VIEW vw_Server AS
SELECT
s.*,
count(ConnectionName) FILTER (WHERE Active) AS ActiveConnections
FROM Server s
LEFT JOIN ServerConnection USING(Server)
GROUP BY s.Server;
CREATE OR REPLACE VIEW vw_DatabaseConnection AS
SELECT *
FROM Database
JOIN ServerConnection USING(Server, Environment, Database, NodeId)
JOIN Server USING(Server);
CREATE OR REPLACE VIEW vw_ActiveTasks AS
SELECT
Server,
sc.Environment,
sc.Database,
sc.NodeId,
TaskChainTaskId,
tct.Task,
tct.SubTask,
COALESCE(t.TaskBody, st.TaskBody) AS TaskBody,
ConnectionNumber
FROM ServerConnection sc
JOIN TaskChainTask tct USING(TaskChainTaskId)
LEFT JOIN Task t ON
t.Task = tct.Task AND tct.SubTask = ''
LEFT JOIN SubTask st ON
st.Task = tct.Task
AND st.SubTask = tct.SubTask
AND tct.SubTask != ''
WHERE
Active
AND COALESCE(t.TaskBody, st.TaskBody) != '';
CREATE TABLE ActiveTaskAudit AS
SELECT
now() AS AuditTime,
vat.*
FROM vw_ActiveTasks vat
LIMIT 0;
CREATE TABLE MasterLog
(
MasterLogId BIGSERIAL PRIMARY KEY,
Happened TIMESTAMPTZ DEFAULT clock_timestamp(),
Level TEXT,
Log TEXT
);
CREATE TABLE AdministrativeCommand
(
Entered TIMESTAMPTZ DEFAULT NOW(),
Consumed TIMESTAMPTZ,
Query TEXT,
Result JSON,
ErrorMessage TEXT
);
CREATE UNIQUE INDEX ON AdministrativeCommand((1));
END;
$$ LANGUAGE PLPGSQL;
/* initialize the schema if needed, other stuff always. */
DO
$$
BEGIN
PERFORM 1 FROM GlobalControl;
EXCEPTION WHEN OTHERS THEN
PERFORM InitializeServer();
END;
$$;
CREATE OR REPLACE FUNCTION DidSucceed(TaskResult_t) RETURNS BOOL AS
$$
SELECT $1
IN ('SUCCESS_COMPLETION', 'SUCCESS_EMPTY', 'SUCCESS_VIA_SUBTASKS');
$$ LANGUAGE SQL IMMUTABLE;
CREATE OR REPLACE FUNCTION ShouldCancel(TaskResult_t) RETURNS BOOL AS
$$
SELECT $1
IN ('FAILED_VIA_TIMEOUT', 'FAILED_VIA_CANCEL');
$$ LANGUAGE SQL IMMUTABLE;
/* When a task connection is no longer determined to be blocking, it's time to
* resolve it and capture the error message (if any).
*/
CREATE OR REPLACE FUNCTION ReapTask(
_ConnectionName TEXT,
_WantError BOOL DEFAULT true,
TaskExecutionState OUT TaskExecutionState_t,
ErrorMessage OUT TEXT) RETURNS RECORD AS
$$
DECLARE
_Error TEXT;
BEGIN
/* all errors have to be trapped due to not being able to handle thrown
* canclled queries. We can check the error message to see if the query
* succeeded or not.
*/
PERFORM dblink_get_result(_ConnectionName, FALSE);
_Error := dblink_error_message(_ConnectionName);
IF _Error = 'OK'
THEN
TaskExecutionState := 'FINISHED';
ELSE
TaskExecutionState := 'FAILED';
END IF;
ErrorMessage := _Error;
/* dblink requires an extra call to reset connection state. This can also
* fail if the connection is dead.
*/
PERFORM dblink_get_result(_ConnectionName, FALSE);
_Error := dblink_error_message(_ConnectionName);
IF _Error LIKE '%server closed the connection%'
THEN
TaskExecutionState := 'RESET';
ErrorMessage := _Error;
END IF;
/* If there is an error on the task, disconnect the connection as a
* precaution
*/
IF TaskExecutionState = 'FAILED'
THEN
PERFORM Disconnect(_ConnectionName);
END IF;
END;
$$ LANGUAGE PLPGSQL;
CREATE OR REPLACE FUNCTION CancelTask(_ConnectionName TEXT) RETURNS VOID AS
$$
DECLARE
s vw_DatabaseConnection;
BEGIN
SELECT INTO s *
FROM vw_DatabaseConnection WHERE ConnectionName = _ConnectionName;
PERFORM Log(format(
'Cancelling connection %s id %s environment %s server %s node %s',
_ConnectionName,
s.TaskChainTaskId,
s.Environment,
s.Server,
s.NodeId));
PERFORM dblink_cancel_query(_ConnectionName);
UPDATE ServerConnection SET Cancelling = true
WHERE ConnectionName = s.ConnectionName;
END;
$$ LANGUAGE PLPGSQL;
/*
* Is the task still running?
*
* XXX: Bizarrely, the only way to determine if the connection has been
* reset without blocking is to call 'dblink_get_notify' twice so that
* the connection errror is established. This could probably be fixed
* by having dblink_is_busy check the response of PQconsumeInput.
*/
CREATE OR REPLACE FUNCTION TaskRunning(
_ConnectionName TEXT) RETURNS TaskExecutionState_t AS
$$
DECLARE
_ErrorMessage TEXT;
BEGIN
IF dblink_is_busy(_ConnectionName) = 0
THEN
RETURN 'FINISHED';
END IF;
PERFORM * from dblink_get_notify(_ConnectionName);
PERFORM * from dblink_get_notify(_ConnectionName);
_ErrorMessage := dblink_error_message(_ConnectionName);
IF _ErrorMessage LIKE 'server closed the connection%'
OR _ErrorMessage LIKE 'connection not open'
THEN
PERFORM Log('WARNING', format('connection: %s is dead via %s',
_ConnectionName, _ErrorMessage));
/* connection dead. pull the result while ignoring the error. */
PERFORM dblink_get_result(_ConnectionName, false);
/* clear connection status */
PERFORM dblink_get_result(_ConnectionName);
RETURN 'RESET';
END IF;
RETURN 'RUNNING';
END;
$$ LANGUAGE PLPGSQL;
CREATE OR REPLACE FUNCTION Log(
TEXT,
TEXT = NULL,
_Debug BOOL = false) RETURNS VOID AS
$$
DECLARE
_Message TEXT;
_Level TEXT;
BEGIN
IF $2 = 'DEBUG' AND _Debug IS NOT true
THEN
RETURN;
END IF;
IF $2 IS NULL
THEN
_Message := $1;
_Level = 'NOTICE';
ELSEIF $1 = 'DEBUG'
THEN
_Message := $2;
_Level = 'NOTICE';
ELSE
_Message := $2;
_Level = upper($1);
END IF;
INSERT INTO MasterLog(Level, Log) VALUES (_Level, _Message);
IF _Level = 'NOTICE'
THEN
RAISE NOTICE '%: %', clock_timestamp()::timestamp(1), _Message;
ELSIF _Level = 'WARNING'
THEN
RAISE WARNING '%: %', clock_timestamp()::timestamp(1), _Message;
ELSE
PERFORM * FROM
dblink_exec('dbname=' || current_database(),
format(
$q$INSERT INTO MasterLog(Level, Log) VALUES ('ERROR', %s)$q$,
quote_literal(_message)));
RAISE EXCEPTION '%: %', clock_timestamp()::timestamp(1), _Message;
END IF;
END;
$$ LANGUAGE PLPGSQL;
CREATE OR REPLACE FUNCTION Debug(ANYELEMENT) RETURNS ANYELEMENT AS
$$
BEGIN
RAISE NOTICE '%', $1;
RETURN $1;
END;
$$ LANGUAGE PLPGSQL;
CREATE OR REPLACE FUNCTION TaskClientThreads(_Server TEXT) RETURNS INT AS
$$
SELECT TaskClientThreads FROM Server WHERE Server = _Server;
$$ LANGUAGE SQL STABLE;
/* Connect to client databases, make sure they are all configured */
CREATE OR REPLACE FUNCTION InitializeClientDatabases() RETURNS VOID AS
$$
DECLARE
s Server;
d Database;
_HasClientDatabase BOOL;
_Version INT;
BEGIN
/* Initialize Client servers */
FOR d IN SELECT * FROM Database WHERE Server IN(
SELECT Server FROM Server WHERE Configured)
LOOP
/* test database connection and get version information */
SELECT v INTO _Version
FROM dblink(
d.ConnectionString,
$q$SELECT setting FROM pg_settings WHERE name = 'server_version_num'$q$)
AS R(v INT);
/* if we got this far we are able to connect :-). */
UPDATE Database SET Version = _Version WHERE Database = d.Database;
END LOOP;
END;
$$ LANGUAGE PLPGSQL;
/* Can this task chain be run? Return true if it can, false if it can be
* queue. If it is not able to be queued, return an error.
*/
CREATE OR REPLACE FUNCTION TaskChainCanStart(
_TaskChainConfigurationName TEXT,
_Tasks TEXT[],
_ConcurrencyModel ConcurrencyModel_t,
_Environment TEXT) RETURNS BOOL AS
$$
DECLARE
_Running BOOL;
_AlreadyQueued BOOL;
BEGIN
IF _ConcurrencyModel = 'NOBLOCK'
THEN
RETURN true;
END IF;
IF _ConcurrencyModel = 'BLOCK'
THEN
IF _TaskChainConfigurationName IS NULL
THEN
SELECT INTO _Running
EXISTS
(
SELECT 1
FROM TaskChainTask tct
JOIN TaskChain tc USING(TaskChainId)
WHERE
Task = any(_Tasks)
AND tc.Environment = _Environment
AND tct.Finished IS NULL
AND tc.Finished IS NULL
);
ELSE
SELECT INTO _Running
EXISTS
(
SELECT 1
FROM TaskChain
WHERE
TaskChainConfigurationName = _TaskChainConfigurationName
AND Environment = _Environment
AND Started IS NOT NULL AND Finished IS NULL
);
END IF;
IF _Running
THEN
PERFORM Log('ERROR', format(
'Can not start task chain configuration %s while already running',
COALESCE(_TaskChainConfigurationName, '<unnamed>')));
END IF;
RETURN true;
END IF;
IF _ConcurrencyModel IN ('QUEUE', 'QUEUE_ONCE')
THEN
IF _TaskChainConfigurationName IS NULL
THEN
SELECT INTO _Running, _AlreadyQueued
COUNT(DISTINCT TaskChainId) FILTER (WHERE tct.Started IS NOT NULL) > 0,
COUNT(DISTINCT TaskChainId) FILTER (WHERE tct.Started IS NULL) > 0
FROM TaskChainTask tct
JOIN TaskChain tc USING(TaskChainId)
WHERE
Task = any(_Tasks)
AND tc.Environment = _Environment
AND tct.Finished IS NULL
AND tc.Finished IS NULL;
ELSE
SELECT INTO _Running, _AlreadyQueued
COUNT(*) FILTER (WHERE Started IS NOT NULL) > 0,
COUNT(*) FILTER (WHERE Started IS NULL) > 0
FROM TaskChain
WHERE
TaskChainConfigurationName = _TaskChainConfigurationName
AND Environment = _Environment
AND Finished IS NULL;
END IF;
IF _ConcurrencyModel = 'QUEUE_ONCE' AND _AlreadyQueued
THEN
PERFORM Log('ERROR', format(
'Can not queue task chain configuration %s while already queued',
COALESCE(_TaskChainConfigurationName, '<unnamed>')));
END IF;
/* If it the task is already running, place it on the queue */
RETURN NOT _Running;
END IF;
IF _ConcurrencyModel = 'MERGE'
THEN
PERFORM Log('ERROR', 'Unimplemented');
END IF;
END;
$$ LANGUAGE PLPGSQL;
/* Sets up a task chain based on an array of tasks */
CREATE OR REPLACE FUNCTION CreateTaskChain(
_Tasks TEXT[], /* these are the tasks that *must* be run */
_Environment TEXT DEFAULT 'Production',
_WantParents BOOL DEFAULT true,
_WantChildren BOOL DEFAULT true,
_TaskChainConfigurationName TEXT DEFAULT NULL,
_ForceConcurrencyModel ConcurrencyModel_t DEFAULT NULL,
_User TEXT DEFAULT 'System') RETURNS BIGINT AS
$$
DECLARE
r RECORD;
_TaskChainId BIGINT;
tcc TaskChainConfiguration;
_CanStart BOOL;
BEGIN
PERFORM Log(format(
'User %s is requesting %s in %s',
_User,
'New Task Chain ' || COALESCE(_TaskChainConfigurationName, ''),
_Environment
));
/* The task chain may or may not be able to be immediately started. */
_CanStart := TaskChainCanStart(
_TaskChainConfigurationName,
_Tasks,
COALESCE(
_ForceConcurrencyModel,
(
SELECT DefaultConcurrencyModel
FROM TaskChainConfiguration
WHERE TaskChainConfigurationName = _TaskChainConfigurationName
),
'QUEUE_ONCE'::ConcurrencyModel_t),
_Environment);
INSERT INTO TaskChain(
Environment,
TaskChainConfigurationName,
Started)
VALUES (
_Environment,
_TaskChainConfigurationName,
CASE WHEN _CanStart THEN clock_timestamp() END)
RETURNING TaskChainId INTO _TaskChainId;
/* which tasks are we running? */
WITH tasks AS
(
SELECT Task
FROM
(
SELECT Task FROM DependentTasks(_Tasks, _WantParents, _WantChildren)
UNION ALL SELECT unnest(_Tasks) WHERE NOT _WantParents
) q
JOIN Task t USING(Task)
WHERE
t.RunSchedule IS NULL
OR Extract('dow' FROM now()) = ANY(t.RunSchedule)
)
INSERT INTO TaskChainTask(
TaskChainId,
Environment,
Database,
NodeId,
Task,
SubTask,
Eligible)
SELECT
_TaskChainId,
_Environment,
Database,
NodeId,
Task,
/* subtask is defined '' by convention, as we can't insert NULL */
'',
/*
* regular tasks are entered eligible if they have no parent dependencies
* in the chain
*/
NOT EXISTS (
SELECT 1 FROM tasks
JOIN TaskDependency td ON
td.ChildTask = t.Task
AND td.ParentTask = tasks.Task
)
FROM Task t
JOIN Database USING(Database)
WHERE
t.Task IN (
SELECT Task FROM tasks
)
AND Environment = _Environment;
INSERT INTO TaskChainTask(
TaskChainId,
Environment,
Database,
NodeId,
Task,
SubTask,
Eligible)
SELECT
_TaskChainId,
_Environment,
Database,
NodeId,
Task,
SubTask,
false /* subtasks are always intialized false */
FROM SubTask st
JOIN Task t USING(Task)
JOIN Database d USING(Database)
WHERE
t.Task IN (
SELECT Task FROM DependentTasks(_Tasks, _WantParents, _WantChildren)
UNION ALL SELECT unnest(_Tasks) WHERE NOT _WantParents
)
AND (t.RunSchedule IS NULL
OR Extract('dow' FROM now()) IN (SELECT unnest(t.RunSchedule))
)
AND (
st.RunSchedule iS NULL
OR Extract ('dow' FROM now()) IN (select unnest(st.RunSchedule))
)
AND Environment = _Environment;
RETURN _TaskChainId;
END;
$$ LANGUAGE PLPGSQL;
/* non array variant of CreateTaskChain */
CREATE OR REPLACE FUNCTION CreateTaskChainFromTask(
_Task TEXT,
_Environment TEXT DEFAULT 'Production',
_WantParents BOOL DEFAULT true,
_WantChildren BOOL DEFAULT true,
_ForceConcurrencyModel ConcurrencyModel_t DEFAULT NULL,
_User TEXT DEFAULT 'System') RETURNS BIGINT AS
$$
SELECT CreateTaskChain(
array[_Task],
_Environment,
_WantParents,
_WantChildren,
_User)
$$ LANGUAGE SQL;
/* Creates a task chain from a named configuration */
CREATE OR REPLACE FUNCTION CreateTaskChain(
_TaskChainConfigurationName TEXT,
_Environment TEXT DEFAULT 'Production',
_WantParents BOOL DEFAULT true,
_WantChildren BOOL DEFAULT true,
_ForceConcurrencyModel ConcurrencyModel_t DEFAULT NULL,
_User TEXT DEFAULT 'System') RETURNS BIGINT AS
$$
SELECT CreateTaskChain(
array(
SELECT Task
FROM TaskChainConfigurationTask
WHERE TaskChainConfigurationName = _TaskChainConfigurationName
),
_Environment,
_WantParents,
_WantChildren,
_TaskChainConfigurationName,
_ForceConcurrencyModel,
_User)
FROM TaskChainConfiguration
WHERE TaskChainConfigurationName = _TaskChainConfigurationName;
$$ LANGUAGE SQL;
CREATE OR REPLACE FUNCTION FinishTaskChain(
_TaskChainId BIGINT) RETURNS VOID AS
$$
DECLARE
r RECORD;
BEGIN
UPDATE TaskChain SET Finished = clock_timestamp()
WHERE TaskChainId = _TaskChainId;
END;
$$ LANGUAGE PLPGSQL;
CREATE OR REPLACE FUNCTION DependentTasks(
_Tasks TEXT[],
_WantParents BOOL DEFAULT false,
_WantChildren BOOL DEFAULT true,
Task OUT TEXT,
CanRun OUT BOOL,
AllNodes OUT BOOL) RETURNS SETOF RECORD AS
$$
WITH RECURSIVE c AS
(
SELECT
t.Task,
NULL::TEXT AS ParentTask,
NULL::TEXT AS ChildTask,
NULL::BOOL AS CanRun,
NULL::BOOL AS AllNodes
FROM Task t
WHERE
Task = any(_Tasks)
AND _WantParents
UNION ALL SELECT
p.Task,
d.ParentTask,
d.ChildTask,
d.CanRun,
d.AllNodes
FROM c
JOIN TaskDependency d ON d.ChildTask = c.Task
JOIN Task p ON p.Task = d.ParentTask
),
p AS
(
SELECT
t.Task,
NULL::TEXT AS ParentTask,
NULL::TEXT AS ChildTask,
NULL::BOOL AS CanRun,
NULL::BOOL AS AllNodes
FROM Task t
WHERE
Task = any(_Tasks)
AND _WantChildren
UNION ALL SELECT
c.Task,
d.ParentTask,
d.ChildTask,
d.CanRun,
/* went travelling downstream, the all nodes bit carries so that
* once you've seen it in the tree, it is always true. The idea is that
* a cross node dependency, once identified, must always be honored.
*/
COALESCE(p.AllNodes, false) OR d.AllNodes
FROM p
JOIN TaskDependency d ON d.ParentTask = p.Task
JOIN Task c ON c.Task = d.ChildTask
)
SELECT DISTINCT
Task,
CanRun,
AllNodes
FROM
(
SELECT * FROM p
/* children does not include self */
WHERE Task NOT IN (SELECT Unnest(_Tasks))
/* parents includes self */
UNION ALL SELECT * FROM c
) q;
$$ LANGUAGE SQL;
CREATE OR REPLACE FUNCTION DependentTasks(
_Task TEXT,
_WantParents BOOL DEFAULT false,
_WantChildren BOOL DEFAULT true,
Task OUT TEXT,
CanRun OUT BOOL,
AllNodes OUT BOOL) RETURNS SETOF RECORD AS
$$
SELECT DependentTasks(array[_Task], _WantParents, _WantChildren);
$$ LANGUAGE SQL;
CREATE OR REPLACE FUNCTION Duration(_i INTERVAL) RETURNS TEXT AS
$$
SELECT
CASE WHEN _i IS NOT NULL THEN
format('%sh %s',
floor(extract('epoch' from _i) / 3600),
to_char(_i, 'MIm SSs'))
ELSE NULL END;
$$ LANGUAGE SQL;
/* Internal cancel function. This can only be called by the same process
* running main() because only the process running dblink connections can
* cancel them.
*/
CREATE OR REPLACE FUNCTION CancelTaskChain(
_TaskChainId BIGINT,
_Pause BOOL) RETURNS VOID AS
$$
DECLARE
q TEXT;
n RECORD;
sc ServerConnection;
_ConnectionNames TEXT[];
gc GlobalControl;
BEGIN
SELECT INTO gc * FROM GlobalControl;
IF pg_backend_pid() != gc.MainProcessPid
THEN
PERFORM Log('ERROR',
'This function must be called from main process. Call Cancel()');
END IF;
PERFORM Log(format('Cancelling task chain id %s for %s',
_TaskChainId,
CASE WHEN _Pause THEN 'Pause' ELSE 'Administrative Cancel' END));
FOR sc IN
SELECT *
FROM ServerConnection
WHERE
Active
AND TaskChainTaskId IN (
SELECT tct.TaskChainTaskId
FROM TaskChainTask tct
WHERE TaskChainId = _TaskChainId
)
LOOP
/* connections are now ready to go */
PERFORM TaskFinished(
sc.TaskChainTaskId,
CASE WHEN NOT _Pause THEN 'FAILED_VIA_CANCEL' END::TaskResult_t,
CASE WHEN NOT _Pause THEN
'Cancelled while in progress due to administrative cancel' END,
_Pause);
END LOOP;
END;
$$ LANGUAGE PLPGSQL;
/* Cancel all running tasks (optionally to a specific chain) and pause
* execution. All chains so paused must be so marked.
*
* The list of paused task chains will be returned; if a task chain is not
* running it will not be returned.
*/
CREATE OR REPLACE FUNCTION Pause(
_TaskChainId BIGINT DEFAULT NULL,
_User TEXT DEFAULT 'System',
TaskChainId OUT BIGINT) RETURNS SETOF BIGINT AS
$$
BEGIN
PERFORM Log(format(
'User %s is pausing %s',
_User,
CASE WHEN _TaskChainId IS NULL
THEN 'all task chains'
ELSE 'task chain ' || _TaskChainId
END));
RETURN QUERY UPDATE TaskChain tc SET PendingAction = 'PAUSE'
WHERE
(tc.TaskChainId = _TaskChainId OR _TaskChainId IS NULL)
AND Finished IS NULL
RETURNING tc.TaskChainId;
END;
$$ LANGUAGE PLPGSQL;
/* Resume all paused task chains or a specific paused task chain.
*
* The list of resumed task chains will be returned; if a task chain is not
* considered able to be resumed it will not be returned.
*/
CREATE OR REPLACE FUNCTION Resume(
_TaskChainId BIGINT DEFAULT NULL,
_User TEXT DEFAULT 'System',
TaskChainId OUT BIGINT) RETURNS SETOF BIGINT AS
$$
BEGIN
PERFORM Log(format(
'User %s is resuming %s',
_User,
CASE WHEN _TaskChainId IS NULL
THEN 'all task chains'
ELSE 'task chain ' || _TaskChainId
END));
RETURN QUERY UPDATE TaskChain tc SET
Paused = false,
PauseEndTime = clock_timestamp()
WHERE
(
tc.TaskChainId = _TaskChainId
OR _TaskChainId IS NULL
)
AND Paused = true
AND Finished IS NULL
RETURNING tc.TaskChainId;
END;
$$ LANGUAGE PLPGSQL;
/* Cancel all running task chains or a specific running task chain.
*
* The list of cancelled task chains will be returned; if a task chain is not
* considered able to be cancelled it will not be returned.
*
* This will only work correctly if called from the procedure which owns the
* connections being cancelled.
*/
CREATE OR REPLACE FUNCTION Cancel(
_TaskChainId BIGINT DEFAULT NULL,
_User TEXT DEFAULT 'System',
TaskChainId OUT BIGINT) RETURNS SETOF BIGINT AS
$$
BEGIN
PERFORM Log(format(
'User %s is cancelling %s',
_User,
CASE WHEN _TaskChainId IS NULL
THEN 'all task chains'
ELSE 'task chain ' || _TaskChainId
END));
RETURN QUERY UPDATE TaskChain tc SET PendingAction = 'CANCEL'
WHERE
(tc.TaskChainId = _TaskChainId OR _TaskChainId IS NULL)
AND Finished IS NULL
RETURNING tc.TaskChainId;
END;
$$ LANGUAGE PLPGSQL;
/* Is it appropriate to run the task at this time? */
CREATE OR REPLACE FUNCTION TimeEligible(
_StartTime TIME,
_EndTime TIME,
_CurrentTime TIME DEFAULT now()::TIME) RETURNS BOOL AS
$$
SELECT
CASE
WHEN _StartTime IS NULL AND _EndTime IS NULL
THEN true
WHEN _StartTime IS NOT NULL AND _EndTime IS NULL
THEN _CurrentTime >= _StartTime
WHEN _StartTime IS NULL AND _EndTime IS NOT NULL
THEN _CurrentTime <= _EndTime
WHEN _StartTime = _EndTime
THEN false
WHEN _StartTime < _EndTime
THEN _CurrentTime BETWEEN _StartTime AND _EndTime
WHEN _StartTime > _EndTime
THEN _CurrentTime > _StartTime OR _CurrentTime < _EndTime
END
$$ LANGUAGE SQL IMMUTABLE;
/* Returns task to run so that all dependencies (other tasks, time of day etc)
* must met. If there tasks ready for execution, the one with the earliest
* entered time will be returned.
*
* NULL is returned if there is nothing to do.
*/
CREATE OR REPLACE FUNCTION GetTask(
TaskChainTaskId OUT BIGINT) RETURNS BIGINT AS
$$
BEGIN
SELECT INTO GetTask.TaskChainTaskId tct.TaskChainTaskId
FROM TaskChainTask tct
JOIN Database USING(Environment, Database, NodeId)
JOIN TaskChain tc USING(TaskChainId)
JOIN vw_Server USING (Server)
JOIN Task t USING(Task)
WHERE
/* Are we eligible for dependency reasons? */
Eligible
AND ActiveConnections < TaskClientThreads
/* Task chain must be qualifying */
AND tc.Started IS NOT NULL
AND tc.Finished IS NULL
AND NOT tc.Paused
/* Don't return any tasks for chains with pending administrative action */
AND tc.PendingAction IS NULL
/* Is the task eligible to run at this time? */
AND TimeEligible(t.StartTime, t.EndTime)
/* The task must not already have started... */
AND tct.Started IS NULL
AND tct.Finished IS NULL
ORDER BY
tct.Yielded NULLS FIRST,
t.TaskOrder,
tct.TaskChainTaskId LIMIT 1;
END;
$$ LANGUAGE PLPGSQL;
CREATE OR REPLACE FUNCTION GetTasks(
TaskChainTaskId OUT BIGINT,
_YieldTime INTERVAL DEFAULT '15 minutes'::INTERVAL) RETURNS SETOF BIGINT AS
$$
BEGIN
RETURN QUERY
SELECT q.TaskChainTaskId
FROM
(
SELECT
tct.TaskChainTaskId,
TaskClientThreads - ActiveConnections AS AvailableThreads,
row_number() OVER (
PARTITION BY Server
ORDER BY COALESCE(Parent.Started, tct.Started)) AS Count
FROM TaskChainTask tct
JOIN Database USING(Environment, Database, NodeId)
JOIN TaskChain tc USING(TaskChainId)
JOIN vw_Server USING (Server)
JOIN Task t USING(Task)
LEFT JOIN TaskChainTask parent ON
tct.SubTask != ''
AND parent.Subtask = ''
AND parent.Task = tct.Task
AND parent.TaskChainId = tct.TaskChainId
AND parent.Environment = tct.Environment
AND parent.Database = tct.Database
AND parent.NodeId = tct.NodeId
WHERE
/* Are we eligible for dependency reasons? */
tct.Eligible
/* This is on off switch at server level */
AND ActiveConnections < TaskClientThreads
/* Task chain must be qualifying */
AND tc.Started IS NOT NULL
AND tc.Finished IS NULL
AND NOT tc.Paused
AND (tct.Yielded IS NULL OR tct.Yielded < now() - _YieldTime)
/* Don't return any tasks for chains with pending administrative action */
AND tc.PendingAction IS NULL
/* Is the task eligible to run at this time? */
AND TimeEligible(t.StartTime, t.EndTime)
/* The task must not already have started... */
AND tct.Started IS NULL
AND tct.Finished IS NULL
) q
WHERE Count <= AvailableThreads;
END;
$$ LANGUAGE PLPGSQL;
/*
* Runs a task. Returns true if generally anything interesting happened;
* the task was run or there was some kind of failure. If no connection was
* available, (this may happen if the server the task is trying to get a
* connection to is completely down), false is returned to inform the main
* stored procedure loop to chill out for a bit
*/
CREATE OR REPLACE FUNCTION RunTask(
_TaskChainTaskId BIGINT) RETURNS BOOL AS
$$
DECLARE
tct TaskChainTask;
r INT;
_ConnectionName TEXT;
gc GlobalControl;
_TaskBody TEXT;
_d RECORD;
_t RECORD;
_ErrorMessage TEXT;
BEGIN
PERFORM Log(format('Starting Task id %s', _TaskChainTaskId));
SELECT INTO gc * FROM GlobalControl;
SELECT INTO tct * FROM TaskChainTask
WHERE TaskChainTaskId = _TaskChainTaskId;
IF tct.SubTask = ''
THEN
SELECT INTO _TaskBody TaskBody FROM Task WHERE Task = tct.Task;
ELSE
IF COALESCE(tct.TaskBody,'') = '' THEN
SELECT INTO _TaskBody TaskBody
FROM SubTask
WHERE
Task = tct.Task
AND SubTask = tct.SubTask;
ELSE
/* if subtask was dynamically created sub taskbody will be in taskchaintask */
_TaskBody := tct.TaskBody;
END IF;
END IF;
IF NOT FOUND
THEN
PERFORM Log('ERROR', format(
'Could not resolve Task %s%s for ID %s',
tct.Task,
CASE WHEN tct.SubTask != '' THEN '/' || tct.SubTask END,
_TaskChainTaskId));
END IF;
IF tct.TaskChainTaskId IS NULL
THEN
PERFORM Log('ERROR', format('Tried to run invalid id %s', _TaskChainTaskId));
END IF;
IF _TaskBody = '' AND tct.SubTask = ''
THEN
/* Parent task -- NOP. Just mark started. */
PERFORM TaskStarted(_TaskChainTaskId, FALSE);
/* if there are no sub tasks, immediately finish the task */
IF NOT EXISTS (
SELECT 1 FROM TaskChainTask s
WHERE
s.TaskChainId = tct.TaskChainId
AND s.Task = tct.Task
AND s.SubTask != '')
THEN
PERFORM TaskFinished(
tct.TaskChainTaskId,
'SUCCESS_EMPTY',
'Auto finishing task with no subtasks');
END IF;
RETURN TRUE;
END IF;
PERFORM Log(format(
'Running task %s%s id %s Environment %s Database %s NodeId %s',
tct.Task,
CASE WHEN tct.SubTask != '' THEN '/' || tct.SubTask END,
_TaskChainTaskId,
tct.Environment,
tct.Database,
tct.NodeId));
/* Get a connection to the database */
_ConnectionName := DblinkConnection(
tct.Environment,
tct.Database,
tct.NodeId);
IF _ConnectionName IS NULL
THEN
_ErrorMessage := format(
'Could not get connection to database %s node %s for id %s',
tct.Database, tct.NodeId, _TaskChainTaskId);
PERFORM Log('WARNING', _ErrorMessage);
/* in case a connection is not available, YIELD the task for later execution by
* by appending [PGTASK.YIELD] keyword in the errormessage , this will prevent the
* mainloop from getting same task instead of processing other pending eligible tasks
*/
PERFORM TaskFinished(
tct.TaskChainTaskId,
'SUCCESS_EMPTY',
format('[PGTASK.YIELD] %s', _ErrorMessage));
RETURN FALSE;
END IF;
UPDATE ServerConnection SET TaskChainTaskId = _TaskChainTaskId
WHERE
ConnectionName = _ConnectionName;
SELECT INTO _d
LogMinMessages,
LogErrorVerbosity
FROM Database d
WHERE d.Database = tct.Database
AND d.Environment = tct.Environment
AND d.NodeId = tct.NodeId
LIMIT 1;
SELECT INTO _t
LogMinMessages,
LogErrorVerbosity
FROM Task t
WHERE t.task = tct.task;
PERFORM Log('DEBUG', format('Got connection %s', _ConnectionName), gc.Debug);
/* run it */
BEGIN
SELECT INTO r dblink_send_query(
_ConnectionName,
FormatTaskQuery(
CASE WHEN gc.Debug THEN gc.DebugQuery ELSE _TaskBody END,
tct.Task,
tct.SubTask,
_TaskChainTaskId,
gc.TaskMinMessages,
COALESCE(_t.LogMinMessages, _d.LogMinMessages, gc.LogMinMessages, 'NOTICE'),
COALESCE(_t.LogErrorVerbosity, _d.LogErrorVerbosity, gc.LogErrorVerbosity, 'TERSE')));
IF r = 0
THEN
PERFORM Log('WARNING',
format('Failed to send query for id %s', _TaskChainTaskId));
RETURN FALSE;
END IF;
EXCEPTION WHEN OTHERS THEN
PERFORM Log('WARNING', format(
'Got %s while attempting to run task id %s',
SQLERRM,
_TaskChainTaskId));
RETURN FALSE;
END;
/* Mark task started */
PERFORM TaskStarted(_TaskChainTaskId);
UPDATE ServerConnection SET TaskChainTaskId = _TaskChainTaskId
WHERE ConnectionName = _ConnectionName;
RETURN TRUE;
END;
$$ LANGUAGE PLPGSQL;
/*
* Mark a task as having been started.
*/
CREATE OR REPLACE FUNCTION TaskStarted(
_TaskChainTaskId BIGINT,
_MarkConnection BOOL DEFAULT true) RETURNS VOID AS
$$
DECLARE
gc GlobalControl;
BEGIN
SELECT INTO gc * FROM GlobalControl;
UPDATE TaskChainTask SET Started = clock_timestamp()
WHERE
TaskChainTaskId = _TaskChainTaskId
AND Started IS NULL;
/* Mark the connection the task is attached to as active, but only if this is
* a task that requires actually doing something (that is, not a task that
* contains subtasks).
*/
IF _MarkConnection
THEN
UPDATE ServerConnection SET Active = true
WHERE TaskChainTaskId = _TaskChainTaskId;
END IF;
END;
$$ LANGUAGE PLPGSQL;
/*
* Marks a task as finished. All dependent tasks are cancelled on the node
* for canRun = false. Task failures can travel through dependencies across
* nodes if and only if the can node flag is set.
*
* In a pause scenario, the task is not finished, however the connection is
* released.
*
*/
CREATE OR REPLACE FUNCTION TasksFinished(
_TaskChainTaskIds BIGINT[],
_TaskResult TaskResult_t,
_ErrorMessage TEXT,
_Pause BOOL DEFAULT FALSE) RETURNS VOID AS
$$
DECLARE
_TasksLeft INT;
_Found BOOL;
_Finished TIMESTAMPTZ;
_TaskChainComplete BOOL;
t TaskChainTask[];
_FinishedString TEXT;
/* if a task was yielded by the source for some reason then
* the task is pushed for later execution
*/
_Yield BOOL := FALSE;
gc GlobalControl;
_ActiveTasks BIGINT[];
BEGIN
SELECT INTO gc * FROM GlobalControl;
/* save off the tasks to cancel now, so that we can clear the cancel flag
* before the cancel is thrown (otherwise we get repeated processing of the
* of cancellation event).
*/
SELECT INTO _ActiveTasks array_agg(TaskChainTaskId)
FROM ServerConnection
WHERE
TaskChainTaskId = ANY(_TaskChainTaskIds)
AND Active;
/* Mark the connection the task is attached to as inactive */
UPDATE ServerConnection SET Active = false
WHERE TaskChainTaskId = ANY(_ActiveTasks);
/* Check if the finished task was actually yielded, if it was
* yielded then reset task status so it can be called again
*/
_Yield := (_ErrorMessage LIKE '%[PGTASK.YIELD]%');
IF _Yield THEN
/* reset status of Task if it was yielded */
WITH tct AS
(
UPDATE TaskChainTask SET
Started = NULL,
Yielded = CASE WHEN _Yield THEN clock_timestamp() END
WHERE
TaskChainTaskId = ANY(_TaskChainTaskIds)
AND TaskResult IS NULL
RETURNING *
)
SELECT INTO t array_agg(tct) FROM tct;
ELSE
WITH tct AS
(
UPDATE TaskChainTask SET
Started =
CASE WHEN NOT _Pause THEN COALESCE(Started, clock_timestamp()) END,
Finished = CASE WHEN NOT _Pause THEN clock_timestamp() END,
TaskResult = CASE WHEN NOT _Pause THEN _TaskResult END,
ErrorMessage = CASE WHEN NOT _Pause THEN _ErrorMessage END
WHERE
TaskChainTaskId = ANY(_TaskChainTaskIds)
AND TaskResult IS NULL
RETURNING *
)
SELECT INTO t array_agg(tct) FROM tct;
END IF;
/* log finshed tasks. If all tasks were determined to be cancelled for
* other reasons (typically due to a admin cancel chaining into a query
* cancel), do nothing.
*/
IF array_upper(t, 1) > 0
THEN
SELECT INTO _FinishedString
CASE
WHEN array_upper(t, 1) = 1
THEN format(
'Task %s%s id %s Environment %s Database %s Node %s',
t[1].Task,
CASE WHEN t[1].SubTask != '' THEN '/' || t[1].SubTask END,
t[1].TaskChainTaskId,
t[1].Environment,
t[1].Database,
t[1].NodeId)
ELSE format('%s tasks', array_upper(t, 1))
END;
PERFORM Log(format(
'%s %s Result %s Message %s',
CASE WHEN _Pause THEN 'Pausing' WHEN _Yield THEN 'Yielding' ELSE 'Finishing' END,
_FinishedString,
_TaskResult,
_ErrorMessage));
END IF;
IF ShouldCancel(_TaskResult) OR _Pause
THEN
/* There are scenarios where the task may still be running when an attempt
* is made to finish it e.g. cancelling situations
*/
PERFORM CancelTask(ConnectionName)
FROM ServerConnection
WHERE
TaskChainTaskId = ANY(_ActiveTasks);
END IF;
IF NOT _Pause
THEN
/* are there any tasks that have yet to finish? */
SELECT INTO _TaskChainComplete NOT EXISTS
(
SELECT 1 FROM TaskChainTask
WHERE
Finished IS NULL
AND TaskChainId = t[1].TaskChainId
);
PERFORM FinishTaskChain(t[1].TaskChainId)
WHERE _TaskChainComplete;
END IF;
END;
$$ LANGUAGE PLPGSQL;
CREATE OR REPLACE FUNCTION TaskFinished(
_TaskChainTaskId BIGINT,
_TaskResult TaskResult_t,
_ErrorMessage TEXT,
_Pause BOOL DEFAULT FALSE) RETURNS VOID AS
$$
SELECT TasksFinished(
array[_TaskChainTaskId],
_TaskResult,
_ErrorMessage,
_Pause);
$$ LANGUAGE SQL;
/*
* Function checks whether the task chain is paused
*/
CREATE OR REPLACE FUNCTION Paused(_TaskChainId BIGINT) RETURNS BOOL AS
$$
BEGIN
RETURN (SELECT IsPaused FROM TaskChain WHERE TaskChainId = _TaskChainId);
END;
$$ LANGUAGE PLPGSQL;
/* Welp, sometimes things don't work out as we've expected. Upon start up,
* check and see if there are any tasks running in any server/database
* this pgtask instance is coordinating. If there are, destroy them as the
* expectation is that they are running untracked.
*/
CREATE OR REPLACE FUNCTION NukeRunningTasks() RETURNS VOID AS
$$
DECLARE
d Database;
_q TEXT;
BEGIN
PERFORM Log('Cleaning possibly bad pgtask connections');
_q := format($q$
SELECT pg_terminate_backend(pid)
FROM pg_stat_activity
WHERE
query LIKE '%%%s%%'
AND pid != pg_backend_pid()
$q$, (SELECT ServerId FROM GlobalControl));
FOR d IN
SELECT * FROM Database
LOOP
BEGIN
PERFORM Log(format(
'Cleaning pgtask connections for database %s environment %s node %s',
d.Database,
d.Environment,
d.NodeId));
/* paranoia: kill any running pgtask queries */
PERFORM * FROM dblink(
d.ConnectionString,
_q ) AS R(V TEXT);
EXCEPTION WHEN OTHERS THEN
PERFORM Log(format(
'Got %s when attempting to kill connections on databsae %s'
' environment %s node %s',
SQLERRM,
d.Database,
d.Environment,
d.NodeId));
END;
END LOOP;
END;
$$ LANGUAGE PLPGSQL;
CREATE OR REPLACE FUNCTION CreateDblinkConnections() RETURNS VOID AS
$$
BEGIN
PERFORM Log('Initializing Connections');
/* There is no confidence in tracked server connections across startup */
TRUNCATE ServerConnection;
PERFORM dblink_disconnect(c) FROM unnest(dblink_get_connections()) c;
INSERT INTO ServerConnection(
Server,
ConnectionNumber,
ConnectionName,
Active,
Cancelling)
SELECT
Server,
Number,
format('%s-%s', Server, Number),
false,
false
FROM
(
SELECT
Server,
/* SRF in tlist is dodgy but works in this simple case */
generate_series(1, TaskClientThreads) AS Number
FROM Server
) q;
PERFORM NukeRunningTasks();
END;
$$ LANGUAGE PLPGSQL;
CREATE OR REPLACE FUNCTION VerifyDblinkConnection(
_Environment TEXT,
_Database TEXT,
_NodeId INT,
ConnectionName INOUT TEXT) RETURNS TEXT AS
$$
BEGIN
PERFORM Log(format(
'Testing connection %s to environment %s database %s NodeId %s',
ConnectionName,
_Environment,
_Database,
_NodeId));
PERFORM * FROM dblink(ConnectionName, 'SELECT 0') AS R(v int);
EXCEPTION WHEN OTHERS THEN
/* uh oh. Try to reconnect. */
RAISE WARNING
'Got % when attempting to use connection % to environment % database % nodeid %',
SQLERRM,
ConnectionName,
_Environment,
_Database,
_NodeId;
BEGIN
PERFORM dblink_disconnect(ConnectionName);
EXCEPTION WHEN OTHERS THEN
RAISE WARNING
'Got % when attempting to disconnect connection % from environment % database % nodeid %',
SQLERRM,
ConnectionName,
_Environment,
_Database,
_NodeId;
BEGIN
PERFORM dblink_connect(ConnectionName,
CASE WHEN ServerType = 'POSTGRES'::ServerType_t
THEN ConnectionString
ELSE ProxyConnectionString()
END )
FROM Database
JOIN Server USING(Server)
WHERE
Environment = _Environment
AND Database = _Database
AND NodeId = _NodeId;
PERFORM Log('WARNING', format(
'Reestablished connection %s to environment %s database %s nodeid %s',
ConnectionName,
_Environment,
_Database,
_NodeId));
EXCEPTION WHEN OTHERS THEN
PERFORM Log('WARNING', format(
'Punting: Got %s when attempting to disconnect connection %s '
'from environment %s database %s nodeid %s',
SQLERRM,
ConnectionName,
_Environment,
_Database,
_NodeId));
/* Could not reconnect. Mark the connection is not active and punt */
UPDATE ServerConnection sc SET
Environment = NULL,
Database = NULL,
NodeId = NULL,
Active = false
WHERE sc.ConnectionName = VerifyDblinkConnection.ConnectionName;
ConnectionName := NULL;
RETURN;
END;
END;
END;
$$ LANGUAGE PLPGSQL;
CREATE OR REPLACE FUNCTION SetKeepAlives() RETURNS TRIGGER AS
$$
BEGIN
/* is this postgres databse? or are keepalives already set? */
PERFORM 1 FROM Server
WHERE
Server = NEW.Server
AND ServerType = 'POSTGRES'
AND NOT NEW.ConnectionString ~* 'keepalive';
IF NOT FOUND
THEN
RETURN NEW;
END IF;
NEW.ConnectionString := NEW.ConnectionString ||
' keepalives_idle=300 keepalives_interval=5 keepalives_count=5';
RETURN NEW;
END;
$$ LANGUAGE PLPGSQL;
CREATE TRIGGER OnDatabaseChange
BEFORE INSERT OR UPDATE ON Database FOR EACH ROW
EXECUTE PROCEDURE SetKeepAlives();
CREATE OR REPLACE FUNCTION ProxyConnectionString() RETURNS TEXT AS
$$
SELECT format(
'host=localhost dbname=%s user=%s',
current_database(),
current_user)
$$ LANGUAGE SQL;
/* Fetch a connection from the server pool. Prefer a conenction that is
* already connected to the database but if necessary reconnect to the
* database if one is not available already connected.
*
* If the connection is already established, Test it. If the test fails,
* mark the database invalid.
*
* Returns connection name. If no connection is available, returns NULL
*/
CREATE OR REPLACE FUNCTION DblinkConnection(
_Environment TEXT,
_Database TEXT,
_NodeId INT,
ConnectionName OUT TEXT) RETURNS TEXT AS
$$
DECLARE
_Connected BOOL;
BEGIN
/* is there a connection available that is already pinned to the database? */
SELECT
sc.ConnectionName
INTO DblinkConnection.ConnectionName
FROM ServerConnection sc
WHERE
Environment = _Environment
AND Database = _Database
AND NodeId = _NodeId
AND NOT Active
AND NOT Cancelling
LIMIT 1;
IF NOT FOUND
THEN
/*
* no existing connections available. look for free connection on the
* database's server.
*/
SELECT
sc.ConnectionName,
Database IS NOT NULL
INTO
DblinkConnection.ConnectionName,
_Connected
FROM ServerConnection sc
WHERE
Server = (
SELECT Server
FROM Database d
WHERE
d.Environment = _Environment
AND d.Database = _Database
AND d.NodeId = _NodeId)
AND NOT Active
AND NOT Cancelling
ORDER BY random()
LIMIT 1;
IF NOT FOUND
THEN
/* no connections are available. This is not supposed to happen but
* nothing to do but return NULL and punt.
*/
PERFORM Log(
'WARNING',
format(
'Not able to acquire connection to environment %s database %s Node %s',
_Environment,
_Database,
_NodeId));
RETURN;
END IF;
BEGIN
/* There is a connection available that is not attached to the database
* we want. If it is connected to another database, disconnect it and
* connect to the correct database
*/
IF _Connected
THEN
PERFORM dblink_disconnect(ConnectionName);
END IF;
PERFORM dblink_connect(ConnectionName,
CASE WHEN ServerType = 'POSTGRES'::ServerType_t
THEN ConnectionString
ELSE ProxyConnectionString()
END)
FROM Database
JOIN Server USING(Server)
WHERE
Environment = _Environment
AND Database = _Database
AND NodeId = _NodeId;
EXCEPTION WHEN OTHERS THEN
PERFORM Log('WARNING', format(
'Got %s when attempting to reconnect connection %s'
' to environment %s database %s node %s',
SQLERRM,
ConnectionName,
_Environment,
_Database,
_NodeId));
UPDATE ServerConnection sc SET
Environment = NULL,
Database = NULL,
NodeId = NULL
WHERE sc.ConnectionName = DblinkConnection.ConnectionName;
ConnectionName := NULL;
RETURN;
END;
END IF;
ConnectionName := VerifyDblinkConnection(
_Environment,
_Database,
_NodeId,
ConnectionName);
UPDATE ServerConnection sc SET
Environment = _Environment,
Database = _Database,
NodeId = _NodeId
WHERE
sc.ConnectionName = DblinkConnection.ConnectionName
AND DblinkConnection.ConnectionName IS NOT NULL;
END;
$$ LANGUAGE PLPGSQL;
/* wrap a task with do..execute so that we don't have to worry about columns.
*
* For a complicated set of reasons, client_min_messages, not log_min_messages,
* controls how pgtask logs task output to the local database log. The default
* client_min_messags is 'NOTICE', which may tend to be overly verbose by
* default. We therefore grab the setting here and invoke it as part of task
* execution.
*/
DROP FUNCTION IF EXISTS FormatTaskQuery(TEXT, TEXT, TEXT, BIGINT, TEXT);
CREATE OR REPLACE FUNCTION FormatTaskQuery(
_TaskBody TEXT,
_Task TEXT,
_SubTask TEXT,
_TaskChainTaskId BIGINT,
_TaskMinMessages TEXT,
_TaskLogMinMessage TEXT,
_TaskLogErrorVerbosity TEXT) RETURNS TEXT AS
$$
SELECT format(
E'DO /* running from pgtask instance %s for task %s%s task id [%s] */'
' \n$pgtask$\nBEGIN\n'
' DISCARD TEMP;\n'
' SET LOCAL client_min_messages = %s;'
' SET LOCAL log_min_messages = %s;'
' SET LOCAL log_error_verbosity = %s;'
' EXECUTE\n$pgtaskbody$\n/* Task body starts here */'
'\n\n%s;\n\n/* Task body ends here */\n$pgtaskbody$;'
'\nEND;\n$pgtask$',
(SELECT ServerId FROM GlobalControl),
_Task,
CASE WHEN _SubTask != '' THEN '/' || _Subtask END,
_TaskChainTaskId,
_TaskMinMessages,
_TaskLogMinMessage,
_TaskLogErrorVerbosity,
_TaskBody);
$$ LANGUAGE SQL IMMUTABLE;
CREATE OR REPLACE FUNCTION Disconnect(_ConnectionName TEXT) RETURNS VOID AS
$$
BEGIN
UPDATE ServerConnection sc SET
Environment = NULL,
Database = NULL,
NodeId = NULL,
Active = false,
Cancelling = false,
TaskChainTaskId = NULL
WHERE sc.ConnectionName = _ConnectionName;
PERFORM dblink_disconnect(_ConnectionName);
END;
$$ LANGUAGE PLPGSQL;
CREATE OR REPLACE FUNCTION ConnectionDead(
_ConnectionName TEXT) RETURNS VOID AS
$$
BEGIN
PERFORM Log(format('Marking connection %s dead', _ConnectionName));
PERFORM TaskFinished(
TaskChainTaskId,
'FAILED_VIA_CONNECTION_RESET',
'Connection to remote server has gone bad')
FROM ServerConnection
WHERE ConnectionName = _ConnectionName;
/* Mark connection bad */
PERFORM Disconnect(_ConnectionName);
END;
$$ LANGUAGE PLPGSQL;
/*
* Check and see if tasks are still running, and, if they are determined to be
* finished, clean them up.
*/
CREATE OR REPLACE FUNCTION CheckForFinishedTasks(
DidStuff OUT BOOL) RETURNS BOOL AS
$$
DECLARE
r RECORD;
_DidSucceed BOOL;
_ErrorMessage TEXT;
_State TaskExecutionState_t;
_Cancelling BOOL;
_Active BOOL;
/* Controls how cancellations are managed. Values:
* DELAYED: continue to loop and want for dblink_get_result
* to return as cancelled and reap query normally. This is the
* best in terms of behavior, but unfortunately, there are now
* some bugs (in dblink), that cause dblink to not be able to
* pick up the cancellation and so the connection gets hung up
* forever.
* DISCONNECT: send a cancel, but then disconnect the query without waiting.
* IMMEDIATE: fire an immediate dblink_get_result, waiting for return.
* this is the safest, but can hang the main thread.
*/
_CancelMode TEXT DEFAULT 'DISCONNECT';
BEGIN
DidStuff := false;
FOR r IN
SELECT
sc.*,
d.ConnectionString,
COALESCE(parent.Started, tct.Started)
+ MaxRunTime < clock_timestamp() AS Cancel,
MaxRunTime,
tct.Task,
tct.SubTask
FROM ServerConnection sc
JOIN Database d USING(Environment, NodeId, Database)
JOIN TaskChainTask tct USING(TaskChainTaskId)
JOIN Task t USING(Task)
/* subtasks will use parent task start task start time for start time since
* the maximum runtime is established with the parent task
*/
LEFT JOIN TaskChainTask parent ON
tct.SubTask != ''
AND parent.Subtask = ''
AND parent.Task = tct.Task
AND parent.TaskChainId = tct.TaskChainId
AND parent.Environment = tct.Environment
AND parent.Database = tct.Database
AND parent.NodeId = tct.NodeId
WHERE (Active OR Cancelling)
LOOP
SELECT INTO
_Active,
_Cancelling
Active,
Cancelling
FROM ServerConnection
WHERE ConnectionName = r.ConnectionName;
/*
* Tasks can fail after the cursor is etablished due to triggers. Recheck
* the WHERE condition to ensure we have not alreday handled the connection.
*/
IF NOT _Active AND NOT _Cancelling
THEN
CONTINUE;
END IF;
_State := TaskRunning(r.ConnectionName);
IF _State != 'RUNNING'
THEN
DidStuff := true;
END IF;
IF _State = 'FINISHED' OR
(
_State = 'RUNNING'
AND _Cancelling
AND _CancelMode = 'IMMEDIATE'
)
THEN
SELECT INTO _State, _ErrorMessage TaskExecutionState, ErrorMessage
FROM ReapTask(r.ConnectionName);
IF _Cancelling
THEN
PERFORM Log(format(
'Reaped cancelling task %s/%s (id %s node %s) on connection %s',
r.Task,
r.SubTask,
r.TaskChainTaskId,
r.NodeId,
r.ConnectionName));
UPDATE ServerConnection SET
Cancelling = false,
TaskChainTaskId = NULL
WHERE ConnectionName = r.ConnectionName;
ELSE
PERFORM TaskFinished(
r.TaskChainTaskId,
CASE _State
WHEN 'FINISHED'::TaskExecutionState_t THEN 'SUCCESS_COMPLETION'
WHEN 'FAILED'::TaskExecutionState_t THEN 'FAILED_VIA_EXECUTION'
END::TaskResult_t,
_ErrorMessage)
WHERE _State IN ('FINISHED', 'FAILED');
END IF;
ELSIF _State = 'RUNNING'
AND _Cancelling
AND _CancelMode = 'DISCONNECT'
THEN
PERFORM Log(format(
'Force closed connection on '
'cancelling task %s/%s (id %s node %s) on connection %s',
r.Task,
r.SubTask,
r.TaskChainTaskId,
r.NodeId,
r.ConnectionName));
PERFORM Disconnect(r.ConnectionName);
ELSIF r.Cancel AND NOT _Cancelling
THEN
_ErrorMessage := format(
'Cancelling task due to exceeding maximum run time of %s',
r.MaxRunTime);
PERFORM TaskFinished(
r.TaskChainTaskId,
'FAILED_VIA_TIMEOUT',
_ErrorMessage);
END IF;
IF _State = 'RESET'
THEN
IF _Cancelling
THEN
UPDATE ServerConnection SET
Cancelling = false,
TaskChainTaskId = NULL
WHERE ConnectionName = r.ConnectionName;
END IF;
PERFORM Log(format(
'Got RESET state while checking connection status of %s',
r.ConnectionName));
PERFORM ConnectionDead(r.ConnectionName);
END IF;
END LOOP;
END;
$$ LANGUAGE PLPGSQL;
CREATE OR REPLACE FUNCTION CheckForAdmnistrativeActions() RETURNS BOOL AS
$$
DECLARE
_TaskChains TaskChain[];
_AdminstrativeQuery TEXT;
_AdminstrativeResult JSON;
BEGIN
SELECT INTO _AdminstrativeQuery Query
FROM AdministrativeCommand
WHERE Consumed IS NULL;
IF _AdminstrativeQuery IS NOT NULL
THEN
BEGIN
PERFORM Log('WARNING', format('Got administrative command %s',
_AdminstrativeQuery));
EXECUTE format('SELECT to_json(array_agg(q)) FROM (%s) q',
_AdminstrativeQuery) INTO _AdminstrativeResult;
UPDATE AdministrativeCommand SET
Consumed = clock_timestamp(),
Result = _AdminstrativeResult,
ErrorMessage = NULL;
EXCEPTION WHEN OTHERS THEN
PERFORM Log('WARNING', format(
'Got error %s when running administrative command %s',
SQLERRM,
_AdminstrativeQuery));
UPDATE AdministrativeCommand SET
Consumed = clock_timestamp(),
Result = NULL,
ErrorMessage = SQLERRM;
END;
END IF;
/* gather up the task chains to cancel */
SELECT INTO _TaskChains array_agg(tc) FROM TaskChain tc
WHERE
PendingAction IS NOT NULL
AND Finished IS NULL;
IF _TaskChains IS NULL OR array_upper(_TaskChains, 1) = 0
THEN
RETURN false;
END IF;
/* ... and cancel them. */
PERFORM CancelTaskChain(TaskChainId, PendingAction = 'PAUSE')
FROM
(
SELECT * FROM unnest(_TaskChains) tc
) q;
UPDATE TaskChain tc
SET
PendingAction = NULL,
Finished = CASE WHEN tc.PendingAction = 'CANCEL' THEN clock_timestamp() END,
ErrorMessage = CASE WHEN tc.PendingAction = 'CANCEL'
THEN 'Administrative Cancel' END,
Paused = tc.PendingAction = 'PAUSE',
PauseStartTime =
CASE WHEN tc.PendingAction = 'CANCEL' THEN clock_timestamp() END
WHERE tc.TaskChainId IN (SELECT TaskChainId FROM unnest(_TaskChains));
RETURN true;
END;
$$ LANGUAGE PLPGSQL;
/*
* Load task configurations from a structurd json. As much as possible,
* the changes will be reconciled in so that RI deletions do not have to be
* cascaded.
*
* XXX: Should loaded configurations be tagged with some some kind of name
* so that downstream adjusttments are simpler to make? this shold
* also facilitate saving independent sections of the configuration that
* are managed from different areas.
*
* XXX: Can this be allowed to run while any impacted task chains are in
* progress?
*
* XXX: After any database/server changes are made, it's probably a good idea
* to flush and reboot any server connections. What if a connection_string
* changes?
*/
CREATE OR REPLACE FUNCTION LoadConfiguration(
_Configuration JSON) RETURNS VOID AS
$$
DECLARE
r RECORD;
t JSONB;
BEGIN
WITH n AS
(
SELECT * FROM json_populate_recordset(
NULL::Environment,
(_Configuration)->'environments')
)
INSERT INTO Environment
SELECT * FROM n
WHERE NOT EXISTS (
SELECT 1 FROM Environment e2
WHERE e2.Environment = n.Environment
);
/* Reconcile servers from input json. We don't want to blow out the table
* for RI reasons. Also try to bring in defaults.
*/
CREATE TEMP TABLE ServerDefaults (LIKE Server INCLUDING DEFAULTS) ON COMMIT DROP;
ALTER TABLE ServerDefaults ALTER Server DROP NOT NULL;
INSERT INTO ServerDefaults DEFAULT VALUES;
CREATE TEMP TABLE servers (LIKE Server);
t := to_jsonb((SELECT sd FROM ServerDefaults sd));
FOR r IN SELECT json_array_elements((_Configuration)->'servers') j
LOOP
INSERT INTO servers SELECT * FROM jsonb_populate_record(
NULL::Server,
t || jsonb_strip_nulls(r.j::jsonb));
END LOOP;
INSERT INTO Server SELECT * FROM servers s
WHERE NOT EXISTS (
SELECT 1 FROM Server WHERE Server.Server = s.Server
);
UPDATE Server s SET
TaskClientThreads = s_new.TaskClientThreads,
ServerType = s_new.ServerType
FROM Servers s_new
WHERE s.Server = s_new.Server;
DELETE FROM TaskChainConfiguration WHERE TaskChainConfigurationName IN (
SELECT
json_array_elements(
_Configuration->'taskChainConfigurations')->>'taskchainconfigurationname'
);
/* Reconcile databases */
CREATE TEMP TABLE databases ON COMMIT DROP AS
SELECT * FROM json_populate_recordset(
NULL::Database,
(_Configuration)->'databases');
INSERT INTO Database SELECT * FROM databases d
WHERE NOT EXISTS (
SELECT 1 FROM Database WHERE
Database.Environment = d.Environment
AND Database.Database = d.Database
AND Database.NodeId = d.NodeId
);
UPDATE Database d SET
Server = d_new.Server,
ConnectionString = d_new.ConnectionString
FROM databases d_new
WHERE
d.Environment = d_new.Environment
AND d.Database = d_new.Database
AND d.NodeId = d_new.NodeId;
/* Reconcile tasks. */
/* insert task in such a way as to preserve defaults */
CREATE TEMP TABLE TaskDefaults (LIKE Task INCLUDING DEFAULTS) ON COMMIT DROP;
ALTER TABLE TaskDefaults ALTER Task DROP NOT NULL;
INSERT INTO TaskDefaults DEFAULT VALUES;
CREATE TEMP TABLE tasks (LIKE Task);
t := to_jsonb((SELECT td FROM TaskDefaults td));
FOR r IN SELECT json_array_elements((_Configuration)->'tasks') j
LOOP
INSERT INTO tasks SELECT * FROM jsonb_populate_record(
NULL::Task,
t || jsonb_strip_nulls(r.j::jsonb));
END LOOP;
/* Clean out tasks that were attached to databases we are loading but are
* no longer attached to them
*/
DELETE FROM Task
WHERE
Database IN (SELECT Database FROM Databases)
AND Task NOT IN (SELECT Task FROM Tasks);
INSERT INTO Task SELECT * FROM tasks t
WHERE NOT EXISTS (
SELECT 1 FROM Task
WHERE Task.Task = t.Task
);
UPDATE Task SET
TaskBody = t_new.TaskBody,
RunSchedule = t_new.RunSchedule,
Database = t_new.Database,
StartTime = t_new.StartTime,
EndTime = t_new.EndTime,
MaxRunTime = t_new.MaxRunTime,
IsConcurrent = t_new.IsConcurrent,
IsTaskChainConfiguration = t_new.IsTaskChainConfiguration,
TaskOrder = t_new.TaskOrder,
SubTasksMustComplete = t_new.SubTasksMustComplete
FROM tasks t_new
WHERE Task.Task = t_new.Task;
/* reconcile subtasks */
CREATE TEMP TABLE subtasks ON COMMIT DROP AS
SELECT * FROM json_populate_recordset(
NULL::SubTask,
(_Configuration)->'subTasks');
/* Get rid of subtasks that are attached to any task we are inserting
* but are not in the subtasks being inserted.
*/
DELETE FROM SubTask
WHERE
Task IN (SELECT Task FROM subtasks)
AND SubTask NOT IN (SELECT SubTask FROM subtasks);
INSERT INTO SubTask SELECT * FROM subtasks s
WHERE NOT EXISTS (
SELECT 1 FROM SubTask
WHERE SubTask.Task = s.Task
AND SubTask.SubTask = s.SubTask
);
UPDATE SubTask s SET
TaskBody = s_new.TaskBody,
RunSchedule = s_new.RunSchedule
FROM subtasks s_new
WHERE
s.Task = s_new.Task
AND s.SubTask = s_new.SubTask;
/* Reconcile TaskDependencies */
CREATE TEMP TABLE taskdependencies ON COMMIT DROP AS SELECT *
FROM json_populate_recordset(
NULL::TaskDependency,
(_Configuration)->'taskDependencies');
DELETE FROM TaskDependency WHERE
(
ParentTask IN (SELECT Task FROM tasks)
OR ChildTask IN (SELECT Task FROM tasks)
)
AND (ParentTask, ChildTask) NOT IN (
SELECT ParentTask, ChildTask FROM taskdependencies
);
INSERT INTO TaskDependency
SELECT * FROM taskdependencies td
WHERE NOT EXISTS (
SELECT 1 FROM TaskDependency
WHERE
TaskDependency.ParentTask = td.ParentTask
AND TaskDependency.ChildTask = td.ChildTask
);
UPDATE TaskDependency td SET
AllNodes = td_new.AllNodes,
CanRun = td_new.CanRun
FROM taskdependencies td_new
WHERE
td.ChildTask = td_new.ChildTask
AND td.ParentTask = td_new.ParentTask;
/* update named task chain configuraitonns. Since there is no RI on this
* table, simply dump and reloading should do the tick.
*/
CREATE TEMP TABLE tcc (LIKE TaskChainConfiguration
INCLUDING DEFAULTS) ON COMMIT DROP;
ALTER TABLE tcc ALTER TaskChainConfigurationName DROP NOT NULL;
INSERT INTO tcc DEFAULT VALUES;
t := to_jsonb((SELECT tcc FROM tcc));
FOR r IN SELECT json_array_elements((_Configuration)->'taskChainConfigurations') j
LOOP
INSERT INTO TaskChainConfiguration SELECT * FROM jsonb_populate_record(
NULL::TaskChainConfiguration,
t || jsonb_strip_nulls(r.j::jsonb));
END LOOP;
INSERT INTO TaskChainConfigurationTask SELECT * FROM json_populate_recordset(
NULL::TaskChainConfigurationTask,
(_Configuration)->'taskChainConfigurationTasks');
END;
$$ LANGUAGE PLPGSQL;
/*
* Saves some or all of the task configuration.
*
* XXX: Tagged configurations is not a completely developed concept.
*/
CREATE OR REPLACE FUNCTION SaveTaskConfiguration() RETURNS JSON AS
$$
BEGIN
END;
$$ LANGUAGE PLPGSQL;
CREATE OR REPLACE FUNCTION TaskChainUpdate() RETURNS TRIGGER AS
$$
DECLARE
_FailedTasks INT;
BEGIN
IF OLD.Finished IS NULL AND NEW.Finished IS NOT NULL
THEN
SELECT INTO _FailedTasks COUNT(*)
FROM TaskChainTask
WHERE
DidSucceed(TaskResult) IS DISTINCT FROM true
AND TaskChainId = NEW.TaskChainId;
IF _FailedTasks = 0
THEN
NEW.DidSucceed = true;
ELSE
NEW.DidSucceed = false;
NEW.ErrorMessage = format('%s tasks did not complete', _FailedTasks);
END IF;
END IF;
RETURN NEW;
END;
$$ LANGUAGE PLPGSQL;
DROP TRIGGER IF EXISTS OnTaskChainUpdate ON TaskChain;
CREATE TRIGGER OnTaskChainUpdate AFTER UPDATE ON TaskChain
FOR EACH ROW EXECUTE PROCEDURE TaskChainUpdate();
CREATE OR REPLACE FUNCTION CheckAndFailRelatedSubTasks(
_SubTask TaskChainTask) RETURNS VOID AS
$$
DECLARE
_SubTasksMustComplete BOOL;
BEGIN
/* Task finished normally. */
IF DidSucceed(_SubTask.TaskResult)
THEN
RETURN;
END IF;
/* The task failed. What is the controlling task behavior? */
SELECT INTO _SubTasksMustComplete SubTasksMustComplete
FROM Task
WHERE Task = _SubTask.Task;
/* all other subtasks running or in queue should be failed if reason for
* this check was triggered by 'FAILED_VIA_TIMEOUT'
*/
IF (NOT _SubTasksMustComplete AND _SubTask.TaskResult != 'FAILED_VIA_TIMEOUT')
THEN
/* We don't have to worry about other subtasks */
RETURN;
END IF;
/* uh oh. A subtask failed. Any other subtasks that are running
* should be cancelled and the rest of them failed
*/
PERFORM TasksFinished(
array_agg(sc.TaskChainTaskId),
'FAILED_VIA_CANCEL'::TaskResult_t,
'Cancelled while in progress due to failure of other tasks')
FROM ServerConnection sc
WHERE
Active
AND TaskChainTaskId IN (
SELECT tct.TaskChainTaskId
FROM TaskChainTask tct
WHERE
TaskChainId = _SubTask.TaskChainId
AND NodeId = _Subtask.NodeId
AND Task = _SubTask.Task
AND SubTask != _SubTask.SubTask
AND IsSubTask(_SubTask.SubTask)
);
/* mark any yet to be started subtasks as failed. */
PERFORM TasksFinished(TaskChainTasks, TaskResult, ErrorMessage)
FROM
(
SELECT
array_agg(tct.TaskChainTaskId) AS TaskChainTasks,
'FAILED_VIA_SUBTASKS'::TaskResult_t AS TaskResult,
format(
'Auto cancelling as task %s/%s (id %s node %s) failed',
_SubTask.Task,
_SubTask.SubTask,
_SubTask.TaskChainTaskId,
_SubTask.NodeId) AS ErrorMessage
FROM TaskChainTask tct
WHERE
IsSubTask(tct.SubTask)
AND tct.Task = _SubTask.Task
AND tct.TaskChainId = _SubTask.TaskChainId
AND tct.Finished IS NULL
AND tct.NodeId = _SubTask.NodeId
AND SubTask != _SubTask.SubTask
GROUP BY 2, 3
) q;
END;
$$ LANGUAGE PLPGSQL;
/*
* Called when (and only when) the last subtask has completed
*/
CREATE OR REPLACE FUNCTION FinishParentTaskFromSubTask(
_SubTask TaskChainTask) RETURNS VOID AS
$$
DECLARE
_ParentTaskChainTaskId BIGINT;
_ParentFinished TIMESTAMPTZ;
_Failures INT;
BEGIN
/* If the parent task is already marked failed due to cascaded dependency
* failure, we don't need to record everything. However, if the failure
* was bubbled up from one or more subtasks failing, the parent task would
* not have been marked failed yet.
*/
SELECT
TaskChainTaskId,
Finished
INTO
_ParentTaskChainTaskId,
_ParentFinished
FROM TaskChainTask p
WHERE
TaskChainId = _SubTask.TaskChainId
AND Task = _SubTask.Task
AND NOT IsSubTask(SubTask)
AND p.NodeId = _SubTask.NodeId;
IF _ParentTaskChainTaskId IS NULL
THEN
PERFORM Log('ERROR',
format(
'Unable to resolve parent task for %',
to_json(_SubTask)));
END IF;
IF _ParentFinished IS NOT NULL
THEN
RETURN;
END IF;
SELECT count(*) AS FailedTasks INTO _Failures
FROM TaskChainTask p
WHERE
p.TaskChainId = _SubTask.TaskChainId
AND p.Task = _SubTask.Task
AND p.NodeId = _SubTask.NodeId
AND IsSubTask(p.SubTask)
AND NOT DidSucceed(TaskResult);
PERFORM TaskFinished(
tct.TaskChainTaskId,
CASE WHEN _Failures = 0
THEN 'SUCCESS_VIA_SUBTASKS'
ELSE 'FAILED_VIA_SUBTASKS'
END::TaskResult_t,
CASE WHEN _Failures > 0
THEN format('%s subtasks failed', _Failures)
END)
FROM TaskChainTask tct
WHERE
TaskChainId = _SubTask.TaskChainId
AND Task = _SubTask.Task
AND SubTask = ''
AND tct.NodeId = _SubTask.NodeId
AND NOT EXISTS (
SELECT 1 FROM TaskChainTask p
WHERE
p.TaskChainId = _SubTask.TaskChainId
AND p.Task = _SubTask.Task
AND IsSubTask(p.SubTask)
AND p.Finished IS NULL
AND p.NodeId = _SubTask.NodeId
);
END;
$$ LANGUAGE PLPGSQL;
/*
* When a task completes, some tasks can become eligible to run as maintained
* by the 'eligible' flag. Mark them ready to go
*/
CREATE OR REPLACE FUNCTION SetDependentTasksEligible(
_Task TaskChainTask) RETURNS VOID AS
$$
BEGIN
UPDATE TaskChainTask tct SET Eligible = true
FROM TaskDependency td
WHERE
tct.TaskChainId = _Task.TaskChainId
/* don't mess with subtasks */
AND tct.SubTask = ''
/* is task dependent on currently completing task */
AND tct.Task = td.ChildTask
AND td.ParentTask = _Task.Task
/* guard against tasks auto cancelled */
AND tct.Finished IS NULL
/* no point in setting it if it is already set */
AND tct.Eligible IS NOT TRUE
/* if the dependency is 'node local', we only mark eligibility for the
* local node, else all nodes; This also applies to the negative
* dependency check.
*/
AND (td.AllNodes OR _Task.NodeId = tct.NodeId)
AND (DidSucceed(_Task.TaskResult) OR td.CanRun)
/* does the task being examined have any ancestors that are incomplete */
AND NOT EXISTS
(
SELECT 1 FROM TaskChainTask parent
JOIN TaskDependency td ON parent.Task = td.ParentTask
WHERE
parent.TaskChainId = _Task.TaskChainId
AND td.ChildTask = tct.Task
AND parent.Finished IS NULL
AND (td.AllNodes OR parent.NodeId = tct.NodeId)
);
END;
$$ LANGUAGE PLPGSQL;
CREATE OR REPLACE FUNCTION HandleTaskFailure(
_Task TaskChainTask) RETURNS VOID AS
$$
BEGIN
/*
* The task failed, all direct dependencies of the task must also fail.
* if the allnodes flag is set, that dependency will additionally cross
* so that other nodes will fail the task well.
*/
PERFORM TasksFinished(TaskChainTasks, TaskResult, ErrorMessage)
FROM
(
SELECT
array_agg(tct.TaskChainTaskId) AS TaskChainTasks,
'FAILED_VIA_DEPENDENCY'::TaskResult_t AS TaskResult,
format(
'Auto cancelling as Task %s (id %s node %s) failed',
_Task.Task,
_Task.TaskChainTaskId,
_Task.NodeId) AS ErrorMessage
FROM TaskChainTask tct
JOIN TaskDependency td ON
td.ParentTask = _Task.Task
AND td.ChildTask = tct.Task
WHERE
td.ChildTask = tct.Task
AND tct.TaskChainId = _Task.TaskChainId
AND (tct.NodeId = _Task.NodeId OR td.AllNodes)
AND tct.Finished IS NULL
AND tct.SubTask = ''
AND td.CanRun IS NOT TRUE
GROUP BY 2, 3
) q;
/* If the parent task has failed via subtasks, there is no point to
* mark subtasks for failure.
*/
IF _Task.TaskResult IN ('SUCCESS_VIA_SUBTASKS', 'FAILED_VIA_SUBTASKS')
THEN
RETURN;
END IF;
/* Subtasks are automatically marked failed if the parent task fails */
PERFORM TasksFinished(TaskChainTasks, TaskResult, ErrorMessage)
FROM
(
SELECT
array_agg(tct.TaskChainTaskId) AS TaskChainTasks,
'FAILED_VIA_PARENT_TASK'::TaskResult_t AS TaskResult,
format(
'Auto cancelling as Parent Task %s (id %s node %s) failed',
_Task.Task,
_Task.TaskChainTaskId,
_Task.NodeId) AS ErrorMessage
FROM TaskChainTask tct
WHERE
tct.SubTask != ''
AND tct.Task = _Task.Task
AND tct.TaskChainId = _Task.TaskChainId
AND tct.Finished IS NULL
AND tct.NodeId = _Task.NodeId
GROUP BY 2, 3
) q;
END;
$$ LANGUAGE PLPGSQL;
CREATE OR REPLACE FUNCTION TaskChainTaskUpdate() RETURNS TRIGGER AS
$$
DECLARE
d TIMESTAMPTZ;
BEGIN
/* If the task has started, all subtasks (if any) become eligible */
IF NOT IsSubTask(NEW.SubTask)
AND OLD.Started IS NULL
AND NEW.Started IS NOT NULL
THEN
UPDATE TaskChainTask SET Eligible = true
WHERE
TaskChainId = NEW.TaskChainId
AND Task = NEW.Task
AND NodeId = NEW.NodeId
AND IsSubTask(SubTask);
END IF;
/* Is this task newly finishing? if not, punt */
IF NOT (NEW.Finished IS NOT NULL AND OLD.Finished IS NULL)
THEN
RETURN NEW;
END IF;
IF NEW.Eligible
THEN
/* If the task is finished, clear the eligibility flag */
NEW.Eligible = false;
END IF;
/* If a subtask completes, look for an uncompleted subtask at same level. If
* none found, mark the task complete as well.
*/
IF IsSubTask(NEW.SubTask)
THEN
PERFORM 1
FROM TaskChainTask p
WHERE
p.TaskChainId = NEW.TaskChainId
AND p.Task = NEW.Task
AND p.NodeId = NEW.NodeId
AND IsSubTask(p.SubTask)
AND Started IS NOT NULL
AND Finished IS NULL LIMIT 1;
IF FOUND
THEN
/* There is at least one incomplete subtask. Did we fail and does that
* failure requre handling?
*/
PERFORM CheckAndFailRelatedSubTasks(NEW);
ELSE
PERFORM FinishParentTaskFromSubTask(NEW);
END IF;
RETURN NEW;
END IF;
/* If the task is regular task and completes, there is an important processing
* descision to be made: failed tasks can (depending on how dependency is
* configured) cascade the failure to dependent tasks while successfully
* completed tasks may cause other dependent tasks to beecome eligible to
* be run.
*/
IF NOT DidSucceed(NEW.TaskResult)
THEN
PERFORM HandleTaskFailure(NEW);
END IF;
PERFORM SetDependentTasksEligible(NEW);
RETURN NEW;
END;
$$ LANGUAGE PLPGSQL;
DROP TRIGGER IF EXISTS OnTaskChainTaskUpdate ON TaskChainTask;
CREATE TRIGGER OnTaskChainTaskUpdate AFTER UPDATE ON TaskChainTask
FOR EACH ROW EXECUTE PROCEDURE TaskChainTaskUpdate();
CREATE OR REPLACE FUNCTION CleanupAfterRestart() RETURNS VOID AS
$$
BEGIN
/* This is not the same backend as the one that initialized the dblink
* connections. There is no chance of recovery of task execution state.
*/
IF pg_backend_pid() != (SELECT MainProcessPid FROM GlobalControl)
THEN
TRUNCATE ServerConnection;
END IF;
/* if the dblink connections are still active, grab any reponses that are
* still outstanding. This will only work if we restarted frome the
* same database session that initiated the tasks (which is a bit of a
* hail mary).
*/
BEGIN
PERFORM CheckForFinishedTasks();
EXCEPTION WHEN OTHERS THEN
PERFORM Log('WARNING', format(
'Attempted to recover old tasks, but was not able to do so via %s',
SQLERRM));
END;
/* at this point, any task that was assume to be running before startup
* must now be considered failed. Likewise, any running task chains will
* be assumed dead.
*
* XXX: There may be a more clever way to try to deal with this, but there
* are a lot of complex failure modes and nuking everything is by
* far the simplest approach.
*/
ALTER TABLE TaskChainTask DISABLE TRIGGER OnTaskChainTaskUpdate;
PERFORM TaskFinished(
tct.TaskChainTaskId,
'FAILED_VIA_RESTART',
'Presumed failed due to spurious pgtask restart')
FROM TaskChainTask tct
WHERE
tct.Finished IS NULL
AND EXISTS (
SELECT 1 FROM TaskChainTask tct2
WHERE
tct2.TaskChainId = tct.TaskChainId
AND tct2.Started IS NOT NULL
);
UPDATE TaskChain tc SET
Finished = now(),
DidSucceed = false,
ErrorMessage = 'Presumed failed due to spurious pgtask restart'
WHERE
Finished IS NULL
AND Started IS NOT NULL
AND EXISTS (
SELECT 1 FROM TaskChainTask tct
WHERE
tct.TaskChainId = tc.TaskChainId
AND tct.Started IS NOT NULL
);
ALTER TABLE TaskChainTask ENABLE TRIGGER OnTaskChainTaskUpdate;
END;
$$ LANGUAGE PLPGSQL;
CREATE OR REPLACE FUNCTION CheckForTimedOutIdleTasksInternal(
_TaskChainId BIGINT) RETURNS VOID AS
$$
BEGIN
SET LOCAL enable_nestloop to false;
PERFORM
TasksFinished(
array_agg(tct.TaskChainTaskId),
'FAILED_VIA_TIMEOUT',
'Cancelling idle tasks that have timed out')
FROM TaskChainTask tct
JOIN TaskChain tc USING(TaskChainId)
JOIN Task USING(Task)
JOIN TaskChainTask parent ON
tct.SubTask != ''
AND parent.Subtask = ''
AND parent.Task = tct.Task
AND parent.TaskChainId = _TaskChainId
AND tct.TaskChainId = _TaskChainId
AND parent.NodeId = tct.NodeId
WHERE
tct.Started IS NULL
AND tct.Finished IS NULL
AND parent.Started + MaxRunTime < now();
END;
$$ LANGUAGE PLPGSQL;
CREATE OR REPLACE FUNCTION CheckForTimedOutIdleTasks() RETURNS VOID AS
$$
DECLARE
_TaskChainId BIGINT;
BEGIN
FOR _TaskChainId IN
SELECT TaskChainId
FROM TaskChain tc
WHERE
tc.Started IS NOT NULL
AND tc.Finished IS NULL
LOOP
PERFORM CheckForTimedOutIdleTasksInternal(_TaskChainId);
END LOOP;
END;
$$ LANGUAGE PLPGSQL;
CREATE OR REPLACE FUNCTION AuditRunningTasks(
_AuditInterval INTERVAL,
_LastAudit TIMESTAMPTZ) RETURNS VOID AS
$$
BEGIN
IF _LastAudit + _AuditInterval > now()
THEN
RETURN;
END IF;
INSERT INTO ActiveTaskAudit
SELECT
now(),
vat.*
FROM vw_ActiveTasks vat;
UPDATE GlobalControl SET LastAudit = now();
END;
$$ LANGUAGE PLPGSQL;
CREATE OR REPLACE FUNCTION ProcessTasks() RETURNS BOOL AS
$$
DECLARE
_DidStuff BOOL;
_RanTasks BOOL;
_TaskChainTaskId BIGINT;
gc GlobalControl;
BEGIN
SELECT INTO gc * FROM GlobalControl;
/* record tasks that are running on semi-frequent basis */
PERFORM AuditRunningTasks(gc.AuditInterval, gc.LastAudit);
/* check for tasks to cancel, etc */
_DidStuff := CheckForAdmnistrativeActions();
/* If there were any cancels/pauses/etc, exit out in order to flush and
* commit the transaction so that no further processing is done. This is
* done to avoid conflicting trigger effects.
*/
IF _DidStuff
THEN
RETURN true;
END IF;
/* Are there any tasks that were waiting and timed out? */
IF COALESCE(gc.IdleTaskTimeoutLastCheck, now())
+ gc.IdleTaskTimeoutInterval > now()
THEN
PERFORM CheckForTimedOutIdleTasks();
UPDATE GlobalControl SET IdleTaskTimeoutLastCheck = now();
END IF;
PERFORM CheckForTimedOutIdleTasks();
/* reap tasks that resolved in the background */
_DidStuff := CheckForFinishedTasks();
FOR _TaskChainTaskId IN SELECT * FROM GetTasks()
LOOP
_RanTasks := RunTask(_TaskChainTaskId);
_DidStuff := _DidStuff OR _RanTasks;
END LOOP;
/* if there is a queued task chain, start it */
UPDATE TaskChain tc SET Started = clock_timestamp()
WHERE
tc.Started IS NULL
AND tc.TaskChainConfigurationName IS NOT NULL
AND tc.Finished IS NULL
AND NOT EXISTS (
/* is there a queued or running task ahead of me? */
SELECT 1 FROM TaskChain tc2
WHERE
tc2.TaskChainConfigurationName = tc.TaskChainConfigurationName
AND tc2.Environment = tc.Environment
AND tc2.TaskChainId < tc.TaskChainId
AND tc2.Finished IS NULL
);
RETURN _DidStuff;
END;
$$ LANGUAGE PLPGSQL;
CREATE OR REPLACE PROCEDURE MainLoop() AS
$$
DECLARE
_DidStuff BOOL;
_MessagedNothing BOOL DEFAULT false;
gc GlobalControl;
BEGIN
SELECT INTO gc * FROM GlobalControl;
PERFORM Log('Initializaton complete. Entering main server loop');
LOOP
_DidStuff := false;
SELECT INTO gc * FROM GlobalControl;
_DidStuff := ProcessTasks();
/* flush transaction */
COMMIT;
IF NOT _DidStuff
THEN
IF EXISTS(
SELECT 1 FROM ServerConnection WHERE Active)
THEN
PERFORM pg_sleep(gc.MainLoopSleepSecondsActive);
ELSE
IF NOT _MessagedNothing
THEN
PERFORM Log('Nothing to do...sleeping');
_MessagedNothing := true;
END IF;
PERFORM pg_sleep(gc.MainLoopSleepSecondsIdle);
END IF;
ELSE
_MessagedNothing := false;
END IF;
END LOOP;
END;
$$ LANGUAGE PLPGSQL;
/*
* Main server process.
*/
CREATE OR REPLACE PROCEDURE Main(
_ForceRestart BOOL DEFAULT false) AS
$$
DECLARE
gc GlobalControl;
_Locked BOOL;
s Server;
_TaskChainTaskId BIGINT;
_MessagedNothing BOOL DEFAULT false;
tct TaskChainTask;
BEGIN
/* There must only be one :-D */
SELECT INTO _Locked pg_try_advisory_lock(0);
IF NOT _Locked
THEN
IF NOT _ForceRestart
THEN
RETURN;
END IF;
SELECT INTO gc * FROM GlobalControl;
PERFORM Log(
'Attemping to kill old main process and take over (good luck!)');
PERFORM pg_terminate_backend(gc.MainProcessPid);
PERFORM pg_advisory_lock(0);
END IF;
/* set start time so that uptime can be determined */
UPDATE GlobalControl SET StartTime = clock_timestamp();
PERFORM Log('pgtask advisory lock acquired. Starting Initializaton');
SELECT INTO gc * FROM GlobalControl;
/* If the exit was unclean, there may be tasks left in indeterminate state.
* Do some housekeeping to deal with that.
*/
IF gc.GracefulShutdown IS DISTINCT FROM TRUE
THEN
PERFORM CleanupAfterRestart();
END IF;
/* Mark the main process pid so that we can guard future restart operations */
UPDATE GlobalControl SET
MainProcessPid = pg_backend_pid(),
GracefulShutdown = FALSE;
/* Create our connection table. This is temporary as upon restart there is
* no reason to believe any connection tracking data is correct.
*/
PERFORM CreateDblinkConnections();
CALL MainLoop();
END;
$$ LANGUAGE PLPGSQL;
/*
* Create subtask based on the subtask json for a given parenttask
* Used by subtask gnerator tasks to create subtasks at runtime
*/
CREATE OR REPLACE FUNCTION CreateTaskChainSubtask(
_TaskChainTaskId BIGINT, /* TaskChainTaskId for prep Task */
_Task TEXT , /* Parent Task name */
_SubTasks JSONB
) RETURNS VOID AS
$$
DECLARE
_TaskChainTaskRec RECORD;
BEGIN
/* get valid taskchaintask record */
SELECT tct.* INTO _TaskChainTaskRec
FROM TaskChainTask tct
JOIN TaskChainTask ptct
ON tct.taskChainId = ptct.taskChainId
AND tct.environment = ptct.environment
AND tct.database = ptct.database
AND tct.nodeid = ptct.nodeId
WHERE ptct.TaskChainTaskId = _TaskChainTaskId
AND tct.Task = _Task
AND tct.Started IS NULL
AND tct.Finished IS NULL;
IF FOUND THEN
PERFORM Log('Adding Subtasks via subtask generator for TaskChainTaskId:' ||
_TaskChainTaskId::TEXT);
INSERT INTO TaskChainTask(
TaskChainId,
Environment,
Database,
NodeId,
Task,
SubTask,
TaskBody,
Eligible)
SELECT
_TaskChainTaskRec.TaskChainId,
_TaskChainTaskRec.Environment,
_TaskChainTaskRec.Database,
_TaskChainTaskRec.NodeId,
_TaskChainTaskRec.Task,
s.SubTask,
s.TaskBody,
true
FROM jsonb_populate_recordset(
NULL::subTask,
_SubTasks->'subTasks') s;
ELSE
PERFORM Log('WARNING','Failed to Add Subtasks via subtask generator
for TaskChainTaskId:' || _TaskChainTaskId::TEXT);
END IF;
END;
$$ LANGUAGE PLPGSQL;