Commit 7965cc61 authored by lukas.burgey's avatar lukas.burgey
Browse files

Refactor the broker module

parent 60923cf0
from logging import getLogger
from json import dumps
from django.db import models
from django.core.cache import cache
import pika
from pika import BlockingConnection, ConnectionParameters, BasicProperties
from pika.credentials import PlainCredentials
LOGGER = getLogger(__name__)
def publish_to_user(user, obj):
from . import serializers
RabbitMQInstance.load().publish_to_user(
user,
dumps(serializers.UpdateSerializer(obj).data),
)
def publish_deployment_state(deployment_state):
from .serializers.clients import DeploymentStateSerializer
RabbitMQInstance.load().publish_deployment_state(
deployment_state,
dumps(DeploymentStateSerializer(deployment_state).data),
)
# singleton for simple configs
# https://steelkiwi.com/blog/practical-application-singleton-design-pattern/
class SingletonModel(models.Model):
class Meta:
abstract = True
def set_cache(self):
cache.set(self.__class__.__name__, self)
# pylint: disable=invalid-name, arguments-differ
def save(self, *args, **kwargs):
self.pk = 1
super(SingletonModel, self).save(*args, **kwargs)
self.set_cache()
@classmethod
def load(cls):
if cache.get(cls.__name__) is None:
obj, created = cls.objects.get_or_create(pk=1)
if not created:
obj.set_cache()
return cache.get(cls.__name__)
# The rabbitmq instance located on this host (localhost)
class RabbitMQInstance(SingletonModel):
class RabbitMQInstance:
@property
def host(self):
......@@ -84,7 +40,7 @@ class RabbitMQInstance(SingletonModel):
def _params(self):
# we set NO port here, we use the default (probably 5672)
# this requires the
return pika.ConnectionParameters(
return ConnectionParameters(
host=self.host,
virtual_host=self.vhost,
credentials=PlainCredentials(
......@@ -102,7 +58,7 @@ class RabbitMQInstance(SingletonModel):
def _open_connection(self):
try:
# start a new connection
return pika.BlockingConnection(self._params)
return BlockingConnection(self._params)
except Exception as e:
LOGGER.exception('RabbitMQ connection error: %s', e)
raise e
......@@ -118,7 +74,7 @@ class RabbitMQInstance(SingletonModel):
exchange=exchange,
routing_key=routing_key,
body=body,
properties=pika.BasicProperties(
properties=BasicProperties(
delivery_mode=1,
),
)
......@@ -148,8 +104,3 @@ class RabbitMQInstance(SingletonModel):
str(user.id),
msg,
)
# we keep this, as removing this breaks migrations
def exchanges_default():
return []
......@@ -8,9 +8,9 @@ import django.db.models.deletion
import django.utils.timezone
import django_mysql.models
import feudal.backend.models.auth
import feudal.backend.models.brokers
import feudal.backend.models.deployments
import feudal.backend.models.users
import feudal.backend.brokers
class Migration(migrations.Migration):
......@@ -110,7 +110,7 @@ class Migration(migrations.Migration):
('id', models.AutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
('host', models.CharField(default='localhost', max_length=150)),
('vhost', models.CharField(default='/', max_length=150)),
('exchanges', django_mysql.models.JSONField(blank=True, default=feudal.backend.models.brokers.exchanges_default, null=True)),
('exchanges', django_mysql.models.JSONField(blank=True, default=None, null=True)),
('port', models.IntegerField(default=15672)),
('username', models.CharField(default='guest', max_length=150)),
('password', models.CharField(default='guest', max_length=150)),
......
from json import dumps
from logging import getLogger
from django.conf import settings
......@@ -11,9 +12,9 @@ from django_mysql.models import JSONField
from polymorphic.models import PolymorphicModel
from feudal.backend.models import Site, Service
from feudal.backend.models.brokers import publish_to_user, publish_deployment_state
from feudal.backend.models.users import User, SSHPublicKey
from feudal.backend.models.auth.vos import VO
from feudal.backend.brokers import RabbitMQInstance
LOGGER = getLogger(__name__)
......@@ -172,11 +173,10 @@ class Deployment(PolymorphicModel):
if settings.DEBUG_PUBLISHING:
LOGGER.debug(self.msg('publish_to_user: {}'.format(self.state_target)))
publish_to_user(
from feudal.backend.models.serializers import UpdateSerializer
RabbitMQInstance().publish_to_user(
self.user,
{
'deployment': self,
},
dumps(UpdateSerializer({'deployment': self}).data),
)
def publish(self):
......@@ -592,11 +592,10 @@ class DeploymentState(models.Model):
if settings.DEBUG_PUBLISHING:
LOGGER.debug(self.msg('publish_to_user'))
publish_to_user(
from feudal.backend.models.serializers import UpdateSerializer
RabbitMQInstance().publish_to_user(
self.user,
{
'deployment_state': self,
},
dumps(UpdateSerializer({'deployment_state': self}).data),
)
def publish_to_client(self):
......@@ -618,7 +617,11 @@ class DeploymentState(models.Model):
if settings.DEBUG_PUBLISHING:
LOGGER.debug(self.msg('publish_to_client'))
publish_deployment_state(self)
from feudal.backend.models.serializers.clients import DeploymentStateSerializer
RabbitMQInstance().publish_deployment_state(
self,
dumps(DeploymentStateSerializer(self).data),
)
def msg(self, msg):
return ' {} - {}'.format(self, msg)
......
......@@ -6,10 +6,10 @@ from rest_framework.serializers import ModelSerializer, DictField, ListField
from rest_polymorphic.serializers import PolymorphicSerializer
from feudal.backend.models import Service
from feudal.backend.models.brokers import RabbitMQInstance
from feudal.backend.models.users import User, SSHPublicKey
from feudal.backend.models.deployments import Deployment, VODeployment, ServiceDeployment, DeploymentState, CredentialState
from feudal.backend.models.auth.serializers.clients import VOSerializer
from feudal.backend.brokers import RabbitMQInstance
class ServiceSerializer(ModelSerializer):
......
......@@ -8,6 +8,7 @@ from feudal.backend.models.deployments import (
Deployment, get_deployment,
DEPLOYED, NOT_DEPLOYED, DEPLOYMENT_PENDING, REMOVAL_PENDING,
)
import feudal.backend.brokers
LOGGER = logging.getLogger(__name__)
......
......@@ -9,8 +9,8 @@ from django.contrib.sessions.models import Session
from feudal.backend.models import Site, Service
from feudal.backend.models.users import User
from feudal.backend.models.brokers import RabbitMQInstance
from feudal.backend.models.auth.vos import Group, Entitlement
from feudal.backend.brokers import RabbitMQInstance
LOGGER = logging.getLogger(__name__)
......@@ -19,7 +19,7 @@ DENY = HttpResponse('deny')
def _valid_vhost(request):
if request.POST.get('vhost') == RabbitMQInstance.load().vhost:
if request.POST.get('vhost') == RabbitMQInstance().vhost:
return True
LOGGER.error('illegal vhost requested')
return False
......@@ -139,7 +139,7 @@ def _resource_authorized_apiclient(request):
and name.startswith('amq.gen-')
) or (
resource == 'exchange'
and name in RabbitMQInstance.load().exchanges
and name in RabbitMQInstance().exchanges
and 'write' not in permission
)
......
from django.test import TestCase, override_settings
from django.test.client import Client
from feudal.backend.models.auth.vos import Group, Entitlement
from feudal.backend.models.brokers import RabbitMQInstance
from feudal.backend.models.users import User
from feudal.backend.models.auth import OIDCConfig
from feudal.backend.models import Site, Service
from feudal.backend.brokers import RabbitMQInstance
@override_settings(DEBUG=True, DEBUG_AUTH=True)
......@@ -45,7 +44,7 @@ class AuthorizationTestCase(TestCase):
cls.service.vos.add(cls.group)
cls.service.vos.add(cls.entitlement)
cls.broker = RabbitMQInstance.load()
cls.broker = RabbitMQInstance()
def setUp(self):
self.client = Client()
......
......@@ -12,8 +12,8 @@ from rest_framework.response import Response
from feudal.backend.models import Site, Service
from feudal.backend.models.auth.vos import VO, Group, Entitlement
from feudal.backend.models.deployments import NOT_DEPLOYED
from feudal.backend.models.brokers import RabbitMQInstance
from feudal.backend.models.serializers import clients
from feudal.backend.brokers import RabbitMQInstance
from feudal.backend.permissions import DownstreamOnly
LOGGER = logging.getLogger(__name__)
......@@ -182,7 +182,7 @@ class ConfigurationView(views.APIView):
self.apply_vos_to_services(request)
# initialize the broker, just in case
broker = RabbitMQInstance.load()
broker = RabbitMQInstance()
broker.initialize()
return Response({
......
Markdown is supported
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment