Skip to content

Instantly share code, notes, and snippets.

@pjaudiomv
Created April 24, 2024 18:58
Show Gist options
  • Save pjaudiomv/6e0ef15068c87c5b06bf311b7db15194 to your computer and use it in GitHub Desktop.
Save pjaudiomv/6e0ef15068c87c5b06bf311b7db15194 to your computer and use it in GitHub Desktop.
Terragrunt to Terraform migration. Useful for migrating lots of statefiles down to a single statefile
#!/usr/bin/env python3
import argparse
import json
import subprocess
from collections import defaultdict
# IE.
# parse-state -o terragrunt.tfstate -n terraform.tfstate -m ec2_bastion
def parse_args():
parser = argparse.ArgumentParser(description="Tool to move Terraform state resources.")
parser.add_argument('-a', '--apply', action='store_true', help='Apply changes directly using Terraform.')
parser.add_argument('-o', '--old_state', default='terragrunt.tfstate', help='Path to the old state file.')
parser.add_argument('-m', '--module_name', help='Specific module name to target the move operation.')
parser.add_argument('-n', '--new_state', default='terraform.tfstate', help='Path to the new state file where resources will be moved.')
return parser.parse_args()
def confirm(prompt):
while True:
choice = input(f"{prompt} (y/n): ").lower()
if choice in ['y', 'n']:
return choice == 'y'
print("Invalid input, please enter 'y' or 'n'.")
def load_json_data(file_path):
try:
with open(file_path, 'r') as file:
return json.load(file)
except (IOError, ValueError) as e:
print(f"Error reading {file_path}: {e}")
exit(1)
def extract_resources(data):
resources = data.get('resources', [])
resource_dict = defaultdict(list)
for resource in resources:
if resource['mode'] != 'data':
module_prefix = f"{resource['module']}." if 'module' in resource else ""
for instance in resource.get('instances', []):
index = instance.get('index_key')
identifier = f"{module_prefix}{resource['type']}.{resource['name']}"
if index is not None:
formatted_index = f"[{index}]" if isinstance(index, int) else f"[\"{index}\"]"
resource_dict[identifier].append(f"{identifier}{formatted_index}")
else:
resource_dict[identifier].append(identifier)
return [item for sublist in sorted(resource_dict.values(), key=lambda x: x[0]) for item in sublist]
def move_resources(old_state, new_state, resources, module_prefix):
"""Move resources from one state file to another."""
for resource in resources:
command = [
"terraform", "state", "mv",
"-state", old_state,
"-state-out", new_state,
resource, f"{module_prefix}{resource}"
]
result = subprocess.run(command, capture_output=True, text=True)
if result.returncode != 0:
print(f"Error moving {resource}: {result.stderr}")
else:
print(f"Successfully moved {resource}")
def main():
args = parse_args()
data = load_json_data(args.old_state)
resources = extract_resources(data)
module_prefix = f"module.{args.module_name}." if args.module_name else ""
if args.apply:
if confirm("Are you sure you want to move these resources?"):
move_resources(args.old_state, args.new_state, resources, module_prefix)
else:
print("Operation cancelled.")
else:
for resource in resources:
print(f"terraform state mv -state={args.old_state} -state-out={args.new_state} '{resource}' '{module_prefix}{resource}'")
if __name__ == "__main__":
main()
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment