[CrateDB] Add support for the data export subsystem

etc/test/cratedb.ini

@@ -3,6 +3,18 @@
 ; ######################################
 
 
+; ==========================================
+; General settings
+; ==========================================
+
+; http server
+[kotori]
+
+; TODO: Refactor to [http] section, or, even better, into the channel configuration.
+http_listen = localhost
+http_port = 24643
+
+
 ; =====================
 ; Connectivity settings
 ; =====================
@@ -39,3 +51,25 @@ type = application
 realm = mqttkit-2
 mqtt_topics = mqttkit-2/#
 application = kotori.daq.application.mqttkit:mqttkit_application
+
+[mqttkit-2.http-api-generic]
+enable = true
+
+type = application
+application = kotori.io.protocol.forwarder:boot
+
+realm = mqttkit-2
+source = http:/api/mqttkit-2/{address:.*}/{slot:(data|event)} [POST]
+target = mqtt:/mqttkit-2/{address}/{slot}.json
+
+[mqttkit-2.cratedb-data-export]
+enable = true
+
+type = application
+application = kotori.io.protocol.forwarder:boot
+
+realm = mqttkit-2
+source = http:/api/{realm:mqttkit-2}/{network:.*}/{gateway:.*}/{node:.*}/{slot:(data|event)}.{suffix} [GET]
+target = cratedb:/{database}?measurement={measurement}
+transform = kotori.daq.strategy.wan:WanBusStrategy.topology_to_storage,
+ kotori.io.protocol.cratedb:QueryTransformer.transform

etc/test/main.ini

@@ -15,7 +15,7 @@ include =
 ; http server
 [kotori]
 
-; TODO: Refactor to [http] section.
+; TODO: Refactor to [http] section, or, even better, into the channel configuration.
 http_listen = localhost
 http_port = 24642
 

kotori/daq/storage/cratedb.py

@@ -1,25 +1,29 @@
 # -*- coding: utf-8 -*-
 # (c) 2023 Andreas Motl <[email protected]>
 import calendar
+import functools
 import json
+from collections import OrderedDict
 from decimal import Decimal
 from copy import deepcopy
-from datetime import datetime, date
+from datetime import datetime, date, timezone
 
 import crate.client.http
 import pytz
 import requests
 from crate import client
+from crate.client.converter import DefaultTypeConverter
 from crate.client.exceptions import ProgrammingError
 from funcy import project
+from munch import Munch
 from twisted.logger import Logger
 
 from kotori.daq.storage.util import format_chunk
 
 log = Logger()
 
 
-class CrateDBAdapter(object):
+class CrateDBAdapter:
  """
  Kotori database backend adapter for CrateDB.
 
@@ -86,6 +90,79 @@ def create_table(self, tablename):
  cursor.execute(sql_ddl)
  cursor.close()
 
+ def query(self, expression: str, tdata: Munch = None):
+ """
+ Query CrateDB and respond with results in suitable shape.
+
+ Make sure to synchronize data by using `REFRESH TABLE ...` before running
+ the actual `SELECT` statement. This is applicable in test case scenarios.
+
+ Response format::
+
+ [
+ {
+ "time": ...,
+ "tags": {"city": "berlin", "location": "balcony"},
+ "fields": {"temperature": 42.42, "humidity": 84.84},
+ },
+ ...
+ ]
+
+ TODO: Unify with `kotori.test.util:CrateDBWrapper.query`.
+ """
+
+ log.info(f"Database query: {expression}")
+
+ tdata = tdata or {}
+
+ # Before reading data from CrateDB, synchronize it.
+ # Currently, it is mostly needed to satisfy synchronization constraints when running the test suite.
+ # However, users also may expect to see data "immediately". On the other hand, in order to satisfy
+ # different needs, this should be made configurable per realm, channel and/or request.
+ # TODO: Maybe just _optionally_ synchronize with the database when reading data.
+ if tdata:
+ refresh_sql = f"REFRESH TABLE {self.get_tablename(tdata)}"
+ self.execute(refresh_sql)
+
+ def dict_from_row(columns, row):
+ """
+ https://stackoverflow.com/questions/3300464/how-can-i-get-dict-from-sqlite-query
+ https://stackoverflow.com/questions/4147707/python-mysqldb-sqlite-result-as-dictionary
+ """
+ return dict(zip(columns, row))
+
+ def record_from_dict(item):
+ record = OrderedDict()
+ record.update({"time": item["time"]})
+ record.update(item["tags"])
+ record.update(item["fields"])
+ return record
+
+ # Query database, with convenience data type converters. Assume timestamps to be in UTC.
+ cursor = self.db_client.cursor(converter=DefaultTypeConverter(), time_zone=timezone.utc)
+ cursor.execute(expression)
+ data_raw = cursor.fetchall()
+
+ # Provide fully-qualified records to downstream components, including column names.
+ column_names = [column_info[0] for column_info in cursor.description]
+ data_tags_fields = map(functools.partial(dict_from_row, column_names), data_raw)
+
+ # Bring results into classic "records" shape.
+ data_records = map(record_from_dict, data_tags_fields)
+
+ cursor.close()
+ return data_records
+
+ def execute(self, expression: str):
+ """
+ Execute a database query, using a cursor, and return its results.
+ """
+ cursor = self.db_client.cursor()
+ cursor.execute(expression)
+ result = cursor._result
+ cursor.close()
+ return result
+
  def write(self, meta, data):
  """
  Format ingress data chunk and store it into database table.

