Skip to content

Commit

Permalink
HDDS-12080. move ratis metadata direcotry cleaner to OzoneContainer i…
Browse files Browse the repository at this point in the history 
…nitilizer
  • Loading branch information
Slava Tutrinov committed Jan 20, 2025
1 parent 1507d3f commit cd4cb8a
Show file tree
Hide file tree
Showing 31 changed files with 242 additions and 56 deletions.
2 changes: 1 addition & 1 deletion ...main/java/org/apache/hadoop/ozone/container/common/statemachine/DatanodeStateMachine.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -181,7 +181,7 @@ public DatanodeStateMachine(HddsDatanodeService hddsDatanodeService,
constructionLock.writeLock().lock();
try {
container = new OzoneContainer(hddsDatanodeService, this.datanodeDetails,
conf, context, certClient, secretKeyClient);
conf, context, certClient, secretKeyClient, connectionManager);
} finally {
constructionLock.writeLock().unlock();
}
Expand Down
19 changes: 0 additions & 19 deletions ...in/java/org/apache/hadoop/ozone/container/common/states/endpoint/VersionEndpointTask.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -16,16 +16,11 @@
*/
package org.apache.hadoop.ozone.container.common.states.endpoint;

import java.io.File;
import java.io.IOException;
import java.net.BindException;
import java.util.Arrays;
import java.util.Objects;
import java.util.concurrent.Callable;

