feat: add GreenletProfiler

.github/workflows/CI.yaml

@@ -223,6 +223,10 @@ jobs:
             path: tests/e2e/case/http/e2e.yaml
           - name: Kafka
             path: tests/e2e/case/kafka/e2e.yaml
+          - name: profiling_threading
+            path: tests/e2e/case/profiling/threading/e2e.yaml
+          - name: profiling_greenlet
+            path: tests/e2e/case/profiling/greenlet/e2e.yaml
       fail-fast: false
     steps:
       - name: Checkout source codes

skywalking/profile/profile_context.py

@@ -34,6 +34,24 @@
 from skywalking.utils.time import current_milli_time
 
 
+THREAD_MODEL = 'thread'
+try:
+    from gevent import monkey
+    import greenlet
+    from gevent.exceptions import BlockingSwitchOutError
+
+    if monkey.is_module_patched('threading'):
+        if greenlet.__version__ < '2.0.0':
+            # todo: greenlet will raise a segment fault with signal 11 when it upgrade to 2.0.0
+            # this issue may be caused by gevent's compatibility with greenlet
+            # we should do some tests when gevent release a new version to verify if this issue would be fixed
+            THREAD_MODEL = 'greenlet'
+        else:
+            logger.warn('greenlet profiler can not work with version >= 2.0.0')
+except ImportError:
+    pass
+
+
 class ProfileTaskExecutionContext:
     def __init__(self, task: ProfileTask):
         self.task = task  # type: ProfileTask
@@ -44,15 +62,29 @@ def __init__(self, task: ProfileTask):
         self._profiling_stop_event = None  # type: Optional[Event]
 
     def start_profiling(self):
-        profile_thread = ProfileThread(self)
-        self._profiling_stop_event = Event()
+        if THREAD_MODEL == 'greenlet':
+            # GreenletProfiler will be started when it is created
+            pass
 
-        self._profiling_thread = Thread(target=profile_thread.start, args=[self._profiling_stop_event], daemon=True)
-        self._profiling_thread.start()
+        else:
+            profile_thread = ProfileThread(self)
+            self._profiling_stop_event = Event()
+
+            self._profiling_thread = Thread(target=profile_thread.start, args=[self._profiling_stop_event], daemon=True)
+            self._profiling_thread.start()
 
     def stop_profiling(self):
-        if self._profiling_thread is not None and self._profiling_stop_event is not None:
-            self._profiling_stop_event.set()
+        if THREAD_MODEL == 'greenlet':
+            for profiler in self.profiling_segment_slots:
+                if profiler and isinstance(profiler, GreenletProfiler):
+                    profiler.stop_profiling()
+
+        else:
+            if (
+                self._profiling_thread is not None
+                and self._profiling_stop_event is not None
+            ):
+                self._profiling_stop_event.set()
 
     def attempt_profiling(self, trace_context: SpanContext, segment_id: str, first_span_opname: str) -> \
             ProfileStatusReference:
@@ -78,10 +110,24 @@ def attempt_profiling(self, trace_context: SpanContext, segment_id: str, first_s
                                                            using_slot_cnt + 1):
             return ProfileStatusReference.create_with_none()
 
-        thread_profiler = ThreadProfiler(trace_context=trace_context,
-                                         segment_id=segment_id,
-                                         profiling_thread=current_thread(),
-                                         profile_context=self)
+        if THREAD_MODEL == 'greenlet':
+            curr = greenlet.getcurrent()
+            thread_profiler = GreenletProfiler(
+                trace_context=trace_context,
+                segment_id=segment_id,
+                profiling_thread=curr,
+                profile_context=self,
+            )
+            thread_profiler.start_profiling(self)
+
+        else:
+            # default is thread
+            thread_profiler = ThreadProfiler(
+                trace_context=trace_context,
+                segment_id=segment_id,
+                profiling_thread=current_thread(),
+                profile_context=self,
+            )
 
         slot_length = self.profiling_segment_slots.length()
         for idx in range(slot_length):
@@ -139,9 +185,8 @@ def profiling(self, context: ProfileTaskExecutionContext):
             profilers = self._task_execution_context.profiling_segment_slots
 
             for profiler in profilers:  # type: ThreadProfiler
-                if profiler is None:
+                if profiler is None or isinstance(profiler, GreenletProfiler):
                     continue
-
                 if profiler.profile_status.get() is ProfileStatus.PENDING:
                     profiler.start_profiling_if_need()
                 elif profiler.profile_status.get() is ProfileStatus.PROFILING:
@@ -221,3 +266,96 @@ def build_snapshot(self) -> Optional[TracingThreadSnapshot]:
 
     def matches(self, trace_context: SpanContext) -> bool:
         return self.trace_context == trace_context
+
+
+class GreenletProfiler:
+    def __init__(
+        self,
+        trace_context: SpanContext,
+        segment_id: str,
+        profiling_thread,  # greenlet
+        profile_context: ProfileTaskExecutionContext,
+    ):
+        self._task_execution_service = profile.profile_task_execution_service
+        self.trace_context = trace_context
+        self._segment_id = segment_id
+        self._profiling_thread = profiling_thread
+        self._profile_context = profile_context
+        self._profile_start_time = -1
+        self.profiling_max_time_mills = config.profile_duration * 60 * 1000
+
+        self.dump_sequence = 0
+
+        self.profile_status = ProfileStatusReference.create_with_pending()
+
+    def stop_profiling(self):
+
+        curr = self._profiling_thread
+        curr.settrace(self._old_trace)
+        self.profile_status.update_status(ProfileStatus.STOPPED)
+
+    def build_snapshot(self) -> Optional[TracingThreadSnapshot]:
+        stack_list = []
+        extracted = traceback.extract_stack(self._profiling_thread.gr_frame)
+        for idx, item in enumerate(extracted):
+            if idx > config.profile_dump_max_stack_depth:
+                break
+
+            code_sig = f'{item.filename}.{item.name}: {item.lineno}'
+            stack_list.append(code_sig)
+
+        # if is first dump, check is can start profiling
+        if (
+            self.dump_sequence == 0
+            and not self._profile_context.is_start_profileable()
+        ):
+            return None
+
+        current_time = current_milli_time()
+        snapshot = TracingThreadSnapshot(
+            self._profile_context.task.task_id,
+            self._segment_id,
+            self.dump_sequence,
+            current_time,
+            stack_list,
+                    )
+        self.dump_sequence += 1
+        return snapshot
+
+
+    def start_profiling(self, context: ProfileTaskExecutionContext):
+        self._task_execution_context = context
+        try:
+            curr = self._profiling_thread
+
+            def callback(event, args):
+                origin, target = args
+                if origin == curr or target == curr:
+                    try:
+                        snapshot = self.build_snapshot()
+                        if snapshot is not None:
+                            agent.add_profiling_snapshot(snapshot)
+                        else:
+                            # tell execution context current tracing thread dump failed, stop it
+                            # todo test it
+                            self._profile_context.stop_tracing_profile(self.trace_context)
+                    except BlockingSwitchOutError:
+                        self._profile_context.stop_tracing_profile(self.trace_context)
+                    except Exception as e:
+                        logger.error(f'build and add snapshot failed. error: {e}')
+                        self._profile_context.stop_tracing_profile(self.trace_context)
+                        raise e
+
+
+            self.profile_status.update_status(ProfileStatus.PROFILING)
+            self._old_trace = curr.settrace(callback)
+
+        except Exception as e:
+            logger.error('profiling task fail. task_id:[%s] error:[%s]', self._profiling_context.task.task_id, e)
+            # todo test this can current stop profile task or not
+            self.profiling_context.stop_current_profile_task(
+                self._task_execution_context
+            )
+
+    def matches(self, trace_context: SpanContext) -> bool:
+        return self.trace_context == trace_context

