kafka sasl elk

### 如何在Kafka中配置SASL以与ELK堆栈集成

#### 配置Kafka Broker支持SASL/GSSAPI (Kerberos)

为了使Kafka能够通过SASL机制进行身份验证，需修改`server.properties`文件来启用相应的协议。具体来说，在监听器设置中加入带有SASL前缀的地址，并指定所使用的安全协议为GSSAPI：

listeners=SASL_PLAINTEXT://192.168.10.2:9092
security.inter.broker.protocol=SASL_PLAINTEXT
sasl.mechanism.inter.broker.protocol=GSSAPI

同时还需要定义允许客户端连接的方式以及对应的认证方法[^1]。

对于Kerberos而言，还需额外编辑JAAS配置文件（通常命名为`kafka_server_jaas.conf`），该文件应放置于启动命令可访问的位置并包含类似下面的内容:

KafkaServer {
    com.sun.security.auth.module.Krb5LoginModule required
    useKeyTab=true
    storeKey=true
    keyTab="/etc/security/keytabs/kafka.service.keytab"
    principal="kafka/hostname@REALM.COM";
};
Client {
    com.sun.security.auth.module.Krb5LoginModule required
    useTicketCache=true;
};

此部分用于告知JVM如何获取必要的凭证以便参与Kerberos交换过程。

#### 修改Logstash中的Kafka插件参数实现通信加密

当准备就绪之后，则转向调整Logstash侧的相关设定。假设已经安装好了logstash-input-kafka插件，那么只需更新input区块内的选项即可满足需求：

input {
  kafka {
    bootstrap_servers => "broker1:9092,broker2:9092"
    topics => ["my-topic"]
    group_id => "my-group"

    codec => json {}

    sasl_mechanisms => "GSSAPI"
    security_protocol => "SASL_PLAINTEXT"
    
    jaas_path => "/path/to/jaas.conf" # JAAS配置路径
  }
}

这里特别注意`sasl_mechanisms`, `security_protocol` 和 `jaas_path`三个属性，它们共同作用确保了Logstash能按照预期方式同启用了SASL的安全型Kafka实例交互数据流[^4]。

最后一步就是重启所有涉及的服务组件——包括但不限于ZooKeeper、Kafka Brokers还有Logstash本身，从而让新的更改生效[^2]。