Add Neo4j plugin. (#312)

* Add Neo4j plugin.

---------

Co-authored-by: 吴晟 Wu Sheng <wu.sheng@foxmail.com>

Author: Jedore

CHANGELOG.md

@@ -1,6 +1,8 @@
 ## Change Logs
 
 ### 1.1.0
+- Plugins:
+  - Add neo4j plugin.(#312)
 
 - Fixes:
   - Fix unexpected 'No active span' IllegalStateError (#311)

docs/en/setup/Plugins.md

@@ -33,6 +33,7 @@ or a limitation of SkyWalking auto-instrumentation (welcome to contribute!)
 | [kafka-python](https://kafka-python.readthedocs.io) | Python >=3.7 - ['2.0'];  | `sw_kafka` |
 | [loguru](https://pypi.org/project/loguru/) | Python >=3.7 - ['0.6.0', '0.7.0'];  | `sw_loguru` |
 | [mysqlclient](https://mysqlclient.readthedocs.io/) | Python >=3.7 - ['2.1.*'];  | `sw_mysqlclient` |
+| [neo4j](https://neo4j.com/docs/python-manual/5/) | Python >=3.7 - ['5.*'];  | `sw_neo4j` |
 | [psycopg[binary]](https://www.psycopg.org/) | Python >=3.11 - ['3.1.*']; Python >=3.7 - ['3.0.18', '3.1.*'];  | `sw_psycopg` |
 | [psycopg2-binary](https://www.psycopg.org/) | Python >=3.10 - NOT SUPPORTED YET; Python >=3.7 - ['2.9'];  | `sw_psycopg2` |
 | [pymongo](https://pymongo.readthedocs.io) | Python >=3.7 - ['3.11.*'];  | `sw_pymongo` |
@@ -53,6 +54,8 @@ in SkyWalking currently. Celery clients can use whatever protocol they want.
 - While Falcon is instrumented, only Hug is tested.
 Hug is believed to be abandoned project, use this plugin with a bit more caution.
 Instead of Hug, plugin test should move to test actual Falcon.
+- The Neo4j plugin integrates neo4j python driver 5.x.x versions which
+support both Neo4j 5 and 4.4 DBMS.
 - The websocket instrumentation only traces client side connection handshake,
 the actual message exchange (send/recv) is not traced since injecting headers to socket message
 body is the only way to propagate the trace context, which requires customization of message structure

poetry.lock

@@ -132,6 +132,7 @@ websockets = "^10.4"
 loguru = "^0.6.0"
 httpx = "^0.23.3"
 confluent-kafka = "^2.0.2"
+neo4j = "^5.9.0"
 
 [tool.poetry.group.lint.dependencies]
 pylint = '2.13.9'

pyproject.toml

@@ -38,6 +38,7 @@ class Component(Enum):
     RabbitmqConsumer = 53
     Elasticsearch = 47
     HBase = 94
+    Neo4j = 112
     Urllib3 = 7006
     Sanic = 7007
     AioHttp = 7008

skywalking/__init__.py

@@ -0,0 +1,96 @@
+#
+# 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.
+#
+
+import json
+
+from skywalking import Layer, Component, config
+from skywalking.trace.context import get_context
+from skywalking.trace.tags import TagDbType, TagDbInstance, TagDbStatement, TagDbSqlParameters
+
+link_vector = ['https://neo4j.com/docs/python-manual/5/']
+support_matrix = {
+    'neo4j': {
+        '>=3.7': ['5.*'],
+    }
+}
+note = """The Neo4j plugin integrates neo4j python driver 5.x.x versions which
+support both Neo4j 5 and 4.4 DBMS."""
+
+
+def install():
+    from neo4j import AsyncSession, Session
+    from neo4j._sync.work.transaction import TransactionBase
+    from neo4j._async.work.transaction import AsyncTransactionBase
+
+    _session_run = Session.run
+    _async_session_run = AsyncSession.run
+    _transaction_run = TransactionBase.run
+    _async_transaction_run = AsyncTransactionBase.run
+
+    def _archive_span(span, database, query, parameters, **kwargs):
+        span.layer = Layer.Database
+        span.tag(TagDbType('Neo4j'))
+        span.tag(TagDbInstance(database or ''))
+        span.tag(TagDbStatement(query))
+
+        parameters = dict(parameters or {}, **kwargs)
+        if config.plugin_sql_parameters_max_length and parameters:
+            parameter = json.dumps(parameters, ensure_ascii=False)
+            max_len = config.plugin_sql_parameters_max_length
+            parameter = f'{parameter[0:max_len]}...' if len(
+                parameter) > max_len else parameter
+            span.tag(TagDbSqlParameters(f'[{parameter}]'))
+
+    def get_peer(address):
+        return f'{address.host}:{address.port}'
+
+    def _sw_session_run(self, query, parameters, **kwargs):
+        with get_context().new_exit_span(
+                op='Neo4j/Session/run',
+                peer=get_peer(self._pool.address),
+                component=Component.Neo4j) as span:
+            _archive_span(span, self._config.database, query, parameters, **kwargs)
+            return _session_run(self, query, parameters, **kwargs)
+
+    def _sw_transaction_run(self, query, parameters=None, **kwargs):
+        with get_context().new_exit_span(
+                op='Neo4j/Transaction/run',
+                peer=get_peer(self._connection.unresolved_address),
+                component=Component.Neo4j) as span:
+            _archive_span(span, self._database, query, parameters, **kwargs)
+            return _transaction_run(self, query, parameters, **kwargs)
+
+    async def _sw_async_session_run(self, query, parameters, **kwargs):
+        with get_context().new_exit_span(
+                op='Neo4j/AsyncSession/run',
+                peer=get_peer(self._pool.address),
+                component=Component.Neo4j) as span:
+            _archive_span(span, self._config.database, query, parameters, **kwargs)
+            return await _async_session_run(self, query, parameters, **kwargs)
+
+    async def _sw_async_transaction_run(self, query, parameters, **kwargs):
+        with get_context().new_exit_span(
+                op='Neo4j/AsyncTransaction/run',
+                peer=get_peer(self._connection.unresolved_address),
+                component=Component.Neo4j) as span:
+            _archive_span(span, self._database, query, parameters, **kwargs)
+            return await _async_transaction_run(self, query, parameters, **kwargs)
+
+    Session.run = _sw_session_run
+    AsyncSession.run = _sw_async_session_run
+    TransactionBase.run = _sw_transaction_run
+    AsyncTransactionBase.run = _sw_async_transaction_run

skywalking/plugins/sw_neo4j.py

@@ -0,0 +1,16 @@
+#
+# 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.
+#

tests/plugin/data/sw_neo4j/__init__.py

@@ -0,0 +1,84 @@
+#
+# 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.
+#
+
+version: '2.1'
+
+services:
+  collector:
+    extends:
+      service: collector
+      file: ../../docker-compose.base.yml
+
+  neo4j:
+    image:  neo4j:4.4-community
+    hostname: neo4j
+    ports:
+      - 7687:7687
+    environment:
+      - NEO4J_dbms_security_auth__enabled=false
+    healthcheck:
+      test: ["CMD", "bash", "-c", "cat < /dev/null > /dev/tcp/127.0.0.1/7687"]
+      interval: 5s
+      timeout: 60s
+      retries: 120
+    networks:
+      - beyond
+
+  provider:
+    extends:
+      service: agent
+      file: ../../docker-compose.base.yml
+    ports:
+      - 9091:9091
+    volumes:
+      - .:/app
+    command: ['bash', '-c', 'pip3 install uvicorn && pip3 install fastapi && pip3 install -r /app/requirements.txt && sw-python run python3 /app/services/provider.py']
+    depends_on:
+      collector:
+        condition: service_healthy
+      neo4j:
+        condition: service_healthy
+    healthcheck:
+      test: ["CMD", "bash", "-c", "cat < /dev/null > /dev/tcp/127.0.0.1/9091"]
+      interval: 5s
+      timeout: 60s
+      retries: 120
+    environment:
+      SW_AGENT_NAME: provider
+      SW_AGENT_LOGGING_LEVEL: DEBUG
+      SW_PLUGIN_SQL_PARAMETERS_MAX_LENGTH: 512
+
+  consumer:
+    extends:
+      service: agent
+      file: ../../docker-compose.base.yml
+    ports:
+      - 9090:9090
+    volumes:
+      - .:/app
+    command: ['bash', '-c', 'pip3 install uvicorn && pip3 install fastapi && pip3 install -r /app/requirements.txt && sw-python run python3 /app/services/consumer.py']
+    depends_on:
+      collector:
+        condition: service_healthy
+      provider:
+        condition: service_healthy
+    environment:
+      SW_AGENT_NAME: consumer
+      SW_AGENT_LOGGING_LEVEL: DEBUG
+
+networks:
+  beyond: