Skip to content
Merged
Show file tree
Hide file tree
Changes from 3 commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -29,7 +29,6 @@
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.StringJoiner;
import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;
Expand All @@ -51,6 +50,7 @@
import org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.ContainerReplicaProto.State;
import org.apache.hadoop.hdds.scm.ContainerPlacementStatus;
import org.apache.hadoop.hdds.scm.PlacementPolicy;
import org.apache.hadoop.hdds.scm.container.replication.ReplicationManagerMetrics;
import org.apache.hadoop.hdds.scm.events.SCMEvents;
import org.apache.hadoop.hdds.scm.ha.SCMContext;
import org.apache.hadoop.hdds.scm.ha.SCMService;
Expand All @@ -60,10 +60,6 @@
import org.apache.hadoop.hdds.scm.node.states.NodeNotFoundException;
import org.apache.hadoop.hdds.scm.server.StorageContainerManager;
import org.apache.hadoop.hdds.server.events.EventPublisher;
import org.apache.hadoop.metrics2.MetricsCollector;
import org.apache.hadoop.metrics2.MetricsInfo;
import org.apache.hadoop.metrics2.MetricsSource;
import org.apache.hadoop.metrics2.lib.DefaultMetricsSystem;
import org.apache.hadoop.ozone.common.statemachine.InvalidStateTransitionException;
import org.apache.hadoop.ozone.protocol.commands.CloseContainerCommand;
import org.apache.hadoop.ozone.protocol.commands.CommandForDatanode;
Expand All @@ -89,7 +85,7 @@
* that the containers are properly replicated. Replication Manager deals only
* with Quasi Closed / Closed container.
*/
public class ReplicationManager implements MetricsSource, SCMService {
public class ReplicationManager implements SCMService {

public static final Logger LOG =
LoggerFactory.getLogger(ReplicationManager.class);
Expand Down Expand Up @@ -166,6 +162,11 @@ public class ReplicationManager implements MetricsSource, SCMService {
private final long waitTimeInMillis;
private long lastTimeToBeReadyInMillis = 0;

/**
* Replication progress related metrics.
*/
private ReplicationManagerMetrics metrics;

/**
* Constructs ReplicationManager instance with the given configuration.
*
Expand Down Expand Up @@ -197,6 +198,7 @@ public ReplicationManager(final ConfigurationSource conf,
HddsConfigKeys.HDDS_SCM_WAIT_TIME_AFTER_SAFE_MODE_EXIT,
HddsConfigKeys.HDDS_SCM_WAIT_TIME_AFTER_SAFE_MODE_EXIT_DEFAULT,
TimeUnit.MILLISECONDS);
this.metrics = null;

// register ReplicationManager to SCMServiceManager.
serviceManager.register(this);
Expand All @@ -212,10 +214,7 @@ public ReplicationManager(final ConfigurationSource conf,
public synchronized void start() {

if (!isRunning()) {
DefaultMetricsSystem.instance().register(METRICS_SOURCE_NAME,
"SCM Replication manager (closed container replication) related "
+ "metrics",
this);
metrics = ReplicationManagerMetrics.create(this);
LOG.info("Starting Replication Monitor Thread.");
running = true;
replicationMonitor = new Thread(this::run);
Expand Down Expand Up @@ -261,7 +260,7 @@ public synchronized void stop() {
inflightReplication.clear();
inflightDeletion.clear();
running = false;
DefaultMetricsSystem.instance().unregisterSource(METRICS_SOURCE_NAME);
metrics.unRegister();
notifyAll();
} else {
LOG.info("Replication Monitor Thread is not running.");
Expand Down Expand Up @@ -353,12 +352,20 @@ private void processContainer(ContainerInfo container) {
*/
updateInflightAction(container, inflightReplication,
action -> replicas.stream()
.anyMatch(r -> r.getDatanodeDetails().equals(action.datanode)));
.anyMatch(r -> r.getDatanodeDetails().equals(action.datanode)),
()-> metrics.incrNumReplicationCmdsTimeout(),
() -> {
metrics.incrNumReplicationCmdsCompleted();
metrics.incrNumReplicationBytesCompleted(
container.getUsedBytes());
});

updateInflightAction(container, inflightDeletion,
action -> replicas.stream()
.noneMatch(r ->
r.getDatanodeDetails().equals(action.datanode)));
r.getDatanodeDetails().equals(action.datanode)),
() -> metrics.incrNumDeletionCmdsTimeout(),
() -> metrics.incrNumDeletionCmdsCompleted());

/*
* If container is under deleting and all it's replicas are deleted,
Expand Down Expand Up @@ -440,10 +447,14 @@ private void processContainer(ContainerInfo container) {
* @param container Container to update
* @param inflightActions inflightReplication (or) inflightDeletion
* @param filter filter to check if the operation is completed
* @param timeoutCounter update timeout metrics
* @param completedCounter update completed metrics
*/
private void updateInflightAction(final ContainerInfo container,
final Map<ContainerID, List<InflightAction>> inflightActions,
final Predicate<InflightAction> filter) {
final Predicate<InflightAction> filter,
final Runnable timeoutCounter,
final Runnable completedCounter) {
final ContainerID id = container.containerID();
final long deadline = Time.monotonicNow() - rmConf.getEventTimeout();
if (inflightActions.containsKey(id)) {
Expand All @@ -458,6 +469,12 @@ private void updateInflightAction(final ContainerInfo container,
if (health != NodeState.HEALTHY || a.time < deadline
|| filter.test(a)) {
iter.remove();

if (a.time < deadline) {
timeoutCounter.run();
} else if (filter.test(a)) {
completedCounter.run();
}
}
} catch (NodeNotFoundException e) {
// Should not happen, but if it does, just remove the action as the
Expand Down Expand Up @@ -1035,6 +1052,9 @@ private void sendReplicateCommand(final ContainerInfo container,
inflightReplication.computeIfAbsent(id, k -> new ArrayList<>());
sendAndTrackDatanodeCommand(datanode, replicateCommand,
action -> inflightReplication.get(id).add(action));

metrics.incrNumReplicationCmdsSent();
metrics.incrNumReplicationBytesTotal(container.getUsedBytes());
}

/**
Expand All @@ -1058,6 +1078,8 @@ private void sendDeleteCommand(final ContainerInfo container,
inflightDeletion.computeIfAbsent(id, k -> new ArrayList<>());
sendAndTrackDatanodeCommand(datanode, deleteCommand,
action -> inflightDeletion.get(id).add(action));

metrics.incrNumDeletionCmdsSent();
}

/**
Expand Down Expand Up @@ -1143,20 +1165,10 @@ private boolean isOpenContainerHealthy(
.allMatch(r -> ReplicationManager.compareState(state, r.getState()));
}

@Override
public void getMetrics(MetricsCollector collector, boolean all) {
collector.addRecord(ReplicationManager.class.getSimpleName())
.addGauge(ReplicationManagerMetrics.INFLIGHT_REPLICATION,
inflightReplication.size())
.addGauge(ReplicationManagerMetrics.INFLIGHT_DELETION,
inflightDeletion.size())
.endRecord();
}

/**
* Wrapper class to hold the InflightAction with its start time.
*/
private static final class InflightAction {
static final class InflightAction {

private final DatanodeDetails datanode;
private final long time;
Expand All @@ -1166,6 +1178,11 @@ private InflightAction(final DatanodeDetails datanode,
this.datanode = datanode;
this.time = time;
}

@VisibleForTesting
public DatanodeDetails getDatanode() {
return datanode;
}
}

/**
Expand Down Expand Up @@ -1239,34 +1256,6 @@ public int getMaintenanceReplicaMinimum() {
}
}

/**
* Metric name definitions for Replication manager.
*/
public enum ReplicationManagerMetrics implements MetricsInfo {

INFLIGHT_REPLICATION("Tracked inflight container replication requests."),
INFLIGHT_DELETION("Tracked inflight container deletion requests.");

private final String desc;

ReplicationManagerMetrics(String desc) {
this.desc = desc;
}

@Override
public String description() {
return desc;
}

@Override
public String toString() {
return new StringJoiner(", ", this.getClass().getSimpleName() + "{", "}")
.add("name=" + name())
.add("description=" + desc)
.toString();
}
}

@Override
public void notifyStatusChanged() {
serviceLock.lock();
Expand Down Expand Up @@ -1304,4 +1293,16 @@ public boolean shouldRun() {
public String getServiceName() {
return ReplicationManager.class.getSimpleName();
}

public ReplicationManagerMetrics getMetrics() {
return this.metrics;
}

public Map<ContainerID, List<InflightAction>> getInflightReplication() {
return inflightReplication;
}

public Map<ContainerID, List<InflightAction>> getInflightDeletion() {
return inflightDeletion;
}
}
Original file line number Diff line number Diff line change
@@ -0,0 +1,154 @@
/**
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with this
* work for additional information regarding copyright ownership. The ASF
* licenses this file to you under the Apache License, Version 2.0 (the
* "License"); you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
* <p/>
* http://www.apache.org/licenses/LICENSE-2.0
* <p/>
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
* WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
* License for the specific language governing permissions and limitations under
* the License.
*/
package org.apache.hadoop.hdds.scm.container.replication;

import org.apache.hadoop.hdds.scm.container.ReplicationManager;
import org.apache.hadoop.metrics2.annotation.Metric;
import org.apache.hadoop.metrics2.annotation.Metrics;
import org.apache.hadoop.metrics2.lib.DefaultMetricsSystem;
import org.apache.hadoop.metrics2.lib.MutableCounterLong;
import org.apache.hadoop.metrics2.lib.MutableGaugeLong;
import org.apache.hadoop.ozone.OzoneConsts;

/**
* Class contains metrics related to ReplicationManager.
*/
@Metrics(about = "Replication Manager Metrics", context = OzoneConsts.OZONE)
public final class ReplicationManagerMetrics {

public static final String METRICS_SOURCE_NAME =
ReplicationManagerMetrics.class.getSimpleName();

@Metric("Tracked inflight container replication requests.")
private MutableGaugeLong inflightReplication;

@Metric("Tracked inflight container deletion requests.")
private MutableGaugeLong inflightDeletion;

@Metric("Number of replication commands sent.")
private MutableCounterLong numReplicationCmdsSent;

@Metric("Number of replication commands completed.")
private MutableCounterLong numReplicationCmdsCompleted;

@Metric("Number of replication commands timeout.")
private MutableCounterLong numReplicationCmdsTimeout;

@Metric("Number of deletion commands sent.")
private MutableCounterLong numDeletionCmdsSent;

@Metric("Number of deletion commands completed.")
private MutableCounterLong numDeletionCmdsCompleted;

@Metric("Number of deletion commands timeout.")
private MutableCounterLong numDeletionCmdsTimeout;

@Metric("Number of replication bytes total.")
private MutableCounterLong numReplicationBytesTotal;

@Metric("Number of replication bytes completed.")
private MutableCounterLong numReplicationBytesCompleted;

private ReplicationManager replicationManager;

public ReplicationManagerMetrics(ReplicationManager manager) {
this.replicationManager = manager;
}

public static ReplicationManagerMetrics create(ReplicationManager manager) {
return DefaultMetricsSystem.instance().register(METRICS_SOURCE_NAME,
"SCM Replication manager (closed container replication) related "
+ "metrics",
new ReplicationManagerMetrics(manager));

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

thanks for the work! NIT: if RM start->stop->start, ReplicationManagerMetrics#create will be call twice ,and thus new ReplicationManagerMetrics(manager) will be called twice, it seems a little confuse to create the ReplicationManagerMetrics twice. i think it is better off using register instead of create for ReplicationManagerMetrics, and making ReplicationManagerMetrics singleton

@guihecheng guihecheng Jul 26, 2021

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Oh, thanks for the comments, it seems to be reasonable to adopt singleton pattern at first glance.
But actually there are things to be made clear:

  1. When does start->stop->start happen? and the purpose ?
  2. If we adopt singleton, then there's a behavior change:
    • With singleton, we have only one metrics object, and after restart RM(without restarting the daemon of course) will start counting from an old metrics base and are these metrics consistent with the inflightReplication/inflightDeletion maps in RM?
    • Without singleton, RM restart with a freshly created object all init to 0.
      So what is the desired behavior?

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

In general I prefer to avoid singletons, as they can make testing difficult, especially around Mini-Cluster tests.

I think its reasonable for a RM restart to reset the counts to zero, as it is a new instance. A RM restart should only happen in tests, as to restart it in a real cluster, you must restart all of SCM.

}

public void unRegister() {
DefaultMetricsSystem.instance().unregisterSource(METRICS_SOURCE_NAME);
}

public void incrNumReplicationCmdsSent() {
this.numReplicationCmdsSent.incr();
}

public void incrNumReplicationCmdsCompleted() {
this.numReplicationCmdsCompleted.incr();
}

public void incrNumReplicationCmdsTimeout() {
this.numReplicationCmdsTimeout.incr();
}

public void incrNumDeletionCmdsSent() {
this.numDeletionCmdsSent.incr();
}

public void incrNumDeletionCmdsCompleted() {
this.numDeletionCmdsCompleted.incr();
}

public void incrNumDeletionCmdsTimeout() {
this.numDeletionCmdsTimeout.incr();
}

public void incrNumReplicationBytesTotal(long bytes) {
this.numReplicationBytesTotal.incr(bytes);
}

public void incrNumReplicationBytesCompleted(long bytes) {
this.numReplicationBytesCompleted.incr(bytes);
}

public long getInflightReplication() {
return replicationManager.getInflightReplication().size();
}

public long getInflightDeletion() {
return replicationManager.getInflightDeletion().size();
}

public long getNumReplicationCmdsSent() {
return this.numReplicationCmdsSent.value();
}

public long getNumReplicationCmdsCompleted() {
return this.numReplicationCmdsCompleted.value();
}

public long getNumReplicationCmdsTimeout() {
return this.numReplicationCmdsTimeout.value();
}

public long getNumDeletionCmdsSent() {
return this.numDeletionCmdsSent.value();
}

public long getNumDeletionCmdsCompleted() {
return this.numDeletionCmdsCompleted.value();
}

public long getNumDeletionCmdsTimeout() {
return this.numDeletionCmdsTimeout.value();
}

public long getNumReplicationBytesTotal() {
return this.numReplicationBytesTotal.value();
}

public long getNumReplicationBytesCompleted() {
return this.numReplicationBytesCompleted.value();
}
}
Loading