Uneven pipeline distribution across indexers

[earlbread]

Hello,

Is there a way to evenly distribute indexing pipelines across all Indexers?

We are currently using 16 Indexers(K8s pods), each with 8 CPU cores and 32GB of memory. The index configuration sets num_pipelines to 32, and the Kafka topic also has 32 partitions. However, it appears that only a few Indexers are assigned the pipelines, while the rest remain underutilized. It seems that the only way to increase throughput is to add more CPU.

Is there a method to ensure that the pipelines are evenly distributed across all Indexers?

We are currently using the qw-airmail-20250522-hotfix version.

Current resource usage:

Indexing Throughput:

Below is the current index configuration:

Source

[
  {
    "version": "0.9",
    "source_id": "kafka_to_access_log_v2_quickwit",
    "num_pipelines": 32,
    "enabled": true,
    "source_type": "kafka",
    "params": {
      "topic": "log.common.access_log_v2_quickwit",
      "client_params": {
        "bootstrap.servers": "b-1.loggingkafka.euq7fc.c3.kafka.ap-northeast-2.amazonaws.com,b-2.loggingkafka.euq7fc.c3.kafka.ap-northeast-2.amazonaws.com,b-3.loggingkafka.euq7fc.c3.kafka.ap-northeast-2.amazonaws.com",
        "group.id": "log.quickwit.access_log_v2"
      }
    },
    "transform": {
      "script": "# `occurredAt` is the timestamp of the log, but `time` is the timestamp of the log entry.\n# So we use `occurred_at` as `time` if it exists.\nif exists(.data.accessLogV2.occurredAt) {\n  .time = .data.accessLogV2.occurredAt\n}\n# unwrapped data fields\nunwrappedDataFields = merge(., {\n  \"requestId\": .data.accessLogV2.requestId,\n  \"ip\": .data.accessLogV2.ip,\n  \"sessionId\": .data.accessLogV2.sessionId,\n  \"userAgent\": .data.accessLogV2.userAgent,\n  \"xUserAgent\": .data.accessLogV2.xUserAgent,\n  \"userId\": .data.accessLogV2.userId,\n  \"deviceId\": .data.accessLogV2.deviceId,\n  \"latency\": .data.accessLogV2.latency,\n  \"latencyNs\": .data.accessLogV2.latencyNs,\n  \"occurredAt\": .data.accessLogV2.occurredAt,\n  \"httpStatusCode\": .data.accessLogV2.httpStatusCode,\n  \"httpMethod\": .data.accessLogV2.httpMethod,\n  \"httpPath\": .data.accessLogV2.httpPath,\n  \"grpcStatusCode\": .data.accessLogV2.grpcStatusCode,\n  \"grpcMethod\": .data.accessLogV2.grpcMethod,\n  \"extra\": .data.accessLogV2.extra,\n  \"kafkaConsumerGroupId\": .data.accessLogV2.kafkaConsumerGroupId,\n  \"kafkaConsumerClientId\": .data.accessLogV2.kafkaConsumerClientId,\n  \"kafkaConsumerHostName\": .data.accessLogV2.kafkaConsumerHostName,\n  \"kafkaTopic\": .data.accessLogV2.kafkaTopic,\n  \"kafkaPartition\": .data.accessLogV2.kafkaPartition,\n  \"kafkaOffset\": .data.accessLogV2.kafkaOffset,\n  \"kafkaMessageKey\": .data.accessLogV2.kafkaMessageKey,\n  \"kafkaConsumingResult\":.data.accessLogV2.kafkaConsumingResult\n})\n\n# remove unused fields\ndel(unwrappedDataFields.data)\ndel(unwrappedDataFields.specversion)\ndel(unwrappedDataFields.type)\n\n. = unwrappedDataFields\n",
      "timezone": "UTC"
    },
    "input_format": "json"
  },
  {
    "version": "0.9",
    "source_id": "_ingest-api-source",
    "num_pipelines": 1,
    "enabled": true,
    "source_type": "ingest-api",
    "input_format": "json"
  },
  {
    "version": "0.9",
    "source_id": "_ingest-source",
    "num_pipelines": 1,
    "enabled": true,
    "source_type": "ingest",
    "input_format": "json"
  },
  {
    "version": "0.9",
    "source_id": "_ingest-cli-source",
    "num_pipelines": 1,
    "enabled": true,
    "source_type": "ingest-cli",
    "input_format": "json"
  }
]

