Merge lp:~brian-murray/daisy/no-pycassa into lp:daisy

[Brian Murray]

[Steve Langasek (vorlon)]

review: Needs Fixing

=== modified file 'daisy/submit.py'
--- daisy/submit.py	2019-08-12 16:09:19 +0000
+++ daisy/submit.py	2019-10-01 22:54:14 +0000
@@ -27,12 +27,19 @@
 
 from oopsrepository import config as oopsconfig
 from oopsrepository import oopses
+<<<<<<< TREE
 import pycassa
 from pycassa.cassandra.ttypes import NotFoundException, InvalidRequestException
 from pycassa.pool import MaximumRetryException
+=======
+
+from apport import Report
+from binascii import hexlify, unhexlify
+from cassandra import WriteTimeout
+from cassandra.query import SimpleStatement
+>>>>>>> MERGE-SOURCE
 
 from daisy import config
-import apport
 from daisy import utils
 from daisy.metrics import get_metrics
 
@@ -47,6 +54,7 @@
 metrics = get_metrics('daisy.%s' % socket.gethostname())
 logger = logging.getLogger('gunicorn.error')
 
+<<<<<<< TREE
 
 def update_release_pkg_counter(column_family, release, src_package, date):
     column_family.insert('%s:%s' % (release, src_package), {date: 1})
@@ -56,6 +64,32 @@
 
 def create_report_from_bson(data):
     report = apport.Report()
+=======
+counters_update = None
+proposed_counters_update = None
+
+def update_counters(_session, release, src_package, date, src_version=None):
+    cql_release = release.replace("'", "''")
+    if src_version:
+        key = '%s:%s:%s' % (cql_release, src_package, src_version)
+    else:
+        key = '%s:%s' % (cql_release, src_package)
+    _session.execute(counters_update, [key, date])
+
+
+def update_proposed_counters(_session, release, src_package, date,
+                             src_version=None):
+    cql_release = release.replace("'", "''")
+    if src_version:
+        key = '%s:%s:%s' % (cql_release, src_package, src_version)
+    else:
+        key = '%s:%s' % (cql_release, src_package)
+    _session.execute(proposed_counters_update, [key, date])
+
+
+def create_minimal_report_from_bson(data):
+    report = Report()
+>>>>>>> MERGE-SOURCE
     for key in data:
         try:
             report[key.encode('UTF-8')] = data[key].encode('UTF-8')
@@ -159,7 +193,15 @@
     proc_status = data.get('ProcStatus', '')
     if date and exec_path and proc_status and system_token:
         try:
+<<<<<<< TREE
             reported_crash_ids = systemoopshashes_cf.get(system_token)
+=======
+            cql_system_token = '0x' + hexlify(system_token)
+            results = _session.execute(
+                SimpleStatement("SELECT column1 FROM \"%s\" WHERE key = %s LIMIT 1"
+                                % ('SystemOOPSHashes', cql_system_token)))
+            reported_crash_ids = (row[0] for row in results)
+>>>>>>> MERGE-SOURCE
             crash_id = '%s:%s:%s' % (date, exec_path, proc_status)
             if type(crash_id) == unicode:
                 crash_id = crash_id.encode('utf-8')
