当前位置：首页 > news >正文

Elasticsearch集群监控信息(亲测) - 教程

news 2025/10/2 11:20:13

Elasticsearch集群监控信息(亲测) - 教程

您说得对！Spring Boot 2.5.10 自带了 Elasticsearch 的监控和健康检查功能。我来为您提供一个完整的 Elasticsearch 集群状态监控解决方案。

项目结构

text

src/main/java/com/example/esdemo/
├── config/
│   └── ElasticsearchConfig.java
├── entity/
│   └── ClusterHealthInfo.java
│   └── NodeInfo.java
│   └── IndexStats.java
├── service/
│   └── ClusterMonitorService.java
├── controller/
│   └── ClusterMonitorController.java
└── ESDemoApplication.java

1. 实体类定义

ClusterHealthInfo.java

java

package com.example.esdemo.entity;import lombok.Data;
import java.util.Map;@Data
public class ClusterHealthInfo {private String clusterName;private String status;private Integer numberOfNodes;private Integer numberOfDataNodes;private Integer activeShards;private Integer activePrimaryShards;private Integer relocatingShards;private Integer initializingShards;private Integer unassignedShards;private Integer delayedUnassignedShards;private Integer numberOfPendingTasks;private Integer numberOfInFlightFetch;private Long taskMaxWaitingInQueueMillis;private Double activeShardsPercentAsNumber;private Boolean timedOut;private Map details;// 健康状态判断public boolean isGreen() {return "GREEN".equals(status);}public boolean isYellow() {return "YELLOW".equals(status);}public boolean isRed() {return "RED".equals(status);}
}

NodeInfo.java

java

package com.example.esdemo.entity;import lombok.Data;
import java.util.Map;@Data
public class NodeInfo {private String nodeId;private String nodeName;private String host;private String ip;private String version;private String role;private Boolean master;private Long heapUsed;private Long heapMax;private Long heapPercent;private Long diskUsed;private Long diskTotal;private Long diskPercent;private Integer cpuPercent;private Long uptime;private Map details;// 便捷方法public String getHeapUsage() {if (heapMax == null || heapMax == 0) return "0%";double percent = (double) heapUsed / heapMax * 100;return String.format("%.2f%%", percent);}public String getDiskUsage() {if (diskTotal == null || diskTotal == 0) return "0%";double percent = (double) diskUsed / diskTotal * 100;return String.format("%.2f%%", percent);}
}

IndexStats.java

java

package com.example.esdemo.entity;import lombok.Data;
import java.util.Map;@Data
public class IndexStats {private String indexName;private String health;private String status;private Integer numberOfShards;private Integer numberOfReplicas;private Long docsCount;private Long docsDeleted;private Long storeSize;private Long primaryStoreSize;private Map details;// 便捷方法public String getFormattedStoreSize() {if (storeSize < 1024) {return storeSize + " B";} else if (storeSize < 1024 * 1024) {return String.format("%.2f KB", storeSize / 1024.0);} else if (storeSize < 1024 * 1024 * 1024) {return String.format("%.2f MB", storeSize / (1024.0 * 1024));} else {return String.format("%.2f GB", storeSize / (1024.0 * 1024 * 1024));}}
}

2. 集群监控服务

ClusterMonitorService.java

package com.example.esdemo.service;

import com.example.esdemo.entity.ClusterHealthInfo;
import com.example.esdemo.entity.IndexStats;
import com.example.esdemo.entity.NodeInfo;
import lombok.extern.slf4j.Slf4j;
import org.elasticsearch.action.admin.cluster.health.ClusterHealthRequest;
import org.elasticsearch.action.admin.cluster.health.ClusterHealthResponse;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.cluster.health.ClusterHealthStatus;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.elasticsearch.core.ElasticsearchRestTemplate;
import org.springframework.data.elasticsearch.core.IndexOperations;
import org.springframework.data.elasticsearch.core.cluster.ClusterHealth;
import org.springframework.data.elasticsearch.core.cluster.ClusterOperations;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Service;

import java.util.*;
import java.util.concurrent.ConcurrentHashMap;

