summaryrefslogtreecommitdiff
path: root/cloudinit/reporting/handlers.py
blob: 0a8c7af3428becd10de9a0d5bd631189a0923446 (plain)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
# This file is part of cloud-init. See LICENSE file for license information.

import abc
import fcntl
import json
import os
import queue
import struct
import threading
import time
import uuid
from datetime import datetime

from cloudinit import log as logging
from cloudinit.registry import DictRegistry
from cloudinit import (url_helper, util)

LOG = logging.getLogger(__name__)


class ReportException(Exception):
    pass


class ReportingHandler(metaclass=abc.ABCMeta):
    """Base class for report handlers.

    Implement :meth:`~publish_event` for controlling what
    the handler does with an event.
    """

    @abc.abstractmethod
    def publish_event(self, event):
        """Publish an event."""

    def flush(self):
        """Ensure ReportingHandler has published all events"""


class LogHandler(ReportingHandler):
    """Publishes events to the cloud-init log at the ``DEBUG`` log level."""

    def __init__(self, level="DEBUG"):
        super(LogHandler, self).__init__()
        if isinstance(level, int):
            pass
        else:
            input_level = level
            try:
                level = getattr(logging, level.upper())
            except Exception:
                LOG.warning("invalid level '%s', using WARN", input_level)
                level = logging.WARN
        self.level = level

    def publish_event(self, event):
        logger = logging.getLogger(
            '.'.join(['cloudinit', 'reporting', event.event_type, event.name]))
        logger.log(self.level, event.as_string())


class PrintHandler(ReportingHandler):
    """Print the event as a string."""

    def publish_event(self, event):
        print(event.as_string())


class WebHookHandler(ReportingHandler):
    def __init__(self, endpoint, consumer_key=None, token_key=None,
                 token_secret=None, consumer_secret=None, timeout=None,
                 retries=None):
        super(WebHookHandler, self).__init__()

        if any([consumer_key, token_key, token_secret, consumer_secret]):
            self.oauth_helper = url_helper.OauthUrlHelper(
                consumer_key=consumer_key, token_key=token_key,
                token_secret=token_secret, consumer_secret=consumer_secret)
        else:
            self.oauth_helper = None
        self.endpoint = endpoint
        self.timeout = timeout
        self.retries = retries
        self.ssl_details = util.fetch_ssl_details()

    def publish_event(self, event):
        if self.oauth_helper:
            readurl = self.oauth_helper.readurl
        else:
            readurl = url_helper.readurl
        try:
            return readurl(
                self.endpoint, data=json.dumps(event.as_dict()),
                timeout=self.timeout,
                retries=self.retries, ssl_details=self.ssl_details)
        except Exception:
            LOG.warning("failed posting event: %s", event.as_string())


