1. Elasticsearch搜索引擎运维监控概述

Elasticsearch作为分布式搜索引擎,在生产环境中需要专业的运维监控和管理。本文将详细介绍Elasticsearch集群监控、索引管理、查询优化、性能调优的完整解决方案,帮助运维人员有效管理Elasticsearch集群。

1.1 核心挑战

  1. 集群监控: 实时监控Elasticsearch集群和节点状态
  2. 索引管理: 管理索引生命周期和分片分配
  3. 查询优化: 优化搜索查询性能和资源使用
  4. 性能调优: 优化Elasticsearch性能和吞吐量
  5. 故障诊断: 快速定位Elasticsearch相关问题

1.2 技术架构

1
2
3
4
5
Elasticsearch监控 → 数据采集 → 性能分析 → 告警通知 → 自动优化
↓ ↓ ↓ ↓ ↓
集群指标 → 监控代理 → 数据存储 → 告警引擎 → 调优脚本
↓ ↓ ↓ ↓ ↓
索引管理 → 查询优化 → 性能调优 → 自动修复 → 运维记录

2. Elasticsearch监控系统

2.1 Maven依赖配置

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
<!-- pom.xml -->
<dependencies>
<!-- Spring Boot Web -->
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-web</artifactId>
</dependency>

<!-- Spring Boot Data Redis -->
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-data-redis</artifactId>
</dependency>

<!-- Elasticsearch Client -->
<dependency>
<groupId>org.elasticsearch.client</groupId>
<artifactId>elasticsearch-rest-high-level-client</artifactId>
<version>7.17.9</version>
</dependency>

<!-- Micrometer监控 -->
<dependency>
<groupId>io.micrometer</groupId>
<artifactId>micrometer-registry-prometheus</artifactId>
</dependency>

<!-- MyBatis Plus -->
<dependency>
<groupId>com.baomidou</groupId>
<artifactId>mybatis-plus-boot-starter</artifactId>
<version>3.5.2</version>
</dependency>
</dependencies>

2.2 应用配置

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
# application.yml
server:
port: 8080

spring:
redis:
host: localhost
port: 6379
database: 0

# Elasticsearch监控配置
elasticsearch-monitor:
cluster-nodes: "localhost:9200" # ES集群节点
cluster-name: "production-cluster" # 集群名称
collection-interval: 10000 # 采集间隔(毫秒)
query-timeout: 30000 # 查询超时时间(毫秒)
shard-threshold: 80 # 分片使用率告警阈值(%)
index-threshold: 90 # 索引使用率告警阈值(%)

3. Elasticsearch监控服务

3.1 Elasticsearch监控实体类

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
/**
* Elasticsearch集群监控数据实体类
*/
@Data
@TableName("elasticsearch_cluster_monitor")
public class ElasticsearchClusterMonitor {

@TableId(type = IdType.AUTO)
private Long id; // 主键ID

private String clusterName; // 集群名称

private String hostname; // 主机名

private String ip; // IP地址

private Integer nodeCount; // 节点数量

private Integer indexCount; // 索引数量

private Integer shardCount; // 分片数量

private Long totalDocuments; // 总文档数

private Long totalSize; // 总大小(字节)

private String clusterStatus; // 集群状态

private String clusterHealth; // 集群健康状态

private Integer activeShards; // 活跃分片数

private Integer unassignedShards; // 未分配分片数

private Integer relocatingShards; // 重定位分片数

private Integer initializingShards; // 初始化分片数

private Long searchQueries; // 搜索查询数

private Long indexingQueries; // 索引查询数

private Double avgSearchTime; // 平均搜索时间(毫秒)

private Double avgIndexingTime; // 平均索引时间(毫秒)

private Long memoryUsed; // 内存使用量(字节)

private Long memoryTotal; // 总内存(字节)

private Double memoryUsage; // 内存使用率

private Long diskUsed; // 磁盘使用量(字节)

private Long diskTotal; // 总磁盘空间(字节)

private Double diskUsage; // 磁盘使用率

private Date collectTime; // 采集时间

private Date createTime; // 创建时间
}

/**
* Elasticsearch索引监控数据实体类
*/
@Data
@TableName("elasticsearch_index_monitor")
public class ElasticsearchIndexMonitor {

@TableId(type = IdType.AUTO)
private Long id; // 主键ID

private String clusterName; // 集群名称

private String indexName; // 索引名称

private String indexStatus; // 索引状态

private Integer shardCount; // 分片数量

private Integer replicaCount; // 副本数量

private Long documentCount; // 文档数量

private Long indexSize; // 索引大小(字节)

private Long storeSize; // 存储大小(字节)

private Double compressionRatio; // 压缩比

private Long searchQueries; // 搜索查询数

private Long indexingQueries; // 索引查询数

private Double avgSearchTime; // 平均搜索时间(毫秒)

private Double avgIndexingTime; // 平均索引时间(毫秒)

private String indexHealth; // 索引健康状态

private Date collectTime; // 采集时间

private Date createTime; // 创建时间
}

3.2 Elasticsearch监控服务

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
/**
* Elasticsearch监控服务
* 负责Elasticsearch集群和索引数据的采集、存储和分析
*/
@Service
public class ElasticsearchMonitorService {

@Autowired
private ElasticsearchClusterMonitorMapper elasticsearchClusterMonitorMapper;

@Autowired
private ElasticsearchIndexMonitorMapper elasticsearchIndexMonitorMapper;

@Autowired
private RedisTemplate<String, Object> redisTemplate;

@Autowired
private AlertService alertService;

private RestHighLevelClient client;

/**
* 初始化Elasticsearch客户端
*/
@PostConstruct
public void initElasticsearchClient() {
try {
// 创建客户端
RestClientBuilder builder = RestClient.builder(
new HttpHost("localhost", 9200, "http")
);

client = new RestHighLevelClient(builder);

log.info("Elasticsearch客户端初始化成功");

} catch (Exception e) {
log.error("初始化Elasticsearch客户端失败: {}", e.getMessage(), e);
}
}

/**
* 采集Elasticsearch集群数据
* 定期采集Elasticsearch集群和索引信息
*/
@Scheduled(fixedRate = 10000) // 每10秒执行一次
public void collectElasticsearchData() {
try {
// 1. 采集集群信息
collectClusterInfo();

// 2. 采集索引信息
collectIndexInfo();

} catch (Exception e) {
log.error("采集Elasticsearch数据失败: {}", e.getMessage(), e);
}
}

/**
* 采集集群信息
*/
private void collectClusterInfo() {
try {
// 1. 获取集群信息
ElasticsearchClusterInfo clusterInfo = getClusterInfo();

// 2. 创建集群监控数据
ElasticsearchClusterMonitor monitorData = createClusterMonitorData(clusterInfo);

// 3. 保存到数据库
elasticsearchClusterMonitorMapper.insert(monitorData);

// 4. 更新缓存
updateClusterCache(monitorData);

// 5. 检查集群告警
checkClusterAlert(monitorData);

log.debug("采集集群信息: nodeCount={}, indexCount={}",
monitorData.getNodeCount(), monitorData.getIndexCount());

} catch (Exception e) {
log.error("采集集群信息失败: {}", e.getMessage(), e);
}
}

/**
* 获取集群信息
*/
private ElasticsearchClusterInfo getClusterInfo() {
ElasticsearchClusterInfo clusterInfo = new ElasticsearchClusterInfo();

try {
// 获取集群健康状态
ClusterHealthRequest healthRequest = new ClusterHealthRequest();
ClusterHealthResponse healthResponse = client.cluster().health(healthRequest, RequestOptions.DEFAULT);

clusterInfo.setClusterName(healthResponse.getClusterName());
clusterInfo.setClusterStatus(healthResponse.getStatus().toString());
clusterInfo.setClusterHealth(healthResponse.getStatus().toString());
clusterInfo.setNodeCount(healthResponse.getNumberOfNodes());
clusterInfo.setActiveShards(healthResponse.getActiveShards());
clusterInfo.setUnassignedShards(healthResponse.getUnassignedShards());
clusterInfo.setRelocatingShards(healthResponse.getRelocatingShards());
clusterInfo.setInitializingShards(healthResponse.getInitializingShards());

// 获取集群统计信息
setClusterStatistics(clusterInfo);

// 获取资源使用信息
setResourceUsageInfo(clusterInfo);

} catch (Exception e) {
log.error("获取集群信息失败: {}", e.getMessage(), e);
}

return clusterInfo;
}

/**
* 设置集群统计信息
*/
private void setClusterStatistics(ElasticsearchClusterInfo clusterInfo) {
try {
// 获取索引统计信息
IndicesStatsRequest statsRequest = new IndicesStatsRequest();
IndicesStatsResponse statsResponse = client.indices().stats(statsRequest, RequestOptions.DEFAULT);

clusterInfo.setIndexCount(statsResponse.getIndices().size());
clusterInfo.setShardCount(statsResponse.getTotal().getShards().getTotal());

// 计算总文档数和总大小
long totalDocuments = 0;
long totalSize = 0;

for (IndexStats indexStats : statsResponse.getIndices().values()) {
totalDocuments += indexStats.getTotal().getDocs().getCount();
totalSize += indexStats.getTotal().getStore().getSizeInBytes();
}

clusterInfo.setTotalDocuments(totalDocuments);
clusterInfo.setTotalSize(totalSize);

// 获取搜索和索引统计
clusterInfo.setSearchQueries(statsResponse.getTotal().getSearch().getTotal().getQueryTotal());
clusterInfo.setIndexingQueries(statsResponse.getTotal().getIndexing().getTotal().getIndexTotal());

// 计算平均时间
if (statsResponse.getTotal().getSearch().getTotal().getQueryTotal() > 0) {
clusterInfo.setAvgSearchTime(statsResponse.getTotal().getSearch().getTotal().getQueryTimeInMillis() /
(double) statsResponse.getTotal().getSearch().getTotal().getQueryTotal());
}

if (statsResponse.getTotal().getIndexing().getTotal().getIndexTotal() > 0) {
clusterInfo.setAvgIndexingTime(statsResponse.getTotal().getIndexing().getTotal().getIndexTimeInMillis() /
(double) statsResponse.getTotal().getIndexing().getTotal().getIndexTotal());
}

} catch (Exception e) {
log.error("设置集群统计信息失败: {}", e.getMessage(), e);
}
}

/**
* 设置资源使用信息
*/
private void setResourceUsageInfo(ElasticsearchClusterInfo clusterInfo) {
try {
// 获取节点统计信息
NodesStatsRequest nodesStatsRequest = new NodesStatsRequest();
NodesStatsResponse nodesStatsResponse = client.nodes().stats(nodesStatsRequest, RequestOptions.DEFAULT);

long totalMemory = 0;
long usedMemory = 0;
long totalDisk = 0;
long usedDisk = 0;

for (NodeStats nodeStats : nodesStatsResponse.getNodes()) {
// 内存信息
if (nodeStats.getJvm() != null && nodeStats.getJvm().getMem() != null) {
totalMemory += nodeStats.getJvm().getMem().getHeapMax().getBytes();
usedMemory += nodeStats.getJvm().getMem().getHeapUsed().getBytes();
}

// 磁盘信息
if (nodeStats.getFs() != null) {
totalDisk += nodeStats.getFs().getTotal().getTotal().getBytes();
usedDisk += nodeStats.getFs().getTotal().getTotal().getBytes() -
nodeStats.getFs().getTotal().getAvailable().getBytes();
}
}

clusterInfo.setMemoryTotal(totalMemory);
clusterInfo.setMemoryUsed(usedMemory);
clusterInfo.setMemoryUsage(totalMemory > 0 ? (double) usedMemory / totalMemory * 100 : 0);

clusterInfo.setDiskTotal(totalDisk);
clusterInfo.setDiskUsed(usedDisk);
clusterInfo.setDiskUsage(totalDisk > 0 ? (double) usedDisk / totalDisk * 100 : 0);

} catch (Exception e) {
log.error("设置资源使用信息失败: {}", e.getMessage(), e);
}
}

/**
* 创建集群监控数据
*/
private ElasticsearchClusterMonitor createClusterMonitorData(ElasticsearchClusterInfo clusterInfo) {
ElasticsearchClusterMonitor monitorData = new ElasticsearchClusterMonitor();

// 设置基本信息
monitorData.setClusterName(clusterInfo.getClusterName());
monitorData.setHostname(getHostname());
monitorData.setIp(getLocalIpAddress());
monitorData.setCollectTime(new Date());
monitorData.setCreateTime(new Date());

// 设置集群信息
monitorData.setNodeCount(clusterInfo.getNodeCount());
monitorData.setIndexCount(clusterInfo.getIndexCount());
monitorData.setShardCount(clusterInfo.getShardCount());

// 设置数据信息
monitorData.setTotalDocuments(clusterInfo.getTotalDocuments());
monitorData.setTotalSize(clusterInfo.getTotalSize());

// 设置状态信息
monitorData.setClusterStatus(clusterInfo.getClusterStatus());
monitorData.setClusterHealth(clusterInfo.getClusterHealth());
monitorData.setActiveShards(clusterInfo.getActiveShards());
monitorData.setUnassignedShards(clusterInfo.getUnassignedShards());
monitorData.setRelocatingShards(clusterInfo.getRelocatingShards());
monitorData.setInitializingShards(clusterInfo.getInitializingShards());

// 设置性能信息
monitorData.setSearchQueries(clusterInfo.getSearchQueries());
monitorData.setIndexingQueries(clusterInfo.getIndexingQueries());
monitorData.setAvgSearchTime(clusterInfo.getAvgSearchTime());
monitorData.setAvgIndexingTime(clusterInfo.getAvgIndexingTime());

// 设置资源信息
monitorData.setMemoryUsed(clusterInfo.getMemoryUsed());
monitorData.setMemoryTotal(clusterInfo.getMemoryTotal());
monitorData.setMemoryUsage(clusterInfo.getMemoryUsage());
monitorData.setDiskUsed(clusterInfo.getDiskUsed());
monitorData.setDiskTotal(clusterInfo.getDiskTotal());
monitorData.setDiskUsage(clusterInfo.getDiskUsage());

return monitorData;
}

/**
* 采集索引信息
*/
private void collectIndexInfo() {
try {
// 获取所有索引
GetIndexRequest getIndexRequest = new GetIndexRequest("*");
GetIndexResponse getIndexResponse = client.indices().get(getIndexRequest, RequestOptions.DEFAULT);

for (String indexName : getIndexResponse.getIndices()) {
try {
// 获取索引统计信息
IndicesStatsRequest statsRequest = new IndicesStatsRequest();
statsRequest.indices(indexName);
IndicesStatsResponse statsResponse = client.indices().stats(statsRequest, RequestOptions.DEFAULT);

IndexStats indexStats = statsResponse.getIndex(indexName);
if (indexStats != null) {
// 创建索引监控数据
ElasticsearchIndexMonitor indexMonitor = createIndexMonitorData(indexName, indexStats);

// 保存到数据库
elasticsearchIndexMonitorMapper.insert(indexMonitor);

// 更新缓存
updateIndexCache(indexMonitor);

// 检查索引告警
checkIndexAlert(indexMonitor);
}

} catch (Exception e) {
log.error("处理索引失败: indexName={}, error={}", indexName, e.getMessage());
}
}

} catch (Exception e) {
log.error("采集索引信息失败: {}", e.getMessage(), e);
}
}

/**
* 创建索引监控数据
*/
private ElasticsearchIndexMonitor createIndexMonitorData(String indexName, IndexStats indexStats) {
ElasticsearchIndexMonitor indexMonitor = new ElasticsearchIndexMonitor();

// 设置基本信息
indexMonitor.setClusterName("production-cluster");
indexMonitor.setIndexName(indexName);
indexMonitor.setCollectTime(new Date());
indexMonitor.setCreateTime(new Date());

// 设置索引信息
indexMonitor.setShardCount(indexStats.getTotal().getShards().getTotal());
indexMonitor.setReplicaCount(indexStats.getTotal().getShards().getReplicas());
indexMonitor.setDocumentCount(indexStats.getTotal().getDocs().getCount());
indexMonitor.setIndexSize(indexStats.getTotal().getStore().getSizeInBytes());
indexMonitor.setStoreSize(indexStats.getTotal().getStore().getSizeInBytes());

// 设置性能信息
indexMonitor.setSearchQueries(indexStats.getTotal().getSearch().getTotal().getQueryTotal());
indexMonitor.setIndexingQueries(indexStats.getTotal().getIndexing().getTotal().getIndexTotal());

// 计算平均时间
if (indexStats.getTotal().getSearch().getTotal().getQueryTotal() > 0) {
indexMonitor.setAvgSearchTime(indexStats.getTotal().getSearch().getTotal().getQueryTimeInMillis() /
(double) indexStats.getTotal().getSearch().getTotal().getQueryTotal());
}

if (indexStats.getTotal().getIndexing().getTotal().getIndexTotal() > 0) {
indexMonitor.setAvgIndexingTime(indexStats.getTotal().getIndexing().getTotal().getIndexTimeInMillis() /
(double) indexStats.getTotal().getIndexing().getTotal().getIndexTotal());
}

// 设置健康状态
indexMonitor.setIndexHealth("green");

return indexMonitor;
}

/**
* 更新集群缓存
*/
private void updateClusterCache(ElasticsearchClusterMonitor monitorData) {
try {
String cacheKey = "elasticsearch:cluster:" + monitorData.getClusterName();
redisTemplate.opsForValue().set(cacheKey, monitorData, Duration.ofMinutes(5));

} catch (Exception e) {
log.warn("更新集群缓存失败: {}", e.getMessage());
}
}

/**
* 更新索引缓存
*/
private void updateIndexCache(ElasticsearchIndexMonitor indexMonitor) {
try {
String cacheKey = "elasticsearch:index:" + indexMonitor.getIndexName();
redisTemplate.opsForValue().set(cacheKey, indexMonitor, Duration.ofMinutes(5));

} catch (Exception e) {
log.warn("更新索引缓存失败: {}", e.getMessage());
}
}

/**
* 检查集群告警
*/
private void checkClusterAlert(ElasticsearchClusterMonitor monitorData) {
try {
String alertType = null;
String alertLevel = null;
String alertMessage = null;

// 检查集群健康状态告警
if ("red".equals(monitorData.getClusterHealth())) {
alertType = "ELASTICSEARCH_CLUSTER_RED";
alertLevel = "CRITICAL";
alertMessage = "Elasticsearch集群状态为红色";
} else if ("yellow".equals(monitorData.getClusterHealth())) {
alertType = "ELASTICSEARCH_CLUSTER_YELLOW";
alertLevel = "WARNING";
alertMessage = "Elasticsearch集群状态为黄色";
}

// 检查未分配分片告警
if (monitorData.getUnassignedShards() > 0) {
alertType = "ELASTICSEARCH_UNASSIGNED_SHARDS";
alertLevel = "WARNING";
alertMessage = String.format("Elasticsearch未分配分片: %d", monitorData.getUnassignedShards());
}

// 检查内存使用率告警
if (monitorData.getMemoryUsage() > 90) {
alertType = "ELASTICSEARCH_MEMORY_HIGH";
alertLevel = "CRITICAL";
alertMessage = String.format("Elasticsearch内存使用率过高: %.2f%%", monitorData.getMemoryUsage());
} else if (monitorData.getMemoryUsage() > 80) {
alertType = "ELASTICSEARCH_MEMORY_WARNING";
alertLevel = "WARNING";
alertMessage = String.format("Elasticsearch内存使用率较高: %.2f%%", monitorData.getMemoryUsage());
}

// 检查磁盘使用率告警
if (monitorData.getDiskUsage() > 90) {
alertType = "ELASTICSEARCH_DISK_HIGH";
alertLevel = "CRITICAL";
alertMessage = String.format("Elasticsearch磁盘使用率过高: %.2f%%", monitorData.getDiskUsage());
} else if (monitorData.getDiskUsage() > 80) {
alertType = "ELASTICSEARCH_DISK_WARNING";
alertLevel = "WARNING";
alertMessage = String.format("Elasticsearch磁盘使用率较高: %.2f%%", monitorData.getDiskUsage());
}

// 发送告警
if (alertType != null) {
sendClusterAlert(monitorData, alertType, alertLevel, alertMessage);
}

} catch (Exception e) {
log.error("检查集群告警失败: {}", e.getMessage(), e);
}
}

/**
* 检查索引告警
*/
private void checkIndexAlert(ElasticsearchIndexMonitor indexMonitor) {
try {
String alertType = null;
String alertLevel = null;
String alertMessage = null;

// 检查索引健康状态告警
if ("red".equals(indexMonitor.getIndexHealth())) {
alertType = "ELASTICSEARCH_INDEX_RED";
alertLevel = "CRITICAL";
alertMessage = String.format("Elasticsearch索引状态为红色: %s", indexMonitor.getIndexName());
} else if ("yellow".equals(indexMonitor.getIndexHealth())) {
alertType = "ELASTICSEARCH_INDEX_YELLOW";
alertLevel = "WARNING";
alertMessage = String.format("Elasticsearch索引状态为黄色: %s", indexMonitor.getIndexName());
}

// 检查搜索时间告警
if (indexMonitor.getAvgSearchTime() > 1000) {
alertType = "ELASTICSEARCH_SEARCH_SLOW";
alertLevel = "WARNING";
alertMessage = String.format("Elasticsearch搜索时间过长: %s, %.2fms",
indexMonitor.getIndexName(), indexMonitor.getAvgSearchTime());
}

// 发送告警
if (alertType != null) {
sendIndexAlert(indexMonitor, alertType, alertLevel, alertMessage);
}

} catch (Exception e) {
log.error("检查索引告警失败: {}", e.getMessage(), e);
}
}

/**
* 发送集群告警
*/
private void sendClusterAlert(ElasticsearchClusterMonitor monitorData, String alertType, String alertLevel, String alertMessage) {
try {
String alertKey = "elasticsearch:cluster:alert:" + monitorData.getClusterName() + ":" + alertType;
Boolean hasAlert = redisTemplate.hasKey(alertKey);

if (hasAlert == null || !hasAlert) {
AlertMessage alert = new AlertMessage();
alert.setType(alertType);
alert.setLevel(alertLevel);
alert.setMessage(alertMessage);
alert.setTimestamp(new Date());
alert.setHostname(monitorData.getHostname());

alertService.sendAlert(alert);

redisTemplate.opsForValue().set(alertKey, "1", Duration.ofMinutes(5));

log.warn("发送集群告警: clusterName={}, type={}, level={}",
monitorData.getClusterName(), alertType, alertLevel);
}

} catch (Exception e) {
log.error("发送集群告警失败: {}", e.getMessage(), e);
}
}

/**
* 发送索引告警
*/
private void sendIndexAlert(ElasticsearchIndexMonitor indexMonitor, String alertType, String alertLevel, String alertMessage) {
try {
String alertKey = "elasticsearch:index:alert:" + indexMonitor.getIndexName() + ":" + alertType;
Boolean hasAlert = redisTemplate.hasKey(alertKey);

if (hasAlert == null || !hasAlert) {
AlertMessage alert = new AlertMessage();
alert.setType(alertType);
alert.setLevel(alertLevel);
alert.setMessage(alertMessage);
alert.setTimestamp(new Date());
alert.setHostname(indexMonitor.getClusterName());

alertService.sendAlert(alert);

redisTemplate.opsForValue().set(alertKey, "1", Duration.ofMinutes(5));

log.warn("发送索引告警: indexName={}, type={}, level={}",
indexMonitor.getIndexName(), alertType, alertLevel);
}

} catch (Exception e) {
log.error("发送索引告警失败: {}", e.getMessage(), e);
}
}

/**
* 获取实时集群数据
*/
public ElasticsearchClusterMonitor getRealTimeClusterData(String clusterName) {
String cacheKey = "elasticsearch:cluster:" + clusterName;
return (ElasticsearchClusterMonitor) redisTemplate.opsForValue().get(cacheKey);
}

/**
* 获取实时索引数据
*/
public ElasticsearchIndexMonitor getRealTimeIndexData(String indexName) {
String cacheKey = "elasticsearch:index:" + indexName;
return (ElasticsearchIndexMonitor) redisTemplate.opsForValue().get(cacheKey);
}

/**
* 获取主机名
*/
private String getHostname() {
try {
return InetAddress.getLocalHost().getHostName();
} catch (UnknownHostException e) {
return "unknown";
}
}

/**
* 获取本地IP地址
*/
private String getLocalIpAddress() {
try {
return InetAddress.getLocalHost().getHostAddress();
} catch (UnknownHostException e) {
return "127.0.0.1";
}
}

/**
* 关闭Elasticsearch客户端
*/
@PreDestroy
public void shutdownElasticsearchClient() {
try {
if (client != null) {
client.close();
}

log.info("Elasticsearch客户端关闭成功");

} catch (Exception e) {
log.error("关闭Elasticsearch客户端失败: {}", e.getMessage(), e);
}
}
}

