@Test public void testGetNodeStatsInJSON() throws Exception { List<ReplicaId> replicaIds = new ArrayList<>(); PartitionId partitionId; DataNodeId dataNodeId = new MockDataNodeId(Collections.singletonList(new Port(6667, PortType.PLAINTEXT)), Collections.singletonList("/tmp"), "DC1"); Map<PartitionId, Store> storeMap = new HashMap<>(); List<StatsSnapshot> partitionClassSnapshots = new ArrayList<>(); List<StatsSnapshot> accountSnapshots = new ArrayList<>(); for (int i = 0; i < 2; i++) { partitionId = new MockPartitionId(i, (i % 2 == 0) ? MockClusterMap.DEFAULT_PARTITION_CLASS : MockClusterMap.SPECIAL_PARTITION_CLASS, Collections.singletonList((MockDataNodeId) dataNodeId), 0); Map<StatsReportType, StatsSnapshot> allSnapshots = generateRandomSnapshot(); partitionClassSnapshots.add(allSnapshots.get(StatsReportType.PARTITION_CLASS_REPORT)); accountSnapshots.add(allSnapshots.get(StatsReportType.ACCOUNT_REPORT)); storeMap.put(partitionId, new MockStore(new MockStoreStats(allSnapshots, false))); replicaIds.add(partitionId.getReplicaIds().get(0)); } StorageManager storageManager = new MockStorageManager(storeMap, dataNodeId); StatsManager statsManager = new StatsManager(storageManager, replicaIds, new MetricRegistry(), statsManagerConfig, new MockTime(), null); StatsSnapshot expectAccountSnapshot = new StatsSnapshot(0L, new HashMap<>()); StatsSnapshot expectPartitionClassSnapshot = new StatsSnapshot(0L, new HashMap<>()); for (int i = 0; i < accountSnapshots.size(); ++i) { Map<String, StatsSnapshot> partitionToAccountSnapshot = new HashMap<>(); Map<String, StatsSnapshot> partitionToPartitionClassSnapshot = new HashMap<>(); Map<String, StatsSnapshot> partitionClassSnapshotMap = new HashMap<>(); String partitionIdStr = String.valueOf(i); String partitionClassStr = i % 2 == 0 ? MockClusterMap.DEFAULT_PARTITION_CLASS : MockClusterMap.SPECIAL_PARTITION_CLASS; partitionToAccountSnapshot.put(partitionIdStr, accountSnapshots.get(i)); partitionToPartitionClassSnapshot.put(partitionIdStr, partitionClassSnapshots.get(i)); partitionClassSnapshotMap.put(partitionClassStr, new StatsSnapshot(partitionClassSnapshots.get(i).getValue(), partitionToPartitionClassSnapshot)); StatsSnapshot.aggregate(expectAccountSnapshot, new StatsSnapshot(accountSnapshots.get(i).getValue(), partitionToAccountSnapshot)); StatsSnapshot.aggregate(expectPartitionClassSnapshot, new StatsSnapshot(partitionClassSnapshots.get(i).getValue(), partitionClassSnapshotMap)); } for (StatsReportType type : EnumSet.of(StatsReportType.ACCOUNT_REPORT, StatsReportType.PARTITION_CLASS_REPORT)) { String statsInJSON = statsManager.getNodeStatsInJSON(type); StatsSnapshot actualSnapshot = mapper.readValue(statsInJSON, StatsWrapper.class).getSnapshot(); switch (type) { case ACCOUNT_REPORT: assertTrue("Mismatch in aggregated node stats at account level", expectAccountSnapshot.equals(actualSnapshot)); break; case PARTITION_CLASS_REPORT: assertTrue("Mismatch in aggregated node stats at partitionClass level", expectPartitionClassSnapshot.equals(actualSnapshot)); break; default: throw new IllegalArgumentException("Unrecognized stats report type: " + type); } } }
String getNodeStatsInJSON(StatsReportType statsReportType) { String statsWrapperJSON = ""; logger.info("Aggregating node-level stats for Helix report"); try { long totalFetchAndAggregateStartTimeMs = time.milliseconds(); StatsSnapshot combinedSnapshot = new StatsSnapshot(0L, new HashMap<>()); long totalValue = 0; List<PartitionId> unreachablePartitions = new ArrayList<>(); Set<PartitionId> partitionsBeforeAggregation = new HashSet<>(partitionToReplicaMap.keySet()); Iterator<PartitionId> iterator = partitionsBeforeAggregation.iterator(); while (iterator.hasNext()) { PartitionId partitionId = iterator.next(); long fetchSnapshotStartTimeMs = time.milliseconds(); StatsSnapshot statsSnapshot = fetchSnapshot(partitionId, unreachablePartitions, statsReportType); if (statsSnapshot != null) { Map<String, StatsSnapshot> combinedSnapshotSubMap = combinedSnapshot.getSubMap(); switch (statsReportType) { case ACCOUNT_REPORT: combinedSnapshotSubMap.put(partitionId.toString(), statsSnapshot); break; case PARTITION_CLASS_REPORT: StatsSnapshot partitionClassSnapshot = combinedSnapshotSubMap.getOrDefault(partitionId.getPartitionClass(), new StatsSnapshot(0L, new HashMap<>())); partitionClassSnapshot.setValue(partitionClassSnapshot.getValue() + statsSnapshot.getValue()); partitionClassSnapshot.getSubMap().put(partitionId.toString(), statsSnapshot); combinedSnapshotSubMap.put(partitionId.getPartitionClass(), partitionClassSnapshot); break; default: throw new IllegalArgumentException("Unrecognized stats report type: " + statsReportType); } totalValue += statsSnapshot.getValue(); } metrics.fetchAndAggregateTimePerStoreMs.update(time.milliseconds() - fetchSnapshotStartTimeMs); } combinedSnapshot.setValue(totalValue); List<String> examinedUnreachableStores = examineUnreachablePartitions(unreachablePartitions); metrics.totalFetchAndAggregateTimeMs.update(time.milliseconds() - totalFetchAndAggregateStartTimeMs); StatsHeader statsHeader = new StatsHeader(StatsHeader.StatsDescription.STORED_DATA_SIZE, time.milliseconds(), partitionsBeforeAggregation.size(), partitionsBeforeAggregation.size() - unreachablePartitions.size(), examinedUnreachableStores); statsWrapperJSON = mapper.writeValueAsString(new StatsWrapper(statsHeader, combinedSnapshot)); logger.info("Node-level stats aggregated for Helix report"); } catch (Exception | Error e) { metrics.statsAggregationFailureCount.inc(); logger.error("Exception while aggregating stats for Helix report", e); } return statsWrapperJSON; }
StatsManager { String getNodeStatsInJSON(StatsReportType statsReportType) { String statsWrapperJSON = ""; logger.info("Aggregating node-level stats for Helix report"); try { long totalFetchAndAggregateStartTimeMs = time.milliseconds(); StatsSnapshot combinedSnapshot = new StatsSnapshot(0L, new HashMap<>()); long totalValue = 0; List<PartitionId> unreachablePartitions = new ArrayList<>(); Set<PartitionId> partitionsBeforeAggregation = new HashSet<>(partitionToReplicaMap.keySet()); Iterator<PartitionId> iterator = partitionsBeforeAggregation.iterator(); while (iterator.hasNext()) { PartitionId partitionId = iterator.next(); long fetchSnapshotStartTimeMs = time.milliseconds(); StatsSnapshot statsSnapshot = fetchSnapshot(partitionId, unreachablePartitions, statsReportType); if (statsSnapshot != null) { Map<String, StatsSnapshot> combinedSnapshotSubMap = combinedSnapshot.getSubMap(); switch (statsReportType) { case ACCOUNT_REPORT: combinedSnapshotSubMap.put(partitionId.toString(), statsSnapshot); break; case PARTITION_CLASS_REPORT: StatsSnapshot partitionClassSnapshot = combinedSnapshotSubMap.getOrDefault(partitionId.getPartitionClass(), new StatsSnapshot(0L, new HashMap<>())); partitionClassSnapshot.setValue(partitionClassSnapshot.getValue() + statsSnapshot.getValue()); partitionClassSnapshot.getSubMap().put(partitionId.toString(), statsSnapshot); combinedSnapshotSubMap.put(partitionId.getPartitionClass(), partitionClassSnapshot); break; default: throw new IllegalArgumentException("Unrecognized stats report type: " + statsReportType); } totalValue += statsSnapshot.getValue(); } metrics.fetchAndAggregateTimePerStoreMs.update(time.milliseconds() - fetchSnapshotStartTimeMs); } combinedSnapshot.setValue(totalValue); List<String> examinedUnreachableStores = examineUnreachablePartitions(unreachablePartitions); metrics.totalFetchAndAggregateTimeMs.update(time.milliseconds() - totalFetchAndAggregateStartTimeMs); StatsHeader statsHeader = new StatsHeader(StatsHeader.StatsDescription.STORED_DATA_SIZE, time.milliseconds(), partitionsBeforeAggregation.size(), partitionsBeforeAggregation.size() - unreachablePartitions.size(), examinedUnreachableStores); statsWrapperJSON = mapper.writeValueAsString(new StatsWrapper(statsHeader, combinedSnapshot)); logger.info("Node-level stats aggregated for Helix report"); } catch (Exception | Error e) { metrics.statsAggregationFailureCount.inc(); logger.error("Exception while aggregating stats for Helix report", e); } return statsWrapperJSON; } }
StatsManager { String getNodeStatsInJSON(StatsReportType statsReportType) { String statsWrapperJSON = ""; logger.info("Aggregating node-level stats for Helix report"); try { long totalFetchAndAggregateStartTimeMs = time.milliseconds(); StatsSnapshot combinedSnapshot = new StatsSnapshot(0L, new HashMap<>()); long totalValue = 0; List<PartitionId> unreachablePartitions = new ArrayList<>(); Set<PartitionId> partitionsBeforeAggregation = new HashSet<>(partitionToReplicaMap.keySet()); Iterator<PartitionId> iterator = partitionsBeforeAggregation.iterator(); while (iterator.hasNext()) { PartitionId partitionId = iterator.next(); long fetchSnapshotStartTimeMs = time.milliseconds(); StatsSnapshot statsSnapshot = fetchSnapshot(partitionId, unreachablePartitions, statsReportType); if (statsSnapshot != null) { Map<String, StatsSnapshot> combinedSnapshotSubMap = combinedSnapshot.getSubMap(); switch (statsReportType) { case ACCOUNT_REPORT: combinedSnapshotSubMap.put(partitionId.toString(), statsSnapshot); break; case PARTITION_CLASS_REPORT: StatsSnapshot partitionClassSnapshot = combinedSnapshotSubMap.getOrDefault(partitionId.getPartitionClass(), new StatsSnapshot(0L, new HashMap<>())); partitionClassSnapshot.setValue(partitionClassSnapshot.getValue() + statsSnapshot.getValue()); partitionClassSnapshot.getSubMap().put(partitionId.toString(), statsSnapshot); combinedSnapshotSubMap.put(partitionId.getPartitionClass(), partitionClassSnapshot); break; default: throw new IllegalArgumentException("Unrecognized stats report type: " + statsReportType); } totalValue += statsSnapshot.getValue(); } metrics.fetchAndAggregateTimePerStoreMs.update(time.milliseconds() - fetchSnapshotStartTimeMs); } combinedSnapshot.setValue(totalValue); List<String> examinedUnreachableStores = examineUnreachablePartitions(unreachablePartitions); metrics.totalFetchAndAggregateTimeMs.update(time.milliseconds() - totalFetchAndAggregateStartTimeMs); StatsHeader statsHeader = new StatsHeader(StatsHeader.StatsDescription.STORED_DATA_SIZE, time.milliseconds(), partitionsBeforeAggregation.size(), partitionsBeforeAggregation.size() - unreachablePartitions.size(), examinedUnreachableStores); statsWrapperJSON = mapper.writeValueAsString(new StatsWrapper(statsHeader, combinedSnapshot)); logger.info("Node-level stats aggregated for Helix report"); } catch (Exception | Error e) { metrics.statsAggregationFailureCount.inc(); logger.error("Exception while aggregating stats for Helix report", e); } return statsWrapperJSON; } StatsManager(StorageManager storageManager, List<? extends ReplicaId> replicaIds, MetricRegistry registry,
      StatsManagerConfig config, Time time, ClusterParticipant clusterParticipant); }
