Postgres Proでジョブスケジューラができること

タスクのスケジューラ（スケジューラ）は、常にデータベースの世界で不可欠なツールとは見なされませんでした。 それはすべてDBMSの目的と起源に依存していました。 従来の商用DBMS（Oracle、DB2、MS SQL）は、スケジューラなしでは想像することは絶対に不可能です。 一方、スケジューラがないために、このファッショナブルなNoSQL DBMSの選択を拒否する潜在的なMongoDBユーザーを想像することは困難です。 （ちなみに、「タスクスケジューラ」という用語はDBMSのロシア語のコンテキストで使用され、クエリプランナー-クエリプランナーと区別します。簡潔にするため、ここではスケジューラと呼びます）。



PostgreSQLは、オープンソースであり、DIYライフスタイル（「自分でやる」）でコミュニティの伝統を吸収してきましたが、今日では、少なくとも副次的な商用DBMSであると定期的に主張しています。 このことから、PostgreSQLには単にスケジューラーが必要であり、このスケジューラーはデータベース管理者とユーザーにとって便利である必要があるということが自動的に続きます。 また、商用DBMSの完全に機能する機能を再現することは望ましいことですが、独自の機能を追加することもできます。



スケジューラの必要性は、産業運用でベースを使用する場合に最も顕著です。 データベースの実験のためにサーバーを割り当てられた開発者にとって、スケジューラは一般に役に立たない：必要であれば、彼はOS（cronまたはUnixで）によって必要なすべての操作を計画します。 しかし、大砲を撃つために真面目な会社の労働基地に立ち入ることは許可されません。 重要な管理上のニュアンスがあります。つまり、もはやニュアンスではありませんが、決定的な理由ではないにしても、重大な理由です。データベース管理者とシステム管理者は、異なるタスクを持つ単なる異なる人ではありません。 彼らは会社の異なる部門に属している可能性があり、場合によっては異なるフロアに座っていることもあります。 理想的には、データベース管理者はその実行可能性をサポートし、その進化を監視し、システム管理者の責任領域はOSとネットワークの実行可能性です。



したがって、データベース管理者は、サーバー上で必要な作業の必要なセットを実行するツールを持っている必要があります。 Oracleスケジューラに関する資料で、 「Oracleスケジューラを使用すると、データベース中心のアプリケーションを構築するときに、さまざまなプラットフォームでOS固有のタスクスケジューラ（cron、at）を使用する必要がなくなります」と言うのは当然です。 つまり、データベース管理者はすべてを行うことができます。特に、OSメカニズムにガイドされていないOracle管理者を想像するのは難しいためです。 彼は、システム管理者に毎回走ったり、OSによって日常的な操作が必要なときに手紙を書いたりする必要はありません。



Oracle、DB2、MS SQLなどの商用DBMSに典型的なスケジューラ要件は次のとおりです。



プランナーはできる必要があります

• スケジュールされた作業を開始する、
• 作業の実行を制御し、必要に応じてタスクを削除できるようにします。
• 限られた期間（ウィンドウ内）でタスクを実行する
• タスクのシーケンスを構築するには（前のタスクが完了した後に次のタスクが実行されます）、
• 1つのトランザクションで複数のリクエストを完了することができ、
• 1つのデータベースで定義されたタスクを複数のデータベースで実行し、
• OSの（メイン）機能を使用し、
• スケジュールの一部のタスクが完了しなかった場合、管理者に通知します。
• 1回限りのタスクを実行します。

最後の点は明らかではないようです。1回限りのタスクを実行できるスケジューラ以外にも、他にも多くの通常のツールがあります。 しかし、これは完全に通常の実行モードではありません。 たとえば、分離ジョブモード：私たちは、それが発生したプロセスから（一時的または永続的に）切断されたタスクについて話している。 作業を完了すると、切断されたプロセスは、それを開始したプロセスに再度接続し（成功または失敗の完了のシグナルを送信）、結果を通知するか、結果を（ファイルまたはデータベーステーブルに）書き込むことができます。 一部のDBMSスケジューラは、DBMS自体を停止および開始できます（このようなタスクは設定していません）。

PostgreSQLとそのエージェント

DBMS自体の「外側」と「内側」のさまざまな方法でタスクを解決することができます。 完全に機能するスケジューラを作成するための最も深刻な試みは、pgAdmin III / IVで配布されるpgAgentです。 商用版-EnterpriseDB配布キット-pgAdmin GUIに統合されており、クロスプラットフォームで使用できます。



