zuul/zuul
Code Issues Proposed changes

Add option to pause tenant event processing

Browse Source 
This adds the ability to pause event processing for a tenant.
This can be useful in cases where an external system is not
behaving properly.

Change-Id: Ia6714d8ea472feef8a769d333deaf873b7cf8501
This commit is contained in:
James E. Blair
2025-05-22 09:34:52 -07:00
parent 2b6de82da3
commit f8dd51f93c
11 changed files with 538 additions and 27 deletions
Download Patch File Download Diff File Expand all files Collapse all files
doc/source/developer/zookeeper.rst
+5
View File
@@ -255,6 +255,11 @@ This is a reference for object layout in Zookeeper.
The pipeline trigger event queue.
.. path:: zuul/events/tenant/<tenant>/state
A Znode that, if exists, can be used to pause event
processing.
.. path:: zuul/executor/unzoned
:type: JobRequestQueue
doc/source/operation.rst
+28
View File
@@ -91,6 +91,34 @@ intended to process events should not set this option on any
schedulers. To use this option on a standby or testing cluster, set
it on all schedulers.
Managing Event Processing
~~~~~~~~~~~~~~~~~~~~~~~~~
If an issue with external systems is affecting Zuul, tenant
administrators may suspend event processing until the issue is resolved.
Three options are available:
* Pausing trigger event queue processing to prevent Zuul from adding
new items into pipelines until the queue is unpaused
* Pausing trigger and result queue processing to prevent Zuul from
adding new items into pipelines or reporting items that are already
in existing pipelines until it is unpaused. This may include
merging changes in the case of a :term:`gate` pipeline.
* Discarding trigger events to cause Zuul to disregard trigger events
(with no processing of the backlog) until further notice.
If any of the above settings are enabled for a tenant, a banner will
be displayed on the status page indicating that queue processing is
paused, and including a reason (if any was supplied).
There are two ways to manage queue processing for a tenant. The first
is using the web interface: authenticate as a tenant administrator,
and a "Manage Queues" button will appear at the top of the status
page; click that and fill out the form. The second is using the
``zuul-client tenant-state`` command.
.. _backup:
Backup and Restoration
releasenotes/notes/tenant-state-bca37a9624b70712.yaml
+5
View File
@@ -0,0 +1,5 @@
---
features:
- |
Tenant administrators may now use the web UI or zuul-client
commands to pause and unpause event processing for a tenant.
tests/unit/test_web.py
+107
View File
@@ -3649,6 +3649,113 @@ class TestTenantScopedWebApi(BaseTestWeb):
self.assertTrue(data['zuul']['admin'] is False, data)
self.assertTrue(data['zuul']['scope'] == ['tenant-one'], data)
def test_state_post(self):
self.executor_server.hold_jobs_in_build = False
A = self.fake_gerrit.addFakeChange('org/project', 'master', 'A')
self.fake_gerrit.addEvent(A.getPatchsetCreatedEvent(1))
self.waitUntilSettled()
self.assertHistory([
dict(name='project-merge', result='SUCCESS', changes='1,1'),
dict(name='project-test1', result='SUCCESS', changes='1,1'),
dict(name='project-test2', result='SUCCESS', changes='1,1'),
], ordered=False)
authz = {'iss': 'zuul_operator',
'aud': 'zuul.example.com',
'sub': 'testuser',
'zuul': {
'admin': ['tenant-one', ],
},
'exp': int(time.time()) + 3600,
'iat': int(time.time())}
token = jwt.encode(authz, key='NoDanaOnlyZuul',
algorithm='HS256')
args = {
'trigger_queue_paused': True,
'reason': 'test trigger paused',
}
req = self.post_url(
'api/tenant/tenant-one/state',
headers={'Authorization': 'Bearer %s' % token},
json=args)
self.assertEqual(200, req.status_code, req.text)
time.sleep(1)
B = self.fake_gerrit.addFakeChange('org/project', 'master', 'B')
self.fake_gerrit.addEvent(B.getPatchsetCreatedEvent(1))
time.sleep(5)
self.assertHistory([
dict(name='project-merge', result='SUCCESS', changes='1,1'),
dict(name='project-test1', result='SUCCESS', changes='1,1'),
dict(name='project-test2', result='SUCCESS', changes='1,1'),
], ordered=False)
self.assertEqual(0, len(self.builds))
args = {
'trigger_queue_paused': True,
'result_queue_paused': True,
'reason': "test result paused",
}
req = self.post_url(
'api/tenant/tenant-one/state',
headers={'Authorization': 'Bearer %s' % token},
json=args)
self.assertEqual(200, req.status_code, req.text)
time.sleep(5)
self.assertHistory([
dict(name='project-merge', result='SUCCESS', changes='1,1'),
dict(name='project-test1', result='SUCCESS', changes='1,1'),
dict(name='project-test2', result='SUCCESS', changes='1,1'),
], ordered=False)
self.assertEqual(0, len(self.builds))
args = {
'trigger_queue_paused': False,
'result_queue_paused': False,
'reason': None,
}
req = self.post_url(
'api/tenant/tenant-one/state',
headers={'Authorization': 'Bearer %s' % token},
json=args)
self.assertEqual(200, req.status_code, req.text)
self.waitUntilSettled()
self.assertHistory([
dict(name='project-merge', result='SUCCESS', changes='1,1'),
dict(name='project-test1', result='SUCCESS', changes='1,1'),
dict(name='project-test2', result='SUCCESS', changes='1,1'),
dict(name='project-merge', result='SUCCESS', changes='2,1'),
dict(name='project-test1', result='SUCCESS', changes='2,1'),
dict(name='project-test2', result='SUCCESS', changes='2,1'),
], ordered=False)
args = {
'trigger_queue_discarding': True,
'reason': 'test discarding',
}
req = self.post_url(
'api/tenant/tenant-one/state',
headers={'Authorization': 'Bearer %s' % token},
json=args)
self.assertEqual(200, req.status_code, req.text)
time.sleep(1)
C = self.fake_gerrit.addFakeChange('org/project', 'master', 'C')
self.fake_gerrit.addEvent(C.getPatchsetCreatedEvent(1))
self.waitUntilSettled()
self.assertHistory([
dict(name='project-merge', result='SUCCESS', changes='1,1'),
dict(name='project-test1', result='SUCCESS', changes='1,1'),
dict(name='project-test2', result='SUCCESS', changes='1,1'),
dict(name='project-merge', result='SUCCESS', changes='2,1'),
dict(name='project-test1', result='SUCCESS', changes='2,1'),
dict(name='project-test2', result='SUCCESS', changes='2,1'),
], ordered=False)
class TestTenantScopedWebApiWithAccessRules(TestTenantScopedWebApi):
config_file = 'zuul-admin-web.conf'
web/src/api.js
+14
View File
@@ -184,6 +184,19 @@ function fetchChangeStatus(apiPrefix, changeId) {
return makeRequest(apiPrefix + 'status/change/' + changeId)
}
function setTenantState(apiPrefix, discardTriggerEvents, pauseTriggerQueue, pauseResultQueue, reason) {
return makeRequest(
apiPrefix + 'state',
'post',
{
trigger_queue_discarding: discardTriggerEvents,
trigger_queue_paused: pauseTriggerQueue,
result_queue_paused: pauseResultQueue,
reason: reason,
}
)
}
function fetchFreezeJob(apiPrefix, pipelineName, projectName, branchName, jobName) {
return makeRequest(apiPrefix +
'pipeline/' + pipelineName +
@@ -427,4 +440,5 @@ export {
getLogFile,
getStreamUrl,
promote,
setTenantState,
}
web/src/containers/status/PauseModal.jsx
+138
View File
@@ -0,0 +1,138 @@
// Copyright 2025 Acme Gating, LLC
//
// Licensed under the Apache License, Version 2.0 (the "License"); you may
// not use this file except in compliance with the License. You may obtain
// a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
// WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
// License for the specific language governing permissions and limitations
// under the License.
import React, { useState } from 'react'
import PropTypes from 'prop-types'
import { useDispatch, useSelector } from 'react-redux'
import {
Button,
FormGroup,
Modal,
ModalVariant,
Radio,
TextInput,
} from '@patternfly/react-core'
import { addApiError } from '../../actions/notifications'
import { setTenantState } from '../../api'
function PauseModal({isOpen, setOpen}) {
const tenant = useSelector((state) => state.tenant)
const [reason, setReason] = useState('')
const [queueState, setQueueState] = useState('normal')
const [discardTrigger, setDiscardTrigger] = useState(false)
const [pauseTrigger, setPauseTrigger] = useState(false)
const [pauseResult, setPauseResult] = useState(false)
const dispatch = useDispatch()
return (
<Modal
variant={ModalVariant.small}
isOpen={isOpen}
title="Manage Tenant Event Processing"
onClose={() => { setOpen(false) }}
actions={[
<Button key="confirm" variant="primary"
onClick={() => {
setOpen(false)
setTenantState(
tenant.apiPrefix,
discardTrigger,
pauseTrigger,
pauseResult,
reason)
.catch(error => {
dispatch(addApiError(error))
})
}}>
Confirm
</Button>,
<Button key="cancel" variant="link"
onClick={() => { setOpen(false) }}>
Cancel</Button>,
]}>
<p>You can pause trigger or result event processing for this tenant, or discard trigger events. Trigger events cause new items to appear in pipelines. Result events cause item results to be reported (and potentially, changes merged).</p>
<FormGroup
label="Pause"
fieldId="pause-form-unpaused">
<Radio
id="pause-form-unpaused"
label="Unpaused"
isChecked={queueState === 'normal'}
onChange={() => {
setQueueState('normal')
setDiscardTrigger(false)
setPauseTrigger(false)
setPauseResult(false)
}}
/>
<Radio
id="pause-form-trigger-queue-paused"
label="Pause trigger queue"
isChecked={queueState === 'pause-trigger'}
onChange={() => {
setQueueState('pause-trigger')
setDiscardTrigger(false)
setPauseTrigger(true)
setPauseResult(false)
}}
/>
<Radio
id="pause-form-result-queue-paused"
label="Pause trigger and result queues"
isChecked={queueState === 'pause-result'}
onChange={() => {
setQueueState('pause-result')
setDiscardTrigger(false)
setPauseTrigger(true)
setPauseResult(true)
}}
/>
<Radio
id="pause-form-trigger-discard"
label="Discard trigger events"
isChecked={queueState === 'discard-trigger'}
onChange={() => {
setQueueState('discard-trigger')
setDiscardTrigger(true)
setPauseTrigger(false)
setPauseResult(false)
}}
/>
</FormGroup>
<FormGroup
label="Reason"
fieldId="pause-form-reason"
helperText="This explanation will appear on the status page.">
<TextInput
value={reason}
isRequired
type="text"
id="pause-form-reason"
name="pauseReason"
onChange={(value) => { setReason(value) }}
/>
</FormGroup>
</Modal>
)
}
PauseModal.propTypes = {
isOpen: PropTypes.bool,
setOpen: PropTypes.object,
}
export default (PauseModal)
web/src/pages/PipelineOverview.jsx
+71 -1
View File
@@ -20,6 +20,8 @@ import PropTypes from 'prop-types'
import * as moment_tz from 'moment-timezone'
import {
Banner,
Button,
Gallery,
GalleryItem,
Level,
@@ -31,10 +33,19 @@ import {
ToolbarContent,
ToolbarItem,
Tooltip,
Flex,
FlexItem,
} from '@patternfly/react-core'
import { StreamIcon } from '@patternfly/react-icons'
import {
StreamIcon,
ExclamationTriangleIcon,
} from '@patternfly/react-icons'
import PipelineSummary from '../containers/status/PipelineSummary'
import PauseModal from '../containers/status/PauseModal'
import { fetchStatusIfNeeded } from '../actions/status'
import { clearQueue } from '../actions/statusExpansion'
@@ -114,6 +125,47 @@ PipelineGallery.propTypes = {
sortKey: PropTypes.string,
}
function renderBanner(status)
{
if (!status || !status.state) {
return <></>
}
// This method is able to describe states that the client libraries
// do not support creating (for example, buth discarding and pausing
// the trigger event queue).
const msgs = []
const queues = []
if (status.state.trigger_queue_discarding)
{
msgs.push('Discarding trigger events')
}
if (status.state.trigger_queue_paused) {
queues.push('Trigger')
}
if (status.state.result_queue_paused) {
queues.push('Result')
}
if (queues.length) {
msgs.push(`${queues.join(', ')} event queue${queues.length>1?'s':''} paused`)
}
if (!msgs.length) {
return <></>
}
const msg = `${msgs.join(', ')}: ${status.state.reason || 'no reason supplied'}`
return (
<Banner screenReaderText="Warning banner" variant="warning">
<Flex spaceItems={{ default: 'spaceItemsSm' }}>
<FlexItem>
<ExclamationTriangleIcon />
</FlexItem>
<FlexItem>{msg}</FlexItem>
</Flex>
</Banner>
)
}
function getPipelines(status, location, filterCategories) {
let pipelines = []
let stats = {}
@@ -142,6 +194,7 @@ function getPipelines(status, location, filterCategories) {
function PipelineOverviewPage() {
const status = useSelector((state) => state.status.status)
const user = useSelector((state) => state.user)
const filterCategories = [
{
@@ -183,6 +236,7 @@ function PipelineOverviewPage() {
const filters = getFiltersFromUrl(location, filterCategories)
const filterActive = isFilterActive(filters)
const [showPauseModal, setShowPauseModal] = useState(false)
const [showAllPipelines, setShowAllPipelines] = useState(
filterActive || localStorage.getItem('zuul_show_all_pipelines') === 'true')
const [expandAll, setExpandAll] = useState(
@@ -284,6 +338,9 @@ function PipelineOverviewPage() {
return (
<>
{renderBanner(status)}
<PageSection
variant={darkMode ? PageSectionVariants.dark : PageSectionVariants.light}
className="zuul-toolbar-section"
@@ -319,6 +376,15 @@ function PipelineOverviewPage() {
<LevelItem>
<Toolbar>
<ToolbarContent style={{paddingRight: '0'}}>
{(user.isAdmin && user.scope.indexOf(tenant.name) !== -1) && (
<ToolbarItem>
<Button onClick={() => {setShowPauseModal(true)}}>
Manage Events
</Button>
</ToolbarItem>
)}
<ToolbarStatsGroup>
<ToolbarStatsItem
name="events"
@@ -367,6 +433,10 @@ function PipelineOverviewPage() {
sortKey={currentSortKey.key}
/>
</PageSection>
<PauseModal
isOpen={showPauseModal}
setOpen={setShowPauseModal}
/>
</>
)
}
zuul/model.py
+49 -1
View File
@@ -1,5 +1,5 @@
# Copyright 2012 Hewlett-Packard Development Company, L.P.
# Copyright 2021-2024 Acme Gating, LLC
# Copyright 2021-2025 Acme Gating, LLC
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
@@ -10975,3 +10975,51 @@ class AuthZRuleTree(object):
def __repr__(self):
return '<AuthZRuleTree [ %s ]>' % self.ruletree
class TenantEventState(zkobject.ZKObject):
def __init__(self, path):
super().__init__()
self._set(_path=path)
self.reset()
def reset(self):
self._set(
trigger_queue_discarding=False,
trigger_queue_paused=False,
result_queue_paused=False,
reason=None,
)
def getPath(self):
return self._path
def toDict(self):
data = {
"trigger_queue_discarding": self.trigger_queue_discarding,
"trigger_queue_paused": self.trigger_queue_paused,
"result_queue_paused": self.result_queue_paused,
"reason": self.reason,
}
return data
def serialize(self, context):
data = self.toDict()
r = json.dumps(data, sort_keys=True).encode("utf8")
return r
def deserialize(self, raw, context, extra=None):
if not raw:
raw = {}
r = super().deserialize(raw, context)
return r
def internalCreate(self, context):
data = self._trySerialize(context)
try:
self._save(context, data)
except NoNodeError:
try:
self._save(context, data, create=True)
except NodeExistsError:
self._save(context, data)
zuul/scheduler.py
+36 -14
View File
@@ -2,7 +2,7 @@
# Copyright 2013 OpenStack Foundation
# Copyright 2013 Antoine "hashar" Musso
# Copyright 2013 Wikimedia Foundation Inc.
# Copyright 2021-2024 Acme Gating, LLC
# Copyright 2021-2025 Acme Gating, LLC
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
@@ -2208,6 +2208,8 @@ class Scheduler(threading.Thread):
if not tenant:
continue
tenant_state = self.event_watcher.tenant_state[tenant_name]
# This will also forward events for the pipelines
# (e.g. enqueue or dequeue events) to the matching
# pipeline event queues that are processed afterwards.
@@ -2227,11 +2229,14 @@ class Scheduler(threading.Thread):
# Get tenant again, as it might have been updated
# by a tenant reconfig or layout change.
tenant = self.abide.tenants[tenant_name]
if not self._stopped:
if (not self._stopped and
not tenant_state.trigger_queue_paused):
# This will forward trigger events to pipeline
# event queues that are processed below.
self.process_tenant_trigger_queue(tenant)
elif tenant_state.trigger_queue_paused:
self.log.info("Trigger queue paused for tenant %s",
tenant.name)
self.process_pipelines(tenant, tlock)
except PendingReconfiguration:
@@ -2406,13 +2411,25 @@ class Scheduler(threading.Thread):
if manager.state.old_queues:
self._reenqueuePipeline(tenant, manager, ctx)
tenant_state = self.event_watcher.tenant_state[tenant.name]
with self.statsd_timer(f'{stats_key}.event_process'):
self.process_pipeline_management_queue(
tenant, tenant_lock, manager)
# Give result events priority -- they let us stop builds,
# whereas trigger events cause us to execute builds.
self.process_pipeline_result_queue(tenant, tenant_lock, manager)
self.process_pipeline_trigger_queue(tenant, tenant_lock, manager)
if not tenant_state.result_queue_paused:
self.process_pipeline_result_queue(
tenant, tenant_lock, manager)
else:
self.log.info("Result queue paused for tenant %s",
tenant.name)
if not tenant_state.trigger_queue_paused:
self.process_pipeline_trigger_queue(
tenant, tenant_lock, manager)
else:
self.log.info("Trigger queue paused for tenant %s",
tenant.name)
self.abortIfPendingReconfig(tenant_lock)
try:
with self.statsd_timer(f'{stats_key}.process'):
@@ -2496,18 +2513,23 @@ class Scheduler(threading.Thread):
# Get the ltime of the last reconfiguration event
self.trigger_events[tenant.name].refreshMetadata()
tenant_state = self.event_watcher.tenant_state[tenant.name]
for event in self.trigger_events[tenant.name]:
log = get_annotated_logger(self.log, event.zuul_event_id)
log.debug("Forwarding trigger event %s", event)
try:
trigger_span = tracing.restoreSpanContext(
event.span_context)
with self.tracer.start_as_current_span(
"TenantTriggerEventProcessing",
links=[
trace.Link(trigger_span.get_span_context())
]):
self._forward_trigger_event(event, tenant)
if not tenant_state.trigger_queue_discarding:
log.debug("Forwarding trigger event %s", event)
trigger_span = tracing.restoreSpanContext(
event.span_context)
with self.tracer.start_as_current_span(
"TenantTriggerEventProcessing",
links=[
trace.Link(
trigger_span.get_span_context())
]):
self._forward_trigger_event(event, tenant)
else:
log.debug("Discarding trigger event %s", event)
except Exception:
log.exception("Unable to forward event %s "
"to tenant %s", event, tenant.name)
zuul/web/__init__.py
+47 -1
View File
@@ -1,5 +1,5 @@
# Copyright (c) 2017 Red Hat
# Copyright 2021-2024 Acme Gating, LLC
# Copyright 2021-2025 Acme Gating, LLC
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
@@ -69,11 +69,13 @@ from zuul.zk import ZooKeeperClient
from zuul.zk.components import COMPONENT_REGISTRY, WebComponent
from zuul.zk.config_cache import SystemConfigCache, UnparsedConfigCache
from zuul.zk.event_queues import (
EventWatcher,
TenantManagementEventQueue,
TenantTriggerEventQueue,
PipelineManagementEventQueue,
PipelineResultEventQueue,
PipelineTriggerEventQueue,
TENANT_EVENT_STATE,
)
from zuul.zk.executor import ExecutorApi
from zuul.zk.image_registry import (
@@ -1353,6 +1355,39 @@ class ZuulWebAPI(object):
return True
@cherrypy.expose
@cherrypy.tools.json_in()
@cherrypy.tools.json_out(content_type='application/json; charset=utf-8')
@cherrypy.tools.handle_options(allowed_methods=['POST', ])
@cherrypy.tools.check_tenant_auth(require_admin=True)
def state_post(self, tenant_name, tenant, auth):
body = cherrypy.request.json
current_state = self.zuulweb.event_watcher.tenant_state[tenant.name]
with self.zuulweb.createZKContext(None, self.log) as ctx:
path = TENANT_EVENT_STATE.format(tenant=tenant_name)
tqs = model.TenantEventState(path)
self.log.info('User %s setting tenant %s state to %s',
auth.uid, tenant_name, body)
reason = body.get('reason')
if reason:
# We limit the reason to 4096 chars to limit the
# zk node size.
reason = str(reason[:4096])
tqs._set(
trigger_queue_discarding=bool(
body.get('trigger_queue_discarding',
current_state.trigger_queue_discarding)),
trigger_queue_paused=bool(
body.get('trigger_queue_paused',
current_state.trigger_queue_paused)),
result_queue_paused=bool(
body.get('result_queue_paused',
current_state.result_queue_paused)),
reason=reason,
)
tqs.internalCreate(ctx)
@cherrypy.expose
@cherrypy.tools.json_out(content_type='application/json; charset=utf-8')
@cherrypy.tools.handle_options()
@@ -1796,6 +1831,10 @@ class ZuulWebAPI(object):
data['trigger_event_queue'] = {}
data['trigger_event_queue']['length'] = len(
self.zuulweb.trigger_events[tenant.name])
data['state'] =\
self.zuulweb.event_watcher.tenant_state[tenant.name].toDict()
data['management_event_queue'] = {}
data['management_event_queue']['length'] = len(
self.zuulweb.management_events[tenant.name]
@@ -2925,6 +2964,11 @@ class ZuulWeb(object):
controller=api,
conditions=dict(method=['POST']),
action='autohold_project_post')
route_map.connect(
'api',
'/api/tenant/{tenant_name}/state',
controller=api,
action='state_post')
route_map.connect(
'api',
'/api/tenant/{tenant_name}/project/{project_name:.*}/enqueue',
@@ -3072,6 +3116,8 @@ class ZuulWeb(object):
self.zk_client, self.hostname, version=get_version_string())
self.component_info.register()
self.event_watcher = EventWatcher(self.zk_client, None)
self.monitoring_server = MonitoringServer(self.config, 'web',
self.component_info)
self.monitoring_server.start()
zuul/zk/event_queues.py
+38 -10
View File
@@ -1,4 +1,5 @@
# Copyright 2020 BMW Group
# Copyright 2021-2025 Acme Gating, LLC
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
@@ -60,6 +61,7 @@ MANAGEMENT_EVENT_TYPE_MAP = {
# /zuul/events/tenant TENANT_ROOT
# /{tenant} TENANT_NAME_ROOT
# /state TENANT_EVENT_STATE
# /management TENANT_MANAGEMENT_ROOT
# /queue TENANT_MANAGEMENT_QUEUE
# /data [side channel data]
@@ -92,6 +94,7 @@ PIPELINE_TRIGGER_ROOT = PIPELINE_NAME_ROOT + "/trigger"
PIPELINE_TRIGGER_QUEUE = PIPELINE_TRIGGER_ROOT + "/queue"
PIPELINE_RESULT_ROOT = PIPELINE_NAME_ROOT + "/result"
PIPELINE_RESULT_QUEUE = PIPELINE_RESULT_ROOT + "/queue"
TENANT_EVENT_STATE = TENANT_NAME_ROOT + "/state"
CONNECTION_ROOT = "/zuul/events/connection"
@@ -121,6 +124,9 @@ class EventWatcher(ZooKeeperSimpleBase):
self.callback = callback
self.watched_tenants = set()
self.watched_pipelines = set()
self.tenant_state = DefaultKeyDict(
lambda tenant_name: model.TenantEventState(
TENANT_EVENT_STATE.format(tenant=tenant_name)))
self.kazoo_client.ensure_path(TENANT_ROOT)
self.kazoo_client.ChildrenWatch(TENANT_ROOT, self._tenantWatch)
@@ -134,18 +140,25 @@ class EventWatcher(ZooKeeperSimpleBase):
if tenant_name in self.watched_tenants:
continue
for path in (TENANT_MANAGEMENT_QUEUE,
TENANT_TRIGGER_QUEUE):
path = path.format(tenant=tenant_name)
self.kazoo_client.ensure_path(path)
if self.callback:
# only set these watches if we're in the scheduler context
for path in (TENANT_MANAGEMENT_QUEUE,
TENANT_TRIGGER_QUEUE):
path = path.format(tenant=tenant_name)
self.kazoo_client.ensure_path(path)
self.kazoo_client.ChildrenWatch(
path, self._eventWatch, send_event=True)
pipelines_path = PIPELINE_ROOT.format(tenant=tenant_name)
self.kazoo_client.ensure_path(pipelines_path)
self.kazoo_client.ChildrenWatch(
path, self._eventWatch, send_event=True)
pipelines_path = PIPELINE_ROOT.format(tenant=tenant_name)
self.kazoo_client.ensure_path(pipelines_path)
self.kazoo_client.ChildrenWatch(
pipelines_path, self._makePipelineWatcher(tenant_name))
pipelines_path, self._makePipelineWatcher(tenant_name))
# both scheduler and zuul-web are interested in these
state_path = TENANT_EVENT_STATE.format(tenant=tenant_name)
self.kazoo_client.DataWatch(
state_path,
self._makeStateWatcher(tenant_name))
self.watched_tenants.add(tenant_name)
def _pipelineWatch(self, tenant_name, pipelines):
@@ -185,6 +198,21 @@ class EventWatcher(ZooKeeperSimpleBase):
elif event.type == EventType.CHILD:
self.callback()
def _makeStateWatcher(self, tenant_name):
def watch(data=None, event=None):
return self._stateWatch(tenant_name, data, event)
return watch
def _stateWatch(self, tenant_name,
data=None, stat=None):
if data is None:
self.tenant_state[tenant_name].reset()
else:
self.tenant_state[tenant_name]._updateFromRaw(
data, stat, None, None)
if self.callback:
self.callback()
class ZooKeeperEventQueue(ZooKeeperSimpleBase, Iterable):
"""Abstract API for events via ZooKeeper