check_elasticsearch

#!/usr/bin/env python
# -*- coding: utf-8 -*-

import argparse
import json
import ssl
from base64 import b64encode
import nagiosplugin
import nagiosplugin.state
import urllib2


SHARD_STATE = {'UNASSIGNED': 1,
				'INITIALIZING': 2,
				'STARTED': 3,
				'RELOCATING': 4,
				'MISSING_PRIMARY': 5}

NODE_METRICS = [["storesize", 'indices.store.size_in_bytes', "B"],
				["documents", 'indices.docs.count', None],

				["index_ops", 'indices.indexing.index_total', None],
				["index_time", 'indices.indexing.index_time_in_millis', "ms"],

				["flush_ops", 'indices.flush.total', None],
				["flush_time", 'indices.flush.total_time_in_millis', "ms"],

				["throttle_time", "indices.store.throttle_time_in_millis", "ms"],

				["index_ops", "indices.indexing.index_total", None],
				["index_time", "indices.indexing.index_time_in_millis", "ms"],
				["delete_ops", "indices.indexing.delete_total", None],
				["delete_time", "indices.indexing.delete_time_in_millis", "ms"],

				["get_ops", "indices.get.total", None],
				["get_time", "indices.get.time_in_millis", "ms"],
				["exists_ops", "indices.get.exists_total", None],
				["exists_time", "indices.get.exists_time_in_millis", "ms"],
				["missing_ops", "indices.get.missing_total", None],
				["missing_time", "indices.get.missing_time_in_millis", "ms"],

				["query_ops", 'indices.search.query_total', None],
				["query_time", 'indices.search.query_time_in_millis', "ms"],
				["fetch_ops", "indices.search.fetch_total", None],
				["fetch_time", "indices.search.fetch_time_in_millis", "ms"],

				["merge_ops", "indices.merges.total", None],
				["merge_time", "indices.merges.time_in_millis", "ms"],

				["refresh_ops", "indices.refresh.total", None],
				["refresh_time", "indices.refresh.total_time_in_millis", "ms"],
]


class ESShard(object):
	def __init__(self, state):
		self.state = state


class ESIndex(object):
	def __init__(self, name, n_shards, n_replicas):
		self.name = name
		self.n_shards = n_shards
		self.n_replicas = n_replicas


class ESNode(object):
	def __init__(self, name=None, esid=None, attributes=None):
		if attributes is None:
			attributes = {}

		self.esid = esid
		self.name = name
		self.attributes = attributes


class ElasticSearchAPI(object):
	def __init__(self, host, port, prefix, scheme = "http", username = None, password = None, cert = None, key = None, insecure = False, ca = False):
		"""
		Initialize ElasticSearch API class
		@param host: API host
		@param port: API port
		@param prefix: API URL Prefix
		@param scheme: API scheme
		@param username: API username
		@param password: API password
		@param cert: API certificate
		@param key: API key
		@param insecure: API verify
		@param ca: API ca
		"""
		self.host = host
		self.port = port
		self.prefix = prefix
		self.scheme = scheme
		self.username = username
		self.password = password
		self.cert = cert
		self.key = key
		self.insecure = insecure
		self.ca = ca

		self.version = None

		if len(prefix) > 0 and not self.prefix.endswith('/'):
			self.prefix += '/'

		self.get_initial_data()

	def get_initial_data(self):
		""" Request initial data from ElasticSearch Server
		"""
		# Request "about" info, so we can figure out the ES version
		es_about = self.get_api_data('')
		self.version = es_about['version']['number']

	def get_health(self):
		""" Request health status from ElasticServer Server
		"""
		return self.get_api_data('_cluster/health')

	def get_state(self):
		return self.get_api_data('_cluster/state')

	def get_stats(self):
		return self.get_api_data('_nodes/_local/stats')

	def get_api_data(self, suffix):
		uri = (r'%s://%s:%d/%s%s' %
		      (self.scheme, self.host, self.port, self.prefix, suffix))
		request = urllib2.Request(uri)
		if self.username:
			authheader = b64encode('%s:%s' % (self.username, self.password))
			request.add_header("Authorization", "Basic %s" % authheader)
		ctx = None
		if self.scheme == "https":
			if self.insecure:
				ctx = ssl._create_unverified_context()
			elif self.cert and self.key:
				ctx = ssl.create_default_context(ssl.Purpose.CLIENT_AUTH)
				ctx.load_cert_chain(self.cert, self.key)
			else:
				ctx = ssl.create_default_context()
		return self.__get_json(request,ctx)

	@staticmethod
	def __get_json(request, ctx):
		try:
			f = urllib2.urlopen(request, context=ctx)
		except urllib2.HTTPError as e:
			raise RuntimeError('Elastic Search API Failure: %s' % str(e))
		except urllib2.URLError as e:
			raise RuntimeError('Elastic Search API URLError: %s' % str(e))

		body = f.read()

		try:
			j = json.loads(body)
		except ValueError as msg:
			raise RuntimeError("Elastic Search API returned non-JSON value: %s" % str(msg))

		return j


