core/source/logic/conf.py

def conf_schema_active(
):
	return {
		"description": "whether the check shall be executed",
		"type": "boolean",
		"default": True,
	}
	

def conf_schema_threshold(
):
	return {
		"description": "how often a condition has to occur in order to be reported",
		"type": "integer",
		"minimum": 1,
		"default": 3,
	}
	

def conf_schema_annoy(
):
	return {
		"description": "whether notifications about non-ok states shall be kept sending after the threshold has been surpassed",
		"type": "boolean",
		"default": False,
	}
	

def conf_schema_interval(
	allow_null,
	default
):
	return {
		"anyOf": [
			{
				"description": "in seconds",
				"type": ("integer" if allow_null else ["null", "integer"]),
				"exclusiveMinimum": 0,
			},
			{
				"description": "as text",
				"type": "string",
				"enum": [
					"minute",
					"hour",
					"day",
					"week",
				]
			},
		],
		"default": default,
	}
	

def conf_schema_schedule(
):
	return {
		"type": "object",
		"additionalProperties": False,
		"properties": {
			"regular_interval": conf_schema_interval(False, (60 * 60)),
			"attentive_interval": conf_schema_interval(False, (60 * 2)),
			"reminding_interval": conf_schema_interval(True, (60 * 60 * 24)),
		},
		"required": [
		],
	}
	

def conf_schema_notifications(
	notification_channel_implementations
):
	return {
		"type": "array",
		"items": {
			"anyOf": list(
				map(
					lambda pair: {
						"title": ("notification channel '%s'" % pair[0]),
						"type": "object",
						"unevaluatedProperties": False,
						"properties": {
							"kind": {
								"type": "string",
								"enum": [pair[0]]
							},
							"parameters": pair[1].parameters_schema(),
						},
						"required": [
							"kind",
							"parameters"
						],
					},
					notification_channel_implementations.items()
				)
			)
		},
		"default": [
			{
				"kind": "console",
				"parameters": {
				}
			},
		],
	}
	

def conf_schema_root(
	check_kind_implementations,
	notification_channel_implementations
):
	return {
		"type": "object",
		"additionalProperties": False,
		"properties": {
			"defaults": {
				"description": "default values for checks",
				"type": "object",
				"additionalProperties": False,
				"properties": {
					"active": conf_schema_active(),
					"threshold": conf_schema_threshold(),
					"annoy": conf_schema_annoy(),
					"schedule": conf_schema_schedule(),
					"notifications": conf_schema_notifications(notification_channel_implementations),
				},
				"required": [
				],
			},
			"includes": {
				"type": "array",
				"items": {
					"type": "string"
				},
				"default": [],
				"description": "list of relative or absolute paths to other hmdl files on the local machine, which shall be subsumed in the overall monitoring task"
			},
			"checks": {
				"type": "array",
				"items": {
					"allOf": [
						{
							"description": "should represent a specific check",
							"type": "object",
							"unevaluatedProperties": False,
							"properties": {
								"name": {
									"type": "string"
								},
								"title": {
									"type": "string"
								},
								"active": conf_schema_active(),
								"threshold": conf_schema_threshold(),
								"annoy": conf_schema_annoy(),
								"schedule": conf_schema_schedule(),
								"notifications": conf_schema_notifications(notification_channel_implementations),
							},
							"required": [
								"name",
							],
						},
						{
							"anyOf": list(
								map(
									lambda pair: {
										"title": ("check '%s'" % pair[0]),
										"type": "object",
										"unevaluatedProperties": False,
										"properties": {
											"kind": {
												"type": "string",
												"enum": [pair[0]]
											},
											"parameters": pair[1].parameters_schema(),
											"custom": {
												"description": "custom data, which shall be attached to notifications",
												"default": None,
											},
										},
										"required": [
											"kind",
											"parameters",
										]
									},
									check_kind_implementations.items()
								)
							),
						},
					]
				}
			}
		},
		"required": [
		]
	}
	

def conf_normalize_interval(
	interval_raw
):
	if (interval_raw is None):
		return None
	else:
		if (type(interval_raw) == int):
			return interval_raw
		elif (type(interval_raw) == str):
			map_ = {
				"minute": (60),
				"hour": (60 * 60),
				"day": (60 * 60 * 24),
				"week": (60 * 60 * 24 * 7),
			}
			if (interval_raw not in map_):
				raise ValueError("invalid string interval value: %s" % interval_raw)
			else:
				return map_[interval_raw]
		else:
			raise ValueError("invalid type for interval value")
		

def conf_normalize_schedule(
	node
):
	node_ = dict_merge(
		{
			"regular_interval": (60 * 60),
			"attentive_interval": (60 * 2),
			"reminding_interval": (60 * 60 * 24),
		},
		node
	)
	return {
		"regular_interval": conf_normalize_interval(node_["regular_interval"]),
		"attentive_interval": conf_normalize_interval(node_["attentive_interval"]),
		"reminding_interval": conf_normalize_interval(node_["reminding_interval"]),
	}
	

def conf_normalize_notification(
	notification_channel_implementations,
	node
):
	if (node["kind"] not in notification_channel_implementations):
		raise ValueError("invalid notification kind: %s" % notification["kind"])
	else:
		return {
			"kind": node["kind"],
			"parameters": notification_channel_implementations[node["kind"]].normalize_conf_node(node["parameters"]),
		}
	

