_extract_frame_info() - Code Metrics - Crystalnix/omaha-server - Measure and Improve Code Quality continuously with Scrutinizer

_extract_frame_info() F
last analyzed 2018-05-22 07:05 UTC

↳ Parent: Project

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	11
dl	0
loc	42
rs	3.1764
c	0
b	0
f	0

How to fix Complexity

# This Source Code Form is subject to the terms of the Mozilla Public
# License, v. 2.0. If a copy of the MPL was not distributed with this
# file, You can obtain one at http://mozilla.org/MPL/2.0/.

"""This module provides a function that will transate Minidump Stackwalk pipe
dump into a json format.

{
  # "status": string, // OK | ERROR_* | SYMBOL_SUPPLIER_INTERRUPTED
  "system_info": {
    "os": string,
    "os_ver": string,
    "cpu_arch": string, // x86 | amd64 | arm | ppc | sparc
    "cpu_info": string,
    "cpu_count": int
  },
  "crash_info": {
    "type": string,
    "crash_address": string, // 0x[[:xdigit:]]+
    "crashing_thread": int // | null
  }
  "main_module": int, // index into modules
  "modules": [
         // zero or more
    {
      "base_addr": string, // 0x[[:xdigit:]]+
      "debug_file": string,
      "debug_id": string, // [[:xdigit:]]{33}
      "end_addr": string, // 0x[[:xdigit:]]+
      "filename": string,
      "version": string
    }
  ],
  "thread_count": int,
  "threads": [
    // for i in range(thread_count)
    {
      "frame_count": int,
      "frames": [
        // for i in range(frame_count)
        {
          "frame": int,
          "module": string,   // optional
          "function": string, // optional
          "file": string,     // optional
          "line": int,        // optional
          "offset": string,         // 0x[[:xdigit:]]+
          "module_offset": string,  // 0x[[:xdigit:]]+ , optional
          "function_offset": string // 0x[[:xdigit:]]+ , optional
        }
      ]
    }
  ],
  // repeated here for ease of searching
  // (max 10 frames)
  "crashing_thread": {
    "threads_index": int,
    "total_frames": int,
    "frames": [
      // for i in range(length)
      {
        // as per "frames" entries from "threads" above
      }
    ]
  }
}
"""


class DotDict(dict):
    __getattr__ = dict.__getitem__
    __setattr__ = dict.__setitem__
    __delattr__ = dict.__delitem__


class DotDictWithPut(DotDict):
    #--------------------------------------------------------------------------
    def put_if_not_none(self, key, value):
        if value is not None and value != '':
            self[key] = value


def pipe_dump_to_json_dump(pipe_dump_iterable):
    """given a list (or any iterable) of strings representing a MDSW pipe dump,
    this function will convert it into a json format."""
    json_dump = DotDict()
    crashing_thread = None
    module_counter = 0
    for a_line in pipe_dump_iterable:
        parts = a_line.split('|')
        if parts[0] == 'OS':
            _extract_OS_info(parts, json_dump)
        elif parts[0] == 'CPU':
            _extract_CPU_info(parts, json_dump)
        elif parts[0] == 'Crash':
            crashing_thread = _extract_crash_info(parts, json_dump)
        elif parts[0] == 'Module':
            _extract_module_info(parts, json_dump, module_counter)
            module_counter += 1
        else:
            try:
                thread_number = int(parts[0])
            except (ValueError, IndexError):
                continue  # unknow line type, ignore it
            _extract_frame_info(parts, json_dump)
    try:
        json_dump.thread_count = len(json_dump.threads)
    except KeyError:  # no threads were over found, 'threads' key was not made
        json_dump.thread_count = 0
    if crashing_thread is not None:
        crashing_thread_frames = DotDict()
        crashing_thread_frames.threads_index = crashing_thread
        crashing_thread_frames.total_frames = \
            len(json_dump.threads[crashing_thread].frames)
        crashing_thread_frames.frames = \
            json_dump.threads[crashing_thread].frames[:10]
        json_dump.crashing_thread = crashing_thread_frames
    return json_dump


#------------------------------------------------------------------------------
def _get(indexable_container, index, default):
    """like 'get' on a dict, but it works on lists, too"""
    try:
        return indexable_container[index]
    except (IndexError, KeyError):
        return default


#------------------------------------------------------------------------------
def _get_int(indexable_container, index, default):
    """try to get an int from an indexable container.  If that fails
    return the default"""
    try:
        return int(indexable_container[index])
    # exceptions separated to make case coverage clearer
    except (IndexError, KeyError):
        # item not found in the container
        return default
    except ValueError:
        # conversion to integer has failed
        return default


#------------------------------------------------------------------------------
def _extract_OS_info(os_line, json_dump):
    """given a pipe dump OS line, extract the parts and put them in their
    proper location within the json_dump"""
    system_info = DotDictWithPut()
    system_info.put_if_not_none('os', _get(os_line, 1, None))
    system_info.put_if_not_none('os_ver', _get(os_line, 2, None))
    if 'system_info' in json_dump:
        json_dump.system_info.update(system_info)
    else:
        json_dump.system_info = system_info


#------------------------------------------------------------------------------
def _extract_CPU_info(cpu_line, json_dump):
    """given a pipe dump CPU line, extract the parts and put them in their
    proper location within the json_dump"""
    system_info = DotDictWithPut()
    system_info.put_if_not_none('cpu_arch', _get(cpu_line, 1, None))
    system_info.put_if_not_none('cpu_info', _get(cpu_line, 2, None))
    system_info.put_if_not_none('cpu_count', _get_int(cpu_line, 3, None))
    if 'system_info' in json_dump:
        json_dump.system_info.update(system_info)
    else:
        json_dump.system_info = system_info


#------------------------------------------------------------------------------
def _extract_crash_info(crash_line, json_dump):
    """given a pipe dump CRASH line, extract the parts and put them in their
    proper location within the json_dump"""
    crash_info = DotDictWithPut()
    crash_info.put_if_not_none('type', _get(crash_line, 1, None))
    crash_info.put_if_not_none('crash_address', _get(crash_line, 2, None))
    crash_info.put_if_not_none('crashing_thread', _get_int(crash_line, 3, None))
    json_dump.crash_info = crash_info
    return crash_info.get('crashing_thread', None)


#------------------------------------------------------------------------------
def _extract_module_info(module_line, json_dump, module_counter):
    """given a pipe dump Module line, extract the parts and put them in their
    proper location within the json_dump"""
    module = DotDictWithPut()
    module.put_if_not_none('filename', _get(module_line, 1, None))
    module.put_if_not_none('version', _get(module_line, 2, None))
    module.put_if_not_none('debug_file', _get(module_line, 3, None))
    module.put_if_not_none('debug_id', _get(module_line, 4, None))
    module.put_if_not_none('base_addr', _get(module_line, 5, None))
    module.put_if_not_none('end_addr', _get(module_line, 6, None))
    is_main_module = _get_int(module_line, 7, 0)
    if is_main_module:
        json_dump.main_module = module_counter
    if 'modules' not in json_dump:
        json_dump.modules = []
    json_dump.modules.append(module)


#------------------------------------------------------------------------------
def _extract_frame_info(frame_line, json_dump):
    """given a pipe dump Frame line, extract the parts and put them in their
    proper location within the json_dump"""
    if 'threads' not in json_dump:
        json_dump.threads = []
    thread_number = _get_int(frame_line, 0, None)
    if thread_number is None:
        return
    if thread_number >= len(json_dump.threads):
        # threads are supposed to arrive in order.  We've not seen this thread
        # before, fill in a new entry in the 'threads' section of the json_dump
        # making sure that intervening missing threads have empty thread data
        for i in range(thread_number - len(json_dump.threads) + 1):
            thread = DotDict()
            thread.frame_count = 0
            thread.frames = []
            json_dump.threads.append(thread)
    # collect frame info from the pipe dump line
    tmp_frame = _get_int(frame_line, 1, None)
    tmp_module = _get(frame_line, 2, None)
    tmp_function = _get(frame_line, 3, None)
    tmp_file = _get(frame_line, 4, None)
    tmp_line = _get_int(frame_line, 5, None)
    tmp_offset = _get(frame_line, 6, None)
    frame = DotDictWithPut()
    frame.put_if_not_none('frame', tmp_frame)
    frame.put_if_not_none('filename', tmp_module)
    frame.put_if_not_none('function', tmp_function)
    frame.put_if_not_none('abs_path', tmp_file)
    frame.put_if_not_none('lineno', tmp_line)
    if tmp_file and tmp_line is not None:
        # skip offset entirely
        pass
    elif not tmp_file and tmp_function:
        frame.function_offset = tmp_offset
    elif not tmp_function and tmp_module:
        frame.module_offset = tmp_offset
    else:
        frame.offset = tmp_offset
    # save the frame info into the json
    json_dump.threads[thread_number].frames.append(frame)
    json_dump.threads[thread_number].frame_count += 1


1			# This Source Code Form is subject to the terms of the Mozilla Public
2			# License, v. 2.0. If a copy of the MPL was not distributed with this
3			# file, You can obtain one at http://mozilla.org/MPL/2.0/.
4
5			"""This module provides a function that will transate Minidump Stackwalk pipe
6			dump into a json format.
7
8			{
9			# "status": string, // OK \| ERROR_* \| SYMBOL_SUPPLIER_INTERRUPTED
10			"system_info": {
11			"os": string,
12			"os_ver": string,
13			"cpu_arch": string, // x86 \| amd64 \| arm \| ppc \| sparc
14			"cpu_info": string,
15			"cpu_count": int
16			},
17			"crash_info": {
18			"type": string,
19			"crash_address": string, // 0x[[:xdigit:]]+
20			"crashing_thread": int // \| null
21			}
22			"main_module": int, // index into modules
23			"modules": [
24			// zero or more
25			{
26			"base_addr": string, // 0x[[:xdigit:]]+
27			"debug_file": string,
28			"debug_id": string, // [[:xdigit:]]{33}
29			"end_addr": string, // 0x[[:xdigit:]]+
30			"filename": string,
31			"version": string
32			}
33			],
34			"thread_count": int,
35			"threads": [
36			// for i in range(thread_count)
37			{
38			"frame_count": int,
39			"frames": [
40			// for i in range(frame_count)
41			{
42			"frame": int,
43			"module": string, // optional
44			"function": string, // optional
45			"file": string, // optional
46			"line": int, // optional
47			"offset": string, // 0x[[:xdigit:]]+
48			"module_offset": string, // 0x[[:xdigit:]]+ , optional
49			"function_offset": string // 0x[[:xdigit:]]+ , optional
50			}
51			]
52			}
53			],
54			// repeated here for ease of searching
55			// (max 10 frames)
56			"crashing_thread": {
57			"threads_index": int,
58			"total_frames": int,
59			"frames": [
60			// for i in range(length)
61			{
62			// as per "frames" entries from "threads" above
63			}
64			]
65			}
66			}
67			"""
68
69
70			class DotDict(dict):
71			__getattr__ = dict.__getitem__
72			__setattr__ = dict.__setitem__
73			__delattr__ = dict.__delitem__
74
75
76			class DotDictWithPut(DotDict):
77			#--------------------------------------------------------------------------
78			def put_if_not_none(self, key, value):
79			if value is not None and value != '':
80			self[key] = value
81
82
83			def pipe_dump_to_json_dump(pipe_dump_iterable):
84			"""given a list (or any iterable) of strings representing a MDSW pipe dump,
85			this function will convert it into a json format."""
86			json_dump = DotDict()
87			crashing_thread = None
88			module_counter = 0
89			for a_line in pipe_dump_iterable:
90			parts = a_line.split('\|')
91			if parts[0] == 'OS':
92			_extract_OS_info(parts, json_dump)
93			elif parts[0] == 'CPU':
94			_extract_CPU_info(parts, json_dump)
95			elif parts[0] == 'Crash':
96			crashing_thread = _extract_crash_info(parts, json_dump)
97			elif parts[0] == 'Module':
98			_extract_module_info(parts, json_dump, module_counter)
99			module_counter += 1
100			else:
101			try:
102			thread_number = int(parts[0])
103			except (ValueError, IndexError):
104			continue # unknow line type, ignore it
105			_extract_frame_info(parts, json_dump)
106			try:
107			json_dump.thread_count = len(json_dump.threads)
108			except KeyError: # no threads were over found, 'threads' key was not made
109			json_dump.thread_count = 0
110			if crashing_thread is not None:
111			crashing_thread_frames = DotDict()
112			crashing_thread_frames.threads_index = crashing_thread
113			crashing_thread_frames.total_frames = \
114			len(json_dump.threads[crashing_thread].frames)
115			crashing_thread_frames.frames = \
116			json_dump.threads[crashing_thread].frames[:10]
117			json_dump.crashing_thread = crashing_thread_frames
118			return json_dump
119
120
121			#------------------------------------------------------------------------------
122			def _get(indexable_container, index, default):
123			"""like 'get' on a dict, but it works on lists, too"""
124			try:
125			return indexable_container[index]
126			except (IndexError, KeyError):
127			return default
128
129
130			#------------------------------------------------------------------------------
131			def _get_int(indexable_container, index, default):
132			"""try to get an int from an indexable container. If that fails
133			return the default"""
134			try:
135			return int(indexable_container[index])
136			# exceptions separated to make case coverage clearer
137			except (IndexError, KeyError):
138			# item not found in the container
139			return default
140			except ValueError:
141			# conversion to integer has failed
142			return default
143
144
145			#------------------------------------------------------------------------------
146			def _extract_OS_info(os_line, json_dump):
147			"""given a pipe dump OS line, extract the parts and put them in their
148			proper location within the json_dump"""
149			system_info = DotDictWithPut()
150			system_info.put_if_not_none('os', _get(os_line, 1, None))
151			system_info.put_if_not_none('os_ver', _get(os_line, 2, None))
152			if 'system_info' in json_dump:
153			json_dump.system_info.update(system_info)
154			else:
155			json_dump.system_info = system_info
156
157
158			#------------------------------------------------------------------------------
159			def _extract_CPU_info(cpu_line, json_dump):
160			"""given a pipe dump CPU line, extract the parts and put them in their
161			proper location within the json_dump"""
162			system_info = DotDictWithPut()
163			system_info.put_if_not_none('cpu_arch', _get(cpu_line, 1, None))
164			system_info.put_if_not_none('cpu_info', _get(cpu_line, 2, None))
165			system_info.put_if_not_none('cpu_count', _get_int(cpu_line, 3, None))
166			if 'system_info' in json_dump:
167			json_dump.system_info.update(system_info)
168			else:
169			json_dump.system_info = system_info
170
171
172			#------------------------------------------------------------------------------
173			def _extract_crash_info(crash_line, json_dump):
174			"""given a pipe dump CRASH line, extract the parts and put them in their
175			proper location within the json_dump"""
176			crash_info = DotDictWithPut()
177			crash_info.put_if_not_none('type', _get(crash_line, 1, None))
178			crash_info.put_if_not_none('crash_address', _get(crash_line, 2, None))
179			crash_info.put_if_not_none('crashing_thread', _get_int(crash_line, 3, None))
180			json_dump.crash_info = crash_info
181			return crash_info.get('crashing_thread', None)
182
183
184			#------------------------------------------------------------------------------
185			def _extract_module_info(module_line, json_dump, module_counter):
186			"""given a pipe dump Module line, extract the parts and put them in their
187			proper location within the json_dump"""
188			module = DotDictWithPut()
189			module.put_if_not_none('filename', _get(module_line, 1, None))
190			module.put_if_not_none('version', _get(module_line, 2, None))
191			module.put_if_not_none('debug_file', _get(module_line, 3, None))
192			module.put_if_not_none('debug_id', _get(module_line, 4, None))
193			module.put_if_not_none('base_addr', _get(module_line, 5, None))
194			module.put_if_not_none('end_addr', _get(module_line, 6, None))
195			is_main_module = _get_int(module_line, 7, 0)
196			if is_main_module:
197			json_dump.main_module = module_counter
198			if 'modules' not in json_dump:
199			json_dump.modules = []
200			json_dump.modules.append(module)
201
202
203			#------------------------------------------------------------------------------
204			def _extract_frame_info(frame_line, json_dump):
205			"""given a pipe dump Frame line, extract the parts and put them in their
206			proper location within the json_dump"""
207			if 'threads' not in json_dump:
208			json_dump.threads = []
209			thread_number = _get_int(frame_line, 0, None)
210			if thread_number is None:
211			return
212			if thread_number >= len(json_dump.threads):
213			# threads are supposed to arrive in order. We've not seen this thread
214			# before, fill in a new entry in the 'threads' section of the json_dump
215			# making sure that intervening missing threads have empty thread data
216			for i in range(thread_number - len(json_dump.threads) + 1):
217			thread = DotDict()
218			thread.frame_count = 0
219			thread.frames = []
220			json_dump.threads.append(thread)
221			# collect frame info from the pipe dump line
222			tmp_frame = _get_int(frame_line, 1, None)
223			tmp_module = _get(frame_line, 2, None)
224			tmp_function = _get(frame_line, 3, None)
225			tmp_file = _get(frame_line, 4, None)
226			tmp_line = _get_int(frame_line, 5, None)
227			tmp_offset = _get(frame_line, 6, None)
228			frame = DotDictWithPut()
229			frame.put_if_not_none('frame', tmp_frame)
230			frame.put_if_not_none('filename', tmp_module)
231			frame.put_if_not_none('function', tmp_function)
232			frame.put_if_not_none('abs_path', tmp_file)
233			frame.put_if_not_none('lineno', tmp_line)
234			if tmp_file and tmp_line is not None:
235			# skip offset entirely
236			pass
237			elif not tmp_file and tmp_function:
238			frame.function_offset = tmp_offset
239			elif not tmp_function and tmp_module:
240			frame.module_offset = tmp_offset
241			else:
242			frame.offset = tmp_offset
243			# save the frame info into the json
244			json_dump.threads[thread_number].frames.append(frame)
245			json_dump.threads[thread_number].frame_count += 1
246

Crystalnix / omaha-server

_extract_frame_info() F last analyzed 2018-05-22 07:05 UTC

Complexity

Size

Duplication

Importance

How to fix Complexity

Complexity

Duplication Side-by-Side

Filter issues like

_extract_frame_info() F
last analyzed 2018-05-22 07:05 UTC