pgAgentは次のことができます。

• タスクを実行する
• SQLスクリプト（異なるデータベースを含む）および/またはシェル/バッチスクリプトで構成されるタスクのシーケンスを実行します。
• 稼働していないウィンドウを設定します（たとえば、週末に何らかのアクションを実行しないでください）。

このスケジューラはPostgreSQLの拡張機能として機能しますが、DBMSの「内部」でタスクを実行するのではなく、独自の「外部」デーモンを作成します。



このアプローチには欠点があります。 それらの中で重要なのは：



pgAgentによって起動されたすべてのジョブは、エージェントを起動したユーザーの権限で実行されます。 SQLクエリは、データベースに接続しているユーザーの権限で実行されます。 シェルスクリプトは、pgAgentデーモン（またはWindows上のサービス）が実行されているユーザーの権利で実行されます。 したがって、セキュリティのために、タスクを作成および実行できるユーザーを制御する必要があります。 さらに、パスワードは接続文字列に含めることはできません。Unixではpsコマンドの出力およびデータベース起動スクリプトに表示され、Windowsでは暗号化されていないテキストとしてレジストリに保存されるためです。

（ pgAdmin 4 1.6ドキュメントから ）。



このソリューションでは、pgAgentは指定された間隔でデータベースサーバーをポーリングします（作業に関する情報はデータベーステーブルに格納されているため）。ジョブがあります。 したがって、何らかの理由で作業を開始すべき時点でエージェントが動作しない場合、エージェントが動作を開始するまでエージェントは起動しません。



さらに、サーバーへの接続は、可能な接続のプールを消費します。その最大数は、構成パラメーターmax_connectionsによって決まります。 エージェントが多くのプロセスを生成し、管理者が警戒していない場合、これは問題になる可能性があります。



DBMSに完全に統合された（DBMSの内部の）スケジューラを作成すると、これらの問題がなくなります。 また、psqlなど、データベースにアクセスするための最小限のインターフェイスに慣れているユーザーにとっては特に便利です。

pgpro_schedulerとそのスケジュール

2016年の終わりに、Postgres ProfessionalはDBMSに完全に統合された独自のスケジューラーの作成を開始しました。 現在では、顧客によって使用され、詳細に文書化されています。 スケジューラは、pgpro_schedulerと呼ばれる拡張機能（アドオンモジュール）として作成され、最初のバージョンから開始されるPostgres Pro Enterpriseの商用バージョンの一部として出荷されます。 開発者-Vladimir Ershov。



DBMS構成ファイルにインストールするとき、構成ファイルにshared_preload_libraries = 'pgpro_scheduler'



を含めることを忘れないでください。 拡張機能( CREATE EXTENSION pgpro_scheduler;)



をインストールしたら、構成ファイルの行（schedule.enabled = on）で有効にし、スケジューラーの対象となるデータベースを一覧表示する必要があります（例： schedule.database = 'database1,database2'



）。



当初から、pgpro_schedulerを、JSONで記述された構成を使用して、会社にとって有機的な現代的なスタイルで作成することが決定されました。 これは、たとえば、スケジューラをアプリケーションに統合できるWebサービスの作成者にとって便利です。 しかし、JSONを使用したくない人のために、通常の変数の形式でパラメーターを取る関数があります。 スケジューラにはDBMSディストリビューションキットが付属しており、クロスプラットフォームです。



pgpro_schedulerは外部のデーモンやサービスを起動しませんが、postmaster-バックグラウンドプロセスの子であるバックグラウンドワーカープロセスを作成します。 「ワーカー」の数はpgpro_scheduler構成で指定されますが、一般的なサーバー構成によって制限されます。 スケジューラーは、実際に最も一般的なSQLコマンドを制限なしで受け取るため、利用可能なPostgres言語で関数を実行できます。 JSON構造に複数のSQLクエリが含まれている場合、それらは（特定の構文に従う場合）単一のトランザクション内で実行できます。



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







これは次と同等です：



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





そして、各リクエストがトランザクション内にある場合：



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



-つまり、最後のパラメータなしで、デフォルトで。



リストの2番目のコマンドでエラーが発生するとします（もちろん、 SELECT 2



