[Spring] PostgreSQL Read/Write Replication with R2DBC

안녕하세요? 제이온입니다.

오늘은 인프라 환경에서 PostgreSQL Read/Write Replication 세팅이 되어 있다고 가정할 때, 스프링 애플리케이션 쪽에서 Replication을 위한 설정을 어떤 것을 해야 하는지, 그리고 테스트는 어떻게 해야 하는지에 관한 가이드를 작성해 보겠습니다.

 

Read/Write End-Point 분리하기

먼저 프로덕션 코드 상에서 읽기 작업 시 Read용 데이터베이스로, 쓰기 작업시 Write용 데이터베이스로 End-Point를 분리하여 쿼리를 요청해야 합니다.

 

가장 쉬운 방법은 ReadConnectionFactory, WriteConnectionFactory 2개의 빈을 생성하고 Repository에 그 빈들을 모두 넣어준 후, 적절히 읽기 작업시 ReadConnectionFactory를 쓰고 쓰기 작업시 WriteConnectionFactory를 쓰는 것입니다.

 

하지만 위 방법은 Repository 내부에서 개발자가 실수로 Connection을 관리할 위험성이 있으므로, 저는 ConnectionFactory는 하나만 두고 특정 기준에 따라 동적으로 ReadConnectionFactory나 WriteConnectionFactory로 라우팅하는 방법을 채택했습니다.

 

class MultiTenantRoutingConnectionFactory : AbstractRoutingConnectionFactory() {

    companion object {
        const val READ_DATA_SOURCE_KEY = "readDataSourceKey"
        const val WRITE_DATA_SOURCE_KEY = "writeDataSourceKey"
    }

    private val logger = PortLoggerFactory.getPortLogger(this.javaClass)

    override fun determineCurrentLookupKey(): Mono<Any> =
        TransactionSynchronizationManager.forCurrentTransaction().map { transactionSynchronizeManager ->
            logger.info("it.getCurrentTransactionName() : ${transactionSynchronizeManager.currentTransactionName}")
            logger.info("it.isActualTransactionActive() : ${transactionSynchronizeManager.isActualTransactionActive}")
            logger.info("it.isCurrentTransactionReadOnly() : ${transactionSynchronizeManager.isCurrentTransactionReadOnly}")

            val dataSourceType = if (transactionSynchronizeManager.isActualTransactionActive) {
                logger.info("isActualTransactionActive....")
                if (transactionSynchronizeManager.isCurrentTransactionReadOnly) {
                    READ_DATA_SOURCE_KEY
                } else {
                    WRITE_DATA_SOURCE_KEY
                }
            } else {
                READ_DATA_SOURCE_KEY
            }
            logger.info("> current dataSourceType: $dataSourceType")
            dataSourceType
        }
}

 

r2dbc는 AbstractRoutingConnectionFactory 클래스를 상속하고 determineCurrentLookupKey() 메서드를 재정의함으로써 ConnectionFactory를 라우팅할 수 있습니다.

 

저는 Transaction의 Read-Only가 true일 때는 ReadConnectionFactory를, false일 때는 WriteConnectionFactory로 선택하도록 구현하였습니다.

 

