ClickHouse¶

name: clickhouse

services:
  clickhouse:
    image: clickhouse/clickhouse-server:26.3
    container_name: clickhouse
    restart: unless-stopped
    ports:
      - "8123:8123"  # HTTP 端口
      - "9000:9000"  # TCP 端口
      # "9009:9009"  # 集群内的 TCP 通信端口
    ulimits:
      nofile: # (1)
        soft: 262144
        hard: 262144
    environment:
      CLICKHOUSE_DB: ${CLICKHOUSE_DB}
      CLICKHOUSE_USER: ${CLICKHOUSE_USER}
      CLICKHOUSE_PASSWORD: ${CLICKHOUSE_PASSWORD}
    volumes:
      - /srv/clickhouse/data:/var/lib/clickhouse
      - /srv/clickhouse/logs:/var/log/clickhouse-server
      # ./init:/docker-entrypoint-initdb.d:ro
      # ./config/users.xml:/etc/clickhouse-server/users.d/users.xml:ro

<clickhouse>
  <users>
    <grafana>
      <password>grafana_password</password>
      <networks>
        <ip>10.0.0.1</ip>
      </networks>
      <profile>grafana_readonly</profile>
      <quota>default</quota>
      <readonly>1</readonly>
    </grafana>

    <mirrors>
      <password>mirrors_password</password>
      <networks>
        <ip>10.0.0.0/24</ip>
      </networks>
      <profile>default</profile>
      <quota>default</quota>
      <grants>
        <query>GRANT SELECT ON mirrors.*</query>
        <query>GRANT INSERT ON mirrors.*</query>
      </grants>
    </mirrors>
  </users>
</clickhouse>

<clickhouse>
  <profiles>
    <grafana_readonly>
      <readonly>1</readonly>
      <max_execution_time>300</max_execution_time>
      <max_result_rows>100000</max_result_rows>
      <result_overflow_mode>throw</result_overflow_mode>
      <constraints>
        <limit>
          <min>0</min>
          <max>10000000</max>
          <changeable_in_readonly/>
        </limit>
        <max_execution_time>
          <changeable_in_readonly/>
        </max_execution_time>
        <additional_table_filters>
          <changeable_in_readonly/>
        </additional_table_filters>
        <use_query_cache>
          <min>0</min>
          <max>1</max>
          <changeable_in_readonly/>
        </use_query_cache>
        <query_cache_ttl>
          <min>1</min>
          <max>300</max>
          <changeable_in_readonly/>
        </query_cache_ttl>
      </constraints>
    </grafana_readonly>
  </profiles>
</clickhouse>

<clickhouse>
  <compression>
    <case>
      <min_part_size>0</min_part_size>
      <min_part_size_ratio>0</min_part_size_ratio>
      <method>zstd</method>
      <level>3</level>
    </case>
  </compression>
</clickhouse>

<clickhouse>
  <max_thread_pool_size>4096</max_thread_pool_size>
  <max_thread_pool_free_size>512</max_thread_pool_free_size>
  <thread_pool_queue_size>4096</thread_pool_queue_size>
  <background_pool_size>16</background_pool_size>
  <background_merges_mutations_concurrency_ratio>2</background_merges_mutations_concurrency_ratio>
  <background_schedule_pool_size>64</background_schedule_pool_size>
  <background_move_pool_size>8</background_move_pool_size>
</clickhouse>

<clickhouse>
  <profiles>
    <default>
      <max_threads>16</max_threads>
      <max_insert_threads>8</max_insert_threads>
    </default>
  </profiles>
</clickhouse>

SELECT name, value, default, changed, description
FROM system.server_settings
WHERE name IN
(
  'max_thread_pool_size',
  'max_thread_pool_free_size',
  'thread_pool_queue_size',
  'background_pool_size',
  'background_merges_mutations_concurrency_ratio',
  'background_schedule_pool_size',
  'background_move_pool_size'
)
ORDER BY name;

