Merge branch '35-add-flower-for-celery-monitoring' into 'develop'

Resolve "Add flower service"

Closes #35

See merge request verbose-equals-true/django-postgres-vue-gitlab-ecs!83

Author: briancaffey

.env.template

@@ -18,8 +18,8 @@ GOOGLE_OAUTH2_SECRET=abcde-fghij
 SOCIAL_AUTH_FACEBOOK_KEY = ''
 SOCIAL_AUTH_FACEBOOK_SECRET = ''
 
-CELERY_BROKER_URL=redis://redis:6379
-CELERY_RESULT_BACKEND=redis://redis:6379
+CELERY_BROKER_URL=redis://redis:6379/0
+CELERY_RESULT_BACKEND=redis://redis:6379/1
 
 DJANGO_EMAIL_HOST=mailhog
 DJANGO_EMAIL_PORT=1025

awscdk/awscdk/backend.py

@@ -57,9 +57,9 @@ def __init__(self, scope: core.Construct, id: str, **kwargs,) -> None:
             "BackendTarget",
             port=80,
             targets=[self.backend_service],
-            priority=1,
+            priority=2,
             path_patterns=["*"],
             health_check=elbv2.HealthCheck(
-                healthy_http_codes="200-299", path="/api/hello-world",
+                healthy_http_codes="200-299", path="/api/health-check/",
             ),
         )

awscdk/awscdk/cdk_app_root.py

@@ -13,6 +13,7 @@
 from ecs import EcsStack
 from env_vars import Variables
 from static_site_bucket import StaticSiteStack
+from flower import FlowerServiceStack
 
 from backend import BackendServiceStack
 from backend_tasks import BackendTasksStack