class HyperVKvpReportingHandler(ReportingHandler):
    """
    Reports events to a Hyper-V host using Key-Value-Pair exchange protocol
    and can be used to obtain high level diagnostic information from the host.

    To use this facility, the KVP user-space daemon (hv_kvp_daemon) has to be
    running. It reads the kvp_file when the host requests the guest to
    enumerate the KVP's.

    This reporter collates all events for a module (origin|name) in a single
    json string in the dictionary.

    For more information, see
    https://technet.microsoft.com/en-us/library/dn798287.aspx#Linux%20guests
    """
    HV_KVP_EXCHANGE_MAX_VALUE_SIZE = 2048
    # The maximum value size expected in Azure
    HV_KVP_AZURE_MAX_VALUE_SIZE = 1024
    HV_KVP_EXCHANGE_MAX_KEY_SIZE = 512
    HV_KVP_RECORD_SIZE = (HV_KVP_EXCHANGE_MAX_KEY_SIZE +
                          HV_KVP_EXCHANGE_MAX_VALUE_SIZE)
    EVENT_PREFIX = 'CLOUD_INIT'
    MSG_KEY = 'msg'
    RESULT_KEY = 'result'
    DESC_IDX_KEY = 'msg_i'
    JSON_SEPARATORS = (',', ':')
    KVP_POOL_FILE_GUEST = '/var/lib/hyperv/.kvp_pool_1'
    _already_truncated_pool_file = False

    def __init__(self,
                 kvp_file_path=KVP_POOL_FILE_GUEST,
                 event_types=None):
        super(HyperVKvpReportingHandler, self).__init__()
        self._kvp_file_path = kvp_file_path
        HyperVKvpReportingHandler._truncate_guest_pool_file(
            self._kvp_file_path)

        self._event_types = event_types
        self.q = queue.Queue()
        self.incarnation_no = self._get_incarnation_no()
        self.event_key_prefix = u"{0}|{1}".format(self.EVENT_PREFIX,
                                                  self.incarnation_no)
        self.publish_thread = threading.Thread(
            target=self._publish_event_routine
        )
        self.publish_thread.daemon = True
        self.publish_thread.start()

    @classmethod
    def _truncate_guest_pool_file(cls, kvp_file):
        """
        Truncate the pool file if it has not been truncated since boot.
        This should be done exactly once for the file indicated by
        KVP_POOL_FILE_GUEST constant above. This method takes a filename
        so that we can use an arbitrary file during unit testing.
        Since KVP is a best-effort telemetry channel we only attempt to
        truncate the file once and only if the file has not been modified
        since boot. Additional truncation can lead to loss of existing
        KVPs.
        """
        if cls._already_truncated_pool_file:
            return
        boot_time = time.time() - float(util.uptime())
        try:
            if os.path.getmtime(kvp_file) < boot_time:
                with open(kvp_file, "w"):
                    pass
        except (OSError, IOError) as e:
            LOG.warning("failed to truncate kvp pool file, %s", e)
        finally:
            cls._already_truncated_pool_file = True

    def _get_incarnation_no(self):
        """
        use the time passed as the incarnation number.
        the incarnation number is the number which are used to
        distinguish the old data stored in kvp and the new data.
        """
        uptime_str = util.uptime()
        try:
            return int(time.time() - float(uptime_str))
        except ValueError:
            LOG.warning("uptime '%s' not in correct format.", uptime_str)
            return 0

    def _iterate_kvps(self, offset):
        """iterate the kvp file from the current offset."""
        with open(self._kvp_file_path, 'rb') as f:
            fcntl.flock(f, fcntl.LOCK_EX)
            f.seek(offset)
            record_data = f.read(self.HV_KVP_RECORD_SIZE)
            while len(record_data) == self.HV_KVP_RECORD_SIZE:
                kvp_item = self._decode_kvp_item(record_data)
                yield kvp_item
                record_data = f.read(self.HV_KVP_RECORD_SIZE)
            fcntl.flock(f, fcntl.LOCK_UN)

    def _event_key(self, event):
        """
        the event key format is:
        CLOUD_INIT|<incarnation number>|<event_type>|<event_name>|<uuid>
        [|subevent_index]
        """
        return u"{0}|{1}|{2}|{3}".format(self.event_key_prefix,
                                         event.event_type, event.name,
                                         uuid.uuid4())

    def _encode_kvp_item(self, key, value):
        data = struct.pack(
            "%ds%ds"
            % (
                self.HV_KVP_EXCHANGE_MAX_KEY_SIZE,
                self.HV_KVP_EXCHANGE_MAX_VALUE_SIZE,
            ),
            key.encode("utf-8"),
            value.encode("utf-8"),
        )
        return data

    def _decode_kvp_item(self, record_data):
        record_data_len = len(record_data)
        if record_data_len != self.HV_KVP_RECORD_SIZE:
            raise ReportException(
                "record_data len not correct {0} {1}."
                .format(record_data_len, self.HV_KVP_RECORD_SIZE))
        k = (record_data[0:self.HV_KVP_EXCHANGE_MAX_KEY_SIZE].decode('utf-8')
                                                             .strip('\x00'))
        v = (
            record_data[
                self.HV_KVP_EXCHANGE_MAX_KEY_SIZE:self.HV_KVP_RECORD_SIZE
            ].decode('utf-8').strip('\x00'))

        return {'key': k, 'value': v}

    def _append_kvp_item(self, record_data):
        with open(self._kvp_file_path, 'ab') as f:
            fcntl.flock(f, fcntl.LOCK_EX)
            for data in record_data:
                f.write(data)
            f.flush()
            fcntl.flock(f, fcntl.LOCK_UN)

    def _break_down(self, key, meta_data, description):
        del meta_data[self.MSG_KEY]
        des_in_json = json.dumps(description)
        des_in_json = des_in_json[1:(len(des_in_json) - 1)]
        i = 0
        result_array = []
        message_place_holder = "\"" + self.MSG_KEY + "\":\"\""
        while True:
            meta_data[self.DESC_IDX_KEY] = i
            meta_data[self.MSG_KEY] = ''
            data_without_desc = json.dumps(meta_data,
                                           separators=self.JSON_SEPARATORS)
            room_for_desc = (
                self.HV_KVP_AZURE_MAX_VALUE_SIZE -
                len(data_without_desc) - 8)
            value = data_without_desc.replace(
                message_place_holder,
                '"{key}":"{desc}"'.format(
                    key=self.MSG_KEY, desc=des_in_json[:room_for_desc]))
            subkey = "{}|{}".format(key, i)
            result_array.append(self._encode_kvp_item(subkey, value))
            i += 1
            des_in_json = des_in_json[room_for_desc:]
            if len(des_in_json) == 0:
                break
        return result_array

    def _encode_event(self, event):
        """
        encode the event into kvp data bytes.
        if the event content reaches the maximum length of kvp value.
        then it would be cut to multiple slices.
        """
        key = self._event_key(event)
        meta_data = {
            "name": event.name,
            "type": event.event_type,
            "ts": (datetime.utcfromtimestamp(event.timestamp)
                   .isoformat() + 'Z'),
        }
        if hasattr(event, self.RESULT_KEY):
            meta_data[self.RESULT_KEY] = event.result
        meta_data[self.MSG_KEY] = event.description
        value = json.dumps(meta_data, separators=self.JSON_SEPARATORS)
        # if it reaches the maximum length of kvp value,
        # break it down to slices.
        # this should be very corner case.
        if len(value) > self.HV_KVP_AZURE_MAX_VALUE_SIZE:
            return self._break_down(key, meta_data, event.description)
        else:
            data = self._encode_kvp_item(key, value)
            return [data]

    def _publish_event_routine(self):
        while True:
            items_from_queue = 0
            try:
                event = self.q.get(block=True)
                items_from_queue += 1
                encoded_data = []
                while event is not None:
                    encoded_data += self._encode_event(event)
                    try:
                        # get all the rest of the events in the queue
                        event = self.q.get(block=False)
                        items_from_queue += 1
                    except queue.Empty:
                        event = None
                try:
                    self._append_kvp_item(encoded_data)
                except (OSError, IOError) as e:
                    LOG.warning("failed posting events to kvp, %s", e)
                finally:
                    for _ in range(items_from_queue):
                        self.q.task_done()
            # when main process exits, q.get() will through EOFError
            # indicating we should exit this thread.
            except EOFError:
                return

    # since the saving to the kvp pool can be a time costing task
    # if the kvp pool already contains a chunk of data,
    # so defer it to another thread.
    def publish_event(self, event):
        if not self._event_types or event.event_type in self._event_types:
            self.q.put(event)

    def flush(self):
        LOG.debug('HyperVReportingHandler flushing remaining events')
        self.q.join()


available_handlers = DictRegistry()
available_handlers.register_item('log', LogHandler)
available_handlers.register_item('print', PrintHandler)
available_handlers.register_item('webhook', WebHookHandler)
available_handlers.register_item('hyperv', HyperVKvpReportingHandler)

# vi: ts=4 expandtab