Skip to content

backend-implementation-guide.md

Latest commit

 

History

History
255 lines (167 loc) · 13.5 KB

backend-implementation-guide.md

File metadata and controls

255 lines (167 loc) · 13.5 KB

Backend Implementation Guide

HostWatch supports three storage backends: InfluxDB 1.8, InfluxDB 2.7, and Elasticsearch. All three inherit from the same TSDBClient abstract base class, but the underlying mechanics differ significantly. This document walks through each method, how each backend implements it, and the problems that came up while unifying them.

The ABC and Registry

TSDBClient is an abstract base class defined in src/registry.py. It declares seven abstract methods that every backend must implement:

class TSDBClient(ABC):

    @abstractmethod
    def create_database(self) -> None: ...

    @abstractmethod
    def batch_write(self, points: list[dict]) -> None: ...

    @abstractmethod
    def query(self, query: Any) -> list[dict[str, Any]]: ...

    @abstractmethod
    def get_metrics(self, measurement: str, time_range: str, host: str | None = None) -> list[dict[str, Any]]: ...

    @abstractmethod
    def delete_metric_data(self, key: str | None = None, tags: dict | None = None) -> None: ...

    @abstractmethod
    def create_or_alter_retention_policy(self, name: str, duration: str) -> None: ...

    @abstractmethod
    def drop_database(self) -> None: ...

Every backend inherits from TSDBClient (e.g., class InfluxDB1Client(TSDBClient)). If a backend is missing any of the seven methods, Python raises a TypeError the moment you try to instantiate it, not later when the dashboard or collector happens to call the missing method. This makes it safe to add new backends: you either implement the full contract or the app won't start.

The registry maps backend names to factory functions. When the collector or dashboard calls get_client(), the registry reads TSDB_BACKEND from the environment and returns the matching client instance. The caller never sees a concrete class. It only sees TSDBClient.

Backend registration uses a decorator:

@register_backend("influxdb1")
def _create_influxdb1():
    from src.database.influxdb1 import InfluxDB1Client
    return InfluxDB1Client()

The lazy import inside the factory means the InfluxDB library is only loaded when that backend is actually selected. If you're running Elasticsearch, the influxdb package is never imported.

Method-by-Method Breakdown

create_database()

Sets up the storage layer. Called once at startup before any writes.

Backend What it does
InfluxDB 1.8 Calls CREATE DATABASE. Idempotent, silently succeeds if the database already exists.
InfluxDB 2.x Looks up the bucket by name via the Buckets API. Creates it if not found. Buckets are the 2.x equivalent of databases.
Elasticsearch Checks if the index exists via indices.exists(). If not, creates it with explicit mappings for @timestamp (date), measurement (keyword), tags (object), and fields (object).

batch_write(points)

Writes a batch of metric points. The collector calls this every COLLECTION_INTERVAL seconds with all collected metrics.

All three backends receive the same input format from the collector:

{
    "name": "cpu",
    "tags": {"host": "docker-host"},
    "values": {"load_percent_avg": 42.5, "cpu_count": 8}
}

Each backend transforms this into its native write format:

Backend Transformation Write mechanism
InfluxDB 1.8 Maps name -> measurement, values -> fields, adds ISO timestamp. write_points(). The client library converts dicts to line protocol internally.
InfluxDB 2.x Same dict transformation as 1.8. write_api.write() with bucket and org parameters. Uses synchronous write mode.
Elasticsearch Wraps each point as a document with @timestamp, measurement, tags, and fields as top-level keys. Prepends each with a {"index": {...}} action line. bulk() API. Sends all documents in a single HTTP request.

query(query)

Executes a raw query written in the backend's native language. This is the method exposed in the dashboard's Query Explorer, where the end user types a query and sees results.

The input problem

Each database speaks a completely different query language:

Backend Query language Example
InfluxDB 1.8 InfluxQL (SQL-like) SELECT * FROM "cpu" WHERE time > now() - 1h
InfluxDB 2.x Flux (functional/pipe-based) from(bucket: "hostwatch") |> range(start: -1h) |> filter(...)
Elasticsearch Query DSL (JSON) {"bool": {"must": [{"term": {"measurement": "cpu"}}]}}

The ABC declares query(self, query: Any). It intentionally accepts Any because there's no way to define a single query format that all three understand. The burden of writing the correct query falls on the end user in the Query Explorer. They know which backend they're running and write the query in the appropriate language.

But this creates a problem for the dashboard: the Query Explorer sends user input from a text area, which is always a string. InfluxDB backends already expect strings, so that works. Elasticsearch expects a dict. Rather than making the dashboard aware of which backend is active, the Elasticsearch client handles this internally. If query() receives a string, it parses it with json.loads() before executing. The caller always sends a string, and each backend deals with it.

The output problem

The three databases return data in completely different structures:

  • InfluxDB 1.8 returns a ResultSet object. Calling result.get_points() yields flat dicts with time, tag values, and field values as top-level keys. This is already close to list[dict].

  • InfluxDB 2.x returns Flux tables where each field is a separate record. A single CPU measurement with load_percent_avg and cpu_count comes back as two separate records, not one. The client groups records by (time, measurement) and merges them into a single dict per timestamp, reconstructing the point shape that was originally written.

  • Elasticsearch returns a search response with hits.hits, where each hit has a _source containing the raw document. The document has nested fields and tags objects. The client flattens these into a single dict, spreading fields and tags alongside time and measurement, so the output shape matches the other backends.

The result is that all three return list[dict] with the same general shape: a time key and metric values as flat keys. The dashboard renders whatever comes back without needing to know which backend produced it.

Elasticsearch result cap

The Elasticsearch client passes size=10000 to search(). If a query matches more than 10,000 documents, the excess is silently dropped. InfluxDB has no such cap.

validate_query(query)

Each backend runs validation before executing a raw query to prevent destructive operations. This is not part of the ABC. It's an internal method called by query().

