datanode疑似内存泄漏，v2.4.17版本，想咨询一下是否有已知的内存问题。

[qiqiqi998]

运行期间，只会周期性的对每个集合进行数据写入，查询和删除，期间数据总量不变。

写入操作如下：
针对20个集合中的10个分区以并发20，每个30分钟执行插入、查询、删除5000条数据的操作。并对数据做校验。

如下是运行4天的datanode的内存情况：

[xiaofan-luan]

可能不是泄漏，只是delete， bloomfilter的数目增多了。
可以尝试pprof看看具体是什么数据结构占了内存么？

[yhmo]

频繁地建表删表，跟表相关的资源不是立即释放的，一般都有滞后性。当你操作的速度大于它内部管理链条传导速度资源释放的速度，那天平就会向坏的一侧倾斜。

[Niceies]

已经没有频繁建表删表（1天一次），但是会有全量的删除数据和插入数据，这个是否也有滞后性？当我停了一晚上后，内存仍然没有下来，看日志也没有什么输出动作，这种是已经坏了必须重启？

[yhmo]

没有oom就放着，没必要自己去重启

[Niceies]

datanode/indexnode 究竟存储什么数据，处理完合并数据和创建索引不应该就不存储数据了么我理解这两个组件

[yhmo]

datanode: 负责segment的持久化，从pulsar/kafka中消费dml数据，把segment的数据往S3/minio上存储；负责segment的合并任务，小segment合并为大segment；负责清除segment中被删除的数据(如果segment中被删除的数据达到一个阈值)；对于有clustering key的表，做clustering，重建segment。
indexnode: 负责segment的索引创建，一旦有新的segment产生，就为其构建索引任务。所谓新的segment产生，并不是只有insert才会有新segment，其他操作比如compaction, clustering, 清除segment中被删数据这些都会产生新的segment

对于比较隐晦的内存问题，你可在issues里创建issue，把pprof的前后对比贴上，把具体的操作流程详细描述，每个组件都有专门的工程师负责。

[Niceies]

我的监控系统曲线也碰到了这个问题，同样的2.4.17版本，如果我这边代码仓库合入这个改动重新编译是否可行（不清楚是否其他合并请求有关联改动），已经测试一段时间了，再升级版本到v2.4.21工作量比较大

[yhmo]

改动只有4行，只是增加了一小段逻辑，在出现error时把出错的task删除，可以pick过去，没有别的关联。
https://github.com/milvus-io/milvus/pull/38686/files

[Niceies]

使用最新版镜像milvusdb/milvus:v2.4.22，设置参数dataNode.clusteringCompaction.memoryBufferRatio: 0.05
dataCoord.segment.maxSize: 512
写入操作如下：
针对20个集合中的10个分区，每10分钟并发执行插入、创建索引、查询、删除分区条数据的操作。每个集合数据25万，整体500万数据
测试脚本如下(环境上没有influxDB，相关代码操作去掉就可以跑)：

# coding=utf-8
"""
milvus查询压测脚本
"""
import random
import time
from datetime import datetime
import numpy as np
from pymilvus import connections, Collection
import multiprocessing
import threading
import string
from faker import Faker
from influxdb_client import InfluxDBClient, Point, WritePrecision
import argparse

# 环境标记
env_tag = "stability_cluster"


#########################内置参数#################################
# InfluxDB 连接参数
url = f"http://10.128.184.208:32555"  # InfluxDB 的 URL
org = "test"  # 您的组织名称
token = "i1Mk4CZFibE39xNlNV0A_XjRCkxdwAZDbJHyndZcH9LMedch8LU1JiStfGI_hldkclOe-R3V7kiaTCRpExFoFw=="
bucket = "test"          # 您的桶名称
pic_format = ["JPEG", "HEIF", "PNG", "GIF", "WEBP", "SVG"]
# 分区信息
partitions = [f"partition_{i}" for i in range(10)]
data_num = 2000  # 每次插入的数量
history_data_num = 25000
lable = "label_insert_test"
delete_expr = f"label == '{lable}'"
#########################内置参数#################################


