What Job Scheduler Can Do in Postgres Pro

The scheduler of tasks (scheduler) not at all times was considered an obligatory tool in the world of databases. It all depended on the purpose and origin of the DBMS. Classic commercial DBMS (Oracle, DB2, MS SQL) is absolutely impossible to imagine without a scheduler. On the other hand, it is hard to imagine a potential MongoDB user who would refuse to choose this fashionable NoSQL DBMS due to the lack of a scheduler. (By the way, the term “task scheduler” in the Russian context is used by the DBMS to distinguish it from the query planner - query planner, for the sake of brevity we will call it the scheduler here).

PostgreSQL, being an Open Source and having absorbed the traditions of the community with a DIY lifestyle (“do it yourself”), nowadays regularly claims to be at least a deputy commercial DBMS. It automatically follows from this that PostgreSQL is simply required to have a scheduler, and that this scheduler should be convenient for the database administrator and for the user. And that it is desirable to reproduce the fully functional capabilities of commercial DBMSs, although it would be nice to add something of your own.

The need for a scheduler is most evident when working with the base in industrial operation. To the developer, who was allocated a server for experimenting with the database, the scheduler, in general, is useless: if necessary, he will plan all the necessary operations by the means of the OS (cron or at in Unix). But he will not be allowed to enter the working base in a serious company for a cannon shot. There is an important administrative nuance, that is, it is no longer a nuance, but a serious, if not decisive reason: the database administrator and system administrator are not just different people with different tasks. It is possible that they belong to different divisions of the company, and maybe even sit on different floors. Ideally, the database administrator supports its viability and monitors its evolution, and the system administrator's area of ​​responsibility is the viability of the OS and network.

Therefore, the database administrator must have a tool to perform the necessary set of possible work on the server. It is not without reason that the materials about the Oracle scheduler say that "Oracle Scheduler eliminates the need to use OS-specific task schedulers (cron, at) for different platforms when building a database-centric application." That is, the database administrator can do everything, especially since it is difficult to imagine an Oracle administrator who is not guided by the OS mechanisms. He does not have to run to the system administrator every time or write letters to him when routine operations are required by OS tools.

Here are the scheduler requirements typical of commercial DBMSs such as Oracle, DB2, MS SQL: The

scheduler must be able to

• start scheduled work,
• control the execution of work, be able to remove tasks, if necessary,
• run tasks in a limited period of time (in the window),
• to build a sequence of tasks (the next starts to be executed after the previous)
• Be able to complete several requests in one transaction,
• run a task defined in one database on several,
• use the (main) features of the OS,
• notify the administrator if some tasks from the schedule were not completed,
• perform one-time tasks.

The last point, as it were, is not obvious: there are quite a few other standard tools besides the scheduler that can run one-time tasks. But this is not an entirely normal execution mode. For example, detached job mode: we are talking about a task that, as it were, is disconnected (temporarily or permanently) from the process that caused it. Having done the work, the disconnected process can again contact the process that started it (sending it a signal of successful or unsuccessful completion), inform it of the result, or write the result (to a file or database table). Some DBMS schedulers can stop and start the DBMS itself (we did not set such a task).

PostgreSQL and its agent

The tasks set can be solved in different ways: “outside” and “inside” the DBMS itself. The most serious attempt to make a fully functional scheduler is pgAgent, distributed with pgAdmin III / IV. In the commercial version - in the EnterpriseDB distribution kit - it is integrated into the pgAdmin GUI and can be used cross-platform.

pgAgent can:

• run tasks
• run sequences of tasks consisting of SQL scripts (including on different databases) and / or shell / batch scripts,
• set non-working windows (for example, DO NOT perform some action on weekends).

This scheduler works as a PostgreSQL extension, but it does not execute tasks “inside” the DBMS, but creates its own, “external” daemons.

There are disadvantages to this approach. Among them are important:

All tasks launched by pgAgent will be executed with the rights of the user who launched the agent. SQL queries will be executed with the rights of the user connected to the database. Shell scripts will be executed with the rights of the user on whose behalf the pgAgent daemon (or service on Windows) is running. Therefore, for security, you will have to control users who can create and run tasks. In addition, the password cannot be included in the connection string, since on Unix it will be visible in the output of the ps command and in the database start script, and on Windows it will be stored in the registry as unencrypted text.
(from the pgAdmin 4 1.6 documentation ).