def conf_normalize_defaults(
	notification_channel_implementations,
	node
):
	node_ = dict_merge(
		{
			"active": True,
			"threshold": 3,
			"annoy": False,
			"schedule": {
				"regular_interval": (60 * 60),
				"attentive_interval": (60 * 2),
				"reminding_interval": (60 * 60 * 24),
			},
			"notifications": [
			],
		},
		node
	)
	return {
		"active": node_["active"],
		"threshold": node_["threshold"],
		"annoy": node_["annoy"],
		"schedule": node_["schedule"],
		"notifications": list(
			map(
				lambda x: conf_normalize_notification(notification_channel_implementations, x),
				node_["notifications"]
			)
		),
	}
	

def conf_normalize_check(
	check_kind_implementations,
	notification_channel_implementations,
	defaults,
	node
):
	if ("name" not in node):
		raise ValueError("missing mandatory field in 'check' node: 'name'")
	else:
		if ("kind" not in node):
			raise ValueError("missing mandatory field in 'check' node: 'kind'")
		else:
			if (node["kind"] not in check_kind_implementations):
				raise ValueError("invalid check kind: '%s'" % node["kind"])
			else:
				node_ = dict_merge(
					dict_merge(
						defaults,
						{
							"title": node["name"],
							"parameters": {},
							"custom": None,
						},
					),
					node
				)
				node__ = {}
				if True:
					node__["name"] = node_["name"]
				if True:
					node__["title"] = node_["title"]
				if ("active" in node_):
					node__["active"] = node_["active"]
				if ("threshold" in node_):
					node__["threshold"] = node_["threshold"]
				if ("annoy" in node_):
					node__["annoy"] = node_["annoy"]
				if ("schedule" in node_):
					node__["schedule"] = conf_normalize_schedule(node_["schedule"])
				if ("notifications" in node_):
					node__["notifications"] = list(
						map(
							lambda x: conf_normalize_notification(notification_channel_implementations, x),
							node_["notifications"]
						)
					)
				if ("kind" in node_):
					node__["kind"] = node_["kind"]
				if True:
					node__["parameters"] = check_kind_implementations[node_["kind"]].normalize_conf_node(node_["parameters"])
				if ("custom" in node_):
					node__["custom"] = node_["custom"]
				return node__
	

def conf_normalize_root(
	check_kind_implementations,
	notification_channel_implementations,
	node,
	options = None
):
	options = dict_merge(
		{
			"use_implicit_default_values": True,
		},
		({} if (options is None) else options)
	)
	counts = {}
	checks_raw = (
		node["checks"]
		if ("checks" in node) else
		[]
	)
	for node_ in checks_raw:
		if (node_["name"] not in counts):
			counts[node_["name"]] = 0
		counts[node_["name"]] += 1
	fails = list(filter(lambda pair: (pair[1] > 1), counts.items()))
	if (len(fails) > 0):
		raise ValueError(
			string_coin(
				"ambiguous check names: {{names}}",
				{
					"names": ",".join(map(lambda pair: pair[0], fails)),
				}
			)
		)
	else:
		defaults_raw = (
			node["defaults"]
			if ("defaults" in node) else
			{}
		)
		defaults = (
			conf_normalize_defaults(
				notification_channel_implementations,
				defaults_raw
			)
			if options["use_implicit_default_values"] else
			defaults_raw
		)
		includes = (
			node["includes"]
			if ("includes" in node) else
			[]
		)
		return {
			"defaults": defaults,
			"includes": includes,
			"checks": list(
				map(
					lambda node_: conf_normalize_check(
						check_kind_implementations,
						notification_channel_implementations,
						defaults,
						node_
					),
					checks_raw
				)
			)
		}
	

def conf_load(
	check_kind_implementations,
	notification_channel_implementations,
	path,
	options = None
):
	options = dict_merge(
		{
			"root": True,
			"already_included": set([]),
		},
		({} if (options is None) else options)
	)
	if (path in options["already_included"]):
		raise ValueError("circular dependency detected")
	else:
		conf_raw = _json.loads(file_read(path))
		includes = (
			conf_raw["includes"]
			if ("includes" in conf_raw) else
			[]
		)
		for index in range(len(includes)):
			path_ = includes[index]
			sub_conf = conf_load(
				check_kind_implementations,
				notification_channel_implementations,
				(
					path_
					if _os.path.isabs(path_) else
					_os.path.join(_os.path.dirname(path), path_)
				),
				{
					"root": False,
					"already_included": (options["already_included"] | {path})
				}
			)
			if (not "checks" in conf_raw):
				conf_raw["checks"] = []
			conf_raw["checks"].extend(
				list(
					map(
						lambda check: dict_merge(
							check,
							{
								"name": string_coin(
									"{{prefix}}.{{original_name}}",
									{
										"prefix": _os.path.basename(path_).split(".")[0],
										"original_name": check["name"],
									}
								),
							}
						),
						sub_conf["checks"]
					)
				)
			)
			conf_raw["includes"] = []
		return conf_normalize_root(
			check_kind_implementations,
			notification_channel_implementations,
			conf_raw,
			{
				"use_implicit_default_values": options["root"],
			}
		)