|
14 | 14 | # See the License for the specific language governing permissions and |
15 | 15 | # limitations under the License. |
16 | 16 | # |
| 17 | +import os |
| 18 | +import json |
| 19 | +import platform |
| 20 | +import socket |
| 21 | +from abc import ABC, abstractmethod |
| 22 | +from typing import List |
17 | 23 |
|
| 24 | +from skywalking import config |
| 25 | +from skywalking.loggings import logger |
| 26 | +from skywalking.protocol.common.Common_pb2 import KeyStringValuePair |
18 | 27 |
|
19 | | -class ServiceManagementClient(object): |
20 | | - def send_instance_props(self): |
| 28 | + |
| 29 | +class ServiceManagementClient(ABC): |
| 30 | + """ |
| 31 | + Used to register service and instance to OAP. |
| 32 | + """ |
| 33 | + |
| 34 | + def __init__(self): |
| 35 | + self.sent_properties_counter = 0 |
| 36 | + |
| 37 | + @abstractmethod |
| 38 | + def send_instance_props(self) -> None: |
| 39 | + """ |
| 40 | + Unique to each protocol, send instance properties to OAP. |
| 41 | + """ |
| 42 | + raise NotImplementedError() |
| 43 | + |
| 44 | + def refresh_instance_props(self) -> None: |
| 45 | + """ |
| 46 | + Periodically refresh the instance properties to prevent loss on OAP TTL records expiration. |
| 47 | + Default: 30 * 10 seconds |
| 48 | + """ |
| 49 | + self.sent_properties_counter += 1 |
| 50 | + if self.sent_properties_counter % config.collector_properties_report_period_factor == 0: |
| 51 | + self.send_instance_props() |
| 52 | + |
| 53 | + @staticmethod |
| 54 | + def get_instance_properties() -> List[dict]: |
| 55 | + """ |
| 56 | + Get current running Python interpreter's system properties. |
| 57 | + Returns: [{'key': str, 'value': str}, ...] |
| 58 | + """ |
| 59 | + try: |
| 60 | + properties = [ |
| 61 | + {'key': 'language', 'value': 'python'}, |
| 62 | + {'key': 'OS Name', 'value': os.name}, |
| 63 | + {'key': 'Process No.', 'value': str(os.getpid())}, |
| 64 | + {'key': 'hostname', 'value': socket.gethostname()}, |
| 65 | + {'key': 'ipv4', 'value': '; '.join(socket.gethostbyname_ex(socket.gethostname())[2])}, |
| 66 | + {'key': 'python_implementation', 'value': platform.python_implementation()}, |
| 67 | + {'key': 'python_version', 'value': platform.python_version()}, |
| 68 | + ] |
| 69 | + |
| 70 | + except Exception as e: # noqa |
| 71 | + logger.exception('Failed to get OS info, fallback to basic properties.') |
| 72 | + properties = [ |
| 73 | + {'key': 'language', 'value': 'python'}, |
| 74 | + {'key': 'Process No.', 'value': str(os.getpid())}, |
| 75 | + ] |
| 76 | + |
| 77 | + namespace = config.namespace |
| 78 | + if namespace: |
| 79 | + properties.append({'key': 'namespace', 'value': namespace}) |
| 80 | + |
| 81 | + instance_properties_json = config.instance_properties_json |
| 82 | + if instance_properties_json: |
| 83 | + # load instance properties from json string |
| 84 | + json_properties = json.loads(instance_properties_json) |
| 85 | + for key, value in json_properties.items(): |
| 86 | + properties.append({'key': key, 'value': value}) |
| 87 | + |
| 88 | + return properties |
| 89 | + |
| 90 | + def get_instance_properties_proto(self) -> List[KeyStringValuePair]: |
| 91 | + """ |
| 92 | + Converts to protobuf format. |
| 93 | + Returns: [KeyStringValuePair, ...] |
| 94 | + """ |
| 95 | + return [KeyStringValuePair(key=prop['key'], value=prop['value']) for prop in self.get_instance_properties()] |
| 96 | + |
| 97 | + def send_heart_beat(self) -> None: |
| 98 | + """ |
| 99 | + Each protocol must implement this method to send heart beat to OAP. |
| 100 | + Returns: None |
| 101 | + """ |
21 | 102 | raise NotImplementedError() |
22 | 103 |
|
23 | | - def send_heart_beat(self): |
| 104 | + |
| 105 | +class Service(ABC): |
| 106 | + @abstractmethod |
| 107 | + def report(self, segment: bytes) -> None: |
24 | 108 | raise NotImplementedError() |
25 | 109 |
|
26 | 110 |
|
27 | | -class TraceSegmentReportService(object): |
| 111 | +class TraceSegmentReportService(Service): |
| 112 | + @abstractmethod |
28 | 113 | def report(self, generator): |
29 | 114 | raise NotImplementedError() |
30 | 115 |
|
31 | 116 |
|
32 | | -class MeterReportService(object): |
| 117 | +class MeterReportService(Service): |
| 118 | + @abstractmethod |
33 | 119 | def report(self, generator): |
34 | 120 | raise NotImplementedError() |
35 | 121 |
|
36 | 122 |
|
37 | | -class LogDataReportService(object): |
| 123 | +class LogDataReportService(Service): |
| 124 | + @abstractmethod |
38 | 125 | def report(self, generator): |
39 | 126 | raise NotImplementedError() |
40 | 127 |
|
41 | 128 |
|
42 | | -class ProfileTaskChannelService(object): |
| 129 | +class ProfileTaskChannelService(Service): |
| 130 | + @abstractmethod |
43 | 131 | def do_query(self): |
44 | 132 | raise NotImplementedError() |
45 | 133 |
|
46 | | - def send(self, generator): |
| 134 | + @abstractmethod |
| 135 | + def report(self, generator): |
47 | 136 | raise NotImplementedError() |
0 commit comments