Skip to content

Commit a001b78

Browse files
tvernumtvernum
committed
Refactor security index migration to its own class
This moves a number of fields and methods from the `Security` plugin class into a new `SecurityMigrations.Manager` class. This refactor makes it easier to reason about the behaviour of the migration process which will allow us to make it project aware
1 parent d2854f5 commit a001b78

File tree

2 files changed

+102
-87
lines changed

2 files changed

+102
-87
lines changed

‎x-pack/plugin/security/src/main/java/org/elasticsearch/xpack/security/Security.java‎

Lines changed: 4 additions & 87 deletions
Original file line numberDiff line numberDiff line change
@@ -16,8 +16,6 @@
1616
import org.elasticsearch.ElasticsearchException;
1717
import org.elasticsearch.ElasticsearchSecurityException;
1818
import org.elasticsearch.ElasticsearchStatusException;
19-
import org.elasticsearch.ExceptionsHelper;
20-
import org.elasticsearch.ResourceAlreadyExistsException;
2119
import org.elasticsearch.TransportVersion;
2220
import org.elasticsearch.action.ActionListener;
2321
import org.elasticsearch.action.ActionResponse;
@@ -30,7 +28,6 @@
3028
import org.elasticsearch.cluster.ClusterState;
3129
import org.elasticsearch.cluster.metadata.IndexNameExpressionResolver;
3230
import org.elasticsearch.cluster.metadata.IndexTemplateMetadata;
33-
import org.elasticsearch.cluster.metadata.ProjectId;
3431
import org.elasticsearch.cluster.node.DiscoveryNode;
3532
import org.elasticsearch.cluster.node.DiscoveryNodes;
3633
import org.elasticsearch.cluster.project.ProjectResolver;
@@ -61,11 +58,9 @@
6158
import org.elasticsearch.common.util.concurrent.ThreadContext;
6259
import org.elasticsearch.common.util.concurrent.ThrottledTaskRunner;
6360
import org.elasticsearch.common.util.set.Sets;
64-
import org.elasticsearch.core.FixForMultiProject;
6561
import org.elasticsearch.core.IOUtils;
6662
import org.elasticsearch.core.Nullable;
6763
import org.elasticsearch.core.Releasable;
68-
import org.elasticsearch.core.TimeValue;
6964
import org.elasticsearch.env.Environment;
7065
import org.elasticsearch.features.FeatureService;
7166
import org.elasticsearch.features.NodeFeature;
@@ -224,7 +219,6 @@
224219
import org.elasticsearch.xpack.core.security.authz.store.ReservedRolesStore;
225220
import org.elasticsearch.xpack.core.security.authz.store.RoleRetrievalResult;
226221
import org.elasticsearch.xpack.core.security.support.Automatons;
227-
import org.elasticsearch.xpack.core.security.support.SecurityMigrationTaskParams;
228222
import org.elasticsearch.xpack.core.security.user.AnonymousUser;
229223
import org.elasticsearch.xpack.core.ssl.SSLConfigurationSettings;
230224
import org.elasticsearch.xpack.core.ssl.SSLService;
@@ -419,8 +413,6 @@
419413
import org.elasticsearch.xpack.security.support.QueryableBuiltInRolesProviderFactory;
420414
import org.elasticsearch.xpack.security.support.QueryableBuiltInRolesSynchronizer;
421415
import org.elasticsearch.xpack.security.support.ReloadableSecurityComponent;
422-
import org.elasticsearch.xpack.security.support.SecurityIndexManager;
423-
import org.elasticsearch.xpack.security.support.SecurityMigrationExecutor;
424416
import org.elasticsearch.xpack.security.support.SecurityMigrations;
425417
import org.elasticsearch.xpack.security.support.SecuritySystemIndices;
426418
import org.elasticsearch.xpack.security.transport.SecurityHttpSettings;
@@ -449,7 +441,6 @@
449441
import java.util.Objects;
450442
import java.util.Set;
451443
import java.util.concurrent.Executor;
452-
import java.util.concurrent.atomic.AtomicInteger;
453444
import java.util.function.BiConsumer;
454445
import java.util.function.Function;
455446
import java.util.function.Predicate;
@@ -487,8 +478,6 @@ public class Security extends Plugin
487478