In this solution, pgAgent polls the database server at specified intervals (since information about the work is stored in the database tables): are there any jobs. Therefore, if for some reason the agent will not work at the moment when the work should start, it will not start until the agent starts working.

In addition, any connection to the server consumes a pool of possible connections, the maximum number of which is determined by the configuration parameter max_connections. If the agent spawns many processes and the administrator has not been vigilant, this can be a problem.

Creating a scheduler fully integrated into the DBMS (“inside” the DBMS) eliminates these problems. And it is especially convenient for those users who are accustomed to minimalist interfaces for accessing the database, such as psql.

pgpro_scheduler and its schedule

At the end of 2016, Postgres Professional began to create its own scheduler, fully integrated into the DBMS. Now it is used by customers and is documented in detail. The scheduler was created as an extension (add-on module), called pgpro_scheduler, and is shipped as part of the commercial version of Postgres Pro Enterprise starting with its first version. Developer - Vladimir Ershov.

When installing it in the DBMS configuration files, you must remember to include it in the configuration file shared_preload_libraries = 'pgpro_scheduler'. Having installed the extension (CREATE EXTENSION pgpro_scheduler;), you need to enable it with a line in the configuration file (schedule.enabled = on) and let it list which databases will fall under the action of the scheduler (for example schedule.database = 'database1,database2').

From the very beginning it was decided to create pgpro_scheduler in a modern style, organic for the company - with the configuration written in JSON. This is convenient, for example, for creators of Web services who can integrate the scheduler into their applications. But for those who do not want to use JSON, there are functions that take parameters in the form of ordinary variables. The scheduler comes with the DBMS distribution kit and is cross-platform.

pgpro_scheduler does not start external daemons or services, but creates background worker processes that are children of the postmaster - background processes. The number of "workers" is specified in the pgpro_scheduler configuration, but is limited by the general server configuration. The scheduler actually receives the most common SQL commands, without any restrictions, so you can run functions in any Postgres language available. If several SQL queries are included in the JSON structure, then they (when following a certain syntax) can be executed within a single transaction:

SELECT schedule.create_job('{"commands": [ "SELECT 1", "SELECT 2", "SELECT 3"], "cron": "23 23 */2 * *" }');

this is equivalent to:

SELECT schedule.create_job('{"commands": [ "SELECT 1", "SELECT 2", "SELECT 3"], "cron": "23 23 */2 * *","use_same_transaction": true}');
and if each query is in its own transaction, then:

SELECT schedule.create_job('{"commands": [ "SELECT 1", "SELECT 2", "SELECT 3" ], "cron": "23 23 */2 * *" }');- that is, without the last parameter, by default.

Suppose an error occurs in the second command on the list (of course, inSELECT 2it is unlikely to happen, but imagine some kind of "dumb" request). If executed in one transaction, all results will be rolled back, but a message about the collapse of the second command will appear in the scheduler’s log. The same message will appear if separate transactions are executed, but the result of the first will be saved (the third transaction will not be executed).

When pgpro_scheduler starts, a group of background workers always starts working with its own hierarchy: one worker, with the rank of supervisor of the scheduler, controls workers with the rank of database managers - one for each database specified in the configuration line. Managers, in turn, control workers directly serving tasks. Supervisors and managers are fairly easy processes, so if the scheduler serves even dozens of databases, this does not affect the overall system load. And then workers are launched in each database according to the needs for processing requests. In total, they should fit into the limitation of the DBMS max_worker_processes. A group of teams for instant execution of tasks uses resources in a different way, but more on that later.


Fig. 1 The main mode of operation of pgpro_scheduler

pgpro_schedulerThis is an extension of Postgres. Therefore, it is installed on a specific database. This creates several system tables in the schedule scheme, by default they are not visible to the user. The database now knows 2 new, special data types: cron_rec and cron_job, with which it will be possible to work through SQL queries. There is a log table that does not duplicate the DBMS log. Information about the success or failure of the scheduler jobs is available only through the extension functions pgpro_scheduler. This is done so that one user of the scheduler does not know about the activities of another user of the scheduler. The functions make it possible to selectively view the log starting from a certain date, for example:

SELECT * from schedule.get_user_log() WHERE started > now() - INTERVAL '1 day';

You can create a task using JSON using the functionschedule.create_job(data jsonb). The only argument to this function is a JSONB object with job information. Examples will follow.

This object may contain the following keys, some of which may be omitted:

• name - job name;
• node - the name of the node (in case of work in the multimaster architecture);
• command - the set of SQL queries that will be executed are specified as an array;
• run_as - the user on behalf of which the commands will be executed;

There are several types of schedule presentations to choose from: for those who are accustomed to cron, a crontab-style line that defines the execution schedule. But you can use the rule - then the schedule will be presented as a JSONB object (see description below). Another option: date - a set of specific dates for which the execution of commands is scheduled. They can be combined, but at least one option must be involved. This will look, for example, like this: "cron":"55 7 * * *"- from the example that can be seen below.
In addition, there are many more useful parameters that can be found in the documentation . Among them:

• start_date and end_date - the beginning and end of the interval in which the execution of the scheduled command is possible (may be NULL);
• max_instances - the maximum number of job instances that can be run simultaneously. 1 by default;
• max_run_time - determines the maximum duration of the job. Set in interval type format. If this field contains NULL or is not specified, time is not limited. The default value is NULL;
• onrollback - SQL query that will be executed when the main transaction fails. By default, the request is undefined;
• next_time_statement - SQL query that will be executed to calculate the next time the job starts. It must necessarily return a value in the format timestamp with time zone;

The schedule can be set as a string in the crontab style (cron key) or as a JSONB object (rule key). They may contain the following keys:

• minutes - minutes; array of integers in the range 0 ... 59;
• hours - hours; an array of integers in the range 0 ... 23;
• days - days of the month; array of integers in the range 1 ... 31;
• months - months; array of integers in the range 1 ... 12;
• wdays - days of the week; an array of integers in the range 0 ... 6 (0 - Sunday);
• onstart - integer value 0 or 1; if this value is 1, the task will be executed only once when the scheduler starts.

A task can also be scheduled for a specific date or for a set of dates. That is, in principle, the task can be one-time, although for one-time tasks you can use the special one-time job mode with other function calls.


Figure 1. Scheduler Process Hierarchy

Fieldnext_time_statementmay contain an SQL query that will be executed after the main transaction to calculate the next start time. If this key is defined, the time of the first start of the task will be calculated according to the methods described above, but the next start will be scheduled for the time that this request will return. This request should return a record containing in the first field a value of type timestamp with time zone. If the return value is of a different type or an error occurs while executing the request, the task is marked as failed and its further execution is canceled.

This request will be executed in any state of completion of the main transaction. You can get the transaction completion status in it from the Postgres Pro Enterprise variableschedule.transaction_state:

• success - transaction completed successfully
• failure - transaction completed with an error

As can be seen even from the abbreviated description, the set of features is rich. There are about 40 functions that work with the pgpro_scheduler application. You can create tasks, cancel them, see their status, filter information about tasks by users and other criteria.

Once, but without waiting in line

As stated, there is still an important class of tasks for the scheduler: the formation of separate, non-periodic tasks using the one-time job mechanism. If the run_after parameter is not set, then in this mode the scheduler can start executing the task immediately at the moment of receipt - up to the time interval for polling the table into which the task is written. In the current implementation, the interval is fixed at 1 second. background workers start in advance and wait "under the pairs" for the job to appear, rather than starting as needed, as in schedule mode. Their number is determined by the schedule.max_parallel_workers parameter. The corresponding number of requests can be processed in parallel.


Fig. 2 One-time job mode.

The main function that forms the task looks like this:

schedule.submit_job(query text [options...])

This function has, in accordance with its specifics, which was at the beginning, subtle settings. The max_duration parameter sets the maximum execution time. If the work is not done in the allotted time, the task is canceled (by default, the execution time is unlimited). max_wait_interval refers not to the time of work, but to the wait time for the start of work. If the DBMS does not find during this period of time "workers" who are ready to take up execution, the task is removed. The interesting depends_on parameter sets the array of jobs (in one-time mode), after which you need to start this job.

