一、工具简洁

1、工具介绍

Kubernetes Deployment日志并行搜索工具是一个高效的命令行工具，用来在一个Deployment多个Pod中查询指定日志内容。脚本通过多线程并行处理，大大提升了在多个Pod中查找关键信息的效率。

2、核心功能介绍

• 灵活搜索模式：支持普通文本搜索和正则表达式搜索；
• 上下文显示：可自定义上文和下文内容；
• 结果高亮：在终端中突出显示匹配的文本；
• 进度显示：实时显示搜索进度；
• 结果保存：自动将搜索结果保存到文件；
• 日志截取：可只搜索最近N行日志
• 并行搜索：多线程并发搜索，显著提高效率（10个线程并发搜索时间减少80%以上）。

二、 工具安装与使用

1、环境要求

（1）Python 3.6+
（2）kubectl命令行工具（已配置好Kubernetes集群访问权限）

2、安装步骤

（1）安装s命令

cd /usr/local/bin
vim s

#!/usr/bin/env python3

import argparse
import subprocess
import json
import re
import sys
import os
import time
from concurrent.futures import ThreadPoolExecutor, as_completed
from datetime import datetime
import signal

class Colors:
    RED = '\033[91m'
    GREEN = '\033[92m'
    YELLOW = '\033[93m'
    BLUE = '\033[94m'
    MAGENTA = '\033[95m'
    CYAN = '\033[96m'
    WHITE = '\033[97m'
    BOLD = '\033[1m'
    UNDERLINE = '\033[4m'
    END = '\033[0m'

