blob: dc26ffb26eaf97d9f9ee46217035f0dafd63833d [file]
#!/usr/bin/env python3
#
# Copyright 2025, The Android Open Source Project
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
"""A script to generate a pprof report from Android OAT (oatdump) files."""
import argparse
import gzip
import os
import re
import subprocess
import sys
# BEGIN profile_pb2 import
try:
import profile_pb2
except ImportError:
# Allow running the script from a checkout, not as a python_binary_host
ANDROID_BUILD_TOP = os.environ.get('ANDROID_BUILD_TOP')
if ANDROID_BUILD_TOP:
import_path = ANDROID_BUILD_TOP + '/system/extras/simpleperf/scripts'
else:
import_path = os.getcwd() + '/system/extras/simpleperf/scripts'
sys.path.append(import_path)
import profile_pb2
# END profile_pb2 import
# Common file extensions for OAT/ODEX files.
OAT_FILE_EXTENSIONS = (
'.odex',
'.oat',
'@classes.dex',
'@classes.odex',
)
def run_command(command, decode=True):
"""Executes a shell command and returns its output, handling errors."""
try:
process = subprocess.run(
command, check=True, capture_output=True, shell=True
)
if decode:
return process.stdout.decode('utf-8', errors='ignore').strip()
return process.stdout
except subprocess.CalledProcessError as e:
sys.stderr.write(
f"Error running command: '{command}'\nStderr:"
f" {e.stderr.decode('utf-8', errors='ignore')}\n"
)
return None
def load_proguard_map(map_file_path):
"""Parse a Proguard map file and return a deobfuscation mapping."""
mapping = {'classes': {}, 'methods': {}}
current_class_map = None
class_regex = re.compile(r'^(.*) -> (.*):')
method_regex = re.compile(r'^\s+\S+\s+([^\s(]+)\(.*\)\s+->\s+(.*)')
with open(map_file_path, 'r', encoding='utf-8', errors='ignore') as f:
for line in f:
class_match = class_regex.match(line)
if class_match:
original_name, obfuscated_name = class_match.groups()
mapping['classes'][obfuscated_name] = original_name
current_class_map = {
'obfuscated': obfuscated_name,
'original': original_name,
}
elif current_class_map:
method_match = method_regex.match(line)
if method_match:
original_method, obfuscated_method = method_match.groups()
full_obfuscated_key = (
f"{current_class_map['obfuscated']}.{obfuscated_method}"
)
mapping['methods'][full_obfuscated_key] = original_method
return mapping
def stream_and_parse_oatdump(paths, proguard_map=None, is_host_file=False):
"""Yields (oat_file, dex_file, package, class, method, size) tuples from oatdump."""
current_dex_file = None
current_obfuscated_class_name = None
current_method_name = None
dex_file_regex = re.compile(r'^location:\s*(.*?)$')
class_regex = re.compile(r'^\s*[0-9a-fA-F]+:\s+L([^;]+);')
method_regex = re.compile(
r'\s(?:[^\s.]+\.)*([a-zA-Z0-9_$<>]+\(.*\))\s+\(dex_method_idx='
)
code_size_regex = re.compile(r'^\s+CODE:.*size=(\d+)')
for path in paths:
print(f'Parsing {path}...')
current_oat_file = path
process = None
if is_host_file:
try:
lines_iterator = open(path, 'r', encoding='utf-8', errors='ignore')
except FileNotFoundError:
sys.stderr.write(f'Error: Oatdump file not found at {path}.\n')
continue
else:
oatdump_command = (
f'adb shell oatdump --oat-file={path} --no-disassemble --no-dump:vmap'
)
process = subprocess.Popen(
oatdump_command,
shell=True,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
text=True,
errors='ignore',
)
lines_iterator = process.stdout
for line in lines_iterator:
oat_dex_file_match = dex_file_regex.search(line)
if oat_dex_file_match:
current_dex_file = oat_dex_file_match.group(1)
current_obfuscated_class_name = None
current_method_name = None
continue
if current_dex_file:
class_match = class_regex.search(line)
if class_match:
current_obfuscated_class_name = class_match.group(1).replace('/', '.')
current_method_name = None
continue
if current_obfuscated_class_name:
method_match = method_regex.search(line)
if method_match:
current_method_name = method_match.group(1)
continue
if not current_method_name:
continue
size_match = code_size_regex.search(line)
if not size_match:
continue
code_size = int(size_match.group(1))
if code_size > 0:
class_to_process = current_obfuscated_class_name
method_to_process = current_method_name
if proguard_map:
class_to_process = proguard_map['classes'].get(
current_obfuscated_class_name, current_obfuscated_class_name
)
method_name_part = method_to_process.split('(')[0]
params = method_to_process[len(method_name_part) :]
full_obfuscated_key = (
f'{current_obfuscated_class_name}.{method_name_part}'
)
deobfuscated_method_name = proguard_map['methods'].get(
full_obfuscated_key
)
if deobfuscated_method_name:
method_to_process = deobfuscated_method_name + params
package, class_name = (
(class_to_process.rsplit('.', 1))
if '.' in class_to_process
else ('default', class_to_process)
)
if package and class_name and method_to_process:
yield (
current_oat_file,
current_dex_file,
package,
class_name,
method_to_process,
code_size,
)
current_method_name = None
if is_host_file and lines_iterator:
lines_iterator.close()
if process:
stderr_output = process.stderr.read()
if process.wait() != 0:
sys.stderr.write(
f'Warning: Failed to get oatdump for {path}.\nStderr:'
f' {stderr_output}\n'
)
def generate_pprof_report(data, output_path, dex_file_oriented, argv):
"""Generates a pprof file from the parsed oatdump data."""
profile = profile_pb2.Profile()
# Use a dictionary for efficient string-to-ID mapping and a list to maintain
# the order for the protobuf string table.
string_to_id = {'': 0}
id_to_string = ['']
functions = {}
locations = {}
mappings = {}
def get_string_id(s):
"""Adds a string to the string table and returns its ID."""
string_id = string_to_id.get(s)
if string_id is None:
new_id = len(id_to_string)
string_to_id[s] = new_id
id_to_string.append(s)
string_id = new_id
return string_id
st_size = profile.sample_type.add()
st_size.type = get_string_id('space')
st_size.unit = get_string_id('bytes')
for oat_file, dex_file, package_name, class_name, method_name, size in data:
if size == 0:
continue
sample = profile.sample.add()
sample.value.append(size)
# Generate a stack trace for the sample with the following hierarchy:
# - Oat file path (to identify where the code comes from)
# - [Optional] Dex file path (to identify where the dex code comes from)
# - Package name parts, e.g. 'android', 'app'
# - Class name, e.g. 'Activity'
# - Fully qualified method name, e.g. 'android.app.Activity.onCreate(android.os.Bundle)'
stack_frames = []
stack_frames.append((oat_file, os.path.basename(oat_file)))
if dex_file_oriented:
stack_frames.append((dex_file, dex_file))
package_parts = package_name.split('.')
current_package = ''
for part in package_parts:
current_package = f'{current_package}.{part}' if current_package else part
stack_frames.append((current_package, current_package))
full_class_name = f'{current_package}.{class_name}'
stack_frames.append((full_class_name, full_class_name))
full_method_name = f'{package_name}.{class_name}.{method_name}'
stack_frames.append((full_method_name, full_method_name))
location_ids = []
if oat_file not in mappings:
mapping = profile.mapping.add()
mapping.id = len(profile.mapping)
mapping.filename = get_string_id(oat_file)
mappings[oat_file] = mapping.id
mapping_id = mappings[oat_file]
for i, (unique_id, label) in enumerate(stack_frames):
if unique_id not in functions:
func = profile.function.add()
func.id = len(profile.function)
func.name = get_string_id(label)
func.filename = get_string_id(oat_file)
functions[unique_id] = func.id
func_id = functions[unique_id]
loc_key = (func_id, mapping_id)
if loc_key not in locations:
loc = profile.location.add()
loc.id = len(profile.location)
loc.mapping_id = mapping_id
line = loc.line.add()
line.function_id = func_id
locations[loc_key] = loc.id
location_ids.append(locations[loc_key])
sample.location_id.extend(reversed(location_ids))
comment = f"{os.path.basename(argv[0])} {' '.join(argv[1:])}"
profile.comment.append(get_string_id(comment))
profile.string_table.extend(id_to_string)
with gzip.open(output_path, 'wb') as f:
f.write(profile.SerializeToString())
def get_oat_files_for_package(package_name):
"""Finds all OAT/ODEX files for a given package under the primary ABI."""
command_output = run_command(f'adb shell pm art dump {package_name}')
if not command_output:
sys.stderr.write(f'Error: Could not find oat file for {package_name}.\n')
return []
paths = set()
expect_location_for_primary_abi = False
location_regex = re.compile(r'^\s*\[location is (.*?)\]$')
for line in command_output.splitlines():
if expect_location_for_primary_abi:
location_match = location_regex.search(line)
if location_match:
location = location_match.group(1)
if location.endswith(OAT_FILE_EXTENSIONS):
paths.add(location)
expect_location_for_primary_abi = False
elif '[primary-abi]' in line:
expect_location_for_primary_abi = True
continue
if not paths:
sys.stderr.write(f'Error: Failed to parse art dump for {package_name}:\n')
sys.stderr.write(command_output)
sys.stderr.write('\n')
return None
return list(paths)
def get_oat_files_for_process(process_identifier):
"""Finds all .oat and .odex files for a given process name or PID."""
pid = run_command(f'adb shell pidof -s {process_identifier}')
if not pid:
sys.stderr.write(f'Error: Could not find PID for "{process_identifier}".\n')
return None
maps_content = run_command(f'adb shell cat /proc/{pid}/maps')
if not maps_content:
sys.stderr.write(f'Error: Could not read maps for PID {pid}.\n')
return None
oat_files = set()
for line in maps_content.splitlines():
path = line.split()[-1]
if path.endswith(OAT_FILE_EXTENSIONS):
oat_files.add(path)
return list(oat_files)
def ensure_adb_root():
"""Ensures adb connection has root access.
Raises:
subprocess.CalledProcessError: If an adb command fails.
FileNotFoundError: If 'adb' command is not found.
SystemExit: If root access cannot be obtained.
"""
try:
# Attempt to restart adbd as root. This command will block until adbd restarts.
run_command('adb root')
# Wait for the device to reconnect after adb root.
run_command('adb wait-for-device')
# Verify root access by checking the user ID.
result_id = run_command('adb shell id -u')
if result_id != '0':
sys.stderr.write(
'Error: Failed to get root access via adb. Please ensure your device'
' is rooted or adbd can be restarted as root.\n'
)
sys.exit(1)
except subprocess.CalledProcessError as e:
command = e.cmd if isinstance(e.cmd, str) else ' '.join(e.cmd)
sys.stderr.write(
f'Error: adb command failed during root check. Is adb installed and'
f' device connected?\n'
)
sys.stderr.write(f'Command: {command}\n')
sys.stderr.write(f'Stderr: {e.stderr}\n')
sys.exit(1)
except FileNotFoundError:
sys.stderr.write(
"Error: 'adb' command not found. Please ensure adb is in your PATH.\n"
)
sys.exit(1)
def main(argv):
parser = argparse.ArgumentParser(
description='Generate a pprof report from Android OAT files.'
)
group = parser.add_mutually_exclusive_group(required=True)
group.add_argument(
'--package', help='Package name of the app on the connected device.'
)
group.add_argument(
'--process', help='Process name or PID on the connected device.'
)
group.add_argument(
'--oatdump-file', help='Path to a single oatdump output file on the host.'
)
parser.add_argument(
'--output',
default='oatdump_report.pprof',
help='Path to save the generated pprof file.',
)
parser.add_argument(
'--map', help='Optional path to a Proguard deobfuscation map file.'
)
parser.add_argument(
'--dex-file-oriented',
action='store_true',
help='If set, groups the flame graph by dex files.',
)
args = parser.parse_args(argv)
oat_files = []
is_host = False
if args.oatdump_file:
oat_files = [args.oatdump_file]
is_host = True
else:
ensure_adb_root()
if args.package:
oat_files = get_oat_files_for_package(args.package)
elif args.process:
oat_files = get_oat_files_for_process(args.process)
if not oat_files:
sys.stderr.write('Error: Could not find any OAT/ODEX files to process.\n')
sys.exit(1)
proguard_map = load_proguard_map(args.map) if args.map else None
parsed_data = list(
stream_and_parse_oatdump(oat_files, proguard_map, is_host_file=is_host)
)
if not parsed_data:
print('No methods found in oatdump. The report will be empty.')
else:
print(f'Generating pprof report to {args.output}...')
generate_pprof_report(parsed_data, args.output, args.dex_file_oriented, sys.argv)
print('Done.')
print(f'Report saved to: {os.path.abspath(args.output)}')
print(f'Tip: To upload to pprof web UI, run:')
print(f'pprof -flame {os.path.abspath(args.output)}')
if __name__ == '__main__':
main(sys.argv[1:])