Useful option -resubmit_limit- sets the maximum number of restart attempts. Let's say the task starts a procedure that starts sending a message to the mail. The mail server, however, is in no hurry to receive it, and by timeout or generally due to a lack of communication, the process ends to resume again immediately or after a specified time. Without limitation in resubmit_limit, attempts will continue until the victorious end.

Seasonings and desserts

At the beginning, detached jobs were mentioned. In the current version, the process that launched the one-time task erases its existence in anticipation of the result. The overhead of background worker is small, there is no point in stopping it. It is important that the fulfillment or non-fulfillment of the task will not pass without a trace, we can learn about his fate from a request to the log of the scheduler available to us, and not just to the database administrator. This is not the only way to track a transaction even if it is rolled back: Postgres Pro Enterprise has an offline transaction mechanism that can be used for the same purpose. But in this case, the result will be written to the DBMS log, and not to the "personal" log of the user who launched the scheduler.

If the user of the scheduler needs to plan or just run some OS commands with the rights that are available to him inside the OS, he can easily do this through the scheduler using the programming languages ​​available to him. Suppose he decided to use untrusted Perl:

CREATE LANGUAGE plperlu;

After that, you can write such a function, for example, as a regular request:

DO LANGUAGE 'plperlu' $$
system('cat /etc/postgresql/9.6/main/pg_hba.conf > $HOME/conf_tmp');
$$;

Life example: 1. storage of irrelevant logs

For starters, a simplified example of managing partitions from a scheduler. Suppose we split the site visit logs into sections by month. We don’t want to store sections of two-year-old freshness and younger on expensive fast disks, and dump the rest into another table space corresponding to other, cheaper media, while preserving, however, full-fledged search and other operations on all logs (with a table not divided into sections, this impossible). We use convenient section management functions in the pg_pathman extension. In the postgresql.conf file should be the line shared_preload_libraries = 'pg_pathman, pgpro_scheduler'.

CREATE EXTENSION pg_pathman; CREATE EXTENSION pgpro_scheduler;

Configurable: There may be several bases. In this case, they are listed with a comma inside the quotation marks. - reread configuration changes without restarting Postgres.

ALTER SYSTEM SET schedule.enabled = on;
ALTER SYSTEM SET schedule.database = 'test_db';


SELECT pg_reload_conf();

CREATE TABLE partitioned_log(id int NOT NULL, visit timestamp NOT NULL);

We just created a parent table that we will partition into sections. This is a tribute to the traditional PostgreSQL syntax based on table inheritance. Now, in Postgres Pro Enterprise, you can create partitions not in 2 stages (first, an empty parent table, then define partitions), but immediately define partitions. In this case, we will use the convenient pg_pathman function, which allows you to first set the approximate number of sections. As filling in, the necessary sections will be created automatically:

SELECT create_range_partitions('partitioned_log','visit', '2015-01-01'::date, '1 month'::interval, 10);

We have set 10 initial sections, one for a month, starting from January 1. 2015. Fill them with some amount of data.

INSERT INTO partitioned_log SELECT i, '2015-01-01'::date + 60*60*i*random()::int*'1 second'::interval visit FROM generate_series(1,24*365) AS g(i);

You can monitor the number of sections as follows:

SELECT count(*) FROM pathman_partition_list WHERE parent='partitioned_log'::regclass;

Starting INSERT, “twisting” the start date and / or factors before random, make the number of sections a little more than 24 (2 years).

We create a directory in the OS and the corresponding table space where obsolete logs will be stored:

CREATE TABLESPACE archive LOCATION '/tmp/archive';

And, finally, the function that the scheduler will run daily: It takes as parameters: the name of the partitioned table , the suffix that will be added to the name of the moved section , the table space (archive) and the number of months is the border of the 1st freshness logs (24). To warm up, we will set up a one-time task: Having executed, the scheduler will display the task id. You can see its status in the views and . The tasks assigned by the function do not fall into the scheduler log . To make it more convenient to play with the scheduler and sections, you can create a function

CREATE OR REPLACE FUNCTION move_oldest_to_archive(parent_name text, suffix text, tblsp_name text, months_hot int) RETURNS int AS
$$
DECLARE
       i int;
       part_rename_sql text;
       part_chtblsp_sql text;
       part_name text;