def generate_random_lowercase_string(length=6):
    # 定义可用字符，只包含小写字母
    characters = string.ascii_lowercase  # 只包含小写字母
    # 随机选择指定长度的字符
    random_string = ''.join(random.choice(characters) for _ in range(length))
    return random_string


def wait_for_data_complete(collection_client, partition_name, target_num, wait_time=60):
    s = 0
    while True:
        count = collection_client.query(expr="", output_fields=["count(*)"], partition_names=[partition_name])
        count = count[0]["count(*)"]
        # print(f"当前分区{partition_name}数据量：{count}")
        if count == target_num:
            return True
        else:
            time.sleep(1)
            s += 1
            if s >= wait_time:
                return False


def insert_vectors(collection_tag, ip):
    # 创建ifluxdb连接
    influxdb_client = InfluxDBClient(url=url, token=token, org=org)

    # 定义集合名称
    alias = generate_random_lowercase_string()
    # 创建milvus连接
    connections.connect(host=ip, port='32085',
                        user='root',  # 替换为您的用户名
                        password='Milvus',  # 替换为您的密码
                        alias=alias
                        )
    if collection_tag > 20 and collection_tag % 20 == 0:
        collection_tag = 20
    elif collection_tag > 20:
        collection_tag = collection_tag % 20
    else:
        collection_tag = collection_tag

    collection_name = f"collection_{collection_tag}"
    collection = Collection(name=collection_name, using=alias)
    # 查询改集合的向量维度，生成随机查询向量
    schema = collection.schema
    c_dim = [field for field in schema.fields if field.name == 'image_vector'][0].params["dim"]

    fake = Faker()
    while True:
        print(f"{collection_name} start at {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
        for j in range(10):
            partition_name = f"partition_{j}"
            try:
                if not wait_for_data_complete(collection, partition_name, history_data_num, wait_time=3):
                    # 插入数据前先删除lable为label_insert_test的数据
                    collection.delete(delete_expr, partition_name=partition_name)

                # 创建一个随机数生成器，用于生成测试向量数据
                rng = np.random.default_rng(seed=15631)

                # 生成随机向量
                vectors = rng.random((data_num, c_dim))
                # 生成文件名、标签、上传时间、大小和格式
                filenames = [f"image_{i}.jpg" for i in range(data_num)]
                labels = [lable]*data_num
                sizes = [1024 * (i + 1) for i in range(data_num)]
                formats = (pic_format*data_num)[:data_num]
                contents = [fake.text(max_nb_chars=200) for i in range(data_num)]
                # 插入数据,一旦有问题就报错停止等待检查

                start_time = time.time()
                collection.insert([vectors, filenames, labels, sizes, formats, contents], partition_name=partition_name)
                duration = time.time() - start_time

                # 成功写入influxdb
                point = Point("wps").tag("env_tag", env_tag).field("value", 1.0).time(datetime.utcnow(),
                                                                                      WritePrecision.NS)
                with influxdb_client.write_api() as write_api:
                    write_api.write(bucket=bucket, org=org, record=point)
                # 写入时延到influxdb
                point = Point("wps_time").tag("env_tag", env_tag).field("value", duration).time(datetime.utcnow(),
                                                                                                WritePrecision.NS)
                with influxdb_client.write_api() as write_api:
                    write_api.write(bucket=bucket, org=org, record=point)

                # 校验数据
                if not wait_for_data_complete(collection, partition_name, history_data_num+data_num):
                    print(f"{datetime.now().strftime('%Y-%m-%d %H:%M:%S')} -- 集合：{collection_name}， 分区：{partition_name}在插入{data_num}条数据后，60秒后，还是没达到预期数据量：{history_data_num+data_num}")

                # 删除lable为label_insert_test的数据
                collection.delete(delete_expr, partition_name=partition_name)

                # 校验数据
                if not wait_for_data_complete(collection, partition_name, history_data_num):
                    print(f"集合：{collection_name}， 分区：{partition_name}在删除{data_num}条数据后，60秒后，还是没达到预期数据量：{history_data_num}")
            except Exception as e:
                print(f"{datetime.now().strftime('%Y-%m-%d %H:%M:%S')} -- 集合：{collection_name}， 分区：{partition_name}插入{data_num}条 报错信息：{str(e)}")
        # 10fenzhong写一次
        time.sleep(600)
        print(f"{collection_name} end at {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")


def main(ip, num_processes):
    processes = []

    # 创建多个进程
    for i in range(num_processes):
        process = multiprocessing.Process(target=insert_vectors, args=(i+1, ip, ))
        processes.append(process)

    # 等待所有进程完成
    for process in processes:
        process.start()

    # 等待所有进程完成
    for process in processes:
        process.join()


if __name__ == "__main__":
    # parser = argparse.ArgumentParser()
    # # 添加位置参数
    # parser.add_argument('param1', type=str, help='第一个参数：IP地址')
    # parser.add_argument('param2', type=int, help='第二个参数：并发数')
    # # 解析参数
    # args = parser.parse_args()
    # # 将参数传递给函数
    # print(f"以并发数{args.param2}向milvus打压，milvus IP地址：{args.param1}")
    # main(args.param1, args.param2)
    main("10.128.184.208", 20)
    # insert_vectors(20, "10.128.184.208")

昨晚跑了3小时，然后停了放了一晚上
datanode：



indexnode：



datanode(Pprof):
2-27 晚上8点半：
datanode-2-27.pb.gz
2-28 早上9点半：
datanode--2-28.pb.gz
indexnode(Pprof)：
2-27 晚上8点半：
indexnode-2-27.pb.gz
2-28 早上9点半：
indexnode-2-28.gz

不太明白这个曲线，可以看到的是在每次完成500万增删后datanode的最低点都会增加几M，停止后会抖上去，indexnode是一直涨，希望能获取原理说明

[axfor]

@yhmo

[yhmo]

@xiaocai2333 唯一怀疑的是index任务数量是否过多，从监控上看也就一千左右，每个任务的信息在内存中最多也就是几KB级别，所以一千个也就最多几十MB内存。

另外，你给的indexnode-2-27.pb.gz indexnode-2-28.gz这两个文件，打开来看，27号和28号的相差无几。
这是27号：

这是28号：

23617.76KB vs 24097.18KB，并无法证明28号比27号使用内存有明显增长。

[Niceies]

go pprof统计不到cgo代码的内存，是否可能是c库konewhere这一块内存使用有增长？因为从监控和实际上我看进程的res指标是对的上的

[yhmo]

目前knowere已知的问题里没有关于memory leak的。你到这里https://github.com/zilliztech/knowhere/issues 开一个issue吧，看knowhere team有没有办法定位问题。

[Niceies]

好的，提到这里关联了：zilliztech/knowhere#1105

[xiaofan-luan]

2.4存在过一些channel泄漏的问题，不过泄漏量非常少，可能也就几M。
频繁创建删除表会比较明显。
另外，很多情况并不是真的泄漏，只是因为内存分配器没有及时把内存返回给用户。
调整一下jemalloc的参数也会有帮助

[Niceies]

最后的一次测试是使用最新的版本v2.4.22来测试的 ,也没有创建删除表。只有重复插入查询删除数据。我在milvus文档中未搜索到jemalloc相关参数配置，我应该如何调整在进行验证？

[xiaofan-luan]

重复插入删除数据，是会有内存上升的，直到compaction才会被清理掉

[Niceies]

上面截图也有，跑了3小时候停了一晚上，内存在某个时间段突然抖上去后平缓了。这种也是正常的吗