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,
013 * software distributed under the License is distributed on an
014 * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
015 * KIND, either express or implied.  See the License for the
016 * specific language governing permissions and limitations
017 * under the License.
018 */
019package org.apache.hadoop.hbase.replication.regionserver;
020
021import java.util.Date;
022import java.util.HashMap;
023import java.util.List;
024import java.util.ArrayList;
025import java.util.Map;
026
027import org.apache.yetus.audience.InterfaceAudience;
028import org.apache.hadoop.hbase.shaded.protobuf.generated.ClusterStatusProtos;
029import org.apache.hadoop.hbase.util.EnvironmentEdgeManager;
030import org.apache.hadoop.hbase.util.Strings;
031
032/**
033 * This class is used for exporting some of the info from replication metrics
034 */
035@InterfaceAudience.Private
036public class ReplicationLoad {
037
038  // Empty load instance.
039  public static final ReplicationLoad EMPTY_REPLICATIONLOAD = new ReplicationLoad();
040
041  private List<MetricsSource> sourceMetricsList;
042  private MetricsSink sinkMetrics;
043
044  private List<ClusterStatusProtos.ReplicationLoadSource> replicationLoadSourceList;
045  private ClusterStatusProtos.ReplicationLoadSink replicationLoadSink;
046
047  /** default constructor */
048  public ReplicationLoad() {
049    super();
050  }
051
052  /**
053   * buildReplicationLoad
054   * @param srMetricsList
055   * @param skMetrics
056   */
057
058  public void buildReplicationLoad(final List<MetricsSource> srMetricsList,
059      final MetricsSink skMetrics) {
060    this.sourceMetricsList = srMetricsList;
061    this.sinkMetrics = skMetrics;
062
063    // build the SinkLoad
064    ClusterStatusProtos.ReplicationLoadSink.Builder rLoadSinkBuild =
065        ClusterStatusProtos.ReplicationLoadSink.newBuilder();
066    rLoadSinkBuild.setAgeOfLastAppliedOp(sinkMetrics.getAgeOfLastAppliedOp());
067    rLoadSinkBuild.setTimeStampsOfLastAppliedOp(sinkMetrics.getTimestampOfLastAppliedOp());
068    this.replicationLoadSink = rLoadSinkBuild.build();
069
070    // build the SourceLoad List
071    Map<String, ClusterStatusProtos.ReplicationLoadSource> replicationLoadSourceMap =
072        new HashMap<>();
073    for (MetricsSource sm : this.sourceMetricsList) {
074      // Get the actual peer id
075      String peerId = sm.getPeerID();
076      String[] parts = peerId.split("-", 2);
077      peerId = parts.length != 1 ? parts[0] : peerId;
078
079      long ageOfLastShippedOp = sm.getAgeOfLastShippedOp();
080      int sizeOfLogQueue = sm.getSizeOfLogQueue();
081      long timeStampOfLastShippedOp = sm.getTimestampOfLastShippedOp();
082      long replicationLag;
083      long timePassedAfterLastShippedOp =
084          EnvironmentEdgeManager.currentTime() - timeStampOfLastShippedOp;
085      if (sizeOfLogQueue != 0) {
086        // err on the large side
087        replicationLag = Math.max(ageOfLastShippedOp, timePassedAfterLastShippedOp);
088      } else if (timePassedAfterLastShippedOp < 2 * ageOfLastShippedOp) {
089        replicationLag = ageOfLastShippedOp; // last shipped happen recently
090      } else {
091        // last shipped may happen last night,
092        // so NO real lag although ageOfLastShippedOp is non-zero
093        replicationLag = 0;
094      }
095
096      ClusterStatusProtos.ReplicationLoadSource rLoadSource = replicationLoadSourceMap.get(peerId);
097      if (rLoadSource != null) {
098        ageOfLastShippedOp = Math.max(rLoadSource.getAgeOfLastShippedOp(), ageOfLastShippedOp);
099        sizeOfLogQueue += rLoadSource.getSizeOfLogQueue();
100        timeStampOfLastShippedOp = Math.min(rLoadSource.getTimeStampOfLastShippedOp(),
101          timeStampOfLastShippedOp);
102        replicationLag = Math.max(rLoadSource.getReplicationLag(), replicationLag);
103      }
104      ClusterStatusProtos.ReplicationLoadSource.Builder rLoadSourceBuild =
105          ClusterStatusProtos.ReplicationLoadSource.newBuilder();
106      rLoadSourceBuild.setPeerID(peerId);
107      rLoadSourceBuild.setAgeOfLastShippedOp(ageOfLastShippedOp);
108      rLoadSourceBuild.setSizeOfLogQueue(sizeOfLogQueue);
109      rLoadSourceBuild.setTimeStampOfLastShippedOp(timeStampOfLastShippedOp);
110      rLoadSourceBuild.setReplicationLag(replicationLag);
111
112      replicationLoadSourceMap.put(peerId, rLoadSourceBuild.build());
113    }
114    this.replicationLoadSourceList = new ArrayList<>(replicationLoadSourceMap.values());
115  }
116
117  /**
118   * sourceToString
119   * @return a string contains sourceReplicationLoad information
120   */
121  public String sourceToString() {
122    if (this.sourceMetricsList == null) return null;
123
124    StringBuilder sb = new StringBuilder();
125
126    for (ClusterStatusProtos.ReplicationLoadSource rls : this.replicationLoadSourceList) {
127
128      sb = Strings.appendKeyValue(sb, "\n           PeerID", rls.getPeerID());
129      sb = Strings.appendKeyValue(sb, "AgeOfLastShippedOp", rls.getAgeOfLastShippedOp());
130      sb = Strings.appendKeyValue(sb, "SizeOfLogQueue", rls.getSizeOfLogQueue());
131      sb =
132          Strings.appendKeyValue(sb, "TimestampsOfLastShippedOp",
133            (new Date(rls.getTimeStampOfLastShippedOp()).toString()));
134      sb = Strings.appendKeyValue(sb, "Replication Lag", rls.getReplicationLag());
135    }
136
137    return sb.toString();
138  }
139
140  /**
141   * sinkToString
142   * @return a string contains sinkReplicationLoad information
143   */
144  public String sinkToString() {
145    if (this.replicationLoadSink == null) return null;
146
147    StringBuilder sb = new StringBuilder();
148    sb =
149        Strings.appendKeyValue(sb, "AgeOfLastAppliedOp",
150          this.replicationLoadSink.getAgeOfLastAppliedOp());
151    sb =
152        Strings.appendKeyValue(sb, "TimestampsOfLastAppliedOp",
153          (new Date(this.replicationLoadSink.getTimeStampsOfLastAppliedOp()).toString()));
154
155    return sb.toString();
156  }
157
158  public ClusterStatusProtos.ReplicationLoadSink getReplicationLoadSink() {
159    return this.replicationLoadSink;
160  }
161
162  public List<ClusterStatusProtos.ReplicationLoadSource> getReplicationLoadSourceList() {
163    return this.replicationLoadSourceList;
164  }
165
166  /**
167   * @see java.lang.Object#toString()
168   */
169  @Override
170  public String toString() {
171    return this.sourceToString() + System.getProperty("line.separator") + this.sinkToString();
172  }
173
174}