diff --git a/.github/workflows/run_tests.yaml b/.github/workflows/run_tests.yaml index b6a8f210d..b9d9b64e1 100644 --- a/.github/workflows/run_tests.yaml +++ b/.github/workflows/run_tests.yaml @@ -16,6 +16,27 @@ on: type: string required: false jobs: + # Go unit tests were previously run by nothing at all, which is how pkg/model's registry + # concurrency test sat broken for two years without anyone noticing. Runs beside the e2e + # job, so it costs no extra wall clock. -race is the point: those tests exist to catch + # locking regressions and only the detector can see them. + go_unit_tests: + name: Go unit tests (race) + runs-on: ubuntu-latest + timeout-minutes: 20 + steps: + - uses: actions/checkout@v4 + + - uses: actions/setup-go@v5 + with: + # go.mod is the single source of the build Go version. + go-version-file: go.mod + + # -vet=off is the project convention: plain `go test` fails to build a few packages + # on pre-existing vet noise unrelated to the tests. + - name: go test -race + run: go test -vet=off -race -count=1 ./pkg/... + run_tests: name: Run Tests runs-on: ubuntu-latest diff --git a/config/config.yaml b/config/config.yaml index c932c84d1..b2acb4344 100644 --- a/config/config.yaml +++ b/config/config.yaml @@ -502,6 +502,16 @@ reconcile: # a.k.a replication lag - calculated as "MAX(absolute_delay) FROM system.replicas" # is within this specified delay (in seconds) delay: 10 + # Optional replicated-host catch-up gate before advancing to the next host. + # Disabled by default to preserve existing reconcile behavior. + catchUp: + enabled: "false" + # Per-host wall-clock budget for the catch-up gate, in seconds. + timeout: 900 + onTimeout: "abort" + health: + pollInterval: 10 + successThreshold: 6 probes: # Whether the operator during host launch procedure should wait for startup probe to succeed. # In case probe is unspecified wait is assumed to be completed successfully. diff --git a/deploy/builder/templates-config/config.yaml b/deploy/builder/templates-config/config.yaml index de55a41d8..2f4b0cb6c 100644 --- a/deploy/builder/templates-config/config.yaml +++ b/deploy/builder/templates-config/config.yaml @@ -496,6 +496,16 @@ reconcile: # a.k.a replication lag - calculated as "MAX(absolute_delay) FROM system.replicas" # is within this specified delay (in seconds) delay: 10 + # Optional replicated-host catch-up gate before advancing to the next host. + # Disabled by default to preserve existing reconcile behavior. + catchUp: + enabled: "false" + # Per-host wall-clock budget for the catch-up gate, in seconds. + timeout: 900 + onTimeout: "abort" + health: + pollInterval: 10 + successThreshold: 6 probes: # Whether the operator during host launch procedure should wait for startup probe to succeed. # In case probe is unspecified wait is assumed to be completed successfully. diff --git a/deploy/builder/templates-install-bundle/clickhouse-operator-install-yaml-template-01-section-crd-02-chopconf.yaml b/deploy/builder/templates-install-bundle/clickhouse-operator-install-yaml-template-01-section-crd-02-chopconf.yaml index e2e49703c..fe0639c22 100644 --- a/deploy/builder/templates-install-bundle/clickhouse-operator-install-yaml-template-01-section-crd-02-chopconf.yaml +++ b/deploy/builder/templates-install-bundle/clickhouse-operator-install-yaml-template-01-section-crd-02-chopconf.yaml @@ -447,6 +447,41 @@ spec: delay: type: integer description: "replication max absolute delay to consider replica is not delayed" + catchUp: + type: object + description: "Replicated-host catch-up gate to run before advancing to the next host of the shard" + properties: + enabled: + <<: *TypeStringBool + description: "Whether to run the replicated-host catch-up gate" + timeout: + type: integer + description: "Per-host wall-clock budget for the catch-up gate, in seconds. Omit to use the default" + minimum: 1 + onTimeout: + type: string + description: | + What to do when the gate does not complete within timeout. + abort (default) — stop the reconcile + proceed — advance to the next host without writing the caught-up marker + enum: + - "" + - "Abort" + - "abort" + - "Proceed" + - "proceed" + health: + type: object + description: "Stable-health window required after the host synced its replicated objects" + properties: + pollInterval: + type: integer + description: "How often to re-check host health, in seconds. Omit to use the default" + minimum: 1 + successThreshold: + type: integer + description: "How many consecutive healthy checks conclude the gate. Omit to use the default" + minimum: 1 probes: type: object description: "What probes the operator should wait during host launch procedure" diff --git a/deploy/helm/clickhouse-operator/crds/CustomResourceDefinition-clickhouseoperatorconfigurations.clickhouse.altinity.com.yaml b/deploy/helm/clickhouse-operator/crds/CustomResourceDefinition-clickhouseoperatorconfigurations.clickhouse.altinity.com.yaml index 6073e41a8..5af0fc296 100644 --- a/deploy/helm/clickhouse-operator/crds/CustomResourceDefinition-clickhouseoperatorconfigurations.clickhouse.altinity.com.yaml +++ b/deploy/helm/clickhouse-operator/crds/CustomResourceDefinition-clickhouseoperatorconfigurations.clickhouse.altinity.com.yaml @@ -447,6 +447,41 @@ spec: delay: type: integer description: "replication max absolute delay to consider replica is not delayed" + catchUp: + type: object + description: "Replicated-host catch-up gate to run before advancing to the next host of the shard" + properties: + enabled: + !!merge <<: *TypeStringBool + description: "Whether to run the replicated-host catch-up gate" + timeout: + type: integer + description: "Per-host wall-clock budget for the catch-up gate, in seconds. Omit to use the default" + minimum: 1 + onTimeout: + type: string + description: | + What to do when the gate does not complete within timeout. + abort (default) — stop the reconcile + proceed — advance to the next host without writing the caught-up marker + enum: + - "" + - "Abort" + - "abort" + - "Proceed" + - "proceed" + health: + type: object + description: "Stable-health window required after the host synced its replicated objects" + properties: + pollInterval: + type: integer + description: "How often to re-check host health, in seconds. Omit to use the default" + minimum: 1 + successThreshold: + type: integer + description: "How many consecutive healthy checks conclude the gate. Omit to use the default" + minimum: 1 probes: type: object description: "What probes the operator should wait during host launch procedure" diff --git a/deploy/helm/clickhouse-operator/values.yaml b/deploy/helm/clickhouse-operator/values.yaml index eca2c5e79..6e3e3b4e1 100644 --- a/deploy/helm/clickhouse-operator/values.yaml +++ b/deploy/helm/clickhouse-operator/values.yaml @@ -804,6 +804,16 @@ configs: # a.k.a replication lag - calculated as "MAX(absolute_delay) FROM system.replicas" # is within this specified delay (in seconds) delay: 10 + # Optional replicated-host catch-up gate before advancing to the next host. + # Disabled by default to preserve existing reconcile behavior. + catchUp: + enabled: "false" + # Per-host wall-clock budget for the catch-up gate, in seconds. + timeout: 900 + onTimeout: "abort" + health: + pollInterval: 10 + successThreshold: 6 probes: # Whether the operator during host launch procedure should wait for startup probe to succeed. # In case probe is unspecified wait is assumed to be completed successfully. diff --git a/deploy/operator/clickhouse-operator-install-ansible.yaml b/deploy/operator/clickhouse-operator-install-ansible.yaml index bcc35d782..dd05a6553 100644 --- a/deploy/operator/clickhouse-operator-install-ansible.yaml +++ b/deploy/operator/clickhouse-operator-install-ansible.yaml @@ -4160,6 +4160,41 @@ spec: delay: type: integer description: "replication max absolute delay to consider replica is not delayed" + catchUp: + type: object + description: "Replicated-host catch-up gate to run before advancing to the next host of the shard" + properties: + enabled: + <<: *TypeStringBool + description: "Whether to run the replicated-host catch-up gate" + timeout: + type: integer + description: "Per-host wall-clock budget for the catch-up gate, in seconds. Omit to use the default" + minimum: 1 + onTimeout: + type: string + description: | + What to do when the gate does not complete within timeout. + abort (default) — stop the reconcile + proceed — advance to the next host without writing the caught-up marker + enum: + - "" + - "Abort" + - "abort" + - "Proceed" + - "proceed" + health: + type: object + description: "Stable-health window required after the host synced its replicated objects" + properties: + pollInterval: + type: integer + description: "How often to re-check host health, in seconds. Omit to use the default" + minimum: 1 + successThreshold: + type: integer + description: "How many consecutive healthy checks conclude the gate. Omit to use the default" + minimum: 1 probes: type: object description: "What probes the operator should wait during host launch procedure" @@ -6193,6 +6228,16 @@ data: # a.k.a replication lag - calculated as "MAX(absolute_delay) FROM system.replicas" # is within this specified delay (in seconds) delay: 10 + # Optional replicated-host catch-up gate before advancing to the next host. + # Disabled by default to preserve existing reconcile behavior. + catchUp: + enabled: "false" + # Per-host wall-clock budget for the catch-up gate, in seconds. + timeout: 900 + onTimeout: "abort" + health: + pollInterval: 10 + successThreshold: 6 probes: # Whether the operator during host launch procedure should wait for startup probe to succeed. # In case probe is unspecified wait is assumed to be completed successfully. diff --git a/deploy/operator/clickhouse-operator-install-bundle-v1beta1.yaml b/deploy/operator/clickhouse-operator-install-bundle-v1beta1.yaml index dd1f3724c..e849dc947 100644 --- a/deploy/operator/clickhouse-operator-install-bundle-v1beta1.yaml +++ b/deploy/operator/clickhouse-operator-install-bundle-v1beta1.yaml @@ -4127,6 +4127,41 @@ spec: delay: type: integer description: "replication max absolute delay to consider replica is not delayed" + catchUp: + type: object + description: "Replicated-host catch-up gate to run before advancing to the next host of the shard" + properties: + enabled: + !!merge <<: *TypeStringBool + description: "Whether to run the replicated-host catch-up gate" + timeout: + type: integer + description: "Per-host wall-clock budget for the catch-up gate, in seconds. Omit to use the default" + minimum: 1 + onTimeout: + type: string + description: | + What to do when the gate does not complete within timeout. + abort (default) — stop the reconcile + proceed — advance to the next host without writing the caught-up marker + enum: + - "" + - "Abort" + - "abort" + - "Proceed" + - "proceed" + health: + type: object + description: "Stable-health window required after the host synced its replicated objects" + properties: + pollInterval: + type: integer + description: "How often to re-check host health, in seconds. Omit to use the default" + minimum: 1 + successThreshold: + type: integer + description: "How many consecutive healthy checks conclude the gate. Omit to use the default" + minimum: 1 probes: type: object description: "What probes the operator should wait during host launch procedure" @@ -6392,6 +6427,16 @@ data: # a.k.a replication lag - calculated as "MAX(absolute_delay) FROM system.replicas" # is within this specified delay (in seconds) delay: 10 + # Optional replicated-host catch-up gate before advancing to the next host. + # Disabled by default to preserve existing reconcile behavior. + catchUp: + enabled: "false" + # Per-host wall-clock budget for the catch-up gate, in seconds. + timeout: 900 + onTimeout: "abort" + health: + pollInterval: 10 + successThreshold: 6 probes: # Whether the operator during host launch procedure should wait for startup probe to succeed. # In case probe is unspecified wait is assumed to be completed successfully. diff --git a/deploy/operator/clickhouse-operator-install-bundle.yaml b/deploy/operator/clickhouse-operator-install-bundle.yaml index be12c80dd..71a1f3578 100644 --- a/deploy/operator/clickhouse-operator-install-bundle.yaml +++ b/deploy/operator/clickhouse-operator-install-bundle.yaml @@ -4153,6 +4153,41 @@ spec: delay: type: integer description: "replication max absolute delay to consider replica is not delayed" + catchUp: + type: object + description: "Replicated-host catch-up gate to run before advancing to the next host of the shard" + properties: + enabled: + <<: *TypeStringBool + description: "Whether to run the replicated-host catch-up gate" + timeout: + type: integer + description: "Per-host wall-clock budget for the catch-up gate, in seconds. Omit to use the default" + minimum: 1 + onTimeout: + type: string + description: | + What to do when the gate does not complete within timeout. + abort (default) — stop the reconcile + proceed — advance to the next host without writing the caught-up marker + enum: + - "" + - "Abort" + - "abort" + - "Proceed" + - "proceed" + health: + type: object + description: "Stable-health window required after the host synced its replicated objects" + properties: + pollInterval: + type: integer + description: "How often to re-check host health, in seconds. Omit to use the default" + minimum: 1 + successThreshold: + type: integer + description: "How many consecutive healthy checks conclude the gate. Omit to use the default" + minimum: 1 probes: type: object description: "What probes the operator should wait during host launch procedure" @@ -6452,6 +6487,16 @@ data: # a.k.a replication lag - calculated as "MAX(absolute_delay) FROM system.replicas" # is within this specified delay (in seconds) delay: 10 + # Optional replicated-host catch-up gate before advancing to the next host. + # Disabled by default to preserve existing reconcile behavior. + catchUp: + enabled: "false" + # Per-host wall-clock budget for the catch-up gate, in seconds. + timeout: 900 + onTimeout: "abort" + health: + pollInterval: 10 + successThreshold: 6 probes: # Whether the operator during host launch procedure should wait for startup probe to succeed. # In case probe is unspecified wait is assumed to be completed successfully. diff --git a/deploy/operator/clickhouse-operator-install-template-v1beta1.yaml b/deploy/operator/clickhouse-operator-install-template-v1beta1.yaml index 6d7b55de3..9dfc1a428 100644 --- a/deploy/operator/clickhouse-operator-install-template-v1beta1.yaml +++ b/deploy/operator/clickhouse-operator-install-template-v1beta1.yaml @@ -4127,6 +4127,41 @@ spec: delay: type: integer description: "replication max absolute delay to consider replica is not delayed" + catchUp: + type: object + description: "Replicated-host catch-up gate to run before advancing to the next host of the shard" + properties: + enabled: + !!merge <<: *TypeStringBool + description: "Whether to run the replicated-host catch-up gate" + timeout: + type: integer + description: "Per-host wall-clock budget for the catch-up gate, in seconds. Omit to use the default" + minimum: 1 + onTimeout: + type: string + description: | + What to do when the gate does not complete within timeout. + abort (default) — stop the reconcile + proceed — advance to the next host without writing the caught-up marker + enum: + - "" + - "Abort" + - "abort" + - "Proceed" + - "proceed" + health: + type: object + description: "Stable-health window required after the host synced its replicated objects" + properties: + pollInterval: + type: integer + description: "How often to re-check host health, in seconds. Omit to use the default" + minimum: 1 + successThreshold: + type: integer + description: "How many consecutive healthy checks conclude the gate. Omit to use the default" + minimum: 1 probes: type: object description: "What probes the operator should wait during host launch procedure" @@ -6139,6 +6174,16 @@ data: # a.k.a replication lag - calculated as "MAX(absolute_delay) FROM system.replicas" # is within this specified delay (in seconds) delay: 10 + # Optional replicated-host catch-up gate before advancing to the next host. + # Disabled by default to preserve existing reconcile behavior. + catchUp: + enabled: "false" + # Per-host wall-clock budget for the catch-up gate, in seconds. + timeout: 900 + onTimeout: "abort" + health: + pollInterval: 10 + successThreshold: 6 probes: # Whether the operator during host launch procedure should wait for startup probe to succeed. # In case probe is unspecified wait is assumed to be completed successfully. diff --git a/deploy/operator/clickhouse-operator-install-template.yaml b/deploy/operator/clickhouse-operator-install-template.yaml index a2240a7d9..96fa5c8ef 100644 --- a/deploy/operator/clickhouse-operator-install-template.yaml +++ b/deploy/operator/clickhouse-operator-install-template.yaml @@ -4153,6 +4153,41 @@ spec: delay: type: integer description: "replication max absolute delay to consider replica is not delayed" + catchUp: + type: object + description: "Replicated-host catch-up gate to run before advancing to the next host of the shard" + properties: + enabled: + <<: *TypeStringBool + description: "Whether to run the replicated-host catch-up gate" + timeout: + type: integer + description: "Per-host wall-clock budget for the catch-up gate, in seconds. Omit to use the default" + minimum: 1 + onTimeout: + type: string + description: | + What to do when the gate does not complete within timeout. + abort (default) — stop the reconcile + proceed — advance to the next host without writing the caught-up marker + enum: + - "" + - "Abort" + - "abort" + - "Proceed" + - "proceed" + health: + type: object + description: "Stable-health window required after the host synced its replicated objects" + properties: + pollInterval: + type: integer + description: "How often to re-check host health, in seconds. Omit to use the default" + minimum: 1 + successThreshold: + type: integer + description: "How many consecutive healthy checks conclude the gate. Omit to use the default" + minimum: 1 probes: type: object description: "What probes the operator should wait during host launch procedure" @@ -6186,6 +6221,16 @@ data: # a.k.a replication lag - calculated as "MAX(absolute_delay) FROM system.replicas" # is within this specified delay (in seconds) delay: 10 + # Optional replicated-host catch-up gate before advancing to the next host. + # Disabled by default to preserve existing reconcile behavior. + catchUp: + enabled: "false" + # Per-host wall-clock budget for the catch-up gate, in seconds. + timeout: 900 + onTimeout: "abort" + health: + pollInterval: 10 + successThreshold: 6 probes: # Whether the operator during host launch procedure should wait for startup probe to succeed. # In case probe is unspecified wait is assumed to be completed successfully. diff --git a/deploy/operator/clickhouse-operator-install-tf.yaml b/deploy/operator/clickhouse-operator-install-tf.yaml index 5c841acea..a4976a208 100644 --- a/deploy/operator/clickhouse-operator-install-tf.yaml +++ b/deploy/operator/clickhouse-operator-install-tf.yaml @@ -4160,6 +4160,41 @@ spec: delay: type: integer description: "replication max absolute delay to consider replica is not delayed" + catchUp: + type: object + description: "Replicated-host catch-up gate to run before advancing to the next host of the shard" + properties: + enabled: + <<: *TypeStringBool + description: "Whether to run the replicated-host catch-up gate" + timeout: + type: integer + description: "Per-host wall-clock budget for the catch-up gate, in seconds. Omit to use the default" + minimum: 1 + onTimeout: + type: string + description: | + What to do when the gate does not complete within timeout. + abort (default) — stop the reconcile + proceed — advance to the next host without writing the caught-up marker + enum: + - "" + - "Abort" + - "abort" + - "Proceed" + - "proceed" + health: + type: object + description: "Stable-health window required after the host synced its replicated objects" + properties: + pollInterval: + type: integer + description: "How often to re-check host health, in seconds. Omit to use the default" + minimum: 1 + successThreshold: + type: integer + description: "How many consecutive healthy checks conclude the gate. Omit to use the default" + minimum: 1 probes: type: object description: "What probes the operator should wait during host launch procedure" @@ -6193,6 +6228,16 @@ data: # a.k.a replication lag - calculated as "MAX(absolute_delay) FROM system.replicas" # is within this specified delay (in seconds) delay: 10 + # Optional replicated-host catch-up gate before advancing to the next host. + # Disabled by default to preserve existing reconcile behavior. + catchUp: + enabled: "false" + # Per-host wall-clock budget for the catch-up gate, in seconds. + timeout: 900 + onTimeout: "abort" + health: + pollInterval: 10 + successThreshold: 6 probes: # Whether the operator during host launch procedure should wait for startup probe to succeed. # In case probe is unspecified wait is assumed to be completed successfully. diff --git a/deploy/operator/parts/crd.yaml b/deploy/operator/parts/crd.yaml index 397cfd9cc..e59383d34 100644 --- a/deploy/operator/parts/crd.yaml +++ b/deploy/operator/parts/crd.yaml @@ -8725,6 +8725,48 @@ spec: delay: type: integer description: "replication max absolute delay to consider replica is not delayed" + catchUp: + type: object + description: "Replicated-host catch-up gate to run before advancing to the next host of the shard" + properties: + enabled: + # StringBool is polymorphic — accepts native YAML + # bool (true/false), integer (0/1), or string from + # the recognized vocabulary. Normalized by + # pkg/apis/common/types StringBool.UnmarshalJSON. + # Structural-schema rules don't natively support + # bool|int|string union, so we use the documented + # escape hatch x-kubernetes-preserve-unknown-fields. + x-kubernetes-preserve-unknown-fields: true + description: "Whether to run the replicated-host catch-up gate" + timeout: + type: integer + description: "Per-host wall-clock budget for the catch-up gate, in seconds. Omit to use the default" + minimum: 1 + onTimeout: + type: string + description: | + What to do when the gate does not complete within timeout. + abort (default) — stop the reconcile + proceed — advance to the next host without writing the caught-up marker + enum: + - "" + - "Abort" + - "abort" + - "Proceed" + - "proceed" + health: + type: object + description: "Stable-health window required after the host synced its replicated objects" + properties: + pollInterval: + type: integer + description: "How often to re-check host health, in seconds. Omit to use the default" + minimum: 1 + successThreshold: + type: integer + description: "How many consecutive healthy checks conclude the gate. Omit to use the default" + minimum: 1 probes: type: object description: "What probes the operator should wait during host launch procedure" diff --git a/docs/chi-examples/99-clickhouseoperatorconfiguration-max.yaml b/docs/chi-examples/99-clickhouseoperatorconfiguration-max.yaml new file mode 100644 index 000000000..e7a0a3e16 --- /dev/null +++ b/docs/chi-examples/99-clickhouseoperatorconfiguration-max.yaml @@ -0,0 +1,858 @@ +# Comprehensive ClickHouseOperatorConfiguration (CHOPCONF) example covering every +# option the chopconf CRD exposes. Counterpart of CHI's 99-clickhouseinstallation-max.yaml. +# +# Designed as documentation, not for direct deployment. Values shown are the +# operator's own defaults, so applying this file wholesale is close to a no-op - +# but see the merge rules below before copying any of it into a real CR. +# +# For a short, deployable starting point use 70-chop-config.yaml instead. +# +# --------------------------------------------------------------------------- +# Two forms of the same settings +# --------------------------------------------------------------------------- +# The operator reads its configuration from two places: +# +# 1. Its own ConfigMap - the file shipped as `config/config.yaml`, mounted at +# /etc/clickhouse-operator/config.yaml. Flat-rooted: sections start at the +# top level, with no apiVersion/kind/spec wrapper. +# 2. A ClickHouseOperatorConfiguration custom resource - this file's shape. +# Everything under `spec:` here matches the ConfigMap's top level. +# +# The CR is merged ON TOP of the ConfigMap, so a CR only needs to carry the +# settings it actually changes. +# +# --------------------------------------------------------------------------- +# Merge rules - read these before copying anything +# --------------------------------------------------------------------------- +# 1. A non-empty value PINS that setting for the lifetime of the CR. It wins +# over the operator's built-in default, including future defaults changed +# by a later release. Delete a key rather than restating the value you +# believe is already the default. +# 2. List-valued settings APPEND to the operator's list, they do not replace +# it. Restating a default list doubles it - e.g. repeating the default +# `networksIP` yields four entries, not two. The single exception is +# `clickhouse.metrics.excludeRegexp`, which replaces. +# 3. The operator reads this CR at startup. `watch.configuration.onChange` +# below governs what happens when it changes afterwards. +# +# A handful of settings exist in the ConfigMap only and cannot be expressed in +# this CR at all; they are called out in place. + +apiVersion: "clickhouse.altinity.com/v1" +kind: "ClickHouseOperatorConfiguration" +metadata: + name: "chop-config-max" +spec: + ################################################ + ## + ## Watch section + ## + ################################################ + watch: + # Namespaces where clickhouse-operator watches for events. + # Concurrently running operators should watch on different namespaces. + # `include` and `exclude` accept literal namespace names or regexp patterns. + # Empty `include` watches the operator's own namespace (or all namespaces when + # the operator runs in `kube-system`); use [".*"] to force watch-all elsewhere. + # Empty `exclude` matches none. `exclude` is applied after `include`. + namespaces: + include: [] + exclude: [] + + # Behavior when ClickHouseOperatorConfiguration changes: none | restart + configuration: + onChange: restart + + ################################################ + ## + ## ClickHouse section + ## + ################################################ + clickhouse: + configuration: + ################################################ + ## + ## Configuration files section + ## + ################################################ + file: + # Each 'path' can be either absolute or relative. + # In case path is absolute - it is used as is + # In case path is relative - it is relative to the folder where the operator's + # configuration file is located. + path: + # Path to the folder where ClickHouse configuration files common for all instances within a CHI are located. + common: chi/config.d + # Path to the folder where ClickHouse configuration files unique for each instance (host) within a CHI are located. + host: chi/conf.d + # Path to the folder where ClickHouse configuration files with users' settings are located. + # Files are common for all instances within a CHI. + user: chi/users.d + ################################################ + ## + ## Configuration users section + ## + ################################################ + user: + # Default settings for user accounts, created by the operator. + # IMPORTANT. These are not access credentials or settings for the 'default' user account, + # it is a template for filling out missing fields for all user accounts to be created by the operator, + # with the following EXCEPTIONS: + # 1. 'default' user account DOES NOT use provided password, but uses all the rest of the fields. + # Password for 'default' user account has to be provided explicitly, if to be used. + # 2. CHOP user account DOES NOT use: + # - profile setting. It uses predefined profile called 'clickhouse_operator' + # - quota setting. It uses empty quota name. + # - networks IP setting. Operator specifies 'networks/ip' user setting to match operators' pod IP only. + # - password setting. Password for CHOP account is used from 'clickhouse.access.*' section + default: + # Default values for ClickHouse user account(s) created by the operator + # 1. user/profile - string + # 2. user/quota - string + # 3. user/networks/ip - multiple strings + # 4. user/password - string + # These values can be overwritten on per-user basis. + profile: "default" + quota: "default" + # APPENDS to the operator's list. Listing the two defaults here would + # produce four entries - specify only additional networks. + networksIP: + - "::1" + - "127.0.0.1" + password: "default" + ################################################ + ## + ## Configuration network section + ## + ################################################ + network: + # Default host_regexp to limit network connectivity from outside + hostRegexpTemplate: "(chi-{chi}-[^.]+\\d+-\\d+|clickhouse\\-{chi})\\.{namespace}\\.svc\\.cluster\\.local$" + + ################################################ + ## + ## Configuration restart policy section + ## Describes what configuration changes require a ClickHouse restart + ## + ################################################ + configurationRestartPolicy: + rules: + # IMPORTANT! + # Special version of "*" - default version - has to satisfy all ClickHouse versions. + # Default version will also be used in case ClickHouse version is unknown. + # ClickHouse version may be unknown due to host being down - for example, because of incorrect "settings" section. + # ClickHouse is not willing to start in case incorrect/unknown settings are provided in config file. + - version: "*" + rules: + # see https://kb.altinity.com/altinity-kb-setup-and-maintenance/altinity-kb-server-config-files/#server-config-configxml-sections-which-dont-require-restart + # to be replaced with "select * from system.server_settings where changeable_without_restart = 'No'" + + - settings/*: "yes" + + # single values + - settings/access_control_path: "no" + - settings/dictionaries_config: "no" + - settings/max_server_memory_*: "no" + - settings/max_*_to_drop: "no" + - settings/max_concurrent_queries: "no" + - settings/models_config: "no" + - settings/user_defined_executable_functions_config: "no" + + # structured XML + - settings/logger/*: "no" + - settings/macros/*: "no" + - settings/remote_servers/*: "no" + - settings/user_directories/*: "no" + + # these settings should not lead to pod restarts + - settings/display_secrets_in_show_and_select: "no" + + - zookeeper/*: "no" + + - files/*.xml: "yes" + - files/config.d/*.xml: "yes" + - files/config.d/*dict*.xml: "no" + - files/config.d/*no_restart*: "no" + + # exceptions in default profile + - profiles/default/background_*_pool_size: "yes" + - profiles/default/max_*_for_server: "yes" + - version: "21.*" + rules: + - settings/logger: "yes" + + ################################################ + ## + ## Access to ClickHouse instances + ## + ################################################ + access: + # Possible values for 'scheme' are: + # 1. http - force http to be used to connect to ClickHouse instances + # 2. https - force https to be used to connect to ClickHouse instances + # 3. Auto - either http or https is selected based on open ports + # Coerced http -> https when security.policy is Enforced. + scheme: "Auto" + # ClickHouse credentials (username, password and port) to be used by the operator + # to connect to ClickHouse instances. These credentials are used for: + # 1. Metrics requests + # 2. Schema maintenance + # User with these credentials can be specified in additional ClickHouse .xml config files, + # located in 'clickhouse.configuration.file.path.user' folder. + # Prefer the `secret` reference below over inline credentials. + username: "" + password: "" + # Location of the k8s Secret with username and password to be used by the operator + # to connect to ClickHouse instances. Can be used instead of the explicit + # username/password above. Secret should have two keys: `username` and `password`. + secret: + # Empty `namespace` means the k8s Secret is looked up in the same namespace + # where the operator's pod is running. + namespace: "" + # Empty `name` means no k8s Secret would be looked for + name: "clickhouse-operator" + # Port where to connect to ClickHouse instances to + port: 8123 + + # `rootCA`: inline PEM CA bundle the operator uses to verify the ClickHouse + # server certificate when connecting over https (scheme: https, or Auto when + # only TLS ports are open). Verification is enforced when TLS hardening is + # opted in - security.clickhouse.tls.verify: Strict, or a non-empty + # minVersion/serverName; otherwise the CA is loaded but verification stays + # relaxed for backward compatibility. + rootCA: "" + # `rootCASecretRef`: alternate source - read the PEM CA from a Kubernetes + # Secret in the operator's own namespace instead of inlining it above. The + # operator resolves it into `rootCA` once at config load (rotate the Secret + + # restart the operator to pick up a new CA). Mutually exclusive with the + # inline `rootCA` above (inline wins). Empty `name` = not used. When `key` is + # empty, the operator tries "ca.crt" then "tls.crt". + rootCASecretRef: + name: "" + key: "" + + # Timeouts used to limit connection and queries from the operator to ClickHouse + # instances. Specified in seconds. + timeouts: + # Timeout to set up a connection from the operator to ClickHouse instances. + connect: 5 + # Timeout to perform an SQL query from the operator to ClickHouse instances. + query: 4 + + ################################################ + ## + ## Addons specify additional configuration sections applied per ClickHouse version + ## + ################################################ + addons: + rules: + - version: "*" + spec: + configuration: + users: + profiles: + quotas: + settings: + files: + - version: ">= 23.3" + spec: + configuration: + ### + ### users.d is global while description depends on CH version which may vary on per-host basis + ### In case of global-ness this may be better to implement via auto-templates + ### + ### As a solution, this may be applied on the whole cluster based on any of its hosts + ### + ### What to do when host is just created? CH version is not known prior to CH started and user config is required before CH started. + ### We do not have any info about the cluster on initial creation + ### + users: + "{clickhouseOperatorUser}/access_management": 1 + "{clickhouseOperatorUser}/named_collection_control": 1 + "{clickhouseOperatorUser}/show_named_collections": 1 + "{clickhouseOperatorUser}/show_named_collections_secrets": 1 + profiles: + quotas: + settings: + files: + - version: ">= 23.5" + spec: + configuration: + users: + profiles: + clickhouse_operator/format_display_secrets_in_show_and_select: 1 + quotas: + settings: + ## + ## this may be added on per-host basis into host's conf.d folder + ## + display_secrets_in_show_and_select: 1 + files: + + ################################################ + ## + ## Metrics collection from ClickHouse instances + ## + ################################################ + metrics: + # Timeouts used to limit connection and queries from the metrics exporter to + # ClickHouse instances. Specified in seconds. + timeouts: + # Timeout used to limit metrics collection request. + # Upon reaching this timeout metrics collection is aborted and no more metrics + # are collected in this cycle. All collected metrics are returned. + collect: 9 + # Regexp to match tables in system database to fetch metrics from. + # Multiple tables can be matched using regexp. Matched tables are merged using merge() table function. + # Default is "^(metrics|custom_metrics)$" which fetches from both system.metrics and system.custom_metrics. + tablesRegexp: "^(metrics|custom_metrics)$" + # List of regexps to match ClickHouse metrics to exclude from collection/export. + # Regexps match internal metric names before Prometheus normalization and prefixing. + # Default is the per-CPU OS metrics filter shown below; set to [] to disable. + # Unlike other lists in this file, this one REPLACES rather than appends. + excludeRegexp: + - "^metric\\.(OS.*CPU[0-9]+|CPUFrequencyMHz_[0-9]+)$" + + ################################################ + ## + ## Keeper section + ## + ## NOTE: this section is not declared in the chopconf CRD schema. It survives + ## in a CR (the schema preserves unknown top-level sections) but gets no + ## validation. The ConfigMap form is the supported way to set it. + ## + ################################################ + keeper: + configuration: + file: + path: + # Path to the folder where Keeper configuration files common for all instances within a CHK are located. + common: chk/keeper_config.d + # Path to the folder where Keeper configuration files unique for each instance (host) within a CHK are located. + host: chk/conf.d + # Path to the folder where Keeper configuration files with users' settings are located. + user: chk/users.d + + ################################################ + ## + ## Security section (operator-wide defaults; a CHI can override per-cluster + ## via `spec.configuration.clusters[].security`) + ## + ## Shape is target-scoped: security.... + ## ClickHouse-client TLS lives under `clickhouse.tls.*`, ZooKeeper / + ## Keeper-client TLS under `zookeeper.tls.*`, Kubernetes-client toggles + ## under `kubernetes.*`. IPC and FIPS are operator-internal - no CHI override. + ## + ## Three orthogonal hardening axes live in this block: + ## 1. security.policy - TLS-hardening master switch. + ## 2. security.fips.enforced - FIPS cryptographic-module gate (Fatals + ## at startup if the binary lacks GOFIPS140). + ## 3. security.images.policy - workload supply-chain gate (rejects + ## CH/Keeper images that lack "fips" in + ## their tag). + ## Each axis is opt-in and independent of the other two. + ## + ## See docs/security_hardening.md for the design and per-knob semantics. + ## + ## NOTE: the CRD models this whole block as a free-form object, so none of the + ## fields below are schema-validated. A typo here is accepted silently. + ## + ################################################ + security: + clickhouse: + # TLS verification for outbound ClickHouse client connections the operator + # makes (schemer, health probes, /metrics scraping helpers). + # + # Each field below is a TRISTATE: explicit value, empty string (""), or + # absent. Empty/absent means "inherit" - for back-compat the operator's + # baseline default is still PERMISSIVE (`InsecureSkipVerify=true`, Go-default + # TLS version), matching pre-0.27.1 behavior. Set explicit values here to + # tighten across every CHI managed by this operator. A CHI may override + # per-cluster via `spec.configuration.clusters[].security.clickhouse.tls`. + tls: + # `verify`: does the client verify the server's certificate chain? + # "Strict" - verify server cert against `rootCA` (or system roots if + # empty). Hostname must match `serverName` (or the dial + # host if `serverName` is empty). MITM-resistant. + # "None" - skip verification entirely. Connection is encrypted but + # an attacker can intercept transparently. Useful only for + # development/self-signed clusters. + # "" - preserve legacy behavior: equivalent to "None" today, but + # re-evaluated by the future master switch. Will become + # "Strict" once the FIPS profile is enforced. + verify: "" + # `minVersion`: minimum negotiated TLS version. Empty uses Go stdlib + # default (currently TLS 1.2). FIPS Strict coerces to "1.3". + # "1.2" | "1.3" | "" + minVersion: "" + # `serverName`: SNI name + name that the server cert must match when + # `verify=Strict`. Empty derives it from the dial host (typically the + # pod's headless-service FQDN). Set this when the cert is issued to a + # different name than the dial address (e.g. a wildcard or service-CN). + serverName: "" + # `rootCA`: PEM-encoded CA bundle used to validate the server cert when + # `verify=Strict`. Accepts raw PEM or base64-wrapped PEM. Empty means + # use the system CA roots from the operator pod's trust store. + rootCA: "" + # `rootCASecretRef`: alternate source - read the PEM CA from a Kubernetes + # Secret. The operator resolves it at CHI normalize time and inlines the + # value into `rootCA`. Mutually exclusive with the inline `rootCA` above + # - setting both aborts the CR with reason RootCAConflict. + # + # `key` defaulting: when omitted, the operator tries "ca.crt" first + # (cert-manager / kubernetes.io/tls convention), then "tls.crt" as a + # fallback. Override `key:` for hand-rolled Secrets with custom layouts. + # + # Namespace: SecretKeySelector has no namespace field. The Secret is + # expected in the CHI's namespace (for cluster-level refs) or the + # operator's namespace (for CHOP-config-level refs). + # + # Missing Secret/key aborts the CR with reason RootCASecretUnresolved. + # There is no silent fallback to system roots, because empty CA + + # Verify=Strict would refuse every dial. + rootCASecretRef: + name: "" + key: "" + zookeeper: + # TLS knobs for the operator's ZooKeeper / Keeper client. Existing ZK TLS + # already loads cert/key/CA + ServerName separately; these knobs add + # MinVersion + InsecureSkipVerify control on top of that path. + tls: + # Same tristate semantics as `clickhouse.tls.verify`, but the ZK baseline + # is more conservative: when the existing ZK TLS path is active (cert+key+CA + # are wired up), "Strict" is the effective default. Set "None" here to + # opt out of cert verification for a development ZK ensemble. + verify: "" + # Same semantics as `clickhouse.tls.minVersion`. Empty = Go default (1.2). + minVersion: "" + kubernetes: + tls: + # `verify`: TLS posture applied as a LOAD-TIME GATE against the kubeconfig. + # Unlike clickhouse/zookeeper, the operator does NOT build the kubeconfig + # tls.Config - client-go reads TLSClientConfig.Insecure from disk. This + # knob only refuses or permits startup based on what the kubeconfig says. + # "Strict" - refuse startup if the kubeconfig has Insecure=true. + # "None" - explicit opt-in: permit an insecure kubeconfig. + # "" - preserve current behavior (kubeconfig wins). + verify: "" + # `minVersion`: floor TLS at this protocol version. Declared here for + # shape uniformity and FIPS coercion symmetry, but NOT yet enforced on + # the operator's K8s API transport. + # "1.2" | "1.3" | "" + minVersion: "" + # Operator<->metrics-exporter REST channel (`/chi` on port 8888) hardening. + # Plain (default) preserves today's behavior: server binds all interfaces, + # no auth. Secure rejects non-loopback callers at the /chi handler AND + # requires an X-CHOP-Token bearer-token header on every request. The + # operator provisions the token at startup (32 bytes from crypto/rand, + # hex-encoded) into a shared Pod-local emptyDir volume mounted into + # both containers. + ipc: + # `mode`: IPC channel posture. + # "Plain" - default. Server binds all interfaces, no auth required. + # "Secure" - bind loopback only, require X-CHOP-Token on every call. + mode: "Plain" + # `bindHost`: address the IPC server binds to in Secure mode. Empty + # defaults to "127.0.0.1". Ignored in Plain mode (server still binds + # all interfaces). + bindHost: "" + # `tokenPath`: filesystem path to the shared Pod-local token file used + # by both containers in Secure mode. Empty defaults to + # "/etc/clickhouse-operator-ipc/token". An advanced GitOps/Vault use + # case - sourcing the token from a Kubernetes Secret - is supported + # via a Deployment volume override (no CRD field). See + # docs/security_hardening.md -> Externally-managed token (advanced). + tokenPath: "" + # Axis 1 - Operator-wide TLS-hardening master switch. Permissive (default) + # preserves 0.27.0 behavior - no coercion, no rejection. Enforced coerces + # all transport knobs above to their Strict positions at startup + # (clickhouse.tls.verify=Strict, clickhouse.tls.minVersion=1.3, + # zookeeper.tls.verify=Strict, zookeeper.tls.minVersion=1.3, + # kubernetes.tls.verify=Strict, kubernetes.tls.minVersion=1.3, + # ipc.mode=Secure), re-registers the ClickHouse legacy TLS config to + # verifying mode (no InsecureSkipVerify), coerces clickhouse.access.scheme + # http->https, rejects ZK `digest:` auth files, and rejects CHIs that + # cannot be served in a FIPS-compatible posture (e.g. CHIs referencing + # plain-text external ZooKeeper). Transport hardening only - does NOT + # assert the binary is FIPS-linked; that is the orthogonal `fips.enforced` + # axis below. + policy: Permissive + # Axis 2 - FIPS cryptographic-module gate. Orthogonal to `policy`. When + # `enforced: true`, the operator Fatals at startup unless the binary was + # built with GOFIPS140 and crypto/fips140 reports Enabled (i.e. the + # process is running with GODEBUG=fips140=on or fips140=only). Also + # triggers the same TLS coercions as `policy: Enforced` - a FIPS-asserted + # operator necessarily wants verified TLS. + # + # The default `altinity/clickhouse-operator` and `altinity/metrics-exporter` + # images are FIPS 140-3 compatible (not certified): built with + # GOFIPS140=v1.0.0 and run with GODEBUG=fips140=on. Setting + # `enforced: true` asserts that posture at startup. + fips: + enforced: false + images: + # Axis 3 - Workload supply-chain gate, orthogonal to `policy` and + # `fips.enforced`. Permissive (default) accepts any image; FIPSRequired + # refuses CRs whose CH/Keeper images lack the "fips" tag substring + # (admission) AND aborts running CRs whose `SELECT version()` lacks + # "fips" (post-Ready confirmation). See docs/security_hardening_fips.md + # -> "security.images.policy: FIPSRequired" for the full policy details + # and recovery procedure. + policy: Permissive + + ################################################ + ## + ## Template(s) management section + ## + ################################################ + template: + chi: + # CHI template updates handling policy + # Possible policy values: + # - ReadOnStart. Accept CHIT updates on the operator's start only. + # - ApplyOnNextReconcile. Accept CHIT updates at all time. Apply new CHITs on next regular reconcile of the CHI + policy: ApplyOnNextReconcile + + # Path to the folder where ClickHouseInstallation templates .yaml manifests are located. + # Templates are added to the list of all templates and used when CHI is reconciled. + # Templates are applied in sorted alpha-numeric order. + path: chi/templates.d + + ################################################ + ## + ## Reconcile section + ## + ################################################ + reconcile: + # Reconcile runtime settings + runtime: + # Max number of concurrent CHI reconciles in progress + reconcileCHIsThreadsNumber: 10 + # Max number of concurrent CHK reconciles in progress + reconcileCHKsThreadsNumber: 1 + + # The operator reconciles shards concurrently in each CHI with the following limitations: + # 1. Number of shards being reconciled (and thus having hosts down) in each CHI concurrently + # can not be greater than 'reconcileShardsThreadsNumber'. + # 2. Percentage of shards being reconciled (and thus having hosts down) in each CHI concurrently + # can not be greater than 'reconcileShardsMaxConcurrencyPercent'. + # 3. The first shard is always reconciled alone. Concurrency starts from the second shard and onward. + # Thus limiting number of shards being reconciled (and thus having hosts down) in each CHI by both number and percentage + + # Max number of concurrent shard reconciles within one cluster in progress + reconcileShardsThreadsNumber: 5 + # Max percentage of concurrent shard reconciles within one cluster in progress + reconcileShardsMaxConcurrencyPercent: 50 + + # Reconcile StatefulSet scenario + statefulSet: + # Create StatefulSet scenario + create: + # What to do in case created StatefulSet is not in 'Ready' after `reconcile.statefulSet.update.timeout` seconds + # Possible options: + # 1. abort - abort the process, do nothing with the problematic StatefulSet, leave it as it is, + # do not try to fix or delete or update it, just abort reconcile cycle. + # Do not proceed to the next StatefulSet(s) and wait for an admin to assist. + # 2. delete - delete newly created problematic StatefulSet and follow 'abort' path afterwards. + # 3. ignore - ignore an error, pretend nothing happened, continue reconcile and move on to the next StatefulSet. + onFailure: ignore + + # Update StatefulSet scenario + update: + # How many seconds to wait for created/updated StatefulSet to be 'Ready' + timeout: 300 + # How many seconds to wait between checks/polls for created/updated StatefulSet status + pollInterval: 5 + # What to do in case updated StatefulSet is not in 'Ready' after `reconcile.statefulSet.update.timeout` seconds + # Possible options: + # 1. abort - abort the process, do nothing with the problematic StatefulSet, leave it as it is, + # do not try to fix or delete or update it, just abort reconcile cycle. + # Do not proceed to the next StatefulSet(s) and wait for an admin to assist. + # 2. rollback - delete Pod and rollback StatefulSet to previous Generation. + # Pod would be recreated by StatefulSet based on rollback-ed StatefulSet configuration. + # Follow 'abort' path afterwards. + # 3. ignore - ignore an error, pretend nothing happened, continue reconcile and move on to the next StatefulSet. + onFailure: abort + + # Recreate StatefulSet scenario + recreate: + # What to do in case the operator is in need to recreate a StatefulSet? + # Possible options: + # 1. abort - abort the process, do nothing with the problematic StatefulSet, leave it as it is, + # do not try to fix or delete or update it, just abort reconcile cycle. + # Do not proceed to the next StatefulSet(s) and wait for an admin to assist. + # 2. recreate - proceed and recreate StatefulSet. + + # Triggered when PVC data loss or missing volumes are detected. + # `abort` is the setting that stops the operator from recreating a host + # whose volume was lost - the safe choice where data loss must be + # investigated by an admin rather than healed automatically. + onDataLoss: recreate + # Triggered when StatefulSet update fails or StatefulSet is not ready + onUpdateFailure: recreate + + # Reconcile Host scenario + host: + # The operator during reconcile procedure should wait for a ClickHouse host to achieve the following conditions: + wait: + # Whether the operator during reconcile procedure should wait for a ClickHouse host: + # - to be excluded from a ClickHouse cluster + # - to complete all running queries + # - to be included into a ClickHouse cluster + # respectfully before moving forward with host reconcile + exclude: "true" + queries: "true" + include: "false" + # The operator during reconcile procedure should wait for replicas to catch-up + # replication delay a.k.a replication lag for the following replicas + replicas: + # All replicas (new and known earlier) are explicitly requested to wait for replication to catch-up + all: "no" + # New replicas only are requested to wait for replication to catch-up + new: "yes" + # Replication catch-up is considered to be completed as soon as replication delay + # a.k.a replication lag - calculated as "MAX(absolute_delay) FROM system.replicas" + # is within this specified delay (in seconds) + delay: 10 + # Opt-in gate that holds the rolling reconcile until a recreated replica has + # actually rebuilt from its peers, instead of trusting the local + # "MAX(absolute_delay)" probe above. Aimed at local/direct-attached storage + # recovery, where a pod can report a healthy delay before asynchronous + # loading has exposed every replicated object. + catchUp: + # Whether the gate is active. Default is off - the legacy delay probe above is used. + enabled: "no" + # Per-host wall-clock budget for the whole gate, in seconds. Must be >= 1. + # Omit to use the default. + timeout: 900 + # What to do when the gate does not complete within `timeout`: + # - abort stop the reconcile (default) + # - proceed advance to the next host without writing the caught-up marker, + # so a later reconcile retries the catch-up + # Accepted in either case, like the other enum-valued options. + onTimeout: "abort" + # Stable-health window required after the host has synced its replicated objects. + # Health is read from system.replicas: is_readonly = 0, is_session_expired = 0 and + # absolute_delay within `delay` above. + health: + # Seconds between health checks. Must be >= 1. Omit to use the default. + pollInterval: 10 + # Consecutive healthy checks required before the caught-up marker is written. + # Must be >= 1. Omit to use the default. + successThreshold: 6 + probes: + # Whether the operator during host launch procedure should wait for startup probe to succeed. + # In case probe is unspecified wait is assumed to be completed successfully. + # Default option value is to do not wait. + startup: "no" + # Whether the operator during host launch procedure should wait for readiness probe to succeed. + # In case probe is unspecified wait is assumed to be completed successfully. + # Default option value is to wait. + readiness: "yes" + + # The operator during reconcile procedure should drop the following entities: + drop: + replicas: + # Whether the operator during reconcile procedure should drop replicas when replica is deleted + onDelete: "yes" + # Whether the operator during reconcile procedure should drop replicas when replica volume is lost + onLostVolume: "yes" + # Whether the operator during reconcile procedure should drop active replicas when replica is deleted or recreated + active: "no" + + # Operator-wide default reconcile hooks, inherited by every CHI this operator + # manages. A CHI can declare its own under `spec.reconcile.host.hooks`; see + # docs/chi-examples/23-reconcile-hooks-*.yaml for the per-CHI form. + # + # KNOWN LIMITATION: the chopconf CRD does not declare the `events` and + # `failurePolicy` fields that the CHI CRD declares, so the API server strips + # them from a ClickHouseOperatorConfiguration CR. A hook with no `events` never + # matches and never fires, and a stripped `failurePolicy` falls back to `Fail`. + # Until the schema is fixed, set operator-wide hooks through the operator's + # ConfigMap (config.yaml), which is not schema-filtered. The block below is + # therefore shown commented out. + # + # hooks: + # pre: + # - events: + # - HostReconcileStarted + # target: host + # failurePolicy: Fail + # sql: + # queries: + # - "SYSTEM STOP DISTRIBUTED SENDS" + # post: + # - events: + # - HostReconcileCompleted + # target: host + # failurePolicy: Ignore + # shell: + # container: clickhouse + # command: + # - "/bin/sh" + # - "-c" + # - "echo reconciled" + + ################################################ + ## + ## Coordination with external systems during reconcile + ## + ################################################ + coordination: + keeper: + # How long the operator waits for a referenced ClickHouseKeeper to become ready + # before aborting CHI reconcile. In seconds. + readyTimeout: 120 + # Reaction when a referenced CHK resource changes: + # none (default) - do nothing + # reconcile - trigger CHI reconcile + onKeeperResourceUpdate: none + + ################################################ + ## + ## Auto-recovery from aborted/completed reconcile + ## + ################################################ + recovery: + # Recovery scopes keyed by the CHI .status.status they apply to. + # Each scope contains on: mappings that apply while the CHI + # is in that status. + onStatus: + # Recovery for a CHI whose .status.status is Aborted (reconcile did not + # complete) when one of its host pods transitions to Ready. + aborted: + # Action when a pod belonging to an Aborted CHI transitions to Ready: + # retry (default) - re-enqueue the CHI for reconcile + # none - do nothing, CHI stays Aborted; an operator user must + # edit the CR spec to retrigger normalize + onPodReady: retry + # Recovery for a CHI whose .status.status is Completed (fully reconciled) + # when one of its host pods regresses to Ready=False and stays NotReady. + completed: + # Action when a Completed CHI's pod flips Ready=True -> Ready=False and + # stays NotReady for at least onPodNotReadyThreshold: + # none (default) - do nothing + # retry - re-enqueue the CHI so the stuck host is force-restarted + # OFF by default: force-recreating a Completed CHI's pod is destructive - it + # can interrupt a replica's in-progress recovery and means hard downtime for + # a single-replica shard. Opt in with `retry` only where that is acceptable. + onPodNotReady: none + # Minimum duration a pod must stay Ready=False before recovery fires, once + # enabled (Go duration string; default 5m). Raise it for slow-recovering replicas. + onPodNotReadyThreshold: 5m + + ################################################ + ## + ## Annotations management section + ## + ################################################ + annotation: + # Applied when: + # 1. Propagating annotations from the CHI's `metadata.annotations` to child objects' `metadata.annotations`, + # 2. Propagating annotations from the CHI Template's `metadata.annotations` to CHI's `metadata.annotations`, + # Include annotations from the following list: + # Applied only when not empty. Empty list means "include all, no selection" + include: [] + # Exclude annotations from the following list: + exclude: [] + + ################################################ + ## + ## Labels management section + ## + ################################################ + label: + # Applied when: + # 1. Propagating labels from the CHI's `metadata.labels` to child objects' `metadata.labels`, + # 2. Propagating labels from the CHI Template's `metadata.labels` to CHI's `metadata.labels`, + # Include labels from the following list: + # Applied only when not empty. Empty list means "include all, no selection" + include: [] + # Exclude labels from the following list: + # Applied only when not empty. Empty list means "nothing to exclude, no selection" + exclude: [] + # Whether to append *Scope* labels to StatefulSet and Pod. + # Full list of available *scope* labels check in 'labeler.go' + # LabelShardScopeIndex + # LabelReplicaScopeIndex + # LabelCHIScopeIndex + # LabelCHIScopeCycleSize + # LabelCHIScopeCycleIndex + # LabelCHIScopeCycleOffset + # LabelClusterScopeIndex + # LabelClusterScopeCycleSize + # LabelClusterScopeCycleIndex + # LabelClusterScopeCycleOffset + appendScope: "no" + + ################################################ + ## + ## Metrics management section + ## + ## Note: this section governs labels the operator ATTACHES to the metrics it + ## exports. Settings for reading metrics OUT of ClickHouse live under + ## `clickhouse.metrics` above - the two are deliberately distinct. + ## + ################################################ + metrics: + labels: + # Labels to omit from exported operator metrics. Empty list exports all of them. + # Use it to drop high-cardinality labels that inflate the metrics store. + exclude: [] + + ################################################ + ## + ## Status management section + ## + ## Which optional fields the operator maintains in a CR's `.status`. Each one + ## costs an additional status write per reconcile step, so the more verbose + ## history fields are off by default. + ## + ################################################ + status: + fields: + # Last action performed on the CR + action: "false" + # Rolling history of recent actions + actions: "false" + # Last error encountered + error: "true" + # Rolling history of recent errors + errors: "true" + + ################################################ + ## + ## StatefulSet management section + ## + ################################################ + statefulSet: + # How many old ControllerRevisions to retain for each StatefulSet the operator + # creates. 0 keeps none. + revisionHistoryLimit: 0 + + ################################################ + ## + ## Pod management section + ## + ################################################ + pod: + # Grace period for Pod termination. + # How many seconds to wait between sending + # SIGTERM and SIGKILL during Pod termination process. + # Increase this number in case of slow shutdown. + terminationGracePeriod: 30 + + ################################################ + ## + ## Log parameters section + ## + ## These mirror the glog flags the operator binary accepts. + ## + ################################################ + logger: + logtostderr: "true" + alsologtostderr: "false" + v: "1" + stderrthreshold: "" + vmodule: "" + log_backtrace_at: "" diff --git a/docs/operator_configuration.md b/docs/operator_configuration.md index 7e4d58af1..4196b7a5e 100644 --- a/docs/operator_configuration.md +++ b/docs/operator_configuration.md @@ -222,6 +222,96 @@ spec: See [Keeper Reference](keeper_reference.md) for details on how CHI references CHK resources. +### Replicated Host Catch-Up Gate + +The operator can optionally block a rolling host reconcile until a recreated replicated +ClickHouse host catches up to a bounded replication baseline. This is an operator +rolling gate, not a readiness probe. It is disabled by default. + +This is especially useful for local or direct-attached storage deployments, including +NVMe-backed Local PVs, where a recreated pod may start with an empty or replaced disk +and must rebuild replicated data from peer replicas before the operator rolls the next +host. + +Three changes to the surrounding catch-up behaviour are **not** gated on `catchUp.enabled`, so they +apply even with this gate off: + +1. A host that lost its storage volume is forced to catch up before it is returned to service: + its `status.hostsWithReplicaCaughtUp` entry is invalidated and the wait runs. This overrides + `reconcile.host.wait.replicas.all` and `.new` — a host whose disk is gone waits even when both + are `no`, because a marker describing a disk that no longer exists is not evidence of anything. + Note the wait itself is not time-capped, so a replica that cannot converge will stall that + CHI's reconcile; it is visible as `InProgress` with periodic replication-lag log lines, and + editing the CHI cancels the stalled pass. +2. A reconcile cancelled while a host is still catching up no longer records the marker — a + cancelled wait is not evidence that the replica caught up. +3. The catch-up wait now runs before the host is restored to normal priority in `remote_servers`, + rather than after, so a host that was excluded stays deprioritized for the duration of the wait + instead of receiving distributed queries while still behind. + +The marker path only polls the local host's `MAX(absolute_delay)` from `system.replicas` before +writing `status.hostsWithReplicaCaughtUp`, which is weak for recreated-host recovery +because the metric is limited to replicated objects already loaded and visible on that +local server. During recreated-host recovery, asynchronous database/table loading may +not have exposed all replicated objects on the local host yet, and a local delay metric +cannot discover replicated objects that exist on peers or issue a ClickHouse sync +barrier for their known parts. The catch-up gate adds those checks before the operator +advances to the next host. + +```yaml +spec: + reconcile: + host: + wait: + replicas: + catchUp: + enabled: "false" + timeout: 900 + onTimeout: "abort" + health: + pollInterval: 10 + successThreshold: 6 +``` + +| Setting | Default | Description | +|---|---|---| +| `enabled` | `"false"` | Enables the replicated-host catch-up gate. Existing replica-delay behavior is unchanged when disabled. | +| `timeout` | `900` | Per-host gate budget in seconds; omit it to take the default. The CRD requires `>= 1`, and the config-file path falls back to the default for anything `<= 0`, so the gate is never unbounded - otherwise `onTimeout` could never fire. | +| `onTimeout` | `"abort"` | `abort` stops reconcile on the gate deadline. `proceed` advances without writing the caught-up marker, so a later reconcile can try again. Accepted in either case, like the other enum-valued options. | +| `health.pollInterval` | `10` | Seconds between post-sync health checks; omit it to take the default. CRD requires `>= 1`. | +| `health.successThreshold` | `6` | Consecutive healthy checks required after sync before the caught-up marker is written; omit it to take the default. CRD requires `>= 1`. | + +When enabled, the gate waits for asynchronous database loading when ClickHouse exposes +`system.asynchronous_loader`, discovers replicated objects from the peer replicas of the same shard, syncs +`Replicated` databases with `SYSTEM SYNC DATABASE REPLICA`, syncs replicated tables +with `SYSTEM SYNC REPLICA ... LIGHTWEIGHT` (full `SYSTEM SYNC REPLICA` when the ClickHouse version is older than 23.4 or cannot be determined), and then requires a stable health window. +Health is based on `system.replicas`: `is_readonly = 0`, `is_session_expired = 0`, and +`absolute_delay <= reconcile.host.wait.replicas.delay`. + +The `LIGHTWEIGHT` baseline is the time when the sync command runs. It waits for the +relevant part-acquisition work known at that point; it does not require +`system.replication_queue` to become empty and does not block forever on unrelated +merges, mutations, or new ingest that arrives after the sync command. + +Hard failures always abort regardless of `onTimeout`: query or connection failure, +parent reconcile context cancellation, failed/canceled async load jobs, readonly +replicas, and expired Keeper sessions. The caught-up marker is written only after real +success or when peer discovery confirms that there are no replicated objects to sync. + +Manual local-PV/data-loss validation: + +1. Create a CHI with a replicated shard and `catchUp.enabled: "true"`. +2. Wait for the current hosts to become caught up and confirm + `status.hostsWithReplicaCaughtUp` contains the host FQDNs. +3. Simulate storage loss for one host, for example by removing the local PV/PVC data + in a test environment. +4. Reconcile the CHI and confirm the operator removes the stale caught-up marker for + the recreated host. +5. Confirm the recreated host runs the catch-up gate and the next host in the shard does + not advance while the recreated host is still behind. +6. Allow replication to catch up and confirm the recreated host receives the + caught-up marker again, then the next host proceeds. + ## Security The `security:` block at the chopconf top level (sibling of `clickhouse:`) holds operator-wide hardening defaults across three orthogonal axes: transport hardening (`security.policy`), FIPS cryptographic-module enforcement (`security.fips.enforced`), and workload supply-chain gating (`security.images.policy`). Per-component sub-blocks under it cover ClickHouse-client TLS, ZooKeeper-client TLS, Kubernetes-client TLS, and the operator↔metrics-exporter IPC channel. diff --git a/pkg/apis/clickhouse-keeper.altinity.com/v1/type_status.go b/pkg/apis/clickhouse-keeper.altinity.com/v1/type_status.go index 937128d88..767383b40 100644 --- a/pkg/apis/clickhouse-keeper.altinity.com/v1/type_status.go +++ b/pkg/apis/clickhouse-keeper.altinity.com/v1/type_status.go @@ -184,6 +184,20 @@ func (s *Status) PushHostReplicaCaughtUp(host string) { }) } +// RemoveHostReplicaCaughtUp removes host from the list of hosts with replica caught-up +func (s *Status) RemoveHostReplicaCaughtUp(host string) { + host = util.NormalizeFQDN(host) + doWithWriteLock(s, func(s *Status) { + hosts := s.HostsWithReplicaCaughtUp[:0] + for _, caughtUpHost := range s.HostsWithReplicaCaughtUp { + if caughtUpHost != host { + hosts = append(hosts, caughtUpHost) + } + } + s.HostsWithReplicaCaughtUp = hosts + }) +} + // PushHostTablesCreated pushes host to the list of hosts with created tables func (s *Status) PushHostTablesCreated(host string) { host = util.NormalizeFQDN(host) diff --git a/pkg/apis/clickhouse.altinity.com/v1/interface.go b/pkg/apis/clickhouse.altinity.com/v1/interface.go index ddd52b5e7..35fdc32a0 100644 --- a/pkg/apis/clickhouse.altinity.com/v1/interface.go +++ b/pkg/apis/clickhouse.altinity.com/v1/interface.go @@ -101,6 +101,7 @@ type IStatus interface { GetHostsWithReplicaCaughtUp() []string PushHostTablesCreated(host string) PushHostReplicaCaughtUp(host string) + RemoveHostReplicaCaughtUp(host string) HasNormalizedCRCompleted() bool diff --git a/pkg/apis/clickhouse.altinity.com/v1/type_configuration_chop.go b/pkg/apis/clickhouse.altinity.com/v1/type_configuration_chop.go index a11de3b1b..aa28d10e7 100644 --- a/pkg/apis/clickhouse.altinity.com/v1/type_configuration_chop.go +++ b/pkg/apis/clickhouse.altinity.com/v1/type_configuration_chop.go @@ -66,6 +66,13 @@ const ( // RecoveryActionRetry means re-enqueue CHI for reconcile (default). RecoveryActionRetry = "Retry" + // CatchUpOnTimeoutAbort means stop the reconcile when the replicated host sync gate + // reaches its deadline (default). + CatchUpOnTimeoutAbort = "Abort" + // CatchUpOnTimeoutProceed means advance to the next host on the gate deadline without + // writing the caught-up marker, so a later reconcile retries the catch-up. + CatchUpOnTimeoutProceed = "Proceed" + // defaultCompletedOnPodNotReadyThreshold is the minimum time a pod must remain in // Ready=False before the operator considers the host stuck and re-enqueues a reconcile defaultCompletedOnPodNotReadyThreshold = 5 * time.Minute @@ -226,6 +233,15 @@ const ( defaultMaxReplicationDelay = 10 ) +const ( + defaultReconcileHostWaitReplicasCatchUpOnTimeout = CatchUpOnTimeoutAbort + // defaultReconcileHostWaitReplicasCatchUpTimeoutSeconds bounds one host catch-up. 0 would mean + // "never give up" - the gate would poll until the reconcile context dies, with no hard failure. + defaultReconcileHostWaitReplicasCatchUpTimeoutSeconds = 900 + defaultReconcileHostWaitReplicasCatchUpHealthPollSeconds = 10 + defaultReconcileHostWaitReplicasCatchUpHealthSuccessThreshold = 6 +) + // OperatorConfig specifies operator configuration // !!! IMPORTANT !!! // !!! IMPORTANT !!! @@ -721,6 +737,7 @@ func (wait ReconcileHostWait) Normalize() ReconcileHostWait { // Default update timeout in seconds wait.Replicas.Delay = types.NewInt32(defaultMaxReplicationDelay) } + wait.Replicas.CatchUp = wait.Replicas.CatchUp.Normalize() if wait.Probes == nil { wait.Probes = &ReconcileHostWaitProbes{} @@ -759,9 +776,116 @@ func (drop ReconcileHostDrop) MergeFrom(from ReconcileHostDrop) ReconcileHostDro } type ReconcileHostWaitReplicas struct { - All *types.StringBool `json:"all,omitempty" yaml:"all,omitempty"` - New *types.StringBool `json:"new,omitempty" yaml:"new,omitempty"` - Delay *types.Int32 `json:"delay,omitempty" yaml:"delay,omitempty"` + All *types.StringBool `json:"all,omitempty" yaml:"all,omitempty"` + New *types.StringBool `json:"new,omitempty" yaml:"new,omitempty"` + Delay *types.Int32 `json:"delay,omitempty" yaml:"delay,omitempty"` + CatchUp *ReconcileHostWaitReplicasCatchUp `json:"catchUp,omitempty" yaml:"catchUp,omitempty"` +} + +// ReconcileHostWaitReplicasCatchUp configures a replicated-host catch-up gate before advancing shard rolling reconcile. +type ReconcileHostWaitReplicasCatchUp struct { + Enabled *types.StringBool `json:"enabled,omitempty" yaml:"enabled,omitempty"` + Timeout *types.Int32 `json:"timeout,omitempty" yaml:"timeout,omitempty"` + OnTimeout *types.String `json:"onTimeout,omitempty" yaml:"onTimeout,omitempty"` + Health *ReconcileHostWaitReplicasCatchUpHealth `json:"health,omitempty" yaml:"health,omitempty"` +} + +// ReconcileHostWaitReplicasCatchUpHealth configures the stable-health window after replicated-host sync. +type ReconcileHostWaitReplicasCatchUpHealth struct { + PollInterval *types.Int32 `json:"pollInterval,omitempty" yaml:"pollInterval,omitempty"` + SuccessThreshold *types.Int32 `json:"successThreshold,omitempty" yaml:"successThreshold,omitempty"` +} + +// isValidReconcileHostWaitReplicasCatchUpOnTimeout accepts either canonical spelling in any case - +// the CRD lists both, matching how every other enum-valued option in this config is handled. +func isValidReconcileHostWaitReplicasCatchUpOnTimeout(value string) bool { + return strings.EqualFold(value, CatchUpOnTimeoutAbort) || strings.EqualFold(value, CatchUpOnTimeoutProceed) +} + +func (catchUpConfig *ReconcileHostWaitReplicasCatchUp) Normalize() *ReconcileHostWaitReplicasCatchUp { + if catchUpConfig == nil { + catchUpConfig = &ReconcileHostWaitReplicasCatchUp{} + } + catchUpConfig.Enabled = catchUpConfig.Enabled.Normalize(false) + if (catchUpConfig.Timeout == nil) || (catchUpConfig.Timeout.Value() <= 0) { + catchUpConfig.Timeout = types.NewInt32(defaultReconcileHostWaitReplicasCatchUpTimeoutSeconds) + } + if !isValidReconcileHostWaitReplicasCatchUpOnTimeout(catchUpConfig.OnTimeout.Value()) { + catchUpConfig.OnTimeout = types.NewString(defaultReconcileHostWaitReplicasCatchUpOnTimeout) + } + catchUpConfig.Health = catchUpConfig.Health.Normalize() + return catchUpConfig +} + +func (health *ReconcileHostWaitReplicasCatchUpHealth) Normalize() *ReconcileHostWaitReplicasCatchUpHealth { + if health == nil { + health = &ReconcileHostWaitReplicasCatchUpHealth{} + } + if (health.PollInterval == nil) || (health.PollInterval.Value() <= 0) { + health.PollInterval = types.NewInt32(defaultReconcileHostWaitReplicasCatchUpHealthPollSeconds) + } + if (health.SuccessThreshold == nil) || (health.SuccessThreshold.Value() <= 0) { + health.SuccessThreshold = types.NewInt32(defaultReconcileHostWaitReplicasCatchUpHealthSuccessThreshold) + } + return health +} + +func (catchUpConfig *ReconcileHostWaitReplicasCatchUp) MergeFrom(from *ReconcileHostWaitReplicasCatchUp) *ReconcileHostWaitReplicasCatchUp { + if from == nil { + return catchUpConfig + } + if catchUpConfig == nil { + catchUpConfig = &ReconcileHostWaitReplicasCatchUp{} + } + catchUpConfig.Enabled = catchUpConfig.Enabled.MergeFrom(from.Enabled) + catchUpConfig.Timeout = catchUpConfig.Timeout.MergeFrom(from.Timeout) + catchUpConfig.OnTimeout = catchUpConfig.OnTimeout.MergeFrom(from.OnTimeout) + catchUpConfig.Health = catchUpConfig.Health.MergeFrom(from.Health) + return catchUpConfig +} + +func (health *ReconcileHostWaitReplicasCatchUpHealth) MergeFrom(from *ReconcileHostWaitReplicasCatchUpHealth) *ReconcileHostWaitReplicasCatchUpHealth { + if from == nil { + return health + } + if health == nil { + health = &ReconcileHostWaitReplicasCatchUpHealth{} + } + health.PollInterval = health.PollInterval.MergeFrom(from.PollInterval) + health.SuccessThreshold = health.SuccessThreshold.MergeFrom(from.SuccessThreshold) + return health +} + +func (catchUpConfig *ReconcileHostWaitReplicasCatchUp) IsEnabled() bool { + return (catchUpConfig != nil) && catchUpConfig.Enabled.Value() +} + +func (catchUpConfig *ReconcileHostWaitReplicasCatchUp) GetTimeout() int { + if (catchUpConfig == nil) || (catchUpConfig.Timeout == nil) || (catchUpConfig.Timeout.Value() <= 0) { + return defaultReconcileHostWaitReplicasCatchUpTimeoutSeconds + } + return catchUpConfig.Timeout.IntValue() +} + +func (catchUpConfig *ReconcileHostWaitReplicasCatchUp) GetOnTimeout() string { + if (catchUpConfig == nil) || !isValidReconcileHostWaitReplicasCatchUpOnTimeout(catchUpConfig.OnTimeout.Value()) { + return defaultReconcileHostWaitReplicasCatchUpOnTimeout + } + return catchUpConfig.OnTimeout.Value() +} + +func (catchUpConfig *ReconcileHostWaitReplicasCatchUp) GetPollInterval() int { + if (catchUpConfig == nil) || (catchUpConfig.Health == nil) || (catchUpConfig.Health.PollInterval == nil) || (catchUpConfig.Health.PollInterval.Value() <= 0) { + return defaultReconcileHostWaitReplicasCatchUpHealthPollSeconds + } + return catchUpConfig.Health.PollInterval.IntValue() +} + +func (catchUpConfig *ReconcileHostWaitReplicasCatchUp) GetSuccessThreshold() int { + if (catchUpConfig == nil) || (catchUpConfig.Health == nil) || (catchUpConfig.Health.SuccessThreshold == nil) || (catchUpConfig.Health.SuccessThreshold.Value() <= 0) { + return defaultReconcileHostWaitReplicasCatchUpHealthSuccessThreshold + } + return catchUpConfig.Health.SuccessThreshold.IntValue() } func (r *ReconcileHostWaitReplicas) MergeFrom(from *ReconcileHostWaitReplicas) *ReconcileHostWaitReplicas { @@ -782,6 +906,7 @@ func (r *ReconcileHostWaitReplicas) MergeFrom(from *ReconcileHostWaitReplicas) * r.All = r.All.MergeFrom(from.All) r.New = r.New.MergeFrom(from.New) r.Delay = r.Delay.MergeFrom(from.Delay) + r.CatchUp = r.CatchUp.MergeFrom(from.CatchUp) return r } diff --git a/pkg/apis/clickhouse.altinity.com/v1/type_configuration_chop_catchup_test.go b/pkg/apis/clickhouse.altinity.com/v1/type_configuration_chop_catchup_test.go new file mode 100644 index 000000000..57021f639 --- /dev/null +++ b/pkg/apis/clickhouse.altinity.com/v1/type_configuration_chop_catchup_test.go @@ -0,0 +1,65 @@ +package v1 + +import ( + "strings" + "testing" + + "github.com/altinity/clickhouse-operator/pkg/apis/common/types" +) + +func TestReconcileHostWaitReplicasCatchUpNormalizeDefaults(t *testing.T) { + var catchUpConfig *ReconcileHostWaitReplicasCatchUp + catchUpConfig = catchUpConfig.Normalize() + if catchUpConfig.IsEnabled() { + t.Fatalf("enabled must default to false") + } + if catchUpConfig.GetTimeout() != 900 { + t.Fatalf("timeout default = %d, want 900", catchUpConfig.GetTimeout()) + } + if !strings.EqualFold(catchUpConfig.GetOnTimeout(), CatchUpOnTimeoutAbort) { + t.Fatalf("onTimeout default = %q, want %q", catchUpConfig.GetOnTimeout(), CatchUpOnTimeoutAbort) + } + if catchUpConfig.GetPollInterval() != 10 || catchUpConfig.GetSuccessThreshold() != 6 { + t.Fatalf("health defaults = %d/%d, want 10/6", catchUpConfig.GetPollInterval(), catchUpConfig.GetSuccessThreshold()) + } +} + +func TestReconcileHostWaitReplicasCatchUpNormalizeRejectsInvalid(t *testing.T) { + catchUpConfig := &ReconcileHostWaitReplicasCatchUp{ + Timeout: types.NewInt32(-5), + OnTimeout: types.NewString("explode"), + Health: &ReconcileHostWaitReplicasCatchUpHealth{ + PollInterval: types.NewInt32(0), + SuccessThreshold: types.NewInt32(-1), + }, + } + catchUpConfig = catchUpConfig.Normalize() + if !strings.EqualFold(catchUpConfig.GetOnTimeout(), CatchUpOnTimeoutAbort) { + t.Fatalf("invalid enums must fall back to defaults") + } + if catchUpConfig.GetTimeout() != 900 || catchUpConfig.GetPollInterval() != 10 || catchUpConfig.GetSuccessThreshold() != 6 { + t.Fatalf("invalid numerics must fall back to defaults") + } +} + +func TestReconcileHostWaitReplicasCatchUpMergeFromPrefersLocal(t *testing.T) { + localSyncConfig := (&ReconcileHostWaitReplicasCatchUp{Enabled: types.NewStringBool(true)}).Normalize() + parentSyncConfig := (&ReconcileHostWaitReplicasCatchUp{Enabled: types.NewStringBool(false), Timeout: types.NewInt32(30)}).Normalize() + mergedSyncConfig := localSyncConfig.MergeFrom(parentSyncConfig) + if !mergedSyncConfig.IsEnabled() { + t.Fatalf("merge must prefer local enabled=true") + } +} + +// Normalize must keep whatever case it is given rather than discarding it as invalid and +// silently reverting to the default. The CRD advertises the two canonical spellings, so those +// are what an API user can set; the config-file path is not schema-checked, hence the +// arbitrary-case entry below. +func TestReconcileHostWaitReplicasCatchUpOnTimeoutAcceptsEitherCase(t *testing.T) { + for _, onTimeout := range []string{"abort", "Abort", "proceed", "Proceed", "PROCEED"} { + catchUpConfig := (&ReconcileHostWaitReplicasCatchUp{OnTimeout: types.NewString(onTimeout)}).Normalize() + if catchUpConfig.GetOnTimeout() != onTimeout { + t.Fatalf("onTimeout %q was not accepted, got %q", onTimeout, catchUpConfig.GetOnTimeout()) + } + } +} diff --git a/pkg/apis/clickhouse.altinity.com/v1/type_host.go b/pkg/apis/clickhouse.altinity.com/v1/type_host.go index d3aeb1c1e..e530b98c6 100644 --- a/pkg/apis/clickhouse.altinity.com/v1/type_host.go +++ b/pkg/apis/clickhouse.altinity.com/v1/type_host.go @@ -66,6 +66,7 @@ type HostRuntime struct { reconcileAttributes *types.ReconcileAttributes `json:"-" yaml:"-" testdiff:"ignore"` replicas *types.Int32 `json:"-" yaml:"-"` hasData bool `json:"-" yaml:"-"` + forceReplicaCatchUp bool `json:"-" yaml:"-"` // CurStatefulSet is a current stateful set, fetched from k8s CurStatefulSet *apps.StatefulSet `json:"-" yaml:"-" testdiff:"ignore"` @@ -736,6 +737,20 @@ func (host *Host) SetHasData(hasData bool) { host.Runtime.hasData = hasData } +func (host *Host) IsForceReplicaCatchUp() bool { + if host == nil { + return false + } + return host.Runtime.forceReplicaCatchUp +} + +func (host *Host) SetForceReplicaCatchUp(force bool) { + if host == nil { + return + } + host.Runtime.forceReplicaCatchUp = force +} + func (host *Host) IsZero() bool { return host == nil } diff --git a/pkg/apis/clickhouse.altinity.com/v1/type_host_runtime_test.go b/pkg/apis/clickhouse.altinity.com/v1/type_host_runtime_test.go new file mode 100644 index 000000000..1dd1edf63 --- /dev/null +++ b/pkg/apis/clickhouse.altinity.com/v1/type_host_runtime_test.go @@ -0,0 +1,20 @@ +package v1 + +import "testing" + +func TestHostForceReplicaCatchUpDefaultsFalseAndCanBeSet(t *testing.T) { + host := &Host{} + if host.IsForceReplicaCatchUp() { + t.Fatalf("force replica catch-up must default to false") + } + + host.SetForceReplicaCatchUp(true) + if !host.IsForceReplicaCatchUp() { + t.Fatalf("force replica catch-up must be true after SetForceReplicaCatchUp(true)") + } + + host.SetForceReplicaCatchUp(false) + if host.IsForceReplicaCatchUp() { + t.Fatalf("force replica catch-up must be false after SetForceReplicaCatchUp(false)") + } +} diff --git a/pkg/apis/clickhouse.altinity.com/v1/type_status.go b/pkg/apis/clickhouse.altinity.com/v1/type_status.go index 5ae4417fb..feff96770 100644 --- a/pkg/apis/clickhouse.altinity.com/v1/type_status.go +++ b/pkg/apis/clickhouse.altinity.com/v1/type_status.go @@ -208,6 +208,20 @@ func (s *Status) PushHostReplicaCaughtUp(host string) { }) } +// RemoveHostReplicaCaughtUp removes host from the list of hosts with replica caught-up +func (s *Status) RemoveHostReplicaCaughtUp(host string) { + host = util.NormalizeFQDN(host) + doWithWriteLock(s, func(s *Status) { + hosts := s.HostsWithReplicaCaughtUp[:0] + for _, caughtUpHost := range s.HostsWithReplicaCaughtUp { + if caughtUpHost != host { + hosts = append(hosts, caughtUpHost) + } + } + s.HostsWithReplicaCaughtUp = hosts + }) +} + // PushHostTablesCreated pushes host to the list of hosts with created tables func (s *Status) PushHostTablesCreated(host string) { host = util.NormalizeFQDN(host) diff --git a/pkg/apis/clickhouse.altinity.com/v1/type_status_test.go b/pkg/apis/clickhouse.altinity.com/v1/type_status_test.go index e30b0653f..0a72ef677 100644 --- a/pkg/apis/clickhouse.altinity.com/v1/type_status_test.go +++ b/pkg/apis/clickhouse.altinity.com/v1/type_status_test.go @@ -8,6 +8,7 @@ import ( "github.com/stretchr/testify/require" "github.com/altinity/clickhouse-operator/pkg/apis/common/types" + "github.com/altinity/clickhouse-operator/pkg/util" ) var normalizedChiA = &ClickHouseInstallation{} @@ -109,6 +110,21 @@ func TestCopyFromUsedTemplates(t *testing.T) { }) } +func TestRemoveHostReplicaCaughtUp(t *testing.T) { + const fqdn = "chi-x-default-0-0" + status := &Status{} + status.PushHostReplicaCaughtUp(fqdn) + status.PushHostReplicaCaughtUp("chi-x-default-0-1") + + status.RemoveHostReplicaCaughtUp(fqdn) + + for _, host := range status.GetHostsWithReplicaCaughtUp() { + if host == util.NormalizeFQDN(fqdn) { + t.Fatalf("host should have been removed: %v", status.GetHostsWithReplicaCaughtUp()) + } + } +} + // NB: These tests mostly exist to exercise synchronization and detect regressions related to them via the // Golang race detector. See: https://go.dev/blog/race-detector // In short, add -race to the go test flags when running this. diff --git a/pkg/apis/clickhouse.altinity.com/v1/zz_generated.deepcopy.go b/pkg/apis/clickhouse.altinity.com/v1/zz_generated.deepcopy.go index dd8160f73..1495497b2 100644 --- a/pkg/apis/clickhouse.altinity.com/v1/zz_generated.deepcopy.go +++ b/pkg/apis/clickhouse.altinity.com/v1/zz_generated.deepcopy.go @@ -2918,6 +2918,11 @@ func (in *ReconcileHostWaitReplicas) DeepCopyInto(out *ReconcileHostWaitReplicas *out = new(types.Int32) **out = **in } + if in.CatchUp != nil { + in, out := &in.CatchUp, &out.CatchUp + *out = new(ReconcileHostWaitReplicasCatchUp) + (*in).DeepCopyInto(*out) + } return } @@ -2931,6 +2936,68 @@ func (in *ReconcileHostWaitReplicas) DeepCopy() *ReconcileHostWaitReplicas { return out } +// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil. +func (in *ReconcileHostWaitReplicasCatchUp) DeepCopyInto(out *ReconcileHostWaitReplicasCatchUp) { + *out = *in + if in.Enabled != nil { + in, out := &in.Enabled, &out.Enabled + *out = new(types.StringBool) + **out = **in + } + if in.Timeout != nil { + in, out := &in.Timeout, &out.Timeout + *out = new(types.Int32) + **out = **in + } + if in.OnTimeout != nil { + in, out := &in.OnTimeout, &out.OnTimeout + *out = new(types.String) + **out = **in + } + if in.Health != nil { + in, out := &in.Health, &out.Health + *out = new(ReconcileHostWaitReplicasCatchUpHealth) + (*in).DeepCopyInto(*out) + } + return +} + +// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new ReconcileHostWaitReplicasCatchUp. +func (in *ReconcileHostWaitReplicasCatchUp) DeepCopy() *ReconcileHostWaitReplicasCatchUp { + if in == nil { + return nil + } + out := new(ReconcileHostWaitReplicasCatchUp) + in.DeepCopyInto(out) + return out +} + +// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil. +func (in *ReconcileHostWaitReplicasCatchUpHealth) DeepCopyInto(out *ReconcileHostWaitReplicasCatchUpHealth) { + *out = *in + if in.PollInterval != nil { + in, out := &in.PollInterval, &out.PollInterval + *out = new(types.Int32) + **out = **in + } + if in.SuccessThreshold != nil { + in, out := &in.SuccessThreshold, &out.SuccessThreshold + *out = new(types.Int32) + **out = **in + } + return +} + +// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new ReconcileHostWaitReplicasCatchUpHealth. +func (in *ReconcileHostWaitReplicasCatchUpHealth) DeepCopy() *ReconcileHostWaitReplicasCatchUpHealth { + if in == nil { + return nil + } + out := new(ReconcileHostWaitReplicasCatchUpHealth) + in.DeepCopyInto(out) + return out +} + // DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil. func (in *ReconcileMacros) DeepCopyInto(out *ReconcileMacros) { *out = *in diff --git a/pkg/controller/chi/worker-catchup-gate_test.go b/pkg/controller/chi/worker-catchup-gate_test.go new file mode 100644 index 000000000..f390ab845 --- /dev/null +++ b/pkg/controller/chi/worker-catchup-gate_test.go @@ -0,0 +1,76 @@ +package chi + +import ( + "errors" + "testing" + "time" + + common "github.com/altinity/clickhouse-operator/pkg/controller/common" + a "github.com/altinity/clickhouse-operator/pkg/controller/common/announcer" +) + +func healthWindowStepForTest(counter int, ok bool, threshold int) (int, bool) { + return healthWindowStep(counter, ok, threshold) +} + +func TestHealthWindowConsecutive(t *testing.T) { + counter := 0 + done := false + for i := 0; i < 6; i++ { + counter, done = healthWindowStepForTest(counter, true, 6) + } + if !done || counter != 6 { + t.Fatalf("6 consecutive OK must satisfy threshold; counter=%d done=%v", counter, done) + } +} + +func TestHealthWindowResetsOnFailure(t *testing.T) { + counter, _ := healthWindowStepForTest(0, true, 6) + counter, _ = healthWindowStepForTest(counter, true, 6) + counter, done := healthWindowStepForTest(counter, false, 6) + if counter != 0 || done { + t.Fatalf("not-OK poll must reset counter; counter=%d done=%v", counter, done) + } +} + +func TestOnSoftTimeoutNeverPushesMarker(t *testing.T) { + // Both spellings of each value, because this is where the case-insensitive comparison + // actually happens - the config layer only validates, it does not canonicalize, so a + // case-sensitive check here would silently abort a reconcile configured with "Proceed". + for _, proceed := range []string{"proceed", "Proceed"} { + advance, pushMarker, err := onSoftTimeout(proceed) + if !advance || pushMarker || err != nil { + t.Fatalf("%s => advance without marker; got advance=%v push=%v err=%v", proceed, advance, pushMarker, err) + } + } + + for _, abort := range []string{"abort", "Abort", ""} { + advance, pushMarker, err := onSoftTimeout(abort) + if advance || pushMarker || !errors.Is(err, common.ErrCRUDAbort) { + t.Fatalf("%q => abort without marker; got advance=%v push=%v err=%v", abort, advance, pushMarker, err) + } + } +} + +func TestCatchUpGateHealthStepTreatsHardFailAsNotReadyBeforeDeadline(t *testing.T) { + counter, done, hardDeadline := catchUpGateHealthStep(3, true, true, 6, time.Second) + if counter != 0 || done || hardDeadline { + t.Fatalf("hard health before deadline must reset and keep waiting; counter=%d done=%v hardDeadline=%v", counter, done, hardDeadline) + } +} + +func TestCatchUpGateHealthStepReturnsHardFailAtDeadline(t *testing.T) { + counter, done, hardDeadline := catchUpGateHealthStep(3, true, true, 6, 0) + if counter != 0 || done || !hardDeadline { + t.Fatalf("hard health at deadline must hard fail; counter=%d done=%v hardDeadline=%v", counter, done, hardDeadline) + } +} + +func TestReplicaCatchUpGateEventReasonDistinguishesProceedWithoutMarker(t *testing.T) { + if got := replicaCatchUpGateEventReason(true); got != a.EventReasonReconcileCompleted { + t.Fatalf("caught-up sync gate must report completed event; got %s", got) + } + if got := replicaCatchUpGateEventReason(false); got == a.EventReasonReconcileCompleted { + t.Fatalf("proceed without marker must not report completed event") + } +} diff --git a/pkg/controller/chi/worker-deleter.go b/pkg/controller/chi/worker-deleter.go index 6a91eff90..88b14eeeb 100644 --- a/pkg/controller/chi/worker-deleter.go +++ b/pkg/controller/chi/worker-deleter.go @@ -534,8 +534,8 @@ func (w *worker) deleteHost(ctx context.Context, chi *api.ClickHouseInstallation return nil } - w.a.V(2).M(host).S().Info(host.Runtime.Address.HostName) - defer w.a.V(2).M(host).E().Info(host.Runtime.Address.HostName) + w.a.V(2).M(host).S().Info("%s", host.Runtime.Address.HostName) + defer w.a.V(2).M(host).E().Info("%s", host.Runtime.Address.HostName) w.a.V(1). WithEvent(host.GetCR(), a.EventActionDelete, a.EventReasonDeleteStarted). diff --git a/pkg/controller/chi/worker-reconciler-chi.go b/pkg/controller/chi/worker-reconciler-chi.go index 00c416da5..fb94d89a6 100644 --- a/pkg/controller/chi/worker-reconciler-chi.go +++ b/pkg/controller/chi/worker-reconciler-chi.go @@ -219,7 +219,7 @@ func (w *worker) buildCR(ctx context.Context, _cr *api.ClickHouseInstallation) * actionPlan := api.MakeActionPlan(cr.GetAncestorT(), cr) cr.EnsureRuntime().ActionPlan = actionPlan cr.EnsureStatus().SetActionPlan(actionPlan) - w.a.V(1).M(cr).Info(actionPlan.Log("buildCR")) + w.a.V(1).M(cr).Info("%s", actionPlan.Log("buildCR")) return cr } @@ -1123,6 +1123,7 @@ func (w *worker) reconcileHostMain(ctx context.Context, host *api.Host) error { w.a.V(1).M(host).F().Warning("Data loss detected for host: %s. Aborting reconcile as configured (onDataLoss: abort)", host.GetName()) return common.ErrCRUDAbort } + w.forceReplicaCatchUpAfterStorageLoss(host, w.c.namer.Name(interfaces.NameFQDN, host)) stsReconcileOpts, migrateTableOpts = w.hostPVCsDataLossDetectedOptions(host) w.a.V(1). M(host).F(). @@ -1133,6 +1134,7 @@ func (w *worker) reconcileHostMain(ctx context.Context, host *api.Host) error { return common.ErrCRUDAbort } // stsReconcileOpts, migrateTableOpts = w.hostPVCsDataVolumeMissedDetectedOptions(host) + w.forceReplicaCatchUpAfterStorageLoss(host, w.c.namer.Name(interfaces.NameFQDN, host)) stsReconcileOpts, migrateTableOpts = w.hostPVCsDataLossDetectedOptions(host) w.a.V(1). M(host).F(). @@ -1195,6 +1197,14 @@ func (w *worker) prepareStsReconcileOptsWaitSection(host *api.Host, opts *statef return opts } +// forceReplicaCatchUpAfterStorageLoss invalidates the persisted caught-up marker of a host that lost +// its storage. Unconditional by design: the stale marker is a correctness bug on its own (the host is +// empty yet listed as caught-up), and clearing it must not depend on the sync gate being enabled. +func (w *worker) forceReplicaCatchUpAfterStorageLoss(host *api.Host, fqdn string) { + host.SetForceReplicaCatchUp(true) + host.GetCR().IEnsureStatus().RemoveHostReplicaCaughtUp(fqdn) +} + func (w *worker) reconcileHostPVCs(ctx context.Context, host *api.Host) storage.ErrorDataPersistence { return storage.NewStorageReconciler( w.task, diff --git a/pkg/controller/chi/worker-reconciler-chi_test.go b/pkg/controller/chi/worker-reconciler-chi_test.go index 28381362c..119b6ebf7 100644 --- a/pkg/controller/chi/worker-reconciler-chi_test.go +++ b/pkg/controller/chi/worker-reconciler-chi_test.go @@ -23,6 +23,8 @@ import ( core "k8s.io/api/core/v1" api "github.com/altinity/clickhouse-operator/pkg/apis/clickhouse.altinity.com/v1" + "github.com/altinity/clickhouse-operator/pkg/apis/common/types" + "github.com/altinity/clickhouse-operator/pkg/chop" ) // sts is a small builder for an apps/v1 StatefulSet with a single-container pod template @@ -47,6 +49,51 @@ func hostWith(cur, desired *apps.StatefulSet) *api.Host { return h } +func withReplicaCatchUpGate(t *testing.T, enabled bool) { + t.Helper() + cfg := chop.Config() + prev := cfg.Reconcile.Host.Wait.Replicas.CatchUp + t.Cleanup(func() { + cfg.Reconcile.Host.Wait.Replicas.CatchUp = prev + }) + cfg.Reconcile.Host.Wait.Replicas.CatchUp = (&api.ReconcileHostWaitReplicasCatchUp{ + Enabled: types.NewStringBool(enabled), + }).Normalize() +} + +func hostWithReplicaCaughtUpMarker(fqdn string) *api.Host { + cr := &api.ClickHouseInstallation{} + host := &api.Host{} + host.SetCR(cr) + host.GetCR().IEnsureStatus().PushHostReplicaCaughtUp(fqdn) + return host +} + +// The stale caught-up marker is invalid regardless of the gate - clearing it is unconditional. +func TestForceReplicaCatchUpAfterStorageLossClearsMarkerWhenCatchUpGateDisabled(t *testing.T) { + const fqdn = "chi-x-default-0-0" + withReplicaCatchUpGate(t, false) + host := hostWithReplicaCaughtUpMarker(fqdn) + w := &worker{} + + w.forceReplicaCatchUpAfterStorageLoss(host, fqdn) + + require.True(t, host.IsForceReplicaCatchUp()) + require.False(t, host.HasListedReplicaCaughtUp(fqdn)) +} + +func TestForceReplicaCatchUpAfterStorageLossClearsMarkerWhenCatchUpGateEnabled(t *testing.T) { + const fqdn = "chi-x-default-0-0" + withReplicaCatchUpGate(t, true) + host := hostWithReplicaCaughtUpMarker(fqdn) + w := &worker{} + + w.forceReplicaCatchUpAfterStorageLoss(host, fqdn) + + require.True(t, host.IsForceReplicaCatchUp()) + require.False(t, host.HasListedReplicaCaughtUp(fqdn)) +} + // TestHostRequiresStatefulSetRollout exercises the pure decision function that gates // the pre-rollout software restart in reconcileHostStatefulSet. // diff --git a/pkg/controller/chi/worker-secret.go b/pkg/controller/chi/worker-secret.go index 4b15ad52a..42028bd20 100644 --- a/pkg/controller/chi/worker-secret.go +++ b/pkg/controller/chi/worker-secret.go @@ -25,8 +25,8 @@ import ( // reconcileSecret reconciles core.Secret func (w *worker) reconcileSecret(ctx context.Context, cr api.ICustomResource, secret *core.Secret) error { - w.a.V(2).M(cr).S().Info(secret.Name) - defer w.a.V(2).M(cr).E().Info(secret.Name) + w.a.V(2).M(cr).S().Info("%s", secret.Name) + defer w.a.V(2).M(cr).E().Info("%s", secret.Name) // Check whether this object already exists if _, err := w.c.getSecret(ctx, secret); err == nil { diff --git a/pkg/controller/chi/worker-service.go b/pkg/controller/chi/worker-service.go index 99f4550b3..8a7c13d50 100644 --- a/pkg/controller/chi/worker-service.go +++ b/pkg/controller/chi/worker-service.go @@ -29,8 +29,8 @@ import ( // reconcileService reconciles core.Service func (w *worker) reconcileService(ctx context.Context, cr chi.ICustomResource, service, prevService *core.Service) error { - w.a.V(2).M(cr).S().Info(service.GetName()) - defer w.a.V(2).M(cr).E().Info(service.GetName()) + w.a.V(2).M(cr).S().Info("%s", service.GetName()) + defer w.a.V(2).M(cr).E().Info("%s", service.GetName()) // Check whether this object already exists curService, err := w.c.getService(ctx, service) diff --git a/pkg/controller/chi/worker-status-helpers.go b/pkg/controller/chi/worker-status-helpers.go index 003f84bdf..a9e3d7f48 100644 --- a/pkg/controller/chi/worker-status-helpers.go +++ b/pkg/controller/chi/worker-status-helpers.go @@ -16,6 +16,7 @@ package chi import ( "context" + "errors" "time" core "k8s.io/api/core/v1" @@ -193,6 +194,45 @@ func (w *worker) hasUnhealthyHosts(ctx context.Context, cr *api.ClickHouseInstal return found } +func (w *worker) catchUpHealthOK(ctx context.Context, host *api.Host, deadline time.Time) (ok bool, hardFail bool, err error) { + clusterSchemer := w.ensureClusterSchemer(host) + readHealth := func(read func(context.Context, *api.Host) (int, error)) (int, bool, error) { + if contextError := ctx.Err(); contextError != nil { + return 0, false, contextError + } + queryCtx, cancel := context.WithDeadline(ctx, deadline) + defer cancel() + healthValue, queryErr := read(queryCtx, host) + if contextError := ctx.Err(); contextError != nil { + return 0, false, contextError + } + if (queryCtx.Err() != nil) || errors.Is(queryErr, context.DeadlineExceeded) { + return 0, true, nil + } + if queryErr != nil { + return 0, false, queryErr + } + return healthValue, false, nil + } + + readonly, notReady, err := readHealth(clusterSchemer.HostMaxIsReadonly) + if (err != nil) || notReady { + return false, false, err + } + sessionExpired, notReady, err := readHealth(clusterSchemer.HostMaxIsSessionExpired) + if (err != nil) || notReady { + return false, false, err + } + replicaDelay, notReady, err := readHealth(clusterSchemer.HostMaxReplicaDelay) + if (err != nil) || notReady { + return false, false, err + } + if (readonly != 0) || (sessionExpired != 0) { + return false, true, nil + } + return replicaDelay <= chop.Config().Reconcile.Host.Wait.Replicas.Delay.IntValue(), false, nil +} + // isOperatorIPTheSame reports whether the operator pod IP still matches the // CHOpIP persisted on the CR from the previous reconcile. A changed IP must // force reconcile so clickhouse-operator user networks/host_regexp are refreshed. @@ -271,6 +311,11 @@ func (w *worker) doesHostHaveNoRunningQueries(ctx context.Context, host *api.Hos return n <= 1 } +// doesHostHaveNoReplicationDelay is a poll predicate, so returning false means "keep waiting". +// +// A failed query yields a delay of 0, which reads as "no lag" and hands out a caught-up verdict +// the host never earned. Answering false instead is worse: the poll driving this predicate is +// uncapped, so an unreachable host would be polled forever and its reconcile thread pinned. func (w *worker) doesHostHaveNoReplicationDelay(ctx context.Context, host *api.Host) bool { delay, _ := w.ensureClusterSchemer(host).HostMaxReplicaDelay(ctx, host) log.V(1).Info("replication lag %d host: %s", delay, host.GetName()) diff --git a/pkg/controller/chi/worker-wait-exclude-include-restart.go b/pkg/controller/chi/worker-wait-exclude-include-restart.go index c47296a09..080dc5991 100644 --- a/pkg/controller/chi/worker-wait-exclude-include-restart.go +++ b/pkg/controller/chi/worker-wait-exclude-include-restart.go @@ -16,19 +16,43 @@ package chi import ( "context" + "errors" + "fmt" + "strings" "time" log "github.com/altinity/clickhouse-operator/pkg/announcer" api "github.com/altinity/clickhouse-operator/pkg/apis/clickhouse.altinity.com/v1" "github.com/altinity/clickhouse-operator/pkg/apis/common/types" "github.com/altinity/clickhouse-operator/pkg/chop" + "github.com/altinity/clickhouse-operator/pkg/controller/chi/cmd_queue" + common "github.com/altinity/clickhouse-operator/pkg/controller/common" a "github.com/altinity/clickhouse-operator/pkg/controller/common/announcer" "github.com/altinity/clickhouse-operator/pkg/controller/common/poller" "github.com/altinity/clickhouse-operator/pkg/controller/common/poller/domain" "github.com/altinity/clickhouse-operator/pkg/interfaces" + "github.com/altinity/clickhouse-operator/pkg/model/chi/schemer" "github.com/altinity/clickhouse-operator/pkg/util" ) +const ( + // replicationCatchUpPassTimeout bounds how long one reconcile pass waits for a host to catch + // up. It is not a budget for the whole catch-up - the replica fetches from its peers whether + // or not the operator is watching - so expiry costs nothing but the wait, and the CR is + // re-enqueued to resume it. + replicationCatchUpPassTimeout = 15 * time.Minute + // replicationCatchUpRetryDelay spaces out those retries so a replica that never converges + // re-checks periodically instead of spinning. + replicationCatchUpRetryDelay = 1 * time.Minute +) + +var ( + // errReplicationCatchUpNotFinished is returned when the per-pass wait expired with the host + // still behind. It is distinct from a hard failure: the caller keeps the host out of the + // Service and schedules another pass rather than aborting the reconcile. + errReplicationCatchUpNotFinished = errors.New("host has not caught up within this reconcile pass") +) + // waitForIPAddresses waits for all pods to get IP address assigned func (w *worker) waitForIPAddresses(ctx context.Context, cr *api.ClickHouseInstallation) { if util.IsContextDone(ctx) { @@ -145,6 +169,13 @@ func (w *worker) shouldWaitReplicationHost(host *api.Host) bool { host.Runtime.Address.ReplicaIndex, host.Runtime.Address.ShardIndex, host.Runtime.Address.ClusterName) return false + case host.IsForceReplicaCatchUp(): + w.a.V(1). + M(host).F(). + Info("Force replica catch-up after data loss. Host/shard/cluster: %d/%d/%s", + host.Runtime.Address.ReplicaIndex, host.Runtime.Address.ShardIndex, host.Runtime.Address.ClusterName) + return true + case host.IsFirstInCluster(): w.a.V(1). M(host).F(). @@ -191,6 +222,21 @@ func (w *worker) shouldWaitReplicationHost(host *api.Host) bool { return false } +func healthWindowStep(counter int, ok bool, threshold int) (int, bool) { + if !ok { + return 0, false + } + counter++ + return counter, counter >= threshold +} + +func onSoftTimeout(onTimeout string) (advance bool, pushMarker bool, err error) { + if strings.EqualFold(onTimeout, api.CatchUpOnTimeoutProceed) { + return true, false, nil + } + return false, false, common.ErrCRUDAbort +} + // includeHost includes host back into all activities - such as cluster, service, etc func (w *worker) includeHost(ctx context.Context, host *api.Host) error { w.a.V(1). @@ -198,9 +244,17 @@ func (w *worker) includeHost(ctx context.Context, host *api.Host) error { Info("Include host into cluster. Host/shard/cluster: %d/%d/%s", host.Runtime.Address.ReplicaIndex, host.Runtime.Address.ShardIndex, host.Runtime.Address.ClusterName) - // w.includeHostIntoClickHouseCluster(ctx, host) - w.ascendHostInClickHouseCluster(ctx, host) + catchUpGateEnabled := chop.Config().Reconcile.Host.Wait.Replicas.CatchUp.IsEnabled() + // Catch up FIRST, ascend afterwards. A host that was excluded is still carrying the low + // priority descendHostInClickHouseCluster gave it, so distributed queries keep preferring its + // up-to-date peers for the duration of the wait. (A host that was never excluded - a brand new + // one, or one the shard-safety guard declined to drain - is at normal priority throughout; + // ordering only matters for the excluded case.) The ascend is unconditional so a host whose + // catch-up failed still returns to normal priority in this pass: a conditional ascend would + // leave it deprioritized until some later pass regenerates the common ConfigMap, and once the + // CR reaches Completed the reconcile early-exit means that may be a long way off. err := w.catchReplicationLag(ctx, host) + w.ascendHostInClickHouseCluster(ctx, host) if err == nil { w.a.V(1). M(host).F(). @@ -212,6 +266,9 @@ func (w *worker) includeHost(ctx context.Context, host *api.Host) error { M(host).F(). Warning("Will NOT include host into cluster due to replication lag. Host/shard/cluster: %d/%d/%s", host.Runtime.Address.ReplicaIndex, host.Runtime.Address.ShardIndex, host.Runtime.Address.ClusterName) + if catchUpGateEnabled { + return err + } } return nil @@ -330,7 +387,34 @@ func (w *worker) catchReplicationLag(ctx context.Context, host *api.Host) error // Host is alive but catching up - add to monitoring so metrics are collected during the wait w.addHostToMonitoring(host) - err := w.waitHostHasNoReplicationDelay(ctx, host) + var err error + if chop.Config().Reconcile.Host.Wait.Replicas.CatchUp.IsEnabled() { + var caughtUp bool + caughtUp, err = w.runReplicaCatchUpGate(ctx, host) + if err == nil { + w.a.V(1). + M(host).F(). + WithEvent(host.GetCR(), a.EventActionReconcile, replicaCatchUpGateEventReason(caughtUp)). + Info("Wait for host to catch replication lag - %s "+ + "Host/shard/cluster: %d/%d/%s", + replicaCatchUpGateResultLabel(caughtUp), + host.Runtime.Address.ReplicaIndex, host.Runtime.Address.ShardIndex, host.Runtime.Address.ClusterName, + ) + } else { + w.a.V(1). + M(host).F(). + WithEvent(host.GetCR(), a.EventActionReconcile, a.EventReasonReconcileFailed). + Info("Wait for host to catch replication lag - FAILED "+ + "Host/shard/cluster: %d/%d/%s"+ + "err: %v ", + host.Runtime.Address.ReplicaIndex, host.Runtime.Address.ShardIndex, host.Runtime.Address.ClusterName, + err, + ) + } + return err + } + + err = w.waitHostHasNoReplicationDelay(ctx, host) if err == nil { w.a.V(1). M(host).F(). @@ -341,6 +425,11 @@ func (w *worker) catchReplicationLag(ctx context.Context, host *api.Host) error ) host.GetCR().IEnsureStatus().PushHostReplicaCaughtUp(w.c.namer.Name(interfaces.NameFQDN, host)) + } else if errors.Is(err, errReplicationCatchUpNotFinished) { + // Ran out of pass time, not a failure. Leave the host out of the Service - it is knowingly + // behind - and schedule another pass to resume the wait, so this releases the reconcile + // worker instead of holding it until the replica converges. + w.scheduleReplicationCatchUpRetry(host) } else { w.a.V(1). M(host).F(). @@ -356,6 +445,142 @@ func (w *worker) catchReplicationLag(ctx context.Context, host *api.Host) error return err } +func (w *worker) runReplicaCatchUpGate(ctx context.Context, host *api.Host) (bool, error) { + catchUpConfig := chop.Config().Reconcile.Host.Wait.Replicas.CatchUp + clusterSchemer := w.ensureClusterSchemer(host) + hostFQDN := w.c.namer.Name(interfaces.NameFQDN, host) + deadline := catchUpGateDeadline(catchUpConfig.GetTimeout()) + + failSoft := func(reason string) (bool, error) { + advance, _, err := onSoftTimeout(catchUpConfig.GetOnTimeout()) + if advance { + w.a.M(host).F().Warning("sync gate %s; proceeding without caught-up marker (onTimeout=proceed)", reason) + } + return false, err + } + classifyErr := func(err error) (bool, error) { + if err == nil { + return false, nil + } + if contextError := ctx.Err(); contextError != nil { + return false, contextError + } + if errors.Is(err, schemer.ErrGateDeadline) { + return failSoft("timed out") + } + return false, err + } + + if err := clusterSchemer.HostAsyncLoadBarrier(ctx, host, deadline); err != nil { + return classifyErr(err) + } + replicatedObjects, err := clusterSchemer.PeerReplicatedObjectCount(ctx, host, deadline) + if err != nil { + return classifyErr(err) + } + if replicatedObjects == 0 { + host.GetCR().IEnsureStatus().PushHostReplicaCaughtUp(hostFQDN) + return true, nil + } + if err := clusterSchemer.HostSyncReplicatedObjects(ctx, host, deadline); err != nil { + return classifyErr(err) + } + + healthCounter := 0 + for { + ok, hardFail, healthErr := w.catchUpHealthOK(ctx, host, deadline) + if healthErr != nil { + return classifyErr(healthErr) + } + + remaining := time.Until(deadline) + var done bool + var hardDeadline bool + healthCounter, done, hardDeadline = catchUpGateHealthStep(healthCounter, ok, hardFail, catchUpConfig.GetSuccessThreshold(), remaining) + if hardDeadline { + return false, catchUpGateHardFailError(host) + } + if done { + host.GetCR().IEnsureStatus().PushHostReplicaCaughtUp(hostFQDN) + return true, nil + } + + if remaining <= 0 { + return failSoft("health window not satisfied") + } + sleepDuration := time.Duration(catchUpConfig.GetPollInterval()) * time.Second + if sleepDuration > remaining { + sleepDuration = remaining + } + select { + case <-ctx.Done(): + return false, ctx.Err() + case <-time.After(sleepDuration): + if hardFail && !time.Now().Before(deadline) { + return false, catchUpGateHardFailError(host) + } + } + } +} + +func catchUpGateHealthStep(counter int, ok bool, hardFail bool, threshold int, remaining time.Duration) (int, bool, bool) { + if hardFail { + return 0, false, remaining <= 0 + } + nextCounter, done := healthWindowStep(counter, ok, threshold) + return nextCounter, done, false +} + +func catchUpGateHardFailError(host *api.Host) error { + return fmt.Errorf("host %s readonly or session-expired; refusing to advance", host.GetName()) +} + +func replicaCatchUpGateEventReason(caughtUp bool) string { + if caughtUp { + return a.EventReasonReconcileCompleted + } + return a.EventReasonReconcileProceed +} + +func replicaCatchUpGateResultLabel(caughtUp bool) string { + if caughtUp { + return "COMPLETED" + } + return "PROCEEDED without caught-up marker" +} + +// catchUpGateDeadline turns the configured budget into an absolute deadline. The caller passes +// GetTimeout(), which substitutes the default for a nil or non-positive value, so the budget is +// always positive - the gate has no unbounded mode. +func catchUpGateDeadline(timeoutSeconds int) time.Time { + return time.Now().Add(time.Duration(timeoutSeconds) * time.Second) +} + +// scheduleReplicationCatchUpRetry re-enqueues the CR so a later pass resumes a catch-up that did +// not finish within replicationCatchUpPassTimeout. Mirrors the stuck-host recovery scheduler: +// the queue coalesces by handle, so repeated scheduling cannot pile up work. +func (w *worker) scheduleReplicationCatchUpRetry(host *api.Host) { + // NewReconcileCHI takes the concrete CHI; GetCR() is the shared interface, and CHK has no + // catch-up wait, so a failed assertion simply means there is nothing to re-enqueue. + cr, ok := host.GetCR().(*api.ClickHouseInstallation) + if !ok || (cr == nil) { + return + } + + w.a.V(1). + M(host).F(). + WithEvent(cr, a.EventActionReconcile, a.EventReasonReplicationCatchUpRescheduled). + Warning("Host has not caught up within %s - left out of the service, re-enqueue in %s. Host/shard/cluster: %d/%d/%s", + replicationCatchUpPassTimeout, replicationCatchUpRetryDelay, + host.Runtime.Address.ReplicaIndex, host.Runtime.Address.ShardIndex, host.Runtime.Address.ClusterName, + ) + + scheduled := cr + time.AfterFunc(replicationCatchUpRetryDelay, func() { + w.c.enqueueObject(cmd_queue.NewReconcileCHI(cmd_queue.ReconcileAdd, nil, scheduled)) + }) +} + // shouldExcludeHost determines whether host to be excluded from cluster before reconcile func (w *worker) shouldExcludeHost(ctx context.Context, host *api.Host) bool { switch { @@ -553,9 +778,31 @@ func (w *worker) waitHostHasNoActiveQueries(ctx context.Context, host *api.Host) return domain.PollHost(ctx, host, w.doesHostHaveNoRunningQueries) } -// waitHostHasNoReplicationDelay +// waitHostHasNoReplicationDelay waits until the host reports a replication lag within the +// configured limit, for at most replicationCatchUpPassTimeout. +// +// The bound is per reconcile pass, not a budget for the whole catch-up: the replica fetches from +// its peers regardless of whether the operator is watching, so giving up here loses no progress. +// On expiry the caller leaves the host out of the Service - it is knowingly behind - and +// re-enqueues the CR, so a slow replica converges over several passes while a replica that can +// never converge stays visible instead of holding a reconcile worker for good. func (w *worker) waitHostHasNoReplicationDelay(ctx context.Context, host *api.Host) error { - return domain.PollHost(ctx, host, w.doesHostHaveNoReplicationDelay, &poller.Options{Timeout: time.Hour * 24 * 365 * 100}) + err := domain.PollHost(ctx, host, w.doesHostHaveNoReplicationDelay, &poller.Options{Timeout: replicationCatchUpPassTimeout}) + if err != nil { + return err + } + // The poller reports a cancelled context as success, and QueryHostInt answers a cancelled + // context with a delay of 0, so without this check an interrupted reconcile would look like + // a host that caught up - and the caller would persist that verdict. + if util.IsContextDone(ctx) { + return common.ErrCRUDAbort + } + // Poll() also returns nil when it simply ran out of time, so re-check the predicate: without + // this an expired wait is indistinguishable from a host that caught up. + if !w.doesHostHaveNoReplicationDelay(ctx, host) { + return errReplicationCatchUpNotFinished + } + return nil } // waitHostRestart diff --git a/pkg/controller/common/announcer/event-emitter.go b/pkg/controller/common/announcer/event-emitter.go index ac54daf34..0c76bf415 100644 --- a/pkg/controller/common/announcer/event-emitter.go +++ b/pkg/controller/common/announcer/event-emitter.go @@ -47,6 +47,7 @@ const ( EventReasonReconcileInProgress = "ReconcileInProgress" EventReasonReconcileCompleted = "ReconcileCompleted" EventReasonReconcileFailed = "ReconcileFailed" + EventReasonReconcileProceed = "ReconcileProceed" EventReasonCreateStarted = "CreateStarted" EventReasonCreateInProgress = "CreateInProgress" EventReasonCreateCompleted = "CreateCompleted" @@ -84,6 +85,11 @@ const ( // The shard keeps serving; the reconcile retries once a peer is back. EventReasonHostReconcileDeferredShardSafety = "HostReconcileDeferredShardSafety" + // EventReasonReplicationCatchUpRescheduled fires when a host did not catch up within one + // reconcile pass. The host is left out of the Service - it is knowingly behind - and the CR is + // re-enqueued so a later pass resumes the wait instead of holding a reconcile worker. + EventReasonReplicationCatchUpRescheduled = "ReplicationCatchUpRescheduled" + // EventReasonHookSkippedUnreachableHost fires when a cluster-scoped reconcile hook does not // run on one of its target hosts because that host's pod cannot serve SQL - during a // scale-up it may not exist yet. The hook still succeeds on the hosts it could reach. diff --git a/pkg/model/chi/schemer/schemer.go b/pkg/model/chi/schemer/schemer.go index 8e5273714..e7b0cadc0 100644 --- a/pkg/model/chi/schemer/schemer.go +++ b/pkg/model/chi/schemer/schemer.go @@ -16,6 +16,8 @@ package schemer import ( "context" + "errors" + "fmt" "time" log "github.com/altinity/clickhouse-operator/pkg/announcer" @@ -34,6 +36,14 @@ type ClusterSchemer struct { version *swversion.SoftWareVersion } +type replicatedTable struct { + DatabaseName string + TableName string +} + +// ErrGateDeadline marks the shared sync-gate deadline being reached. +var ErrGateDeadline = errors.New("sync gate deadline exceeded") + // NewClusterSchemer creates new Schemer object func NewClusterSchemer(clusterConnectionParams *clickhouse.ClusterConnectionParams, version *swversion.SoftWareVersion) *ClusterSchemer { return &ClusterSchemer{ @@ -180,7 +190,107 @@ func (s *ClusterSchemer) HostClickHouseVersion(ctx context.Context, host *api.Ho // HostMaxReplicaDelay returns max replica delay on the host func (s *ClusterSchemer) HostMaxReplicaDelay(ctx context.Context, host *api.Host) (int, error) { - return s.QueryHostInt(ctx, host, s.sqlMaxReplicaDelay()) + replicaDelay, err := s.QueryHostInt(ctx, host, s.sqlMaxReplicaDelay()) + if contextError := ctx.Err(); contextError != nil { + return 0, contextError + } + return replicaDelay, err +} + +func (s *ClusterSchemer) HostMaxIsReadonly(ctx context.Context, host *api.Host) (int, error) { + readonly, err := s.QueryHostInt(ctx, host, s.sqlReplicaHealth("is_readonly")) + if contextError := ctx.Err(); contextError != nil { + return 0, contextError + } + return readonly, err +} + +func (s *ClusterSchemer) HostMaxIsSessionExpired(ctx context.Context, host *api.Host) (int, error) { + sessionExpired, err := s.QueryHostInt(ctx, host, s.sqlReplicaHealth("is_session_expired")) + if contextError := ctx.Err(); contextError != nil { + return 0, contextError + } + return sessionExpired, err +} + +func (s *ClusterSchemer) PeerReplicatedObjectCount(ctx context.Context, host *api.Host, deadline time.Time) (int, error) { + databaseNames, replicatedTables, err := s.peerReplicatedObjects(ctx, host, deadline) + if err != nil { + return 0, err + } + return len(databaseNames) + len(replicatedTables), nil +} + +func (s *ClusterSchemer) HostAsyncLoadBarrier(ctx context.Context, host *api.Host, deadline time.Time) error { + for { + asyncLoaderExists, err := s.queryHostIntWithDeadline(ctx, host, deadline, s.sqlAsyncLoaderTableExists()) + if err != nil { + return err + } + if asyncLoaderExists == 0 { + return nil + } + + pendingLoadJobs, failedLoadJobs, err := s.queryHostIntPairWithDeadline(ctx, host, deadline, s.sqlAsyncLoaderState()) + if err != nil { + return err + } + if failedLoadJobs > 0 { + failedLoadJob, detailErr := s.queryHostStringWithDeadline(ctx, host, deadline, s.sqlAsyncLoaderFailedDetails()) + if detailErr != nil { + return detailErr + } + return fmt.Errorf("async loader failed or canceled job: %s", failedLoadJob) + } + if pendingLoadJobs == 0 { + return nil + } + if err := waitForNextGatePoll(ctx, deadline); err != nil { + return err + } + } +} + +func (s *ClusterSchemer) HostSyncReplicatedObjects(ctx context.Context, host *api.Host, deadline time.Time) error { + // LIGHTWEIGHT is available since 23.4 only. When the version is unknown (digest-pinned + // or non-numeric image tag) or older, fall back to plain SYSTEM SYNC REPLICA rather than + // failing the reconcile - the gate must never be harder to pass than the plain wait it replaces. + lightweight := s.version.Matches(">= 23.4") + if !lightweight { + log.V(1).M(host).F().Info("SYSTEM SYNC REPLICA LIGHTWEIGHT is unavailable for version %s - falling back to full SYNC REPLICA", s.version) + } + + if err := s.HostAsyncLoadBarrier(ctx, host, deadline); err != nil { + return err + } + + databaseNames, _, err := s.peerReplicatedObjects(ctx, host, deadline) + if err != nil { + return err + } + for _, databaseName := range databaseNames { + if err := s.execHostWithDeadline(ctx, host, deadline, s.sqlSyncDatabaseReplica(databaseName)); err != nil { + return err + } + } + + if err := s.HostAsyncLoadBarrier(ctx, host, deadline); err != nil { + return err + } + + _, replicatedTables, err := s.peerReplicatedObjects(ctx, host, deadline) + if err != nil { + return err + } + for _, replicatedTable := range replicatedTables { + if err := s.execHostWithDeadline(ctx, host, deadline, s.sqlWaitLoadingParts(replicatedTable.DatabaseName, replicatedTable.TableName)); err != nil { + return err + } + if err := s.execHostWithDeadline(ctx, host, deadline, s.sqlSyncReplica(replicatedTable.DatabaseName, replicatedTable.TableName, lightweight)); err != nil { + return err + } + } + return nil } // HostShutdown shutdown a host @@ -204,3 +314,212 @@ func debugCreateSQLs(names, sqls []string, err error) ([]string, []string) { } return names, sqls } + +func (s *ClusterSchemer) peerReplicatedObjects(ctx context.Context, host *api.Host, deadline time.Time) ([]string, []replicatedTable, error) { + if _, err := gateRemaining(ctx, deadline); err != nil { + return nil, nil, err + } + + // Replication is a per-shard property - discover replicated objects from the shard peers only. + // A cluster-wide scan would drag tables that live on other shards into this host's catch-up. + peers := s.Names(interfaces.NameFQDNs, host, api.ChiShard{}, true) + if len(peers) == 0 { + return nil, nil, nil + } + + queryCtx, cancel, err := gateQueryContext(ctx, deadline) + if err != nil { + return nil, nil, err + } + defer cancel() + + queryResult, err := s.Cluster.SetHosts(peers).QueryAny(queryCtx, s.sqlReplicatedObjects()) + if mappedErr := gateQueryError(ctx, queryCtx, err); mappedErr != nil { + return nil, nil, mappedErr + } + if queryResult == nil { + return nil, nil, fmt.Errorf("empty replicated object discovery result from peers %v", peers) + } + defer queryResult.Close() + + databaseNames := make([]string, 0) + replicatedTables := make([]replicatedTable, 0) + for queryResult.Rows.Next() { + var objectType string + var databaseName string + var tableName string + if err := queryResult.Rows.Scan(&objectType, &databaseName, &tableName); err != nil { + if mappedErr := gateQueryError(ctx, queryCtx, err); mappedErr != nil { + return nil, nil, mappedErr + } + return nil, nil, err + } + switch objectType { + case "database": + databaseNames = append(databaseNames, databaseName) + case "table": + replicatedTables = append(replicatedTables, replicatedTable{ + DatabaseName: databaseName, + TableName: tableName, + }) + default: + return nil, nil, fmt.Errorf("unknown replicated object type %q", objectType) + } + } + if err := queryResult.Rows.Err(); err != nil { + if mappedErr := gateQueryError(ctx, queryCtx, err); mappedErr != nil { + return nil, nil, mappedErr + } + return nil, nil, err + } + if mappedErr := gateQueryError(ctx, queryCtx, nil); mappedErr != nil { + return nil, nil, mappedErr + } + return databaseNames, replicatedTables, nil +} + +func (s *ClusterSchemer) execHostWithDeadline(ctx context.Context, host *api.Host, deadline time.Time, querySQL string) error { + remaining, err := gateRemaining(ctx, deadline) + if err != nil { + return err + } + + opts := clickhouse.NewQueryOptions() + opts.SetRetry(false) + opts.SetQueryTimeout(remaining) + + err = s.ExecHost(ctx, host, []string{querySQL}, opts) + if contextError := ctx.Err(); contextError != nil { + return contextError + } + if errors.Is(err, context.DeadlineExceeded) { + return ErrGateDeadline + } + return err +} + +func (s *ClusterSchemer) queryHostIntWithDeadline(ctx context.Context, host *api.Host, deadline time.Time, querySQL string) (int, error) { + queryCtx, cancel, err := gateQueryContext(ctx, deadline) + if err != nil { + return 0, err + } + defer cancel() + + queryValue, err := s.QueryHostInt(queryCtx, host, querySQL) + if mappedErr := gateQueryError(ctx, queryCtx, err); mappedErr != nil { + return 0, mappedErr + } + return queryValue, nil +} + +func (s *ClusterSchemer) queryHostStringWithDeadline(ctx context.Context, host *api.Host, deadline time.Time, querySQL string) (string, error) { + queryCtx, cancel, err := gateQueryContext(ctx, deadline) + if err != nil { + return "", err + } + defer cancel() + + queryValue, err := s.QueryHostString(queryCtx, host, querySQL) + if mappedErr := gateQueryError(ctx, queryCtx, err); mappedErr != nil { + return "", mappedErr + } + return queryValue, nil +} + +func (s *ClusterSchemer) queryHostIntPairWithDeadline(ctx context.Context, host *api.Host, deadline time.Time, querySQL string) (int, int, error) { + queryCtx, cancel, err := gateQueryContext(ctx, deadline) + if err != nil { + return 0, 0, err + } + defer cancel() + + queryResult, err := s.QueryHost(queryCtx, host, querySQL) + if mappedErr := gateQueryError(ctx, queryCtx, err); mappedErr != nil { + return 0, 0, mappedErr + } + if queryResult == nil { + return 0, 0, fmt.Errorf("empty query result") + } + defer queryResult.Close() + + if !queryResult.Rows.Next() { + if err := queryResult.Rows.Err(); err != nil { + if mappedErr := gateQueryError(ctx, queryCtx, err); mappedErr != nil { + return 0, 0, mappedErr + } + return 0, 0, err + } + return 0, 0, fmt.Errorf("found no rows") + } + + var firstValue int + var secondValue int + if err := queryResult.Rows.Scan(&firstValue, &secondValue); err != nil { + if mappedErr := gateQueryError(ctx, queryCtx, err); mappedErr != nil { + return 0, 0, mappedErr + } + return 0, 0, err + } + if err := queryResult.Rows.Err(); err != nil { + if mappedErr := gateQueryError(ctx, queryCtx, err); mappedErr != nil { + return 0, 0, mappedErr + } + return 0, 0, err + } + if mappedErr := gateQueryError(ctx, queryCtx, nil); mappedErr != nil { + return 0, 0, mappedErr + } + return firstValue, secondValue, nil +} + +func gateQueryContext(ctx context.Context, deadline time.Time) (context.Context, context.CancelFunc, error) { + remaining, err := gateRemaining(ctx, deadline) + if err != nil { + return nil, nil, err + } + queryCtx, cancel := context.WithTimeout(ctx, remaining) + return queryCtx, cancel, nil +} + +func gateRemaining(ctx context.Context, deadline time.Time) (time.Duration, error) { + if contextError := ctx.Err(); contextError != nil { + return 0, contextError + } + remaining := time.Until(deadline) + if remaining <= 0 { + return 0, ErrGateDeadline + } + return remaining, nil +} + +func gateQueryError(parentCtx, queryCtx context.Context, err error) error { + if contextError := parentCtx.Err(); contextError != nil { + return contextError + } + if errors.Is(err, context.DeadlineExceeded) || errors.Is(queryCtx.Err(), context.DeadlineExceeded) { + return ErrGateDeadline + } + if contextError := queryCtx.Err(); contextError != nil { + return contextError + } + return err +} + +func waitForNextGatePoll(ctx context.Context, deadline time.Time) error { + remaining, err := gateRemaining(ctx, deadline) + if err != nil { + return err + } + sleepDuration := time.Second + if remaining < sleepDuration { + sleepDuration = remaining + } + timer := time.NewTimer(sleepDuration) + defer timer.Stop() + select { + case <-ctx.Done(): + return ctx.Err() + case <-timer.C: + return nil + } +} diff --git a/pkg/model/chi/schemer/sql.go b/pkg/model/chi/schemer/sql.go index b8182c46c..fd4c25499 100644 --- a/pkg/model/chi/schemer/sql.go +++ b/pkg/model/chi/schemer/sql.go @@ -17,6 +17,7 @@ package schemer import ( "context" "fmt" + "strings" "github.com/MakeNowJust/heredoc" @@ -91,6 +92,91 @@ func (s *ClusterSchemer) sqlSyncTable(ctx context.Context, host *api.Host) ([]st return names, sqlStatements, nil } +func (s *ClusterSchemer) sqlReplicaHealth(column string) string { + return fmt.Sprintf("SELECT coalesce(max(%s),0) FROM system.replicas", column) +} + +// sqlSyncReplica waits for the local replica to fetch the replication log of the specified table. +// LIGHTWEIGHT (23.4+) waits for metadata/entry fetch only, full sync waits for parts as well. +// SYSTEM statements accept no SETTINGS clause - the wait is bounded by the query context deadline. +func (s *ClusterSchemer) sqlSyncReplica(databaseName, tableName string, lightweight bool) string { + sql := fmt.Sprintf(`SYSTEM SYNC REPLICA "%s"."%s"`, quoteIdent(databaseName), quoteIdent(tableName)) + if lightweight { + sql += " LIGHTWEIGHT" + } + return sql +} + +func (s *ClusterSchemer) sqlSyncDatabaseReplica(databaseName string) string { + return fmt.Sprintf(`SYSTEM SYNC DATABASE REPLICA "%s"`, quoteIdent(databaseName)) +} + +func (s *ClusterSchemer) sqlWaitLoadingParts(databaseName, tableName string) string { + return fmt.Sprintf(`SYSTEM WAIT LOADING PARTS "%s"."%s"`, quoteIdent(databaseName), quoteIdent(tableName)) +} + +func (s *ClusterSchemer) sqlAsyncLoaderTableExists() string { + return "SELECT count() FROM system.tables WHERE database='system' AND name='asynchronous_loader'" +} + +func (s *ClusterSchemer) sqlAsyncLoaderState() string { + return heredoc.Doc(` + SELECT + countIf(status = 'PENDING' OR is_executing = 1 OR is_ready = 1 OR is_blocked = 1), + countIf(status IN ('FAILED', 'CANCELED')) + FROM + system.asynchronous_loader + WHERE + startsWith(job, 'startup ') AND + (position(job, ' database ') > 0 OR position(job, ' table ') > 0) + `) +} + +func (s *ClusterSchemer) sqlAsyncLoaderFailedDetails() string { + return heredoc.Doc(` + SELECT + concat(job, ': ', status, ifNull(concat(': ', exception), '')) + FROM + system.asynchronous_loader + WHERE + startsWith(job, 'startup ') AND + (position(job, ' database ') > 0 OR position(job, ' table ') > 0) AND + status IN ('FAILED', 'CANCELED') + LIMIT 1 + `) +} + +func (s *ClusterSchemer) sqlReplicatedObjects() string { + // Runs on a shard peer, against its LOCAL system tables. Replication is per-shard, so the set of + // objects this host has to catch up on is exactly the set its shard peer already serves. + return heredoc.Docf(` + SELECT + 'database' AS object_type, + name AS database, + '' AS table_name + FROM system.databases + WHERE + name NOT IN (%s) AND + engine = 'Replicated' + UNION ALL + SELECT + 'table' AS object_type, + database, + name AS table_name + FROM system.tables + WHERE + database NOT IN (%s) AND + engine LIKE 'Replicated%%' + `, + ignoredDBs, + ignoredDBs, + ) +} + +func quoteIdent(identifier string) string { + return strings.ReplaceAll(identifier, `"`, `""`) +} + func (s *ClusterSchemer) sqlCreateDatabaseDistributed(cluster string) string { var createDatabaseStmt string switch { diff --git a/pkg/model/chi/schemer/sql_sync_test.go b/pkg/model/chi/schemer/sql_sync_test.go new file mode 100644 index 000000000..f5b6e32e2 --- /dev/null +++ b/pkg/model/chi/schemer/sql_sync_test.go @@ -0,0 +1,132 @@ +package schemer + +import ( + "context" + "errors" + "strings" + "testing" + "time" + + api "github.com/altinity/clickhouse-operator/pkg/apis/clickhouse.altinity.com/v1" + "github.com/altinity/clickhouse-operator/pkg/apis/swversion" +) + +func TestQuoteIdentDoublesQuotes(t *testing.T) { + if got := quoteIdent(`my"db`); got != `my""db` { + t.Fatalf("quoteIdent must double embedded quotes; got %q", got) + } +} + +func TestSQLReplicaHealthShape(t *testing.T) { + schemer := &ClusterSchemer{} + sql := schemer.sqlReplicaHealth("is_readonly") + if !strings.Contains(sql, "coalesce(max(is_readonly),0)") || !strings.Contains(sql, "system.replicas") { + t.Fatalf("health SQL wrong: %s", sql) + } +} + +func TestHostMaxReplicaDelayReturnsCanceledContext(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + cancel() + + delay, err := (&ClusterSchemer{}).HostMaxReplicaDelay(ctx, &api.Host{}) + if delay != 0 || !errors.Is(err, context.Canceled) { + t.Fatalf("canceled context must be returned; delay=%d err=%v", delay, err) + } +} + +func TestSQLSyncReplicaLightweight(t *testing.T) { + schemer := &ClusterSchemer{} + sql := schemer.sqlSyncReplica(`my"db`, "tbl", true) + if !strings.HasSuffix(sql, "LIGHTWEIGHT") { + t.Fatalf("table sync must end with LIGHTWEIGHT: %s", sql) + } + if !strings.Contains(sql, `"my""db"."tbl"`) { + t.Fatalf("identifiers must be quoted and escaped: %s", sql) + } +} + +func TestSQLSyncDatabaseReplicaHasNoLightweight(t *testing.T) { + schemer := &ClusterSchemer{} + sql := schemer.sqlSyncDatabaseReplica("db") + if strings.Contains(sql, "LIGHTWEIGHT") { + t.Fatalf("DATABASE REPLICA takes no LIGHTWEIGHT modifier: %s", sql) + } + if !strings.Contains(sql, "SYSTEM SYNC DATABASE REPLICA") || !strings.Contains(sql, `"db"`) { + t.Fatalf("wrong DB-sync stmt: %s", sql) + } +} + +func TestSQLWaitLoadingPartsShape(t *testing.T) { + schemer := &ClusterSchemer{} + sql := schemer.sqlWaitLoadingParts("db", "tbl") + if !strings.Contains(sql, "SYSTEM WAIT LOADING PARTS") || !strings.Contains(sql, `"db"."tbl"`) { + t.Fatalf("wrong wait-loading-parts stmt: %s", sql) + } +} + +// SYSTEM statements have no SETTINGS production - appending one is a parse-time SYNTAX_ERROR (Code 62). +func TestSQLSyncStatementsCarryNoSettingsClause(t *testing.T) { + schemer := &ClusterSchemer{} + for _, sql := range []string{ + schemer.sqlSyncReplica("db", "tbl", true), + schemer.sqlSyncReplica("db", "tbl", false), + schemer.sqlSyncDatabaseReplica("db"), + schemer.sqlWaitLoadingParts("db", "tbl"), + } { + if strings.Contains(sql, "SETTINGS") { + t.Fatalf("SYSTEM statement must carry no SETTINGS clause: %s", sql) + } + } +} + +func TestSQLSyncReplicaLightweightToggle(t *testing.T) { + schemer := &ClusterSchemer{} + if !strings.HasSuffix(schemer.sqlSyncReplica("db", "tbl", true), "LIGHTWEIGHT") { + t.Fatalf("lightweight variant must end with LIGHTWEIGHT") + } + if strings.Contains(schemer.sqlSyncReplica("db", "tbl", false), "LIGHTWEIGHT") { + t.Fatalf("fallback variant must not use LIGHTWEIGHT") + } +} + +func TestSQLAsyncLoaderStateShape(t *testing.T) { + schemer := &ClusterSchemer{} + sql := schemer.sqlAsyncLoaderState() + if !strings.Contains(sql, "countIf(status = 'PENDING'") || !strings.Contains(sql, "status IN ('FAILED', 'CANCELED')") { + t.Fatalf("async loader state SQL must count pending and failed jobs: %s", sql) + } + if !strings.Contains(sql, "startsWith(job, 'startup ')") || !strings.Contains(sql, " database ") { + t.Fatalf("async loader state SQL must filter relevant startup load jobs: %s", sql) + } +} + +// An unknown or pre-23.4 version must NOT fail the gate - it falls back to full SYNC REPLICA. +func TestHostSyncReplicatedObjectsFailsOpenOnOldVersion(t *testing.T) { + for _, version := range []string{"23.3.22", "0.0.1"} { + schemer := &ClusterSchemer{version: swversion.NewSoftWareVersion(version)} + err := schemer.HostSyncReplicatedObjects(context.Background(), &api.Host{}, time.Now().Add(-time.Second)) + // The version decision is taken before the async-load barrier, so an expired deadline + // proves the gate got past it: a hard-fail would surface the version error here instead + // of ErrGateDeadline. + if !errors.Is(err, ErrGateDeadline) { + t.Fatalf("version %s must not hard-fail the gate, want ErrGateDeadline, got %v", version, err) + } + } +} + +func TestHostAsyncLoadBarrierReturnsGateDeadlineWhenExpired(t *testing.T) { + schemer := &ClusterSchemer{} + err := schemer.HostAsyncLoadBarrier(context.Background(), &api.Host{}, time.Now().Add(-time.Second)) + if !errors.Is(err, ErrGateDeadline) { + t.Fatalf("expected ErrGateDeadline, got %v", err) + } +} + +func TestPeerReplicatedObjectCountReturnsGateDeadlineWhenExpired(t *testing.T) { + schemer := &ClusterSchemer{} + _, err := schemer.PeerReplicatedObjectCount(context.Background(), &api.Host{}, time.Now().Add(-time.Second)) + if !errors.Is(err, ErrGateDeadline) { + t.Fatalf("expected ErrGateDeadline, got %v", err) + } +} diff --git a/pkg/model/registry_test.go b/pkg/model/registry_test.go index e5a7a7317..91afb1e29 100644 --- a/pkg/model/registry_test.go +++ b/pkg/model/registry_test.go @@ -1,5 +1,3 @@ -//go:build race - package model import ( @@ -70,18 +68,18 @@ func Test_Registry_BasicOperations_ConcurrencyTest(t *testing.T) { go func() { startWg.Done() startWg.Wait() // Block until the other goroutine has begun execution - reg.RegisterConfigMap(testCmA) - reg.RegisterPVC(testPvcA) + reg.RegisterConfigMap(&testCmA) + reg.RegisterPVC(&testPvcA) doneWg.Done() }() go func() { startWg.Done() startWg.Wait() // Block until the other goroutine has begun execution - reg.RegisterConfigMap(testCmA) - reg.RegisterConfigMap(testCmAOtherNamespace) - reg.RegisterConfigMap(testCmB) - reg.RegisterPVC(testPvcB) + reg.RegisterConfigMap(&testCmA) + reg.RegisterConfigMap(&testCmAOtherNamespace) + reg.RegisterConfigMap(&testCmB) + reg.RegisterPVC(&testPvcB) doneWg.Done() }() @@ -101,7 +99,7 @@ func Test_Registry_BasicOperations_ConcurrencyTest(t *testing.T) { &testPvcA: PVC, &testPvcB: PVC, } { - if got := reg.hasEntity(expectedEntityType, *expectedMetaObj); !got { + if got := reg.hasEntity(expectedEntityType, expectedMetaObj); !got { t.Errorf( "Expected registry to contain entity type %s:{Namespace = %s, Name = %s}", expectedEntityType, @@ -117,20 +115,20 @@ func Test_Registry_BasicOperations_ConcurrencyTest(t *testing.T) { go func() { startWg.Done() - startWg.Wait() // Block until the other goroutine has begun execution - reg.RegisterPVC(testPvcD) // Add a net-new PVC (both goroutines) - reg.deleteEntity(ConfigMap, testCmAOtherNamespace) // Delete testCmAOtherNamespace (only this goroutine) - reg.deleteEntity(ConfigMap, testCmB) // Delete testCmB (both goroutines) + startWg.Wait() // Block until the other goroutine has begun execution + reg.RegisterPVC(&testPvcD) // Add a net-new PVC (both goroutines) + reg.deleteEntity(ConfigMap, &testCmAOtherNamespace) // Delete testCmAOtherNamespace (only this goroutine) + reg.deleteEntity(ConfigMap, &testCmB) // Delete testCmB (both goroutines) doneWg.Done() }() go func() { startWg.Done() - startWg.Wait() // Block until the other goroutine has begun execution - reg.RegisterPVC(testPvcC) // Add a net-new PVC (only this goroutine) - reg.RegisterPVC(testPvcD) // Add a net-new PVC (both goroutines) - reg.deleteEntity(ConfigMap, testCmB) // Delete testCmB (both goroutines) - reg.deleteEntity(PVC, testPvcB) // Delete testPvcB (only this goroutine) + startWg.Wait() // Block until the other goroutine has begun execution + reg.RegisterPVC(&testPvcC) // Add a net-new PVC (only this goroutine) + reg.RegisterPVC(&testPvcD) // Add a net-new PVC (both goroutines) + reg.deleteEntity(ConfigMap, &testCmB) // Delete testCmB (both goroutines) + reg.deleteEntity(PVC, &testPvcB) // Delete testPvcB (only this goroutine) doneWg.Done() }() @@ -152,7 +150,7 @@ func Test_Registry_BasicOperations_ConcurrencyTest(t *testing.T) { &testPvcC: PVC, // We added testPvcC (one of the goroutines) &testPvcD: PVC, // We added testPvcD (both goroutines tried) } { - if got := reg.hasEntity(expectedEntityType, *expectedMetaObj); !got { + if got := reg.hasEntity(expectedEntityType, expectedMetaObj); !got { t.Errorf( "Expected registry to contain entity type %s:{Namespace = %s, Name = %s}", expectedEntityType, @@ -171,7 +169,7 @@ func Test_Registry_BasicOperations_ConcurrencyTest(t *testing.T) { go func() { startWg.Done() startWg.Wait() // Block until the other goroutine has begun execution - reg.Walk(func(entityType EntityType, meta v1.ObjectMeta) { + reg.Walk(func(entityType EntityType, meta v1.Object) { threadAObjsSeen++ }) doneWg.Done() @@ -181,7 +179,7 @@ func Test_Registry_BasicOperations_ConcurrencyTest(t *testing.T) { go func() { startWg.Done() startWg.Wait() // Block until the other goroutine has begun execution - reg.Walk(func(entityType EntityType, meta v1.ObjectMeta) { + reg.Walk(func(entityType EntityType, meta v1.Object) { threadBObjsSeen++ }) doneWg.Done() @@ -199,3 +197,72 @@ func Test_Registry_BasicOperations_ConcurrencyTest(t *testing.T) { ) } } + +// Test_Registry_ConcurrentReadersAndWriters runs readers concurrently with writers. +// The test above only ever pairs writers with writers, and its Walk phase runs after all +// mutation has finished, so a missing *read* lock is invisible to it: stripping the +// RLock/RUnlock pairs out of objectMetaSet leaves it green. This one reports a data race +// for that same edit, which is the more likely regression of the two. +func Test_Registry_ConcurrentReadersAndWriters(t *testing.T) { + const iterations = 200 + + reg := NewRegistry() + // Pre-register so the readers exercise populated entity types, not only the + // create-on-miss path, and so there is a stable entry to assert on at the end. + reg.RegisterConfigMap(&testCmA) + reg.RegisterPVC(&testPvcA) + + churn := func() { + for i := 0; i < iterations; i++ { + reg.RegisterConfigMap(&testCmB) + reg.RegisterPVC(&testPvcB) + reg.deleteEntity(ConfigMap, &testCmB) + reg.deleteEntity(PVC, &testPvcB) + } + } + + // Point lookups on exactly the keys the writers add and remove. + pointRead := func() { + for i := 0; i < iterations; i++ { + reg.hasEntity(ConfigMap, &testCmB) + reg.hasEntity(PVC, &testPvcB) + } + } + + // Whole-map iteration during mutation, dereferencing what it yields - a bare count + // would never touch the shared objects the registry hands out. + iterate := func() { + for i := 0; i < iterations; i++ { + reg.Walk(func(entityType EntityType, meta v1.Object) { + _ = meta.GetName() + _ = meta.GetNamespace() + _ = len(meta.GetLabels()) + }) + _ = reg.Len(ConfigMap) + } + } + + workers := []func(){churn, churn, pointRead, pointRead, iterate, iterate} + + startWg := sync.WaitGroup{} + doneWg := sync.WaitGroup{} + startWg.Add(len(workers)) + doneWg.Add(len(workers)) + for _, worker := range workers { + go func(run func()) { + startWg.Done() + startWg.Wait() // Block until every goroutine has begun execution + run() + doneWg.Done() + }(worker) + } + doneWg.Wait() + + // The pre-registered entries are never deleted, so they must survive the churn. + if !reg.hasEntity(ConfigMap, &testCmA) { + t.Errorf("expected %s to survive concurrent churn", testCmA.Name) + } + if !reg.hasEntity(PVC, &testPvcA) { + t.Errorf("expected %s to survive concurrent churn", testPvcA.Name) + } +} diff --git a/tests/e2e/manifests/chi/test-079-sync-gate-1.yaml b/tests/e2e/manifests/chi/test-079-sync-gate-1.yaml new file mode 100644 index 000000000..1fa76d8bd --- /dev/null +++ b/tests/e2e/manifests/chi/test-079-sync-gate-1.yaml @@ -0,0 +1,17 @@ +apiVersion: "clickhouse.altinity.com/v1" +kind: "ClickHouseInstallation" +metadata: + name: "test-079-sync-gate" +spec: + useTemplates: + - name: clickhouse-version + configuration: + zookeeper: + nodes: + - host: zookeeper + port: 2181 + clusters: + - name: "default" + layout: + shardsCount: 1 + replicasCount: 1 diff --git a/tests/e2e/manifests/chi/test-079-sync-gate-2.yaml b/tests/e2e/manifests/chi/test-079-sync-gate-2.yaml new file mode 100644 index 000000000..e452c1b2a --- /dev/null +++ b/tests/e2e/manifests/chi/test-079-sync-gate-2.yaml @@ -0,0 +1,17 @@ +apiVersion: "clickhouse.altinity.com/v1" +kind: "ClickHouseInstallation" +metadata: + name: "test-079-sync-gate" +spec: + useTemplates: + - name: clickhouse-version + configuration: + zookeeper: + nodes: + - host: zookeeper + port: 2181 + clusters: + - name: "default" + layout: + shardsCount: 1 + replicasCount: 3 diff --git a/tests/e2e/manifests/chopconf/test-079-sync-gate-off.yaml b/tests/e2e/manifests/chopconf/test-079-sync-gate-off.yaml new file mode 100644 index 000000000..f6736b172 --- /dev/null +++ b/tests/e2e/manifests/chopconf/test-079-sync-gate-off.yaml @@ -0,0 +1,13 @@ +apiVersion: "clickhouse.altinity.com/v1" +kind: "ClickHouseOperatorConfiguration" +metadata: + name: "sync-gate-off" +spec: + reconcile: + host: + wait: + replicas: + all: "false" + new: "false" + catchUp: + enabled: "false" diff --git a/tests/e2e/manifests/chopconf/test-079-sync-gate.yaml b/tests/e2e/manifests/chopconf/test-079-sync-gate.yaml new file mode 100644 index 000000000..8aa5956d9 --- /dev/null +++ b/tests/e2e/manifests/chopconf/test-079-sync-gate.yaml @@ -0,0 +1,20 @@ +apiVersion: "clickhouse.altinity.com/v1" +kind: "ClickHouseOperatorConfiguration" +metadata: + name: "sync-gate" +spec: + reconcile: + host: + wait: + replicas: + catchUp: + enabled: "true" + # Matches the shipped default. The scenario holds the gate open across its own setup - + # table poll, a 30s settle, then three negative probes - which is minutes, so a 120s + # budget would expire mid-test and abort the reconcile before the test resumes + # replicated sends, leaving the gate's release half unreachable. + timeout: 900 + onTimeout: "abort" + health: + pollInterval: 5 + successThreshold: 3 diff --git a/tests/e2e/test_operator.py b/tests/e2e/test_operator.py index ca305eca2..43e620d3a 100644 --- a/tests/e2e/test_operator.py +++ b/tests/e2e/test_operator.py @@ -5841,7 +5841,7 @@ def test_010056(self): assert out != "0" with And("Replica still should be unready after reconcile timeout"): - ready = kubectl.get_field("pod", f"chi-{chi}-{cluster}-0-1-0", ".metadata.labels.clickhouse\.altinity\.com\/ready") + ready = kubectl.get_field("pod", f"chi-{chi}-{cluster}-0-1-0", r".metadata.labels.clickhouse\.altinity\.com\/ready") print(f"ready label={ready}") assert ready != "yes", error("Replica should be unready") @@ -5867,7 +5867,7 @@ def test_010056(self): with Then("Replica should become ready"): kubectl.wait_field("pod", f"chi-{chi}-{cluster}-0-1-0", - ".metadata.labels.clickhouse\.altinity\.com\/ready", value="yes") + r".metadata.labels.clickhouse\.altinity\.com\/ready", value="yes") with And("Replication delay should be zero"): out = clickhouse.query(chi, "select max(absolute_delay) from system.replicas", host=f"chi-{chi}-{cluster}-0-1-0") @@ -7240,6 +7240,243 @@ def test_010072(self): with Finally("I clean up"): delete_test_namespace() + +@TestScenario +@Name("test_010079. Test replicated host catch-up gate") +def test_010079(self): + create_shell_namespace_clickhouse_template() + + with Given("I enable replicated host catch-up gate"): + util.apply_operator_config("manifests/chopconf/test-079-sync-gate.yaml") + + with And("The chopconf CR retains the catchUp block (CRD schema must not prune it)"): + # A ClickHouseOperatorConfiguration CRD without the catchUp sub-schema silently drops + # `catchUp:` on apply, and the gate would then be OFF while the test claims it is ON. + applied = kubectl.get("chopconf", "sync-gate", ns=current().context.operator_namespace) + applied_catch_up = applied["spec"]["reconcile"]["host"]["wait"]["replicas"].get("catchUp") + assert applied_catch_up is not None, error("chopconf CRD pruned the catchUp block") + assert applied_catch_up["enabled"] == "true", error(f"catch-up gate is not enabled: {applied_catch_up}") + + util.require_keeper(keeper_type=self.context.keeper_type) + + manifest = "manifests/chi/test-079-sync-gate-1.yaml" + chi = yaml_manifest.get_name(util.get_full_path(manifest)) + cluster = "default" + source_host = f"chi-{chi}-{cluster}-0-0-0" + delayed_replica_host = f"chi-{chi}-{cluster}-0-1-0" + next_replica_host = f"chi-{chi}-{cluster}-0-2-0" + delayed_replica_fqdn = f"chi-{chi}-{cluster}-0-1.{current().context.test_namespace}.svc.cluster.local" + + def get_replica_caught_up_hosts(): + chi_status = kubectl.get("chi", chi).get("status") or {} + return chi_status.get("hostsWithReplicaCaughtUp") or [] + + def wait_table_exists_on_delayed_replica(): + table_exists = "0" + for attempt_index in range(1, 11): + table_exists = clickhouse.query_with_error( + chi, + "select count() from system.tables where name='test_079'", + host=delayed_replica_host, + ) + if table_exists == "1": + break + retry_sleep(attempt_index, 10, "Table is not ready on delayed replica") + assert table_exists == "1", error("Table was not created on a new replica") + + def wait_replica_caught_up_marker(): + caught_up_hosts = [] + for attempt_index in range(1, 25): + caught_up_hosts = get_replica_caught_up_hosts() + if delayed_replica_fqdn in caught_up_hosts: + break + retry_sleep(attempt_index, 5, "Replica caught-up marker is not ready") + assert delayed_replica_fqdn in caught_up_hosts, error("Replica caught-up marker was not written") + + def wait_delayed_replica_row_count(expected_count): + row_count = "" + for attempt_index in range(1, 13): + row_count = clickhouse.query(chi, "select count() from test_079", host=delayed_replica_host) + if row_count == expected_count: + break + retry_sleep(attempt_index, 5, "Table data is not yet replicated") + assert row_count == expected_count, error("Table data has not been replicated") + + with Given("CHI is installed"): + kubectl.create_and_check( + manifest=manifest, + check={ + "pod_count": 1, + "apply_templates": { + current().context.clickhouse_template, + }, + "do_not_delete": 1, + }, + ) + + with Then("Create a replicated table"): + clickhouse.query( + chi, + "CREATE TABLE test_079 (a Int64) Engine = ReplicatedMergeTree('/clickhouse/tables/{database}/{table}', '{replica}') ORDER BY a PARTITION BY a", + ) + clickhouse.query(chi, "INSERT INTO test_079 SELECT 1") + + with And("STOP REPLICATED SENDS"): + clickhouse.query(chi, "SYSTEM STOP REPLICATED SENDS", host=source_host) + + with When("Scale to three replicas while the new replica is delayed"): + kubectl.create_and_check( + manifest="manifests/chi/test-079-sync-gate-2.yaml", + check={ + "do_not_delete": 1, + "pod_count": 2, + "chi_status": "InProgress", + }, + ) + + with Then("Table should be created on the delayed replica"): + wait_table_exists_on_delayed_replica() + + with And("Table should have no data replicated"): + query_result = clickhouse.query(chi, "select count() from test_079", host=delayed_replica_host) + assert query_result == "0", error("Table data has been replicated") + + with And("Replication delay should be non-zero"): + replica_delay = clickhouse.query( + chi, + "select max(absolute_delay) from system.replicas", + host=delayed_replica_host, + ) + print(f"max(absolute_delay)={replica_delay}") + assert replica_delay != "0" + + with And("Wait for the catch-up gate to observe the delayed replica"): + time.sleep(30) + + with And("Delayed replica should not have a caught-up marker"): + caught_up_hosts = get_replica_caught_up_hosts() + print(yaml.safe_dump(caught_up_hosts)) + assert delayed_replica_fqdn not in caught_up_hosts + + with And("Next replica should not be created while the gate waits"): + pod_count = kubectl.get_count("pod", chi=chi) + assert pod_count == 2, error(f"Expected 2 pods while gate waits, got {pod_count}") + next_replica_pod = kubectl.get("pod", next_replica_host, ok_to_fail=True) + assert next_replica_pod is None, error("Next replica should not be created before sync completes") + + with And("Delayed replica should still be unready"): + ready_label = kubectl.get_field( + "pod", + delayed_replica_host, + r".metadata.labels.clickhouse\.altinity\.com\/ready", + ) + print(f"ready label={ready_label}") + assert ready_label != "yes", error("Delayed replica should be unready") + + with When("START REPLICATED SENDS"): + clickhouse.query(chi, "SYSTEM START REPLICATED SENDS", host=source_host) + + # Not And(): clickhouse.query() above opens no TestFlows step, so an And() here would be + # the block's first child and would have no sibling to inherit its subtype from. + with When("Live inserts continue after sync starts"): + clickhouse.query(chi, "INSERT INTO test_079 SELECT number + 2 FROM numbers(5)", host=source_host) + + with Then("Delayed replica should receive a caught-up marker"): + wait_replica_caught_up_marker() + + with And("Delayed replica should become ready"): + kubectl.wait_field( + "pod", + delayed_replica_host, + r".metadata.labels.clickhouse\.altinity\.com\/ready", + value="yes", + ) + + with And("Next replica should be created after sync completes"): + kubectl.wait_object("pod", "", label=f"-l clickhouse.altinity.com/chi={chi}", count=3) + kubectl.wait_field( + "pod", + next_replica_host, + r".metadata.labels.clickhouse\.altinity\.com\/ready", + value="yes", + ) + + with And("Live inserts should be visible on the synced replica"): + wait_delayed_replica_row_count("6") + + with Finally("I clean up"): + delete_test_namespace() + + +@TestScenario +@Tags("HEAVY") +@Name("test_010079_2. Sync gate OFF control: rolling reconcile advances past a delayed replica") +def test_010079_2(self): + """No-waits baseline for test_010079. Same fixture (replicated table, REPLICATED SENDS + stopped, scale 1 -> 3), with the catch-up gate disabled AND wait.replicas.all/new off, so no + catch-up wait of any kind applies. The reconcile MUST advance to the third replica while + the second one is still behind. + + Note this disables all three knobs together, so it establishes that the fixture itself + does not stall - it does not isolate the gate from the pre-existing replication-delay + wait. Isolating those would need a third scenario with the gate off but wait.replicas.new + left on.""" + create_shell_namespace_clickhouse_template() + + with Given("I disable the replicated host catch-up gate"): + util.apply_operator_config("manifests/chopconf/test-079-sync-gate-off.yaml") + + util.require_keeper(keeper_type=self.context.keeper_type) + + manifest = "manifests/chi/test-079-sync-gate-1.yaml" + chi = yaml_manifest.get_name(util.get_full_path(manifest)) + cluster = "default" + source_host = f"chi-{chi}-{cluster}-0-0-0" + delayed_replica_host = f"chi-{chi}-{cluster}-0-1-0" + delayed_replica_fqdn = f"chi-{chi}-{cluster}-0-1.{current().context.test_namespace}.svc.cluster.local" + + with Given("CHI is installed"): + kubectl.create_and_check( + manifest=manifest, + check={ + "pod_count": 1, + "apply_templates": {current().context.clickhouse_template}, + "do_not_delete": 1, + }, + ) + + with Then("Create a replicated table and stop replicated sends"): + clickhouse.query( + chi, + "CREATE TABLE test_079 (a Int64) Engine = ReplicatedMergeTree('/clickhouse/tables/{database}/{table}', '{replica}') ORDER BY a PARTITION BY a", + ) + clickhouse.query(chi, "INSERT INTO test_079 SELECT 1") + clickhouse.query(chi, "SYSTEM STOP REPLICATED SENDS", host=source_host) + + with When("Scale to three replicas while the new replica is delayed"): + kubectl.create_and_check( + manifest="manifests/chi/test-079-sync-gate-2.yaml", + check={"do_not_delete": 1, "pod_count": 3}, + ) + + with Then("All three pods exist even though the second replica is behind"): + assert kubectl.get_count("pod", chi=chi) == 3, error("gate-OFF reconcile must not stop at 2 pods") + replica_delay = clickhouse.query( + chi, "select max(absolute_delay) from system.replicas", host=delayed_replica_host + ) + print(f"max(absolute_delay)={replica_delay}") + + with And("No caught-up marker is written for the delayed replica"): + chi_status = kubectl.get("chi", chi).get("status") or {} + assert delayed_replica_fqdn not in (chi_status.get("hostsWithReplicaCaughtUp") or []), error( + "caught-up marker must not be written when the gate is disabled" + ) + + with Finally("I clean up"): + clickhouse.query_with_error(chi, "SYSTEM START REPLICATED SENDS", host=source_host) + delete_test_namespace() + + @TestScenario @Tags("HEAVY") @Requirements(RQ_SRS_026_ClickHouseOperator_EnableHttps("1.0"))