Merge branch 'develop' with daisy-chain architecture

README.md

-# rabbitmq_agents
+# rabbitmq_agents
\ No newline at end of file
+
+This repo keeps different rabbitmq agents that help in account creation on OHPC system.
+
+It has 2 branches ```develop``` and ```production``` , that houses agents based on where they are launched
+
+## Using RCRMQ class
+
+- First, rename `rabbit_config.py.example` to `rabbit_config.py`
+
+- Modify config file, at least the `Password` needs to be your own passwod
+
+- In your code:
+
+```
+# import the class
+from rc_rmq import RCRMQ
+
+# instantiate an instance
+rc_rmq = RCRMQ({'exchange': 'RegUsr'})
+
+# publish a message to message queue
+rc_rmq.publish_msg({
+  'routing_key': 'your_key',
+  'msg': {
+    'type': 'warning',
+    'content': 'this is warning'
+  }
+})
+
+# to consume message from a queue
+# you have to first define callback function
+# with parameters: channel, method, properties, body
+def callback_function(ch, method, properties, body):
+  msg = json.loads(body)
+  print("get msg: {}".format(msg['username')
+
+  # this will stop the consumer
+  rc_rmq.stop_consumer()
+
+# start consume messagre from queue with callback function
+rc_rmq.start_consume({
+  'queue': 'queue_name',
+  'cb': callback_function
+})
+
+```

flask_producer.py

+#!/usr/bin/env python
+import pika
+import sys
+import socket
+import json
+import rabbit_config as rcfg
+
+if len(sys.argv) < 3:
+    sys.stderr.write("Usage: {} TAG USERNAME ".format(sys.argv[0]))
+    exit(1)
+
+node = sys.argv[1]
+user_name = sys.argv[2]
+
+message = {
+    "username": user_name,
+    "fullname": "Full Name",
+    "reason": "Reason1, Reason2."
+}
+
+hostname = socket.gethostname().split(".", 1)[0]
+connect_host = rcfg.Server if hostname != rcfg.Server else "localhost"
+
+# Set up credentials to connect to RabbitMQ server
+credentials = pika.PlainCredentials(rcfg.User, rcfg.Password)
+parameters = pika.ConnectionParameters(connect_host,
+                                   rcfg.Port,
+                                   rcfg.VHost,
+                                   credentials)
+
+# Establish connection to RabbitMQ server
+connection = pika.BlockingConnection(parameters)
+channel = connection.channel()
+channel.exchange_declare(exchange=rcfg.Exchange, exchange_type='direct')
+
+channel.basic_publish(exchange=rcfg.Exchange, routing_key=node, body=json.dumps(message))
+print(" [x] Sent {}: {}".format(node, json.dumps(message)))
+
+# creates a named queue
+result = channel.queue_declare(queue=user_name, exclusive=False, durable=True)
+
+# bind the queue with exchange
+channel.queue_bind(exchange=rcfg.Exchange, queue=user_name, routing_key=user_name)
+
+def work(ch, method, properties, body):
+    msg = json.loads(body)
+    print("Received message from {}: \n\t{}".format(method.routing_key, msg))
+    #queue_unbind(queue, exchange=None, routing_key=None, arguments=None, callback=None)
+    channel.queue_delete(method.routing_key)
+
+# ingest messages, and assume delivered via auto_ack
+channel.basic_consume(queue=sys.argv[2], on_message_callback=work, auto_ack=True)
+print("Subscribing to queue: {}".format(sys.argv[2]))
+
+# initiate message ingestion
+try:
+    channel.start_consuming()
+except KeyboardInterrupt:
+    print("Disconnecting from broker.")
+    channel.stop_consuming()
+connection.close()

ohpc_account_create.py

+#!/usr/bin/env python
+import pika # python client
+import sys
+import rabbit_config as rcfg
+import socket
+import subprocess
+import time
+import json
+from pwd import getpwnam
+
+hostname = socket.gethostname().split(".", 1)[0]
+connect_host = rcfg.Server if hostname != rcfg.Server else "localhost"
+queue_name = "ohpc_account_create"
+duration = 2
+
+# Set up credentials to connect to RabbitMQ server
+credentials = pika.PlainCredentials(rcfg.User, rcfg.Password)
+parameters = pika.ConnectionParameters(connect_host,
+                                   rcfg.Port,
+                                   rcfg.VHost,
+                                   credentials)
+
+# Establish connection to RabbitMQ server
+connection = pika.BlockingConnection(parameters)
+channel = connection.channel()
+
+print("connection established. Listening for messages:")
+
+# create exchange to pass messages
+channel.exchange_declare(exchange=rcfg.Exchange, exchange_type='direct')
+
+# creates a random name for the newly generated queue
+result = channel.queue_declare(queue=queue_name, exclusive=False)
+
+channel.queue_bind(exchange=rcfg.Exchange, queue=queue_name, routing_key=queue_name)
+
+def ohpc_account_create(ch, method, properties, body):
+    msg = json.loads(body)
+    print("Message received {}".format(msg))
+    username = msg['username']
+    try:
+        subprocess.call(["sudo", "useradd", username])
+        print("User {} has been added to {}".format(username, hostname))
+    except:
+        print("Failed to create user")
+
+    channel.basic_ack(delivery_tag=method.delivery_tag)
+    msg['uid'] = getpwnam(username).pw_uid
+    msg['gid'] = getpwnam(username).pw_gid
+    channel.basic_publish(exchange=rcfg.Exchange, routing_key='ood_account_create', body=json.dumps(msg))
+
+
+# ingest messages
+channel.basic_consume(queue=queue_name, on_message_callback=ohpc_account_create)
+
+# initiate message ingestion
+try:
+    channel.start_consuming()
+except KeyboardInterrupt:
+    print("Disconnecting from broker.")
+    channel.stop_consuming()
+
+connection.close()

