libservice 1.0.3

Library for building InfraSonar services

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:

Environment variable	Default	Description
THINGSDB_HOSTLIST	thingsdb:9200	ThingsDB host list.
THINGSDB_TOKEN	empty	Token for authentication (required).
THINGSDB_SCOPE	//data	Collection scope for data.
HUB_HOST	hub	Hub host
HUB_PORT	8700	Hub port
SLEEP_TIME	2	Sleep time in seconds in each iteration.
LOG_LEVEL	warning	Log level (error, warning, info, debug).
LOG_COLORIZED	0	Either 0 (=disabled) or 1 (=enabled).
LOG_FMT	%y%m...	Default format is %y%m%d %H:%M:%S.
DRY_RUN	empty	If enabled, result data will be printed to stdout instead of send to the hub.

Usage

from asyncio import AbstractEventLoop
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[
            dict | None, dict | None]:
        # 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 or to return an error together with the result.
        # For example:
        #
        #   return state, CheckException('Incomplete result').to_dict()
        #
        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,
      no_count=False)  # When True, the check(s) do not count (counter + lastseen)

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()

Check is an asset Id is scheduled

In some cases, the dmarc service as an example, you might want to check if an asset Id is scheduled and ask for the container Id. As the Asset() instance is only available during the check process, you need to verify this in a different way.

from libservice.serviceroom import service_room

container_id = service_room.get_container_id(asset_id)
if container_id is None:
    # missing asset ID
    ...