Kubernetes 集群与 Pod 状态巡检笔记

本文将完整巡检脚本和使用说明集中保存，适用于生产环境的只读状态巡检。执行前请确认目标集群、账号权限和检查范围。

阅读导航

• 完整脚本
• 使用说明

完整脚本

将下面代码完整保存为 k8s-inspect.sh，使用 Bash 执行。脚本内部使用 Python 标准库解析 Kubernetes JSON，避免通过文本列或关键字匹配误判状态。依赖 Bash、Python 3.8+ 和 kubectl，不需要安装第三方 Python 包。

#!/usr/bin/env bash
# 只读 Kubernetes 巡检；需要 bash、python3 >= 3.8、kubectl。
set -euo pipefail
umask 077
command -v python3 >/dev/null 2>&1 || { echo '错误：需要 python3 >= 3.8' >&2; exit 3; }
exec python3 - "$@" <<'PY'
import argparse
import collections
import datetime as dt
import json
import os
import re
import shutil
import subprocess
import sys
import time
from decimal import Decimal, InvalidOperation

UTC = dt.timezone.utc


def timestamp(value):
    if not value:
        return None
    try:
        return dt.datetime.fromisoformat(value.replace('Z', '+00:00')).astimezone(UTC)
    except (ValueError, TypeError):
        return None


def quantity(value):
    """Kubernetes Quantity -> Decimal 基本单位（CPU 核，内存字节）。"""
    match = re.fullmatch(r'([+-]?(?:\d+(?:\.\d*)?|\.\d+)(?:[eE][+-]?\d+)?)([a-zA-Z]*)', str(value))
    if not match:
        raise ValueError('无法解析资源数量')
    factors = {'': 1, 'n': Decimal('1e-9'), 'u': Decimal('1e-6'), 'm': Decimal('.001'),
               'k': 1000, 'K': 1000, 'M': 1000**2, 'G': 1000**3, 'T': 1000**4,
               'P': 1000**5, 'E': 1000**6,
               **{s + 'i': 1024**i for i, s in enumerate('KMGTPE', 1)}}
    if match[2] not in factors:
        raise ValueError('未知资源单位')
    return Decimal(match[1]) * factors[match[2]]


def clean(value):
    # 避免终端控制字符；不把任意事件消息、日志或容器环境变量写入报告。
    return re.sub(r'[\x00-\x1f\x7f-\x9f]', '?', str(value))[:2000]


