diff --git a/.flake8 b/.flake8 index 1313a0c..c835618 100644 --- a/.flake8 +++ b/.flake8 @@ -4,8 +4,39 @@ exclude = __pycache__, docs, .idea, - venv + venv, + out, + scripts, max-line-length = 120 per-file-ignores = # ignore unused imports in __init__ files - */__init__.py: F401 \ No newline at end of file + */__init__.py: F401 + # supress some docstring requirements in tests + test/__init__.py: D104 + test/test_deep/__init__.py: D104 + test/test_deep/*/__init__.py: D104,D107 + test/test_deep/*/test_*.py: D101,D102,D107,D100,D105 + test/test_deep/test_*.py: D101,D102,D107,D100,D105 + test/test_deep/test_target.py: D102,D107,D103 + # these files are from OTEL so should use OTEL license. + */deep/api/types.py: NCF102 + */deep/api/resource/__init__.py: NCF102 + */deep/api/attributes/__init__.py: NCF102 + +detailed-output = True +copyright-regex = + '# Copyright \(C\) [0-9]{4} Intergral GmbH' + '#' + '# This program is free software: you can redistribute it and/or modify' + '# it under the terms of the GNU Affero General Public License as published by' + '# the Free Software Foundation, either version 3 of the License, or' + '# \(at your option\) any later version.' + '#' + '# This program is distributed in the hope that it will be useful,' + '# but WITHOUT ANY WARRANTY; without even the implied warranty of' + '# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the' + '# GNU Affero General Public License for more details.' + '#' + '# You should have received a copy of the GNU Affero General Public License' + '# along with this program. If not, see .' + '' diff --git a/.idea/misc.xml b/.idea/misc.xml index c7dbe3f..eaee005 100644 --- a/.idea/misc.xml +++ b/.idea/misc.xml @@ -1,5 +1,12 @@ + + + + + + diff --git a/CHANGELOG.md b/CHANGELOG.md new file mode 100644 index 0000000..8eb8867 --- /dev/null +++ b/CHANGELOG.md @@ -0,0 +1,18 @@ +# main (unreleased) + +- **[CHANGE]**: change(build): add doc string check to flake8 [#14](https://github.com/intergral/deep/pull/14) [@Umaaz](https://github.com/Umaaz) +- **[FEATURE]**: feat(logging): initial implementation of log points [#3](https://github.com/intergral/deep/pull/3) [@Umaaz](https://github.com/Umaaz) +- **[BUGFIX]**: feat(api): add api function to register tracepoint directly [#8](https://github.com/intergral/deep/pull/8) [@Umaaz](https://github.com/Umaaz) + +# 1.0.1 (22/06/2023) + +- **[BUGFIX]**: fix(config): correct env name on look up [#1](https://github.com/intergral/deep/pull/1) [@Umaaz](https://github.com/Umaaz) + + \ No newline at end of file diff --git a/dev-requirements.txt b/dev-requirements.txt index 6b0128b..326ed26 100644 --- a/dev-requirements.txt +++ b/dev-requirements.txt @@ -6,4 +6,7 @@ pdoc3 mkdocstrings-python build twine -certifi>=2023.7.22 # not directly required, pinned by Snyk to avoid a vulnerability \ No newline at end of file +flake8-docstrings +certifi>=2023.7.22 # not directly required, pinned by Snyk to avoid a vulnerability +flake8-header-validator>=0.0.3 +setuptools>=65.5.1 # not directly required, pinned by Snyk to avoid a vulnerability diff --git a/dev/test-server/src/test-server/server.py b/dev/test-server/src/test-server/server.py index 18bfff9..0c7ee59 100644 --- a/dev/test-server/src/test-server/server.py +++ b/dev/test-server/src/test-server/server.py @@ -9,6 +9,11 @@ # but WITHOUT ANY WARRANTY; without even the implied warranty of # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the # GNU Affero General Public License for more details. +# +# You should have received a copy of the GNU Affero General Public License +# along with this program. If not, see . + +"""This is a basic example of setting up a GRPC server to consume Deep protobuf messages.""" from concurrent import futures @@ -23,6 +28,7 @@ def serve(): + """Set up and start a GRPC service on port 43315 to server Deep clients.""" server = grpc.server(futures.ThreadPoolExecutor(max_workers=10)) deepproto.proto.poll.v1.poll_pb2_grpc.add_PollConfigServicer_to_server( @@ -34,14 +40,19 @@ def serve(): class SnapshotServicer(SnapshotServiceServicer): + """Create class to handle snapshot send events.""" def send(self, request, context): + """Receive and process a snapshot request.""" print("hit", request.ID, request.attributes) return SnapshotResponse() class PollServicer(PollConfigServicer): + """Create a class to handle poll requests.""" + def poll(self, request, context): + """Receive and process poll requests.""" print(request, context, context.invocation_metadata()) response = PollResponse(ts_nanos=request.ts_nanos, current_hash="123", response=[ TracePointConfig(ID="17", path="/simple-app/simple_test.py", line_number=31, diff --git a/examples/simple-app-custom-logging/src/simple-app/base_test.py b/examples/simple-app-custom-logging/src/simple-app/base_test.py new file mode 100644 index 0000000..20d06e9 --- /dev/null +++ b/examples/simple-app-custom-logging/src/simple-app/base_test.py @@ -0,0 +1,43 @@ +# Copyright (C) 2023 Intergral GmbH +# +# This program is free software: you can redistribute it and/or modify +# it under the terms of the GNU Affero General Public License as published by +# the Free Software Foundation, either version 3 of the License, or +# (at your option) any later version. +# +# This program is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +# GNU Affero General Public License for more details. +# +# You should have received a copy of the GNU Affero General Public License +# along with this program. If not, see . + +"""A simple test object for examples.""" + +import random +import uuid + + +class BaseTest: + """A basic test that is used in examples.""" + + def new_id(self): + """Create new id.""" + return str(uuid.uuid4()) + + def next_max(self): + """Create new random max.""" + return random.randint(1, 101) + + def make_char_count_map(self, in_str): + """Create char count map.""" + res = {} + + for i in range(0, len(in_str)): + c = in_str[i] + if c not in res: + res[c] = 0 + else: + res[c] = res[c] + 1 + return res diff --git a/examples/simple-app-custom-logging/src/simple-app/main.py b/examples/simple-app-custom-logging/src/simple-app/main.py index 5b013f1..5bfa014 100644 --- a/examples/simple-app-custom-logging/src/simple-app/main.py +++ b/examples/simple-app-custom-logging/src/simple-app/main.py @@ -9,14 +9,53 @@ # but WITHOUT ANY WARRANTY; without even the implied warranty of # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the # GNU Affero General Public License for more details. +# +# You should have received a copy of the GNU Affero General Public License +# along with this program. If not, see . +"""Simple Example of using Deep, with custom log config.""" import os +import signal +import time import deep +from simple_test import SimpleTest + + +class GracefulKiller: + """Ensure clean shutdown.""" + + kill_now = False + + def __init__(self): + """Crate new killer.""" + signal.signal(signal.SIGINT, self.exit_gracefully) + signal.signal(signal.SIGTERM, self.exit_gracefully) + + def exit_gracefully(self, *args): + """Exit example.""" + self.kill_now = True + + +def main(): + """Run the example.""" + killer = GracefulKiller() + ts = SimpleTest("This is a test") + while not killer.kill_now: + try: + ts.message(ts.new_id()) + except BaseException as e: + print(e) + ts.reset() + + time.sleep(0.1) + if __name__ == '__main__': deep.start({ 'SERVICE_URL': 'localhost:43315', 'LOGGING_CONF': "%s/logging.conf" % os.path.dirname(os.path.realpath(__file__)) }) + print("app running") + main() diff --git a/examples/simple-app-custom-logging/src/simple-app/simple_test.py b/examples/simple-app-custom-logging/src/simple-app/simple_test.py new file mode 100644 index 0000000..5811e00 --- /dev/null +++ b/examples/simple-app-custom-logging/src/simple-app/simple_test.py @@ -0,0 +1,73 @@ +# Copyright (C) 2023 Intergral GmbH +# +# This program is free software: you can redistribute it and/or modify +# it under the terms of the GNU Affero General Public License as published by +# the Free Software Foundation, either version 3 of the License, or +# (at your option) any later version. +# +# This program is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +# GNU Affero General Public License for more details. +# +# You should have received a copy of the GNU Affero General Public License +# along with this program. If not, see . + +"""A simple test object for examples.""" + +import time + +from base_test import BaseTest + + +class SimpleTest(BaseTest): + """A basic test that is used in examples.""" + + def __init__(self, test_name): + """Create new test object.""" + super().__init__() + self._started_at = round(time.time() * 1000) + self.__cnt = 0 + self.char_counter = {} + self.test_name = test_name + self.max_executions = self.next_max() + + def message(self, uuid): + """Print message to console.""" + print("%s:%s" % (self.__cnt, uuid)) + self.__cnt += 1 + self.check_end(self.__cnt, self.max_executions) + + info = self.make_char_count_map(uuid) + self.merge(self.char_counter, info) + if self.__cnt % 30 == 0: + self.dump() + + def merge(self, char_counter, new_info): + """Merge captured data.""" + for key in new_info: + new_val = new_info[key] + + if key not in char_counter: + char_counter[key] = new_val + else: + char_counter[key] = new_val + char_counter[key] + + def dump(self): + """Dump message to console.""" + print(self.char_counter) + self.char_counter = {} + + def check_end(self, value, max_executions): + """Check if we are at end.""" + if value > max_executions: + raise Exception("Hit max executions %s %s " % (value, max_executions)) + + def __str__(self) -> str: + """Represent this as a string.""" + return self.__class__.__name__ + ":" + self.test_name + ":" + str(self._started_at) + + def reset(self): + """Reset the count.""" + self.__cnt = 0 + self.max_executions = self.next_max() diff --git a/examples/simple-app-docker/src/simple-app/base_test.py b/examples/simple-app-docker/src/simple-app/base_test.py index 0f05dba..20d06e9 100644 --- a/examples/simple-app-docker/src/simple-app/base_test.py +++ b/examples/simple-app-docker/src/simple-app/base_test.py @@ -9,20 +9,29 @@ # but WITHOUT ANY WARRANTY; without even the implied warranty of # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the # GNU Affero General Public License for more details. +# +# You should have received a copy of the GNU Affero General Public License +# along with this program. If not, see . + +"""A simple test object for examples.""" import random import uuid class BaseTest: + """A basic test that is used in examples.""" def new_id(self): + """Create new id.""" return str(uuid.uuid4()) def next_max(self): + """Create new random max.""" return random.randint(1, 101) def make_char_count_map(self, in_str): + """Create char count map.""" res = {} for i in range(0, len(in_str)): diff --git a/examples/simple-app-docker/src/simple-app/main.py b/examples/simple-app-docker/src/simple-app/main.py index b30ba64..1b0efd3 100644 --- a/examples/simple-app-docker/src/simple-app/main.py +++ b/examples/simple-app-docker/src/simple-app/main.py @@ -1,4 +1,4 @@ -# Copyright (C) 2023 Intergral GmbH +# Copyright (C) 2024 Intergral GmbH # # This program is free software: you can redistribute it and/or modify # it under the terms of the GNU Affero General Public License as published by @@ -9,6 +9,12 @@ # but WITHOUT ANY WARRANTY; without even the implied warranty of # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the # GNU Affero General Public License for more details. +# +# You should have received a copy of the GNU Affero General Public License +# along with this program. If not, see . + +"""Simple Example of using Deep.""" + import signal import time @@ -17,17 +23,22 @@ class GracefulKiller: + """Ensure clean shutdown.""" + kill_now = False def __init__(self): + """Crate new killer.""" signal.signal(signal.SIGINT, self.exit_gracefully) signal.signal(signal.SIGTERM, self.exit_gracefully) def exit_gracefully(self, *args): + """Exit example.""" self.kill_now = True def main(): + """Run the example.""" killer = GracefulKiller() ts = SimpleTest("This is a test") while not killer.kill_now: diff --git a/examples/simple-app-docker/src/simple-app/simple_test.py b/examples/simple-app-docker/src/simple-app/simple_test.py index c724cb7..5811e00 100644 --- a/examples/simple-app-docker/src/simple-app/simple_test.py +++ b/examples/simple-app-docker/src/simple-app/simple_test.py @@ -9,6 +9,11 @@ # but WITHOUT ANY WARRANTY; without even the implied warranty of # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the # GNU Affero General Public License for more details. +# +# You should have received a copy of the GNU Affero General Public License +# along with this program. If not, see . + +"""A simple test object for examples.""" import time @@ -16,27 +21,30 @@ class SimpleTest(BaseTest): + """A basic test that is used in examples.""" - def __init__(self, test_name) -> None: + def __init__(self, test_name): + """Create new test object.""" super().__init__() - self.started_at = round(time.time() * 1000) - self.cnt = 0 + self._started_at = round(time.time() * 1000) + self.__cnt = 0 self.char_counter = {} self.test_name = test_name self.max_executions = self.next_max() def message(self, uuid): - print("%s:%s" % (self.cnt, uuid)) - self.cnt += 1 - self.check_end(self.cnt, self.max_executions) + """Print message to console.""" + print("%s:%s" % (self.__cnt, uuid)) + self.__cnt += 1 + self.check_end(self.__cnt, self.max_executions) info = self.make_char_count_map(uuid) self.merge(self.char_counter, info) - if self.cnt % 30 == 0: + if self.__cnt % 30 == 0: self.dump() def merge(self, char_counter, new_info): - + """Merge captured data.""" for key in new_info: new_val = new_info[key] @@ -46,16 +54,20 @@ def merge(self, char_counter, new_info): char_counter[key] = new_val + char_counter[key] def dump(self): + """Dump message to console.""" print(self.char_counter) self.char_counter = {} def check_end(self, value, max_executions): + """Check if we are at end.""" if value > max_executions: raise Exception("Hit max executions %s %s " % (value, max_executions)) def __str__(self) -> str: - return self.__class__.__name__ + ":" + self.test_name + ":" + str(self.started_at) + """Represent this as a string.""" + return self.__class__.__name__ + ":" + self.test_name + ":" + str(self._started_at) def reset(self): - self.cnt = 0 + """Reset the count.""" + self.__cnt = 0 self.max_executions = self.next_max() diff --git a/examples/simple-app/src/simple-app/base_test.py b/examples/simple-app/src/simple-app/base_test.py index 0f05dba..20d06e9 100644 --- a/examples/simple-app/src/simple-app/base_test.py +++ b/examples/simple-app/src/simple-app/base_test.py @@ -9,20 +9,29 @@ # but WITHOUT ANY WARRANTY; without even the implied warranty of # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the # GNU Affero General Public License for more details. +# +# You should have received a copy of the GNU Affero General Public License +# along with this program. If not, see . + +"""A simple test object for examples.""" import random import uuid class BaseTest: + """A basic test that is used in examples.""" def new_id(self): + """Create new id.""" return str(uuid.uuid4()) def next_max(self): + """Create new random max.""" return random.randint(1, 101) def make_char_count_map(self, in_str): + """Create char count map.""" res = {} for i in range(0, len(in_str)): diff --git a/examples/simple-app/src/simple-app/main.py b/examples/simple-app/src/simple-app/main.py index d1e6620..6b6c515 100644 --- a/examples/simple-app/src/simple-app/main.py +++ b/examples/simple-app/src/simple-app/main.py @@ -9,6 +9,12 @@ # but WITHOUT ANY WARRANTY; without even the implied warranty of # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the # GNU Affero General Public License for more details. +# +# You should have received a copy of the GNU Affero General Public License +# along with this program. If not, see . + +"""Example of using Deep with OTEl.""" + import signal import time @@ -23,17 +29,22 @@ class GracefulKiller: + """Ensure clean shutdown.""" + kill_now = False def __init__(self): + """Crate new killer.""" signal.signal(signal.SIGINT, self.exit_gracefully) signal.signal(signal.SIGTERM, self.exit_gracefully) def exit_gracefully(self, *args): + """Exit example.""" self.kill_now = True def main(): + """Run the example.""" killer = GracefulKiller() ts = SimpleTest("This is a test") while not killer.kill_now: diff --git a/examples/simple-app/src/simple-app/simple_test.py b/examples/simple-app/src/simple-app/simple_test.py index 3a7134f..5811e00 100644 --- a/examples/simple-app/src/simple-app/simple_test.py +++ b/examples/simple-app/src/simple-app/simple_test.py @@ -9,6 +9,11 @@ # but WITHOUT ANY WARRANTY; without even the implied warranty of # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the # GNU Affero General Public License for more details. +# +# You should have received a copy of the GNU Affero General Public License +# along with this program. If not, see . + +"""A simple test object for examples.""" import time @@ -16,8 +21,10 @@ class SimpleTest(BaseTest): + """A basic test that is used in examples.""" - def __init__(self, test_name) -> None: + def __init__(self, test_name): + """Create new test object.""" super().__init__() self._started_at = round(time.time() * 1000) self.__cnt = 0 @@ -26,6 +33,7 @@ def __init__(self, test_name) -> None: self.max_executions = self.next_max() def message(self, uuid): + """Print message to console.""" print("%s:%s" % (self.__cnt, uuid)) self.__cnt += 1 self.check_end(self.__cnt, self.max_executions) @@ -36,7 +44,7 @@ def message(self, uuid): self.dump() def merge(self, char_counter, new_info): - + """Merge captured data.""" for key in new_info: new_val = new_info[key] @@ -46,16 +54,20 @@ def merge(self, char_counter, new_info): char_counter[key] = new_val + char_counter[key] def dump(self): + """Dump message to console.""" print(self.char_counter) self.char_counter = {} def check_end(self, value, max_executions): + """Check if we are at end.""" if value > max_executions: raise Exception("Hit max executions %s %s " % (value, max_executions)) def __str__(self) -> str: + """Represent this as a string.""" return self.__class__.__name__ + ":" + self.test_name + ":" + str(self._started_at) def reset(self): + """Reset the count.""" self.__cnt = 0 self.max_executions = self.next_max() diff --git a/scripts/gendocs.py b/scripts/gendocs.py index 7151ed5..c4ae52a 100644 --- a/scripts/gendocs.py +++ b/scripts/gendocs.py @@ -13,6 +13,8 @@ # GNU Affero General Public License for more details. # # You should have received a copy of the GNU Affero General Public License +# along with this program. If not, see .# +# You should have received a copy of the GNU Affero General Public License # along with this program. If not, see . import glob diff --git a/src/deep/__init__.py b/src/deep/__init__.py index 06203e6..054b31a 100644 --- a/src/deep/__init__.py +++ b/src/deep/__init__.py @@ -9,6 +9,12 @@ # but WITHOUT ANY WARRANTY; without even the implied warranty of # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the # GNU Affero General Public License for more details. +# +# You should have received a copy of the GNU Affero General Public License +# along with this program. If not, see . + +"""Main entry for Deep Client.""" + import inspect import os @@ -19,6 +25,7 @@ def start(config=None): """ Start DEEP. + :param config: a custom config :return: the created Deep instance """ diff --git a/src/deep/api/__init__.py b/src/deep/api/__init__.py index db8f090..74e5c02 100644 --- a/src/deep/api/__init__.py +++ b/src/deep/api/__init__.py @@ -9,5 +9,10 @@ # but WITHOUT ANY WARRANTY; without even the implied warranty of # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the # GNU Affero General Public License for more details. +# +# You should have received a copy of the GNU Affero General Public License +# along with this program. If not, see . + +"""Main api for Deep.""" from deep.api.deep import Deep diff --git a/src/deep/api/attributes/__init__.py b/src/deep/api/attributes/__init__.py index 3f95fde..28d477a 100644 --- a/src/deep/api/attributes/__init__.py +++ b/src/deep/api/attributes/__init__.py @@ -12,6 +12,8 @@ # See the License for the specific language governing permissions and # limitations under the License. +"""Check and create attributes.""" + import threading from collections import OrderedDict from typing import MutableMapping, Optional, Union, Sequence @@ -43,7 +45,8 @@ def _clean_attribute_value( def _clean_attribute( key: str, value: types.AttributeValue, max_len: Optional[int] ) -> Optional[types.AttributeValue]: - """Checks if attribute value is valid and cleans it if required. + """ + Check if attribute value is valid and cleans it if required. The function returns the cleaned value or None if the value is not valid. @@ -57,7 +60,6 @@ def _clean_attribute( - Its length is greater than the maximum allowed length. - It needs to be encoded/decoded e.g, bytes to strings. """ - if not (key and isinstance(key, str)): logging.warning("invalid key `%s`. must be non-empty string.", key) return None @@ -132,6 +134,14 @@ def __init__( immutable: bool = True, max_value_len: Optional[int] = None, ): + """ + Create new attributes. + + :param maxlen: max number of attributes + :param attributes: existing attributes to copy + :param immutable: are these attributes immutable + :param max_value_len: max length of the attribute values + """ if maxlen is not None: if not isinstance(maxlen, int) or maxlen < 0: raise ValueError( @@ -148,14 +158,17 @@ def __init__( self._immutable = immutable def __repr__(self): + """Represent this as a string.""" return ( f"{type(self).__name__}({dict(self._dict)}, maxlen={self.maxlen})" ) def __getitem__(self, key): + """Get attribute value.""" return self._dict[key] def __setitem__(self, key, value): + """Set attribute value.""" if getattr(self, "_immutable", False): raise TypeError with self._lock: @@ -176,21 +189,26 @@ def __setitem__(self, key, value): self._dict[key] = value def __delitem__(self, key): + """Delete item from attributes.""" if getattr(self, "_immutable", False): raise TypeError with self._lock: del self._dict[key] def __iter__(self): + """Create iterator.""" with self._lock: return iter(self._dict.copy()) def __len__(self): + """Get number of attributes.""" return len(self._dict) def copy(self): + """Create a copy of these attributes.""" return self._dict.copy() def merge_in(self, attributes): + """Merge in another attributes object.""" for k, v in attributes.items(): self[k] = v diff --git a/src/deep/api/auth/__init__.py b/src/deep/api/auth/__init__.py index 6a74f14..c032295 100644 --- a/src/deep/api/auth/__init__.py +++ b/src/deep/api/auth/__init__.py @@ -9,6 +9,11 @@ # but WITHOUT ANY WARRANTY; without even the implied warranty of # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the # GNU Affero General Public License for more details. +# +# You should have received a copy of the GNU Affero General Public License +# along with this program. If not, see . + +"""Services for customizing auth connection.""" import abc import base64 @@ -19,23 +24,33 @@ class UnknownAuthProvider(Exception): - """This exception is thrown when the configured auth provider cannot be loaded""" + """This exception is thrown when the configured auth provider cannot be loaded.""" + pass class AuthProvider(abc.ABC): """ - This is the abstract class to define an AuthProvider. The 'provide' function will be called - when the system needs to get an auth token. + This is the abstract class to define an AuthProvider. + + The 'provide' function will be called when the system needs to get an auth token. """ - def __init__(self, config) -> None: + def __init__(self, config: ConfigService) -> None: + """ + Create a new auth provider. + + :param config: the deep config service + """ self._config = config @staticmethod def get_provider(config: ConfigService) -> Optional['AuthProvider']: """ + Get the provider to use. + Static function to load the correct auth provider based on the current config. + :param config: The agent config :return: the loaded provider :raises: UnknownAuthProvider if we cannot load the provider configured @@ -54,18 +69,26 @@ def get_provider(config: ConfigService) -> Optional['AuthProvider']: @abc.abstractmethod def provide(self): """ + Provide the auth metadata. + This is called when we need to get the auth for the request. + :return: a list of tuples to be attached to the outbound request """ raise NotImplementedError() class BasicAuthProvider(AuthProvider): - """ - This is a provider for http basic auth. This expects the config to provide a username and password. - """ + """This is a provider for http basic auth. This expects the config to provide a username and password.""" def provide(self): + """ + Provide the auth metadata. + + This is called when we need to get the auth for the request. + + :return: a list of tuples to be attached to the outbound request + """ username = self._config.SERVICE_USERNAME password = self._config.SERVICE_PASSWORD if username is not None and password is not None: diff --git a/src/deep/api/deep.py b/src/deep/api/deep.py index 0b557a4..86c5a68 100644 --- a/src/deep/api/deep.py +++ b/src/deep/api/deep.py @@ -9,6 +9,11 @@ # but WITHOUT ANY WARRANTY; without even the implied warranty of # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the # GNU Affero General Public License for more details. +# +# You should have received a copy of the GNU Affero General Public License +# along with this program. If not, see .from typing import Dict, List + +"""The main services for Deep.""" from typing import Dict, List from deep.api.plugin import load_plugins @@ -17,7 +22,6 @@ from deep.config import ConfigService from deep.config.tracepoint_config import TracepointConfigService from deep.grpc import GRPCService -from deep.logging.tracepoint_logger import TracepointLogger from deep.poll import LongPoll from deep.processor import TriggerHandler from deep.push import PushService @@ -26,11 +30,18 @@ class Deep: """ + The main service for deep. + This type acts as the main service for DEEP. It will initialise the other services and bind then together. DEEP is so small there is no need for service injection work. """ def __init__(self, config: 'ConfigService'): + """ + Create new deep service. + + :param config: the config to use. + """ self.started = False self.config = config self.grpc = GRPCService(self.config) @@ -41,23 +52,35 @@ def __init__(self, config: 'ConfigService'): self.trigger_handler = TriggerHandler(config, self.push) def start(self): + """Start Deep.""" if self.started: return plugins, attributes = load_plugins() self.config.plugins = plugins - self.config.resource = Resource.create(attributes) + self.config.resource = Resource.create(attributes.copy()) self.trigger_handler.start() self.grpc.start() self.poll.start() self.started = True def shutdown(self): + """Shutdown deep.""" if not self.started: return self.task_handler.flush() self.started = False + def register_tracepoint(self, path: str, line: int, args: Dict[str, str] = None, watches: List[str] = None) -> 'TracepointRegistration': + """ + Register a new tracepoint. + + :param path: the source path + :param line: the line number + :param args: the args + :param watches: the watches + :return: the new registration + """ if watches is None: watches = [] if args is None: @@ -67,18 +90,22 @@ def register_tracepoint(self, path: str, line: int, args: Dict[str, str] = None, class TracepointRegistration: - _cfg: TracePointConfig - _tpServ: TracepointConfigService + """Registration of a new tracepoint.""" def __init__(self, cfg: TracePointConfig, tracepoints: TracepointConfigService): + """ + Create a new registration. + + :param cfg: the created config + :param tracepoints: the config service + """ self._cfg = cfg self._tpServ = tracepoints def get(self) -> TracePointConfig: + """Get the created tracepoint.""" return self._cfg def unregister(self): + """Remove this custom tracepoint.""" self._tpServ.remove_custom(self._cfg) - - def tracepoint_logger(self, logger: 'TracepointLogger'): - self.config.tracepoint_logger = logger diff --git a/src/deep/api/plugin/__init__.py b/src/deep/api/plugin/__init__.py index 4a32805..affc03f 100644 --- a/src/deep/api/plugin/__init__.py +++ b/src/deep/api/plugin/__init__.py @@ -9,10 +9,19 @@ # but WITHOUT ANY WARRANTY; without even the implied warranty of # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the # GNU Affero General Public License for more details. +# +# You should have received a copy of the GNU Affero General Public License +# along with this program. If not, see . + +"""Load and handle plugins.""" import abc import os from importlib import import_module +from typing import TYPE_CHECKING + +if TYPE_CHECKING: + from typing import Tuple from deep import logging from deep.api.attributes import BoundedAttributes @@ -24,7 +33,7 @@ ] -def plugin_generator(configured): +def __plugin_generator(configured): for plugin in configured: try: module, cls = plugin.rsplit(".", 1) @@ -36,10 +45,17 @@ def plugin_generator(configured): ) -def load_plugins(): +def load_plugins() -> 'Tuple[list[Plugin], BoundedAttributes]': + """ + Load all the deep plugins. + + Attempt to load each plugin, if successful merge a attributes list of each plugin. + + :return: the loaded plugins and attributes. + """ bounded_attributes = BoundedAttributes(immutable=False) loaded = [] - for plugin in plugin_generator(DEEP_PLUGINS): + for plugin in __plugin_generator(DEEP_PLUGINS): try: plugin_instance = plugin() if not plugin_instance.is_active(): @@ -56,10 +72,13 @@ def load_plugins(): class Plugin(abc.ABC): """ + A deep Plugin. + This type defines a plugin for deep, these plugins allow for extensions to how deep decorates data. """ def __init__(self, name=None): + """Create a new.""" super(Plugin, self).__init__() if name is None: self._name = self.__class__.__name__ @@ -68,24 +87,42 @@ def __init__(self, name=None): @property def name(self): + """The name of the plugin.""" return self._name - def is_active(self): - # type: ()-> bool + def is_active(self) -> bool: + """ + Is the plugin active. + + Check the value of the environment value of the module name + class name. It set to + 'false' this plugin is not active. + """ getenv = os.getenv("{0}.{1}".format(self.__class__.__module__, self.__class__.__name__), 'True') return str2bool(getenv) @abc.abstractmethod - def load_plugin(self) -> None: + def load_plugin(self) -> BoundedAttributes: + """ + Load the plugin. + + :return: any values to attach to the client resource. + """ raise NotImplementedError() @abc.abstractmethod def collect_attributes(self) -> BoundedAttributes: + """ + Collect attributes to attach to snapshot. + + :return: the attributes to attach. + """ raise NotImplementedError() class DidNotEnable(Exception): """ + Raised when failed to load plugin. + The plugin could not be enabled due to a trivial user error like `otel` not being installed for the `OTelPlugin`. """ diff --git a/src/deep/api/plugin/otel.py b/src/deep/api/plugin/otel.py index dcea449..5268248 100644 --- a/src/deep/api/plugin/otel.py +++ b/src/deep/api/plugin/otel.py @@ -9,6 +9,11 @@ # but WITHOUT ANY WARRANTY; without even the implied warranty of # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the # GNU Affero General Public License for more details. +# +# You should have received a copy of the GNU Affero General Public License +# along with this program. If not, see . + +"""Provide plugin for Deep to connect to OTEL.""" from typing import Optional @@ -22,40 +27,19 @@ raise DidNotEnable("opentelemetry is not installed", e) -def span_name(span): - # type: (_Span)-> Optional[str] - return span.name if span.name else None - - -def span_id(span): - # type: (_Span)-> Optional[str] - return (format_span_id(span.context.span_id)) if span else None - - -def trace_id(span): - # type: (_Span)-> Optional[str] - return (format_trace_id(span.context.trace_id)) if span else None - - -def get_span(): - # type: () -> Optional[_Span] - span = trace.get_current_span() - if isinstance(span, _Span): - return span - return None - - -def format_span_id(_id): - return format(_id, "016x") - - -def format_trace_id(_id): - return format(_id, "032x") - - class OTelPlugin(Plugin): + """ + Deep Otel plugin. + + Provide span and trace information to the snapshot. + """ def load_plugin(self) -> Optional[BoundedAttributes]: + """ + Load the plugin. + + :return: any values to attach to the client resource. + """ provider = trace.get_tracer_provider() if isinstance(provider, TracerProvider): # noinspection PyUnresolvedReferences @@ -65,11 +49,47 @@ def load_plugin(self) -> Optional[BoundedAttributes]: return None def collect_attributes(self) -> Optional[BoundedAttributes]: - span = get_span() + """ + Collect attributes to attach to snapshot. + + :return: the attributes to attach. + """ + span = OTelPlugin.__get_span() if span is not None: return BoundedAttributes(attributes={ - "span_name": span_name(span), - "trace_id": trace_id(span), - "span_id": span_id(span) + "span_name": OTelPlugin.__span_name(span), + "trace_id": OTelPlugin.__trace_id(span), + "span_id": OTelPlugin.__span_id(span) }) return None + + @staticmethod + def __span_name(span): + # type: (_Span)-> Optional[str] + return span.name if span.name else None + + @staticmethod + def __span_id(span): + # type: (_Span)-> Optional[str] + return (OTelPlugin.__format_span_id(span.context.__span_id)) if span else None + + @staticmethod + def __trace_id(span): + # type: (_Span)-> Optional[str] + return (OTelPlugin.__format_trace_id(span.context.__trace_id)) if span else None + + @staticmethod + def __get_span(): + # type: () -> Optional[_Span] + span = trace.get_current_span() + if isinstance(span, _Span): + return span + return None + + @staticmethod + def __format_span_id(_id): + return format(_id, "016x") + + @staticmethod + def __format_trace_id(_id): + return format(_id, "032x") diff --git a/src/deep/api/plugin/python.py b/src/deep/api/plugin/python.py index b4a7933..2a08f31 100644 --- a/src/deep/api/plugin/python.py +++ b/src/deep/api/plugin/python.py @@ -11,8 +11,12 @@ # GNU Affero General Public License for more details. # # You should have received a copy of the GNU Affero General Public License +# along with this program. If not, see .# +# You should have received a copy of the GNU Affero General Public License # along with this program. If not, see . +"""Simple plugin for deep, decorating some python information.""" + import platform import threading @@ -21,12 +25,28 @@ class PythonPlugin(Plugin): + """ + Deep python plugin. + + This plugin provides the python version to the resource, and the thread name to the attributes. + """ + def load_plugin(self): + """ + Load the plugin. + + :return: any values to attach to the client resource. + """ return BoundedAttributes(attributes={ "python_version": platform.python_version(), }) def collect_attributes(self): + """ + Collect attributes to attach to snapshot. + + :return: the attributes to attach. + """ thread = threading.current_thread() return BoundedAttributes(attributes={ diff --git a/src/deep/api/resource/__init__.py b/src/deep/api/resource/__init__.py index 4451f6a..3984ea2 100644 --- a/src/deep/api/resource/__init__.py +++ b/src/deep/api/resource/__init__.py @@ -12,6 +12,8 @@ # See the License for the specific language governing permissions and # limitations under the License. +"""Constant values for Resource data.""" + import abc import os import typing @@ -108,6 +110,12 @@ class Resource: def __init__( self, attributes: Attributes, schema_url: typing.Optional[str] = None ): + """ + Create new resource. + + :param attributes: the attributes + :param schema_url: the schema url + """ self._attributes = BoundedAttributes(attributes=attributes) if schema_url is None: schema_url = "" @@ -118,7 +126,8 @@ def create( attributes: typing.Optional[Attributes] = None, schema_url: typing.Optional[str] = None, ) -> "Resource": - """Creates a new `Resource` from attributes. + """ + Create a new `Resource` from attributes. Args: attributes: Optional zero or more key-value pairs. @@ -146,18 +155,24 @@ def create( @staticmethod def get_empty() -> "Resource": + """Get an empty resource.""" return _EMPTY_RESOURCE @property def attributes(self) -> BoundedAttributes: + """The underlying attributes for the resource.""" return self._attributes @property def schema_url(self) -> str: + """The schema url for the resource.""" return self._schema_url def merge(self, other: "Resource") -> "Resource": - """Merges this resource and an updating resource into a new `Resource`. + """ + Merge another resource into this one. + + Merges this resource and an updating resource into a new `Resource`. If a key exists on both the old and updating resource, the value of the updating resource will override the old resource value. @@ -193,6 +208,7 @@ def merge(self, other: "Resource") -> "Resource": return Resource(merged_attributes, schema_url) def __eq__(self, other: object) -> bool: + """Check if other object is equals to this one.""" if not isinstance(other, Resource): return False return ( @@ -201,11 +217,13 @@ def __eq__(self, other: object) -> bool: ) def __hash__(self): + """Create hash value for this object.""" return hash( f"{dumps(self._attributes.copy(), sort_keys=True)}|{self._schema_url}" ) def to_json(self, indent=4) -> str: + """Convert this object to json.""" return dumps( { "attributes": dict(self._attributes), @@ -226,17 +244,35 @@ def to_json(self, indent=4) -> str: class ResourceDetector(abc.ABC): + """Detect the resource information for Deep.""" + def __init__(self, raise_on_error=False): + """ + Create a new detector. + + :param raise_on_error: should raise exception on error + """ self.raise_on_error = raise_on_error @abc.abstractmethod def detect(self) -> "Resource": + """ + Create a resource. + + :return: the created resrouce + """ raise NotImplementedError() class DeepResourceDetector(ResourceDetector): - # pylint: disable=no-self-use + """Detect the resource information for Deep.""" + def detect(self) -> "Resource": + """ + Create a resource from the discovered environment data. + + :return: the created resource + """ env_resources_items = os.environ.get(DEEP_RESOURCE_ATTRIBUTES) env_resource_map = {} diff --git a/src/deep/api/tracepoint/__init__.py b/src/deep/api/tracepoint/__init__.py index 37c5fc6..2b15065 100644 --- a/src/deep/api/tracepoint/__init__.py +++ b/src/deep/api/tracepoint/__init__.py @@ -9,6 +9,11 @@ # but WITHOUT ANY WARRANTY; without even the implied warranty of # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the # GNU Affero General Public License for more details. +# +# You should have received a copy of the GNU Affero General Public License +# along with this program. If not, see . + +"""Internal types for GRPC data.""" from .eventsnapshot import StackFrame, EventSnapshot, Variable, VariableId, WatchResult from .tracepoint_config import TracePointConfig diff --git a/src/deep/api/tracepoint/eventsnapshot.py b/src/deep/api/tracepoint/eventsnapshot.py index 1b90a5d..197802b 100644 --- a/src/deep/api/tracepoint/eventsnapshot.py +++ b/src/deep/api/tracepoint/eventsnapshot.py @@ -9,9 +9,14 @@ # but WITHOUT ANY WARRANTY; without even the implied warranty of # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the # GNU Affero General Public License for more details. +# +# You should have received a copy of the GNU Affero General Public License +# along with this program. If not, see . + +"""Types for the captured data.""" import random -from typing import List, Dict, Optional +from typing import Optional, Dict, List from deep.api.attributes import BoundedAttributes from deep.api.resource import Resource @@ -19,14 +24,21 @@ class EventSnapshot: - """ - This is the model for the snapshot that is uploaded to the services - """ + """This is the model for the snapshot that is uploaded to the services.""" def __init__(self, tracepoint, ts, resource, frames, var_lookup: Dict[str, 'Variable']): + """ + Create a new snapshot object. + + :param tracepoint: the tracepoint object + :param ts: the time in nanoseconds + :param resource: the client resource + :param frames: the captured frames + :param var_lookup: the captured variables. + """ self._id = random.getrandbits(128) self._tracepoint = tracepoint - self._var_lookup: Dict[str, 'Variable'] = var_lookup + self._var_lookup: dict[str, 'Variable'] = var_lookup self._ts_nanos = ts self._frames = frames self._watches = [] @@ -37,77 +49,101 @@ def __init__(self, tracepoint, ts, resource, frames, var_lookup: Dict[str, 'Vari self._log = None def complete(self): + """Close and complete the snapshot.""" if not self._open: return self._duration_nanos = time_ns() - self._ts_nanos self._open = False def is_open(self): + """Is this snapshot still open.""" return self._open def add_watch_result(self, watch_result: 'WatchResult'): + """ + Append a watch result to the snapshot. + + :param watch_result: the result to append. + :return: + """ if self.is_open(): self.watches.append(watch_result) def merge_var_lookup(self, lookup: Dict[str, 'Variable']): + """ + Merge additional variables into the var lookup. + + :param lookup: the values to merge + """ if self.is_open(): self._var_lookup.update(lookup) @property def id(self): + """The id of this snapshot.""" return self._id @property def tracepoint(self): + """The tracepoint that triggered this snapshot.""" return self._tracepoint @property def var_lookup(self): + """The captured var lookup.""" return self._var_lookup @property def ts_nanos(self): + """The time in nanoseconds, this snapshot was triggered.""" return self._ts_nanos @property def frames(self): + """The captured frames.""" return self._frames @property def watches(self): + """The watch results.""" return self._watches @property def attributes(self) -> BoundedAttributes: + """The snapshot attributes.""" return self._attributes @property def duration_nanos(self): + """The duration in nanoseconds.""" return self._duration_nanos @property def resource(self): + """The client resource information.""" return self._resource @property def log_msg(self): + """Get the processed log message.""" return self._log @log_msg.setter def log_msg(self, msg): + """Set the processed log message.""" self._log = msg def __str__(self) -> str: + """Represent this as a string.""" return str(self.__dict__) def __repr__(self) -> str: + """Represent this as a string.""" return self.__str__() class StackFrame: - """ - This represents a frame of code that is being executed - """ + """This represents a frame of code that is being executed.""" def __init__(self, file_name, @@ -123,6 +159,22 @@ def __init__(self, transpiled_column_number=0, app_frame=False ): + """ + Create a new StackFrame object. + + :param file_name: The full file path + :param short_path: The short file path + :param method_name: The method name + :param line_number: The line number + :param variables: Variables captured on this frame + :param class_name: The class name + :param is_async: Is the frame an async frame + :param column_number: The column number + :param transpiled_file_name: The transpiled file name + :param transpiled_line_number: The transpiled line number + :param transpiled_column_number: The transpiled column number + :param app_frame: Is this frame in the user app + """ self._file_name = file_name self._short_path = short_path self._method_name = method_name @@ -138,61 +190,75 @@ def __init__(self, @property def file_name(self): + """The full file path.""" return self._file_name @property def short_path(self): + """The short file path.""" return self._short_path @property def method_name(self): + """The method name.""" return self._method_name @property def line_number(self): + """The line number.""" return self._line_number @property def class_name(self): + """The class name.""" return self._class_name @property def is_async(self): + """Is the frame an async frame.""" return self._async @property def column_number(self): + """The column number.""" return self._column_number @property def transpiled_file_name(self): + """The transpiled file name.""" return self._transpiled_file_name @property def transpiled_line_number(self): + """The transpiled line number.""" return self._transpiled_line_number @property def transpiled_column_number(self): + """The transpiled column number.""" return self._transpiled_column_number @property def variables(self): + """Variables captured on this frame.""" return self._variables @property def app_frame(self): + """Is this frame in the user app.""" return self._app_frame def __str__(self) -> str: + """Represent this as a string.""" return str(self.__dict__) def __repr__(self) -> str: + """Represent this as a string.""" return self.__str__() class Variable: - """This represents a captured variable value""" + """This represents a captured variable value.""" def __init__(self, var_type, @@ -201,6 +267,15 @@ def __init__(self, children, truncated, ): + """ + Create a new Variable object. + + :param var_type: the type of the variable. + :param value: the value as a string + :param var_hash: the identity hash of the value + :param children: list of child VariableIds + :param truncated: is the value string truncated. + """ self._type = var_type self._value = value self._hash = var_hash @@ -209,34 +284,51 @@ def __init__(self, @property def type(self): + """The type of this value.""" return self._type @property def value(self): + """The string value of variable..""" return self._value @property def hash(self): + """The identity hash of this value.""" return self._hash @property def children(self) -> List['VariableId']: + """The children of this value.""" return self._children @property def truncated(self): + """Is the string value truncated.""" return self._truncated def __str__(self) -> str: + """Represent this as a string.""" return str(self.__dict__) def __repr__(self) -> str: + """Represent this as a string.""" return self.__str__() class VariableId: """ - This represents an variable id, that is used for de duplication + This represents a variable id, that is used for de duplication. + + A VariableID is a pointer to a reference within the var lookup of the snapshot. Each VariableID can have + different names and modifiers, but point to the same value. + + e.g. + val = "Ben" + name = val + + Both 'val' and 'name' have the value 'Ben' to prevent duplication of this in the var lookup, we use the + VariableId to point to the value using the vid property. """ def __init__(self, @@ -245,6 +337,14 @@ def __init__(self, modifiers=None, original_name=None ): + """ + Create a new variable object. + + :param vid: the variable id + :param name: the variable name + :param modifiers: the variable modifiers + :param original_name: the original name + """ if modifiers is None: modifiers = [] self._vid = vid @@ -254,27 +354,34 @@ def __init__(self, @property def vid(self): + """Get variable id.""" return self._vid @property def name(self): + """Get variable name.""" return self._name @property def original_name(self): + """Get variable original name.""" return self._original_name @property def modifiers(self): + """Get variable modifiers.""" return self._modifiers def __str__(self) -> str: + """Represent this as a string.""" return str(self.__dict__) def __repr__(self) -> str: + """Represent this as a string.""" return self.__str__() def __eq__(self, o) -> bool: + """Check if the variable id matches.""" if not isinstance(o, VariableId): return False @@ -285,27 +392,35 @@ def __eq__(self, o) -> bool: class WatchResult: - """ - This is the result of a watch expression - """ + """This is the result of a watch expression.""" def __init__(self, expression: str, result: Optional['VariableId'], error: Optional[str] = None ): + """ + Create new watch result. + + :param expression: the expression used + :param result: the result of the expression + :param error: the error captured during execution + """ self._expression = expression self._result = result self._error = error @property def expression(self) -> str: + """The watch expression.""" return self._expression @property def result(self) -> Optional['VariableId']: + """The good result.""" return self._result @property def error(self) -> Optional[str]: + """The error.""" return self._error diff --git a/src/deep/api/tracepoint/tracepoint_config.py b/src/deep/api/tracepoint/tracepoint_config.py index 53618b2..2b16342 100644 --- a/src/deep/api/tracepoint/tracepoint_config.py +++ b/src/deep/api/tracepoint/tracepoint_config.py @@ -9,6 +9,12 @@ # but WITHOUT ANY WARRANTY; without even the implied warranty of # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the # GNU Affero General Public License for more details. +# +# You should have received a copy of the GNU Affero General Public License +# along with this program. If not, see . + +"""Internal type for configured tracepoints.""" + from typing import List # Below are constants used in the configuration of a tracepoint @@ -55,7 +61,10 @@ def frame_type_ordinal(frame_type) -> int: """ - Convert a frame type to an ordinal (essentially making it an enum). This is useful for ordering. + Convert a frame type to an ordinal (essentially making it an enum). + + This is useful for ordering. + :param frame_type: the frame type :return: the ordinal of the type """ @@ -70,17 +79,22 @@ def frame_type_ordinal(frame_type) -> int: class TracepointWindow: - """ - This is used to handle validating the time frame for the tracepoint - """ + """This is used to handle validating the time frame for the tracepoint.""" - def __init__(self, start, end): + def __init__(self, start: int, end: int): + """ + Create a new tracepoint window. + + :param start: the window start time + :param end: the window end time + """ self._start = start self._end = end def in_window(self, ts): """ - Is the provided time in the configured window + Is the provided time in the configured window. + :param ts: time in ms :return: true, if the time is within the configured window, else false """ @@ -102,11 +116,21 @@ def in_window(self, ts): class TracePointConfig: """ - This represents the configuration of a single tracepoint, this is a python version of the GRPC - data collected from the LongPoll. + This represents the configuration of a single tracepoint. + + This is a python version of the GRPC data collected from the LongPoll. """ def __init__(self, tp_id: str, path: str, line_no: int, args: dict, watches: List[str]): + """ + Create a new tracepoint config. + + :param tp_id: the tracepoint id + :param path: the tracepoint source file + :param line_no: the tracepoint line number + :param args: the tracepoint args + :param watches: the tracepoint watches + """ self._id = tp_id self._path = path self._line_no = line_no @@ -117,36 +141,43 @@ def __init__(self, tp_id: str, path: str, line_no: int, args: dict, watches: Lis @property def id(self): + """The tracepoint id.""" return self._id @property def path(self): + """The tracepoint source file.""" return self._path @property def line_no(self): + """The tracepoint line number.""" return self._line_no @property def args(self): + """The tracepoint args.""" return self._args @property def watches(self): + """The tracepoint watches.""" return self._watches @property def frame_type(self): + """The tracepoint frame type.""" return self.get_arg(FRAME_TYPE, SINGLE_FRAME_TYPE) @property def stack_type(self): + """The tracepoint stack type.""" return self.get_arg(STACK_TYPE, STACK) @property def fire_count(self): """ - Get the allowed number of triggers + Get the allowed number of triggers. :return: the configured number of triggers, or -1 for unlimited triggers """ @@ -154,14 +185,29 @@ def fire_count(self): @property def condition(self): + """The tracepoint condition.""" return self.get_arg(CONDITION, None) def get_arg(self, name: str, default_value: any): + """ + Get an arg from tracepoint args. + + :param name: the argument name + :param default_value: the default value + :return: the value, or the default value + """ if name in self._args: return self._args[name] return default_value def get_arg_int(self, name: str, default_value: int): + """ + Get an argument from the args as an int. + + :param name: the argument name + :param default_value: the default value to use. + :return: the value as an int, or the default value + """ try: return int(self.get_arg(name, default_value)) except ValueError: @@ -169,7 +215,10 @@ def get_arg_int(self, name: str, default_value: int): def can_trigger(self, ts): """ - Check if the tracepoint can trigger, this is to check the config. e.g. fire count, fire windows etc + Check if the tracepoint can trigger. + + This is to check the config. e.g. fire count, fire windows etc + :param ts: the time the tracepoint has been triggered :return: true, if we should collect data; else false """ @@ -191,34 +240,58 @@ def can_trigger(self, ts): return True def record_triggered(self, ts): - """This is called when the tracepoint has been processed.""" + """ + Record a fire. + + Call this to record this tracepoint being triggered. + + :param ts: the time in nanoseconds + """ self._stats.fire(ts) def __str__(self) -> str: + """Represent this object as a string.""" return str({'id': self._id, 'path': self._path, 'line_no': self._line_no, 'args': self._args, 'watches': self._watches}) def __repr__(self) -> str: + """Represent this object as a string.""" return self.__str__() class TracepointExecutionStats: - """ - This keeps track of the tracepoint stats, so we can check fire counts etc - """ + """This keeps track of the tracepoint stats, so we can check fire counts etc.""" def __init__(self): + """Create a new stats object.""" self._fire_count = 0 self._last_fire = 0 - def fire(self, ts): + def fire(self, ts: int): + """ + Record a fire. + + Call this to record this tracepoint being triggered. + + :param ts: the time in nanoseconds + """ self._fire_count += 1 self._last_fire = ts @property def fire_count(self): + """ + The number of times this tracepoint has fired. + + :return: the number of times this has fired. + """ return self._fire_count @property def last_fire(self): + """ + The time this tracepoint last fired. + + :return: the time in nanoseconds. + """ return self._last_fire diff --git a/src/deep/api/types.py b/src/deep/api/types.py index 6e86052..dc4e7a2 100644 --- a/src/deep/api/types.py +++ b/src/deep/api/types.py @@ -12,6 +12,8 @@ # See the License for the specific language governing permissions and # limitations under the License. +"""Types used for attributes in resources.""" + from typing import Mapping, Optional, Sequence, Tuple, Union AttributeValue = Union[ diff --git a/src/deep/config/__init__.py b/src/deep/config/__init__.py index 7cdb62c..4d27d80 100644 --- a/src/deep/config/__init__.py +++ b/src/deep/config/__init__.py @@ -9,15 +9,22 @@ # but WITHOUT ANY WARRANTY; without even the implied warranty of # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the # GNU Affero General Public License for more details. +# +# You should have received a copy of the GNU Affero General Public License +# along with this program. If not, see . + +""" +Config values for deep. + +Here we have the initial values for the config, there can be set as either static values, environment values + or functions. +""" import os import sys from .config_service import ConfigService -# Here we have the initial values for the config, there can be set as either static values, environment -# values or functions. - LOGGING_CONF = os.getenv('DEEP_LOGGING_CONF', None) '''The path to the logging config file to use''' @@ -36,7 +43,11 @@ # noinspection PyPep8Naming def IN_APP_INCLUDE(): - """The packages to mark as in app packages. (default: ''). Must be a command (,) seperated list.""" + """ + Get the included app packages. + + The packages to mark as in app packages. (default: ''). Must be a command (,) seperated list. + """ user_defined = os.getenv('DEEP_IN_APP_INCLUDE', None) if user_defined is None: return [] @@ -47,7 +58,11 @@ def IN_APP_INCLUDE(): # noinspection PyPep8Naming def IN_APP_EXCLUDE(): - """The packages to mark as NOT in app packages. (default: ''). Must be a command (,) seperated list.""" + """ + Get the exclude app packages. + + The packages to mark as NOT in app packages. (default: ''). Must be a command (,) seperated list. + """ user_defined = os.getenv('DEEP_IN_APP_EXCLUDE', None) if user_defined is None: user_defined = [] diff --git a/src/deep/config/config_service.py b/src/deep/config/config_service.py index 0fdeeb1..5d06f85 100644 --- a/src/deep/config/config_service.py +++ b/src/deep/config/config_service.py @@ -9,6 +9,11 @@ # but WITHOUT ANY WARRANTY; without even the implied warranty of # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the # GNU Affero General Public License for more details. +# +# You should have received a copy of the GNU Affero General Public License +# along with this program. If not, see . + +"""Service for handling deep config.""" import os from typing import Any, List, Dict @@ -16,29 +21,31 @@ from deep import logging from deep.api.plugin import Plugin from deep.api.resource import Resource -from deep.config.tracepoint_config import TracepointConfigService +from deep.config.tracepoint_config import TracepointConfigService, ConfigUpdateListener from deep.logging.tracepoint_logger import DefaultLogger, TracepointLogger class ConfigService: - """ - This is the main service that handles config for DEEP. - """ + """This is the main service that handles config for DEEP.""" def __init__(self, custom: Dict[str, any]): """ - Create a new config object + Create a new config object. + :param custom: any custom values that are passed to DEEP """ self._plugins = [] self.__custom = custom self._resource = None self._tracepoint_config = TracepointConfigService() - self._tracepoint_logger: 'TracepointLogger' = DefaultLogger(self) + self._tracepoint_logger: 'TracepointLogger' = DefaultLogger() def __getattribute__(self, name: str) -> Any: """ + Get attribute from config. + A custom attribute processor to load the config values + :param name: the key to load :return: the loaded value or None """ @@ -75,41 +82,68 @@ def __getattribute__(self, name: str) -> Any: return attr def __setattr__(self, name: str, value: Any) -> None: + """Set attribute on this config.""" super().__setattr__(name, value) def set_task_handler(self, task_handler): + """ + Set the task handler to use. + + :param task_handler: the taskhandler + """ self._tracepoint_config.set_task_handler(task_handler) @property def resource(self) -> Resource: + """Get the resource that describes this client.""" return self._resource @resource.setter def resource(self, new_resource): + """Set the resource that describes this client.""" self._resource = new_resource @property def plugins(self) -> List[Plugin]: + """Get the active deep client plugins.""" return self._plugins @plugins.setter def plugins(self, plugins): + """Set the active deep client plugins.""" self._plugins = plugins @property def tracepoints(self) -> 'TracepointConfigService': + """The tracepoint config service.""" return self._tracepoint_config - def add_listener(self, listener): + def add_listener(self, listener: 'ConfigUpdateListener'): + """ + Add a new listener to the config. + + :param listener: the listener to add + """ self._tracepoint_config.add_listener(listener) @property def tracepoint_logger(self) -> 'TracepointLogger': + """Get the tracepoint logger.""" return self._tracepoint_logger @tracepoint_logger.setter def tracepoint_logger(self, logger: 'TracepointLogger'): + """Set the tracepoint logger.""" self._tracepoint_logger = logger def log_tracepoint(self, log_msg: str, tp_id: str, snap_id: str): + """ + Log the dynamic log message. + + Pass the processed log to the tracepoint logger. + + :param (str) log_msg: the log message to log + :param (str) tp_id: the id of the tracepoint that generated this log + :param (str) snap_id: the is of the snapshot that was created by this tracepoint + """ self._tracepoint_logger.log_tracepoint(log_msg, tp_id, snap_id) diff --git a/src/deep/config/tracepoint_config.py b/src/deep/config/tracepoint_config.py index a4df6a1..bc4f77f 100644 --- a/src/deep/config/tracepoint_config.py +++ b/src/deep/config/tracepoint_config.py @@ -9,6 +9,11 @@ # but WITHOUT ANY WARRANTY; without even the implied warranty of # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the # GNU Affero General Public License for more details. +# +# You should have received a copy of the GNU Affero General Public License +# along with this program. If not, see . + +"""Maintain the current config of the tracepoints.""" import abc import logging @@ -19,26 +24,33 @@ class TracepointConfigService: - """This service deals with new responses from the LongPoll""" + """This service deals with new responses from the LongPoll.""" def __init__(self) -> None: + """Create new tracepoint config service.""" self._custom = [] self._tracepoint_config = [] self._current_hash = None self._last_update = 0 self._task_handler = None - self._listeners = [] + self._listeners: list[ConfigUpdateListener] = [] def update_no_change(self, ts): """ + Update no change detected. + This is called when the response says the config has not changed + :param ts: the ts of the last poll, in ms """ self._last_update = ts def update_new_config(self, ts, new_hash, new_config): """ + Update to the new config. + This is called when there is a change in the config, this will trigger a call to all listeners + :param ts: the ts of the last poll, in ms :param new_hash: the new config hash :param new_config: the new config values @@ -48,9 +60,9 @@ def update_new_config(self, ts, new_hash, new_config): self._last_update = ts self._current_hash = new_hash self._tracepoint_config = new_config - self.trigger_update(old_hash, old_config) + self.__trigger_update(old_hash, old_config) - def trigger_update(self, old_hash, old_config): + def __trigger_update(self, old_hash, old_config): ts = self._last_update if self._task_handler is not None: future = self._task_handler.submit_task(self.update_listeners, self._last_update, old_hash, @@ -58,11 +70,25 @@ def trigger_update(self, old_hash, old_config): future.add_done_callback(lambda _: logging.debug("Completed processing new config %s", ts)) def set_task_handler(self, task_handler): - """Link in task handler""" + """ + Set the task handler to use. + + :param task_handler: the taskhandler + """ self._task_handler = task_handler def update_listeners(self, ts, old_hash, current_hash, old_config, new_config): - """This is called to update any listeners that the config has changed""" + """ + Update the registered listeners. + + This is called to update any listeners that the config has changed + + :param ts: the ts of the update + :param old_hash: the old hash + :param current_hash: the new hash value + :param old_config: the old config + :param new_config: the new config + """ listeners_copy = self._listeners.copy() for listeners in listeners_copy: try: @@ -70,25 +96,56 @@ def update_listeners(self, ts, old_hash, current_hash, old_config, new_config): except Exception: logging.exception("Error updating listener %s", listeners) - def add_listener(self, listener): - """Add a new listener to the config""" + def add_listener(self, listener: 'ConfigUpdateListener'): + """ + Add a new listener to the config. + + :param listener: the listener to add + """ self._listeners.append(listener) @property def current_config(self): + """ + The current tracepoint config. + + :return: the config + """ return self._tracepoint_config @property def current_hash(self): + """ + The current hash. + + The hash is updated only when the config is changed. It is used by the server and client to + reduce the number of updates. + + :return: the current hash. + """ return self._current_hash def add_custom(self, path: str, line: int, args: Dict[str, str], watches: List[str]) -> TracePointConfig: + """ + Crate a new tracepoint from the input. + + :param path: the source file name + :param line: the source line number + :param args: the tracepoint args + :param watches: the tracepoint watches + :return: the new TracePointConfig + """ config = TracePointConfig(str(uuid.uuid4()), path, line, args, watches) self._custom.append(config) - self.trigger_update(None, None) + self.__trigger_update(None, None) return config def remove_custom(self, config: TracePointConfig): + """ + Remove a custom tracepoint config. + + :param config: the config to remove + """ for idx, cfg in enumerate(self._custom): if cfg.id == config.id: del self._custom[idx] @@ -96,14 +153,13 @@ def remove_custom(self, config: TracePointConfig): class ConfigUpdateListener(abc.ABC): - """ - Class to describe a config listener - """ + """Class to describe a config listener.""" @abc.abstractmethod def config_change(self, ts, old_hash, current_hash, old_config, new_config): """ - Called when the config has changed + Process an update to the tracepoint config. + :param ts: the ts of the new config :param old_hash: the old config hash :param current_hash: the new config hash diff --git a/src/deep/grpc/__init__.py b/src/deep/grpc/__init__.py index 43c5aea..93faa7e 100644 --- a/src/deep/grpc/__init__.py +++ b/src/deep/grpc/__init__.py @@ -9,6 +9,17 @@ # but WITHOUT ANY WARRANTY; without even the implied warranty of # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the # GNU Affero General Public License for more details. +# +# You should have received a copy of the GNU Affero General Public License +# along with this program. If not, see . + +""" +Collection of functions to convert to protobuf version of types. + +We do not use the protobuf types throughout the project as they do not autocomplete or +have type definitions that work in IDE. It also makes it easier to deal with agent functionality by +having local types we can modify. +""" # noinspection PyUnresolvedReferences from deepproto.proto.common.v1.common_pb2 import KeyValue, AnyValue, ArrayValue, KeyValueList @@ -20,6 +31,12 @@ def convert_value(value): + """ + Convert a value from the python type. + + :param value: the value to convert + :return: the value wrapped in the appropriate AnyValue type. + """ """Convert the attributes to jaeger tags.""" if isinstance(value, bool): return AnyValue(bool_value=value) @@ -32,29 +49,41 @@ def convert_value(value): if isinstance(value, bytes): return AnyValue(bytes_value=value) if isinstance(value, dict): - return AnyValue(kvlist_value=value_as_dict(value)) + return AnyValue(kvlist_value=__value_as_dict(value)) if isinstance(value, list): - return AnyValue(array_value=value_as_list(value)) + return AnyValue(array_value=__value_as_list(value)) return None -def value_as_dict(value): +def __value_as_dict(value): return KeyValueList(values=[KeyValue(key=k, value=convert_value(v)) for k, v in value.items()]) -def value_as_list(value): +def __value_as_list(value): return ArrayValue(values=[convert_value(val) for val in value]) def convert_resource(resource): - return convert_attributes(resource.attributes) + """ + Convert a internal resource to GRPC type. + + :param resource: the resource to convert + :return: the converted type as GRPC. + """ + return __convert_attributes(resource.attributes) -def convert_attributes(attributes): +def __convert_attributes(attributes): return Resource(dropped_attributes_count=attributes.dropped, attributes=[KeyValue(key=k, value=convert_value(v)) for k, v in attributes.items()]) def convert_response(response): + """ + Convert a response from GRPC to internal types. + + :param response: the grpc response. + :return: the internal types for tracepoints + """ return [TracePointConfig(r.ID, r.path, r.line_number, dict(r.args), [w for w in r.watches]) for r in response] diff --git a/src/deep/grpc/grpc_service.py b/src/deep/grpc/grpc_service.py index be225ef..f1608d4 100644 --- a/src/deep/grpc/grpc_service.py +++ b/src/deep/grpc/grpc_service.py @@ -9,20 +9,29 @@ # but WITHOUT ANY WARRANTY; without even the implied warranty of # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the # GNU Affero General Public License for more details. +# +# You should have received a copy of the GNU Affero General Public License +# along with this program. If not, see . + +"""Service for connecting to GRPC channel.""" import grpc from deep import logging from deep.api.auth import AuthProvider +from deep.config import ConfigService from deep.utils import str2bool class GRPCService: - """ - This service handles config and initialising the GRPc channel that will be used - """ + """This service handles config and initialising the GRPc channel that will be used.""" + + def __init__(self, config: ConfigService): + """ + Create a new grpc service. - def __init__(self, config): + :param config: the deep config + """ self.channel = None self._config = config self._service_url = config.SERVICE_URL @@ -30,6 +39,7 @@ def __init__(self, config): self._metadata = None def start(self): + """Start and connect the GRPC channel.""" if str2bool(self._secure): logging.info("Connecting securely") logging.debug("Connecting securely to: %s", self._service_url) @@ -41,7 +51,10 @@ def start(self): def metadata(self): """ - Call this to get any metadata that should be attached to calls + Get GRPC metadata. + + Call this to get any metadata that should be attached to calls. + :return: list of metadata """ if self._metadata is None: diff --git a/src/deep/logging/__init__.py b/src/deep/logging/__init__.py index 79f9dc4..2e36656 100644 --- a/src/deep/logging/__init__.py +++ b/src/deep/logging/__init__.py @@ -9,6 +9,11 @@ # but WITHOUT ANY WARRANTY; without even the implied warranty of # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the # GNU Affero General Public License for more details. +# +# You should have received a copy of the GNU Affero General Public License +# along with this program. If not, see . + +"""Deep client logging api.""" import logging import logging.config @@ -16,25 +21,65 @@ def warning(msg, *args, **kwargs): + """ + Log a message at warning level. + + :param msg: the message to log + :param args: the args for the log + :param kwargs: the kwargs + """ logging.getLogger("deep").warning(msg, *args, **kwargs) def info(msg, *args, **kwargs): + """ + Log a message at info level. + + :param msg: the message to log + :param args: the args for the log + :param kwargs: the kwargs + """ logging.getLogger("deep").info(msg, *args, **kwargs) def debug(msg, *args, **kwargs): + """ + Log a message at debug level. + + :param msg: the message to log + :param args: the args for the log + :param kwargs: the kwargs + """ logging.getLogger("deep").debug(msg, *args, **kwargs) def error(msg, *args, **kwargs): + """ + Log a message at error level. + + :param msg: the message to log + :param args: the args for the log + :param kwargs: the kwargs + """ logging.getLogger("deep").debug(msg, *args, **kwargs) def exception(msg, *args, exc_info=True, **kwargs): + """ + Log a message with the exception data. + + :param msg: the message to log + :param args: the args for the log + :param kwargs: the kwargs + """ logging.getLogger("deep").exception(msg, *args, exc_info=exc_info, **kwargs) def init(cfg): + """ + Configure the deep log provider. + + :param cfg: the config for deep. + """ log_conf = cfg.LOGGING_CONF or "%s/logging.conf" % os.path.dirname(os.path.realpath(__file__)) logging.config.fileConfig(fname=log_conf, disable_existing_loggers=False) diff --git a/src/deep/logging/tracepoint_logger.py b/src/deep/logging/tracepoint_logger.py index c464203..f1fbc31 100644 --- a/src/deep/logging/tracepoint_logger.py +++ b/src/deep/logging/tracepoint_logger.py @@ -12,25 +12,43 @@ # # You should have received a copy of the GNU Affero General Public License # along with this program. If not, see . -import abc -from typing import TYPE_CHECKING -if TYPE_CHECKING: - from deep.config import ConfigService +"""Service for customizing the tracepoint logging.""" + +import abc from deep import logging class TracepointLogger(abc.ABC): + """ + This defines how a tracepoint logger should interact with Deep. + + This can be registered with Deep to provide customization to the way Deep will log dynamic log + messages injected via tracepoints. + """ @abc.abstractmethod def log_tracepoint(self, log_msg: str, tp_id: str, snap_id: str): + """ + Log the dynamic log message. + + :param (str) log_msg: the log message to log + :param (str) tp_id: the id of the tracepoint that generated this log + :param (str) snap_id: the is of the snapshot that was created by this tracepoint + """ pass class DefaultLogger(TracepointLogger): - def __init__(self, _config: 'ConfigService'): - self._config = _config + """The default tracepoint logger used by Deep.""" def log_tracepoint(self, log_msg: str, tp_id: str, snap_id: str): + """ + Log the dynamic log message. + + :param (str) log_msg: the log message to log + :param (str) tp_id: the id of the tracepoint that generated this log + :param (str) snap_id: the is of the snapshot that was created by this tracepoint + """ logging.info(log_msg + " snapshot=%s tracepoint=%s" % (snap_id, tp_id)) diff --git a/src/deep/poll/__init__.py b/src/deep/poll/__init__.py index 9152da6..9eca63e 100644 --- a/src/deep/poll/__init__.py +++ b/src/deep/poll/__init__.py @@ -9,6 +9,11 @@ # but WITHOUT ANY WARRANTY; without even the implied warranty of # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the # GNU Affero General Public License for more details. +# +# You should have received a copy of the GNU Affero General Public License +# along with this program. If not, see . + +"""Service for long poll.""" from deep.poll.poll import LongPoll diff --git a/src/deep/poll/poll.py b/src/deep/poll/poll.py index 7d82b62..3a14b55 100644 --- a/src/deep/poll/poll.py +++ b/src/deep/poll/poll.py @@ -9,6 +9,16 @@ # but WITHOUT ANY WARRANTY; without even the implied warranty of # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the # GNU Affero General Public License for more details. +# +# You should have received a copy of the GNU Affero General Public License +# along with this program. If not, see . + +""" +Long poll service for maintaining tracepoint config. + +Deep needs to maintain a config of the tracepoints configured by the users. The clients do this by + periodically polling the long poll service. +""" # noinspection PyUnresolvedReferences from deepproto.proto.poll.v1.poll_pb2 import PollRequest, ResponseType @@ -16,36 +26,41 @@ from deep import logging from deep.config import ConfigService -from deep.grpc import convert_resource, convert_response +from deep.grpc import convert_resource, convert_response, GRPCService from deep.utils import time_ns, RepeatedTimer class LongPoll(object): - """ - This service deals with polling the remote service to get the tracepoint configs - """ - config: ConfigService + """This service deals with polling the remote service to get the tracepoint configs.""" + + def __init__(self, config: ConfigService, grpc: GRPCService): + """ + Create a new long poll service. - def __init__(self, config, grpc): + :param config: the deep config service + :param grpc: the grpc service being used + """ self.config = config self.grpc = grpc self.timer = None def start(self): + """Start the long poll service.""" logging.info("Starting Long Poll system") if self.timer is not None: self.timer.stop() self.timer = RepeatedTimer("Tracepoint Long Poll", self.config.POLL_TIMER, self.poll) - self.initial_poll() + self.__initial_poll() self.timer.start() - def initial_poll(self): + def __initial_poll(self): try: self.poll() except Exception: logging.exception("Initial poll failed. Will continue with interval.") def poll(self): + """Check with the Deep servers for changes to the tracepoint config.""" stub = PollConfigStub(self.grpc.channel) request = PollRequest(ts_nanos=time_ns(), current_hash=self.config.tracepoints.current_hash, resource=convert_resource(self.config.resource)) diff --git a/src/deep/processor/__init__.py b/src/deep/processor/__init__.py index 3b59695..85505af 100644 --- a/src/deep/processor/__init__.py +++ b/src/deep/processor/__init__.py @@ -9,6 +9,11 @@ # but WITHOUT ANY WARRANTY; without even the implied warranty of # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the # GNU Affero General Public License for more details. +# +# You should have received a copy of the GNU Affero General Public License +# along with this program. If not, see . + +"""Handlers for processing tracepoint hits.""" from .trigger_handler import TriggerHandler diff --git a/src/deep/processor/bfs/__init__.py b/src/deep/processor/bfs/__init__.py index 580bd09..ed6edb0 100644 --- a/src/deep/processor/bfs/__init__.py +++ b/src/deep/processor/bfs/__init__.py @@ -9,6 +9,16 @@ # but WITHOUT ANY WARRANTY; without even the implied warranty of # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the # GNU Affero General Public License for more details. +# +# You should have received a copy of the GNU Affero General Public License +# along with this program. If not, see . + +""" +Breadth first search functions. + +To improve the performance and usefulness of the variable data gathered we use a Breadth First Search (BFS) +approach. This means scanning all local values before proceeding to the next depth on each. +""" import abc from typing import Callable, List @@ -19,58 +29,96 @@ class Node: """This is a Node that is used within the Breadth First Search of variables.""" - def __init__(self, value=None, children: List['Node'] = None, parent=None): + def __init__(self, value: 'NodeValue' = None, children: List['Node'] = None, parent: 'ParentNode' = None): + """ + Create a new node to process. + + :param (NodValue) value: the value to process + :param (list) children: the child nodes for this value + :param (ParentNode) parent: the parent node for this node + """ if children is None: children = [] self._value: 'NodeValue' = value - self._children: List['Node'] = children + self._children: list['Node'] = children self._parent: 'ParentNode' = parent self._depth = 0 @property def parent(self) -> 'ParentNode': + """Get the parent node.""" return self._parent @parent.setter def parent(self, parent: 'ParentNode'): + """Set the parent node.""" self._parent = parent def add_children(self, children: List['Node']): + """ + Add children to this node. + + :param (list) children: the children to add + """ for child in children: child._depth = self._depth + 1 self._children.append(child) @property def value(self) -> 'NodeValue': + """The node value.""" return self._value @property def depth(self): + """The node value.""" return self._depth @property def children(self) -> List['Node']: + """The node children.""" return self._children def __str__(self) -> str: + """Convert to string.""" return str(self.__dict__) def __repr__(self) -> str: + """Convert to string.""" return self.__str__() class ParentNode(abc.ABC): - """This represents the parent node - simple used to attach children to the parent if they are processed""" + """This represents the parent node - simple used to attach children to the parent if they are processed.""" @abc.abstractmethod def add_child(self, child: VariableId): + """ + Add a child to this parent. + + :param child: the child to add. + """ raise NotImplementedError class NodeValue: - """The variable value the node represents""" + """The variable value the node represents.""" def __init__(self, name: str, value: any, original_name=None): + """ + Create a new node value. + + It is possible to rename variables by providing an original name. This is used when dealing with + 'private' variables in calsses. + + e.g. A variable called _NodeValue__name is used by python to represent the private variable __name. This + is not known by devs, so we rename the variable to __name, and keep the original name as _NodeValue__name, + so we can show this if required. + + :param name: the name of the variable at this scope. + :param value: the value of the variable + :param original_name: the original name + """ self.name = name if original_name is not None and name != original_name: self.original_name = original_name @@ -79,16 +127,25 @@ def __init__(self, name: str, value: any, original_name=None): self.value = value def __str__(self) -> str: + """Parse the value into a string.""" return str(self.__dict__) def __repr__(self) -> str: + """Parse the value into a string.""" return self.__str__() def breadth_first_search(node: 'Node', consumer: Callable[['Node'], bool]): """ - To improve the performance and usefulness of the variable data gathered we use a Breadth First Search (BFS) - approach. This means scanning all local values before proceeding to the next depth on each. + Search for variables using BFS. + + Starting from the provided node, and using the consumer. Search for variables using BFS. + + We call consume, which will add all the child nodes to thr passed node. The return will then tell us to process + these or not. If we process them then we append the children to the queue. + + By using this queue approach we will process all the top level variables, then all of their children, and so + on until we are complete. :param node: the initial node to start the search :param consumer: the consumer to call on each node diff --git a/src/deep/processor/frame_collector.py b/src/deep/processor/frame_collector.py index 545b4ad..2b120ee 100644 --- a/src/deep/processor/frame_collector.py +++ b/src/deep/processor/frame_collector.py @@ -9,6 +9,11 @@ # but WITHOUT ANY WARRANTY; without even the implied warranty of # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the # GNU Affero General Public License for more details. +# +# You should have received a copy of the GNU Affero General Public License +# along with this program. If not, see . + +"""Processing for frame collection.""" import abc from typing import Dict, Tuple, List, Optional @@ -23,11 +28,15 @@ class FrameCollector(Collector): - """ - This deals with collecting data from the paused frames. - """ + """This deals with collecting data from the paused frames.""" def __init__(self, frame, config: ConfigService): + """ + Create a new collector. + + :param frame: the frame data + :param config: the deep config service + """ self._var_cache: Dict[str, str] = {} self._config = config self._has_time_exceeded = False @@ -39,25 +48,49 @@ def __init__(self, frame, config: ConfigService): @property def frame_config(self) -> FrameProcessorConfig: + """ + The frame config. + + :return: the frame config + """ return self._frame_config @abc.abstractmethod def configure_self(self): + """Process the filtered tracepoints to configure this processor.""" pass def add_child_to_lookup(self, parent_id: str, child: VariableId): + """ + Add a child variable to the var lookup parent. + + :param parent_id: the internal id of the parent + :param child: the child VariableId to append + :return: + """ self._var_lookup[parent_id].children.append(child) def log_tracepoint(self, log_msg: str, tp_id: str, snap_id: str): + """Send the processed log to the log handler.""" self._config.log_tracepoint(log_msg, tp_id, snap_id) def process_log(self, tp, log_msg) -> Tuple[str, List[WatchResult], Dict[str, Variable]]: + """ + Process a log message. + + :param tp: the tracepoint config + :param log_msg: the log message + :returns: + (str) log_msg: the processed log message + (list) watches: the watch results from the log + (dict) vars: the collected vars for the watches + """ frame_col = self watch_results = [] _var_lookup = {} class FormatDict(dict): - """This type is used in the log process to ensure that missing values are formatted don't error""" + """This type is used in the log process to ensure that missing values are formatted don't error.""" def __missing__(self, key): return "{%s}" % key @@ -65,8 +98,12 @@ def __missing__(self, key): import string class FormatExtractor(string.Formatter): - """This type allows us to use watches within log strings and collect the watch - as well as interpolate the values""" + """ + Allows logs to be formatted correctly. + + This type allows us to use watches within log strings and collect the watch + as well as interpolate the values. + """ def get_field(self, field_name, args, kwargs): # evaluate watch @@ -83,6 +120,7 @@ def get_field(self, field_name, args, kwargs): def eval_watch(self, watch: str) -> Tuple[WatchResult, Dict[str, Variable], str]: """ Evaluate an expression in the current frame. + :param watch: The watch expression to evaluate. :return: Tuple with WatchResult, collected variables, and the log string for the expression """ @@ -91,7 +129,7 @@ def eval_watch(self, watch: str) -> Tuple[WatchResult, Dict[str, Variable], str] try: result = eval(watch, None, self._frame.f_locals) - watch_var, var_lookup, log_str = self.process_watch_result_breadth_first(watch, result) + watch_var, var_lookup, log_str = self.__process_watch_result_breadth_first(watch, result) # again we reset the local version of the var lookup. self._var_lookup = {} return WatchResult(watch, watch_var), var_lookup, log_str @@ -101,7 +139,8 @@ def eval_watch(self, watch: str) -> Tuple[WatchResult, Dict[str, Variable], str] def process_frame(self): """ - This is the main variable processing + Start processing the frame. + :return: Tuple of collected frames and variables """ current_frame = self._frame @@ -132,13 +171,13 @@ def _process_frame(self, frame, process_vars): var_ids = [] # only process vars if we are under the time limit - if process_vars and not self.time_exceeded(): + if process_vars and not self.__time_exceeded(): var_ids = self.process_frame_variables_breadth_first(f_locals) short_path, app_frame = self.parse_short_name(filename) return StackFrame(filename, short_path, func_name, lineno, var_ids, class_name, app_frame=app_frame) - def time_exceeded(self): + def __time_exceeded(self): if self._has_time_exceeded: return self._has_time_exceeded @@ -146,7 +185,7 @@ def time_exceeded(self): self._has_time_exceeded = duration > self._frame_config.max_tp_process_time return self._has_time_exceeded - def is_app_frame(self, filename: str) -> Tuple[bool, Optional[str]]: + def __is_app_frame(self, filename: str) -> Tuple[bool, Optional[str]]: in_app_include = self._config.IN_APP_INCLUDE in_app_exclude = self._config.IN_APP_EXCLUDE @@ -165,7 +204,8 @@ def is_app_frame(self, filename: str) -> Tuple[bool, Optional[str]]: def process_frame_variables_breadth_first(self, f_locals): """ - Here we start the BFS process for the frame. + Process the variables on a frame. + :param f_locals: the frame locals. :return: the list of var ids for the frame. """ @@ -185,7 +225,8 @@ def add_child(self, child): def search_function(self, node: Node) -> bool: """ - This is the search function to use during BFS + Process a node using breadth first approach. + :param node: the current node we are process :return: True, if we want to continue with the nodes children """ @@ -212,11 +253,16 @@ def search_function(self, node: Node) -> bool: return True def check_var_count(self): + """ + Check if we have exceeded our var count. + + :return: True, if we should continue. + """ if len(self._var_cache) > self._frame_config.max_variables: return False return True - def process_watch_result_breadth_first(self, watch: str, result: any) -> ( + def __process_watch_result_breadth_first(self, watch: str, result: any) -> ( Tuple)[VariableId, Dict[str, Variable], str]: identity_hash_id = str(id(result)) @@ -249,21 +295,53 @@ def add_child(self, child): return VariableId(var_id, watch), self._var_lookup, str(result) def check_id(self, identity_hash_id): + """ + Check if the identity_hash_id is known to us, and return the lookup id. + + :param identity_hash_id: the id of the object + :return: the lookup id used + """ if identity_hash_id in self._var_cache: return self._var_cache[identity_hash_id] return None def new_var_id(self, identity_hash_id: str) -> str: + """ + Create a new cache id for the lookup. + + :param identity_hash_id: the id of the object + :return: the new lookup id + """ var_count = len(self._var_cache) new_id = str(var_count + 1) self._var_cache[identity_hash_id] = new_id return new_id def append_variable(self, var_id, variable): + """ + Append a variable to var lookup using the var id. + + :param var_id: the internal variable id + :param variable: the variable data to append + """ self._var_lookup[var_id] = variable - def parse_short_name(self, filename) -> Tuple[str, bool]: - is_app_frame, match = self.is_app_frame(filename) + def parse_short_name(self, filename: str) -> Tuple[str, bool]: + """ + Process a file name into a shorter version. + + By default, the file names in python are the absolute path to the file on disk. These can be quite long, + so we try to shorten the names by looking at the APP_ROOT and converting the file name into a relative path. + + e.g. if the file name is '/dev/python/custom_service/api/handler.py' and the APP_ROOT is + '/dev/python/custom_service' then we shorten the path to 'custom_service/api/handler.py'. + + :param (str) filename: the file name + :returns: + (str) filename: the new file name + (bool) is_app_frame: True if the file is an application frame file + """ + is_app_frame, match = self.__is_app_frame(filename) if match is not None: return filename[len(match):], is_app_frame return filename, is_app_frame diff --git a/src/deep/processor/frame_config.py b/src/deep/processor/frame_config.py index 3ffe7e6..fb8cd7f 100644 --- a/src/deep/processor/frame_config.py +++ b/src/deep/processor/frame_config.py @@ -9,6 +9,11 @@ # but WITHOUT ANY WARRANTY; without even the implied warranty of # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the # GNU Affero General Public License for more details. +# +# You should have received a copy of the GNU Affero General Public License +# along with this program. If not, see . + +"""Configuration options for tracepoint processing.""" from deep.api.tracepoint.tracepoint_config import SINGLE_FRAME_TYPE, STACK, \ frame_type_ordinal, STACK_TYPE, FRAME_TYPE, \ @@ -16,9 +21,8 @@ class FrameProcessorConfig: - """ - This is the config for a data collection. - """ + """This is the config for a data collection.""" + DEFAULT_MAX_VAR_DEPTH = 5 DEFAULT_MAX_VARIABLES = 1000 DEFAULT_MAX_COLLECTION_SIZE = 10 @@ -29,6 +33,7 @@ class FrameProcessorConfig: DEFAULT_PROFILE_INTERVAL = 10 def __init__(self): + """Create a new config.""" self._frame_type = None self._stack_type = None self._max_var_depth = -1 @@ -40,19 +45,22 @@ def __init__(self): def process_tracepoint(self, tp: TracePointConfig): """ + Process a tracepoint into this config. + Each tracepoint can have a different config we want to re-configure to the lowest impact. e.g. if all tracepoints are single frame, then do not collect all frames. :param tp: the tracepoint to process """ - self._max_var_depth = FrameProcessorConfig.get_max_or_default(tp.args, 'MAX_VAR_DEPTH', self._max_var_depth) - self._max_variables = FrameProcessorConfig.get_max_or_default(tp.args, 'MAX_VARIABLES', self._max_variables) - self._max_collection_size = FrameProcessorConfig.get_max_or_default(tp.args, 'MAX_COLLECTION_SIZE', - self._max_collection_size) - self._max_string_length = FrameProcessorConfig.get_max_or_default(tp.args, 'MAX_STRING_LENGTH', - self._max_string_length) - self._max_watch_vars = FrameProcessorConfig.get_max_or_default(tp.args, 'MAX_WATCH_VARS', self._max_watch_vars) - self._max_tp_process_time = FrameProcessorConfig.get_max_or_default(tp.args, 'MAX_TP_PROCESS_TIME', - self._max_tp_process_time) + self._max_var_depth = FrameProcessorConfig.__get_max_or_default(tp.args, 'MAX_VAR_DEPTH', self._max_var_depth) + self._max_variables = FrameProcessorConfig.__get_max_or_default(tp.args, 'MAX_VARIABLES', self._max_variables) + self._max_collection_size = FrameProcessorConfig.__get_max_or_default(tp.args, 'MAX_COLLECTION_SIZE', + self._max_collection_size) + self._max_string_length = FrameProcessorConfig.__get_max_or_default(tp.args, 'MAX_STRING_LENGTH', + self._max_string_length) + self._max_watch_vars = FrameProcessorConfig.__get_max_or_default(tp.args, 'MAX_WATCH_VARS', + self._max_watch_vars) + self._max_tp_process_time = FrameProcessorConfig.__get_max_or_default(tp.args, 'MAX_TP_PROCESS_TIME', + self._max_tp_process_time) # use the highest collection type - results can be trimmed during pre upload processing frame_type = tp.get_arg(FRAME_TYPE, None) @@ -72,7 +80,6 @@ def process_tracepoint(self, tp: TracePointConfig): def close(self): """Close the config, to check for any unconfirmed parts, and set them to defaults.""" - # todo: What if one tp has 'MAX_VARS' as 10, but others do not have it set. self._max_var_depth = FrameProcessorConfig.DEFAULT_MAX_VAR_DEPTH if self._max_var_depth == -1 \ @@ -97,44 +104,100 @@ def close(self): self._stack_type = STACK @staticmethod - def get_max_or_default(config, key, default_value): + def __get_max_or_default(config, key, default_value): if key in config: return max(int(config[key]), default_value) return default_value @property - def frame_type(self): + def frame_type(self) -> str: + """ + Get the frame type. + + :return: the frame type + """ return self._frame_type @property - def stack_type(self): + def stack_type(self) -> str: + """ + Get the stack type. + + :return: the stack type + """ return self._stack_type @property - def max_var_depth(self): + def max_var_depth(self) -> int: + """ + Get the maximum depth of variables to process. + + Values deeper than this will be ignored. + + :return: the maximum variable depth + """ return self._max_var_depth @property - def max_variables(self): + def max_variables(self) -> int: + """ + Get the maximum number of variables to process. + + Any additional variables will not be processed or attached to the snapshots. + + :return: the maximum number of variables + """ return self._max_variables @property - def max_collection_size(self): + def max_collection_size(self) -> int: + """ + Get the maximum size of a collection. + + Collections larger than this should be truncated. + + :return: the maximum collection size + """ return self._max_collection_size @property - def max_string_length(self): + def max_string_length(self) -> int: + """ + Get the maximum length of a string. + + Strings longer than this value should be truncated. + + :return: the maximum string length + """ return self._max_string_length @property - def max_watch_vars(self): + def max_watch_vars(self) -> int: + """ + Get the maximum number of variables to collect for a watch. + + :return: the max variables + """ return self._max_watch_vars @property - def max_tp_process_time(self): + def max_tp_process_time(self) -> int: + """ + Get the maximum time we should spend processing a tracepoint. + + :return: the max time + """ return self._max_tp_process_time - def should_collect_vars(self, current_frame_index): + def should_collect_vars(self, current_frame_index: int) -> bool: + """ + Check if we can collect data for a frame. + + Frame indexes start from 0 (as the current frame) and increase as we go back up the stack. + + :param (int) current_frame_index: the current frame index. + :return (bool): if we should collect the frame vars. + """ if self._frame_type == NO_FRAME_TYPE: return False if current_frame_index == 0: diff --git a/src/deep/processor/frame_processor.py b/src/deep/processor/frame_processor.py index 32c0ab4..bdfaf1b 100644 --- a/src/deep/processor/frame_processor.py +++ b/src/deep/processor/frame_processor.py @@ -9,6 +9,19 @@ # but WITHOUT ANY WARRANTY; without even the implied warranty of # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the # GNU Affero General Public License for more details. +# +# You should have received a copy of the GNU Affero General Public License +# along with this program. If not, see . + +""" +Handle Frame data processing. + +When processing a frame we need to ensure that the matched tracepoints can fire and that we collect +the appropriate information. We need to process the conditions and fire rates of the tracepoints, and check the +configs to collect the smallest amount of data possible. +""" + +from types import FrameType from typing import List from deep import logging @@ -20,19 +33,26 @@ class FrameProcessor(FrameCollector): - """ - This handles a 'hit' and starts the process of collecting the data. - """ + """This handles a 'hit' and starts the process of collecting the data.""" + _filtered_tracepoints: List[TracePointConfig] - def __init__(self, tracepoints: List[TracePointConfig], frame, config: ConfigService): + def __init__(self, tracepoints: List[TracePointConfig], frame: FrameType, config: ConfigService): + """ + Create a new processor. + + :param tracepoints: the tracepoints for the triggering event + :param frame: the frame data + :param config: the deep config service + """ super().__init__(frame, config) self._tracepoints = tracepoints self._filtered_tracepoints = [] - def collect(self): + def collect(self) -> List[EventSnapshot]: """ - Here we start the data collection process + Collect the snapshot data for the available tracepoints. + :return: list of completed snapshots """ snapshots = [] @@ -69,7 +89,10 @@ def collect(self): def can_collect(self): """ + Check if we can collect data. + Check if the tracepoints can fire given their configs. Checking time windows, fire rates etc. + :return: True, if any tracepoint can fire """ for tp in self._tracepoints: @@ -79,7 +102,13 @@ def can_collect(self): return len(self._filtered_tracepoints) > 0 - def condition_passes(self, tp): + def condition_passes(self, tp: TracePointConfig) -> bool: + """ + Check if the tracepoint condition passes. + + :param (TracePointConfig) tp: the tracepoint to check + :return: True, if the condition passes + """ condition = tp.condition if condition is None or condition == "": # There is no condition so return True @@ -96,15 +125,18 @@ def condition_passes(self, tp): return False def configure_self(self): - """ - Using the filtered tracepoints, re-configure the frame config for minimum collection - :return: - """ + """Process the filtered tracepoints to configure this processor.""" for tp in self._filtered_tracepoints: self._frame_config.process_tracepoint(tp) self._frame_config.close() - def process_attributes(self, tp): + def process_attributes(self, tp: TracePointConfig) -> BoundedAttributes: + """ + Process the attributes for a tracepoint. + + :param (TracePointConfig) tp: the tracepoint to process. + :return (BoundedAttributes): the attributes for the tracepoint + """ attributes = { "tracepoint": tp.id, "path": tp.path, diff --git a/src/deep/processor/trigger_handler.py b/src/deep/processor/trigger_handler.py index 63ac4c6..bd8fa06 100644 --- a/src/deep/processor/trigger_handler.py +++ b/src/deep/processor/trigger_handler.py @@ -9,6 +9,18 @@ # but WITHOUT ANY WARRANTY; without even the implied warranty of # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the # GNU Affero General Public License for more details. +# +# You should have received a copy of the GNU Affero General Public License +# along with this program. If not, see . + +""" +Handle events from the python engine to trigger tracepoints. + +Using the `sys.settrace` and `threading.settrace` functions we register a function to +process events from the python engine. When we get an event we are interested in (e.g. line) we can match that +to a tracepoint config and if we have a match process the frame data to collect a snapshot, add logs or any +other supported action. +""" import logging import os @@ -21,27 +33,39 @@ from deep.push import PushService -def add_or_get(target, key, default_value): - if key not in target: - target[key] = default_value - return target[key] - - class TracepointHandlerUpdateListener(ConfigUpdateListener): - """ - This is the listener that connects the config to the handler - """ + """This is the listener that connects the config to the handler.""" def __init__(self, handler): + """ + Create a new update listener. + + :param handler: the handler to call when a new tracepoint config is ready + """ self._handler = handler + @staticmethod + def __add_or_get(target, key, default_value): + if key not in target: + target[key] = default_value + return target[key] + def config_change(self, ts, old_hash, current_hash, old_config, new_config): + """ + Process an update to the tracepoint config. + + :param ts: the ts of the new config + :param old_hash: the old config hash + :param current_hash: the new config hash + :param old_config: the old config + :param new_config: the new config + """ sorted_config = {} for tracepoint in new_config: path = os.path.basename(tracepoint.path) line_no = tracepoint.line_no - by_file = add_or_get(sorted_config, path, {}) - by_line = add_or_get(by_file, line_no, []) + by_file = self.__add_or_get(sorted_config, path, {}) + by_line = self.__add_or_get(by_file, line_no, []) by_line.append(tracepoint) self._handler.new_config(sorted_config) @@ -49,17 +73,25 @@ def config_change(self, ts, old_hash, current_hash, old_config, new_config): class TriggerHandler: """ - This is the handler for the tracepoints. This is where we 'listen' for a hit, and determine if we - should collect data. + This is the handler for the tracepoints. + + This is where we 'listen' for a hit, and determine if we should collect data. """ def __init__(self, config: ConfigService, push_service: PushService): + """ + Create a new tigger handler. + + :param config: the config service + :param push_service: the push service + """ self._push_service = push_service self._tp_config = [] self._config = config self._config.add_listener(TracepointHandlerUpdateListener(self)) def start(self): + """Start the trigger handler.""" # if we call settrace we cannot use debugger, # so we allow the settrace to be disabled, so we can at least debug around it if self._config.NO_TRACE: @@ -68,23 +100,32 @@ def start(self): threading.settrace(self.trace_call) def new_config(self, new_config): + """ + Process a new tracepoint config. + + Called when a change to the tracepoint config is processed. + + :param new_config: the new config to use + """ self._tp_config = new_config def trace_call(self, frame, event, arg): """ - This is called by python with the current frame data + Process the data for a trace call. + + This is called by the python engine when an event is about to be called. + :param frame: the current frame :param event: the event 'line', 'call', etc. That we are processing. :param arg: the args :return: None to ignore other calls, or our self to continue """ - # return if we do not have any tracepoints if len(self._tp_config) == 0: return None - tracepoints_for_file, tracepoints_for_line = self.tracepoints_for(os.path.basename(frame.f_code.co_filename), - frame.f_lineno) + tracepoints_for_file, tracepoints_for_line = self.__tracepoints_for(os.path.basename(frame.f_code.co_filename), + frame.f_lineno) # return if this is not a 'line' event if event != 'line': @@ -96,7 +137,7 @@ def trace_call(self, frame, event, arg): self.process_tracepoints(tracepoints_for_line, frame) return self.trace_call - def tracepoints_for(self, filename, lineno): + def __tracepoints_for(self, filename, lineno): if filename in self._tp_config: filename_ = self._tp_config[filename] if lineno in filename_: @@ -106,7 +147,7 @@ def tracepoints_for(self, filename, lineno): def process_tracepoints(self, tracepoints_for, frame): """ - We have some tracepoints, now check if we can collect + We have some tracepoints, now check if we can collect. :param tracepoints_for: tracepoints for the file/line :param frame: the frame data diff --git a/src/deep/processor/variable_processor.py b/src/deep/processor/variable_processor.py index c143a65..e6ab7c2 100644 --- a/src/deep/processor/variable_processor.py +++ b/src/deep/processor/variable_processor.py @@ -9,6 +9,17 @@ # but WITHOUT ANY WARRANTY; without even the implied warranty of # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the # GNU Affero General Public License for more details. +# +# You should have received a copy of the GNU Affero General Public License +# along with this program. If not, see . + +""" +A set of functions to collect and process variable data. + +There are many things to consider when collecting that data from a variable. Here we try to manage the collection +as best we can without affecting the original data source. As a result we have different ways to collect the data +and many options to consider when collecting. +""" import abc from typing import List @@ -28,6 +39,7 @@ 'unicode', 'long', ] +"""A list of types that do not have child nodes, or only have child nodes we do not want to process.""" LIST_LIKE_TYPES = [ 'frozenset', @@ -35,6 +47,7 @@ 'list', 'tuple', ] +"""A list of types that we should handle like lists.""" ITER_LIKE_TYPES = [ 'list_iterator', @@ -42,25 +55,41 @@ 'list_reverseiterator', 'listreverseiterator', ] +"""A list of types that we should handle like iterators.""" +# We cannot process child nodes of iterators so add the iterator types to the no child types. NO_CHILD_TYPES += ITER_LIKE_TYPES class Collector(abc.ABC): + """A type that is used to manage variable collection.""" @property @abc.abstractmethod def frame_config(self) -> FrameProcessorConfig: + """ + The frame config. + + :return: the frame config + """ pass @abc.abstractmethod - def add_child_to_lookup(self, variable_id, child): + def add_child_to_lookup(self, parent_id: str, child: VariableId): + """ + Add a child variable to the var lookup parent. + + :param parent_id: the internal id of the parent + :param child: the child VariableId to append + :return: + """ pass @abc.abstractmethod def check_id(self, identity_hash_id: str) -> str: """ - Check if the identity_hash_id is known to us, and return the lookup id + Check if the identity_hash_id is known to us, and return the lookup id. + :param identity_hash_id: the id of the object :return: the lookup id used """ @@ -69,33 +98,47 @@ def check_id(self, identity_hash_id: str) -> str: @abc.abstractmethod def new_var_id(self, identity_hash_id: str) -> str: """ - Create a new cache id for the lookup + Create a new cache id for the lookup. + :param identity_hash_id: the id of the object :return: the new lookup id """ pass @abc.abstractmethod - def append_variable(self, var_id, variable): + def append_variable(self, var_id: str, variable: Variable): + """ + Append a variable to var lookup using the var id. + + :param var_id: the internal variable id + :param variable: the variable data to append + """ pass class VariableResponse: + """The response from processing a variable.""" + def __init__(self, variable_id, process_children=True): + """Create a new response object.""" self.__variable_id = variable_id self.__process_children = process_children @property def variable_id(self): + """The variable id data for the processed variable.""" return self.__variable_id @property def process_children(self): + """Can we process the children of the value.""" return self.__process_children def var_modifiers(var_name: str) -> List[str]: """ + Process access modifiers. + Python does not have true access modifiers. The convention is to use leading underscores, one for protected, two for private. @@ -113,7 +156,8 @@ def var_modifiers(var_name: str) -> List[str]: def variable_to_string(variable_type, var_value): """ - Convert the variable to a string + Convert the variable to a string. + :param variable_type: the variable type :param var_value: the variable value :return: a string of the value @@ -134,11 +178,11 @@ def variable_to_string(variable_type, var_value): def process_variable(frame_collector: Collector, node: NodeValue) -> VariableResponse: """ Process the variable into a serializable type. + :param frame_collector: the collector being used :param node: the variable node to process :return: a response to determine if we continue """ - # get the variable hash id identity_hash_id = str(id(node.value)) # guess the modifiers @@ -173,7 +217,8 @@ def process_variable(frame_collector: Collector, node: NodeValue) -> VariableRes def truncate_string(string, max_length): """ - Truncate the incoming string to the specified length + Truncate the incoming string to the specified length. + :param string: the string to truncate :param max_length: the length to truncated to :return: a tuple of the new string, and if it was truncated @@ -188,8 +233,9 @@ def process_child_nodes( frame_depth: int ) -> List[Node]: """ - Processing the children how we get the list of new variables to process. The method changes depending on - the type we are processing. + Collect the child nodes for this variable. + + Child node collection is performed via a variety of functions based on the type of the variable we are processing. :param frame_collector: the collector we are using :param variable_id: the variable if to attach children to @@ -218,7 +264,8 @@ def add_child(self, child: VariableId): def correct_names(name, val): """ - If a value is 'private' then python will rename the value to be prefixed with the class name + If a value is 'private' then python will rename the value to be prefixed with the class name. + :param name: the name of the class :param val: the variable name we are modifying :return: the new name to use @@ -232,7 +279,8 @@ def correct_names(name, val): def find_children_for_parent(frame_collector: Collector, parent_node: ParentNode, value: any, variable_type: type): """ - Scan the parent for children based on the type + Scan the parent for children based on the type. + :param frame_collector: the collector we are using :param parent_node: the parent node :param value: the variable value we are processing @@ -254,14 +302,38 @@ def find_children_for_parent(frame_collector: Collector, parent_node: ParentNode return [] -def process_dict_breadth_first(parent_node, type_name, value, func=lambda x, y: y): +def process_dict_breadth_first(parent_node, type_name, value, func=lambda x, y: y) -> List[Node]: + """ + Process a dict value. + + Take a dict and collect all the child nodes for the dict. + + :param (ParentNode) parent_node: the node that represents the list, to be used as the parent for the returned nodes + :param (str) type_name: the name of the type we are processing + :param (any) value: the list value to process + :param (Callable) func: an optional function to preprocess values + + :param func: + :return (list): the collected child nodes + """ # we wrap the keys() in a call to list to prevent concurrent changes return [Node(value=NodeValue(func(type_name, key), value[key], key), parent=parent_node) for key in list(value.keys()) if key in value] -def process_list_breadth_first(frame_collector: Collector, parent_node: ParentNode, value): +def process_list_breadth_first(frame_collector: Collector, parent_node: ParentNode, value) -> List[Node]: + """ + Process a list value. + + Take a list and collect all the child nodes for the list. Returned list is + limited by the config 'max_collection_size'. + + :param (Collector) frame_collector: the collector that is managing this collection + :param (ParentNode) parent_node: the node that represents the list, to be used as the parent for the returned nodes + :param (any) value: the list value to process + :return (list): the collected child nodes + """ nodes = [] total = 0 for val_ in tuple(value): diff --git a/src/deep/push/__init__.py b/src/deep/push/__init__.py index e913d6a..fc873dd 100644 --- a/src/deep/push/__init__.py +++ b/src/deep/push/__init__.py @@ -9,6 +9,17 @@ # but WITHOUT ANY WARRANTY; without even the implied warranty of # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the # GNU Affero General Public License for more details. +# +# You should have received a copy of the GNU Affero General Public License +# along with this program. If not, see . + +""" +Collection of functions to convert to protobuf version of types. + +We do not use the protobuf types throughout the project as they do not autocomplete or +have type definitions that work in IDE. It also makes it easier to deal with agent functionality by +having local types we can modify. +""" import logging @@ -27,16 +38,16 @@ from ..grpc import convert_value -def convert_tracepoint(tracepoint: TrPoCo): +def __convert_tracepoint(tracepoint: TrPoCo): return TracePointConfig(ID=tracepoint.id, path=tracepoint.path, line_number=tracepoint.line_no, args=tracepoint.args, watches=tracepoint.watches) -def convert_frame(frame: StFr): +def __convert_frame(frame: StFr): return StackFrame(file_name=frame.file_name, short_path=frame.short_path, method_name=frame.method_name, line_number=frame.line_number, class_name=frame.class_name, is_async=frame.is_async, - column_number=frame.column_number, variables=[convert_variable_id(v) for v in frame.variables], + column_number=frame.column_number, variables=[__convert_variable_id(v) for v in frame.variables], app_frame=frame.app_frame, transpiled_file_name=frame.transpiled_file_name, transpiled_line_number=frame.transpiled_line_number, @@ -44,36 +55,42 @@ def convert_frame(frame: StFr): ) -def convert_watch(watch: WaRe): - return WatchResult(expression=watch.expression, good_result=convert_variable_id(watch.result), +def __convert_watch(watch: WaRe): + return WatchResult(expression=watch.expression, good_result=__convert_variable_id(watch.result), error_result=watch.error) -def convert_variable(variable: Var): +def __convert_variable(variable: Var): return Variable(type=variable.type, value=variable.value, hash=variable.hash, - children=[convert_variable_id(c) for c in variable.children], truncated=variable.truncated) + children=[__convert_variable_id(c) for c in variable.children], truncated=variable.truncated) -def convert_variable_id(variable: VarId): +def __convert_variable_id(variable: VarId): if variable is None: return None return VariableID(ID=variable.vid, name=variable.name, modifiers=variable.modifiers, original_name=variable.original_name) -def convert_lookup(var_lookup): +def __convert_lookup(var_lookup): converted = {} for k, v in var_lookup.items(): - converted[k] = convert_variable(v) + converted[k] = __convert_variable(v) return converted def convert_snapshot(snapshot: EventSnapshot) -> Snapshot: + """ + Convert a snapshot from internal model to protobuf model. + + :param (EventSnapshot) snapshot: the internal snapshot model + :return (Snapshot): the protobuf model of the snapshot + """ try: - return Snapshot(ID=snapshot.id.to_bytes(16, "big"), tracepoint=convert_tracepoint(snapshot.tracepoint), - var_lookup=convert_lookup(snapshot.var_lookup), - ts_nanos=snapshot.ts_nanos, frames=[convert_frame(f) for f in snapshot.frames], - watches=[convert_watch(w) for w in snapshot.watches], + return Snapshot(ID=snapshot.id.to_bytes(16, "big"), tracepoint=__convert_tracepoint(snapshot.tracepoint), + var_lookup=__convert_lookup(snapshot.var_lookup), + ts_nanos=snapshot.ts_nanos, frames=[__convert_frame(f) for f in snapshot.frames], + watches=[__convert_watch(w) for w in snapshot.watches], attributes=[KeyValue(key=k, value=convert_value(v)) for k, v in snapshot.attributes.items()], duration_nanos=snapshot.duration_nanos, resource=[KeyValue(key=k, value=convert_value(v)) for k, v in diff --git a/src/deep/push/push_service.py b/src/deep/push/push_service.py index 2ddb9b5..3b4d8fe 100644 --- a/src/deep/push/push_service.py +++ b/src/deep/push/push_service.py @@ -9,6 +9,11 @@ # but WITHOUT ANY WARRANTY; without even the implied warranty of # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the # GNU Affero General Public License for more details. +# +# You should have received a copy of the GNU Affero General Public License +# along with this program. If not, see . + +"""Provide service for pushing events to Deep services.""" from deepproto.proto.tracepoint.v1.tracepoint_pb2_grpc import SnapshotServiceStub @@ -18,17 +23,23 @@ class PushService: - """ - This service deals with pushing the snapshots to the service endpoints - """ + """This service deals with pushing the snapshots to the service endpoints.""" def __init__(self, config, grpc, task_handler): + """ + Create a service to handle push events. + + :param config: the current deep config + :param grpc: the grpc service to use to send events + :param task_handler: the task handler to offload tasks to + """ self.config = config self.grpc = grpc self.task_handler = task_handler def push_snapshot(self, snapshot: EventSnapshot): - self.decorate(snapshot) + """Push a snapshot to the deep services.""" + self.__decorate(snapshot) task = self.task_handler.submit_task(self._push_task, snapshot) task.add_done_callback( lambda _: logging.debug("Completed uploading snapshot %s", snapshot_id_as_hex_str(snapshot.id))) @@ -42,7 +53,7 @@ def _push_task(self, snapshot): stub.send(converted, metadata=self.grpc.metadata()) - def decorate(self, snapshot): + def __decorate(self, snapshot): plugins = self.config.plugins for plugin in plugins: try: diff --git a/src/deep/task/__init__.py b/src/deep/task/__init__.py index d724994..fcbdd4e 100644 --- a/src/deep/task/__init__.py +++ b/src/deep/task/__init__.py @@ -9,6 +9,11 @@ # but WITHOUT ANY WARRANTY; without even the implied warranty of # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the # GNU Affero General Public License for more details. +# +# You should have received a copy of the GNU Affero General Public License +# along with this program. If not, see . + +"""Provides processing options for tasks on background threads.""" import logging import threading @@ -17,11 +22,16 @@ class IllegalStateException(BaseException): + """This is raised when we are in an incompatible state.""" + pass class TaskHandler: + """Allow processing of tasks without blocking the current thread.""" + def __init__(self): + """Create a new TaskHandler to process tasks in a separate thread.""" self._pool = ThreadPoolExecutor(max_workers=2) self._pending = {} self._job_id = 0 @@ -34,12 +44,19 @@ def _next_id(self): next_id = self._job_id return next_id - def check_open(self): + def __check_open(self): if not self._open: raise IllegalStateException def submit_task(self, task, *args) -> Future: - self.check_open() + """ + Submit a task to be processed in the task thread. + + :param task: the task function to process + :param args: the args to pass to the function + :return: a future that can be listened to for completion + """ + self.__check_open() next_id = self._next_id() # there is an at exit in threading that prevents submitting tasks after shutdown, but no api to check this future = self._pool.submit(task, *args) @@ -56,6 +73,7 @@ def callback(future: Future): return future def flush(self): + """Await completion of all pending tasks.""" self._open = False if len(self._pending) > 0: for key in dict(self._pending).keys(): diff --git a/src/deep/utils.py b/src/deep/utils.py index 1c9eb29..f6fe38a 100644 --- a/src/deep/utils.py +++ b/src/deep/utils.py @@ -9,6 +9,11 @@ # but WITHOUT ANY WARRANTY; without even the implied warranty of # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the # GNU Affero General Public License for more details. +# +# You should have received a copy of the GNU Affero General Public License +# along with this program. If not, see . + +"""A collection of util functions to perform common or repeated actions.""" import logging import time @@ -16,7 +21,7 @@ def snapshot_id_as_hex_str(snapshot_id): - """"Convert a snapshot if to a hex string.""" + """Convert a snapshot if to a hex string.""" return snapshot_id.to_bytes(16, 'big').hex() @@ -50,7 +55,7 @@ def reduce_list(key, update_value, default_value, lst): def str2bool(string): """ - Convert a string to a boolean + Convert a string to a boolean. :param string: the string to convert :return: True, if string is yes, true, t or 1. (case insensitive) @@ -62,6 +67,15 @@ class RepeatedTimer: """Repeat `function` every `interval` seconds.""" def __init__(self, name, interval, function, *args, **kwargs): + """ + Create a new RepeatTimer. + + :param name: the name of the timer + :param interval: the time in seconds between each execution + :param function: the function to repeat + :param args: the arguments for the function + :param kwargs: the kwargs for the function + """ self.name = name self.interval = interval self.function = function @@ -73,9 +87,11 @@ def __init__(self, name, interval, function, *args, **kwargs): self.thread.daemon = True def start(self): + """Start the thread to run the timer.""" self.thread.start() def stop(self): + """Stop and shutdown the timer.""" self.event.set() self.thread.join() diff --git a/src/deep/version.py b/src/deep/version.py index 5c36829..45b0adb 100644 --- a/src/deep/version.py +++ b/src/deep/version.py @@ -9,6 +9,11 @@ # but WITHOUT ANY WARRANTY; without even the implied warranty of # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the # GNU Affero General Public License for more details. +# +# You should have received a copy of the GNU Affero General Public License +# along with this program. If not, see . + +"""Version information about deep.""" __version__ = "0.0.13" # this version is set by the build, but not updated in the code. """The version of the agent that is running.""" diff --git a/test/test_deep/__init__.py b/test/test_deep/__init__.py index a22412a..53e9b3b 100644 --- a/test/test_deep/__init__.py +++ b/test/test_deep/__init__.py @@ -9,3 +9,6 @@ # but WITHOUT ANY WARRANTY; without even the implied warranty of # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the # GNU Affero General Public License for more details. +# +# You should have received a copy of the GNU Affero General Public License +# along with this program. If not, see . diff --git a/test/test_deep/auth/__init__.py b/test/test_deep/auth/__init__.py index a22412a..53e9b3b 100644 --- a/test/test_deep/auth/__init__.py +++ b/test/test_deep/auth/__init__.py @@ -9,3 +9,6 @@ # but WITHOUT ANY WARRANTY; without even the implied warranty of # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the # GNU Affero General Public License for more details. +# +# You should have received a copy of the GNU Affero General Public License +# along with this program. If not, see . diff --git a/test/test_deep/auth/test_auth.py b/test/test_deep/auth/test_auth.py index 85a6311..b0e33d1 100644 --- a/test/test_deep/auth/test_auth.py +++ b/test/test_deep/auth/test_auth.py @@ -9,6 +9,10 @@ # but WITHOUT ANY WARRANTY; without even the implied warranty of # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the # GNU Affero General Public License for more details. +# +# You should have received a copy of the GNU Affero General Public License +# along with this program. If not, see . + import unittest from deep.api.auth import AuthProvider diff --git a/test/test_deep/config/__init__.py b/test/test_deep/config/__init__.py index a22412a..53e9b3b 100644 --- a/test/test_deep/config/__init__.py +++ b/test/test_deep/config/__init__.py @@ -9,3 +9,6 @@ # but WITHOUT ANY WARRANTY; without even the implied warranty of # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the # GNU Affero General Public License for more details. +# +# You should have received a copy of the GNU Affero General Public License +# along with this program. If not, see . diff --git a/test/test_deep/config/test_config.py b/test/test_deep/config/test_config.py index cc9cfb4..f235ca0 100644 --- a/test/test_deep/config/test_config.py +++ b/test/test_deep/config/test_config.py @@ -9,6 +9,9 @@ # but WITHOUT ANY WARRANTY; without even the implied warranty of # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the # GNU Affero General Public License for more details. +# +# You should have received a copy of the GNU Affero General Public License +# along with this program. If not, see . import sys import unittest diff --git a/test/test_deep/config/test_config_service.py b/test/test_deep/config/test_config_service.py index 3d7ba48..85d6f03 100644 --- a/test/test_deep/config/test_config_service.py +++ b/test/test_deep/config/test_config_service.py @@ -9,6 +9,10 @@ # but WITHOUT ANY WARRANTY; without even the implied warranty of # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the # GNU Affero General Public License for more details. +# +# You should have received a copy of the GNU Affero General Public License +# along with this program. If not, see . + import os import unittest diff --git a/test/test_deep/grpc/__init__.py b/test/test_deep/grpc/__init__.py index a22412a..53e9b3b 100644 --- a/test/test_deep/grpc/__init__.py +++ b/test/test_deep/grpc/__init__.py @@ -9,3 +9,6 @@ # but WITHOUT ANY WARRANTY; without even the implied warranty of # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the # GNU Affero General Public License for more details. +# +# You should have received a copy of the GNU Affero General Public License +# along with this program. If not, see . diff --git a/test/test_deep/grpc/test_grpc.py b/test/test_deep/grpc/test_grpc.py index 7ac8e4a..efeebd2 100644 --- a/test/test_deep/grpc/test_grpc.py +++ b/test/test_deep/grpc/test_grpc.py @@ -9,6 +9,9 @@ # but WITHOUT ANY WARRANTY; without even the implied warranty of # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the # GNU Affero General Public License for more details. +# +# You should have received a copy of the GNU Affero General Public License +# along with this program. If not, see . import unittest diff --git a/test/test_deep/processor/__init__.py b/test/test_deep/processor/__init__.py index 80f521d..b6573b1 100644 --- a/test/test_deep/processor/__init__.py +++ b/test/test_deep/processor/__init__.py @@ -9,8 +9,13 @@ # but WITHOUT ANY WARRANTY; without even the implied warranty of # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the # GNU Affero General Public License for more details. +# +# You should have received a copy of the GNU Affero General Public License +# along with this program. If not, see . class MockFrame: + """A Frame used during testing to Mock a debug Frame.""" + def __init__(self, _locals=None): if _locals is None: _locals = {} diff --git a/test/test_deep/processor/test_variable_processor.py b/test/test_deep/processor/test_variable_processor.py index 1b9fd7e..215b0d9 100644 --- a/test/test_deep/processor/test_variable_processor.py +++ b/test/test_deep/processor/test_variable_processor.py @@ -9,6 +9,10 @@ # but WITHOUT ANY WARRANTY; without even the implied warranty of # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the # GNU Affero General Public License for more details. +# +# You should have received a copy of the GNU Affero General Public License +# along with this program. If not, see . + import unittest from parameterized import parameterized @@ -21,9 +25,7 @@ class MockVariable(Variable): - """ - We do not want to test the hash as this is the memory address so hard to verify in the tests - """ + """We do not want to test the hash as this is the memory address so hard to verify in the tests.""" def __eq__(self, o: object) -> bool: diff --git a/test/test_deep/tracepoint/__init__.py b/test/test_deep/tracepoint/__init__.py index a22412a..962577d 100644 --- a/test/test_deep/tracepoint/__init__.py +++ b/test/test_deep/tracepoint/__init__.py @@ -1,4 +1,4 @@ -# Copyright (C) 2023 Intergral GmbH +# Copyright (C) 2024 Intergral GmbH # # This program is free software: you can redistribute it and/or modify # it under the terms of the GNU Affero General Public License as published by @@ -9,3 +9,6 @@ # but WITHOUT ANY WARRANTY; without even the implied warranty of # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the # GNU Affero General Public License for more details. +# +# You should have received a copy of the GNU Affero General Public License +# along with this program. If not, see . diff --git a/test/test_deep/tracepoint/test_tracepoint_config.py b/test/test_deep/tracepoint/test_tracepoint_config.py index 2483ca6..f223fcc 100644 --- a/test/test_deep/tracepoint/test_tracepoint_config.py +++ b/test/test_deep/tracepoint/test_tracepoint_config.py @@ -9,6 +9,10 @@ # but WITHOUT ANY WARRANTY; without even the implied warranty of # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the # GNU Affero General Public License for more details. +# +# You should have received a copy of the GNU Affero General Public License +# along with this program. If not, see . + import unittest from deep.api.tracepoint.tracepoint_config import TracepointWindow, TracePointConfig, FIRE_PERIOD, FIRE_COUNT