diff --git a/collector/collector/__init__.py b/collector/collector/__init__.py new file mode 100644 index 0000000..9ad9be3 --- /dev/null +++ b/collector/collector/__init__.py @@ -0,0 +1,47 @@ +# +# Copyright 2015-2017 Intel Corporation +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +# + +from flask import Flask +from . import config +from flask_sqlalchemy import SQLAlchemy +from logging.handlers import RotatingFileHandler + + +def configure_app(config_object, app): + app.config.from_object(config_object) + db = SQLAlchemy(app) + +app = Flask(__name__) +db = SQLAlchemy() +app.config.from_object(config.Config) + +try: + # try importing from the local dev configuration if it exists + from . import config_local + app.config.from_object(config_local.Config) +except Exception as e: + print(e) + pass + +from .model import * +from . import report_handler + +handler = RotatingFileHandler(app.config['LOG_FILE'], maxBytes=10000, backupCount=1) +handler.setLevel(app.config['LOG_LEVEL']) +app.logger.addHandler(handler) + + +# vi: ts=4 et sw=4 sts=4 diff --git a/collector/collector/config.py b/collector/collector/config.py new file mode 100644 index 0000000..7026c02 --- /dev/null +++ b/collector/collector/config.py @@ -0,0 +1,55 @@ +# +# Copyright 2015-2017 Intel Corporation +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +# + +import logging + + +class Config(object): + DEBUG = False + TESTING = False + LOG_LEVEL = logging.ERROR + SQLALCHEMY_DATABASE_URI = 'postgresql://postgres:@@db_password@@@localhost/telemdb' + SQLALCHEMY_TRACK_MODIFICATIONS = True + LOG_FILE = 'handler.log' + + # When PURGE_OLD_RECORDS == True then a purging system of old records will + # be triggered daily. If this variable is not present, then no purging will be done. + # If the purging system is enabled, then the following two variables must be set + PURGE_OLD_RECORDS = True + # The maximum retention time in days for records stored in the database + # which do not match the filters in PURGE_FILTERED_RECORDS. + # Use 0 to avoid deletion of all unfiltered records. + MAX_DAYS_KEEP_UNFILTERED_RECORDS = 35 + # See config_example.py for details about PURGE_FILTERED_RECORDS + PURGE_FILTERED_RECORDS = { + "classification": { + "org.clearlinux/hello/world": 1, + } + } + + # The Telemetry ID (TID) accepted by this `collector` app. The ID should be a + # random UUID, generated with (for example) `uuidgen`. The default value + # set here is used for records from the Clear Linux OS for Intel + # Architecture. + TELEMETRY_ID = "6907c830-eed9-4ce9-81ae-76daf8d88f0f" + + +class Testing(Config): + TESTING = True + SQLALCHEMY_DATABASE_URI = 'postgresql://postgres:@@db_password@@@localhost/testdb' + + +# vi: ts=4 et sw=4 sts=4 diff --git a/collector/collector/config_example.py b/collector/collector/config_example.py new file mode 100644 index 0000000..ddb3c8d --- /dev/null +++ b/collector/collector/config_example.py @@ -0,0 +1,78 @@ +# +# Copyright 2015-2017 Intel Corporation +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +# + +import logging + + +class Config(object): + DEBUG = False + TESTING = False + LOG_LEVEL = logging.ERROR + SQLALCHEMY_DATABASE_URI = 'postgresql://postgres:@@db_password@@@localhost/telemdb' + SQLALCHEMY_TRACK_MODIFICATIONS = True + LOG_FILE = 'handler.log' + + # When PURGE_OLD_RECORDS == True then a purging system of old records will + # be triggered daily. If this variable is not present, then no purging will be done. + # If the purging system is enabled, then the following two variables must be set + PURGE_OLD_RECORDS = True + # The maximum retention time in days for records stored in the database + # which do not match the filters in PURGE_FILTERED_RECORDS. + # Use 0 to avoid deletion of all unfiltered records. + MAX_DAYS_KEEP_UNFILTERED_RECORDS = 35 + # A dictionary in the following format: + # { + # "": { + # "": , + # ... + # }, + # ... + # } + # Currently supported fields to filter: + # [ + # 'severity', + # 'classification', + # 'machine_id' + # ] + # Use 0 to avoid deletion of records that matches the filter. + # If you do not want to filter records to delete, just set an empty dict '{}' + PURGE_FILTERED_RECORDS = { + "severity": { + 1: 5, + 4: 0 + }, + "classification": { + "org.clearlinux/mce/*": 0, + "org.clearlinux/hello/world": 1, + "org.clearlinux/heartbeat/ping": 1, + } + } + + # The Telemetry ID (TID) accepted by this `collector` app. The ID should be a + # random UUID, generated with (for example) `uuidgen`. The default value + # set here is used for records from the Clear Linux OS for Intel + # Architecture. + TELEMETRY_ID = "6907c830-eed9-4ce9-81ae-76daf8d88f0f" + + +class Testing(Config): + TESTING = True + SQLALCHEMY_DATABASE_URI = 'postgresql://postgres:@@db_password@@@localhost/testdb' + SQLALCHEMY_TRACK_MODIFICATIONS = True + PURGE_OLD_RECORDS = True + + +# vi: ts=4 et sw=4 sts=4 diff --git a/collector/collector/config_local.py b/collector/collector/config_local.py new file mode 100644 index 0000000..667d5f8 --- /dev/null +++ b/collector/collector/config_local.py @@ -0,0 +1,61 @@ +# +# Copyright 2015-2017 Intel Corporation +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +# + +# This configuration file can be used for a local development debug server +# at localhost:5000. Overrides the config module. + +import logging + + +class Config(object): + DEBUG = True + TESTING = False + LOG_LEVEL = logging.ERROR + + # If your telemdb database password is not 'postgres', update this line. + SQLALCHEMY_DATABASE_URI = 'postgresql://postgres:postgres@localhost/telemdb' + + SQLALCHEMY_TRACK_MODIFICATIONS = True + LOG_FILE = 'handler.log' + + # When PURGE_OLD_RECORDS == True then a purging system of old records will + # be triggered daily. If this variable is not present, then no purging will be done. + # If the purging system is enabled, then the following two variables must be set + PURGE_OLD_RECORDS = True + # The maximum retention time in days for records stored in the database + # which do not match the filters in PURGE_FILTERED_RECORDS. + # Use 0 to avoid deletion of all unfiltered records. + MAX_DAYS_KEEP_UNFILTERED_RECORDS = 35 + # See config_example.py for details about PURGE_FILTERED_RECORDS + PURGE_FILTERED_RECORDS = { + "classification": { + "org.clearlinux/hello/world": 1, + } + } + # Custom Payload transformations + # POST_PROCESSING_PARSERS = ["demo"] + + +class Testing(Config): + TESTING = True + + # If your testdb database password is not 'postgres', update this line. + SQLALCHEMY_DATABASE_URI = 'postgresql://postgres:postgres@localhost/testdb' + + SQLALCHEMY_TRACK_MODIFICATIONS = True + + +# vi: ts=4 et sw=4 sts=4 diff --git a/collector/collector/crash.py b/collector/collector/crash.py new file mode 120000 index 0000000..68dab23 --- /dev/null +++ b/collector/collector/crash.py @@ -0,0 +1 @@ +../../shared/crash.py \ No newline at end of file diff --git a/collector/collector/lib/exceptions.py b/collector/collector/lib/exceptions.py new file mode 100644 index 0000000..482d8b9 --- /dev/null +++ b/collector/collector/lib/exceptions.py @@ -0,0 +1,6 @@ + + +class PlugablePayloadParserException(Exception): + + def __init__(self, message): + self.__str__ = message diff --git a/collector/collector/lib/parser.py b/collector/collector/lib/parser.py new file mode 100644 index 0000000..401564d --- /dev/null +++ b/collector/collector/lib/parser.py @@ -0,0 +1,28 @@ +# +# Copyright 2018 Intel Corporation +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +# + +try: + from uwsgidecorators import spool +except ImportError: + def spool(f): + f.spool = f + return f + +# Alias for uwsgi spool feature. Using this feature to async data processing from +# different probes. +# Aliasing this feature will make it easier in the future to move to a different +# product if scaling becomes a problem. +parser_spooler = spool diff --git a/collector/collector/lib/validation.py b/collector/collector/lib/validation.py new file mode 100644 index 0000000..1e832a1 --- /dev/null +++ b/collector/collector/lib/validation.py @@ -0,0 +1,213 @@ +# +# Copyright 2015-2018 Intel Corporation +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +# + +import string +from .. import app +MAX_NUM_RECORDS = '1000' + +REQUIRED_HEADERS_V1 = ( + 'Arch', + 'Build', + 'Creation-Timestamp', + 'Classification', + 'Host-Type', + 'Kernel-Version', + 'Machine-Id', + 'Severity', + 'Record-Format-Version', +) + +REQUIRED_HEADERS_V2 = REQUIRED_HEADERS_V1 + ( + 'Payload-Format-Version', + 'System-Name', + 'X-Telemetry-Tid', +) + +REQUIRED_HEADERS_V3 = REQUIRED_HEADERS_V2 + ( + 'Board-Name', + 'Bios-Version', + 'Cpu-Model', +) + +REQUIRED_HEADERS_V4 = REQUIRED_HEADERS_V3 + ( + 'Event-Id', +) + +# see config.py for the meaning of this value +TELEMETRY_ID = app.config.get("TELEMETRY_ID", "6907c830-eed9-4ce9-81ae-76daf8d88f0f") + +SEVERITY_VALUES = [1, 2, 3, 4] +VALID_RECORD_FORMAT_VERSIONS = [1, 2, 3, 4] +POSTGRES_INT_MAX = 2147483647 +MAXLEN_PRINTABLE = 200 + + +class InvalidUsage(Exception): + status_code = 400 + + def __init__(self, message, status_code=None, payload=None): + Exception.__init__(self) + self.message = message + if status_code is not None: + self.status_code = status_code + self.payload = payload + app.logger.error("InvalidUsage ({}): {}".format(self.status_code, self.message)) + + def to_dict(self): + rv = dict(self.payload or ()) + rv['message'] = self.message + return rv + + def __str__(self): + return self.message + + +def validate_headers(headers, required_headers): + """ Check for every single required header to be + in the request """ + req_headers = dict(headers) + req_headers_keys = req_headers.keys() + for header in required_headers: + if header not in req_headers_keys: + raise InvalidUsage("Record-Format-Version headers are invalid, {} missing".format(header), 400) + return True + + +def is_not_none(v): + return v is not None + + +def is_a_number(n): + return str(n).isdigit() + + +def value_is_printable(a_value): + return all([x in string.printable for x in a_value]) + + +def record_format_version_validation(record_format_version): + return all([is_not_none(record_format_version), is_a_number(record_format_version)]) and int(record_format_version) in VALID_RECORD_FORMAT_VERSIONS + + +def record_format_version_headers_validation(record_format_version, headers): + # Validate required headers based on Record Version + req_headrs = { + '1': REQUIRED_HEADERS_V1, + '2': REQUIRED_HEADERS_V2, + '3': REQUIRED_HEADERS_V3, + '4': REQUIRED_HEADERS_V4, + } + reqs = req_headrs.get(record_format_version, None) + if reqs is None: + return False + return validate_headers(headers, reqs) + + +def validation_tid_header(tid): + return tid == TELEMETRY_ID + + +def validate_severity(severity): + return is_a_number(severity) and int(severity) in SEVERITY_VALUES + + +def validate_classification(classification): + return is_not_none(classification) and len(classification.split('/')) == 3 + + +def validate_machine_id(machine_id): + return is_not_none(machine_id) and len(machine_id) <= 32 + + +def validate_timestamp(timestamp): + return all([is_not_none(timestamp), is_a_number(timestamp)]) + + +def validate_architecture(arch): + return arch in ["armv7l", "armv6l", "aarch64", "amd64", "sparc64", "ppc64", "i686", "i386", "x86_64", "ppc"] + + +def validate_host_type(host_type): + return is_not_none(host_type) and len(host_type) < 250 + + +def validate_kernel_version(kernel_version): + """ makes sure that the kernel version string has at least 2 numbers + version, major and minor revision + """ + try: + version, major_revision, _ = str(kernel_version).split('.', maxsplit=2) + return all([is_a_number(x) for x in [version, major_revision]]) + except ValueError as e: + print(e) + return False + + +def validate_payload_format_version(payload_format_version): + return int(payload_format_version) < POSTGRES_INT_MAX + + +def validate_x_header(header_value): + return is_not_none(header_value) and len(header_value) < MAXLEN_PRINTABLE and value_is_printable(header_value) + + +def validate_created(created): + return is_a_number(created) + + +def validate_record_limit(limit): + return is_a_number(limit) and limit <= MAX_NUM_RECORDS + + +def validate_event_id(header_value): + return len(header_value) == 32 and len([v for v in header_value if v in "0123456789abcdef"]) == 32 + + +def validate_header(name, value, expected=None): + if expected is None: + return { + 'tid_header': validation_tid_header, + 'record_format_version': record_format_version_validation, + 'payload_format_version': validate_payload_format_version, + 'severity': validate_severity, + 'classification': validate_classification, + 'machine_id': validate_machine_id, + 'timestamp': validate_timestamp, + 'architecture': validate_architecture, + 'host_type': validate_host_type, + 'kernel_version': validate_kernel_version, + 'board_name': validate_x_header, + 'cpu_model': validate_x_header, + 'bios_version': validate_x_header, + 'build': validate_x_header, + 'event_id': validate_event_id, + }.get(name, lambda x: False)(value) + else: + return value == expected + + +def validate_query(name, value): + return { + 'id': is_a_number, + 'ts_capture': is_a_number, + 'severity': validate_severity, + 'classification': validate_classification, + 'build': validate_x_header, + 'limit': validate_record_limit, + 'machine_id': validate_machine_id, + 'created_in_days': validate_created, + 'created_in_sec': validate_created, + }.get(name, lambda x: False)(value) diff --git a/collector/collector/log_requests.py b/collector/collector/log_requests.py new file mode 100644 index 0000000..49422dc --- /dev/null +++ b/collector/collector/log_requests.py @@ -0,0 +1,36 @@ +# +# Copyright 2015-2017 Intel Corporation +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +# + +from flask import request, current_app, app +import logging + + +@app.before_request +def before_request(): + headers = request.headers + payload = request.data + print('Before request') + current_app.logger.info('\t'.join([ + datetime.datetime.today().ctime(), + request.remote_addr, + request.method, + request.url, + str(request.data), + ', '.join([': '.join(x) for x in request.headers])]) + ) + + +# vi: ts=4 et sw=4 sts=4 diff --git a/collector/collector/model.py b/collector/collector/model.py new file mode 120000 index 0000000..ac6394a --- /dev/null +++ b/collector/collector/model.py @@ -0,0 +1 @@ +../../shared/model.py \ No newline at end of file diff --git a/collector/collector/parsers/README.md b/collector/collector/parsers/README.md new file mode 100644 index 0000000..4d08b5d --- /dev/null +++ b/collector/collector/parsers/README.md @@ -0,0 +1,68 @@ +## Plugable payload parsers + +### Overview + +A plugable parser is a python module that encapsulates a data transformation. +This transformation is applied to telemetry message payload if the classification +matches one of the classifications that the parser defines during plugin +registration. + +Plugin registration occurs during collector initialization, to enable a plugin +the plugin needs to be listed in the array ```POST_PROCESSING_PARSERS``` in +collector configuration. + +For example to enable the ```demo``` parser plugin that comes with the telemetry +backend we need to add a line like the following to collector configuration: + +```python + +class Config(object): + + # ... some configuration values + + POST_PROCESSING_PARSERS = ["demo"] + + # ... more configuration values + +``` + + +### Plugable parser `installation` + +The plugable parser module needs to be located under parsers directory, see +the following plugin parser source tree: + +```bash +/collector/collector/ + parsers/ + / + __init__.py + main.py +``` + +* `````` is the parser module name and should live under +collector/parsers +* ```main.py``` is the entry point for plugin registration. + +The parser entry point `must have` the following members: + +* An array called ```CLASSIFICATIONS``` with a list of message classifications. Any message +that matches one of these classifications will be parsed by this plugin. + +* A function signature ```parse_payload(kwargs)``` +which is the transformation that will be applied to the payload. + +### Other considerations + +Plugable parsers are applied asynchronously and no major consideration is required for +most use cases. When in doubt keep in mind the following: + +* Payload transformations are applied after the message record (including payload) +is safely stored. +* It is a correct assumption to think that the entire record can +be queried from storage. +* It is not possible to register multiple plugins for the same classification, +the correct way to go about doing this is to encapsulate the multiple +transformations in one plugin. This is by design, remember that transformations +are applied asynchronously therefore a given plugin execution order is not +guaranteed. diff --git a/collector/collector/parsers/demo/__init__.py b/collector/collector/parsers/demo/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/collector/collector/parsers/demo/main.py b/collector/collector/parsers/demo/main.py new file mode 100644 index 0000000..91b0411 --- /dev/null +++ b/collector/collector/parsers/demo/main.py @@ -0,0 +1,35 @@ +# +# Copyright 2018 Intel Corporation +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +# + + +import time +import base64 +from collector.lib.parser import parser_spooler + +CLASSIFICATIONS = ['org.clearlinux/telemetry/b64payload'] + + +@parser_spooler +def parse_payload(**kwargs): + """ + :param kwargs: classification=, + record_id=, + payload= + :return: None + """ + print("Processing data") + print(base64.b64decode(kwargs.get('payload'))) + print("Data processed") diff --git a/collector/collector/parsers/demo/models.py b/collector/collector/parsers/demo/models.py new file mode 100644 index 0000000..23a7f70 --- /dev/null +++ b/collector/collector/parsers/demo/models.py @@ -0,0 +1,26 @@ +# +# Copyright 2018 Intel Corporation +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +# + +from collector.models import db + +class Message(db.Model): + __tablename__ = 'demo_message' + record_id = db.Column(db.Integer, db.ForeignKey('records.id')) + payload = buildstamp = db.Column(db.String, default='') + + def __init__(self, **kwargs): + self.record_id = kwargs.get('record_id', None) + self.payload = kwargs.get('payload', None) diff --git a/collector/collector/purge.py b/collector/collector/purge.py new file mode 100644 index 0000000..cb6808a --- /dev/null +++ b/collector/collector/purge.py @@ -0,0 +1,38 @@ +# +# Copyright 2015-2017 Intel Corporation +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +# + +from .model import Record +from . import app + +try: + import uwsgi + from uwsgidecorators import cron + + PURGE_OLD_RECORDS = app.config.get("PURGE_OLD_RECORDS", True) + + # Runs cron job at 4:30 every day + @cron(30, 4, -1, -1, -1, target='spooler') + def purge_task(signum): + if PURGE_OLD_RECORDS: + app.logger.info("Running cron job for purging records") + with app.app_context(): + Record.delete_records() + +except ImportError: + app.logger.info("Import error for uwsgi") + + +# vi: ts=4 et sw=4 sts=4 diff --git a/collector/collector/report_handler.py b/collector/collector/report_handler.py new file mode 100644 index 0000000..669da85 --- /dev/null +++ b/collector/collector/report_handler.py @@ -0,0 +1,346 @@ +# +# Copyright 2015-2017 Intel Corporation +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +# + +import re +import time +import datetime +import importlib +from flask import request +from flask import jsonify +from flask import redirect +from .lib.exceptions import PlugablePayloadParserException +from .lib.validation import ( + validate_header, + validate_query, + MAX_NUM_RECORDS, + record_format_version_headers_validation, + InvalidUsage) +from .model import ( + Classification, + Build) +from .crash import ( + process_guilties, + is_crash_classification) +from .purge import * + +tm_version_regex = re.compile("^[0-9]+\.[0-9]+$") +client_id_regex = re.compile( + "^[0-9]+[ \t]+[-/.:_+*a-zA-Z0-9]+[ \t]+[-/.:_+*a-zA-Z0-9]+$") +ts_v3_regex = re.compile("^[0-9]+$") +# FIXME: configurable limits +max_payload_len_inline = 30 * 1024 # 30k +max_payload_len = 300 * 1024 # 300k +MAX_INTERVAL_SEC = 24 * 60 * 60 * 30 # 30 days in seconds + + +# This variable is loaded during app initialization from +# values on config.POST_PROCESSING_PARSERS +POST_PROCESSING_PARSERS = {} + + +@app.errorhandler(InvalidUsage) +def handle_invalid_usage(error): + response = jsonify(error.to_dict()) + response.status_code = error.status_code + return response + + +@app.before_request +def before_request(): + headers = request.headers + payload = request.data + app.logger.info('\t'.join([ + datetime.datetime.today().ctime(), + request.method, + request.url, + str(request.data), + ', '.join([': '.join(x) for x in request.headers])]) + ) + + +def clean_build_n_value(build): + # It is common to see the build numbers quoted in os-release. We don't + # need the quotes, so strip them from the semantic value. + _build = build.replace('"', '').replace("'", "") + return _build + + +def validate_header_value(header_value, record_name, err_msg): + try: + if validate_header(record_name, header_value) is True: + return header_value + except Exception as e: + err_msg = "Error parsing {}, {}".format(record_name, e) + raise InvalidUsage(err_msg, 400) + + +def validate_record_v3_headers(board_name, cpu_model, bios_version): + validate_header_value(board_name, "board_name", "board name is invalid") + validate_header_value(cpu_model, "cpu_model", "cpu model is invalid") + validate_header_value(bios_version, "bios_version", "BIOS version is invalid") + + +def collector_post_handler(): + + # The collector only accepts records with the configured TID value. + # Make sure the TID in the collector config.py matches the TID + # configured for telemetrics-client on the systems from which this + # collector receives records. + tid_header = request.headers.get("X-Telemetry-TID") + validate_header_value(tid_header, "tid_header", "Telemetry ID mismatch") + + record_format_version = request.headers.get("Record-Format-Version") + validate_header_value(record_format_version, "record_format_version", "Record-Format-Version is invalid") + + record_format_version_headers_validation(record_format_version, request.headers) + + severity = request.headers.get("Severity") + validate_header_value(severity, "severity", "severity value is out of range") + + classification = request.headers.get("Classification") + validate_header_value(classification, "classification", "Classification value is invalid") + + machine_id = request.headers.get("Machine-Id") + validate_header_value(machine_id, "machine_id", "Machine id value is invalid") + + timestamp = request.headers.get("Creation-Timestamp") + validate_header_value(timestamp, "timestamp", "timestamp is invalid") + + ts_capture = int(timestamp) + ts_reception = time.time() + + architecture = request.headers.get("Arch") + validate_header_value(architecture, "architecture", "architecture is invalid") + + host_type = request.headers.get("Host-Type") + validate_header_value(host_type, "host_type", "host type is invalid") + + kernel_version = request.headers.get("Kernel-Version") + validate_header_value(kernel_version, "kernel_version", "kernel version is invalid") + + # Record V3 format headers + board_name = "N/A" + cpu_model = "N/A" + bios_version = "N/A" + if record_format_version >= '3': + board_name = request.headers.get("Board-Name") + cpu_model = request.headers.get("Cpu-Model") + bios_version = request.headers.get("Bios-Version") + validate_record_v3_headers(board_name, cpu_model, bios_version) + + # Record V4 format headers + event_id = "N/A" + if record_format_version >= '4': + event_id = request.headers.get("Event-Id") + validate_header_value(event_id, "event_id", "Event id is invalid") + + os_name = request.headers.get('System-Name') + os_name = os_name.replace('"', '').replace("'", "") + build = request.headers.get('Build') + build = clean_build_n_value(build) + # The build number is stored as a string in the database, but if this + # record is from a Clear Linux OS system, only accept an integer. + # Otherwise, loosen the restriction to the characters listed for + # VERSION_ID in os-release(5) in addition to the capital letters A-Z. + if os_name == 'clear-linux-os': + build_regex = re.compile(r"^[0-9]+$") + if not build_regex.match(build): + raise InvalidUsage("Clear Linux OS build version has invalid characters") + else: + build_regex = re.compile(r"^[-_a-zA-Z0-9.]+$") + if not build_regex.match(build): + raise InvalidUsage("Build version has invalid characters") + + payload_format_version = request.headers.get("Payload-Format-Version") + validate_header_value(payload_format_version, "payload_format_version", + "Payload format version outside of range supported") + + external = request.headers.get('X-CLR-External') + if external and external == "true": + external = True + else: + external = False + + try: + # prefer UTF-8, if possible + payload = request.data.decode('utf-8') + except UnicodeError: + # fallback to Latin-1, since it accepts all byte values + payload = request.data.decode('latin-1') + + db_class = Classification.query.filter_by(classification=classification).first() + if db_class is None: + db_class = Classification(classification) + + db_build = Build.query.filter_by(build=build).first() + if db_build is None: + db_build = Build(build) + + db_rec = Record.create(machine_id, host_type, severity, db_class, db_build, architecture, kernel_version, + record_format_version, ts_capture, ts_reception, payload_format_version, os_name, + board_name, bios_version, cpu_model, event_id, external, payload) + + # TODO: This should become a plugable parser + if is_crash_classification(classification): + # must pass args as bytes to uwsgi under Python 3 + process_guilties(klass=classification.encode(), id=str(db_rec.id).encode()) + + if classification in POST_PROCESSING_PARSERS.keys(): + POST_PROCESSING_PARSERS[classification](classification=classification.encode(), + record_id=str(db_rec.id).encode(), + payload=payload.encode()) + + resp = jsonify(db_rec.to_dict()) + resp.status_code = 201 + return resp + + +def validate_query_value(query_value, query_name, err_msg): + try: + if validate_query(query_name, query_value) is True: + return query_value + except Exception as e: + err_msg = "Error parsing {}, {}".format(query_name, e) + raise InvalidUsage(err_msg, 400) + + +def get_records_api_handler(): + + # Validate query parameters correctness + severity = request.args.get("severity", None) + if severity is not None: + validate_query_value(severity, "severity", "Severity should be a numeric value") + + classification = request.args.get('classification', None) + if classification is not None: + validate_query_value(classification, "classification", "Classification value is invalid") + + build = request.args.get('build', None) + if build is not None: + build = clean_build_n_value(build) + validate_query_value(build, "build", "Build value is invalid") + + machine_id = request.args.get('machine_id', None) + if machine_id is not None: + validate_query_value(machine_id, "machine_id", "Machine id value is invalid") + + created_in_days = request.args.get('created_in_days', None) + if created_in_days is not None: + validate_query_value(created_in_days, "created_in_days", "Created (in days) value is invalid") + + created_in_sec = request.args.get('created_in_sec', None) + if created_in_sec is not None: + validate_query_value(created_in_sec, "created_in_sec", "Created (in seconds) value is invalid") + + from_id = request.args.get('from_id', None) + if from_id is not None: + validate_query_value(from_id, "id", "Provided record id value is invalid") + + ts_capture = request.args.get('ts_capture', None) + if ts_capture is not None: + validate_query_value(ts_capture, "ts_capture", "Time stamp from record capture") + + limit = request.args.get('limit', MAX_NUM_RECORDS) + if limit != MAX_NUM_RECORDS: + validate_query_value(limit, "limit", "Record limit value is invalid") + + # Transform days interval to seconds + if created_in_days is not None: + created_in_days = int(created_in_days) + interval_sec = 24 * 60 * 60 * created_in_days + elif created_in_sec is not None: + interval_sec = int(created_in_sec) + else: + interval_sec = MAX_INTERVAL_SEC + + if interval_sec is not None and interval_sec > MAX_INTERVAL_SEC: + interval_sec = MAX_INTERVAL_SEC + + records = Record.query_records(build, classification, severity, machine_id, limit, interval_sec, + from_id=from_id, ts_capture=ts_capture) + record_list = [Record.to_dict(rec) for rec in records] + + return jsonify(records=record_list) + +# ########## Routes ########### + + +@app.route("/", methods=['GET', 'POST']) +@app.route("/v2/collector", methods=['GET', 'POST']) +def handler(): + if request.method == 'POST': + return collector_post_handler() + else: + return redirect("/telemetryui", code=302) + + +@app.route("/api/records", methods=['GET']) +def records_api_handler(): + """ + query filters for simple query: + classification + severity + build + machine_id + created_in_days - records created after given days + created_in_sec - records created after given seconds + + TODO: Advanced query with pagination and following parameters? + client_created_after - timestamp + client_created_before - timestamp + server_created_after - timestamp + server_created_before - timestamp + + """ + return get_records_api_handler() + + +def verify_parser_module(parser_module): + + if getattr(parser_module, 'CLASSIFICATIONS', None) is None: + raise PlugablePayloadParserException('Parser {} should have a CLASSIFICATIONS field') + + if getattr(parser_module, 'parse_payload', None) is None: + raise PlugablePayloadParserException('Parser {} should have a parse_payload method') + + +def load_parser(parser_name): + global POST_PROCESSING_PARSERS + + try: + parser_module = importlib.import_module("collector.parsers.{}.main".format(parser_name)) + verify_parser_module(parser_module) + for parser_classification in parser_module.CLASSIFICATIONS: + if parser_classification in POST_PROCESSING_PARSERS.keys(): + raise PlugablePayloadParserException("Parser plugin for class" + " {} is already registered".format(parser_classification)) + POST_PROCESSING_PARSERS[parser_classification] = parser_module.parse_payload + app.logger.info(" * Parser: {} registered for class: {}".format(parser_name, parser_classification)) + except PlugablePayloadParserException as e: + print(e.str()) + except ImportError as ie: + print(ie) + + +def load_parsers(): + pp_parsers = app.config.get('POST_PROCESSING_PARSERS', []) + for pp_parser in pp_parsers: + load_parser(pp_parser) + + +load_parsers() + +# vi: ts=4 et sw=4 sts=4 diff --git a/collector/collector/tests/api.py b/collector/collector/tests/api.py new file mode 100644 index 0000000..efd7591 --- /dev/null +++ b/collector/collector/tests/api.py @@ -0,0 +1,48 @@ +# +# Copyright 2015-2017 Intel Corporation +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +# + +import json +import unittest +from collector.tests.testcase import ( + RecordTestCases, + get_record) + + +class TestHandler(RecordTestCases): + + def test_query_report(self): + rec = get_record() + response = self.client.post('/', headers=rec, data='test') + self.assertTrue(response.status_code == 201) + filters = { + 'severity': 1, + } + response = self.client.get('/api/records', query_string=filters) + resp_obj = json.loads(response.data.decode('utf-8')) + self.assertEqual(len(resp_obj['records']), 1) + filters1 = { + 'build': '17780', + } + response = self.client.get('/api/records', query_string=filters1) + resp_obj = json.loads(response.data.decode('utf-8')) + self.assertEqual(len(resp_obj['records']), 0) + + +if __name__ == '__main__': + unittest.main() + + +# vi: ts=4 et sw=4 sts=4 diff --git a/collector/collector/tests/headers.py b/collector/collector/tests/headers.py new file mode 100644 index 0000000..389130b --- /dev/null +++ b/collector/collector/tests/headers.py @@ -0,0 +1,173 @@ +# +# Copyright 2015-2017 Intel Corporation +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +# + +import unittest +import json +from collector.tests.testcase import ( + RecordTestCases, + classification, + severity, + kernel_version, + record_version, + machine_id, + host_type, + arch, + build, + timestamp, + tid, + board_name, + cpu_model, + bios_version, + system_name, + payload_version, + event_id, + get_record_v1, + get_record_v2, + get_record_v3, + get_record_v4,) + + +class TestHandlerRecordV2(RecordTestCases): + """ + Tests record creation for record version 2 and headers validation + """ + @staticmethod + def get_version_records(): + return get_record_v2() + + def test_record_created(self): + headers = get_record_v2() + data = "hello" + response = self.client.post('/', headers=headers, data=data) + self.assertTrue(response.status_code == 201, response.data.decode('utf-8')) + json_resp = json.loads(response.data.decode('utf-8')) + self.assertTrue(json_resp['classification'] == headers[classification]) + self.assertTrue(str(json_resp['severity']) == str(headers[severity])) + self.assertTrue(json_resp['kernel_version'] == headers[kernel_version]) + self.assertTrue(str(json_resp['record_format_version']) == str(headers[record_version])) + self.assertTrue(json_resp['machine_id'] == headers[machine_id]) + self.assertTrue(json_resp['machine_type'] == headers[host_type]) + self.assertTrue(json_resp['arch'] == headers[arch]) + self.assertTrue(json_resp['build'] == headers[build]) + self.assertTrue(json_resp['payload'] == data) + + def missing_header(self, header, header_name): + headers = get_record_v2() + del headers[header_name] + response = self.client.post('/', headers=headers, data='test') + self.assertTrue(response.status_code == 400, response.data.decode('utf-8')) + + def test_post_fail_missing_classifiction(self): + self.missing_header('Classification', severity) + + def test_post_fail_missing_severity(self): + self.missing_header('Severity', severity) + + def test_post_fail_missing_kernel_version(self): + self.missing_header('Kernel-Version', kernel_version) + + def test_post_fail_missing_host_type(self): + self.missing_header('Host-Type', host_type) + + def test_post_fail_missing_machine_id(self): + self.missing_header('Machine-Id', machine_id) + + def test_post_fail_missing_arch(self): + self.missing_header('Arch', arch) + + def test_post_fail_missing_build(self): + self.missing_header('Build', build) + + def test_post_fail_missing_record_version(self): + self.missing_header('Record-Format-Version', record_version) + + +class TestHandlerRecordTransitionV2toV3(RecordTestCases): + """ + This test case tests a transition where the client is in record + version 2 though is sending v3 headers, make sure the server + understands the record as v2 and do not have problems creating it + """ + def test_record_created(self): + headers = get_record_v3() + data = "hello" + response = self.client.post('/', headers=headers, data=data) + self.assertTrue(response.status_code == 201, response.data.decode('utf-8')) + json_resp = json.loads(response.data.decode('utf-8')) + self.assertTrue(json_resp['classification'] == headers[classification]) + self.assertTrue(int(json_resp['severity']) == int(headers[severity])) + self.assertTrue(json_resp['kernel_version'] == headers[kernel_version]) + self.assertTrue(int(json_resp['record_format_version']) == int(headers[record_version])) + self.assertTrue(json_resp['machine_id'] == headers[machine_id]) + self.assertTrue(json_resp['machine_type'] == headers[host_type]) + self.assertTrue(json_resp['arch'] == headers[arch]) + self.assertTrue(json_resp['build'] == headers[build]) + + +class TestHandlerRecordV3(RecordTestCases): + """ + Test record v3 making sure to validate expected headers for v3 + if record version is properly set to 3. + """ + @staticmethod + def get_version_records(): + return get_record_v3() + + def test_post_record_v3_with_headers_v2(self): + headers = get_record_v2() + headers.update({record_version: 3, }) + response = self.client.post('/', headers=headers, data='test') + self.assertTrue(response.status_code == 400, response.data.decode('utf-8')) + + def test_post_record_v3(self): + headers = get_record_v3() + response = self.client.post('/', headers=headers, data='test') + self.assertTrue(response.status_code == 201, response.data.decode('utf-8')) + + def test_post_missing_cpu_model(self): + self.missing_header('Cpu-Model', cpu_model) + + def test_post_missing_board_name(self): + self.missing_header('Board-Name', board_name) + + def test_post_missing_bios_version(self): + self.missing_header('Bios-Version', bios_version) + + +class TestHandlerRecordV4(RecordTestCases): + """ + Test record v4 + """ + @staticmethod + def get_version_records(): + return get_record_v4() + + def test_post_record_v4(self): + headers = get_record_v4() + response = self.client.post('/', headers=headers, data='test') + self.assertTrue(response.status_code == 201, response.data.decode('utf-8')) + + def test_post_missing_eid(self): + self.missing_header('Event-Id', event_id) + + +if __name__ == '__main__' and __package__ is None: + from os import sys, path + sys.path.append(path.dirname(path.dirname(path.abspath(__file__)))) + unittest.main() + + +# vi: ts=4 et sw=4 sts=4 diff --git a/collector/collector/tests/parsers.py b/collector/collector/tests/parsers.py new file mode 100644 index 0000000..fcb69a8 --- /dev/null +++ b/collector/collector/tests/parsers.py @@ -0,0 +1,53 @@ +# +# Copyright 2018 Intel Corporation +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +# + +import base64 +import unittest +from collector import app +from collector.parsers.demo import main +from collector.report_handler import load_parsers +from collector.tests.testcase import ( + RecordTestCases, + get_record_v3,) + + +class TestCasesParserPlugin(RecordTestCases): + """ + Load parsers programmatically for test + """ + def setUp(self): + RecordTestCases.setUp(self) + app.config['POST_PROCESSING_PARSERS'] = ["demo"] + load_parsers() + + +class TestParserPlugin(TestCasesParserPlugin): + """ + Simple test to make sure that plugins parsers are working + """ + def test_record_created(self): + headers = get_record_v3() + data = b'Hello World' + _data = base64.b64encode(data) + headers['classification'] = main.CLASSIFICATIONS + response = self.client.post('/', headers=headers, data=_data) + self.assertTrue(response.status_code == 201, response.data.decode('utf-8')) + + +if __name__ == '__main__' and __package__ is None: + from os import sys, path + sys.path.append(path.dirname(path.dirname(path.abspath(__file__)))) + unittest.main() diff --git a/collector/collector/tests/purging.py b/collector/collector/tests/purging.py new file mode 100644 index 0000000..e5ab494 --- /dev/null +++ b/collector/collector/tests/purging.py @@ -0,0 +1,115 @@ +# +# Copyright 2015-2017 Intel Corporation +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +# + +import unittest +import time +from collector import db +from flask import current_app +from collector.model import ( + app, + Record, + Classification, + Build) + +from collector.tests.testcase import ( + RecordTestCases, + get_record,) + + +def get_insert_params(days_old, severity, classification): + record = get_record() + record["classification"] = classification + record["severity"] = severity + db_class = Classification.query.filter_by(classification=record["classification"]).first() + if db_class is None: + db_class = Classification(record["classification"]) + db_build = Build.query.filter_by(build=record["build"]).first() + if db_build is None: + db_build = Build(record["build"]) + return [ + record["machine_id"], + record["host_type"], + record["severity"], + db_class, + db_build, + record["arch"], + record["kernel_version"], + record["record_format_version"], + int(time.time()-3600*24*days_old), + int(time.time()-3600*24*days_old), + record["payload_format_version"], + record["system_name"], + record["board_name"], + record["bios_version"], + record["cpu_model"], + "39cc109a1079df96376693ebc7a0f632", + False, + "Test" + ] + + +class TestPurging(RecordTestCases): + """ Generic object for telemetry record tests """ + + def setUp(self): + app.testing = True + app.config.from_object('config_local.Testing') + app.config["MAX_DAYS_KEEP_UNFILTERED_RECORDS"] = 5 + app.config["PURGE_FILTERED_RECORDS"] = { + "severity": { + 1: 1, + 4: 0 + }, + "classification": { + "test/keep/one": 0, + "test/discard/*": 1, + } + } + app.debug = False + self.app_context = app.app_context() + self.app_context.push() + db.init_app(current_app) + db.create_all() + self.client = app.test_client() + + def test_purge_delete(self): + Record.create(*get_insert_params(2, 1, "test/discard/one")) + Record.create(*get_insert_params(2, 2, "test/discard/two")) + Record.create(*get_insert_params(2, 2, "test/discard/three")) + Record.create(*get_insert_params(2, 4, "test/discard/two")) + Record.create(*get_insert_params(2, 4, "test/discard/three")) + Record.create(*get_insert_params(6, 2, "test/test/one")) + Record.create(*get_insert_params(2, 1, "test/keep/one")) + self.assertTrue(Record.query.count() == 7) + Record.delete_records() + self.assertTrue(Record.query.count() == 0) + + def test_purge_keep(self): + Record.create(*get_insert_params(6, 2, "test/keep/one")) + Record.create(*get_insert_params(3, 2, "test/test/one")) + Record.create(*get_insert_params(6, 4, "test/test/one")) + Record.create(*get_insert_params(2, 4, "test/discard/three")) + Record.create(*get_insert_params(6, 2, "test/test/one")) + Record.create(*get_insert_params(2, 1, "test/keep/one")) + self.assertTrue(Record.query.count() == 6) + Record.delete_records() + self.assertTrue(len(Record.query.all()) == 3) + + +if __name__ == '__main__' and __package__ is None: + from os import sys, path + sys.path.append(path.dirname(path.dirname(path.abspath(__file__)))) + unittest.main() diff --git a/collector/collector/tests/testcase.py b/collector/collector/tests/testcase.py new file mode 100644 index 0000000..4accf57 --- /dev/null +++ b/collector/collector/tests/testcase.py @@ -0,0 +1,143 @@ +# +# Copyright 2015-2017 Intel Corporation +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +# + +import unittest +from collector import ( + app, + db, + report_handler,) +from flask import current_app + +classification = 'classification' +severity = 'severity' +kernel_version = 'kernel_version' +record_version = 'record_format_version' +machine_id = 'machine_id' +host_type = 'host_type' +arch = 'arch' +build = 'build' +timestamp = 'creation_timestamp' +tid = 'X-Telemetry-Tid' +board_name = 'Board-Name' +cpu_model = 'Cpu-Model' +bios_version = 'Bios-Version' +system_name = 'System-Name' +payload_version = 'Payload-Format-Version' +event_id = 'Event-Id' + + +REQUIRED_HEADERS_V1 = ( + 'Arch', + 'Build', + 'Creation-Timestamp', + 'Classification', + 'Host-Type', + 'Kernel-Version', + 'Machine-Id', + 'Severity', + 'Record-Format-Version', +) + + +def get_record_v1(): + return { + arch: 'x86_64', + build: '550', + timestamp: 1483232401, + classification: 'a/b/c', + host_type: 'LenovoT20', + kernel_version: '3.16.4-123.generic', + machine_id: '1234', + severity: 2, + record_version: 1, + } + + +def get_record_v2(): + v2 = get_record_v1() + v2.update({ + record_version: 2, + tid: '6907c830-eed9-4ce9-81ae-76daf8d88f0f', + system_name: 'clear-linux-os', + payload_version: 1 + }) + return v2 + + +def get_record_v3(): + v3 = get_record_v2() + v3.update({ + record_version: 3, + board_name: 'D54250WYK|Intel Corporation', + cpu_model: 'Intel(R) Core(TM) i5-4250U CPU @ 1.30GHz', + bios_version: 'WYLPT10H.86A.0041.2015.0720.1108', + + }) + return v3 + + +def get_record_v4(): + v4 = get_record_v3() + v4.update({ + record_version: 4, + event_id: '39cc109a1079df96376693ebc7a0f632', + }) + return v4 + + +def get_record(): + return { + "X-Telemetry-TID": "6907c830-eed9-4ce9-81ae-76daf8d88f0f", + "record_format_version": "2", + "severity": "1", + "classification": "org.clearlinux/hello/world", + "machine_id": "clr-linux-avj01", + "creation_timestamp": "1505235249", + "arch": "x86_64", + "host_type": "blank|blank|blank", + "kernel_version": "4.12.5-374.native", + "system_name": "clear-linux-os", + "build": "17700", + "payload_format_version": "1", + "board_name": "D54250WYK|Intel Corporation", + "cpu_model": "Intel(R) Core(TM) i5-4250U CPU @ 1.30GHz", + "bios_version": "WYLPT10H.86A.0041.2015.0720.1108" + } + + +class RecordTestCases(unittest.TestCase): + """ Generic object for telemetry record tests """ + + def setUp(self): + app.testing = True + app.config.from_object('collector.config_local.Testing') + app.debug = False + self.app_context = app.app_context() + self.app_context.push() + db.init_app(current_app) + db.create_all() + self.client = app.test_client() + + def tearDown(self): + db.session.remove() + db.drop_all() + self.app_context.pop() + + def missing_header(self, header, header_name): + headers = self.get_version_records() + del headers[header_name] + response = self.client.post('/', headers=headers, data='test') + self.assertTrue(response.status_code == 400) diff --git a/collector/collector/tests/validation.py b/collector/collector/tests/validation.py new file mode 100644 index 0000000..217043b --- /dev/null +++ b/collector/collector/tests/validation.py @@ -0,0 +1,45 @@ +# +# Copyright 2015-2017 Intel Corporation +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +# + +import unittest +from collector.tests.testcase import ( + RecordTestCases, + get_record_v3, + kernel_version) + + +class TestHandlerRecordValidation(RecordTestCases): + """ + Test record validation + """ + @staticmethod + def get_version_records(): + return get_record_v3() + + def test_post_kernel_version_validation_1(self): + headers = get_record_v3() + headers.update({kernel_version: '3.16.generic'}) + response = self.client.post('/', headers=headers, data='test') + self.assertTrue(response.status_code == 201, response.data.decode('utf-8')) + + +if __name__ == '__main__' and __package__ is None: + from os import sys, path + sys.path.append(path.dirname(path.dirname(path.abspath(__file__)))) + unittest.main() + + +# vi: ts=4 et sw=4 sts=4 diff --git a/scripts/collector_uwsgi.ini b/scripts/collector_uwsgi.ini new file mode 100644 index 0000000..44ddc4c --- /dev/null +++ b/scripts/collector_uwsgi.ini @@ -0,0 +1,50 @@ +[uwsgi] +#application's base folder +chdir = @@install_path@@collector/ + +#python module to import +virtualenv = @@install_path@@venv/ +module = collector:app +plugins = python3 + +#socket file's location +socket = @@install_path@@collector/%n.sock + +#permissions for the socket file +chmod-socket = 644 + +processes = 4 + +threads = 2 + +#enable thread support.Need to figure out the optimal no of threads. +enable-threads = true + +#location of log files +logto = /var/log/uwsgi/%n.log + +#maximum size of log file before rotation (100MB) +log-maxsize = 104857600 + +#backup log file (created after rotation) +log-backupname = /var/log/uwsgi/%n.log.bk + +#if request takes more than this parameter(in sec), request will be dropped +harakiri = 60 + +#respawn processes after serving 5000 requests +max-requests = 5000 + +#get verbose logs when a process gets stuck +harakiri-verbose = true + +#http://uwsgi-docs.readthedocs.org/en/latest/Tracebacker.html#combining-the-tracebacker-with-harakiri +#Traceback is automatically logged during harakiri phase. +py-tracebacker=collectorsocket + +#enable stats. Use this for fine-tuning no of processes. +#connect uwsgitop to the stats socket as: uwsgitop /tmp/collectorstats.socket +stats=/tmp/collectorstats.socket + +#for asynchronous processing (e.g. updating guilty data for crash records) +spooler = %(chdir)/uwsgi-spool diff --git a/scripts/nginx.conf b/scripts/nginx.conf index d34b847..09b05d3 100644 --- a/scripts/nginx.conf +++ b/scripts/nginx.conf @@ -20,10 +20,10 @@ http { tcp_nodelay on; keepalive_timeout 65; types_hash_max_size 2048; - server_tokens off; include /etc/nginx/mime.types; default_type application/octet-stream; + server_tokens off; include /etc/nginx/conf.d/*.conf; diff --git a/shared/crash.py b/shared/crash.py new file mode 100644 index 0000000..bec57a6 --- /dev/null +++ b/shared/crash.py @@ -0,0 +1,433 @@ +# +# Copyright 2015-2017 Intel Corporation +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +# + +import json +from operator import itemgetter +from collections import namedtuple +import subprocess +import re +from .model import Record, GuiltyBlacklist +from . import app + +try: + from uwsgidecorators import spool +except ImportError: + def spool(f): + f.spool = f + return f + +filters = [] + +# Groups for the frame_pattern below +# 1 - frame number + one space +# 2 - function name + optional arguments +# 3 - rest of the line +# 4 - module name (inside the []) +# 5 - optional frame source file and line number info + +# TODO: The current c++filt logic depends on properly subsituting c++filt +# output for the function name. Thus, it is very, very important to keep a +# capture group that extends from the function name to the end of the frame as +# long as this logic remains the same. Probably better to rework the code to +# *not* destructively overwrite the backtrace field. Maybe store the filtered +# output in a different field. + +frame_pattern = "^(#\d+ )(.+)( - \[(.*)\](.*))$" + +backtrace_classes = [ + 'org.clearlinux/crash/clr', + 'org.clearlinux/kernel/bug', + 'org.clearlinux/kernel/stackoverflow', + 'org.clearlinux/kernel/warning' +] + +other_classes = [ + 'org.clearlinux/crash/unknown', + 'org.clearlinux/crash/clr-build', + 'org.clearlinux/crash/error' +] + + +def get_all_classes(): + return backtrace_classes + other_classes + + +def get_backtrace_classes(): + return backtrace_classes + + +def get_other_classes(): + return other_classes + + +def is_crash_classification(klass): + return (klass in backtrace_classes) and True or False + + +def is_blacklisted(function, module): + funcmod = (function, module) + if funcmod in filters: + return True + + return False + + +def demangle_backtrace(bt): + new_bt = [] + # FIXME: c++filt should be replaced before demangle_backtrace is used again + prog = '/usr/bin/c++filt' + + frame_regex = re.compile(frame_pattern) + lines = bt.splitlines() + + for line in lines: + m = frame_regex.match(line) + if m: + func = m.group(2) + + # A frame with missing symbols is a special case, so skip it + if func == '???': + new_bt.append(line) + continue + + # FIXME: this logic will break once the crash probe starts sending + # function argument values; make this more generic! + if func[-2:] == '()': + # The crash probe adds the () to the function name, but c++filt + # cannot demangle a symbol with the () suffix + func_name = func[:-2] + else: + # Assume already demangled, or this is from a kernel crash record + new_bt.append(line) + continue + + try: + new_func = subprocess.check_output([prog, func_name], universal_newlines=True) + except: + new_bt.append(line) + continue + + # c++filt adds a trailing newline to the output + new_func = new_func.rstrip() + + # Restore () if this was not a mangled symbol + if new_func == func_name: + new_func = func_name + '()' + + repl_str = r'\1{}\3'.format(new_func) + new_line = frame_regex.sub(repl_str, line) + new_bt.append(new_line) + else: + new_bt.append(line) + + return '\n'.join(new_bt) + + +def find_guilty(backtrace): + frame_regex = re.compile(frame_pattern) + + guilty = {} + + lines = backtrace.splitlines() + + first_unknown = None + in_backtrace = False + prev_frame = None + found_match = False + found_unknown = False + + # Begin guilty detection process + for line in lines[1:]: + m = frame_regex.match(line) + + if m: + # Either this is the first frame of the backtrace, or we are still + # iterating through the backtrace. + in_backtrace = True + + func = m.group(2) + mod = m.group(4) + + # Only consider blacklisted function/module pairs as a last resort. + # It's likely that the blacklisted pairs will never be chosen as + # worthy candidates... if they are, the guilty blacklist may be + # filtering too much. + if is_blacklisted(func, mod): + prev_frame = (func, mod) + continue + + # Consider the first frame without function symbols ('???') only if + # there are no function symbols for any frames lower in the stack. + if (func == '???' or func[:2] == '? ') and not found_unknown: + found_unknown = True + first_unknown = (func, mod) + prev_frame = (func, mod) + continue + elif func == '???': + # In this case, we've already encountered a frame with missing + # function symbols, so skip it, but save the info for backup. + prev_frame = (func, mod) + continue + + # If the previous three conditional checks fail, then we have found + # the best guilty candidate: it is not in the blacklist, and it has + # function symbols. + guilty['function'] = func + guilty['module'] = mod + guilty['count'] = 1 + + found_match = True + return (guilty, found_match) + + elif in_backtrace: + # We have processed the entire backtrace for the crashing thread of + # the process, but no solid guilty has been found. Since we only + # consider the crashing thread for guilty detection, stop iterating + # through the remainder of the threads at this point. + break + + # Implement a backup plan to ensure that a guilty is chosen. + if found_unknown: + # Take preference for '???' + guilty['function'] = first_unknown[0] + guilty['module'] = first_unknown[1] + guilty['count'] = 1 + found_match = True + elif prev_frame: + # Choose the previous frame as a last resort + guilty['function'] = prev_frame[0] + guilty['module'] = prev_frame[1] + guilty['count'] = 1 + found_match = True + + return (guilty, found_match) + + +def _process_guilties(args): + if isinstance(args['klass'], bytes): + klass = args['klass'].decode() + else: + klass = args['klass'] + # In case the caller does not check for proper classification, bail early + if not is_crash_classification(klass): + return + if 'id' in args: + record_id = int(args['id']) + else: + record_id = None + global filters + with app.app_context(): + crashes = Record.get_new_crash_records(classes=get_backtrace_classes(), id=record_id) + filters = GuiltyBlacklist.get_guilties() + for rec in crashes: + if rec.payload: + # TODO: re-add demangling capabilities when a better solution is found + # TODO: update the rec.payload field as well + Record.commit_guilty_changes() + g, match = find_guilty(rec.payload) + if match: + function = g['function'] + module = g['module'] + db_guilty = Record.get_guilty_for_funcmod(function, module) + if db_guilty is None: + db_guilty = Record.init_guilty(function, module) + Record.create_guilty_for_record(rec, db_guilty) + Record.set_processed_flag(rec) + + Record.commit_guilty_changes() + + +@spool +def process_guilties(args): + _process_guilties(args) + + +def process_guilties_sync(**args): + _process_guilties(args) + + +def guilty_list_per_build(guilties): + # TODO: should compute max values per build with a subquery instead + build_maxcount = {} + + buildset = set() + buildlist = [] + newlist = [] + + for g in guilties: + found_entry = False + guilty_str = g[0] + ' - [' + g[1] + ']' + build, count, guilty_id, comment = (g[2], g[3], g[4], g[5]) + for i, n in enumerate(newlist): + if guilty_str == n['guilty']: + newlist[i]['total'] += count + newlist[i]['builds'].append((build, count)) + if build in build_maxcount: + build_maxcount[build] = max(build_maxcount[build], count) + else: + build_maxcount[build] = count + found_entry = True + break + + if found_entry: + continue + + entry = {} + entry['guilty'] = guilty_str + entry['total'] = count + entry['guilty_id'] = guilty_id + entry['comment'] = comment + entry['builds'] = [] + entry['builds'].append((build, count)) + if build in build_maxcount: + build_maxcount[build] = max(build_maxcount[build], count) + else: + build_maxcount[build] = count + newlist.append(entry) + + # We only care about the top 10 guilties + newlist = sorted(newlist, key=itemgetter('total'), reverse=True)[:10] + for guilty in newlist: + for build in guilty['builds']: + buildset.add(build[0]) + + buildlist = list(buildset) + buildlist = sorted(buildlist, key=lambda b: int(b[0]), reverse=True) + + # For crashes not occuring in a particular build, provide a "0" value for + # the count. This simplifies table generation in the jinja template. + for i, g in enumerate(newlist): + builds, counts = list(zip(*g['builds'])) + counter = 0 + for b in buildlist: + if b not in builds: + newlist[i]['builds'].insert(counter, (b, "0")) + counter += 1 + + for i, b in enumerate(buildlist): + buildlist[i] = (b, build_maxcount[b]) + + buildlist = sorted(buildlist, key=lambda b: int(b[0]), reverse=True) + + for i, b in enumerate(newlist): + newlist[i]['builds'] = sorted(newlist[i]['builds'], key=lambda b: int(b[0]), reverse=True) + + return (buildlist, newlist) + + +def guilty_list_for_build(guilties, filter='overall'): + newlist = [] + + for g in guilties: + found_entry = False + guilty_str = g[0] + ' - [' + g[1] + ']' + build, count, guilty_id, comment = (g[2], g[3], g[4], g[5]) + for i, n in enumerate(newlist): + if guilty_str == n['guilty'] and filter in ['overall', build]: + newlist[i]['total'] += count + found_entry = True + break + + if found_entry: + continue + + if filter in ['overall', build]: + entry = {} + entry['guilty'] = guilty_str + entry['total'] = count + entry['guilty_id'] = guilty_id + entry['comment'] = comment + newlist.append(entry) + + # We only care about the top 10 guilties + newlist = sorted(newlist, key=itemgetter('total'), reverse=True)[:10] + + return newlist + + +def get_all_funcmods(): + frame_regex = re.compile(frame_pattern) + funcmodset = set() + funcmodlist = [] + backtraces = Record.get_crash_backtraces(classes=get_backtrace_classes()) + + for b in backtraces: + lines = b[0].splitlines() + for line in lines: + match = frame_regex.match(line) + if match: + funcmodset.add((match.group(2), match.group(4))) + + return sorted(funcmodset) + + +def parse_backtrace(backtrace): + program_regex = re.compile('^Process: (.*)$') + pid_regex = re.compile('^PID: ([0-9]+)$') + signal_regex = re.compile('^Signal: ([0-9]+)$') + bt_header_regex = re.compile('^Backtrace \(TID ([0-9]+)\):$') + frame_regex = re.compile(frame_pattern) + + Crash = namedtuple('Crash', ['record_id', 'program', 'pid', 'signal', 'backtrace']) + program = '' + pid = '' + signal = '' + frames = [] + parsed_header = False + + lines = backtrace[0].splitlines() + for line in lines: + # Header info + match = program_regex.match(line) + if match: + program = match.group(1) + continue + match = pid_regex.match(line) + if match: + pid = match.group(1) + continue + match = signal_regex.match(line) + if match: + signal = match.group(1) + continue + + # We only care about the backtrace from the crashing thread, which + # is listed first in the payload. + match = bt_header_regex.match(line) + if match and not parsed_header: + parsed_header = True + elif match: + break + + match = frame_regex.match(line) + if match: + frames.append((match.group(2), match.group(4), match.group(5))) + + # Populate a namedtuple for convenience + record_id = backtrace[1] + c = Crash(record_id, program, pid, signal, frames) + return c + + +def explode_backtraces(classes=None, guilty_id=None, machine_id=None, build=None): + crashes = [] + backtraces = Record.get_crash_backtraces(classes, guilty_id, machine_id, build) + for b in backtraces: + crashes.append(parse_backtrace(b)) + return crashes + + +# vi: ts=4 et sw=4 sts=4 diff --git a/shared/model.py b/shared/model.py new file mode 100644 index 0000000..775840b --- /dev/null +++ b/shared/model.py @@ -0,0 +1,619 @@ +# +# Copyright 2015-2017 Intel Corporation +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +# + +import itertools +from flask_sqlalchemy import SQLAlchemy +from sqlalchemy.sql.expression import cast +from sqlalchemy.sql.expression import desc +from sqlalchemy.sql.expression import case +from time import time, localtime, strftime, mktime, strptime, gmtime +from distutils.version import LooseVersion + +from . import app + +db = SQLAlchemy(app) + + +class Guilty(db.Model): + __tablename__ = 'guilty' + id = db.Column(db.Integer, primary_key=True) + function = db.Column(db.String) + module = db.Column(db.String) + comment = db.Column(db.String) + hide = db.Column(db.Boolean, default=False) + + def __init__(self, func, mod): + self.function = func + self.module = mod + + @staticmethod + def update_comment(guilty_id, comment): + guilty = Guilty.query.filter_by(id=guilty_id).first() + guilty.comment = comment + db.session.commit() + + @staticmethod + def get_function(guilty_id): + guilty = Guilty.query.filter_by(id=guilty_id).first() + return guilty and guilty.function or "" + + @staticmethod + def get_module(guilty_id): + guilty = Guilty.query.filter_by(id=guilty_id).first() + return guilty and guilty.module or "" + + @staticmethod + def update_hidden(guilty_id, status): + guilty = Guilty.query.filter_by(id=guilty_id).first() + guilty.hide = status + db.session.commit() + + @staticmethod + def get_hidden_value(guilty_id): + guilty = Guilty.query.filter_by(id=guilty_id).first() + return guilty and guilty.hide or False + + @staticmethod + def get_hidden_guilties(): + q = db.session.query(Guilty.id, Guilty.function, Guilty.module) + q = q.filter(Guilty.hide == True) + q = q.order_by(Guilty.function) + return q.all() + + +class Record(db.Model): + __tablename__ = 'records' + id = db.Column(db.Integer, primary_key=True) + architecture = db.Column(db.Text) + bios_version = db.Column(db.Text, default='') + board_name = db.Column(db.Text, default='') + build = db.Column(db.Text, nullable=False) + classification = db.Column(db.Text, nullable=False) + cpu_model = db.Column(db.Text, default='') + event_id = db.Column(db.Text, default='') + external = db.Column(db.Boolean, default=False) + host_type = db.Column(db.Text, default='') + kernel_version = db.Column(db.Text, default=0) + machine_id = db.Column(db.Text, default='') + payload_version = db.Column(db.Integer) + record_version = db.Column(db.Integer, default=0) + severity = db.Column(db.Integer) + system_name = db.Column(db.Text) + timestamp_client = db.Column(db.Numeric) + timestamp_server = db.Column(db.Numeric, nullable=False) + payload = db.Column(db.Text, nullable=False) + + processed = db.Column(db.Boolean, default=False) + guilty_id = db.Column(db.Integer, db.ForeignKey('guilty.id')) + + guilty = db.Column(db.Text, default='') + guilty = db.relationship('Guilty', backref=db.backref('records', lazy='dynamic'), lazy='joined') + + def __init__(self, machine_id, host_type, severity, classification, build, architecture, kernel_version, + record_version, ts_capture, ts_reception, payload_version, system_name, + board_name, bios_version, cpu_model, event_id, external, payload): + self.machine_id = machine_id + self.host_type = host_type + self.architecture = architecture + self.classification = classification + self.build = build + self.kernel_version = kernel_version + self.record_version = record_version + self.severity = severity + self.timestamp_client = ts_capture + self.timestamp_server = ts_reception + self.payload_version = payload_version + self.system_name = system_name + self.external = external + self.board_name = board_name + self.bios_version = bios_version + self.cpu_model = cpu_model + self.event_id = event_id + + try: + self.payload = payload.encode('utf-8') + except UnicodeError: + self.payload = payload.encode('latin-1') + + def __repr__(self): + return "".format(self.id, self.classification, self.build, strftime("%a, %d %b %Y %H:%M:%S", localtime(self.timestamp_client))) + + def __str__(self): + return str(self.to_dict()) + + def to_dict(self): + record = { + 'id': self.id, + 'machine_id': self.machine_id, + 'machine_type': self.host_type, + 'arch': self.architecture, + 'build': self.build, + 'kernel_version': self.kernel_version, + 'ts_capture': strftime('%Y-%m-%d %H:%M:%S UTC', gmtime(self.timestamp_client)), + 'ts_reception': strftime('%Y-%m-%d %H:%M:%S UTC', gmtime(self.timestamp_server)), + 'severity': self.severity, + 'classification': self.classification, + 'record_version': self.record_version, + 'payload': self.payload, + 'board_name': self.board_name, + 'bios_version': self.bios_version, + 'cpu_model': self.cpu_model, + 'event_id': self.event_id, + 'external': self.external, + } + return record + + # for the exported CSV rows + def to_list(self): + record = [ + self.id, + self.external, + self.timestamp_server, + self.severity, + self.classification, + self.build, + self.machine_id, + self.payload + ] + return record + + @staticmethod + def list(): + return Record.query.all() + + @staticmethod + def create(machine_id, host_type, severity, classification, build, architecture, kernel_version, + record_version, ts_capture, ts_reception, payload_version, system_name, + board_name, bios_version, cpu_model, event_id, external, payload): + try: + record = Record(machine_id, host_type, severity, classification, build, architecture, kernel_version, + record_version, ts_capture, ts_reception, payload_version, system_name, + board_name, bios_version, cpu_model, event_id, external, payload) + db.session.add(record) + db.session.commit() + return record + except: + db.session.rollback() + raise + + @staticmethod + def query_records(build, classification, severity, machine_id, limit, + interval_sec=None, ts_capture=None, from_id=None): + records = Record.query + if build is not None: + records = records.filter_by(build=build) + if classification is not None: + records = records.filter_by(classification=classification) + if severity is not None: + records = records.filter(Record.severity == severity) + if machine_id is not None: + records = records.filter(Record.machine_id == machine_id) + if from_id is not None: + records = records.filter(Record.id >= from_id) + if ts_capture is not None: + records = records.filter(Record.timestamp_client > ts_capture) + + if interval_sec is not None: + current_time = time() + secs_in_past = current_time - interval_sec + # Due to time skew on client systems, delayed sends due to + # spooling, etc, timestamp_server works better as the reference + # timestamp. + records = records.filter(Record.timestamp_server > secs_in_past) + + records = records.order_by(Record.id.desc()) + + if limit is not None: + records = records.limit(limit) + + return records.all() + + @staticmethod + def get_record(record_id): + record = Record.query.filter_by(id=record_id).first() + return record + + @staticmethod + def filter_records(build, classification, severity, machine_id=None, system_name=None, limit=None, from_date=None, + to_date=None, payload=None, not_payload=None, data_source=None): + records = Record.query + if build is not None: + records = records.filter_by(build=build) + if classification is not None: + if isinstance(classification, list): + records = records.filter(Record.classification.in_(classification)) + else: + records = records.filter(Record.classification.like(classification)) + if severity is not None: + records = records.filter(Record.severity == severity) + if system_name is not None: + records = records.filter(Record.system_name == system_name) + if machine_id is not None: + records = records.filter(Record.machine_id == machine_id) + if from_date is not None: + from_date = mktime(strptime(from_date, "%Y-%m-%d")) + records = records.filter(Record.timestamp_client >= from_date) + if to_date is not None: + to_date = mktime(strptime(to_date, "%Y-%m-%d")) + records = records.filter(Record.timestamp_client < to_date) + if payload is not None: + records = records.filter(Record.payload.op('~')(payload)) + if not_payload is not None: + records = records.filter(~Record.payload.op('~')(not_payload)) + if data_source is not None: + if data_source == "external": + records = records.filter(Record.external == True) + elif data_source == "internal": + records = records.filter(Record.external == False) + + records = records.order_by(Record.id.desc()) + + if limit is not None: + records = records.limit(limit) + + return records + + @staticmethod + def delete_records(): + MAX_DAYS_KEEP_UNFILTERED_RECORDS = app.config.get("MAX_DAYS_KEEP_UNFILTERED_RECORDS", 35) + PURGE_FILTERED_RECORDS = app.config.get("PURGE_FILTERED_RECORDS", {}) + try: + def purge_field(field): + for name in PURGE_FILTERED_RECORDS[field].keys(): + if PURGE_FILTERED_RECORDS[field][name]: + age = time() - PURGE_FILTERED_RECORDS[field][name] * 24 * 60 * 60 + q = db.session.query(Record) + if field == 'classification': + q = q.filter(Record.classification.like(name.replace("*", "%"))) + else: + q = q.filter(getattr(Record, field) == name) + q = q.filter(Record.timestamp_server < age) + if q.all(): + count = db.session.query(Record).filter(Record.id.in_([x.id for x in q.all()])).delete(synchronize_session=False) + print("Deleted {} {} records".format(count, name)) + for field in PURGE_FILTERED_RECORDS.keys(): + purge_field(field) + if MAX_DAYS_KEEP_UNFILTERED_RECORDS: + unfiltered_age = time() - MAX_DAYS_KEEP_UNFILTERED_RECORDS * 24 * 60 * 60 + q = db.session.query(Record.id) + for field in PURGE_FILTERED_RECORDS.keys(): + if field == 'classification': + for classification in PURGE_FILTERED_RECORDS[field].keys(): + q = q.filter(~Record.classification.like(classification.replace("*", "%"))) + else: + for name in PURGE_FILTERED_RECORDS[field].keys(): + q = q.filter(getattr(Record, field) != name) + q = q.filter(Record.timestamp_server < unfiltered_age) + if q.all(): + count = db.session.query(Record).filter(Record.id.in_([x.id for x in q.all()])).delete(synchronize_session=False) + print("Deleted {} old records".format(count)) + db.session.commit() + except Exception as e: + app.logger.error("Record purging failed") + app.logger.error(e) + db.session.rollback() + + @staticmethod + def get_recordcnts_by_build(): + q = db.session.query(Record.build, db.func.count(Record.id)) + q = q.filter(Record.build.op('~')('^[0-9]+$')) + q = q.group_by(Record.build).order_by(cast(Record.build, db.Integer)).all() + return q + + @staticmethod + def get_builds(): + q = db.session.query(Record.build).distinct() + q = q.order_by(Record.build) + return sorted(q.all(), key=lambda x: LooseVersion(x[0]), reverse=True) + + @staticmethod + def get_recordcnts_by_classification(): + q = db.session.query(Record.classification, db.func.count(Record.id).label('total')) + q = q.group_by(Record.classification) + q = q.order_by(desc('total')) + return q.all() + + @staticmethod + def expand_class(D): + A, B, C = D + return ["{}/*".format(A), "{}/{}/*".format(A, B), "{}/{}/{}".format(A, B, C)] + + @staticmethod + def get_classifications(with_regex=False): + q = db.session.query(Record.classification).distinct() + if with_regex: + classes = [Record.expand_class(c[0].split('/')) for c in q.all()] + return sorted(set(itertools.chain(*classes))) + else: + return q.all() + + @staticmethod + def get_os_map(): + q = db.session.query(Record.system_name, Record.build).order_by(Record.system_name).group_by(Record.system_name, Record.build).all() + result = {} + for x in q: + result.setdefault(x[0], []).append(x[1]) + return result + + @staticmethod + def get_recordcnts_by_machine_type(): + q = db.session.query(Record.host_type, db.func.count(Record.id).label('total')) + q = q.group_by(Record.host_type) + q = q.order_by(desc('total')) + return q.all() + + @staticmethod + def get_recordcnts_by_severity(): + q = db.session.query(Record.severity, db.func.count(Record.id)).group_by(Record.severity).all() + return q + + @staticmethod + def get_crashcnts_by_class(classes=None): + q = db.session.query(Record.classification, db.func.count(Record.id)) + if classes: + q = q.filter(Record.classification.in_(classes)) + else: + q = q.filter(Record.classification.like('org.clearlinux/crash/%')) + q = q.group_by(Record.classification) + return q.all() + + @staticmethod + def get_crashcnts_by_build(classes=None): + q = db.session.query(Record.build, db.func.count(Record.id)) + if not classes: + classes = ['org.clearlinux/crash/clr'] + q = q.filter(Record.classification.in_(classes)) + q = q.filter(Record.build.op('~')('^[0-9]+$')) + q = q.group_by(Record.build) + q = q.order_by(desc(cast(Record.build, db.Integer))) + q = q.limit(10) + return q.all() + + @staticmethod + def get_top_crash_guilties(classes=None): + q = db.session.query(Guilty.function, Guilty.module, Record.build, db.func.count(Record.id).label('total'), Guilty.id, Guilty.comment) + q = q.join(Record) + if not classes: + classes = ['org.clearlinux/crash/clr'] + q = q.filter(Record.classification.in_(classes)) + q = q.filter(Record.build.op('~')('^[0-9][0-9]+$')) + q = q.filter(cast(Record.build, db.Integer) <= 100000) + q = q.filter(Guilty.hide == False) + q = q.group_by(Guilty.function, Guilty.module, Guilty.comment, Guilty.id, Record.build) + q = q.order_by(desc(cast(Record.build, db.Integer)), desc('total')) + # query for records created in the last week (~ 10 Clear builds) + q = q.filter(Record.build.in_(sorted(tuple(set([x[2] for x in q.all()])), key=lambda x: int(x))[-8:])) + interval_sec = 24 * 60 * 60 * 7 + current_time = time() + sec_in_past = current_time - interval_sec + q = q.filter(Record.timestamp_client > sec_in_past) + return q.all() + + @staticmethod + def get_new_crash_records(classes=None, id=None): + q = db.session.query(Record) + if not classes: + classes = ['org.clearlinux/crash/clr'] + q = q.filter(Record.classification.in_(classes)) + q = q.filter(Record.system_name == 'clear-linux-os') + q = q.filter(Record.processed == False) + if id: + q = q.filter(Record.id == id) + return q.all() + + @staticmethod + def set_processed_flag(record): + record.processed = True + + @staticmethod + def get_guilty_for_funcmod(func, mod): + q = db.session.query(Guilty).filter_by(function=func, module=mod).first() + return q + + @staticmethod + def get_guilty_id_for_record(record_id): + q = db.session.query(Guilty.id).join(Record) + q = q.filter(Record.id == record_id) + return q.first() + + @staticmethod + def init_guilty(func, mod): + return Guilty(func, mod) + + @staticmethod + def create_guilty_for_record(record, guilty): + record.guilty = guilty + + @staticmethod + def commit_guilty_changes(): + # just commit for now + db.session.commit() + + @staticmethod + def get_crash_backtraces(classes=None, guilty_id=None, machine_id=None, build=None, most_recent=None, record_id=None): + q = db.session.query(Record.payload, Record.id) + # Short circuit if we know the record ID + if record_id: + q = q.filter(Record.id == record_id) + return q.first() + if build: + q = q.filter(Record.build == build) + if not classes: + classes = ['org.clearlinux/crash/clr'] + q = q.filter(Record.classification.in_(classes)) + q = q.filter(Record.system_name == 'clear-linux-os') + if guilty_id: + q = q.filter(Record.guilty_id == guilty_id) + if machine_id: + q = q.filter(Record.machine_id == machine_id) + if most_recent: + interval_sec = 24 * 60 * 60 * int(most_recent) + current_time = time() + sec_in_past = current_time - interval_sec + q = q.filter(Record.timestamp_client > sec_in_past) + return q.all() + + @staticmethod + def reset_processed_records(classes=None, id=None): + q = db.session.query(Record) + if not classes: + classes = ['org.clearlinux/crash/clr'] + q = q.filter(Record.classification.in_(classes)) + q = q.filter(Record.system_name == 'clear-linux-os') + if id: + q = q.filter(Record.id == id) + records = q.all() + for r in records: + r.processed = False + db.session.commit() + + @staticmethod + def get_machine_ids_for_guilty(id, most_recent=None): + q = db.session.query(Record.build, Record.machine_id, db.func.count(Record.id).label('total'), Record.guilty_id) + q = q.filter(Record.guilty_id == id) + q = q.filter(Record.system_name == 'clear-linux-os') + q = q.filter(Record.build.op('~')('^[0-9][0-9]+$')) + q = q.group_by(Record.build, Record.machine_id, Record.guilty_id) + q = q.order_by(desc(cast(Record.build, db.Integer)), desc('total')) + if most_recent: + interval_sec = 24 * 60 * 60 * int(most_recent) + current_time = time() + sec_in_past = current_time - interval_sec + q = q.filter(Record.timestamp_client > sec_in_past) + return q.all() + + @staticmethod + def get_update_msgs(): + q = db.session.query(Record.payload) + q = q.filter(Record.classification == "org.clearlinux/swupd-client/update") + + sec_2_weeks = 24 * 60 * 60 * 7 + current_time = time() + time_2_weeks_ago = current_time - sec_2_weeks + + # query for records created in that last 2 weeks + q = q.filter(Record.timestamp_client > time_2_weeks_ago) + return q.all() + + @staticmethod + def get_swupd_msgs(most_recent=None): + q = db.session.query(Record.timestamp_client, Record.machine_id, Record.payload) + q = q.filter(Record.classification.like('org.clearlinux/swupd-client/%')) + + if most_recent: + interval_sec = 24 * 60 * 60 * int(most_recent) + current_time = time() + sec_in_past = current_time - interval_sec + q = q.filter(Record.timestamp_client > sec_in_past) + + q = q.order_by(desc(Record.timestamp_client)) + return q + + @staticmethod + def get_heartbeat_msgs(most_recent=None): + # These two expressions are SQL CASE conditional expressions, later + # used within count(distinct ...) aggregates for the query. + internal_expr = case([(Record.external == False, Record.machine_id), ]).label('internal_count') + external_expr = case([(Record.external == True, Record.machine_id), ]).label('external_count') + + q = db.session.query(Record.build, db.func.count(db.distinct(internal_expr)), db.func.count(db.distinct(external_expr))) + q = q.filter(Record.classification == "org.clearlinux/heartbeat/ping") + q = q.filter(Record.system_name == 'clear-linux-os') + q = q.group_by(Record.build) + + if most_recent: + interval_sec = 24 * 60 * 60 * int(most_recent) + current_time = time() + sec_in_past = current_time - interval_sec + q = q.filter(Record.timestamp_client > sec_in_past) + + q = q.order_by(cast(Record.build, db.Integer)) + return q.all() + + +class GuiltyBlacklist(db.Model): + __tablename__ = 'guilty_blacklisted' + id = db.Column(db.Integer, primary_key=True) + function = db.Column(db.String) + module = db.Column(db.String) + + def __init__(self, func, mod): + self.function = func + self.module = mod + + def __repr__(self): + return "".format(self.id, self.function, self.module) + + def __str__(self): + return str(self.to_dict()) + + def to_dict(self): + guilty = { + 'function': self.function, + 'module': self.module + } + return guilty + + @staticmethod + def add(func, mod): + try: + g = GuiltyBlacklist(func, mod) + db.session.add(g) + db.session.commit() + return g + except: + db.session.rollback() + raise + + @staticmethod + def remove(func, mod): + q = db.session.query(GuiltyBlacklist) + q = q.filter_by(function=func, module=mod) + entry = q.first() + db.session.delete(entry) + db.session.commit() + + @staticmethod + def get_guilties(): + q = db.session.query(GuiltyBlacklist.function, GuiltyBlacklist.module) + q = q.order_by(GuiltyBlacklist.function) + return q.all() + + @staticmethod + def exists(func, mod): + q = db.session.query(GuiltyBlacklist.function, GuiltyBlacklist.module) + q = q.filter_by(function=func, module=mod) + return len(q.all()) != 0 and True or False + + @staticmethod + def update(to_add, to_remove): + try: + for i in to_add: + if not GuiltyBlacklist.exists(i[0], i[1]): + g = GuiltyBlacklist(i[0], i[1]) + db.session.add(g) + for i in to_remove: + if GuiltyBlacklist.exists(i[0], i[1]): + q = db.session.query(GuiltyBlacklist) + q = q.filter_by(function=i[0], module=i[1]) + entry = q.first() + db.session.delete(entry) + db.session.commit() + except: + db.session.rollback() + raise + +# vi: ts=4 et sw=4 sts=4 diff --git a/telemetryui/telemetryui/crash.py b/telemetryui/telemetryui/crash.py deleted file mode 100644 index bec57a6..0000000 --- a/telemetryui/telemetryui/crash.py +++ /dev/null @@ -1,433 +0,0 @@ -# -# Copyright 2015-2017 Intel Corporation -# -# Licensed under the Apache License, Version 2.0 (the "License"); -# you may not use this file except in compliance with the License. -# You may obtain a copy of the License at -# -# http://www.apache.org/licenses/LICENSE-2.0 -# -# Unless required by applicable law or agreed to in writing, software -# distributed under the License is distributed on an "AS IS" BASIS, -# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -# See the License for the specific language governing permissions and -# limitations under the License. -# - -import json -from operator import itemgetter -from collections import namedtuple -import subprocess -import re -from .model import Record, GuiltyBlacklist -from . import app - -try: - from uwsgidecorators import spool -except ImportError: - def spool(f): - f.spool = f - return f - -filters = [] - -# Groups for the frame_pattern below -# 1 - frame number + one space -# 2 - function name + optional arguments -# 3 - rest of the line -# 4 - module name (inside the []) -# 5 - optional frame source file and line number info - -# TODO: The current c++filt logic depends on properly subsituting c++filt -# output for the function name. Thus, it is very, very important to keep a -# capture group that extends from the function name to the end of the frame as -# long as this logic remains the same. Probably better to rework the code to -# *not* destructively overwrite the backtrace field. Maybe store the filtered -# output in a different field. - -frame_pattern = "^(#\d+ )(.+)( - \[(.*)\](.*))$" - -backtrace_classes = [ - 'org.clearlinux/crash/clr', - 'org.clearlinux/kernel/bug', - 'org.clearlinux/kernel/stackoverflow', - 'org.clearlinux/kernel/warning' -] - -other_classes = [ - 'org.clearlinux/crash/unknown', - 'org.clearlinux/crash/clr-build', - 'org.clearlinux/crash/error' -] - - -def get_all_classes(): - return backtrace_classes + other_classes - - -def get_backtrace_classes(): - return backtrace_classes - - -def get_other_classes(): - return other_classes - - -def is_crash_classification(klass): - return (klass in backtrace_classes) and True or False - - -def is_blacklisted(function, module): - funcmod = (function, module) - if funcmod in filters: - return True - - return False - - -def demangle_backtrace(bt): - new_bt = [] - # FIXME: c++filt should be replaced before demangle_backtrace is used again - prog = '/usr/bin/c++filt' - - frame_regex = re.compile(frame_pattern) - lines = bt.splitlines() - - for line in lines: - m = frame_regex.match(line) - if m: - func = m.group(2) - - # A frame with missing symbols is a special case, so skip it - if func == '???': - new_bt.append(line) - continue - - # FIXME: this logic will break once the crash probe starts sending - # function argument values; make this more generic! - if func[-2:] == '()': - # The crash probe adds the () to the function name, but c++filt - # cannot demangle a symbol with the () suffix - func_name = func[:-2] - else: - # Assume already demangled, or this is from a kernel crash record - new_bt.append(line) - continue - - try: - new_func = subprocess.check_output([prog, func_name], universal_newlines=True) - except: - new_bt.append(line) - continue - - # c++filt adds a trailing newline to the output - new_func = new_func.rstrip() - - # Restore () if this was not a mangled symbol - if new_func == func_name: - new_func = func_name + '()' - - repl_str = r'\1{}\3'.format(new_func) - new_line = frame_regex.sub(repl_str, line) - new_bt.append(new_line) - else: - new_bt.append(line) - - return '\n'.join(new_bt) - - -def find_guilty(backtrace): - frame_regex = re.compile(frame_pattern) - - guilty = {} - - lines = backtrace.splitlines() - - first_unknown = None - in_backtrace = False - prev_frame = None - found_match = False - found_unknown = False - - # Begin guilty detection process - for line in lines[1:]: - m = frame_regex.match(line) - - if m: - # Either this is the first frame of the backtrace, or we are still - # iterating through the backtrace. - in_backtrace = True - - func = m.group(2) - mod = m.group(4) - - # Only consider blacklisted function/module pairs as a last resort. - # It's likely that the blacklisted pairs will never be chosen as - # worthy candidates... if they are, the guilty blacklist may be - # filtering too much. - if is_blacklisted(func, mod): - prev_frame = (func, mod) - continue - - # Consider the first frame without function symbols ('???') only if - # there are no function symbols for any frames lower in the stack. - if (func == '???' or func[:2] == '? ') and not found_unknown: - found_unknown = True - first_unknown = (func, mod) - prev_frame = (func, mod) - continue - elif func == '???': - # In this case, we've already encountered a frame with missing - # function symbols, so skip it, but save the info for backup. - prev_frame = (func, mod) - continue - - # If the previous three conditional checks fail, then we have found - # the best guilty candidate: it is not in the blacklist, and it has - # function symbols. - guilty['function'] = func - guilty['module'] = mod - guilty['count'] = 1 - - found_match = True - return (guilty, found_match) - - elif in_backtrace: - # We have processed the entire backtrace for the crashing thread of - # the process, but no solid guilty has been found. Since we only - # consider the crashing thread for guilty detection, stop iterating - # through the remainder of the threads at this point. - break - - # Implement a backup plan to ensure that a guilty is chosen. - if found_unknown: - # Take preference for '???' - guilty['function'] = first_unknown[0] - guilty['module'] = first_unknown[1] - guilty['count'] = 1 - found_match = True - elif prev_frame: - # Choose the previous frame as a last resort - guilty['function'] = prev_frame[0] - guilty['module'] = prev_frame[1] - guilty['count'] = 1 - found_match = True - - return (guilty, found_match) - - -def _process_guilties(args): - if isinstance(args['klass'], bytes): - klass = args['klass'].decode() - else: - klass = args['klass'] - # In case the caller does not check for proper classification, bail early - if not is_crash_classification(klass): - return - if 'id' in args: - record_id = int(args['id']) - else: - record_id = None - global filters - with app.app_context(): - crashes = Record.get_new_crash_records(classes=get_backtrace_classes(), id=record_id) - filters = GuiltyBlacklist.get_guilties() - for rec in crashes: - if rec.payload: - # TODO: re-add demangling capabilities when a better solution is found - # TODO: update the rec.payload field as well - Record.commit_guilty_changes() - g, match = find_guilty(rec.payload) - if match: - function = g['function'] - module = g['module'] - db_guilty = Record.get_guilty_for_funcmod(function, module) - if db_guilty is None: - db_guilty = Record.init_guilty(function, module) - Record.create_guilty_for_record(rec, db_guilty) - Record.set_processed_flag(rec) - - Record.commit_guilty_changes() - - -@spool -def process_guilties(args): - _process_guilties(args) - - -def process_guilties_sync(**args): - _process_guilties(args) - - -def guilty_list_per_build(guilties): - # TODO: should compute max values per build with a subquery instead - build_maxcount = {} - - buildset = set() - buildlist = [] - newlist = [] - - for g in guilties: - found_entry = False - guilty_str = g[0] + ' - [' + g[1] + ']' - build, count, guilty_id, comment = (g[2], g[3], g[4], g[5]) - for i, n in enumerate(newlist): - if guilty_str == n['guilty']: - newlist[i]['total'] += count - newlist[i]['builds'].append((build, count)) - if build in build_maxcount: - build_maxcount[build] = max(build_maxcount[build], count) - else: - build_maxcount[build] = count - found_entry = True - break - - if found_entry: - continue - - entry = {} - entry['guilty'] = guilty_str - entry['total'] = count - entry['guilty_id'] = guilty_id - entry['comment'] = comment - entry['builds'] = [] - entry['builds'].append((build, count)) - if build in build_maxcount: - build_maxcount[build] = max(build_maxcount[build], count) - else: - build_maxcount[build] = count - newlist.append(entry) - - # We only care about the top 10 guilties - newlist = sorted(newlist, key=itemgetter('total'), reverse=True)[:10] - for guilty in newlist: - for build in guilty['builds']: - buildset.add(build[0]) - - buildlist = list(buildset) - buildlist = sorted(buildlist, key=lambda b: int(b[0]), reverse=True) - - # For crashes not occuring in a particular build, provide a "0" value for - # the count. This simplifies table generation in the jinja template. - for i, g in enumerate(newlist): - builds, counts = list(zip(*g['builds'])) - counter = 0 - for b in buildlist: - if b not in builds: - newlist[i]['builds'].insert(counter, (b, "0")) - counter += 1 - - for i, b in enumerate(buildlist): - buildlist[i] = (b, build_maxcount[b]) - - buildlist = sorted(buildlist, key=lambda b: int(b[0]), reverse=True) - - for i, b in enumerate(newlist): - newlist[i]['builds'] = sorted(newlist[i]['builds'], key=lambda b: int(b[0]), reverse=True) - - return (buildlist, newlist) - - -def guilty_list_for_build(guilties, filter='overall'): - newlist = [] - - for g in guilties: - found_entry = False - guilty_str = g[0] + ' - [' + g[1] + ']' - build, count, guilty_id, comment = (g[2], g[3], g[4], g[5]) - for i, n in enumerate(newlist): - if guilty_str == n['guilty'] and filter in ['overall', build]: - newlist[i]['total'] += count - found_entry = True - break - - if found_entry: - continue - - if filter in ['overall', build]: - entry = {} - entry['guilty'] = guilty_str - entry['total'] = count - entry['guilty_id'] = guilty_id - entry['comment'] = comment - newlist.append(entry) - - # We only care about the top 10 guilties - newlist = sorted(newlist, key=itemgetter('total'), reverse=True)[:10] - - return newlist - - -def get_all_funcmods(): - frame_regex = re.compile(frame_pattern) - funcmodset = set() - funcmodlist = [] - backtraces = Record.get_crash_backtraces(classes=get_backtrace_classes()) - - for b in backtraces: - lines = b[0].splitlines() - for line in lines: - match = frame_regex.match(line) - if match: - funcmodset.add((match.group(2), match.group(4))) - - return sorted(funcmodset) - - -def parse_backtrace(backtrace): - program_regex = re.compile('^Process: (.*)$') - pid_regex = re.compile('^PID: ([0-9]+)$') - signal_regex = re.compile('^Signal: ([0-9]+)$') - bt_header_regex = re.compile('^Backtrace \(TID ([0-9]+)\):$') - frame_regex = re.compile(frame_pattern) - - Crash = namedtuple('Crash', ['record_id', 'program', 'pid', 'signal', 'backtrace']) - program = '' - pid = '' - signal = '' - frames = [] - parsed_header = False - - lines = backtrace[0].splitlines() - for line in lines: - # Header info - match = program_regex.match(line) - if match: - program = match.group(1) - continue - match = pid_regex.match(line) - if match: - pid = match.group(1) - continue - match = signal_regex.match(line) - if match: - signal = match.group(1) - continue - - # We only care about the backtrace from the crashing thread, which - # is listed first in the payload. - match = bt_header_regex.match(line) - if match and not parsed_header: - parsed_header = True - elif match: - break - - match = frame_regex.match(line) - if match: - frames.append((match.group(2), match.group(4), match.group(5))) - - # Populate a namedtuple for convenience - record_id = backtrace[1] - c = Crash(record_id, program, pid, signal, frames) - return c - - -def explode_backtraces(classes=None, guilty_id=None, machine_id=None, build=None): - crashes = [] - backtraces = Record.get_crash_backtraces(classes, guilty_id, machine_id, build) - for b in backtraces: - crashes.append(parse_backtrace(b)) - return crashes - - -# vi: ts=4 et sw=4 sts=4 diff --git a/telemetryui/telemetryui/crash.py b/telemetryui/telemetryui/crash.py new file mode 120000 index 0000000..68dab23 --- /dev/null +++ b/telemetryui/telemetryui/crash.py @@ -0,0 +1 @@ +../../shared/crash.py \ No newline at end of file diff --git a/telemetryui/telemetryui/model.py b/telemetryui/telemetryui/model.py deleted file mode 100644 index 775840b..0000000 --- a/telemetryui/telemetryui/model.py +++ /dev/null @@ -1,619 +0,0 @@ -# -# Copyright 2015-2017 Intel Corporation -# -# Licensed under the Apache License, Version 2.0 (the "License"); -# you may not use this file except in compliance with the License. -# You may obtain a copy of the License at -# -# http://www.apache.org/licenses/LICENSE-2.0 -# -# Unless required by applicable law or agreed to in writing, software -# distributed under the License is distributed on an "AS IS" BASIS, -# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -# See the License for the specific language governing permissions and -# limitations under the License. -# - -import itertools -from flask_sqlalchemy import SQLAlchemy -from sqlalchemy.sql.expression import cast -from sqlalchemy.sql.expression import desc -from sqlalchemy.sql.expression import case -from time import time, localtime, strftime, mktime, strptime, gmtime -from distutils.version import LooseVersion - -from . import app - -db = SQLAlchemy(app) - - -class Guilty(db.Model): - __tablename__ = 'guilty' - id = db.Column(db.Integer, primary_key=True) - function = db.Column(db.String) - module = db.Column(db.String) - comment = db.Column(db.String) - hide = db.Column(db.Boolean, default=False) - - def __init__(self, func, mod): - self.function = func - self.module = mod - - @staticmethod - def update_comment(guilty_id, comment): - guilty = Guilty.query.filter_by(id=guilty_id).first() - guilty.comment = comment - db.session.commit() - - @staticmethod - def get_function(guilty_id): - guilty = Guilty.query.filter_by(id=guilty_id).first() - return guilty and guilty.function or "" - - @staticmethod - def get_module(guilty_id): - guilty = Guilty.query.filter_by(id=guilty_id).first() - return guilty and guilty.module or "" - - @staticmethod - def update_hidden(guilty_id, status): - guilty = Guilty.query.filter_by(id=guilty_id).first() - guilty.hide = status - db.session.commit() - - @staticmethod - def get_hidden_value(guilty_id): - guilty = Guilty.query.filter_by(id=guilty_id).first() - return guilty and guilty.hide or False - - @staticmethod - def get_hidden_guilties(): - q = db.session.query(Guilty.id, Guilty.function, Guilty.module) - q = q.filter(Guilty.hide == True) - q = q.order_by(Guilty.function) - return q.all() - - -class Record(db.Model): - __tablename__ = 'records' - id = db.Column(db.Integer, primary_key=True) - architecture = db.Column(db.Text) - bios_version = db.Column(db.Text, default='') - board_name = db.Column(db.Text, default='') - build = db.Column(db.Text, nullable=False) - classification = db.Column(db.Text, nullable=False) - cpu_model = db.Column(db.Text, default='') - event_id = db.Column(db.Text, default='') - external = db.Column(db.Boolean, default=False) - host_type = db.Column(db.Text, default='') - kernel_version = db.Column(db.Text, default=0) - machine_id = db.Column(db.Text, default='') - payload_version = db.Column(db.Integer) - record_version = db.Column(db.Integer, default=0) - severity = db.Column(db.Integer) - system_name = db.Column(db.Text) - timestamp_client = db.Column(db.Numeric) - timestamp_server = db.Column(db.Numeric, nullable=False) - payload = db.Column(db.Text, nullable=False) - - processed = db.Column(db.Boolean, default=False) - guilty_id = db.Column(db.Integer, db.ForeignKey('guilty.id')) - - guilty = db.Column(db.Text, default='') - guilty = db.relationship('Guilty', backref=db.backref('records', lazy='dynamic'), lazy='joined') - - def __init__(self, machine_id, host_type, severity, classification, build, architecture, kernel_version, - record_version, ts_capture, ts_reception, payload_version, system_name, - board_name, bios_version, cpu_model, event_id, external, payload): - self.machine_id = machine_id - self.host_type = host_type - self.architecture = architecture - self.classification = classification - self.build = build - self.kernel_version = kernel_version - self.record_version = record_version - self.severity = severity - self.timestamp_client = ts_capture - self.timestamp_server = ts_reception - self.payload_version = payload_version - self.system_name = system_name - self.external = external - self.board_name = board_name - self.bios_version = bios_version - self.cpu_model = cpu_model - self.event_id = event_id - - try: - self.payload = payload.encode('utf-8') - except UnicodeError: - self.payload = payload.encode('latin-1') - - def __repr__(self): - return "".format(self.id, self.classification, self.build, strftime("%a, %d %b %Y %H:%M:%S", localtime(self.timestamp_client))) - - def __str__(self): - return str(self.to_dict()) - - def to_dict(self): - record = { - 'id': self.id, - 'machine_id': self.machine_id, - 'machine_type': self.host_type, - 'arch': self.architecture, - 'build': self.build, - 'kernel_version': self.kernel_version, - 'ts_capture': strftime('%Y-%m-%d %H:%M:%S UTC', gmtime(self.timestamp_client)), - 'ts_reception': strftime('%Y-%m-%d %H:%M:%S UTC', gmtime(self.timestamp_server)), - 'severity': self.severity, - 'classification': self.classification, - 'record_version': self.record_version, - 'payload': self.payload, - 'board_name': self.board_name, - 'bios_version': self.bios_version, - 'cpu_model': self.cpu_model, - 'event_id': self.event_id, - 'external': self.external, - } - return record - - # for the exported CSV rows - def to_list(self): - record = [ - self.id, - self.external, - self.timestamp_server, - self.severity, - self.classification, - self.build, - self.machine_id, - self.payload - ] - return record - - @staticmethod - def list(): - return Record.query.all() - - @staticmethod - def create(machine_id, host_type, severity, classification, build, architecture, kernel_version, - record_version, ts_capture, ts_reception, payload_version, system_name, - board_name, bios_version, cpu_model, event_id, external, payload): - try: - record = Record(machine_id, host_type, severity, classification, build, architecture, kernel_version, - record_version, ts_capture, ts_reception, payload_version, system_name, - board_name, bios_version, cpu_model, event_id, external, payload) - db.session.add(record) - db.session.commit() - return record - except: - db.session.rollback() - raise - - @staticmethod - def query_records(build, classification, severity, machine_id, limit, - interval_sec=None, ts_capture=None, from_id=None): - records = Record.query - if build is not None: - records = records.filter_by(build=build) - if classification is not None: - records = records.filter_by(classification=classification) - if severity is not None: - records = records.filter(Record.severity == severity) - if machine_id is not None: - records = records.filter(Record.machine_id == machine_id) - if from_id is not None: - records = records.filter(Record.id >= from_id) - if ts_capture is not None: - records = records.filter(Record.timestamp_client > ts_capture) - - if interval_sec is not None: - current_time = time() - secs_in_past = current_time - interval_sec - # Due to time skew on client systems, delayed sends due to - # spooling, etc, timestamp_server works better as the reference - # timestamp. - records = records.filter(Record.timestamp_server > secs_in_past) - - records = records.order_by(Record.id.desc()) - - if limit is not None: - records = records.limit(limit) - - return records.all() - - @staticmethod - def get_record(record_id): - record = Record.query.filter_by(id=record_id).first() - return record - - @staticmethod - def filter_records(build, classification, severity, machine_id=None, system_name=None, limit=None, from_date=None, - to_date=None, payload=None, not_payload=None, data_source=None): - records = Record.query - if build is not None: - records = records.filter_by(build=build) - if classification is not None: - if isinstance(classification, list): - records = records.filter(Record.classification.in_(classification)) - else: - records = records.filter(Record.classification.like(classification)) - if severity is not None: - records = records.filter(Record.severity == severity) - if system_name is not None: - records = records.filter(Record.system_name == system_name) - if machine_id is not None: - records = records.filter(Record.machine_id == machine_id) - if from_date is not None: - from_date = mktime(strptime(from_date, "%Y-%m-%d")) - records = records.filter(Record.timestamp_client >= from_date) - if to_date is not None: - to_date = mktime(strptime(to_date, "%Y-%m-%d")) - records = records.filter(Record.timestamp_client < to_date) - if payload is not None: - records = records.filter(Record.payload.op('~')(payload)) - if not_payload is not None: - records = records.filter(~Record.payload.op('~')(not_payload)) - if data_source is not None: - if data_source == "external": - records = records.filter(Record.external == True) - elif data_source == "internal": - records = records.filter(Record.external == False) - - records = records.order_by(Record.id.desc()) - - if limit is not None: - records = records.limit(limit) - - return records - - @staticmethod - def delete_records(): - MAX_DAYS_KEEP_UNFILTERED_RECORDS = app.config.get("MAX_DAYS_KEEP_UNFILTERED_RECORDS", 35) - PURGE_FILTERED_RECORDS = app.config.get("PURGE_FILTERED_RECORDS", {}) - try: - def purge_field(field): - for name in PURGE_FILTERED_RECORDS[field].keys(): - if PURGE_FILTERED_RECORDS[field][name]: - age = time() - PURGE_FILTERED_RECORDS[field][name] * 24 * 60 * 60 - q = db.session.query(Record) - if field == 'classification': - q = q.filter(Record.classification.like(name.replace("*", "%"))) - else: - q = q.filter(getattr(Record, field) == name) - q = q.filter(Record.timestamp_server < age) - if q.all(): - count = db.session.query(Record).filter(Record.id.in_([x.id for x in q.all()])).delete(synchronize_session=False) - print("Deleted {} {} records".format(count, name)) - for field in PURGE_FILTERED_RECORDS.keys(): - purge_field(field) - if MAX_DAYS_KEEP_UNFILTERED_RECORDS: - unfiltered_age = time() - MAX_DAYS_KEEP_UNFILTERED_RECORDS * 24 * 60 * 60 - q = db.session.query(Record.id) - for field in PURGE_FILTERED_RECORDS.keys(): - if field == 'classification': - for classification in PURGE_FILTERED_RECORDS[field].keys(): - q = q.filter(~Record.classification.like(classification.replace("*", "%"))) - else: - for name in PURGE_FILTERED_RECORDS[field].keys(): - q = q.filter(getattr(Record, field) != name) - q = q.filter(Record.timestamp_server < unfiltered_age) - if q.all(): - count = db.session.query(Record).filter(Record.id.in_([x.id for x in q.all()])).delete(synchronize_session=False) - print("Deleted {} old records".format(count)) - db.session.commit() - except Exception as e: - app.logger.error("Record purging failed") - app.logger.error(e) - db.session.rollback() - - @staticmethod - def get_recordcnts_by_build(): - q = db.session.query(Record.build, db.func.count(Record.id)) - q = q.filter(Record.build.op('~')('^[0-9]+$')) - q = q.group_by(Record.build).order_by(cast(Record.build, db.Integer)).all() - return q - - @staticmethod - def get_builds(): - q = db.session.query(Record.build).distinct() - q = q.order_by(Record.build) - return sorted(q.all(), key=lambda x: LooseVersion(x[0]), reverse=True) - - @staticmethod - def get_recordcnts_by_classification(): - q = db.session.query(Record.classification, db.func.count(Record.id).label('total')) - q = q.group_by(Record.classification) - q = q.order_by(desc('total')) - return q.all() - - @staticmethod - def expand_class(D): - A, B, C = D - return ["{}/*".format(A), "{}/{}/*".format(A, B), "{}/{}/{}".format(A, B, C)] - - @staticmethod - def get_classifications(with_regex=False): - q = db.session.query(Record.classification).distinct() - if with_regex: - classes = [Record.expand_class(c[0].split('/')) for c in q.all()] - return sorted(set(itertools.chain(*classes))) - else: - return q.all() - - @staticmethod - def get_os_map(): - q = db.session.query(Record.system_name, Record.build).order_by(Record.system_name).group_by(Record.system_name, Record.build).all() - result = {} - for x in q: - result.setdefault(x[0], []).append(x[1]) - return result - - @staticmethod - def get_recordcnts_by_machine_type(): - q = db.session.query(Record.host_type, db.func.count(Record.id).label('total')) - q = q.group_by(Record.host_type) - q = q.order_by(desc('total')) - return q.all() - - @staticmethod - def get_recordcnts_by_severity(): - q = db.session.query(Record.severity, db.func.count(Record.id)).group_by(Record.severity).all() - return q - - @staticmethod - def get_crashcnts_by_class(classes=None): - q = db.session.query(Record.classification, db.func.count(Record.id)) - if classes: - q = q.filter(Record.classification.in_(classes)) - else: - q = q.filter(Record.classification.like('org.clearlinux/crash/%')) - q = q.group_by(Record.classification) - return q.all() - - @staticmethod - def get_crashcnts_by_build(classes=None): - q = db.session.query(Record.build, db.func.count(Record.id)) - if not classes: - classes = ['org.clearlinux/crash/clr'] - q = q.filter(Record.classification.in_(classes)) - q = q.filter(Record.build.op('~')('^[0-9]+$')) - q = q.group_by(Record.build) - q = q.order_by(desc(cast(Record.build, db.Integer))) - q = q.limit(10) - return q.all() - - @staticmethod - def get_top_crash_guilties(classes=None): - q = db.session.query(Guilty.function, Guilty.module, Record.build, db.func.count(Record.id).label('total'), Guilty.id, Guilty.comment) - q = q.join(Record) - if not classes: - classes = ['org.clearlinux/crash/clr'] - q = q.filter(Record.classification.in_(classes)) - q = q.filter(Record.build.op('~')('^[0-9][0-9]+$')) - q = q.filter(cast(Record.build, db.Integer) <= 100000) - q = q.filter(Guilty.hide == False) - q = q.group_by(Guilty.function, Guilty.module, Guilty.comment, Guilty.id, Record.build) - q = q.order_by(desc(cast(Record.build, db.Integer)), desc('total')) - # query for records created in the last week (~ 10 Clear builds) - q = q.filter(Record.build.in_(sorted(tuple(set([x[2] for x in q.all()])), key=lambda x: int(x))[-8:])) - interval_sec = 24 * 60 * 60 * 7 - current_time = time() - sec_in_past = current_time - interval_sec - q = q.filter(Record.timestamp_client > sec_in_past) - return q.all() - - @staticmethod - def get_new_crash_records(classes=None, id=None): - q = db.session.query(Record) - if not classes: - classes = ['org.clearlinux/crash/clr'] - q = q.filter(Record.classification.in_(classes)) - q = q.filter(Record.system_name == 'clear-linux-os') - q = q.filter(Record.processed == False) - if id: - q = q.filter(Record.id == id) - return q.all() - - @staticmethod - def set_processed_flag(record): - record.processed = True - - @staticmethod - def get_guilty_for_funcmod(func, mod): - q = db.session.query(Guilty).filter_by(function=func, module=mod).first() - return q - - @staticmethod - def get_guilty_id_for_record(record_id): - q = db.session.query(Guilty.id).join(Record) - q = q.filter(Record.id == record_id) - return q.first() - - @staticmethod - def init_guilty(func, mod): - return Guilty(func, mod) - - @staticmethod - def create_guilty_for_record(record, guilty): - record.guilty = guilty - - @staticmethod - def commit_guilty_changes(): - # just commit for now - db.session.commit() - - @staticmethod - def get_crash_backtraces(classes=None, guilty_id=None, machine_id=None, build=None, most_recent=None, record_id=None): - q = db.session.query(Record.payload, Record.id) - # Short circuit if we know the record ID - if record_id: - q = q.filter(Record.id == record_id) - return q.first() - if build: - q = q.filter(Record.build == build) - if not classes: - classes = ['org.clearlinux/crash/clr'] - q = q.filter(Record.classification.in_(classes)) - q = q.filter(Record.system_name == 'clear-linux-os') - if guilty_id: - q = q.filter(Record.guilty_id == guilty_id) - if machine_id: - q = q.filter(Record.machine_id == machine_id) - if most_recent: - interval_sec = 24 * 60 * 60 * int(most_recent) - current_time = time() - sec_in_past = current_time - interval_sec - q = q.filter(Record.timestamp_client > sec_in_past) - return q.all() - - @staticmethod - def reset_processed_records(classes=None, id=None): - q = db.session.query(Record) - if not classes: - classes = ['org.clearlinux/crash/clr'] - q = q.filter(Record.classification.in_(classes)) - q = q.filter(Record.system_name == 'clear-linux-os') - if id: - q = q.filter(Record.id == id) - records = q.all() - for r in records: - r.processed = False - db.session.commit() - - @staticmethod - def get_machine_ids_for_guilty(id, most_recent=None): - q = db.session.query(Record.build, Record.machine_id, db.func.count(Record.id).label('total'), Record.guilty_id) - q = q.filter(Record.guilty_id == id) - q = q.filter(Record.system_name == 'clear-linux-os') - q = q.filter(Record.build.op('~')('^[0-9][0-9]+$')) - q = q.group_by(Record.build, Record.machine_id, Record.guilty_id) - q = q.order_by(desc(cast(Record.build, db.Integer)), desc('total')) - if most_recent: - interval_sec = 24 * 60 * 60 * int(most_recent) - current_time = time() - sec_in_past = current_time - interval_sec - q = q.filter(Record.timestamp_client > sec_in_past) - return q.all() - - @staticmethod - def get_update_msgs(): - q = db.session.query(Record.payload) - q = q.filter(Record.classification == "org.clearlinux/swupd-client/update") - - sec_2_weeks = 24 * 60 * 60 * 7 - current_time = time() - time_2_weeks_ago = current_time - sec_2_weeks - - # query for records created in that last 2 weeks - q = q.filter(Record.timestamp_client > time_2_weeks_ago) - return q.all() - - @staticmethod - def get_swupd_msgs(most_recent=None): - q = db.session.query(Record.timestamp_client, Record.machine_id, Record.payload) - q = q.filter(Record.classification.like('org.clearlinux/swupd-client/%')) - - if most_recent: - interval_sec = 24 * 60 * 60 * int(most_recent) - current_time = time() - sec_in_past = current_time - interval_sec - q = q.filter(Record.timestamp_client > sec_in_past) - - q = q.order_by(desc(Record.timestamp_client)) - return q - - @staticmethod - def get_heartbeat_msgs(most_recent=None): - # These two expressions are SQL CASE conditional expressions, later - # used within count(distinct ...) aggregates for the query. - internal_expr = case([(Record.external == False, Record.machine_id), ]).label('internal_count') - external_expr = case([(Record.external == True, Record.machine_id), ]).label('external_count') - - q = db.session.query(Record.build, db.func.count(db.distinct(internal_expr)), db.func.count(db.distinct(external_expr))) - q = q.filter(Record.classification == "org.clearlinux/heartbeat/ping") - q = q.filter(Record.system_name == 'clear-linux-os') - q = q.group_by(Record.build) - - if most_recent: - interval_sec = 24 * 60 * 60 * int(most_recent) - current_time = time() - sec_in_past = current_time - interval_sec - q = q.filter(Record.timestamp_client > sec_in_past) - - q = q.order_by(cast(Record.build, db.Integer)) - return q.all() - - -class GuiltyBlacklist(db.Model): - __tablename__ = 'guilty_blacklisted' - id = db.Column(db.Integer, primary_key=True) - function = db.Column(db.String) - module = db.Column(db.String) - - def __init__(self, func, mod): - self.function = func - self.module = mod - - def __repr__(self): - return "".format(self.id, self.function, self.module) - - def __str__(self): - return str(self.to_dict()) - - def to_dict(self): - guilty = { - 'function': self.function, - 'module': self.module - } - return guilty - - @staticmethod - def add(func, mod): - try: - g = GuiltyBlacklist(func, mod) - db.session.add(g) - db.session.commit() - return g - except: - db.session.rollback() - raise - - @staticmethod - def remove(func, mod): - q = db.session.query(GuiltyBlacklist) - q = q.filter_by(function=func, module=mod) - entry = q.first() - db.session.delete(entry) - db.session.commit() - - @staticmethod - def get_guilties(): - q = db.session.query(GuiltyBlacklist.function, GuiltyBlacklist.module) - q = q.order_by(GuiltyBlacklist.function) - return q.all() - - @staticmethod - def exists(func, mod): - q = db.session.query(GuiltyBlacklist.function, GuiltyBlacklist.module) - q = q.filter_by(function=func, module=mod) - return len(q.all()) != 0 and True or False - - @staticmethod - def update(to_add, to_remove): - try: - for i in to_add: - if not GuiltyBlacklist.exists(i[0], i[1]): - g = GuiltyBlacklist(i[0], i[1]) - db.session.add(g) - for i in to_remove: - if GuiltyBlacklist.exists(i[0], i[1]): - q = db.session.query(GuiltyBlacklist) - q = q.filter_by(function=i[0], module=i[1]) - entry = q.first() - db.session.delete(entry) - db.session.commit() - except: - db.session.rollback() - raise - -# vi: ts=4 et sw=4 sts=4 diff --git a/telemetryui/telemetryui/model.py b/telemetryui/telemetryui/model.py new file mode 120000 index 0000000..ac6394a --- /dev/null +++ b/telemetryui/telemetryui/model.py @@ -0,0 +1 @@ +../../shared/model.py \ No newline at end of file