openrvdas.logger.writers.grafana_live_writer

  1#!/usr/bin/env python3
  2"""
  3This writer pushes data to Grafana Live using the /api/live/push endpoint
  4with InfluxDB Line Protocol format. This is a supported, stable API.
  5
  6NOTE: This code was created substantially via Google Gemini. The usual
  7AI code disclaimers apply - watch your topknot...
  8
  9SECURITY - AUTHENTICATION TOKEN:
 10The Grafana Service Account Token can be provided in three ways, checked in
 11the following order of priority:
 12
 131. token_file (BEST):
 14   Path to a file containing the token (e.g., /etc/secrets/grafana_token).
 15   Set file permissions to 600 so only the process owner can read it.
 16
 172. GRAFANA_API_TOKEN (GOOD):
 18   Environment variable. Better than hardcoding, but visible to root.
 19
 203. api_token (LEAST SECURE):
 21   Passed directly as a string argument. Discouraged for production use
 22   as it may appear in version control or process lists.
 23"""
 24
 25import logging
 26import json
 27import time
 28import threading
 29import queue
 30import urllib.request
 31import urllib.error
 32import os
 33from typing import Union
 34from urllib.parse import quote
 35
 36from logger.utils.das_record import DASRecord  # noqa: E402
 37from logger.writers.writer import Writer  # noqa: E402
 38
 39
 40class GrafanaLiveWriter(Writer):
 41    """
 42    Write data records to Grafana Live via HTTP Push using InfluxDB Line Protocol.
 43
 44    Grafana Live supports ingesting data via InfluxDB Line Protocol at:
 45        POST /api/live/push/{stream_id}
 46
 47    The 'stream_id' passed during initialization is treated as the 'base_stream'.
 48    Records are written to dynamic streams constructed as:
 49        {base_stream}/{data_id}/{message_type}
 50
 51    This allows a single Writer to populate multiple Grafana Live channels
 52    based on the content of the records.
 53
 54    Example:
 55        writer = GrafanaLiveWriter(
 56            host='localhost:3000',
 57            stream_id='openrvdas',  # Base stream ID
 58            token_file='/etc/secrets/grafana_token'
 59        )
 60        # A record with data_id='mru' and message_type='rotation' will be pushed to:
 61        # /api/live/push/openrvdas/mru/rotation
 62    """
 63
 64    def __init__(self, host, stream_id, api_token=None, token_file=None,
 65                 secure=False, measurement_name=None, batch_size=1,
 66                 queue_size=1000, **kwargs):
 67        """
 68        Initialize GrafanaLiveWriter.
 69
 70        Args:
 71            host (str): Grafana host (e.g., 'localhost:3000')
 72            stream_id (str): Base Stream ID (e.g., 'openrvdas')
 73            api_token (str): Direct token string (Low priority security).
 74            token_file (str): Path to file containing token (High priority security).
 75            secure (bool): Use HTTPS instead of HTTP
 76            measurement_name (str): Override measurement name (uses message_type if None)
 77            batch_size (int): Number of records to batch (1 = no batching)
 78            queue_size (int): Maximum queue size before dropping records
 79            quiet (bool): Suppress routine logging
 80        """
 81        super().__init__(**kwargs)  # processes 'quiet' and type hints
 82
 83        # -----------------------------------------------------------
 84        # SECURITY LOGIC: File > Env > Argument
 85        # -----------------------------------------------------------
 86        self.api_token = None
 87
 88        # 1. Try reading from a secure file (Best)
 89        if token_file:
 90            try:
 91                # Expand ~ to user home if necessary
 92                expanded_path = os.path.expanduser(token_file)
 93                if os.path.exists(expanded_path):
 94                    with open(expanded_path, 'r') as f:
 95                        self.api_token = f.read().strip()
 96                else:
 97                    logging.warning(f"Grafana token_file not found: {token_file}")
 98            except IOError as e:
 99                logging.error(f"Could not read Grafana token file: {e}")
