From 3be3748d7fb41e4fcaf850d5b81b1285951fd266 Mon Sep 17 00:00:00 2001 From: Roie Danino Date: Mon, 28 Jul 2025 15:56:43 +0300 Subject: [PATCH 1/7] PYTHON-PROJECT-WIDE: replaced print() calls with proper logging (#512) * PYTHON-PROJECT-WIDE: replaced print() calls with proper logging Signed-off-by: Roie Danino * BENCHMARK/KVBENCH: fixed identations Signed-off-by: Roie Danino * SRC/API/PYTHON: added a configuration file and a centrelized nixl_logging script with support for NIXL_LOG_LEVEL env Signed-off-by: Roie Danino * NIXL_LOGGING: remove trailing spaces, fix identations Signed-off-by: Roie Danino * EXAMPLES/PYTHON: fixed logging imports Signed-off-by: Roie Danino * BENCHMARK/KVBENCH: fixed logging imports Signed-off-by: Roie Danino * BENCHMARK/KVBENCH/COMMANDS: fixed blank lines Signed-off-by: Roie Danino * EXAMPLES/PYTHON: fixed blank lines Signed-off-by: Roie Danino * SRC/API/PYTHON: moved nixl_logging.py to api/python Signed-off-by: Roie Danino * BENCHMARK/KVBENCH: fixed identations and blank lines Signed-off-by: Roie Danino * BENCHMARK/KVBENCH: fixed identations and spaces Signed-off-by: Roie Danino * BENCHMARK/KVBENCH: fixed identations and spaces Signed-off-by: Roie Danino * BENCHMARK/KVBENCH: fixed identations and spaces Signed-off-by: Roie Danino * CONTRIB: install nixl python package in manylinux docker Signed-off-by: Roie Danino * CONTRIB: moved up installing nixl package Signed-off-by: Roie Danino * .GITLAB: added a CI step for checking there are no print() in the code and only logger is being used Signed-off-by: Roie Danino * .GITLAB: added missing copyrights Signed-off-by: Roie Danino * TEST/PYTHON: change debug logs to info, move script to .ci/scripts Signed-off-by: Roie Danino * EAMPLES/PYTHON: fix pre-commit checks Signed-off-by: Roie Danino * TEST/PYTHON: fixed '-' * 40 log lines Signed-off-by: Roie Danino * EXAMPLES/PYTHON: merge log lines Signed-off-by: Roie Danino * CONTRIB/MANYLINUX: removed unnecessary pip install Signed-off-by: Roie Danino * EXAMPLES/PYTHON: fixed formatting Signed-off-by: Roie Danino * CONTRIB: added back the installtion of the nixl pack in manylinux docker Signed-off-by: Roie Danino * CONTRIB: can't assume nixl package was installed in a script within contrib folder Signed-off-by: Roie Danino * CONTRIB: moved import logging below hashlib Signed-off-by: Roie Danino * .CI/SCRIPTS: added port selection script as test_cpp and test_python might run in parallel Signed-off-by: Roie Danino * .CI/SCRIPTS: make server_port.sh executable Signed-off-by: Roie Danino * .CI/SCRIPTS: removed server_port.sh, lets keep it out of this PR Signed-off-by: Roie Danino * BENCHMAKR/KVBENCH: log -> logger for consistency Signed-off-by: Roie Danino * API/PYTHON: simplified logging.py code, and minor fixes Signed-off-by: Roie Danino * SRC/API/PYTHON: replaced format strings with %s formatting in logger lines Signed-off-by: Roie Danino * BENCHMARK/KVBENCH: replaced all logging line f-strings with % formatting (optimization/compatibility) Signed-off-by: Roie Danino --------- Signed-off-by: Roie Danino Co-authored-by: Yossi Itigin (cherry picked from commit 29ae0ad636ef8188b21eb4f9d7cddbe2ae8aa18f) --- .ci/scripts/check_prints.sh | 60 ++++++++++ .github/workflows/python-checks.yml | 19 +++ benchmark/kvbench/commands/nixlbench.py | 1 - benchmark/kvbench/models/model_config.py | 6 +- benchmark/kvbench/runtime/etcd_rt.py | 21 ++-- .../kvbench/test/custom_traffic_perftest.py | 47 +++++--- .../kvbench/test/inference_workload_matgen.py | 22 ++-- .../sequential_custom_traffic_perftest.py | 68 ++++++----- benchmark/kvbench/test/traffic_pattern.py | 3 - .../src/runtime/etcd/test_etcd_runtime.py | 30 ++--- contrib/wheel_add_ucx_plugins.py | 25 ++-- examples/python/blocking_send_recv_example.py | 25 ++-- examples/python/nixl_api_example.py | 60 +++++----- examples/python/nixl_gds_example.py | 41 ++++--- examples/python/partial_md_example.py | 30 ++--- examples/python/query_mem_example.py | 58 ++++----- src/api/python/_api.py | 59 +++++----- src/api/python/logging.py | 110 ++++++++++++++++++ src/api/python/meson.build | 2 + src/api/python/python_logging.ini | 43 +++++++ test/python/desc_perf.py | 8 +- test/python/prep_xfer_perf.py | 40 ++++--- test/python/test_nixl_bindings.py | 45 +++---- 23 files changed, 574 insertions(+), 249 deletions(-) create mode 100755 .ci/scripts/check_prints.sh create mode 100644 .github/workflows/python-checks.yml create mode 100644 src/api/python/logging.py create mode 100644 src/api/python/python_logging.ini diff --git a/.ci/scripts/check_prints.sh b/.ci/scripts/check_prints.sh new file mode 100755 index 0000000000..33d52a754e --- /dev/null +++ b/.ci/scripts/check_prints.sh @@ -0,0 +1,60 @@ +#!/bin/bash +# SPDX-FileCopyrightText: Copyright (c) 2025 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +# Check if a path is provided as an argument +if [ -z "$1" ]; then + echo "Usage: $0 " + echo "Example: $0 ./path/to/python/directory" + exit 1 +fi + +DIR_PATH="$1" + +# Validate that the provided path is a directory +if [ ! -d "$DIR_PATH" ]; then + echo "Error: The provided path '$DIR_PATH' is not a valid directory." + exit 1 +fi + +echo "Checking for BUILT-IN 'print()' calls in Python files within: $DIR_PATH" +echo "---------------------------------------------------------------------" + +found_print=false + +# Find all Python files and process them +while read -r py_file; do + # Use grep to find 'print()' calls with line numbers, then filter out method calls. + # First grep: finds all occurrences of 'print(' with word boundary. + # Second grep: filters out lines where 'print(' is preceded by a dot and optional whitespace. + MATCHES=$(grep -nE '\bprint\s*\(' "$py_file" | grep -vE '\.[[:space:]]*print\s*\(') + + if [ -n "$MATCHES" ]; then + echo "Found built-in 'print()' in: $py_file" + echo "${MATCHES//$'\n'/$'\n' Line }" # Indent and prepend "Line " + echo # Add a blank line for readability + found_print=true + fi +done < <(find "$DIR_PATH" -name "*.py") + +echo "---------------------------------------------------------------------" + +if [ "$found_print" = true ]; then + echo "One or more Python files in '$DIR_PATH' contain built-in 'print()' calls." + exit 1 +else + echo "No built-in 'print()' calls found in any Python files within '$DIR_PATH'." + exit 0 +fi diff --git a/.github/workflows/python-checks.yml b/.github/workflows/python-checks.yml new file mode 100644 index 0000000000..092c2ee8ab --- /dev/null +++ b/.github/workflows/python-checks.yml @@ -0,0 +1,19 @@ +name: Python Checks + +on: [pull_request] + +jobs: + python-checks: + runs-on: ubuntu-latest + strategy: + matrix: + path: + - ./src + - ./test + - ./benchmark + - ./examples/python + steps: + - uses: actions/checkout@v3 + - name: Check for print() calls in ${{ matrix.path }} + run: | + ./.ci/scripts/check_prints.sh ${{ matrix.path }} diff --git a/benchmark/kvbench/commands/nixlbench.py b/benchmark/kvbench/commands/nixlbench.py index 21d55d47f4..cb30f7c208 100644 --- a/benchmark/kvbench/commands/nixlbench.py +++ b/benchmark/kvbench/commands/nixlbench.py @@ -333,7 +333,6 @@ def should_include(name, value, include_defaults=False): for key, value in params.items(): if value is not None: merged_params[key] = value - # print(json.dumps(merged_params)) return merged_params else: # for text format, exclude defaults to keep command concise for name, value in params.items(): diff --git a/benchmark/kvbench/models/model_config.py b/benchmark/kvbench/models/model_config.py index c3b3fb66df..540cddcfd8 100644 --- a/benchmark/kvbench/models/model_config.py +++ b/benchmark/kvbench/models/model_config.py @@ -20,6 +20,10 @@ import yaml # type: ignore +from nixl.logging import get_logger + +logger = get_logger(__name__) + @dataclass class StrategyConfig: @@ -121,7 +125,7 @@ def from_yaml_files(cls, yaml_paths: List[str]) -> "ModelConfig": config_dict = yaml.safe_load(f) config = config.update(config_dict) else: - print(f"Warning: Config file not found: {path}") + logger.warning("Config file not found: %s", path) return config diff --git a/benchmark/kvbench/runtime/etcd_rt.py b/benchmark/kvbench/runtime/etcd_rt.py index 4f3adecdbc..da70c4cb2f 100644 --- a/benchmark/kvbench/runtime/etcd_rt.py +++ b/benchmark/kvbench/runtime/etcd_rt.py @@ -13,7 +13,6 @@ # See the License for the specific language governing permissions and # limitations under the License. -import logging import os import pickle import re @@ -23,9 +22,11 @@ import etcd3 +from nixl.logging import get_logger + from .rt_base import ReduceOp, _RTUtils -log = logging.getLogger(__name__) +logger = get_logger(__name__) def int_to_bytes(val: int) -> bytes: @@ -75,7 +76,7 @@ def __init__( f"Invalid etcd endpoint format: {etcd_endpoints}, expected format is [http://]host[:port]" ) - log.info(f"ETCD client initialized with host {host} & port {port}") + logger.info("ETCD client initialized with host %s & port %d", host, port) try: self.client = etcd3.client(host=host, port=port) @@ -83,7 +84,7 @@ def __init__( raise ValueError(f"Failed to initialize ETCD client: {e}") if self.rank == 0: - log.info(f"Wiping ETCD prefix {self.prefix}") + logger.info("Wiping ETCD prefix %s", self.prefix) self.client.delete_prefix(self.prefix) def destroy_dist(self): @@ -124,7 +125,13 @@ def barrier(self, ranks: Optional[List[int]] = None, timeout_sec=600): ): if timeout_sec and time.time() - start_time > timeout_sec: raise TimeoutError( - f"[Rank {self.rank}] ROOT - Barrier {key} timed out after {timeout_sec} seconds, current value: {self.client.get(key)}, waiting for val={len(ranks)} (i.e all the ranks have entered the barrier), (ranks: {ranks})" + "[Rank %d] ROOT - Barrier %s timed out after %.3f seconds, current value: %s, waiting for val=%d (i.e all the ranks have entered the barrier), (ranks: %s)", + self.rank, + key, + timeout_sec, + self.client.get(key), + len(ranks), + ranks, ) else: my_index = ranks.index(self.rank) @@ -207,7 +214,7 @@ def all_reduce( val = self.client.get(f"{self.prefix}/all_reduce/{dest_rank}")[0] vals.append(pickle.loads(val)) - print(vals) + logger.debug("All reduce values: %s", vals) if op == ReduceOp.SUM: final_val = [sum(col) for col in zip(*vals)] elif op == ReduceOp.AVG: @@ -235,7 +242,7 @@ def _get_group_id(self, ranks: List[int]) -> int: if not os.environ.get("NIXL_ETCD_NAMESPACE"): - log.warning( + logger.warning( "Environment variable NIXL_ETCD_NAMESPACE is not set, using default prefix /nixl/kvbench. " "Note that it can lead to conflicts if multiple instances of KVBench are running. " "To avoid this, set NIXL_ETCD_NAMESPACE to a unique value for each instance of KVBench. " diff --git a/benchmark/kvbench/test/custom_traffic_perftest.py b/benchmark/kvbench/test/custom_traffic_perftest.py index 6a3bcf2cf1..3bc456940d 100644 --- a/benchmark/kvbench/test/custom_traffic_perftest.py +++ b/benchmark/kvbench/test/custom_traffic_perftest.py @@ -13,7 +13,6 @@ # See the License for the specific language governing permissions and # limitations under the License. -import logging import time from test.traffic_pattern import TrafficPattern from typing import Literal, Optional, Tuple @@ -24,8 +23,9 @@ from tabulate import tabulate from nixl._api import nixl_agent +from nixl.logging import get_logger -log = logging.getLogger(__name__) +logger = get_logger(__name__) class NixlHandle: @@ -62,13 +62,20 @@ def __init__( if shards > 1: raise ValueError("Sharding is not supported yet") - log.debug( - f"[Rank {dist_rt.get_rank()}] Initializing NixlBuffer with size {size}, device {device}, shards {shards}, fill_value {fill_value}" + logger.debug( + "[Rank %d] Initializing NixlBuffer with size %d, device %s, shards %d, fill_value %d", + dist_rt.get_rank(), + size, + device, + shards, + fill_value, ) self.buf = torch.full((size,), fill_value, dtype=dtype, device=device) - log.debug( - f"[Rank {dist_rt.get_rank()}] Registering memory for buffer {self.buf}" + logger.debug( + "[Rank %d] Registering memory for buffer %s", + dist_rt.get_rank(), + self.buf, ) self.reg_descs = nixl_agent.get_reg_descs(self.buf) assert ( @@ -137,7 +144,7 @@ def _barrier_tp(self, tp: TrafficPattern, senders_only=True): def _share_md(self) -> None: """Share agent metadata between all ranks. (Need to be run after registering buffers)""" - log.debug(f"[Rank {self.my_rank}] Sharing MD") + logger.debug("[Rank %d] Sharing MD", self.my_rank) md = self.nixl_agent.get_agent_metadata() mds = dist_rt.allgather_obj(md) for other_rank, metadata in enumerate(mds): @@ -202,7 +209,7 @@ def _prepare_tp( send_bufs, recv_bufs = self._get_bufs(tp) - log.debug(f"[Rank {self.my_rank}] Sharing recv buf descs") + logger.debug("[Rank %d] Sharing recv buf descs", self.my_rank) dst_bufs_descs = self._share_recv_buf_descs(recv_bufs) handles: list[NixlHandle] = [] @@ -212,8 +219,12 @@ def _prepare_tp( xfer_desc = self.nixl_agent.get_xfer_descs(buf) - log.debug( - f"[Rank {self.my_rank}] Initializing xfer for {other} - xfer desc: {xfer_desc}, dst buf desc: {dst_bufs_descs[other]}" + logger.debug( + "[Rank %d] Initializing xfer for %d - xfer desc: %s, dst buf desc: %s", + self.my_rank, + other, + xfer_desc, + dst_bufs_descs[other], ) handle = self.nixl_agent.initialize_xfer( "WRITE", @@ -268,11 +279,11 @@ def _wait(self, handles: list[NixlHandle]): handles = pending def _destroy(self, handles: list[NixlHandle]): - log.debug(f"[Rank {self.my_rank}] Releasing XFER handles") + logger.debug("[Rank %d] Releasing XFER handles", self.my_rank) for handle in handles: self.nixl_agent.release_xfer_handle(handle.handle) - log.debug(f"[Rank {self.my_rank}] Removing remote agents") + logger.debug("[Rank %d] Removing remote agents", self.my_rank) for other_rank in range(self.world_size): if other_rank == self.my_rank: continue @@ -281,7 +292,7 @@ def _destroy(self, handles: list[NixlHandle]): self._destroy_buffers() def _destroy_buffers(self): - log.debug(f"[Rank {self.my_rank}] Destroying buffers") + logger.debug("[Rank %d] Destroying buffers", self.my_rank) self.send_buf.destroy() self.recv_buf.destroy() @@ -301,14 +312,14 @@ def _verify_tp( for r, recv_buf in enumerate(recv_bufs): if recv_buf is None: if tp.matrix[r][self.my_rank] > 0: - log.error( + logger.error( f"Rank {self.my_rank} expected {tp.matrix[r][self.my_rank]} bytes from rank {r}, but got 0" ) raise RuntimeError("Buffer verification failed") continue if print_recv_buffers: - log.info(f"Recv buffer {r}:\n{recv_buf.buf}") + logger.info("Recv buffer %d:\n%s", r, recv_buf.buf) # recv_buf has to be filled with the rank of the sender # and its size has to be the same as matrix[r][my_rank] @@ -337,7 +348,7 @@ def run( Returns: Total execution time in seconds """ - log.debug(f"[Rank {self.my_rank}] Running CT perftest") + logger.debug("[Rank %d] Running CT perftest", self.my_rank) self._share_md() handles, send_bufs, recv_bufs = self._prepare_tp(self.traffic_pattern) @@ -377,7 +388,9 @@ def run( total_size_gb, ] ] - print(tabulate(data, headers=headers, floatfmt=".6f")) + logger.info( + f"Performance metrics:\n{tabulate(data, headers=headers, floatfmt='.6f')}" + ) if verify_buffers: self._verify_tp(self.traffic_pattern, recv_bufs, print_recv_buffers) diff --git a/benchmark/kvbench/test/inference_workload_matgen.py b/benchmark/kvbench/test/inference_workload_matgen.py index 42d42a0ebe..dd23647fb7 100644 --- a/benchmark/kvbench/test/inference_workload_matgen.py +++ b/benchmark/kvbench/test/inference_workload_matgen.py @@ -57,6 +57,10 @@ import yaml from tqdm import tqdm +from nixl.logging import get_logger + +logger = get_logger(__name__) + @dataclass class ModelConfig: @@ -190,7 +194,7 @@ def gen_batches( curr_mem = 0 if curr: batches.append(Batch(user_requests=curr)) - print(f"Last batch is incomplete, his size is {len(curr)}") + logger.warning("Last batch is incomplete, with size %d", len(curr)) return batches @@ -268,8 +272,6 @@ def gen_matrix( num_peers = int(num_peers) buf_size = kv_slice_size / num_peers - # print(f"kv_size: {format_size(kv_size)}, kv_slice_size: {format_size(kv_slice_size)}, buf_size: {format_size(buf_size)}, num_peers: {num_peers}") - mat = np.zeros((world_size, world_size)) dst_iter = iter(decode_worker) @@ -389,11 +391,11 @@ def main( decode_workers = reordered - print(f"Prefill workers: {prefill_workers}") - print(f"Decode workers: {decode_workers}") + logger.info("Prefill workers: %s", prefill_workers) + logger.info("Decode workers: %s", decode_workers) batches = gen_batches(num_user_requests, task_config, model_config) - print(f"Generated {len(batches)} batches") + logger.info("Generated %d batches", len(batches)) matrices = gen_matrices_and_compute_time( batches, prefill_workers, @@ -406,7 +408,7 @@ def main( # Save matrices and metadata to files results_dir = results_dir or Path(f"matrices_{world_size}ranks") results_dir = Path(results_dir) - print(f"Saving {len(matrices)} matrices to {results_dir}") + logger.info("Saving %d matrices to %s", len(matrices), results_dir) results_dir.mkdir(parents=True, exist_ok=True) metadata: dict[str, Any] = { @@ -435,7 +437,7 @@ def main( metadata_path = results_dir / "metadata.yaml" with open(metadata_path, "w") as f: yaml.dump(metadata, f) - print(f"Saved metadata to {metadata_path}") + logger.info("Saved metadata to %s", metadata_path) if __name__ == "__main__": @@ -603,10 +605,6 @@ def generate( max_batch_mem=max_batch_mem, ) - # world_size = num_prefill_nodes * prefill_tp + num_decode_nodes * decode_tp - # print(f"World size: {world_size}") - # print(f"Model config: {model_config}") - main( num_user_requests=num_user_requests, task_config=task_config, diff --git a/benchmark/kvbench/test/sequential_custom_traffic_perftest.py b/benchmark/kvbench/test/sequential_custom_traffic_perftest.py index 278ac7476c..27af3da055 100644 --- a/benchmark/kvbench/test/sequential_custom_traffic_perftest.py +++ b/benchmark/kvbench/test/sequential_custom_traffic_perftest.py @@ -16,7 +16,6 @@ """Sequential is different from multi in that every rank processes only one TP at a time, but they can process different ones""" import json -import logging import time from collections import defaultdict from itertools import chain @@ -29,8 +28,9 @@ from tabulate import tabulate from nixl._api import nixl_agent +from nixl.logging import get_logger -log = logging.getLogger(__name__) +logger = get_logger(__name__) class SequentialCTPerftest(CTPerftest): @@ -59,7 +59,7 @@ def __init__( self.n_isolation_iters = n_isolation_iters self.warmup_iters = warmup_iters - log.debug(f"[Rank {self.my_rank}] Initializing Nixl agent") + logger.debug("[Rank %d] Initializing Nixl agent", self.my_rank) self.nixl_agent = nixl_agent(f"{self.my_rank}") for tp in self.traffic_patterns: @@ -71,7 +71,7 @@ def __init__( self.recv_buf_by_mem_type: dict[str, NixlBuffer] = {} def _init_buffers(self): - log.debug(f"[Rank {self.my_rank}] Initializing buffers") + logger.debug("[Rank %d] Initializing buffers", self.my_rank) max_src_by_mem_type = defaultdict(int) max_dst_by_mem_type = defaultdict(int) @@ -98,14 +98,14 @@ def _init_buffers(self): ) def _destroy_buffers(self): - log.debug(f"[Rank {self.my_rank}] Destroying buffers") + logger.debug("[Rank %d] Destroying buffers", self.my_rank) for buf in chain( self.send_buf_by_mem_type.values(), self.recv_buf_by_mem_type.values() ): buf.destroy() def _get_bufs(self, tp: TrafficPattern): - log.debug(f"[Rank {self.my_rank}] Getting buffers for TP {tp.id}") + logger.debug("[Rank %d] Getting buffers for TP %s", self.my_rank, tp.id) send_bufs = [None for _ in range(self.world_size)] recv_bufs = [None for _ in range(self.world_size)] @@ -152,7 +152,7 @@ def run( This method initializes and executes multiple traffic patterns simultaneously, measures their performance, and optionally verifies the results. """ - log.debug(f"[Rank {self.my_rank}] Running sequential CT perftest") + logger.debug("[Rank %d] Running sequential CT perftest", self.my_rank) self._init_buffers() self._share_md() @@ -168,7 +168,7 @@ def run( tp_bufs = [] s = time.time() - log.info(f"[Rank {self.my_rank}] Preparing TPs") + logger.info("[Rank %d] Preparing TPs", self.my_rank) for i, tp in enumerate(self.traffic_patterns): handles, send_bufs, recv_bufs = self._prepare_tp(tp) tp_bufs.append((send_bufs, recv_bufs)) @@ -190,8 +190,9 @@ def run( dist_rt.barrier() # Isolated mode - Measure SOL for every matrix - log.info( - f"[Rank {self.my_rank}] Running isolated benchmark (to measure perf without noise)" + logger.info( + "[Rank %d] Running isolated benchmark (to measure perf without noise)", + self.my_rank, ) my_isolated_tp_latencies: list[float] = [0 for _ in tp_handles] @@ -211,8 +212,13 @@ def run( my_isolated_tp_latencies[tp_ix] += e - t self._barrier_tp(tp) - log.debug( - f"[Rank {self.my_rank}] Ran {self.n_isolation_iters} isolated iters for tp {tp_ix}/{len(tp_handles)}, took {e - t} secs" + logger.debug( + "[Rank %d] Ran %d isolated iters for tp %d/%d, took %.3f secs", + self.my_rank, + self.n_isolation_iters, + tp_ix, + len(tp_handles), + e - t, ) my_isolated_tp_latencies[tp_ix] /= self.n_isolation_iters @@ -230,18 +236,21 @@ def run( if tp_lats: isolated_tp_latencies_ms.append(max(tp_lats) * 1e3) - log.info(f"[Rank {self.my_rank}] Running workload benchmark") + logger.info("[Rank %d] Running workload benchmark", self.my_rank) # Workload mode - Measure perf of the matrices while running the full workload for iter_ix in range(self.n_iters): - log.debug( - f"[Rank {self.my_rank}] Running iteration {iter_ix + 1}/{self.n_iters}" + logger.debug( + "[Rank %d] Running iteration %d/%d", + self.my_rank, + iter_ix + 1, + self.n_iters, ) iter_metadata = results["metadata"]["iters"][iter_ix] tp_starts: list[float | None] = [None] * len(tp_handles) tp_ends: list[float | None] = [None] * len(tp_handles) - log.debug(f"[Rank {self.my_rank}] Warmup done.") + logger.debug("[Rank %d] Warmup done.", self.my_rank) dist_rt.barrier(timeout_sec=None) iter_metadata["start_ts"] = time.time() @@ -256,14 +265,22 @@ def run( time.sleep(tp.sleep_before_launch_sec) # Run TP - log.debug(f"[Rank {self.my_rank}] Running TP {tp_ix}/{len(tp_handles)}") + logger.debug( + "[Rank %d] Running TP %d/%d", + self.my_rank, + tp_ix, + len(tp_handles), + ) tp_start_ts = time.time() self._run_tp(handles, blocking=True) tp_end_ts = time.time() - log.debug( - f"[Rank {self.my_rank}] TP {tp_ix} took {tp_end_ts - tp_start_ts} seconds" + logger.debug( + "[Rank %d] TP %d took %.3f seconds", + self.my_rank, + tp_ix, + tp_end_ts - tp_start_ts, ) tp_starts[tp_ix] = tp_start_ts @@ -316,9 +333,8 @@ def run( ] for i, tp in enumerate(self.traffic_patterns) ] - print( - f"Iteration {iter_ix + 1}/{self.n_iters}\n", - tabulate(data, headers=headers, floatfmt=".3f"), + logger.info( + f"Iteration {iter_ix + 1}/{self.n_iters}\n{tabulate(data, headers=headers, floatfmt='.3f')}" ) if verify_buffers: @@ -357,12 +373,12 @@ def run( results["metadata"]["finished_ts"] = time.time() if json_output_path and self.my_rank == 0: - log.info(f"Saving results to {json_output_path}") + logger.info("Saving results to %s", json_output_path) with open(json_output_path, "w") as f: json.dump(results, f) # Destroy - log.info(f"[Rank {self.my_rank}] Finished run, destroying objects") + logger.info("[Rank %d] Finished run, destroying objects", self.my_rank) self._destroy(handles) def _write_yaml_results( @@ -411,6 +427,6 @@ def _write_yaml_results( try: with open(output_path, "w") as f: yaml.dump(results, f, default_flow_style=False, sort_keys=False) - log.info(f"Results saved to YAML file: {output_path}") + logger.info("Results saved to YAML file: %s", output_path) except Exception as e: - log.error(f"Failed to write YAML results to {output_path}: {e}") + logger.error("Failed to write YAML results to %s: %s", output_path, e) diff --git a/benchmark/kvbench/test/traffic_pattern.py b/benchmark/kvbench/test/traffic_pattern.py index e771e58551..1f612279af 100644 --- a/benchmark/kvbench/test/traffic_pattern.py +++ b/benchmark/kvbench/test/traffic_pattern.py @@ -12,15 +12,12 @@ # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. -import logging from dataclasses import dataclass, field from typing import ClassVar, Literal, Optional import numpy as np import torch -log = logging.getLogger(__name__) - @dataclass class TrafficPattern: diff --git a/benchmark/nixlbench/src/runtime/etcd/test_etcd_runtime.py b/benchmark/nixlbench/src/runtime/etcd/test_etcd_runtime.py index d869dd99b4..20ffaaa920 100755 --- a/benchmark/nixlbench/src/runtime/etcd/test_etcd_runtime.py +++ b/benchmark/nixlbench/src/runtime/etcd/test_etcd_runtime.py @@ -22,15 +22,19 @@ import os import sys +from nixl.logging import get_logger + # Add the kvbench runtime path sys.path.insert(0, os.path.join(os.path.dirname(__file__), "../../../kvbench/runtime")) +logger = get_logger(__name__) + try: from etcd_rt import _EtcdDistUtils def test_basic_functionality(): """Test basic rank and size functionality""" - print("Testing basic functionality...") + logger.info("Testing basic functionality...") # Initialize runtime - modify size based on how many processes you're running runtime = _EtcdDistUtils(etcd_endpoints="http://localhost:2379", size=2) @@ -38,37 +42,37 @@ def test_basic_functionality(): rank = runtime.get_rank() world_size = runtime.get_world_size() - print(f"Rank: {rank}, World Size: {world_size}") + logger.info("Rank: %d, World Size: %d", rank, world_size) # Test barrier - print(f"Rank {rank}: Before barrier") + logger.info("Rank %d: Before barrier", rank) runtime.barrier() - print(f"Rank {rank}: After barrier") + logger.info("Rank %d: After barrier", rank) # Test allgather my_data = {"rank": rank, "message": f"Hello from rank {rank}"} - print(f"Rank {rank}: Gathering data...") + logger.info("Rank %d: Gathering data...", rank) try: all_data = runtime.allgather_obj(my_data) - print(f"Rank {rank}: Gathered data from all ranks:") + logger.info("Rank %d: Gathered data from all ranks:", rank) for i, data in enumerate(all_data): - print(f" Rank {i}: {data}") + logger.info(" Rank %d: %s", i, data) except Exception as e: - print(f"Rank {rank}: Allgather failed: {e}") + logger.error("Rank %d: Allgather failed: %s", rank, e) # Test barrier again runtime.barrier() - print(f"Rank {rank}: Test completed successfully!") + logger.info("Rank %d: Test completed successfully!", rank) if __name__ == "__main__": test_basic_functionality() except ImportError as e: - print(f"Import error: {e}") - print("Make sure the etcd_runtime module is built and accessible") - print("Also ensure the etcd server is running at http://localhost:2379") + logger.error("Import error: %s", e) + logger.error("Make sure the etcd_runtime module is built and accessible") + logger.error("Also ensure the etcd server is running at http://localhost:2379") sys.exit(1) except Exception as e: - print(f"Runtime error: {e}") + logger.error("Runtime error: %s", e) sys.exit(1) diff --git a/contrib/wheel_add_ucx_plugins.py b/contrib/wheel_add_ucx_plugins.py index 124ff81ce3..d9bd8d0698 100755 --- a/contrib/wheel_add_ucx_plugins.py +++ b/contrib/wheel_add_ucx_plugins.py @@ -19,11 +19,14 @@ import base64 import csv import hashlib +import logging import os import shutil import tempfile import zipfile +logger = logging.getLogger(__name__) + def extract_wheel(wheel_path): """ @@ -32,7 +35,7 @@ def extract_wheel(wheel_path): Path to the temporary directory. The caller is responsible for cleaning up the directory. """ temp_dir = tempfile.mkdtemp() - print(f"Extracting wheel {wheel_path} to {temp_dir}") + logger.info("Extracting wheel %s to %s", wheel_path, temp_dir) with zipfile.ZipFile(wheel_path, "r") as zip_ref: zip_ref.extractall(temp_dir) return temp_dir @@ -82,7 +85,7 @@ def create_wheel(wheel_path, temp_dir): """ Create a wheel from a temporary directory. """ - print(f"Creating wheel {wheel_path} from {temp_dir}") + logger.info("Creating wheel %s from %s", wheel_path, temp_dir) update_wheel_record_file(temp_dir) with zipfile.ZipFile( wheel_path, "w", compression=zipfile.ZIP_DEFLATED, compresslevel=9 @@ -190,7 +193,7 @@ def add_plugins(wheel_path, sys_plugins_dir, install_dirname): if not os.path.exists(pkg_libs_dir): raise FileNotFoundError(f"nixl.libs directory not found in wheel: {wheel_path}") - print("Listing existing libs:") + logger.debug("Listing existing libs:") name_map = get_repaired_lib_name_map(pkg_libs_dir) # Ensure that all of them in name_map have RPATH set to $ORIGIN @@ -203,20 +206,20 @@ def add_plugins(wheel_path, sys_plugins_dir, install_dirname): rpath = "$ORIGIN" else: rpath = "$ORIGIN:" + rpath - print(f"Setting rpath for {fpath} to {rpath}") + logger.debug("Setting rpath for %s to %s", fpath, rpath) ret = os.system(f"patchelf --set-rpath '{rpath}' {fpath}") if ret != 0: raise RuntimeError(f"Failed to set rpath for {fpath}") pkg_plugins_dir = os.path.join(pkg_libs_dir, install_dirname) - print(f"Copying plugins from {sys_plugins_dir} to {pkg_plugins_dir}") + logger.debug("Copying plugins from %s to %s", sys_plugins_dir, pkg_plugins_dir) copied_files = copytree(sys_plugins_dir, pkg_plugins_dir) if not copied_files: raise RuntimeError(f"No plugins found in {sys_plugins_dir}") # Patch all libs to load plugin deps from the wheel for fname in copied_files: - print(f"Patching {fname}") + logger.debug("Patching %s", fname) fpath = os.path.join(pkg_plugins_dir, fname) if os.path.isfile(fpath) and ".so" in fname: rpath = os.popen(f"patchelf --print-rpath {fpath}").read().strip() @@ -224,7 +227,7 @@ def add_plugins(wheel_path, sys_plugins_dir, install_dirname): rpath = "$ORIGIN/..:$ORIGIN" else: rpath = "$ORIGIN/..:$ORIGIN:" + rpath - print(f"Setting rpath for {fpath} to {rpath}") + logger.debug("Setting rpath for %s to %s", fpath, rpath) ret = os.system(f"patchelf --set-rpath '{rpath}' {fpath}") if ret != 0: raise RuntimeError(f"Failed to set rpath for {fpath}") @@ -234,7 +237,9 @@ def add_plugins(wheel_path, sys_plugins_dir, install_dirname): base_name = libname.split(".")[0] if base_name in name_map: packaged_name = name_map[base_name] - print(f"Replacing {libname} with {packaged_name} in {fpath}") + logger.debug( + "Replacing %s with %s in %s", libname, packaged_name, fpath + ) ret = os.system( f"patchelf --replace-needed {libname} {packaged_name} {fpath}" ) @@ -243,7 +248,7 @@ def add_plugins(wheel_path, sys_plugins_dir, install_dirname): f"Failed to replace {libname} with {packaged_name} in {fpath}" ) # Check that there is no breakage introduced in the patched lib - print(f"Checking that {fpath} loads") + logger.debug("Checking that %s loads", fpath) original_deps = get_lib_deps(os.path.join(sys_plugins_dir, fname)) for libname, libpath in get_lib_deps(fpath).items(): if libpath is None: @@ -255,7 +260,7 @@ def add_plugins(wheel_path, sys_plugins_dir, install_dirname): create_wheel(wheel_path, temp_dir) shutil.rmtree(temp_dir) - print(f"Added plugins to wheel: {wheel_path}") + logger.info("Added plugins to wheel: %s", wheel_path) def main(): diff --git a/examples/python/blocking_send_recv_example.py b/examples/python/blocking_send_recv_example.py index 878d145ed9..da4064c535 100755 --- a/examples/python/blocking_send_recv_example.py +++ b/examples/python/blocking_send_recv_example.py @@ -20,6 +20,9 @@ import torch from nixl._api import nixl_agent, nixl_agent_config +from nixl.logging import get_logger + +logger = get_logger(__name__) def parse_args(): @@ -58,11 +61,11 @@ def parse_args(): else: tensors = [torch.zeros(10, dtype=torch.float32) for _ in range(2)] - print(f"{args.mode} Tensors: {tensors}") + logger.info("Running test with %s tensors in mode %s", tensors, args.mode) reg_descs = agent.register_memory(tensors) if not reg_descs: # Same as reg_descs if successful - print("Memory registration failed.") + logger.error("Memory registration failed.") exit() # Target code @@ -78,7 +81,7 @@ def parse_args(): agent.send_notif("initiator", target_desc_str) - print("Waiting for transfer") + logger.info("Waiting for transfer") # Waiting for transfer # For now the notification is just UUID, could be any python bytes. @@ -88,7 +91,7 @@ def parse_args(): continue # Initiator code else: - print("Initiator sending to " + args.ip) + logger.info("Initiator sending to %s", args.ip) agent.fetch_remote_metadata("target", args.ip, args.port) agent.send_local_metadata(args.ip, args.port) @@ -105,24 +108,24 @@ def parse_args(): while not ready: ready = agent.check_remote_metadata("target") - print("Ready for transfer") + logger.info("Ready for transfer") xfer_handle = agent.initialize_xfer( "READ", initiator_descs, target_descs, "target", "UUID" ) if not xfer_handle: - print("Creating transfer failed.") + logger.error("Creating transfer failed.") exit() state = agent.transfer(xfer_handle) if state == "ERR": - print("Posting transfer failed.") + logger.error("Posting transfer failed.") exit() while True: state = agent.check_xfer_state(xfer_handle) if state == "ERR": - print("Transfer got to Error state.") + logger.error("Transfer got to Error state.") exit() elif state == "DONE": break @@ -130,9 +133,9 @@ def parse_args(): # Verify data after read for i, tensor in enumerate(tensors): if not torch.allclose(tensor, torch.ones(10)): - print(f"Data verification failed for tensor {i}.") + logger.error("Data verification failed for tensor %d.", i) exit() - print(f"{args.mode} Data verification passed - {tensors}") + logger.info("%s Data verification passed", args.mode) if args.mode != "target": agent.remove_remote_agent("target") @@ -141,4 +144,4 @@ def parse_args(): agent.deregister_memory(reg_descs) - print("Test Complete.") + logger.info("Test Complete.") diff --git a/examples/python/nixl_api_example.py b/examples/python/nixl_api_example.py index 316e84378d..67b3f65367 100755 --- a/examples/python/nixl_api_example.py +++ b/examples/python/nixl_api_example.py @@ -22,13 +22,17 @@ import nixl._utils as nixl_utils from nixl._api import nixl_agent, nixl_agent_config +from nixl.logging import get_logger + +# Configure logging +logger = get_logger(__name__) + if __name__ == "__main__": buf_size = 256 # Allocate memory and register with NIXL - print("Using NIXL Plugins from:") - print(os.environ["NIXL_PLUGIN_DIR"]) + logger.info("Using NIXL Plugins from:\n%s", os.environ["NIXL_PLUGIN_DIR"]) # Example using nixl_agent_config agent_config = nixl_agent_config(backends=["UCX"]) @@ -37,14 +41,17 @@ plugin_list = nixl_agent1.get_plugin_list() assert "UCX" in plugin_list - print("Plugin parameters") - print(nixl_agent1.get_plugin_mem_types("UCX")) - print(nixl_agent1.get_plugin_params("UCX")) + logger.info( + "Plugin parameters:\n%s\n%s", + nixl_agent1.get_plugin_mem_types("UCX"), + nixl_agent1.get_plugin_params("UCX"), + ) - print("\nLoaded backend parameters") - print(nixl_agent1.get_backend_mem_types("UCX")) - print(nixl_agent1.get_backend_params("UCX")) - print() + logger.info( + "Backend parameters:\n%s\n%s", + nixl_agent1.get_backend_mem_types("UCX"), + nixl_agent1.get_backend_params("UCX"), + ) addr1 = nixl_utils.malloc_passthru(buf_size * 2) addr2 = addr1 + buf_size @@ -66,7 +73,9 @@ assert agent1_xfer_descs == agent1_xfer_descs_np assert agent1_reg_descs == agent1_reg_descs_np - print(agent1_reg_descs, agent1_reg_descs_np) + logger.debug( + "Registration descriptors: %s %s", agent1_reg_descs, agent1_reg_descs_np + ) # Just for tensor test tensors = [torch.zeros(10, dtype=torch.float32) for _ in range(2)] @@ -92,7 +101,7 @@ # Exchange metadata meta = nixl_agent1.get_agent_metadata() remote_name = nixl_agent2.add_remote_agent(meta) - print("Loaded name from metadata:", remote_name, flush=True) + logger.info("Loaded name from metadata: %s", remote_name) serdes = nixl_agent1.get_serialized_descs(agent1_reg_descs) src_descs_recvd = nixl_agent2.deserialize_descs(serdes) @@ -103,7 +112,7 @@ "READ", agent2_xfer_descs, agent1_xfer_descs, remote_name, b"UUID1" ) if not xfer_handle_1: - print("Creating transfer failed.") + logger.error("Creating transfer failed.") exit() # test multiple postings @@ -118,16 +127,16 @@ if not init_done: state = nixl_agent2.check_xfer_state(xfer_handle_1) if state == "ERR": - print("Transfer got to Error state.") + logger.error("Transfer got to Error state.") exit() elif state == "DONE": init_done = True - print("Initiator done") + logger.info("Initiator done") if not target_done: if nixl_agent1.check_remote_xfer_done("initiator", b"UUID1"): target_done = True - print("Target done") + logger.info("Target done") # prep transfer mode local_prep_handle = nixl_agent2.prep_xfer_dlist( @@ -145,30 +154,29 @@ test_notif = str.encode("DESCS: ") + serdes nixl_agent2.send_notif(remote_name, test_notif) - print("sent notif ") - print(test_notif) + logger.info("sent notif: \n%s", test_notif) notif_recv = False while not notif_recv: notif_map = nixl_agent1.get_new_notifs() if "initiator" in notif_map: - print("received message from initiator") + logger.info("received message from initiator") for msg in notif_map["initiator"]: if msg == test_notif: notif_recv = True - print("notif test complete, doing transfer 2\n") + logger.info("notif test complete, doing transfer 2") xfer_handle_2 = nixl_agent2.make_prepped_xfer( "WRITE", local_prep_handle, [0, 1], remote_prep_handle, [1, 0], b"UUID2" ) if not local_prep_handle or not remote_prep_handle: - print("Preparing transfer side handles failed.") + logger.error("Preparing transfer side handles failed.") exit() if not xfer_handle_2: - print("Make prepped transfer failed.") + logger.error("Make prepped transfer failed.") exit() state = nixl_agent2.transfer(xfer_handle_2) @@ -177,22 +185,22 @@ target_done = False init_done = False - print("Transfer 2 started") + logger.info("Transfer 2 started") while (not init_done) or (not target_done): if not init_done: state = nixl_agent2.check_xfer_state(xfer_handle_2) if state == "ERR": - print("Transfer got to Error state.") + logger.error("Transfer got to Error state.") exit() elif state == "DONE": init_done = True - print("Initiator done") + logger.info("Initiator done") if not target_done: if nixl_agent1.check_remote_xfer_done("initiator", b"UUID2"): target_done = True - print("Target done") + logger.info("Target done") nixl_agent2.release_xfer_handle(xfer_handle_1) nixl_agent2.release_xfer_handle(xfer_handle_2) @@ -205,4 +213,4 @@ nixl_utils.free_passthru(addr1) nixl_utils.free_passthru(addr3) - print("Test Complete.") + logger.info("Test Complete.") diff --git a/examples/python/nixl_gds_example.py b/examples/python/nixl_gds_example.py index c0d9c0c5c1..065968ca1e 100755 --- a/examples/python/nixl_gds_example.py +++ b/examples/python/nixl_gds_example.py @@ -20,17 +20,21 @@ import nixl._utils as nixl_utils from nixl._api import nixl_agent, nixl_agent_config +from nixl.logging import get_logger + +# Configure logging +logger = get_logger(__name__) + if __name__ == "__main__": buf_size = 16 * 4096 # Allocate memory and register with NIXL if len(sys.argv) < 2: - print("Please specify file path in argv") + logger.error("Please specify file path in argv") exit(0) - print("Using NIXL Plugins from:") - print(os.environ["NIXL_PLUGIN_DIR"]) + logger.info("Using NIXL Plugins from:\n%s", os.environ["NIXL_PLUGIN_DIR"]) agent_config = nixl_agent_config(backends=[]) nixl_agent1 = nixl_agent("GDSTester", agent_config) @@ -38,16 +42,19 @@ plugin_list = nixl_agent1.get_plugin_list() assert "GDS" in plugin_list - print("Plugin parameters") - print(nixl_agent1.get_plugin_mem_types("GDS")) - print(nixl_agent1.get_plugin_params("GDS")) + logger.info( + "Plugin parameters:\n%s\n%s\n", + nixl_agent1.get_plugin_mem_types("GDS"), + nixl_agent1.get_plugin_params("GDS"), + ) nixl_agent1.create_backend("GDS") - print("\nLoaded backend parameters") - print(nixl_agent1.get_backend_mem_types("GDS")) - print(nixl_agent1.get_backend_params("GDS")) - print() + logger.info( + "Backend parameters:\n%s\n%s\n", + nixl_agent1.get_backend_mem_types("GDS"), + nixl_agent1.get_backend_params("GDS"), + ) # get DRAM buf and initialize it to 0xba for verification addr1 = nixl_utils.malloc_passthru(buf_size) @@ -78,7 +85,7 @@ "WRITE", agent1_xfer1_descs, agent1_xfer_files, "GDSTester" ) if not xfer_handle_1: - print("Creating transfer failed.") + logger.error("Creating transfer failed.") exit() state = nixl_agent1.transfer(xfer_handle_1) @@ -89,18 +96,18 @@ while not done: state = nixl_agent1.check_xfer_state(xfer_handle_1) if state == "ERR": - print("Transfer got to Error state.") + logger.error("Transfer got to Error state.") exit() elif state == "DONE": done = True - print("Initiator done") + logger.info("Initiator done") # read file data back into second buffer xfer_handle_2 = nixl_agent1.initialize_xfer( "READ", agent1_xfer2_descs, agent1_xfer_files, "GDSTester" ) if not xfer_handle_2: - print("Creating transfer failed.") + logger.error("Creating transfer failed.") exit() state = nixl_agent1.transfer(xfer_handle_2) @@ -111,11 +118,11 @@ while not done: state = nixl_agent1.check_xfer_state(xfer_handle_2) if state == "ERR": - print("Transfer got to Error state.") + logger.error("Transfer got to Error state.") exit() elif state == "DONE": done = True - print("Initiator done") + logger.info("Initiator done") # transfer verification nixl_utils.verify_transfer(addr1, addr2, buf_size) @@ -130,4 +137,4 @@ os.close(agent1_fd) - print("Test Complete.") + logger.info("Test Complete.") diff --git a/examples/python/partial_md_example.py b/examples/python/partial_md_example.py index d84a1a762d..431eac311d 100755 --- a/examples/python/partial_md_example.py +++ b/examples/python/partial_md_example.py @@ -21,6 +21,10 @@ import nixl._utils as nixl_utils from nixl._api import nixl_agent, nixl_agent_config from nixl._bindings import nixlNotFoundError +from nixl.logging import get_logger + +# Configure logging +logger = get_logger(__name__) def exchange_target_metadata( @@ -75,13 +79,13 @@ def invalidate_target_metadata( ) args = parser.parse_args() - print("Using NIXL Plugins from:") - print(os.environ["NIXL_PLUGIN_DIR"]) + logger.info("Using NIXL Plugins from:") + logger.info(os.environ["NIXL_PLUGIN_DIR"]) if args.etcd: etcd_endpoints = os.getenv("NIXL_ETCD_ENDPOINTS", "") if etcd_endpoints: - print("NIXL_ETCD_ENDPOINTS is set, using endpoints: ", etcd_endpoints) + logger.info("NIXL_ETCD_ENDPOINTS is set, using endpoints: ", etcd_endpoints) else: raise ValueError( "NIXL_ETCD_ENDPOINTS is not set, but --etcd flag is provided" @@ -89,7 +93,7 @@ def invalidate_target_metadata( else: etcd_endpoints = "" del os.environ["NIXL_ETCD_ENDPOINTS"] - print("NIXL_ETCD_ENDPOINTS is not set, using socket exchange") + logger.info("NIXL_ETCD_ENDPOINTS is not set, using socket exchange") # Needed for socket exchange ip_addr = "127.0.0.1" @@ -167,16 +171,16 @@ def invalidate_target_metadata( if not init_done: state = init_agent.check_xfer_state(xfer_handle_1) if state == "ERR": - print("Transfer got to Error state.") + logger.error("Transfer got to Error state.") exit() elif state == "DONE": init_done = True - print("Initiator done") + logger.info("Initiator done") if not target_done: if target_agent.check_remote_xfer_done("initiator", b"UUID1"): target_done = True - print("Target done") + logger.info("Target done") # Second set of descs was not sent, should fail try: @@ -184,9 +188,9 @@ def invalidate_target_metadata( "READ", init_xfer_descs, target_xfer_descs2, "target", b"UUID1" ) except nixlNotFoundError: - print("Correct exception") + logger.info("Correct exception") else: - print("Incorrect success") + logger.error("Incorrect success") os.abort() # Now send rest of descs @@ -224,16 +228,16 @@ def invalidate_target_metadata( if not init_done: state = init_agent.check_xfer_state(xfer_handle_2) if state == "ERR": - print("Transfer got to Error state.") + logger.error("Transfer got to Error state.") exit() elif state == "DONE": init_done = True - print("Initiator done") + logger.info("Initiator done") if not target_done: if target_agent.check_remote_xfer_done("initiator", b"UUID1"): target_done = True - print("Target done") + logger.info("Target done") init_agent.release_xfer_handle(xfer_handle_1) init_agent.release_xfer_handle(xfer_handle_2) @@ -249,4 +253,4 @@ def invalidate_target_metadata( del init_agent del target_agent - print("Test Complete.") + logger.info("Test Complete.") diff --git a/examples/python/query_mem_example.py b/examples/python/query_mem_example.py index 9cf38cb84b..2d210616f1 100755 --- a/examples/python/query_mem_example.py +++ b/examples/python/query_mem_example.py @@ -21,18 +21,24 @@ try: from nixl._api import nixl_agent, nixl_agent_config + from nixl.logging import get_logger + + logger = get_logger(__name__) NIXL_AVAILABLE = True except ImportError: - print("NIXL API missing install NIXL.") + import logging + + logger = logging.getLogger(__name__) + logger.error("NIXL API missing install NIXL.") NIXL_AVAILABLE = False if __name__ == "__main__": - print("NIXL queryMem Python API Example") - print("=" * 40) + logger.info("NIXL queryMem Python API Example") + logger.info("=" * 40) if not NIXL_AVAILABLE: - print("Skipping example - NIXL bindings not available") + logger.warning("Skipping example - NIXL bindings not available") sys.exit(0) # Create temporary test files @@ -49,17 +55,17 @@ non_existent_file = "/tmp/nixl_example_nonexistent.txt" try: - print("Using NIXL Plugins from:") - print(os.environ["NIXL_PLUGIN_DIR"]) + logger.info("Using NIXL Plugins from:") + logger.info(os.environ["NIXL_PLUGIN_DIR"]) # Create an NIXL agent - print("Creating NIXL agent...") + logger.info("Creating NIXL agent...") config = nixl_agent_config(False, False, 0, []) agent = nixl_agent("example_agent", config) # Prepare a list of tuples as file paths in metaInfo field for querying. # Addr and length and devID fields are set to 0 for file queries. - print("Preparing file paths for querying...") + logger.info("Preparing file paths for querying...") file_paths = [ (0, 0, 0, temp_files[0]), # Existing file 1 (0, 0, 0, temp_files[1]), # Existing file 2 @@ -68,57 +74,57 @@ ] # Query memory using queryMem - print("Querying memory/storage information...") + logger.info("Querying memory/storage information...") # Try to create a backend with POSIX plugin try: params = agent.get_plugin_params("POSIX") agent.create_backend("POSIX", params) - print("Created backend: POSIX") + logger.info("Created backend: POSIX") # Query with specific backend resp = agent.query_memory(file_paths, "POSIX", mem_type="FILE") except Exception as e: - print(f"POSIX backend creation failed: {e}") + logger.exception("POSIX backend creation failed: %s", e) # Try MOCK_DRAM as fallback try: params = agent.get_plugin_params("MOCK_DRAM") agent.create_backend("MOCK_DRAM", params) - print("Created backend: MOCK_DRAM") + logger.info("Created backend: MOCK_DRAM") # Query with specific backend resp = agent.query_memory(file_paths, "MOCK_DRAM", mem_type="FILE") except Exception as e2: - print(f"MOCK_DRAM also failed: {e2}") - print("No working backends available") + logger.exception("MOCK_DRAM also failed: %s", e2) + logger.exception("No working backends available") sys.exit(0) # Display results - print(f"\nQuery results ({len(resp)} responses):") - print("-" * 50) + logger.info("\nQuery results (%d responses):", len(resp)) + logger.info("-" * 50) for i, result in enumerate(resp): - print(f"Descriptor {i}:") + logger.info("Descriptor %d:", i) if result is not None: - print(f" File size: {result.get('size', 'N/A')} bytes") - print(f" File mode: {result.get('mode', 'N/A')}") - print(f" Modified time: {result.get('mtime', 'N/A')}") + logger.info(" File size: %s bytes", result.get("size", "N/A")) + logger.info(" File mode: %s", result.get("mode", "N/A")) + logger.info(" Modified time: %s", result.get("mtime", "N/A")) else: - print(" File does not exist or is not accessible") - print() + logger.info(" File does not exist or is not accessible") + logger.info("") - print("Example completed successfully!") + logger.info("Example completed successfully!") except Exception as e: - print(f"Error in example: {e}") + logger.exception("Error in example: %s", e) import traceback traceback.print_exc() finally: # Clean up temporary files - print("Cleaning up temporary files...") + logger.info("Cleaning up temporary files...") for temp_file_path in temp_files: if os.path.exists(temp_file_path): os.unlink(temp_file_path) - print(f"Removed: {temp_file_path}") + logger.info("Removed: %s", temp_file_path) diff --git a/src/api/python/_api.py b/src/api/python/_api.py index 58db3f2215..b6be714539 100644 --- a/src/api/python/_api.py +++ b/src/api/python/_api.py @@ -20,6 +20,10 @@ import torch import nixl._bindings as nixlBind +from nixl.logging import get_logger + +# Get logger using centralized configuration +logger = get_logger(__name__) DEFAULT_COMM_PORT = nixlBind.DEFAULT_COMM_PORT @@ -76,7 +80,7 @@ def __init__( ): if nixl_conf and instantiate_all: instantiate_all = False - print( + logger.warning( "Ignoring instantiate_all based on the provided config in agent creation." ) if not nixl_conf: @@ -105,7 +109,7 @@ def __init__( self.plugin_list = self.agent.getAvailPlugins() if len(self.plugin_list) == 0: - print("No plugins available, cannot start transfers!") + logger.error("No plugins available, cannot start transfers!") raise RuntimeError("No plugins available for NIXL, cannot start transfers!") self.plugin_b_options: dict[str, dict[str, str]] = {} @@ -125,10 +129,9 @@ def __init__( for bknd in nixl_conf.backends: # TODO: populate init from nixl_conf when added if bknd not in self.plugin_list: - print( - "Skipping backend registration", + logger.warning( + "Skipping backend registration %s due to the missing plugin.", bknd, - "due to the missing plugin.", ) else: self.create_backend(bknd, init) @@ -147,7 +150,7 @@ def __init__( "READ": nixlBind.NIXL_READ, } - print("Initialized NIXL agent:", agent_name) + logger.info("Initialized NIXL agent: %s", agent_name) """ @brief Get the list of available plugins. @@ -169,7 +172,9 @@ def get_plugin_mem_types(self, backend: str) -> list[str]: if backend in self.plugin_mem_types: return self.plugin_mem_types[backend] else: - print("Plugin", backend, "is not available to get its supported mem types.") + logger.warning( + "Plugin %s is not available to get its supported mem types.", backend + ) return [] """ @@ -184,7 +189,7 @@ def get_plugin_params(self, backend: str) -> dict[str, str]: if backend in self.plugin_b_options: return self.plugin_b_options[backend] else: - print("Plugin", backend, "is not available to get its parameters.") + logger.warning("Plugin %s is not available to get its parameters.", backend) return {} """ @@ -201,8 +206,8 @@ def get_backend_mem_types(self, backend: str) -> list[str]: if backend in self.backend_mems: return self.backend_mems[backend] else: - print( - "Backend", backend, "not instantiated to get its supported mem types." + logger.warning( + "Backend %s not instantiated to get its supported mem types.", backend ) return [] @@ -220,7 +225,9 @@ def get_backend_params(self, backend: str) -> dict[str, str]: if backend in self.backend_options: return self.backend_options[backend] else: - print("Backend", backend, "not instantiated to get its parameters.") + logger.warning( + "Backend %s not instantiated to get its parameters.", backend + ) return {} """ @@ -238,7 +245,7 @@ def create_backend(self, backend: str, initParams: dict[str, str] = {}): ) self.backend_mems[backend] = mem_types self.backend_options[backend] = backend_options - print("Backend", backend, "was instantiated") + logger.info("Backend %s was instantiated", backend) """ @brief Register memory regions, optionally with specified backends. @@ -793,7 +800,7 @@ def get_xfer_descs( if isinstance(descs, nixlBind.nixlXferDList): return descs elif isinstance(descs, nixlBind.nixlRegDList): - print("RegList type detected for transfer, please use XferList") + logger.error("RegList type detected for transfer, please use XferList") new_descs = None elif isinstance(descs[0], tuple): if mem_type is not None and len(descs[0]) == 3: @@ -801,10 +808,10 @@ def get_xfer_descs( self.nixl_mems[mem_type], descs, is_sorted ) elif mem_type is None: - print("Please specify a mem type if not using Tensors") + logger.error("Please specify a mem type if not using Tensors") new_descs = None else: - print("3-tuple list needed for transfer") + logger.error("3-tuple list needed for transfer") new_descs = None elif isinstance(descs, np.ndarray): if mem_type is not None and descs.ndim == 2 and descs.shape[1] == 3: @@ -812,10 +819,10 @@ def get_xfer_descs( self.nixl_mems[mem_type], descs, is_sorted ) elif mem_type is None: - print("Please specify a mem type if not using Tensors") + logger.error("Please specify a mem type if not using Tensors") new_descs = None else: - print( + logger.error( "Nx3 shape required for transfer descriptor list from numpy array" ) new_descs = None @@ -833,7 +840,7 @@ def get_xfer_descs( is_sorted, ) else: - print("Please use a list of contiguous Tensors") + logger.error("Please use a list of contiguous Tensors") new_descs = None elif isinstance(descs[0], torch.Tensor): # List[torch.Tensor]: tensor_type = descs[0].device @@ -843,7 +850,7 @@ def get_xfer_descs( if descs[i].device != tensor_type: return None if not descs[i].is_contiguous(): - print("Please use a list of contiguous Tensors") + logger.error("Please use a list of contiguous Tensors") return None base_addr = descs[i].data_ptr() region_len = descs[i].numel() * descs[i].element_size() @@ -887,7 +894,7 @@ def get_reg_descs( if isinstance(descs, nixlBind.nixlRegDList): return descs elif isinstance(descs, nixlBind.nixlXferDList): - print("XferList type detected for registration, please use RegList") + logger.error("XferList type detected for registration, please use RegList") new_descs = None elif isinstance(descs[0], tuple): if mem_type is not None and len(descs[0]) == 4: @@ -895,10 +902,10 @@ def get_reg_descs( self.nixl_mems[mem_type], descs, is_sorted ) elif mem_type is None: - print("Please specify a mem type if not using Tensors") + logger.error("Please specify a mem type if not using Tensors") new_descs = None else: - print("4-tuple list needed for registration") + logger.error("4-tuple list needed for registration") new_descs = None elif isinstance(descs, np.ndarray): if mem_type is not None and descs.ndim == 2 and descs.shape[1] == 3: @@ -906,10 +913,10 @@ def get_reg_descs( self.nixl_mems[mem_type], descs, is_sorted ) elif mem_type is None: - print("Please specify a mem type if not using Tensors") + logger.error("Please specify a mem type if not using Tensors") new_descs = None else: - print( + logger.error( "Nx3 shape required for transfer descriptor list from numpy array" ) new_descs = None @@ -927,7 +934,7 @@ def get_reg_descs( is_sorted, ) else: - print("Please use a list of contiguous Tensors") + logger.error("Please use a list of contiguous Tensors") new_descs = None elif isinstance(descs[0], torch.Tensor): # List[torch.Tensor]: tensor_type = descs[0].device @@ -937,7 +944,7 @@ def get_reg_descs( if descs[i].device != tensor_type: return None if not descs[i].is_contiguous(): - print("Please use a list of contiguous Tensors") + logger.error("Please use a list of contiguous Tensors") return None base_addr = descs[i].data_ptr() region_len = descs[i].numel() * descs[i].element_size() diff --git a/src/api/python/logging.py b/src/api/python/logging.py new file mode 100644 index 0000000000..668440c9bc --- /dev/null +++ b/src/api/python/logging.py @@ -0,0 +1,110 @@ +# SPDX-FileCopyrightText: Copyright (c) 2025 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +""" +Centralized logging configuration for NIXL. + +This module provides a single point of configuration for logging across the entire NIXL project. +It supports: +1. Loading configuration from logging.ini file +2. Override log levels using NIXL_LOG_LEVEL environment variable +3. Fallback to sensible defaults if configuration files are missing + +Usage: + from nixl_logging import get_logger + + logger = get_logger(__name__) + logger.info("This is a log message") + +Or for backward compatibility: + import nixl_logging + import logging + + logger = logging.getLogger(__name__) + logger.info("This is a log message") +""" + +import logging +import logging.config +import os +import socket + +_logging_configured = False + + +class HostnameFilter(logging.Filter): + """Filter that adds hostname to log records.""" + + def __init__(self): + super().__init__() + self._hostname = socket.gethostname() + + def filter(self, record): + record.hostname = self._hostname + return True + + +def set_log_level_by_env(nixl_logger: logging.Logger) -> None: + # Override log level from environment variable if set + env_log_level = os.getenv("NIXL_LOG_LEVEL") + if env_log_level: + try: + # Convert string to logging level + numeric_level = getattr(logging, env_log_level.upper(), None) + if not isinstance(numeric_level, int): + raise ValueError(f"Invalid log level: {env_log_level}") + + # Set the level for the nixl logger + nixl_logger.setLevel(numeric_level) + + nixl_logger.info( + "Log level set to %s from NIXL_LOG_LEVEL environment variable", + env_log_level.upper(), + ) + except (ValueError, AttributeError) as e: + nixl_logger.warning( + "Invalid NIXL_LOG_LEVEL value '%s': %s. Using configuration default.", + env_log_level, + e, + ) + + +def setup_logging() -> None: + global _logging_configured + + if _logging_configured: + return + + # Find python_logging.ini in the same directory as this file (project root) + current_dir = os.path.dirname(os.path.abspath(__file__)) + config_file_path = os.path.join(current_dir, "python_logging.ini") + config_file = os.path.abspath(config_file_path) + + logging.config.fileConfig(config_file, disable_existing_loggers=False) + + nixl_logger = logging.getLogger("nixl") + hostname_filter = HostnameFilter() + nixl_logger.addFilter(hostname_filter) + set_log_level_by_env(nixl_logger) + + _logging_configured = True + + +def get_logger(name: str) -> logging.Logger: + setup_logging() + # Convert module name to nixl hierarchy + # e.g., '_api' -> 'nixl.api', 'test_nixl_bindings' -> 'nixl.test_nixl_bindings' + clean_name = name.lstrip("_") # Remove leading underscores + return logging.getLogger(f"nixl.{clean_name}") diff --git a/src/api/python/meson.build b/src/api/python/meson.build index 6f7eb80eda..1ce26b8bd8 100644 --- a/src/api/python/meson.build +++ b/src/api/python/meson.build @@ -18,3 +18,5 @@ py = import('python').find_installation('python3', pure: false) py.install_sources('_api.py', subdir: ('nixl')) py.install_sources('__init__.py', subdir: ('nixl')) py.install_sources('py.typed', subdir: ('nixl')) +py.install_sources('logging.py', subdir: ('nixl')) +py.install_sources('python_logging.ini', subdir: ('nixl')) diff --git a/src/api/python/python_logging.ini b/src/api/python/python_logging.ini new file mode 100644 index 0000000000..c7d50a9dd7 --- /dev/null +++ b/src/api/python/python_logging.ini @@ -0,0 +1,43 @@ +# SPDX-FileCopyrightText: Copyright (c) 2025 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +[loggers] +keys=root,nixl + +[handlers] +keys=consoleHandler + +[formatters] +keys=simpleFormatter + +[logger_root] +level=WARNING +handlers= + +[logger_nixl] +level=INFO +handlers=consoleHandler +qualname=nixl +propagate=0 + +[handler_consoleHandler] +class=StreamHandler +level=INFO +formatter=simpleFormatter +args=(sys.stdout,) + +[formatter_simpleFormatter] +format=%(asctime)s [%(hostname)s:%(process)d] NIXL %(levelname)-7s %(filename)s:%(lineno)d %(message)s +datefmt=%Y-%m-%d %H:%M:%S diff --git a/test/python/desc_perf.py b/test/python/desc_perf.py index a1e8309875..7c40a9b7e3 100755 --- a/test/python/desc_perf.py +++ b/test/python/desc_perf.py @@ -19,6 +19,9 @@ import nixl._utils as nixl_utils from nixl._api import nixl_agent +from nixl.logging import get_logger + +logger = get_logger(__name__) if __name__ == "__main__": desc_count = 24 * 64 * 1024 @@ -35,7 +38,8 @@ assert descs.descCount() == desc_count - print( - "Time per desc add in us:", (1000000.0 * (end_time - start_time)) / desc_count + logger.info( + "Time per desc add in us: %f", + (1000000.0 * (end_time - start_time)) / desc_count, ) nixl_utils.free_passthru(addr) diff --git a/test/python/prep_xfer_perf.py b/test/python/prep_xfer_perf.py index c462e1d607..0ca48ea064 100755 --- a/test/python/prep_xfer_perf.py +++ b/test/python/prep_xfer_perf.py @@ -21,6 +21,9 @@ import nixl._utils as nixl_utils from nixl._api import nixl_agent, nixl_agent_config +from nixl.logging import get_logger + +logger = get_logger(__name__) def init_agent(): @@ -34,12 +37,12 @@ def prep_handles(agent: nixl_agent, xfer_dlist, reg_dlist, indices): xfer_dlist_trim = reg_dlist.trim() elapsed = time.perf_counter() - start assert xfer_dlist_trim.descCount() == xfer_dlist.descCount() - print(f"Trim nixlRegDList:\t{elapsed:.4f} sec") + logger.info("Trim nixlRegDList:\t%.4f sec", elapsed) start = time.perf_counter() assert agent.register_memory(reg_dlist) is not None elapsed = time.perf_counter() - start - print(f"register_memory:\t{elapsed:.4f} sec") + logger.info("register_memory:\t%.4f sec", elapsed) start = time.perf_counter() local_prep_handle = agent.prep_xfer_dlist( @@ -47,13 +50,13 @@ def prep_handles(agent: nixl_agent, xfer_dlist, reg_dlist, indices): ) elapsed = time.perf_counter() - start assert local_prep_handle - print(f"prep_xfer_dlist INIT:\t{elapsed:.4f} sec") + logger.info("prep_xfer_dlist INIT:\t%.4f sec", elapsed) start = time.perf_counter() remote_prep_handle = agent.prep_xfer_dlist("agent", xfer_dlist, "DRAM", False) elapsed = time.perf_counter() - start assert remote_prep_handle - print(f"prep_xfer_dlist SELF:\t{elapsed:.4f} sec") + logger.info("prep_xfer_dlist SELF:\t%.4f sec", elapsed) start = time.perf_counter() xfer_handle = agent.make_prepped_xfer( @@ -61,15 +64,15 @@ def prep_handles(agent: nixl_agent, xfer_dlist, reg_dlist, indices): ) elapsed = time.perf_counter() - start assert xfer_handle - print(f"make_prepped_xfer:\t{elapsed:.4f} sec") + logger.info("make_prepped_xfer:\t%.4f sec", elapsed) return local_prep_handle, remote_prep_handle, xfer_handle def perf_test_list(num_descs: int, addr_base: int, length: int): - print("-" * 40) - print("Starting list test...") - print("-" * 40) + logger.info("-" * 40) + logger.info("Starting list test...") + logger.info("-" * 40) agent = init_agent() descs_list = [(addr_base + i * length, length, 0) for i in range(num_descs)] indices = list(range(num_descs)) @@ -78,7 +81,7 @@ def perf_test_list(num_descs: int, addr_base: int, length: int): xfer_dlist = agent.get_xfer_descs(descs_list, "DRAM", False) elapsed = time.perf_counter() - start assert xfer_dlist.descCount() == num_descs - print(f"get_xfer_descs:\t\t{elapsed:.4f} sec") + logger.info("get_xfer_descs:\t\t%.4f sec", elapsed) blob_descs_list = [ (addr_base + i * length, length, 0, b"") for i in range(num_descs) @@ -87,7 +90,7 @@ def perf_test_list(num_descs: int, addr_base: int, length: int): reg_dlist = agent.get_reg_descs(blob_descs_list, "DRAM", False) elapsed = time.perf_counter() - start assert reg_dlist.descCount() == num_descs - print(f"get_reg_descs:\t\t{elapsed:.4f} sec") + logger.info("get_reg_descs:\t\t%.4f sec", elapsed) local_prep_handle, remote_prep_handle, xfer_handle = prep_handles( agent, xfer_dlist, reg_dlist, indices @@ -97,9 +100,9 @@ def perf_test_list(num_descs: int, addr_base: int, length: int): def perf_test_array(num_descs: int, addr_base: int, length: int): - print("-" * 40) - print("Starting array test...") - print("-" * 40) + logger.info("-" * 40) + logger.info("Starting array test...") + logger.info("-" * 40) agent = init_agent() descs_np = np.zeros((num_descs, 3), dtype=np.uint64) indices = np.arange(num_descs) @@ -111,13 +114,13 @@ def perf_test_array(num_descs: int, addr_base: int, length: int): xfer_dlist = agent.get_xfer_descs(descs_np, "DRAM", False) elapsed = time.perf_counter() - start assert xfer_dlist.descCount() == num_descs - print(f"get_xfer_descs:\t\t{elapsed:.4f} sec") + logger.info("get_xfer_descs:\t\t%.4f sec", elapsed) start = time.perf_counter() reg_dlist = agent.get_reg_descs(descs_np, "DRAM", False) elapsed = time.perf_counter() - start assert reg_dlist.descCount() == num_descs - print(f"get_reg_descs:\t\t{elapsed:.4f} sec") + logger.info("get_reg_descs:\t\t%.4f sec", elapsed) local_prep_handle, remote_prep_handle, xfer_handle = prep_handles( agent, xfer_dlist, reg_dlist, indices @@ -138,8 +141,7 @@ def perf_test_array(num_descs: int, addr_base: int, length: int): ) args = parser.parse_args() - print("Using NIXL Plugins from:") - print(os.environ["NIXL_PLUGIN_DIR"]) + logger.info("Using NIXL Plugins from:\n%s", os.environ["NIXL_PLUGIN_DIR"]) # Example using nixl_agent_config agent = init_agent() @@ -147,7 +149,9 @@ def perf_test_array(num_descs: int, addr_base: int, length: int): num_descs = 2**8 length = 1024 addr_base = nixl_utils.malloc_passthru(num_descs * length) - print(f"Performance test: Creating nixlXferDList with {num_descs} descriptors") + logger.info( + "Performance test: Creating nixlXferDList with %d descriptors", num_descs + ) if args.mode == "list": perf_test_list(num_descs, addr_base, length) diff --git a/test/python/test_nixl_bindings.py b/test/python/test_nixl_bindings.py index b73aede291..c30352ee72 100644 --- a/test/python/test_nixl_bindings.py +++ b/test/python/test_nixl_bindings.py @@ -19,6 +19,9 @@ import nixl._bindings as nixl import nixl._utils as nixl_utils +from nixl.logging import get_logger + +logger = get_logger(__name__) # These should automatically be run by pytest because of function names @@ -33,7 +36,7 @@ def test_list(): pickled_list = pickle.dumps(test_list) - print(pickled_list) + logger.info(pickled_list) unpickled_list = pickle.loads(pickled_list) @@ -41,7 +44,7 @@ def test_list(): assert test_list.getType() == nixl.DRAM_SEG - print(test_list.descCount()) + logger.info("Descriptor count: %s", test_list.descCount()) assert test_list.descCount() == 3 test_list.remDesc(1) @@ -89,10 +92,8 @@ def test_agent(): meta1 = agent1.getLocalMD() meta2 = agent2.getLocalMD() - print("Agent1 MD: ") - print(meta1) - print("Agent2 MD: ") - print(meta2) + logger.info("Agent1 MD: \n%s", meta1) + logger.info("Agent2 MD: \n%s", meta2) ret_name = agent1.loadRemoteMD(meta2) assert ret_name.decode(encoding="UTF-8") == name2 @@ -108,23 +109,23 @@ def test_agent(): dst_list = nixl.nixlXferDList(nixl.DRAM_SEG, False) dst_list.addDesc((addr2 + offset, req_size, 0)) - print("Transfer from " + str(addr1 + offset) + " to " + str(addr2 + offset)) + logger.info("Transfer from %s to %s", str(addr1 + offset), str(addr2 + offset)) noti_str = "n\0tification" - print(noti_str) + logger.info("Notification string: %s", noti_str) - print(src_list) - print(dst_list) + logger.info("Source list: %s", src_list) + logger.info("Destination list: %s", dst_list) handle = agent1.createXferReq(nixl.NIXL_WRITE, src_list, dst_list, name2, noti_str) assert handle != 0 - print(handle) + logger.info("Transfer handle: %s", handle) status = agent1.postXferReq(handle) assert status == nixl.NIXL_SUCCESS or status == nixl.NIXL_IN_PROG - print("Transfer posted") + logger.info("Transfer posted") notifMap = {} @@ -139,10 +140,10 @@ def test_agent(): nixl_utils.verify_transfer(addr1 + offset, addr2 + offset, req_size) assert len(notifMap[name1]) == 1 - print(notifMap[name1][0]) + logger.info("Received notification: %s", notifMap[name1][0]) assert notifMap[name1][0] == noti_str.encode() - print("Transfer verified") + logger.info("Transfer verified") agent1.releaseXferReq(handle) @@ -193,8 +194,9 @@ def test_query_mem(): assert len(resp) == 0 except Exception as e: # Some backends might not support queryMem, which is okay - print( - f"queryMem with empty list failed (expected for some backends): {e}" + logger.exception( + "queryMem with empty list failed (expected for some backends): %s", + e, ) # Test 2: Query with actual file descriptors @@ -228,16 +230,19 @@ def test_query_mem(): except Exception as e: # Some backends might not support queryMem, which is okay - print(f"queryMem failed (expected for some backends): {e}") + logger.exception( + "queryMem failed (expected for some backends): %s", + e, + ) except Exception as e: - print(f"Backend creation failed: {e}") + logger.exception("Backend creation failed: %s", e) # Try MOCK_DRAM as fallback try: params, mems = agent.getPluginParams("MOCK_DRAM") backend = agent.createBackend("MOCK_DRAM", params) - print("Using MOCK_DRAM backend") + logger.info("Using MOCK_DRAM backend") except Exception as e2: - print(f"MOCK_DRAM also failed: {e2}") + logger.exception("MOCK_DRAM also failed: %s", e2) return finally: From fc580fe270c8272250878a2de2a0ad53c3f3b4e9 Mon Sep 17 00:00:00 2001 From: Roie Danino Date: Tue, 29 Jul 2025 13:26:08 +0300 Subject: [PATCH 2/7] SRC/API/PYTHON: removed hostname from log format Signed-off-by: Roie Danino --- src/api/python/logging.py | 14 -------------- src/api/python/python_logging.ini | 2 +- 2 files changed, 1 insertion(+), 15 deletions(-) diff --git a/src/api/python/logging.py b/src/api/python/logging.py index 668440c9bc..16c6c91a2f 100644 --- a/src/api/python/logging.py +++ b/src/api/python/logging.py @@ -44,18 +44,6 @@ _logging_configured = False -class HostnameFilter(logging.Filter): - """Filter that adds hostname to log records.""" - - def __init__(self): - super().__init__() - self._hostname = socket.gethostname() - - def filter(self, record): - record.hostname = self._hostname - return True - - def set_log_level_by_env(nixl_logger: logging.Logger) -> None: # Override log level from environment variable if set env_log_level = os.getenv("NIXL_LOG_LEVEL") @@ -95,8 +83,6 @@ def setup_logging() -> None: logging.config.fileConfig(config_file, disable_existing_loggers=False) nixl_logger = logging.getLogger("nixl") - hostname_filter = HostnameFilter() - nixl_logger.addFilter(hostname_filter) set_log_level_by_env(nixl_logger) _logging_configured = True diff --git a/src/api/python/python_logging.ini b/src/api/python/python_logging.ini index c7d50a9dd7..010516bda6 100644 --- a/src/api/python/python_logging.ini +++ b/src/api/python/python_logging.ini @@ -39,5 +39,5 @@ formatter=simpleFormatter args=(sys.stdout,) [formatter_simpleFormatter] -format=%(asctime)s [%(hostname)s:%(process)d] NIXL %(levelname)-7s %(filename)s:%(lineno)d %(message)s +format=%(asctime)s NIXL %(levelname)-7s %(filename)s:%(lineno)d %(message)s datefmt=%Y-%m-%d %H:%M:%S From 285d5be43ace0ba3e2f9d958657560e1b605fa03 Mon Sep 17 00:00:00 2001 From: Roie Danino Date: Tue, 29 Jul 2025 13:35:32 +0300 Subject: [PATCH 3/7] SRC/API/PYTHON: removed unused import Signed-off-by: Roie Danino --- src/api/python/logging.py | 1 - 1 file changed, 1 deletion(-) diff --git a/src/api/python/logging.py b/src/api/python/logging.py index 16c6c91a2f..dbe357e67f 100644 --- a/src/api/python/logging.py +++ b/src/api/python/logging.py @@ -39,7 +39,6 @@ import logging import logging.config import os -import socket _logging_configured = False From b2f38763a4566885643116e20f893a20e7976c20 Mon Sep 17 00:00:00 2001 From: Roie Danino Date: Tue, 29 Jul 2025 14:14:08 +0300 Subject: [PATCH 4/7] SRC/API/PYTHON: checking if setting raiseExceptions = True will fail CI on tracebacks Signed-off-by: Roie Danino --- src/api/python/logging.py | 1 + 1 file changed, 1 insertion(+) diff --git a/src/api/python/logging.py b/src/api/python/logging.py index dbe357e67f..f57cc7145d 100644 --- a/src/api/python/logging.py +++ b/src/api/python/logging.py @@ -83,6 +83,7 @@ def setup_logging() -> None: nixl_logger = logging.getLogger("nixl") set_log_level_by_env(nixl_logger) + logging.raiseExceptions = True _logging_configured = True From f9ec6884e80da972fdb6ce7d23446ebc81849f95 Mon Sep 17 00:00:00 2001 From: Roie Danino Date: Thu, 31 Jul 2025 13:00:58 +0300 Subject: [PATCH 5/7] SRC/API/PYTHON: changed file configuration to dict config, fixed log lines, added env for raising logger exceptions Signed-off-by: Roie Danino --- .gitlab/test_python.sh | 2 ++ examples/python/partial_md_example.py | 5 ++-- examples/python/query_mem_example.py | 4 +-- src/api/python/logging.py | 43 +++++++++++++++++++-------- test/python/test_nixl_bindings.py | 2 +- 5 files changed, 37 insertions(+), 19 deletions(-) diff --git a/.gitlab/test_python.sh b/.gitlab/test_python.sh index 13869e90ab..3eebb70dea 100755 --- a/.gitlab/test_python.sh +++ b/.gitlab/test_python.sh @@ -44,6 +44,8 @@ export CPATH=${INSTALL_DIR}/include:$CPATH export PATH=${INSTALL_DIR}/bin:$PATH export PKG_CONFIG_PATH=${INSTALL_DIR}/lib/pkgconfig:$PKG_CONFIG_PATH export NIXL_PLUGIN_DIR=${INSTALL_DIR}/lib/$ARCH-linux-gnu/plugins +# Raise exceptions for logging errors +export NIXL_DEBUG_LOGGING=yes pip3 install --break-system-packages . pip3 install --break-system-packages pytest diff --git a/examples/python/partial_md_example.py b/examples/python/partial_md_example.py index 431eac311d..596decee0f 100755 --- a/examples/python/partial_md_example.py +++ b/examples/python/partial_md_example.py @@ -79,13 +79,12 @@ def invalidate_target_metadata( ) args = parser.parse_args() - logger.info("Using NIXL Plugins from:") - logger.info(os.environ["NIXL_PLUGIN_DIR"]) + logger.info("Using NIXL Plugins from: %s", os.environ["NIXL_PLUGIN_DIR"]) if args.etcd: etcd_endpoints = os.getenv("NIXL_ETCD_ENDPOINTS", "") if etcd_endpoints: - logger.info("NIXL_ETCD_ENDPOINTS is set, using endpoints: ", etcd_endpoints) + logger.info("NIXL_ETCD_ENDPOINTS is set, using endpoints: %s", etcd_endpoints) else: raise ValueError( "NIXL_ETCD_ENDPOINTS is not set, but --etcd flag is provided" diff --git a/examples/python/query_mem_example.py b/examples/python/query_mem_example.py index 2d210616f1..bf169b90a8 100755 --- a/examples/python/query_mem_example.py +++ b/examples/python/query_mem_example.py @@ -55,8 +55,8 @@ non_existent_file = "/tmp/nixl_example_nonexistent.txt" try: - logger.info("Using NIXL Plugins from:") - logger.info(os.environ["NIXL_PLUGIN_DIR"]) + logger.info("Using NIXL Plugins from: %s", + os.environ["NIXL_PLUGIN_DIR"]) # Create an NIXL agent logger.info("Creating NIXL agent...") diff --git a/src/api/python/logging.py b/src/api/python/logging.py index f57cc7145d..5a2d47704f 100644 --- a/src/api/python/logging.py +++ b/src/api/python/logging.py @@ -16,12 +16,6 @@ """ Centralized logging configuration for NIXL. -This module provides a single point of configuration for logging across the entire NIXL project. -It supports: -1. Loading configuration from logging.ini file -2. Override log levels using NIXL_LOG_LEVEL environment variable -3. Fallback to sensible defaults if configuration files are missing - Usage: from nixl_logging import get_logger @@ -42,6 +36,32 @@ _logging_configured = False +LOGGING_CONFIG = { + 'version': 1, + 'disable_existing_loggers': False, + 'formatters': { + 'simpleFormatter': { + 'format': '%(asctime)s NIXL %(levelname)-7s %(filename)s:%(lineno)d %(message)s', + 'datefmt': '%Y-%m-%d %H:%M:%S' + } + }, + 'handlers': { + 'consoleHandler': { + 'class': 'logging.StreamHandler', + 'level': 'INFO', + 'formatter': 'simpleFormatter', + 'stream': 'ext://sys.stdout' + } + }, + 'loggers': { + 'nixl': { + 'level': 'INFO', + 'handlers': ['consoleHandler'], + 'propagate': False + } + } +} + def set_log_level_by_env(nixl_logger: logging.Logger) -> None: # Override log level from environment variable if set @@ -74,16 +94,13 @@ def setup_logging() -> None: if _logging_configured: return - # Find python_logging.ini in the same directory as this file (project root) - current_dir = os.path.dirname(os.path.abspath(__file__)) - config_file_path = os.path.join(current_dir, "python_logging.ini") - config_file = os.path.abspath(config_file_path) - - logging.config.fileConfig(config_file, disable_existing_loggers=False) + # Use dictionary-based configuration + logging.config.dictConfig(LOGGING_CONFIG) nixl_logger = logging.getLogger("nixl") set_log_level_by_env(nixl_logger) - logging.raiseExceptions = True + + logging.raiseExceptions = os.getenv("NIXL_DEBUG_LOGGING", "").lower() in ("true", "1", "yes") _logging_configured = True diff --git a/test/python/test_nixl_bindings.py b/test/python/test_nixl_bindings.py index c30352ee72..3c59330148 100644 --- a/test/python/test_nixl_bindings.py +++ b/test/python/test_nixl_bindings.py @@ -36,7 +36,7 @@ def test_list(): pickled_list = pickle.dumps(test_list) - logger.info(pickled_list) + logger.info("Pickled list: %s", pickled_list) unpickled_list = pickle.loads(pickled_list) From 05101936dbcada07823f0fc0e7b25faba938a3a4 Mon Sep 17 00:00:00 2001 From: Roie Danino Date: Thu, 31 Jul 2025 13:04:54 +0300 Subject: [PATCH 6/7] EXAMPLES/PYTHON: fixed formatting Signed-off-by: Roie Danino --- examples/python/partial_md_example.py | 4 ++- examples/python/query_mem_example.py | 3 +- src/api/python/logging.py | 42 +++++++++++++-------------- 3 files changed, 25 insertions(+), 24 deletions(-) diff --git a/examples/python/partial_md_example.py b/examples/python/partial_md_example.py index 596decee0f..bd09a6dbcf 100755 --- a/examples/python/partial_md_example.py +++ b/examples/python/partial_md_example.py @@ -84,7 +84,9 @@ def invalidate_target_metadata( if args.etcd: etcd_endpoints = os.getenv("NIXL_ETCD_ENDPOINTS", "") if etcd_endpoints: - logger.info("NIXL_ETCD_ENDPOINTS is set, using endpoints: %s", etcd_endpoints) + logger.info( + "NIXL_ETCD_ENDPOINTS is set, using endpoints: %s", etcd_endpoints + ) else: raise ValueError( "NIXL_ETCD_ENDPOINTS is not set, but --etcd flag is provided" diff --git a/examples/python/query_mem_example.py b/examples/python/query_mem_example.py index bf169b90a8..8e5fcb6deb 100755 --- a/examples/python/query_mem_example.py +++ b/examples/python/query_mem_example.py @@ -55,8 +55,7 @@ non_existent_file = "/tmp/nixl_example_nonexistent.txt" try: - logger.info("Using NIXL Plugins from: %s", - os.environ["NIXL_PLUGIN_DIR"]) + logger.info("Using NIXL Plugins from: %s", os.environ["NIXL_PLUGIN_DIR"]) # Create an NIXL agent logger.info("Creating NIXL agent...") diff --git a/src/api/python/logging.py b/src/api/python/logging.py index 5a2d47704f..d92e76df54 100644 --- a/src/api/python/logging.py +++ b/src/api/python/logging.py @@ -37,29 +37,25 @@ _logging_configured = False LOGGING_CONFIG = { - 'version': 1, - 'disable_existing_loggers': False, - 'formatters': { - 'simpleFormatter': { - 'format': '%(asctime)s NIXL %(levelname)-7s %(filename)s:%(lineno)d %(message)s', - 'datefmt': '%Y-%m-%d %H:%M:%S' + "version": 1, + "disable_existing_loggers": False, + "formatters": { + "simpleFormatter": { + "format": "%(asctime)s NIXL %(levelname)-7s %(filename)s:%(lineno)d %(message)s", + "datefmt": "%Y-%m-%d %H:%M:%S", } }, - 'handlers': { - 'consoleHandler': { - 'class': 'logging.StreamHandler', - 'level': 'INFO', - 'formatter': 'simpleFormatter', - 'stream': 'ext://sys.stdout' + "handlers": { + "consoleHandler": { + "class": "logging.StreamHandler", + "level": "INFO", + "formatter": "simpleFormatter", + "stream": "ext://sys.stdout", } }, - 'loggers': { - 'nixl': { - 'level': 'INFO', - 'handlers': ['consoleHandler'], - 'propagate': False - } - } + "loggers": { + "nixl": {"level": "INFO", "handlers": ["consoleHandler"], "propagate": False} + }, } @@ -99,8 +95,12 @@ def setup_logging() -> None: nixl_logger = logging.getLogger("nixl") set_log_level_by_env(nixl_logger) - - logging.raiseExceptions = os.getenv("NIXL_DEBUG_LOGGING", "").lower() in ("true", "1", "yes") + + logging.raiseExceptions = os.getenv("NIXL_DEBUG_LOGGING", "").lower() in ( + "true", + "1", + "yes", + ) _logging_configured = True From 52a168a37d3f8fca363241102938e661ddc7eaa7 Mon Sep 17 00:00:00 2001 From: Roie Danino Date: Wed, 6 Aug 2025 09:17:46 +0000 Subject: [PATCH 7/7] BENCHMARK/KVBENCH/TEST: fixed review - %s, removing ini file Signed-off-by: Roie Danino --- .../kvbench/test/custom_traffic_perftest.py | 3 +- src/api/python/meson.build | 1 - src/api/python/python_logging.ini | 43 ------------------- 3 files changed, 2 insertions(+), 45 deletions(-) delete mode 100644 src/api/python/python_logging.ini diff --git a/benchmark/kvbench/test/custom_traffic_perftest.py b/benchmark/kvbench/test/custom_traffic_perftest.py index 3bc456940d..838f859c20 100644 --- a/benchmark/kvbench/test/custom_traffic_perftest.py +++ b/benchmark/kvbench/test/custom_traffic_perftest.py @@ -389,7 +389,8 @@ def run( ] ] logger.info( - f"Performance metrics:\n{tabulate(data, headers=headers, floatfmt='.6f')}" + "Performance metrics:\n%s", + tabulate(data, headers=headers, floatfmt=".6f"), ) if verify_buffers: diff --git a/src/api/python/meson.build b/src/api/python/meson.build index 1ce26b8bd8..828c814ef9 100644 --- a/src/api/python/meson.build +++ b/src/api/python/meson.build @@ -19,4 +19,3 @@ py.install_sources('_api.py', subdir: ('nixl')) py.install_sources('__init__.py', subdir: ('nixl')) py.install_sources('py.typed', subdir: ('nixl')) py.install_sources('logging.py', subdir: ('nixl')) -py.install_sources('python_logging.ini', subdir: ('nixl')) diff --git a/src/api/python/python_logging.ini b/src/api/python/python_logging.ini deleted file mode 100644 index 010516bda6..0000000000 --- a/src/api/python/python_logging.ini +++ /dev/null @@ -1,43 +0,0 @@ -# SPDX-FileCopyrightText: Copyright (c) 2025 NVIDIA CORPORATION & AFFILIATES. All rights reserved. -# SPDX-License-Identifier: Apache-2.0 -# -# Licensed under the Apache License, Version 2.0 (the "License"); -# you may not use this file except in compliance with the License. -# You may obtain a copy of the License at -# -# http://www.apache.org/licenses/LICENSE-2.0 -# -# Unless required by applicable law or agreed to in writing, software -# distributed under the License is distributed on an "AS IS" BASIS, -# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -# See the License for the specific language governing permissions and -# limitations under the License. - -[loggers] -keys=root,nixl - -[handlers] -keys=consoleHandler - -[formatters] -keys=simpleFormatter - -[logger_root] -level=WARNING -handlers= - -[logger_nixl] -level=INFO -handlers=consoleHandler -qualname=nixl -propagate=0 - -[handler_consoleHandler] -class=StreamHandler -level=INFO -formatter=simpleFormatter -args=(sys.stdout,) - -[formatter_simpleFormatter] -format=%(asctime)s NIXL %(levelname)-7s %(filename)s:%(lineno)d %(message)s -datefmt=%Y-%m-%d %H:%M:%S