[proxy] Fix regression in logging

For some reason, `tracing::instrument` proc_macro doesn't always print
elements specified via `fields()` or even show that it's impossible
(e.g. there's no Display impl).

Work around this using the `?foo` notation.

Before:
2023-04-03T14:48:06.017504Z  INFO handle_client🤝 received SslRequest

After:
2023-04-03T14:51:24.424176Z  INFO handle_client{session_id=7bd07be8-3462-404e-8ccc-0a5332bf3ace}🤝 received SslRequest

.github/actions/allure-report/action.yml

@@ -15,32 +15,10 @@ outputs:
   report-url:
     description: 'Allure report URL'
     value: ${{ steps.generate-report.outputs.report-url }}
-  report-json-url:
-    description: 'Allure report JSON URL'
-    value: ${{ steps.generate-report.outputs.report-json-url }}
 
 runs:
   using: "composite"
-
   steps:
-    # We're using some of env variables quite offen, so let's set them once.
-    #
-    # It would be nice to have them set in common runs.env[0] section, but it doesn't work[1]
-    #
-    # - [0] https://docs.github.com/en/actions/creating-actions/metadata-syntax-for-github-actions#runsenv
-    # - [1] https://github.com/neondatabase/neon/pull/3907#discussion_r1154703456
-    #
-    - name: Set common environment variables
-      shell: bash -euxo pipefail {0}
-      run: |
-        echo "BUILD_TYPE=${BUILD_TYPE}"   >> $GITHUB_ENV
-        echo "BUCKET=${BUCKET}"           >> $GITHUB_ENV
-        echo "TEST_OUTPUT=${TEST_OUTPUT}" >> $GITHUB_ENV
-      env:
-        BUILD_TYPE: ${{ inputs.build_type }}
-        BUCKET: neon-github-public-dev
-        TEST_OUTPUT: /tmp/test_output
-
     - name: Validate input parameters
       shell: bash -euxo pipefail {0}
       run: |
@@ -98,14 +76,16 @@ runs:
           rm -f ${ALLURE_ZIP}
         fi
       env:
-        ALLURE_VERSION: 2.21.0
-        ALLURE_ZIP_MD5: c8db4dd8e2a7882583d569ed2c82879c
+        ALLURE_VERSION: 2.19.0
+        ALLURE_ZIP_MD5: ced21401a1a8b9dfb68cee9e4c210464
 
     - name: Upload Allure results
       if: ${{ inputs.action == 'store' }}
       env:
         REPORT_PREFIX: reports/${{ steps.calculate-vars.outputs.KEY }}/${{ inputs.build_type }}
         RAW_PREFIX: reports-raw/${{ steps.calculate-vars.outputs.KEY }}/${{ inputs.build_type }}
+        TEST_OUTPUT: /tmp/test_output
+        BUCKET: neon-github-public-dev
         TEST_SELECTION: ${{ steps.calculate-vars.outputs.TEST_SELECTION }}
       shell: bash -euxo pipefail {0}
       run: |
@@ -124,7 +104,7 @@ runs:
         EOF
         cat <<EOF > $TEST_OUTPUT/allure/results/environment.properties
           TEST_SELECTION=${{ inputs.test_selection }}
-          BUILD_TYPE=${BUILD_TYPE}
+          BUILD_TYPE=${{ inputs.build_type }}
         EOF
 
         ARCHIVE="${GITHUB_RUN_ID}-${TEST_SELECTION}-${GITHUB_RUN_ATTEMPT}-$(date +%s).tar.zst"
@@ -133,12 +113,13 @@ runs:
         tar -C ${TEST_OUTPUT}/allure/results -cf ${ARCHIVE} --zstd .
         aws s3 mv --only-show-errors ${ARCHIVE} "s3://${BUCKET}/${RAW_PREFIX}/${ARCHIVE}"
 
-    # Potentially we could have several running build for the same key (for example for the main branch), so we use improvised lock for this
+    # Potentially we could have several running build for the same key (for example for the main branch),  so we use improvised lock for this
     - name: Acquire Allure lock
       if: ${{ inputs.action == 'generate' }}
       shell: bash -euxo pipefail {0}
       env:
         LOCK_FILE: reports/${{ steps.calculate-vars.outputs.KEY }}/lock.txt
+        BUCKET: neon-github-public-dev
         TEST_SELECTION: ${{ steps.calculate-vars.outputs.TEST_SELECTION }}
       run: |
         LOCK_TIMEOUT=300 # seconds
@@ -168,6 +149,8 @@ runs:
       env:
         REPORT_PREFIX: reports/${{ steps.calculate-vars.outputs.KEY }}/${{ inputs.build_type }}
         RAW_PREFIX: reports-raw/${{ steps.calculate-vars.outputs.KEY }}/${{ inputs.build_type }}
+        TEST_OUTPUT: /tmp/test_output
+        BUCKET: neon-github-public-dev
       shell: bash -euxo pipefail {0}
       run: |
         # Get previously uploaded data for this run
@@ -203,24 +186,24 @@ runs:
         REPORT_URL=https://${BUCKET}.s3.amazonaws.com/${REPORT_PREFIX}/${GITHUB_RUN_ID}/index.html
 
         # Generate redirect
-        cat <<EOF > ${TEST_OUTPUT}/allure/index.html
+        cat <<EOF > ./index.html
           <!DOCTYPE html>
 
           <meta charset="utf-8">
           <title>Redirecting to ${REPORT_URL}</title>
           <meta http-equiv="refresh" content="0; URL=${REPORT_URL}">
         EOF
-        aws s3 cp --only-show-errors ${TEST_OUTPUT}/allure/index.html "s3://${BUCKET}/${REPORT_PREFIX}/latest/index.html"
+        aws s3 cp --only-show-errors ./index.html "s3://${BUCKET}/${REPORT_PREFIX}/latest/index.html"
 
         echo "[Allure Report](${REPORT_URL})" >> ${GITHUB_STEP_SUMMARY}
         echo "report-url=${REPORT_URL}" >> $GITHUB_OUTPUT
-        echo "report-json-url=${REPORT_URL%/index.html}/data/suites.json" >> $GITHUB_OUTPUT
 
     - name: Release Allure lock
       if: ${{ inputs.action == 'generate' && always() }}
       shell: bash -euxo pipefail {0}
       env:
         LOCK_FILE: reports/${{ steps.calculate-vars.outputs.KEY }}/lock.txt
+        BUCKET: neon-github-public-dev
         TEST_SELECTION: ${{ steps.calculate-vars.outputs.TEST_SELECTION }}
       run: |
         aws s3 cp --only-show-errors "s3://${BUCKET}/${LOCK_FILE}" ./lock.txt || exit 0
@@ -229,16 +212,11 @@ runs:
           aws s3 rm "s3://${BUCKET}/${LOCK_FILE}"
         fi
 
-    - name: Cleanup
-      if: always()
-      shell: bash -euxo pipefail {0}
-      run: |
-        rm -rf ${TEST_OUTPUT}/allure
-
     - uses: actions/github-script@v6
       if: ${{ inputs.action == 'generate' && always() }}
       env:
         REPORT_URL: ${{ steps.generate-report.outputs.report-url }}
+        BUILD_TYPE: ${{ inputs.build_type }}
         SHA: ${{ github.event.pull_request.head.sha || github.sha }}
       with:
         script: |

.github/actions/run-python-test-set/action.yml

@@ -44,10 +44,6 @@ inputs:
     description: 'Secret access key'
     required: false
     default: ''
-  rerun_flaky:
-    description: 'Whether to rerun flaky tests'
-    required: false
-    default: 'false'
 
 runs:
   using: "composite"
@@ -105,7 +101,6 @@ runs:
         COMPATIBILITY_SNAPSHOT_DIR: /tmp/compatibility_snapshot_pg14
         ALLOW_BACKWARD_COMPATIBILITY_BREAKAGE: contains(github.event.pull_request.labels.*.name, 'backward compatibility breakage')
         ALLOW_FORWARD_COMPATIBILITY_BREAKAGE: contains(github.event.pull_request.labels.*.name, 'forward compatibility breakage')
-        RERUN_FLAKY: ${{ inputs.rerun_flaky }}
       shell: bash -euxo pipefail {0}
       run: |
         # PLATFORM will be embedded in the perf test report
@@ -148,13 +143,6 @@ runs:
           EXTRA_PARAMS="--out-dir $PERF_REPORT_DIR $EXTRA_PARAMS"
         fi
 
-        if [ "${RERUN_FLAKY}" == "true" ]; then
-          mkdir -p $TEST_OUTPUT
-          poetry run ./scripts/flaky_tests.py "${TEST_RESULT_CONNSTR}" --days 10 --output "$TEST_OUTPUT/flaky.json"
-
-          EXTRA_PARAMS="--flaky-tests-json $TEST_OUTPUT/flaky.json $EXTRA_PARAMS"
-        fi
-
         if [[ "${{ inputs.build_type }}" == "debug" ]]; then
           cov_prefix=(scripts/coverage "--profraw-prefix=$GITHUB_JOB" --dir=/tmp/coverage run)
         elif [[ "${{ inputs.build_type }}" == "release" ]]; then

.github/workflows/build_and_test.yml

@@ -335,9 +335,6 @@ jobs:
           real_s3_region: us-west-2
           real_s3_access_key_id: "${{ secrets.AWS_ACCESS_KEY_ID_CI_TESTS_S3 }}"
           real_s3_secret_access_key: "${{ secrets.AWS_SECRET_ACCESS_KEY_CI_TESTS_S3 }}"
-          rerun_flaky: true
-        env:
-          TEST_RESULT_CONNSTR: ${{ secrets.REGRESS_TEST_RESULT_CONNSTR }}
 
       - name: Merge and upload coverage data
         if: matrix.build_type == 'debug'
@@ -374,88 +371,42 @@ jobs:
       # XXX: no coverage data handling here, since benchmarks are run on release builds,
       # while coverage is currently collected for the debug ones
 
-  create-test-report:
+  merge-allure-report:
     runs-on: [ self-hosted, gen3, small ]
     container:
       image: 369495373322.dkr.ecr.eu-central-1.amazonaws.com/rust:pinned
       options: --init
     needs: [ regress-tests, benchmarks ]
     if: ${{ !cancelled() }}
-
+    strategy:
+      fail-fast: false
+      matrix:
+        build_type: [ debug, release ]
     steps:
-      - uses: actions/checkout@v3
+      - name: Checkout
+        uses: actions/checkout@v3
+        with:
+          submodules: false
 
-      - name: Create Allure report (debug)
-        if: ${{ !cancelled() }}
-        id: create-allure-report-debug
+      - name: Create Allure report
+        id: create-allure-report
         uses: ./.github/actions/allure-report
         with:
           action: generate
-          build_type: debug
-
-      - name: Create Allure report (release)
-        if: ${{ !cancelled() }}
-        id: create-allure-report-release
-        uses: ./.github/actions/allure-report
-        with:
-          action: generate
-          build_type: release
-
-      - uses: actions/github-script@v6
-        if: >
-          !cancelled() &&
-          github.event_name == 'pull_request' && (
-            steps.create-allure-report-debug.outputs.report-url ||
-            steps.create-allure-report-release.outputs.report-url
-          )
-        with:
-          script: |
-            const reports = [{
-              buildType: "debug",
-              reportUrl: "${{ steps.create-allure-report-debug.outputs.report-url }}",
-              jsonUrl:   "${{ steps.create-allure-report-debug.outputs.report-json-url }}",
-            }, {
-              buildType: "release",
-              reportUrl: "${{ steps.create-allure-report-release.outputs.report-url }}",
-              jsonUrl:   "${{ steps.create-allure-report-release.outputs.report-json-url }}",
-            }]
-
-            const script = require("./scripts/pr-comment-test-report.js")
-            await script({
-              github,
-              context,
-              fetch,
-              reports,
-            })
+          build_type: ${{ matrix.build_type }}
 
       - name: Store Allure test stat in the DB
-        if: >
-          !cancelled() && (
-            steps.create-allure-report-debug.outputs.report-url ||
-            steps.create-allure-report-release.outputs.report-url
-          )
+        if: ${{ steps.create-allure-report.outputs.report-url }}
         env:
+          BUILD_TYPE: ${{ matrix.build_type }}
           SHA: ${{ github.event.pull_request.head.sha || github.sha }}
-          REPORT_JSON_URL_DEBUG: ${{ steps.create-allure-report-debug.outputs.report-json-url }}
-          REPORT_JSON_URL_RELEASE: ${{ steps.create-allure-report-release.outputs.report-json-url }}
+          REPORT_URL: ${{ steps.create-allure-report.outputs.report-url }}
           TEST_RESULT_CONNSTR: ${{ secrets.REGRESS_TEST_RESULT_CONNSTR }}
         run: |
+          curl --fail --output suites.json ${REPORT_URL%/index.html}/data/suites.json
           ./scripts/pysync
 
-          for report_url in $REPORT_JSON_URL_DEBUG $REPORT_JSON_URL_RELEASE; do
-            if [ -z "$report_url" ]; then
-              continue
-            fi
-
-            if [[ "$report_url" == "$REPORT_JSON_URL_DEBUG" ]]; then
-              BUILD_TYPE=debug
-            else
-              BUILD_TYPE=release
-            fi
-
-            curl --fail --output suites.json "${report_url}"
-            DATABASE_URL="$TEST_RESULT_CONNSTR" poetry run python3 scripts/ingest_regress_test_result.py --revision ${SHA} --reference ${GITHUB_REF} --build-type ${BUILD_TYPE} --ingest suites.json
-          done
+          DATABASE_URL="$TEST_RESULT_CONNSTR" poetry run python3 scripts/ingest_regress_test_result.py --revision ${SHA} --reference ${GITHUB_REF} --build-type ${BUILD_TYPE} --ingest suites.json
 
   coverage-report:
     runs-on: [ self-hosted, gen3, small ]

compute_tools/src/bin/compute_ctl.rs

@@ -34,23 +34,22 @@ use std::fs::File;
 use std::panic;
 use std::path::Path;
 use std::process::exit;
-use std::sync::{Arc, Condvar, Mutex};
+use std::sync::{Arc, RwLock};
 use std::{thread, time::Duration};
 
 use anyhow::{Context, Result};
 use chrono::Utc;
 use clap::Arg;
 use tracing::{error, info};
-use url::Url;
 
 use compute_tools::compute::{ComputeMetrics, ComputeNode, ComputeState, ComputeStatus};
-use compute_tools::configurator::launch_configurator;
 use compute_tools::http::api::launch_http_server;
 use compute_tools::logger::*;
 use compute_tools::monitor::launch_monitor;
 use compute_tools::params::*;
 use compute_tools::pg_helpers::*;
 use compute_tools::spec::*;
+use url::Url;
 
 fn main() -> Result<()> {
     init_tracing_and_logging(DEFAULT_LOG_LEVEL)?;
@@ -63,7 +62,7 @@ fn main() -> Result<()> {
     let connstr = matches
         .get_one::<String>("connstr")
         .expect("Postgres connection string is required");
-    let spec_json = matches.get_one::<String>("spec");
+    let spec = matches.get_one::<String>("spec");
     let spec_path = matches.get_one::<String>("spec-path");
 
     let compute_id = matches.get_one::<String>("compute-id");
@@ -72,110 +71,40 @@ fn main() -> Result<()> {
     // Try to use just 'postgres' if no path is provided
     let pgbin = matches.get_one::<String>("pgbin").unwrap();
 
-    let mut spec = Default::default();
-    let mut spec_set = false;
-    let mut live_config_allowed = false;
-    match spec_json {
+    let spec: ComputeSpec = match spec {
         // First, try to get cluster spec from the cli argument
-        Some(json) => {
-            spec = serde_json::from_str(json)?;
-            spec_set = true;
-        }
+        Some(json) => serde_json::from_str(json)?,
         None => {
             // Second, try to read it from the file if path is provided
             if let Some(sp) = spec_path {
                 let path = Path::new(sp);
                 let file = File::open(path)?;
-                spec = serde_json::from_reader(file)?;
-                spec_set = true;
+                serde_json::from_reader(file)?
             } else if let Some(id) = compute_id {
                 if let Some(cp_base) = control_plane_uri {
-                    live_config_allowed = true;
-                    if let Ok(s) = get_spec_from_control_plane(cp_base, id) {
-                        spec = s;
-                        spec_set = true;
-                    }
+                    let cp_uri = format!("{cp_base}/management/api/v1/{id}/spec");
+                    let jwt: String = match std::env::var("NEON_CONSOLE_JWT") {
+                        Ok(v) => v,
+                        Err(_) => "".to_string(),
+                    };
+
+                    reqwest::blocking::Client::new()
+                        .get(cp_uri)
+                        .header("Authorization", jwt)
+                        .send()?
+                        .json()?
                 } else {
-                    panic!("must specify both --control-plane-uri and --compute-id or none");
+                    panic!(
+                        "must specify --control-plane-uri \"{:#?}\" and --compute-id \"{:#?}\"",
+                        control_plane_uri, compute_id
+                    );
                 }
             } else {
-                panic!(
-                    "compute spec should be provided by one of the following ways: \
-                    --spec OR --spec-path OR --control-plane-uri and --compute-id"
-                );
+                panic!("compute spec should be provided via --spec or --spec-path argument");
             }
         }
     };
 
-    let mut new_state = ComputeState::new();
-    if spec_set {
-        new_state.spec = spec;
-    }
-    // Volatile compute state under mutex and condition variable to notify everyone
-    // who is interested in the state changes.
-    let compute_state = (Mutex::new(new_state), Condvar::new());
-    let compute_node = ComputeNode {
-        start_time: Utc::now(),
-        connstr: Url::parse(connstr).context("cannot parse connstr as a URL")?,
-        pgdata: pgdata.to_string(),
-        pgbin: pgbin.to_string(),
-        live_config_allowed,
-        metrics: ComputeMetrics::default(),
-        state: compute_state,
-    };
-    let compute = Arc::new(compute_node);
-
-    // Launch http service first, so we were able to serve control-plane
-    // requests, while configuration is still in progress.
-    let _http_handle = launch_http_server(&compute).expect("cannot launch http endpoint thread");
-
-    if !spec_set {
-        // No spec provided, hang waiting for it.
-        info!("no compute spec provided, waiting");
-        let (state, state_changed) = &compute.state;
-        let mut state = state.lock().unwrap();
-        while state.status != ComputeStatus::ConfigurationPending {
-            state = state_changed.wait(state).unwrap();
-
-            if state.status == ComputeStatus::ConfigurationPending {
-                info!("got spec, continue configuration");
-                // Spec is already set by the http server handler.
-                break;
-            }
-        }
-    }
-
-    // We got all we need, fill in the state.
-    let (state, _) = &compute.state;
-    let mut state = state.lock().unwrap();
-    let pageserver_connstr = state
-        .spec
-        .cluster
-        .settings
-        .find("neon.pageserver_connstring")
-        .expect("pageserver connstr should be provided");
-    let storage_auth_token = state.spec.storage_auth_token.clone();
-    let tenant = state
-        .spec
-        .cluster
-        .settings
-        .find("neon.tenant_id")
-        .expect("tenant id should be provided");
-    let timeline = state
-        .spec
-        .cluster
-        .settings
-        .find("neon.timeline_id")
-        .expect("tenant id should be provided");
-    let startup_tracing_context = state.spec.startup_tracing_context.clone();
-
-    state.pageserver_connstr = pageserver_connstr;
-    state.storage_auth_token = storage_auth_token;
-    state.tenant = tenant;
-    state.timeline = timeline;
-    state.status = ComputeStatus::Init;
-    drop(state);
-
     // Extract OpenTelemetry context for the startup actions from the spec, and
     // attach it to the current tracing context.
     //
@@ -191,7 +120,7 @@ fn main() -> Result<()> {
     // postgres is configured and up-and-running, we exit this span. Any other
     // actions that are performed on incoming HTTP requests, for example, are
     // performed in separate spans.
-    let startup_context_guard = if let Some(ref carrier) = startup_tracing_context {
+    let startup_context_guard = if let Some(ref carrier) = spec.startup_tracing_context {
         use opentelemetry::propagation::TextMapPropagator;
         use opentelemetry::sdk::propagation::TraceContextPropagator;
         Some(TraceContextPropagator::new().extract(carrier).attach())
@@ -199,10 +128,42 @@ fn main() -> Result<()> {
         None
     };
 
-    // Launch remaining service threads
+    let pageserver_connstr = spec
+        .cluster
+        .settings
+        .find("neon.pageserver_connstring")
+        .expect("pageserver connstr should be provided");
+    let storage_auth_token = spec.storage_auth_token.clone();
+    let tenant = spec
+        .cluster
+        .settings
+        .find("neon.tenant_id")
+        .expect("tenant id should be provided");
+    let timeline = spec
+        .cluster
+        .settings
+        .find("neon.timeline_id")
+        .expect("tenant id should be provided");
+
+    let compute_state = ComputeNode {
+        start_time: Utc::now(),
+        connstr: Url::parse(connstr).context("cannot parse connstr as a URL")?,
+        pgdata: pgdata.to_string(),
+        pgbin: pgbin.to_string(),
+        spec,
+        tenant,
+        timeline,
+        pageserver_connstr,
+        storage_auth_token,
+        metrics: ComputeMetrics::default(),
+        state: RwLock::new(ComputeState::new()),
+    };
+    let compute = Arc::new(compute_state);
+
+    // Launch service threads first, so we were able to serve availability
+    // requests, while configuration is still in progress.
+    let _http_handle = launch_http_server(&compute).expect("cannot launch http endpoint thread");
     let _monitor_handle = launch_monitor(&compute).expect("cannot launch compute monitor thread");
-    let _configurator_handle =
-        launch_configurator(&compute).expect("cannot launch configurator thread");
 
     // Start Postgres
     let mut delay_exit = false;
@@ -211,8 +172,7 @@ fn main() -> Result<()> {
         Ok(pg) => Some(pg),
         Err(err) => {
             error!("could not start the compute node: {:?}", err);
-            let (state, _) = &compute.state;
-            let mut state = state.lock().unwrap();
+            let mut state = compute.state.write().unwrap();
             state.error = Some(format!("{:?}", err));
             state.status = ComputeStatus::Failed;
             drop(state);
@@ -302,7 +262,7 @@ fn cli() -> clap::Command {
             Arg::new("control-plane-uri")
                 .short('p')
                 .long("control-plane-uri")
-                .value_name("CONTROL_PLANE_API_BASE_URI"),
+                .value_name("CONTROL_PLANE"),
         )
 }
 

compute_tools/src/compute.rs

@@ -20,8 +20,7 @@ use std::path::Path;
 use std::process::{Command, Stdio};
 use std::str::FromStr;
 use std::sync::atomic::{AtomicU64, Ordering};
-// use std::sync::RwLock;
-use std::sync::{Condvar, Mutex};
+use std::sync::RwLock;
 
 use anyhow::{Context, Result};
 use chrono::{DateTime, Utc};
@@ -42,22 +41,16 @@ pub struct ComputeNode {
     pub connstr: url::Url,
     pub pgdata: String,
     pub pgbin: String,
+    pub spec: ComputeSpec,
+    pub tenant: String,
+    pub timeline: String,
+    pub pageserver_connstr: String,
+    pub storage_auth_token: Option<String>,
     pub metrics: ComputeMetrics,
-    // We only allow live re- / configuration of the compute node if
-    // it uses 'pull model', i.e. it can go to control-plane and fetch
-    // the latest configuration. Otherwise, there could be a case:
-    // - we start compute with some spec provided as argument
-    // - we push new spec and it does reconfiguration
-    // - but then something happens and compute pod / VM is destroyed,
-    //   so k8s controller starts it again with the **old** spec
-    pub live_config_allowed: bool,
-    /// Volatile part of the `ComputeNode`, which should be used under `Mutex`.
-    /// Coupled with `Condvar` to allow notifying HTTP API and configurator
-    /// thread about state changes. To allow HTTP API server to serving status
-    /// requests, while configuration is in progress, lock should be held only
-    /// for short periods of time to do read/write, not the whole configuration
-    /// process.
-    pub state: (Mutex<ComputeState>, Condvar),
+    /// Volatile part of the `ComputeNode` so should be used under `RwLock`
+    /// to allow HTTP API server to serve status requests, while configuration
+    /// is in progress.
+    pub state: RwLock<ComputeState>,
 }
 
 fn rfc3339_serialize<S>(x: &DateTime<Utc>, s: S) -> Result<S::Ok, S::Error>
@@ -67,7 +60,7 @@ where
     x.to_rfc3339().serialize(s)
 }
 
-#[derive(Serialize, Clone)]
+#[derive(Serialize)]
 #[serde(rename_all = "snake_case")]
 pub struct ComputeState {
     pub status: ComputeStatus,
@@ -75,27 +68,14 @@ pub struct ComputeState {
     #[serde(serialize_with = "rfc3339_serialize")]
     pub last_active: DateTime<Utc>,
     pub error: Option<String>,
-    #[serde(skip_serializing)]
-    pub spec: ComputeSpec,
-    pub tenant: String,
-    pub timeline: String,
-    #[serde(skip_serializing)]
-    pub pageserver_connstr: String,
-    #[serde(skip_serializing)]
-    pub storage_auth_token: Option<String>,
 }
 
 impl ComputeState {
     pub fn new() -> Self {
         Self {
-            status: ComputeStatus::Empty,
+            status: ComputeStatus::Init,
             last_active: Utc::now(),
             error: None,
-            spec: ComputeSpec::default(),
-            tenant: String::new(),
-            timeline: String::new(),
-            pageserver_connstr: String::new(),
-            storage_auth_token: None,
         }
     }
 }
@@ -106,25 +86,12 @@ impl Default for ComputeState {
     }
 }
 
-#[derive(Serialize, Clone, Copy, PartialEq, Eq, Debug)]
+#[derive(Serialize, Clone, Copy, PartialEq, Eq)]
 #[serde(rename_all = "snake_case")]
 pub enum ComputeStatus {
-    // Spec wasn't provided as start, waiting for it to be
-    // provided by control-plane.
-    Empty,
-    // Compute node has spec and initial startup and
-    // configuration is in progress.
     Init,
-    // Compute is configured and running.
     Running,
-    // Either startup or configuration failed,
-    // compute will exit soon or is waiting for
-    // control-plane to terminate it.
     Failed,
-    // Control-plane requested reconfiguration.
-    ConfigurationPending,
-    // New spec is being applied.
-    Configuration,
 }
 
 #[derive(Default, Serialize)]
@@ -137,16 +104,11 @@ pub struct ComputeMetrics {
 
 impl ComputeNode {
     pub fn set_status(&self, status: ComputeStatus) {
-        let (state, state_changed) = &self.state;
-        let mut state = state.lock().unwrap();
-        state.status = status;
-        state_changed.notify_all();
+        self.state.write().unwrap().status = status;
     }
 
     pub fn get_status(&self) -> ComputeStatus {
-        // self.state.read().unwrap().status
-        let (state, _) = &self.state;
-        state.lock().unwrap().status
+        self.state.read().unwrap().status
     }
 
     // Remove `pgdata` directory and create it again with right permissions.
@@ -162,15 +124,15 @@ impl ComputeNode {
 
     // Get basebackup from the libpq connection to pageserver using `connstr` and
     // unarchive it to `pgdata` directory overriding all its previous content.
-    #[instrument(skip(self, compute_state))]
-    fn get_basebackup(&self, compute_state: &ComputeState, lsn: &str) -> Result<()> {
+    #[instrument(skip(self))]
+    fn get_basebackup(&self, lsn: &str) -> Result<()> {
         let start_time = Utc::now();
 
-        let mut config = postgres::Config::from_str(&compute_state.pageserver_connstr)?;
+        let mut config = postgres::Config::from_str(&self.pageserver_connstr)?;
 
         // Use the storage auth token from the config file, if given.
         // Note: this overrides any password set in the connection string.
-        if let Some(storage_auth_token) = &compute_state.storage_auth_token {
+        if let Some(storage_auth_token) = &self.storage_auth_token {
             info!("Got storage auth token from spec file");
             config.password(storage_auth_token);
         } else {
@@ -179,14 +141,8 @@ impl ComputeNode {
 
         let mut client = config.connect(NoTls)?;
         let basebackup_cmd = match lsn {
-            "0/0" => format!(
-                "basebackup {} {}",
-                &compute_state.tenant, &compute_state.timeline
-            ), // First start of the compute
-            _ => format!(
-                "basebackup {} {} {}",
-                &compute_state.tenant, &compute_state.timeline, lsn
-            ),
+            "0/0" => format!("basebackup {} {}", &self.tenant, &self.timeline), // First start of the compute
+            _ => format!("basebackup {} {} {}", &self.tenant, &self.timeline, lsn),
         };
         let copyreader = client.copy_out(basebackup_cmd.as_str())?;
 
@@ -213,14 +169,14 @@ impl ComputeNode {
 
     // Run `postgres` in a special mode with `--sync-safekeepers` argument
     // and return the reported LSN back to the caller.
-    #[instrument(skip(self, storage_auth_token))]
-    fn sync_safekeepers(&self, storage_auth_token: Option<String>) -> Result<String> {
+    #[instrument(skip(self))]
+    fn sync_safekeepers(&self) -> Result<String> {
         let start_time = Utc::now();
 
         let sync_handle = Command::new(&self.pgbin)
             .args(["--sync-safekeepers"])
             .env("PGDATA", &self.pgdata) // we cannot use -D in this mode
-            .envs(if let Some(storage_auth_token) = &storage_auth_token {
+            .envs(if let Some(storage_auth_token) = &self.storage_auth_token {
                 vec![("NEON_AUTH_TOKEN", storage_auth_token)]
             } else {
                 vec![]
@@ -261,9 +217,9 @@ impl ComputeNode {
 
     /// Do all the preparations like PGDATA directory creation, configuration,
     /// safekeepers sync, basebackup, etc.
-    #[instrument(skip(self, compute_state))]
-    pub fn prepare_pgdata(&self, compute_state: &ComputeState) -> Result<()> {
-        let spec = &compute_state.spec;
+    #[instrument(skip(self))]
+    pub fn prepare_pgdata(&self) -> Result<()> {
+        let spec = &self.spec;
         let pgdata_path = Path::new(&self.pgdata);
 
         // Remove/create an empty pgdata directory and put configuration there.
@@ -272,18 +228,18 @@ impl ComputeNode {
 
         info!("starting safekeepers syncing");
         let lsn = self
-            .sync_safekeepers(compute_state.storage_auth_token.clone())
+            .sync_safekeepers()
             .with_context(|| "failed to sync safekeepers")?;
         info!("safekeepers synced at LSN {}", lsn);
 
         info!(
             "getting basebackup@{} from pageserver {}",
-            lsn, &compute_state.pageserver_connstr
+            lsn, &self.pageserver_connstr
         );
-        self.get_basebackup(compute_state, &lsn).with_context(|| {
+        self.get_basebackup(&lsn).with_context(|| {
             format!(
                 "failed to get basebackup@{} from pageserver {}",
-                lsn, &compute_state.pageserver_connstr
+                lsn, &self.pageserver_connstr
             )
         })?;
 
@@ -296,16 +252,13 @@ impl ComputeNode {
     /// Start Postgres as a child process and manage DBs/roles.
     /// After that this will hang waiting on the postmaster process to exit.
     #[instrument(skip(self))]
-    pub fn start_postgres(
-        &self,
-        storage_auth_token: Option<String>,
-    ) -> Result<std::process::Child> {
+    pub fn start_postgres(&self) -> Result<std::process::Child> {
         let pgdata_path = Path::new(&self.pgdata);
 
         // Run postgres as a child process.
         let mut pg = Command::new(&self.pgbin)
             .args(["-D", &self.pgdata])
-            .envs(if let Some(storage_auth_token) = &storage_auth_token {
+            .envs(if let Some(storage_auth_token) = &self.storage_auth_token {
                 vec![("NEON_AUTH_TOKEN", storage_auth_token)]
             } else {
                 vec![]
@@ -318,9 +271,8 @@ impl ComputeNode {
         Ok(pg)
     }
 
-    /// Do initial configuration of the already started Postgres.
-    #[instrument(skip(self, compute_state))]
-    pub fn apply_config(&self, compute_state: &ComputeState) -> Result<()> {
+    #[instrument(skip(self))]
+    pub fn apply_config(&self) -> Result<()> {
         // If connection fails,
         // it may be the old node with `zenith_admin` superuser.
         //
@@ -351,62 +303,19 @@ impl ComputeNode {
         };
 
         // Proceed with post-startup configuration. Note, that order of operations is important.
-        handle_roles(&compute_state.spec, &mut client)?;
-        handle_databases(&compute_state.spec, &mut client)?;
-        handle_role_deletions(&compute_state.spec, self.connstr.as_str(), &mut client)?;
-        handle_grants(&compute_state.spec, self.connstr.as_str(), &mut client)?;
+        handle_roles(&self.spec, &mut client)?;
+        handle_databases(&self.spec, &mut client)?;
+        handle_role_deletions(self, &mut client)?;
+        handle_grants(self, &mut client)?;
         create_writability_check_data(&mut client)?;
-        handle_extensions(&compute_state.spec, &mut client)?;
+        handle_extensions(&self.spec, &mut client)?;
 
         // 'Close' connection
         drop(client);
 
         info!(
             "finished configuration of compute for project {}",
-            compute_state.spec.cluster.cluster_id
-        );
-
-        Ok(())
-    }
-
-    // We could've wrapped this around `pg_ctl reload`, but right now we don't use
-    // `pg_ctl` for start / stop, so this just seems much easier to do as we already
-    // have opened connection to Postgres and superuser access.
-    #[instrument(skip(self, client))]
-    fn pg_reload_conf(&self, client: &mut Client) -> Result<()> {
-        client.simple_query("SELECT pg_reload_conf()")?;
-        Ok(())
-    }
-
-    /// Similar to `apply_config()`, but does a bit different sequence of operations,
-    /// as it's used to reconfigure a previously started and configured Postgres node.
-    #[instrument(skip(self))]
-    pub fn reconfigure(&self) -> Result<()> {
-        let (state, _) = &self.state;
-        let spec = state.lock().unwrap().spec.clone();
-
-        // Write new config
-        let pgdata_path = Path::new(&self.pgdata);
-        config::write_postgres_conf(&pgdata_path.join("postgresql.conf"), &spec)?;
-
-        let mut client = Client::connect(self.connstr.as_str(), NoTls)?;
-        self.pg_reload_conf(&mut client)?;
-
-        // Proceed with post-startup configuration. Note, that order of operations is important.
-        handle_roles(&spec, &mut client)?;
-        handle_databases(&spec, &mut client)?;
-        handle_role_deletions(&spec, self.connstr.as_str(), &mut client)?;
-        handle_grants(&spec, self.connstr.as_str(), &mut client)?;
-        handle_extensions(&spec, &mut client)?;
-
-        // 'Close' connection
-        drop(client);
-
-        let unknown_op = "unknown".to_string();
-        let op_id = spec.operation_uuid.as_ref().unwrap_or(&unknown_op);
-        info!(
-            "finished reconfiguration of compute node for operation {}",
-            op_id
+            self.spec.cluster.cluster_id
         );
 
         Ok(())
@@ -414,24 +323,21 @@ impl ComputeNode {
 
     #[instrument(skip(self))]
     pub fn start_compute(&self) -> Result<std::process::Child> {
-        // let compute_state = self.state.read().unwrap().clone();
-        let (state, _) = &self.state;
-        let compute_state = state.lock().unwrap().clone();
         info!(
             "starting compute for project {}, operation {}, tenant {}, timeline {}",
-            compute_state.spec.cluster.cluster_id,
-            compute_state.spec.operation_uuid.as_ref().unwrap(),
-            compute_state.tenant,
-            compute_state.timeline,
+            self.spec.cluster.cluster_id,
+            self.spec.operation_uuid.as_ref().unwrap(),
+            self.tenant,
+            self.timeline,
         );
 
-        self.prepare_pgdata(&compute_state)?;
+        self.prepare_pgdata()?;
 
         let start_time = Utc::now();
 
-        let pg = self.start_postgres(compute_state.storage_auth_token.clone())?;
+        let pg = self.start_postgres()?;
 
-        self.apply_config(&compute_state)?;
+        self.apply_config()?;
 
         let startup_end_time = Utc::now();
         self.metrics.config_ms.store(

compute_tools/src/configurator.rs

@@ -1,53 +0,0 @@
-use std::sync::Arc;
-use std::thread;
-
-use anyhow::Result;
-use tracing::{error, info, instrument};
-
-use crate::compute::{ComputeNode, ComputeStatus};
-
-#[instrument(skip(compute))]
-fn configurator_main_loop(compute: &Arc<ComputeNode>) {
-    info!("waiting for reconfiguration requests");
-    let (state, state_changed) = &compute.state;
-    loop {
-        let state = state.lock().unwrap();
-        let mut state = state_changed.wait(state).unwrap();
-
-        if state.status == ComputeStatus::ConfigurationPending {
-            info!("got configuration request");
-            state.status = ComputeStatus::Configuration;
-            state_changed.notify_all();
-            drop(state);
-
-            let mut new_status = ComputeStatus::Failed;
-            if let Err(e) = compute.reconfigure() {
-                error!("could not configure compute node: {}", e);
-            } else {
-                new_status = ComputeStatus::Running;
-                info!("compute node configured");
-            }
-
-            // XXX: used to test that API is blocking
-            // std::thread::sleep(std::time::Duration::from_millis(2000));
-
-            compute.set_status(new_status);
-        } else if state.status == ComputeStatus::Failed {
-            info!("compute node is now in Failed state, exiting");
-            break;
-        } else {
-            info!("woken up for compute status: {:?}, sleeping", state.status);
-        }
-    }
-}
-
-pub fn launch_configurator(compute: &Arc<ComputeNode>) -> Result<thread::JoinHandle<()>> {
-    let compute = Arc::clone(compute);
-
-    Ok(thread::Builder::new()
-        .name("compute-configurator".into())
-        .spawn(move || {
-            configurator_main_loop(&compute);
-            info!("configurator thread is exited");
-        })?)
-}

compute_tools/src/http/api.rs

@@ -3,9 +3,7 @@ use std::net::SocketAddr;
 use std::sync::Arc;
 use std::thread;
 
-use crate::compute::{ComputeNode, ComputeStatus};
-use crate::http::models::{ConfigurationRequest, GenericAPIError};
-
+use crate::compute::ComputeNode;
 use anyhow::Result;
 use hyper::service::{make_service_fn, service_fn};
 use hyper::{Body, Method, Request, Response, Server, StatusCode};
@@ -25,9 +23,7 @@ async fn routes(req: Request<Body>, compute: &Arc<ComputeNode>) -> Response<Body
         // Serialized compute state.
         (&Method::GET, "/status") => {
             info!("serving /status GET request");
-            // let state = compute.state.read().unwrap();
-            let (state, _) = &compute.state;
-            let state = state.lock().unwrap();
+            let state = compute.state.read().unwrap();
             Response::new(Body::from(serde_json::to_string(&*state).unwrap()))
         }
 
@@ -41,29 +37,12 @@ async fn routes(req: Request<Body>, compute: &Arc<ComputeNode>) -> Response<Body
         // Collect Postgres current usage insights
         (&Method::GET, "/insights") => {
             info!("serving /insights GET request");
-            let status = compute.get_status();
-            if status != ComputeStatus::Running {
-                let msg = format!("compute is not running, current status: {:?}", status);
-                error!(msg);
-                return Response::new(Body::from(msg));
-            }
-
             let insights = compute.collect_insights().await;
             Response::new(Body::from(insights))
         }
 
         (&Method::POST, "/check_writability") => {
             info!("serving /check_writability POST request");
-            let status = compute.get_status();
-            if status != ComputeStatus::Running {
-                let msg = format!(
-                    "invalid compute status for check_writability request: {:?}",
-                    status
-                );
-                error!(msg);
-                return Response::new(Body::from(msg));
-            }
-
             let res = crate::checker::check_writability(compute).await;
             match res {
                 Ok(_) => Response::new(Body::from("true")),
@@ -82,61 +61,6 @@ async fn routes(req: Request<Body>, compute: &Arc<ComputeNode>) -> Response<Body
             ))
         }
 
-        // Accept spec in JSON format and request compute configuration from
-        // the configurator thread. If anything goes wrong after we set the
-        // compute state to `ConfigurationPending` and / or sent spec to the
-        // configurator thread, we basically leave compute in the potentially
-        // wrong state. That said, it's control-plane's responsibility to
-        // watch compute state after reconfiguration request and to clean
-        // restart in case of errors.
-        (&Method::POST, "/configure") => {
-            info!("serving /configure POST request");
-            if !compute.live_config_allowed {
-                let msg = "live reconfiguration is not allowed for this compute node";
-                error!(msg);
-                return render_json_error(msg, StatusCode::PRECONDITION_FAILED);
-            }
-
-            let body_bytes = hyper::body::to_bytes(req.into_body()).await.unwrap();
-            let spec_raw = String::from_utf8(body_bytes.to_vec()).unwrap();
-            if let Ok(request) = serde_json::from_str::<ConfigurationRequest>(&spec_raw) {
-                let spec = request.spec;
-                let (state, state_changed) = &compute.state;
-                let mut state = state.lock().unwrap();
-                if !(state.status == ComputeStatus::Empty || state.status == ComputeStatus::Running)
-                {
-                    let msg = format!(
-                        "invalid compute status for reconfiguration request: {}",
-                        serde_json::to_string(&*state).unwrap()
-                    );
-                    error!(msg);
-                    return render_json_error(&msg, StatusCode::PRECONDITION_FAILED);
-                }
-                state.spec = spec;
-                state.status = ComputeStatus::ConfigurationPending;
-                state_changed.notify_all();
-                drop(state);
-                info!("set new spec and notified configurator");
-
-                let (state, state_changed) = &compute.state;
-                let mut state = state.lock().unwrap();
-                while state.status != ComputeStatus::Running {
-                    state = state_changed.wait(state).unwrap();
-                    info!(
-                        "waiting for compute to become Running, current status: {:?}",
-                        state.status
-                    );
-                }
-
-                // Return current compute state if everything went well.
-                Response::new(Body::from(serde_json::to_string(&*state).unwrap()))
-            } else {
-                let msg = "invalid spec";
-                error!(msg);
-                render_json_error(msg, StatusCode::BAD_REQUEST)
-            }
-        }
-
         // Return the `404 Not Found` for any other routes.
         _ => {
             let mut not_found = Response::new(Body::from("404 Not Found"));
@@ -146,16 +70,6 @@ async fn routes(req: Request<Body>, compute: &Arc<ComputeNode>) -> Response<Body
     }
 }
 
-fn render_json_error(e: &str, status: StatusCode) -> Response<Body> {
-    let error = GenericAPIError {
-        error: e.to_string(),
-    };
-    Response::builder()
-        .status(status)
-        .body(Body::from(serde_json::to_string(&error).unwrap()))
-        .unwrap()
-}
-
 // Main Hyper HTTP server function that runs it and blocks waiting on it forever.
 #[tokio::main]
 async fn serve(state: Arc<ComputeNode>) {
@@ -196,6 +110,7 @@ async fn serve(state: Arc<ComputeNode>) {
 /// Launch a separate Hyper HTTP API server thread and return its `JoinHandle`.
 pub fn launch_http_server(state: &Arc<ComputeNode>) -> Result<thread::JoinHandle<()>> {
     let state = Arc::clone(state);
+
     Ok(thread::Builder::new()
         .name("http-endpoint".into())
         .spawn(move || serve(state))?)

compute_tools/src/http/mod.rs

@@ -1,2 +1 @@
 pub mod api;
-pub mod models;

compute_tools/src/http/models.rs

@@ -1,16 +0,0 @@
-use serde::{Deserialize, Serialize};
-
-use crate::spec::ComputeSpec;
-
-/// We now pass only `spec` in the configuration request, but later we can
-/// extend it and something like `restart: bool` or something else. So put
-/// `spec` into a struct initially to be more flexible in the future.
-#[derive(Deserialize, Debug)]
-pub struct ConfigurationRequest {
-    pub spec: ComputeSpec,
-}
-
-#[derive(Serialize, Debug)]
-pub struct GenericAPIError {
-    pub error: String,
-}

compute_tools/src/http/openapi_spec.yaml

@@ -11,7 +11,7 @@ paths:
     get:
       tags:
       - Info
-      summary: Get compute node internal status.
+      summary: Get compute node internal status
       description: ""
       operationId: getComputeStatus
       responses:
@@ -26,7 +26,7 @@ paths:
     get:
       tags:
       - Info
-      summary: Get compute node startup metrics in JSON format.
+      summary: Get compute node startup metrics in JSON format
       description: ""
       operationId: getComputeMetricsJSON
       responses:
@@ -41,9 +41,9 @@ paths:
     get:
       tags:
       - Info
-      summary: Get current compute insights in JSON format.
+      summary: Get current compute insights in JSON format
       description: |
-        Note, that this doesn't include any historical data.
+        Note, that this doesn't include any historical data
       operationId: getComputeInsights
       responses:
         200:
@@ -56,12 +56,12 @@ paths:
   /info:
     get:
       tags:
-      - Info
-      summary: Get info about the compute pod / VM.
+      - "info"
+      summary: Get info about the compute Pod/VM
       description: ""
       operationId: getInfo
       responses:
-        200:
+        "200":
           description: Info
           content:
             application/json:
@@ -72,7 +72,7 @@ paths:
     post:
       tags:
       - Check
-      summary: Check that we can write new data on this compute.
+      summary: Check that we can write new data on this compute
       description: ""
       operationId: checkComputeWritability
       responses:
@@ -82,57 +82,9 @@ paths:
             text/plain:
               schema:
                 type: string
-                description: Error text or 'true' if check passed.
+                description: Error text or 'true' if check passed
                 example: "true"
 
-  /configure:
-    post:
-      tags:
-      - Configure
-      summary: Request compute node configuration.
-      description: |
-        This is a blocking API endpoint, i.e. it blocks waiting until
-        compute is finished configuration and is in `Running` state.
-        Optional non-blocking mode could be added later. Currently,
-        it's also assumed that reconfiguration doesn't require restart.
-      operationId: configureCompute
-      requestBody:
-        description: Configuration request.
-        required: true
-        content:
-          application/json:
-            schema:
-              type: object
-              required:
-                - spec
-              properties:
-                spec:
-                  # XXX: I don't want to explain current spec in the OpenAPI format,
-                  # as it could be changed really soon. Consider doing it later.
-                  type: object
-      responses:
-        200:
-          description: Compute configuration finished.
-          content:
-            application/json:
-              schema:
-                $ref: "#/components/schemas/ComputeState"
-        400:
-          description: Provided spec is invalid.
-          content:
-            application/json:
-              schema:
-                $ref: "#/components/schemas/GenericError"
-        412:
-          description: |
-            It's not possible to do live-configuration of the compute.
-            It's either in the wrong state, or compute doesn't use pull
-            mode of configuration.
-          content:
-            application/json:
-              schema:
-                $ref: "#/components/schemas/GenericError"
-
 components:
   securitySchemes:
     JWT:
@@ -143,7 +95,7 @@ components:
   schemas:
     ComputeMetrics:
       type: object
-      description: Compute startup metrics.
+      description: Compute startup metrics
       required:
         - sync_safekeepers_ms
         - basebackup_ms
@@ -161,7 +113,7 @@ components:
 
     Info:
       type: object
-      description: Information about VM/Pod.
+      description: Information about VM/Pod
       required:
         - num_cpus
       properties:
@@ -178,26 +130,17 @@ components:
           $ref: '#/components/schemas/ComputeStatus'
         last_active:
           type: string
-          description: The last detected compute activity timestamp in UTC and RFC3339 format.
+          description: The last detected compute activity timestamp in UTC and RFC3339 format
           example: "2022-10-12T07:20:50.52Z"
         error:
           type: string
-          description: Text of the error during compute startup, if any.
-          example: ""
-        tenant:
-          type: string
-          description: Identifier of the current tenant served by compute node, if any.
-          example: c9269c359e9a199fad1ea0981246a78f
-        timeline:
-          type: string
-          description: Identifier of the current timeline served by compute node, if any.
-          example: ece7de74d4b8cbe5433a68ce4d1b97b4
+          description: Text of the error during compute startup, if any
 
     ComputeInsights:
       type: object
       properties:
         pg_stat_statements:
-          description: Contains raw output from pg_stat_statements in JSON format.
+          description: Contains raw output from pg_stat_statements in JSON format
           type: array
           items:
             type: object
@@ -208,19 +151,6 @@ components:
         - init
         - failed
         - running
-      example: running
-
-    #
-    # Errors
-    #
-
-    GenericError:
-      type: object
-      required:
-        - error
-      properties:
-        error:
-          type: string
 
 security:
   - JWT: []

compute_tools/src/lib.rs

@@ -4,7 +4,6 @@
 //!
 pub mod checker;
 pub mod config;
-pub mod configurator;
 pub mod http;
 #[macro_use]
 pub mod logger;

compute_tools/src/monitor.rs

@@ -46,9 +46,7 @@ fn watch_compute_activity(compute: &ComputeNode) {
                             AND usename != 'cloud_admin';", // XXX: find a better way to filter other monitors?
                         &[],
                     );
-                // let mut last_active = compute.state.read().unwrap().last_active;
-                let (state, _) = &compute.state;
-                let mut last_active = state.lock().unwrap().last_active;
+                let mut last_active = compute.state.read().unwrap().last_active;
 
                 if let Ok(backs) = backends {
                     let mut idle_backs: Vec<DateTime<Utc>> = vec![];
@@ -89,8 +87,7 @@ fn watch_compute_activity(compute: &ComputeNode) {
                 }
 
                 // Update the last activity in the shared state if we got a more recent one.
-                let (state, _) = &compute.state;
-                let mut state = state.lock().unwrap();
+                let mut state = compute.state.write().unwrap();
                 if last_active > state.last_active {
                     state.last_active = last_active;
                     debug!("set the last compute activity time to: {}", last_active);

compute_tools/src/pg_helpers.rs

@@ -17,7 +17,7 @@ const POSTGRES_WAIT_TIMEOUT: Duration = Duration::from_millis(60 * 1000); // mil
 
 /// Rust representation of Postgres role info with only those fields
 /// that matter for us.
-#[derive(Clone, Deserialize, Debug)]
+#[derive(Clone, Deserialize)]
 pub struct Role {
     pub name: PgIdent,
     pub encrypted_password: Option<String>,
@@ -26,7 +26,7 @@ pub struct Role {
 
 /// Rust representation of Postgres database info with only those fields
 /// that matter for us.
-#[derive(Clone, Deserialize, Debug)]
+#[derive(Clone, Deserialize)]
 pub struct Database {
     pub name: PgIdent,
     pub owner: PgIdent,
@@ -36,7 +36,7 @@ pub struct Database {
 /// Common type representing both SQL statement params with or without value,
 /// like `LOGIN` or `OWNER username` in the `CREATE/ALTER ROLE`, and config
 /// options like `wal_level = logical`.
-#[derive(Clone, Deserialize, Debug)]
+#[derive(Clone, Deserialize)]
 pub struct GenericOption {
     pub name: String,
     pub value: Option<String>,

compute_tools/src/spec.rs

@@ -8,13 +8,14 @@ use postgres::{Client, NoTls};
 use serde::Deserialize;
 use tracing::{info, info_span, instrument, span_enabled, warn, Level};
 
+use crate::compute::ComputeNode;
 use crate::config;
 use crate::params::PG_HBA_ALL_MD5;
 use crate::pg_helpers::*;
 
 /// Cluster spec or configuration represented as an optional number of
 /// delta operations + final cluster state description.
-#[derive(Clone, Deserialize, Debug, Default)]
+#[derive(Clone, Deserialize)]
 pub struct ComputeSpec {
     pub format_version: f32,
     pub timestamp: String,
@@ -30,7 +31,7 @@ pub struct ComputeSpec {
 
 /// Cluster state seen from the perspective of the external tools
 /// like Rails web console.
-#[derive(Clone, Deserialize, Debug, Default)]
+#[derive(Clone, Deserialize)]
 pub struct Cluster {
     pub cluster_id: String,
     pub name: String,
@@ -46,36 +47,13 @@ pub struct Cluster {
 /// - DROP ROLE
 /// - ALTER ROLE name RENAME TO new_name
 /// - ALTER DATABASE name RENAME TO new_name
-#[derive(Clone, Deserialize, Debug)]
+#[derive(Clone, Deserialize)]
 pub struct DeltaOp {
     pub action: String,
     pub name: PgIdent,
     pub new_name: Option<PgIdent>,
 }
 
-/// Request spec from the control-plane by compute_id. If `NEON_CONSOLE_JWT`
-/// env variable is set, it will be used for authorization.
-pub fn get_spec_from_control_plane(base_uri: &str, compute_id: &str) -> Result<ComputeSpec> {
-    let cp_uri = format!("{base_uri}/management/api/v2/computes/{compute_id}/spec");
-    let jwt: String = match std::env::var("NEON_CONSOLE_JWT") {
-        Ok(v) => v,
-        Err(_) => "".to_string(),
-    };
-    info!("getting spec from control plane: {}", cp_uri);
-
-    // TODO: check the response. We should distinguish cases when it's
-    // - network error, then retry
-    // - no spec for compute yet, then wait
-    // - compute id is unknown or any other error, then bail out
-    let spec = reqwest::blocking::Client::new()
-        .get(cp_uri)
-        .header("Authorization", jwt)
-        .send()?
-        .json()?;
-
-    Ok(spec)
-}
-
 /// It takes cluster specification and does the following:
 /// - Serialize cluster config and put it into `postgresql.conf` completely rewriting the file.
 /// - Update `pg_hba.conf` to allow external connections.
@@ -248,8 +226,8 @@ pub fn handle_roles(spec: &ComputeSpec, client: &mut Client) -> Result<()> {
 
 /// Reassign all dependent objects and delete requested roles.
 #[instrument(skip_all)]
-pub fn handle_role_deletions(spec: &ComputeSpec, connstr: &str, client: &mut Client) -> Result<()> {
-    if let Some(ops) = &spec.delta_operations {
+pub fn handle_role_deletions(node: &ComputeNode, client: &mut Client) -> Result<()> {
+    if let Some(ops) = &node.spec.delta_operations {
         // First, reassign all dependent objects to db owners.
         info!("reassigning dependent objects of to-be-deleted roles");
 
@@ -266,7 +244,7 @@ pub fn handle_role_deletions(spec: &ComputeSpec, connstr: &str, client: &mut Cli
             // Check that role is still present in Postgres, as this could be a
             // restart with the same spec after role deletion.
             if op.action == "delete_role" && existing_roles.iter().any(|r| r.name == op.name) {
-                reassign_owned_objects(spec, connstr, &op.name)?;
+                reassign_owned_objects(node, &op.name)?;
             }
         }
 
@@ -290,10 +268,10 @@ pub fn handle_role_deletions(spec: &ComputeSpec, connstr: &str, client: &mut Cli
 }
 
 // Reassign all owned objects in all databases to the owner of the database.
-fn reassign_owned_objects(spec: &ComputeSpec, connstr: &str, role_name: &PgIdent) -> Result<()> {
-    for db in &spec.cluster.databases {
+fn reassign_owned_objects(node: &ComputeNode, role_name: &PgIdent) -> Result<()> {
+    for db in &node.spec.cluster.databases {
         if db.owner != *role_name {
-            let mut conf = Config::from_str(connstr)?;
+            let mut conf = Config::from_str(node.connstr.as_str())?;
             conf.dbname(&db.name);
 
             let mut client = conf.connect(NoTls)?;
@@ -438,7 +416,9 @@ pub fn handle_databases(spec: &ComputeSpec, client: &mut Client) -> Result<()> {
 /// Grant CREATE ON DATABASE to the database owner and do some other alters and grants
 /// to allow users creating trusted extensions and re-creating `public` schema, for example.
 #[instrument(skip_all)]
-pub fn handle_grants(spec: &ComputeSpec, connstr: &str, client: &mut Client) -> Result<()> {
+pub fn handle_grants(node: &ComputeNode, client: &mut Client) -> Result<()> {
+    let spec = &node.spec;
+
     info!("cluster spec grants:");
 
     // We now have a separate `web_access` role to connect to the database
@@ -470,8 +450,8 @@ pub fn handle_grants(spec: &ComputeSpec, connstr: &str, client: &mut Client) ->
     // Do some per-database access adjustments. We'd better do this at db creation time,
     // but CREATE DATABASE isn't transactional. So we cannot create db + do some grants
     // atomically.
-    for db in &spec.cluster.databases {
-        let mut conf = Config::from_str(connstr)?;
+    for db in &node.spec.cluster.databases {
+        let mut conf = Config::from_str(node.connstr.as_str())?;
         conf.dbname(&db.name);
 
         let mut db_client = conf.connect(NoTls)?;

libs/remote_storage/src/lib.rs

@@ -78,6 +78,9 @@ impl RemotePath {
 /// providing basic CRUD operations for storage files.
 #[async_trait::async_trait]
 pub trait RemoteStorage: Send + Sync + 'static {
+    /// Lists all items the storage has right now.
+    async fn list(&self) -> anyhow::Result<Vec<RemotePath>>;
+
     /// Lists all top level subdirectories for a given prefix
     /// Note: here we assume that if the prefix is passed it was obtained via remote_object_id
     /// which already takes into account any kind of global prefix (prefix_in_bucket for S3 or storage_root for LocalFS)

libs/remote_storage/src/local_fs.rs

@@ -73,8 +73,10 @@ impl LocalFs {
             Ok(None)
         }
     }
+}
 
-    #[cfg(test)]
+#[async_trait::async_trait]
+impl RemoteStorage for LocalFs {
     async fn list(&self) -> anyhow::Result<Vec<RemotePath>> {
         Ok(get_all_files(&self.storage_root, true)
             .await?
@@ -89,10 +91,7 @@ impl LocalFs {
             })
             .collect())
     }
-}
 
-#[async_trait::async_trait]
-impl RemoteStorage for LocalFs {
     async fn list_prefixes(
         &self,
         prefix: Option<&RemotePath>,

libs/remote_storage/src/s3_bucket.rs

@@ -275,6 +275,50 @@ impl<S: AsyncRead> AsyncRead for RatelimitedAsyncRead<S> {
 
 #[async_trait::async_trait]
 impl RemoteStorage for S3Bucket {
+    async fn list(&self) -> anyhow::Result<Vec<RemotePath>> {
+        let mut document_keys = Vec::new();
+
+        let mut continuation_token = None;
+        loop {
+            let _guard = self
+                .concurrency_limiter
+                .acquire()
+                .await
+                .context("Concurrency limiter semaphore got closed during S3 list")?;
+
+            metrics::inc_list_objects();
+
+            let fetch_response = self
+                .client
+                .list_objects_v2()
+                .bucket(self.bucket_name.clone())
+                .set_prefix(self.prefix_in_bucket.clone())
+                .delimiter(REMOTE_STORAGE_PREFIX_SEPARATOR.to_string())
+                .set_continuation_token(continuation_token)
+                .set_max_keys(self.max_keys_per_list_response)
+                .send()
+                .await
+                .map_err(|e| {
+                    metrics::inc_list_objects_fail();
+                    e
+                })?;
+            document_keys.extend(
+                fetch_response
+                    .contents
+                    .unwrap_or_default()
+                    .into_iter()
+                    .filter_map(|o| Some(self.s3_object_to_relative_path(o.key()?))),
+            );
+
+            match fetch_response.next_continuation_token {
+                Some(new_token) => continuation_token = Some(new_token),
+                None => break,
+            }
+        }
+
+        Ok(document_keys)
+    }
+
     /// See the doc for `RemoteStorage::list_prefixes`
     /// Note: it wont include empty "directories"
     async fn list_prefixes(

libs/remote_storage/src/simulate_failures.rs

@@ -20,6 +20,7 @@ pub struct UnreliableWrapper {
 /// Used to identify retries of different unique operation.
 #[derive(Debug, Hash, Eq, PartialEq)]
 enum RemoteOp {
+    List,
     ListPrefixes(Option<RemotePath>),
     Upload(RemotePath),
     Download(RemotePath),
@@ -74,6 +75,12 @@ impl UnreliableWrapper {
 
 #[async_trait::async_trait]
 impl RemoteStorage for UnreliableWrapper {
+    /// Lists all items the storage has right now.
+    async fn list(&self) -> anyhow::Result<Vec<RemotePath>> {
+        self.attempt(RemoteOp::List)?;
+        self.inner.list().await
+    }
+
     async fn list_prefixes(
         &self,
         prefix: Option<&RemotePath>,

poetry.lock

@@ -1,4 +1,4 @@
-# This file is automatically @generated by Poetry 1.4.1 and should not be changed by hand.
+# This file is automatically @generated by Poetry 1.4.0 and should not be changed by hand.
 
 [[package]]
 name = "aiohttp"
@@ -79,35 +79,37 @@ sa = ["sqlalchemy[postgresql-psycopg2binary] (>=1.3,<1.5)"]
 
 [[package]]
 name = "allure-pytest"
-version = "2.13.1"
+version = "2.10.0"
 description = "Allure pytest integration"
 category = "main"
 optional = false
 python-versions = "*"
 files = [
-    {file = "allure-pytest-2.13.1.tar.gz", hash = "sha256:68d69456eeb65af4061ec06a80bc941163b0616e8216554d36b070a6bf070e08"},
-    {file = "allure_pytest-2.13.1-py3-none-any.whl", hash = "sha256:a8de2fc3b3effe2d8f98801646920de3f055b779710f4c806dbee7c613c24633"},
+    {file = "allure-pytest-2.10.0.tar.gz", hash = "sha256:3b2ab67629f4cbd8617abd817d2b22292c6eb7efd5584f992d1af8143aea6ee7"},
+    {file = "allure_pytest-2.10.0-py3-none-any.whl", hash = "sha256:08274096594758447db54c3b2c382526ee04f1fe12119cdaee92d2d93c84b530"},
 ]
 
 [package.dependencies]
-allure-python-commons = "2.13.1"
+allure-python-commons = "2.10.0"
 pytest = ">=4.5.0"
+six = ">=1.9.0"
 
 [[package]]
 name = "allure-python-commons"
-version = "2.13.1"
+version = "2.10.0"
 description = "Common module for integrate allure with python-based frameworks"
 category = "main"
 optional = false
-python-versions = ">=3.6"
+python-versions = ">=3.5"
 files = [
-    {file = "allure-python-commons-2.13.1.tar.gz", hash = "sha256:3fc13e1da8ebb23f9ab5c9c72ad04595023cdd5078dbb8604939997faebed5cb"},
-    {file = "allure_python_commons-2.13.1-py3-none-any.whl", hash = "sha256:d08e04867bddf44fef55def3d67f4bc25af58a1bf9fcffcf4ec3331f7f2ef0d0"},
+    {file = "allure-python-commons-2.10.0.tar.gz", hash = "sha256:d4d31344b0f0037a4a11e16b91b28cf0eeb23ffa0e50c27fcfc6aabe72212d3c"},
+    {file = "allure_python_commons-2.10.0-py3-none-any.whl", hash = "sha256:2a717e8ca8d296bf89cd57f38fc3c21893bd7ea8cd02a6ae5420e6d1a6eda5d0"},
 ]
 
 [package.dependencies]
 attrs = ">=16.0.0"
 pluggy = ">=0.4.0"
+six = ">=1.9.0"
 
 [[package]]
 name = "async-timeout"
@@ -1930,22 +1932,6 @@ pytest = [
     {version = ">=6.2.4", markers = "python_version >= \"3.10\""},
 ]
 
-[[package]]
-name = "pytest-rerunfailures"
-version = "11.1.2"
-description = "pytest plugin to re-run tests to eliminate flaky failures"
-category = "main"
-optional = false
-python-versions = ">=3.7"
-files = [
-    {file = "pytest-rerunfailures-11.1.2.tar.gz", hash = "sha256:55611661e873f1cafa384c82f08d07883954f4b76435f4b8a5b470c1954573de"},
-    {file = "pytest_rerunfailures-11.1.2-py3-none-any.whl", hash = "sha256:d21fe2e46d9774f8ad95f1aa799544ae95cac3a223477af94aa985adfae92b7e"},
-]
-
-[package.dependencies]
-packaging = ">=17.1"
-pytest = ">=5.3"
-
 [[package]]
 name = "pytest-timeout"
 version = "2.1.0"
@@ -2611,4 +2597,4 @@ testing = ["func-timeout", "jaraco.itertools", "pytest (>=6)", "pytest-black (>=
 [metadata]
 lock-version = "2.0"
 python-versions = "^3.9"
-content-hash = "b689ffd6eae32b966f1744b5ac3343fe0dd26b31ee1f50e13daf5045ee0623e1"
+content-hash = "2515a9320c2960076012fbc036fb33c4f6a23515c8d143785931dc18c6722d91"

pyproject.toml

@@ -26,7 +26,7 @@ prometheus-client = "^0.14.1"
 pytest-timeout = "^2.1.0"
 Werkzeug = "^2.2.3"
 pytest-order = "^1.0.1"
-allure-pytest = "^2.13.1"
+allure-pytest = "^2.10.0"
 pytest-asyncio = "^0.19.0"
 toml = "^0.10.2"
 psutil = "^5.9.4"
@@ -34,7 +34,6 @@ types-psutil = "^5.9.5.4"
 types-toml = "^0.10.8"
 pytest-httpserver = "^1.0.6"
 aiohttp = "3.7.4"
-pytest-rerunfailures = "^11.1.2"
 
 [tool.poetry.group.dev.dependencies]
 black = "^23.1.0"
@@ -70,9 +69,6 @@ strict = true
 module = [
     "asyncpg.*",
     "pg8000.*",
-    "allure.*",
-    "allure_commons.*",
-    "allure_pytest.*",
 ]
 ignore_missing_imports = true
 

scripts/flaky_tests.py

@@ -1,87 +0,0 @@
-#! /usr/bin/env python3
-
-import argparse
-import json
-import logging
-from collections import defaultdict
-from typing import DefaultDict, Dict
-
-import psycopg2
-import psycopg2.extras
-
-# We call the test "flaky" if it failed at least once on the main branch in the last N=10 days.
-FLAKY_TESTS_QUERY = """
-    SELECT
-        DISTINCT parent_suite, suite, test
-    FROM
-        (
-            SELECT
-                revision,
-                jsonb_array_elements(data -> 'children') -> 'name' as parent_suite,
-                jsonb_array_elements(jsonb_array_elements(data -> 'children') -> 'children') -> 'name' as suite,
-                jsonb_array_elements(jsonb_array_elements(jsonb_array_elements(data -> 'children') -> 'children') -> 'children') -> 'name' as test,
-                jsonb_array_elements(jsonb_array_elements(jsonb_array_elements(data -> 'children') -> 'children') -> 'children') -> 'status' as status,
-                to_timestamp((jsonb_array_elements(jsonb_array_elements(jsonb_array_elements(data -> 'children') -> 'children') -> 'children') -> 'time' -> 'start')::bigint / 1000)::date as timestamp
-            FROM
-                regress_test_results
-            WHERE
-                reference = 'refs/heads/main'
-        ) data
-    WHERE
-        timestamp > CURRENT_DATE - INTERVAL '%s' day
-        AND status::text IN ('"failed"', '"broken"')
-    ;
-"""
-
-
-def main(args: argparse.Namespace):
-    connstr = args.connstr
-    interval_days = args.days
-    output = args.output
-
-    res: DefaultDict[str, DefaultDict[str, Dict[str, bool]]]
-    res = defaultdict(lambda: defaultdict(dict))
-
-    logging.info("connecting to the database...")
-    with psycopg2.connect(connstr, connect_timeout=10) as conn:
-        with conn.cursor(cursor_factory=psycopg2.extras.DictCursor) as cur:
-            logging.info("fetching flaky tests...")
-            cur.execute(FLAKY_TESTS_QUERY, (interval_days,))
-            rows = cur.fetchall()
-
-    for row in rows:
-        logging.info(f"\t{row['parent_suite'].replace('.', '/')}/{row['suite']}.py::{row['test']}")
-        res[row["parent_suite"]][row["suite"]][row["test"]] = True
-
-    logging.info(f"saving results to {output.name}")
-    json.dump(res, output, indent=2)
-
-
-if __name__ == "__main__":
-    parser = argparse.ArgumentParser(description="Detect flaky tests in the last N days")
-    parser.add_argument(
-        "--output",
-        type=argparse.FileType("w"),
-        default="flaky.json",
-        help="path to output json file (default: flaky.json)",
-    )
-    parser.add_argument(
-        "--days",
-        required=False,
-        default=10,
-        type=int,
-        help="how many days to look back for flaky tests (default: 10)",
-    )
-    parser.add_argument(
-        "connstr",
-        help="connection string to the test results database",
-    )
-    args = parser.parse_args()
-
-    level = logging.INFO
-    logging.basicConfig(
-        format="%(message)s",
-        level=level,
-    )
-
-    main(args)

scripts/pr-comment-test-report.js

@@ -1,125 +0,0 @@
-//
-// The script parses Allure reports and posts a comment with a summary of the test results to the PR.
-// It accepts an array of items and creates a comment with a summary for each one (for "release" and "debug", together or separately if any of them failed to be generated).
-//
-// The comment is updated on each run with the latest results.
-//
-// It is designed to be used with actions/github-script from GitHub Workflows:
-// - uses: actions/github-script@v6
-//   with:
-//     script: |
-//       const script = require("./scripts/pr-comment-test-report.js")
-//       await script({
-//         github,
-//         context,
-//         fetch,
-//         reports: [{...}, ...], // each report is expected to have "buildType", "reportUrl", and "jsonUrl" properties
-//       })
-//
-
-module.exports = async ({ github, context, fetch, reports }) => {
-    // Marker to find the comment in the subsequent runs
-    const startMarker = `<!--AUTOMATIC COMMENT START #${context.payload.number}-->`
-    // GitHub bot id taken from (https://api.github.com/users/github-actions[bot])
-    const githubActionsBotId = 41898282
-    // The latest commit in the PR URL
-    const commitUrl = `${context.serverUrl}/${context.repo.owner}/${context.repo.repo}/pull/${context.payload.number}/commits/${context.sha}`
-    // Commend body itself
-    let commentBody = `${startMarker}\n### Test results for ${commitUrl}:\n___\n`
-
-    // Common parameters for GitHub API requests
-    const ownerRepoParams = {
-        owner: context.repo.owner,
-        repo: context.repo.repo,
-    }
-
-    for (const report of reports) {
-        const {buildType, reportUrl, jsonUrl} = report
-
-        if (!reportUrl || !jsonUrl) {
-            console.warn(`"reportUrl" or "jsonUrl" aren't set for ${buildType} build`)
-            continue
-        }
-
-        const suites = await (await fetch(jsonUrl)).json()
-
-        // Allure distinguishes "failed" (with an assertion error) and "broken" (with any other error) tests.
-        // For this report it's ok to treat them in the same way (as failed).
-        failedTests = []
-        passedTests = []
-        skippedTests = []
-
-        retriedTests = []
-        retriedStatusChangedTests = []
-
-        for (const parentSuite of suites.children) {
-            for (const suite of parentSuite.children) {
-                for (const test of suite.children) {
-                    pytestName = `${parentSuite.name.replace(".", "/")}/${suite.name}.py::${test.name}`
-                    test.pytestName = pytestName
-
-                    if (test.status === "passed") {
-                        passedTests.push(test);
-                    } else if (test.status === "failed" || test.status === "broken") {
-                        failedTests.push(test);
-                    } else if (test.status === "skipped") {
-                        skippedTests.push(test);
-                    }
-
-                    if (test.retriesCount > 0) {
-                        retriedTests.push(test);
-
-                        if (test.retriedStatusChangedTests) {
-                            retriedStatusChangedTests.push(test);
-                        }
-                    }
-                }
-            }
-        }
-
-        const totalTestsCount = failedTests.length + passedTests.length + skippedTests.length
-        commentBody += `#### ${buildType} build: ${totalTestsCount} tests run: ${passedTests.length} passed, ${failedTests.length} failed, ${skippedTests.length} ([full report](${reportUrl}))\n`
-        if (failedTests.length > 0) {
-            commentBody += `Failed tests:\n`
-            for (const test of failedTests) {
-                const allureLink = `${reportUrl}#suites/${test.parentUid}/${test.uid}`
-
-                commentBody += `- [\`${test.pytestName}\`](${allureLink})`
-                if (test.retriesCount > 0) {
-                    commentBody += ` (ran [${test.retriesCount + 1} times](${allureLink}/retries))`
-                }
-                commentBody += "\n"
-            }
-            commentBody += "\n"
-        }
-        if (retriedStatusChangedTests > 0) {
-            commentBody += `Flaky tests:\n`
-            for (const test of retriedStatusChangedTests) {
-                const status = test.status === "passed" ? ":white_check_mark:" : ":x:"
-                commentBody += `- ${status} [\`${test.pytestName}\`](${reportUrl}#suites/${test.parentUid}/${test.uid}/retries)\n`
-            }
-            commentBody += "\n"
-        }
-        commentBody += "___\n"
-    }
-
-    const { data: comments } = await github.rest.issues.listComments({
-        issue_number: context.payload.number,
-        ...ownerRepoParams,
-    })
-
-    const comment = comments.find(comment => comment.user.id === githubActionsBotId && comment.body.startsWith(startMarker))
-    if (comment) {
-        await github.rest.issues.updateComment({
-            comment_id: comment.id,
-            body: commentBody,
-            ...ownerRepoParams,
-        })
-    } else {
-        await github.rest.issues.createComment({
-            issue_number: context.payload.number,
-            body: commentBody,
-            ...ownerRepoParams,
-        })
-    }
-}

test_runner/conftest.py

@@ -4,5 +4,4 @@ pytest_plugins = (
     "fixtures.pg_stats",
     "fixtures.compare_fixtures",
     "fixtures.slow",
-    "fixtures.flaky",
 )

test_runner/fixtures/flaky.py

@@ -1,58 +0,0 @@
-import json
-from pathlib import Path
-from typing import List
-
-import pytest
-from _pytest.config import Config
-from _pytest.config.argparsing import Parser
-from allure_commons.types import LabelType
-from allure_pytest.utils import allure_name, allure_suite_labels
-
-from fixtures.log_helper import log
-
-"""
-The plugin reruns flaky tests.
-It uses `pytest.mark.flaky` provided by `pytest-rerunfailures` plugin and flaky tests detected by `scripts/flaky_tests.py`
-
-Note: the logic of getting flaky tests is extracted to a separate script to avoid running it for each of N xdist workers
-"""
-
-
-def pytest_addoption(parser: Parser):
-    parser.addoption(
-        "--flaky-tests-json",
-        action="store",
-        type=Path,
-        help="Path to json file with flaky tests generated by scripts/flaky_tests.py",
-    )
-
-
-def pytest_collection_modifyitems(config: Config, items: List[pytest.Item]):
-    if not config.getoption("--flaky-tests-json"):
-        return
-
-    # Any error with getting flaky tests aren't critical, so just do not rerun any tests
-    flaky_json = config.getoption("--flaky-tests-json")
-    if not flaky_json.exists():
-        return
-
-    content = flaky_json.read_text()
-    try:
-        flaky_tests = json.loads(content)
-    except ValueError:
-        log.error(f"Can't parse {content} as json")
-        return
-
-    for item in items:
-        # Use the same logic for constructing test name as Allure does (we store allure-provided data in DB)
-        # Ref https://github.com/allure-framework/allure-python/blob/2.13.1/allure-pytest/src/listener.py#L98-L100
-        allure_labels = dict(allure_suite_labels(item))
-        parent_suite = str(allure_labels.get(LabelType.PARENT_SUITE))
-        suite = str(allure_labels.get(LabelType.SUITE))
-        params = item.callspec.params if hasattr(item, "callspec") else {}
-        name = allure_name(item, params)
-
-        if flaky_tests.get(parent_suite, {}).get(suite, {}).get(name, False):
-            # Rerun 3 times = 1 original run + 2 reruns
-            log.info(f"Marking {item.nodeid} as flaky. It will be rerun up to 3 times")
-            item.add_marker(pytest.mark.flaky(reruns=2))

test_runner/fixtures/utils.py

@@ -7,7 +7,7 @@ import time
 from pathlib import Path
 from typing import Any, Callable, Dict, List, Tuple, TypeVar
 
-import allure
+import allure  # type: ignore
 from psycopg2.extensions import cursor
 
 from fixtures.log_helper import log