/**
* Elasticsearch集群信息实体类
*/
@Data
public class ElasticsearchClusterInfo {
private String clusterName; // 集群名称
private String clusterStatus; // 集群状态
private String clusterHealth; // 集群健康状态
private Integer nodeCount; // 节点数量
private Integer indexCount; // 索引数量
private Integer shardCount; // 分片数量
private Long totalDocuments; // 总文档数
private Long totalSize; // 总大小
private Integer activeShards; // 活跃分片数
private Integer unassignedShards; // 未分配分片数
private Integer relocatingShards; // 重定位分片数
private Integer initializingShards; // 初始化分片数
private Long searchQueries; // 搜索查询数
private Long indexingQueries; // 索引查询数
private Double avgSearchTime; // 平均搜索时间
private Double avgIndexingTime; // 平均索引时间
private Long memoryUsed; // 内存使用量
private Long memoryTotal; // 总内存
private Double memoryUsage; // 内存使用率
private Long diskUsed; // 磁盘使用量
private Long diskTotal; // 总磁盘空间
private Double diskUsage; // 磁盘使用率
}

4. Elasticsearch管理服务

4.1 Elasticsearch管理服务

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
/**
* Elasticsearch管理服务
* 提供Elasticsearch集群管理功能
*/
@Service
public class ElasticsearchManagementService {

@Autowired
private RestHighLevelClient client;

@Autowired
private AlertService alertService;

/**
* 创建索引
*/
public void createIndex(String indexName, String mapping) {
try {
CreateIndexRequest request = new CreateIndexRequest(indexName);

if (mapping != null && !mapping.isEmpty()) {
request.mapping(mapping, XContentType.JSON);
}

CreateIndexResponse response = client.indices().create(request, RequestOptions.DEFAULT);

log.info("创建索引成功: indexName={}, acknowledged={}",
indexName, response.isAcknowledged());

} catch (Exception e) {
log.error("创建索引失败: indexName={}, error={}", indexName, e.getMessage(), e);
}
}

/**
* 删除索引
*/
public void deleteIndex(String indexName) {
try {
DeleteIndexRequest request = new DeleteIndexRequest(indexName);
DeleteIndexResponse response = client.indices().delete(request, RequestOptions.DEFAULT);

log.info("删除索引成功: indexName={}, acknowledged={}",
indexName, response.isAcknowledged());

} catch (Exception e) {
log.error("删除索引失败: indexName={}, error={}", indexName, e.getMessage(), e);
}
}

/**
* 更新索引映射
*/
public void updateIndexMapping(String indexName, String mapping) {
try {
PutMappingRequest request = new PutMappingRequest(indexName);
request.source(mapping, XContentType.JSON);

AcknowledgedResponse response = client.indices().putMapping(request, RequestOptions.DEFAULT);

log.info("更新索引映射成功: indexName={}, acknowledged={}",
indexName, response.isAcknowledged());

} catch (Exception e) {
log.error("更新索引映射失败: indexName={}, error={}", indexName, e.getMessage(), e);
}
}

/**
* 更新索引设置
*/
public void updateIndexSettings(String indexName, String settings) {
try {
UpdateSettingsRequest request = new UpdateSettingsRequest(indexName);
request.settings(settings, XContentType.JSON);

AcknowledgedResponse response = client.indices().putSettings(request, RequestOptions.DEFAULT);

log.info("更新索引设置成功: indexName={}, acknowledged={}",
indexName, response.isAcknowledged());

} catch (Exception e) {
log.error("更新索引设置失败: indexName={}, error={}", indexName, e.getMessage(), e);
}
}

/**
* 刷新索引
*/
public void refreshIndex(String indexName) {
try {
RefreshRequest request = new RefreshRequest(indexName);
RefreshResponse response = client.indices().refresh(request, RequestOptions.DEFAULT);

log.info("刷新索引成功: indexName={}, totalShards={}",
indexName, response.getTotalShards());

} catch (Exception e) {
log.error("刷新索引失败: indexName={}, error={}", indexName, e.getMessage(), e);
}
}

/**
* 强制合并索引
*/
public void forceMergeIndex(String indexName) {
try {
ForceMergeRequest request = new ForceMergeRequest(indexName);
request.maxNumSegments(1);

ForceMergeResponse response = client.indices().forcemerge(request, RequestOptions.DEFAULT);

log.info("强制合并索引成功: indexName={}, totalShards={}",
indexName, response.getTotalShards());

} catch (Exception e) {
log.error("强制合并索引失败: indexName={}, error={}", indexName, e.getMessage(), e);
}
}

/**
* 获取索引列表
*/
public List<String> getIndices() {
try {
GetIndexRequest request = new GetIndexRequest("*");
GetIndexResponse response = client.indices().get(request, RequestOptions.DEFAULT);

return Arrays.asList(response.getIndices());

} catch (Exception e) {
log.error("获取索引列表失败: {}", e.getMessage(), e);
return new ArrayList<>();
}
}

/**
* 获取索引信息
*/
public Map<String, Object> getIndexInfo(String indexName) {
try {
GetIndexRequest request = new GetIndexRequest(indexName);
GetIndexResponse response = client.indices().get(request, RequestOptions.DEFAULT);

Map<String, Object> indexInfo = new HashMap<>();
indexInfo.put("name", indexName);
indexInfo.put("mappings", response.getMappings().get(indexName).getSourceAsMap());
indexInfo.put("settings", response.getSettings().get(indexName).getAsMap());

return indexInfo;

} catch (Exception e) {
log.error("获取索引信息失败: indexName={}, error={}", indexName, e.getMessage(), e);
return new HashMap<>();
}
}

/**
* 执行搜索查询
*/
public SearchResponse executeSearch(String indexName, String query) {
try {
SearchRequest request = new SearchRequest(indexName);
request.source(query, XContentType.JSON);

return client.search(request, RequestOptions.DEFAULT);

} catch (Exception e) {
log.error("执行搜索查询失败: indexName={}, query={}, error={}",
indexName, query, e.getMessage(), e);
return null;
}
}

/**
* 批量索引文档
*/
public void bulkIndex(String indexName, List<Map<String, Object>> documents) {
try {
BulkRequest request = new BulkRequest();

for (Map<String, Object> document : documents) {
IndexRequest indexRequest = new IndexRequest(indexName);
indexRequest.source(document);
request.add(indexRequest);
}

BulkResponse response = client.bulk(request, RequestOptions.DEFAULT);

log.info("批量索引文档成功: indexName={}, totalItems={}, hasFailures={}",
indexName, response.getItems().length, response.hasFailures());

} catch (Exception e) {
log.error("批量索引文档失败: indexName={}, error={}", indexName, e.getMessage(), e);
}
}

/**
* 获取集群健康状态
*/
public ClusterHealthResponse getClusterHealth() {
try {
ClusterHealthRequest request = new ClusterHealthRequest();
return client.cluster().health(request, RequestOptions.DEFAULT);

} catch (Exception e) {
log.error("获取集群健康状态失败: {}", e.getMessage(), e);
return null;
}
}

/**
* 获取节点信息
*/
public NodesInfoResponse getNodesInfo() {
try {
NodesInfoRequest request = new NodesInfoRequest();
return client.nodes().info(request, RequestOptions.DEFAULT);

} catch (Exception e) {
log.error("获取节点信息失败: {}", e.getMessage(), e);
return null;
}
}
}

5. 总结

本文详细介绍了Elasticsearch搜索引擎运维监控与管理的完整解决方案,包括:

5.1 核心技术点

  1. 集群监控: 实时监控Elasticsearch集群和节点状态
  2. 索引监控: 监控索引健康状态和性能指标
  3. 查询优化: 优化搜索查询性能和资源使用
  4. 资源管理: 管理内存、磁盘使用情况
  5. 告警通知: 多级告警、智能通知

5.2 架构优势

  1. 实时监控: 10秒间隔的实时Elasticsearch数据采集
  2. 智能告警: 基于阈值的智能告警机制
  3. 自动化管理: 自动化的索引和集群管理
  4. 多维度分析: 集群、索引、性能等多维度分析

5.3 最佳实践

  1. 监控策略: 设置合理的Elasticsearch监控阈值
  2. 管理策略: 根据业务需求执行索引管理
  3. 性能优化: 合理配置Elasticsearch参数
  4. 预防措施: 提前预防Elasticsearch集群问题

通过以上架构设计,可以构建完善的Elasticsearch运维监控系统,实现Elasticsearch集群的有效管理和优化。