class ElasticSearchSummary(nagiosplugin.Summary):
	def ok(self, results):
		return "Monitoring cluster '%s'" % (results['cluster_name'].metric.value)


class ESClusterHealth(nagiosplugin.Context):
	def evaluate(self, metric, resource):
		msg = "Cluster status is %s" % metric.value

		if metric.value == 'green':
			return self.result_cls(nagiosplugin.state.Ok, None, metric)
		elif metric.value == 'yellow':
			return self.result_cls(nagiosplugin.state.Warn, msg, metric)
		elif metric.value == 'red':
			return self.result_cls(nagiosplugin.state.Critical, msg, metric)
		else:
			return self.result_cls(nagiosplugin.state.Unknown, "Cluster status is unknown: %s" % metric.value, metric)


class ESClusterName(nagiosplugin.Context):
	pass


class ShardStateContext(nagiosplugin.Context):
	# Assertion:  Each shard has one primary in STARTED or RELOCATING state.
	def evaluate(self, metric, resource):
		shard_states = metric.value
		for idx_name in shard_states:
			for shard_no in shard_states[idx_name]:
				if shard_states[idx_name][shard_no] == SHARD_STATE['MISSING_PRIMARY']:
					msg = "Index '%s' missing primary on shard %d" % (idx_name, shard_no)
					return self.result_cls(nagiosplugin.state.Critical, msg, metric)
				elif shard_states[idx_name][shard_no] not in (SHARD_STATE['STARTED'], SHARD_STATE['RELOCATING']):
					msg = "Index '%s' primary down on shard %d" % (idx_name, shard_no)
					return self.result_cls(nagiosplugin.state.Critical, msg, metric)

		return self.result_cls(nagiosplugin.state.Ok, None, metric)


class ShardReplicaContext(nagiosplugin.Context):
	# Assertion: Each shard has replicas in STARTED state
	def evaluate(self, metric, resource):
		shard_replica = metric.value
		for idx_name in shard_replica:
			for shard_no in shard_replica[idx_name]:
				if shard_replica[idx_name][shard_no]["status"] == "MISSING_REPLICA":
					msg = "Index '%s' missing replica on shard %d" % (idx_name, shard_no)
					return self.result_cls(nagiosplugin.state.Warn, msg, metric)

				for replica_state in shard_replica[idx_name][shard_no]['replicas_state']:
					if replica_state not in (SHARD_STATE['STARTED'], SHARD_STATE['RELOCATING']):
						msg = "Index '%s' replica down on shard %d" % (idx_name, shard_no)
						return self.result_cls(nagiosplugin.state.Warn, msg, metric)

		return self.result_cls(nagiosplugin.state.Ok, None, metric)


