Failed to reload the collections after restarting the milvus cluster.

[tianshihan818]

@yhmo @xiaofan-luan Hello!

I am inserting batch data continually, while the milvus cluster crashed suddenly. I cheked the k8s events and the pods logs that time. It reports some nodes have untolerated taint {node.kubernetes.io/disk-pressure: } so k8s evicted and killed some pods in milvus, and because most of the nodes have this taint label, the above pods cannot be scheduled and restarted successfully, then the whole milvus cluster almost crashed.

The thing I am sure is the disk space is enough, so I guess the disk pressure is mainly more from high disk i/o rate, may dues to the huge amount of inserting logs & indexing files persistence and loading segments operation?

And the weird thing is, if I restart the milvus cluster (by helm, using commands like helm uninstall {milvus_instance_name} and helm install {milvus_instance_name} zilliztech/milvus -f ./values.yaml), the new milvus cluster will restart normally and the nodes are healthy without any taints at first. Then the milvus will loading the collections automatically that were loaded before crashed (collection A with 1 million entites, collection B with 300 million entites), but it fails and leads to the same crash scene above.

I have tried 5 times, it always run into crashing. Finally, I tried to release the loading for these collections as soon as the new milvus restarted. Then I create a new collection, so I can insert the data again with this new collection.

I have some questions:

1. Does the collection in milvus has some limit on their entites amounts?
   I tried an inserting test on the low dimension (such as 64-dim) vectors before, one collection is able to contain more than 2 billion entites. I checked the limitation of milvus from your documents, it said no entites limits within collections. But I am not sure about the high dimension (such as 768-dim) scene.

2. Will the loading operation at one time cause the high disk pressure?
   I checked the working pipeline of the querynode, I know it is in charge of the loading collections according to the segments batch. Suppose for a large scale collection like 300 million, it might contains 700+ flushed segments with each about 400k rows. Reload this collection at one time would lead to a high disk reading pressure, I think. Is there any better way for this?

3. Does more shards increase the disk usage?
   My target is inserting 1+ billion 768-dim vectors, so at fisrt I create a collection with 8 shards and 4 datanodes (the milvus resources scale is deployed according to the Milvus sizing tool). By the way, to reduce the memory usage, I select the IVF_SQ8 indexing type. From what I know, more shards means more growing segments, thus more compaction operations. So under the high rate of compactions, will the disk usage exceed the ideal estimation?

Hoping you will give me some advice :) Thank you all so much!

[yhmo]

We didn't limit the maximum number of entities in a collection.
Our QA team has tested 1 billion 768dim vectors in one collection, all the functions work fine. @yanliang567 can confirm.
The test case is a simple collectin with one primary key field and a 768dim vector field. If your collection has more scalar fields, it will require more resources.

Loading process will use multi-threads to load segments. The disk reading pressure might be high.
In the milvus.yaml, we can control the concurrency by these configurations:

common:
  threadCoreCoefficient:
    highPriority: 10 # This parameter specify how many times the number of threads is the number of cores in high priority pool
    middlePriority: 5 # This parameter specify how many times the number of threads is the number of cores in middle priority pool
    lowPriority: 1 # This parameter specify how many times the number of threads is the number of cores in low priority pool

If a querynode's CPU core is 2, there might be 10 * 2 threads to read files parallelly.
So, you can decrease the value of highPriority and middlePriority to 1 to reduce the reading pressure.

Shards define the number of "data-channel" of a collection. Each data-channel consumes data from pulsar/kafka to generate segments. Segments come from the same data-channel can be merged. Segments come from different data-channels cannot be merged.
So, if the collection has 8 shards, the collection has at least 8 segments.

By default, each segment size is 512MB to 1024MB, the number of segments mainly depends on how much data is inserted, slightly affected by shards.
Assume you inserted 100GB segments, segment.maxSize=1024MB.
If shards=1, there might be 1000 segments in the collection.
If shards=8, there might be 1008 segments in the collection.

Compaction has its machinery to manage tasks. I think no much difference for shards=1 or shards=8.

[tianshihan818]

Thanks for your timely reply! It really helps me a lot!

By the way, could you give me some info about the previous 1 billion 768-dim insertion test? such as the deployment scale of milvus, the size of test machines or the k8s cluster, and better if you could give an example of the test scripts. @yanliang567

Or, if in your convenience, could you check my deployment information below?

# The base
6 nodes, each 256cpu/500Gi/3.5T*4

# The deployment method
on k8s, by helm, using milvus v2.4.1

