use super::*;
impl RdsService {
pub(super) async fn create_db_instance_read_replica(
&self,
request: &AwsRequest,
) -> Result<AwsResponse, AwsServiceError> {
let db_instance_identifier = required_query_param(request, "DBInstanceIdentifier")?;
let source_db_instance_identifier =
optional_query_param(request, "SourceDBInstanceIdentifier")
.or_else(|| optional_query_param(request, "SourceDBClusterIdentifier"))
.ok_or_else(|| db_instance_not_found("(none)"))?;
let (source_instance, db_name) = {
let mut accounts = self.state.write();
let state = accounts.get_or_create(&request.account_id);
if !state.begin_instance_creation(&db_instance_identifier) {
return Err(AwsServiceError::aws_error(
StatusCode::CONFLICT,
"DBInstanceAlreadyExists",
format!("DBInstance {db_instance_identifier} already exists."),
));
}
let source_instance = match state.instances.get(&source_db_instance_identifier).cloned()
{
Some(inst) => inst,
None => {
state.cancel_instance_creation(&db_instance_identifier);
return Err(db_instance_not_found(&source_db_instance_identifier));
}
};
let default_db = default_db_name(&source_instance.engine);
let db_name = source_instance
.db_name
.as_deref()
.unwrap_or(default_db)
.to_string();
(source_instance, db_name)
};
let runtime = self.runtime.as_ref().ok_or_else(|| {
AwsServiceError::aws_error(
StatusCode::SERVICE_UNAVAILABLE,
"InsufficientDBInstanceCapacity",
"Docker/Podman is required for RDS read replicas but is not available",
)
})?;
let dump_data = match runtime
.dump_database(
&source_db_instance_identifier,
&source_instance.engine,
&source_instance.master_username,
&source_instance.master_user_password,
&db_name,
)
.await
{
Ok(data) => data,
Err(e) => {
self.state
.write()
.get_or_create(&request.account_id)
.cancel_instance_creation(&db_instance_identifier);
return Err(runtime_error_to_service_error(e));
}
};
let (dbi_resource_id, db_instance_arn) = {
let accounts = self.state.read();
let empty = RdsState::new(&request.account_id, &request.region);
let s = accounts.get(&request.account_id).unwrap_or(&empty);
(
s.next_dbi_resource_id(),
s.db_instance_arn(&db_instance_identifier),
)
};
let created_at = Utc::now();
let running = match runtime
.ensure_postgres(
&db_instance_identifier,
&source_instance.engine,
&source_instance.engine_version,
&source_instance.master_username,
&source_instance.master_user_password,
&db_name,
&request.account_id,
&request.region,
)
.await
{
Ok(running) => running,
Err(e) => {
self.state
.write()
.get_or_create(&request.account_id)
.cancel_instance_creation(&db_instance_identifier);
return Err(runtime_error_to_service_error(e));
}
};
if let Err(e) = runtime
.restore_database(
&db_instance_identifier,
&source_instance.engine,
&source_instance.master_username,
&source_instance.master_user_password,
&db_name,
&dump_data,
)
.await
{
self.state
.write()
.get_or_create(&request.account_id)
.cancel_instance_creation(&db_instance_identifier);
runtime.stop_container(&db_instance_identifier).await;
return Err(runtime_error_to_service_error(e));
}
let replica = build_read_replica_instance(
&db_instance_identifier,
db_instance_arn,
dbi_resource_id,
created_at,
&source_db_instance_identifier,
&source_instance,
&running,
);
let source_missing = {
let mut accounts = self.state.write();
let state = accounts.get_or_create(&request.account_id);
match state.instances.get_mut(&source_db_instance_identifier) {
Some(source) => {
source
.read_replica_db_instance_identifiers
.push(db_instance_identifier.clone());
state.finish_instance_creation(replica.clone());
false
}
None => {
state.cancel_instance_creation(&db_instance_identifier);
true
}
}
};
if source_missing {
runtime.stop_container(&db_instance_identifier).await;
return Err(db_instance_not_found(&source_db_instance_identifier));
}
self.emit_event(
RdsSourceType::DbInstance,
&db_instance_identifier,
&replica.db_instance_arn,
"RDS-EVENT-0005",
&["creation", "read replica"],
"Read replica DB instance created",
);
Ok(AwsResponse::xml(
StatusCode::OK,
query_response_xml(
"CreateDBInstanceReadReplica",
RDS_NS,
&format!(
"<DBInstance>{}</DBInstance>",
db_instance_xml(&replica, None)
),
&request.request_id,
),
))
}
}