|
# This file is part of Buildbot. Buildbot is free software: you can # redistribute it and/or modify it under the terms of the GNU General Public # License as published by the Free Software Foundation, version 2. # # This program is distributed in the hope that it will be useful, but WITHOUT # ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS # FOR A PARTICULAR PURPOSE. See the GNU General Public License for more # details. # # You should have received a copy of the GNU General Public License along with # this program; if not, write to the Free Software Foundation, Inc., 51 # Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA. # # Copyright Buildbot Team Members
"""Push events to an abstract receiver.
Implements the HTTP receiver."""
assert json
MemoryQueue, PersistentQueue
"""Event streamer to a abstract channel.
It uses IQueue to batch push requests and queue the data when the receiver is down. When a PersistentQueue object is used, the items are saved to disk on master shutdown so they can be pushed back when the master is restarted. """
bufferDelay=1, retryDelay=5, blackList=None): """ @serverPushCb: callback to be used. It receives 'self' as parameter. It should call self.queueNextServerPush() when it's done to queue the next push. It is guaranteed that the queue is not empty when this function is called. @queue: a item queue that implements IQueue. @path: path to save config. @filter: when True (default), removes all "", None, False, [] or {} entries. @bufferDelay: amount of time events are queued before sending, to reduce the number of push requests rate. This is the delay between the end of a request to initializing a new one. @retryDelay: amount of time between retries when no items were pushed on last serverPushCb call. @blackList: events that shouldn't be sent. """ StatusReceiverMultiService.__init__(self)
# Parameters. self.queue = queue if self.queue is None: self.queue = MemoryQueue() self.queue = IndexedQueue(self.queue) self.path = path self.filter = filter self.bufferDelay = bufferDelay self.retryDelay = retryDelay if not callable(serverPushCb): raise NotImplementedError('Please pass serverPushCb parameter.') def hookPushCb(): # Update the index so we know if the next push succeed or not, don't # update the value when the queue is empty. if not self.queue.nbItems(): return self.lastIndex = self.queue.getIndex() return serverPushCb(self) self.serverPushCb = hookPushCb self.blackList = blackList
# Other defaults. # IDelayedCall object that represents the next queued push. self.task = None self.stopped = False self.lastIndex = -1 self.state = {} self.state['started'] = str(datetime.datetime.utcnow()) self.state['next_id'] = 1 self.state['last_id_pushed'] = 0 # Try to load back the state. if self.path and os.path.isdir(self.path): state_path = os.path.join(self.path, 'state') if os.path.isfile(state_path): with open(state_path, 'r') as f: self.state.update(json.load(f))
if self.queue.nbItems(): # Last shutdown was not clean, don't wait to send events. self.queueNextServerPush()
"""Starting up.""" StatusReceiverMultiService.startService(self) self.status = self.parent.getStatus() self.status.subscribe(self) self.initialPush()
"""Returns if the "virtual pointer" in the queue advanced.""" return self.lastIndex <= self.queue.getIndex()
"""Queue the next push or call it immediately.
Called to signal new items are available to be sent or on shutdown. A timer should be queued to trigger a network request or the callback should be called immediately. If a status push is already queued, ignore the current call.""" # Determine the delay. if self.wasLastPushSuccessful(): if self.stopped: # Shutting down. delay = 0 else: # Normal case. delay = self.bufferDelay else: if self.stopped: # Too bad, we can't do anything now, we're shutting down and the # receiver is also down. We'll just save the objects to disk. return else: # The server is inaccessible, retry less often. delay = self.retryDelay
# Cleanup a previously queued task if necessary. if self.task: # Warning: we could be running inside the task. if self.task.active(): # There was already a task queue, don't requeue it, just let it # go. return else: if self.task.active(): # There was a task queued but it is requested to call it # *right now* so cancel it. self.task.cancel() # Otherwise, it was just a stray object. self.task = None
# Do the queue/direct call. if delay: # Call in delay seconds. self.task = reactor.callLater(delay, self.serverPushCb) elif self.stopped: if not self.queue.nbItems(): return # Call right now, we're shutting down. @defer.inlineCallbacks def BlockForEverythingBeingSent(): yield defer.maybeDeferred(self.serverPushCb()) return BlockForEverythingBeingSent() else: # delay should never be 0. That can cause Buildbot to spin tightly # trying to push events that may not be received well by a status # listener. log.err('Did not expect delay to be 0, but it is.') return
"""Shutting down.""" self.finalPush() self.stopped = True if (self.task and self.task.active()): # We don't have time to wait, force an immediate call. self.task.cancel() self.task = None d = self.queueNextServerPush() elif self.wasLastPushSuccessful(): d = self.queueNextServerPush() else: d = defer.succeed(None)
# We're dying, make sure we save the results. self.queue.save() if self.path and os.path.isdir(self.path): state_path = os.path.join(self.path, 'state') with open(state_path, 'w') as f: json.dump(self.state, f, sort_keys=True, indent=2) # Make sure all Deferreds are called on time and in a sane order. defers = filter(None, [d, StatusReceiverMultiService.stopService(self)]) return defer.DeferredList(defers)
"""Push a new event.
The new event will be either: - Queued in memory to reduce network usage - Queued to disk when the sink server is down - Pushed (along the other queued items) to the server """ if self.blackList and event in self.blackList: return # First, generate the packet. packet = {} packet['id'] = self.state['next_id'] self.state['next_id'] += 1 packet['timestamp'] = str(datetime.datetime.utcnow()) packet['project'] = self.status.getTitle() packet['started'] = self.state['started'] packet['event'] = event packet['payload'] = {} for obj_name, obj in objs.items(): if hasattr(obj, 'asDict'): obj = obj.asDict() if self.filter: obj = FilterOut(obj) packet['payload'][obj_name] = obj self.queue.pushItem(packet) if self.task is None or not self.task.active(): # No task queued since it was probably idle, let's queue a task. return self.queueNextServerPush()
#### Events
# Push everything we want to push from the initial configuration. self.push('start', status=self.status)
self.push('shutdown', status=self.status)
self.push('requestSubmitted', request=request)
self.push('requestCancelled', builder=builder, request=request)
self.push('buildsetSubmitted', buildset=buildset)
self.push('builderAdded', builderName=builderName, builder=builder) return self
self.push('builderChangedState', builderName=builderName, state=state)
self.push('buildStarted', build=build) return self
self.push('buildETAUpdate', build=build, ETA=ETA)
self.push('stepStarted', properties=build.getProperties().asList(), step=step)
self.push('stepTextChanged', properties=build.getProperties().asList(), step=step, text=text)
self.push('stepText2Changed', properties=build.getProperties().asList(), step=step, text2=text2)
self.push('stepETAUpdate', properties=build.getProperties().asList(), step=step, ETA=ETA, expectations=expectations)
self.push('logStarted', properties=build.getProperties().asList(), step=step)
self.push('logFinished', properties=build.getProperties().asList(), step=step)
self.push('stepFinished', properties=build.getProperties().asList(), step=step)
self.push('buildFinished', build=build)
self.push('buildedRemoved', builderName=builderName)
self.push('changeAdded', change=change)
self.push('slaveConnected', slave=self.status.getSlave(slavename))
self.push('slaveDisconnected', slavename=slavename)
"""Event streamer to a HTTP server."""
maxDiskItems=None, chunkSize=200, maxHttpRequestSize=2**20, extra_post_params=None, **kwargs): """ @serverUrl: Base URL to be used to push events notifications. @maxMemoryItems: Maximum number of items to keep queued in memory. @maxDiskItems: Maximum number of items to buffer to disk, if 0, doesn't use disk at all. @debug: Save the json with nice formatting. @chunkSize: maximum number of items to send in each at each HTTP POST. @maxHttpRequestSize: limits the size of encoded data for AE, the default is 1MB. """ if not serverUrl: raise config.ConfigErrors(['HttpStatusPush requires a serverUrl'])
# Parameters. self.serverUrl = serverUrl self.extra_post_params = extra_post_params or {} self.debug = debug self.chunkSize = chunkSize self.lastPushWasSuccessful = True self.maxHttpRequestSize = maxHttpRequestSize if maxDiskItems != 0: # The queue directory is determined by the server url. path = ('events_' + urlparse.urlparse(self.serverUrl)[1].split(':')[0]) queue = PersistentQueue( primaryQueue=MemoryQueue(maxItems=maxMemoryItems), secondaryQueue=DiskQueue(path, maxItems=maxDiskItems)) else: path = None queue = MemoryQueue(maxItems=maxMemoryItems)
# Use the unbounded method. StatusPush.__init__(self, serverPushCb=HttpStatusPush.pushHttp, queue=queue, path=path, **kwargs)
return self.lastPushWasSuccessful
"""Pops items from the pending list.
They must be queued back on failure.""" if self.wasLastPushSuccessful(): chunkSize = self.chunkSize else: chunkSize = 1
while True: items = self.queue.popChunk(chunkSize) if self.debug: packets = json.dumps(items, indent=2, sort_keys=True) else: packets = json.dumps(items, separators=(',',':')) params = {'packets': packets} params.update(self.extra_post_params) data = urllib.urlencode(params) if (not self.maxHttpRequestSize or len(data) < self.maxHttpRequestSize): return (data, items)
if chunkSize == 1: # This packet is just too large. Drop this packet. log.msg("ERROR: packet %s was dropped, too large: %d > %d" % (items[0]['id'], len(data), self.maxHttpRequestSize)) chunkSize = self.chunkSize else: # Try with half the packets. chunkSize /= 2 self.queue.insertBackChunk(items)
"""Do the HTTP POST to the server.""" (encoded_packets, items) = self.popChunk()
def Success(result): """Queue up next push.""" log.msg('Sent %d events to %s' % (len(items), self.serverUrl)) self.lastPushWasSuccessful = True return self.queueNextServerPush()
def Failure(result): """Insert back items not sent and queue up next push.""" # Server is now down. log.msg('Failed to push %d events to %s: %s' % (len(items), self.serverUrl, str(result))) self.queue.insertBackChunk(items) if self.stopped: # Bad timing, was being called on shutdown and the server died # on us. Make sure the queue is saved since we just queued back # items. self.queue.save() self.lastPushWasSuccessful = False return self.queueNextServerPush()
# Trigger the HTTP POST request. headers = {'Content-Type': 'application/x-www-form-urlencoded'} connection = client.getPage(self.serverUrl, method='POST', postdata=encoded_packets, headers=headers, agent='buildbot') connection.addCallbacks(Success, Failure) return connection
# vim: set ts=4 sts=4 sw=4 et: |