Commit c27408ac authored by Stuart Fisher's avatar Stuart Fisher
Browse files

initial commit

parents
Pipeline #45926 failed with stages
in 54 seconds
# Byte-compiled / optimized / DLL files
__pycache__/
*.py[cod]
*$py.class
.DS_Store
\#*\#
\.\#*
*~
*.bck
build
dist
*.swp
.coverage
.coverage*
# for eclipse, vscode and other IDE users
.settings/
.project
.pydevproject
.cache/
*.egg-info/
.eggs
.pytest_cache/
.vscode/*
*.so
stages:
- style
- tests
- deploy
services:
- esrfbcu/daiquiri-testdb:1.19
check_style:
stage: style
image: continuumio/miniconda3:latest
script:
- pip install black==19.10b0
- LC_ALL=C.UTF-8 black --check --safe .
check_lint:
stage: style
image: continuumio/miniconda3:latest
script:
- pip install flake8
- git diff -U0 origin/master...$CI_COMMIT_SHA | flake8 --diff
allow_failure: true
.template_test:
stage: tests
image: continuumio/miniconda3:latest
script:
- conda create -q --yes --name test
- source activate test
- pip install .
- pytest $PYTEST_ARGS
test:
extends: .template_test
except:
- master
test_cov:
extends: .template_test
only:
- master
artifacts:
when: always
paths:
- htmlcov
expire_in: 7 days
variables:
PYTEST_ARGS: '--cov status_monitor --cov-report html:htmlcov'
pages:
stage: deploy
before_script:
- ''
script:
- mkdir public
- mv htmlcov public/coverage
artifacts:
paths:
- public
expire_in: 7 days
only:
- tags
- master
# Infrastructure Monitoring
Monitor infrastructure using simple tests
* Ping a host
* Make a http request
* Check a database is available
* Check zocalo services are available
Test results are outputted in json format to be ingested in a basic web status page
## Installation
```bash
pip install -e .
cp examples/tests.yml tests.yml
```
## Configuration
Define some tests:
Ping a host:
```yaml
mygroup:
- type: ping
host: localhost
name: Ping localhost
```
Make a request:
```yaml
mygroup:
- type: request
url: https://localhost
name: Test localhost server
# optionally allow this test to fail
allow_fail: true
```
db:
- type: ping
host: localhost
name: Ping db host
- type: ispyb
config: config/ispyb-sp.cfg
proposal_code: blc
proposal_number: '00001'
name: Test db query
zocalo:
- type: zocalo
host: localhost:61613
name: Test zocalo services
services:
- Dispatcher
host:
- type: request
url: https://localhost:8080/api/components/config
name: Test daiquiri server
allow_fail: true
[aliases]
test = pytest
[flake8]
ignore = E501,W503,E203,W504,E251,E262,E265,E266,W291,W293
#!/usr/bin/env python
from setuptools import setup
requirements = ["zocalo", "ispyb", "setuptools", "pyyaml", "marshmallow"]
setup_requirements = []
test_requirements = ["pytest"]
setup(
author="ESRF BCU",
author_email="bcu@esrf.fr",
classifiers=[
"Development Status :: 3 - Alpha",
"Intended Audience :: Developers",
"License :: OSI Approved :: BSD License",
"Natural Language :: English",
"Programming Language :: Python :: 3",
"Programming Language :: Python :: 3.7",
"Programming Language :: Python :: 3.8",
"Programming Language :: Python :: 3.9",
],
description="Infrastructure status monitor",
entry_points={"console_scripts": ["status_monitor = status_monitor.run:run"]},
install_requires=requirements,
license="BSD license",
include_package_data=True,
keywords="infrastructure tests",
name="status-monitor",
packages=["status_monitor"],
package_dir={"": "src"},
python_requires=">=3.7",
setup_requires=setup_requirements,
test_suite="tests",
tests_require=test_requirements,
url="https://gitlab.esrf.fr/ui/status-monitor",
zip_safe=False,
)
import os
import importlib
import time
import json
from datetime import datetime
from dataclasses import asdict
import yaml
from status_monitor.tests.test import TestResult, TestResultSchema, TestStatus
def load_config():
config_yml = os.environ.get("STATUS_MONITOR_CONFIG", "examples/tests.yml")
if not os.path.exists(config_yml):
raise AttributeError(f"Cannot find config file: {config_yml}")
config = {}
with open(config_yml, "r") as stream:
config = yaml.safe_load(stream)
return config
def run():
config = load_config()
collected_tests = {}
collected = 0
for name, tests in config.items():
for config in tests:
try:
mod = importlib.import_module(f"status_monitor.tests.{config['type']}")
except ModuleNotFoundError:
print(f"Couldnt load module {config['type']}")
# raise
else:
if name not in collected_tests:
collected_tests[name] = []
cls = getattr(mod, config["type"].title())
inst = cls(config)
collected_tests[name].append(inst)
collected += 1
print(f"Collected {collected} tests")
results = []
failed = 0
allow_failed = 0
passed = 0
for group, test_instances in collected_tests.items():
for test in test_instances:
start = time.time()
status_code, response = test.run()
if status_code == 0:
passed += 1
else:
if test.allow_fail:
allow_failed += 1
allowed = "(allowed)" if test.allow_fail else ""
print(f"- Test failed {allowed} for: '{test.name}' in '{group}'")
print(f" {response}")
failed += 1
results.append(
TestResult(
success=TestStatus.SUCCESS if status_code == 0 else TestStatus.FAIL,
name=test.name,
group=group,
type=test.__class__.__name__.lower(),
time=datetime.now(),
duration=time.time() - start,
response=response,
)
)
output = {}
sch = TestResultSchema()
for result in results:
if result.group not in output:
output[result.group] = []
marshalled_output = sch.dump(asdict(result))
output[result.group].append(marshalled_output)
with open("output.json", "w") as out:
out.write(json.dumps(output, indent=2))
print(f"{passed} passed, {failed} failed ({allow_failed} allowed to fail)")
if failed > allow_failed:
exit(1)
from marshmallow import fields
import ispyb
from status_monitor.tests.test import Test, TestSchema
class ISPyBSchema(TestSchema):
config = fields.Str(required=True)
proposal_code = fields.Str(required=True)
proposal_number = fields.Str(required=True)
class Ispyb(Test):
schema = ISPyBSchema
def run(self):
try:
ispyb_config = self._config["config"]
i = ispyb.open(ispyb_config)
result = i.core.retrieve_proposal_title(
self._config["proposal_code"], self._config["proposal_number"]
)
if len(result):
return 0, str(result[0]["title"])
else:
return 1, "Proposal not found"
except Exception as e:
return 1, str(e)
import subprocess
import shlex
from marshmallow import fields
from status_monitor.tests.test import Test, TestSchema
class PingSchema(TestSchema):
host = fields.Str(required=True)
class Ping(Test):
schema = PingSchema
def run(self):
ping = subprocess.Popen(
shlex.split(f"ping -c 1 {self._config['host']}"),
stderr=subprocess.PIPE,
stdout=subprocess.PIPE,
)
out, err = ping.communicate()
if ping.returncode == 0:
return 0, ""
else:
return 1, err
import requests
from marshmallow import fields
from status_monitor.tests.test import Test, TestSchema
class RequestSchema(TestSchema):
url = fields.Url(required=True)
class Request(Test):
schema = RequestSchema
def run(self):
try:
resp = requests.get(
self._config["url"], verify=self._config.get("verify_ssl", True)
)
if resp.status_code == 200:
return 0, ""
else:
return 1, resp.status_code
except Exception as e:
return 1, str(e)
from enum import IntEnum
from dataclasses import dataclass
from datetime import datetime
from abc import abstractmethod, ABC
from marshmallow import Schema, ValidationError, INCLUDE, fields
class TestSchema(Schema):
pass
class Test(ABC):
schema = None
def __init__(self, config):
if self.schema:
try:
sch_inst = self.schema()
sch_inst.load(config, unknown=INCLUDE)
except ValidationError:
print(f"Error loading test configuration: {config}")
raise
else:
pass
self._config = config
@property
def name(self):
return self._config["name"]
@property
def allow_fail(self):
return self._config.get("allow_fail", False)
@abstractmethod
def run(self):
pass
class TestStatus(IntEnum):
SUCCESS = 1
FAIL = 0
@dataclass(frozen=True)
class TestResult:
success: TestStatus
type: str
name: str
group: str
response: str
time: datetime
duration: int
class TestResultSchema(Schema):
success = fields.Int()
type = fields.Str()
name = fields.Str()
response = fields.Str()
time = fields.DateTime()
duration = fields.Float()
class Meta:
datetimeformat = "%d-%m-%Y %H:%M:%S"
import sys
import time
import threading
from marshmallow import fields
import workflows
from zocalo.configuration import transport_from_config
from status_monitor.tests.test import Test, TestSchema
class ZocaloSchema(TestSchema):
host = fields.Str(required=True)
services = fields.List(fields.Str(), required=True)
timeout = fields.Int()
class Zocalo(Test):
schema = ZocaloSchema
def run(self):
try:
self._lock = threading.RLock()
self._services = {}
transport_from_config(env="test" if "--test" in sys.argv else "live")
transport = workflows.transport.lookup("stomp")()
transport.connect()
transport.subscribe_broadcast(
"transient.status", self.update_status, retroactive=True
)
timeout = self._config.get("timeout", 10)
time.sleep(timeout)
services = sorted(self._services.keys())
searched = sorted(self._config["services"])
missing = list(set(searched) - set(services))
if not missing:
return 0, ""
else:
extra = list(set(services) - set(searched))
return (
1,
f"Some services were not found wihtin timeout {timeout}: {missing}, extras found: {extra}",
)
except Exception as e:
return 1, str(e)
def update_status(self, header, message):
with self._lock:
self._services[message["serviceclass"]] = {
"workflows": message["workflows"],
"zocalo": message["zocalo"],
"statustext": message["statustext"],
}
[ispyb_mariadb_sp]
user = test
pw = test
host = daiquiri-testdb
port = 3306
db = test
reconn_attempts = 6
reconn_delay = 1
test:
- type: ping
host: google.com
name: Ping google
- type: request
url: http://google.com
name: Make request to google
- type: ispyb
config: tests/config/ispyb-sp.cfg
proposal_code: blc
proposal_number: '00001'
name: Test db query
\ No newline at end of file
import os
from status_monitor.run import run
def test_simple():
os.environ[
"STATUS_MONITOR_CONFIG"
] = f"{os.path.dirname(__file__)}/config/tests.yml"
run()
Supports Markdown
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment