Add ESF specific User-Agent header in outgoing Elasticsearch requests

CHANGELOG.md

@@ -1,3 +1,7 @@
+### v1.11.0 - 2023/11/27
+##### Features
+* Add user agent with information about ESF version and host environment: [#537](https://github.com/elastic/elastic-serverless-forwarder/pull/537)
+
 ### v1.10.0 - 2023/10/27
 ##### Features
 * Move `_id` field to `@metadata._id` in logstash output: [#507](https://github.com/elastic/elastic-serverless-forwarder/pull/507)

docs/en/aws-elastic-serverless-forwarder.asciidoc

@@ -11,7 +11,7 @@
 
 The Elastic Serverless Forwarder is an Amazon Web Services ({aws}) Lambda function that ships logs from your {aws} environment to Elastic.
 
-The Elastic Serverless Forwarder works with {stack} 7.16 and later.
+The Elastic Serverless Forwarder works with {stack} 7.17 and later.
 
 IMPORTANT: Using Elastic Serverless Forwarder may result in additional charges. To learn
 how to minimize additional charges, refer to <<preventing-unexpected-costs>>.

requirements-tests.txt

@@ -8,6 +8,6 @@ orjson==3.9.10
 pysimdjson==5.0.2
 python-rapidjson==1.13
 cysimdjson==23.8
-responses==0.23.3
+responses==0.24.1
 testcontainers==3.7.1
 pyOpenSSL==23.3.0

requirements.txt

@@ -1,9 +1,9 @@
 elastic-apm==6.19.0
-boto3==1.28.80
+boto3==1.29.7
 ecs_logging==2.1.0
-elasticsearch==7.16.3
+elasticsearch==7.17.9
 PyYAML==6.0.1
 aws_lambda_typing==2.18.0
 ujson==5.8.0
 requests==2.31.0
-urllib3==1.26.15
+urllib3==1.26.18

share/environment.py

@@ -0,0 +1,17 @@
+# Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+# or more contributor license agreements. Licensed under the Elastic License 2.0;
+# you may not use this file except in compliance with the Elastic License 2.0.
+
+import os
+import platform
+
+
+def is_aws() -> bool:
+    return os.getenv("AWS_EXECUTION_ENV") is not None
+
+
+def get_environment() -> str:
+    if is_aws():
+        return os.environ["AWS_EXECUTION_ENV"]
+    else:
+        return f"Python/{platform.python_version()} {platform.system()}/{platform.machine()}"

share/utils.py

@@ -2,7 +2,13 @@
 # or more contributor license agreements. Licensed under the Elastic License 2.0;
 # you may not use this file except in compliance with the Elastic License 2.0.
 import hashlib
+import sys
 
 
 def get_hex_prefix(src: str) -> str:
     return hashlib.sha3_384(src.encode("utf-8")).hexdigest()
+
+
+def create_user_agent(esf_version: str, environment: str = sys.version) -> str:
+    """Creates the 'User-Agent' header given ESF version and running environment"""
+    return f"ElasticServerlessForwarder/{esf_version} ({environment})"

share/version.py

@@ -0,0 +1,5 @@
+# Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+# or more contributor license agreements. Licensed under the Elastic License 2.0;
+# you may not use this file except in compliance with the Elastic License 2.0.
+
+version = "1.11.0"

shippers/es.py

@@ -10,7 +10,10 @@
 from elasticsearch.helpers import bulk as es_bulk
 from elasticsearch.serializer import Serializer
 
+import share.utils
 from share import json_dumper, json_parser, normalise_event, shared_logger
+from share.environment import get_environment
+from share.version import version
 
 from .shipper import EventIdGeneratorCallable, ReplayHandlerCallable
 
@@ -119,6 +122,9 @@ def _elasticsearch_client(**es_client_kwargs: Any) -> Elasticsearch:
         es_client_kwargs["max_retries"] = 4
         es_client_kwargs["http_compress"] = True
         es_client_kwargs["retry_on_timeout"] = True
+        es_client_kwargs["headers"] = {
+            "User-Agent": share.utils.create_user_agent(esf_version=version, environment=get_environment())
+        }
 
         return Elasticsearch(**es_client_kwargs)
 

tests/handlers/aws/test_integrations.py

@@ -74,7 +74,7 @@ def setUpClass(cls) -> None:
 
         lsc = LocalStackContainer(image="localstack/localstack:1.4.0")
         lsc.with_env("EAGER_SERVICE_LOADING", "1")
-        lsc.with_services("kinesis", "logs", "s3", "sqs", "secretsmanager")
+        lsc.with_services("kinesis", "logs", "s3", "sqs", "secretsmanager", "ec2")
         cls.localstack = lsc.start()
 
         session = boto3.Session(region_name=_AWS_REGION)

tests/share/test_environment.py

@@ -0,0 +1,17 @@
+# Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+# or more contributor license agreements. Licensed under the Elastic License 2.0;
+# you may not use this file except in compliance with the Elastic License 2.0.
+
+import os
+from unittest import mock
+
+import pytest
+
+from share.environment import get_environment
+
+
+@pytest.mark.unit
+@mock.patch.dict(os.environ, {"AWS_EXECUTION_ENV": "AWS_Lambda_Python3.9"})
+def test_aws_environment() -> None:
+    environment = get_environment()
+    assert environment == "AWS_Lambda_Python3.9"

tests/shippers/test_es.py

@@ -12,6 +12,9 @@
 import pytest
 from elasticsearch import SerializationError
 
+import share
+from share.environment import get_environment
+from share.version import version
 from shippers import ElasticsearchShipper, JSONSerializer
 
 _now = datetime.datetime.utcnow().strftime("%Y-%m-%dT%H:%M:%S.%fZ")
@@ -48,7 +51,7 @@
 
 class MockTransport(elasticsearch.Transport):
     def __init__(self, *args: Any, **kwargs: Any) -> None:
-        pass
+        self.kwargs = kwargs
 
 
 class MockClient(elasticsearch.Elasticsearch):
@@ -61,13 +64,20 @@ def __init__(self, *args: Any, **kwargs: Any) -> None:
 _failures = []
 
 
-def mock_bulk(client: Any, actions: list[dict[str, Any]], **kwargs: Any) -> tuple[int, list[dict[str, Any]]]:
+def mock_bulk(
+    client: elasticsearch.Elasticsearch, actions: list[dict[str, Any]], **kwargs: Any
+) -> tuple[int, list[dict[str, Any]]]:
     global _documents
     _documents = [actions]
+    assert client.transport.kwargs["headers"] == {
+        "User-Agent": share.utils.create_user_agent(esf_version=version, environment=get_environment())
+    }
     return len(actions), []
 
 
-def mock_bulk_failure(client: Any, actions: list[dict[str, Any]], **kwargs: Any) -> tuple[int, list[dict[str, Any]]]:
+def mock_bulk_failure(
+    client: elasticsearch.Elasticsearch, actions: list[dict[str, Any]], **kwargs: Any
+) -> tuple[int, list[dict[str, Any]]]:
     global _failures
     _failures = list(map(lambda action: {"create": {"_id": action["_id"], "error": "an error"}}, actions))
     return len(actions), _failures

tests/testcontainers/es.py

@@ -31,7 +31,7 @@ class ElasticsearchContainer(DockerContainer):  # type: ignore
     """
 
     _DEFAULT_IMAGE = "docker.elastic.co/elasticsearch/elasticsearch"
-    _DEFAULT_VERSION = "7.16.3"
+    _DEFAULT_VERSION = "7.17.9"
     _DEFAULT_PORT = 9200
     _DEFAULT_USERNAME = DEFAULT_USERNAME
     _DEFAULT_PASSWORD = DEFAULT_PASSWORD

tests/testcontainers/logstash.py

@@ -38,7 +38,7 @@ class LogstashContainer(DockerContainer):  # type: ignore
     """
 
     _DEFAULT_IMAGE = "docker.elastic.co/logstash/logstash"
-    _DEFAULT_VERSION = "7.16.0"
+    _DEFAULT_VERSION = "7.17.0"
     _DEFAULT_PORT = 5044
     _DEFAULT_API_PORT = 9600
     _DEFAULT_USERNAME = "USERNAME"