WIP tests

Author: Hyask

src/daisy/submit.py

@@ -28,11 +28,9 @@
 from cassandra import WriteTimeout
 from cassandra.cqlengine.query import DoesNotExist
 
-from daisy import utils
 from daisy.metrics import get_metrics
-from errortracker import cassandra_schema
+from errortracker import cassandra_schema, oopses, utils
 from errortracker.cassandra import cassandra_session
-from oopsrepository import oopses
 
 metrics = get_metrics("daisy.%s" % socket.gethostname())
 logger = logging.getLogger("gunicorn.error")

src/daisy/submit_core.py

@@ -19,7 +19,6 @@
 import logging
 import os
 import random
-import shutil
 import socket
 from datetime import datetime
 from pathlib import Path
@@ -29,9 +28,7 @@
 
 # from daisy import config
 from daisy.metrics import get_metrics
-from errortracker import cassandra_schema, config
-
-from . import utils
+from errortracker import amqp_utils, cassandra_schema, config
 
 metrics = get_metrics("daisy.%s" % socket.gethostname())
 logger = logging.getLogger("gunicorn.error")
@@ -247,8 +244,8 @@ def write_to_amqp(message, arch):
             connection = amqp.Connection(host=config.amqp_host)
             connection.connect()
             channel = connection.channel()
-    except utils.amqplib_error_types as e:
-        if utils.is_amqplib_connection_error(e):
+    except amqp_utils.amqplib_error_types as e:
+        if amqp_utils.is_amqplib_connection_error(e):
             # Could not connect
             return False
         # Unknown error mode : don't hide it.
@@ -267,8 +264,8 @@ def write_to_amqp(message, arch):
         msg = "%s added to %s queue" % (message.split(":")[0], queue)
         logger.info(msg)
         queued = True
-    except utils.amqplib_error_types as e:
-        if utils.is_amqplib_connection_error(e):
+    except amqp_utils.amqplib_error_types as e:
+        if amqp_utils.is_amqplib_connection_error(e):
             # Could not connect / interrupted connection
             queued = False
         # Unknown error mode : don't hide it.

src/errortracker/amqp_utils.py

@@ -0,0 +1,31 @@
+import socket
+
+from amqp import ConnectionError as AMQPConnectionException
+
+# From oops-amqp
+# These exception types always indicate an AMQP connection error/closure.
+# However you should catch amqplib_error_types and post-filter with
+# is_amqplib_connection_error.
+amqplib_connection_errors = (socket.error, AMQPConnectionException)
+# A tuple to reduce duplication in different code paths. Lists the types of
+# exceptions legitimately raised by amqplib when the AMQP server goes down.
+# Not all exceptions *will* be such errors - use is_amqplib_connection_error to
+# do a second-stage filter after catching the exception.
+amqplib_error_types = amqplib_connection_errors + (IOError,)
+
+
+# From oops-amqp
+def is_amqplib_ioerror(e):
+    """Returns True if e is an amqplib internal exception."""
+    # Raised by amqplib rather than socket.error on ssl issues and short reads.
+    if type(e) is not IOError:
+        return False
+    if e.args == ("Socket error",) or e.args == ("Socket closed",):
+        return True
+    return False
+
+
+# From oops-amqp
+def is_amqplib_connection_error(e):
+    """Return True if e was (probably) raised due to a connection issue."""
+    return isinstance(e, amqplib_connection_errors) or is_amqplib_ioerror(e)

src/errortracker/cassandra.py

@@ -11,22 +11,22 @@
 _session = None
 
 
-def setup_cassandra():
+def setup_cassandra(keyspace="crashdb", replication_factor=3):
     # cassandra wants this environment variable to be set, otherwise issues a warning. Let's please it.
     os.environ["CQLENG_ALLOW_SCHEMA_MANAGEMENT"] = "1"
     auth_provider = PlainTextAuthProvider(
         username=config.cassandra_username, password=config.cassandra_password
     )
     connection.setup(
         config.cassandra_hosts,
-        "crashdb",
+        keyspace,
         auth_provider=auth_provider,
         load_balancing_policy=RoundRobinPolicy(),
         protocol_version=4,
     )