tests/e2e/base/Dockerfile.e2e

@@ -30,7 +30,7 @@ ENV PATH="/skywalking-python/venv/bin:$PATH"
 
 RUN pip install requests kafka-python
 # Extra dependencies for e2e services
-RUN pip install fastapi uvicorn aiohttp
+RUN pip install fastapi uvicorn aiohttp flask
 
 # Entrypoint with agent attached
 Entrypoint ["sw-python", "run"]

tests/e2e/case/expected/profile-create.yml

@@ -0,0 +1,17 @@
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements.  See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License.  You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+id: {{ notEmpty .id }}
+errorreason: null

tests/e2e/case/expected/profile-list-finished.yml

@@ -0,0 +1,34 @@
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements.  See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License.  You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+{{- contains . }}
+- id: {{ notEmpty .id }}
+  serviceid: {{ b64enc "e2e-service-provider" }}.1
+  servicename: ""
+  endpointname: /artist
+  starttime: {{ gt .starttime 0 }}
+  duration: 5
+  mindurationthreshold: 0
+  dumpperiod: 10
+  maxsamplingcount: 5
+  logs:
+    {{- contains .logs }}
+    - id: {{ notEmpty .id }}
+      instanceid: {{ b64enc "e2e-service-provider" }}.1_{{ b64enc "provider1" }}
+      operationtype: NOTIFIED 
+      instancename: ""
+      operationtime: {{ gt .operationtime 0 }}
+    {{- end }}
+{{- end }}

tests/e2e/case/expected/profile-list-notified.yml

@@ -0,0 +1,34 @@
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements.  See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License.  You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+{{- contains . }}
+- id: {{ notEmpty .id }}
+  serviceid: {{ b64enc "e2e-service-provider" }}.1
+  servicename: ""
+  endpointname: /artist
+  starttime: {{ gt .starttime 0 }}
+  duration: 5
+  mindurationthreshold: 0
+  dumpperiod: 10
+  maxsamplingcount: 5
+  logs:
+    {{- contains .logs }}
+    - id: {{ notEmpty .id }}
+      instanceid: {{ b64enc "e2e-service-provider" }}.1_{{ b64enc "provider1" }}
+      operationtype: NOTIFIED
+      instancename: ""
+      operationtime: {{ gt .operationtime 0 }}
+    {{- end }}
+{{- end }}

tests/e2e/case/expected/profile-segment-analyze.yml

@@ -0,0 +1,28 @@
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements.  See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License.  You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+tip: null
+trees:
+  {{- contains .trees }}
+  - elements:
+    {{- contains .elements }}
+    - id: "{{ notEmpty .id }}"
+      parentid: "{{ notEmpty .parentid }}"
+      codesignature: "/services/provider.py.artist: 29"
+      duration: {{ gt .duration 0 }}
+      durationchildexcluded: {{ ge .durationchildexcluded 0 }}
+      count: {{ gt .count 0 }}
+    {{- end }}
+  {{- end }}

tests/e2e/case/expected/profile-segment-detail.yml

@@ -0,0 +1,38 @@
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements.  See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License.  You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+spans:
+{{- contains .spans }}
+- spanid: 0
+  parentspanid: -1
+  servicecode: e2e-service-provider
+  serviceinstancename: ""
+  starttime: {{ gt .starttime 0 }}
+  endtime: {{ gt .endtime 0 }}
+  endpointname: /artist
+  type: Entry
+  peer: {{ notEmpty .peer }}
+  component: {{ notEmpty .component }}
+  iserror: false
+  layer: Http
+  tags:
+    {{- contains .tags }}
+    - key: http.url
+      value: {{ notEmpty .value }}
+    - key: http.method
+      value: POST
+    {{- end }}
+  logs: []
+{{- end }}