BEGIN
       i=0;
       FOR part_name IN SELECT partition FROM pathman_partition_list WHERE parent=parent_name::regclass and partition::text NOT LIKE '%'||suffix ORDER BY range_max OFFSET months_hot LOOP
              i:=i+1;
              part_rename_sql:=format('ALTER TABLE %I RENAME to %I', part_name, part_name||'_'||suffix);
              part_chtblsp_sql:=format('ALTER TABLE %I SET TABLESPACE %I', part_name, tblsp_name);
              EXECUTE part_chtblsp_sql;
              EXECUTE part_rename_sql;
              RAISE NOTICE 'executed %, %',part_rename_sql,part_chtblsp_sql;
       END LOOP;
       RETURN i;
END;
$$ LANGUAGE plpgsql;

(partitioned_log)(archived)



SELECT schedule.submit_job(query := $$select move_oldest_to_archive('partitioned_log','archived', 'archive', 24);$$);

schedule.job_statusschedule.all_job_statussubmit_job()

unarchive(parent_name text, suffix text), rolling back changes (we do not give to save space).

It can also be started from the scheduler, but using the run_after parameter, which sets the delay time in seconds, so that we have time to think whether we did the right thing:

SELECT schedule.submit_job(query := $$'select unarchive('partitioned_log','archived');',run_after='10'$$);

and if it is wrong, you can cancel it by a function schedule.cancel_job(id);
After making sure that everything works as intended, you can put the task (now in JSON syntax) already in the schedule:

SELECT schedule.create_job($${"commands":"SELECT move_oldest_to_archive('partitioned_log','archived', 'archive', 24);","cron":"55 7 * * *"}$$);

That is, every morning at five minutes to eight the scheduler will check if it is time to move the outdated partitions to the “cold” archive and move if it's time. Status at this time can be checked with the log Scheduler: schedule.get_log();

Life example: 2. we spread banners on servers

We show how one of the typical tasks is solved, in which the scheduled work is required and one-time tasks are used.

We have a content delivery network (CDN). We are going to lay out banners on several websites included in it, which users from advertising agencies automatically uploaded to the directory reserved for them. Let's create a function that initializes the download of the banner to the server. For each server, it creates a download task, as well as a task that expects all created downloads and assigns the correct status to the banner when the downloads are completed: And this function simulates sending a banner to the server (it actually just sleeps for a while): This function Based on the status of banner downloads to the server, it will determine which status to put on the banner:

DROP SCHEMA IF EXISTS banners CASCADE;
CREATE SCHEMA banners;

SET search_path TO 'banners';

CREATE TYPE banner_status_t AS enum ('submitted', 'distributing', 'ready', 'error');
CREATE TYPE cdn_dist_status_t AS enum ('submitted', 'processing', 'ready', 'error');

CREATE TABLE banners (
      id SERIAL PRIMARY KEY,
      title text,
      file text,
      status banner_status_t DEFAULT 'submitted'
);

CREATE TABLE cdn_servers(
      id SERIAL PRIMARY KEY,
      title text,
      address text,
      active boolean
);

CREATE TABLE banner_on_cdn(
      banner_id int,
      server_id int,
      created timestamp with time zone DEFAULT now(),
      started timestamp with time zone,
      finished timestamp with time zone,
      url text,
      error text,
      status cdn_dist_status_t DEFAULT 'submitted'
);

CREATE INDEX banner_on_cdn_banner_server_idx ON banner_on_cdn (banner_id, server_id);
CREATE INDEX banner_on_cdn_url_idx ON banner_on_cdn (url);



CREATE FUNCTION start_banner_upload(bid int) RETURNS bigint AS
$BODY$
DECLARE
      job_id bigint;
      r record;
      dep bigint[];
      sql text;
      len int;
