- Remove the temporary gRPC log channel (#148)

- Add consideration for log filterers
- Modify all affected Docs

Signed-off-by: YihaoChen <[email protected]>

Author: Superskyyy

README.md

@@ -56,7 +56,7 @@ Alternatively, you can also pass the configurations via environment variables (s
 All supported environment variables can be found [here](docs/EnvVars.md)
 
 ## Report logs with Python Agent
-The Python agent is capable of reporting collected logs to the backend(SkyWalking OAP/ [SkyWalking Satellite Sidecar](https://github.com/apache/skywalking-satellite)), enabling Log & Trace Correlation.
+The Python agent is capable of reporting collected logs to the backend(SkyWalking OAP), enabling Log & Trace Correlation.
 
 Please refer to the [Log Reporter Doc](docs/LogReporter.md) for a detailed guide.
 

docs/EnvVars.md

@@ -33,9 +33,8 @@ Environment Variable | Description | Default
 | `SW_AGENT_PROFILE_ACTIVE` | If `True`, Python agent will enable profile when user create a new profile task. Otherwise disable profile. | `False` |
 | `SW_PROFILE_TASK_QUERY_INTERVAL` | The number of seconds between two profile task query. | `20` |
 | `SW_AGENT_LOG_REPORTER_ACTIVE` | If `True`, Python agent will report collected logs to the OAP or Satellite. Otherwise, it disables the feature. | `False` |
-| `SW_AGENT_LOG_COLLECTOR_BACKEND_SERVICES` | The log reporter will use a separate gRPC channel until the [Satellite](https://github.com/apache/skywalking-satellite) project is ready. | `127.0.0.1:11800` |
 | `SW_AGENT_LOG_REPORTER_BUFFER_SIZE` | The maximum queue backlog size for sending log data to backend, logs beyond this are silently dropped. | `10000` |
-| `SW_AGENT_LOG_REPORTER_MESSAGE_SIZE` | Max message size allowed for log transmission. | `10485760` |
 | `SW_AGENT_LOG_REPORTER_LEVEL` | This config specifies the logger levels of concern, any logs with a level below the config will be ignored. | `WARNING` |
+| `SW_AGENT_LOG_IGNORE_FILTER` | This config customizes whether to ignore the application-defined logger filters, if `True`, all logs are reported disregarding any filter rules. | `False` |
 | `SW_AGENT_LOG_REPORTER_FORMATTED` | If `True`, the log reporter will transmit the logs as formatted. Otherwise, puts logRecord.msg and logRecord.args into message content and tags(`argument.n`), respectively. Along with an `exception` tag if an exception was raised. | `True` |
 | `SW_AGENT_LOG_REPORTER_LAYOUT` | The log reporter formats the logRecord message based on the layout given. | `%(asctime)s [%(threadName)s] %(levelname)s %(name)s - %(message)s` |

docs/LogReporter.md

@@ -9,16 +9,13 @@ To utilize this feature, you will need to add some new configurations to the age
 from skywalking import agent, config
 
 config.init(collector_address='127.0.0.1:11800', service_name='your awesome service',
-                log_grpc_reporter_active=True, log_grpc_collector_address='127.0.0.1:11800')
+                log_grpc_reporter_active=True)
 agent.start()
 ``` 
 
 `log_grpc_reporter_active=True` - Enables the log reporter.
 
-`log_grpc_collector_address` - For now, the log reporter uses a separate gRPC channel(will be merged upon the [SkyWalking Satellite Sidecar](https://github.com/apache/skywalking-satellite) project matures). 
-If you would like to use the Satellite sidecar, you will need to configure an address pointing to its gatherer. Otherwise, you can simply keep the address the same as the OAP.
-
-`log_grpc_reporter_max_buffer_size` and  `log_grpc_reporter_max_message_size` - Used to limit the reporting overhead.
+`log_grpc_reporter_max_buffer_size` - The maximum queue backlog size for sending log data to backend, logs beyond this are silently dropped.
 
 Alternatively, you can pass configurations through environment variables. 
 Please refer to [EnvVars.md](EnvVars.md) for the list of environment variables associated with the log reporter.
@@ -31,6 +28,21 @@ In other words, the agent ignores some unwanted logs based on your level thresho
 
 Note that it also works with your custom logger levels, simply specify its string name in the config.
 
+### Ignore log filters
+The following config is disabled by default. When enabled, the log reporter will collect logs disregarding your custom log filters.
+
+For example, if you attach the filter below to the logger - the default behavior of log reporting aligns with the filter
+(not reporting any logs with a message starting with `SW test`)
+```python
+class AppFilter(logging.Filter):
+    def filter(self, record):
+        return not record.getMessage().startswith('SW test')
+
+logger.addFilter(AppFilter())
+```
+However, if you do would like to report those filtered logs, set the `log_grpc_reporter_ignore_filter` to `True`.
+
+
 ## Formatting
 Note that regardless of the formatting, Python agent will always report the following three tags - 
 
@@ -46,7 +58,7 @@ If not set, the agent uses the layout below by default, else the agent uses your
 
 `'%(asctime)s [%(threadName)s] %(levelname)s %(name)s - %(message)s'`
 
-If the layout is set to `None`, the reported log content will only contain the pre-formatted `LogRecord.message`(`msg % args`) without any additional styles and information.
+If the layout is set to `None`, the reported log content will only contain the pre-formatted `LogRecord.message`(`msg % args`) without any additional styles, information or extra fields.
 
 ### Transmit un-formatted logs
 You can also choose to report the log messages without any formatting.
@@ -55,7 +67,7 @@ It separates the raw log msg `logRecord.msg` and `logRecord.args`, then puts the
 Note when you set `log_grpc_reporter_formatted` to False, it ignores your custom layout introduced above.
 
 As an example, the following code:
-```Python
+```python
 logger.info("SW test log %s %s %s", 'arg0', 'arg1', 'arg2')
 ```
 

skywalking/agent/__init__.py

@@ -26,7 +26,6 @@
 from skywalking import config, plugins
 from skywalking import loggings
 from skywalking.agent.protocol import Protocol
-from skywalking.agent.protocol.grpc_log import GrpcLogProtocol
 from skywalking.command import command_service
 from skywalking.config import profile_active, profile_task_query_interval
 from skywalking.loggings import logger
@@ -35,7 +34,7 @@
     from skywalking.trace.context import Segment
 
 __started = False
-__protocol = __log_protocol = Protocol()  # type: Protocol
+__protocol = Protocol()  # type: Protocol
 __heartbeat_thread = __report_thread = __log_report_thread = __query_profile_thread = __command_dispatch_thread \
     = __queue = __log_queue = __finished = None
 
@@ -60,20 +59,10 @@ def __report():
         __finished.wait(0)
 
 
-def __log_heartbeat():
+def __report_log():
     while not __finished.is_set():
         try:
-            __log_protocol.heartbeat()
-        except Exception as exc:
-            logger.error(str(exc))
-
-        __finished.wait(30)
-
-
-def __log_report():
-    while not __finished.is_set():
-        try:
-            __log_protocol.report(__log_queue)
+            __protocol.report_log(__log_queue)
         except Exception as exc:
             logger.error(str(exc))
 
@@ -112,17 +101,15 @@ def __init_threading():
 
     if config.log_grpc_reporter_active:
         __log_queue = Queue(maxsize=config.log_grpc_reporter_max_buffer_size)
-        __log_heartbeat_thread = Thread(name='LogHeartbeatThread', target=__log_heartbeat, daemon=True)
-        __log_report_thread = Thread(name='LogReportThread', target=__log_report, daemon=True)
-        __log_heartbeat_thread.start()
+        __log_report_thread = Thread(name='LogReportThread', target=__report_log, daemon=True)
         __log_report_thread.start()
 
     if profile_active:
         __query_profile_thread.start()
 
 
 def __init():
-    global __protocol, __log_protocol
+    global __protocol
 
     if config.protocol == 'grpc':
         from skywalking.agent.protocol.grpc import GrpcProtocol
@@ -135,9 +122,8 @@ def __init():
         __protocol = KafkaProtocol()
 
     plugins.install()
-    if config.log_grpc_reporter_active:
+    if config.log_grpc_reporter_active:  # todo - Add support for printing traceID/ context in logs
         from skywalking import log
-        __log_protocol = GrpcLogProtocol()
         log.install()
 
     __init_threading()
@@ -147,7 +133,7 @@ def __fini():
     __protocol.report(__queue, False)
     __queue.join()
     if config.log_grpc_reporter_active:
-        __log_protocol.report(__log_queue, False)
+        __protocol.report_log(__log_queue, False)
         __log_queue.join()
     __finished.set()
 

skywalking/agent/protocol/__init__.py

@@ -35,5 +35,8 @@ def heartbeat(self):
     def report(self, queue: Queue, block: bool = True):
         raise NotImplementedError()
 
+    def report_log(self, queue: Queue, block: bool = True):
+        raise NotImplementedError()
+
     def query_profile_commands(self):
         pass

skywalking/agent/protocol/grpc.py

@@ -23,12 +23,13 @@
 import grpc
 from skywalking.protocol.common.Common_pb2 import KeyStringValuePair
 from skywalking.protocol.language_agent.Tracing_pb2 import SegmentObject, SpanObject, Log, SegmentReference
+from skywalking.protocol.logging.Logging_pb2 import LogData
 
 from skywalking import config
 from skywalking.agent import Protocol
 from skywalking.agent.protocol.interceptors import header_adder_interceptor
 from skywalking.client.grpc import GrpcServiceManagementClient, GrpcTraceSegmentReportService, \
-    GrpcProfileTaskChannelService
+    GrpcProfileTaskChannelService, GrpcLogDataReportService
 from skywalking.loggings import logger
 from skywalking.trace.segment import Segment
 
@@ -52,6 +53,7 @@ def __init__(self):
         self.service_management = GrpcServiceManagementClient(self.channel)
         self.traces_reporter = GrpcTraceSegmentReportService(self.channel)
         self.profile_query = GrpcProfileTaskChannelService(self.channel)
+        self.log_reporter = GrpcLogDataReportService(self.channel)
 
     def _cb(self, state):
         logger.debug('grpc channel connectivity changed, [%s -> %s]', self.state, state)
@@ -67,11 +69,6 @@ def heartbeat(self):
                 self.service_management.send_instance_props()
                 self.properties_sent = True
 
-            logger.debug(
-                'segment reporter service heart beats, [%s], [%s]',
-                config.service_name,
-                config.service_instance,
-            )
             self.service_management.send_heart_beat()
 
         except grpc.RpcError:
@@ -142,3 +139,27 @@ def generator():
             self.traces_reporter.report(generator())
         except grpc.RpcError:
             self.on_error()
+
+    def report_log(self, queue: Queue, block: bool = True):
+        start = time()
+
+        def generator():
+            while True:
+                try:
+                    timeout = config.QUEUE_TIMEOUT - int(time() - start)  # type: int
+                    if timeout <= 0:
+                        return
+                    log_data = queue.get(block=block, timeout=timeout)  # type: LogData
+                except Empty:
+                    return
+
+                queue.task_done()
+
+                logger.debug('Reporting Log')
+
+                yield log_data
+
+        try:
+            self.log_reporter.report(generator())
+        except grpc.RpcError:
+            self.on_error()

skywalking/agent/protocol/grpc_log.py

@@ -28,6 +28,7 @@
 from skywalking.client import ServiceManagementClient, TraceSegmentReportService, ProfileTaskChannelService, \
     LogDataReportService
 from skywalking.command import command_service
+from skywalking.loggings import logger
 from skywalking.profile import profile_task_execution_service
 
 
@@ -43,6 +44,11 @@ def send_instance_props(self):
         ))
 
     def send_heart_beat(self):
+        logger.debug(
+            'service heart beats, [%s], [%s]',
+            config.service_name,
+            config.service_instance,
+        )
         self.service_stub.keepAlive(InstancePingPkg(
             service=config.service_name,
             serviceInstance=config.service_instance,
@@ -57,6 +63,14 @@ def report(self, generator):
         self.report_stub.collect(generator)
 
 
+class GrpcLogDataReportService(LogDataReportService):
+    def __init__(self, channel: grpc.Channel):
+        self.report_stub = LogReportServiceStub(channel)
+
+    def report(self, generator):
+        self.report_stub.collect(generator)
+
+
 class GrpcProfileTaskChannelService(ProfileTaskChannelService):
     def __init__(self, channel: grpc.Channel):
         self.task_stub = ProfileTaskStub(channel)
@@ -71,11 +85,3 @@ def do_query(self):
 
         commands = self.task_stub.getProfileTaskCommands(query)
         command_service.receive_command(commands)
-
-
-class GrpcLogDataReportService(LogDataReportService):
-    def __init__(self, channel: grpc.Channel):
-        self.report_stub = LogReportServiceStub(channel)
-
-    def report(self, generator):
-        self.report_stub.collect(generator)

skywalking/client/grpc.py

@@ -70,10 +70,10 @@
 # NOTE - Log reporting requires a separate channel, will merge in the future.
 log_grpc_reporter_active = True if os.getenv('SW_AGENT_LOG_REPORTER_ACTIVE') and \
                          os.getenv('SW_AGENT_LOG_REPORTER_ACTIVE') == 'True' else False  # type: bool
-log_grpc_collector_address = os.getenv('SW_AGENT_LOG_COLLECTOR_BACKEND_SERVICES') or '127.0.0.1:11800'  # type: str
 log_grpc_reporter_max_buffer_size = int(os.getenv('SW_AGENT_LOG_REPORTER_BUFFER_SIZE') or '10000')  # type: int
-log_grpc_reporter_max_message_size = int(os.getenv('SW_AGENT_LOG_REPORTER_MESSAGE_SIZE') or '10485760')  # type: int
 log_grpc_reporter_level = os.getenv('SW_AGENT_LOG_REPORTER_LEVEL') or 'WARNING'  # type: str
+log_grpc_reporter_ignore_filter = True if os.getenv('SW_AGENT_LOG_IGNORE_FILTER') and \
+                         os.getenv('SW_AGENT_LOG_REPORTER_FORMATTED') == 'True' else False  # type: bool
 log_grpc_reporter_formatted = False if os.getenv('SW_AGENT_LOG_REPORTER_FORMATTED') and \
                          os.getenv('SW_AGENT_LOG_REPORTER_FORMATTED') == 'False' else True  # type: bool
 log_grpc_reporter_layout = os.getenv('SW_AGENT_LOG_REPORTER_LAYOUT') or \