最新发现一个比较有意思的库ksniff，它是一个kubectl 插件，使用tcpdump来远程捕获Kubernetes集群中的pod流量并保存到文件或输出到wireshark中，发布网络问题定位。使用方式如下：
kubectl sniff hello-minikube-7c77b68cff-qbvsd -c hello-minikube
要知道很多pod中其实是没有tcpdump这个可执行文件的，那它是如何在Kubernetes集群的Pod中远程执行tcpdump命令的？又是如何倒出Pod的tcpdump的输出并将输出直接传递给wireshark的？下面分析一下该工具的实现方式。
ksniff有两种运行模式：特权模式和非特权模式。首先看下非特权模式。
非特权模式
非特权模式的运行逻辑为：
- 找到本地的tcpdump可执行文件路径
- 将本地的tcpdump上传到远端pod中
- 远程执行pod的tcpdump命令，并将输出重定向到文件或wireshark
上传tcpdump可执行文件
ksniff使用tar命令对tcpdump可执行文件进行打包，然后通过client-go的remotecommand库将其解压到pod中，最后执行tcpdump命令即可：
fileContent, err := ioutil.ReadFile(req.Src) //读取tcpdump可执行文件
if err != nil {
return 0, err
}
tarFile, err := WrapAsTar(destFileName, fileContent)//将使用tar命令对tcpdump进行打包
if err != nil {
return 0, err
}
stdIn := bytes.NewReader(tarFile) //通过标准输入传递给容器
tarCmd := []string{"tar", "-xf", "-"} //构建解压命令
destDir := path.Dir(req.Dst)
if len(destDir) > 0 {
tarCmd = append(tarCmd, "-C", destDir)
}
execTarRequest := ExecCommandRequest{
KubeRequest: KubeRequest{
Clientset: req.Clientset,
RestConfig: req.RestConfig,
Namespace: req.Namespace,
Pod: req.Pod,
Container: req.Container,
},
Command: tarCmd,
StdIn: stdIn,
StdOut: stdOut,
StdErr: stdErr,
}
exitCode, err := PodExecuteCommand(execTarRequest)
tar打包的实现如下：
func WrapAsTar(fileNameOnTar string, fileContent []byte) ([]byte, error) {
var buf bytes.Buffer
tw := tar.NewWriter(&buf)
hdr := &tar.Header{
Name: fileNameOnTar,
Mode: 0755,
Size: int64(len(fileContent)),
}
if err := tw.WriteHeader(hdr); err != nil {
return nil, err
}
if _, err := tw.Write(fileContent); err != nil {
return nil, err
}
if err := tw.Close(); err != nil {
return nil, err
}
return buf.Bytes(), nil
}
远程执行命令
下面是远程在pod中执行命令的代码，是client-go remotecommand库的标准用法，没有什么特别之处：
func (k *KubernetesApiServiceImpl) ExecuteCommand(podName string, containerName string, command []string, stdOut io.Writer) (int, error) {
log.Infof("executing command: '%s' on container: '%s', pod: '%s', namespace: '%s'", command, containerName, podName, k.targetNamespace)
stdErr := new(Writer)
executeTcpdumpRequest := ExecCommandRequest{
KubeRequest: KubeRequest{
Clientset: k.clientset,
RestConfig: k.restConfig,
Namespace: k.targetNamespace,
Pod: podName,
Container: containerName,
},
Command: command,
StdErr: stdErr,
StdOut: stdOut,
}
exitCode, err := PodExecuteCommand(executeTcpdumpRequest)
if err != nil {
log.WithError(err).Errorf("failed executing command: '%s', exitCode: '%d', stdErr: '%s'",
command, exitCode, stdErr.Output)
return exitCode, err
}
log.Infof("command: '%s' executing successfully exitCode: '%d', stdErr :'%s'", command, exitCode, stdErr.Output)
return exitCode, err
}
func PodExecuteCommand(req ExecCommandRequest) (int, error) {
execRequest := req.Clientset.CoreV1().RESTClient().Post().
Resource("pods").
Name(req.Pod).
Namespace(req.Namespace).
SubResource("exec")
execRequest.VersionedParams(&corev1.PodExecOptions{
Container: req.Container,
Command: req.Command,
Stdin: req.StdIn != nil,
Stdout: req.StdOut != nil,
Stderr: req.StdErr != nil,
TTY: false,
}, scheme.ParameterCodec)
exec, err := remotecommand.NewSPDYExecutor(req.RestConfig, "POST", execRequest.URL())
if err != nil {
return 0, err
}
err = exec.Stream(remotecommand.StreamOptions{
Stdin: req.StdIn,
Stdout: req.StdOut, //重定向的输出，可以是文件或wireshark
Stderr: req.StdErr,
Tty: false,
})
var exitCode = 0
if err != nil {
if exitErr, ok := err.(utilexec.ExitError); ok && exitErr.Exited() {
exitCode = exitErr.ExitStatus()
err = nil
}
}
return exitCode, err
}
执行tcpdump命令
该步骤就是组装远程命令，并在目标pod中执行即可：
func (u *StaticTcpdumpSnifferService) Start(stdOut io.Writer) error {
log.Info("start sniffing on remote container")
command := []string{u.settings.UserSpecifiedRemoteTcpdumpPath, "-i", u.settings.UserSpecifiedInterface,
"-U", "-w", "-", u.settings.UserSpecifiedFilter}
exitCode, err := u.kubernetesApiService.ExecuteCommand(u.settings.UserSpecifiedPodName, u.settings.UserSpecifiedContainer, command, stdOut)
if err != nil || exitCode != 0 {
return errors.Errorf("executing sniffer failed, exit code: '%d'", exitCode)
}
log.Infof("done sniffing on remote container")
return nil
}
wireshark库支持输入重定向，使用o.wireshark.StdinPipe()创建出输入之后，将其作为远程调用tcpdump命令的StreamOptions.Stdout的参数即可将pod的输出重定向到wireshark中：
title := fmt.Sprintf("gui.window_title:%s/%s/%s", o.resultingContext.Namespace, o.settings.UserSpecifiedPodName, o.settings.UserSpecifiedContainer)
o.wireshark = exec.Command("wireshark", "-k", "-i", "-", "-o", title)
stdinWriter, err := o.wireshark.StdinPipe() //创建输入
if err != nil {
return err
}
go func() {
err := o.snifferService.Start(stdinWriter)//将wireshark创建的输入作为pod的输出
if err != nil {
log.WithError(err).Errorf("failed to start remote sniffing, stopping wireshark")
_ = o.wireshark.Process.Kill()
}
}()
err = o.wireshark.Run()
特权模式
特权模式的处理有一些复杂，该模式下，ksniff会在目标pod所在的node节点(通过目标pod的pod.Spec.NodeName字段获取)上创建一个权限为privileged的pod，并挂载主机的/目录和默认的容器socket，然后在特权pod内调用对应的容器运行时命令来执行tcpdump命令。ksniff支持三种常见的容器运行时：docker、cri-o和containerd，对应的容器运行时的默认目录如下：
/var/run/docker.sock
/var/run/crio/crio.sock
/run/containerd/containerd.sock
由于特权模式可能会创建一个新的pod，因此在命令执行完之后需要清理掉新建的pod。
区分容器运行时
特权模式下会调用目标节点上的容器运行时命令，不同容器运行时的命令是不同的，那么ksniff是如何区分不同的容器运行时呢？
ksniff会通过kubernetes clientset来获取目标pod信息，通过pod.status.containerStatuses.containerID字段来确定所使用的CRI，如下例，其CRI为containerd，containerId为0f76ee399228ed02f8ba13a6bbec6bb8b696f4f1997176882b309edbe3a56ee1：
status:
containerStatuses:
- containerID: containerd://0f76ee399228ed02f8ba13a6bbec6bb8b696f4f1997176882b309edbe3a56ee1
....
容器运行时和ContainerId的获取方式如下：
func (o *Ksniff) findContainerId(pod *corev1.Pod) error {
for _, containerStatus := range pod.Status.ContainerStatuses {
if o.settings.UserSpecifiedContainer == containerStatus.Name {
result := strings.Split(containerStatus.ContainerID, "://")
if len(result) != 2 {
break
}
o.settings.DetectedContainerRuntime = result[0] //获取容器运行时
o.settings.DetectedContainerId = result[1] //获取containerID
return nil
}
}
return errors.Errorf("couldn't find container: '%s' in pod: '%s'", o.settings.UserSpecifiedContainer, o.settings.UserSpecifiedPodName)
}
不同运行时执行tcpdump命令
下面看下不同运行时是如何执行tcpdump命令的。
Containerd
Containerd会在特权pod内通过crictl pull来拉取tcpdump镜像并启动tcpdump容器，使其和目标容器(containerId)共享相同的网络命名空间，这样就可以使用tcpdump抓取目标容器的报文。在命令执行完之后需要清理创建出来的tcpdump容器。
func (d *ContainerdBridge) BuildTcpdumpCommand(containerId *string, netInterface string, filter string, pid *string, socketPath string, tcpdumpImage string) []string {
d.tcpdumpContainerName = "ksniff-container-" + utils.GenerateRandomString(8)
d.socketPath = socketPath
tcpdumpCommand := fmt.Sprintf("tcpdump -i %s -U -w - %s", netInterface, filter)
shellScript := fmt.Sprintf(`
set -ex
export CONTAINERD_SOCKET="%s"
export CONTAINERD_NAMESPACE="k8s.io"
export CONTAINER_RUNTIME_ENDPOINT="unix:///host${CONTAINERD_SOCKET}"
export IMAGE_SERVICE_ENDPOINT=${CONTAINER_RUNTIME_ENDPOINT}
crictl pull %s >/dev/null
netns=$(crictl inspect %s | jq '.info.runtimeSpec.linux.namespaces[] | select(.type == "network") | .path' | tr -d '"')
exec chroot /host ctr -a ${CONTAINERD_SOCKET} run --rm --with-ns "network:${netns}" %s %s %s
`, d.socketPath, tcpdumpImage, *containerId, tcpdumpImage, d.tcpdumpContainerName, tcpdumpCommand)
command := []string{"/bin/sh", "-c", shellScript}
return command
}
func (d *ContainerdBridge) BuildCleanupCommand() []string {
shellScript := fmt.Sprintf(`
set -ex
export CONTAINERD_SOCKET="%s"
export CONTAINERD_NAMESPACE="k8s.io"
export CONTAINER_ID="%s"
chroot /host ctr -a ${CONTAINERD_SOCKET} task kill -s SIGKILL ${CONTAINER_ID}
`, d.socketPath, d.tcpdumpContainerName)
command := []string{"/bin/sh", "-c", shellScript}
return command
}
Cri-o
Cri-o通过nsenter指定目标容器的进程进入目标网络命名空间来执行tcpdump命令，由于它没有使用tcpdump镜像，因此要求目标节点上需要存在tcpdump可执行文件：
func (c *CrioBridge) BuildTcpdumpCommand(containerId *string, netInterface string, filter string, pid *string, socketPath string, tcpdumpImage string) []string {
return []string{"nsenter", "-n", "-t", *pid, "--", "tcpdump", "-i", netInterface, "-U", "-w", "-", filter}
}
这种方式下没有在特权pod内部创建容器，因此不需要清理环境。
docker
docker的处理方式和containerd类似，也是通过启动tcpdump容器，并和目标容器共享网络命名空间实现的：
func (d *DockerBridge) BuildTcpdumpCommand(containerId *string, netInterface string, filter string, pid *string, socketPath string, tcpdumpImage string) []string {
d.tcpdumpContainerName = "ksniff-container-" + utils.GenerateRandomString(8)
containerNameFlag := fmt.Sprintf("--name=%s", d.tcpdumpContainerName)
command := []string{"docker", "--host", "unix://" + socketPath,
"run", "--rm", "--log-driver", "none", containerNameFlag,
fmt.Sprintf("--net=container:%s", *containerId), tcpdumpImage, "-i",
netInterface, "-U", "-w", "-", filter}
d.cleanupCommand = []string{"docker", "--host", "unix://" + socketPath,
"rm", "-f", d.tcpdumpContainerName}
return command
}
func (d *DockerBridge) BuildCleanupCommand() []string {
return d.cleanupCommand
}
环境清理
由于特权模式下创建了特权pod，containerd和docker还会在特权pod内创建tcpdump容器，因此在进行环境清理时需要清理掉创建出来的tcpdump容器，然后再清理掉特权pod：
func (p *PrivilegedPodSnifferService) Cleanup() error {
command := p.runtimeBridge.BuildCleanupCommand()
if command != nil {
log.Infof("removing privileged container: '%s'", p.privilegedContainerName)
exitCode, err := p.kubernetesApiService.ExecuteCommand(p.privilegedPod.Name, p.privilegedContainerName, command, &kube.NopWriter{})
if err != nil {
log.WithError(err).Errorf("failed to remove privileged container: '%s', exit code: '%d', "+
"please manually remove it", p.privilegedContainerName, exitCode)
} else {
log.Infof("privileged container: '%s' removed successfully", p.privilegedContainerName)
}
}
if p.privilegedPod != nil {
log.Infof("removing pod: '%s'", p.privilegedPod.Name)
err := p.kubernetesApiService.DeletePod(p.privilegedPod.Name)
if err != nil {
log.WithError(err).Errorf("failed to remove pod: '%s", p.privilegedPod.Name)
return err
}
log.Infof("pod: '%s' removed successfully", p.privilegedPod.Name)
}
return nil
}
总结
非特权模式的实现比较简单，不需要考虑容器运行时的问题，但它也有一个缺点，就是需要考虑目标容器的运行环境，比如32位/64位、amd/arm等，可能需要在本地准备多套tcpdump来满足不同的容器运行环境。
特权模式的实现相对比较复杂，如果还有其他的运行时，就需要对ksniff进行功能扩展。且有些集群节点上可能会禁用特权pod，导致该方法行不通。
尽管存在一些使用上的限制，但本文在文件上传以及对不同容器运行时方面的处理还是很值得借鉴的。