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
|
# Copyright 2016 Red Hat, Inc.
#
# 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 logging
import time
from uuid import uuid4
from opentelemetry import trace
from zuul.driver import Driver, TriggerInterface
from zuul.driver.zuul.zuulmodel import ZuulTriggerEvent
from zuul.driver.zuul import zuulmodel
from zuul.driver.zuul import zuultrigger
from zuul.lib.logutil import get_annotated_logger
PARENT_CHANGE_ENQUEUED = 'parent-change-enqueued'
PROJECT_CHANGE_MERGED = 'project-change-merged'
class ZuulDriver(Driver, TriggerInterface):
name = 'zuul'
log = logging.getLogger("zuul.ZuulTrigger")
tracer = trace.get_tracer("zuul")
def __init__(self):
self.parent_change_enqueued_events = {}
self.project_change_merged_events = {}
def registerScheduler(self, scheduler):
self.sched = scheduler
def reconfigure(self, tenant):
for pipeline in tenant.layout.pipelines.values():
for ef in pipeline.manager.event_filters:
if not isinstance(ef.trigger, zuultrigger.ZuulTrigger):
continue
if PARENT_CHANGE_ENQUEUED in ef._types:
# parent-change-enqueued events need to be filtered by
# pipeline
for pipeline in ef._pipelines:
key = (tenant.name, pipeline)
self.parent_change_enqueued_events[key] = True
elif PROJECT_CHANGE_MERGED in ef._types:
self.project_change_merged_events[tenant.name] = True
def onChangeMerged(self, tenant, change, source):
# Called each time zuul merges a change
if self.project_change_merged_events.get(tenant.name):
span = trace.get_current_span()
link_attributes = {"rel": "ChangeMerged"}
link = trace.Link(span.get_span_context(),
attributes=link_attributes)
attributes = {"event_type": PROJECT_CHANGE_MERGED}
with self.tracer.start_as_current_span(
"ZuulEvent", links=[link], attributes=attributes):
try:
self._createProjectChangeMergedEvents(change, source)
except Exception:
self.log.exception(
"Unable to create project-change-merged events for "
"%s" % (change,))
def onChangeEnqueued(self, tenant, change, pipeline, event):
log = get_annotated_logger(self.log, event)
# Called each time a change is enqueued in a pipeline
tenant_events = self.parent_change_enqueued_events.get(
(tenant.name, pipeline.name))
log.debug("onChangeEnqueued %s", tenant_events)
if tenant_events:
span = trace.get_current_span()
link_attributes = {"rel": "ChangeEnqueued"}
link = trace.Link(span.get_span_context(),
attributes=link_attributes)
attributes = {"event_type": PARENT_CHANGE_ENQUEUED}
with self.tracer.start_as_current_span(
"ZuulEvent", links=[link], attributes=attributes):
try:
self._createParentChangeEnqueuedEvents(
change, pipeline, tenant, event)
except Exception:
log.exception(
"Unable to create parent-change-enqueued events for "
"%s in %s" % (change, pipeline))
def _createProjectChangeMergedEvents(self, change, source):
changes = source.getProjectOpenChanges(
change.project)
for open_change in changes:
self._createProjectChangeMergedEvent(open_change)
def _createProjectChangeMergedEvent(self, change):
event = ZuulTriggerEvent()
event.type = PROJECT_CHANGE_MERGED
event.trigger_name = self.name
event.connection_name = "zuul"
event.project_hostname = change.project.canonical_hostname
event.project_name = change.project.name
event.change_number = change.number
event.branch = change.branch
event.change_url = change.url
event.patch_number = change.patchset
event.ref = change.ref
event.zuul_event_id = str(uuid4().hex)
event.timestamp = time.time()
self.sched.addTriggerEvent(self.name, event)
def _createParentChangeEnqueuedEvents(self, change, pipeline, tenant,
event):
log = get_annotated_logger(self.log, event)
log.debug("Checking for changes needing %s:" % change)
if not hasattr(change, 'needed_by_changes'):
log.debug(" %s does not support dependencies" % type(change))
return
# This is very inefficient, especially on systems with large
# numbers of github installations. This can be improved later
# with persistent storage of dependency information.
needed_by_changes = set(
pipeline.manager.resolveChangeReferences(change.needed_by_changes))
for source in self.sched.connections.getSources():
log.debug(" Checking source: %s",
source.connection.connection_name)
new_changes = source.getChangesDependingOn(change, None, tenant)
log.debug(" Source %s found %s changes",
source.connection.connection_name,
len(new_changes))
needed_by_changes.update(new_changes)
log.debug(" Following changes: %s", needed_by_changes)
for needs in needed_by_changes:
self._createParentChangeEnqueuedEvent(needs, pipeline)
def _createParentChangeEnqueuedEvent(self, change, pipeline):
event = ZuulTriggerEvent()
event.type = PARENT_CHANGE_ENQUEUED
event.connection_name = "zuul"
event.trigger_name = self.name
event.pipeline_name = pipeline.name
event.project_hostname = change.project.canonical_hostname
event.project_name = change.project.name
event.change_number = change.number
event.branch = change.branch
event.change_url = change.url
event.patch_number = change.patchset
event.ref = change.ref
event.zuul_event_id = str(uuid4().hex)
event.timestamp = time.time()
self.sched.addTriggerEvent(self.name, event)
def getTrigger(self, connection_name, config=None):
return zuultrigger.ZuulTrigger(self, config)
def getTriggerSchema(self):
return zuultrigger.getSchema()
def getTriggerEventClass(self):
return zuulmodel.ZuulTriggerEvent
|
