From ea373d207ec99d7132b4dd4a25726a3e55ecf96b Mon Sep 17 00:00:00 2001 From: "James E. Blair" Date: Fri, 10 Jul 2020 11:31:13 -0700 Subject: [PATCH] Vendor gear for submit-log-processors-jobs We use the gear library to submit log processing jobs in the post-run playbooks. That is installed in the ansible virtualenvs on our vms, but it is not installed in the virtualenvs on the zuul-executor image. Aside from our use of it, there isn't a compelling reason to ask the Zuul project to add that to the zuul-executor image. And we would prefer to continue using the upstream image rather than make our own. It seems the simplest way to avoid any further complexity here is just to vendor the gear library. It's small and does not change often. Change-Id: I733361be3b7eaaa60dacc0b09112eebaa06e0d9a --- .../library/gear/__init__.py | 3526 +++++++++++++++++ .../library/gear/acl.py | 289 ++ .../library/gear/constants.py | 83 + 3 files changed, 3898 insertions(+) create mode 100644 roles/submit-log-processor-jobs/library/gear/__init__.py create mode 100644 roles/submit-log-processor-jobs/library/gear/acl.py create mode 100644 roles/submit-log-processor-jobs/library/gear/constants.py diff --git a/roles/submit-log-processor-jobs/library/gear/__init__.py b/roles/submit-log-processor-jobs/library/gear/__init__.py new file mode 100644 index 0000000..f7fc767 --- /dev/null +++ b/roles/submit-log-processor-jobs/library/gear/__init__.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 + +from gear import constants +from 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(): + l = "%s\tregister=%s\tinvoke=%s\tgrant=%s\n" % ( + entry.subject, entry.register, entry.invoke, entry.grant) + request.connection.sendRaw(l.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/library/gear/acl.py b/roles/submit-log-processor-jobs/library/gear/acl.py new file mode 100644 index 0000000..07c9e10 --- /dev/null +++ b/roles/submit-log-processor-jobs/library/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/library/gear/constants.py b/roles/submit-log-processor-jobs/library/gear/constants.py new file mode 100644 index 0000000..2751278 --- /dev/null +++ b/roles/submit-log-processor-jobs/library/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'