Skip to content

Commit 489e2d6

Browse files
prwhelanprzemekwitek
prwhelan
and
przemekwitek
authored
[8.x] [Transform] Delete Alias Write Index (elastic#122074) (elastic#122405)
* [Transform] Delete Alias Write Index (elastic#122074) 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 elastic#121913 Co-authored-by: Przemysław Witek <[email protected]> * Update for 8.x api --------- Co-authored-by: Przemysław Witek <[email protected]>
1 parent e58c38c commit 489e2d6

File tree

4 files changed

+175
-31
lines changed

4 files changed

+175
-31
lines changed

‎docs/changelog/122074.yaml‎

Lines changed: 8 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,8 @@
1+
pr: 122074
2+
summary: If the Transform is configured to write to an alias as its destination index,
3+
when the delete_dest_index parameter is set to true, then the Delete API will now
4+
delete the write index backing the alias
5+
area: Transform
6+
type: bug
7+
issues:
8+
- 121913

‎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/TransformRestTestCase.java‎

Lines changed: 1 addition & 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 {

‎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(destIndex);
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 static 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)