Skip to main content

Health Check API for multiprocessing python apps

Project description

A Health Check API Library for Multiprocessing Python Apps

license

This library adds a health check REST API to your multiprocessing apps. You can add health check function calls to your functions and library will track the function calls. This library supports multiprocessing threads. You can fetch a single overall app status by fetching http://<ip>:<port>/healthcheck, a single overall app liveness by fetching http://<ip>:<port>/liveness.

Influenced by prometheus client mp exporter. Health check functions will write healthy and liveness results <pid>.json file located in directory defined by PY_HEALTH_MULTIPROC_DIR. If the directory doesn't exist, health checks won't work.

Please clear the directory content before running your app. REST API

Each health check class will be run every 10 seconds by default. You can change this value by setting PY_HEALTH_RUN_PERIOD.

Usage

You can register your functions with add_check() decorator. You can set a timeout for your functions with set_timeout() if you process needs to check in regularly.

import multiprocessing as mp
import time

import healthcheck_python as hp


class P1(mp.Process):
	def __init__(self, timeout=0):
		super().__init__()
		self._stop_bit = mp.Event()
		self.timeout = timeout

	def close(self) -> None:
		self._stop_bit.set()

	def healthcheck(self):
		return True, "Healthcheck is OK"

	def do_something(self):
		time.sleep(5)

	def run(self):
		hp.init_check(timeout=self.timeout)
		hp.add_check(self.healthcheck)

		hp.live()
		while not self._stop_bit.is_set():
			hp.healthy()
			self.do_something()


hp.start_http_server(port=8080)

p1 = P1(timeout=10)
p1.start()

p2 = P1()
p2.start()

time.sleep(30)

p1.close()
p2.close()

p1.join()
p2.join()

It can also be used with threading. You need to create separate Healthcheck instances and start instances in your threads.

import multiprocessing as mp
import threading
import time

import healthcheck_python as hp
from healthcheck_python import create_health_check

HEALTHCHECK_P1 = create_health_check("P1", timeout=5)
HEALTHCHECK_P2 = create_health_check("P2")


class P1(threading.Thread):
	def __init__(self, timeout=0):
		super().__init__()
		self._stop_bit = mp.Event()
		self.timeout = timeout

	def close(self) -> None:
		self._stop_bit.set()

	def healthcheck(self):
		return True, "Healthcheck is OK"

	def do_something(self):
		time.sleep(5)

	def run(self):
		HEALTHCHECK_P1.add_check(self.healthcheck)
		HEALTHCHECK_P1.start()

		HEALTHCHECK_P1.live()
		while not self._stop_bit.is_set():
			HEALTHCHECK_P1.healthy()
			self.do_something()


class P2(threading.Thread):
	def __init__(self, timeout=0):
		super().__init__()
		self._stop_bit = mp.Event()
		self.timeout = timeout

	def close(self) -> None:
		self._stop_bit.set()

	def healthcheck(self):
		return True, "Healthcheck is OK"

	def do_something(self):
		time.sleep(5)

	def run(self):
		HEALTHCHECK_P2.add_check(self.healthcheck)
		HEALTHCHECK_P2.start()

		HEALTHCHECK_P2.live()
		while not self._stop_bit.is_set():
			HEALTHCHECK_P2.healthy()
			self.do_something()


hp.init_check(timeout=0)
hp.start_http_server(port=8080)

p1 = P1(timeout=10)
p1.start()

p2 = P2()
p2.start()

time.sleep(30)

p1.close()
p2.close()

p1.join()
p2.join()
$ curl http://localhost:8080/healthcheck
{"hostname": "my_app", "status": "success", "timestamp": 1684406235.474363, "results": [[{"checker": "healthcheck", "output": "Healthcheck is OK", "passed": true, "timestamp": 1684406230.9507005, "response_time": 5e-06}, {"checker": "P1", "output": "", "passed": true, "timestamp": 1684406230.9507082, "response_time": 0}]]}
$ curl http://localhost:8080/liveness
{"hostname": "my_app", "liveness": true, "timestamp": 1684406208.784097}

Project details


Download files

Download the file for your platform. If you're not sure which to choose, learn more about installing packages.

Source Distribution

healthcheck_python-1.2.1.tar.gz (10.4 kB view details)

Uploaded Source

Built Distribution

healthcheck_python-1.2.1-py3-none-any.whl (13.4 kB view details)

Uploaded Python 3

File details

Details for the file healthcheck_python-1.2.1.tar.gz.

File metadata

  • Download URL: healthcheck_python-1.2.1.tar.gz
  • Upload date:
  • Size: 10.4 kB
  • Tags: Source
  • Uploaded using Trusted Publishing? No
  • Uploaded via: poetry/1.6.1 CPython/3.11.5 Linux/6.7.4-200.fc39.x86_64

File hashes

Hashes for healthcheck_python-1.2.1.tar.gz
Algorithm Hash digest
SHA256 0a43c0ec307aa3e814a53cc2e1d2b70538c4e1bc161ede14cfca73206912646c
MD5 b28fde4264f1a7bbdc0df2433c454890
BLAKE2b-256 730cbed99f02e0529c0e95f6e4a96f1d82bafda50091c0d119df1e3e7922659d

See more details on using hashes here.

File details

Details for the file healthcheck_python-1.2.1-py3-none-any.whl.

File metadata

  • Download URL: healthcheck_python-1.2.1-py3-none-any.whl
  • Upload date:
  • Size: 13.4 kB
  • Tags: Python 3
  • Uploaded using Trusted Publishing? No
  • Uploaded via: poetry/1.6.1 CPython/3.11.5 Linux/6.7.4-200.fc39.x86_64

File hashes

Hashes for healthcheck_python-1.2.1-py3-none-any.whl
Algorithm Hash digest
SHA256 fe6b9ac6f6872f3fbabfffb0a4332351fc3d545e33f866b9bcf418edacb7018c
MD5 06d989f0d5bd45f4f6851cc6fb2740fc
BLAKE2b-256 751c2036140f14d5306b60191aacea227ea4ca864939219223e42570a1555159

See more details on using hashes here.

Supported by

AWS Cloud computing and Security Sponsor Datadog Monitoring Fastly CDN Google Download Analytics Pingdom Monitoring Sentry Error logging StatusPage Status page