Python library for building InfraSonar Services

This library is created for building InfraSonar services.

Environment variable

You might want to implement configuration which applies for all assets in all containers, but still don't want to hard-code this setting in your check. Think for example about a API URL which is always the same, except maybe for a development environment.

The following environment variable are required for a service to work and are always set by InfraSonar:

Usage

from asyncio import AbstractEventLoop
from typing import Tuple, Optional
from libservice import start, Asset, CheckBase


class MyCheck(CheckBase):
    # Use CheckBaseMulti if you want to perform checks for multiple assets
    # combined. Sometimes this can be useful as you might be able to combine
    # multiple assets in a single request.
    key = 'my_check'

    @classmethod
    async def run(cls, ts: float, asset: Asset) -> Tuple[
            Optional[dict], Optional[dict]]:
        # Return with the state and optionally an error dict which can be
        # created using CheckException(my_error_message).to_dict().
        # Alternatively, you can rase a CheckException. The return error is
        # especially useful with CheckBaseMulti where only a single asset might
        # fail.
        return {
          'my_type': [
            {'name': 'my_item'}
          ]
        }, None


def start_func(loop: AbstractEventLoop):
    pass  # optional init function

def close_func(loop: AbstractEventLoop):
    pass  # optional close function


if __name__ == '__main__':
    start(
      collector_key='my_server',
      version='0.1.0',
      checks=(MyCheck, ),
      start_func=start_func,
      close_func=close_func)

ASCII item names

InfraSonar requires each item to have a unique name property. The value for name must be a string with ASCII compatible character. When your name is not guaranteed to be ASCII compatible, the following code replaces the incompatible characters with question marks (?):

name = name.encode('ascii', errors='replace').decode()