BEGIN
      UPDATE banners SET status = 'distributing' WHERE id = bid;
      dep := '{}'::bigint[];
      FOR r IN SELECT * FROM cdn_servers WHERE active is TRUE LOOP
             -- для каждого сервера создаем задачу для загрузки
             INSERT INTO banner_on_cdn (banner_id, server_id) VALUES (bid, r.id);
             sql := format('select banners.send_banner_to_server(%s, %s)', bid, r.id);
             job_id := schedule.submit_job(
                    sql,
                    name := format('send banner id = %s to server %s', bid, r.title)
             );
             -- собираем идетификаторы созданных задач в массив
             dep := array_append(dep, job_id);
      END LOOP;
      len := array_length(dep, 1);
      IF len = 0 THEN
             UPDATE banners SET status = error WHERE id = bid;
             RETURN NULL;
      END IF;
      -- создаем задачу, которая будет выполненна сразу после завершения задач,
      -- идентификаторы которых мы собрали в массив dep
      job_id = schedule.submit_job(
             format('SELECT banners.finalize_banner(%s)', bid),
             depends_on := dep,
             name := format('finalization of banner %s', bid)
      );
      RETURN job_id;
END
$BODY$
LANGUAGE plpgsql SET search_path FROM CURRENT;



CREATE FUNCTION send_banner_to_server(bid int, sid int)
RETURNS boolean AS
$BODY$
DECLARE
      banner record;
      server record;
BEGIN
      SELECT * from banners WHERE id = bid LIMIT 1 INTO banner;
      SELECT * from cdn_servers WHERE id = sid LIMIT 1 INTO server;

      UPDATE banner_on_cdn SET
             status = 'processing',
             started = now()
      WHERE
             banner_id = bid AND server_id = sid;
             
      PERFORM pg_sleep((random()*10)::int);
      UPDATE banner_on_cdn SET
             url = 'http://' || server.address || '/' || banner.file,
             status = 'ready',
             finished = now()
      WHERE
             banner_id = bid AND server_id = sid;

      RETURN TRUE;
END;
$BODY$
LANGUAGE plpgsql set search_path FROM CURRENT;



CREATE FUNCTION finalize_banner(bid int)
RETURNS boolean AS
$BODY$
DECLARE
      N int;
BEGIN
      SELECT count(*) FROM banner_on_cdn WHERE banner_id = bid AND status IN ('submitted', 'processing') INTO N;
      IF N > 0 THEN -- не все загрузки еще завершились
             RETURN FALSE;
      END IF;
      SELECT count(*) FROM banner_on_cdn WHERE banner_id = bid AND status IN ('error') INTO N;
      IF N > 0 THEN -- загрузки прошли с ошибками
             UPDATE banners SET status = 'error' WHERE id = bid;
             RETURN FALSE;
      END IF;
      -- все хорошо
      UPDATE banners SET status = 'ready' WHERE id = bid;
      RETURN TRUE;
END;
$BODY$
LANGUAGE plpgsql set search_path FROM CURRENT;

This feature will schedule to check if there are raw banners. And, if necessary, start processing the banner: Now let's take care of the data. Let's create a list of servers: Let's create a couple of banners: And, finally, we will schedule the task of checking for newly arrived banners that need to be laid out on the servers. The task will be performed every minute: That's all, the pictures will be laid out on sites, you can relax.

CREATE FUNCTION check_banners() RETURNS int AS
$BODY$
DECLARE
      r record;
      N int;
BEGIN
      N := 0;
      FOR r IN SELECT * from banners WHERE status = 'submitted' FOR UPDATE LOOP
             PERFORM start_banner_upload(r.id);
             N := N + 1;
      END LOOP;

      RETURN N;
END;
$BODY$
LANGUAGE plpgsql SET search_path FROM CURRENT;



INSERT INTO cdn_servers (title, address, active)
      VALUES ('server #1', 'cdn1.local', true);
INSERT INTO cdn_servers (title, address, active)
      VALUES ('server #2', 'cdn2.local', true);
INSERT INTO cdn_servers (title, address, active)
      VALUES ('server #3', 'cdn3.local', true);
INSERT INTO cdn_servers (title, address, active)
      VALUES ('server #4', 'cdn4.local', true);



INSERT INTO banners (title, file) VALUES ('banner #1', 'bbb1.jpg');
INSERT INTO banners (title, file) VALUES ('banner #2', 'bbb2.jpg');



SELECT schedule.create_job('* * * * *', 'select banners.check_banners()');

RESET search_path;

Afterword

As a Post Scriptum, we inform that the pgpro_scheduler scheduler works not only on a separate server, but also in the multimaster cluster configuration. But this is a topic for another discussion.

And as a Post Post Scriptum - that in the future plans to integrate the scheduler into the currently created graphical administration shell.