Merge branch 'control-persist-only-if-sync' into safekeeper-append-perf

.gitignore

@@ -11,3 +11,6 @@ test_output/
 # Coverage
 *.profraw
 *.profdata
+
+tmp1
+tmp2

Cargo.lock

@@ -2349,6 +2349,7 @@ dependencies = [
  "humantime",
  "hyper",
  "lazy_static",
+ "log",
  "pageserver",
  "postgres",
  "postgres-protocol",
@@ -2362,7 +2363,6 @@ dependencies = [
  "tempfile",
  "tokio",
  "tokio-stream",
- "tracing",
  "walkdir",
  "workspace_hack",
  "zenith_metrics",

README.md

@@ -1,6 +1,6 @@
 # Zenith
 
-Zenith is an serverless open source alternative to AWS Aurora Postgres. It separates storage and compute and substitutes PostgreSQL storage layer by redistributes data across a cluster of nodes. 
+Zenith substitutes PostgreSQL storage layer and redistributes data across a cluster of nodes
 
 ## Architecture overview
 

control_plane/src/compute.rs

@@ -199,24 +199,17 @@ impl PostgresNode {
         })
     }
 
-    fn sync_safekeepers(&self, auth_token: &Option<String>) -> Result<Lsn> {
+    fn sync_safekeepers(&self) -> Result<Lsn> {
         let pg_path = self.env.pg_bin_dir().join("postgres");
-        let mut cmd = Command::new(&pg_path);
-
-        cmd.arg("--sync-safekeepers")
+        let sync_handle = Command::new(pg_path)
+            .arg("--sync-safekeepers")
             .env_clear()
             .env("LD_LIBRARY_PATH", self.env.pg_lib_dir().to_str().unwrap())
             .env("DYLD_LIBRARY_PATH", self.env.pg_lib_dir().to_str().unwrap())
             .env("PGDATA", self.pgdata().to_str().unwrap())
             .stdout(Stdio::piped())
             // Comment this to avoid capturing stderr (useful if command hangs)
-            .stderr(Stdio::piped());
-
-        if let Some(token) = auth_token {
-            cmd.env("ZENITH_AUTH_TOKEN", token);
-        }
-
-        let sync_handle = cmd
+            .stderr(Stdio::piped())
             .spawn()
             .expect("postgres --sync-safekeepers failed to start");
 
@@ -326,11 +319,8 @@ impl PostgresNode {
             } else {
                 ""
             };
-            // NOTE avoiding spaces in connection string, because it is less error prone if we forward it somewhere.
-            // Also note that not all parameters are supported here. Because in compute we substitute $ZENITH_AUTH_TOKEN
-            // We parse this string and build it back with token from env var, and for simplicity rebuild
-            // uses only needed variables namely host, port, user, password.
-            format!("postgresql://no_user:{}@{}:{}", password, host, port)
+
+            format!("host={} port={} password={}", host, port, password)
         };
         conf.append("shared_preload_libraries", "zenith");
         conf.append_line("");
@@ -368,7 +358,7 @@ impl PostgresNode {
         Ok(())
     }
 
-    fn load_basebackup(&self, auth_token: &Option<String>) -> Result<()> {
+    fn load_basebackup(&self) -> Result<()> {
         let backup_lsn = if let Some(lsn) = self.lsn {
             Some(lsn)
         } else if self.uses_wal_proposer {
@@ -376,7 +366,7 @@ impl PostgresNode {
             // latest data from the pageserver. That is a bit clumsy but whole bootstrap
             // procedure evolves quite actively right now, so let's think about it again
             // when things would be more stable (TODO).
-            let lsn = self.sync_safekeepers(auth_token)?;
+            let lsn = self.sync_safekeepers()?;
             if lsn == Lsn(0) {
                 None
             } else {
@@ -427,6 +417,7 @@ impl PostgresNode {
         .env_clear()
         .env("LD_LIBRARY_PATH", self.env.pg_lib_dir().to_str().unwrap())
         .env("DYLD_LIBRARY_PATH", self.env.pg_lib_dir().to_str().unwrap());
+
         if let Some(token) = auth_token {
             cmd.env("ZENITH_AUTH_TOKEN", token);
         }
@@ -460,7 +451,7 @@ impl PostgresNode {
         fs::write(&postgresql_conf_path, postgresql_conf)?;
 
         // 3. Load basebackup
-        self.load_basebackup(auth_token)?;
+        self.load_basebackup()?;
 
         if self.lsn.is_some() {
             File::create(self.pgdata().join("standby.signal"))?;

control_plane/src/safekeeper.rs

@@ -15,11 +15,13 @@ use reqwest::blocking::{Client, RequestBuilder, Response};
 use reqwest::{IntoUrl, Method};
 use thiserror::Error;
 use zenith_utils::http::error::HttpErrorBody;
+use zenith_utils::postgres_backend::AuthType;
 
 use crate::local_env::{LocalEnv, SafekeeperConf};
 use crate::read_pidfile;
 use crate::storage::PageServerNode;
 use zenith_utils::connstring::connection_address;
+use zenith_utils::connstring::connection_host_port;
 
 #[derive(Error, Debug)]
 pub enum SafekeeperHttpError {
@@ -114,6 +116,17 @@ impl SafekeeperNode {
         );
         io::stdout().flush().unwrap();
 
+        // Configure connection to page server
+        //
+        // FIXME: We extract the host and port from the connection string instead of using
+        // the connection string directly, because the 'safekeeper' binary expects
+        // host:port format. That's a bit silly when we already have a full libpq connection
+        // string at hand.
+        let pageserver_conn = {
+            let (host, port) = connection_host_port(&self.pageserver.pg_connection_config);
+            format!("{}:{}", host, port)
+        };
+
         let listen_pg = format!("localhost:{}", self.conf.pg_port);
         let listen_http = format!("localhost:{}", self.conf.http_port);
 
@@ -121,6 +134,7 @@ impl SafekeeperNode {
         cmd.args(&["-D", self.datadir_path().to_str().unwrap()])
             .args(&["--listen-pg", &listen_pg])
             .args(&["--listen-http", &listen_http])
+            .args(&["--pageserver", &pageserver_conn])
             .args(&["--recall", "1 second"])
             .arg("--daemonize")
             .env_clear()
@@ -129,6 +143,10 @@ impl SafekeeperNode {
             cmd.arg("--no-sync");
         }
 
+        if self.env.pageserver.auth_type == AuthType::ZenithJWT {
+            cmd.env("PAGESERVER_AUTH_TOKEN", &self.env.pageserver.auth_token);
+        }
+
         let var = "LLVM_PROFILE_FILE";
         if let Some(val) = std::env::var_os(var) {
             cmd.env(var, val);

test_runner/batch_others/test_wal_acceptor.py

@@ -340,11 +340,10 @@ class ProposerPostgres:
                 "synchronous_standby_names = 'walproposer'\n",
                 f"zenith.zenith_timeline = '{self.timeline_id}'\n",
                 f"zenith.zenith_tenant = '{self.tenant_id}'\n",
-                f"zenith.page_server_connstring = ''\n",
                 f"wal_acceptors = '{wal_acceptors}'\n",
             ])
 
-    def sync_safekeepers(self, timeout=60) -> str:
+    def sync_safekeepers(self) -> str:
         """
         Run 'postgres --sync-safekeepers'.
         Returns execution result, which is commit_lsn after sync.
@@ -355,7 +354,7 @@ class ProposerPostgres:
             "PGDATA": self.pg_data_dir_path(),
         }
 
-        basepath = self.pg_bin.run_capture(command, env, timeout=timeout)
+        basepath = self.pg_bin.run_capture(command, env)
         stdout_filename = basepath + '.stdout'
 
         with open(stdout_filename, 'r') as stdout_f:
@@ -438,90 +437,3 @@ def test_timeline_status(zenith_env_builder: ZenithEnvBuilder):
 
     epoch_after_reboot = wa_http_cli.timeline_status(tenant_id, timeline_id).acceptor_epoch
     assert epoch_after_reboot > epoch
-
-
-class WalAppender:
-    """Helper for appending WAL to safekeepers, keeps track of last inserted lsn."""
-
-    # 0/16B9188 is good lsn to start with, it's valid and not in the segment start, nor in zero segment
-    def __init__(self,
-                 acceptors,
-                 tenant_id,
-                 timeline_id,
-                 epoch_start_lsn=0x16B9188,
-                 begin_lsn=0x16B9188):
-        self.acceptors = acceptors
-        self.epoch_start_lsn = epoch_start_lsn
-        self.begin_lsn = begin_lsn
-        self.tenant_id = tenant_id
-        self.timeline_id = timeline_id
-        self.flush_lsns = dict()
-
-    def append(self, i, term, lm_message="message", lm_prefix="", set_commit_lsn=False):
-        """Append new logical message to i'th safekeeper."""
-        lsn = self.flush_lsns.get(i, self.begin_lsn)
-        req = {
-            "lm_prefix": lm_prefix,
-            "lm_message": lm_message,
-            "set_commit_lsn": set_commit_lsn,
-            "term": term,
-            "begin_lsn": lsn,
-            "epoch_start_lsn": self.epoch_start_lsn,
-            "truncate_lsn": self.begin_lsn,
-        }
-
-        res = self.acceptors[i].append_logical_message(
-            self.tenant_id,
-            self.timeline_id,
-            req,
-        )
-
-        end_lsn = res["inserted_wal"]["end_lsn"]
-        self.flush_lsns[i] = end_lsn
-
-        return res
-
-    def debug_print(self):
-        """Print lsn for each acceptor."""
-        for i, lsn in self.flush_lsns.items():
-            print(f'end_lsn for acceptors[{i}] = {lsn_to_hex(lsn)}')
-
-
-# one safekeeper with old term and a lot of non-commited wal
-def test_sync_safekeepers_old_term_ahead(repo_dir: str,
-                                         pg_bin: PgBin,
-                                         wa_factory: WalAcceptorFactory):
-    wa_factory.start_n_new(3)
-
-    timeline_id = uuid.uuid4().hex
-    tenant_id = uuid.uuid4().hex
-
-    # write config for proposer
-    pgdata_dir = os.path.join(repo_dir, "proposer_pgdata")
-    pg = ProposerPostgres(pgdata_dir, pg_bin, timeline_id, tenant_id)
-    pg.create_dir_config(wa_factory.get_connstrs())
-
-    # append wal to safekeepers
-    appender = WalAppender(wa_factory.instances, tenant_id, timeline_id)
-
-    appender.append(0, term=1, lm_message="msg1")
-    appender.append(1, term=1, lm_message="msg1")
-    appender.append(2, term=1, lm_message="msg1")
-
-    appender.append(2, term=1, lm_message="msg1")
-    appender.append(2, term=1, lm_message="msg1")
-    appender.append(2, term=1, lm_message="msg1")
-    appender.append(2, term=1, lm_message="msg1")
-
-    new_epoch = appender.flush_lsns[0]
-    appender.epoch_start_lsn = new_epoch
-
-    appender.append(0, term=2, lm_message="msg2")
-    appender.append(1, term=2, lm_message="msg2")
-
-    # run sync safekeepers
-    # FIXME: fails with timeout
-    lsn_after_sync = pg.sync_safekeepers()
-    print(f"lsn after sync = {lsn_after_sync}")
-
-    appender.debug_print()

walkeeper/Cargo.toml

@@ -16,7 +16,7 @@ routerify = "2"
 fs2 = "0.4.3"
 lazy_static = "1.4.0"
 serde_json = "1"
-tracing = "0.1.27"
+log = "0.4.14"
 clap = "2.33.0"
 daemonize = "0.4.1"
 rust-s3 = { version = "0.27.0-rc4", features = ["no-verify-ssl"] }

walkeeper/src/bin/append_test.rs

@@ -0,0 +1,91 @@
+//
+// Performance test for append handling in safekeeper.
+//
+use anyhow::Result;
+use clap::{App, Arg};
+use const_format::formatcp;
+use daemonize::Daemonize;
+use log::*;
+use std::path::{Path, PathBuf};
+use std::thread;
+use walkeeper::defaults::{DEFAULT_HTTP_LISTEN_ADDR, DEFAULT_PG_LISTEN_ADDR};
+use walkeeper::http;
+use walkeeper::s3_offload;
+use walkeeper::wal_service;
+use walkeeper::SafeKeeperConf;
+use walkeeper::timeline::{GlobalTimelines, Timeline};
+use walkeeper::timeline::CreateControlFile;
+use walkeeper::json_ctrl;
+use zenith_utils::http::endpoint;
+use zenith_utils::shutdown::exit_now;
+use zenith_utils::signals;
+use zenith_utils::{logging, tcp_listener, GIT_VERSION};
+use zenith_utils::zid::{self, ZTimelineId, ZTenantId};
+use std::sync::Arc;
+
+use walkeeper::safekeeper::{AcceptorProposerMessage, AppendResponse};
+use walkeeper::safekeeper::{
+    AppendRequest, AppendRequestHeader, ProposerAcceptorMessage, ProposerElected, ProposerGreeting,
+};
+use walkeeper::safekeeper::{SafeKeeperState, Term, TermHistory, TermSwitchEntry};
+use walkeeper::send_wal::SendWalHandler;
+use walkeeper::timeline::TimelineTools;
+use postgres_ffi::pg_constants;
+use postgres_ffi::xlog_utils;
+use postgres_ffi::{uint32, uint64, Oid, XLogRecord};
+use zenith_utils::lsn::Lsn;
+use zenith_utils::postgres_backend::PostgresBackend;
+use zenith_utils::pq_proto::{BeMessage, RowDescriptor, TEXT_OID};
+
+fn main() -> Result<()> {
+    zenith_metrics::set_common_metrics_prefix("safekeeper");
+    let log_file = logging::init("safekeeper.log", false)?;
+    info!("version: {}", GIT_VERSION);
+    
+    let timeline_id = ZTimelineId::generate();
+    let tenant_id = ZTenantId::generate();
+    let create = CreateControlFile::True;
+    let conf = SafeKeeperConf::default();
+
+    let mut timeline = GlobalTimelines::get(&conf, tenant_id, timeline_id, create)?;
+    
+    json_ctrl::prepare_safekeeper(tenant_id ,timeline_id, &mut timeline)?;
+
+    let term = 1;
+    let epoch_start_lsn = Lsn::from(0x16B9188);
+    let mut lsn = epoch_start_lsn;
+
+    json_ctrl::send_proposer_elected(&timeline, term, lsn)?;
+
+    let message = "a".repeat(1024 * 8);
+
+    info!("Starting test");
+
+    let now = std::time::Instant::now();
+    let test_duration = std::time::Duration::from_secs(10);
+    let timeout = now.checked_add(test_duration).unwrap();
+
+    let mut total_count = 0;
+
+    while std::time::Instant::now() < timeout {
+        let result = json_ctrl::append_logical_message(&timeline, json_ctrl::AppendLogicalMessage{
+            lm_prefix: "".to_string(),
+            lm_message: message.clone(),
+            set_commit_lsn: true,
+            send_proposer_elected: false,
+            term,
+            epoch_start_lsn,
+            begin_lsn: lsn,
+            truncate_lsn: lsn,
+        })?;
+
+        lsn = result.end_lsn;
+        total_count += 1;
+    }
+
+    info!("Total count: {}", total_count);
+    info!("Test duration: {}s", test_duration.as_secs_f64());
+    info!("Appends per second: {}", total_count as f64 / test_duration.as_secs_f64());
+
+    Ok(())
+}

walkeeper/src/bin/safekeeper.rs

@@ -5,9 +5,9 @@ use anyhow::Result;
 use clap::{App, Arg};
 use const_format::formatcp;
 use daemonize::Daemonize;
+use log::*;
 use std::path::{Path, PathBuf};
 use std::thread;
-use tracing::*;
 use walkeeper::defaults::{DEFAULT_HTTP_LISTEN_ADDR, DEFAULT_PG_LISTEN_ADDR};
 use walkeeper::http;
 use walkeeper::s3_offload;
@@ -44,9 +44,6 @@ fn main() -> Result<()> {
                 .takes_value(true)
                 .help(formatcp!("http endpoint address for metrics on ip:port (default: {DEFAULT_HTTP_LISTEN_ADDR})")),
         )
-        // FIXME this argument is no longer needed since pageserver address is forwarded from compute.
-        // However because this argument is in use by console's e2e tests lets keep it for now and remove separately.
-        // So currently it is a noop.
         .arg(
             Arg::with_name("pageserver")
                 .short("p")
@@ -104,6 +101,10 @@ fn main() -> Result<()> {
         conf.listen_http_addr = addr.to_owned();
     }
 
+    if let Some(addr) = arg_matches.value_of("pageserver") {
+        conf.pageserver_addr = Some(addr.to_owned());
+    }
+
     if let Some(ttl) = arg_matches.value_of("ttl") {
         conf.ttl = Some(humantime::parse_duration(ttl)?);
     }

walkeeper/src/json_ctrl.rs

@@ -9,8 +9,9 @@
 use anyhow::{anyhow, Result};
 use bytes::{BufMut, Bytes, BytesMut};
 use crc32c::crc32c_append;
+use log::*;
 use serde::{Deserialize, Serialize};
-use tracing::*;
+use std::sync::Arc;
 
 use crate::safekeeper::{AcceptorProposerMessage, AppendResponse};
 use crate::safekeeper::{
@@ -18,31 +19,32 @@ use crate::safekeeper::{
 };
 use crate::safekeeper::{SafeKeeperState, Term, TermHistory, TermSwitchEntry};
 use crate::send_wal::SendWalHandler;
-use crate::timeline::TimelineTools;
+use crate::timeline::{TimelineTools, Timeline};
 use postgres_ffi::pg_constants;
 use postgres_ffi::xlog_utils;
 use postgres_ffi::{uint32, uint64, Oid, XLogRecord};
 use zenith_utils::lsn::Lsn;
 use zenith_utils::postgres_backend::PostgresBackend;
 use zenith_utils::pq_proto::{BeMessage, RowDescriptor, TEXT_OID};
+use zenith_utils::zid::{ZTimelineId, ZTenantId};
 
 #[derive(Serialize, Deserialize, Debug)]
-struct AppendLogicalMessage {
+pub struct AppendLogicalMessage {
     // prefix and message to build LogicalMessage
-    lm_prefix: String,
-    lm_message: String,
+    pub lm_prefix: String,
+    pub lm_message: String,
 
     // if true, commit_lsn will match flush_lsn after append
-    set_commit_lsn: bool,
+    pub set_commit_lsn: bool,
 
     // if true, ProposerElected will be sent before append
-    send_proposer_elected: bool,
+    pub send_proposer_elected: bool,
 
     // fields from AppendRequestHeader
-    term: Term,
-    epoch_start_lsn: Lsn,
-    begin_lsn: Lsn,
-    truncate_lsn: Lsn,
+    pub term: Term,
+    pub epoch_start_lsn: Lsn,
+    pub begin_lsn: Lsn,
+    pub truncate_lsn: Lsn,
 }
 
 #[derive(Serialize, Deserialize)]
@@ -71,14 +73,14 @@ pub fn handle_json_ctrl(
     info!("JSON_CTRL request: {:?}", append_request);
 
     // need to init safekeeper state before AppendRequest
-    prepare_safekeeper(swh)?;
+    prepare_safekeeper(swh.tenantid.unwrap(), swh.timelineid.unwrap(), swh.timeline.get())?;
 
     // if send_proposer_elected is true, we need to update local history
     if append_request.send_proposer_elected {
-        send_proposer_elected(swh, append_request.term, append_request.epoch_start_lsn)?;
+        send_proposer_elected(swh.timeline.get(), append_request.term, append_request.epoch_start_lsn)?;
     }
 
-    let inserted_wal = append_logical_message(swh, append_request)?;
+    let inserted_wal = append_logical_message(swh.timeline.get(), append_request)?;
     let response = AppendResult {
         state: swh.timeline.get().get_info(),
         inserted_wal,
@@ -98,28 +100,28 @@ pub fn handle_json_ctrl(
 
 /// Prepare safekeeper to process append requests without crashes,
 /// by sending ProposerGreeting with default server.wal_seg_size.
-fn prepare_safekeeper(swh: &mut SendWalHandler) -> Result<()> {
+pub fn prepare_safekeeper(tenant_id: ZTenantId, timeline_id: ZTimelineId, timeline: &Arc<Timeline>) -> Result<()> {
     let greeting_request = ProposerAcceptorMessage::Greeting(ProposerGreeting {
         protocol_version: 1, // current protocol
         pg_version: 0,       // unknown
         proposer_id: [0u8; 16],
         system_id: 0,
-        ztli: swh.timelineid.unwrap(),
-        tenant_id: swh.tenantid.unwrap(),
-        tli: 0,
+        ztli: timeline_id,
+        tenant_id: tenant_id,
+        tli: xlog_utils::PG_TLI,
         wal_seg_size: pg_constants::WAL_SEGMENT_SIZE as u32, // 16MB, default for tests
     });
 
-    let response = swh.timeline.get().process_msg(&greeting_request)?;
+    let response = timeline.process_msg(&greeting_request)?;
     match response {
         Some(AcceptorProposerMessage::Greeting(_)) => Ok(()),
         _ => anyhow::bail!("not GreetingResponse"),
     }
 }
 
-fn send_proposer_elected(swh: &mut SendWalHandler, term: Term, lsn: Lsn) -> Result<()> {
+pub fn send_proposer_elected(timeline: &Arc<Timeline>, term: Term, lsn: Lsn) -> Result<()> {
     // add new term to existing history
-    let history = swh.timeline.get().get_info().acceptor_state.term_history;
+    let history = timeline.get_info().acceptor_state.term_history;
     let history = history.up_to(lsn.checked_sub(1u64).unwrap());
     let mut history_entries = history.0;
     history_entries.push(TermSwitchEntry { term, lsn });
@@ -131,25 +133,25 @@ fn send_proposer_elected(swh: &mut SendWalHandler, term: Term, lsn: Lsn) -> Resu
         term_history: history,
     });
 
-    swh.timeline.get().process_msg(&proposer_elected_request)?;
+    timeline.process_msg(&proposer_elected_request)?;
     Ok(())
 }
 
 #[derive(Serialize, Deserialize)]
-struct InsertedWAL {
-    begin_lsn: Lsn,
-    end_lsn: Lsn,
-    append_response: AppendResponse,
+pub struct InsertedWAL {
+    pub begin_lsn: Lsn,
+    pub end_lsn: Lsn,
+    pub append_response: AppendResponse,
 }
 
 /// Extend local WAL with new LogicalMessage record. To do that,
 /// create AppendRequest with new WAL and pass it to safekeeper.
-fn append_logical_message(
-    swh: &mut SendWalHandler,
+pub fn append_logical_message(
+    timeline: &Arc<Timeline>,
     msg: AppendLogicalMessage,
 ) -> Result<InsertedWAL> {
     let wal_data = encode_logical_message(msg.lm_prefix, msg.lm_message);
-    let sk_state = swh.timeline.get().get_info();
+    let sk_state = timeline.get_info();
 
     let begin_lsn = msg.begin_lsn;
     let end_lsn = begin_lsn + wal_data.len() as u64;
@@ -173,7 +175,7 @@ fn append_logical_message(
         wal_data: Bytes::from(wal_data),
     });
 
-    let response = swh.timeline.get().process_msg(&append_request)?;
+    let response = timeline.process_msg(&append_request)?;
 
     let append_response = match response {
         Some(AcceptorProposerMessage::AppendResponse(resp)) => resp,
@@ -222,11 +224,15 @@ fn encode_logical_message(prefix: String, message: String) -> Vec<u8> {
 
     let mainrdata = logical_message.encode();
     let mainrdata_len: usize = mainrdata.len() + prefix_bytes.len() + message_bytes.len();
-    // only short mainrdata is supported for now
-    assert!(mainrdata_len <= 255);
-    let mainrdata_len = mainrdata_len as u8;
 
-    let mut data: Vec<u8> = vec![pg_constants::XLR_BLOCK_ID_DATA_SHORT, mainrdata_len];
+    let mut data: Vec<u8> = vec![];
+    if mainrdata_len <= 255 {
+        data.put_u8(pg_constants::XLR_BLOCK_ID_DATA_SHORT);
+        data.put_u8(mainrdata_len as u8);
+    } else {
+        data.put_u8(pg_constants::XLR_BLOCK_ID_DATA_LONG);
+        data.put_u64_le(mainrdata_len as u64);
+    }
     data.extend_from_slice(&mainrdata);
     data.extend_from_slice(&prefix_bytes);
     data.extend_from_slice(message_bytes);

walkeeper/src/lib.rs

@@ -2,6 +2,7 @@
 use std::path::PathBuf;
 use std::time::Duration;
 
+use std::env;
 use zenith_utils::zid::ZTimelineId;
 
 pub mod http;
@@ -12,7 +13,6 @@ pub mod s3_offload;
 pub mod safekeeper;
 pub mod send_wal;
 pub mod timeline;
-pub mod upgrade;
 pub mod wal_service;
 
 pub mod defaults {
@@ -39,6 +39,9 @@ pub struct SafeKeeperConf {
     pub no_sync: bool,
     pub listen_pg_addr: String,
     pub listen_http_addr: String,
+    pub pageserver_addr: Option<String>,
+    // TODO (create issue) this is temporary, until protocol between PG<->SK<->PS rework
+    pub pageserver_auth_token: Option<String>,
     pub ttl: Option<Duration>,
     pub recall_period: Option<Duration>,
 }
@@ -58,10 +61,12 @@ impl Default for SafeKeeperConf {
             workdir: PathBuf::from("./"),
             daemonize: false,
             no_sync: false,
+            pageserver_addr: None,
             listen_pg_addr: defaults::DEFAULT_PG_LISTEN_ADDR.to_string(),
             listen_http_addr: defaults::DEFAULT_PG_LISTEN_ADDR.to_string(),
             ttl: None,
             recall_period: None,
+            pageserver_auth_token: env::var("PAGESERVER_AUTH_TOKEN").ok(),
         }
     }
 }

walkeeper/src/receive_wal.rs

@@ -5,8 +5,8 @@
 use anyhow::{bail, Context, Result};
 use bytes::Bytes;
 use bytes::BytesMut;
+use log::*;
 use postgres::{Client, Config, NoTls};
-use tracing::*;
 
 use std::net::SocketAddr;
 use std::thread;
@@ -28,22 +28,20 @@ pub struct ReceiveWalConn<'pg> {
     pg_backend: &'pg mut PostgresBackend,
     /// The cached result of `pg_backend.socket().peer_addr()` (roughly)
     peer_addr: SocketAddr,
-    /// Pageserver connection string forwarded from compute
-    /// NOTE that it is allowed to operate without a pageserver.
-    /// So if compute has no pageserver configured do not use it.
-    pageserver_connstr: Option<String>,
 }
 
 ///
 /// Periodically request pageserver to call back.
 /// If pageserver already has replication channel, it will just ignore this request
 ///
-fn request_callback(
-    conf: SafeKeeperConf,
-    pageserver_connstr: String,
-    timelineid: ZTimelineId,
-    tenantid: ZTenantId,
-) {
+fn request_callback(conf: SafeKeeperConf, timelineid: ZTimelineId, tenantid: ZTenantId) {
+    let ps_addr = conf.pageserver_addr.unwrap();
+    let ps_connstr = format!(
+        "postgresql://no_user:{}@{}/no_db",
+        &conf.pageserver_auth_token.unwrap_or_default(),
+        ps_addr
+    );
+
     // use Config parsing because SockAddr parsing doesnt allow to use host names instead of ip addresses
     let me_connstr = format!("postgresql://no_user@{}/no_db", conf.listen_pg_addr);
     let me_conf: Config = me_connstr.parse().unwrap();
@@ -56,18 +54,15 @@ fn request_callback(
     loop {
         info!(
             "requesting page server to connect to us: start {} {}",
-            pageserver_connstr, callme
+            ps_connstr, callme
         );
-        match Client::connect(&pageserver_connstr, NoTls) {
+        match Client::connect(&ps_connstr, NoTls) {
             Ok(mut client) => {
                 if let Err(e) = client.simple_query(&callme) {
                     error!("Failed to send callme request to pageserver: {}", e);
                 }
             }
-            Err(e) => error!(
-                "Failed to connect to pageserver {}: {}",
-                &pageserver_connstr, e
-            ),
+            Err(e) => error!("Failed to connect to pageserver {}: {}", &ps_connstr, e),
         }
 
         if let Some(period) = conf.recall_period {
@@ -79,15 +74,11 @@ fn request_callback(
 }
 
 impl<'pg> ReceiveWalConn<'pg> {
-    pub fn new(
-        pg: &'pg mut PostgresBackend,
-        pageserver_connstr: Option<String>,
-    ) -> ReceiveWalConn<'pg> {
+    pub fn new(pg: &'pg mut PostgresBackend) -> ReceiveWalConn<'pg> {
         let peer_addr = *pg.get_peer_addr();
         ReceiveWalConn {
             pg_backend: pg,
             peer_addr,
-            pageserver_connstr,
         }
     }
 
@@ -116,8 +107,6 @@ impl<'pg> ReceiveWalConn<'pg> {
 
     /// Receive WAL from wal_proposer
     pub fn run(&mut self, swh: &mut SendWalHandler) -> Result<()> {
-        let _enter = info_span!("WAL acceptor", timeline = %swh.timelineid.unwrap()).entered();
-
         // Notify the libpq client that it's allowed to send `CopyData` messages
         self.pg_backend
             .write_message(&BeMessage::CopyBothResponse)?;
@@ -138,17 +127,17 @@ impl<'pg> ReceiveWalConn<'pg> {
             _ => bail!("unexpected message {:?} instead of greeting", msg),
         }
 
-        // Need to establish replication channel with page server.
-        // Add far as replication in postgres is initiated by receiver, we should use callme mechanism
-        if let Some(ref pageserver_connstr) = self.pageserver_connstr {
+        // if requested, ask pageserver to fetch wal from us
+        // xxx: this place seems not really fitting
+        if swh.conf.pageserver_addr.is_some() {
+            // Need to establish replication channel with page server.
+            // Add far as replication in postgres is initiated by receiver, we should use callme mechanism
             let conf = swh.conf.clone();
             let timelineid = swh.timeline.get().timelineid;
-            // copy to safely move to a thread
-            let pageserver_connstr = pageserver_connstr.to_owned();
             let _ = thread::Builder::new()
                 .name("request_callback thread".into())
                 .spawn(move || {
-                    request_callback(conf, pageserver_connstr, timelineid, tenant_id);
+                    request_callback(conf, timelineid, tenant_id);
                 })
                 .unwrap();
         }

walkeeper/src/replication.rs

@@ -5,6 +5,7 @@ use crate::send_wal::SendWalHandler;
 use crate::timeline::{ReplicaState, Timeline, TimelineTools};
 use anyhow::{anyhow, Context, Result};
 use bytes::Bytes;
+use log::*;
 use postgres_ffi::xlog_utils::{
     get_current_timestamp, TimestampTz, XLogFileName, MAX_SEND_SIZE, PG_TLI,
 };
@@ -19,7 +20,6 @@ use std::sync::Arc;
 use std::thread::sleep;
 use std::time::Duration;
 use std::{str, thread};
-use tracing::*;
 use zenith_utils::bin_ser::BeSer;
 use zenith_utils::lsn::Lsn;
 use zenith_utils::postgres_backend::PostgresBackend;
@@ -177,8 +177,6 @@ impl ReplicationConn {
         pgb: &mut PostgresBackend,
         cmd: &Bytes,
     ) -> Result<()> {
-        let _enter = info_span!("WAL sender", timeline = %swh.timelineid.unwrap()).entered();
-
         // spawn the background thread which receives HotStandbyFeedback messages.
         let bg_timeline = Arc::clone(swh.timeline.get());
         let bg_stream_in = self.stream_in.take().unwrap();

walkeeper/src/s3_offload.rs

@@ -3,6 +3,7 @@
 //
 
 use anyhow::Result;
+use log::*;
 use postgres_ffi::xlog_utils::*;
 use s3::bucket::Bucket;
 use s3::creds::Credentials;
@@ -15,7 +16,6 @@ use std::path::Path;
 use std::time::SystemTime;
 use tokio::runtime;
 use tokio::time::sleep;
-use tracing::*;
 use walkdir::WalkDir;
 
 use crate::SafeKeeperConf;

walkeeper/src/safekeeper.rs

@@ -8,13 +8,13 @@ use bytes::Buf;
 use bytes::BufMut;
 use bytes::Bytes;
 use bytes::BytesMut;
+use log::*;
 use pageserver::waldecoder::WalStreamDecoder;
 use postgres_ffi::xlog_utils::TimeLineID;
 use serde::{Deserialize, Serialize};
 use std::cmp::min;
 use std::fmt;
 use std::io::Read;
-use tracing::*;
 
 use lazy_static::lazy_static;
 
@@ -30,7 +30,7 @@ use zenith_utils::pq_proto::SystemId;
 use zenith_utils::zid::{ZTenantId, ZTimelineId};
 
 pub const SK_MAGIC: u32 = 0xcafeceefu32;
-pub const SK_FORMAT_VERSION: u32 = 2;
+pub const SK_FORMAT_VERSION: u32 = 1;
 const SK_PROTOCOL_VERSION: u32 = 1;
 const UNKNOWN_SERVER_VERSION: u32 = 0;
 
@@ -102,7 +102,7 @@ impl fmt::Debug for TermHistory {
 }
 
 /// Unique id of proposer. Not needed for correctness, used for monitoring.
-pub type PgUuid = [u8; 16];
+type PgUuid = [u8; 16];
 
 /// Persistent consensus state of the acceptor.
 #[derive(Debug, Clone, Serialize, Deserialize)]
@@ -140,9 +140,12 @@ pub struct ServerInfo {
 }
 
 /// Persistent information stored on safekeeper node
-/// On disk data is prefixed by magic and format version and followed by checksum.
 #[derive(Debug, Clone, Serialize, Deserialize)]
 pub struct SafeKeeperState {
+    /// magic for verifying content the control file
+    pub magic: u32,
+    /// safekeeper format version
+    pub format_version: u32,
     /// persistent acceptor state
     pub acceptor_state: AcceptorState,
     /// information about server
@@ -163,6 +166,8 @@ pub struct SafeKeeperState {
 impl SafeKeeperState {
     pub fn new() -> SafeKeeperState {
         SafeKeeperState {
+            magic: SK_MAGIC,
+            format_version: SK_FORMAT_VERSION,
             acceptor_state: AcceptorState {
                 term: 0,
                 term_history: TermHistory::empty(),
@@ -409,8 +414,8 @@ impl AcceptorProposerMessage {
 }
 
 pub trait Storage {
-    /// Persist safekeeper state on disk.
-    fn persist(&mut self, s: &SafeKeeperState) -> Result<()>;
+    /// Persist safekeeper state on disk, optionally syncing it.
+    fn persist(&mut self, s: &SafeKeeperState, sync: bool) -> Result<()>;
     /// Write piece of wal in buf to disk and sync it.
     fn write_wal(&mut self, server: &ServerInfo, startpos: Lsn, buf: &[u8]) -> Result<()>;
     // Truncate WAL at specified LSN
@@ -563,7 +568,7 @@ where
         self.s.server.ztli = msg.ztli;
         self.s.server.wal_seg_size = msg.wal_seg_size;
         self.storage
-            .persist(&self.s)
+            .persist(&self.s, true)
             .with_context(|| "failed to persist shared state")?;
 
         self.metrics = SafeKeeperMetrics::new(self.s.server.ztli);
@@ -593,7 +598,7 @@ where
         if self.s.acceptor_state.term < msg.term {
             self.s.acceptor_state.term = msg.term;
             // persist vote before sending it out
-            self.storage.persist(&self.s)?;
+            self.storage.persist(&self.s, true)?;
             resp.term = self.s.acceptor_state.term;
             resp.vote_given = true as u64;
         }
@@ -605,7 +610,7 @@ where
     fn bump_if_higher(&mut self, term: Term) -> Result<()> {
         if self.s.acceptor_state.term < term {
             self.s.acceptor_state.term = term;
-            self.storage.persist(&self.s)?;
+            self.storage.persist(&self.s, true)?;
         }
         Ok(())
     }
@@ -644,7 +649,7 @@ where
         self.flush_lsn = msg.start_streaming_at;
         // and now adopt term history from proposer
         self.s.acceptor_state.term_history = msg.term_history.clone();
-        self.storage.persist(&self.s)?;
+        self.storage.persist(&self.s, true)?;
 
         info!("start receiving WAL since {:?}", msg.start_streaming_at);
 
@@ -697,15 +702,17 @@ where
                 );
                 self.decoder = WalStreamDecoder::new(msg.h.begin_lsn);
             }
-            self.decoder.feed_bytes(&msg.wal_data);
-            loop {
-                match self.decoder.poll_decode()? {
-                    None => break, // no full record yet
-                    Some((lsn, _rec)) => {
-                        last_rec_lsn = lsn;
-                    }
-                }
-            }
+            // XXX: decoder is disabled for this test
+            // self.decoder.feed_bytes(&msg.wal_data);
+            // loop {
+            //     match self.decoder.poll_decode()? {
+            //         None => break, // no full record yet
+            //         Some((lsn, _rec)) => {
+            //             last_rec_lsn = lsn;
+            //         }
+            //     }
+            // }
+            last_rec_lsn = msg.h.begin_lsn + msg.wal_data.len() as u64;
 
             // If this was the first record we ever receieved, remember LSN to help
             // find_end_of_wal skip the hole in the beginning.
@@ -750,7 +757,7 @@ where
         }
 
         if sync_control_file {
-            self.storage.persist(&self.s)?;
+            self.storage.persist(&self.s, true)?;
         }
 
         let resp = self.append_response();
@@ -776,7 +783,7 @@ mod tests {
     }
 
     impl Storage for InMemoryStorage {
-        fn persist(&mut self, s: &SafeKeeperState) -> Result<()> {
+        fn persist(&mut self, s: &SafeKeeperState, _sync: bool) -> Result<()> {
             self.persisted_state = s.clone();
             Ok(())
         }

walkeeper/src/send_wal.rs

@@ -46,7 +46,6 @@ impl postgres_backend::Handler for SendWalHandler {
         if let Some(app_name) = sm.params.get("application_name") {
             self.appname = Some(app_name.clone());
         }
-
         Ok(())
     }
 
@@ -76,17 +75,7 @@ impl postgres_backend::Handler for SendWalHandler {
         } else if query_string.starts_with(b"START_REPLICATION") {
             ReplicationConn::new(pgb).run(self, pgb, &query_string)?;
         } else if query_string.starts_with(b"START_WAL_PUSH") {
-            // TODO: this repeats query decoding logic from page_service so it is probably
-            // a good idea to refactor it in pgbackend and pass string to process query instead of bytes
-            let decoded_query_string = match query_string.last() {
-                Some(0) => std::str::from_utf8(&query_string[..query_string.len() - 1])?,
-                _ => std::str::from_utf8(&query_string)?,
-            };
-            let pageserver_connstr = decoded_query_string
-                .split_whitespace()
-                .nth(1)
-                .map(|s| s.to_owned());
-            ReceiveWalConn::new(pgb, pageserver_connstr)
+            ReceiveWalConn::new(pgb)
                 .run(self)
                 .with_context(|| "failed to run ReceiveWalConn")?;
         } else if query_string.starts_with(b"JSON_CTRL") {

walkeeper/src/timeline.rs

@@ -2,9 +2,9 @@
 //! persistence and support for interaction between sending and receiving wal.
 
 use anyhow::{anyhow, bail, ensure, Context, Result};
-use byteorder::{LittleEndian, ReadBytesExt, WriteBytesExt};
 use fs2::FileExt;
 use lazy_static::lazy_static;
+use log::*;
 use postgres_ffi::xlog_utils::{find_end_of_wal, PG_TLI};
 use std::cmp::{max, min};
 use std::collections::HashMap;
@@ -13,7 +13,6 @@ use std::io::{Read, Seek, SeekFrom, Write};
 use std::path::PathBuf;
 use std::sync::{Arc, Condvar, Mutex};
 use std::time::Duration;
-use tracing::*;
 use zenith_metrics::{register_histogram_vec, Histogram, HistogramVec, DISK_WRITE_SECONDS_BUCKETS};
 use zenith_utils::bin_ser::LeSer;
 use zenith_utils::lsn::Lsn;
@@ -24,7 +23,6 @@ use crate::safekeeper::{
     AcceptorProposerMessage, ProposerAcceptorMessage, SafeKeeper, SafeKeeperState, ServerInfo,
     Storage, SK_FORMAT_VERSION, SK_MAGIC,
 };
-use crate::upgrade::upgrade_control_file;
 use crate::SafeKeeperConf;
 use postgres_ffi::xlog_utils::{XLogFileName, XLOG_BLCKSZ};
 use std::convert::TryInto;
@@ -36,7 +34,7 @@ const CONTROL_FILE_NAME_PARTIAL: &str = "safekeeper.control.partial";
 // dedicated lockfile to prevent running several safekeepers on the same data
 const LOCK_FILE_NAME: &str = "safekeeper.lock";
 const POLL_STATE_TIMEOUT: Duration = Duration::from_secs(1);
-pub const CHECKSUM_SIZE: usize = std::mem::size_of::<u32>();
+const CHECKSUM_SIZE: usize = std::mem::size_of::<u32>();
 
 /// Replica status: host standby feedback + disk consistent lsn
 #[derive(Debug, Clone, Copy)]
@@ -85,13 +83,20 @@ pub enum CreateControlFile {
 }
 
 lazy_static! {
-    static ref PERSIST_CONTROL_FILE_SECONDS: HistogramVec = register_histogram_vec!(
-        "safekeeper_persist_control_file_seconds",
+    static ref PERSIST_SYNC_CONTROL_FILE_SECONDS: HistogramVec = register_histogram_vec!(
+        "safekeeper_persist_sync_control_file_seconds",
         "Seconds to persist and sync control file, grouped by timeline",
         &["timeline_id"],
         DISK_WRITE_SECONDS_BUCKETS.to_vec()
     )
-    .expect("Failed to register safekeeper_persist_control_file_seconds histogram vec");
+    .expect("Failed to register safekeeper_persist_sync_control_file_seconds histogram vec");
+    static ref PERSIST_NOSYNC_CONTROL_FILE_SECONDS: HistogramVec = register_histogram_vec!(
+        "safekeeper_persist_nosync_control_file_seconds",
+        "Seconds to persist and sync control file, grouped by timeline",
+        &["timeline_id"],
+        DISK_WRITE_SECONDS_BUCKETS.to_vec()
+    )
+    .expect("Failed to register safekeeper_persist_nosync_control_file_seconds histogram vec");
 }
 
 impl SharedState {
@@ -129,7 +134,7 @@ impl SharedState {
         timelineid: ZTimelineId,
         create: CreateControlFile,
     ) -> Result<Self> {
-        let (file_storage, state) = FileStorage::load_from_control_file(conf, timelineid, create)
+        let (file_storage, state) = SharedState::load_from_control_file(conf, timelineid, create)
             .with_context(|| "failed to load from control file")?;
         let flush_lsn = if state.server.wal_seg_size != 0 {
             let wal_dir = conf.timeline_dir(&timelineid);
@@ -150,6 +155,111 @@ impl SharedState {
             replicas: Vec::new(),
         })
     }
+
+    /// Fetch and lock control file (prevent running more than one instance of safekeeper)
+    /// If create=false and file doesn't exist, bails out.
+    fn load_from_control_file(
+        conf: &SafeKeeperConf,
+        timelineid: ZTimelineId,
+        create: CreateControlFile,
+    ) -> Result<(FileStorage, SafeKeeperState)> {
+        let timeline_dir = conf.timeline_dir(&timelineid);
+
+        let control_file_path = timeline_dir.join(CONTROL_FILE_NAME);
+        let lock_file_path = timeline_dir.join(LOCK_FILE_NAME);
+
+        info!(
+            "loading control file {}, create={:?} lock file {:?}",
+            control_file_path.display(),
+            create,
+            lock_file_path.display(),
+        );
+
+        let lock_file = File::create(&lock_file_path).with_context(|| "failed to open lockfile")?;
+
+        // Lock file to prevent two or more active safekeepers
+        lock_file.try_lock_exclusive().map_err(|e| {
+            anyhow!(
+                "control file {:?} is locked by some other process: {}",
+                &control_file_path,
+                e
+            )
+        })?;
+
+        let mut control_file = OpenOptions::new()
+            .read(true)
+            .write(true)
+            .create(matches!(create, CreateControlFile::True))
+            .open(&control_file_path)
+            .with_context(|| {
+                format!(
+                    "failed to open control file at {}",
+                    control_file_path.display(),
+                )
+            })?;
+
+        // Empty file is legit on 'create', don't try to deser from it.
+        let state = if control_file.metadata().unwrap().len() == 0 {
+            if let CreateControlFile::False = create {
+                bail!("control file is empty");
+            }
+            SafeKeeperState::new()
+        } else {
+            let mut buf = Vec::new();
+            control_file
+                .read_to_end(&mut buf)
+                .with_context(|| "failed to read control file")?;
+
+            let calculated_checksum = crc32c::crc32c(&buf[..buf.len() - CHECKSUM_SIZE]);
+
+            let expected_checksum_bytes: &[u8; CHECKSUM_SIZE] =
+                buf[buf.len() - CHECKSUM_SIZE..].try_into()?;
+            let expected_checksum = u32::from_le_bytes(*expected_checksum_bytes);
+
+            ensure!(
+                calculated_checksum == expected_checksum,
+                format!(
+                    "safe keeper state checksum mismatch expected {} got {}",
+                    expected_checksum, calculated_checksum
+                )
+            );
+
+            let state =
+                SafeKeeperState::des(&buf[..buf.len() - CHECKSUM_SIZE]).with_context(|| {
+                    format!(
+                        "failed to deserialize safe keeper state from control file at {}",
+                        control_file_path.display(),
+                    )
+                })?;
+
+            if state.magic != SK_MAGIC {
+                bail!("bad control file magic: {}", state.magic);
+            }
+            if state.format_version != SK_FORMAT_VERSION {
+                bail!(
+                    "Got incompatible format version, expected {}, got {}",
+                    SK_FORMAT_VERSION,
+                    state.format_version,
+                );
+            }
+            state
+        };
+
+        let timelineid_str = format!("{}", timelineid);
+
+        Ok((
+            FileStorage {
+                lock_file,
+                timeline_dir,
+                conf: conf.clone(),
+                persist_sync_control_file_seconds: PERSIST_SYNC_CONTROL_FILE_SECONDS
+                    .with_label_values(&[&timelineid_str]),
+                persist_nosync_control_file_seconds: PERSIST_NOSYNC_CONTROL_FILE_SECONDS
+                    .with_label_values(&[&timelineid_str]),
+            },
+            state,
+        ))
+    }
 }
 
 /// Database instance (tenant)
@@ -331,123 +441,20 @@ struct FileStorage {
     // save timeline dir to avoid reconstructing it every time
     timeline_dir: PathBuf,
     conf: SafeKeeperConf,
-    persist_control_file_seconds: Histogram,
-}
-
-impl FileStorage {
-    // Check the magic/version in the on-disk data and deserialize it, if possible.
-    fn deser_sk_state(buf: &mut &[u8]) -> Result<SafeKeeperState> {
-        // Read the version independent part
-        let magic = buf.read_u32::<LittleEndian>()?;
-        if magic != SK_MAGIC {
-            bail!(
-                "bad control file magic: {:X}, expected {:X}",
-                magic,
-                SK_MAGIC
-            );
-        }
-        let version = buf.read_u32::<LittleEndian>()?;
-        if version == SK_FORMAT_VERSION {
-            let res = SafeKeeperState::des(buf)?;
-            return Ok(res);
-        }
-        // try to upgrade
-        upgrade_control_file(buf, version)
-    }
-
-    /// Fetch and lock control file (prevent running more than one instance of safekeeper)
-    /// If create=false and file doesn't exist, bails out.
-    fn load_from_control_file(
-        conf: &SafeKeeperConf,
-        timelineid: ZTimelineId,
-        create: CreateControlFile,
-    ) -> Result<(FileStorage, SafeKeeperState)> {
-        let timeline_dir = conf.timeline_dir(&timelineid);
-
-        let control_file_path = timeline_dir.join(CONTROL_FILE_NAME);
-        let lock_file_path = timeline_dir.join(LOCK_FILE_NAME);
-
-        info!(
-            "loading control file {}, create={:?} lock file {:?}",
-            control_file_path.display(),
-            create,
-            lock_file_path.display(),
-        );
-
-        let lock_file = File::create(&lock_file_path).with_context(|| "failed to open lockfile")?;
-
-        // Lock file to prevent two or more active safekeepers
-        lock_file.try_lock_exclusive().map_err(|e| {
-            anyhow!(
-                "control file {:?} is locked by some other process: {}",
-                &control_file_path,
-                e
-            )
-        })?;
-
-        let mut control_file = OpenOptions::new()
-            .read(true)
-            .write(true)
-            .create(matches!(create, CreateControlFile::True))
-            .open(&control_file_path)
-            .with_context(|| {
-                format!(
-                    "failed to open control file at {}",
-                    control_file_path.display(),
-                )
-            })?;
-
-        // Empty file is legit on 'create', don't try to deser from it.
-        let state = if control_file.metadata().unwrap().len() == 0 {
-            if let CreateControlFile::False = create {
-                bail!("control file is empty");
-            }
-            SafeKeeperState::new()
-        } else {
-            let mut buf = Vec::new();
-            control_file
-                .read_to_end(&mut buf)
-                .with_context(|| "failed to read control file")?;
-
-            let calculated_checksum = crc32c::crc32c(&buf[..buf.len() - CHECKSUM_SIZE]);
-
-            let expected_checksum_bytes: &[u8; CHECKSUM_SIZE] =
-                buf[buf.len() - CHECKSUM_SIZE..].try_into()?;
-            let expected_checksum = u32::from_le_bytes(*expected_checksum_bytes);
-
-            ensure!(
-                calculated_checksum == expected_checksum,
-                format!(
-                    "safekeeper control file checksum mismatch: expected {} got {}",
-                    expected_checksum, calculated_checksum
-                )
-            );
-
-            FileStorage::deser_sk_state(&mut &buf[..buf.len() - CHECKSUM_SIZE]).with_context(
-                || format!("while reading control file {}", control_file_path.display(),),
-            )?
-        };
-
-        let timelineid_str = format!("{}", timelineid);
-
-        Ok((
-            FileStorage {
-                lock_file,
-                timeline_dir,
-                conf: conf.clone(),
-                persist_control_file_seconds: PERSIST_CONTROL_FILE_SECONDS
-                    .with_label_values(&[&timelineid_str]),
-            },
-            state,
-        ))
-    }
+    persist_sync_control_file_seconds: Histogram,
+    persist_nosync_control_file_seconds: Histogram,
 }
 
 impl Storage for FileStorage {
     // persists state durably to underlying storage
     // for description see https://lwn.net/Articles/457667/
-    fn persist(&mut self, s: &SafeKeeperState) -> Result<()> {
-        let _timer = &self.persist_control_file_seconds.start_timer();
+    fn persist(&mut self, s: &SafeKeeperState, sync: bool) -> Result<()> {
+        let _timer = if sync {
+            &self.persist_sync_control_file_seconds
+        } else {
+            &self.persist_nosync_control_file_seconds
+        }
+        .start_timer();
 
         // write data to safekeeper.control.partial
         let control_partial_path = self.timeline_dir.join(CONTROL_FILE_NAME_PARTIAL);
@@ -457,11 +464,7 @@ impl Storage for FileStorage {
                 &control_partial_path.display()
             )
         })?;
-        let mut buf: Vec<u8> = Vec::new();
-        buf.write_u32::<LittleEndian>(SK_MAGIC)?;
-        buf.write_u32::<LittleEndian>(SK_FORMAT_VERSION)?;
-        s.ser_into(&mut buf)?;
-
+        let mut buf = s.ser().with_context(|| "failed to serialize state")?;
         // calculate checksum before resize
         let checksum = crc32c::crc32c(&buf);
         buf.extend_from_slice(&checksum.to_le_bytes());
@@ -473,32 +476,36 @@ impl Storage for FileStorage {
             )
         })?;
 
-        // fsync the file
-        control_partial.sync_all().with_context(|| {
-            format!(
-                "failed to sync partial control file at {}",
-                control_partial_path.display()
-            )
-        })?;
+        if sync {
+            // fsync the file
+            control_partial.sync_all().with_context(|| {
+                format!(
+                    "failed to sync partial control file at {}",
+                    control_partial_path.display()
+                )
+            })?;
+        }
 
         let control_path = self.timeline_dir.join(CONTROL_FILE_NAME);
 
         // rename should be atomic
         fs::rename(&control_partial_path, &control_path)?;
-        // this sync is not required by any standard but postgres does this (see durable_rename)
-        File::open(&control_path)
-            .and_then(|f| f.sync_all())
-            .with_context(|| {
-                format!(
-                    "failed to sync control file at: {}",
-                    &control_path.display()
-                )
-            })?;
+        if sync {
+            // this sync is not required by any standard but postgres does this (see durable_rename)
+            File::open(&control_path)
+                .and_then(|f| f.sync_all())
+                .with_context(|| {
+                    format!(
+                        "failed to sync control file at: {}",
+                        &control_path.display()
+                    )
+                })?;
 
-        // fsync the directory (linux specific)
-        File::open(&self.timeline_dir)
-            .and_then(|f| f.sync_all())
-            .with_context(|| "failed to sync control file directory")?;
+            // fsync the directory (linux specific)
+            File::open(&self.timeline_dir)
+                .and_then(|f| f.sync_all())
+                .with_context(|| "failed to sync control file directory")?;
+        }
         Ok(())
     }
 
@@ -675,7 +682,7 @@ mod test {
     use super::FileStorage;
     use crate::{
         safekeeper::{SafeKeeperState, Storage},
-        timeline::{CreateControlFile, CONTROL_FILE_NAME},
+        timeline::{CreateControlFile, SharedState, CONTROL_FILE_NAME},
         SafeKeeperConf,
     };
     use anyhow::Result;
@@ -697,7 +704,7 @@ mod test {
     ) -> Result<(FileStorage, SafeKeeperState)> {
         fs::create_dir_all(&conf.timeline_dir(&timeline_id))
             .expect("failed to create timeline dir");
-        FileStorage::load_from_control_file(conf, timeline_id, create)
+        SharedState::load_from_control_file(conf, timeline_id, create)
     }
 
     #[test]
@@ -710,7 +717,9 @@ mod test {
                     .expect("failed to read state");
             // change something
             state.wal_start_lsn = Lsn(42);
-            storage.persist(&state).expect("failed to persist state");
+            storage
+                .persist(&state, true)
+                .expect("failed to persist state");
         }
 
         let (_, state) = load_from_control_file(&conf, timeline_id, CreateControlFile::False)
@@ -728,7 +737,9 @@ mod test {
                     .expect("failed to read state");
             // change something
             state.wal_start_lsn = Lsn(42);
-            storage.persist(&state).expect("failed to persist state");
+            storage
+                .persist(&state, true)
+                .expect("failed to persist state");
         }
         let control_path = conf.timeline_dir(&timeline_id).join(CONTROL_FILE_NAME);
         let mut data = fs::read(&control_path).unwrap();
@@ -738,7 +749,7 @@ mod test {
         match load_from_control_file(&conf, timeline_id, CreateControlFile::False) {
             Err(err) => assert!(err
                 .to_string()
-                .contains("safekeeper control file checksum mismatch")),
+                .contains("safe keeper state checksum mismatch")),
             Ok(_) => panic!("expected error"),
         }
     }

walkeeper/src/upgrade.rs

@@ -1,60 +0,0 @@
-//! Code to deal with safekeeper control file upgrades
-use crate::safekeeper::{
-    AcceptorState, PgUuid, SafeKeeperState, ServerInfo, Term, TermHistory, TermSwitchEntry,
-};
-use anyhow::{bail, Result};
-use serde::{Deserialize, Serialize};
-use tracing::*;
-use zenith_utils::{bin_ser::LeSer, lsn::Lsn};
-
-/// Persistent consensus state of the acceptor.
-#[derive(Debug, Clone, Serialize, Deserialize)]
-struct AcceptorStateV1 {
-    /// acceptor's last term it voted for (advanced in 1 phase)
-    term: Term,
-    /// acceptor's epoch (advanced, i.e. bumped to 'term' when VCL is reached).
-    epoch: Term,
-}
-
-#[derive(Debug, Clone, Serialize, Deserialize)]
-struct SafeKeeperStateV1 {
-    /// persistent acceptor state
-    acceptor_state: AcceptorStateV1,
-    /// information about server
-    server: ServerInfo,
-    /// Unique id of the last *elected* proposer we dealed with. Not needed
-    /// for correctness, exists for monitoring purposes.
-    proposer_uuid: PgUuid,
-    /// part of WAL acknowledged by quorum and available locally
-    commit_lsn: Lsn,
-    /// minimal LSN which may be needed for recovery of some safekeeper (end_lsn
-    /// of last record streamed to everyone)
-    truncate_lsn: Lsn,
-    // Safekeeper starts receiving WAL from this LSN, zeros before it ought to
-    // be skipped during decoding.
-    wal_start_lsn: Lsn,
-}
-
-pub fn upgrade_control_file(buf: &[u8], version: u32) -> Result<SafeKeeperState> {
-    // migrate to storing full term history
-    if version == 1 {
-        info!("reading safekeeper control file version {}", version);
-        let oldstate = SafeKeeperStateV1::des(&buf[..buf.len()])?;
-        let ac = AcceptorState {
-            term: oldstate.acceptor_state.term,
-            term_history: TermHistory(vec![TermSwitchEntry {
-                term: oldstate.acceptor_state.epoch,
-                lsn: Lsn(0),
-            }]),
-        };
-        return Ok(SafeKeeperState {
-            acceptor_state: ac,
-            server: oldstate.server.clone(),
-            proposer_uuid: oldstate.proposer_uuid,
-            commit_lsn: oldstate.commit_lsn,
-            truncate_lsn: oldstate.truncate_lsn,
-            wal_start_lsn: oldstate.wal_start_lsn,
-        });
-    }
-    bail!("unsupported safekeeper control file version {}", version)
-}

walkeeper/src/wal_service.rs

@@ -3,10 +3,9 @@
 //!   receive WAL from wal_proposer and send it to WAL receivers
 //!
 use anyhow::Result;
-use regex::Regex;
+use log::*;
 use std::net::{TcpListener, TcpStream};
 use std::thread;
-use tracing::*;
 
 use crate::send_wal::SendWalHandler;
 use crate::SafeKeeperConf;
@@ -34,19 +33,9 @@ pub fn thread_main(conf: SafeKeeperConf, listener: TcpListener) -> Result<()> {
     }
 }
 
-// Get unique thread id (Rust internal), with ThreadId removed for shorter printing
-fn get_tid() -> u64 {
-    let tids = format!("{:?}", thread::current().id());
-    let r = Regex::new(r"ThreadId\((\d+)\)").unwrap();
-    let caps = r.captures(&tids).unwrap();
-    caps.get(1).unwrap().as_str().parse().unwrap()
-}
-
 /// This is run by `thread_main` above, inside a background thread.
 ///
 fn handle_socket(socket: TcpStream, conf: SafeKeeperConf) -> Result<()> {
-    let _enter = info_span!("", tid = ?get_tid()).entered();
-
     socket.set_nodelay(true)?;
 
     let mut conn_handler = SendWalHandler::new(conf);