You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@druid.apache.org by GitBox <gi...@apache.org> on 2019/03/18 19:32:06 UTC
[GitHub] [incubator-druid] leventov commented on a change in pull request
#4320: Replace usages of CountingMap with Object2LongMap
leventov commented on a change in pull request #4320: Replace usages of CountingMap with Object2LongMap
URL: https://github.com/apache/incubator-druid/pull/4320#discussion_r266606780
##########
File path: server/src/main/java/io/druid/server/coordinator/helper/DruidCoordinatorLogger.java
##########
@@ -213,91 +197,92 @@ public Number transformEntry(String key, AtomicLong value)
}
// Emit coordinator metrics
- final Set<Map.Entry<String, LoadQueuePeon>> peonEntries = params.getLoadManagementPeons().entrySet();
- for (Map.Entry<String, LoadQueuePeon> entry : peonEntries) {
- String serverName = entry.getKey();
- LoadQueuePeon queuePeon = entry.getValue();
- emitter.emit(
- new ServiceMetricEvent.Builder()
- .setDimension(DruidMetrics.SERVER, serverName).build(
- "segment/loadQueue/size", queuePeon.getLoadQueueSize()
- )
- );
- emitter.emit(
- new ServiceMetricEvent.Builder()
- .setDimension(DruidMetrics.SERVER, serverName).build(
- "segment/loadQueue/failed", queuePeon.getAndResetFailedAssignCount()
- )
- );
- emitter.emit(
- new ServiceMetricEvent.Builder()
- .setDimension(DruidMetrics.SERVER, serverName).build(
- "segment/loadQueue/count", queuePeon.getSegmentsToLoad().size()
- )
- );
- emitter.emit(
- new ServiceMetricEvent.Builder()
- .setDimension(DruidMetrics.SERVER, serverName).build(
- "segment/dropQueue/count", queuePeon.getSegmentsToDrop().size()
- )
- );
- }
- for (Map.Entry<String, AtomicLong> entry : coordinator.getSegmentAvailability().entrySet()) {
- String datasource = entry.getKey();
- Long count = entry.getValue().get();
- emitter.emit(
+ params
+ .getLoadManagementPeons()
+ .forEach(
+ (final String serverName, final LoadQueuePeon queuePeon) -> {
+ emitter.emit(
+ new ServiceMetricEvent.Builder()
+ .setDimension(DruidMetrics.SERVER, serverName).build(
+ "segment/loadQueue/size", queuePeon.getLoadQueueSize()
+ )
+ );
+ emitter.emit(
+ new ServiceMetricEvent.Builder()
+ .setDimension(DruidMetrics.SERVER, serverName).build(
+ "segment/loadQueue/failed", queuePeon.getAndResetFailedAssignCount()
+ )
+ );
+ emitter.emit(
+ new ServiceMetricEvent.Builder()
+ .setDimension(DruidMetrics.SERVER, serverName).build(
+ "segment/loadQueue/count", queuePeon.getSegmentsToLoad().size()
+ )
+ );
+ emitter.emit(
+ new ServiceMetricEvent.Builder()
+ .setDimension(DruidMetrics.SERVER, serverName).build(
+ "segment/dropQueue/count", queuePeon.getSegmentsToDrop().size()
+ )
+ );
+ }
+ );
+
+ coordinator.getSegmentAvailability().object2LongEntrySet().forEach(
+ (final Object2LongMap.Entry<String> entry) -> {
+ final String dataSource = entry.getKey();
+ final long count = entry.getLongValue();
+ emitter.emit(
new ServiceMetricEvent.Builder()
- .setDimension(DruidMetrics.DATASOURCE, datasource).build(
- "segment/unavailable/count", count
+ .setDimension(DruidMetrics.DATASOURCE, dataSource).build(
+ "segment/unavailable/count", count
)
- );
- }
- for (Map.Entry<String, CountingMap<String>> entry : coordinator.getReplicationStatus().entrySet()) {
- String tier = entry.getKey();
- CountingMap<String> datasourceAvailabilities = entry.getValue();
- for (Map.Entry<String, AtomicLong> datasourceAvailability : datasourceAvailabilities.entrySet()) {
- String datasource = datasourceAvailability.getKey();
- Long count = datasourceAvailability.getValue().get();
- emitter.emit(
+ );
+ }
+ );
+
+ coordinator.getReplicationStatus().forEach(
+ (final String tier, final Object2LongMap<String> status) -> {
+ for (final Object2LongMap.Entry<String> entry : status.object2LongEntrySet()) {
+ final String dataSource = entry.getKey();
+ final long count = entry.getLongValue();
+
+ emitter.emit(
new ServiceMetricEvent.Builder()
- .setDimension(DruidMetrics.TIER, tier)
- .setDimension(DruidMetrics.DATASOURCE, datasource).build(
- "segment/underReplicated/count", count
+ .setDimension(DruidMetrics.TIER, tier)
+ .setDimension(DruidMetrics.DATASOURCE, dataSource).build(
+ "segment/underReplicated/count", count
)
- );
- }
- }
+ );
+ }
+ }
+ );
// Emit segment metrics
- CountingMap<String> segmentSizes = new CountingMap<String>();
- CountingMap<String> segmentCounts = new CountingMap<String>();
- for (DruidDataSource dataSource : params.getDataSources()) {
- for (DataSegment segment : dataSource.getSegments()) {
- segmentSizes.add(dataSource.getName(), segment.getSize());
- segmentCounts.add(dataSource.getName(), 1L);
- }
- }
- for (Map.Entry<String, Long> entry : segmentSizes.snapshot().entrySet()) {
- String dataSource = entry.getKey();
- Long size = entry.getValue();
- emitter.emit(
- new ServiceMetricEvent.Builder()
- .setDimension(DruidMetrics.DATASOURCE, dataSource).build(
- "segment/size", size
- )
- );
- }
- for (Map.Entry<String, Long> entry : segmentCounts.snapshot().entrySet()) {
- String dataSource = entry.getKey();
- Long count = entry.getValue();
- emitter.emit(
- new ServiceMetricEvent.Builder()
- .setDimension(DruidMetrics.DATASOURCE, dataSource).build(
- "segment/count", count
- )
- );
- }
+ final Stream<DataSegment> allSegments = params
Review comment:
Not sure why here segments are first merged from a per-data source grouping into a single Stream, and then grouped by data source again. I'll fix this.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org