API - Add AGENT_HTTP Transport (#8)

API - Add AGENT_HTTP Transport

Author: jaygel179

requirements.txt

@@ -1,3 +1,4 @@
+flake8
 mock==2.0.0
 protobuf==3.9.1
 pytest==4.3.0

setup.cfg

@@ -11,13 +11,16 @@ exclude =
     README.md,
     LICENSE.md,
     requirements.txt,
+    *protos*,
 
 
 [coverage:run]
 include =
     stackify/*
 omit =
-    *tests*
+    *tests*,
+    *handler_backport.py,
+    *protos*,
 
 [tool:pytest]
 python_files=tests.py test.py test_*.py *_test.py tests_*.py *_tests.py

stackify/constants.py

@@ -7,8 +7,9 @@
 
 # using `%2F` instead of `/` as per package documentation
 DEFAULT_SOCKET_FILE = '%2Fusr%2Flocal%2Fstackify%2Fstackify.sock'
+DEFAULT_HTTP_ENDPOINT = 'https://localhost:10601'
 SOCKET_URL = 'http+unix://' + DEFAULT_SOCKET_FILE
-SOCKET_LOG_URL = '/log'
+AGENT_LOG_URL = '/log'
 
 API_REQUEST_INTERVAL_IN_SEC = 30
 
@@ -36,3 +37,4 @@
 
 TRANSPORT_TYPE_DEFAULT = 'default'
 TRANSPORT_TYPE_AGENT_SOCKET = 'agent_socket'
+TRANSPORT_TYPE_AGENT_HTTP = 'agent_http'

stackify/handler.py

@@ -16,7 +16,7 @@
 from stackify.constants import MAX_BATCH
 from stackify.constants import QUEUE_SIZE
 from stackify.timer import RepeatedTimer
-from stackify.transport import Transport
+from stackify.transport import configure_transport
 
 
 internal_logger = logging.getLogger(__name__)
@@ -70,7 +70,7 @@ def __init__(self, queue_, max_batch=MAX_BATCH, config=None, **kwargs):
 
         self.max_batch = max_batch
         self.messages = []
-        self.transport = Transport(config, **kwargs)
+        self.transport = configure_transport(config, **kwargs)
         self.timer = RepeatedTimer(API_REQUEST_INTERVAL_IN_SEC, self.send_group)
 
         self._started = False

stackify/timer.py

@@ -27,10 +27,12 @@ def _time(self):
 
     def start(self):
         if not self._started:
+            self._started = True
             self.thread.setDaemon(True)
             self.thread.start()
 
     def stop(self):
         if self._started:
+            self._started = False
             self.event.set()
             self.thread.join()

stackify/transport/__init__.py

@@ -1,17 +1,14 @@
 import logging
 
-from stackify.constants import LOG_SAVE_URL
-from stackify.constants import SOCKET_LOG_URL
+from stackify.constants import TRANSPORT_TYPE_AGENT_HTTP
 from stackify.constants import TRANSPORT_TYPE_AGENT_SOCKET
 from stackify.constants import TRANSPORT_TYPE_DEFAULT
-from stackify.transport.agent import AgentSocket
-from stackify.transport.agent.message import Log
-from stackify.transport.agent.message import LogGroup
+from stackify.transport.agent import AgentSocketTransport
+from stackify.transport.agent import AgentHTTPTransport
 from stackify.transport.application import get_configuration
 from stackify.transport.application import EnvironmentDetail
-from stackify.transport.default import HTTPClient
-from stackify.transport.default.log import LogMsg
-from stackify.transport.default.log import LogMsgGroup
+from stackify.transport.default import DefaultTransport
+
 
 internal_logger = logging.getLogger(__name__)
 
@@ -24,93 +21,34 @@ class TransportTypes(object):
     Types:
     * DEFAULT - HTTP transport that will directly send logs to the Platform
     * AGENT_SOCKET - HTTP warapped Unix Socket Domain that will send logs to the StackifyAgent
+    * AGENT_HTTP - HTTP transport that will send logs to the Agent using HTTP requests
     """
 
     DEFAULT = TRANSPORT_TYPE_DEFAULT
     AGENT_SOCKET = TRANSPORT_TYPE_AGENT_SOCKET
+    AGENT_HTTP = TRANSPORT_TYPE_AGENT_HTTP
 
     @classmethod
     def get_transport(self, api_config=None, env_details=None):
         # determine which transport to use depening on users config
         if api_config.transport == self.AGENT_SOCKET:
             internal_logger.debug('Setting Agent Socket Transport.')
-            api_config.transport = self.AGENT_SOCKET
-            return AgentSocket()
+            return AgentSocketTransport(api_config, env_details)
+
+        if api_config.transport == self.AGENT_HTTP:
+            internal_logger.debug('Setting Agent HTTP Transport.')
+            return AgentHTTPTransport(api_config, env_details)
 
         internal_logger.debug('Setting Default Transport.')
         api_config.transport = self.DEFAULT
-        return HTTPClient(api_config, env_details)
-
-    @classmethod
-    def create_message(self, record, api_config, env_details):
-        # create message depending on which transport
-        if api_config.transport == self.AGENT_SOCKET:
-            return Log(record, api_config, env_details).get_object()
-
-        msg = LogMsg()
-        msg.from_record(record)
-        return msg
-
-    @classmethod
-    def create_group_message(self, messages, api_config, env_details):
-        # create group message depending on which transport
-        if api_config.transport == self.AGENT_SOCKET:
-            return LogGroup(messages, api_config, env_details).get_object()
-
-        return LogMsgGroup(messages)
-
-    @classmethod
-    def get_log_url(self, api_config):
-        # return log url depending on which transport
-        if api_config.transport == self.AGENT_SOCKET:
-            return api_config.socket_url + SOCKET_LOG_URL
-
-        return LOG_SAVE_URL
-
-    @classmethod
-    def prepare_message(self, api_config, message):
-        # convert message depending on which transport
-        if api_config.transport == self.AGENT_SOCKET:
-            return message.SerializeToString()
-
-        return message
-
-
-class Transport(object):
-    """
-    Transport base class
-    """
-
-    def __init__(self, config=None, **kwargs):
-        self.api_config = config or get_configuration(**kwargs)
-        self.env_details = EnvironmentDetail(self.api_config)
-        self._transport = TransportTypes.get_transport(
-            self.api_config,
-            self.env_details,
-        )
-
-    def create_message(self, record):
-        # create message from record
-        return TransportTypes.create_message(
-            record,
-            self.api_config,
-            self.env_details,
-        )
+        return DefaultTransport(api_config, env_details)
 
-    def create_group_message(self, messages):
-        # create group message from list of records
-        return TransportTypes.create_group_message(
-            messages,
-            self.api_config,
-            self.env_details,
-        )
 
-    def send(self, group_message):
-        # send group message
-        try:
-            self._transport.send(
-                TransportTypes.get_log_url(self.api_config),
-                TransportTypes.prepare_message(self.api_config, group_message),
-            )
-        except Exception as e:
-            internal_logger.error('Request error: {}'.format(e))
+def configure_transport(config=None, **kwargs):
+    # return which transport to use depending on users input
+    api_config = config or get_configuration(**kwargs)
+    env_details = EnvironmentDetail(api_config)
+    return TransportTypes.get_transport(
+        api_config,
+        env_details,
+    )

stackify/transport/agent/__init__.py

@@ -1 +1,30 @@
-from .agent_socket import AgentSocket  # noqa
+import logging
+
+from stackify.constants import AGENT_LOG_URL
+from stackify.transport.agent import agent_http
+from stackify.transport.agent import agent_socket
+from stackify.transport.base import AgentBaseTransport
+
+internal_logger = logging.getLogger(__name__)
+
+
+class AgentSocketTransport(AgentBaseTransport):
+    """
+    Agent Socket Transport handles sending of logs using Unix Socket Domain
+    """
+
+    def __init__(self, api_config, env_details):
+        super(AgentSocketTransport, self).__init__(api_config, env_details)
+        self.url = api_config.socket_url + AGENT_LOG_URL
+        self._transport = agent_socket.AgentSocket()
+
+
+class AgentHTTPTransport(AgentBaseTransport):
+    """
+    Agent HTTP Transport handles sending of logs using HTTP requests
+    """
+
+    def __init__(self, api_config, env_details):
+        super(AgentHTTPTransport, self).__init__(api_config, env_details)
+        self.url = api_config.http_endpoint + AGENT_LOG_URL
+        self._transport = agent_http.AgentHTTP()

stackify/transport/agent/agent_http.py

@@ -0,0 +1,26 @@
+import logging
+import requests
+import retrying
+
+internal_logger = logging.getLogger(__name__)
+
+
+class AgentHTTP(object):
+    """
+    AgentHTTP class that handles HTTP post requests
+    """
+
+    def _post(self, url, payload):
+        headers = {
+            'Content-Type': 'application/x-protobuf',
+        }
+        try:
+            return requests.post(url, payload, headers=headers)
+        except Exception as e:
+            internal_logger.debug('HTTP transport exception: {}.'.format(e))
+            raise
+
+    @retrying.retry(wait_exponential_multiplier=1000, stop_max_delay=32000)
+    def send(self, url, payload):
+        # send payload through socket domain using _post method
+        return self._post(url, payload)

stackify/transport/agent/agent_socket.py

@@ -42,4 +42,4 @@ def _post(self, url, payload):
     @retrying.retry(wait_exponential_multiplier=1000, stop_max_delay=32000)
     def send(self, url, payload):
         # send payload through socket domain using _post method
-        self._post(url, payload)
+        return self._post(url, payload)

stackify/transport/agent/message.py

@@ -122,4 +122,4 @@ def __init__(self, messages, api_config, env_details, logger=None):
         log_group.application_location = env_details.appLocation
         log_group.logger = logger or __name__
         log_group.platform = 'python'
-        log_group.logs.MergeFrom(messages)
+        log_group.logs.extend(messages)