ログクラスタリングでログの圧縮率を向上させる

1AddItemAsync called with userId=ea894cf4-a9b8-11f0-956c-4a218c6deb45, productId=0PUK6V6EV0, quantity=4
2GetCartAsync called with userId=7f3e16e6-a9f9-11f0-956c-4a218c6deb45
3AddItemAsync called with userId=a79c1e20-a9a0-11f0-956c-4a218c6deb45, productId=LS4PSXUNUM, quantity=3
4GetCartAsync called with userId=9a89945c-a9f9-11f0-8bd1-ee6fbde68079

1wget https://datasets-documentation.s3.eu-west-3.amazonaws.com/otel_demo/logs_recommendation.sample

1#!/usr/bin/env python3
2import sys
3from collections import defaultdict
4from drain3 import TemplateMiner
5from drain3.template_miner_config import TemplateMinerConfig
6
7def main():
8    lines = [ln.strip() for ln in sys.stdin if ln.strip()]
9    cfg = TemplateMinerConfig(); cfg.config_file = None
10    miner = TemplateMiner(None, cfg)
11
12    counts, templates, total = defaultdict(int), {}, 0
13    for raw in lines:
14        r = miner.add_log_message(raw)
15        cid = r["cluster_id"]; total += 1
16        counts[cid] += 1
17        templates[cid] = r["template_mined"]
18
19    items = [ (cnt, templates[cid]) for cid, cnt in counts.items() ]
20    items.sort(key=lambda x: (-x[0], x[1]))
21
22    for cnt, tmpl in items:
23        cov = (cnt / total * 100.0) if total else 0.0
24        print(f"{cov:.2f}\t{tmpl}")
25
26if __name__ == "__main__":
27    main()

1$ cat logs_recommendation.sample | python3 drain3_min.py
250.01	2025-09-15 <*> INFO [main] [recommendation_server.py:47] <*> <*> resource.service.name=recommendation trace_sampled=True] - Receive ListRecommendations for product <*> <*> <*> <*> <*>
349.99	Receive ListRecommendations for product <*> <*> <*> <*> <*>

1<functions>
2  <function>
3    <type>executable_pool</type>
4    <name>drain3_miner</name>
5    <return_type>Array(String)</return_type>
6    <return_name>result</return_name>
7    <argument>
8      <type>Array(String)</type>
9      <name>values</name>
10    </argument>
11    <format>JSONEachRow</format>
12    <command>drain3_miner.py</command>
13    <execute_direct>1</execute_direct> 
14    <pool_size>1</pool_size>
15    <max_command_execution_time>100</max_command_execution_time>
16    <command_read_timeout>100000</command_read_timeout>
17    <send_chunk_header>false</send_chunk_header>
18  </function>
19</functions>

# Install drain3 for all users
sudo pip install drain3

# Verify the clickhouse user has access to it
sudo -u clickhouse python3 -c "import drain3"

1-- Create table
2CREATE TABLE raw_logs
3(
4    `Body` String,
5    `ServiceName` String
6)
7ORDER BY tuple();
8
9-- Insert nginx access logs
10INSERT INTO raw_logs SELECT line As Body, 'nginx' as ServiceName FROM s3('https://datasets-documentation.s3.eu-west-3.amazonaws.com/http_logs/nginx-66.log.gz', 'LineAsString')
11
12-- Insert recommendation service logs
13INSERT INTO raw_logs SELECT line As Body, 'recommendation' as ServiceName FROM s3('https://datasets-documentation.s3.eu-west-3.amazonaws.com/otel_demo/logs_recommendation.log.gz', 'LineAsString')
14
15-- Insert cart service logs
16INSERT INTO raw_logs SELECT line As Body, 'cart' as ServiceName FROM s3('https://datasets-documentation.s3.eu-west-3.amazonaws.com/otel_demo/logs_cart.log.gz', 'LineAsString')

1WITH drain3_miner(groupArray(Body)) AS results
2SELECT
3    JSONExtractString(arrayJoin(results), 'template') AS template,
4    JSONExtractUInt(arrayJoin(results), 'count') AS count,
5    JSONExtractFloat(arrayJoin(results), 'coverage') AS coverage
6FROM
7(
8    SELECT Body
9    FROM raw_logs
10    WHERE (ServiceName = 'recommendation') AND (randCanonical() < 0.1)
11    LIMIT 10000
12)
13FORMAT VERTICAL

Row 1:
──────
template: <*> <*> INFO [main] [recommendation_server.py:47] <*> <*> resource.service.name=recommendation trace_sampled=True] - Receive ListRecommendations for product <*> <*> <*> <*> <*>
count:    5068
coverage: 50.68

Row 2:
──────
template: Receive ListRecommendations for product <*> <*> <*> <*> <*>
count:    4931
coverage: 49.31

