You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@skywalking.apache.org by "Superskyyy (via GitHub)" <gi...@apache.org> on 2023/02/16 06:01:48 UTC
[GitHub] [skywalking-python] Superskyyy opened a new pull request, #287: Refactor agent startup logic and experimental os.fork support
Superskyyy opened a new pull request, #287:
URL: https://github.com/apache/skywalking-python/pull/287
This PR is made of two tightly coupled parts:
1. Total rewrite of agent startup logic from module functions -> singleton class. (some other logic was changed in meter to fix wrong forking behavior)
2. Provide experimental support for os.fork(), exposed as an option.
3. A demo directory to provide easier access to oap/kafka/demoservices (for contributors).
Minor changes:
1. Docs: fixed some missed ones over time.
2. Fixed a redis bug.
<!--
â ī¸ Please make sure to read this template first, pull requests that don't accord with this template
maybe closed without notice.
Texts surrounded by `<` and `>` are meant to be replaced by you, e.g. <framework name>, <issue number>.
Put an `x` in the `[ ]` to mark the item as CHECKED. `[x]`
-->
<!-- ==== đ Remove this line WHEN AND ONLY WHEN you're fixing a bug, follow the checklist đ ====
### Fix <bug description or bug issue link>
- [ ] Add a unit test to verify that the fix works.
- [ ] Explain briefly why the bug exists and how to fix it.
==== đ Remove this line WHEN AND ONLY WHEN you're fixing a bug, follow the checklist đ ==== -->
<!-- ==== đ Remove this line WHEN AND ONLY WHEN you're improving the performance, follow the checklist đ ====
### Improve the performance of <class or module or ...>
- [ ] Add a benchmark for the improvement, refer to [the existing ones](https://github.com/apache/skywalking/blob/master/apm-commons/apm-datacarrier/src/test/java/org/apache/skywalking/apm/commons/datacarrier/LinkedArrayBenchmark.java)
- [ ] The benchmark result.
```text
<Paste the benchmark results here>
```
- [ ] Links/URLs to the theory proof or discussion articles/blogs. <links/URLs here>
==== đ Remove this line WHEN AND ONLY WHEN you're improving the performance, follow the checklist đ ==== -->
<!-- ==== đ Remove this line WHEN AND ONLY WHEN you're adding a new feature, follow the checklist đ ====
### <Feature description>
- [ ] If this is non-trivial feature, paste the links/URLs to the design doc.
- [ ] Update the documentation to include this new feature.
- [ ] Tests(including UT, IT, E2E) are added to verify the new feature.
- [ ] If it's UI related, attach the screenshots below.
- [ ] I have rebuilt the `Configuration.md` documentation by running `make doc-gen`
==== đ Remove this line WHEN AND ONLY WHEN you're adding a new feature, follow the checklist đ ==== -->
<!-- ==== đą Remove this line WHEN AND ONLY WHEN you're adding or modifying a plugin instrumentation, follow the checklist đ ====
### <Feature description>
- [ ] If adding a new plugin, add a component id in [the main repo](https://github.com/apache/skywalking/blob/master/oap-server/server-starter/src/main/resources/component-libraries.yml)
- [ ] If adding a new plugin, add a logo in [the UI repo](https://github.com/apache/skywalking-booster-ui/tree/main/src/assets/img/technologies)
- [ ] I have added the library to `pyproject.toml` (plugin group) by running `poetry add library --group plugins`
- [ ] I have rebuilt the `Plugins.md` documentation by running `make doc-gen`
==== đą Remove this line WHEN AND ONLY WHEN you're adding a new feature, follow the checklist đ ==== -->
- [ ] If this pull request closes/resolves/fixes an existing issue, replace the issue url. Closes: <URL to main repo issue>
- [ ] Update the [`CHANGELOG.md`](https://github.com/apache/skywalking-python/blob/master/CHANGELOG.md).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@skywalking.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [skywalking-python] wu-sheng merged pull request #287: Refactor agent startup logic and experimental os.fork support
Posted by "wu-sheng (via GitHub)" <gi...@apache.org>.
wu-sheng merged PR #287:
URL: https://github.com/apache/skywalking-python/pull/287
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@skywalking.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [skywalking-python] sonatype-lift[bot] commented on a diff in pull request #287: Refactor agent startup logic and experimental os.fork support
Posted by "sonatype-lift[bot] (via GitHub)" <gi...@apache.org>.
sonatype-lift[bot] commented on code in PR #287:
URL: https://github.com/apache/skywalking-python/pull/287#discussion_r1108048061
##########
skywalking/agent/__init__.py:
##########
@@ -16,269 +16,353 @@
#
import atexit
+import functools
+import os
+import sys
from queue import Queue, Full
from threading import Thread, Event
-from typing import TYPE_CHECKING
+from typing import TYPE_CHECKING, Optional
from skywalking import config, plugins
from skywalking import loggings
-from skywalking import profile
from skywalking import meter
+from skywalking import profile
from skywalking.agent.protocol import Protocol
from skywalking.command import command_service
from skywalking.loggings import logger
from skywalking.profile.profile_task import ProfileTask
from skywalking.profile.snapshot import TracingThreadSnapshot
-from skywalking.protocol.logging.Logging_pb2 import LogData
from skywalking.protocol.language_agent.Meter_pb2 import MeterData
+from skywalking.protocol.logging.Logging_pb2 import LogData
+from skywalking.utils.singleton import Singleton
if TYPE_CHECKING:
from skywalking.trace.context import Segment
-__started = False
-__protocol = None # type: Protocol
-__heartbeat_thread = __report_thread = __log_report_thread = __query_profile_thread = __command_dispatch_thread \
- = __send_profile_thread = __queue = __log_queue = __snapshot_queue = __meter_queue = __finished = None
-
-
-def __heartbeat():
- wait = base = 30
-
- while not __finished.is_set():
- try:
- __protocol.heartbeat()
- wait = base # reset to base wait time on success
- except Exception as exc:
- logger.error(str(exc))
- wait = min(60, wait * 2 or 1) # double wait time with each consecutive error up to a maximum
-
- __finished.wait(wait)
-
-def __report():
- wait = base = 0
-
- while not __finished.is_set():
+def report_with_backoff(reporter_name, init_wait):
+ """
+ An exponential backoff for retrying reporters.
+ """
+
+ def backoff_decorator(func):
+ @functools.wraps(func)
+ def backoff_wrapper(self, *args, **kwargs):
+ wait = base = init_wait
+ while not self._finished.is_set():
+ try:
+ func(self, *args, **kwargs)
+ wait = base # reset to base wait time on success
+ except Exception: # noqa
+ wait = min(60, wait * 2 or 1) # double wait time with each consecutive error up to a maximum
+ logger.exception(f'Exception in {reporter_name} service in pid {os.getpid()}, '
+ f'retry in {wait} seconds')
+ self._finished.wait(wait)
+ logger.info('finished reporter thread')
+ return backoff_wrapper
+ return backoff_decorator
+
+
+class SkyWalkingAgent(Singleton):
+ """
+ The main singleton class and entrypoint of SkyWalking Python Agent.
+ Upon fork(), original instance rebuild everything (queues, threads, instrumentation) by
+ calling the fork handlers in the class instance.
+ """
+ __started: bool = False # shared by all instances
+
+ def __init__(self):
+ """
+ Protocol is one of gRPC, HTTP and Kafka that
+ provides clients to reporters to communicate with OAP backend.
+ """
+ self.started_pid = None
+ self.__protocol: Optional[Protocol] = None
+ self._finished: Optional[Event] = None
+
+ def __bootstrap(self):
+ # when forking, already instrumented modules must not be instrumented again
+ # otherwise it will cause double instrumentation! (we should provide an un-instrument method)
+ if config.agent_protocol == 'grpc':
+ from skywalking.agent.protocol.grpc import GrpcProtocol
+ self.__protocol = GrpcProtocol()
+ elif config.agent_protocol == 'http':
+ from skywalking.agent.protocol.http import HttpProtocol
+ self.__protocol = HttpProtocol()
+ elif config.agent_protocol == 'kafka':
+ from skywalking.agent.protocol.kafka import KafkaProtocol
+ self.__protocol = KafkaProtocol()
+
+ # Initialize queues for segment, log, meter and profiling snapshots
+ self.__segment_queue: Optional[Queue] = None
+ self.__log_queue: Optional[Queue] = None
+ self.__meter_queue: Optional[Queue] = None
+ self.__snapshot_queue: Optional[Queue] = None
+
+ # Start reporter threads and register queues
+ self.__init_threading()
+
+ def __init_threading(self) -> None:
+ """
+ This method initializes all the queues and threads for the agent and reporters.
+ Upon os.fork(), callback will reinitialize threads and queues by calling this method
+
+ Heartbeat thread is started by default.
+ Segment reporter thread and segment queue is created by default.
+ All other queues and threads depends on user configuration.
+ """
+ self._finished = Event()
+
+ __heartbeat_thread = Thread(name='HeartbeatThread', target=self.__heartbeat, daemon=True)
+ __heartbeat_thread.start()
+
+ self.__segment_queue = Queue(maxsize=config.agent_trace_reporter_max_buffer_size)
+ __segment_report_thread = Thread(name='SegmentReportThread', target=self.__report_segment, daemon=True)
+ __segment_report_thread.start()
+
+ if config.agent_meter_reporter_active:
+ self.__meter_queue = Queue(maxsize=config.agent_meter_reporter_max_buffer_size)
+ __meter_report_thread = Thread(name='MeterReportThread', target=self.__report_meter, daemon=True)
+ __meter_report_thread.start()
+
+ if config.agent_pvm_meter_reporter_active:
+ from skywalking.meter.pvm.cpu_usage import CPUUsageDataSource
+ from skywalking.meter.pvm.gc_data import GCDataSource
+ from skywalking.meter.pvm.mem_usage import MEMUsageDataSource
+ from skywalking.meter.pvm.thread_data import ThreadDataSource
+
+ MEMUsageDataSource().register()
+ CPUUsageDataSource().register()
+ GCDataSource().register()
+ ThreadDataSource().register()
+
+ if config.agent_log_reporter_active:
+ self.__log_queue = Queue(maxsize=config.agent_log_reporter_max_buffer_size)
+ __log_report_thread = Thread(name='LogReportThread', target=self.__report_log, daemon=True)
+ __log_report_thread.start()
+
+ if config.agent_profile_active:
+ # Now only profiler receives commands from OAP
+ __command_dispatch_thread = Thread(name='CommandDispatchThread', target=self.__command_dispatch,
+ daemon=True)
+ __command_dispatch_thread.start()
+
+ self.__snapshot_queue = Queue(maxsize=config.agent_profile_snapshot_transport_buffer_size)
+
+ __query_profile_thread = Thread(name='QueryProfileCommandThread', target=self.__query_profile_command,
+ daemon=True)
+ __query_profile_thread.start()
+
+ __send_profile_thread = Thread(name='SendProfileSnapShotThread', target=self.__send_profile_snapshot,
+ daemon=True)
+ __send_profile_thread.start()
+
+ @staticmethod # for now
+ def __fork_before() -> None:
+ """
+ This handles explicit fork() calls. The child process will not have a running thread, so we need to
+ revive all of them. The parent process will continue to run as normal.
+
+ This does not affect pre-forking server support, which are handled separately.
+ """
+ # possible deadlock would be introduced if some queue is in use when fork() is called and
+ # therefore child process will inherit a locked queue. To avoid this and have side benefit
+ # of a clean queue in child process (prevent duplicated reporting), we simply restart the agent and
+ # reinitialize all queues and threads.
+ logger.warning('SkyWalking Python agent fork support is currently experimental, '
+ 'please report issues if you encounter any.')
+
+ @staticmethod # for now
+ def __fork_after_in_parent() -> None:
+ """
+ Something to do after fork() in parent process
+ """
+ ...
+
+ def __fork_after_in_child(self) -> None:
+ """
+ Simply restart the agent after we detect a fork() call
+ """
+ self.start()
+ logger.info('SkyWalking Python agent spawned in child after fork() call.')
+
+ def start(self) -> None:
+ """
+ Start would be called by user or os.register_at_fork() callback
+ Start will proceed if and only if the agent is not started in the
+ current process.
+
+ When os.fork(), the service instance should be changed to a new one by appending pid.
+ """
+ python_version: tuple = sys.version_info[:2]
+ if python_version[0] < 3 and python_version[1] < 7:
Review Comment:
<picture><img alt="21% of developers fix this issue" src="https://lift.sonatype.com/api/commentimage/fixrate/21/display.svg"></picture>
<b>*Unsupported operand:</b>* `<` is not supported for operand types `typing.Union[int, str]` and `int`.
---
<details><summary>âšī¸ Expand to see all <b>@sonatype-lift</b> commands</summary>
You can reply with the following commands. For example, reply with ***@sonatype-lift ignoreall*** to leave out all findings.
| **Command** | **Usage** |
| ------------- | ------------- |
| `@sonatype-lift ignore` | Leave out the above finding from this PR |
| `@sonatype-lift ignoreall` | Leave out all the existing findings from this PR |
| `@sonatype-lift exclude <file\|issue\|path\|tool>` | Exclude specified `file\|issue\|path\|tool` from Lift findings by updating your config.toml file |
**Note:** When talking to LiftBot, you need to **refresh** the page to see its response.
<sub>[Click here](https://github.com/apps/sonatype-lift/installations/new) to add LiftBot to another repo.</sub></details>
---
<b>Help us improve LIFT! (<i>Sonatype LiftBot external survey</i>)</b>
Was this a good recommendation for you? <sub><small>Answering this survey will not impact your Lift settings.</small></sub>
[â[đ Not relevant](https://www.sonatype.com/lift-comment-rating?comment=392497579&lift_comment_rating=1)â] - [â[đ Won't fix](https://www.sonatype.com/lift-comment-rating?comment=392497579&lift_comment_rating=2)â] - [â[đ Not critical, will fix](https://www.sonatype.com/lift-comment-rating?comment=392497579&lift_comment_rating=3)â] - [â[đ Critical, will fix](https://www.sonatype.com/lift-comment-rating?comment=392497579&lift_comment_rating=4)â] - [â[đ Critical, fixing now](https://www.sonatype.com/lift-comment-rating?comment=392497579&lift_comment_rating=5)â]
##########
skywalking/agent/__init__.py:
##########
@@ -16,269 +16,353 @@
#
import atexit
+import functools
+import os
+import sys
from queue import Queue, Full
from threading import Thread, Event
-from typing import TYPE_CHECKING
+from typing import TYPE_CHECKING, Optional
from skywalking import config, plugins
from skywalking import loggings
-from skywalking import profile
from skywalking import meter
+from skywalking import profile
from skywalking.agent.protocol import Protocol
from skywalking.command import command_service
from skywalking.loggings import logger
from skywalking.profile.profile_task import ProfileTask
from skywalking.profile.snapshot import TracingThreadSnapshot
-from skywalking.protocol.logging.Logging_pb2 import LogData
from skywalking.protocol.language_agent.Meter_pb2 import MeterData
+from skywalking.protocol.logging.Logging_pb2 import LogData
+from skywalking.utils.singleton import Singleton
if TYPE_CHECKING:
from skywalking.trace.context import Segment
-__started = False
-__protocol = None # type: Protocol
-__heartbeat_thread = __report_thread = __log_report_thread = __query_profile_thread = __command_dispatch_thread \
- = __send_profile_thread = __queue = __log_queue = __snapshot_queue = __meter_queue = __finished = None
-
-
-def __heartbeat():
- wait = base = 30
-
- while not __finished.is_set():
- try:
- __protocol.heartbeat()
- wait = base # reset to base wait time on success
- except Exception as exc:
- logger.error(str(exc))
- wait = min(60, wait * 2 or 1) # double wait time with each consecutive error up to a maximum
-
- __finished.wait(wait)
-
-def __report():
- wait = base = 0
-
- while not __finished.is_set():
+def report_with_backoff(reporter_name, init_wait):
+ """
+ An exponential backoff for retrying reporters.
+ """
+
+ def backoff_decorator(func):
+ @functools.wraps(func)
+ def backoff_wrapper(self, *args, **kwargs):
+ wait = base = init_wait
+ while not self._finished.is_set():
+ try:
+ func(self, *args, **kwargs)
+ wait = base # reset to base wait time on success
+ except Exception: # noqa
+ wait = min(60, wait * 2 or 1) # double wait time with each consecutive error up to a maximum
+ logger.exception(f'Exception in {reporter_name} service in pid {os.getpid()}, '
+ f'retry in {wait} seconds')
+ self._finished.wait(wait)
+ logger.info('finished reporter thread')
+ return backoff_wrapper
+ return backoff_decorator
+
+
+class SkyWalkingAgent(Singleton):
+ """
+ The main singleton class and entrypoint of SkyWalking Python Agent.
+ Upon fork(), original instance rebuild everything (queues, threads, instrumentation) by
+ calling the fork handlers in the class instance.
+ """
+ __started: bool = False # shared by all instances
+
+ def __init__(self):
+ """
+ Protocol is one of gRPC, HTTP and Kafka that
+ provides clients to reporters to communicate with OAP backend.
+ """
+ self.started_pid = None
+ self.__protocol: Optional[Protocol] = None
+ self._finished: Optional[Event] = None
+
+ def __bootstrap(self):
+ # when forking, already instrumented modules must not be instrumented again
+ # otherwise it will cause double instrumentation! (we should provide an un-instrument method)
+ if config.agent_protocol == 'grpc':
+ from skywalking.agent.protocol.grpc import GrpcProtocol
+ self.__protocol = GrpcProtocol()
+ elif config.agent_protocol == 'http':
+ from skywalking.agent.protocol.http import HttpProtocol
+ self.__protocol = HttpProtocol()
+ elif config.agent_protocol == 'kafka':
+ from skywalking.agent.protocol.kafka import KafkaProtocol
+ self.__protocol = KafkaProtocol()
+
+ # Initialize queues for segment, log, meter and profiling snapshots
+ self.__segment_queue: Optional[Queue] = None
+ self.__log_queue: Optional[Queue] = None
+ self.__meter_queue: Optional[Queue] = None
+ self.__snapshot_queue: Optional[Queue] = None
+
+ # Start reporter threads and register queues
+ self.__init_threading()
+
+ def __init_threading(self) -> None:
+ """
+ This method initializes all the queues and threads for the agent and reporters.
+ Upon os.fork(), callback will reinitialize threads and queues by calling this method
+
+ Heartbeat thread is started by default.
+ Segment reporter thread and segment queue is created by default.
+ All other queues and threads depends on user configuration.
+ """
+ self._finished = Event()
+
+ __heartbeat_thread = Thread(name='HeartbeatThread', target=self.__heartbeat, daemon=True)
+ __heartbeat_thread.start()
+
+ self.__segment_queue = Queue(maxsize=config.agent_trace_reporter_max_buffer_size)
+ __segment_report_thread = Thread(name='SegmentReportThread', target=self.__report_segment, daemon=True)
+ __segment_report_thread.start()
+
+ if config.agent_meter_reporter_active:
+ self.__meter_queue = Queue(maxsize=config.agent_meter_reporter_max_buffer_size)
+ __meter_report_thread = Thread(name='MeterReportThread', target=self.__report_meter, daemon=True)
+ __meter_report_thread.start()
+
+ if config.agent_pvm_meter_reporter_active:
+ from skywalking.meter.pvm.cpu_usage import CPUUsageDataSource
+ from skywalking.meter.pvm.gc_data import GCDataSource
+ from skywalking.meter.pvm.mem_usage import MEMUsageDataSource
+ from skywalking.meter.pvm.thread_data import ThreadDataSource
+
+ MEMUsageDataSource().register()
+ CPUUsageDataSource().register()
+ GCDataSource().register()
+ ThreadDataSource().register()
+
+ if config.agent_log_reporter_active:
+ self.__log_queue = Queue(maxsize=config.agent_log_reporter_max_buffer_size)
+ __log_report_thread = Thread(name='LogReportThread', target=self.__report_log, daemon=True)
+ __log_report_thread.start()
+
+ if config.agent_profile_active:
+ # Now only profiler receives commands from OAP
+ __command_dispatch_thread = Thread(name='CommandDispatchThread', target=self.__command_dispatch,
+ daemon=True)
+ __command_dispatch_thread.start()
+
+ self.__snapshot_queue = Queue(maxsize=config.agent_profile_snapshot_transport_buffer_size)
+
+ __query_profile_thread = Thread(name='QueryProfileCommandThread', target=self.__query_profile_command,
+ daemon=True)
+ __query_profile_thread.start()
+
+ __send_profile_thread = Thread(name='SendProfileSnapShotThread', target=self.__send_profile_snapshot,
+ daemon=True)
+ __send_profile_thread.start()
+
+ @staticmethod # for now
+ def __fork_before() -> None:
+ """
+ This handles explicit fork() calls. The child process will not have a running thread, so we need to
+ revive all of them. The parent process will continue to run as normal.
+
+ This does not affect pre-forking server support, which are handled separately.
+ """
+ # possible deadlock would be introduced if some queue is in use when fork() is called and
+ # therefore child process will inherit a locked queue. To avoid this and have side benefit
+ # of a clean queue in child process (prevent duplicated reporting), we simply restart the agent and
+ # reinitialize all queues and threads.
+ logger.warning('SkyWalking Python agent fork support is currently experimental, '
+ 'please report issues if you encounter any.')
+
+ @staticmethod # for now
+ def __fork_after_in_parent() -> None:
+ """
+ Something to do after fork() in parent process
+ """
+ ...
+
+ def __fork_after_in_child(self) -> None:
+ """
+ Simply restart the agent after we detect a fork() call
+ """
+ self.start()
+ logger.info('SkyWalking Python agent spawned in child after fork() call.')
+
+ def start(self) -> None:
+ """
+ Start would be called by user or os.register_at_fork() callback
+ Start will proceed if and only if the agent is not started in the
+ current process.
+
+ When os.fork(), the service instance should be changed to a new one by appending pid.
+ """
+ python_version: tuple = sys.version_info[:2]
+ if python_version[0] < 3 and python_version[1] < 7:
Review Comment:
<picture><img alt="21% of developers fix this issue" src="https://lift.sonatype.com/api/commentimage/fixrate/21/display.svg"></picture>
<b>*Unsupported operand:</b>* `<` is not supported for operand types `typing.Union[int, str]` and `int`.
---
<details><summary>âšī¸ Expand to see all <b>@sonatype-lift</b> commands</summary>
You can reply with the following commands. For example, reply with ***@sonatype-lift ignoreall*** to leave out all findings.
| **Command** | **Usage** |
| ------------- | ------------- |
| `@sonatype-lift ignore` | Leave out the above finding from this PR |
| `@sonatype-lift ignoreall` | Leave out all the existing findings from this PR |
| `@sonatype-lift exclude <file\|issue\|path\|tool>` | Exclude specified `file\|issue\|path\|tool` from Lift findings by updating your config.toml file |
**Note:** When talking to LiftBot, you need to **refresh** the page to see its response.
<sub>[Click here](https://github.com/apps/sonatype-lift/installations/new) to add LiftBot to another repo.</sub></details>
---
<b>Help us improve LIFT! (<i>Sonatype LiftBot external survey</i>)</b>
Was this a good recommendation for you? <sub><small>Answering this survey will not impact your Lift settings.</small></sub>
[â[đ Not relevant](https://www.sonatype.com/lift-comment-rating?comment=392497580&lift_comment_rating=1)â] - [â[đ Won't fix](https://www.sonatype.com/lift-comment-rating?comment=392497580&lift_comment_rating=2)â] - [â[đ Not critical, will fix](https://www.sonatype.com/lift-comment-rating?comment=392497580&lift_comment_rating=3)â] - [â[đ Critical, will fix](https://www.sonatype.com/lift-comment-rating?comment=392497580&lift_comment_rating=4)â] - [â[đ Critical, fixing now](https://www.sonatype.com/lift-comment-rating?comment=392497580&lift_comment_rating=5)â]
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@skywalking.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org