Cnchi/logging_utils.py

   1 #!/usr/bin/env python
   2 # -*- coding: utf-8 -*-
   3 #
   4 # logging_utils.py
   5 #
   6 # Copyright © 2015-2018 Antergos
   7 #
   8 # This file is part of Cnchi.
   9 #
  10 # Cnchi is free software; you can redistribute it and/or modify
  11 # it under the terms of the GNU General Public License as published by
  12 # the Free Software Foundation; either version 3 of the License, or
  13 # (at your option) any later version.
  14 #
  15 # Cnchi is distributed in the hope that it will be useful,
  16 # but WITHOUT ANY WARRANTY; without even the implied warranty of
  17 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  18 # GNU General Public License for more details.
  19 #
  20 # The following additional terms are in effect as per Section 7 of the license:
  21 #
  22 # The preservation of all legal notices and author attributions in
  23 # the material or in the Appropriate Legal Notices displayed
  24 # by works containing it is required.
  25 #
  26 # You should have received a copy of the GNU General Public License
  27 # along with Cnchi; If not, see <http://www.gnu.org/licenses/>.
  28
  29 """ Logging utils to ease log calls """
  30
  31 import logging
  32 import uuid
  33 import json
  34 import os
  35 import requests
  36
  37 from info import CNCHI_VERSION, CNCHI_RELEASE_STAGE
  38
  39
  40 class Singleton(type):
  41     """ Single instance """
  42     _instance = None
  43
  44     def __call__(cls, *args, **kwargs):
  45         if not cls._instance:
  46             cls._instance = super(Singleton, cls).__call__(*args, **kwargs)
  47         return cls._instance
  48
  49     def __new__(cls, *args, **kwargs):
  50         obj = super().__new__(cls, *args, **kwargs)
  51         obj.ip_addr = None
  52         obj.install_id = None
  53         obj.api_key = None
  54         obj.have_install_id = False
  55         obj.after_location_screen = False
  56         obj.name = 'Cnchi'
  57         obj.description = 'Installer'
  58         obj.key = 'X-{}-{}'.format(obj.name, obj.description)
  59
  60         return obj
  61
  62
  63 class ContextFilter(logging.Filter, metaclass=Singleton):
  64     """ Context filter for logging methods to send logs to bugsnag """
  65     LOG_FOLDER = '/var/log/cnchi'
  66
  67     def __init__(self):
  68         super().__init__()
  69         self.api_key = self.get_bugsnag_api()
  70         self.have_install_id = False
  71         self.after_location_screen = False
  72         self.install_id = ""
  73         self.ip_addr = '1.2.3.4'
  74
  75     def filter(self, record):
  76         uid = str(uuid.uuid1()).split("-")
  77         record.uuid = uid[3] + "-" + uid[1] + "-" + uid[2] + "-" + uid[4]
  78         record.ip_addr = self.ip_addr
  79         record.install_id = self.install_id
  80         return True
  81
  82     def get_and_save_install_id(self, is_location_screen=False):
  83         """ Obtain an install identification """
  84         if self.have_install_id:
  85             return self.install_id
  86
  87         if is_location_screen:
  88             self.after_location_screen = True
  89
  90         if CNCHI_RELEASE_STAGE == 'development':
  91             self.install_id = 'development'
  92             self.ip_addr = '1.2.3.4'
  93             self.have_install_id = True
  94             return 'development'
  95
  96         info = {'ip': '1.2.3.4', 'id': '0'}
  97         url = self.get_url_for_id_request()
  98         headers = {self.api_key: CNCHI_VERSION}
  99
 100         try:
 101             req = requests.get(url, headers=headers)
 102             req.raise_for_status()
 103             info = json.loads(req.json())
 104         except Exception as err:
 105             logger = logging.getLogger()
 106             msg = "Unable to get an Id for this installation. Error: {0}".format(err.args)
 107             logger.debug(msg)
 108
 109         try:
 110             self.ip_addr = info['ip']
 111             self.install_id = info['id']
 112             self.have_install_id = True
 113         except (TypeError, KeyError):
 114             self.have_install_id = False
 115
 116         return self.install_id
 117
 118     @staticmethod
 119     def get_bugsnag_api():
 120         """ Gets bugsnag API key """
 121         config_path = '/etc/cnchi.conf'
 122         alt_config_path = '/usr/share/cnchi/data/cnchi.conf'
 123         bugsnag_api = None
 124
 125         if (not os.path.exists(config_path) and
 126                 os.path.exists(alt_config_path)):
 127             config_path = alt_config_path
 128
 129         if os.path.exists(config_path):
 130             with open(config_path) as bugsnag_conf:
 131                 bugsnag_api = bugsnag_conf.readline().strip()
 132
 133         return bugsnag_api
 134
 135     def get_url_for_id_request(self):
 136         """ Constructs bugsnag url """
 137         build_server = None
 138
 139         if self.api_key and CNCHI_RELEASE_STAGE != 'development':
 140             parts = {
 141                 1: 'com', 2: 'http', 3: 'hook', 4: 'build',
 142                 5: 'antergos', 6: 'cnchi', 7: '://'
 143             }
 144             build_server = '{}{}{}.{}.{}/{}?{}={}'.format(
 145                 parts[2], parts[7], parts[4], parts[5],
 146                 parts[1], parts[3], parts[6], self.api_key
 147             )
 148         return build_server
 149
 150     @staticmethod
 151     def filter_log_lines(log):
 152         """ Filter log lines """
 153         keep_lines = []
 154         look_for = ['[WARNING]', '[ERROR]']
 155         log_lines = log.readlines()
 156
 157         for i, log_line in enumerate(log_lines):
 158             for pattern in look_for:
 159                 if pattern in log_line:
 160                     try:
 161                         if 10 < i < (len(log_lines) - 10):
 162                             keep_lines.extend([log_lines[l]
 163                                                for l in range(i - 10, i + 10)])
 164                         elif i < 10:
 165                             keep_lines.extend([log_lines[l]
 166                                                for l in range(0, i)])
 167                         elif i > (len(log_lines) - 10):
 168                             keep_lines.extend([log_lines[l]
 169                                                for l in range(i, len(log_lines))])
 170                     except (IndexError, KeyError) as err:
 171                         print(err)
 172
 173         return keep_lines
 174
 175     def bugsnag_before_notify_callback(self, notification=None):
 176         """ Filter unwanted notifications here """
 177         if notification is not None:
 178             excluded = ["No such interface '/org/freedesktop/UPower'"]
 179
 180             if any(True for pattern in excluded if pattern in str(notification.exception)):
 181                 return False
 182
 183             if self.after_location_screen and not self.have_install_id:
 184                 self.get_and_save_install_id()
 185
 186             notification.user = {"id": self.ip_addr,
 187                                  "name": self.install_id,
 188                                  "install_id": self.install_id}
 189
 190             logs = [
 191                 os.path.join(ContextFilter.LOG_FOLDER, '{0}.log'.format(n))
 192                 for n in ['cnchi', 'cnchi-alpm', 'pacman', 'postinstall']]
 193             missing = [f for f in logs if not os.path.exists(f)]
 194             if missing:
 195                 for log in missing:
 196                     open(log, 'a').close()
 197
 198             with open(logs[0], 'r') as cnchi:
 199                 with open(logs[1], 'r') as pacman:
 200                     with open(logs[2], 'r') as postinstall:
 201                         log_dict = {'pacman': pacman,
 202                                     'postinstall': postinstall}
 203                         parse = {
 204                             log: [line.strip() for line in log_dict[log]]
 205                             for log in log_dict
 206                         }
 207                         parse['cnchi'] = self.filter_log_lines(cnchi)
 208                         notification.add_tab('logs', parse)
 209
 210             return notification
 211         return False
 212
 213     def send_install_result(self, result):
 214         """ Sends install result to bugsnag server (result: str) """
 215         try:
 216             if self.after_location_screen and not self.have_install_id:
 217                 self.get_and_save_install_id()
 218             build_server = self.get_url_for_id_request()
 219             if build_server and self.install_id:
 220                 url = "{0}&install_id={1}&result={2}"
 221                 url = url.format(build_server, self.install_id, result)
 222                 headers = {self.api_key: CNCHI_VERSION}
 223                 req = requests.get(url, headers=headers)
 224                 json.loads(req.json())
 225         except Exception as ex:
 226             logger = logging.getLogger()
 227             template = "Can't send install result. An exception of type {0} occured. "
 228             template += "Arguments:\n{1!r}"
 229             message = template.format(type(ex).__name__, ex.args)
 230             logger.error(message)