# The main deployment values
dataCoordinator:
  replicas: 1
  resources:
    limits:
      cpu: "1"
      memory: 1Gi

dataNode:
  replicas: 4
  resources:
    limits:
      cpu: 2
      memory: 8Gi

etcd:
  persistence:
    size: 30Gi
  replicaCount: 3
  resources:
    limits:
      cpu: 8
      memory: 16Gi
    requests:
      cpu: 8
      memory: 16Gi

extraConfigFiles:
  user.yaml: |+
    # If I want to control the concurrency of the loading, I should fill in the configurations here.
    # common:
    #   threadCoreCoefficient:
    #     highPriority: 10
    #     middlePriority: 5
    #     lowPriority: 1  

indexCoordinator:
  replicas: 1
  resources:
    limits:
      cpu: "1"
      memory: 1Gi

indexNode:
  replicas: 4
  resources:
    limits:
      cpu: 8
      memory: 16Gi

minio:
  persistence:
    size: 10500Gi
  resources:
    limits:
      cpu: 8
      memory: 32Gi

proxy:
  replicas: 1
  resources:
    limits:
      cpu: 2
      memory: 8Gi

pulsar:
  autorecovery:
    resources:
      requests:
        cpu: 1
        memory: 512Mi
  bookkeeper:
    configData:
      PULSAR_GC: |
        -Dio.netty.leakDetectionLevel=disabled -Dio.netty.recycler.linkCapacity=1024 -XX:+UseG1GC -XX:MaxGCPauseMillis=10 -XX:+ParallelRefProcEnabled -XX:+UnlockExperimentalVMOptions -XX:+DoEscapeAnalysis -XX:ParallelGCThreads=32 -XX:ConcGCThreads=32 -XX:G1NewSizePercent=50 -XX:+DisableExplicitGC -XX:-ResizePLAB -XX:+ExitOnOutOfMemoryError -XX:+PerfDisableSharedMem -XX:+PrintGCDetails
      PULSAR_MEM: |
        -Xms16384M -Xmx16384M -XX:MaxDirectMemorySize=32768M
      nettyMaxFrameSizeBytes: "104867840"
    replicaCount: 3
    resources:
      requests:
        cpu: 8
        memory: 64Gi
    volumes:
      journal:
        name: journal
        size: 1150Gi
      ledgers:
        name: ledgers
        size: 2870Gi
  broker:
    component: broker
    configData:
      PULSAR_GC: |
        -Dio.netty.leakDetectionLevel=disabled -Dio.netty.recycler.linkCapacity=1024 -XX:+ParallelRefProcEnabled -XX:+UnlockExperimentalVMOptions -XX:+DoEscapeAnalysis -XX:ParallelGCThreads=32 -XX:ConcGCThreads=32 -XX:G1NewSizePercent=50 -XX:+DisableExplicitGC -XX:-ResizePLAB -XX:+ExitOnOutOfMemoryError
      PULSAR_MEM: |
        -Xms16384M -Xmx16384M -XX:MaxDirectMemorySize=32768M
      backlogQuotaDefaultLimitGB: "8"
      backlogQuotaDefaultRetentionPolicy: producer_exception
      defaultRetentionSizeInMB: "-1"
      defaultRetentionTimeInMinutes: "10080"
      maxMessageSize: "104857600"
      subscriptionExpirationTimeMinutes: "3"
      ttlDurationDefaultInSeconds: "259200"
    replicaCount: 2
    resources:
      requests:
        cpu: 8
        memory: 72Gi
  maxMessageSize: "5242880"

  proxy:
    configData:
      PULSAR_MEM: |
        -Xms8192M -Xmx8192M -XX:MaxDirectMemorySize=8192M
      httpNumThreads: "100"
    replicaCount: 2
    resources:
      requests:
        cpu: 8
        memory: 18Gi

  zookeeper:
    configData:
      PULSAR_GC: |
        -Dcom.sun.management.jmxremote -Djute.maxbuffer=10485760 -XX:+ParallelRefProcEnabled -XX:+UnlockExperimentalVMOptions -XX:+DoEscapeAnalysis -XX:+DisableExplicitGC -XX:+PerfDisableSharedMem -Dzookeeper.forceSync=no
      PULSAR_MEM: |
        -Xms4096M -Xmx4096M
    resources:
      requests:
        cpu: 4
        memory: 8Gi
    volumes:
      data:
        size: 20Gi