100
101        # 2. Try Environment Variable (Good)
102        if not self.api_token:
103            self.api_token = os.environ.get('GRAFANA_API_TOKEN')
104
105        # 3. Try direct argument (Development only)
106        if not self.api_token:
107            self.api_token = api_token
108
109        # -----------------------------------------------------------
110
111        if not host or not stream_id or not self.api_token:
112            raise RuntimeError(
113                'GrafanaLiveWriter requires host, stream_id, and a valid api_token '
114                '(provided via token_file, GRAFANA_API_TOKEN env var, or api_token arg).'
115            )
116
117        # Sanitize Host
118        self.host = host.strip()
119        for prefix in ['http://', 'https://', 'ws://', 'wss://']:
120            if self.host.startswith(prefix):
121                self.host = self.host[len(prefix):]
122        if self.host.endswith('/'):
123            self.host = self.host[:-1]
124
125        # Store Base Stream ID
126        raw_id = str(stream_id).strip()
127        if raw_id.endswith('/'):
128            raw_id = raw_id[:-1]
129
130        # We store the raw base ID here. Construction happens in write()
131        self.base_stream_id = raw_id
132
133        self.api_token = self.api_token.strip()
134        self.measurement_name = measurement_name
135        self.batch_size = max(1, batch_size)
136        self.protocol = 'https' if secure else 'http'
137
138        # Base API URL
139        self.base_api_url = f'{self.protocol}://{self.host}/api/live/push'
140
141        logging.info(f'GrafanaLiveWriter initialized. Host: {self.host}, '
142                     f'Base Stream: {self.base_stream_id}')
143        logging.info(f'  Batch size: {batch_size}')
144
145        # Queue holds tuples: (target_stream_id, payload_string)
146        self.queue = queue.Queue(maxsize=queue_size)
147
148        # Statistics
149        self.stats = {
150            'sent': 0,
151            'dropped': 0,
152            'errors': 0,
153            'last_error': None
154        }
155        self.stats_lock = threading.Lock()
156
157        # Background worker
158        self.stop_event = threading.Event()
159        self.thread = threading.Thread(target=self._http_worker, daemon=True)
160        self.thread.start()
161
162    def _http_worker(self):
163        """Background worker that sends batched data to Grafana."""
164        # Batches is a dict mapping stream_id -> list of payload strings
165        # Example: { 'openrvdas/mru/rot': ['line1', 'line2'], 'openrvdas/pos/loc': ['line3'] }
166        batches = {}
167        pending_count = 0
168        last_send = time.time()
169
170        while not self.stop_event.is_set():
171            try:
172                # Get item with timeout to allow checking stop_event
173                try:
174                    # Queue items are (stream_id, payload_str)
175                    stream_id, payload_str = self.queue.get(timeout=1.0)
176
177                    if stream_id not in batches:
178                        batches[stream_id] = []
179
180                    batches[stream_id].append(payload_str)
181                    pending_count += 1
182                    self.queue.task_done()
183                except queue.Empty:
184                    pass
185
186                # Send batches if threshold reached or timeout elapsed
187                time_since_last = time.time() - last_send
188                should_send = (pending_count >= self.batch_size or
189                               (pending_count > 0 and time_since_last > 1.0))
190
191                if should_send:
192                    # Send each stream's batch separately
193                    for target_stream, batch_list in batches.items():
194                        if batch_list:
195                            self._send_batch(target_stream, batch_list)
196
197                    # Reset state
198                    batches = {}
199                    pending_count = 0
200                    last_send = time.time()
201
202            except Exception as e:
203                logging.error(f'Unexpected error in GrafanaLiveWriter worker: {e}')
204                # Don't tight loop on error
205                time.sleep(1)
206
207        # Send any remaining batches on shutdown
208        for target_stream, batch_list in batches.items():
209            if batch_list:
210                self._send_batch(target_stream, batch_list)
211
212    def _send_batch(self, stream_id, batch):
213        """Send a batch of line protocol records to a specific Grafana stream."""
214        safe_stream_id = quote(stream_id, safe='')
215        url = f"{self.base_api_url}/{safe_stream_id}"
216
217        try:
218            # Join batch with newlines
219            payload = '\n'.join(batch)
220            data_bytes = payload.encode('utf-8')
221
222            req = urllib.request.Request(url, data=data_bytes, method='POST')
223            req.add_header('Authorization', f'Bearer {self.api_token}')
224            req.add_header('Content-Type', 'text/plain')
225
226            with urllib.request.urlopen(req, timeout=5) as response:
227                if response.status == 204 or response.status == 200:
228                    with self.stats_lock:
229                        self.stats['sent'] += len(batch)
230                    logging.debug(f'Sent {len(batch)} records to Grafana stream {stream_id}')
231
232        except urllib.error.HTTPError as e:
233            error_msg = f'HTTP {e.code}: {e.reason}'
234            with self.stats_lock:
235                self.stats['errors'] += 1
236                self.stats['last_error'] = error_msg
237
238            logging.error(f'Grafana Push Error {error_msg} | URL: {url}')
239
240            # Read error body for details
241            try:
242                error_body = e.read().decode('utf-8')
243                logging.error(f'Error details: {error_body}')
244            except Exception:
245                pass
246
247            if e.code in (401, 403):
248                logging.error('Authentication error - check API token')
249                time.sleep(5)
250            elif e.code == 404:
251                logging.error('Endpoint not found - check Grafana version supports Live')
252                time.sleep(5)
253
254        except urllib.error.URLError as e:
255            error_msg = f'Connection error: {e.reason}'
256            with self.stats_lock:
257                self.stats['errors'] += 1
258                self.stats['last_error'] = error_msg
259
260            logging.error(f'Grafana Connection Error: {e.reason}')
261            time.sleep(1)
262
263        except Exception as e:
264            error_msg = str(e)
265            with self.stats_lock:
266                self.stats['errors'] += 1
267                self.stats['last_error'] = error_msg
268
269            logging.error(f'Unexpected error sending to Grafana: {e}')
270
271    def _normalize_record(self, record):
272        """Extract data_id, message_type, timestamp, and fields from record."""
273        if isinstance(record, str):
274            try:
275                record = DASRecord(json_str=record)
276            except json.JSONDecodeError:
277                return None, None, None, None
278
279        if isinstance(record, dict):
280            msg_type = record.get('message_type')
281            ts = record.get('timestamp', time.time())
282            did = record.get('data_id')
283            fields = record.get('fields', record)
284
285            return did, msg_type, ts, fields
286
287        if isinstance(record, DASRecord):
288            return record.data_id, record.message_type, record.timestamp, record.fields
289
290        return None, None, None, None
291
292    def _escape_key(self, key):
293        """Escape special characters in tag/field keys."""
294        return key.replace(',', '\\,').replace('=', '\\=').replace(' ', '\\ ')
295
296    def _escape_tag_value(self, value):
297        """Escape special characters in tag values."""
298        return str(value).replace(',', '\\,').replace('=', '\\=').replace(' ', '\\ ')
299
300    def _format_line_protocol(self, measurement, fields, timestamp, tags=None):
301        """
302        Format data in InfluxDB Line Protocol.
303
304        Format: measurement[,tag=value...] field=value[,field=value...] [timestamp]
305        """
306        if not fields:
307            return None
308
309        # Format fields
310        field_parts = []
311        for k, v in fields.items():
312            k_escaped = self._escape_key(k)
313
314            # Format value based on type
315            if isinstance(v, str):
316                v_escaped = v.replace('"', '\\"')
317                field_parts.append(f'{k_escaped}="{v_escaped}"')
318            elif isinstance(v, bool):
319                field_parts.append(f'{k_escaped}={str(v).upper()}')
320            elif isinstance(v, int):
321                field_parts.append(f'{k_escaped}={v}i')
322            elif isinstance(v, float):
323                field_parts.append(f'{k_escaped}={v}')
324            else:
325                v_escaped = str(v).replace('"', '\\"')
326                field_parts.append(f'{k_escaped}="{v_escaped}"')
327
328        if not field_parts:
329            return None
330
331        # Escape and format measurement name
332        measurement = self._escape_key(measurement.replace(' ', '_'))
333
334        # Format tags if provided
335        tag_str = ''
336        if tags:
337            tag_parts = [f'{self._escape_key(k)}={self._escape_tag_value(v)}'
338                         for k, v in tags.items()]
339            if tag_parts:
340                tag_str = ',' + ','.join(tag_parts)
341
342        # Format fields
343        field_str = ','.join(field_parts)
344
345        # Timestamp in nanoseconds
346        ts_ns = int(timestamp * 1e9)
347
348        return f'{measurement}{tag_str} {field_str} {ts_ns}'
349
350    def write(self, record: Union[DASRecord, dict, str]):
351        """
352        Write a record to Grafana Live.
353
354        The stream ID for the record is constructed as:
355            {base_stream}/{data_id}/{message_type}
356
357        Args:
358            record: DASRecord, dict, or JSON string
359        """
360        # See if it's something we can process, and if not, try digesting
361        if not self.can_process_record(record):  # inherited from BaseModule()
362            self.digest_record(record)  # inherited from BaseModule()
363            return
364
365        data_id, message_type, timestamp, fields = self._normalize_record(record)
366
367        if not fields:
368            return
369
370        # Determine measurement name (keep existing logic for Line Protocol)
371        measurement = message_type or self.measurement_name or data_id or 'default'
372
373        # Optional: Add data_id as a tag for better querying
374        tags = {'data_id': data_id} if data_id else None
375
376        # Format payload
377        payload = self._format_line_protocol(measurement, fields, timestamp, tags)
378
379        if payload:
380            # Honestly, this section is a bit voodoo. The current setup works
381            # both when there is a message_id and when there isn't, and I'm not
382            # sure why.
383
384            # Construct dynamic Stream ID
385            # Safe quoting of components to ensure valid URL structure
386            # Default to 'unknown' if parts are missing to avoid malformed URLs
387            safe_data_id = quote(str(data_id or 'unknown'), safe='')
388
389            # The message_id gets folded in by _format_line_protocol(), so
390            # don't need to duplicate (Gemini-written) code below
391            #safe_msg_type = quote(str(message_type or 'unknown'), safe='')
392            #target_stream_id = (f"{self.base_stream_id}/{safe_data_id}/"
393            #                    f"{safe_msg_type}")
394
395            # Without the logic below, we get duplicate data_ids when there's
396            # no message_id. I...don't understand, but I'm going with it.
397            if message_type:
398                target_stream_id = f"{self.base_stream_id}/{safe_data_id}"
399            else:
400                target_stream_id = f"{self.base_stream_id}"
401
402            try:
403                # Put tuple (stream_id, payload) into queue
404                self.queue.put_nowait((target_stream_id, payload))
405            except queue.Full:
406                with self.stats_lock:
407                    self.stats['dropped'] += 1
408                logging.warning('GrafanaLiveWriter queue full; dropping record')
409
410    def get_stats(self):
411        """Get writer statistics."""
412        with self.stats_lock:
413            return self.stats.copy()
414
415    def stop(self):
416        """Gracefully stop the writer."""
417        logging.info('Stopping GrafanaLiveWriter...')
418        self.stop_event.set()
419        self.thread.join(timeout=5)
420        logging.info(f'GrafanaLiveWriter stopped. Stats: {self.get_stats()}')