import org.apache.commons.io.FileUtils;
import org.apache.hadoop.hdds.conf.ConfigurationSource;
import org.apache.hadoop.hdds.protocol.proto.HddsProtos;
import org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.SCMVersionResponseProto;
import org.apache.hadoop.ozone.OzoneConsts;
import org.apache.hadoop.ozone.container.common.DatanodeLayoutStorage;
Expand Down Expand Up @@ -99,20 +94,6 @@ public EndpointStateMachine.EndPointStates call() throws Exception {
layoutStorage.setClusterId(clusterId);
layoutStorage.persistCurrentState();

HddsProtos.NodeState nodePreviousState = rpcEndPoint.getEndPoint()
.getNodePreviousState(ozoneContainer.getDatanodeDetails().getUuid());

if (nodePreviousState != null && nodePreviousState.equals(HddsProtos.NodeState.DEAD)) {
ozoneContainer.getMetaVolumeSet().getVolumeMap().forEach((key, value) ->
Arrays.asList(Objects.requireNonNull(value.getStorageDir().listFiles())).stream().filter(File::isDirectory).forEach(f -> {
try {
FileUtils.deleteDirectory(f);
} catch (IOException e) {
LOG.warn("Failed to delete directory {}", f.getAbsolutePath(), e);
}
}));
}

// Start the container services after getting the version information
ozoneContainer.start(clusterId);
}
Expand Down
67 changes: 63 additions & 4 deletions ...ner-service/src/main/java/org/apache/hadoop/ozone/container/ozoneimpl/OzoneContainer.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -21,10 +21,13 @@
import com.google.common.annotations.VisibleForTesting;
import com.google.common.collect.Maps;
import com.google.common.util.concurrent.ThreadFactoryBuilder;
import org.apache.commons.io.FileUtils;
import org.apache.hadoop.hdds.HddsUtils;
import org.apache.hadoop.hdds.conf.ConfigurationSource;
import org.apache.hadoop.hdds.protocol.DatanodeDetails;
import org.apache.hadoop.hdds.protocol.DatanodeDetails.Port.Name;
import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.ContainerType;
import org.apache.hadoop.hdds.protocol.proto.HddsProtos;
import org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos;
import org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.ContainerReplicaProto;
import org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.IncrementalContainerReportProto;
Expand All @@ -49,6 +52,7 @@
import org.apache.hadoop.ozone.container.common.interfaces.Handler;
import org.apache.hadoop.ozone.container.common.report.IncrementalReportSender;
import org.apache.hadoop.ozone.container.common.statemachine.DatanodeConfiguration;
import org.apache.hadoop.ozone.container.common.statemachine.SCMConnectionManager;
import org.apache.hadoop.ozone.container.common.statemachine.StateContext;
import org.apache.hadoop.ozone.container.common.transport.server.XceiverServerGrpc;
import org.apache.hadoop.ozone.container.common.transport.server.XceiverServerSpi;
Expand All @@ -73,14 +77,19 @@
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.File;
import java.io.IOException;
import java.net.InetSocketAddress;
import java.time.Duration;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.HashMap;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
Expand Down Expand Up @@ -135,6 +144,7 @@ public class OzoneContainer {
private ScheduledExecutorService dbCompactionExecutorService;

private final ContainerMetrics metrics;
private final SCMConnectionManager scmConnectionManager;
private WitnessedContainerMetadataStore witnessedContainerMetadataStore;

enum InitializingStatus {
Expand All @@ -151,12 +161,14 @@ enum InitializingStatus {
* @throws IOException
*/
public OzoneContainer(HddsDatanodeService hddsDatanodeService,
DatanodeDetails datanodeDetails, ConfigurationSource conf,
StateContext context, CertificateClient certClient,
SecretKeyVerifierClient secretKeyClient) throws IOException {
DatanodeDetails datanodeDetails, ConfigurationSource conf,
StateContext context, CertificateClient certClient,
SecretKeyVerifierClient secretKeyClient,
SCMConnectionManager scmConnectionManager) throws IOException {
config = conf;
this.datanodeDetails = datanodeDetails;
this.context = context;
this.scmConnectionManager = scmConnectionManager;
this.volumeChecker = new StorageVolumeChecker(conf, new Timer(),
datanodeDetails.threadNamePrefix());

Expand Down Expand Up @@ -229,6 +241,8 @@ public OzoneContainer(HddsDatanodeService hddsDatanodeService,
*/
controller = new ContainerController(containerSet, handlers);

cleanUpRatisMetadataDirectory();

writeChannel = XceiverServerRatis.newXceiverServerRatis(hddsDatanodeService,
datanodeDetails, config, hddsDispatcher, controller, certClient,
context);
Expand Down Expand Up @@ -300,7 +314,7 @@ public OzoneContainer(HddsDatanodeService hddsDatanodeService,
public OzoneContainer(
DatanodeDetails datanodeDetails, ConfigurationSource conf,
StateContext context) throws IOException {
this(null, datanodeDetails, conf, context, null, null);
this(null, datanodeDetails, conf, context, null, null, null);
}

public GrpcTlsConfig getTlsClientConfig() {
Expand Down Expand Up @@ -659,4 +673,49 @@ public void compactDb() {
public DatanodeDetails getDatanodeDetails() {
return datanodeDetails;
}

private void cleanUpRatisMetadataDirectory()
throws IOException {
if (scmConnectionManager != null) {
Collection<InetSocketAddress> scmAddressesForDatanodes = HddsUtils.getSCMAddressForDatanodes(config);
for (InetSocketAddress scmAddress : scmAddressesForDatanodes) {
scmConnectionManager.addSCMServer(scmAddress, context.getThreadNamePrefix());
context.addEndpoint(scmAddress);
}

scmConnectionManager.getValues().stream()
.filter(endPoint -> !endPoint.isPassive())
.findFirst()
.ifPresent(rpcEndPoint -> {
try {
// Check the previous state of the datanode stored on the SCM side
HddsProtos.NodeState nodePreviousState = rpcEndPoint.getEndPoint()
.getNodeState(datanodeDetails.getUuid());

if (HddsProtos.NodeState.DEAD.equals(nodePreviousState)) {
LOG.info("The node previous state is DEAD, let's clean up the RATIS/THREE pipelines");
// OK, the node was previously marked as DEAD, let's clean up the
// RATIS/THREE pipelines (aka raft-groups)
this.getMetaVolumeSet().getVolumeMap().forEach((key, value) ->
Arrays.stream(Objects.requireNonNull(value.getStorageDir()
// don't touch the directory with volume check info
.listFiles((dir, name) -> !name.equals("tmp"))))
.filter(File::isDirectory) // only directories
.forEach(directory -> {
try {
FileUtils.deleteDirectory(directory);
LOG.info("Delete directory: {}", directory.getAbsolutePath());
} catch (IOException e) {
LOG.warn("Failed to delete directory: {}", directory);
}
}));
}

} catch (IOException e) {
LOG.error(String.format("Failed to get datanode previous state with SCM: %s", e.getMessage()), e);
}
});
}
}

}
9 changes: 3 additions & 6 deletions ...vice/src/main/java/org/apache/hadoop/ozone/protocol/StorageContainerDatanodeProtocol.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -18,8 +18,8 @@

import org.apache.hadoop.hdds.annotation.InterfaceAudience;
import org.apache.hadoop.hdds.protocol.proto.HddsProtos.ExtendedDatanodeDetailsProto;
import org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos;
import org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.NodePreviousStateResponseProto;
import org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.NodeStateRequestProto;
import org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.NodeStateResponseProto;
import org.apache.hadoop.hdds.protocol.proto
.StorageContainerDatanodeProtocolProtos.PipelineReportsProto;
import org.apache.hadoop.hdds.protocol.proto
Expand Down Expand Up @@ -91,9 +91,6 @@ SCMRegisteredResponseProto register(
PipelineReportsProto pipelineReports,
LayoutVersionProto layoutInfo) throws IOException;

NodePreviousStateResponseProto
getNodePreviousState(
StorageContainerDatanodeProtocolProtos.NodePreviousStateRequestProto
request) throws IOException;
NodeStateResponseProto getNodeState(NodeStateRequestProto request) throws IOException;

}
17 changes: 8 additions & 9 deletions ...pache/hadoop/ozone/protocolPB/StorageContainerDatanodeProtocolClientSideTranslatorPB.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -21,10 +21,9 @@
import org.apache.hadoop.hdds.protocol.proto.HddsProtos;
import org.apache.hadoop.hdds.protocol.proto.HddsProtos
.ExtendedDatanodeDetailsProto;
import org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos;
import org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.LayoutVersionProto;
import org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.NodePreviousStateRequestProto;
import org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.NodePreviousStateResponseProto;
import org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.NodeStateRequestProto;
import org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.NodeStateResponseProto;
import org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.PipelineReportsProto;
import org.apache.hadoop.hdds.protocol.proto
.StorageContainerDatanodeProtocolProtos.ContainerReportsProto;
Expand Down Expand Up @@ -184,17 +183,17 @@ public SCMRegisteredResponseProto register(
.getRegisterResponse();
}

public HddsProtos.NodeState getNodePreviousState(UUID datanodeUuid) throws IOException {
NodePreviousStateRequestProto request = NodePreviousStateRequestProto.newBuilder()
public HddsProtos.NodeState getNodeState(UUID datanodeUuid) throws IOException {
NodeStateRequestProto request = NodeStateRequestProto.newBuilder()
.setDatanodeUUID(datanodeUuid.toString())
.build();
return getNodePreviousState(request).getPreviousState();
return getNodeState(request).getNodeState();

}

@Override
public NodePreviousStateResponseProto getNodePreviousState(NodePreviousStateRequestProto request) throws IOException {
return submitRequest(Type.NodePreviousState, builder -> builder.setNodePreviousStateRequest(request))
.getNodePreviousStateResponse();
public NodeStateResponseProto getNodeState(NodeStateRequestProto request) throws IOException {
return submitRequest(Type.NodePreviousState, builder -> builder.setNodeStateRequest(request))
.getNodeStateResponse();
}
}
4 changes: 2 additions & 2 deletions ...pache/hadoop/ozone/protocolPB/StorageContainerDatanodeProtocolServerSideTranslatorPB.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -122,8 +122,8 @@ public SCMDatanodeResponse processMessage(SCMDatanodeRequest request)
return SCMDatanodeResponse.newBuilder()
.setCmdType(cmdType)
.setStatus(Status.OK)
.setNodePreviousStateResponse(
impl.getNodePreviousState(request.getNodePreviousStateRequest()))
.setNodeStateResponse(
impl.getNodeState(request.getNodeStateRequest()))
.build();
default:
throw new ServiceException("Unknown command type: " + cmdType);
Expand Down
6 changes: 4 additions & 2 deletions ...container-service/src/test/java/org/apache/hadoop/ozone/container/common/ScmTestMock.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -26,6 +26,8 @@
.StorageContainerDatanodeProtocolProtos.ContainerReplicaProto;
import org.apache.hadoop.hdds.protocol.proto
.StorageContainerDatanodeProtocolProtos.ContainerReportsProto;
import org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.NodeStateRequestProto;
import org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.NodeStateResponseProto;
import org.apache.hadoop.hdds.protocol.proto
.StorageContainerDatanodeProtocolProtos.PipelineReportsProto;
import org.apache.hadoop.hdds.protocol.proto
Expand Down Expand Up @@ -256,8 +258,8 @@ private void sleepIfNeeded() {
}

@Override
public StorageContainerDatanodeProtocolProtos.NodePreviousStateResponseProto getNodePreviousState(
StorageContainerDatanodeProtocolProtos.NodePreviousStateRequestProto request) throws IOException {
public NodeStateResponseProto getNodeState(
NodeStateRequestProto request) throws IOException {
return null;
}

Expand Down
Loading

0 comments on commit cd4cb8a

Please sign in to comment.