pulsar阻塞导致logstash无法接入日志

郑文峰的博客

使用dify对接飞书多维表格使用n8n对接飞书多维表格服务启动时出现 OOM 一次服务升级时pg表DDL执行超时失败 Go语言高效IO缓冲技术详解 Go语言延迟初始化(Lazy Initialization)最佳实践 Go语言字符串拼接性能对比与优化指南 Go语言结构体内存对齐完全指南 Go语言空结构体：零内存消耗的高效编程 Go语言堆栈分配与逃逸分析深度解析 Go语言原子操作完全指南 Go语言内存预分配完全指南 Go语言不可变数据共享：无锁并发编程实践 Go语言零拷贝技术完全指南 Go语言遍历性能深度解析：从原理到优化实践 Go语言Interface Boxing原理与性能优化指南 Go协程池深度解析：原理、实现与最佳实践使用etcd分布式锁导致的协程泄露与死锁问题基于pre-commit的Python代码规范落地实践初识 MCP Server django-prometheus使用及源码分析 kube-proxy源码分析 kubernetes service如何通过iptables转发 tcp缓存引起的日志丢失 django-apschedule定时任务异常停止理解calico容器网络通信方案原理理解flannel的三种容器网络方案原理理解Linux IPIP隧道理解VXLAN网络理解Linux TunTap设备快速了解iptables kafka中listener和advertised.listeners的作用 django rest_framework 分页 django后端服务、logstash和flink接入VictoriaMetrics指标监控 python中import原理 docker容器单机网络手动实现docker容器bridge网络模型 mysql之MVCC原理 mysql之日志使用java开发logstash的filter插件使用python实现单例模式的三种方式 redis之缓存 redis之分片集群 redis之哨兵机制 redis之主从库同步 redis之持久化 redis之五种基本数据类型 go中如何处理error pod中将代码与运行环境分离 ddt源码分析 python装饰器的使用方法读书笔记:如何阅读一本书使用ddt实现unittest的参数化测试使用kubeadm安装k8s 优化gin表单的错误提示信息 gin中validator模块的源码分析 go简单使用grpc python简单使用grpc k8s之PV、PVC和StorageClass k8s之StatefulSet k8s之DaemonSet k8s之Job和CronJob k8s之ConfigMap和Secret k8s之Service k8s之Pod k8s之Deployment 容器的本质 docker容器 python迭代器与生成器 python元编程 python垃圾回收机制 python上下文管理器 django rest_framework使用jwt django rest_framework异常处理 django rest_framework 自定义文档 django压缩文件下载 django rest_framework使用pytest单元测试 django restframework choice 自定义输出数据 django Filtering 使用 django viewset 和 Router 配合使用时报的错 django model的序列化 django中使用AbStractUser django.core.exceptions.ImproperlyConfigured Application labels aren't unique, duplicates users django 中 media配置 django 外键引用自身和on_delete参数 django 警告 while time zone support is active Flask使用flask_socketio实现websocket flask结合mongo tornado 文件上传 tornado 使用jwt完成用户异步认证 tornado 用户密码 bcrypt加密 tornado 结合wtforms使用表单操作 tornado finish和write区别 tornado 使用peewee-async 完成异步orm数据库操作 pyspark streaming简介和消费 kafka示例使用hue创建ozzie的pyspark action workflow count的性能优化 django rest_framework Authentication django celery 结合使用网站

zhengwenfeng · 2024-10-30 · via 郑文峰的博客

# 背景

在logstash中，input管道接收tcp和udp的日志，然后再通过output管道将日志输出到pulsar中，如下图所示：

当前的问题是，在使用tcp和udp接入方式接入一段时间日志之后，日志突然无法接入了，在pulsar中的对应topic没有新的日志生产进来了。

# 检查网络情况

在容器中抓包，判断数据包是否进入到容器中

因为服务都是容器化部署的，所以可以使用nsenter -t $Pid -n来进入到容器的network namespace中，然后再使用tcpdump -i any port 514 -vvnn来抓取514端口的数据包。

