blob: 5a30ff3b73e35a0a4c5fe6417d545f4e45d218b3 [file] [log] [blame]
# Copyright 2016 The Chromium OS Authors. All rights reserved.
# Use of this source code is governed by a BSD-style license that can be
# found in the LICENSE file.
from __future__ import print_function
import logging
import os
import threading
import time
import instalog_common # pylint: disable=W0611
from instalog import flow_policy
from instalog import json_utils
from instalog import plugin_sandbox
from instalog import plugin_base
from instalog.external.jsonrpclib import SimpleJSONRPCServer
# Fix for bug b/30904731: Import _strptime manually. Otherwise,
# threads may initially raise the exception `AttributeError: _strptime`.
import _strptime # pylint: disable=unused-import
# Possible daemon states.
STARTING = 'STARTING'
UP = 'UP'
STOPPING = 'STOPPING'
DOWN = 'DOWN'
class Instalog(plugin_sandbox.CoreAPI):
def __init__(self, node_id, data_dir, cli_hostname, cli_port, buffer_plugin,
input_plugins=None, output_plugins=None):
"""Constructor.
Args:
node_id: ID of this Instalog node.
data_dir: Path to Instalog's state directory. Plugin state directories
will be stored here.
cli_hostname: Hostname used for the CLI RPC server.
cli_port: Port used for the CLI RPC server.
buffer_plugin: Configuration dict for the buffer plugin. Keys should
consist of:
- plugin: Required, plugin module name.
- args: Optional, defines plugin arguments.
input_plugins: List of configuration dicts for input plugins.
Configuration dicts should be the same format as
that of buffer_plugin, with the addition of:
- targets: Optional, defines target plugins.
output_plugins: List of configuration dicts for output plugins.
- plugin: Required, plugin module name.
- args: Optional, defines plugin arguments.
- allow: Optional, defines flow policy allow rules.
- deny: Optional, defines flow policy deny rules.
- targets: Optional, defines target plugins.
"""
self._rpc_lock = threading.Lock()
self._state = DOWN
# Store the node ID.
self._node_id = node_id
# Ensure we have a working data directory.
self._data_dir = data_dir
if not os.path.exists(self._data_dir):
os.makedirs(self._data_dir)
# Create plugin sandboxes.
self._PreprocessConfigEntries(input_plugins, output_plugins)
self._buffer = self._ConfigEntryToSandbox(
plugin_base.BufferPlugin, 'buffer', buffer_plugin)
self._plugins = {}
self._plugins.update(self._ConfigEntriesToSandboxes(
plugin_base.InputPlugin, input_plugins))
self._plugins.update(self._ConfigEntriesToSandboxes(
plugin_base.OutputPlugin, output_plugins))
# Start the RPC server.
self._rpc_server = SimpleJSONRPCServer.SimpleJSONRPCServer(
(cli_hostname, cli_port))
self._rpc_server.register_function(self.IsUp)
self._rpc_server.register_function(self.Stop)
self._rpc_server.register_function(self.Inspect)
self._rpc_server.register_function(self.Flush)
self._rpc_thread = threading.Thread(target=self._rpc_server.serve_forever)
self._rpc_thread.start()
def _ShutdownRPCServer(self):
def ShutdownThread():
self._rpc_server.shutdown()
self._rpc_server.server_close()
t = threading.Thread(target=ShutdownThread)
t.start()
def _PreprocessConfigEntries(self, input_plugins, output_plugins):
"""Preprocesses config entries to allow the "targets" argument."""
# Ensure that plugin IDs don't overlap across input and output.
if any([plugin_id in output_plugins for plugin_id in input_plugins]):
raise ValueError
# Next, convert 'targets' entries to corresponding allow policy rule.
for dct in [input_plugins, output_plugins]:
for plugin_id, plugin_config in dct.iteritems():
if 'targets' in plugin_config:
targets = plugin_config.pop('targets')
if not isinstance(targets, list):
targets = [targets]
for target in targets:
if target not in output_plugins:
raise plugin_base.ConfigError(
'Non-existent target output plugin ID `%s\' referenced in '
'plugin `%s\' config' % (target, plugin_id))
target_allow = output_plugins[target].setdefault('allow', [])
target_allow.append({'rule': 'history',
'plugin_id': plugin_id,
'position': -1})
# Ensure that all output plugins have at least one event source.
for plugin_id, plugin_config in output_plugins.iteritems():
if not plugin_config.get('allow'):
raise plugin_base.ConfigError(
'No plugin is targetting output plugin `%s\'. Please (1) disable '
'this plugin, (2) add allow/deny rules, or (3) configure '
'`targets\' of another plugin to point to it.' % plugin_id)
def _ConfigEntryToSandbox(self, superclass, plugin_id, config):
"""Parses configuration for a particular plugin entry.
Returns:
PluginSandbox object representing the plugin.
Raises:
ConfigError if the config dict does not include the plugin module to load.
"""
# The plugin type is included along with its configuration. Extract it.
if not isinstance(config, dict) or 'plugin' not in config:
raise plugin_base.ConfigError(
'Plugin %s must have a config dictionary which includes the key '
'`plugin` to specify which plugin module to load' % plugin_id)
plugin_type = config.pop('plugin')
allow = config.pop('allow', [])
deny = config.pop('deny', [])
args = config.pop('args', {})
# Disallow recursion by default. Any events emitted by a plugin should
# never be processed by that plugin again.
enable_recursion = config.pop('enable_recursion', False)
if config:
raise plugin_base.ConfigError(
'Plugin %s has extra arguments: %s' % (plugin_id, ', '.join(config)))
# Create FlowPolicy object.
policy = flow_policy.FlowPolicy(allow, deny)
if not enable_recursion:
policy.deny.append(
flow_policy.HistoryRule(plugin_id=plugin_id,
node_id=self._node_id))
# Make sure we have a store_path and data_dir for the plugin.
store_path = os.path.join(self._data_dir, '%s.json' % plugin_id)
data_dir = os.path.join(self._data_dir, plugin_id)
if not os.path.exists(data_dir):
os.makedirs(data_dir)
return plugin_sandbox.PluginSandbox(
plugin_type=plugin_type,
plugin_id=plugin_id,
superclass=superclass,
config=args,
policy=policy,
store_path=store_path,
data_dir=data_dir,
core_api=self)
def _ConfigEntriesToSandboxes(self, superclass, entries):
plugins = {}
for plugin_id, plugin_config in entries.iteritems():
# Parse this particular plugin entry and add to the _plugins map.
plugin_entry = self._ConfigEntryToSandbox(
superclass=superclass,
plugin_id=plugin_id,
config=plugin_config)
plugins[plugin_id] = plugin_entry
return plugins
def _StartBuffer(self):
logging.info('Starting buffer...')
self._buffer.Start(True)
self._SyncConsumerList()
def _SyncConsumerList(self):
"""Synchronizes consumer list with buffer."""
consumers = [plugin.plugin_id for plugin in self._plugins.values()
if plugin.GetSuperclass() is plugin_base.OutputPlugin]
buffer_consumers = self._buffer.CallPlugin('ListConsumers').keys()
logging.info('Syncing consumer lists')
logging.debug('Our consumer list: %s', consumers)
logging.debug('Buffer consumer list: %s', buffer_consumers)
for c in buffer_consumers:
if c not in consumers:
self._buffer.CallPlugin('RemoveConsumer', c)
for c in consumers:
if c not in buffer_consumers:
self._buffer.CallPlugin('AddConsumer', c)
def Run(self):
try:
self._state = STARTING
self._Start()
plugin_states = {}
for plugin in self._plugins.values():
plugin_states[plugin] = plugin.GetState()
while self._state not in (STOPPING, DOWN):
# If Instalog is just starting, check to see that all plugins have left
# the STARTING state. When this occurs, Instalog's state should change
# to UP.
if (self._state is STARTING and
all([state is not plugin_sandbox.STARTING
for state in plugin_states.values()])):
self._state = UP
for plugin in self._plugins.values():
plugin.AdvanceState()
if plugin_states[plugin] != plugin.GetState():
logging.info('Plugin %s changed state from %s to %s',
plugin.plugin_id, plugin_states[plugin],
plugin.GetState())
plugin_states[plugin] = plugin.GetState()
time.sleep(1)
except Exception as e:
logging.exception(e)
# In case there was some error in the Run function (exception or otherwise),
# call Stop synchronously at the end just in case.
self.Stop(True)
logging.warning('Stopped')
def _Start(self):
logging.info('Starting buffer...')
self._StartBuffer()
logging.info('Started buffer')
for plugin in self._plugins.values():
logging.info('Starting %s...', plugin.plugin_id)
plugin.Start()
for plugin in self._plugins.values():
plugin.AdvanceState(True)
logging.info('Started %s', plugin.plugin_id)
def IsUp(self):
with self._rpc_lock:
return self._state is UP
def Stop(self, sync=False):
"""Stops Instalog.
Args:
sync: If true, only returns when Instalog has stopped running.
"""
# If a Stop is already in process, return immediately.
if self._state in (STOPPING, DOWN):
return
# If called in asynchronous mode, kick off a thread to perform the stop.
if not sync:
threading.Thread(target=self.Stop, args=(True,)).start()
return
self._ShutdownRPCServer()
with self._rpc_lock:
self._state = STOPPING
for plugin in self._plugins.values():
if plugin.IsLoaded():
logging.info('Stopping %s...', plugin.plugin_id)
plugin.Stop()
for plugin in self._plugins.values():
plugin.AdvanceState(True)
logging.info('Stopped %s', plugin.plugin_id)
logging.info('Stopping buffer...')
self._buffer.Stop()
logging.info('Stopped buffer')
self._state = DOWN
def Inspect(self, plugin_id, json_path):
with self._rpc_lock:
if plugin_id not in self._plugins:
return False, 'Plugin `%s\' not found' % plugin_id
try:
store_data = self._plugins[plugin_id].store
return True, json_utils.JSONEncoder().encode(
json_utils.WalkJSONPath(json_path, store_data))
except Exception as e:
return False, ('Error on inspect with JSON path `%s\': %s'
% (json_path, e.message))
def Flush(self, plugin_id, timeout):
with self._rpc_lock:
if plugin_id not in self._plugins:
return False, 'Plugin `%s\' not found' % plugin_id
plugin = self._plugins[plugin_id]
if plugin.GetSuperclass() is not plugin_base.OutputPlugin:
return False, ('Only output plugins can be flushed (tried: `%s\')'
% plugin_id)
if plugin.Flush(timeout, True):
progress = plugin.GetProgress()
return True, ('Flushed `%s\' successfully (%d / %d events)'
% (plugin_id, progress[0], progress[1]))
else:
progress = plugin.GetProgress()
return False, ('Flush for `%s\' failed within the specified timeout '
'(%d / %d events)'
% (plugin_id, progress[0], progress[1]))
############################################################
# Functions below implement plugin_base.CoreAPI.
############################################################
def Emit(self, plugin, events):
"""Emits given events from the specified plugin.
Args:
plugin: PluginSandbox object of plugin performing Emit.
events: List of events to be emitted.
Returns:
True if successful, False if any failure occurred.
Raises:
PluginCallError if Buffer fails unexpectedly.
"""
return self._buffer.CallPlugin('Produce', events)
def NewStream(self, plugin):
"""Creates a new BufferEventStream for the specified plugin.
Args:
plugin: PluginSandbox object requesting BufferEventStream.
Returns:
Object implementing plugin_base.BufferEventStream.
Raises:
PluginCallError if Buffer fails unexpectedly.
"""
return self._buffer.CallPlugin('Consume', plugin.plugin_id)
def GetProgress(self, plugin):
"""Returns the current progress through buffer for the specified plugin.
Args:
plugin: PluginSandbox object requesting BufferEventStream.
Returns:
A tuple (completed_count, total_count) representing how many Events have
been processed so far, and how many exist in total.
Raises:
PluginCallError if Buffer fails unexpectedly.
"""
return self._buffer.CallPlugin('ListConsumers')[plugin.plugin_id]
def GetNodeID(self):
"""Returns the ID of this Instalog node.
Returns:
A string representing the ID of this Instalog node.
"""
return self._node_id