001/* 002 * Licensed to the Apache Software Foundation (ASF) under one 003 * or more contributor license agreements. See the NOTICE file 004 * distributed with this work for additional information 005 * regarding copyright ownership. The ASF licenses this file 006 * to you under the Apache License, Version 2.0 (the 007 * "License"); you may not use this file except in compliance 008 * with the License. You may obtain a copy of the License at 009 * 010 * http://www.apache.org/licenses/LICENSE-2.0 011 * 012 * Unless required by applicable law or agreed to in writing, software 013 * distributed under the License is distributed on an "AS IS" BASIS, 014 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 015 * See the License for the specific language governing permissions and 016 * limitations under the License. 017 */ 018package org.apache.hadoop.hbase; 019 020import edu.umd.cs.findbugs.annotations.Nullable; 021import java.util.ArrayList; 022import java.util.Collection; 023import java.util.Collections; 024import java.util.HashMap; 025import java.util.List; 026import java.util.Map; 027import java.util.Set; 028import java.util.TreeMap; 029import java.util.TreeSet; 030import java.util.stream.Collectors; 031import org.apache.hadoop.hbase.replication.ReplicationLoadSink; 032import org.apache.hadoop.hbase.replication.ReplicationLoadSource; 033import org.apache.hadoop.hbase.util.Bytes; 034import org.apache.hadoop.hbase.util.EnvironmentEdgeManager; 035import org.apache.hadoop.hbase.util.Strings; 036import org.apache.yetus.audience.InterfaceAudience; 037 038import org.apache.hbase.thirdparty.com.google.common.base.Preconditions; 039 040import org.apache.hadoop.hbase.shaded.protobuf.ProtobufUtil; 041import org.apache.hadoop.hbase.shaded.protobuf.generated.ClusterStatusProtos; 042import org.apache.hadoop.hbase.shaded.protobuf.generated.HBaseProtos; 043 044@InterfaceAudience.Private 045public final class ServerMetricsBuilder { 046 047 public static ServerMetrics of(ServerName sn) { 048 return newBuilder(sn).build(); 049 } 050 051 public static ServerMetrics of(ServerName sn, int versionNumber, String version) { 052 return newBuilder(sn).setVersionNumber(versionNumber).setVersion(version).build(); 053 } 054 055 public static ServerMetrics toServerMetrics(ClusterStatusProtos.LiveServerInfo serverInfo) { 056 return toServerMetrics(ProtobufUtil.toServerName(serverInfo.getServer()), 0, "0.0.0", 057 serverInfo.getServerLoad()); 058 } 059 060 public static ServerMetrics toServerMetrics(ServerName serverName, 061 ClusterStatusProtos.ServerLoad serverLoadPB) { 062 return toServerMetrics(serverName, 0, "0.0.0", serverLoadPB); 063 } 064 065 public static ServerMetrics toServerMetrics(ServerName serverName, int versionNumber, 066 String version, ClusterStatusProtos.ServerLoad serverLoadPB) { 067 return ServerMetricsBuilder.newBuilder(serverName) 068 .setRequestCountPerSecond(serverLoadPB.getNumberOfRequests()) 069 .setRequestCount(serverLoadPB.getTotalNumberOfRequests()) 070 .setInfoServerPort(serverLoadPB.getInfoServerPort()) 071 .setMaxHeapSize(new Size(serverLoadPB.getMaxHeapMB(), Size.Unit.MEGABYTE)) 072 .setUsedHeapSize(new Size(serverLoadPB.getUsedHeapMB(), Size.Unit.MEGABYTE)) 073 .setCoprocessorNames(serverLoadPB.getCoprocessorsList().stream() 074 .map(HBaseProtos.Coprocessor::getName).collect(Collectors.toList())) 075 .setRegionMetrics(serverLoadPB.getRegionLoadsList().stream() 076 .map(RegionMetricsBuilder::toRegionMetrics).collect(Collectors.toList())) 077 .setUserMetrics(serverLoadPB.getUserLoadsList().stream() 078 .map(UserMetricsBuilder::toUserMetrics).collect(Collectors.toList())) 079 .setReplicationLoadSources(serverLoadPB.getReplLoadSourceList().stream() 080 .map(ProtobufUtil::toReplicationLoadSource).collect(Collectors.toList())) 081 .setReplicationLoadSink(serverLoadPB.hasReplLoadSink() 082 ? ProtobufUtil.toReplicationLoadSink(serverLoadPB.getReplLoadSink()) 083 : null) 084 .setTasks(serverLoadPB.getTasksList().stream().map(ProtobufUtil::getServerTask) 085 .collect(Collectors.toList())) 086 .setRegionCachedInfo(serverLoadPB.getRegionCachedInfoMap()) 087 .setReportTimestamp(serverLoadPB.getReportEndTime()) 088 .setLastReportTimestamp(serverLoadPB.getReportStartTime()).setVersionNumber(versionNumber) 089 .setVersion(version).build(); 090 } 091 092 public static List<HBaseProtos.Coprocessor> toCoprocessor(Collection<String> names) { 093 return names.stream().map(n -> HBaseProtos.Coprocessor.newBuilder().setName(n).build()) 094 .collect(Collectors.toList()); 095 } 096 097 public static ClusterStatusProtos.ServerLoad toServerLoad(ServerMetrics metrics) { 098 ClusterStatusProtos.ServerLoad.Builder builder = ClusterStatusProtos.ServerLoad.newBuilder() 099 .setNumberOfRequests(metrics.getRequestCountPerSecond()) 100 .setTotalNumberOfRequests(metrics.getRequestCount()) 101 .setInfoServerPort(metrics.getInfoServerPort()) 102 .setMaxHeapMB((int) metrics.getMaxHeapSize().get(Size.Unit.MEGABYTE)) 103 .setUsedHeapMB((int) metrics.getUsedHeapSize().get(Size.Unit.MEGABYTE)) 104 .addAllCoprocessors(toCoprocessor(metrics.getCoprocessorNames())) 105 .addAllRegionLoads(metrics.getRegionMetrics().values().stream() 106 .map(RegionMetricsBuilder::toRegionLoad).collect(Collectors.toList())) 107 .addAllUserLoads(metrics.getUserMetrics().values().stream() 108 .map(UserMetricsBuilder::toUserMetrics).collect(Collectors.toList())) 109 .addAllReplLoadSource(metrics.getReplicationLoadSourceList().stream() 110 .map(ProtobufUtil::toReplicationLoadSource).collect(Collectors.toList())) 111 .addAllTasks( 112 metrics.getTasks().stream().map(ProtobufUtil::toServerTask).collect(Collectors.toList())) 113 .putAllRegionCachedInfo(metrics.getRegionCachedInfo()) 114 .setReportStartTime(metrics.getLastReportTimestamp()) 115 .setReportEndTime(metrics.getReportTimestamp()); 116 if (metrics.getReplicationLoadSink() != null) { 117 builder.setReplLoadSink(ProtobufUtil.toReplicationLoadSink(metrics.getReplicationLoadSink())); 118 } 119 return builder.build(); 120 } 121 122 public static ServerMetricsBuilder newBuilder(ServerName sn) { 123 return new ServerMetricsBuilder(sn); 124 } 125 126 private final ServerName serverName; 127 private int versionNumber; 128 private String version = "0.0.0"; 129 private long requestCountPerSecond; 130 private long requestCount; 131 private Size usedHeapSize = Size.ZERO; 132 private Size maxHeapSize = Size.ZERO; 133 private int infoServerPort; 134 private List<ReplicationLoadSource> sources = Collections.emptyList(); 135 @Nullable 136 private ReplicationLoadSink sink = null; 137 private final Map<byte[], RegionMetrics> regionStatus = new TreeMap<>(Bytes.BYTES_COMPARATOR); 138 private final Map<byte[], UserMetrics> userMetrics = new TreeMap<>(Bytes.BYTES_COMPARATOR); 139 private final Set<String> coprocessorNames = new TreeSet<>(); 140 private long reportTimestamp = EnvironmentEdgeManager.currentTime(); 141 private long lastReportTimestamp = 0; 142 private final List<ServerTask> tasks = new ArrayList<>(); 143 private Map<String, Integer> regionCachedInfo = new HashMap<>(); 144 145 private ServerMetricsBuilder(ServerName serverName) { 146 this.serverName = serverName; 147 } 148 149 public ServerMetricsBuilder setVersionNumber(int versionNumber) { 150 this.versionNumber = versionNumber; 151 return this; 152 } 153 154 public ServerMetricsBuilder setVersion(String version) { 155 this.version = version; 156 return this; 157 } 158 159 public ServerMetricsBuilder setRequestCountPerSecond(long value) { 160 this.requestCountPerSecond = value; 161 return this; 162 } 163 164 public ServerMetricsBuilder setRequestCount(long value) { 165 this.requestCount = value; 166 return this; 167 } 168 169 public ServerMetricsBuilder setUsedHeapSize(Size value) { 170 this.usedHeapSize = value; 171 return this; 172 } 173 174 public ServerMetricsBuilder setMaxHeapSize(Size value) { 175 this.maxHeapSize = value; 176 return this; 177 } 178 179 public ServerMetricsBuilder setInfoServerPort(int value) { 180 this.infoServerPort = value; 181 return this; 182 } 183 184 public ServerMetricsBuilder setReplicationLoadSources(List<ReplicationLoadSource> value) { 185 this.sources = value; 186 return this; 187 } 188 189 public ServerMetricsBuilder setReplicationLoadSink(ReplicationLoadSink value) { 190 this.sink = value; 191 return this; 192 } 193 194 public ServerMetricsBuilder setRegionMetrics(List<RegionMetrics> value) { 195 value.forEach(v -> this.regionStatus.put(v.getRegionName(), v)); 196 return this; 197 } 198 199 public ServerMetricsBuilder setUserMetrics(List<UserMetrics> value) { 200 value.forEach(v -> this.userMetrics.put(v.getUserName(), v)); 201 return this; 202 } 203 204 public ServerMetricsBuilder setCoprocessorNames(List<String> value) { 205 coprocessorNames.addAll(value); 206 return this; 207 } 208 209 public ServerMetricsBuilder setReportTimestamp(long value) { 210 this.reportTimestamp = value; 211 return this; 212 } 213 214 public ServerMetricsBuilder setLastReportTimestamp(long value) { 215 this.lastReportTimestamp = value; 216 return this; 217 } 218 219 public ServerMetricsBuilder setTasks(List<ServerTask> tasks) { 220 this.tasks.addAll(tasks); 221 return this; 222 } 223 224 public ServerMetricsBuilder setRegionCachedInfo(Map<String, Integer> value) { 225 this.regionCachedInfo = value; 226 return this; 227 } 228 229 public ServerMetrics build() { 230 return new ServerMetricsImpl(serverName, versionNumber, version, requestCountPerSecond, 231 requestCount, usedHeapSize, maxHeapSize, infoServerPort, sources, sink, regionStatus, 232 coprocessorNames, reportTimestamp, lastReportTimestamp, userMetrics, tasks, regionCachedInfo); 233 } 234 235 private static class ServerMetricsImpl implements ServerMetrics { 236 private final ServerName serverName; 237 private final int versionNumber; 238 private final String version; 239 private final long requestCountPerSecond; 240 private final long requestCount; 241 private final Size usedHeapSize; 242 private final Size maxHeapSize; 243 private final int infoServerPort; 244 private final List<ReplicationLoadSource> sources; 245 @Nullable 246 private final ReplicationLoadSink sink; 247 private final Map<byte[], RegionMetrics> regionStatus; 248 private final Set<String> coprocessorNames; 249 private final long reportTimestamp; 250 private final long lastReportTimestamp; 251 private final Map<byte[], UserMetrics> userMetrics; 252 private final List<ServerTask> tasks; 253 private final Map<String, Integer> regionCachedInfo; 254 255 ServerMetricsImpl(ServerName serverName, int versionNumber, String version, 256 long requestCountPerSecond, long requestCount, Size usedHeapSize, Size maxHeapSize, 257 int infoServerPort, List<ReplicationLoadSource> sources, ReplicationLoadSink sink, 258 Map<byte[], RegionMetrics> regionStatus, Set<String> coprocessorNames, long reportTimestamp, 259 long lastReportTimestamp, Map<byte[], UserMetrics> userMetrics, List<ServerTask> tasks, 260 Map<String, Integer> regionCachedInfo) { 261 this.serverName = Preconditions.checkNotNull(serverName); 262 this.versionNumber = versionNumber; 263 this.version = version; 264 this.requestCountPerSecond = requestCountPerSecond; 265 this.requestCount = requestCount; 266 this.usedHeapSize = Preconditions.checkNotNull(usedHeapSize); 267 this.maxHeapSize = Preconditions.checkNotNull(maxHeapSize); 268 this.infoServerPort = infoServerPort; 269 this.sources = Preconditions.checkNotNull(sources); 270 this.sink = sink; 271 this.regionStatus = Preconditions.checkNotNull(regionStatus); 272 this.userMetrics = Preconditions.checkNotNull(userMetrics); 273 this.coprocessorNames = Preconditions.checkNotNull(coprocessorNames); 274 this.reportTimestamp = reportTimestamp; 275 this.lastReportTimestamp = lastReportTimestamp; 276 this.tasks = tasks; 277 this.regionCachedInfo = regionCachedInfo; 278 } 279 280 @Override 281 public ServerName getServerName() { 282 return serverName; 283 } 284 285 @Override 286 public int getVersionNumber() { 287 return versionNumber; 288 } 289 290 @Override 291 public String getVersion() { 292 return version; 293 } 294 295 @Override 296 public long getRequestCountPerSecond() { 297 return requestCountPerSecond; 298 } 299 300 @Override 301 public long getRequestCount() { 302 return requestCount; 303 } 304 305 @Override 306 public Size getUsedHeapSize() { 307 return usedHeapSize; 308 } 309 310 @Override 311 public Size getMaxHeapSize() { 312 return maxHeapSize; 313 } 314 315 @Override 316 public int getInfoServerPort() { 317 return infoServerPort; 318 } 319 320 @Override 321 public List<ReplicationLoadSource> getReplicationLoadSourceList() { 322 return Collections.unmodifiableList(sources); 323 } 324 325 @Override 326 public Map<String, List<ReplicationLoadSource>> getReplicationLoadSourceMap() { 327 Map<String, List<ReplicationLoadSource>> sourcesMap = new HashMap<>(); 328 for (ReplicationLoadSource loadSource : sources) { 329 sourcesMap.computeIfAbsent(loadSource.getPeerID(), peerId -> new ArrayList<>()) 330 .add(loadSource); 331 } 332 return sourcesMap; 333 } 334 335 @Override 336 public ReplicationLoadSink getReplicationLoadSink() { 337 return sink; 338 } 339 340 @Override 341 public Map<byte[], RegionMetrics> getRegionMetrics() { 342 return Collections.unmodifiableMap(regionStatus); 343 } 344 345 @Override 346 public Map<byte[], UserMetrics> getUserMetrics() { 347 return Collections.unmodifiableMap(userMetrics); 348 } 349 350 @Override 351 public Set<String> getCoprocessorNames() { 352 return Collections.unmodifiableSet(coprocessorNames); 353 } 354 355 @Override 356 public long getReportTimestamp() { 357 return reportTimestamp; 358 } 359 360 @Override 361 public long getLastReportTimestamp() { 362 return lastReportTimestamp; 363 } 364 365 @Override 366 public List<ServerTask> getTasks() { 367 return tasks; 368 } 369 370 @Override 371 public Map<String, Integer> getRegionCachedInfo() { 372 return Collections.unmodifiableMap(regionCachedInfo); 373 } 374 375 @Override 376 public String toString() { 377 int storeCount = 0; 378 int storeFileCount = 0; 379 int storeRefCount = 0; 380 int maxCompactedStoreFileRefCount = 0; 381 long uncompressedStoreFileSizeMB = 0; 382 long storeFileSizeMB = 0; 383 long memStoreSizeMB = 0; 384 long storefileIndexSizeKB = 0; 385 long rootLevelIndexSizeKB = 0; 386 long readRequestsCount = 0; 387 long writeRequestsCount = 0; 388 long filteredReadRequestsCount = 0; 389 long bloomFilterSizeMB = 0; 390 long compactingCellCount = 0; 391 long compactedCellCount = 0; 392 for (RegionMetrics r : getRegionMetrics().values()) { 393 storeCount += r.getStoreCount(); 394 storeFileCount += r.getStoreFileCount(); 395 storeRefCount += r.getStoreRefCount(); 396 int currentMaxCompactedStoreFileRefCount = r.getMaxCompactedStoreFileRefCount(); 397 maxCompactedStoreFileRefCount = 398 Math.max(maxCompactedStoreFileRefCount, currentMaxCompactedStoreFileRefCount); 399 uncompressedStoreFileSizeMB += 400 (long) r.getUncompressedStoreFileSize().get(Size.Unit.MEGABYTE); 401 storeFileSizeMB += (long) r.getStoreFileSize().get(Size.Unit.MEGABYTE); 402 memStoreSizeMB += (long) r.getMemStoreSize().get(Size.Unit.MEGABYTE); 403 storefileIndexSizeKB += 404 (long) r.getStoreFileUncompressedDataIndexSize().get(Size.Unit.KILOBYTE); 405 readRequestsCount += r.getReadRequestCount(); 406 writeRequestsCount += r.getWriteRequestCount(); 407 filteredReadRequestsCount += r.getFilteredReadRequestCount(); 408 rootLevelIndexSizeKB += (long) r.getStoreFileRootLevelIndexSize().get(Size.Unit.KILOBYTE); 409 bloomFilterSizeMB += (long) r.getBloomFilterSize().get(Size.Unit.MEGABYTE); 410 compactedCellCount += r.getCompactedCellCount(); 411 compactingCellCount += r.getCompactingCellCount(); 412 } 413 StringBuilder sb = Strings.appendKeyValue(new StringBuilder(), "requestsPerSecond", 414 Double.valueOf(getRequestCountPerSecond())); 415 Strings.appendKeyValue(sb, "numberOfOnlineRegions", 416 Integer.valueOf(getRegionMetrics().size())); 417 Strings.appendKeyValue(sb, "usedHeapMB", getUsedHeapSize()); 418 Strings.appendKeyValue(sb, "maxHeapMB", getMaxHeapSize()); 419 Strings.appendKeyValue(sb, "numberOfStores", storeCount); 420 Strings.appendKeyValue(sb, "numberOfStorefiles", storeFileCount); 421 Strings.appendKeyValue(sb, "storeRefCount", storeRefCount); 422 Strings.appendKeyValue(sb, "maxCompactedStoreFileRefCount", maxCompactedStoreFileRefCount); 423 Strings.appendKeyValue(sb, "storefileUncompressedSizeMB", uncompressedStoreFileSizeMB); 424 Strings.appendKeyValue(sb, "storefileSizeMB", storeFileSizeMB); 425 if (uncompressedStoreFileSizeMB != 0) { 426 Strings.appendKeyValue(sb, "compressionRatio", 427 String.format("%.4f", (float) storeFileSizeMB / (float) uncompressedStoreFileSizeMB)); 428 } 429 Strings.appendKeyValue(sb, "memstoreSizeMB", memStoreSizeMB); 430 Strings.appendKeyValue(sb, "readRequestsCount", readRequestsCount); 431 Strings.appendKeyValue(sb, "filteredReadRequestsCount", filteredReadRequestsCount); 432 Strings.appendKeyValue(sb, "writeRequestsCount", writeRequestsCount); 433 Strings.appendKeyValue(sb, "rootIndexSizeKB", rootLevelIndexSizeKB); 434 Strings.appendKeyValue(sb, "totalStaticIndexSizeKB", storefileIndexSizeKB); 435 Strings.appendKeyValue(sb, "totalStaticBloomSizeKB", bloomFilterSizeMB); 436 Strings.appendKeyValue(sb, "totalCompactingKVs", compactingCellCount); 437 Strings.appendKeyValue(sb, "currentCompactedKVs", compactedCellCount); 438 float compactionProgressPct = Float.NaN; 439 if (compactingCellCount > 0) { 440 compactionProgressPct = Float.valueOf((float) compactedCellCount / compactingCellCount); 441 } 442 Strings.appendKeyValue(sb, "compactionProgressPct", compactionProgressPct); 443 Strings.appendKeyValue(sb, "coprocessors", getCoprocessorNames()); 444 return sb.toString(); 445 } 446 } 447}