kotori/io/export/database.py

@@ -31,6 +31,11 @@ def query(self):
  settings=self.settings.influxdb,
  database=bucket.tdata.database,
  )
+ elif "cratedb" in self.settings:
+ from kotori.daq.storage.cratedb import CrateDBAdapter
+ database = CrateDBAdapter(
+ settings=self.settings.cratedb,
+ )
  else:
  log.warn("No time-series database configured")
  return

kotori/io/protocol/cratedb.py

@@ -0,0 +1,40 @@
+# -*- coding: utf-8 -*-
+# (c) 2023 Andreas Motl <[email protected]>
+from twisted.logger import Logger
+
+from kotori.io.protocol.util import compute_daterange
+
+log = Logger()
+
+
+class QueryTransformer:
+
+ @classmethod
+ def transform(cls, data):
+ """
+ Compute CrateDB query expression from data in transformation dictionary.
+ Also compute date range from query parameters "from" and "to".
+ """
+
+ log.info(f"Querying database: {data}")
+
+ # The PyInfluxQL query generator is versatile enough to be used for all SQL databases.
+ from pyinfluxql import Query
+
+ # TODO: Use ".date_range" API method
+ time_begin, time_end = compute_daterange(data.get('from'), data.get('to'))
+
+ # TODO: Add querying by tags.
+ tags = {}
+ # tags = CrateDBAdapter.get_tags(data)
+
+ table = f"{data.database}.{data.measurement}"
+ expression = Query('*').from_(table).where(time__gte=time_begin, time__lte=time_end, **tags)
+
+ result = {
+ 'expression': str(expression),
+ 'time_begin': time_begin,
+ 'time_end': time_end,
+ }
+
+ return result

kotori/io/protocol/target.py

@@ -61,8 +61,8 @@ def setupService(self):
  except Exception as ex:
  log.failure("Connecting to MQTT broker failed: {ex}", ex=last_error_and_traceback())
 
- elif self.scheme == 'influxdb':
-  # InfluxDB has no subsystem service, it's just an adapter
+ # CrateDB and InfluxDB are not subsystem services, they are just adapters.
+ elif self.scheme in ['cratedb', 'influxdb']:
  pass
 
  else:
@@ -87,7 +87,7 @@ def emit(self, uri, bucket):
  # TODO: Use threads.deferToThread here?
  return self.downstream.publish(topic, payload)
 
- elif self.scheme == 'influxdb':
+ elif self.scheme in ['cratedb', 'influxdb']:
 
  # InfluxDB query wrapper using expression derived from transformation data
  dfq = DataFrameQuery(settings=self.settings, bucket=bucket)

test/settings/mqttkit.py

@@ -16,6 +16,7 @@ class TestSettings:
  cratedb_measurement_events = 'foo_bar_events'
  mqtt2_topic_json = 'mqttkit-2/itest/foo/bar/data.json'
  grafana2_dashboards = ['mqttkit-2-itest', 'mqttkit-2-itest3']
+ io_channel_path = '/mqttkit-2/itest/foo/bar/data'
 
  # InfluxDB settings.
  influx_database = 'mqttkit_1_itest'

test/test_export.py

@@ -23,6 +23,25 @@
 ts_to = '2020-03-10T23:59:59.000Z'
 
 
+@pytest_twisted.inlineCallbacks
+@pytest.mark.http
+@pytest.mark.export
+@pytest.mark.cratedb
+def test_export_cratedb_general(machinery_cratedb, reset_cratedb):
+ """
+ Submit single reading in JSON format to HTTP API and proof
+ it can be retrieved back from the HTTP API in different formats.
+
+ This uses CrateDB as timeseries database.
+ """
+
+ channel_path = settings.io_channel_path
+ http_submit = functools.partial(http_json_sensor, port=24643)
+ http_fetch = functools.partial(http_get_data, port=24643)
+
+ yield verify_export_general(channel_path, http_submit, http_fetch)
+
+
 @pytest_twisted.inlineCallbacks
 @pytest.mark.http
 @pytest.mark.export

test/util.py

@@ -259,6 +259,9 @@ def query(self):
  },
  ...
  ]
+
+ TODO: Refactor to / unify with `kotori.daq.storage.cratedb:CrateDBAdapter.query`.
+ TODO: Add dict-based record retrieval as a row factory to `crate` library.
  """
  logger.info('CrateDB: Querying database')
  db_table = self.get_tablename()
@@ -280,6 +283,8 @@ def query(self):
  def execute(self, expression):
  """
  Actually execute the database query, using a cursor.
+
+ TODO: Use `kotori.daq.storage.cratedb:CrateDBAdapter.execute`.
  """
  cursor = self.client.cursor()
  cursor.execute(expression)