mirrors/milvus
0
0 Fork 0
mirror of https://github.com/milvus-io/milvus.git synced 2026-10-29 01:16:06 +08:00
Code Activity
Files
master
milvus/tests/python_client/scale/test_query_node_scale.py
T
JamesxiaofanluanClaude Opus 5
 f929d68ba6 test: clear the ruff baseline in tests/ so the lint gate reflects the PR under review (#52115) 
issue: #52114

Makes `ruff check` and `ruff format --check` pass over all of `tests/`,
so the `Ruff (changed files only)` job stops failing PRs for violations
they did not introduce (most recently #51903, which got 69 errors from
an 8-line change).

Almost all of this is mechanical. The behavioural changes are called out
individually below: one `NameError` fix, three shadowed duplicate
definitions, 13 assertion gaps, and two undefined names that only became
visible once star imports were expanded.

### Layout

Seven reviewable commits:

| commit | what | files |
|---|---|---|
| 1 | `ruff format` | 170 |
| 2 | `ruff check --fix` (safe fixes only, no `--unsafe-fixes`) | 156 |
| 3 | F821 fix + `tests/ruff.toml` | 4 |
| 4 | three shadowed duplicate definitions | 4 |
| 5 | close 13 assertion gaps behind F841 | 7 |
| 6 | expand star imports, re-enable F403/F405/F821 | 33 |
| 7 | enable E402, E712, E722 | 16 |

1745 violations → 0. Only `tests/` is touched.

### The undefined-name bug

`chaos/test_chaos_memory_stress.py`:
`TestMemoryStressReplica.prepare_collection` referenced bare `nb` and
`dim`. Those are class attributes, never module-level names, so the
fixture raised `NameError` the moment it ran — and the line directly
above it already used `self.dim`. Changed to `self.nb` / `self.dim`.
This was the only F821 in `tests/`; **the rule stays enabled**.

### Three shadowed duplicate definitions

Three places define the same name twice in the same scope. Python keeps
the later one, so the earlier definition is dead while still reading as
live code — a test that looks like it exists and never runs.

**`restful_client/testcases/test_vector_operations.py` —
`TestDeleteVector`**

```python
@pytest.mark.L0
class TestDeleteVector(TestBase):          # line 778 — shadowed, never ran
    def test_delete_vector_default(...)    # 2x2 parametrized

@pytest.mark.L1
class TestDeleteVector(TestBase):          # line 838 — wins
    def test_delete_vector_with_invalid_api_key(...)
    def test_delete_vector_with_invalid_collection_name(...)
    def test_delete_vector_with_non_primary_key(...)
```

The second holds only negative cases, and the file already uses
`TestInsertVectorNegative` / `TestSearchVectorNegative`, so this is a
copy whose class name was never updated. Renamed it to
`TestDeleteVectorNegative`.

**This is the one to look at:** it brings back an `L0` case, so 4
parametrizations of `test_delete_vector_default` start running in PR CI.
They have never executed, so if the restful delete path has drifted
since they were written, this is where it surfaces.

**`python_client/testcases/test_compaction.py` —
`test_compact_after_delete`**

Defined twice in `TestCompactionParams`, covering different things: the
shadowed one (L3, `@parametrize("delete_pos", [1, tmp_nb // 2])`) covers
*partial* deletes plus a compaction-plan check; the surviving one (L2)
deletes everything. There is already a separate
`test_compact_after_delete_all`, so partial-delete coverage existed only
in the shadowed case and was not running. Renamed it to
`test_compact_after_delete_partial`; the surviving definition is
untouched.

**`python_client/utils/util_pymilvus.py` — `add_vector_field`**

Defined twice with different signatures:

```python
def add_vector_field(entities, is_normal=False):     # line 455 — shadowed
    ...
    entities.append(field)
    return entities

def add_vector_field(nb, dimension=default_dim):     # line 527 — wins
    field_name = gen_unique_str()
    field = {...}          # built, then discarded
    return field_name      # returns only the name
```

Removed the second — both the broken one and the one that was winning.
Grepped the whole repo: **neither definition has a caller**; every
`add_vector_field` hit is something else
(`test_milvus_client_add_vector_field` test methods, the
`Op.add_vector_field` enum member in `chaos/checker.py`, a `Checker`
method of the same name). Dead-code removal, not an API change.

**Net effect on what executes:** `test_delete_vector_default` ×4
(**L0**) and `test_compact_after_delete_partial` ×2 (L3) start running;
nothing stops running; the 3 negative cases keep running under the new
class name. Computed by parsing each file before and after and resolving
shadowing the way Python does, not by reading the diff.

F811 is now clean across `tests/` with no per-file suppressions.

### Star imports expanded, F403/F405/F821 re-enabled

33 of 307 files used `from utils.util_pymilvus import *` and friends.
Inside those, pyflakes cannot tell a typo from a star-imported name — it
downgrades everything unresolved to F405, and **F821 never fires**. The
`NameError` fixed in commit 3 was only catchable because that file
happens not to use a star import.

The 379 F405 hits were only **58 distinct names**. All are now explicit,
resolved by walking the star modules in source order (later import wins,
matching Python). Names that `util_pymilvus` merely re-exports come from
their real source now — `DataType` from `pymilvus`, `np` from `numpy`,
`log` from `utils.util_log` — rather than being laundered through it.

Turning F821 on immediately caught **two real undefined names**, both in
`rate_limit/test_rate_limit.py` (L3, so it has never run in CI):

```python
for collection_num in range(collectionRateLimit):   # no such name
#                          ^ the parametrize arg is `collection_rate_limit`, and it is
#                            a string ("10"/"500"), so range() needs int().
#                            The same test uses the correct name 30 lines below.

mil.uninstall(release_name, namespace=NAMESPACE)    # no such name
#  ^ MilvusOperator.uninstall has exactly this signature, and other suites do the
#    same under `mic`/`milvus_op` — the instantiation was just missing.
```

Both fixed. `test_milvus_client_data_integrity_new.py` needed 15 `noqa`
markers instead: its `prepare_data` fixture injects those names via
`globals()[field] = ...`, which no static analysis can follow. That
block has other problems — a `float is None` check against the builtin,
an always-true `if "is null" or "IS NULL" in expression`, leftover
`binbin_debug1` logging — but rewriting fixture logic does not belong
here; noted in #52116.

### 13 assertion gaps closed

Of ~80 F841 hits, most are harmless scratch values. 13 were a call whose
result was bound and then never checked. Highlights:

- `test_milvus_client_database.py`: two tests apply a non-existent
property key (`data.replica.number`, note `data.` not `database.`),
describe the database, and drop it without checking anything. Their
docstrings say `actual: run successfully` / `nothing changed` — the
intent was to prove nothing changed. Now asserted.
- `test_milvus_client_alias.py`: two assertions were commented out,
**and they were commented out for a reason** — `list_aliases()[0]`
returns a dict, so `assert alias in aliases` would have tested dict
keys. Rewritten against `[0]["aliases"]` (how the rest of the file does
it) and re-enabled. In the skipped `alias_default` test the calls were
also missing the required `collection_name` argument.

The remaining ~65 F841 stay, and the rule stays off.

### E402, E712, E722 also enabled

**E722** (37 sites): mechanical `except:` → `except Exception:`. Checked
the two ways this could change behaviour, both empty: no bare-except
try-block can raise `SystemExit`/`KeyboardInterrupt`, and **none calls
`pytest.skip`/`fail`/`xfail`**. That second one matters — pytest's
`Skipped` derives from `BaseException`, so a bare except silently eats
it while `except Exception` would not. 0 of 37, so behaviour is
unchanged.

**E402** (5 sites): imports that were just in the wrong place — `import
re` sat 70 lines into `check/param_check.py`, and the two
`resource_group` modules imported `rich` after their module-level
constants. Nothing to do with `sys.path`, which is what an earlier
version of this description wrongly claimed.

**E712** (5 sites), gone through one at a time rather than
bulk-rewritten:

| site | value | change |
|---|---|---|
| `test_milvus_client_partition.py` | `has_partition()` | → `is False` —
stricter than `not result`, which would pass on `None` |
| `test_restart.py` | JSON status flag | → `is True` |
| `test_restart.py` | local `break_flag` | → plain truth test, it is a
real bool |
| `test_query.py` | wrapper `check_result` | → `is True` |
| `test_milvus_client_ttl.py` | `hit.get(bool_field)` | **kept as `==
False` + noqa** |

That last one is the reason this rule cannot be bulk-fixed: `.get()`
returns `None` for a missing field, so `not ...` would pass exactly
where the assertion should fail, and `is False` would break on a numpy
bool.

### What ruff.toml still disables

| rule | n | reason |
|---|---|---|
| F841 | ~65 | tests deliberately keep call results; the 13 that were
genuine assertion gaps are fixed above |
| UP031 | 95 | pure style rewrite that catches no bugs, and 27 sites use
`% (...)` tuples where it is not equivalent |

Everything else pyflakes/pycodestyle checks is on, including `F401`,
`F811` and `F821`, so new code cannot introduce dead, duplicate or
undefined names. Tracked in #52116.

### Verification

Not just "ruff is green" — the mechanical passes were checked for
semantic drift:

- **Format pass:** parsed all 307 files before and after and compared
ASTs with string constants normalized away → **zero structural
differences**. The only AST-visible deltas are 146 string-constant
changes, every one whitespace-only (docstring indentation / trailing
whitespace); zero non-whitespace string changes.
- **Import removals:** 142 imports were dropped. For each, checked
whether the bound name still appears anywhere in its file. 115 have no
remaining reference at all. The 27 that still matched textually were
each read by hand and are all comments (`# pdb.set_trace()`, the
commented-out checker dict in `test_load_with_checker.py`), docstring
prose (`"input random collection name"`, `"Tuple[List[Dict]]"`), string
literals (`"Collection ID"`, the `"config"` dict key), a different
binding (`from time import sleep`), or attribute access through another
module (`ct.CaseLabel`, `np.random`). **No real reference was removed.**
- **Python version:** the pyupgrade rewrites emit `datetime.UTC` (3.11+)
and `X | None` (3.10+). `tests/.python-version` is 3.12,
`tests/scripts/python-env.sh` defaults to 3.12, and
`tests/docker/Dockerfile` is `python:3.12-bullseye`.
- **`compileall`** over all 307 files under Python 3.12 passes.
- No external reference to `TestDeleteVector`,
`test_compact_after_delete`, or `add_vector_field` exists anywhere in
the repo (`*.py`, `*.sh`, `*.yaml`, `*.json`).

### Conflict surface

Of the 100 currently-open PRs, 9 touch `tests/*.py` and only **2**
overlap with this change: #51903 (3 of its 5 files) and #51856 (1 file).
The other 7 touch files that were already format-clean. Worth merging
promptly to keep that number small.

`.pre-commit-config.yaml` needs no change — it already runs with
`--config=tests/ruff.toml`, so local hooks and CI stay in sync.

---------

Signed-off-by: xiaofanluan <xf@hjjaq.com>
Co-authored-by: xiaofanluan <xf@hjjaq.com>
Co-authored-by: Claude Opus 5 (1M context) <noreply@anthropic.com>
2026-08-03 14:54:18 +08:00

379 lines
16 KiB
Python
Raw Permalink Blame History

import random
import threading
import time
import pytest
from base.collection_wrapper import ApiCollectionWrapper
from base.utility_wrapper import ApiUtilityWrapper
from common import common_func as cf
from common import common_type as ct
from common.common_type import CaseLabel, CheckTasks
from common.milvus_sys import MilvusSys
from customize.milvus_operator import MilvusOperator
from pymilvus import Index, MilvusException, connections
from scale import constants, scale_common
from utils.util_k8s import read_pod_log, wait_pods_ready
from utils.util_log import test_log as log
from utils.util_pymilvus import get_latest_tag
from utils.wrapper import counter
nb = 10000
default_index_params = {"index_type": "IVF_SQ8", "metric_type": "L2", "params": {"nlist": 64}}
def verify_load_balance(c_name, host, port=19530):
"""
verify load balance is available after scale
"""
connections.connect("default", host=host, port=port)
# verify load balance
utility_w = ApiUtilityWrapper()
collection_w = ApiCollectionWrapper()
collection_w.init_collection(c_name)
ms = MilvusSys()
res, _ = utility_w.get_query_segment_info(collection_w.name)
log.debug(res)
segment_distribution = cf.get_segment_distribution(res)
all_querynodes = [node["identifier"] for node in ms.query_nodes]
assert len(all_querynodes) > 1
all_querynodes = sorted(
all_querynodes,
key=lambda x: len(segment_distribution[x]["sealed"]) if x in segment_distribution else 0,
reverse=True,
)
log.debug(all_querynodes)
src_node_id = all_querynodes[0]
des_node_ids = all_querynodes[1:]
sealed_segment_ids = segment_distribution[src_node_id]["sealed"]
# load balance
utility_w.load_balance(collection_w.name, src_node_id, des_node_ids, sealed_segment_ids)
# get segments distribution after load balance
res, _ = utility_w.get_query_segment_info(collection_w.name)
log.debug(res)
segment_distribution = cf.get_segment_distribution(res)
sealed_segment_ids_after_load_banalce = segment_distribution[src_node_id]["sealed"]
# assert src node has no sealed segments
assert sealed_segment_ids_after_load_banalce == []
des_sealed_segment_ids = []
for des_node_id in des_node_ids:
des_sealed_segment_ids += segment_distribution[des_node_id]["sealed"]
# assert sealed_segment_ids is subset of des_sealed_segment_ids
assert set(sealed_segment_ids).issubset(des_sealed_segment_ids)
@pytest.mark.tags(CaseLabel.L3)
class TestQueryNodeScale:
@pytest.mark.tags(CaseLabel.L3)
def test_scale_query_node(self, host):
"""
target: test scale queryNode
method: 1.deploy milvus cluster with 1 queryNode
2.prepare work (connect, create, insert, index and load)
3.continuously search (daemon thread)
4.expand queryNode from 2 to 5
5.continuously insert new data (daemon thread)
6.shrink queryNode from 5 to 3
expected: Verify milvus remains healthy and search successfully during scale
"""
release_name = "scale-query"
image_tag = get_latest_tag()
image = f"{constants.IMAGE_REPOSITORY}:{image_tag}"
query_config = {
"metadata.namespace": constants.NAMESPACE,
"spec.mode": "cluster",
"metadata.name": release_name,
"spec.components.image": image,
"spec.components.proxy.serviceType": "LoadBalancer",
"spec.components.queryNode.replicas": 1,
"spec.config.common.retentionDuration": 60,
}
mic = MilvusOperator()
mic.install(query_config)
if mic.wait_for_healthy(release_name, constants.NAMESPACE, timeout=1800):
host = mic.endpoint(release_name, constants.NAMESPACE).split(":")[0]
else:
raise MilvusException(message="Milvus healthy timeout 1800s")
try:
# connect
connections.add_connection(default={"host": host, "port": 19530})
connections.connect(alias="default")
# create
c_name = cf.gen_unique_str("scale_query")
# c_name = 'scale_query_DymS7kI4'
collection_w = ApiCollectionWrapper()
utility_w = ApiUtilityWrapper()
collection_w.init_collection(name=c_name, schema=cf.gen_default_collection_schema())
# insert two segments
for i in range(30):
df = cf.gen_default_dataframe_data(nb)
collection_w.insert(df)
log.debug(collection_w.num_entities)
# create index
collection_w.create_index(ct.default_float_vec_field_name, default_index_params, timeout=60)
assert collection_w.has_index()[0]
assert collection_w.index()[0] == Index(
collection_w.collection, ct.default_float_vec_field_name, default_index_params
)
# load
collection_w.load()
# scale queryNode to 5
mic.upgrade(release_name, {"spec.components.queryNode.replicas": 5}, constants.NAMESPACE)
@counter
def do_search():
"""do search"""
search_res, is_succ = collection_w.search(
cf.gen_vectors(1, ct.default_dim),
ct.default_float_vec_field_name,
ct.default_search_params,
ct.default_limit,
check_task=CheckTasks.check_nothing,
)
assert len(search_res) == 1
return search_res, is_succ
def loop_search():
"""continuously search"""
while True:
do_search()
threading.Thread(target=loop_search, args=(), daemon=True).start()
# wait new QN running, continuously insert
mic.wait_for_healthy(release_name, constants.NAMESPACE)
wait_pods_ready(constants.NAMESPACE, f"app.kubernetes.io/instance={release_name}")
# verify load balance
verify_load_balance(c_name, host=host)
@counter
def do_insert():
"""do insert"""
return collection_w.insert(cf.gen_default_dataframe_data(1000), check_task=CheckTasks.check_nothing)
def loop_insert():
"""loop insert"""
while True:
do_insert()
threading.Thread(target=loop_insert, args=(), daemon=True).start()
log.debug(collection_w.num_entities)
time.sleep(20)
log.debug("Expand querynode test finished")
mic.upgrade(release_name, {"spec.components.queryNode.replicas": 3}, constants.NAMESPACE)
mic.wait_for_healthy(release_name, constants.NAMESPACE)
wait_pods_ready(constants.NAMESPACE, f"app.kubernetes.io/instance={release_name}")
log.debug(collection_w.num_entities)
time.sleep(60)
scale_common.check_succ_rate(do_search)
scale_common.check_succ_rate(do_insert)
log.debug("Shrink querynode test finished")
except Exception as e:
raise Exception(str(e))
finally:
label = f"app.kubernetes.io/instance={release_name}"
log.info("Start to export milvus pod logs")
read_pod_log(namespace=constants.NAMESPACE, label_selector=label, release_name=release_name)
mic.uninstall(release_name, namespace=constants.NAMESPACE)
def test_scale_query_node_replicas(self):
"""
target: test scale out querynode when load multi replicas
method: 1.Deploy cluster with 5 querynodes
2.Create collection with 2 shards
3.Insert 10 segments and flushed
4.Load collection with 2 replicas
5.Scale out querynode from 5 to 6 while search and insert growing data
expected: Verify search succ rate is 100%
"""
release_name = "scale-replica"
image_tag = get_latest_tag()
image = f"{constants.IMAGE_REPOSITORY}:{image_tag}"
query_config = {
"metadata.namespace": constants.NAMESPACE,
"metadata.name": release_name,
"spec.mode": "cluster",
"spec.components.image": image,
"spec.components.proxy.serviceType": "LoadBalancer",
"spec.components.queryNode.replicas": 5,
"spec.config.common.retentionDuration": 60,
}
mic = MilvusOperator()
mic.install(query_config)
if mic.wait_for_healthy(release_name, constants.NAMESPACE, timeout=1800):
host = mic.endpoint(release_name, constants.NAMESPACE).split(":")[0]
else:
raise MilvusException(message="Milvus healthy timeout 1800s")
try:
scale_querynode = random.choice([6, 7, 4, 3])
connections.connect("scale-replica", host=host, port=19530)
collection_w = ApiCollectionWrapper()
collection_w.init_collection(
name=cf.gen_unique_str("scale_out"),
schema=cf.gen_default_collection_schema(),
using="scale-replica",
shards_num=3,
)
# insert 10 sealed segments
for i in range(5):
df = cf.gen_default_dataframe_data(nb=nb, start=i * nb)
collection_w.insert(df)
assert collection_w.num_entities == (i + 1) * nb
collection_w.load(replica_number=2)
@counter
def do_search():
"""do search"""
search_res, is_succ = collection_w.search(
cf.gen_vectors(1, ct.default_dim),
ct.default_float_vec_field_name,
ct.default_search_params,
ct.default_limit,
check_task=CheckTasks.check_nothing,
)
assert len(search_res) == 1
return search_res, is_succ
def loop_search():
"""continuously search"""
while True:
do_search()
threading.Thread(target=loop_search, args=(), daemon=True).start()
# scale out
mic.upgrade(release_name, {"spec.components.queryNode.replicas": scale_querynode}, constants.NAMESPACE)
mic.wait_for_healthy(release_name, constants.NAMESPACE)
wait_pods_ready(constants.NAMESPACE, f"app.kubernetes.io/instance={release_name}")
log.debug("Scale out querynode success")
time.sleep(100)
scale_common.check_succ_rate(do_search)
log.debug("Scale out test finished")
except Exception as e:
raise Exception(str(e))
finally:
label = f"app.kubernetes.io/instance={release_name}"
log.info("Start to export milvus pod logs")
read_pod_log(namespace=constants.NAMESPACE, label_selector=label, release_name=release_name)
mic.uninstall(release_name, namespace=constants.NAMESPACE)
def test_scale_in_query_node_less_than_replicas(self):
"""
target: test scale in cluster and querynode < replica
method: 1.Deploy cluster with 3 querynodes
2.Create and insert data, flush
3.Load collection with 2 replica number
4.Scale in querynode from 3 to 1 and query
5.Scale out querynode from 1 back to 3
expected: Verify search successfully after scale out
"""
release_name = "scale-in-query"
image_tag = get_latest_tag()
image = f"{constants.IMAGE_REPOSITORY}:{image_tag}"
query_config = {
"metadata.namespace": constants.NAMESPACE,
"metadata.name": release_name,
"spec.mode": "cluster",
"spec.components.image": image,
"spec.components.proxy.serviceType": "LoadBalancer",
"spec.components.queryNode.replicas": 2,
"spec.config.common.retentionDuration": 60,
}
mic = MilvusOperator()
mic.install(query_config)
if mic.wait_for_healthy(release_name, constants.NAMESPACE, timeout=1800):
host = mic.endpoint(release_name, constants.NAMESPACE).split(":")[0]
else:
raise MilvusException(message="Milvus healthy timeout 1800s")
try:
# prepare collection
connections.connect("scale-in", host=host, port=19530)
utility_w = ApiUtilityWrapper()
collection_w = ApiCollectionWrapper()
collection_w.init_collection(
name=cf.gen_unique_str("scale_in"), schema=cf.gen_default_collection_schema(), using="scale-in"
)
collection_w.insert(cf.gen_default_dataframe_data())
assert collection_w.num_entities == ct.default_nb
# load multi replicas and search success
collection_w.load(replica_number=2)
search_res, is_succ = collection_w.search(
cf.gen_vectors(1, ct.default_dim),
ct.default_float_vec_field_name,
ct.default_search_params,
ct.default_limit,
)
assert len(search_res[0].ids) == ct.default_limit
log.info("Search successfully after load with 2 replicas")
log.debug(collection_w.get_replicas()[0])
log.debug(utility_w.get_query_segment_info(collection_w.name, using="scale-in"))
# scale in querynode from 2 to 1, less than replica number
log.debug("Scale in querynode from 2 to 1")
mic.upgrade(release_name, {"spec.components.queryNode.replicas": 1}, constants.NAMESPACE)
mic.wait_for_healthy(release_name, constants.NAMESPACE)
wait_pods_ready(constants.NAMESPACE, f"app.kubernetes.io/instance={release_name}")
# search and not assure success
collection_w.search(
cf.gen_vectors(1, ct.default_dim),
ct.default_float_vec_field_name,
ct.default_search_params,
ct.default_limit,
check_task=CheckTasks.check_nothing,
)
log.debug(collection_w.get_replicas(check_task=CheckTasks.check_nothing)[0])
# scale querynode from 1 back to 2
mic.upgrade(release_name, {"spec.components.queryNode.replicas": 2}, constants.NAMESPACE)
mic.wait_for_healthy(release_name, constants.NAMESPACE)
wait_pods_ready(constants.NAMESPACE, f"app.kubernetes.io/instance={release_name}")
# verify search success
collection_w.search(
cf.gen_vectors(1, ct.default_dim),
ct.default_float_vec_field_name,
ct.default_search_params,
ct.default_limit,
)
# Verify replica info is correct
replicas = collection_w.get_replicas()[0]
assert len(replicas.groups) == 2
for group in replicas.groups:
assert len(group.group_nodes) == 1
# Verify loaded segment info is correct
seg_info = utility_w.get_query_segment_info(collection_w.name, using="scale-in")[0]
num_entities = 0
for seg in seg_info:
assert len(seg.nodeIds) == 2
num_entities += seg.num_rows
assert num_entities == ct.default_nb
except Exception as e:
raise Exception(str(e))
finally:
label = f"app.kubernetes.io/instance={release_name}"
log.info("Start to export milvus pod logs")
read_pod_log(namespace=constants.NAMESPACE, label_selector=label, release_name=release_name)
mic.uninstall(release_name, namespace=constants.NAMESPACE)
View Git Blame Copy Permalink