fix(elasticsearch): gate rolling restarts on cluster health only, not on shard relocations

.github/workflows/test_contracts.yml

@@ -0,0 +1,73 @@
+---
+name: Test Contracts
+on:
+  workflow_dispatch:
+  pull_request:
+    types: [labeled, opened, synchronize, reopened]
+    paths:
+      - 'roles/**'
+      - 'tests/integration/**'
+      - 'tests/fakes/**'
+      - '.github/workflows/test_contracts.yml'
+  push:
+    branches:
+      - main
+  merge_group:
+
+permissions:
+  contents: read
+
+concurrency:
+  group: ${{ github.workflow }}-${{ github.ref }}
+  cancel-in-progress: true
+
+jobs:
+  contracts:
+    # Contract tests are fast, self-contained, and never need Incus — run on
+    # GitHub-hosted ubuntu so a self-hosted-runner outage does not block them.
+    runs-on: ubuntu-latest
+    timeout-minutes: 10
+    steps:
+      - name: Check out code
+        uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1  # v7.0.1
+
+      - name: Set up Python
+        uses: actions/setup-python@a26af69be951a213d495a4c3e4e4022e16d87065  # v7.0.0
+        with:
+          python-version: '3.12'
+
+      - name: Install ansible-core
+        run: |
+          python -m pip install --upgrade pip
+          # Match the requirements-test.txt floor so contract tests exercise
+          # the same ansible-core the molecule matrix does.
+          pip install 'ansible-core>=2.18,<2.19'
+
+      - name: Install collection into ansible-galaxy path
+        run: |
+          mkdir -p "$HOME/.ansible/collections/ansible_collections/oddly"
+          ln -s "$PWD" "$HOME/.ansible/collections/ansible_collections/oddly/elasticstack"
+          ansible-galaxy collection install community.general community.crypto ansible.posix
+
+      - name: Run contract tests
+        run: |
+          set -euo pipefail
+          cd tests/integration
+          fail=0
+          for pb in *_contract.yml; do
+            # If a sibling *_inventory.ini exists, use it; otherwise the
+            # test targets localhost directly.
+            inv="${pb%_contract.yml}_inventory.ini"
+            if [ -f "$inv" ]; then
+              args=(-i "$inv")
+            else
+              args=(-i localhost,)
+            fi
+            echo "::group::${pb}"
+            if ! ansible-playbook "${args[@]}" -c local "$pb"; then
+              echo "::error::contract test ${pb} failed"
+              fail=1
+            fi
+            echo "::endgroup::"
+          done
+          exit "$fail"

roles/elasticsearch/tasks/restart_and_verify_elasticsearch_rolling_node.yml

@@ -41,10 +41,7 @@
         force_basic_auth: "{{ elasticsearch_security | bool }}"
         validate_certs: "{{ elasticsearch_validate_api_certs }}"
       register: _elasticsearch_pre_restart_health
-      until:
-        - (_elasticsearch_pre_restart_health.json.status | default('')) in _elasticsearch_health_statuses
-        - (_elasticsearch_pre_restart_health.json.relocating_shards | default(0) | int) == 0
-        - (_elasticsearch_pre_restart_health.json.initializing_shards | default(0) | int) == 0
+      until: (_elasticsearch_pre_restart_health.json.status | default('')) in _elasticsearch_health_statuses
       retries: "{{ elasticsearch_config_restart_health_retries }}"
       delay: "{{ elasticsearch_config_restart_health_delay }}"
       changed_when: false
@@ -187,10 +184,7 @@
         force_basic_auth: "{{ elasticsearch_security | bool }}"
         validate_certs: "{{ elasticsearch_validate_api_certs }}"
       register: _elasticsearch_post_restart_health
-      until:
-        - (_elasticsearch_post_restart_health.json.status | default('')) in _elasticsearch_health_statuses
-        - (_elasticsearch_post_restart_health.json.relocating_shards | default(0) | int) == 0
-        - (_elasticsearch_post_restart_health.json.initializing_shards | default(0) | int) == 0
+      until: (_elasticsearch_post_restart_health.json.status | default('')) in _elasticsearch_health_statuses
       retries: "{{ elasticsearch_config_restart_health_retries }}"
       delay: "{{ elasticsearch_config_restart_health_delay }}"
       changed_when: false