class ShardReplicaLocation(nagiosplugin.Context):
	def evaluate(self, metric, resource):
		replica_loc = metric.value

		for idx_name in replica_loc:
			for shard_no in replica_loc[idx_name]:
				if replica_loc[idx_name][shard_no]['vulnerable_shards'].isdisjoint(replica_loc[idx_name][shard_no]['local_shards']):
					# Suppress the problem unless at least one of the vulnerable shards is on this data node.
					continue

				if len(replica_loc[idx_name][shard_no]['loc_redundancy']) == 1:
					loc = ",".join(list(replica_loc[idx_name][shard_no]['loc_redundancy'])[0])
					msg = "Index '%s' shard %d only exists in location '%s'" % (idx_name, shard_no, loc)
					return self.result_cls(nagiosplugin.state.Warn, msg, metric)

		return self.result_cls(nagiosplugin.state.Ok, None, metric)


class ElasticSearchCheck(nagiosplugin.Resource):
	def __init__(self, args):
		self.args = args

	def probe(self):
		host = self.args.host
		port = int(self.args.port)
		prefix = self.args.prefix
		scheme = self.args.scheme
		username = self.args.username
		password = self.args.password
		cert = self.args.cert
		key = self.args.key
		insecure = self.args.insecure
		ca = self.args.ca

		failure_domain = []
		if (isinstance(self.args.failure_domain, str) and
					self.args.failure_domain):
			failure_domain.extend(self.args.failure_domain.split(","))

		es_server = ElasticSearchAPI(host, port, prefix, scheme, username, password, cert, key, insecure, ca)

		# Request cluster 'health'.  /_cluster/health is like a tl;dr for /_cluster/state (see below).  There is very little useful
		# information here.  We are primarily interested in ES' cluster 'health colour':  a little rating ES gives itself to describe
		# how much pain it is in.
		es_health = es_server.get_health()

		# Request cluster 'state'.  This be where all the meat at, yo. Here, we can see a list of all nodes, indexes, and shards in
		# the cluster.  This response will also contain a map detailing where all shards are living at this point in time.
		es_state = es_server.get_state()

		# Request a bunch of useful numbers that we export as perfdata. Details like the number of get, search, and indexing operations come from here.
		es_stats = es_server.get_stats()

		myid = list(es_stats['nodes'].keys())[0]

		n_nodes = es_health['number_of_nodes']
		n_dnodes = es_health['number_of_data_nodes']

		# Unlike n_dnodes (the number of data nodes), we must compute the number of master-eligible nodes ourselves.
		n_mnodes = 0
		for esid in es_state['nodes']:
			master_elig = True

			# ES will never elect 'client' nodes as masters.
			try:
				master_elig = not (booleanise(es_state['nodes'][esid]['attributes']['client']))
			except KeyError as e:
				if e.args[0] != 'client':
					raise

			try:
				master_elig = (booleanise(es_state['nodes'][esid]['attributes']['master']))
			except KeyError as e:
				if e.args[0] != 'master':
					raise

			if master_elig:
				n_mnodes += 1

		n_active_shards = es_health['active_shards']
		n_relocating_shards = es_health['relocating_shards']
		n_initialising_shards = es_health['initializing_shards']
		n_unassigned_shards = es_health['unassigned_shards']
		n_shards = (n_active_shards + n_relocating_shards + n_initialising_shards + n_unassigned_shards)

		# Map construction
		#

		# String all the dumb ES* objects into a bunch of transitive
		# associations so that we may make some useful assertions about
		# them.
		esid_node_map = {}  # ESID : <ESNode>
		index_primary_map = {}  # <ESIndex> : { 0: <ESShard>, ... }
		name_index_map = {}  # 'bar' : <ESIndex>
		name_node_map = {}  # 'foo' : <ESNode>
		node_esid_map = {}  # <ESNode> : ESID
		node_location_map = {}  # <ESNode> : ('mars',)
		node_shard_map = {}  # <ESNode> : [ <ESShard>, ... ]
		primary_replica_map = {}  # <ESShard> : [ <ESShard>, ... ]
		shard_location_map = {}  # <ESShard> : ('mars',)

		# Build node maps:
		#
		# - esid_node_map
		#     - name_node_map
		#     - node_esid_map
		#     - node_location_map (data nodes only)
		#
		nodes = es_state['nodes']
		for n in nodes:
			name = nodes[n]['name']
			attrs = nodes[n]['attributes']
			node = ESNode(name, n, attrs)

			name_node_map[name] = node
			esid_node_map[n] = node
			node_esid_map[node] = n
			node_shard_map[node] = []

			if failure_domain:
				node_location_map[node] = tuple()
				try:
					node_location_map[node] = (tuple(map(lambda a: attrs[a], failure_domain)))
				except KeyError as e:
					# Nodes that do not store shards (e.g.: 'client'
					# nodes) cannot be expected to have been configured
					# with locational attributes.
					if 'data' not in attrs or booleanise(attrs['data']):
						missing_attr = e.args[0]
						raise nagiosplugin.CheckError("Node '%s' missing location attribute '%s'" % (name, missing_attr))

		# Build index maps:
		#
		#     - name_index_map
		#
		indices = es_state['metadata']['indices']
		n_indices = len(indices)
		n_closed_indices = 0
		for i in indices:
			if indices[i]["state"] == "close":
				n_closed_indices += 1
				continue
			idx_stns = indices[i]['settings']
			if version(es_server.version) < version("1.0.0"):
				idx = ESIndex(i, int(idx_stns['index.number_of_shards']), int(idx_stns['index.number_of_replicas']))
			else:
				idx = ESIndex(i, int(idx_stns['index']['number_of_shards']), int(idx_stns['index']['number_of_replicas']))
			name_index_map[i] = idx

		# Build shard maps:
		#
		#     - index_primary_map
		#     - node_shard_map
		#     - primary_replica_map
		#     - shard_location_map
		#
		for i in name_index_map:
			idx = name_index_map[i]

			if idx not in index_primary_map:
				index_primary_map[idx] = dict(map(lambda m: (m, None), range(idx.n_shards)))

			idx_shards = (es_state['routing_table']['indices'][i]['shards'])
			for d in idx_shards:
				primary = None
				replicas = []
				for s in idx_shards[d]:
					shard = ESShard(SHARD_STATE[s['state'].upper()])

					if s['primary']:
						primary = shard
					else:
						replicas.append(shard)

					if s['state'] != 'UNASSIGNED':
						node = esid_node_map[s['node']]

						node_shard_map[node].append(shard)

						if len(failure_domain) > 0:
							loc = node_location_map[esid_node_map[s['node']]]
							shard_location_map[shard] = loc

				index_primary_map[idx][int(d)] = primary

				if primary is not None:
					primary_replica_map[primary] = replicas

		# Yield cluster-wide metrics first.  If you monitor all of your ES
		# cluster nodes with this plugin, they should all report the
		# same figures for these labels.  Not ideal, but 'tis better to
		# graph this data multiple times than not graph it at all.
		metrics = [["cluster_health", es_health["status"]],
					["cluster_name", es_health["cluster_name"]],
					["cluster_nodes", n_nodes],
					["cluster_mnodes", n_mnodes],
					["cluster_data_nodes", n_dnodes],
					["cluster_act_shards", n_active_shards],
					["cluster_reloc_shards", n_relocating_shards],
					["cluster_init_shards", n_initialising_shards],
					["cluster_unass_shards", n_unassigned_shards],
					["cluster_tot_shards", n_shards],
					["cluster_tot_indices", n_indices],
					["cluster_clsd_indices", n_closed_indices]]

		for metric in metrics:
			if len(metric) == 2:
				label, value = metric
				yield nagiosplugin.Metric("%s" % label, value, context='%s' % label)
			else:
				continue

		# Yield node-specific metric
		for metric in NODE_METRICS:
			label, path, unit = metric
			keys = path.split(".")
			value = es_stats['nodes'][myid]

			for key in keys:
				if value is None:
					break
				try:
					value = value[key]
				except KeyError:
					value = None
					break

			if value is not None:
				yield nagiosplugin.Metric("%s" % label, value, uom=unit, context='%s' % label)

		# Collect data for shard primaries
		shard_states = {}
		for idx_name, idx in name_index_map.items():
			shard_states[idx_name] = {}
			for shard_no in range(idx.n_shards):
				primary = index_primary_map[idx][shard_no]
				if primary is None:
					shard_states[idx_name][shard_no] = SHARD_STATE['MISSING_PRIMARY']
				else:
					shard_states[idx_name][shard_no] = primary.state

		yield nagiosplugin.Metric("shard_states", shard_states)

		# Collect data for shard replicas nodes
		shard_replica = {}
		for idx_name, idx in name_index_map.items():
			expect_replicas = idx.n_replicas
			shard_replica[idx_name] = {}

			for shard_no in range(idx.n_shards):
				shard_replica[idx_name][shard_no] = {}
				primary = index_primary_map[idx][shard_no]
				if primary is None:
					continue

				has_replicas = len(primary_replica_map[primary])
				if has_replicas < expect_replicas:
					shard_replica[idx_name][shard_no]["status"] = "MISSING_REPLICA"
				else:
					shard_replica[idx_name][shard_no]["status"] = "OK"

				shard_replica[idx_name][shard_no]['replicas_state'] = []
				for replica in primary_replica_map[primary]:
					shard_replica[idx_name][shard_no]['replicas_state'].append(replica.state)
		yield nagiosplugin.Metric("shard_replica", shard_replica)

		# Collect replica location data
		replica_loc = {}
		if failure_domain:
			for idx_name, idx in name_index_map.items():
				replica_loc[idx_name] = {}

				# Suppress this test if the index has not been
				# configured with replicas.
				if idx.n_replicas == 0:
					continue

				for shard_no in range(idx.n_shards):
					replica_loc[idx_name][shard_no] = {'loc_redundancy': set(), 'vulnerable_shards': set()}

					primary = index_primary_map[idx][shard_no]

					if primary is None:
						continue

					try:
						loc = shard_location_map[primary]
					except KeyError:
						continue
					replica_loc[idx_name][shard_no]['loc_redundancy'].add(loc)
					replica_loc[idx_name][shard_no]['vulnerable_shards'].add(primary)

					for replica in primary_replica_map[primary]:
						try:
							loc = shard_location_map[replica]
						except KeyError:
							continue
						replica_loc[idx_name][shard_no]['loc_redundancy'].add(loc)
						replica_loc[idx_name][shard_no]['vulnerable_shards'].add(replica)

					replica_loc[idx_name][shard_no]['local_shards'] = set(node_shard_map[esid_node_map[myid]])
		yield nagiosplugin.Metric("replica_location", replica_loc)


