diff --git a/.github/benchmark-configs.json b/.github/benchmark-configs.json new file mode 100644 index 0000000000000..5b44198cd3b8e --- /dev/null +++ b/.github/benchmark-configs.json @@ -0,0 +1,161 @@ +{ + "name": "Cluster and opensearch-benchmark configurations", + "id_1": { + "description": "Indexing only configuration for NYC_TAXIS workload", + "supported_major_versions": ["2", "3"], + "cluster-benchmark-configs": { + "SINGLE_NODE_CLUSTER": "true", + "MIN_DISTRIBUTION": "true", + "TEST_WORKLOAD": "nyc_taxis", + "WORKLOAD_PARAMS": "{\"number_of_replicas\":\"0\",\"number_of_shards\":\"1\"}", + "EXCLUDE_TASKS": "type:search", + "CAPTURE_NODE_STAT": "true" + }, + "cluster_configuration": { + "size": "Single-Node", + "data_instance_config": "4vCPU, 32G Mem, 16G Heap" + } + }, + "id_2": { + "description": "Indexing only configuration for HTTP_LOGS workload", + "supported_major_versions": ["2", "3"], + "cluster-benchmark-configs": { + "SINGLE_NODE_CLUSTER": "true", + "MIN_DISTRIBUTION": "true", + "TEST_WORKLOAD": "http_logs", + "WORKLOAD_PARAMS": "{\"number_of_replicas\":\"0\",\"number_of_shards\":\"1\"}", + "EXCLUDE_TASKS": "type:search", + "CAPTURE_NODE_STAT": "true" + }, + "cluster_configuration": { + "size": "Single-Node", + "data_instance_config": "4vCPU, 32G Mem, 16G Heap" + } + }, + "id_3": { + "description": "Search only test-procedure for NYC_TAXIS, uses snapshot to restore the data for OS-3.0.0", + "supported_major_versions": ["3"], + "cluster-benchmark-configs": { + "SINGLE_NODE_CLUSTER": "true", + "MIN_DISTRIBUTION": "true", + "TEST_WORKLOAD": "nyc_taxis", + "WORKLOAD_PARAMS": "{\"snapshot_repo_name\":\"benchmark-workloads-repo-300\",\"snapshot_bucket_name\":\"benchmark-workload-snapshots\",\"snapshot_region\":\"us-east-1\",\"snapshot_base_path\":\"workload-snapshots-300\",\"snapshot_name\":\"nyc_taxis_1_shard\"}", + "CAPTURE_NODE_STAT": "true", + "TEST_PROCEDURE": "restore-from-snapshot" + }, + "cluster_configuration": { + "size": "Single-Node", + "data_instance_config": "4vCPU, 32G Mem, 16G Heap" + } + }, + "id_4": { + "description": "Search only test-procedure for HTTP_LOGS, uses snapshot to restore the data for OS-3.0.0", + "supported_major_versions": ["3"], + "cluster-benchmark-configs": { + "SINGLE_NODE_CLUSTER": "true", + "MIN_DISTRIBUTION": "true", + "TEST_WORKLOAD": "http_logs", + "WORKLOAD_PARAMS": "{\"snapshot_repo_name\":\"benchmark-workloads-repo-300\",\"snapshot_bucket_name\":\"benchmark-workload-snapshots\",\"snapshot_region\":\"us-east-1\",\"snapshot_base_path\":\"workload-snapshots-300\",\"snapshot_name\":\"http_logs_1_shard\"}", + "CAPTURE_NODE_STAT": "true", + "TEST_PROCEDURE": "restore-from-snapshot" + }, + "cluster_configuration": { + "size": "Single-Node", + "data_instance_config": "4vCPU, 32G Mem, 16G Heap" + } + }, + "id_5": { + "description": "Search only test-procedure for HTTP_LOGS, uses snapshot to restore the data for OS-3.0.0", + "supported_major_versions": ["3"], + "cluster-benchmark-configs": { + "SINGLE_NODE_CLUSTER": "true", + "MIN_DISTRIBUTION": "true", + "TEST_WORKLOAD": "big5", + "WORKLOAD_PARAMS": "{\"snapshot_repo_name\":\"benchmark-workloads-repo-300\",\"snapshot_bucket_name\":\"benchmark-workload-snapshots\",\"snapshot_region\":\"us-east-1\",\"snapshot_base_path\":\"workload-snapshots-300\",\"snapshot_name\":\"big5_1_shard\"}", + "CAPTURE_NODE_STAT": "true", + "TEST_PROCEDURE": "restore-from-snapshot" + }, + "cluster_configuration": { + "size": "Single-Node", + "data_instance_config": "4vCPU, 32G Mem, 16G Heap" + } + }, + "id_6": { + "description": "Search only test-procedure for NYC_TAXIS, uses snapshot to restore the data for OS-2.x", + "supported_major_versions": ["2"], + "cluster-benchmark-configs": { + "SINGLE_NODE_CLUSTER": "true", + "MIN_DISTRIBUTION": "true", + "TEST_WORKLOAD": "nyc_taxis", + "WORKLOAD_PARAMS": "{\"snapshot_repo_name\":\"benchmark-workloads-repo\",\"snapshot_bucket_name\":\"benchmark-workload-snapshots\",\"snapshot_region\":\"us-east-1\",\"snapshot_base_path\":\"workload-snapshots\",\"snapshot_name\":\"nyc_taxis_1_shard\"}", + "CAPTURE_NODE_STAT": "true", + "TEST_PROCEDURE": "restore-from-snapshot" + }, + "cluster_configuration": { + "size": "Single-Node", + "data_instance_config": "4vCPU, 32G Mem, 16G Heap" + } + }, + "id_7": { + "description": "Search only test-procedure for HTTP_LOGS, uses snapshot to restore the data for OS-2.x", + "supported_major_versions": ["2"], + "cluster-benchmark-configs": { + "SINGLE_NODE_CLUSTER": "true", + "MIN_DISTRIBUTION": "true", + "TEST_WORKLOAD": "http_logs", + "WORKLOAD_PARAMS": "{\"snapshot_repo_name\":\"benchmark-workloads-repo\",\"snapshot_bucket_name\":\"benchmark-workload-snapshots\",\"snapshot_region\":\"us-east-1\",\"snapshot_base_path\":\"workload-snapshots\",\"snapshot_name\":\"http_logs_1_shard\"}", + "CAPTURE_NODE_STAT": "true", + "TEST_PROCEDURE": "restore-from-snapshot" + }, + "cluster_configuration": { + "size": "Single-Node", + "data_instance_config": "4vCPU, 32G Mem, 16G Heap" + } + }, + "id_8": { + "description": "Search only test-procedure for HTTP_LOGS, uses snapshot to restore the data for OS-2.x", + "supported_major_versions": ["2"], + "cluster-benchmark-configs": { + "SINGLE_NODE_CLUSTER": "true", + "MIN_DISTRIBUTION": "true", + "TEST_WORKLOAD": "big5", + "WORKLOAD_PARAMS": "{\"snapshot_repo_name\":\"benchmark-workloads-repo\",\"snapshot_bucket_name\":\"benchmark-workload-snapshots\",\"snapshot_region\":\"us-east-1\",\"snapshot_base_path\":\"workload-snapshots\",\"snapshot_name\":\"big5_1_shard\"}", + "CAPTURE_NODE_STAT": "true", + "TEST_PROCEDURE": "restore-from-snapshot" + }, + "cluster_configuration": { + "size": "Single-Node", + "data_instance_config": "4vCPU, 32G Mem, 16G Heap" + } + }, + "id_9": { + "description": "Indexing and search configuration for pmc workload", + "supported_major_versions": ["2", "3"], + "cluster-benchmark-configs": { + "SINGLE_NODE_CLUSTER": "true", + "MIN_DISTRIBUTION": "true", + "TEST_WORKLOAD": "pmc", + "WORKLOAD_PARAMS": "{\"number_of_replicas\":\"0\",\"number_of_shards\":\"1\"}", + "CAPTURE_NODE_STAT": "true" + }, + "cluster_configuration": { + "size": "Single-Node", + "data_instance_config": "4vCPU, 32G Mem, 16G Heap" + } + }, + "id_10": { + "description": "Indexing only configuration for stack-overflow workload", + "supported_major_versions": ["2", "3"], + "cluster-benchmark-configs": { + "SINGLE_NODE_CLUSTER": "true", + "MIN_DISTRIBUTION": "true", + "TEST_WORKLOAD": "so", + "WORKLOAD_PARAMS": "{\"number_of_replicas\":\"0\",\"number_of_shards\":\"1\"}", + "CAPTURE_NODE_STAT": "true" + }, + "cluster_configuration": { + "size": "Single-Node", + "data_instance_config": "4vCPU, 32G Mem, 16G Heap" + } + } +} diff --git a/.github/workflows/add-performance-comment.yml b/.github/workflows/add-performance-comment.yml new file mode 100644 index 0000000000000..b522d348c84b2 --- /dev/null +++ b/.github/workflows/add-performance-comment.yml @@ -0,0 +1,25 @@ +name: Performance Label Action + +on: + pull_request_target: + types: [labeled] + +jobs: + add-comment: + if: github.event.label.name == 'Performance' + runs-on: ubuntu-latest + permissions: + pull-requests: write + + steps: + - name: Add comment to PR + uses: actions/github-script@v6 + with: + github-token: ${{secrets.GITHUB_TOKEN}} + script: | + github.rest.issues.createComment({ + issue_number: context.issue.number, + owner: context.repo.owner, + repo: context.repo.repo, + body: "Hello!\nWe have added a performance benchmark workflow that runs by adding a comment on the PR.\n Please refer https://github.com/opensearch-project/OpenSearch/blob/main/PERFORMANCE_BENCHMARKS.md on how to run benchmarks on pull requests." + }) diff --git a/.github/workflows/benchmark-pull-request.yml b/.github/workflows/benchmark-pull-request.yml new file mode 100644 index 0000000000000..2e2e83eb132de --- /dev/null +++ b/.github/workflows/benchmark-pull-request.yml @@ -0,0 +1,163 @@ +name: Run performance benchmark on pull request +on: + issue_comment: + types: [created] +jobs: + run-performance-benchmark-on-pull-request: + if: ${{ (github.event.issue.pull_request) && (contains(github.event.comment.body, '"run-benchmark-test"')) }} + runs-on: ubuntu-latest + permissions: + id-token: write + contents: read + issues: write + pull-requests: write + steps: + - name: Checkout Repository + uses: actions/checkout@v3 + - name: Set up required env vars + run: | + echo "PR_NUMBER=${{ github.event.issue.number }}" >> $GITHUB_ENV + echo "REPOSITORY=${{ github.event.repository.full_name }}" >> $GITHUB_ENV + OPENSEARCH_VERSION=$(awk -F '=' '/^opensearch[[:space:]]*=/ {gsub(/[[:space:]]/, "", $2); print $2}' buildSrc/version.properties) + echo "OPENSEARCH_VERSION=$OPENSEARCH_VERSION" >> $GITHUB_ENV + major_version=$(echo $OPENSEARCH_VERSION | cut -d'.' -f1) + echo "OPENSEARCH_MAJOR_VERSION=$major_version" >> $GITHUB_ENV + echo "USER_TAGS=pull_request_number:${{ github.event.issue.number }},repository:OpenSearch" >> $GITHUB_ENV + - name: Check comment format + id: check_comment + uses: actions/github-script@v6 + with: + script: | + const fs = require('fs'); + const comment = context.payload.comment.body; + let commentJson; + try { + commentJson = JSON.parse(comment); + } catch (error) { + core.setOutput('invalid', 'true'); + return; + } + if (!commentJson.hasOwnProperty('run-benchmark-test')) { + core.setOutput('invalid', 'true'); + return; + } + const configId = commentJson['run-benchmark-test']; + let benchmarkConfigs; + try { + benchmarkConfigs = JSON.parse(fs.readFileSync('.github/benchmark-configs.json', 'utf8')); + } catch (error) { + core.setFailed('Failed to read benchmark-configs.json'); + return; + } + const openSearchMajorVersion = process.env.OPENSEARCH_MAJOR_VERSION; + console.log('MAJOR_VERSION', openSearchMajorVersion) + if (!benchmarkConfigs.hasOwnProperty(configId) || + !benchmarkConfigs[configId].supported_major_versions.includes(openSearchMajorVersion)) { + core.setOutput('invalid', 'true'); + return; + } + const clusterBenchmarkConfigs = benchmarkConfigs[configId]['cluster-benchmark-configs']; + for (const [key, value] of Object.entries(clusterBenchmarkConfigs)) { + core.exportVariable(key, value); + } + - name: Post invalid format comment + if: steps.check_comment.outputs.invalid == 'true' + uses: actions/github-script@v6 + with: + github-token: ${{secrets.GITHUB_TOKEN}} + script: | + github.rest.issues.createComment({ + issue_number: context.issue.number, + owner: context.repo.owner, + repo: context.repo.repo, + body: 'Invalid comment format or config id. Please refer to https://github.com/opensearch-project/OpenSearch/blob/main/PERFORMANCE_BENCHMARKS.md on how to run benchmarks on pull requests.' + }) + - name: Fail workflow for invalid comment + if: steps.check_comment.outputs.invalid == 'true' + run: | + echo "Invalid comment format detected. Failing the workflow." + exit 1 + - id: get_approvers + run: | + echo "approvers=$(cat .github/CODEOWNERS | grep '^\*' | tr -d '* ' | sed 's/@/,/g' | sed 's/,//1')" >> $GITHUB_OUTPUT + - uses: trstringer/manual-approval@v1 + if: (!contains(steps.get_approvers.outputs.approvers, github.event.comment.user.login)) + with: + secret: ${{ github.TOKEN }} + approvers: ${{ steps.get_approvers.outputs.approvers }} + minimum-approvals: 1 + issue-title: 'Request to approve/deny benchmark run for PR #${{ env.PR_NUMBER }}' + issue-body: "Please approve or deny the benchmark run for PR #${{ env.PR_NUMBER }}" + exclude-workflow-initiator-as-approver: false + - name: Get PR Details + id: get_pr + uses: actions/github-script@v7 + with: + script: | + const issue = context.payload.issue; + const prNumber = issue.number; + console.log(`Pull Request Number: ${prNumber}`); + + const { data: pull_request } = await github.rest.pulls.get({ + owner: context.repo.owner, + repo: context.repo.repo, + pull_number: prNumber, + }); + + return { + "headRepoFullName": pull_request.head.repo.full_name, + "headRef": pull_request.head.ref + }; + - name: Set pr details env vars + run: | + echo '${{ steps.get_pr.outputs.result }}' | jq -r '.headRepoFullName' + echo '${{ steps.get_pr.outputs.result }}' | jq -r '.headRef' + headRepo=$(echo '${{ steps.get_pr.outputs.result }}' | jq -r '.headRepoFullName') + headRef=$(echo '${{ steps.get_pr.outputs.result }}' | jq -r '.headRef') + echo "prHeadRepo=$headRepo" >> $GITHUB_ENV + echo "prHeadRef=$headRef" >> $GITHUB_ENV + - name: Checkout PR Repo + uses: actions/checkout@v2 + with: + repository: ${{ env.prHeadRepo }} + ref: ${{ env.prHeadRef }} + token: ${{ secrets.GITHUB_TOKEN }} + - name: Setup Java + uses: actions/setup-java@v1 + with: + java-version: 21 + - name: Build and Assemble OpenSearch from PR + run: | + ./gradlew :distribution:archives:linux-tar:assemble -Dbuild.snapshot=false + - name: Configure AWS credentials + uses: aws-actions/configure-aws-credentials@v4 + with: + role-to-assume: ${{ secrets.UPLOAD_ARCHIVE_ARTIFACT_ROLE }} + role-session-name: publish-to-s3 + aws-region: us-west-2 + - name: Push to S3 + run: | + aws s3 cp distribution/archives/linux-tar/build/distributions/opensearch-min-$OPENSEARCH_VERSION-linux-x64.tar.gz s3://${{ secrets.ARCHIVE_ARTIFACT_BUCKET_NAME }}/PR-$PR_NUMBER/ + echo "DISTRIBUTION_URL=${{ secrets.ARTIFACT_BUCKET_CLOUDFRONT_URL }}/PR-$PR_NUMBER/opensearch-min-$OPENSEARCH_VERSION-linux-x64.tar.gz" >> $GITHUB_ENV + - name: Checkout opensearch-build repo + uses: actions/checkout@v4 + with: + repository: opensearch-project/opensearch-build + ref: main + path: opensearch-build + - name: Trigger jenkins workflow to run gradle check + run: | + cat $GITHUB_ENV + bash opensearch-build/scripts/benchmark/benchmark-pull-request.sh ${{ secrets.JENKINS_PR_BENCHMARK_GENERIC_WEBHOOK_TOKEN }} + - name: Update PR with Job Url + uses: actions/github-script@v6 + with: + github-token: ${{ secrets.GITHUB_TOKEN }} + script: | + const workflowUrl = process.env.WORKFLOW_URL; + github.rest.issues.createComment({ + issue_number: context.issue.number, + owner: context.repo.owner, + repo: context.repo.repo, + body: `The Jenkins job url is ${workflowUrl} . Final results will be published once the job is completed.` + }) diff --git a/CHANGELOG.md b/CHANGELOG.md index f8c7b0c1cc940..f3b471989ad1c 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -20,6 +20,7 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/), - Add matchesPluginSystemIndexPattern to SystemIndexRegistry ([#14750](https://github.com/opensearch-project/OpenSearch/pull/14750)) - Add Plugin interface for loading application based configuration templates (([#14659](https://github.com/opensearch-project/OpenSearch/issues/14659))) - Refactor remote-routing-table service inline with remote state interfaces([#14668](https://github.com/opensearch-project/OpenSearch/pull/14668)) +- Add prefix mode verification setting for repository verification (([#14790](https://github.com/opensearch-project/OpenSearch/pull/14790))) - Reduce logging in DEBUG for MasterService:run ([#14795](https://github.com/opensearch-project/OpenSearch/pull/14795)) ### Dependencies @@ -76,6 +77,9 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/), - Refactoring Grok.validatePatternBank by using an iterative approach ([#14206](https://github.com/opensearch-project/OpenSearch/pull/14206)) - Update help output for _cat ([#14722](https://github.com/opensearch-project/OpenSearch/pull/14722)) - Fix bulk upsert ignores the default_pipeline and final_pipeline when auto-created index matches the index template ([#12891](https://github.com/opensearch-project/OpenSearch/pull/12891)) +- Fix NPE in ReplicaShardAllocator ([#14385](https://github.com/opensearch-project/OpenSearch/pull/14385)) +- Fix constant_keyword field type used when creating index ([#14807](https://github.com/opensearch-project/OpenSearch/pull/14807)) +- Use circuit breaker in InternalHistogram when adding empty buckets ([#14754](https://github.com/opensearch-project/OpenSearch/pull/14754)) ### Security diff --git a/PERFORMANCE_BENCHMARKS.md b/PERFORMANCE_BENCHMARKS.md new file mode 100644 index 0000000000000..252c4ae312136 --- /dev/null +++ b/PERFORMANCE_BENCHMARKS.md @@ -0,0 +1,112 @@ +# README: Running Performance Benchmarks on Pull Requests + +## Overview + +`benchmark-pull-request` GitHub Actions workflow is designed to automatically run performance benchmarks on a pull request when a specific comment is made on the pull request. This ensures that performance benchmarks are consistently and accurately applied to code changes, helping maintain the performance standards of the repository. + +## Workflow Trigger + +The workflow is triggered when a new comment is created on a pull request. Specifically, it checks for the presence of the `"run-benchmark-test"` keyword in the comment body. If this keyword is detected, the workflow proceeds to run the performance benchmarks. + +## Key Steps in the Workflow + +1. **Check Comment Format and Configuration:** + - Validates the format of the comment to ensure it contains the required `"run-benchmark-test"` keyword and is in json format. + - Extracts the benchmark configuration ID from the comment and verifies if it exists in the `benchmark-config.json` file. + - Checks if the extracted configuration ID is supported for the current OpenSearch major version. + +2. **Post Invalid Format Comment:** + - If the comment format is invalid or the configuration ID is not supported, a comment is posted on the pull request indicating the problem, and the workflow fails. + +3. **Manual Approval (if necessary):** + - Fetches the list of approvers from the `.github/CODEOWNERS` file. + - If the commenter is not one of the maintainers, a manual approval request is created. The workflow pauses until an approver approves or denies the benchmark run by commenting appropriate word on the issue. + - The issue for approval request is auto-closed once the approver is done adding appropriate comment + +4. **Build and Assemble OpenSearch:** + - Builds and assembles (x64-linux tar) the OpenSearch distribution from the pull request code changes. + +5. **Upload to S3:** + - Configures AWS credentials and uploads the assembled OpenSearch distribution to an S3 bucket for further use in benchmarking. + - The S3 bucket is fronted by cloudfront to only allow downloads. + - The lifecycle policy on the S3 bucket will delete the uploaded artifacts after 30-days. + +6. **Trigger Jenkins Workflow:** + - Triggers a Jenkins workflow to run the benchmark tests using a webhook token. + +7. **Update Pull Request with Job URL:** + - Posts a comment on the pull request with the URL of the Jenkins job. The final benchmark results will be posted once the job completes. + - To learn about how benchmark job works see https://github.com/opensearch-project/opensearch-build/tree/main/src/test_workflow#benchmarking-tests + +## How to Use This Workflow + +1. **Ensure `benchmark-config.json` is Up-to-Date:** + - The `benchmark-config.json` file should contain valid benchmark configurations with supported major versions and cluster-benchmark configurations. + +2. **Add the Workflow to Your Repository:** + - Save the workflow YAML file (typically named `benchmark.yml`) in the `.github/workflows` directory of your repository. + +3. **Make a Comment to Trigger the Workflow:** + - On any pull request issue, make a comment containing the keyword `"run-benchmark-test"` along with the configuration ID. For example: + ```json + {"run-benchmark-test": "id_1"} + ``` + +4. **Monitor Workflow Progress:** + - The workflow will validate the comment, check for approval (if necessary), build the OpenSearch distribution, and trigger the Jenkins job. + - A comment will be posted on the pull request with the URL of the Jenkins job. You can monitor the progress and final results there as well. + +## Example Comment Format + +To run the benchmark with configuration ID `id_1`, post the following comment on the pull request issue: +```json +{"run-benchmark-test": "id_1"} +``` + +## How to add a new benchmark configuration + +The benchmark-config.json file accepts the following schema. +```json +{ + "id_": { + "description": "Short description of the configuration", + "supported_major_versions": ["2", "3"], + "cluster-benchmark-configs": { + "SINGLE_NODE_CLUSTER": "Use single node cluster for benchmarking, accepted values are \"true\" or \"false\"", + "MIN_DISTRIBUTION": "Use OpenSearch min distribution, should always be \"true\"", + "MANAGER_NODE_COUNT": "For multi-node cluster tests, number of cluster manager nodes, empty value defaults to 3.", + "DATA_NODE_COUNT": "For multi-node cluster tests, number of data nodes, empty value defaults to 2.", + "DATA_INSTANCE_TYPE": "EC2 instance type for data node, empty defaults to r5.xlarge.", + "DATA_NODE_STORAGE": "Data node ebs block storage size, empty value defaults to 100Gb", + "JVM_SYS_PROPS": "A comma-separated list of key=value pairs that will be added to jvm.options as JVM system properties", + "ADDITIONAL_CONFIG": "Additional space delimited opensearch.yml config parameters. e.g., `search.concurrent_segment_search.enabled:true`", + "TEST_WORKLOAD": "The workload name from OpenSearch Benchmark Workloads. https://github.com/opensearch-project/opensearch-benchmark-workloads. Default is nyc_taxis", + "WORKLOAD_PARAMS": "With this parameter you can inject variables into workloads, e.g.{\"number_of_replicas\":\"0\",\"number_of_shards\":\"3\"}. See https://opensearch.org/docs/latest/benchmark/reference/commands/command-flags/#workload-params", + "EXCLUDE_TASKS": "Defines a comma-separated list of test procedure tasks not to run. e.g. type:search, see https://opensearch.org/docs/latest/benchmark/reference/commands/command-flags/#exclude-tasks", + "INCLUDE_TASKS": "Defines a comma-separated list of test procedure tasks to run. By default, all tasks listed in a test procedure array are run. See https://opensearch.org/docs/latest/benchmark/reference/commands/command-flags/#include-tasks", + "TEST_PROCEDURE": "Defines a test procedure to use. e.g., `append-no-conflicts,significant-text`. Uses default if none provided. See https://opensearch.org/docs/latest/benchmark/reference/commands/command-flags/#test-procedure", + "CAPTURE_NODE_STAT": "Enable opensearch-benchmark node-stats telemetry to capture system level metrics like cpu, jvm etc., see https://opensearch.org/docs/latest/benchmark/reference/telemetry/#node-stats" + }, + "cluster_configuration": { + "size": "Single-Node/Multi-Node", + "data_instance_config": "data-instance-config, e.g., 4vCPU, 32G Mem, 16G Heap" + } + } +} +``` +To add a new test configuration that are suitable to your changes please create a new PR to add desired cluster and benchmark configurations. + +## How to compare my results against baseline? + +Apart from just running benchmarks the user will also be interested in how their change is performing against current OpenSearch distribution with the exactly same cluster and benchmark configurations. +The user can refer to https://s12d.com/basline-dashboards (WIP) to access baseline data for their workload, this data is generated by our nightly benchmark runs on latest build distribution artifacts for 3.0 and 2.x. +In the future, we will add the [compare](https://opensearch.org/docs/latest/benchmark/reference/commands/compare/) feature of opensearch-benchmark to run comparison and publish data on the PR as well. + +## Notes + +- Ensure all required secrets (e.g., `GITHUB_TOKEN`, `UPLOAD_ARCHIVE_ARTIFACT_ROLE`, `ARCHIVE_ARTIFACT_BUCKET_NAME`, `JENKINS_PR_BENCHMARK_GENERIC_WEBHOOK_TOKEN`) are properly set in the repository secrets. +- The `CODEOWNERS` file should list the GitHub usernames of approvers for the benchmark process. + +By following these instructions, repository maintainers can ensure consistent and automated performance benchmarking for all code changes introduced via pull requests. + + diff --git a/modules/ingest-common/src/yamlRestTest/resources/rest-api-spec/test/ingest/70_bulk.yml b/modules/ingest-common/src/yamlRestTest/resources/rest-api-spec/test/ingest/70_bulk.yml index 8830503940f4d..36b2b5351dcad 100644 --- a/modules/ingest-common/src/yamlRestTest/resources/rest-api-spec/test/ingest/70_bulk.yml +++ b/modules/ingest-common/src/yamlRestTest/resources/rest-api-spec/test/ingest/70_bulk.yml @@ -176,8 +176,8 @@ teardown: --- "Test bulk upsert honors default_pipeline and final_pipeline when the auto-created index matches with the index template": - skip: - version: " - 2.99.99" - reason: "fixed in 3.0.0" + version: " - 2.15.99" + reason: "fixed in 2.16.0" - do: indices.put_index_template: name: test_for_bulk_upsert_index_template diff --git a/rest-api-spec/src/main/resources/rest-api-spec/test/index/110_constant_keyword.yml b/rest-api-spec/src/main/resources/rest-api-spec/test/index/110_constant_keyword.yml new file mode 100644 index 0000000000000..9864bfbbb26e9 --- /dev/null +++ b/rest-api-spec/src/main/resources/rest-api-spec/test/index/110_constant_keyword.yml @@ -0,0 +1,70 @@ +--- +# The test setup includes: +# - Create index with constant_keyword field type +# - Check mapping +# - Index two example documents +# - Search +# - Delete Index when connection is teardown + +"Mappings and Supported queries": + - skip: + version: " - 2.99.99" + reason: "fixed in 3.0.0" + + # Create index with constant_keyword field type + - do: + indices.create: + index: test + body: + mappings: + properties: + genre: + type: "constant_keyword" + value: "1" + + # Index document + - do: + index: + index: test + id: 1 + body: { + "genre": "1" + } + + - do: + index: + index: test + id: 2 + body: { + "genre": 1 + } + + - do: + indices.refresh: + index: test + + # Check mapping + - do: + indices.get_mapping: + index: test + - is_true: test.mappings + - match: { test.mappings.properties.genre.type: constant_keyword } + - length: { test.mappings.properties.genre: 2 } + + # Verify Document Count + - do: + search: + body: { + query: { + match_all: {} + } + } + + - length: { hits.hits: 2 } + - match: { hits.hits.0._source.genre: "1" } + - match: { hits.hits.1._source.genre: 1 } + + # Delete Index when connection is teardown + - do: + indices.delete: + index: test diff --git a/rest-api-spec/src/main/resources/rest-api-spec/test/indices.put_alias/10_basic.yml b/rest-api-spec/src/main/resources/rest-api-spec/test/indices.put_alias/10_basic.yml index e78a5cf93c666..41f87c1df28ed 100644 --- a/rest-api-spec/src/main/resources/rest-api-spec/test/indices.put_alias/10_basic.yml +++ b/rest-api-spec/src/main/resources/rest-api-spec/test/indices.put_alias/10_basic.yml @@ -273,8 +273,8 @@ --- "Can set is_hidden": - skip: - version: " - 2.99.99" - reason: "Fix was introduced in 3.0.0" + version: " - 2.15.99" + reason: "Fix was introduced in 2.16.0" - do: indices.create: index: test_index @@ -295,8 +295,8 @@ --- "Throws exception with invalid parameters": - skip: - version: " - 2.99.99" - reason: "Fix was introduced in 3.0.0" + version: " - 2.15.99" + reason: "Fix was introduced in 2.16.0" - do: indices.create: diff --git a/server/src/internalClusterTest/java/org/opensearch/remotemigration/MigrationBaseTestCase.java b/server/src/internalClusterTest/java/org/opensearch/remotemigration/MigrationBaseTestCase.java index 2bea36ed80c9f..e4e681a5433b5 100644 --- a/server/src/internalClusterTest/java/org/opensearch/remotemigration/MigrationBaseTestCase.java +++ b/server/src/internalClusterTest/java/org/opensearch/remotemigration/MigrationBaseTestCase.java @@ -46,6 +46,7 @@ import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicLong; +import static org.opensearch.cluster.metadata.IndexMetadata.SETTING_NUMBER_OF_REPLICAS; import static org.opensearch.cluster.routing.allocation.decider.EnableAllocationDecider.CLUSTER_ROUTING_REBALANCE_ENABLE_SETTING; import static org.opensearch.gateway.remote.RemoteClusterStateService.REMOTE_CLUSTER_STATE_ENABLED_SETTING; import static org.opensearch.node.remotestore.RemoteStoreNodeService.MIGRATION_DIRECTION_SETTING; @@ -277,4 +278,30 @@ protected IndexShard getIndexShard(String dataNode, String indexName) throws Exe IndexService indexService = indicesService.indexService(new Index(indexName, uuid)); return indexService.getShard(0); } + + public void changeReplicaCountAndEnsureGreen(int replicaCount, String indexName) { + assertAcked( + client().admin() + .indices() + .prepareUpdateSettings(indexName) + .setSettings(Settings.builder().put(SETTING_NUMBER_OF_REPLICAS, replicaCount)) + ); + ensureGreen(indexName); + } + + public void completeDocRepToRemoteMigration() { + assertTrue( + internalCluster().client() + .admin() + .cluster() + .prepareUpdateSettings() + .setPersistentSettings( + Settings.builder() + .putNull(REMOTE_STORE_COMPATIBILITY_MODE_SETTING.getKey()) + .putNull(MIGRATION_DIRECTION_SETTING.getKey()) + ) + .get() + .isAcknowledged() + ); + } } diff --git a/server/src/internalClusterTest/java/org/opensearch/remotemigration/RemoteMigrationIndexMetadataUpdateIT.java b/server/src/internalClusterTest/java/org/opensearch/remotemigration/RemoteMigrationIndexMetadataUpdateIT.java index 216c104dfecc1..b55219e1cb37f 100644 --- a/server/src/internalClusterTest/java/org/opensearch/remotemigration/RemoteMigrationIndexMetadataUpdateIT.java +++ b/server/src/internalClusterTest/java/org/opensearch/remotemigration/RemoteMigrationIndexMetadataUpdateIT.java @@ -546,6 +546,73 @@ public void testRemoteIndexPathFileExistsAfterMigration() throws Exception { assertTrue(Arrays.stream(files).anyMatch(file -> file.toString().contains(fileNamePrefix))); } + /** + * Scenario: + * Creates an index with 1 pri 1 rep setup with 3 docrep nodes (1 cluster manager + 2 data nodes), + * initiate migration and create 3 remote nodes (1 cluster manager + 2 data nodes) and moves over + * only primary shard copy of the index + * After the primary shard copy is relocated, decrease replica count to 0, stop all docrep nodes + * and conclude migration. Remote store index settings should be applied to the index at this point. + */ + public void testIndexSettingsUpdateDuringReplicaCountDecrement() throws Exception { + String indexName = "migration-index-replica-decrement"; + String docrepClusterManager = internalCluster().startClusterManagerOnlyNode(); + + logger.info("---> Starting 2 docrep nodes"); + List docrepNodeNames = internalCluster().startDataOnlyNodes(2); + internalCluster().validateClusterFormed(); + + logger.info("---> Creating index with 1 primary and 1 replica"); + Settings oneReplica = Settings.builder() + .put(IndexMetadata.SETTING_NUMBER_OF_REPLICAS, 1) + .put(IndexMetadata.SETTING_NUMBER_OF_SHARDS, 1) + .build(); + createIndexAndAssertDocrepProperties(indexName, oneReplica); + + int docsToIndex = randomIntBetween(10, 100); + logger.info("---> Indexing {} on both indices", docsToIndex); + indexBulk(indexName, docsToIndex); + + logger.info( + "---> Stopping shard rebalancing to ensure shards do not automatically move over to newer nodes after they are launched" + ); + stopShardRebalancing(); + + logger.info("---> Starting 3 remote store enabled nodes"); + initDocRepToRemoteMigration(); + setAddRemote(true); + internalCluster().startClusterManagerOnlyNode(); + List remoteNodeNames = internalCluster().startDataOnlyNodes(2); + internalCluster().validateClusterFormed(); + + String primaryNode = primaryNodeName(indexName); + + logger.info("---> Moving over primary to remote store enabled nodes"); + assertAcked( + client().admin() + .cluster() + .prepareReroute() + .add(new MoveAllocationCommand(indexName, 0, primaryNode, remoteNodeNames.get(0))) + .execute() + .actionGet() + ); + waitForRelocation(); + waitNoPendingTasksOnAll(); + + logger.info("---> Reducing replica count to 0 for the index"); + changeReplicaCountAndEnsureGreen(0, indexName); + + logger.info("---> Stopping all docrep nodes"); + internalCluster().stopRandomNode(InternalTestCluster.nameFilter(docrepClusterManager)); + for (String node : docrepNodeNames) { + internalCluster().stopRandomNode(InternalTestCluster.nameFilter(node)); + } + internalCluster().validateClusterFormed(); + completeDocRepToRemoteMigration(); + waitNoPendingTasksOnAll(); + assertRemoteProperties(indexName); + } + private void createIndexAndAssertDocrepProperties(String index, Settings settings) { createIndexAssertHealthAndDocrepProperties(index, settings, this::ensureGreen); } diff --git a/server/src/main/java/org/opensearch/action/admin/cluster/settings/TransportClusterUpdateSettingsAction.java b/server/src/main/java/org/opensearch/action/admin/cluster/settings/TransportClusterUpdateSettingsAction.java index 216e1fb2ed1cc..3988d50b2ce1e 100644 --- a/server/src/main/java/org/opensearch/action/admin/cluster/settings/TransportClusterUpdateSettingsAction.java +++ b/server/src/main/java/org/opensearch/action/admin/cluster/settings/TransportClusterUpdateSettingsAction.java @@ -42,7 +42,6 @@ import org.opensearch.cluster.ClusterState; import org.opensearch.cluster.block.ClusterBlockException; import org.opensearch.cluster.block.ClusterBlockLevel; -import org.opensearch.cluster.metadata.IndexMetadata; import org.opensearch.cluster.metadata.IndexNameExpressionResolver; import org.opensearch.cluster.metadata.Metadata; import org.opensearch.cluster.node.DiscoveryNode; @@ -59,17 +58,13 @@ import org.opensearch.common.settings.SettingsException; import org.opensearch.core.action.ActionListener; import org.opensearch.core.common.io.stream.StreamInput; -import org.opensearch.index.remote.RemoteMigrationIndexMetadataUpdater; import org.opensearch.node.remotestore.RemoteStoreNodeService; import org.opensearch.threadpool.ThreadPool; import org.opensearch.transport.TransportService; import java.io.IOException; -import java.util.Collection; -import java.util.Set; -import java.util.stream.Collectors; -import static org.opensearch.index.remote.RemoteMigrationIndexMetadataUpdater.indexHasAllRemoteStoreRelatedMetadata; +import static org.opensearch.index.remote.RemoteStoreUtils.checkAndFinalizeRemoteStoreMigration; /** * Transport action for updating cluster settings @@ -262,13 +257,14 @@ public void onFailure(String source, Exception e) { @Override public ClusterState execute(final ClusterState currentState) { - validateCompatibilityModeSettingRequest(request, state); - final ClusterState clusterState = updater.updateSettings( + boolean isCompatibilityModeChanging = validateCompatibilityModeSettingRequest(request, state); + ClusterState clusterState = updater.updateSettings( currentState, clusterSettings.upgradeSettings(request.transientSettings()), clusterSettings.upgradeSettings(request.persistentSettings()), logger ); + clusterState = checkAndFinalizeRemoteStoreMigration(isCompatibilityModeChanging, request, clusterState, logger); changed = clusterState != currentState; return clusterState; } @@ -278,19 +274,23 @@ public ClusterState execute(final ClusterState currentState) { /** * Runs various checks associated with changing cluster compatibility mode + * * @param request cluster settings update request, for settings to be updated and new values * @param clusterState current state of cluster, for information on nodes + * @return true if the incoming cluster settings update request is switching compatibility modes */ - public void validateCompatibilityModeSettingRequest(ClusterUpdateSettingsRequest request, ClusterState clusterState) { + public boolean validateCompatibilityModeSettingRequest(ClusterUpdateSettingsRequest request, ClusterState clusterState) { Settings settings = Settings.builder().put(request.persistentSettings()).put(request.transientSettings()).build(); if (RemoteStoreNodeService.REMOTE_STORE_COMPATIBILITY_MODE_SETTING.exists(settings)) { - String value = RemoteStoreNodeService.REMOTE_STORE_COMPATIBILITY_MODE_SETTING.get(settings).mode; validateAllNodesOfSameVersion(clusterState.nodes()); - if (RemoteStoreNodeService.CompatibilityMode.STRICT.mode.equals(value)) { + if (RemoteStoreNodeService.REMOTE_STORE_COMPATIBILITY_MODE_SETTING.get( + settings + ) == RemoteStoreNodeService.CompatibilityMode.STRICT) { validateAllNodesOfSameType(clusterState.nodes()); - validateIndexSettings(clusterState); } + return true; } + return false; } /** @@ -310,31 +310,18 @@ private void validateAllNodesOfSameVersion(DiscoveryNodes discoveryNodes) { * @param discoveryNodes current discovery nodes in the cluster */ private void validateAllNodesOfSameType(DiscoveryNodes discoveryNodes) { - Set nodeTypes = discoveryNodes.getNodes() + boolean allNodesDocrepEnabled = discoveryNodes.getNodes() .values() .stream() - .map(DiscoveryNode::isRemoteStoreNode) - .collect(Collectors.toSet()); - if (nodeTypes.size() != 1) { + .allMatch(discoveryNode -> discoveryNode.isRemoteStoreNode() == false); + boolean allNodesRemoteStoreEnabled = discoveryNodes.getNodes() + .values() + .stream() + .allMatch(discoveryNode -> discoveryNode.isRemoteStoreNode()); + if (allNodesDocrepEnabled == false && allNodesRemoteStoreEnabled == false) { throw new SettingsException( "can not switch to STRICT compatibility mode when the cluster contains both remote and non-remote nodes" ); } } - - /** - * Verifies that while trying to switch to STRICT compatibility mode, - * all indices in the cluster have {@link RemoteMigrationIndexMetadataUpdater#indexHasAllRemoteStoreRelatedMetadata(IndexMetadata)} as true. - * If not, throws {@link SettingsException} - * @param clusterState current cluster state - */ - private void validateIndexSettings(ClusterState clusterState) { - Collection allIndicesMetadata = clusterState.metadata().indices().values(); - if (allIndicesMetadata.isEmpty() == false - && allIndicesMetadata.stream().anyMatch(indexMetadata -> indexHasAllRemoteStoreRelatedMetadata(indexMetadata) == false)) { - throw new SettingsException( - "can not switch to STRICT compatibility mode since all indices in the cluster does not have remote store based index settings" - ); - } - } } diff --git a/server/src/main/java/org/opensearch/gateway/ReplicaShardAllocator.java b/server/src/main/java/org/opensearch/gateway/ReplicaShardAllocator.java index d9474b32bdbf6..aaf0d696e1444 100644 --- a/server/src/main/java/org/opensearch/gateway/ReplicaShardAllocator.java +++ b/server/src/main/java/org/opensearch/gateway/ReplicaShardAllocator.java @@ -100,7 +100,10 @@ protected Runnable cancelExistingRecoveryForBetterMatch( Metadata metadata = allocation.metadata(); RoutingNodes routingNodes = allocation.routingNodes(); ShardRouting primaryShard = allocation.routingNodes().activePrimary(shard.shardId()); - assert primaryShard != null : "the replica shard can be allocated on at least one node, so there must be an active primary"; + if (primaryShard == null) { + logger.trace("{}: no active primary shard found or allocated, letting actual allocation figure it out", shard); + return null; + } assert primaryShard.currentNodeId() != null; final DiscoveryNode primaryNode = allocation.nodes().get(primaryShard.currentNodeId()); diff --git a/server/src/main/java/org/opensearch/gateway/remote/RemoteClusterStateService.java b/server/src/main/java/org/opensearch/gateway/remote/RemoteClusterStateService.java index 7e7a93e1d42ec..e7ca3e8aa7594 100644 --- a/server/src/main/java/org/opensearch/gateway/remote/RemoteClusterStateService.java +++ b/server/src/main/java/org/opensearch/gateway/remote/RemoteClusterStateService.java @@ -258,7 +258,7 @@ public RemoteClusterStateManifestInfo writeFullMetadata(ClusterState clusterStat uploadedMetadataResults.uploadedIndicesRoutingMetadata.size() ); } else { - logger.info( + logger.debug( "writing cluster state took [{}ms]; " + "wrote full state with [{}] indices, [{}] indicesRouting and global metadata", durationMillis, uploadedMetadataResults.uploadedIndexMetadata.size(), @@ -457,8 +457,8 @@ public RemoteClusterStateManifestInfo writeIncrementalMetadata( customsDiff.getUpserts().size() ); } else { - logger.info("{}; {}", clusterStateUploadTimeMessage, metadataUpdateMessage); - logger.info( + logger.debug("{}; {}", clusterStateUploadTimeMessage, metadataUpdateMessage); + logger.debug( "writing cluster state for version [{}] took [{}ms]; " + "wrote metadata for [{}] indices and skipped [{}] unchanged indices, coordination metadata updated : [{}], " + "settings metadata updated : [{}], templates metadata updated : [{}], custom metadata updated : [{}]", diff --git a/server/src/main/java/org/opensearch/index/mapper/ConstantKeywordFieldMapper.java b/server/src/main/java/org/opensearch/index/mapper/ConstantKeywordFieldMapper.java index f4730c70362d1..2edd817f61f61 100644 --- a/server/src/main/java/org/opensearch/index/mapper/ConstantKeywordFieldMapper.java +++ b/server/src/main/java/org/opensearch/index/mapper/ConstantKeywordFieldMapper.java @@ -68,11 +68,11 @@ private static ConstantKeywordFieldMapper toType(FieldMapper in) { */ public static class Builder extends ParametrizedFieldMapper.Builder { - private final Parameter value; + private final Parameter value = Parameter.stringParam(valuePropertyName, false, m -> toType(m).value, null); public Builder(String name, String value) { super(name); - this.value = Parameter.stringParam(valuePropertyName, false, m -> toType(m).value, value); + this.value.setValue(value); } @Override diff --git a/server/src/main/java/org/opensearch/index/remote/RemoteStoreUtils.java b/server/src/main/java/org/opensearch/index/remote/RemoteStoreUtils.java index 9a9de6c819424..a5e0c10f72301 100644 --- a/server/src/main/java/org/opensearch/index/remote/RemoteStoreUtils.java +++ b/server/src/main/java/org/opensearch/index/remote/RemoteStoreUtils.java @@ -11,17 +11,23 @@ import org.apache.logging.log4j.LogManager; import org.apache.logging.log4j.Logger; import org.opensearch.Version; +import org.opensearch.action.admin.cluster.settings.ClusterUpdateSettingsRequest; import org.opensearch.cluster.ClusterState; import org.opensearch.cluster.metadata.IndexMetadata; +import org.opensearch.cluster.metadata.Metadata; import org.opensearch.cluster.node.DiscoveryNode; import org.opensearch.cluster.node.DiscoveryNodes; +import org.opensearch.cluster.routing.RoutingTable; import org.opensearch.common.collect.Tuple; import org.opensearch.common.settings.Settings; import org.opensearch.node.remotestore.RemoteStoreNodeAttribute; +import org.opensearch.node.remotestore.RemoteStoreNodeService; import java.nio.ByteBuffer; import java.util.Arrays; import java.util.Base64; +import java.util.Collection; +import java.util.Collections; import java.util.HashMap; import java.util.List; import java.util.Locale; @@ -29,7 +35,9 @@ import java.util.Objects; import java.util.Optional; import java.util.function.Function; +import java.util.stream.Collectors; +import static org.opensearch.index.remote.RemoteMigrationIndexMetadataUpdater.indexHasRemoteStoreSettings; import static org.opensearch.indices.RemoteStoreSettings.CLUSTER_REMOTE_STORE_PATH_HASH_ALGORITHM_SETTING; import static org.opensearch.indices.RemoteStoreSettings.CLUSTER_REMOTE_STORE_PATH_TYPE_SETTING; import static org.opensearch.indices.RemoteStoreSettings.CLUSTER_REMOTE_STORE_TRANSLOG_METADATA; @@ -214,13 +222,13 @@ public static Map determineRemoteStoreCustomMetadataDuringMigrat // does not support custom metadata. // https://github.com/opensearch-project/OpenSearch/issues/13745 boolean blobStoreMetadataEnabled = false; - boolean translogMetadata = Version.CURRENT.compareTo(minNodeVersion) <= 0 + boolean translogMetadata = Version.V_2_15_0.compareTo(minNodeVersion) <= 0 && CLUSTER_REMOTE_STORE_TRANSLOG_METADATA.get(clusterSettings) && blobStoreMetadataEnabled; remoteCustomData.put(IndexMetadata.TRANSLOG_METADATA_KEY, Boolean.toString(translogMetadata)); - RemoteStoreEnums.PathType pathType = Version.CURRENT.compareTo(minNodeVersion) <= 0 + RemoteStoreEnums.PathType pathType = Version.V_2_15_0.compareTo(minNodeVersion) <= 0 ? CLUSTER_REMOTE_STORE_PATH_TYPE_SETTING.get(clusterSettings) : RemoteStoreEnums.PathType.FIXED; RemoteStoreEnums.PathHashAlgorithm pathHashAlgorithm = pathType == RemoteStoreEnums.PathType.FIXED @@ -250,4 +258,119 @@ public static Map getRemoteStoreRepoName(DiscoveryNodes discover .findFirst(); return remoteNode.map(RemoteStoreNodeAttribute::getDataRepoNames).orElseGet(HashMap::new); } + + /** + * Invoked after a cluster settings update. + * Checks if the applied cluster settings has switched the cluster to STRICT mode. + * If so, checks and applies appropriate index settings depending on the current set + * of node types in the cluster + * This has been intentionally done after the cluster settings update + * flow. That way we are not interfering with the usual settings update + * and the cluster state mutation that comes along with it + * + * @param isCompatibilityModeChanging flag passed from cluster settings update call to denote if a compatibility mode change has been done + * @param request request payload passed from cluster settings update + * @param currentState cluster state generated after changing cluster settings were applied + * @param logger Logger reference + * @return Mutated cluster state with remote store index settings applied, no-op if the cluster is not switching to `STRICT` compatibility mode + */ + public static ClusterState checkAndFinalizeRemoteStoreMigration( + boolean isCompatibilityModeChanging, + ClusterUpdateSettingsRequest request, + ClusterState currentState, + Logger logger + ) { + if (isCompatibilityModeChanging && isSwitchToStrictCompatibilityMode(request)) { + return finalizeMigration(currentState, logger); + } + return currentState; + } + + /** + * Finalizes the docrep to remote-store migration process by applying remote store based index settings + * on indices that are missing them. No-Op if all indices already have the settings applied through + * IndexMetadataUpdater + * + * @param incomingState mutated cluster state after cluster settings were applied + * @return new cluster state with index settings updated + */ + public static ClusterState finalizeMigration(ClusterState incomingState, Logger logger) { + Map discoveryNodeMap = incomingState.nodes().getNodes(); + if (discoveryNodeMap.isEmpty() == false) { + // At this point, we have already validated that all nodes in the cluster are of uniform type. + // Either all of them are remote store enabled, or all of them are docrep enabled + boolean remoteStoreEnabledNodePresent = discoveryNodeMap.values().stream().findFirst().get().isRemoteStoreNode(); + if (remoteStoreEnabledNodePresent == true) { + List indicesWithoutRemoteStoreSettings = getIndicesWithoutRemoteStoreSettings(incomingState, logger); + if (indicesWithoutRemoteStoreSettings.isEmpty() == true) { + logger.info("All indices in the cluster has remote store based index settings"); + } else { + Metadata mutatedMetadata = applyRemoteStoreSettings(incomingState, indicesWithoutRemoteStoreSettings, logger); + return ClusterState.builder(incomingState).metadata(mutatedMetadata).build(); + } + } else { + logger.debug("All nodes in the cluster are not remote nodes. Skipping."); + } + } + return incomingState; + } + + /** + * Filters out indices which does not have remote store based + * index settings applied even after all shard copies have + * migrated to remote store enabled nodes + */ + private static List getIndicesWithoutRemoteStoreSettings(ClusterState clusterState, Logger logger) { + Collection allIndicesMetadata = clusterState.metadata().indices().values(); + if (allIndicesMetadata.isEmpty() == false) { + List indicesWithoutRemoteSettings = allIndicesMetadata.stream() + .filter(idxMd -> indexHasRemoteStoreSettings(idxMd.getSettings()) == false) + .collect(Collectors.toList()); + logger.debug( + "Attempting to switch to strict mode. Count of indices without remote store settings {}", + indicesWithoutRemoteSettings.size() + ); + return indicesWithoutRemoteSettings; + } + return Collections.emptyList(); + } + + /** + * Applies remote store index settings through {@link RemoteMigrationIndexMetadataUpdater} + */ + private static Metadata applyRemoteStoreSettings( + ClusterState clusterState, + List indicesWithoutRemoteStoreSettings, + Logger logger + ) { + Metadata.Builder metadataBuilder = Metadata.builder(clusterState.getMetadata()); + RoutingTable currentRoutingTable = clusterState.getRoutingTable(); + DiscoveryNodes currentDiscoveryNodes = clusterState.getNodes(); + Settings currentClusterSettings = clusterState.metadata().settings(); + for (IndexMetadata indexMetadata : indicesWithoutRemoteStoreSettings) { + IndexMetadata.Builder indexMetadataBuilder = IndexMetadata.builder(indexMetadata); + RemoteMigrationIndexMetadataUpdater indexMetadataUpdater = new RemoteMigrationIndexMetadataUpdater( + currentDiscoveryNodes, + currentRoutingTable, + indexMetadata, + currentClusterSettings, + logger + ); + indexMetadataUpdater.maybeAddRemoteIndexSettings(indexMetadataBuilder, indexMetadata.getIndex().getName()); + metadataBuilder.put(indexMetadataBuilder); + } + return metadataBuilder.build(); + } + + /** + * Checks if the incoming cluster settings payload is attempting to switch + * the cluster to `STRICT` compatibility mode + * Visible only for tests + */ + public static boolean isSwitchToStrictCompatibilityMode(ClusterUpdateSettingsRequest request) { + Settings incomingSettings = Settings.builder().put(request.persistentSettings()).put(request.transientSettings()).build(); + return RemoteStoreNodeService.REMOTE_STORE_COMPATIBILITY_MODE_SETTING.get( + incomingSettings + ) == RemoteStoreNodeService.CompatibilityMode.STRICT; + } } diff --git a/server/src/main/java/org/opensearch/repositories/blobstore/BlobStoreRepository.java b/server/src/main/java/org/opensearch/repositories/blobstore/BlobStoreRepository.java index 02290b6a5e566..c4908f8c5fc4b 100644 --- a/server/src/main/java/org/opensearch/repositories/blobstore/BlobStoreRepository.java +++ b/server/src/main/java/org/opensearch/repositories/blobstore/BlobStoreRepository.java @@ -109,6 +109,7 @@ import org.opensearch.core.xcontent.XContentParser; import org.opensearch.index.mapper.MapperService; import org.opensearch.index.remote.RemoteStorePathStrategy; +import org.opensearch.index.remote.RemoteStorePathStrategy.PathInput; import org.opensearch.index.snapshots.IndexShardRestoreFailedException; import org.opensearch.index.snapshots.IndexShardSnapshotStatus; import org.opensearch.index.snapshots.blobstore.BlobStoreIndexShardSnapshot; @@ -157,6 +158,7 @@ import java.util.List; import java.util.Locale; import java.util.Map; +import java.util.Objects; import java.util.Optional; import java.util.Set; import java.util.concurrent.BlockingQueue; @@ -174,6 +176,8 @@ import java.util.stream.LongStream; import java.util.stream.Stream; +import static org.opensearch.index.remote.RemoteStoreEnums.PathHashAlgorithm.FNV_1A_COMPOSITE_1; +import static org.opensearch.index.remote.RemoteStoreEnums.PathType.HASHED_PREFIX; import static org.opensearch.index.snapshots.blobstore.BlobStoreIndexShardSnapshot.FileInfo.canonicalName; import static org.opensearch.repositories.blobstore.ChecksumBlobStoreFormat.SNAPSHOT_ONLY_FORMAT_PARAMS; @@ -302,6 +306,16 @@ public abstract class BlobStoreRepository extends AbstractLifecycleComponent imp Setting.Property.NodeScope ); + /** + * Setting to enable prefix mode verification. In this mode, a hashed string is prepended at the prefix of the base + * path during repository verification. + */ + public static final Setting PREFIX_MODE_VERIFICATION_SETTING = Setting.boolSetting( + "prefix_mode_verification", + false, + Setting.Property.NodeScope + ); + protected volatile boolean supportURLRepo; private volatile int maxShardBlobDeleteBatch; @@ -369,6 +383,8 @@ public abstract class BlobStoreRepository extends AbstractLifecycleComponent imp private final boolean isSystemRepository; + private final boolean prefixModeVerification; + private final Object lock = new Object(); private final SetOnce blobContainer = new SetOnce<>(); @@ -426,6 +442,7 @@ protected BlobStoreRepository( readRepositoryMetadata(repositoryMetadata); isSystemRepository = SYSTEM_REPOSITORY_SETTING.get(metadata.settings()); + prefixModeVerification = PREFIX_MODE_VERIFICATION_SETTING.get(metadata.settings()); this.namedXContentRegistry = namedXContentRegistry; this.threadPool = clusterService.getClusterApplierService().threadPool(); this.clusterService = clusterService; @@ -767,6 +784,10 @@ protected BlobStore getBlobStore() { return blobStore.get(); } + boolean getPrefixModeVerification() { + return prefixModeVerification; + } + /** * maintains single lazy instance of {@link BlobContainer} */ @@ -1918,7 +1939,7 @@ public String startVerification() { } else { String seed = UUIDs.randomBase64UUID(); byte[] testBytes = Strings.toUTF8Bytes(seed); - BlobContainer testContainer = blobStore().blobContainer(basePath().add(testBlobPrefix(seed))); + BlobContainer testContainer = testContainer(seed); BytesArray bytes = new BytesArray(testBytes); if (isSystemRepository == false) { try (InputStream stream = bytes.streamInput()) { @@ -1936,12 +1957,26 @@ public String startVerification() { } } + /** + * Returns the blobContainer depending on the seed and {@code prefixModeVerification}. + */ + private BlobContainer testContainer(String seed) { + BlobPath testBlobPath; + if (prefixModeVerification == true) { + PathInput pathInput = PathInput.builder().basePath(basePath()).indexUUID(seed).build(); + testBlobPath = HASHED_PREFIX.path(pathInput, FNV_1A_COMPOSITE_1); + } else { + testBlobPath = basePath(); + } + assert Objects.nonNull(testBlobPath); + return blobStore().blobContainer(testBlobPath.add(testBlobPrefix(seed))); + } + @Override public void endVerification(String seed) { if (isReadOnly() == false) { try { - final String testPrefix = testBlobPrefix(seed); - blobStore().blobContainer(basePath().add(testPrefix)).delete(); + testContainer(seed).delete(); } catch (Exception exp) { throw new RepositoryVerificationException(metadata.name(), "cannot delete test data at " + basePath(), exp); } @@ -3266,7 +3301,7 @@ public void verify(String seed, DiscoveryNode localNode) { ); } } else { - BlobContainer testBlobContainer = blobStore().blobContainer(basePath().add(testBlobPrefix(seed))); + BlobContainer testBlobContainer = testContainer(seed); try { BytesArray bytes = new BytesArray(seed); try (InputStream stream = bytes.streamInput()) { diff --git a/server/src/main/java/org/opensearch/search/aggregations/bucket/histogram/InternalHistogram.java b/server/src/main/java/org/opensearch/search/aggregations/bucket/histogram/InternalHistogram.java index a27c689127ac9..a988b911de5a3 100644 --- a/server/src/main/java/org/opensearch/search/aggregations/bucket/histogram/InternalHistogram.java +++ b/server/src/main/java/org/opensearch/search/aggregations/bucket/histogram/InternalHistogram.java @@ -395,6 +395,7 @@ private void addEmptyBuckets(List list, ReduceContext reduceContext) { // fill with empty buckets for (double key = round(emptyBucketInfo.minBound); key <= emptyBucketInfo.maxBound; key = nextKey(key)) { iter.add(new Bucket(key, 0, keyed, format, reducedEmptySubAggs)); + reduceContext.consumeBucketsAndMaybeBreak(0); } } else { Bucket first = list.get(iter.nextIndex()); @@ -402,11 +403,12 @@ private void addEmptyBuckets(List list, ReduceContext reduceContext) { // fill with empty buckets until the first key for (double key = round(emptyBucketInfo.minBound); key < first.key; key = nextKey(key)) { iter.add(new Bucket(key, 0, keyed, format, reducedEmptySubAggs)); + reduceContext.consumeBucketsAndMaybeBreak(0); } } // now adding the empty buckets within the actual data, - // e.g. if the data series is [1,2,3,7] there're 3 empty buckets that will be created for 4,5,6 + // e.g. if the data series is [1,2,3,7] there are 3 empty buckets that will be created for 4,5,6 Bucket lastBucket = null; do { Bucket nextBucket = list.get(iter.nextIndex()); @@ -414,6 +416,7 @@ private void addEmptyBuckets(List list, ReduceContext reduceContext) { double key = nextKey(lastBucket.key); while (key < nextBucket.key) { iter.add(new Bucket(key, 0, keyed, format, reducedEmptySubAggs)); + reduceContext.consumeBucketsAndMaybeBreak(0); key = nextKey(key); } assert key == nextBucket.key || Double.isNaN(nextBucket.key) : "key: " + key + ", nextBucket.key: " + nextBucket.key; @@ -424,6 +427,7 @@ private void addEmptyBuckets(List list, ReduceContext reduceContext) { // finally, adding the empty buckets *after* the actual data (based on the extended_bounds.max requested by the user) for (double key = nextKey(lastBucket.key); key <= emptyBucketInfo.maxBound; key = nextKey(key)) { iter.add(new Bucket(key, 0, keyed, format, reducedEmptySubAggs)); + reduceContext.consumeBucketsAndMaybeBreak(0); } } } diff --git a/server/src/test/java/org/opensearch/action/support/clustermanager/TransportClusterManagerNodeActionTests.java b/server/src/test/java/org/opensearch/action/support/clustermanager/TransportClusterManagerNodeActionTests.java index 35c5c5e605b4d..37e884502b613 100644 --- a/server/src/test/java/org/opensearch/action/support/clustermanager/TransportClusterManagerNodeActionTests.java +++ b/server/src/test/java/org/opensearch/action/support/clustermanager/TransportClusterManagerNodeActionTests.java @@ -47,7 +47,6 @@ import org.opensearch.common.settings.Settings; import org.opensearch.common.settings.SettingsException; import org.opensearch.common.unit.TimeValue; -import org.opensearch.common.util.FeatureFlags; import org.opensearch.common.util.concurrent.ThreadContext; import org.opensearch.core.action.ActionListener; import org.opensearch.core.action.ActionResponse; @@ -85,8 +84,6 @@ import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; -import static org.opensearch.common.util.FeatureFlags.REMOTE_STORE_MIGRATION_EXPERIMENTAL; -import static org.opensearch.index.remote.RemoteMigrationIndexMetadataUpdaterTests.createIndexMetadataWithDocrepSettings; import static org.opensearch.index.remote.RemoteMigrationIndexMetadataUpdaterTests.createIndexMetadataWithRemoteStoreSettings; import static org.opensearch.node.remotestore.RemoteStoreNodeAttribute.REMOTE_STORE_SEGMENT_REPOSITORY_NAME_ATTRIBUTE_KEY; import static org.opensearch.node.remotestore.RemoteStoreNodeAttribute.REMOTE_STORE_TRANSLOG_REPOSITORY_NAME_ATTRIBUTE_KEY; @@ -718,9 +715,6 @@ protected void masterOperation(Task task, Request request, ClusterState state, A } public void testDontAllowSwitchingToStrictCompatibilityModeForMixedCluster() { - Settings nodeSettings = Settings.builder().put(REMOTE_STORE_MIGRATION_EXPERIMENTAL, "true").build(); - FeatureFlags.initializeFeatureFlags(nodeSettings); - // request to change cluster compatibility mode to STRICT Settings currentCompatibilityModeSettings = Settings.builder() .put(REMOTE_STORE_COMPATIBILITY_MODE_SETTING.getKey(), RemoteStoreNodeService.CompatibilityMode.MIXED) @@ -809,84 +803,7 @@ public void testDontAllowSwitchingToStrictCompatibilityModeForMixedCluster() { transportClusterUpdateSettingsAction.validateCompatibilityModeSettingRequest(request, sameTypeClusterState); } - public void testDontAllowSwitchingToStrictCompatibilityModeWithoutRemoteIndexSettings() { - Settings nodeSettings = Settings.builder().put(REMOTE_STORE_MIGRATION_EXPERIMENTAL, "true").build(); - FeatureFlags.initializeFeatureFlags(nodeSettings); - Settings currentCompatibilityModeSettings = Settings.builder() - .put(REMOTE_STORE_COMPATIBILITY_MODE_SETTING.getKey(), RemoteStoreNodeService.CompatibilityMode.MIXED) - .build(); - Settings intendedCompatibilityModeSettings = Settings.builder() - .put(REMOTE_STORE_COMPATIBILITY_MODE_SETTING.getKey(), RemoteStoreNodeService.CompatibilityMode.STRICT) - .build(); - ClusterUpdateSettingsRequest request = new ClusterUpdateSettingsRequest(); - request.persistentSettings(intendedCompatibilityModeSettings); - DiscoveryNode remoteNode1 = new DiscoveryNode( - UUIDs.base64UUID(), - buildNewFakeTransportAddress(), - getRemoteStoreNodeAttributes(), - DiscoveryNodeRole.BUILT_IN_ROLES, - Version.CURRENT - ); - DiscoveryNode remoteNode2 = new DiscoveryNode( - UUIDs.base64UUID(), - buildNewFakeTransportAddress(), - getRemoteStoreNodeAttributes(), - DiscoveryNodeRole.BUILT_IN_ROLES, - Version.CURRENT - ); - DiscoveryNodes discoveryNodes = DiscoveryNodes.builder() - .add(remoteNode1) - .localNodeId(remoteNode1.getId()) - .add(remoteNode2) - .localNodeId(remoteNode2.getId()) - .build(); - AllocationService allocationService = new AllocationService( - new AllocationDeciders(Collections.singleton(new MaxRetryAllocationDecider())), - new TestGatewayAllocator(), - new BalancedShardsAllocator(Settings.EMPTY), - EmptyClusterInfoService.INSTANCE, - EmptySnapshotsInfoService.INSTANCE - ); - TransportClusterUpdateSettingsAction transportClusterUpdateSettingsAction = new TransportClusterUpdateSettingsAction( - transportService, - clusterService, - threadPool, - allocationService, - new ActionFilters(Collections.emptySet()), - new IndexNameExpressionResolver(new ThreadContext(Settings.EMPTY)), - clusterService.getClusterSettings() - ); - - Metadata nonRemoteIndexMd = Metadata.builder(createIndexMetadataWithDocrepSettings("test")) - .persistentSettings(currentCompatibilityModeSettings) - .build(); - final ClusterState clusterState = ClusterState.builder(ClusterName.DEFAULT) - .metadata(nonRemoteIndexMd) - .nodes(discoveryNodes) - .build(); - final SettingsException exception = expectThrows( - SettingsException.class, - () -> transportClusterUpdateSettingsAction.validateCompatibilityModeSettingRequest(request, clusterState) - ); - assertEquals( - "can not switch to STRICT compatibility mode since all indices in the cluster does not have remote store based index settings", - exception.getMessage() - ); - - Metadata remoteIndexMd = Metadata.builder(createIndexMetadataWithRemoteStoreSettings("test")) - .persistentSettings(currentCompatibilityModeSettings) - .build(); - ClusterState clusterStateWithRemoteIndices = ClusterState.builder(ClusterName.DEFAULT) - .metadata(remoteIndexMd) - .nodes(discoveryNodes) - .build(); - transportClusterUpdateSettingsAction.validateCompatibilityModeSettingRequest(request, clusterStateWithRemoteIndices); - } - public void testDontAllowSwitchingCompatibilityModeForClusterWithMultipleVersions() { - Settings nodeSettings = Settings.builder().put(REMOTE_STORE_MIGRATION_EXPERIMENTAL, "true").build(); - FeatureFlags.initializeFeatureFlags(nodeSettings); - // request to change cluster compatibility mode boolean toStrictMode = randomBoolean(); Settings currentCompatibilityModeSettings = Settings.builder() @@ -988,5 +905,4 @@ private Map getRemoteStoreNodeAttributes() { remoteStoreNodeAttributes.put(REMOTE_STORE_TRANSLOG_REPOSITORY_NAME_ATTRIBUTE_KEY, "my-translog-repo-1"); return remoteStoreNodeAttributes; } - } diff --git a/server/src/test/java/org/opensearch/gateway/ReplicaShardBatchAllocatorTests.java b/server/src/test/java/org/opensearch/gateway/ReplicaShardBatchAllocatorTests.java index 2e148c2bc8130..526a3990955b8 100644 --- a/server/src/test/java/org/opensearch/gateway/ReplicaShardBatchAllocatorTests.java +++ b/server/src/test/java/org/opensearch/gateway/ReplicaShardBatchAllocatorTests.java @@ -644,6 +644,25 @@ public void testDoNotCancelForBrokenNode() { assertThat(allocation.routingNodes().shardsWithState(ShardRoutingState.UNASSIGNED), empty()); } + public void testDoNotCancelForInactivePrimaryNode() { + RoutingAllocation allocation = oneInactivePrimaryOnNode1And1ReplicaRecovering(yesAllocationDeciders(), null); + testBatchAllocator.addData( + node1, + null, + "MATCH", + null, + new StoreFileMetadata("file1", 10, "MATCH_CHECKSUM", MIN_SUPPORTED_LUCENE_VERSION) + ).addData(node2, randomSyncId(), null, new StoreFileMetadata("file1", 10, "MATCH_CHECKSUM", MIN_SUPPORTED_LUCENE_VERSION)); + + testBatchAllocator.processExistingRecoveries( + allocation, + Collections.singletonList(new ArrayList<>(allocation.routingNodes().shardsWithState(ShardRoutingState.INITIALIZING))) + ); + + assertThat(allocation.routingNodesChanged(), equalTo(false)); + assertThat(allocation.routingNodes().shardsWithState(ShardRoutingState.UNASSIGNED), empty()); + } + public void testAllocateUnassignedBatchThrottlingAllocationDeciderIsHonoured() throws InterruptedException { ClusterSettings clusterSettings = new ClusterSettings(Settings.EMPTY, ClusterSettings.BUILT_IN_CLUSTER_SETTINGS); AllocationDeciders allocationDeciders = randomAllocationDeciders( @@ -872,6 +891,41 @@ private RoutingAllocation onePrimaryOnNode1And1ReplicaRecovering(AllocationDecid ); } + private RoutingAllocation oneInactivePrimaryOnNode1And1ReplicaRecovering(AllocationDeciders deciders, UnassignedInfo unassignedInfo) { + ShardRouting primaryShard = TestShardRouting.newShardRouting(shardId, node1.getId(), true, ShardRoutingState.INITIALIZING); + RoutingTable routingTable = RoutingTable.builder() + .add( + IndexRoutingTable.builder(shardId.getIndex()) + .addIndexShard( + new IndexShardRoutingTable.Builder(shardId).addShard(primaryShard) + .addShard( + TestShardRouting.newShardRouting( + shardId, + node2.getId(), + null, + false, + ShardRoutingState.INITIALIZING, + unassignedInfo + ) + ) + .build() + ) + ) + .build(); + ClusterState state = ClusterState.builder(org.opensearch.cluster.ClusterName.CLUSTER_NAME_SETTING.getDefault(Settings.EMPTY)) + .routingTable(routingTable) + .nodes(DiscoveryNodes.builder().add(node1).add(node2)) + .build(); + return new RoutingAllocation( + deciders, + new RoutingNodes(state, false), + state, + ClusterInfo.EMPTY, + SnapshotShardSizeInfo.EMPTY, + System.nanoTime() + ); + } + private RoutingAllocation onePrimaryOnNode1And1ReplicaRecovering(AllocationDeciders deciders) { return onePrimaryOnNode1And1ReplicaRecovering(deciders, new UnassignedInfo(UnassignedInfo.Reason.CLUSTER_RECOVERED, null)); } diff --git a/server/src/test/java/org/opensearch/index/mapper/ConstantKeywordFieldMapperTests.java b/server/src/test/java/org/opensearch/index/mapper/ConstantKeywordFieldMapperTests.java index 65dd3b6447663..e9d0b6d826ade 100644 --- a/server/src/test/java/org/opensearch/index/mapper/ConstantKeywordFieldMapperTests.java +++ b/server/src/test/java/org/opensearch/index/mapper/ConstantKeywordFieldMapperTests.java @@ -105,6 +105,14 @@ public void testMissingDefaultIndexMapper() throws Exception { assertThat(e.getMessage(), containsString("Field [field] is missing required parameter [value]")); } + public void testBuilderToXContent() throws IOException { + ConstantKeywordFieldMapper.Builder builder = new ConstantKeywordFieldMapper.Builder("name", "value1"); + XContentBuilder xContentBuilder = JsonXContent.contentBuilder().startObject(); + builder.toXContent(xContentBuilder, false); + xContentBuilder.endObject(); + assertEquals("{\"value\":\"value1\"}", xContentBuilder.toString()); + } + private final SourceToParse source(CheckedConsumer build) throws IOException { XContentBuilder builder = JsonXContent.contentBuilder().startObject(); build.accept(builder); diff --git a/server/src/test/java/org/opensearch/index/remote/RemoteStoreUtilsTests.java b/server/src/test/java/org/opensearch/index/remote/RemoteStoreUtilsTests.java index 15915ee431972..ec48032df4a15 100644 --- a/server/src/test/java/org/opensearch/index/remote/RemoteStoreUtilsTests.java +++ b/server/src/test/java/org/opensearch/index/remote/RemoteStoreUtilsTests.java @@ -9,15 +9,29 @@ package org.opensearch.index.remote; import org.opensearch.Version; +import org.opensearch.action.admin.cluster.settings.ClusterUpdateSettingsRequest; +import org.opensearch.cluster.ClusterName; +import org.opensearch.cluster.ClusterState; import org.opensearch.cluster.metadata.IndexMetadata; import org.opensearch.cluster.metadata.Metadata; +import org.opensearch.cluster.node.DiscoveryNode; +import org.opensearch.cluster.node.DiscoveryNodeRole; import org.opensearch.cluster.node.DiscoveryNodes; +import org.opensearch.cluster.routing.IndexRoutingTable; +import org.opensearch.cluster.routing.IndexShardRoutingTable; +import org.opensearch.cluster.routing.RoutingTable; +import org.opensearch.cluster.routing.ShardRoutingState; +import org.opensearch.cluster.routing.TestShardRouting; +import org.opensearch.common.UUIDs; import org.opensearch.common.blobstore.BlobMetadata; import org.opensearch.common.blobstore.support.PlainBlobMetadata; import org.opensearch.common.settings.Settings; +import org.opensearch.core.index.Index; +import org.opensearch.core.index.shard.ShardId; import org.opensearch.index.shard.IndexShardTestUtils; import org.opensearch.index.store.RemoteSegmentStoreDirectory; import org.opensearch.index.translog.transfer.TranslogTransferMetadata; +import org.opensearch.indices.replication.common.ReplicationType; import org.opensearch.node.remotestore.RemoteStoreNodeAttribute; import org.opensearch.test.OpenSearchTestCase; @@ -28,11 +42,15 @@ import java.util.LinkedList; import java.util.List; import java.util.Map; +import java.util.UUID; import java.util.stream.Collectors; import static org.opensearch.cluster.metadata.IndexMetadata.REMOTE_STORE_CUSTOM_KEY; +import static org.opensearch.index.remote.RemoteMigrationIndexMetadataUpdaterTests.createIndexMetadataWithDocrepSettings; import static org.opensearch.index.remote.RemoteStoreUtils.URL_BASE64_CHARSET; import static org.opensearch.index.remote.RemoteStoreUtils.determineTranslogMetadataEnabled; +import static org.opensearch.index.remote.RemoteStoreUtils.finalizeMigration; +import static org.opensearch.index.remote.RemoteStoreUtils.isSwitchToStrictCompatibilityMode; import static org.opensearch.index.remote.RemoteStoreUtils.longToCompositeBase64AndBinaryEncoding; import static org.opensearch.index.remote.RemoteStoreUtils.longToUrlBase64; import static org.opensearch.index.remote.RemoteStoreUtils.urlBase64ToLong; @@ -42,6 +60,9 @@ import static org.opensearch.index.store.RemoteSegmentStoreDirectory.MetadataFilenameUtils.METADATA_PREFIX; import static org.opensearch.index.store.RemoteSegmentStoreDirectory.MetadataFilenameUtils.SEPARATOR; import static org.opensearch.index.translog.transfer.TranslogTransferMetadata.METADATA_SEPARATOR; +import static org.opensearch.node.remotestore.RemoteStoreNodeAttribute.REMOTE_STORE_SEGMENT_REPOSITORY_NAME_ATTRIBUTE_KEY; +import static org.opensearch.node.remotestore.RemoteStoreNodeAttribute.REMOTE_STORE_TRANSLOG_REPOSITORY_NAME_ATTRIBUTE_KEY; +import static org.opensearch.node.remotestore.RemoteStoreNodeService.REMOTE_STORE_COMPATIBILITY_MODE_SETTING; public class RemoteStoreUtilsTests extends OpenSearchTestCase { @@ -398,4 +419,122 @@ private static Map getCustomDataMap(int option) { ); } + public void testFinalizeMigrationWithAllRemoteNodes() { + String migratedIndex = "migrated-index"; + Settings mockSettings = Settings.builder().put(REMOTE_STORE_COMPATIBILITY_MODE_SETTING.getKey(), "strict").build(); + DiscoveryNode remoteNode1 = new DiscoveryNode( + UUIDs.base64UUID(), + buildNewFakeTransportAddress(), + getRemoteStoreNodeAttributes(), + DiscoveryNodeRole.BUILT_IN_ROLES, + Version.CURRENT + ); + DiscoveryNode remoteNode2 = new DiscoveryNode( + UUIDs.base64UUID(), + buildNewFakeTransportAddress(), + getRemoteStoreNodeAttributes(), + DiscoveryNodeRole.BUILT_IN_ROLES, + Version.CURRENT + ); + DiscoveryNodes discoveryNodes = DiscoveryNodes.builder() + .add(remoteNode1) + .localNodeId(remoteNode1.getId()) + .add(remoteNode2) + .localNodeId(remoteNode2.getId()) + .build(); + Metadata docrepIdxMetadata = createIndexMetadataWithDocrepSettings(migratedIndex); + assertDocrepSettingsApplied(docrepIdxMetadata.index(migratedIndex)); + Metadata remoteIndexMd = Metadata.builder(docrepIdxMetadata).persistentSettings(mockSettings).build(); + ClusterState clusterStateWithDocrepIndexSettings = ClusterState.builder(ClusterName.DEFAULT) + .metadata(remoteIndexMd) + .nodes(discoveryNodes) + .routingTable(createRoutingTableAllShardsStarted(migratedIndex, 1, 1, remoteNode1, remoteNode2)) + .build(); + Metadata mutatedMetadata = finalizeMigration(clusterStateWithDocrepIndexSettings, logger).metadata(); + assertTrue(mutatedMetadata.index(migratedIndex).getVersion() > docrepIdxMetadata.index(migratedIndex).getVersion()); + assertRemoteSettingsApplied(mutatedMetadata.index(migratedIndex)); + } + + public void testFinalizeMigrationWithAllDocrepNodes() { + String docrepIndex = "docrep-index"; + Settings mockSettings = Settings.builder().put(REMOTE_STORE_COMPATIBILITY_MODE_SETTING.getKey(), "strict").build(); + DiscoveryNode docrepNode1 = new DiscoveryNode(UUIDs.base64UUID(), buildNewFakeTransportAddress(), Version.CURRENT); + DiscoveryNode docrepNode2 = new DiscoveryNode(UUIDs.base64UUID(), buildNewFakeTransportAddress(), Version.CURRENT); + DiscoveryNodes discoveryNodes = DiscoveryNodes.builder() + .add(docrepNode1) + .localNodeId(docrepNode1.getId()) + .add(docrepNode2) + .localNodeId(docrepNode2.getId()) + .build(); + Metadata docrepIdxMetadata = createIndexMetadataWithDocrepSettings(docrepIndex); + assertDocrepSettingsApplied(docrepIdxMetadata.index(docrepIndex)); + Metadata remoteIndexMd = Metadata.builder(docrepIdxMetadata).persistentSettings(mockSettings).build(); + ClusterState clusterStateWithDocrepIndexSettings = ClusterState.builder(ClusterName.DEFAULT) + .metadata(remoteIndexMd) + .nodes(discoveryNodes) + .routingTable(createRoutingTableAllShardsStarted(docrepIndex, 1, 1, docrepNode1, docrepNode2)) + .build(); + Metadata mutatedMetadata = finalizeMigration(clusterStateWithDocrepIndexSettings, logger).metadata(); + assertEquals(docrepIdxMetadata.index(docrepIndex).getVersion(), mutatedMetadata.index(docrepIndex).getVersion()); + assertDocrepSettingsApplied(mutatedMetadata.index(docrepIndex)); + } + + public void testIsSwitchToStrictCompatibilityMode() { + Settings mockSettings = Settings.builder().put(REMOTE_STORE_COMPATIBILITY_MODE_SETTING.getKey(), "strict").build(); + ClusterUpdateSettingsRequest request = new ClusterUpdateSettingsRequest(); + request.persistentSettings(mockSettings); + assertTrue(isSwitchToStrictCompatibilityMode(request)); + + mockSettings = Settings.builder().put(REMOTE_STORE_COMPATIBILITY_MODE_SETTING.getKey(), "mixed").build(); + request.persistentSettings(mockSettings); + assertFalse(isSwitchToStrictCompatibilityMode(request)); + } + + private void assertRemoteSettingsApplied(IndexMetadata indexMetadata) { + assertTrue(IndexMetadata.INDEX_REMOTE_STORE_ENABLED_SETTING.get(indexMetadata.getSettings())); + assertTrue(IndexMetadata.INDEX_REMOTE_TRANSLOG_REPOSITORY_SETTING.exists(indexMetadata.getSettings())); + assertTrue(IndexMetadata.INDEX_REMOTE_SEGMENT_STORE_REPOSITORY_SETTING.exists(indexMetadata.getSettings())); + assertEquals(ReplicationType.SEGMENT, IndexMetadata.INDEX_REPLICATION_TYPE_SETTING.get(indexMetadata.getSettings())); + } + + private void assertDocrepSettingsApplied(IndexMetadata indexMetadata) { + assertFalse(IndexMetadata.INDEX_REMOTE_STORE_ENABLED_SETTING.get(indexMetadata.getSettings())); + assertFalse(IndexMetadata.INDEX_REMOTE_TRANSLOG_REPOSITORY_SETTING.exists(indexMetadata.getSettings())); + assertFalse(IndexMetadata.INDEX_REMOTE_SEGMENT_STORE_REPOSITORY_SETTING.exists(indexMetadata.getSettings())); + assertEquals(ReplicationType.DOCUMENT, IndexMetadata.INDEX_REPLICATION_TYPE_SETTING.get(indexMetadata.getSettings())); + } + + private RoutingTable createRoutingTableAllShardsStarted( + String indexName, + int numberOfShards, + int numberOfReplicas, + DiscoveryNode primaryHostingNode, + DiscoveryNode replicaHostingNode + ) { + RoutingTable.Builder builder = RoutingTable.builder(); + Index index = new Index(indexName, UUID.randomUUID().toString()); + + IndexRoutingTable.Builder indexRoutingTableBuilder = IndexRoutingTable.builder(index); + for (int i = 0; i < numberOfShards; i++) { + ShardId shardId = new ShardId(index, i); + IndexShardRoutingTable.Builder indexShardRoutingTable = new IndexShardRoutingTable.Builder(shardId); + indexShardRoutingTable.addShard( + TestShardRouting.newShardRouting(shardId, primaryHostingNode.getId(), true, ShardRoutingState.STARTED) + ); + for (int j = 0; j < numberOfReplicas; j++) { + indexShardRoutingTable.addShard( + TestShardRouting.newShardRouting(shardId, replicaHostingNode.getId(), false, ShardRoutingState.STARTED) + ); + } + indexRoutingTableBuilder.addIndexShard(indexShardRoutingTable.build()); + } + return builder.add(indexRoutingTableBuilder.build()).build(); + } + + private Map getRemoteStoreNodeAttributes() { + Map remoteStoreNodeAttributes = new HashMap<>(); + remoteStoreNodeAttributes.put(REMOTE_STORE_SEGMENT_REPOSITORY_NAME_ATTRIBUTE_KEY, "my-segment-repo-1"); + remoteStoreNodeAttributes.put(REMOTE_STORE_TRANSLOG_REPOSITORY_NAME_ATTRIBUTE_KEY, "my-translog-repo-1"); + return remoteStoreNodeAttributes; + } } diff --git a/server/src/test/java/org/opensearch/repositories/blobstore/BlobStoreRepositoryTests.java b/server/src/test/java/org/opensearch/repositories/blobstore/BlobStoreRepositoryTests.java index 2445cad01574c..bd47507da4863 100644 --- a/server/src/test/java/org/opensearch/repositories/blobstore/BlobStoreRepositoryTests.java +++ b/server/src/test/java/org/opensearch/repositories/blobstore/BlobStoreRepositoryTests.java @@ -255,6 +255,28 @@ public void testBadChunksize() throws Exception { ); } + public void testPrefixModeVerification() throws Exception { + final Client client = client(); + final Path location = OpenSearchIntegTestCase.randomRepoPath(node().settings()); + final String repositoryName = "test-repo"; + AcknowledgedResponse putRepositoryResponse = client.admin() + .cluster() + .preparePutRepository(repositoryName) + .setType(REPO_TYPE) + .setSettings( + Settings.builder() + .put(node().settings()) + .put("location", location) + .put(BlobStoreRepository.PREFIX_MODE_VERIFICATION_SETTING.getKey(), true) + ) + .get(); + assertTrue(putRepositoryResponse.isAcknowledged()); + + final RepositoriesService repositoriesService = getInstanceFromNode(RepositoriesService.class); + final BlobStoreRepository repository = (BlobStoreRepository) repositoriesService.repository(repositoryName); + assertTrue(repository.getPrefixModeVerification()); + } + public void testFsRepositoryCompressDeprecatedIgnored() { final Path location = OpenSearchIntegTestCase.randomRepoPath(node().settings()); final Settings settings = Settings.builder().put(node().settings()).put("location", location).build(); diff --git a/server/src/test/java/org/opensearch/search/aggregations/bucket/histogram/InternalHistogramTests.java b/server/src/test/java/org/opensearch/search/aggregations/bucket/histogram/InternalHistogramTests.java index 288b22ccfcc92..98c6ac2b3de45 100644 --- a/server/src/test/java/org/opensearch/search/aggregations/bucket/histogram/InternalHistogramTests.java +++ b/server/src/test/java/org/opensearch/search/aggregations/bucket/histogram/InternalHistogramTests.java @@ -33,10 +33,15 @@ package org.opensearch.search.aggregations.bucket.histogram; import org.apache.lucene.tests.util.TestUtil; +import org.opensearch.core.common.breaker.CircuitBreaker; +import org.opensearch.core.common.breaker.CircuitBreakingException; import org.opensearch.search.DocValueFormat; import org.opensearch.search.aggregations.BucketOrder; +import org.opensearch.search.aggregations.InternalAggregation; import org.opensearch.search.aggregations.InternalAggregations; +import org.opensearch.search.aggregations.MultiBucketConsumerService; import org.opensearch.search.aggregations.ParsedMultiBucketAggregation; +import org.opensearch.search.aggregations.pipeline.PipelineAggregator; import org.opensearch.test.InternalAggregationTestCase; import org.opensearch.test.InternalMultiBucketAggregationTestCase; @@ -47,6 +52,8 @@ import java.util.Map; import java.util.TreeMap; +import org.mockito.Mockito; + public class InternalHistogramTests extends InternalMultiBucketAggregationTestCase { private boolean keyed; @@ -123,6 +130,42 @@ public void testHandlesNaN() { ); } + public void testCircuitBreakerWhenAddEmptyBuckets() { + String name = randomAlphaOfLength(5); + double interval = 1; + double lowerBound = 1; + double upperBound = 1026; + List bucket1 = List.of( + new InternalHistogram.Bucket(lowerBound, 1, false, format, InternalAggregations.EMPTY) + ); + List bucket2 = List.of( + new InternalHistogram.Bucket(upperBound, 1, false, format, InternalAggregations.EMPTY) + ); + BucketOrder order = BucketOrder.key(true); + InternalHistogram.EmptyBucketInfo emptyBucketInfo = new InternalHistogram.EmptyBucketInfo( + interval, + 0, + lowerBound, + upperBound, + InternalAggregations.EMPTY + ); + InternalHistogram histogram1 = new InternalHistogram(name, bucket1, order, 0, emptyBucketInfo, format, false, null); + InternalHistogram histogram2 = new InternalHistogram(name, bucket2, order, 0, emptyBucketInfo, format, false, null); + + CircuitBreaker breaker = Mockito.mock(CircuitBreaker.class); + Mockito.when(breaker.addEstimateBytesAndMaybeBreak(0, "allocated_buckets")).thenThrow(CircuitBreakingException.class); + + MultiBucketConsumerService.MultiBucketConsumer bucketConsumer = new MultiBucketConsumerService.MultiBucketConsumer(0, breaker); + InternalAggregation.ReduceContext reduceContext = InternalAggregation.ReduceContext.forFinalReduction( + null, + null, + bucketConsumer, + PipelineAggregator.PipelineTree.EMPTY + ); + expectThrows(CircuitBreakingException.class, () -> histogram1.reduce(List.of(histogram1, histogram2), reduceContext)); + Mockito.verify(breaker, Mockito.times(1)).addEstimateBytesAndMaybeBreak(0, "allocated_buckets"); + } + @Override protected void assertReduced(InternalHistogram reduced, List inputs) { TreeMap expectedCounts = new TreeMap<>(); diff --git a/test/framework/src/main/java/org/opensearch/test/OpenSearchIntegTestCase.java b/test/framework/src/main/java/org/opensearch/test/OpenSearchIntegTestCase.java index 7a50502e418e2..9853cef482254 100644 --- a/test/framework/src/main/java/org/opensearch/test/OpenSearchIntegTestCase.java +++ b/test/framework/src/main/java/org/opensearch/test/OpenSearchIntegTestCase.java @@ -152,6 +152,7 @@ import org.opensearch.node.remotestore.RemoteStoreNodeService; import org.opensearch.plugins.NetworkPlugin; import org.opensearch.plugins.Plugin; +import org.opensearch.repositories.blobstore.BlobStoreRepository; import org.opensearch.repositories.fs.ReloadableFsRepository; import org.opensearch.script.MockScriptService; import org.opensearch.search.MockSearchService; @@ -386,6 +387,8 @@ public abstract class OpenSearchIntegTestCase extends OpenSearchTestCase { protected static final String REMOTE_BACKED_STORAGE_REPOSITORY_NAME = "test-remote-store-repo"; + private static Boolean prefixModeVerificationEnable; + private Path remoteStoreRepositoryPath; private ReplicationType randomReplicationType; @@ -394,6 +397,7 @@ public abstract class OpenSearchIntegTestCase extends OpenSearchTestCase { @BeforeClass public static void beforeClass() throws Exception { + prefixModeVerificationEnable = randomBoolean(); testClusterRule.beforeClass(); } @@ -2645,16 +2649,21 @@ private static Settings buildRemoteStoreNodeAttributes( segmentRepoName ); + String prefixModeVerificationSuffix = BlobStoreRepository.PREFIX_MODE_VERIFICATION_SETTING.getKey(); + Settings.Builder settings = Settings.builder() .put("node.attr." + REMOTE_STORE_SEGMENT_REPOSITORY_NAME_ATTRIBUTE_KEY, segmentRepoName) .put(segmentRepoTypeAttributeKey, segmentRepoType) .put(segmentRepoSettingsAttributeKeyPrefix + "location", segmentRepoPath) + .put(segmentRepoSettingsAttributeKeyPrefix + prefixModeVerificationSuffix, prefixModeVerificationEnable) .put("node.attr." + REMOTE_STORE_TRANSLOG_REPOSITORY_NAME_ATTRIBUTE_KEY, translogRepoName) .put(translogRepoTypeAttributeKey, translogRepoType) .put(translogRepoSettingsAttributeKeyPrefix + "location", translogRepoPath) + .put(translogRepoSettingsAttributeKeyPrefix + prefixModeVerificationSuffix, prefixModeVerificationEnable) .put("node.attr." + REMOTE_STORE_CLUSTER_STATE_REPOSITORY_NAME_ATTRIBUTE_KEY, segmentRepoName) .put(stateRepoTypeAttributeKey, segmentRepoType) - .put(stateRepoSettingsAttributeKeyPrefix + "location", segmentRepoPath); + .put(stateRepoSettingsAttributeKeyPrefix + "location", segmentRepoPath) + .put(stateRepoSettingsAttributeKeyPrefix + prefixModeVerificationSuffix, prefixModeVerificationEnable); if (withRateLimiterAttributes) { settings.put(segmentRepoSettingsAttributeKeyPrefix + "compress", randomBoolean())