ood_account_create.py

+#!/usr/bin/env python
+import pika # python client
+import sys
+import rabbit_config as rcfg
+import socket
+import subprocess
+import time
+import json
+
+hostname = socket.gethostname().split(".", 1)[0]
+connect_host = rcfg.Server if hostname != rcfg.Server else "localhost"
+queue_name = "ood_account_create"
+duration = 2
+
+# Set up credentials to connect to RabbitMQ server
+credentials = pika.PlainCredentials(rcfg.User, rcfg.Password)
+parameters = pika.ConnectionParameters(connect_host,
+                                   rcfg.Port,
+                                   rcfg.VHost,
+                                   credentials)
+
+# Establish connection to RabbitMQ server
+connection = pika.BlockingConnection(parameters)
+channel = connection.channel()
+
+print("connection established. Listening for messages:")
+
+# create exchange to pass messages
+channel.exchange_declare(exchange=rcfg.Exchange, exchange_type='direct')
+
+# creates a random name for the newly generated queue
+result = channel.queue_declare(queue=queue_name, exclusive=False)
+
+channel.queue_bind(exchange=rcfg.Exchange, queue=queue_name, routing_key=queue_name)
+
+def ood_account_create(ch, method, properties, body):
+    msg = json.loads(body)
+    print("Message received {}".format(msg))
+    username = msg['username']
+    user_uid = str(msg['uid'])
+    user_gid = str(msg['gid'])
+    try:
+        subprocess.call(["sudo", "groupadd", "-r", "-g", user_gid, username])
+        subprocess.call(["sudo", "useradd", "-u", user_uid, "-g", user_gid, username])
+        print("User {} has been added to {}".format(username, hostname))
+    except:
+        print("Failed to create user")
+
+    channel.basic_ack(delivery_tag=method.delivery_tag)
+
+    channel.basic_publish(exchange=rcfg.Exchange, routing_key='slurm_add_account', body=json.dumps(msg))
+
+
+# ingest messages
+channel.basic_consume(queue=queue_name, on_message_callback=ood_account_create)
+
+# initiate message ingestion
+try:
+    channel.start_consuming()
+except KeyboardInterrupt:
+    print("Disconnecting from broker.")
+    channel.stop_consuming()
+
+connection.close()

rabbit_config.py.example

+Exchange = 'RegUsr'
+User = 'reggie'
+Password = 'CHANGE_IT_TO_YOUR_OWN_PASSWORD'
+VHost = '/'
+Server = 'ohpc'
+Port = 5672

rc_rmq.py

+import json
+import pika
+import socket
+import rabbit_config as rcfg
+
+class RCRMQ(object):
+
+    USER = 'guest'
+    PASSWORD = 'guest'
+    HOST = 'localhost'
+    PORT = 5672
+    VHOST = '/'
+    EXCHANGE = ''
+    EXCHANGE_TYPE = 'direct'
+    QUEUE = None
+    DURABLE = True
+    ROUTING_KEY = None
+
+    def __init__(self, config=None):
+        if config:
+            if 'exchange' in config:
+                self.EXCHANGE = config['exchange']
+            if 'exchange_type' in config:
+                self.EXCHANGE_TYPE = config['exchange_type']
+
+        hostname = socket.gethostname().split(".", 1)[0]
+
+        self.HOST = rcfg.Server if hostname != rcfg.Server else "localhost"
+        self.USER = rcfg.User
+        self.PASSWORD = rcfg.Password
+        self.VHOST = rcfg.VHost
+        self.PORT = rcfg.Port
+        self._parameters = pika.ConnectionParameters(
+                self.HOST,
+                self.PORT,
+                self.VHOST,
+                pika.PlainCredentials(self.USER, self.PASSWORD))
+
+    def connect(self):
+        self._connection = pika.BlockingConnection(self._parameters)
+        self._channel = self._connection.channel()
+        self._channel.exchange_declare(
+                exchange=self.EXCHANGE, 
+                exchange_type=self.EXCHANGE_TYPE,
+                durable=True)
+
+        if self.QUEUE is not None:
+            self._channel.queue_declare(queue=self.QUEUE, durable=self.DURABLE)
+            self._channel.queue_bind(exchange=self.EXCHANGE,
+                    queue=self.QUEUE,
+                    routing_key=self.ROUTING_KEY)
+
+    def disconnect(self):
+        self._channel.close()
+        self._connection.close()
+
+    def delete_queue(self):
+        self._channel.queue_delete(self.QUEUE)
+
+    def publish_msg(self, obj):
+        if 'routing_key' in obj:
+            self.ROUTING_KEY = obj['routing_key']
+
+        self.connect()
+
+        self._channel.basic_publish(exchange=self.EXCHANGE,
+                routing_key=self.ROUTING_KEY,
+                body=json.dumps(obj['msg']))
+
+        self.disconnect()
+
+    def start_consume(self, obj):
+        if 'queue' in obj:
+            self.QUEUE = obj['queue']
+            self.ROUTING_KEY = obj['routing_key'] if 'routing_key' in obj else self.QUEUE
+        if 'durable' in obj:
+            self.DURABLE = obj['durable']
+
+        self.connect()
+
+        self._consumer_tag = self._channel.basic_consume(self.QUEUE,obj['cb'])
+        try:
+            self._channel.start_consuming()
+        except KeyboardInterrupt:
+            self._channel.stop_consuming()
+
+        self.disconnect()
+
+    def stop_consume(self):
+        self._channel.basic_cancel(self._consumer_tag)