이렇게 세팅한 MultiTenantRoutingConnectionFactory를 살짝만 응용하여 아래와 같이 ConnectionFactory를 빈으로 등록합니다.

 

    @Primary
    @Bean
    override fun connectionFactory(): ConnectionFactory {
        val multiTenantRoutingConnectionFactory = MultiTenantRoutingConnectionFactory()

        val factories = mapOf(
            MultiTenantRoutingConnectionFactory.READ_DATA_SOURCE_KEY to readConnectionFactory(),
            MultiTenantRoutingConnectionFactory.WRITE_DATA_SOURCE_KEY to writeConnectionFactory()
        )

        multiTenantRoutingConnectionFactory.setLenientFallback(true)
        multiTenantRoutingConnectionFactory.setDefaultTargetConnectionFactory(readConnectionFactory())
        multiTenantRoutingConnectionFactory.setTargetConnectionFactories(factories)
        return multiTenantRoutingConnectionFactory
    }

    @Bean
    fun readConnectionFactory(): ConnectionFactory {
        val connectionFactory = ConnectionFactories.get(
            builder()
                .option(DRIVER, "pool")
                .option(PROTOCOL, "postgresql")
                .option(HOST, env.getProperty("spring.r2dbc.postgresql.read.host", "localhost"))
                .option(PORT, env.getProperty("spring.r2dbc.postgresql.read.port", "25432").toInt())
                .option(DATABASE, env.getProperty("spring.r2dbc.postgresql.read.name", "port"))
                .option(USER, env.getProperty("spring.r2dbc.postgresql.read.username", "postgres"))
                .option(PASSWORD, env.getProperty("spring.r2dbc.postgresql.read.password", "postgres"))
                .build()
        )
        println(env.getProperty("spring.r2dbc.postgresql.read.port", "25432").toInt())
        return configurateConnectionPool(connectionFactory)
    }

    @Bean
    fun writeConnectionFactory(): ConnectionFactory {
        val connectionFactory = ConnectionFactories.get(
            builder()
                .option(DRIVER, "pool")
                .option(PROTOCOL, "postgresql")
                .option(HOST, env.getProperty("spring.r2dbc.postgresql.write.host", "localhost"))
                .option(PORT, env.getProperty("spring.r2dbc.postgresql.write.port", "5432").toInt())
                .option(DATABASE, env.getProperty("spring.r2dbc.postgresql.write.name", "port"))
                .option(USER, env.getProperty("spring.r2dbc.postgresql.write.username", "postgres"))
                .option(PASSWORD, env.getProperty("spring.r2dbc.postgresql.write.password", "postgres"))
                .build()
        )
        println(env.getProperty("spring.r2dbc.postgresql.write.port", "5432").toInt())
        return configurateConnectionPool(connectionFactory)
    }

    @Bean
    fun reactiveTransactionManager(connectionFactory: ConnectionFactory): ReactiveTransactionManager {
        return R2dbcTransactionManager(TransactionAwareConnectionFactoryProxy(connectionFactory))
    }

 

코드는 간단합니다. 아까 라우팅할 때 쓰인 KEY값을 실제 ConnectionFactory에 연결만 하면 끝입니다. 참고로, 스프링 트랜잭션을 사용하기 위해 ReactiveTransactionManager 빈은 꼭 등록이 되어야 합니다.

 

Repository는 위의 connectionFactory 객체를 주입받고, Application은 트랜잭션을 설정할 때 read-only 값을 설정한다면 성공적으로 Read/Write End-Point를 나눌 수 있게 됩니다.

 

@Repository
class SpringDataTestRepository(
    connectionFactory: ConnectionFactory
) : TestRepository {
    val template: R2dbcEntityTemplate = R2dbcEntityTemplate(connectionFactory)

    // ...
}

@Service
class TestService(
    val testRepository: TestRepository,
) {
    @Transactional(readOnly = true)
    suspend fun getTest(request: TestRqeust): TestResult =
        // ...
}

 

테스트 컨테이너를 활용하여 Read/Write Replication 검증하기

위 코드처럼 DB Read/Write End-Point를 나누는 것은 단순히 쿼리 요청 커넥션만 분리하는 일이므로 어렵지 않습니다. 하지만 실제로 테스트 상에서 End-Point가 잘 분리되는지, 더 나아가 Write 서버가 Read 서버로 동기화를 잘 해주는지 까지 테스트하고 싶다면 어떻게 할 수 있을까요?

 

저는 실제 운영 DB와의 스펙을 일치시키기 위해 TestContainers 라이브러리를 활용한 테스트 컨테이너로 테스트해 보겠습니다.

 

먼저 아래와 같이 gradle에 의존성을 추가합니다. 현 프로젝트는 테스트 컨테이너를 testFixture 모듈 안에서 구현하였으므로 testFixturesImplementation을 사용하였습니다.

 

testFixturesImplementation("org.testcontainers:testcontainers:${Version.testContainers}")
testFixturesImplementation("org.testcontainers:r2dbc:${Version.testContainers}")
testFixturesImplementation("org.testcontainers:postgresql:${Version.testContainers}")
testFixturesImplementation("org.testcontainers:junit-jupiter:${Version.testContainers}")
testFixturesImplementation("org.springframework.boot:spring-boot-starter-data-r2dbc")
testFixturesImplementation("io.kotest:kotest-runner-junit5:${Version.kotest}")
testFixturesImplementation("io.r2dbc:r2dbc-postgresql:${Version.r2dbcPostgresql}")

 

