How to create one kafka source, multiple indexes

[YangTao0]

I'm a quickwit newbie。in my case, We have about 50T of data a day。We need to store this data in different indexes according to servername classification to facilitate quick query.
At the beginning we used vector to get data from kafka and then parsed and sent the large Ingest API source，like this:

[sinks.to_quickwit_applogs]
type = "http"
method = "post"
inputs = [ "app_pod_parser_qw","app_vm_parser_qw"]
encoding.codec = "json"
framing.method = "newline_delimited"
uri = "http://127.0.0.1:7280/api/v1/{{ servicename }}/ingest"
batch.max_bytes = 102400

But Ingest API does not support HA, and the performance is not satisfactory。
So We try to use kafka data source，I'm not sure how to configure，We have multiple topics and multiple brokers in different clusters：

version: 0.8
source_id: app_k8s_vm
source_type: kafka
num_pipelines: 4
params:
  topic: app_k8s_vm
  client_params:
    bootstrap.servers: example01.com:9092,example02.com:9092
transform:
  script: |
    .message = downcase(string!(.message))
    .timestamp = to_unix_timestamp(parse_timestamp!(.@timestamp, format: "%+"),"milliseconds")

[fulmicoton]

Creating one kafka source to feed different indexes is not really possible.

painful workaround 1:
Create one source per index and use transform to filter out the message not targetted for your index.
If you have N index, your data will be read from kafka N times instead of 1. This is VERY inefficient as N grows.

workaround 2:
If you have the same document mapping for all indexes and you wanted different indexes for search efficiency only, one thing you can do is have a single index and use partitioning.

[YangTao0]

workaround 2
Yes, we have the same document mapping, 40 billion data a day, search efficiency needs to be improved。
one thing you can do is have a single index and use partitioning.
use partitioning？Is there any relevant documentation?

[fulmicoton]

We don't have a tutorial explaining that, but you can see some explanation in
https://quickwit.io/docs/main-branch/overview/concepts/querying#tag-pruning
and
https://quickwit.io/docs/main-branch/overview/concepts/querying#partitioning

You need to set 3 things.

partition_key
max_num_partitions
tag_fields

[YangTao0]

I read the documentation and modified config:

index_id: "applogs"
index_uri: "s3://applogs-quickwit-1303155120/applogs"
doc_mapping:
  mode: lenient
  store_source: false
  field_mappings:
    - name: timestamp
      type: datetime
      input_formats:
        - "%Y-%m-%d %H:%M:%S.%f"
        - iso8601
        - rfc3339
        - unix_timestamp
      output_format: unix_timestamp_millis
      fast_precision: milliseconds
      fast: true
    - name: servicename
      type: text
      tokenizer: raw
      fast: true
    - name: message
      type: text
      tokenizer: default
      record: position
  timestamp_field: timestamp
  tag_fields: []
  partition_key: servicename
  max_num_partitions: 200

search_settings:
  default_search_fields: [servicename, message]

indexing_settings:
  commit_timeout_secs: 10
  split_num_docs_target: 10000000
  resources:
    heap_size: 2GB

retention:
  period: 5 days
  schedule: daily

Query:

curl 'http://127.0.0.1:7280/api/v1/applogs/search/stream?snippet_fields=corecapi&query=servicename:corecapi+AND+capiproxy-0add5019-479817-1081036&fast_field=timestamp'

we have many topics，so can I create multiple kafka data sources and write to one index? example:

app_k8s_vm.yaml
./quickwit source create --index applogs --source-config app_k8s_vm.yaml
version: 0.8
source_id: app_k8s_vm
source_type: kafka
num_pipelines: 4
params:
  topic: app_k8s_vm
  client_params:
    bootstrap.servers: example01.com:9092,example02.com:9092
transform:
  script: |
    .message = downcase(string!(.message))
    .timestamp = to_unix_timestamp(parse_timestamp!(.@timestamp, format: "%+"),"milliseconds")