@@ -300,6 +342,17 @@
        If a core file is to be requested, return (True, 'UUID CORE')
        If no further action is needed, return (True, 'UUID OOPSID')
     '''
+<<<<<<< TREE
+=======
+    indexes_select = None
+    if not indexes_select:
+        indexes_select = \
+            _session.prepare('SELECT value FROM "Indexes" WHERE key = ? and column1 = ? LIMIT 1')
+    stacktrace_select = None
+    if not stacktrace_select:
+        stacktrace_select = \
+            _session.prepare('SELECT value FROM "Stacktrace" WHERE key = ? and column1 = ? LIMIT 1')
+>>>>>>> MERGE-SOURCE
 
     indexes_fam = pycassa.ColumnFamily(_pool, 'Indexes')
     oops_cf = pycassa.ColumnFamily(_pool, 'OOPS')
@@ -369,10 +422,20 @@
         if crash_sig and not retry:
             # The crash is a duplicate so we don't need this data.
             # Stacktrace, and ThreadStacktrace were already not accepted
+<<<<<<< TREE
             if 'ProcMaps' in report:
                 unneeded_columns = ['Disassembly', 'ProcMaps', 'ProcStatus',
                                     'Registers', 'StacktraceTop']
                 oops_cf.remove(oops_id, columns=unneeded_columns)
+=======
+            if 'ProcMaps' in data:
+                oops_delete = _session.prepare('DELETE FROM "OOPS" WHERE key = ? AND column1 = ?')
+                unneeded_columns = ('Disassembly', 'ProcMaps', 'ProcStatus',
+                                    'Registers', 'StacktraceTop')
+                for unneeded_column in unneeded_columns:
+                    _session.execute(oops_delete,
+                                     [oops_id, unneeded_column])
+>>>>>>> MERGE-SOURCE
             # We have already retraced for this address signature, so this
             # crash can be immediately bucketed.
             utils.bucket(oops_config, oops_id, crash_sig, data)

=== modified file 'daisy/submit_core.py'
--- daisy/submit_core.py	2019-08-12 16:09:19 +0000
+++ daisy/submit_core.py	2019-10-01 22:54:14 +0000
@@ -54,53 +54,31 @@
             return False
     return True
 
-def swift_delete_ignoring_error(conn, bucket, oops_id):
-    global _cached_swift
-    import swiftclient
+def swift_delete_ignoring_error(swift_cmd, bucket, oops_id):
+    from subprocess import check_call, CalledProcessError
+    swift_delete_cmd = swift_cmd + ['delete', bucket, oops_id]
     try:
-        conn.delete_object(bucket, oops_id)
-    except swiftclient.ClientException:
-        metrics.meter('swift_delete_error')
+        check_call(swift_delete_cmd)
+    except CalledProcessError:
+        pass
 
 def write_to_swift(environ, fileobj, oops_id, provider_data):
     '''Write the core file to OpenStack Swift.'''
-    global _cached_swift
-    import swiftclient
-    opts = {'tenant_name': provider_data['os_tenant_name'],
-            'region_name': provider_data['os_region_name']}
-    if not _cached_swift:
-        _cached_swift = swiftclient.client.Connection(
-                                     provider_data['os_auth_url'],
-                                     provider_data['os_username'],
-                                     provider_data['os_password'],
-                                     os_options=opts,
-                                     auth_version='2.0')
-    # if there is no token we should recreate the connection
-    if not _cached_swift.token:
-        _cached_swift = swiftclient.client.Connection(
-                                     provider_data['os_auth_url'],
-                                     provider_data['os_username'],
-                                     provider_data['os_password'],
-                                     os_options=opts,
-                                     auth_version='2.0')
-    # it seems to still be None sometimes
-    msg = 'swift_token: %s' % (_cached_swift.token)
-    logger.info(msg)
+    from subprocess import check_call, CalledProcessError
+
+    swift_cmd = ['/usr/bin/swift',
+                 '--os-auth-url', '%s' % provider_data['os_auth_url'],
+                 '--os-username', '%s' % provider_data['os_username'],
+                 '--os-password', '%s' % provider_data['os_password'],
+                 '--os-tenant-name', '%s' % provider_data['os_tenant_name'],
+                 '--os-region-name', '%s' % provider_data['os_region_name'],
+                 '--auth-version', '2.0']
     bucket = provider_data['bucket']
-    _cached_swift.http_conn = None
-    if (provider_data.get('usage_max_mb')):
-        headers = _cached_swift.head_account()
-        bytes_used = int(headers.get('x-account-bytes-used', 0))
-        # Keep a reference to the number of bytes used by swift in a possible
-        # future OOPS report. We may find that a series of OOPSes are actually
-        # related to heavy load on Swift, as has been the case before.
-        environ['swift.bytes_used'] = bytes_used
-        if (not write_policy_allow(oops_id, bytes_used, provider_data)):
-            msg = '%s bytes_used issue' % (oops_id)
-            logger.info(msg)
-            return False
-
-    _cached_swift.put_container(bucket)
+    swift_post_cmd = swift_cmd + ['post', bucket]
+    try:
+        check_call(swift_post_cmd)
+    except CalledProcessError:
+        return False
     try:
         coredir = '/tmp/cores-%s' % os.getpid()
         if not os.path.exists(coredir):
@@ -117,25 +95,21 @@
             t_size = os.path.getsize(t.name)
             msg = '%s has a %i byte core file' % (oops_id, t_size)
             logger.info(msg)
-            # Don't set a content_length (that we don't have) to force a chunked
-            # transfer.
-            _cached_swift.put_object(bucket, oops_id, t, content_length=t_size)
-    except IOError, e:
-        swift_delete_ignoring_error(_cached_swift, bucket, oops_id)
-        if e.message == 'request data read error':
-            msg = 'IOError when trying to add (%s) to bucket: %s' % (oops_id, str(e))
-            logger.info(msg)
-            return False
-        else:
-            msg = 'IOError when trying to add (%s) to bucket: %s' % (oops_id, str(e))
-            logger.info(msg)
-            metrics.meter('swift_ioerror')
-            raise
-    except swiftclient.ClientException as e:
-        msg = 'ClientException when trying to add (%s) to bucket: %s' % (oops_id, str(e))
-        logger.info(msg)
-        metrics.meter('swift_client_exception')
-        swift_delete_ignoring_error(_cached_swift, bucket, oops_id)
+            swift_upload_cmd = swift_cmd + ['upload', '--object-name',
+                                            oops_id, bucket,
+                                            os.path.join(coredir, t.name)]
+            check_call(swift_upload_cmd)
+    except CalledProcessError as e:
+        swift_delete_ignoring_error(swift_cmd, bucket, oops_id)
+        msg = 'CalledProcessError when trying to add (%s) to bucket: %s' % \
+               (oops_id, str(e.returncode))
+        logger.info(msg)
+        return False
+    except IOError as e:
+        swift_delete_ignoring_error(swift_cmd, bucket, oops_id)
+        msg = 'IOError when trying to add (%s) to bucket: %s' % \
+               (oops_id, str(e))
+        logger.info(msg)
         return False
     msg = 'CORE for (%s) written to bucket' % (oops_id)
     logger.info(msg)