Open Telemetry Transport

agent_build_refactored/container_images/image_builders.py

@@ -230,6 +230,8 @@ def create_agent_filesystem(
         for third_party_lib in third_party_lis_root.iterdir():
             if third_party_lib.name == "tcollector":
                 continue
+            if third_party_lib.name == "opentelemetry":
+                continue
             if third_party_lib.is_dir():
                 shutil.rmtree(third_party_lib)
             if third_party_lib.is_file():

agent_build_refactored/managed_packages/managed_packages_builders.py

@@ -848,6 +848,10 @@ def _build_package_root(self):
             SOURCE_ROOT / "scalyr_agent/third_party/tcollector",
             third_party_libs_dir / "tcollector",
         )
+        shutil.copytree(
+            SOURCE_ROOT / "scalyr_agent/third_party/opentelemetry",
+            third_party_libs_dir / "opentelemetry",
+        )
         shutil.copy2(
             SOURCE_ROOT / "scalyr_agent/third_party/__init__.py",
             third_party_libs_dir / "__init__.py",

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/agent_main.py

@@ -597,7 +597,7 @@ def __perform_config_checks(self, no_check_remote):
         )
 
         # Send a test message to the server to make sure everything works.  If not, print a decent error message.
-        if not no_check_remote or self.__config.use_new_ingestion:
+        if self.__config.transport == "scalyr" and (not no_check_remote or self.__config.use_new_ingestion):
             client = create_client(self.__config, quiet=True)
             try:
                 ping_result = client.ping()

scalyr_agent/client_auth.py