app_k8s_pod.yaml
./quickwit source create --index applogs --source-config app_k8s_pod.yaml
version: 0.8
source_id: app_k8s_pod
source_type: kafka
num_pipelines: 4
params:
  topic: app_k8s_pod
  client_params:
    bootstrap.servers: example01.com:9092,example02.com:9092
transform:
  script: |
    .message = downcase(string!(.message))
    .timestamp = to_unix_timestamp(parse_timestamp!(.@timestamp, format: "%+"),"milliseconds")

[fulmicoton]

you need servicename as a tag field.

[fulmicoton]

we have many topics，so can I create multiple kafka data sources and write to one index

Yes... But I don't understand, you have many topics. I thought you have 1 topic for all services.
Do their service overlap?

Can you say more about the number of topics/ the number of services
and the relationship between topic and services?

[YangTao0]

Yes, in fact we only need two topics to solve the problem, but the current situation is like this after I took over. Of course, merging into two topics will be my subsequent work. two topics are necessary ，Because we have two data centers, I need to avoid data transmission across dedicated lines, which are expensive
Currently, we have 4 topics in a data center, which store the logs of all applications(about 300 servernames). They are in the same data format.

[YangTao0]

I get an error：

2024-09-26T13:57:50.022Z ERROR quickwit_actors::spawn_builder: actor-failure cause=Failed to open file for write: 'IoError { io_error: Custom { kind: Other, error: "Directory kill switch was activated." }, filepath: "f4f9281f3ce8442fba71bd0a23b03c6f.term" }'

Caused by:
    IoError 'Custom { kind: Other, error: "Directory kill switch was activated." }' while opening file for write: 'f4f9281f3ce8442fba71bd0a23b03c6f.term'. exit_status=Failure(Failed to open file for write: 'IoError { io_error: Custom { kind: Other, error: "Directory kill switch was activated." }, filepath: "f4f9281f3ce8442fba71bd0a23b03c6f.term" }'

This is my configuration, is there anything wrong with it?

version: 0.8
cluster_id: quickwit-cluster
node_id: quickwit01
listen_address: 0.0.0.0
rest:
   listen_port: 7280
   cors_allow_origins:
     - "*"
grpc:
   max_message_size: 100 MiB
advertise_address: 10.221.7.208
peer_seeds:
   - quickwit01
   - quickwit02
data_dir: /data/qwdata
metastore_uri: postgres://postgres:xxx@127.0.0.1:5432/quickwit_metastore
default_index_root_uri: s3://applogs-quickwit/applogs
storage:
  s3:
     access_key_id: "xxx"
     secret_access_key: "xxx"
     region: "xxx"
     endpoint: "xxx"
     force_path_style_access: ${QW_S3_FORCE_PATH_STYLE_ACCESS:-false}
     disable_multi_object_delete: false
     disable_multipart_upload: false
metastore:
   postgres:
     min_connections: 0
     max_connections: 50
     acquire_connection_timeout: 10s
     idle_connection_timeout: 10min
     max_connection_lifetime: 30min

indexer:
  enable_otlp_endpoint: false
  #enable_otlp_endpoint: ${QW_ENABLE_OTLP_ENDPOINT:-true}
  split_store_max_num_bytes: 140G
  split_store_max_num_splits: 1000
  max_concurrent_split_uploads: 36
  max_merge_write_throughput: 100mb
  merge_concurrency: 16
ingest_api:
   max_queue_memory_usage: 8GiB
   max_queue_disk_usage: 100GiB
searcher:
   fast_field_cache_capacity: 16G
   split_footer_cache_capacity: 500M
   max_num_concurrent_split_streams: 100
   partial_request_cache_capacity: 64M
   max_num_concurrent_split_searches: 100

jaeger:
  enable_endpoint: true
  lookback_period_hours: 24
  max_trace_duration_secs: 600
  max_fetch_spans: 1000