You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@skywalking.apache.org by ke...@apache.org on 2023/02/10 07:39:46 UTC
[skywalking-python] branch master updated: Periodically report instance properties (#279)
This is an automated email from the ASF dual-hosted git repository.
kezhenxu94 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/skywalking-python.git
The following commit(s) were added to refs/heads/master by this push:
new d756d69 Periodically report instance properties (#279)
d756d69 is described below
commit d756d69b1529e83c26e86a8bb1cf6f97a41f642c
Author: Superskyyy (AWAY, busy graduating | Debug 人) <Su...@outlook.com>
AuthorDate: Fri Feb 10 02:39:40 2023 -0500
Periodically report instance properties (#279)
---
CHANGELOG.md | 1 +
Makefile | 2 +-
docs/en/setup/Configuration.md | 5 +-
skywalking/agent/__init__.py | 32 +-------
skywalking/agent/protocol/__init__.py | 31 +++-----
skywalking/agent/protocol/grpc.py | 7 +-
skywalking/agent/protocol/http.py | 19 +++--
skywalking/agent/protocol/kafka.py | 12 ++-
skywalking/client/__init__.py | 105 +++++++++++++++++++++++++--
skywalking/client/grpc.py | 26 +++----
skywalking/client/http.py | 27 ++-----
skywalking/client/kafka.py | 14 ++--
skywalking/config.py | 32 ++++++--
tests/e2e/base/docker-compose.base.yml | 1 +
tests/e2e/case/expected/service-instance.yml | 25 ++++---
tools/config_doc_gen.py | 4 +-
16 files changed, 213 insertions(+), 130 deletions(-)
diff --git a/CHANGELOG.md b/CHANGELOG.md
index 1f2f5ab..35106d6 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -41,6 +41,7 @@
- Avoid reporting meaningless tracecontext with logs when there's no active span, UI will now show empty traceID (#272)
- Fix exception handler in profile_context (#273)
- Add namespace suffix to service name (#275)
+ - Add periodical instance property report to prevent data loss (#279)
- Docs:
- New documentation on how to test locally (#222)
diff --git a/Makefile b/Makefile
index 6012402..ea6ebd0 100644
--- a/Makefile
+++ b/Makefile
@@ -27,7 +27,7 @@ endif
.PHONY: env
env: poetry gen
- poetry install
+ poetry install --all-extras
poetry run pip install --upgrade pip
.PHONY: poetry poetry-fallback
diff --git a/docs/en/setup/Configuration.md b/docs/en/setup/Configuration.md
index 24ee266..d9f28d9 100644
--- a/docs/en/setup/Configuration.md
+++ b/docs/en/setup/Configuration.md
@@ -36,8 +36,9 @@ export SW_AGENT_YourConfiguration=YourValue
| Configuration | Environment Variable | Type | Default Value | Description |
| :------------ | :------------ | :------------ | :------------ | :------------ |
| heartbeat_period | SW_AGENT_HEARTBEAT_PERIOD | <class 'int'> | 30 | The agent will exchange heartbeat message with SkyWalking OAP backend every `period` seconds |
-| service_instance_property_report_factor | SW_AGENT_SERVICE_INSTANCE_PROPERTY_REPORT_FACTOR | <class 'int'> | 10 | The agent will report service instance properties every `factor * heartbeat period` seconds default: 10*30 = 300 seconds (TODO) |
-| experimental_fork_support | SW_AGENT_EXPERIMENTAL_FORK_SUPPORT | <class 'bool'> | False | The agent will try to restart itself in any os.fork()-ed child process. Important note: it's not suitable for large numbered, short-lived processes such as multiprocessing.Pool, as each one will introduce overhead and create numerous instances in SkyWalking dashboard in format of `service_instance-child-<pid>` (TODO) |
+| collector_properties_report_period_factor | SW_AGENT_COLLECTOR_PROPERTIES_REPORT_PERIOD_FACTOR | <class 'int'> | 10 | The agent will report service instance properties every `factor * heartbeat period` seconds default: 10*30 = 300 seconds |
+| instance_properties_json | SW_AGENT_INSTANCE_PROPERTIES_JSON | <class 'str'> | | A custom JSON string to be reported as service instance properties, e.g. `{"key": "value"}` |
+| experimental_fork_support | SW_AGENT_EXPERIMENTAL_FORK_SUPPORT | <class 'bool'> | False | The agent will try to restart itself in any os.fork()-ed child process. Important Note: it's not suitable for short-lived processes as each one will introduce overhead and create a new instance in SkyWalking dashboard in format of `service_instance-child-<pid>` (TODO) |
| queue_timeout | SW_AGENT_QUEUE_TIMEOUT | <class 'int'> | 1 | DANGEROUS - This option controls the interval of each bulk report from telemetry data queues Do not modify unless you have evaluated its impact given your service load. |
### SW_PYTHON Auto Instrumentation CLI
| Configuration | Environment Variable | Type | Default Value | Description |
diff --git a/skywalking/agent/__init__.py b/skywalking/agent/__init__.py
index a8fb212..9d61487 100644
--- a/skywalking/agent/__init__.py
+++ b/skywalking/agent/__init__.py
@@ -16,7 +16,6 @@
#
import atexit
-import os
from queue import Queue, Full
from threading import Thread, Event
from typing import TYPE_CHECKING
@@ -61,7 +60,7 @@ def __report():
while not __finished.is_set():
try:
- __protocol.report(__queue) # is blocking actually, blocks for max config.queue_timeout seconds
+ __protocol.report_segment(__queue) # is blocking actually, blocks for max config.queue_timeout seconds
wait = base
except Exception as exc:
logger.error(str(exc))
@@ -89,7 +88,7 @@ def __send_profile_snapshot():
while not __finished.is_set():
try:
- __protocol.send_snapshot(__snapshot_queue)
+ __protocol.report_snapshot(__snapshot_queue)
wait = base
except Exception as exc:
logger.error(str(exc))
@@ -198,7 +197,7 @@ def __init():
def __fini():
- __protocol.report(__queue, False)
+ __protocol.report_segment(__queue, False)
__queue.join()
if config.log_reporter_active:
@@ -206,31 +205,12 @@ def __fini():
__log_queue.join()
if config.profiler_active:
- __protocol.send_snapshot(__snapshot_queue, False)
+ __protocol.report_snapshot(__snapshot_queue, False)
__snapshot_queue.join()
__finished.set()
-def __fork_before():
- if config.protocol != 'http':
- logger.warning(f'fork() not currently supported with {config.protocol} protocol')
-
- # TODO: handle __queue and __finished correctly (locks, mutexes, etc...), need to lock before fork and unlock after
- # if possible, or ensure they are not locked in threads (end threads and restart after fork?)
-
- __protocol.fork_before()
-
-
-def __fork_after_in_parent():
- __protocol.fork_after_in_parent()
-
-
-def __fork_after_in_child():
- __protocol.fork_after_in_child()
- __init_threading()
-
-
def start():
global __started
if __started:
@@ -256,10 +236,6 @@ def start():
atexit.register(__fini)
- if (hasattr(os, 'register_at_fork')):
- os.register_at_fork(before=__fork_before, after_in_parent=__fork_after_in_parent,
- after_in_child=__fork_after_in_child)
-
def stop():
atexit.unregister(__fini)
diff --git a/skywalking/agent/protocol/__init__.py b/skywalking/agent/protocol/__init__.py
index f804eb5..bf6b427 100644
--- a/skywalking/agent/protocol/__init__.py
+++ b/skywalking/agent/protocol/__init__.py
@@ -19,38 +19,31 @@ from abc import ABC, abstractmethod
from queue import Queue
-class BaseProtocol(ABC):
+class Protocol(ABC):
@abstractmethod
def heartbeat(self):
raise NotImplementedError()
@abstractmethod
- def report(self, queue: Queue, block: bool = True):
+ def report_segment(self, queue: Queue, block: bool = True):
raise NotImplementedError()
@abstractmethod
def report_log(self, queue: Queue, block: bool = True):
raise NotImplementedError()
-
-class Protocol(BaseProtocol):
- def fork_before(self):
- pass
-
- def fork_after_in_parent(self):
- pass
-
- def fork_after_in_child(self):
- pass
-
+ @abstractmethod
def report_meter(self, queue: Queue, block: bool = True):
- pass
+ raise NotImplementedError()
- def query_profile_commands(self):
- pass
+ @abstractmethod
+ def report_snapshot(self, queue: Queue, block: bool = True):
+ raise NotImplementedError()
- def send_snapshot(self, queue: Queue, block: bool = True):
- pass
+ @abstractmethod
+ def query_profile_commands(self):
+ raise NotImplementedError()
+ @abstractmethod
def notify_profile_task_finish(self, task):
- pass
+ raise NotImplementedError()
diff --git a/skywalking/agent/protocol/grpc.py b/skywalking/agent/protocol/grpc.py
index 9055b1e..d873830 100644
--- a/skywalking/agent/protocol/grpc.py
+++ b/skywalking/agent/protocol/grpc.py
@@ -90,7 +90,7 @@ class GrpcProtocol(Protocol):
self.channel.unsubscribe(self._cb)
self.channel.subscribe(self._cb, try_to_connect=True)
- def report(self, queue: Queue, block: bool = True):
+ def report_segment(self, queue: Queue, block: bool = True):
start = None
def generator():
@@ -214,7 +214,6 @@ class GrpcProtocol(Protocol):
if logger_debug_enabled:
logger.debug('Reporting Meter')
-
yield meter_data
try:
@@ -223,7 +222,7 @@ class GrpcProtocol(Protocol):
self.on_error()
raise
- def send_snapshot(self, queue: Queue, block: bool = True):
+ def report_snapshot(self, queue: Queue, block: bool = True):
start = None
def generator():
@@ -255,7 +254,7 @@ class GrpcProtocol(Protocol):
yield transform_snapshot
try:
- self.profile_channel.send(generator())
+ self.profile_channel.report(generator())
except grpc.RpcError:
self.on_error()
raise
diff --git a/skywalking/agent/protocol/http.py b/skywalking/agent/protocol/http.py
index 0272fba..695604e 100644
--- a/skywalking/agent/protocol/http.py
+++ b/skywalking/agent/protocol/http.py
@@ -33,17 +33,13 @@ class HttpProtocol(Protocol):
self.traces_reporter = HttpTraceSegmentReportService()
self.log_reporter = HttpLogDataReportService()
- def fork_after_in_child(self):
- self.service_management.fork_after_in_child()
- self.traces_reporter.fork_after_in_child()
-
def heartbeat(self):
if not self.properties_sent:
self.service_management.send_instance_props()
self.properties_sent = True
self.service_management.send_heart_beat()
- def report(self, queue: Queue, block: bool = True):
+ def report_segment(self, queue: Queue, block: bool = True):
start = None
def generator():
@@ -102,3 +98,16 @@ class HttpProtocol(Protocol):
self.log_reporter.report(generator=generator())
except Exception:
pass
+
+ # meter support requires OAP side HTTP handler to be implemented
+ def report_meter(self, queue: Queue, block: bool = True):
+ ...
+
+ def report_snapshot(self, queue: Queue, block: bool = True):
+ ...
+
+ def query_profile_commands(self):
+ ...
+
+ def notify_profile_task_finish(self, task):
+ ...
diff --git a/skywalking/agent/protocol/kafka.py b/skywalking/agent/protocol/kafka.py
index da9ec87..e5483f7 100644
--- a/skywalking/agent/protocol/kafka.py
+++ b/skywalking/agent/protocol/kafka.py
@@ -45,7 +45,7 @@ class KafkaProtocol(Protocol):
def heartbeat(self):
self.service_management.send_heart_beat()
- def report(self, queue: Queue, block: bool = True):
+ def report_segment(self, queue: Queue, block: bool = True):
start = None
def generator():
@@ -162,3 +162,13 @@ class KafkaProtocol(Protocol):
yield meter_data
self.meter_reporter.report(generator=generator())
+
+ # TODO: implement profiling for kafka
+ def report_snapshot(self, queue: Queue, block: bool = True):
+ ...
+
+ def query_profile_commands(self):
+ ...
+
+ def notify_profile_task_finish(self, task):
+ ...
diff --git a/skywalking/client/__init__.py b/skywalking/client/__init__.py
index cdd4519..420d1a2 100644
--- a/skywalking/client/__init__.py
+++ b/skywalking/client/__init__.py
@@ -14,34 +14,123 @@
# See the License for the specific language governing permissions and
# limitations under the License.
#
+import os
+import json
+import platform
+import socket
+from abc import ABC, abstractmethod
+from typing import List
+from skywalking import config
+from skywalking.loggings import logger
+from skywalking.protocol.common.Common_pb2 import KeyStringValuePair
-class ServiceManagementClient(object):
- def send_instance_props(self):
+
+class ServiceManagementClient(ABC):
+ """
+ Used to register service and instance to OAP.
+ """
+
+ def __init__(self):
+ self.sent_properties_counter = 0
+
+ @abstractmethod
+ def send_instance_props(self) -> None:
+ """
+ Unique to each protocol, send instance properties to OAP.
+ """
+ raise NotImplementedError()
+
+ def refresh_instance_props(self) -> None:
+ """
+ Periodically refresh the instance properties to prevent loss on OAP TTL records expiration.
+ Default: 30 * 10 seconds
+ """
+ self.sent_properties_counter += 1
+ if self.sent_properties_counter % config.collector_properties_report_period_factor == 0:
+ self.send_instance_props()
+
+ @staticmethod
+ def get_instance_properties() -> List[dict]:
+ """
+ Get current running Python interpreter's system properties.
+ Returns: [{'key': str, 'value': str}, ...]
+ """
+ try:
+ properties = [
+ {'key': 'language', 'value': 'python'},
+ {'key': 'OS Name', 'value': os.name},
+ {'key': 'Process No.', 'value': str(os.getpid())},
+ {'key': 'hostname', 'value': socket.gethostname()},
+ {'key': 'ipv4', 'value': '; '.join(socket.gethostbyname_ex(socket.gethostname())[2])},
+ {'key': 'python_implementation', 'value': platform.python_implementation()},
+ {'key': 'python_version', 'value': platform.python_version()},
+ ]
+
+ except Exception as e: # noqa
+ logger.exception('Failed to get OS info, fallback to basic properties.')
+ properties = [
+ {'key': 'language', 'value': 'python'},
+ {'key': 'Process No.', 'value': str(os.getpid())},
+ ]
+
+ namespace = config.namespace
+ if namespace:
+ properties.append({'key': 'namespace', 'value': namespace})
+
+ instance_properties_json = config.instance_properties_json
+ if instance_properties_json:
+ # load instance properties from json string
+ json_properties = json.loads(instance_properties_json)
+ for key, value in json_properties.items():
+ properties.append({'key': key, 'value': value})
+
+ return properties
+
+ def get_instance_properties_proto(self) -> List[KeyStringValuePair]:
+ """
+ Converts to protobuf format.
+ Returns: [KeyStringValuePair, ...]
+ """
+ return [KeyStringValuePair(key=prop['key'], value=prop['value']) for prop in self.get_instance_properties()]
+
+ def send_heart_beat(self) -> None:
+ """
+ Each protocol must implement this method to send heart beat to OAP.
+ Returns: None
+ """
raise NotImplementedError()
- def send_heart_beat(self):
+
+class Service(ABC):
+ @abstractmethod
+ def report(self, segment: bytes) -> None:
raise NotImplementedError()
-class TraceSegmentReportService(object):
+class TraceSegmentReportService(Service):
+ @abstractmethod
def report(self, generator):
raise NotImplementedError()
-class MeterReportService(object):
+class MeterReportService(Service):
+ @abstractmethod
def report(self, generator):
raise NotImplementedError()
-class LogDataReportService(object):
+class LogDataReportService(Service):
+ @abstractmethod
def report(self, generator):
raise NotImplementedError()
-class ProfileTaskChannelService(object):
+class ProfileTaskChannelService(Service):
+ @abstractmethod
def do_query(self):
raise NotImplementedError()
- def send(self, generator):
+ @abstractmethod
+ def report(self, generator):
raise NotImplementedError()
diff --git a/skywalking/client/grpc.py b/skywalking/client/grpc.py
index a62cb8a..1ebcd97 100644
--- a/skywalking/client/grpc.py
+++ b/skywalking/client/grpc.py
@@ -15,7 +15,6 @@
# limitations under the License.
#
-import os
import grpc
from skywalking import config
@@ -25,7 +24,6 @@ from skywalking.command import command_service
from skywalking.loggings import logger, logger_debug_enabled
from skywalking.profile import profile_task_execution_service
from skywalking.profile.profile_task import ProfileTask
-from skywalking.protocol.common.Common_pb2 import KeyStringValuePair
from skywalking.protocol.language_agent.Tracing_pb2_grpc import TraceSegmentReportServiceStub
from skywalking.protocol.logging.Logging_pb2_grpc import LogReportServiceStub
from skywalking.protocol.management.Management_pb2 import InstancePingPkg, InstanceProperties
@@ -37,33 +35,31 @@ from skywalking.protocol.profile.Profile_pb2_grpc import ProfileTaskStub
class GrpcServiceManagementClient(ServiceManagementClient):
def __init__(self, channel: grpc.Channel):
+ super().__init__()
+ self.instance_properties = self.get_instance_properties_proto()
self.service_stub = ManagementServiceStub(channel)
def send_instance_props(self):
- # TODO: other properties periodically | matching behavior of java agent
- properties = [
- KeyStringValuePair(key='language', value='python'),
- KeyStringValuePair(key='Process No.', value=str(os.getpid())),
- ]
- if config.namespace:
- properties.append(KeyStringValuePair(key='namespace', value=config.namespace))
self.service_stub.reportInstanceProperties(InstanceProperties(
service=config.service_name,
serviceInstance=config.service_instance,
- properties=properties,
+ properties=self.instance_properties,
))
def send_heart_beat(self):
+ self.refresh_instance_props()
+
+ self.service_stub.keepAlive(InstancePingPkg(
+ service=config.service_name,
+ serviceInstance=config.service_instance,
+ ))
+
if logger_debug_enabled:
logger.debug(
'service heart beats, [%s], [%s]',
config.service_name,
config.service_instance,
)
- self.service_stub.keepAlive(InstancePingPkg(
- service=config.service_name,
- serviceInstance=config.service_instance,
- ))
class GrpcTraceSegmentReportService(TraceSegmentReportService):
@@ -107,7 +103,7 @@ class GrpcProfileTaskChannelService(ProfileTaskChannelService):
commands = self.profile_stub.getProfileTaskCommands(query)
command_service.receive_command(commands)
- def send(self, generator):
+ def report(self, generator):
self.profile_stub.collectSnapshot(generator)
def finish(self, task: ProfileTask):
diff --git a/skywalking/client/http.py b/skywalking/client/http.py
index cc7506a..7b8313e 100644
--- a/skywalking/client/http.py
+++ b/skywalking/client/http.py
@@ -15,8 +15,6 @@
# limitations under the License.
#
import json
-import os
-
import requests
from google.protobuf import json_format
@@ -27,31 +25,26 @@ from skywalking.loggings import logger, logger_debug_enabled
class HttpServiceManagementClient(ServiceManagementClient):
def __init__(self):
+ super().__init__()
+ self.instance_properties = self.get_instance_properties()
+
proto = 'https://' if config.force_tls else 'http://'
self.url_instance_props = f"{proto}{config.collector_address.rstrip('/')}/v3/management/reportProperties"
self.url_heart_beat = f"{proto}{config.collector_address.rstrip('/')}/v3/management/keepAlive"
self.session = requests.Session()
- def fork_after_in_child(self):
- self.session.close()
- self.session = requests.Session()
-
def send_instance_props(self):
- properties = [
- {'key': 'language', 'value': 'python'},
- {'key': 'Process No.', 'value': str(os.getpid())},
- ]
- if config.namespace:
- properties.append({'key': 'namespace', 'value': config.namespace})
res = self.session.post(self.url_instance_props, json={
'service': config.service_name,
'serviceInstance': config.service_instance,
- 'properties': properties,
+ 'properties': self.instance_properties,
})
if logger_debug_enabled:
logger.debug('heartbeat response: %s', res)
def send_heart_beat(self):
+ self.refresh_instance_props()
+
if logger_debug_enabled:
logger.debug(
'service heart beats, [%s], [%s]',
@@ -72,10 +65,6 @@ class HttpTraceSegmentReportService(TraceSegmentReportService):
self.url_report = f"{proto}{config.collector_address.rstrip('/')}/v3/segment"
self.session = requests.Session()
- def fork_after_in_child(self):
- self.session.close()
- self.session = requests.Session()
-
def report(self, generator):
for segment in generator:
res = self.session.post(self.url_report, json={
@@ -127,10 +116,6 @@ class HttpLogDataReportService(LogDataReportService):
self.url_report = f"{proto}{config.collector_address.rstrip('/')}/v3/logs"
self.session = requests.Session()
- def fork_after_in_child(self):
- self.session.close()
- self.session = requests.Session()
-
def report(self, generator):
log_batch = [json.loads(json_format.MessageToJson(log_data)) for log_data in generator]
if log_batch: # prevent empty batches
diff --git a/skywalking/client/kafka.py b/skywalking/client/kafka.py
index ab24eb9..16fdaba 100644
--- a/skywalking/client/kafka.py
+++ b/skywalking/client/kafka.py
@@ -23,7 +23,6 @@ from kafka import KafkaProducer
from skywalking import config
from skywalking.client import MeterReportService, ServiceManagementClient, TraceSegmentReportService, LogDataReportService
from skywalking.loggings import logger, logger_debug_enabled
-from skywalking.protocol.common.Common_pb2 import KeyStringValuePair
from skywalking.protocol.language_agent.Meter_pb2 import MeterDataCollection
from skywalking.protocol.management.Management_pb2 import InstancePingPkg, InstanceProperties
@@ -58,6 +57,9 @@ __init_kafka_configs()
class KafkaServiceManagementClient(ServiceManagementClient):
def __init__(self):
+ super().__init__()
+ self.instance_properties = self.get_instance_properties_proto()
+
if logger_debug_enabled:
logger.debug('kafka reporter configs: %s', kafka_configs)
self.producer = KafkaProducer(**kafka_configs)
@@ -67,16 +69,10 @@ class KafkaServiceManagementClient(ServiceManagementClient):
self.send_instance_props()
def send_instance_props(self):
- properties = [
- KeyStringValuePair(key='language', value='python'),
- KeyStringValuePair(key='Process No.', value=str(os.getpid())),
- ]
- if config.namespace:
- properties.append(KeyStringValuePair(key='namespace', value=config.namespace))
instance = InstanceProperties(
service=config.service_name,
serviceInstance=config.service_instance,
- properties=properties,
+ properties=self.instance_properties,
)
key = bytes(self.topic_key_register + instance.serviceInstance, encoding='utf-8')
@@ -84,6 +80,8 @@ class KafkaServiceManagementClient(ServiceManagementClient):
self.producer.send(topic=self.topic, key=key, value=value)
def send_heart_beat(self):
+ self.refresh_instance_props()
+
if logger_debug_enabled:
logger.debug(
'service heart beats, [%s], [%s]',
diff --git a/skywalking/config.py b/skywalking/config.py
index 2b2d618..4cc97e0 100644
--- a/skywalking/config.py
+++ b/skywalking/config.py
@@ -32,8 +32,8 @@ any documentation to reflect changes here, just make sure to run `make doc-gen`
import os
import re
import uuid
-
from typing import List, Pattern
+import warnings
RE_IGNORE_PATH: Pattern = re.compile('^$')
RE_HTTP_IGNORE_METHOD: Pattern = RE_IGNORE_PATH
@@ -87,11 +87,13 @@ logging_level: str = os.getenv('SW_AGENT_LOGGING_LEVEL', 'INFO')
# The agent will exchange heartbeat message with SkyWalking OAP backend every `period` seconds
heartbeat_period: int = int(os.getenv('SW_AGENT_HEARTBEAT_PERIOD', '30'))
# The agent will report service instance properties every
-# `factor * heartbeat period` seconds default: 10*30 = 300 seconds (TODO)
-service_instance_property_report_factor = int(os.getenv('SW_AGENT_SERVICE_INSTANCE_PROPERTY_REPORT_FACTOR', '10'))
-# The agent will try to restart itself in any os.fork()-ed child process. Important note: it's not suitable for
-# large numbered, short-lived processes such as multiprocessing.Pool, as each one will introduce overhead and create
-# numerous instances in SkyWalking dashboard in format of `service_instance-child-<pid>` (TODO)
+# `factor * heartbeat period` seconds default: 10*30 = 300 seconds
+collector_properties_report_period_factor = int(os.getenv('SW_AGENT_COLLECTOR_PROPERTIES_REPORT_PERIOD_FACTOR', '10'))
+# A custom JSON string to be reported as service instance properties, e.g. `{"key": "value"}`
+instance_properties_json: str = os.getenv('SW_INSTANCE_PROPERTIES_JSON', '')
+# The agent will try to restart itself in any os.fork()-ed child process. Important Note: it's not suitable for
+# short-lived processes as each one will introduce overhead and create a new instance in SkyWalking dashboard
+# in format of `service_instance-child-<pid>` (TODO)
experimental_fork_support: bool = os.getenv('SW_AGENT_EXPERIMENTAL_FORK_SUPPORT', '').lower() == 'true'
# DANGEROUS - This option controls the interval of each bulk report from telemetry data queues
# Do not modify unless you have evaluated its impact given your service load.
@@ -222,6 +224,23 @@ def init(**kwargs) -> None:
glob[key] = val
+def finalize_feature() -> None:
+ """
+ Examine reporter configuration and warn users about the incompatibility of protocol vs features
+ """
+ global profiler_active, meter_reporter_active
+
+ if protocol == 'http' and (profiler_active or meter_reporter_active):
+ profiler_active = False
+ meter_reporter_active = False
+ warnings.warn('HTTP protocol does not support meter reporter and profiler. Please use gRPC protocol if you '
+ 'would like to use both features.')
+ elif protocol == 'kafka' and profiler_active:
+ profiler_active = False
+ warnings.warn('Kafka protocol does not support profiler. Please use gRPC protocol if you would like to use '
+ 'this feature.')
+
+
def finalize_name() -> None:
"""
This function concatenates the serviceName according to
@@ -274,5 +293,6 @@ def finalize() -> None:
"""
invokes finalizers
"""
+ finalize_feature()
finalize_regex()
finalize_name()
diff --git a/tests/e2e/base/docker-compose.base.yml b/tests/e2e/base/docker-compose.base.yml
index 8ab138d..ecebe43 100644
--- a/tests/e2e/base/docker-compose.base.yml
+++ b/tests/e2e/base/docker-compose.base.yml
@@ -48,6 +48,7 @@ services:
SW_AGENT_LOG_REPORTER_SAFE_MODE: "True"
SW_AGENT_METER_REPORTER_ACTIVE: "True"
SW_AGENT_PVM_METER_REPORTER_ACTIVE: "True"
+ SW_INSTANCE_PROPERTIES_JSON: '{"custom_property":"The answer to all is 42"}'
healthcheck:
test: [ "CMD", "bash", "-c", "cat < /dev/null > /dev/tcp/127.0.0.1/9090" ]
interval: 5s
diff --git a/tests/e2e/case/expected/service-instance.yml b/tests/e2e/case/expected/service-instance.yml
index b1ec846..3f7ebe4 100644
--- a/tests/e2e/case/expected/service-instance.yml
+++ b/tests/e2e/case/expected/service-instance.yml
@@ -20,16 +20,21 @@
name: {{ notEmpty .name }}
attributes:
{{- contains .attributes }}
-# TODO: The following attributes implementation is missing
-# - name: OS Name
-# value: linux
-# - name: hostname
-# value: {{ notEmpty .value }}
-# - name: Process No.
-# value: {{ notEmpty .value }}
+ - name: OS Name
+ value: "posix"
+ - name: Process No.
+ value: {{ notEmpty .value }}
+ - name: hostname
+ value: {{ notEmpty .value }}
+ - name: python_implementation
+ value: "CPython"
+ - name: python_version
+ value: {{ notEmpty .value }}
- name: ipv4s
- value: ""
- {{- end}}
+ value: {{ notEmpty .value }}
+ - name: custom_property
+ value: "The answer to all is 42"
+ {{- end }}
language: PYTHON
instanceuuid: {{ b64enc "e2e-service-provider" }}.1_{{ b64enc "provider1" }}
-{{- end}}
+{{- end}}
\ No newline at end of file
diff --git a/tools/config_doc_gen.py b/tools/config_doc_gen.py
index 56da558..f9d5995 100644
--- a/tools/config_doc_gen.py
+++ b/tools/config_doc_gen.py
@@ -124,8 +124,8 @@ def config_env_var_verify():
with open('skywalking/config.py', 'r') as config_file:
data = config_file.read().replace('\n', '')
for each in options_with_default_value_and_type.keys():
- if f"'SW_AGENT_{each.upper()}'" not in data:
- raise Exception(f'Environment variable SW_AGENT_{each.upper()} is not found in config.py\n'
+ if f'_{each.upper()}' not in data:
+ raise Exception(f'Environment variable for {each.upper()} is not found in config.py\n'
f'This means you have a mismatch of config.py variable and env var name')