04616fce8b
Fleet ranked endpoints by raw critical-event counts, so a machine with a noisy but harmless event log outranked one with a mapped drive down or a blocking service stopped. Volume is not severity, and the aggregation ran across the whole fleet on every page load. Rank operational blockers above degraded conditions above event noise, and cache the verdict per asset so Fleet reads a row instead of recomputing 48h of history. Transitions are written only when a verdict changes, capped at 50 rows per asset. Absence of a policy binding must never suppress a verdict: drives and events are not policy-gated, and services fall back to a built-in default impact when no mapping applies. Event 1001 drops to informational because the WER bucket pairs with 1000 and double-counted every crash. Parallelize the Fleet first-paint loaders that were awaiting serially.
417 lines
16 KiB
Python
417 lines
16 KiB
Python
"""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
|