|
| 1 | +/* |
| 2 | + * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one |
| 3 | + * or more contributor license agreements. Licensed under the Elastic License; |
| 4 | + * you may not use this file except in compliance with the Elastic License. |
| 5 | + */ |
| 6 | +package org.elasticsearch.upgrades; |
| 7 | + |
| 8 | +import org.elasticsearch.Version; |
| 9 | +import org.elasticsearch.client.Request; |
| 10 | +import org.elasticsearch.client.Response; |
| 11 | +import org.elasticsearch.cluster.routing.Murmur3HashFunction; |
| 12 | +import org.elasticsearch.common.Strings; |
| 13 | +import org.elasticsearch.common.UUIDs; |
| 14 | +import org.elasticsearch.common.settings.Settings; |
| 15 | +import org.elasticsearch.common.xcontent.XContentType; |
| 16 | +import org.elasticsearch.xpack.core.watcher.condition.AlwaysCondition; |
| 17 | + |
| 18 | +import java.io.IOException; |
| 19 | +import java.util.Map; |
| 20 | +import java.util.concurrent.TimeUnit; |
| 21 | + |
| 22 | +import static org.elasticsearch.xpack.watcher.actions.ActionBuilders.loggingAction; |
| 23 | +import static org.elasticsearch.xpack.watcher.client.WatchSourceBuilders.watchBuilder; |
| 24 | +import static org.elasticsearch.xpack.watcher.input.InputBuilders.simpleInput; |
| 25 | +import static org.elasticsearch.xpack.watcher.trigger.TriggerBuilders.schedule; |
| 26 | +import static org.elasticsearch.xpack.watcher.trigger.schedule.Schedules.interval; |
| 27 | +import static org.hamcrest.Matchers.equalTo; |
| 28 | +import static org.hamcrest.Matchers.greaterThan; |
| 29 | + |
| 30 | +/** |
| 31 | + * This rolling upgrade node tests whether watcher is able to update the watch status after execution in a mixed cluster. |
| 32 | + * |
| 33 | + * Versions before 6.7.0 the watch status was using the version to do optimistic locking, after 6.7.0 sequence number and |
| 34 | + * primary term are used. The problem was that bwc logic was forgotten to be added, so in a mixed versions cluster, when |
| 35 | + * a watch is executed and its watch status is updated then an update request using sequence number / primary term as |
| 36 | + * way to do optimistic locking can be sent to nodes that don't support this. |
| 37 | + * |
| 38 | + * This test tries to simulate a situation where the bug manifests. This requires watches to be run by multiple nodes |
| 39 | + * holding a .watches index shard. |
| 40 | + */ |
| 41 | +public class WatcherUpgradeIT extends AbstractUpgradeTestCase { |
| 42 | + |
| 43 | + public void testWatchesKeepRunning() throws Exception { |
| 44 | + if (UPGRADED_FROM_VERSION.before(Version.V_6_0_0)) { |
| 45 | + logger.info("Skipping test. Upgrading from before 6.0 makes this test too complicated."); |
| 46 | + return; |
| 47 | + } |
| 48 | + |
| 49 | + final int numWatches = 16; |
| 50 | + |
| 51 | + if (CLUSTER_TYPE.equals(ClusterType.OLD)) { |
| 52 | + final String watch = watchBuilder() |
| 53 | + .trigger(schedule(interval("5s"))) |
| 54 | + .input(simpleInput()) |
| 55 | + .condition(AlwaysCondition.INSTANCE) |
| 56 | + .addAction("_action1", loggingAction("{{ctx.watch_id}}")) |
| 57 | + .buildAsBytes(XContentType.JSON) |
| 58 | + .utf8ToString(); |
| 59 | + |
| 60 | + for (int i = 0; i < numWatches; i++) { |
| 61 | + // Using a random id helps to distribute the watches between watcher services on the different nodes with |
| 62 | + // a .watches index shard: |
| 63 | + String id = UUIDs.randomBase64UUID(); |
| 64 | + logger.info("Adding watch [{}/{}]", id, Math.floorMod(Murmur3HashFunction.hash(id), 3)); |
| 65 | + Request putWatchRequest = new Request("PUT", "/_xpack/watcher/watch/" + id); |
| 66 | + putWatchRequest.setJsonEntity(watch); |
| 67 | + assertOK(client().performRequest(putWatchRequest)); |
| 68 | + |
| 69 | + if (i == 0) { |
| 70 | + // Increasing the number of replicas to makes it more likely that an upgraded node sends an |
| 71 | + // update request (in order to update watch status) to a non upgraded node. |
| 72 | + Request updateSettingsRequest = new Request("PUT", "/.watches/_settings"); |
| 73 | + updateSettingsRequest.setJsonEntity(Strings.toString(Settings.builder() |
| 74 | + .put("index.number_of_replicas", 2) |
| 75 | + .put("index.auto_expand_replicas", (String) null) |
| 76 | + .build())); |
| 77 | + assertOK(client().performRequest(updateSettingsRequest)); |
| 78 | + ensureAllWatchesIndexShardsStarted(); |
| 79 | + } |
| 80 | + } |
| 81 | + } else { |
| 82 | + ensureAllWatchesIndexShardsStarted(); |
| 83 | + // Restarting watcher helps to ensure that after a node upgrade each node will be executing watches: |
| 84 | + // (and not that a non upgraded node is in charge of watches that an upgraded node should run) |
| 85 | + assertOK(client().performRequest(new Request("POST", "/_xpack/watcher/_stop"))); |
| 86 | + assertOK(client().performRequest(new Request("POST", "/_xpack/watcher/_start"))); |
| 87 | + |
| 88 | + // Casually checking whether watches are executing: |
| 89 | + for (int i = 0; i < 10; i++) { |
| 90 | + int previous = getWatchHistoryEntriesCount(); |
| 91 | + assertBusy(() -> { |
| 92 | + Integer totalHits = getWatchHistoryEntriesCount(); |
| 93 | + assertThat(totalHits, greaterThan(previous)); |
| 94 | + }, 30, TimeUnit.SECONDS); |
| 95 | + } |
| 96 | + } |
| 97 | + } |
| 98 | + |
| 99 | + private int getWatchHistoryEntriesCount() throws IOException { |
| 100 | + Request refreshRequest = new Request("POST", "/.watcher-history-*/_refresh"); |
| 101 | + assertOK(client().performRequest(refreshRequest)); |
| 102 | + |
| 103 | + Request searchRequest = new Request("GET", "/.watcher-history-*/_search"); |
| 104 | + searchRequest.setJsonEntity("{\"query\": {\"match\": {\"state\": {\"query\": \"executed\"}}}}"); |
| 105 | + |
| 106 | + Response response = client().performRequest(searchRequest); |
| 107 | + assertEquals(200, response.getStatusLine().getStatusCode()); |
| 108 | + Map<String, Object> responseBody = entityAsMap(response); |
| 109 | + return (Integer) ((Map<?, ?>) responseBody.get("hits")).get("total"); |
| 110 | + } |
| 111 | + |
| 112 | + private void ensureAllWatchesIndexShardsStarted() throws Exception { |
| 113 | + assertBusy(() -> { |
| 114 | + Request request = new Request("GET", "/_cluster/health/.watches"); |
| 115 | + Response response = client().performRequest(request); |
| 116 | + assertEquals(200, response.getStatusLine().getStatusCode()); |
| 117 | + Map<String, Object> responseBody = entityAsMap(response); |
| 118 | + int activeShards = (int) responseBody.get("active_shards"); |
| 119 | + assertThat(activeShards, equalTo(3)); |
| 120 | + }, 30, TimeUnit.SECONDS); |
| 121 | + } |
| 122 | + |
| 123 | +} |
0 commit comments