FcgiNet.rx_nbytes() - Code Metrics - Inspection of "added php-fpm reader for fpm, opcache, apc metrics..." - s4z/plumd - Measure and Improve Code Quality continuously with Scrutinizer

Completed
Push — master ( 55c3a8...4ee339 )

by Kenny
created 2016-08-28 12:34 UTC
FcgiNet.rx_nbytes() B

↳ Parent: FcgiNet
Complexity

Conditions
Size

Total Lines
Duplication

Lines	0
Ratio	0 %
Importance

Changes	1
Bugs	0	Features	1
Metric	Value
cc	4
c	1
b	0
f	1
dl	0
loc	25
rs	8.5806
# -*- coding: utf-8 -*-

__author__ = 'Kenny Freeman'
__email__ = '[email protected]'
__license__ = "ISCL"
__docformat__ = 'reStructuredText'

import sys

PY3 = sys.version_info > (3,)

import json
import struct
import socket
from collections import deque

import plumd
import plumd.plugins


class FcgiParameterProblem(Exception):
    """Generic problem with FastCGI request data."""
    pass

class FcgiRequestFailed(Exception):
    """Generic FastCGI request failed exception."""
    pass

# Number of bytes in a FCGI_Header
FCGI_HEADER_LEN         =  8

# Version of the FastCGI Protocol
FCGI_VERSION_1          =  1

# Possible values for the type in FCGI_Header
FCGI_BEGIN_REQUEST      =  1
#FCGI_ABORT_REQUEST      =  2
FCGI_END_REQUEST        =  3
FCGI_PARAMS             =  4
FCGI_STDIN              =  5
FCGI_STDOUT             =  6
FCGI_STDERR             =  7
FCGI_DATA               =  8
#FCGI_GET_VALUES         =  9
#FCGI_GET_VALUES_RESULT  = 10
FCGI_UNKNOWN_TYPE       = 11
FCGI_MAXTYPE = FCGI_UNKNOWN_TYPE

# Values for role in FCGI_BeginRequestBody
FCGI_RESPONDER  = 1

# Values for protocolStatus in FCGI_EndRequestBody
FCGI_REQUEST_COMPLETE = 0
FCGI_CANT_MPX_CONN    = 1
FCGI_OVERLOADED       = 2
FCGI_UNKNOWN_ROLE     = 3

# Numbe rof bytes in a FCGI_BeginRequestBody, etc
FCGI_BEGIN_REQUEST_LEN = 8


def fcgi_header(dat):
    """Return a dict with values unpacked from dat:

        dat = {
            'version': FCGI_VERSION_1,
            'type': FCGI_PARAMS,
            'id': 1,
            'data': 512, # data length
            'padding': 0 # padding length
        }

    :param dat: A packed FastCGI header string
    :type dat: str
    :rtype: dict
    """
    keys = ['version', 'type', 'id', 'data_len', 'pad_len']
    vals = struct.unpack("!BBHHBx", dat)
    return dict(zip(keys, vals))

def fcgi_end_request(dat):
    """Return a tuple of (appstatus, protocolstatus) from the end request
    body dat.

    :param dat: A packed string representing the FCGI_EndRequestBody
    :type dat: str
    :rtype: tuple(int, int)
    """
    astat, pstat = struct.unpack('!LB3x', dat)
    return (astat, pstat)

def fcgi_unpack(buff):
    """Return a dict containing all key, value pairs in the buffer.

    :param buff: str to unpack pairs from
    :type buff: str
    :rtype: dict
    """
    ret = {}
    maxi = len(buff)
    i = 0
    while i < maxi:
        # extract klen, vlen then key, val
        klen = ord(buff[i][0])
        if klen < 128:
            i += 1
        else:
            klen = struct.unpack("!L", struct_bytes)[0] & 0x7fffffff
            i += 4
        # now get the val length
        vlen = ord(buff[i][0])
        if vlen < 128:
            i += 1
        else:
            vlen = struct.unpack("!L", struct_bytes)[0] & 0x7fffffff
            i += 4
        # get the values
        key = buff[i:i+klen]
        i += klen
        val = buff[i:i+vlen]
        i += vlen
        ret[key] = val
    return ret

