diff --git a/client/rest-high-level/src/main/java/org/elasticsearch/client/WatcherClient.java b/client/rest-high-level/src/main/java/org/elasticsearch/client/WatcherClient.java index 71247764566..c06493aea73 100644 --- a/client/rest-high-level/src/main/java/org/elasticsearch/client/WatcherClient.java +++ b/client/rest-high-level/src/main/java/org/elasticsearch/client/WatcherClient.java @@ -32,6 +32,8 @@ import org.elasticsearch.client.watcher.DeleteWatchRequest; import org.elasticsearch.client.watcher.DeleteWatchResponse; import org.elasticsearch.client.watcher.PutWatchRequest; import org.elasticsearch.client.watcher.PutWatchResponse; +import org.elasticsearch.client.watcher.WatcherStatsRequest; +import org.elasticsearch.client.watcher.WatcherStatsResponse; import java.io.IOException; @@ -237,4 +239,31 @@ public final class WatcherClient { ActivateWatchResponse::fromXContent, listener, singleton(404)); } + /** + * Get the watcher stats + * See + * the docs for more. + * @param request the request + * @param options the request options (e.g. headers), use {@link RequestOptions#DEFAULT} if nothing needs to be customized + * @return the response + * @throws IOException in case there is a problem sending the request or parsing back the response + */ + public WatcherStatsResponse watcherStats(WatcherStatsRequest request, RequestOptions options) throws IOException { + return restHighLevelClient.performRequestAndParseEntity(request, WatcherRequestConverters::watcherStats, options, + WatcherStatsResponse::fromXContent, emptySet()); + } + + /** + * Asynchronously get the watcher stats + * See + * the docs for more. + * @param request the request + * @param options the request options (e.g. headers), use {@link RequestOptions#DEFAULT} if nothing needs to be customized + * @param listener the listener to be notified upon request completion + */ + public void watcherStatsAsync(WatcherStatsRequest request, RequestOptions options, ActionListener listener) { + restHighLevelClient.performRequestAsyncAndParseEntity(request, WatcherRequestConverters::watcherStats, options, + WatcherStatsResponse::fromXContent, listener, emptySet()); + } + } diff --git a/client/rest-high-level/src/main/java/org/elasticsearch/client/WatcherRequestConverters.java b/client/rest-high-level/src/main/java/org/elasticsearch/client/WatcherRequestConverters.java index 57e817e083a..a017779495b 100644 --- a/client/rest-high-level/src/main/java/org/elasticsearch/client/WatcherRequestConverters.java +++ b/client/rest-high-level/src/main/java/org/elasticsearch/client/WatcherRequestConverters.java @@ -20,6 +20,7 @@ package org.elasticsearch.client; import org.apache.http.client.methods.HttpDelete; +import org.apache.http.client.methods.HttpGet; import org.apache.http.client.methods.HttpPost; import org.apache.http.client.methods.HttpPut; import org.apache.http.entity.ByteArrayEntity; @@ -29,6 +30,7 @@ import org.elasticsearch.client.watcher.ActivateWatchRequest; import org.elasticsearch.client.watcher.AckWatchRequest; import org.elasticsearch.client.watcher.StartWatchServiceRequest; import org.elasticsearch.client.watcher.StopWatchServiceRequest; +import org.elasticsearch.client.watcher.WatcherStatsRequest; import org.elasticsearch.common.bytes.BytesReference; import org.elasticsearch.client.watcher.DeleteWatchRequest; import org.elasticsearch.client.watcher.PutWatchRequest; @@ -115,4 +117,25 @@ final class WatcherRequestConverters { Request request = new Request(HttpPut.METHOD_NAME, endpoint); return request; } + + static Request watcherStats(WatcherStatsRequest watcherStatsRequest) { + RequestConverters.EndpointBuilder builder = new RequestConverters.EndpointBuilder().addPathPartAsIs("_xpack", "watcher", "stats"); + String endpoint = builder.build(); + Request request = new Request(HttpGet.METHOD_NAME, endpoint); + RequestConverters.Params parameters = new RequestConverters.Params(request); + StringBuilder metric = new StringBuilder(); + if (watcherStatsRequest.includeCurrentWatches()) { + metric.append("current_watches"); + } + if (watcherStatsRequest.includeQueuedWatches()) { + if (metric.length() > 0) { + metric.append(","); + } + metric.append("queued_watches"); + } + if (metric.length() > 0) { + parameters.putParam("metric", metric.toString()); + } + return request; + } } diff --git a/client/rest-high-level/src/main/java/org/elasticsearch/client/watcher/ExecutionPhase.java b/client/rest-high-level/src/main/java/org/elasticsearch/client/watcher/ExecutionPhase.java new file mode 100644 index 00000000000..9cdfcfba153 --- /dev/null +++ b/client/rest-high-level/src/main/java/org/elasticsearch/client/watcher/ExecutionPhase.java @@ -0,0 +1,50 @@ +/* + * Licensed to Elasticsearch 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.client.watcher; + +public enum ExecutionPhase { + + // awaiting execution of the watch + AWAITS_EXECUTION(false), + // initial phase, watch execution has started, but the input is not yet processed + STARTED(false), + // input is being executed + INPUT(false), + // condition phase is being executed + CONDITION(false), + // transform phase (optional, depends if a global transform was configured in the watch) + WATCH_TRANSFORM(false), + // actions phase, all actions, including specific action transforms + ACTIONS(false), + // missing watch, failed execution of input/condition/transform, + ABORTED(true), + // successful run + FINISHED(true); + + private final boolean sealed; + + ExecutionPhase(boolean sealed) { + this.sealed = sealed; + } + + public boolean sealed() { + return sealed; + } +} diff --git a/client/rest-high-level/src/main/java/org/elasticsearch/client/watcher/QueuedWatch.java b/client/rest-high-level/src/main/java/org/elasticsearch/client/watcher/QueuedWatch.java new file mode 100644 index 00000000000..e091c15e812 --- /dev/null +++ b/client/rest-high-level/src/main/java/org/elasticsearch/client/watcher/QueuedWatch.java @@ -0,0 +1,91 @@ +/* + * Licensed to Elasticsearch 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.client.watcher; + +import org.elasticsearch.common.ParseField; +import org.elasticsearch.common.xcontent.ConstructingObjectParser; +import org.joda.time.DateTime; + +import java.util.Objects; + +public class QueuedWatch { + + @SuppressWarnings("unchecked") + public static final ConstructingObjectParser PARSER = + new ConstructingObjectParser<>("watcher_stats_node", true, (args, c) -> new QueuedWatch( + (String) args[0], + (String) args[1], + DateTime.parse((String) args[2]), + DateTime.parse((String) args[3]) + )); + + static { + PARSER.declareString(ConstructingObjectParser.constructorArg(), new ParseField("watch_id")); + PARSER.declareString(ConstructingObjectParser.constructorArg(), new ParseField("watch_record_id")); + PARSER.declareString(ConstructingObjectParser.constructorArg(), new ParseField("triggered_time")); + PARSER.declareString(ConstructingObjectParser.constructorArg(), new ParseField("execution_time")); + } + + + private final String watchId; + private final String watchRecordId; + private final DateTime triggeredTime; + private final DateTime executionTime; + + public QueuedWatch(String watchId, String watchRecordId, DateTime triggeredTime, DateTime executionTime) { + this.watchId = watchId; + this.watchRecordId = watchRecordId; + this.triggeredTime = triggeredTime; + this.executionTime = executionTime; + } + + public String getWatchId() { + return watchId; + } + + public String getWatchRecordId() { + return watchRecordId; + } + + public DateTime getTriggeredTime() { + return triggeredTime; + } + + public DateTime getExecutionTime() { + return executionTime; + } + + @Override + public boolean equals(Object o) { + if (this == o) return true; + if (o == null || getClass() != o.getClass()) return false; + QueuedWatch that = (QueuedWatch) o; + return Objects.equals(watchId, that.watchId) && + Objects.equals(watchRecordId, that.watchRecordId) && + Objects.equals(triggeredTime, that.triggeredTime) && + Objects.equals(executionTime, that.executionTime); + } + + @Override + public int hashCode() { + + return Objects.hash(watchId, watchRecordId, triggeredTime, executionTime); + } +} diff --git a/client/rest-high-level/src/main/java/org/elasticsearch/client/watcher/WatchExecutionSnapshot.java b/client/rest-high-level/src/main/java/org/elasticsearch/client/watcher/WatchExecutionSnapshot.java new file mode 100644 index 00000000000..d1dfcd1b0b2 --- /dev/null +++ b/client/rest-high-level/src/main/java/org/elasticsearch/client/watcher/WatchExecutionSnapshot.java @@ -0,0 +1,123 @@ +/* + * Licensed to Elasticsearch 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.client.watcher; + +import org.elasticsearch.common.ParseField; +import org.elasticsearch.common.xcontent.ConstructingObjectParser; +import org.joda.time.DateTime; + +import java.util.Arrays; +import java.util.List; +import java.util.Locale; +import java.util.Objects; + +public class WatchExecutionSnapshot { + @SuppressWarnings("unchecked") + public static final ConstructingObjectParser PARSER = + new ConstructingObjectParser<>("watcher_stats_node", true, (args, c) -> new WatchExecutionSnapshot( + (String) args[0], + (String) args[1], + DateTime.parse((String) args[2]), + DateTime.parse((String) args[3]), + ExecutionPhase.valueOf(((String) args[4]).toUpperCase(Locale.ROOT)), + args[5] == null ? null : ((List) args[5]).toArray(new String[0]), + args[6] == null ? null : ((List) args[6]).toArray(new String[0]) + )); + + static { + PARSER.declareString(ConstructingObjectParser.constructorArg(), new ParseField("watch_id")); + PARSER.declareString(ConstructingObjectParser.constructorArg(), new ParseField("watch_record_id")); + PARSER.declareString(ConstructingObjectParser.constructorArg(), new ParseField("triggered_time")); + PARSER.declareString(ConstructingObjectParser.constructorArg(), new ParseField("execution_time")); + PARSER.declareString(ConstructingObjectParser.constructorArg(), new ParseField("execution_phase")); + PARSER.declareStringArray(ConstructingObjectParser.optionalConstructorArg(), new ParseField("executed_actions")); + PARSER.declareStringArray(ConstructingObjectParser.optionalConstructorArg(), new ParseField("stack_trace")); + } + + private final String watchId; + private final String watchRecordId; + private final DateTime triggeredTime; + private final DateTime executionTime; + private final ExecutionPhase phase; + private final String[] executedActions; + private final String[] executionStackTrace; + + public WatchExecutionSnapshot(String watchId, String watchRecordId, DateTime triggeredTime, DateTime executionTime, + ExecutionPhase phase, String[] executedActions, String[] executionStackTrace) { + this.watchId = watchId; + this.watchRecordId = watchRecordId; + this.triggeredTime = triggeredTime; + this.executionTime = executionTime; + this.phase = phase; + this.executedActions = executedActions; + this.executionStackTrace = executionStackTrace; + } + + public String getWatchId() { + return watchId; + } + + public String getWatchRecordId() { + return watchRecordId; + } + + public DateTime getTriggeredTime() { + return triggeredTime; + } + + public DateTime getExecutionTime() { + return executionTime; + } + + public ExecutionPhase getPhase() { + return phase; + } + + public String[] getExecutedActions() { + return executedActions; + } + + public String[] getExecutionStackTrace() { + return executionStackTrace; + } + + @Override + public boolean equals(Object o) { + if (this == o) return true; + if (o == null || getClass() != o.getClass()) return false; + WatchExecutionSnapshot that = (WatchExecutionSnapshot) o; + return Objects.equals(watchId, that.watchId) && + Objects.equals(watchRecordId, that.watchRecordId) && + Objects.equals(triggeredTime, that.triggeredTime) && + Objects.equals(executionTime, that.executionTime) && + phase == that.phase && + Arrays.equals(executedActions, that.executedActions) && + Arrays.equals(executionStackTrace, that.executionStackTrace); + } + + @Override + public int hashCode() { + + int result = Objects.hash(watchId, watchRecordId, triggeredTime, executionTime, phase); + result = 31 * result + Arrays.hashCode(executedActions); + result = 31 * result + Arrays.hashCode(executionStackTrace); + return result; + } +} diff --git a/client/rest-high-level/src/main/java/org/elasticsearch/client/watcher/WatcherMetaData.java b/client/rest-high-level/src/main/java/org/elasticsearch/client/watcher/WatcherMetaData.java new file mode 100644 index 00000000000..1d84fb943f0 --- /dev/null +++ b/client/rest-high-level/src/main/java/org/elasticsearch/client/watcher/WatcherMetaData.java @@ -0,0 +1,54 @@ +/* + * Licensed to Elasticsearch 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.client.watcher; + +import java.util.Objects; + +public class WatcherMetaData { + + private final boolean manuallyStopped; + + public WatcherMetaData(boolean manuallyStopped) { + this.manuallyStopped = manuallyStopped; + } + + public boolean manuallyStopped() { + return manuallyStopped; + } + @Override + public String toString() { + return "manuallyStopped["+ manuallyStopped +"]"; + } + + @Override + public boolean equals(Object o) { + if (this == o) return true; + if (o == null || getClass() != o.getClass()) return false; + + WatcherMetaData action = (WatcherMetaData) o; + + return manuallyStopped == action.manuallyStopped; + } + + @Override + public int hashCode() { + return Objects.hash(manuallyStopped); + } +} diff --git a/client/rest-high-level/src/main/java/org/elasticsearch/client/watcher/WatcherState.java b/client/rest-high-level/src/main/java/org/elasticsearch/client/watcher/WatcherState.java new file mode 100644 index 00000000000..710f9e437c5 --- /dev/null +++ b/client/rest-high-level/src/main/java/org/elasticsearch/client/watcher/WatcherState.java @@ -0,0 +1,54 @@ +/* + * Licensed to Elasticsearch 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.client.watcher; + +public enum WatcherState { + + /** + * The watcher plugin is not running and not functional. + */ + STOPPED(0), + + /** + * The watcher plugin is performing the necessary operations to get into a started state. + */ + STARTING(1), + + /** + * The watcher plugin is running and completely functional. + */ + STARTED(2), + + /** + * The watcher plugin is shutting down and not functional. + */ + STOPPING(3); + + private final byte id; + + WatcherState(int id) { + this.id = (byte) id; + } + + public byte getId() { + return id; + } + +} + diff --git a/client/rest-high-level/src/main/java/org/elasticsearch/client/watcher/WatcherStatsRequest.java b/client/rest-high-level/src/main/java/org/elasticsearch/client/watcher/WatcherStatsRequest.java new file mode 100644 index 00000000000..5b48978c88a --- /dev/null +++ b/client/rest-high-level/src/main/java/org/elasticsearch/client/watcher/WatcherStatsRequest.java @@ -0,0 +1,53 @@ +/* + * Licensed to Elasticsearch 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.client.watcher; + +import org.elasticsearch.client.Validatable; + +/** + * A request to explicitly acknowledge a watch. + */ +public class WatcherStatsRequest implements Validatable { + + private final boolean includeCurrentWatches; + private final boolean includeQueuedWatches; + + public WatcherStatsRequest( ) { + this(true, true); + } + + public WatcherStatsRequest(boolean includeCurrentWatches, boolean includeQueuedWatches) { + this.includeCurrentWatches = includeCurrentWatches; + this.includeQueuedWatches = includeQueuedWatches; + } + + public boolean includeCurrentWatches() { + return includeCurrentWatches; + } + + public boolean includeQueuedWatches() { + return includeQueuedWatches; + } + + @Override + public String toString() { + return "stats [current=" + includeCurrentWatches + ", " + "queued=" + includeQueuedWatches + "]"; + } +} diff --git a/client/rest-high-level/src/main/java/org/elasticsearch/client/watcher/WatcherStatsResponse.java b/client/rest-high-level/src/main/java/org/elasticsearch/client/watcher/WatcherStatsResponse.java new file mode 100644 index 00000000000..708954e666b --- /dev/null +++ b/client/rest-high-level/src/main/java/org/elasticsearch/client/watcher/WatcherStatsResponse.java @@ -0,0 +1,229 @@ +/* + * Licensed to Elasticsearch 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.client.watcher; + +import org.elasticsearch.client.NodesResponseHeader; +import org.elasticsearch.common.ParseField; +import org.elasticsearch.common.collect.Tuple; +import org.elasticsearch.common.xcontent.ConstructingObjectParser; +import org.elasticsearch.common.xcontent.XContentParser; + +import java.io.IOException; +import java.util.List; +import java.util.Locale; +import java.util.Map; +import java.util.Objects; + +/** + * The response from an 'ack watch' request. + */ +public class WatcherStatsResponse { + + private final List nodes; + private final NodesResponseHeader header; + private final String clusterName; + + private final WatcherMetaData watcherMetaData; + + public WatcherStatsResponse(NodesResponseHeader header, String clusterName, WatcherMetaData watcherMetaData, List nodes) { + this.nodes = nodes; + this.header = header; + this.clusterName = clusterName; + this.watcherMetaData = watcherMetaData; + } + + /** + * @return the status of the requested watch. If an action was + * successfully acknowledged, this will be reflected in its status. + */ + public WatcherMetaData getWatcherMetaData() { + return watcherMetaData; + } + + /** + * returns a list of nodes that returned stats + */ + public List getNodes() { + return nodes; + } + + /** + * Gets information about the number of total, successful and failed nodes the request was run on. + * Also includes exceptions if relevant. + */ + public NodesResponseHeader getHeader() { + return header; + } + + /** + * Get the cluster name associated with all of the nodes. + * + * @return Never {@code null}. + */ + public String getClusterName() { + return clusterName; + } + + @SuppressWarnings("unchecked") + private static ConstructingObjectParser PARSER = + new ConstructingObjectParser<>("watcher_stats_response", true, + a -> new WatcherStatsResponse((NodesResponseHeader) a[0], (String) a[1], new WatcherMetaData((boolean) a[2]), + (List) a[3])); + + static { + PARSER.declareObject(ConstructingObjectParser.constructorArg(), NodesResponseHeader::fromXContent, new ParseField("_nodes")); + PARSER.declareString(ConstructingObjectParser.constructorArg(), new ParseField("cluster_name")); + PARSER.declareBoolean(ConstructingObjectParser.constructorArg(), new ParseField("manually_stopped")); + PARSER.declareObjectArray(ConstructingObjectParser.constructorArg(), (p, c) -> Node.PARSER.apply(p, null), + new ParseField("stats")); + } + + public static WatcherStatsResponse fromXContent(XContentParser parser) throws IOException { + return PARSER.parse(parser, null); + } + + @Override + public boolean equals(Object o) { + if (this == o) return true; + if (o == null || getClass() != o.getClass()) return false; + WatcherStatsResponse that = (WatcherStatsResponse) o; + return Objects.equals(nodes, that.nodes) && + Objects.equals(header, that.header) && + Objects.equals(clusterName, that.clusterName) && + Objects.equals(watcherMetaData, that.watcherMetaData); + } + + @Override + public int hashCode() { + + return Objects.hash(nodes, header, clusterName, watcherMetaData); + } + + public static class Node { + @SuppressWarnings("unchecked") + public static final ConstructingObjectParser PARSER = + new ConstructingObjectParser<>("watcher_stats_node", true, (args, c) -> new Node( + (String) args[0], + WatcherState.valueOf(((String) args[1]).toUpperCase(Locale.ROOT)), + (long) args[2], + ((Tuple) args[3]).v1(), + ((Tuple) args[3]).v2(), + (List) args[4], + (List) args[5], + (Map) args[6] + + )); + + private static final ConstructingObjectParser, Void> THREAD_POOL_PARSER = + new ConstructingObjectParser<>("execution_thread_pool", true, (args, id) -> new Tuple<>((Long) args[0], (Long) args[1])); + + static { + PARSER.declareString(ConstructingObjectParser.constructorArg(), new ParseField("node_id")); + PARSER.declareString(ConstructingObjectParser.constructorArg(), new ParseField("watcher_state")); + PARSER.declareLong(ConstructingObjectParser.constructorArg(), new ParseField("watch_count")); + PARSER.declareObject(ConstructingObjectParser.constructorArg(), THREAD_POOL_PARSER::apply, + new ParseField("execution_thread_pool")); + PARSER.declareObjectArray(ConstructingObjectParser.optionalConstructorArg(), WatchExecutionSnapshot.PARSER, + new ParseField("current_watches")); + PARSER.declareObjectArray(ConstructingObjectParser.optionalConstructorArg(), QueuedWatch.PARSER, + new ParseField("queued_watches")); + PARSER.declareObject(ConstructingObjectParser.optionalConstructorArg(), (p, c) -> p.map(), new ParseField("stats")); + + THREAD_POOL_PARSER.declareLong(ConstructingObjectParser.constructorArg(), new ParseField("queue_size")); + THREAD_POOL_PARSER.declareLong(ConstructingObjectParser.constructorArg(), new ParseField("max_size")); + } + + private final String nodeId; + + private WatcherState watcherState; + private long watchesCount; + private long threadPoolQueueSize; + private long threadPoolMaxSize; + private List snapshots; + private List queuedWatches; + private Map stats; + + + public Node(String nodeId, WatcherState watcherState, long watchesCount, long threadPoolQueueSize, long threadPoolMaxSize, + List snapshots, List queuedWatches, Map stats) { + this.nodeId = nodeId; + this.watcherState = watcherState; + this.watchesCount = watchesCount; + this.threadPoolQueueSize = threadPoolQueueSize; + this.threadPoolMaxSize = threadPoolMaxSize; + this.snapshots = snapshots; + this.queuedWatches = queuedWatches; + this.stats = stats; + } + + public String getNodeId() { + return nodeId; + } + + public long getWatchesCount() { + return watchesCount; + } + + public WatcherState getWatcherState() { + return watcherState; + } + + public long getThreadPoolQueueSize() { + return threadPoolQueueSize; + } + + public long getThreadPoolMaxSize() { + return threadPoolMaxSize; + } + + public List getSnapshots() { + return snapshots; + } + + public List getQueuedWatches() { + return queuedWatches; + } + + public Map getStats() { + return stats; + } + + @Override + public boolean equals(Object o) { + if (this == o) return true; + if (o == null || getClass() != o.getClass()) return false; + Node node = (Node) o; + return watchesCount == node.watchesCount && + threadPoolQueueSize == node.threadPoolQueueSize && + threadPoolMaxSize == node.threadPoolMaxSize && + Objects.equals(nodeId, node.nodeId) && + watcherState == node.watcherState && + Objects.equals(snapshots, node.snapshots) && + Objects.equals(queuedWatches, node.queuedWatches) && + Objects.equals(stats, node.stats); + } + + @Override + public int hashCode() { + return Objects.hash(nodeId, watcherState, watchesCount, threadPoolQueueSize, threadPoolMaxSize, snapshots, queuedWatches, + stats); + } + } +} diff --git a/client/rest-high-level/src/test/java/org/elasticsearch/client/NodesResponseHeaderTestUtils.java b/client/rest-high-level/src/test/java/org/elasticsearch/client/NodesResponseHeaderTestUtils.java new file mode 100644 index 00000000000..09ec3f38bfd --- /dev/null +++ b/client/rest-high-level/src/test/java/org/elasticsearch/client/NodesResponseHeaderTestUtils.java @@ -0,0 +1,49 @@ +/* + * Licensed to Elasticsearch 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.client; + +import org.elasticsearch.ElasticsearchException; +import org.elasticsearch.common.xcontent.ToXContent; +import org.elasticsearch.common.xcontent.XContentBuilder; + +import java.io.IOException; + +public class NodesResponseHeaderTestUtils { + + public static void toXContent(NodesResponseHeader header, String clusterName, XContentBuilder builder) throws IOException { + builder.startObject("_nodes"); + builder.field("total", header.getTotal()); + builder.field("successful", header.getSuccessful()); + builder.field("failed", header.getFailed()); + + if (header.getFailures().isEmpty() == false) { + builder.startArray("failures"); + for (ElasticsearchException failure : header.getFailures()) { + builder.startObject(); + failure.toXContent(builder, ToXContent.EMPTY_PARAMS); + builder.endObject(); + } + builder.endArray(); + } + + builder.endObject(); + builder.field("cluster_name", clusterName); + } + +} diff --git a/client/rest-high-level/src/test/java/org/elasticsearch/client/WatcherIT.java b/client/rest-high-level/src/test/java/org/elasticsearch/client/WatcherIT.java index c4a2242f901..8d1429d319b 100644 --- a/client/rest-high-level/src/test/java/org/elasticsearch/client/WatcherIT.java +++ b/client/rest-high-level/src/test/java/org/elasticsearch/client/WatcherIT.java @@ -32,6 +32,9 @@ import org.elasticsearch.client.watcher.ActivateWatchRequest; import org.elasticsearch.client.watcher.ActivateWatchResponse; import org.elasticsearch.client.watcher.StartWatchServiceRequest; import org.elasticsearch.client.watcher.StopWatchServiceRequest; +import org.elasticsearch.client.watcher.WatcherState; +import org.elasticsearch.client.watcher.WatcherStatsRequest; +import org.elasticsearch.client.watcher.WatcherStatsResponse; import org.elasticsearch.common.bytes.BytesArray; import org.elasticsearch.common.bytes.BytesReference; import org.elasticsearch.common.xcontent.XContentType; @@ -41,8 +44,10 @@ import org.elasticsearch.client.watcher.PutWatchRequest; import org.elasticsearch.client.watcher.PutWatchResponse; import org.elasticsearch.rest.RestStatus; +import static org.hamcrest.Matchers.empty; import static org.hamcrest.Matchers.is; import static org.hamcrest.Matchers.lessThan; +import static org.hamcrest.Matchers.not; public class WatcherIT extends ESRestHighLevelClientTestCase { @@ -50,12 +55,22 @@ public class WatcherIT extends ESRestHighLevelClientTestCase { AcknowledgedResponse response = highLevelClient().watcher().startWatchService(new StartWatchServiceRequest(), RequestOptions.DEFAULT); assertTrue(response.isAcknowledged()); + + WatcherStatsResponse stats = highLevelClient().watcher().watcherStats(new WatcherStatsRequest(), RequestOptions.DEFAULT); + assertFalse(stats.getWatcherMetaData().manuallyStopped()); + assertThat(stats.getNodes(), not(empty())); + for(WatcherStatsResponse.Node node : stats.getNodes()) { + assertEquals(WatcherState.STARTED, node.getWatcherState()); + } } public void testStopWatchService() throws Exception { AcknowledgedResponse response = highLevelClient().watcher().stopWatchService(new StopWatchServiceRequest(), RequestOptions.DEFAULT); assertTrue(response.isAcknowledged()); + + WatcherStatsResponse stats = highLevelClient().watcher().watcherStats(new WatcherStatsRequest(), RequestOptions.DEFAULT); + assertTrue(stats.getWatcherMetaData().manuallyStopped()); } public void testPutWatch() throws Exception { @@ -169,4 +184,15 @@ public class WatcherIT extends ESRestHighLevelClientTestCase { highLevelClient().watcher().activateWatch(new ActivateWatchRequest(watchId), RequestOptions.DEFAULT)); assertEquals(RestStatus.NOT_FOUND, exception.status()); } + + public void testWatcherStatsMetrics() throws Exception { + boolean includeCurrent = randomBoolean(); + boolean includeQueued = randomBoolean(); + WatcherStatsRequest request = new WatcherStatsRequest(includeCurrent, includeQueued); + + WatcherStatsResponse stats = highLevelClient().watcher().watcherStats(request, RequestOptions.DEFAULT); + assertThat(stats.getNodes(), not(empty())); + assertEquals(includeCurrent, stats.getNodes().get(0).getSnapshots() != null); + assertEquals(includeQueued, stats.getNodes().get(0).getQueuedWatches() != null); + } } diff --git a/client/rest-high-level/src/test/java/org/elasticsearch/client/WatcherRequestConvertersTests.java b/client/rest-high-level/src/test/java/org/elasticsearch/client/WatcherRequestConvertersTests.java index b0b04fd0e5b..2712dbc0438 100644 --- a/client/rest-high-level/src/test/java/org/elasticsearch/client/WatcherRequestConvertersTests.java +++ b/client/rest-high-level/src/test/java/org/elasticsearch/client/WatcherRequestConvertersTests.java @@ -20,25 +20,34 @@ package org.elasticsearch.client; import org.apache.http.client.methods.HttpDelete; +import org.apache.http.client.methods.HttpGet; import org.apache.http.client.methods.HttpPost; import org.apache.http.client.methods.HttpPut; -import org.elasticsearch.client.watcher.DeactivateWatchRequest; -import org.elasticsearch.client.watcher.ActivateWatchRequest; import org.elasticsearch.client.watcher.AckWatchRequest; -import org.elasticsearch.client.watcher.StartWatchServiceRequest; -import org.elasticsearch.client.watcher.StopWatchServiceRequest; -import org.elasticsearch.common.bytes.BytesArray; -import org.elasticsearch.common.xcontent.XContentType; +import org.elasticsearch.client.watcher.ActivateWatchRequest; +import org.elasticsearch.client.watcher.DeactivateWatchRequest; import org.elasticsearch.client.watcher.DeleteWatchRequest; import org.elasticsearch.client.watcher.PutWatchRequest; +import org.elasticsearch.client.watcher.StartWatchServiceRequest; +import org.elasticsearch.client.watcher.StopWatchServiceRequest; +import org.elasticsearch.client.watcher.WatcherStatsRequest; +import org.elasticsearch.common.Strings; +import org.elasticsearch.common.bytes.BytesArray; +import org.elasticsearch.common.xcontent.XContentType; import org.elasticsearch.test.ESTestCase; import java.io.ByteArrayOutputStream; import java.util.HashMap; +import java.util.HashSet; import java.util.Map; +import java.util.Set; import java.util.StringJoiner; +import static org.hamcrest.CoreMatchers.equalTo; +import static org.hamcrest.Matchers.hasKey; +import static org.hamcrest.Matchers.hasSize; import static org.hamcrest.Matchers.is; +import static org.hamcrest.Matchers.not; import static org.hamcrest.Matchers.nullValue; public class WatcherRequestConvertersTests extends ESTestCase { @@ -130,4 +139,31 @@ public class WatcherRequestConvertersTests extends ESTestCase { assertEquals("/_xpack/watcher/watch/" + watchId + "/_activate", request.getEndpoint()); assertThat(request.getEntity(), nullValue()); } + + public void testWatcherStatsRequest() { + boolean includeCurrent = randomBoolean(); + boolean includeQueued = randomBoolean(); + + WatcherStatsRequest watcherStatsRequest = new WatcherStatsRequest(includeCurrent, includeQueued); + + Request request = WatcherRequestConverters.watcherStats(watcherStatsRequest); + assertThat(request.getEndpoint(), equalTo("/_xpack/watcher/stats")); + assertThat(request.getMethod(), equalTo(HttpGet.METHOD_NAME)); + if (includeCurrent || includeQueued) { + assertThat(request.getParameters(), hasKey("metric")); + Set metric = Strings.tokenizeByCommaToSet(request.getParameters().get("metric")); + assertThat(metric, hasSize((includeCurrent?1:0) + (includeQueued?1:0))); + Set expectedMetric = new HashSet<>(); + if (includeCurrent) { + expectedMetric.add("current_watches"); + } + if (includeQueued) { + expectedMetric.add("queued_watches"); + } + assertThat(metric, equalTo(expectedMetric)); + } else { + assertThat(request.getParameters(), not(hasKey("metric"))); + } + assertThat(request.getEntity(), nullValue()); + } } diff --git a/client/rest-high-level/src/test/java/org/elasticsearch/client/documentation/WatcherDocumentationIT.java b/client/rest-high-level/src/test/java/org/elasticsearch/client/documentation/WatcherDocumentationIT.java index ac4fca82b2e..74562a1d17f 100644 --- a/client/rest-high-level/src/test/java/org/elasticsearch/client/documentation/WatcherDocumentationIT.java +++ b/client/rest-high-level/src/test/java/org/elasticsearch/client/documentation/WatcherDocumentationIT.java @@ -37,6 +37,8 @@ import org.elasticsearch.client.watcher.DeactivateWatchResponse; import org.elasticsearch.client.watcher.StartWatchServiceRequest; import org.elasticsearch.client.watcher.StopWatchServiceRequest; import org.elasticsearch.client.watcher.WatchStatus; +import org.elasticsearch.client.watcher.WatcherStatsRequest; +import org.elasticsearch.client.watcher.WatcherStatsResponse; import org.elasticsearch.common.bytes.BytesArray; import org.elasticsearch.common.bytes.BytesReference; import org.elasticsearch.common.xcontent.XContentType; @@ -46,6 +48,7 @@ import org.elasticsearch.client.watcher.PutWatchRequest; import org.elasticsearch.client.watcher.PutWatchResponse; import org.elasticsearch.rest.RestStatus; +import java.util.List; import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; @@ -100,7 +103,7 @@ public class WatcherDocumentationIT extends ESRestHighLevelClientTestCase { } }; // end::start-watch-service-execute-listener - + CountDownLatch latch = new CountDownLatch(1); listener = new LatchedActionListener<>(listener, latch); @@ -408,4 +411,49 @@ public class WatcherDocumentationIT extends ESRestHighLevelClientTestCase { } } + public void testWatcherStats() throws Exception { + RestHighLevelClient client = highLevelClient(); + + { + //tag::watcher-stats-request + WatcherStatsRequest request = new WatcherStatsRequest(true, true); + //end::watcher-stats-request + + //tag::watcher-stats-execute + WatcherStatsResponse response = client.watcher().watcherStats(request, RequestOptions.DEFAULT); + //end::watcher-stats-execute + + //tag::watcher-stats-response + List nodes = response.getNodes(); // <1> + //end::watcher-stats-response + } + + { + WatcherStatsRequest request = new WatcherStatsRequest(); + + // tag::watcher-stats-execute-listener + ActionListener listener = new ActionListener() { + @Override + public void onResponse(WatcherStatsResponse response) { + // <1> + } + + @Override + public void onFailure(Exception e) { + // <2> + } + }; + // end::watcher-stats-execute-listener + + CountDownLatch latch = new CountDownLatch(1); + listener = new LatchedActionListener<>(listener, latch); + + // tag::watcher-stats-execute-async + client.watcher().watcherStatsAsync(request, RequestOptions.DEFAULT, listener); // <1> + // end::watcher-stats-execute-async + + assertTrue(latch.await(30L, TimeUnit.SECONDS)); + } + } + } diff --git a/client/rest-high-level/src/test/java/org/elasticsearch/client/watcher/WatcherStatsResponseTests.java b/client/rest-high-level/src/test/java/org/elasticsearch/client/watcher/WatcherStatsResponseTests.java new file mode 100644 index 00000000000..f1efe13c761 --- /dev/null +++ b/client/rest-high-level/src/test/java/org/elasticsearch/client/watcher/WatcherStatsResponseTests.java @@ -0,0 +1,188 @@ +/* + * Licensed to Elasticsearch 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.client.watcher; + +import org.elasticsearch.client.NodesResponseHeader; +import org.elasticsearch.client.NodesResponseHeaderTestUtils; +import org.elasticsearch.common.xcontent.XContentBuilder; +import org.elasticsearch.test.ESTestCase; +import org.joda.time.DateTime; +import org.joda.time.DateTimeZone; + +import java.io.IOException; +import java.util.ArrayList; +import java.util.Collections; +import java.util.HashMap; +import java.util.List; +import java.util.Locale; +import java.util.Map; + +import static org.elasticsearch.test.AbstractXContentTestCase.xContentTester; + +public class WatcherStatsResponseTests extends ESTestCase { + + public void testFromXContent() throws IOException { + xContentTester( + this::createParser, + this::createTestInstance, + this::toXContent, + WatcherStatsResponse::fromXContent) + .supportsUnknownFields(true) + .randomFieldsExcludeFilter(field -> field.endsWith("stats")) + .test(); + } + + private void toXContent(WatcherStatsResponse response, XContentBuilder builder) throws IOException { + builder.startObject(); + NodesResponseHeaderTestUtils.toXContent(response.getHeader(), response.getClusterName(), builder); + toXContent(response.getWatcherMetaData(), builder); + builder.startArray("stats"); + for (WatcherStatsResponse.Node node : response.getNodes()) { + toXContent(node, builder); + } + builder.endArray(); + builder.endObject(); + } + + private void toXContent(WatcherMetaData metaData, XContentBuilder builder) throws IOException { + builder.field("manually_stopped", metaData.manuallyStopped()); + } + + private void toXContent(WatcherStatsResponse.Node node, XContentBuilder builder) throws IOException { + builder.startObject(); + builder.field("node_id", node.getNodeId()); + builder.field("watcher_state", node.getWatcherState().toString().toLowerCase(Locale.ROOT)); + builder.field("watch_count", node.getWatchesCount()); + builder.startObject("execution_thread_pool"); + builder.field("queue_size", node.getThreadPoolQueueSize()); + builder.field("max_size", node.getThreadPoolMaxSize()); + builder.endObject(); + + if (node.getSnapshots() != null) { + builder.startArray("current_watches"); + for (WatchExecutionSnapshot snapshot : node.getSnapshots()) { + toXContent(snapshot, builder); + } + builder.endArray(); + } + if (node.getQueuedWatches() != null) { + builder.startArray("queued_watches"); + for (QueuedWatch queuedWatch : node.getQueuedWatches()) { + toXContent(queuedWatch, builder); + } + builder.endArray(); + } + if (node.getStats() != null) { + builder.field("stats", node.getStats()); + } + builder.endObject(); + } + + private void toXContent(WatchExecutionSnapshot snapshot, XContentBuilder builder) throws IOException { + builder.startObject(); + builder.field("watch_id", snapshot.getWatchId()); + builder.field("watch_record_id", snapshot.getWatchRecordId()); + builder.timeField("triggered_time", snapshot.getTriggeredTime()); + builder.timeField("execution_time", snapshot.getExecutionTime()); + builder.field("execution_phase", snapshot.getPhase()); + if (snapshot.getExecutedActions() != null) { + builder.startArray("executed_actions"); + for (String executedAction : snapshot.getExecutedActions()) { + builder.value(executedAction); + } + builder.endArray(); + } + if (snapshot.getExecutionStackTrace() != null) { + builder.startArray("stack_trace"); + for (String element : snapshot.getExecutionStackTrace()) { + builder.value(element); + } + builder.endArray(); + } + builder.endObject(); + } + + private void toXContent(QueuedWatch queuedWatch, XContentBuilder builder) throws IOException { + builder.startObject(); + builder.field("watch_id", queuedWatch.getWatchId()); + builder.field("watch_record_id", queuedWatch.getWatchRecordId()); + builder.timeField("triggered_time", queuedWatch.getTriggeredTime()); + builder.timeField("execution_time", queuedWatch.getExecutionTime()); + builder.endObject(); + } + + protected WatcherStatsResponse createTestInstance() { + int nodeCount = randomInt(10); + List nodes = new ArrayList<>(nodeCount); + for (int i = 0; i < nodeCount; i++) { + List snapshots = null; + if (randomBoolean()) { + int snapshotCount = randomInt(10); + snapshots = new ArrayList<>(snapshotCount); + + for (int j = 0; j < snapshotCount; j++) { + String[] actions = null; + if (randomBoolean()) { + actions = new String[randomInt(10)]; + for (int k = 0; k < actions.length; k++) { + actions[k] = randomAlphaOfLength(10); + } + } + String[] stackTrace = null; + if (randomBoolean()) { + stackTrace = new String[randomInt(10)]; + for (int k = 0; k < stackTrace.length; k++) { + stackTrace[k] = randomAlphaOfLength(10); + } + } + snapshots.add(new WatchExecutionSnapshot(randomAlphaOfLength(10), randomAlphaOfLength(10), + new DateTime(randomInt(), DateTimeZone.UTC), new DateTime(randomInt(), DateTimeZone.UTC), + randomFrom(ExecutionPhase.values()), actions, stackTrace)); + } + } + + List queuedWatches = null; + if(randomBoolean()) { + int queuedWatchCount = randomInt(10); + queuedWatches = new ArrayList<>(queuedWatchCount); + for (int j=0; j stats = null; + if (randomBoolean()) { + int statsCount = randomInt(10); + stats = new HashMap<>(statsCount); + for (int j=0; j> * <<{upid}-ack-watch>> * <<{upid}-activate-watch>> +* <<{upid}-watcher-stats>> include::watcher/start-watch-service.asciidoc[] include::watcher/stop-watch-service.asciidoc[] @@ -381,6 +382,7 @@ include::watcher/delete-watch.asciidoc[] include::watcher/ack-watch.asciidoc[] include::watcher/deactivate-watch.asciidoc[] include::watcher/activate-watch.asciidoc[] +include::watcher/watcher-stats.asciidoc[] == Graph APIs diff --git a/docs/java-rest/high-level/watcher/watcher-stats.asciidoc b/docs/java-rest/high-level/watcher/watcher-stats.asciidoc new file mode 100644 index 00000000000..7fd27053fcb --- /dev/null +++ b/docs/java-rest/high-level/watcher/watcher-stats.asciidoc @@ -0,0 +1,32 @@ +-- +:api: watcher-stats +:request: WatcherStatsRequest +:response: WatcherStatsResponse +-- +[id="{upid}-{api}"] +=== Watcher Stats API + +[id="{upid}-{api}-request"] +==== Execution + +{ref}/watcher-api-stats.html[Watcher Stats] returns the current {watcher} metrics. +Submit the following request to get the stats: + +["source","java",subs="attributes,callouts,macros"] +-------------------------------------------------- +include-tagged::{doc-tests-file}[{api}-request] +-------------------------------------------------- + +[id="{upid}-{api}-response"] +==== Response + +The returned `AcknowledgeResponse` contains a value on whether or not the request +was received: + +["source","java",subs="attributes,callouts,macros"] +-------------------------------------------------- +include-tagged::{doc-tests-file}[{api}-response] +-------------------------------------------------- +<1> A boolean value of `true` if successfully received, `false` otherwise. + +include::../execution.asciidoc[]