Add support for CMS Monitoring VictoriaMetrics back-end

spider_cms.py

@@ -98,6 +98,9 @@ def main():
     parser.add_argument(
         "--feed_amq", action="store_true", dest="feed_amq", help="Feed to CERN AMQ"
     )
+    parser.add_argument(
+        "--feed_vm", action="store_true", dest="feed_vm", help="Feed to CMS Monitoring VictoriaMetrics"
+    )
 
     parser.add_argument(
         "--schedd_filter",
@@ -210,6 +213,14 @@ def main():
             "[default: %(default)s]"
         ),
     )
+    parser.add_argument(
+        "--vm_url",
+        default="http://cms-monitoring.cern.ch:30422",
+        type=str,
+        dest="vm_url",
+        help="CMS Monitoring VM url to be used "
+        "[default: %(default)s]",
+    )
     parser.add_argument(
         "--log_dir",
         default="log/",

src/htcondor_es/history.py

@@ -112,6 +112,12 @@ def process_schedd(
                             for id_, dict_ad in ad_list
                         ]
                         htcondor_es.amq.post_ads(data_for_amq, metadata=metadata)
+                    if args.feed_vm:
+                        data = [
+                            (id_, convert_dates_to_millisecs(dict_ad))
+                            for id_, dict_ad in ad_list
+                        ]
+                        htcondor_es.vm.post_ads(args.vm_url, data, metadata=metadata)
 
                 logging.debug(
                     "...posting %d ads from %s (process_schedd)",

src/htcondor_es/queues.py

@@ -317,6 +317,16 @@ def _callback_amq(result):
             )
             futures.append(("UPLOADER_AMQ", future))
 
+        if args.feed_vm and not args.read_only:
+            data_bunch = [
+                (id_, convert_dates_to_millisecs(dict_ad)) for id_, dict_ad in bunch
+            ]
+            future = upload_pool.apply_async(
+                htcondor_es.vm.post_ads,
+                args=(args.vm_url, data_bunch, metadata)
+            )
+            futures.append(("UPLOADER_VM", future))
+
         logging.info("Starting new uploader, %d items in queue" % output_queue.qsize())
 
     listener.join()

src/htcondor_es/vm.py

@@ -0,0 +1,21 @@
+import os
+import time
+import logging
+import requests
+
+
+def post_ads(url, ads, metadata=None):
+    "Post classAds docs into VM url"
+    if not len(ads):
+        logging.warning("No new documents found")
+        return
+
+    starttime = time.time()
+    failed = []
+    data = [ad for _, ad in ads]
+    req = requests.post(url, data)
+    if req.status_code != 200:
+        failed = data
+        logging.warning("Fail to send to {}, reason {}".format(url, req.reason))
+    elapsed = time.time() - starttime
+    return (len(ads) - len(failed), len(ads), elapsed)