"""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