Row 3:
──────
template: 2025-09-27 02:00:00,319 WARNING [opentelemetry.exporter.otlp.proto.grpc.exporter] [exporter.py:328] [trace_id=0 span_id=0 resource.service.name=recommendation trace_sampled=False] - Transient error StatusCode.UNAVAILABLE encountered while exporting logs to my-hyperdx-hdx-oss-v2-otel-collector:4317, retrying in 1s.
count:    1
coverage: 0.01

1CREATE TABLE logs_structured
2(
3    `ServiceName` LowCardinality(String),
4    `TemplateNumber` UInt8,
5    `Extracted` Map(LowCardinality(String), String)
6) ORDER BY (ServiceName, TemplateNumber)

1CREATE MATERIALIZED VIEW IF NOT EXISTS mv_logs_structured_min
2TO logs_structured
3AS
4SELECT
5    ServiceName,
6    /* which template matched */
7   multiIf(m1, 1, m2, 2, 0) AS TemplateNumber,
8    /* extracted fields as Map(LowCardinality(String), String) */
9    CAST(
10    multiIf(
11      m1,
12      map(
13        'date',           g1_1,
14        'time',           g1_2,
15        'service_name',   g1_3,
16        'trace_sampled',  g1_4,
17        'prod_1',         g1_5,
18        'prod_2',         g1_6,
19        'prod_3',         g1_7,
20        'prod_4',         g1_8,
21        'prod_5',         g1_9
22      ),
23      m2,
24      map(
25        'prod_1', g2_1,
26        'prod_2', g2_2,
27        'prod_3', g2_3,
28        'prod_4', g2_4,
29        'prod_5', g2_5
30      ),
31      map()                   -- else: empty map
32    ),
33    'Map(LowCardinality(String), String)'
34  ) AS Extracted
35FROM
36(
37    /* compute once per row */
38    WITH
39        '^([^\\s]+) ([^\\s]+) INFO \[main\] \[recommendation_server.py:47\] \[trace_id=([^\\s]+) span_id=([^\\s]+) resource\.service\.name=recommendation trace_sampled=True\] - Receive ListRecommendations for product ids:\[([^\\s]+) ([^\\s]+) ([^\\s]+) ([^\\s]+) ([^\\s]+)\]$' AS pattern1,
40        '^Receive ListRecommendations for product ([^\\s]+) ([^\\s]+) ([^\\s]+) ([^\\s]+) ([^\\s]+)$' AS pattern2
41
42    SELECT
43        *,
44        match(Body, pattern1) AS m1,
45        match(Body, pattern2) AS m2,
46
47        extractAllGroups(Body, pattern1) AS g1,
48        extractAllGroups(Body, pattern2) AS g2,
49
50        /* pick first (and only) match’s capture groups */
51        arrayElement(arrayElement(g1, 1), 1) AS g1_1,
52        arrayElement(arrayElement(g1, 1), 2) AS g1_2,
53        arrayElement(arrayElement(g1, 1), 3) AS g1_3,
54        arrayElement(arrayElement(g1, 1), 4) AS g1_4,
55        arrayElement(arrayElement(g1, 1), 5) AS g1_5,
56        arrayElement(arrayElement(g1, 1), 6) AS g1_6,
57        arrayElement(arrayElement(g1, 1), 7) AS g1_7,
58        arrayElement(arrayElement(g1, 1), 7) AS g1_8,
59        arrayElement(arrayElement(g1, 1), 7) AS g1_9,
60
61        arrayElement(arrayElement(g2, 1), 1) AS g2_1,
62        arrayElement(arrayElement(g2, 1), 2) AS g2_2,
63        arrayElement(arrayElement(g2, 1), 3) AS g2_3,
64        arrayElement(arrayElement(g2, 1), 4) AS g2_4,
65        arrayElement(arrayElement(g2, 1), 5) AS g2_5
66
67    FROM raw_logs where ServiceName='recommendation'
68) WHERE m1 OR m2;

1CREATE TABLE raw_logs_tmp as raw_logs
2EXCHANGE TABLES raw_logs AND raw_logs_tmp
3INSERT INTO raw_logs SELECT * FROM raw_logs_tmp

1SELECT
2    ServiceName,
3    TemplateNumber,
4    count()
5FROM logs_structured
6GROUP BY
7    ServiceName,
8    TemplateNumber
9ORDER BY
10    ServiceName ASC,
11    TemplateNumber ASC

┌─ServiceName────┬─TemplateNumber─┬──count()─┐
│ cart           │              0 │    66162 │
│ cart           │              3 │ 76793139 │
│ cart           │              4 │ 61116119 │
│ cart           │              5 │ 41877952 │
│ cart           │              6 │  1738375 │
│ nginx          │              0 │       16 │
│ nginx          │              7 │ 66747274 │
│ recommendation │              0 │     5794 │
│ recommendation │              1 │ 10537999 │
│ recommendation │              2 │ 10565640 │   └────────────────┴────────────────┴──────────┘