class Inspector:
    def __init__(self, args):
        self.a = args
        self.now = dt.datetime.now(UTC)
        self.findings = []
        self.coverage = []
        self.counts = {}
        self.start = time.monotonic()
        self.context = args.context
        self.base = ['kubectl', '--request-timeout=' + str(args.request_timeout) + 's']
        if args.kubeconfig:
            self.base += ['--kubeconfig', args.kubeconfig]
        self.scope = ['-n', args.namespace] if args.namespace else ['-A']

    def add(self, level, kind, obj, detail):
        self.findings.append(dict(level=level, kind=kind, object=clean(obj), detail=clean(detail)))

    def ident(self, obj):
        m = obj.get('metadata', {})
        return (m.get('namespace', '') + '/' + m.get('name', '?')).lstrip('/')

    def age(self, value):
        t = timestamp(value)
        return max(0, (self.now - t).total_seconds()) if t else None

    def old(self, value, seconds):
        age = self.age(value)
        return age is not None and age >= seconds

    def run(self, argv):
        remaining = self.a.total_timeout - (time.monotonic() - self.start)
        if remaining <= 0:
            raise RuntimeError('巡检总采集时限已到')
        try:
            p = subprocess.run(self.base + argv, stdout=subprocess.PIPE, stderr=subprocess.PIPE,
                               timeout=min(self.a.command_timeout, remaining), check=False)
        except subprocess.TimeoutExpired:
            raise RuntimeError('命令超时') from None
        except OSError:
            raise RuntimeError('kubectl 无法执行') from None
        if p.returncode:
            # 不回显 stderr：认证插件、代理可能把凭据写入错误消息。
            err = p.stderr.decode('utf-8', errors='replace')
            category = next((label for token, label in [
                ('Forbidden', 'RBAC 权限不足'), ('Unauthorized', '认证失败'),
                ('NotFound', 'API/资源不存在'), ('could not find the requested resource', 'API/资源不存在'),
                ('x509', 'TLS 证书错误'), ('connection refused', '连接被拒绝'),
                ('i/o timeout', '网络超时')] if token in err), '请求失败（原始错误未写入报告）')
            raise RuntimeError(f'{category}，退出码 {p.returncode}')
        return p.stdout.decode('utf-8')

    def collect(self, name, resource=None, path=None, extra=None, scoped=True):
        try:
            cmd = ['get', '--raw', path] if path else ['get', resource] + (self.scope if scoped else []) + [
                '-o', 'json', '--chunk-size=' + str(self.a.chunk_size)] + (extra or [])
            data = json.loads(self.run(cmd))
            if not isinstance(data, dict) or not isinstance(data.get('items'), list):
                raise ValueError('缺少 items 数组')
            self.coverage.append(dict(check=name, status='OK', count=len(data['items'])))
            self.counts[name] = len(data['items'])
            return data['items']
        except (RuntimeError, ValueError, UnicodeError) as exc:
            self.coverage.append(dict(check=name, status='UNKNOWN'))
            detail = str(exc) if isinstance(exc, RuntimeError) else '返回数据格式异常'
            self.add('UNKNOWN', '采集', name, detail)
            return None

    def nodes(self, items):
        for obj in items:
            key = self.ident(obj)
            status = obj.get('status', {})
            cond = {c['type']: c for c in status.get('conditions', [])}
            if cond.get('Ready', {}).get('status') != 'True':
                self.add('CRITICAL', 'Node', key, 'Ready=' + cond.get('Ready', {}).get('status', '缺失'))
            for name in ['MemoryPressure', 'DiskPressure', 'PIDPressure', 'NetworkUnavailable']:
                c = cond.get(name, {})
                if c.get('status') == 'True':
                    self.add('CRITICAL', 'Node', key, f'{name}=True reason={c.get("reason", "")}')
                elif c.get('status') == 'Unknown':
                    self.add('WARNING', 'Node', key, f'{name}=Unknown')
            if obj.get('spec', {}).get('unschedulable'):
                self.add('WARNING', 'Node', key, '已 cordon，禁止调度（可能是计划维护）')
        if not items:
            self.add('CRITICAL', 'Node', '集群', '未发现任何节点')

    def pods(self, items):
        severe_wait = {'CrashLoopBackOff', 'ImagePullBackOff', 'ErrImagePull', 'CreateContainerConfigError',
                       'CreateContainerError', 'RunContainerError', 'InvalidImageName', 'ErrImageNeverPull'}
        for obj in items:
            key = self.ident(obj)
            m, s = obj.get('metadata', {}), obj.get('status', {})
            phase = s.get('phase', 'Unknown')
            if m.get('deletionTimestamp'):
                grace = obj.get('spec', {}).get('terminationGracePeriodSeconds', 30)
                if self.old(m['deletionTimestamp'], max(self.a.terminating_seconds, grace)):
                    self.add('WARNING', 'Pod', key, '长时间 Terminating；检查 finalizer、节点与存储卸载')
                continue
            if phase == 'Succeeded':
                continue
            if phase in ('Failed', 'Unknown'):
                self.add('CRITICAL', 'Pod', key, f'phase={phase} reason={s.get("reason", "")} node={obj.get("spec", {}).get("nodeName", "未调度")}')
            age = self.age(m.get('creationTimestamp'))
            mature = age is not None and age >= self.a.pending_seconds
            cond = {c['type']: c for c in s.get('conditions', [])}
            if phase == 'Pending' and mature:
                self.add('WARNING', 'Pod', key, f'Pending 已持续至少 {int(age)} 秒（Pod 年龄）；PodScheduled={cond.get("PodScheduled", {}).get("status", "缺失")}')
            ready = cond.get('Ready', {})
            if phase == 'Running' and ready.get('status') != 'True':
                # 优先使用本次 Ready 变化时间，避免老 Pod 短暂探针失败立即报警。
                since = ready.get('lastTransitionTime') or m.get('creationTimestamp')
                if self.old(since, self.a.notready_seconds):
                    self.add('WARNING', 'Pod', key, f'Running 但 Ready={ready.get("status", "缺失")} reason={ready.get("reason", "")}')
            for field in ('initContainerStatuses', 'containerStatuses', 'ephemeralContainerStatuses'):
                for c in s.get(field, []):
                    target = key + ':' + c.get('name', '?')
                    state = c.get('state', {})
                    waiting = state.get('waiting', {})
                    reason = waiting.get('reason', '')
                    if reason in severe_wait:
                        self.add('CRITICAL', 'Container', target, f'{field}: waiting={reason}')
                    elif waiting and mature:
                        self.add('WARNING', 'Container', target, f'{field}: waiting={reason or "未知"}')
                    terminated = state.get('terminated', {})
                    if terminated and (terminated.get('exitCode', 0) != 0 or terminated.get('reason') == 'OOMKilled'):
                        self.add('CRITICAL', 'Container', target, f'当前终止 reason={terminated.get("reason", "")} exitCode={terminated.get("exitCode")}')
                    last = c.get('lastState', {}).get('terminated', {})
                    if last and self.age(last.get('finishedAt')) is not None and not self.old(last.get('finishedAt'), self.a.history_minutes * 60):
                        if last.get('exitCode', 0) != 0 or last.get('reason') == 'OOMKilled':
                            self.add('WARNING', 'Container', target, f'近期上次终止 reason={last.get("reason", "")} exitCode={last.get("exitCode")} finishedAt={last.get("finishedAt")}')
                    if c.get('restartCount', 0) >= self.a.restart_threshold:
                        self.add('WARNING', 'Container', target, f'生命周期累计重启={c["restartCount"]}（不是时间窗口增量）')

    def workloads(self, kind, items):
        for o in items:
            key = self.ident(o)
            spec, s, m = o.get('spec', {}), o.get('status', {}), o.get('metadata', {})
            if m.get('deletionTimestamp'):
                continue
            if kind == 'Deployment' and spec.get('paused'):
                self.add('WARNING', kind, key, 'Deployment 已暂停更新')
            if s.get('observedGeneration', 0) < m.get('generation', 0):
                self.add('WARNING', kind, key, '控制器尚未观察到最新 generation')
            if kind in ('Deployment', 'StatefulSet'):
                desired = spec.get('replicas', 1)
                ready = s.get('readyReplicas', 0)
                if ready < desired:
                    self.add('CRITICAL' if ready == 0 else 'WARNING', kind, key, f'ready={ready}/{desired}（发布或扩容期间也可能出现）')
                if kind == 'Deployment' and s.get('updatedReplicas', 0) < desired:
                    self.add('WARNING', kind, key, f'updated={s.get("updatedReplicas", 0)}/{desired}')
                if kind == 'Deployment' and s.get('availableReplicas', 0) < desired:
                    self.add('WARNING', kind, key, f'available={s.get("availableReplicas", 0)}/{desired}')
                if kind == 'StatefulSet' and spec.get('updateStrategy', {}).get('type', 'RollingUpdate') == 'RollingUpdate':
                    partition = spec.get('updateStrategy', {}).get('rollingUpdate', {}).get('partition', 0)
                    if s.get('updatedReplicas', 0) < max(0, desired - partition):
                        self.add('WARNING', kind, key, '更新副本未达到 RollingUpdate partition 要求')
            elif kind == 'DaemonSet':
                desired, ready = s.get('desiredNumberScheduled', 0), s.get('numberReady', 0)
                if ready < desired:
                    self.add('CRITICAL' if ready == 0 else 'WARNING', kind, key, f'ready={ready}/{desired}')
                if s.get('numberUnavailable', 0) or s.get('numberMisscheduled', 0):
                    self.add('WARNING', kind, key, f'unavailable={s.get("numberUnavailable", 0)} misscheduled={s.get("numberMisscheduled", 0)}')
                if spec.get('updateStrategy', {}).get('type', 'RollingUpdate') == 'RollingUpdate' and s.get('updatedNumberScheduled', 0) < desired:
                    self.add('WARNING', kind, key, f'updated={s.get("updatedNumberScheduled", 0)}/{desired}')
            for c in s.get('conditions', []):
                if (c.get('type'), c.get('status')) in [('Progressing', 'False'), ('ReplicaFailure', 'True')]:
                    self.add('CRITICAL', kind, key, f'{c["type"]}={c["status"]} reason={c.get("reason", "")}')

    def jobs(self, items):
        for o in items:
            s = o.get('status', {})
            for c in s.get('conditions', []):
                if c.get('type') in ('Failed', 'FailureTarget') and c.get('status') == 'True':
                    self.add('CRITICAL', 'Job', self.ident(o), f'{c["type"]} reason={c.get("reason", "")}')
            if s.get('failed', 0) and not any(c.get('type') in ('Failed', 'Complete', 'FailureTarget') and c.get('status') == 'True' for c in s.get('conditions', [])):
                self.add('WARNING', 'Job', self.ident(o), f'失败尝试={s["failed"]} active={s.get("active", 0)}，可能仍在重试')

    def pvcs(self, items):
        for o in items:
            phase = o.get('status', {}).get('phase', 'Unknown')
            if phase == 'Lost':
                self.add('CRITICAL', 'PVC', self.ident(o), 'phase=Lost')
            elif phase != 'Bound' and self.old(o.get('metadata', {}).get('creationTimestamp'), self.a.pending_seconds):
                self.add('WARNING', 'PVC', self.ident(o), f'phase={phase}；WaitForFirstConsumer 可能是预期状态')
            for c in o.get('status', {}).get('conditions', []):
                if c.get('status') == 'True':
                    self.add('WARNING', 'PVC', self.ident(o), f'condition={c.get("type")} reason={c.get("reason", "")}')

    def events(self, items):
        for o in items:
            latest = o.get('series', {}).get('lastObservedTime') or o.get('lastTimestamp') or o.get('eventTime') or o.get('metadata', {}).get('creationTimestamp')
            age = self.age(latest)
            if age is None:
                self.add('UNKNOWN', 'Event', self.ident(o), '无法确定事件时间')
            elif age <= self.a.history_minutes * 60:
                ref = o.get('involvedObject', {})
                target = '/'.join(filter(None, [ref.get('namespace'), ref.get('kind'), ref.get('name')]))
                self.add('WARNING', 'Event', target, f'reason={o.get("reason", "")} count={o.get("series", {}).get("count", o.get("count", 1))} last={latest}（可能已恢复）')

    def metrics(self, nodes, pods):
        # 使用结构化 Metrics API，避免解析 kubectl top 的列格式。
        for kind, source, path in [
            ('Node', nodes, '/apis/metrics.k8s.io/v1beta1/nodes'),
            ('Pod', pods, '/apis/metrics.k8s.io/v1beta1/' + ('namespaces/' + self.a.namespace + '/pods' if self.a.namespace else 'pods'))]:
            if source is None:
                continue
            items = self.collect(kind + 'Metrics', path=path)
            if items is None:
                continue
            lookup = {self.ident(o): o for o in source}
            seen = set()
            for metric in items:
                key = self.ident(metric)
                if key not in lookup:
                    continue
                seen.add(key)
                if timestamp(metric.get('timestamp')) is None or self.old(metric.get('timestamp'), self.a.metrics_max_age):
                    self.add('UNKNOWN', kind + 'Metrics', key, '指标时间缺失或已过期')
                    continue
                obj = lookup[key]
                if kind == 'Node':
                    self.check_usage(kind, key, metric.get('usage', {}), obj.get('status', {}).get('allocatable', {}))
                else:
                    specs = {c['name']: c for field in ('containers', 'initContainers', 'ephemeralContainers') for c in obj.get('spec', {}).get(field, [])}
                    for c in metric.get('containers', []):
                        limits = specs.get(c['name'], {}).get('resources', {}).get('limits', {})
                        self.check_usage('Container', key + ':' + c['name'], c.get('usage', {}), limits)
            for key, obj in lookup.items():
                eligible = kind == 'Node' or (obj.get('status', {}).get('phase') == 'Running' and not obj.get('metadata', {}).get('deletionTimestamp') and self.old(obj.get('status', {}).get('startTime') or obj.get('metadata', {}).get('creationTimestamp'), self.a.pending_seconds))
                if eligible and key not in seen:
                    self.add('UNKNOWN', kind + 'Metrics', key, '缺少资源使用指标')

    def check_usage(self, kind, key, usage, capacity):
        for resource in ('cpu', 'memory'):
            if resource not in capacity:
                continue  # 容器未配置 limit，不把缺少 limit 当成超限。
            try:
                cap = quantity(capacity[resource])
                used = quantity(usage[resource])
                if cap <= 0:
                    continue
                percent = float(used / cap * 100)
                if percent >= self.a.usage_percent:
                    base = 'allocatable' if kind == 'Node' else 'limit'
                    self.add('WARNING', kind + 'Usage', key, f'{resource}={percent:.1f}% of {base}，单次采样')
            except (ValueError, KeyError, InvalidOperation):
                self.add('UNKNOWN', kind + 'Metrics', key, f'{resource} 指标缺失或无法解析')

    def execute(self):
        if not self.context:
            try:
                self.context = self.run(['config', 'current-context']).strip()
                if not self.context:
                    raise RuntimeError('当前 context 为空')
            except RuntimeError as exc:
                self.add('UNKNOWN', '连接', 'context', str(exc))
                return
        # 固定本次 context，后续不修改 kubeconfig，不受 current-context 被切换影响。
        self.base += ['--context', self.context]
        try:
            result = self.run(['get', '--raw', '/readyz']).strip()
            if result != 'ok':
                raise RuntimeError('readyz 未返回 ok')
            self.coverage.append(dict(check='APIServerReadyz', status='OK'))
        except RuntimeError as exc:
            self.add('UNKNOWN', 'APIServer', '/readyz', str(exc))
            self.coverage.append(dict(check='APIServerReadyz', status='UNKNOWN'))
        nodes = None
        if not self.a.skip_cluster_checks:
            nodes = self.collect('Nodes', 'nodes', scoped=False)
            if nodes is not None:
                self.nodes(nodes)
        else:
            self.coverage.append(dict(check='Nodes', status='SKIPPED'))
        pods = self.collect('Pods', 'pods')
        if pods is not None:
            self.pods(pods)
        for name, res, handler in [
            ('Deployment', 'deployments.apps', None), ('StatefulSet', 'statefulsets.apps', None),
            ('DaemonSet', 'daemonsets.apps', None), ('Job', 'jobs.batch', self.jobs),
            ('PVC', 'persistentvolumeclaims', self.pvcs),
            ('WarningEvents', 'events', self.events)]:
            items = self.collect(name, res, extra=['--field-selector=type=Warning'] if name == 'WarningEvents' else None)
            if items is not None:
                if handler:
                    handler(items)
                else:
                    self.workloads(name, items)
        if self.a.metrics:
            self.metrics(nodes, pods)
        else:
            self.coverage.append(dict(check='Metrics', status='SKIPPED'))

    def report(self):
        counts = collections.Counter(f['level'] for f in self.findings)
        # 采集不完整优先用 3；报告同时保留已经发现的 CRITICAL。
        code = 3 if counts['UNKNOWN'] else 2 if counts['CRITICAL'] else 1 if counts['WARNING'] else 0
        order = {'CRITICAL': 0, 'WARNING': 1, 'UNKNOWN': 2}
        self.findings.sort(key=lambda f: (order[f['level']], f['kind'], f['object']))
        report = dict(schema_version=1, started_at=self.now.isoformat(), finished_at=dt.datetime.now(UTC).isoformat(),
                      context=self.context, namespace=self.a.namespace or 'ALL', exit_code=code,
                      summary={k: counts[k] for k in ('CRITICAL', 'WARNING', 'UNKNOWN')},
                      coverage=self.coverage, object_counts=self.counts, findings=self.findings)
        lines = ['Kubernetes 只读巡检报告', f'时间(UTC): {report["started_at"]}',
                 f'Context: {clean(self.context)} | 命名空间: {clean(report["namespace"])}',
                 f'退出码: {code} | CRITICAL={counts["CRITICAL"]} WARNING={counts["WARNING"]} UNKNOWN={counts["UNKNOWN"]}',
                 '范围: API 状态快照；不等同于业务、网络、存储和控制平面完整健康验收。', '', '采集覆盖:']
        for c in self.coverage:
            lines.append(f'  {c["check"]}: {c["status"]}' + (f' ({c["count"]} 个对象)' if 'count' in c else ''))
        lines += ['', '异常明细（不截断条目）:']
        lines += [f'[{f["level"]}] {f["kind"]} {f["object"]} | {f["detail"]}' for f in self.findings]
        if not self.findings:
            lines.append('已启用且成功采集的检查项未发现异常。')
        content = '\n'.join(lines) + '\n'
        if self.a.output_dir:
            os.makedirs(self.a.output_dir, mode=0o700, exist_ok=True)
            # 独占创建；不覆盖已有文件或跟随同名文件符号链接。
            stem = 'k8s-inspect-' + self.now.strftime('%Y%m%dT%H%M%S%fZ') + '-' + str(os.getpid())
            for suffix, body in [('.txt', content), ('.json', json.dumps(report, ensure_ascii=False, indent=2) + '\n')]:
                path = os.path.join(self.a.output_dir, stem + suffix)
                fd = os.open(path, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600)
                with os.fdopen(fd, 'w', encoding='utf-8') as stream:
                    stream.write(body)
                print('报告已保存: ' + path, file=sys.stderr)
        print(content, end='')
        return code


