第98集Cassandra数据库运维监控与优化实战

1. Cassandra数据库运维监控概述

Cassandra作为分布式NoSQL数据库，在生产环境中需要专业的运维监控和管理。本文将详细介绍Cassandra集群监控、节点管理、性能调优、数据修复的完整解决方案，帮助运维人员有效管理Cassandra集群。

1.1 核心挑战

集群监控: 实时监控Cassandra集群和节点状态
节点管理: 管理节点加入、离开和故障恢复
性能调优: 优化Cassandra性能和吞吐量
数据修复: 修复数据一致性和完整性
故障诊断: 快速定位Cassandra相关问题

1.2 技术架构

Cassandra监控 → 数据采集 → 性能分析 → 告警通知 → 自动优化
   ↓        ↓         ↓         ↓         ↓
集群指标 → 监控代理 → 数据存储 → 告警引擎 → 调优脚本
   ↓        ↓         ↓         ↓         ↓
节点管理 → 数据修复 → 性能调优 → 自动修复 → 运维记录

2. Cassandra监控系统

2.1 Maven依赖配置

<!-- 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>
    
    <!-- Cassandra Driver -->
    <dependency>
        <groupId>com.datastax.oss</groupId>
        <artifactId>java-driver-core</artifactId>
        <version>4.15.0</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 应用配置

# application.yml
server:
  port: 8080

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

# Cassandra监控配置
cassandra-monitor:
  contact-points: "localhost:9042"     # Cassandra集群地址
  cluster-name: "production-cluster"    # 集群名称
  keyspace: "monitoring"               # 监控键空间
  collection-interval: 10000           # 采集间隔(毫秒)
  node-alert-threshold: 80             # 节点资源告警阈值(%)
  compaction-threshold: 90             # 压缩告警阈值(%)

3. Cassandra监控服务

3.1 Cassandra监控实体类

/**
 * Cassandra集群监控数据实体类
 */
@Data
@TableName("cassandra_cluster_monitor")
public class CassandraClusterMonitor {
    
    @TableId(type = IdType.AUTO)
    private Long id;                    // 主键ID
    
    private String clusterName;          // 集群名称
    
    private String hostname;             // 主机名
    
    private String ip;                   // IP地址
    
    private Integer nodeCount;           // 节点数量
    
    private Integer keyspaceCount;        // 键空间数量
    
    private Integer tableCount;          // 表数量
    
    private Long totalDataSize;          // 总数据大小
    
    private Long totalDiskSpace;         // 总磁盘空间
    
    private Double diskUsage;            // 磁盘使用率
    
    private Integer activeConnections;   // 活跃连接数
    
    private Double throughput;            // 吞吐量(操作/秒)
    
    private Double avgLatency;            // 平均延迟(毫秒)
    
    private Integer pendingCompactions;  // 待压缩任务数
    
    private Integer failedOperations;    // 失败操作数
    
    private String clusterStatus;        // 集群状态
    
    private Date collectTime;            // 采集时间
    
    private Date createTime;             // 创建时间
}

/**
 * Cassandra节点监控数据实体类
 */
@Data
@TableName("cassandra_node_monitor")
public class CassandraNodeMonitor {
    
    @TableId(type = IdType.AUTO)
    private Long id;                    // 主键ID
    
    private String clusterName;          // 集群名称
    
    private String nodeAddress;          // 节点地址
    
    private String nodeStatus;           // 节点状态
    
    private String nodeState;            // 节点状态
    
    private Long dataSize;               // 数据大小
    
    private Long diskSpaceUsed;          // 已使用磁盘空间
    
    private Long diskSpaceTotal;         // 总磁盘空间
    
    private Double diskUsage;            // 磁盘使用率
    
    private Long memoryUsed;             // 内存使用量
    
    private Long memoryTotal;            // 总内存
    
    private Double memoryUsage;          // 内存使用率
    
    private Double cpuUsage;             // CPU使用率
    
    private Integer activeConnections;   // 活跃连接数
    
    private Long readOperations;          // 读操作数
    
    private Long writeOperations;        // 写操作数
    
    private Double readLatency;          // 读延迟
    
    private Double writeLatency;         // 写延迟
    
    private Integer pendingCompactions;  // 待压缩任务数
    
    private String nodeStatus;           // 节点状态
    
    private Date collectTime;            // 采集时间
    
    private Date createTime;             // 创建时间
}

3.2 Cassandra监控服务

/**
 * Cassandra监控服务
 * 负责Cassandra集群和节点数据的采集、存储和分析
 */
@Service
public class CassandraMonitorService {

    @Autowired
    private CassandraClusterMonitorMapper cassandraClusterMonitorMapper;
    
    @Autowired
    private CassandraNodeMonitorMapper cassandraNodeMonitorMapper;
    
    @Autowired
    private RedisTemplate<String, Object> redisTemplate;
    
    @Autowired
    private AlertService alertService;
    
    private CqlSession cqlSession;
    private Cluster cluster;

    /**
     * 初始化Cassandra客户端
     */
    @PostConstruct
    public void initCassandraClient() {
        try {
            // 创建CqlSession
            cqlSession = CqlSession.builder()
                    .addContactPoint(new InetSocketAddress("localhost", 9042))
                    .withLocalDatacenter("datacenter1")
                    .build();
            
            // 创建Cluster
            cluster = Cluster.builder()
                    .addContactPoint("localhost")
                    .withPort(9042)
                    .build();
            
            log.info("Cassandra客户端初始化成功");
            
        } catch (Exception e) {
            log.error("初始化Cassandra客户端失败: {}", e.getMessage(), e);
        }
    }

    /**
     * 采集Cassandra集群数据
     * 定期采集Cassandra集群和节点信息
     */
    @Scheduled(fixedRate = 10000) // 每10秒执行一次
    public void collectCassandraData() {
        try {
            // 1. 采集集群信息
            collectClusterInfo();
            
            // 2. 采集节点信息
            collectNodeInfo();
            
        } catch (Exception e) {
            log.error("采集Cassandra数据失败: {}", e.getMessage(), e);
        }
    }

    /**
     * 采集集群信息
     */
    private void collectClusterInfo() {
        try {
            // 1. 获取集群信息
            CassandraClusterInfo clusterInfo = getClusterInfo();
            
            // 2. 创建集群监控数据
            CassandraClusterMonitor monitorData = createClusterMonitorData(clusterInfo);
            
            // 3. 保存到数据库
            cassandraClusterMonitorMapper.insert(monitorData);
            
            // 4. 更新缓存
            updateClusterCache(monitorData);
            
            // 5. 检查集群告警
            checkClusterAlert(monitorData);
            
            log.debug("采集集群信息: nodeCount={}, keyspaceCount={}", 
                    monitorData.getNodeCount(), monitorData.getKeyspaceCount());
                    
        } catch (Exception e) {
            log.error("采集集群信息失败: {}", e.getMessage(), e);
        }
    }

    /**
     * 获取集群信息
     */
    private CassandraClusterInfo getClusterInfo() {
        CassandraClusterInfo clusterInfo = new CassandraClusterInfo();
        
        try {
            // 获取集群元数据
            Metadata metadata = cqlSession.getMetadata();
            
            // 获取节点信息
            Map<InetSocketAddress, Node> nodes = metadata.getNodes();
            clusterInfo.setNodeCount(nodes.size());
            
            // 获取键空间信息
            Map<CqlIdentifier, KeyspaceMetadata> keyspaces = metadata.getKeyspaces();
            clusterInfo.setKeyspaceCount(keyspaces.size());
            
            // 计算表数量
            int tableCount = 0;
            for (KeyspaceMetadata keyspace : keyspaces.values()) {
                tableCount += keyspace.getTables().size();
            }
            clusterInfo.setTableCount(tableCount);
            
            // 获取集群统计信息
            setClusterStatistics(clusterInfo);
            
            // 设置集群状态
            clusterInfo.setClusterStatus(determineClusterStatus(clusterInfo));
            
        } catch (Exception e) {
            log.error("获取集群信息失败: {}", e.getMessage(), e);
        }
        
        return clusterInfo;
    }

    /**
     * 设置集群统计信息
     */
    private void setClusterStatistics(CassandraClusterInfo clusterInfo) {
        try {
            // 获取集群统计信息
            ResultSet resultSet = cqlSession.execute("SELECT * FROM system.size_estimates");
            
            long totalDataSize = 0;
            for (Row row : resultSet) {
                totalDataSize += row.getLong("mean_partition_size");
            }
            clusterInfo.setTotalDataSize(totalDataSize);
            
            // 设置其他统计信息
            clusterInfo.setTotalDiskSpace(calculateTotalDiskSpace());
            clusterInfo.setDiskUsage(calculateDiskUsage());
            clusterInfo.setActiveConnections(calculateActiveConnections());
            clusterInfo.setThroughput(calculateThroughput());
            clusterInfo.setAvgLatency(calculateAvgLatency());
            clusterInfo.setPendingCompactions(calculatePendingCompactions());
            clusterInfo.setFailedOperations(calculateFailedOperations());
            
        } catch (Exception e) {
            log.error("设置集群统计信息失败: {}", e.getMessage(), e);
        }
    }

    /**
     * 计算总磁盘空间
     */
    private long calculateTotalDiskSpace() {
        try {
            ResultSet resultSet = cqlSession.execute("SELECT * FROM system.local");
            Row row = resultSet.one();
            return row != null ? row.getLong("total_space") : 0;
        } catch (Exception e) {
            return 0;
        }
    }

    /**
     * 计算磁盘使用率
     */
    private double calculateDiskUsage() {
        try {
            ResultSet resultSet = cqlSession.execute("SELECT * FROM system.local");
            Row row = resultSet.one();
            if (row != null) {
                long usedSpace = row.getLong("used_space");
                long totalSpace = row.getLong("total_space");
                return totalSpace > 0 ? (double) usedSpace / totalSpace * 100 : 0;
            }
        } catch (Exception e) {
            log.error("计算磁盘使用率失败: {}", e.getMessage());
        }
        return 0;
    }

    /**
     * 计算活跃连接数
     */
    private int calculateActiveConnections() {
        try {
            ResultSet resultSet = cqlSession.execute("SELECT * FROM system.clients");
            int count = 0;
            for (Row row : resultSet) {
                count++;
            }
            return count;
        } catch (Exception e) {
            return 0;
        }
    }

    /**
     * 计算吞吐量
     */
    private double calculateThroughput() {
        try {
            ResultSet resultSet = cqlSession.execute("SELECT * FROM system.local");
            Row row = resultSet.one();
            return row != null ? row.getDouble("throughput") : 0;
        } catch (Exception e) {
            return 0;
        }
    }

    /**
     * 计算平均延迟
     */
    private double calculateAvgLatency() {
        try {
            ResultSet resultSet = cqlSession.execute("SELECT * FROM system.local");
            Row row = resultSet.one();
            return row != null ? row.getDouble("avg_latency") : 0;
        } catch (Exception e) {
            return 0;
        }
    }

    /**
     * 计算待压缩任务数
     */
    private int calculatePendingCompactions() {
        try {
            ResultSet resultSet = cqlSession.execute("SELECT * FROM system.compaction_history");
            int count = 0;
            for (Row row : resultSet) {
                if ("PENDING".equals(row.getString("status"))) {
                    count++;
                }
            }
            return count;
        } catch (Exception e) {
            return 0;
        }
    }