만약 트랜잭션의 read-only 값에 따라 커넥션이 잘 라우팅되어 전달되는지 확인하고 싶다면, 아래처럼 테스트 컨테이너를 심플하게 2개 띄우는 코드를 작성하면 됩니다.

 

class TestPostgreSQLContainer : PostgreSQLContainer<TestPostgreSQLContainer>("docker.io/bitnami/postgresql:13") {

    companion object {

        private lateinit var writeInstance: TestPostgreSQLContainer
        private lateinit var readInstance: TestPostgreSQLContainer

        private lateinit var writeDatabaseClient: DatabaseClient
        private lateinit var readDatabaseClient: DatabaseClient

        private lateinit var writeConnectionFactory: PostgresqlConnectionFactory
        private lateinit var readConnectionFactory: PostgresqlConnectionFactory

        init {
            start()
        }

        private fun getConnectionFactoryOption(instance: TestPostgreSQLContainer): ConnectionFactoryOptions {
            return PostgreSQLR2DBCDatabaseContainer.getOptions(instance)
        }

        fun sql(sql: String) {
            writeDatabaseClient.sql(sql).then().block()
        }

        fun start() {
            if (!Companion::writeInstance.isInitialized) {
                writeInstance = TestPostgreSQLContainer()
                    .withUsername("postgres")
                    .withPassword("postgres")
                    .withStartupTimeout(Duration.ofSeconds(60))
                    .withDatabaseName("master-database")
                    .apply { start() }

                writeConnectionFactory = PostgresqlConnectionFactoryProvider().create(getConnectionFactoryOption(writeInstance))
                writeDatabaseClient = DatabaseClient.create(writeConnectionFactory)
            }

            if (!Companion::readInstance.isInitialized) {
                readInstance = TestPostgreSQLContainer()
                    .withUsername("postgres")
                    .withPassword("postgres")
                    .withStartupTimeout(Duration.ofSeconds(60))
                    .withDatabaseName("slave-database")
                    .apply { start() }

                readConnectionFactory = PostgresqlConnectionFactoryProvider().create(getConnectionFactoryOption(readInstance))
                readDatabaseClient = DatabaseClient.create(readConnectionFactory)
            }
        }

        fun stop() {
            writeInstance.stop()
            readInstance.stop()
        }
    }
}

 

이 코드에 문제점은 무엇일까요? Transaction의 read-only 값에 따라 Connection 라우팅 테스트는 성공하겠지만, Write 데이터베이스에 데이터를 업데이트하면, Read 데이터베이스에는 그 데이터가 동기화되지 않아서 통합 테스트시 대부분의 테스트가 실패하게 됩니다. 따라서 Write 데이터베이스의 데이터를 Read 데이터베이스에 반영해 주는 설정이 필요합니다.

 

PostgreSQL에서 Read/Write Replication를 구현하는 원리

테스트를 위한 세팅에 앞서, PostgreSQL은 Read/Write Replication을 어떻게 하는지 살펴 봅시다.

 

Log Shipping

 

• file-based 복제로 master 서버의 WAL file이 생성되면 이 파일을 scp를 통해 standby 서버로 전달하여 반영합니다.
• master 서버의 wal file이 생길 때 까지 replication gap이 발생할 수 있다는 단점이 있습니다.

 

Logical Replication

 

• pub / sub 구조로 양방향 replication이 가능합니다.
• 특정 테이블만 복제하는 partial replication만 가능합니다.
• DDL 복제는 안되기 때문에 각각 별도로 수행해 줘야 한다는 단점이 있습니다.

 

Streaming Replication

 

• Master 서버는 standby 서버에게 transaction log entires를 전달하고 standby 서버는 WAL file을 기다리지 않고 record 단위로 복제를 수행하는 방식입니다.
• 일반적으로 가장 많이 사용되는 복제 방식이지만, standby 서버에 긴 장애가 발생할 경우 master 서버의 wal 파일을 재활용하므로 데이터가 유실될 가능성이 있습니다. 가령, standby 서버에 긴 장애가 발생하였고, 그동안 master 서버가 가장 오래된 wal 파일을 덮어 써 버린다면, standby 서버가 복구된 이후에도 유실된 wal 내용을 복원할 방법이 없습니다.
• 따라서 일반적으로 Streaming Replication 방식을 사용하되, 만약을 위해 Log Shipping 방식도 사용하는 편입니다.

 

