Merge pull request #56 from ARGOeu/devel

Release Version:   0.3.5

argo-nagios-ams-publisher.spec

@@ -10,7 +10,7 @@
 %endif
 
 Name:           argo-nagios-ams-publisher
-Version:        0.3.4
+Version:        0.3.5
 Release:        1%{mydist}
 Summary:        Bridge from Nagios to the ARGO Messaging system
 
@@ -109,6 +109,8 @@ if ! /usr/bin/getent group nagiocmd &>/dev/null; then
 fi
 
 %changelog
+* Wed Apr 17 2019 Daniel Vrcic <dvrcic@srce.hr> - 0.3.5-1%{?dist}
+- ARGO-1726 Pass site name in metric results  
 * Wed Mar 6 2019 Daniel Vrcic <dvrcic@srce.hr> - 0.3.4-1%{?dist}
 - verbose log messages for errors not handled in argo-ams-library
 * Tue Feb  5 2019 Daniel Vrcic <dvrcic@srce.hr> - 0.3.3-1%{?dist}

pymod/metrictoqueue.py

@@ -17,10 +17,12 @@
 conf = '/etc/argo-nagios-ams-publisher/ams-publisher.conf'
 logfile = '/var/log/argo-nagios-ams-publisher/ams-publisher.log'
 
+
 def seteuser(user):
     os.setegid(user.pw_gid)
     os.seteuid(user.pw_uid)
 
+
 def build_msg(args, *headers):
     msg = Message()
     msg.header = dict()
@@ -35,13 +37,14 @@ def build_msg(args, *headers):
     msg.header.update({'status': status.encode('utf-8')})
     msg.header.update({'monitoring_host': nagioshost.encode('utf-8')})
 
-    for bs in ['summary', 'message', 'vofqan', 'voname', 'roc', 'actual_data']:
+    for bs in ['summary', 'message', 'vofqan', 'voname', 'roc', 'actual_data', 'site']:
         code = "msg.body += '%s: ' + args.%s.encode(\'utf-8\') + '\\n' if args.%s else ''" % (bs, bs, bs)
         exec code
 
     msg.text = True
     return msg
 
+
 def main():
     parser = argparse.ArgumentParser()
     lobj = log.Logger(sys.argv[0], logfile)
@@ -67,6 +70,7 @@ def main():
     parser.add_argument('--summary', required=False, type=str)
     parser.add_argument('--vofqan', required=False, type=str)
     parser.add_argument('--voname', required=False, type=str)
+    parser.add_argument('--site', required=False, type=str)
 
     args = parser.parse_args()
 

pymod/publish.py

@@ -11,6 +11,7 @@
 from argo_nagios_ams_publisher.stats import StatSig
 from argo_ams_library.amsexceptions import AmsConnectionException, AmsServiceException
 
+
 class Publish(StatSig):
     """
        Base publisher class that initialize statistic data
@@ -27,6 +28,7 @@ def _increm_intervalcounters(self, num):
         counter[now] = num + counter.get(now, 0)
         self.shared.statint[self.name]['published_periodic'] += num
 
+
 class FilePublisher(Publish):
     """
        Publisher that write the messages into a file. Used only for debugging
@@ -57,6 +59,7 @@ def write(self, num=0):
             self.shared.log.error(e)
             return False, published
 
+
 class MessagingPublisher(Publish):
     """
        MessagingPublisher class that dispatch messages to ARGO Messaging
@@ -97,6 +100,7 @@ def _avro_serialize(msg):
         plainmsg = dict()
         plainmsg.update(msg.header)
         plainmsg.update(self.body2dict(msg.body))
+        plainmsg.update(tags=self.tag2dict(msg.body))
         timestamp = plainmsg.get('timestamp', None)
 
         m = None
@@ -107,17 +111,36 @@ def _avro_serialize(msg):
 
         return _part_date(timestamp), m
 
-    def body2dict(self, body):
-        d = dict()
+    def _extract_body(self, body, fields, maps=None):
+        msg = dict()
+
         bodylines = body.split('\n')
         for line in bodylines:
             split = line.split(': ', 1)
             if len(split) > 1:
                 key = split[0]
                 value = split[1]
-                d[key] = value.decode('utf-8', 'replace')
 
-        return d
+                if key not in set(fields):
+                    continue
+
+                if maps and key in maps:
+                    key = maps[key]
+
+                msg[key] = value.decode('utf-8', 'replace')
+
+        return msg
+
+    def body2dict(self, body):
+        body_fields = ['summary', 'message', 'actual_data']
+
+        return self._extract_body(body, body_fields)
+
+    def tag2dict(self, body):
+        tag_fields = ['vofqan', 'voname', 'roc', 'site']
+        body_to_tagname = dict(site='endpoint_group')
+
+        return self._extract_body(body, tag_fields, body_to_tagname)
 
     def _write(self, msgs):
         t = 1

pymod/run.py

@@ -47,8 +47,8 @@ def init_dirq_consume(workers, daemonized, sockstat):
 
         if not getattr(shared, 'runtime', False):
             shared.runtime = dict()
-            shared.runtime['started']  = datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')
-            shared.runtime['started_epoch']  = str(int(time.time()))
+            shared.runtime['started'] = datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')
+            shared.runtime['started_epoch'] = str(int(time.time()))
 
         if shared.general['publishmsgfile']:
             shared.runtime.update(publisher=FilePublisher)