From 7775c5a510882eae3fc5e09789d0c3370e3765e3 Mon Sep 17 00:00:00 2001 From: guzzijones12 Date: Mon, 31 Aug 2026 09:48:16 -0400 Subject: [PATCH 1/7] sensor state, sensor retry config, sensor backoff config, sensor death trigger --- CHANGELOG.rst | 13 ++ conf/st2.conf.sample | 6 + st2api/st2api/controllers/v1/sensors.py | 113 ++++++++++++ .../unit/controllers/v1/test_sensortypes.py | 84 +++++++++ st2client/st2client/commands/sensor.py | 8 +- st2client/st2client/models/reactor.py | 2 +- st2client/tests/unit/test_sensor_commands.py | 103 +++++++++++ st2common/st2common/constants/sensors.py | 24 +++ st2common/st2common/constants/triggers.py | 10 +- st2common/st2common/models/api/sensor.py | 29 ++++ st2common/st2common/models/db/__init__.py | 1 + .../st2common/models/db/sensor_instance.py | 76 ++++++++ st2common/st2common/openapi.yaml | 10 ++ st2common/st2common/openapi.yaml.j2 | 10 ++ .../st2common/persistence/sensor_instance.py | 29 ++++ .../tests/unit/test_db_sensor_instance.py | 90 ++++++++++ .../st2reactor/container/process_container.py | 131 +++++++++++++- st2reactor/st2reactor/sensor/config.py | 30 ++++ .../tests/unit/test_process_container.py | 164 ++++++++++++++++++ st2tests/st2tests/config.py | 26 +++ 20 files changed, 950 insertions(+), 9 deletions(-) create mode 100644 st2client/tests/unit/test_sensor_commands.py create mode 100644 st2common/st2common/models/db/sensor_instance.py create mode 100644 st2common/st2common/persistence/sensor_instance.py create mode 100644 st2common/tests/unit/test_db_sensor_instance.py diff --git a/CHANGELOG.rst b/CHANGELOG.rst index 5719ee2d16..a2a2872e46 100644 --- a/CHANGELOG.rst +++ b/CHANGELOG.rst @@ -36,6 +36,19 @@ Changed Added ~~~~~ * added raw_string type to allow template strings to pass through variable processing (by @guzzijones12@gmail.com) #6351 +* Added a new internal ``core.st2.sensor.process_abandoned`` trigger which is emitted when the sensor + container permanently gives up on a sensor after exceeding the maximum number of respawn attempts. + This lets users define their own alerting (email, Slack, workflow, etc.) via a rule instead of a + sensor failing silently. +* Added a ``SensorInstanceDB`` collection tracking each sensor's runtime health (status, hostname, pid, + exit code, respawn count, last updated). The health fields are merged into the ``/v1/sensors`` API + response and shown by ``st2 sensor list`` / ``st2 sensor get``, and can be filtered with + ``st2 sensor list --status=abandoned``. +* Added ``[sensorcontainer].max_respawn_count``, ``[sensorcontainer].respawn_delay`` and + ``[sensorcontainer].respawn_backoff_factor`` config options to control how many times a crashed + sensor is respawned before being abandoned and how long to wait between attempts. The backoff + factor defaults to ``1`` (a constant ``respawn_delay`` between attempts); a value greater than ``1`` + grows the delay exponentially. 3.9.0 - October 10, 2025 ------------------------ diff --git a/conf/st2.conf.sample b/conf/st2.conf.sample index 27a2eb0a86..37b17b9b52 100644 --- a/conf/st2.conf.sample +++ b/conf/st2.conf.sample @@ -306,8 +306,14 @@ version = 4 [sensorcontainer] # location of the logging.conf file logging = /etc/st2/logging.sensorcontainer.conf +# Maximum number of times to respawn a sensor after it exits with a non-zero code before giving up and firing the st2.sensor.process_abandoned trigger. +max_respawn_count = 2 # Provider of sensor node partition config. partition_provider = name:default +# Exponential backoff multiplier applied to respawn_delay per attempt. 1 (default) means a constant delay between attempts; a value > 1 grows the delay exponentially (respawn_delay * factor ** (attempt - 1)). +respawn_backoff_factor = 1.0 +# Base delay (in seconds) to wait between sensor respawn attempts. +respawn_delay = 2.5 # name of the sensor node. sensor_node_name = sensornode1 # Run in a single sensor mode where parent process exits when a sensor crashes / dies. This is useful in environments where partitioning, sensor process life cycle and failover is handled by a 3rd party service such as kubernetes. diff --git a/st2api/st2api/controllers/v1/sensors.py b/st2api/st2api/controllers/v1/sensors.py index b62b56c92d..457a1d7927 100644 --- a/st2api/st2api/controllers/v1/sensors.py +++ b/st2api/st2api/controllers/v1/sensors.py @@ -18,14 +18,17 @@ from st2common import log as logging from st2common.persistence.sensor import SensorType +from st2common.persistence.sensor_instance import SensorInstance from st2common.models.api.sensor import SensorTypeAPI from st2common.exceptions.apivalidation import ValueValidationException from st2common.validators.api.misc import validate_not_part_of_system_pack +from st2common.util import isotime from st2api.controllers import resource from st2api.controllers.controller_transforms import transform_to_bool from st2common.rbac.types import PermissionType from st2common.rbac.backends import get_rbac_backend from st2common.router import abort +from st2common.router import Response http_client = six.moves.http_client @@ -40,12 +43,25 @@ class SensorTypeController(resource.ContentPackResourceController): "pack": "pack", "enabled": "enabled", "trigger": "trigger_types", + # "ref" maps to a "ref__in" filter so it accepts one or more refs. It is + # also used internally to constrain results when filtering by "status" + # (which lives in the separate SensorInstanceDB collection). + "ref": "ref.in", } filter_transform_functions = {"enabled": transform_to_bool} options = {"sort": ["pack", "name"]} + # Runtime health fields merged in from SensorInstanceDB (keyed by sensor ref). + HEALTH_ATTRIBUTES = [ + "status", + "hostname", + "pid", + "exit_code", + "respawn_count", + ] + def get_all( self, exclude_attributes=None, @@ -56,6 +72,30 @@ def get_all( requester_user=None, **raw_filters, ): + # "status" is not a SensorTypeDB field - it lives in SensorInstanceDB. + # Resolve it to the set of matching sensor refs and constrain the query. + status = raw_filters.pop("status", None) + if status: + try: + refs = [ + instance.ref + for instance in SensorInstance.query( + status=status, only_fields=["ref"] + ) + ] + except Exception: + LOG.warning( + "Failed to resolve sensor refs for status filter", exc_info=True + ) + refs = [] + + if not refs: + resp = Response(json=[]) + resp.headers["X-Total-Count"] = "0" + return resp + + raw_filters["ref"] = refs + return super(SensorTypeController, self)._get_all( exclude_fields=exclude_attributes, include_fields=include_attributes, @@ -72,6 +112,79 @@ def get_one(self, ref_or_id, requester_user): ref_or_id, requester_user=requester_user, permission_type=permission_type ) + def resources_model_filter( + self, + model, + instances, + requester_user=None, + offset=0, + eop=0, + **from_model_kwargs, + ): + # List path (get_all): batch-fetch health records once to avoid N+1. + page = list(instances[offset:eop]) + health_by_ref = self._get_health_by_ref( + [getattr(instance, "ref", None) for instance in page] + ) + + result = [] + for instance in page: + # Call the base per-item converter directly (not self.resource_model_filter) + # so we don't issue a second, per-item health query. + item = super(SensorTypeController, self).resource_model_filter( + model=model, + instance=instance, + requester_user=requester_user, + **from_model_kwargs, + ) + if item: + self._apply_health(item, health_by_ref.get(instance.ref)) + result.append(item) + return result + + def resource_model_filter( + self, model, instance, requester_user=None, **from_model_kwargs + ): + # Single path (get_one): convert then merge runtime health. + item = super(SensorTypeController, self).resource_model_filter( + model=model, + instance=instance, + requester_user=requester_user, + **from_model_kwargs, + ) + if item: + health_by_ref = self._get_health_by_ref([getattr(instance, "ref", None)]) + self._apply_health(item, health_by_ref.get(instance.ref)) + return item + + def _get_health_by_ref(self, refs): + """ + Return a ``ref -> SensorInstanceDB`` map for the provided sensor refs. + """ + refs = [ref for ref in refs if ref] + if not refs: + return {} + + try: + instances = SensorInstance.query(ref__in=refs) + return {instance.ref: instance for instance in instances} + except Exception: + LOG.warning("Failed to load sensor health records", exc_info=True) + return {} + + def _apply_health(self, item, instance_db): + """ + Merge runtime health fields from a SensorInstanceDB record onto the + SensorTypeAPI instance. Fields default to None when no record exists. + """ + for attribute in self.HEALTH_ATTRIBUTES: + setattr(item, attribute, getattr(instance_db, attribute, None)) + + updated_at = getattr(instance_db, "updated_at", None) + item.updated_at = ( + isotime.format(updated_at, offset=False) if updated_at else None + ) + def put(self, sensor_type, ref_or_id, requester_user): # Note: Right now this function only supports updating of "enabled" # attribute on the SensorType model. diff --git a/st2api/tests/unit/controllers/v1/test_sensortypes.py b/st2api/tests/unit/controllers/v1/test_sensortypes.py index ea32f0e904..23b2bbad03 100644 --- a/st2api/tests/unit/controllers/v1/test_sensortypes.py +++ b/st2api/tests/unit/controllers/v1/test_sensortypes.py @@ -20,6 +20,11 @@ import st2common.bootstrap.sensorsregistrar as sensors_registrar from st2api.controllers.v1.sensors import SensorTypeController +from st2common.constants.sensors import SENSOR_STATUS_RUNNING +from st2common.constants.sensors import SENSOR_STATUS_ABANDONED +from st2common.models.db.sensor_instance import SensorInstanceDB +from st2common.persistence.sensor_instance import SensorInstance + from st2tests.api import FunctionalTest from st2tests.api import APIControllerWithIncludeAndExcludeFilterTestCase @@ -122,6 +127,85 @@ def test_get_one_doesnt_exist(self): resp = self.app.get("/v1/sensortypes/1", expect_errors=True) self.assertEqual(resp.status_int, http_client.NOT_FOUND) + def tearDown(self): + super(SensorTypeControllerTestCase, self).tearDown() + # Remove any runtime health records created by health tests so they do + # not leak into other tests. + for instance in SensorInstance.get_all(): + SensorInstance.delete(instance) + + @staticmethod + def _create_health_record(ref, pack, status, **kwargs): + instance_db = SensorInstanceDB(ref=ref, pack=pack, status=status, **kwargs) + return SensorInstance.add_or_update(instance_db) + + def test_get_one_merges_health_fields(self): + ref = f"{DUMMY_PACK_1}.SampleSensor" + self._create_health_record( + ref, + DUMMY_PACK_1, + SENSOR_STATUS_RUNNING, + hostname="sensor-node-1", + pid=4321, + exit_code=0, + respawn_count=0, + ) + + resp = self.app.get(f"/v1/sensortypes/{ref}") + self.assertEqual(resp.status_int, http_client.OK) + self.assertEqual(resp.json["status"], SENSOR_STATUS_RUNNING) + self.assertEqual(resp.json["hostname"], "sensor-node-1") + self.assertEqual(resp.json["pid"], 4321) + self.assertEqual(resp.json["exit_code"], 0) + self.assertEqual(resp.json["respawn_count"], 0) + self.assertIsNotNone(resp.json["updated_at"]) + + def test_get_one_no_health_record_returns_null_fields(self): + # A sensor which has never run has no SensorInstanceDB record; the + # health fields must still be present and null. + resp = self.app.get(f"/v1/sensortypes/{DUMMY_PACK_1}.SampleSensor") + self.assertEqual(resp.status_int, http_client.OK) + self.assertIsNone(resp.json["status"]) + self.assertIsNone(resp.json["hostname"]) + self.assertIsNone(resp.json["pid"]) + self.assertIsNone(resp.json["updated_at"]) + + def test_get_all_merges_health_fields(self): + running_ref = f"{DUMMY_PACK_1}.SampleSensor" + abandoned_ref = f"{DUMMY_PACK_1}.SampleSensor2" + self._create_health_record( + running_ref, DUMMY_PACK_1, SENSOR_STATUS_RUNNING, pid=100 + ) + self._create_health_record( + abandoned_ref, DUMMY_PACK_1, SENSOR_STATUS_ABANDONED, exit_code=1 + ) + + resp = self.app.get("/v1/sensortypes") + self.assertEqual(resp.status_int, http_client.OK) + status_by_ref = {item["ref"]: item["status"] for item in resp.json} + self.assertEqual(status_by_ref[running_ref], SENSOR_STATUS_RUNNING) + self.assertEqual(status_by_ref[abandoned_ref], SENSOR_STATUS_ABANDONED) + # Sensor without a health record reports a null status. + self.assertIsNone(status_by_ref[f"{DUMMY_PACK_1}.SampleSensor3"]) + + def test_get_all_status_filter(self): + running_ref = f"{DUMMY_PACK_1}.SampleSensor" + abandoned_ref = f"{DUMMY_PACK_1}.SampleSensor2" + self._create_health_record(running_ref, DUMMY_PACK_1, SENSOR_STATUS_RUNNING) + self._create_health_record(abandoned_ref, DUMMY_PACK_1, SENSOR_STATUS_ABANDONED) + + resp = self.app.get("/v1/sensortypes?status=abandoned") + self.assertEqual(resp.status_int, http_client.OK) + self.assertEqual(len(resp.json), 1) + self.assertEqual(resp.json[0]["ref"], abandoned_ref) + self.assertEqual(resp.json[0]["status"], SENSOR_STATUS_ABANDONED) + + def test_get_all_status_filter_no_matches(self): + # No sensor is in the requested status - the result is empty. + resp = self.app.get("/v1/sensortypes?status=abandoned") + self.assertEqual(resp.status_int, http_client.OK) + self.assertEqual(len(resp.json), 0) + def test_disable_and_enable_sensor(self): # Verify initial state resp = self.app.get(f"/v1/sensortypes/{DUMMY_PACK_1}.SampleSensor") diff --git a/st2client/st2client/commands/sensor.py b/st2client/st2client/commands/sensor.py index 28c796ecac..9f14f22460 100644 --- a/st2client/st2client/commands/sensor.py +++ b/st2client/st2client/commands/sensor.py @@ -40,7 +40,7 @@ def __init__(self, description, app, subparsers, parent_parser=None): class SensorListCommand(resource.ContentPackResourceListCommand): - display_attributes = ["ref", "pack", "description", "enabled"] + display_attributes = ["ref", "pack", "enabled", "status", "updated_at"] class SensorGetCommand(resource.ContentPackResourceGetCommand): @@ -55,6 +55,12 @@ class SensorGetCommand(resource.ContentPackResourceGetCommand): "entry_point", "artifact_uri", "trigger_types", + "status", + "hostname", + "pid", + "exit_code", + "respawn_count", + "updated_at", ] diff --git a/st2client/st2client/models/reactor.py b/st2client/st2client/models/reactor.py index 0d89d8cdd8..fb185360cf 100644 --- a/st2client/st2client/models/reactor.py +++ b/st2client/st2client/models/reactor.py @@ -25,7 +25,7 @@ class Sensor(core.Resource): _plural = "Sensortypes" - _repr_attributes = ["name", "pack"] + _repr_attributes = ["name", "pack", "status"] class TriggerType(core.Resource): diff --git a/st2client/tests/unit/test_sensor_commands.py b/st2client/tests/unit/test_sensor_commands.py new file mode 100644 index 0000000000..a7b598f513 --- /dev/null +++ b/st2client/tests/unit/test_sensor_commands.py @@ -0,0 +1,103 @@ +# 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 __future__ import absolute_import + +import json + +import mock + +from tests import base + +from st2client.shell import Shell +from st2client.utils import httpclient + +__all__ = ["SensorCommandTestCase"] + + +SENSOR_RUNNING = { + "id": "1", + "ref": "wolfpack.SensorA", + "pack": "wolfpack", + "name": "SensorA", + "enabled": True, + "status": "running", + "hostname": "sensor-node-1", + "pid": 1234, + "exit_code": None, + "respawn_count": 0, + "updated_at": "2026-08-31T00:00:00.000000Z", +} + +SENSOR_ABANDONED = { + "id": "2", + "ref": "wolfpack.SensorB", + "pack": "wolfpack", + "name": "SensorB", + "enabled": True, + "status": "abandoned", + "hostname": "sensor-node-1", + "pid": None, + "exit_code": 1, + "respawn_count": 2, + "updated_at": "2026-08-31T00:00:00.000000Z", +} + + +class SensorCommandTestCase(base.BaseCLITestCase): + def __init__(self, *args, **kwargs): + super(SensorCommandTestCase, self).__init__(*args, **kwargs) + self.shell = Shell() + + @mock.patch.object( + httpclient.HTTPClient, + "get", + mock.MagicMock( + return_value=base.FakeResponse( + json.dumps([SENSOR_RUNNING, SENSOR_ABANDONED]), 200, "OK", {} + ) + ), + ) + def test_sensor_list_renders_status_column(self): + return_code = self.shell.run(["sensor", "list"]) + self.assertEqual(return_code, 0) + + stdout = self.stdout.getvalue() + # The status column header and both runtime status values are rendered. + self.assertIn("status", stdout) + self.assertIn("running", stdout) + self.assertIn("abandoned", stdout) + + @mock.patch.object( + httpclient.HTTPClient, + "get", + mock.MagicMock( + return_value=base.FakeResponse(json.dumps(SENSOR_ABANDONED), 200, "OK", {}) + ), + ) + def test_sensor_get_shows_health_fields(self): + return_code = self.shell.run(["sensor", "get", "wolfpack.SensorB"]) + self.assertEqual(return_code, 0) + + stdout = self.stdout.getvalue() + # get uses display_attributes = ["all"], so every health field is shown. + for expected in [ + "status", + "abandoned", + "hostname", + "exit_code", + "respawn_count", + ]: + self.assertIn(expected, stdout) diff --git a/st2common/st2common/constants/sensors.py b/st2common/st2common/constants/sensors.py index a2d7903d18..02c956cc62 100644 --- a/st2common/st2common/constants/sensors.py +++ b/st2common/st2common/constants/sensors.py @@ -21,3 +21,27 @@ KVSTORE_PARTITION_LOADER = "kvstore" FILE_PARTITION_LOADER = "file" HASH_PARTITION_LOADER = "hash" + +# Sensor respawn / retry defaults (overridable via the [sensorcontainer] config +# group: max_respawn_count, respawn_delay, respawn_backoff_factor). +# +# How many times to subsequently respawn a sensor after a non-zero exit before +# giving up (and firing the process_abandoned trigger). +DEFAULT_SENSOR_MAX_RESPAWN_COUNT = 2 +# Base delay (in seconds) to wait between respawn attempts. +DEFAULT_SENSOR_RESPAWN_DELAY = 2.5 +# Exponential backoff multiplier applied to the base delay per attempt. A value +# of 1 (the default) means a constant delay between attempts; a value > 1 grows +# the delay exponentially (delay * factor ** (attempt - 1)). +DEFAULT_SENSOR_RESPAWN_BACKOFF_FACTOR = 1.0 + +# Runtime status values recorded for a running sensor instance (SensorInstanceDB) +SENSOR_STATUS_RUNNING = "running" +SENSOR_STATUS_STOPPED = "stopped" +SENSOR_STATUS_ABANDONED = "abandoned" + +SENSOR_STATUSES = [ + SENSOR_STATUS_RUNNING, + SENSOR_STATUS_STOPPED, + SENSOR_STATUS_ABANDONED, +] diff --git a/st2common/st2common/constants/triggers.py b/st2common/st2common/constants/triggers.py index 4751ec372e..1b5c69c491 100644 --- a/st2common/st2common/constants/triggers.py +++ b/st2common/st2common/constants/triggers.py @@ -133,6 +133,14 @@ "payload_schema": {"type": "object", "properties": {"object": {}}}, } +SENSOR_ABANDONED_TRIGGER = { + "name": "st2.sensor.process_abandoned", + "pack": SYSTEM_PACK_NAME, + "description": "Trigger indicating the sensor process was abandoned after " + "exceeding the maximum number of respawn attempts.", + "payload_schema": {"type": "object", "properties": {"object": {}}}, +} + # KeyValuePair resource triggers KEY_VALUE_PAIR_CREATE_TRIGGER = { "name": "st2.key_value_pair.create", @@ -173,7 +181,7 @@ ACTION_FILE_WRITTEN_TRIGGER, INQUIRY_TRIGGER, ], - "sensor": [SENSOR_SPAWN_TRIGGER, SENSOR_EXIT_TRIGGER], + "sensor": [SENSOR_SPAWN_TRIGGER, SENSOR_EXIT_TRIGGER, SENSOR_ABANDONED_TRIGGER], "key_value_pair": [ KEY_VALUE_PAIR_CREATE_TRIGGER, KEY_VALUE_PAIR_UPDATE_TRIGGER, diff --git a/st2common/st2common/models/api/sensor.py b/st2common/st2common/models/api/sensor.py index 3de248bc11..1426037ab4 100644 --- a/st2common/st2common/models/api/sensor.py +++ b/st2common/st2common/models/api/sensor.py @@ -53,6 +53,35 @@ class SensorTypeAPI(BaseAPI): "type": "string", "default": "", }, + # Runtime health fields. These are read-only and are merged in from + # the separate SensorInstanceDB collection by the API controller; + # they are not part of to_model() and are never persisted back onto + # SensorTypeDB. + "status": { + "description": "Current runtime status of the sensor " + "(running, stopped, abandoned). Null if the sensor has never run.", + "type": ["string", "null"], + }, + "hostname": { + "description": "Host of the sensor container which owns this sensor.", + "type": ["string", "null"], + }, + "pid": { + "description": "PID of the sensor process when running.", + "type": ["integer", "null"], + }, + "exit_code": { + "description": "Exit code of the last observed process exit.", + "type": ["integer", "null"], + }, + "respawn_count": { + "description": "Respawn attempts for the current failure streak.", + "type": ["integer", "null"], + }, + "updated_at": { + "description": "Timestamp when the runtime status was last updated.", + "type": ["string", "null"], + }, }, "additionalProperties": False, } diff --git a/st2common/st2common/models/db/__init__.py b/st2common/st2common/models/db/__init__.py index 2782c80b61..63018cc1e7 100644 --- a/st2common/st2common/models/db/__init__.py +++ b/st2common/st2common/models/db/__init__.py @@ -85,6 +85,7 @@ "st2common.models.db.rule_enforcement", "st2common.models.db.runner", "st2common.models.db.sensor", + "st2common.models.db.sensor_instance", "st2common.models.db.trace", "st2common.models.db.trigger", "st2common.models.db.webhook", diff --git a/st2common/st2common/models/db/sensor_instance.py b/st2common/st2common/models/db/sensor_instance.py new file mode 100644 index 0000000000..356f1d9683 --- /dev/null +++ b/st2common/st2common/models/db/sensor_instance.py @@ -0,0 +1,76 @@ +# 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 __future__ import absolute_import +import mongoengine as me + +from st2common.models.db import MongoDBAccess +from st2common.models.db import stormbase +from st2common.fields import ComplexDateTimeField +from st2common.util import date as date_utils + +__all__ = ["SensorInstanceDB"] + + +class SensorInstanceDB(stormbase.StormFoundationDB): + """ + Runtime health/state of a sensor as observed by the sensor container. + + This is intentionally separate from ``SensorTypeDB`` (which holds static + registration metadata and is overwritten on every pack (re-)registration). + A single record exists per sensor ``ref`` and is updated in place on each + lifecycle transition (spawn / exit / abandon). + + Attribute: + ref - Sensor reference ("."); correlation key. + pack - Name of the content pack this sensor belongs to. + status - Current runtime status (running, stopped, abandoned). + hostname - Host of the sensor container which owns this sensor. + pid - PID of the sensor process (when running). + exit_code - Exit code of the last observed process exit. + respawn_count - Number of respawn attempts for the current failure streak. + updated_at - Timestamp of the last status update. + """ + + ref = me.StringField(required=True, unique=True) + pack = me.StringField(required=True) + status = me.StringField( + required=True, help_text="The current runtime status of the sensor." + ) + hostname = me.StringField( + help_text="Host of the sensor container which owns this sensor." + ) + pid = me.IntField(help_text="PID of the sensor process when running.") + exit_code = me.IntField(help_text="Exit code of the last observed process exit.") + respawn_count = me.IntField( + default=0, help_text="Respawn attempts for the current failure streak." + ) + updated_at = ComplexDateTimeField( + default=date_utils.get_datetime_utc_now, + help_text="The timestamp when the status was last updated.", + ) + + meta = { + "indexes": [ + {"fields": ["ref"]}, + {"fields": ["status"]}, + {"fields": ["pack"]}, + ] + } + + +sensor_instance_access = MongoDBAccess(SensorInstanceDB) + +MODELS = [SensorInstanceDB] diff --git a/st2common/st2common/openapi.yaml b/st2common/st2common/openapi.yaml index 5cffb48d1e..fd1096a455 100644 --- a/st2common/st2common/openapi.yaml +++ b/st2common/st2common/openapi.yaml @@ -3195,6 +3195,16 @@ paths: in: query description: Enabled filter type: string + - name: ref + in: query + description: Sensor ref filter (one or more "." refs) + type: array + items: + type: string + - name: status + in: query + description: Runtime status filter (running, stopped, abandoned) + type: string x-parameters: - name: user in: context diff --git a/st2common/st2common/openapi.yaml.j2 b/st2common/st2common/openapi.yaml.j2 index 3bf36161cb..45c98b658b 100644 --- a/st2common/st2common/openapi.yaml.j2 +++ b/st2common/st2common/openapi.yaml.j2 @@ -3191,6 +3191,16 @@ paths: in: query description: Enabled filter type: string + - name: ref + in: query + description: Sensor ref filter (one or more "." refs) + type: array + items: + type: string + - name: status + in: query + description: Runtime status filter (running, stopped, abandoned) + type: string x-parameters: - name: user in: context diff --git a/st2common/st2common/persistence/sensor_instance.py b/st2common/st2common/persistence/sensor_instance.py new file mode 100644 index 0000000000..1d06d32436 --- /dev/null +++ b/st2common/st2common/persistence/sensor_instance.py @@ -0,0 +1,29 @@ +# 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 __future__ import absolute_import + +from st2common.models.db.sensor_instance import sensor_instance_access +from st2common.persistence.base import Access + +__all__ = ["SensorInstance"] + + +class SensorInstance(Access): + impl = sensor_instance_access + + @classmethod + def _get_impl(cls): + return cls.impl diff --git a/st2common/tests/unit/test_db_sensor_instance.py b/st2common/tests/unit/test_db_sensor_instance.py new file mode 100644 index 0000000000..8c41db4de5 --- /dev/null +++ b/st2common/tests/unit/test_db_sensor_instance.py @@ -0,0 +1,90 @@ +# 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 __future__ import absolute_import + +from st2common.constants.sensors import SENSOR_STATUS_RUNNING +from st2common.constants.sensors import SENSOR_STATUS_STOPPED +from st2common.constants.sensors import SENSOR_STATUS_ABANDONED +from st2common.models.db.sensor_instance import SensorInstanceDB +from st2common.persistence.sensor_instance import SensorInstance + +from st2tests import DbTestCase +from st2tests.base import CleanDbTestCase + +from tests.unit.base import BaseDBModelCRUDTestCase + +__all__ = ["SensorInstanceDBModelCRUDTestCase", "SensorInstanceQueryTestCase"] + + +class SensorInstanceDBModelCRUDTestCase(BaseDBModelCRUDTestCase, DbTestCase): + model_class = SensorInstanceDB + persistance_class = SensorInstance + model_class_kwargs = { + "ref": "wolfpack.StupidSensor", + "pack": "wolfpack", + "status": SENSOR_STATUS_RUNNING, + "hostname": "sensor-node-1", + "pid": 1234, + "exit_code": 0, + "respawn_count": 0, + } + update_attribute_name = "status" + # updated_at is populated by a default callable, not part of the kwargs. + skip_check_attribute_names = ["updated_at"] + + +class SensorInstanceQueryTestCase(CleanDbTestCase): + def _create(self, ref, pack, status, **kwargs): + instance_db = SensorInstanceDB(ref=ref, pack=pack, status=status, **kwargs) + return SensorInstance.add_or_update(instance_db) + + def test_query_by_ref(self): + self._create("wolfpack.SensorA", "wolfpack", SENSOR_STATUS_RUNNING, pid=10) + self._create("wolfpack.SensorB", "wolfpack", SENSOR_STATUS_STOPPED) + + instance_db = SensorInstance.query(ref="wolfpack.SensorA").first() + self.assertIsNotNone(instance_db) + self.assertEqual(instance_db.ref, "wolfpack.SensorA") + self.assertEqual(instance_db.status, SENSOR_STATUS_RUNNING) + self.assertEqual(instance_db.pid, 10) + + def test_query_by_status(self): + self._create("wolfpack.SensorA", "wolfpack", SENSOR_STATUS_RUNNING) + self._create("wolfpack.SensorB", "wolfpack", SENSOR_STATUS_ABANDONED) + self._create("wolfpack.SensorC", "wolfpack", SENSOR_STATUS_ABANDONED) + + abandoned = list(SensorInstance.query(status=SENSOR_STATUS_ABANDONED)) + self.assertEqual(len(abandoned), 2) + refs = sorted(instance.ref for instance in abandoned) + self.assertEqual(refs, ["wolfpack.SensorB", "wolfpack.SensorC"]) + + def test_status_transition_upsert_by_ref(self): + # Simulate the container upsert pattern: look up by ref, mutate in place, + # persist - the record count stays at one and the id is stable. + ref = "wolfpack.SensorA" + created = self._create(ref, "wolfpack", SENSOR_STATUS_RUNNING, pid=99) + + instance_db = SensorInstance.query(ref=ref).first() + instance_db.status = SENSOR_STATUS_ABANDONED + instance_db.exit_code = 1 + instance_db.respawn_count = 2 + updated = SensorInstance.add_or_update(instance_db) + + self.assertEqual(created.id, updated.id) + self.assertEqual(SensorInstance.query(ref=ref).count(), 1) + self.assertEqual(updated.status, SENSOR_STATUS_ABANDONED) + self.assertEqual(updated.exit_code, 1) + self.assertEqual(updated.respawn_count, 2) diff --git a/st2reactor/st2reactor/container/process_container.py b/st2reactor/st2reactor/container/process_container.py index 890bcccbb9..9ee2db0719 100644 --- a/st2reactor/st2reactor/container/process_container.py +++ b/st2reactor/st2reactor/container/process_container.py @@ -19,6 +19,7 @@ import sys import time import json +import socket import subprocess from collections import defaultdict @@ -28,13 +29,23 @@ from st2common import log as logging from st2common.util import concurrency +from st2common.util import date as date_utils from st2common.constants.error_messages import PACK_VIRTUALENV_DOESNT_EXIST from st2common.constants.system import API_URL_ENV_VARIABLE_NAME from st2common.constants.system import AUTH_TOKEN_ENV_VARIABLE_NAME from st2common.constants.triggers import SENSOR_SPAWN_TRIGGER, SENSOR_EXIT_TRIGGER +from st2common.constants.triggers import SENSOR_ABANDONED_TRIGGER +from st2common.constants.sensors import SENSOR_STATUS_RUNNING +from st2common.constants.sensors import SENSOR_STATUS_STOPPED +from st2common.constants.sensors import SENSOR_STATUS_ABANDONED +from st2common.constants.sensors import DEFAULT_SENSOR_MAX_RESPAWN_COUNT +from st2common.constants.sensors import DEFAULT_SENSOR_RESPAWN_DELAY +from st2common.constants.sensors import DEFAULT_SENSOR_RESPAWN_BACKOFF_FACTOR from st2common.constants.exit_codes import SUCCESS_EXIT_CODE from st2common.constants.exit_codes import FAILURE_EXIT_CODE +from st2common.models.db.sensor_instance import SensorInstanceDB from st2common.models.system.common import ResourceReference +from st2common.persistence.sensor_instance import SensorInstance from st2common.services.access import create_token from st2common.transport.reactor import TriggerDispatcher from st2common.util.api import get_full_public_api_url @@ -52,15 +63,19 @@ WRAPPER_SCRIPT_NAME = "sensor_wrapper.py" WRAPPER_SCRIPT_PATH = os.path.join(BASE_DIR, WRAPPER_SCRIPT_NAME) -# How many times to try to subsequently respawn a sensor after a non-zero exit before giving up -SENSOR_MAX_RESPAWN_COUNTS = 2 +# How many times to try to subsequently respawn a sensor after a non-zero exit before giving up. +# Note: This is the default; the effective value is configurable via +# [sensorcontainer].max_respawn_count. +SENSOR_MAX_RESPAWN_COUNTS = DEFAULT_SENSOR_MAX_RESPAWN_COUNT # How many seconds after the sensor has been started we should wait before considering sensor as # being started and running successfully SENSOR_SUCCESSFUL_START_THRESHOLD = 10 -# How long to wait (in seconds) before respawning a dead process -SENSOR_RESPAWN_DELAY = 2.5 +# How long to wait (in seconds) before respawning a dead process. +# Note: This is the default; the effective value is configurable via +# [sensorcontainer].respawn_delay. +SENSOR_RESPAWN_DELAY = DEFAULT_SENSOR_RESPAWN_DELAY # How long to wait for process to exit after sending SIGTERM signal. If the process doesn't # exit in this amount of seconds, SIGKILL signal will be sent to the process. @@ -114,6 +129,25 @@ def __init__( self._dispatcher = dispatcher or TriggerDispatcher(LOG) + self._hostname = socket.gethostname() + + # Respawn / retry behavior (configurable via the [sensorcontainer] group). + # Fall back to the module-level defaults if the config group is not + # registered (e.g. when the container is constructed outside a fully + # configured service). + sensor_container_cfg = getattr(cfg.CONF, "sensorcontainer", None) + self._max_respawn_count = getattr( + sensor_container_cfg, "max_respawn_count", DEFAULT_SENSOR_MAX_RESPAWN_COUNT + ) + self._respawn_delay = getattr( + sensor_container_cfg, "respawn_delay", DEFAULT_SENSOR_RESPAWN_DELAY + ) + self._respawn_backoff_factor = getattr( + sensor_container_cfg, + "respawn_backoff_factor", + DEFAULT_SENSOR_RESPAWN_BACKOFF_FACTOR, + ) + self._stopped = False self._exit_code = None # exit code with which this process should exit @@ -381,6 +415,9 @@ def _spawn_sensor_process(self, sensor): self._sensor_start_times[sensor_id] = int(time.time()) self._dispatch_trigger_for_sensor_spawn(sensor=sensor, process=process, cmd=cmd) + self._update_sensor_instance( + sensor=sensor, status=SENSOR_STATUS_RUNNING, pid=process.pid + ) return process @@ -399,6 +436,10 @@ def _stop_sensor_process(self, sensor_id, exit_timeout=PROCESS_EXIT_TIMEOUT): """ process = self._processes[sensor_id] + # Grab a reference to the sensor object before it is deleted below so we + # can record its stopped state once the process has exited. + sensor = self._sensors.get(sensor_id) + # Delete sensor before terminating process so that it will not be # respawned during termination self._delete_sensor(sensor_id) @@ -423,6 +464,11 @@ def _stop_sensor_process(self, sensor_id, exit_timeout=PROCESS_EXIT_TIMEOUT): # Process hasn't exited yet, forcefully kill it process.kill() + if sensor: + self._update_sensor_instance( + sensor=sensor, status=SENSOR_STATUS_STOPPED, exit_code=status + ) + def _respawn_sensor(self, sensor_id, sensor, exit_code): """ Method for respawning a sensor which died with a non-zero exit code. @@ -450,12 +496,38 @@ def _respawn_sensor(self, sensor_id, sensor, exit_code): if not should_respawn: LOG.debug("Not respawning a dead sensor", extra=extra) + + # Distinguish a permanent give-up (crashed and exceeded the max + # respawn attempts) from an intentional clean exit (exit_code == 0). + # Only the former is an "abandoned" sensor worth alerting on. + respawn_count = self._sensor_respawn_counts[sensor_id] + if exit_code != 0 and respawn_count >= self._max_respawn_count: + LOG.warning( + "Sensor %s abandoned after %s respawn attempts", + sensor_id, + respawn_count, + extra=extra, + ) + self._dispatch_trigger_for_sensor_abandon( + sensor=sensor, exit_code=exit_code, respawn_count=respawn_count + ) + self._update_sensor_instance( + sensor=sensor, + status=SENSOR_STATUS_ABANDONED, + exit_code=exit_code, + ) return LOG.debug("Respawning dead sensor", extra=extra) self._sensor_respawn_counts[sensor_id] += 1 - sleep_delay = SENSOR_RESPAWN_DELAY * self._sensor_respawn_counts[sensor_id] + respawn_count = self._sensor_respawn_counts[sensor_id] + # Wait respawn_delay * backoff_factor ** (attempt - 1) seconds before + # respawning. With the default backoff factor of 1 this is a constant + # respawn_delay; a factor > 1 grows the delay exponentially per attempt. + sleep_delay = self._respawn_delay * ( + self._respawn_backoff_factor ** (respawn_count - 1) + ) concurrency.sleep(sleep_delay) try: @@ -475,7 +547,7 @@ def _should_respawn_sensor(self, sensor_id, sensor, exit_code): return False respawn_count = self._sensor_respawn_counts[sensor_id] - if respawn_count >= SENSOR_MAX_RESPAWN_COUNTS: + if respawn_count >= self._max_respawn_count: LOG.debug("Sensor has already been respawned max times, giving up") return False @@ -543,6 +615,53 @@ def _dispatch_trigger_for_sensor_exit(self, sensor, exit_code): payload = {"id": sensor["class_name"], "timestamp": now, "exit_code": exit_code} self._dispatcher.dispatch(trigger, payload=payload) + def _dispatch_trigger_for_sensor_abandon(self, sensor, exit_code, respawn_count): + trigger = ResourceReference.to_string_reference( + name=SENSOR_ABANDONED_TRIGGER["name"], pack=SENSOR_ABANDONED_TRIGGER["pack"] + ) + now = int(time.time()) + payload = { + "id": sensor["class_name"], + "timestamp": now, + "exit_code": exit_code, + "respawn_count": respawn_count, + } + self._dispatcher.dispatch(trigger, payload=payload) + + def _update_sensor_instance(self, sensor, status, exit_code=None, pid=None): + """ + Upsert the persisted runtime health record (SensorInstanceDB) for a sensor. + + A single record exists per sensor ``ref`` and is updated in place on each + lifecycle transition. Failures here must never crash the container loop, + so all DB errors are caught and logged. + """ + ref = sensor["ref"] + + try: + instance_db = SensorInstance.query(ref=ref).first() + + if not instance_db: + instance_db = SensorInstanceDB( + ref=ref, pack=sensor["pack"], status=status + ) + + instance_db.pack = sensor["pack"] + instance_db.status = status + instance_db.hostname = self._hostname + instance_db.pid = pid + instance_db.exit_code = exit_code + instance_db.respawn_count = self._sensor_respawn_counts[ref] + instance_db.updated_at = date_utils.get_datetime_utc_now() + + SensorInstance.add_or_update( + instance_db, publish=False, dispatch_trigger=False + ) + except Exception: + LOG.warning( + "Failed to update sensor instance state for %s", ref, exc_info=True + ) + def _delete_sensor(self, sensor_id): """ Delete / reset all the internal state about a particular sensor. diff --git a/st2reactor/st2reactor/sensor/config.py b/st2reactor/st2reactor/sensor/config.py index f54a602167..7e3ff8bedd 100644 --- a/st2reactor/st2reactor/sensor/config.py +++ b/st2reactor/st2reactor/sensor/config.py @@ -19,6 +19,9 @@ from st2common import config as st2cfg from st2common.constants.sensors import DEFAULT_PARTITION_LOADER +from st2common.constants.sensors import DEFAULT_SENSOR_MAX_RESPAWN_COUNT +from st2common.constants.sensors import DEFAULT_SENSOR_RESPAWN_DELAY +from st2common.constants.sensors import DEFAULT_SENSOR_RESPAWN_BACKOFF_FACTOR from st2common.constants.system import VERSION_STRING from st2common.constants.system import DEFAULT_CONFIG_FILE_PATH @@ -92,6 +95,33 @@ def _register_sensor_container_opts(ignore_errors=False): other_opts, group="sensorcontainer", ignore_errors=ignore_errors ) + # Sensor respawn / retry options + respawn_opts = [ + cfg.IntOpt( + "max_respawn_count", + default=DEFAULT_SENSOR_MAX_RESPAWN_COUNT, + help="Maximum number of times to respawn a sensor after it exits with a " + "non-zero code before giving up and firing the " + "st2.sensor.process_abandoned trigger.", + ), + cfg.FloatOpt( + "respawn_delay", + default=DEFAULT_SENSOR_RESPAWN_DELAY, + help="Base delay (in seconds) to wait between sensor respawn attempts.", + ), + cfg.FloatOpt( + "respawn_backoff_factor", + default=DEFAULT_SENSOR_RESPAWN_BACKOFF_FACTOR, + help="Exponential backoff multiplier applied to respawn_delay per attempt. " + "1 (default) means a constant delay between attempts; a value > 1 grows " + "the delay exponentially (respawn_delay * factor ** (attempt - 1)).", + ), + ] + + st2cfg.do_register_opts( + respawn_opts, group="sensorcontainer", ignore_errors=ignore_errors + ) + # CLI options cli_opts = [ cfg.StrOpt( diff --git a/st2reactor/tests/unit/test_process_container.py b/st2reactor/tests/unit/test_process_container.py index 747618a4f0..baf33d3ac9 100644 --- a/st2reactor/tests/unit/test_process_container.py +++ b/st2reactor/tests/unit/test_process_container.py @@ -20,7 +20,10 @@ from mock import MagicMock, Mock, patch import unittest +from oslo_config import cfg + from st2reactor.container.process_container import ProcessSensorContainer +from st2reactor.container.process_container import SENSOR_MAX_RESPAWN_COUNTS from st2common.util import concurrency from st2common.models.db.pack import PackDB from st2common.persistence.pack import Pack @@ -180,3 +183,164 @@ def test_dispatch_triggers_on_spawn_exit(self): "exit_code": 1, }, ) + + @patch.object(time, "time", MagicMock(return_value=1439441533)) + def test_dispatch_trigger_for_sensor_abandon(self): + mock_dispatcher = Mock() + process_container = ProcessSensorContainer( + None, poll_interval=0.1, dispatcher=mock_dispatcher + ) + sensor = {"class_name": "pack.StupidSensor"} + + process_container._dispatch_trigger_for_sensor_abandon( + sensor, exit_code=1, respawn_count=SENSOR_MAX_RESPAWN_COUNTS + ) + mock_dispatcher.dispatch.assert_called_with( + "core.st2.sensor.process_abandoned", + payload={ + "id": "pack.StupidSensor", + "timestamp": 1439441533, + "exit_code": 1, + "respawn_count": SENSOR_MAX_RESPAWN_COUNTS, + }, + ) + + @patch.object(time, "time", MagicMock(return_value=1439441533)) + def test_respawn_dispatches_abandoned_after_max_respawns(self): + mock_dispatcher = Mock() + process_container = ProcessSensorContainer( + None, poll_interval=0.1, dispatcher=mock_dispatcher + ) + # Avoid touching the database for the health record update. + process_container._update_sensor_instance = Mock() + + sensor_id = "wolfpack.StupidSensor" + sensor = {"class_name": sensor_id, "ref": sensor_id, "pack": "wolfpack"} + + # Simulate a sensor which has already been respawned the maximum number + # of times and just crashed again with a non-zero exit code. + process_container._sensor_respawn_counts[sensor_id] = SENSOR_MAX_RESPAWN_COUNTS + process_container._respawn_sensor( + sensor_id=sensor_id, sensor=sensor, exit_code=1 + ) + + mock_dispatcher.dispatch.assert_called_with( + "core.st2.sensor.process_abandoned", + payload={ + "id": sensor_id, + "timestamp": 1439441533, + "exit_code": 1, + "respawn_count": SENSOR_MAX_RESPAWN_COUNTS, + }, + ) + process_container._update_sensor_instance.assert_called_once() + + def test_respawn_clean_exit_does_not_dispatch_abandoned(self): + mock_dispatcher = Mock() + process_container = ProcessSensorContainer( + None, poll_interval=0.1, dispatcher=mock_dispatcher + ) + process_container._update_sensor_instance = Mock() + + sensor_id = "wolfpack.StupidSensor" + sensor = {"class_name": sensor_id, "ref": sensor_id, "pack": "wolfpack"} + + # A clean exit (exit_code == 0) must never be treated as "abandoned". + process_container._sensor_respawn_counts[sensor_id] = SENSOR_MAX_RESPAWN_COUNTS + process_container._respawn_sensor( + sensor_id=sensor_id, sensor=sensor, exit_code=0 + ) + + self.assertFalse(mock_dispatcher.dispatch.called) + self.assertFalse(process_container._update_sensor_instance.called) + + def test_respawn_settings_default_from_config(self): + process_container = ProcessSensorContainer(None, poll_interval=0.1) + self.assertEqual( + process_container._max_respawn_count, + cfg.CONF.sensorcontainer.max_respawn_count, + ) + self.assertEqual( + process_container._respawn_delay, cfg.CONF.sensorcontainer.respawn_delay + ) + self.assertEqual( + process_container._respawn_backoff_factor, + cfg.CONF.sensorcontainer.respawn_backoff_factor, + ) + + def test_max_respawn_count_config_override(self): + # With a higher max_respawn_count, a sensor that has been respawned the + # old default number of times should still be respawned (not abandoned). + cfg.CONF.set_override("max_respawn_count", 5, group="sensorcontainer") + try: + mock_dispatcher = Mock() + process_container = ProcessSensorContainer( + None, poll_interval=0.1, dispatcher=mock_dispatcher + ) + self.assertEqual(process_container._max_respawn_count, 5) + + sensor_id = "wolfpack.StupidSensor" + self.assertTrue( + process_container._should_respawn_sensor( + sensor_id=sensor_id, sensor={}, exit_code=1 + ) + ) + + # At respawn_count == 5 (the configured max) it must give up. + process_container._sensor_respawn_counts[sensor_id] = 5 + self.assertFalse( + process_container._should_respawn_sensor( + sensor_id=sensor_id, sensor={}, exit_code=1 + ) + ) + finally: + cfg.CONF.clear_override("max_respawn_count", group="sensorcontainer") + + @patch.object(ProcessSensorContainer, "_spawn_sensor_process", MagicMock()) + @patch.object(concurrency, "sleep", MagicMock()) + def test_respawn_delay_constant_with_default_backoff(self): + # Default backoff factor of 1 -> constant respawn_delay between attempts. + cfg.CONF.set_override("respawn_delay", 3.0, group="sensorcontainer") + cfg.CONF.set_override("respawn_backoff_factor", 1.0, group="sensorcontainer") + try: + process_container = ProcessSensorContainer(None, poll_interval=0.1) + sensor_id = "wolfpack.StupidSensor" + sensor = {"class_name": sensor_id, "ref": sensor_id, "pack": "wolfpack"} + + process_container._respawn_sensor( + sensor_id=sensor_id, sensor=sensor, exit_code=1 + ) + concurrency.sleep.assert_called_with(3.0) + + # Second attempt: still constant with a backoff factor of 1. + process_container._respawn_sensor( + sensor_id=sensor_id, sensor=sensor, exit_code=1 + ) + concurrency.sleep.assert_called_with(3.0) + finally: + cfg.CONF.clear_override("respawn_delay", group="sensorcontainer") + cfg.CONF.clear_override("respawn_backoff_factor", group="sensorcontainer") + + @patch.object(ProcessSensorContainer, "_spawn_sensor_process", MagicMock()) + @patch.object(concurrency, "sleep", MagicMock()) + def test_respawn_delay_exponential_backoff(self): + # backoff factor of 2 -> delay grows exponentially: 2.5, 5.0, 10.0, ... + cfg.CONF.set_override("respawn_delay", 2.5, group="sensorcontainer") + cfg.CONF.set_override("respawn_backoff_factor", 2.0, group="sensorcontainer") + # Raise the cap so all three attempts respawn. + cfg.CONF.set_override("max_respawn_count", 10, group="sensorcontainer") + try: + process_container = ProcessSensorContainer(None, poll_interval=0.1) + sensor_id = "wolfpack.StupidSensor" + sensor = {"class_name": sensor_id, "ref": sensor_id, "pack": "wolfpack"} + + expected = [2.5, 5.0, 10.0] + for expected_delay in expected: + process_container._respawn_sensor( + sensor_id=sensor_id, sensor=sensor, exit_code=1 + ) + concurrency.sleep.assert_called_with(expected_delay) + finally: + cfg.CONF.clear_override("respawn_delay", group="sensorcontainer") + cfg.CONF.clear_override("respawn_backoff_factor", group="sensorcontainer") + cfg.CONF.clear_override("max_respawn_count", group="sensorcontainer") diff --git a/st2tests/st2tests/config.py b/st2tests/st2tests/config.py index ebfddc8397..4f9313a6ff 100644 --- a/st2tests/st2tests/config.py +++ b/st2tests/st2tests/config.py @@ -26,6 +26,9 @@ from st2common.constants.garbage_collection import DEFAULT_COLLECTION_INTERVAL from st2common.constants.garbage_collection import DEFAULT_SLEEP_DELAY from st2common.constants.sensors import DEFAULT_PARTITION_LOADER +from st2common.constants.sensors import DEFAULT_SENSOR_MAX_RESPAWN_COUNT +from st2common.constants.sensors import DEFAULT_SENSOR_RESPAWN_DELAY +from st2common.constants.sensors import DEFAULT_SENSOR_RESPAWN_BACKOFF_FACTOR from st2tests.fixturesloader import get_fixtures_packs_base_path CONF = cfg.CONF @@ -467,6 +470,29 @@ def _register_sensor_container_opts(): _register_opts(other_opts, group="sensorcontainer") + # Sensor respawn / retry options + respawn_opts = [ + cfg.IntOpt( + "max_respawn_count", + default=DEFAULT_SENSOR_MAX_RESPAWN_COUNT, + help="Maximum number of times to respawn a sensor after it exits with a " + "non-zero code before giving up and firing the " + "st2.sensor.process_abandoned trigger.", + ), + cfg.FloatOpt( + "respawn_delay", + default=DEFAULT_SENSOR_RESPAWN_DELAY, + help="Base delay (in seconds) to wait between sensor respawn attempts.", + ), + cfg.FloatOpt( + "respawn_backoff_factor", + default=DEFAULT_SENSOR_RESPAWN_BACKOFF_FACTOR, + help="Exponential backoff multiplier applied to respawn_delay per attempt.", + ), + ] + + _register_opts(respawn_opts, group="sensorcontainer") + # CLI options cli_opts = [ cfg.StrOpt( From 2cc08b9ec5a76f2a826725644f96eaf65f590d1c Mon Sep 17 00:00:00 2001 From: guzzijones12 Date: Mon, 31 Aug 2026 10:21:17 -0400 Subject: [PATCH 2/7] comment on explicit sensor fields; catch mongoengine exception instead of exception --- st2api/st2api/controllers/v1/sensors.py | 9 +++++++-- 1 file changed, 7 insertions(+), 2 deletions(-) diff --git a/st2api/st2api/controllers/v1/sensors.py b/st2api/st2api/controllers/v1/sensors.py index 457a1d7927..ede1e52a9c 100644 --- a/st2api/st2api/controllers/v1/sensors.py +++ b/st2api/st2api/controllers/v1/sensors.py @@ -15,6 +15,7 @@ import six from mongoengine import ValidationError +from mongoengine.errors import MongoEngineException from st2common import log as logging from st2common.persistence.sensor import SensorType @@ -54,6 +55,10 @@ class SensorTypeController(resource.ContentPackResourceController): options = {"sort": ["pack", "name"]} # Runtime health fields merged in from SensorInstanceDB (keyed by sensor ref). + # This is an explicit allow-list: it must stay in sync with the read-only + # health fields declared on the SensorTypeAPI schema. Adding a field to + # SensorInstanceDB does not expose it here until it is added to both places. + # (updated_at is handled separately in _apply_health.) HEALTH_ATTRIBUTES = [ "status", "hostname", @@ -83,7 +88,7 @@ def get_all( status=status, only_fields=["ref"] ) ] - except Exception: + except MongoEngineException: LOG.warning( "Failed to resolve sensor refs for status filter", exc_info=True ) @@ -168,7 +173,7 @@ def _get_health_by_ref(self, refs): try: instances = SensorInstance.query(ref__in=refs) return {instance.ref: instance for instance in instances} - except Exception: + except MongoEngineException: LOG.warning("Failed to load sensor health records", exc_info=True) return {} From 238221c55e3182d83c4b6dd77ac769751358a932 Mon Sep 17 00:00:00 2001 From: guzzijones12 Date: Thu, 27 Aug 2026 15:59:21 -0400 Subject: [PATCH 3/7] fix orquesta hash --- lockfiles/st2.lock | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/lockfiles/st2.lock b/lockfiles/st2.lock index 546f12698e..7a9f94672c 100644 --- a/lockfiles/st2.lock +++ b/lockfiles/st2.lock @@ -3085,7 +3085,7 @@ "artifacts": [ { "algorithm": "sha256", - "hash": "491767e81c1bb11a54fb68d1a24119bdeede593a2beccca5bc09bfed36fdb35c", + "hash": "b9feb1769b48102061fe4fc59b2f5ad600bc2ac0b55cf12ef5fe49464ac0d230", "url": "git+https://github.com/StackStorm/orquesta.git" } ], From 2e06edbd5e4dddfefd0ab91fc835eb3cb8df00e1 Mon Sep 17 00:00:00 2001 From: guzzijones12 Date: Mon, 31 Aug 2026 10:39:14 -0400 Subject: [PATCH 4/7] sensor changelog entry --- CHANGELOG.rst | 3 +++ 1 file changed, 3 insertions(+) diff --git a/CHANGELOG.rst b/CHANGELOG.rst index a2a2872e46..9b19ec06ec 100644 --- a/CHANGELOG.rst +++ b/CHANGELOG.rst @@ -44,6 +44,9 @@ Added exit code, respawn count, last updated). The health fields are merged into the ``/v1/sensors`` API response and shown by ``st2 sensor list`` / ``st2 sensor get``, and can be filtered with ``st2 sensor list --status=abandoned``. +* Added a ``status`` column to ``st2 sensor list`` and the sensor runtime health fields (``status``, + ``hostname``, ``pid``, ``exit_code``, ``respawn_count``, ``updated_at``) to ``st2 sensor get`` output + in the ``st2`` CLI. (by @guzzijones12@gmail.com) * Added ``[sensorcontainer].max_respawn_count``, ``[sensorcontainer].respawn_delay`` and ``[sensorcontainer].respawn_backoff_factor`` config options to control how many times a crashed sensor is respawned before being abandoned and how long to wait between attempts. The backoff From efa0a38aef8b0a1560c90d427bb90ed9326349d1 Mon Sep 17 00:00:00 2001 From: guzzijones12 Date: Mon, 31 Aug 2026 10:41:26 -0400 Subject: [PATCH 5/7] update api schema gen --- contrib/schemas/sensor.json | 42 +++++++++++++++++++++++++++++++++++++ 1 file changed, 42 insertions(+) diff --git a/contrib/schemas/sensor.json b/contrib/schemas/sensor.json index 6193bf521f..ae66c6a023 100644 --- a/contrib/schemas/sensor.json +++ b/contrib/schemas/sensor.json @@ -89,6 +89,48 @@ "description": "Path to the metadata file relative to the pack directory.", "type": "string", "default": "" + }, + "status": { + "description": "Current runtime status of the sensor (running, stopped, abandoned). Null if the sensor has never run.", + "type": [ + "string", + "null" + ] + }, + "hostname": { + "description": "Host of the sensor container which owns this sensor.", + "type": [ + "string", + "null" + ] + }, + "pid": { + "description": "PID of the sensor process when running.", + "type": [ + "integer", + "null" + ] + }, + "exit_code": { + "description": "Exit code of the last observed process exit.", + "type": [ + "integer", + "null" + ] + }, + "respawn_count": { + "description": "Respawn attempts for the current failure streak.", + "type": [ + "integer", + "null" + ] + }, + "updated_at": { + "description": "Timestamp when the runtime status was last updated.", + "type": [ + "string", + "null" + ] } }, "additionalProperties": false From f4a0a78f691c0f302bfae8e662e8b655c4047b60 Mon Sep 17 00:00:00 2001 From: guzzijones12 Date: Mon, 31 Aug 2026 11:17:38 -0400 Subject: [PATCH 6/7] add new python file to pants --- st2common/BUILD | 1 + 1 file changed, 1 insertion(+) diff --git a/st2common/BUILD b/st2common/BUILD index 48a0726dfd..daee88493d 100644 --- a/st2common/BUILD +++ b/st2common/BUILD @@ -66,6 +66,7 @@ st2_component_python_distribution( "./st2common/models/db/timer.py", # used by st2api "./st2common/models/db/webhook.py", # used by st2api "./st2common/persistence/execution_queue.py", # used by st2scheduler (in st2actions) + "./st2common/persistence/sensor_instance.py", # used by st2api and sensor container (in st2reactor) "./st2common/stream", # used by st2stream "./st2common/transport/consumers.py", # used by st2actions- and st2reactor-related services "./st2common/util/actionalias_helpstring.py", # used by st2api From 314989da8bef56e63bed46c5ae6aeb8ab0928942 Mon Sep 17 00:00:00 2001 From: guzzijones12 Date: Mon, 31 Aug 2026 11:48:58 -0400 Subject: [PATCH 7/7] filter out updated_at for sensor db api --- st2api/st2api/controllers/v1/sensors.py | 28 ++++++++++++++----- .../unit/controllers/v1/test_sensortypes.py | 22 +++++++++++++++ 2 files changed, 43 insertions(+), 7 deletions(-) diff --git a/st2api/st2api/controllers/v1/sensors.py b/st2api/st2api/controllers/v1/sensors.py index ede1e52a9c..f613acca03 100644 --- a/st2api/st2api/controllers/v1/sensors.py +++ b/st2api/st2api/controllers/v1/sensors.py @@ -54,11 +54,7 @@ class SensorTypeController(resource.ContentPackResourceController): options = {"sort": ["pack", "name"]} - # Runtime health fields merged in from SensorInstanceDB (keyed by sensor ref). - # This is an explicit allow-list: it must stay in sync with the read-only - # health fields declared on the SensorTypeAPI schema. Adding a field to - # SensorInstanceDB does not expose it here until it is added to both places. - # (updated_at is handled separately in _apply_health.) + # Explicit allow-list of runtime health fields merged in from SensorInstanceDB; must stay in sync with the read-only fields on SensorTypeAPI (updated_at handled separately in _apply_health). HEALTH_ATTRIBUTES = [ "status", "hostname", @@ -67,6 +63,9 @@ class SensorTypeController(resource.ContentPackResourceController): "respawn_count", ] + # Synthetic fields merged from SensorInstanceDB (not SensorTypeDB fields); stripped from include/exclude lists before querying and re-applied in _apply_health. + _SYNTHETIC_HEALTH_FIELDS = set(HEALTH_ATTRIBUTES) | {"updated_at"} + def get_all( self, exclude_attributes=None, @@ -77,8 +76,7 @@ def get_all( requester_user=None, **raw_filters, ): - # "status" is not a SensorTypeDB field - it lives in SensorInstanceDB. - # Resolve it to the set of matching sensor refs and constrain the query. + # "status" lives in SensorInstanceDB, not SensorTypeDB - resolve it to matching refs and constrain the query. status = raw_filters.pop("status", None) if status: try: @@ -101,6 +99,22 @@ def get_all( raw_filters["ref"] = refs + # Health fields are merged in post-query; strip them from include/exclude lists but keep "ref" for correlation. + if include_attributes: + include_attributes = [ + attribute + for attribute in include_attributes + if attribute not in self._SYNTHETIC_HEALTH_FIELDS + ] + if "ref" not in include_attributes: + include_attributes.append("ref") + if exclude_attributes: + exclude_attributes = [ + attribute + for attribute in exclude_attributes + if attribute not in self._SYNTHETIC_HEALTH_FIELDS + ] + return super(SensorTypeController, self)._get_all( exclude_fields=exclude_attributes, include_fields=include_attributes, diff --git a/st2api/tests/unit/controllers/v1/test_sensortypes.py b/st2api/tests/unit/controllers/v1/test_sensortypes.py index 23b2bbad03..77e5cde3a3 100644 --- a/st2api/tests/unit/controllers/v1/test_sensortypes.py +++ b/st2api/tests/unit/controllers/v1/test_sensortypes.py @@ -206,6 +206,28 @@ def test_get_all_status_filter_no_matches(self): self.assertEqual(resp.status_int, http_client.OK) self.assertEqual(len(resp.json), 0) + def test_get_all_include_attributes_with_health_fields(self): + # The CLI (st2 sensor list) requests health fields via include_attributes. + # They are not SensorTypeDB fields, so they must be stripped before the + # query (otherwise mongoengine 400s) and still merged in from the health + # record afterwards. This mirrors the request st2-self-check issues. + running_ref = f"{DUMMY_PACK_1}.SampleSensor" + self._create_health_record( + running_ref, DUMMY_PACK_1, SENSOR_STATUS_RUNNING, pid=100 + ) + + resp = self.app.get( + "/v1/sensortypes?include_attributes=ref,pack,enabled,status,updated_at" + ) + self.assertEqual(resp.status_int, http_client.OK) + item_by_ref = {item["ref"]: item for item in resp.json} + # DB-backed included field is present. + self.assertIn("pack", item_by_ref[running_ref]) + # Synthetic health fields are merged in even though they were stripped + # from the DB query. + self.assertEqual(item_by_ref[running_ref]["status"], SENSOR_STATUS_RUNNING) + self.assertIsNotNone(item_by_ref[running_ref]["updated_at"]) + def test_disable_and_enable_sensor(self): # Verify initial state resp = self.app.get(f"/v1/sensortypes/{DUMMY_PACK_1}.SampleSensor")