Doc mapping

{
  "doc_mapping_uid": "00000000000000000000000000",
  "mode": "dynamic",
  "dynamic_mapping": {
    "indexed": true,
    "tokenizer": "raw",
    "record": "basic",
    "stored": true,
    "expand_dots": true,
    "fast": {
      "normalizer": "raw"
    }
  },
  "field_mappings": [
    {
      "name": "id",
      "type": "text",
      "description": "unique identifier for the event",
      "fast": false,
      "fieldnorms": false,
      "indexed": true,
      "record": "basic",
      "stored": true,
      "tokenizer": "raw"
    },
    {
      "name": "specversion",
      "type": "text",
      "description": "version information about the CloudEvents specification",
      "fast": false,
      "fieldnorms": false,
      "indexed": false,
      "stored": false
    },
    {
      "name": "source",
      "type": "text",
      "description": "information about where the event occurred",
      "fast": false,
      "fieldnorms": false,
      "indexed": true,
      "record": "basic",
      "stored": true,
      "tokenizer": "raw"
    },
    {
      "name": "subject",
      "type": "text",
      "description": "detailed information about the source where the event occurred",
      "fast": false,
      "fieldnorms": false,
      "indexed": true,
      "record": "basic",
      "stored": true,
      "tokenizer": "raw"
    },
    {
      "name": "time",
      "type": "datetime",
      "description": "timestamp of the event",
      "fast": true,
      "fast_precision": "seconds",
      "indexed": true,
      "input_formats": [
        "unix_timestamp",
        "iso8601"
      ],
      "output_format": "unix_timestamp_nanos",
      "stored": true
    },
    {
      "name": "datacontenttype",
      "type": "text",
      "description": "content type of the data",
      "fast": false,
      "fieldnorms": false,
      "indexed": true,
      "record": "basic",
      "stored": true,
      "tokenizer": "raw"
    },
    {
      "name": "requestId",
      "type": "text",
      "description": "request id",
      "fast": false,
      "fieldnorms": false,
      "indexed": true,
      "record": "basic",
      "stored": true,
      "tokenizer": "raw"
    },
    {
      "name": "ip",
      "type": "ip",
      "description": "ip address",
      "fast": true,
      "indexed": true,
      "stored": true
    },
    {
      "name": "userAgent",
      "type": "text",
      "fast": false,
      "fieldnorms": false,
      "indexed": true,
      "record": "basic",
      "stored": true,
      "tokenizer": "default"
    },
    {
      "name": "xUserAgent",
      "type": "text",
      "fast": false,
      "fieldnorms": false,
      "indexed": true,
      "record": "basic",
      "stored": true,
      "tokenizer": "default"
    },
    {
      "name": "userId",
      "type": "text",
      "fast": false,
      "fieldnorms": false,
      "indexed": true,
      "record": "basic",
      "stored": true,
      "tokenizer": "raw"
    },
    {
      "name": "deviceId",
      "type": "text",
      "description": "device's id eg) 6ae1f6a6-107d-3183-a1df-adf7368f9d10",
      "fast": false,
      "fieldnorms": false,
      "indexed": true,
      "record": "basic",
      "stored": true,
      "tokenizer": "raw"
    },
    {
      "name": "latency",
      "type": "text",
      "description": "latency of the request, but already in latencyNs, so we don't need to index it",
      "fast": false,
      "fieldnorms": false,
      "indexed": false,
      "stored": false
    },
    {
      "name": "latencyNs",
      "type": "i64",
      "coerce": true,
      "description": "latency of the request in nanoseconds",
      "fast": true,
      "indexed": true,
      "output_format": "number",
      "stored": true
    },
    {
      "name": "occurredAt",
      "type": "datetime",
      "description": "timestamp of the log entry",
      "fast": false,
      "fast_precision": "seconds",
      "indexed": true,
      "input_formats": [
        "unix_timestamp",
        "iso8601"
      ],
      "output_format": "unix_timestamp_nanos",
      "stored": true
    },
    {
      "name": "httpStatusCode",
      "type": "i64",
      "coerce": true,
      "description": "HTTP status code",
      "fast": true,
      "indexed": true,
      "output_format": "number",
      "stored": true
    },
    {
      "name": "httpMethod",
      "type": "text",
      "description": "HTTP method",
      "fast": {
        "normalizer": "raw"
      },
      "fieldnorms": false,
      "indexed": true,
      "record": "basic",
      "stored": true,
      "tokenizer": "default"
    },
    {
      "name": "httpPath",
      "type": "text",
      "description": "HTTP path",
      "fast": {
        "normalizer": "raw"
      },
      "fieldnorms": false,
      "indexed": true,
      "record": "basic",
      "stored": true,
      "tokenizer": "default"
    },
    {
      "name": "grpcStatusCode",
      "type": "i64",
      "coerce": true,
      "description": "gRPC status code",
      "fast": true,
      "indexed": true,
      "output_format": "number",
      "stored": true
    },
    {
      "name": "grpcMethod",
      "type": "text",
      "description": "gRPC method",
      "fast": {
        "normalizer": "raw"
      },
      "fieldnorms": false,
      "indexed": true,
      "record": "basic",
      "stored": true,
      "tokenizer": "default"
    },
    {
      "name": "extra",
      "type": "json",
      "description": "extra fields",
      "expand_dots": true,
      "fast": false,
      "indexed": true,
      "record": "basic",
      "stored": true,
      "tokenizer": "raw"
    },
    {
      "name": "kafkaConsumerGroupId",
      "type": "text",
      "description": "Kafka Consumer Group Id",
      "fast": false,
      "fieldnorms": false,
      "indexed": true,
      "record": "basic",
      "stored": true,
      "tokenizer": "default"
    },
    {
      "name": "kafkaConsumerClientId",
      "type": "text",
      "description": "Kafka Consumer Client Id",
      "fast": false,
      "fieldnorms": false,
      "indexed": true,
      "record": "basic",
      "stored": true,
      "tokenizer": "default"
    },
    {
      "name": "kafkaConsumerHostName",
      "type": "text",
      "description": "Kafka Consumer Host Name",
      "fast": false,
      "fieldnorms": false,
      "indexed": true,
      "record": "basic",
      "stored": true,
      "tokenizer": "default"
    },
    {
      "name": "kafkaTopic",
      "type": "text",
      "description": "Kafka Topic",
      "fast": false,
      "fieldnorms": false,
      "indexed": true,
      "record": "basic",
      "stored": true,
      "tokenizer": "default"
    },
    {
      "name": "kafkaPartition",
      "type": "i64",
      "coerce": true,
      "description": "Kafka Partition",
      "fast": false,
      "indexed": true,
      "output_format": "number",
      "stored": true
    },
    {
      "name": "kafkaOffset",
      "type": "i64",
      "coerce": true,
      "description": "Kafka Offset",
      "fast": false,
      "indexed": true,
      "output_format": "number",
      "stored": true
    },
    {
      "name": "kafkaMessageKey",
      "type": "text",
      "description": "Kafka Message Key",
      "fast": false,
      "fieldnorms": false,
      "indexed": true,
      "record": "basic",
      "stored": true,
      "tokenizer": "raw"
    },
    {
      "name": "kafkaConsumingResult",
      "type": "text",
      "description": "Kafka Consuming Result",
      "fast": {
        "normalizer": "raw"
      },
      "fieldnorms": false,
      "indexed": true,
      "record": "basic",
      "stored": true,
      "tokenizer": "default"
    },
    {
      "name": "env",
      "type": "text",
      "description": "represents environmental information as an extension field in the cloudEvents specification",
      "fast": false,
      "fieldnorms": false,
      "indexed": true,
      "record": "basic",
      "stored": true,
      "tokenizer": "raw"
    },
    {
      "name": "region",
      "type": "text",
      "description": "represents region information as an extension field in the cloudEvents specification",
      "fast": {
        "normalizer": "raw"
      },
      "fieldnorms": false,
      "indexed": true,
      "record": "basic",
      "stored": true,
      "tokenizer": "raw"
    },
    {
      "name": "namespace",
      "type": "text",
      "description": "represents namespace information as an extension field in the cloudEvents specification",
      "fast": {
        "normalizer": "raw"
      },
      "fieldnorms": false,
      "indexed": true,
      "record": "basic",
      "stored": true,
      "tokenizer": "raw"
    }
  ],
  "timestamp_field": "time",
  "tag_fields": [
    "region"
  ],
  "partition_key": "namespace",
  "max_num_partitions": 200,
  "index_field_presence": false,
  "store_document_size": false,
  "store_source": false,
  "tokenizers": []
}