1ALTER TABLE logs_structured 
2ADD COLUMN  Body String ALIAS multiIf(
3        TemplateNumber=1, 
4        format('{0} {1} INFO [main] [recommendation_server.py:47] resource.service.name={2} trace_sampled={3}] - Receive ListRecommendations for product {4} {5} {6} {7} {8}',Extracted['date'],Extracted['time'],Extracted['service_name'],Extracted['trace_sampled'],Extracted['prod_1'],Extracted['prod_2'],Extracted['prod_3'],Extracted['prod_4'],Extracted['prod_5']),
5        TemplateNumber=2, 
6        format('Receive ListRecommendations for product {0} {1} {2} {3} {4}',Extracted['prod_1'],Extracted['prod_2'],Extracted['prod_3'],Extracted['prod_4'],Extracted['prod_5']),
7        TemplateNumber=3, 
8        format('GetCartAsync called with userId={0}',Extracted['user_id']),
9        TemplateNumber=4, 
10        'info: cart.cartstore.ValkeyCartStore[0]',
11        TemplateNumber=5, 
12        format('AddItemAsync called with userId={0}, productId={1}, quantity={2}', Extracted['user_id'], Extracted['product_id'], Extracted['quantity']),
13        TemplateNumber=6, 
14        format('EmptyCartAsync called with userId={0}',Extracted['user_id']),
15        TemplateNumber=7, 
16        format('{0} - {1} [{2}] "{3} {4} {5}" {6} {7} "{8}" "{9}"', Extracted['remote_addr'], Extracted['remote_user'], Extracted['time_local'], Extracted['request_type'], Extracted['request_path'], Extracted['request_protocol'], Extracted['status'], Extracted['size'], Extracted['referer'], Extracted['user_agent']),
17        '')

1SELECT Body
2FROM logs_structured
3WHERE ServiceName = 'nginx'
4LIMIT 1
5FORMAT vertical

Row 1:
──────
Body: 66.249.66.92 - - [2019-02-10 03:10:02] "GET /static/images/amp/third-party/footer-mobile.png HTTP/1.1" 200 62894 "-" "Googlebot-Image/1.0"

1SELECT Body
2FROM raw_logs
3WHERE Body = '66.249.66.92 - - [2019-02-10 03:10:02] "GET /static/images/amp/third-party/footer-mobile.png HTTP/1.1" 200 62894 "-" "Googlebot-Image/1.0"'
4LIMIT 1
5FORMAT vertical

Row 1:
──────
Body: 66.249.66.92 - - [2019-02-10 03:10:02] "GET /static/images/amp/third-party/footer-mobile.png HTTP/1.1" 200 62894 "-" "Googlebot-Image/1.0"

1SELECT
2    `table`,
3    formatReadableSize(sum(data_compressed_bytes)) AS compressed_size,
4    formatReadableSize(sum(data_uncompressed_bytes)) AS uncompressed_size
5FROM system.parts
6WHERE ((`table` = 'raw_logs') OR (`table` = 'logs_structured')) AND active
7GROUP BY `table`
8FORMAT VERTICAL

Row 1:
──────
table:             raw_logs
compressed_size:   2.00 GiB
uncompressed_size: 37.67 GiB

Row 2:
──────
table:             logs_structured
compressed_size:   1.71 GiB
uncompressed_size: 29.95 GiB

Uncompressed original size: 37.67 GiB
Compressed size on raw logs: 2.00 GiB - (18x compression ratio)
Compressed size on structured logs: 1.71 GiB - (22x compression ratio)

