JIRA_URL=sample_url
JIRA_USER_NAME=sample_jira_user
-JIRA_USER_PASSWORD=sample_jira_pass
\ No newline at end of file
+JIRA_USER_PASSWORD=sample_jira_pass
+
+# Rabbitmq
+RABBITMQ_USER=opnfv
+RABBITMQ_PASSWORD=opnfvopnfv
--- /dev/null
+##############################################################################
+# Copyright (c) 2016 Max Breitenfeldt and others.
+#
+# All rights reserved. This program and the accompanying materials
+# are made available under the terms of the Apache License, Version 2.0
+# which accompanies this distribution, and is available at
+# http://www.apache.org/licenses/LICENSE-2.0
+##############################################################################
+
+import jsonpickle
+import pika
+
+
+class Message(object):
+ def __init__(self, type, topic, content):
+ self.type = type
+ self.topic = topic
+ self.content = content
+
+
+class Notification(object):
+ """
+ This class can be used by the dashboard and the labs to exchange notifications about booking
+ events and pod status. It utilizes rabbitmq to communicate.
+
+ Notifications are associated to an event and to a topic.
+ Events are:
+ [ 'booking_start', 'booking_end']
+ The topic is usually a POD name, ie:
+ 'Intel POD 2'
+ """
+
+ def __init__(self, dashboard_url, user=None, password=None, verbose=False):
+ self.rabbitmq_broker = dashboard_url
+ self.verbose = verbose
+ if user is None and password is None:
+ self._connection = pika.BlockingConnection(pika.ConnectionParameters(
+ host=self.rabbitmq_broker))
+ else:
+ self.credentials = pika.PlainCredentials(user, password)
+ self._connection = pika.BlockingConnection(pika.ConnectionParameters(
+ credentials=self.credentials,
+ host=self.rabbitmq_broker))
+ self._registry = {}
+ self._channel = self._connection.channel()
+ self._channel.exchange_declare(exchange='notifications', type='topic')
+ self._result = self._channel.queue_declare(exclusive=True, durable=True)
+ self._queue_name = self._result.method.queue
+
+ def __enter__(self):
+ return self
+
+ def __exit__(self, exc_type, exc_val, exc_tb):
+ self._connection.close()
+
+ def register(self, function, topic, type='all'):
+ """
+ Registers a function to be called for the specified event.
+ :param function: the function to register
+ :param event: the event type
+ :param regex: a regex to specify for wich topics the function will be called. Some
+ possible Expressions can be:
+ 'Intel POD 2' : Intel POD 2
+ """
+
+ if topic not in self._registry:
+ self._registry[topic] = [(function, type)]
+ else:
+ self._registry[topic].append((function, type))
+
+ def receive(self):
+ """
+ Start receiving notifications. This is a blocking operation, if a notification is received,
+ the registered functions will be called.
+ """
+ if self.verbose:
+ print('Start receiving Notifications. Keys: ', self._registry.keys())
+ self._receive_message(self._registry.keys())
+
+ def send(self, message):
+ """
+ Send an event notification.
+ :param event: the event type
+ :param topic: the pod name
+ :param content: a JSON-serializable dictionary
+ """
+ self._send_message(message)
+
+ def _send_message(self, message):
+ routing_key = message.topic
+ message_json = jsonpickle.encode(message)
+ self._channel.basic_publish(exchange='notifications',
+ routing_key=routing_key,
+ body=message_json,
+ properties=pika.BasicProperties(
+ content_type='application/json',
+ delivery_mode=2, # make message persistent
+ ))
+ if self.verbose:
+ print(" [x] Sent %r:%r" % (routing_key, message_json))
+
+ def _receive_message(self, binding_keys):
+ for key in binding_keys:
+ self._channel.queue_bind(exchange='notifications',
+ queue=self._queue_name,
+ routing_key=key)
+ self._channel.basic_consume(self._message_callback,
+ queue=self._queue_name)
+ self._channel.start_consuming()
+
+ def _message_callback(self, ch, method, properties, body):
+ if self.verbose:
+ print(" [x] Got %r:%r" % (method.routing_key, body))
+ if method.routing_key not in self._registry:
+ return
+ for func, type in self._registry[method.routing_key]:
+ message = jsonpickle.decode(body.decode())
+ if message.type == type:
+ func(message)
+ ch.basic_ack(delivery_tag=method.delivery_tag)
- postgres
env_file: config.env
volumes:
- - ./src:/src
+ - ./:/pharos_dashboard
- /var/lib/pharos_dashboard/static:/static
- /var/lib/pharos_dashboard/media:/media
expose:
rabbitmq:
restart: always
- image: rabbitmq:latest
+ build: ./rabbitmq/
container_name: rm01
- expose:
- - "5672"
-
+ env_file: config.env
+ ports:
+ - "5672:5672"
+
worker:
restart: always
build: ./worker/
- command: bash -c "celery -A pharos_dashboard worker -l info -B --schedule=~/celerybeat-schedule""
+ command: bash -c "celery -A pharos_dashboard worker -l info -B --schedule=~/celerybeat-schedule"
env_file: config.env
links:
- postgres
- rabbitmq
volumes:
- - ./src:/src
-
\ No newline at end of file
+ - ./:/pharos_dashboard
+
--- /dev/null
+FROM rabbitmq
+
+ADD init.sh /init.sh
+CMD ["/init.sh"]
\ No newline at end of file
--- /dev/null
+#!/bin/sh
+
+# Create Rabbitmq user
+( sleep 20 ; \
+rabbitmqctl add_user $RABBITMQ_USER $RABBITMQ_PASSWORD 2>/dev/null ; \
+rabbitmqctl set_user_tags $RABBITMQ_USER administrator ; \
+rabbitmqctl set_permissions -p / $RABBITMQ_USER ".*" ".*" ".*" ; \
+echo "*** User '$RABBITMQ_USER' with password '$RABBITMQ_PASSWORD' completed. ***") &
+
+rabbitmq-server $@
--- /dev/null
+##############################################################################
+# Copyright (c) 2016 Max Breitenfeldt and others.
+#
+# All rights reserved. This program and the accompanying materials
+# are made available under the terms of the Apache License, Version 2.0
+# which accompanies this distribution, and is available at
+# http://www.apache.org/licenses/LICENSE-2.0
+##############################################################################
\ No newline at end of file
# -*- coding: utf-8 -*-
-# Generated by Django 1.10 on 2016-09-23 11:36
+# Generated by Django 1.10 on 2016-11-03 13:33
from __future__ import unicode_literals
from django.db import migrations, models
def get_content(self):
return {
- 'start': self.booking.start.isoformat(),
- 'end': self.booking.end.isoformat(),
+ 'resource_id': self.booking.resource.id,
+ 'booking_id': self.booking.id,
'user': self.booking.user.username,
- 'purpose': self.booking.purpose
+ 'user_id': self.booking.user.id,
}
def save(self, *args, **kwargs):
notifications = self.booking.bookingnotification_set.filter(type=self.type).exclude(
id=self.id)
- if notifications.count() > 0:
- raise ValueError('Doubled Notification')
- return super(BookingNotification, self).save(*args, **kwargs)
\ No newline at end of file
+ #if notifications.count() > 0:
+ # raise ValueError('Doubled Notification')
+ return super(BookingNotification, self).save(*args, **kwargs)
##############################################################################
+import os
+import sys
from datetime import timedelta
from celery import shared_task
from django.utils import timezone
from notification.models import BookingNotification
-from notification_framework.notification import Notification
+
+# this adds the top level directory to the python path, this is needed so that we can access the
+# notification library
+sys.path.append(os.path.join(os.path.dirname(__file__), '..', '..'))
+
+from dashboard_notification.notification import Notification, Message
@shared_task
def send_booking_notifications():
- messaging = Notification(dashboard_url=settings.RABBITMQ_URL)
-
- now = timezone.now()
- notifications = BookingNotification.objects.filter(submitted=False,
- submit_time__gt=now,
- submit_time__lt=now + timedelta(minutes=5))
- for notification in notifications:
- messaging.send(notification.type, notification.booking.resource.name,
- notification.get_content())
- notification.submitted = True
- notification.save()
+ with Notification(dashboard_url=settings.RABBITMQ_URL) as messaging:
+ now = timezone.now()
+ notifications = BookingNotification.objects.filter(submitted=False,
+ submit_time__gt=now - timedelta(minutes=1),
+ submit_time__lt=now + timedelta(minutes=5))
+ for notification in notifications:
+ message = Message(type=notification.type, topic=notification.booking.resource.name,
+ content=notification.get_content())
+ messaging.send(message)
+ notification.submitted = True
+ notification.save()
+
+@shared_task
+def notification_debug():
+ with Notification(dashboard_url=settings.RABBITMQ_URL) as messaging:
+ notifications = BookingNotification.objects.all()
+ for notification in notifications:
+ message = Message(type=notification.type, topic=notification.booking.resource.name,
+ content=notification.get_content())
+ messaging.send(message)
+++ /dev/null
-##############################################################################
-# Copyright (c) 2016 Max Breitenfeldt and others.
-#
-# All rights reserved. This program and the accompanying materials
-# are made available under the terms of the Apache License, Version 2.0
-# which accompanies this distribution, and is available at
-# http://www.apache.org/licenses/LICENSE-2.0
-##############################################################################
-
-
-import json
-import re
-
-import pika
-
-
-class Notification(object):
- """
- This class can be used by the dashboard and the labs to exchange notifications about booking
- events and pod status. It utilizes rabbitmq to communicate.
-
- Notifications are associated to an event and to a topic.
- Events are:
- [ 'booking_start', 'booking_stop', 'pod_status' ]
- The topic is usually a POD name, ie:
- 'Intel POD 2'
- """
-
- def __init__(self, dashboard_url, verbose=False):
- self.rabbitmq_broker = dashboard_url
- self.verbose = verbose
- self._registry = {}
-
- self.connection = pika.BlockingConnection(pika.ConnectionParameters(
- host=self.rabbitmq_broker))
- self.channel = self.connection.channel()
-
- self.channel.exchange_declare(exchange='notifications', type='topic')
-
- self.result = self.channel.queue_declare(exclusive=True)
- self.queue_name = self.result.method.queue
-
- def register(self, function, event, regex):
- """
- Registers a function to be called for the specified event.
- :param function: the function to register
- :param event: the event type
- :param regex: a regex to specify for wich topics the function will be called. Some
- possible Expressions can be:
- 'Intel POD 2' : Intel POD 2
- 'Intel POD .*' : All Intel Pods
- '.*' : All Topics
- """
-
- if event not in self._registry:
- self._registry[event] = [(function, regex)]
- else:
- self._registry[event].append((function, regex))
-
- def receive(self):
- """
- Start receiving notifications. This is a blocking operation, if a notification is received,
- the registered functions will be called.
- """
- if self.verbose:
- print('Start receiving Notifications. Keys: ', self._registry.keys())
- self._receive_message(self._registry.keys())
-
- def send(self, event, topic, content):
- """
- Send an event notification.
- :param event: the event type
- :param topic: the pod name
- :param content: a JSON-serializable dictionary
- """
- message = {
- 'event': event,
- 'topic': topic,
- 'content': content
- }
- self._send_message(message)
-
- def _send_message(self, event):
- routing_key = event['type']
- message = json.dumps(event)
- self.channel.basic_publish(exchange='notifications',
- routing_key=routing_key,
- body=message,
- properties=pika.BasicProperties(
- content_type='application/json'
- ))
- if self.verbose:
- print(" [x] Sent %r:%r" % (routing_key, message))
-
- def _receive_message(self, binding_keys):
- for key in binding_keys:
- self.channel.queue_bind(exchange='notifications',
- queue=self.queue_name,
- routing_key=key)
- self.channel.basic_consume(self._message_callback,
- queue=self.queue_name,
- no_ack=True)
- self.channel.start_consuming()
-
- def _message_callback(self, ch, method, properties, body):
- if self.verbose:
- print(" [x] Got %r:%r" % (method.routing_key, body))
- if method.routing_key not in self._registry:
- return
- for func, regex in self._registry[method.routing_key]:
- message = json.loads(body.decode())
- match = re.match(regex, message['topic'])
- if match:
- func(body)
RUN mkdir /config
ADD ./requirements.txt /config/
RUN pip install -r /config/requirements.txt
-RUN mkdir /src;
-WORKDIR /src
+RUN mkdir -p /pharos_dashboard/src
+WORKDIR /pharos_dashboard/src
djangorestframework==3.4.6
gunicorn==19.6.0
jira==1.0.7
+jsonpickle==0.9.3
oauth2==1.9.0.post1
oauthlib==1.1.2
pika==0.10.0
RUN pip install -r /config/requirements.txt
RUN useradd -ms /bin/bash celery
USER celery
-WORKDIR /src
+WORKDIR /pharos_dashboard/src
djangorestframework==3.4.6
gunicorn==19.6.0
jira==1.0.7
+jsonpickle==0.9.3
oauth2==1.9.0.post1
oauthlib==1.1.2
pika==0.10.0
Code Review / pharos.git / commitdiff