def positive(value):
    number = int(value)
    if number <= 0:
        raise argparse.ArgumentTypeError('必须为正整数')
    return number


def main():
    p = argparse.ArgumentParser(prog='k8s-inspect.sh', description='只读 K8s 巡检；退出码 0=无异常 1=告警 2=严重 3=采集/执行不完整。')
    p.add_argument('--context', default='', help='指定 kubeconfig context；默认固定当前 context')
    p.add_argument('--kubeconfig', help='kubeconfig 文件路径')
    p.add_argument('-n', '--namespace', help='仅检查指定命名空间；默认所有命名空间')
    p.add_argument('--skip-cluster-checks', action='store_true', help='跳过集群节点检查；仍检查 API readyz')
    p.add_argument('--metrics', action='store_true', help='启用 Metrics API；不可用时报告 UNKNOWN')
    p.add_argument('--output-dir', help='保存 TXT/JSON 报告的目录')
    for flag, default, help_text in [
        ('request-timeout', 15, '单次 API 请求超时秒数'), ('command-timeout', 60, '单个 kubectl 总超时秒数'),
        ('total-timeout', 600, '全部采集总时限秒数'), ('chunk-size', 200, '资源列表分页大小'),
        ('pending-seconds', 300, 'Pending、容器等待、PVC 非 Bound 的年龄阈值'),
        ('notready-seconds', 300, 'Running Pod 持续不 Ready 阈值'),
        ('terminating-seconds', 600, '终止超时阈值（兼顾配置的优雅退出期）'),
        ('restart-threshold', 5, '单容器生命周期累计重启阈值'),
        ('history-minutes', 60, '近期 Warning 事件和上次异常终止的回看分钟数'),
        ('usage-percent', 85, 'CPU/内存使用率告警百分比'), ('metrics-max-age', 180, '指标最大年龄秒数')]:
        p.add_argument('--' + flag, type=positive, default=default, help=help_text + f'；默认 {default}')
    a = p.parse_args()
    if a.namespace and not re.fullmatch(r'[a-z0-9](?:[-a-z0-9]{0,61}[a-z0-9])?', a.namespace):
        p.error('namespace 格式无效')
    if a.usage_percent > 100:
        p.error('usage-percent 必须在 1~100 之间')
    if not shutil.which('kubectl'):
        print('错误：找不到 kubectl', file=sys.stderr)
        return 3
    inspector = Inspector(a)
    inspector.execute()
    return inspector.report()