rc_util.py

+from rc_rmq import RCRMQ
+import json
+
+rc_rmq = RCRMQ({'exchange': 'Register'})
+confirm_rmq = RCRMQ({'exchange': 'Confirm'})
+tasks = {'ohpc_account': False, 'ohpc_homedir': False, 'ood_account': False, 'slurm_account': False}
+
+def add_account(username, full='', reason=''):
+  rc_rmq.publish_msg({
+    'routing_key': 'ohpc_account',
+    'msg': {
+      "username": username,
+      "fullname": full,
+      "reason": reason
+    }
+  })
+
+def worker(ch, method, properties, body):
+    msg = json.loads(body)
+    task = msg['task']
+    print("get msg: {}".format(task))
+    tasks[task] = msg['success']
+
+    # Check if all tasks are done
+    done = True
+    for key, status in tasks.items():
+        if not status:
+            print("{} is not done yet.".format(key))
+            done = False 
+    if done:
+        confirm_rmq.stop_consume()
+        confirm_rmq.delete_queue()
+  
+def consume(username, callback, debug=False):
+    if debug:
+        sleep(5)
+    else:
+        confirm_rmq.start_consume({
+            'queue': username,
+            'cb': callback
+        })
+  
+    return { 'success' : True }

requirements.txt

+pika==1.1.0

slurm_agent.py

+#!/usr/bin/env python
+import pika # python client
+import sys
+import rabbit_config as rcfg
+import socket
+import subprocess
+import time
+import json
+
+hostname = socket.gethostname().split(".", 1)[0]
+connect_host = rcfg.Server if hostname != rcfg.Server else "localhost"
+queue_name = "slurm_add_account"
+duration = 2
+
+# Set up credentials to connect to RabbitMQ server
+credentials = pika.PlainCredentials(rcfg.User, rcfg.Password)
+parameters = pika.ConnectionParameters(connect_host,
+                                   rcfg.Port,
+                                   rcfg.VHost,
+                                   credentials)
+
+# Establish connection to RabbitMQ server
+connection = pika.BlockingConnection(parameters)
+channel = connection.channel()
+
+print("connection established. Listening for messages:")
+
+# create exchange to pass messages
+channel.exchange_declare(exchange=rcfg.Exchange, exchange_type='direct')
+
+# creates a random name for the newly generated queue
+result = channel.queue_declare(queue=queue_name, exclusive=False)
+
+channel.queue_bind(exchange=rcfg.Exchange, queue=queue_name, routing_key=queue_name)
+
+def slurm_account_create(ch, method, properties, body):
+    msg = json.loads(body)
+    print("Message received {}".format(msg))
+    username = msg['username']
+    try:
+        subprocess.call(["sudo", "sacctmgr", "add", "account", username, "-i",  "Descripition: Add user"])
+        subprocess.call(["sudo", "sacctmgr", "add", "user", username, "account="+username, "-i"])
+        print("SLURM account for user {} has been added".format(username))
+    except:
+        print("Failed to create user")
+    
+    channel.basic_ack(delivery_tag=method.delivery_tag)
+        
+    channel.basic_publish(exchange=rcfg.Exchange, routing_key=username, body=json.dumps(msg))
+    
+
+# ingest messages
+channel.basic_consume(queue=queue_name, on_message_callback=slurm_account_create)
+
+# initiate message ingestion
+try:
+    channel.start_consuming()
+except KeyboardInterrupt:
+    print("Disconnecting from broker.")
+    channel.stop_consuming()
+
+connection.close()
+
+        
+