use std::fmt;
use std::fs;
use std::path::Path;
use std::thread;
use std::time;
use serde_json as json;
use crate::api::ApiResult;
use crate::latest;
use crate::output::Output;
use crate::show::Show;
use crate::util;
use crate::VolumeCreate;
const SECOND: time::Duration = time::Duration::from_secs(1);
#[derive(Debug)]
pub(crate) struct Volume {
api: super::Api,
path: String,
}
impl Volume {
pub(crate) fn new(api: super::Api) -> Self {
let path = String::from("/volumes");
Self { api, path }
}
pub(crate) fn show(&self, volume: impl AsRef<str>) -> Result<String, anyhow::Error> {
self.get_volume(volume.as_ref()).map(Show::show)
}
pub(crate) fn create(
&self,
create: &VolumeCreate,
wait: bool,
) -> Result<String, anyhow::Error> {
match create {
VolumeCreate::Body { path } => self.create_from_file(path, wait),
VolumeCreate::Template { path } => self.generate_template(path),
}
}
pub(crate) fn create_from_file(
&self,
path: &Path,
wait: bool,
) -> Result<String, anyhow::Error> {
let text = fs::read_to_string(path)?;
let body = json::from_str(&text)?;
self.create_impl(body, wait)
}
pub(crate) fn generate_template(&self, path: &Path) -> Result<String, anyhow::Error> {
let realm1 = latest::AwsReplicaCreateRealm {
realm: "AWS_Account1".to_string(),
region: "us-west-2".to_string(),
subnet: "subnet-1234567890".to_string(),
};
let realm2 = latest::AzureReplicaCreateRealm {
realm: "Azure_Subscription1".to_string(),
resource_group: "EU_Deployment".to_string(),
subnet: "lore/default".to_string(),
};
let realm3 = latest::AwsReplicaCreateRealm {
realm: "AWS_Account2".to_string(),
region: "eu-west-2".to_string(),
subnet: "subnet-1234567890".to_string(),
};
let replica1 = latest::ReplicaCreate {
name: "sfo".to_string(),
realm: latest::ReplicaCreateRealm::Aws(realm1),
primary: true,
..Default::default()
};
let replica2 = latest::ReplicaCreate {
name: "ewr".to_string(),
realm: latest::ReplicaCreateRealm::Azure(realm2),
..Default::default()
};
let replica3 = latest::ReplicaCreate {
name: "fra".to_string(),
realm: latest::ReplicaCreateRealm::Aws(realm3),
..Default::default()
};
let template = latest::VolumeCreate {
name: String::from("dragon"),
description: Some(String::from("Golden dragon")),
size_gb: Some(17),
replicas: vec![replica1, replica2, replica3],
..Default::default()
};
let text = json::to_string_pretty(&template)?;
fs::write(path, text)?;
Ok(format!("Template created: {}", path.display()))
}
pub(crate) fn add_replica_from_file(
&self,
volume: impl AsRef<str>,
path: impl AsRef<Path>,
wait: bool,
) -> Result<String, anyhow::Error> {
let text = fs::read_to_string(path)?;
let body = json::from_str(&text)?;
self.add_replica_impl(volume, body, wait)
}
pub(crate) fn delete(
&self,
volume: impl AsRef<str>,
delete_native: bool,
force: bool,
wait: bool,
) -> Result<String, anyhow::Error> {
let volume = volume.as_ref();
log::debug!("sending delete volume");
let path = format!("{}/{}", self.path, volume);
let params = [("delete_native", delete_native), ("force", force)];
let volume: Output<latest::Volume> = self.api.del(path, ¶ms)?;
if wait {
log::debug!("deleting volume status {}", volume.status.value);
let name = &volume.name;
for delay in util::stepping_down_delay(16 * SECOND) {
thread::sleep(delay);
match self.get_volume(name) {
Ok(volume) => {
if volume.status.value != "deleting" && volume.status.value != "online" {
log::debug!(
"Stop polling deleting volume (status: {})",
volume.status.value
);
break;
}
}
Err(e) => {
log::debug!("Stop polling deleting volume (get_volume returned: {})", e);
break;
}
};
}
}
Ok(String::new())
}
pub(crate) fn list(&self) -> Result<String, anyhow::Error> {
self.api
.get::<_, Vec<latest::Volume>>(&self.path)
.map(|v| v.into_iter().map(|v| v.name).collect::<Vec<_>>().join("\n"))
}
pub(crate) fn set_primary(
&self,
volume: impl AsRef<str>,
replica: impl AsRef<str>,
) -> Result<String, anyhow::Error> {
let volume = volume.as_ref();
let replica = replica.as_ref();
log::info!(
"volume \"{}\": selecting replica \"{}\" as primary",
volume,
replica
);
let path = format!("{}/{}/replicas/{}/primary", self.path, volume, replica);
let replica_response = self.api.put::<_, latest::Replica>(path)?;
log::debug!(
"set-primary replica status {}",
replica_response.status.value
);
for delay in util::stepping_down_delay(3 * SECOND) {
thread::sleep(delay);
let volume = self.get_volume(&volume)?;
let is_primary = volume
.replicas
.iter()
.find(|r| r.name == replica)
.map(|r| r.primary)
.unwrap_or_default();
if is_primary && volume.status.value == "online" {
break;
}
}
Ok(String::new())
}
pub(crate) fn add_replica_impl(
&self,
volume: impl AsRef<str>,
add_replica: latest::ReplicaCreate,
wait: bool,
) -> Result<String, anyhow::Error> {
let volume = volume.as_ref();
let replica = add_replica.name.to_owned();
log::info!("volume \"{}\": adding replica \"{}\"", volume, replica);
let path = format!("{}/{}/replicas", self.path, volume);
let response = self.api.post::<_, _, latest::Replica>(path, add_replica)?;
if !response.is_typed() {
return Ok(response.show());
}
if wait {
log::debug!("add-replica replica status {}", response.status.value);
for delay in util::stepping_down_delay(3 * SECOND) {
thread::sleep(delay);
let volume = self.get_volume(&volume)?;
let replica = volume.replicas.iter().find(|r| r.name == replica);
match replica {
Some(replica) if replica.status.value == "online" => break,
_ => continue,
}
}
}
Ok(String::new())
}
pub(crate) fn remove_replica(
&self,
volume: impl AsRef<str>,
replica: impl AsRef<str>,
wait: bool,
) -> Result<String, anyhow::Error> {
let volume = volume.as_ref();
let replica = replica.as_ref();
log::info!("volume \"{}\": removing replica \"{}\"", volume, replica);
let path = format!("{}/{}/replicas/{}", self.path, volume, replica);
let response = self
.api
.del::<_, Option<(&str, &str)>, &str, &str, latest::Replica>(path, None)?;
if !response.is_typed() {
return Ok(response.show());
}
if wait {
log::debug!("remove-replica replica status {}", response.status.value);
for delay in util::stepping_down_delay(3 * SECOND) {
thread::sleep(delay);
let volume = self.get_volume(&volume)?;
let replica = volume.replicas.iter().find(|r| r.name == replica);
match replica {
None => break,
_ => continue,
}
}
}
Ok(String::new())
}
pub(crate) fn snapshot_list(&self, volume: impl AsRef<str>) -> Result<String, anyhow::Error> {
let volume = volume.as_ref();
log::info!("volume \"{}\": listing snapshots", volume);
let path = format!("{}/{}/snapshots", self.path, volume);
self.api
.get::<_, Vec<latest::Snapshot>>(path)
.map(|v| v.into_iter().map(|v| v.name).collect::<Vec<_>>().join("\n"))
}
pub(crate) fn snapshot_create(
&self,
volume: impl AsRef<str>,
name: impl AsRef<str>,
wait: bool,
) -> Result<String, anyhow::Error> {
let volume = volume.as_ref();
let name = name.as_ref();
log::info!("volume \"{}\": creating snapshot \"{}\"", volume, name);
let path = format!("{}/{}/snapshots", self.path, volume);
let create = latest::SnapshotCreate {
name: Some(name.to_string()),
};
let snapshot = self.api.post::<_, _, latest::Snapshot>(path, create)?;
if !snapshot.is_typed() {
return Ok(snapshot.show());
}
if wait {
log::debug!("snapshot-create status {}", snapshot.status.value);
for delay in util::stepping_down_delay(10 * SECOND) {
thread::sleep(delay);
let snapshot = self.get_snapshot(volume, name)?;
if snapshot.status.value == "online" {
break;
} else {
continue;
}
}
}
Ok(String::new())
}
pub(crate) fn snapshot_delete(
&self,
volume: impl AsRef<str>,
name: impl AsRef<str>,
wait: bool,
) -> Result<String, anyhow::Error> {
let volume = volume.as_ref();
let name = name.as_ref();
log::info!("volume \"{}\": deleting snapshot \"{}\"", volume, name);
let path = format!("{}/{}/snapshots/{}", self.path, volume, name);
let params = [("delete", "sure")];
let snapshot: Output<latest::Snapshot> = self.api.del(path, ¶ms)?;
if !snapshot.is_typed() {
return Ok(snapshot.show());
}
if wait {
log::debug!("snapshot-delete status {}", snapshot.status.value);
for delay in util::stepping_down_delay(2 * SECOND) {
thread::sleep(delay);
match self.get_snapshot(volume, name) {
Ok(snapshot) => {
let status = &snapshot.status.value;
if status != "deleting" && status != "online" {
log::debug!("Stop polling deleting snapshot (status: {})", status);
break;
} else {
continue;
}
}
Err(e) => {
log::debug!(
"Stop polling deleting snapshot (get_snapshot returned: {})",
e
);
break;
}
}
}
}
Ok(String::new())
}
pub(crate) fn snapshot_export_aws(
&self,
volume: impl AsRef<str>,
name: impl AsRef<str>,
realm: impl AsRef<str>,
region: impl AsRef<str>,
) -> Result<String, anyhow::Error> {
let volume = volume.as_ref();
let name = name.as_ref().to_string();
log::info!("volume \"{}\": exporting snapshot \"{}\"", volume, name);
let path = format!("{}/{}/snapshots/{}/exports", self.path, volume, name);
let export = latest::SnapshotExportCreate::aws(realm, region);
let export = self
.api
.post::<_, _, latest::SnapshotExport>(path, export)?;
if !export.is_typed() {
return Ok(export.show());
}
let snapshot_id = export.aws_snapshot_id().unwrap_or_default();
Ok(snapshot_id)
}
pub(crate) fn wait_for_volume_status(
&self,
volume: &str,
replica: Option<&str>,
primary: bool,
status: &str,
) -> Result<String, anyhow::Error> {
for delay in util::stepping_down_delay(8 * SECOND) {
let volume = self.get_volume(volume)?;
if let Some(replica) = replica {
let replica = volume
.replicas
.iter()
.find(|r| r.name == replica)
.filter(|replica| !primary || replica.primary)
.ok_or_else(|| anyhow::Error::msg("No such replica"))?;
if replica.status.value == status {
break;
}
} else if volume.status.value == status {
break;
}
thread::sleep(delay)
}
Ok(String::new())
}
fn get_volume(&self, volume: impl fmt::Display) -> ApiResult<latest::Volume> {
let path = format!("{}/{}", self.path, volume);
self.api.get(path)
}
fn get_snapshot(
&self,
volume: impl fmt::Display,
snapshot: impl fmt::Display,
) -> ApiResult<latest::Snapshot> {
let path = format!("{}/{}/snapshots/{}", self.path, volume, snapshot);
self.api.get(path)
}
fn create_impl(
&self,
create: latest::VolumeCreate,
wait: bool,
) -> Result<String, anyhow::Error> {
let volume = self.api.post::<_, _, latest::Volume>(&self.path, create)?;
if !volume.is_typed() {
return Ok(volume.show());
}
if wait {
log::debug!("creating volume status {}", volume.status.value);
let name = &volume.name;
for delay in util::stepping_down_delay(16 * SECOND) {
thread::sleep(delay);
let volume = self.get_volume(name)?;
log::debug!("polling creating volume status: {}", volume.status.value);
if volume.status.value == "online" || volume.status.value == "error" {
log::debug!(
"Stop polling creating volume (status: {})",
volume.status.value
);
break;
}
}
}
Ok(String::new())
}
}