@@ -264,10 +258,7 @@
         force_basic_auth: "{{ elasticsearch_security | bool }}"
         validate_certs: "{{ elasticsearch_validate_api_certs }}"
       register: _elasticsearch_post_restart_health
-      until:
-        - (_elasticsearch_post_restart_health.json.status | default('')) in _elasticsearch_health_statuses
-        - (_elasticsearch_post_restart_health.json.relocating_shards | default(0) | int) == 0
-        - (_elasticsearch_post_restart_health.json.initializing_shards | default(0) | int) == 0
+      until: (_elasticsearch_post_restart_health.json.status | default('')) in _elasticsearch_health_statuses
       retries: "{{ elasticsearch_config_restart_health_retries }}"
       delay: "{{ elasticsearch_config_restart_health_delay }}"
       changed_when: false

tests/fakes/fake_es_rolling_api.py

@@ -8,11 +8,12 @@
 
 
 class State:
-    def __init__(self, nodes, log_path, persistent_settings, fail_nodes_on):
+    def __init__(self, nodes, log_path, persistent_settings, fail_nodes_on, in_flight):
         self.nodes = nodes
         self.log_path = log_path
         self.persistent_settings = persistent_settings
         self.fail_nodes_on = fail_nodes_on
+        self.in_flight = in_flight
         self.lock = threading.Lock()
 
     def log(self, port, method, path, body):
@@ -59,8 +60,8 @@ def do_GET(self):
                 self._send_json(
                     {
                         "status": "green",
-                        "relocating_shards": 0,
-                        "initializing_shards": 0,
+                        "relocating_shards": state.in_flight,
+                        "initializing_shards": state.in_flight,
                     }
                 )
                 return
@@ -125,6 +126,12 @@ def main():
         default="",
         help="Comma-separated ports whose /_cat/nodes endpoint returns 503",
     )
+    parser.add_argument(
+        "--in-flight",
+        type=int,
+        default=0,
+        help="Shard count reported on both relocating_shards and initializing_shards while green",
+    )
     args = parser.parse_args()
 
     ports = [int(port) for port in args.ports.split(",")]
@@ -136,6 +143,7 @@ def main():
         args.log,
         json.loads(args.persistent_settings),
         fail_nodes_on,
+        args.in_flight,
     )
 
     for port in ports:

tests/integration/rolling_restart_contract.yml

@@ -182,6 +182,105 @@
               - "ansible_failed_result.msg is search('elastic user password is unavailable')"
           run_once: true
 
