将OpenStack Ceilometer计量通知导出至Kafka的最优方案

我之前在OpenStack环境里折腾过把Ceilometer指标数据同步到Kafka的需求，刚好踩过你提到的这两个方案的坑，分享点实际操作的经验给你：

方案一：文件发布器 + Kafka Connect FileSource Connector #

这个方案的核心是让Ceilometer把数据写到日志文件，再用Kafka Connect把文件内容同步到Kafka，好处是不用额外加中间件，架构简单。

• 第一步：配置Ceilometer文件发布器
  找到Ceilometer的主配置文件（通常是/etc/ceilometer/ceilometer.conf），修改发布器配置：

  [publisher]
  publisher = file:///var/log/ceilometer/metrics.log
  

  别忘了给这个日志文件设置正确的权限，要保证Ceilometer进程能写，Kafka Connect进程能读，比如设置属组为ceilometer:kafka。

• 第二步：配置Kafka Connect的FileSource连接器
  创建一个连接器配置文件（比如ceilometer-file-source.json），内容如下：

  {
    "name": "ceilometer-file-source",
    "config": {
      "connector.class": "org.apache.kafka.connect.file.FileStreamSourceConnector",
      "tasks.max": "1",
      "file": "/var/log/ceilometer/metrics.log",
      "topic": "ceilometer-metrics",
      "key.converter": "org.apache.kafka.connect.storage.StringConverter",
      "value.converter": "org.apache.kafka.connect.json.JsonConverter",
      "value.converter.schemas.enable": "false"
    }
  }
  

  这里我用了JSON转换器，因为Ceilometer的日志可以配置成结构化JSON格式，后续处理更方便。如果你的日志是纯文本，把value.converter改成StringConverter就行。

  然后通过Kafka Connect的API启动连接器：

  curl -X POST -H "Content-Type: application/json" --data @ceilometer-file-source.json http://<你的Connect主机>:8083/connectors
  

• 实操踩坑点

  • 一定要配置日志轮转！用logrotate的时候，建议用copytruncate模式，这样Kafka Connect不会因为文件被截断而丢失追踪位置；如果用重命名+新建文件的方式，要确保Kafka Connect能识别到新文件的inode变化。
  • 可以在Ceilometer配置里开启结构化日志输出，比如在ceilometer.conf里加：

    [logging]
    format = json
    

    这样同步到Kafka的数据就是标准JSON，后续监控系统解析起来更轻松。

方案二：AMQP通知器 + Kafka Connect AMQP Source Connector #

这个方案是用OpenStack原生的AMQP通知机制，先把Ceilometer数据发到AMQP代理（比如RabbitMQ），再用Kafka Connect同步到Kafka，好处是数据实时性更好，符合OpenStack的生态设计。

• 第一步：部署并配置RabbitMQ
  先装RabbitMQ，然后给Ceilometer创建专属的虚拟主机、用户和权限：

  rabbitmqctl add_vhost ceilometer
  rabbitmqctl add_user ceilometer_user 你的密码
  rabbitmqctl set_permissions -p ceilometer ceilometer_user ".*" ".*" ".*"
  

• 第二步：配置Ceilometer的AMQP通知器
  修改ceilometer.conf里的通知驱动配置：

  [notification]
  driver = messagingv2
  transport_url = amqp://ceilometer_user:你的密码@<RabbitMQ主机>:5672/ceilometer
  topic = ceilometer.notifications
  

  配置完重启Ceilometer服务，检查日志有没有连接RabbitMQ的错误，确保数据能正常发出去。

• 第三步：配置Kafka Connect的AMQP连接器
  我用的是Confluent的AMQP Source连接器，先把插件包放到Kafka Connect的插件目录里，然后创建配置文件（比如ceilometer-amqp-source.json）：

  {
    "name": "ceilometer-amqp-source",
    "config": {
      "connector.class": "io.confluent.connect.amqp.AmqpSourceConnector",
      "tasks.max": "2",
      "amqp.host": "<RabbitMQ主机>",
      "amqp.port": "5672",
      "amqp.virtual.host": "ceilometer",
      "amqp.username": "ceilometer_user",
      "amqp.password": "你的密码",
      "amqp.queue": "ceilometer-kafka-bridge",
      "kafka.topic": "ceilometer-metrics",
      "key.converter": "org.apache.kafka.connect.storage.StringConverter",
      "value.converter": "org.apache.kafka.connect.json.JsonConverter",
      "value.converter.schemas.enable": "false"
    }
  }
  

  注意要先在RabbitMQ里创建这个队列，并绑定到Ceilometer的通知topic上：

  rabbitmqadmin declare queue name=ceilometer-kafka-bridge vhost=ceilometer
  rabbitmqadmin declare binding vhost=ceilometer source=ceilometer.notifications destination=ceilometer-kafka-bridge destination_type=queue
  

  最后同样用API启动连接器就行。

• 实操踩坑点

  • RabbitMQ要配置持久化，避免重启或者宕机丢数据；同时调整队列的prefetch count，根据你的消息量设定合适的值，防止Consumer过载。
  • 如果用开源的AMQP连接器，要注意版本兼容性，最好和你的Kafka Connect版本匹配。

两个方案的对比 #

根据你的监控系统需求选就行，如果追求简单快捷选文件方案，如果对实时性和可靠性要求高选AMQP方案。

内容的提问来源于stack exchange，提问作者user2523783