Skip to content

Commit 2d7f1dc

Browse files
marcopegmarcopeg
committed
Update queue counters on dropping a document
closes #33
1 parent 03a0891 commit 2d7f1dc

File tree

3 files changed

+186
-34
lines changed

3 files changed

+186
-34
lines changed

‎ssr/feature/api-v1/routes/v1-document-drop.js

Lines changed: 35 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,32 @@
11
const schema = require('./v1-document-drop.schema');
22

3+
const getMetricsToDecrease = ({ status_prev, version_prev }) => {
4+
const queries = ['cnt', `v${version_prev}`];
5+
6+
switch (status_prev) {
7+
case 3:
8+
queries.push('cpl');
9+
break;
10+
case 2:
11+
queries.push('act');
12+
break;
13+
case 1:
14+
queries.push('pnd');
15+
break;
16+
case 0:
17+
queries.push('pln');
18+
break;
19+
case -1:
20+
queries.push('kll');
21+
break;
22+
}
23+
24+
return queries;
25+
};
26+
27+
const buildDecrementSql = (queueName) => (metric) =>
28+
`SELECT FROM "fetchq"."metric_log_decrement"('${queueName}', '${metric}', 1);`;
29+
330
const v1QueueDocumentDrop = {
431
method: 'POST',
532
url: '/api/v1/queues/:name/drop/:subject',
@@ -11,8 +38,10 @@ const v1QueueDocumentDrop = {
1138
try {
1239
const _sql = `
1340
DELETE FROM "fetchq_data"."${params.name}__docs"
14-
WHERE "subject" = '${params.subject}';
41+
WHERE "subject" = '${params.subject}'
42+
RETURNING status AS status_prev, version AS version_prev;
1543
`;
44+
1645
const res = await fetchq.pool.query(_sql);
1746

1847
// Handle subject or queue not existing
@@ -27,18 +56,15 @@ const v1QueueDocumentDrop = {
2756
});
2857
}
2958

30-
// Log decrement of the amount of items in the queue
31-
// this is useful to keep the pagination correct.
32-
const _sqlCnt = `SELECT FROM "fetchq"."metric_log_decrement"('${params.name}', 'cnt', 1)`;
33-
await fetchq.pool.query(_sqlCnt);
34-
35-
// TODO: it should look into the status of the deleted document
36-
// and update counters accordingly.
59+
// Update the metrics according to the previous status of the document:
60+
const toSql = buildDecrementSql(params.name);
61+
const _sqlMetrics = getMetricsToDecrease(res.rows[0]).map(toSql).join('');
62+
_sqlMetrics && (await fetchq.pool.query(_sqlMetrics));
3763

3864
reply.send({
3965
success: true,
4066
data: {
41-
_sql,
67+
_sql: [_sql, ..._sqlMetrics].join('\n'),
4268
},
4369
});
4470
} catch (err) {

‎ssr/feature/api-v1/routes/v1-document-drop.test.e2e.js

Lines changed: 119 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,3 +1,15 @@
1+
const getQueueMetrics = async (queue) => {
2+
const _sqlCnt = `SELECT * FROM fetchq.metric_get('${queue}')`;
3+
const res = await global.query(_sqlCnt);
4+
return res.rows.reduce(
5+
(acc, curr) => ({
6+
...acc,
7+
[curr.metric]: Number(curr.current_value),
8+
}),
9+
{},
10+
);
11+
};
12+
113
describe('v1QueueDocumentDrop', () => {
214
beforeEach(global.resetSchema);
315

@@ -55,4 +67,111 @@ describe('v1QueueDocumentDrop', () => {
5567
expect(calls[0][0].response.status).toBe(404);
5668
expect(calls[0][0].response.data.success).toBe(false);
5769
});
70+
71+
it('should update counters from ACTIVE status', async () => {
72+
await global.query(`SELECT FROM fetchq.queue_create('q1')`);
73+
await global.query(`SELECT FROM fetchq.queue_set_max_attempts('q1', 1)`);
74+
await global.query(`SELECT FROM fetchq.doc_push('q1', 's1')`);
75+
await global.query(`SELECT FROM fetchq.doc_pick('q1', 0, 1, '1y')`);
76+
await global.query(`SELECT FROM fetchq.mnt()`);
77+
78+
const m1 = await getQueueMetrics('q1');
79+
expect(m1).toEqual({ act: 1, cnt: 1, ent: 1, pkd: 1, pnd: 0, v0: 1 });
80+
81+
await global.post('/api/v1/queues/q1/drop/s1');
82+
await global.query(`SELECT FROM fetchq.mnt()`);
83+
84+
const m2 = await getQueueMetrics('q1');
85+
expect(m2).toEqual({ act: 0, cnt: 0, ent: 1, pkd: 1, pnd: 0, v0: 0 });
86+
});
87+
88+
it('should update counters from PLANNED status', async () => {
89+
await global.query(`SELECT FROM fetchq.queue_create('q1')`);
90+
await global.query(`SELECT FROM fetchq.queue_set_max_attempts('q1', 1)`);
91+
await global.query(
92+
`SELECT FROM fetchq.doc_push('q1', 's1', 0, 0, NOW() + INTERVAL '1m', '{}')`,
93+
);
94+
await global.query(`SELECT FROM fetchq.mnt()`);
95+
96+
const m1 = await getQueueMetrics('q1');
97+
expect(m1).toEqual({ cnt: 1, ent: 1, pln: 1, v0: 1 });
98+
99+
await global.post('/api/v1/queues/q1/drop/s1');
100+
await global.query(`SELECT FROM fetchq.mnt()`);
101+
102+
const m2 = await getQueueMetrics('q1');
103+
expect(m2).toEqual({ cnt: 0, ent: 1, pln: 0, v0: 0 });
104+
});
105+
106+
it('should update counters from COMPLETED status', async () => {
107+
await global.query(`SELECT FROM fetchq.queue_create('q1')`);
108+
await global.query(`SELECT FROM fetchq.queue_set_max_attempts('q1', 1)`);
109+
await global.query(`SELECT FROM fetchq.doc_push('q1', 's1')`);
110+
await global.query(`SELECT FROM fetchq.doc_pick('q1', 0, 1, '1y')`);
111+
await global.query(`SELECT FROM fetchq.doc_complete('q1', 's1')`);
112+
await global.query(`SELECT FROM fetchq.mnt()`);
113+
114+
const m1 = await getQueueMetrics('q1');
115+
expect(m1).toEqual({
116+
act: 0,
117+
cnt: 1,
118+
cpl: 1,
119+
ent: 1,
120+
pkd: 1,
121+
pnd: 0,
122+
prc: 1,
123+
v0: 1,
124+
});
125+
126+
await global.post('/api/v1/queues/q1/drop/s1');
127+
await global.query(`SELECT FROM fetchq.mnt()`);
128+
129+
const m2 = await getQueueMetrics('q1');
130+
expect(m2).toEqual({
131+
act: 0,
132+
cnt: 0,
133+
cpl: 0,
134+
ent: 1,
135+
pkd: 1,
136+
pnd: 0,
137+
prc: 1,
138+
v0: 0,
139+
});
140+
});
141+
142+
it('should update counters from KILLED status', async () => {
143+
await global.query(`SELECT FROM fetchq.queue_create('q1')`);
144+
await global.query(`SELECT FROM fetchq.queue_set_max_attempts('q1', 1)`);
145+
await global.query(`SELECT FROM fetchq.doc_push('q1', 's1')`);
146+
await global.query(`SELECT FROM fetchq.doc_pick('q1', 0, 1, '1y')`);
147+
await global.query(`SELECT FROM fetchq.doc_kill('q1', 's1')`);
148+
await global.query(`SELECT FROM fetchq.mnt()`);
149+
150+
const m1 = await getQueueMetrics('q1');
151+
expect(m1).toEqual({
152+
act: 0,
153+
cnt: 1,
154+
kll: 1,
155+
ent: 1,
156+
pkd: 1,
157+
pnd: 0,
158+
prc: 1,
159+
v0: 1,
160+
});
161+
162+
await global.post('/api/v1/queues/q1/drop/s1');
163+
await global.query(`SELECT FROM fetchq.mnt()`);
164+
165+
const m2 = await getQueueMetrics('q1');
166+
expect(m2).toEqual({
167+
act: 0,
168+
cnt: 0,
169+
kll: 0,
170+
ent: 1,
171+
pkd: 1,
172+
pnd: 0,
173+
prc: 1,
174+
v0: 0,
175+
});
176+
});
58177
});

‎ssr/feature/api-v1/routes/v1-document-play.js

Lines changed: 32 additions & 25 deletions
Original file line numberDiff line numberDiff line change
@@ -1,32 +1,34 @@
11
const schema = require('./v1-document-play.schema');
22

3-
const getUpdateMetricsSql = (queueName, prevStatus) => {
4-
if (prevStatus === 2) {
5-
return `
6-
SELECT FROM "fetchq"."metric_log_increment"('${queueName}', 'pnd', 1);
7-
SELECT FROM "fetchq"."metric_log_decrement"('${queueName}', 'act', 1);
8-
`;
9-
}
10-
if (prevStatus === 0) {
11-
return `
12-
SELECT FROM "fetchq"."metric_log_increment"('${queueName}', 'pnd', 1);
13-
SELECT FROM "fetchq"."metric_log_decrement"('${queueName}', 'pln', 1);
14-
`;
15-
}
16-
if (prevStatus === 3) {
17-
return `
18-
SELECT FROM "fetchq"."metric_log_increment"('${queueName}', 'pnd', 1);
19-
SELECT FROM "fetchq"."metric_log_decrement"('${queueName}', 'cpl', 1);
20-
`;
21-
}
22-
if (prevStatus === -1) {
23-
return `
24-
SELECT FROM "fetchq"."metric_log_increment"('${queueName}', 'pnd', 1);
25-
SELECT FROM "fetchq"."metric_log_decrement"('${queueName}', 'kll', 1);
26-
`;
3+
const getUpdateMetricsSql = ({ status_prev }) => {
4+
const increment = [];
5+
const decrement = [];
6+
7+
switch (status_prev) {
8+
case 3:
9+
increment.push('pnd');
10+
decrement.push('cpl');
11+
break;
12+
case 2:
13+
increment.push('pnd');
14+
decrement.push('act');
15+
break;
16+
case 0:
17+
increment.push('pnd');
18+
decrement.push('pln');
19+
break;
20+
case -1:
21+
increment.push('pnd');
22+
decrement.push('kll');
23+
break;
2724
}
25+
26+
return [increment, decrement];
2827
};
2928

29+
const buildMetricSql = (queueName, action) => (metric) =>
30+
`SELECT FROM "fetchq"."metric_log_${action}"('${queueName}', '${metric}', 1);`;
31+
3032
/**
3133
* POST://api/v1/queues/:name/play/:subject
3234
*/
@@ -71,7 +73,12 @@ const v1QueueDocumentPlay = {
7173
const doc = res.rows[0];
7274

7375
// Update the metrics according to the previous status of the document:
74-
const _sqlMetrics = getUpdateMetricsSql(params.name, doc.status_prev);
76+
const [increment, decrement] = getUpdateMetricsSql(doc);
77+
const _sqlMetrics = [
78+
...increment.map(buildMetricSql(params.name, 'increment')),
79+
...decrement.map(buildMetricSql(params.name, 'decrement')),
80+
].join('');
81+
7582
_sqlMetrics && (await fetchq.pool.query(_sqlMetrics));
7683

7784
reply.send({

0 commit comments

Comments
 (0)