#!/usr/bin/env python3
#
# This file is open source software, licensed to you under the terms
# of the Apache License, Version 2.0 (the "License").  See the NOTICE file
# distributed with this work for additional information regarding copyright
# ownership.  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.
#
# Copyright (C) 2017 ScyllaDB

import argparse
from typing import Any, Optional, Sequence, TextIO
import unittest
import sys

from addr2line import BacktraceResolver


def read_backtrace(stdin: TextIO):
    """
    Read stdin char-by-char and stop when when user pressed Ctrl+D or the
    Enter twice. Altough reading char-by-char is slow this won't be a
    problem here as backtraces shouldn't be huge.
    """
    linefeeds = 0
    line: list[str] = []

    while True:
        char = stdin.read(1)

        if char == '\n':
            linefeeds += 1

            if len(line) > 0:
                yield ''.join(line)
                line = []
        else:
            line.append(char)
            linefeeds = 0

        if char == '' or linefeeds > 1:
            break


TestResult = dict[str, Any]
MaybeTestResult = Optional[TestResult]
TestList = Sequence[tuple[str, MaybeTestResult]]


class TestStringMethods(unittest.TestCase):

    def setUp(self):
        self.parser = BacktraceResolver.BacktraceParser()

    def _test(self, cases: TestList):
        for line, expected in cases:
            res = self.parser(line.strip() + '\n')
            self.assertEqual(res, expected, f"failed to parse {line}")

    def test_separator(self):
        data = [('---', {'type': BacktraceResolver.BacktraceParser.Type.SEPARATOR})]
        self._test(data)

    def test_addresses_only(self):
        data = [
            (
                '0x12f34',
                {
                    'type': BacktraceResolver.BacktraceParser.Type.ADDRESS,
                    'prefix': None,
                    'addresses': [{'path': None, 'addr': '0x12f34'}],
                },
            ),
            (
                '0xa1234 0xb4567',
                {
                    'type': BacktraceResolver.BacktraceParser.Type.ADDRESS,
                    'prefix': None,
                    'addresses': [
                        {'path': None, 'addr': '0xa1234'},
                        {'path': None, 'addr': '0xb4567'},
                    ],
                },
            ),
            (
                '	0xa1234 /my/path+0xb4567',
                {
                    'type': BacktraceResolver.BacktraceParser.Type.ADDRESS,
                    'prefix': None,
                    'addresses': [
                        {'path': None, 'addr': '0xa1234'},
                        {'path': '/my/path', 'addr': '0xb4567'},
                    ],
                },
            ),
            (
                '/my/path+0x12f34',
                {
                    'type': BacktraceResolver.BacktraceParser.Type.ADDRESS,
                    'prefix': None,
                    'addresses': [{'path': '/my/path', 'addr': '0x12f34'}],
                },
            ),
            (
                ' /my/path+0x12f34',
                {
                    'type': BacktraceResolver.BacktraceParser.Type.ADDRESS,
                    'prefix': None,
                    'addresses': [{'path': '/my/path', 'addr': '0x12f34'}],
                },
            ),
        ]
        self._test(data)

    def test_without_prefix(self):
        data = [
            (
                'Some prefix 0x12f34',
                {
                    'type': BacktraceResolver.BacktraceParser.Type.ADDRESS,
                    'prefix': None,
                    'addresses': [{'path': None, 'addr': '0x12f34'}],
                },
            ),
            (
                'Some prefix /my/path+0x12f34',
                {
                    'type': BacktraceResolver.BacktraceParser.Type.ADDRESS,
                    'prefix': None,
                    'addresses': [{'path': '/my/path', 'addr': '0x12f34'}],
                },
            ),
        ]
        self._test(data)

    def test_reactor_stall_reports(self):
        data = [
            (
                'Reactor stalled on shard 1. Backtrace: 0x12f34',
                {
                    'type': BacktraceResolver.BacktraceParser.Type.ADDRESS,
                    'prefix': 'Reactor stalled on shard 1. Backtrace:',
                    'addresses': [{'path': None, 'addr': '0x12f34'}],
                },
            ),
            (
                'Reactor stalled on shard 1. Backtrace: 0xa1234 0xb5678',
                {
                    'type': BacktraceResolver.BacktraceParser.Type.ADDRESS,
                    'prefix': 'Reactor stalled on shard 1. Backtrace:',
                    'addresses': [
                        {'path': None, 'addr': '0xa1234'},
                        {'path': None, 'addr': '0xb5678'},
                    ],
                },
            ),
            (
                'Reactor stalled on shard 1. Backtrace: /my/path+0xabcd',
                {
                    'type': BacktraceResolver.BacktraceParser.Type.ADDRESS,
                    'prefix': 'Reactor stalled on shard 1. Backtrace:',
                    'addresses': [{'path': '/my/path', 'addr': '0xabcd'}],
                },
            ),
            (
                'Apr 28 11:42:58 ip-172-31-2-154.ec2.internal scylla[10612]: Reactor stalled for 260 ms on shard 20.',
                None,
            ),
            ('Apr 28 11:42:58 ip-172-31-2-154.ec2.internal scylla[10612]: Backtrace:', None),
            (
                'Apr 28 11:42:58 ip-172-31-2-154.ec2.internal scylla[10612]: 0x0000000003163dc2',
                {
                    'type': BacktraceResolver.BacktraceParser.Type.ADDRESS,
                    'prefix': 'Apr 28 11:42:58 ip-172-31-2-154.ec2.internal scylla[10612]:',
                    'addresses': [{'path': None, 'addr': '0x0000000003163dc2'}],
                },
            ),
        ]
        self._test(data)

    def test_boost_failure(self):
        data = [
            (
                'Expected partition_end(), but got end of stream, at 0xa1234 0xb5678 /my/path+0xabcd',
                {
                    'type': BacktraceResolver.BacktraceParser.Type.ADDRESS,
                    'prefix': 'Expected partition_end(), but got end of stream, at',
                    'addresses': [
                        {'path': None, 'addr': '0xa1234'},
                        {'path': None, 'addr': '0xb5678'},
                        {'path': '/my/path', 'addr': '0xabcd'},
                    ],
                },
            )
        ]
        self._test(data)

    def test_asan_failure(self):
        data = [
            (
                '==16118==ERROR: AddressSanitizer: heap-use-after-free on address 0x60700019c710 at pc 0x000014d24643 bp 0x7ffc51f72220 sp 0x7ffc51f72218',
                None,
            ),
            ('READ of size 8 at 0x60700019c710 thread T0', None),
            (
                '#0 0x14d24642  (/jenkins/workspace/scylla-enterprise/dtest-debug/scylla/.ccm/scylla-repository/1a5173bd45d01697d98ba2a7645f5d86afb2d0be/scylla/libexec/scylla+0x14d24642)',
                {
                    'type': BacktraceResolver.BacktraceParser.Type.ADDRESS,
                    'prefix': None,
                    'addresses': [
                        {
                            'path': '/jenkins/workspace/scylla-enterprise/dtest-debug/scylla/.ccm/scylla-repository/1a5173bd45d01697d98ba2a7645f5d86afb2d0be/scylla/libexec/scylla',
                            'addr': '0x14d24642',
                        }
                    ],
                },
            ),
            (
                '    #1 0xd8d910f  (/home/myhome/.dtest/dtest-84j9064d/test/node1/bin/scylla+0xd8d910f) (BuildId: 05a1d3d58d2b07e526decdad717e71a4590be2e0)',
                {
                    'type': BacktraceResolver.BacktraceParser.Type.ADDRESS,
                    'prefix': None,
                    'addresses': [
                        {
                            'path': '/home/myhome/.dtest/dtest-84j9064d/test/node1/bin/scylla',
                            'addr': '0xd8d910f',
                        }
                    ],
                },
            ),
        ]
        self._test(data)

    def test_thrown_exception(self):
        data = [
            (
                'seastar::internal::backtraced<std::runtime_error> (throw_with_backtrace_exception_logging Backtrace: 0x42bc95 /lib64/libc.so.6+0x281e1 0x412cfd)',
                {
                    'type': BacktraceResolver.BacktraceParser.Type.ADDRESS,
                    'prefix': 'seastar::internal::backtraced<std::runtime_error> (throw_with_backtrace_exception_logging Backtrace:',
                    'addresses': [
                        {'path': None, 'addr': '0x42bc95'},
                        {'path': '/lib64/libc.so.6', 'addr': '0x281e1'},
                        {'path': None, 'addr': '0x412cfd'},
                    ],
                },
            ),
            (
                'seastar::nested_exception: seastar::internal::backtraced<std::runtime_error> (inner Backtrace: 0x42bc95 /lib64/libc.so.6+0x281e1 0x412cfd) (while cleaning up after unknown_obj)',
                {
                    'type': BacktraceResolver.BacktraceParser.Type.ADDRESS,
                    'prefix': 'seastar::nested_exception: seastar::internal::backtraced<std::runtime_error> (inner Backtrace:',
                    'addresses': [
                        {'path': None, 'addr': '0x42bc95'},
                        {'path': '/lib64/libc.so.6', 'addr': '0x281e1'},
                        {'path': None, 'addr': '0x412cfd'},
                    ],
                },
            ),
            (
                'seastar::nested_exception: seastar::internal::backtraced<std::runtime_error> (inner Backtrace: 0x42bc95 /lib64/libc.so.6+0x281e1 0x412cfd) '
                '(while cleaning up after seastar::internal::backtraced<std::runtime_error> (outer Backtrace: 0x1234 /lib64/libc.so.6+0x5678 0xabcd))',
                {
                    'type': BacktraceResolver.BacktraceParser.Type.ADDRESS,
                    'prefix': 'seastar::nested_exception: seastar::internal::backtraced<std::runtime_error> (inner Backtrace: 0x42bc95 /lib64/libc.so.6+0x281e1 0x412cfd)'
                    ' (while cleaning up after seastar::internal::backtraced<std::runtime_error> (outer Backtrace:',
                    'addresses': [
                        {'path': None, 'addr': '0x1234'},
                        {'path': '/lib64/libc.so.6', 'addr': '0x5678'},
                        {'path': None, 'addr': '0xabcd'},
                    ],
                },
            ),
        ]
        self._test(data)

    def test_assertion_failure(self):
        data = [
            (
                "2022-04-15T06:19:24+00:00 gemini-1tb-10h-master-db-node-c0c7fc43-4 !    INFO | "
                "scylla: sstables/consumer.hh:610: future<> data_consumer::continuous_data_consumer"
                "<sstables::index_consume_entry_context<sstables::index_consumer>>::fast_forward_to"
                "(size_t, size_t) [StateProcessor = sstables::index_consume_entry_context"
                "<sstables::index_consumer>]: Assertion `begin >= _stream_position.position' failed.",
                None,
            ),
            (
                '2022-04-15T06:19:24+00:00 gemini-1tb-10h-master-db-node-c0c7fc43-4 !    INFO |   Backtrace:',
                None,
            ),
            (
                '2022-04-15T06:19:24+00:00 gemini-1tb-10h-master-db-node-c0c7fc43-4 !    INFO |   0x199d312',
                {
                    'type': BacktraceResolver.BacktraceParser.Type.ADDRESS,
                    'prefix': None,
                    'addresses': [{'path': None, 'addr': '0x199d312'}],
                },
            ),
            (
                '2022-04-15T06:19:24+00:00 gemini-1tb-10h-master-db-node-c0c7fc43-4 !    INFO |   (throw_with_backtrace_exception_logging Backtrace: 0x42bc95 /lib64/libc.so.6+0x281e1 0x412cfd)',
                {
                    'type': BacktraceResolver.BacktraceParser.Type.ADDRESS,
                    'prefix': '2022-04-15T06:19:24+00:00 gemini-1tb-10h-master-db-node-c0c7fc43-4 !    INFO |   (throw_with_backtrace_exception_logging Backtrace:',
                    'addresses': [
                        {'path': None, 'addr': '0x42bc95'},
                        {'path': '/lib64/libc.so.6', 'addr': '0x281e1'},
                        {'path': None, 'addr': '0x412cfd'},
                    ],
                },
            ),
        ]
        self._test(data)

    def test_segfault_failure(self):
        data = [
            ('[2022-04-19T23:09:28.311Z] Segmentation fault on shard 1.', None),
            ('[2022-04-19T23:09:28.311Z] Backtrace:', None),
            (
                '[2022-04-19T23:09:28.311Z]   0x461bbb8',
                {
                    'type': BacktraceResolver.BacktraceParser.Type.ADDRESS,
                    'prefix': None,
                    'addresses': [{'path': None, 'addr': '0x461bbb8'}],
                },
            ),
            (
                '[2022-04-19T23:09:28.311Z]   /lib64/libpthread.so.0+0x92a4',
                {
                    'type': BacktraceResolver.BacktraceParser.Type.ADDRESS,
                    'prefix': None,
                    'addresses': [{'path': '/lib64/libpthread.so.0', 'addr': '0x92a4'}],
                },
            ),
            (
                '#0 0x19c01681 (/path/to/scylla+0xdeadbeef)',
                {
                    'type': BacktraceResolver.BacktraceParser.Type.ADDRESS,
                    'prefix': None,
                    'addresses': [{'path': '/path/to/scylla', 'addr': '0x19c01681'}],
                },
            ),
            (
                '#1 0x00007fd2dab4f950 abort (libc.so.6 + 0x26950)',
                {
                    'type': BacktraceResolver.BacktraceParser.Type.ADDRESS,
                    'prefix': None,
                    'addresses': [{'path': 'libc.so.6', 'addr': '0x00007fd2dab4f950'}],
                },
            ),
            (
                '#2 0x00000000015c4cd3  n/a (/path/to/scylla + 0x15c4cd3)',
                {
                    'type': BacktraceResolver.BacktraceParser.Type.ADDRESS,
                    'prefix': None,
                    'addresses': [{'path': '/path/to/scylla', 'addr': '0x00000000015c4cd3'}],
                },
            ),
            ('kernel callstack: ', None),
            (
                'kernel callstack: 0xffffffffffffff80',
                {
                    'type': BacktraceResolver.BacktraceParser.Type.ADDRESS,
                    'prefix': 'kernel callstack: ',
                    'addresses': [{'path': '<kernel>', 'addr': '0xffffffffffffff80'}],
                },
            ),
            (
                'kernel callstack: 0xffffffffffffff80 0xffffffffaf15ccca',
                {
                    'type': BacktraceResolver.BacktraceParser.Type.ADDRESS,
                    'prefix': 'kernel callstack: ',
                    'addresses': [
                        {'path': '<kernel>', 'addr': '0xffffffffffffff80'},
                        {'path': '<kernel>', 'addr': '0xffffffffaf15ccca'},
                    ],
                },
            ),
        ]
        self._test(data)