@@ -55,6 +56,11 @@ def __init__(
         self.static_site_stack = StaticSiteStack(self, "StaticSiteStack")
         self.static_site_bucket = self.static_site_stack.static_site_bucket
 
+        self.backend_assets = BackendAssetsStack(self, "BackendAssetsStack")
+        self.backend_assets_bucket = self.backend_assets.assets_bucket
+
+        self.cloudfront = CloudFrontStack(self, "CloudFrontStack")
+
         if os.path.isdir("./quasar/dist/pwa"):
             s3_deployment.BucketDeployment(
                 self,
@@ -64,11 +70,6 @@ def __init__(
                 distribution=self.cloudfront.distribution,
             )
 
-        self.backend_assets = BackendAssetsStack(self, "BackendAssetsStack")
-        self.backend_assets_bucket = self.backend_assets.assets_bucket
-
-        self.cloudfront = CloudFrontStack(self, "CloudFrontStack")
-
         self.ecs = EcsStack(self, "EcsStack")
 
         self.rds = RdsStack(self, "RdsStack")
@@ -93,6 +94,7 @@ def __init__(
         )
 
         self.backend_service = BackendServiceStack(self, "BackendServiceStack")
+        self.flower_service = FlowerServiceStack(self, "FlowerServiceStack")
 
         self.celery_default_service = CeleryDefaultServiceStack(
             self, "CeleryDefaultServiceStack"

awscdk/awscdk/celery_default.py

@@ -4,7 +4,6 @@
     aws_ecs as ecs,
     aws_logs as logs,
     aws_cloudformation as cloudformation,
-    aws_sqs as sqs,
 )
 
 

awscdk/awscdk/cloudfront.py

@@ -65,10 +65,11 @@ def __init__(self, scope: core.Construct, id: str, **kwargs,) -> None:
                     behaviors=[
                         cloudfront.Behavior(
                             allowed_methods=ALL_METHODS,
+                            forwarded_values={"query_string": True},
                             path_pattern=path_pattern,
-                            max_ttl=core.Duration.seconds(0),
                             min_ttl=core.Duration.seconds(0),
-                            forwarded_values={"query_string": True},
+                            default_ttl=core.Duration.seconds(0),
+                            max_ttl=core.Duration.seconds(0),
                         )
                         for path_pattern in ["/static/*", "/media/*"]
                     ],

awscdk/awscdk/flower.py

@@ -0,0 +1,67 @@
+import os
+
+from aws_cdk import (
+    core,
+    aws_ec2 as ec2,
+    aws_ecs as ecs,
+    aws_logs as logs,
+    aws_cloudformation as cloudformation,
+    aws_elasticloadbalancingv2 as elbv2,
+)
+
+
+class FlowerServiceStack(cloudformation.NestedStack):
+    def __init__(self, scope: core.Construct, id: str, **kwargs,) -> None:
+        super().__init__(
+            scope, id, **kwargs,
+        )
+
+        self.flower_task = ecs.FargateTaskDefinition(self, "FlowerTask")
+
+        FLOWER_PASSWORD = os.environ.get("FLOWER_PASSWORD", "flowerpassword")
+        REDIS_SERVICE_HOST = (
+            scope.elasticache.elasticache.attr_redis_endpoint_address
+        )
+        CELERY_BROKER_URL = f"redis://{REDIS_SERVICE_HOST}:6379/0"
+        self.flower_task.add_container(
+            "FlowerContainer",
+            image=ecs.ContainerImage.from_registry("mher/flower"),
+            logging=ecs.LogDrivers.aws_logs(
+                stream_prefix="FlowerContainer",
+                log_retention=logs.RetentionDays.ONE_DAY,
+            ),
+            command=[
+                "--url_prefix=flower",
+                f"--broker={CELERY_BROKER_URL}",
+                f"--basic_auth=flower:{FLOWER_PASSWORD}",
+            ],
+        )
+
+        port_mapping = ecs.PortMapping(
+            container_port=5555, protocol=ecs.Protocol.TCP
+        )
+        self.flower_task.default_container.add_port_mappings(port_mapping)
+
+        self.flower_service = ecs.FargateService(
+            self,
+            "FlowerService",
+            task_definition=self.flower_task,
+            assign_public_ip=True,
+            cluster=scope.ecs.cluster,
+            security_group=ec2.SecurityGroup.from_security_group_id(
+                self,
+                "FlowerServiceSecurityGroup",
+                security_group_id=scope.vpc.vpc_default_security_group,
+            ),
+        )
+
+        scope.https_listener.add_targets(
+            "FlowerTarget",
+            port=80,
+            targets=[self.flower_service],
+            priority=1,
+            path_patterns=["/flower/*", "flower/*"],
+            health_check=elbv2.HealthCheck(
+                healthy_http_codes="200-401", path="/flower/"
+            ),
+        )

backend/apps/core/management/commands/__init__.py

@@ -6,21 +6,32 @@
 https://avilpage.com/2017/05/how-to-auto-reload-celery-workers-in-development.html
 """
 
+import os
 import shlex
 import subprocess
 
 from django.core.management.base import BaseCommand
 from django.utils import autoreload
 
 
-def restart_celery():
-    cmd = "pkill -9 celery"
+def restart_celery(queue=None, concurrency=None):
+    cmd = 'pkill -9 celery'
     subprocess.call(shlex.split(cmd))
-    cmd = "celery worker --app=backend.celery_app:app --loglevel=info --concurrency=2 --max-memory-per-child=150000"  # noqa
+    cmd = f"celery worker --app=backend.celery_app:app --loglevel=info -Q {queue} -n worker-{queue}@%h --concurrency={os.environ.get('CONCURRENT_WORKERS', 2)} --max-memory-per-child=150000"  # noqa
     subprocess.call(shlex.split(cmd))
 
 
 class Command(BaseCommand):
+    def add_arguments(self, parser):
+        parser.add_argument(
+            '-q', '--queue', nargs=1, default='celery', type=str
+        )
+        parser.add_argument('-c', '--concurrency', type=str)
+
     def handle(self, *args, **options):
-        print("Starting celery worker with autoreload...")
-        autoreload.run_with_reloader(restart_celery)
+        queue = options['queue'][0]
+        concurrency = options['concurrency'] or 1
+        print('Starting celery worker with autoreload...')
+        autoreload.run_with_reloader(
+            restart_celery, queue=queue, concurrency=concurrency
+        )

backend/apps/core/management/commands/watch_celery.py

@@ -15,7 +15,6 @@ class BaseTask(celery.Task):
 @task(bind=True, base=BaseTask)
 def debug_task(self):
     time.sleep(10)
-    print("Task is done")
 
 
 @periodic_task(
@@ -30,9 +29,15 @@ def debug_periodic_task():
 @task(bind=True, base=BaseTask)
 def send_test_email_task(self):
     send_mail(
-        "Subject here",
-        "Here is the message.",
+        "Email subject",
+        "Email message.",
         "from@example.com",
         ["to@example.com"],
         fail_silently=False,
     )
+
+
+@task(bind=True, base=BaseTask)
+def sleep_task(self, seconds):
+    time.sleep(int(seconds))
+    return f"Slept {seconds} seconds"

backend/apps/core/tasks.py

@@ -3,9 +3,8 @@
 from . import views
 
 urlpatterns = [
-    path("", views.home, name="home"),
-    path("hello-world", views.hello_world, name="hello-world"),
-    path("debug-task/", views.debug_task_view, name="debug-task",),
+    path("health-check/", views.health_check, name="health-check"),
+    path("celery/sleep-task/", views.sleep_task_view, name="sleep-task"),
     path(
         "debug/send-test-email/",
         views.send_test_email,