Index settings

{
  "commit_timeout_secs": 60,
  "docstore_compression_level": 8,
  "docstore_blocksize": 1000000,
  "split_num_docs_target": 10000000,
  "merge_policy": {
    "type": "stable_log",
    "min_level_num_docs": 100000,
    "merge_factor": 10,
    "max_merge_factor": 12,
    "maturation_period": "2days"
  },
  "resources": {
    "heap_size": "2.0 GB"
  }
}

[rdettai]

#5808 should solve this issue. I'm just making some final verifications before merging it.

[earlbread]

@rdettai Thank you for your prompt response!
I've also been following that PR closely. In the meantime, is there any workaround or alternative approach you’d recommend while we wait for it to be merged?

[fulmicoton-dd]

@earlbread do you have a single topic? If so setting a mcpu for your indexers to 1mcpu should solve the problem too.

[earlbread]

@fulmicoton-dd
Yes, I'm using a single topic.
Thanks for the suggestion, I’ll give it a try.
Just to confirm — by setting 1mcpu, do you mean configuring the cpu_capacity in the node configuration?

[fulmicoton-dd]

yes. It should spawn the same number of pipeline on each of your nodes.

[earlbread]

Thanks! I'll give it a try and let you know how it goes.

[rdettai]

FYI, I tested the scheduler in your setup, and it already gives the distribution you expect (2 pipeline per node), regardless of the cpu_capacity:

    #[test]
    fn test_place_earlbread() {
        let mut problem =
            SchedulingProblem::with_indexer_cpu_capacities((0..16).map(|_| mcpu(8_000)).collect());
        problem.add_source(32, NonZeroU32::new(4_000).unwrap());
        let mut solution = problem.new_solution();
        solution = solve(problem, solution);
        for assignment in solution.indexer_assignments.iter() {
            assert_eq!(assignment.num_shards(0), 2);
        }
    }

