diff --git a/plugins/endpoints/api/routes.py b/plugins/endpoints/api/routes.py index 17bac052..44355086 100644 --- a/plugins/endpoints/api/routes.py +++ b/plugins/endpoints/api/routes.py @@ -27,7 +27,7 @@ from shopdb.api import ( from ..models import ( EndpointState, EndpointEvent, EndpointCommand, EndpointApplication, - OperatorReport, + EndpointHealthState, OperatorReport, ) from ..commands import COMMAND_ALLOWLIST, is_allowed_action, ALLOWED_ACTIONS_FOR, row_actions_for, requires_target, action_targetkind, elevated_permission_for from ..eventcatalog import CRITICAL_EVENT_IDS, event_severity, event_title @@ -258,7 +258,10 @@ def list_endpoints(): # one query per endpoint for machines and another for events. assetids = [state.assetid for state, *_ in items] machines = _machines_for_assets(assetids) - critical = _critical_counts_for_assets(assetids) + states = {state.assetid: state for state, *_ in items} + computertypes = {state.assetid: computertypeid + for state, _computerid, computertypeid, *_ in items} + health = _health_for_assets(states, computertypes) data = [] for state, computerid, computertypeid, computertype, assetnumber, rank in items: @@ -270,7 +273,7 @@ def list_endpoints(): row['uptimedisplay'] = _format_uptime(state.uptimeseconds) row['status'] = ('never', 'quiet', 'reporting')[rank] row['machines'] = machines.get(state.assetid, []) - row['health'] = _health_for(state, critical.get(state.assetid)) + row['health'] = health.get(state.assetid) data.append(row) return paginated_response(data, page, per_page, total) @@ -352,68 +355,41 @@ def _machines_for_assets(assetids): return out -def _critical_counts_for_assets(assetids): - """{assetid: (total, dominant_eventid)} of critical events in the window. +def _health_for_assets(states, computertypes): + """{assetid: {'level', 'label'}} for a page, from the cached rollup. - Broken down BY EVENT ID, not just totalled. A bare count says "418 - critical", which tells an operator nothing about what to do; naming the - event that dominates - service timeout, application crash - is the - difference between a number and a lead. + ONE query for the page instead of the old per-page critical-event + aggregate, and no ranking work here - the verdict was decided by + services/health.py in the same commit as the telemetry it came from. + + Endpoints with no cached row have not checked in since the rollup shipped. + Those are evaluated read-only for display rather than left blank, because a + blank Health column reads as "fine" and not doing that is this page's whole + job. Nothing is persisted on a GET: writing here would invent a firstseenat + no agent ever observed, and would make a page load look like an + observation. """ - if not assetids: + if not states: return {} - cutoff = _naive_utcnow() - timedelta(hours=48) - rows = (db.session.query(EndpointEvent.assetid, EndpointEvent.eventid, - db.func.count(EndpointEvent.endpointeventid)) - .filter(EndpointEvent.assetid.in_(assetids), - EndpointEvent.eventid.in_(CRITICAL_EVENT_IDS), - EndpointEvent.eventtime.isnot(None), - EndpointEvent.eventtime >= cutoff) - .group_by(EndpointEvent.assetid, EndpointEvent.eventid) - .all()) - perasset = {} - for assetid, eventid, count in rows: - total, dominant, best = perasset.get(assetid, (0, None, 0)) - total += count - if count > best: - dominant, best = eventid, count - perasset[assetid] = (total, dominant, best) - return {assetid: (total, dominant) - for assetid, (total, dominant, _best) in perasset.items()} + from ..services.health import _verdict + assetids = list(states) + cached = {row.assetid: row for row in + EndpointHealthState.query + .filter(EndpointHealthState.assetid.in_(assetids)) + .all()} -def _health_for(state, critical): - """Worst-of rollup: a stopped watched service, or a recent critical event. - - Deliberately a PHRASE, not a score. "eDNC stopped" tells an operator what - to do; a percentage does not - and neither does "418 critical", which is - why the dominant event is named rather than merely counted. - """ - if critical: - total, dominant = critical - title = event_title(dominant) if dominant else 'critical' - label = title if total == 1 else '{} x{}'.format(title, total) - return {'level': 'critical', 'label': label} - - stopped = [] - try: - for entry in json.loads(state.servicesjson or '[]'): - if not _service_running(entry): - stopped.append(_service_name(entry)) - except (TypeError, ValueError): - # Malformed payload is not a health verdict - say nothing rather than - # claiming the endpoint is unhealthy. - return {'level': 'unknown', 'label': None} - - if stopped: - first = stopped[0] - extra = f' +{len(stopped) - 1}' if len(stopped) > 1 else '' - return {'level': 'warning', 'label': f'{first} stopped{extra}'} - - if state.lastreportedat is None: - return {'level': 'unknown', 'label': 'no agent'} - return {'level': 'ok', 'label': 'ok'} + now = _naive_utcnow() + out = {} + for assetid, state in states.items(): + row = cached.get(assetid) + if row is not None: + out[assetid] = {'level': row.level, 'label': row.label} + continue + verdict = _verdict(state, computertypes.get(assetid), assetid, now) + out[assetid] = {'level': verdict['level'], 'label': verdict['label']} + return out @endpoints_bp.route('/', methods=['GET']) diff --git a/plugins/endpoints/eventcatalog.py b/plugins/endpoints/eventcatalog.py index d382eeee..82389ac8 100644 --- a/plugins/endpoints/eventcatalog.py +++ b/plugins/endpoints/eventcatalog.py @@ -29,8 +29,15 @@ EVENT_CATALOG = { # code) and is the one that says what actually broke; 1001 is the WER bucket # record Windows writes alongside it; 1002 is a hang rather than a crash, so # it is a warning - the app is unresponsive, not gone. + # + # 1001 is deliberately NOT critical. Windows writes it ALONGSIDE 1000 for the + # same crash, so counting both double-counted every fault: a box with 2200 + # crashes reported "Application crash x4400". It also won the dominant-event + # tie often enough to displace 1000, which is the record that actually names + # the faulting module. Kept curated so the history panel still shows it, but + # it no longer drives the fleet health verdict - 1000 speaks for the crash. 1000: ('Application error', 'critical', 'Application'), - 1001: ('Application crash', 'critical', 'Application'), + 1001: ('Application crash', 'info', 'Application'), 1002: ('Application hang', 'warning', 'Application'), } diff --git a/plugins/endpoints/frontend/views/EndpointsList.vue b/plugins/endpoints/frontend/views/EndpointsList.vue index 7b5ab62d..e640bb8b 100644 --- a/plugins/endpoints/frontend/views/EndpointsList.vue +++ b/plugins/endpoints/frontend/views/EndpointsList.vue @@ -276,23 +276,26 @@ onMounted(() => { async function loadFilterOptions() { // listAll, not list: perpage is clamped to 100 server-side, so a long type or // location list would silently lose its tail in a dropdown. - try { - computertypes.value = await computersApi.types.listAll() - } catch (error) { - computertypes.value = [] - } - try { - locations.value = await locationsApi.listAll() - } catch (error) { - locations.value = [] - } + // Issued together: neither list feeds the other, so awaiting them in sequence + // only added a round trip to first paint. Each keeps its own fallback so one + // failing dropdown does not empty the other. + const [types, places] = await Promise.all([ + computersApi.types.listAll().catch(() => []), + locationsApi.listAll().catch(() => []) + ]) + computertypes.value = types + locations.value = places } async function loadSummary() { try { - const response = await endpointsApi.dashboardSummary() + // Issued together: the stability call does not depend on the summary, so + // sequencing them just stacked the slower request behind the faster one. + const [response, stability] = await Promise.all([ + endpointsApi.dashboardSummary(), + endpointsApi.dashboardStability() + ]) summary.value = response.data.data || null - const stability = await endpointsApi.dashboardStability() criticalcount.value = (stability.data.data || []).length } catch (error) { // The strip is a convenience, not the page. If it fails the table still diff --git a/plugins/endpoints/migrations/versions/0017_endpoint_health_rollup.py b/plugins/endpoints/migrations/versions/0017_endpoint_health_rollup.py new file mode 100644 index 00000000..34f77ae4 --- /dev/null +++ b/plugins/endpoints/migrations/versions/0017_endpoint_health_rollup.py @@ -0,0 +1,133 @@ +"""Add endpoint health rollup tables. + +Three tables behind the Fleet Health column: + + endpointhealthstates one cached verdict per endpoint, read directly by + the Fleet list instead of recomputing a 48h + fleet-wide event aggregate on every page load + endpointhealthtransitions capped history, written on CHANGE not on sample + endpointmonitorimpacts per-PC-type operational impact for a monitor + +No backfill. An endpoint with no row has simply not checked in since this +shipped, and the first check-in writes one. Backfilling would have to invent a +firstseenat for every existing condition, and a fabricated "broken since" is +worse than an absent one - an operator reading it has no way to tell it apart +from an observed timestamp. + +endpointmonitorimpacts ships EMPTY on purpose. Absence of a row means the +built-in default impact applies, so an endpoint with no policies behaves +exactly as it did before this migration. Seeding defaults here would freeze +today's guesses into rows nobody could distinguish from deliberate choices. + +Idempotent: every create checks the live schema first, so a re-run and a +db.create_all() site both stay clean on strict MySQL 8. +""" +from alembic import op + +import sqlalchemy as sa + + +# revision identifiers, used by Alembic. +revision = 'endpoints0017healthrollup' +down_revision = 'endpoints0016dellupdates' +branch_labels = None +depends_on = None + +HEALTHSTATES = 'endpointhealthstates' +TRANSITIONS = 'endpointhealthtransitions' +IMPACTS = 'endpointmonitorimpacts' + + +def _has_table(table_name): + from sqlalchemy import inspect + + return inspect(op.get_bind()).has_table(table_name) + + +def upgrade(): + if not _has_table(HEALTHSTATES): + op.create_table( + HEALTHSTATES, + sa.Column('endpointhealthstateid', sa.Integer(), nullable=False), + sa.Column('assetid', sa.Integer(), nullable=False), + sa.Column('level', sa.String(length=16), nullable=False, + server_default='unknown'), + sa.Column('label', sa.String(length=255), nullable=True), + sa.Column('impactclass', sa.String(length=16), nullable=True), + sa.Column('sourcekind', sa.String(length=16), nullable=False, + server_default='none'), + sa.Column('sourceref', sa.String(length=64), nullable=True), + sa.Column('firstseenat', sa.DateTime(), nullable=True), + sa.Column('lastconfirmedat', sa.DateTime(), nullable=True), + sa.Column('lastevaluatedat', sa.DateTime(), nullable=False), + sa.Column('flapcount', sa.Integer(), nullable=False, + server_default='0'), + sa.ForeignKeyConstraint(['assetid'], ['assets.assetid'], + ondelete='CASCADE'), + sa.PrimaryKeyConstraint('endpointhealthstateid'), + # One live verdict per endpoint. Unique rather than indexed: two + # rows for one asset is not a state this table should represent. + sa.UniqueConstraint('assetid', + name='uq_endpointhealthstates_assetid'), + ) + op.create_index('idx_endpointhealthstates_asset', HEALTHSTATES, + ['assetid']) + op.create_index('idx_endpointhealthstates_impactclass', HEALTHSTATES, + ['impactclass']) + + if not _has_table(TRANSITIONS): + op.create_table( + TRANSITIONS, + sa.Column('endpointhealthtransitionid', sa.Integer(), + nullable=False), + sa.Column('assetid', sa.Integer(), nullable=False), + sa.Column('level', sa.String(length=16), nullable=False), + sa.Column('label', sa.String(length=255), nullable=True), + sa.Column('impactclass', sa.String(length=16), nullable=True), + sa.Column('sourcekind', sa.String(length=16), nullable=False, + server_default='none'), + sa.Column('sourceref', sa.String(length=64), nullable=True), + sa.Column('openedat', sa.DateTime(), nullable=False), + sa.Column('resolvedat', sa.DateTime(), nullable=True), + sa.ForeignKeyConstraint(['assetid'], ['assets.assetid'], + ondelete='CASCADE'), + sa.PrimaryKeyConstraint('endpointhealthtransitionid'), + ) + # Retention reads newest-first per asset; the resolved index serves the + # "never evict an unresolved row" filter without scanning history. + op.create_index('idx_endpointhealthtransitions_assetopened', + TRANSITIONS, ['assetid', 'openedat']) + op.create_index('idx_endpointhealthtransitions_assetresolved', + TRANSITIONS, ['assetid', 'resolvedat']) + + if not _has_table(IMPACTS): + op.create_table( + IMPACTS, + sa.Column('endpointmonitorimpactid', sa.Integer(), nullable=False), + sa.Column('endpointmonitorid', sa.Integer(), nullable=False), + # Plain integer, not a ForeignKey: computertypes belongs to the + # computers plugin and owns its own migration chain. A cross-plugin + # FK would couple the two chains' ordering for no benefit the + # unique constraint does not already provide. + sa.Column('computertypeid', sa.Integer(), nullable=False), + sa.Column('impactclass', sa.String(length=16), nullable=False, + server_default='degraded'), + sa.Column('createdat', sa.DateTime(), nullable=False), + sa.ForeignKeyConstraint(['endpointmonitorid'], + ['endpointmonitors.endpointmonitorid'], + ondelete='CASCADE'), + sa.PrimaryKeyConstraint('endpointmonitorimpactid'), + sa.UniqueConstraint('endpointmonitorid', 'computertypeid', + name='uq_endpointmonitorimpacts_monitortype'), + ) + op.create_index('idx_endpointmonitorimpacts_computertypeid', IMPACTS, + ['computertypeid']) + + +def downgrade(): + # Dropped newest-first so the FK to endpointmonitors goes before anything + # that might reference it. All three are caches or mappings - nothing here + # is a system of record, so a downgrade loses no telemetry. + for table_name in (IMPACTS, TRANSITIONS, HEALTHSTATES): + if _has_table(table_name): + op.drop_table(table_name) diff --git a/plugins/endpoints/models/__init__.py b/plugins/endpoints/models/__init__.py index 929540f3..6912fa78 100644 --- a/plugins/endpoints/models/__init__.py +++ b/plugins/endpoints/models/__init__.py @@ -5,8 +5,12 @@ from .endpoint import ( ) from .monitor import EndpointMonitor from .operatorreport import OperatorReport +from .health import ( + EndpointHealthState, EndpointHealthTransition, EndpointMonitorImpact, +) __all__ = [ 'EndpointState', 'EndpointEvent', 'EndpointCommand', 'EndpointApplication', 'EndpointMonitor', 'OperatorReport', + 'EndpointHealthState', 'EndpointHealthTransition', 'EndpointMonitorImpact', ] diff --git a/plugins/endpoints/models/health.py b/plugins/endpoints/models/health.py new file mode 100644 index 00000000..c91ddefe --- /dev/null +++ b/plugins/endpoints/models/health.py @@ -0,0 +1,267 @@ +"""Endpoint health rollup: cached verdict, transition history, impact mapping. + +Three tables, one job: decide what the Fleet Health column says about an +endpoint, decide it ONCE when the data changes, and keep enough history to +answer "when did this break". + +Why this exists at all. The verdict used to be computed on every Fleet page +load by aggregating 48h of endpointevents fleet-wide, which cost more the +longer the fleet ran, and it ranked a raw Windows event count above every +other signal. A box with a crash loop reported "Application crash x5861" while +its dead eDNC service - the thing that actually stops work on the floor - was +structurally unreachable, because the event branch returned first. + +The model here separates two things the old rollup conflated: + + events Point-in-time facts. They happened, they are counted, they age + out of the window on their own. + conditions A stopped service or a disconnected drive is not an event, it is + a STATE with a duration. It opens, it persists, and it is only + over when something observes it to be over. + +Ranking a condition against an event count is comparing incompatible things. +Conditions are pinned until explicitly resolved; events are ranked beneath +them. That is the whole reason the Health column can be trusted now. +""" + +from datetime import datetime + +from shopdb.api import db + + +def _utciso(value): + """Naive-UTC datetime -> ISO string marked UTC, or None.""" + return value.isoformat() + 'Z' if value else None + + +# Operational impact, in the site's terms rather than the OS's. Windows calls a +# crash severe; the floor calls a disconnected drive a stoppage. These disagree, +# and the floor wins - which is why this vocabulary is not the event catalog's +# 'critical'/'warning'/'info'. +IMPACT_BLOCKER = 'blocker' +IMPACT_DEGRADED = 'degraded' +IMPACT_INFORMATIONAL = 'informational' + +IMPACT_CLASSES = (IMPACT_BLOCKER, IMPACT_DEGRADED, IMPACT_INFORMATIONAL) + +# Lower sorts worse, matching _DRIVE_STATUS_RANK's existing convention so the +# two ladders in this plugin do not read in opposite directions. +IMPACT_RANK = { + IMPACT_BLOCKER: 0, + IMPACT_DEGRADED: 1, + IMPACT_INFORMATIONAL: 2, +} + +# Where a verdict came from. Kept alongside sourceref so the UI can deep-link +# back to the monitor or the event that produced the label. +SOURCE_DRIVE = 'drive' +SOURCE_SERVICE = 'service' +SOURCE_PROCESS = 'process' +SOURCE_EVENT = 'event' +SOURCE_NONE = 'none' + + +class EndpointHealthState(db.Model): + """The current health verdict for one endpoint - the Fleet page's only read. + + Cached deliberately. The Fleet list joins this and reads it directly; it + does not recompute anything per row. Written in the SAME commit as the + check-in that produced it, so the cached verdict and the telemetry it + reflects can never disagree (the ledger pattern this codebase already uses + for quantityonhand). + """ + + __tablename__ = 'endpointhealthstates' + + endpointhealthstateid = db.Column(db.Integer, primary_key=True) + + # One row per asset. Unique rather than a plain index: two live verdicts + # for one endpoint is not a state this table should be able to represent. + assetid = db.Column( + db.Integer, + db.ForeignKey('assets.assetid', ondelete='CASCADE'), + nullable=False, + unique=True, + ) + + # Display vocabulary, matching what the Fleet renderer already understands: + # 'ok', 'warning', 'critical', 'unknown'. + level = db.Column(db.String(16), nullable=False, default='unknown') + + # The phrase an operator reads ("eDNC stopped", "S: disconnected"). A + # PHRASE, not a score - it has to say what to do. + label = db.Column(db.String(255), nullable=True) + + # Null for an ok/unknown verdict: nothing is wrong, so nothing has impact. + impactclass = db.Column(db.String(16), nullable=True) + + sourcekind = db.Column(db.String(16), nullable=False, default=SOURCE_NONE) + + # Monitor id, drive letter or Windows event id, depending on sourcekind. + # Free text because those identifier spaces have nothing in common. + sourceref = db.Column(db.String(64), nullable=True) + + # When the CURRENT condition opened. Survives re-evaluation: a service that + # has been down for three days keeps its original firstseenat, so the UI can + # say how long rather than only that it is down now. + firstseenat = db.Column(db.DateTime, nullable=True) + + # Last time an agent explicitly observed this endpoint. Drives staleness: + # a pinned blocker on an endpoint that stopped reporting is neither + # resolvable nor trustworthy, and must not render as present tense. + lastconfirmedat = db.Column(db.DateTime, nullable=True) + + lastevaluatedat = db.Column(db.DateTime, nullable=False, + default=datetime.utcnow) + + # Transitions inside the trailing window. A service flapping ten times an + # hour is a different fault from one cleanly down, and dispatching someone + # to a service that has already restarted twice wastes a trip. + flapcount = db.Column(db.Integer, nullable=False, default=0) + + __table_args__ = ( + db.Index('idx_endpointhealthstates_asset', 'assetid'), + db.Index('idx_endpointhealthstates_impactclass', 'impactclass'), + ) + + def __repr__(self): + return f"" + + def to_dict(self): + return { + 'assetid': self.assetid, + 'level': self.level, + 'label': self.label, + 'impactclass': self.impactclass, + 'sourcekind': self.sourcekind, + 'sourceref': self.sourceref, + 'firstseenat': _utciso(self.firstseenat), + 'lastconfirmedat': _utciso(self.lastconfirmedat), + 'lastevaluatedat': _utciso(self.lastevaluatedat), + 'flapcount': self.flapcount, + } + + +class EndpointHealthTransition(db.Model): + """One verdict CHANGE for an endpoint. Written on change, never on sample. + + Storing a row per check-in would make the last 50 rows cover a few minutes + and say nothing. Storing a row per change makes 50 rows cover months on a + stable endpoint, compresses "nothing happened" to zero cost, and answers + the question operators actually ask - when did this break - directly rather + than by inferring it from a gap between samples. + + It also makes flapping measurable: 50 transitions in an hour and 50 over a + year are the same row count and obviously different faults. + """ + + __tablename__ = 'endpointhealthtransitions' + + endpointhealthtransitionid = db.Column(db.Integer, primary_key=True) + + assetid = db.Column( + db.Integer, + db.ForeignKey('assets.assetid', ondelete='CASCADE'), + nullable=False, + ) + + level = db.Column(db.String(16), nullable=False) + label = db.Column(db.String(255), nullable=True) + impactclass = db.Column(db.String(16), nullable=True) + sourcekind = db.Column(db.String(16), nullable=False, default=SOURCE_NONE) + sourceref = db.Column(db.String(64), nullable=True) + + openedat = db.Column(db.DateTime, nullable=False, default=datetime.utcnow) + + # Null means STILL OPEN. Retention never evicts an unresolved row: at 50 + # deep a long-running blocker's opening transition would otherwise age out + # behind a burst of unrelated churn, deleting the record of why the thing + # currently on screen is on screen. + resolvedat = db.Column(db.DateTime, nullable=True) + + __table_args__ = ( + db.Index('idx_endpointhealthtransitions_assetopened', + 'assetid', 'openedat'), + db.Index('idx_endpointhealthtransitions_assetresolved', + 'assetid', 'resolvedat'), + ) + + def __repr__(self): + return (f"") + + def to_dict(self): + return { + 'endpointhealthtransitionid': self.endpointhealthtransitionid, + 'assetid': self.assetid, + 'level': self.level, + 'label': self.label, + 'impactclass': self.impactclass, + 'sourcekind': self.sourcekind, + 'sourceref': self.sourceref, + 'openedat': _utciso(self.openedat), + 'resolvedat': _utciso(self.resolvedat), + } + + +class EndpointMonitorImpact(db.Model): + """How much a given monitor matters ON A GIVEN PC TYPE. + + eDNC stopped on a shopfloor PC halts work; the same service stopped on a + lobby display is noise. Impact is therefore a property of the monitor IN + CONTEXT, not of the monitor and not of the event type - which is why it + cannot live on endpointmonitors and could never have lived in the event + catalog. + + Why a plugin-owned side table rather than a column on the policy action: + monitors bind to PC types through configurationpolicyactions, which is a + CORE table shared by every plugin. Adding an endpoints-specific impact + column there would be a contract-surface change imposed on unrelated + plugins. This keeps the concept inside the plugin that owns it. + + ABSENCE OF A ROW IS THE DEFAULT, and that is load-bearing. An endpoint with + no policies, no PC type, or no monitors at all resolves to the built-in + default impact for its source kind - the same verdict the old rollup gave. + The no-policy path is the NORMAL path here, not a special case. + """ + + __tablename__ = 'endpointmonitorimpacts' + + endpointmonitorimpactid = db.Column(db.Integer, primary_key=True) + + endpointmonitorid = db.Column( + db.Integer, + db.ForeignKey('endpointmonitors.endpointmonitorid', + ondelete='CASCADE'), + nullable=False, + ) + + # Plain integer, not a ForeignKey: computertypes belongs to the computers + # plugin and owns its own migration chain. A cross-plugin FK would couple + # the two chains' ordering for no benefit the unique constraint below does + # not already provide. + computertypeid = db.Column(db.Integer, nullable=False) + + impactclass = db.Column(db.String(16), nullable=False, + default=IMPACT_DEGRADED) + + createdat = db.Column(db.DateTime, nullable=False, default=datetime.utcnow) + + __table_args__ = ( + db.UniqueConstraint('endpointmonitorid', 'computertypeid', + name='uq_endpointmonitorimpacts_monitortype'), + db.Index('idx_endpointmonitorimpacts_computertypeid', 'computertypeid'), + ) + + def __repr__(self): + return (f"") + + def to_dict(self): + return { + 'endpointmonitorimpactid': self.endpointmonitorimpactid, + 'endpointmonitorid': self.endpointmonitorid, + 'computertypeid': self.computertypeid, + 'impactclass': self.impactclass, + 'createdat': _utciso(self.createdat), + } diff --git a/plugins/endpoints/plugin.py b/plugins/endpoints/plugin.py index f3cdb870..e573648f 100644 --- a/plugins/endpoints/plugin.py +++ b/plugins/endpoints/plugin.py @@ -1274,6 +1274,27 @@ class EndpointsPlugin(BasePlugin): # flush, not commit: the collector dispatcher owns the transaction. db.session.flush() + # Health rollup, AFTER the flush so the events just written are visible + # to the evaluator's query in this same transaction. Recomputed here, + # per asset, rather than by the Fleet page across the whole fleet on + # every load - the verdict changes when telemetry changes, so this is + # where it belongs, and it lands in the SAME commit as the telemetry it + # was derived from. + # + # Fail-open, matching the reconciler above: a health error must never + # fail an otherwise-good check-in. The previous verdict simply stands, + # which is correct - a failed evaluation is not evidence of health. + try: + from .services.health import evaluate_endpoint_health + evaluate_endpoint_health( + assetid, state, + computertypeid=getattr(computer, 'computertypeid', None)) + db.session.flush() + except Exception: + logger.exception( + 'health evaluation failed for asset %s; leaving verdict ' + 'unchanged', assetid) + return { 'action': action, 'assetid': assetid, diff --git a/plugins/endpoints/services/health.py b/plugins/endpoints/services/health.py new file mode 100644 index 00000000..cac91f48 --- /dev/null +++ b/plugins/endpoints/services/health.py @@ -0,0 +1,416 @@ +"""Evaluate one endpoint's health verdict and record it when it changes. + +Scoped to ONE assetid on purpose. The old rollup aggregated 48h of events +fleet-wide on every Fleet page load; this runs when the data actually changes +(a check-in, or an explicit refresh from PC detail) and narrows to a single +asset, which the existing idx_endpointevents_assetevent (assetid, eventtime) +composite already serves. The Fleet page then reads a cached row per endpoint +and recomputes nothing. + +THE ORDERING RULE, which is the whole point of this module: + + 1. Unresolved BLOCKER conditions (drive down, blocking service stopped) + 2. DEGRADED conditions (stalled service, non-blocking stop) + 3. Events, ranked by catalog severity + 4. ok / unknown + +A condition outranks an event count no matter how large the count. That is the +inversion of the old behaviour, where the event branch returned first and a +crash loop permanently hid a dead service. + +THE NO-POLICY RULE, which is what keeps it safe: + + A policy binding may only RAISE impact. It may never gate PRESENCE. + +An endpoint with no policies, no PC type, or no monitors still gets a verdict. +Drives and events are not policy-gated at all, so they work with zero +configuration; services fall back to a built-in default impact. Absence of an +EndpointMonitorImpact row is the normal path, not an error path. +""" + +import json +import logging +from datetime import datetime, timedelta + +from shopdb.api import db + +from ..eventcatalog import CRITICAL_EVENT_IDS, event_severity, event_title +from ..models.endpoint import EndpointEvent +from ..models.health import ( + EndpointHealthState, EndpointHealthTransition, EndpointMonitorImpact, + IMPACT_BLOCKER, IMPACT_DEGRADED, IMPACT_INFORMATIONAL, IMPACT_RANK, + SOURCE_DRIVE, SOURCE_EVENT, SOURCE_NONE, SOURCE_PROCESS, SOURCE_SERVICE, +) + +logger = logging.getLogger(__name__) + +# Keep the last 50 transitions per asset, but do not pay for a DELETE on every +# write. Let history grow to the high-water mark and then trim back, which +# amortises the delete to roughly once per ten transitions and leaves the write +# path a plain insert almost every time. +TRANSITION_RETENTION = 50 +TRANSITION_HIGHWATER = 60 + +# Window for the flap counter and the event rollup. Matches the 48h the +# stability card used, so the number on screen does not change meaning. +HEALTH_WINDOW_HOURS = 48 + +# A drive that is unavailable stops work on the floor. This is the one signal +# that is a blocker with NO configuration at all - it never consults policy, +# so it survives an endpoint that has none. What counts as unavailable is +# _mapped_drive_rows' decision, not a second list maintained here. + +# Default impact when no EndpointMonitorImpact row applies. This is the +# no-policy path, and it reproduces the old rollup's verdict rather than +# inventing a new one: a stopped watched service was a warning before and is +# 'degraded' now. +_DEFAULT_IMPACT_BY_SOURCE = { + SOURCE_DRIVE: IMPACT_BLOCKER, + SOURCE_SERVICE: IMPACT_DEGRADED, + SOURCE_PROCESS: IMPACT_DEGRADED, + SOURCE_EVENT: IMPACT_INFORMATIONAL, +} + +_LEVEL_BY_IMPACT = { + IMPACT_BLOCKER: 'critical', + IMPACT_DEGRADED: 'warning', + IMPACT_INFORMATIONAL: 'warning', +} + + +def _naive_utcnow(): + return datetime.utcnow() + + +def impact_for_monitor(endpointmonitorid, computertypeid): + """Impact class for a monitor on a PC type, or None if unmapped. + + Returns None rather than raising or defaulting when there is no mapping, + no PC type, or no monitor id. The caller supplies the default; this + function only reports what policy explicitly said, so "policy said + degraded" and "policy said nothing" stay distinguishable. + """ + if endpointmonitorid is None or computertypeid is None: + return None + row = (EndpointMonitorImpact.query + .filter_by(endpointmonitorid=endpointmonitorid, + computertypeid=computertypeid) + .first()) + return row.impactclass if row is not None else None + + +def _drive_conditions(state): + """Open drive conditions from the last user-context snapshot. + + Not policy-gated: mapped drives arrive with the user-context payload and + are evaluated for every endpoint regardless of configuration. On a shop + floor a dead share is a stoppage, so this is a blocker by default and does + not wait for anyone to declare it one. + """ + from ..api.routes import _mapped_drive_rows + + raw = getattr(state, 'usercontextjson', None) + if not raw: + return [] + + try: + context = json.loads(raw) + except (TypeError, ValueError): + # A malformed payload is not a health verdict. Say nothing rather than + # claiming the endpoint is either broken or fine. + logger.warning('malformed usercontextjson for asset %s', + getattr(state, 'assetid', None)) + return [] + + if not isinstance(context, dict): + return [] + + out = [] + for row in _mapped_drive_rows(context): + # Reuse the card's own verdict rather than re-deriving one from the + # raw status string. _mapped_drive_rows already resolves the two + # conflicting sources, treats unknown as unhealthy, and is the thing + # the PC detail card renders - so the Health column and the drive card + # cannot disagree about whether a drive is down. + if row['statecolor'] == 'green': + continue + out.append({ + 'impactclass': IMPACT_BLOCKER, + 'sourcekind': SOURCE_DRIVE, + 'sourceref': row['drive'], + 'label': '{} {}'.format(row['drive'], row['statuslabel']), + }) + return out + + +def _service_conditions(state, computertypeid): + """Open service conditions from the last services snapshot. + + Distinguishes STALLED from STOPPED, which the old rollup could not: it + read a running/not-running boolean, so a service sitting in stop-pending + was reported identically to one cleanly down. _service_transitional() has + existed for a while; nothing was calling it. + + Policy-gated only in the sense that an endpoint with no monitors reports no + services and therefore has no service conditions - which is the existing + contract, not a regression. Where a service IS reported, an unmapped + monitor falls back to the built-in default impact. + """ + from ..api.routes import (_service_name, _service_running, + _service_transitional) + + raw = getattr(state, 'servicesjson', None) + if not raw: + # Nothing watched, or nothing reported. Distinct from malformed: this + # is "no condition", and it must read as ok rather than unknown or the + # Fleet page would light up every unpolicied endpoint. + return [] + + try: + entries = json.loads(raw) or [] + except (TypeError, ValueError): + # Malformed IS unknown - we genuinely cannot tell. Signalled to the + # caller with a sentinel so it does not get mistaken for "healthy". + return None + + out = [] + for entry in entries: + if not isinstance(entry, dict): + continue + if _service_running(entry): + continue + + name = _service_name(entry) + monitorid = entry.get('monitorid') or entry.get('endpointmonitorid') + raw_state = entry.get('state') or entry.get('status') or '' + stalled = _service_transitional(raw_state) + + mapped = impact_for_monitor(monitorid, computertypeid) + if mapped is not None: + impactclass = mapped + elif stalled: + # A stalled service MAY be a blocker but we cannot know without + # someone saying so. Degraded until policy says otherwise - it is + # visible, but it does not cry wolf. + impactclass = IMPACT_DEGRADED + else: + impactclass = _DEFAULT_IMPACT_BY_SOURCE[SOURCE_SERVICE] + + kind = (entry.get('kind') or SOURCE_SERVICE).strip().lower() + sourcekind = SOURCE_PROCESS if kind == 'process' else SOURCE_SERVICE + + out.append({ + 'impactclass': impactclass, + 'sourcekind': sourcekind, + 'sourceref': str(monitorid) if monitorid is not None else name, + 'label': '{} {}'.format(name, 'stalled' if stalled else 'stopped'), + }) + return out + + +def _event_condition(assetid, cutoff): + """Worst recent event for one asset, as a pseudo-condition, or None. + + Narrowed to a single assetid so the (assetid, eventtime) composite serves + it. Counts are still shown, but a count can no longer outrank a real + condition - it is ranked beneath every one of them. + """ + if not CRITICAL_EVENT_IDS: + return None + + rows = (db.session.query(EndpointEvent.eventid, + db.func.count(EndpointEvent.endpointeventid)) + .filter(EndpointEvent.assetid == assetid, + EndpointEvent.eventid.in_(CRITICAL_EVENT_IDS), + EndpointEvent.eventtime.isnot(None), + EndpointEvent.eventtime >= cutoff) + .group_by(EndpointEvent.eventid) + .all()) + if not rows: + return None + + total = sum(count for _eventid, count in rows) + dominant = max(rows, key=lambda row: row[1])[0] + title = event_title(dominant) + label = title if total == 1 else '{} x{}'.format(title, total) + + severity = event_severity(dominant) + impactclass = (IMPACT_DEGRADED if severity == 'critical' + else IMPACT_INFORMATIONAL) + + return { + 'impactclass': impactclass, + 'sourcekind': SOURCE_EVENT, + 'sourceref': str(dominant), + 'label': label, + } + + +def _verdict(state, computertypeid, assetid, now): + """Resolve the single worst signal into a verdict dict. + + Order is impact class first, source second. A flat "conditions always beat + events" rule would eventually put a stopped print spooler above filesystem + corruption; ranking by impact avoids that while still guaranteeing a + blocker outranks any event. + """ + cutoff = now - timedelta(hours=HEALTH_WINDOW_HOURS) + + services = _service_conditions(state, computertypeid) + if services is None: + return { + 'level': 'unknown', + 'label': None, + 'impactclass': None, + 'sourcekind': SOURCE_NONE, + 'sourceref': None, + } + + candidates = list(_drive_conditions(state)) + list(services) + + event = _event_condition(assetid, cutoff) + if event is not None: + candidates.append(event) + + if not candidates: + if getattr(state, 'lastreportedat', None) is None: + return { + 'level': 'unknown', + 'label': 'no agent', + 'impactclass': None, + 'sourcekind': SOURCE_NONE, + 'sourceref': None, + } + return { + 'level': 'ok', + 'label': 'ok', + 'impactclass': None, + 'sourcekind': SOURCE_NONE, + 'sourceref': None, + } + + # Conditions before events at equal impact: a live condition is actionable + # now, an event is a report of something that already finished happening. + def sort_key(candidate): + return ( + IMPACT_RANK.get(candidate['impactclass'], 99), + 1 if candidate['sourcekind'] == SOURCE_EVENT else 0, + ) + + worst = sorted(candidates, key=sort_key)[0] + return { + 'level': _LEVEL_BY_IMPACT.get(worst['impactclass'], 'warning'), + 'label': worst['label'], + 'impactclass': worst['impactclass'], + 'sourcekind': worst['sourcekind'], + 'sourceref': worst['sourceref'], + } + + +def _same_condition(existing, verdict): + """Is this the same ongoing condition, or a new one? + + Compared on identity (what and where), NOT on the label: a crash count + ticking from x12 to x13 is the same condition and must not open a new + transition, or the history would fill with noise and the flap counter + would be meaningless. + """ + if existing is None: + return False + return (existing.level == verdict['level'] + and existing.impactclass == verdict['impactclass'] + and existing.sourcekind == verdict['sourcekind'] + and existing.sourceref == verdict['sourceref']) + + +def _trim_transitions(assetid): + """Enforce retention, keeping every unresolved row regardless of age.""" + total = (db.session.query(db.func.count( + EndpointHealthTransition.endpointhealthtransitionid)) + .filter(EndpointHealthTransition.assetid == assetid) + .scalar()) or 0 + if total <= TRANSITION_HIGHWATER: + return + + keep = [row.endpointhealthtransitionid for row in + (db.session.query( + EndpointHealthTransition.endpointhealthtransitionid) + .filter(EndpointHealthTransition.assetid == assetid) + .order_by(EndpointHealthTransition.openedat.desc()) + .limit(TRANSITION_RETENTION) + .all())] + + query = (EndpointHealthTransition.query + .filter(EndpointHealthTransition.assetid == assetid, + EndpointHealthTransition.resolvedat.isnot(None))) + if keep: + query = query.filter( + ~EndpointHealthTransition.endpointhealthtransitionid.in_(keep)) + query.delete(synchronize_session=False) + + +def evaluate_endpoint_health(assetid, state, computertypeid=None, now=None): + """Recompute one endpoint's verdict and record a transition if it changed. + + Does NOT commit. The caller owns the transaction so the verdict lands in + the same commit as the telemetry it was derived from - the cached row and + the data behind it can never disagree. + + Fail-open is the caller's job too: a health error must never fail an + otherwise-good check-in. + """ + now = now or _naive_utcnow() + verdict = _verdict(state, computertypeid, assetid, now) + + existing = (EndpointHealthState.query + .filter_by(assetid=assetid) + .first()) + + unchanged = _same_condition(existing, verdict) + + if existing is None: + existing = EndpointHealthState(assetid=assetid) + db.session.add(existing) + + if not unchanged: + # Close the outgoing condition before opening the new one, so the two + # never overlap in history. + (EndpointHealthTransition.query + .filter(EndpointHealthTransition.assetid == assetid, + EndpointHealthTransition.resolvedat.is_(None)) + .update({'resolvedat': now}, synchronize_session=False)) + + db.session.add(EndpointHealthTransition( + assetid=assetid, + level=verdict['level'], + label=verdict['label'], + impactclass=verdict['impactclass'], + sourcekind=verdict['sourcekind'], + sourceref=verdict['sourceref'], + openedat=now, + )) + + existing.firstseenat = now + window = now - timedelta(hours=HEALTH_WINDOW_HOURS) + existing.flapcount = (db.session.query(db.func.count( + EndpointHealthTransition.endpointhealthtransitionid)) + .filter(EndpointHealthTransition.assetid == assetid, + EndpointHealthTransition.openedat >= window) + .scalar()) or 0 + + _trim_transitions(assetid) + + existing.level = verdict['level'] + existing.label = verdict['label'] + existing.impactclass = verdict['impactclass'] + existing.sourcekind = verdict['sourcekind'] + existing.sourceref = verdict['sourceref'] + existing.lastevaluatedat = now + + # Only an explicit observation confirms an endpoint. A verdict recomputed + # from a stale snapshot must NOT refresh this, or a pinned blocker on a + # silent endpoint would look freshly verified forever. + if getattr(state, 'lastreportedat', None) is not None: + existing.lastconfirmedat = state.lastreportedat + + return existing diff --git a/tests/test_plugins/test_endpoints_health.py b/tests/test_plugins/test_endpoints_health.py new file mode 100644 index 00000000..fe3bc999 --- /dev/null +++ b/tests/test_plugins/test_endpoints_health.py @@ -0,0 +1,320 @@ +"""Health rollup: conditions outrank event counts, and no policy still works. + +The bug this locks down: the Fleet Health column ranked a raw Windows event +count above every other signal, so a box with a crash loop reported +"Application error x5861" while its dead eDNC service was structurally +unreachable - the event branch returned before services were ever considered. + +The other half is the no-policy path. An endpoint with no policies, no PC type, +and no monitor impact rows is the NORMAL case here, not an edge case, and it +must still produce a verdict. A policy binding may only RAISE impact; it may +never gate whether health is computed at all. +""" + +import json +from datetime import datetime, timedelta + +import pytest + +from shopdb.core.models import Asset, AssetType +from plugins.endpoints.models import ( + EndpointState, EndpointEvent, EndpointMonitor, + EndpointHealthState, EndpointHealthTransition, EndpointMonitorImpact, +) +from plugins.endpoints.models.health import ( + IMPACT_BLOCKER, IMPACT_DEGRADED, SOURCE_DRIVE, SOURCE_EVENT, + SOURCE_SERVICE, +) +from plugins.endpoints.services.health import ( + HEALTH_WINDOW_HOURS, TRANSITION_HIGHWATER, TRANSITION_RETENTION, + evaluate_endpoint_health, +) + +NOW = datetime(2025, 1, 15, 12, 0, 0) + +# A curated CRITICAL id, so the event path is genuinely exercised rather than +# filtered out by the catalog. 1000 is 'Application error'; note 1001 was +# deliberately demoted to informational and would not surface here. +CRASH_EVENTID = 1000 + + +@pytest.fixture +def endpoint(db): + """One computer asset with a state row. No policies, no PC type.""" + assettype = AssetType.query.filter_by(assettype='computer').first() + if not assettype: + assettype = AssetType(assettype='computer') + db.session.add(assettype) + db.session.flush() + + asset = Asset(assetnumber='HEALTH-PC-1', assettypeid=assettype.assettypeid) + db.session.add(asset) + db.session.flush() + + state = EndpointState(assetid=asset.assetid, lastreportedat=NOW) + db.session.add(state) + db.session.commit() + return state + + +def _services(*entries): + return json.dumps(list(entries)) + + +def _drives(*statuses): + """A usercontext payload with one netuse drive per (letter, status).""" + return json.dumps({ + 'netuse': [{'localname': letter, 'remotename': r'\\server\share', + 'status': status} + for letter, status in statuses] + }) + + +def _events(db, assetid, count, hoursago=1): + for index in range(count): + db.session.add(EndpointEvent( + assetid=assetid, + eventid=CRASH_EVENTID, + eventrecordid=index + 1, + eventchannel='Application', + eventtime=NOW - timedelta(hours=hoursago), + )) + db.session.commit() + + +def _monitor(db, servicename): + monitor = EndpointMonitor(kind='service', displayname=servicename, + servicename=servicename) + db.session.add(monitor) + db.session.flush() + return monitor + + +def test_no_policy_endpoint_still_gets_a_verdict(db, endpoint): + """Zero policies, zero monitors, zero impact rows - still resolves. + + The whole no-policy path in one assertion: absence of configuration is the + default, not a failure, and it must not read as unknown either. + """ + assert EndpointMonitorImpact.query.count() == 0 + assert EndpointMonitor.query.count() == 0 + + result = evaluate_endpoint_health(endpoint.assetid, endpoint, now=NOW) + db.session.commit() + + assert result.level == 'ok' + assert result.impactclass is None + + +def test_stopped_service_falls_back_to_degraded_without_an_impact_row( + db, endpoint): + """No EndpointMonitorImpact row means the built-in default applies. + + This is the verdict the old rollup gave a stopped watched service, kept + deliberately so shipping the mapping table changes nothing until someone + actually maps something. + """ + endpoint.servicesjson = _services( + {'name': 'eDNC', 'state': 'Stopped', 'monitorid': 4242}) + db.session.commit() + + result = evaluate_endpoint_health(endpoint.assetid, endpoint, now=NOW) + db.session.commit() + + assert result.impactclass == IMPACT_DEGRADED + assert result.level == 'warning' + assert result.label == 'eDNC stopped' + + +def test_impact_is_a_property_of_the_monitor_in_context(db, endpoint): + """Policy may RAISE impact, and only for the PC type it was mapped to. + + Same service, same telemetry, two PC types: a blocker on the shopfloor + machine that runs parts, merely degraded on one that does not. + """ + monitor = _monitor(db, 'eDNC') + db.session.add(EndpointMonitorImpact( + endpointmonitorid=monitor.endpointmonitorid, + computertypeid=7, + impactclass=IMPACT_BLOCKER, + createdat=NOW)) + endpoint.servicesjson = _services( + {'name': 'eDNC', 'state': 'Stopped', + 'monitorid': monitor.endpointmonitorid}) + db.session.commit() + + raised = evaluate_endpoint_health(endpoint.assetid, endpoint, + computertypeid=7, now=NOW) + db.session.commit() + assert raised.impactclass == IMPACT_BLOCKER + assert raised.level == 'critical' + + # PC type 9 has no mapping for this monitor, so the default applies. + default = evaluate_endpoint_health(endpoint.assetid, endpoint, + computertypeid=9, now=NOW) + db.session.commit() + assert default.impactclass == IMPACT_DEGRADED + + +def test_a_dead_drive_outranks_a_large_crash_count(db, endpoint): + """THE REGRESSION. 200 crashes must not hide a disconnected drive.""" + endpoint.usercontextjson = _drives(('S:', 'Disconnected')) + db.session.commit() + _events(db, endpoint.assetid, 200) + + result = evaluate_endpoint_health(endpoint.assetid, endpoint, now=NOW) + db.session.commit() + + assert result.sourcekind == SOURCE_DRIVE + assert result.impactclass == IMPACT_BLOCKER + assert 'S:' in result.label + + +def test_a_stopped_service_outranks_an_event_at_equal_impact(db, endpoint): + """A live condition beats a report of something that already finished.""" + endpoint.servicesjson = _services({'name': 'eDNC', 'state': 'Stopped'}) + db.session.commit() + _events(db, endpoint.assetid, 50) + + result = evaluate_endpoint_health(endpoint.assetid, endpoint, now=NOW) + db.session.commit() + + assert result.sourcekind == SOURCE_SERVICE + + +def test_events_still_surface_when_nothing_is_structurally_wrong(db, endpoint): + """Demoting events below conditions must not mean discarding them.""" + _events(db, endpoint.assetid, 3) + + result = evaluate_endpoint_health(endpoint.assetid, endpoint, now=NOW) + db.session.commit() + + assert result.sourcekind == SOURCE_EVENT + assert result.label.endswith('x3') + + +def test_events_outside_the_window_are_not_counted(db, endpoint): + """The window is a window, not a running total since install.""" + _events(db, endpoint.assetid, 5, hoursago=HEALTH_WINDOW_HOURS + 1) + + result = evaluate_endpoint_health(endpoint.assetid, endpoint, now=NOW) + db.session.commit() + + assert result.level == 'ok' + + +def test_a_stalled_service_is_distinguished_from_a_stopped_one(db, endpoint): + """stop-pending is not the same fault as cleanly down, and now says so.""" + endpoint.servicesjson = _services( + {'name': 'eDNC', 'state': 'Stop Pending'}) + db.session.commit() + + result = evaluate_endpoint_health(endpoint.assetid, endpoint, now=NOW) + db.session.commit() + + assert result.label == 'eDNC stalled' + + +def test_malformed_services_reads_as_unknown_not_as_healthy(db, endpoint): + """Unparseable telemetry means we cannot tell - never claim ok.""" + endpoint.servicesjson = '{not json' + db.session.commit() + + result = evaluate_endpoint_health(endpoint.assetid, endpoint, now=NOW) + db.session.commit() + + assert result.level == 'unknown' + + +def test_a_ticking_event_count_is_the_same_condition(db, endpoint): + """x2 -> x3 must not churn history, reset firstseenat, or flap. + + Identity is what and where, not the rendered label. If the label drove + this, a crash loop would open a new transition every single check-in. + """ + _events(db, endpoint.assetid, 2) + evaluate_endpoint_health(endpoint.assetid, endpoint, now=NOW) + db.session.commit() + + db.session.add(EndpointEvent( + assetid=endpoint.assetid, eventid=CRASH_EVENTID, eventrecordid=99, + eventchannel='Application', eventtime=NOW - timedelta(hours=1))) + db.session.commit() + + # Still inside the 48h window: the condition genuinely persists, so this + # isolates "same condition" from "events aged out". + later = NOW + timedelta(hours=1) + result = evaluate_endpoint_health(endpoint.assetid, endpoint, now=later) + db.session.commit() + + assert result.label.endswith('x3') + assert result.firstseenat == NOW + assert EndpointHealthTransition.query.filter_by( + assetid=endpoint.assetid).count() == 1 + + +def test_recovery_resolves_the_open_transition(db, endpoint): + """A condition is over only when something observes it to be over.""" + endpoint.servicesjson = _services({'name': 'eDNC', 'state': 'Stopped'}) + db.session.commit() + evaluate_endpoint_health(endpoint.assetid, endpoint, now=NOW) + db.session.commit() + + endpoint.servicesjson = _services({'name': 'eDNC', 'state': 'Running'}) + later = NOW + timedelta(hours=1) + db.session.commit() + result = evaluate_endpoint_health(endpoint.assetid, endpoint, now=later) + db.session.commit() + + assert result.level == 'ok' + opening = (EndpointHealthTransition.query + .filter_by(assetid=endpoint.assetid, sourcekind=SOURCE_SERVICE) + .first()) + assert opening.resolvedat == later + + +def test_transitions_are_capped_but_unresolved_rows_are_never_trimmed( + db, endpoint): + """History is capped; the row explaining what is on screen still stays. + + Flap the service past the high-water mark, then leave it down. Retention + must bound the history AND keep the open transition, because that row is + what the UI reads to say how long the current fault has been running. + """ + for index in range(TRANSITION_HIGHWATER + 6): + running = index % 2 == 0 + endpoint.servicesjson = _services( + {'name': 'eDNC', 'state': 'Running' if running else 'Stopped'}) + db.session.commit() + evaluate_endpoint_health( + endpoint.assetid, endpoint, now=NOW + timedelta(minutes=index)) + db.session.commit() + + rows = EndpointHealthTransition.query.filter_by( + assetid=endpoint.assetid).all() + assert TRANSITION_RETENTION <= len(rows) <= TRANSITION_HIGHWATER + + unresolved = [row for row in rows if row.resolvedat is None] + assert len(unresolved) == 1 + + # Flapping is counted, and there is exactly one live verdict throughout. + state = EndpointHealthState.query.filter_by( + assetid=endpoint.assetid).one() + assert state.flapcount > 1 + + +def test_lastconfirmedat_does_not_advance_without_an_observation(db, endpoint): + """A pinned blocker on a silent endpoint must not look freshly verified.""" + endpoint.servicesjson = _services({'name': 'eDNC', 'state': 'Stopped'}) + db.session.commit() + evaluate_endpoint_health(endpoint.assetid, endpoint, now=NOW) + db.session.commit() + + # Re-evaluated two days later with no new check-in: lastreportedat is + # untouched, so confirmation must not move either. + result = evaluate_endpoint_health(endpoint.assetid, endpoint, + now=NOW + timedelta(days=2)) + db.session.commit() + + assert result.lastconfirmedat == NOW + assert result.lastevaluatedat == NOW + timedelta(days=2)