def fcgi_pack(vals):
    """Pack a dict of key=value pairs into a buffer.

    :param vals: A dict of key=value pairs
    :type vals: dict
    :rtype: str
    """
    ret = deque()
    # pack: key_len+val_len+key+val
    for key, val in vals.items():
        val = str(val) if val is not None else ""
        klen = len(key)
        vlen = len(val)

        # does key_len fit in a char?
        if klen < 128:
            ret.append(chr(klen))
        else:
            # should check if len(key) > 0x7fffffff..
            #long value:      0x80000000L = 2147483648
            ret.append(struct.pack('!L', klen | 0x80000000))

        # same, does it fit in a single char?
        if vlen < 128:
            ret.append(chr(vlen))
        else:
            # should check if len(val) > 0x7fffffff..
            #long value:      0x80000000L = 2147483648
            ret.append(struct.pack('!L', vlen | 0x80000000))
        ret.append(str(key))
        ret.append(str(val))
    return "".join(ret)

def fcgi_type_begin():
    """Return a FCGI_BeginRequestRecord Responder request.

    :rtype: str
    """
    header = struct.pack("!BBHHBx", FCGI_VERSION_1, FCGI_BEGIN_REQUEST,
                         1, FCGI_BEGIN_REQUEST_LEN, 0)
    body = struct.pack('!HB5x', FCGI_RESPONDER, 0)
    return "".join([header, body])

def fcgi_type_params(params):
    """Return a FCGI_PARAMS record.

    :param params: A str returned from fcgi_pack.
    :type params: str
    :rtype: str
    """
    header = struct.pack("!BBHHBx", FCGI_VERSION_1, FCGI_PARAMS,
                         1, len(params), 0)
    return "".join([header, params])

def fcgi_type_data(data):
    """Return a FCGI_DATA record.

    :param data: A payload string
    :type data: str
    :rtype: str
    """
    header = struct.pack("!BBHHBx", FCGI_VERSION_1, FCGI_DATA,
                         1, len(data), 0)
    return "".join([header, data])


def fcgi_type_stdin(data):
    """Return a FCGI_STDIN record.

    :param data: The data to send to stdin
    :type data: str
    :rtype: str
    """
    header = struct.pack("!BBHHBx", FCGI_VERSION_1, FCGI_STDIN,
                         1, len(data), 0)
    return "".join([header, data])


class FcgiNet(object):
    """A socket reader/writer that reads/writes FastCGI records."""

    def __init__(self, addr, timeout):
        """A helper class to read/write FcgiHeaders to/from a server.

        :param addr: The FastCGI host:port to connect to.
        :type addr: str
        :param to: An optional timeout
        :type to: int
        :rtype: FcgiNet
        """
        self.addr = addr
        self.timeout = timeout
        self.socket = None

    def connect(self):
        if self.socket:
            self.socket.close()
        try:
            # connect
            self.socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
            # set timeout for socket operations
            self.socket.settimeout(self.timeout)
            self.socket.connect(self.addr)
        except Exception as e:
            msg = "could not connect: {0}"
            raise FcgiRequestFailed(msg.format(e))

    def rx_nbytes(self, nbytes):
        """Read nbytes from our socket, return a tuple of the number of bytes
        read and the data read.

        :param nbytes: The number of bytes to attempt to read
        :type nbytes: int

        :rtype: tuple(int, str)
        :raises: FcgiRequestFailed
        """
        dat = deque()
        nbytes_rx = 0
        while nbytes_rx < nbytes:
            buff = None
            try:
                buff = self.socket.recv(nbytes)
            except socket.error as e:
                msg = "Exception reading from remote host {0}:{1}: {2}"
                msg = msg.format(self.addr[0], self.addr[1], e)
                raise FcgiRequestFailed(msg)
            if not buff:
                break
            dat.append(buff)
            nbytes_rx += len(buff)
        return (nbytes_rx, "".join(dat))

    def rx(self):
        """Read a FastCGI header and payload from our socket and return it as a
        a tuple of (header, buffer).

        :raises:
            FcgiRequestFailed

        :rtype: tuple(FcgiHeader, dict)
        """
        # hdr.data will raise ValueError if data length != FCGI_HEADER_LEN
        hlen, dat  = self.rx_nbytes(FCGI_HEADER_LEN)
        if hlen != FCGI_HEADER_LEN:
            msg = "Excpecting header of {0} bytes, received {1}"
            raise FcgiRequestFailed(msg.format(FCGI_HEADER_LEN, hlen))

        hdr = fcgi_header(dat)
        dat = ""
        dat_len = hdr['data_len']
        pad_len = hdr['pad_len']

        if dat_len:
            dlen, dat = self.rx_nbytes(dat_len)
            if dlen != dat_len:
                msg = "Excpecting data of {0} bytes, received {1}"
                raise FcgiRequestFailed(msg.format(dat_len, dlen))

        if pad_len:
            plen, pad = self.rx_nbytes(pad_len)
            if plen != pad_len:
                msg = "Excpecting padding of {0} bytes, received {1}"
                raise FcgiRequestFailed(msg.format(pad_len, plen))

        return (hdr, dat)

    def tx(self, dat):
        """Send a FastCGI Record to our socket.

        :param record: The FcgiRecord to send
        :type record: FcgiRecord
        """
        if not self.socket:
            self.connect()
        self.socket.sendall(dat)

    def end(self):
        """Close the socket."""
        if self.socket:
            self.socket.close()