queryCoordinator:
  replicas: 1
  resources:
    limits:
      cpu: "1"
      memory: 1Gi

queryNode:
  replicas: 16
  resources:
    limits:
      cpu: 16
      memory: 128Gi

rootCoordinator:
  replicas: 1
  resources:
    limits:
      cpu: 2
      memory: 4Gi

# My testing steps
1. create a collection (schema: auto-id field + 768-dim float vector field; partition: default; shard: 4/8 both tried)
2. generate random data and insert (single call pymilvus insert API, insert 10000 entites each time)
3. create IVF_SQ8 index (metric type: L2; nlist: default 1024) and then load data after the entites reached about 1 million level
4. Then insert continually, and milvus will create index & load data automatically

[tianshihan818]

Moreover, I found that the RT of single batch insertion increased a lot when there are so many segments merging. Is this normal?

RT is increasing like:

When most segments merged, RT back to normal:

I compute the RT by marking the timestamp before and after the pymilvus insert API calling. I know it is not the real time cost, it seems like the duration from [proxy received the insertion request] to [proxy writed the data into message store (pulsar) and reported task is done].

I am just curious about how to balance the writing rate and the cluster stability.

[yanliang567]

Generally, the Milvus config looks good to me.

1. about the reload failure, please offer the full milvus pods logs for investigation. My guess is that you have 16 querynode pods, but they could share a few underlying servers, which makes disk IO a bottleneck, but usually it only leads to a timeout error instead of crash. So we need completed logs to know what is hanpening.
2. about the RT of inserting, I suggest you use 4 shards(as you have 4 datanodes), insert less than 1000 *4(shards) in one batch. Also if you could offer Milvus mertic for us for investigation.

[tianshihan818]

Thanks for your advice!

I found the reload failure crash is due to the increasing disk usage of the directory /var at the milvus restarting & auto reloading scene, when its usage reached to the threshold, k8s will tags the node with taints node.kubernetes.io/disk-pressure then evict the pods on the node.

I will try to turn on the milvus service monitor, reproduce more tests and grab the metrics, then I will show you the results here later.

[xiaofan-luan]

what disk and index are you using?
For in memory index(like hnsw), only load use disk, and load should be a one time work.
For on disk index, search will read from disk as well.
300M data is not that large and the cluster you are dploying is large enough. 700 segemnts also seems to be reasonable for your data size

[tianshihan818]

Hi! Sorry for my late response.
I use NVMe SSD disk, and IVF_SQ8 index.

And I done several tests, found out these:

1. Every time I create a new collection and keep inserting into it (also create the index and load the collection when the entities arrives about 1 million), the milvus cluster is stable before the colection entities num reached near to 300 million, after that the milvus cluster will crash. I store some pod logs automatically when it crashed (using kubectl logs pod commands), but sorry don't have the metrics data at that moment. I will try to grab some metrics later. The logs seem to show the etcd service is abnormal at that time. Here are the pods logs:
   It might be many files because there are many querynodes, and I stored logs for 3 retry times.
   crash_when_300_million.zip

2. For the reload failure problem, last time I thought k8s tags nodes with taints node.kubernetes.io/disk-pressure is due to high disk i/o pressure. However, I checked this taint description from the k8s document, it said it occurred by the insufficient disk space of the root file system or image file system on the node. So I reproduced this test and found the disk usage of the directory /var will increase a lot during the reload process. Here are some pictures:

• Restart milvus cluster and reload the collection automatically, one node for example, other node is almost the same
  
  

• After reloading for a while, the /var disk usage reached the threshold on some nodes
  

The /var directory seems default for storing the pods logs and containers runtime of k8s. The auto reload process after restarting the milvus cluster at once cause high usage on /var, when the node's /var disk usage reached to 85%, k8s will evict some pods on this node, and if there are not available nodes for scheduling, these pods cannot be recovered, thus milvus crashed.

I wait the milvus cluster recovered back to normal for a time, then I try to load one collection with 300 million entities. By the way, the querynodes config is 16 * 16cpu128Gi, and I decreased the options below threadCoreCoefficient to 1 as @yhmo suggested. The /var disk usage still reached to the threshold. Do you think 215 GB for /var is too small under this scene?

[xiaofan-luan]

问题是这样的，使用IVFPQ索引后，只有量化数据，没有原始数据
所以milvus会加载一份原始数据到内存中，默认目录就是/var目录

如果你不需要原始数据，关闭预热就可以，方式已经给出了。
如果你需要原始数据查询，需要把目录配置对。