diff --git a/src/main/java/org/elasticsearch/action/ActionModule.java b/src/main/java/org/elasticsearch/action/ActionModule.java index 8fc26892a48..3bfe69e0ec7 100644 --- a/src/main/java/org/elasticsearch/action/ActionModule.java +++ b/src/main/java/org/elasticsearch/action/ActionModule.java @@ -78,6 +78,10 @@ import org.elasticsearch.action.admin.indices.template.put.PutIndexTemplateActio import org.elasticsearch.action.admin.indices.template.put.TransportPutIndexTemplateAction; import org.elasticsearch.action.admin.indices.validate.query.TransportValidateQueryAction; import org.elasticsearch.action.admin.indices.validate.query.ValidateQueryAction; +import org.elasticsearch.action.admin.indices.warmer.delete.DeleteWarmerAction; +import org.elasticsearch.action.admin.indices.warmer.delete.TransportDeleteWarmerAction; +import org.elasticsearch.action.admin.indices.warmer.put.PutWarmerAction; +import org.elasticsearch.action.admin.indices.warmer.put.TransportPutWarmerAction; import org.elasticsearch.action.bulk.BulkAction; import org.elasticsearch.action.bulk.TransportBulkAction; import org.elasticsearch.action.bulk.TransportShardBulkAction; @@ -182,6 +186,8 @@ public class ActionModule extends AbstractModule { registerAction(FlushAction.INSTANCE, TransportFlushAction.class); registerAction(OptimizeAction.INSTANCE, TransportOptimizeAction.class); registerAction(ClearIndicesCacheAction.INSTANCE, TransportClearIndicesCacheAction.class); + registerAction(PutWarmerAction.INSTANCE, TransportPutWarmerAction.class); + registerAction(DeleteWarmerAction.INSTANCE, TransportDeleteWarmerAction.class); registerAction(IndexAction.INSTANCE, TransportIndexAction.class); registerAction(GetAction.INSTANCE, TransportGetAction.class); diff --git a/src/main/java/org/elasticsearch/action/admin/indices/warmer/delete/DeleteWarmerAction.java b/src/main/java/org/elasticsearch/action/admin/indices/warmer/delete/DeleteWarmerAction.java new file mode 100644 index 00000000000..c88b998f387 --- /dev/null +++ b/src/main/java/org/elasticsearch/action/admin/indices/warmer/delete/DeleteWarmerAction.java @@ -0,0 +1,45 @@ +/* + * Licensed to ElasticSearch and Shay Banon under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. ElasticSearch licenses this + * file to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.elasticsearch.action.admin.indices.warmer.delete; + +import org.elasticsearch.action.admin.indices.IndicesAction; +import org.elasticsearch.client.IndicesAdminClient; + +/** + */ +public class DeleteWarmerAction extends IndicesAction { + + public static final DeleteWarmerAction INSTANCE = new DeleteWarmerAction(); + public static final String NAME = "indices/warmer/delete"; + + private DeleteWarmerAction() { + super(NAME); + } + + @Override + public DeleteWarmerResponse newResponse() { + return new DeleteWarmerResponse(); + } + + @Override + public DeleteWarmerRequestBuilder newRequestBuilder(IndicesAdminClient client) { + return new DeleteWarmerRequestBuilder(client); + } +} diff --git a/src/main/java/org/elasticsearch/action/admin/indices/warmer/delete/DeleteWarmerRequest.java b/src/main/java/org/elasticsearch/action/admin/indices/warmer/delete/DeleteWarmerRequest.java new file mode 100644 index 00000000000..9519acc9ad9 --- /dev/null +++ b/src/main/java/org/elasticsearch/action/admin/indices/warmer/delete/DeleteWarmerRequest.java @@ -0,0 +1,112 @@ +/* + * Licensed to Elastic Search and Shay Banon under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. Elastic Search licenses this + * file to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.elasticsearch.action.admin.indices.warmer.delete; + +import org.elasticsearch.action.ActionRequestValidationException; +import org.elasticsearch.action.support.master.MasterNodeOperationRequest; +import org.elasticsearch.common.Nullable; +import org.elasticsearch.common.io.stream.StreamInput; +import org.elasticsearch.common.io.stream.StreamOutput; + +import java.io.IOException; + +/** + * A request to delete an index warmer. + */ +public class DeleteWarmerRequest extends MasterNodeOperationRequest { + + private String name; + + private String[] indices; + + DeleteWarmerRequest() { + } + + /** + * Constructs a new delete warmer request for the specified name. + * + * @param name: the name (or wildcard expression) of the warmer to match, null to delete all. + */ + public DeleteWarmerRequest(String name) { + this.name = name; + } + + @Override + public ActionRequestValidationException validate() { + ActionRequestValidationException validationException = null; + return validationException; + } + + /** + * The name to delete. + */ + @Nullable + String name() { + return name; + } + + /** + * The name (or wildcard expression) of the index warmer to delete, or null + * to delete all warmers. + */ + public DeleteWarmerRequest name(@Nullable String name) { + this.name = name; + return this; + } + + /** + * Sets the indices this put mapping operation will execute on. + */ + public DeleteWarmerRequest indices(String[] indices) { + this.indices = indices; + return this; + } + + /** + * The indices the mappings will be put. + */ + public String[] indices() { + return indices; + } + + @Override + public void readFrom(StreamInput in) throws IOException { + super.readFrom(in); + name = in.readOptionalUTF(); + indices = new String[in.readVInt()]; + for (int i = 0; i < indices.length; i++) { + indices[i] = in.readUTF(); + } + } + + @Override + public void writeTo(StreamOutput out) throws IOException { + super.writeTo(out); + out.writeOptionalUTF(name); + if (indices == null) { + out.writeVInt(0); + } else { + out.writeVInt(indices.length); + for (String index : indices) { + out.writeUTF(index); + } + } + } +} \ No newline at end of file diff --git a/src/main/java/org/elasticsearch/action/admin/indices/warmer/delete/DeleteWarmerRequestBuilder.java b/src/main/java/org/elasticsearch/action/admin/indices/warmer/delete/DeleteWarmerRequestBuilder.java new file mode 100644 index 00000000000..ceadce77315 --- /dev/null +++ b/src/main/java/org/elasticsearch/action/admin/indices/warmer/delete/DeleteWarmerRequestBuilder.java @@ -0,0 +1,62 @@ +/* + * Licensed to ElasticSearch and Shay Banon under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. ElasticSearch licenses this + * file to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.elasticsearch.action.admin.indices.warmer.delete; + +import org.elasticsearch.action.ActionListener; +import org.elasticsearch.action.admin.indices.support.BaseIndicesRequestBuilder; +import org.elasticsearch.client.IndicesAdminClient; +import org.elasticsearch.common.unit.TimeValue; + +/** + * + */ +public class DeleteWarmerRequestBuilder extends BaseIndicesRequestBuilder { + + public DeleteWarmerRequestBuilder(IndicesAdminClient indicesClient) { + super(indicesClient, new DeleteWarmerRequest()); + } + + public DeleteWarmerRequestBuilder setIndices(String... indices) { + request.indices(indices); + return this; + } + + /** + * The name (or wildcard expression) of the index warmer to delete, or null + * to delete all warmers. + */ + public DeleteWarmerRequestBuilder setName(String name) { + request.name(name); + return this; + } + + /** + * Sets the master node timeout in case the master has not yet been discovered. + */ + public DeleteWarmerRequestBuilder setMasterNodeTimeout(TimeValue timeout) { + request.masterNodeTimeout(timeout); + return this; + } + + @Override + protected void doExecute(ActionListener listener) { + client.deleteWarmer(request, listener); + } +} \ No newline at end of file diff --git a/src/main/java/org/elasticsearch/action/admin/indices/warmer/delete/DeleteWarmerResponse.java b/src/main/java/org/elasticsearch/action/admin/indices/warmer/delete/DeleteWarmerResponse.java new file mode 100644 index 00000000000..0a2d91be9ef --- /dev/null +++ b/src/main/java/org/elasticsearch/action/admin/indices/warmer/delete/DeleteWarmerResponse.java @@ -0,0 +1,60 @@ +/* + * Licensed to Elastic Search and Shay Banon under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. Elastic Search licenses this + * file to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.elasticsearch.action.admin.indices.warmer.delete; + +import org.elasticsearch.action.ActionResponse; +import org.elasticsearch.common.io.stream.StreamInput; +import org.elasticsearch.common.io.stream.StreamOutput; +import org.elasticsearch.common.io.stream.Streamable; + +import java.io.IOException; + +/** + * A response for a delete warmer. + */ +public class DeleteWarmerResponse implements ActionResponse, Streamable { + + private boolean acknowledged; + + DeleteWarmerResponse() { + } + + DeleteWarmerResponse(boolean acknowledged) { + this.acknowledged = acknowledged; + } + + public boolean acknowledged() { + return acknowledged; + } + + public boolean getAcknowledged() { + return acknowledged(); + } + + @Override + public void readFrom(StreamInput in) throws IOException { + acknowledged = in.readBoolean(); + } + + @Override + public void writeTo(StreamOutput out) throws IOException { + out.writeBoolean(acknowledged); + } +} diff --git a/src/main/java/org/elasticsearch/action/admin/indices/warmer/delete/TransportDeleteWarmerAction.java b/src/main/java/org/elasticsearch/action/admin/indices/warmer/delete/TransportDeleteWarmerAction.java new file mode 100644 index 00000000000..247e75b6f47 --- /dev/null +++ b/src/main/java/org/elasticsearch/action/admin/indices/warmer/delete/TransportDeleteWarmerAction.java @@ -0,0 +1,175 @@ +/* + * Licensed to Elastic Search and Shay Banon under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. Elastic Search licenses this + * file to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.elasticsearch.action.admin.indices.warmer.delete; + +import com.google.common.collect.Lists; +import org.elasticsearch.ElasticSearchException; +import org.elasticsearch.action.ActionListener; +import org.elasticsearch.action.support.master.TransportMasterNodeOperationAction; +import org.elasticsearch.cluster.ClusterService; +import org.elasticsearch.cluster.ClusterState; +import org.elasticsearch.cluster.ProcessedClusterStateUpdateTask; +import org.elasticsearch.cluster.block.ClusterBlockException; +import org.elasticsearch.cluster.block.ClusterBlockLevel; +import org.elasticsearch.cluster.metadata.IndexMetaData; +import org.elasticsearch.cluster.metadata.MetaData; +import org.elasticsearch.common.inject.Inject; +import org.elasticsearch.common.regex.Regex; +import org.elasticsearch.common.settings.Settings; +import org.elasticsearch.index.Index; +import org.elasticsearch.indices.IndexMissingException; +import org.elasticsearch.search.warmer.IndexWarmerMissingException; +import org.elasticsearch.search.warmer.IndexWarmersMetaData; +import org.elasticsearch.threadpool.ThreadPool; +import org.elasticsearch.transport.TransportService; + +import java.util.List; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.atomic.AtomicReference; + +/** + * Delete index warmer. + */ +public class TransportDeleteWarmerAction extends TransportMasterNodeOperationAction { + + @Inject + public TransportDeleteWarmerAction(Settings settings, TransportService transportService, ClusterService clusterService, ThreadPool threadPool) { + super(settings, transportService, clusterService, threadPool); + } + + @Override + protected String executor() { + return ThreadPool.Names.MANAGEMENT; + } + + @Override + protected String transportAction() { + return DeleteWarmerAction.NAME; + } + + @Override + protected DeleteWarmerRequest newRequest() { + return new DeleteWarmerRequest(); + } + + @Override + protected DeleteWarmerResponse newResponse() { + return new DeleteWarmerResponse(); + } + + @Override + protected void doExecute(DeleteWarmerRequest request, ActionListener listener) { + // update to concrete indices + request.indices(clusterService.state().metaData().concreteIndices(request.indices())); + super.doExecute(request, listener); + } + + @Override + protected ClusterBlockException checkBlock(DeleteWarmerRequest request, ClusterState state) { + return state.blocks().indicesBlockedException(ClusterBlockLevel.METADATA, request.indices()); + } + + @Override + protected DeleteWarmerResponse masterOperation(final DeleteWarmerRequest request, ClusterState state) throws ElasticSearchException { + final AtomicReference failureRef = new AtomicReference(); + final CountDownLatch latch = new CountDownLatch(1); + + clusterService.submitStateUpdateTask("delete_warmer [" + request.name() + "]", new ProcessedClusterStateUpdateTask() { + @Override + public ClusterState execute(ClusterState currentState) { + MetaData.Builder mdBuilder = MetaData.builder().metaData(currentState.metaData()); + + boolean globalFoundAtLeastOne = false; + for (String index : request.indices()) { + IndexMetaData indexMetaData = currentState.metaData().index(index); + if (indexMetaData == null) { + throw new IndexMissingException(new Index(index)); + } + IndexWarmersMetaData warmers = indexMetaData.custom(IndexWarmersMetaData.TYPE); + if (warmers != null) { + List entries = Lists.newArrayList(); + for (IndexWarmersMetaData.Entry entry : warmers.entries()) { + if (request.name() == null || Regex.simpleMatch(request.name(), entry.name())) { + globalFoundAtLeastOne = true; + // don't add it... + } else { + entries.add(entry); + } + } + // a change, update it... + if (entries.size() != warmers.entries().size()) { + warmers = new IndexWarmersMetaData(entries.toArray(new IndexWarmersMetaData.Entry[entries.size()])); + IndexMetaData.Builder indexBuilder = IndexMetaData.newIndexMetaDataBuilder(indexMetaData).putCustom(IndexWarmersMetaData.TYPE, warmers); + mdBuilder.put(indexBuilder); + } + } + } + + if (!globalFoundAtLeastOne) { + if (request.name() == null) { + // full match, just return with no failure + return currentState; + } + throw new IndexWarmerMissingException(request.name()); + } + + if (logger.isInfoEnabled()) { + for (String index : request.indices()) { + IndexMetaData indexMetaData = currentState.metaData().index(index); + if (indexMetaData == null) { + throw new IndexMissingException(new Index(index)); + } + IndexWarmersMetaData warmers = indexMetaData.custom(IndexWarmersMetaData.TYPE); + if (warmers != null) { + for (IndexWarmersMetaData.Entry entry : warmers.entries()) { + if (Regex.simpleMatch(request.name(), entry.name())) { + logger.info("[{}] delete warmer [{}]", index, entry.name()); + } + } + } + } + } + + return ClusterState.builder().state(currentState).metaData(mdBuilder).build(); + } + + @Override + public void clusterStateProcessed(ClusterState clusterState) { + latch.countDown(); + } + }); + + try { + latch.await(); + } catch (InterruptedException e) { + failureRef.set(e); + } + + if (failureRef.get() != null) { + if (failureRef.get() instanceof ElasticSearchException) { + throw (ElasticSearchException) failureRef.get(); + } else { + throw new ElasticSearchException(failureRef.get().getMessage(), failureRef.get()); + } + } + + return new DeleteWarmerResponse(true); + } +} diff --git a/src/main/java/org/elasticsearch/action/admin/indices/warmer/put/PutWarmerAction.java b/src/main/java/org/elasticsearch/action/admin/indices/warmer/put/PutWarmerAction.java new file mode 100644 index 00000000000..ef1e9945d0f --- /dev/null +++ b/src/main/java/org/elasticsearch/action/admin/indices/warmer/put/PutWarmerAction.java @@ -0,0 +1,45 @@ +/* + * Licensed to ElasticSearch and Shay Banon under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. ElasticSearch licenses this + * file to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.elasticsearch.action.admin.indices.warmer.put; + +import org.elasticsearch.action.admin.indices.IndicesAction; +import org.elasticsearch.client.IndicesAdminClient; + +/** + */ +public class PutWarmerAction extends IndicesAction { + + public static final PutWarmerAction INSTANCE = new PutWarmerAction(); + public static final String NAME = "indices/warmer/put"; + + private PutWarmerAction() { + super(NAME); + } + + @Override + public PutWarmerResponse newResponse() { + return new PutWarmerResponse(); + } + + @Override + public PutWarmerRequestBuilder newRequestBuilder(IndicesAdminClient client) { + return new PutWarmerRequestBuilder(client); + } +} diff --git a/src/main/java/org/elasticsearch/action/admin/indices/warmer/put/PutWarmerRequest.java b/src/main/java/org/elasticsearch/action/admin/indices/warmer/put/PutWarmerRequest.java new file mode 100644 index 00000000000..cc623674aed --- /dev/null +++ b/src/main/java/org/elasticsearch/action/admin/indices/warmer/put/PutWarmerRequest.java @@ -0,0 +1,119 @@ +/* + * Licensed to Elastic Search and Shay Banon under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. Elastic Search licenses this + * file to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.elasticsearch.action.admin.indices.warmer.put; + +import org.elasticsearch.action.ActionRequestValidationException; +import org.elasticsearch.action.search.SearchRequest; +import org.elasticsearch.action.search.SearchRequestBuilder; +import org.elasticsearch.action.support.master.MasterNodeOperationRequest; +import org.elasticsearch.common.Nullable; +import org.elasticsearch.common.io.stream.StreamInput; +import org.elasticsearch.common.io.stream.StreamOutput; + +import java.io.IOException; + +import static org.elasticsearch.action.ValidateActions.addValidationError; + +/** + * A request to put a search warmer. + */ +public class PutWarmerRequest extends MasterNodeOperationRequest { + + private String name; + + private SearchRequest searchRequest; + + PutWarmerRequest() { + + } + + /** + * Constructs a new warmer. + * + * @param name The name of the warmer. + */ + public PutWarmerRequest(String name) { + this.name = name; + } + + /** + * Sets the name of the warmer. + */ + public PutWarmerRequest name(String name) { + this.name = name; + return this; + } + + String name() { + return this.name; + } + + /** + * Sets the search request to warm. + */ + public PutWarmerRequest searchRequest(SearchRequest searchRequest) { + this.searchRequest = searchRequest; + return this; + } + + /** + * Sets the search request to warm. + */ + public PutWarmerRequest searchRequest(SearchRequestBuilder searchRequest) { + this.searchRequest = searchRequest.request(); + return this; + } + + @Nullable + SearchRequest searchRequest() { + return this.searchRequest; + } + + @Override + public ActionRequestValidationException validate() { + ActionRequestValidationException validationException = searchRequest.validate(); + if (name == null) { + validationException = addValidationError("name is missing", validationException); + } + return validationException; + } + + @Override + public void readFrom(StreamInput in) throws IOException { + super.readFrom(in); + name = in.readUTF(); + if (in.readBoolean()) { + searchRequest = new SearchRequest(); + searchRequest.readFrom(in); + } + } + + @Override + public void writeTo(StreamOutput out) throws IOException { + super.writeTo(out); + out.writeUTF(name); + if (searchRequest == null) { + out.writeBoolean(false); + } else { + out.writeBoolean(true); + searchRequest.writeTo(out); + } + } +} diff --git a/src/main/java/org/elasticsearch/action/admin/indices/warmer/put/PutWarmerRequestBuilder.java b/src/main/java/org/elasticsearch/action/admin/indices/warmer/put/PutWarmerRequestBuilder.java new file mode 100644 index 00000000000..065ccd37947 --- /dev/null +++ b/src/main/java/org/elasticsearch/action/admin/indices/warmer/put/PutWarmerRequestBuilder.java @@ -0,0 +1,78 @@ +/* + * Licensed to ElasticSearch and Shay Banon under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. ElasticSearch licenses this + * file to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.elasticsearch.action.admin.indices.warmer.put; + +import org.elasticsearch.action.ActionListener; +import org.elasticsearch.action.admin.indices.support.BaseIndicesRequestBuilder; +import org.elasticsearch.action.search.SearchRequest; +import org.elasticsearch.action.search.SearchRequestBuilder; +import org.elasticsearch.client.IndicesAdminClient; +import org.elasticsearch.common.unit.TimeValue; + +/** + * + */ +public class PutWarmerRequestBuilder extends BaseIndicesRequestBuilder { + + public PutWarmerRequestBuilder(IndicesAdminClient indicesClient, String name) { + super(indicesClient, new PutWarmerRequest().name(name)); + } + + public PutWarmerRequestBuilder(IndicesAdminClient indicesClient) { + super(indicesClient, new PutWarmerRequest()); + } + + /** + * Sets the name of the warmer. + */ + public PutWarmerRequestBuilder setName(String name) { + request.name(name); + return this; + } + + /** + * Sets the search request to use to warm the index when applicable. + */ + public PutWarmerRequestBuilder setSearchRequest(SearchRequest searchRequest) { + request.searchRequest(searchRequest); + return this; + } + + /** + * Sets the search request to use to warm the index when applicable. + */ + public PutWarmerRequestBuilder setSearchRequest(SearchRequestBuilder searchRequest) { + request.searchRequest(searchRequest); + return this; + } + + /** + * Sets the master node timeout in case the master has not yet been discovered. + */ + public PutWarmerRequestBuilder setMasterNodeTimeout(TimeValue timeout) { + request.masterNodeTimeout(timeout); + return this; + } + + @Override + protected void doExecute(ActionListener listener) { + client.putWarmer(request, listener); + } +} \ No newline at end of file diff --git a/src/main/java/org/elasticsearch/action/admin/indices/warmer/put/PutWarmerResponse.java b/src/main/java/org/elasticsearch/action/admin/indices/warmer/put/PutWarmerResponse.java new file mode 100644 index 00000000000..fe0daaf6678 --- /dev/null +++ b/src/main/java/org/elasticsearch/action/admin/indices/warmer/put/PutWarmerResponse.java @@ -0,0 +1,67 @@ +/* + * Licensed to ElasticSearch and Shay Banon under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. ElasticSearch licenses this + * file to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.elasticsearch.action.admin.indices.warmer.put; + +import org.elasticsearch.action.ActionResponse; +import org.elasticsearch.common.io.stream.StreamInput; +import org.elasticsearch.common.io.stream.StreamOutput; +import org.elasticsearch.common.io.stream.Streamable; + +import java.io.IOException; + +/** + * The response of put warmer operation. + */ +public class PutWarmerResponse implements ActionResponse, Streamable { + + private boolean acknowledged; + + PutWarmerResponse() { + + } + + PutWarmerResponse(boolean acknowledged) { + this.acknowledged = acknowledged; + } + + /** + * Has the put warmer been ack'ed. + */ + public boolean acknowledged() { + return acknowledged; + } + + /** + * Has the put warmer been ack'ed. + */ + public boolean getAcknowledged() { + return acknowledged(); + } + + @Override + public void readFrom(StreamInput in) throws IOException { + acknowledged = in.readBoolean(); + } + + @Override + public void writeTo(StreamOutput out) throws IOException { + out.writeBoolean(acknowledged); + } +} \ No newline at end of file diff --git a/src/main/java/org/elasticsearch/action/admin/indices/warmer/put/TransportPutWarmerAction.java b/src/main/java/org/elasticsearch/action/admin/indices/warmer/put/TransportPutWarmerAction.java new file mode 100644 index 00000000000..8de1f07f0dd --- /dev/null +++ b/src/main/java/org/elasticsearch/action/admin/indices/warmer/put/TransportPutWarmerAction.java @@ -0,0 +1,178 @@ +/* + * Licensed to ElasticSearch and Shay Banon under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. ElasticSearch licenses this + * file to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.elasticsearch.action.admin.indices.warmer.put; + +import org.elasticsearch.ElasticSearchException; +import org.elasticsearch.action.search.SearchResponse; +import org.elasticsearch.action.search.TransportSearchAction; +import org.elasticsearch.action.support.master.TransportMasterNodeOperationAction; +import org.elasticsearch.cluster.ClusterService; +import org.elasticsearch.cluster.ClusterState; +import org.elasticsearch.cluster.ProcessedClusterStateUpdateTask; +import org.elasticsearch.cluster.block.ClusterBlockException; +import org.elasticsearch.cluster.block.ClusterBlockLevel; +import org.elasticsearch.cluster.metadata.IndexMetaData; +import org.elasticsearch.cluster.metadata.MetaData; +import org.elasticsearch.common.BytesHolder; +import org.elasticsearch.common.inject.Inject; +import org.elasticsearch.common.settings.Settings; +import org.elasticsearch.index.Index; +import org.elasticsearch.indices.IndexMissingException; +import org.elasticsearch.search.warmer.IndexWarmersMetaData; +import org.elasticsearch.threadpool.ThreadPool; +import org.elasticsearch.transport.TransportService; + +import java.util.ArrayList; +import java.util.Arrays; +import java.util.List; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.atomic.AtomicReference; + +/** + * Put warmer action. + */ +public class TransportPutWarmerAction extends TransportMasterNodeOperationAction { + + private final TransportSearchAction searchAction; + + @Inject + public TransportPutWarmerAction(Settings settings, TransportService transportService, ClusterService clusterService, ThreadPool threadPool, + TransportSearchAction searchAction) { + super(settings, transportService, clusterService, threadPool); + this.searchAction = searchAction; + } + + @Override + protected String executor() { + return ThreadPool.Names.MANAGEMENT; + } + + @Override + protected String transportAction() { + return PutWarmerAction.NAME; + } + + @Override + protected PutWarmerRequest newRequest() { + return new PutWarmerRequest(); + } + + @Override + protected PutWarmerResponse newResponse() { + return new PutWarmerResponse(); + } + + @Override + protected ClusterBlockException checkBlock(PutWarmerRequest request, ClusterState state) { + String[] concreteIndices = clusterService.state().metaData().concreteIndices(request.searchRequest().indices()); + return state.blocks().indicesBlockedException(ClusterBlockLevel.METADATA, concreteIndices); + } + + @Override + protected PutWarmerResponse masterOperation(final PutWarmerRequest request, ClusterState state) throws ElasticSearchException { + + // first execute the search request, see that its ok... + SearchResponse searchResponse = searchAction.execute(request.searchRequest()).actionGet(); + // check no shards errors + //TODO: better failure to raise... + if (searchResponse.failedShards() > 0) { + throw new ElasticSearchException("search failed with failed shards: " + Arrays.toString(searchResponse.shardFailures())); + } + + // all is well, continue to the cluster service + + final AtomicReference failureRef = new AtomicReference(); + final CountDownLatch latch = new CountDownLatch(1); + + clusterService.submitStateUpdateTask("put_warmer [" + request.name() + "]", new ProcessedClusterStateUpdateTask() { + @Override + public ClusterState execute(ClusterState currentState) { + MetaData metaData = currentState.metaData(); + String[] concreteIndices = metaData.concreteIndices(request.searchRequest().indices()); + + + BytesHolder source = null; + if (request.searchRequest().source() != null && request.searchRequest().source().length > 0) { + source = new BytesHolder(request.searchRequest().source(), request.searchRequest().sourceOffset(), request.searchRequest().sourceLength()); + } else if (request.searchRequest().extraSource() != null && request.searchRequest().extraSource().length > 0) { + source = new BytesHolder(request.searchRequest().extraSource(), request.searchRequest().extraSourceOffset(), request.searchRequest().extraSourceLength()); + } + + // now replace it on the metadata + MetaData.Builder mdBuilder = MetaData.builder().metaData(currentState.metaData()); + + for (String index : concreteIndices) { + IndexMetaData indexMetaData = metaData.index(index); + if (indexMetaData == null) { + throw new IndexMissingException(new Index(index)); + } + IndexWarmersMetaData warmers = indexMetaData.custom(IndexWarmersMetaData.TYPE); + if (warmers == null) { + logger.info("[{}] putting warmer [{}]", index, request.name()); + warmers = new IndexWarmersMetaData(new IndexWarmersMetaData.Entry(request.name(), request.searchRequest().types(), source)); + } else { + boolean found = false; + List entries = new ArrayList(warmers.entries().size() + 1); + for (IndexWarmersMetaData.Entry entry : warmers.entries()) { + if (entry.name().equals(request.name())) { + found = true; + entries.add(new IndexWarmersMetaData.Entry(request.name(), request.searchRequest().types(), source)); + } else { + entries.add(entry); + } + } + if (!found) { + logger.info("[{}] put warmer [{}]", index, request.name()); + entries.add(new IndexWarmersMetaData.Entry(request.name(), request.searchRequest().types(), source)); + } else { + logger.info("[{}] update warmer [{}]", index, request.name()); + } + warmers = new IndexWarmersMetaData(entries.toArray(new IndexWarmersMetaData.Entry[entries.size()])); + } + IndexMetaData.Builder indexBuilder = IndexMetaData.newIndexMetaDataBuilder(indexMetaData).putCustom(IndexWarmersMetaData.TYPE, warmers); + mdBuilder.put(indexBuilder); + } + + return ClusterState.builder().state(currentState).metaData(mdBuilder).build(); + } + + @Override + public void clusterStateProcessed(ClusterState clusterState) { + latch.countDown(); + } + }); + + try { + latch.await(); + } catch (InterruptedException e) { + failureRef.set(e); + } + + if (failureRef.get() != null) { + if (failureRef.get() instanceof ElasticSearchException) { + throw (ElasticSearchException) failureRef.get(); + } else { + throw new ElasticSearchException(failureRef.get().getMessage(), failureRef.get()); + } + } + + return new PutWarmerResponse(true); + } +} \ No newline at end of file diff --git a/src/main/java/org/elasticsearch/client/IndicesAdminClient.java b/src/main/java/org/elasticsearch/client/IndicesAdminClient.java index 4e852109aa8..588b154b8b8 100644 --- a/src/main/java/org/elasticsearch/client/IndicesAdminClient.java +++ b/src/main/java/org/elasticsearch/client/IndicesAdminClient.java @@ -84,6 +84,12 @@ import org.elasticsearch.action.admin.indices.template.put.PutIndexTemplateRespo import org.elasticsearch.action.admin.indices.validate.query.ValidateQueryRequest; import org.elasticsearch.action.admin.indices.validate.query.ValidateQueryRequestBuilder; import org.elasticsearch.action.admin.indices.validate.query.ValidateQueryResponse; +import org.elasticsearch.action.admin.indices.warmer.delete.DeleteWarmerRequest; +import org.elasticsearch.action.admin.indices.warmer.delete.DeleteWarmerRequestBuilder; +import org.elasticsearch.action.admin.indices.warmer.delete.DeleteWarmerResponse; +import org.elasticsearch.action.admin.indices.warmer.put.PutWarmerRequest; +import org.elasticsearch.action.admin.indices.warmer.put.PutWarmerRequestBuilder; +import org.elasticsearch.action.admin.indices.warmer.put.PutWarmerResponse; import org.elasticsearch.common.Nullable; /** @@ -570,4 +576,34 @@ public interface IndicesAdminClient { * Validate a query for correctness. */ ValidateQueryRequestBuilder prepareValidateQuery(String... indices); + + /** + * Puts an index search warmer to be applies when applicable. + */ + ActionFuture putWarmer(PutWarmerRequest request); + + /** + * Puts an index search warmer to be applies when applicable. + */ + void putWarmer(PutWarmerRequest request, ActionListener listener); + + /** + * Puts an index search warmer to be applies when applicable. + */ + PutWarmerRequestBuilder preparePutWarmer(String name); + + /** + * Deletes an index warmer. + */ + ActionFuture deleteWarmer(DeleteWarmerRequest request); + + /** + * Deletes an index warmer. + */ + void deleteWarmer(DeleteWarmerRequest request, ActionListener listener); + + /** + * Deletes an index warmer. + */ + DeleteWarmerRequestBuilder prepareDeleteWarmer(); } diff --git a/src/main/java/org/elasticsearch/client/support/AbstractIndicesAdminClient.java b/src/main/java/org/elasticsearch/client/support/AbstractIndicesAdminClient.java index 61f16aa6a7f..54ef607684c 100644 --- a/src/main/java/org/elasticsearch/client/support/AbstractIndicesAdminClient.java +++ b/src/main/java/org/elasticsearch/client/support/AbstractIndicesAdminClient.java @@ -105,6 +105,14 @@ import org.elasticsearch.action.admin.indices.validate.query.ValidateQueryAction import org.elasticsearch.action.admin.indices.validate.query.ValidateQueryRequest; import org.elasticsearch.action.admin.indices.validate.query.ValidateQueryRequestBuilder; import org.elasticsearch.action.admin.indices.validate.query.ValidateQueryResponse; +import org.elasticsearch.action.admin.indices.warmer.delete.DeleteWarmerAction; +import org.elasticsearch.action.admin.indices.warmer.delete.DeleteWarmerRequest; +import org.elasticsearch.action.admin.indices.warmer.delete.DeleteWarmerRequestBuilder; +import org.elasticsearch.action.admin.indices.warmer.delete.DeleteWarmerResponse; +import org.elasticsearch.action.admin.indices.warmer.put.PutWarmerAction; +import org.elasticsearch.action.admin.indices.warmer.put.PutWarmerRequest; +import org.elasticsearch.action.admin.indices.warmer.put.PutWarmerRequestBuilder; +import org.elasticsearch.action.admin.indices.warmer.put.PutWarmerResponse; import org.elasticsearch.client.internal.InternalIndicesAdminClient; import org.elasticsearch.common.Nullable; @@ -437,4 +445,34 @@ public abstract class AbstractIndicesAdminClient implements InternalIndicesAdmin public ValidateQueryRequestBuilder prepareValidateQuery(String... indices) { return new ValidateQueryRequestBuilder(this).setIndices(indices); } + + @Override + public ActionFuture putWarmer(PutWarmerRequest request) { + return execute(PutWarmerAction.INSTANCE, request); + } + + @Override + public void putWarmer(PutWarmerRequest request, ActionListener listener) { + execute(PutWarmerAction.INSTANCE, request, listener); + } + + @Override + public PutWarmerRequestBuilder preparePutWarmer(String name) { + return new PutWarmerRequestBuilder(this, name); + } + + @Override + public ActionFuture deleteWarmer(DeleteWarmerRequest request) { + return execute(DeleteWarmerAction.INSTANCE, request); + } + + @Override + public void deleteWarmer(DeleteWarmerRequest request, ActionListener listener) { + execute(DeleteWarmerAction.INSTANCE, request, listener); + } + + @Override + public DeleteWarmerRequestBuilder prepareDeleteWarmer() { + return new DeleteWarmerRequestBuilder(this); + } } diff --git a/src/main/java/org/elasticsearch/cluster/metadata/IndexMetaData.java b/src/main/java/org/elasticsearch/cluster/metadata/IndexMetaData.java index ace9836bf43..24e751f16f2 100644 --- a/src/main/java/org/elasticsearch/cluster/metadata/IndexMetaData.java +++ b/src/main/java/org/elasticsearch/cluster/metadata/IndexMetaData.java @@ -40,6 +40,7 @@ import org.elasticsearch.common.xcontent.XContentBuilder; import org.elasticsearch.common.xcontent.XContentFactory; import org.elasticsearch.common.xcontent.XContentParser; import org.elasticsearch.rest.RestStatus; +import org.elasticsearch.search.warmer.IndexWarmersMetaData; import java.io.IOException; import java.util.Arrays; @@ -66,12 +67,17 @@ public class IndexMetaData { T fromXContent(XContentParser parser) throws IOException; - void toXContent(T customIndexMetaData, XContentBuilder builder, ToXContent.Params params); + void toXContent(T customIndexMetaData, XContentBuilder builder, ToXContent.Params params) throws IOException; } } public static Map customFactories = new HashMap(); + static { + // register non plugin custom metadata + registerFactory(IndexWarmersMetaData.TYPE, IndexWarmersMetaData.FACTORY); + } + /** * Register a custom index meta data factory. Make sure to call it from a static block. */ @@ -296,6 +302,10 @@ public class IndexMetaData { return this.customs; } + public T custom(String type) { + return (T) customs.get(type); + } + @Nullable public DiscoveryNodeFilters includeFilters() { return includeFilters; @@ -514,7 +524,7 @@ public class IndexMetaData { builder.endArray(); for (Map.Entry entry : indexMetaData.customs().entrySet()) { - builder.startObject(entry.getKey()); + builder.startObject(entry.getKey(), XContentBuilder.FieldCaseConversion.NONE); lookupFactorySafe(entry.getKey()).toXContent(entry.getValue(), builder, params); builder.endObject(); } diff --git a/src/main/java/org/elasticsearch/cluster/service/InternalClusterService.java b/src/main/java/org/elasticsearch/cluster/service/InternalClusterService.java index 299ad6bdcdc..47965c9f58b 100644 --- a/src/main/java/org/elasticsearch/cluster/service/InternalClusterService.java +++ b/src/main/java/org/elasticsearch/cluster/service/InternalClusterService.java @@ -220,6 +220,9 @@ public class InternalClusterService extends AbstractLifecycleComponent readMap() throws IOException { return (Map) readGenericValue(); diff --git a/src/main/java/org/elasticsearch/common/io/stream/StreamOutput.java b/src/main/java/org/elasticsearch/common/io/stream/StreamOutput.java index 637506a2002..5ea6774f24e 100644 --- a/src/main/java/org/elasticsearch/common/io/stream/StreamOutput.java +++ b/src/main/java/org/elasticsearch/common/io/stream/StreamOutput.java @@ -203,6 +203,13 @@ public abstract class StreamOutput extends OutputStream { writeBytes(b, off, len); } + public void writeStringArray(String[] array) throws IOException { + writeVInt(array.length); + for (String s : array) { + writeUTF(s); + } + } + public void writeMap(@Nullable Map map) throws IOException { writeGenericValue(map); } diff --git a/src/main/java/org/elasticsearch/common/xcontent/XContent.java b/src/main/java/org/elasticsearch/common/xcontent/XContent.java index a92b0fbb3f2..1232e2af6fc 100644 --- a/src/main/java/org/elasticsearch/common/xcontent/XContent.java +++ b/src/main/java/org/elasticsearch/common/xcontent/XContent.java @@ -19,12 +19,12 @@ package org.elasticsearch.common.xcontent; +import org.elasticsearch.common.BytesHolder; + import java.io.*; /** * A generic abstraction on top of handling content, inspired by JSON and pull parsing. - * - * */ public interface XContent { @@ -65,6 +65,11 @@ public interface XContent { */ XContentParser createParser(byte[] data, int offset, int length) throws IOException; + /** + * Creates a parser over the provided bytes. + */ + XContentParser createParser(BytesHolder bytes) throws IOException; + /** * Creates a parser over the provided reader. */ diff --git a/src/main/java/org/elasticsearch/common/xcontent/XContentFactory.java b/src/main/java/org/elasticsearch/common/xcontent/XContentFactory.java index 4e44d03ea82..7cd950187cf 100644 --- a/src/main/java/org/elasticsearch/common/xcontent/XContentFactory.java +++ b/src/main/java/org/elasticsearch/common/xcontent/XContentFactory.java @@ -22,6 +22,7 @@ package org.elasticsearch.common.xcontent; import org.codehaus.jackson.smile.SmileConstants; import org.elasticsearch.ElasticSearchIllegalArgumentException; import org.elasticsearch.ElasticSearchParseException; +import org.elasticsearch.common.BytesHolder; import org.elasticsearch.common.xcontent.json.JsonXContent; import org.elasticsearch.common.xcontent.smile.SmileXContent; @@ -32,8 +33,6 @@ import java.util.Arrays; /** * A one stop to use {@link org.elasticsearch.common.xcontent.XContent} and {@link XContentBuilder}. - * - * */ public class XContentFactory { @@ -138,6 +137,13 @@ public class XContentFactory { return xContent(data, 0, data.length); } + /** + * Guesses the content type based on the provided bytes. + */ + public static XContent xContent(BytesHolder bytes) { + return xContent(bytes.bytes(), bytes.offset(), bytes.length()); + } + /** * Guesses the content type based on the provided bytes. */ diff --git a/src/main/java/org/elasticsearch/common/xcontent/json/JsonXContent.java b/src/main/java/org/elasticsearch/common/xcontent/json/JsonXContent.java index 35498c38eb8..f16a332eeef 100644 --- a/src/main/java/org/elasticsearch/common/xcontent/json/JsonXContent.java +++ b/src/main/java/org/elasticsearch/common/xcontent/json/JsonXContent.java @@ -23,6 +23,7 @@ import org.codehaus.jackson.JsonEncoding; import org.codehaus.jackson.JsonFactory; import org.codehaus.jackson.JsonGenerator; import org.codehaus.jackson.JsonParser; +import org.elasticsearch.common.BytesHolder; import org.elasticsearch.common.io.FastStringReader; import org.elasticsearch.common.xcontent.*; @@ -30,8 +31,6 @@ import java.io.*; /** * A JSON based content implementation using Jackson. - * - * */ public class JsonXContent implements XContent { @@ -92,6 +91,11 @@ public class JsonXContent implements XContent { return new JsonXContentParser(jsonFactory.createJsonParser(data, offset, length)); } + @Override + public XContentParser createParser(BytesHolder bytes) throws IOException { + return createParser(bytes.bytes(), bytes.offset(), bytes.length()); + } + @Override public XContentParser createParser(Reader reader) throws IOException { return new JsonXContentParser(jsonFactory.createJsonParser(reader)); diff --git a/src/main/java/org/elasticsearch/common/xcontent/smile/SmileXContent.java b/src/main/java/org/elasticsearch/common/xcontent/smile/SmileXContent.java index 91a67a33ad7..8bdf9544586 100644 --- a/src/main/java/org/elasticsearch/common/xcontent/smile/SmileXContent.java +++ b/src/main/java/org/elasticsearch/common/xcontent/smile/SmileXContent.java @@ -22,6 +22,7 @@ package org.elasticsearch.common.xcontent.smile; import org.codehaus.jackson.JsonEncoding; import org.codehaus.jackson.smile.SmileFactory; import org.codehaus.jackson.smile.SmileGenerator; +import org.elasticsearch.common.BytesHolder; import org.elasticsearch.common.io.FastStringReader; import org.elasticsearch.common.xcontent.*; import org.elasticsearch.common.xcontent.json.JsonXContentParser; @@ -30,8 +31,6 @@ import java.io.*; /** * A JSON based content implementation using Jackson. - * - * */ public class SmileXContent implements XContent { @@ -91,6 +90,11 @@ public class SmileXContent implements XContent { return new SmileXContentParser(smileFactory.createJsonParser(data, offset, length)); } + @Override + public XContentParser createParser(BytesHolder bytes) throws IOException { + return createParser(bytes.bytes(), bytes.offset(), bytes.length()); + } + @Override public XContentParser createParser(Reader reader) throws IOException { return new JsonXContentParser(smileFactory.createJsonParser(reader)); diff --git a/src/main/java/org/elasticsearch/index/engine/robin/RobinEngine.java b/src/main/java/org/elasticsearch/index/engine/robin/RobinEngine.java index 1671b06dae3..0b979ffcac8 100644 --- a/src/main/java/org/elasticsearch/index/engine/robin/RobinEngine.java +++ b/src/main/java/org/elasticsearch/index/engine/robin/RobinEngine.java @@ -55,6 +55,7 @@ import org.elasticsearch.index.similarity.SimilarityService; import org.elasticsearch.index.store.Store; import org.elasticsearch.index.translog.Translog; import org.elasticsearch.index.translog.TranslogStreams; +import org.elasticsearch.indices.warmer.IndicesWarmer; import org.elasticsearch.indices.warmer.InternalIndicesWarmer; import org.elasticsearch.threadpool.ThreadPool; @@ -157,7 +158,7 @@ public class RobinEngine extends AbstractIndexShardComponent implements Engine { @Inject public RobinEngine(ShardId shardId, @IndexSettings Settings indexSettings, ThreadPool threadPool, - IndexSettingsService indexSettingsService, @Nullable InternalIndicesWarmer warmer, + IndexSettingsService indexSettingsService, @Nullable IndicesWarmer warmer, Store store, SnapshotDeletionPolicy deletionPolicy, Translog translog, MergePolicyProvider mergePolicyProvider, MergeSchedulerProvider mergeScheduler, AnalysisService analysisService, SimilarityService similarityService, @@ -175,7 +176,7 @@ public class RobinEngine extends AbstractIndexShardComponent implements Engine { this.threadPool = threadPool; this.indexSettingsService = indexSettingsService; - this.warmer = warmer; + this.warmer = (InternalIndicesWarmer) warmer; this.store = store; this.deletionPolicy = deletionPolicy; this.translog = translog; diff --git a/src/main/java/org/elasticsearch/rest/action/RestActionModule.java b/src/main/java/org/elasticsearch/rest/action/RestActionModule.java index a211ee97d63..2472c9c33ee 100644 --- a/src/main/java/org/elasticsearch/rest/action/RestActionModule.java +++ b/src/main/java/org/elasticsearch/rest/action/RestActionModule.java @@ -56,6 +56,9 @@ import org.elasticsearch.rest.action.admin.indices.template.delete.RestDeleteInd import org.elasticsearch.rest.action.admin.indices.template.get.RestGetIndexTemplateAction; import org.elasticsearch.rest.action.admin.indices.template.put.RestPutIndexTemplateAction; import org.elasticsearch.rest.action.admin.indices.validate.query.RestValidateQueryAction; +import org.elasticsearch.rest.action.admin.indices.warmer.delete.RestDeleteWarmerAction; +import org.elasticsearch.rest.action.admin.indices.warmer.get.RestGetWarmerAction; +import org.elasticsearch.rest.action.admin.indices.warmer.put.RestPutWarmerAction; import org.elasticsearch.rest.action.bulk.RestBulkAction; import org.elasticsearch.rest.action.count.RestCountAction; import org.elasticsearch.rest.action.delete.RestDeleteAction; @@ -120,6 +123,10 @@ public class RestActionModule extends AbstractModule { bind(RestPutIndexTemplateAction.class).asEagerSingleton(); bind(RestDeleteIndexTemplateAction.class).asEagerSingleton(); + bind(RestPutWarmerAction.class).asEagerSingleton(); + bind(RestDeleteWarmerAction.class).asEagerSingleton(); + bind(RestGetWarmerAction.class).asEagerSingleton(); + bind(RestPutMappingAction.class).asEagerSingleton(); bind(RestDeleteMappingAction.class).asEagerSingleton(); bind(RestGetMappingAction.class).asEagerSingleton(); diff --git a/src/main/java/org/elasticsearch/rest/action/admin/indices/warmer/delete/RestDeleteWarmerAction.java b/src/main/java/org/elasticsearch/rest/action/admin/indices/warmer/delete/RestDeleteWarmerAction.java new file mode 100644 index 00000000000..38d104bf930 --- /dev/null +++ b/src/main/java/org/elasticsearch/rest/action/admin/indices/warmer/delete/RestDeleteWarmerAction.java @@ -0,0 +1,79 @@ +/* + * Licensed to Elastic Search and Shay Banon under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. Elastic Search licenses this + * file to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.elasticsearch.rest.action.admin.indices.warmer.delete; + +import org.elasticsearch.action.ActionListener; +import org.elasticsearch.action.admin.indices.warmer.delete.DeleteWarmerRequest; +import org.elasticsearch.action.admin.indices.warmer.delete.DeleteWarmerResponse; +import org.elasticsearch.client.Client; +import org.elasticsearch.common.inject.Inject; +import org.elasticsearch.common.settings.Settings; +import org.elasticsearch.common.xcontent.XContentBuilder; +import org.elasticsearch.rest.*; +import org.elasticsearch.rest.action.support.RestActions; +import org.elasticsearch.rest.action.support.RestXContentBuilder; + +import java.io.IOException; + +import static org.elasticsearch.rest.RestRequest.Method.DELETE; +import static org.elasticsearch.rest.RestStatus.OK; + +/** + */ +public class RestDeleteWarmerAction extends BaseRestHandler { + + @Inject + public RestDeleteWarmerAction(Settings settings, Client client, RestController controller) { + super(settings, client); + controller.registerHandler(DELETE, "/{index}/_warmer", this); + controller.registerHandler(DELETE, "/{index}/_warmer/{name}", this); + controller.registerHandler(DELETE, "/{index}/{type}/_warmer/{name}", this); + } + + @Override + public void handleRequest(final RestRequest request, final RestChannel channel) { + DeleteWarmerRequest deleteWarmerRequest = new DeleteWarmerRequest(request.param("name")) + .indices(RestActions.splitIndices(request.param("index"))); + client.admin().indices().deleteWarmer(deleteWarmerRequest, new ActionListener() { + @Override + public void onResponse(DeleteWarmerResponse response) { + try { + XContentBuilder builder = RestXContentBuilder.restContentBuilder(request); + builder.startObject() + .field("ok", true) + .field("acknowledged", response.acknowledged()); + builder.endObject(); + channel.sendResponse(new XContentRestResponse(request, OK, builder)); + } catch (IOException e) { + onFailure(e); + } + } + + @Override + public void onFailure(Throwable e) { + try { + channel.sendResponse(new XContentThrowableRestResponse(request, e)); + } catch (IOException e1) { + logger.error("Failed to send failure response", e1); + } + } + }); + } +} diff --git a/src/main/java/org/elasticsearch/rest/action/admin/indices/warmer/get/RestGetWarmerAction.java b/src/main/java/org/elasticsearch/rest/action/admin/indices/warmer/get/RestGetWarmerAction.java new file mode 100644 index 00000000000..7052a01f8cd --- /dev/null +++ b/src/main/java/org/elasticsearch/rest/action/admin/indices/warmer/get/RestGetWarmerAction.java @@ -0,0 +1,137 @@ +/* + * Licensed to Elastic Search and Shay Banon under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. Elastic Search licenses this + * file to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.elasticsearch.rest.action.admin.indices.warmer.get; + +import org.elasticsearch.action.ActionListener; +import org.elasticsearch.action.admin.cluster.state.ClusterStateRequest; +import org.elasticsearch.action.admin.cluster.state.ClusterStateResponse; +import org.elasticsearch.client.Client; +import org.elasticsearch.client.Requests; +import org.elasticsearch.cluster.metadata.IndexMetaData; +import org.elasticsearch.cluster.metadata.MetaData; +import org.elasticsearch.common.inject.Inject; +import org.elasticsearch.common.regex.Regex; +import org.elasticsearch.common.settings.Settings; +import org.elasticsearch.common.xcontent.XContentBuilder; +import org.elasticsearch.index.Index; +import org.elasticsearch.indices.IndexMissingException; +import org.elasticsearch.rest.*; +import org.elasticsearch.rest.action.support.RestXContentBuilder; +import org.elasticsearch.search.warmer.IndexWarmerMissingException; +import org.elasticsearch.search.warmer.IndexWarmersMetaData; + +import java.io.IOException; + +import static org.elasticsearch.rest.RestRequest.Method.GET; +import static org.elasticsearch.rest.RestStatus.OK; +import static org.elasticsearch.rest.action.support.RestActions.splitIndices; + +/** + * + */ +public class RestGetWarmerAction extends BaseRestHandler { + + @Inject + public RestGetWarmerAction(Settings settings, Client client, RestController controller) { + super(settings, client); + + controller.registerHandler(GET, "/{index}/_warmer", this); + controller.registerHandler(GET, "/{index}/_warmer/{name}", this); + controller.registerHandler(GET, "/{index}/{type}/_warmer/{name}", this); + } + + @Override + public void handleRequest(final RestRequest request, final RestChannel channel) { + final String[] indices = splitIndices(request.param("index")); + final String name = request.param("name"); + + ClusterStateRequest clusterStateRequest = Requests.clusterStateRequest() + .filterAll() + .filterMetaData(false) + .filteredIndices(indices); + + client.admin().cluster().state(clusterStateRequest, new ActionListener() { + @Override + public void onResponse(ClusterStateResponse response) { + try { + MetaData metaData = response.state().metaData(); + + if (indices.length == 1 && metaData.indices().isEmpty()) { + channel.sendResponse(new XContentThrowableRestResponse(request, new IndexMissingException(new Index(indices[0])))); + return; + } + + XContentBuilder builder = RestXContentBuilder.restContentBuilder(request); + builder.startObject(); + + boolean wroteOne = false; + for (IndexMetaData indexMetaData : metaData) { + IndexWarmersMetaData warmers = indexMetaData.custom(IndexWarmersMetaData.TYPE); + if (warmers == null) { + continue; + } + + boolean foundOne = false; + for (IndexWarmersMetaData.Entry entry : warmers.entries()) { + if (name == null || Regex.simpleMatch(name, entry.name())) { + foundOne = true; + wroteOne = true; + break; + } + } + + if (foundOne) { + builder.startObject(indexMetaData.index(), XContentBuilder.FieldCaseConversion.NONE); + builder.startObject(IndexWarmersMetaData.TYPE, XContentBuilder.FieldCaseConversion.NONE); + for (IndexWarmersMetaData.Entry entry : warmers.entries()) { + if (name == null || Regex.simpleMatch(name, entry.name())) { + IndexWarmersMetaData.FACTORY.toXContent(entry, builder, request); + } + } + builder.endObject(); + builder.endObject(); + } + } + + builder.endObject(); + + if (!wroteOne && name != null) { + // did not find any... + channel.sendResponse(new XContentThrowableRestResponse(request, new IndexWarmerMissingException(name))); + return; + } + + channel.sendResponse(new XContentRestResponse(request, OK, builder)); + } catch (Exception e) { + onFailure(e); + } + } + + @Override + public void onFailure(Throwable e) { + try { + channel.sendResponse(new XContentThrowableRestResponse(request, e)); + } catch (IOException e1) { + logger.error("Failed to send failure response", e1); + } + } + }); + } +} diff --git a/src/main/java/org/elasticsearch/rest/action/admin/indices/warmer/put/RestPutWarmerAction.java b/src/main/java/org/elasticsearch/rest/action/admin/indices/warmer/put/RestPutWarmerAction.java new file mode 100644 index 00000000000..8f4432db6cc --- /dev/null +++ b/src/main/java/org/elasticsearch/rest/action/admin/indices/warmer/put/RestPutWarmerAction.java @@ -0,0 +1,82 @@ +/* + * Licensed to Elastic Search and Shay Banon under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. Elastic Search licenses this + * file to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.elasticsearch.rest.action.admin.indices.warmer.put; + +import org.elasticsearch.action.ActionListener; +import org.elasticsearch.action.admin.indices.warmer.put.PutWarmerRequest; +import org.elasticsearch.action.admin.indices.warmer.put.PutWarmerResponse; +import org.elasticsearch.action.search.SearchRequest; +import org.elasticsearch.client.Client; +import org.elasticsearch.common.inject.Inject; +import org.elasticsearch.common.settings.Settings; +import org.elasticsearch.common.xcontent.XContentBuilder; +import org.elasticsearch.rest.*; +import org.elasticsearch.rest.action.support.RestActions; +import org.elasticsearch.rest.action.support.RestXContentBuilder; + +import java.io.IOException; + +import static org.elasticsearch.rest.RestRequest.Method.PUT; +import static org.elasticsearch.rest.RestStatus.OK; + +/** + */ +public class RestPutWarmerAction extends BaseRestHandler { + + @Inject + public RestPutWarmerAction(Settings settings, Client client, RestController controller) { + super(settings, client); + controller.registerHandler(PUT, "/{index}/_warmer/{name}", this); + controller.registerHandler(PUT, "/{index}/{type}/_warmer/{name}", this); + } + + @Override + public void handleRequest(final RestRequest request, final RestChannel channel) { + PutWarmerRequest putWarmerRequest = new PutWarmerRequest(request.param("name")); + SearchRequest searchRequest = new SearchRequest(RestActions.splitIndices(request.param("index"))) + .types(RestActions.splitTypes(request.param("type"))) + .source(request.contentByteArray(), request.contentByteArrayOffset(), request.contentLength(), request.contentUnsafe()); + putWarmerRequest.searchRequest(searchRequest); + client.admin().indices().putWarmer(putWarmerRequest, new ActionListener() { + @Override + public void onResponse(PutWarmerResponse response) { + try { + XContentBuilder builder = RestXContentBuilder.restContentBuilder(request); + builder.startObject() + .field("ok", true) + .field("acknowledged", response.acknowledged()); + builder.endObject(); + channel.sendResponse(new XContentRestResponse(request, OK, builder)); + } catch (IOException e) { + onFailure(e); + } + } + + @Override + public void onFailure(Throwable e) { + try { + channel.sendResponse(new XContentThrowableRestResponse(request, e)); + } catch (IOException e1) { + logger.error("Failed to send failure response", e1); + } + } + }); + } +} diff --git a/src/main/java/org/elasticsearch/search/SearchService.java b/src/main/java/org/elasticsearch/search/SearchService.java index 956ab8b83f4..822ab3a4e53 100644 --- a/src/main/java/org/elasticsearch/search/SearchService.java +++ b/src/main/java/org/elasticsearch/search/SearchService.java @@ -25,6 +25,7 @@ import org.apache.lucene.search.TopDocs; import org.elasticsearch.ElasticSearchException; import org.elasticsearch.action.search.SearchType; import org.elasticsearch.cluster.ClusterService; +import org.elasticsearch.cluster.metadata.IndexMetaData; import org.elasticsearch.common.Nullable; import org.elasticsearch.common.Unicode; import org.elasticsearch.common.component.AbstractLifecycleComponent; @@ -43,6 +44,7 @@ import org.elasticsearch.index.shard.ShardId; import org.elasticsearch.index.shard.service.IndexShard; import org.elasticsearch.indices.IndicesLifecycle; import org.elasticsearch.indices.IndicesService; +import org.elasticsearch.indices.warmer.IndicesWarmer; import org.elasticsearch.script.ScriptService; import org.elasticsearch.search.dfs.CachedDfSource; import org.elasticsearch.search.dfs.DfsPhase; @@ -52,6 +54,7 @@ import org.elasticsearch.search.internal.InternalScrollSearchRequest; import org.elasticsearch.search.internal.InternalSearchRequest; import org.elasticsearch.search.internal.SearchContext; import org.elasticsearch.search.query.*; +import org.elasticsearch.search.warmer.IndexWarmersMetaData; import org.elasticsearch.threadpool.ThreadPool; import java.io.IOException; @@ -73,6 +76,8 @@ public class SearchService extends AbstractLifecycleComponent { private final IndicesService indicesService; + private final IndicesWarmer indicesWarmer; + private final ScriptService scriptService; private final DfsPhase dfsPhase; @@ -96,12 +101,13 @@ public class SearchService extends AbstractLifecycleComponent { private final ImmutableMap elementParsers; @Inject - public SearchService(Settings settings, ClusterService clusterService, IndicesService indicesService, IndicesLifecycle indicesLifecycle, ThreadPool threadPool, + public SearchService(Settings settings, ClusterService clusterService, IndicesService indicesService, IndicesLifecycle indicesLifecycle, IndicesWarmer indicesWarmer, ThreadPool threadPool, ScriptService scriptService, DfsPhase dfsPhase, QueryPhase queryPhase, FetchPhase fetchPhase) { super(settings); this.threadPool = threadPool; this.clusterService = clusterService; this.indicesService = indicesService; + this.indicesWarmer = indicesWarmer; this.scriptService = scriptService; this.dfsPhase = dfsPhase; this.queryPhase = queryPhase; @@ -120,6 +126,8 @@ public class SearchService extends AbstractLifecycleComponent { indicesLifecycle.addListener(indicesLifecycleListener); this.keepAliveReaper = threadPool.scheduleWithFixedDelay(new Reaper(), keepAliveInterval); + + this.indicesWarmer.addListener(new SearchWarmer()); } @Override @@ -454,13 +462,17 @@ public class SearchService extends AbstractLifecycleComponent { return context; } - private SearchContext createContext(InternalSearchRequest request) throws ElasticSearchException { + SearchContext createContext(InternalSearchRequest request) throws ElasticSearchException { + return createContext(request, null); + } + + SearchContext createContext(InternalSearchRequest request, @Nullable Engine.Searcher searcher) throws ElasticSearchException { IndexService indexService = indicesService.indexServiceSafe(request.index()); IndexShard indexShard = indexService.shardSafe(request.shardId()); SearchShardTarget shardTarget = new SearchShardTarget(clusterService.localNode().id(), request.index(), request.shardId()); - Engine.Searcher engineSearcher = indexShard.searcher(); + Engine.Searcher engineSearcher = searcher == null ? indexShard.searcher() : searcher; SearchContext context = new SearchContext(idGenerator.incrementAndGet(), request, shardTarget, engineSearcher, indexService, indexShard, scriptService); SearchContext.setCurrent(context); try { @@ -613,6 +625,44 @@ public class SearchService extends AbstractLifecycleComponent { } } + class SearchWarmer implements IndicesWarmer.Listener { + + @Override + public String executor() { + return ThreadPool.Names.SEARCH; + } + + @Override + public void warm(ShardId shardId, IndexMetaData indexMetaData, Engine.Searcher search) { + IndexWarmersMetaData custom = indexMetaData.custom(IndexWarmersMetaData.TYPE); + if (custom == null) { + return; + } + for (IndexWarmersMetaData.Entry entry : custom.entries()) { + SearchContext context = null; + try { + long now = System.nanoTime(); + InternalSearchRequest request = new InternalSearchRequest(shardId.index().name(), shardId.id(), indexMetaData.numberOfShards(), SearchType.COUNT) + .source(entry.source().bytes(), entry.source().offset(), entry.source().length()) + .types(entry.types()); + context = createContext(request, search); + queryPhase.execute(context); + long took = System.nanoTime() - now; + if (logger.isTraceEnabled()) { + logger.trace("[{}][{}] warmed [{}], took [{}]", shardId.index().name(), shardId.id(), entry.name(), TimeValue.timeValueNanos(took)); + } + } catch (Throwable t) { + logger.warn("[{}][{}] warmer [{}] failed", t, shardId.index().name(), shardId.id(), entry.name()); + } finally { + if (context != null) { + freeContext(context); + cleanContext(context); + } + } + } + } + } + class CleanContextOnIndicesLifecycleListener extends IndicesLifecycle.Listener { @Override diff --git a/src/main/java/org/elasticsearch/search/internal/InternalSearchRequest.java b/src/main/java/org/elasticsearch/search/internal/InternalSearchRequest.java index bb2b5ba22c9..1b7f8ae92b5 100644 --- a/src/main/java/org/elasticsearch/search/internal/InternalSearchRequest.java +++ b/src/main/java/org/elasticsearch/search/internal/InternalSearchRequest.java @@ -177,8 +177,9 @@ public class InternalSearchRequest implements Streamable { return types; } - public void types(String[] types) { + public InternalSearchRequest types(String[] types) { this.types = types; + return this; } @Override diff --git a/src/main/java/org/elasticsearch/search/warmer/IndexWarmerMissingException.java b/src/main/java/org/elasticsearch/search/warmer/IndexWarmerMissingException.java new file mode 100644 index 00000000000..af829a4a4df --- /dev/null +++ b/src/main/java/org/elasticsearch/search/warmer/IndexWarmerMissingException.java @@ -0,0 +1,46 @@ +/* + * Licensed to Elastic Search and Shay Banon under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. Elastic Search licenses this + * file to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.elasticsearch.search.warmer; + +import org.elasticsearch.ElasticSearchException; +import org.elasticsearch.rest.RestStatus; + +/** + * + */ +public class IndexWarmerMissingException extends ElasticSearchException { + + private final String name; + + public IndexWarmerMissingException(String name) { + super("index_warmer [" + name + "] missing"); + this.name = name; + } + + public String name() { + return this.name; + } + + + @Override + public RestStatus status() { + return RestStatus.NOT_FOUND; + } +} diff --git a/src/main/java/org/elasticsearch/search/warmer/IndexWarmersMetaData.java b/src/main/java/org/elasticsearch/search/warmer/IndexWarmersMetaData.java new file mode 100644 index 00000000000..75918153987 --- /dev/null +++ b/src/main/java/org/elasticsearch/search/warmer/IndexWarmersMetaData.java @@ -0,0 +1,182 @@ +/* + * Licensed to ElasticSearch and Shay Banon under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. ElasticSearch licenses this + * file to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.elasticsearch.search.warmer; + +import com.google.common.collect.ImmutableList; +import org.elasticsearch.cluster.metadata.IndexMetaData; +import org.elasticsearch.common.BytesHolder; +import org.elasticsearch.common.Nullable; +import org.elasticsearch.common.Strings; +import org.elasticsearch.common.io.stream.StreamInput; +import org.elasticsearch.common.io.stream.StreamOutput; +import org.elasticsearch.common.xcontent.ToXContent; +import org.elasticsearch.common.xcontent.XContentBuilder; +import org.elasticsearch.common.xcontent.XContentFactory; +import org.elasticsearch.common.xcontent.XContentParser; + +import java.io.IOException; +import java.util.ArrayList; +import java.util.List; +import java.util.Map; + +/** + */ +public class IndexWarmersMetaData implements IndexMetaData.Custom { + + public static final String TYPE = "warmers"; + + public static final Factory FACTORY = new Factory(); + + static { + IndexMetaData.registerFactory(TYPE, FACTORY); + } + + public static class Entry { + private final String name; + private final String[] types; + private final BytesHolder source; + + public Entry(String name, String[] types, BytesHolder source) { + this.name = name; + this.types = types == null ? Strings.EMPTY_ARRAY : types; + this.source = source; + } + + public String name() { + return this.name; + } + + public String[] types() { + return this.types; + } + + @Nullable + public BytesHolder source() { + return this.source; + } + } + + private final ImmutableList entries; + + + public IndexWarmersMetaData(Entry... entries) { + this.entries = ImmutableList.copyOf(entries); + } + + public ImmutableList entries() { + return this.entries; + } + + public static class Factory implements IndexMetaData.Custom.Factory { + + @Override + public String type() { + return TYPE; + } + + @Override + public IndexWarmersMetaData readFrom(StreamInput in) throws IOException { + Entry[] entries = new Entry[in.readVInt()]; + for (int i = 0; i < entries.length; i++) { + entries[i] = new Entry(in.readUTF(), in.readStringArray(), in.readBoolean() ? in.readBytesHolder() : null); + } + return new IndexWarmersMetaData(entries); + } + + @Override + public void writeTo(IndexWarmersMetaData warmers, StreamOutput out) throws IOException { + out.writeVInt(warmers.entries().size()); + for (Entry entry : warmers.entries()) { + out.writeUTF(entry.name()); + out.writeStringArray(entry.types()); + if (entry.source() == null) { + out.writeBoolean(false); + } else { + out.writeBoolean(true); + out.writeBytesHolder(entry.source()); + } + } + } + + @Override + public IndexWarmersMetaData fromXContent(XContentParser parser) throws IOException { + // we get here after we are at warmers token + String currentFieldName = null; + XContentParser.Token token; + List entries = new ArrayList(); + while ((token = parser.nextToken()) != XContentParser.Token.END_OBJECT) { + if (token == XContentParser.Token.FIELD_NAME) { + currentFieldName = parser.currentName(); + } else if (token == XContentParser.Token.START_OBJECT) { + String name = currentFieldName; + List types = new ArrayList(2); + BytesHolder source = null; + while ((token = parser.nextToken()) != XContentParser.Token.END_OBJECT) { + if (token == XContentParser.Token.FIELD_NAME) { + currentFieldName = parser.currentName(); + } else if (token == XContentParser.Token.START_ARRAY) { + if ("types".equals(currentFieldName)) { + while ((token = parser.nextToken()) != XContentParser.Token.END_ARRAY) { + types.add(parser.text()); + } + } + } else if (token == XContentParser.Token.START_OBJECT) { + if ("source".equals(currentFieldName)) { + XContentBuilder builder = XContentFactory.jsonBuilder().map(parser.mapOrdered()); + source = new BytesHolder(builder.underlyingBytes(), 0, builder.underlyingBytesLength()); + } + } else if (token == XContentParser.Token.VALUE_EMBEDDED_OBJECT) { + if ("source".equals(currentFieldName)) { + source = new BytesHolder(parser.binaryValue()); + } + } + } + entries.add(new Entry(name, types.size() == 0 ? Strings.EMPTY_ARRAY : types.toArray(new String[types.size()]), source)); + } + } + return new IndexWarmersMetaData(entries.toArray(new Entry[entries.size()])); + } + + @Override + public void toXContent(IndexWarmersMetaData warmers, XContentBuilder builder, ToXContent.Params params) throws IOException { + //No need, IndexMetaData already writes it + //builder.startObject(TYPE, XContentBuilder.FieldCaseConversion.NONE); + for (Entry entry : warmers.entries()) { + toXContent(entry, builder, params); + } + //No need, IndexMetaData already writes it + //builder.endObject(); + } + + public void toXContent(Entry entry, XContentBuilder builder, ToXContent.Params params) throws IOException { + boolean binary = params.paramAsBoolean("binary", false); + builder.startObject(entry.name(), XContentBuilder.FieldCaseConversion.NONE); + builder.field("types", entry.types()); + builder.field("source"); + if (binary) { + builder.value(entry.source().bytes(), entry.source().offset(), entry.source().length()); + } else { + Map mapping = XContentFactory.xContent(entry.source()).createParser(entry.source()).mapOrderedAndClose(); + builder.map(mapping); + } + builder.endObject(); + } + } +} diff --git a/src/test/java/org/elasticsearch/test/integration/indices/wamer/LocalGatewayIndicesWarmerTests.java b/src/test/java/org/elasticsearch/test/integration/indices/wamer/LocalGatewayIndicesWarmerTests.java new file mode 100644 index 00000000000..52a661ffedb --- /dev/null +++ b/src/test/java/org/elasticsearch/test/integration/indices/wamer/LocalGatewayIndicesWarmerTests.java @@ -0,0 +1,136 @@ +/* + * Licensed to ElasticSearch and Shay Banon under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. ElasticSearch licenses this + * file to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.elasticsearch.test.integration.indices.wamer; + +import org.elasticsearch.action.admin.cluster.health.ClusterHealthResponse; +import org.elasticsearch.cluster.ClusterState; +import org.elasticsearch.common.logging.ESLogger; +import org.elasticsearch.common.logging.Loggers; +import org.elasticsearch.common.settings.ImmutableSettings; +import org.elasticsearch.env.NodeEnvironment; +import org.elasticsearch.gateway.Gateway; +import org.elasticsearch.index.query.QueryBuilders; +import org.elasticsearch.node.internal.InternalNode; +import org.elasticsearch.search.warmer.IndexWarmersMetaData; +import org.elasticsearch.test.integration.AbstractNodesTests; +import org.hamcrest.Matchers; +import org.testng.annotations.AfterMethod; +import org.testng.annotations.Test; + +import static org.elasticsearch.common.settings.ImmutableSettings.settingsBuilder; +import static org.hamcrest.MatcherAssert.assertThat; +import static org.hamcrest.Matchers.equalTo; + +/** + */ +public class LocalGatewayIndicesWarmerTests extends AbstractNodesTests { + + private final ESLogger logger = Loggers.getLogger(LocalGatewayIndicesWarmerTests.class); + + @AfterMethod + public void cleanAndCloseNodes() throws Exception { + for (int i = 0; i < 10; i++) { + if (node("node" + i) != null) { + node("node" + i).stop(); + // since we store (by default) the index snapshot under the gateway, resetting it will reset the index data as well + if (((InternalNode) node("node" + i)).injector().getInstance(NodeEnvironment.class).hasNodeFile()) { + ((InternalNode) node("node" + i)).injector().getInstance(Gateway.class).reset(); + } + } + } + closeAllNodes(); + } + + @Test + public void testStatePersistence() throws Exception { + logger.info("--> cleaning nodes"); + buildNode("node1", settingsBuilder().put("gateway.type", "local")); + buildNode("node2", settingsBuilder().put("gateway.type", "local")); + cleanAndCloseNodes(); + + logger.info("--> starting 1 nodes"); + startNode("node1", settingsBuilder().put("gateway.type", "local")); + + logger.info("--> putting two templates"); + client("node1").admin().indices().prepareCreate("test") + .setSettings(ImmutableSettings.settingsBuilder().put("index.number_of_shards", 1)) + .execute().actionGet(); + + client("node1").admin().cluster().prepareHealth().setWaitForYellowStatus().execute().actionGet(); + + client("node1").admin().indices().preparePutWarmer("warmer_1") + .setSearchRequest(client("node1").prepareSearch("test").setQuery(QueryBuilders.termQuery("field", "value1"))) + .execute().actionGet(); + client("node1").admin().indices().preparePutWarmer("warmer_2") + .setSearchRequest(client("node1").prepareSearch("test").setQuery(QueryBuilders.termQuery("field", "value2"))) + .execute().actionGet(); + + logger.info("--> verify warmers are registered in cluster state"); + ClusterState clusterState = client("node1").admin().cluster().prepareState().execute().actionGet().state(); + IndexWarmersMetaData warmersMetaData = clusterState.metaData().index("test").custom(IndexWarmersMetaData.TYPE); + assertThat(warmersMetaData, Matchers.notNullValue()); + assertThat(warmersMetaData.entries().size(), equalTo(2)); + + logger.info("--> close the node"); + closeNode("node1"); + + logger.info("--> starting the node again..."); + startNode("node1", settingsBuilder().put("gateway.type", "local")); + + ClusterHealthResponse healthResponse = client("node1").admin().cluster().prepareHealth().setWaitForYellowStatus().execute().actionGet(); + assertThat(healthResponse.timedOut(), equalTo(false)); + + logger.info("--> verify warmers are recovered"); + clusterState = client("node1").admin().cluster().prepareState().execute().actionGet().state(); + IndexWarmersMetaData recoveredWarmersMetaData = clusterState.metaData().index("test").custom(IndexWarmersMetaData.TYPE); + assertThat(recoveredWarmersMetaData.entries().size(), equalTo(warmersMetaData.entries().size())); + for (int i = 0; i < warmersMetaData.entries().size(); i++) { + assertThat(recoveredWarmersMetaData.entries().get(i).name(), equalTo(warmersMetaData.entries().get(i).name())); + assertThat(recoveredWarmersMetaData.entries().get(i).source(), equalTo(warmersMetaData.entries().get(i).source())); + } + + logger.info("--> delete warmer warmer_1"); + client("node1").admin().indices().prepareDeleteWarmer().setIndices("test").setName("warmer_1").execute().actionGet(); + + logger.info("--> verify warmers (delete) are registered in cluster state"); + clusterState = client("node1").admin().cluster().prepareState().execute().actionGet().state(); + warmersMetaData = clusterState.metaData().index("test").custom(IndexWarmersMetaData.TYPE); + assertThat(warmersMetaData, Matchers.notNullValue()); + assertThat(warmersMetaData.entries().size(), equalTo(1)); + + logger.info("--> close the node"); + closeNode("node1"); + + logger.info("--> starting the node again..."); + startNode("node1", settingsBuilder().put("gateway.type", "local")); + + healthResponse = client("node1").admin().cluster().prepareHealth().setWaitForYellowStatus().execute().actionGet(); + assertThat(healthResponse.timedOut(), equalTo(false)); + + logger.info("--> verify warmers are recovered"); + clusterState = client("node1").admin().cluster().prepareState().execute().actionGet().state(); + recoveredWarmersMetaData = clusterState.metaData().index("test").custom(IndexWarmersMetaData.TYPE); + assertThat(recoveredWarmersMetaData.entries().size(), equalTo(warmersMetaData.entries().size())); + for (int i = 0; i < warmersMetaData.entries().size(); i++) { + assertThat(recoveredWarmersMetaData.entries().get(i).name(), equalTo(warmersMetaData.entries().get(i).name())); + assertThat(recoveredWarmersMetaData.entries().get(i).source(), equalTo(warmersMetaData.entries().get(i).source())); + } + } +} diff --git a/src/test/java/org/elasticsearch/test/integration/indices/wamer/SimpleIndicesWarmerTests.java b/src/test/java/org/elasticsearch/test/integration/indices/wamer/SimpleIndicesWarmerTests.java new file mode 100644 index 00000000000..99675f3072a --- /dev/null +++ b/src/test/java/org/elasticsearch/test/integration/indices/wamer/SimpleIndicesWarmerTests.java @@ -0,0 +1,73 @@ +/* + * Licensed to ElasticSearch and Shay Banon under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. ElasticSearch licenses this + * file to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.elasticsearch.test.integration.indices.wamer; + +import org.elasticsearch.client.Client; +import org.elasticsearch.common.settings.ImmutableSettings; +import org.elasticsearch.index.query.QueryBuilders; +import org.elasticsearch.test.integration.AbstractNodesTests; +import org.testng.annotations.AfterClass; +import org.testng.annotations.BeforeClass; +import org.testng.annotations.Test; + +/** + */ +public class SimpleIndicesWarmerTests extends AbstractNodesTests { + + private Client client; + + @BeforeClass + public void createNodes() throws Exception { + startNode("node1"); + startNode("node2"); + client = getClient(); + } + + @AfterClass + public void closeNodes() { + client.close(); + closeAllNodes(); + } + + protected Client getClient() { + return client("node2"); + } + + @Test + public void simpleWarmerTests() { + client.admin().indices().prepareDelete().execute().actionGet(); + + client.admin().indices().prepareCreate("test") + .setSettings(ImmutableSettings.settingsBuilder().put("index.number_of_shards", 1)) + .execute().actionGet(); + + client.admin().cluster().prepareHealth().setWaitForGreenStatus().execute().actionGet(); + + client.admin().indices().preparePutWarmer("warmer_1") + .setSearchRequest(client.prepareSearch("test").setQuery(QueryBuilders.termQuery("field", "value1"))) + .execute().actionGet(); + client.admin().indices().preparePutWarmer("warmer_2") + .setSearchRequest(client.prepareSearch("test").setQuery(QueryBuilders.termQuery("field", "value2"))) + .execute().actionGet(); + + client.prepareIndex("test", "type1", "1").setSource("field", "value1").setRefresh(true).execute().actionGet(); + client.prepareIndex("test", "type1", "2").setSource("field", "value2").setRefresh(true).execute().actionGet(); + } +}