Skip to content

Commit

Permalink
[Refactor] ClusterInfo to use j.util.Map instead of ImmutableOpenMap (#…
Browse files Browse the repository at this point in the history 
…7126)

With java.util.Map immutability and collection improvements the
hppc ImmutableOpenMap is not needed in ClusterInfo. This commit
refactors ClusterInfo to use java Maps and Immutable Collections and
further trim the dependency on the aging hppc library.

Signed-off-by: Nicholas Walter Knize <[email protected]>
  • Loading branch information
@nknize
nknize committed Jun 23, 2023
1 parent 79bf769 commit 5a8e522
Show file tree
Hide file tree
Showing 13 changed files with 309 additions and 389 deletions.
27 changes: 13 additions & 14 deletions server/src/internalClusterTest/java/org/opensearch/cluster/ClusterInfoServiceIT.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -32,7 +32,6 @@

package org.opensearch.cluster;

import com.carrotsearch.hppc.cursors.ObjectCursor;
import org.opensearch.OpenSearchException;
import org.opensearch.action.ActionListener;
import org.opensearch.action.ActionRequest;
Expand All @@ -47,7 +46,6 @@
import org.opensearch.cluster.routing.allocation.decider.EnableAllocationDecider;
import org.opensearch.cluster.service.ClusterService;
import org.opensearch.common.Strings;
import org.opensearch.common.collect.ImmutableOpenMap;
import org.opensearch.common.settings.Settings;
import org.opensearch.common.unit.TimeValue;
import org.opensearch.index.IndexService;
Expand All @@ -69,6 +67,7 @@
import java.util.Collection;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.Locale;
import java.util.Set;
import java.util.concurrent.atomic.AtomicBoolean;
Expand Down Expand Up @@ -174,24 +173,24 @@ public void testClusterInfoServiceCollectsInformation() {
infoService.setUpdateFrequency(TimeValue.timeValueMillis(200));
ClusterInfo info = infoService.refresh();
assertNotNull("info should not be null", info);
ImmutableOpenMap<String, DiskUsage> leastUsages = info.getNodeLeastAvailableDiskUsages();
ImmutableOpenMap<String, DiskUsage> mostUsages = info.getNodeMostAvailableDiskUsages();
ImmutableOpenMap<String, Long> shardSizes = info.shardSizes;
final Map<String, DiskUsage> leastUsages = info.getNodeLeastAvailableDiskUsages();
final Map<String, DiskUsage> mostUsages = info.getNodeMostAvailableDiskUsages();
final Map<String, Long> shardSizes = info.shardSizes;
assertNotNull(leastUsages);
assertNotNull(shardSizes);
assertThat("some usages are populated", leastUsages.values().size(), Matchers.equalTo(2));
assertThat("some shard sizes are populated", shardSizes.values().size(), greaterThan(0));
for (ObjectCursor<DiskUsage> usage : leastUsages.values()) {
logger.info("--> usage: {}", usage.value);
assertThat("usage has be retrieved", usage.value.getFreeBytes(), greaterThan(0L));
for (Map.Entry<String, DiskUsage> usage : leastUsages.entrySet()) {
logger.info("--> usage: {}", usage.getValue());
assertThat("usage has be retrieved", usage.getValue().getFreeBytes(), greaterThan(0L));
}
for (ObjectCursor<DiskUsage> usage : mostUsages.values()) {
logger.info("--> usage: {}", usage.value);
assertThat("usage has be retrieved", usage.value.getFreeBytes(), greaterThan(0L));
for (DiskUsage usage : mostUsages.values()) {
logger.info("--> usage: {}", usage);
assertThat("usage has be retrieved", usage.getFreeBytes(), greaterThan(0L));
}
for (ObjectCursor<Long> size : shardSizes.values()) {
logger.info("--> shard size: {}", size.value);
assertThat("shard size is greater than 0", size.value, greaterThanOrEqualTo(0L));
for (Long size : shardSizes.values()) {
logger.info("--> shard size: {}", size);
assertThat("shard size is greater than 0", size, greaterThanOrEqualTo(0L));
}
ClusterService clusterService = internalTestCluster.getInstance(ClusterService.class, internalTestCluster.getClusterManagerName());
ClusterState state = clusterService.state();
Expand Down
2 changes: 1 addition & 1 deletion ...erTest/java/org/opensearch/cluster/routing/allocation/decider/DiskThresholdDeciderIT.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -574,7 +574,7 @@ private void refreshDiskUsage() {
// if the nodes were all under the low watermark already (but unbalanced) then a change in the disk usage doesn't trigger a reroute
// even though it's now possible to achieve better balance, so we have to do an explicit reroute. TODO fix this?
if (StreamSupport.stream(clusterInfoService.getClusterInfo().getNodeMostAvailableDiskUsages().values().spliterator(), false)
.allMatch(cur -> cur.value.getFreeBytes() > WATERMARK_BYTES)) {
.allMatch(cur -> cur.getFreeBytes() > WATERMARK_BYTES)) {
assertAcked(client().admin().cluster().prepareReroute());
}

Expand Down
82 changes: 36 additions & 46 deletions server/src/main/java/org/opensearch/cluster/ClusterInfo.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -34,9 +34,7 @@

import com.carrotsearch.hppc.ObjectHashSet;
import com.carrotsearch.hppc.cursors.ObjectCursor;
import com.carrotsearch.hppc.cursors.ObjectObjectCursor;
import org.opensearch.cluster.routing.ShardRouting;
import org.opensearch.common.collect.ImmutableOpenMap;
import org.opensearch.common.io.stream.StreamInput;
import org.opensearch.common.io.stream.StreamOutput;
import org.opensearch.common.io.stream.Writeable;
Expand All @@ -47,6 +45,7 @@
import org.opensearch.index.store.StoreStats;

import java.io.IOException;
import java.util.Collections;
import java.util.Map;
import java.util.Objects;

Expand All @@ -59,15 +58,15 @@
* @opensearch.internal
*/
public class ClusterInfo implements ToXContentFragment, Writeable {
private final ImmutableOpenMap<String, DiskUsage> leastAvailableSpaceUsage;
private final ImmutableOpenMap<String, DiskUsage> mostAvailableSpaceUsage;
final ImmutableOpenMap<String, Long> shardSizes;
private final Map<String, DiskUsage> leastAvailableSpaceUsage;
private final Map<String, DiskUsage> mostAvailableSpaceUsage;
final Map<String, Long> shardSizes; // pkg-private for testing only
public static final ClusterInfo EMPTY = new ClusterInfo();
final ImmutableOpenMap<ShardRouting, String> routingToDataPath;
final ImmutableOpenMap<NodeAndPath, ReservedSpace> reservedSpace;
final Map<ShardRouting, String> routingToDataPath;
final Map<NodeAndPath, ReservedSpace> reservedSpace;

protected ClusterInfo() {
this(ImmutableOpenMap.of(), ImmutableOpenMap.of(), ImmutableOpenMap.of(), ImmutableOpenMap.of(), ImmutableOpenMap.of());
this(Map.of(), Map.of(), Map.of(), Map.of(), Map.of());
}

/**
Expand All @@ -81,11 +80,11 @@ protected ClusterInfo() {
* @see #shardIdentifierFromRouting
*/
public ClusterInfo(
ImmutableOpenMap<String, DiskUsage> leastAvailableSpaceUsage,
ImmutableOpenMap<String, DiskUsage> mostAvailableSpaceUsage,
ImmutableOpenMap<String, Long> shardSizes,
ImmutableOpenMap<ShardRouting, String> routingToDataPath,
ImmutableOpenMap<NodeAndPath, ReservedSpace> reservedSpace
final Map<String, DiskUsage> leastAvailableSpaceUsage,
final Map<String, DiskUsage> mostAvailableSpaceUsage,
final Map<String, Long> shardSizes,
final Map<ShardRouting, String> routingToDataPath,
final Map<NodeAndPath, ReservedSpace> reservedSpace
) {
this.leastAvailableSpaceUsage = leastAvailableSpaceUsage;
this.shardSizes = shardSizes;
Expand All @@ -106,48 +105,39 @@ public ClusterInfo(StreamInput in) throws IOException {
reservedSpaceMap = Map.of();
}

ImmutableOpenMap.Builder<String, DiskUsage> leastBuilder = ImmutableOpenMap.builder();
this.leastAvailableSpaceUsage = leastBuilder.putAll(leastMap).build();
ImmutableOpenMap.Builder<String, DiskUsage> mostBuilder = ImmutableOpenMap.builder();
this.mostAvailableSpaceUsage = mostBuilder.putAll(mostMap).build();
ImmutableOpenMap.Builder<String, Long> sizeBuilder = ImmutableOpenMap.builder();
this.shardSizes = sizeBuilder.putAll(sizeMap).build();
ImmutableOpenMap.Builder<ShardRouting, String> routingBuilder = ImmutableOpenMap.builder();
this.routingToDataPath = routingBuilder.putAll(routingMap).build();
ImmutableOpenMap.Builder<NodeAndPath, ReservedSpace> reservedSpaceBuilder = ImmutableOpenMap.builder();
this.reservedSpace = reservedSpaceBuilder.putAll(reservedSpaceMap).build();
this.leastAvailableSpaceUsage = Collections.unmodifiableMap(leastMap);
this.mostAvailableSpaceUsage = Collections.unmodifiableMap(mostMap);
this.shardSizes = Collections.unmodifiableMap(sizeMap);
this.routingToDataPath = Collections.unmodifiableMap(routingMap);
this.reservedSpace = Collections.unmodifiableMap(reservedSpaceMap);
}

@Override
public void writeTo(StreamOutput out) throws IOException {
out.writeVInt(this.leastAvailableSpaceUsage.size());
for (ObjectObjectCursor<String, DiskUsage> c : this.leastAvailableSpaceUsage) {
out.writeString(c.key);
c.value.writeTo(out);
}
out.writeMap(this.leastAvailableSpaceUsage, StreamOutput::writeString, (o, v) -> v.writeTo(o));
out.writeMap(this.mostAvailableSpaceUsage, StreamOutput::writeString, (o, v) -> v.writeTo(o));
out.writeMap(this.shardSizes, StreamOutput::writeString, (o, v) -> out.writeLong(v == null ? -1 : v));
out.writeMap(this.routingToDataPath, (o, k) -> k.writeTo(o), StreamOutput::writeString);
if (out.getVersion().onOrAfter(StoreStats.RESERVED_BYTES_VERSION)) {
out.writeMap(this.reservedSpace);
out.writeMap(this.reservedSpace, (o, v) -> v.writeTo(o), (o, v) -> v.writeTo(o));
}
}

public XContentBuilder toXContent(XContentBuilder builder, Params params) throws IOException {
builder.startObject("nodes");
{
for (ObjectObjectCursor<String, DiskUsage> c : this.leastAvailableSpaceUsage) {
builder.startObject(c.key);
for (Map.Entry<String, DiskUsage> c : this.leastAvailableSpaceUsage.entrySet()) {
builder.startObject(c.getKey());
{ // node
builder.field("node_name", c.value.getNodeName());
builder.field("node_name", c.getValue().getNodeName());
builder.startObject("least_available");
{
c.value.toShortXContent(builder);
c.getValue().toShortXContent(builder);
}
builder.endObject(); // end "least_available"
builder.startObject("most_available");
{
DiskUsage most = this.mostAvailableSpaceUsage.get(c.key);
DiskUsage most = this.mostAvailableSpaceUsage.get(c.getKey());
if (most != null) {
most.toShortXContent(builder);
}
Expand All @@ -160,26 +150,26 @@ public XContentBuilder toXContent(XContentBuilder builder, Params params) throws
builder.endObject(); // end "nodes"
builder.startObject("shard_sizes");
{
for (ObjectObjectCursor<String, Long> c : this.shardSizes) {
builder.humanReadableField(c.key + "_bytes", c.key, new ByteSizeValue(c.value));
for (Map.Entry<String, Long> c : this.shardSizes.entrySet()) {
builder.humanReadableField(c.getKey() + "_bytes", c.getKey(), new ByteSizeValue(c.getValue()));
}
}
builder.endObject(); // end "shard_sizes"
builder.startObject("shard_paths");
{
for (ObjectObjectCursor<ShardRouting, String> c : this.routingToDataPath) {
builder.field(c.key.toString(), c.value);
for (Map.Entry<ShardRouting, String> c : this.routingToDataPath.entrySet()) {
builder.field(c.getKey().toString(), c.getValue());
}
}
builder.endObject(); // end "shard_paths"
builder.startArray("reserved_sizes");
{
for (ObjectObjectCursor<NodeAndPath, ReservedSpace> c : this.reservedSpace) {
for (Map.Entry<NodeAndPath, ReservedSpace> c : this.reservedSpace.entrySet()) {
builder.startObject();
{
builder.field("node_id", c.key.nodeId);
builder.field("path", c.key.path);
c.value.toXContent(builder, params);
builder.field("node_id", c.getKey().nodeId);
builder.field("path", c.getKey().path);
c.getValue().toXContent(builder, params);
}
builder.endObject(); // NodeAndPath
}
Expand All @@ -192,16 +182,16 @@ public XContentBuilder toXContent(XContentBuilder builder, Params params) throws
* Returns a node id to disk usage mapping for the path that has the least available space on the node.
* Note that this does not take account of reserved space: there may be another path with less available _and unreserved_ space.
*/
public ImmutableOpenMap<String, DiskUsage> getNodeLeastAvailableDiskUsages() {
return this.leastAvailableSpaceUsage;
public Map<String, DiskUsage> getNodeLeastAvailableDiskUsages() {
return Collections.unmodifiableMap(this.leastAvailableSpaceUsage);
}

/**
* Returns a node id to disk usage mapping for the path that has the most available space on the node.
* Note that this does not take account of reserved space: there may be another path with more available _and unreserved_ space.
*/
public ImmutableOpenMap<String, DiskUsage> getNodeMostAvailableDiskUsages() {
return this.mostAvailableSpaceUsage;
public Map<String, DiskUsage> getNodeMostAvailableDiskUsages() {
return Collections.unmodifiableMap(this.mostAvailableSpaceUsage);
}

/**
Expand Down
Loading

0 comments on commit 5a8e522

Please sign in to comment.