Repository: dmwm/wmcore
Issue: MCFakeFile IntegrityError in LQ

Bug found in cmssrv217
https://cmsweb.cern.ch/couchdb/_utils/document.html?wmstats_logdb/d890febe83d602b5db7abce4e5264659

the agent even created jobs (that probably means WMBS inserts worked), but the workflow was stuck in acquired since it was failing to be injected into dbsbuffer_file.

I think it's a side effect of the workflow cleanup/completion separation, since the agent cleans these things in a later stage (when archiving(?)).

In case logDB entry gets cleaned up, this is the error message:

```
2015-07-29 00:01:18,792:ERROR:WorkQueue:jbadillo_ACDC_SUS-RunIIWinter15pLHE-00011_00010_v0_STEP0ATCERN_150708_150727_7499, /acdc/jbadillo_ACDC_SUS-RunIIWinter15pLHE-00011_00010_v0_STEP0ATCERN_150708_150727_7499/:pdmvserv_SUS-RunIIWinter15pLHE-00011_00010_v0_STEP0ATCERN_150704_201230_7297:Production/0/26 creating subscription failed in LQ: (IntegrityError) (1062, "Duplicate entry 'MCFakeFile-pdmvserv_SUS-RunIIWinter15pLHE-00011_00010_v0_STEP0AT' for key 'lfn'") 'insert into dbsbuffer_file(lfn, filesize, events, dataset_algo, status, workflow)\\n                values (%s, %s, %s, %s, %s, %s)' [('MCFakeFile-pdmvserv_SUS-RunIIWinter15pLHE-00011_00010_v0_STEP0ATCERN_150704_201230_7297-Production-b9ea0183968f67ba39db2a964f0565e6', 0, 500000, 1L, 'GLOBAL', 19221L), ('MCFakeFile-pdmvserv_SUS-RunIIWinter15pLHE-00011_00010_v0_STEP0ATCERN_150704_201230_7297-Production-b9ea0183968f67ba39db2a964f0565e6', 0, 200000, 1L, 'GLOBAL', 19221L)]
```


Changed files:
test/python/WMCore_t/ACDC_t/DataCollectionService_t.py, src/python/WMCore/ACDC/DataCollectionService.py

diff --git a/src/python/WMCore/ACDC/DataCollectionService.py b/src/python/WMCore/ACDC/DataCollectionService.py
index cf551bf43..52d00b786 100644
--- a/src/python/WMCore/ACDC/DataCollectionService.py
+++ b/src/python/WMCore/ACDC/DataCollectionService.py
@@ -20,6 +20,31 @@ from WMCore.WMException      import WMException
 from WMCore.DataStructs.File import File
 from WMCore.DataStructs.Run  import Run
 
