honeymoose/OpenSearch
1
0
Fork 0
mirror of https://github.com/honeymoose/OpenSearch.git synced 2025-03-09 14:34:43 +00:00
Code Issues Packages Projects Releases Wiki Activity

Changed watch_record xcontent structure

Browse Source 
- Renamed `watch_execution` to `execution_result`
- Renamed `actions_results` to `actions`
- Renamed `input_result` to `input`
- Renamed `condition_result` to `condition`
- Updated the `watch_history.json` template to reflect the changes, also added concrete mappings for action types (such that field that should not be analized will be mapped as `not_analyzed`
- Fixed a bug in `WatchUtils.createSearchRequestFromPrototype` where the document types were ignored.

Also, changed the `actions` (fka `actions_results`) from an object to an array. the action id is not part of the action objects (indicated by the `id` field). For example:

```
{
  "actions" : [
    {
      "id" : "my_email",
      "email" : {
        ...
      }
    }
  ]
}
```

The reason for this change is to make the path to the action fields predictable deterministic. With the object structure, where the actions were keyed by their `id`, the path to the action fields depended on the action id, which is unpredictable and continuously changing from one action to another. This made it impossible to properly analyze the action data using aggregations (as aggs require full path into the aggregated fields).

With this change, the mappings of `watch_record` changed as well where the `actions` are not defined as nested type, yet it is still configured to include all the fields in the root object. We do this so in the future, when appropriate support will be added to kibana, it'll be able to apply nested aggregations on the actions, enabling correct/safe multi-dimensional aggregations. In the mean time however, while kibana doesn't support nested aggregations, we still need to have all the fields indexed on the root, so at least a single dimensional aggregations can be safely applied.

Also, change the `input` and `condition` objects in the `watch_record` such that their mappings are disabled. The main reason for this is the fact that a lot of the inputs use elements that can be configured in many ways, but the mappings are too strict to accept it. For example, a template can be configured as a `string` or as an `object`.

Original commit: elastic/x-pack-elasticsearch@83464a0c71
This commit is contained in:
uboness 2015-05-08 22:35:48 +02:00
parent 7ff92c8f96
commit 46dfa059fd
27 changed files with 537 additions and 229 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

11
rest-api-spec/test/execute_watch/10_basic.yaml
View File

@ -44,7 +44,7 @@
}
},
"actions" : {
"EmailAdmin" : {
"email_admin" : {
"email" : {
"to" : "someone@domain.host.com",
"subject" : "404 recently encountered"
@ -74,9 +74,10 @@
"record_execution" : true
}
- match: { "watch_id": "my_exe_watch" }
- match: { "watch_execution.condition_result.always": {} }
- match: { "state": "executed" }
- match: { "trigger_event.manual.schedule.scheduled_time": "2015-05-05T20:58:02.443Z" }
- match: { "watch_execution.input_result.simple.payload.foo": "bar" }
- match: { "watch_execution.actions_results.EmailAdmin.email.success" : true }
- match: { "watch_execution.actions_results.EmailAdmin.email.simulated_email.subject" : "404 recently encountered" }
- match: { "execution_result.condition.always": {} }
- match: { "execution_result.input.simple.payload.foo": "bar" }
- match: { "execution_result.actions.0.id" : "email_admin" }
- match: { "execution_result.actions.0.email.success" : true }
- match: { "execution_result.actions.0.email.simulated_email.subject" : "404 recently encountered" }

9
rest-api-spec/test/execute_watch/20_minimal_body.yaml
View File

@ -21,7 +21,7 @@
"script" : "ctx.payload.count == 1"
},
"actions" : {
"Logging" : {
"logging" : {
"logging" : {
"text" : "foobar"
}
@ -44,7 +44,8 @@
}
- match: { "watch_id": "my_logging_watch" }
- match: { "watch_execution.condition_result.script.met": true }
- match: { "execution_result.condition.script.met": true }
- match: { "state": "executed" }
- match: { "watch_execution.actions_results.Logging.logging.success" : true }
- match: { "watch_execution.actions_results.Logging.logging.logged_text" : "foobar" }
- match: { "execution_result.actions.0.id" : "logging" }
- match: { "execution_result.actions.0.logging.success" : true }
- match: { "execution_result.actions.0.logging.logged_text" : "foobar" }

15
src/main/java/org/elasticsearch/watcher/actions/ActionRegistry.java
View File

@ -53,15 +53,14 @@ public class ActionRegistry {
public ExecutableActions.Results parseResults(Wid wid, XContentParser parser) throws IOException {
Map<String, ActionWrapper.Result> results = new HashMap<>();
String id = null;
if (parser.currentToken() != XContentParser.Token.START_ARRAY) {
throw new ActionException("could not parse action results for watch [{}]. expected an array of actions, but found [{}]", parser.currentToken());
}
XContentParser.Token token;
while ((token = parser.nextToken()) != XContentParser.Token.END_OBJECT) {
if (token == XContentParser.Token.FIELD_NAME) {
id = parser.currentName();
} else if (token == XContentParser.Token.START_OBJECT && id != null) {
ActionWrapper.Result result = ActionWrapper.Result.parse(wid, id, parser, this, transformRegistry);
results.put(id, result);
}
while ((token = parser.nextToken()) != XContentParser.Token.END_ARRAY) {
ActionWrapper.Result result = ActionWrapper.Result.parse(wid, parser, this, transformRegistry);
results.put(result.id(), result);
}
return new ExecutableActions.Results(results);
}

70
src/main/java/org/elasticsearch/watcher/actions/ActionWrapper.java
View File

@ -6,9 +6,13 @@
package org.elasticsearch.watcher.actions;
import org.elasticsearch.common.Nullable;
import org.elasticsearch.common.ParseField;
import org.elasticsearch.common.bytes.BytesReference;
import org.elasticsearch.common.xcontent.ToXContent;
import org.elasticsearch.common.xcontent.XContentBuilder;
import org.elasticsearch.common.xcontent.XContentHelper;
import org.elasticsearch.common.xcontent.XContentParser;
import org.elasticsearch.common.xcontent.json.JsonXContent;
import org.elasticsearch.watcher.execution.WatchExecutionContext;
import org.elasticsearch.watcher.execution.Wid;
import org.elasticsearch.watcher.transform.ExecutableTransform;
@ -18,6 +22,8 @@ import org.elasticsearch.watcher.watch.Payload;
import java.io.IOException;
import static org.elasticsearch.common.xcontent.XContentFactory.jsonBuilder;
/**
*
*/
@ -174,6 +180,7 @@ public class ActionWrapper implements ToXContent {
@Override
public XContentBuilder toXContent(XContentBuilder builder, Params params) throws IOException {
builder.startObject();
builder.field(Field.ID.getPreferredName(), id);
if (transform != null) {
builder.startObject(Transform.Field.TRANSFORM_RESULT.getPreferredName())
.field(transform.type(), transform, params)
@ -183,35 +190,70 @@ public class ActionWrapper implements ToXContent {
return builder.endObject();
}
static Result parse(Wid wid, String actionId, XContentParser parser, ActionRegistry actionRegistry, TransformRegistry transformRegistry) throws IOException {
static Result parse(Wid wid, XContentParser parser, ActionRegistry actionRegistry, TransformRegistry transformRegistry) throws IOException {
assert parser.currentToken() == XContentParser.Token.START_OBJECT;
String id = null;
Transform.Result transformResult = null;
Action.Result actionResult = null;
ActionFactory actionFactory = null;
BytesReference actionResultSource = null;
String currentFieldName = null;
XContentParser.Token token;
while ((token = parser.nextToken()) != XContentParser.Token.END_OBJECT) {
if (token == XContentParser.Token.FIELD_NAME) {
currentFieldName = parser.currentName();
} else {
if (Transform.Field.TRANSFORM.match(currentFieldName)) {
transformResult = transformRegistry.parseResult(wid.watchId(), parser);
} else if (Transform.Field.TRANSFORM.match(currentFieldName)) {
transformResult = transformRegistry.parseResult(wid.watchId(), parser);
} else if (Field.ID.match(currentFieldName)) {
if (token == XContentParser.Token.VALUE_STRING) {
id = parser.text();
} else {
// it's the type of the action
ActionFactory actionFactory = actionRegistry.factory(currentFieldName);
if (actionFactory == null) {
throw new ActionException("could not parse action result [{}/{}]. unknown action type [{}]", wid, actionId, currentFieldName);
}
actionResult = actionFactory.parseResult(wid, actionId, parser);
throw new ActionException("could not parse action result for watch [{}]. expected a string value for [{}] but found [{}] instead", wid, currentFieldName, token);
}
} else {
// it's the type of the action
// here we don't directly parse the action type. instead we'll collect
// the bytes of the structure that makes the action result. The reason
// for this is that we want to make sure to pass the action id to the
// action factory when we parse the result (so that error messages will
// point to the action result that failed to parse). It's an overhead,
// but for worth it for usability purposes.
actionFactory = actionRegistry.factory(currentFieldName);
if (actionFactory == null) {
throw new ActionException("could not parse action result for watch [{}]. unknown action type [{}]", wid, currentFieldName);
}
// it would have been nice if we had access to the underlying byte offset
// of the parser... but for now we'll just need to create a new json
// builder with its own (new) byte array and copy over the content.
XContentBuilder resultBuilder = jsonBuilder();
XContentHelper.copyCurrentStructure(resultBuilder.generator(), parser);
actionResultSource = resultBuilder.bytes();
}
}
if (actionResult == null) {
throw new ActionException("could not parse watch action result [{}/{}]. missing action result type", wid, actionId);
if (id == null) {
throw new ActionException("could not parse watch action result for watch [{}]. missing required [{}] field", wid, Field.ID.getPreferredName());
}
return new Result(actionId, transformResult, actionResult);
if (actionFactory == null) {
throw new ActionException("could not parse watch action result for watch [{}]. missing action result type", wid);
}
assert actionResultSource != null : "if we parsed the type name we must have collected the type bytes";
parser = JsonXContent.jsonXContent.createParser(actionResultSource);
parser.nextToken();
Action.Result actionResult = actionFactory.parseResult(wid, id, parser);
return new Result(id, transformResult, actionResult);
}
}
interface Field {
ParseField ID = new ParseField("id");
}
}

6
src/main/java/org/elasticsearch/watcher/actions/ExecutableActions.java
View File

@ -97,11 +97,11 @@ public class ExecutableActions implements Iterable<ActionWrapper>, ToXContent {
@Override
public XContentBuilder toXContent(XContentBuilder builder, Params params) throws IOException {
builder.startObject();
builder.startArray();
for (ActionWrapper.Result result : results.values()) {
builder.field(result.id(), result, params);
result.toXContent(builder, params);
}
return builder.endObject();
return builder.endArray();
}
}
}

63
src/main/java/org/elasticsearch/watcher/execution/WatchExecutionResult.java
View File

@ -13,7 +13,6 @@ import org.elasticsearch.common.xcontent.XContentBuilder;
import org.elasticsearch.common.xcontent.XContentParser;
import org.elasticsearch.watcher.WatcherException;
import org.elasticsearch.watcher.actions.ActionRegistry;
import org.elasticsearch.watcher.actions.ActionWrapper;
import org.elasticsearch.watcher.actions.ExecutableActions;
import org.elasticsearch.watcher.condition.Condition;
import org.elasticsearch.watcher.condition.ConditionRegistry;
@ -80,21 +79,21 @@ public class WatchExecutionResult implements ToXContent {
@Override
public XContentBuilder toXContent(XContentBuilder builder, Params params) throws IOException {
builder.startObject();
WatcherDateUtils.writeDate(Parser.EXECUTION_TIME_FIELD.getPreferredName(), builder, executionTime);
WatcherDateUtils.writeDate(Field.EXECUTION_TIME.getPreferredName(), builder, executionTime);
if (inputResult != null) {
builder.startObject(Parser.INPUT_RESULT_FIELD.getPreferredName())
builder.startObject(Field.INPUT.getPreferredName())
.field(inputResult.type(), inputResult, params)
.endObject();
}
if (conditionResult != null) {
builder.startObject(Parser.CONDITION_RESULT_FIELD.getPreferredName())
builder.startObject(Field.CONDITION.getPreferredName())
.field(conditionResult.type(), conditionResult, params)
.endObject();
}
if (throttleResult != null && throttleResult.throttle()) {
builder.field(Parser.THROTTLED.getPreferredName(), throttleResult.throttle());
builder.field(Field.THROTTLED.getPreferredName(), throttleResult.throttle());
if (throttleResult.reason() != null) {
builder.field(Parser.THROTTLE_REASON.getPreferredName(), throttleResult.reason());
builder.field(Field.THROTTLE_REASON.getPreferredName(), throttleResult.reason());
}
}
if (transformResult != null) {
@ -102,24 +101,13 @@ public class WatchExecutionResult implements ToXContent {
.field(transformResult.type(), transformResult, params)
.endObject();
}
builder.startObject(Parser.ACTIONS_RESULTS.getPreferredName());
for (ActionWrapper.Result actionResult : actionsResults) {
builder.field(actionResult.id(), actionResult, params);
}
builder.endObject();
builder.field(Field.ACTIONS.getPreferredName(), actionsResults, params);
builder.endObject();
return builder;
}
public static class Parser {
public static final ParseField EXECUTION_TIME_FIELD = new ParseField("execution_time");
public static final ParseField INPUT_RESULT_FIELD = new ParseField("input_result");
public static final ParseField CONDITION_RESULT_FIELD = new ParseField("condition_result");
public static final ParseField ACTIONS_RESULTS = new ParseField("actions_results");
public static final ParseField THROTTLED = new ParseField("throttled");
public static final ParseField THROTTLE_REASON = new ParseField("throttle_reason");
public static WatchExecutionResult parse(Wid wid, XContentParser parser, ConditionRegistry conditionRegistry, ActionRegistry actionRegistry,
InputRegistry inputRegistry, TransformRegistry transformRegistry) throws IOException {
DateTime executionTime = null;
@ -135,43 +123,56 @@ public class WatchExecutionResult implements ToXContent {
while ((token = parser.nextToken()) != XContentParser.Token.END_OBJECT) {
if (token == XContentParser.Token.FIELD_NAME) {
currentFieldName = parser.currentName();
} else if (EXECUTION_TIME_FIELD.match(currentFieldName)) {
} else if (Field.EXECUTION_TIME.match(currentFieldName)) {
try {
executionTime = WatcherDateUtils.parseDate(currentFieldName, parser, UTC);
} catch (WatcherDateUtils.ParseException pe) {
throw new WatcherException("unable to parse watch execution [{}]. failed to parse date field [{}]", pe, wid, currentFieldName);
throw new WatcherException("could not parse watch execution [{}]. failed to parse date field [{}]", pe, wid, currentFieldName);
}
} else if (token.isValue()) {
if (THROTTLE_REASON.match(currentFieldName)) {
if (Field.THROTTLE_REASON.match(currentFieldName)) {
throttleReason = parser.text();
} else if (THROTTLED.match(currentFieldName)) {
} else if (Field.THROTTLED.match(currentFieldName)) {
throttled = parser.booleanValue();
} else {
throw new WatcherException("unable to parse watch execution [{}]. unexpected field [{}]", wid, currentFieldName);
throw new WatcherException("could not parse watch execution [{}]. unexpected field [{}]", wid, currentFieldName);
}
} else if (token == XContentParser.Token.START_ARRAY) {
if (Field.ACTIONS.match(currentFieldName)) {
actionResults = actionRegistry.parseResults(wid, parser);
} else {
throw new WatcherException("could not parse watch execution [{}]. unexpected field [{}]", wid, currentFieldName);
}
} else if (token == XContentParser.Token.START_OBJECT) {
if (INPUT_RESULT_FIELD.match(currentFieldName)) {
if (Field.INPUT.match(currentFieldName)) {
inputResult = inputRegistry.parseResult(wid.watchId(), parser);
} else if (CONDITION_RESULT_FIELD.match(currentFieldName)) {
} else if (Field.CONDITION.match(currentFieldName)) {
conditionResult = conditionRegistry.parseResult(wid.watchId(), parser);
} else if (Transform.Field.TRANSFORM_RESULT.match(currentFieldName)) {
transformResult = transformRegistry.parseResult(wid.watchId(), parser);
} else if (ACTIONS_RESULTS.match(currentFieldName)) {
actionResults = actionRegistry.parseResults(wid, parser);
} else {
throw new WatcherException("unable to parse watch execution. unexpected field [" + currentFieldName + "]");
throw new WatcherException("could not parse watch execution [{}]. unexpected field [{}]", wid, currentFieldName);
}
} else {
throw new WatcherException("unable to parse watch execution. unexpected token [" + token + "]");
throw new WatcherException("could not parse watch execution [{}]. unexpected token [{}]", wid, token);
}
}
if (executionTime == null) {
throw new WatcherException("unable to parse watch execution [{}]. missing required date field [{}]", wid, EXECUTION_TIME_FIELD.getPreferredName());
throw new WatcherException("could not parse watch execution [{}]. missing required date field [{}]", wid, Field.EXECUTION_TIME.getPreferredName());
}
Throttler.Result throttleResult = throttled ? Throttler.Result.throttle(throttleReason) : Throttler.Result.NO;
return new WatchExecutionResult(executionTime, inputResult, conditionResult, throttleResult, transformResult, actionResults);
}
}
interface Field {
ParseField EXECUTION_TIME = new ParseField("execution_time");
ParseField INPUT = new ParseField("input");
ParseField CONDITION = new ParseField("condition");
ParseField ACTIONS = new ParseField("actions");
ParseField THROTTLED = new ParseField("throttled");
ParseField THROTTLE_REASON = new ParseField("throttle_reason");
}
}

2
src/main/java/org/elasticsearch/watcher/history/HistoryStore.java
View File

@ -368,7 +368,7 @@ public class HistoryStore extends AbstractComponent implements NodeSettingsServi
private SearchRequest createScanSearchRequest(WatchRecord.State recordState) {
SearchSourceBuilder sourceBuilder = new SearchSourceBuilder()
.query(QueryBuilders.termQuery(WatchRecord.Parser.STATE_FIELD.getPreferredName(), recordState.id()))
.query(QueryBuilders.termQuery(WatchRecord.Field.STATE.getPreferredName(), recordState.id()))
.size(scrollSize)
.version(true);

40
src/main/java/org/elasticsearch/watcher/history/WatchRecord.java
View File

@ -131,8 +131,8 @@ public class WatchRecord implements ToXContent {
@Override
public XContentBuilder toXContent(XContentBuilder builder, Params params) throws IOException {
builder.startObject();
builder.field(Parser.WATCH_ID_FIELD.getPreferredName(), watchId);
builder.startObject(Parser.TRIGGER_EVENT_FIELD.getPreferredName())
builder.field(Field.WATCH_ID.getPreferredName(), watchId);
builder.startObject(Field.TRIGGER_EVENT.getPreferredName())
.field(triggerEvent.type(), triggerEvent, params)
.endObject();
builder.startObject(Watch.Parser.INPUT_FIELD.getPreferredName())
@ -141,17 +141,17 @@ public class WatchRecord implements ToXContent {
builder.startObject(Watch.Parser.CONDITION_FIELD.getPreferredName())
.field(condition.type(), condition, params)
.endObject();
builder.field(Parser.STATE_FIELD.getPreferredName(), state.id());
builder.field(Field.STATE.getPreferredName(), state.id());
if (message != null) {
builder.field(Parser.MESSAGE_FIELD.getPreferredName(), message);
builder.field(Field.MESSAGE.getPreferredName(), message);
}
if (metadata != null) {
builder.field(Parser.METADATA_FIELD.getPreferredName(), metadata);
builder.field(Field.METADATA.getPreferredName(), metadata);
}
if (execution != null) {
builder.field(Parser.WATCH_EXECUTION_FIELD.getPreferredName(), execution, params);
builder.field(Field.EXECUTION_RESULT.getPreferredName(), execution, params);
}
builder.endObject();
@ -209,13 +209,6 @@ public class WatchRecord implements ToXContent {
public static class Parser extends AbstractComponent {
public static final ParseField WATCH_ID_FIELD = new ParseField("watch_id");
public static final ParseField TRIGGER_EVENT_FIELD = new ParseField("trigger_event");
public static final ParseField MESSAGE_FIELD = new ParseField("message");
public static final ParseField STATE_FIELD = new ParseField("state");
public static final ParseField METADATA_FIELD = new ParseField("metadata");
public static final ParseField WATCH_EXECUTION_FIELD = new ParseField("watch_execution");
private final ConditionRegistry conditionRegistry;
private final ActionRegistry actionRegistry;
private final InputRegistry inputRegistry;
@ -257,21 +250,21 @@ public class WatchRecord implements ToXContent {
record.input = inputRegistry.parse(id, parser);
} else if (Watch.Parser.CONDITION_FIELD.match(currentFieldName)) {
record.condition = conditionRegistry.parseCondition(id, parser);
} else if (METADATA_FIELD.match(currentFieldName)) {
} else if (Field.METADATA.match(currentFieldName)) {
record.metadata = parser.map();
} else if (WATCH_EXECUTION_FIELD.match(currentFieldName)) {
} else if (Field.EXECUTION_RESULT.match(currentFieldName)) {
record.execution = WatchExecutionResult.Parser.parse(record.id, parser, conditionRegistry, actionRegistry, inputRegistry, transformRegistry);
} else if (TRIGGER_EVENT_FIELD.match(currentFieldName)) {
} else if (Field.TRIGGER_EVENT.match(currentFieldName)) {
record.triggerEvent = triggerService.parseTriggerEvent(record.watchId, id, parser);
} else {
throw new WatcherException("could not parse watch record [{}]. unexpected field [{}]", id, currentFieldName);
}
} else if (token.isValue()) {
if (WATCH_ID_FIELD.match(currentFieldName)) {
if (Field.WATCH_ID.match(currentFieldName)) {
record.watchId = parser.text();
} else if (MESSAGE_FIELD.match(currentFieldName)) {
} else if (Field.MESSAGE.match(currentFieldName)) {
record.message = parser.textOrNull();
} else if (STATE_FIELD.match(currentFieldName)) {
} else if (Field.STATE.match(currentFieldName)) {
record.state = State.resolve(parser.text());
} else {
throw new WatcherException("could not parse watch record [{}]. unexpected field [{}]", id, currentFieldName);
@ -290,4 +283,13 @@ public class WatchRecord implements ToXContent {
return record;
}
}
public interface Field {
ParseField WATCH_ID = new ParseField("watch_id");
ParseField TRIGGER_EVENT = new ParseField("trigger_event");
ParseField MESSAGE = new ParseField("message");
ParseField STATE = new ParseField("state");
ParseField METADATA = new ParseField("metadata");
ParseField EXECUTION_RESULT = new ParseField("execution_result");
}
}

3
src/main/java/org/elasticsearch/watcher/support/WatcherUtils.java
View File

@ -59,7 +59,8 @@ public final class WatcherUtils {
SearchRequest request = new SearchRequest(requestPrototype)
.indicesOptions(requestPrototype.indicesOptions())
.searchType(requestPrototype.searchType())
.indices(requestPrototype.indices());
.indices(requestPrototype.indices())
.types(requestPrototype.types());
// TODO: Revise this search template conversion code once search templates in core have been refactored once ES 2.0 is released.
// Due the inconsistency with templates in ES 1.x, we maintain our own template format.

8
src/main/java/org/elasticsearch/watcher/transform/search/SearchTransform.java
View File

@ -81,6 +81,10 @@ public class SearchTransform implements Transform {
this.executedRequest = executedRequest;
}
public SearchRequest executedRequest() {
return executedRequest;
}
@Override
protected XContentBuilder xContentBody(XContentBuilder builder, Params params) throws IOException {
builder.field(Field.EXECUTED_REQUEST.getPreferredName());
@ -122,10 +126,6 @@ public class SearchTransform implements Transform {
return new SearchTransform.Result(executedRequest, payload);
}
public SearchRequest executedRequest() {
return executedRequest;
}
}
public static class Builder implements Transform.Builder<SearchTransform> {

238
src/main/resources/watch_history.json
View File

@ -3,64 +3,56 @@
"order": 2147483647,
"settings": {
"index.number_of_shards": 1,
"index.mapper.dynamic" : false
"index.mapper.dynamic": false
},
"mappings": {
"watch_record": {
"dynamic_templates" : [
"dynamic_templates": [
{
"email_not_analyzed_fields": {
"path_match": ".*\\.email\\.(from|to|cc|bcc|reply_to|id)",
"match_pattern" : "regex",
"match_mapping_type": "string",
"disabled_fields": {
"path_match": "execution_result\\.input\\..+\\.payload",
"match_pattern": "regex",
"mapping": {
"type": "string",
"index": "not_analyzed"
}
}
},
{
"time_fields": {
"path_match": ".*(execution|triggered|scheduled)_time",
"match_pattern" : "regex",
"mapping": {
"type": "date"
}
}
},
{
"http_not_analyzed_fields": {
"path_match": "watch_execution\\.((actions_results\\..+\\.webhook\\.request)|input_result\\.http\\.sent_request)\\.(path|host)",
"match_pattern" : "regex",
"mapping": {
"type": "string",
"index": "not_analyzed"
}
}
},
{
"disabled_fields" : {
"path_match": "watch_execution\\.input_result\\..+\\.payload",
"match_pattern" : "regex",
"mapping": {
"type" : "object",
"enabled" : false
"type": "object",
"enabled": false
}
}
}
],
"dynamic" : "strict",
"_all" : {
"enabled" : false
"dynamic": "strict",
"_all": {
"enabled": false
},
"properties": {
"watch_id": {
"type": "string",
"index": "not_analyzed"
},
"trigger_event" : {
"type" : "object",
"dynamic" : true
"trigger_event": {
"type": "object",
"dynamic": true,
"properties": {
"schedule": {
"type": "object",
"dynamic": true,
"properties": {
"triggered_time": {
"type": "date"
},
"scheduled_time": {
"type": "date"
}
}
}
}
},
"input": {
"type": "object",
"enabled": false
},
"condition": {
"type": "object",
"enabled": false
},
"state": {
"type": "string",
@ -69,20 +61,154 @@
"message": {
"type": "string"
},
"condition" : {
"type" : "object",
"dynamic" : true
"execution_result": {
"type": "object",
"dynamic": true,
"properties": {
"execution_time": {
"type": "date"
},
"input": {
"type": "object",
"dynamic": true,
"properties": {
"search": {
"type": "object",
"dynamic": true,
"properties": {
"executed_request": {
"type": "object",
"dynamic": true,
"properties": {
"search_type": {
"type": "string",
"index": "not_analyzed"
},
"indices": {
"type": "string",
"index": "not_analyzed"
},
"types": {
"type": "string",
"index": "not_analyzed"
}
}
}
}
},
"http": {
"type": "object",
"dynamic": true,
"properties": {
"sent_request": {
"type": "object",
"dynamic": true,
"properties": {
"path": {
"type": "string",
"index": "not_analyzed"
},
"host": {
"type": "string",
"index": "not_analyzed"
}
}
}
}
}
}
},
"actions": {
"type": "nested",
"include_in_parent": true,
"dynamic": true,
"properties": {
"email": {
"type": "object",
"dynamic": true,
"properties": {
"email": {
"type": "object",
"dynamic": true,
"properties": {
"id": {
"type": "string",
"index": "not_analyzed"
},
"from": {
"type": "string",
"index": "not_analyzed"
},
"reply_to": {
"type": "string",
"index": "not_analyzed"
},
"to": {
"type": "string",
"index": "not_analyzed"
},
"cc": {
"type": "string",
"index": "not_analyzed"
},
"bcc": {
"type": "string",
"index": "not_analyzed"
}
}
}
}
},
"webhook": {
"type": "object",
"dynamic": true,
"properties": {
"request": {
"type": "object",
"dynamic": true,
"properties": {
"path": {
"type": "string",
"index": "not_analyzed"
},
"host": {
"type": "string",
"index": "not_analyzed"
}
}
}
}
},
"index": {
"type": "object",
"dynamic": true,
"properties": {
"response": {
"type": "object",
"dynamic": true,
"properties": {
"_index": {
"type": "string",
"index": "not_analyzed"
},
"_type": {
"type": "string",
"index": "not_analyzed"
},
"_id": {
"type": "string",
"index": "not_analyzed"
}
}
}
}
}
}
}
}
},
"input" : {
"type" : "object",
"dynamic" : true
},
"watch_execution" : {
"type" : "object",
"dynamic" : true
},
"metadata" : {
"type" : "object",
"metadata": {
"type": "object",
"dynamic": true
}
}

5
src/test/java/org/elasticsearch/watcher/license/LicenseIntegrationTests.java
View File

@ -162,11 +162,10 @@ public class LicenseIntegrationTests extends AbstractWatcherIntegrationTests {
// and last... lets verify that we have throttled watches due to license expiration
long throttledCount = docCount(HistoryStore.INDEX_PREFIX + "*", HistoryStore.DOC_TYPE, filteredQuery(
matchQuery("watch_execution.throttle_reason", "watcher license expired"),
termFilter("watch_execution.throttled", true)));
matchQuery("execution_result.throttle_reason", "watcher license expired"),
termFilter("execution_result.throttled", true)));
assertThat(throttledCount, is(1L));
//=====
// now... lets verify that all the watcher APIs are blocked when the license is disabled
//=====

2
src/test/java/org/elasticsearch/watcher/test/AbstractWatcherIntegrationTests.java
View File

@ -326,7 +326,7 @@ public abstract class AbstractWatcherIntegrationTests extends ElasticsearchInteg
.get();
assertThat("could not find executed watch record", searchResponse.getHits().getTotalHits(), greaterThanOrEqualTo(minimumExpectedWatchActionsWithActionPerformed));
if (assertConditionMet) {
assertThat((Integer) XContentMapValues.extractValue("watch_execution.input_result.search.payload.hits.total", searchResponse.getHits().getAt(0).sourceAsMap()), greaterThanOrEqualTo(1));
assertThat((Integer) XContentMapValues.extractValue("execution_result.input.search.payload.hits.total", searchResponse.getHits().getAt(0).sourceAsMap()), greaterThanOrEqualTo(1));
}
}
});

11
src/test/java/org/elasticsearch/watcher/test/WatcherTestUtils.java
View File

@ -202,7 +202,6 @@ public final class WatcherTestUtils {
new Watch.Status());
}
public static ScriptServiceProxy getScriptServiceProxy(ThreadPool tp) throws IOException {
Settings settings = ImmutableSettings.settingsBuilder()
.put(ScriptService.DISABLE_DYNAMIC_SCRIPTING_SETTING, "none")
@ -216,12 +215,12 @@ public final class WatcherTestUtils {
return ScriptServiceProxy.of(new ScriptService(settings, new Environment(), engineServiceSet, new ResourceWatcherService(settings, tp), nodeSettingsService));
}
public static SearchType getRandomSupportedSearchType() {
Set<SearchType> searchTypes = new HashSet<>();
searchTypes.addAll(Arrays.asList(SearchType.values()));
searchTypes.remove(SearchType.SCAN);
return randomFrom(searchTypes.toArray(new SearchType[searchTypes.size()]));
return randomFrom(
SearchType.COUNT,
SearchType.DFS_QUERY_AND_FETCH,
SearchType.DFS_QUERY_THEN_FETCH,
SearchType.DFS_QUERY_AND_FETCH);
}
public static boolean areJsonEquivalent(String json1, String json2) throws IOException {

10
src/test/java/org/elasticsearch/watcher/test/integration/BasicWatcherTests.java
View File

@ -14,7 +14,6 @@ import org.elasticsearch.common.joda.time.DateTime;
import org.elasticsearch.common.unit.TimeValue;
import org.elasticsearch.common.xcontent.ToXContent;
import org.elasticsearch.common.xcontent.XContentBuilder;
import org.elasticsearch.script.ScriptService;
import org.elasticsearch.search.builder.SearchSourceBuilder;
import org.elasticsearch.test.junit.annotations.TestLogging;
import org.elasticsearch.watcher.WatcherException;
@ -25,6 +24,7 @@ import org.elasticsearch.watcher.condition.ConditionBuilders;
import org.elasticsearch.watcher.history.HistoryStore;
import org.elasticsearch.watcher.support.WatcherUtils;
import org.elasticsearch.watcher.support.template.Template;
import org.elasticsearch.watcher.support.xcontent.XContentSource;
import org.elasticsearch.watcher.test.AbstractWatcherIntegrationTests;
import org.elasticsearch.watcher.transport.actions.delete.DeleteWatchResponse;
import org.elasticsearch.watcher.transport.actions.get.GetWatchResponse;
@ -37,8 +37,6 @@ import org.elasticsearch.watcher.trigger.schedule.support.WeekTimes;
import org.elasticsearch.watcher.watch.WatchStore;
import org.junit.Test;
import java.util.Map;
import static org.elasticsearch.common.xcontent.XContentFactory.jsonBuilder;
import static org.elasticsearch.index.query.FilterBuilders.rangeFilter;
import static org.elasticsearch.index.query.QueryBuilders.*;
@ -365,10 +363,8 @@ public class BasicWatcherTests extends AbstractWatcherIntegrationTests {
.setSize(1)
.get();
assertHitCount(searchResponse, 1);
Map payload = (Map) ((Map)((Map)((Map) searchResponse.getHits().getAt(0).sourceAsMap().get("watch_execution")).get("input_result")).get("search")).get("payload");
assertThat(payload.size(), equalTo(1));
assertThat(((Map) payload.get("hits")).size(), equalTo(1));
assertThat((Integer) ((Map) payload.get("hits")).get("total"), equalTo(1));
XContentSource source = new XContentSource(searchResponse.getHits().getAt(0).getSourceRef());
assertThat(source.getValue("execution_result.input.search.payload.hits.total"), equalTo((Object) 1));
}
@Test

32
src/test/java/org/elasticsearch/watcher/test/integration/BootStrapTests.java
View File

@ -144,8 +144,8 @@ public class BootStrapTests extends AbstractWatcherIntegrationTests {
String index = HistoryStore.getHistoryIndexNameForTime(now);
client().prepareIndex(index, HistoryStore.DOC_TYPE, wid.value())
.setSource(jsonBuilder().startObject()
.field(WatchRecord.Parser.WATCH_ID_FIELD.getPreferredName(), wid.watchId())
.startObject(WatchRecord.Parser.TRIGGER_EVENT_FIELD.getPreferredName())
.field(WatchRecord.Field.WATCH_ID.getPreferredName(), wid.watchId())
.startObject(WatchRecord.Field.TRIGGER_EVENT.getPreferredName())
.field(event.type(), event)
.endObject()
.startObject(Watch.Parser.CONDITION_FIELD.getPreferredName())
@ -154,7 +154,7 @@ public class BootStrapTests extends AbstractWatcherIntegrationTests {
.startObject(Watch.Parser.INPUT_FIELD.getPreferredName())
.startObject("none").endObject()
.endObject()
.field(WatchRecord.Parser.STATE_FIELD.getPreferredName(), WatchRecord.State.AWAITS_EXECUTION)
.field(WatchRecord.Field.STATE.getPreferredName(), WatchRecord.State.AWAITS_EXECUTION)
.endObject())
.setConsistencyLevel(WriteConsistencyLevel.ALL)
.setRefresh(true)
@ -164,8 +164,8 @@ public class BootStrapTests extends AbstractWatcherIntegrationTests {
wid = new Wid("_id", 2, now);
client().prepareIndex(index, HistoryStore.DOC_TYPE, wid.value())
.setSource(jsonBuilder().startObject()
.field(WatchRecord.Parser.WATCH_ID_FIELD.getPreferredName(), wid.watchId())
.startObject(WatchRecord.Parser.TRIGGER_EVENT_FIELD.getPreferredName())
.field(WatchRecord.Field.WATCH_ID.getPreferredName(), wid.watchId())
.startObject(WatchRecord.Field.TRIGGER_EVENT.getPreferredName())
.field(event.type(), event)
.endObject()
.startObject(Watch.Parser.CONDITION_FIELD.getPreferredName())
@ -174,7 +174,7 @@ public class BootStrapTests extends AbstractWatcherIntegrationTests {
.startObject(Watch.Parser.INPUT_FIELD.getPreferredName())
.startObject("none").endObject()
.endObject()
.field(WatchRecord.Parser.STATE_FIELD.getPreferredName(), WatchRecord.State.AWAITS_EXECUTION)
.field(WatchRecord.Field.STATE.getPreferredName(), WatchRecord.State.AWAITS_EXECUTION)
.endObject())
.setConsistencyLevel(WriteConsistencyLevel.ALL)
.setRefresh(true)
@ -184,8 +184,8 @@ public class BootStrapTests extends AbstractWatcherIntegrationTests {
wid = new Wid("_id", 2, now);
client().prepareIndex(index, HistoryStore.DOC_TYPE, wid.value())
.setSource(jsonBuilder().startObject()
.field(WatchRecord.Parser.WATCH_ID_FIELD.getPreferredName(), wid.watchId())
.startObject(WatchRecord.Parser.TRIGGER_EVENT_FIELD.getPreferredName())
.field(WatchRecord.Field.WATCH_ID.getPreferredName(), wid.watchId())
.startObject(WatchRecord.Field.TRIGGER_EVENT.getPreferredName())
.startObject("unknown").endObject()
.endObject()
.startObject(Watch.Parser.CONDITION_FIELD.getPreferredName())
@ -194,7 +194,7 @@ public class BootStrapTests extends AbstractWatcherIntegrationTests {
.startObject(Watch.Parser.INPUT_FIELD.getPreferredName())
.startObject("none").endObject()
.endObject()
.field(WatchRecord.Parser.STATE_FIELD.getPreferredName(), WatchRecord.State.AWAITS_EXECUTION)
.field(WatchRecord.Field.STATE.getPreferredName(), WatchRecord.State.AWAITS_EXECUTION)
.endObject())
.setConsistencyLevel(WriteConsistencyLevel.ALL)
.setRefresh(true)
@ -218,8 +218,8 @@ public class BootStrapTests extends AbstractWatcherIntegrationTests {
String index = HistoryStore.getHistoryIndexNameForTime(now);
client().prepareIndex(index, HistoryStore.DOC_TYPE, wid.value())
.setSource(jsonBuilder().startObject()
.field(WatchRecord.Parser.WATCH_ID_FIELD.getPreferredName(), wid.value())
.startObject(WatchRecord.Parser.TRIGGER_EVENT_FIELD.getPreferredName())
.field(WatchRecord.Field.WATCH_ID.getPreferredName(), wid.value())
.startObject(WatchRecord.Field.TRIGGER_EVENT.getPreferredName())
.field(event.type(), event)
.endObject()
.startObject(Watch.Parser.CONDITION_FIELD.getPreferredName())
@ -228,7 +228,7 @@ public class BootStrapTests extends AbstractWatcherIntegrationTests {
.startObject(Watch.Parser.INPUT_FIELD.getPreferredName())
.startObject("none").endObject()
.endObject()
.field(WatchRecord.Parser.STATE_FIELD.getPreferredName(), WatchRecord.State.AWAITS_EXECUTION)
.field(WatchRecord.Field.STATE.getPreferredName(), WatchRecord.State.AWAITS_EXECUTION)
.endObject())
.setConsistencyLevel(WriteConsistencyLevel.ALL)
.setRefresh(true)
@ -240,8 +240,8 @@ public class BootStrapTests extends AbstractWatcherIntegrationTests {
refresh();
SearchResponse searchResponse = client().prepareSearch(index).get();
assertHitCount(searchResponse, 1);
assertThat(searchResponse.getHits().getAt(0).sourceAsMap().get(WatchRecord.Parser.WATCH_ID_FIELD.getPreferredName()).toString(), Matchers.equalTo(wid.value()));
assertThat(searchResponse.getHits().getAt(0).sourceAsMap().get(WatchRecord.Parser.STATE_FIELD.getPreferredName()).toString(), Matchers.equalTo(WatchRecord.State.DELETED_WHILE_QUEUED.toString()));
assertThat(searchResponse.getHits().getAt(0).sourceAsMap().get(WatchRecord.Field.WATCH_ID.getPreferredName()).toString(), Matchers.equalTo(wid.value()));
assertThat(searchResponse.getHits().getAt(0).sourceAsMap().get(WatchRecord.Field.STATE.getPreferredName()).toString(), Matchers.equalTo(WatchRecord.State.DELETED_WHILE_QUEUED.toString()));
}
@ -365,7 +365,7 @@ public class BootStrapTests extends AbstractWatcherIntegrationTests {
state = randomFrom(WatchRecord.State.AWAITS_EXECUTION, WatchRecord.State.CHECKING, WatchRecord.State.EXECUTION_NOT_NEEDED, WatchRecord.State.EXECUTED);
}
client().prepareUpdate(index, HistoryStore.DOC_TYPE, watchRecord.id().value())
.setDoc(WatchRecord.Parser.STATE_FIELD.getPreferredName(), state.id())
.setDoc(WatchRecord.Field.STATE.getPreferredName(), state.id())
.get();
if (state == WatchRecord.State.AWAITS_EXECUTION) {
awaitsExecution++;
@ -448,7 +448,7 @@ public class BootStrapTests extends AbstractWatcherIntegrationTests {
@Override
public void run() {
long count = docCount(HistoryStore.INDEX_PREFIX + "*", HistoryStore.DOC_TYPE,
termQuery(WatchRecord.Parser.STATE_FIELD.getPreferredName(), WatchRecord.State.EXECUTED.id()));
termQuery(WatchRecord.Field.STATE.getPreferredName(), WatchRecord.State.EXECUTED.id()));
assertThat(count, is(totalHistoryEntries));
}
}, 30, TimeUnit.SECONDS);

9
src/test/java/org/elasticsearch/watcher/test/integration/EmailSecretsIntegrationTests.java
View File

@ -26,6 +26,7 @@ import org.junit.After;
import org.junit.Test;
import javax.mail.internet.MimeMessage;
import java.util.List;
import java.util.Map;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
@ -143,9 +144,11 @@ public class EmailSecretsIntegrationTests extends AbstractWatcherIntegrationTest
.get();
assertThat(executeResponse, notNullValue());
contentSource = executeResponse.getSource();
value = contentSource.getValue("watch_execution.actions_results._email.email.success");
assertThat(value, notNullValue());
assertThat(value, is((Object) Boolean.TRUE));
value = contentSource.getValue("execution_result.actions.email.success");
assertThat(value, instanceOf(List.class));
List<Boolean> values = (List<Boolean>) value;
assertThat(values, hasSize(1));
assertThat(values, hasItem(Boolean.TRUE));
if (!latch.await(5, TimeUnit.SECONDS)) {
fail("waiting too long for the email to be sent");

10
src/test/java/org/elasticsearch/watcher/test/integration/HistoryTemplateEmailMappingsTests.java
View File

@ -102,11 +102,11 @@ public class HistoryTemplateEmailMappingsTests extends AbstractWatcherIntegratio
assertWatchWithMinimumActionsCount("_id", WatchRecord.State.EXECUTED, 1);
SearchResponse response = client().prepareSearch(HistoryStore.INDEX_PREFIX + "*").setSource(searchSource()
.aggregation(terms("from").field("watch_execution.actions_results._email.email.email.from"))
.aggregation(terms("to").field("watch_execution.actions_results._email.email.email.to"))
.aggregation(terms("cc").field("watch_execution.actions_results._email.email.email.cc"))
.aggregation(terms("bcc").field("watch_execution.actions_results._email.email.email.bcc"))
.aggregation(terms("reply_to").field("watch_execution.actions_results._email.email.email.reply_to"))
.aggregation(terms("from").field("execution_result.actions.email.email.from"))
.aggregation(terms("to").field("execution_result.actions.email.email.to"))
.aggregation(terms("cc").field("execution_result.actions.email.email.cc"))
.aggregation(terms("bcc").field("execution_result.actions.email.email.bcc"))
.aggregation(terms("reply_to").field("execution_result.actions.email.email.reply_to"))
.buildAsBytes())
.get();

5
src/test/java/org/elasticsearch/watcher/test/integration/HistoryTemplateHttpMappingsTests.java
View File

@ -98,8 +98,9 @@ public class HistoryTemplateHttpMappingsTests extends AbstractWatcherIntegration
assertWatchWithMinimumActionsCount("_id", WatchRecord.State.EXECUTED, 1);
SearchResponse response = client().prepareSearch(HistoryStore.INDEX_PREFIX + "*").setSource(searchSource()
.aggregation(terms("input_result_path").field("watch_execution.input_result.http.sent_request.path"))
.aggregation(terms("webhook_path").field("watch_execution.actions_results._webhook.webhook.request.path"))
.aggregation(terms("input_result_path").field("execution_result.input.http.sent_request.path"))
.aggregation(terms("input_result_host").field("execution_result.input.http.sent_request.host"))
.aggregation(terms("webhook_path").field("execution_result.actions.webhook.request.path"))
.buildAsBytes())
.get();

100
src/test/java/org/elasticsearch/watcher/test/integration/HistoryTemplateSearchInputMappingsTests.java Normal file
View File

@ -0,0 +1,100 @@
/*
* Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
* or more contributor license agreements. Licensed under the Elastic License;
* you may not use this file except in compliance with the Elastic License.
*/
package org.elasticsearch.watcher.test.integration;
import org.elasticsearch.action.search.SearchRequest;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.action.search.SearchType;
import org.elasticsearch.search.aggregations.Aggregations;
import org.elasticsearch.search.aggregations.bucket.terms.Terms;
import org.elasticsearch.watcher.history.HistoryStore;
import org.elasticsearch.watcher.history.WatchRecord;
import org.elasticsearch.watcher.test.AbstractWatcherIntegrationTests;
import org.elasticsearch.watcher.transport.actions.put.PutWatchResponse;
import org.junit.Test;
import static org.elasticsearch.search.aggregations.AggregationBuilders.terms;
import static org.elasticsearch.search.builder.SearchSourceBuilder.searchSource;
import static org.elasticsearch.watcher.actions.ActionBuilders.loggingAction;
import static org.elasticsearch.watcher.client.WatchSourceBuilders.watchBuilder;
import static org.elasticsearch.watcher.condition.ConditionBuilders.alwaysCondition;
import static org.elasticsearch.watcher.input.InputBuilders.searchInput;
import static org.elasticsearch.watcher.trigger.TriggerBuilders.schedule;
import static org.elasticsearch.watcher.trigger.schedule.Schedules.interval;
import static org.hamcrest.Matchers.is;
import static org.hamcrest.Matchers.notNullValue;
/**
* This test makes sure that the http host and path fields in the watch_record action result are
* not analyzed so they can be used in aggregations
*/
public class HistoryTemplateSearchInputMappingsTests extends AbstractWatcherIntegrationTests {
@Override
protected boolean timeWarped() {
return true; // just to have better control over the triggers
}
@Override
protected boolean enableShield() {
return false; // remove shield noise from this test
}
@Test
public void testHttpFields() throws Exception {
String index = "the-index";
String type = "the-type";
createIndex(index);
index(index, type, "{}");
flush();
refresh();
PutWatchResponse putWatchResponse = watcherClient().preparePutWatch("_id").setSource(watchBuilder()
.trigger(schedule(interval("5s")))
.input(searchInput(new SearchRequest().indices(index).types(type).searchType(SearchType.QUERY_AND_FETCH)))
.condition(alwaysCondition())
.addAction("logger", loggingAction("indexed")))
.get();
assertThat(putWatchResponse.isCreated(), is(true));
timeWarp().scheduler().trigger("_id");
flush();
refresh();
// the action should fail as no email server is available
assertWatchWithMinimumActionsCount("_id", WatchRecord.State.EXECUTED, 1);
SearchResponse response = client().prepareSearch(HistoryStore.INDEX_PREFIX + "*").setSource(searchSource()
.aggregation(terms("input_search_type").field("execution_result.input.search.executed_request.search_type"))
.aggregation(terms("input_indices").field("execution_result.input.search.executed_request.indices"))
.aggregation(terms("input_types").field("execution_result.input.search.executed_request.types"))
.buildAsBytes())
.get();
assertThat(response, notNullValue());
assertThat(response.getHits().getTotalHits(), is(1L));
Aggregations aggs = response.getAggregations();
assertThat(aggs, notNullValue());
Terms terms = aggs.get("input_search_type");
assertThat(terms, notNullValue());
assertThat(terms.getBuckets().size(), is(1));
assertThat(terms.getBucketByKey("query_and_fetch"), notNullValue());
assertThat(terms.getBucketByKey("query_and_fetch").getDocCount(), is(1L));
terms = aggs.get("input_indices");
assertThat(terms, notNullValue());
assertThat(terms.getBuckets().size(), is(1));
assertThat(terms.getBucketByKey(index), notNullValue());
assertThat(terms.getBucketByKey(index).getDocCount(), is(1L));
terms = aggs.get("input_types");
assertThat(terms, notNullValue());
assertThat(terms.getBuckets().size(), is(1));
assertThat(terms.getBucketByKey(type), notNullValue());
assertThat(terms.getBucketByKey(type).getDocCount(), is(1L));
}
}

2
src/test/java/org/elasticsearch/watcher/test/integration/HistoryTemplateTimeMappingsTests.java
View File

@ -76,7 +76,7 @@ public class HistoryTemplateTimeMappingsTests extends AbstractWatcherIntegration
logger.info("checking index [{}] with metadata:\n[{}]", metadatas.key, metadata.source().toString());
assertThat(extractValue("properties.trigger_event.properties.schedule.properties.scheduled_time.type", source), is((Object) "date"));
assertThat(extractValue("properties.trigger_event.properties.schedule.properties.triggered_time.type", source), is((Object) "date"));
assertThat(extractValue("properties.watch_execution.properties.execution_time.type", source), is((Object) "date"));
assertThat(extractValue("properties.execution_result.properties.execution_time.type", source), is((Object) "date"));
} catch (IOException e) {
throw new RuntimeException(e);
}

8
src/test/java/org/elasticsearch/watcher/test/integration/HttpInputIntegrationTest.java
View File

@ -17,12 +17,12 @@ import org.elasticsearch.watcher.history.HistoryStore;
import org.elasticsearch.watcher.support.http.HttpRequestTemplate;
import org.elasticsearch.watcher.support.http.auth.basic.BasicAuth;
import org.elasticsearch.watcher.support.template.Template;
import org.elasticsearch.watcher.support.xcontent.XContentSource;
import org.elasticsearch.watcher.test.AbstractWatcherIntegrationTests;
import org.elasticsearch.watcher.trigger.schedule.IntervalSchedule;
import org.junit.Test;
import java.net.InetSocketAddress;
import java.util.Map;
import static org.elasticsearch.common.xcontent.XContentFactory.jsonBuilder;
import static org.elasticsearch.index.query.QueryBuilders.matchQuery;
@ -146,10 +146,8 @@ public class HttpInputIntegrationTest extends AbstractWatcherIntegrationTests {
.setSize(1)
.get();
assertHitCount(searchResponse, 1);
Map payload = (Map) ((Map)((Map)((Map) searchResponse.getHits().getAt(0).sourceAsMap().get("watch_execution")).get("input_result")).get("http")).get("payload");
assertThat(payload.size(), equalTo(1));
assertThat(((Map) payload.get("hits")).size(), equalTo(1));
assertThat((Integer) ((Map) payload.get("hits")).get("total"), equalTo(1));
XContentSource source = new XContentSource(searchResponse.getHits().getAt(0).getSourceRef());
assertThat(source.getValue("execution_result.input.http.payload.hits.total"), equalTo((Object) 1));
}
}

27
src/test/java/org/elasticsearch/watcher/test/integration/HttpSecretsIntegrationTests.java
View File

@ -32,6 +32,7 @@ import org.junit.Before;
import org.junit.Test;
import java.net.BindException;
import java.util.List;
import java.util.Map;
import static org.elasticsearch.common.joda.time.DateTimeZone.UTC;
@ -152,7 +153,7 @@ public class HttpSecretsIntegrationTests extends AbstractWatcherIntegrationTests
.get();
assertThat(executeResponse, notNullValue());
contentSource = executeResponse.getSource();
value = contentSource.getValue("watch_execution.input_result.http.http_status");
value = contentSource.getValue("execution_result.input.http.http_status");
assertThat(value, notNullValue());
assertThat(value, is((Object) 200));
@ -224,13 +225,25 @@ public class HttpSecretsIntegrationTests extends AbstractWatcherIntegrationTests
.get();
assertThat(executeResponse, notNullValue());
contentSource = executeResponse.getSource();
value = contentSource.getValue("watch_execution.actions_results._webhook.webhook.response.status");
value = contentSource.getValue("execution_result.actions.webhook.response.status");
assertThat(value, instanceOf(List.class));
assertThat(value, notNullValue());
assertThat(value, is((Object) 200));
value = contentSource.getValue("watch_execution.actions_results._webhook.webhook.request.auth.username");
assertThat(value, notNullValue()); // the auth username exists
value = contentSource.getValue("watch_execution.actions_results._webhook.webhook.request.auth.password");
assertThat(value, nullValue()); // but the auth password was filtered out
List<Number> values = (List<Number>) value;
assertThat(values, hasSize(1));
assertThat(values, hasItem(200));
value = contentSource.getValue("execution_result.actions.webhook.request.auth.username");
assertThat(value, notNullValue());
assertThat(value, instanceOf(List.class));
values = (List<Number>) value;
assertThat(values, hasSize(1)); // the auth username exists
value = contentSource.getValue("execution_result.actions.webhook.request.auth.password");
assertThat(value, notNullValue());
assertThat(value, instanceOf(List.class));
values = (List<Number>) value;
assertThat(values, hasSize(0)); // but the auth password was filtered out
RecordedRequest request = webServer.takeRequest();
assertThat(request.getHeader("Authorization"), equalTo(ApplicableBasicAuth.headerValue(USERNAME, PASSWORD.toCharArray())));

8
src/test/java/org/elasticsearch/watcher/test/integration/WatchThrottleTests.java
View File

@ -105,7 +105,7 @@ public class WatchThrottleTests extends AbstractWatcherIntegrationTests {
assertThat(parsedWatch.status().ackStatus().state(), is(Watch.Status.AckStatus.State.AWAITS_EXECUTION));
long throttledCount = docCount(HistoryStore.INDEX_PREFIX + "*", null,
matchQuery(WatchRecord.Parser.STATE_FIELD.getPreferredName(), WatchRecord.State.THROTTLED.id()));
matchQuery(WatchRecord.Field.STATE.getPreferredName(), WatchRecord.State.THROTTLED.id()));
assertThat(throttledCount, greaterThan(0L));
}
@ -167,7 +167,7 @@ public class WatchThrottleTests extends AbstractWatcherIntegrationTests {
assertThat(actionsCount, is(2L));
long throttledCount = docCount(HistoryStore.INDEX_PREFIX + "*", null,
matchQuery(WatchRecord.Parser.STATE_FIELD.getPreferredName(), WatchRecord.State.THROTTLED.id()));
matchQuery(WatchRecord.Field.STATE.getPreferredName(), WatchRecord.State.THROTTLED.id()));
assertThat(throttledCount, is(1L));
} else {
@ -191,7 +191,7 @@ public class WatchThrottleTests extends AbstractWatcherIntegrationTests {
assertThat(actionsCount, is(1L));
long throttledCount = docCount(HistoryStore.INDEX_PREFIX + "*", null,
matchQuery(WatchRecord.Parser.STATE_FIELD.getPreferredName(), WatchRecord.State.THROTTLED.id()));
matchQuery(WatchRecord.Field.STATE.getPreferredName(), WatchRecord.State.THROTTLED.id()));
assertThat(throttledCount, greaterThanOrEqualTo(1L));
}
}, 5, TimeUnit.SECONDS);
@ -308,7 +308,7 @@ public class WatchThrottleTests extends AbstractWatcherIntegrationTests {
assertThat(actionsCount, is(2L));
long throttledCount = docCount(HistoryStore.INDEX_PREFIX + "*", null,
matchQuery(WatchRecord.Parser.STATE_FIELD.getPreferredName(), WatchRecord.State.THROTTLED.id()));
matchQuery(WatchRecord.Field.STATE.getPreferredName(), WatchRecord.State.THROTTLED.id()));
assertThat(throttledCount, is(1L));
}
}

19
src/test/java/org/elasticsearch/watcher/test/integration/WebhookHttpsIntegrationTests.java
View File

@ -13,7 +13,6 @@ import org.elasticsearch.ElasticsearchException;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.common.settings.ImmutableSettings;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.xcontent.support.XContentMapValues;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.watcher.actions.ActionBuilders;
import org.elasticsearch.watcher.history.HistoryStore;
@ -24,6 +23,7 @@ import org.elasticsearch.watcher.support.http.HttpRequestTemplate;
import org.elasticsearch.watcher.support.http.Scheme;
import org.elasticsearch.watcher.support.http.auth.basic.BasicAuth;
import org.elasticsearch.watcher.support.template.Template;
import org.elasticsearch.watcher.support.xcontent.XContentSource;
import org.elasticsearch.watcher.test.AbstractWatcherIntegrationTests;
import org.junit.After;
import org.junit.Before;
@ -33,6 +33,7 @@ import java.net.BindException;
import java.net.URISyntaxException;
import java.nio.file.Path;
import java.nio.file.Paths;
import java.util.List;
import static org.elasticsearch.test.hamcrest.ElasticsearchAssertions.assertNoFailures;
import static org.elasticsearch.watcher.client.WatchSourceBuilders.watchBuilder;
@ -40,7 +41,7 @@ import static org.elasticsearch.watcher.condition.ConditionBuilders.alwaysCondit
import static org.elasticsearch.watcher.input.InputBuilders.simpleInput;
import static org.elasticsearch.watcher.trigger.TriggerBuilders.schedule;
import static org.elasticsearch.watcher.trigger.schedule.Schedules.interval;
import static org.hamcrest.Matchers.equalTo;
import static org.hamcrest.Matchers.*;
/**
*/
@ -115,11 +116,19 @@ public class WebhookHttpsIntegrationTests extends AbstractWatcherIntegrationTest
assertThat(recordedRequest.getBody().readUtf8Line(), equalTo("{key=value}"));
SearchResponse response = client().prepareSearch(HistoryStore.INDEX_PREFIX + "*")
.setQuery(QueryBuilders.termQuery(WatchRecord.Parser.STATE_FIELD.getPreferredName(), "executed"))
.setQuery(QueryBuilders.termQuery(WatchRecord.Field.STATE.getPreferredName(), "executed"))
.get();
assertNoFailures(response);
assertThat(XContentMapValues.extractValue("watch_execution.actions_results._id.webhook.response.body", response.getHits().getAt(0).sourceAsMap()).toString(), equalTo("body"));
assertThat(XContentMapValues.extractValue("watch_execution.actions_results._id.webhook.response.status", response.getHits().getAt(0).sourceAsMap()).toString(), equalTo("200"));
XContentSource source = new XContentSource(response.getHits().getAt(0).sourceRef());
List<String> bodies = source.getValue("execution_result.actions.webhook.response.body");
assertThat(bodies, notNullValue());
assertThat(bodies, hasSize(1));
assertThat(bodies, hasItem("body"));
List<Number> statuses = source.getValue("execution_result.actions.webhook.response.status");
assertThat(statuses, notNullValue());
assertThat(statuses, hasSize(1));
assertThat(statuses, hasItem(200));
}
@Test

14
src/test/java/org/elasticsearch/watcher/test/integration/WebhookIntegrationTests.java
View File

@ -26,6 +26,7 @@ import org.junit.Before;
import org.junit.Test;
import java.net.BindException;
import java.util.List;
import static org.elasticsearch.test.hamcrest.ElasticsearchAssertions.assertNoFailures;
import static org.elasticsearch.watcher.client.WatchSourceBuilders.watchBuilder;
@ -33,8 +34,7 @@ import static org.elasticsearch.watcher.condition.ConditionBuilders.alwaysCondit
import static org.elasticsearch.watcher.input.InputBuilders.simpleInput;
import static org.elasticsearch.watcher.trigger.TriggerBuilders.schedule;
import static org.elasticsearch.watcher.trigger.schedule.Schedules.interval;
import static org.hamcrest.Matchers.anyOf;
import static org.hamcrest.Matchers.equalTo;
import static org.hamcrest.Matchers.*;
/**
*/
@ -93,13 +93,17 @@ public class WebhookIntegrationTests extends AbstractWatcherIntegrationTests {
assertThat(recordedRequest.getBody().readUtf8Line(), equalTo("{key=value}"));
SearchResponse response = client().prepareSearch(HistoryStore.INDEX_PREFIX + "*")
.setQuery(QueryBuilders.termQuery(WatchRecord.Parser.STATE_FIELD.getPreferredName(), "executed"))
.setQuery(QueryBuilders.termQuery(WatchRecord.Field.STATE.getPreferredName(), "executed"))
.get();
assertNoFailures(response);
XContentSource source = new XContentSource(response.getHits().getAt(0).getSourceRef());
assertThat(source.getValue("watch_execution.actions_results._id.webhook.response.body"), equalTo((Object) "body"));
assertThat(source.getValue("watch_execution.actions_results._id.webhook.response.status"), equalTo((Object) 200));
List<String> bodies = source.getValue("execution_result.actions.webhook.response.body");
assertThat(bodies, notNullValue());
assertThat(bodies, hasItem("body"));
List<Number> statuses = source.getValue("execution_result.actions.webhook.response.status");
assertThat(statuses, notNullValue());
assertThat(statuses, hasItem(200));
}
@Test

39
src/test/java/org/elasticsearch/watcher/transform/search/SearchTransformTests.java
View File

@ -69,10 +69,23 @@ public class SearchTransformTests extends ElasticsearchIntegrationTest {
@Override
public Settings nodeSettings(int nodeOrdinal) {
//Set path so ScriptService will pick up the test scripts
return settingsBuilder().put(super.nodeSettings(nodeOrdinal))
return settingsBuilder()
.put(super.nodeSettings(nodeOrdinal))
.put("path.conf", this.getResource("config").getPath()).build();
}
@Override
public Settings indexSettings() {
return settingsBuilder()
.put(super.indexSettings())
// we have to test this on an index that has at least 2 shards. Otherwise when searching indices with
// a single shard the QUERY_THEN_FETCH search type will change to QUERY_AND_FETCH during execution.
.put("index.number_of_shards", randomIntBetween(2, 5))
.build();
}
@Test
public void testApply() throws Exception {
@ -128,8 +141,6 @@ public class SearchTransformTests extends ElasticsearchIntegrationTest {
//
// we then do a search for document 3, and compare the response to the payload returned by the transform
index("idx", "type", "1", doc("2015-01-01T00:00:00", "val_1"));
index("idx", "type", "2", doc("2015-01-02T00:00:00", "val_2"));
index("idx", "type", "3", doc("2015-01-03T00:00:00", "val_3"));
@ -323,24 +334,26 @@ public class SearchTransformTests extends ElasticsearchIntegrationTest {
@Test
public void testDifferentSearchType() throws Exception {
SearchSourceBuilder searchSourceBuilder = searchSource().query(
filteredQuery(matchQuery("event_type", "a"), rangeFilter("_timestamp").from("{{ctx.trigger.scheduled_time}}||-30s").to("{{ctx.trigger.triggered_time}}"))
);
SearchType searchType = getRandomSupportedSearchType();
SearchSourceBuilder searchSourceBuilder = searchSource().query(filteredQuery(
matchQuery("event_type", "a"),
rangeFilter("_timestamp")
.from("{{ctx.trigger.scheduled_time}}||-30s")
.to("{{ctx.trigger.triggered_time}}")));
final SearchType searchType = getRandomSupportedSearchType();
SearchRequest request = client()
.prepareSearch()
.prepareSearch("test-search-index")
.setSearchType(searchType)
.setIndices("test-search-index")
.request()
.source(searchSourceBuilder);
SearchTransform.Result result = executeSearchTransform(request);
assertThat((Integer) XContentMapValues.extractValue("hits.total", result.payload().data()), equalTo(0));
assertNotNull(result.executedRequest());
assertEquals(result.executedRequest().searchType(), searchType);
assertArrayEquals(result.executedRequest().indices(), request.indices());
assertEquals(result.executedRequest().indicesOptions(), request.indicesOptions());
assertThat(result.executedRequest(), notNullValue());
assertThat(result.executedRequest().searchType(), is(searchType));
assertThat(result.executedRequest().indices(), arrayContainingInAnyOrder(request.indices()));
assertThat(result.executedRequest().indicesOptions(), equalTo(request.indicesOptions()));
}
@Test