Sensors and Triggers

Sensors

Sensors are a way to integrate external systems and events with StackStorm. Sensors are pieces of Python code that either periodically poll some external system, or passively wait for inbound events. They then inject triggers into StackStorm, which can be matched by rules, for potential action execution.

Sensors are written in Python, and must follow the StackStorm-defined sensor interface requirements.

Triggers

Triggers are StackStorm constructs that identify the incoming events to StackStorm. A trigger is a tuple of type (string) and optional parameters (object). Rules are written to work with triggers. Sensors typically register triggers though this is not strictly required. For example, there is a generic webhooks trigger registered with StackStorm, which does not require a custom sensor.

Internal Triggers

By default StackStorm emits some internal triggers which you can leverage in rules. Those triggers can be distinguished from non-system triggers since they are prefixed with st2..

A list of available triggers for each resource is included below:

Action

Reference	Description	Properties
core.st2.generic.actiontrigger	Trigger encapsulating the completion of an action execution.	execution_id, status, start_timestamp, action_name, action_ref, runner_ref, parameters, result
core.st2.generic.notifytrigger	Notification trigger.	execution_id, status, start_timestamp, end_timestamp, action_ref, runner_ref, channel, route, message, data
core.st2.action.file_written	Trigger encapsulating action file being written on disk.	ref, file_path, host_info
core.st2.generic.inquiry	Trigger indicating a new “inquiry” has entered “pending” status	id, route

Sensor

Reference	Description	Properties
core.st2.sensor.process_spawn	Trigger indicating sensor process is started up.	object
core.st2.sensor.process_exit	Trigger indicating sensor process is stopped.	object

Key Value Pair

Reference	Description	Properties
core.st2.key_value_pair.create	Trigger encapsulating datastore item creation.	object
core.st2.key_value_pair.update	Trigger encapsulating datastore set action.	object
core.st2.key_value_pair.value_change	Trigger encapsulating a change of datastore item value.	old_object, new_object
core.st2.key_value_pair.delete	Trigger encapsulating datastore item deletion.	object

Creating a Sensor

Creating a sensor involves writing a Python file and a YAML metadata file that defines the sensor. Here’s a minimal skeleton example. This is the metadata file:

---
  class_name: "SampleSensor"
  entry_point: "sample_sensor.py"
  description: "Sample sensor that emits triggers."
  trigger_types:
    -
      name: "event"
      description: "An example trigger."
      payload_schema:
        type: "object"
        properties:
          executed_at:
            type: "string"
            format: "date-time"
            default: "2014-07-30 05:04:24.578325"

And this is the corresponding Python skeleton:

# Copyright 2020 The StackStorm Authors.
# Copyright 2019 Extreme Networks, 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 st2reactor.sensor.base import Sensor


class SampleSensor(Sensor):
    """
    * self.sensor_service
        - provides utilities like
            - get_logger() - returns logger instance specific to this sensor.
            - dispatch() for dispatching triggers into the system.
    * self._config
        - contains parsed configuration that was specified as
          config.yaml in the pack.
    """

    def setup(self):
        # Setup stuff goes here. For example, you might establish connections
        # to external system once and reuse it. This is called only once by the system.
        pass

    def run(self):
        # This is where the crux of the sensor work goes.
        # This is called once by the system.
        # (If you want to sleep for regular intervals and keep
        # interacting with your external system, you'd inherit from PollingSensor.)
        # For example, let's consider a simple flask app. You'd run the flask app here.
        # You can dispatch triggers using sensor_service like so:
        # self.sensor_service(trigger, payload, trace_tag)
        #   # You can refer to the trigger as dict
        #   # { "name": ${trigger_name}, "pack": ${trigger_pack} }
        #   # or just simply by reference as string.
        #   # i.e. dispatch(${trigger_pack}.${trigger_name}, payload)
        #   # E.g.: dispatch('examples.foo_sensor', {'k1': 'stuff', 'k2': 'foo'})
        #   # trace_tag is a tag you would like to associate with the dispatched TriggerInstance
        #   # Typically the trace_tag is unique and a reference to an external event.
        pass

    def cleanup(self):
        # This is called when the st2 system goes down. You can perform cleanup operations like
        # closing the connections to external system here.
        pass

    def add_trigger(self, trigger):
        # This method is called when trigger is created
        pass

    def update_trigger(self, trigger):
        # This method is called when trigger is updated
        pass

    def remove_trigger(self, trigger):
        # This method is called when trigger is deleted
        pass

This is a bare minimum version of what a sensor looks like. For a more complete implementation of a sensor that actually injects triggers into the system, look at the examples section below.

Your sensor should generate triggers in Python dict form:

trigger = 'pack.name'
payload = {
    'executed_at': '2014-08-01T00:00:00.000000Z'
}
trace_tag = external_event_id

The sensor injects such triggers by using the sensor_service passed into the sensor on instantiation.