+
+def mergeFakeFiles(chunkFiles):
+    """
+    _mergeFakeFiles_
+
+    Receive a list of dicts with acdc files info and merge them together
+    (sum up events) belonging to the same MCFakeFile.
+    """
+    # do not do anything for real files
+    if not chunkFiles[0]['lfn'].startswith('MCFakeFile'):
+        return chunkFiles
+
+    logging.info("Merging %d ACDC FakeFiles...", len(chunkFiles))
+    mergedFiles = {}
+    for acdcFile in chunkFiles:
+        if acdcFile['lfn'] not in mergedFiles:
+            mergedFiles[acdcFile['lfn']] = acdcFile
+        else:
+            mergedFiles[acdcFile['lfn']]['events'] += acdcFile['events']
+            mergedFiles[acdcFile['lfn']]['runs'][0]['lumis'].extend(acdcFile['runs'][0]['lumis'])
+    logging.info("resulted in %d final ACDC FakeFiles.", len(mergedFiles))
+
+    return mergedFiles.values()
+
+
 class ACDCDCSException(WMException):
     """
     Yet another dummy variable class
@@ -236,7 +261,7 @@ class DataCollectionService(CouchService):
         files = self._getFilesetInfo(collectionName, filesetName, 
                                        user, group, chunkOffset, chunkSize)
 
-
+        files = mergeFakeFiles(files)
         for fileInfo in files:
             newFile = File(lfn = fileInfo["lfn"], size = fileInfo["size"],
                            events = fileInfo["events"], parents = set(fileInfo["parents"]),
@@ -250,6 +275,7 @@ class DataCollectionService(CouchService):
 
         return chunkFiles
 
+
     @CouchUtils.connectToCouch
     def getProductionACDCInfo(self, collectionID, taskName, user = "cmsdataops",
                         group = "cmsdataops"):
diff --git a/test/python/WMCore_t/ACDC_t/DataCollectionService_t.py b/test/python/WMCore_t/ACDC_t/DataCollectionService_t.py
index 47b24e03f..f2372c6d0 100644
--- a/test/python/WMCore_t/ACDC_t/DataCollectionService_t.py
+++ b/test/python/WMCore_t/ACDC_t/DataCollectionService_t.py
@@ -9,9 +9,8 @@ Copyright (c) 2010 Fermilab. All rights reserved.
 
 import unittest
 
-
 from WMQuality.TestInitCouchApp import TestInitCouchApp
-from WMCore.ACDC.DataCollectionService import DataCollectionService
+from WMCore.ACDC.DataCollectionService import DataCollectionService, mergeFakeFiles
 from WMCore.WMBS.Job import Job
 from WMCore.DataStructs.File import File
 from WMCore.DataStructs.Run import Run
@@ -295,5 +294,86 @@ class DataCollectionService_t(unittest.TestCase):
                          "Error: Whitelist for run 3 is wrong.")
         return
 
+    def notestMergeFakeFiles(self):
+        """
+        _testMergeFakeFiles_
+
+        Verify that we can merge MCFakeFiles together when a fileset contains
+        several failures for the same input fake file.
+        """
+        originalFiles = [{'checksums': {},
+                          'events': 500000,
+                          'first_event': 1,
+                          'id': 40,
+                          'last_event': 0,
+                          'lfn': 'MCFakeFile-File1',
+                          'locations': ['T1_DE_KIT_Disk'],
+                          'merged': '0',
+                          'parents': [],
+                          'runs': [{'lumis': [1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11], 'run_number': 1}],
+                          'size': 0},
+                         {'checksums': {},
+                          'events': 500000,
+                          'first_event': 1000001,
+                          'id': 40,
+                          'last_event': 0,
+                          'lfn': 'MCFakeFile-File2',
+                          'locations': ['T1_DE_KIT_Disk'],
+                          'merged': '0',
+                          'parents': [],
+                          'runs': [{'lumis': [21, 22, 23, 24, 25, 26, 27, 28, 29, 30, 31],
+                                    'run_number': 1}],
+                          'size': 0},
+                         {'checksums': {},
+                          'events': 500000,
+                          'first_event': 2000001,
+                          'id': 40,
+                          'last_event': 0,
+                          'lfn': 'MCFakeFile-File1',
+                          'locations': ['T1_DE_KIT_Disk'],
+                          'merged': '0',
+                          'parents': [],
+                          'runs': [{'lumis': [41, 42, 43, 44, 45, 46, 47, 48, 49, 50, 51],
+                                    'run_number': 1}],
+                          'size': 0},
+                         {'checksums': {},
+                          'events': 500000,
+                          'first_event': 7000001,
+                          'id': 40,
+                          'last_event': 0,
+                          'lfn': 'MCFakeFile-File3',
+                          'locations': ['T1_DE_KIT_Disk'],
+                          'merged': '0',
+                          'parents': [],
+                          'runs': [{'lumis': [81, 82, 83, 84, 85, 86, 87, 88, 89, 90, 91],
+                                    'run_number': 1}],
+                          'size': 0},
+                         {'checksums': {},
+                          'events': 500000,
+                          'first_event': 4000001,
+                          'id': 40,
+                          'last_event': 0,
+                          'lfn': 'MCFakeFile-File3',
+                          'locations': ['T1_DE_KIT_Disk'],
+                          'merged': '0',
+                          'parents': [],
+                          'runs': [{'lumis': [51, 52, 53, 54, 55, 56, 57, 58, 59, 60, 61],
+                                    'run_number': 1}],
+                          'size': 0}]
+
+        mergedFiles = mergeFakeFiles(originalFiles)
+        self.assertEqual(len(mergedFiles), 3, "Error: wrong number of files.")
+
+        totalEvents = 0
+        for job in mergedFiles:
+            totalEvents += job['events']
+        self.assertEqual(totalEvents, 2500000, "Error: wrong number of total events.")
+
+        for job in mergedFiles:
+            if job['lfn'] == 'MCFakeFile-File1':
+                lumiList= job['runs'][0]['lumis']
+        self.assertEqual(len(lumiList), 22)
+
+
 if __name__ == '__main__':
     unittest.main()