mirror of
https://github.com/honeymoose/OpenSearch.git
synced 2025-02-05 20:48:22 +00:00
Stores security headers with the LifecyclePolicy and uses them for AsyncSteps (#30657)
* Stores security headers with the LifecyclePolicy and uses them for AsyncSteps x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/ClientHelp er.java x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/indexlifec ycle/IndexLifecycleMetadata.java x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/indexlifec ycle/LifecyclePolicyMetadata.java x-pack/plugin/core/src/test/java/org/elasticsearch/xpack/core/indexlifec ycle/LifecyclePolicyMetadataTests.java x-pack/plugin/index-lifecycle/src/main/java/org/elasticsearch/xpack/inde xlifecycle/IndexLifecycleRunner.java x-pack/plugin/index-lifecycle/src/main/java/org/elasticsearch/xpack/inde xlifecycle/LifecyclePolicyClient.java x-pack/plugin/index-lifecycle/src/main/java/org/elasticsearch/xpack/inde xlifecycle/PolicyStepsRegistry.java x-pack/plugin/index-lifecycle/src/main/java/org/elasticsearch/xpack/inde xlifecycle/action/TransportDeleteLifcycleAction.java x-pack/plugin/index-lifecycle/src/main/java/org/elasticsearch/xpack/inde xlifecycle/action/TransportGetLifecycleAction.java x-pack/plugin/index-lifecycle/src/main/java/org/elasticsearch/xpack/inde xlifecycle/action/TransportPutLifecycleAction.java x-pack/plugin/index-lifecycle/src/test/java/org/elasticsearch/xpack/inde xlifecycle/ExecuteStepsUpdateTaskTests.java x-pack/plugin/index-lifecycle/src/test/java/org/elasticsearch/xpack/inde xlifecycle/IndexLifecycleMetadataTests.java x-pack/plugin/index-lifecycle/src/test/java/org/elasticsearch/xpack/inde xlifecycle/IndexLifecycleRunnerTests.java x-pack/plugin/index-lifecycle/src/test/java/org/elasticsearch/xpack/inde xlifecycle/IndexLifecycleServiceTests.java x-pack/plugin/index-lifecycle/src/test/java/org/elasticsearch/xpack/inde xlifecycle/LifecyclePolicyClientTests.java x-pack/plugin/index-lifecycle/src/test/java/org/elasticsearch/xpack/inde xlifecycle/PolicyStepsRegistryTests.java * Small renaming and Javadocs x-pack/plugin/index-lifecycle/src/main/java/org/elasticsearch/xpack/inde xlifecycle/LifecyclePolicyClient.java -> x-pack/plugin/index-lifecycle/src/main/java/org/elasticsearch/xpack/inde xlifecycle/LifecyclePolicySecurityClient.java x-pack/plugin/index-lifecycle/src/main/java/org/elasticsearch/xpack/inde xlifecycle/PolicyStepsRegistry.java x-pack/plugin/index-lifecycle/src/test/java/org/elasticsearch/xpack/inde xlifecycle/LifecyclePolicyClientTests.java * Fixes checkstyle x-pack/plugin/index-lifecycle/src/main/java/org/elasticsearch/xpack/inde xlifecycle/LifecyclePolicySecurityClient.java * Fixes checkstyle x-pack/plugin/index-lifecycle/src/test/java/org/elasticsearch/xpack/inde xlifecycle/LifecyclePolicyClientTests.java * Fixes Tests x-pack/plugin/index-lifecycle/src/test/java/org/elasticsearch/xpack/inde xlifecycle/ExecuteStepsUpdateTaskTests.java
This commit is contained in:
Colin Goodheart-Smithe
GitHub
parent
46a3aba798
commit
aa61a1ea62
@ -40,6 +40,7 @@ public final class ClientHelper {
|
||||
public static final String SECURITY_ORIGIN = "security";
|
||||
public static final String WATCHER_ORIGIN = "watcher";
|
||||
public static final String ML_ORIGIN = "ml";
|
||||
public static final String INDEX_LIFECYCLE_ORIGIN = "index_lifecycle";
|
||||
public static final String MONITORING_ORIGIN = "monitoring";
|
||||
public static final String DEPRECATION_ORIGIN = "deprecation";
|
||||
public static final String PERSISTENT_TASK_ORIGIN = "persistent_tasks";
|
||||
|
||||
@ -26,6 +26,8 @@ import java.util.List;
|
||||
import java.util.Map;
|
||||
import java.util.Objects;
|
||||
import java.util.TreeMap;
|
||||
import java.util.function.Function;
|
||||
import java.util.stream.Collectors;
|
||||
|
||||
|
||||
public class IndexLifecycleMetadata implements MetaData.Custom {
|
||||
@ -35,40 +37,45 @@ public class IndexLifecycleMetadata implements MetaData.Custom {
|
||||
@SuppressWarnings("unchecked")
|
||||
public static final ConstructingObjectParser<IndexLifecycleMetadata, Void> PARSER = new ConstructingObjectParser<>(
|
||||
TYPE, a -> new IndexLifecycleMetadata(
|
||||
ObjectParserUtils.convertListToMapValues(LifecyclePolicy::getName, (List<LifecyclePolicy>) a[0])));
|
||||
ObjectParserUtils.convertListToMapValues(LifecyclePolicyMetadata::getName, (List<LifecyclePolicyMetadata>) a[0])));
|
||||
static {
|
||||
PARSER.declareNamedObjects(ConstructingObjectParser.constructorArg(), (p, c, n) -> LifecyclePolicy.parse(p, n),
|
||||
PARSER.declareNamedObjects(ConstructingObjectParser.constructorArg(), (p, c, n) -> LifecyclePolicyMetadata.parse(p, n),
|
||||
v -> {
|
||||
throw new IllegalArgumentException("ordered " + POLICIES_FIELD.getPreferredName() + " are not supported");
|
||||
}, POLICIES_FIELD);
|
||||
}
|
||||
|
||||
private final Map<String, LifecyclePolicy> policies;
|
||||
private final Map<String, LifecyclePolicyMetadata> policyMetadatas;
|
||||
|
||||
public IndexLifecycleMetadata(Map<String, LifecyclePolicy> policies) {
|
||||
this.policies = Collections.unmodifiableMap(policies);
|
||||
public IndexLifecycleMetadata(Map<String, LifecyclePolicyMetadata> policies) {
|
||||
this.policyMetadatas = Collections.unmodifiableMap(policies);
|
||||
}
|
||||
|
||||
public IndexLifecycleMetadata(StreamInput in) throws IOException {
|
||||
int size = in.readVInt();
|
||||
TreeMap<String, LifecyclePolicy> policies = new TreeMap<>();
|
||||
TreeMap<String, LifecyclePolicyMetadata> policies = new TreeMap<>();
|
||||
for (int i = 0; i < size; i++) {
|
||||
policies.put(in.readString(), new LifecyclePolicy(in));
|
||||
policies.put(in.readString(), new LifecyclePolicyMetadata(in));
|
||||
}
|
||||
this.policies = policies;
|
||||
this.policyMetadatas = policies;
|
||||
}
|
||||
|
||||
@Override
|
||||
public void writeTo(StreamOutput out) throws IOException {
|
||||
out.writeVInt(policies.size());
|
||||
for (Map.Entry<String, LifecyclePolicy> entry : policies.entrySet()) {
|
||||
out.writeVInt(policyMetadatas.size());
|
||||
for (Map.Entry<String, LifecyclePolicyMetadata> entry : policyMetadatas.entrySet()) {
|
||||
out.writeString(entry.getKey());
|
||||
entry.getValue().writeTo(out);
|
||||
}
|
||||
}
|
||||
|
||||
public Map<String, LifecyclePolicyMetadata> getPolicyMetadatas() {
|
||||
return policyMetadatas;
|
||||
}
|
||||
|
||||
public Map<String, LifecyclePolicy> getPolicies() {
|
||||
return policies;
|
||||
return policyMetadatas.values().stream().map(LifecyclePolicyMetadata::getPolicy)
|
||||
.collect(Collectors.toMap(LifecyclePolicy::getName, Function.identity()));
|
||||
}
|
||||
|
||||
@Override
|
||||
@ -78,7 +85,7 @@ public class IndexLifecycleMetadata implements MetaData.Custom {
|
||||
|
||||
@Override
|
||||
public XContentBuilder toXContent(XContentBuilder builder, Params params) throws IOException {
|
||||
builder.field(POLICIES_FIELD.getPreferredName(), policies);
|
||||
builder.field(POLICIES_FIELD.getPreferredName(), policyMetadatas);
|
||||
return builder;
|
||||
}
|
||||
|
||||
@ -99,7 +106,7 @@ public class IndexLifecycleMetadata implements MetaData.Custom {
|
||||
|
||||
@Override
|
||||
public int hashCode() {
|
||||
return Objects.hash(policies);
|
||||
return Objects.hash(policyMetadatas);
|
||||
}
|
||||
|
||||
@Override
|
||||
@ -111,7 +118,7 @@ public class IndexLifecycleMetadata implements MetaData.Custom {
|
||||
return false;
|
||||
}
|
||||
IndexLifecycleMetadata other = (IndexLifecycleMetadata) obj;
|
||||
return Objects.equals(policies, other.policies);
|
||||
return Objects.equals(policyMetadatas, other.policyMetadatas);
|
||||
}
|
||||
|
||||
@Override
|
||||
@ -121,20 +128,21 @@ public class IndexLifecycleMetadata implements MetaData.Custom {
|
||||
|
||||
public static class IndexLifecycleMetadataDiff implements NamedDiff<MetaData.Custom> {
|
||||
|
||||
final Diff<Map<String, LifecyclePolicy>> policies;
|
||||
final Diff<Map<String, LifecyclePolicyMetadata>> policies;
|
||||
|
||||
IndexLifecycleMetadataDiff(IndexLifecycleMetadata before, IndexLifecycleMetadata after) {
|
||||
this.policies = DiffableUtils.diff(before.policies, after.policies, DiffableUtils.getStringKeySerializer());
|
||||
this.policies = DiffableUtils.diff(before.policyMetadatas, after.policyMetadatas, DiffableUtils.getStringKeySerializer());
|
||||
}
|
||||
|
||||
public IndexLifecycleMetadataDiff(StreamInput in) throws IOException {
|
||||
this.policies = DiffableUtils.readJdkMapDiff(in, DiffableUtils.getStringKeySerializer(), LifecyclePolicy::new,
|
||||
this.policies = DiffableUtils.readJdkMapDiff(in, DiffableUtils.getStringKeySerializer(), LifecyclePolicyMetadata::new,
|
||||
IndexLifecycleMetadataDiff::readLifecyclePolicyDiffFrom);
|
||||
}
|
||||
|
||||
@Override
|
||||
public MetaData.Custom apply(MetaData.Custom part) {
|
||||
TreeMap<String, LifecyclePolicy> newPolicies = new TreeMap<>(policies.apply(((IndexLifecycleMetadata) part).policies));
|
||||
TreeMap<String, LifecyclePolicyMetadata> newPolicies = new TreeMap<>(
|
||||
policies.apply(((IndexLifecycleMetadata) part).policyMetadatas));
|
||||
return new IndexLifecycleMetadata(newPolicies);
|
||||
}
|
||||
|
||||
@ -148,8 +156,8 @@ public class IndexLifecycleMetadata implements MetaData.Custom {
|
||||
return TYPE;
|
||||
}
|
||||
|
||||
static Diff<LifecyclePolicy> readLifecyclePolicyDiffFrom(StreamInput in) throws IOException {
|
||||
return AbstractDiffable.readDiffFrom(LifecyclePolicy::new, in);
|
||||
static Diff<LifecyclePolicyMetadata> readLifecyclePolicyDiffFrom(StreamInput in) throws IOException {
|
||||
return AbstractDiffable.readDiffFrom(LifecyclePolicyMetadata::new, in);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@ -0,0 +1,100 @@
|
||||
/*
|
||||
* Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
|
||||
* or more contributor license agreements. Licensed under the Elastic License;
|
||||
* you may not use this file except in compliance with the Elastic License.
|
||||
*/
|
||||
|
||||
package org.elasticsearch.xpack.core.indexlifecycle;
|
||||
|
||||
import org.elasticsearch.cluster.AbstractDiffable;
|
||||
import org.elasticsearch.cluster.Diffable;
|
||||
import org.elasticsearch.common.ParseField;
|
||||
import org.elasticsearch.common.io.stream.StreamInput;
|
||||
import org.elasticsearch.common.io.stream.StreamOutput;
|
||||
import org.elasticsearch.common.xcontent.ConstructingObjectParser;
|
||||
import org.elasticsearch.common.xcontent.ObjectParser.ValueType;
|
||||
import org.elasticsearch.common.xcontent.ToXContentObject;
|
||||
import org.elasticsearch.common.xcontent.XContentBuilder;
|
||||
import org.elasticsearch.common.xcontent.XContentParser;
|
||||
|
||||
import java.io.IOException;
|
||||
import java.util.Map;
|
||||
import java.util.Objects;
|
||||
|
||||
public class LifecyclePolicyMetadata extends AbstractDiffable<LifecyclePolicyMetadata>
|
||||
implements ToXContentObject, Diffable<LifecyclePolicyMetadata> {
|
||||
|
||||
public static final ParseField POLICY = new ParseField("policy");
|
||||
public static final ParseField HEADERS = new ParseField("headers");
|
||||
@SuppressWarnings("unchecked")
|
||||
public static final ConstructingObjectParser<LifecyclePolicyMetadata, String> PARSER = new ConstructingObjectParser<>("policy_metadata",
|
||||
a -> new LifecyclePolicyMetadata((LifecyclePolicy) a[0], (Map<String, String>) a[1]));
|
||||
static {
|
||||
PARSER.declareObject(ConstructingObjectParser.constructorArg(), (p, c) -> LifecyclePolicy.parse(p, c), POLICY);
|
||||
PARSER.declareField(ConstructingObjectParser.constructorArg(), p -> p.mapStrings(), HEADERS, ValueType.OBJECT);
|
||||
}
|
||||
|
||||
public static LifecyclePolicyMetadata parse(XContentParser parser, String name) {
|
||||
return PARSER.apply(parser, name);
|
||||
}
|
||||
|
||||
private final LifecyclePolicy policy;
|
||||
private final Map<String, String> headers;
|
||||
|
||||
public LifecyclePolicyMetadata(LifecyclePolicy policy, Map<String, String> headers) {
|
||||
this.policy = policy;
|
||||
this.headers = headers;
|
||||
}
|
||||
|
||||
@SuppressWarnings("unchecked")
|
||||
public LifecyclePolicyMetadata(StreamInput in) throws IOException {
|
||||
this.policy = new LifecyclePolicy(in);
|
||||
this.headers = (Map<String, String>) in.readGenericValue();
|
||||
}
|
||||
|
||||
public Map<String, String> getHeaders() {
|
||||
return headers;
|
||||
}
|
||||
|
||||
public LifecyclePolicy getPolicy() {
|
||||
return policy;
|
||||
}
|
||||
|
||||
public String getName() {
|
||||
return policy.getName();
|
||||
}
|
||||
|
||||
@Override
|
||||
public XContentBuilder toXContent(XContentBuilder builder, Params params) throws IOException {
|
||||
builder.startObject();
|
||||
builder.field(POLICY.getPreferredName(), policy);
|
||||
builder.field(HEADERS.getPreferredName(), headers);
|
||||
builder.endObject();
|
||||
return builder;
|
||||
}
|
||||
|
||||
@Override
|
||||
public void writeTo(StreamOutput out) throws IOException {
|
||||
policy.writeTo(out);
|
||||
out.writeGenericValue(headers);
|
||||
}
|
||||
|
||||
@Override
|
||||
public int hashCode() {
|
||||
return Objects.hash(policy, headers);
|
||||
}
|
||||
|
||||
@Override
|
||||
public boolean equals(Object obj) {
|
||||
if (obj == null) {
|
||||
return false;
|
||||
}
|
||||
if (getClass() != obj.getClass()) {
|
||||
return false;
|
||||
}
|
||||
LifecyclePolicyMetadata other = (LifecyclePolicyMetadata) obj;
|
||||
return Objects.equals(policy, other.policy) &&
|
||||
Objects.equals(headers, other.headers);
|
||||
}
|
||||
|
||||
}
|
||||
@ -0,0 +1,90 @@
|
||||
/*
|
||||
* Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
|
||||
* or more contributor license agreements. Licensed under the Elastic License;
|
||||
* you may not use this file except in compliance with the Elastic License.
|
||||
*/
|
||||
|
||||
package org.elasticsearch.xpack.core.indexlifecycle;
|
||||
|
||||
import org.elasticsearch.cluster.ClusterModule;
|
||||
import org.elasticsearch.common.ParseField;
|
||||
import org.elasticsearch.common.io.stream.NamedWriteableRegistry;
|
||||
import org.elasticsearch.common.io.stream.Writeable.Reader;
|
||||
import org.elasticsearch.common.xcontent.NamedXContentRegistry;
|
||||
import org.elasticsearch.common.xcontent.XContentParser;
|
||||
import org.elasticsearch.test.AbstractSerializingTestCase;
|
||||
import org.junit.Before;
|
||||
|
||||
import java.io.IOException;
|
||||
import java.util.ArrayList;
|
||||
import java.util.Arrays;
|
||||
import java.util.HashMap;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
|
||||
public class LifecyclePolicyMetadataTests extends AbstractSerializingTestCase<LifecyclePolicyMetadata> {
|
||||
|
||||
private String lifecycleName;
|
||||
|
||||
@Before
|
||||
public void setup() {
|
||||
lifecycleName = randomAlphaOfLength(20); // NORELEASE we need to randomise the lifecycle name rather
|
||||
// than use the same name for all instances
|
||||
}
|
||||
|
||||
@Override
|
||||
protected NamedWriteableRegistry getNamedWriteableRegistry() {
|
||||
return new NamedWriteableRegistry(
|
||||
Arrays.asList(new NamedWriteableRegistry.Entry(LifecycleAction.class, MockAction.NAME, MockAction::new),
|
||||
new NamedWriteableRegistry.Entry(LifecycleType.class, TestLifecycleType.TYPE, (in) -> TestLifecycleType.INSTANCE)));
|
||||
}
|
||||
|
||||
@Override
|
||||
protected NamedXContentRegistry xContentRegistry() {
|
||||
List<NamedXContentRegistry.Entry> entries = new ArrayList<>(ClusterModule.getNamedXWriteables());
|
||||
entries.add(new NamedXContentRegistry.Entry(LifecycleAction.class, new ParseField(MockAction.NAME), MockAction::parse));
|
||||
entries.add(new NamedXContentRegistry.Entry(LifecycleType.class, new ParseField(TestLifecycleType.TYPE),
|
||||
(p) -> TestLifecycleType.INSTANCE));
|
||||
return new NamedXContentRegistry(entries);
|
||||
}
|
||||
|
||||
@Override
|
||||
protected LifecyclePolicyMetadata doParseInstance(XContentParser parser) throws IOException {
|
||||
return LifecyclePolicyMetadata.parse(parser, lifecycleName);
|
||||
}
|
||||
|
||||
@Override
|
||||
protected LifecyclePolicyMetadata createTestInstance() {
|
||||
Map<String, String> headers = new HashMap<>();
|
||||
int numberHeaders = between(0, 10);
|
||||
for (int i = 0; i < numberHeaders; i++) {
|
||||
headers.put(randomAlphaOfLength(10), randomAlphaOfLength(10));
|
||||
}
|
||||
return new LifecyclePolicyMetadata(LifecyclePolicyTests.randomLifecyclePolicy(lifecycleName), headers);
|
||||
}
|
||||
|
||||
@Override
|
||||
protected Reader<LifecyclePolicyMetadata> instanceReader() {
|
||||
return LifecyclePolicyMetadata::new;
|
||||
}
|
||||
|
||||
@Override
|
||||
protected LifecyclePolicyMetadata mutateInstance(LifecyclePolicyMetadata instance) throws IOException {
|
||||
LifecyclePolicy policy = instance.getPolicy();
|
||||
Map<String, String> headers = instance.getHeaders();
|
||||
switch (between(0, 1)) {
|
||||
case 0:
|
||||
policy = new LifecyclePolicy(TestLifecycleType.INSTANCE, policy.getName() + randomAlphaOfLengthBetween(1, 5),
|
||||
policy.getPhases());
|
||||
break;
|
||||
case 1:
|
||||
headers = new HashMap<>(headers);
|
||||
headers.put(randomAlphaOfLength(11), randomAlphaOfLength(11));
|
||||
break;
|
||||
default:
|
||||
throw new AssertionError("Illegal randomisation branch");
|
||||
}
|
||||
return new LifecyclePolicyMetadata(policy, headers);
|
||||
}
|
||||
|
||||
}
|
||||
@ -238,7 +238,7 @@ public class IndexLifecycleRunner {
|
||||
}
|
||||
|
||||
private void moveToErrorStep(Index index, String policy, StepKey currentStepKey, Exception e) {
|
||||
logger.debug("policy [" + policy + "] for index [" + index.getName() + "] failed on step [" + currentStepKey
|
||||
logger.error("policy [" + policy + "] for index [" + index.getName() + "] failed on step [" + currentStepKey
|
||||
+ "]. Moving to ERROR step.", e);
|
||||
clusterService.submitStateUpdateTask("ILM", new MoveToErrorStepUpdateTask(index, policy, currentStepKey, e, nowSupplier));
|
||||
}
|
||||
|
||||
@ -0,0 +1,56 @@
|
||||
/*
|
||||
* Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
|
||||
* or more contributor license agreements. Licensed under the Elastic License;
|
||||
* you may not use this file except in compliance with the Elastic License.
|
||||
*/
|
||||
|
||||
package org.elasticsearch.xpack.indexlifecycle;
|
||||
|
||||
import org.elasticsearch.action.Action;
|
||||
import org.elasticsearch.action.ActionListener;
|
||||
import org.elasticsearch.action.ActionRequest;
|
||||
import org.elasticsearch.action.ActionRequestBuilder;
|
||||
import org.elasticsearch.action.ActionResponse;
|
||||
import org.elasticsearch.client.Client;
|
||||
import org.elasticsearch.client.support.AbstractClient;
|
||||
import org.elasticsearch.xpack.core.ClientHelper;
|
||||
import org.elasticsearch.xpack.core.indexlifecycle.Step;
|
||||
|
||||
import java.util.Map;
|
||||
|
||||
/**
|
||||
* This class wraps a client and calls the client using the headers provided in
|
||||
* constructor. The intent is to abstract away the fact that there are headers
|
||||
* so {@link Step}s etc. can call this client as if it was a normal client.
|
||||
*
|
||||
* Note: This client will not close the wrapped {@link Client} instance since
|
||||
* the intent is that the wrapped client is shared between multiple instances of
|
||||
* this class.
|
||||
*/
|
||||
public class LifecyclePolicySecurityClient extends AbstractClient {
|
||||
|
||||
private Client client;
|
||||
private Map<String, String> headers;
|
||||
private String origin;
|
||||
|
||||
public LifecyclePolicySecurityClient(Client client, String origin, Map<String, String> headers) {
|
||||
super(client.settings(), client.threadPool());
|
||||
this.client = client;
|
||||
this.origin = origin;
|
||||
this.headers = headers;
|
||||
}
|
||||
|
||||
@Override
|
||||
public void close() {
|
||||
// Doesn't close the wrapped client since this client object is shared
|
||||
// among multiple instances
|
||||
}
|
||||
|
||||
@Override
|
||||
protected <Request extends ActionRequest, Response extends ActionResponse,
|
||||
RequestBuilder extends ActionRequestBuilder<Request, Response, RequestBuilder>> void doExecute(
|
||||
Action<Request, Response, RequestBuilder> action, Request request, ActionListener<Response> listener) {
|
||||
ClientHelper.executeWithHeadersAsync(headers, origin, client, action, request, listener);
|
||||
}
|
||||
|
||||
}
|
||||
@ -9,9 +9,10 @@ import org.elasticsearch.client.Client;
|
||||
import org.elasticsearch.cluster.ClusterState;
|
||||
import org.elasticsearch.cluster.Diff;
|
||||
import org.elasticsearch.cluster.DiffableUtils;
|
||||
import org.elasticsearch.xpack.core.ClientHelper;
|
||||
import org.elasticsearch.xpack.core.indexlifecycle.ErrorStep;
|
||||
import org.elasticsearch.xpack.core.indexlifecycle.IndexLifecycleMetadata;
|
||||
import org.elasticsearch.xpack.core.indexlifecycle.LifecyclePolicy;
|
||||
import org.elasticsearch.xpack.core.indexlifecycle.LifecyclePolicyMetadata;
|
||||
import org.elasticsearch.xpack.core.indexlifecycle.Step;
|
||||
|
||||
import java.util.HashMap;
|
||||
@ -23,7 +24,7 @@ import java.util.function.LongSupplier;
|
||||
|
||||
public class PolicyStepsRegistry {
|
||||
// keeps track of existing policies in the cluster state
|
||||
private SortedMap<String, LifecyclePolicy> lifecyclePolicyMap;
|
||||
private SortedMap<String, LifecyclePolicyMetadata> lifecyclePolicyMap;
|
||||
// keeps track of what the first step in a policy is
|
||||
private Map<String, Step> firstStepMap;
|
||||
// keeps track of a mapping from policy/step-name to respective Step
|
||||
@ -35,14 +36,14 @@ public class PolicyStepsRegistry {
|
||||
this.stepMap = new HashMap<>();
|
||||
}
|
||||
|
||||
PolicyStepsRegistry(SortedMap<String, LifecyclePolicy> lifecyclePolicyMap,
|
||||
PolicyStepsRegistry(SortedMap<String, LifecyclePolicyMetadata> lifecyclePolicyMap,
|
||||
Map<String, Step> firstStepMap, Map<String, Map<Step.StepKey, Step>> stepMap) {
|
||||
this.lifecyclePolicyMap = lifecyclePolicyMap;
|
||||
this.firstStepMap = firstStepMap;
|
||||
this.stepMap = stepMap;
|
||||
}
|
||||
|
||||
SortedMap<String, LifecyclePolicy> getLifecyclePolicyMap() {
|
||||
SortedMap<String, LifecyclePolicyMetadata> getLifecyclePolicyMap() {
|
||||
return lifecyclePolicyMap;
|
||||
}
|
||||
|
||||
@ -58,17 +59,19 @@ public class PolicyStepsRegistry {
|
||||
@SuppressWarnings({ "unchecked", "rawtypes" })
|
||||
public void update(ClusterState currentState, Client client, LongSupplier nowSupplier) {
|
||||
IndexLifecycleMetadata meta = currentState.metaData().custom(IndexLifecycleMetadata.TYPE);
|
||||
Diff<Map<String, LifecyclePolicy>> diff = DiffableUtils.diff(lifecyclePolicyMap, meta.getPolicies(),
|
||||
Diff<Map<String, LifecyclePolicyMetadata>> diff = DiffableUtils.diff(lifecyclePolicyMap, meta.getPolicyMetadatas(),
|
||||
DiffableUtils.getStringKeySerializer());
|
||||
DiffableUtils.MapDiff<String, LifecyclePolicy, DiffableUtils.KeySerializer<String>> mapDiff = (DiffableUtils.MapDiff) diff;
|
||||
DiffableUtils.MapDiff<String, LifecyclePolicyMetadata, DiffableUtils.KeySerializer<String>> mapDiff = (DiffableUtils.MapDiff) diff;
|
||||
if (mapDiff.getUpserts().isEmpty() == false) {
|
||||
for (LifecyclePolicy policy : mapDiff.getUpserts().values()) {
|
||||
lifecyclePolicyMap.put(policy.getName(), policy);
|
||||
List<Step> policyAsSteps = policy.toSteps(client, nowSupplier);
|
||||
for (LifecyclePolicyMetadata policyMetadata : mapDiff.getUpserts().values()) {
|
||||
LifecyclePolicySecurityClient policyClient = new LifecyclePolicySecurityClient(client, ClientHelper.INDEX_LIFECYCLE_ORIGIN,
|
||||
policyMetadata.getHeaders());
|
||||
lifecyclePolicyMap.put(policyMetadata.getName(), policyMetadata);
|
||||
List<Step> policyAsSteps = policyMetadata.getPolicy().toSteps(policyClient, nowSupplier);
|
||||
if (policyAsSteps.isEmpty() == false) {
|
||||
firstStepMap.put(policy.getName(), policyAsSteps.get(0));
|
||||
stepMap.put(policy.getName(), new HashMap<>());
|
||||
Map<Step.StepKey, Step> stepMapForPolicy = stepMap.get(policy.getName());
|
||||
firstStepMap.put(policyMetadata.getName(), policyAsSteps.get(0));
|
||||
stepMap.put(policyMetadata.getName(), new HashMap<>());
|
||||
Map<Step.StepKey, Step> stepMapForPolicy = stepMap.get(policyMetadata.getName());
|
||||
for (Step step : policyAsSteps) {
|
||||
assert ErrorStep.NAME.equals(step.getKey().getName()) == false;
|
||||
stepMapForPolicy.put(step.getKey(), step);
|
||||
|
||||
@ -22,7 +22,7 @@ import org.elasticsearch.common.settings.Settings;
|
||||
import org.elasticsearch.threadpool.ThreadPool;
|
||||
import org.elasticsearch.transport.TransportService;
|
||||
import org.elasticsearch.xpack.core.indexlifecycle.IndexLifecycleMetadata;
|
||||
import org.elasticsearch.xpack.core.indexlifecycle.LifecyclePolicy;
|
||||
import org.elasticsearch.xpack.core.indexlifecycle.LifecyclePolicyMetadata;
|
||||
import org.elasticsearch.xpack.core.indexlifecycle.LifecycleSettings;
|
||||
import org.elasticsearch.xpack.core.indexlifecycle.action.DeleteLifecycleAction;
|
||||
import org.elasticsearch.xpack.core.indexlifecycle.action.DeleteLifecycleAction.Request;
|
||||
@ -74,10 +74,10 @@ public class TransportDeleteLifcycleAction extends TransportMasterNodeAction<Req
|
||||
}
|
||||
ClusterState.Builder newState = ClusterState.builder(currentState);
|
||||
IndexLifecycleMetadata currentMetadata = currentState.metaData().custom(IndexLifecycleMetadata.TYPE);
|
||||
if (currentMetadata.getPolicies().containsKey(request.getPolicyName()) == false) {
|
||||
if (currentMetadata.getPolicyMetadatas().containsKey(request.getPolicyName()) == false) {
|
||||
throw new ResourceNotFoundException("Lifecycle policy not found: {}", request.getPolicyName());
|
||||
}
|
||||
SortedMap<String, LifecyclePolicy> newPolicies = new TreeMap<>(currentMetadata.getPolicies());
|
||||
SortedMap<String, LifecyclePolicyMetadata> newPolicies = new TreeMap<>(currentMetadata.getPolicyMetadatas());
|
||||
newPolicies.remove(request.getPolicyName());
|
||||
IndexLifecycleMetadata newMetadata = new IndexLifecycleMetadata(newPolicies);
|
||||
newState.metaData(MetaData.builder(currentState.getMetaData())
|
||||
|
||||
@ -18,9 +18,9 @@ import org.elasticsearch.common.inject.Inject;
|
||||
import org.elasticsearch.common.settings.Settings;
|
||||
import org.elasticsearch.threadpool.ThreadPool;
|
||||
import org.elasticsearch.transport.TransportService;
|
||||
import org.elasticsearch.xpack.core.indexlifecycle.action.GetLifecycleAction;
|
||||
import org.elasticsearch.xpack.core.indexlifecycle.IndexLifecycleMetadata;
|
||||
import org.elasticsearch.xpack.core.indexlifecycle.LifecyclePolicy;
|
||||
import org.elasticsearch.xpack.core.indexlifecycle.action.GetLifecycleAction;
|
||||
import org.elasticsearch.xpack.core.indexlifecycle.action.GetLifecycleAction.Request;
|
||||
import org.elasticsearch.xpack.core.indexlifecycle.action.GetLifecycleAction.Response;
|
||||
|
||||
|
||||
@ -20,14 +20,17 @@ import org.elasticsearch.common.inject.Inject;
|
||||
import org.elasticsearch.common.settings.Settings;
|
||||
import org.elasticsearch.threadpool.ThreadPool;
|
||||
import org.elasticsearch.transport.TransportService;
|
||||
import org.elasticsearch.xpack.core.indexlifecycle.action.PutLifecycleAction;
|
||||
import org.elasticsearch.xpack.core.ClientHelper;
|
||||
import org.elasticsearch.xpack.core.indexlifecycle.IndexLifecycleMetadata;
|
||||
import org.elasticsearch.xpack.core.indexlifecycle.LifecyclePolicy;
|
||||
import org.elasticsearch.xpack.core.indexlifecycle.LifecyclePolicyMetadata;
|
||||
import org.elasticsearch.xpack.core.indexlifecycle.action.PutLifecycleAction;
|
||||
import org.elasticsearch.xpack.core.indexlifecycle.action.PutLifecycleAction.Request;
|
||||
import org.elasticsearch.xpack.core.indexlifecycle.action.PutLifecycleAction.Response;
|
||||
|
||||
import java.util.Map;
|
||||
import java.util.SortedMap;
|
||||
import java.util.TreeMap;
|
||||
import java.util.stream.Collectors;
|
||||
|
||||
public class TransportPutLifecycleAction extends TransportMasterNodeAction<Request, Response> {
|
||||
|
||||
@ -61,12 +64,17 @@ public class TransportPutLifecycleAction extends TransportMasterNodeAction<Reque
|
||||
public ClusterState execute(ClusterState currentState) throws Exception {
|
||||
ClusterState.Builder newState = ClusterState.builder(currentState);
|
||||
IndexLifecycleMetadata currentMetadata = currentState.metaData().custom(IndexLifecycleMetadata.TYPE);
|
||||
if (currentMetadata.getPolicies().containsKey(request.getPolicy().getName())) {
|
||||
if (currentMetadata.getPolicyMetadatas().containsKey(request.getPolicy().getName())) {
|
||||
throw new ResourceAlreadyExistsException("Lifecycle policy already exists: {}",
|
||||
request.getPolicy().getName());
|
||||
}
|
||||
SortedMap<String, LifecyclePolicy> newPolicies = new TreeMap<>(currentMetadata.getPolicies());
|
||||
newPolicies.put(request.getPolicy().getName(), request.getPolicy());
|
||||
SortedMap<String, LifecyclePolicyMetadata> newPolicies = new TreeMap<>(currentMetadata.getPolicyMetadatas());
|
||||
|
||||
Map<String, String> filteredHeaders = threadPool.getThreadContext().getHeaders().entrySet().stream()
|
||||
.filter(e -> ClientHelper.SECURITY_HEADER_FILTERS.contains(e.getKey()))
|
||||
.collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
|
||||
LifecyclePolicyMetadata lifecyclePolicyMetadata = new LifecyclePolicyMetadata(request.getPolicy(), filteredHeaders);
|
||||
newPolicies.put(lifecyclePolicyMetadata.getName(), lifecyclePolicyMetadata);
|
||||
IndexLifecycleMetadata newMetadata = new IndexLifecycleMetadata(newPolicies);
|
||||
newState.metaData(MetaData.builder(currentState.getMetaData())
|
||||
.putCustom(IndexLifecycleMetadata.TYPE, newMetadata).build());
|
||||
|
||||
@ -7,6 +7,7 @@ package org.elasticsearch.xpack.indexlifecycle;
|
||||
|
||||
import org.elasticsearch.ElasticsearchException;
|
||||
import org.elasticsearch.Version;
|
||||
import org.elasticsearch.client.Client;
|
||||
import org.elasticsearch.cluster.ClusterName;
|
||||
import org.elasticsearch.cluster.ClusterState;
|
||||
import org.elasticsearch.cluster.metadata.IndexMetaData;
|
||||
@ -21,6 +22,7 @@ import org.elasticsearch.node.Node;
|
||||
import org.elasticsearch.test.ESTestCase;
|
||||
import org.elasticsearch.xpack.core.indexlifecycle.IndexLifecycleMetadata;
|
||||
import org.elasticsearch.xpack.core.indexlifecycle.LifecyclePolicy;
|
||||
import org.elasticsearch.xpack.core.indexlifecycle.LifecyclePolicyMetadata;
|
||||
import org.elasticsearch.xpack.core.indexlifecycle.LifecycleSettings;
|
||||
import org.elasticsearch.xpack.core.indexlifecycle.MockAction;
|
||||
import org.elasticsearch.xpack.core.indexlifecycle.MockStep;
|
||||
@ -32,6 +34,7 @@ import org.elasticsearch.xpack.core.indexlifecycle.TestLifecycleType;
|
||||
import org.elasticsearch.xpack.indexlifecycle.IndexLifecycleRunnerTests.MockClusterStateWaitStep;
|
||||
import org.elasticsearch.xpack.indexlifecycle.IndexLifecycleRunnerTests.MockInitializePolicyContextStep;
|
||||
import org.junit.Before;
|
||||
import org.mockito.Mockito;
|
||||
|
||||
import java.io.IOException;
|
||||
import java.util.Arrays;
|
||||
@ -56,9 +59,12 @@ public class ExecuteStepsUpdateTaskTests extends ESTestCase {
|
||||
private MockClusterStateWaitStep secondStep;
|
||||
private MockClusterStateWaitStep allClusterSecondStep;
|
||||
private MockStep thirdStep;
|
||||
private Client client;
|
||||
|
||||
@Before
|
||||
public void prepareState() {
|
||||
client = Mockito.mock(Client.class);
|
||||
Mockito.when(client.settings()).thenReturn(Settings.EMPTY);
|
||||
firstStep = new MockInitializePolicyContextStep(firstStepKey, secondStepKey);
|
||||
secondStep = new MockClusterStateWaitStep(secondStepKey, thirdStepKey);
|
||||
secondStep.setWillComplete(true);
|
||||
@ -75,9 +81,9 @@ public class ExecuteStepsUpdateTaskTests extends ESTestCase {
|
||||
Collections.singletonMap(mixedPhase.getName(), mixedPhase));
|
||||
LifecyclePolicy allClusterPolicy = new LifecyclePolicy(TestLifecycleType.INSTANCE, allClusterPolicyName,
|
||||
Collections.singletonMap(allClusterPhase.getName(), allClusterPhase));
|
||||
Map<String, LifecyclePolicy> policyMap = new HashMap<>();
|
||||
policyMap.put(mixedPolicyName, mixedPolicy);
|
||||
policyMap.put(allClusterPolicyName, allClusterPolicy);
|
||||
Map<String, LifecyclePolicyMetadata> policyMap = new HashMap<>();
|
||||
policyMap.put(mixedPolicyName, new LifecyclePolicyMetadata(mixedPolicy, Collections.emptyMap()));
|
||||
policyMap.put(allClusterPolicyName, new LifecyclePolicyMetadata(allClusterPolicy, Collections.emptyMap()));
|
||||
policyStepsRegistry = new PolicyStepsRegistry();
|
||||
|
||||
IndexMetaData indexMetadata = IndexMetaData.builder(randomAlphaOfLength(5))
|
||||
@ -102,7 +108,7 @@ public class ExecuteStepsUpdateTaskTests extends ESTestCase {
|
||||
.metaData(metaData)
|
||||
.nodes(DiscoveryNodes.builder().localNodeId(nodeId).masterNodeId(nodeId).add(masterNode).build())
|
||||
.build();
|
||||
policyStepsRegistry.update(clusterState, null, () -> 0L);
|
||||
policyStepsRegistry.update(clusterState, client, () -> 0L);
|
||||
}
|
||||
|
||||
public void testExecuteAllUntilEndOfPolicy() throws IOException {
|
||||
|
||||
@ -19,12 +19,13 @@ import org.elasticsearch.common.xcontent.XContentParser;
|
||||
import org.elasticsearch.test.AbstractDiffableSerializationTestCase;
|
||||
import org.elasticsearch.xpack.core.indexlifecycle.DeleteAction;
|
||||
import org.elasticsearch.xpack.core.indexlifecycle.IndexLifecycleMetadata;
|
||||
import org.elasticsearch.xpack.core.indexlifecycle.IndexLifecycleMetadata.IndexLifecycleMetadataDiff;
|
||||
import org.elasticsearch.xpack.core.indexlifecycle.LifecycleAction;
|
||||
import org.elasticsearch.xpack.core.indexlifecycle.LifecyclePolicy;
|
||||
import org.elasticsearch.xpack.core.indexlifecycle.LifecyclePolicyMetadata;
|
||||
import org.elasticsearch.xpack.core.indexlifecycle.LifecycleType;
|
||||
import org.elasticsearch.xpack.core.indexlifecycle.Phase;
|
||||
import org.elasticsearch.xpack.core.indexlifecycle.TestLifecycleType;
|
||||
import org.elasticsearch.xpack.core.indexlifecycle.IndexLifecycleMetadata.IndexLifecycleMetadataDiff;
|
||||
|
||||
import java.io.IOException;
|
||||
import java.util.ArrayList;
|
||||
@ -41,7 +42,7 @@ public class IndexLifecycleMetadataTests extends AbstractDiffableSerializationTe
|
||||
@Override
|
||||
protected IndexLifecycleMetadata createTestInstance() {
|
||||
int numPolicies = randomInt(5);
|
||||
SortedMap<String, LifecyclePolicy> policies = new TreeMap<>();
|
||||
SortedMap<String, LifecyclePolicyMetadata> policies = new TreeMap<>();
|
||||
for (int i = 0; i < numPolicies; i++) {
|
||||
int numberPhases = randomInt(5);
|
||||
Map<String, Phase> phases = new HashMap<>(numberPhases);
|
||||
@ -55,7 +56,8 @@ public class IndexLifecycleMetadataTests extends AbstractDiffableSerializationTe
|
||||
phases.put(phaseName, new Phase(phaseName, after, actions));
|
||||
}
|
||||
String policyName = randomAlphaOfLength(10);
|
||||
policies.put(policyName, new LifecyclePolicy(TestLifecycleType.INSTANCE, policyName, phases));
|
||||
policies.put(policyName, new LifecyclePolicyMetadata(new LifecyclePolicy(TestLifecycleType.INSTANCE, policyName, phases),
|
||||
Collections.emptyMap()));
|
||||
}
|
||||
return new IndexLifecycleMetadata(policies);
|
||||
}
|
||||
@ -89,10 +91,11 @@ public class IndexLifecycleMetadataTests extends AbstractDiffableSerializationTe
|
||||
@Override
|
||||
protected MetaData.Custom mutateInstance(MetaData.Custom instance) {
|
||||
IndexLifecycleMetadata metadata = (IndexLifecycleMetadata) instance;
|
||||
Map<String, LifecyclePolicy> policies = metadata.getPolicies();
|
||||
Map<String, LifecyclePolicyMetadata> policies = metadata.getPolicyMetadatas();
|
||||
policies = new TreeMap<>(policies);
|
||||
String policyName = randomAlphaOfLength(10);
|
||||
policies.put(policyName, new LifecyclePolicy(TestLifecycleType.INSTANCE, policyName, Collections.emptyMap()));
|
||||
policies.put(policyName, new LifecyclePolicyMetadata(
|
||||
new LifecyclePolicy(TestLifecycleType.INSTANCE, policyName, Collections.emptyMap()), Collections.emptyMap()));
|
||||
return new IndexLifecycleMetadata(policies);
|
||||
}
|
||||
|
||||
|
||||
@ -28,6 +28,7 @@ import org.elasticsearch.xpack.core.indexlifecycle.ClusterStateWaitStep;
|
||||
import org.elasticsearch.xpack.core.indexlifecycle.ErrorStep;
|
||||
import org.elasticsearch.xpack.core.indexlifecycle.InitializePolicyContextStep;
|
||||
import org.elasticsearch.xpack.core.indexlifecycle.LifecyclePolicy;
|
||||
import org.elasticsearch.xpack.core.indexlifecycle.LifecyclePolicyMetadata;
|
||||
import org.elasticsearch.xpack.core.indexlifecycle.LifecycleSettings;
|
||||
import org.elasticsearch.xpack.core.indexlifecycle.MockStep;
|
||||
import org.elasticsearch.xpack.core.indexlifecycle.Step;
|
||||
@ -47,7 +48,7 @@ import static org.hamcrest.Matchers.equalTo;
|
||||
public class IndexLifecycleRunnerTests extends ESTestCase {
|
||||
|
||||
private PolicyStepsRegistry createOneStepPolicyStepRegistry(String policyName, Step step) {
|
||||
SortedMap<String, LifecyclePolicy> lifecyclePolicyMap = null; // Not used in this test
|
||||
SortedMap<String, LifecyclePolicyMetadata> lifecyclePolicyMap = null; // Not used in this test
|
||||
Map<String, Step> firstStepMap = new HashMap<>();
|
||||
firstStepMap.put(policyName, step);
|
||||
Map<String, Map<StepKey, Step>> stepMap = new HashMap<>();
|
||||
@ -395,7 +396,7 @@ public class IndexLifecycleRunnerTests extends ESTestCase {
|
||||
}
|
||||
|
||||
public void testGetCurrentStep() {
|
||||
SortedMap<String, LifecyclePolicy> lifecyclePolicyMap = null; // Not used in the methods tested here
|
||||
SortedMap<String, LifecyclePolicyMetadata> lifecyclePolicyMap = null; // Not used in the methods tested here
|
||||
String policyName = "policy_1";
|
||||
String otherPolicyName = "other_policy";
|
||||
StepKey firstStepKey = new StepKey("phase_1", "action_1", "step_1");
|
||||
|
||||
@ -184,7 +184,7 @@ public class IndexLifecycleServiceTests extends ESTestCase {
|
||||
ClusterStateUpdateTask updateTask = (ClusterStateUpdateTask) invocationOnMock.getArguments()[1];
|
||||
ClusterState newState = updateTask.execute(state);
|
||||
IndexLifecycleMetadata indexLifecycleMetadata = newState.metaData().custom(IndexLifecycleMetadata.TYPE);
|
||||
assertThat(indexLifecycleMetadata.getPolicies(), equalTo(Collections.emptySortedMap()));
|
||||
assertThat(indexLifecycleMetadata.getPolicyMetadatas(), equalTo(Collections.emptySortedMap()));
|
||||
installedEvent.set(new ClusterChangedEvent(event.source(), newState, state));
|
||||
return null;
|
||||
}).when(clusterService).submitStateUpdateTask(anyString(), any(ClusterStateUpdateTask.class));
|
||||
|
||||
@ -0,0 +1,134 @@
|
||||
/*
|
||||
* Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
|
||||
* or more contributor license agreements. Licensed under the Elastic License;
|
||||
* you may not use this file except in compliance with the Elastic License.
|
||||
*/
|
||||
|
||||
package org.elasticsearch.xpack.indexlifecycle;
|
||||
|
||||
import org.elasticsearch.action.ActionListener;
|
||||
import org.elasticsearch.action.search.SearchAction;
|
||||
import org.elasticsearch.action.search.SearchRequest;
|
||||
import org.elasticsearch.action.search.SearchResponse;
|
||||
import org.elasticsearch.client.Client;
|
||||
import org.elasticsearch.common.settings.Settings;
|
||||
import org.elasticsearch.common.util.concurrent.ThreadContext;
|
||||
import org.elasticsearch.test.ESTestCase;
|
||||
import org.elasticsearch.threadpool.ThreadPool;
|
||||
import org.elasticsearch.xpack.core.ClientHelper;
|
||||
import org.mockito.Mockito;
|
||||
|
||||
import java.util.Collections;
|
||||
import java.util.HashMap;
|
||||
import java.util.Map;
|
||||
import java.util.concurrent.CountDownLatch;
|
||||
|
||||
import static org.hamcrest.Matchers.equalTo;
|
||||
import static org.mockito.Matchers.anyObject;
|
||||
import static org.mockito.Mockito.doAnswer;
|
||||
import static org.mockito.Mockito.mock;
|
||||
import static org.mockito.Mockito.when;
|
||||
|
||||
public class LifecyclePolicyClientTests extends ESTestCase {
|
||||
|
||||
public void testExecuteWithHeadersAsyncNoHeaders() throws InterruptedException {
|
||||
final ThreadContext threadContext = new ThreadContext(Settings.EMPTY);
|
||||
final Client client = mock(Client.class);
|
||||
Mockito.when(client.settings()).thenReturn(Settings.EMPTY);
|
||||
final ThreadPool threadPool = mock(ThreadPool.class);
|
||||
when(client.threadPool()).thenReturn(threadPool);
|
||||
when(threadPool.getThreadContext()).thenReturn(threadContext);
|
||||
|
||||
final CountDownLatch latch = new CountDownLatch(2);
|
||||
final ActionListener<SearchResponse> listener = ActionListener.wrap(v -> {
|
||||
assertTrue(threadContext.getHeaders().isEmpty());
|
||||
latch.countDown();
|
||||
}, e -> fail(e.getMessage()));
|
||||
|
||||
doAnswer(invocationOnMock -> {
|
||||
assertTrue(threadContext.getHeaders().isEmpty());
|
||||
latch.countDown();
|
||||
((ActionListener<?>) invocationOnMock.getArguments()[2]).onResponse(null);
|
||||
return null;
|
||||
}).when(client).execute(anyObject(), anyObject(), anyObject());
|
||||
|
||||
SearchRequest request = new SearchRequest("foo");
|
||||
|
||||
try (LifecyclePolicySecurityClient policyClient = new LifecyclePolicySecurityClient(client, ClientHelper.INDEX_LIFECYCLE_ORIGIN,
|
||||
Collections.emptyMap())) {
|
||||
policyClient.execute(SearchAction.INSTANCE, request, listener);
|
||||
}
|
||||
|
||||
latch.await();
|
||||
}
|
||||
|
||||
public void testExecuteWithHeadersAsyncWrongHeaders() throws InterruptedException {
|
||||
final ThreadContext threadContext = new ThreadContext(Settings.EMPTY);
|
||||
final Client client = mock(Client.class);
|
||||
Mockito.when(client.settings()).thenReturn(Settings.EMPTY);
|
||||
final ThreadPool threadPool = mock(ThreadPool.class);
|
||||
when(client.threadPool()).thenReturn(threadPool);
|
||||
when(threadPool.getThreadContext()).thenReturn(threadContext);
|
||||
|
||||
final CountDownLatch latch = new CountDownLatch(2);
|
||||
final ActionListener<SearchResponse> listener = ActionListener.wrap(v -> {
|
||||
assertTrue(threadContext.getHeaders().isEmpty());
|
||||
latch.countDown();
|
||||
}, e -> fail(e.getMessage()));
|
||||
|
||||
doAnswer(invocationOnMock -> {
|
||||
assertTrue(threadContext.getHeaders().isEmpty());
|
||||
latch.countDown();
|
||||
((ActionListener<?>) invocationOnMock.getArguments()[2]).onResponse(null);
|
||||
return null;
|
||||
}).when(client).execute(anyObject(), anyObject(), anyObject());
|
||||
|
||||
SearchRequest request = new SearchRequest("foo");
|
||||
Map<String, String> headers = new HashMap<>(1);
|
||||
headers.put("foo", "foo");
|
||||
headers.put("bar", "bar");
|
||||
|
||||
try (LifecyclePolicySecurityClient policyClient = new LifecyclePolicySecurityClient(client, ClientHelper.INDEX_LIFECYCLE_ORIGIN,
|
||||
headers)) {
|
||||
policyClient.execute(SearchAction.INSTANCE, request, listener);
|
||||
}
|
||||
|
||||
latch.await();
|
||||
}
|
||||
|
||||
public void testExecuteWithHeadersAsyncWithHeaders() throws Exception {
|
||||
final ThreadContext threadContext = new ThreadContext(Settings.EMPTY);
|
||||
final Client client = mock(Client.class);
|
||||
Mockito.when(client.settings()).thenReturn(Settings.EMPTY);
|
||||
final ThreadPool threadPool = mock(ThreadPool.class);
|
||||
when(client.threadPool()).thenReturn(threadPool);
|
||||
when(threadPool.getThreadContext()).thenReturn(threadContext);
|
||||
|
||||
final CountDownLatch latch = new CountDownLatch(2);
|
||||
final ActionListener<SearchResponse> listener = ActionListener.wrap(v -> {
|
||||
assertTrue(threadContext.getHeaders().isEmpty());
|
||||
latch.countDown();
|
||||
}, e -> fail(e.getMessage()));
|
||||
|
||||
doAnswer(invocationOnMock -> {
|
||||
assertThat(threadContext.getHeaders().size(), equalTo(2));
|
||||
assertThat(threadContext.getHeaders().get("es-security-runas-user"), equalTo("foo"));
|
||||
assertThat(threadContext.getHeaders().get("_xpack_security_authentication"), equalTo("bar"));
|
||||
latch.countDown();
|
||||
((ActionListener<?>) invocationOnMock.getArguments()[2]).onResponse(null);
|
||||
return null;
|
||||
}).when(client).execute(anyObject(), anyObject(), anyObject());
|
||||
|
||||
SearchRequest request = new SearchRequest("foo");
|
||||
Map<String, String> headers = new HashMap<>(1);
|
||||
headers.put("es-security-runas-user", "foo");
|
||||
headers.put("_xpack_security_authentication", "bar");
|
||||
|
||||
try (LifecyclePolicySecurityClient policyClient = new LifecyclePolicySecurityClient(client, ClientHelper.INDEX_LIFECYCLE_ORIGIN,
|
||||
headers)) {
|
||||
policyClient.execute(SearchAction.INSTANCE, request, listener);
|
||||
}
|
||||
|
||||
latch.await();
|
||||
}
|
||||
}
|
||||
@ -6,22 +6,27 @@
|
||||
package org.elasticsearch.xpack.indexlifecycle;
|
||||
|
||||
import org.elasticsearch.Version;
|
||||
import org.elasticsearch.client.Client;
|
||||
import org.elasticsearch.cluster.ClusterName;
|
||||
import org.elasticsearch.cluster.ClusterState;
|
||||
import org.elasticsearch.cluster.metadata.MetaData;
|
||||
import org.elasticsearch.cluster.node.DiscoveryNode;
|
||||
import org.elasticsearch.cluster.node.DiscoveryNodes;
|
||||
import org.elasticsearch.common.settings.Settings;
|
||||
import org.elasticsearch.common.transport.TransportAddress;
|
||||
import org.elasticsearch.node.Node;
|
||||
import org.elasticsearch.test.ESTestCase;
|
||||
import org.elasticsearch.xpack.core.indexlifecycle.ErrorStep;
|
||||
import org.elasticsearch.xpack.core.indexlifecycle.IndexLifecycleMetadata;
|
||||
import org.elasticsearch.xpack.core.indexlifecycle.LifecyclePolicy;
|
||||
import org.elasticsearch.xpack.core.indexlifecycle.LifecyclePolicyMetadata;
|
||||
import org.elasticsearch.xpack.core.indexlifecycle.LifecyclePolicyTests;
|
||||
import org.elasticsearch.xpack.core.indexlifecycle.MockStep;
|
||||
import org.elasticsearch.xpack.core.indexlifecycle.Step;
|
||||
import org.mockito.Mockito;
|
||||
|
||||
import java.util.Collections;
|
||||
import java.util.HashMap;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
|
||||
@ -90,9 +95,17 @@ public class PolicyStepsRegistryTests extends ESTestCase {
|
||||
}
|
||||
|
||||
public void testUpdateFromNothingToSomethingToNothing() {
|
||||
Client client = Mockito.mock(Client.class);
|
||||
Mockito.when(client.settings()).thenReturn(Settings.EMPTY);
|
||||
LifecyclePolicy newPolicy = LifecyclePolicyTests.randomLifecyclePolicy(randomAlphaOfLength(5));
|
||||
List<Step> policySteps = newPolicy.toSteps(null, () -> 0L);
|
||||
Map<String, LifecyclePolicy> policyMap = Collections.singletonMap(newPolicy.getName(), newPolicy);
|
||||
List<Step> policySteps = newPolicy.toSteps(client, () -> 0L);
|
||||
Map<String, String> headers = new HashMap<>();
|
||||
if (randomBoolean()) {
|
||||
headers.put(randomAlphaOfLength(10), randomAlphaOfLength(10));
|
||||
headers.put(randomAlphaOfLength(10), randomAlphaOfLength(10));
|
||||
}
|
||||
Map<String, LifecyclePolicyMetadata> policyMap = Collections.singletonMap(newPolicy.getName(),
|
||||
new LifecyclePolicyMetadata(newPolicy, headers));
|
||||
MetaData metaData = MetaData.builder()
|
||||
.persistentSettings(settings(Version.CURRENT).build())
|
||||
.putCustom(IndexLifecycleMetadata.TYPE, new IndexLifecycleMetadata(policyMap))
|
||||
@ -110,10 +123,12 @@ public class PolicyStepsRegistryTests extends ESTestCase {
|
||||
PolicyStepsRegistry registry = new PolicyStepsRegistry();
|
||||
|
||||
// add new policy
|
||||
registry.update(currentState, null, () -> 0L);
|
||||
registry.update(currentState, client, () -> 0L);
|
||||
|
||||
assertThat(registry.getFirstStep(newPolicy.getName()), equalTo(policySteps.get(0)));
|
||||
assertThat(registry.getLifecyclePolicyMap().size(), equalTo(1));
|
||||
assertNotNull(registry.getLifecyclePolicyMap().get(newPolicy.getName()));
|
||||
assertThat(registry.getLifecyclePolicyMap().get(newPolicy.getName()).getHeaders(), equalTo(headers));
|
||||
assertThat(registry.getFirstStepMap().size(), equalTo(1));
|
||||
assertThat(registry.getStepMap().size(), equalTo(1));
|
||||
Map<Step.StepKey, Step> registeredStepsForPolicy = registry.getStepMap().get(newPolicy.getName());
|
||||
@ -123,10 +138,10 @@ public class PolicyStepsRegistryTests extends ESTestCase {
|
||||
assertThat(registry.getStep(newPolicy.getName(), step.getKey()), equalTo(step));
|
||||
}
|
||||
|
||||
Map<String, LifecyclePolicy> registryPolicyMap = registry.getLifecyclePolicyMap();
|
||||
Map<String, LifecyclePolicyMetadata> registryPolicyMap = registry.getLifecyclePolicyMap();
|
||||
Map<String, Step> registryFirstStepMap = registry.getFirstStepMap();
|
||||
Map<String, Map<Step.StepKey, Step>> registryStepMap = registry.getStepMap();
|
||||
registry.update(currentState, null, () -> 0L);
|
||||
registry.update(currentState, client, () -> 0L);
|
||||
assertThat(registry.getLifecyclePolicyMap(), equalTo(registryPolicyMap));
|
||||
assertThat(registry.getFirstStepMap(), equalTo(registryFirstStepMap));
|
||||
assertThat(registry.getStepMap(), equalTo(registryStepMap));
|
||||
@ -136,16 +151,24 @@ public class PolicyStepsRegistryTests extends ESTestCase {
|
||||
.metaData(
|
||||
MetaData.builder(metaData)
|
||||
.putCustom(IndexLifecycleMetadata.TYPE, new IndexLifecycleMetadata(Collections.emptyMap()))).build();
|
||||
registry.update(currentState, null, () -> 0L);
|
||||
registry.update(currentState, client, () -> 0L);
|
||||
assertTrue(registry.getLifecyclePolicyMap().isEmpty());
|
||||
assertTrue(registry.getFirstStepMap().isEmpty());
|
||||
assertTrue(registry.getStepMap().isEmpty());
|
||||
}
|
||||
|
||||
public void testUpdateChangedPolicy() {
|
||||
Client client = Mockito.mock(Client.class);
|
||||
Mockito.when(client.settings()).thenReturn(Settings.EMPTY);
|
||||
String policyName = randomAlphaOfLengthBetween(5, 10);
|
||||
LifecyclePolicy newPolicy = LifecyclePolicyTests.randomLifecyclePolicy(policyName);
|
||||
Map<String, LifecyclePolicy> policyMap = Collections.singletonMap(newPolicy.getName(), newPolicy);
|
||||
Map<String, String> headers = new HashMap<>();
|
||||
if (randomBoolean()) {
|
||||
headers.put(randomAlphaOfLength(10), randomAlphaOfLength(10));
|
||||
headers.put(randomAlphaOfLength(10), randomAlphaOfLength(10));
|
||||
}
|
||||
Map<String, LifecyclePolicyMetadata> policyMap = Collections.singletonMap(newPolicy.getName(),
|
||||
new LifecyclePolicyMetadata(newPolicy, headers));
|
||||
MetaData metaData = MetaData.builder()
|
||||
.persistentSettings(settings(Version.CURRENT).build())
|
||||
.putCustom(IndexLifecycleMetadata.TYPE, new IndexLifecycleMetadata(policyMap))
|
||||
@ -160,7 +183,7 @@ public class PolicyStepsRegistryTests extends ESTestCase {
|
||||
.build();
|
||||
PolicyStepsRegistry registry = new PolicyStepsRegistry();
|
||||
// add new policy
|
||||
registry.update(currentState, null, () -> 0L);
|
||||
registry.update(currentState, client, () -> 0L);
|
||||
|
||||
// swap out policy
|
||||
newPolicy = LifecyclePolicyTests.randomLifecyclePolicy(policyName);
|
||||
@ -168,8 +191,10 @@ public class PolicyStepsRegistryTests extends ESTestCase {
|
||||
.metaData(
|
||||
MetaData.builder(metaData)
|
||||
.putCustom(IndexLifecycleMetadata.TYPE,
|
||||
new IndexLifecycleMetadata(Collections.singletonMap(policyName, newPolicy)))).build();
|
||||
registry.update(currentState, null, () -> 0L);
|
||||
new IndexLifecycleMetadata(Collections.singletonMap(policyName,
|
||||
new LifecyclePolicyMetadata(newPolicy, Collections.emptyMap())))))
|
||||
.build();
|
||||
registry.update(currentState, client, () -> 0L);
|
||||
// TODO(talevy): assert changes... right now we do not support updates to policies. will require internal cleanup
|
||||
}
|
||||
}
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user