    /**
     * 计算失败操作数
     */
    private int calculateFailedOperations() {
        try {
            ResultSet resultSet = cqlSession.execute("SELECT * FROM system.local");
            Row row = resultSet.one();
            return row != null ? row.getInt("failed_operations") : 0;
        } catch (Exception e) {
            return 0;
        }
    }

    /**
     * 确定集群状态
     */
    private String determineClusterStatus(CassandraClusterInfo clusterInfo) {
        String status = "Healthy";
        
        if (clusterInfo.getDiskUsage() > 90) {
            status = "HighDiskUsage";
        } else if (clusterInfo.getPendingCompactions() > 100) {
            status = "HighCompactionLoad";
        } else if (clusterInfo.getFailedOperations() > 10) {
            status = "HighFailureRate";
        } else if (clusterInfo.getAvgLatency() > 100) {
            status = "HighLatency";
        }
        
        return status;
    }

    /**
     * 创建集群监控数据
     */
    private CassandraClusterMonitor createClusterMonitorData(CassandraClusterInfo clusterInfo) {
        CassandraClusterMonitor monitorData = new CassandraClusterMonitor();
        
        // 设置基本信息
        monitorData.setClusterName(clusterInfo.getClusterName());
        monitorData.setHostname(getHostname());
        monitorData.setIp(getLocalIpAddress());
        monitorData.setCollectTime(new Date());
        monitorData.setCreateTime(new Date());
        
        // 设置集群信息
        monitorData.setNodeCount(clusterInfo.getNodeCount());
        monitorData.setKeyspaceCount(clusterInfo.getKeyspaceCount());
        monitorData.setTableCount(clusterInfo.getTableCount());
        
        // 设置统计信息
        monitorData.setTotalDataSize(clusterInfo.getTotalDataSize());
        monitorData.setTotalDiskSpace(clusterInfo.getTotalDiskSpace());
        monitorData.setDiskUsage(clusterInfo.getDiskUsage());
        monitorData.setActiveConnections(clusterInfo.getActiveConnections());
        
        // 设置性能信息
        monitorData.setThroughput(clusterInfo.getThroughput());
        monitorData.setAvgLatency(clusterInfo.getAvgLatency());
        
        // 设置状态信息
        monitorData.setPendingCompactions(clusterInfo.getPendingCompactions());
        monitorData.setFailedOperations(clusterInfo.getFailedOperations());
        monitorData.setClusterStatus(clusterInfo.getClusterStatus());
        
        return monitorData;
    }

    /**
     * 采集节点信息
     */
    private void collectNodeInfo() {
        try {
            // 获取所有节点
            Metadata metadata = cqlSession.getMetadata();
            Map<InetSocketAddress, Node> nodes = metadata.getNodes();
            
            for (Map.Entry<InetSocketAddress, Node> entry : nodes.entrySet()) {
                try {
                    InetSocketAddress address = entry.getKey();
                    Node node = entry.getValue();
                    
                    // 创建节点监控数据
                    CassandraNodeMonitor nodeMonitor = createNodeMonitorData(address, node);
                    
                    // 保存到数据库
                    cassandraNodeMonitorMapper.insert(nodeMonitor);
                    
                    // 更新缓存
                    updateNodeCache(nodeMonitor);
                    
                    // 检查节点告警
                    checkNodeAlert(nodeMonitor);
                    
                } catch (Exception e) {
                    log.error("处理节点失败: address={}, error={}", 
                            entry.getKey(), e.getMessage());
                }
            }
            
        } catch (Exception e) {
            log.error("采集节点信息失败: {}", e.getMessage(), e);
        }
    }

    /**
     * 创建节点监控数据
     */
    private CassandraNodeMonitor createNodeMonitorData(InetSocketAddress address, Node node) {
        CassandraNodeMonitor nodeMonitor = new CassandraNodeMonitor();
        
        // 设置基本信息
        nodeMonitor.setClusterName("production-cluster");
        nodeMonitor.setNodeAddress(address.toString());
        nodeMonitor.setCollectTime(new Date());
        nodeMonitor.setCreateTime(new Date());
        
        // 设置节点状态
        nodeMonitor.setNodeStatus(node.getState().toString());
        nodeMonitor.setNodeState(node.getState().toString());
        
        // 设置资源信息
        setNodeResourceInfo(nodeMonitor, address);
        
        // 设置性能信息
        setNodePerformanceInfo(nodeMonitor, address);
        
        return nodeMonitor;
    }

    /**
     * 设置节点资源信息
     */
    private void setNodeResourceInfo(CassandraNodeMonitor nodeMonitor, InetSocketAddress address) {
        try {
            // 获取节点资源信息
            ResultSet resultSet = cqlSession.execute("SELECT * FROM system.local WHERE key = ?", address.toString());
            Row row = resultSet.one();
            
            if (row != null) {
                nodeMonitor.setDataSize(row.getLong("data_size"));
                nodeMonitor.setDiskSpaceUsed(row.getLong("disk_space_used"));
                nodeMonitor.setDiskSpaceTotal(row.getLong("disk_space_total"));
                nodeMonitor.setDiskUsage((double) row.getLong("disk_space_used") / row.getLong("disk_space_total") * 100);
                nodeMonitor.setMemoryUsed(row.getLong("memory_used"));
                nodeMonitor.setMemoryTotal(row.getLong("memory_total"));
                nodeMonitor.setMemoryUsage((double) row.getLong("memory_used") / row.getLong("memory_total") * 100);
                nodeMonitor.setCpuUsage(row.getDouble("cpu_usage"));
            }
            
        } catch (Exception e) {
            log.error("设置节点资源信息失败: {}", e.getMessage(), e);
        }
    }

    /**
     * 设置节点性能信息
     */
    private void setNodePerformanceInfo(CassandraNodeMonitor nodeMonitor, InetSocketAddress address) {
        try {
            // 获取节点性能信息
            ResultSet resultSet = cqlSession.execute("SELECT * FROM system.local WHERE key = ?", address.toString());
            Row row = resultSet.one();
            
            if (row != null) {
                nodeMonitor.setActiveConnections(row.getInt("active_connections"));
                nodeMonitor.setReadOperations(row.getLong("read_operations"));
                nodeMonitor.setWriteOperations(row.getLong("write_operations"));
                nodeMonitor.setReadLatency(row.getDouble("read_latency"));
                nodeMonitor.setWriteLatency(row.getDouble("write_latency"));
                nodeMonitor.setPendingCompactions(row.getInt("pending_compactions"));
            }
            
        } catch (Exception e) {
            log.error("设置节点性能信息失败: {}", e.getMessage(), e);
        }
    }

    /**
     * 更新集群缓存
     */
    private void updateClusterCache(CassandraClusterMonitor monitorData) {
        try {
            String cacheKey = "cassandra:cluster:" + monitorData.getClusterName();
            redisTemplate.opsForValue().set(cacheKey, monitorData, Duration.ofMinutes(5));
            
        } catch (Exception e) {
            log.warn("更新集群缓存失败: {}", e.getMessage());
        }
    }

    /**
     * 更新节点缓存
     */
    private void updateNodeCache(CassandraNodeMonitor nodeMonitor) {
        try {
            String cacheKey = "cassandra:node:" + nodeMonitor.getNodeAddress();
            redisTemplate.opsForValue().set(cacheKey, nodeMonitor, Duration.ofMinutes(5));
            
        } catch (Exception e) {
            log.warn("更新节点缓存失败: {}", e.getMessage());
        }
    }