def main():
    description = 'Massage and pass addresses to the real addr2line for symbol lookup.'
    epilog = '''
There are three operational modes:
  1) If -f is specified input will be read from FILE
  2) If -f is omitted and there are ADDRESS args they will be read as input
  3) If -f is omitted and there are no ADDRESS args input will be read from stdin
'''

    cmdline_parser = argparse.ArgumentParser(
        description=description,
        epilog=epilog,
        formatter_class=argparse.RawDescriptionHelpFormatter,
    )

    cmdline_parser.add_argument(
        '-e',
        '--executable',
        type=str,
        required=True,
        metavar='EXECUTABLE',
        dest='executable',
        help='The executable where the addresses originate from',
    )

    cmdline_parser.add_argument(
        '-f',
        '--file',
        type=str,
        required=False,
        metavar='FILE',
        dest='file',
        help='The file containing the addresses',
    )

    cmdline_parser.add_argument(
        '-b',
        '--before',
        type=int,
        metavar='BEFORE',
        default=1,
        help='Non-backtrace lines to print before resolved backtraces for context.'
        ' Set to 0 to print only resolved backtraces.'
        ' Set to -1 to print all non-backtrace lines. Default is 1.',
    )

    cmdline_parser.add_argument(
        '-m',
        '--match',
        type=str,
        metavar='MATCH',
        help='Only resolve backtraces whose non-backtrace lines match the regular-expression.'
        ' The amount of non-backtrace lines considered can be set with --before.'
        ' By default no matching is performed.',
    )

    cmdline_parser.add_argument(
        '-a',
        '--addr2line',
        type=str,
        metavar='CMD_PATH',
        default='addr2line',
        help='The path or name of the addr2line command, which should behave as and '
        'accept the same options as binutils addr2line or llvm-addr2line.',
    )

    cmdline_parser.add_argument(
        '-v',
        '--verbose',
        action='store_true',
        default=False,
        help='Make resolved backtraces verbose, prepend to each line the module'
        ' it originates from, as well as the address being resolved',
    )

    cmdline_parser.add_argument(
        '-d', '--debug', action='store_true', help='Emit debug logging to stderr.'
    )

    cmdline_parser.add_argument(
        '-t', '--test', action='store_true', default=False, help='Self-test'
    )

    cmdline_parser.add_argument(
        'addresses', type=str, metavar='ADDRESS', nargs='*', help='Addresses to parse'
    )

    cmdline_parser.add_argument(
        '--kallsyms',
        type=str,
        metavar='KALLSYMS',
        default='/proc/kallsyms',
        help='Alternative path for kallsyms file to resolve kernel addresses',
    )

    args = cmdline_parser.parse_args()

    if args.addresses and args.file:
        print("Cannot use both -f and ADDRESS")
        cmdline_parser.print_help()

    if args.file:
        lines = open(args.file, 'r')
    elif args.addresses:
        lines = args.addresses
    else:
        if sys.stdin.isatty():
            lines = read_backtrace(sys.stdin)
        else:
            lines = sys.stdin

    with BacktraceResolver(
        executable=args.executable,
        kallsyms=args.kallsyms,
        before_lines=args.before,
        context_re=args.match,
        verbose=args.verbose,
        cmd_path=args.addr2line,
        debug=args.debug,
    ) as resolve:
        for line in list(lines):
            resolve(line.strip() + '\n')


if __name__ == '__main__':
    cmdline_parser = argparse.ArgumentParser(add_help=False)
    cmdline_parser.add_argument(
        '-t', '--test', action='store_true', default=False, help='Self-test'
    )
    args, unrecognized = cmdline_parser.parse_known_args()

    if args.test:
        unittest.main(argv=[sys.argv[0]] + unrecognized)
    else:
        main()