Add OpAMP agent

.github/workflows/ci.yml

@@ -65,7 +65,7 @@ jobs:
   test:
     runs-on: ubuntu-latest
     env:
-      py39: 3.9
+      py39: "3.9"
       py310: "3.10"
       py311: "3.11"
       py312: "3.12"
@@ -82,4 +82,16 @@ jobs:
         python-version: ${{ env[matrix.python-version] }}
         architecture: "x64"
     - run: pip install -r dev-requirements.txt
+    - name: run recorded tests with python 3.10+ where urllib3 2.x is supported
+      run: pip install pytest-vcr
+      if: ${{ matrix.python-version != 'py39' }}
     - run: pytest --with-integration-tests
+
+  typecheck:
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v4
+      - uses: ./.github/actions/env-install
+      - run: pip install -r dev-requirements.txt
+      - run: pip install pyright
+      - run: pyright

dev-requirements.txt

@@ -112,7 +112,7 @@ pyproject-hooks==1.2.0
     # via
     #   build
     #   pip-tools
-pytest==8.4.0
+pytest==8.4.1
     # via elastic-opentelemetry (pyproject.toml)
 requests==2.32.4
     # via
@@ -132,8 +132,10 @@ typing-extensions==4.14.0
     #   opentelemetry-resourcedetector-gcp
     #   opentelemetry-sdk
     #   opentelemetry-semantic-conventions
-urllib3==2.4.0
+urllib3==2.5.0
     # via requests
+uuid-utils==0.11.0
+    # via elastic-opentelemetry (pyproject.toml)
 wheel==0.45.1
     # via pip-tools
 wrapt==1.17.2

opamp-gen-requirements.txt

@@ -0,0 +1,5 @@
+# Use caution when bumping this version to ensure compatibility with the currently supported protobuf version.
+# Pinning this to the oldest grpcio version that supports protobuf 5 helps avoid RuntimeWarning messages
+# from the generated protobuf code and ensures continued stability for newer grpcio versions.
+grpcio-tools==1.63.2
+mypy-protobuf~=3.5.0

pyproject.toml

@@ -38,10 +38,11 @@ dependencies = [
     "opentelemetry-sdk-extension-aws ~= 2.1.0",
     "opentelemetry-semantic-conventions == 0.55b1",
     "packaging",
+    "uuid-utils",
 ]
 
 [project.optional-dependencies]
-dev = ["pytest", "pip-tools", "oteltest==0.24.0", "leb128"]
+dev = ["pytest", "pip-tools", "oteltest==0.24.0", "leb128", "pytest-vcr ; python_version > '3.9'"]
 
 [project.entry-points.opentelemetry_configurator]
 configurator = "elasticotel.distro:ElasticOpenTelemetryConfigurator"
@@ -86,9 +87,26 @@ build-backend = "setuptools.build_meta"
 [tool.ruff]
 target-version = "py38"
 line-length = 120
+extend-exclude = [
+  "*_pb2*.py*",
+]
 
 [tool.ruff.lint.isort]
 known-third-party = [
   "opentelemetry",
 ]
 known-first-party = ["elasticotel"]
+
+[tool.pyright]
+typeCheckingMode = "standard"
+pythonVersion = "3.9"
+
+include = [
+  "src/elasticotel",
+  "src/opentelemetry",
+]
+
+exclude = [
+  "**/__pycache__",
+  "src/opentelemetry/_opamp/proto",
+]

scripts/license_headers_check.sh

