How to create a flow for counter type metric continuous aggregation

[dozer47528]

I have a metric like this:

CREATE TABLE requests (
                                  service_name STRING,
                                  service_ip STRING,
                                  value INT,
                                  ts TIMESTAMP TIME INDEX
);

Data Example:

svc1, 10.0.0.1, 100, 2024-10-18 19:00:00
svc1, 10.0.0.2, 100, 2024-10-18 19:00:00
svc1, 10.0.0.1, 200, 2024-10-18 19:00:30
svc1, 10.0.0.2, 200, 2024-10-18 19:00:30
svc1, 10.0.0.1, 300, 2024-10-18 19:01:00
svc1, 10.0.0.2, 100, 2024-10-18 19:01:00
svc1, 10.0.0.1, 400, 2024-10-18 19:01:30
svc1, 10.0.0.2, 200, 2024-10-18 19:01:30

And I will query this metric by this promQL sum(increase(requests{service_name='svc1'}[1m]))

I would like to create a flow which can remove service_ip column. For long-term storage, I don't need the service_ip column.

Which aggregation func should I use for creating a flow?

CREATE FLOW requests_long_term
SINK TO requests_without_ip
AS
SELECT
    service_name,
    sum(value) as sum_value,
FROM requests
GROUP BY
    service_name

If I use sum for aggregation func, the result would be:

svc1, 200, 2024-10-18 19:00:00
svc1, 400, 2024-10-18 19:00:30
svc1, 400, 2024-10-18 19:01:00
svc1, 600, 2024-10-18 19:01:30

For this data, How can I get the correct result? If I still use increase like this: sum(increase(requests_without_ip{service_name='svc1'}[1m])).

The result will be wrong.

[discord9]

To remove a column from a table, you can use Flow for filtering, as it does not perform any aggregation here. For an ad-hoc query that identifies all increases in the last minute and sums them, you would utilize PromQL. This requires a two-stage solution:

First, use Flow to filter out the service_ip column:

CREATE TABLE requests (
    service_name STRING,
    service_ip STRING,
    val INT,
    ts TIMESTAMP TIME INDEX
);

CREATE TABLE requests_without_ip (
    service_name STRING,
    val INT,
    ts TIMESTAMP TIME INDEX,
    update_at TIMESTAMP, -- auto generated column by flow engine, will be optional in next release
-- by then you can use `ALTER TABLE requests_without_ip DROP COLUMN update_at` to remove it
);

CREATE FLOW requests_long_term
SINK TO requests_without_ip
AS
SELECT
    service_name,
    val,
    ts
FROM requests;

INSERT INTO requests VALUES 
    ("svc1", "10.0.0.1", 100, "2024-10-18 19:00:00"),
    ("svc1", "10.0.0.2", 100, "2024-10-18 19:00:00"),
    ("svc1", "10.0.0.1", 200, "2024-10-18 19:00:30"),
    ("svc1", "10.0.0.2", 200, "2024-10-18 19:00:30"),
    ("svc1", "10.0.0.1", 300, "2024-10-18 19:01:00"),
    ("svc1", "10.0.0.2", 100, "2024-10-18 19:01:01"), 
-- notice here since ts is time index, so having same pk tag and timeindex will make `100` overwrite `400`, so a tiny difference is needed here
    ("svc1", "10.0.0.1", 400, "2024-10-18 19:01:30"),
    ("svc1", "10.0.0.2", 200, "2024-10-18 19:01:31");

admin flush_flow('requests_long_term'); -- flush the flow to make sure the data is in the table

SELECT * FROM requests_without_ip;

Then you can use PromQL to find the increase in the last minute:

sum(increase(requests_without_ip{service_name='svc1'}[1m]))

That should give you the sum of the increase in the last minute for svc1.

[dozer47528]

If just remove the column, how can the PromQL engine calcualte the correct result?

before aggregate:

    ("svc1", "10.0.0.1", 100, "2024-10-18 19:00:00"),
    ("svc1", "10.0.0.2", 100, "2024-10-18 19:00:30"),
    ("svc1", "10.0.0.1", 200, "2024-10-18 19:01:00"),
    ("svc1", "10.0.0.2", 200, "2024-10-18 19:01:30"),

after aggregate:

    ("svc1", 100, "2024-10-18 19:00:00"),
    ("svc1", 100, "2024-10-18 19:00:30"),
    ("svc1", 200, "2024-10-18 19:01:00"),
    ("svc1", 200, "2024-10-18 19:01:30"),

The result of sum(increase(requests_without_ip{service_name='svc1'}[1m])) would be 100. But should be 200.

[discord9]

If just remove the column, how can the PromQL engine calcualte the correct result?

before aggregate:

    ("svc1", "10.0.0.1", 100, "2024-10-18 19:00:00"),
    ("svc1", "10.0.0.2", 100, "2024-10-18 19:00:30"),
    ("svc1", "10.0.0.1", 200, "2024-10-18 19:01:00"),
    ("svc1", "10.0.0.2", 200, "2024-10-18 19:01:30"),

after aggregate:

    ("svc1", 100, "2024-10-18 19:00:00"),
    ("svc1", 100, "2024-10-18 19:00:30"),
    ("svc1", 200, "2024-10-18 19:01:00"),
    ("svc1", 200, "2024-10-18 19:01:30"),

The result of sum(increase(requests_without_ip{service_name='svc1'}[1m])) would be 100. But should be 200.

oh I think I get what you mean now, so ip actually matter in this query? Like:

("svc1", "10.0.0.1", 100, "2024-10-18 19:00:00"), --> ("svc1", "10.0.0.1", 200, "2024-10-18 19:01:00"), makes 100
("svc1", "10.0.0.2", 100, "2024-10-18 19:00:30"), --> ("svc1", "10.0.0.2", 200, "2024-10-18 19:01:30"), makes another 100, combine to 200?

Then maybe you can run your promql query(Modified with this ip included) in request table, and set a TTL(Time To Live) For request table(For like one or two minutes), and use flow to continouos remove server_ip column and save to requests_without_ip then? If I didn't misunderstand you?