0001-cron-allow-some-cron-workers-to-be-run-in-parallel-3.patch

Frédéric Péters, 09 septembre 2022 13:38

Voir les différences: en ligne côte à côte

Subject: [PATCH] cron: allow some cron workers to be run in parallel (#33280)

 tests/test_publisher.py                | 427 ++++++++++++++-----------
 wcs/qommon/cron.py                     |  13 +-
 wcs/qommon/management/commands/cron.py | 113 ++++---
 wcs/settings.py                        |   3 +
 wcs/sql.py                             |  29 ++
 5 files changed, 344 insertions(+), 241 deletions(-)

     import datetime
     import glob
     import io
     import json
     import os
-...
     import re
     import shutil
     import sys
     import tempfile
     import time
     import xml.etree.ElementTree as ET
     import zipfile
-...
     import pytest
     from django.core.management import call_command
     from django.core.management.base import CommandError
     from django.http import Http404
     from django.test import override_settings
     from django.utils.timezone import localtime
     from quixote import cleanup
     from quixote import cleanup, get_publisher
     from quixote.http_request import Upload
     from wcs import sql
     from wcs.qommon import get_publisher_class
     from wcs.qommon.afterjobs import AfterJob
     from wcs.qommon.cron import CronJob
-...
         assert pub.cfg['sp'] == {'what': 'ever'}
     def test_cron_command(settings):
         pub = create_temporary_pub()
         def clear_log_file():
             now = localtime()
             with open(os.path.join(pub.APP_DIR, 'cron.log-%s' % now.strftime('%Y%m%d')), 'w'):
     def clear_log_files():
         now = localtime()
         with open(os.path.join(get_publisher().APP_DIR, 'cron.log-%s' % now.strftime('%Y%m%d')), 'w'):
             pass
         for filepath in glob.glob(os.path.join(get_publisher().APP_DIR, '*', 'cron.log-*')):
             with open(filepath, 'w'):
                 pass
         def get_logs():
             now = localtime()
             with open(os.path.join(pub.APP_DIR, 'cron.log-%s' % now.strftime('%Y%m%d'))) as fd:
                 lines = fd.readlines()
             lines = [line[33:].strip() for line in lines]  # 33 chars for date & time
             return lines
         offset = ord(settings.SECRET_KEY[-1]) % 60
         with mock.patch('tempfile.gettempdir') as gettempdir:
             gettempdir.side_effect = lambda: pub.app_dir
             hostnames = ['example.net', 'foo.bar', 'something.com']
             for hostname in hostnames:
                 if not os.path.exists(os.path.join(pub.APP_DIR, hostname)):
                     os.mkdir(os.path.join(pub.APP_DIR, hostname))
                     # add a config.pck with postgresql configuration
                     with open(os.path.join(pub.APP_DIR, hostname, 'config.pck'), 'wb') as fd:
                         pickle.dump(pub.cfg, file=fd)
     def get_logs(hostname=None):
         pub = get_publisher()
         now = localtime()
         if hostname:
             base_dir = os.path.join(pub.APP_DIR, hostname)
         else:
             base_dir = pub.APP_DIR
         with open(os.path.join(base_dir, 'cron.log-%s' % now.strftime('%Y%m%d'))) as fd:
             lines = fd.readlines()
         lines = [line[33:].strip() for line in lines]  # 33 chars for date & time
         return lines
             with mock.patch('wcs.qommon.management.commands.cron.cron_worker') as cron_worker:
                 with mock.patch('wcs.qommon.publisher.QommonPublisher.get_tenants') as mock_tenants:
                     mock_tenants.return_value = [
                         Tenant(os.path.join(pub.app_dir, x)) for x in ('example.net', 'foo.bar', 'something.com')
+                    ]
                     clear_log_file()
                     call_command('cron')
                     assert cron_worker.call_count == 3
                     assert get_logs() == [
                         'starting cron (minutes offset is %s)' % offset,
                         '[tenant example.net] start',
                         '[tenant foo.bar] start',
                         '[tenant something.com] start',
+                    ]
                     cron_worker.reset_mock()
                     clear_log_file()
                     call_command('cron', domain='example.net')
                     assert cron_worker.call_count == 1
                     assert get_logs() == [
                         'starting cron (minutes offset is %s)' % offset,
                         '[tenant example.net] start',
+                    ]
                     cron_worker.reset_mock()
                     # disable cron on something.com
                     site_options_path = os.path.join(pub.APP_DIR, 'something.com', 'site-options.cfg')
                     with open(site_options_path, 'w') as fd:
                         fd.write(
                             '''\
                             [variables]
                             disable_cron_jobs = True
                             '''
+                        )
                     clear_log_file()
                     call_command('cron')
                     assert cron_worker.call_count == 2
                     assert get_logs() == [
                         'starting cron (minutes offset is %s)' % offset,
                         '[tenant example.net] start',
                         '[tenant foo.bar] start',
+                    ]
                     cron_worker.reset_mock()
                     os.unlink(site_options_path)
             # simulate another locked cron
             from wcs.qommon.vendor import locket
             lockfile = os.path.join(tempfile.gettempdir(), 'wcs-cron-in-progress.lock')
             with locket.lock_file(lockfile, timeout=0):
                 with mock.patch('wcs.qommon.management.commands.cron.cron_worker') as cron_worker:
                     call_command('cron')  # silent by default (verbosity=0)
                     assert cron_worker.call_count == 0
                     call_command('cron', verbosity=2)  # same if verbosity>0
                     assert cron_worker.call_count == 0
                     with mock.patch('wcs.qommon.management.commands.cron.JUMP_TIMEOUT_INTERVAL', -1):
                         with pytest.raises(CommandError, match='can not start cron job.*seems old'):
                             call_command('cron')
                     assert cron_worker.call_count == 0
             # verify that the lock is released
             with mock.patch('wcs.qommon.management.commands.cron.cron_worker') as cron_worker:
                 call_command('cron', domain='example.net')
                 assert cron_worker.call_count == 1
             # simulate a cron crash
             with mock.patch('wcs.qommon.management.commands.cron.cron_worker') as cron_worker:
                 cron_worker.side_effect = NotImplementedError
                 with pytest.raises(NotImplementedError):
                     call_command('cron', domain='example.net')
                 assert cron_worker.call_count == 1
             # verify that the lock is released
             with mock.patch('wcs.qommon.management.commands.cron.cron_worker') as cron_worker:
                 call_command('cron', domain='example.net')
                 assert cron_worker.call_count == 1
     def get_sql_cron_statuses():
         conn, cur = sql.get_connection_and_cursor()
         cur.execute("SELECT key, value FROM wcs_meta WHERE key LIKE 'cron-status-%'")
         rows = cur.fetchall()
         conn.commit()
         cur.close()
         return dict(rows)
             # disable cron system
             with override_settings(DISABLE_CRON_JOBS=True):
                 with mock.patch('wcs.qommon.management.commands.cron.cron_worker') as cron_worker:
                     call_command('cron', domain='example.net')
                     assert cron_worker.call_count == 0
             # run a specific job
             jobs = []
     def test_cron_command(settings):
         pub = create_temporary_pub()
         offset = ord(settings.SECRET_KEY[-1]) % 60
             def job1(pub, job=None):
                 jobs.append('job1')
             def job2(pub, job=None):
                 jobs.append('job2')
             def job3(pub, job=None):
                 jobs.append('job3')
                 for key in ['foo', 'bar', 'blah']:
                     with job.log_long_job(key):
                         pass
             @classmethod
             def register_test_cronjobs(cls):
                 cls.register_cronjob(CronJob(job1, days=[10]))
                 cls.register_cronjob(CronJob(job2, name='job2', days=[10]))
                 cls.register_cronjob(CronJob(job3, name='job3', days=[10]))
             with mock.patch('wcs.publisher.WcsPublisher.register_cronjobs', register_test_cronjobs):
                 get_publisher_class().cronjobs = []
                 call_command('cron', job_name='job1', domain='example.net')
                 assert jobs == []
                 get_publisher_class().cronjobs = []
                 clear_log_file()
                 call_command('cron', job_name='job2', domain='example.net')
                 assert jobs == ['job2']
         hostnames = ['example.net', 'foo.bar', 'something.com']
         for hostname in hostnames:
             if not os.path.exists(os.path.join(pub.APP_DIR, hostname)):
                 os.mkdir(os.path.join(pub.APP_DIR, hostname))
                 # add a config.pck with postgresql configuration
                 with open(os.path.join(pub.APP_DIR, hostname, 'config.pck'), 'wb') as fd:
                     pickle.dump(pub.cfg, file=fd)
         with mock.patch('wcs.qommon.management.commands.cron.cron_worker') as cron_worker:
             with mock.patch('wcs.qommon.publisher.QommonPublisher.get_tenants') as mock_tenants:
                 mock_tenants.return_value = [
                     Tenant(os.path.join(pub.app_dir, x)) for x in ('example.net', 'foo.bar', 'something.com')
+                ]
                 clear_log_files()
                 call_command('cron')
                 assert cron_worker.call_count == 3
                 assert get_logs() == [
                     'starting cron (minutes offset is %s)' % offset,
                     '[tenant example.net] start',
+                ]
                 get_publisher_class().cronjobs = []
                 jobs = []
                 clear_log_file()
                 with mock.patch('wcs.qommon.cron.CronJob.LONG_JOB_DURATION', 0):
                     call_command('cron', job_name='job2', domain='example.net')
                 cron_worker.reset_mock()
                 clear_log_files()
                 call_command('cron', domain='example.net')
                 assert cron_worker.call_count == 1
                 assert get_logs() == [
                     'starting cron (minutes offset is %s)' % offset,
                     '[tenant example.net] start',
                     '[tenant example.net] long job: job2 (took 0 minutes)',
+                ]
                 assert jobs == ['job2']
                 get_publisher_class().cronjobs = []
                 jobs = []
                 clear_log_file()
                 with mock.patch('wcs.qommon.cron.CronJob.LONG_JOB_DURATION', 0):
                     call_command('cron', job_name='job3', domain='example.net')
                 cron_worker.reset_mock()
                 # check we're still running them all
                 call_command('cron')
                 assert cron_worker.call_count == 3
                 cron_worker.reset_mock()
                 assert get_sql_cron_statuses() == {
                     'cron-status-example.net': 'needed',
                     'cron-status-foo.bar': 'needed',
                     'cron-status-something.com': 'needed',
+                }
                 # disable cron on something.com
                 site_options_path = os.path.join(pub.APP_DIR, 'something.com', 'site-options.cfg')
                 with open(site_options_path, 'w') as fd:
                     fd.write(
                         '''\
                         [variables]
                         disable_cron_jobs = True
                         '''
+                    )
                 clear_log_files()
                 call_command('cron')
                 assert cron_worker.call_count == 2
                 assert get_logs() == [
                     'starting cron (minutes offset is %s)' % offset,
                     '[tenant example.net] start',
                     '[tenant example.net] job3: running on "foo" took 0 minutes',
                     '[tenant example.net] job3: running on "bar" took 0 minutes',
                     '[tenant example.net] job3: running on "blah" took 0 minutes',
                     '[tenant example.net] long job: job3 (took 0 minutes)',
+                ]
                 assert jobs == ['job3']
                 os.unlink(site_options_path)
                 assert get_sql_cron_statuses() == {
                     'cron-status-example.net': 'needed',
                     'cron-status-foo.bar': 'needed',
                     'cron-status-something.com': 'needed',
+                }
                 # simulate a running cron
                 cron_worker.reset_mock()
                 settings.CRON_WORKERS = 1
                 get_publisher().set_tenant_by_hostname('example.net')
                 sql.mark_cron_status('running')
                 call_command('cron')
                 assert cron_worker.call_count == 0
                 assert get_sql_cron_statuses() == {
                     'cron-status-example.net': 'running',
                     'cron-status-foo.bar': 'needed',
                     'cron-status-something.com': 'needed',
+                }
                 # with one more worker, the two other tenants can be run sequentially
                 cron_worker.reset_mock()
                 settings.CRON_WORKERS = 2
                 call_command('cron')
                 assert cron_worker.call_count == 2
                 assert get_sql_cron_statuses() == {
                     'cron-status-example.net': 'running',
                     'cron-status-foo.bar': 'done',
                     'cron-status-something.com': 'done',
+                }
                 get_publisher().set_tenant_by_hostname('example.net')
                 sql.mark_cron_status('running')
                 get_publisher().set_tenant_by_hostname('foo.bar')
                 sql.mark_cron_status('running')
                 get_publisher().set_tenant_by_hostname('something.com')
                 sql.mark_cron_status('needed')
                 assert get_sql_cron_statuses() == {
                     'cron-status-example.net': 'running',
                     'cron-status-foo.bar': 'running',
                     'cron-status-something.com': 'needed',
+                }
                 cron_worker.reset_mock()
                 call_command('cron')
                 assert cron_worker.call_count == 0
                 shutil.rmtree(os.path.join(pub.APP_DIR, 'foo.bar'))
                 shutil.rmtree(os.path.join(pub.APP_DIR, 'something.com'))
         get_publisher().set_tenant_by_hostname('example.net')
         sql.mark_cron_status('needed')
         # simulate a cron crash
         with mock.patch('wcs.qommon.management.commands.cron.cron_worker') as cron_worker:
             cron_worker.side_effect = NotImplementedError
             with pytest.raises(NotImplementedError):
                 call_command('cron')
             assert cron_worker.call_count == 1
         # verify that the job is not marked as running
         assert get_sql_cron_statuses().get('cron-status-example.net') == 'done'
         # disable cron system
         with override_settings(DISABLE_CRON_JOBS=True):
             with mock.patch('wcs.qommon.management.commands.cron.cron_worker') as cron_worker:
                 call_command('cron', domain='example.net')
                 assert cron_worker.call_count == 0
     def test_cron_command_jobs(settings):
         create_temporary_pub()
         # run a specific job
         jobs = []
         def job1(pub, job=None):
             jobs.append('job1')
         def job2(pub, job=None):
             jobs.append('job2')
         def job3(pub, job=None):
             jobs.append('job3')
             for key in ['foo', 'bar', 'blah']:
                 with job.log_long_job(key):
                     pass
         @classmethod
         def register_test_cronjobs(cls):
             cls.register_cronjob(CronJob(job1, days=[10]))
             cls.register_cronjob(CronJob(job2, name='job2', days=[10]))
             cls.register_cronjob(CronJob(job3, name='job3', days=[10]))
         get_publisher().set_tenant_by_hostname('example.net')
         sql.mark_cron_status('needed')
         with mock.patch('wcs.publisher.WcsPublisher.register_cronjobs', register_test_cronjobs):
             get_publisher_class().cronjobs = []
             call_command('cron', job_name='job1', domain='example.net')
             assert jobs == []
             get_publisher_class().cronjobs = []
             clear_log_files()
             call_command('cron', job_name='job2', domain='example.net')
             assert jobs == ['job2']
             assert get_logs('example.net') == ['start']
             get_publisher_class().cronjobs = []
             jobs = []
             clear_log_files()
             with mock.patch('wcs.qommon.cron.CronJob.LONG_JOB_DURATION', 0):
                 call_command('cron', job_name='job2', domain='example.net')
             assert get_logs('example.net') == ['start', 'long job: job2 (took 0 minutes)']
             assert jobs == ['job2']
             get_publisher_class().cronjobs = []
             jobs = []
             clear_log_files()
             with mock.patch('wcs.qommon.cron.CronJob.LONG_JOB_DURATION', 0):
                 call_command('cron', job_name='job3', domain='example.net')
             assert get_logs('example.net') == [
                 'start',
                 'job3: running on "foo" took 0 minutes',
                 'job3: running on "bar" took 0 minutes',
                 'job3: running on "blah" took 0 minutes',
                 'long job: job3 (took 0 minutes)',
+            ]
             assert jobs == ['job3']
     def test_cron_command_delayed_jobs(settings, freezer):
         pub = create_temporary_pub()
         offset = ord(settings.SECRET_KEY[-1]) % 60
         with mock.patch('tempfile.gettempdir') as gettempdir:
             gettempdir.side_effect = lambda: pub.app_dir
             jobs = []
         jobs = []
             def job1(pub, job=None):
                 jobs.append('job1')
         def job1(pub, job=None):
             jobs.append('job1')
             def job2(pub, job=None):
                 jobs.append('job2')
         def job2(pub, job=None):
             jobs.append('job2')
             def job3(pub, job=None):
                 jobs.append('job3')
         def job3(pub, job=None):
             jobs.append('job3')
             start_time = datetime.datetime(2021, 4, 6, 2, offset)
         start_time = datetime.datetime(2021, 4, 6, 2, offset)
             @classmethod
             def register_test_cronjobs(cls):
                 cls.register_cronjob(CronJob(job1, minutes=[0, 3]))
                 cls.register_cronjob(CronJob(job2, minutes=[2]))
                 cls.register_cronjob(CronJob(job3, minutes=[10]))
         @classmethod
         def register_test_cronjobs(cls):
             cls.register_cronjob(CronJob(job1, minutes=[0, 3]))
             cls.register_cronjob(CronJob(job2, minutes=[2]))
             cls.register_cronjob(CronJob(job3, minutes=[10]))
             freezer.move_to(start_time)
             with mock.patch('wcs.publisher.WcsPublisher.register_cronjobs', register_test_cronjobs):
                 get_publisher_class().cronjobs = []
                 call_command('cron', domain='example.net')
                 assert jobs == ['job1']
         freezer.move_to(start_time)
         with mock.patch('wcs.publisher.WcsPublisher.register_cronjobs', register_test_cronjobs):
             get_publisher_class().cronjobs = []
             call_command('cron', domain='example.net')
             assert jobs == ['job1']
             # reset cronjobs
             pub.cronjobs = []
         # reset cronjobs
         pub.cronjobs = []
             jobs = []
         jobs = []
             def job1_delay(pub, job=None):
                 jobs.append('job1')
                 freezer.move_to(datetime.timedelta(minutes=5))
         def job1_delay(pub, job=None):
             jobs.append('job1')
             freezer.move_to(datetime.timedelta(minutes=5))
             @classmethod
             def register_test_cronjobs_2(cls):
                 cls.register_cronjob(CronJob(job1_delay, minutes=[0, 3]))
                 cls.register_cronjob(CronJob(job2, minutes=[2]))
                 cls.register_cronjob(CronJob(job3, minutes=[10]))
         @classmethod
         def register_test_cronjobs_2(cls):
             cls.register_cronjob(CronJob(job1_delay, minutes=[0, 3]))
             cls.register_cronjob(CronJob(job2, minutes=[2]))
             cls.register_cronjob(CronJob(job3, minutes=[10]))
             freezer.move_to(start_time)
             with mock.patch('wcs.publisher.WcsPublisher.register_cronjobs', register_test_cronjobs_2):
                 get_publisher_class().cronjobs = []
                 call_command('cron', domain='example.net')
                 assert jobs == ['job1', 'job2']
         freezer.move_to(start_time)
         with mock.patch('wcs.publisher.WcsPublisher.register_cronjobs', register_test_cronjobs_2):
             get_publisher_class().cronjobs = []
             call_command('cron', domain='example.net')
             assert jobs == ['job1', 'job2']
     def test_clean_afterjobs():

                     self.log('long job: %s (took %s minutes)' % (self.name, minutes))
         @staticmethod
         def log(message, with_tenant=True):
             tenant_prefix = ''
         def log(message, in_tenant=True):
             now = localtime()
             if with_tenant:
                 tenant_prefix = '[tenant %s] ' % get_publisher().tenant.hostname
             with open(os.path.join(get_publisher().APP_DIR, 'cron.log-%s' % now.strftime('%Y%m%d')), 'a+') as fd:
                 fd.write('%s %s%s\n' % (now.isoformat(), tenant_prefix, message))
             if in_tenant:
                 base_dir = get_publisher().tenant.directory
             else:
                 base_dir = get_publisher().APP_DIR
             with open(os.path.join(base_dir, 'cron.log-%s' % now.strftime('%Y%m%d')), 'a+') as fd:
                 fd.write('%s %s\n' % (now.isoformat(), message))
         def is_time(self, timetuple):
             minutes = self.minutes

     # You should have received a copy of the GNU General Public License
     # along with this program; if not, see <http://www.gnu.org/licenses/>.
     import os
     import tempfile
     import time
     from django.conf import settings
     from django.core.management.base import BaseCommand, CommandError
     from django.core.management.base import BaseCommand
     from wcs import sql
     from wcs.qommon.cron import CronJob, cron_worker
     from wcs.qommon.publisher import get_publisher_class
     from wcs.qommon.vendor import locket
     from wcs.wf.jump import JUMP_TIMEOUT_INTERVAL
     class Command(BaseCommand):
-...
+            )
             parser.add_argument('--job', dest='job_name', metavar='NAME')
         def handle(self, verbosity, **options):
         def handle(self, verbosity, domain=None, job_name=None, **options):
             if getattr(settings, 'DISABLE_CRON_JOBS', False) and not options['force_job']:
                 if verbosity > 1:
                     print('Command is ignored because DISABLE_CRON_JOBS is set in settings')
                 return
             lockfile = os.path.join(tempfile.gettempdir(), 'wcs-cron-in-progress.lock')
             if options.get('domain'):
                 domains = [options.get('domain')]
             if domain:
                 domains = [domain]
             else:
                 domains = [x.hostname for x in get_publisher_class().get_tenants()]
             try:
                 with locket.lock_file(lockfile, timeout=0):
                     if verbosity > 2:
                         print('cron start (lock %s)' % lockfile)
                     now = time.localtime()
                     publisher_class = get_publisher_class()
                     publisher_class.register_cronjobs()
                     publisher = publisher_class.create_publisher()
                     offset = ord(settings.SECRET_KEY[-1]) % 60
                     CronJob.log('starting cron (minutes offset is %s)' % offset, with_tenant=False)
                     for hostname in domains:
                         publisher.set_tenant_by_hostname(hostname)
                         if publisher.get_site_option('disable_cron_jobs', 'variables'):
                             if verbosity > 1:
                                 print('cron ignored on %s because DISABLE_CRON_JOBS is set' % hostname)
                             continue
                         if not publisher.has_postgresql_config():
                             if verbosity > 1:
                                 print('cron ignored on %s because it has no PostgreSQL configuration' % hostname)
                             continue
             if not job_name and verbosity > 2:
                 print('cron start')
             now = time.localtime()
             publisher_class = get_publisher_class()
             publisher_class.register_cronjobs()
             publisher = publisher_class.create_publisher()
             offset = ord(settings.SECRET_KEY[-1]) % 60
             if not job_name:
                 CronJob.log('starting cron (minutes offset is %s)' % offset, in_tenant=False)
             tenant_status = {
                 'needed': [],
                 'running': [],
                 'done': [],
+            }
             for hostname in domains:
                 publisher.set_tenant_by_hostname(hostname)
                 if publisher.get_site_option('disable_cron_jobs', 'variables'):
                     if verbosity > 1:
                         print('cron ignored on %s because DISABLE_CRON_JOBS is set' % hostname)
                     continue
                 if not publisher.has_postgresql_config():
                     if verbosity > 1:
                         print('cron ignored on %s because it has no PostgreSQL configuration' % hostname)
                     continue
                 if domain:
                     cron_status = 'ignored'
                 else:
                     cron_status = sql.get_and_update_cron_status()
                     tenant_status[cron_status].append(hostname)
                 if not options['force_job']:
                     if len(tenant_status['running']) >= settings.CRON_WORKERS:
                         if cron_status == 'needed':
                             # unmark current tenant as being running
                             sql.mark_cron_status('needed')
                         if verbosity > 1:
                             print('cron work on %s' % hostname)
                         CronJob.log('start')
                         try:
                             cron_worker(publisher, now, job_name=options.get('job_name'))
                         except Exception as e:
                             CronJob.log('aborted (%r)' % e)
                             raise e
                 if verbosity > 2:
                     print('cron end (release lock %s)' % lockfile)
             except locket.LockError:
                 age = int(time.time() - os.stat(lockfile).st_mtime)
                 if age > JUMP_TIMEOUT_INTERVAL * 10:
                     raise CommandError(
                         'can not start cron job, locked by %s for %s seconds (seems old)' % (lockfile, age)
+                    )
                 if verbosity > 0:
                     print('can not start cron job, locked by %s for %s seconds' % (lockfile, age))
                             print(hostname, 'skip running, too many workers')
                         break
                     if cron_status in ('running', 'done'):
                         if verbosity > 1:
                             print(hostname, 'skip running, already handled')
                         continue
                 if verbosity > 1:
                     print('cron work on %s' % hostname)
                 CronJob.log('start')
                 try:
                     cron_worker(publisher, now, job_name=job_name)
                 except Exception as e:
                     CronJob.log('aborted (%r)' % e)
                     raise e
                 finally:
                     if not domain:
                         tenant_status['needed'].remove(hostname)
                         tenant_status['done'].append(hostname)
                         sql.mark_cron_status('done')
             if tenant_status['done'] and not (tenant_status['needed'] or tenant_status['running']):
                 # reset everything if no hostnames are marked as needed or running
                 for hostname in tenant_status['done']:
                     publisher.set_tenant_by_hostname(hostname)
                     sql.mark_cron_status('needed')
             if verbosity > 1:
                 print('cron end')

     # workalendar config
     WORKING_DAY_CALENDAR = 'workalendar.europe.France'
     # CRON_WORKERS
     CRON_WORKERS = os.cpu_count() // 2 + 1
     local_settings_file = os.environ.get(
         'WCS_SETTINGS_FILE', os.path.join(os.path.dirname(__file__), 'local_settings.py')
+    )

         return sql_level
     @guard_postgres
     def get_and_update_cron_status():
         conn, cur = get_connection_and_cursor()
         do_meta_table(conn, cur, insert_current_sql_level=False)
         key = 'cron-status-%s' % get_publisher().tenant.hostname
         cur.execute("SELECT value FROM wcs_meta WHERE key = %s", (key,))
         row = cur.fetchone()
         if row is None:
             cur.execute("INSERT INTO wcs_meta (id, key, value) VALUES (DEFAULT, %s, 'running')", (key,))
             status = 'needed'
         elif row[0] == 'needed':
             cur.execute("UPDATE wcs_meta SET value = 'running', updated_at = NOW() WHERE key = %s", (key,))
             status = 'needed'
         else:
             status = row[0]
         conn.commit()
         cur.close()
         return status
     @guard_postgres
     def mark_cron_status(status):
         conn, cur = get_connection_and_cursor()
         key = 'cron-status-%s' % get_publisher().tenant.hostname
         cur.execute("UPDATE wcs_meta SET value = %s, updated_at = NOW() WHERE key = %s", (status, key))
         conn.commit()
         cur.close()
     @guard_postgres
     def is_reindex_needed(index, conn, cur):
         do_meta_table(conn, cur, insert_current_sql_level=False)
+    -

Projet

Général

Profil

Produits Entr'ouvert » w.c.s.

0001-cron-allow-some-cron-workers-to-be-run-in-parallel-3.patch