-
Notifications
You must be signed in to change notification settings - Fork 24.9k
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Browse files
Browse the repository at this point in the history
this commit introduces a new Rollup ILM Action that allows indices to be rolled up according to a specific rollup config. The action also allows for the new rolled up index to be associated with a different policy than the original/source index.
- Loading branch information
Showing
21 changed files
with
881 additions
and
42 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,56 @@ | ||
| [role="xpack"] | ||
| [[ilm-rollup]] | ||
| === Rollup | ||
|
|
||
| Phases allowed: hot, cold. | ||
|
|
||
| Aggregates an index's time series data and stores the results in a new read-only | ||
| index. For example, you can roll up hourly data into daily or weekly summaries. | ||
|
|
||
| For more information about rollup, see the <<rollup-api, rollup action documentation>> | ||
|
|
||
| The name of the rolled up index will be the original index name of the managed index prefixed | ||
| with `rollup-`. | ||
|
|
||
| [[ilm-rollup-options]] | ||
| ==== Rollup options | ||
| `config`:: | ||
| (Required, integer) | ||
| The rollup configuration, a more detailed description of the | ||
| rollup configuration specification can be found <<rollup-api-request-body,here>>. | ||
|
|
||
| `rollup_policy`:: | ||
| (Optional, string) | ||
| The name of an <<index-lifecycle-management, {ilm}>> ({ilm-init}) policy to associate | ||
| with the newly created rollup index. | ||
|
|
||
| [[ilm-rollup-ex]] | ||
| ==== Example | ||
|
|
||
| [source,console] | ||
| -------------------------------------------------- | ||
| PUT _ilm/policy/my_policy | ||
| { | ||
| "policy": { | ||
| "phases": { | ||
| "cold": { | ||
| "actions": { | ||
| "rollup" : { | ||
| "config": { | ||
| "groups": { | ||
| "date_histogram": { | ||
| "field": "@timestamp", | ||
| "calendar_interval": "1y" | ||
| } | ||
| }, | ||
| "metrics": [ | ||
| { "field": "temperature", "metrics": [ "avg" ] } | ||
| ] | ||
| } | ||
| } | ||
| } | ||
| } | ||
| } | ||
| } | ||
| } | ||
| -------------------------------------------------- |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
136 changes: 136 additions & 0 deletions
136
x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/ilm/RollupILMAction.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,136 @@ | ||
| /* | ||
| * 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.ilm; | ||
|
|
||
| import org.elasticsearch.client.Client; | ||
| import org.elasticsearch.common.Nullable; | ||
| import org.elasticsearch.common.ParseField; | ||
| import org.elasticsearch.common.Strings; | ||
| 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; | ||
| import org.elasticsearch.common.xcontent.XContentBuilder; | ||
| import org.elasticsearch.common.xcontent.XContentParser; | ||
| import org.elasticsearch.xpack.core.ilm.Step.StepKey; | ||
| import org.elasticsearch.xpack.core.rollup.RollupActionConfig; | ||
|
|
||
| import java.io.IOException; | ||
| import java.util.Arrays; | ||
| import java.util.List; | ||
| import java.util.Objects; | ||
|
|
||
| /** | ||
| * A {@link LifecycleAction} which calls {@link org.elasticsearch.xpack.core.rollup.action.RollupAction} on an index | ||
| */ | ||
| public class RollupILMAction implements LifecycleAction { | ||
| public static final String NAME = "rollup"; | ||
|
|
||
| private static final ParseField CONFIG_FIELD = new ParseField("config"); | ||
| private static final ParseField POLICY_FIELD = new ParseField("rollup_policy"); | ||
|
|
||
| @SuppressWarnings("unchecked") | ||
| private static final ConstructingObjectParser<RollupILMAction, Void> PARSER = new ConstructingObjectParser<>(NAME, | ||
| a -> new RollupILMAction((RollupActionConfig) a[0], (String) a[1])); | ||
|
|
||
| private final RollupActionConfig config; | ||
| private final String rollupPolicy; | ||
|
|
||
| static { | ||
| PARSER.declareField(ConstructingObjectParser.constructorArg(), | ||
| (p, c) -> RollupActionConfig.fromXContent(p), CONFIG_FIELD, ObjectParser.ValueType.OBJECT); | ||
| PARSER.declareString(ConstructingObjectParser.optionalConstructorArg(), POLICY_FIELD); | ||
| } | ||
|
|
||
| public static RollupILMAction parse(XContentParser parser) { | ||
| return PARSER.apply(parser, null); | ||
| } | ||
|
|
||
| public RollupILMAction(RollupActionConfig config, @Nullable String rollupPolicy) { | ||
| this.config = config; | ||
| this.rollupPolicy = rollupPolicy; | ||
| } | ||
|
|
||
| public RollupILMAction(StreamInput in) throws IOException { | ||
| this(new RollupActionConfig(in), in.readOptionalString()); | ||
| } | ||
|
|
||
| @Override | ||
| public String getWriteableName() { | ||
| return NAME; | ||
| } | ||
|
|
||
| RollupActionConfig config() { | ||
| return config; | ||
| } | ||
|
|
||
| String rollupPolicy() { | ||
| return rollupPolicy; | ||
| } | ||
|
|
||
| @Override | ||
| public XContentBuilder toXContent(XContentBuilder builder, Params params) throws IOException { | ||
| builder.startObject(); | ||
| builder.field(CONFIG_FIELD.getPreferredName(), config); | ||
| if (rollupPolicy != null) { | ||
| builder.field(POLICY_FIELD.getPreferredName(), rollupPolicy); | ||
| } | ||
| builder.endObject(); | ||
| return builder; | ||
| } | ||
|
|
||
| @Override | ||
| public void writeTo(StreamOutput out) throws IOException { | ||
| config.writeTo(out); | ||
| out.writeOptionalString(rollupPolicy); | ||
| } | ||
|
|
||
| @Override | ||
| public boolean isSafeAction() { | ||
| return false; | ||
| } | ||
|
|
||
| @Override | ||
| public List<Step> toSteps(Client client, String phase, StepKey nextStepKey) { | ||
| StepKey checkNotWriteIndex = new StepKey(phase, NAME, CheckNotDataStreamWriteIndexStep.NAME); | ||
| StepKey readOnlyKey = new StepKey(phase, NAME, ReadOnlyStep.NAME); | ||
| StepKey rollupKey = new StepKey(phase, NAME, NAME); | ||
| CheckNotDataStreamWriteIndexStep checkNotWriteIndexStep = new CheckNotDataStreamWriteIndexStep(checkNotWriteIndex, | ||
| readOnlyKey); | ||
| ReadOnlyStep readOnlyStep = new ReadOnlyStep(readOnlyKey, rollupKey, client); | ||
| if (rollupPolicy == null) { | ||
| Step rollupStep = new RollupStep(rollupKey, nextStepKey, client, config); | ||
| return Arrays.asList(checkNotWriteIndexStep, readOnlyStep, rollupStep); | ||
| } else { | ||
| StepKey updateRollupIndexPolicyStepKey = new StepKey(phase, NAME, UpdateRollupIndexPolicyStep.NAME); | ||
| Step rollupStep = new RollupStep(rollupKey, updateRollupIndexPolicyStepKey, client, config); | ||
| Step updateRollupIndexPolicyStep = new UpdateRollupIndexPolicyStep(updateRollupIndexPolicyStepKey, nextStepKey, | ||
| client, rollupPolicy); | ||
| return Arrays.asList(checkNotWriteIndexStep, readOnlyStep, rollupStep, updateRollupIndexPolicyStep); | ||
| } | ||
| } | ||
|
|
||
| @Override | ||
| public boolean equals(Object o) { | ||
| if (this == o) return true; | ||
| if (o == null || getClass() != o.getClass()) return false; | ||
|
|
||
| RollupILMAction that = (RollupILMAction) o; | ||
|
|
||
| return Objects.equals(this.config, that.config) | ||
| && Objects.equals(this.rollupPolicy, that.rollupPolicy); | ||
| } | ||
|
|
||
| @Override | ||
| public int hashCode() { | ||
| return Objects.hash(config, rollupPolicy); | ||
| } | ||
|
|
||
| @Override | ||
| public String toString() { | ||
| return Strings.toString(this); | ||
| } | ||
| } |
71 changes: 71 additions & 0 deletions
71
x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/ilm/RollupStep.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,71 @@ | ||
| /* | ||
| * 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.ilm; | ||
|
|
||
| import org.elasticsearch.action.ActionListener; | ||
| import org.elasticsearch.client.Client; | ||
| import org.elasticsearch.cluster.ClusterState; | ||
| import org.elasticsearch.cluster.ClusterStateObserver; | ||
| import org.elasticsearch.cluster.metadata.IndexMetadata; | ||
| import org.elasticsearch.xpack.core.rollup.RollupActionConfig; | ||
| import org.elasticsearch.xpack.core.rollup.action.RollupAction; | ||
|
|
||
| import java.util.Objects; | ||
|
|
||
| /** | ||
| * Rolls up index using a {@link RollupActionConfig} | ||
| */ | ||
| public class RollupStep extends AsyncActionStep { | ||
| public static final String NAME = "rollup"; | ||
| public static final String ROLLUP_INDEX_NAME_PREFIX = "rollup-"; | ||
|
|
||
| private final RollupActionConfig config; | ||
|
|
||
| public RollupStep(StepKey key, StepKey nextStepKey, Client client, RollupActionConfig config) { | ||
| super(key, nextStepKey, client); | ||
| this.config = config; | ||
| } | ||
|
|
||
| public static String getRollupIndexName(String index) { | ||
| return ROLLUP_INDEX_NAME_PREFIX + index; | ||
| } | ||
|
|
||
| @Override | ||
| public boolean isRetryable() { | ||
| return true; | ||
| } | ||
|
|
||
| @Override | ||
| public void performAction(IndexMetadata indexMetadata, ClusterState currentState, ClusterStateObserver observer, Listener listener) { | ||
| String originalIndex = indexMetadata.getIndex().getName(); | ||
| RollupAction.Request request = new RollupAction.Request(originalIndex, getRollupIndexName(originalIndex), config); | ||
| // currently RollupAction always acknowledges action was complete when no exceptions are thrown. | ||
| getClient().execute(RollupAction.INSTANCE, request, | ||
| ActionListener.wrap(response -> listener.onResponse(true), listener::onFailure)); | ||
| } | ||
|
|
||
| public RollupActionConfig getConfig() { | ||
| return config; | ||
| } | ||
|
|
||
| @Override | ||
| public int hashCode() { | ||
| return Objects.hash(super.hashCode(), config); | ||
| } | ||
|
|
||
| @Override | ||
| public boolean equals(Object obj) { | ||
| if (obj == null) { | ||
| return false; | ||
| } | ||
| if (getClass() != obj.getClass()) { | ||
| return false; | ||
| } | ||
| RollupStep other = (RollupStep) obj; | ||
| return super.equals(obj) | ||
| && Objects.equals(config, other.config); | ||
| } | ||
| } |
Oops, something went wrong.