Skip to content

Significantly improve logbook performance when selecting entities #71657

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
balloob merged 17 commits into from
May 11, 2022
Merged

Significantly improve logbook performance when selecting entities #71657

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
17 commits
Select commit Hold shift + click to select a range
486ce45
tweak
bdraco May 10, 2022
642cb5b
working
bdraco May 10, 2022
b1ccb92
adjust
bdraco May 10, 2022
bcdd3dc
adjust
bdraco May 10, 2022
1e976c7
adjust
bdraco May 10, 2022
383f1ce
adjust
bdraco May 10, 2022
b843f6e
tweak
bdraco May 10, 2022
c553583
fix cache key
bdraco May 10, 2022
a1045a7
tweaks
bdraco May 10, 2022
9a16091
more tweaks
bdraco May 10, 2022
17b5706
more tweaks
bdraco May 10, 2022
cdb86e7
more tweaks
bdraco May 10, 2022
05d6e6e
more tweaks
bdraco May 10, 2022
1333c84
tweak
bdraco May 10, 2022
3d6ecc4
tweak
bdraco May 10, 2022
4f00ff4
add coverage to make sure we did not break caching
bdraco May 10, 2022
5582956
entity_matches_only
bdraco May 10, 2022
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
277 changes: 25 additions & 252 deletions homeassistant/components/logbook/__init__.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,23 +1,18 @@
"""Event parser and human readable log generator."""
from __future__ import annotations

from collections.abc import Callable, Generator, Iterable
from collections.abc import Callable, Generator
from contextlib import suppress
from datetime import datetime as dt, timedelta
from http import HTTPStatus
import json
import logging
import re
from typing import Any, cast

from aiohttp import web
import sqlalchemy
from sqlalchemy import lambda_stmt, select
from sqlalchemy.engine.row import Row
from sqlalchemy.orm import aliased
from sqlalchemy.orm.query import Query
from sqlalchemy.sql.expression import literal
from sqlalchemy.sql.lambdas import StatementLambdaElement
from sqlalchemy.sql.selectable import Select
import voluptuous as vol

from homeassistant.components import frontend
Expand All @@ -27,15 +22,8 @@
sqlalchemy_filter_from_include_exclude_conf,
)
from homeassistant.components.http import HomeAssistantView
from homeassistant.components.proximity import DOMAIN as PROXIMITY_DOMAIN
from homeassistant.components.recorder import get_instance
from homeassistant.components.recorder.models import (
EventData,
Events,
StateAttributes,
States,
process_timestamp_to_utc_isoformat,
)
from homeassistant.components.recorder.models import process_timestamp_to_utc_isoformat
from homeassistant.components.recorder.util import session_scope
from homeassistant.components.script import EVENT_SCRIPT_STARTED
from homeassistant.components.sensor import ATTR_STATE_CLASS, DOMAIN as SENSOR_DOMAIN
Expand Down Expand Up @@ -75,66 +63,32 @@
from homeassistant.loader import bind_hass
import homeassistant.util.dt as dt_util

ENTITY_ID_JSON_TEMPLATE = '%"entity_id":"{}"%'
from .queries import statement_for_request

_LOGGER = logging.getLogger(__name__)


FRIENDLY_NAME_JSON_EXTRACT = re.compile('"friendly_name": ?"([^"]+)"')
ENTITY_ID_JSON_EXTRACT = re.compile('"entity_id": ?"([^"]+)"')
DOMAIN_JSON_EXTRACT = re.compile('"domain": ?"([^"]+)"')
ICON_JSON_EXTRACT = re.compile('"icon": ?"([^"]+)"')
ATTR_MESSAGE = "message"

CONTINUOUS_DOMAINS = {PROXIMITY_DOMAIN, SENSOR_DOMAIN}
CONTINUOUS_ENTITY_ID_LIKE = [f"{domain}.%" for domain in CONTINUOUS_DOMAINS]

DOMAIN = "logbook"

EMPTY_JSON_OBJECT = "{}"
UNIT_OF_MEASUREMENT_JSON = '"unit_of_measurement":'
UNIT_OF_MEASUREMENT_JSON_LIKE = f"%{UNIT_OF_MEASUREMENT_JSON}%"
HA_DOMAIN_ENTITY_ID = f"{HA_DOMAIN}._"

CONFIG_SCHEMA = vol.Schema(
{DOMAIN: INCLUDE_EXCLUDE_BASE_FILTER_SCHEMA}, extra=vol.ALLOW_EXTRA
)

HOMEASSISTANT_EVENTS = [
EVENT_HOMEASSISTANT_START,
EVENT_HOMEASSISTANT_STOP,
]
HOMEASSISTANT_EVENTS = {EVENT_HOMEASSISTANT_START, EVENT_HOMEASSISTANT_STOP}

