113 lines
4.3 KiB
YAML
113 lines
4.3 KiB
YAML
apiVersion: v1
|
|
kind: ConfigMap
|
|
metadata:
|
|
name: {{ include "thermo.fullname" . }}-writer
|
|
labels:
|
|
{{- include "thermo.labels" . | nindent 4 }}
|
|
data:
|
|
redpanda.yaml: |
|
|
http:
|
|
enabled: true
|
|
address: 0.0.0.0:4195
|
|
|
|
input:
|
|
kafka:
|
|
addresses:
|
|
{{- toYaml .Values.writer.kafka.brokers | nindent 10 }}
|
|
topics:
|
|
- {{ .Values.writer.kafka.topic }}
|
|
consumer_group: {{ .Values.writer.kafka.consumerGroup }}
|
|
start_from_oldest: {{ .Values.writer.kafka.startFromOldest }}
|
|
|
|
pipeline:
|
|
processors:
|
|
- mapping: |
|
|
root = if this.event_id.type() != "string" ||
|
|
this.timestamp.type() != "string" ||
|
|
this.source.type() != "string" ||
|
|
this.event_type.type() != "string" ||
|
|
this.severity.type() != "string" ||
|
|
this.data.type() != "object" ||
|
|
this.data.measurements.type() != "object" {
|
|
deleted()
|
|
} else {
|
|
this
|
|
}
|
|
|
|
output:
|
|
sql_raw:
|
|
driver: postgres
|
|
dsn: ${THERMO_POSTGRES_DSN}
|
|
max_in_flight: 1
|
|
query: |
|
|
WITH payload AS (
|
|
SELECT $1::jsonb AS document
|
|
), event_data AS (
|
|
SELECT
|
|
(document->>'event_id')::uuid AS event_id,
|
|
COALESCE(document->>'timestamp', document#>>'{data,observed_at}')::timestamptz AS observed_at,
|
|
COALESCE(document->>'ingested_at', document->>'timestamp')::timestamptz AS ingested_at,
|
|
document->>'source' AS source,
|
|
document->>'event_type' AS event_type,
|
|
document->>'severity' AS severity,
|
|
document#>>'{data,collector_id}' AS collector_id,
|
|
COALESCE(
|
|
document#>>'{data,device_id}',
|
|
document#>>'{data,sensor_id}',
|
|
document#>>'{data,device_address}'
|
|
) AS device_id,
|
|
document AS raw_event,
|
|
COALESCE(document#>'{data,measurements}', '{}'::jsonb) AS measurements
|
|
FROM payload
|
|
), stored_event AS (
|
|
INSERT INTO sensor_events (
|
|
event_id, observed_at, ingested_at, source, event_type,
|
|
severity, collector_id, device_id, raw_event
|
|
)
|
|
SELECT
|
|
event_id, observed_at, ingested_at, source, event_type,
|
|
severity, collector_id, device_id, raw_event
|
|
FROM event_data
|
|
ON CONFLICT (event_id) DO NOTHING
|
|
RETURNING event_id
|
|
)
|
|
INSERT INTO sensor_measurements (
|
|
event_id, observed_at, source, collector_id, device_id,
|
|
metric_name, numeric_value, text_value, unit, source_updated_at
|
|
)
|
|
SELECT
|
|
event.event_id,
|
|
event.observed_at,
|
|
event.source,
|
|
event.collector_id,
|
|
event.device_id,
|
|
measurement.key,
|
|
CASE
|
|
WHEN jsonb_typeof(measurement.value) = 'number'
|
|
THEN measurement.value::text::double precision
|
|
WHEN jsonb_typeof(measurement.value) = 'boolean'
|
|
THEN CASE WHEN measurement.value::boolean THEN 1 ELSE 0 END
|
|
WHEN jsonb_typeof(measurement.value->'value') = 'number'
|
|
THEN (measurement.value->>'value')::double precision
|
|
WHEN jsonb_typeof(measurement.value->'value') = 'boolean'
|
|
THEN CASE WHEN (measurement.value->>'value')::boolean THEN 1 ELSE 0 END
|
|
ELSE NULL
|
|
END,
|
|
CASE
|
|
WHEN jsonb_typeof(measurement.value) = 'string'
|
|
THEN measurement.value#>>'{}'
|
|
WHEN jsonb_typeof(measurement.value->'value') = 'string'
|
|
THEN measurement.value->>'value'
|
|
ELSE NULL
|
|
END,
|
|
measurement.value->>'unit',
|
|
NULLIF(measurement.value->>'updated_at', '')::timestamptz
|
|
FROM event_data AS event
|
|
CROSS JOIN LATERAL jsonb_each(event.measurements) AS measurement
|
|
WHERE
|
|
jsonb_typeof(measurement.value) IN ('number', 'boolean', 'string')
|
|
OR jsonb_typeof(measurement.value->'value') IN ('number', 'boolean', 'string')
|
|
ON CONFLICT (event_id, metric_name) DO NOTHING;
|
|
args_mapping: |
|
|
root = [this.string()]
|