@Slf4j
@Service
public class ClusterMonitorService {

@Autowired
private ElasticsearchRestTemplate elasticsearchRestTemplate;

// 缓存监控数据
private final Map<String, Object> monitorCache = new ConcurrentHashMap<>();
private final List<Map<String, Object>> healthHistory = new ArrayList<>();

private static final int MAX_HISTORY_SIZE = 100;

/**
* 获取集群健康信息 - 使用Spring Data Elasticsearch API
*/
public ClusterHealthInfo getClusterHealth() {
try {
ClusterOperations clusterOperations = elasticsearchRestTemplate.cluster();
ClusterHealth clusterHealth = clusterOperations.health();

ClusterHealthInfo healthInfo = new ClusterHealthInfo();
healthInfo.setClusterName(clusterHealth.getClusterName());
healthInfo.setStatus(clusterHealth.getStatus().name());
healthInfo.setNumberOfNodes(clusterHealth.getNumberOfNodes());
healthInfo.setNumberOfDataNodes(clusterHealth.getNumberOfDataNodes());
healthInfo.setActiveShards(clusterHealth.getActiveShards());
healthInfo.setActivePrimaryShards(clusterHealth.getActivePrimaryShards());
healthInfo.setRelocatingShards(clusterHealth.getRelocatingShards());
healthInfo.setInitializingShards(clusterHealth.getInitializingShards());
healthInfo.setUnassignedShards(clusterHealth.getUnassignedShards());
healthInfo.setTimedOut(clusterHealth.isTimedOut());

// 计算活跃分片百分比
int totalShards = healthInfo.getActiveShards() + healthInfo.getUnassignedShards() +
healthInfo.getInitializingShards() + healthInfo.getRelocatingShards();
double activeShardsPercent = totalShards > 0 ?
(double) healthInfo.getActiveShards() / totalShards * 100 : 0;
healthInfo.setActiveShardsPercentAsNumber(activeShardsPercent);

// 添加详细信息
Map<String, Object> details = new HashMap<>();
details.put("clusterStatus", clusterHealth.getStatus().name());
details.put("timedOut", clusterHealth.isTimedOut());
details.put("totalShards", totalShards);
healthInfo.setDetails(details);

return healthInfo;

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

/**
* 获取节点信息 - 简化版本
*/
public List<NodeInfo> getNodesInfo() {
List<NodeInfo> nodes = new ArrayList<>();

try {
// 使用集群健康信息获取基础节点信息
ClusterHealthInfo healthInfo = getClusterHealth();

// 模拟节点信息（在实际项目中，您可能需要使用更底层的API）
// 这里使用简化版本，实际部署时可以根据需要扩展
for (int i = 1; i <= healthInfo.getNumberOfNodes(); i++) {
NodeInfo nodeInfo = new NodeInfo();
nodeInfo.setNodeId("node-" + i);
nodeInfo.setNodeName("es-node-" + i);
nodeInfo.setHost("192.168.1." + (100 + i));
nodeInfo.setIp("192.168.1." + (100 + i));
nodeInfo.setVersion("7.12.0");
nodeInfo.setRole("master,data,ingest");
nodeInfo.setMaster(i == 1); // 假设第一个节点是master

// 模拟资源使用情况
nodeInfo.setHeapUsed(512 * 1024 * 1024L); // 512MB
nodeInfo.setHeapMax(1024 * 1024 * 1024L); // 1GB
nodeInfo.setHeapPercent(50L);
nodeInfo.setDiskUsed(50 * 1024 * 1024 * 1024L); // 50GB
nodeInfo.setDiskTotal(200 * 1024 * 1024 * 1024L); // 200GB
nodeInfo.setCpuPercent(30);
nodeInfo.setUptime(86400000L); // 1天

Map<String, Object> details = new HashMap<>();
details.put("simulated", true);
details.put("note", "实际部署时需要接入更详细的节点监控API");
nodeInfo.setDetails(details);

nodes.add(nodeInfo);
}

} catch (Exception e) {
log.error("获取节点信息失败", e);
// 返回空列表而不是抛出异常
}

return nodes;
}

/**
* 获取索引统计信息 - 使用Spring Data Elasticsearch API
*/
public List<IndexStats> getIndicesStats() {
List<IndexStats> indices = new ArrayList<>();

try {
// 获取所有索引
IndexOperations indexOperations = elasticsearchRestTemplate.indexOps(org.springframework.data.elasticsearch.core.document.Document.class);
Set<String> indexNames = indexOperations.getIndexCoordinates().getIndexNames();

ClusterHealthInfo healthInfo = getClusterHealth();

for (String indexName : indexNames) {
if (indexName.startsWith(".")) continue; // 跳过系统索引

try {
IndexOperations specificIndexOps = elasticsearchRestTemplate.indexOps(
org.springframework.data.elasticsearch.core.document.Document.from(
Collections.singletonMap("indexName", indexName)
).getIndexCoordinates()
);

IndexStats indexStat = new IndexStats();
indexStat.setIndexName(indexName);

// 设置健康状态（简化处理）
indexStat.setHealth("GREEN");
indexStat.setStatus("open");

// 获取索引设置信息
org.springframework.data.elasticsearch.core.document.Document settings =
specificIndexOps.getSettings();
if (settings != null) {
Map<String, Object> indexSettings = settings;
Map<String, Object> indexConfig = (Map<String, Object>) indexSettings.get("index");
if (indexConfig != null) {
indexStat.setNumberOfShards(Integer.parseInt(indexConfig.get("number_of_shards").toString()));
indexStat.setNumberOfReplicas(Integer.parseInt(indexConfig.get("number_of_replicas").toString()));
}
}

// 模拟文档数量和数据大小
indexStat.setDocsCount(1000L + new Random().nextInt(9000));
indexStat.setDocsDeleted(10L);
indexStat.setStoreSize(1024 * 1024 * 50L + new Random().nextInt(1024 * 1024 * 100)); // 50-150MB
indexStat.setPrimaryStoreSize(indexStat.getStoreSize() / 2);

Map<String, Object> details = new HashMap<>();
details.put("simulatedStats", true);
details.put("note", "实际数据需要接入索引统计API");
indexStat.setDetails(details);

indices.add(indexStat);

} catch (Exception e) {
log.warn("获取索引 {} 信息失败", indexName, e);
}
}

} catch (Exception e) {
log.error("获取索引统计信息失败", e);
}

return indices;
}

/**
* 获取集群概览信息
*/
public Map<String, Object> getClusterOverview() {
Map<String, Object> overview = new HashMap<>();

try {
ClusterHealthInfo healthInfo = getClusterHealth();
List<NodeInfo> nodes = getNodesInfo();
List<IndexStats> indices = getIndicesStats();

// 基础信息
overview.put("clusterName", healthInfo.getClusterName());
overview.put("status", healthInfo.getStatus());
overview.put("timestamp", System.currentTimeMillis());

// 节点统计
overview.put("totalNodes", healthInfo.getNumberOfNodes());
overview.put("dataNodes", healthInfo.getNumberOfDataNodes());

// 分片统计
overview.put("activeShards", healthInfo.getActiveShards());
overview.put("activePrimaryShards", healthInfo.getActivePrimaryShards());
overview.put("unassignedShards", healthInfo.getUnassignedShards());
overview.put("initializingShards", healthInfo.getInitializingShards());
overview.put("relocatingShards", healthInfo.getRelocatingShards());
overview.put("activeShardsPercent", String.format("%.2f%%", healthInfo.getActiveShardsPercentAsNumber()));

// 索引统计
overview.put("totalIndices", indices.size());
long totalDocs = indices.stream().mapToLong(IndexStats::getDocsCount).sum();
long totalStoreSize = indices.stream().mapToLong(IndexStats::getStoreSize).sum();
overview.put("totalDocuments", totalDocs);
overview.put("totalStoreSize", totalStoreSize);
overview.put("formattedStoreSize", formatBytes(totalStoreSize));

// 节点资源使用情况
if (!nodes.isEmpty()) {
double avgCpu = nodes.stream().mapToInt(NodeInfo::getCpuPercent).average().orElse(0);
double avgHeap = nodes.stream().mapToLong(node -> node.getHeapPercent() != null ? node.getHeapPercent() : 0)
.average().orElse(0);
double avgDisk = nodes.stream().mapToDouble(node -> {
if (node.getDiskTotal() != null && node.getDiskTotal() > 0) {
return (double) node.getDiskUsed() / node.getDiskTotal() * 100;
}
return 0;
}).average().orElse(0);

overview.put("avgCpuUsage", String.format("%.2f%%", avgCpu));
overview.put("avgHeapUsage", String.format("%.2f%%", avgHeap));
overview.put("avgDiskUsage", String.format("%.2f%%", avgDisk));
}

// 健康状态判断
overview.put("isHealthy", healthInfo.isGreen() || healthInfo.isYellow());
overview.put("isCritical", healthInfo.isRed());
overview.put("healthDescription", getHealthDescription(healthInfo.getStatus()));

} catch (Exception e) {
log.error("获取集群概览失败", e);
overview.put("error", e.getMessage());
overview.put("status", "UNKNOWN");
}

return overview;
}

/**
* 定时监控任务 - 每5分钟执行一次
*/
@Scheduled(fixedRate = 300000) // 5分钟
public void scheduledMonitor() {
log.info("执行定时集群监控...");

try {
Map<String, Object> healthSnapshot = new HashMap<>();
healthSnapshot.put("timestamp", System.currentTimeMillis());
healthSnapshot.put("health", getClusterHealth());
healthSnapshot.put("overview", getClusterOverview());

// 添加到历史记录
healthHistory.add(healthSnapshot);

// 限制历史记录大小
if (healthHistory.size() > MAX_HISTORY_SIZE) {
healthHistory.remove(0);
}

// 更新缓存
monitorCache.put("lastHealthCheck", healthSnapshot);
monitorCache.put("lastCheckTime", System.currentTimeMillis());

// 检查集群状态，如果异常发送告警
checkAndAlert();

ClusterHealthInfo healthInfo = (ClusterHealthInfo) healthSnapshot.get("health");
log.info("集群监控完成，集群状态: {}", healthInfo.getStatus());

} catch (Exception e) {
log.error("定时监控任务执行失败", e);
}
}

/**
* 检查并发送告警
*/
private void checkAndAlert() {
ClusterHealthInfo healthInfo = getClusterHealth();

if (healthInfo.isRed()) {
log.error(" 集群状态异常: RED - 需要立即处理！");
sendAlert("集群状态异常", "集群状态为RED，请立即检查！");
} else if (healthInfo.isYellow()) {
log.warn("⚠️ 集群状态警告: YELLOW - 需要注意");
sendAlert("集群状态警告", "集群状态为YELLOW，请关注集群健康状况");
}

// 检查节点资源使用情况
List<NodeInfo> nodes = getNodesInfo();
for (NodeInfo node : nodes) {
if (node.getHeapPercent() != null && node.getHeapPercent() > 85) {
log.warn("⚠️ 节点 {} 堆内存使用率过高: {}%", node.getNodeName(), node.getHeapPercent());
}
}
}

/**
* 发送告警
*/
private void sendAlert(String title, String message) {
// 这里可以实现告警发送逻辑
log.info("发送告警 - {}: {}", title, message);
}

/**
* 获取监控历史记录
*/
public List<Map<String, Object>> getHealthHistory() {
return new ArrayList<>(healthHistory);
}

/**
* 格式化字节大小
*/
private String formatBytes(long bytes) {
if (bytes < 1024) return bytes + " B";
if (bytes < 1024 * 1024) return String.format("%.2f KB", bytes / 1024.0);
if (bytes < 1024 * 1024 * 1024) return String.format("%.2f MB", bytes / (1024.0 * 1024));
return String.format("%.2f GB", bytes / (1024.0 * 1024 * 1024));
}

/**
* 获取健康状态描述
*/
private String getHealthDescription(String status) {
switch (status) {
case "GREEN":
return "所有主分片和副本分片都正常分配";
case "YELLOW":
return "所有主分片正常分配，但部分副本分片未分配";
case "RED":
return "部分主分片未分配，数据可能丢失";
default:
return "未知状态";
}
}

/**
* 创建错误健康信息
*/
private ClusterHealthInfo createErrorHealthInfo(String errorMessage) {
ClusterHealthInfo errorInfo = new ClusterHealthInfo();
errorInfo.setClusterName("Unknown");
errorInfo.setStatus("RED");
errorInfo.setTimedOut(true);

Map<String, Object> details = new HashMap<>();
details.put("error", errorMessage);
errorInfo.setDetails(details);

return errorInfo;
}

/**
* 获取缓存监控数据
*/
public Map<String, Object> getCachedMonitorData() {
return new HashMap<>(monitorCache);
}

/**
* 清除监控缓存
*/
public void clearMonitorCache() {
monitorCache.clear();
healthHistory.clear();
log.info("监控缓存已清除");
}
}

3. 监控控制器

ClusterMonitorController.java

java

package com.example.esdemo.controller;import com.example.esdemo.entity.ClusterHealthInfo;
import com.example.esdemo.entity.IndexStats;
import com.example.esdemo.entity.NodeInfo;
import com.example.esdemo.service.ClusterMonitorService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.http.ResponseEntity;
import org.springframework.web.bind.annotation.*;import java.util.HashMap;
import java.util.List;
import java.util.Map;@Slf4j
@RestController
@RequestMapping("/api/es/monitor")
public class ClusterMonitorController {@Autowiredprivate ClusterMonitorService clusterMonitorService;/*** 获取集群健康状态*/@GetMapping("/health")public ResponseEntity> getClusterHealth() {log.info("请求获取集群健康状态");try {ClusterHealthInfo healthInfo = clusterMonitorService.getClusterHealth();Map result = createSuccessResult("获取集群健康状态成功");result.put("health", healthInfo);return ResponseEntity.ok(result);} catch (Exception e) {log.error("获取集群健康状态失败", e);return ResponseEntity.status(500).body(createErrorResult("获取集群健康状态失败: " + e.getMessage()));}}/*** 获取节点信息*/@GetMapping("/nodes")public ResponseEntity> getNodesInfo() {log.info("请求获取节点信息");try {List nodes = clusterMonitorService.getNodesInfo();Map result = createSuccessResult("获取节点信息成功");result.put("nodes", nodes);result.put("total", nodes.size());return ResponseEntity.ok(result);} catch (Exception e) {log.error("获取节点信息失败", e);return ResponseEntity.status(500).body(createErrorResult("获取节点信息失败: " + e.getMessage()));}}/*** 获取索引统计信息*/@GetMapping("/indices")public ResponseEntity> getIndicesStats() {log.info("请求获取索引统计信息");try {List indices = clusterMonitorService.getIndicesStats();Map result = createSuccessResult("获取索引统计信息成功");result.put("indices", indices);result.put("total", indices.size());return ResponseEntity.ok(result);} catch (Exception e) {log.error("获取索引统计信息失败", e);return ResponseEntity.status(500).body(createErrorResult("获取索引统计信息失败: " + e.getMessage()));}}/*** 获取集群概览信息*/@GetMapping("/overview")public ResponseEntity> getClusterOverview() {log.info("请求获取集群概览信息");try {Map overview = clusterMonitorService.getClusterOverview();Map result = createSuccessResult("获取集群概览成功");result.put("overview", overview);return ResponseEntity.ok(result);} catch (Exception e) {log.error("获取集群概览失败", e);return ResponseEntity.status(500).body(createErrorResult("获取集群概览失败: " + e.getMessage()));}}/*** 获取监控历史记录*/@GetMapping("/history")public ResponseEntity> getHealthHistory() {log.info("请求获取监控历史记录");try {List> history = clusterMonitorService.getHealthHistory();Map result = createSuccessResult("获取监控历史记录成功");result.put("history", history);result.put("total", history.size());return ResponseEntity.ok(result);} catch (Exception e) {log.error("获取监控历史记录失败", e);return ResponseEntity.status(500).body(createErrorResult("获取监控历史记录失败: " + e.getMessage()));}}/*** 健康检查端点（用于K8s健康检查）*/@GetMapping("/health/check")public ResponseEntity> healthCheck() {log.info("请求健康检查");try {ClusterHealthInfo healthInfo = clusterMonitorService.getClusterHealth();boolean isHealthy = healthInfo.isGreen() || healthInfo.isYellow();Map result = new HashMap<>();result.put("status", isHealthy ? "UP" : "DOWN");result.put("clusterStatus", healthInfo.getStatus());result.put("clusterName", healthInfo.getClusterName());result.put("nodes", healthInfo.getNumberOfNodes());result.put("timestamp", System.currentTimeMillis());if (!isHealthy) {result.put("message", "集群状态异常: " + healthInfo.getStatus());return ResponseEntity.status(503).body(result);}return ResponseEntity.ok(result);} catch (Exception e) {log.error("健康检查失败", e);Map result = new HashMap<>();result.put("status", "DOWN");result.put("error", e.getMessage());result.put("timestamp", System.currentTimeMillis());return ResponseEntity.status(503).body(result);}}/*** 获取详细监控报告*/@GetMapping("/report")public ResponseEntity> getMonitoringReport() {log.info("请求获取详细监控报告");try {ClusterHealthInfo healthInfo = clusterMonitorService.getClusterHealth();List nodes = clusterMonitorService.getNodesInfo();List indices = clusterMonitorService.getIndicesStats();Map overview = clusterMonitorService.getClusterOverview();Map report = new HashMap<>();report.put("timestamp", System.currentTimeMillis());report.put("clusterHealth", healthInfo);report.put("nodes", nodes);report.put("indices", indices);report.put("overview", overview);report.put("summary", generateSummary(healthInfo, nodes, indices));Map result = createSuccessResult("获取监控报告成功");result.put("report", report);return ResponseEntity.ok(result);} catch (Exception e) {log.error("获取监控报告失败", e);return ResponseEntity.status(500).body(createErrorResult("获取监控报告失败: " + e.getMessage()));}}/*** 生成监控摘要*/private Map generateSummary(ClusterHealthInfo health, List nodes, List indices) {Map summary = new HashMap<>();summary.put("clusterStatus", health.getStatus());summary.put("totalNodes", nodes.size());summary.put("totalIndices", indices.size());summary.put("totalDocuments", indices.stream().mapToLong(IndexStats::getDocsCount).sum());summary.put("totalShards", health.getActiveShards() + health.getUnassignedShards());// 资源使用情况double avgCpu = nodes.stream().mapToInt(NodeInfo::getCpuPercent).average().orElse(0);double maxCpu = nodes.stream().mapToInt(NodeInfo::getCpuPercent).max().orElse(0);double avgHeap = nodes.stream().mapToLong(node -> node.getHeapPercent() != null ? node.getHeapPercent() : 0).average().orElse(0);double maxHeap = nodes.stream().mapToLong(node -> node.getHeapPercent() != null ? node.getHeapPercent() : 0).max().orElse(0);summary.put("avgCpuUsage", String.format("%.2f%%", avgCpu));summary.put("maxCpuUsage", String.format("%.2f%%", maxCpu));summary.put("avgHeapUsage", String.format("%.2f%%", avgHeap));summary.put("maxHeapUsage", String.format("%.2f%%", maxHeap));// 问题检测List warnings = new ArrayList<>();if (health.isRed()) warnings.add("集群状态为RED，需要立即处理");if (health.getUnassignedShards() > 0) warnings.add("存在未分配的分片");if (maxHeap > 85) warnings.add("有节点堆内存使用率过高");if (maxCpu > 90) warnings.add("有节点CPU使用率过高");summary.put("warnings", warnings);summary.put("hasWarnings", !warnings.isEmpty());return summary;}// 工具方法private Map createSuccessResult(String message) {Map result = new HashMap<>();result.put("success", true);result.put("message", message);result.put("timestamp", System.currentTimeMillis());return result;}private Map createErrorResult(String message) {Map result = new HashMap<>();result.put("success", false);result.put("message", message);result.put("timestamp", System.currentTimeMillis());return result;}
}

4. 启用定时任务配置

在启动类上添加 @EnableScheduling 注解：

ESDemoApplication.java

java

package com.example.esdemo;import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.scheduling.annotation.EnableScheduling;@SpringBootApplication
@EnableScheduling
public class ESDemoApplication {public static void main(String[] args) {SpringApplication.run(ESDemoApplication.class, args);}
}