diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..681af72 --- /dev/null +++ b/.gitignore @@ -0,0 +1 @@ +src/ps_collector/*.pyc diff --git a/Dockerfile b/Dockerfile new file mode 100644 index 0000000..937b421 --- /dev/null +++ b/Dockerfile @@ -0,0 +1,15 @@ + + +FROM python:2-alpine3.7 + +RUN apk update && apk add ca-certificates && rm -rf /var/cache/apk/* + +ADD . /ps_collector +WORKDIR /ps_collector +RUN pip install -r requirements.txt + +RUN python setup.py install + +EXPOSE 8000 + +CMD ps-collector diff --git a/Makefile b/Makefile deleted file mode 100644 index e67c08d..0000000 --- a/Makefile +++ /dev/null @@ -1,41 +0,0 @@ -prefix := /usr -localstatedir := /var -sysconfdir := /etc -bindir := $(prefix)/bin -datadir := $(prefix)/share -initrddir := $(sysconfdir)/rc.d/init.d -libexecdir := $(prefix)/libexec -mandir := $(prefix)/share/man - - - -_default: - @echo "No default. Try 'make install'" - -install: - # Install executables - install -d $(DESTDIR)/$(libexecdir)/rsv - cp -r libexec/probes $(DESTDIR)/$(libexecdir)/rsv/ - cp -r libexec/metrics $(DESTDIR)/$(libexecdir)/rsv/ - # Install configuration - install -d $(DESTDIR)/$(sysconfdir)/rsv/meta - cp -r etc/meta/metrics $(DESTDIR)/$(sysconfdir)/rsv/meta/ - cp -r etc/metrics $(DESTDIR)/$(sysconfdir)/rsv/ - # Install configuration files for message broker - install -d $(DESTDIR)/$(sysconfdir)/rsv/stompclt - cp -r etc/stompclt $(DESTDIR)/$(sysconfdir)/rsv/ - # Install the simplevisor init script - install -d $(DESTDIR)/$(initrddir) - install -m 0755 init/simplevisor.init $(DESTDIR)/$(initrddir)/simplevisor - # Install the /var/rsv directory - install -d $(DESTDIR)/$(localstatedir)/rsv - install -d $(DESTDIR)/$(localstatedir)/rsv/localenv - # Install the message passing directory - install -d $(DESTDIR)/$(localstatedir)/run/rsv-perfsonar - #Install condor-cron configs - install -d $(DESTDIR)/$(sysconfdir)/condor-cron/config.d - cp -r etc/condor-cron/config.d $(DESTDIR)/$(sysconfdir)/condor-cron/ - - -.PHONY: _default install - diff --git a/README b/README deleted file mode 100644 index a31cb53..0000000 --- a/README +++ /dev/null @@ -1,6 +0,0 @@ -Usage -1. Install the rsv-perfsonar RPM -2. Enable the local network probe -rsv-control --enable org.osg.local.network-monitoring-local --host myhost - -Where myhost is the host where rsvpersonar is installed diff --git a/README.md b/README.md new file mode 100644 index 0000000..a170e84 --- /dev/null +++ b/README.md @@ -0,0 +1,30 @@ + +PS Collector +============ + +The PS Collector queries [perfSONAR](https://www.perfsonar.net/) instances and sends the data onto a RabbitMQ message bus. + +The collector is packages in a docker conatiner available at: + + +## Configuration + +The default configuration is in configs/config.ini. Additional configuration can be added +to config.d directory, usually installed in /etc/ps-collector/config.d/. An example RabbitMQ +configuration is: + + [rabbitmq] + + username = nma + password = password + rabbit_host = example.com + virtual_host = vhost + queue = osg-nma-q + exchange = osg.ps.raw + routing_key = osg-nma-q + +Place this in a file in the config.d directory. + +## Running the Collector + +An example `docker-compose` file is provided. It is recommended to modify the `docker-compose.yml` file to run and start the collector. diff --git a/bin/ps-collector b/bin/ps-collector new file mode 100755 index 0000000..edf6b39 --- /dev/null +++ b/bin/ps-collector @@ -0,0 +1,5 @@ +#!/usr/bin/env python + +import ps_collector.scheduler + +ps_collector.scheduler.main() diff --git a/configs/10-site-local.ini b/configs/10-site-local.ini new file mode 100644 index 0000000..8b5ded9 --- /dev/null +++ b/configs/10-site-local.ini @@ -0,0 +1,6 @@ + +# +# This is the config file for site-configs. +# Items in this file (or the directory) will override +# default configuration values. +# diff --git a/configs/config.ini b/configs/config.ini new file mode 100644 index 0000000..2a71b3e --- /dev/null +++ b/configs/config.ini @@ -0,0 +1,52 @@ +# +# WARNING: DO NOT EDIT THIS CONFIGURATION FILE. +# It will be overwritten on upgrade. +# +# Instead, put site customizations in /etc/ps-collector/config.d +# + +[General] + +# The location of the config.d directory; place +# site-custom config files here. +#config_directory = /etc/ps-collector/config.d +config_directory = ./config.d + +# The location of the logging configuration +logging_configuration = configs/logging-config.ini + +[Scheduler] + +# The amount of time, in minutes, between +# queries to the mesh configuration. +mesh_interval = 15 + +# The amount of time, in minutes, between +# queries to the remote perfSonar endpoints. +probe_interval = 5 + +# Number of processes to query all of the perfsonar hosts +# Experimentally determined that 200 is a reasonable number +pool_size = 200 + +[Mesh] + +# The mesh configuration to utilize for discovering +# perfSonar endpoints to monitor. +endpoint = https://psconfig.opensciencegrid.org/pub/config + + +[org.osg.general.perfsonar-rabbitmq-simple args] + +summary = False + +maxstart = 32800 +allowedEvents = packet-loss-rate,packet-trace,packet-retransmits,throughput,throughput-subintervals,failures,packet-count-sent,packet-count-lost,histogram-owdelay,histogram-ttl,packet-retransmits-subintervals,packet-loss-rate-bidir,packet-count-lost-bidir +tmpdirectory = /var/lib/ps-collector + +usercert = /etc/grid-security/ps_collector/cert.pem +userkey = /etc/grid-security/ps_collector/key.pem + +#mq max message size +mq-max-message-size = 10000 + diff --git a/configs/logging-config.ini b/configs/logging-config.ini new file mode 100644 index 0000000..1117179 --- /dev/null +++ b/configs/logging-config.ini @@ -0,0 +1,24 @@ + +[loggers] +keys=root + +[handlers] +keys=stdout + +[formatters] +keys=default + +[logger_root] +level=INFO +handlers=stdout + +[handler_stdout] +class=StreamHandler +level=NOTSET +formatter=default +args=(sys.stdout,) + +[formatter_default] +format=%(asctime)s %(levelname)s %(process)d %(name)s: %(message)s +class=logging.Formatter + diff --git a/configs/ps-collector.service b/configs/ps-collector.service new file mode 100644 index 0000000..80ecbdd --- /dev/null +++ b/configs/ps-collector.service @@ -0,0 +1,17 @@ +[Unit] +Description=Data collector for perfSonar endpoints +Requires=network-online.target +After=network-online.target + +[Service] +ExecStart=/usr/bin/ps-collector +User=pscollector +Group=pscollector +Type=simple +Restart=on-abort +RestartSec=1 min +KillMode=control-group +LimitNOFILE=65536 + +[Install] +WantedBy=multi-user.target diff --git a/configs/ps-collector.state b/configs/ps-collector.state new file mode 100644 index 0000000..7a088ee --- /dev/null +++ b/configs/ps-collector.state @@ -0,0 +1,6 @@ + +NOTE: + +This directory contains the state files for the ps-collector service. +They record the last successful timestamp of data collected by the service, +hashed by the name of the remote perfSonar endpoint. diff --git a/etc/condor-cron/config.d/50-rsv-perfsonar.config b/etc/condor-cron/config.d/50-rsv-perfsonar.config deleted file mode 100644 index 87475a8..0000000 --- a/etc/condor-cron/config.d/50-rsv-perfsonar.config +++ /dev/null @@ -1,4 +0,0 @@ -# Prevent swaping from having too many probes running at same time -SCHEDD_INTERVAL = 200 -START_LOCAL_UNIVERSE = TotalLocalJobsRunning < 500 -START_SCHEDULER_UNIVERSE = TotalSchedulerJobsRunning < 500 diff --git a/etc/meta/metrics/org.osg.general.perfsonar-activemq-simple.meta b/etc/meta/metrics/org.osg.general.perfsonar-activemq-simple.meta deleted file mode 100644 index 5af2c3b..0000000 --- a/etc/meta/metrics/org.osg.general.perfsonar-activemq-simple.meta +++ /dev/null @@ -1,9 +0,0 @@ -[org.osg.general.perfsonar-activemq-simple] -default-cron-interval = */10 * * * * -execute = local -service-type = Perfsonar-Monitor -output-format = brief -enable-by-default = False -description = Uploads the information to myosg of a perfsonar ndoe -need-proxy = False -[org.osg.general.perfsonar-activemq-simple env] diff --git a/etc/meta/metrics/org.osg.general.perfsonar-rabbitmq-simple.meta b/etc/meta/metrics/org.osg.general.perfsonar-rabbitmq-simple.meta deleted file mode 100644 index f627bb3..0000000 --- a/etc/meta/metrics/org.osg.general.perfsonar-rabbitmq-simple.meta +++ /dev/null @@ -1,9 +0,0 @@ -[org.osg.general.perfsonar-rabbitmq-simple] -default-cron-interval = */10 * * * * -execute = local -service-type = Perfsonar-Monitor -output-format = brief -enable-by-default = False -description = Uploads the information to myosg of a perfsonar ndoe -need-proxy = False -[org.osg.general.perfsonar-rabbitmq-simple env] diff --git a/etc/meta/metrics/org.osg.general.perfsonar-simple.meta b/etc/meta/metrics/org.osg.general.perfsonar-simple.meta deleted file mode 100644 index 449ba2b..0000000 --- a/etc/meta/metrics/org.osg.general.perfsonar-simple.meta +++ /dev/null @@ -1,9 +0,0 @@ -[org.osg.general.perfsonar-simple] -default-cron-interval = */10 * * * * -execute = local -service-type = Perfsonar-Monitor -output-format = brief -enable-by-default = False -description = Uploads the information to myosg of a perfsonar ndoe -need-proxy = False -[org.osg.general.perfsonar-simple env] diff --git a/etc/meta/metrics/org.osg.local.network-monitoring-local.meta b/etc/meta/metrics/org.osg.local.network-monitoring-local.meta deleted file mode 100644 index 824a341..0000000 --- a/etc/meta/metrics/org.osg.local.network-monitoring-local.meta +++ /dev/null @@ -1,11 +0,0 @@ -[org.osg.local.network-monitoring-local] - -default-cron-interval = */10 * * * * -execute = local -service-type = Perfsonar-Monitor -output-format = brief -need-proxy = false -enable-by-default = false -description = To be written - -[org.osg.local.network-monitoring-local env] \ No newline at end of file diff --git a/etc/metrics/org.osg.general.perfsonar-activemq-simple.conf b/etc/metrics/org.osg.general.perfsonar-activemq-simple.conf deleted file mode 100644 index 543fe6d..0000000 --- a/etc/metrics/org.osg.general.perfsonar-activemq-simple.conf +++ /dev/null @@ -1,23 +0,0 @@ -[org.osg.general.perfsonar-activemq-simple] -cron-interval = */15 * * * * -probe-interval = 60 -job-timeout = 1500 -no-ping = True -[org.osg.general.perfsonar-activemq-simple args] -#This is the good key -start = 26052 -timeout = 1400 -debug = False -summary = True -#maxstart = 36800 -maxstart = 2800 -allowedEvents = packet-loss-rate,packet-trace,packet-retransmits,throughput,throughput-subintervals,failures,packet-count-sent,packet-count-lost,histogram-owdelay,histogram-ttl,packet-retransmits-subintervals,packet-loss-rate-bidir,packet-count-lost-bidir -directoryqueue = /scratch/rsv-perfsonar -tmpdirectory = /usr/local/rsv-perfsonar-timestamps/ -usercert = /etc/grid-security/rsv/rsvcert.pem -userkey = /etc/grid-security/rsv/rsvkey.pem -#mq max message size -mq-max-message-size = 10000 -# Added the granularity 60s is the default -# See https://dirq.readthedocs.io/en/latest/queuesimple.html#dirq.QueueSimple.QueueSimple -granularity=5 diff --git a/etc/metrics/org.osg.general.perfsonar-rabbitmq-simple.conf b/etc/metrics/org.osg.general.perfsonar-rabbitmq-simple.conf deleted file mode 100644 index ccf964d..0000000 --- a/etc/metrics/org.osg.general.perfsonar-rabbitmq-simple.conf +++ /dev/null @@ -1,28 +0,0 @@ -[org.osg.general.perfsonar-rabbitmq-simple] -cron-interval = */15 * * * * -probe-interval = 60 -job-timeout = 1500 -no-ping = True -[org.osg.general.perfsonar-rabbitmq-simple args] -#This is the good key -start = 26052 -timeout = 1400 -debug = False -summary = True -#maxstart = 36800 -maxstart = 32800 -allowedEvents = packet-loss-rate,packet-trace,packet-retransmits,throughput,throughput-subintervals,failures,packet-count-sent,packet-count-lost,histogram-owdelay,histogram-ttl,packet-retransmits-subintervals,packet-loss-rate-bidir,packet-count-lost-bidir -tmpdirectory = /usr/local/rsv-perfsonar-timestamps/ -usercert = /etc/grid-security/rsv/rsvcert.pem -userkey = /etc/grid-security/rsv/rsvkey.pem -#mq max message size -username = emfajard -password = TbE^ gTdfg$ -rabbit_host = event-itb.grid.iu.edu -virtual_host = osg-nma -queue = osg-nma-q -exchange = osg.ps.raw -routing_key = osg-nma-q -#mq max message size -mq-max-message-size = 10000 - diff --git a/etc/metrics/org.osg.general.perfsonar-simple.conf b/etc/metrics/org.osg.general.perfsonar-simple.conf deleted file mode 100644 index 349ede5..0000000 --- a/etc/metrics/org.osg.general.perfsonar-simple.conf +++ /dev/null @@ -1,21 +0,0 @@ -[org.osg.general.perfsonar-simple] -cron-interval = */15 * * * * -probe-interval = 60 -job-timeout = 1500 -no-ping = True -[org.osg.general.perfsonar-simple args] -username = perfsonar3 -#This is the good key -key = a4dfdf481a73f37b9dad39528c6150f2944e0934 -goc = http://fermicloud077.fnal.gov:9090 -start = 26052 -timeout = 1400 -debug = False -summary = True -maxstart = 2800 -allowedEvents = packet-loss-rate,packet-trace,packet-retransmits,throughput,throughput-subintervals,failures,packet-count-sent,packet-count-lost, histogram-owdelay,histogram-ttl,packet-retransmits-subintervals,packet-loss-rate-bidir,packet-count-lost-bidir -tmpdirectory = /usr/local/rsv-perfsonar-timestamps/ -usercert = /etc/grid-security/rsv/rsvcert.pem -userkey = /etc/grid-security/rsv/rsvkey.pem - - diff --git a/etc/metrics/org.osg.local.network-monitoring-local.conf b/etc/metrics/org.osg.local.network-monitoring-local.conf deleted file mode 100644 index a7ad972..0000000 --- a/etc/metrics/org.osg.local.network-monitoring-local.conf +++ /dev/null @@ -1,10 +0,0 @@ -[org.osg.local.network-monitoring-local] -cron-interval = */30 * * * * -[org.osg.local.network-monitoring-local args] -#super_mesh = https://meshconfig.opensciencegrid.org/pub/config -#mesh0 = https://meshconfig.opensciencegrid.org/pub/config/us-atlas -#mesh1 = https://meshconfig.opensciencegrid.org/pub/config/us-cms -mesh0 = http://meshconfig.opensciencegrid.org/pub/config/ps-testbed -dmetric0 = org.osg.general.perfsonar-simple -#dmetric1 = org.osg.general.perfsonar-activemq-simple -#dmetric2 = org.osg.general.perfsonar-rabbitmq-simple \ No newline at end of file diff --git a/etc/stompclt/default.conf b/etc/stompclt/default.conf deleted file mode 100644 index acd9b64..0000000 --- a/etc/stompclt/default.conf +++ /dev/null @@ -1,14 +0,0 @@ -# define the source broker - - - scheme = x509 - cert = /etc/grid-security/rsv/rsvcert.pem # change to certificate path - key = /etc/grid-security/rsv/rsvkey.pem # change to key path - - - - path = /var/run/rsv-perfsonar/ # change this to your directory queue (i.e. should be the same as self.dq) - -# miscellaneous options -loop = true -remove = true diff --git a/etc/stompclt/simplevisor.cfg b/etc/stompclt/simplevisor.cfg deleted file mode 100644 index bcdaee1..0000000 --- a/etc/stompclt/simplevisor.cfg +++ /dev/null @@ -1,39 +0,0 @@ - - store = /var/cache/simplevisor/publisher-simplevisor.json - pidfile = /var/run/publisher-simplevisor.pid - interval = 30 - log = file - logfile = /var/log/publisher-simplevisor.log - loglevel = info - - - type = supervisor - name = perfsonar-supervisor - strategy = one_for_one - - - type = supervisor - name = perfsonar1 - strategy = one_for_one - - - type = service - name = dashb-test-mb - expected = running - start = /usr/bin/stompclt --outgoing-broker-uri stomp+ssl://netmon-test-mb.cern.ch:61523 --conf /etc/rsv/stompclt/default.conf --pidfile /var/lock/perfsonar-stompclt-dashb-test-mb.pid --daemon - stop = /usr/bin/stompclt --pidfile /var/lock/perfsonar-stompclt-dashb-test-mb.pid --quit - status = /usr/bin/stompclt --pidfile /var/lock/perfsonar-stompclt-dashb-test-mb.pid --status - -# One of these needed for every broker -# -# type = service -# name = mb1062.cern.ch -# expected = running -# start = /usr/bin/stompclt --outgoing-broker-uri stomp+ssl://mb106.cern.ch:61123 --conf /etc/rsv-perfsonar/trunk/etc/stompclt/default.conf --pidfile /var/lock/perfsonar-stompclt-mb1062.pid --daemon -# stop = /usr/bin/stompclt --pidfile /var/lock/perfsonar-stompclt-mb1062.pid --quit -# status = /usr/bin/stompclt --pidfile /var/lock/perfsonar-stompclt-mb1062.pid --status -# - - - - diff --git a/init/simplevisor.init b/init/simplevisor.init deleted file mode 100755 index 6f94c66..0000000 --- a/init/simplevisor.init +++ /dev/null @@ -1,107 +0,0 @@ -#!/bin/sh - -### BEGIN INIT INFO -# Provides: simplevisor -# Required-Start: $syslog -# Required-Stop: $syslog -# Default-Start: 2 3 4 5 -# Default-Stop: 0 1 6 -# Short-Description: Start simplevisor -# Description: simplevisor send rsv-perfsonar matrics to cern MQ -### END INIT INFO - -# Source function library. -[ -f /etc/rc.d/init.d/functions ] || exit 0 -. /etc/rc.d/init.d/functions - -cmd="simplevisor --conf /etc/rsv/stompclt/simplevisor.cfg start" -user="" - -name=`basename $0` -pid_file="/var/run/publisher-simplevisor.pid" #needs to match what is in the config - -get_pid() { - cat "$pid_file" -} - -is_running() { - [ -f "$pid_file" ] && ps `get_pid` > /dev/null 2>&1 -} - -case "$1" in - start) - if is_running; then - echo "Already started" - else - echo "Starting $name" - mkdir -p /var/lib/simplevisor - if [ -z "$user" ]; then - chown $user:$user /var/lib/simplevisor - daemon $cmd & - else - daemon --user "$user" $cmd & - fi - echo -n "Waiting for $name to start " - for i in {1..60} - do - if is_running; then - exit 0 - fi - echo -n "." - sleep 1 - done - echo "Failed to start in 30 seconds" - exit 1 - fi - ;; - stop) - if is_running; then - echo -n "Stopping $name (could take up to a minute) " - kill `get_pid` - for i in {1..60} - do - if ! is_running; then - break - fi - - echo -n "." - sleep 1 - done - echo - - if is_running; then - echo "Not stopped; may still be shutting down or shutdown may have failed" - exit 1 - else - echo "Stopped" - if [ -f "$pid_file" ]; then - rm "$pid_file" - fi - fi - else - echo "Not running" - fi - ;; - restart) - $0 stop - if is_running; then - echo "Unable to stop, will not attempt to start" - exit 1 - fi - $0 start - ;; - status) - if is_running; then - echo "Running" - else - echo "Stopped" - exit 1 - fi - ;; - *) - echo "Usage: $0 {start|stop|restart|status}" - exit 1 - ;; -esac - -exit 0 diff --git a/libexec/metrics/org.osg.general.perfsonar-activemq-simple b/libexec/metrics/org.osg.general.perfsonar-activemq-simple deleted file mode 120000 index c92749b..0000000 --- a/libexec/metrics/org.osg.general.perfsonar-activemq-simple +++ /dev/null @@ -1 +0,0 @@ -../probes/perfsonar-activemq-simple-local-probe \ No newline at end of file diff --git a/libexec/metrics/org.osg.general.perfsonar-rabbitmq-simple b/libexec/metrics/org.osg.general.perfsonar-rabbitmq-simple deleted file mode 120000 index 3d7aeaf..0000000 --- a/libexec/metrics/org.osg.general.perfsonar-rabbitmq-simple +++ /dev/null @@ -1 +0,0 @@ -../probes/perfsonar-rabbitmq-simple-local-probe \ No newline at end of file diff --git a/libexec/metrics/org.osg.general.perfsonar-simple b/libexec/metrics/org.osg.general.perfsonar-simple deleted file mode 120000 index 9cef995..0000000 --- a/libexec/metrics/org.osg.general.perfsonar-simple +++ /dev/null @@ -1 +0,0 @@ -../probes/perfsonar-simple-local-probe \ No newline at end of file diff --git a/libexec/metrics/org.osg.local.network-monitoring-local b/libexec/metrics/org.osg.local.network-monitoring-local deleted file mode 120000 index d6fef1d..0000000 --- a/libexec/metrics/org.osg.local.network-monitoring-local +++ /dev/null @@ -1 +0,0 @@ -../probes/network-monitoring-local-probe \ No newline at end of file diff --git a/libexec/probes/network-monitoring-local-probe b/libexec/probes/network-monitoring-local-probe deleted file mode 100755 index 89fec27..0000000 --- a/libexec/probes/network-monitoring-local-probe +++ /dev/null @@ -1,232 +0,0 @@ -#!/usr/bin/python - -import os -import re -import subprocess -import sys -import urllib2 -import rsvprobe -import socket -try:#No json on python2.4 using simplejson for now - import json -except ImportError: - import simplejson as json - -# The maximun number of meshes allowed -max_mesh = 100 -standard_metrics = ['org.osg.general.perfsonar-activemq-simple', 'org.osg.general.perfsonar-rabbitmq-simple', 'org.osg.general.perfsonar-simple'] - -sys.path.insert(0, '.') - - -class NetworkMonitoringLocalProbe(rsvprobe.RSVProbe): - """ - This master probe reads a json and parses it to obtain the urls - of several perfsonar nodes then fires up other dummy probes that contact - the perfsonar nodes and upload the information to myosg - """ - - def __init__(self): - rsvprobe.RSVProbe.__init__(self) - self.metric = "" - self.short_metric = "" - metric = rsvprobe.RSVMetric("Perfsonar-Monitor", - "org.osg.local.network-monitoring-local", rsvprobe.RSVMetric.STATUS) - self.supported_metrics = [metric] - self.details = "---\n" - #Add the probe specific options - # For addin up to max_mesh different specific meshes - for mesh_num in range(0, max_mesh): - mesh_opt = "mesh%d=" % mesh_num - self.addopt("", mesh_opt, "--mesh# url of the personar mesh (ex. --mesh0 http://meshconfig.opensciencegrid.org/pub/config/ps-testbed --mesh2 url2") - # Adding a new super_mesh which is a json with several other meshes - self.addopt("", "super_mesh=", "--super_mesh# url of the goc mesh which has some others (ex. --super_mesh https://meshconfig.opensciencegrid.org/pub/config") - self.meshlist = [] - self.super_mesh = "" - self.nodesIP = [] - - ### Added support for multiple metrics - self.dummyMetrics = [] - for metrics_num in range(0, 3): - metric_opt = "dmetric%d=" % metrics_num - self.addopt("", metric_opt, "--dmetricX metric name (ex. --dmetric0 org.osg.general.osg-version)") - - # Returns a list of ip addres that the hostname is associated with ipv4/ipv6 - def hostnameToIP(self, hostname): - try: - addrTup = socket.getaddrinfo(hostname, 80, 0, 0, socket.IPPROTO_TCP) - except socket.gaierror: - self.add_message("Host %s ip not found, ignoring"%hostname) - return [] - #addr = socket.gethostbyname(hostname) - address = [] - for tup in addrTup: - if len(tup) == 5: - ip = tup[4][0] - address.append(ip) - return address - - def parseSuperMesh(self, super_mesh): - self.add_message("Parsing super mesh %s" % super_mesh) - req = urllib2.Request(super_mesh) - opener = urllib2.build_opener() - f = opener.open(req) - try: - data = json.loads(f.read()) - except ValueError: - self.return_unknown("Invalid json at %s" % (super_mesh)) - for mesh in data: - self.meshlist.append(mesh['include'][0]) - - - # Reads the url from the json url and returns the url nodes of the perfsonar nodes - def parseJsonUrl(self, jsonurl): - self.add_message("Obtaining nodes from mesh %s" % jsonurl) - clean_url = urllib2.quote(jsonurl, safe="%/:=&?~#+!$,;'@()*[]") - req = urllib2.Request(clean_url) - opener = urllib2.build_opener() - f = opener.open(req) - try: - data = json.loads(f.read()) - except ValueError: - raise - if 'tests' not in data: - raise Exception('jsoneror', 'json not correctly formatted') - nodes = [] - for test in data['tests']: - for key in test['members'].keys(): - if 'member' in key: - for node in test['members'][key]: - hostname = node.encode('utf-8') - ips = self.hostnameToIP(hostname) - # Check the ips are not already in nodesIP - for ip in ips: - if not ip in self.nodesIP: - #Each time an ip is not found in the list - self.nodesIP.append(ip) - if not hostname in nodes: - nodes.append(hostname) - return nodes - - def parseopt(self): - """parse options specific to network monitroing probe and return options, optlist and reminder to allow further processing - """ - options, optlist, remainder = rsvprobe.RSVProbe.parseopt(self) - for opt, arg in options: - # Adding the extra meshes - if 'mesh' in opt and ('super' not in opt): - if arg not in self.meshlist: - self.meshlist.append(arg) - if 'super' in opt: - self.super_mesh = arg - elif 'dmetric' in opt: - if arg not in self.dummyMetrics: - self.dummyMetrics.append(arg) - if self.host == self.localhost: - self.is_local = True - else: - self.is_local = False - return options, optlist, remainder - - # Enables a dummy probe for each node in nodes - def enableDummyProbe(self, nodes, metric='org.osg.general.perfsonar-simple'): - for node in nodes: - cmd = "rsv-control --enable --host %s %s" % (node, metric) - ec, out = rsvprobe.run_command(cmd) - # make a warning if the dummy metric failed - result = out.split("\n") - if 'ERROR' in result: - self.add_warning("Failed to enable probe failed for node: %s" % (node), exit_code=0) - cmd = "rsv-control --on --host %s %s" % (node, metric) - ec, out = rsvprobe.run_command(cmd) - result = out.split("\n") - if 'ERROR' in result: - self.add_warning("Failed to turn on probe failed for node: %s, metric %s" % (node, metric), exit_code=0) - - # Disable the dummy probe for a list of nodes - def disableDummyProbe(self, nodes, metric='org.osg.general.perfsonar-simple'): - for node in nodes: - if len(node) < 1: - continue - cmd = "rsv-control --disable --host %s %s" % (node, metric) - ec, out = rsvprobe.run_command(cmd) - # make a warning if the dummy metric failed - result = out.split("\n") - if 'ERROR' in result: - self.add_warning("Failed to enable probe failed for node: %s" % (node), exit_code=0) - cmd = "rsv-control --off --host %s %s" % (node, metric) - ec, out = rsvprobe.run_command(cmd) - result = out.split("\n") - if 'ERROR' in result: - self.add_warning("Failed to turn off probe failed for node: %s" % (node), exit_code=0) - - #Returns a list of nodes that are currently enabled for a given metric - def getListEnablednodes(self, metric): - cmd = "rsv-control --list --wide %s | grep Metric | awk '{print $5}'" % (metric) - ec, out = rsvprobe.run_command(cmd) - result = out.split("\n") - return result - - def getListNodesToDisable(self, actualNodes, enabledNodes): - nodesToDisable = [] - for node in enabledNodes: - if node not in actualNodes and len(node)>0: - nodesToDisable.append(node) - return nodesToDisable - - def getListNodesToEnable(self, actualNodes, enabledNodes): - nodesToEnable = [] - for node in actualNodes: - if node not in enabledNodes: - nodesToEnable.append(node) - return nodesToEnable - - def run(self): - """Main routine for the probe""" - self.parseopt() - if len(self.super_mesh)>0: - self.parseSuperMesh(self.super_mesh) - #Actual nodes are the nodes that should be on according to the last reading of the mesh_rul - actualNodes = [] - failures = False - for mesh_url in self.meshlist: - try: - actualNodes += self.parseJsonUrl(mesh_url) - except AttributeError as inst: - self.add_message(inst) - except Exception as inst: - failures = True - self.add_message("Invalid json at %s" % (mesh_url)) - #self.add_message(inst.strerror) - if failures: - self.return_unknown("Problem parsing a URL") - - for metric in self.dummyMetrics: - self.add_message("Proccesing metric: %s" % metric) - #Enabled nodes is the list of currently enabled nodes - self.add_message("Getting list of enabled nodes") - enabledNodes = self.getListEnablednodes(metric) - self.add_message("Calculating list of enabled and disabled nodes") - nodesToEnable = self.getListNodesToEnable(actualNodes, enabledNodes) - nodesToDisable = self.getListNodesToDisable(actualNodes, enabledNodes) - self .add_message("Enabling nodes %s " % nodesToEnable) - self.enableDummyProbe(nodesToEnable, metric) - self.add_message("Disabling nodes %s " % nodesToDisable) - self.disableDummyProbe(nodesToDisable, metric) - - # This step is to disable the metrics that are no longer present - for metric in standard_metrics: - if metric not in self.dummyMetrics: - # So if a metric is no longer present turn off all of the corresponding probes - self.add_message("Disabling nodes for metric %s: cause the metric is no longer in the metrics conf file " % metric) - nodesToDisable = self.getListEnablednodes(metric) - self.add_message("Disabling nodes %s " % nodesToDisable) - self.disableDummyProbe(nodesToDisable, metric) - self.return_ok("Everything OK") - -def main(): - probe = NetworkMonitoringLocalProbe() - return probe.run() - -if __name__ == '__main__': - sys.exit(main()) diff --git a/libexec/probes/perfsonar-activemq-simple-local-probe b/libexec/probes/perfsonar-activemq-simple-local-probe deleted file mode 100755 index 884a343..0000000 --- a/libexec/probes/perfsonar-activemq-simple-local-probe +++ /dev/null @@ -1,26 +0,0 @@ -#!/usr/bin/python -#import perfsonar_simple_local_probe -#import perfsonar_basic_probe -from perfsonar_basic_probe import * - -#sys.path.insert(0, '.') -class PerfsonarActiveMQSimpleProbe(PerfsonarSimpleProbe): - """ - TODO - write description of the probe - """ - def __init__(self): - PerfsonarSimpleProbe.__init__(self) - self.metricName = "org.osg.general.perfsonar-activemq-simple" - metric = rsvprobe.RSVMetric("Perfsonar-Monitor", - self.metricName, rsvprobe.RSVMetric.STATUS) - self.supported_metrics = [metric] - self.addopt("", 'directoryqueue=', "--directoryqueue a dir for stompctl") - self.addopt("", 'granularity=', "--granularity the granularity of the directory quee") - self.addopt("", 'mq-max-message-size=', "--mq-max-message-size the size in KB of the biggest message allowed in the Mq") - -def main(): - probe = PerfsonarActiveMQSimpleProbe() - return probe.run() - -if __name__ == '__main__': - sys.exit(main()) diff --git a/libexec/probes/perfsonar-rabbitmq-simple-local-probe b/libexec/probes/perfsonar-rabbitmq-simple-local-probe deleted file mode 100755 index 6c6edd6..0000000 --- a/libexec/probes/perfsonar-rabbitmq-simple-local-probe +++ /dev/null @@ -1,31 +0,0 @@ -#!/usr/bin/python -#import perfsonar_simple_local_probe -#import perfsonar_basic_probe -from perfsonar_basic_probe import * - -#sys.path.insert(0, '.') -class PerfsonarRabbitMQSimpleProbe(PerfsonarSimpleProbe): - """ - TODO - write description of the probe - """ - def __init__(self): - PerfsonarSimpleProbe.__init__(self) - self.metricName = "org.osg.general.perfsonar-rabbitmq-simple" - metric = rsvprobe.RSVMetric("Perfsonar-Monitor", - self.metricName, rsvprobe.RSVMetric.STATUS) - self.supported_metrics = [metric] - self.addopt("", 'mq-max-message-size=', "--mq-max-message-size the size in KB of the biggest message allowed in the Mq") - self.addopt("", 'username=', "--username for the RabbitMq") - self.addopt("", 'password=', "--password for the Rabbit Mq") - self.addopt("", 'rabbit_host=', "--host hostname where the Rabbit Mq is hosted") - self.addopt("", 'virtual_host=',"--virtual host of the rabbit mq") - self.addopt("", 'queue=', "--queue for the rabbit mq") - self.addopt("", 'exchange=', "--exchange for the rabbit mq") - self.addopt("", 'routing_key=', "--routing_key for the rabbit mq") - -def main(): - probe = PerfsonarRabbitMQSimpleProbe() - return probe.run() - -if __name__ == '__main__': - sys.exit(main()) diff --git a/libexec/probes/perfsonar-simple-local-probe b/libexec/probes/perfsonar-simple-local-probe deleted file mode 100755 index 6c606b4..0000000 --- a/libexec/probes/perfsonar-simple-local-probe +++ /dev/null @@ -1,26 +0,0 @@ -#!/usr/bin/python - -from perfsonar_basic_probe import * - -#sys.path.insert(0, '.') -class PerfsonarEsmondProbe(PerfsonarSimpleProbe): - """ - Probe that reads from a perfsonar and posts it to an esmond central data store - """ - def __init__(self): - PerfsonarSimpleProbe.__init__(self) - self.metricName = "org.osg.general.perfsonar-simple" - metric = rsvprobe.RSVMetric("Perfsonar-Esmond-Uploader", - self.metricName, rsvprobe.RSVMetric.STATUS) - self.supported_metrics = [metric] - self.addopt("", "username=", "--username username the username for uploading data to the goc") - self.addopt("", "key=", "--key key the key for uploading data to the goc") - self.addopt("", "goc=", "--goc url the url for where to upload the data (i.e http://psds0.opensciengrid.org)") - -def main(): - probe = PerfsonarEsmondProbe() - return probe.run() - -if __name__ == '__main__': - sys.exit(main()) - diff --git a/libexec/probes/perfsonar_basic_probe.py b/libexec/probes/perfsonar_basic_probe.py deleted file mode 100644 index b0f99fa..0000000 --- a/libexec/probes/perfsonar_basic_probe.py +++ /dev/null @@ -1,143 +0,0 @@ -#!/usr/bin/python - -import os -import re -import subprocess -import sys -import rsvprobe -import time -import random - -import time -from time import strftime -from time import localtime - -sys.path.insert(0, '.') -class PerfsonarSimpleProbe(rsvprobe.RSVProbe): - """ - TODO - write description of the probe - """ - def __init__(self): - rsvprobe.RSVProbe.__init__(self) - metric = rsvprobe.RSVMetric("Perfsonar-Monitor", - "org.osg.general.perfsonar-simple", rsvprobe.RSVMetric.STATUS) - self.supported_metrics = [metric] - self.details = "---\n" - # Add the config directory - self.conf_dir = os.path.join("/", "etc", "rsv", "metrics") - self.start = 900 - self.debug = False - self.maxstart = 43200 - self.summaries = False - self.allowedEvents = "packet-loss-rate, throughput, packet-trace, packet-retransmits, histogram-owdelay, packet-count-sent,packet-count-lost" - self.soft_timeout = 1400 - self.tmpdirectory = '/tmp/rsv-perfsonar/' - # Add the options so the parsing knows what to expect - self.addopt("", "start=", "--start How back in history get data from perfsonar ndoe in secs (i.e 43200)") - # sleep left here but not used anymore just for compatiblity - self.addopt("", "debug=", "--debug True or False. For if extra debugging is needed") - self.addopt("", "timeout=", "--timeout Seconds. A softimeout for how long the probes are allowed to run") - self.addopt("", "summary=", "--summary True or False. Read and upload data summaries or not") - self.addopt("", "maxstart=", "--maxstart the max number in seconds it will go in the past to retrieve information") - self.addopt("", "allowedEvents=", "--allowedEvents a list with the allowed event types") - self.addopt("", 'tmpdirectory=', "--tmpdirectory a directory to store temporary timestamps must be read/writebl by rsv") - - def parseopt(self): - """parse options specific to network monitroing probe and return options, optlist and reminder to - allow further processing - """ - options, optlist, remainder = rsvprobe.RSVProbe.parseopt(self) - for opt, arg in options: - if opt == '--start': - self.start = arg - elif opt == '--debug': - self.debug = arg - elif opt == '--timeout': - self.soft_timeout = arg - elif opt == '--summary': - self.summaries = arg - elif opt == '--maxstart': - self.maxstart = arg - elif opt == '--tmpdirectory': - self.tmpdirectory = arg - elif opt == '--allowedEvents': - # Replacing white spaces with nothing - self.allowedEvents = arg.replace(" ", '') - if self.host == self.localhost: - self.is_local = True - else: - self.is_local = False - return options, optlist, remainder - - def ReadTimeStampFile(self, filename): - if not os.path.isfile(filename): - return 1 - else: - nfile = open(filename, 'r') - return nfile.readline() - - def createDir(self, directory): - if not os.path.exists(directory): - os.makedirs(directory) - return directory - - def WriteNewTimestamp(self, filename, starttime): - nfile = open(filename, 'w') - time = strftime("%a, %d %b %Y %H:%M:%S", starttime) - nfile.write(time) - - def computeStartTime(self, filename): - timestamp = self.ReadTimeStampFile(filename) - timeFormat = "%a, %d %b %Y %H:%M:%S" - if not timestamp == 1:#The probe ran succesfully: - try: - oldTime = time.strptime(timestamp, timeFormat) - self.add_message("Last succesfull run at %s" % timestamp) - newStart = time.mktime(localtime())-time.mktime(oldTime) - except ValueError: - self.out_debug("Timestamp %s, in file %s not in correct format %s" % (timestamp, filename, timeFormat)) - newStart = self.start - # If it the value is less than the one allowed use maxstart - if newStart < int(self.maxstart): - self.start = int(newStart) - else: - self.add_message("previous time_start %s too old. set to maxstart: %s" % (newStart, self.maxstart) ) - self.start = int(self.maxstart) - else: - self.add_message("No previous sucesfull run found") - - def runCallerScript(self): - cmd = "source ./uploader.sh %s %d %s %s" % (self.host, int(self.start), self.soft_timeout, self.metricName) - self.add_message("Command call %s" % cmd) - ec, out = rsvprobe.run_command(cmd, workdir="/usr/libexec/rsv/probes/worker-scripts") - return out - - def run(self): - """Main routine for the probe""" - self.parseopt() - #In /var/log/rsv/metrics/ each probe creates a dir with the last time it ran succesfully if it is not yet there - basedir = os.path.join("/var/log/rsv/metrics/", self.supported_metrics[0].name) - basedir = self.createDir(basedir) - directoryFile = self.createDir(os.path.join(basedir,self.host)) - timeFile = os.path.join(directoryFile, 'timeFile.out') - # computeStartTime basically reads the file assumes the date there is the time_end in the filter probe of the last succesfull probe - # and computes how long in the past it has to look back, basically now - time_end of last sucesfull run. - self.computeStartTime(timeFile) - startTime = localtime() - # Some dictionaries where last ran actually happen will be created on the /tmps - tmpdirectory = os.path.join(self.tmpdirectory, self.supported_metrics[0].name, self.host) - self.tmpdirectory = self.createDir(tmpdirectory) - # Calling the caller script and parsing it is output - out = self.runCallerScript() - reverOut = out.split('\n') - reverOut.reverse() - for line in reverOut: - self.add_message(line) - if 'ERROR' in out: - self.return_critical("Failed running the caller") - self.WriteNewTimestamp(timeFile, startTime) - if 'WARNING' in out: - self.return_warning('WARNING: timeout ') - self.return_ok("OK") - - diff --git a/libexec/probes/worker-scripts/uploader.sh b/libexec/probes/worker-scripts/uploader.sh deleted file mode 100755 index 878a58a..0000000 --- a/libexec/probes/worker-scripts/uploader.sh +++ /dev/null @@ -1,6 +0,0 @@ -#!/bin/bash -host=$1 start=$2 timeout=$3 metricName=$4 -#enabling the python2.7 enviroment -. /var/rsv/localenv/bin/activate -python uploader/caller.py -s $start -u $host -t $timeout -C $metricName -deactivate diff --git a/libexec/probes/worker-scripts/uploader/activemquploader.py b/libexec/probes/worker-scripts/uploader/activemquploader.py deleted file mode 100644 index 157072b..0000000 --- a/libexec/probes/worker-scripts/uploader/activemquploader.py +++ /dev/null @@ -1,119 +0,0 @@ -from uploader import * - -# Need to push to cern message queue - -from messaging.message import Message -from messaging.queue.dqs import DQS - -class ActiveMQUploader(Uploader): - - def __init__(self, start = 1600, connect = 'iut2-net3.iu.edu', metricName='org.osg.general-perfsonar-simple.conf'): - Uploader.__init__(self, start, connect, metricName) - self.maxMQmessageSize = self.readConfigFile('mq-max-message-size') - #Code to allow publishing data to the mq - self.mq = None - self.dq = self.readConfigFile('directoryqueue') - self.granularity = int(self.readConfigFile('granularity')) - if self.dq != None and self.dq!='None': - try: - self.mq = DQS(path=self.dq, granularity=self.granularity) - except Exception as e: - self.add2log("Unable to create dirq %s, exception was %s, " % (self.dq, e)) - - - # Publish summaries to Mq - def publishSToMq(self, arguments, event_types, summaries, summaries_data): - # the max size limit in KB but python expects it in bytes - size_limit = self.maxMQmessageSize * 1000 - for event in summaries_data.keys(): - if not summaries_data[event]: - continue - msg_head = { 'input-source' : arguments['input_source'], - 'input-destination' : arguments['input_destination'], - 'org_metadata_key' : arguments['org_metadata_key'], - 'event-type' : event, - 'rsv-timestamp' : "%s" % time.time(), - 'summaries' : 1, - 'destination' : '/topic/perfsonar.summary.' + event } - msg_body = { 'meta': arguments } - msg_body['summaries'] = summaries_data[event] - size_summ = self.total_size(summaries_data[event]) - msg = Message(body=json.dumps(msg_body), header=msg_head) - size_msg = msg.size() - #self.add2log("Message size: %s" % size_msg) - # if size of the message is larger than 10MB discarrd - if size_msg > size_limit or sys.getsizeof(json.dumps(msg_body)) > size_limit or size_summ > size_limit: - self.add2log("Size of message body bigger than limit, discarding") - continue - # add to mq - try: - self.mq.add_message(msg) - except Exception as e: - self.add2log("Failed to add message to mq %s, exception was %s" % (self.dq, e)) - - # Publish message to Mq - def publishRToMq(self, arguments, event_types, datapoints): - for event in datapoints.keys(): - # filter events for mq (must be subset of the probe's filter) - if event not in self.allowedEvents: - continue - # skip events that have no datapoints - if not datapoints[event]: - continue - # compose msg - msg_head = { 'input-source' : arguments['input_source'], - 'input-destination' : arguments['input_destination'], - 'org_metadata_key' : arguments['org_metadata_key'], - # including the timestart of the smalles measureement - 'ts_start' : min(datapoints[event].keys()), - 'event-type' : event, - 'rsv-timestamp' : "%s" % time.time(), - 'summaries' : 0, - 'destination' : '/topic/perfsonar.raw.' + event} - msg_body = { 'meta': arguments } - msg_body['datapoints'] = datapoints[event] - msg = Message(body=json.dumps(msg_body), header=msg_head) - # add to mq - try: - self.mq.add_message(msg) - except Exception as e: - self.add2log("Failed to add message to mq %s, exception was %s" % (self.dq, e)) - - def postData(self, arguments, event_types, summaries, summaries_data, metadata_key, datapoints): - summary= self.summary - disp = self.debug - lenght_post = -1 - arguments['org_metadata_key'] = metadata_key - for event_type in datapoints.keys(): - if len(datapoints[event_type])>lenght_post: - lenght_post = len(datapoints[event_type]) - if lenght_post == 0: - self.add2log("No new datapoints skipping posting for efficiency") - return - if self.mq and summaries_data: - self.add2log("posting new summaries") - self.publishSToMq(arguments, event_types, summaries, summaries_data) - step_size = 100 - for step in range(0, lenght_post, step_size): - chunk_datapoints = {} - for event_type in datapoints.keys(): - chunk_datapoints[event_type] = {} - if len(datapoints[event_type].keys())>0: - pointsconsider = sorted(datapoints[event_type].keys())[step:step+step_size] - for point in pointsconsider: - chunk_datapoints[event_type][point] = datapoints[event_type][point] - if self.mq: - self.publishRToMq(arguments, event_types, chunk_datapoints) - # Updating the checkpoint files for each host/metric and metadata - for event_type in datapoints.keys(): - if len(datapoints[event_type].keys()) > 0: - if event_type not in self.time_starts: - self.time_starts[event_type] = 0 - next_time_start = max(datapoints[event_type].keys())+1 - if next_time_start > self.time_starts[event_type]: - self.time_starts[event_type] = int(next_time_start) - f = open(self.tmpDir + metadata_key, 'w') - f.write(json.dumps(self.time_starts)) - f.close() - self.add2log("posting NEW METADATA/DATA to Cern Active MQ %s" % metadata_key) - diff --git a/libexec/probes/worker-scripts/uploader/caller.py b/libexec/probes/worker-scripts/uploader/caller.py deleted file mode 100644 index a1c3fd1..0000000 --- a/libexec/probes/worker-scripts/uploader/caller.py +++ /dev/null @@ -1,60 +0,0 @@ -from time import sleep -from time import strftime -from time import localtime -import sys -import signal -import os -from optparse import OptionParser - -try: - from activemquploader import * - from esmonduploader import * - from rabbitmquploader import * -except Exception as err: - print "ERROR:! Importing libraries! Exception: \"%s\" of type: \"%s\" was thrown! Quitting out." % (err,type(err)) - sys.exit(1) - -# Set command line options -parser = OptionParser() -parser.add_option('-s', '--start', help='set start time for gathering data (default is -12 hours)', dest='start', default=960) -parser.add_option('-u', '--url', help='set url to gather data from (default is http://hcc-pki-ps02.unl.edu)', dest='url', default='http://hcc-pki-ps02.unl.edu') -parser.add_option('-t', '--timeout', help='the maxtimeout that the probe is allowed to run in secs', dest='timeout', default=1000, action='store') -# Name of the metric -parser.add_option('-C','--metric', help='Metric name', default='org.osg.general-perfsonar-simple', dest='metricName', action='store') -(opts, args) = parser.parse_args() - -caller = None -metricName = opts.metricName -if metricName == 'org.osg.general.perfsonar-activemq-simple': - caller = ActiveMQUploader(start=int(opts.start), connect=opts.url, metricName=metricName) -elif metricName == 'org.osg.general.perfsonar-simple': - caller = EsmondUploader(start=int(opts.start), connect=opts.url, metricName=metricName) -elif metricName == 'org.osg.general.perfsonar-rabbitmq-simple': - caller = RabbitMQUploader(start=int(opts.start), connect=opts.url, metricName=metricName) -else: - print "ERROR: metric not supporred" - - -def get_post(): - try: - caller.getData() - caller.add2log("Finished getting and posting data succesfully!") - sys.exit(0) - except Exception, err: - print Exception, err - print "ERROR:! Unsuccessful! Exception: \"%s\" of type: \"%s\" was thrown! Quitting out." % (err,type(err)) - sys.exit(1) - - -def handler(signum, frame): - caller.add2log('WARNING: Running time took more than %d seconds' % int(opts.timeout)) - sys.exit(0) - -# Option: Get and Post Metadata -if True: - # Implementing some timeout - signal.signal(signal.SIGALRM, handler) - signal.alarm(int(opts.timeout)) - get_post() - signal.alarm(0) - diff --git a/libexec/probes/worker-scripts/uploader/esmonduploader.py b/libexec/probes/worker-scripts/uploader/esmonduploader.py deleted file mode 100644 index d73edce..0000000 --- a/libexec/probes/worker-scripts/uploader/esmonduploader.py +++ /dev/null @@ -1,126 +0,0 @@ -from uploader import * -# Esmond libraries to post back to the data store -from esmond_client.perfsonar.post import MetadataPost, EventTypePost, EventTypeBulkPost -from esmond_client.perfsonar.post import EventTypeBulkPostWarning, EventTypePostWarning - -class EsmondUploader(Uploader): - - def __init__(self, start = 1600, connect = 'iut2-net3.iu.edu', metricName='org.osg.general-perfsonar-simple.conf'): - Uploader.__init__(self, start = start, connect = connect, metricName = metricName) - self.username = self.readConfigFile('username') - self.key = self.readConfigFile('key') - self.goc = self.readConfigFile('goc') - - gfilters = ApiFilters() - # gfiltesrs and in general g* means connecting to the cassandra db at the central place ie goc - gfilters.verbose = False - gfilters.time_start = int(self.time_end - 5*start) - gfilters.time_end = self.time_end - gfilters.input_source = connect - self.gconn = ApiConnect(self.goc, gfilters) - - - def postMetaData(self, arguments, event_types, summaries, summaries_data, metadata_key, datapoints, summary = True, disp=False): - mp = MetadataPost(self.goc, username=self.username, api_key=self.key, **arguments) - for event_type in summaries.keys(): - mp.add_event_type(event_type) - if summary: - summary_window_map = {} - #organize summaries windows by type so that all windows of the same type are in an array - for summy in summaries[event_type]: - if summy[0] not in summary_window_map: - summary_window_map[summy[0]] = [] - summary_window_map[summy[0]].append(summy[1]) - #Add each summary type once and give the post object the array of windows - for summary_type in summary_window_map: - mp.add_summary_type(event_type, summary_type, summary_window_map[summary_type]) - # Added the old metadata key - mp.add_freeform_key_value("org_metadata_key", metadata_key) - new_meta = mp.post_metadata() - return new_meta - - # Post data points from a metadata - def postBulkData(self, new_meta, old_metadata_key, datapoints, disp=False): - et = EventTypeBulkPost(self.goc, username=self.username, api_key=self.key, metadata_key=new_meta.metadata_key) - for event_type in datapoints.keys(): - for epoch in datapoints[event_type]: - # packet-loss-rate is read as a float but should be uploaded as a dict with denominator and numerator - if event_type in ['packet-loss-rate', 'packet-loss-rate-bidir']: - # Some extra protection incase the number of datapoints in packet-loss-setn and packet-loss-rate does not match - packetcountsent = 210 - packetcountlost = 0 - specialTypes = ['packet-count-sent', 'packet-count-lost'] - if event_type == 'packet-loss-rate-bidir': - specialTypes = ['packet-count-sent', 'packet-count-lost-bidir'] - for specialType in specialTypes: - if not epoch in datapoints[specialType].keys(): - self.add2log("Something went wrong time epoch %s not found for %s fixing it" % (specialType, epoch)) - time.sleep(5) - datapoints_added = self.getMissingData(epoch, old_metadata_key, specialType) - # Try to get the data once more because we know it is there - - try: - value = datapoints_added[specialType][epoch] - except Exception as err: - datapoints_added[specialType][epoch] = 0 - value = datapoints_added[specialType][epoch] - datapoints[specialType][epoch] = value - et.add_data_point(specialType, epoch, value) - - packetcountsent = datapoints['packet-count-sent'][epoch] - if event_type == 'packet-loss-rate-bidir': - packetcountlost = datapoints['packet-count-lost-bidir'][epoch] - else: - packetcountlost = datapoints['packet-count-lost'][epoch] - et.add_data_point(event_type, epoch, {'denominator': packetcountsent, 'numerator': packetcountlost}) - # For the rests the data points are uploaded as they are read - else: - # datapoint are tuples the first field is epoc the second the value - et.add_data_point(event_type, epoch, datapoints[event_type][epoch]) - if disp: - self.add2log("Datapoints to upload:") - self.add2log(et.json_payload()) - with warnings.catch_warnings(record=True) as w: - warnings.simplefilter('error', EventTypePostWarning) - try: - et.post_data() - # Some EventTypePostWarning went wrong: - except Exception as err: - self.add2log("Probably this data already existed") - #self.postDataSlow(json.loads(et.json_payload()), new_meta.metadata_key, datapoints, disp) - for event_type in datapoints.keys(): - if len(datapoints[event_type].keys()) > 0: - if event_type not in self.time_starts: - self.time_starts[event_type] = 0 - next_time_start = max(datapoints[event_type].keys())+1 - if next_time_start > self.time_starts[event_type]: - self.time_starts[event_type] = int(next_time_start) - f = open(self.tmpDir + old_metadata_key, 'w') - f.write(json.dumps(self.time_starts)) - f.close() - self.add2log("posting NEW METADATA/DATA %s" % new_meta.metadata_key) - - def postData(self, arguments, event_types, summaries, summaries_data, metadata_key, datapoints, summary = True, disp=False): - lenght_post = -1 - for event_type in datapoints.keys(): - if len(datapoints[event_type])>lenght_post: - lenght_post = len(datapoints[event_type]) - new_meta = self.postMetaData(arguments, event_types, summaries, summaries_data, metadata_key, datapoints, summary, disp) - # Catching bad posts - if new_meta is None: - raise Exception("Post metadata empty, possible problem with user and key") - if lenght_post == 0: - self.add2log("No new datapoints skipping posting for efficiency") - return - step_size = 100 - for step in range(0, lenght_post, step_size): - chunk_datapoints = {} - for event_type in datapoints.keys(): - chunk_datapoints[event_type] = {} - if len(datapoints[event_type].keys())>0: - pointsconsider = sorted(datapoints[event_type].keys())[step:step+step_size] - for point in pointsconsider: - chunk_datapoints[event_type][point] = datapoints[event_type][point] - self.postBulkData(new_meta, metadata_key, chunk_datapoints, disp=False) - - diff --git a/logrotate/rsv-perfsonar-metrics.logrotate b/logrotate/rsv-perfsonar-metrics.logrotate deleted file mode 100644 index 31ae984..0000000 --- a/logrotate/rsv-perfsonar-metrics.logrotate +++ /dev/null @@ -1,17 +0,0 @@ -/var/log/rsv/metrics/*org.osg.general.perfsonar-simple.log { - weekly - rotate 2 - compress - compressoptions -9f - copytruncate - missingok -} - -/var/log/rsv/metrics/*org.osg.local.network-monitoring-local.log { - weekly - rotate 2 - compress - compressoptions -9f - copytruncate - missingok -} diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..e6f6207 --- /dev/null +++ b/requirements.txt @@ -0,0 +1,4 @@ +schedule +pika +esmond-client +prometheus_client diff --git a/rpm/ps-collector.spec b/rpm/ps-collector.spec new file mode 100644 index 0000000..6d98628 --- /dev/null +++ b/rpm/ps-collector.spec @@ -0,0 +1,68 @@ +Name: ps-collector +Version: 0.1.0 +Release: 1%{?dist} +Summary: perfSonar data collector +Packager: SAND-CI +Group: Applications/Monitoring +License: Apache 2.0 +URL: https://sand-ci.org + +Source0: %{name}-%{version}.tar.gz + +Requires(pre): shadow-utils + +BuildRequires: python2-devel +BuildRequires: python-setuptools + +BuildRequires: systemd +%{?systemd_requires} + +BuildArch: noarch + +# Requires: python2-schedule + +%description +%{summary} + +%prep +%setup -n %{name}-%{version} + +%build +%py2_build + +%install +%py2_install + +%pre +getent group pscollector >/dev/null || groupadd -f -r pscollector +if ! getent passwd pscollector >/dev/null ; then + useradd -r -g pscollector -d %{_localstatedir}/lib/%{name} -s /sbin/nologin -c "Runtime account for the ps-collector" pscollector +fi +exit 0 + +%files +%doc README +%defattr(-,root,root,-) +%config %{_sysconfdir}/%{name}/config.ini +%config %{_sysconfdir}/%{name}/logging-config.ini +%config(noreplace) %{_sysconfdir}/%{name}/config.d/* +%attr(-,pscollector,pscollector) %{_localstatedir}/lib/%{name} +%{_bindir}/%{name} +%{_unitdir}/%{name}.service + +%{python2_sitelib}/ps_collector +%{python2_sitelib}/ps_collector-*.egg-info + +%post +%systemd_post ps-collector.service + +%preun +%systemd_preun ps-collector.service + +%postun +%systemd_postun_with_restart ps-collector.service + +%changelog +* Tue Dec 18 2018 Brian Bockelman - 0.1.0-1 +- Initial version of the new ps-collector software. + diff --git a/setup.py b/setup.py new file mode 100644 index 0000000..526688e --- /dev/null +++ b/setup.py @@ -0,0 +1,24 @@ + +""" +Install file for the perfSonar collector project. +""" + +import setuptools + +setuptools.setup(name="ps-collector", + version="0.1.0", + description="A daemon for aggregating perfSonar measurements", + author_email="discuss@sand-ci.org", + author="Brian Bockelman", + url="https://sand-ci.org", + package_dir={"": "src"}, + packages=["ps_collector"], + scripts=['bin/ps-collector'], + install_requires=['schedule', 'pika', 'esmond-client'], + data_files=[('/etc/ps-collector', ['configs/config.ini', 'configs/logging-config.ini']), + ('/etc/ps-collector/config.d', ['configs/10-site-local.ini']), + ('/usr/lib/systemd/system', ['configs/ps-collector.service']), + ('/var/lib/ps-collector', ['configs/ps-collector.state']) + ] + + ) diff --git a/libexec/probes/worker-scripts/uploader/SSLNodeInfo.py b/src/ps_collector/SSLNodeInfo.py similarity index 100% rename from libexec/probes/worker-scripts/uploader/SSLNodeInfo.py rename to src/ps_collector/SSLNodeInfo.py diff --git a/libexec/probes/worker-scripts/uploader/SocksApiConnect.py b/src/ps_collector/SocksApiConnect.py similarity index 100% rename from libexec/probes/worker-scripts/uploader/SocksApiConnect.py rename to src/ps_collector/SocksApiConnect.py diff --git a/libexec/probes/worker-scripts/uploader/SocksSSLApiConnect.py b/src/ps_collector/SocksSSLApiConnect.py similarity index 100% rename from libexec/probes/worker-scripts/uploader/SocksSSLApiConnect.py rename to src/ps_collector/SocksSSLApiConnect.py diff --git a/src/ps_collector/__init__.py b/src/ps_collector/__init__.py new file mode 100644 index 0000000..7952340 --- /dev/null +++ b/src/ps_collector/__init__.py @@ -0,0 +1,11 @@ + +import sharedrabbitmq + +# Shared RabbitMQ +shared_rabbitmq = None + +def get_rabbitmq_connection(cp): + global shared_rabbitmq + if shared_rabbitmq == None: + shared_rabbitmq = sharedrabbitmq.SharedRabbitMQ(cp) + return shared_rabbitmq \ No newline at end of file diff --git a/src/ps_collector/config.py b/src/ps_collector/config.py new file mode 100644 index 0000000..f43534d --- /dev/null +++ b/src/ps_collector/config.py @@ -0,0 +1,27 @@ + +import ConfigParser +import logging.config +import os + + +def get_config(): + cp = ConfigParser.ConfigParser() + config_file = os.environ.get("PS_COLLECTOR_CONFIG", "/etc/ps-collector/config.ini") + cp.read(config_file) + if cp.has_section("General") and cp.has_option("General", "config_directory"): + config_dir = cp.get("General", "config_directory") + file_list = list(os.listdir(config_dir)) + file_list.sort() + for fname in file_list: + if fname.startswith(".") or fname.endswith(".rpmsave") or fname.endswith(".rpmnew"): + continue + cp.read(os.path.join(config_dir, fname)) + return cp + + +def setup_logging(cp): + config_file = "/etc/ps-collector/logging-config.ini" + if cp.has_option("General", "logging_configuration"): + config_file = cp.get("General", "logging_configuration") + logging.config.fileConfig(config_file) + diff --git a/src/ps_collector/docker-compose.yml b/src/ps_collector/docker-compose.yml new file mode 100644 index 0000000..824f99d --- /dev/null +++ b/src/ps_collector/docker-compose.yml @@ -0,0 +1,12 @@ +version: '3' +services: + perfsonar: + build: . + volumes: + - ./config.d:/etc/ps-collector/config.d + - /etc/grid-security/rsv/rsvcert.pem:/etc/grid-security/ps_collector/cert.pem + - /etc/grid-security/rsv/rsvkey.pem:/etc/grid-security/ps_collector/key.pem + - /var/lib/ps-collector:/var/lib/ps-collector + restart: always + ports: + - "8000:8000" diff --git a/src/ps_collector/mesh.py b/src/ps_collector/mesh.py new file mode 100644 index 0000000..84d87e6 --- /dev/null +++ b/src/ps_collector/mesh.py @@ -0,0 +1,59 @@ +from __future__ import print_function + +import requests +import json +from urlparse import urlparse + + +class Mesh: + def __init__(self, base_url = "https://psconfig.opensciencegrid.org/pub/config"): + self.base_url = base_url + + def get_nodes(self): + """ + Get all the nodes we should collect statistics + + :returns list: List of nodes in the mesh + """ + # Get the top level config, with list of URLs + meshes = self._download_toplevel() + nodes = set() + for mesh_url in meshes: + nodes.update(self._download_nodes(mesh_url)) + + return nodes + + + def _download_nodes(self, mesh_url): + """ + Download the nodes from a single mesh + """ + nodes = set() + + response = requests.get(mesh_url) + response_json = response.json() + + for org in response_json.get('organizations', []): + for site in org.get('sites', []): + for host in site.get('hosts', []): + for url in host.get('measurement_archives', []): + parsed = urlparse(url['read_url']) + nodes.update([parsed.netloc]) + return nodes + + def _download_toplevel(self): + """ + Download the list of URLs for the meshes + """ + to_return = [] + response = requests.get(self.base_url) + for sub_mesh in response.json(): + to_return.append(sub_mesh['include'][0]) + return to_return + + +if __name__ == "__main__": + mesh = Mesh() + print(mesh.get_nodes()) + + diff --git a/src/ps_collector/monitoring.py b/src/ps_collector/monitoring.py new file mode 100644 index 0000000..d1eea56 --- /dev/null +++ b/src/ps_collector/monitoring.py @@ -0,0 +1,17 @@ + +import multiprocessing +import time + +class timed_execution: + + def __init__(self, endpoint, q): + self.start_time = 0 + self.q = q + self.endpoint = endpoint + + def __enter__(self): + self.start_time = time.time() + + def __exit__(self, type, value, traceback): + self.q.put([self.endpoint, time.time() - self.start_time]) + diff --git a/libexec/probes/worker-scripts/uploader/rabbitmquploader.py b/src/ps_collector/rabbitmquploader.py similarity index 61% rename from libexec/probes/worker-scripts/uploader/rabbitmquploader.py rename to src/ps_collector/rabbitmquploader.py index a0bb367..b15b4ec 100644 --- a/libexec/probes/worker-scripts/uploader/rabbitmquploader.py +++ b/src/ps_collector/rabbitmquploader.py @@ -1,32 +1,28 @@ -from uploader import * +from uploader import Uploader # Need to push to Rabbit mq import pika +from ps_collector.sharedrabbitmq import SharedRabbitMQ +import ps_collector +import time +import json +import tempfile +import os +import errno class RabbitMQUploader(Uploader): - def __init__(self, start = 1600, connect = 'iut2-net3.iu.edu', metricName='org.osg.general-perfsonar-simple.conf'): - Uploader.__init__(self, start, connect, metricName) + def __init__(self, start = 1600, connect = 'iut2-net3.iu.edu', + metricName='org.osg.general.perfsonar-rabbitmq-simple', + config = None, log = None): + Uploader.__init__(self, start, connect, metricName, config, log) + + self.channel = ps_collector.get_rabbitmq_connection(config).createChannel() self.maxMQmessageSize = self.readConfigFile('mq-max-message-size') - self.username = self.readConfigFile('username') - self.password = self.readConfigFile('password') - self.rabbithost = self.readConfigFile('rabbit_host') - self.virtual_host = self.readConfigFile('virtual_host') - self.queue = self.readConfigFile('queue') - self.exchange = self.readConfigFile('exchange') - self.routing_key = self.readConfigFile('routing_key') - self.connection = None - self.channel = None - try: - credentials = pika.PlainCredentials(self.username, self.password) - self.parameters = pika.ConnectionParameters(host=self.rabbithost,virtual_host=self.virtual_host,credentials=credentials) - self.ch_prop = pika.BasicProperties(delivery_mode = 2) #Make message persistent - except Exception as e: - self.add2log("ERROR: Unable to create dirq channgel, exception was %s, " % (repr(e))) def __del__(self): + self.log.exception("Del odd") if self.channel and self.channel.is_open: self.channel.close() - self.connection.close() # Publish summaries to Mq @@ -47,24 +43,24 @@ def SendMessagetoMQ(self, msg_body, event): size_msg = self.total_size(msg_body) # if size of the message is larger than 10MB discarrd if size_msg > size_limit: - self.add2log("Size of message body bigger than limit, discarding") + self.log.warning("Size of message body bigger than limit, discarding") return # add to mq result = None for tries in range(5): try: - result = self.channel.basic_publish(exchange = self.exchange, + result = self.channel.basic_publish(exchange = self.config.get('rabbitmq', 'exchange'), routing_key = 'perfsonar.raw.' + event, body = json.dumps(msg_body), - properties = self.ch_prop) + properties = pika.BasicProperties(delivery_mode = 2)) break if not result: raise Exception('ERROR: Exception publishing to rabbit MQ', 'Problem publishing to mq') except Exception as e: - self.add2log("Restarting pika connection,, exception was %s, " % (repr(e))) - self.restartPikaConnection() + self.log.exception("Restarting pika connection,, exception was %s, " % (repr(e))) + ps_collector.get_rabbitmq_connection(self.config).createChannel() if result == None: - self.add2log("ERROR: Failed to send message to mq, exception was %s" % (repr(e))) + self.log.error("ERROR: Failed to send message to mq, exception was %s" % (repr(e))) # Publish message to Mq def publishRToMq(self, arguments, event_types, datapoints): @@ -85,21 +81,6 @@ def publishRToMq(self, arguments, event_types, datapoints): msg_body = { 'meta': arguments } msg_body['datapoints'] = datapoints[event] self.SendMessagetoMQ(msg_body, event) - - def restartPikaConnection(self): - if self.channel and self.channel.is_open: - try: - self.channel.close() - except Exception as e: - self.add2log("Failed to close the channel exception was %s" % (repr(e))) - if self.connection and self.connection.is_open: - try: - self.connection.close() - except Exception as e: - self.add2log("Failed to close the connection exception was %s" % (repr(e))) - self.connection = pika.BlockingConnection(self.parameters) - self.channel = self.connection.channel() - def postData(self, arguments, event_types, summaries, summaries_data, metadata_key, datapoints): summary= self.summary @@ -110,22 +91,18 @@ def postData(self, arguments, event_types, summaries, summaries_data, metadata_k if len(datapoints[event_type])>lenght_post: lenght_post = len(datapoints[event_type]) if lenght_post == 0: - self.add2log("No new datapoints skipping posting for efficiency") + self.log.info("No new datapoints skipping posting for efficiency") return # Now that we know we have data to send, actually connect upstream. - if self.connection is None: - try: - self.restartPikaConnection() - except Exception as e: - self.add2log("Unable to create channel to RabbitMQ, exception was %s" % repr(e)) - return + if self.channel is None: + self.channel = ps_collector.get_rabbitmq_connection(self.config).createChannel() if summaries_data: - self.add2log("posting new summaries") + self.log.info("posting new summaries") self.publishSToMq(arguments, event_types, summaries, summaries_data) step_size = 200 - self.add2log("Length of the post: %s " % lenght_post) + self.log.info("Length of the post: %s " % lenght_post) for step in range(0, lenght_post, step_size): chunk_datapoints = {} for event_type in datapoints.keys(): @@ -143,7 +120,27 @@ def postData(self, arguments, event_types, summaries, summaries_data, metadata_k next_time_start = max(chunk_datapoints[event_type].keys())+1 if next_time_start > self.time_starts[event_type]: self.time_starts[event_type] = int(next_time_start) - f = open(self.tmpDir + metadata_key, 'w') - f.write(json.dumps(self.time_starts)) - f.close() - self.add2log("posting NEW METADATA/DATA to rabbitMQ %s" % metadata_key) + + self.writeCheckpoint(metadata_key, self.time_starts) + self.log.info("posting NEW METADATA/DATA to rabbitMQ %s" % metadata_key) + + def writeCheckpoint(self, metadata_key, times): + """ + Perform an atomic write to the checkpoint file + """ + # Make sure that the directory exists + try: + os.makedirs(self.tmpDir) + except OSError as e: + if e.errno != errno.EEXIST: + raise + + # Open a temporary file + (file_handle, tmp_filename) = tempfile.mkstemp(dir=self.tmpDir) + wrapped_file = os.fdopen(file_handle, 'w') + wrapped_file.write(json.dumps(times)) + wrapped_file.close() + os.rename(tmp_filename, self.tmpDir + metadata_key) + + + diff --git a/src/ps_collector/scheduler.py b/src/ps_collector/scheduler.py new file mode 100644 index 0000000..e895881 --- /dev/null +++ b/src/ps_collector/scheduler.py @@ -0,0 +1,169 @@ +from __future__ import print_function + +import functools +import logging +import multiprocessing +from multiprocessing import Queue +import Queue +import time + +import schedule +from prometheus_client import Summary, Gauge +from prometheus_client import start_http_server + + +import ps_collector.config +import ps_collector.sharedrabbitmq +from ps_collector.rabbitmquploader import RabbitMQUploader +from ps_collector.mesh import Mesh +import ps_collector +from ps_collector.monitoring import timed_execution + +# The conversion factor from minutes to seconds: +MINUTE = 60 + +log = None + +communication_manager = multiprocessing.Manager() +communication_queue = communication_manager.Queue() + +class SchedulerState(object): + + def __init__(self, cp, pool, log): + self.pool = pool + self.cp = cp + self.probes = set() + self.futures = {} + self.log = log + + +IN_PROGRESS = Gauge("ps_inprogress_requests", "Number of requests queued or running") +request_summary = Summary('ps_request_latency_seconds', 'How long the request to the remote perfsonar took', ['endpoint']) +NUM_ENDPOINTS = Gauge("ps_num_endpoint", "How many endpoints are being queried") + +def query_ps_child(cp, endpoint): + reverse_dns = endpoint.split(".") + reverse_dns = ".".join(reverse_dns[::-1]) + log = logging.getLogger("perfSonar.{}".format(reverse_dns)) + log.info("I query endpoint {}.".format(endpoint)) + with timed_execution(endpoint, communication_queue): + RabbitMQUploader(connect=endpoint, config=cp, log = log).getData() + +def query_ps(state, endpoint): + old_future = state.futures.get(endpoint) + if old_future: + if not old_future.ready(): + state.log.info("Prior probe {} is still running; skipping query.".format(endpoint)) + return + # For now, ignore the result. + try: + old_future.get() + except Exception as e: + state.log.exception("Failed to get data last time:") + + result = state.pool.apply_async(query_ps_child, (state.cp, endpoint)) + IN_PROGRESS.inc() + state.futures[endpoint] = result + + +def query_ps_mesh(state): + state.log.info("Querying PS mesh") + # TODO: get a list of endpoints from the configured mesh config. + + mesh_endpoint = state.cp.get("Mesh", "endpoint") + + mesh = Mesh(state.cp.get("Mesh", "endpoint")) + endpoints = mesh.get_nodes() + state.log.info("Nodes: %s", endpoints) + + running_probes = set(state.probes) + probes_to_stop = running_probes.difference(endpoints) + probes_to_start = endpoints.difference(running_probes) + + for probe in probes_to_stop: + state.log.debug("Stopping probe: %s", probe) + NUM_ENDPOINTS.dec() + state.probes.remove(probe) + schedule.clear(probe) + future = state.futures.get(probe) + if not future: + continue + future.wait() + + default_probe_interval = state.cp.getint("Scheduler", "probe_interval") * MINUTE + + for probe in probes_to_start: + state.log.debug("Adding probe: %s", probe) + NUM_ENDPOINTS.inc() + state.probes.add(probe) + probe_interval = default_probe_interval + if state.cp.has_section(probe) and state.cp.has_option("interval"): + probe_interval = state.cp.getint(probe, "interval") * MINUTE + + query_ps_job = functools.partial(query_ps, state, probe) + # Run the probe the first time + query_ps_job() + schedule.every(probe_interval).to(probe_interval + MINUTE).seconds.do(query_ps_job).tag(probe) + + time.sleep(5) + logging.debug("Finished querying mesh") + + +def cleanup_futures(state): + # Loop through the futures, cleaning up those that are done, and dec the gauge + for endpoint in state.futures: + cur_future = state.futures[endpoint] + if cur_future: + if cur_future.ready(): + try: + cur_future.get() + except Exception as e: + state.log.exception("Failed to get data last time for endpoint {0}:".format(endpoint)) + state.futures[endpoint] = None + IN_PROGRESS.dec() + +def main(): + global MINUTE + cp = ps_collector.config.get_config() + if cp.has_option("Scheduler", "debug"): + if cp.get("Scheduler", "debug").lower() == "true": + MINUTE = 1 + ps_collector.config.setup_logging(cp) + global log + log = logging.getLogger("scheduler") + + pool_size = 5 + if cp.has_option("Scheduler", "pool_size"): + pool_size = cp.getint("Scheduler", "pool_size") + pool = multiprocessing.Pool(pool_size) + + state = SchedulerState(cp, pool, log) + + # Query the mesh the first time + query_ps_mesh(state) + + query_ps_mesh_job = functools.partial(query_ps_mesh, state) + cleanup_futures_job = functools.partial(cleanup_futures, state) + + mesh_interval_s = cp.getint("Scheduler", "mesh_interval") * MINUTE + log.info("Will update the mesh config every %d seconds.", mesh_interval_s) + schedule.every(mesh_interval_s).to(mesh_interval_s + MINUTE).seconds.do(query_ps_mesh_job) + + schedule.every(10).seconds.do(cleanup_futures_job) + + # Start the prometheus webserver + start_http_server(8000) + try: + while True: + schedule.run_pending() + while True: + try: + item = communication_queue.get(True, 1) + print("Got from queue: {0}".format(item)) + request_summary.labels(item[0]).observe(item[1]) + except Queue.Empty: + break + except: + pool.terminate() + pool.join() + raise diff --git a/src/ps_collector/sharedrabbitmq.py b/src/ps_collector/sharedrabbitmq.py new file mode 100644 index 0000000..c6acaf3 --- /dev/null +++ b/src/ps_collector/sharedrabbitmq.py @@ -0,0 +1,37 @@ + +import pika + + +# Singleton from: +# https://python-3-patterns-idioms-test.readthedocs.io/en/latest/Singleton.html +class SharedRabbitMQ: + def __init__(self, config): + # Create the connection to the bus + self.config = config + self.username = self._readConfig('username') + self.password = self._readConfig('password') + self.rabbithost = self._readConfig('rabbit_host') + self.virtual_host = self._readConfig('virtual_host') + self.queue = self._readConfig('queue') + self.exchange = self._readConfig('exchange') + self.routing_key = self._readConfig('routing_key') + self.createConection() + + def createConection(self): + + credentials = pika.PlainCredentials(self.username, self.password) + self.parameters = pika.ConnectionParameters(host=self.rabbithost,virtual_host=self.virtual_host,credentials=credentials) + self.conn = pika.BlockingConnection(self.parameters) + + def _readConfig(self, option): + return self.config.get('rabbitmq', option) + + def createChannel(self): + """ + Create a channel and return it + """ + if not self.conn.is_open: + self.createConection() + return self.conn.channel() + + diff --git a/libexec/probes/worker-scripts/uploader/uploader.py b/src/ps_collector/uploader.py similarity index 70% rename from libexec/probes/worker-scripts/uploader/uploader.py rename to src/ps_collector/uploader.py index 650bb15..dfc0ded 100644 --- a/libexec/probes/worker-scripts/uploader/uploader.py +++ b/src/ps_collector/uploader.py @@ -10,7 +10,7 @@ # Using the esmond_client instead of the rpm from esmond_client.perfsonar.query import ApiFilters -from esmond_client.perfsonar.query import ApiConnect +from esmond_client.perfsonar.query import ApiConnect, ApiConnectWarning # New module with socks5 OR SSL connection that inherits ApiConnect from SocksSSLApiConnect import SocksSSLApiConnect @@ -22,20 +22,19 @@ filters = ApiFilters() class Uploader(object): - - def add2log(self, log): - print strftime("%a, %d %b %Y %H:%M:%S", localtime()), str(log) - def __init__(self, start = 1600, connect = 'iut2-net3.iu.edu', metricName='org.osg.general-perfsonar-simple.conf'): + def __init__(self, start = 1600, + connect = 'iut2-net3.iu.edu', + metricName='org.osg.general-perfsonar-simple.conf', + config = None, log = None): + self.log = log + self.config = config ######################################## ### New Section to read directly the configuration file self.metricName = metricName - conf_dir = os.path.join("/", "etc", "rsv", "metrics") - self.configFile = os.path.join(conf_dir, metricName + ".conf") - self.add2log("Configuration File: %s" % self.configFile) - self.config = ConfigParser.RawConfigParser() + #conf_dir = os.path.join("/", "etc", "rsv", "metrics") ######################################## - self.debug = self.str2bool(self.readConfigFile('debug')) + self.debug = self.str2bool(self.readConfigFile('debug', "false")) verbose = self.debug # Filter variables filters.verbose = verbose @@ -50,8 +49,8 @@ def __init__(self, start = 1600, connect = 'iut2-net3.iu.edu', metricName='org.o # For logging pourposes filterDates = (strftime("%a, %d %b %Y %H:%M:%S ", time.gmtime(self.time_start)), strftime("%a, %d %b %Y %H:%M:%S", time.gmtime(self.time_end))) #filterDates = (strftime("%a, %d %b %Y %H:%M:%S ", time.gmtime(filters.time_start))) - self.add2log("Data interval is from %s to %s" %filterDates) - self.add2log("Metada interval is from %s to now" % (filters.time_start)) + self.log.info("Data interval is from %s to %s" %filterDates) + self.log.info("Metada interval is from %s to now" % (filters.time_start)) # Connection info to the perfsonar remote host that the data is gathered from. self.connect = connect self.conn = SocksSSLApiConnect("http://"+self.connect, filters) @@ -70,13 +69,13 @@ def __init__(self, start = 1600, connect = 'iut2-net3.iu.edu', metricName='org.o # Get Data def getData(self): disp = self.debug - self.add2log("Only reading data for event types: %s" % (str(self.allowedEvents))) + self.log.info("Only reading data for event types: %s" % (str(self.allowedEvents))) summary = self.summary disp = self.debug if summary: - self.add2log("Reading Summaries") + self.log.info("Reading Summaries") else: - self.add2log("Omiting Sumaries") + self.log.info("Omitting Summaries") period = 3600 for new_time_start in range(self.time_start, self.time_end, period): self.getDataHourChunks(new_time_start, new_time_start + period) @@ -84,31 +83,45 @@ def getData(self): def getDataHourChunks(self, time_start, time_end): filters.time_start = time_start filters.time_end = time_end + + # Do we have to use Socks? if self.useSSL == True: self.conn = SocksSSLApiConnect("https://"+self.connect, filters) else: self.conn = SocksSSLApiConnect("http://"+self.connect, filters) metadata = self.conn.get_metadata() + #self.conn = None + #metadata = None + #try: + # self.conn = ApiConnect("http://" + self.connect, filters) + # metadata = self.conn.get_metadata() + #except ApiConnectWarning as apiwarning: + # self.log.exception("Unable to connect to host") + # return try: #Test to see if https connection is succesfull md = metadata.next() self.readMetaData(md) except StopIteration: - self.add2log("There is no metadat in this time range") + self.log.warning("There is no metadat in this time range") return except ConnectionError as e: #Test to see if https connection is sucesful - self.add2log("Unable to connect to %s, exception was %s, trying SSL" % ("http://"+self.connect, type(e))) + self.log.exception("Unable to connect to %s, exception was %s, trying SSL" % ("http://"+self.connect, type(e))) try: metadata = self.conn.get_metadata(cert=self.cert, key=self.certkey) md = metadata.next() self.useSSL = True self.readMetaData(md) except StopIteration: - self.add2log("There is no metadat in this time range") + self.log.warning("There is no metadata in this time range") except ConnectionError as e: raise Exception("Unable to connect to %s, exception was %s, " % ("https://"+self.connect, type(e))) + except Exception as e: + self.log.exception("Failed to read metadata") + return + for md in metadata: self.readMetaData(md) @@ -136,10 +149,10 @@ def readMetaData(self, md): event_types = md.event_types metadata_key = md.metadata_key # print extra debugging only if requested - self.add2log("Reading New METADATA/DATA %s" % (md.metadata_key)) + self.log.info("Reading New METADATA/DATA %s" % (md.metadata_key)) if disp: - self.add2log("Posting args: ") - self.add2log(arguments) + self.log.debug("Posting args: ") + self.log.debug(arguments) # Get Events and Data Payload summaries = {} summaries_data = {} @@ -154,10 +167,10 @@ def readMetaData(self, md): self.time_starts = json.loads(f.read()) f.close() except IOError: - self.add2log("first time for %s" % (md.metadata_key)) + self.log.exception("first time for %s" % (md.metadata_key)) except ValueError: # decoding failed - self.add2log("first time for %s" % (md.metadata_key)) + self.log.exception("first time for %s" % (md.metadata_key)) for et in md.get_all_event_types(): if self.useSSL: etSSL = EventTypeSSL(et, self.cert, self.certkey) @@ -167,7 +180,7 @@ def readMetaData(self, md): et.filters.time_start = filters.time_start if et.event_type in self.time_starts.keys(): et.filters.time_start = self.time_starts[et.event_type] - self.add2log("loaded previous time_start %s" % et.filters.time_start) + self.log.info("loaded previous time_start %s" % et.filters.time_start) et.filters.time_end = filters.time_end if et.filters.time_end < et.filters.time_start: continue @@ -208,18 +221,22 @@ def readMetaData(self, md): tup = (dp.ts_epoch, dp.val) datapoints[eventype][dp.ts_epoch] = dp.val # print debugging data - self.add2log("For event type %s, %d new data points" %(eventype, len(datapoints[eventype]))) + self.log.info("For event type %s, %d new data points" %(eventype, len(datapoints[eventype]))) if len(datapoints[eventype]) > 0 and not isinstance(tup[1], (dict,list)): # picking the first one as the sample datapointSample[eventype] = tup[1] - self.add2log("Sample of the data being posted %s" % datapointSample) + self.log.debug("Sample of the data being posted %s" % datapointSample) try: self.postData(arguments, event_types, summaries, summaries_data, metadata_key, datapoints) except Exception as e: - raise Exception("Unable to post because exception: %s" % repr(e)) + import traceback + traceback.print_exc() + raise e + #raise Exception("Unable to post because exception: %s" % repr(e)) # Experimental function to try to recover from missing packet-count-sent or packet-count-lost data def getMissingData(self, timestamp, metadata_key, event_type): + self.log.info("Trying to get missing data") disp = self.debug filtersEsp = ApiFilters() filtersEsp.verbose = disp @@ -243,7 +260,7 @@ def getMissingData(self, timestamp, metadata_key, event_type): dpay = et.get_data() for dp in dpay.data: if dp.ts_epoch == timestamp: - self.add2log("point found") + self.log.debug("point found") datapoints[event_type][dp.ts_epoch] = dp.val return datapoints @@ -251,49 +268,48 @@ def getMissingData(self, timestamp, metadata_key, event_type): def postData(self, arguments, event_types, summaries, summaries_data, metadata_key, datapoints): pass - def readConfigFile(self, key): + def readConfigFile(self, key, default = ""): section = self.metricName + " args" - ret = self.config.read(self.configFile) try: return self.config.get(section, key) except ConfigParser.NoOptionError: - self.add2log("ERROR: config knob %s not found in file: %s"% (self.key, self.configFile)) - raise Exception("ERROR: config knob %s not found in file: %s"% (self.key, self.configFile)) + self.log.warning("Error getting key: %s from config file", key) + return default def str2bool(self,word): return word.lower() in ("true") def total_size(o, handlers={}, verbose=False): - """ Returns the approximate memory footprint an object and all of its contents. + """ Returns the approximate memory footprint an object and all of its contents. - Automatically finds the contents of the following builtin containers and - their subclasses: tuple, list, deque, dict, set and frozenset. - To search other containers, add handlers to iterate over their contents: + Automatically finds the contents of the following builtin containers and + their subclasses: tuple, list, deque, dict, set and frozenset. + To search other containers, add handlers to iterate over their contents: - handlers = {SomeContainerClass: iter, - OtherContainerClass: OtherContainerClass.get_elements} - - """ - dict_handler = lambda d: chain.from_iterable(d.items()) - all_handlers = {tuple: iter, - list: iter, - dict: dict_handler, - set: iter, - frozenset: iter, - } - #all_handlers.update(handlers) # user handlers take precedence - seen = set() # track which object id's have already been seen - default_size = sys.getsizeof(0) # estimate sizeof object without __sizeof__ + handlers = {SomeContainerClass: iter, + OtherContainerClass: OtherContainerClass.get_elements} + + """ + dict_handler = lambda d: chain.from_iterable(d.items()) + all_handlers = {tuple: iter, + list: iter, + dict: dict_handler, + set: iter, + frozenset: iter, + } + #all_handlers.update(handlers) # user handlers take precedence + seen = set() # track which object id's have already been seen + default_size = sys.getsizeof(0) # estimate sizeof object without __sizeof__ - def sizeof(o): - if id(o) in seen: # do not double count the same object - return 0 - seen.add(id(o)) - s = sys.getsizeof(o, default_size) + def sizeof(o): + if id(o) in seen: # do not double count the same object + return 0 + seen.add(id(o)) + s = sys.getsizeof(o, default_size) - for typ, handler in all_handlers.items(): - if isinstance(o, typ): - s += sum(map(sizeof, handler(o))) - break - return s + for typ, handler in all_handlers.items(): + if isinstance(o, typ): + s += sum(map(sizeof, handler(o))) + break + return s