Stores security headers with the LifecyclePolicy and uses them for AsyncSteps

x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/ClientHelper.java

@@ -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";

x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/indexlifecycle/IndexLifecycleMetadata.java

@@ -26,6 +26,8 @@
 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 Diff<Custom> diff(Custom previousState) {
 
     @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 EnumSet<MetaData.XContentContext> context() {
 
     @Override
     public int hashCode() {
-        return Objects.hash(policies);
+        return Objects.hash(policyMetadatas);
     }
 
     @Override
@@ -111,7 +118,7 @@ public boolean equals(Object obj) {
             return false;
         }
         IndexLifecycleMetadata other = (IndexLifecycleMetadata) obj;
-        return Objects.equals(policies, other.policies);
+        return Objects.equals(policyMetadatas, other.policyMetadatas);
     }
 
     @Override
@@ -121,20 +128,21 @@ public String toString() {
 
     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 String getWriteableName() {
             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);
         }
     }
 }

x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/indexlifecycle/LifecyclePolicyMetadata.java

@@ -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);
+    }
+
+}

x-pack/plugin/core/src/test/java/org/elasticsearch/xpack/core/indexlifecycle/LifecyclePolicyMetadataTests.java

@@ -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);
+    }
+
+}

x-pack/plugin/index-lifecycle/src/main/java/org/elasticsearch/xpack/indexlifecycle/IndexLifecycleRunner.java

@@ -238,7 +238,7 @@ private void moveToStep(Index index, String policy, StepKey currentStepKey, Step
     }
 
     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));
     }

x-pack/plugin/index-lifecycle/src/main/java/org/elasticsearch/xpack/indexlifecycle/LifecyclePolicySecurityClient.java

@@ -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);
+    }
+
+}