From e1316cfaa5da09428007ce1251844e64c320f01b Mon Sep 17 00:00:00 2001 From: root Date: May 10 2022 15:08:26 +0000 Subject: Automatic update of defaults --- diff --git a/roles/build-pages/tasks/main.yaml b/roles/build-pages/tasks/main.yaml index 0d946da..aa3017a 100644 --- a/roles/build-pages/tasks/main.yaml +++ b/roles/build-pages/tasks/main.yaml @@ -12,7 +12,7 @@ - name: "Ensure pages-artifacts exists" file: - path: "{{ workspace }}/{{ pages_build_dir }}" + path: "{{ ansible_user_dir }}/zuul-output/logs/pages" state: directory - name: "Check site content type" @@ -34,7 +34,7 @@ when: success_check|length > 1 - name: "Build sphinx site" - command: "sphinx-build -b html . {{ workspace }}/{{ pages_build_dir }}/" + command: "sphinx-build -b html . {{ ansible_user_dir }}/zuul-output/logs/pages" args: chdir: "{{ source }}" when: @@ -42,7 +42,7 @@ - success_check.0.item.type == "sphinx" - name: "Build pelican site" - command: "pelican content -o {{ workspace }}/{{ pages_build_dir }}/" + command: "pelican content -o {{ ansible_user_dir }}/zuul-output/logs/pages" args: chdir: "{{ source }}" when: @@ -50,5 +50,14 @@ - success_check.0.item.type == "pelican" - name: "Copy {{ source }} to {{ pages_build_dir }}/" - shell: "cp -Rf {{ source }}/* {{ workspace }}/{{ pages_build_dir }}/" + shell: "cp -Rf {{ source }}/* {{ ansible_user_dir }}/zuul-output/logs/pages" when: not success_check + +- name: Define zuul artifacts + delegate_to: localhost + zuul_return: + data: + zuul: + artifacts: + - name: "Pages preview" + url: "pages/" diff --git a/roles/submit-log-processor-jobs/README.rst b/roles/submit-log-processor-jobs/README.rst index ccb2604..b691652 100644 --- a/roles/submit-log-processor-jobs/README.rst +++ b/roles/submit-log-processor-jobs/README.rst @@ -4,3 +4,5 @@ This role is a container for an Ansible module which processes a log directory and submits jobs to a log processing gearman queue. The role itself performs no actions, and is intended only to be used by other roles as a dependency to supply the module. + +This role could be updated during upgrade diff --git a/roles/submit-log-processor-jobs/library/submit_log_processor_jobs.py b/roles/submit-log-processor-jobs/library/submit_log_processor_jobs.py index d6e078f..99464d6 100644 --- a/roles/submit-log-processor-jobs/library/submit_log_processor_jobs.py +++ b/roles/submit-log-processor-jobs/library/submit_log_processor_jobs.py @@ -18,11 +18,12 @@ import os import json import re +import traceback from ansible.module_utils.six.moves import urllib -from ansible.module_utils.basic import AnsibleModule, get_exception +from ansible.module_utils.basic import AnsibleModule -import gear +import ansible.module_utils.gear as gear class FileMatcher(object): @@ -38,16 +39,7 @@ class FileMatcher(object): class File(object): def __init__(self, name, tags): - # Note that even if we upload a .gz we want to use the logical - # non compressed name for handling (it is easier on humans). - # The reason we can get away with this is that this name is used - # to construct the log_url below. The server serving that - # log_url treats foo.txt and foo.txt.gz as being the same content - # and serves both paths from the same backend content. - if name.endswith('.gz'): - self._name = name[:-3] - else: - self._name = name + self._name = name self._tags = tags @property @@ -128,8 +120,12 @@ class LogMatcher(object): def makeEvent(self, file_object): out_event = {} out_event["fields"] = self.makeFields(file_object.name) - out_event["tags"] = [os.path.basename(file_object.name)] + \ - file_object.tags + basename = os.path.basename(file_object.name) + out_event["tags"] = [basename] + file_object.tags + if basename.endswith(".gz"): + # Backward compat for e-r which relies on tag values + # without the .gx suffix + out_event["tags"].append(basename[:-3]) return out_event def makeFields(self, filename): @@ -141,9 +137,14 @@ class LogMatcher(object): fields["build_status"] = self.success and 'SUCCESS' or 'FAILURE' # TODO: this is too simplistic for zuul v3 multinode jobs if hosts: - fields["build_node"] = hosts[0]['nodepool']['label'] + node = hosts[0] + fields["build_node"] = node['nodepool']['label'] + fields["build_hostids"] = [h['nodepool']['host_id'] for h in hosts + if 'host_id' in h['nodepool']] + fields["node_provider"] = node['nodepool']['provider'] else: fields["build_node"] = 'zuul-executor' + fields["node_provider"] = 'local' # TODO: should be build_executor, or removed completely fields["build_master"] = zuul['executor']['hostname'] @@ -153,7 +154,6 @@ class LogMatcher(object): # TODO(clarkb) can we do better without duplicated data here? fields["build_uuid"] = zuul['build'] fields["build_short_uuid"] = fields["build_uuid"][:7] - fields["build_set"] = zuul["buildset"] # TODO: this should be build_pipeline fields["build_queue"] = zuul['pipeline'] # TODO: this is not interesteding anymore @@ -166,15 +166,12 @@ class LogMatcher(object): fields["build_patchset"] = zuul['patchset'] elif 'newrev' in zuul: fields["build_newrev"] = zuul.get('newrev', 'UNKNOWN') - if hosts: - fields["node_provider"] = hosts[0]['nodepool']['provider'] - else: - fields["node_provider"] = 'local' log_url = urllib.parse.urljoin(self.log_url, filename) fields["log_url"] = log_url - fields["tenant"] = zuul["tenant"] if 'executor' in zuul and 'hostname' in zuul['executor']: fields["zuul_executor"] = zuul['executor']['hostname'] + if 'attempts' in zuul: + fields["zuul_attempts"] = zuul['attempts'] return fields @@ -209,9 +206,9 @@ def main(): results['jobs'].append(handle) module.exit_json(**results) except Exception: - e = get_exception() + tb = traceback.format_exc() module.fail_json(msg='Unknown error', - details=repr(e), + details=tb, **results) diff --git a/roles/submit-log-processor-jobs/module_utils/gear.py b/roles/submit-log-processor-jobs/module_utils/gear.py new file mode 100644 index 0000000..329cdae --- /dev/null +++ b/roles/submit-log-processor-jobs/module_utils/gear.py @@ -0,0 +1,3526 @@ +# Copyright 2013-2014 OpenStack Foundation +# +# 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 errno +import logging +import os +import select +import six +import socket +import ssl +import struct +import threading +import time +import uuid as uuid_module + +import ansible.module_utils.gear_constants as constants +from ansible.module_utils.gear_acl import ACLError, ACLEntry, ACL # noqa + +try: + import Queue as queue_mod +except ImportError: + import queue as queue_mod + +try: + import statsd +except ImportError: + statsd = None + +PRECEDENCE_NORMAL = 0 +PRECEDENCE_LOW = 1 +PRECEDENCE_HIGH = 2 + + +class ConnectionError(Exception): + pass + + +class InvalidDataError(Exception): + pass + + +class ConfigurationError(Exception): + pass + + +class NoConnectedServersError(Exception): + pass + + +class UnknownJobError(Exception): + pass + + +class InterruptedError(Exception): + pass + + +class TimeoutError(Exception): + pass + + +class GearmanError(Exception): + pass + + +class DisconnectError(Exception): + pass + + +class RetryIOError(Exception): + pass + + +def convert_to_bytes(data): + try: + data = data.encode('utf8') + except AttributeError: + pass + return data + + +class Task(object): + def __init__(self): + self._wait_event = threading.Event() + + def setComplete(self): + self._wait_event.set() + + def wait(self, timeout=None): + """Wait for a response from Gearman. + + :arg int timeout: If not None, return after this many seconds if no + response has been received (default: None). + """ + + self._wait_event.wait(timeout) + return self._wait_event.is_set() + + +class SubmitJobTask(Task): + def __init__(self, job): + super(SubmitJobTask, self).__init__() + self.job = job + + +class OptionReqTask(Task): + pass + + +class Connection(object): + """A Connection to a Gearman Server. + + :arg str client_id: The client ID associated with this connection. + It will be appending to the name of the logger (e.g., + gear.Connection.client_id). Defaults to 'unknown'. + :arg bool keepalive: Whether to use TCP keepalives + :arg int tcp_keepidle: Idle time after which to start keepalives sending + :arg int tcp_keepintvl: Interval in seconds between TCP keepalives + :arg int tcp_keepcnt: Count of TCP keepalives to send before disconnect + """ + + def __init__(self, host, port, ssl_key=None, ssl_cert=None, ssl_ca=None, + client_id='unknown', keepalive=False, tcp_keepidle=7200, + tcp_keepintvl=75, tcp_keepcnt=9): + self.log = logging.getLogger("gear.Connection.%s" % (client_id,)) + self.host = host + self.port = port + self.ssl_key = ssl_key + self.ssl_cert = ssl_cert + self.ssl_ca = ssl_ca + self.keepalive = keepalive + self.tcp_keepcnt = tcp_keepcnt + self.tcp_keepintvl = tcp_keepintvl + self.tcp_keepidle = tcp_keepidle + + self.use_ssl = False + if all([self.ssl_key, self.ssl_cert, self.ssl_ca]): + self.use_ssl = True + + self.input_buffer = b'' + self.need_bytes = False + self.echo_lock = threading.Lock() + self.send_lock = threading.Lock() + self._init() + + def _init(self): + self.conn = None + self.connected = False + self.connect_time = None + self.related_jobs = {} + self.pending_tasks = [] + self.admin_requests = [] + self.echo_conditions = {} + self.options = set() + self.changeState("INIT") + + def changeState(self, state): + # The state variables are provided as a convenience (and used by + # the Worker implementation). They aren't used or modified within + # the connection object itself except to reset to "INIT" immediately + # after reconnection. + self.log.debug("Setting state to: %s" % state) + self.state = state + self.state_time = time.time() + + def __repr__(self): + return '' % ( + id(self), self.host, self.port) + + def connect(self): + """Open a connection to the server. + + :raises ConnectionError: If unable to open the socket. + """ + + self.log.debug("Connecting to %s port %s" % (self.host, self.port)) + s = None + for res in socket.getaddrinfo(self.host, self.port, + socket.AF_UNSPEC, socket.SOCK_STREAM): + af, socktype, proto, canonname, sa = res + try: + s = socket.socket(af, socktype, proto) + if self.keepalive and hasattr(socket, 'TCP_KEEPIDLE'): + s.setsockopt(socket.SOL_SOCKET, socket.SO_KEEPALIVE, 1) + s.setsockopt(socket.IPPROTO_TCP, socket.TCP_KEEPIDLE, + self.tcp_keepidle) + s.setsockopt(socket.IPPROTO_TCP, socket.TCP_KEEPINTVL, + self.tcp_keepintvl) + s.setsockopt(socket.IPPROTO_TCP, socket.TCP_KEEPCNT, + self.tcp_keepcnt) + elif self.keepalive: + self.log.warning('Keepalive requested but not available ' + 'on this platform') + except socket.error: + s = None + continue + + if self.use_ssl: + self.log.debug("Using SSL") + context = ssl.SSLContext(ssl.PROTOCOL_TLSv1) + context.verify_mode = ssl.CERT_REQUIRED + context.check_hostname = False + context.load_cert_chain(self.ssl_cert, self.ssl_key) + context.load_verify_locations(self.ssl_ca) + s = context.wrap_socket(s, server_hostname=self.host) + + try: + s.connect(sa) + except socket.error: + s.close() + s = None + continue + break + if s is None: + self.log.debug("Error connecting to %s port %s" % ( + self.host, self.port)) + raise ConnectionError("Unable to open socket") + self.log.info("Connected to %s port %s" % (self.host, self.port)) + self.conn = s + self.connected = True + self.connect_time = time.time() + self.input_buffer = b'' + self.need_bytes = False + + def disconnect(self): + """Disconnect from the server and remove all associated state + data. + """ + + if self.conn: + try: + self.conn.close() + except Exception: + pass + + self.log.info("Disconnected from %s port %s" % (self.host, self.port)) + self._init() + + def reconnect(self): + """Disconnect from and reconnect to the server, removing all + associated state data. + """ + self.disconnect() + self.connect() + + def sendRaw(self, data): + """Send raw data over the socket. + + :arg bytes data The raw data to send + """ + with self.send_lock: + sent = 0 + while sent < len(data): + try: + sent += self.conn.send(data) + except ssl.SSLError as e: + if e.errno == ssl.SSL_ERROR_WANT_READ: + continue + elif e.errno == ssl.SSL_ERROR_WANT_WRITE: + continue + else: + raise + + def sendPacket(self, packet): + """Send a packet to the server. + + :arg Packet packet: The :py:class:`Packet` to send. + """ + self.log.info("Sending packet to %s: %s" % (self, packet)) + self.sendRaw(packet.toBinary()) + + def _getAdminRequest(self): + return self.admin_requests.pop(0) + + def _readRawBytes(self, bytes_to_read): + while True: + try: + buff = self.conn.recv(bytes_to_read) + except ssl.SSLError as e: + if e.errno == ssl.SSL_ERROR_WANT_READ: + continue + elif e.errno == ssl.SSL_ERROR_WANT_WRITE: + continue + else: + raise + break + return buff + + def _putAdminRequest(self, req): + self.admin_requests.insert(0, req) + + def readPacket(self): + """Read one packet or administrative response from the server. + + :returns: The :py:class:`Packet` or :py:class:`AdminRequest` read. + :rtype: :py:class:`Packet` or :py:class:`AdminRequest` + """ + # This handles non-blocking or blocking IO. + datalen = 0 + code = None + ptype = None + admin = None + admin_request = None + need_bytes = self.need_bytes + raw_bytes = self.input_buffer + try: + while True: + try: + if not raw_bytes or need_bytes: + segment = self._readRawBytes(4096) + if not segment: + # This occurs when the connection is closed. The + # the connect method will reset input_buffer and + # need_bytes for us. + return None + raw_bytes += segment + need_bytes = False + except RetryIOError: + if admin_request: + self._putAdminRequest(admin_request) + raise + if admin is None: + if raw_bytes[0:1] == b'\x00': + admin = False + else: + admin = True + admin_request = self._getAdminRequest() + if admin: + complete, remainder = admin_request.isComplete(raw_bytes) + if remainder is not None: + raw_bytes = remainder + if complete: + return admin_request + else: + length = len(raw_bytes) + if code is None and length >= 12: + code, ptype, datalen = struct.unpack('!4sii', + raw_bytes[:12]) + if length >= datalen + 12: + end = 12 + datalen + p = Packet(code, ptype, raw_bytes[12:end], + connection=self) + raw_bytes = raw_bytes[end:] + return p + # If we don't return a packet above then we need more data + need_bytes = True + finally: + self.input_buffer = raw_bytes + self.need_bytes = need_bytes + + def hasPendingData(self): + return self.input_buffer != b'' + + def sendAdminRequest(self, request, timeout=90): + """Send an administrative request to the server. + + :arg AdminRequest request: The :py:class:`AdminRequest` to send. + :arg numeric timeout: Number of seconds to wait until the response + is received. If None, wait forever (default: 90 seconds). + :raises TimeoutError: If the timeout is reached before the response + is received. + """ + self.admin_requests.append(request) + self.sendRaw(request.getCommand()) + complete = request.waitForResponse(timeout) + if not complete: + raise TimeoutError() + + def echo(self, data=None, timeout=30): + """Perform an echo test on the server. + + This method waits until the echo response has been received or the + timeout has been reached. + + :arg bytes data: The data to request be echoed. If None, a random + unique byte string will be generated. + :arg numeric timeout: Number of seconds to wait until the response + is received. If None, wait forever (default: 30 seconds). + :raises TimeoutError: If the timeout is reached before the response + is received. + """ + if data is None: + data = uuid_module.uuid4().hex.encode('utf8') + self.echo_lock.acquire() + try: + if data in self.echo_conditions: + raise InvalidDataError("This client is already waiting on an " + "echo response of: %s" % data) + condition = threading.Condition() + self.echo_conditions[data] = condition + finally: + self.echo_lock.release() + + self.sendEchoReq(data) + + condition.acquire() + condition.wait(timeout) + condition.release() + + if data in self.echo_conditions: + return data + raise TimeoutError() + + def sendEchoReq(self, data): + p = Packet(constants.REQ, constants.ECHO_REQ, data) + self.sendPacket(p) + + def handleEchoRes(self, data): + condition = None + self.echo_lock.acquire() + try: + condition = self.echo_conditions.get(data) + if condition: + del self.echo_conditions[data] + finally: + self.echo_lock.release() + + if not condition: + return False + condition.notifyAll() + return True + + def handleOptionRes(self, option): + self.options.add(option) + + +class AdminRequest(object): + """Encapsulates a request (and response) sent over the + administrative protocol. This is a base class that may not be + instantiated dircectly; a subclass implementing a specific command + must be used instead. + + :arg list arguments: A list of byte string arguments for the command. + + The following instance attributes are available: + + **response** (bytes) + The response from the server. + **arguments** (bytes) + The argument supplied with the constructor. + **command** (bytes) + The administrative command. + """ + + command = None + arguments = [] + response = None + _complete_position = 0 + + def __init__(self, *arguments): + self.wait_event = threading.Event() + self.arguments = arguments + if type(self) == AdminRequest: + raise NotImplementedError("AdminRequest must be subclassed") + + def __repr__(self): + return '' % ( + id(self), self.command) + + def getCommand(self): + cmd = self.command + if self.arguments: + cmd += b' ' + b' '.join(self.arguments) + cmd += b'\n' + return cmd + + def isComplete(self, data): + x = -1 + start = self._complete_position + start = max(self._complete_position - 4, 0) + end_index_newline = data.find(b'\n.\n', start) + end_index_return = data.find(b'\r\n.\r\n', start) + if end_index_newline != -1: + x = end_index_newline + 3 + elif end_index_return != -1: + x = end_index_return + 5 + elif data.startswith(b'.\n'): + x = 2 + elif data.startswith(b'.\r\n'): + x = 3 + self._complete_position = len(data) + if x != -1: + self.response = data[:x] + return (True, data[x:]) + else: + return (False, None) + + def setComplete(self): + self.wait_event.set() + + def waitForResponse(self, timeout=None): + self.wait_event.wait(timeout) + return self.wait_event.is_set() + + +class StatusAdminRequest(AdminRequest): + """A "status" administrative request. + + The response from gearman may be found in the **response** attribute. + """ + command = b'status' + + def __init__(self): + super(StatusAdminRequest, self).__init__() + + +class ShowJobsAdminRequest(AdminRequest): + """A "show jobs" administrative request. + + The response from gearman may be found in the **response** attribute. + """ + command = b'show jobs' + + def __init__(self): + super(ShowJobsAdminRequest, self).__init__() + + +class ShowUniqueJobsAdminRequest(AdminRequest): + """A "show unique jobs" administrative request. + + The response from gearman may be found in the **response** attribute. + """ + + command = b'show unique jobs' + + def __init__(self): + super(ShowUniqueJobsAdminRequest, self).__init__() + + +class CancelJobAdminRequest(AdminRequest): + """A "cancel job" administrative request. + + :arg str handle: The job handle to be canceled. + + The response from gearman may be found in the **response** attribute. + """ + + command = b'cancel job' + + def __init__(self, handle): + handle = convert_to_bytes(handle) + super(CancelJobAdminRequest, self).__init__(handle) + + def isComplete(self, data): + end_index_newline = data.find(b'\n') + if end_index_newline != -1: + x = end_index_newline + 1 + self.response = data[:x] + return (True, data[x:]) + else: + return (False, None) + + +class VersionAdminRequest(AdminRequest): + """A "version" administrative request. + + The response from gearman may be found in the **response** attribute. + """ + + command = b'version' + + def __init__(self): + super(VersionAdminRequest, self).__init__() + + def isComplete(self, data): + end_index_newline = data.find(b'\n') + if end_index_newline != -1: + x = end_index_newline + 1 + self.response = data[:x] + return (True, data[x:]) + else: + return (False, None) + + +class WorkersAdminRequest(AdminRequest): + """A "workers" administrative request. + + The response from gearman may be found in the **response** attribute. + """ + command = b'workers' + + def __init__(self): + super(WorkersAdminRequest, self).__init__() + + +class Packet(object): + """A data packet received from or to be sent over a + :py:class:`Connection`. + + :arg bytes code: The Gearman magic code (:py:data:`constants.REQ` or + :py:data:`constants.RES`) + :arg bytes ptype: The packet type (one of the packet types in + constants). + :arg bytes data: The data portion of the packet. + :arg Connection connection: The connection on which the packet + was received (optional). + :raises InvalidDataError: If the magic code is unknown. + """ + + def __init__(self, code, ptype, data, connection=None): + if not isinstance(code, bytes) and not isinstance(code, bytearray): + raise TypeError("code must be of type bytes or bytearray") + if code[0:1] != b'\x00': + raise InvalidDataError("First byte of packet must be 0") + self.code = code + self.ptype = ptype + if not isinstance(data, bytes) and not isinstance(data, bytearray): + raise TypeError("data must be of type bytes or bytearray") + self.data = data + self.connection = connection + + def __repr__(self): + ptype = constants.types.get(self.ptype, 'UNKNOWN') + try: + extra = self._formatExtraData() + except Exception: + extra = '' + return '' % (id(self), ptype, extra) + + def __eq__(self, other): + if not isinstance(other, Packet): + return False + if (self.code == other.code and + self.ptype == other.ptype and + self.data == other.data): + return True + return False + + def __ne__(self, other): + return not self.__eq__(other) + + def _formatExtraData(self): + if self.ptype in [constants.JOB_CREATED, + constants.JOB_ASSIGN, + constants.GET_STATUS, + constants.STATUS_RES, + constants.WORK_STATUS, + constants.WORK_COMPLETE, + constants.WORK_FAIL, + constants.WORK_EXCEPTION, + constants.WORK_DATA, + constants.WORK_WARNING]: + return ' handle: %s' % self.getArgument(0) + + if self.ptype == constants.JOB_ASSIGN_UNIQ: + return (' handle: %s function: %s unique: %s' % + (self.getArgument(0), + self.getArgument(1), + self.getArgument(2))) + + if self.ptype in [constants.SUBMIT_JOB, + constants.SUBMIT_JOB_BG, + constants.SUBMIT_JOB_HIGH, + constants.SUBMIT_JOB_HIGH_BG, + constants.SUBMIT_JOB_LOW, + constants.SUBMIT_JOB_LOW_BG, + constants.SUBMIT_JOB_SCHED, + constants.SUBMIT_JOB_EPOCH]: + return ' function: %s unique: %s' % (self.getArgument(0), + self.getArgument(1)) + + if self.ptype in [constants.CAN_DO, + constants.CANT_DO, + constants.CAN_DO_TIMEOUT]: + return ' function: %s' % (self.getArgument(0),) + + if self.ptype == constants.SET_CLIENT_ID: + return ' id: %s' % (self.getArgument(0),) + + if self.ptype in [constants.OPTION_REQ, + constants.OPTION_RES]: + return ' option: %s' % (self.getArgument(0),) + + if self.ptype == constants.ERROR: + return ' code: %s message: %s' % (self.getArgument(0), + self.getArgument(1)) + return '' + + def toBinary(self): + """Return a Gearman wire protocol binary representation of the packet. + + :returns: The packet in binary form. + :rtype: bytes + """ + b = struct.pack('!4sii', self.code, self.ptype, len(self.data)) + b = bytearray(b) + b += self.data + return b + + def getArgument(self, index, last=False): + """Get the nth argument from the packet data. + + :arg int index: The argument index to look up. + :arg bool last: Whether this is the last argument (and thus + nulls should be ignored) + :returns: The argument value. + :rtype: bytes + """ + + parts = self.data.split(b'\x00') + if not last: + return parts[index] + return b'\x00'.join(parts[index:]) + + def getJob(self): + """Get the :py:class:`Job` associated with the job handle in + this packet. + + :returns: The :py:class:`Job` for this packet. + :rtype: Job + :raises UnknownJobError: If the job is not known. + """ + handle = self.getArgument(0) + job = self.connection.related_jobs.get(handle) + if not job: + raise UnknownJobError() + return job + + +class BaseClientServer(object): + def __init__(self, client_id=None): + if client_id: + self.client_id = convert_to_bytes(client_id) + self.log = logging.getLogger("gear.BaseClientServer.%s" % + (self.client_id,)) + else: + self.client_id = None + self.log = logging.getLogger("gear.BaseClientServer") + self.running = True + self.active_connections = [] + self.inactive_connections = [] + + self.connection_index = -1 + # A lock and notification mechanism to handle not having any + # current connections + self.connections_condition = threading.Condition() + + # A pipe to wake up the poll loop in case it needs to restart + self.wake_read, self.wake_write = os.pipe() + + self.poll_thread = threading.Thread(name="Gearman client poll", + target=self._doPollLoop) + self.poll_thread.daemon = True + self.poll_thread.start() + self.connect_thread = threading.Thread(name="Gearman client connect", + target=self._doConnectLoop) + self.connect_thread.daemon = True + self.connect_thread.start() + + def _doConnectLoop(self): + # Outer run method of the reconnection thread + while self.running: + self.connections_condition.acquire() + while self.running and not self.inactive_connections: + self.log.debug("Waiting for change in available servers " + "to reconnect") + self.connections_condition.wait() + self.connections_condition.release() + self.log.debug("Checking if servers need to be reconnected") + try: + if self.running and not self._connectLoop(): + # Nothing happened + time.sleep(2) + except Exception: + self.log.exception("Exception in connect loop:") + + def _connectLoop(self): + # Inner method of the reconnection loop, triggered by + # a connection change + success = False + for conn in self.inactive_connections[:]: + self.log.debug("Trying to reconnect %s" % conn) + try: + conn.reconnect() + except ConnectionError: + self.log.debug("Unable to connect to %s" % conn) + continue + except Exception: + self.log.exception("Exception while connecting to %s" % conn) + continue + + try: + self._onConnect(conn) + except Exception: + self.log.exception("Exception while performing on-connect " + "tasks for %s" % conn) + continue + self.connections_condition.acquire() + self.inactive_connections.remove(conn) + self.active_connections.append(conn) + self.connections_condition.notifyAll() + os.write(self.wake_write, b'1\n') + self.connections_condition.release() + + try: + self._onActiveConnection(conn) + except Exception: + self.log.exception("Exception while performing active conn " + "tasks for %s" % conn) + + success = True + return success + + def _onConnect(self, conn): + # Called immediately after a successful (re-)connection + pass + + def _onActiveConnection(self, conn): + # Called immediately after a connection is activated + pass + + def _lostConnection(self, conn): + # Called as soon as a connection is detected as faulty. Remove + # it and return ASAP and let the connection thread deal with it. + self.log.debug("Marking %s as disconnected" % conn) + self.connections_condition.acquire() + try: + # NOTE(notmorgan): In the loop below it is possible to change the + # jobs list on the connection. In python 3 .values() is an iter not + # a static list, meaning that a change will break the for loop + # as the object being iterated on will have changed in size. + jobs = list(conn.related_jobs.values()) + if conn in self.active_connections: + self.active_connections.remove(conn) + if conn not in self.inactive_connections: + self.inactive_connections.append(conn) + finally: + self.connections_condition.notifyAll() + self.connections_condition.release() + for job in jobs: + self.handleDisconnect(job) + + def _doPollLoop(self): + # Outer run method of poll thread. + while self.running: + self.connections_condition.acquire() + while self.running and not self.active_connections: + self.log.debug("Waiting for change in available connections " + "to poll") + self.connections_condition.wait() + self.connections_condition.release() + try: + self._pollLoop() + except socket.error as e: + if e.errno == errno.ECONNRESET: + self.log.debug("Connection reset by peer") + # This will get logged later at info level as + # "Marking ... as disconnected" + except Exception: + self.log.exception("Exception in poll loop:") + + def _pollLoop(self): + # Inner method of poll loop + self.log.debug("Preparing to poll") + poll = select.poll() + bitmask = (select.POLLIN | select.POLLERR | + select.POLLHUP | select.POLLNVAL) + # Reverse mapping of fd -> connection + conn_dict = {} + for conn in self.active_connections: + poll.register(conn.conn.fileno(), bitmask) + conn_dict[conn.conn.fileno()] = conn + # Register the wake pipe so that we can break if we need to + # reconfigure connections + poll.register(self.wake_read, bitmask) + while self.running: + self.log.debug("Polling %s connections" % + len(self.active_connections)) + ret = poll.poll() + for fd, event in ret: + if fd == self.wake_read: + self.log.debug("Woken by pipe") + while True: + if os.read(self.wake_read, 1) == b'\n': + break + return + conn = conn_dict[fd] + if event & select.POLLIN: + # Process all packets that may have been read in this + # round of recv's by readPacket. + while True: + self.log.debug("Processing input on %s" % conn) + p = conn.readPacket() + if p: + if isinstance(p, Packet): + self.handlePacket(p) + else: + self.handleAdminRequest(p) + else: + self.log.debug("Received no data on %s" % conn) + self._lostConnection(conn) + return + if not conn.hasPendingData(): + break + else: + self.log.debug("Received error event on %s" % conn) + self._lostConnection(conn) + return + + def handlePacket(self, packet): + """Handle a received packet. + + This method is called whenever a packet is received from any + connection. It normally calls the handle method appropriate + for the specific packet. + + :arg Packet packet: The :py:class:`Packet` that was received. + """ + + self.log.info("Received packet from %s: %s" % (packet.connection, + packet)) + start = time.time() + if packet.ptype == constants.JOB_CREATED: + self.handleJobCreated(packet) + elif packet.ptype == constants.WORK_COMPLETE: + self.handleWorkComplete(packet) + elif packet.ptype == constants.WORK_FAIL: + self.handleWorkFail(packet) + elif packet.ptype == constants.WORK_EXCEPTION: + self.handleWorkException(packet) + elif packet.ptype == constants.WORK_DATA: + self.handleWorkData(packet) + elif packet.ptype == constants.WORK_WARNING: + self.handleWorkWarning(packet) + elif packet.ptype == constants.WORK_STATUS: + self.handleWorkStatus(packet) + elif packet.ptype == constants.STATUS_RES: + self.handleStatusRes(packet) + elif packet.ptype == constants.GET_STATUS: + self.handleGetStatus(packet) + elif packet.ptype == constants.JOB_ASSIGN_UNIQ: + self.handleJobAssignUnique(packet) + elif packet.ptype == constants.JOB_ASSIGN: + self.handleJobAssign(packet) + elif packet.ptype == constants.NO_JOB: + self.handleNoJob(packet) + elif packet.ptype == constants.NOOP: + self.handleNoop(packet) + elif packet.ptype == constants.SUBMIT_JOB: + self.handleSubmitJob(packet) + elif packet.ptype == constants.SUBMIT_JOB_BG: + self.handleSubmitJobBg(packet) + elif packet.ptype == constants.SUBMIT_JOB_HIGH: + self.handleSubmitJobHigh(packet) + elif packet.ptype == constants.SUBMIT_JOB_HIGH_BG: + self.handleSubmitJobHighBg(packet) + elif packet.ptype == constants.SUBMIT_JOB_LOW: + self.handleSubmitJobLow(packet) + elif packet.ptype == constants.SUBMIT_JOB_LOW_BG: + self.handleSubmitJobLowBg(packet) + elif packet.ptype == constants.SUBMIT_JOB_SCHED: + self.handleSubmitJobSched(packet) + elif packet.ptype == constants.SUBMIT_JOB_EPOCH: + self.handleSubmitJobEpoch(packet) + elif packet.ptype == constants.GRAB_JOB_UNIQ: + self.handleGrabJobUniq(packet) + elif packet.ptype == constants.GRAB_JOB: + self.handleGrabJob(packet) + elif packet.ptype == constants.PRE_SLEEP: + self.handlePreSleep(packet) + elif packet.ptype == constants.SET_CLIENT_ID: + self.handleSetClientID(packet) + elif packet.ptype == constants.CAN_DO: + self.handleCanDo(packet) + elif packet.ptype == constants.CAN_DO_TIMEOUT: + self.handleCanDoTimeout(packet) + elif packet.ptype == constants.CANT_DO: + self.handleCantDo(packet) + elif packet.ptype == constants.RESET_ABILITIES: + self.handleResetAbilities(packet) + elif packet.ptype == constants.ECHO_REQ: + self.handleEchoReq(packet) + elif packet.ptype == constants.ECHO_RES: + self.handleEchoRes(packet) + elif packet.ptype == constants.ERROR: + self.handleError(packet) + elif packet.ptype == constants.ALL_YOURS: + self.handleAllYours(packet) + elif packet.ptype == constants.OPTION_REQ: + self.handleOptionReq(packet) + elif packet.ptype == constants.OPTION_RES: + self.handleOptionRes(packet) + else: + self.log.error("Received unknown packet: %s" % packet) + end = time.time() + self.reportTimingStats(packet.ptype, end - start) + + def reportTimingStats(self, ptype, duration): + """Report processing times by packet type + + This method is called by handlePacket to report how long + processing took for each packet. The default implementation + does nothing. + + :arg bytes ptype: The packet type (one of the packet types in + constants). + :arg float duration: The time (in seconds) it took to process + the packet. + """ + pass + + def _defaultPacketHandler(self, packet): + self.log.error("Received unhandled packet: %s" % packet) + + def handleJobCreated(self, packet): + return self._defaultPacketHandler(packet) + + def handleWorkComplete(self, packet): + return self._defaultPacketHandler(packet) + + def handleWorkFail(self, packet): + return self._defaultPacketHandler(packet) + + def handleWorkException(self, packet): + return self._defaultPacketHandler(packet) + + def handleWorkData(self, packet): + return self._defaultPacketHandler(packet) + + def handleWorkWarning(self, packet): + return self._defaultPacketHandler(packet) + + def handleWorkStatus(self, packet): + return self._defaultPacketHandler(packet) + + def handleStatusRes(self, packet): + return self._defaultPacketHandler(packet) + + def handleGetStatus(self, packet): + return self._defaultPacketHandler(packet) + + def handleJobAssignUnique(self, packet): + return self._defaultPacketHandler(packet) + + def handleJobAssign(self, packet): + return self._defaultPacketHandler(packet) + + def handleNoJob(self, packet): + return self._defaultPacketHandler(packet) + + def handleNoop(self, packet): + return self._defaultPacketHandler(packet) + + def handleSubmitJob(self, packet): + return self._defaultPacketHandler(packet) + + def handleSubmitJobBg(self, packet): + return self._defaultPacketHandler(packet) + + def handleSubmitJobHigh(self, packet): + return self._defaultPacketHandler(packet) + + def handleSubmitJobHighBg(self, packet): + return self._defaultPacketHandler(packet) + + def handleSubmitJobLow(self, packet): + return self._defaultPacketHandler(packet) + + def handleSubmitJobLowBg(self, packet): + return self._defaultPacketHandler(packet) + + def handleSubmitJobSched(self, packet): + return self._defaultPacketHandler(packet) + + def handleSubmitJobEpoch(self, packet): + return self._defaultPacketHandler(packet) + + def handleGrabJobUniq(self, packet): + return self._defaultPacketHandler(packet) + + def handleGrabJob(self, packet): + return self._defaultPacketHandler(packet) + + def handlePreSleep(self, packet): + return self._defaultPacketHandler(packet) + + def handleSetClientID(self, packet): + return self._defaultPacketHandler(packet) + + def handleCanDo(self, packet): + return self._defaultPacketHandler(packet) + + def handleCanDoTimeout(self, packet): + return self._defaultPacketHandler(packet) + + def handleCantDo(self, packet): + return self._defaultPacketHandler(packet) + + def handleResetAbilities(self, packet): + return self._defaultPacketHandler(packet) + + def handleEchoReq(self, packet): + return self._defaultPacketHandler(packet) + + def handleEchoRes(self, packet): + return self._defaultPacketHandler(packet) + + def handleError(self, packet): + return self._defaultPacketHandler(packet) + + def handleAllYours(self, packet): + return self._defaultPacketHandler(packet) + + def handleOptionReq(self, packet): + return self._defaultPacketHandler(packet) + + def handleOptionRes(self, packet): + return self._defaultPacketHandler(packet) + + def handleAdminRequest(self, request): + """Handle an administrative command response from Gearman. + + This method is called whenever a response to a previously + issued administrative command is received from one of this + client's connections. It normally releases the wait lock on + the initiating AdminRequest object. + + :arg AdminRequest request: The :py:class:`AdminRequest` that + initiated the received response. + """ + + self.log.info("Received admin data %s" % request) + request.setComplete() + + def shutdown(self): + """Close all connections and stop all running threads. + + The object may no longer be used after shutdown is called. + """ + if self.running: + self.log.debug("Beginning shutdown") + self._shutdown() + self.log.debug("Beginning cleanup") + self._cleanup() + self.log.debug("Finished shutdown") + else: + self.log.warning("Shutdown called when not currently running. " + "Ignoring.") + + def _shutdown(self): + # The first part of the shutdown process where all threads + # are told to exit. + self.running = False + self.connections_condition.acquire() + try: + self.connections_condition.notifyAll() + os.write(self.wake_write, b'1\n') + finally: + self.connections_condition.release() + + def _cleanup(self): + # The second part of the shutdown process where we wait for all + # threads to exit and then clean up. + self.poll_thread.join() + self.connect_thread.join() + for connection in self.active_connections: + connection.disconnect() + self.active_connections = [] + self.inactive_connections = [] + os.close(self.wake_read) + os.close(self.wake_write) + + +class BaseClient(BaseClientServer): + def __init__(self, client_id='unknown'): + super(BaseClient, self).__init__(client_id) + self.log = logging.getLogger("gear.BaseClient.%s" % (self.client_id,)) + # A lock to use when sending packets that set the state across + # all known connections. Note that it doesn't necessarily need + # to be used for all broadcasts, only those that affect multi- + # connection state, such as setting options or functions. + self.broadcast_lock = threading.RLock() + + def addServer(self, host, port=4730, + ssl_key=None, ssl_cert=None, ssl_ca=None, + keepalive=False, tcp_keepidle=7200, tcp_keepintvl=75, + tcp_keepcnt=9): + """Add a server to the client's connection pool. + + Any number of Gearman servers may be added to a client. The + client will connect to all of them and send jobs to them in a + round-robin fashion. When servers are disconnected, the + client will automatically remove them from the pool, + continuously try to reconnect to them, and return them to the + pool when reconnected. New servers may be added at any time. + + This is a non-blocking call that will return regardless of + whether the initial connection succeeded. If you need to + ensure that a connection is ready before proceeding, see + :py:meth:`waitForServer`. + + When using SSL connections, all SSL files must be specified. + + :arg str host: The hostname or IP address of the server. + :arg int port: The port on which the gearman server is listening. + :arg str ssl_key: Path to the SSL private key. + :arg str ssl_cert: Path to the SSL certificate. + :arg str ssl_ca: Path to the CA certificate. + :arg bool keepalive: Whether to use TCP keepalives + :arg int tcp_keepidle: Idle time after which to start keepalives + sending + :arg int tcp_keepintvl: Interval in seconds between TCP keepalives + :arg int tcp_keepcnt: Count of TCP keepalives to send before disconnect + :raises ConfigurationError: If the host/port combination has + already been added to the client. + """ + + self.log.debug("Adding server %s port %s" % (host, port)) + + self.connections_condition.acquire() + try: + for conn in self.active_connections + self.inactive_connections: + if conn.host == host and conn.port == port: + raise ConfigurationError("Host/port already specified") + conn = Connection(host, port, ssl_key, ssl_cert, ssl_ca, + self.client_id, keepalive, tcp_keepidle, + tcp_keepintvl, tcp_keepcnt) + self.inactive_connections.append(conn) + self.connections_condition.notifyAll() + finally: + self.connections_condition.release() + + def _checkTimeout(self, start_time, timeout): + if time.time() - start_time > timeout: + raise TimeoutError() + + def waitForServer(self, timeout=None): + """Wait for at least one server to be connected. + + Block until at least one gearman server is connected. + + :arg numeric timeout: Number of seconds to wait for a connection. + If None, wait forever (default: no timeout). + :raises TimeoutError: If the timeout is reached before any server + connects. + """ + + connected = False + start_time = time.time() + while self.running: + self.connections_condition.acquire() + while self.running and not self.active_connections: + if timeout is not None: + self._checkTimeout(start_time, timeout) + self.log.debug("Waiting for at least one active connection") + self.connections_condition.wait(timeout=1) + if self.active_connections: + self.log.debug("Active connection found") + connected = True + self.connections_condition.release() + if connected: + return + + def getConnection(self): + """Return a connected server. + + Finds the next scheduled connected server in the round-robin + rotation and returns it. It is not usually necessary to use + this method external to the library, as more consumer-oriented + methods such as submitJob already use it internally, but is + available nonetheless if necessary. + + :returns: The next scheduled :py:class:`Connection` object. + :rtype: :py:class:`Connection` + :raises NoConnectedServersError: If there are not currently + connected servers. + """ + + conn = None + try: + self.connections_condition.acquire() + if not self.active_connections: + raise NoConnectedServersError("No connected Gearman servers") + + self.connection_index += 1 + if self.connection_index >= len(self.active_connections): + self.connection_index = 0 + conn = self.active_connections[self.connection_index] + finally: + self.connections_condition.release() + return conn + + def broadcast(self, packet): + """Send a packet to all currently connected servers. + + :arg Packet packet: The :py:class:`Packet` to send. + """ + connections = self.active_connections[:] + for connection in connections: + try: + self.sendPacket(packet, connection) + except Exception: + # Error handling is all done by sendPacket + pass + + def sendPacket(self, packet, connection): + """Send a packet to a single connection, removing it from the + list of active connections if that fails. + + :arg Packet packet: The :py:class:`Packet` to send. + :arg Connection connection: The :py:class:`Connection` on + which to send the packet. + """ + try: + connection.sendPacket(packet) + return + except Exception: + self.log.exception("Exception while sending packet %s to %s" % + (packet, connection)) + # If we can't send the packet, discard the connection + self._lostConnection(connection) + raise + + def handleEchoRes(self, packet): + """Handle an ECHO_RES packet. + + Causes the blocking :py:meth:`Connection.echo` invocation to + return. + + :arg Packet packet: The :py:class:`Packet` that was received. + :returns: None + """ + packet.connection.handleEchoRes(packet.getArgument(0, True)) + + def handleError(self, packet): + """Handle an ERROR packet. + + Logs the error. + + :arg Packet packet: The :py:class:`Packet` that was received. + :returns: None + """ + self.log.error("Received ERROR packet: %s: %s" % + (packet.getArgument(0), + packet.getArgument(1))) + try: + task = packet.connection.pending_tasks.pop(0) + task.setComplete() + except Exception: + self.log.exception("Exception while handling error packet:") + self._lostConnection(packet.connection) + + +class Client(BaseClient): + """A Gearman client. + + You may wish to subclass this class in order to override the + default event handlers to react to Gearman events. Be sure to + call the superclass event handlers so that they may perform + job-related housekeeping. + + :arg str client_id: The client ID to provide to Gearman. It will + appear in administrative output and be appended to the name of + the logger (e.g., gear.Client.client_id). Defaults to + 'unknown'. + """ + + def __init__(self, client_id='unknown'): + super(Client, self).__init__(client_id) + self.log = logging.getLogger("gear.Client.%s" % (self.client_id,)) + self.options = set() + + def __repr__(self): + return '' % id(self) + + def _onConnect(self, conn): + # Called immediately after a successful (re-)connection + self.broadcast_lock.acquire() + try: + super(Client, self)._onConnect(conn) + for name in self.options: + self._setOptionConnection(name, conn) + finally: + self.broadcast_lock.release() + + def _setOptionConnection(self, name, conn): + # Set an option on a connection + packet = Packet(constants.REQ, constants.OPTION_REQ, name) + task = OptionReqTask() + try: + conn.pending_tasks.append(task) + self.sendPacket(packet, conn) + except Exception: + # Error handling is all done by sendPacket + task = None + return task + + def setOption(self, name, timeout=30): + """Set an option for all connections. + + :arg str name: The option name to set. + :arg int timeout: How long to wait (in seconds) for a response + from the server before giving up (default: 30 seconds). + :returns: True if the option was set on all connections, + otherwise False + :rtype: bool + """ + tasks = {} + name = convert_to_bytes(name) + self.broadcast_lock.acquire() + + try: + self.options.add(name) + connections = self.active_connections[:] + for connection in connections: + task = self._setOptionConnection(name, connection) + if task: + tasks[task] = connection + finally: + self.broadcast_lock.release() + + success = True + for task in tasks.keys(): + complete = task.wait(timeout) + conn = tasks[task] + if not complete: + self.log.error("Connection %s timed out waiting for a " + "response to an option request: %s" % + (conn, name)) + self._lostConnection(conn) + continue + if name not in conn.options: + success = False + return success + + def submitJob(self, job, background=False, precedence=PRECEDENCE_NORMAL, + timeout=30): + """Submit a job to a Gearman server. + + Submits the provided job to the next server in this client's + round-robin connection pool. + + If the job is a foreground job, updates will be made to the + supplied :py:class:`Job` object as they are received. + + :arg Job job: The :py:class:`Job` to submit. + :arg bool background: Whether the job should be backgrounded. + :arg int precedence: Whether the job should have normal, low, or + high precedence. One of :py:data:`PRECEDENCE_NORMAL`, + :py:data:`PRECEDENCE_LOW`, or :py:data:`PRECEDENCE_HIGH` + :arg int timeout: How long to wait (in seconds) for a response + from the server before giving up (default: 30 seconds). + :raises ConfigurationError: If an invalid precendence value + is supplied. + """ + if job.unique is None: + unique = b'' + else: + unique = job.binary_unique + data = b'\x00'.join((job.binary_name, unique, job.binary_arguments)) + if background: + if precedence == PRECEDENCE_NORMAL: + cmd = constants.SUBMIT_JOB_BG + elif precedence == PRECEDENCE_LOW: + cmd = constants.SUBMIT_JOB_LOW_BG + elif precedence == PRECEDENCE_HIGH: + cmd = constants.SUBMIT_JOB_HIGH_BG + else: + raise ConfigurationError("Invalid precedence value") + else: + if precedence == PRECEDENCE_NORMAL: + cmd = constants.SUBMIT_JOB + elif precedence == PRECEDENCE_LOW: + cmd = constants.SUBMIT_JOB_LOW + elif precedence == PRECEDENCE_HIGH: + cmd = constants.SUBMIT_JOB_HIGH + else: + raise ConfigurationError("Invalid precedence value") + packet = Packet(constants.REQ, cmd, data) + attempted_connections = set() + while True: + if attempted_connections == set(self.active_connections): + break + conn = self.getConnection() + task = SubmitJobTask(job) + conn.pending_tasks.append(task) + attempted_connections.add(conn) + try: + self.sendPacket(packet, conn) + except Exception: + # Error handling is all done by sendPacket + continue + complete = task.wait(timeout) + if not complete: + self.log.error("Connection %s timed out waiting for a " + "response to a submit job request: %s" % + (conn, job)) + self._lostConnection(conn) + continue + if not job.handle: + self.log.error("Connection %s sent an error in " + "response to a submit job request: %s" % + (conn, job)) + continue + job.connection = conn + return + raise GearmanError("Unable to submit job to any connected servers") + + def handleJobCreated(self, packet): + """Handle a JOB_CREATED packet. + + Updates the appropriate :py:class:`Job` with the newly + returned job handle. + + :arg Packet packet: The :py:class:`Packet` that was received. + :returns: The :py:class:`Job` object associated with the job request. + :rtype: :py:class:`Job` + """ + task = packet.connection.pending_tasks.pop(0) + if not isinstance(task, SubmitJobTask): + msg = ("Unexpected response received to submit job " + "request: %s" % packet) + self.log.error(msg) + self._lostConnection(packet.connection) + raise GearmanError(msg) + + job = task.job + job.handle = packet.data + packet.connection.related_jobs[job.handle] = job + task.setComplete() + self.log.debug("Job created; %s" % job) + return job + + def handleWorkComplete(self, packet): + """Handle a WORK_COMPLETE packet. + + Updates the referenced :py:class:`Job` with the returned data + and removes it from the list of jobs associated with the + connection. + + :arg Packet packet: The :py:class:`Packet` that was received. + :returns: The :py:class:`Job` object associated with the job request. + :rtype: :py:class:`Job` + """ + + job = packet.getJob() + data = packet.getArgument(1, True) + if data: + job.data.append(data) + job.complete = True + job.failure = False + del packet.connection.related_jobs[job.handle] + self.log.debug("Job complete; %s data: %s" % + (job, job.data)) + return job + + def handleWorkFail(self, packet): + """Handle a WORK_FAIL packet. + + Updates the referenced :py:class:`Job` with the returned data + and removes it from the list of jobs associated with the + connection. + + :arg Packet packet: The :py:class:`Packet` that was received. + :returns: The :py:class:`Job` object associated with the job request. + :rtype: :py:class:`Job` + """ + + job = packet.getJob() + job.complete = True + job.failure = True + del packet.connection.related_jobs[job.handle] + self.log.debug("Job failed; %s" % job) + return job + + def handleWorkException(self, packet): + """Handle a WORK_Exception packet. + + Updates the referenced :py:class:`Job` with the returned data + and removes it from the list of jobs associated with the + connection. + + :arg Packet packet: The :py:class:`Packet` that was received. + :returns: The :py:class:`Job` object associated with the job request. + :rtype: :py:class:`Job` + """ + + job = packet.getJob() + job.exception = packet.getArgument(1, True) + job.complete = True + job.failure = True + del packet.connection.related_jobs[job.handle] + self.log.debug("Job exception; %s exception: %s" % + (job, job.exception)) + return job + + def handleWorkData(self, packet): + """Handle a WORK_DATA packet. + + Updates the referenced :py:class:`Job` with the returned data. + + :arg Packet packet: The :py:class:`Packet` that was received. + :returns: The :py:class:`Job` object associated with the job request. + :rtype: :py:class:`Job` + """ + + job = packet.getJob() + data = packet.getArgument(1, True) + if data: + job.data.append(data) + self.log.debug("Job data; job: %s data: %s" % + (job, job.data)) + return job + + def handleWorkWarning(self, packet): + """Handle a WORK_WARNING packet. + + Updates the referenced :py:class:`Job` with the returned data. + + :arg Packet packet: The :py:class:`Packet` that was received. + :returns: The :py:class:`Job` object associated with the job request. + :rtype: :py:class:`Job` + """ + + job = packet.getJob() + data = packet.getArgument(1, True) + if data: + job.data.append(data) + job.warning = True + self.log.debug("Job warning; %s data: %s" % + (job, job.data)) + return job + + def handleWorkStatus(self, packet): + """Handle a WORK_STATUS packet. + + Updates the referenced :py:class:`Job` with the returned data. + + :arg Packet packet: The :py:class:`Packet` that was received. + :returns: The :py:class:`Job` object associated with the job request. + :rtype: :py:class:`Job` + """ + + job = packet.getJob() + job.numerator = packet.getArgument(1) + job.denominator = packet.getArgument(2) + try: + job.fraction_complete = (float(job.numerator) / + float(job.denominator)) + except Exception: + job.fraction_complete = None + self.log.debug("Job status; %s complete: %s/%s" % + (job, job.numerator, job.denominator)) + return job + + def handleStatusRes(self, packet): + """Handle a STATUS_RES packet. + + Updates the referenced :py:class:`Job` with the returned data. + + :arg Packet packet: The :py:class:`Packet` that was received. + :returns: The :py:class:`Job` object associated with the job request. + :rtype: :py:class:`Job` + """ + + job = packet.getJob() + job.known = (packet.getArgument(1) == b'1') + job.running = (packet.getArgument(2) == b'1') + job.numerator = packet.getArgument(3) + job.denominator = packet.getArgument(4) + + try: + job.fraction_complete = (float(job.numerator) / + float(job.denominator)) + except Exception: + job.fraction_complete = None + return job + + def handleOptionRes(self, packet): + """Handle an OPTION_RES packet. + + Updates the set of options for the connection. + + :arg Packet packet: The :py:class:`Packet` that was received. + :returns: None. + """ + task = packet.connection.pending_tasks.pop(0) + if not isinstance(task, OptionReqTask): + msg = ("Unexpected response received to option " + "request: %s" % packet) + self.log.error(msg) + self._lostConnection(packet.connection) + raise GearmanError(msg) + + packet.connection.handleOptionRes(packet.getArgument(0)) + task.setComplete() + + def handleDisconnect(self, job): + """Handle a Gearman server disconnection. + + If the Gearman server is disconnected, this will be called for any + jobs currently associated with the server. + + :arg Job packet: The :py:class:`Job` that was running when the server + disconnected. + """ + return job + + +class FunctionRecord(object): + """Represents a function that should be registered with Gearman. + + This class only directly needs to be instatiated for use with + :py:meth:`Worker.setFunctions`. If a timeout value is supplied, + the function will be registered with CAN_DO_TIMEOUT. + + :arg str name: The name of the function to register. + :arg numeric timeout: The timeout value (optional). + """ + def __init__(self, name, timeout=None): + self.name = name + self.timeout = timeout + + def __repr__(self): + return '' % ( + id(self), self.name, self.timeout) + + +class BaseJob(object): + def __init__(self, name, arguments, unique=None, handle=None): + self._name = convert_to_bytes(name) + self._validate_arguments(arguments) + self._arguments = convert_to_bytes(arguments) + self._unique = convert_to_bytes(unique) + self.handle = handle + self.connection = None + + def _validate_arguments(self, arguments): + if (not isinstance(arguments, bytes) and + not isinstance(arguments, bytearray)): + raise TypeError("arguments must be of type bytes or bytearray") + + @property + def arguments(self): + return self._arguments + + @arguments.setter + def arguments(self, value): + self._arguments = value + + @property + def unique(self): + return self._unique + + @unique.setter + def unique(self, value): + self._unique = value + + @property + def name(self): + if isinstance(self._name, six.binary_type): + return self._name.decode('utf-8') + return self._name + + @name.setter + def name(self, value): + if isinstance(value, six.text_type): + value = value.encode('utf-8') + self._name = value + + @property + def binary_name(self): + return self._name + + @property + def binary_arguments(self): + return self._arguments + + @property + def binary_unique(self): + return self._unique + + def __repr__(self): + return '' % ( + id(self), self.handle, self.name, self.unique) + + +class WorkerJob(BaseJob): + """A job that Gearman has assigned to a Worker. Not intended to + be instantiated directly, but rather returned by + :py:meth:`Worker.getJob`. + + :arg str handle: The job handle assigned by gearman. + :arg str name: The name of the job. + :arg bytes arguments: The opaque data blob passed to the worker + as arguments. + :arg str unique: A byte string to uniquely identify the job to Gearman + (optional). + + The following instance attributes are available: + + **name** (str) + The name of the job. Assumed to be utf-8. + **arguments** (bytes) + The opaque data blob passed to the worker as arguments. + **unique** (str or None) + The unique ID of the job (if supplied). + **handle** (bytes) + The Gearman job handle. + **connection** (:py:class:`Connection` or None) + The connection associated with the job. Only set after the job + has been submitted to a Gearman server. + """ + + def __init__(self, handle, name, arguments, unique=None): + super(WorkerJob, self).__init__(name, arguments, unique, handle) + + def sendWorkData(self, data=b''): + """Send a WORK_DATA packet to the client. + + :arg bytes data: The data to be sent to the client (optional). + """ + + data = self.handle + b'\x00' + data + p = Packet(constants.REQ, constants.WORK_DATA, data) + self.connection.sendPacket(p) + + def sendWorkWarning(self, data=b''): + """Send a WORK_WARNING packet to the client. + + :arg bytes data: The data to be sent to the client (optional). + """ + + data = self.handle + b'\x00' + data + p = Packet(constants.REQ, constants.WORK_WARNING, data) + self.connection.sendPacket(p) + + def sendWorkStatus(self, numerator, denominator): + """Send a WORK_STATUS packet to the client. + + Sends a numerator and denominator that together represent the + fraction complete of the job. + + :arg numeric numerator: The numerator of the fraction complete. + :arg numeric denominator: The denominator of the fraction complete. + """ + + data = (self.handle + b'\x00' + + str(numerator).encode('utf8') + b'\x00' + + str(denominator).encode('utf8')) + p = Packet(constants.REQ, constants.WORK_STATUS, data) + self.connection.sendPacket(p) + + def sendWorkComplete(self, data=b''): + """Send a WORK_COMPLETE packet to the client. + + :arg bytes data: The data to be sent to the client (optional). + """ + + data = self.handle + b'\x00' + data + p = Packet(constants.REQ, constants.WORK_COMPLETE, data) + self.connection.sendPacket(p) + + def sendWorkFail(self): + "Send a WORK_FAIL packet to the client." + + p = Packet(constants.REQ, constants.WORK_FAIL, self.handle) + self.connection.sendPacket(p) + + def sendWorkException(self, data=b''): + """Send a WORK_EXCEPTION packet to the client. + + :arg bytes data: The exception data to be sent to the client + (optional). + """ + + data = self.handle + b'\x00' + data + p = Packet(constants.REQ, constants.WORK_EXCEPTION, data) + self.connection.sendPacket(p) + + +class Worker(BaseClient): + """A Gearman worker. + + :arg str client_id: The client ID to provide to Gearman. It will + appear in administrative output and be appended to the name of + the logger (e.g., gear.Worker.client_id). + :arg str worker_id: The client ID to provide to Gearman. It will + appear in administrative output and be appended to the name of + the logger (e.g., gear.Worker.client_id). This parameter name + is deprecated, use client_id instead. + """ + + job_class = WorkerJob + + def __init__(self, client_id=None, worker_id=None): + if not client_id or worker_id: + raise Exception("A client_id must be provided") + if worker_id: + client_id = worker_id + super(Worker, self).__init__(client_id) + self.log = logging.getLogger("gear.Worker.%s" % (self.client_id,)) + self.worker_id = client_id + self.functions = {} + self.job_lock = threading.Lock() + self.waiting_for_jobs = 0 + self.job_queue = queue_mod.Queue() + + def __repr__(self): + return '' % id(self) + + def registerFunction(self, name, timeout=None): + """Register a function with Gearman. + + If a timeout value is supplied, the function will be + registered with CAN_DO_TIMEOUT. + + :arg str name: The name of the function to register. + :arg numeric timeout: The timeout value (optional). + """ + name = convert_to_bytes(name) + self.functions[name] = FunctionRecord(name, timeout) + if timeout: + self._sendCanDoTimeout(name, timeout) + else: + self._sendCanDo(name) + + connections = self.active_connections[:] + for connection in connections: + if connection.state == "SLEEP": + connection.changeState("IDLE") + self._updateStateMachines() + + def unRegisterFunction(self, name): + """Remove a function from Gearman's registry. + + :arg str name: The name of the function to remove. + """ + name = convert_to_bytes(name) + del self.functions[name] + self._sendCantDo(name) + + def setFunctions(self, functions): + """Replace the set of functions registered with Gearman. + + Accepts a list of :py:class:`FunctionRecord` objects which + represents the complete set of functions that should be + registered with Gearman. Any existing functions will be + unregistered and these registered in their place. If the + empty list is supplied, then the Gearman registered function + set will be cleared. + + :arg list functions: A list of :py:class:`FunctionRecord` objects. + """ + + self._sendResetAbilities() + self.functions = {} + for f in functions: + if not isinstance(f, FunctionRecord): + raise InvalidDataError( + "An iterable of FunctionRecords is required.") + self.functions[f.name] = f + for f in self.functions.values(): + if f.timeout: + self._sendCanDoTimeout(f.name, f.timeout) + else: + self._sendCanDo(f.name) + + def _sendCanDo(self, name): + self.broadcast_lock.acquire() + try: + p = Packet(constants.REQ, constants.CAN_DO, name) + self.broadcast(p) + finally: + self.broadcast_lock.release() + + def _sendCanDoTimeout(self, name, timeout): + self.broadcast_lock.acquire() + try: + data = name + b'\x00' + timeout + p = Packet(constants.REQ, constants.CAN_DO_TIMEOUT, data) + self.broadcast(p) + finally: + self.broadcast_lock.release() + + def _sendCantDo(self, name): + self.broadcast_lock.acquire() + try: + p = Packet(constants.REQ, constants.CANT_DO, name) + self.broadcast(p) + finally: + self.broadcast_lock.release() + + def _sendResetAbilities(self): + self.broadcast_lock.acquire() + try: + p = Packet(constants.REQ, constants.RESET_ABILITIES, b'') + self.broadcast(p) + finally: + self.broadcast_lock.release() + + def _sendPreSleep(self, connection): + p = Packet(constants.REQ, constants.PRE_SLEEP, b'') + self.sendPacket(p, connection) + + def _sendGrabJobUniq(self, connection=None): + p = Packet(constants.REQ, constants.GRAB_JOB_UNIQ, b'') + if connection: + self.sendPacket(p, connection) + else: + self.broadcast(p) + + def _onConnect(self, conn): + self.broadcast_lock.acquire() + try: + # Called immediately after a successful (re-)connection + p = Packet(constants.REQ, constants.SET_CLIENT_ID, self.client_id) + conn.sendPacket(p) + super(Worker, self)._onConnect(conn) + for f in self.functions.values(): + if f.timeout: + data = f.name + b'\x00' + f.timeout + p = Packet(constants.REQ, constants.CAN_DO_TIMEOUT, data) + else: + p = Packet(constants.REQ, constants.CAN_DO, f.name) + conn.sendPacket(p) + conn.changeState("IDLE") + finally: + self.broadcast_lock.release() + # Any exceptions will be handled by the calling function, and the + # connection will not be put into the pool. + + def _onActiveConnection(self, conn): + self.job_lock.acquire() + try: + if self.waiting_for_jobs > 0: + self._updateStateMachines() + finally: + self.job_lock.release() + + def _updateStateMachines(self): + connections = self.active_connections[:] + + for connection in connections: + if (connection.state == "IDLE" and self.waiting_for_jobs > 0): + self._sendGrabJobUniq(connection) + connection.changeState("GRAB_WAIT") + if (connection.state != "IDLE" and self.waiting_for_jobs < 1): + connection.changeState("IDLE") + + def getJob(self): + """Get a job from Gearman. + + Blocks until a job is received. This method is re-entrant, so + it is safe to call this method on a single worker from + multiple threads. In that case, one of them at random will + receive the job assignment. + + :returns: The :py:class:`WorkerJob` assigned. + :rtype: :py:class:`WorkerJob`. + :raises InterruptedError: If interrupted (by + :py:meth:`stopWaitingForJobs`) before a job is received. + """ + self.job_lock.acquire() + try: + # self.running gets cleared during _shutdown(), before the + # stopWaitingForJobs() is called. This check has to + # happen with the job_lock held, otherwise there would be + # a window for race conditions between manipulation of + # "running" and "waiting_for_jobs". + if not self.running: + raise InterruptedError() + + self.waiting_for_jobs += 1 + self.log.debug("Get job; number of threads waiting for jobs: %s" % + self.waiting_for_jobs) + + try: + job = self.job_queue.get(False) + except queue_mod.Empty: + job = None + + if not job: + self._updateStateMachines() + + finally: + self.job_lock.release() + + if not job: + job = self.job_queue.get() + + self.log.debug("Received job: %s" % job) + if job is None: + raise InterruptedError() + return job + + def stopWaitingForJobs(self): + """Interrupts all running :py:meth:`getJob` calls, which will raise + an exception. + """ + + self.job_lock.acquire() + try: + while True: + connections = self.active_connections[:] + now = time.time() + ok = True + for connection in connections: + if connection.state == "GRAB_WAIT": + # Replies to GRAB_JOB should be fast, give up if we've + # been waiting for more than 5 seconds. + if now - connection.state_time > 5: + self._lostConnection(connection) + else: + ok = False + if ok: + break + else: + self.job_lock.release() + time.sleep(0.1) + self.job_lock.acquire() + + while self.waiting_for_jobs > 0: + self.waiting_for_jobs -= 1 + self.job_queue.put(None) + + self._updateStateMachines() + finally: + self.job_lock.release() + + def _shutdown(self): + self.job_lock.acquire() + try: + # The upstream _shutdown() will clear the "running" bool. Because + # that is a variable which is used for proper synchronization of + # the exit within getJob() which might be about to be called from a + # separate thread, it's important to call it with a proper lock + # being held. + super(Worker, self)._shutdown() + finally: + self.job_lock.release() + self.stopWaitingForJobs() + + def handleNoop(self, packet): + """Handle a NOOP packet. + + Sends a GRAB_JOB_UNIQ packet on the same connection. + GRAB_JOB_UNIQ will return jobs regardless of whether they have + been specified with a unique identifier when submitted. If + they were not, then :py:attr:`WorkerJob.unique` attribute + will be None. + + :arg Packet packet: The :py:class:`Packet` that was received. + """ + + self.job_lock.acquire() + try: + if packet.connection.state == "SLEEP": + self.log.debug("Sending GRAB_JOB_UNIQ") + self._sendGrabJobUniq(packet.connection) + packet.connection.changeState("GRAB_WAIT") + else: + self.log.debug("Received unexpecetd NOOP packet on %s" % + packet.connection) + finally: + self.job_lock.release() + + def handleNoJob(self, packet): + """Handle a NO_JOB packet. + + Sends a PRE_SLEEP packet on the same connection. + + :arg Packet packet: The :py:class:`Packet` that was received. + """ + self.job_lock.acquire() + try: + if packet.connection.state == "GRAB_WAIT": + self.log.debug("Sending PRE_SLEEP") + self._sendPreSleep(packet.connection) + packet.connection.changeState("SLEEP") + else: + self.log.debug("Received unexpected NO_JOB packet on %s" % + packet.connection) + finally: + self.job_lock.release() + + def handleJobAssign(self, packet): + """Handle a JOB_ASSIGN packet. + + Adds a WorkerJob to the internal queue to be picked up by any + threads waiting in :py:meth:`getJob`. + + :arg Packet packet: The :py:class:`Packet` that was received. + """ + + handle = packet.getArgument(0) + name = packet.getArgument(1) + arguments = packet.getArgument(2, True) + return self._handleJobAssignment(packet, handle, name, + arguments, None) + + def handleJobAssignUnique(self, packet): + """Handle a JOB_ASSIGN_UNIQ packet. + + Adds a WorkerJob to the internal queue to be picked up by any + threads waiting in :py:meth:`getJob`. + + :arg Packet packet: The :py:class:`Packet` that was received. + """ + + handle = packet.getArgument(0) + name = packet.getArgument(1) + unique = packet.getArgument(2) + if unique == b'': + unique = None + arguments = packet.getArgument(3, True) + return self._handleJobAssignment(packet, handle, name, + arguments, unique) + + def _handleJobAssignment(self, packet, handle, name, arguments, unique): + job = self.job_class(handle, name, arguments, unique) + job.connection = packet.connection + + self.job_lock.acquire() + try: + packet.connection.changeState("IDLE") + self.waiting_for_jobs -= 1 + self.log.debug("Job assigned; number of threads waiting for " + "jobs: %s" % self.waiting_for_jobs) + self.job_queue.put(job) + + self._updateStateMachines() + finally: + self.job_lock.release() + + +class Job(BaseJob): + """A job to run or being run by Gearman. + + :arg str name: The name of the job. + :arg bytes arguments: The opaque data blob to be passed to the worker + as arguments. + :arg str unique: A byte string to uniquely identify the job to Gearman + (optional). + + The following instance attributes are available: + + **name** (str) + The name of the job. Assumed to be utf-8. + **arguments** (bytes) + The opaque data blob passed to the worker as arguments. + **unique** (str or None) + The unique ID of the job (if supplied). + **handle** (bytes or None) + The Gearman job handle. None if no job handle has been received yet. + **data** (list of byte-arrays) + The result data returned from Gearman. Each packet appends an + element to the list. Depending on the nature of the data, the + elements may need to be concatenated before use. This is returned + as a snapshot copy of the data to prevent accidental attempts at + modification which will be lost. + **exception** (bytes or None) + Exception information returned from Gearman. None if no exception + has been received. + **warning** (bool) + Whether the worker has reported a warning. + **complete** (bool) + Whether the job is complete. + **failure** (bool) + Whether the job has failed. Only set when complete is True. + **numerator** (bytes or None) + The numerator of the completion ratio reported by the worker. + Only set when a status update is sent by the worker. + **denominator** (bytes or None) + The denominator of the completion ratio reported by the + worker. Only set when a status update is sent by the worker. + **fraction_complete** (float or None) + The fractional complete ratio reported by the worker. Only set when + a status update is sent by the worker. + **known** (bool or None) + Whether the job is known to Gearman. Only set by handleStatusRes() in + response to a getStatus() query. + **running** (bool or None) + Whether the job is running. Only set by handleStatusRes() in + response to a getStatus() query. + **connection** (:py:class:`Connection` or None) + The connection associated with the job. Only set after the job + has been submitted to a Gearman server. + """ + + data_type = list + + def __init__(self, name, arguments, unique=None): + super(Job, self).__init__(name, arguments, unique) + self._data = self.data_type() + self._exception = None + self.warning = False + self.complete = False + self.failure = False + self.numerator = None + self.denominator = None + self.fraction_complete = None + self.known = None + self.running = None + + @property + def binary_data(self): + for value in self._data: + if isinstance(value, six.text_type): + value = value.encode('utf-8') + yield value + + @property + def data(self): + return self._data + + @data.setter + def data(self, value): + if not isinstance(value, self.data_type): + raise ValueError( + "data attribute must be {}".format(self.data_type)) + self._data = value + + @property + def exception(self): + return self._exception + + @exception.setter + def exception(self, value): + self._exception = value + + +class TextJobArguments(object): + """Assumes utf-8 arguments in addition to name + + If one is always dealing in valid utf-8, using this job class relieves one + of the need to encode/decode constantly.""" + + def _validate_arguments(self, arguments): + pass + + @property + def arguments(self): + args = self._arguments + if isinstance(args, six.binary_type): + return args.decode('utf-8') + return args + + @arguments.setter + def arguments(self, value): + if not isinstance(value, six.binary_type): + value = value.encode('utf-8') + self._arguments = value + + +class TextJobUnique(object): + """Assumes utf-8 unique + + If one is always dealing in valid utf-8, using this job class relieves one + of the need to encode/decode constantly.""" + + @property + def unique(self): + unique = self._unique + if isinstance(unique, six.binary_type): + return unique.decode('utf-8') + return unique + + @unique.setter + def unique(self, value): + if not isinstance(value, six.binary_type): + value = value.encode('utf-8') + self._unique = value + + +class TextList(list): + def append(self, x): + if isinstance(x, six.binary_type): + x = x.decode('utf-8') + super(TextList, self).append(x) + + def extend(self, iterable): + def _iter(): + for value in iterable: + if isinstance(value, six.binary_type): + yield value.decode('utf-8') + else: + yield value + super(TextList, self).extend(_iter) + + def insert(self, i, x): + if isinstance(x, six.binary_type): + x = x.decode('utf-8') + super(TextList, self).insert(i, x) + + +class TextJob(TextJobArguments, TextJobUnique, Job): + """ Sends and receives UTF-8 arguments and data. + + Use this instead of Job when you only expect to send valid UTF-8 through + gearman. It will automatically encode arguments and work data as UTF-8, and + any jobs fetched from this worker will have their arguments and data + decoded assuming they are valid UTF-8, and thus return strings. + + Attributes and method signatures are thes ame as Job except as noted here: + + ** arguments ** (str) This will be returned as a string. + ** data ** (tuple of str) This will be returned as a tuble of strings. + + """ + + data_type = TextList + + @property + def exception(self): + exception = self._exception + if isinstance(exception, six.binary_type): + return exception.decode('utf-8') + return exception + + @exception.setter + def exception(self, value): + if not isinstance(value, six.binary_type): + value = value.encode('utf-8') + self._exception = value + + +class TextWorkerJob(TextJobArguments, TextJobUnique, WorkerJob): + """ Sends and receives UTF-8 arguments and data. + + See TextJob. sendWorkData and sendWorkWarning accept strings + and will encode them as UTF-8. + """ + def sendWorkData(self, data=''): + """Send a WORK_DATA packet to the client. + + :arg str data: The data to be sent to the client (optional). + """ + if isinstance(data, six.text_type): + data = data.encode('utf8') + return super(TextWorkerJob, self).sendWorkData(data) + + def sendWorkWarning(self, data=''): + """Send a WORK_WARNING packet to the client. + + :arg str data: The data to be sent to the client (optional). + """ + + if isinstance(data, six.text_type): + data = data.encode('utf8') + return super(TextWorkerJob, self).sendWorkWarning(data) + + def sendWorkComplete(self, data=''): + """Send a WORK_COMPLETE packet to the client. + + :arg str data: The data to be sent to the client (optional). + """ + if isinstance(data, six.text_type): + data = data.encode('utf8') + return super(TextWorkerJob, self).sendWorkComplete(data) + + def sendWorkException(self, data=''): + """Send a WORK_EXCEPTION packet to the client. + + :arg str data: The data to be sent to the client (optional). + """ + + if isinstance(data, six.text_type): + data = data.encode('utf8') + return super(TextWorkerJob, self).sendWorkException(data) + + +class TextWorker(Worker): + """ Sends and receives UTF-8 only. + + See TextJob. + + """ + + job_class = TextWorkerJob + + +class BaseBinaryJob(object): + """ For the case where non-utf-8 job names are needed. It will function + exactly like Job, except that the job name will not be decoded.""" + + @property + def name(self): + return self._name + + +class BinaryWorkerJob(BaseBinaryJob, WorkerJob): + pass + + +class BinaryJob(BaseBinaryJob, Job): + pass + + +# Below are classes for use in the server implementation: +class ServerJob(BinaryJob): + """A job record for use in a server. + + :arg str name: The name of the job. + :arg bytes arguments: The opaque data blob to be passed to the worker + as arguments. + :arg str unique: A byte string to uniquely identify the job to Gearman + (optional). + + The following instance attributes are available: + + **name** (str) + The name of the job. + **arguments** (bytes) + The opaque data blob passed to the worker as arguments. + **unique** (str or None) + The unique ID of the job (if supplied). + **handle** (bytes or None) + The Gearman job handle. None if no job handle has been received yet. + **data** (list of byte-arrays) + The result data returned from Gearman. Each packet appends an + element to the list. Depending on the nature of the data, the + elements may need to be concatenated before use. + **exception** (bytes or None) + Exception information returned from Gearman. None if no exception + has been received. + **warning** (bool) + Whether the worker has reported a warning. + **complete** (bool) + Whether the job is complete. + **failure** (bool) + Whether the job has failed. Only set when complete is True. + **numerator** (bytes or None) + The numerator of the completion ratio reported by the worker. + Only set when a status update is sent by the worker. + **denominator** (bytes or None) + The denominator of the completion ratio reported by the + worker. Only set when a status update is sent by the worker. + **fraction_complete** (float or None) + The fractional complete ratio reported by the worker. Only set when + a status update is sent by the worker. + **known** (bool or None) + Whether the job is known to Gearman. Only set by handleStatusRes() in + response to a getStatus() query. + **running** (bool or None) + Whether the job is running. Only set by handleStatusRes() in + response to a getStatus() query. + **client_connection** :py:class:`Connection` + The client connection associated with the job. + **worker_connection** (:py:class:`Connection` or None) + The worker connection associated with the job. Only set after the job + has been assigned to a worker. + """ + + def __init__(self, handle, name, arguments, client_connection, + unique=None): + super(ServerJob, self).__init__(name, arguments, unique) + self.handle = handle + self.client_connection = client_connection + self.worker_connection = None + del self.connection + + +class ServerAdminRequest(AdminRequest): + """An administrative request sent to a server.""" + + def __init__(self, connection): + super(ServerAdminRequest, self).__init__() + self.connection = connection + + def isComplete(self, data): + end_index_newline = data.find(b'\n') + if end_index_newline != -1: + self.command = data[:end_index_newline] + # Remove newline from data + x = end_index_newline + 1 + return (True, data[x:]) + else: + return (False, None) + + +class NonBlockingConnection(Connection): + """A Non-blocking connection to a Gearman Client.""" + + def __init__(self, host, port, ssl_key=None, ssl_cert=None, + ssl_ca=None, client_id='unknown'): + super(NonBlockingConnection, self).__init__( + host, port, ssl_key, + ssl_cert, ssl_ca, client_id) + self.send_queue = [] + + def connect(self): + super(NonBlockingConnection, self).connect() + if self.connected and self.conn: + self.conn.setblocking(0) + + def _readRawBytes(self, bytes_to_read): + try: + buff = self.conn.recv(bytes_to_read) + except ssl.SSLError as e: + if e.errno == ssl.SSL_ERROR_WANT_READ: + raise RetryIOError() + elif e.errno == ssl.SSL_ERROR_WANT_WRITE: + raise RetryIOError() + raise + except socket.error as e: + if e.errno == errno.EAGAIN: + # Read operation would block, we're done until + # epoll flags this connection again + raise RetryIOError() + raise + return buff + + def sendPacket(self, packet): + """Append a packet to this connection's send queue. The Client or + Server must manage actually sending the data. + + :arg :py:class:`Packet` packet The packet to send + + """ + self.log.debug("Queuing packet to %s: %s" % (self, packet)) + self.send_queue.append(packet.toBinary()) + self.sendQueuedData() + + def sendRaw(self, data): + """Append raw data to this connection's send queue. The Client or + Server must manage actually sending the data. + + :arg bytes data The raw data to send + + """ + self.log.debug("Queuing data to %s: %s" % (self, data)) + self.send_queue.append(data) + self.sendQueuedData() + + def sendQueuedData(self): + """Send previously queued data to the socket.""" + try: + while len(self.send_queue): + data = self.send_queue.pop(0) + r = 0 + try: + r = self.conn.send(data) + except ssl.SSLError as e: + if e.errno == ssl.SSL_ERROR_WANT_READ: + raise RetryIOError() + elif e.errno == ssl.SSL_ERROR_WANT_WRITE: + raise RetryIOError() + else: + raise + except socket.error as e: + if e.errno == errno.EAGAIN: + self.log.debug("Write operation on %s would block" + % self) + raise RetryIOError() + else: + raise + finally: + data = data[r:] + if data: + self.send_queue.insert(0, data) + except RetryIOError: + pass + + +class ServerConnection(NonBlockingConnection): + """A Connection to a Gearman Client.""" + + def __init__(self, addr, conn, use_ssl, client_id): + if client_id: + self.log = logging.getLogger("gear.ServerConnection.%s" % + (client_id,)) + else: + self.log = logging.getLogger("gear.ServerConnection") + self.send_queue = [] + self.admin_requests = [] + self.host = addr[0] + self.port = addr[1] + self.conn = conn + self.conn.setblocking(0) + self.input_buffer = b'' + self.need_bytes = False + self.use_ssl = use_ssl + self.client_id = None + self.functions = set() + self.related_jobs = {} + self.ssl_subject = None + if self.use_ssl: + for x in conn.getpeercert()['subject']: + if x[0][0] == 'commonName': + self.ssl_subject = x[0][1] + self.log.debug("SSL subject: %s" % self.ssl_subject) + self.changeState("INIT") + + def _getAdminRequest(self): + return ServerAdminRequest(self) + + def _putAdminRequest(self, req): + # The server does not need to keep track of admin requests + # that have been partially received; it will simply create a + # new instance the next time it tries to read. + pass + + def __repr__(self): + return '' % ( + id(self), self.client_id, self.host, self.port) + + +class Server(BaseClientServer): + """A simple gearman server implementation for testing + (not for production use). + + :arg int port: The TCP port on which to listen. + :arg str ssl_key: Path to the SSL private key. + :arg str ssl_cert: Path to the SSL certificate. + :arg str ssl_ca: Path to the CA certificate. + :arg str statsd_host: statsd hostname. None means disabled + (the default). + :arg str statsd_port: statsd port (defaults to 8125). + :arg str statsd_prefix: statsd key prefix. + :arg str client_id: The ID associated with this server. + It will be appending to the name of the logger (e.g., + gear.Server.server_id). Defaults to None (unused). + :arg ACL acl: An :py:class:`ACL` object if the server should apply + access control rules to its connections. + :arg str host: Host name or IPv4/IPv6 address to bind to. Defaults + to "whatever getaddrinfo() returns", which might be IPv4-only. + :arg bool keepalive: Whether to use TCP keepalives + :arg int tcp_keepidle: Idle time after which to start keepalives sending + :arg int tcp_keepintvl: Interval in seconds between TCP keepalives + :arg int tcp_keepcnt: Count of TCP keepalives to send before disconnect + """ + + edge_bitmask = select.EPOLLET + error_bitmask = (select.EPOLLERR | select.EPOLLHUP | edge_bitmask) + read_bitmask = (select.EPOLLIN | error_bitmask) + readwrite_bitmask = (select.EPOLLOUT | read_bitmask) + + def __init__(self, port=4730, ssl_key=None, ssl_cert=None, ssl_ca=None, + statsd_host=None, statsd_port=8125, statsd_prefix=None, + server_id=None, acl=None, host=None, keepalive=False, + tcp_keepidle=7200, tcp_keepintvl=75, tcp_keepcnt=9): + self.port = port + self.ssl_key = ssl_key + self.ssl_cert = ssl_cert + self.ssl_ca = ssl_ca + self.high_queue = [] + self.normal_queue = [] + self.low_queue = [] + self.jobs = {} + self.running_jobs = 0 + self.waiting_jobs = 0 + self.total_jobs = 0 + self.functions = set() + self.max_handle = 0 + self.acl = acl + self.connect_wake_read, self.connect_wake_write = os.pipe() + self.poll = select.epoll() + # Reverse mapping of fd -> connection + self.connection_map = {} + + self.use_ssl = False + if all([self.ssl_key, self.ssl_cert, self.ssl_ca]): + self.use_ssl = True + + # Get all valid passive listen addresses, then sort by family to prefer + # ipv6 if available. + addrs = socket.getaddrinfo(host, self.port, socket.AF_UNSPEC, + socket.SOCK_STREAM, 0, + socket.AI_PASSIVE | + socket.AI_ADDRCONFIG) + addrs.sort(key=lambda addr: addr[0], reverse=True) + for res in addrs: + af, socktype, proto, canonname, sa = res + try: + self.socket = socket.socket(af, socktype, proto) + self.socket.setsockopt(socket.SOL_SOCKET, + socket.SO_REUSEADDR, 1) + if keepalive and hasattr(socket, 'TCP_KEEPIDLE'): + self.socket.setsockopt(socket.SOL_SOCKET, + socket.SO_KEEPALIVE, 1) + self.socket.setsockopt(socket.IPPROTO_TCP, + socket.TCP_KEEPIDLE, tcp_keepidle) + self.socket.setsockopt(socket.IPPROTO_TCP, + socket.TCP_KEEPINTVL, tcp_keepintvl) + self.socket.setsockopt(socket.IPPROTO_TCP, + socket.TCP_KEEPCNT, tcp_keepcnt) + elif keepalive: + self.log.warning('Keepalive requested but not available ' + 'on this platform') + except socket.error: + self.socket = None + continue + try: + self.socket.bind(sa) + self.socket.listen(1) + except socket.error: + self.socket.close() + self.socket = None + continue + break + + if self.socket is None: + raise Exception("Could not open socket") + + if port == 0: + self.port = self.socket.getsockname()[1] + + super(Server, self).__init__(server_id) + + # Register the wake pipe so that we can break if we need to + # reconfigure connections + self.poll.register(self.wake_read, self.read_bitmask) + + if server_id: + self.log = logging.getLogger("gear.Server.%s" % (self.client_id,)) + else: + self.log = logging.getLogger("gear.Server") + + if statsd_host: + if not statsd: + self.log.error("Unable to import statsd module") + self.statsd = None + else: + self.statsd = statsd.StatsClient(statsd_host, + statsd_port, + statsd_prefix) + else: + self.statsd = None + + def _doConnectLoop(self): + while self.running: + try: + self.connectLoop() + except Exception: + self.log.exception("Exception in connect loop:") + time.sleep(1) + + def connectLoop(self): + poll = select.poll() + bitmask = (select.POLLIN | select.POLLERR | + select.POLLHUP | select.POLLNVAL) + # Register the wake pipe so that we can break if we need to + # shutdown. + poll.register(self.connect_wake_read, bitmask) + poll.register(self.socket.fileno(), bitmask) + while self.running: + ret = poll.poll() + for fd, event in ret: + if fd == self.connect_wake_read: + self.log.debug("Accept woken by pipe") + while True: + if os.read(self.connect_wake_read, 1) == b'\n': + break + return + if event & select.POLLIN: + self.log.debug("Accepting new connection") + c, addr = self.socket.accept() + if self.use_ssl: + context = ssl.SSLContext(ssl.PROTOCOL_TLSv1) + context.verify_mode = ssl.CERT_REQUIRED + context.load_cert_chain(self.ssl_cert, self.ssl_key) + context.load_verify_locations(self.ssl_ca) + c = context.wrap_socket(c, server_side=True) + conn = ServerConnection(addr, c, self.use_ssl, + self.client_id) + self.log.info("Accepted connection %s" % (conn,)) + self.connections_condition.acquire() + try: + self.active_connections.append(conn) + self._registerConnection(conn) + self.connections_condition.notifyAll() + finally: + self.connections_condition.release() + + def readFromConnection(self, conn): + while True: + self.log.debug("Processing input on %s" % conn) + try: + p = conn.readPacket() + except RetryIOError: + # Read operation would block, we're done until + # epoll flags this connection again + return + if p: + if isinstance(p, Packet): + self.handlePacket(p) + else: + self.handleAdminRequest(p) + else: + self.log.debug("Received no data on %s" % conn) + raise DisconnectError() + + def writeToConnection(self, conn): + self.log.debug("Processing output on %s" % conn) + conn.sendQueuedData() + + def _processPollEvent(self, conn, event): + # This should do whatever is necessary to process a connection + # that has triggered a poll event. It should generally not + # raise exceptions so as to avoid restarting the poll loop. + # The exception handlers here can raise exceptions and if they + # do, it's okay, the poll loop will be restarted. + try: + if event & (select.EPOLLERR | select.EPOLLHUP): + self.log.debug("Received error event on %s: %s" % ( + conn, event)) + raise DisconnectError() + if event & (select.POLLIN | select.POLLOUT): + self.readFromConnection(conn) + self.writeToConnection(conn) + except socket.error as e: + if e.errno == errno.ECONNRESET: + self.log.debug("Connection reset by peer: %s" % (conn,)) + self._lostConnection(conn) + return + raise + except DisconnectError: + # Our inner method says we should quietly drop + # this connection + self._lostConnection(conn) + return + except Exception: + self.log.exception("Exception reading or writing " + "from %s:" % (conn,)) + self._lostConnection(conn) + return + + def _flushAllConnections(self): + # If we need to restart the poll loop, we need to make sure + # there are no pending data on any connection. Simulate poll + # in+out events on every connection. + # + # If this method raises an exception, the poll loop wil + # restart again. + # + # No need to get the lock since this is called within the poll + # loop and therefore the list in guaranteed never to shrink. + connections = self.active_connections[:] + for conn in connections: + self._processPollEvent(conn, select.POLLIN | select.POLLOUT) + + def _doPollLoop(self): + # Outer run method of poll thread. + while self.running: + try: + self._pollLoop() + except Exception: + self.log.exception("Exception in poll loop:") + + def _pollLoop(self): + # Inner method of poll loop. + self.log.debug("Preparing to poll") + # Ensure there are no pending data. + self._flushAllConnections() + while self.running: + self.log.debug("Polling %s connections" % + len(self.active_connections)) + ret = self.poll.poll() + # Since we're using edge-triggering, we need to make sure + # that every file descriptor in 'ret' is processed. + for fd, event in ret: + if fd == self.wake_read: + # This means we're exiting, so we can ignore the + # rest of 'ret'. + self.log.debug("Woken by pipe") + while True: + if os.read(self.wake_read, 1) == b'\n': + break + return + # In the unlikely event this raises an exception, the + # loop will be restarted. + conn = self.connection_map[fd] + self._processPollEvent(conn, event) + + def _shutdown(self): + super(Server, self)._shutdown() + os.write(self.connect_wake_write, b'1\n') + + def _cleanup(self): + super(Server, self)._cleanup() + self.socket.close() + os.close(self.connect_wake_read) + os.close(self.connect_wake_write) + + def _registerConnection(self, conn): + # Register the connection with the poll object + # Call while holding the connection condition + self.log.debug("Registering %s" % conn) + self.connection_map[conn.conn.fileno()] = conn + self.poll.register(conn.conn.fileno(), self.readwrite_bitmask) + + def _unregisterConnection(self, conn): + # Unregister the connection with the poll object + # Call while holding the connection condition + self.log.debug("Unregistering %s" % conn) + fd = conn.conn.fileno() + if fd not in self.connection_map: + return + try: + self.poll.unregister(fd) + except KeyError: + pass + try: + del self.connection_map[fd] + except KeyError: + pass + + def _lostConnection(self, conn): + # Called as soon as a connection is detected as faulty. + self.log.info("Marking %s as disconnected" % conn) + self.connections_condition.acquire() + self._unregisterConnection(conn) + try: + # NOTE(notmorgan): In the loop below it is possible to change the + # jobs list on the connection. In python 3 .values() is an iter not + # a static list, meaning that a change will break the for loop + # as the object being iterated on will have changed in size. + jobs = list(conn.related_jobs.values()) + if conn in self.active_connections: + self.active_connections.remove(conn) + finally: + self.connections_condition.notifyAll() + self.connections_condition.release() + for job in jobs: + if job.worker_connection == conn: + # the worker disconnected, alert the client + try: + p = Packet(constants.REQ, constants.WORK_FAIL, job.handle) + if job.client_connection: + job.client_connection.sendPacket(p) + except Exception: + self.log.exception("Sending WORK_FAIL to client after " + "worker disconnect failed:") + self._removeJob(job) + try: + conn.conn.shutdown(socket.SHUT_RDWR) + except socket.error as e: + if e.errno != errno.ENOTCONN: + self.log.exception("Unable to shutdown socket " + "for connection %s" % (conn,)) + except Exception: + self.log.exception("Unable to shutdown socket " + "for connection %s" % (conn,)) + try: + conn.conn.close() + except Exception: + self.log.exception("Unable to close socket " + "for connection %s" % (conn,)) + self._updateStats() + + def _removeJob(self, job, dequeue=True): + # dequeue is tri-state: True, False, or a specific queue + if job.client_connection: + try: + del job.client_connection.related_jobs[job.handle] + except KeyError: + pass + if job.worker_connection: + try: + del job.worker_connection.related_jobs[job.handle] + except KeyError: + pass + try: + del self.jobs[job.handle] + except KeyError: + pass + if dequeue is True: + # Search all queues for the job + try: + self.high_queue.remove(job) + except ValueError: + pass + try: + self.normal_queue.remove(job) + except ValueError: + pass + try: + self.low_queue.remove(job) + except ValueError: + pass + elif dequeue is not False: + # A specific queue was supplied + dequeue.remove(job) + # If dequeue is false, no need to remove from any queue + self.total_jobs -= 1 + if job.running: + self.running_jobs -= 1 + else: + self.waiting_jobs -= 1 + + def getQueue(self): + """Returns a copy of all internal queues in a flattened form. + + :returns: The Gearman queue. + :rtype: list of :py:class:`WorkerJob`. + """ + ret = [] + for queue in [self.high_queue, self.normal_queue, self.low_queue]: + ret += queue + return ret + + def handleAdminRequest(self, request): + self.log.info("Received admin request %s" % (request,)) + + if request.command.startswith(b'cancel job'): + self.handleCancelJob(request) + elif request.command.startswith(b'status'): + self.handleStatus(request) + elif request.command.startswith(b'workers'): + self.handleWorkers(request) + elif request.command.startswith(b'acl list'): + self.handleACLList(request) + elif request.command.startswith(b'acl grant'): + self.handleACLGrant(request) + elif request.command.startswith(b'acl revoke'): + self.handleACLRevoke(request) + elif request.command.startswith(b'acl self-revoke'): + self.handleACLSelfRevoke(request) + + self.log.debug("Finished handling admin request %s" % (request,)) + + def _cancelJob(self, request, job, queue): + if self.acl: + if not self.acl.canInvoke(request.connection.ssl_subject, + job.name): + self.log.info("Rejecting cancel job from %s for %s " + "due to ACL" % + (request.connection.ssl_subject, job.name)) + request.connection.sendRaw(b'ERR PERMISSION_DENIED\n') + return + self._removeJob(job, dequeue=queue) + self._updateStats() + request.connection.sendRaw(b'OK\n') + return + + def handleCancelJob(self, request): + words = request.command.split() + handle = words[2] + + if handle in self.jobs: + for queue in [self.high_queue, self.normal_queue, self.low_queue]: + for job in queue: + if handle == job.handle: + return self._cancelJob(request, job, queue) + request.connection.sendRaw(b'ERR UNKNOWN_JOB\n') + + def handleACLList(self, request): + if self.acl is None: + request.connection.sendRaw(b'ERR ACL_DISABLED\n') + return + for entry in self.acl.getEntries(): + acl = "%s\tregister=%s\tinvoke=%s\tgrant=%s\n" % ( + entry.subject, entry.register, entry.invoke, entry.grant) + request.connection.sendRaw(acl.encode('utf8')) + request.connection.sendRaw(b'.\n') + + def handleACLGrant(self, request): + # acl grant register worker .* + words = request.command.split(None, 4) + verb = words[2] + subject = words[3] + + if self.acl is None: + request.connection.sendRaw(b'ERR ACL_DISABLED\n') + return + if not self.acl.canGrant(request.connection.ssl_subject): + request.connection.sendRaw(b'ERR PERMISSION_DENIED\n') + return + try: + if verb == 'invoke': + self.acl.grantInvoke(subject, words[4]) + elif verb == 'register': + self.acl.grantRegister(subject, words[4]) + elif verb == 'grant': + self.acl.grantGrant(subject) + else: + request.connection.sendRaw(b'ERR UNKNOWN_ACL_VERB\n') + return + except ACLError as e: + self.log.info("Error in grant command: %s" % (e.message,)) + request.connection.sendRaw(b'ERR UNABLE %s\n' % (e.message,)) + return + request.connection.sendRaw(b'OK\n') + + def handleACLRevoke(self, request): + # acl revoke register worker + words = request.command.split() + verb = words[2] + subject = words[3] + + if self.acl is None: + request.connection.sendRaw(b'ERR ACL_DISABLED\n') + return + if subject != request.connection.ssl_subject: + if not self.acl.canGrant(request.connection.ssl_subject): + request.connection.sendRaw(b'ERR PERMISSION_DENIED\n') + return + try: + if verb == 'invoke': + self.acl.revokeInvoke(subject) + elif verb == 'register': + self.acl.revokeRegister(subject) + elif verb == 'grant': + self.acl.revokeGrant(subject) + elif verb == 'all': + try: + self.acl.remove(subject) + except ACLError: + pass + else: + request.connection.sendRaw(b'ERR UNKNOWN_ACL_VERB\n') + return + except ACLError as e: + self.log.info("Error in revoke command: %s" % (e.message,)) + request.connection.sendRaw(b'ERR UNABLE %s\n' % (e.message,)) + return + request.connection.sendRaw(b'OK\n') + + def handleACLSelfRevoke(self, request): + # acl self-revoke register + words = request.command.split() + verb = words[2] + + if self.acl is None: + request.connection.sendRaw(b'ERR ACL_DISABLED\n') + return + subject = request.connection.ssl_subject + try: + if verb == 'invoke': + self.acl.revokeInvoke(subject) + elif verb == 'register': + self.acl.revokeRegister(subject) + elif verb == 'grant': + self.acl.revokeGrant(subject) + elif verb == 'all': + try: + self.acl.remove(subject) + except ACLError: + pass + else: + request.connection.sendRaw(b'ERR UNKNOWN_ACL_VERB\n') + return + except ACLError as e: + self.log.info("Error in self-revoke command: %s" % (e.message,)) + request.connection.sendRaw(b'ERR UNABLE %s\n' % (e.message,)) + return + request.connection.sendRaw(b'OK\n') + + def _getFunctionStats(self): + functions = {} + for function in self.functions: + # Total, running, workers + functions[function] = [0, 0, 0] + for job in self.jobs.values(): + if job.name not in functions: + functions[job.name] = [0, 0, 0] + functions[job.name][0] += 1 + if job.running: + functions[job.name][1] += 1 + for connection in self.active_connections: + for function in connection.functions: + if function not in functions: + functions[function] = [0, 0, 0] + functions[function][2] += 1 + return functions + + def handleStatus(self, request): + functions = self._getFunctionStats() + for name, values in functions.items(): + request.connection.sendRaw( + ("%s\t%s\t%s\t%s\n" % + (name.decode('utf-8'), values[0], values[1], + values[2])).encode('utf8')) + request.connection.sendRaw(b'.\n') + + def handleWorkers(self, request): + for connection in self.active_connections: + fd = connection.conn.fileno() + ip = connection.host + client_id = connection.client_id or b'-' + functions = b' '.join(connection.functions).decode('utf8') + request.connection.sendRaw(("%s %s %s : %s\n" % + (fd, ip, client_id.decode('utf8'), + functions)) + .encode('utf8')) + request.connection.sendRaw(b'.\n') + + def wakeConnection(self, connection): + p = Packet(constants.RES, constants.NOOP, b'') + if connection.state == 'SLEEP': + connection.changeState("AWAKE") + connection.sendPacket(p) + + def wakeConnections(self, job=None): + p = Packet(constants.RES, constants.NOOP, b'') + for connection in self.active_connections: + if connection.state == 'SLEEP': + if ((job and job.name in connection.functions) or + (job is None)): + connection.changeState("AWAKE") + connection.sendPacket(p) + + def reportTimingStats(self, ptype, duration): + """Report processing times by packet type + + This method is called by handlePacket to report how long + processing took for each packet. If statsd is configured, + timing and counts are reported with the key + "prefix.packet.NAME". + + :arg bytes ptype: The packet type (one of the packet types in + constants). + :arg float duration: The time (in seconds) it took to process + the packet. + """ + if not self.statsd: + return + ptype = constants.types.get(ptype, 'UNKNOWN') + key = 'packet.%s' % ptype + self.statsd.timing(key, int(duration * 1000)) + self.statsd.incr(key) + + def _updateStats(self): + if not self.statsd: + return + + # prefix.queue.total + # prefix.queue.running + # prefix.queue.waiting + self.statsd.gauge('queue.total', self.total_jobs) + self.statsd.gauge('queue.running', self.running_jobs) + self.statsd.gauge('queue.waiting', self.waiting_jobs) + + def _handleSubmitJob(self, packet, precedence, background=False): + name = packet.getArgument(0) + unique = packet.getArgument(1) + if not unique: + unique = None + arguments = packet.getArgument(2, True) + if self.acl: + if not self.acl.canInvoke(packet.connection.ssl_subject, name): + self.log.info("Rejecting SUBMIT_JOB from %s for %s " + "due to ACL" % + (packet.connection.ssl_subject, name)) + self.sendError(packet.connection, 0, + 'Permission denied by ACL') + return + self.max_handle += 1 + handle = ('H:%s:%s' % (packet.connection.host, + self.max_handle)).encode('utf8') + if not background: + conn = packet.connection + else: + conn = None + job = ServerJob(handle, name, arguments, conn, unique) + p = Packet(constants.RES, constants.JOB_CREATED, handle) + packet.connection.sendPacket(p) + self.jobs[handle] = job + self.total_jobs += 1 + self.waiting_jobs += 1 + if not background: + packet.connection.related_jobs[handle] = job + if precedence == PRECEDENCE_HIGH: + self.high_queue.append(job) + elif precedence == PRECEDENCE_NORMAL: + self.normal_queue.append(job) + elif precedence == PRECEDENCE_LOW: + self.low_queue.append(job) + self._updateStats() + self.wakeConnections(job) + + def handleSubmitJob(self, packet): + return self._handleSubmitJob(packet, PRECEDENCE_NORMAL) + + def handleSubmitJobHigh(self, packet): + return self._handleSubmitJob(packet, PRECEDENCE_HIGH) + + def handleSubmitJobLow(self, packet): + return self._handleSubmitJob(packet, PRECEDENCE_LOW) + + def handleSubmitJobBg(self, packet): + return self._handleSubmitJob(packet, PRECEDENCE_NORMAL, + background=True) + + def handleSubmitJobHighBg(self, packet): + return self._handleSubmitJob(packet, PRECEDENCE_HIGH, background=True) + + def handleSubmitJobLowBg(self, packet): + return self._handleSubmitJob(packet, PRECEDENCE_LOW, background=True) + + def getJobForConnection(self, connection, peek=False): + for queue in [self.high_queue, self.normal_queue, self.low_queue]: + for job in queue: + if job.name in connection.functions: + if not peek: + queue.remove(job) + connection.related_jobs[job.handle] = job + job.worker_connection = connection + job.running = True + self.waiting_jobs -= 1 + self.running_jobs += 1 + self._updateStats() + return job + return None + + def handleGrabJobUniq(self, packet): + job = self.getJobForConnection(packet.connection) + if job: + self.sendJobAssignUniq(packet.connection, job) + else: + self.sendNoJob(packet.connection) + + def sendJobAssignUniq(self, connection, job): + unique = job.binary_unique + if not unique: + unique = b'' + data = b'\x00'.join((job.handle, job.name, unique, job.arguments)) + p = Packet(constants.RES, constants.JOB_ASSIGN_UNIQ, data) + connection.sendPacket(p) + + def sendNoJob(self, connection): + p = Packet(constants.RES, constants.NO_JOB, b'') + connection.sendPacket(p) + + def handlePreSleep(self, packet): + packet.connection.changeState("SLEEP") + if self.getJobForConnection(packet.connection, peek=True): + self.wakeConnection(packet.connection) + + def handleWorkComplete(self, packet): + self.handlePassthrough(packet, True) + + def handleWorkFail(self, packet): + self.handlePassthrough(packet, True) + + def handleWorkException(self, packet): + self.handlePassthrough(packet, True) + + def handleWorkData(self, packet): + self.handlePassthrough(packet) + + def handleWorkWarning(self, packet): + self.handlePassthrough(packet) + + def handleWorkStatus(self, packet): + handle = packet.getArgument(0) + job = self.jobs.get(handle) + if not job: + self.log.info("Received packet %s for unknown job" % (packet,)) + return + job.numerator = packet.getArgument(1) + job.denominator = packet.getArgument(2) + self.handlePassthrough(packet) + + def handlePassthrough(self, packet, finished=False): + handle = packet.getArgument(0) + job = self.jobs.get(handle) + if not job: + self.log.info("Received packet %s for unknown job" % (packet,)) + return + packet.code = constants.RES + if job.client_connection: + job.client_connection.sendPacket(packet) + if finished: + self._removeJob(job, dequeue=False) + self._updateStats() + + def handleSetClientID(self, packet): + name = packet.getArgument(0) + packet.connection.client_id = name + + def sendError(self, connection, code, text): + data = (str(code).encode('utf8') + b'\x00' + + str(text).encode('utf8') + b'\x00') + p = Packet(constants.RES, constants.ERROR, data) + connection.sendPacket(p) + + def handleCanDo(self, packet): + name = packet.getArgument(0) + if self.acl: + if not self.acl.canRegister(packet.connection.ssl_subject, name): + self.log.info("Ignoring CAN_DO from %s for %s due to ACL" % + (packet.connection.ssl_subject, name)) + # CAN_DO normally does not merit a response so it is + # not clear that it is appropriate to send an ERROR + # response at this point. + return + self.log.debug("Adding function %s to %s" % (name, packet.connection)) + packet.connection.functions.add(name) + self.functions.add(name) + + def handleCantDo(self, packet): + name = packet.getArgument(0) + self.log.debug("Removing function %s from %s" % + (name, packet.connection)) + packet.connection.functions.remove(name) + + def handleResetAbilities(self, packet): + self.log.debug("Resetting functions for %s" % packet.connection) + packet.connection.functions = set() + + def handleGetStatus(self, packet): + handle = packet.getArgument(0) + self.log.debug("Getting status for %s" % handle) + + known = 0 + running = 0 + numerator = b'' + denominator = b'' + job = self.jobs.get(handle) + if job: + known = 1 + if job.running: + running = 1 + numerator = job.numerator or b'' + denominator = job.denominator or b'' + + data = (handle + b'\x00' + + str(known).encode('utf8') + b'\x00' + + str(running).encode('utf8') + b'\x00' + + numerator + b'\x00' + + denominator) + p = Packet(constants.RES, constants.STATUS_RES, data) + packet.connection.sendPacket(p) diff --git a/roles/submit-log-processor-jobs/module_utils/gear_acl.py b/roles/submit-log-processor-jobs/module_utils/gear_acl.py new file mode 100644 index 0000000..07c9e10 --- /dev/null +++ b/roles/submit-log-processor-jobs/module_utils/gear_acl.py @@ -0,0 +1,289 @@ +# Copyright 2014 OpenStack Foundation +# +# 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 re + + +class ACLError(Exception): + pass + + +class ACLEntry(object): + """An access control list entry. + + :arg str subject: The SSL certificate Subject Common Name to which + the entry applies. + + :arg str register: A regular expression that matches the jobs that + connections with this certificate are permitted to register. + + :arg str invoke: A regular expression that matches the jobs that + connections with this certificate are permitted to invoke. + Also implies the permission to cancel the same set of jobs in + the queue. + + :arg boolean grant: A flag indicating whether connections with + this certificate are permitted to grant access to other + connections. Also implies the permission to revoke access + from other connections. The ability to self-revoke access is + always implied. + """ + + def __init__(self, subject, register=None, invoke=None, grant=False): + self.subject = subject + self.setRegister(register) + self.setInvoke(invoke) + self.setGrant(grant) + + def __repr__(self): + return ('' % + (self.subject, self.register, self.invoke, self.grant)) + + def isEmpty(self): + """Checks whether this entry grants any permissions at all. + + :returns: False if any permission is granted, otherwise True. + """ + if (self.register is None and + self.invoke is None and + self.grant is False): + return True + return False + + def canRegister(self, name): + """Check whether this subject is permitted to register a function. + + :arg str name: The function name to check. + :returns: A boolean indicating whether the action should be permitted. + """ + if self.register is None: + return False + if not self._register.match(name): + return False + return True + + def canInvoke(self, name): + """Check whether this subject is permitted to register a function. + + :arg str name: The function name to check. + :returns: A boolean indicating whether the action should be permitted. + """ + if self.invoke is None: + return False + if not self._invoke.match(name): + return False + return True + + def setRegister(self, register): + """Sets the functions that this subject can register. + + :arg str register: A regular expression that matches the jobs that + connections with this certificate are permitted to register. + """ + self.register = register + if register: + try: + self._register = re.compile(register) + except re.error as e: + raise ACLError('Regular expression error: %s' % (e.message,)) + else: + self._register = None + + def setInvoke(self, invoke): + """Sets the functions that this subject can invoke. + + :arg str invoke: A regular expression that matches the jobs that + connections with this certificate are permitted to invoke. + """ + self.invoke = invoke + if invoke: + try: + self._invoke = re.compile(invoke) + except re.error as e: + raise ACLError('Regular expression error: %s' % (e.message,)) + else: + self._invoke = None + + def setGrant(self, grant): + """Sets whether this subject can grant ACLs to others. + + :arg boolean grant: A flag indicating whether connections with + this certificate are permitted to grant access to other + connections. Also implies the permission to revoke access + from other connections. The ability to self-revoke access is + always implied. + """ + self.grant = grant + + +class ACL(object): + """An access control list. + + ACLs are deny-by-default. The checked actions are only allowed if + there is an explicit rule in the ACL granting permission for a + given client (identified by SSL certificate Common Name Subject) + to perform that action. + """ + + def __init__(self): + self.subjects = {} + + def add(self, entry): + """Add an ACL entry. + + :arg Entry entry: The :py:class:`ACLEntry` to add. + :raises ACLError: If there is already an entry for the subject. + """ + if entry.subject in self.subjects: + raise ACLError("An ACL entry for %s already exists" % + (entry.subject,)) + self.subjects[entry.subject] = entry + + def remove(self, subject): + """Remove an ACL entry. + + :arg str subject: The SSL certificate Subject Common Name to + remove from the ACL. + :raises ACLError: If there is no entry for the subject. + """ + if subject not in self.subjects: + raise ACLError("There is no ACL entry for %s" % (subject,)) + del self.subjects[subject] + + def getEntries(self): + """Return a list of current ACL entries. + + :returns: A list of :py:class:`ACLEntry` objects. + """ + items = list(self.subjects.items()) + items.sort(key=lambda a: a[0]) + return [x[1] for x in items] + + def canRegister(self, subject, name): + """Check whether a subject is permitted to register a function. + + :arg str subject: The SSL certificate Subject Common Name to + check against. + :arg str name: The function name to check. + :returns: A boolean indicating whether the action should be permitted. + """ + entry = self.subjects.get(subject) + if entry is None: + return False + return entry.canRegister(name) + + def canInvoke(self, subject, name): + """Check whether a subject is permitted to invoke a function. + + :arg str subject: The SSL certificate Subject Common Name to + check against. + :arg str name: The function name to check. + :returns: A boolean indicating whether the action should be permitted. + """ + entry = self.subjects.get(subject) + if entry is None: + return False + return entry.canInvoke(name) + + def canGrant(self, subject): + """Check whether a subject is permitted to grant access to others. + + :arg str subject: The SSL certificate Subject Common Name to + check against. + :returns: A boolean indicating whether the action should be permitted. + """ + entry = self.subjects.get(subject) + if entry is None: + return False + if not entry.grant: + return False + return True + + def grantInvoke(self, subject, invoke): + """Grant permission to invoke certain functions. + + :arg str subject: The SSL certificate Subject Common Name to which + the entry applies. + :arg str invoke: A regular expression that matches the jobs + that connections with this certificate are permitted to + invoke. Also implies the permission to cancel the same + set of jobs in the queue. + """ + e = self.subjects.get(subject) + if not e: + e = ACLEntry(subject) + self.add(e) + e.setInvoke(invoke) + + def grantRegister(self, subject, register): + """Grant permission to register certain functions. + + :arg str subject: The SSL certificate Subject Common Name to which + the entry applies. + :arg str register: A regular expression that matches the jobs that + connections with this certificate are permitted to register. + """ + e = self.subjects.get(subject) + if not e: + e = ACLEntry(subject) + self.add(e) + e.setRegister(register) + + def grantGrant(self, subject): + """Grant permission to grant permissions to other connections. + + :arg str subject: The SSL certificate Subject Common Name to which + the entry applies. + """ + e = self.subjects.get(subject) + if not e: + e = ACLEntry(subject) + self.add(e) + e.setGrant(True) + + def revokeInvoke(self, subject): + """Revoke permission to invoke all functions. + + :arg str subject: The SSL certificate Subject Common Name to which + the entry applies. + """ + e = self.subjects.get(subject) + if e: + e.setInvoke(None) + if e.isEmpty(): + self.remove(subject) + + def revokeRegister(self, subject): + """Revoke permission to register all functions. + + :arg str subject: The SSL certificate Subject Common Name to which + the entry applies. + """ + e = self.subjects.get(subject) + if e: + e.setRegister(None) + if e.isEmpty(): + self.remove(subject) + + def revokeGrant(self, subject): + """Revoke permission to grant permissions to other connections. + + :arg str subject: The SSL certificate Subject Common Name to which + the entry applies. + """ + e = self.subjects.get(subject) + if e: + e.setGrant(False) + if e.isEmpty(): + self.remove(subject) diff --git a/roles/submit-log-processor-jobs/module_utils/gear_constants.py b/roles/submit-log-processor-jobs/module_utils/gear_constants.py new file mode 100644 index 0000000..2751278 --- /dev/null +++ b/roles/submit-log-processor-jobs/module_utils/gear_constants.py @@ -0,0 +1,83 @@ +# Copyright 2013 OpenStack Foundation +# +# 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. + +""" +Protocol Constants +================== + +These are not necessary for normal API usage. See the `Gearman +protocol reference `_ for an explanation +of each of these. + +Magic Codes +----------- + +.. py:data:: REQ + + The Gearman magic code for a request. + +.. py:data:: RES + + The Gearman magic code for a response. + +Packet Types +------------ + +""" + +types = { + 1: 'CAN_DO', + 2: 'CANT_DO', + 3: 'RESET_ABILITIES', + 4: 'PRE_SLEEP', + # unused + 6: 'NOOP', + 7: 'SUBMIT_JOB', + 8: 'JOB_CREATED', + 9: 'GRAB_JOB', + 10: 'NO_JOB', + 11: 'JOB_ASSIGN', + 12: 'WORK_STATUS', + 13: 'WORK_COMPLETE', + 14: 'WORK_FAIL', + 15: 'GET_STATUS', + 16: 'ECHO_REQ', + 17: 'ECHO_RES', + 18: 'SUBMIT_JOB_BG', + 19: 'ERROR', + 20: 'STATUS_RES', + 21: 'SUBMIT_JOB_HIGH', + 22: 'SET_CLIENT_ID', + 23: 'CAN_DO_TIMEOUT', + 24: 'ALL_YOURS', + 25: 'WORK_EXCEPTION', + 26: 'OPTION_REQ', + 27: 'OPTION_RES', + 28: 'WORK_DATA', + 29: 'WORK_WARNING', + 30: 'GRAB_JOB_UNIQ', + 31: 'JOB_ASSIGN_UNIQ', + 32: 'SUBMIT_JOB_HIGH_BG', + 33: 'SUBMIT_JOB_LOW', + 34: 'SUBMIT_JOB_LOW_BG', + 35: 'SUBMIT_JOB_SCHED', + 36: 'SUBMIT_JOB_EPOCH', +} + +for i, name in types.items(): + globals()[name] = i + __doc__ += '\n.. py:data:: %s\n' % name + +REQ = b'\x00REQ' +RES = b'\x00RES' diff --git a/roles/submit-logstash-jobs/tasks/main.yaml b/roles/submit-logstash-jobs/tasks/main.yaml index 2313040..f7e2c15 100644 --- a/roles/submit-logstash-jobs/tasks/main.yaml +++ b/roles/submit-logstash-jobs/tasks/main.yaml @@ -7,4 +7,4 @@ success: "{{ zuul_success }}" host_vars: "{{ hostvars }}" path: "{{ zuul.executor.log_root }}" - log_url: "{{ (lookup('file', zuul.executor.result_data_file) | from_json).get('zuul').get('log_url') }}" + log_url: "{{ (lookup('file', zuul.executor.result_data_file) | from_json).get('data').get('zuul').get('log_url') }}" diff --git a/roles/upload-pages/tasks/main.yaml b/roles/upload-pages/tasks/main.yaml index 924e49c..ad73f34 100644 --- a/roles/upload-pages/tasks/main.yaml +++ b/roles/upload-pages/tasks/main.yaml @@ -3,7 +3,7 @@ path: "{{ zuul_pagesserver_root }}/{{ zuul.project.name }}" state: directory recurse: yes - mode: 0775 + mode: '0775' - name: Check fqdn and vhost_name variables fail: @@ -25,7 +25,7 @@ template: src: templates/vhost.conf.j2 dest: "{{ zuul_pagesvhosts_root }}/pages-{{ zuul.project.name | regex_replace('/', '_') | regex_replace('\\.\\.', '') }}.conf" - mode: 0644 + mode: '0644' register: apache_conf - name: Upload website to publication server