-    management.create_keyspace_simple("crashdb", 3)
+    management.create_keyspace_simple(name=keyspace, replication_factor=replication_factor)
     sync_schema()
-    # connection.execute("USE crashdb;")
+    # connection.execute(f"USE {keyspace};")
 
 
 def sync_schema():

src/errortracker/cassandra_schema.py

@@ -4,7 +4,7 @@
 
 
 class ErrorTrackerTable(models.Model):
-    __keyspace__ = "crashdb"
+    # __table_name_case_sensitive__ is deprecated already, but let's keep it in case we run on older machines.
     __table_name_case_sensitive__ = True
     __abstract__ = True
 

src/oopsrepository/oopses.py src/errortracker/oopses.pysrc/oopsrepository/oopses.py renamed to src/errortracker/oopses.py

@@ -13,10 +13,8 @@
 import uuid
 from hashlib import md5, sha1
 
-from cassandra import ConsistencyLevel
-from cassandra.auth import PlainTextAuthProvider
-from cassandra.cluster import Cluster
 from cassandra.query import SimpleStatement
+from cassandra.cqlengine.query import BatchQuery
 
 from errortracker import cassandra_schema
 from errortracker.cassandra import cassandra_session
@@ -27,7 +25,33 @@
 _cassandra_session = None
 
 
-def insert(config, oopsid, oops_json, user_token=None, fields=None, proposed_pkg=False):
+def prune():
+    """Remove OOPSES that are over 30 days old."""
+    # Find days to prune
+    days = set()
+    prune_to = time.strftime("%Y%m%d", time.gmtime(time.time() - MONTH))
+    for dayoops in cassandra_schema.DayOOPS.objects.distinct(["key"]):
+        key = dayoops.key.decode()
+        if key < prune_to:
+            days.add(key)
+    if not days:
+        return
+    # collect all the oopses (buffers all in memory; may want to make
+    # incremental in future)
+    for day in days:
+        oops_ids = list(
+            set(
+                dayoops.value
+                for dayoops in cassandra_schema.DayOOPS.filter(key=day.encode()).only(["value"])
+            )
+        )
+        with BatchQuery() as b:
+            cassandra_schema.DayOOPS.objects.batch(b).filter(key=day.encode()).delete()
+            for id in oops_ids:
+                cassandra_schema.OOPS.objects.batch(b).filter(key=id).delete()
+
+
+def insert(oopsid, oops_json, user_token=None, fields=None, proposed_pkg=False) -> str:
     """Insert an OOPS into the system.
 
     :return: The day which the oops was filed under.
@@ -39,8 +63,8 @@ def insert(config, oopsid, oops_json, user_token=None, fields=None, proposed_pkg
     assert isinstance(oops_dict, dict)
     insert_dict = {}
     for key, value in list(oops_dict.items()):
-        insert_dict[key] = json.dumps(value)
-    return _insert(config, oopsid, insert_dict, user_token, fields, proposed_pkg)
+        insert_dict[key] = str(value)
+    return _insert(oopsid, insert_dict, user_token, fields, proposed_pkg)
 
 
 def insert_dict(
@@ -50,7 +74,7 @@ def insert_dict(
     fields=None,
     proposed_pkg=False,
     ttl=None,
-):
+) -> str:
     """Insert an OOPS into the system.
 
     :return: The day which the oops was filed under.
