Cluster runs for backups (#27)

* selects for cluster

* cluster runs

* fixes

* jobs are scheduling, but no connection

* works

docker-compose.dev.yml

@@ -86,6 +86,8 @@ services:
       - zookeeper
     ports:
       - "9000:9000"
+    volumes:
+      - ./docker/clickhouse-server/config.d:/etc/clickhouse-server/config.d
 
   zookeeper:
     image: zookeeper:3.7.0

docker/clickhouse-server/config.d/config.xml

@@ -0,0 +1,23 @@
+<clickhouse>
+    <!-- Listen wildcard address to allow accepting connections from other containers and host
+    network. -->
+    <listen_host>::</listen_host>
+    <listen_host>0.0.0.0</listen_host>
+    <listen_try>1</listen_try>
+
+    <!--
+       <logger>
+           <console>1</console>
+       </logger>
+       -->
+    <remote_servers>
+        <housewatch>
+            <shard>
+                <replica>
+                    <host>clickhouse</host>
+                    <port>9000</port>
+                </replica>
+            </shard>
+        </housewatch>
+    </remote_servers>
+</clickhouse>

frontend/src/pages/Backups/ScheduledBackups.tsx

@@ -1,13 +1,15 @@
 import React, { useEffect, useState } from 'react'
 import { usePollingEffect } from '../../utils/usePollingEffect'
 import { ColumnType } from 'antd/es/table'
-import { Switch, Table, Button, Form, Input, Modal, Tag, Col, Progress, Row, Tooltip, notification } from 'antd'
+import { Switch, Select, Table, Button, Form, Input, Modal, Tag, Col, Progress, Row, Tooltip, notification } from 'antd'
+import { Clusters } from '../Clusters/Clusters'
 
 interface ScheduleRow {
     id: string
     created_at: string
     enabled: boolean
     last_run_time: string
+    cluster: string
     schedule: string
     table: string
     database: string
@@ -21,6 +23,7 @@ interface Backups {
 }
 
 type FieldType = {
+    cluster?: string
     schedule?: string
     database?: string
     table?: string
@@ -35,6 +38,9 @@ export default function ScheduledBackups() {
     const [loadingBackups, setLoadingBackups] = useState(false)
     const [open, setOpen] = useState(false)
     const [confirmLoading, setConfirmLoading] = useState(false)
+    const [clusters, setClusters] = useState<Clusters>({
+        clusters: [],
+    })
 
     const [form] = Form.useForm() // Hook to get form API
 
@@ -79,6 +85,15 @@ export default function ScheduledBackups() {
         } catch (err) {
             notification.error({ message: 'Failed to load data' })
         }
+
+        try {
+            const res = await fetch('/api/clusters')
+            const resJson = await res.json()
+            const clusters = { clusters: resJson }
+            setClusters(clusters)
+        } catch (err) {
+            notification.error({ message: 'Failed to load data' })
+        }
     }
 
     useEffect(() => {
@@ -110,7 +125,7 @@ export default function ScheduledBackups() {
                 return <Switch defaultChecked={sched.enabled} onChange={toggleEnabled} />
             },
         },
-        { title: 'Enabled', dataIndex: 'enabled' },
+        { title: 'Cluster', dataIndex: 'cluster' },
         { title: 'Schedule', dataIndex: 'schedule' },
         { title: 'Last Run Time', dataIndex: 'last_run_time' },
         { title: 'Database', dataIndex: 'database' },
@@ -149,6 +164,13 @@ export default function ScheduledBackups() {
                     initialValues={{ remember: true }}
                     autoComplete="on"
                 >
+                    <Form.Item name="cluster" label="Cluster">
+                        <Select>
+                            {clusters.clusters.map(cluster => (
+                                <Select.Option value={cluster.cluster}>{cluster.cluster}</Select.Option>
+                            ))}
+                        </Select>
+                    </Form.Item>
                     <Form.Item<FieldType>
                         label="Schedule"
                         name="schedule"

frontend/src/pages/Clusters/Clusters.tsx

@@ -23,7 +23,7 @@ interface Cluster {
     nodes: ClusterNode[]
 }
 
-interface Clusters {
+export interface Clusters {
     clusters: Cluster[]
 }
 
@@ -70,7 +70,7 @@ export default function Clusters() {
             <br />
             <Row gutter={8} style={{ paddingBottom: 8 }}>
                 <ul>
-                    {clusters.clusters.map((cluster) => (
+                    {clusters.clusters.map(cluster => (
                         <>
                             <h1 key={cluster.cluster}>{cluster.cluster}</h1>
                             <Table columns={columns} dataSource={cluster.nodes} loading={loadingClusters} />

housewatch/celery.py

@@ -46,7 +46,13 @@ def schedule_backups():
     backups = ScheduledBackup.objects.filter(enabled=True)
     now = timezone.now()
     for backup in backups:
-        nr = croniter(backup.schedule, backup.last_run_time).get_next(datetime)
+        lrt = backup.last_run_time
+        if lrt is None:
+            lrt = backup.created_at
+        nr = croniter(backup.schedule, lrt).get_next(datetime)
+        if nr.tzinfo is None:
+            nr = timezone.make_aware(nr)
+        logger.info("Checking backup", backup_id=backup.id, next_run=nr, now=now)
         if nr < now:
             run_backup.delay(backup.id)
             backup.last_run_time = now

housewatch/clickhouse/backups.py

@@ -1,7 +1,7 @@
 import structlog
 from collections import defaultdict
 from datetime import datetime
-from housewatch.clickhouse.client import run_query
+from housewatch.clickhouse.client import run_query, run_query_on_shards
 from housewatch.models.backup import ScheduledBackup, ScheduledBackupRun
 
 from django.conf import settings
@@ -28,13 +28,26 @@ def get_backup(backup, cluster=None):
         return run_query(QUERY, {"uuid": backup}, use_cache=False)
 
 
-def create_table_backup(database, table, bucket, path, aws_key=None, aws_secret=None):
+def create_table_backup(database, table, bucket, path, cluster=None, aws_key=None, aws_secret=None):
     if aws_key is None or aws_secret is None:
         aws_key = settings.AWS_ACCESS_KEY_ID
         aws_secret = settings.AWS_SECRET_ACCESS_KEY
     QUERY = """BACKUP TABLE %(database)s.%(table)s
     TO S3('https://%(bucket)s.s3.amazonaws.com/%(path)s', '%(aws_key)s', '%(aws_secret)s')
     ASYNC"""
+    if cluster:
+        return run_query_on_shards(
+            QUERY,
+            {
+                "database": database,
+                "table": table,
+                "bucket": bucket,
+                "path": path,
+                "aws_key": aws_key,
+                "aws_secret": aws_secret,
+            },
+            cluster=cluster,
+        )
     return run_query(
         QUERY,
         {
@@ -49,13 +62,26 @@ def create_table_backup(database, table, bucket, path, aws_key=None, aws_secret=
     )
 
 
-def create_database_backup(database, bucket, path, aws_key=None, aws_secret=None):
+def create_database_backup(database, bucket, path, cluster=None, aws_key=None, aws_secret=None):
     if aws_key is None or aws_secret is None:
         aws_key = settings.AWS_ACCESS_KEY_ID
         aws_secret = settings.AWS_SECRET_ACCESS_KEY
     QUERY = """BACKUP DATABASE %(database)s 
                 TO S3('https://%(bucket)s.s3.amazonaws.com/%(path)s', '%(aws_key)s', '%(aws_secret)s')
                 ASYNC"""
+    if cluster:
+        return run_query_on_shards(
+            QUERY,
+            {
+                "database": database,
+                "table": table,
+                "bucket": bucket,
+                "path": path,
+                "aws_key": aws_key,
+                "aws_secret": aws_secret,
+            },
+            cluster=cluster,
+        )
     return run_query(
         QUERY,
         {
@@ -78,6 +104,7 @@ def run_backup(backup_id):
             backup.database,
             backup.bucket,
             path,
+            backup.cluster,
             backup.aws_access_key_id,
             backup.aws_secret_access_key,
         )[0]["id"]
@@ -87,6 +114,7 @@ def run_backup(backup_id):
             backup.table,
             backup.bucket,
             path,
+            backup.cluster,
             backup.aws_access_key_id,
             backup.aws_secret_access_key,
         )[0]["id"]

housewatch/clickhouse/client.py

@@ -31,7 +31,7 @@
 def run_query_on_shards(
     query: str,
     params: Dict[str, str | int] = {},
-    settings: Dict[str, str | int] = {},
+    query_settings: Dict[str, str | int] = {},
     query_id: Optional[str] = None,
     substitute_params: bool = True,
     cluster: Optional[str] = None,
@@ -54,7 +54,7 @@ def run_query_on_shards(
             send_receive_timeout=30,
             password=settings.CLICKHOUSE_PASSWORD,
         )
-        result = client.execute(final_query, settings=settings, with_column_types=True, query_id=query_id)
+        result = client.execute(final_query, settings=query_settings, with_column_types=True, query_id=query_id)
         response = []
         for res in result[0]:
             item = {}

housewatch/clickhouse/clusters.py

@@ -23,7 +23,7 @@ def get_shards(cluster):
     cluster = get_cluster(cluster)
     nodes = defaultdict(list)
     for node in cluster:
-        nodes[node["shard_number"]].append(node)
+        nodes[node["shard_num"]].append(node)
     return nodes
 
 

housewatch/migrations/0009_scheduledbackup_cluster_alter_scheduledbackup_id.py

@@ -0,0 +1,24 @@
+# Generated by Django 4.1.1 on 2023-08-24 01:33
+
+from django.db import migrations, models
+import uuid
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('housewatch', '0008_remove_scheduledbackup_aws_endpoint_url_and_more'),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name='scheduledbackup',
+            name='cluster',
+            field=models.CharField(max_length=255, null=True),
+        ),
+        migrations.AlterField(
+            model_name='scheduledbackup',
+            name='id',
+            field=models.UUIDField(default=uuid.uuid4, editable=False, primary_key=True, serialize=False),
+        ),
+    ]

housewatch/models/backup.py

@@ -16,6 +16,7 @@ class ScheduledBackup(models.Model):
     schedule: models.CharField = models.CharField(max_length=255)
     table: models.CharField = models.CharField(max_length=255, null=True)
     database: models.CharField = models.CharField(max_length=255)
+    cluster: models.CharField = models.CharField(max_length=255, null=True)
     bucket: models.CharField = models.CharField(max_length=255)
     path: models.CharField = models.CharField(max_length=255)
     # if set these will override the defaults from settings