    /**
     * 检查集群告警
     */
    private void checkClusterAlert(CassandraClusterMonitor monitorData) {
        try {
            String alertType = null;
            String alertLevel = null;
            String alertMessage = null;
            
            // 检查磁盘使用率告警
            if (monitorData.getDiskUsage() > 90) {
                alertType = "CASSANDRA_DISK_USAGE_HIGH";
                alertLevel = "CRITICAL";
                alertMessage = String.format("Cassandra磁盘使用率过高: %.2f%%", monitorData.getDiskUsage());
            } else if (monitorData.getDiskUsage() > 80) {
                alertType = "CASSANDRA_DISK_USAGE_WARNING";
                alertLevel = "WARNING";
                alertMessage = String.format("Cassandra磁盘使用率较高: %.2f%%", monitorData.getDiskUsage());
            }
            
            // 检查待压缩任务告警
            if (monitorData.getPendingCompactions() > 100) {
                alertType = "CASSANDRA_COMPACTION_HIGH";
                alertLevel = "WARNING";
                alertMessage = String.format("Cassandra待压缩任务过多: %d", monitorData.getPendingCompactions());
            }
            
            // 检查失败操作告警
            if (monitorData.getFailedOperations() > 10) {
                alertType = "CASSANDRA_FAILURES_HIGH";
                alertLevel = "WARNING";
                alertMessage = String.format("Cassandra失败操作过多: %d", monitorData.getFailedOperations());
            }
            
            // 发送告警
            if (alertType != null) {
                sendClusterAlert(monitorData, alertType, alertLevel, alertMessage);
            }
            
        } catch (Exception e) {
            log.error("检查集群告警失败: {}", e.getMessage(), e);
        }
    }

    /**
     * 检查节点告警
     */
    private void checkNodeAlert(CassandraNodeMonitor nodeMonitor) {
        try {
            String alertType = null;
            String alertLevel = null;
            String alertMessage = null;
            
            // 检查节点状态告警
            if ("DOWN".equals(nodeMonitor.getNodeStatus())) {
                alertType = "CASSANDRA_NODE_DOWN";
                alertLevel = "CRITICAL";
                alertMessage = String.format("Cassandra节点离线: %s", nodeMonitor.getNodeAddress());
            }
            
            // 检查磁盘使用率告警
            if (nodeMonitor.getDiskUsage() > 90) {
                alertType = "CASSANDRA_NODE_DISK_HIGH";
                alertLevel = "CRITICAL";
                alertMessage = String.format("Cassandra节点磁盘使用率过高: %s, %.2f%%", 
                        nodeMonitor.getNodeAddress(), nodeMonitor.getDiskUsage());
            }
            
            // 检查内存使用率告警
            if (nodeMonitor.getMemoryUsage() > 90) {
                alertType = "CASSANDRA_NODE_MEMORY_HIGH";
                alertLevel = "WARNING";
                alertMessage = String.format("Cassandra节点内存使用率过高: %s, %.2f%%", 
                        nodeMonitor.getNodeAddress(), nodeMonitor.getMemoryUsage());
            }
            
            // 发送告警
            if (alertType != null) {
                sendNodeAlert(nodeMonitor, alertType, alertLevel, alertMessage);
            }
            
        } catch (Exception e) {
            log.error("检查节点告警失败: {}", e.getMessage(), e);
        }
    }

    /**
     * 发送集群告警
     */
    private void sendClusterAlert(CassandraClusterMonitor monitorData, String alertType, String alertLevel, String alertMessage) {
        try {
            String alertKey = "cassandra: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 sendNodeAlert(CassandraNodeMonitor nodeMonitor, String alertType, String alertLevel, String alertMessage) {
        try {
            String alertKey = "cassandra:node:alert:" + nodeMonitor.getNodeAddress() + ":" + 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(nodeMonitor.getNodeAddress());
                
                alertService.sendAlert(alert);
                
                redisTemplate.opsForValue().set(alertKey, "1", Duration.ofMinutes(5));
                
                log.warn("发送节点告警: nodeAddress={}, type={}, level={}", 
                        nodeMonitor.getNodeAddress(), alertType, alertLevel);
            }
            
        } catch (Exception e) {
            log.error("发送节点告警失败: {}", e.getMessage(), e);
        }
    }

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