では発生しそうにありませんが、ある種の「ダム」クエリを想像してください）。 1つのトランザクションで実行された場合、すべての結果がロールバックされますが、2番目のコマンドの崩壊に関するメッセージがスケジューラーのログに表示されます。 別のトランザクションが実行された場合も同じメッセージが表示されますが、最初のトランザクションの結果は保存されます（3番目のトランザクションは実行されません）。



pgpro_schedulerが起動すると、バックグラウンドワーカーのグループは常に独自の階層で作業を開始します。1人のワーカーは、スケジューラーのスーパーバイザーのランクで、データベースマネージャーのランクでワーカーを制御します。 マネージャーは、タスクを直接処理するワーカーを制御します。 スーパーバイザとマネージャは非常に簡単なプロセスであるため、スケジューラが数十ものデータベースを処理する場合、これはシステム全体の負荷に影響しません。 そして、リクエストを処理するニーズに応じて、各データベースでワーカーが起動されます。 合計で、DBMS max_worker_processesの制限に適合する必要があります。 タスクの即時実行のためのチームのグループは、リソースを異なる方法で使用しますが、それについては後で詳しく説明します。





図1 pgpro_schedulerのメイン動作モード



pgpro_scheduler



はPostgres拡張機能です。 したがって、特定のデータベースにインストールされます。 これにより、スケジュールスキームにいくつかのシステムテーブルが作成されます。デフォルトでは、それらはユーザーには表示されません。 データベースは、cron_recとcron_jobの2つの新しい特別なデータ型を認識できるようになりました。これらを使用して、SQLクエリを処理できます。 DBMSログを複製しないログテーブルがあります。 スケジューラジョブの成功または失敗に関する情報は、pgpro_scheduler拡張機能を介してのみ利用できます。 これは、スケジューラのあるユーザーがスケジューラの別のユーザーのアクティビティを知らないようにするために行われます。 これらの機能により、特定の日付から開始してログを選択的に表示することが可能になります。例：



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







schedule.create_job(data jsonb)



関数を使用して、JSONを使用してタスクを作成できます。 この関数の唯一の引数は、ジョブ情報を持つJSONBオブジェクトです。 次に例を示します。



このオブジェクトには次のキーが含まれる場合がありますが、その一部は省略可能です。

• name-ジョブ名;
• node-ノードの名前（マルチマスターアーキテクチャで作業する場合）;
• command-実行されるSQLクエリのセットは配列として指定されます。
• run_as-コマンドを実行するユーザー。

スケジュール表現のタイプを選択できます：cronに慣れている人向け-実行スケジュールを設定するcrontabスタイルの行。 ただし、ルールを使用できます。スケジュールはJSONBオブジェクトとして表示されます（以下の説明を参照）。 別のオプション：日付-コマンドの実行がスケジュールされている特定の日付のセット。 これらは組み合わせることができますが、少なくとも1つのオプションが必要です。 たとえば、次のようになります。 "cron":"55 7 * * *"



-以下に示す例から。

さらに、 ドキュメントで見つけることができるより多くの有用なパラメーターがあります 。 それらの中には：

• start_dateおよびend_date-スケジュールされたコマンドの実行が可能な間隔の開始と終了（NULLの場合があります）。
• max_instances-同時に実行できるジョブインスタンスの最大数。 デフォルトでは1。
• max_run_time-ジョブの最大期間を決定します。 間隔タイプの形式で設定します。 このフィールドにNULLが含まれているか、指定されていない場合、時間は制限されません。 デフォルト値はNULLです。
• onrollback-メイントランザクションが失敗したときに実行されるSQLクエリ。 デフォルトでは、リクエストは未定義です。
• next_time_statement-次回ジョブが開始する時間を計算するために実行されるSQLクエリ。 必ずタイムゾーン付きのタイムスタンプ形式で値を返す必要があります。

スケジュールは、crontab（cronキー）のスタイルの文字列またはJSONBオブジェクト（ルールキー）として設定できます。 次のキーが含まれる場合があります。

• 分-分; 0〜59の範囲の整数の配列。
• 時間-時間; 0〜23の範囲の整数の配列。
• 日-月の日; 1〜31の範囲の整数の配列。
• 月-月; 1〜12の範囲の整数の配列。
• wdays-曜日; 0〜6（0-日曜日）の範囲の整数の配列。
• onstart-整数値0または1。 この値が1の場合、スケジューラーの開始時にタスクは1回だけ実行されます。