TestContainers에서 PostgreSQL Read/Write Replication 구현하기

위에 PostgreSQL Replication 원리를 소개하였지만, 실제로 세팅하는 과정은 여러 conf 파일을 수정하고 wal를 위한 디렉토리를 만들고, master 서버와 slave 서버 간의 포트를 열어주고, 리플리케이션을 위한 유저를 만드는 등 꽤나 복잡한 설정이 들어갑니다.

 

이들을 하려면 할 수 있겠지만, 테스트 자동화하는 입장에서 제대로 동작하도록 만드는 것은 꽤나 까다롭습니다. 그렇다면, 우리는 어떻게 테스트 컨테이너에서 PostgreSQL Replication을 구현할 수 있을까요?

 

다행히, Replication을 편하게 할 수 있도록 이미 만들어진 bitnami/postgresql:13라는 이미지가 존재합니다! (https://hub.docker.com/r/bitnami/postgresql)

 

해당 사이트에 접속해 보면 아주 친절하게도 Streaming Replication 방식의 Replication 세팅 가이드가 나와 있습니다.

 

 

저는 이 포스팅에 상세한 Replication 세팅법을 적지는 않고, 바로 docker-compose.yml을 사용하여 테스트 컨테이너를 설정해 보겠습니다.

 

우선 다음과 같이 docker-compose.yml을 작성하고, classpath에 해당하는 디렉토리에 저장해 줍니다.

 

version: '3'

networks:
  my-network:

services:
  postgresql-master:
    image: 'docker.io/bitnami/postgresql:13'
    environment:
      - POSTGRESQL_USERNAME=postgres
      - POSTGRESQL_DATABASE=my_database
      - POSTGRESQL_PASSWORD=postgres
      - POSTGRESQL_PGAUDIT_LOG=READ,WRITE
      - POSTGRESQL_LOG_HOSTNAME=true
      - POSTGRESQL_REPLICATION_MODE=master
      - POSTGRESQL_REPLICATION_USER=repl_user
      - POSTGRESQL_REPLICATION_PASSWORD=repl_password
    networks:
      my-network:
        aliases:
          - postgresql
  postgresql-slave:
    image: 'docker.io/bitnami/postgresql:13'
    depends_on:
      - postgresql-master
    environment:
      - POSTGRESQL_USERNAME=postgres
      - POSTGRESQL_PASSWORD=postgres
      - POSTGRESQL_MASTER_HOST=postgresql-master
      - POSTGRESQL_PGAUDIT_LOG=READ,WRITE
      - POSTGRESQL_LOG_HOSTNAME=true
      - POSTGRESQL_REPLICATION_MODE=slave
      - POSTGRESQL_REPLICATION_USER=repl_user
      - POSTGRESQL_REPLICATION_PASSWORD=repl_password
      - POSTGRESQL_MASTER_PORT_NUMBER=5432
    networks:
      my-network:

 

기존 bitnami/postgresql 세팅과 거의 유사합니다. 저는 두 DB를 하나의 도커 네트워크로 속하도록 설정하였고, 명시적인 포트 포워딩은 하지 않았습니다. 특히 포트 포워딩 쪽은 병렬 테스트할 때 컨테이너가 2개 이상 띄워지는 것을 대비해야 하기 때문입니다. (안 그러면 포트가 겹치는 ㅠㅠ)

 

그리고 다음과 같이 Docker-Compose를 활용한 테스트 컨테이너 코드를 작성해 줍니다.

 

class TestDockerComposeContainer :
    DockerComposeContainer<TestDockerComposeContainer>(
        File(MountableFile.forClasspathResource("testcontainers/docker-compose.yml").resolvedPath)
    ) {

    companion object {

        private lateinit var testDockerComposeContainer: TestDockerComposeContainer
        private lateinit var masterDatabaseClient: DatabaseClient

        lateinit var masterConnectionFactory: PostgresqlConnectionFactory

        init {
            start()
        }

        fun sql(sql: String) {
            masterDatabaseClient.sql(sql).then().block()
            Thread.sleep(100)
        }

        fun start() {
            if (!Companion::testDockerComposeContainer.isInitialized) {
                val masterContainerName = "postgresql-master"
                val slaveContainerName = "postgresql-slave"

                testDockerComposeContainer = TestDockerComposeContainer()
                    .withExposedService(masterContainerName, 5432)
                    .withExposedService(slaveContainerName, 5432)
                    .apply {
                        start()
                    }
                Thread.sleep(3000)

                val masterContainerState = testDockerComposeContainer.getContainerByServiceName("${masterContainerName}_1").get()
                val masterContainerPort = testDockerComposeContainer.getServicePort(masterContainerName, 5432)

                masterConnectionFactory = masterContainerState.run {
                    PostgresqlConnectionFactoryProvider().create(
                        ConnectionFactoryOptions.builder()
                            .option(ConnectionFactoryOptions.DRIVER, "pool")
                            .option(ConnectionFactoryOptions.PROTOCOL, "postgresql")
                            .option(ConnectionFactoryOptions.HOST, this.host.toString())
                            .option(ConnectionFactoryOptions.PORT, masterContainerPort)
                            .option(ConnectionFactoryOptions.DATABASE, "my_database")
                            .option(ConnectionFactoryOptions.USER, "postgres")
                            .option(ConnectionFactoryOptions.PASSWORD, "postgres")
                            .build()
                    )
                }
                masterDatabaseClient = DatabaseClient.create(masterConnectionFactory)

                val slaveContainerState = testDockerComposeContainer.getContainerByServiceName("${slaveContainerName}_1").get()
                val slaveContainerPort = testDockerComposeContainer.getServicePort(slaveContainerName, 5432)

                System.setProperty("spring.r2dbc.postgresql.write.host", masterContainerState.host)
                System.setProperty("spring.r2dbc.postgresql.write.port", masterContainerPort.toString())
                System.setProperty("spring.r2dbc.postgresql.write.name", "my_database")
                System.setProperty("spring.r2dbc.postgresql.write.username", "postgres")
                System.setProperty("spring.r2dbc.postgresql.write.password", "postgres")

                System.setProperty("spring.r2dbc.postgresql.read.host", slaveContainerState.host)
                System.setProperty("spring.r2dbc.postgresql.read.port", slaveContainerPort.toString())
                System.setProperty("spring.r2dbc.postgresql.read.name", "my_database")
                System.setProperty("spring.r2dbc.postgresql.read.username", "postgres")
                System.setProperty("spring.r2dbc.postgresql.read.password", "postgres")
            }
        }

        fun stop() {
            testDockerComposeContainer.stop()
        }
    }
}

 

코드 자체는 어렵지 않으나, 몇 가지 주의해야 할 세팅이 있습니다.

 

• docker-compose.yml 파일 명시
  • DockerComposeContainer 객체를 생성할 때 docker-compose.yml이 담긴 java.io.File 객체를 전달해야 하는데, File 객체 생성 시 classpath를 지정하는 설정을 못 찾았습니다.
  • 그래서 TestContainers에서 제공하는 MountableFile를 활용하여 classpath로 docker-compsoe.yml을 전달하고 실제 파일 경로를 얻어 왔습니다.
• withExposedService
  • 특정 컨테이너에 대해 랜덤 포트 포워딩을 할 수 있습니다.
  • 위와 같이 설정하면 xxxx:5432와 같이 포트 포워딩됩니다.
• start
  • start() 메서드를 실행하면 docker-compose up이 되는 것이지, 실제로 테스트 컨테이너가 구동될 때까지 기다리지는 않습니다.
  • 그래서 명시적으로 3초를 대기해 줬습니다. Wait 객체를 활용하여 더 우아하게 기다리는 방법이 있었으나 제대로 동작하게끔 세팅을 못해서 일단 조금 비효율적인 방법으로 뒀습니다.
• getContainerByServiceName("${masterContainerName}_1")
  • 이상하게도 getServicePort()나 getServiceHost()는 컨테이너의 이름을 넣어도 잘 포트나 호스트 정보를 얻어올 수 있는데, getContainerByServiceName()할 때만 컨테이너 이름 뒤에 _1을 붙여야 합니다.
  • 이건 뾰족한 우회법을 못 찾아서 이렇게 뒀습니다.

 

이렇게 세팅을 완료하면 테스트 컨테이너가 2개 띄워지고, 쓰기 작업 시 Read 데이터 베이스에도 정상적으로 동기화가 성공합니다.