<clickhouse>
  <max_concurrent_queries>64</max_concurrent_queries>
  <max_concurrent_select_queries>48</max_concurrent_select_queries>
  <max_concurrent_insert_queries>16</max_concurrent_insert_queries>
  <concurrent_threads_soft_limit_ratio_to_cores>2</concurrent_threads_soft_limit_ratio_to_cores>
  <concurrent_threads_scheduler>fair_round_robin</concurrent_threads_scheduler>
  <max_server_memory_usage_to_ram_ratio>0.80</max_server_memory_usage_to_ram_ratio>
</clickhouse>

<clickhouse>
  <profiles>
    <grafana_readonly>
      <max_threads>8</max_threads>
      <max_memory_usage>8589934592</max_memory_usage>
      <max_memory_usage_for_user>25769803776</max_memory_usage_for_user>
      <max_execution_time>300</max_execution_time>
    </grafana_readonly>
  </profiles>
</clickhouse>

CREATE TABLE mirrors.access_log (
    `timestamp` Float64 CODEC(DoubleDelta, Default),
    `event_time` DateTime64(3, 'UTC') MATERIALIZED fromUnixTimestamp64Milli(toInt64(timestamp * 1000)) CODEC(DoubleDelta, Default),
    `serverip` LowCardinality(String),
    `method` LowCardinality(String),
    `scheme` LowCardinality(String),
    `url` String CODEC(ZSTD(3)),
    `status` UInt16,
    `size` UInt64,
    `resp_time` Float32,
    `http_host` LowCardinality(String),
    `referer` String CODEC(ZSTD(3)),
    `user_agent` String CODEC(ZSTD(3)),
    `request_id` String DEFAULT hex(cityHash64(tuple(event_time, clientip))) CODEC(ZSTD(3)),
    `proto` LowCardinality(String),
    `proxied` LowCardinality(String),
    `alpn` LowCardinality(String),
    `clientip` IPv6 CODEC(ZSTD(1)),
    `source` LowCardinality(String) DEFAULT '',
    `repo` LowCardinality(String) MATERIALIZED extract_repo("url"),
    `ip` IPv6 ALIAS clientip
)
ENGINE = ReplacingMergeTree
PARTITION BY toDate(event_time)
PRIMARY KEY event_time
ORDER BY (event_time, request_id)
SETTINGS index_granularity = 8192;

-- 提取 URL path 中的第一段目录
CREATE FUNCTION extract_repo_dir AS (path) ->
if(
  match(path, '^/*[^/?]+/'),
  extract(path, '^/*([^/?]+)/'),
  '/'
);

CREATE FUNCTION extract_repo AS (path) ->
multiIf(
  has([
    'adoptium', 'alpine', 'anaconda', -- 一大串仓库名
    'zerotier',
    '/'
  ], extract_repo_dir(path)), extract_repo_dir(path),
  has(['assets', 'static', 'status', '.well-known'], extract_repo_dir(path)), '/',
  has(['misc'], extract_repo_dir(path)), 'ustclug',
  '<invalid>'
);

CREATE FUNCTION ip_prefix AS (ip, ipv4_length, ipv6_length) ->
if(
  isIPAddressInRange(ip, '::ffff:0.0.0.0/96'),
  if(ipv4_length = 32,
    replaceOne(toString(tupleElement(IPv6CIDRToRange(ip, toUInt8(96 + ipv4_length)), 1)), '::ffff:', ''),
    concat(replaceOne(toString(tupleElement(IPv6CIDRToRange(ip, toUInt8(96 + ipv4_length)), 1)), '::ffff:', ''), '/', toString(ipv4_length))
  ),
  if(ipv6_length = 128,
    toString(tupleElement(IPv6CIDRToRange(ip, ipv6_length), 1)),
    concat(toString(tupleElement(IPv6CIDRToRange(ip, ipv6_length), 1)), '/', toString(ipv6_length))
  )
);