StatsManager { String getNodeStatsInJSON(StatsReportType statsReportType) { String statsWrapperJSON = ""; logger.info("Aggregating node-level stats for Helix report"); try { long totalFetchAndAggregateStartTimeMs = time.milliseconds(); StatsSnapshot combinedSnapshot = new StatsSnapshot(0L, new HashMap<>()); long totalValue = 0; List<PartitionId> unreachablePartitions = new ArrayList<>(); Set<PartitionId> partitionsBeforeAggregation = new HashSet<>(partitionToReplicaMap.keySet()); Iterator<PartitionId> iterator = partitionsBeforeAggregation.iterator(); while (iterator.hasNext()) { PartitionId partitionId = iterator.next(); long fetchSnapshotStartTimeMs = time.milliseconds(); StatsSnapshot statsSnapshot = fetchSnapshot(partitionId, unreachablePartitions, statsReportType); if (statsSnapshot != null) { Map<String, StatsSnapshot> combinedSnapshotSubMap = combinedSnapshot.getSubMap(); switch (statsReportType) { case ACCOUNT_REPORT: combinedSnapshotSubMap.put(partitionId.toString(), statsSnapshot); break; case PARTITION_CLASS_REPORT: StatsSnapshot partitionClassSnapshot = combinedSnapshotSubMap.getOrDefault(partitionId.getPartitionClass(), new StatsSnapshot(0L, new HashMap<>())); partitionClassSnapshot.setValue(partitionClassSnapshot.getValue() + statsSnapshot.getValue()); partitionClassSnapshot.getSubMap().put(partitionId.toString(), statsSnapshot); combinedSnapshotSubMap.put(partitionId.getPartitionClass(), partitionClassSnapshot); break; default: throw new IllegalArgumentException("Unrecognized stats report type: " + statsReportType); } totalValue += statsSnapshot.getValue(); } metrics.fetchAndAggregateTimePerStoreMs.update(time.milliseconds() - fetchSnapshotStartTimeMs); } combinedSnapshot.setValue(totalValue); List<String> examinedUnreachableStores = examineUnreachablePartitions(unreachablePartitions); metrics.totalFetchAndAggregateTimeMs.update(time.milliseconds() - totalFetchAndAggregateStartTimeMs); StatsHeader statsHeader = new StatsHeader(StatsHeader.StatsDescription.STORED_DATA_SIZE, time.milliseconds(), partitionsBeforeAggregation.size(), partitionsBeforeAggregation.size() - unreachablePartitions.size(), examinedUnreachableStores); statsWrapperJSON = mapper.writeValueAsString(new StatsWrapper(statsHeader, combinedSnapshot)); logger.info("Node-level stats aggregated for Helix report"); } catch (Exception | Error e) { metrics.statsAggregationFailureCount.inc(); logger.error("Exception while aggregating stats for Helix report", e); } return statsWrapperJSON; } StatsManager(StorageManager storageManager, List<? extends ReplicaId> replicaIds, MetricRegistry registry,
      StatsManagerConfig config, Time time, ClusterParticipant clusterParticipant);  }
StatsManager { String getNodeStatsInJSON(StatsReportType statsReportType) { String statsWrapperJSON = ""; logger.info("Aggregating node-level stats for Helix report"); try { long totalFetchAndAggregateStartTimeMs = time.milliseconds(); StatsSnapshot combinedSnapshot = new StatsSnapshot(0L, new HashMap<>()); long totalValue = 0; List<PartitionId> unreachablePartitions = new ArrayList<>(); Set<PartitionId> partitionsBeforeAggregation = new HashSet<>(partitionToReplicaMap.keySet()); Iterator<PartitionId> iterator = partitionsBeforeAggregation.iterator(); while (iterator.hasNext()) { PartitionId partitionId = iterator.next(); long fetchSnapshotStartTimeMs = time.milliseconds(); StatsSnapshot statsSnapshot = fetchSnapshot(partitionId, unreachablePartitions, statsReportType); if (statsSnapshot != null) { Map<String, StatsSnapshot> combinedSnapshotSubMap = combinedSnapshot.getSubMap(); switch (statsReportType) { case ACCOUNT_REPORT: combinedSnapshotSubMap.put(partitionId.toString(), statsSnapshot); break; case PARTITION_CLASS_REPORT: StatsSnapshot partitionClassSnapshot = combinedSnapshotSubMap.getOrDefault(partitionId.getPartitionClass(), new StatsSnapshot(0L, new HashMap<>())); partitionClassSnapshot.setValue(partitionClassSnapshot.getValue() + statsSnapshot.getValue()); partitionClassSnapshot.getSubMap().put(partitionId.toString(), statsSnapshot); combinedSnapshotSubMap.put(partitionId.getPartitionClass(), partitionClassSnapshot); break; default: throw new IllegalArgumentException("Unrecognized stats report type: " + statsReportType); } totalValue += statsSnapshot.getValue(); } metrics.fetchAndAggregateTimePerStoreMs.update(time.milliseconds() - fetchSnapshotStartTimeMs); } combinedSnapshot.setValue(totalValue); List<String> examinedUnreachableStores = examineUnreachablePartitions(unreachablePartitions); metrics.totalFetchAndAggregateTimeMs.update(time.milliseconds() - totalFetchAndAggregateStartTimeMs); StatsHeader statsHeader = new StatsHeader(StatsHeader.StatsDescription.STORED_DATA_SIZE, time.milliseconds(), partitionsBeforeAggregation.size(), partitionsBeforeAggregation.size() - unreachablePartitions.size(), examinedUnreachableStores); statsWrapperJSON = mapper.writeValueAsString(new StatsWrapper(statsHeader, combinedSnapshot)); logger.info("Node-level stats aggregated for Helix report"); } catch (Exception | Error e) { metrics.statsAggregationFailureCount.inc(); logger.error("Exception while aggregating stats for Helix report", e); } return statsWrapperJSON; } StatsManager(StorageManager storageManager, List<? extends ReplicaId> replicaIds, MetricRegistry registry,
      StatsManagerConfig config, Time time, ClusterParticipant clusterParticipant);   }