def booleanise(b):
	"""Normalise a 'stringified' Boolean to a proper Python Boolean.

	ElasticSearch has a habit of returning "true" and "false" in its
	JSON responses when it should be returning `true` and `false`.  If
	`b` looks like a stringified Boolean true, return True.  If `b`
	looks like a stringified Boolean false, return False.

	Raise ValueError if we don't know what `b` is supposed to represent.

	"""
	s = str(b)
	if s.lower() == "true":
		return True
	if s.lower() == "false":
		return False

	raise ValueError("I don't know how to coerce %r to a bool" % b)


def version(version_string):
	"""Accept a typical version string (ex: 1.0.1) and return a tuple
	of ints, allowing for reasonable comparison."""
	return tuple([int(i) for i in version_string.split('.')])


@nagiosplugin.guarded
def main():
	argp = argparse.ArgumentParser()
	argp.add_argument('-v', '--verbose', action='count', default=0)  # FIXME: change to 0

	argp.add_argument('-f', '--failure-domain', help="A comma-separated list of ElasticSearch attributes that make up your cluster's failure domain[0]. This should be the same list "
						"of attributes that ElasticSearch's location-aware shard allocator has been configured with. If this option is supplied, additional checks are carried out to ensure that primary "
						"and replica shards are not stored in the same failure domain. [0]: http://www.elasticsearch.org/guide/en/elasticsearch/reference/current/modules-cluster.html")

	argp.add_argument('-H', '--host', help="Hostname or network address to probe. The ElasticSearch API should be listening here. Defaults to 'localhost'.", default='localhost')

	argp.add_argument('-m', '--master-nodes', type=int, help="Issue a warning if the number of master-eligible nodes in the cluster drops below this number. By default, do not monitor the number of nodes in the cluster.", default=None)

	argp.add_argument('-p', '--port', type=int, help="TCP port to probe. The ElasticSearch API should be listening here. Defaults to 9200.", default=9200)

	argp.add_argument('-s', '--scheme', type=str, help="URL scheme to use: http or https. Defaults to http.", default='http')

	argp.add_argument('-U', '--username', type=str, help="Username for basic http authentication. Defaults to None.", default=None)

	argp.add_argument('-P', '--password', type=str, help="Password for basic http authentication. Defaults to None.", default=None)

	argp.add_argument('-c', '--cert', type=str, help="Path to certificate for user certificate authentication. Defaults to None.", default=None)

	argp.add_argument('-a', '--ca', type=str, help="Path to certificate authority. Defaults to None i.e. only the system's CA bundle will be used.", default=None)

	argp.add_argument('-k', '--key', type=str, help="Path to key for user certificate authentication. Defaults to None.", default=None)

	argp.add_argument('-I', '--insecure', action='store_true', help="Insecure SSL connection."
"Defaults to False.", default=False)

	argp.add_argument('-r', '--prefix', help="Optional prefix (e.g. 'es') for the ElasticSearch API. Defaults to ''.", default='')
	args = argp.parse_args()

	if args.master_nodes is not None:
		if int(args.master_nodes) < 1:
			argp.error("'master_nodes' must be greater than zero")
			raise ValueError("'master_nodes' must be greater than zero")

	check = nagiosplugin.Check(ElasticSearchCheck(args))

	if args.master_nodes is not None:
		check.add(nagiosplugin.ScalarContext('cluster_mnodes', critical="%d:%d" % (args.master_nodes, args.master_nodes), fmt_metric='{value} master nodes in cluster'))
	else:
		check.add(nagiosplugin.ScalarContext('cluster_mnodes', fmt_metric='{value} master nodes in cluster'))
	check.add(nagiosplugin.ScalarContext('cluster_nodes'))
	check.add(nagiosplugin.ScalarContext('cluster_data_nodes'))
	check.add(nagiosplugin.ScalarContext('cluster_act_shards'))
	check.add(nagiosplugin.ScalarContext('cluster_reloc_shards'))
	check.add(nagiosplugin.ScalarContext('cluster_init_shards'))
	check.add(nagiosplugin.ScalarContext('cluster_unass_shards'))
	check.add(nagiosplugin.ScalarContext('cluster_tot_shards'))
	check.add(nagiosplugin.ScalarContext('cluster_tot_indices'))
	check.add(nagiosplugin.ScalarContext('cluster_clsd_indices'))

	for metric in NODE_METRICS:
		check.add(nagiosplugin.ScalarContext(metric[0]))

	check.add(ShardStateContext('shard_states'))
	check.add(ShardReplicaContext('shard_replica'))
	check.add(ShardReplicaLocation('replica_location'))
	check.add(ESClusterHealth('cluster_health'))
	check.add(ESClusterName('cluster_name', 'Monitoring cluster {value}'))

	check.add(ElasticSearchSummary())

	check.main(verbose=args.verbose)

if __name__ == '__main__':
	main()