self.sensor_service.dispatch(trigger=trigger, payload=payload, trace_tag=trace_tag)

If you want a sensor that polls an external system at regular intervals, you can use a PollingSensor instead of Sensor as the base class.

# Copyright 2020 The StackStorm Authors.
# Copyright 2019 Extreme Networks, 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 st2reactor.sensor.base import PollingSensor


class SamplePollingSensor(PollingSensor):
    """
    * self.sensor_service
        - provides utilities like
            get_logger() for writing to logs.
            dispatch() for dispatching triggers into the system.
    * self._config
        - contains configuration that was specified as
          config.yaml in the pack.
    * self._poll_interval
        - indicates the interval between two successive poll() calls.
    """

    def setup(self):
        # Setup stuff goes here. For example, you might establish connections
        # to external system once and reuse it. This is called only once by the system.
        pass

    def poll(self):
        # This is where the crux of the sensor work goes.
        # This is called every self._poll_interval.
        # For example, let's assume you want to query ec2 and get
        # health information about your instances:
        #   some_data = aws_client.get('')
        #   payload = self._to_payload(some_data)
        #   # _to_triggers is something you'd write to convert the data format you have
        #   # into a standard python dictionary. This should follow the payload schema
        #   # registered for the trigger.
        #   self.sensor_service.dispatch(trigger, payload)
        #   # You can refer to the trigger as dict
        #   # { "name": ${trigger_name}, "pack": ${trigger_pack} }
        #   # or just simply by reference as string.
        #   # i.e. dispatch(${trigger_pack}.${trigger_name}, payload)
        #   # E.g.: dispatch('examples.foo_sensor', {'k1': 'stuff', 'k2': 'foo'})
        #   # trace_tag is a tag you would like to associate with the dispatched TriggerInstance
        #   # Typically the trace_tag is unique and a reference to an external event.
        pass

    def cleanup(self):
        # This is called when the st2 system goes down. You can perform cleanup operations like
        # closing the connections to external system here.
        pass

    def add_trigger(self, trigger):
        # This method is called when trigger is created
        pass

    def update_trigger(self, trigger):
        # This method is called when trigger is updated
        pass

    def remove_trigger(self, trigger):
        # This method is called when trigger is deleted
        pass

Polling Sensors also require a poll_interval parameter in the metadata file. This defines (in seconds) how frequently the poll() method is called.

How Sensors are Run

Each sensor runs as a separate process. The st2sensorcontainer (see Overview) starts sensor_wrapper.py which wraps your Sensor class (such as SampleSensor or SamplePollingSensor above) in a st2reactor.container.sensor_wrapper.SensorWrapper.

Sensor Service

As you can see in the example above, a sensor_service is passed to each sensor class constructor on instantiation.

The Sensor service provides different services to the sensor via public methods. The most important one is the dispatch method which allows sensors to inject triggers into the system.

All public methods are described below:

Common Operations

1. dispatch(trigger, payload, trace_tag)

This method allows the sensor to inject triggers into the system.

For example:

trigger = 'pack.name'
payload = {
    'executed_at': '2014-08-01T00:00:00.000000Z'
}
trace_tag = uuid.uuid4().hex

self.sensor_service.dispatch(trigger=trigger, payload=payload, trace_tag=trace_tag)

2. get_logger(name)

This method allows the sensor instance to retrieve the logger instance which is specific to that sensor.

For example:

self._logger = self.sensor_service.get_logger(name=self.__class__.__name__)
self._logger.debug('Polling 3rd party system for information')

Datastore Management Operations

In addition to the trigger injection, the sensor service also provides functionality for reading and manipulating the datastore.

Each sensor has a namespace which is local to it and by default, all the datastore operations operate on the keys in that sensor-local namespace. If you want to operate on a global namespace, you need to pass the local=False argument to the datastore manipulation method.

Among other reasons, this functionality is useful if you want to persist temporary data between sensor runs.

A good example of this functionality in action is TwitterSensor. The Twitter sensor persists the ID of the last processed tweet after every poll in the datastore. This way if the sensor is restarted or if it crashes, the sensor can resume from where it left off without injecting duplicate triggers into the system.

For the implementation, see twitter_search_sensor.py in StackStorm Exchange

1. list_values(local=True, prefix=None)

This method allows you to list the values in the datastore. You can also filter by key name prefix (key name starts with) by passing prefix argument to the method:

kvps = self.sensor_service.list_values(local=False, prefix='cmdb.')

for kvp in kvps:
    print(kvp.name)
    print(kvp.value)

2. get_value(name, local=True, decrypt=False)

This method allows you to retrieve a single value from the datastore:

kvp = self.sensor_service.get_value('cmdb.api_host')
print(kvp.name)

If the value is encrypted, you can decrypt it with this:

kvp = self.sensor_service.get_value('cmdb.api_password', decrypt=True)
print(kvp.name)

