Skip to content
Open
Show file tree
Hide file tree
Changes from all 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 @@ -799,6 +799,31 @@ CompletableFuture<RegisterResult> registerProducerOffsets(
*/
CompletableFuture<ClusterHealth> getClusterHealth();

/**
* Describe the replica load of each tablet server in the cluster asynchronously.
*
* <p>The returned list contains {@link TabletServerDescription} counters per tablet server.
* Live tablet servers are always included, even when they host no replicas.
*
* <p>This API is designed for operational tooling (e.g., a Kubernetes operator):
*
* <ul>
* <li>scale-in safety gate - a tablet server may only be removed once its {@code numReplicas}
* is {@code 0}, since terminating a non-empty server causes under-replication or data
* unavailability;
* <li>rolling-upgrade gate - only proceed to the next server when the previous one is green
* again ({@code inSyncReplicas == numReplicas && activeLeaderReplicas ==
* numLeaderReplicas}) and {@link #getClusterHealth()} is not {@link
* ClusterHealthStatus#RED}, since a leaderless bucket may not surface in the per-server
* counters (see {@link TabletServerDescription});
* <li>reporting per-server tablet load in cluster status.
* </ul>
*
* @return a {@link CompletableFuture} that completes with the per tablet server replica loads.
* @since 1.0
*/
CompletableFuture<List<TabletServerDescription>> describeTabletServers();

/**
* List per-bucket remote log manifest entries for a table or partition scope.
*
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -67,6 +67,7 @@
import org.apache.fluss.rpc.messages.DatabaseExistsResponse;
import org.apache.fluss.rpc.messages.DeleteProducerOffsetsRequest;
import org.apache.fluss.rpc.messages.DescribeClusterConfigsRequest;
import org.apache.fluss.rpc.messages.DescribeTabletServersRequest;
import org.apache.fluss.rpc.messages.DropAclsRequest;
import org.apache.fluss.rpc.messages.DropDatabaseRequest;
import org.apache.fluss.rpc.messages.DropTableRequest;
Expand Down Expand Up @@ -942,6 +943,12 @@ public CompletableFuture<ClusterHealth> getClusterHealth() {
.thenApply(ClientRpcMessageUtils::toClusterHealth);
}

@Override
public CompletableFuture<List<TabletServerDescription>> describeTabletServers() {
return gateway.describeTabletServers(new DescribeTabletServersRequest())
.thenApply(ClientRpcMessageUtils::toTabletServerDescriptions);
}

@VisibleForTesting
public AdminGateway getAdminGateway() {
return gateway;
Expand Down
Original file line number Diff line number Diff line change
@@ -0,0 +1,158 @@
/*
* 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
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* 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.fluss.client.admin;

import org.apache.fluss.annotation.PublicEvolving;
import org.apache.fluss.cluster.rebalance.ServerTag;

import javax.annotation.Nullable;

import java.util.Objects;
import java.util.Optional;

/**
* Per tablet server replica load returned by {@link Admin#describeTabletServers()}.
*
* <p>It reports the replica counters of a single tablet server: how many replicas it hosts, how
* many of those are in sync, how many buckets it leads, and how many of those leaders are active.
* The {@link ClusterHealth} aggregates the same counters across all tablet servers.
*
* <p>Operational tooling can derive from it:
*
* <ul>
* <li>a scale-in safety gate: the server is empty and safe to remove iff {@code numReplicas ==
* 0};
* <li>a per-server health (green) predicate for rolling upgrades: {@code inSyncReplicas ==
* numReplicas && activeLeaderReplicas == numLeaderReplicas}.
* </ul>
*
* <p>Note: a bucket without an elected leader is attributed to no server's {@code
* numLeaderReplicas}, so the per-server values sum to the cluster-wide {@link
* ClusterHealth#getNumLeaderReplicas()} only when every bucket has an elected leader. Such a bucket
* does not always surface through {@code inSyncReplicas < numReplicas}: when the last ISR member
* goes offline, it is kept in the ISR so that a leader can be re-elected later, and the assigned
* servers still look green by the predicate above. A rolling-upgrade gate should therefore
* additionally require {@link Admin#getClusterHealth()} to not report {@link
* ClusterHealthStatus#RED}, which counts leaderless buckets unconditionally.
*
* @since 1.0
*/
@PublicEvolving
public final class TabletServerDescription {

private final int serverId;
private final int numReplicas;
private final int inSyncReplicas;
private final int numLeaderReplicas;
private final int activeLeaderReplicas;
private final @Nullable ServerTag serverTag;

public TabletServerDescription(
int serverId,
int numReplicas,
int inSyncReplicas,
int numLeaderReplicas,
int activeLeaderReplicas,
@Nullable ServerTag serverTag) {
this.serverId = serverId;
this.numReplicas = numReplicas;
this.inSyncReplicas = inSyncReplicas;
this.numLeaderReplicas = numLeaderReplicas;
this.activeLeaderReplicas = activeLeaderReplicas;
this.serverTag = serverTag;
}

public int getServerId() {
return serverId;
}

/** Number of replicas assigned to this tablet server. */
public int getNumReplicas() {
return numReplicas;
}

/** Number of assigned replicas that are in the ISR of their bucket. */
public int getInSyncReplicas() {
return inSyncReplicas;
}

/** Number of hosted replicas that are the elected leader of their bucket. */
public int getNumLeaderReplicas() {
return numLeaderReplicas;
}

/** Number of leader replicas on this server that are currently active. */
public int getActiveLeaderReplicas() {
return activeLeaderReplicas;
}

/**
* The {@link ServerTag} set on this server via {@link Admin#addServerTag}, or empty if the
* server is untagged. A tagged server is being drained, so it is reported even when it is dead
* and hosts no replicas.
*/
public Optional<ServerTag> getServerTag() {
return Optional.ofNullable(serverTag);
}

@Override
public boolean equals(Object o) {
if (this == o) {
return true;
}
if (!(o instanceof TabletServerDescription)) {
return false;
}
TabletServerDescription that = (TabletServerDescription) o;
return serverId == that.serverId
&& numReplicas == that.numReplicas
&& inSyncReplicas == that.inSyncReplicas
&& numLeaderReplicas == that.numLeaderReplicas
&& activeLeaderReplicas == that.activeLeaderReplicas
&& serverTag == that.serverTag;
}

@Override
public int hashCode() {
return Objects.hash(
serverId,
numReplicas,
inSyncReplicas,
numLeaderReplicas,
activeLeaderReplicas,
serverTag);
}

@Override
public String toString() {
return "TabletServerDescription{"
+ "serverId="
+ serverId
+ ", numReplicas="
+ numReplicas
+ ", inSyncReplicas="
+ inSyncReplicas
+ ", numLeaderReplicas="
+ numLeaderReplicas
+ ", activeLeaderReplicas="
+ activeLeaderReplicas
+ ", serverTag="
+ serverTag
+ '}';
}
}
Original file line number Diff line number Diff line change
Expand Up @@ -21,6 +21,7 @@
import org.apache.fluss.client.admin.ClusterHealthStatus;
import org.apache.fluss.client.admin.OffsetSpec;
import org.apache.fluss.client.admin.ProducerOffsetsResult;
import org.apache.fluss.client.admin.TabletServerDescription;
import org.apache.fluss.client.lookup.LookupBatch;
import org.apache.fluss.client.lookup.PrefixLookupBatch;
import org.apache.fluss.client.metadata.AcquireKvSnapshotLeaseResult;
Expand All @@ -35,6 +36,7 @@
import org.apache.fluss.cluster.rebalance.RebalanceProgress;
import org.apache.fluss.cluster.rebalance.RebalanceResultForBucket;
import org.apache.fluss.cluster.rebalance.RebalanceStatus;
import org.apache.fluss.cluster.rebalance.ServerTag;
import org.apache.fluss.config.cluster.AlterConfigOpType;
import org.apache.fluss.config.cluster.ColumnPositionType;
import org.apache.fluss.config.cluster.ConfigEntry;
Expand All @@ -55,6 +57,7 @@
import org.apache.fluss.rpc.messages.AlterDatabaseRequest;
import org.apache.fluss.rpc.messages.AlterTableRequest;
import org.apache.fluss.rpc.messages.CreatePartitionRequest;
import org.apache.fluss.rpc.messages.DescribeTabletServersResponse;
import org.apache.fluss.rpc.messages.DropPartitionRequest;
import org.apache.fluss.rpc.messages.GetClusterHealthResponse;
import org.apache.fluss.rpc.messages.GetFileSystemSecurityTokenResponse;
Expand Down Expand Up @@ -898,4 +901,21 @@ private static ClusterHealthStatus toClusterHealthStatus(int pbStatus) {
return ClusterHealthStatus.UNKNOWN;
}
}