ALL_EVENT_TYPES_EXCEPT_STATE_CHANGED = [
ALL_EVENT_TYPES_EXCEPT_STATE_CHANGED = (
EVENT_LOGBOOK_ENTRY,
EVENT_CALL_SERVICE,
*HOMEASSISTANT_EVENTS,
]

ALL_EVENT_TYPES = [
EVENT_STATE_CHANGED,
*ALL_EVENT_TYPES_EXCEPT_STATE_CHANGED,
]


EVENT_COLUMNS = [
Events.event_type.label("event_type"),
Events.event_data.label("event_data"),
Events.time_fired.label("time_fired"),
Events.context_id.label("context_id"),
Events.context_user_id.label("context_user_id"),
Events.context_parent_id.label("context_parent_id"),
]

STATE_COLUMNS = [
States.state.label("state"),
States.entity_id.label("entity_id"),
States.attributes.label("attributes"),
StateAttributes.shared_attrs.label("shared_attrs"),
]

EMPTY_STATE_COLUMNS = [
literal(value=None, type_=sqlalchemy.String).label("state"),
literal(value=None, type_=sqlalchemy.String).label("entity_id"),
literal(value=None, type_=sqlalchemy.Text).label("attributes"),
literal(value=None, type_=sqlalchemy.Text).label("shared_attrs"),
]
)

SCRIPT_AUTOMATION_EVENTS = {EVENT_AUTOMATION_TRIGGERED, EVENT_SCRIPT_STARTED}

Expand Down Expand Up @@ -295,7 +249,6 @@ async def get(

hass = request.app["hass"]

entity_matches_only = "entity_matches_only" in request.query
context_id = request.query.get("context_id")

if entity_ids and context_id:
Expand All @@ -313,7 +266,6 @@ def json_events() -> web.Response:
entity_ids,
self.filters,
self.entities_filter,
entity_matches_only,
context_id,
)
)
Expand Down Expand Up @@ -416,7 +368,6 @@ def _get_events(
entity_ids: list[str] | None = None,
filters: Filters | None = None,
entities_filter: EntityFilter | Callable[[str], bool] | None = None,
entity_matches_only: bool = False,
context_id: str | None = None,
) -> list[dict[str, Any]]:
"""Get events for a period of time."""
Expand All @@ -428,10 +379,13 @@ def _get_events(
event_data_cache: dict[str, dict[str, Any]] = {}
context_lookup: dict[str | None, Row | None] = {None: None}
event_cache = EventCache(event_data_cache)
external_events = hass.data.get(DOMAIN, {})
external_events: dict[
str, tuple[str, Callable[[LazyEventPartialState], dict[str, Any]]]
] = hass.data.get(DOMAIN, {})
context_augmenter = ContextAugmenter(
context_lookup, entity_name_cache, external_events, event_cache
)
event_types = (*ALL_EVENT_TYPES_EXCEPT_STATE_CHANGED, *external_events)

def yield_rows(query: Query) -> Generator[Row, None, None]:
"""Yield Events that are not filtered away."""
Expand All @@ -441,6 +395,8 @@ def yield_rows(query: Query) -> Generator[Row, None, None]:
rows = query.yield_per(1000)
for row in rows:
context_lookup.setdefault(row.context_id, row)
if row.context_only:
continue
event_type = row.event_type
if event_type != EVENT_CALL_SERVICE and (
event_type == EVENT_STATE_CHANGED
Expand All @@ -451,22 +407,15 @@ def yield_rows(query: Query) -> Generator[Row, None, None]:
if entity_ids is not None:
entities_filter = generate_filter([], entity_ids, [], [])

event_types = [
*ALL_EVENT_TYPES_EXCEPT_STATE_CHANGED,
*hass.data.get(DOMAIN, {}),
]
entity_filter = None
if entity_ids is None and filters:
entity_filter = filters.entity_filter() # type: ignore[no-untyped-call]
stmt = _generate_logbook_query(
start_day,
end_day,
event_types,
entity_ids,
entity_filter,
entity_matches_only,
context_id,
stmt = statement_for_request(
start_day, end_day, event_types, entity_ids, filters, context_id
)
if _LOGGER.isEnabledFor(logging.DEBUG):
_LOGGER.debug(
"Literal statement: %s",
stmt.compile(compile_kwargs={"literal_binds": True}),
)

with session_scope(hass=hass) as session:
return list(
_humanify(
Expand All @@ -479,182 +428,6 @@ def yield_rows(query: Query) -> Generator[Row, None, None]:
)


def _generate_logbook_query(
start_day: dt,
end_day: dt,
event_types: list[str],
entity_ids: list[str] | None = None,
entity_filter: Any | None = None,
entity_matches_only: bool = False,
context_id: str | None = None,
) -> StatementLambdaElement:
"""Generate a logbook query lambda_stmt."""
stmt = lambda_stmt(
lambda: _generate_events_query_without_states()
.where((Events.time_fired > start_day) & (Events.time_fired < end_day))
.where(Events.event_type.in_(event_types))
.outerjoin(EventData, (Events.data_id == EventData.data_id))
)
if entity_ids is not None:
if entity_matches_only:
# When entity_matches_only is provided, contexts and events that do not
# contain the entity_ids are not included in the logbook response.
stmt.add_criteria(
lambda s: s.where(_apply_event_entity_id_matchers(entity_ids)),
track_on=entity_ids,
)
stmt += lambda s: s.union_all(
_generate_states_query()
.filter((States.last_updated > start_day) & (States.last_updated < end_day))
.where(States.entity_id.in_(entity_ids))
)
else:
if context_id is not None:
# Once all the old `state_changed` events
# are gone from the database remove the
# union_all(_generate_legacy_events_context_id_query()....)
stmt += lambda s: s.where(Events.context_id == context_id).union_all(
_generate_legacy_events_context_id_query()
.where((Events.time_fired > start_day) & (Events.time_fired < end_day))
.where(Events.context_id == context_id),
_generate_states_query()
.where(
(States.last_updated > start_day) & (States.last_updated < end_day)
)
.outerjoin(Events, (States.event_id == Events.event_id))
.where(States.context_id == context_id),
)
elif entity_filter is not None:
stmt += lambda s: s.union_all(
_generate_states_query()
.where(
(States.last_updated > start_day) & (States.last_updated < end_day)
)
.where(entity_filter)
)
else:
stmt += lambda s: s.union_all(
_generate_states_query().where(
(States.last_updated > start_day) & (States.last_updated < end_day)
)
)

stmt += lambda s: s.order_by(Events.time_fired)
return stmt


def _generate_events_query_without_data() -> Select:
return select(
literal(value=EVENT_STATE_CHANGED, type_=sqlalchemy.String).label("event_type"),
literal(value=None, type_=sqlalchemy.Text).label("event_data"),
States.last_changed.label("time_fired"),
States.context_id.label("context_id"),
States.context_user_id.label("context_user_id"),
States.context_parent_id.label("context_parent_id"),
literal(value=None, type_=sqlalchemy.Text).label("shared_data"),
*STATE_COLUMNS,
)


def _generate_legacy_events_context_id_query() -> Select:
"""Generate a legacy events context id query that also joins states."""
# This can be removed once we no longer have event_ids in the states table
return (
select(
*EVENT_COLUMNS,
literal(value=None, type_=sqlalchemy.String).label("shared_data"),
States.state,
States.entity_id,
States.attributes,
StateAttributes.shared_attrs,
)
.outerjoin(States, (Events.event_id == States.event_id))
.where(States.last_updated == States.last_changed)
.where(_not_continuous_entity_matcher())
.outerjoin(
StateAttributes, (States.attributes_id == StateAttributes.attributes_id)
)
)


def _generate_events_query_without_states() -> Select:
return select(
*EVENT_COLUMNS, EventData.shared_data.label("shared_data"), *EMPTY_STATE_COLUMNS
)


def _generate_states_query() -> Select:
old_state = aliased(States, name="old_state")
return (
_generate_events_query_without_data()
.outerjoin(old_state, (States.old_state_id == old_state.state_id))
.where(_missing_state_matcher(old_state))
.where(_not_continuous_entity_matcher())
.where(States.last_updated == States.last_changed)
.outerjoin(
StateAttributes, (States.attributes_id == StateAttributes.attributes_id)
)
)


def _missing_state_matcher(old_state: States) -> sqlalchemy.and_:
# The below removes state change events that do not have
# and old_state or the old_state is missing (newly added entities)
# or the new_state is missing (removed entities)
return sqlalchemy.and_(
old_state.state_id.isnot(None),
(States.state != old_state.state),
States.state.isnot(None),
)


def _not_continuous_entity_matcher() -> sqlalchemy.or_:
"""Match non continuous entities."""
return sqlalchemy.or_(
_not_continuous_domain_matcher(),
sqlalchemy.and_(
_continuous_domain_matcher, _not_uom_attributes_matcher()
).self_group(),
)


def _not_continuous_domain_matcher() -> sqlalchemy.and_:
"""Match not continuous domains."""
return sqlalchemy.and_(
*[
~States.entity_id.like(entity_domain)
for entity_domain in CONTINUOUS_ENTITY_ID_LIKE
],
).self_group()


def _continuous_domain_matcher() -> sqlalchemy.or_:
"""Match continuous domains."""
return sqlalchemy.or_(
*[
States.entity_id.like(entity_domain)
for entity_domain in CONTINUOUS_ENTITY_ID_LIKE
],
).self_group()


def _not_uom_attributes_matcher() -> Any:
"""Prefilter ATTR_UNIT_OF_MEASUREMENT as its much faster in sql."""
return ~StateAttributes.shared_attrs.like(
UNIT_OF_MEASUREMENT_JSON_LIKE
) | ~States.attributes.like(UNIT_OF_MEASUREMENT_JSON_LIKE)


def _apply_event_entity_id_matchers(entity_ids: Iterable[str]) -> sqlalchemy.or_:
"""Create matchers for the entity_id in the event_data."""
ors = []
for entity_id in entity_ids:
like = ENTITY_ID_JSON_TEMPLATE.format(entity_id)
ors.append(Events.event_data.like(like))
ors.append(EventData.shared_data.like(like))
return sqlalchemy.or_(*ors)


def _keep_row(
hass: HomeAssistant,
event_type: str,
Expand Down
Loading