    /**
     * 获取实时节点数据
     */
    public CassandraNodeMonitor getRealTimeNodeData(String nodeAddress) {
        String cacheKey = "cassandra:node:" + nodeAddress;
        return (CassandraNodeMonitor) 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";
        }
    }

    /**
     * 关闭Cassandra客户端
     */
    @PreDestroy
    public void shutdownCassandraClient() {
        try {
            if (cqlSession != null) {
                cqlSession.close();
            }
            
            if (cluster != null) {
                cluster.close();
            }
            
            log.info("Cassandra客户端关闭成功");
            
        } catch (Exception e) {
            log.error("关闭Cassandra客户端失败: {}", e.getMessage(), e);
        }
    }
}

/**
 * Cassandra集群信息实体类
 */
@Data
public class CassandraClusterInfo {
    private String clusterName;          // 集群名称
    private Integer nodeCount;           // 节点数量
    private Integer keyspaceCount;       // 键空间数量
    private Integer tableCount;          // 表数量
    private Long totalDataSize;          // 总数据大小
    private Long totalDiskSpace;         // 总磁盘空间
    private Double diskUsage;            // 磁盘使用率
    private Integer activeConnections;   // 活跃连接数
    private Double throughput;           // 吞吐量
    private Double avgLatency;           // 平均延迟
    private Integer pendingCompactions;  // 待压缩任务数
    private Integer failedOperations;    // 失败操作数
    private String clusterStatus;        // 集群状态
}

4. Cassandra管理服务

4.1 Cassandra管理服务

/**
 * Cassandra管理服务
 * 提供Cassandra集群管理功能
 */
@Service
public class CassandraManagementService {

    @Autowired
    private CqlSession cqlSession;
    
    @Autowired
    private AlertService alertService;

    /**
     * 创建键空间
     */
    public void createKeyspace(String keyspaceName, String replicationStrategy, int replicationFactor) {
        try {
            String cql = String.format(
                "CREATE KEYSPACE IF NOT EXISTS %s WITH REPLICATION = {'class': '%s', 'replication_factor': %d}",
                keyspaceName, replicationStrategy, replicationFactor
            );
            
            cqlSession.execute(cql);
            
            log.info("创建键空间成功: keyspaceName={}, replicationStrategy={}, replicationFactor={}", 
                    keyspaceName, replicationStrategy, replicationFactor);
            
        } catch (Exception e) {
            log.error("创建键空间失败: keyspaceName={}, error={}", keyspaceName, e.getMessage(), e);
        }
    }

    /**
     * 删除键空间
     */
    public void dropKeyspace(String keyspaceName) {
        try {
            String cql = String.format("DROP KEYSPACE IF EXISTS %s", keyspaceName);
            cqlSession.execute(cql);
            
            log.info("删除键空间成功: keyspaceName={}", keyspaceName);
            
        } catch (Exception e) {
            log.error("删除键空间失败: keyspaceName={}, error={}", keyspaceName, e.getMessage(), e);
        }
    }

    /**
     * 创建表
     */
    public void createTable(String keyspaceName, String tableName, String columns) {
        try {
            String cql = String.format(
                "CREATE TABLE IF NOT EXISTS %s.%s (%s)",
                keyspaceName, tableName, columns
            );
            
            cqlSession.execute(cql);
            
            log.info("创建表成功: keyspaceName={}, tableName={}", keyspaceName, tableName);
            
        } catch (Exception e) {
            log.error("创建表失败: keyspaceName={}, tableName={}, error={}", 
                    keyspaceName, tableName, e.getMessage(), e);
        }
    }