if __name__ == '__main__':
    try:
        sys.exit(main())
    except KeyboardInterrupt:
        print('巡检已中断', file=sys.stderr)
        sys.exit(3)
    except (OSError, ValueError, TypeError, KeyError):
        # 不打印 traceback 或任意外部错误，避免报告携带凭据。
        print('巡检执行或报告写入失败，请检查输入数据、目录权限和磁盘空间；结果不完整。', file=sys.stderr)
        sys.exit(3)
PY

使用说明

脚本：k8s-inspect.sh。需要 Bash、Python 3.8+、kubectl，以及可访问目标集群的 kubeconfig。kubectl 版本应与集群版本兼容。

这是单次 API 状态巡检，适合作为人工巡检或定时任务的基础。不是业务可用性验证，也不能保证发现所有故障。首次上线建议人工执行，结合维护窗口、业务特点调整阈值。

1. 直接使用

chmod 750 k8s-inspect.sh

# 显式指定集群，检查所有命名空间，终端显示所有异常并保存 TXT/JSON。
./k8s-inspect.sh --context prod-cluster --output-dir ./reports

# 指定 kubeconfig。
./k8s-inspect.sh --kubeconfig /path/to/kubeconfig \
  --context prod-cluster --output-dir ./reports

# 只检查一个命名空间；仍然检查所有节点和 API readyz。
./k8s-inspect.sh --context prod-cluster -n production --output-dir ./reports

