# Luna Streams v.0.6.21 documentation

## Monitoring

## Data for monitoring

Now we monitor two types of events for monitoring: request and error. First type is all requests, second is failed requests only. Every event is a point in the time series. The point is represented as union of the following data:

> - series name (now requests and errors)

> - start request time

> - tags, indexed data in storage, dictionary: keys - string tag names, values - string, integer, float

> - fields, non indexed data in storage, dictionary: keys - string tag names, values - string, integer, float

‘Requests’ series. Triggered on every request. Each point contains a data about corresponding request (execution time and etc).

> - tags

>   > tag name | description
>   > --- | ---
>   > service | always “luna-streams”
>   > route | concatenation of a request method and a request resource (POST:/streams)
>   > status_code | http status code of response

> - fields

>   > fields | description
>   > --- | ---
>   > request_id | request id
>   > execution_time | request execution time

‘Errors’ series. Triggered on failed request. Each point contains error_code of luna error.

> - tags

>   > tag name | description
>   > --- | ---
>   > service | always “luna-streams”
>   > route | concatenation of a request method and a request resource (POST:/streams)
>   > status_code | http status code of response
>   > error_code | luna error code

> - fields

>   > fields | description
>   > --- | ---
>   > request_id | request id

‘Licensing’ series. Triggered when the service starts and every 60 seconds. Each point contains license check data.

> - tags

>   > tag name | description
>   > service | always “luna-streams”
>   > license_status | license status (“ok”, “warning”, “error”, “exception”)

> - fields

>   > fields | description
>   > license_streams_limit_rate | the percentage of used streams
>   > warnings | license warning messages
>   > errors | license error messages

## Database

Monitoring is implemented as data sending to an influx database. You can setup your database credentials in configuration file in section “monitoring”.

## Classes

Module contains points for monitoring’s.

class luna_streams.crutches_on_wheels.cow.monitoring.points.BaseMonitoringPoint(eventTime)

  Abstract class for points

  eventTime

    event time as timestamp

    Type:

      float

  abstract property fields: Dict[str, int | float | str]

    Get tags from point. We supposed that fields are not indexing data

    Return type:

      `Dict`[`str`, `Union`[`int`, `float`, `str`]]

    Returns:

      dict with fields.

  abstract property tags: Dict[str, int | float | str]

    Get tags from point. We supposed that tags are indexing data

    Return type:

      `Dict`[`str`, `Union`[`int`, `float`, `str`]]

    Returns:

      dict with tags.

class luna_streams.crutches_on_wheels.cow.monitoring.points.BaseRequestMonitoringPoint(requestId, resource, method, requestTime, service, statusCode)

  Base class for point which is associated with requests.

  requestId

    request id

    Type:

      str

  route

    concatenation of a request method and a request resource

    Type:

      str

  service

    service name

    Type:

      str

  requestTime

    a request processing start timestamp

    Type:

      float

  statusCode

    status code of a request response.

    Type:

      int

  property fields: Dict[str, int | float | str]

    Get fields

    Returns:

      “request_id”

    Return type:

      dict with following keys

  property tags: Dict[str, int | float | str]

    Get tags

    Returns:

      “route”, “service”, “status_code”

    Return type:

      dict with following keys

class luna_streams.crutches_on_wheels.cow.monitoring.points.DataForMonitoring(tags=<factory>, fields=<factory>)

  Class fo storing an additional data for monitoring.

class luna_streams.crutches_on_wheels.cow.monitoring.points.RequestErrorMonitoringPoint(requestId, resource, method, errorCode, service, requestTime, statusCode, additionalTags=None, additionalFields=None)

  Request monitoring point is suspended for monitoring requests errors (error codes)

  errorCode

    error code

    Type:

      int

  additionalTags

    additional tags which was specified for the request

    Type:

      dict

  additionalFields

    additional fields which was specified for the request

    Type:

      dict

  property fields: Dict[str, int | float | str]

    Get fields.

    Return type:

      `Dict`[`str`, `Union`[`int`, `float`, `str`]]

    Returns:

      dict with base fields and additional tags

  series: str = 'errors'

    series “errors”

  property tags: Dict[str, int | float | str]

    Get tags.

    Return type:

      `Dict`[`str`, `Union`[`int`, `float`, `str`]]

    Returns:

      dict with base tags, “error_code” and additional tags

