Plugins

The service supports the system of plugins. Plugins must be written in the Python programming language.

Plugin types

There are two sorts of plugins:

• On event plugin. The plugin is triggered when an event occurs. The plugin should implement a callback function. This function is called on each event of the corresponding type. The set of event types is defined by the service developers.

  event type	description
  monitoring_event	Event contains monitoring points for sending to a custom monitoring system
  sending_event	Event contains handler-generated event data for sending to the third-party source

  Monitoring plugin example:

  This plugin demonstrates the sending of a request monitoring data to another service. All monitoring plugins must implement the BaseRequestMonitoringPlugin abstract class.

  Event sending plugin example:

  from abc import abstractmethod
  from typing import List, Optional, Union
  
  from aiohttp import ClientSession
  
  from classes.event import Event
  from classes.schemas.event_raw import RawEvent
  from crutches_on_wheels.cow.plugins.plugins_meta.base_plugins import BaseEventPlugin
  from crutches_on_wheels.cow.utils.log import Logger
  from crutches_on_wheels.cow.web.application import LunaApplication
  
  
  class BaseEventSendingPlugin(BaseEventPlugin):
      """Base class for event sending."""
  
      # event name for triggering callback
      eventName = "sending_event"
  
      @abstractmethod
      async def sendEvents(
          self,
          events: Union[List[Event], List[RawEvent]],
          handlerId: str,
          accountId: str,
          requestId: str,
          createEventTime: str,
          endEventTime: str,
          logger: Logger,
      ) -> None:
          """
          Callback that is triggered on every success request to handlers.
  
          Args:
              events: event list
              handlerId: handler id
              accountId: account id
              requestId: request id
              createEventTime: event creation time
              endEventTime: event end time
              logger: logger
  
          """
  
      async def handleEvent(
          self,
          events: Union[List[Event], List[RawEvent]],
          handlerId: str,
          accountId: str,
          requestId: str,
          createEventTime: str,
          endEventTime: str,
          logger: Logger,
      ):
          """
          Handle events.
          Args:
              events: event list
              handlerId: handler id
              accountId: account id
              requestId: request id
              createEventTime: event creation time
              endEventTime: event end time
              logger: logger
  
          """
          await self.sendEvents(events, handlerId, accountId, requestId, createEventTime, endEventTime, logger)
  
  
  class EventSendingPlugin(BaseEventSendingPlugin):
      """Sends events to the third-party source. Only one instance of this class exist during the program execution."""
  
      def __init__(self, app: LunaApplication):
          super().__init__(app)
          self.url = "http://127.0.0.1:5020/1/buckets"
          self.session: Optional[ClientSession] = None
          self.bucket = "plugin_test_bucket"
  
      async def close(self):
          """Stop plugin. Close all open connections and etc."""
          if self.session:
              await self.session.close()
  
      async def initialize(self):
          """Initialize plugin."""
          self.session = ClientSession()
          async with self.session.post(f"{self.url}?bucket={self.bucket}") as resp:
              if resp.status not in (201, 409):
                  response = await resp.json()
                  raise RuntimeError(f"failed create bucket, {self.bucket}, response: {response}")
  
      async def sendEvents(
          self,
          events: Union[List[Event], List[RawEvent]],
          handlerId: str,
          accountId: str,
          requestId: str,
          createEventTime: str,
          endEventTime: str,
          logger: Logger,
      ) -> None:
  
          logger.debug(
              f"Plugin 'EventsOnFinishExampleClass' get events, request_id: {requestId}, "
              f"event_create_time: {createEventTime}, event_end_time: {endEventTime}"
          )
          prepareEvents = []
          for event in events:
              if isinstance(event, Event):
                  serializationEvent = event.asDict()
              else:
                  serializationEvent = event.asHandlerEventDict()
              prepareEvents.append(serializationEvent)
          msg = {
              "handler_id": handlerId,
              "account_id": accountId,
              "Luna-Request-Id": requestId,
              "events": prepareEvents,
              "event_create_time": createEventTime,
              "event_end_time": endEventTime,
          }
          async with ClientSession() as session:
              async with session.post(f"{self.url}/{self.bucket}/objects", json=msg) as resp:
                  logger.debug(resp.status)
                  logger.debug(await resp.text())
  

  This plugin demonstrates the sending of a handler generates event data to the third-party source. All event sending plugins must implement the BaseEventSendingPlugin abstract class.

• Background plugin. This sort of plugin is intended for background work.

  The background plugin can implement:

  • custom route

  • background monitoring of service resources

  • collaboration of an event plugin and a background plugin (batching monitoring points)

  • connection to other data sources (Redis, RabbitMQ) and their data processing

  Plugin example:

  This plugin demonstrates background work and implements a route. All background plugins must implement the BaseRequestMonitoringPlugin abstract class.

Enable plugin

If the user implements a plugin, the file with the plugin should be added to the luna_handlers/plugins directory of the service. The plugin filename should be added to the LUNA_HANDLERS_ACTIVE_PLUGINS configuration setting.

LUNA_HANDLERS_ACTIVE_PLUGINS = [event_sending_plugin_example]

List should contains filenames without extension (.py).