public static List<TabletServerDescription> toTabletServerDescriptions(
DescribeTabletServersResponse resp) {
return resp.getTabletServersList().stream()
.map(
load ->
new TabletServerDescription(
load.getServerId(),
load.getNumReplicas(),
load.getInSyncReplicas(),
load.getNumLeaderReplicas(),
load.getActiveLeaderReplicas(),
load.hasServerTag()
? ServerTag.valueOf(load.getServerTag())
: null))
.collect(Collectors.toList());
}
}
Original file line number Diff line number Diff line change
Expand Up @@ -2305,6 +2305,14 @@ public void testAddAndRemoveServerTags() throws Exception {
.containsEntry(0, ServerTag.PERMANENT_OFFLINE)
.containsEntry(1, ServerTag.PERMANENT_OFFLINE);

// the tags are also visible via describeTabletServers.
List<TabletServerDescription> described = admin.describeTabletServers().get();
assertThat(getTabletServerDescription(described, 0).getServerTag())
.contains(ServerTag.PERMANENT_OFFLINE);
assertThat(getTabletServerDescription(described, 1).getServerTag())
.contains(ServerTag.PERMANENT_OFFLINE);
assertThat(getTabletServerDescription(described, 2).getServerTag()).isNotPresent();

// 3.add different server tag for server 0,2. error will be thrown and tag for 2 will not be
// added.
assertThatThrownBy(
Expand Down Expand Up @@ -2860,4 +2868,94 @@ void testClusterHealthDuringRollingUpgrade() throws Exception {
assertThat(afterRecovery.getNumLeaderReplicas())
.isEqualTo(afterRecovery.getActiveLeaderReplicas());
}

@Test
void testDescribeTabletServersDuringRollingUpgrade() throws Exception {
TablePath tablePath = TablePath.of("test_db", "describe_tablet_servers_table");
TableDescriptor tableDescriptor =
TableDescriptor.builder().schema(DEFAULT_SCHEMA).distributedBy(3, "id").build();
long tableId = createTable(tablePath, tableDescriptor, true);
waitAllReplicasReady(tableId, 3);

// Phase 1: Cluster is healthy - every live server is reported, hosts replicas of the
// created table (replication factor 3 on 3 servers) and is green. The cluster is shared
// with other tests, so wait until residue from them (e.g. a recovering ISR) has settled
// before taking the snapshot asserted below.
waitUntil(
() ->
admin.describeTabletServers().get().stream()
.allMatch(FlussAdminITCase::isServerGreen),
Duration.ofMinutes(1),
"All tablet servers should be green before the rolling upgrade starts");

List<TabletServerDescription> servers = admin.describeTabletServers().get();
assertThat(servers).extracting(TabletServerDescription::getServerId).contains(0, 1, 2);
for (TabletServerDescription server : servers) {
assertThat(server.getNumReplicas()).isGreaterThan(0);
assertThat(isServerGreen(server)).isTrue();
}

// The per-server counters must sum up to the cluster-wide health counters.

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

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

A leaderless bucket is counted by cluster health but by no server here, so these sums can differ while every server still looks green. The waitUntil above doesn't rule that out. testPerServerSumsReconcileWithClusterHealth already covers this.
Do we need it?

ClusterHealth health = admin.getClusterHealth().get();
assertThat(servers.stream().mapToInt(TabletServerDescription::getNumReplicas).sum())
.isEqualTo(health.getNumReplicas());
assertThat(servers.stream().mapToInt(TabletServerDescription::getInSyncReplicas).sum())
.isEqualTo(health.getInSyncReplicas());
assertThat(servers.stream().mapToInt(TabletServerDescription::getNumLeaderReplicas).sum())
.isEqualTo(health.getNumLeaderReplicas());
assertThat(
servers.stream()
.mapToInt(TabletServerDescription::getActiveLeaderReplicas)
.sum())
.isEqualTo(health.getActiveLeaderReplicas());

// Phase 2: Stop one tablet server (simulate server crash during rolling upgrade). It must
// still be reported with its assigned replicas, but no longer green - an operator must
// not treat it as safe to remove.
int stoppedServerId = 0;
FLUSS_CLUSTER_EXTENSION.stopTabletServer(stoppedServerId);
FLUSS_CLUSTER_EXTENSION.assertHasTabletServerNumber(2);

for (int bucket = 0; bucket < 3; bucket++) {
TableBucket tb = new TableBucket(tableId, bucket);
FLUSS_CLUSTER_EXTENSION.waitUntilReplicaShrinkFromIsr(tb, stoppedServerId);
}

TabletServerDescription stopped =
getTabletServerDescription(admin.describeTabletServers().get(), stoppedServerId);
assertThat(stopped.getNumReplicas()).isGreaterThan(0);
assertThat(stopped.getInSyncReplicas()).isLessThan(stopped.getNumReplicas());

// Phase 3: Restart the server and wait until every server is green again.
FLUSS_CLUSTER_EXTENSION.startTabletServer(stoppedServerId);
FLUSS_CLUSTER_EXTENSION.assertHasTabletServerNumber(3);

for (int bucket = 0; bucket < 3; bucket++) {
TableBucket tb = new TableBucket(tableId, bucket);
FLUSS_CLUSTER_EXTENSION.waitUntilReplicaExpandToIsr(tb, stoppedServerId);
}

waitUntil(
() ->
admin.describeTabletServers().get().stream()
.allMatch(FlussAdminITCase::isServerGreen),
Duration.ofMinutes(1),
"All tablet servers should become green again after server restart");
}

private static TabletServerDescription getTabletServerDescription(
List<TabletServerDescription> servers, int serverId) {
return servers.stream()
.filter(server -> server.getServerId() == serverId)
.findFirst()
.orElseThrow(
() ->
new AssertionError(
"no description reported for tablet server " + serverId));
}

private static boolean isServerGreen(TabletServerDescription server) {
return server.getInSyncReplicas() == server.getNumReplicas()
&& server.getActiveLeaderReplicas() == server.getNumLeaderReplicas();
}
}
Original file line number Diff line number Diff line change
Expand Up @@ -26,6 +26,7 @@
import org.apache.fluss.client.admin.OffsetSpec;
import org.apache.fluss.client.admin.ProducerOffsetsResult;
import org.apache.fluss.client.admin.RegisterResult;
import org.apache.fluss.client.admin.TabletServerDescription;
import org.apache.fluss.client.metadata.ActiveKvSnapshots;
import org.apache.fluss.client.metadata.KvSnapshotMetadata;
import org.apache.fluss.client.metadata.KvSnapshots;
Expand Down Expand Up @@ -326,6 +327,11 @@ public CompletableFuture<ClusterHealth> getClusterHealth() {
throw new UnsupportedOperationException("Not implemented in TestAdminAdapter");
}

@Override
public CompletableFuture<List<TabletServerDescription>> describeTabletServers() {
throw new UnsupportedOperationException("Not implemented in TestAdminAdapter");
}

@Override
public CompletableFuture<List<RemoteLogManifestInfo>> listRemoteLogManifests(
long tableId, @Nullable Long partitionId) {
Expand Down
Loading
Loading