@@ -66,7 +90,7 @@ def _insert(
     fields=None,
     proposed_pkg=False,
     ttl=None,
-):
+) -> str:
     """Internal function. Do not call this directly.
 
     :param oopsid: The identifier for this OOPS.
@@ -91,7 +115,7 @@ def _insert(
         cassandra_schema.OOPS.create(key=oopsid.encode(), column1=key, value=value).ttl(ttl)
 
     automated_testing = False
-    if user_token.startswith("deadbeef"):
+    if user_token and user_token.startswith("deadbeef"):
         automated_testing = True
 
     cassandra_schema.DayOOPS.create(key=day_key.encode(), column1=now_uuid, value=oopsid.encode())

src/daisy/utils.py src/errortracker/utils.pysrc/daisy/utils.py renamed to src/errortracker/utils.py

@@ -1,23 +1,10 @@
 import logging
 import re
-import socket
 import uuid
 
 import apt
-from amqp import ConnectionError as AMQPConnectionException
 
-from oopsrepository import oopses
-
-# From oops-amqp
-# These exception types always indicate an AMQP connection error/closure.
-# However you should catch amqplib_error_types and post-filter with
-# is_amqplib_connection_error.
-amqplib_connection_errors = (socket.error, AMQPConnectionException)
-# A tuple to reduce duplication in different code paths. Lists the types of
-# exceptions legitimately raised by amqplib when the AMQP server goes down.
-# Not all exceptions *will* be such errors - use is_amqplib_connection_error to
-# do a second-stage filter after catching the exception.
-amqplib_error_types = amqplib_connection_errors + (IOError,)
+from errortracker import oopses
 
 EOL_RELEASES = {
     "Ubuntu 10.04": "lucid",
@@ -193,18 +180,6 @@ def bucket(oops_config, oops_id, crash_signature, report_dict):
             oopses.update_errors_by_release(oops_config, oops_uuid, system_uuid, release)
 
 
-def attach_error_report(report, context):
-    # We only attach error report that was submitted by the client if we've hit
-    # a MaximumRetryException from Cassandra.
-    if "type" in report and report["type"] == "MaximumRetryException":
-        env = context["wsgi_environ"]
-        if "wsgi.input.decoded" in env:
-            data = env["wsgi.input.decoded"]
-            if "req_vars" not in report:
-                report["req_vars"] = {}
-            report["req_vars"]["wsgi.input.decoded"] = data
-
-
 def retraceable_release(release):
     if release in EOL_RELEASES:
         logging.info("%s is EoL, not retraceable", release)
@@ -237,31 +212,7 @@ def blocklisted_device(system_token):
     Used for devices that have repeatedly failed to submit a crash.
     """
 
-    blocklist = [
-        # 20150814 - OOPS count was at 43
-        "2f175cea621bda810f267f1da46409a111f58011435f410aa198362e9372da78b6fafe6827ff26e025a5ab7d2859346de6b188f0622118c15a119c58ca538acb",
-        # 20150826 - OOPS count was at 18
-        "81b75a0bdd531a5c02a4455b05674ea45fbb65324bcae5fe51659bce850aa40bcd1388e3eed4d46ce9abb4e56d1dd7dde45ded473995feb0ac2c01518a841efe",
-        # 20150903 - OOPS count was at 27
-        "b5329547bdab8adea4245399ff9656ca608e825425fbb0ad2c68e182b75ce80c13f9186e4e9b8e7a17dd15dd196b12a65e1b7f513184296320dad50c587754f5",
-    ]
+    blocklist = []
     if system_token in blocklist:
         return True
     return False
-
-
-# From oops-amqp
-def is_amqplib_ioerror(e):
-    """Returns True if e is an amqplib internal exception."""
-    # Raised by amqplib rather than socket.error on ssl issues and short reads.
-    if type(e) is not IOError:
-        return False
-    if e.args == ("Socket error",) or e.args == ("Socket closed",):
-        return True
-    return False
-
-
-# From oops-amqp
-def is_amqplib_connection_error(e):
-    """Return True if e was (probably) raised due to a connection issue."""
-    return isinstance(e, amqplib_connection_errors) or is_amqplib_ioerror(e)