in mysqloperator/controller/innodbcluster/cluster_api.py [0:0]
def parse(self, spec: dict, prefix: str) -> None:
# TODO other storage types ....
storagespec = dget_dict(spec, "storage", prefix)
if "ociObjectStorage" in storagespec:
ocistorage = dget_dict(storagespec, "ociObjectStorage", prefix+".storage")
self.oci_credentials = dget_str(ocistorage, "credentials", prefix+".storage.credentials")
if "s3" in storagespec:
s3storage = dget_dict(storagespec, "s3", prefix+".storage")
self.s3_region = dget_str(s3storage, "region", prefix+".storage.s3.region")
self.s3_bucket= dget_str(s3storage, "region", prefix+".storage.s3.bucket")
self.s3_object_key_prefix = dget_str(s3storage, "region", prefix+".storage.s3.object_key_prefix")
self.s3_credentials = dget_str(s3storage, "region", prefix+".storage.s3.credentials")
self.s3_host = dget_str(s3storage, "region", prefix+".storage.s3.host", default_value="")
if (not self.oci_credentials and not self.s3_bucket) or (self.oci_credentials and self.s3_bucket):
raise kopf.TemporaryError("Need one of either s3 or ociObjectStorage for MEB Restore")
self.full_backup = dget_str(spec, "fullBackup", prefix)
self.incremental_backups = dget_list(spec, "incrementalBackups", prefix, default_value = [])
if "pitr" in spec:
pitr_spec = dget_dict(spec, "pitr", prefix)
self.pitr_backup_file = dget_str(pitr_spec, "backupFile", prefix+".pitr", default_value="")
self.pitr_binlog_name = dget_str(pitr_spec, "binlogName", prefix+".pitr", default_value="") # cluster.name)
self.pitr_gtid_purge = dget_str(pitr_spec, "gtidPurge", prefix+".pitr", default_value="")
if "end" in pitr_spec:
pitr_end_spec = dget_dict(pitr_spec, "end", prefix+".pitr.end")
if "afterGtids" in pitr_end_spec:
self.pitr_end_term = "SQL_AFTER_GTIDS"
self.pitr_end_value = dget_str(pitr_end_spec, "afterGtids",
prefix+".pitr.end.afterGtids")
elif "beforeGtids" in pitr_end_spec:
self.pitr_end_term = "SQL_BEFORE_GTIDS"
self.pitr_end_value = dget_str(pitr_end_spec, "beforeGtids",
prefix+".pitr.end.beforeGtids")