class LogSearcher:
    def __init__(self, args):
        self.deployment = args.deployment
        self.namespace = args.namespace or "default"
        self.search_text = args.search_text
        self.threads = args.threads or 10
        self.before_context = args.before_context or 0
        self.after_context = args.after_context or 0
        self.context_lines = args.context or 0
        self.tail_lines = args.tail or 1000
        self.use_regex = args.regex
        self.timeout = args.timeout or 30
        self.compiled_pattern = None

        if not self.deployment:
            print(f"{Colors.RED}错误: 必须指定deployment名称{Colors.END}")
            sys.exit(1)

        if not self.search_text:
            print(f"{Colors.RED}错误: 必须指定要搜索的日志内容{Colors.END}")
            sys.exit(1)

        # 如果指定了-c参数，则覆盖before_context和after_context
        if self.context_lines > 0:
            self.before_context = self.context_lines
            self.after_context = self.context_lines

        if self.use_regex:
            try:
                self.compiled_pattern = re.compile(self.search_text, re.IGNORECASE)
            except re.error as e:
                print(f"{Colors.RED}错误: 无效的正则表达式: {e}{Colors.END}")
                sys.exit(1)

    def run_command(self, cmd: list, capture_output: bool = True):
        """执行命令并返回结果 - Python 3.6兼容版本"""
        try:
            if capture_output:
                result = subprocess.run(
                    cmd,
                    stdout=subprocess.PIPE,
                    stderr=subprocess.PIPE,
                    universal_newlines=True,
                    timeout=self.timeout
                )
            else:
                result = subprocess.run(
                    cmd,
                    universal_newlines=True,
                    timeout=self.timeout
                )

            if result.returncode == 0:
                return True, result.stdout.strip()
            else:
                return False, result.stderr.strip()
        except subprocess.TimeoutExpired:
            return False, f"命令执行超时: {' '.join(cmd)}"
        except Exception as e:
            return False, f"执行命令出错: {e}"

    def get_deployment_pods(self) -> list:
        """获取Deployment的所有Pod"""
        print(f"{Colors.BLUE}获取Deployment '{self.deployment}' 的Pod列表...{Colors.END}")

        cmd = ["kubectl", "get", "deployment", self.deployment,
               "-n", self.namespace, "-o", "json"]

        success, output = self.run_command(cmd)
        if not success:
            print(f"{Colors.RED}错误: 无法获取Deployment: {output}{Colors.END}")
            sys.exit(1)

        try:
            deployment_info = json.loads(output)
            selector = deployment_info.get("spec", {}).get("selector", {}).get("matchLabels", {})
            if not selector:
                print(f"{Colors.RED}错误: 无法获取Deployment的标签选择器{Colors.END}")
                sys.exit(1)

            selector_str = ",".join([f"{k}={v}" for k, v in selector.items()])
            print(f"{Colors.YELLOW}标签选择器: {selector_str}{Colors.END}")

            cmd = ["kubectl", "get", "pods", "-n", self.namespace,
                   "-l", selector_str, "-o", "jsonpath='{.items[*].metadata.name}'"]

            success, output = self.run_command(cmd)
            if not success:
                print(f"{Colors.RED}错误: 无法获取Pod列表: {output}{Colors.END}")
                sys.exit(1)

            if output.startswith("'") and output.endswith("'"):
                output = output[1:-1]

            pods = [pod for pod in output.split() if pod]
            if not pods:
                print(f"{Colors.RED}错误: 未找到Deployment相关的Pod{Colors.END}")
                sys.exit(1)

            print(f"{Colors.GREEN}找到 {len(pods)} 个Pod:{Colors.END}")
            for pod in pods:
                print(f"  {pod}")
            print()
            return pods

        except json.JSONDecodeError as e:
            print(f"{Colors.RED}错误: 解析Deployment信息失败: {e}{Colors.END}")
            sys.exit(1)
        except Exception as e:
            print(f"{Colors.RED}错误: {e}{Colors.END}")
            sys.exit(1)

    def get_pod_logs(self, pod_name: str):
        """获取单个Pod的日志"""
        cmd = ["kubectl", "logs", pod_name, "-n", self.namespace]

        if self.tail_lines > 0:
            cmd.extend(["--tail", str(self.tail_lines)])

        success, output = self.run_command(cmd)
        if not success:
            print(f"{Colors.YELLOW}警告: 无法获取Pod {pod_name} 的日志: {output}{Colors.END}")
            return None

        return output

    def search_in_text(self, text: str, pod_name: str) -> list:
        """在文本中搜索匹配的内容"""
        results = []
        lines = text.splitlines()
        total_lines = len(lines)

        if self.use_regex:
            # 使用正则表达式搜索
            for i, line in enumerate(lines):
                if self.compiled_pattern.search(line):
                    result = {
                        "pod": pod_name,
                        "line_number": i + 1,
                        "content": line,
                        "before_context": [],
                        "after_context": []
                    }

                    # 添加上文 (Before context)
                    if self.before_context > 0:
                        start = max(0, i - self.before_context)
                        for j in range(start, i):
                            result["before_context"].append({
                                "line": j + 1,
                                "content": lines[j]
                            })

                    # 添加下文 (After context)
                    if self.after_context > 0:
                        end = min(total_lines, i + self.after_context + 1)
                        for j in range(i + 1, end):
                            result["after_context"].append({
                                "line": j + 1,
                                "content": lines[j]
                            })

                    results.append(result)
        else:
            # 使用普通文本搜索
            search_lower = self.search_text.lower()
            for i, line in enumerate(lines):
                if search_lower in line.lower():
                    result = {
                        "pod": pod_name,
                        "line_number": i + 1,
                        "content": line,
                        "before_context": [],
                        "after_context": []
                    }

                    # 添加上文 (Before context)
                    if self.before_context > 0:
                        start = max(0, i - self.before_context)
                        for j in range(start, i):
                            result["before_context"].append({
                                "line": j + 1,
                                "content": lines[j]
                            })

                    # 添加下文 (After context)
                    if self.after_context > 0:
                        end = min(total_lines, i + self.after_context + 1)
                        for j in range(i + 1, end):
                            result["after_context"].append({
                                "line": j + 1,
                                "content": lines[j]
                            })

                    results.append(result)

        return results

    def highlight_match(self, text: str) -> str:
        """高亮显示匹配的文本"""
        if self.use_regex:
            try:
                return self.compiled_pattern.sub(
                    f"{Colors.RED}{Colors.BOLD}\\g<0>{Colors.END}",
                    text
                )
            except:
                return text
        else:
            try:
                pattern = re.compile(f"({re.escape(self.search_text)})", re.IGNORECASE)
                return pattern.sub(
                    f"{Colors.RED}{Colors.BOLD}\\1{Colors.END}",
                    text
                )
            except:
                return text

    def process_pod(self, pod_name: str):
        """处理单个Pod的搜索"""
        logs = self.get_pod_logs(pod_name)
        if logs is None:
            return pod_name, [], 0

        results = self.search_in_text(logs, pod_name)
        return pod_name, results, len(logs.splitlines())

    def search_all_pods(self, pods: list):
        """并行搜索所有Pod的日志"""
        print(f"{Colors.BLUE}开始并行搜索 (使用 {self.threads} 个线程)...{Colors.END}")
        print(f"{Colors.YELLOW}搜索内容: {self.search_text}{Colors.END}")
        if self.use_regex:
            print(f"{Colors.YELLOW}使用正则表达式模式{Colors.END}")
        if self.tail_lines > 0:
            print(f"{Colors.YELLOW}只搜索最后 {self.tail_lines} 行日志{Colors.END}")

        if self.before_context > 0 or self.after_context > 0:
            if self.before_context == self.after_context:
                print(f"{Colors.YELLOW}显示上下文: 前后各 {self.before_context} 行{Colors.END}")
            else:
                if self.before_context > 0:
                    print(f"{Colors.YELLOW}显示上文: {self.before_context} 行{Colors.END}")
                if self.after_context > 0:
                    print(f"{Colors.YELLOW}显示下文: {self.after_context} 行{Colors.END}")

        print()

        total_results = []
        total_matches = 0
        pods_with_matches = 0

        start_time = time.time()
        processed_count = 0

        print(f"{Colors.CYAN}正在搜索Pod日志...{Colors.END}")

        with ThreadPoolExecutor(max_workers=self.threads) as executor:
            future_to_pod = {}
            for pod in pods:
                future = executor.submit(self.process_pod, pod)
                future_to_pod[future] = pod

            for future in as_completed(future_to_pod):
                pod_name = future_to_pod[future]
                try:
                    pod_name, results, log_lines = future.result(timeout=self.timeout)

                    processed_count += 1
                    progress = processed_count / len(pods) * 100
                    sys.stdout.write(f"\r{Colors.CYAN}进度: [{processed_count}/{len(pods)}] {progress:.1f}%{Colors.END}")
                    sys.stdout.flush()

                    if results:
                        pods_with_matches += 1
                        total_matches += len(results)

                        print(f"\n{Colors.YELLOW}{'='*80}{Colors.END}")
                        print(f"{Colors.GREEN}[Pod: {pod_name}] 找到 {len(results)} 处匹配 (共{log_lines}行日志):{Colors.END}")

                        for idx, result in enumerate(results):
                            print(f"\n{Colors.MAGENTA}匹配 #{idx+1}:{Colors.END}")

                            if result["before_context"]:
                                print(f"{Colors.CYAN}上文 (前 {len(result['before_context'])} 行):{Colors.END}")
                                for ctx in result["before_context"]:
                                    print(f"  {Colors.BLUE}第 {ctx['line']:4d} 行:{Colors.END} {ctx['content']}")

                            print(f"{Colors.RED}{Colors.BOLD}>>> 第 {result['line_number']:4d} 行:{Colors.END} {self.highlight_match(result['content'])}")

                            if result["after_context"]:
                                print(f"{Colors.CYAN}下文 (后 {len(result['after_context'])} 行):{Colors.END}")
                                for ctx in result["after_context"]:
                                    print(f"  {Colors.BLUE}第 {ctx['line']:4d} 行:{Colors.END} {ctx['content']}")

                        total_results.extend(results)

                except Exception as e:
                    print(f"\n{Colors.RED}处理Pod {pod_name} 时出错: {e}{Colors.END}")

        elapsed_time = time.time() - start_time

        print(f"\n\n{Colors.YELLOW}{'='*80}{Colors.END}")
        print(f"{Colors.BLUE}搜索完成!{Colors.END}")
        print(f"{Colors.GREEN}总耗时: {elapsed_time:.2f} 秒{Colors.END}")
        print(f"{Colors.GREEN}已检查 Pod 数量: {len(pods)}{Colors.END}")
        print(f"{Colors.GREEN}找到匹配的 Pod 数量: {pods_with_matches}{Colors.END}")
        print(f"{Colors.GREEN}总匹配行数: {total_matches}{Colors.END}")

        if total_results:
            self.save_results(total_results)

        return total_results

    def save_results(self, results: list):
        """保存搜索结果到文件"""
        timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
        filename = f"search_results_{self.deployment}_{timestamp}.log"

        try:
            with open(filename, 'w', encoding='utf-8') as f:
                f.write(f"搜索时间: {datetime.now()}\n")
                f.write(f"Deployment: {self.deployment}\n")
                f.write(f"命名空间: {self.namespace}\n")
                f.write(f"搜索内容: {self.search_text}\n")
                f.write(f"使用正则: {self.use_regex}\n")
                if self.before_context > 0:
                    f.write(f"显示上文: {self.before_context} 行\n")
                if self.after_context > 0:
                    f.write(f"显示下文: {self.after_context} 行\n")
                f.write(f"搜索最后 {self.tail_lines} 行日志\n")
                f.write("=" * 80 + "\n\n")

                current_pod = None
                for result_idx, result in enumerate(results):
                    if result["pod"] != current_pod:
                        current_pod = result["pod"]
                        f.write(f"\n{'='*80}\n")
                        f.write(f"Pod: {current_pod}\n")
                        f.write(f"{'='*80}\n\n")

                    f.write(f"匹配 #{result_idx+1}:\n")

                    if result["before_context"]:
                        f.write(f"上文 (前 {len(result['before_context'])} 行):\n")
                        for ctx in result["before_context"]:
                            f.write(f"  第 {ctx['line']:4d} 行: {ctx['content']}\n")

                    f.write(f">>> 第 {result['line_number']:4d} 行: {result['content']}\n")

                    if result["after_context"]:
                        f.write(f"下文 (后 {len(result['after_context'])} 行):\n")
                        for ctx in result["after_context"]:
                            f.write(f"  第 {ctx['line']:4d} 行: {ctx['content']}\n")

                    f.write("\n")

            print(f"{Colors.GREEN}搜索结果已保存到: {filename}{Colors.END}")
        except Exception as e:
            print(f"{Colors.YELLOW}警告: 保存结果到文件失败: {e}{Colors.END}")

