Skip to content

Commit fcb70d8

Browse files
prwhelanprwhelan
committed
[Transform] Delete Alias Write Index
When the Transform is configured to write to an alias, specifying `DELETE _transform/<id>?delete_dest_index` will follow the alias to the concrete destination index. Fix #121913
1 parent 5302589 commit fcb70d8

File tree

5 files changed

+231
-34
lines changed

5 files changed

+231
-34
lines changed

‎docs/reference/transform/apis/delete-transform.asciidoc‎

Lines changed: 4 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -23,7 +23,7 @@ Deletes an existing {transform}.
2323
[[delete-transform-prereqs]]
2424
== {api-prereq-title}
2525

26-
* Requires the `manage_transform` cluster privilege. This privilege is included
26+
* Requires the `manage_transform` cluster privilege. This privilege is included
2727
in the `transform_admin` built-in role.
2828
* Before you can delete the {transform}, you must stop it.
2929

@@ -44,8 +44,9 @@ current state. The default value is `false`, meaning that the {transform} must b
4444

4545
`delete_dest_index`::
4646
(Optional, Boolean) When `true`, the destination index is deleted together with
47-
the {transform}. The default value is `false`, meaning that the destination
48-
index will not be deleted.
47+
the {transform}. If the destination index is an alias, then the alias's backing
48+
write index will be deleted. The default value is `false`, meaning that the
49+
destination index will not be deleted.
4950

5051
`timeout`::
5152
(Optional, time)

‎x-pack/plugin/transform/qa/single-node-tests/src/javaRestTest/java/org/elasticsearch/xpack/transform/integration/TransformDeleteIT.java‎

Lines changed: 86 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -110,7 +110,7 @@ public void testDeleteWithParamDeletesAutoCreatedDestinationIndex() throws Excep
110110

111111
deleteTransform(transformId, false, true);
112112
assertFalse(indexExists(transformDest));
113-
assertFalse(aliasExists(transformDest));
113+
assertFalse(aliasExists(transformDestAlias));
114114
}
115115

116116
public void testDeleteWithParamDeletesManuallyCreatedDestinationIndex() throws Exception {
@@ -139,7 +139,7 @@ public void testDeleteWithParamDeletesManuallyCreatedDestinationIndex() throws E
139139
assertFalse(aliasExists(transformDestAlias));
140140
}
141141