def fcgi_call(addr, timeout, script, root, query):
    """Call <script> on local/remote FastCGI server <addr> and return results.

    The script is called with no arguments and minimal FastCGI parameters
    are set.

    :param addr: The host:port to connect to
    :type addr: str
    :param timeout: The tcp timeout to set on the socket
    :type timeout: int
    :param script: The name of the script to call
    :type script: str
    :param root: The root directory the script lives in
    :type root: str
    :param query: The query string for the request (eg. json&full)
    :type query: str
    :rval: tuple()
    :raises: FcgiRequestFailed
    """
    net = FcgiNet(addr, timeout)
    sout = deque()
    serr = deque()
    r_begin = fcgi_type_begin()
    params = {
        "QUERY_STRING": "json",
        "REQUEST_METHOD": "GET",
        "CONTENT_TYPE": "",
        "CONTENT_LENGTH": "",
        "SCRIPT_NAME": script,
        "REQUEST_URI": script,
        "DOCUMENT_ROOT": root,
        "SERVER_PROTOCOL": "HTTP/1.1",
        "GATEWAY_INTERFACE": "CGI/1.1",
        "SERVER_SOFTWARE": "fcgi/1.0",
        "REMOTE_ADDR": "127.0.0.1",
        "REMOTE_PORT": "0",
        "SERVER_ADDR": "127.0.0.1",
        "SERVER_PORT": "80",
        "SERVER_NAME": "localhost",
        "PATH_INFO": script,
        "SCRIPT_FILENAME": "{0}/{1}".format(root, script)
    }
    param_dat = fcgi_pack(params)
    r_params = fcgi_type_params(param_dat)
    r_stdin = fcgi_type_stdin("")
    net.tx("".join([r_begin, r_params, r_stdin]))
    while True:
        hdr, buff = net.rx()
        if hdr['type'] == FCGI_STDOUT:
            sout.append(buff)
            continue
        elif hdr['type'] == FCGI_STDERR:
            serr.append(buff)
            continue
        elif hdr['type'] == FCGI_END_REQUEST:
            net.end()
            astatus, pstatus = fcgi_end_request(buff)
            if pstatus != FCGI_REQUEST_COMPLETE or astatus != 0:
                msg = "bad proto/app status: {0}, {1}".format(pstatus, astatus)
                raise FcgiRequestFailed(msg)
            break
        net.end()
        msg = "unexpected response: {0} : {1}"
        raise FcgiRequestFailed(msg.format(hdr, buff))
    net.end()
    return ("".join(sout), "".join(serr))