CREATE TABLE mirrors.repo_names (
  name String,
  canonical_name String,
  updated_at DateTime64(3) DEFAULT now64()
)
ENGINE = ReplacingMergeTree(updated_at)
ORDER BY name;

CREATE DICTIONARY mirrors.repo_names_dict (
  name String,
  canonical_name String,
  updated_at DateTime
)
PRIMARY KEY name
SOURCE(CLICKHOUSE(
  DB 'mirrors'
  TABLE 'repo_names'
))
LIFETIME(MIN 0 MAX 300)
LAYOUT(HASHED());

INSERT INTO mirrors.repo_names (name, canonical_name)
SELECT name, name
FROM (
  SELECT arrayJoin([
    'adoptium', 'alpine', 'anaconda', -- 一大串仓库名
    'zerotier',
    'help', '/'
  ]) AS name
);

INSERT INTO mirrors.repo_names (name, canonical_name) VALUES
  ('archive.raspberrypi.org', 'raspberrypi'),
  ('assets', '/'),
  ('static', '/'),
  ('status', '/'),
  ('.well-known', '/'),
  ('misc', 'ustclug');

CREATE OR REPLACE FUNCTION extract_repo AS (path) ->
  dictGetOrDefault(
    'mirrors.repo_names_dict',
    'canonical_name',
    extract_repo_dir(path),
    '<invalid>'
  );

ALTER TABLE mirrors.access_log
  MODIFY COLUMN `repo` LowCardinality(String) MATERIALIZED extract_repo("url");

ALTER TABLE mirrors.access_log
  MATERIALIZE COLUMN `repo`;