142-
public void testDeleteWithParamDoesNotDeleteManuallySetUpAlias() throws Exception {
142+
public void testDeleteWithManuallyCreatedIndexAndManuallyCreatedAlias() throws Exception {
143143
String transformId = "transform-4";
144144
String transformDest = transformId + "_idx";
145145
String transformDestAlias = transformId + "_alias";
@@ -158,31 +158,106 @@ public void testDeleteWithParamDoesNotDeleteManuallySetUpAlias() throws Exceptio
158158
assertTrue(indexExists(transformDest));
159159
assertTrue(aliasExists(transformDestAlias));
160160

161+
deleteTransform(transformId, false, true);
162+
assertFalse(indexExists(transformDest));
163+
assertFalse(aliasExists(transformDestAlias));
164+
}
165+
166+
public void testDeleteDestinationIndexIsNoOpWhenNoDestinationIndexExists() throws Exception {
167+
String transformId = "transform-5";
168+
String transformDest = transformId + "_idx";
169+
String transformDestAlias = transformId + "_alias";
170+
setupDataAccessRole(DATA_ACCESS_ROLE, REVIEWS_INDEX_NAME, transformDest, transformDestAlias);
171+
172+
createTransform(transformId, transformDest, transformDestAlias);
173+
assertFalse(indexExists(transformDest));
174+
assertFalse(aliasExists(transformDestAlias));
175+
176+
deleteTransform(transformId, false, true);
177+
assertFalse(indexExists(transformDest));
178+
assertFalse(aliasExists(transformDestAlias));
179+
}
180+
181+
public void testDeleteWithAliasPointingToManyIndices() throws Exception {
182+
var transformId = "transform-6";
183+
var transformDest = transformId + "_idx";
184+
var otherIndex = "some-other-index-6";
185+
String transformDestAlias = transformId + "_alias";
186+
setupDataAccessRole(DATA_ACCESS_ROLE, REVIEWS_INDEX_NAME, transformDest, otherIndex, transformDestAlias);
187+
188+
createIndex(transformDest, null, null, "\"" + transformDestAlias + "\": { \"is_write_index\": true }");
189+
createIndex(otherIndex, null, null, "\"" + transformDestAlias + "\": {}");
190+
191+
assertTrue(indexExists(transformDest));
192+
assertTrue(indexExists(otherIndex));
193+
assertTrue(aliasExists(transformDestAlias));
194+
195+
createTransform(transformId, transformDestAlias, null);
196+
197+
startTransform(transformId);
198+
waitForTransformCheckpoint(transformId, 1);
199+
200+
stopTransform(transformId, false);
201+
202+
assertTrue(indexExists(transformDest));
203+
assertTrue(indexExists(otherIndex));
204+
assertTrue(aliasExists(transformDestAlias));
205+
206+
deleteTransform(transformId, false, true);
207+
208+
assertFalse(indexExists(transformDest));
209+
assertTrue(indexExists(otherIndex));
210+
assertTrue(aliasExists(transformDestAlias));
211+
}
212+
213+
public void testDeleteWithNoWriteIndexThrowsException() throws Exception {
214+
var transformId = "transform-7";
215+
var transformDest = transformId + "_idx";
216+
var otherIndex = "some-other-index-7";
217+
String transformDestAlias = transformId + "_alias";
218+
setupDataAccessRole(DATA_ACCESS_ROLE, REVIEWS_INDEX_NAME, transformDest, otherIndex, transformDestAlias);
219+
220+
createIndex(transformDest, null, null, "\"" + transformDestAlias + "\": {}");
221+
222+
assertTrue(indexExists(transformDest));
223+
assertTrue(aliasExists(transformDestAlias));
224+
225+
createTransform(transformId, transformDestAlias, null);
226+
227+
createIndex(otherIndex, null, null, "\"" + transformDestAlias + "\": {}");
228+
assertTrue(indexExists(otherIndex));
229+
161230
ResponseException e = expectThrows(ResponseException.class, () -> deleteTransform(transformId, false, true));
162231
assertThat(
163232
e.getMessage(),
164233
containsString(
165234
Strings.format(
166-
"The provided expression [%s] matches an alias, specify the corresponding concrete indices instead.",
235+
"Cannot disambiguate destination index alias [%s]. Alias points to many indices with no clear write alias."
236+
+ " Retry with delete_dest_index=false and manually clean up destination index.",
167237
transformDestAlias
168238
)
169239
)
170240
);
171241
}
172242

173-
public void testDeleteDestinationIndexIsNoOpWhenNoDestinationIndexExists() throws Exception {
174-
String transformId = "transform-5";
175-
String transformDest = transformId + "_idx";
176-
String transformDestAlias = transformId + "_alias";
177-
setupDataAccessRole(DATA_ACCESS_ROLE, REVIEWS_INDEX_NAME, transformDest, transformDestAlias);
243+
public void testDeleteWithAlreadyDeletedIndex() throws Exception {
244+
var transformId = "transform-8";
245+
var transformDest = transformId + "_idx";
246+
setupDataAccessRole(DATA_ACCESS_ROLE, REVIEWS_INDEX_NAME, transformDest);
247+
248+
createIndex(transformDest);
249+
250+
assertTrue(indexExists(transformDest));
251+
252+
createTransform(transformId, transformDest, null);
253+
254+
deleteIndex(transformDest);
178255

179-
createTransform(transformId, transformDest, transformDestAlias);
180256
assertFalse(indexExists(transformDest));
181-
assertFalse(aliasExists(transformDestAlias));
182257

183258
deleteTransform(transformId, false, true);
259+
184260
assertFalse(indexExists(transformDest));
185-
assertFalse(aliasExists(transformDestAlias));
186261
}
187262

188263
private void createTransform(String transformId, String destIndex, String destAlias) throws IOException {

‎x-pack/plugin/transform/qa/single-node-tests/src/javaRestTest/java/org/elasticsearch/xpack/transform/integration/TransformResetIT.java‎

Lines changed: 29 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -138,6 +138,35 @@ public void testResetDeletesDestinationIndex() throws Exception {
138138
assertFalse(indexExists(transformDest));
139139
}
140140

141+
public void testResetDeletesDestinationAlias() throws Exception {
142+
String transformId = "transform-3";
143+
String transformDest = transformId + "_idx";
144+
setupDataAccessRole(DATA_ACCESS_ROLE, REVIEWS_INDEX_NAME, transformDest);
145+
146+
final Request createTransformRequest = createRequestWithAuth(
147+
"PUT",
148+
getTransformEndpoint() + transformId,
149+
BASIC_AUTH_VALUE_TRANSFORM_ADMIN_1
150+
);
151+
String config = createConfig(transformDest);
152+
createTransformRequest.setJsonEntity(config);
153+
Map<String, Object> createTransformResponse = entityAsMap(client().performRequest(createTransformRequest));
154+
assertThat(createTransformResponse.get("acknowledged"), equalTo(Boolean.TRUE));
155+
156+
assertFalse(indexExists(transformDest));
157+
158+
startTransform(transformId);
159+
waitForTransformCheckpoint(transformId, 1);
160+
stopTransform(transformId, false);
161+
assertTrue(indexExists(transformDest));
162+
163+
var transformDestAlias = createAlias(transformDest);
164+
updateTransformIndex(transformId, transformDestAlias);
165+
166+
resetTransform(transformId, false);
167+
assertFalse(indexExists(transformDest));
168+
}
169+
141170
private static String createConfig(String transformDestIndex) {
142171
boolean isContinuous = randomBoolean();
143172
return Strings.format("""

‎x-pack/plugin/transform/qa/single-node-tests/src/javaRestTest/java/org/elasticsearch/xpack/transform/integration/TransformRestTestCase.java‎

Lines changed: 32 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -412,7 +412,7 @@ protected void updateTransform(String transformId, String update, boolean deferV
412412
}
413413
updateTransformRequest.setJsonEntity(update);
414414

415-
client().performRequest(updateTransformRequest);
415+
assertOKAndConsume(client().performRequest(updateTransformRequest));
416416
}
417417

418418
protected void startTransform(String transformId) throws IOException {
@@ -684,4 +684,35 @@ protected void assertOneCount(String query, String field, int expected) throws I
684684
int actual = (Integer) ((List<?>) XContentMapValues.extractValue(field, searchResult)).get(0);
685685
assertEquals(expected, actual);
686686
}
687+
688+
protected String createAlias(String index) throws IOException {
689+
var alias = index + "_alias";
690+
var createNewIndexRequest = new Request("POST", Strings.format("/_create_from/%s/%s-new", index, index));
691+
assertOKAndConsume(adminClient().performRequest(createNewIndexRequest));
692+
var createAlias = new Request("POST", "/_aliases");
693+
createAlias.setJsonEntity(Strings.format("""
694+
{
695+
"actions": [
696+
{
697+
"add": {
698+
"index": "%s",
699+
"alias": "%s"
700+
}
701+
}
702+
]
703+
}
704+
""", index, alias));
705+
assertOKAndConsume(adminClient().performRequest(createAlias));
706+
return alias;
707+
}
708+
709+
protected void updateTransformIndex(String transformId, String index) throws IOException {
710+
updateTransform(transformId, Strings.format("""
711+
{
712+
"dest": {
713+
"index": "%s"
714+
}
715+
}
716+
""", index), false);
717+
}
687718
}

‎x-pack/plugin/transform/src/main/java/org/elasticsearch/xpack/transform/action/TransportDeleteTransformAction.java‎

Lines changed: 80 additions & 19 deletions
Original file line numberDiff line numberDiff line change
@@ -10,9 +10,13 @@
1010
import org.apache.logging.log4j.Logger;
1111
import org.elasticsearch.ElasticsearchStatusException;
1212
import org.elasticsearch.action.ActionListener;
13+
import org.elasticsearch.action.admin.indices.alias.get.GetAliasesAction;
14+
import org.elasticsearch.action.admin.indices.alias.get.GetAliasesRequest;
15+
import org.elasticsearch.action.admin.indices.alias.get.GetAliasesResponse;
1316
import org.elasticsearch.action.admin.indices.delete.DeleteIndexRequest;
1417
import org.elasticsearch.action.admin.indices.delete.TransportDeleteIndexAction;
1518
import org.elasticsearch.action.support.ActionFilters;
19+
import org.elasticsearch.action.support.SubscribableListener;
1620
import org.elasticsearch.action.support.master.AcknowledgedResponse;
1721
import org.elasticsearch.action.support.master.AcknowledgedTransportMasterNodeAction;
1822
import org.elasticsearch.client.internal.Client;
@@ -27,6 +31,7 @@
2731
import org.elasticsearch.index.IndexNotFoundException;
2832
import org.elasticsearch.injection.guice.Inject;
2933
import org.elasticsearch.rest.RestStatus;
34+
import org.elasticsearch.rest.action.admin.indices.AliasesNotFoundException;
3035
import org.elasticsearch.tasks.Task;
3136
import org.elasticsearch.tasks.TaskId;
3237
import org.elasticsearch.threadpool.ThreadPool;
@@ -42,6 +47,8 @@
4247
import org.elasticsearch.xpack.transform.persistence.TransformConfigManager;
4348
import org.elasticsearch.xpack.transform.transforms.TransformTask;
4449

50+
import java.util.Objects;
51+
4552
import static org.elasticsearch.xpack.core.ClientHelper.TRANSFORM_ORIGIN;
4653
import static org.elasticsearch.xpack.core.ClientHelper.executeAsyncWithOrigin;
4754
import static org.elasticsearch.xpack.core.ClientHelper.executeWithHeadersAsync;
@@ -146,20 +153,31 @@ private void deleteDestinationIndex(
146153
TimeValue timeout,
147154
ActionListener<AcknowledgedResponse> listener
148155
) {
149-
// <3> Check if the error is "index not found" error. If so, just move on. The index is already deleted.
150-
ActionListener<AcknowledgedResponse> deleteDestIndexListener = ActionListener.wrap(listener::onResponse, e -> {
151-
if (e instanceof IndexNotFoundException) {
152-
listener.onResponse(AcknowledgedResponse.TRUE);
153-
} else {
154-
listener.onFailure(e);
155-
}
156-
});
156+
getTransformConfig(transformId).<AcknowledgedResponse>andThen((l, r) -> deleteDestinationIndex(r.v1(), parentTaskId, timeout, l))
157+
.addListener(listener.delegateResponse((l, e) -> {
158+
if (e instanceof IndexNotFoundException) {
159+
l.onResponse(AcknowledgedResponse.TRUE);
160+
} else {
161+
l.onFailure(e);
162+
}
163+
}));
164+
}
157165

158-
// <2> Delete destination index
159-
ActionListener<Tuple<TransformConfig, SeqNoPrimaryTermAndIndex>> getTransformConfigurationListener = ActionListener.wrap(
160-
transformConfigAndVersion -> {
161-
TransformConfig config = transformConfigAndVersion.v1();
162-
String destIndex = config.getDestination().getIndex();
166+
private SubscribableListener<Tuple<TransformConfig, SeqNoPrimaryTermAndIndex>> getTransformConfig(String transformId) {
167+
return SubscribableListener.newForked(l -> transformConfigManager.getTransformConfigurationForUpdate(transformId, l));
168+
}
169+
170+
/**
171+
* Delete the destination index. If the Transform is configured to write to an alias, then follow that alias to the concrete index.
172+
*/
173+
private void deleteDestinationIndex(
174+
TransformConfig config,
175+
TaskId parentTaskId,
176+
TimeValue timeout,
177+
ActionListener<AcknowledgedResponse> listener
178+
) {
179+
SubscribableListener.<String>newForked(l -> resolveDestinationIndex(config, parentTaskId, timeout, l))
180+
.<AcknowledgedResponse>andThen((l, destIndex) -> {
163181
DeleteIndexRequest deleteDestIndexRequest = new DeleteIndexRequest(destIndex);
164182
deleteDestIndexRequest.ackTimeout(timeout);
165183
deleteDestIndexRequest.setParentTask(parentTaskId);
@@ -169,14 +187,57 @@ private void deleteDestinationIndex(
169187
client,
170188
TransportDeleteIndexAction.TYPE,
171189
deleteDestIndexRequest,
172-
deleteDestIndexListener
190+
l
173191
);
174-
},
175-
listener::onFailure
176-
);
192+
})
193+
.addListener(listener);
194+
}
195+
196+
private void resolveDestinationIndex(TransformConfig config, TaskId parentTaskId, TimeValue timeout, ActionListener<String> listener) {
197+
var destIndex = config.getDestination().getIndex();
198+
var responseListener = ActionListener.<GetAliasesResponse>wrap(r -> findDestinationIndexInAliases(r, destIndex, listener), e -> {
199+
if (e instanceof AliasesNotFoundException) {
200+
// no alias == the destIndex is our concrete index
201+
listener.onResponse(destIndex);
202+
} else {
203+
listener.onFailure(e);
204+
}
205+
});
206+
207+
GetAliasesRequest request = new GetAliasesRequest(timeout, config.getDestination().getIndex());
208+
request.setParentTask(parentTaskId);
209+
executeWithHeadersAsync(config.getHeaders(), TRANSFORM_ORIGIN, client, GetAliasesAction.INSTANCE, request, responseListener);
210+
}
177211

178-
// <1> Fetch transform configuration
179-
transformConfigManager.getTransformConfigurationForUpdate(transformId, getTransformConfigurationListener);
212+
private void findDestinationIndexInAliases(GetAliasesResponse aliases, String destIndex, ActionListener<String> listener) {
213+
var indexToAliases = aliases.getAliases();
214+
if (indexToAliases.isEmpty()) {
215+
// if the alias list is empty, that means the index is a concrete index
216+
listener.onResponse(destIndex);
217+
} else if (indexToAliases.size() == 1) {
218+
// if there is one value, the alias will treat it as the write index, so it's our destination index
219+
listener.onResponse(indexToAliases.keySet().iterator().next());
220+
} else {
221+
// if there is more than one index, there may be more than one alias for each index
222+
// we have to search for the alias that matches our destination index name AND is declared the write index for that alias
223+
indexToAliases.entrySet().stream().map(entry -> {
224+
if (entry.getValue().stream().anyMatch(md -> destIndex.equals(md.getAlias()) && Boolean.TRUE.equals(md.writeIndex()))) {
225+
return entry.getKey();
226+
} else {
227+
return null;
228+
}
229+
}).filter(Objects::nonNull).findFirst().ifPresentOrElse(listener::onResponse, () -> {
230+
listener.onFailure(
231+
new ElasticsearchStatusException(
232+
"Cannot disambiguate destination index alias ["
233+
+ destIndex
234+
+ "]. Alias points to many indices with no clear write alias. Retry with delete_dest_index=false and manually"
235+
+ " clean up destination index.",
236+
RestStatus.CONFLICT
237+
)
238+
);
239+
});
240+
}
180241
}
181242

182243
@Override

0 commit comments

Comments
 (0)