# 命名空间账号没有节点读取权限时，明确跳过节点检查。
# 此时报告中的 Nodes 为 SKIPPED，不能理解为节点正常。
./k8s-inspect.sh --context prod-cluster -n production \
  --skip-cluster-checks --output-dir ./reports

# 开启 CPU / 内存使用率检查（需要 Metrics API 及权限）。
./k8s-inspect.sh --context prod-cluster --metrics --output-dir ./reports

./k8s-inspect.sh --help

不要使用 sudo，除非本机凭据访问确实需要；通常普通用户即可执行。脚本不会切换或修改 current-context，每次运行固定所选 context。默认未指定 --context 时读取当前 context，生产环境建议始终显式指定。

2. 能发现哪些异常

Pod Succeeded 不作为异常。副本期望为 0 不单独告警。已进入删除流程的 Pod 只检查删除是否超时，避免正常退出被当成故障。

事件和异常不截断条目，也不默认排除 kube-system。各检查独立采集，因此结果不是同一时刻的原子快照，发布、扩容过程中可能看到暂时不一致。

3. 默认阈值与调整

./k8s-inspect.sh --context prod-cluster \
  --pending-seconds 600 --notready-seconds 300 \
  --restart-threshold 10 --history-minutes 30 \
  --metrics --usage-percent 90 --output-dir ./reports

