Copy the prometheus transferred files function into the instrument server calls

Author: stephen-riggs

src/murfey/client/multigrid_control.py

@@ -223,6 +223,13 @@ def rsync_result(update: RSyncerUpdate):
                     self.rsync_processes[update.base_path].enqueue(update.file_path)
 
             self.rsync_processes[source].subscribe(rsync_result)
+            self.rsync_processes[source].subscribe(
+                partial(
+                    self._increment_transferred_files_prometheus,
+                    destination=destination,
+                    source=str(source),
+                )
+            )
             self.rsync_processes[source].subscribe(
                 partial(
                     self._increment_transferred_files,
@@ -466,6 +473,33 @@ def _increment_file_count(
         }
         requests.post(url, json=data)
 
+    # Prometheus can handle higher traffic so update for every transferred file rather
+    # than batching as we do for the Murfey database updates in _increment_transferred_files
+    def _increment_transferred_files_prometheus(
+        self, update: RSyncerUpdate, source: str, destination: str
+    ):
+        if update.outcome is TransferResult.SUCCESS:
+            url = f"{str(self._environment.url.geturl())}/visits/{str(self._environment.visit)}/increment_rsync_transferred_files_prometheus"
+            data_files = (
+                [update]
+                if update.file_path.suffix in self._data_suffixes
+                and any(
+                    substring in update.file_path.name
+                    for substring in self._data_substrings
+                )
+                else []
+            )
+            data = {
+                "source": source,
+                "destination": destination,
+                "session_id": self.session_id,
+                "increment_count": 1,
+                "bytes": update.file_size,
+                "increment_data_count": len(data_files),
+                "data_bytes": sum(f.file_size for f in data_files),
+            }
+            requests.post(url, json=data)
+
     def _increment_transferred_files(
         self, updates: List[RSyncerUpdate], source: str, destination: str
     ):