Backend Forbidden patterns Validation style
InfluxDB 1.8 drop, delete, alter, create, into Substring match on the lowercased query string.
InfluxDB 2.x drop(, delete(, |> to(, experimental Substring match, but uses function call syntax (with parentheses) to reduce false positives.
Elasticsearch delete, drop, alter Converts the query dict to a string via str(), then does substring matching on the lowercased result.

get_metrics(measurement, time_range, host=None)

The query() method works, but it requires the caller to know the query language. The dashboard's built-in charts need to fetch metrics without caring about the backend. That's what get_metrics() solves.

Each backend constructs the query in its native language internally and delegates to self.query():

Backend Generated query
InfluxDB 1.8 SELECT * FROM "{measurement}" WHERE time > now() - {time_range} [AND "host" = '{host}'] ORDER BY time ASC
InfluxDB 2.x from(bucket: "...") |> range(start: -{time_range}) |> filter(fn: (r) => r._measurement == "{measurement}") [|> filter(fn: (r) => r.host == "{host}")]
Elasticsearch {"bool": {"must": [{"term": {"measurement": ...}}, {"range": {"@timestamp": {"gte": "now-{time_range}"}}}]}} with optional {"term": {"tags.host": host}}

Since each implementation delegates to self.query(), the result normalization described above kicks in automatically. The dashboard just calls client.get_metrics("cpu", "1h") and gets back a list[dict] regardless of the backend.

create_or_alter_retention_policy(name, duration)

Controls how long data is kept before automatic deletion.

Backend Mechanism Duration parsing
InfluxDB 1.8 Native retention policies. Checks if a policy with the given name exists, creates or alters it accordingly. Duration is passed directly to InfluxDB (e.g., 30d). None needed. InfluxDB parses duration strings natively.
InfluxDB 2.x Bucket retention rules via BucketRetentionRules. Looks up the bucket, sets every_seconds on its retention rules, then updates the bucket. Parses 30d into seconds using a unit mapping (d -> 86400, w -> 604800, etc.).
Elasticsearch Index Lifecycle Management (ILM). Creates or updates a lifecycle policy with a delete phase set to min_age. Parses 30d into days. Durations shorter than a day (seconds, minutes, hours) are rounded up to 1 day, since ILM operates at day granularity.

Caveat (Elasticsearch granularity): ILM policies work at day-level precision. A duration of 6h would be rounded up to 1d. InfluxDB retention policies support second-level precision.

Caveat (InfluxDB 2.x): The name parameter is accepted for interface compatibility but is effectively ignored. InfluxDB 2.x ties retention directly to the bucket, not to a named policy. There's only one retention rule per bucket.

delete_metric_data(key=None, tags=None)

Deletes stored metrics. Supports deleting by measurement name, by tags, or everything.

Backend Delete all Delete by key Delete by tags
InfluxDB 1.8 DROP SERIES FROM /.*/ delete_series(measurement=key) delete_series(tags=tags)
InfluxDB 2.x Delete API with empty predicate, full time range (epoch to now). Predicate: _measurement="key" Predicate: tag_key="tag_value" joined with AND
Elasticsearch delete_by_query with match_all delete_by_query with {"term": {"measurement": key}} delete_by_query with {"term": {"tags.tag_key": value}}

drop_database()

Destroys the entire storage container. Destructive and should not be called during normal collector/dashboard operation. It exists in the ABC for completeness and cleanup tooling.

Backend What it does
InfluxDB 1.8 DROP DATABASE. Removes the database and all data.
InfluxDB 2.x Deletes the bucket via Buckets API. Checks existence first.
Elasticsearch indices.delete(). Removes the index and all documents. Checks existence first.

Output Format Comparison

All query() and get_metrics() calls return list[dict]. Here's what a CPU metric looks like from each backend:

InfluxDB 1.8:

{"time": "2026-03-21T10:00:00Z", "host": "docker-host", "load_percent_avg": 42.5, "cpu_count": 8}

InfluxDB 2.x:

{"time": "2026-03-21T10:00:00+00:00", "measurement": "cpu", "load_percent_avg": 42.5, "cpu_count": 8}

Elasticsearch:

{"time": "2026-03-21T10:00:00+00:00", "measurement": "cpu", "host": "docker-host", "load_percent_avg": 42.5, "cpu_count": 8}

The differences are minor:

  • InfluxDB 1.8 includes tags (host) as flat keys but doesn't include measurement (it's implicit in the query).
  • InfluxDB 2.x includes measurement but tags may or may not appear depending on the Flux query.
  • Elasticsearch includes everything: measurement, tags, and fields are all flattened into the dict.

The dashboard handles these differences gracefully by checking for column existence before rendering charts (if "load_percent_avg" in df.columns).

Adding a New Backend

  1. Create a new directory under src/database/ (e.g., src/database/timescaledb/).
  2. Implement a client class that inherits from TSDBClient and provides all seven abstract methods. If any method is missing, Python will raise TypeError at instantiation.
from src.registry import TSDBClient

class TimescaleDBClient(TSDBClient):

    def create_database(self) -> None: ...
    def batch_write(self, points: list[dict]) -> None: ...
    def query(self, query): ...
    def get_metrics(self, measurement, time_range="1h", host=None): ...
    def delete_metric_data(self, key=None, tags=None): ...
    def create_or_alter_retention_policy(self, name, duration): ...
    def drop_database(self) -> None: ...
  1. Register it in src/registry.py:
@register_backend("timescaledb")
def _create_timescaledb():
    from src.database.timescaledb import TimescaleDBClient
    return TimescaleDBClient()
  1. Add the database service in docker-compose.yml with a profile, and pass TSDB_BACKEND=timescaledb to the collector and dashboard.

No changes needed in the collector, dashboard, or any other existing backend.

For how HostWatch's patterns map to OpenWISP Monitoring's backend system, see mapping-to-openwisp-monitoring.md.