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.hbtop.mode;
019
020import java.util.ArrayList;
021import java.util.Arrays;
022import java.util.Collections;
023import java.util.HashMap;
024import java.util.HashSet;
025import java.util.List;
026import java.util.Map;
027import java.util.Set;
028import java.util.stream.Collectors;
029import org.apache.hadoop.hbase.ClusterMetrics;
030import org.apache.hadoop.hbase.ServerMetrics;
031import org.apache.hadoop.hbase.UserMetrics;
032import org.apache.hadoop.hbase.hbtop.Record;
033import org.apache.hadoop.hbase.hbtop.RecordFilter;
034import org.apache.hadoop.hbase.hbtop.field.Field;
035import org.apache.hadoop.hbase.hbtop.field.FieldInfo;
036import org.apache.hadoop.hbase.hbtop.field.FieldValue;
037import org.apache.hadoop.hbase.hbtop.field.FieldValueType;
038import org.apache.yetus.audience.InterfaceAudience;
039
040/**
041 * Implementation for {@link ModeStrategy} for client Mode.
042 */
043@InterfaceAudience.Private
044public final class ClientModeStrategy implements ModeStrategy {
045
046  private final List<FieldInfo> fieldInfos =
047    Arrays.asList(new FieldInfo(Field.CLIENT, 0, true), new FieldInfo(Field.USER_COUNT, 5, true),
048      new FieldInfo(Field.REQUEST_COUNT_PER_SECOND, 10, true),
049      new FieldInfo(Field.READ_REQUEST_COUNT_PER_SECOND, 10, true),
050      new FieldInfo(Field.WRITE_REQUEST_COUNT_PER_SECOND, 10, true),
051      new FieldInfo(Field.FILTERED_READ_REQUEST_COUNT_PER_SECOND, 10, true),
052      new FieldInfo(Field.HOST_ADDRESS, 0, true), new FieldInfo(Field.USER_NAME, 0, true),
053      new FieldInfo(Field.CLIENT_VERSION, 0, true), new FieldInfo(Field.SERVICE_NAME, 0, true));
054  private final Map<String, RequestCountPerSecond> requestCountPerSecondMap = new HashMap<>();
055
056  ClientModeStrategy() {
057  }
058
059  @Override
060  public List<FieldInfo> getFieldInfos() {
061    return fieldInfos;
062  }
063
064  @Override
065  public Field getDefaultSortField() {
066    return Field.REQUEST_COUNT_PER_SECOND;
067  }
068
069  @Override
070  public List<Record> getRecords(ClusterMetrics clusterMetrics,
071    List<RecordFilter> pushDownFilters) {
072    List<Record> records = createRecords(clusterMetrics);
073    return aggregateRecordsAndAddDistinct(
074      ModeStrategyUtils.applyFilterAndGet(records, pushDownFilters), Field.CLIENT, Field.USER,
075      Field.USER_COUNT);
076  }
077
078  List<Record> createRecords(ClusterMetrics clusterMetrics) {
079    List<Record> ret = new ArrayList<>();
080    for (ServerMetrics serverMetrics : clusterMetrics.getLiveServerMetrics().values()) {
081      long lastReportTimestamp = serverMetrics.getLastReportTimestamp();
082      serverMetrics.getUserMetrics().values()
083        .forEach(um -> um.getClientMetrics().values()
084          .forEach(clientMetrics -> ret.add(createRecord(um.getNameAsString(), clientMetrics,
085            lastReportTimestamp, serverMetrics.getServerName().getServerName()))));
086    }
087    return ret;
088  }
089
090  /**
091   * Aggregate the records and count the unique values for the given distinctField
092   * @param records               records to be processed
093   * @param groupBy               Field on which group by needs to be done
094   * @param distinctField         Field whose unique values needs to be counted
095   * @param uniqueCountAssignedTo a target field to which the unique count is assigned to
096   * @return aggregated records
097   */
098  List<Record> aggregateRecordsAndAddDistinct(List<Record> records, Field groupBy,
099    Field distinctField, Field uniqueCountAssignedTo) {
100    List<Record> result = new ArrayList<>();
101    records.stream().collect(Collectors.groupingBy(r -> r.get(groupBy))).values().forEach(val -> {
102      Set<FieldValue> distinctValues = new HashSet<>();
103      Map<Field, FieldValue> map = new HashMap<>();
104      for (Record record : val) {
105        for (Map.Entry<Field, FieldValue> field : record.entrySet()) {
106          if (distinctField.equals(field.getKey())) {
107            // We will not be adding the field in the new record whose distinct count is required
108            distinctValues.add(record.get(distinctField));
109          } else {
110            if (field.getKey().getFieldValueType() == FieldValueType.STRING) {
111              map.put(field.getKey(), field.getValue());
112            } else {
113              if (map.get(field.getKey()) == null) {
114                map.put(field.getKey(), field.getValue());
115              } else {
116                map.put(field.getKey(), map.get(field.getKey()).plus(field.getValue()));
117              }
118            }
119          }
120        }
121      }
122      // Add unique count field
123      map.put(uniqueCountAssignedTo, uniqueCountAssignedTo.newValue(distinctValues.size()));
124      result.add(
125        Record.ofEntries(map.entrySet().stream().map(k -> Record.entry(k.getKey(), k.getValue()))));
126    });
127    return result;
128  }
129
130  Record createRecord(String user, UserMetrics.ClientMetrics clientMetrics,
131    long lastReportTimestamp, String server) {
132    Record.Builder builder = Record.builder();
133    String client = clientMetrics.getHostName();
134    builder.put(Field.CLIENT, clientMetrics.getHostName());
135    String mapKey = client + "$" + user + "$" + server;
136    RequestCountPerSecond requestCountPerSecond = requestCountPerSecondMap.get(mapKey);
137    if (requestCountPerSecond == null) {
138      requestCountPerSecond = new RequestCountPerSecond();
139      requestCountPerSecondMap.put(mapKey, requestCountPerSecond);
140    }
141    requestCountPerSecond.refresh(lastReportTimestamp, clientMetrics.getReadRequestsCount(),
142      clientMetrics.getFilteredReadRequestsCount(), clientMetrics.getWriteRequestsCount());
143    builder.put(Field.REQUEST_COUNT_PER_SECOND, requestCountPerSecond.getRequestCountPerSecond());
144    builder.put(Field.READ_REQUEST_COUNT_PER_SECOND,
145      requestCountPerSecond.getReadRequestCountPerSecond());
146    builder.put(Field.WRITE_REQUEST_COUNT_PER_SECOND,
147      requestCountPerSecond.getWriteRequestCountPerSecond());
148    builder.put(Field.FILTERED_READ_REQUEST_COUNT_PER_SECOND,
149      requestCountPerSecond.getFilteredReadRequestCountPerSecond());
150    builder.put(Field.HOST_ADDRESS, clientMetrics.getHostAddress());
151    builder.put(Field.USER_NAME, clientMetrics.getUserName());
152    builder.put(Field.CLIENT_VERSION, clientMetrics.getClientVersion());
153    builder.put(Field.SERVICE_NAME, clientMetrics.getServiceName());
154    builder.put(Field.USER, user);
155    return builder.build();
156  }
157
158  @Override
159  public DrillDownInfo drillDown(Record selectedRecord) {
160    List<RecordFilter> initialFilters = Collections.singletonList(
161      RecordFilter.newBuilder(Field.CLIENT).doubleEquals(selectedRecord.get(Field.CLIENT)));
162    return new DrillDownInfo(Mode.USER, initialFilters);
163  }
164}