From 21237b333ca2e6766a74dc28c5f788160349a9ce Mon Sep 17 00:00:00 2001
From: Gordon Chung <chungg@ca.ibm.com>
Date: Thu, 20 Mar 2014 22:46:37 -0400
Subject: [PATCH] fix create_or_update logic to avoid rollbacks

check if entry exists rather than blindly adding entries to table.
this will avoid unnecessary rollback overhead.

Closes-Bug: #1295504
Change-Id: I923cc35ed4c9155ffa04ac189555cdbcb38bcbcf
---
 ceilometer/storage/impl_sqlalchemy.py | 55 +++++++++------------------
 1 file changed, 19 insertions(+), 36 deletions(-)

diff --git a/ceilometer/storage/impl_sqlalchemy.py b/ceilometer/storage/impl_sqlalchemy.py
index 075545a441..84e05b6a1c 100644
--- a/ceilometer/storage/impl_sqlalchemy.py
+++ b/ceilometer/storage/impl_sqlalchemy.py
@@ -272,56 +272,39 @@ class Connection(base.Connection):
     def _create_or_update(session, model_class, _id, source=None, **kwargs):
         if not _id:
             return None
-
         try:
-            # create a nested session for the case of two call of
-            # record_metering_data run in parallel to not fail the
-            # record of this sample
-            # (except for sqlite, that doesn't support nested
-            # transaction and doesn't have concurrency problem)
-            nested = session.connection().dialect.name != 'sqlite'
+            with session.begin(subtransactions=True):
+                obj = session.query(model_class).get(str(_id))
+                if obj is None:
+                    obj = model_class(id=str(_id))
+                    session.add(obj)
 
-            # raise dbexc.DBDuplicateEntry manually for sqlite
-            # to not break the current session
-            if not nested and session.query(model_class).get(str(_id)):
-                raise dbexc.DBDuplicateEntry()
-
-            with session.begin(nested=nested,
-                               subtransactions=not nested):
-                obj = model_class(id=str(_id))
-                session.add(obj)
+                if source and not filter(lambda x: x.id == source.id,
+                                         obj.sources):
+                    obj.sources.append(source)
+                for k in kwargs:
+                    setattr(obj, k, kwargs[k])
         except dbexc.DBDuplicateEntry:
             # requery the object from the db if this is an other
             # parallel/previous call of record_metering_data that
             # have successfully created this object
-            obj = session.query(model_class).get(str(_id))
-
-        # update the object
-        if source and not filter(lambda x: x.id == source.id, obj.sources):
-            obj.sources.append(source)
-        for k in kwargs:
-            setattr(obj, k, kwargs[k])
+            obj = Connection._create_or_update(session, model_class,
+                                               _id, source, **kwargs)
         return obj
 
     @staticmethod
     def _create_meter(session, name, type, unit):
         try:
-            nested = session.connection().dialect.name != 'sqlite'
-            if not nested and session.query(models.Meter)\
+            with session.begin(subtransactions=True):
+                obj = session.query(models.Meter)\
                     .filter(models.Meter.name == name)\
                     .filter(models.Meter.type == type)\
-                    .filter(models.Meter.unit == unit).count() > 0:
-                raise dbexc.DBDuplicateEntry()
-
-            with session.begin(nested=nested,
-                               subtransactions=not nested):
-                obj = models.Meter(name=name, type=type, unit=unit)
-                session.add(obj)
+                    .filter(models.Meter.unit == unit).first()
+                if obj is None:
+                    obj = models.Meter(name=name, type=type, unit=unit)
+                    session.add(obj)
         except dbexc.DBDuplicateEntry:
-            obj = session.query(models.Meter)\
-                .filter(models.Meter.name == name)\
-                .filter(models.Meter.type == type)\
-                .filter(models.Meter.unit == unit).first()
+            obj = Connection._create_meter(session, name, type, unit)
 
         return obj