@@ -17,14 +17,15 @@
 
 if [ $# -eq 0 ]
 then
-   FILES=$(find . \( -name "*.py" -o -name "*.c" -o -name "*.sh" \) -size +1c -not -path "./dist/*" -not -path "./build/*" -not -path "./venv*/*")
+    FILES=$(git ls-files | grep -e "\.py$" -e "\.c$" -e "\.sh$" | grep -v -e "/proto/" | xargs -r -d'\n' -I{} find {} -size +1c)
 else
     FILES=$@
 fi
 
 LICENSE_HEADER="Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one"
+UPSTREAM_LICENSE_HEADER="Copyright The OpenTelemetry Authors"
 
-MISSING=$(grep --files-without-match "$LICENSE_HEADER" ${FILES})
+MISSING=$(grep --files-without-match -e "$LICENSE_HEADER" -e "$UPSTREAM_LICENSE_HEADER" ${FILES})
 
 if [ -z "$MISSING" ]
 then

scripts/opamp_proto_codegen.sh

@@ -0,0 +1,81 @@
+#!/bin/bash
+# Copyright The OpenTelemetry Authors
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+# Regenerate python code from opamp protos in
+# https://github.com/open-telemetry/opamp-spec
+#
+# To use, update OPAMP_SPEC_REPO_BRANCH_OR_COMMIT variable below to a commit hash or
+# tag in opentelemtry-proto repo that you want to build off of. Then, just run
+# this script to update the proto files. Commit the changes as well as any
+# fixes needed in the OTLP exporter.
+#
+# Optional envars:
+#   OPAMP_SPEC_REPO_DIR - the path to an existing checkout of the opamp-spec repo
+
+# Pinned commit/branch/tag for the current version used in the opamp python package.
+OPAMP_SPEC_REPO_BRANCH_OR_COMMIT="v0.12.0"
+
+set -e
+
+OPAMP_SPEC_REPO_DIR=${OPAMP_SPEC_REPO_DIR:-"/tmp/opamp-spec"}
+# root of opentelemetry-python repo
+repo_root="$(git rev-parse --show-toplevel)"
+venv_dir="/tmp/opamp_proto_codegen_venv"
+proto_output_dir="$repo_root/src/opentelemetry/_opamp/proto"
+
+# run on exit even if crash
+cleanup() {
+    echo "Deleting $venv_dir"
+    rm -rf $venv_dir
+}
+trap cleanup EXIT
+
+echo "Creating temporary virtualenv at $venv_dir using $(python3 --version)"
+python3 -m venv $venv_dir
+source $venv_dir/bin/activate
+python -m pip install \
+    -c $repo_root/opamp-gen-requirements.txt \
+    grpcio-tools mypy-protobuf
+echo 'python -m grpc_tools.protoc --version'
+python -m grpc_tools.protoc --version
+
+# Clone the proto repo if it doesn't exist
+if [ ! -d "$OPAMP_SPEC_REPO_DIR" ]; then
+    git clone https://github.com/open-telemetry/opamp-spec.git $OPAMP_SPEC_REPO_DIR
+fi
+
+# Pull in changes and switch to requested branch
+(
+    cd $OPAMP_SPEC_REPO_DIR
+    git fetch --all
+    git checkout $OPAMP_SPEC_REPO_BRANCH_OR_COMMIT
+    # pull if OPAMP_SPEC_BRANCH_OR_COMMIT is not a detached head
+    git symbolic-ref -q HEAD && git pull --ff-only || true
+)
+
+cd $proto_output_dir
+
+# clean up old generated code
+find . -regex ".*_pb2.*\.pyi?" -exec rm {} +
+
+# generate proto code for all protos
+all_protos=$(find $OPAMP_SPEC_REPO_DIR/ -name "*.proto")
+python -m grpc_tools.protoc \
+    -I $OPAMP_SPEC_REPO_DIR/proto \
+    --python_out=. \
+    --mypy_out=. \
+    $all_protos
+
+sed -i -e 's/import anyvalue_pb2 as anyvalue__pb2/from . import anyvalue_pb2 as anyvalue__pb2/' opamp_pb2.py

src/elasticotel/distro/__init__.py

@@ -16,6 +16,7 @@
 
 import logging
 import os
+from urllib.parse import urlparse, urlunparse
 
 from opentelemetry.environment_variables import (
     OTEL_LOGS_EXPORTER,
@@ -35,17 +36,56 @@
     OTEL_EXPORTER_OTLP_METRICS_TEMPORALITY_PREFERENCE,
     OTEL_EXPORTER_OTLP_PROTOCOL,
 )
+from opentelemetry.sdk.resources import OTELResourceDetector
 from opentelemetry.util._importlib_metadata import EntryPoint
+from opentelemetry._opamp.agent import OpAMPAgent
+from opentelemetry._opamp.client import OpAMPClient
+from opentelemetry._opamp.proto import opamp_pb2 as opamp_pb2
 
-from elasticotel.distro.environment_variables import ELASTIC_OTEL_SYSTEM_METRICS_ENABLED
+from elasticotel.distro.environment_variables import ELASTIC_OTEL_OPAMP_ENDPOINT, ELASTIC_OTEL_SYSTEM_METRICS_ENABLED
 from elasticotel.distro.resource_detectors import get_cloud_resource_detectors
+from elasticotel.distro.config import opamp_handler
 
 
 logger = logging.getLogger(__name__)
 
 
 class ElasticOpenTelemetryConfigurator(_OTelSDKConfigurator):
-    pass
+    def _configure(self, **kwargs):
+        super()._configure(**kwargs)
+
+        enable_opamp = False
+        endpoint = os.environ.get(ELASTIC_OTEL_OPAMP_ENDPOINT)
+        if endpoint:
+            parsed = urlparse(endpoint)
+            enable_opamp = parsed.scheme in ("http", "https") and parsed.netloc
+            if enable_opamp:
+                if not parsed.path:
+                    parsed = parsed._replace(path="/v1/opamp")
+
+                endpoint_url = urlunparse(parsed)
+                # this is not great but we don't have the calculated resource attributes around
+                resource = OTELResourceDetector().detect()
+                agent_identifying_attributes = {
+                    "service.name": resource.attributes.get("service.name"),
+                }
+                if deployment_environment_name := resource.attributes.get(
+                    "deployment.environment.name", resource.attributes.get("deployment.environment")
+                ):
+                    agent_identifying_attributes["deployment.environment.name"] = deployment_environment_name
+
+                opamp_client = OpAMPClient(
+                    endpoint=endpoint_url,
+                    agent_identifying_attributes=agent_identifying_attributes,
+                )
+                opamp_agent = OpAMPAgent(
+                    interval=30,
+                    message_handler=opamp_handler,
+                    client=opamp_client,
+                )
+                opamp_agent.start()
+            else:
+                logger.warning("Found invalid value for OpAMP endpoint")
 
 
 class ElasticOpenTelemetryDistro(BaseDistro):
@@ -63,7 +103,7 @@ def load_instrumentor(self, entry_point: EntryPoint, **kwargs):
                 instrumentor_kwargs["config"] = {
                     k: v for k, v in SYSTEM_METRICS_DEFAULT_CONFIG.items() if k.startswith("process.runtime")
                 }
-        instrumentor_class(**instrumentor_kwargs).instrument(**kwargs)
+        instrumentor_class(**instrumentor_kwargs).instrument(**kwargs)  # type: ignore[reportCallIssue]
 
     def _configure(self, **kwargs):
         os.environ.setdefault(OTEL_TRACES_EXPORTER, "otlp")

src/elasticotel/distro/config.py

@@ -0,0 +1,57 @@
+# Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+# or more contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright
+# ownership. Elasticsearch B.V. licenses this file to you under
+# the Apache License, Version 2.0 (the "License"); you may
+# not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import logging
+
+from opentelemetry._opamp import messages
+from opentelemetry._opamp.client import OpAMPClient
+from opentelemetry._opamp.proto import opamp_pb2 as opamp_pb2
+
+
+logger = logging.getLogger(__name__)
+
+_LOG_LEVELS_MAP = {
+    "trace": 5,
+    "debug": logging.DEBUG,
+    "info": logging.INFO,
+    "warn": logging.WARNING,
+    "error": logging.ERROR,
+    "fatal": logging.CRITICAL,
+    "off": 1000,
+}
+
+
+def opamp_handler(client: OpAMPClient, message: opamp_pb2.ServerToAgent):
+    if not message.remote_config:
+        return
+
+    for config_filename, config in messages._decode_remote_config(message.remote_config):
+        # we don't have standardized config values so limit to configs coming from our backend
+        if config_filename == "elastic":
+            logger.debug("Config %s: %s", config_filename, config)
+            # when config option has default value you don't get it so need to handle the default
+            config_logging_level = config.get("logging_level")
+            if config_logging_level is not None:
+                logging_level = _LOG_LEVELS_MAP.get(config_logging_level)  # type: ignore[reportArgumentType]
+            else:
+                logging_level = logging.INFO
+
+            if logging_level is None:
+                logger.warning("Logging level not handled: %s", config_logging_level)
+            else:
+                # update upstream and distro logging levels
+                logging.getLogger("opentelemetry").setLevel(logging_level)
+                logging.getLogger("elasticotel").setLevel(logging_level)

src/elasticotel/distro/environment_variables.py

@@ -22,3 +22,12 @@
 
 **Default value:** ``false``
 """
+
+ELASTIC_OTEL_OPAMP_ENDPOINT = "ELASTIC_OTEL_OPAMP_ENDPOINT"
+"""
+.. envvar:: ELASTIC_OTEL_OPAMP_ENDPOINT
+
+OpAMP Endpoint URL.
+
+**Default value:** ``not set``
+"""

src/opentelemetry/_opamp/README.md

@@ -0,0 +1,7 @@
+# opamp
+
+opamp is an OpAMP protocol implementation.
+
+Implementation tries to be agnostic to the transport libraries and protocols used but since it's only HTTP for now that
+may be achieved once more transport implementation appears.
+

src/opentelemetry/_opamp/__init__.py

@@ -0,0 +1 @@
+