[Perl] DAQ system for the FAPG.
perf serve charts from reading rollups
- backfill and maintain minute and hour aggregates in SQLite - query chart series from rollups instead of raw readings - preserve sample counts and gap detection across rollup boundaries - label the rollup variability band as the observed minimum-maximum range
Changed files
roles/daq-recorder/bin/fapg-daq-mqtt-sqlite
@@ -141,6 +141,8 @@
141
141
}
142
142
);
143
143
144
Added:
init_reading_rollups($dbh);
145
Added:
144
146
$dbh->do(
145
147
q{
146
148
CREATE TABLE IF NOT EXISTS node_status (
@@ -180,6 +182,168 @@
180
182
q{
181
183
CREATE INDEX IF NOT EXISTS node_status_topic_idx
182
184
ON node_status(topic)
185
Added:
}
186
Added:
);
187
Added:
}
188
Added:
189
Added:
sub init_reading_rollups {
190
Added:
my ($dbh) = @_;
191
Added:
192
Added:
$dbh->do(
193
Added:
q{
194
Added:
CREATE TABLE IF NOT EXISTS reading_rollups (
195
Added:
probe TEXT NOT NULL,
196
Added:
bucket_seconds INTEGER NOT NULL,
197
Added:
bucket_epoch INTEGER NOT NULL,
198
Added:
sample_count INTEGER NOT NULL,
199
Added:
value_total REAL NOT NULL,
200
Added:
value_min REAL NOT NULL,
201
Added:
value_max REAL NOT NULL,
202
Added:
first_sample_epoch INTEGER NOT NULL,
203
Added:
last_sample_epoch INTEGER NOT NULL,
204
Added:
max_gap_seconds INTEGER NOT NULL DEFAULT 0,
205
Added:
PRIMARY KEY (probe, bucket_seconds, bucket_epoch)
206
Added:
) WITHOUT ROWID
207
Added:
}
208
Added:
);
209
Added:
210
Added:
for my $bucket_seconds ( 60, 3_600 ) {
211
Added:
$dbh->do(
212
Added:
qq{
213
Added:
WITH samples AS (
214
Added:
SELECT
215
Added:
id,
216
Added:
probe,
217
Added:
CAST(
218
Added:
CAST(strftime('%s', COALESCE(timestamp, received_at)) AS INTEGER)
219
Added:
/ $bucket_seconds AS INTEGER
220
Added:
) * $bucket_seconds AS bucket_epoch,
221
Added:
CAST(strftime('%s', COALESCE(timestamp, received_at)) AS INTEGER)
222
Added:
AS sample_epoch,
223
Added:
CAST(value AS REAL) AS value
224
Added:
FROM readings
225
Added:
WHERE valid = 1
226
Added:
AND probe IS NOT NULL
227
Added:
AND value IS NOT NULL
228
Added:
AND COALESCE(timestamp, received_at) IS NOT NULL
229
Added:
),
230
Added:
ordered AS (
231
Added:
SELECT
232
Added:
*,
233
Added:
sample_epoch - LAG(sample_epoch) OVER (
234
Added:
PARTITION BY probe, bucket_epoch
235
Added:
ORDER BY sample_epoch, id
236
Added:
) AS gap_seconds
237
Added:
FROM samples
238
Added:
)
239
Added:
INSERT INTO reading_rollups (
240
Added:
probe, bucket_seconds, bucket_epoch, sample_count,
241
Added:
value_total, value_min, value_max,
242
Added:
first_sample_epoch, last_sample_epoch, max_gap_seconds
243
Added:
)
244
Added:
SELECT
245
Added:
probe,
246
Added:
$bucket_seconds,
247
Added:
bucket_epoch,
248
Added:
COUNT(*),
249
Added:
SUM(value),
250
Added:
MIN(value),
251
Added:
MAX(value),
252
Added:
MIN(sample_epoch),
253
Added:
MAX(sample_epoch),
254
Added:
MAX(COALESCE(gap_seconds, 0))
255
Added:
FROM ordered
256
Added:
WHERE NOT EXISTS (
257
Added:
SELECT 1 FROM reading_rollups
258
Added:
WHERE bucket_seconds = $bucket_seconds
259
Added:
LIMIT 1
260
Added:
)
261
Added:
GROUP BY probe, bucket_epoch
262
Added:
}
263
Added:
);
264
Added:
}
265
Added:
266
Added:
$dbh->do(
267
Added:
q{
268
Added:
CREATE TRIGGER IF NOT EXISTS readings_rollup_insert
269
Added:
AFTER INSERT ON readings
270
Added:
WHEN NEW.valid = 1
271
Added:
AND NEW.probe IS NOT NULL
272
Added:
AND NEW.value IS NOT NULL
273
Added:
AND COALESCE(NEW.timestamp, NEW.received_at) IS NOT NULL
274
Added:
BEGIN
275
Added:
INSERT INTO reading_rollups (
276
Added:
probe, bucket_seconds, bucket_epoch, sample_count,
277
Added:
value_total, value_min, value_max,
278
Added:
first_sample_epoch, last_sample_epoch, max_gap_seconds
279
Added:
)
280
Added:
VALUES (
281
Added:
NEW.probe,
282
Added:
60,
283
Added:
CAST(
284
Added:
CAST(strftime('%s', COALESCE(NEW.timestamp, NEW.received_at)) AS INTEGER)
285
Added:
/ 60 AS INTEGER
286
Added:
) * 60,
287
Added:
1,
288
Added:
CAST(NEW.value AS REAL),
289
Added:
CAST(NEW.value AS REAL),
290
Added:
CAST(NEW.value AS REAL),
291
Added:
CAST(strftime('%s', COALESCE(NEW.timestamp, NEW.received_at)) AS INTEGER),
292
Added:
CAST(strftime('%s', COALESCE(NEW.timestamp, NEW.received_at)) AS INTEGER),
293
Added:
0
294
Added:
)
295
Added:
ON CONFLICT (probe, bucket_seconds, bucket_epoch) DO UPDATE SET
296
Added:
sample_count = sample_count + 1,
297
Added:
value_total = value_total + excluded.value_total,
298
Added:
value_min = MIN(value_min, excluded.value_min),
299
Added:
value_max = MAX(value_max, excluded.value_max),
300
Added:
max_gap_seconds = MAX(
301
Added:
max_gap_seconds,
302
Added:
CASE
303
Added:
WHEN excluded.first_sample_epoch > last_sample_epoch
304
Added:
THEN excluded.first_sample_epoch - last_sample_epoch
305
Added:
ELSE 0
306
Added:
END
307
Added:
),
308
Added:
first_sample_epoch = MIN(first_sample_epoch, excluded.first_sample_epoch),
309
Added:
last_sample_epoch = MAX(last_sample_epoch, excluded.last_sample_epoch);
310
Added:
311
Added:
INSERT INTO reading_rollups (
312
Added:
probe, bucket_seconds, bucket_epoch, sample_count,
313
Added:
value_total, value_min, value_max,
314
Added:
first_sample_epoch, last_sample_epoch, max_gap_seconds
315
Added:
)
316
Added:
VALUES (
317
Added:
NEW.probe,
318
Added:
3600,
319
Added:
CAST(
320
Added:
CAST(strftime('%s', COALESCE(NEW.timestamp, NEW.received_at)) AS INTEGER)
321
Added:
/ 3600 AS INTEGER
322
Added:
) * 3600,
323
Added:
1,
324
Added:
CAST(NEW.value AS REAL),
325
Added:
CAST(NEW.value AS REAL),
326
Added:
CAST(NEW.value AS REAL),
327
Added:
CAST(strftime('%s', COALESCE(NEW.timestamp, NEW.received_at)) AS INTEGER),
328
Added:
CAST(strftime('%s', COALESCE(NEW.timestamp, NEW.received_at)) AS INTEGER),
329
Added:
0
330
Added:
)
331
Added:
ON CONFLICT (probe, bucket_seconds, bucket_epoch) DO UPDATE SET
332
Added:
sample_count = sample_count + 1,
333
Added:
value_total = value_total + excluded.value_total,
334
Added:
value_min = MIN(value_min, excluded.value_min),
335
Added:
value_max = MAX(value_max, excluded.value_max),
336
Added:
max_gap_seconds = MAX(
337
Added:
max_gap_seconds,
338
Added:
CASE
339
Added:
WHEN excluded.first_sample_epoch > last_sample_epoch
340
Added:
THEN excluded.first_sample_epoch - last_sample_epoch
341
Added:
ELSE 0
342
Added:
END
343
Added:
),
344
Added:
first_sample_epoch = MIN(first_sample_epoch, excluded.first_sample_epoch),
345
Added:
last_sample_epoch = MAX(last_sample_epoch, excluded.last_sample_epoch);
346
Added:
END
183
347
}
184
348
);
185
349
}
roles/dashboard/lib/FAPG/DAQ/Dashboard/Controller/Reading.pm
@@ -92,95 +92,52 @@
92
92
json => { error => "Unknown timeframe: $timeframe", },
93
93
) unless defined $window;
94
94
95
Removed:
my $unit = ( grep { $_->{key} eq $probe } $self->probes->@* )[0]{unit};
96
Removed:
my $minutes = $self->sqlite->db->query(
95
Added:
my $unit = ( grep { $_->{key} eq $probe } $self->probes->@* )[0]{unit};
96
Added:
my $rollup_seconds
97
Added:
= $window->{bucket_stride} >= 3_600 ? 3_600 : $MINUTE_SECONDS;
98
Added:
my $rollups = $self->sqlite->db->query(
97
99
q{
98
Removed:
WITH samples AS (
100
Added:
WITH ordered AS (
99
101
SELECT
100
Removed:
id,
101
Removed:
CAST(strftime('%s', COALESCE(timestamp, received_at)) AS INTEGER)
102
Removed:
AS sample_epoch,
103
Removed:
CAST(value AS REAL) AS value
104
Removed:
FROM readings
105
Removed:
WHERE probe = ?
106
Removed:
AND value IS NOT NULL
107
Removed:
AND COALESCE(timestamp, received_at) >= ?
108
Removed:
AND COALESCE(timestamp, received_at) < ?
109
Removed:
),
110
Removed:
ordered AS (
111
Removed:
SELECT
112
Removed:
id,
113
Removed:
sample_epoch,
114
Removed:
value,
115
Removed:
CAST(sample_epoch / 60 AS INTEGER) * 60 AS minute_epoch,
116
Removed:
sample_epoch - LAG(sample_epoch) OVER (
117
Removed:
ORDER BY sample_epoch, id
102
Added:
bucket_epoch AS minute_epoch,
103
Added:
sample_count AS count,
104
Added:
value_total / sample_count AS median,
105
Added:
value_min AS lower,
106
Added:
value_max AS upper,
107
Added:
max_gap_seconds,
108
Added:
first_sample_epoch - LAG(last_sample_epoch) OVER (
109
Added:
ORDER BY bucket_epoch
118
110
) AS gap_seconds
119
Removed:
FROM samples
120
Removed:
),
121
Removed:
ranked AS (
122
Removed:
SELECT
123
Removed:
*,
124
Removed:
ROW_NUMBER() OVER (
125
Removed:
PARTITION BY minute_epoch
126
Removed:
ORDER BY value, id
127
Removed:
) AS value_rank,
128
Removed:
COUNT(*) OVER (
129
Removed:
PARTITION BY minute_epoch
130
Removed:
) AS value_count
131
Removed:
FROM ordered
111
Added:
FROM reading_rollups
112
Added:
WHERE probe = ?
113
Added:
AND bucket_seconds = ?
114
Added:
AND bucket_epoch >= ?
115
Added:
AND bucket_epoch < ?
132
116
)
133
117
SELECT
134
118
minute_epoch,
135
Removed:
COUNT(*) AS count,
136
Removed:
AVG(
137
Removed:
CASE
138
Removed:
WHEN value_rank IN (
139
Removed:
CAST((value_count + 1) / 2 AS INTEGER),
140
Removed:
CAST((value_count + 2) / 2 AS INTEGER)
141
Removed:
) THEN value
142
Removed:
END
143
Removed:
) AS median,
144
Removed:
MAX(
145
Removed:
CASE
146
Removed:
WHEN value_rank = CASE
147
Removed:
WHEN value_count >= 5 THEN MAX(
148
Removed:
2,
149
Removed:
CAST((value_count + 9) / 10 AS INTEGER)
150
Removed:
)
151
Removed:
ELSE 1
152
Removed:
END THEN value
153
Removed:
END
154
Removed:
) AS lower,
155
Removed:
MAX(
156
Removed:
CASE
157
Removed:
WHEN value_rank = CASE
158
Removed:
WHEN value_count >= 5 THEN MIN(
159
Removed:
value_count - 1,
160
Removed:
CAST((9 * value_count + 9) / 10 AS INTEGER)
161
Removed:
)
162
Removed:
ELSE value_count
163
Removed:
END THEN value
164
Removed:
END
165
Removed:
) AS upper,
166
Removed:
MAX(
167
Removed:
CASE
168
Removed:
WHEN gap_seconds > CAST(? AS INTEGER) THEN 1
169
Removed:
ELSE 0
170
Removed:
END
171
Removed:
) AS gap
172
Removed:
FROM ranked
173
Removed:
GROUP BY minute_epoch
119
Added:
count,
120
Added:
median,
121
Added:
lower,
122
Added:
upper,
123
Added:
CASE
124
Added:
WHEN max_gap_seconds > CAST(? AS INTEGER)
125
Added:
OR gap_seconds > CAST(? AS INTEGER) THEN 1
126
Added:
ELSE 0
127
Added:
END AS gap
128
Added:
FROM ordered
174
129
ORDER BY minute_epoch
175
130
},
176
131
$probe,
177
Removed:
utc_timestamp( $window->{start_epoch} - ( 2 * $MINUTE_SECONDS ) ),
178
Removed:
$window->{end_iso},
132
Added:
$rollup_seconds,
133
Added:
$window->{start_epoch} - ( 2 * $rollup_seconds ),
134
Added:
$window->{end_epoch},
179
135
$SERIES_GAP_SECONDS,
136
Added:
$SERIES_GAP_SECONDS,
180
137
)->hashes->to_array;
181
138
182
Removed:
my $points
183
Removed:
= aggregate_series( $minutes, $window, $probe, $unit, $smooth );
139
Added:
my $points = aggregate_series( $rollups, $window, $probe, $unit, $smooth,
140
Added:
$rollup_seconds );
184
141
185
142
$self->render(
186
143
json => {
@@ -189,8 +146,8 @@
189
146
range => 0 + $range,
190
147
smooth => 0 + $smooth,
191
148
band => {
192
Removed:
lower_percentile => 10,
193
Removed:
upper_percentile => 90,
149
Added:
lower_statistic => 'minimum',
150
Added:
upper_statistic => 'maximum',
194
151
},
195
152
gap_seconds => $SERIES_GAP_SECONDS,
196
153
readings => $points,
@@ -312,9 +269,11 @@
312
269
return \@points;
313
270
}
314
271
315
Removed:
sub aggregate_series ( $minutes, $window, $probe, $unit, $smooth ) {
272
Added:
sub aggregate_series ( $minutes, $window, $probe, $unit, $smooth,
273
Added:
$rollup_seconds )
274
Added:
{
316
275
my $points = series_points($window);
317
Removed:
my $trends = minute_trends( $minutes, $smooth );
276
Added:
my $trends = minute_trends( $minutes, $smooth, $rollup_seconds );
318
277
my @values;
319
278
320
279
for my $i ( 0 .. $minutes->$#* ) {
@@ -352,7 +311,7 @@
352
311
return $points;
353
312
}
354
313
355
Removed:
sub minute_trends ( $minutes, $smooth ) {
314
Added:
sub minute_trends ( $minutes, $smooth, $rollup_seconds ) {
356
315
my @trends = map { 0 + $_->{median} } @$minutes;
357
316
return \@trends if !$smooth;
358
317
@@ -366,7 +325,7 @@
366
325
last if $minutes->[$next]{gap};
367
326
last
368
327
if $minutes->[$next]{minute_epoch}
369
Removed:
- $minutes->[$segment_end]{minute_epoch} > $MINUTE_SECONDS;
328
Added:
- $minutes->[$segment_end]{minute_epoch} > $rollup_seconds;
370
329
$segment_end = $next;
371
330
}
372
331
roles/dashboard/public/js/dashboard/charts.js
@@ -142,7 +142,7 @@
142
142
pointHoverRadius: 0
143
143
},
144
144
{
145
Removed:
label: "Raw variability (10th–90th percentile)",
145
Added:
label: "Observed range (minimum–maximum)",
146
146
data: series.upper,
147
147
borderColor: TRANSPARENT_CHART_COLOR,
148
148
backgroundColor: RAW_BAND_COLOR,
roles/dashboard/t/04-readings-series.t
@@ -17,12 +17,12 @@
17
17
18
18
$t->get_ok('/api/readings/ec/series?timeframe=hour')
19
19
->status_is(200)
20
Removed:
->json_is( '/probe' => 'ec' )
21
Removed:
->json_is( '/timeframe' => 'hour' )
22
Removed:
->json_is( '/smooth' => 0 )
23
Removed:
->json_is( '/band/lower_percentile' => 10 )
24
Removed:
->json_is( '/band/upper_percentile' => 90 )
25
Removed:
->json_is( '/gap_seconds' => 30 )
20
Added:
->json_is( '/probe' => 'ec' )
21
Added:
->json_is( '/timeframe' => 'hour' )
22
Added:
->json_is( '/smooth' => 0 )
23
Added:
->json_is( '/band/lower_statistic' => 'minimum' )
24
Added:
->json_is( '/band/upper_statistic' => 'maximum' )
25
Added:
->json_is( '/gap_seconds' => 30 )
26
26
->json_hasnt('/readings/30');
27
27
assert_series_reading( $t, 1050, 1000, 1100, 2 );
28
28
@@ -51,20 +51,21 @@
51
51
$t->get_ok('/api/readings/orp/series?timeframe=hour')
52
52
->status_is(200)
53
53
->json_is( '/smooth' => 0 );
54
Removed:
ok(
55
Removed:
scalar( grep { defined $_->{value} && $_->{value} > 0 }
56
Removed:
$t->tx->res->json->{readings}->@* ),
57
Removed:
'unsmoothed series retains the observed value' );
54
Added:
ok( scalar(
55
Added:
grep { defined $_->{value} && $_->{value} > 0 }
56
Added:
$t->tx->res->json->{readings}->@*
57
Added:
),
58
Added:
'unsmoothed series retains the observed value'
59
Added:
);
58
60
59
61
$t->get_ok('/api/readings/orp/series?timeframe=hour&smooth=1')
60
62
->status_is(200)
61
63
->json_is( '/smooth' => 1 );
62
Removed:
ok(
63
Removed:
scalar( grep { defined $_->{value} } $t->tx->res->json->{readings}->@* ),
64
Removed:
'smoothed series contains readings' );
64
Added:
ok( scalar( grep { defined $_->{value} } $t->tx->res->json->{readings}->@* ),
65
Added:
'smoothed series contains readings'
66
Added:
);
65
67
66
Removed:
$t->get_ok('/api/readings/do/series?timeframe=hour&smooth=1')
67
Removed:
->status_is(200);
68
Added:
$t->get_ok('/api/readings/do/series?timeframe=hour&smooth=1')->status_is(200);
68
69
ok( scalar( grep { $_->{gap} } $t->tx->res->json->{readings}->@* ),
69
70
'series marks a data gap' );
70
71
roles/dashboard/t/lib/Dashboard/Test.pm
@@ -44,11 +44,14 @@
44
44
probe TEXT NOT NULL,
45
45
value REAL NOT NULL,
46
46
unit TEXT,
47
Removed:
raw_json TEXT
47
Added:
raw_json TEXT,
48
Added:
valid INTEGER NOT NULL DEFAULT 1
48
49
)
49
50
}
50
51
);
51
52
53
Added:
create_reading_rollups($t);
54
Added:
52
55
$t->app->sqlite->db->query(
53
56
q{
54
57
CREATE TABLE node_status (
@@ -69,6 +72,83 @@
69
72
)
70
73
}
71
74
);
75
Added:
76
Added:
return;
77
Added:
}
78
Added:
79
Added:
sub create_reading_rollups {
80
Added:
my ($t) = @_;
81
Added:
my $db = $t->app->sqlite->db;
82
Added:
83
Added:
$db->query(
84
Added:
q{
85
Added:
CREATE TABLE reading_rollups (
86
Added:
probe TEXT NOT NULL,
87
Added:
bucket_seconds INTEGER NOT NULL,
88
Added:
bucket_epoch INTEGER NOT NULL,
89
Added:
sample_count INTEGER NOT NULL,
90
Added:
value_total REAL NOT NULL,
91
Added:
value_min REAL NOT NULL,
92
Added:
value_max REAL NOT NULL,
93
Added:
first_sample_epoch INTEGER NOT NULL,
94
Added:
last_sample_epoch INTEGER NOT NULL,
95
Added:
max_gap_seconds INTEGER NOT NULL DEFAULT 0,
96
Added:
PRIMARY KEY (probe, bucket_seconds, bucket_epoch)
97
Added:
) WITHOUT ROWID
98
Added:
}
99
Added:
);
100
Added:
101
Added:
for my $bucket_seconds ( 60, 3_600 ) {
102
Added:
$db->query(
103
Added:
qq{
104
Added:
CREATE TRIGGER readings_rollup_${bucket_seconds}_insert
105
Added:
AFTER INSERT ON readings
106
Added:
WHEN NEW.valid = 1
107
Added:
AND NEW.value IS NOT NULL
108
Added:
BEGIN
109
Added:
INSERT INTO reading_rollups (
110
Added:
probe, bucket_seconds, bucket_epoch, sample_count,
111
Added:
value_total, value_min, value_max,
112
Added:
first_sample_epoch, last_sample_epoch, max_gap_seconds
113
Added:
)
114
Added:
VALUES (
115
Added:
NEW.probe,
116
Added:
$bucket_seconds,
117
Added:
CAST(
118
Added:
CAST(strftime('%s', COALESCE(NEW.timestamp, NEW.received_at)) AS INTEGER)
119
Added:
/ $bucket_seconds AS INTEGER
120
Added:
) * $bucket_seconds,
121
Added:
1,
122
Added:
CAST(NEW.value AS REAL),
123
Added:
CAST(NEW.value AS REAL),
124
Added:
CAST(NEW.value AS REAL),
125
Added:
CAST(strftime('%s', COALESCE(NEW.timestamp, NEW.received_at)) AS INTEGER),
126
Added:
CAST(strftime('%s', COALESCE(NEW.timestamp, NEW.received_at)) AS INTEGER),
127
Added:
0
128
Added:
)
129
Added:
ON CONFLICT (probe, bucket_seconds, bucket_epoch) DO UPDATE SET
130
Added:
sample_count = sample_count + 1,
131
Added:
value_total = value_total + excluded.value_total,
132
Added:
value_min = MIN(value_min, excluded.value_min),
133
Added:
value_max = MAX(value_max, excluded.value_max),
134
Added:
max_gap_seconds = MAX(
135
Added:
max_gap_seconds,
136
Added:
CASE
137
Added:
WHEN excluded.first_sample_epoch > last_sample_epoch
138
Added:
THEN excluded.first_sample_epoch - last_sample_epoch
139
Added:
ELSE 0
140
Added:
END
141
Added:
),
142
Added:
first_sample_epoch = MIN(
143
Added:
first_sample_epoch, excluded.first_sample_epoch
144
Added:
),
145
Added:
last_sample_epoch = MAX(
146
Added:
last_sample_epoch, excluded.last_sample_epoch
147
Added:
);
148
Added:
END
149
Added:
}
150
Added:
);
151
Added:
}
72
152
73
153
return;
74
154
}