    /**
     * 删除表
     */
    public void dropTable(String keyspaceName, String tableName) {
        try {
            String cql = String.format("DROP TABLE IF EXISTS %s.%s", keyspaceName, tableName);
            cqlSession.execute(cql);
            
            log.info("删除表成功: keyspaceName={}, tableName={}", keyspaceName, tableName);
            
        } catch (Exception e) {
            log.error("删除表失败: keyspaceName={}, tableName={}, error={}", 
                    keyspaceName, tableName, e.getMessage(), e);
        }
    }

    /**
     * 执行数据修复
     */
    public void repairData(String keyspaceName, String tableName) {
        try {
            String cql = String.format("REPAIR %s.%s", keyspaceName, tableName);
            cqlSession.execute(cql);
            
            log.info("数据修复成功: keyspaceName={}, tableName={}", keyspaceName, tableName);
            
        } catch (Exception e) {
            log.error("数据修复失败: keyspaceName={}, tableName={}, error={}", 
                    keyspaceName, tableName, e.getMessage(), e);
        }
    }

    /**
     * 执行压缩
     */
    public void compactData(String keyspaceName, String tableName) {
        try {
            String cql = String.format("COMPACT %s.%s", keyspaceName, tableName);
            cqlSession.execute(cql);
            
            log.info("数据压缩成功: keyspaceName={}, tableName={}", keyspaceName, tableName);
            
        } catch (Exception e) {
            log.error("数据压缩失败: keyspaceName={}, tableName={}, error={}", 
                    keyspaceName, tableName, e.getMessage(), e);
        }
    }

    /**
     * 获取键空间列表
     */
    public List<String> getKeyspaces() {
        try {
            ResultSet resultSet = cqlSession.execute("SELECT keyspace_name FROM system_schema.keyspaces");
            List<String> keyspaces = new ArrayList<>();
            
            for (Row row : resultSet) {
                keyspaces.add(row.getString("keyspace_name"));
            }
            
            return keyspaces;
            
        } catch (Exception e) {
            log.error("获取键空间列表失败: {}", e.getMessage(), e);
            return new ArrayList<>();
        }
    }

    /**
     * 获取表列表
     */
    public List<String> getTables(String keyspaceName) {
        try {
            String cql = "SELECT table_name FROM system_schema.tables WHERE keyspace_name = ?";
            ResultSet resultSet = cqlSession.execute(cql, keyspaceName);
            List<String> tables = new ArrayList<>();
            
            for (Row row : resultSet) {
                tables.add(row.getString("table_name"));
            }
            
            return tables;
            
        } catch (Exception e) {
            log.error("获取表列表失败: keyspaceName={}, error={}", keyspaceName, e.getMessage(), e);
            return new ArrayList<>();
        }
    }

    /**
     * 获取节点状态
     */
    public Map<String, String> getNodeStatus() {
        try {
            ResultSet resultSet = cqlSession.execute("SELECT peer, state FROM system.peers");
            Map<String, String> nodeStatus = new HashMap<>();
            
            for (Row row : resultSet) {
                String peer = row.getInetAddress("peer").toString();
                String state = row.getString("state");
                nodeStatus.put(peer, state);
            }
            
            return nodeStatus;
            
        } catch (Exception e) {
            log.error("获取节点状态失败: {}", e.getMessage(), e);
            return new HashMap<>();
        }
    }
}

5. 总结

本文详细介绍了Cassandra数据库运维监控与管理的完整解决方案，包括：

5.1 核心技术点

集群监控: 实时监控Cassandra集群和节点状态
节点管理: 管理节点状态和资源使用
性能监控: 监控吞吐量、延迟、压缩任务
数据管理: 管理键空间、表、数据修复
告警通知: 多级告警、智能通知

5.2 架构优势

实时监控: 10秒间隔的实时Cassandra数据采集
智能告警: 基于阈值的智能告警机制
自动化管理: 自动化的键空间和表管理
多维度分析: 集群、节点、性能等多维度分析

5.3 最佳实践

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

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