+- name: Restart fake API reporting relocating shards
+  hosts: localhost
+  gather_facts: false
+  tasks:
+    - name: Stop previous fake API
+      ansible.builtin.shell: |
+        set -o pipefail
+        if [ -f /tmp/elasticstack-rolling-restart-api.pid ]; then
+          kill "$(cat /tmp/elasticstack-rolling-restart-api.pid)" 2>/dev/null || true
+        fi
+      args:
+        executable: /bin/bash
+      changed_when: false
+
+    - name: Reset fake API log
+      ansible.builtin.copy:
+        dest: /tmp/elasticstack-rolling-restart-api.jsonl
+        content: ""
+        mode: "0644"
+
+    - name: Start fake API reporting green with relocating shards
+      ansible.builtin.shell: |
+        set -o pipefail
+        nohup python3 {{ playbook_dir }}/../fakes/fake_es_rolling_api.py \
+          --ports 19200,19201 \
+          --nodes es1,es2 \
+          --in-flight 2 \
+          --log /tmp/elasticstack-rolling-restart-api.jsonl \
+          >/tmp/elasticstack-rolling-restart-api.out 2>&1 &
+        echo $!
+      args:
+        executable: /bin/bash
+      register: _fake_es_api
+      changed_when: true
+
+    - name: Store fake API PID
+      ansible.builtin.copy:
+        dest: /tmp/elasticstack-rolling-restart-api.pid
+        content: "{{ _fake_es_api.stdout }}"
+        mode: "0644"
+
+    - name: Wait for fake API
+      ansible.builtin.uri:
+        url: http://127.0.0.1:19200/_cluster/health
+        method: GET
+      register: _fake_api_health
+      until: _fake_api_health.status | default(0) == 200
+      retries: 10
+      delay: 1
+
+- name: Rolling restart contract - relocating shards do not block a green cluster
+  hosts: elasticsearch
+  gather_facts: false
+  vars:
+    _elasticsearch_restart_test_mode: true
+    elasticsearch_security: false
+    elasticstack_no_log: false
+    elasticstack_elasticsearch_group_name: elasticsearch
+    elasticsearch_http_protocol: http
+    elasticsearch_validate_api_certs: false
+    elasticsearch_config_restart_strategy: rolling
+    elasticsearch_config_restart_flush: false
+    elasticsearch_config_restart_wait_status: green
+    elasticsearch_config_restart_health_retries: 1
+    elasticsearch_config_restart_health_delay: 0
+    elasticsearch_config_restart_node_retries: 1
+    elasticsearch_config_restart_node_delay: 0
+  tasks:
+    - name: Mark all hosts for restart
+      ansible.builtin.set_fact:
+        _elasticsearch_restart_requested: true
+
+    - name: Include rolling restart task  # noqa: run-once[task]
+      ansible.builtin.include_tasks: ../../roles/elasticsearch/tasks/restart_and_verify_elasticsearch_rolling.yml
+      run_once: true
+
+- name: Verify relocating shards did not block the restart
+  hosts: localhost
+  gather_facts: false
+  tasks:
+    - name: Read fake API log
+      ansible.builtin.command:
+        cmd: cat /tmp/elasticstack-rolling-restart-api.jsonl
+      register: _fake_api_log
+      changed_when: false
+
+    # es2 is only drained once es1 cleared its own post-restart health gate.
+    - name: Assert es2 was reached while shards were relocating
+      ansible.builtin.assert:
+        that:
+          - '_fake_api_log.stdout is search(''"method": "PUT", "path": "/_cluster/settings", "port": 19200'')'
+          - '_fake_api_log.stdout is search(''"method": "PUT", "path": "/_cluster/settings", "port": 19201'')'
+
+    - name: Reset fake API log
+      ansible.builtin.copy:
+        dest: /tmp/elasticstack-rolling-restart-api.jsonl
+        content: ""
+        mode: "0644"
+
 - name: Restart fake API with simulated rejoin failure on es1
   hosts: localhost
   gather_facts: false
@@ -254,22 +353,26 @@
       ansible.builtin.set_fact:
         _elasticsearch_restart_requested: "{{ inventory_hostname == 'es1' }}"
 
+    # The include is expected to fail on es1 (simulated rejoin failure). The
+    # peer play below then reads the fake API log to prove cleanup was still
+    # driven by es2. We wrap the include in a block/rescue so that:
+    #  - the failure raised inside the included tasks is caught here,
+    #  - both es1 (which actually failed) and es2 (along for the ride but
+    #    never touched) leave the play clean.
+    # ignore_errors doesn't work here: it doesn't propagate into
+    # include_tasks children reliably, and the role explicitly fail:s with
+    # startup diagnostics on rejoin failure.
     - name: Expect rolling restart to fail but cleanup via peer
       block:
         - name: Include rolling restart task  # noqa: run-once[task]
           ansible.builtin.include_tasks: ../../roles/elasticsearch/tasks/restart_and_verify_elasticsearch_rolling.yml
           run_once: true
-
-        - name: Rejoin failure should have raised  # noqa: run-once[task]
-          ansible.builtin.fail:
-            msg: Simulated rejoin failure did not raise.
-          run_once: true
-
       rescue:
-        - name: Absorb expected failure  # noqa: run-once[task]
+        - name: Absorb expected failure
           ansible.builtin.debug:
-            msg: Rolling restart failed as expected; cleanup should have run via es2.
-          run_once: true
+            msg: >-
+              Rolling restart failed as expected on {{ inventory_hostname }};
+              the following play asserts cleanup landed on the surviving peer.
 
 - name: Verify cleanup used surviving peer
   hosts: localhost