How to route data to influxDB from Sensors

[keenanjohnson]

@damz @akhilgupta1093 and @abhinavtripathy, there is one big architectural decision we have to make regarding the Frog Software for now. I think this will be a relatively short term solution as we will very soon after develop a proper API for data write, etc for frogs.

The question is how should data end up in our database: influxDB?

In previous sensor versions, we used the influxDB Python Client to send the data directly to the cloud hosted DB. This is very simple to implement and would probably work well with our micro-python implementation now. The main downside to this approach is that the devices have to have the write API key on the device, which is a bit of a security / abuse risk, though I think not much given the scale of the project.

The other easy short term solution would be to the route the data through Golioth's LightDB Stream service and then push it to Influx via Webhooks. The data chain would then be frog_sensor -> golioth LightDB -> InfluxDB. This would be nice because we could hide the InfluxDB key in the LightDB backend, removing it from the devices. However, it does introduce an additional system in the loop, which adds complexity.

Thoughts?

[damz]

It depends if we want to keep InfluxDB in the picture or not. I started working on a Golioth -> Pubsub -> BigQuery approach, based on the data model document. But that implies not going to InfluxDB at all, and not having a solution where we have v3 and v4 sensors data in the same place.

[keenanjohnson]

I updated the diagram here for how I solved this here:

I decided that the best way to solve this would be to use a serverless function like Google Cloud Functions to route the data from Golioth using Webhooks and then push it to InfluxDB.

The serverless function I used is shown below:

import functions_framework
from datetime import datetime, timedelta
from influxdb_client import InfluxDBClient, Point, WritePrecision  # type: ignore
from influxdb_client.client.write_api import SYNCHRONOUS  # type: ignore

BUCKET = "keenan-testing"
ORG = "keenan@ribbitnetwork.org"

@functions_framework.http
def hello_http(request):
    request_json = request.get_json(silent=True)
    request_args = request.args

    golioth_msg = request_json
    ribbit_msg = golioth_msg['data']['ribbitnetwork.datapoint']
    gps_fix = ribbit_msg['gps']['has_fix']

    # Check if the gps fix is valid, only send a datapoint if so
    if gps_fix:
        client = InfluxDBClient(url="https://us-central1-1.gcp.cloud2.influxdata.com/", token=SECRET_INFLUX_TOKEN, org="keenan@ribbitnetwork.org")
        write_api = client.write_api(write_options=SYNCHRONOUS)

        # Get metadata
        device_id = golioth_msg['device_id'] + '_golioth_esp32s3'
        timestamp = datetime.utcnow()
        
        # Parse Data
        scd30_co2_ppm = ribbit_msg['scd30']['co2']
        scd30_humidity = ribbit_msg['scd30']['humidity']
        scd30_temp = ribbit_msg['scd30']['temperature']

        gps_lat = ribbit_msg['gps']['latitude']
        gps_lon = ribbit_msg['gps']['longitude']
        gps_alt = ribbit_msg['gps']['altitude']
        
        dps310_pressure = ribbit_msg['dps310']['pressure']
        dps310_temp = ribbit_msg['dps310']['temperature']

        point = (
            Point("ghg_point")
            .tag("host", device_id)
            .field("co2", scd30_co2_ppm)
            .time(timestamp, WritePrecision.NS)
            .field("temperature", scd30_temp)
            .field("humidity", scd30_humidity)
            .field("lat", gps_lat)
            .field("lon", gps_lon)
            .field("alt", gps_alt)
            .field("baro_pressure", dps310_pressure)
            .field("baro_temperature", dps310_temp)
        )
        write_api.write(BUCKET, ORG, point)


    #Must send an HTTP Return Code
    return 'OK'