1-- Create table for cart service logs
2CREATE TABLE logs_service_cart
3(
4    TemplateNumber UInt8,
5    `user_id` Nullable(UUID),
6    `product_id` String,
7    `quantity` String,
8    Body ALIAS multiIf(
9        TemplateNumber=1, format('GetCartAsync called with userId={0}',user_id),
10        TemplateNumber=2, 'info: cart.cartstore.ValkeyCartStore[0]',
11        TemplateNumber=3, format('AddItemAsync called with userId={0}, productId={1}, quantity={2}', user_id, product_id, quantity),
12        TemplateNumber=4, format('EmptyCartAsync called with userId={0}',user_id),
13        '')
14)
15ORDER BY (TemplateNumber, product_id, quantity)
16
17
18-- Create materialized view for cart service logs
19CREATE MATERIALIZED VIEW IF NOT EXISTS mv_logs_cart
20TO logs_service_cart
21AS
22SELECT
23   multiIf(m1, 1, m2, 2, m3, 3, 0) AS TemplateNumber,
24   multiIf(m1, g1_1, m2, Null, m3, g3_1, m4, g4_1, Null) AS user_id,
25   multiIf(m1, '', m2, '', m3, g3_2, '') AS product_id,
26   multiIf(m1, '', m2, '', m3, g3_3, '') AS quantity
27
28FROM
29(
30    WITH
31        '^[\\s]*GetCartAsync called with userId=([^\\s]*)$' AS pattern1,
32        '^info\: cart.cartstore.ValkeyCartStore\[0\]$' AS pattern2,
33        '^[\\s]*AddItemAsync called with userId=([^\\s]+), productId=([^\\s]+), quantity=([^\\s]+)$' AS pattern3,
34        '^[\\s]*EmptyCartAsync called with userId=([^\\s]*)$' AS pattern4
35    SELECT
36        *,
37        match(Body, pattern1) AS m1,
38        match(Body, pattern2) AS m2,
39        match(Body, pattern3) AS m3,
40        match(Body, pattern4) AS m4,
41        extractAllGroups(Body, pattern1) AS g1,
42        extractAllGroups(Body, pattern2) AS g2,
43        extractAllGroups(Body, pattern3) AS g3,
44        extractAllGroups(Body, pattern4) AS g4,
45
46        arrayElement(arrayElement(g1, 1), 1) AS g1_1,
47        arrayElement(arrayElement(g3, 1), 1) AS g3_1,
48        arrayElement(arrayElement(g3, 1), 2) AS g3_2,
49        arrayElement(arrayElement(g3, 1), 3) AS g3_3,
50        arrayElement(arrayElement(g4, 1), 1) AS g4_1
51    FROM raw_logs where ServiceName='cart'
52);

1WITH (
2        SELECT sum(data_uncompressed_bytes)
3        FROM system.parts
4        WHERE (`table` = 'raw_logs') AND active
5    ) AS raw_uncompressed
6SELECT
7    label AS `table`,
8    formatReadableSize(sum(data_uncompressed_bytes)) AS uncompressed_bytes,
9    formatReadableSize(sum(data_compressed_bytes)) AS compressed_bytes,
10    sum(rows) AS nb_of_rows,
11    toUInt32(round(raw_uncompressed / sum(data_compressed_bytes))) AS compression_from_raw
12FROM
13(
14    SELECT
15        if(match(`table`, '^logs_service_'), 'logs_service_*', `table`) AS label,
16        data_uncompressed_bytes,
17        data_compressed_bytes,
18        rows
19    FROM system.parts
20    WHERE active AND ((`table` IN ('raw_logs', 'logs_structured')) OR match(`table`, '^logs_service_'))
21)
22GROUP BY label
23ORDER BY label ASC

┌─table───────────┬─uncompressed─┬─compressed─┬─nb_of_rows─┬─compression_from_raw─┐
│ logs_service_*  │ 16.58 GiB    │ 865.16 MiB │  269448454 │                   45 │
│ logs_structured │ 29.95 GiB    │ 1.71 GiB   │  269448470 │                   22 │
│ raw_logs        │ 37.71 GiB    │ 2.01 GiB   │  269448470 │                   19 │   └─────────────────┴──────────────┴────────────┴────────────┴──────────────────────┘

1SELECT Body
2FROM merge(currentDatabase(), '^logs_service_')
3ORDER BY rand() ASC
4LIMIT 10
5FORMAT TSV

info: cart.cartstore.ValkeyCartStore[0]
AddItemAsync called with userId={userId}, productId={productId}, quantity={quantity}
AddItemAsync called with userId=6dd06afe-a9da-11f0-8754-96b7632aa52f, productId=L9ECAV7KIM, quantity=4
info: cart.cartstore.ValkeyCartStore[0]
GetCartAsync called with userId=c6a2e0fc-a9e5-11f0-a910-e6976c512022
info: cart.cartstore.ValkeyCartStore[0]
GetCartAsync called with userId=0745841e-a970-11f0-ae33-92666e0294bc
info: cart.cartstore.ValkeyCartStore[0]
84.47.202.242 - - [2019-02-21 05:01:17] "GET /image/32964?name=bl1189-13.jpg&wh=300x300 HTTP/1.1" 200 8914 "https://www.zanbil.ir/product/32964/63521/%D9%85%D8%AE%D9%84%D9%88%D8%B7-%DA%A9%D9%86-%D9%85%DB%8C%D8%AF%DB%8C%D8%A7-%D9%85%D8%AF%D9%84-BL1189" "Mozilla/5.0 (Windows NT 6.1; WOW64; rv:64.0) Gecko/20100101 Firefox/64.0"
GetCartAsync called with userId={userId}