Skip to content

Add metrics in Actual States Allocator and Discovery #56

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
HaoSun1993 merged 1 commit into from
Dec 23, 2025
Merged

Add metrics in Actual States Allocator and Discovery #56

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
8 changes: 4 additions & 4 deletions src/main/java/io/clustercontroller/ClusterControllerApplication.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -114,9 +114,9 @@ public ShardAllocator shardAllocator(MetadataStore metadataStore, MetricsProvide
}

@Bean
public ActualAllocationUpdater actualAllocationUpdater(MetadataStore metadataStore) {
public ActualAllocationUpdater actualAllocationUpdater(MetadataStore metadataStore, MetricsProvider metricsProvider) {
log.info("Initializing ActualAllocationUpdater");
return new ActualAllocationUpdater(metadataStore);
return new ActualAllocationUpdater(metadataStore, metricsProvider);
}

/**
Expand All @@ -132,9 +132,9 @@ public GoalStateOrchestrator goalStateOrchestrator(MetadataStore metadataStore,
* Discovery bean for node discovery
*/
@Bean
public Discovery discovery(MetadataStore metadataStore) {
public Discovery discovery(MetadataStore metadataStore, MetricsProvider metricsProvider) {
log.info("Initializing Discovery for multi-cluster support");
return new Discovery(metadataStore);
return new Discovery(metadataStore, metricsProvider);
}

/**
Expand Down
45 changes: 31 additions & 14 deletions src/main/java/io/clustercontroller/allocation/ActualAllocationUpdater.java
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,30 +1,46 @@
package io.clustercontroller.allocation;

import io.clustercontroller.enums.ShardState;
import io.clustercontroller.models.*;
import io.clustercontroller.metrics.MetricsProvider;
import io.clustercontroller.models.Alias;
import io.clustercontroller.models.CoordinatorGoalState;
import io.clustercontroller.models.Index;
import io.clustercontroller.models.IndexSettings;
import io.clustercontroller.models.SearchUnit;
import io.clustercontroller.models.SearchUnitActualState;
import io.clustercontroller.models.SearchUnitGoalState;
import io.clustercontroller.models.ShardAllocation;
import io.clustercontroller.store.MetadataStore;
import lombok.extern.slf4j.Slf4j;
import io.clustercontroller.config.Constants;

import java.util.*;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;

import static io.clustercontroller.metrics.MetricsConstants.UPDATE_ACTUAL_ALLOCATION_FAILURES_METRIC_NAME;
import static io.clustercontroller.metrics.MetricsUtils.buildMetricsTags;

/**
* ActualAllocationUpdater - Responsible for aggregating search unit actual states into index shard actual allocations
*
* <p>
* This task:
* 1. Reads actual states from /search-units/<su-name>/actual-state
* 2. Aggregates this information by index and shard
* 3. Updates actual allocations at /indices/<index-name>/<shard_id>/actual-allocation
* 4. Provides coordinator nodes with a goal-state view of remote_shards used for routing/monitoring
*
* <p>
* Coordinator goal state specifics:
* - For each index and shard, we compute a list of nodes with their role (primary/replica) flag: List<List<NodeRoutingInfo>>.
* - A node is included only when it is converged for that index/shard.
* Converged means: (a) the node's goal state requests that index+shard (with a role), and
* (b) the node's actual state reports that shard in STARTED state.
* - Primaries come from the current actual ingest set; replicas come from the current actual search set,
* and both are filtered by convergence before inclusion.
*
* <p>
* Example:
* - Index "logs-2025" has 2 shards.
* - Shard 0: su-a (primary, STARTED, requested in goal) and su-b (replica, STARTED, requested in goal)
Expand All @@ -36,10 +52,12 @@
public class ActualAllocationUpdater {

private final MetadataStore metadataStore;
private final MetricsProvider metricsProvider;
private static final long ONE_MINUTE_MS = 60 * 1000;

public ActualAllocationUpdater(MetadataStore metadataStore) {
public ActualAllocationUpdater(MetadataStore metadataStore, MetricsProvider metricsProvider) {
this.metadataStore = metadataStore;
this.metricsProvider = metricsProvider;
}

/**
Expand Down Expand Up @@ -77,7 +95,7 @@ public void updateActualAllocations(String clusterId) throws Exception {
* Collect actual allocations from all search units
* Returns: Map<indexName, Map<shardId, Set<unitNames>>>
*/
private Map<String, Map<String, Set<String>>> collectActualAllocations(String clusterId, List<SearchUnit> searchUnits) throws Exception {
private Map<String, Map<String, Set<String>>> collectActualAllocations(String clusterId, List<SearchUnit> searchUnits) {
Map<String, Map<String, Set<String>>> actualAllocations = new HashMap<>();

for (SearchUnit searchUnit : searchUnits) {
Expand Down Expand Up @@ -152,7 +170,7 @@ private Map<String, Map<String, Set<String>>> collectActualAllocations(String cl
/**
* Update actual allocation records based on collected data
*/
private int updateActualAllocationRecords(String clusterId, Map<String, Map<String, Set<String>>> actualAllocations) throws Exception {
private int updateActualAllocationRecords(String clusterId, Map<String, Map<String, Set<String>>> actualAllocations) {
int totalUpdates = 0;

for (Map.Entry<String, Map<String, Set<String>>> indexEntry : actualAllocations.entrySet()) {
Expand All @@ -169,6 +187,10 @@ private int updateActualAllocationRecords(String clusterId, Map<String, Map<Stri
} catch (Exception e) {
log.error("ActualAllocationUpdater - Error updating actual allocation for {}/{}: {}",
indexName, shardId, e.getMessage(), e);
metricsProvider.counter(
UPDATE_ACTUAL_ALLOCATION_FAILURES_METRIC_NAME,
buildMetricsTags(clusterId, indexName, shardId)
).increment();
// Continue with other shards
}
}
Expand Down Expand Up @@ -349,7 +371,6 @@ private Set<String> getAllIndicesWithActualAllocations(String clusterId) throws
/**
* Determines if a search unit is a coordinator node
* Coordinator nodes belong to a separate cluster and are not part of shard allocation
*
* Current logic: role="coordinator" OR name starts with "coordinator"
* TODO: Use cluster_name information to properly identify coordinator nodes
* instead of relying on naming conventions. Coordinator nodes should have
Expand All @@ -368,11 +389,7 @@ private boolean isCoordinatorNode(SearchUnit unit) {

// Check if name starts with "coordinator" prefix (for data nodes acting as coordinators)
String name = unit.getName() != null ? unit.getName().toLowerCase() : "";
if (name.startsWith("coordinator")) {
return true;
}

return false;
return name.startsWith("coordinator");

@TejasNaikk TejasNaikk Dec 23, 2025

Copy link
Copy Markdown
Owner

Choose a reason for hiding this comment

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

make as const

}

/**
Expand Down
19 changes: 14 additions & 5 deletions src/main/java/io/clustercontroller/discovery/Discovery.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2,6 +2,7 @@

import io.clustercontroller.config.Constants;
import io.clustercontroller.enums.HealthState;
import io.clustercontroller.metrics.MetricsProvider;
import io.clustercontroller.models.NodeAttributes;
import io.clustercontroller.models.SearchUnit;
import io.clustercontroller.models.SearchUnitActualState;
Expand All @@ -13,24 +14,29 @@
import java.util.List;
import java.util.Map;

import static io.clustercontroller.metrics.MetricsConstants.CLUSTER_ID_TAG;
import static io.clustercontroller.metrics.MetricsConstants.DISCOVERY_CLEANED_STALE_SEARCH_UNITS_COUNT_METRIC_NAME;

/**
* Handles cluster topology discovery.
* Internal component used by TaskManager.
*
* <p>
* Note: This is a shared component across all clusters in multi-cluster mode.
* Methods accept clusterName as a parameter instead of storing it as a field.
*/
@Slf4j
public class Discovery {

private final MetadataStore metadataStore;
private final MetricsProvider metricsProvider;

public Discovery(MetadataStore metadataStore) {
public Discovery(MetadataStore metadataStore, MetricsProvider metricsProvider) {
this.metadataStore = metadataStore;
this.metricsProvider = metricsProvider;
}


public void discoverSearchUnits(String clusterName) throws Exception {
public void discoverSearchUnits(String clusterName) {
log.info("Discovery - Starting search unit discovery process for cluster: {}", clusterName);

// Discover and update search units from Etcd actual-states
Expand Down Expand Up @@ -75,7 +81,6 @@ private void processAllSearchUnits(String clusterName) {
*/
private void discoverSearchUnitsFromEtcd(String clusterName) {
try {
log.info("Discovery - Discovering search units from Etcd...");
// fetch search units from actual-state paths
List<SearchUnit> etcdSearchUnits = fetchSearchUnitsFromEtcd(clusterName);
log.info("Discovery - Found {} search units from Etcd", etcdSearchUnits.size());
Expand Down Expand Up @@ -175,12 +180,12 @@ private SearchUnit convertActualStateToSearchUnit(SearchUnitActualState actualSt
* Clean up search units with missing or stale actual state timestamp (older than configured timeout)
*/
private void cleanupStaleSearchUnits(String clusterName) {
int deletedCount = 0;
try {
log.info("Discovery - Starting cleanup of stale search units...");

// Get all existing search units from metadata store
List<SearchUnit> allSearchUnits = metadataStore.getAllSearchUnits(clusterName);
int deletedCount = 0;

for (SearchUnit searchUnit : allSearchUnits) {
String unitName = searchUnit.getName();
Expand Down Expand Up @@ -221,6 +226,10 @@ private void cleanupStaleSearchUnits(String clusterName) {

} catch (Exception e) {
log.error("Discovery - Failed to cleanup stale search units: {}", e.getMessage(), e);
} finally {
metricsProvider
.counter(DISCOVERY_CLEANED_STALE_SEARCH_UNITS_COUNT_METRIC_NAME, Map.of(CLUSTER_ID_TAG, clusterName))
.increment(deletedCount);
}
}

Expand Down
2 changes: 2 additions & 0 deletions src/main/java/io/clustercontroller/metrics/MetricsConstants.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -8,6 +8,8 @@ public class MetricsConstants {
public final static String ROLLING_UPDATE_TRANSIT_NODES_PERCENTAGE_METRIC_NAME = "rolling_update_transit_nodes_percentage";
public final static String PLANNED_INGEST_SU_ALLOCATION_METRIC_NAME = "planned_ingest_su_allocation";
public final static String PLANNED_SEARCH_SU_ALLOCATION_METRIC_NAME = "planned_search_su_allocation";
public final static String UPDATE_ACTUAL_ALLOCATION_FAILURES_METRIC_NAME = "update_actual_allocation_failures_count";
public final static String DISCOVERY_CLEANED_STALE_SEARCH_UNITS_COUNT_METRIC_NAME = "discovery_cleaned_stale_search_units_count";
public final static String CLUSTER_ID_TAG = "clusterId";
public final static String INDEX_NAME_TAG = "indexName";
public final static String SHARD_ID_TAG = "shardId";
Expand Down
90 changes: 77 additions & 13 deletions src/test/java/io/clustercontroller/allocation/ActualAllocationUpdaterCleanupTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,18 +1,19 @@
package io.clustercontroller.allocation;

import io.clustercontroller.enums.ShardState;
import io.clustercontroller.metrics.MetricsProvider;
import io.clustercontroller.models.Index;
import io.clustercontroller.models.IndexSettings;
import io.clustercontroller.models.SearchUnit;
import io.clustercontroller.models.SearchUnitActualState;
import io.clustercontroller.models.ShardAllocation;
import io.clustercontroller.store.MetadataStore;
import io.micrometer.core.instrument.Counter;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.mockito.Mock;
import org.mockito.MockitoAnnotations;

import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
Expand All @@ -29,12 +30,20 @@ class ActualAllocationUpdaterCleanupTest {
@Mock
private MetadataStore metadataStore;

@Mock
private MetricsProvider metricsProvider;

@Mock
private Counter mockCounter;

private ActualAllocationUpdater updater;

@BeforeEach
void setUp() {
MockitoAnnotations.openMocks(this);
updater = new ActualAllocationUpdater(metadataStore);
updater = new ActualAllocationUpdater(metadataStore, metricsProvider);
// Use lenient() for metrics since not all tests will trigger metric emissions
lenient().when(metricsProvider.counter(anyString(), anyMap())).thenReturn(mockCounter);
}

// ========== PHASE 1 TESTS: Existing Index - Selective Cleanup ==========
Expand All @@ -49,22 +58,22 @@ void testPhase1_ExistingIndex_ClearsStaleShards() throws Exception {
SearchUnit unit = createSearchUnit("node1");
SearchUnitActualState emptyState = new SearchUnitActualState();
emptyState.setNodeRouting(new HashMap<>());
when(metadataStore.getAllSearchUnits(clusterId)).thenReturn(Arrays.asList(unit));
when(metadataStore.getAllSearchUnits(clusterId)).thenReturn(List.of(unit));
when(metadataStore.getSearchUnitActualState(clusterId, "node1")).thenReturn(emptyState);

// Index config exists
Index indexConfig = createIndex(indexName, 2);
when(metadataStore.getAllIndexConfigs(clusterId)).thenReturn(Arrays.asList(indexConfig));
when(metadataStore.getAllIndexConfigs(clusterId)).thenReturn(List.of(indexConfig));

// Mock that the index has actual-allocation entries in etcd
when(metadataStore.getAllIndicesWithActualAllocations(clusterId))
.thenReturn(Collections.singleton(indexName));

// Stored actual allocations in etcd (stale)
ShardAllocation storedShard0 = createShardAllocation("0", Arrays.asList("node1"), Arrays.asList());
ShardAllocation storedShard1 = createShardAllocation("1", Arrays.asList("node1"), Arrays.asList());
ShardAllocation storedShard0 = createShardAllocation("0", List.of("node1"), List.of());
ShardAllocation storedShard1 = createShardAllocation("1", List.of("node1"), List.of());
when(metadataStore.getAllActualAllocations(clusterId, indexName))
.thenReturn(Arrays.asList(storedShard0, storedShard1));
.thenReturn(List.of(storedShard0, storedShard1));

// When
updater.updateActualAllocations(clusterId);
Expand All @@ -88,9 +97,9 @@ void testPhase2_DeletedIndex_DeletesAllAllocations() throws Exception {

// Node reports having shards from the deleted index
SearchUnit unit = createSearchUnit("node1");
SearchUnitActualState actualState = createActualStateWithShards(deletedIndex, Arrays.asList(0, 1, 2));
SearchUnitActualState actualState = createActualStateWithShards(deletedIndex, List.of(0, 1, 2));

when(metadataStore.getAllSearchUnits(clusterId)).thenReturn(Arrays.asList(unit));
when(metadataStore.getAllSearchUnits(clusterId)).thenReturn(List.of(unit));
when(metadataStore.getSearchUnitActualState(clusterId, "node1")).thenReturn(actualState);

// NO config exists for deleted-index (this makes it a deleted index)
Expand All @@ -101,12 +110,12 @@ void testPhase2_DeletedIndex_DeletesAllAllocations() throws Exception {
.thenReturn(Collections.singleton(deletedIndex));

// The deleted index has actual allocations stored in etcd (orphaned entries)
ShardAllocation shard0 = createShardAllocation("0", Arrays.asList("node1"), Arrays.asList());
ShardAllocation shard1 = createShardAllocation("1", Arrays.asList("node1"), Arrays.asList());
ShardAllocation shard2 = createShardAllocation("2", Arrays.asList("node1"), Arrays.asList());
ShardAllocation shard0 = createShardAllocation("0", List.of("node1"), List.of());
ShardAllocation shard1 = createShardAllocation("1", List.of("node1"), List.of());
ShardAllocation shard2 = createShardAllocation("2", List.of("node1"), List.of());

when(metadataStore.getAllActualAllocations(clusterId, deletedIndex))
.thenReturn(Arrays.asList(shard0, shard1, shard2));
.thenReturn(List.of(shard0, shard1, shard2));

// When
updater.updateActualAllocations(clusterId);
Expand All @@ -123,6 +132,61 @@ void testPhase2_DeletedIndex_DeletesAllAllocations() throws Exception {
verify(metadataStore).deleteActualAllocation(eq(clusterId), eq(deletedIndex), eq("2"));
}


@Test
void testMetricsCounter_IncrementedOnAllocationUpdateFailure() throws Exception {
String clusterId = "test-cluster";
String indexName = "test-index";
String shardId = "0";

SearchUnit unit = createSearchUnit("node1");
SearchUnitActualState actualState = createActualStateWithShards(indexName, List.of(0));
when(metadataStore.getAllSearchUnits(clusterId)).thenReturn(List.of(unit));
when(metadataStore.getSearchUnitActualState(clusterId, "node1")).thenReturn(actualState);
Index indexConfig = createIndex(indexName, 1);
when(metadataStore.getAllIndexConfigs(clusterId)).thenReturn(List.of(indexConfig));
when(metadataStore.getAllIndicesWithActualAllocations(clusterId))
.thenReturn(Collections.singleton(indexName));
when(metadataStore.getActualAllocation(clusterId, indexName, shardId))
.thenThrow(new RuntimeException("Database connection failed"));

updater.updateActualAllocations(clusterId);

verify(metricsProvider).counter(
eq("update_actual_allocation_failures_count"),
argThat(tags ->
tags.get("clusterId").equals(clusterId) &&
tags.get("indexName").equals(indexName) &&
tags.get("shardId").equals(shardId)
)
);
verify(mockCounter).increment();
}

@Test
void testMetricsCounter_NotIncrementedOnSuccess() throws Exception {
String clusterId = "test-cluster";
String indexName = "test-index";

SearchUnit unit = createSearchUnit("node1");
unit.setRole("PRIMARY");
SearchUnitActualState actualState = createActualStateWithShards(indexName, List.of(0));

when(metadataStore.getAllSearchUnits(clusterId)).thenReturn(List.of(unit));
when(metadataStore.getSearchUnitActualState(clusterId, "node1")).thenReturn(actualState);
when(metadataStore.getSearchUnit(clusterId, "node1"))
.thenReturn(java.util.Optional.of(unit));
Index indexConfig = createIndex(indexName, 1);
when(metadataStore.getAllIndexConfigs(clusterId)).thenReturn(List.of(indexConfig));
when(metadataStore.getAllIndicesWithActualAllocations(clusterId))
.thenReturn(Collections.singleton(indexName));
when(metadataStore.getActualAllocation(clusterId, indexName, "0")).thenReturn(null);
doNothing().when(metadataStore).setActualAllocation(eq(clusterId), eq(indexName), eq("0"), any());

updater.updateActualAllocations(clusterId);
verify(metricsProvider, never()).counter(eq("update_actual_allocation_failures_count"), anyMap());
verify(mockCounter, never()).increment();
}
// Helper methods
private Index createIndex(String indexName, int numberOfShards) {
Index index = new Index();
Expand Down
Loading