Repository: rucio/rucio
Issue: Automatix / transmogrifier race condition

Motivation
----------
Transmogrifier is often / sometimes failing to set rules for subscriptions to automatix data. 

It looks like a race condition:  the new version of automatix uses the upload client to create datasets and *then* sets the metadata for the new DIDs.

If transmogrifier runs and processes the new DIDs *before* automatix has finished uploading, it does not find any matching subscriptions as there is not yet metadata associated with the DIDs.  Those DIDs are then no longer new to transmogrifier and will be ignored in future cycles.

This will presumably affect anything created with the UploadClient whose metadata is attached afterwards.

Modification
------------
I think this can be mitigated by simply making sure the transmogrifier sleep cycle is >> the automatix / upload client work cycle.

Ultimately, I guess the upload client needs to know about metadata so that metadata can be attached at DID creation time.





Changed files:
lib/rucio/tests/test_automatix.py, lib/rucio/daemons/automatix/automatix.py, lib/rucio/client/uploadclient.py

diff --git a/lib/rucio/client/uploadclient.py b/lib/rucio/client/uploadclient.py
index c1ff2567c..ff6279347 100644
--- a/lib/rucio/client/uploadclient.py
+++ b/lib/rucio/client/uploadclient.py
@@ -77,6 +77,7 @@ class UploadClient:
             did_name              - Optional: custom did name (Default: name of the file)
             dataset_scope         - Optional: custom dataset scope
             dataset_name          - Optional: custom dataset name
+            dataset_meta          - Optional: custom metadata for dataset
             impl                  - Optional: name of the protocol implementation to be used to upload this item.
             force_scheme          - Optional: force a specific scheme (if PFN upload this will be overwritten) (Default: None)
             pfn                   - Optional: use a given PFN (this sets no_register to True, and no_register becomes mandatory)
@@ -371,6 +372,7 @@ class UploadClient:
                 logger(logging.DEBUG, 'Trying to create dataset: %s' % dataset_did_str)
                 self.client.add_dataset(scope=file['dataset_scope'],
                                         name=file['dataset_name'],
+                                        meta=file.get('dataset_meta'),
                                         rules=[{'account': self.client.account,
                                                 'copies': 1,
                                                 'rse_expression': rse,
diff --git a/lib/rucio/daemons/automatix/automatix.py b/lib/rucio/daemons/automatix/automatix.py
index 835899d33..c27a20701 100644
--- a/lib/rucio/daemons/automatix/automatix.py
+++ b/lib/rucio/daemons/automatix/automatix.py
@@ -217,23 +217,19 @@ def run_once(heartbeat_handler: "HeartbeatHandler", inputfile: str, **_kwargs) -
                 "rse": rse,
                 "path": physical_fname,
             }
+            if set_metadata:
+                file_["dataset_meta"] = metadata
+                if dataset_lifetime:
+                    file_["dataset_meta"]["lifetime"] = dataset_lifetime
             files.append(file_)
         logger(logging.INFO, "Upload %s:%s to %s", scope, dsn, rse)
         upload_client = UploadClient(client)
         ret = upload_client.upload(files)
         if ret == 0:
             logger(logging.INFO, "%s sucessfully registered" % dsn)
-            if set_metadata:
-                client.set_metadata_bulk(
-                    scope=scope, name=dsn, meta=metadata, recursive=False
-                )
-                monitor.record_counter(name="automatix.addnewdataset.done", delta=1)
-                monitor.record_counter(name="automatix.addnewfile.done", delta=nbfiles)
-                timer.record('automatix.datasetinjection')
-            if dataset_lifetime:
-                client.set_metadata(
-                    scope=scope, name=dsn, key="lifetime", value=dataset_lifetime
-                )
+            monitor.record_counter(name="automatix.addnewdataset.done", delta=1)
+            monitor.record_counter(name="automatix.addnewfile.done", delta=nbfiles)
+            timer.record('automatix.datasetinjection')
         else:
             logger(logging.INFO, "Error uploading files")
         for physical_fname in physical_fnames:
diff --git a/lib/rucio/tests/test_automatix.py b/lib/rucio/tests/test_automatix.py
index 235411f47..227f7d942 100644
--- a/lib/rucio/tests/test_automatix.py
+++ b/lib/rucio/tests/test_automatix.py
@@ -23,7 +23,7 @@ import pytest
 
 from rucio.common.config import config_add_section, config_has_section, config_set, config_remove_option
 from rucio.common.types import InternalScope
-from rucio.core.did import list_dids, list_files
+from rucio.core.did import list_dids, list_files, get_metadata
 from rucio.core.scope import add_scope
 from rucio.daemons.automatix.automatix import automatix
 from rucio.rse import rsemanager as rsemgr
@@ -76,6 +76,10 @@ def test_automatix(vo, root_account, rse_factory):
             )
         ]
         assert len(dids) == 1
+        meta = get_metadata(InternalScope(scope, vo), dids[0])
+        assert meta["project"] == project
+        assert meta["datatype"] == 'AOD'
+        assert meta["prod_step"] == 'recon'
         rse_info = rsemgr.get_rse_info(rse, vo)
         files = [file_ for file_ in list_files(InternalScope(scope, vo), dids[0])]
         for file_ in files: