Log Service

The log service: the platform’s pod logs, from the log store.

The Helm chart keeps the last 72 hours of the logs of the app, the Celery workers, beat and MariaDB in Loki, which Grafana Alloy fills. Alloy labels each line with its component (the pod’s app.kubernetes.io/name, e.g. smarter-worker), pod and container, and redacts email addresses, IP addresses, tokens, api keys and passwords before it is stored. See helm/charts/smarter/values.yaml.

Loki does not depend on a cloud, so LogService is implemented by LokiLogService, at smarter_settings.loki_url, for every cloud provider. Tests and local development use InMemoryLogService, installed with configure_logs().

from datetime import timedelta

from smarter.apps.infrastructure.services import infrastructure
from smarter.apps.infrastructure.services.logs import LogQuery

lines = infrastructure.logs.query(
    LogQuery.last(timedelta(hours=1), components=("smarter-worker",), contains=("ERROR",))
)

A LogQuery is structured, not LogQL: the service builds the query itself, and escapes every value, so that a caller, e.g. an LLM’s tool call, cannot inject LogQL.

The Smarter Operations MCP server’s log tools (smarter.apps.mcpserver.logs.tools) and its help desk snapshot (smarter.apps.mcpserver.helpdesk.users) are the service’s callers. The service only reads: nothing is written to Loki, and nothing is cached.

class smarter.apps.infrastructure.services.logs.InMemoryLogService(ready=True, **kwargs)[source]

Bases: LogService

Pod logs in memory, for tests and local development: add lines to lines.

Queries are answered in Python: contains case insensitively, and pattern with re, which accepts the RE2 syntax that Loki uses for the patterns that matter here.

Unlike Loki, a min_level query keeps only the lines whose own level is at least the minimum, which is what callers filter Loki’s result down to.

Parameters:: ready (bool) – Whether the service is ready.

__init__(ready=True, **kwargs)[source]

Create the service, with no lines.

Parameters:

ready (bool) – Whether the service is ready. False makes every call raise InfrastructureNotReadyError, like a platform without a log store.
kwargs – Passed to InfrastructureService.

label_values(label, start, end)[source]

Return the values of a label among the lines of lines in a period.

Parameters:

label (Literal['component', 'pod']) – component or pod.
start (datetime) – The earliest time, in UTC.
end (datetime) – The latest time, in UTC.

Return type:

list[str]

Returns:

The distinct values, sorted.

Raises:

InfrastructureNotReadyError – If the service is not ready.

lines

tests add lines here.

Type:: The log, in any order

matches(query, line)[source]

Return whether a line matches every filter of a query.

Parameters:

query (LogQuery) – The query.
line (LogLine) – The line.

Return type:

bool

Returns:

True if the line is in the query’s period, components and pod, contains each text, has at least the minimum level, and matches the pattern.

queries: Every query, in order, for tests to inspect.

query(query)[source]

Return the lines of lines that match a query, and record the query in queries.

Parameters:: query (LogQuery) – The query.
Return type:: list[LogLine]
Returns:: At most query.limit lines, latest first if query.direction is backward.
Raises:: InfrastructureNotReadyError – If the service is not ready.

property ready: bool: Whether the service is ready, as the constructor set it.

smarter.apps.infrastructure.services.logs.LOG_LEVEL_ORDER: tuple[LogLevel, ...] = (LogLevel.DEBUG, LogLevel.INFO, LogLevel.WARNING, LogLevel.ERROR, LogLevel.CRITICAL)

The levels, least severe first.

(
    <LogLevel.DEBUG: 'debug'>,
    <LogLevel.INFO: 'info'>,
    <LogLevel.WARNING: 'warning'>,
    <LogLevel.ERROR: 'error'>,
    <LogLevel.CRITICAL: 'critical'>,
)

smarter.apps.infrastructure.services.logs.LOG_RETENTION = datetime.timedelta(days=3)

limits_config.retention_period in the Helm chart’s Loki.

Type:: How long the log store keeps a line

class smarter.apps.infrastructure.services.logs.LogLevel(*values)[source]

Bases: StrEnum

The level of a log line, as parse_level() reads it from the line’s text.

Loki stores lines as text, without a level, so the level is read from the words that the platform’s, Celery’s and MariaDB’s formats put near the start of each line. The values are lower case, as an LLM writes them in a tool call, e.g. warning.

CRITICAL = 'critical'

DEBUG = 'debug'

ERROR = 'error'

INFO = 'info'

WARNING = 'warning'

class smarter.apps.infrastructure.services.logs.LogLine(**data)[source]

Bases: BaseModel

One line of a pod’s log, with the labels that Alloy gave it.

It is frozen: a line is read, never changed.

component: str: The pod’s app.kubernetes.io/name, e.g. smarter, smarter-worker or mariadb.

container: str: The container’s name.

property level: LogLevel | None: The line’s level, read from its text by parse_level(), or None if it names none.

line: str

already redacted.

Type:: The line’s text, as stored

model_config = {'frozen': True}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

pod: str: The pod’s name.

timestamp: datetime: When the line was written, in UTC.

class smarter.apps.infrastructure.services.logs.LogQuery(**data)[source]

Bases: BaseModel

A query of the pod logs.

Every value is validated, and the service escapes it, so a query cannot carry LogQL of its own. It is frozen, and refuses fields that it does not define. Build one with last(), for the most recent period, or with start and end. logql() turns it into LogQL.

Parameters:

start – The earliest time, in UTC. It may not be earlier than LOG_RETENTION ago.
end – The latest time, in UTC.
components – The components to read, e.g. ("smarter", "smarter-worker"), or all of them.
pod – One pod’s name, or every pod of the components.
contains – Text that every line must contain, case insensitively.
pattern – A regular expression (RE2 syntax) that every line must match.
min_level – The least severe level to return, e.g. warning for warnings, errors and critical lines. Lines without a level, e.g. a traceback’s continuation, are then left out.
limit – The most lines to return.
direction – backward returns the latest lines first, forward the earliest.

components: tuple[str, ...]: The components to read, each a Kubernetes name, or empty for all of them.

contains: tuple[str, ...]: Texts that every line must contain, case insensitively, each one line of 1 to 200 characters.

direction: Literal['backward', 'forward']: backward for the latest lines first, forward for the earliest.

end: datetime: The latest time, in UTC, later than start.

classmethod last(period, **kwargs)[source]

Return a query of the most recent period, e.g. the last hour.

A period longer than LOG_RETENTION is shortened to it, since Loki keeps nothing older.

Parameters:

period (timedelta) – How far back to read, at most LOG_RETENTION.
kwargs – The query’s other fields.

Return type:

LogQuery

Returns:

The query, from period ago until now.

Raises:

pydantic.ValidationError – If a field is not valid.

limit: int: The most lines to return, from 1 to MAX_QUERY_LINES.

logql()[source]

Return the query as LogQL, with every value escaped.

The stream selector picks the components, as an alternation of their escaped names, or every component, and the pod. Each filter is then a line filter, |~: each contains text, escaped and case insensitive, the pattern, and the levels from min_level up, as LOG_LEVEL_RE2. Every value is escaped for RE2 with re2_escape(), where it is taken literally, and for a LogQL string with logql_string().

Return type:: str
Returns:: e.g. {component=~"smarter|smarter-worker"} |~ "(?i)timeout".

min_level: LogLevel | None: The least severe level to return, or None for every line.

model_config = {'extra': 'forbid', 'frozen': True}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

pattern: str | None: A regular expression (RE2 syntax) of at most 200 characters, that every line must match, or None.

pod: str | None: One pod’s name, or None for every pod of the components.

start: datetime: The earliest time, in UTC, within LOG_RETENTION of now.

classmethod utc(value)[source]

Return a time in UTC: a time without a time zone is taken to be UTC, and any other is converted.

Parameters:: value (datetime) – The start or end.
Return type:: datetime
Returns:: The time, with the UTC time zone.

classmethod valid_components(value)[source]

Check that each component is a Kubernetes name, which cannot carry LogQL.

Parameters:: value (tuple[str, ...]) – The components.
Return type:: tuple[str, ...]
Returns:: The components, unchanged.
Raises:: ValueError – If a component is not a Kubernetes name.

classmethod valid_contains(value)[source]

Check that each text is a single line, of 1 to 200 characters.

Parameters:: value (tuple[str, ...]) – The texts.
Return type:: tuple[str, ...]
Returns:: The texts, unchanged.
Raises:: ValueError – If a text is empty, too long, or has a line break.

classmethod valid_pattern(value)[source]

Check that the pattern is at most 200 characters, and compiles.

It is compiled with Python’s re, which accepts nearly all of RE2’s syntax. A pattern that RE2 rejects anyway is reported by Loki, as a LogServiceError.

Parameters:: value (str | None) – The pattern, or None.
Return type:: str | None
Returns:: The pattern, unchanged.
Raises:: ValueError – If the pattern is too long, or does not compile.

valid_period()[source]

Check that the period ends after it starts, and starts within the retention period.

Five minutes of grace allow for the time between last() and this check.

Return type:: LogQuery
Returns:: The query, unchanged.
Raises:: ValueError – If the period is empty, or starts too long ago.

classmethod valid_pod(value)[source]

Check that the pod is a Kubernetes name, which cannot carry LogQL.

Parameters:: value (str | None) – The pod’s name, or None.
Return type:: str | None
Returns:: The name, unchanged.
Raises:: ValueError – If the name is not a Kubernetes name.

class smarter.apps.infrastructure.services.logs.LogService(provider_name, *args, **kwargs)[source]

Bases: InfrastructureService

The platform’s pod logs, from the log store: the interface of the facade’s .logs.

LokiLogService implements it for every platform, and InMemoryLogService for tests and local development. Callers reach it as smarter.apps.infrastructure.services.infrastructure .logs, never directly.

error_class

The exception that the service raises when its backend fails.

alias of LogServiceError

abstractmethod label_values(label, start, end)[source]

Return the values of a label in a period, e.g. the pods that logged in the last hour.

Parameters:

label (Literal['component', 'pod']) – component or pod.
start (datetime) – The earliest time, in UTC.
end (datetime) – The latest time, in UTC.

Return type:

list[str]

Returns:

The values, sorted.

Raises:

LogServiceError – If the store could not be reached.
InfrastructureNotReadyError – If there is no log store.

abstractmethod query(query)[source]

Return the lines that match a query.

Parameters:

query (LogQuery) – The query.

Return type:

list[LogLine]

Returns:

At most query.limit lines, latest first if query.direction is backward.

Raises:

LogServiceError – If the store could not be reached, or rejected the query.
InfrastructureNotReadyError – If there is no log store.

service_name: str = 'logs': The service’s name in the infrastructure facade, and in its signals.

class smarter.apps.infrastructure.services.logs.LokiLogService(allow_in_tests=False, **kwargs)[source]

Bases: LogService

The pod logs, in Loki at smarter_settings.loki_url.

It is ready when the url is set. In the unit tests, it refuses to query, unless allow_in_tests is True, e.g. with requests mocked. Loki is in the platform’s namespace, so the url is a private, in-cluster address, and redirects are not followed.

Side effects: every query is an HTTP GET to Loki, and the first success or failure after a change of state sends the infrastructure service’s connection signal (see connection_state()).

Parameters:: allow_in_tests (bool) – Query in the unit tests.

__init__(allow_in_tests=False, **kwargs)[source]

Create the service.

Parameters:

allow_in_tests (bool) – Query in the unit tests, e.g. with requests mocked.
kwargs – Passed to InfrastructureService.

label_values(label, start, end)[source]

Return the values of a label in a period, from Loki’s label/<name>/values api.

Side effects: one HTTP GET to Loki.

Parameters:

label (Literal['component', 'pod']) – component or pod.
start (datetime) – The earliest time, in UTC.
end (datetime) – The latest time, in UTC.

Return type:

list[str]

Returns:

The values, sorted.

Raises:

LogServiceError – If Loki could not be reached, or answered with something other than a label values response.
InfrastructureNotReadyError – If there is no log store.

query(query)[source]

Return the lines that match a query, from Loki’s query_range api.

Loki returns one stream for each set of labels, each ordered on its own, so the lines of every stream are merged, and sorted by time, in the query’s direction.

Side effects: one HTTP GET to Loki, and a DEBUG log of the LogQL and the number of lines.

Parameters:

query (LogQuery) – The query.

Return type:

list[LogLine]

Returns:

At most query.limit lines, latest first if query.direction is backward.

Raises:

LogServiceError – If Loki could not be reached, rejected the query, or answered with something other than a query response.
InfrastructureNotReadyError – If there is no log store.

property ready: bool

smarter_settings.loki_url is set, and it is not a unit test, unless allow_in_tests.

Type:: Whether the service can query

smarter.apps.infrastructure.services.logs.MAX_QUERY_LINES = 5000

Loki’s default max_entries_limit_per_query.

Type:: The most lines that one query returns

smarter.apps.infrastructure.services.logs.at_least(level, minimum)[source]

Return whether a level is at least as severe as a minimum.

A line without a level, e.g. a traceback’s continuation, is not.

Parameters:

level (LogLevel | None) – The line’s level, or None.
minimum (LogLevel) – The least severe level, e.g. LogLevel.WARNING.

Return type:

bool

Returns:

True if the level is the minimum, or more severe, by LOG_LEVEL_ORDER.

smarter.apps.infrastructure.services.logs.configure_logs(factory)[source]

Set the factory of the log service that get_logs() returns.

The current service is dropped, so that the next get_logs() creates one with the new factory. Tests install an InMemoryLogService this way, and restore the default after.

Parameters:: factory (Callable[[], LogService] | None) – Returns the service, or None to restore the default, LokiLogService.
Return type:: None

smarter.apps.infrastructure.services.logs.get_logs()[source]

Return the log service, which is created once, by the factory of configure_logs().

Return type:: LogService
Returns:: The service: the factory’s, or a LokiLogService by default.

smarter.apps.infrastructure.services.logs.logql_string(value)[source]

Escape a value for a LogQL double-quoted string, whose escapes are Go’s.

Parameters:: value (str) – The value.
Return type:: str
Returns:: The value with backslashes and double quotes escaped.

smarter.apps.infrastructure.services.logs.parse_level(line)[source]

Return the level of a log line, from its text.

The platform’s lines put their level near the start, so only the first 120 characters are read, which keeps a message that mentions “ERROR” from making an INFO line an error.

Parameters:: line (str) – The line.
Return type:: LogLevel | None
Returns:: The level, or None if the line names none, e.g. a traceback’s continuation.

smarter.apps.infrastructure.services.logs.re2_escape(text)[source]

Escape text so that an RE2 regular expression matches it literally.

Unlike re.escape(), it escapes only RE2_SPECIAL: RE2 rejects an escaped space or letter, which re.escape() can produce.

Parameters:: text (str) – The text.
Return type:: str
Returns:: The text, with each special character escaped with a backslash.