class luna_streams.crutches_on_wheels.cow.monitoring.points.RequestMonitoringPoint(requestId, resource, method, executionTime, requestTime, service, statusCode, additionalTags=None, additionalFields=None)

  Request monitoring point is suspended for monitoring all requests and measure a request time and etc.

  executionTime

    execution time

    Type:

      float

  additionalTags

    additional tags which was specified for the request

    Type:

      dict

  additionalFields

    additional fields which was specified for the request

    Type:

      dict

  property fields: Dict[str, int | float | str]

    Get fields.

    Return type:

      `Dict`[`str`, `Union`[`int`, `float`, `str`]]

    Returns:

      dict with base fields, “execution_time” and additional tags

  series: str = 'requests'

    series “request”

  property tags: Dict[str, int | float | str]

    Get tags.

    Return type:

      `Dict`[`str`, `Union`[`int`, `float`, `str`]]

    Returns:

      dict with base tags and additional tags

luna_streams.crutches_on_wheels.cow.monitoring.points.getRoute(resource, method)

  Get a request route, concatenation of a request method and a request resource :type resource: `str` :param resource: resource :type method: `str` :param method: method

  Returns:

    {resource}”

  Return type:

    “{method}

luna_streams.crutches_on_wheels.cow.monitoring.points.monitorTime(monitoringData, fieldName)

  Context manager for timing execution time.

  Parameters:

    - monitoringData (`DataForMonitoring`) – container for saving result

    - fieldName (`str`) – field name

Module implement base class for monitoring

class luna_streams.crutches_on_wheels.cow.monitoring.base_monitoring.BaseLunaMonitoring

  Base class for monitoring

  abstract flushPoints(points)

    Flush point to monitoring.

    Parameters:

      points (`List`[`BaseMonitoringPoint`]) – point

    Return type:

      `None`

class luna_streams.crutches_on_wheels.cow.monitoring.base_monitoring.LunaRequestInfluxMonitoring(credentials, host='localhost', port=8086, ssl=False, flushingPeriod=1)

  Class for sending data which is associated with request to influx .. attribute:: settings

  > influxdb settings

  > type:

  >   InfluxSettings

  flushingPeriod

    period of flushing points (in seconds)

    Type:

      int

  async close()

    Stop monitoring.

    Return type:

      `None`

  flushPoints(points)

    Flush point to influx.

    Parameters:

      points (`Iterable`[`BaseMonitoringPoint`]) – point

    Return type:

      `None`

  async initialize()

    Initialize monitoring

    Return type:

      `None`

  static prepareSettings(credentials, host, port, ssl)

    Prepare influxdb settings :type credentials: ~T_INFLUX_CREDENTIALS :param credentials: database credentials :type host: `str` :param host: influx host :type port: `int` :param port: influx port :type ssl: `bool` :param ssl: use or not ssl for connecting to influx

    Return type:

      `InfluxSettings`

    Returns:

      influxdb settings container

Module contains classes for sending a data to an influx monitoring.

class luna_streams.crutches_on_wheels.cow.monitoring.influx_adapter.BaseMonitoringAdapter(settings, flushingPeriod)

  Base monitoring adapter.

  backgroundScheduler

    runner for periodic flushing monitoring points

    Type:

      AsyncIOScheduler

  _buffer

    list of buffering points which is waiting sending to influx

    Type:

      List[BaseRequestMonitoringPoint]

  flushingPeriod

    period of flushing points (in seconds)

    Type:

      float

  logger

    logger

    Type:

      Logger

  _influxSettings

    current influx settings

    Type:

      InfluxSettings

  _job

    sending monitoring data job

    Type:

      Job

  addPointsToBuffer(points)

    Add points to buffer.

    Parameters:

      points (`Iterable`[`BaseMonitoringPoint`]) – points

    Return type:

      `None`

  static convertFieldsToInfluxLineProtocol(fields)

    Convert field value to influx line protocol format

    Parameters:

      fields (`dict`) – dict with values to convert

    Retruns:

      line protocol string

    Return type:

      `str`

  generatePointStr(point)

    Generate string from point

    Parameters:

      point (`BaseMonitoringPoint`) – point

    Return type:

      `str`

    Returns:

      influx line protocol string

  initializeScheduler()

    Start the loop for sending data from the buffer to monitoring.

    Return type:

      `None`

  stopScheduler()

    Stop monitoring.

    Return type:

      `None`

  updateFlushingPeriod(newPeriod)

    Update flushing period :type newPeriod: `int` :param newPeriod: new period

class luna_streams.crutches_on_wheels.cow.monitoring.influx_adapter.InfluxMonitoringAdapter(settings, flushingPeriod)

  Influx 2.x adaptor. Suspended to send points to an influxdb

  bucket

    influx bucket name

    Type:

      str

  initializeMonitoring()

    Initialize monitoring.

    Return type:

      `None`

  async stopMonitoring()

    Stop monitoring (cancel all request and stop getting new).

    Return type:

      `None`

class luna_streams.crutches_on_wheels.cow.monitoring.influx_adapter.InfluxSettings(url, bucket, organization, token, ssl)

  Container for influx 2.x settings