sources:
  mirror_logs:
    type: file
    include:
      - /var/log/nginx/mirrors/access_json.log
      - /var/log/nginx/403/access_json.log
      - /var/log/nginx/cacheproxy/*_json.log
    read_from: end
    fingerprint:
      strategy: device_and_inode

transforms:
  parsed_logs:
    type: remap
    inputs:
      - mirror_logs
    source: |
      filename = .file
      . = parse_json!(string!(.message))
      .source = ""
      if contains!(filename, "/mirrors/") {
        .source = "mirrors"
      } else if contains!(filename, "/403/") {
        .source = "403"
      } else if contains!(filename, "/cacheproxy/") {
        .source = "cacheproxy"
      }

sinks:
  mirrorlog:
    type: clickhouse
    inputs:
      - parsed_logs
    endpoint: http://localhost:8123/
    auth:
      strategy: basic
      user: n-buna
      password: "umiyuri kaiteitan"
    database: mirrors
    table: access_log
    format: json_each_row
    buffer:
      type: disk
      max_size: 1073741824

sources:
  nginx_syslog_udp:
    type: syslog
    address: 0.0.0.0:5514
    mode: udp
  nginx_syslog_tcp:
    type: syslog
    address: 0.0.0.0:5514
    mode: tcp

transforms:
  parsed_syslog:
    type: remap
    inputs:
      - nginx_syslog_udp
      - nginx_syslog_tcp
    source: |
      syslog_host = to_string(.hostname) ?? ""
      . = parse_json!(string!(.message))
      .source = syslog_host

for i in mirrors/access_json.log-*.xz; do
  echo "$i" >&2
  xzcat "$i" |
    pv -s $(xz --robot -l "$i" | awk 'NR==2{print $5}') |
    jq '.source = "mirrors"' |
    clickhouse-client --query 'INSERT INTO "mirrors"."access_log" FORMAT JSONEachRow'
done

SELECT
  ip_prefix("ip", 24, 48) AS "IP",
  COUNT(DISTINCT "ip") AS "Unique IPs",
  count() AS "Request Count",
  sum("size") AS "Bytes Sent",
  argMax("url", "event_time") AS "Last URL",
  max("event_time") AS "Last Seen",
  count(DISTINCT "user_agent") AS "Unique UAs"
FROM "mirrors"."access_log"
WHERE $__timeFilter("event_time")
  AND "source" = 'mirrors'
GROUP BY "IP"
ORDER BY "Bytes Sent" DESC, "IP" ASC
LIMIT 50;

WITH
  repo_size AS MATERIALIZED (
    SELECT
      repo,
      sum("size") AS "sum_size"
    FROM "mirrors"."access_log"
    WHERE $__timeFilter(event_time)
    GROUP BY "repo"
  ),
  top_repos AS MATERIALIZED (
    SELECT "repo", "sum_size"
    FROM "repo_size"
    ORDER BY "sum_size" DESC
    LIMIT 20
  )
SELECT
  if("repo" IN (SELECT "repo" FROM "top_repos"), "repo", '[Others]') AS "Repository",
  sum("sum_size") AS "Bytes Sent"
FROM "repo_size"
GROUP BY "Repository"
ORDER BY "Repository" = '[Others]' ASC, "Bytes Sent" DESC;

CREATE TABLE mirrors.access_log_30s
(
    event_time DateTime('UTC') CODEC(DoubleDelta, Default),
    source LowCardinality(String),
    http_host LowCardinality(String),
    repo LowCardinality(String),
    status UInt16,
    method LowCardinality(String),
    pv AggregateFunction(count),
    uv AggregateFunction(uniqCombined64, IPv6),
    bytes AggregateFunction(sum, UInt64),
    resp_time_avg AggregateFunction(avg, Float32)
)
ENGINE = AggregatingMergeTree
PARTITION BY toDate(event_time)
ORDER BY
(
    event_time,
    source,
    http_host,
    repo,
    status,
    method
)
SETTINGS
    index_granularity = 8192,
    default_compression_codec = 'ZSTD(3)';

CREATE MATERIALIZED VIEW mirrors.access_log_30s_mv
TO mirrors.access_log_30s
AS
SELECT
    toDateTime(
      toStartOfInterval(event_time, INTERVAL 30 SECOND),
      'UTC'
    ) AS event_time,
    source,
    http_host,
    repo,
    status,
    method,
    countState() AS pv,
    uniqCombined64State(clientip) AS uv,
    sumState(size) AS bytes,
    avgState(resp_time) AS resp_time_avg
FROM mirrors.access_log
GROUP BY
    event_time,
    source,
    http_host,
    repo,
    status,
    method;

SELECT
  toStartOfInterval(event_time, INTERVAL 10 MINUTE) AS time,
  countMerge(pv) AS pv,
  uniqCombined64Merge(uv) AS uv,
  sumMerge(bytes) AS bytes,
  avgMerge(resp_time_avg) AS avg_resp_time
FROM mirrors.access_log_30s
WHERE $__timeFilter(event_time)
  AND source IN (${source:singlequote})
GROUP BY time
ORDER BY time;

INSERT INTO mirrors.access_log_30s
SELECT
  toDateTime(
    toStartOfInterval(event_time, INTERVAL 30 SECOND),
    'UTC'
  ) AS event_time,
  source,
  http_host,
  repo,
  status,
  method,
  countState(),
  uniqCombined64State(clientip),
  sumState(size),
  avgState(resp_time)
FROM mirrors.access_log
WHERE event_time < toDateTime64('2026-08-09 09:31:00', 3, 'UTC')
GROUP BY
  event_time,
  source,
  http_host,
  repo,
  status,
  method;

WITH
  toStartOfInterval(now('UTC'), INTERVAL 30 SECOND)
    - INTERVAL 5 MINUTE AS stop,
  stop - INTERVAL 24 HOUR AS start
SELECT
  'raw' AS source_table,
  count() AS pv,
  uniqCombined64(clientip) AS uv,
  sum(size) AS bytes,
  avg(resp_time) AS avg_resp_time
FROM mirrors.access_log
WHERE event_time >= start AND event_time < stop
UNION ALL
SELECT
  '30s' AS source_table,
  countMerge(pv),
  uniqCombined64Merge(uv),
  sumMerge(bytes),
  avgMerge(resp_time_avg)
FROM mirrors.access_log_30s
WHERE event_time >= start AND event_time < stop;

SELECT
  repo,
  sumMerge(bytes) AS bytes
FROM mirrors.access_log_30s
WHERE event_time >= {start:DateTime}
  AND event_time < {stop:DateTime}
GROUP BY repo
ORDER BY bytes DESC
LIMIT 20
SETTINGS
  use_query_cache = 1,
  query_cache_ttl = 30;

SELECT
  count() AS entries,
  formatReadableSize(sum(result_size)) AS result_size
FROM system.query_cache;

<clickhouse>
  <metric_log>
    <database>system</database>
    <table>metric_log</table>
    <flush_interval_milliseconds>7500</flush_interval_milliseconds>
    <collect_interval_milliseconds>1000</collect_interval_milliseconds>
    <max_size_rows>1048576</max_size_rows>
    <reserved_size_rows>8192</reserved_size_rows>
    <buffer_size_rows_flush_threshold>524288</buffer_size_rows_flush_threshold>
    <ttl>event_date + INTERVAL 30 DAY DELETE</ttl>
    <flush_on_crash>false</flush_on_crash>
  </metric_log>
</clickhouse>

<clickhouse>
  <background_schedule_pool_log>
    <database>system</database>
    <table>background_schedule_pool_log</table>
    <partition_by>toYYYYMM(event_date)</partition_by>
    <flush_interval_milliseconds>7500</flush_interval_milliseconds>
    <max_size_rows>1048576</max_size_rows>
    <reserved_size_rows>8192</reserved_size_rows>
    <buffer_size_rows_flush_threshold>524288</buffer_size_rows_flush_threshold>
    <duration_threshold_milliseconds>100</duration_threshold_milliseconds>
    <flush_on_crash>false</flush_on_crash>
  </background_schedule_pool_log>
</clickhouse>

SELECT dashboard, title, query
FROM system.dashboards
WHERE dashboard IN ('Overview', 'Memory (host)')
ORDER BY dashboard, title;

SELECT
  $__timeInterval(event_time) AS time,
  avg(ProfileEvent_OSCPUVirtualTimeMicroseconds) / 1000000 AS cores
FROM system.metric_log
WHERE $__timeFilter(event_time)
GROUP BY time
ORDER BY time WITH FILL;

SELECT metric, value, description
FROM system.metrics
WHERE metric IN
(
  'GlobalThread',
  'GlobalThreadActive',
  'GlobalThreadScheduled',
  'BackgroundMergesAndMutationsPoolTask',
  'BackgroundMergesAndMutationsPoolSize',
  'BackgroundMovePoolTask',
  'BackgroundMovePoolSize',
  'BackgroundSchedulePoolTask',
  'BackgroundSchedulePoolSize'
)
ORDER BY metric;

SELECT
  pool,
  count() AS tasks,
  countIf(scheduled) AS scheduled,
  countIf(delayed) AS delayed,
  countIf(executing) AS executing,
  max(elapsed_ms) AS max_elapsed_ms
FROM system.background_schedule_pool
GROUP BY pool
ORDER BY pool;

SELECT time, value, metric
FROM
(
  SELECT
    $__timeInterval(event_time) AS time,
    avg(CurrentMetric_MemoryTracking) AS value,
    'Tracked' AS metric
  FROM system.metric_log
  WHERE $__timeFilter(event_time)
  GROUP BY time
  UNION ALL
  SELECT
    $__timeInterval(event_time) AS time,
    avg(value) AS value,
    'RSS' AS metric
  FROM system.asynchronous_metric_log
  WHERE $__timeFilter(event_time)
    AND metric = 'MemoryResident'
  GROUP BY time
)
ORDER BY time, metric;

SELECT
  $__timeInterval(event_time) AS time,
  avg(CurrentMetric_GlobalThread) AS Created,
  avg(CurrentMetric_GlobalThreadActive) AS Active,
  avg(CurrentMetric_GlobalThreadScheduled) AS Scheduled
FROM system.metric_log
WHERE $__timeFilter(event_time)
GROUP BY time
ORDER BY time WITH FILL;

SELECT
  $__timeInterval(event_time) AS time,
  avg(CurrentMetric_BackgroundMergesAndMutationsPoolTask) AS Active,
  avg(CurrentMetric_BackgroundMergesAndMutationsPoolSize) AS Size,
  100 * Active / nullIf(Size, 0) AS utilization
FROM system.metric_log
WHERE $__timeFilter(event_time)
GROUP BY time
ORDER BY time WITH FILL;

<clickhouse>
  <storage_configuration>
    <disks>
      <mirrors_warm>
        <path>/var/lib/clickhouse-warm/</path>
      </mirrors_warm>
      <mirrors_cold>
        <path>/var/lib/clickhouse-cold/</path>
      </mirrors_cold>
    </disks>
    <policies>
      <mirrors_tiered>
        <volumes>
          <default>
            <disk>default</disk>
          </default>
          <warm>
            <disk>mirrors_warm</disk>
          </warm>
          <cold>
            <disk>mirrors_cold</disk>
          </cold>
        </volumes>
        <move_factor>0.10</move_factor>
      </mirrors_tiered>
    </policies>
  </storage_configuration>
</clickhouse>

ALTER TABLE mirrors.access_log
MODIFY SETTING storage_policy = 'mirrors_tiered';

ALTER TABLE mirrors.access_log
MODIFY TTL
  event_time + INTERVAL 30 DAY TO VOLUME 'warm',
  event_time + INTERVAL 90 DAY TO VOLUME 'cold',
  event_time + INTERVAL 180 DAY DELETE;

SELECT
  estimateCompressionRatio('ZSTD(3)')(event_time) AS zstd3,
  estimateCompressionRatio(
    'DoubleDelta, ZSTD(3)'
  )(event_time) AS doubledelta_zstd3
FROM mirrors.access_log_30s;

SELECT
  table,
  formatReadableSize(sum(data_compressed_bytes)) AS compressed,
  formatReadableSize(sum(data_uncompressed_bytes)) AS uncompressed,
  round(
    sum(data_compressed_bytes) / sum(data_uncompressed_bytes) * 100,
    4
  ) AS compression_percent,
  round(
    sum(data_uncompressed_bytes) / sum(data_compressed_bytes),
    3
  ) AS compression_ratio
FROM system.parts
WHERE active
  AND database = 'mirrors'
GROUP BY table
ORDER BY table;

SELECT
  toStartOfInterval(event_time, INTERVAL 10 MINUTE) AS time,
  count() AS PV
FROM ${clickhouse_adhoc_query:raw}
WHERE $__timeFilter(event_time)
  AND http_host IN (${domain:singlequote})
  AND $__conditionalAll(extract_repo(url) IN (${repo:singlequote}), $repo)
  AND $__conditionalAll(status IN (${status:csv}), $status)
  AND $__conditionalAll(request_method IN (${method:singlequote}), $method)
  AND $__conditionalAll(client_country IN (${country:singlequote}), $country)
  AND $__conditionalAll(client_asn IN (${asn:csv}), $asn)
GROUP BY time
ORDER BY time;

SELECT
  toStartOfInterval(event_time, INTERVAL 10 MINUTE) AS time,
  countMerge(pv) AS PV
FROM mirrorlog.access_log_30s
WHERE $__timeFilter(event_time)
  AND http_host IN (${domain:singlequote})
  AND $__conditionalAll(repo IN (${repo:singlequote}), $repo)
  AND $__conditionalAll(status IN (${status:csv}), $status)
  AND $__conditionalAll(request_method IN (${method:singlequote}), $method)
  AND $__conditionalAll(client_country IN (${country:singlequote}), $country)
  AND $__conditionalAll(client_asn IN (${asn:csv}), $asn)
GROUP BY time
ORDER BY time;