通过抓包是可以抓到数据包，并且看三次握手和数据包都是正常的，说明流量已经进入到容器中了。

查看tcp和udp缓存队列

使用netstat -anp查看tcp和upd缓存队列，发现Recv-Q是一直有堆积的，也就是说，服务端没有在从缓存队列中消费日志，服务端卡住了。

# netstat -anp
Active Internet connections (servers and established)
Proto Recv-Q Send-Q Local Address           Foreign Address         State       PID/Program name  
tcp   286138      0 14.2.19.144:514         10.65.132.32:46778      ESTABLISHED 538152/java   
udp   152543      0 14.2.19.144:514         10.65.132.32:46774      ESTABLISHED 538152/java

1
2
3
4
5

# 检查logstash管道信息

进入到容器中执行以下命令来获取logstash对应管道的信息

curl -XGET 'localhost:9600/_node/stats/pipelines/syslog?pretty'

可以看到input管道中tcp插件的out的数量一直保持不变，说明已经没有接收新的数据包了。

{
    "id": "d1a2689850bcefed823d25f0ae0ee7b863f3843474c6eafd3d3fe5aa7babcded",
    "name": "tcp",
    "events": {
        "queue_push_duration_in_millis": 3728,
        "out": 6680
    }
}

1
2
3
4
5
6
7
8

再看output管道的pulsar插件信息，可以看到in比out多了150，这是因为pipeline.batch.size=50，并且pipeline.workers=3，所以一共有150条日志阻塞在队列中，pulsar-client法发送出去，阻塞了。

{
    "id": "4acf48fc134b333ecb24218723c8fd36d54dc3713cbe66bd199ce4ccb09f2cd1",
    "name": "pulsar",
    "events": {
        "duration_in_millis": 9923,
        "in": 6529,
        "out": 6379
    }
}

1
2
3
4
5
6
7
8
9

通过查看pulsar服务的状态和日志发现其并没有任何异常，手动去生产日志也是可以的，所以暂时排除了pulsar本身的问题。

# 从堆栈中找到蛛丝马迹

通过以下命令来查看logstash的线程堆栈信息

# curl -XGET 'localhost:9600/_node/hot_threads?pretty'

"java.base@11.0.15/jdk.internal.misc.Unsafe.park(Native Method)", "java.base@11.0.15/java.util.concurrent.locks.LockSupport.park(LockSupport.java:194)", "java.base@11.0.15/java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.await(AbstractQueuedSynchronizer.java:2081)", "org.apache.pulsar.client.impl.MemoryLimitController.reserveMemory(MemoryLimitController.java:88)", ...

1
2
3

通过上面的堆栈信息去google上搜索，找到了下面这个链接，发现是pulsar的官方bug，在开启压缩时，并且内存限制的越小，越容易触发该BUG，从而导致pulsar-client发生阻塞。

https://lists.apache.org/thread/76jn7k01ldgkl54n0bpgw3sf2kxqq5q9

# 复现

这里用的是开源的pulsar插件：logstash-output-pulsar (opens new window)

其并没有内存限制配置参数，通过修改源码将pulsar-client的memory_limit参数暴露出来，并将其设置成1，开启压缩，持续构造日志进来，会发现稳定快速复现该问题。

# 修复

查看logstash-output-pulsar项目下的build.gradle文件，其pulsar版本为2.10.2，而在https://github.com/apache/pulsar/pull/21790 提交可以看到在3.2.0版本修复的。

修改项目中pulsar-client版本号，打包出来，再按照上面的复现步骤，发现不会再进行阻塞了。

# 总结

善于利用组件提供的各种工具得到其指标，分析原因，找到蛛丝马迹，再利用google去搜索。

此内容由惯性聚合(RSS阅读器)自动聚合整理，仅供阅读参考。原文来自 — 版权归原作者所有。

推荐订阅源

郑文峰的博客

# 背景

# 检查网络情况

# 检查logstash管道信息

# 从堆栈中找到蛛丝马迹

# 复现

# 修复

# 总结