1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
|
# Copyright 2020 BMW Group
# Copyright 2022 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 json
from collections.abc import MutableMapping
from contextlib import suppress
from functools import total_ordering
import logging
import time
from kazoo.exceptions import NoNodeError
from zuul.zk.components import COMPONENT_REGISTRY
from zuul.zk import sharding, ZooKeeperBase
@total_ordering
class LayoutState:
"""Representation of a tenant's layout state.
The layout state holds information about a certain version of a
tenant's layout. It is used to coordinate reconfigurations across
multiple schedulers by comparing a local tenant layout state
against the current version in Zookeeper. In case it detects that
a local layout state is outdated, this scheduler is not allowed to
process this tenant (events, pipelines, ...) until the layout is
updated.
The important information of the layout state is the logical
timestamp (ltime) that is used to detect if the layout on a
scheduler needs to be updated. The ltime is the last modified
transaction ID (mzxid) of the corresponding Znode in Zookeeper.
The hostname of the scheduler creating the new layout state and the
timestamp of the last reconfiguration are only informational and
may aid in debugging.
"""
def __init__(self, tenant_name, hostname, last_reconfigured, uuid,
branch_cache_min_ltimes, last_reconfigure_event_ltime,
ltime=-1):
self.uuid = uuid
self.ltime = ltime
self.tenant_name = tenant_name
self.hostname = hostname
self.last_reconfigured = last_reconfigured
self.last_reconfigure_event_ltime =\
last_reconfigure_event_ltime
self.branch_cache_min_ltimes = branch_cache_min_ltimes
def toDict(self):
return {
"tenant_name": self.tenant_name,
"hostname": self.hostname,
"last_reconfigured": self.last_reconfigured,
"last_reconfigure_event_ltime":
self.last_reconfigure_event_ltime,
"uuid": self.uuid,
"branch_cache_min_ltimes": self.branch_cache_min_ltimes,
}
@classmethod
def fromDict(cls, data):
return cls(
data["tenant_name"],
data["hostname"],
data["last_reconfigured"],
data.get("uuid"),
data.get("branch_cache_min_ltimes"),
data.get("last_reconfigure_event_ltime", -1),
data.get("ltime", -1),
)
def __eq__(self, other):
if not isinstance(other, LayoutState):
return False
return self.uuid == other.uuid
def __gt__(self, other):
if not isinstance(other, LayoutState):
return False
return self.ltime > other.ltime
def __repr__(self):
return (
f"<{self.__class__.__name__} {self.tenant_name}: "
f"ltime={self.ltime}, "
f"hostname={self.hostname}, "
f"last_reconfigured={self.last_reconfigured}>"
)
class LayoutStateStore(ZooKeeperBase, MutableMapping):
log = logging.getLogger("zuul.LayoutStore")
layout_root = "/zuul/layout"
layout_data_root = "/zuul/layout-data"
def __init__(self, client, callback):
super().__init__(client)
self._watched_tenants = set()
self._callback = callback
self.kazoo_client.ensure_path(self.layout_root)
self.kazoo_client.ChildrenWatch(self.layout_root, self._layoutCallback)
def _layoutCallback(self, tenant_list, event=None):
new_tenants = set(tenant_list) - self._watched_tenants
for tenant_name in new_tenants:
self.kazoo_client.DataWatch(f"{self.layout_root}/{tenant_name}",
self._callbackWrapper)
def _callbackWrapper(self, data, stat, event):
self._callback()
def __getitem__(self, tenant_name):
try:
data, zstat = self.kazoo_client.get(
f"{self.layout_root}/{tenant_name}")
except NoNodeError:
raise KeyError(tenant_name)
if not data:
raise KeyError(tenant_name)
return LayoutState.fromDict({
"ltime": zstat.last_modified_transaction_id,
**json.loads(data)
})
def __setitem__(self, tenant_name, state):
path = f"{self.layout_root}/{tenant_name}"
data = json.dumps(state.toDict(), sort_keys=True).encode("utf-8")
if self.kazoo_client.exists(path):
zstat = self.kazoo_client.set(path, data)
else:
_, zstat = self.kazoo_client.create(path, data, include_data=True)
# Set correct ltime of the layout in Zookeeper
state.ltime = zstat.last_modified_transaction_id
def __delitem__(self, tenant_name):
try:
self.kazoo_client.delete(f"{self.layout_root}/{tenant_name}")
except NoNodeError:
raise KeyError(tenant_name)
def __iter__(self):
try:
tenant_names = self.kazoo_client.get_children(self.layout_root)
except NoNodeError:
return
yield from tenant_names
def __len__(self):
zstat = self.kazoo_client.exists(self.layout_root)
if zstat is None:
return 0
return zstat.children_count
def getMinLtimes(self, layout_state):
if COMPONENT_REGISTRY.model_api < 6:
return None
try:
path = f"{self.layout_data_root}/{layout_state.uuid}"
with sharding.BufferedShardReader(
self.kazoo_client, path) as stream:
data = stream.read()
except NoNodeError:
return None
try:
return json.loads(data)['min_ltimes']
except Exception:
return None
def setMinLtimes(self, layout_state, min_ltimes):
data = dict(min_ltimes=min_ltimes)
encoded_data = json.dumps(data).encode("utf-8")
path = f"{self.layout_data_root}/{layout_state.uuid}"
with sharding.BufferedShardWriter(
self.kazoo_client, path) as stream:
stream.write(encoded_data)
def cleanup(self, delay=300):
self.log.debug("Starting layout data cleanup")
known_layouts = set()
for tenant in self.kazoo_client.get_children(
self.layout_root):
state = self.get(tenant)
if state:
known_layouts.add(state.uuid)
for layout_uuid in self.kazoo_client.get_children(
self.layout_data_root):
if layout_uuid in known_layouts:
continue
path = f"{self.layout_data_root}/{layout_uuid}"
zstat = self.kazoo_client.exists(path)
if zstat is None:
continue
now = time.time()
if now - zstat.created >= delay:
self.log.debug("Deleting unused layout data for %s",
layout_uuid)
with suppress(NoNodeError):
self.kazoo_client.delete(path, recursive=True)
self.log.debug("Finished layout data cleanup")
|
