fix grpc disconnect, SW_AGENT_MAX_BUFFER_SIZE

tom-pytel · tom-pytel · commit 41c222923d3f · 2021-08-03T09:04:45.000-03:00
diff --git a/docs/EnvVars.md b/docs/EnvVars.md
@@ -11,6 +11,7 @@ Environment Variable | Description | Default
 | `SW_AGENT_AUTHENTICATION` | The authentication token to verify that the agent is trusted by the backend OAP, as for how to configure the backend, refer to [the yaml](https://github.com/apache/skywalking/blob/4f0f39ffccdc9b41049903cc540b8904f7c9728e/oap-server/server-bootstrap/src/main/resources/application.yml#L155-L158). | unset |
 | `SW_AGENT_LOGGING_LEVEL` | The logging level, could be one of `CRITICAL`, `FATAL`, `ERROR`, `WARN`(`WARNING`), `INFO`, `DEBUG` | `INFO` |
 | `SW_AGENT_DISABLE_PLUGINS` | The name patterns in CSV pattern, plugins whose name matches one of the pattern won't be installed | `''` |
+| `SW_AGENT_MAX_BUFFER_SIZE` | The maximum queue backlog size for sending the segment data to backend, segments beyond this are silently dropped | `'1000'` |
 | `SW_SQL_PARAMETERS_LENGTH` | The maximum length of the collected parameter, parameters longer than the specified length will be truncated, length 0 turns off parameter tracing | `0` |
 | `SW_PYMONGO_TRACE_PARAMETERS` | Indicates whether to collect the filters of pymongo | `False` |
 | `SW_PYMONGO_PARAMETERS_MAX_LENGTH` | The maximum length of the collected filters, filters longer than the specified length will be truncated |  `512` |
diff --git a/skywalking/agent/__init__.py b/skywalking/agent/__init__.py
@@ -38,24 +38,30 @@
 
 def __heartbeat():
     while not __finished.is_set():
-        if connected():
+        try:
             __protocol.heartbeat()
+        except Exception as exc:
+            logger.error(str(exc))
 
-        __finished.wait(30 if connected() else 3)
+        __finished.wait(30)
 
 
 def __report():
     while not __finished.is_set():
-        if connected():
+        try:
             __protocol.report(__queue)  # is blocking actually, blocks for max config.QUEUE_TIMEOUT seconds
+        except Exception as exc:
+            logger.error(str(exc))
 
-        __finished.wait(1)
+        __finished.wait(0)
 
 
 def __query_profile_command():
     while not __finished.is_set():
-        if connected():
+        try:
             __protocol.query_profile_commands()
+        except Exception as exc:
+            logger.error(str(exc))
 
         __finished.wait(profile_task_query_interval)
 
@@ -68,7 +74,7 @@ def __command_dispatch():
 def __init_threading():
     global __heartbeat_thread, __report_thread,  __query_profile_thread, __command_dispatch_thread, __queue, __finished
 
-    __queue = Queue(maxsize=10000)
+    __queue = Queue(maxsize=config.max_buffer_size)
     __finished = Event()
     __heartbeat_thread = Thread(name='HeartbeatThread', target=__heartbeat, daemon=True)
     __report_thread = Thread(name='ReportThread', target=__report, daemon=True)
@@ -162,10 +168,6 @@ def started():
     return __started
 
 
-def connected():
-    return __protocol.connected()
-
-
 def archive(segment: 'Segment'):
     try:  # unlike checking __queue.full() then inserting, this is atomic
         __queue.put(segment, block=False)
diff --git a/skywalking/agent/protocol/__init__.py b/skywalking/agent/protocol/__init__.py
@@ -29,9 +29,6 @@ def fork_after_in_parent(self):
     def fork_after_in_child(self):
         pass
 
-    def connected(self):
-        return False
-
     def heartbeat(self):
         raise NotImplementedError()
 
diff --git a/skywalking/agent/protocol/grpc.py b/skywalking/agent/protocol/grpc.py
@@ -17,7 +17,7 @@
 
 import logging
 import traceback
-from queue import Queue, Empty, Full
+from queue import Queue, Empty
 from time import time
 
 import grpc
@@ -35,15 +35,13 @@
 
 class GrpcProtocol(Protocol):
     def __init__(self):
+        self.properties_sent = False
         self.state = None
 
         if config.force_tls:
-            self.channel = grpc.secure_channel(config.collector_address, grpc.ssl_channel_credentials(),
-                                               options=(('grpc.max_connection_age_grace_ms',
-                                                         1000 * config.GRPC_TIMEOUT),))
+            self.channel = grpc.secure_channel(config.collector_address, grpc.ssl_channel_credentials())
         else:
-            self.channel = grpc.insecure_channel(config.collector_address, options=(('grpc.max_connection_age_grace_ms',
-                                                 1000 * config.GRPC_TIMEOUT),))
+            self.channel = grpc.insecure_channel(config.collector_address)
 
         if config.authentication:
             self.channel = grpc.intercept_channel(
@@ -58,44 +56,42 @@ def __init__(self):
     def _cb(self, state):
         logger.debug('grpc channel connectivity changed, [%s -> %s]', self.state, state)
         self.state = state
-        if self.connected():
-            try:
-                self.service_management.send_instance_props()
-            except grpc.RpcError:
-                self.on_error()
 
     def query_profile_commands(self):
         logger.debug("query profile commands")
         self.profile_query.do_query()
 
     def heartbeat(self):
         try:
+            if not self.properties_sent:
+                self.service_management.send_instance_props()
+                self.properties_sent = True
+
             self.service_management.send_heart_beat()
+
         except grpc.RpcError:
             self.on_error()
 
-    def connected(self):
-        return self.state == grpc.ChannelConnectivity.READY
-
     def on_error(self):
         traceback.print_exc() if logger.isEnabledFor(logging.DEBUG) else None
         self.channel.unsubscribe(self._cb)
         self.channel.subscribe(self._cb, try_to_connect=True)
 
     def report(self, queue: Queue, block: bool = True):
         start = time()
-        segment = None
 
         def generator():
-            nonlocal segment
-
             while True:
                 try:
-                    timeout = max(0, config.QUEUE_TIMEOUT - int(time() - start))  # type: int
+                    timeout = config.QUEUE_TIMEOUT - int(time() - start)  # type: int
+                    if timeout <= 0:  # this is to make sure we exit eventually instead of being fed continuously
+                        return
                     segment = queue.get(block=block, timeout=timeout)  # type: Segment
                 except Empty:
                     return
 
+                queue.task_done()
+
                 logger.debug('reporting segment %s', segment)
 
                 s = SegmentObject(
@@ -137,16 +133,7 @@ def generator():
 
                 yield s
 
-                queue.task_done()
-
         try:
             self.traces_reporter.report(generator())
-
         except grpc.RpcError:
             self.on_error()
-
-            if segment:
-                try:
-                    queue.put(segment, block=False)
-                except Full:
-                    pass
diff --git a/skywalking/agent/protocol/http.py b/skywalking/agent/protocol/http.py
@@ -35,9 +35,6 @@ def fork_after_in_child(self):
         self.service_management.fork_after_in_child()
         self.traces_reporter.fork_after_in_child()
 
-    def connected(self):
-        return True
-
     def heartbeat(self):
         if not self.properties_sent:
             self.service_management.send_instance_props()
@@ -50,17 +47,19 @@ def report(self, queue: Queue, block: bool = True):
         def generator():
             while True:
                 try:
-                    timeout = max(0, config.QUEUE_TIMEOUT - int(time() - start))  # type: int
+                    timeout = config.QUEUE_TIMEOUT - int(time() - start)  # type: int
+                    if timeout <= 0:  # this is to make sure we exit eventually instead of being fed continuously
+                        return
                     segment = queue.get(block=block, timeout=timeout)  # type: Segment
                 except Empty:
                     return
 
+                queue.task_done()
+
                 logger.debug('reporting segment %s', segment)
 
                 yield segment
 
-                queue.task_done()
-
         try:
             self.traces_reporter.report(generator=generator())
         except Exception:
diff --git a/skywalking/agent/protocol/kafka.py b/skywalking/agent/protocol/kafka.py
@@ -18,6 +18,7 @@
 import logging
 from skywalking.loggings import logger, getLogger
 from queue import Queue, Empty
+from time import time
 
 from skywalking import config
 from skywalking.agent import Protocol
@@ -43,13 +44,20 @@ def heartbeat(self):
         self.service_management.send_heart_beat()
 
     def report(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:  # this is to make sure we exit eventually instead of being fed continuously
+                        return
                     segment = queue.get(block=block)  # type: Segment
                 except Empty:
                     return
 
+                queue.task_done()
+
                 logger.debug('reporting segment %s', segment)
 
                 s = SegmentObject(
@@ -91,6 +99,4 @@ def generator():
 
                 yield s
 
-                queue.task_done()
-
         self.traces_reporter.report(generator())
diff --git a/skywalking/client/grpc.py b/skywalking/client/grpc.py
@@ -60,7 +60,7 @@ def __init__(self, channel: grpc.Channel):
         self.report_stub = TraceSegmentReportServiceStub(channel)
 
     def report(self, generator):
-        self.report_stub.collect(generator, timeout=config.GRPC_TIMEOUT)
+        self.report_stub.collect(generator)
 
 
 class GrpcProfileTaskChannelService(ProfileTaskChannelService):
diff --git a/skywalking/config.py b/skywalking/config.py
@@ -22,10 +22,7 @@
 if TYPE_CHECKING:
     from typing import List
 
-# In order to prevent timeouts and possible segment loss make sure QUEUE_TIMEOUT is always at least few seconds lower
-# than GRPC_TIMEOUT.
-GRPC_TIMEOUT = 300  # type: int
-QUEUE_TIMEOUT = 240  # type: int
+QUEUE_TIMEOUT = 1  # type: int
 
 RE_IGNORE_PATH = re.compile('^$')  # type: re.Pattern
 
@@ -41,6 +38,7 @@
 authentication = os.getenv('SW_AGENT_AUTHENTICATION')  # type: str
 logging_level = os.getenv('SW_AGENT_LOGGING_LEVEL') or 'INFO'  # type: str
 disable_plugins = (os.getenv('SW_AGENT_DISABLE_PLUGINS') or '').split(',')  # type: List[str]
+max_buffer_size = int(os.getenv('SW_AGENT_MAX_BUFFER_SIZE', '1000'))  # type: int
 sql_parameters_length = int(os.getenv('SW_SQL_PARAMETERS_LENGTH') or '0')  # type: int
 pymongo_trace_parameters = True if os.getenv('SW_PYMONGO_TRACE_PARAMETERS') and \
                                    os.getenv('SW_PYMONGO_TRACE_PARAMETERS') == 'True' else False  # type: bool
diff --git a/skywalking/trace/context.py b/skywalking/trace/context.py
@@ -269,4 +269,4 @@ def get_context() -> SpanContext:
     if spans:
         return spans[len(spans) - 1].context
 
-    return SpanContext() if agent.connected() else NoopContext()
+    return SpanContext()