@@ -0,0 +1,97 @@
+import datetime
+import json
+import urllib
+from base64 import b64encode
+
+from urllib.parse import urlparse
+
+import scalyr_agent.scalyr_logging as scalyr_logging
+
+import requests
+
+log = scalyr_logging.getLogger(__name__)
+
+
+class ClientAuth(object):
+    def __init__(self, configuration, headers):
+        log.setLevel(configuration.debug_level)
+        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)
+        elif self.configuration.auth == "basic":
+            self.auth = Basic(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-custom-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
+
+class Basic(object):
+    def __init__(self, configuration, headers):
+        authorization_header_value = b64encode(
+            bytes(('' + configuration.basic_username + ":" + configuration.basic_password).encode("utf-8"))).decode('utf-8')
+        headers["Authorization"]="Basic " + authorization_header_value
+
+    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.oauth_client_id
+        self.client_secret = configuration.oauth_client_secret
+        self.token_url = configuration.oauth_token_url
+        scopes = " ".join(configuration.oauth_scopes)
+        # Payload Body for the token exchange
+        self.auth_request = "grant_type=client_credentials&scope=" + urllib.parse.quote(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():
+            log.info("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):
+        log.log(scalyr_logging.DEBUG_LEVEL_1, "OAuth2 Token Request to %s, body: %s, headers: %s" % (self.token_url, self.auth_request, self.headers))
+        resp = requests.post(self.token_url, data=self.auth_request, headers=self.auth_headers, verify=self.verify_ssl)
+        if resp.status_code == 200:
+            log.log(scalyr_logging.DEBUG_LEVEL_1, "OAUTH Response: %s" % (resp.content))
+            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"])
+            log.log(scalyr_logging.DEBUG_LEVEL_1, "OAUTH Token: %s (expires: %s)" % (self.token, self.expiry_time))
+            return True
+        else:
+            raise Exception("Unable to obtain OAuth2 Token: " + str(resp.status_code) + "(" + str(resp.content) + ")")
+        return False

scalyr_agent/configuration.py

@@ -766,6 +766,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",
         ]
@@ -1498,6 +1500,51 @@ 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 oauth_client_id(self):
+        """Returns the configuration value for 'oauth_client_id'."""
+        return self.__get_config().get_string("oauth_client_id")
+
+    @property
+    def oauth_client_secret(self):
+        """Returns the configuration value for 'oauth_client_secret'."""
+        return self.__get_config().get_string("oauth_client_secret")
+
+    @property
+    def oauth_token_url(self):
+        """Returns the configuration value for 'oauth_token_url'."""
+        return self.__get_config().get_string("oauth_token_url")
+
+    @property
+    def oauth_scopes(self):
+        """Returns the configuration value for 'oauth_scopes'."""
+        return self.__get_config().get_json_array("oauth_scopes")
+
+    @property
+    def basic_username(self):
+        """Returns the configuration value for 'basic_username'."""
+        return self.__get_config().get_string("basic_username")
+
+    @property
+    def basic_password(self):
+        """Returns the configuration value for 'basic_password'."""
+        return self.__get_config().get_string("basic_password")
+
+    @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'."""
@@ -2179,6 +2226,10 @@ def __set_api_key(self, config, api_key):
         if api_key:
             config.put("api_key", api_key)
 
+        # Ingore api_key check if we are not writing to Scalyr/DataSet/SDL
+        if "transport" in config and config.get_string("transport") != "scalyr":
+            return
+
         if "api_key" not in config:
             raise BadConfiguration(
                 'The configuration file is missing the required field "api_key" that '
@@ -2264,6 +2315,36 @@ def __verify_main_config_and_apply_defaults(
         self.__verify_or_set_optional_string(
             config, "api_key", "", description, apply_defaults, env_aware=True
         )
+        self.__verify_or_set_optional_string(
+            config, "transport", "scalyr", description, apply_defaults, env_aware=True
+        )
+        self.__verify_or_set_optional_string(
+            config, "auth", "", description, apply_defaults, env_aware=True
+        )
+        self.__verify_or_set_optional_string(
+            config, "oauth_client_id", "", description, apply_defaults, env_aware=True
+        )
+        self.__verify_or_set_optional_string(
+            config, "oauth_client_secret", "", description, apply_defaults, env_aware=True
+        )
+        self.__verify_or_set_optional_string(
+            config, "oauth_token_url", "", description, apply_defaults, env_aware=True
+        )
+        self.__verify_or_set_optional_array_of_strings(
+            config,
+            "oauth_scopes",
+            [],
+            description,
+            apply_defaults,
+            separators=[None, ","],
+            env_aware=True,
+        )
+        self.__verify_or_set_optional_string(
+            config, "basic_username", "", description, apply_defaults, env_aware=True
+        )
+        self.__verify_or_set_optional_string(
+            config, "basic_password", "", description, apply_defaults, env_aware=True
+        )
         self.__verify_or_set_optional_bool(
             config, "allow_http", False, description, apply_defaults, env_aware=True
         )
@@ -2827,6 +2908,27 @@ def __verify_main_config_and_apply_defaults(
             apply_defaults,
             env_aware=True,
         )
+        self.__verify_or_set_optional_string(
+            config,
+            "transport",
+            "scalyr",
+            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",
@@ -4200,10 +4302,14 @@ def __verify_workers_entry_and_set_defaults(self, worker_entry, entry_index=None
                 worker_entry, "api_key", description % entry_index
             )
 
+        # Only use scalyr_server if our transport is scalyr (but still use it if we don't get a server_url)
+        default_server_url = self.scalyr_server
+        if self.transport != "scalyr" and self.server_url is not None and self.server_url != "":
+            default_server_url = self.server_url
         self.__verify_or_set_optional_string(
             worker_entry,
             "server_url",
-            default_value=self.scalyr_server,
+            default_value=default_server_url,
             config_description=description % entry_index,
         )
 

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)):
     """
@@ -489,9 +490,10 @@ 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
+                                    in self.__pending_add_events_task.receive_response_status
                                 ):
                                     log.error(
                                         "Repeatedly failed to parse response due to exception.  Dropping events",
@@ -586,7 +588,7 @@ def run(self):
                                 full_response = ""
                             else:
                                 (result, bytes_sent, full_response) = get_response()
-                                self.__pending_add_events_task.__receive_response_status = (
+                                self.__pending_add_events_task.receive_response_status = (
                                     result
                                 )
                             blocking_response_time_end = time.time()
@@ -1226,8 +1228,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,