honeymoose
/
OpenSearch
mirror of https://github.com/honeymoose/OpenSearch.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

[7.x] Data stream admin actions are now index-level actions

This commit is contained in:
Dan Hermann 2020-07-10 14:36:18 -05:00 committed by GitHub
parent 7fa9cf601b
commit e01d73c737
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
15 changed files with 352 additions and 120 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
rest-api-spec/src/main/resources/rest-api-spec/api/indices.get_data_stream.json
View File

@ -20,8 +20,8 @@
],
"parts":{
"name":{
"type":"string",
"description":"The name or wildcard expression of the requested data streams"
"type":"list",
"description":"A comma-separated list of data streams to get; use `*` to get all data streams"
}
}
}

2
rest-api-spec/src/main/resources/rest-api-spec/test/indices.data_stream/10_basic.yml
View File

@ -157,7 +157,7 @@ setup:
catch: missing
- match: { status: 404 }
- match: { error.root_cause.0.type: "resource_not_found_exception" }
- match: { error.root_cause.0.type: "index_not_found_exception" }
- do:
indices.get_data_stream:

4
server/src/internalClusterTest/java/org/elasticsearch/action/bulk/BulkIntegrationIT.java
View File

@ -258,7 +258,7 @@ public class BulkIntegrationIT extends ESIntegTestCase {
bulkResponse = client().bulk(bulkRequest).actionGet();
assertThat("bulk failures: " + Strings.toString(bulkResponse), bulkResponse.hasFailures(), is(false));
GetDataStreamAction.Request getDataStreamRequest = new GetDataStreamAction.Request("*");
GetDataStreamAction.Request getDataStreamRequest = new GetDataStreamAction.Request(new String[]{"*"});
GetDataStreamAction.Response getDataStreamsResponse = client().admin().indices().getDataStreams(getDataStreamRequest).actionGet();
assertThat(getDataStreamsResponse.getDataStreams(), hasSize(4));
getDataStreamsResponse.getDataStreams().sort(Comparator.comparing(dataStreamInfo -> dataStreamInfo.getDataStream().getName()));
@ -294,7 +294,7 @@ public class BulkIntegrationIT extends ESIntegTestCase {
BulkResponse bulkResponse = client().bulk(bulkRequest).actionGet();
assertThat("bulk failures: " + Strings.toString(bulkResponse), bulkResponse.hasFailures(), is(false));
GetDataStreamAction.Request getDataStreamRequest = new GetDataStreamAction.Request("*");
GetDataStreamAction.Request getDataStreamRequest = new GetDataStreamAction.Request(new String[]{"*"});
GetDataStreamAction.Response getDataStreamsResponse = client().admin().indices().getDataStreams(getDataStreamRequest).actionGet();
assertThat(getDataStreamsResponse.getDataStreams(), hasSize(0));

8
server/src/internalClusterTest/java/org/elasticsearch/indices/DataStreamIT.java
View File

@ -119,7 +119,7 @@ public class DataStreamIT extends ESIntegTestCase {
createDataStreamRequest = new CreateDataStreamAction.Request("metrics-bar");
client().admin().indices().createDataStream(createDataStreamRequest).get();
GetDataStreamAction.Request getDataStreamRequest = new GetDataStreamAction.Request("*");
GetDataStreamAction.Request getDataStreamRequest = new GetDataStreamAction.Request(new String[]{"*"});
GetDataStreamAction.Response getDataStreamResponse = client().admin().indices().getDataStreams(getDataStreamRequest).actionGet();
getDataStreamResponse.getDataStreams().sort(Comparator.comparing(dataStreamInfo -> dataStreamInfo.getDataStream().getName()));
assertThat(getDataStreamResponse.getDataStreams().size(), equalTo(2));
@ -302,7 +302,7 @@ public class DataStreamIT extends ESIntegTestCase {
verifyDocs(dataStreamName, numDocs, 1, 1);
String backingIndex = DataStream.getDefaultBackingIndexName(dataStreamName, 1);
GetDataStreamAction.Request getDataStreamRequest = new GetDataStreamAction.Request("*");
GetDataStreamAction.Request getDataStreamRequest = new GetDataStreamAction.Request(new String[]{"*"});
GetDataStreamAction.Response getDataStreamResponse = client().admin().indices().getDataStreams(getDataStreamRequest).actionGet();
assertThat(getDataStreamResponse.getDataStreams().size(), equalTo(1));
assertThat(getDataStreamResponse.getDataStreams().get(0).getDataStream().getName(), equalTo(dataStreamName));
@ -530,7 +530,7 @@ public class DataStreamIT extends ESIntegTestCase {
CreateDataStreamAction.Request createDataStreamRequest = new CreateDataStreamAction.Request("logs-foobar");
client().admin().indices().createDataStream(createDataStreamRequest).get();
GetDataStreamAction.Request getDataStreamRequest = new GetDataStreamAction.Request("logs-foobar");
GetDataStreamAction.Request getDataStreamRequest = new GetDataStreamAction.Request(new String[]{"logs-foobar"});
GetDataStreamAction.Response getDataStreamResponse = client().admin().indices().getDataStreams(getDataStreamRequest).actionGet();
assertThat(getDataStreamResponse.getDataStreams().size(), equalTo(1));
assertThat(getDataStreamResponse.getDataStreams().get(0).getDataStream().getName(), equalTo("logs-foobar"));
@ -682,7 +682,7 @@ public class DataStreamIT extends ESIntegTestCase {
indexDocs("metrics-foo", "@timestamp", numDocsFoo);
GetDataStreamAction.Response response =
client().admin().indices().getDataStreams(new GetDataStreamAction.Request("metrics-foo")).actionGet();
client().admin().indices().getDataStreams(new GetDataStreamAction.Request(new String[]{"metrics-foo"})).actionGet();
assertThat(response.getDataStreams().size(), is(1));
GetDataStreamAction.Response.DataStreamInfo metricsFooDataStream = response.getDataStreams().get(0);
assertThat(metricsFooDataStream.getDataStream().getName(), is("metrics-foo"));

20
server/src/internalClusterTest/java/org/elasticsearch/snapshots/DataStreamsSnapshotsIT.java
View File

@ -117,7 +117,8 @@ public class DataStreamsSnapshotsIT extends AbstractSnapshotIntegTestCase {
assertEquals(1, hits.length);
assertEquals(DOCUMENT_SOURCE, hits[0].getSourceAsMap());
GetDataStreamAction.Response ds = client.admin().indices().getDataStreams(new GetDataStreamAction.Request("ds")).get();
GetDataStreamAction.Response ds = client.admin().indices().getDataStreams(
new GetDataStreamAction.Request(new String[]{"ds"})).get();
assertEquals(1, ds.getDataStreams().size());
assertEquals(1, ds.getDataStreams().get(0).getDataStream().getIndices().size());
assertEquals(DS_BACKING_INDEX_NAME, ds.getDataStreams().get(0).getDataStream().getIndices().get(0).getName());
@ -155,7 +156,8 @@ public class DataStreamsSnapshotsIT extends AbstractSnapshotIntegTestCase {
assertEquals(1, hits.length);
assertEquals(DOCUMENT_SOURCE, hits[0].getSourceAsMap());
GetDataStreamAction.Response ds = client.admin().indices().getDataStreams(new GetDataStreamAction.Request("ds")).get();
GetDataStreamAction.Response ds = client.admin().indices().getDataStreams(
new GetDataStreamAction.Request(new String[]{"ds"})).get();
assertEquals(1, ds.getDataStreams().size());
assertEquals(1, ds.getDataStreams().get(0).getDataStream().getIndices().size());
assertEquals(DS_BACKING_INDEX_NAME, ds.getDataStreams().get(0).getDataStream().getIndices().get(0).getName());
@ -188,7 +190,8 @@ public class DataStreamsSnapshotsIT extends AbstractSnapshotIntegTestCase {
.setRenameReplacement("ds2")
.get();
GetDataStreamAction.Response ds = client.admin().indices().getDataStreams(new GetDataStreamAction.Request("ds2")).get();
GetDataStreamAction.Response ds = client.admin().indices().getDataStreams(
new GetDataStreamAction.Request(new String[]{"ds2"})).get();
assertEquals(1, ds.getDataStreams().size());
assertEquals(1, ds.getDataStreams().get(0).getDataStream().getIndices().size());
assertEquals(DS2_BACKING_INDEX_NAME, ds.getDataStreams().get(0).getDataStream().getIndices().get(0).getName());
@ -227,7 +230,7 @@ public class DataStreamsSnapshotsIT extends AbstractSnapshotIntegTestCase {
assertThat(restoreSnapshotResponse.status(), is(RestStatus.OK));
GetDataStreamAction.Request getDSRequest = new GetDataStreamAction.Request("ds");
GetDataStreamAction.Request getDSRequest = new GetDataStreamAction.Request(new String[]{"ds"});
GetDataStreamAction.Response response = client.admin().indices().getDataStreams(getDSRequest).actionGet();
assertThat(response.getDataStreams().get(0).getDataStream().getIndices().get(0).getName(), is(DS_BACKING_INDEX_NAME));
}
@ -261,13 +264,13 @@ public class DataStreamsSnapshotsIT extends AbstractSnapshotIntegTestCase {
assertThat(restoreSnapshotResponse.status(), is(RestStatus.OK));
// assert "ds" was restored as "test-ds" and the backing index has a valid name
GetDataStreamAction.Request getRenamedDS = new GetDataStreamAction.Request("test-ds");
GetDataStreamAction.Request getRenamedDS = new GetDataStreamAction.Request(new String[]{"test-ds"});
GetDataStreamAction.Response response = client.admin().indices().getDataStreams(getRenamedDS).actionGet();
assertThat(response.getDataStreams().get(0).getDataStream().getIndices().get(0).getName(),
is(DataStream.getDefaultBackingIndexName("test-ds", 1L)));
// data stream "ds" should still exist in the system
GetDataStreamAction.Request getDSRequest = new GetDataStreamAction.Request("ds");
GetDataStreamAction.Request getDSRequest = new GetDataStreamAction.Request(new String[]{"ds"});
response = client.admin().indices().getDataStreams(getDSRequest).actionGet();
assertThat(response.getDataStreams().get(0).getDataStream().getIndices().get(0).getName(), is(DS_BACKING_INDEX_NAME));
}
@ -293,7 +296,8 @@ public class DataStreamsSnapshotsIT extends AbstractSnapshotIntegTestCase {
assertEquals(RestStatus.OK, restoreSnapshotResponse.status());
GetDataStreamAction.Response ds = client.admin().indices().getDataStreams(new GetDataStreamAction.Request("ds2")).get();
GetDataStreamAction.Response ds = client.admin().indices().getDataStreams(
new GetDataStreamAction.Request(new String[]{"ds2"})).get();
assertEquals(1, ds.getDataStreams().size());
assertEquals(1, ds.getDataStreams().get(0).getDataStream().getIndices().size());
assertEquals(DS2_BACKING_INDEX_NAME, ds.getDataStreams().get(0).getDataStream().getIndices().get(0).getName());
@ -340,7 +344,7 @@ public class DataStreamsSnapshotsIT extends AbstractSnapshotIntegTestCase {
assertEquals(RestStatus.OK, restoreSnapshotResponse.status());
GetDataStreamAction.Request getRequest = new GetDataStreamAction.Request("ds");
GetDataStreamAction.Request getRequest = new GetDataStreamAction.Request(new String[]{"ds"});
expectThrows(ResourceNotFoundException.class, () -> client.admin().indices().getDataStreams(getRequest).actionGet());
}

14
server/src/main/java/org/elasticsearch/action/admin/indices/datastream/CreateDataStreamAction.java
View File

@ -21,8 +21,10 @@ package org.elasticsearch.action.admin.indices.datastream;
import org.elasticsearch.action.ActionListener;
import org.elasticsearch.action.ActionRequestValidationException;
import org.elasticsearch.action.ActionType;
import org.elasticsearch.action.IndicesRequest;
import org.elasticsearch.action.ValidateActions;
import org.elasticsearch.action.support.ActionFilters;
import org.elasticsearch.action.support.IndicesOptions;
import org.elasticsearch.action.support.master.AcknowledgedRequest;
import org.elasticsearch.action.support.master.AcknowledgedResponse;
import org.elasticsearch.action.support.master.TransportMasterNodeAction;
@ -52,7 +54,7 @@ public class CreateDataStreamAction extends ActionType<AcknowledgedResponse> {
super(NAME, AcknowledgedResponse::new);
}
public static class Request extends AcknowledgedRequest<Request> {
public static class Request extends AcknowledgedRequest<Request> implements IndicesRequest {
private final String name;
@ -92,6 +94,16 @@ public class CreateDataStreamAction extends ActionType<AcknowledgedResponse> {
public int hashCode() {
return Objects.hash(name);
}
@Override
public String[] indices() {
return new String[]{name};
}
@Override
public IndicesOptions indicesOptions() {
return IndicesOptions.strictSingleIndexNoExpandForbidClosed();
}
}
public static class TransportAction extends TransportMasterNodeAction<Request, AcknowledgedResponse> {

40
server/src/main/java/org/elasticsearch/action/admin/indices/datastream/DeleteDataStreamAction.java
View File

@ -20,11 +20,12 @@ package org.elasticsearch.action.admin.indices.datastream;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.elasticsearch.ResourceNotFoundException;
import org.elasticsearch.action.ActionListener;
import org.elasticsearch.action.ActionRequestValidationException;
import org.elasticsearch.action.ActionType;
import org.elasticsearch.action.IndicesRequest;
import org.elasticsearch.action.support.ActionFilters;
import org.elasticsearch.action.support.IndicesOptions;
import org.elasticsearch.action.support.master.AcknowledgedResponse;
import org.elasticsearch.action.support.master.MasterNodeRequest;
import org.elasticsearch.action.support.master.TransportMasterNodeAction;
@ -70,9 +71,9 @@ public class DeleteDataStreamAction extends ActionType<AcknowledgedResponse> {
super(NAME, AcknowledgedResponse::new);
}
public static class Request extends MasterNodeRequest<Request> {
public static class Request extends MasterNodeRequest<Request> implements IndicesRequest.Replaceable {
private final String[] names;
private String[] names;
public Request(String[] names) {
this.names = Objects.requireNonNull(names);
@ -110,6 +111,29 @@ public class DeleteDataStreamAction extends ActionType<AcknowledgedResponse> {
public int hashCode() {
return Arrays.hashCode(names);
}
@Override
public String[] indices() {
return names;
}
@Override
public IndicesOptions indicesOptions() {
// this doesn't really matter since data stream name resolution isn't affected by IndicesOptions and
// a data stream's backing indices are retrieved from its metadata
return IndicesOptions.fromOptions(false, true, true, true, false, false, true, false);
}
@Override
public boolean includeDataStreams() {
return true;
}
@Override
public IndicesRequest indices(String... indices) {
this.names = indices;
return this;
}
}
public static class TransportAction extends TransportMasterNodeAction<Request, AcknowledgedResponse> {
@ -175,16 +199,6 @@ public class DeleteDataStreamAction extends ActionType<AcknowledgedResponse> {
snapshottingDataStreams.addAll(SnapshotsService.snapshottingDataStreams(currentState, dataStreams));
}
if (dataStreams.isEmpty()) {
// if only a match-all pattern was specified and no data streams were found because none exist, do not
// fail with data stream missing exception
if (request.names.length == 1 && Regex.isMatchAllPattern(request.names[0])) {
return currentState;
}
throw new ResourceNotFoundException("data_streams matching [" + Strings.arrayToCommaDelimitedString(request.names) +
"] not found");
}
if (snapshottingDataStreams.isEmpty() == false) {
throw new SnapshotInProgressException("Cannot delete data streams that are being snapshotted: " + snapshottingDataStreams +
". Try again after snapshot finishes or cancel the currently running snapshot.");

71
server/src/main/java/org/elasticsearch/action/admin/indices/datastream/GetDataStreamAction.java
View File

@ -20,12 +20,13 @@ package org.elasticsearch.action.admin.indices.datastream;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.elasticsearch.ResourceNotFoundException;
import org.elasticsearch.action.ActionListener;
import org.elasticsearch.action.ActionRequestValidationException;
import org.elasticsearch.action.ActionResponse;
import org.elasticsearch.action.ActionType;
import org.elasticsearch.action.IndicesRequest;
import org.elasticsearch.action.support.ActionFilters;
import org.elasticsearch.action.support.IndicesOptions;
import org.elasticsearch.action.support.master.MasterNodeReadRequest;
import org.elasticsearch.action.support.master.TransportMasterNodeReadAction;
import org.elasticsearch.cluster.AbstractDiffable;
@ -43,7 +44,6 @@ import org.elasticsearch.common.ParseField;
import org.elasticsearch.common.inject.Inject;
import org.elasticsearch.common.io.stream.StreamInput;
import org.elasticsearch.common.io.stream.StreamOutput;
import org.elasticsearch.common.regex.Regex;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.xcontent.ToXContentObject;
import org.elasticsearch.common.xcontent.XContentBuilder;
@ -53,10 +53,12 @@ import org.elasticsearch.transport.TransportService;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Comparator;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.stream.Collectors;
public class GetDataStreamAction extends ActionType<GetDataStreamAction.Response> {
@ -67,12 +69,12 @@ public class GetDataStreamAction extends ActionType<GetDataStreamAction.Response
super(NAME, Response::new);
}
public static class Request extends MasterNodeReadRequest<Request> {
public static class Request extends MasterNodeReadRequest<Request> implements IndicesRequest.Replaceable {
private final String name;
private String[] names;
public Request(String name) {
this.name = name;
public Request(String[] names) {
this.names = names;
}
@Override
@ -82,13 +84,13 @@ public class GetDataStreamAction extends ActionType<GetDataStreamAction.Response
public Request(StreamInput in) throws IOException {
super(in);
this.name = in.readOptionalString();
this.names = in.readOptionalStringArray();
}
@Override
public void writeTo(StreamOutput out) throws IOException {
super.writeTo(out);
out.writeOptionalString(name);
out.writeOptionalStringArray(names);
}
@Override
@ -96,12 +98,35 @@ public class GetDataStreamAction extends ActionType<GetDataStreamAction.Response
if (this == o) return true;
if (o == null || getClass() != o.getClass()) return false;
Request request = (Request) o;
return Objects.equals(name, request.name);
return Arrays.equals(names, request.names);
}
@Override
public int hashCode() {
return Objects.hash(name);
return Arrays.hashCode(names);
}
@Override
public String[] indices() {
return names;
}
@Override
public IndicesOptions indicesOptions() {
// this doesn't really matter since data stream name resolution isn't affected by IndicesOptions and
// a data stream's backing indices are retrieved from its metadata
return IndicesOptions.fromOptions(false, true, true, true, false, false, true, false);
}
@Override
public boolean includeDataStreams() {
return true;
}
@Override
public IndicesRequest indices(String... indices) {
this.names = indices;
return this;
}
}
@ -260,7 +285,7 @@ public class GetDataStreamAction extends ActionType<GetDataStreamAction.Response
@Override
protected void masterOperation(Request request, ClusterState state,
ActionListener<Response> listener) throws Exception {
List<DataStream> dataStreams = getDataStreams(state, request);
List<DataStream> dataStreams = getDataStreams(state, indexNameExpressionResolver, request);
List<Response.DataStreamInfo> dataStreamInfos = new ArrayList<>(dataStreams.size());
for (DataStream dataStream : dataStreams) {
String indexTemplate = MetadataIndexTemplateService.findV2Template(state.metadata(), dataStream.getName(), false);
@ -279,26 +304,14 @@ public class GetDataStreamAction extends ActionType<GetDataStreamAction.Response
listener.onResponse(new Response(dataStreamInfos));
}
static List<DataStream> getDataStreams(ClusterState clusterState, Request request) {
static List<DataStream> getDataStreams(ClusterState clusterState, IndexNameExpressionResolver iner, Request request) {
List<String> results = iner.dataStreamNames(clusterState, request.indicesOptions(), request.names);
Map<String, DataStream> dataStreams = clusterState.metadata().dataStreams();
// return all data streams if no name was specified
final String requestedName = request.name == null ? "*" : request.name;
final List<DataStream> results = new ArrayList<>();
if (Regex.isSimpleMatchPattern(requestedName)) {
for (Map.Entry<String, DataStream> entry : dataStreams.entrySet()) {
if (Regex.simpleMatch(requestedName, entry.getKey())) {
results.add(entry.getValue());
}
}
} else if (dataStreams.containsKey(request.name)) {
results.add(dataStreams.get(request.name));
} else {
throw new ResourceNotFoundException("data_stream matching [" + request.name + "] not found");
}
results.sort(Comparator.comparing(DataStream::getName));
return results;
return results.stream()
.map(dataStreams::get)
.sorted(Comparator.comparing(DataStream::getName))
.collect(Collectors.toList());
}
@Override

4
server/src/main/java/org/elasticsearch/rest/action/admin/indices/RestGetDataStreamsAction.java
View File

@ -20,6 +20,7 @@ package org.elasticsearch.rest.action.admin.indices;
import org.elasticsearch.action.admin.indices.datastream.GetDataStreamAction;
import org.elasticsearch.client.node.NodeClient;
import org.elasticsearch.common.Strings;
import org.elasticsearch.rest.BaseRestHandler;
import org.elasticsearch.rest.RestRequest;
import org.elasticsearch.rest.action.RestToXContentListener;
@ -44,7 +45,8 @@ public class RestGetDataStreamsAction extends BaseRestHandler {
@Override
protected RestChannelConsumer prepareRequest(RestRequest request, NodeClient client) throws IOException {
GetDataStreamAction.Request getDataStreamsRequest = new GetDataStreamAction.Request(request.param("name"));
GetDataStreamAction.Request getDataStreamsRequest = new GetDataStreamAction.Request(
Strings.splitStringByCommaToArray(request.param("name")));
return channel -> client.admin().indices().getDataStreams(getDataStreamsRequest, new RestToXContentListener<>(channel));
}
}

18
server/src/test/java/org/elasticsearch/action/admin/indices/datastream/DeleteDataStreamRequestTests.java
View File

@ -18,7 +18,6 @@
*/
package org.elasticsearch.action.admin.indices.datastream;
import org.elasticsearch.ResourceNotFoundException;
import org.elasticsearch.Version;
import org.elasticsearch.action.ActionRequestValidationException;
import org.elasticsearch.action.admin.indices.datastream.DeleteDataStreamAction.Request;
@ -29,6 +28,7 @@ import org.elasticsearch.cluster.metadata.DataStream;
import org.elasticsearch.cluster.metadata.IndexMetadata;
import org.elasticsearch.cluster.metadata.Metadata;
import org.elasticsearch.cluster.metadata.MetadataDeleteIndexService;
import org.elasticsearch.common.Strings;
import org.elasticsearch.common.collect.Tuple;
import org.elasticsearch.common.io.stream.Writeable;
import org.elasticsearch.common.settings.Settings;
@ -46,6 +46,7 @@ import java.util.Set;
import java.util.stream.Collectors;
import static org.elasticsearch.cluster.DataStreamTestHelper.createTimestampField;
import static org.hamcrest.Matchers.containsInAnyOrder;
import static org.hamcrest.Matchers.containsString;
import static org.hamcrest.Matchers.equalTo;
import static org.mockito.Matchers.any;
@ -140,11 +141,18 @@ public class DeleteDataStreamRequestTests extends AbstractWireSerializingTestCas
public void testDeleteNonexistentDataStream() {
final String dataStreamName = "my-data-stream";
ClusterState cs = ClusterState.builder(new ClusterName("_name")).build();
String[] dataStreamNames = {"foo", "bar", "baz", "eggplant"};
ClusterState cs = getClusterStateWithDataStreams(org.elasticsearch.common.collect.List.of(
new Tuple<>(dataStreamNames[0], randomIntBetween(1, 3)),
new Tuple<>(dataStreamNames[1], randomIntBetween(1, 3)),
new Tuple<>(dataStreamNames[2], randomIntBetween(1, 3)),
new Tuple<>(dataStreamNames[3], randomIntBetween(1, 3))
), org.elasticsearch.common.collect.List.of());
DeleteDataStreamAction.Request req = new DeleteDataStreamAction.Request(new String[]{dataStreamName});
ResourceNotFoundException e = expectThrows(ResourceNotFoundException.class,
() -> DeleteDataStreamAction.TransportAction.removeDataStream(getMetadataDeleteIndexService(), cs, req));
assertThat(e.getMessage(), containsString("data_streams matching [" + dataStreamName + "] not found"));
ClusterState newState = DeleteDataStreamAction.TransportAction.removeDataStream(getMetadataDeleteIndexService(), cs, req);
assertThat(newState.metadata().dataStreams().size(), equalTo(cs.metadata().dataStreams().size()));
assertThat(newState.metadata().dataStreams().keySet(),
containsInAnyOrder(cs.metadata().dataStreams().keySet().toArray(Strings.EMPTY_ARRAY)));
}
@SuppressWarnings("unchecked")

99
server/src/test/java/org/elasticsearch/action/admin/indices/datastream/GetDataStreamsRequestTests.java
View File

@ -18,21 +18,19 @@
*/
package org.elasticsearch.action.admin.indices.datastream;
import org.elasticsearch.ResourceNotFoundException;
import org.elasticsearch.action.admin.indices.datastream.GetDataStreamAction.Request;
import org.elasticsearch.cluster.ClusterName;
import org.elasticsearch.cluster.ClusterState;
import org.elasticsearch.cluster.DataStreamTestHelper;
import org.elasticsearch.cluster.metadata.DataStream;
import org.elasticsearch.cluster.metadata.IndexMetadata;
import org.elasticsearch.cluster.metadata.Metadata;
import org.elasticsearch.common.collect.Map;
import org.elasticsearch.cluster.metadata.IndexNameExpressionResolver;
import org.elasticsearch.common.collect.Tuple;
import org.elasticsearch.common.io.stream.Writeable;
import org.elasticsearch.index.IndexNotFoundException;
import org.elasticsearch.test.AbstractWireSerializingTestCase;
import java.util.List;
import static org.elasticsearch.cluster.DataStreamTestHelper.createTimestampField;
import static org.elasticsearch.action.admin.indices.datastream.DeleteDataStreamRequestTests.getClusterStateWithDataStreams;
import static org.hamcrest.Matchers.containsString;
import static org.hamcrest.Matchers.equalTo;
@ -45,16 +43,20 @@ public class GetDataStreamsRequestTests extends AbstractWireSerializingTestCase<
@Override
protected Request createTestInstance() {
final String searchParameter;
final String[] searchParameter;
switch (randomIntBetween(1, 4)) {
case 1:
searchParameter = randomAlphaOfLength(8);
searchParameter = generateRandomStringArray(3, 8, false, false);
break;
case 2:
searchParameter = randomAlphaOfLength(8) + "*";
String[] parameters = generateRandomStringArray(3, 8, false, false);
for (int k = 0; k < parameters.length; k++) {
parameters[k] = parameters[k] + "*";
}
searchParameter = parameters;
break;
case 3:
searchParameter = "*";
searchParameter = new String[]{"*"};
break;
default:
searchParameter = null;
@ -65,60 +67,77 @@ public class GetDataStreamsRequestTests extends AbstractWireSerializingTestCase<
public void testGetDataStream() {
final String dataStreamName = "my-data-stream";
IndexMetadata idx = DataStreamTestHelper.createFirstBackingIndex(dataStreamName).build();
DataStream existingDataStream =
new DataStream(dataStreamName, createTimestampField("@timestamp"), org.elasticsearch.common.collect.List.of(idx.getIndex()));
ClusterState cs = ClusterState.builder(new ClusterName("_name"))
.metadata(Metadata.builder().dataStreams(Map.of(dataStreamName, existingDataStream)).build()).build();
GetDataStreamAction.Request req = new GetDataStreamAction.Request(dataStreamName);
List<DataStream> dataStreams = GetDataStreamAction.TransportAction.getDataStreams(cs, req);
ClusterState cs = getClusterStateWithDataStreams(
org.elasticsearch.common.collect.List.of(new Tuple<>(dataStreamName, 1)), org.elasticsearch.common.collect.List.of());
GetDataStreamAction.Request req = new GetDataStreamAction.Request(new String[]{dataStreamName});
List<DataStream> dataStreams = GetDataStreamAction.TransportAction.getDataStreams(cs, new IndexNameExpressionResolver(), req);
assertThat(dataStreams.size(), equalTo(1));
assertThat(dataStreams.get(0).getName(), equalTo(dataStreamName));
}
public void testGetDataStreamsWithWildcards() {
final String[] dataStreamNames = {"my-data-stream", "another-data-stream"};
IndexMetadata idx1 = DataStreamTestHelper.createFirstBackingIndex(dataStreamNames[0]).build();
IndexMetadata idx2 = DataStreamTestHelper.createFirstBackingIndex(dataStreamNames[1]).build();
ClusterState cs = getClusterStateWithDataStreams(
org.elasticsearch.common.collect.List.of(new Tuple<>(dataStreamNames[0], 1), new Tuple<>(dataStreamNames[1], 1)),
org.elasticsearch.common.collect.List.of());
DataStream ds1 = new DataStream(dataStreamNames[0], createTimestampField("@timestamp"),
org.elasticsearch.common.collect.List.of(idx1.getIndex()));
DataStream ds2 = new DataStream(dataStreamNames[1], createTimestampField("@timestamp"),
org.elasticsearch.common.collect.List.of(idx2.getIndex()));
ClusterState cs = ClusterState.builder(new ClusterName("_name"))
.metadata(Metadata.builder().dataStreams(
Map.of(dataStreamNames[0], ds1, dataStreamNames[1], ds2)).build())
.build();
GetDataStreamAction.Request req = new GetDataStreamAction.Request(dataStreamNames[1].substring(0, 5) + "*");
List<DataStream> dataStreams = GetDataStreamAction.TransportAction.getDataStreams(cs, req);
GetDataStreamAction.Request req = new GetDataStreamAction.Request(new String[]{dataStreamNames[1].substring(0, 5) + "*"});
List<DataStream> dataStreams = GetDataStreamAction.TransportAction.getDataStreams(cs, new IndexNameExpressionResolver(), req);
assertThat(dataStreams.size(), equalTo(1));
assertThat(dataStreams.get(0).getName(), equalTo(dataStreamNames[1]));
req = new GetDataStreamAction.Request("*");
dataStreams = GetDataStreamAction.TransportAction.getDataStreams(cs, req);
req = new GetDataStreamAction.Request(new String[]{"*"});
dataStreams = GetDataStreamAction.TransportAction.getDataStreams(cs, new IndexNameExpressionResolver(), req);
assertThat(dataStreams.size(), equalTo(2));
assertThat(dataStreams.get(0).getName(), equalTo(dataStreamNames[1]));
assertThat(dataStreams.get(1).getName(), equalTo(dataStreamNames[0]));
req = new GetDataStreamAction.Request((String) null);
dataStreams = GetDataStreamAction.TransportAction.getDataStreams(cs, req);
req = new GetDataStreamAction.Request((String[]) null);
dataStreams = GetDataStreamAction.TransportAction.getDataStreams(cs, new IndexNameExpressionResolver(), req);
assertThat(dataStreams.size(), equalTo(2));
assertThat(dataStreams.get(0).getName(), equalTo(dataStreamNames[1]));
assertThat(dataStreams.get(1).getName(), equalTo(dataStreamNames[0]));
req = new GetDataStreamAction.Request("matches-none*");
dataStreams = GetDataStreamAction.TransportAction.getDataStreams(cs, req);
req = new GetDataStreamAction.Request(new String[]{"matches-none*"});
dataStreams = GetDataStreamAction.TransportAction.getDataStreams(cs, new IndexNameExpressionResolver(), req);
assertThat(dataStreams.size(), equalTo(0));
}
public void testGetDataStreamsWithoutWildcards() {
final String[] dataStreamNames = {"my-data-stream", "another-data-stream"};
ClusterState cs = getClusterStateWithDataStreams(
org.elasticsearch.common.collect.List.of(new Tuple<>(dataStreamNames[0], 1), new Tuple<>(dataStreamNames[1], 1)),
org.elasticsearch.common.collect.List.of());
GetDataStreamAction.Request req = new GetDataStreamAction.Request(new String[]{dataStreamNames[0], dataStreamNames[1]});
List<DataStream> dataStreams = GetDataStreamAction.TransportAction.getDataStreams(cs, new IndexNameExpressionResolver(), req);
assertThat(dataStreams.size(), equalTo(2));
assertThat(dataStreams.get(0).getName(), equalTo(dataStreamNames[1]));
assertThat(dataStreams.get(1).getName(), equalTo(dataStreamNames[0]));
req = new GetDataStreamAction.Request(new String[]{dataStreamNames[1]});
dataStreams = GetDataStreamAction.TransportAction.getDataStreams(cs, new IndexNameExpressionResolver(), req);
assertThat(dataStreams.size(), equalTo(1));
assertThat(dataStreams.get(0).getName(), equalTo(dataStreamNames[1]));
req = new GetDataStreamAction.Request(new String[]{dataStreamNames[0]});
dataStreams = GetDataStreamAction.TransportAction.getDataStreams(cs, new IndexNameExpressionResolver(), req);
assertThat(dataStreams.size(), equalTo(1));
assertThat(dataStreams.get(0).getName(), equalTo(dataStreamNames[0]));
GetDataStreamAction.Request req2 = new GetDataStreamAction.Request(new String[]{"foo"});
IndexNotFoundException e = expectThrows(IndexNotFoundException.class,
() -> GetDataStreamAction.TransportAction.getDataStreams(cs, new IndexNameExpressionResolver(), req2));
assertThat(e.getMessage(), containsString("no such index [foo]"));
}
public void testGetNonexistentDataStream() {
final String dataStreamName = "my-data-stream";
ClusterState cs = ClusterState.builder(new ClusterName("_name")).build();
GetDataStreamAction.Request req = new GetDataStreamAction.Request(dataStreamName);
ResourceNotFoundException e = expectThrows(ResourceNotFoundException.class,
() -> GetDataStreamAction.TransportAction.getDataStreams(cs, req));
assertThat(e.getMessage(), containsString("data_stream matching [" + dataStreamName + "] not found"));
GetDataStreamAction.Request req = new GetDataStreamAction.Request(new String[]{dataStreamName});
IndexNotFoundException e = expectThrows(IndexNotFoundException.class,
() -> GetDataStreamAction.TransportAction.getDataStreams(cs, new IndexNameExpressionResolver(), req));
assertThat(e.getMessage(), containsString("no such index [" + dataStreamName + "]"));
}
}

2
x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/security/authz/privilege/ClusterPrivilegeResolver.java
View File

@ -212,8 +212,6 @@ public class ClusterPrivilegeResolver {
public static boolean isClusterAction(String actionName) {
return actionName.startsWith("cluster:") ||
actionName.startsWith("indices:admin/template/") ||
// todo: hack until we implement security of data_streams
actionName.startsWith("indices:admin/data_stream/") ||
actionName.startsWith("indices:admin/index_template/");
}

10
x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/security/authz/privilege/IndexPrivilege.java
View File

@ -14,6 +14,9 @@ import org.elasticsearch.action.admin.indices.alias.get.GetAliasesAction;
import org.elasticsearch.action.admin.indices.close.CloseIndexAction;
import org.elasticsearch.action.admin.indices.create.AutoCreateAction;
import org.elasticsearch.action.admin.indices.create.CreateIndexAction;
import org.elasticsearch.action.admin.indices.datastream.CreateDataStreamAction;
import org.elasticsearch.action.admin.indices.datastream.DeleteDataStreamAction;
import org.elasticsearch.action.admin.indices.datastream.GetDataStreamAction;
import org.elasticsearch.action.admin.indices.delete.DeleteIndexAction;
import org.elasticsearch.action.admin.indices.exists.indices.IndicesExistsAction;
import org.elasticsearch.action.admin.indices.exists.types.TypesExistsAction;
@ -63,12 +66,13 @@ public final class IndexPrivilege extends Privilege {
private static final Automaton MONITOR_AUTOMATON = patterns("indices:monitor/*");
private static final Automaton MANAGE_AUTOMATON =
unionAndMinimize(Arrays.asList(MONITOR_AUTOMATON, patterns("indices:admin/*")));
private static final Automaton CREATE_INDEX_AUTOMATON = patterns(CreateIndexAction.NAME, AutoCreateAction.NAME);
private static final Automaton DELETE_INDEX_AUTOMATON = patterns(DeleteIndexAction.NAME);
private static final Automaton CREATE_INDEX_AUTOMATON = patterns(CreateIndexAction.NAME, AutoCreateAction.NAME,
CreateDataStreamAction.NAME);
private static final Automaton DELETE_INDEX_AUTOMATON = patterns(DeleteIndexAction.NAME, DeleteDataStreamAction.NAME);
private static final Automaton VIEW_METADATA_AUTOMATON = patterns(GetAliasesAction.NAME, AliasesExistAction.NAME,
GetIndexAction.NAME, IndicesExistsAction.NAME, GetFieldMappingsAction.NAME + "*", GetMappingsAction.NAME,
ClusterSearchShardsAction.NAME, TypesExistsAction.NAME, ValidateQueryAction.NAME + "*", GetSettingsAction.NAME,
ExplainLifecycleAction.NAME);
ExplainLifecycleAction.NAME, GetDataStreamAction.NAME);
private static final Automaton MANAGE_FOLLOW_INDEX_AUTOMATON = patterns(PutFollowAction.NAME, UnfollowAction.NAME,
CloseIndexAction.NAME + "*");
private static final Automaton MANAGE_LEADER_INDEX_AUTOMATON = patterns(ForgetFollowerAction.NAME + "*");

7
x-pack/plugin/security/src/main/java/org/elasticsearch/xpack/security/authz/AuthorizationService.java
View File

@ -15,6 +15,7 @@ import org.elasticsearch.action.StepListener;
import org.elasticsearch.action.admin.indices.alias.Alias;
import org.elasticsearch.action.admin.indices.alias.IndicesAliasesAction;
import org.elasticsearch.action.admin.indices.create.CreateIndexRequest;
import org.elasticsearch.action.admin.indices.datastream.CreateDataStreamAction;
import org.elasticsearch.action.bulk.BulkItemRequest;
import org.elasticsearch.action.bulk.BulkShardRequest;
import org.elasticsearch.action.bulk.TransportShardBulkAction;
@ -294,11 +295,11 @@ public class AuthorizationService {
}
//if we are creating an index we need to authorize potential aliases created at the same time
if (IndexPrivilege.CREATE_INDEX_MATCHER.test(action)) {
assert request instanceof CreateIndexRequest;
Set<Alias> aliases = ((CreateIndexRequest) request).aliases();
if (aliases.isEmpty()) {
assert (request instanceof CreateIndexRequest) || (request instanceof CreateDataStreamAction.Request);
if (request instanceof CreateDataStreamAction.Request || ((CreateIndexRequest) request).aliases().isEmpty()) {
runRequestInterceptors(requestInfo, authzInfo, authorizationEngine, listener);
} else {
Set<Alias> aliases = ((CreateIndexRequest) request).aliases();
final RequestInfo aliasesRequestInfo = new RequestInfo(authentication, request, IndicesAliasesAction.NAME);
authzEngine.authorizeIndexAction(aliasesRequestInfo, authzInfo,
ril -> {

165
x-pack/plugin/src/test/resources/rest-api-spec/test/security/authz/50_data_streams.yml
View File

@ -15,7 +15,17 @@ setup:
body: >
{
"indices": [
{ "names": ["simple*"], "privileges": ["read", "write", "view_index_metadata"] }
{ "names": ["simple*"], "privileges": ["read", "write", "create_index", "view_index_metadata", "delete_index"] }
]
}
- do:
security.put_role:
name: "data_stream_role2"
body: >
{
"indices": [
{ "names": ["matches_none"], "privileges": ["read", "write", "create_index", "view_index_metadata", "delete_index"] }
]
}
@ -26,16 +36,26 @@ setup:
{
"password" : "x-pack-test-password",
"roles" : [ "data_stream_role" ],
"full_name" : "user with privileges on data streams but not backing indices"
"full_name" : "user with privileges on some data streams"
}
- do:
security.put_user:
username: "no_authz_user"
body: >
{
"password" : "x-pack-test-password",
"roles" : [ "data_stream_role2" ],
"full_name" : "user with privileges on no data streams"
}
- do:
allowed_warnings:
- "index template [my-template1] has index patterns [simple-data-stream1] matching patterns from existing older templates [global] with patterns (global => [*]); this template [my-template1] will take precedence during new index creation"
- "index template [my-template1] has index patterns [s*] matching patterns from existing older templates [global] with patterns (global => [*]); this template [my-template1] will take precedence during new index creation"
indices.put_index_template:
name: my-template1
body:
index_patterns: [simple-data-stream1]
index_patterns: [s*]
template:
mappings:
properties:
@ -51,11 +71,21 @@ teardown:
username: "test_user"
ignore: 404
- do:
security.delete_user:
username: "test_user2"
ignore: 404
- do:
security.delete_role:
name: "data_stream_role"
ignore: 404
- do:
security.delete_role:
name: "data_stream_role2"
ignore: 404
---
"Test backing indices inherit parent data stream privileges":
- skip:
@ -147,3 +177,130 @@ teardown:
indices.delete_data_stream:
name: simple-data-stream1
- is_true: acknowledged
---
"Test that create data stream is limited to authorized namespace":
- skip:
version: " - 7.99.99"
reason: "change to 7.8.99 after backport"
- do:
headers: { Authorization: "Basic dGVzdF91c2VyOngtcGFjay10ZXN0LXBhc3N3b3Jk" } # test_user
indices.create_data_stream:
name: simple-data-stream1
- is_true: acknowledged
- do: # superuser
indices.delete_data_stream:
name: simple-data-stream1
- is_true: acknowledged
- do:
catch: forbidden
headers: { Authorization: "Basic dGVzdF91c2VyOngtcGFjay10ZXN0LXBhc3N3b3Jk" } # test_user
indices.create_data_stream:
name: outside_of_namespace
---
"Test that get data stream is limited to authorized namespace":
- skip:
version: " - 7.99.99"
reason: "change to 7.8.99 after backport"
- do: # superuser
indices.create_data_stream:
name: simple-data-stream1
- is_true: acknowledged
- do: # superuser
indices.create_data_stream:
name: s-outside-of-authed-namespace
- is_true: acknowledged
- do:
headers: { Authorization: "Basic dGVzdF91c2VyOngtcGFjay10ZXN0LXBhc3N3b3Jk" } # test_user
indices.get_data_stream:
name: simple-data-stream1
- length: { data_streams: 1 }
- match: { data_streams.0.name: simple-data-stream1 }
- do: # superuser
indices.get_data_stream:
name: "*"
# superuser should be authorized for both data streams
- length: { data_streams: 2 }
- match: { data_streams.0.name: s-outside-of-authed-namespace }
- match: { data_streams.1.name: simple-data-stream1 }
- do:
headers: { Authorization: "Basic dGVzdF91c2VyOngtcGFjay10ZXN0LXBhc3N3b3Jk" } # test_user
indices.get_data_stream:
name: "*"
# test_user should be authorized for only one data stream
- length: { data_streams: 1 }
- match: { data_streams.0.name: simple-data-stream1 }
- do:
catch: forbidden
headers: { Authorization: "Basic dGVzdF91c2VyOngtcGFjay10ZXN0LXBhc3N3b3Jk" } # test_user
indices.get_data_stream:
name: outside_of_namespace
- do:
headers: { Authorization: "Basic bm9fYXV0aHpfdXNlcjp4LXBhY2stdGVzdC1wYXNzd29yZA==" } # no_authz_user
indices.get_data_stream: {}
# no_authz_user should not be authorized for any data streams
- length: { data_streams: 0 }
- do: # superuser
indices.delete_data_stream:
name: simple-data-stream1
- is_true: acknowledged
- do: # superuser
indices.delete_data_stream:
name: s-outside-of-authed-namespace
- is_true: acknowledged
---
"Test that delete data stream is limited to authorized namespace":
- skip:
version: " - 7.99.99"
reason: "change to 7.8.99 after backport"
- do: # superuser
indices.create_data_stream:
name: simple-data-stream1
- is_true: acknowledged
- do: # superuser
indices.create_data_stream:
name: s-outside-of-authed-namespace
- is_true: acknowledged
- do:
headers: { Authorization: "Basic dGVzdF91c2VyOngtcGFjay10ZXN0LXBhc3N3b3Jk" } # test_user
indices.delete_data_stream:
name: simple-data-stream1
- is_true: acknowledged
- do:
catch: forbidden
headers: { Authorization: "Basic dGVzdF91c2VyOngtcGFjay10ZXN0LXBhc3N3b3Jk" } # test_user
indices.delete_data_stream:
name: s-outside-of-authed-namespace
- do:
catch: forbidden
headers: { Authorization: "Basic bm9fYXV0aHpfdXNlcjp4LXBhY2stdGVzdC1wYXNzd29yZA==" } # no_authz_user
indices.delete_data_stream:
name: simple-data-stream1
- do: # superuser
indices.delete_data_stream:
name: s-outside-of-authed-namespace
- is_true: acknowledged