class Php(plumd.plugins.Reader):
    """Plugin to record php-fpm, opcache and apc metrics."""

    # default config values
    defaults = {
        'poll.interval': 10,
        'fpm_status': "/fpmstatus",         # match pm.status_path from config
        'fpm_status_args': "json",          # query params (json required)
        'fpm_pools': {
            'www': {                        # add an entry per pool to monitor
                'host': '127.0.0.1',
                'port': 9000,
                'script': 'cacheinfo.php',
                'root': '/var/www'
            }
        },
        'record': {
            'apc': [ "expunges","mem_size", "num_entries",  "num_hits",
                     "num_inserts", "num_misses", "num_slots", "ttl" ],
            'apc_sma': [ "avail_mem", "num_seg", "seg_size" ]
        },
        'record_op': {
            "interned_strings_usage": [
                "buffer_size", "free_memory", "number_of_strings",
                "used_memory"
                ],
            "memory_usage": [
                "current_wasted_percentage", "free_memory", "used_memory", "wasted_memory"
                ],
            "opcache_statistics": [
                "blacklist_miss_ratio", "blacklist_misses", "hash_restarts",
                "hits", "manual_restarts", "max_cached_keys", "misses",
                "num_cached_keys","num_cached_scripts", "oom_restarts",
                "opcache_hit_rate"
                ]
        },
        'record_status': [
            "accepted conn", "active processes", "idle processes",
            "listen queue", "listen queue len", "max active processes",
            "max children reached", "max listen queue", "slow requests",
            "total processes"
        ],
        'rename': {
            'interned_strings_usage': 'strings',
            'memory_usage': 'mem',
            'opcache_statistics': 'cache',
            'buffer_size': 'buf_size',
            'free_memory': 'mem_free',
            'number_of_strings': 'num_str',
            'used_memory': 'mem_used',
            'current_wasted_percentage': 'waste_perc',
            'wasted_memory': 'mem_wasted',
            'blacklist_miss_ratio': 'bl_miss_ratio',
            'blacklist_misses': 'bl_miss',
            'hash_restarts': 'h_restarts',
            'manual_restarts': 'm_restarts',
            'max_cached_keys': 'max_keys',
            'num_cached_keys': 'num_keys',
            'num_cached_scripts': 'num_scripts',
            'opcache_hit_rate': 'hit_rate',
            'accepted conn': 'con_accepted',
            'active processes': 'proc_active',
            'idle processes': 'proc_idle',
            'listen queue': 'listen_q',
            'listen queue len': 'listen_q_len',
            'max active processes': 'proc_max_active',
            'max children reached': 'proc_max_child',
            'max listen queue': 'listen_q_max',
            'slow requests': 'req_slow',
            'total processes': 'proc_total'
        },
        'timeout': 10
    }

    def __init__(self, log, config):
        """Plugin to record nginx stub_status metrics.

        :param log: A logger
        :type log: logging.RootLogger
        :param config: a plumd.config.Conf configuration helper instance.
        :type config: plumd.config.Conf
        """
        super(Php, self).__init__(log, config)
        self.config.defaults(Php.defaults)

        self.fpm_status = self.config.get('fpm_status')
        self.fpm_status_args = self.config.get('fpm_status_args')
        self.record_status = self.config.get('record_status')
        self.pools = self.config.get('pools')
        self.timeout = self.config.get('timeout')
        self.record = self.config.get("record")
        self.record_op = self.config.get("record_op")
        self.rename = self.config.get("rename")



    def poll(self):
        """Query PHP-FPM for metrics over a FastCGI connection.

        Records php-fpm pool stats from pm.status_path and also
        output from a php script that returns both opcache and apc metrics.

        The php-fpm pool stats output is expected to be in json by setting a
        QUERY_STRING FastCGI parameter to json (currently does not support
        json&full).

        The opcache and apc metrics are parsed from a custom php script - see
        misc in the git repo for the script source.

        :rtype: ResultSet
        """
        result = plumd.Result("php")
        record = self.config.get('record')
        record_op = self.config.get('record_op')
        rename = self.config.get('rename')
        timeout = self.config.get('timeout')
        record_status = self.config.get('record_status')
        fpm_status = self.config.get('fpm_status')
        fpm_status_args = self.config.get('fpm_status_args')
        fpm_pools = self.config.get('fpm_pools')

        # check each configured pool
        for pname, pconf in fpm_pools.items():
            try:
                addr = (pconf['host'], pconf['port'])
                script = pconf['script']
                root = pconf['root']
            except KeyError as e:
                msg = "Php: pool {0} config invalid: {1}"
                self.log.error(msg.format(pname, e))
                continue
            # call the status script and get its output
            cinfo = {}
            try:
                sout, serr = fcgi_call(addr, timeout, fpm_status, root,
                                       fpm_status_args)
                lines = [ line for line in sout.split("\n")
                          if line.startswith("{\"") ]
                cinfo = json.loads("".join(lines))
            except Exception as e:
                msg = "Php: failed to call {0} for pool {1}: {2}"
                self.log.error(msg.format(fpm_status, pname, e))
                continue

            # record fpm status metrics
            for metric in record_status:
                if metric not in cinfo:
                    continue
                mname = metric if metric not in rename else rename[metric]
                mstr = "{0}.fpm.{1}".format(pname, mname)
                result.add(plumd.Float(mstr, cinfo[metric]))

            # call the cache info script and get its output
            cinfo = {}
            try:
                sout, serr = fcgi_call(addr, timeout, script, root, "")
                lines = [ line for line in sout.split("\n")
                          if line.startswith("{\"") ]
                cinfo = json.loads("".join(lines))
            except Exception as e:
                msg = "Php: failed to call {0} for pool {1}: {2}"
                self.log.error(msg.format(script, pname, e))
                continue

            # apc
            for key, metrics in record.items():
                if key not in cinfo:
                    continue
                mkey = key if key not in rename else rename[key]
                for metric in metrics:
                    if metric not in cinfo[key]:
                        continue
                    mname = metric if metric not in rename else rename[metric]
                    mstr = "{0}.{1}.{2}".format(pname, mkey, mname)
                    result.add(plumd.Float(mstr, cinfo[key][metric]))

            # opcache
            for key, metrics in record_op.items():
                if key not in cinfo['op']:
                    continue
                mkey = ke if key not in rename else rename[key]
                for metric in metrics:
                    if metric not in cinfo['op'][key]:
                        continue
                    mname = metric if metric not in rename else rename[metric]
                    mstr = "{0}.op.{1}.{2}".format(pname, mkey, mname)
                    result.add(plumd.Float(mstr, cinfo['op'][key][metric]))

        return plumd.ResultSet([result])


