ES7.7以后的版本将安全认证功能免费开放了。并将X-pack插件集成了到了开源的ElasticSearch版本中。
下面将图文介绍如何利用X-pack给ElasticSearch相关组件设置用户名和密码：

1. 修改主站点的elasticsearch.yml添加一下行

vim /opt/module/elasticsearch-7.12.0/config/elasticsearch.yml

2. 添加内容

xpack.security.enabled: true

3. 生成证书

cd /opt/module/elasticsearch-7.12.0
bin/elasticsearch-certutil cert -out config/elastic-certificates.p12 -pass ""

生成的证书地址：/opt/module/elasticsearch-7.12.0/config/elastic-certificates.p12

继续修改elasticsearch.yml文件
添加以下四行：

xpack.security.transport.ssl.enabled: true
xpack.security.transport.ssl.verification_mode: certificate
xpack.security.transport.ssl.keystore.path: elastic-certificates.p12
xpack.security.transport.ssl.truststore.path: elastic-certificates.p12

4. 更新集群节点的配置
把elastic-certificates.p12复制到其他节点的config下，同时修改其他节点的yml

sh /opt/bash/xsync /opt/module/elasticsearch-7.12.0/config/elastic-certificates.p12 
sh /opt/bash/batchexec "sudo chown -R bigdata:bigdata /opt/module/elasticsearch-7.12.0/config/elastic-certificates.p12"

其他节点yml文件直接添加内容【注意不要xsync yml到其他机器 配置不一样】

vim /opt/module/elasticsearch-7.12.0/config/elasticsearch.yml

xpack.security.enabled: true
xpack.security.transport.ssl.enabled: true
xpack.security.transport.ssl.verification_mode: certificate
xpack.security.transport.ssl.keystore.path: elastic-certificates.p12
xpack.security.transport.ssl.truststore.path: elastic-certificates.p12

5. 重启ES

jps | grep 'Elasticsearch' | awk '{print $1}' | xargs -n 1 kill -9
sh /opt/module/elasticsearch-7.12.0/bin/elasticsearch -d > /data/log/elasticsearch/elasticsearch.txt 2>&1 &

6. 密码设置

sh /opt/module/elasticsearch-7.12.0/bin/elasticsearch-setup-passwords interactive

** 修改kibana的配置kibana.yml**

vim /opt/module/kibana-7.12.0-linux-x86_64/config/kibana.yml

7. 测试

8. 如果需要修改ES密码

curl -H "Content-Type:application/json" -XPOST -u elastic 'http://$IP:9200/_xpack/security/user/elastic/_password' -d '{ "password" : "密码" }'

9. 代码添加账号密码
9.1 Flink写ES代码

val dnsDomainBlackListEventSinkBuilder: ElasticsearchSink.Builder[DnsDomainBlackListEvent] = new ElasticsearchSink.Builder[DnsDomainBlackListEvent](httpHosts, new DnsDomainBlackListEventEsSinkFunction())
//TODO 如果是加密了
if (ProjectConfiguration.ESUSERNAME != null && ProjectConfiguration.ESPASSWORD !=null) {
  dnsDomainBlackListEventSinkBuilder.setRestClientFactory(new ElaticSearchRestClientFactory)
}

需要实现RestClientFactory

class ElaticSearchRestClientFactory extends RestClientFactory{
  def configureRestClientBuilder(restClientBuilder: RestClientBuilder): Unit = {
    val credentialsProvider: CredentialsProvider = new BasicCredentialsProvider();
    credentialsProvider.setCredentials(AuthScope.ANY,
      new UsernamePasswordCredentials(ProjectConfiguration.ESUSERNAME, ProjectConfiguration.ESPASSWORD));
    restClientBuilder.setHttpClientConfigCallback(new HttpClientConfigCallback() {
      def customizeHttpClient(httpAsyncClientBuilder: HttpAsyncClientBuilder): HttpAsyncClientBuilder = {
          httpAsyncClientBuilder.disableAuthCaching();
          httpAsyncClientBuilder.setDefaultCredentialsProvider(credentialsProvider);
      }
    })
  }
}

9.2 Spark写ES代码

添加两行
.option(“es.net.http.auth.user”,es_user)
.option(“es.net.http.auth.pass”,es_pass)

9.3 Logstash写ES
在logstash的conf文件中添加user和password即可

output{
elasticsearch {
        hosts => [ "$IP:9200" ]
        user => "elastic"
        password => "changeme"
    }
}

9.4 Java ES High Level API代码

import org.apache.http.HttpHost
import org.apache.http.auth.{AuthScope, UsernamePasswordCredentials}
import org.apache.http.impl.client.BasicCredentialsProvider
import org.apache.http.impl.nio.client.HttpAsyncClientBuilder
import org.elasticsearch.client.{RestClient, RestClientBuilder, RestHighLevelClient}

def getClient(hosts: Seq[String], port: Int = 9200, schema: String = "http", account: (String, String) = null): RestHighLevelClient = {
  val httpHosts = hosts.map(h => new HttpHost(h, port, schema))
  val clientBuilder = RestClient.builder(httpHosts:_*)

  if (account == null) {
    new RestHighLevelClient(clientBuilder)
  } else {
    // login with credential
    val credentialsProvider = new BasicCredentialsProvider()
    credentialsProvider.setCredentials(AuthScope.ANY, new UsernamePasswordCredentials(account._1, account._2))

    new RestHighLevelClient(
      clientBuilder.setHttpClientConfigCallback(new RestClientBuilder.HttpClientConfigCallback() {
        def customizeHttpClient(httpClientBuilder: HttpAsyncClientBuilder): HttpAsyncClientBuilder = {
          httpClientBuilder.setDefaultCredentialsProvider(credentialsProvider)
        }}
      )
    )
  }
}