# coding=utf-8 # # Copyright 2011-2015 Splunk, Inc. # # Licensed under the Apache License, Version 2.0 (the "License"): you may # not use this file except in compliance with the License. You may obtain # a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, WITHOUT # WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the # License for the specific language governing permissions and limitations # under the License. from __future__ import absolute_import, division, print_function, unicode_literals from splunklib import six from splunklib.six.moves import map as imap, filter as ifilter from .decorators import ConfigurationSetting from .search_command import SearchCommand class StreamingCommand(SearchCommand): """ Applies a transformation to search results as they travel through the streams pipeline. Streaming commands typically filter, augment, or update, search result records. Splunk will send them in batches of up to 50,000 records. Hence, a search command must be prepared to be invoked many times during the course of pipeline processing. Each invocation should produce a set of results independently usable by downstream processors. By default Splunk may choose to run a streaming command locally on a search head and/or remotely on one or more indexers concurrently. The size and frequency of the search result batches sent to the command will vary based on scheduling considerations. StreamingCommand configuration ============================== You can configure your command for operation under Search Command Protocol (SCP) version 1 or 2. SCP 2 requires Splunk 6.3 or later. """ # region Methods def stream(self, records): """ Generator function that processes and yields event records to the Splunk stream pipeline. You must override this method. """ raise NotImplementedError('StreamingCommand.stream(self, records)') def _execute(self, ifile, process): SearchCommand._execute(self, ifile, self.stream) # endregion class ConfigurationSettings(SearchCommand.ConfigurationSettings): """ Represents the configuration settings that apply to a :class:`StreamingCommand`. """ # region SCP v1/v2 properties required_fields = ConfigurationSetting(doc=''' List of required fields for this search which back-propagates to the generating search. Setting this value enables selected fields mode under SCP 2. Under SCP 1 you must also specify :code:`clear_required_fields=True` to enable selected fields mode. To explicitly select all fields, specify a value of :const:`['*']`. No error is generated if a specified field is missing. Default: :const:`None`, which implicitly selects all fields. Supported by: SCP 1, SCP 2 ''') # endregion # region SCP v1 properties clear_required_fields = ConfigurationSetting(doc=''' :const:`True`, if required_fields represent the *only* fields required. If :const:`False`, required_fields are additive to any fields that may be required by subsequent commands. In most cases, :const:`False` is appropriate for streaming commands. Default: :const:`False` Supported by: SCP 1 ''') local = ConfigurationSetting(doc=''' :const:`True`, if the command should run locally on the search head. Default: :const:`False` Supported by: SCP 1 ''') overrides_timeorder = ConfigurationSetting(doc=''' :const:`True`, if the command changes the order of events with respect to time. Default: :const:`False` Supported by: SCP 1 ''') streaming = ConfigurationSetting(readonly=True, value=True, doc=''' Specifies that the command is streamable. Fixed: :const:`True` Supported by: SCP 1 ''') # endregion # region SCP v2 Properties distributed = ConfigurationSetting(value=True, doc=''' :const:`True`, if this command should be distributed to indexers. Under SCP 1 you must either specify `local = False` or include this line in commands.conf_, if this command should be distributed to indexers. ..code: local = true Default: :const:`True` Supported by: SCP 2 .. commands.conf_: http://docs.splunk.com/Documentation/Splunk/latest/Admin/Commandsconf ''') maxinputs = ConfigurationSetting(doc=''' Specifies the maximum number of events that can be passed to the command for each invocation. This limit cannot exceed the value of `maxresultrows` in limits.conf. Under SCP 1 you must specify this value in commands.conf_. Default: The value of `maxresultrows`. Supported by: SCP 2 ''') type = ConfigurationSetting(readonly=True, value='streaming', doc=''' Command type name. Fixed: :const:`'streaming'` Supported by: SCP 2 ''') # endregion # region Methods @classmethod def fix_up(cls, command): """ Verifies :code:`command` class structure. """ if command.stream == StreamingCommand.stream: raise AttributeError('No StreamingCommand.stream override') return # TODO: Stop looking like a dictionary because we don't obey the semantics # N.B.: Does not use Python 2 dict copy semantics def iteritems(self): iteritems = SearchCommand.ConfigurationSettings.iteritems(self) version = self.command.protocol_version if version == 1: if self.required_fields is None: iteritems = ifilter(lambda name_value: name_value[0] != 'clear_required_fields', iteritems) else: iteritems = ifilter(lambda name_value2: name_value2[0] != 'distributed', iteritems) if not self.distributed: iteritems = imap( lambda name_value1: (name_value1[0], 'stateful') if name_value1[0] == 'type' else (name_value1[0], name_value1[1]), iteritems) return iteritems # N.B.: Does not use Python 3 dict view semantics if not six.PY2: items = iteritems # endregion