Are you sure your Kafka topics are balanced? (I see you are using partitioning in Quickwit, I assume it might be used elsewhere).

(FYI, you can see how pipelines are distributed by looking at the chitchat state with http://localhost:7280/api/developer/debug?node_ids=quickwit-prod-kr-indexer-a-0)

[earlbread]

Thanks for the detailed explanation and the test case.
I checked the Kafka topic, and the partitions do seem to be well balanced.

Partition ID Replicas First Offset Next Offset Message Count
0	15	12823880733	12844733648	20852915
1	54	12777942633	12797690980	19748347
2	46	12818971632	12839405505	20433873
3	62	12812250858	12831524013	19273155
4	23	12806613324	12827927046	21313722
5	31	12785085001	12805247312	20162311
6	12	12784165082	12802293672	18128590
7	53	12836383808	12858143513	21759705
8	41	12786204597	12807357454	21152857
9	65	12782431852	12803591438	21159586
10	24	12796350486	12816924995	20574509
11	36	12792870028	12809779173	16909145
12	14	12818163950	12838763392	20599442
13	56	12760397389	12782554391	22157002
14	42	12804330195	12822049791	17719596
15	63	12800379760	12821058824	20679064
16	46	10819567302	10840607921	21040619
17	62	10863098409	10885033480	21935071
18	24	10832957246	10853723621	20766375
19	36	10854330105	10872299554	17969449
20	12	10837451911	10854848983	17397072
21	53	10881647351	10900845893	19198542
22	41	10853598600	10871515026	17916426
23	65	10833618597	10853851272	20232675
24	21	10831347554	10849486870	18139316
25	35	10847043216	10865595217	18552001
26	14	10860358236	10879640146	19281910
27	56	10830348635	10850295769	19947134
28	42	10859348457	10877522444	18173987
29	63	10856039169	10877676620	21637451
30	26	10849272617	10869483667	20211050
31	32	10854176273	10871451111	17274838

I checked the chitchat debug endpoint, but there are too many task logs, so it's hard to tell if the pipelines are properly distributed. Just in case, I'm attaching the debug logs:

quickwit_debug.json

[rdettai]

You have (at least) another index configured (access_log_v1_to_karrotgateway), that's why your plan is not like the test. #5808 should definitively help (even though your plan will still be a bit unbalanced afterwards).

[earlbread]

Yes, I'm using multiple indexes.
I'll try setting 1mCPU and also check if I can adjust the pipelines for the other indexes.

Thank you!

[earlbread]

I tried setting cpu_capacity to 1m, but it didn’t have any effect. I guess I’ll have to wait for #5808.

[fulmicoton-dd]

This is odd. Can you share your config as seen from quickwit's UI from an indexer node?
http://localhost:7280/ui/node-info

[earlbread]

@fulmicoton-dd Sure, this is my node info:

{
  "cluster_id": "quickwit-quickwit-prod-kr-a",
  "node_id": "quickwit-prod-kr-a-indexer-0",
  "enabled_services": [
    "indexer"
  ],
  "gossip_listen_addr": "0.0.0.0:7282",
  "grpc_listen_addr": "0.0.0.0:7281",
  "gossip_advertise_addr": "10.128.91.185:7282",
  "grpc_advertise_addr": "10.128.91.185:7281",
  "gossip_interval": {
    "secs": 1,
    "nanos": 0
  },
  "peer_seeds": [
    "quickwit-prod-kr-a-headless"
  ],
  "data_dir_path": "/quickwit/qwdata",
  "metastore_uri": "file:///quickwit/qwdata/indexes#polling_interval=30s",
  "default_index_root_uri": "s3://prod-kr-quickwit-index/indexes",
  "rest_config": {
    "listen_addr": "0.0.0.0:7280",
    "cors_allow_origins": [],
    "extra_headers": {},
    "tls": null
  },
  "grpc_config": {
    "max_message_size": "21.0 MB",
    "tls": null
  },
  "storage_configs": {
    "s3": {
      "access_key_id": null,
      "secret_access_key": null,
      "region": "ap-northeast-2",
      "endpoint": "https://s3.ap-northeast-2.amazonaws.com",
      "force_path_style_access": false,
      "disable_multi_object_delete": false,
      "disable_multipart_upload": false
    }
  },
  "metastore_configs": {},
  "indexer_config": {
    "split_store_max_num_bytes": "107.4 GB",
    "split_store_max_num_splits": 1000,
    "max_concurrent_split_uploads": 12,
    "max_merge_write_throughput": null,
    "merge_concurrency": 5,
    "enable_otlp_endpoint": true,
    "enable_cooperative_indexing": false,
    "cpu_capacity": "1m"
  },
  "searcher_config": {
    "aggregation_memory_limit": "500.0 MB",
    "aggregation_bucket_limit": 65000,
    "fast_field_cache_capacity": "1000.0 MB",
    "split_footer_cache_capacity": "500.0 MB",
    "partial_request_cache_capacity": "64.0 MB",
    "max_num_concurrent_split_searches": 100,
    "max_num_concurrent_split_streams": 100,
    "request_timeout_secs": 30,
    "warmup_memory_budget": "100.0 GB",
    "warmup_single_split_initial_allocation": "1000.0 MB"
  },
  "ingest_api_config": {
    "max_queue_memory_usage": "2.1 GB",
    "max_queue_disk_usage": "4.3 GB",
    "replication_factor": 1,
    "content_length_limit": "10.5 MB",
    "shard_throughput_limit": "5.2 MB",
    "shard_burst_limit": "52.4 MB",
    "shard_scale_up_factor": 1.5
  },
  "jaeger_config": {
    "enable_endpoint": true,
    "lookback_period_hours": 72,
    "max_trace_duration_secs": 3600,
    "max_fetch_spans": 10000
  }
}

[fulmicoton-dd]

ok and you are seeing the pipelines not being scheduled on all nodes? I have no explanation for that :-/

[earlbread]

Yes, I expect the pipelines to be evenly distributed across all indexer nodes, but that doesn’t seem to be happening.