def signal_handler(sig, frame):
    """处理Ctrl+C信号"""
    print(f"\n\n{Colors.YELLOW}接收到中断信号，正在退出...{Colors.END}")
    sys.exit(0)

def main():
    """主函数"""
    signal.signal(signal.SIGINT, signal_handler)

    parser = argparse.ArgumentParser(
        description="Kubernetes Deployment日志搜索工具",
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog="""
示例:
  %(prog)s deployment名称 -n 命名空间 -s "error"
  %(prog)s deployment名称 -n 命名空间 -s "404" -t 20
  %(prog)s deployment名称 -n 命名空间 -s "timeout" -B 5 -A 10
  %(prog)s deployment名称 -n 命名空间 -r "error|exception|fatal" -c 3
  %(prog)s deployment名称 -n 命名空间 -s "warning" -T 5000 -B 100 -A 100
        """
    )

    parser.add_argument(
        "deployment",
        help="Kubernetes Deployment名称"
    )

    parser.add_argument(
        "-n", "--namespace",
        help="命名空间 (默认: default)"
    )

    parser.add_argument(
        "-s", "--search-text",
        required=True,
        help="要搜索的日志内容"
    )

    parser.add_argument(
        "-t", "--threads",
        type=int,
        default=10,
        help="并发线程数 (默认: 10)"
    )

    parser.add_argument(
        "-B", "--before-context",
        type=int,
        default=0,
        help="显示匹配行之前的行数 (上文)"
    )

    parser.add_argument(
        "-A", "--after-context",
        type=int,
        default=0,
        help="显示匹配行之后的行数 (下文)"
    )

    parser.add_argument(
        "-c", "--context",
        type=int,
        default=0,
        help="显示匹配行的上下文行数 (前后各c行，会被-B和-A覆盖)"
    )

    parser.add_argument(
        "-T", "--tail",
        type=int,
        default=999999999,
        help="只搜索最后N行日志 (默认: 1000)"
    )

    parser.add_argument(
        "-r", "--regex",
        action="store_true",
        help="使用正则表达式搜索"
    )

    parser.add_argument(
        "--timeout",
        type=int,
        default=30,
        help="单个Pod日志获取超时时间(秒) (默认: 30)"
    )

    args = parser.parse_args()

    try:
        result = subprocess.run(
            ["kubectl", "version", "--client"],
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
            universal_newlines=True
        )
        if result.returncode != 0:
            print(f"{Colors.RED}错误: kubectl配置不正确: {result.stderr}{Colors.END}")
            sys.exit(1)
    except FileNotFoundError:
        print(f"{Colors.RED}错误: 未找到kubectl命令{Colors.END}")
        sys.exit(1)

    searcher = LogSearcher(args)
    pods = searcher.get_deployment_pods()
    searcher.search_all_pods(pods)

if __name__ == "__main__":
    main()

（2）赋予执行权限

chmod +x s

（3）测试使用

s --help

3、参数详解

参数	简写	说明	默认值
--search-text	-s	要搜索的文本内容（必需）	无
--namespace	-n	Kubernetes命名空间	default
--threads	-t	并发线程数	10
--before-context	-B	显示匹配行之前的行数	0
--after-context	-A	显示匹配行之后的行数	0
--context	-c	显示匹配行前后各N行	0
--tail	-T	只搜索最后N行日志	1000
--regex	-r	使用正则表达式搜索	false
--timeout	无	单个Pod超时时间(秒)	30

4、使用案例

（1）查询指定日志

s nginx-feikong -n app -s "404"

（2）使用正则表达式搜索

s nginx-feikong -n app -s "(error|exception|fatal)" -r

（3）显示匹配行的上下文（前后各3行）

s nginx-feikong -n app -s "404" -c 3

（4）只搜索最近1000行日志

s nginx-feikong -n app -s "404" -T 1000

（5）增加并发线程数（适合Pod数量多的情况）

s nginx-feikong -n app -s "404" -t 20