Pending 使用 Pod 创建时间作近似；NotReady 优先使用 Ready 条件的 lastTransitionTime。不提供跨次状态缓存，因此无法判断“最近 N 分钟重启次数”。请用 Prometheus 等时序监控计算速率。

工作负载副本不足按当前状态报告，不设置发布宽限期；维护或滚动更新期间需要人工结合发布状态判断。长优雅退出的 Pod 会在配置阈值与优雅退出时间两者较大者之后告警。

4. 结果与退出码

参数使用错误由参数解析器返回 2，并在 stderr 输出用法；因此接入告警前应先验证固定的调用参数。

TXT 方便阅读，JSON 含 summary、coverage、object_counts、findings，方便自动处理。summary 统计的是异常条目数，同一 Pod 可以有多条异常，不是异常 Pod 个数。

默认不启用 Metrics，coverage 会显示 SKIPPED；开启后 Metrics API 缺失、无权限、指标缺失或过期会标记 UNKNOWN。未配置容器 limit 的资源无法计算其占 limit 比例，跳过该比例判断；这不代表资源使用一定安全。指标为单次采样，不能作为持续高负载结论。

5. 权限与访问范围

所有 Kubernetes 操作均为读取：

• 当前 context：本地 kubectl config current-context；显式指定时无需读取。
• 非资源 URL：GET /readyz。
• 核心 API：list nodes、pods、persistentvolumeclaims、events。
• apps API：list deployments、statefulsets、daemonsets。
• batch API：list jobs。
• 开启 Metrics 后：metrics.k8s.io 的 nodes、pods 列表读取。