タスクは、特定の日付または一連の日付に対してスケジュールすることもできます。 つまり、原則として、タスクは1回だけ実行できますが、1回限りのタスクでは、他の関数呼び出しで特別な1回限りのジョブモードを使用できます。





図1.スケジューラプロセス階層



next_time_statement



フィールドには、メイントランザクションの後に実行されて次の開始時刻を計算するSQLクエリが含まれる場合があります。 このキーが定義されている場合、タスクの最初の開始時刻は上記の方法に従って計算されますが、次の開始はこの要求が返される時間にスケジュールされます。 この要求は、最初のフィールドにタイムゾーン付きタイムスタンプタイプの値を含むレコードを返す必要があります。 戻り値のタイプが異なる場合、または要求の実行中にエラーが発生した場合、タスクは失敗としてマークされ、それ以降の実行はキャンセルされます。



この要求は、メイントランザクションの完了状態で実行されます。 Postgres Pro Enterpriseのschedule.transaction_state:



からトランザクションの完了状態を取得できschedule.transaction_state:

• 成功-トランザクションは正常に完了しました
• 失敗-トランザクションはエラーで完了しました

省略された説明からもわかるように、一連の機能は豊富です。 合計で、pgpro_schedulerアプリケーションで動作する約40の関数があり、タスクの作成、キャンセル、ステータスの表示、ユーザーやその他の基準によるタスクに関する情報のフィルタリングを行うことができます。

一度、ただし順番に待たずに

前述のように、スケジューラには重要なクラスのタスクがあります。1回限りのジョブメカニズムを使用して、個別の非周期的なタスクを形成します。 run_afterパラメーターが設定されていない場合、このモードでは、スケジューラーは受信の瞬間にタスクの実行を開始できます-タスクが書き込まれるテーブルをポーリングする時間間隔まで。 現在の実装では、間隔は1秒に固定されています。 バックグラウンドワーカーは事前に開始し、スケジュールモードのように必要に応じて開始するのではなく、ジョブが表示されるまで「ペアで」待機します。 それらの数は、schedule.max_parallel_workersパラメーターによって決定されます。 対応する数のリクエストを並行して処理できます。





図2ワンタイムジョブモード。



タスクを形成する主な機能は次のようになります。



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







この機能には、最初に説明した詳細に従って、微妙な設定があります。 max_durationパラメーターは、最大実行時間を設定します。 割り当てられた時間内に作業が行われない場合、タスクはキャンセルされます（デフォルトでは、実行時間は無制限です）。 max_wait_intervalは、作業時間ではなく、作業の開始を待つ時間を指します。 この期間中にDBMSが実行を開始する準備ができている「ワーカー」を見つけられない場合、タスクは削除されます。 興味深いdepend_onパラメーターは、ジョブの配列を（ワンタイムモードで）設定します。その後、このジョブを開始する必要があります。



便利なパラメータresubmit_limit



は、再起動の最大試行回数を設定します。 タスクがメールへのメッセージの送信を開始するプロシージャを開始するとします。 ただし、メールサーバーは受信を急ぐことはなく、タイムアウトするか、通常は通信不足のため、プロセスは終了してすぐに、または指定された時間後に再開します。 resubmit_limitに制限はありませんが、試行は勝利するまで続きます。

調味料とデザート

最初に、独立したジョブが言及されました。 現在のバージョンでは、1回限りのタスクを起動したプロセスは、結果を見越してその存在を消去します。 バックグラウンドワーカーのオーバーヘッドは小さく、停止する意味はありません。 タスクのフルフィルメントまたは非フルフィルメントがトレースなしで通過しないことが重要です。データベース管理者だけでなく、利用可能なスケジューラのログへのリクエストから彼の運命について知ることができます。 これは、トランザクションがロールバックされた場合でもトランザクションを追跡する唯一の方法ではありません。PostgresPro Enterpriseには、同じ目的で使用できるオフライントランザクションメカニズムがあります。 ただし、この場合、結果は、スケジューラを起動したユーザーの「個人」ログではなく、DBMSログに書き込まれます。



スケジューラユーザーが、OS内で利用可能な権限でいくつかのOSコマンドをスケジュールするか、単に実行する必要がある場合、利用可能なプログラミング言語を使用してスケジューラを通じてこれを簡単に行うことができます。 信頼できないPerlを使用することにしたとしましょう。



CREATE LANGUAGE plperlu;