"""
Example php script and pm.status_path outputs:

Cache output:
{
    "apc": {
        "expunges": 0,
        "file_upload_progress": 1,
        "mem_size": 65600,
        "memory_type": "mmap",
        "num_entries": 100,
        "num_hits": 0,
        "num_inserts": 100,
        "num_misses": 0,
        "num_slots": 4099,
        "start_time": 1472356538,
        "ttl": 0
    },
    "op": {
        "cache_full": false,
        "interned_strings_usage": {
            "buffer_size": 8388608,
            "free_memory": 8050800,
            "number_of_strings": 3748,
            "used_memory": 337808
        },
        "memory_usage": {
            "current_wasted_percentage": 0.0081896781921387,
            "free_memory": 123263200,
            "used_memory": 10943536,
            "wasted_memory": 10992
        },
        "opcache_enabled": true,
        "opcache_statistics": {
            "blacklist_miss_ratio": 0,
            "blacklist_misses": 0,
            "hash_restarts": 0,
            "hits": 90,
            "last_restart_time": 0,
            "manual_restarts": 0,
            "max_cached_keys": 7963,
            "misses": 5,
            "num_cached_keys": 2,
            "num_cached_scripts": 2,
            "oom_restarts": 0,
            "opcache_hit_rate": 94.736842105263,
            "start_time": 1472356538
        },
        "restart_in_progress": false,
        "restart_pending": false
    }
}



Cache Output w/ SMA:
{
    "apc": {
        "expunges": 0,
        "file_upload_progress": 1,
        "mem_size": 65600,
        "memory_type": "mmap",
        "num_entries": 100,
        "num_hits": 0,
        "num_inserts": 100,
        "num_misses": 0,
        "num_slots": 4099,
        "start_time": 1472356538,
        "ttl": 0
    },
    "apc_sma": {
        "avail_mem": 33452504,
        "num_seg": 1,
        "seg_size": 33554296
    },
    "op": {
        "cache_full": false,
        "interned_strings_usage": {
            "buffer_size": 8388608,
            "free_memory": 8050800,
            "number_of_strings": 3748,
            "used_memory": 337808
        },
        "memory_usage": {
            "current_wasted_percentage": 0.0081896781921387,
            "free_memory": 123263200,
            "used_memory": 10943536,
            "wasted_memory": 10992
        },
        "opcache_enabled": true,
        "opcache_statistics": {
            "blacklist_miss_ratio": 0,
            "blacklist_misses": 0,
            "hash_restarts": 0,
            "hits": 90,
            "last_restart_time": 0,
            "manual_restarts": 0,
            "max_cached_keys": 7963,
            "misses": 5,
            "num_cached_keys": 2,
            "num_cached_scripts": 2,
            "oom_restarts": 0,
            "opcache_hit_rate": 94.736842105263,
            "start_time": 1472356538
        },
        "restart_in_progress": false,
        "restart_pending": false
    }
}





Normal output:
X-Powered-By: PHP/5.4.16
Expires: Thu, 01 Jan 1970 00:00:00 GMT
Cache-Control: no-cache, no-store, must-revalidate, max-age=0
Content-Type: application/json

{
    "accepted conn": 164,
    "active processes": 1,
    "idle processes": 5,
    "listen queue": 0,
    "listen queue len": 128,
    "max active processes": 1,
    "max children reached": 0,
    "max listen queue": 0,
    "pool": "www",
    "process manager": "dynamic",
    "slow requests": 0,
    "start since": 16150,
    "start time": 1472356538,
    "total processes": 6
}


Full output:
X-Powered-By: PHP/5.4.16
Expires: Thu, 01 Jan 1970 00:00:00 GMT
Cache-Control: no-cache, no-store, must-revalidate, max-age=0
Content-Type: application/json

{
    "accepted conn": 162,
    "active processes": 1,
    "idle processes": 5,
    "listen queue": 0,
    "listen queue len": 128,
    "max active processes": 1,
    "max children reached": 0,
    "max listen queue": 0,
    "pool": "www",
    "process manager": "dynamic",
    "processes": [
        {
            "content length": 0,
            "last request cpu": 0.0,
            "last request memory": 262144,
            "pid": 13170,
            "request duration": 790,
            "request method": "GET",
            "request uri": "/fpmstatus?json&full",
            "requests": 29,
            "script": "-",
            "start since": 15954,
            "start time": 1472356538,
            "state": "Idle",
            "user": "-"
        },
        {
            "content length": 0,
            "last request cpu": 0.0,
            "last request memory": 0,
            "pid": 13171,
            "request duration": 219,
            "request method": "GET",
            "request uri": "/fpmstatus?json&full",
            "requests": 29,
            "script": "-",
            "start since": 15954,
            "start time": 1472356538,
            "state": "Running",
            "user": "-"
        },
        {
            "content length": 0,
            "last request cpu": 3802.28,
            "last request memory": 262144,
            "pid": 13172,
            "request duration": 263,
            "request method": "GET",
            "request uri": "/fpmstatus?json",
            "requests": 28,
            "script": "-",
            "start since": 15954,
            "start time": 1472356538,
            "state": "Idle",
            "user": "-"
        },
        {
            "content length": 0,
            "last request cpu": 0.0,
            "last request memory": 262144,
            "pid": 13173,
            "request duration": 216,
            "request method": "GET",
            "request uri": "/fpmstatus?json",
            "requests": 28,
            "script": "-",
            "start since": 15954,
            "start time": 1472356538,
            "state": "Idle",
            "user": "-"
        },
        {
            "content length": 0,
            "last request cpu": 0.0,
            "last request memory": 262144,
            "pid": 13174,
            "request duration": 466,
            "request method": "GET",
            "request uri": "/fpmstatus?json&full",
            "requests": 28,
            "script": "-",
            "start since": 15954,
            "start time": 1472356538,
            "state": "Idle",
            "user": "-"
        },
        {
            "content length": 0,
            "last request cpu": 0.0,
            "last request memory": 262144,
            "pid": 15823,
            "request duration": 780,
            "request method": "GET",
            "request uri": "/fpmstatus?json&full",
            "requests": 20,
            "script": "-",
            "start since": 875,
            "start time": 1472371617,
            "state": "Idle",
            "user": "-"
        }
    ],
    "slow requests": 0,
    "start since": 15954,
    "start time": 1472356538,
    "total processes": 6
}
"""

s4z / plumd

Push — master ( 55c3a8...4ee339 )

FcgiNet.rx_nbytes() B

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like
