core/source/logic/main.py

def main():
	## setup translation for the first time
	translation_initialize("en", env_get_language())
	
	## args
	argumentparser = _argparse.ArgumentParser(
		description = translation_get("help.title"),
		formatter_class = _argparse.ArgumentDefaultsHelpFormatter
	)
	argumentparser.add_argument(
		type = str,
		default = "monitoring.hmdl.json",
		dest = "order_path",
		metavar = "<order-path>",
		help = translation_get("help.args.order_path"),
	)
	argumentparser.add_argument(
		"-x",
		"--erase-state",
		action = "store_true",
		default = False,
		dest = "erase_state",
		help = translation_get("help.args.erase_state"),
	)
	argumentparser.add_argument(
		"-s",
		"--show-schema",
		action = "store_true",
		default = False,
		dest = "show_schema",
		help = translation_get("help.args.show_schema"),
	)
	argumentparser.add_argument(
		"-e",
		"--expose-full-order",
		action = "store_true",
		default = False,
		dest = "expose_full_order",
		help = translation_get("help.args.expose_full_order"),
	)
	### v conf stuff v
	argumentparser.add_argument(
		"-d",
		"--database-path",
		type = str,
		default = None,
		dest = "database_path",
		metavar = "<database-path>",
		help = translation_get("help.args.database_path"),
	)
	argumentparser.add_argument(
		"-m",
		"--mutex-path",
		type = str,
		default = "/tmp/heimdall.lock",
		dest = "mutex_path",
		metavar = "<mutex-path>",
		help = translation_get("help.args.mutex_path"),
	)
	argumentparser.add_argument(
		"-y",
		"--send-ok-notifications",
		action = "store_true",
		default = False,
		dest = "send_ok_notifications",
		help = translation_get("help.args.send_ok_notifications", {"condition_name": translation_get("conditions.ok")}),
	)
	argumentparser.add_argument(
		"-l",
		"--language",
		type = str,
		choices = localization_data.keys(),
		default = None,
		dest = "language",
		metavar = "<language>",
		help = translation_get("help.args.language"),
	)
	argumentparser.add_argument(
		"-t",
		"--time-to-live",
		type = int,
		default = (60 * 60 * 24 * 7),
		dest = "time_to_live",
		metavar = "<time-to-live>",
		help = translation_get("help.args.time_to_live"),
	)
	args = argumentparser.parse_args()
	
	## vars
	id_ = _hashlib.sha256(_os.path.abspath(args.order_path).encode("ascii")).hexdigest()[:8]
	database_path = (
		args.database_path
		if (args.database_path is not None) else
		_os.path.join(
			_tempfile.gettempdir(),
			string_coin("monitoring-state-{{id}}.sqlite", {"id": id_})
		)
	)
	
	## exec
	
	### setup translation for the second time
	if (args.language is not None):
		translation_initialize("en", args.language)
	
	### load check kind implementations
	check_kind_implementations = {
		"script": implementation_check_kind_script(),
		"file_state": implementation_check_kind_file_state(),
		"tls_certificate": implementation_check_kind_tls_certificate(),
		"http_request": implementation_check_kind_http_request(),
		"generic_remote" : implementation_check_kind_generic_remote(),
	}
	
	### load notification channel implementations
	notification_channel_implementations = {
		"console": implementation_notification_channel_console(),
		"email": implementation_notification_channel_email(),
		"libnotify": implementation_notification_channel_libnotify(),
	}
	
	if (args.show_schema):
		_sys.stdout.write(
			_json.dumps(
				order_schema_root(
					check_kind_implementations,
					notification_channel_implementations
				),
				indent = "\t"
			)
			+
			"\n"
		)
	else:
		### get order data
		order = order_load(
			check_kind_implementations,
			notification_channel_implementations,
			_os.path.abspath(args.order_path)
		)
		
		if (args.expose_full_order):
			_sys.stdout.write(_json.dumps(order, indent = "\t") + "\n")
			_sys.exit(1)
		else:
			_sys.stderr.write(
				string_coin(
					"[info] {{label}}: {{path}}\n",
					{
						"label": translation_get("misc.state_file_path"),
						"path": database_path,
					}
				)
			)
			
			### mutex check
			if (_os.path.exists(args.mutex_path)):
				_sys.stderr.write(
					string_coin(
						"[error] {{message}} ({{path}})\n",
						{
							"message": translation_get("misc.still_running"),
							"path": args.mutex_path,
						}
					)
				)
				_sys.exit(2)
			else:
				### setup database
				repository_result_setup(database_path)
				
				### clean database
				count = repository_result_clean(database_path, args.time_to_live, args.erase_state)
				_sys.stderr.write(
					string_coin(
						"[info] {{text}}\n",
						{
							"text": translation_get(
								"misc.cleanup_info",
								{
									"count": ("%u" % count),
								}
							),
						}
					)
				)
				
				file_write(args.mutex_path, "", {"append": True})
				
				### iterate through checks
				for check_data in order["checks"]:
					if (not check_data["active"]):
						pass
					else:
						### get old state and examine whether the check shall be executed
						old_item_state = repository_result_scan(
							database_path,
							check_data["name"],
							check_data["threshold"]
						)
						
						timestamp = get_current_timestamp()
						due = (
							(old_item_state is None)
							or
							(old_item_state["condition"] != enum_condition.ok)
							or
							((timestamp - old_item_state["timestamp"]) >= check_data["schedule"]["regular_interval"])
							or
							(
								(old_item_state["count"] is not None)
								and
								((timestamp - old_item_state["timestamp"]) >= check_data["schedule"]["attentive_interval"])
							)
						)
						if (not due):
							pass
						else:
							_sys.stderr.write(
								string_coin(
									"-- {{check_name}}\n",
									{
										"check_name": check_data["name"],
									}
								)
							)
							
							### execute check and set new state
							try:
								result = check_kind_implementations[check_data["kind"]].run(check_data["parameters"])
							except Exception as error:
								result = {
									"condition": enum_condition.unknown,
									"info": {
										"cause": translation_get("misc.check_procedure_failed"),
										"error": str(error),
									},
								}
							count = (
								1
								if (
									(old_item_state is None)
									or
									(old_item_state["condition"] != result["condition"])
								) else
								(
									(old_item_state["count"] + 1)
									if (
										(old_item_state["count"] is not None)
										and
										((old_item_state["count"] + 1) <= check_data["threshold"])
									) else
									None
								)
							)
							shall_send_notification = (
								(
									(
										(count is not None)
										and
										(count == check_data["threshold"])
									)
									or
									(
										(count is None)
										and
										check_data["annoy"]
									)
									or
									(
										(count is None)
										and
										(
											(old_item_state is not None)
											and
											(old_item_state["last_notification_timestamp"] is not None)
											and
											(check_data["schedule"]["reminding_interval"] is not None)
											and
											(
												(timestamp - old_item_state["last_notification_timestamp"])
												>=
												check_data["schedule"]["reminding_interval"]
											)
										)
									)
								)
								and
								(
									(result["condition"] != enum_condition.ok)
									or
									args.send_ok_notifications
								)
							)
							new_item_state = {
								"timestamp": timestamp,
								"condition": result["condition"],
								"count": count,
								"last_notification_timestamp": (
									timestamp
									if shall_send_notification else
									(
										None
										if (old_item_state is None) else
										old_item_state["last_notification_timestamp"]
									)
								),
							}
							repository_result_add(
								database_path,
								{
									"check_name": check_data["name"],
									"timestamp": timestamp,
									"condition": result["condition"],
									"notification_sent": shall_send_notification,
									"info": result["info"],
								}
							)
							
							### send notifications
							if (not shall_send_notification):
								pass
							else:
								for notification in check_data["notifications"]:
									notification_channel_implementations[notification["kind"]].notify(
										notification["parameters"],
										check_data["name"],
										check_data,
										new_item_state,
										dict_merge(
											(
												{}
												if (check_data["custom"] is None) else
												{"custom": check_data["custom"]}
											),
											result["info"]
										)
									)
				
				if (not _os.exists(args.mutex_path)):
					pass
				else:
					_os.remove(args.mutex_path)
		

main()