所有命名空间检查需要相应集群范围授权。-n 限制命名空间资源，nodes 和 /readyz 仍为集群级检查；--skip-cluster-checks 仅跳过 nodes。事件使用服务端 type=Warning 过滤，但时间窗口在客户端判断。

不要为了运行脚本直接授予 cluster-admin。使用现有的受控只读账号，按需要授权；无权限的项目会明确列为 UNKNOWN。脚本不读取 Secret 资源、不抓日志、不执行容器命令、不重启、不删除、不修改资源。

注意：读取完整 Pod 对象本身可能读取到明文环境变量或 annotations；脚本仅在内存解析，不保存原始对象。报告不输出这些字段或任意事件 message，也不回显 kubectl 原始 stderr，避免认证插件错误带出凭据。报告仍有集群、命名空间、对象名称等运维信息，请放在受控目录。

创建报告时使用独占文件名、0600 权限；新建输出目录为 0700。已有目录的权限不自动修改，目录路径应由可信用户管理。报告写入失败返回 3。脚本不会自动清理历史报告，需要另外制定保留策略。

6. 定时巡检示例

Linux 使用 flock 防止任务重叠。下面是 crontab 示例，需要将路径、context 和 PATH 改为现场值，确保账号能非交互认证：

PATH=/usr/local/bin:/usr/bin:/bin
*/10 * * * * /usr/bin/flock -n /opt/k8s-inspect/inspect.lock /opt/k8s-inspect/k8s-inspect.sh --kubeconfig /opt/k8s-inspect/kubeconfig --context prod-cluster --output-dir /opt/k8s-inspect/reports >> /opt/k8s-inspect/cron.log 2>&1

