diff options
| -rw-r--r-- | Dockerfile | 1 | ||||
| -rw-r--r-- | doc/source/monitoring.rst | 11 | ||||
| -rw-r--r-- | noxfile.py | 2 | ||||
| -rw-r--r-- | requirements.txt | 2 | ||||
| -rwxr-xr-x | tests/make_playbooks.py | 3 | ||||
| -rw-r--r-- | tests/unit/test_connection.py | 192 | ||||
| -rw-r--r-- | tests/unit/test_gerrit.py | 45 | ||||
| -rw-r--r-- | tests/unit/test_git_driver.py | 3 | ||||
| -rw-r--r-- | tests/unit/test_reporting.py | 4 | ||||
| -rw-r--r-- | tests/unit/test_scheduler.py | 4 | ||||
| -rw-r--r-- | web/public/openapi.yaml | 40 | ||||
| -rw-r--r-- | zuul/ansible/logconfig.py | 3 | ||||
| -rw-r--r-- | zuul/configloader.py | 6 | ||||
| -rw-r--r-- | zuul/driver/gerrit/gerritconnection.py | 9 | ||||
| -rw-r--r-- | zuul/driver/sql/alembic/env.py | 3 | ||||
| -rw-r--r-- | zuul/driver/sql/alembic/versions/60c119eb1e3f_use_build_set_results.py | 17 | ||||
| -rw-r--r-- | zuul/driver/sql/alembic/versions/c7467b642498_buildset_updated.py | 17 | ||||
| -rw-r--r-- | zuul/driver/sql/sqlconnection.py | 40 | ||||
| -rw-r--r-- | zuul/lib/repl.py | 6 | ||||
| -rw-r--r-- | zuul/model.py | 11 | ||||
| -rw-r--r-- | zuul/scheduler.py | 123 |
21 files changed, 330 insertions, 212 deletions
diff --git a/Dockerfile b/Dockerfile index 5a17739f0..df326bd8a 100644 --- a/Dockerfile +++ b/Dockerfile @@ -29,6 +29,7 @@ ARG REACT_APP_ENABLE_SERVICE_WORKER # Kubectl/Openshift version/sha ARG OPENSHIFT_URL=https://mirror.openshift.com/pub/openshift-v4/x86_64/clients/ocp/4.11.20/openshift-client-linux-4.11.20.tar.gz ARG OPENSHIFT_SHA=74f252c812932425ca19636b2be168df8fe57b114af6b114283975e67d987d11 +ARG PBR_VERSION= COPY . /tmp/src COPY --from=js-builder /tmp/src/build /tmp/src/zuul/web/static diff --git a/doc/source/monitoring.rst b/doc/source/monitoring.rst index 7f99c7f7f..f40bee445 100644 --- a/doc/source/monitoring.rst +++ b/doc/source/monitoring.rst @@ -444,6 +444,11 @@ These metrics are emitted by the Zuul :ref:`scheduler`: Indicates if the executor is paused. 1 means paused else 0. + .. stat:: pct_used_hdd + :type: gauge + + The used disk on this executor, as a percentage multiplied by 100. + .. stat:: pct_used_ram :type: gauge @@ -711,6 +716,12 @@ These metrics are emitted by the Zuul :ref:`scheduler`: The size of the current connection event queue. + .. stat:: run_handler + :type: timer + + A timer metric reporting the time taken for one scheduler run + handler iteration. + .. stat:: time_query :type: timer diff --git a/noxfile.py b/noxfile.py index e920f053e..30058cef7 100644 --- a/noxfile.py +++ b/noxfile.py @@ -32,7 +32,6 @@ def set_standard_env_vars(session): set_env(session, 'OS_STDERR_CAPTURE', '1') set_env(session, 'OS_STDOUT_CAPTURE', '1') set_env(session, 'OS_TEST_TIMEOUT', '360') - set_env(session, 'SQLALCHEMY_WARN_20', '1') session.env['PYTHONWARNINGS'] = ','.join([ 'always::DeprecationWarning:zuul.driver.sql.sqlconnection', 'always::DeprecationWarning:tests.base', @@ -49,7 +48,6 @@ def set_standard_env_vars(session): @nox.session(python='3') def bindep(session): set_standard_env_vars(session) - set_env(session, 'SQLALCHEMY_WARN_20', '1') session.install('bindep') session.run('bindep', 'test') diff --git a/requirements.txt b/requirements.txt index 082a18043..e12e9cf72 100644 --- a/requirements.txt +++ b/requirements.txt @@ -18,7 +18,7 @@ PrettyTable>=0.6,<0.8 babel>=1.0 netaddr kazoo>=2.9.0 -sqlalchemy<2.0.0 +sqlalchemy>=2.0.0 alembic cryptography>=39.0.0 cachecontrol<0.12.7 diff --git a/tests/make_playbooks.py b/tests/make_playbooks.py index 93c37bc81..cb7a98096 100755 --- a/tests/make_playbooks.py +++ b/tests/make_playbooks.py @@ -40,7 +40,8 @@ def handle_repo(path): config_path = os.path.join(path, fn) break try: - config = yaml.safe_load(open(config_path)) + with open(config_path) as f: + config = yaml.safe_load(f) except Exception: print(" Has yaml errors") return diff --git a/tests/unit/test_connection.py b/tests/unit/test_connection.py index c30e1743d..26a99215e 100644 --- a/tests/unit/test_connection.py +++ b/tests/unit/test_connection.py @@ -65,7 +65,7 @@ class TestSQLConnectionMysql(ZuulTestCase): def _sql_tables_created(self, connection_name): connection = self.scheds.first.connections.connections[connection_name] - insp = sa.engine.reflection.Inspector(connection.engine) + insp = sa.inspect(connection.engine) table_prefix = connection.table_prefix self.assertEqual(self.expected_table_prefix, table_prefix) @@ -82,7 +82,7 @@ class TestSQLConnectionMysql(ZuulTestCase): def _sql_indexes_created(self, connection_name): connection = self.scheds.first.connections.connections[connection_name] - insp = sa.engine.reflection.Inspector(connection.engine) + insp = sa.inspect(connection.engine) table_prefix = connection.table_prefix self.assertEqual(self.expected_table_prefix, table_prefix) @@ -127,7 +127,7 @@ class TestSQLConnectionMysql(ZuulTestCase): engine.connect() as conn: result = conn.execute( - sa.sql.select([reporter.connection.zuul_buildset_table])) + sa.sql.select(reporter.connection.zuul_buildset_table)) buildsets = result.fetchall() self.assertEqual(5, len(buildsets)) @@ -137,107 +137,107 @@ class TestSQLConnectionMysql(ZuulTestCase): buildset3 = buildsets[3] buildset4 = buildsets[4] - self.assertEqual('check', buildset0['pipeline']) - self.assertEqual('org/project', buildset0['project']) - self.assertEqual(1, buildset0['change']) - self.assertEqual('1', buildset0['patchset']) - self.assertEqual('SUCCESS', buildset0['result']) - self.assertEqual('Build succeeded.', buildset0['message']) - self.assertEqual('tenant-one', buildset0['tenant']) + self.assertEqual('check', buildset0.pipeline) + self.assertEqual('org/project', buildset0.project) + self.assertEqual(1, buildset0.change) + self.assertEqual('1', buildset0.patchset) + self.assertEqual('SUCCESS', buildset0.result) + self.assertEqual('Build succeeded.', buildset0.message) + self.assertEqual('tenant-one', buildset0.tenant) self.assertEqual( - 'https://review.example.com/%d' % buildset0['change'], - buildset0['ref_url']) - self.assertNotEqual(None, buildset0['event_id']) - self.assertNotEqual(None, buildset0['event_timestamp']) + 'https://review.example.com/%d' % buildset0.change, + buildset0.ref_url) + self.assertNotEqual(None, buildset0.event_id) + self.assertNotEqual(None, buildset0.event_timestamp) buildset0_builds = conn.execute( - sa.sql.select([ + sa.sql.select( reporter.connection.zuul_build_table - ]).where( + ).where( reporter.connection.zuul_build_table.c.buildset_id == - buildset0['id'] + buildset0.id ) ).fetchall() # Check the first result, which should be the project-merge job self.assertEqual( - 'project-merge', buildset0_builds[0]['job_name']) - self.assertEqual("SUCCESS", buildset0_builds[0]['result']) - self.assertEqual(None, buildset0_builds[0]['log_url']) - self.assertEqual('check', buildset1['pipeline']) - self.assertEqual('master', buildset1['branch']) - self.assertEqual('org/project', buildset1['project']) - self.assertEqual(2, buildset1['change']) - self.assertEqual('1', buildset1['patchset']) - self.assertEqual('FAILURE', buildset1['result']) - self.assertEqual('Build failed.', buildset1['message']) + 'project-merge', buildset0_builds[0].job_name) + self.assertEqual("SUCCESS", buildset0_builds[0].result) + self.assertEqual(None, buildset0_builds[0].log_url) + self.assertEqual('check', buildset1.pipeline) + self.assertEqual('master', buildset1.branch) + self.assertEqual('org/project', buildset1.project) + self.assertEqual(2, buildset1.change) + self.assertEqual('1', buildset1.patchset) + self.assertEqual('FAILURE', buildset1.result) + self.assertEqual('Build failed.', buildset1.message) buildset1_builds = conn.execute( - sa.sql.select([ + sa.sql.select( reporter.connection.zuul_build_table - ]).where( + ).where( reporter.connection.zuul_build_table.c.buildset_id == - buildset1['id'] + buildset1.id ) ).fetchall() # Check the second result, which should be the project-test1 # job which failed self.assertEqual( - 'project-test1', buildset1_builds[1]['job_name']) - self.assertEqual("FAILURE", buildset1_builds[1]['result']) - self.assertEqual(None, buildset1_builds[1]['log_url']) + 'project-test1', buildset1_builds[1].job_name) + self.assertEqual("FAILURE", buildset1_builds[1].result) + self.assertEqual(None, buildset1_builds[1].log_url) buildset2_builds = conn.execute( - sa.sql.select([ + sa.sql.select( reporter.connection.zuul_build_table - ]).where( + ).where( reporter.connection.zuul_build_table.c.buildset_id == - buildset2['id'] + buildset2.id ) ).fetchall() # Check the first result, which should be the project-publish # job self.assertEqual('project-publish', - buildset2_builds[0]['job_name']) - self.assertEqual("SUCCESS", buildset2_builds[0]['result']) + buildset2_builds[0].job_name) + self.assertEqual("SUCCESS", buildset2_builds[0].result) buildset3_builds = conn.execute( - sa.sql.select([ + sa.sql.select( reporter.connection.zuul_build_table - ]).where( + ).where( reporter.connection.zuul_build_table.c.buildset_id == - buildset3['id'] + buildset3.id ) ).fetchall() self.assertEqual( - 'project-test1', buildset3_builds[1]['job_name']) - self.assertEqual('NODE_FAILURE', buildset3_builds[1]['result']) - self.assertEqual(None, buildset3_builds[1]['log_url']) - self.assertIsNotNone(buildset3_builds[1]['start_time']) - self.assertIsNotNone(buildset3_builds[1]['end_time']) + 'project-test1', buildset3_builds[1].job_name) + self.assertEqual('NODE_FAILURE', buildset3_builds[1].result) + self.assertEqual(None, buildset3_builds[1].log_url) + self.assertIsNotNone(buildset3_builds[1].start_time) + self.assertIsNotNone(buildset3_builds[1].end_time) self.assertGreaterEqual( - buildset3_builds[1]['end_time'], - buildset3_builds[1]['start_time']) + buildset3_builds[1].end_time, + buildset3_builds[1].start_time) # Check the paused build result buildset4_builds = conn.execute( - sa.sql.select([ + sa.sql.select( reporter.connection.zuul_build_table - ]).where( + ).where( reporter.connection.zuul_build_table.c.buildset_id == - buildset4['id'] + buildset4.id ).order_by(reporter.connection.zuul_build_table.c.id) ).fetchall() paused_build_events = conn.execute( - sa.sql.select([ + sa.sql.select( reporter.connection.zuul_build_event_table - ]).where( + ).where( reporter.connection.zuul_build_event_table.c.build_id - == buildset4_builds[0]["id"] + == buildset4_builds[0].id ) ).fetchall() @@ -245,16 +245,16 @@ class TestSQLConnectionMysql(ZuulTestCase): pause_event = paused_build_events[0] resume_event = paused_build_events[1] self.assertEqual( - pause_event["event_type"], "paused") - self.assertIsNotNone(pause_event["event_time"]) - self.assertIsNone(pause_event["description"]) + pause_event.event_type, "paused") + self.assertIsNotNone(pause_event.event_time) + self.assertIsNone(pause_event.description) self.assertEqual( - resume_event["event_type"], "resumed") - self.assertIsNotNone(resume_event["event_time"]) - self.assertIsNone(resume_event["description"]) + resume_event.event_type, "resumed") + self.assertIsNotNone(resume_event.event_time) + self.assertIsNone(resume_event.description) self.assertGreater( - resume_event["event_time"], pause_event["event_time"]) + resume_event.event_time, pause_event.event_time) self.executor_server.hold_jobs_in_build = True @@ -333,51 +333,51 @@ class TestSQLConnectionMysql(ZuulTestCase): engine.connect() as conn: result = conn.execute( - sa.sql.select([reporter.connection.zuul_buildset_table]) + sa.sql.select(reporter.connection.zuul_buildset_table) ) buildsets = result.fetchall() self.assertEqual(1, len(buildsets)) buildset0 = buildsets[0] - self.assertEqual('check', buildset0['pipeline']) - self.assertEqual('org/project', buildset0['project']) - self.assertEqual(1, buildset0['change']) - self.assertEqual('1', buildset0['patchset']) - self.assertEqual('SUCCESS', buildset0['result']) - self.assertEqual('Build succeeded.', buildset0['message']) - self.assertEqual('tenant-one', buildset0['tenant']) + self.assertEqual('check', buildset0.pipeline) + self.assertEqual('org/project', buildset0.project) + self.assertEqual(1, buildset0.change) + self.assertEqual('1', buildset0.patchset) + self.assertEqual('SUCCESS', buildset0.result) + self.assertEqual('Build succeeded.', buildset0.message) + self.assertEqual('tenant-one', buildset0.tenant) self.assertEqual( - 'https://review.example.com/%d' % buildset0['change'], - buildset0['ref_url']) + 'https://review.example.com/%d' % buildset0.change, + buildset0.ref_url) buildset0_builds = conn.execute( sa.sql.select( - [reporter.connection.zuul_build_table] + reporter.connection.zuul_build_table ).where( reporter.connection.zuul_build_table.c.buildset_id == - buildset0['id'] + buildset0.id ) ).fetchall() # Check the retry results - self.assertEqual('project-merge', buildset0_builds[0]['job_name']) - self.assertEqual('SUCCESS', buildset0_builds[0]['result']) - self.assertTrue(buildset0_builds[0]['final']) - - self.assertEqual('project-test1', buildset0_builds[1]['job_name']) - self.assertEqual('RETRY', buildset0_builds[1]['result']) - self.assertFalse(buildset0_builds[1]['final']) - self.assertEqual('project-test2', buildset0_builds[2]['job_name']) - self.assertEqual('RETRY', buildset0_builds[2]['result']) - self.assertFalse(buildset0_builds[2]['final']) - - self.assertEqual('project-test1', buildset0_builds[3]['job_name']) - self.assertEqual('SUCCESS', buildset0_builds[3]['result']) - self.assertTrue(buildset0_builds[3]['final']) - self.assertEqual('project-test2', buildset0_builds[4]['job_name']) - self.assertEqual('SUCCESS', buildset0_builds[4]['result']) - self.assertTrue(buildset0_builds[4]['final']) + self.assertEqual('project-merge', buildset0_builds[0].job_name) + self.assertEqual('SUCCESS', buildset0_builds[0].result) + self.assertTrue(buildset0_builds[0].final) + + self.assertEqual('project-test1', buildset0_builds[1].job_name) + self.assertEqual('RETRY', buildset0_builds[1].result) + self.assertFalse(buildset0_builds[1].final) + self.assertEqual('project-test2', buildset0_builds[2].job_name) + self.assertEqual('RETRY', buildset0_builds[2].result) + self.assertFalse(buildset0_builds[2].final) + + self.assertEqual('project-test1', buildset0_builds[3].job_name) + self.assertEqual('SUCCESS', buildset0_builds[3].result) + self.assertTrue(buildset0_builds[3].final) + self.assertEqual('project-test2', buildset0_builds[4].job_name) + self.assertEqual('SUCCESS', buildset0_builds[4].result) + self.assertTrue(buildset0_builds[4].final) self.executor_server.hold_jobs_in_build = True @@ -430,7 +430,7 @@ class TestSQLConnectionMysql(ZuulTestCase): engine.connect() as conn: result = conn.execute( - sa.sql.select([reporter.connection.zuul_buildset_table]) + sa.sql.select(reporter.connection.zuul_buildset_table) ) buildsets = result.fetchall() @@ -439,10 +439,10 @@ class TestSQLConnectionMysql(ZuulTestCase): buildset0_builds = conn.execute( sa.sql.select( - [reporter.connection.zuul_build_table] + reporter.connection.zuul_build_table ).where( reporter.connection.zuul_build_table.c.buildset_id == - buildset0['id'] + buildset0.id ) ).fetchall() @@ -488,7 +488,7 @@ class TestSQLConnectionMysql(ZuulTestCase): engine.connect() as conn: result = conn.execute( - sa.sql.select([reporter.connection.zuul_buildset_table]) + sa.sql.select(reporter.connection.zuul_buildset_table) ) buildsets = result.fetchall() @@ -497,10 +497,10 @@ class TestSQLConnectionMysql(ZuulTestCase): buildset0_builds = conn.execute( sa.sql.select( - [reporter.connection.zuul_build_table] + reporter.connection.zuul_build_table ).where( reporter.connection.zuul_build_table.c.buildset_id == - buildset0['id'] + buildset0.id ) ).fetchall() diff --git a/tests/unit/test_gerrit.py b/tests/unit/test_gerrit.py index 2e3057af6..2a63d5ef8 100644 --- a/tests/unit/test_gerrit.py +++ b/tests/unit/test_gerrit.py @@ -957,3 +957,48 @@ class TestGerritConnection(ZuulTestCase): self.assertEqual(B.queried, 2) self.assertEqual(A.data['status'], 'MERGED') self.assertEqual(B.data['status'], 'MERGED') + + +class TestGerritUnicodeRefs(ZuulTestCase): + config_file = 'zuul-gerrit-web.conf' + tenant_config_file = 'config/single-tenant/main.yaml' + + upload_pack_data = (b'014452944ee370db5c87691e62e0f9079b6281319b4e HEAD' + b'\x00multi_ack thin-pack side-band side-band-64k ' + b'ofs-delta shallow deepen-since deepen-not ' + b'deepen-relative no-progress include-tag ' + b'multi_ack_detailed allow-tip-sha1-in-want ' + b'allow-reachable-sha1-in-want ' + b'symref=HEAD:refs/heads/faster filter ' + b'object-format=sha1 agent=git/2.37.1.gl1\n' + b'003d5f42665d737b3fd4ec22ca0209e6191859f09fd6 ' + b'refs/for/faster\n' + b'004952944ee370db5c87691e62e0f9079b6281319b4e ' + b'refs/heads/foo/\xf0\x9f\x94\xa5\xf0\x9f\x94\xa5' + b'\xf0\x9f\x94\xa5\n' + b'003f52944ee370db5c87691e62e0f9079b6281319b4e ' + b'refs/heads/faster\n0000').decode("utf-8") + + def test_mb_unicode_refs(self): + gerrit_config = { + 'user': 'gerrit', + 'server': 'localhost', + } + driver = GerritDriver() + gerrit = GerritConnection(driver, 'review_gerrit', gerrit_config) + + def _uploadPack(project): + return self.upload_pack_data + + self.patch(gerrit, '_uploadPack', _uploadPack) + + project = gerrit.source.getProject('org/project') + refs = gerrit.getInfoRefs(project) + + self.assertEqual(refs, + {'refs/for/faster': + '5f42665d737b3fd4ec22ca0209e6191859f09fd6', + 'refs/heads/foo/🔥🔥🔥': + '52944ee370db5c87691e62e0f9079b6281319b4e', + 'refs/heads/faster': + '52944ee370db5c87691e62e0f9079b6281319b4e'}) diff --git a/tests/unit/test_git_driver.py b/tests/unit/test_git_driver.py index 06e2ac7c8..95fca30d3 100644 --- a/tests/unit/test_git_driver.py +++ b/tests/unit/test_git_driver.py @@ -62,7 +62,8 @@ class TestGitDriver(ZuulTestCase): # Update zuul.yaml to force a tenant reconfiguration path = os.path.join(self.upstream_root, 'common-config', 'zuul.yaml') - config = yaml.safe_load(open(path, 'r').read()) + with open(path, 'r') as f: + config = yaml.safe_load(f) change = { 'name': 'org/project', 'check': { diff --git a/tests/unit/test_reporting.py b/tests/unit/test_reporting.py index 2cf93cdcb..0c5c5fbc9 100644 --- a/tests/unit/test_reporting.py +++ b/tests/unit/test_reporting.py @@ -151,7 +151,7 @@ class TestReporting(ZuulTestCase): engine.connect() as conn: result = conn.execute( - sa.sql.select([reporter.connection.zuul_buildset_table])) + sa.sql.select(reporter.connection.zuul_buildset_table)) buildsets = result.fetchall() for x in buildsets: @@ -180,7 +180,7 @@ class TestReporting(ZuulTestCase): engine.connect() as conn: result = conn.execute( - sa.sql.select([reporter.connection.zuul_buildset_table])) + sa.sql.select(reporter.connection.zuul_buildset_table)) buildsets = result.fetchall() for x in buildsets: diff --git a/tests/unit/test_scheduler.py b/tests/unit/test_scheduler.py index 30db44b62..1a5657ed6 100644 --- a/tests/unit/test_scheduler.py +++ b/tests/unit/test_scheduler.py @@ -461,6 +461,7 @@ class TestScheduler(ZuulTestCase): 'zuul.mergers.online', value='1', kind='g') self.assertReportedStat('zuul.scheduler.eventqueues.connection.gerrit', value='0', kind='g') + self.assertReportedStat('zuul.scheduler.run_handler', kind='ms') # Catch time / monotonic errors for key in [ @@ -6216,7 +6217,8 @@ For CI problems and help debugging, contact ci@example.org""" build = self.getBuildByName('check-job') inv_path = os.path.join(build.jobdir.root, 'ansible', 'inventory.yaml') - inventory = yaml.safe_load(open(inv_path, 'r')) + with open(inv_path, 'r') as f: + inventory = yaml.safe_load(f) label = inventory['all']['hosts']['controller']['nodepool']['label'] self.assertEqual('slow-label', label) diff --git a/web/public/openapi.yaml b/web/public/openapi.yaml index d69111cf8..cb2e10b37 100644 --- a/web/public/openapi.yaml +++ b/web/public/openapi.yaml @@ -249,7 +249,7 @@ paths: - tenant /api/tenant/{tenant}/key/{project}.pub: get: - operationId: get-project-key + operationId: get-project-secrets-key parameters: - description: The tenant name in: path @@ -275,12 +275,44 @@ paths: ' schema: - description: The project public key + description: The project secrets public key in PKCS8 format type: string - description: Returns the project public key + description: Returns the project public key that is used to encrypt secrets '404': description: Tenant or Project not found - summary: Get a project public key + summary: Get a project public key that is used to encrypt secrets + tags: + - tenant + /api/tenant/{tenant}/project-ssh-key/{project}.pub: + get: + operationId: get-project-ssh-key + parameters: + - description: The tenant name + in: path + name: tenant + required: true + schema: + type: string + - description: The project name + in: path + name: project + required: true + schema: + type: string + responses: + '200': + content: + text/plain: + example: 'ssh-rsa AAAAB3NzaC1yc2EAAAADAQABAAACA + + ' + schema: + description: The project ssh public key in SSH2 format + type: string + description: Returns the project public key that executor adds to SSH agent + '404': + description: Tenant or Project not found + summary: Get a project public key that is used for SSH in post-merge pipelines tags: - tenant /api/tenant/{tenant}/semaphores: diff --git a/zuul/ansible/logconfig.py b/zuul/ansible/logconfig.py index 66881336a..2d7c37463 100644 --- a/zuul/ansible/logconfig.py +++ b/zuul/ansible/logconfig.py @@ -140,7 +140,8 @@ def _read_config_file(filename: str): raise ValueError("Unable to read logging config file at %s" % filename) if os.path.splitext(filename)[1] in ('.yml', '.yaml', '.json'): - return yaml.safe_load(open(filename, 'r')) + with open(filename, 'r') as f: + return yaml.safe_load(f) return filename diff --git a/zuul/configloader.py b/zuul/configloader.py index f01be597b..f9e52595e 100644 --- a/zuul/configloader.py +++ b/zuul/configloader.py @@ -2259,6 +2259,12 @@ class TenantParser(object): job.source_context.branch) with self.unparsed_config_cache.writeLock( job.source_context.project_canonical_name): + # Prevent files cache ltime from going backward + if files_cache.ltime >= job.ltime: + self.log.info( + "Discarding job %s result since the files cache was " + "updated in the meantime", job) + continue # Since the cat job returns all required config files # for ALL tenants the project is a part of, we can # clear the whole cache and then populate it with the diff --git a/zuul/driver/gerrit/gerritconnection.py b/zuul/driver/gerrit/gerritconnection.py index 0a1f0ee61..276365e1d 100644 --- a/zuul/driver/gerrit/gerritconnection.py +++ b/zuul/driver/gerrit/gerritconnection.py @@ -1643,7 +1643,10 @@ class GerritConnection(ZKChangeCacheMixin, ZKBranchCacheMixin, BaseConnection): def getInfoRefs(self, project: Project) -> Dict[str, str]: try: - data = self._uploadPack(project) + # Encode the UTF-8 data back to a byte array, as the size of + # each record in the pack is in bytes, and so the slicing must + # also be done on a byte-basis. + data = self._uploadPack(project).encode("utf-8") except Exception: self.log.error("Cannot get references from %s" % project) raise # keeps error information @@ -1662,7 +1665,9 @@ class GerritConnection(ZKChangeCacheMixin, ZKBranchCacheMixin, BaseConnection): plen -= 4 if len(data) - i < plen: raise Exception("Invalid data in info/refs") - line = data[i:i + plen] + # Once the pack data is sliced, we can safely decode it back + # into a (UTF-8) string. + line = data[i:i + plen].decode("utf-8") i += plen if not read_advertisement: read_advertisement = True diff --git a/zuul/driver/sql/alembic/env.py b/zuul/driver/sql/alembic/env.py index da7b3207f..17b67805e 100644 --- a/zuul/driver/sql/alembic/env.py +++ b/zuul/driver/sql/alembic/env.py @@ -53,7 +53,8 @@ def run_migrations_online(): connectable = engine_from_config( config.get_section(config.config_ini_section), prefix='sqlalchemy.', - poolclass=pool.NullPool) + poolclass=pool.NullPool, + future=True) # we can get the table prefix via the tag object tag = context.get_tag_argument() diff --git a/zuul/driver/sql/alembic/versions/60c119eb1e3f_use_build_set_results.py b/zuul/driver/sql/alembic/versions/60c119eb1e3f_use_build_set_results.py index 67581a6f9..1735d35f3 100644 --- a/zuul/driver/sql/alembic/versions/60c119eb1e3f_use_build_set_results.py +++ b/zuul/driver/sql/alembic/versions/60c119eb1e3f_use_build_set_results.py @@ -24,13 +24,16 @@ def upgrade(table_prefix=''): connection = op.get_bind() connection.execute( - """ - UPDATE {buildset_table} - SET result=( - SELECT CASE score - WHEN 1 THEN 'SUCCESS' - ELSE 'FAILURE' END) - """.format(buildset_table=table_prefix + BUILDSET_TABLE)) + sa.text( + """ + UPDATE {buildset_table} + SET result=( + SELECT CASE score + WHEN 1 THEN 'SUCCESS' + ELSE 'FAILURE' END) + """.format(buildset_table=table_prefix + BUILDSET_TABLE) + ) + ) op.drop_column(table_prefix + BUILDSET_TABLE, 'score') diff --git a/zuul/driver/sql/alembic/versions/c7467b642498_buildset_updated.py b/zuul/driver/sql/alembic/versions/c7467b642498_buildset_updated.py index abfba7247..99d12d750 100644 --- a/zuul/driver/sql/alembic/versions/c7467b642498_buildset_updated.py +++ b/zuul/driver/sql/alembic/versions/c7467b642498_buildset_updated.py @@ -34,13 +34,16 @@ def upgrade(table_prefix=''): connection = op.get_bind() connection.execute( - """ - UPDATE {buildset_table} - SET updated=greatest( - coalesce(first_build_start_time, '1970-01-01 00:00:00'), - coalesce(last_build_end_time, '1970-01-01 00:00:00'), - coalesce(event_timestamp, '1970-01-01 00:00:00')) - """.format(buildset_table=table_prefix + "zuul_buildset")) + sa.text( + """ + UPDATE {buildset_table} + SET updated=greatest( + coalesce(first_build_start_time, '1970-01-01 00:00:00'), + coalesce(last_build_end_time, '1970-01-01 00:00:00'), + coalesce(event_timestamp, '1970-01-01 00:00:00')) + """.format(buildset_table=table_prefix + "zuul_buildset") + ) + ) def downgrade(): diff --git a/zuul/driver/sql/sqlconnection.py b/zuul/driver/sql/sqlconnection.py index b89653bba..2d5c39ec3 100644 --- a/zuul/driver/sql/sqlconnection.py +++ b/zuul/driver/sql/sqlconnection.py @@ -308,27 +308,31 @@ class SQLConnection(BaseConnection): def _migrate(self, revision='head'): """Perform the alembic migrations for this connection""" + # Note that this method needs to be called with an external lock held. + # The reason for this is we retrieve the alembic version and run the + # alembic migrations in different database transactions which opens + # us to races without an external lock. with self.engine.begin() as conn: context = alembic.migration.MigrationContext.configure(conn) current_rev = context.get_current_revision() - self.log.debug('Current migration revision: %s' % current_rev) - - config = alembic.config.Config() - config.set_main_option("script_location", - "zuul:driver/sql/alembic") - config.set_main_option("sqlalchemy.url", - self.connection_config.get('dburi'). - replace('%', '%%')) - - # Alembic lets us add arbitrary data in the tag argument. We can - # leverage that to tell the upgrade scripts about the table prefix. - tag = {'table_prefix': self.table_prefix} - - if current_rev is None and not self.force_migrations: - self.metadata.create_all(self.engine) - alembic.command.stamp(config, revision, tag=tag) - else: - alembic.command.upgrade(config, revision, tag=tag) + self.log.debug('Current migration revision: %s' % current_rev) + + config = alembic.config.Config() + config.set_main_option("script_location", + "zuul:driver/sql/alembic") + config.set_main_option("sqlalchemy.url", + self.connection_config.get('dburi'). + replace('%', '%%')) + + # Alembic lets us add arbitrary data in the tag argument. We can + # leverage that to tell the upgrade scripts about the table prefix. + tag = {'table_prefix': self.table_prefix} + + if current_rev is None and not self.force_migrations: + self.metadata.create_all(self.engine) + alembic.command.stamp(config, revision, tag=tag) + else: + alembic.command.upgrade(config, revision, tag=tag) def onLoad(self, zk_client, component_registry=None): safe_connection = quote_plus(self.connection_name) diff --git a/zuul/lib/repl.py b/zuul/lib/repl.py index ecefae9ea..63a800406 100644 --- a/zuul/lib/repl.py +++ b/zuul/lib/repl.py @@ -26,14 +26,14 @@ class ThreadLocalProxy(object): self.default = default def __getattr__(self, name): - obj = self.files.get(threading.currentThread(), self.default) + obj = self.files.get(threading.current_thread(), self.default) return getattr(obj, name) def register(self, obj): - self.files[threading.currentThread()] = obj + self.files[threading.current_thread()] = obj def unregister(self): - self.files.pop(threading.currentThread()) + self.files.pop(threading.current_thread()) class REPLHandler(socketserver.StreamRequestHandler): diff --git a/zuul/model.py b/zuul/model.py index bfb65fbb7..1d82b5f2c 100644 --- a/zuul/model.py +++ b/zuul/model.py @@ -4467,19 +4467,15 @@ class BuildSet(zkobject.ZKObject): if (COMPONENT_REGISTRY.model_api < 12): return True current = build.getZKVersion() - if current is None: - current = -1 expected = self.build_versions.get(build.uuid, 0) - return expected > current + return expected != current def shouldRefreshJob(self, job): if (COMPONENT_REGISTRY.model_api < 12): return True current = job.getZKVersion() - if current is None: - current = -1 expected = self.job_versions.get(job.name, 0) - return expected > current + return expected != current @property def ref(self): @@ -6028,8 +6024,7 @@ class Bundle: def deserialize(cls, context, queue, items_by_path, data): bundle = cls(data["uuid"]) bundle.items = [ - items_by_path.get(p) or QueueItem.fromZK( - context, p, pipeline=queue.pipeline, queue=queue) + items_by_path.get(p) or QueueItem.fromZK(context, p, queue=queue) for p in data["items"] ] bundle.started_reporting = data["started_reporting"] diff --git a/zuul/scheduler.py b/zuul/scheduler.py index a546339c3..b8314f162 100644 --- a/zuul/scheduler.py +++ b/zuul/scheduler.py @@ -1548,7 +1548,6 @@ class Scheduler(threading.Thread): # This is called in the scheduler loop after another thread submits # a request if self.unparsed_abide.ltime < self.system_config_cache.ltime: - self.log.debug("Updating system config") self.updateSystemConfig() with self.layout_lock: @@ -2126,70 +2125,74 @@ class Scheduler(threading.Thread): return self.log.debug("Run handler awake") self.run_handler_lock.acquire() - try: - if not self._stopped: - self.process_reconfigure_queue() + with self.statsd_timer("zuul.scheduler.run_handler"): + try: + self._run() + except Exception: + self.log.exception("Exception in run handler:") + # There may still be more events to process + self.wake_event.set() + finally: + self.run_handler_lock.release() - if self.unparsed_abide.ltime < self.system_config_cache.ltime: - self.updateSystemConfig() + def _run(self): + if not self._stopped: + self.process_reconfigure_queue() - for tenant_name in self.unparsed_abide.tenants: - if self._stopped: - break + if self.unparsed_abide.ltime < self.system_config_cache.ltime: + self.updateSystemConfig() - tenant = self.abide.tenants.get(tenant_name) - if not tenant: - continue + for tenant_name in self.unparsed_abide.tenants: + if self._stopped: + break - # This will also forward events for the pipelines - # (e.g. enqueue or dequeue events) to the matching - # pipeline event queues that are processed afterwards. - self.process_tenant_management_queue(tenant) + tenant = self.abide.tenants.get(tenant_name) + if not tenant: + continue - if self._stopped: - break + # This will also forward events for the pipelines + # (e.g. enqueue or dequeue events) to the matching + # pipeline event queues that are processed afterwards. + self.process_tenant_management_queue(tenant) - try: - with tenant_read_lock( - self.zk_client, tenant_name, blocking=False - ) as tlock: - if not self.isTenantLayoutUpToDate(tenant_name): - continue + if self._stopped: + break - # Get tenant again, as it might have been updated - # by a tenant reconfig or layout change. - tenant = self.abide.tenants[tenant_name] + try: + with tenant_read_lock( + self.zk_client, tenant_name, blocking=False + ) as tlock: + if not self.isTenantLayoutUpToDate(tenant_name): + continue - if not self._stopped: - # This will forward trigger events to pipeline - # event queues that are processed below. - self.process_tenant_trigger_queue(tenant) + # Get tenant again, as it might have been updated + # by a tenant reconfig or layout change. + tenant = self.abide.tenants[tenant_name] - self.process_pipelines(tenant, tlock) - except LockException: - self.log.debug("Skipping locked tenant %s", - tenant.name) - remote_state = self.tenant_layout_state.get( - tenant_name) - local_state = self.local_layout_state.get( - tenant_name) - if (remote_state is None or - local_state is None or - remote_state > local_state): - # Let's keep looping until we've updated to the - # latest tenant layout. - self.wake_event.set() - except Exception: - self.log.exception("Exception processing tenant %s:", - tenant_name) - # There may still be more events to process - self.wake_event.set() + if not self._stopped: + # This will forward trigger events to pipeline + # event queues that are processed below. + self.process_tenant_trigger_queue(tenant) + + self.process_pipelines(tenant, tlock) + except LockException: + self.log.debug("Skipping locked tenant %s", + tenant.name) + remote_state = self.tenant_layout_state.get( + tenant_name) + local_state = self.local_layout_state.get( + tenant_name) + if (remote_state is None or + local_state is None or + remote_state > local_state): + # Let's keep looping until we've updated to the + # latest tenant layout. + self.wake_event.set() except Exception: - self.log.exception("Exception in run handler:") + self.log.exception("Exception processing tenant %s:", + tenant_name) # There may still be more events to process self.wake_event.set() - finally: - self.run_handler_lock.release() def primeSystemConfig(self): with self.layout_lock: @@ -2207,6 +2210,7 @@ class Scheduler(threading.Thread): def updateSystemConfig(self): with self.layout_lock: + self.log.debug("Updating system config") self.unparsed_abide, self.globals = self.system_config_cache.get() self.ansible_manager = AnsibleManager( default_version=self.globals.default_ansible_version) @@ -2241,7 +2245,12 @@ class Scheduler(threading.Thread): self.zk_client, tenant.name, pipeline.name, blocking=False) as lock,\ self.createZKContext(lock, self.log) as ctx: + self.log.debug("Processing pipeline %s in tenant %s", + pipeline.name, tenant.name) with pipeline.manager.currentContext(ctx): + if ((tenant.name, pipeline.name) in + self._profile_pipelines): + ctx.profile = True with self.statsd_timer(f'{stats_key}.handling'): refreshed = self._process_pipeline( tenant, pipeline) @@ -2310,14 +2319,10 @@ class Scheduler(threading.Thread): stats_key = f'zuul.tenant.{tenant.name}.pipeline.{pipeline.name}' ctx = pipeline.manager.current_context - if (tenant.name, pipeline.name) in self._profile_pipelines: - ctx.profile = True with self.statsd_timer(f'{stats_key}.refresh'): pipeline.change_list.refresh(ctx) pipeline.summary.refresh(ctx) pipeline.state.refresh(ctx) - if (tenant.name, pipeline.name) in self._profile_pipelines: - ctx.profile = False pipeline.state.setDirty(self.zk_client.client) if pipeline.state.old_queues: @@ -2382,6 +2387,8 @@ class Scheduler(threading.Thread): with trigger_queue_lock( self.zk_client, tenant.name, blocking=False ): + self.log.debug("Processing tenant trigger events in %s", + tenant.name) # Update the pipeline changes ctx = self.createZKContext(None, self.log) for pipeline in tenant.layout.pipelines.values(): @@ -2590,6 +2597,8 @@ class Scheduler(threading.Thread): "Skipping management event queue for tenant %s", tenant.name) return + self.log.debug("Processing tenant management events in %s", + tenant.name) self._process_tenant_management_queue(tenant) except LockException: self.log.debug("Skipping locked management event queue" |