その後、このような関数を、たとえば通常のリクエストとして作成できます。



DO LANGUAGE 'plperlu' $$

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

$$;

実例：1.無関係なログの保存

まず、スケジューラからパーティションを管理する簡単な例。 月ごとにサイト訪問ログをセクションに分割するとします。 2歳以下の新鮮なセクションを高価な高速ディスクに保存したくないので、残りを他の安価なメディアに対応する別のテーブルスペースにダンプします。ただし、すべてのログに対する本格的な検索やその他の操作は（セクションに分割されていないテーブルで、不可能）。 pg_pathman拡張機能で便利なセクション管理機能を使用します。 postgresql.confファイルには、shared_preload_libraries = 'pg_pathman、pgpro_scheduler'という行が必要です。



CREATE EXTENSION pg_pathman; CREATE EXTENSION pgpro_scheduler;







構成：



ALTER SYSTEM SET schedule.enabled = on ;

ALTER SYSTEM SET schedule.database = 'test_db' ;







いくつかの拠点があります。 この場合、引用符の中にコンマが付いてリストされます。

SELECT pg_reload_conf();



-Postgresを再起動せずに設定変更を再読み取りします。



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







セクションに分割する親テーブルを作成しました。 これは、テーブル継承に基づいた従来のPostgreSQL構文へのオマージュです。 これで、Postgres Pro Enterpriseでは、2段階（最初に空の親テーブル、次にパーティションを定義）ではなくパーティションを作成できますが、すぐにパーティションを定義できます。 この場合、便利なpg_pathman関数を使用します。これにより、最初にセクションのおおよその数を設定できます。 それらがいっぱいになると、必要なセクションが自動的に作成されます。



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







1月1日から1か月に1回、10の初期セクションを設定します。 2015.ある程度のデータを入力します。



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);







次のようにセクションの数を監視できます。



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







INSERT



開始して、開始日や要素をランダムに「ねじる」ことにより、セクションの数を24（2年）より少し増やします。



OSにディレクトリを作成し、古いログが保存される対応するテーブルスペースを作成します。



CREATE TABLESPACE archive LOCATION '/tmp/archive' ;







そして最後に、スケジューラーが毎日実行する機能：



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)



、テーブルスペース（アーカイブ）、および月数-最初の鮮度ログの境界（24）を取ります。



ウォームアップするために、1回限りのタスクを配置します。



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







実行されると、スケジューラはジョブIDを表示します。 schedule.job_status



およびschedule.all_job_status



ビューでそのステータスを確認できます。 submit_job()



によって割り当てられたジョブは、スケジューラログに記録されません。



スケジューラーとセクションの操作をより便利にするために、変更をロールバックするunarchive(parent_name text , suffix text )



関数unarchive(parent_name text , suffix text )



を作成できます（これはスペースを節約するために使用しません）。



また、スケジューラーから開始することもできますが、遅延時間を秒単位で設定するrun_afterパラメーターを使用して、正しいことをしたかどうかを考える時間を確保します。



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







間違っている場合は、 schedule.cancel_job(id)



関数でキャンセルできます。

すべてが意図したとおりに動作することを確認したら、タスク（現在はJSON構文）を既にスケジュールに入れることができます。



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







つまり、スケジューラは毎朝5分から8時に、古くなったパーティションを「コールド」アーカイブに移動する時間であるかどうかをチェックし、時間があれば移動します。 今回は、スケジューラーのログでステータスを確認できます： schedule.get_log()



;

実例：2.サーバーにバナーを広げます

スケジュールされた作業が必要であり、1回限りのタスクが使用される典型的なタスクの1つがどのように解決されるかを示します。



コンテンツ配信ネットワーク（CDN）があります。 広告代理店のユーザーが予約したディレクトリに自動的にアップロードしたバナーを、それに含まれるいくつかのWebサイトに配置します。



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 ;







この機能は、生のバナーがあるかどうかをチェックするようにスケジュールします。 そして、必要に応じて、バナー処理を開始します。



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;







それだけで、写真はサイトにレイアウトされ、リラックスできます。

あとがき

Post Scriptumとして、pgpro_schedulerスケジューラは別のサーバー上だけでなく、マルチマスタークラスター構成でも動作することをお知らせします。 しかし、これは別の議論のトピックです。



そして、 Post Post Scriptumとして-将来、スケジューラを現在作成されているグラフィカルな管理シェルに統合することを計画しています。