Skip to content

Commit 317af7a

Browse files
nielsbaumanmridula-s109
nielsbauman
authored and
mridula-s109
committed
Make rollup APIs project-aware (elastic#130365)
Updates the rollup APIs to work in a multi-project context (even though that will never actually happen, as rollup is deprecated and the plugin is excluded in serverless).
1 parent d27b749 commit 317af7a

File tree

8 files changed

+138
-56
lines changed

8 files changed

+138
-56
lines changed

‎x-pack/plugin/rollup/src/main/java/org/elasticsearch/xpack/rollup/action/TransportDeleteRollupJobAction.java‎

Lines changed: 11 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -15,6 +15,7 @@
1515
import org.elasticsearch.action.support.tasks.TransportTasksAction;
1616
import org.elasticsearch.cluster.ClusterState;
1717
import org.elasticsearch.cluster.node.DiscoveryNodes;
18+
import org.elasticsearch.cluster.project.ProjectResolver;
1819
import org.elasticsearch.cluster.service.ClusterService;
1920
import org.elasticsearch.common.logging.DeprecationCategory;
2021
import org.elasticsearch.common.logging.DeprecationLogger;
@@ -44,8 +45,15 @@ public class TransportDeleteRollupJobAction extends TransportTasksAction<
4445

4546
private static final DeprecationLogger DEPRECATION_LOGGER = DeprecationLogger.getLogger(TransportDeleteRollupJobAction.class);
4647

48+
private final ProjectResolver projectResolver;
49+
4750
@Inject
48-
public TransportDeleteRollupJobAction(TransportService transportService, ActionFilters actionFilters, ClusterService clusterService) {
51+
public TransportDeleteRollupJobAction(
52+
TransportService transportService,
53+
ActionFilters actionFilters,
54+
ClusterService clusterService,
55+
ProjectResolver projectResolver
56+
) {
4957
super(
5058
DeleteRollupJobAction.NAME,
5159
clusterService,
@@ -55,6 +63,7 @@ public TransportDeleteRollupJobAction(TransportService transportService, ActionF
5563
DeleteRollupJobAction.Response::new,
5664
EsExecutors.DIRECT_EXECUTOR_SERVICE
5765
);
66+
this.projectResolver = projectResolver;
5867
}
5968

6069
@Override
@@ -64,7 +73,7 @@ protected void doExecute(Task task, DeleteRollupJobAction.Request request, Actio
6473
final DiscoveryNodes nodes = state.nodes();
6574

6675
if (nodes.isLocalNodeElectedMaster()) {
67-
PersistentTasksCustomMetadata pTasksMeta = state.getMetadata().getProject().custom(PersistentTasksCustomMetadata.TYPE);
76+
PersistentTasksCustomMetadata pTasksMeta = projectResolver.getProjectMetadata(state).custom(PersistentTasksCustomMetadata.TYPE);
6877
if (pTasksMeta != null && pTasksMeta.getTask(request.getId()) != null) {
6978
super.doExecute(task, request, listener);
7079
} else {

‎x-pack/plugin/rollup/src/main/java/org/elasticsearch/xpack/rollup/action/TransportGetRollupCapsAction.java‎

Lines changed: 11 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -13,6 +13,7 @@
1313
import org.elasticsearch.cluster.metadata.IndexMetadata;
1414
import org.elasticsearch.cluster.metadata.MappingMetadata;
1515
import org.elasticsearch.cluster.metadata.Metadata;
16+
import org.elasticsearch.cluster.project.ProjectResolver;
1617
import org.elasticsearch.cluster.service.ClusterService;
1718
import org.elasticsearch.common.logging.DeprecationCategory;
1819
import org.elasticsearch.common.logging.DeprecationLogger;
@@ -43,9 +44,15 @@ public class TransportGetRollupCapsAction extends HandledTransportAction<GetRoll
4344

4445
private final ClusterService clusterService;
4546
private final Executor managementExecutor;
47+
private final ProjectResolver projectResolver;
4648

4749
@Inject
48-
public TransportGetRollupCapsAction(TransportService transportService, ClusterService clusterService, ActionFilters actionFilters) {
50+
public TransportGetRollupCapsAction(
51+
TransportService transportService,
52+
ClusterService clusterService,
53+
ActionFilters actionFilters,
54+
ProjectResolver projectResolver
55+
) {
4956
// TODO replace SAME when removing workaround for https://github.com/elastic/elasticsearch/issues/97916
5057
super(
5158
GetRollupCapsAction.NAME,
@@ -56,6 +63,7 @@ public TransportGetRollupCapsAction(TransportService transportService, ClusterSe
5663
);
5764
this.clusterService = clusterService;
5865
this.managementExecutor = transportService.getThreadPool().executor(ThreadPool.Names.MANAGEMENT);
66+
this.projectResolver = projectResolver;
5967
}
6068

6169
@Override
@@ -67,7 +75,8 @@ protected void doExecute(Task task, GetRollupCapsAction.Request request, ActionL
6775

6876
private void doExecuteForked(String indexPattern, ActionListener<GetRollupCapsAction.Response> listener) {
6977
Transports.assertNotTransportThread("retrieving rollup job caps may be expensive");
70-
Map<String, RollableIndexCaps> allCaps = getCaps(indexPattern, clusterService.state().getMetadata().getProject().indices());
78+
final var project = projectResolver.getProjectMetadata(clusterService.state());
79+
Map<String, RollableIndexCaps> allCaps = getCaps(indexPattern, project.indices());
7180
listener.onResponse(new GetRollupCapsAction.Response(allCaps));
7281
}
7382

‎x-pack/plugin/rollup/src/main/java/org/elasticsearch/xpack/rollup/action/TransportGetRollupIndexCapsAction.java‎

Lines changed: 8 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -13,6 +13,7 @@
1313
import org.elasticsearch.action.support.HandledTransportAction;
1414
import org.elasticsearch.cluster.metadata.IndexMetadata;
1515
import org.elasticsearch.cluster.metadata.IndexNameExpressionResolver;
16+
import org.elasticsearch.cluster.project.ProjectResolver;
1617
import org.elasticsearch.cluster.service.ClusterService;
1718
import org.elasticsearch.common.logging.DeprecationCategory;
1819
import org.elasticsearch.common.logging.DeprecationLogger;
@@ -46,13 +47,15 @@ public class TransportGetRollupIndexCapsAction extends HandledTransportAction<
4647
private final ClusterService clusterService;
4748
private final IndexNameExpressionResolver resolver;
4849
private final Executor managementExecutor;
50+
private final ProjectResolver projectResolver;
4951

5052
@Inject
5153
public TransportGetRollupIndexCapsAction(
5254
TransportService transportService,
5355
ClusterService clusterService,
5456
ActionFilters actionFilters,
55-
IndexNameExpressionResolver indexNameExpressionResolver
57+
IndexNameExpressionResolver indexNameExpressionResolver,
58+
ProjectResolver projectResolver
5659
) {
5760
// TODO replace SAME when removing workaround for https://github.com/elastic/elasticsearch/issues/97916
5861
super(
@@ -65,6 +68,7 @@ public TransportGetRollupIndexCapsAction(
6568
this.clusterService = clusterService;
6669
this.managementExecutor = transportService.getThreadPool().executor(ThreadPool.Names.MANAGEMENT);
6770
this.resolver = indexNameExpressionResolver;
71+
this.projectResolver = projectResolver;
6872
}
6973

7074
@Override
@@ -80,11 +84,9 @@ protected void doExecute(
8084

8185
private void doExecuteForked(IndicesRequest request, ActionListener<GetRollupIndexCapsAction.Response> listener) {
8286
Transports.assertNotTransportThread("retrieving rollup job index caps may be expensive");
83-
String[] indices = resolver.concreteIndexNames(clusterService.state(), request.indicesOptions(), request);
84-
Map<String, RollableIndexCaps> allCaps = getCapsByRollupIndex(
85-
Arrays.asList(indices),
86-
clusterService.state().getMetadata().getProject().indices()
87-
);
87+
final var project = projectResolver.getProjectMetadata(clusterService.state());
88+
String[] indices = resolver.concreteIndexNames(project, request.indicesOptions(), request);
89+
Map<String, RollableIndexCaps> allCaps = getCapsByRollupIndex(Arrays.asList(indices), project.indices());
8890
listener.onResponse(new GetRollupIndexCapsAction.Response(allCaps));
8991
}
9092

‎x-pack/plugin/rollup/src/main/java/org/elasticsearch/xpack/rollup/action/TransportPutRollupJobAction.java‎

Lines changed: 28 additions & 13 deletions
Original file line numberDiff line numberDiff line change
@@ -29,7 +29,9 @@
2929
import org.elasticsearch.cluster.block.ClusterBlockException;
3030
import org.elasticsearch.cluster.block.ClusterBlockLevel;
3131
import org.elasticsearch.cluster.metadata.MappingMetadata;
32-
import org.elasticsearch.cluster.metadata.Metadata;
32+
import org.elasticsearch.cluster.metadata.ProjectId;
33+
import org.elasticsearch.cluster.metadata.ProjectMetadata;
34+
import org.elasticsearch.cluster.project.ProjectResolver;
3335
import org.elasticsearch.cluster.service.ClusterService;
3436
import org.elasticsearch.common.logging.DeprecationCategory;
3537
import org.elasticsearch.common.logging.DeprecationLogger;
@@ -78,6 +80,7 @@ public class TransportPutRollupJobAction extends AcknowledgedTransportMasterNode
7880

7981
private final PersistentTasksService persistentTasksService;
8082
private final Client client;
83+
private final ProjectResolver projectResolver;
8184

8285
@Inject
8386
public TransportPutRollupJobAction(
@@ -86,7 +89,8 @@ public TransportPutRollupJobAction(
8689
ActionFilters actionFilters,
8790
ClusterService clusterService,
8891
PersistentTasksService persistentTasksService,
89-
Client client
92+
Client client,
93+
ProjectResolver projectResolver
9094
) {
9195
super(
9296
PutRollupJobAction.NAME,
@@ -99,7 +103,7 @@ public TransportPutRollupJobAction(
99103
);
100104
this.persistentTasksService = persistentTasksService;
101105
this.client = client;
102-
106+
this.projectResolver = projectResolver;
103107
}
104108

105109
@Override
@@ -113,10 +117,11 @@ protected void masterOperation(
113117
XPackPlugin.checkReadyForXPackCustomMetadata(clusterState);
114118
checkForDeprecatedTZ(request);
115119

116-
int numberOfCurrentRollupJobs = RollupUsageTransportAction.findNumberOfRollupJobs(clusterState.metadata().getProject());
120+
final var project = projectResolver.getProjectMetadata(clusterState);
121+
int numberOfCurrentRollupJobs = RollupUsageTransportAction.findNumberOfRollupJobs(project);
117122
if (numberOfCurrentRollupJobs == 0) {
118123
try {
119-
boolean hasRollupIndices = hasRollupIndices(clusterState.getMetadata());
124+
boolean hasRollupIndices = hasRollupIndices(project);
120125
if (hasRollupIndices == false) {
121126
listener.onFailure(
122127
new IllegalArgumentException(
@@ -135,6 +140,7 @@ protected void masterOperation(
135140
.fields(request.getConfig().getAllFields().toArray(new String[0]));
136141
fieldCapsRequest.setParentTask(clusterService.localNode().getId(), task.getId());
137142

143+
final var projectId = project.id();
138144
client.fieldCaps(fieldCapsRequest, listener.delegateFailure((l, fieldCapabilitiesResponse) -> {
139145
ActionRequestValidationException validationException = request.validateMappings(fieldCapabilitiesResponse.get());
140146
if (validationException != null) {
@@ -143,7 +149,7 @@ protected void masterOperation(
143149
}
144150

145151
RollupJob job = createRollupJob(request.getConfig(), threadPool);
146-
createIndex(job, l, persistentTasksService, client, LOGGER);
152+
createIndex(projectId, job, l, persistentTasksService, client, LOGGER);
147153
}));
148154
}
149155

@@ -177,6 +183,7 @@ private RollupJob createRollupJob(RollupJobConfig config, ThreadPool threadPool)
177183
}
178184

179185
static void createIndex(
186+
ProjectId projectId,
180187
RollupJob job,
181188
ActionListener<AcknowledgedResponse> listener,
182189
PersistentTasksService persistentTasksService,
@@ -196,10 +203,10 @@ static void createIndex(
196203
client.execute(
197204
TransportCreateIndexAction.TYPE,
198205
request,
199-
ActionListener.wrap(createIndexResponse -> startPersistentTask(job, listener, persistentTasksService), e -> {
206+
ActionListener.wrap(createIndexResponse -> startPersistentTask(projectId, job, listener, persistentTasksService), e -> {
200207
if (e instanceof ResourceAlreadyExistsException) {
201208
logger.debug("Rolled index already exists for rollup job [" + job.getConfig().getId() + "], updating metadata.");
202-
updateMapping(job, listener, persistentTasksService, client, logger, request.masterNodeTimeout());
209+
updateMapping(projectId, job, listener, persistentTasksService, client, logger, request.masterNodeTimeout());
203210
} else {
204211
String msg = "Could not create index for rollup job [" + job.getConfig().getId() + "]";
205212
logger.error(msg);
@@ -245,6 +252,7 @@ static XContentBuilder createMappings(RollupJobConfig config) throws IOException
245252

246253
@SuppressWarnings("unchecked")
247254
static void updateMapping(
255+
ProjectId projectId,
248256
RollupJob job,
249257
ActionListener<AcknowledgedResponse> listener,
250258
PersistentTasksService persistentTasksService,
@@ -301,7 +309,10 @@ static void updateMapping(
301309
client.execute(
302310
TransportPutMappingAction.TYPE,
303311
request,
304-
ActionListener.wrap(putMappingResponse -> startPersistentTask(job, listener, persistentTasksService), listener::onFailure)
312+
ActionListener.wrap(
313+
putMappingResponse -> startPersistentTask(projectId, job, listener, persistentTasksService),
314+
listener::onFailure
315+
)
305316
);
306317
};
307318

@@ -314,17 +325,19 @@ static void updateMapping(
314325
}
315326

316327
static void startPersistentTask(
328+
ProjectId projectId,
317329
RollupJob job,
318330
ActionListener<AcknowledgedResponse> listener,
319331
PersistentTasksService persistentTasksService
320332
) {
321333
assertNoAuthorizationHeader(job.getHeaders());
322-
persistentTasksService.sendStartRequest(
334+
persistentTasksService.sendProjectStartRequest(
335+
projectId,
323336
job.getConfig().getId(),
324337
RollupField.TASK_NAME,
325338
job,
326339
TimeValue.THIRTY_SECONDS /* TODO should this be configurable? longer by default? infinite? */,
327-
ActionListener.wrap(rollupConfigPersistentTask -> waitForRollupStarted(job, listener, persistentTasksService), e -> {
340+
ActionListener.wrap(rollupConfigPersistentTask -> waitForRollupStarted(projectId, job, listener, persistentTasksService), e -> {
328341
if (e instanceof ResourceAlreadyExistsException) {
329342
e = new ElasticsearchStatusException(
330343
"Cannot create job [" + job.getConfig().getId() + "] because it has already been created (task exists)",
@@ -338,11 +351,13 @@ static void startPersistentTask(
338351
}
339352

340353
private static void waitForRollupStarted(
354+
ProjectId projectId,
341355
RollupJob job,
342356
ActionListener<AcknowledgedResponse> listener,
343357
PersistentTasksService persistentTasksService
344358
) {
345359
persistentTasksService.waitForPersistentTaskCondition(
360+
projectId,
346361
job.getConfig().getId(),
347362
Objects::nonNull,
348363
job.getConfig().getTimeout(),
@@ -369,9 +384,9 @@ public void onTimeout(TimeValue timeout) {
369384
);
370385
}
371386

372-
static boolean hasRollupIndices(Metadata metadata) throws IOException {
387+
static boolean hasRollupIndices(ProjectMetadata project) throws IOException {
373388
// Sniffing logic instead of invoking sourceAsMap(), which would materialize the entire mapping as map of maps.
374-
for (var imd : metadata.getProject()) {
389+
for (var imd : project) {
375390
if (imd.mapping() == null) {
376391
continue;
377392
}

‎x-pack/plugin/rollup/src/main/java/org/elasticsearch/xpack/rollup/action/TransportRollupSearchAction.java‎

Lines changed: 7 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -20,6 +20,7 @@
2020
import org.elasticsearch.cluster.metadata.IndexMetadata;
2121
import org.elasticsearch.cluster.metadata.IndexNameExpressionResolver;
2222
import org.elasticsearch.cluster.metadata.Metadata;
23+
import org.elasticsearch.cluster.project.ProjectResolver;
2324
import org.elasticsearch.cluster.service.ClusterService;
2425
import org.elasticsearch.common.Strings;
2526
import org.elasticsearch.common.io.stream.BytesStreamOutput;
@@ -88,6 +89,7 @@ public class TransportRollupSearchAction extends TransportAction<SearchRequest,
8889
private final ScriptService scriptService;
8990
private final ClusterService clusterService;
9091
private final IndexNameExpressionResolver resolver;
92+
private final ProjectResolver projectResolver;
9193
private static final Logger logger = LogManager.getLogger(RollupSearchAction.class);
9294

9395
@Inject
@@ -99,7 +101,8 @@ public TransportRollupSearchAction(
99101
BigArrays bigArrays,
100102
ScriptService scriptService,
101103
ClusterService clusterService,
102-
IndexNameExpressionResolver resolver
104+
IndexNameExpressionResolver resolver,
105+
ProjectResolver projectResolver
103106
) {
104107
super(RollupSearchAction.NAME, actionFilters, transportService.getTaskManager(), EsExecutors.DIRECT_EXECUTOR_SERVICE);
105108
this.client = client;
@@ -108,6 +111,7 @@ public TransportRollupSearchAction(
108111
this.scriptService = scriptService;
109112
this.clusterService = clusterService;
110113
this.resolver = resolver;
114+
this.projectResolver = projectResolver;
111115

112116
transportService.registerRequestHandler(
113117
actionName,
@@ -123,7 +127,8 @@ public TransportRollupSearchAction(
123127
protected void doExecute(Task task, SearchRequest request, ActionListener<SearchResponse> listener) {
124128
DEPRECATION_LOGGER.warn(DeprecationCategory.API, DEPRECATION_KEY, DEPRECATION_MESSAGE);
125129
String[] indices = resolver.concreteIndexNames(clusterService.state(), request);
126-
RollupSearchContext rollupSearchContext = separateIndices(indices, clusterService.state().getMetadata().getProject().indices());
130+
final var project = projectResolver.getProjectMetadata(clusterService.state());
131+
RollupSearchContext rollupSearchContext = separateIndices(indices, project.indices());
127132

128133
MultiSearchRequest msearch = createMSearchRequest(request, registry, rollupSearchContext);
129134

0 commit comments

Comments
 (0)