Skip to content

HDDS-12840. Avoid long containerId in KeyValueContainer #8291

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
szetszwo merged 1 commit into from
Apr 17, 2025
Merged

HDDS-12840. Avoid long containerId in KeyValueContainer #8291

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
167 changes: 62 additions & 105 deletions ...r-service/src/main/java/org/apache/hadoop/ozone/container/keyvalue/KeyValueContainer.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,6 +17,11 @@

package org.apache.hadoop.ozone.container.keyvalue;

import static org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.ContainerDataProto.State.CLOSED;
import static org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.ContainerDataProto.State.CLOSING;
import static org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.ContainerDataProto.State.DELETED;
import static org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.ContainerDataProto.State.QUASI_CLOSED;
import static org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.ContainerDataProto.State.UNHEALTHY;
import static org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.Result.CONTAINER_ALREADY_EXISTS;
import static org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.Result.CONTAINER_FILES_CREATE_ERROR;
import static org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.Result.CONTAINER_INTERNAL_ERROR;
Expand Down Expand Up @@ -44,6 +49,7 @@
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Set;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.ReentrantReadWriteLock;
Expand All @@ -53,7 +59,7 @@
import org.apache.hadoop.hdds.HddsUtils;
import org.apache.hadoop.hdds.conf.ConfigurationSource;
import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos;
import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.ContainerDataProto;
import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.ContainerDataProto.State;
import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.ContainerType;
import org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.ContainerReplicaProto;
import org.apache.hadoop.hdds.scm.container.common.helpers.StorageContainerException;
Expand Down Expand Up @@ -97,7 +103,7 @@ public class KeyValueContainer implements Container<KeyValueContainerData> {
private final Object dumpLock = new Object();

private final KeyValueContainerData containerData;
private ConfigurationSource config;
private final ConfigurationSource config;

// Cache of Blocks (LocalIDs) awaiting final PutBlock call after the stream
// is closed. When a block is added to the DB as part of putBlock, it is
Expand All @@ -114,10 +120,8 @@ public class KeyValueContainer implements Container<KeyValueContainerData> {

public KeyValueContainer(KeyValueContainerData containerData,
ConfigurationSource ozoneConfig) {
Preconditions.checkNotNull(containerData,
"KeyValueContainerData cannot be null");
Preconditions.checkNotNull(ozoneConfig,
"Ozone configuration cannot be null");
Objects.requireNonNull(containerData, "containerData == null");
Objects.requireNonNull(ozoneConfig, "ozoneConfig == null");
this.config = ozoneConfig;
this.containerData = containerData;
if (this.containerData.isOpen() || this.containerData.isClosing()) {
Expand All @@ -140,10 +144,9 @@ public void setCheckChunksFilePath(boolean bCheckChunksDirFilePath) {
@Override
public void create(VolumeSet volumeSet, VolumeChoosingPolicy
volumeChoosingPolicy, String clusterId) throws StorageContainerException {
Preconditions.checkNotNull(volumeChoosingPolicy, "VolumeChoosingPolicy " +
"cannot be null");
Preconditions.checkNotNull(volumeSet, "VolumeSet cannot be null");
Preconditions.checkNotNull(clusterId, "clusterId cannot be null");
Objects.requireNonNull(volumeChoosingPolicy, "VolumeChoosingPolicy == null");
Objects.requireNonNull(volumeSet, "volumeSet == null");
Objects.requireNonNull(clusterId, "clusterId == null");

File containerMetaDataPath = null;
//acquiring volumeset read lock
Expand Down Expand Up @@ -226,13 +229,12 @@ public void create(VolumeSet volumeSet, VolumeChoosingPolicy
FileUtil.fullyDelete(containerMetaDataPath.getParentFile());
}
volumes.remove(containerVolume);
LOG.error("Exception attempting to create container {} on volume {}" +
" remaining volumes to try {}", containerData.getContainerID(),
containerVolume.getHddsRootDir(), volumes.size(), ex);
LOG.error("Failed to create {} on volume {}, remaining volumes: {}]",
containerData, containerVolume.getHddsRootDir(), volumes.size(), ex);
if (volumes.isEmpty()) {
throw new StorageContainerException(
"Container creation failed. " + ex.getMessage(), ex,
CONTAINER_INTERNAL_ERROR);
"Failed to create " + containerData + " on all volumes: " + volumeSet.getVolumesList(),
ex, CONTAINER_INTERNAL_ERROR);
}
}
}
Expand Down Expand Up @@ -291,7 +293,6 @@ public void populatePathFields(String clusterId,
private void writeToContainerFile(File containerFile, boolean isCreate)
throws StorageContainerException {
File tempContainerFile = null;
long containerId = containerData.getContainerID();
try {
tempContainerFile = createTempFile(containerFile);
ContainerDataYaml.createContainerFile(containerData, tempContainerFile);
Expand All @@ -308,14 +309,11 @@ private void writeToContainerFile(File containerFile, boolean isCreate)

} catch (IOException ex) {
onFailure(containerData.getVolume());
String containerExceptionMessage = "Error while creating/updating" +
" container file. ContainerID: " + containerId +
", container path: " + containerFile.getAbsolutePath();
if (tempContainerFile == null) {
containerExceptionMessage += " Temporary file could not be created.";
}
throw new StorageContainerException(containerExceptionMessage, ex,
CONTAINER_FILES_CREATE_ERROR);
final String op = tempContainerFile == null ? "create tmp file for "
: (isCreate ? "create" : "update") + " container file ";
throw new StorageContainerException(
"Failed to " + op + containerFile.getAbsolutePath() + ": " + containerData,
ex, CONTAINER_FILES_CREATE_ERROR);
} finally {
if (tempContainerFile != null && tempContainerFile.exists()) {
if (!tempContainerFile.delete()) {
Expand All @@ -339,7 +337,6 @@ private void updateContainerFile(File containerFile)

@Override
public void delete() throws StorageContainerException {
long containerId = containerData.getContainerID();
try {
// Delete the Container from tmp directory.
File tmpDirectoryPath = KeyValueContainerUtil.getTmpDirectoryPath(
Expand All @@ -353,8 +350,7 @@ public void delete() throws StorageContainerException {
// On datanode shutdown/restart any partial artifacts left
// will be wiped from volume's tmp directory.
onFailure(containerData.getVolume());
String errMsg = String.format("Failed to cleanup container. ID: %d",
containerId);
final String errMsg = "Failed to cleanup " + containerData;
LOG.error(errMsg, ex);
throw new StorageContainerException(errMsg, ex, CONTAINER_INTERNAL_ERROR);
}
Expand All @@ -377,8 +373,7 @@ public void markContainerForClose() throws StorageContainerException {
"Attempting to close a " + getContainerState() + " container.",
CONTAINER_NOT_OPEN);
}
updateContainerData(() ->
containerData.setState(ContainerDataProto.State.CLOSING));
updateContainerState(CLOSING);
// Do not clear the pendingBlockCache here as a follower can still
// receive transactions from leader in CLOSING state. Refer to
// KeyValueHandler#checkContainerOpen()
Expand All @@ -390,37 +385,30 @@ public void markContainerForClose() throws StorageContainerException {
@Override
public void markContainerUnhealthy() throws StorageContainerException {
writeLock();
ContainerDataProto.State prevState = containerData.getState();
final State prevState = containerData.getState();
try {
updateContainerData(() ->
containerData.setState(ContainerDataProto.State.UNHEALTHY));
updateContainerState(UNHEALTHY);
clearPendingPutBlockCache();
} finally {
writeUnlock();
}
LOG.warn("Moving container {} to state {} from state:{}",
containerData.getContainerPath(), containerData.getState(),
prevState);
LOG.warn("Marked container UNHEALTHY from {}: {}", prevState, containerData);
}

@Override
public void markContainerForDelete() {
writeLock();
ContainerDataProto.State prevState = containerData.getState();
final State prevState = containerData.getState();
try {
containerData.setState(ContainerDataProto.State.DELETED);
File containerFile = getContainerFile();
containerData.setState(DELETED);
// update the new container data to .container File
updateContainerFile(containerFile);
updateContainerFile(getContainerFile());
} catch (IOException ioe) {
LOG.error("Exception occur while update container {} state",
containerData.getContainerID(), ioe);
LOG.error("Failed to updateContainerFile: {}", containerData, ioe);
} finally {
writeUnlock();
}
LOG.info("Moving container {} to state {} from state:{}",
containerData.getContainerPath(), containerData.getState(),
prevState);
LOG.warn("Marked container DELETED from {}: {}", prevState, containerData);
}

@Override
Expand All @@ -437,9 +425,7 @@ public void close() throws StorageContainerException {
throw new StorageContainerException(ex, IO_EXCEPTION);
}
closeAndFlushIfNeeded(containerData::closeContainer);
LOG.info("Container {} is closed with bcsId {}.",
containerData.getContainerID(),
containerData.getBlockCommitSequenceId());
LOG.info("Closed container: {}", containerData);
}

@Override
Expand Down Expand Up @@ -475,6 +461,10 @@ private void closeAndFlushIfNeeded(Runnable closer)
}
}

private void updateContainerState(State newState) throws StorageContainerException {
updateContainerData(() -> containerData.setState(newState));
}

/**
*
* Must be invoked with the writeLock held.
Expand All @@ -485,17 +475,14 @@ private void closeAndFlushIfNeeded(Runnable closer)
private void updateContainerData(Runnable update)
throws StorageContainerException {
Preconditions.checkState(hasWriteLock());
ContainerDataProto.State oldState = null;
final State oldState = containerData.getState();
try {
oldState = containerData.getState();
update.run();
File containerFile = getContainerFile();
// update the new container data to .container File
updateContainerFile(containerFile);
updateContainerFile(getContainerFile());

} catch (StorageContainerException ex) {
if (oldState != null
&& containerData.getState() != ContainerDataProto.State.UNHEALTHY) {
if (containerData.getState() != UNHEALTHY) {
// Failed to update .container file. Reset the state to old state only
// if the current state is not unhealthy.
containerData.setState(oldState);
Expand All @@ -522,9 +509,7 @@ private void flushAndSyncDB() throws StorageContainerException {
try {
try (DBHandle db = BlockUtils.getDB(containerData, config)) {
db.getStore().flushLog(true);
LOG.info("Container {} is synced with bcsId {}.",
containerData.getContainerID(),
containerData.getBlockCommitSequenceId());
LOG.info("Synced container: {}", containerData);
}
} catch (StorageContainerException ex) {
throw ex;
Expand Down Expand Up @@ -562,16 +547,12 @@ public void update(Map<String, String> metadata, boolean forceUpdate, String con
// TODO: Now, when writing the updated data to .container file, we are
// holding lock and writing data to disk. We can have async implementation
// to flush the update container data to disk.
long containerId = containerData.getContainerID();
if (!containerData.isValid()) {
LOG.debug("Invalid container data. ContainerID: {}", containerId);
throw new StorageContainerException("Invalid container data. " +
"ContainerID: " + containerId, INVALID_CONTAINER_STATE);
throw new StorageContainerException("Invalid container data: " + containerData, INVALID_CONTAINER_STATE);
}
if (!forceUpdate && !containerData.isOpen()) {
throw new StorageContainerException(
"Updating a closed container without force option is not allowed. " +
"ContainerID: " + containerId, UNSUPPORTED_REQUEST);
"Updating a closed container without force option is disallowed: " + containerData, UNSUPPORTED_REQUEST);
}

Map<String, String> oldMetadata = containerData.getMetadata();
Expand Down Expand Up @@ -616,9 +597,8 @@ public void importContainerData(InputStream input,
byte[] descriptorContent = packer.unpackContainerData(this, input, tmpDir,
destContainerDir);

Preconditions.checkNotNull(descriptorContent,
"Container descriptor is missing from the container archive: "
+ getContainerData().getContainerID());
Objects.requireNonNull(descriptorContent,
() -> "Missing container descriptor from the archive: " + getContainerData());

//now, we have extracted the container descriptor from the previous
//datanode. We can load it and upload it with the current data
Expand Down Expand Up @@ -694,15 +674,10 @@ public void exportContainerData(OutputStream destination,
try {
// Closed/ Quasi closed and unhealthy containers are considered for
// replication by replication manager if they are under-replicated.
ContainerProtos.ContainerDataProto.State state =
getContainerData().getState();
if (!(state == ContainerProtos.ContainerDataProto.State.CLOSED ||
state == ContainerDataProto.State.QUASI_CLOSED
|| state == ContainerDataProto.State.UNHEALTHY)) {
throw new IllegalStateException(
"Only (quasi)closed and unhealthy containers can be exported. " +
"ContainerId=" + getContainerData().getContainerID() +
" is in state " + state);
final State state = getContainerData().getState();
// Only CLOSED, QUASI_CLOSED and UNHEALTHY containers can be exported.
if (state != CLOSED && state != QUASI_CLOSED && state != UNHEALTHY) {
throw new IllegalStateException("Failed to export: Unexpected state in " + getContainerData());
}

try {
Expand Down Expand Up @@ -848,12 +823,9 @@ public void addToPendingPutBlockCache(long localID)
// pendingPutBlockCache is an Empty Set. This should not happen if the
// container is in OPEN or CLOSING state. Log the exception here and
// throw a non-Runtime exception so that putBlock request fails.
String msg = "Failed to add block " + localID + " to " +
"pendingPutBlockCache of container " + containerData.getContainerID()
+ " (state: " + containerData.getState() + ")";
String msg = "Failed to add block " + localID + " to pendingPutBlockCache for " + containerData;
LOG.error(msg, e);
throw new StorageContainerException(msg,
ContainerProtos.Result.CONTAINER_INTERNAL_ERROR);
throw new StorageContainerException(msg, CONTAINER_INTERNAL_ERROR);
}
}

Expand Down Expand Up @@ -925,8 +897,7 @@ private ContainerReplicaProto.State getHddsState()
state = ContainerReplicaProto.State.DELETED;
break;
default:
throw new StorageContainerException("Invalid Container state found: " +
containerData.getContainerID(), INVALID_CONTAINER_STATE);
throw new StorageContainerException("Invalid Container state: " + containerData, INVALID_CONTAINER_STATE);
}
return state;
}
Expand All @@ -941,33 +912,24 @@ public File getContainerDBFile() {

@Override
public boolean shouldScanMetadata() {
boolean shouldScan =
getContainerState() != ContainerDataProto.State.UNHEALTHY;
if (!shouldScan && LOG.isDebugEnabled()) {
LOG.debug("Container {} in state {} should not have its metadata " +
"scanned.",
containerData.getContainerID(), containerData.getState());
final boolean shouldScan = getContainerState() != UNHEALTHY;
if (!shouldScan) {
LOG.debug("Healthy container metadata is not scanned: {}", containerData);
}
return shouldScan;
}

@Override
public ScanResult scanMetaData() throws InterruptedException {
long containerId = containerData.getContainerID();
KeyValueContainerCheck checker =
new KeyValueContainerCheck(containerData.getMetadataPath(), config,
containerId, containerData.getVolume(), this);
return checker.fastCheck();
return new KeyValueContainerCheck(containerData, this, config)
.fastCheck();
}

@Override
public boolean shouldScanData() {
boolean shouldScan =
getContainerState() == ContainerDataProto.State.CLOSED
|| getContainerState() == ContainerDataProto.State.QUASI_CLOSED;
if (!shouldScan && LOG.isDebugEnabled()) {
LOG.debug("Container {} in state {} should not have its data scanned.",
containerData.getContainerID(), containerData.getState());
final boolean shouldScan = getContainerState() == CLOSED || getContainerState() == QUASI_CLOSED;
if (!shouldScan) {
LOG.debug("Healthy container is not scanned: {}", containerData);
}

return shouldScan;
Expand All @@ -981,13 +943,8 @@ public ScanResult scanData(DataTransferThrottler throttler, Canceler canceler)
" done for container in state "
+ containerData.getState());
}

long containerId = containerData.getContainerID();
KeyValueContainerCheck checker =
new KeyValueContainerCheck(containerData.getMetadataPath(), config,
containerId, containerData.getVolume(), this);

return checker.fullCheck(throttler, canceler);
return new KeyValueContainerCheck(containerData, this, config)
.fullCheck(throttler, canceler);
}

/**
Expand Down
4 changes: 4 additions & 0 deletions ...vice/src/main/java/org/apache/hadoop/ozone/container/keyvalue/KeyValueContainerCheck.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -82,6 +82,10 @@ public KeyValueContainerCheck(String metadataPath, ConfigurationSource conf,
this.container = container;
}

KeyValueContainerCheck(KeyValueContainerData data, KeyValueContainer container, ConfigurationSource conf) {
this(data.getMetadataPath(), conf, data.getContainerID(), data.getVolume(), container);
}

/**
* Run basic integrity checks on container metadata.
* These checks do not look inside the metadata files.
Expand Down
Loading