488479
public static final String SECURITY_CRYPTO_THREAD_POOL_NAME = XPackField.SECURITY + "-crypto";
489480

490-
private static final int MAX_SECURITY_MIGRATION_RETRY_COUNT = 10;
491-
492481
// TODO: ip filtering does not actually track license usage yet
493482
public static final LicensedFeature.Momentary IP_FILTERING_FEATURE = LicensedFeature.momentaryLenient(
494483
null,
@@ -628,7 +617,6 @@ public class Security extends Plugin
628617
private final SetOnce<GetBuiltinPrivilegesResponseTranslator> getBuiltinPrivilegesResponseTranslator = new SetOnce<>();
629618
private final SetOnce<HasPrivilegesRequestBuilderFactory> hasPrivilegesRequestBuilderFactory = new SetOnce<>();
630619

631-
private final SetOnce<PersistentTasksService> persistentTasksService = new SetOnce<>();
632620
private final SetOnce<FileRolesStore> fileRolesStore = new SetOnce<>();
633621
private final SetOnce<OperatorPrivileges.OperatorPrivilegesService> operatorPrivilegesService = new SetOnce<>();
634622
private final SetOnce<ReservedRoleMappingAction> reservedRoleMappingAction = new SetOnce<>();
@@ -640,13 +628,8 @@ public class Security extends Plugin
640628
private final SetOnce<FileRoleValidator> fileRoleValidator = new SetOnce<>();
641629
private final SetOnce<SecondaryAuthActions> secondaryAuthActions = new SetOnce<>();
642630
private final SetOnce<QueryableBuiltInRolesProviderFactory> queryableRolesProviderFactory = new SetOnce<>();
643-
private final SetOnce<SecurityMigrationExecutor> securityMigrationExecutor = new SetOnce<>();
644-
645-
// Node local retry count for migration jobs that's checked only on the master node to make sure
646-
// submit migration jobs doesn't get out of hand and retries forever if they fail. Reset by a
647-
// restart or master node change.
648-
private final AtomicInteger nodeLocalMigrationRetryCount = new AtomicInteger(0);
649631

632+
private final SetOnce<SecurityMigrations.Manager> migrationManager = new SetOnce<>();
650633
private final SetOnce<List<Closeable>> closableComponents = new SetOnce<>();
651634

652635
public Security(Settings settings) {
@@ -796,23 +779,7 @@ Collection<Object> createComponents(
796779

797780
systemIndices.init(client, featureService, clusterService, projectResolver);
798781

799-
this.securityMigrationExecutor.set(
800-
new SecurityMigrationExecutor(
801-
SecurityMigrationTaskParams.TASK_NAME,
802-
threadPool.executor(ThreadPool.Names.MANAGEMENT),
803-
systemIndices.getMainIndexManager(),
804-
client,
805-
SecurityMigrations.MIGRATIONS_BY_VERSION
806-
)
807-
);
808-
this.persistentTasksService.set(persistentTasksService);
809-
810-
systemIndices.getMainIndexManager().addStateListener((projectId, oldState, newState) -> {
811-
// Only consider applying migrations if it's the master node and the security index exists
812-
if (clusterService.state().nodes().isLocalNodeElectedMaster() && newState.indexExists()) {
813-
applyPendingSecurityMigrations(projectId, newState);
814-
}
815-
});
782+
this.migrationManager.set(new SecurityMigrations.Manager(clusterService, persistentTasksService, systemIndices));
816783

817784
scriptServiceReference.set(scriptService);
818785
// We need to construct the checks here while the secure settings are still available.
@@ -1362,57 +1329,6 @@ private static boolean isInternalExtension(SecurityExtension extension) {
13621329
return canonicalName.startsWith("org.elasticsearch.xpack.") || canonicalName.startsWith("co.elastic.elasticsearch.");
13631330
}
13641331

1365-
@FixForMultiProject
1366-
// TODO : The migration task needs to be project aware
1367-
private void applyPendingSecurityMigrations(ProjectId projectId, SecurityIndexManager.IndexState newState) {
1368-
// If no migrations have been applied and the security index is on the latest version (new index), all migrations can be skipped
1369-
if (newState.migrationsVersion == 0 && newState.createdOnLatestVersion) {
1370-
submitPersistentMigrationTask(SecurityMigrations.MIGRATIONS_BY_VERSION.lastKey(), false);
1371-
return;
1372-
}
1373-
1374-
Map.Entry<Integer, SecurityMigrations.SecurityMigration> nextMigration = SecurityMigrations.MIGRATIONS_BY_VERSION.higherEntry(
1375-
newState.migrationsVersion
1376-
);
1377-
1378-
// Check if next migration that has not been applied is eligible to run on the current cluster
1379-
if (nextMigration == null
1380-
|| systemIndices.getMainIndexManager().getProject(projectId).isEligibleSecurityMigration(nextMigration.getValue()) == false) {
1381-
// Reset retry counter if all eligible migrations have been applied successfully
1382-
nodeLocalMigrationRetryCount.set(0);
1383-
} else if (nodeLocalMigrationRetryCount.get() > MAX_SECURITY_MIGRATION_RETRY_COUNT) {
1384-
logger.warn("Security migration failed [" + nodeLocalMigrationRetryCount.get() + "] times, restart node to retry again.");
1385-
} else if (systemIndices.getMainIndexManager().getProject(projectId).isReadyForSecurityMigration(nextMigration.getValue())) {
1386-
submitPersistentMigrationTask(newState.migrationsVersion);
1387-
}
1388-
}
1389-
1390-
private void submitPersistentMigrationTask(int migrationsVersion) {
1391-
submitPersistentMigrationTask(migrationsVersion, true);
1392-
}
1393-
1394-
private void submitPersistentMigrationTask(int migrationsVersion, boolean securityMigrationNeeded) {
1395-
nodeLocalMigrationRetryCount.incrementAndGet();
1396-
persistentTasksService.get()
1397-
.sendStartRequest(
1398-
SecurityMigrationTaskParams.TASK_NAME,
1399-
SecurityMigrationTaskParams.TASK_NAME,
1400-
new SecurityMigrationTaskParams(migrationsVersion, securityMigrationNeeded),
1401-
TimeValue.THIRTY_SECONDS /* TODO should this be configurable? longer by default? infinite? */,
1402-
ActionListener.wrap((response) -> {
1403-
logger.debug("Security migration task submitted");
1404-
}, (exception) -> {
1405-
// Do nothing if the task is already in progress
1406-
if (ExceptionsHelper.unwrapCause(exception) instanceof ResourceAlreadyExistsException) {
1407-
// Do not count ResourceAlreadyExistsException as failure
1408-
nodeLocalMigrationRetryCount.decrementAndGet();
1409-
} else {
1410-
logger.warn("Submit security migration task failed: " + exception.getCause());
1411-
}
1412-
})
1413-
);
1414-
}
1415-
14161332
private static Executor buildRoleBuildingExecutor(ThreadPool threadPool, Settings settings) {
14171333
final int allocatedProcessors = EsExecutors.allocatedProcessors(settings);
14181334
final ThrottledTaskRunner throttledTaskRunner = new ThrottledTaskRunner("build_roles", allocatedProcessors, threadPool.generic());
@@ -2558,7 +2474,8 @@ public List<PersistentTasksExecutor<?>> getPersistentTasksExecutor(
25582474
SettingsModule settingsModule,
25592475
IndexNameExpressionResolver expressionResolver
25602476
) {
2561-
return this.securityMigrationExecutor.get() != null ? List.of(this.securityMigrationExecutor.get()) : List.of();
2477+
final SecurityMigrations.Manager manager = this.migrationManager.get();
2478+
return manager == null ? List.of() : List.of(manager.getPersistentTasksExecutor(client, threadPool));
25622479
}
25632480

25642481
List<ReservedProjectStateHandler<?>> reservedProjectStateHandlers() {

‎x-pack/plugin/security/src/main/java/org/elasticsearch/xpack/security/support/SecurityMigrations.java‎

Lines changed: 98 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -9,26 +9,36 @@
99

1010
import org.apache.logging.log4j.LogManager;
1111
import org.apache.logging.log4j.Logger;
12+
import org.elasticsearch.ExceptionsHelper;
13+
import org.elasticsearch.ResourceAlreadyExistsException;
1214
import org.elasticsearch.action.ActionListener;
1315
import org.elasticsearch.action.search.SearchRequest;
1416
import org.elasticsearch.action.support.GroupedActionListener;
1517
import org.elasticsearch.action.support.WriteRequest;
1618
import org.elasticsearch.client.internal.Client;
19+
import org.elasticsearch.cluster.metadata.ProjectId;
20+
import org.elasticsearch.cluster.service.ClusterService;
21+
import org.elasticsearch.core.FixForMultiProject;
22+
import org.elasticsearch.core.TimeValue;
1723
import org.elasticsearch.features.NodeFeature;
1824
import org.elasticsearch.index.query.BoolQueryBuilder;
1925
import org.elasticsearch.index.query.QueryBuilders;
2026
import org.elasticsearch.index.reindex.UpdateByQueryAction;
2127
import org.elasticsearch.index.reindex.UpdateByQueryRequest;
28+
import org.elasticsearch.persistent.PersistentTasksExecutor;
29+
import org.elasticsearch.persistent.PersistentTasksService;
2230
import org.elasticsearch.script.Script;
2331
import org.elasticsearch.script.ScriptType;
2432
import org.elasticsearch.search.builder.SearchSourceBuilder;
33+
import org.elasticsearch.threadpool.ThreadPool;
2534
import org.elasticsearch.xpack.core.security.action.rolemapping.DeleteRoleMappingAction;
2635
import org.elasticsearch.xpack.core.security.action.rolemapping.DeleteRoleMappingRequestBuilder;
2736
import org.elasticsearch.xpack.core.security.action.rolemapping.DeleteRoleMappingResponse;
2837
import org.elasticsearch.xpack.core.security.action.rolemapping.GetRoleMappingsAction;
2938
import org.elasticsearch.xpack.core.security.action.rolemapping.GetRoleMappingsRequestBuilder;
3039
import org.elasticsearch.xpack.core.security.action.rolemapping.GetRoleMappingsResponse;
3140
import org.elasticsearch.xpack.core.security.authc.support.mapper.ExpressionRoleMapping;
41+
import org.elasticsearch.xpack.core.security.support.SecurityMigrationTaskParams;
3242
import org.elasticsearch.xpack.security.support.SecurityIndexManager.IndexState;
3343

3444
import java.util.Arrays;
@@ -37,6 +47,7 @@
3747
import java.util.Map;
3848
import java.util.Set;
3949
import java.util.TreeMap;
50+
import java.util.concurrent.atomic.AtomicInteger;
4051
import java.util.stream.Collectors;
4152

4253
import static org.elasticsearch.xpack.core.ClientHelper.SECURITY_ORIGIN;
@@ -250,4 +261,91 @@ protected static List<String> getDuplicateRoleMappingNames(ExpressionRoleMapping
250261
.toList();
251262
}
252263
}
264+
265+
public static class Manager {
266+
267+
private static final int MAX_SECURITY_MIGRATION_RETRY_COUNT = 10;
268+
269+
private final PersistentTasksService persistentTasksService;
270+
private final SecuritySystemIndices systemIndices;
271+
272+
// Node local retry count for migration jobs that's checked only on the master node to make sure
273+
// submit migration jobs doesn't get out of hand and retries forever if they fail. Reset by a
274+
// restart or master node change.
275+
private final AtomicInteger nodeLocalMigrationRetryCount;
276+
277+
public Manager(ClusterService clusterService, PersistentTasksService persistentTasksService, SecuritySystemIndices systemIndices) {
278+
this.persistentTasksService = persistentTasksService;
279+
this.systemIndices = systemIndices;
280+
this.nodeLocalMigrationRetryCount = new AtomicInteger(0);
281+
systemIndices.getMainIndexManager().addStateListener((projectId, oldState, newState) -> {
282+
// Only consider applying migrations if it's the master node and the security index exists
283+
if (clusterService.state().nodes().isLocalNodeElectedMaster() && newState.indexExists()) {
284+
applyPendingSecurityMigrations(projectId, newState);
285+
}
286+
});
287+
}
288+
289+
@FixForMultiProject
290+
// TODO : The migration task needs to be project aware
291+
private void applyPendingSecurityMigrations(ProjectId projectId, SecurityIndexManager.IndexState newState) {
292+
// If no migrations have been applied and the security index is on the latest version (new index), all migrations can be skipped
293+
if (newState.migrationsVersion == 0 && newState.createdOnLatestVersion) {
294+
submitPersistentMigrationTask(SecurityMigrations.MIGRATIONS_BY_VERSION.lastKey(), false);
295+
return;
296+
}
297+
298+
Map.Entry<Integer, SecurityMigrations.SecurityMigration> nextMigration = SecurityMigrations.MIGRATIONS_BY_VERSION.higherEntry(
299+
newState.migrationsVersion
300+
);
301+
302+
// Check if next migration that has not been applied is eligible to run on the current cluster
303+
if (nextMigration == null
304+
|| systemIndices.getMainIndexManager()
305+
.getProject(projectId)
306+
.isEligibleSecurityMigration(nextMigration.getValue()) == false) {
307+
// Reset retry counter if all eligible migrations have been applied successfully
308+
nodeLocalMigrationRetryCount.set(0);
309+
} else if (nodeLocalMigrationRetryCount.get() > MAX_SECURITY_MIGRATION_RETRY_COUNT) {
310+
logger.warn("Security migration failed [" + nodeLocalMigrationRetryCount.get() + "] times, restart node to retry again.");
311+
} else if (systemIndices.getMainIndexManager().getProject(projectId).isReadyForSecurityMigration(nextMigration.getValue())) {
312+
submitPersistentMigrationTask(newState.migrationsVersion);
313+
}
314+
}
315+
316+
private void submitPersistentMigrationTask(int migrationsVersion) {
317+
submitPersistentMigrationTask(migrationsVersion, true);
318+
}
319+
320+
private void submitPersistentMigrationTask(int migrationsVersion, boolean securityMigrationNeeded) {
321+
nodeLocalMigrationRetryCount.incrementAndGet();
322+
persistentTasksService.sendStartRequest(
323+
SecurityMigrationTaskParams.TASK_NAME,
324+
SecurityMigrationTaskParams.TASK_NAME,
325+
new SecurityMigrationTaskParams(migrationsVersion, securityMigrationNeeded),
326+
TimeValue.THIRTY_SECONDS /* TODO should this be configurable? longer by default? infinite? */,
327+
ActionListener.wrap((response) -> {
328+
logger.debug("Security migration task submitted");
329+
}, (exception) -> {
330+
// Do nothing if the task is already in progress
331+
if (ExceptionsHelper.unwrapCause(exception) instanceof ResourceAlreadyExistsException) {
332+
// Do not count ResourceAlreadyExistsException as failure
333+
nodeLocalMigrationRetryCount.decrementAndGet();
334+
} else {
335+
logger.warn("Submit security migration task failed: " + exception.getCause());
336+
}
337+
})
338+
);
339+
}
340+
341+
public PersistentTasksExecutor<?> getPersistentTasksExecutor(Client client, ThreadPool threadPool) {
342+
return new SecurityMigrationExecutor(
343+
SecurityMigrationTaskParams.TASK_NAME,
344+
threadPool.executor(ThreadPool.Names.MANAGEMENT),
345+
systemIndices.getMainIndexManager(),
346+
client,
347+
SecurityMigrations.MIGRATIONS_BY_VERSION
348+
);
349+
}
350+
}
253351
}

0 commit comments

Comments
 (0)