Open Telemetry Transport

dev-requirements-new.txt

@@ -12,6 +12,9 @@ python-dateutil==2.8.2
 repoze.lru==0.7
 six==1.14.0
 
+# Required for opentelemetry support.
+protobuf==3.17.3
+
 # Required for redis monitor.
 redis==2.10.5
 

scalyr_agent/client_auth.py

@@ -0,0 +1,78 @@
+import datetime
+import json
+import urllib
+from base64 import b64encode
+
+from urllib.parse import urlparse
+
+import requests
+
+
+class ClientAuth(object):
+    def __init__(self, configuration, headers):
+        self.configuration = configuration
+        self.headers = headers
+        if self.configuration.auth == "oauth2":
+            self.auth = OAuth2(self.configuration, self.headers)
+        elif self.configuration.auth == "bearer":
+            self.auth = BearerToken(self.configuration, self.headers)
+        else:
+            self.auth = NoAuth(self.configuration, self.headers)
+
+    def authenticate(self):
+        return self.auth.authenticate()
+
+class NoAuth(object):
+    def __init__(self, configuration, headers):
+        self.headers = headers
+
+    def authenticate(self):
+        # Add a header just so for traceability
+        self.headers["x-no-auth"]="true"
+        return True
+
+# Simple Authorization Header as Bearer Token using `api_key` configuration
+class BearerToken(object):
+    def __init__(self, configuration, headers):
+        headers.set("Authorization", "Bearer " + configuration.api_key)
+
+    def authenticate(self):
+        return True
+
+# Implement https://datatracker.ietf.org/doc/html/rfc6749#section-4.4 flow
+class OAuth2(object):
+    def __init__(self, configuration, headers):
+        self.headers = headers # Headers modified for external requests
+        self.client_id = configuration.client_id
+        self.client_secret = configuration.client_secret
+        self.token_url = configuration.token_url
+        scopes = " ".join(configuration.scopes)
+        # Payload Body for the token exchange
+        self.auth_request = "grant_type=client_credentials&scope=" + urllib.parse.quote_plus(scopes)
+        # Our token to use in requests
+        self.token = None
+        # When the token expires
+        self.expiry_time = datetime.datetime.now()
+        self.verify_ssl = configuration.verify_server_certificate
+        # Authentication Headers for the token exchange
+        authorization_header_value = b64encode( bytes(('' + self.client_id + ':' + self.client_secret).encode("utf-8"))).decode('utf-8')
+        self.auth_headers = { "Content-Type": "application/x-www-form-urlencoded", "Authorization" : "Basic " + authorization_header_value}
+
+    def authenticate(self):
+        if self.token == None or self.expiry_time < datetime.datetime.now():
+            print("Request/Refresh OAuth2 Token")
+            if not self.refresh_token():
+                raise Exception("OAuth2: Unable to refresh token")
+        self.headers["Authorization"]="Bearer " + self.token
+        return True
+
+    def refresh_token(self):
+        resp = requests.post(self.token_url, data=self.auth_request, headers=self.auth_headers, verify=self.verify_ssl)
+        if resp.status_code == 200:
+            auth_response = json.loads(resp.content)
+            self.token = auth_response["access_token"]
+            self.expiry_time = datetime.datetime.now() + datetime.timedelta(seconds=auth_response["expires_in"])
+            return True
+        else:
+            raise Exception("Unable to obtain OAuth2 Token: " + str(resp.status_code) + "(" + str(resp.content) + ")")
+        return False

scalyr_agent/configuration.py

@@ -765,6 +765,8 @@ def print_useful_settings(self, other_config=None):
             "default_sessions_per_worker",
             "default_worker_session_status_message_interval",
             "enable_worker_session_process_metrics_gather",
+            "server_url"
+            "transport",
             # NOTE: It's important we use sanitzed_ version of this method which masks the API key
             "sanitized_worker_configs",
         ]
@@ -1497,6 +1499,41 @@ def api_key(self):
         """Returns the configuration value for 'api_key'."""
         return self.__get_config().get_string("api_key")
 
+    @property
+    def server_url(self):
+        """Returns the configuration value for 'server_url'."""
+        return self.__get_config().get_string("server_url")
+
+    @property
+    def auth(self):
+        """Returns the configuration value for 'auth'."""
+        return self.__get_config().get_string("auth")
+
+    @property
+    def client_id(self):
+        """Returns the configuration value for 'client_id'."""
+        return self.__get_config().get_string("client_id")
+
+    @property
+    def client_secret(self):
+        """Returns the configuration value for 'client_secret'."""
+        return self.__get_config().get_string("client_secret")
+
+    @property
+    def token_url(self):
+        """Returns the configuration value for 'token_url'."""
+        return self.__get_config().get_string("token_url")
+
+    @property
+    def scopes(self):
+        """Returns the configuration value for 'scopes'."""
+        return self.__get_config().get_json_array("scopes")
+
+    @property
+    def transport(self):
+        """Returns the configuration value for 'transport'."""
+        return self.__get_config().get_string("transport")
+
     @property
     def scalyr_server(self):
         """Returns the configuration value for 'scalyr_server'."""
@@ -2826,6 +2863,27 @@ def __verify_main_config_and_apply_defaults(
             apply_defaults,
             env_aware=True,
         )
+        self.__verify_or_set_optional_string(
+            config,
+            "transport",
+            "",
+            description,
+            apply_defaults
+        )
+        self.__verify_or_set_optional_string(
+            config,
+            "server_url",
+            "",
+            description,
+            apply_defaults
+        )
+        self.__verify_or_set_optional_string(
+            config,
+            "auth",
+            "",
+            description,
+            apply_defaults
+        )
         self.__verify_or_set_optional_bool(
             config,
             "use_new_ingestion",

scalyr_agent/copying_manager/worker.py

@@ -212,7 +212,8 @@ def __init__(self, add_events_request, completion_callback):
         # If there is a AddEventsTask object already created for the next request due to pipelining, this is set to it.
         # This must be the next request if this request is successful, otherwise, we will lose bytes.
         self.next_pipelined_task = None
-
+        # last status
+        self.__receive_response_status = ()
 
 class CopyingManagerWorkerSessionInterface(six.with_metaclass(ABCMeta)):
     """
@@ -354,6 +355,7 @@ def __init__(self, configuration, worker_config_entry, session_id, is_daemon=Fal
 
         # The current pending AddEventsTask.  We will retry the contained AddEventsRequest several times.
         self.__pending_add_events_task = None
+        self.__receive_response_status = {}
 
         # The next LogFileProcessor that should have log lines read from it for transmission.
         self.__current_processor = 0
@@ -489,6 +491,7 @@ def run(self):
                         # on the ground and advance.
                         if current_time - last_success > self.__config.max_retry_time:
                             if self.__pending_add_events_task is not None:
+
                                 if (
                                     "parseResponseFailed"
                                     in self.__pending_add_events_task.__receive_response_status
@@ -1226,8 +1229,12 @@ def _init_scalyr_client(self, quiet=False):
         """
         api_key = self.__worker_config_entry["api_key"]
         if self.__config.use_new_ingestion:
-
             self.__new_scalyr_client = create_new_client(self.__config, api_key=api_key)
+        elif self.__config.transport == "otlp":
+            from scalyr_agent.otlp_client import (
+                create_otlp_client,
+            )
+            self.__scalyr_client = create_otlp_client(self.__config, self.__worker_config_entry)
         else:
             self.__scalyr_client = create_client(
                 self.__config,