"""
OneAgent Memcached plugin
=========================
Gathers Memcached stats.
The full description of stats is described at https://github.com/memcached/memcached/blob/master/doc/protocol.txt.
Supported communication protocols are below:
* TCP - simple text lines
* Unix sockets - simple text lines
* UDP - text lines with binary header
All needed configuration is retrieved from parsing process command line.
"""
from ruxit.api.data import PluginMeasurement
from ruxit.api.base_plugin import BasePlugin
from ruxit.api.exceptions import ConfigException
from enum import IntEnum
import os
import stat
import socket
import struct
from contextlib import closing
class _CommunicationType(IntEnum):
tcp = 1
udp = 2
unix = 3
[docs]class MemcachedPlugin(BasePlugin):
"""
Memcached plugin class.
This plugin is stateful - initialize and close methods are defined in order to maintain one server connection for the lifetime of the plugin.
"""
absolute_values = {
"curr_connections",
"limit_maxbytes",
"bytes"
}
"""
Values retrieved from Memcached and send to Ruxit server as absolute values.
"""
per_second_values = {
"cmd_get",
"cmd_set",
"get_hits",
"get_misses",
"evictions",
"bytes_read",
"bytes_written"
}
"""
Values retrieved from Memcached and send to Ruxit server as per second values.
"""
SOCKET_TIMEOUT_SECONDS = 10
[docs] def set_communication_data(self, command_line):
"""
Parse command line and set communication protocol data.
Memcached command parameters are described at http://linux.die.net/man/1/memcached
Following data are discovered:
* TCP interface: IP address and port
* UDP interface: IP address and port
* Unix socket: Unix socket name
"""
self.port = 11211
self.interface = "127.0.0.1"
self.comm_type = _CommunicationType.tcp
commands = command_line.split(" ")
prev_command = ""
for command in commands:
if prev_command == "-s":
self.unix_socket = command
self.comm_type = _CommunicationType.unix
permissions = stat.S_IMODE(os.stat(command).st_mode)
if (permissions & 0o666) == 0o666:
return True
raise ConfigException("Access to unix socket forbidden: %s" % command)
elif prev_command == "-l":
self.interface = command
elif prev_command == "-U":
self.comm_type = _CommunicationType.udp
try:
self.port = int(command)
except ValueError:
self.port = None
elif prev_command == "-p":
self.comm_type = _CommunicationType.tcp
try:
self.port = int(command)
except ValueError:
self.port = None
prev_command = command
return self.port is not None
[docs] def initialize(self, **kwargs):
"""
Initializes Memcached plugin.
Raises:
ruxit.api.exceptions.ConfigException there is no Memcached process or command line isn't recognized
"""
self.entity = kwargs["associated_entity"]
for process in self.entity.processes:
command_line = process.properties.get("CmdLine", "")
if self.set_communication_data(command_line):
return
else:
raise ConfigException("Unable to parse Memcached commandline")
raise ConfigException("There is no Memcached process in PGI")
def _query_tcp(self, bytes_to_read=2048):
with closing(socket.socket(socket.AF_INET, socket.SOCK_STREAM)) as client_socket:
client_socket.settimeout(self.SOCKET_TIMEOUT_SECONDS)
client_socket.connect((self.interface, self.port))
return self._get_client_socket_data(client_socket, bytes_to_read)
def _query_ux_socket(self, bytes_to_read=2048):
with closing((socket.socket(socket.AF_UNIX, socket.SOCK_STREAM))) as client_socket:
client_socket.settimeout(self.SOCKET_TIMEOUT_SECONDS)
client_socket.connect(self.unix_socket)
return self._get_client_socket_data(client_socket, bytes_to_read)
def _query_udp(self):
with closing(socket.socket(socket.AF_INET, socket.SOCK_DGRAM)) as client_socket:
command = struct.pack("!Hhhh", 1, 0, 1, 0) + 'stats\r\n'.encode('utf-8')
client_socket.sendto(command, (self.interface, self.port))
output = ""
client_socket.settimeout(self.SOCKET_TIMEOUT_SECONDS)
data, server = client_socket.recvfrom(8192)
output += data[8:].decode("utf-8")
if "ERROR" in output:
raise ConfigException("'ERROR' in memcached stats output")
elif 'END' not in output:
raise ConfigException('Memcached stats are incomplete')
return output
def _get_client_socket_data(self, client_socket, bytes_to_read):
client_socket.send("stats\r\n".encode())
output = ""
iteration = 1
while "END" not in output:
output += client_socket.recv(bytes_to_read).decode()
if "ERROR" in output:
raise ConfigException("'ERROR' in memcached stats output")
if iteration > 10:
raise ConfigException("Error while getting memcached data")
iteration += 1
return output
[docs] def query(self, **kwargs):
"""
Calls specific communication protocol request, parses response and send data to Ruxit server.
"""
try:
if self.comm_type == _CommunicationType.tcp:
output = self._query_tcp()
elif self.comm_type == _CommunicationType.udp:
output = self._query_udp()
elif self.comm_type == _CommunicationType.unix:
output = self._query_ux_socket()
else:
output = ""
bytes_stat = None
limit_max_bytes = None
for line in output.splitlines():
parsed = line.split(" ")
if "STAT" == parsed[0]:
metric = parsed[1]
if metric in self.absolute_values:
self.results_builder.add_absolute_result(PluginMeasurement(key=parsed[1]+'2', value=parsed[2]))
if metric == 'bytes':
bytes_stat = float(parsed[2])
elif metric == 'limit_maxbytes':
limit_max_bytes = float(parsed[2])
elif metric in self.per_second_values:
self.results_builder.add_per_second_result(PluginMeasurement(key=parsed[1]+'2', value=parsed[2]))
if bytes_stat is not None and limit_max_bytes is not None:
self.results_builder.add_absolute_result(PluginMeasurement(
key='cache_usage2',
value=int(100*(bytes_stat/limit_max_bytes))
))
except ConnectionRefusedError as ex:
raise ConfigException('No connection to port "%s"' % self.port) from ex
except socket.timeout as ex:
raise ConfigException('Socket timed out while getting data') from ex