目录应由运行账号管理；日志应轮转。cron 本身不会把脚本退出码自动转换成消息告警；接入现有监控平台时读取退出码和 JSON。脚本无通知、外发数据功能。

大集群先按命名空间评估运行耗时，再调整分页、超时和巡检频率。分页降低单次 API 响应大小，但 kubectl 和脚本仍会在本地汇总完整列表，内存随对象数量增长。默认串行请求，避免每个 Pod 单独发起 describe/logs 请求。

7. 发现异常后的只读核查

脚本有意只展示结构化状态及 reason。需要详细错误消息、日志时，在受控终端按对象核查，注意输出中的敏感信息：

kubectl --context prod-cluster describe node NODE
kubectl --context prod-cluster -n NAMESPACE describe pod POD
kubectl --context prod-cluster -n NAMESPACE logs POD -c CONTAINER --tail=100
kubectl --context prod-cluster -n NAMESPACE logs POD -c CONTAINER --previous --tail=100
kubectl --context prod-cluster -n NAMESPACE describe deployment DEPLOYMENT
kubectl --context prod-cluster -n NAMESPACE describe pvc PVC

不要仅凭历史 OOM、累计重启次数或 Warning 事件直接重启业务；先判断是否仍在发生及业务影响。

8. 边界与验证情况

本版本不检测：业务 HTTP 成功率、DNS 实际解析、Service/Ingress 端到端连通、空 EndpointSlice、HPA/CronJob/PDB 策略、PV/CSI 后端真实读写、证书到期、etcd 多成员一致性、宿主机完整磁盘/内核状态、GPU 硬件和云平台状态。API readyz 只反映所连接 API 端点的就绪状态，不证明所有控制平面实例健康。

Event 受集群保留时间限制；Pod 被删除后无法从当前对象恢复历史状态。未创建出 Pod 的 CronJob 调度问题也不能仅凭 Pod 列表发现。应结合集群监控、业务探测及平台专用巡检使用。

交付前已通过 Bash 语法检查、ShellCheck，以及 8 组本地模拟测试，覆盖节点/Pod/容器、工作负载、Job/PVC/事件、资源单位与阈值、报告权限、退出码、RBAC 失败、缺失指标和超时。未连接实际 Kubernetes 集群验证；实际版本、RBAC、认证插件和集群规模需要在目标环境首次运行确认。

判断依据参考 Kubernetes 官方文档：Node Status、Node-pressure Eviction、Pod Conditions。