3. set_value(name, value, ttl=None, local=True, encrypt=False)

This method allows you to store (set) a value in the datastore. Optionally you can also specify time to live (TTL) for the stored value:

last_id = 12345
self.sensor_service.set_value(name='last_id', value=str(last_id))

Secret values can be encrypted in the datastore:

ma_password = 'Sup3rS34et'
self.sensor_service.set_value(name='ma_password', value=ma_password, encrypt=True)

4. delete_value(name, local=True)

This method allows you to delete an existing value from the datastore. If a value is not found this method will return False, True otherwise.

self.sensor_service.delete_value(name='my_key')

API Docs

class st2reactor.container.sensor_wrapper.SensorService(sensor_wrapper)[source]

Instance of this class is passed to the sensor instance and exposes “public” methods which can be called by the sensor.

dispatch_with_context(trigger, payload=None, trace_context=None)[source]

Method which dispatches the trigger.

Parameters

trigger (str) – Full name / reference of the trigger.
payload (dict) – Trigger payload.
trace_context (st2common.api.models.api.trace.TraceContext) – Trace context to associate with Trigger.

get_logger(name)[source]: Retrieve an instance of a logger to be used by the sensor class.

Running Your First Sensor

Once you write your own sensor, the following steps can be used to run your sensor for the first time:

Place the sensor Python file and YAML metadata in the default pack in /opt/stackstorm/packs/default/sensors/. Alternatively, you can create a custom pack in /opt/stackstorm/packs/ with the appropriate pack structure (see Create and Contribute a Pack) and place the sensor artifacts there.
Register the sensor with st2ctl. Watch for any errors in sensor registration:
```
st2ctl reload --register-all
```
If there are errors in registration, fix the errors and re-register them using st2ctl reload --register-all.
If registration is successful, the sensor will run automatically.

Once you like your sensor, you can promote it to a pack (if required) by creating a pack in /opt/stackstorm/packs/${pack_name} and moving the sensor artifacts (YAML and Python) to /opt/stackstorm/packs/${pack_name}/sensors/. See Create and Contribute a Pack for how to create a pack.

Examples

This is a working example of a simple sensor that injects a trigger every 60 seconds.

Metadata:

---
class_name: "HelloSensor"
entry_point: "sensor1.py"
description: "Test sensor that emits triggers."
trigger_types:
  -
    name: "event1"
    description: "An example trigger."
    payload_schema:
      type: "object"

Python code:

# Copyright 2020 The StackStorm Authors.
# Copyright 2019 Extreme Networks, 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.

import eventlet

from st2reactor.sensor.base import Sensor


class HelloSensor(Sensor):
    def __init__(self, sensor_service, config):
        super(HelloSensor, self).__init__(sensor_service=sensor_service, config=config)
        self._logger = self.sensor_service.get_logger(name=self.__class__.__name__)
        self._stop = False

    def setup(self):
        pass

    def run(self):
        while not self._stop:
            self._logger.debug("HelloSensor dispatching trigger...")
            count = self.sensor_service.get_value("hello_st2.count") or 0
            payload = {"greeting": "Yo, StackStorm!", "count": int(count) + 1}
            self.sensor_service.dispatch(trigger="hello_st2.event1", payload=payload)
            self.sensor_service.set_value("hello_st2.count", payload["count"])
            eventlet.sleep(60)

    def cleanup(self):
        self._stop = True

    # Methods required for programmable sensors.
    def add_trigger(self, trigger):
        pass

    def update_trigger(self, trigger):
        pass

    def remove_trigger(self, trigger):
        pass

The StackStorm Exchange has many more examples. Here’s just a few:

Passive Sensors

Polling Sensors

Debugging a Sensor From a Pack

If you just want to run a single sensor from a pack and the sensor is already registered, you can use the st2sensorcontainer to run just that single sensor:

sudo /opt/stackstorm/st2/bin/st2sensorcontainer --config-file=/etc/st2/st2.conf --sensor-ref=pack.SensorClassName

For example:

sudo /opt/stackstorm/st2/bin/st2sensorcontainer --config-file=/etc/st2/st2.conf --sensor-ref=git.GitCommitSensor

Sensors and Triggers

Sensors

Triggers

Internal Triggers

Action

Sensor

Key Value Pair

Creating a Sensor

How Sensors are Run

Sensor Service

Common Operations

1. dispatch(trigger, payload, trace_tag)

2. get_logger(name)

Datastore Management Operations

1. list_values(local=True, prefix=None)

2. get_value(name, local=True, decrypt=False)

3. set_value(name, value, ttl=None, local=True, encrypt=False)

4. delete_value(name, local=True)

API Docs

Running Your First Sensor

Examples

Passive Sensors

Polling Sensors

Debugging a Sensor From a Pack

Sharing code between Python Sensors and Actions